generated from Nodarx/template
feat: serve message lists from the cache
Signed-off-by: Sebastian Krupinski <krupinski01@gmail.com>
This commit is contained in:
@@ -22,11 +22,18 @@ use KTXF\Resource\Identifier\CollectionIdentifier;
|
||||
use KTXF\Resource\Identifier\EntityIdentifier;
|
||||
use KTXF\Resource\Identifier\EntityIdentifierInterface;
|
||||
use KTXF\Resource\Range\IRange;
|
||||
use KTXF\Resource\Range\IRangeTally;
|
||||
use KTXF\Resource\Range\RangeAnchorType;
|
||||
use KTXF\Resource\Sort\ISort;
|
||||
use KTXM\ProviderImap\Service\Cache\HarmonizationService;
|
||||
use KTXM\ProviderImap\Service\Cache\MessageDeltaService;
|
||||
use KTXM\ProviderImap\Service\Cache\MessageIngestor;
|
||||
use KTXM\ProviderImap\Service\Cache\MessageQueryBuilder;
|
||||
use KTXM\ProviderImap\Service\Live\LiveMailService;
|
||||
use KTXM\ProviderImap\Stores\MailboxStore;
|
||||
use KTXM\ProviderImap\Stores\MessageFileStore;
|
||||
use KTXM\ProviderImap\Stores\MessageStore;
|
||||
use UnexpectedValueException;
|
||||
|
||||
/**
|
||||
* IMAP mail service that serves reads from the cache.
|
||||
@@ -41,11 +48,18 @@ class CachedService extends ServiceBase
|
||||
/** Seconds after which a mailbox (or the mailbox list) is harmonized again on access */
|
||||
public const FRESHNESS_WINDOW = 60;
|
||||
|
||||
/** Messages hydrated from the cache per meta store query */
|
||||
private const HYDRATE_BATCH_SIZE = 100;
|
||||
|
||||
private ?LiveMailService $liveMail = null;
|
||||
private ?LiveService $live = null;
|
||||
|
||||
public function __construct(
|
||||
private readonly MailboxStore $mailboxStore,
|
||||
private readonly MessageStore $messageStore,
|
||||
private readonly MessageFileStore $fileStore,
|
||||
private readonly MessageIngestor $ingestor,
|
||||
private readonly MessageQueryBuilder $queries,
|
||||
private readonly HarmonizationService $harmonizer,
|
||||
private readonly MessageDeltaService $deltas,
|
||||
) {}
|
||||
@@ -119,14 +133,42 @@ class CachedService extends ServiceBase
|
||||
|
||||
public function entityListBulk(string|int $collection, ?IFilter $filter = null, ?ISort $sort = null, ?IRange $range = null, ?array $properties = null): array
|
||||
{
|
||||
// served from the cache in 5.2b
|
||||
return $this->live()->entityListBulk($collection, $filter, $sort, $range, $properties);
|
||||
return iterator_to_array($this->entityListStream($collection, $filter, $sort, $range, $properties), true);
|
||||
}
|
||||
|
||||
/**
|
||||
* List messages of a mailbox.
|
||||
*
|
||||
* - mailbox not harmonized yet: streamed from the server, each message cached as it passes
|
||||
* - harmonized: UIDs from the meta store (filter / sort translated, range applied like the
|
||||
* live service), content from message.json; a filter on body / full text has the server
|
||||
* find the UIDs (IMAP SEARCH) and only the content comes from the cache
|
||||
*/
|
||||
public function entityListStream(string|int $collection, ?IFilter $filter = null, ?ISort $sort = null, ?IRange $range = null, ?array $properties = null): Generator
|
||||
{
|
||||
// served from the cache in 5.2b
|
||||
return $this->live()->entityListStream($collection, $filter, $sort, $range, $properties);
|
||||
$mailbox = (string) $collection;
|
||||
$state = $this->mailboxStore->state($this->serviceId(), $mailbox);
|
||||
|
||||
if (!$state['harmonizationComplete'] || $state['uidValidity'] === null) {
|
||||
yield from $this->listFromServer($mailbox, $filter, $sort, $range);
|
||||
return;
|
||||
}
|
||||
|
||||
$uidValidity = (int) $state['uidValidity'];
|
||||
|
||||
if ($this->queries->needsServer($filter)) {
|
||||
$uids = $this->liveMail()->entityFind($mailbox, $filter, $sort, $range);
|
||||
} else {
|
||||
$order = $this->queries->sort($sort);
|
||||
$uids = self::applyRange(
|
||||
$this->messageStore->query($this->serviceId(), $mailbox, $uidValidity, $this->queries->filter($filter), $order['sort'], $order['collation']),
|
||||
$range,
|
||||
);
|
||||
}
|
||||
|
||||
foreach ($this->hydrate($mailbox, $uidValidity, $uids) as $entity) {
|
||||
yield $entity->urn() => $entity;
|
||||
}
|
||||
}
|
||||
|
||||
public function entityFetchBulk(EntityIdentifierInterface ...$identifiers): array
|
||||
@@ -211,6 +253,142 @@ class CachedService extends ServiceBase
|
||||
|
||||
// ── Internals ────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Stream a list from the server, caching each message as it passes (cold mailbox).
|
||||
*/
|
||||
private function listFromServer(string $mailbox, ?IFilter $filter, ?ISort $sort, ?IRange $range): Generator
|
||||
{
|
||||
$uidValidity = $this->cacheGeneration($mailbox);
|
||||
|
||||
foreach ($this->liveMail()->entityList($mailbox, $filter, $sort, $range) as $entity) {
|
||||
if ($uidValidity !== null) {
|
||||
$this->ingestQuietly($uidValidity, $entity);
|
||||
}
|
||||
yield $entity->urn() => $entity;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Entities for UIDs in the given order: cached content + meta flags, with messages
|
||||
* missing from the cache (or written with another schema version) fetched from the
|
||||
* server and cached.
|
||||
*
|
||||
* @param int[] $uids
|
||||
* @return Generator<int, EntityResource>
|
||||
*/
|
||||
private function hydrate(string $mailbox, int $uidValidity, array $uids): Generator
|
||||
{
|
||||
foreach (array_chunk($uids, self::HYDRATE_BATCH_SIZE) as $batch) {
|
||||
$metas = $this->messageStore->fetchMany($this->serviceId(), $mailbox, $uidValidity, ...$batch);
|
||||
|
||||
$entities = [];
|
||||
$missing = [];
|
||||
foreach ($batch as $uid) {
|
||||
$entity = isset($metas[$uid]) ? $this->entityFromCache($mailbox, $uidValidity, $metas[$uid]) : null;
|
||||
if ($entity === null) {
|
||||
$missing[] = $uid;
|
||||
continue;
|
||||
}
|
||||
$entities[$uid] = $entity;
|
||||
}
|
||||
|
||||
if ($missing !== []) {
|
||||
foreach ($this->liveMail()->entityFetch($mailbox, ...$missing) as $uid => $entity) {
|
||||
$this->ingestQuietly($uidValidity, $entity);
|
||||
$entities[$uid] = $entity;
|
||||
}
|
||||
}
|
||||
|
||||
foreach ($batch as $uid) {
|
||||
if (isset($entities[$uid])) {
|
||||
yield $uid => $entities[$uid];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* An entity from its meta document and message.json; null when the content is missing or outdated.
|
||||
*/
|
||||
private function entityFromCache(string $mailbox, int $uidValidity, array $meta): ?EntityResource
|
||||
{
|
||||
$content = $this->fileStore->read($this->tenantId(), $this->serviceId(), $mailbox, $uidValidity, (int) $meta['uid']);
|
||||
if ($content === null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
try {
|
||||
return $this->entityFresh()->fromCacheContent($content)->fromCacheMeta($meta);
|
||||
} catch (UnexpectedValueException) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* UIDVALIDITY to cache a not yet harmonized mailbox under; null when it cannot be cached.
|
||||
*/
|
||||
private function cacheGeneration(string $mailbox): ?int
|
||||
{
|
||||
// the mailbox document holds the change sequence, so it has to exist before ingesting
|
||||
if ($this->mailboxStore->fetch($this->serviceId(), $mailbox) === null) {
|
||||
try {
|
||||
$this->harmonizer()->harmonizeMailboxes();
|
||||
} catch (\Throwable) {
|
||||
return null;
|
||||
}
|
||||
if ($this->mailboxStore->fetch($this->serviceId(), $mailbox) === null) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
$state = $this->mailboxStore->state($this->serviceId(), $mailbox);
|
||||
if ($state['uidValidity'] !== null) {
|
||||
return (int) $state['uidValidity'];
|
||||
}
|
||||
|
||||
return $this->liveMail()->mailboxFetch($mailbox)?->uidValidity();
|
||||
}
|
||||
|
||||
/**
|
||||
* Cache a message; a failing cache write must not fail the read that triggered it.
|
||||
*/
|
||||
private function ingestQuietly(int $uidValidity, EntityResource $entity): void
|
||||
{
|
||||
try {
|
||||
$this->ingestor->ingest($this->tenantId(), $this->serviceId(), $uidValidity, $entity);
|
||||
} catch (\Throwable) {
|
||||
// the next harmonization caches it
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply a list range to sorted UIDs, as the live service does: absolute skips `position`
|
||||
* messages, relative starts at the UID given as `position`.
|
||||
*
|
||||
* @param int[] $uids
|
||||
* @return int[]
|
||||
*/
|
||||
private static function applyRange(array $uids, ?IRange $range): array
|
||||
{
|
||||
if (!$range instanceof IRangeTally) {
|
||||
return array_values($uids);
|
||||
}
|
||||
|
||||
$tally = max(0, $range->getTally());
|
||||
if ($tally === 0) {
|
||||
return [];
|
||||
}
|
||||
|
||||
if ($range->getAnchor() === RangeAnchorType::ABSOLUTE) {
|
||||
$start = max(0, (int) $range->getPosition());
|
||||
} else {
|
||||
$index = array_search((int) $range->getPosition(), $uids, true);
|
||||
$start = $index === false ? 0 : $index;
|
||||
}
|
||||
|
||||
return array_values(array_slice($uids, $start, $tally));
|
||||
}
|
||||
|
||||
/**
|
||||
* Build a collection from its cached document; its signature is the delta signature of the mailbox.
|
||||
*/
|
||||
@@ -240,6 +418,11 @@ class CachedService extends ServiceBase
|
||||
return (string) $this->identifier();
|
||||
}
|
||||
|
||||
private function tenantId(): string
|
||||
{
|
||||
return (string) $this->tenantIdentifier();
|
||||
}
|
||||
|
||||
/**
|
||||
* One IMAP connection per service object, shared by the LiveService and harmonization.
|
||||
*/
|
||||
|
||||
Reference in New Issue
Block a user