// Package imapimport implementiert IMP-01: den Scheduler für // periodischen IMAP-Postfach-Abruf mit UID-basiertem Delta-Sync. Baut // auf ING-01 (mail/internal/imap, IMAP-Server-Grundgerüst inkl. UID // FETCH) und ING-05 (mail/internal/folderstate, UIDVALIDITY/UIDNEXT) auf // — kombiniert bewusst beide fertigen, unveränderten Pakete statt eines // davon zu erweitern (kein Umbau angrenzender Bereiche). // // Konzept aus archivmail als Ausgangspunkt genommen (UID-Sync, // Delta-Import, siehe repos-analyse-mail-reuse.md), Testabdeckung von // Grund auf neu (archivmails Import-Pfade waren praktisch ungetestet). // Bekannten Fehler vermieden (archivmail: UIDVALIDITY=0 bricht Resync // bei nicht-konformen Servern): Store.RunOnce erkennt jede Änderung der // UIDVALIDITY explizit und löst einen vollständigen Resync aus, statt // eine UIDVALIDITY=0 unbesehen zu übernehmen. package imapimport import ( "context" _ "embed" "fmt" "github.com/jackc/pgx/v5/pgxpool" ) //go:embed migrations/0001_mail_import_state.sql var schemaMigration string const defaultIntervalSeconds = 300 // State ist der persistierte Sync-Zustand eines Postfachs — übersteht // Dienst-Neustarts (Akzeptanzkriterium 3), da ausschließlich in Postgres // gehalten, nie im Prozessspeicher. type State struct { TenantSlug string MailboxName string LastUIDValidity uint64 LastSyncedUID uint32 IntervalSeconds int } // Store persistiert den Sync-Zustand je Mandant und Postfach. type Store struct { pool *pgxpool.Pool } func NewStore(pool *pgxpool.Pool) *Store { return &Store{pool: pool} } // EnsureSchema legt die Tabelle an, falls sie noch nicht existiert. func (s *Store) EnsureSchema(ctx context.Context) error { if _, err := s.pool.Exec(ctx, schemaMigration); err != nil { return fmt.Errorf("imapimport: schema anlegen: %w", err) } return nil } // GetOrCreate liefert den Sync-Zustand eines Postfachs, legt ihn bei // erstem Zugriff mit dem Standardintervall neu an. func (s *Store) GetOrCreate(ctx context.Context, tenantSlug, mailboxName string) (State, error) { if _, err := s.pool.Exec(ctx, ` INSERT INTO mail_import_state (tenant_slug, mailbox_name, interval_seconds) VALUES ($1, $2, $3) ON CONFLICT (tenant_slug, mailbox_name) DO NOTHING `, tenantSlug, mailboxName, defaultIntervalSeconds); err != nil { return State{}, fmt.Errorf("imapimport: sync-zustand anlegen: %w", err) } return s.Get(ctx, tenantSlug, mailboxName) } // Get liest den aktuellen Sync-Zustand. func (s *Store) Get(ctx context.Context, tenantSlug, mailboxName string) (State, error) { var st State st.TenantSlug = tenantSlug st.MailboxName = mailboxName err := s.pool.QueryRow(ctx, ` SELECT last_uidvalidity, last_synced_uid, interval_seconds FROM mail_import_state WHERE tenant_slug = $1 AND mailbox_name = $2 `, tenantSlug, mailboxName).Scan(&st.LastUIDValidity, &st.LastSyncedUID, &st.IntervalSeconds) if err != nil { return State{}, fmt.Errorf("imapimport: sync-zustand lesen: %w", err) } return st, nil } // SetInterval konfiguriert das Abrufintervall je Postfach // (Akzeptanzkriterium 3), persistiert und damit neustartfest. func (s *Store) SetInterval(ctx context.Context, tenantSlug, mailboxName string, seconds int) error { if _, err := s.pool.Exec(ctx, ` UPDATE mail_import_state SET interval_seconds = $3, updated_at = now() WHERE tenant_slug = $1 AND mailbox_name = $2 `, tenantSlug, mailboxName, seconds); err != nil { return fmt.Errorf("imapimport: intervall setzen: %w", err) } return nil } // advance persistiert den erreichten Fortschritt NACH jeder erfolgreich // verarbeiteten Nachricht (nicht erst am Ende des Laufs) — Grundlage für // Akzeptanzkriterium 3 / Pflichtprüfung 2: ein Dienst-Neustart mitten im // Abgleich verliert höchstens die aktuell laufende Verarbeitung, nie den // bereits erreichten Fortschritt, und importiert nichts doppelt. func (s *Store) advance(ctx context.Context, tenantSlug, mailboxName string, uidvalidity uint64, syncedUID uint32) error { if _, err := s.pool.Exec(ctx, ` UPDATE mail_import_state SET last_uidvalidity = $3, last_synced_uid = $4, updated_at = now() WHERE tenant_slug = $1 AND mailbox_name = $2 `, tenantSlug, mailboxName, uidvalidity, syncedUID); err != nil { return fmt.Errorf("imapimport: fortschritt persistieren: %w", err) } return nil }