feat: harmonize mailbox messages

Signed-off-by: Sebastian Krupinski <krupinski01@gmail.com>

lib/Service/Cache/HarmonizationService.php

@@ -9,13 +9,17 @@ declare(strict_types=1);
 
 namespace KTXM\ProviderImap\Service\Cache;
 
+use KTXM\ProviderImap\Client\Protocol\Command\Argument\FetchOptions;
 use KTXM\ProviderImap\Providers\CollectionResource;
+use KTXM\ProviderImap\Providers\EntityResource;
+use KTXM\ProviderImap\Providers\MessageProperties;
 use KTXM\ProviderImap\Providers\Service;
 use KTXM\ProviderImap\Service\Live\LiveMailService;
 use KTXM\ProviderImap\Stores\MailboxStore;
 use KTXM\ProviderImap\Stores\MessageFileStore;
 use KTXM\ProviderImap\Stores\MessageStore;
 use LogicException;
+use RuntimeException;
 
 /**
  * Brings the cache of one service in line with its IMAP server.
@@ -25,6 +29,12 @@ use LogicException;
  */
 class HarmonizationService
 {
+    /** Messages fetched per UID FETCH when ingesting; each is cached as it streams in */
+    public const INGEST_BATCH_SIZE = 200;
+
+    /** Lifetime of a mailbox harmonization lock in seconds; long enough for a large first run */
+    public const LOCK_TTL = 900;
+
     private ?Service $service = null;
     private ?LiveMailService $live = null;
 
@@ -32,6 +42,7 @@ class HarmonizationService
         private readonly MailboxStore $mailboxStore,
         private readonly MessageStore $messageStore,
         private readonly MessageFileStore $fileStore,
+        private readonly MessageIngestor $ingestor,
     ) {}
 
     /**
@@ -87,6 +98,138 @@ class HarmonizationService
         return $result;
     }
 
+    /**
+     * Harmonize the messages of one mailbox with the server.
+     *
+     * Ingests messages that are not cached yet (newest first), updates changed
+     * flags and removes messages expunged on the server. A changed UIDVALIDITY
+     * discards the mailbox's cache first. Never waits: when another run holds the
+     * mailbox lock, nothing is done.
+     *
+     * @return array{status: 'harmonized'|'skipped', reset: bool, added: int, updated: int, removed: int, complete: bool}
+     */
+    public function harmonizeMessages(string $mailbox): array
+    {
+        [$service, $live] = $this->bound();
+        $tenantId = (string) $service->tenantIdentifier();
+        $serviceId = (string) $service->identifier();
+
+        $result = ['status' => 'skipped', 'reset' => false, 'added' => 0, 'updated' => 0, 'removed' => 0, 'complete' => false];
+
+        // the lock lives on the mailbox document, so it has to exist first
+        if ($this->mailboxStore->fetch($serviceId, $mailbox) === null) {
+            $this->harmonizeMailboxes();
+            if ($this->mailboxStore->fetch($serviceId, $mailbox) === null) {
+                throw new RuntimeException("Mailbox not found on server: {$mailbox}");
+            }
+        }
+
+        $owner = bin2hex(random_bytes(8));
+        if (!$this->mailboxStore->acquireLock($serviceId, $mailbox, $owner, self::LOCK_TTL)) {
+            return $result;
+        }
+
+        try {
+            $state = $this->mailboxStore->state($serviceId, $mailbox);
+            $selected = $live->collectionFetch($mailbox)
+                ?? throw new RuntimeException("Mailbox not found on server: {$mailbox}");
+            $uidValidity = $selected->uidValidity()
+                ?? throw new RuntimeException("Server did not report UIDVALIDITY for mailbox: {$mailbox}");
+
+            // a new UIDVALIDITY invalidates every cached UID of the mailbox
+            if ($state['uidValidity'] !== null && (int) $state['uidValidity'] !== $uidValidity) {
+                $this->messageStore->deleteByMailbox($serviceId, $mailbox);
+                $this->fileStore->deleteByMailbox($tenantId, $serviceId, $mailbox);
+                $result['reset'] = true;
+            }
+
+            $cachedFlags = $this->messageStore->flags($serviceId, $mailbox, $uidValidity);
+
+            // stream every UID and its flags from the server: cached messages get their flags
+            // compared (and are dropped from $cachedFlags), unknown UIDs are collected as new.
+            // No other IMAP command may run until the stream is consumed.
+            $new = [];
+            foreach ($live->entityFlags($mailbox) as $uid => $flags) {
+                if (!isset($cachedFlags[$uid])) {
+                    $new[] = $uid;
+                    continue;
+                }
+
+                $flags = MessageProperties::normalizeFlags($flags);
+                if (!self::sameFlags($flags, $cachedFlags[$uid])) {
+                    $this->messageStore->updateFlags($serviceId, $mailbox, $uidValidity, $uid, $flags);
+                    $result['updated']++;
+                }
+                unset($cachedFlags[$uid]);
+            }
+
+            // new: newest first so recent mail is available soonest
+            rsort($new);
+            $missing = [];
+            foreach (array_chunk($new, self::INGEST_BATCH_SIZE) as $batch) {
+                $ingested = $this->ingestBatch($tenantId, $serviceId, $mailbox, $uidValidity, $batch);
+                $result['added'] += count($ingested);
+                array_push($missing, ...array_diff($batch, $ingested));
+            }
+
+            // expunged: cached UIDs the server did not list; meta documents first, then files
+            $expunged = array_keys($cachedFlags);
+            if ($expunged !== []) {
+                $this->messageStore->delete($serviceId, $mailbox, $uidValidity, ...$expunged);
+                $this->fileStore->delete($tenantId, $serviceId, $mailbox, $uidValidity, ...$expunged);
+                $result['removed'] = count($expunged);
+            }
+
+            // UIDs that could not be fetched (e.g. expunged meanwhile) are retried next run
+            $result['complete'] = $missing === [];
+            $result['status'] = 'harmonized';
+
+            $this->mailboxStore->updateState($serviceId, $mailbox, [
+                'uidValidity'           => $uidValidity,
+                'uidNext'               => $selected->uidNext(),
+                'highestModSeq'         => $selected->highestModSeq(),
+                'harmonizedAt'          => time(),
+                'harmonizationComplete' => $result['complete'],
+            ]);
+        } finally {
+            $this->mailboxStore->releaseLock($serviceId, $mailbox, $owner);
+        }
+
+        return $result;
+    }
+
+    /**
+     * Fetch a batch of messages and cache each one as it streams in.
+     *
+     * @param int[] $uids
+     * @return int[] UIDs that were cached
+     */
+    private function ingestBatch(string $tenantId, string $serviceId, string $mailbox, int $uidValidity, array $uids): array
+    {
+        [$service, $live] = $this->bound();
+        $options = FetchOptions::message()->withBodyText(MessageIngestor::BODY_TEXT_LIMIT);
+
+        $ingested = [];
+        foreach ($live->entityFetch($mailbox, $options, ...$uids) as $message) {
+            $entity = (new EntityResource($service->provider(), $serviceId))->fromImap($message, $mailbox);
+            array_push($ingested, ...$this->ingestor->ingest($tenantId, $serviceId, $uidValidity, $entity));
+        }
+
+        return $ingested;
+    }
+
+    /**
+     * @param string[] $left
+     * @param string[] $right
+     */
+    private static function sameFlags(array $left, array $right): bool
+    {
+        sort($left);
+        sort($right);
+
+        return $left === $right;
+    }
+
     /**
      * Remove a mailbox and everything cached for it: meta documents first, then files, then the mailbox.
      */