update gmail

This commit is contained in:
Paul Lützner committed 2026-07-17 12:47:20 +02:00
1 parent 165e8b599f
commit be5550cd24
24 files changed
+688 -244

No files matched your search

+1
View File
@@ -70,4 +70,5 @@ MERCURE_JWT_SECRET="!ChangeThisMercureHubJWTSecretKey!"
###> google oauth ###
GOOGLE_OAUTH_CLIENT_ID=
GOOGLE_OAUTH_CLIENT_SECRET=
GMAIL_PUBSUB_TOPIC=projects/plMail/topics/gmail-push
###< google oauth ###
+3 -1
View File
@@ -21,7 +21,9 @@ framework:
Symfony\Component\Mailer\Messenger\SendEmailMessage: async
Symfony\Component\Notifier\Message\ChatMessage: async
Symfony\Component\Notifier\Message\SmsMessage: async
App\Message\SyncMailboxMessage: async
App\Message\SyncImapMailboxMessage: async
App\Message\SyncAccountMessage: async
App\Message\SendMessageMessage: async
App\Message\ApplyImapFlagsMessage: async
App\Message\HarvestContactsMessage: async
App\Message\SyncGmailMessageBatchMessage: async
+2 -2
View File
@@ -1508,8 +1508,8 @@ use Symfony\Component\Config\Loader\ParamConfigurator as Param;
* }
* @psalm-type MercureConfig = array{
* hubs?: array<string, array{ // Default: []
* url?: scalar|Param|null, // URL of the hub's publish endpoint
* public_url?: scalar|Param|null, // URL of the hub's public endpoint // Default: null
* url?: scalar|Param|null, // URL of the hub's publish endpoint // Default: null
* public_url?: scalar|Param|null, // URL of the hub's public endpoint
* jwt?: string|array{ // JSON Web Token configuration.
* value?: scalar|Param|null, // JSON Web Token to use to publish to this hub.
* provider?: scalar|Param|null, // The ID of a service to call to provide the JSON Web Token.
+33
View File
@@ -0,0 +1,33 @@
<?php
declare(strict_types=1);
namespace DoctrineMigrations;
use Doctrine\DBAL\Schema\Schema;
use Doctrine\Migrations\AbstractMigration;
/**
* Auto-generated Migration: Please modify to your needs!
*/
final class Version20260717095340 extends AbstractMigration
{
public function getDescription(): string
{
return '';
}
public function up(Schema $schema): void
{
// this up() migration is auto-generated, please modify it to your needs
$this->addSql('ALTER TABLE message ADD gmail_id VARCHAR(255) DEFAULT NULL');
$this->addSql('CREATE INDEX idx_message_gmail_id ON message (gmail_id)');
}
public function down(Schema $schema): void
{
// this down() migration is auto-generated, please modify it to your needs
$this->addSql('DROP INDEX idx_message_gmail_id');
$this->addSql('ALTER TABLE message DROP gmail_id');
}
}
+17 -36
View File
@@ -4,10 +4,9 @@ declare(strict_types=1);
namespace App\Command;
use App\Domain\Enum\MailProvider;
use App\Enum\AuthType;
use App\Repository\MailboxRepository;
use App\Repository\AccountRepository;
use App\Service\Gmail\GmailWatchService;
use DateTimeImmutable;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
@@ -15,13 +14,12 @@ use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;
/**
* Renews Gmail push-notification watches that are expiring within 24 hours.
* Renews Gmail push-notification watches expiring within 24 hours.
*
* Run this daily via cron or Symfony Scheduler:
* Run daily (cron or Symfony Scheduler):
* 0 6 * * * php bin/console app:gmail:renew-watches
*
* Google watch registrations expire after at most 7 days so a daily renewal
* gives us a comfortable 6-day buffer.
* Watches expire after at most 7 days, so a daily renewal gives a 6-day buffer.
*/
#[AsCommand(
name: 'app:gmail:renew-watches',
@@ -30,7 +28,7 @@ use Symfony\Component\Console\Style\SymfonyStyle;
final class GmailWatchRenewalCommand extends Command
{
public function __construct(
private readonly MailboxRepository $mailboxRepository,
private readonly AccountRepository $accountRepository,
private readonly GmailWatchService $watchService,
) {
parent::__construct();
@@ -41,32 +39,19 @@ final class GmailWatchRenewalCommand extends Command
$io = new SymfonyStyle($input, $output);
$io->title('Gmail watch renewal');
$mailboxes = $this->mailboxRepository->findAll();
$renewed = 0;
$skipped = 0;
$accounts = $this->accountRepository->findBy(['isActive' => true]);
$renewed = 0;
$skipped = 0;
foreach ($mailboxes as $mailbox) {
$account = $mailbox->getAccount();
// Only Gmail OAuth accounts
if (false === (
AuthType::OAuth2->value === $account->getAuthType()
&& MailProvider::Google->value === $account->getOauthProvider()
)) {
foreach ($accounts as $account) {
if (false === $account->isGmail()) {
$skipped++;
continue;
}
// Only sync-enabled mailboxes
if (false === $mailbox->isSyncEnabled()) {
$skipped++;
continue;
}
// Renew if expiry is within 24 hours or not yet registered
$expiry = $mailbox->getGmailWatchExpiry();
$expiry = $account->getGmailWatchExpiry();
$needsRenewal = (null === $expiry)
|| ($expiry <= new \DateTimeImmutable('+24 hours'));
|| ($expiry <= new DateTimeImmutable('+24 hours'));
if (false === $needsRenewal) {
$skipped++;
@@ -74,17 +59,13 @@ final class GmailWatchRenewalCommand extends Command
}
try {
$this->watchService->watch($mailbox);
$io->text(sprintf(
'✓ Renewed watch for %s / %s',
$account->getEmail(),
$mailbox->getName(),
));
$this->watchService->watch($account);
$io->text(sprintf('✓ Renewed watch for %s', $account->getEmail()));
$renewed++;
} catch (\Throwable $e) {
$io->error(sprintf(
'Failed to renew watch for mailbox %d: %s',
$mailbox->getId(),
'Failed to renew watch for account %d: %s',
$account->getId(),
$e->getMessage(),
));
}
+62
View File
@@ -0,0 +1,62 @@
<?php
declare(strict_types=1);
namespace App\Command;
use App\Message\SyncAccountMessage;
use App\Repository\AccountRepository;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;
use Symfony\Component\Messenger\MessageBusInterface;
#[AsCommand(
name: 'app:mail:sync',
description: 'Dispatch an account-level sync (IMAP or Gmail) for one or all active accounts',
)]
final class MailSyncCommand extends Command
{
public function __construct(
private readonly AccountRepository $accountRepository,
private readonly MessageBusInterface $bus,
) {
parent::__construct();
}
protected function configure(): void
{
$this->addArgument('account-id', InputArgument::OPTIONAL, 'Sync a single account by ID; omit for all active accounts');
}
protected function execute(InputInterface $input, OutputInterface $output): int
{
$io = new SymfonyStyle($input, $output);
$accountId = $input->getArgument('account-id');
if (null !== $accountId) {
$account = $this->accountRepository->find((int) $accountId);
if (null === $account) {
$io->error(sprintf('Account %s not found.', $accountId));
return Command::FAILURE;
}
$accounts = [$account];
} else {
$accounts = $this->accountRepository->findBy(['isActive' => true]);
}
foreach ($accounts as $account) {
$this->bus->dispatch(new SyncAccountMessage($account->getId()));
$io->text(sprintf('→ dispatched sync for %s (#%d)', $account->getEmail(), $account->getId()));
}
$io->success(sprintf('Dispatched %d account sync(s).', count($accounts)));
return Command::SUCCESS;
}
}
+54
View File
@@ -0,0 +1,54 @@
<?php
declare(strict_types=1);
namespace App\Controller;
use App\Entity\MessagePart;
use App\Service\Mail\AttachmentResolver;
use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
use Symfony\Component\HttpFoundation\BinaryFileResponse;
use Symfony\Component\HttpFoundation\Request;
use Symfony\Component\HttpFoundation\Response;
use Symfony\Component\HttpFoundation\ResponseHeaderBag;
use Symfony\Component\Routing\Attribute\Route;
use Symfony\Component\Security\Http\Attribute\IsGranted;
#[IsGranted('ROLE_USER')]
final class AttachmentController extends AbstractController
{
public function __construct(
private readonly AttachmentResolver $attachmentResolver,
) {}
#[Route('/mail/attachment/{id}', name: 'app_mail_attachment', methods: ['GET'])]
public function serve(MessagePart $part, Request $request): Response
{
$account = $part->getMessage()->getMailbox()->getAccount();
if ($account->getUsr() !== $this->getUser()) {
throw $this->createAccessDeniedException();
}
$absolutePath = $this->attachmentResolver->absolutePathFor($part);
$contentType = $part->getContentType() ?? 'application/octet-stream';
// Only images render inline; everything else (crucially any text/html)
// is forced to download so email-supplied markup never runs on our origin.
$inlineAllowed = true === str_starts_with($contentType, 'image/');
$forceDownload = true === $request->query->getBoolean('download')
|| false === $inlineAllowed;
$response = new BinaryFileResponse($absolutePath);
$response->headers->set('Content-Type', $contentType);
$response->headers->set('X-Content-Type-Options', 'nosniff');
$response->setContentDisposition(
true === $forceDownload
? ResponseHeaderBag::DISPOSITION_ATTACHMENT
: ResponseHeaderBag::DISPOSITION_INLINE,
$part->getFilename() ?? 'attachment',
);
return $response;
}
}
+16
View File
@@ -10,6 +10,7 @@ use Doctrine\Common\Collections\Collection;
use Doctrine\DBAL\Types\Types;
use Doctrine\ORM\Mapping as ORM;
#[ORM\Index(name: 'idx_message_gmail_id', columns: ['gmail_id'])]
#[ORM\Entity(repositoryClass: MessageRepository::class)]
class Message extends MessageModel
{
@@ -28,6 +29,9 @@ class Message extends MessageModel
#[ORM\Column(type: Types::TEXT, nullable: true)]
private ?string $messageId = null;
#[ORM\Column(length: 255, nullable: true)]
private ?string $gmailId = null;
#[ORM\Column(type: Types::TEXT, nullable: true)]
private ?string $subject = null;
@@ -441,4 +445,16 @@ class Message extends MessageModel
$this->cancelled = $cancelled;
return $this;
}
public function getGmailId(): ?string
{
return $this->gmailId;
}
public function setGmailId(?string $gmailId): static
{
$this->gmailId = $gmailId;
return $this;
}
}
@@ -0,0 +1,20 @@
<?php
declare(strict_types=1);
namespace App\Message;
/**
* A chunk of Gmail message IDs to fetch, build, and persist. Dispatched by
* GmailApiSyncer so the work parallelises across workers.
*/
readonly class SyncGmailMessageBatchMessage
{
/**
* @param list<string> $gmailIds
*/
public function __construct(
public int $mailboxId,
public array $gmailIds,
) {}
}
@@ -0,0 +1,107 @@
<?php
declare(strict_types=1);
namespace App\MessageHandler;
use App\Message\SyncGmailMessageBatchMessage;
use App\Repository\MailboxRepository;
use App\Repository\MessageRepository;
use App\Service\Gmail\GmailMessageBuilder;
use App\Service\HarvestContactsService;
use App\Service\Imap\MessageThreader;
use App\Service\Mail\GmailApiClient;
use App\Service\Mail\SyncNotifier;
use DateTimeImmutable;
use Doctrine\ORM\EntityManagerInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
#[AsMessageHandler]
final readonly class SyncGmailMessageBatchHandler
{
public function __construct(
private MailboxRepository $mailboxRepository,
private MessageRepository $messageRepository,
private GmailApiClient $apiClient,
private GmailMessageBuilder $messageBuilder,
private MessageThreader $messageThreader,
private HarvestContactsService $harvestService,
private SyncNotifier $syncNotifier,
private EntityManagerInterface $em,
private LoggerInterface $logger,
) {}
public function __invoke(SyncGmailMessageBatchMessage $message): void
{
$mailbox = $this->mailboxRepository->find($message->mailboxId);
if (null === $mailbox) {
$this->logger->warning('SyncGmailMessageBatch: mailbox not found', [
'mailboxId' => $message->mailboxId,
]);
return;
}
$account = $mailbox->getAccount();
// Dedup inside the batch too — batches can overlap across runs/retries.
$syncedGmailIds = array_flip(
$this->messageRepository->findSyncedGmailIds($mailbox)
);
$toFetch = [];
foreach ($message->gmailIds as $gmailId) {
if (true === isset($syncedGmailIds[$gmailId])) {
continue;
}
$toFetch[] = $gmailId;
}
if (count($toFetch) === 0) {
return;
}
$payloads = $this->apiClient->getMessages($account, $toFetch);
$built = [];
foreach ($payloads as $payload) {
try {
$entity = $this->messageBuilder->build($payload, $mailbox);
$this->em->persist($entity);
$built[] = $entity;
} catch (\Throwable $e) {
$this->logger->error('SyncGmailMessageBatch: build failed', [
'error' => $e->getMessage(),
]);
}
}
$this->em->flush();
foreach ($built as $entity) {
try {
$this->messageThreader->assignThread($entity, $account, $mailbox);
} catch (\Throwable $e) {
$this->logger->error('SyncGmailMessageBatch: threading failed', [
'messageId' => $entity->getId(),
'error' => $e->getMessage(),
]);
}
}
$this->em->flush();
// Harvest only this batch's messages, not the whole mailbox — keeps the
// fan-out linear instead of quadratic.
$this->harvestService->harvestMessages($account->getUsr(), $built);
$mailbox->setUnreadMessages($this->messageRepository->countUnseenForMailbox($mailbox));
$mailbox->setTotalMessages($this->messageRepository->countTotalForMailbox($mailbox));
$mailbox->setSyncedAt(new DateTimeImmutable());
$this->em->flush();
// Progressive UI refresh as each batch lands (publish only, no full harvest).
$this->syncNotifier->publishMailboxSynced($account, $mailbox);
}
}
+3 -4
View File
@@ -28,13 +28,12 @@ class MessageRepository extends ServiceEntityRepository
->getSingleColumnResult();
}
public function findSyncedMessageIds(Mailbox $mailbox): array
public function findSyncedGmailIds(Mailbox $mailbox): array
{
return $this->createQueryBuilder('m')
->select('m.messageId')
->select('m.gmailId')
->where('m.mailbox = :mailbox')
->andWhere('m.messageId IS NOT NULL')
->andWhere('m.imapUid IS NULL')
->andWhere('m.gmailId IS NOT NULL')
->setParameter('mailbox', $mailbox)
->getQuery()
->getSingleColumnResult();
+1 -1
View File
@@ -51,6 +51,6 @@ final readonly class GmailAccountSyncer implements AccountSyncerInterface
$this->gmailApiSyncer->syncIncremental($inbox);
}
return [$inbox->getId()];
return [];
}
}
+82 -144
View File
@@ -5,132 +5,71 @@ declare(strict_types=1);
namespace App\Service\Gmail;
use App\Entity\Mailbox;
use App\Message\SyncGmailMessageBatchMessage;
use App\Repository\MessageRepository;
use App\Service\Imap\MessageThreader;
use App\Service\Mail\GmailApiClient;
use DateTimeImmutable;
use Doctrine\ORM\EntityManagerInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\MessageBusInterface;
/**
* Syncs a Gmail mailbox using the Gmail REST API.
*
* Two entry points:
* initialSync() — fetches all messages, stores historyId for future incremental calls
* syncIncremental() — fetches only what changed since the last stored historyId
*
* The SyncMailboxMessageHandler decides which to call based on whether a
* historyId is already stored on the Mailbox.
* Plans Gmail INBOX sync work and fans it out to SyncGmailMessageBatchMessage
* jobs. Message building/persistence happens in SyncGmailMessageBatchHandler,
* so initial and incremental sync share one build path and parallelise.
*/
final class GmailApiSyncer
{
/** Max messages to fetch per initial-sync page (API max is 500) */
/** Message-list page size (API max is 500). */
private const int PAGE_SIZE = 500;
/** Gmail message IDs per fan-out batch. */
private const int BATCH_SIZE = 100;
public function __construct(
private readonly GmailApiClient $apiClient,
private readonly GmailMessageBuilder $messageBuilder,
private readonly MessageThreader $messageThreader,
private readonly MessageRepository $messageRepository,
private readonly GmailApiClient $apiClient,
private readonly MessageRepository $messageRepository,
private readonly EntityManagerInterface $em,
private readonly LoggerInterface $logger,
private readonly MessageBusInterface $bus,
private readonly LoggerInterface $logger,
) {}
// ── Public entry points ───────────────────────────────────────────────────
/**
* Full initial sync: fetch all INBOX message IDs, build Message entities,
* assign threads, store the current historyId.
*
* Safe to call on subsequent runs — already-synced messages are skipped
* via messageId dedup.
* Snapshot the historyId up front (so a crash mid-fan-out leaves the account
* incremental-ready), list every INBOX message ID, and dispatch batches for
* the ones we don't already have.
*/
public function initialSync(Mailbox $mailbox): void
{
$account = $mailbox->getAccount();
$this->logger->info('GmailApiSyncer: starting initial sync', [
$this->logger->info('GmailApiSyncer: planning initial sync', [
'mailboxId' => $mailbox->getId(),
'account' => $account->getEmail(),
]);
// Grab the current historyId before we start so we don't miss anything
// that arrives while we're paginating
$profile = $this->apiClient->getProfile($account);
$profile = $this->apiClient->getProfile($account);
$currentHistoryId = (string) ($profile['historyId'] ?? '');
// Load existing message IDs for dedup
$syncedMessageIds = array_flip(
$this->messageRepository->findSyncedMessageIds($mailbox)
);
// Persist the snapshot immediately. The backlog is filled by batch jobs
// (by ID) and dedup prevents double inserts, so storing it before the
// messages land is safe and makes a partial run incremental-ready.
if ('' !== $currentHistoryId) {
$account->setGmailHistoryId($currentHistoryId);
$this->em->flush();
}
$messageRefs = $this->apiClient->listMessages($account, [
'labelIds' => 'INBOX',
'maxResults' => self::PAGE_SIZE,
]);
$this->logger->info('GmailApiSyncer: fetched message list', [
'count' => count($messageRefs),
]);
$synced = 0;
foreach ($messageRefs as $ref) {
$gmailId = (string) ($ref['id'] ?? '');
if (true === isset($syncedMessageIds[$gmailId])) {
continue;
}
try {
$payload = $this->apiClient->getMessage($account, $gmailId);
$message = $this->messageBuilder->build($payload, $mailbox);
$this->em->persist($message);
$this->em->flush();
$this->messageThreader->assignThread($message, $account, $mailbox);
$this->em->flush();
$syncedMessageIds[$gmailId] = true;
$synced++;
// Flush in small batches to avoid memory pressure
if (true === ($synced % 50 === 0)) {
$this->em->clear();
$this->logger->info(sprintf('GmailApiSyncer: synced %d messages', $synced));
}
} catch (\Throwable $e) {
$this->logger->error('GmailApiSyncer: failed to sync message', [
'gmailId' => $gmailId,
'error' => $e->getMessage(),
]);
}
}
// Store historyId captured at the start — anything arriving after this
// point will be caught by the next incremental sync
if ('' !== $currentHistoryId) {
$account->setGmailHistoryId($currentHistoryId);
}
$mailbox->setSyncedAt(new DateTimeImmutable());
$mailbox->setTotalMessages(
$this->messageRepository->countTotalForMailbox($mailbox)
);
$mailbox->setUnreadMessages(
$this->messageRepository->countUnseenForMailbox($mailbox)
);
$this->em->flush();
$this->logger->info('GmailApiSyncer: initial sync complete', [
'synced' => $synced,
'historyId' => $currentHistoryId,
]);
$this->dispatchBatches($mailbox, $this->newGmailIds($mailbox, $messageRefs));
}
/**
* Incremental sync: fetch only messages added since the last historyId.
* Updates the stored historyId on completion.
* Read history since the stored historyId, dispatch batches for newly-added
* messages, and advance the stored historyId.
*/
public function syncIncremental(Mailbox $mailbox): void
{
@@ -138,28 +77,27 @@ final class GmailApiSyncer
$startHistoryId = $account->getGmailHistoryId();
if (null === $startHistoryId) {
$this->logger->warning(
'GmailApiSyncer: no historyId stored, falling back to initial sync',
['mailboxId' => $mailbox->getId()],
);
$this->logger->warning('GmailApiSyncer: no historyId stored, running initial sync', [
'mailboxId' => $mailbox->getId(),
]);
$this->initialSync($mailbox);
return;
}
$this->logger->info('GmailApiSyncer: incremental sync', [
'mailboxId' => $mailbox->getId(),
'startHistoryId' => $startHistoryId,
]);
try {
$result = $this->apiClient->listHistory($account, $startHistoryId, [
'labelId' => 'INBOX',
'historyTypes' => 'messageAdded',
'labelId' => 'INBOX',
'historyTypes' => 'messageAdded',
]);
} catch (\Throwable $e) {
// historyId too old (410 Gone) — fall back to initial sync
if (true === str_contains($e->getMessage(), '410')) {
$this->logger->warning('GmailApiSyncer: historyId expired, re-running initial sync');
// A too-old historyId returns 404 or 410 — rebuild from scratch.
if (
true === str_contains($e->getMessage(), '404')
|| true === str_contains($e->getMessage(), '410')
) {
$this->logger->warning('GmailApiSyncer: historyId expired, re-running initial sync', [
'mailboxId' => $mailbox->getId(),
]);
$account->setGmailHistoryId(null);
$this->em->flush();
$this->initialSync($mailbox);
@@ -169,63 +107,63 @@ final class GmailApiSyncer
throw $e;
}
$newHistoryId = $result['historyId'];
$history = $result['history'];
// Collect unique message IDs from messagesAdded events
$gmailIds = [];
foreach ($history as $record) {
$refs = [];
foreach ($result['history'] as $record) {
foreach ($record['messagesAdded'] ?? [] as $added) {
$id = (string) ($added['message']['id'] ?? '');
if ('' !== $id) {
$gmailIds[$id] = true;
$refs[] = ['id' => $id];
}
}
}
$this->logger->info('GmailApiSyncer: new messages from history', [
'count' => count($gmailIds),
]);
$this->dispatchBatches($mailbox, $this->newGmailIds($mailbox, $refs));
// Dedup against already-synced
$syncedMessageIds = array_flip(
$this->messageRepository->findSyncedMessageIds($mailbox)
$account->setGmailHistoryId((string) $result['historyId']);
$mailbox->setSyncedAt(new DateTimeImmutable());
$this->em->flush();
}
/**
* @param list<array{id?: string}> $refs
* @return list<string>
*/
private function newGmailIds(Mailbox $mailbox, array $refs): array
{
$syncedGmailIds = array_flip(
$this->messageRepository->findSyncedGmailIds($mailbox)
);
foreach (array_keys($gmailIds) as $gmailId) {
if (true === isset($syncedMessageIds[$gmailId])) {
$pending = [];
foreach ($refs as $ref) {
$gmailId = (string) ($ref['id'] ?? '');
if ('' === $gmailId) {
continue;
}
try {
$payload = $this->apiClient->getMessage($account, $gmailId);
$message = $this->messageBuilder->build($payload, $mailbox);
$this->em->persist($message);
$this->em->flush();
$this->messageThreader->assignThread($message, $account, $mailbox);
$this->em->flush();
} catch (\Throwable $e) {
$this->logger->error('GmailApiSyncer: failed to sync message', [
'gmailId' => $gmailId,
'error' => $e->getMessage(),
]);
if (true === isset($syncedGmailIds[$gmailId])) {
continue;
}
$pending[] = $gmailId;
}
// Always advance the stored historyId even if no new messages arrived
$account->setGmailHistoryId($newHistoryId);
$mailbox->setSyncedAt(new DateTimeImmutable());
$mailbox->setUnreadMessages(
$this->messageRepository->countUnseenForMailbox($mailbox)
);
$mailbox->setTotalMessages(
$this->messageRepository->countTotalForMailbox($mailbox)
);
$this->em->flush();
return $pending;
}
$this->logger->info('GmailApiSyncer: incremental sync complete', [
'newHistoryId' => $newHistoryId,
/**
* @param list<string> $gmailIds
*/
private function dispatchBatches(Mailbox $mailbox, array $gmailIds): void
{
$chunks = array_chunk($gmailIds, self::BATCH_SIZE);
foreach ($chunks as $chunk) {
$this->bus->dispatch(new SyncGmailMessageBatchMessage($mailbox->getId(), $chunk));
}
$this->logger->info('GmailApiSyncer: fanned out sync', [
'mailboxId' => $mailbox->getId(),
'messages' => count($gmailIds),
'batches' => count($chunks),
]);
}
}
+29 -21
View File
@@ -42,9 +42,8 @@ final class GmailMessageBuilder
$message = new Message();
$message->setMailbox($mailbox);
// Gmail message ID (not RFC Message-ID — that lives in the headers)
// We use it as the imapUid equivalent for dedup; store as messageId.
$gmailId = (string) ($payload['id'] ?? '');
$gmailId = $payload['id'] ?? '';
$message->setGmailId($gmailId);
// ── Headers ───────────────────────────────────────────────────────────
$headers = $this->indexHeaders($payload['payload']['headers'] ?? []);
@@ -215,13 +214,20 @@ final class GmailMessageBuilder
}
}
// Attachment reference (has an attachmentId, no inline data)
if (
true === isset($part['body']['attachmentId'])
&& '' !== ($part['filename'] ?? '')
) {
$hasAttachments = true;
$this->persistAttachmentStub($part, $message, $mailbox, $accountId, $fakeUid);
// Attachment or inline part backed by a separate attachmentId
if (true === isset($part['body']['attachmentId'])) {
$partHeaders = $this->indexHeaders($part['headers'] ?? []);
$filename = (string) ($part['filename'] ?? '');
$hasContentId = '' !== trim((string) ($partHeaders['content-id'] ?? ''), '<> ');
if ('' !== $filename || true === $hasContentId) {
$isInline = $this->persistAttachmentStub($part, $message, $mailbox, $accountId, $fakeUid);
// Only real (non-inline) attachments flip the paperclip.
if (false === $isInline) {
$hasAttachments = true;
}
}
}
// Recurse into sub-parts (multipart/*)
@@ -236,11 +242,11 @@ final class GmailMessageBuilder
}
/**
* Persist a MessagePart stub for an attachment.
* We store the Gmail attachmentId in storagePath for now; a separate job
* can download the content later if needed.
* Persist a MessagePart stub for an attachment. Bytes are fetched lazily by
* AttachmentResolver on first access.
*
* @param array<string,mixed> $part
* @return bool true if the part is inline (referenced from the body)
*/
private function persistAttachmentStub(
array $part,
@@ -248,28 +254,30 @@ final class GmailMessageBuilder
Mailbox $mailbox,
int $accountId,
int $fakeUid,
): void {
): bool {
$partHeaders = $this->indexHeaders($part['headers'] ?? []);
$filename = (string) ($part['filename'] ?? 'attachment');
$contentType = (string) ($part['mimeType'] ?? 'application/octet-stream');
$attachmentId = (string) ($part['body']['attachmentId'] ?? '');
$size = (int) ($part['body']['size'] ?? 0);
$isInline = str_contains(
strtolower((string) ($part['headers']['content-disposition'] ?? '')),
'inline',
);
$size = (int) ($part['body']['size'] ?? 0);
$contentId = trim((string) ($partHeaders['content-id'] ?? ''), '<> ');
$disposition = strtolower((string) ($partHeaders['content-disposition'] ?? ''));
$isInline = true === str_contains($disposition, 'inline') || '' !== $contentId;
$mp = new MessagePart();
$mp->setMessage($message);
$mp->setContentType($contentType);
$mp->setFilename($filename);
$mp->setContentId('' !== $contentId ? $contentId : null);
$mp->setDisposition($isInline ? 'inline' : 'attachment');
$mp->setSize($size);
// Store the Gmail attachment ID so we can fetch later;
// prefix distinguishes it from a real filesystem path
$mp->setStoragePath('gmail://' . $attachmentId);
$mp->setIsInline($isInline);
$this->em->persist($mp);
return $isInline;
}
private function decodeMimeHeader(string $value): string
+2
View File
@@ -9,6 +9,7 @@ use App\Service\Mail\GmailApiClient;
use DateTimeImmutable;
use Doctrine\ORM\EntityManagerInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\DependencyInjection\Attribute\Autowire;
/**
* Registers or renews a Gmail push-notification watch for a mailbox.
@@ -29,6 +30,7 @@ final class GmailWatchService
private readonly GmailApiClient $apiClient,
private readonly EntityManagerInterface $em,
private readonly LoggerInterface $logger,
#[Autowire(env: 'GMAIL_PUBSUB_TOPIC')]
private readonly string $pubSubTopicName,
) {}
+30 -28
View File
@@ -5,6 +5,7 @@ declare(strict_types=1);
namespace App\Service;
use App\Entity\Mailbox;
use App\Entity\Message;
use App\Entity\User;
use App\Repository\ContactRepository;
use App\Repository\MessageRepository;
@@ -19,43 +20,50 @@ final readonly class HarvestContactsService
private ContactRepository $contactRepository,
private LoggerInterface $logger,
) {}
/**
* Harvests all from/to/cc/bcc addresses from the given mailbox's messages
* and upserts them into the contact table for the owning user.
*
* Returns the total number of address occurrences processed.
*/
public function harvestForMailbox(Mailbox $mailbox): int
{
$user = $mailbox->getAccount()->getUsr();
$messages = $this->messageRepository->findByMailboxOrderedByDate($mailbox);
$batch = [];
$total = 0;
$total = $this->upsertFromMessages($user, $messages);
$this->logger->info('HarvestContactsService: mailbox done', [
'mailboxId' => $mailbox->getId(),
'messages' => count($messages),
'addresses' => $total,
]);
return $total;
}
/**
* @param list<Message> $messages
*/
public function harvestMessages(User $user, array $messages): int
{
return $this->upsertFromMessages($user, $messages);
}
/**
* @param iterable<Message> $messages
*/
private function upsertFromMessages(User $user, iterable $messages): int
{
$batch = [];
$total = 0;
foreach ($messages as $msg) {
if ($msg->getFromAddress() !== null && $msg->getFromAddress() !== '') {
$batch[] = [
'email' => $msg->getFromAddress(),
'name' => $msg->getFromName(),
];
$batch[] = ['email' => $msg->getFromAddress(), 'name' => $msg->getFromName()];
}
foreach ([
$msg->getToAddresses(),
$msg->getCcAddresses(),
$msg->getBccAddresses(),
] as $group) {
foreach ([$msg->getToAddresses(), $msg->getCcAddresses(), $msg->getBccAddresses()] as $group) {
if ($group === null) {
continue;
}
foreach ($group as $addr) {
if (isset($addr['address']) && $addr['address'] !== '') {
$batch[] = [
'email' => $addr['address'],
'name' => $addr['name'] ?? null,
];
$batch[] = ['email' => $addr['address'], 'name' => $addr['name'] ?? null];
}
}
}
@@ -72,12 +80,6 @@ final readonly class HarvestContactsService
$total += count($batch);
}
$this->logger->info('HarvestContactsService: mailbox done', [
'mailboxId' => $mailbox->getId(),
'messages' => count($messages),
'addresses' => $total,
]);
return $total;
}
}
+5 -3
View File
@@ -8,6 +8,7 @@ use App\Domain\Helper\ImapConnectionFactory;
use App\Entity\Account;
use App\Entity\Message;
use App\Repository\MailboxRepository;
use App\Service\Mail\AttachmentResolver;
use App\Service\Mail\MailSenderRegistry;
use DateTimeImmutable;
use Doctrine\ORM\EntityManagerInterface;
@@ -26,6 +27,7 @@ class MessageSendService
private readonly AttachmentStorageHelper $attachmentStorage,
private readonly MailSenderRegistry $senderRegistry,
private readonly ImapConnectionFactory $imapConnectionFactory,
private readonly AttachmentResolver $attachmentResolver,
) {
}
@@ -90,7 +92,7 @@ class MessageSendService
$subject = '';
}
$email = (new Email())
$email = new Email()
->from(new Address($account->getEmail(), $fromName))
->subject($subject);
@@ -131,13 +133,13 @@ class MessageSendService
}
$email->embedFromPath(
$this->attachmentStorage->getAbsolutePath($part->getStoragePath()),
$this->attachmentResolver->absolutePathFor($part),
$contentId,
$part->getContentType(),
);
} else {
$email->attachFromPath(
$this->attachmentStorage->getAbsolutePath($part->getStoragePath()),
$this->attachmentResolver->absolutePathFor($part),
$part->getFilename(),
$part->getContentType(),
);
+8 -1
View File
@@ -227,7 +227,14 @@ class MessageSyncer
$part->setDisposition($attachment->getDisposition() ?? 'attachment');
$part->setSize(strlen($content));
$part->setStoragePath($storagePath);
$part->setIsInline(($attachment->getDisposition() ?? '') === 'inline');
$contentId = $attachment->getId();
$normalizedCid = null !== $contentId ? trim((string) $contentId, '<> ') : '';
$part->setContentId('' !== $normalizedCid ? $normalizedCid : null);
$part->setIsInline(
($attachment->getDisposition() ?? '') === 'inline' || '' !== $normalizedCid
);
$this->em->persist($part);
}
+74
View File
@@ -0,0 +1,74 @@
<?php
declare(strict_types=1);
namespace App\Service\Mail;
use App\Domain\Helper\AttachmentStorageHelper;
use App\Entity\MessagePart;
use Doctrine\ORM\EntityManagerInterface;
use Psr\Log\LoggerInterface;
final readonly class AttachmentResolver
{
private const string GMAIL_SCHEME = 'gmail://';
public function __construct(
private AttachmentStorageHelper $attachmentStorage,
private GmailApiClient $apiClient,
private EntityManagerInterface $em,
private LoggerInterface $logger,
) {}
/**
* Absolute filesystem path to the part's content. If it's an unmaterialised
* Gmail attachment, it's downloaded and cached first.
*/
public function absolutePathFor(MessagePart $part): string
{
$storagePath = (string) $part->getStoragePath();
if (false === str_starts_with($storagePath, self::GMAIL_SCHEME)) {
return $this->attachmentStorage->getAbsolutePath($storagePath);
}
return $this->materialiseGmail($part, $storagePath);
}
private function materialiseGmail(MessagePart $part, string $storagePath): string
{
$attachmentId = substr($storagePath, strlen(self::GMAIL_SCHEME));
$message = $part->getMessage();
$gmailId = $message->getGmailId();
if (null === $gmailId || '' === $attachmentId) {
throw new \RuntimeException(sprintf(
'Cannot materialise Gmail attachment for part %d: missing gmailId or attachmentId.',
(int) $part->getId(),
));
}
$mailbox = $message->getMailbox();
$account = $mailbox->getAccount();
$content = $this->apiClient->getAttachment($account, $gmailId, $attachmentId);
$relativePath = $this->attachmentStorage->store(
$account->getId(),
$mailbox->getId(),
abs(crc32($gmailId)),
(string) $part->getFilename(),
$content,
);
$part->setStoragePath($relativePath);
$this->em->flush();
$this->logger->info('AttachmentResolver: materialised Gmail attachment', [
'partId' => $part->getId(),
'gmailId' => $gmailId,
]);
return $this->attachmentStorage->getAbsolutePath($relativePath);
}
}
+56
View File
@@ -68,6 +68,43 @@ final class GmailApiClient
return $messages;
}
/**
* Fetch multiple messages (format=full) concurrently, keyed by Gmail message ID.
* Concurrency is bounded by the HTTP client's connection pool, so a large batch
* won't open a request per ID at once. Failed fetches are dropped; re-runs are
* safe via dedup.
*
* @param list<string> $messageIds
* @return array<string,array<string,mixed>>
*/
public function getMessages(Account $account, array $messageIds): array
{
$token = $this->tokenManager->getValidAccessToken($account);
$responses = [];
foreach ($messageIds as $id) {
$responses[$id] = $this->httpClient->request(
'GET',
self::BASE . '/messages/' . urlencode($id),
[
'auth_bearer' => $token,
'query' => ['format' => 'full'],
],
);
}
$payloads = [];
foreach ($responses as $id => $response) {
try {
$payloads[$id] = $response->toArray();
} catch (\Throwable) {
// Skip; the batch is re-runnable and dedup makes retries safe.
}
}
return $payloads;
}
/**
* Fetch a single message in full RFC-2822 format (raw) so we can parse it
* exactly like an IMAP message.
@@ -192,4 +229,23 @@ final class GmailApiClient
return $response->toArray();
}
/**
* Download a single attachment's bytes. Gmail returns URL-safe base64 in `data`.
*/
public function getAttachment(Account $account, string $messageId, string $attachmentId): string
{
$token = $this->tokenManager->getValidAccessToken($account);
$response = $this->httpClient->request(
'GET',
self::BASE . '/messages/' . urlencode($messageId) . '/attachments/' . urlencode($attachmentId),
['auth_bearer' => $token],
);
$body = $response->toArray();
$data = (string) ($body['data'] ?? '');
return base64_decode(strtr($data, '-_', '+/'));
}
}
+4
View File
@@ -21,7 +21,11 @@ final readonly class SyncNotifier
public function notifyMailboxSynced(Account $account, Mailbox $mailbox): void
{
$this->bus->dispatch(new HarvestContactsMessage($mailbox->getId()));
$this->publishMailboxSynced($account, $mailbox);
}
public function publishMailboxSynced(Account $account, Mailbox $mailbox): void
{
$this->hub->publish(new Update(
topics: [
sprintf('mail/user/%d', $account->getUsr()->getId()),
+63
View File
@@ -0,0 +1,63 @@
<?php
declare(strict_types=1);
namespace App\Twig;
use App\Entity\Message;
use Symfony\Component\Routing\Generator\UrlGeneratorInterface;
use Twig\Extension\AbstractExtension;
use Twig\TwigFilter;
final class MailBodyExtension extends AbstractExtension
{
public function __construct(
private readonly UrlGeneratorInterface $urlGenerator,
) {}
public function getFilters(): array
{
return [
new TwigFilter('resolve_cid', [$this, 'resolveCid']),
];
}
/**
* Rewrite `cid:` references in an email body to point at our attachment
* route, so inline images load lazily through AttachmentResolver.
*/
public function resolveCid(?string $html, Message $message): string
{
if (null === $html || '' === $html) {
return (string) $html;
}
$map = [];
foreach ($message->getMessageParts() as $part) {
$cid = $part->getContentId();
if (null === $cid || '' === $cid) {
continue;
}
$map[strtolower($cid)] = $this->urlGenerator->generate(
'app_mail_attachment',
['id' => $part->getId()],
);
}
if (count($map) === 0) {
return $html;
}
return (string) preg_replace_callback(
'/(["\'])cid:([^"\']+)\1/i',
static function (array $m) use ($map): string {
$quote = $m[1];
$cid = strtolower(trim($m[2]));
return $quote . ($map[$cid] ?? ('cid:' . $m[2])) . $quote;
},
$html,
);
}
}
+15 -2
View File
@@ -34,7 +34,7 @@
{% if message.bodyHtml %}
<div class="prose prose-sm dark:prose-invert max-w-none">
{{ message.bodyHtml|raw }}
{{ message.bodyHtml|resolve_cid(message)|raw }}
</div>
{% else %}
<div class="whitespace-pre-wrap text-sm text-gray-700 dark:text-gray-300">
@@ -45,7 +45,20 @@
{% if message.hasAttachments %}
<div class="mt-6 pt-4 border-t border-gray-100 dark:border-gray-800">
<p class="text-xs font-medium text-gray-400 dark:text-gray-600 mb-2">{{ 'message.attachments'|trans }}</p>
{# attachment list goes here once MessagePart rendering is wired up #}
<div class="flex flex-wrap gap-2">
{% for part in message.messageParts if not part.isInline %}
href="{{ path('app_mail_attachment', { id: part.id }) }}?download=1"
class="inline-flex items-center gap-2 px-3 py-2 rounded-lg text-xs
bg-gray-100 dark:bg-white/5 border border-gray-200/60 dark:border-white/10
text-gray-700 dark:text-gray-300
hover:bg-gray-200 dark:hover:bg-white/10 transition-colors"
>
<i class="fa-solid fa-paperclip text-[11px]" aria-hidden="true"></i>
<span class="truncate max-w-[12rem]">{{ part.filename ?: 'attachment' }}</span>
</a>
{% endfor %}
</div>
</div>
{% endif %}
+1 -1
View File
@@ -100,7 +100,7 @@
{% if message.bodyHtml %}
<div
data-thread-message-target="shadowHost"
data-html="{{ message.bodyHtml|e('html_attr') }}"
data-html="{{ message.bodyHtml|resolve_cid(message)|e('html_attr') }}"
class="w-full"
></div>
{% else %}