// Package reconciliation implements the daily completeness reconciliation // report (PROJ-52). It counts newly archived mails per source (SMTP journal, // IMAP account, POP3 account, bulk import) and per day, persists the counts in // the reconciliation_reports table, and flags significant drops against the // trailing 7-day average via the audit log. // // The reconciliation is deliberately a read-only observer of the emails table // plus the IMAP UID-tracking state (imap_folder_state, PROJ-45). It never // mutates archived mail content and issues no additional IMAP logins — the // IMAP soll/ist comparison reuses the UID high-water marks already persisted by // the sync scheduler. package reconciliation import ( "context" "fmt" "log/slog" "time" "github.com/jackc/pgx/v5/pgxpool" "archivmail/internal/audit" ) // Store owns the reconciliation_reports table and the reconciliation logic. // It uses its own connection pool so the CLI cron command and the daemon can // both operate it independently. type Store struct { pool *pgxpool.Pool logger *slog.Logger audlog *audit.Logger // optional; when nil, anomalies are only logged } // Report is a single persisted reconciliation row for one day and one source. type Report struct { Date time.Time `json:"date"` TenantID *int64 `json:"tenant_id"` SourceType string `json:"source_type"` SourceID *int64 `json:"source_id"` ExpectedCount *int64 `json:"expected_count"` ArchivedCount int64 `json:"archived_count"` Delta *int64 `json:"delta"` } // New connects to PostgreSQL and initialises the reconciliation schema. func New(dsn string, logger *slog.Logger) (*Store, error) { pool, err := pgxpool.New(context.Background(), dsn) if err != nil { return nil, fmt.Errorf("reconciliation: connect: %w", err) } s := &Store{pool: pool, logger: logger} if err := s.initSchema(context.Background()); err != nil { pool.Close() return nil, fmt.Errorf("reconciliation: init schema: %w", err) } return s, nil } // SetAuditLogger wires an audit.Logger so anomalies are persisted as // tenant-visible `reconciliation_anomaly` audit entries. Optional. func (s *Store) SetAuditLogger(a *audit.Logger) { s.audlog = a } // Close releases the connection pool. func (s *Store) Close() { if s.pool != nil { s.pool.Close() } } // initSchema creates the reconciliation_reports table and its indexes. // Idempotent and safe on existing databases (CREATE ... IF NOT EXISTS). func (s *Store) initSchema(ctx context.Context) error { if _, err := s.pool.Exec(ctx, ` CREATE TABLE IF NOT EXISTS reconciliation_reports ( id BIGSERIAL PRIMARY KEY, date DATE NOT NULL, tenant_id BIGINT, source_type TEXT NOT NULL, source_id BIGINT, expected_count BIGINT, archived_count BIGINT NOT NULL DEFAULT 0, delta BIGINT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); `); err != nil { return err } // Upsert key. tenant_id and source_id are nullable and PostgreSQL treats // NULLs as distinct in a plain UNIQUE index, which would allow duplicate // rows for the (tenant-less / smtp) buckets. A COALESCE-based expression // index gives a single deterministic key per (date, tenant, source_type, // source_id). -1 is a safe sentinel because real tenant/account IDs are // positive BIGSERIALs. if _, err := s.pool.Exec(ctx, ` CREATE UNIQUE INDEX IF NOT EXISTS idx_recon_reports_key ON reconciliation_reports (date, COALESCE(tenant_id, -1), source_type, COALESCE(source_id, -1)); `); err != nil { return err } // Lookup index for the dashboard / CSV queries (per AC). if _, err := s.pool.Exec(ctx, ` CREATE INDEX IF NOT EXISTS idx_recon_reports_lookup ON reconciliation_reports (tenant_id, date, source_type); `); err != nil { return err } return nil } // SourceKey returns the canonical source identifier used in the API / CSV: // "smtp", "import", "imap:", "pop3:". func SourceKey(sourceType string, sourceID *int64) string { if sourceID != nil && (sourceType == "imap" || sourceType == "pop3") { return fmt.Sprintf("%s:%d", sourceType, *sourceID) } return sourceType }