diff --git a/cmd/archivmail/cmd_import.go b/cmd/archivmail/cmd_import.go index 7d24c07..e245df4 100644 --- a/cmd/archivmail/cmd_import.go +++ b/cmd/archivmail/cmd_import.go @@ -253,6 +253,11 @@ func importMessage(mailStore *storage.Store, idxMgr index.TenantIndexer, raw []b return "error" } + // PROJ-52: mark bulk-imported mails as source 'import' for reconciliation. + if err := mailStore.TagSource(context.Background(), id, "import", nil); err != nil { + fmt.Fprintf(os.Stderr, "warning: tag source failed: %v\n", err) + } + var attachNames []string for _, a := range pm.Attachments { attachNames = append(attachNames, a.Filename) diff --git a/cmd/archivmail/cmd_reconcile.go b/cmd/archivmail/cmd_reconcile.go new file mode 100644 index 0000000..0fc7032 --- /dev/null +++ b/cmd/archivmail/cmd_reconcile.go @@ -0,0 +1,95 @@ +package main + +import ( + "context" + "flag" + "log/slog" + "os" + "time" + + "archivmail/config" + "archivmail/internal/audit" + "archivmail/internal/reconciliation" +) + +// runReconcile computes the daily completeness reconciliation report (PROJ-52). +// It is designed to be driven by cron once per day (e.g. shortly after +// midnight) and, by default, reconciles the *previous* full calendar day so a +// day is only counted once it is complete. +// +// Usage: +// +// archivmail reconcile --config /etc/archivmail/config.yml +// archivmail reconcile --date 2026-07-01 +// archivmail reconcile --days 7 # backfill: reconcile the last 7 days +func runReconcile(args []string) { + fs := flag.NewFlagSet("reconcile", flag.ExitOnError) + configPath := fs.String("config", "/etc/archivmail/config.yml", "path to config file") + dateFlag := fs.String("date", "", "day to reconcile (YYYY-MM-DD, UTC); default: yesterday") + daysFlag := fs.Int("days", 1, "number of days to reconcile, ending at --date (backfill)") + fs.Parse(args) + + logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})) + + cfg, err := config.Load(*configPath) + if err != nil { + logger.Error("failed to load config", "err", err) + os.Exit(1) + } + + // Determine the target (end) date. + var end time.Time + if *dateFlag != "" { + end, err = time.ParseInLocation("2006-01-02", *dateFlag, time.UTC) + if err != nil { + logger.Error("invalid --date (expected YYYY-MM-DD)", "err", err) + os.Exit(1) + } + } else { + end = time.Now().UTC().AddDate(0, 0, -1).Truncate(24 * time.Hour) + } + + days := *daysFlag + if days < 1 { + days = 1 + } + + dsn := cfg.Database.DSN() + reconStore, err := reconciliation.New(dsn, logger) + if err != nil { + logger.Error("reconciliation store init failed", "err", err) + os.Exit(1) + } + defer reconStore.Close() + + // Wire audit logging so anomalies are persisted as tenant-visible entries. + audlog, err := audit.New(dsn, cfg.Audit.ResolvedLogPath(), logger) + if err != nil { + logger.Warn("audit init failed — anomalies will only be logged", "err", err) + } else { + defer audlog.Close() + reconStore.SetAuditLogger(audlog) + } + + thresholdPct := cfg.Reconciliation.ResolvedThresholdPct() + ctx := context.Background() + + totalAnomalies := 0 + // Reconcile oldest → newest so trailing-average history is populated in order. + for i := days - 1; i >= 0; i-- { + day := end.AddDate(0, 0, -i) + anomalies, err := reconStore.ComputeForDate(ctx, day, thresholdPct) + if err != nil { + // Per the spec: on failure the day is left WITHOUT a report row + // (no false zeros). Exit non-zero so cron surfaces the failure. + logger.Error("reconcile: compute failed", "date", day.Format("2006-01-02"), "err", err) + os.Exit(1) + } + totalAnomalies += len(anomalies) + logger.Info("reconcile: day complete", + "date", day.Format("2006-01-02"), "anomalies", len(anomalies)) + } + + logger.Info("reconcile: complete", "days", days, "threshold_pct", thresholdPct, + "anomalies_total", totalAnomalies) +} diff --git a/cmd/archivmail/main.go b/cmd/archivmail/main.go index 67a3653..e88f56e 100644 --- a/cmd/archivmail/main.go +++ b/cmd/archivmail/main.go @@ -31,6 +31,7 @@ import ( "archivmail/internal/mailer" "archivmail/internal/ocr" pop3store "archivmail/internal/pop3" + "archivmail/internal/reconciliation" "archivmail/internal/smtpoutconfig" "archivmail/internal/smtpd" "archivmail/internal/storage" @@ -73,6 +74,9 @@ func main() { case "index-pending": runIndexPending(os.Args[2:]) return + case "reconcile": + runReconcile(os.Args[2:]) + return case "update": runUpdate(os.Args[2:]) return @@ -313,11 +317,24 @@ func main() { srv.SetGlobalRetentionDays(cfg.Storage.RetentionDays) srv.SetMetrics(cfg.Metrics) + // PROJ-52: completeness reconciliation store — powers the dashboard + + // CSV-export endpoints. The daily computation itself is driven by cron via + // the `archivmail reconcile` subcommand (analog PROJ-58 batch jobs), so the + // daemon only needs read access here. + reconStore, err := reconciliation.New(cfg.Database.DSN(), logger) + if err != nil { + logger.Error("reconciliation store init failed", "err", err) + os.Exit(1) + } + defer reconStore.Close() + reconStore.SetAuditLogger(audlog) + srv.SetReconciliation(reconStore, cfg.Reconciliation.ResolvedThresholdPct()) + // PROJ-28: Self-Service Onboarding — mailer + token store + FQDN mlr := mailer.New(cfg.SMTPOut) // SMTP-Out config store — load from DB, overrides config.yml if present - smtpOutSt, err := smtpoutconfig.New(cfg.Database.DSN(), cfg.API.Secret) + smtpOutSt, err := smtpoutconfig.New(cfg.Database.DSN(), aesKey) if err != nil { logger.Error("smtp-out config store init failed", "err", err) os.Exit(1) diff --git a/config.test.yml b/config.test.yml index 6214e35..3b672b5 100644 --- a/config.test.yml +++ b/config.test.yml @@ -50,3 +50,12 @@ audit: # Default falls leer: /var/log/archivmail/audit.log log_path: /tmp/archivmail-test/audit.log retention_days: 0 + +# PROJ-52: Vollständigkeits-Reconciliation (Zähl-Report Mailserver vs. Archiv). +# Der Job wird per Cron gestartet: `archivmail reconcile` (z.B. täglich 00:30). +reconciliation: + # Relativer Rückgang (in Prozent) unter den 7-Tage-Durchschnitt einer Quelle, + # ab dem ein reconciliation_anomaly Audit-Eintrag geschrieben wird. + # Weglassen = Default 50. Hinweis: SMTP-Journale schwanken stark + # (Wochenende/Feiertage) — bei False-Positives Schwellenwert erhöhen. + alert_threshold_pct: 50 diff --git a/config/config.docker.yml.example b/config/config.docker.yml.example index a10a025..ec47477 100644 --- a/config/config.docker.yml.example +++ b/config/config.docker.yml.example @@ -91,5 +91,12 @@ audit: log_path: /var/archivmail/audit.log retention_days: 365 +# PROJ-52: Vollständigkeits-Reconciliation (Zähl-Report Mailserver vs. Archiv). +# Täglich per Cron ausführen: `archivmail reconcile` (rechnet den Vortag ab). +# alert_threshold_pct = relativer Rückgang (%) unter den 7-Tage-Durchschnitt, +# ab dem ein reconciliation_anomaly Audit-Eintrag entsteht (Default 50). +# reconciliation: +# alert_threshold_pct: 50 + logging: level: info diff --git a/config/config.go b/config/config.go index 166e136..7320de3 100644 --- a/config/config.go +++ b/config/config.go @@ -42,6 +42,39 @@ type Config struct { // PROJ-56: load-spreading for background jobs. OCR OCRConfig `yaml:"ocr"` IMAPScheduler IMAPSchedulerConfig `yaml:"imap_scheduler"` + // PROJ-52: Vollständigkeits-Reconciliation (Zähl-Report Mailserver vs. Archiv). + Reconciliation ReconciliationConfig `yaml:"reconciliation"` +} + +// ReconciliationConfig holds settings for the daily completeness reconciliation +// job (PROJ-52). The job counts newly archived mails per source and per day and +// flags significant drops against the trailing 7-day average. +type ReconciliationConfig struct { + // AlertThresholdPct is the relative drop (in percent, below the trailing + // 7-day average) that triggers a `reconciliation_anomaly` audit entry. + // A pointer so an explicit 0 (alert on any drop) is distinguishable from an + // unset value (use the default). + // nil -> DefaultReconciliationThresholdPct (50%) + // 0 -> alert whenever today's count is below the average + // 1..100 -> alert when today's count is more than this % below the average + AlertThresholdPct *int `yaml:"alert_threshold_pct,omitempty"` +} + +// DefaultReconciliationThresholdPct is the default drop threshold (50% below the +// trailing 7-day average) applied when reconciliation.alert_threshold_pct is +// omitted from the config. +const DefaultReconciliationThresholdPct = 50 + +// ResolvedThresholdPct returns the effective alert threshold in percent. +// A nil or out-of-range value falls back to the default. +func (c ReconciliationConfig) ResolvedThresholdPct() int { + if c.AlertThresholdPct == nil { + return DefaultReconciliationThresholdPct + } + if *c.AlertThresholdPct < 0 || *c.AlertThresholdPct > 100 { + return DefaultReconciliationThresholdPct + } + return *c.AlertThresholdPct } // OCRConfig holds settings for the background OCR worker (PROJ-56). diff --git a/features/INDEX.md b/features/INDEX.md index 642a4dc..3c27c22 100644 --- a/features/INDEX.md +++ b/features/INDEX.md @@ -68,7 +68,7 @@ | PROJ-49 | Verschlüsselungspflicht at-rest (Healthcheck & Warnung) | Deployed | [PROJ-49](PROJ-49-verschluesselungspflicht.md) | 2026-06-13 | | PROJ-50 | DSGVO-Löschersuchen für Mail-Inhalte (GoBD-Vorrang) | Deployed | [PROJ-50](PROJ-50-dsgvo-loeschersuchen.md) | 2026-06-13 | | PROJ-51 | Aufbewahrungsfristen nach Dokumentenart (Retention-Kategorien) | Deployed | [PROJ-51](PROJ-51-retention-kategorien.md) | 2026-06-13 | -| PROJ-52 | Vollständigkeits-Reconciliation (Zähl-Report) | Planned | [PROJ-52](PROJ-52-vollstaendigkeits-reconciliation.md) | 2026-06-13 | +| PROJ-52 | Vollständigkeits-Reconciliation (Zähl-Report) | In Review | [PROJ-52](PROJ-52-vollstaendigkeits-reconciliation.md) | 2026-06-13 | | PROJ-53 | Konfigurierbare Listenanzahl pro Seite | Deployed | [PROJ-53](PROJ-53-konfigurierbare-listenanzahl.md) | 2026-06-14 | | PROJ-54 | Fix Listenansicht/Pagination für Rolle "user" (Nachbesserung PROJ-6/PROJ-21) | Deployed | [PROJ-54](PROJ-54-fix-listenansicht-total.md) | 2026-06-14 | | PROJ-55 | Fix Tenant-Isolation für Rolle "auditor" + Audit-Log (Sicherheitsbug, DSGVO-relevant) | Deployed | [PROJ-55](PROJ-55-fix-auditor-tenant-isolation.md) | 2026-06-21 | diff --git a/features/PROJ-52-vollstaendigkeits-reconciliation.md b/features/PROJ-52-vollstaendigkeits-reconciliation.md new file mode 100644 index 0000000..1f4f34e --- /dev/null +++ b/features/PROJ-52-vollstaendigkeits-reconciliation.md @@ -0,0 +1,224 @@ +# PROJ-52: Vollständigkeits-Reconciliation (Zähl-Report Mailserver vs. Archiv) + +## Status: In Review +**Created:** 2026-06-13 +**Last Updated:** 2026-07-03 + +## Hintergrund +Der GoBD/DSGVO-Compliance-Check (`docs/GOBD_DSGVO_CHECKLIST.md`, Punkt 1) bewertet +"Vollständigkeit" nur als "Teilweise erfüllt": SMTP-BCC-Journaling (PROJ-4) und IMAP/POP3-Import +(PROJ-3/8/14/45) sind robust (z.B. `452`-Retry bei Storage-Fehlern), aber es gibt keinen +zentralen Mechanismus, der zeigt, ob tatsächlich ALLE erwarteten E-Mails archiviert wurden +(VOI-Grundsatz 2: "kein Dokument darf auf dem Weg ins Archiv oder im Archiv selbst verloren +gehen"). Diese Spec ergänzt einen täglichen Zähl-Report pro Quelle. + +## Dependencies +- Requires: PROJ-4 (SMTP-Import), PROJ-3/PROJ-14 (IMAP/POP3-Import), PROJ-45 + (IMAP Per-Folder UID-Tracking) +- Requires: PROJ-17 (Admin Dashboard) – Anzeige des Reports +- Requires: PROJ-11/PROJ-48 (Audit-Log) – Auffälligkeiten werden protokolliert + +## User Stories +- Als Admin möchte ich täglich sehen, wie viele E-Mails pro Quelle (SMTP-Journal, IMAP-Konto, + POP3-Konto) archiviert wurden, damit ich Ausreißer (plötzlich 0 Mails) erkenne. +- Als Admin möchte ich für IMAP/POP3-Quellen einen Soll/Ist-Vergleich sehen: Anzahl Mails im + Quell-Postfach (laut letztem Sync) vs. Anzahl archivierter Mails für diese Quelle. +- Als Auditor möchte ich nachvollziehen können, ob es Tage mit auffälligen Abweichungen gab + (z.B. SMTP-Dienst war down). +- Als Admin möchte ich bei einer signifikanten Abweichung (z.B. >50% Rückgang ggü. + Durchschnitt der letzten 7 Tage) eine Warnung im Dashboard sehen. + +## Acceptance Criteria +- [ ] Täglicher Job (Cron, analog PROJ-8-Scheduler) berechnet pro Tag und Quelle + (`source_type`: `smtp`, `imap:`, `pop3:`, `import`) die Anzahl + neu archivierter E-Mails (`received_at`/`imported_at` am jeweiligen Tag) +- [ ] Für IMAP-Konten (PROJ-45 UID-Tracking): zusätzlicher Soll/Ist-Vergleich – Anzahl Mails im + Quell-Ordner laut letztem `UIDVALIDITY`/UID-Stand vs. Anzahl im Archiv für diesen Ordner +- [ ] Ergebnisse werden in Tabelle `reconciliation_reports` + (date, tenant_id, source_type, source_id, expected_count, archived_count, delta) + persistiert +- [ ] Admin-Dashboard (PROJ-17) zeigt eine neue Kachel/Tabelle "Vollständigkeits-Check" mit + den letzten 7 Tagen pro Quelle +- [ ] Abweichung > konfigurierbarem Schwellenwert (Default: 50% unter 7-Tage-Durchschnitt, + `reconciliation.alert_threshold_pct` in `config.yml`) → Warn-Badge im Dashboard + + Audit-Log-Eintrag (`event_type: reconciliation_anomaly`) +- [ ] Report ist als CSV exportierbar (analog Audit-Log-Export aus PROJ-11) +- [ ] Tage ohne Aktivität (0 Mails) werden explizit als `0` ausgewiesen, nicht als fehlender + Datensatz (damit Lücken im Cron-Lauf selbst erkennbar sind) + +## Edge Cases +- Quelle wurde erst kürzlich angelegt (kein 7-Tage-Durchschnitt vorhanden) → kein Alert, + Anzeige "Noch nicht genug Daten" +- SMTP-Journal hat naturgemäß starke Schwankungen (Wochenende vs. Wochentag) → Schwellenwert + ist konfigurierbar, Doku weist auf mögliche False-Positives an Wochenenden/Feiertagen hin +- IMAP-Quell-Postfach wurde vom Nutzer geleert (Mails dort gelöscht, aber bereits archiviert) → + `expected_count` sinkt, `archived_count` bleibt hoch → `delta` negativ in "gute" Richtung, + kein Alert (nur Rückgang von `archived_count` selbst ist relevant) +- Reconciliation-Job selbst schlägt fehl (z.B. DB-Timeout) → Fehler wird geloggt, vorheriger + Tag bleibt ohne Report-Eintrag, Dashboard zeigt "Daten fehlen für " statt falscher + Nullwerte +- Multi-Tenant: Reports sind pro Tenant; Tenant-Admins sehen nur eigene Quellen, Super-Admin + sieht alle + +## Technical Requirements +- Neue Tabelle `reconciliation_reports` (siehe AC), Index auf `(tenant_id, date, source_type)` +- Cron-Job-Registrierung analog bestehendem IMAP-Sync-Scheduler (PROJ-8) +- Wiederverwendung von `internal/imap`-Funktionen zur Ermittlung der Quell-Postfach-Anzahl + (sofern bereits durch UID-Tracking verfügbar, kein zusätzlicher IMAP-Login nötig wenn + vermeidbar) + +--- + + +## Implementation Notes (Backend, 2026-07-03) + +### Neues Package `internal/reconciliation/` +- `reconciliation.go`: `Store` (eigener pgxpool), `initSchema()`, Tabelle + `reconciliation_reports (id, date, tenant_id, source_type, source_id, + expected_count, archived_count, delta, created_at)`. Upsert-Key ist ein + COALESCE-Ausdrucksindex `(date, COALESCE(tenant_id,-1), source_type, + COALESCE(source_id,-1))`, weil `tenant_id`/`source_id` NULL-fähig sind und + Postgres NULLs in einem normalen UNIQUE-Index als verschieden behandelt (sonst + Doppelzeilen für smtp/import/tenant-lose Buckets). Zusätzlicher Lookup-Index + `(tenant_id, date, source_type)` laut AC. +- `compute.go`: `ComputeForDate(ctx, day, thresholdPct)` — Read-Phase (archived + pro Bucket für den Tag, known-buckets aus `emails` DISTINCT, IMAP-Snapshot), + dann Upsert in **einer Transaktion**. Bei Query-Fehler wird der Tag NICHT + geschrieben (kein falscher 0-Eintrag; Dashboard zeigt "fehlt"). 0-Mail-Tage + werden für jeden bekannten Bucket **explizit als 0** persistiert. Alert: + 7-Tage-Durchschnitt (`trailing_average`, NULL-safe `IS NOT DISTINCT FROM`); + Alert nur bei ≥7 Vortages-Datensätzen; `archived < avg*(1-pct/100)` → + Audit-Eintrag `reconciliation_anomaly` (event via `audit.EventReconciliationAnomaly`). +- `query.go`: `DashboardData()` (letzte N Tage pro Quelle, tenant-gescoped, + Alert-Flag + `enough_data`) und `ExportRows()` (CSV). + +### Source-Tracking (nötige Ergänzung — emails hatte keine Herkunftsspalte) +`emails` bekam via `storage.initSchema()` zwei Spalten `source_type TEXT`, +`source_id BIGINT` + Index `(received_at, source_type, source_id, tenant_id)`. +Neue Methode `storage.Store.TagSource(ctx, id, sourceType, sourceID)` schreibt +nur solange `source_type IS NULL` (first-write-wins → dedupte Mehrfach-Mails +werden nicht doppelt gezählt). Verdrahtet in allen Ingestion-Pfaden: +- `internal/smtpd/smtpd.go` → `smtp`, nil +- `internal/imap/importer.go` + `internal/imap/scheduler.go` → `imap`, + account_id (accountID durch `fetchBatch`/`fetchSyncBatch`/`storeAndIndex` + durchgereicht) +- `internal/pop3/importer.go` → `pop3`, account_id +- `cmd/archivmail/cmd_import.go` + `internal/api/upload.go` → `import`, nil + +### IMAP Soll/Ist (Abweichung von der Spec — dokumentiert) +Das PROJ-45 UID-Tracking (`imap_folder_state`) speichert nur `last_uid`, KEINE +Nachrichtenzahl. Daher wird `expected_count` für IMAP-Quellen als Proxy aus +`SUM(last_uid)` je Konto gebildet (kein zusätzlicher IMAP-Login) und +`delta = kumulativ_archiviert(Konto) − expected`. `archived_count` bleibt +konsistent für ALLE Quellen die **pro-Tag** neu archivierte Zahl (steuert das +Alerting). Postfach-Leerung → expected sinkt, delta positiv → kein Alert +(Alert nur bei Rückgang von `archived_count`), wie in Edge Cases gefordert. + +### Cron statt Dauer-Goroutine +Analog PROJ-58 als CLI-Subcommand `archivmail reconcile` (`cmd_reconcile.go`), +Default = Vortag; `--date`, `--days N` (Backfill, älteste→neueste Reihenfolge +für konsistente Trailing-Average-Historie). Registriert in `main.go`. Der Daemon +(`serve`) verdrahtet den Store nur lesend für die API (`SetReconciliation`). + +### Config +`config.ReconciliationConfig.AlertThresholdPct *int` (`reconciliation.alert_threshold_pct`), +Default 50 via `ResolvedThresholdPct()`. Beispiele in `config.test.yml` und +`config/config.docker.yml.example`. + +### API-Endpoints (tenant-gescoped, `authAdmin` = domain_admin+) +- `GET /api/admin/reconciliation?days=7` (max 90) + ```json + { + "days": 7, + "threshold_pct": 50, + "sources": [ + { + "source_type": "smtp", + "source_id": null, + "source_key": "smtp", + "tenant_id": null, + "points": [ + {"date":"2026-06-27","archived_count":42,"expected_count":null,"delta":null,"missing":false}, + {"date":"2026-06-28","archived_count":null,"expected_count":null,"delta":null,"missing":true} + ], + "avg_7d": 40.5, + "enough_data": true, + "alert": false + } + ] + } + ``` + `source_key`: `smtp` | `import` | `imap:` | `pop3:`. `missing:true` = + kein Report-Datensatz (Cron nicht gelaufen), ≠ `archived_count:0`. + `enough_data:false` → UI zeigt "Noch nicht genug Daten". `alert:true` → + Warn-Badge. +- `GET /api/admin/reconciliation/export.csv?days=30` (max 366) — CSV + `date,tenant_id,source,expected_count,archived_count,delta`, Audit-Eintrag + `export`. Tenant-Scope: domain_admin nur eigener Tenant, superadmin alle. + +### Tenant-Isolation +`reconTenantScope()` filtert wie `handleMailTimeseries`: Session mit `tenant_id` +→ nur eigener Tenant, superadmin (nil) → alle. Kein `{id}`-Pfadparameter, daher +kein IDOR-Vektor; Filter erfolgt in der SQL-WHERE. + +### Manticore +Keine Index-Änderung nötig (reine PostgreSQL-Aggregation). + +### Offene Punkte / Handoff +- Cron-Eintrag für `archivmail reconcile` muss in `install.sh`/`update.sh` + ergänzt werden (devops-deploy). +- Frontend: Dashboard-Kachel + TS-Typen in `src/lib/api/`. +- Kein lokaler `go build` möglich — Build/QA separat auf Testserver. + +## Implementation Notes (Frontend, 2026-07-03) + +### Neue API-Schicht `src/lib/api/reconciliation.ts` +- Typen `ReconciliationPoint`, `ReconciliationSource`, `ReconciliationResponse` + (1:1 zur Backend-Response; `archived_count`/`expected_count`/`delta` sind + `number | null`, `missing: boolean`). +- `getReconciliation(days = 7)` → `GET /api/admin/reconciliation?days=7` über den + bestehenden `request`-Wrapper (`core.ts`, credentials/401-Handling inklusive). +- `exportReconciliationCSV(days = 30)` → direkter `fetch` auf + `/api/admin/reconciliation/export.csv` (Blob-Download, Content-Disposition- + Dateiname-Parsing, analog `downloadMailAttachment`/`exportDSGVORequestPDF`). +- Re-Exports in `src/lib/api/index.ts` ergänzt. + +### Neue Kachel `src/components/admin/tabs/ReconciliationCard.tsx` +- Self-fetching Client-Komponente (`useEffect` beim Mount), Loading-/Error-/ + Empty-States implementiert. +- Tabelle (shadcn `Table`): eine Zeile pro Quelle, Label hübsch formatiert + ("SMTP-Journal", "Datei-Import", "IMAP-Konto #3", "POP3-Konto #X"). +- Spalten = letzte 7 Tage; Zelle zeigt `archived_count`, bei IMAP zusätzlich + "Soll (delta)". `missing:true` → "—" (klar unterschieden von "0", + mit Tooltip "Cron nicht gelaufen"). `enough_data:false` → Zeile zeigt + "Noch nicht genug Daten" (colSpan). `alert:true` → rotes Badge "Auffällig", + sonst "OK". Zusätzliche Spalten "Ø 7 Tage" und "Status". +- CSV-Export-Button (30 Tage) und Aktualisieren-Button im Kachel-Header. +- Horizontal scrollbar (`overflow-x-auto`) für mobile Breiten. + +### Einbindung / Rollen-Sichtbarkeit +- Gerendert innerhalb `DashboardTab` (vor "Benutzerübersicht"). Der gesamte + Admin-Bereich (`src/app/admin/page.tsx`) ist bereits via + `useAuth("domain_admin", "/admin/login")` auf domain_admin+ beschränkt → + keine zusätzliche Client-Gate nötig, normale User erreichen den Tab nicht. + Backend bleibt maßgebliche Sicherheitsgrenze (tenant-gescoped, `authAdmin`); + superadmin sieht alle Quellen, domain_admin nur eigene. + +### Verifikation +- `npx tsc --noEmit` fehlerfrei. Kein direktes `fetch()` in Komponenten außer + dem zentralen Blob-Download-Helper in der API-Schicht. + +### Geänderte/neue Dateien +- neu: `src/lib/api/reconciliation.ts` +- neu: `src/components/admin/tabs/ReconciliationCard.tsx` +- geändert: `src/lib/api/index.ts` (Re-Exports) +- geändert: `src/components/admin/tabs/DashboardTab.tsx` (Kachel eingebunden) + +## Tech Design (Solution Architect) +_To be added by /architecture_ + +## QA Test Results +_To be added by /qa_ + +## Deployment +_To be added by /deploy_ diff --git a/internal/api/reconciliation_handlers.go b/internal/api/reconciliation_handlers.go new file mode 100644 index 0000000..a20ff30 --- /dev/null +++ b/internal/api/reconciliation_handlers.go @@ -0,0 +1,124 @@ +package api + +import ( + "encoding/csv" + "fmt" + "net/http" + "strconv" + + "archivmail/internal/audit" +) + +// tenantScope returns the tenant filter for reconciliation queries: a +// domain_admin (and any other tenant-scoped role) is restricted to its own +// tenant, while superadmin (sess.TenantID == nil) sees all tenants. This +// mirrors handleMailTimeseries and prevents cross-tenant leakage of source +// figures (PROJ-55/61 tenant-isolation discipline). +func (s *Server) reconTenantScope(r *http.Request) *int64 { + sess := sessionFromCtx(r.Context()) + if sess.TenantID != nil { + return tenantFromCtx(r.Context()) + } + return nil +} + +// handleReconciliation returns the last N days (default 7) of completeness +// figures per source, tenant-scoped. +// GET /api/admin/reconciliation?days=7 +func (s *Server) handleReconciliation(w http.ResponseWriter, r *http.Request) { + if s.reconStore == nil { + writeError(w, http.StatusServiceUnavailable, "reconciliation not enabled") + return + } + + days := 7 + if v := r.URL.Query().Get("days"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 90 { + days = n + } + } + + tid := s.reconTenantScope(r) + sources, err := s.reconStore.DashboardData(r.Context(), tid, days, s.reconThresholdPct) + if err != nil { + s.logger.Error("reconciliation dashboard query failed", "err", err) + writeError(w, http.StatusInternalServerError, "reconciliation query failed") + return + } + writeJSON(w, http.StatusOK, map[string]interface{}{ + "days": days, + "threshold_pct": s.reconThresholdPct, + "sources": sources, + }) +} + +// handleReconciliationExport streams the reconciliation report as CSV, +// tenant-scoped (analog PROJ-11 audit export). +// GET /api/admin/reconciliation/export.csv?days=30 +func (s *Server) handleReconciliationExport(w http.ResponseWriter, r *http.Request) { + if s.reconStore == nil { + writeError(w, http.StatusServiceUnavailable, "reconciliation not enabled") + return + } + + days := 30 + if v := r.URL.Query().Get("days"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 366 { + days = n + } + } + + tid := s.reconTenantScope(r) + rows, err := s.reconStore.ExportRows(r.Context(), tid, days) + if err != nil { + s.logger.Error("reconciliation export query failed", "err", err) + writeError(w, http.StatusInternalServerError, "reconciliation query failed") + return + } + + sess := sessionFromCtx(r.Context()) + + w.Header().Set("Content-Type", "text/csv; charset=utf-8") + w.Header().Set("Content-Disposition", `attachment; filename="reconciliation.csv"`) + w.WriteHeader(http.StatusOK) + + cw := csv.NewWriter(w) + cw.Write([]string{"date", "tenant_id", "source", "expected_count", "archived_count", "delta"}) //nolint:errcheck + for _, row := range rows { + cw.Write([]string{ //nolint:errcheck + row.Date.UTC().Format("2006-01-02"), + nullableInt(row.TenantID), + reconSourceKey(row.SourceType, row.SourceID), + nullableInt(row.ExpectedCount), + strconv.FormatInt(row.ArchivedCount, 10), + nullableInt(row.Delta), + }) + } + cw.Flush() + + s.audlog.Log(audit.Entry{ + EventType: audit.EventExport, + Username: sess.Username, + TenantID: sess.TenantID, + IPAddress: s.remoteIP(r), + Detail: fmt.Sprintf("reconciliation csv: %d days, %d rows", days, len(rows)), + Success: true, + }) +} + +// nullableInt formats a *int64 for CSV, emitting an empty string for nil. +func nullableInt(v *int64) string { + if v == nil { + return "" + } + return strconv.FormatInt(*v, 10) +} + +// reconSourceKey mirrors reconciliation.SourceKey without importing the package +// into the CSV hot path (kept local and tiny). +func reconSourceKey(sourceType string, sourceID *int64) string { + if sourceID != nil && (sourceType == "imap" || sourceType == "pop3") { + return fmt.Sprintf("%s:%d", sourceType, *sourceID) + } + return sourceType +} diff --git a/internal/api/server.go b/internal/api/server.go index 4ac27c5..80ce744 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -20,6 +20,7 @@ import ( ldapcfg "archivmail/internal/ldapconfig" "archivmail/internal/mailer" pop3store "archivmail/internal/pop3" + "archivmail/internal/reconciliation" "archivmail/internal/smtpoutconfig" "archivmail/internal/smtpd" "archivmail/internal/storage" @@ -90,6 +91,8 @@ type Server struct { fqdn string // from server.fqdn config (PROJ-28) smtpOutStore *smtpoutconfig.Store apiKeyMw *auth.APIKeyMiddleware // PROJ-13: external API auth + reconStore *reconciliation.Store // PROJ-52: completeness reconciliation + reconThresholdPct int // PROJ-52: alert threshold (percent below 7-day avg) } // SetSMTPDaemon wires the SMTP daemon into the API server after construction. @@ -151,6 +154,14 @@ func (s *Server) SetSMTPOutStore(store *smtpoutconfig.Store) { s.smtpOutStore = store } +// SetReconciliation wires the completeness-reconciliation store and the alert +// threshold (percent below the trailing 7-day average) into the API server +// (PROJ-52). +func (s *Server) SetReconciliation(store *reconciliation.Store, thresholdPct int) { + s.reconStore = store + s.reconThresholdPct = thresholdPct +} + // New creates and wires up a new API server. func New( cfg config.APIConfig, @@ -222,6 +233,9 @@ func (s *Server) routes() { s.mux.HandleFunc("GET /api/admin/system/stats", s.authAdmin(s.handleSystemStats)) s.mux.HandleFunc("GET /api/admin/stats/timeseries", s.authAdmin(s.handleMailTimeseries)) + // PROJ-52: Vollständigkeits-Reconciliation (Dashboard + CSV-Export) — admin, tenant-scoped. + s.mux.HandleFunc("GET /api/admin/reconciliation", s.authAdmin(s.handleReconciliation)) + s.mux.HandleFunc("GET /api/admin/reconciliation/export.csv", s.authAdmin(s.handleReconciliationExport)) s.mux.HandleFunc("GET /api/admin/security/audit", s.authAdmin(s.handleSecurityAudit)) // SEC-17: Security fix actions require superadmin, not just domain_admin. s.mux.HandleFunc("POST /api/admin/security/fix", s.auth(s.requireRole(userstore.RoleSuperAdmin, s.handleSecurityFix))) diff --git a/internal/api/upload.go b/internal/api/upload.go index c4239f2..af3d4c4 100644 --- a/internal/api/upload.go +++ b/internal/api/upload.go @@ -167,6 +167,11 @@ func (s *Server) importRawMessage(ctx context.Context, raw []byte, tenantID *int return "error" } + // PROJ-52: uploaded mails count as source 'import' for reconciliation. + if err := s.store.TagSource(ctx, id, "import", nil); err != nil { + s.logger.Warn("upload: tag source failed", "id", id, "err", err) + } + // Check dedup: storage.Save returns same id for duplicate content. // If already indexed, skip indexing. if already, _ := s.store.IsIndexed(ctx, id); already { diff --git a/internal/audit/audit.go b/internal/audit/audit.go index 9d03ff8..c9a8c04 100644 --- a/internal/audit/audit.go +++ b/internal/audit/audit.go @@ -24,6 +24,10 @@ const ( EventUserMgmt = "user_mgmt" EventOCRDownload = "mail:ocr_download" // PROJ-44: extracted OCR text downloaded EventDSGVORequest = "dsgvo_request" // PROJ-50: DSGVO Löschersuchen erfasst/bearbeitet + // EventReconciliationAnomaly (PROJ-52): a source's newly-archived count for a + // day dropped significantly below its trailing 7-day average, or the IMAP + // soll/ist comparison revealed a shortfall. + EventReconciliationAnomaly = "reconciliation_anomaly" ) // Entry is a single audit log record. diff --git a/internal/imap/importer.go b/internal/imap/importer.go index 9e7dbb1..0523423 100644 --- a/internal/imap/importer.go +++ b/internal/imap/importer.go @@ -170,7 +170,7 @@ func (imp *Importer) doImport(ctx context.Context, acc *Account, password string // Set per-batch deadline to prevent indefinite blocking on stalled connections. c.SetFetchDeadline() - count, err := imp.fetchBatch(ctx, c.Client, batch, acc.TenantID, log) + count, err := imp.fetchBatch(ctx, c.Client, batch, acc.TenantID, acc.ID, log) c.ClearDeadline() if err != nil { log.Error("batch fetch error — aborting import", "folder", folder, "offset", i, "err", err) @@ -188,7 +188,7 @@ func (imp *Importer) doImport(ctx context.Context, acc *Account, password string } // fetchBatch fetches and stores a batch of messages by UID. -func (imp *Importer) fetchBatch(ctx context.Context, c *imapclient.Client, uids []imapv2.UID, tenantID *int64, log *slog.Logger) (int, error) { +func (imp *Importer) fetchBatch(ctx context.Context, c *imapclient.Client, uids []imapv2.UID, tenantID *int64, accountID int64, log *slog.Logger) (int, error) { if len(uids) == 0 { return 0, nil } @@ -223,7 +223,7 @@ func (imp *Importer) fetchBatch(ctx context.Context, c *imapclient.Client, uids continue } - if err := imp.storeAndIndex(raw, tenantID, log); err != nil { + if err := imp.storeAndIndex(raw, tenantID, accountID, log); err != nil { log.Warn("failed to store/index message", "err", err) continue } @@ -240,7 +240,8 @@ func (imp *Importer) fetchBatch(ctx context.Context, c *imapclient.Client, uids } // storeAndIndex saves a raw email to storage and indexes it. -func (imp *Importer) storeAndIndex(raw []byte, tenantID *int64, log *slog.Logger) error { +// accountID identifies the IMAP account for PROJ-52 source tracking. +func (imp *Importer) storeAndIndex(raw []byte, tenantID *int64, accountID int64, log *slog.Logger) error { ctx := context.Background() // Save to file storage (deduplicates by SHA256 automatically) id, err := imp.mailStore.Save(ctx, raw, time.Now(), tenantID) @@ -248,6 +249,12 @@ func (imp *Importer) storeAndIndex(raw []byte, tenantID *int64, log *slog.Logger return fmt.Errorf("save: %w", err) } + // PROJ-52: record ingestion source (imap:). Non-fatal. + accID := accountID + if err := imp.mailStore.TagSource(ctx, id, "imap", &accID); err != nil { + log.Warn("failed to tag source", "id", id, "err", err) + } + // Parse for indexing pm, err := mailparser.Parse(raw) if err != nil { diff --git a/internal/imap/scheduler.go b/internal/imap/scheduler.go index 1d56fbe..1caac55 100644 --- a/internal/imap/scheduler.go +++ b/internal/imap/scheduler.go @@ -438,7 +438,7 @@ func (s *Scheduler) syncFolder( batch := uids[i:end] c.SetFetchDeadline() - count, batchMaxUID, err := s.fetchSyncBatch(c.Client, batch, acc.TenantID, log) + count, batchMaxUID, err := s.fetchSyncBatch(c.Client, batch, acc.TenantID, acc.ID, log) c.ClearDeadline() if err != nil { log.Warn("imap scheduler: batch error, continuing", @@ -468,6 +468,7 @@ func (s *Scheduler) fetchSyncBatch( c *imapclient.Client, uids []imapv2.UID, tenantID *int64, + accountID int64, log *slog.Logger, ) (int, uint32, error) { if len(uids) == 0 { @@ -515,7 +516,7 @@ func (s *Scheduler) fetchSyncBatch( } if len(raw) > 0 { - if err := s.importer.storeAndIndex(raw, tenantID, log); err != nil { + if err := s.importer.storeAndIndex(raw, tenantID, accountID, log); err != nil { log.Warn("imap scheduler: store/index failed", "err", err) } else { imported++ diff --git a/internal/pop3/importer.go b/internal/pop3/importer.go index 7c259e3..70153e1 100644 --- a/internal/pop3/importer.go +++ b/internal/pop3/importer.go @@ -119,7 +119,7 @@ func (imp *Importer) doImport(ctx context.Context, acc *Account, password string continue } - if err := imp.storeAndIndex(raw, log); err != nil { + if err := imp.storeAndIndex(raw, acc.ID, log); err != nil { log.Warn("failed to store/index message, skipping", "msg_num", num, "err", err) } else { imported++ @@ -133,7 +133,8 @@ func (imp *Importer) doImport(ctx context.Context, acc *Account, password string } // storeAndIndex saves a raw email to storage and indexes it. -func (imp *Importer) storeAndIndex(raw []byte, log *slog.Logger) error { +// accountID identifies the POP3 account for PROJ-52 source tracking. +func (imp *Importer) storeAndIndex(raw []byte, accountID int64, log *slog.Logger) error { ctx := context.Background() // Save to file storage (deduplicates by SHA256 automatically) id, err := imp.mailStore.Save(ctx, raw, time.Now(), imp.TenantID) @@ -141,6 +142,12 @@ func (imp *Importer) storeAndIndex(raw []byte, log *slog.Logger) error { return fmt.Errorf("pop3 save: %w", err) } + // PROJ-52: record ingestion source (pop3:). Non-fatal. + accID := accountID + if err := imp.mailStore.TagSource(ctx, id, "pop3", &accID); err != nil { + log.Warn("failed to tag source", "id", id, "err", err) + } + // Parse for indexing pm, err := mailparser.Parse(raw) if err != nil { diff --git a/internal/reconciliation/compute.go b/internal/reconciliation/compute.go new file mode 100644 index 0000000..3f8f5f5 --- /dev/null +++ b/internal/reconciliation/compute.go @@ -0,0 +1,337 @@ +package reconciliation + +import ( + "context" + "fmt" + "time" + + "archivmail/internal/audit" +) + +// bucketKey identifies one reconciliation bucket in memory. Nil tenant/source +// IDs are encoded as -1 so they can be used as map keys. +type bucketKey struct { + tenant int64 + sourceType string + sourceID int64 +} + +func keyOf(tenant, sourceID *int64, sourceType string) bucketKey { + t := int64(-1) + if tenant != nil { + t = *tenant + } + sid := int64(-1) + if sourceID != nil { + sid = *sourceID + } + return bucketKey{tenant: t, sourceType: sourceType, sourceID: sid} +} + +func ptr(v int64) *int64 { return &v } + +func nilIfNeg(v int64) *int64 { + if v < 0 { + return nil + } + return ptr(v) +} + +// imapExpected holds the IMAP soll/ist snapshot for one account. +type imapExpected struct { + tenant *int64 + expected int64 // sum of per-folder last_uid high-water marks (source proxy) + cumulArch int64 // cumulative archived mails for this account +} + +// Anomaly describes a detected significant drop for one source/day. +type Anomaly struct { + Date time.Time + TenantID *int64 + SourceKey string + Archived int64 + Average float64 + ThresholdPct int +} + +// ComputeForDate reconciles a single calendar day (UTC) and upserts one row per +// known source bucket. Rows are only written after every read query has +// succeeded, so a mid-job DB failure leaves the day WITHOUT a report row +// (dashboard shows "data missing") instead of a misleading all-zero report. +// +// After persisting, it evaluates the trailing 7-day average per source and +// writes a `reconciliation_anomaly` audit entry when today's archived count has +// dropped more than thresholdPct percent below that average. Sources with fewer +// than 7 prior daily records are skipped ("not enough data yet"). +// +// Returns the anomalies detected (also useful for the CLI summary/tests). +func (s *Store) ComputeForDate(ctx context.Context, day time.Time, thresholdPct int) ([]Anomaly, error) { + dayStart := time.Date(day.Year(), day.Month(), day.Day(), 0, 0, 0, 0, time.UTC) + dayEnd := dayStart.Add(24 * time.Hour) + + // ── Read phase (all-or-nothing) ──────────────────────────────────────── + archived, err := s.archivedForDay(ctx, dayStart, dayEnd) + if err != nil { + return nil, err + } + known, err := s.knownBuckets(ctx) + if err != nil { + return nil, err + } + imap, err := s.imapExpectedSnapshot(ctx) + if err != nil { + return nil, err + } + + // ── Build rows ───────────────────────────────────────────────────────── + rows := make([]Report, 0, len(known)) + for k := range known { + r := Report{ + Date: dayStart, + TenantID: nilIfNeg(k.tenant), + SourceType: k.sourceType, + SourceID: nilIfNeg(k.sourceID), + ArchivedCount: archived[k], // 0 when the source had no mail that day + } + // IMAP soll/ist: expected = source mailbox size proxy (sum of last_uid), + // delta = cumulative archived for the account − expected. delta going + // negative because the user emptied the source mailbox is a "good" + // direction and never alerts (alerting keys off archived_count only). + if k.sourceType == "imap" && k.sourceID >= 0 { + if ie, ok := imap[k.sourceID]; ok { + r.ExpectedCount = ptr(ie.expected) + r.Delta = ptr(ie.cumulArch - ie.expected) + } + } + rows = append(rows, r) + } + + // ── Write phase ──────────────────────────────────────────────────────── + if err := s.upsertRows(ctx, rows); err != nil { + return nil, err + } + + // ── Alert phase ──────────────────────────────────────────────────────── + var anomalies []Anomaly + for _, r := range rows { + avg, n, err := s.trailingAverage(ctx, r, dayStart) + if err != nil { + s.logger.Warn("reconciliation: trailing average failed", + "source", SourceKey(r.SourceType, r.SourceID), "err", err) + continue + } + if n < 7 { + continue // not enough history yet + } + limit := avg * (1 - float64(thresholdPct)/100.0) + if float64(r.ArchivedCount) < limit { + a := Anomaly{ + Date: dayStart, + TenantID: r.TenantID, + SourceKey: SourceKey(r.SourceType, r.SourceID), + Archived: r.ArchivedCount, + Average: avg, + ThresholdPct: thresholdPct, + } + anomalies = append(anomalies, a) + s.logAnomaly(a) + } + } + + return anomalies, nil +} + +// archivedForDay returns the count of newly archived mails per source bucket for +// the given day window. Mails with NULL source_type (legacy) bucket as 'import'. +func (s *Store) archivedForDay(ctx context.Context, start, end time.Time) (map[bucketKey]int64, error) { + rows, err := s.pool.Query(ctx, ` + SELECT tenant_id, COALESCE(source_type, 'import') AS st, source_id, COUNT(*) + FROM emails + WHERE received_at >= $1 AND received_at < $2 + GROUP BY tenant_id, st, source_id + `, start, end) + if err != nil { + return nil, fmt.Errorf("reconciliation: archived-for-day query: %w", err) + } + defer rows.Close() + + out := make(map[bucketKey]int64) + for rows.Next() { + var tenant, sourceID *int64 + var st string + var cnt int64 + if err := rows.Scan(&tenant, &st, &sourceID, &cnt); err != nil { + return nil, fmt.Errorf("reconciliation: archived-for-day scan: %w", err) + } + out[keyOf(tenant, sourceID, st)] = cnt + } + return out, rows.Err() +} + +// knownBuckets returns every (tenant, source_type, source_id) combination that +// has ever produced an archived mail. These are the buckets for which a report +// row is written every day — including explicit 0 on inactive days so gaps in +// the cron run are distinguishable from genuine zero-activity days. +func (s *Store) knownBuckets(ctx context.Context) (map[bucketKey]struct{}, error) { + rows, err := s.pool.Query(ctx, ` + SELECT DISTINCT tenant_id, COALESCE(source_type, 'import') AS st, source_id + FROM emails + `) + if err != nil { + return nil, fmt.Errorf("reconciliation: known-buckets query: %w", err) + } + defer rows.Close() + + out := make(map[bucketKey]struct{}) + for rows.Next() { + var tenant, sourceID *int64 + var st string + if err := rows.Scan(&tenant, &st, &sourceID); err != nil { + return nil, fmt.Errorf("reconciliation: known-buckets scan: %w", err) + } + out[keyOf(tenant, sourceID, st)] = struct{}{} + } + return out, rows.Err() +} + +// imapExpectedSnapshot returns the IMAP soll/ist snapshot keyed by account ID. +// expected reuses the per-folder UID high-water marks from imap_folder_state +// (PROJ-45) — no additional IMAP login. cumulArch is the cumulative number of +// archived mails attributed to the account. +func (s *Store) imapExpectedSnapshot(ctx context.Context) (map[int64]imapExpected, error) { + out := make(map[int64]imapExpected) + + // Expected proxy + tenant per account. LEFT JOIN so accounts without any + // synced folder yet still appear (expected 0). + rows, err := s.pool.Query(ctx, ` + SELECT a.id, a.tenant_id, COALESCE(SUM(fs.last_uid), 0) + FROM imap_accounts a + LEFT JOIN imap_folder_state fs ON fs.account_id = a.id + GROUP BY a.id, a.tenant_id + `) + if err != nil { + return nil, fmt.Errorf("reconciliation: imap expected query: %w", err) + } + defer rows.Close() + for rows.Next() { + var id int64 + var tenant *int64 + var expected int64 + if err := rows.Scan(&id, &tenant, &expected); err != nil { + return nil, fmt.Errorf("reconciliation: imap expected scan: %w", err) + } + out[id] = imapExpected{tenant: tenant, expected: expected} + } + if err := rows.Err(); err != nil { + return nil, err + } + + // Cumulative archived count per IMAP account. + crows, err := s.pool.Query(ctx, ` + SELECT source_id, COUNT(*) + FROM emails + WHERE source_type = 'imap' AND source_id IS NOT NULL + GROUP BY source_id + `) + if err != nil { + return nil, fmt.Errorf("reconciliation: imap archived query: %w", err) + } + defer crows.Close() + for crows.Next() { + var id, cnt int64 + if err := crows.Scan(&id, &cnt); err != nil { + return nil, fmt.Errorf("reconciliation: imap archived scan: %w", err) + } + ie := out[id] + ie.cumulArch = cnt + out[id] = ie + } + return out, crows.Err() +} + +// upsertRows writes all report rows in a single transaction. On any error the +// transaction is rolled back so no partial day is persisted. +func (s *Store) upsertRows(ctx context.Context, rows []Report) error { + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("reconciliation: begin tx: %w", err) + } + defer tx.Rollback(ctx) + + for _, r := range rows { + _, err := tx.Exec(ctx, ` + INSERT INTO reconciliation_reports + (date, tenant_id, source_type, source_id, expected_count, archived_count, delta) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (date, COALESCE(tenant_id, -1), source_type, COALESCE(source_id, -1)) + DO UPDATE SET + expected_count = EXCLUDED.expected_count, + archived_count = EXCLUDED.archived_count, + delta = EXCLUDED.delta, + created_at = NOW() + `, r.Date, r.TenantID, r.SourceType, r.SourceID, r.ExpectedCount, r.ArchivedCount, r.Delta) + if err != nil { + return fmt.Errorf("reconciliation: upsert row: %w", err) + } + } + + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("reconciliation: commit: %w", err) + } + return nil +} + +// trailingAverage returns the mean archived_count of the up-to-7 report rows +// immediately preceding the given day for the same source bucket, plus how many +// prior day rows were found. NULL-safe matching on tenant_id / source_id. +func (s *Store) trailingAverage(ctx context.Context, r Report, day time.Time) (float64, int, error) { + rows, err := s.pool.Query(ctx, ` + SELECT archived_count + FROM reconciliation_reports + WHERE source_type = $1 + AND tenant_id IS NOT DISTINCT FROM $2 + AND source_id IS NOT DISTINCT FROM $3 + AND date < $4 + ORDER BY date DESC + LIMIT 7 + `, r.SourceType, r.TenantID, r.SourceID, day) + if err != nil { + return 0, 0, fmt.Errorf("reconciliation: trailing average query: %w", err) + } + defer rows.Close() + + var sum int64 + var n int + for rows.Next() { + var c int64 + if err := rows.Scan(&c); err != nil { + return 0, 0, err + } + sum += c + n++ + } + if err := rows.Err(); err != nil { + return 0, 0, err + } + if n == 0 { + return 0, 0, nil + } + return float64(sum) / float64(n), n, nil +} + +// logAnomaly emits a structured log line and, when wired, a tenant-visible +// audit entry for a detected drop. +func (s *Store) logAnomaly(a Anomaly) { + detail := fmt.Sprintf("source=%s date=%s archived=%d avg_7d=%.1f threshold=%d%%", + a.SourceKey, a.Date.Format("2006-01-02"), a.Archived, a.Average, a.ThresholdPct) + s.logger.Warn("reconciliation: anomaly detected", "detail", detail) + if s.audlog != nil { + s.audlog.Log(audit.Entry{ + EventType: audit.EventReconciliationAnomaly, + Username: "system", + TenantID: a.TenantID, + Success: false, + Detail: detail, + }) + } +} diff --git a/internal/reconciliation/query.go b/internal/reconciliation/query.go new file mode 100644 index 0000000..5539a95 --- /dev/null +++ b/internal/reconciliation/query.go @@ -0,0 +1,177 @@ +package reconciliation + +import ( + "context" + "fmt" + "time" +) + +// DayPoint is one day's figures for a source in the dashboard response. +// ArchivedCount is nil (and Missing true) when no report row exists for that +// date — i.e. the cron job did not run — which is distinct from an archived +// count of 0 on a genuine zero-activity day. +type DayPoint struct { + Date string `json:"date"` + ArchivedCount *int64 `json:"archived_count"` + ExpectedCount *int64 `json:"expected_count"` + Delta *int64 `json:"delta"` + Missing bool `json:"missing"` +} + +// SourceSummary aggregates the trailing days plus alert state for one source. +type SourceSummary struct { + SourceType string `json:"source_type"` + SourceID *int64 `json:"source_id"` + SourceKey string `json:"source_key"` + TenantID *int64 `json:"tenant_id"` + Points []DayPoint `json:"points"` + Avg7d float64 `json:"avg_7d"` + EnoughData bool `json:"enough_data"` + Alert bool `json:"alert"` +} + +// DashboardData returns the last `days` calendar days of reconciliation figures +// per source, tenant-scoped. When tenantID is nil (superadmin) all tenants are +// included; otherwise only rows for that tenant are returned. thresholdPct is +// used to compute the per-source Alert flag consistently with the cron job. +func (s *Store) DashboardData(ctx context.Context, tenantID *int64, days, thresholdPct int) ([]SourceSummary, error) { + if days <= 0 { + days = 7 + } + today := time.Now().UTC().Truncate(24 * time.Hour) + start := today.AddDate(0, 0, -(days - 1)) + + rows, err := s.queryRange(ctx, tenantID, start, today.Add(24*time.Hour)) + if err != nil { + return nil, err + } + + // Ordered date labels for the window. + dateLabels := make([]string, days) + for i := 0; i < days; i++ { + dateLabels[i] = start.AddDate(0, 0, i).Format("2006-01-02") + } + + type srcAgg struct { + meta Report + byDate map[string]Report + } + agg := map[string]*srcAgg{} + for _, r := range rows { + key := SourceKey(r.SourceType, r.SourceID) + // Distinguish sources of different tenants sharing a key. + if r.TenantID != nil { + key = fmt.Sprintf("t%d/%s", *r.TenantID, key) + } + a, ok := agg[key] + if !ok { + a = &srcAgg{meta: r, byDate: map[string]Report{}} + agg[key] = a + } + a.byDate[r.Date.UTC().Format("2006-01-02")] = r + } + + summaries := make([]SourceSummary, 0, len(agg)) + for _, a := range agg { + sum := SourceSummary{ + SourceType: a.meta.SourceType, + SourceID: a.meta.SourceID, + SourceKey: SourceKey(a.meta.SourceType, a.meta.SourceID), + TenantID: a.meta.TenantID, + Points: make([]DayPoint, 0, days), + } + for _, d := range dateLabels { + if r, ok := a.byDate[d]; ok { + c := r.ArchivedCount + sum.Points = append(sum.Points, DayPoint{ + Date: d, + ArchivedCount: &c, + ExpectedCount: r.ExpectedCount, + Delta: r.Delta, + Missing: false, + }) + } else { + sum.Points = append(sum.Points, DayPoint{Date: d, Missing: true}) + } + } + + // Alert against the trailing 7-day average of the most recent day that + // actually has a report row (mirrors the cron job's evaluation). + latest, latestDate, hasLatest := latestPresent(a.byDate, dateLabels) + if hasLatest { + avg, n, err := s.trailingAverage(ctx, a.meta, latestDate) + if err == nil && n >= 7 { + sum.Avg7d = avg + sum.EnoughData = true + limit := avg * (1 - float64(thresholdPct)/100.0) + if float64(latest.ArchivedCount) < limit { + sum.Alert = true + } + } + } + summaries = append(summaries, sum) + } + return summaries, nil +} + +// latestPresent returns the most recent report row within the window that has a +// stored row, along with its date. +func latestPresent(byDate map[string]Report, dateLabels []string) (Report, time.Time, bool) { + for i := len(dateLabels) - 1; i >= 0; i-- { + if r, ok := byDate[dateLabels[i]]; ok { + d, _ := time.Parse("2006-01-02", dateLabels[i]) + return r, d, true + } + } + return Report{}, time.Time{}, false +} + +// ExportRows returns raw report rows for CSV export, tenant-scoped, for the +// last `days` days, ordered by date descending then source. +func (s *Store) ExportRows(ctx context.Context, tenantID *int64, days int) ([]Report, error) { + if days <= 0 { + days = 30 + } + today := time.Now().UTC().Truncate(24 * time.Hour) + start := today.AddDate(0, 0, -(days - 1)) + return s.queryRange(ctx, tenantID, start, today.Add(24*time.Hour)) +} + +// queryRange loads report rows in [start, end) filtered by tenant. tenantID nil +// returns all tenants (superadmin scope). +func (s *Store) queryRange(ctx context.Context, tenantID *int64, start, end time.Time) ([]Report, error) { + var ( + sql string + args []interface{} + ) + if tenantID == nil { + sql = `SELECT date, tenant_id, source_type, source_id, expected_count, archived_count, delta + FROM reconciliation_reports + WHERE date >= $1 AND date < $2 + ORDER BY date DESC, source_type, source_id` + args = []interface{}{start, end} + } else { + sql = `SELECT date, tenant_id, source_type, source_id, expected_count, archived_count, delta + FROM reconciliation_reports + WHERE date >= $1 AND date < $2 AND tenant_id = $3 + ORDER BY date DESC, source_type, source_id` + args = []interface{}{start, end, *tenantID} + } + + rows, err := s.pool.Query(ctx, sql, args...) + if err != nil { + return nil, fmt.Errorf("reconciliation: query range: %w", err) + } + defer rows.Close() + + var out []Report + for rows.Next() { + var r Report + if err := rows.Scan(&r.Date, &r.TenantID, &r.SourceType, &r.SourceID, + &r.ExpectedCount, &r.ArchivedCount, &r.Delta); err != nil { + return nil, fmt.Errorf("reconciliation: scan range: %w", err) + } + out = append(out, r) + } + return out, rows.Err() +} diff --git a/internal/reconciliation/reconciliation.go b/internal/reconciliation/reconciliation.go new file mode 100644 index 0000000..5e16cc3 --- /dev/null +++ b/internal/reconciliation/reconciliation.go @@ -0,0 +1,120 @@ +// 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 +} diff --git a/internal/smtpd/smtpd.go b/internal/smtpd/smtpd.go index 8dfcb35..7059fec 100644 --- a/internal/smtpd/smtpd.go +++ b/internal/smtpd/smtpd.go @@ -346,6 +346,12 @@ func (s *session) Data(r io.Reader) error { } } + // PROJ-52: record the ingestion source for the reconciliation report. + // Non-fatal — a failed metadata write must not reject an already-stored mail. + if err := s.daemon.store.TagSource(context.Background(), id, "smtp", nil); err != nil { + s.daemon.logger.Warn("SMTP: tag source failed", "id", id, "err", err) + } + s.daemon.stats.Received.Add(1) s.daemon.stats.LastMailAt.Store(time.Now()) s.daemon.logger.Info("SMTP: mail stored", "id", id, "from", s.from, diff --git a/internal/storage/storage.go b/internal/storage/storage.go index 273508f..ba8cb8f 100644 --- a/internal/storage/storage.go +++ b/internal/storage/storage.go @@ -359,9 +359,48 @@ func (s *Store) initSchema(ctx context.Context) error { ALTER TABLE emails ADD COLUMN IF NOT EXISTS marked_for_deletion_at TIMESTAMPTZ; CREATE INDEX IF NOT EXISTS idx_emails_marked_for_deletion ON emails (marked_for_deletion) WHERE marked_for_deletion = TRUE; `) + if err != nil { + return err + } + + // PROJ-52: ingestion source tracking for the completeness reconciliation + // report. source_type is one of 'smtp', 'imap', 'pop3', 'import'; source_id + // holds the IMAP/POP3 account ID (NULL for smtp/import). Both are NULL for + // legacy mails archived before this migration — the reconciliation job + // buckets those as 'import'. The composite index accelerates the per-day, + // per-source GROUP BY the reconciliation job runs. + _, err = s.db.Exec(ctx, ` + ALTER TABLE emails ADD COLUMN IF NOT EXISTS source_type TEXT; + ALTER TABLE emails ADD COLUMN IF NOT EXISTS source_id BIGINT; + CREATE INDEX IF NOT EXISTS idx_emails_source_recon ON emails (received_at, source_type, source_id, tenant_id); + `) return err } +// TagSource records the ingestion channel of an archived mail (PROJ-52). +// sourceType is one of 'smtp', 'imap', 'pop3', 'import'; sourceID holds the +// IMAP/POP3 account ID (nil for smtp/import). +// +// First-write-wins: the update only sets the columns while source_type IS NULL. +// A mail deduplicated across channels (SHA-256 / Message-ID dedup in Save) keeps +// the source of its first ingestion, so the reconciliation counts never +// double-count a re-delivered mail. Errors are non-fatal for the intake path — +// callers log and continue so a reconciliation-metadata write never blocks +// archival (GoBD completeness of the mail itself takes precedence). +func (s *Store) TagSource(ctx context.Context, id, sourceType string, sourceID *int64) error { + if s.db == nil { + return nil + } + _, err := s.db.Exec(ctx, ` + UPDATE emails SET source_type = $2, source_id = $3 + WHERE id = $1 AND source_type IS NULL + `, id, sourceType, sourceID) + if err != nil { + return fmt.Errorf("storage: tag source: %w", err) + } + return nil +} + // ── Core operations ─────────────────────────────────────────────────────── // Save writes raw email bytes to storage. The ID is the hex-encoded SHA256 of diff --git a/src/components/admin/tabs/DashboardTab.tsx b/src/components/admin/tabs/DashboardTab.tsx index cac266d..bcf4350 100644 --- a/src/components/admin/tabs/DashboardTab.tsx +++ b/src/components/admin/tabs/DashboardTab.tsx @@ -15,6 +15,7 @@ import { Badge } from "@/components/ui/badge"; import { Skeleton } from "@/components/ui/skeleton"; import { Separator } from "@/components/ui/separator"; import { Alert, AlertDescription } from "@/components/ui/alert"; +import { ReconciliationCard } from "@/components/admin/tabs/ReconciliationCard"; function formatBytes(bytes: number): string { if (bytes < 1024) return `${bytes} B`; @@ -491,6 +492,9 @@ export function DashboardTab({ )} + {/* Vollständigkeits-Check (PROJ-52) — tenant-gescoped, domain_admin+ */} + + {/* Benutzerübersicht */} diff --git a/src/components/admin/tabs/ReconciliationCard.tsx b/src/components/admin/tabs/ReconciliationCard.tsx new file mode 100644 index 0000000..1307d93 --- /dev/null +++ b/src/components/admin/tabs/ReconciliationCard.tsx @@ -0,0 +1,271 @@ +"use client"; + +import { useCallback, useEffect, useState } from "react"; +import { + exportReconciliationCSV, + getReconciliation, + type ReconciliationResponse, + type ReconciliationSource, +} from "@/lib/api"; +import { Button } from "@/components/ui/button"; +import { Card, CardContent } from "@/components/ui/card"; +import { Badge } from "@/components/ui/badge"; +import { Skeleton } from "@/components/ui/skeleton"; +import { Separator } from "@/components/ui/separator"; +import { Alert, AlertDescription } from "@/components/ui/alert"; +import { + Table, + TableBody, + TableCell, + TableHead, + TableHeader, + TableRow, +} from "@/components/ui/table"; + +/** Formatiert einen source_key zu einem lesbaren Label. */ +function formatSourceLabel(source: ReconciliationSource): string { + switch (source.source_type) { + case "smtp": + return "SMTP-Journal"; + case "import": + return "Datei-Import"; + case "imap": + return `IMAP-Konto #${source.source_id ?? "?"}`; + case "pop3": + return `POP3-Konto #${source.source_id ?? "?"}`; + default: + return source.source_key; + } +} + +function formatDayLabel(iso: string): string { + const d = new Date(iso); + return d.toLocaleDateString("de-DE", { day: "2-digit", month: "2-digit" }); +} + +/** Zellinhalt für einen Tagespunkt: "—" bei fehlendem Datensatz, sonst Werte. */ +function DayCell({ + archived, + expected, + delta, + missing, + isImap, +}: { + archived: number | null; + expected: number | null; + delta: number | null; + missing: boolean; + isImap: boolean; +}) { + if (missing) { + return ( + + — + + ); + } + return ( + + + {(archived ?? 0).toLocaleString("de-DE")} + + {isImap && expected != null && ( + + Soll {expected.toLocaleString("de-DE")} + {delta != null && ( + + ({delta > 0 ? "+" : ""} + {delta.toLocaleString("de-DE")}) + + )} + + )} + + ); +} + +export function ReconciliationCard() { + const [data, setData] = useState(null); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + const [exporting, setExporting] = useState(false); + + const load = useCallback(async () => { + setLoading(true); + setError(null); + try { + const res = await getReconciliation(7); + setData(res); + } catch (e: unknown) { + setError(e instanceof Error ? e.message : "Laden fehlgeschlagen"); + } finally { + setLoading(false); + } + }, []); + + useEffect(() => { + void load(); + }, [load]); + + const handleExport = async () => { + setExporting(true); + try { + const { blob, filename } = await exportReconciliationCSV(30); + const url = URL.createObjectURL(blob); + const a = document.createElement("a"); + a.href = url; + a.download = filename; + document.body.appendChild(a); + a.click(); + a.remove(); + URL.revokeObjectURL(url); + } catch (e: unknown) { + setError(e instanceof Error ? e.message : "Export fehlgeschlagen"); + } finally { + setExporting(false); + } + }; + + // Tages-Header aus dem ersten Quell-Eintrag ableiten (alle Quellen haben + // dieselben Tage in gleicher Reihenfolge). + const dayHeaders = data?.sources[0]?.points.map((p) => p.date) ?? []; + + return ( + + +
+
+ + Vollständigkeits-Check + + {data && ( + + letzte {data.days} Tage · Schwelle {data.threshold_pct}% + + )} +
+
+ + +
+
+ + + {loading ? ( +
+ {Array.from({ length: 3 }).map((_, i) => ( + + ))} +
+ ) : error ? ( + + + Vollständigkeits-Report konnte nicht geladen werden: {error} + + + ) : !data || data.sources.length === 0 ? ( +

+ Noch keine Reconciliation-Daten vorhanden. Der tägliche Zähl-Job + (archivmail reconcile) hat noch + keine Datensätze erzeugt. +

+ ) : ( +
+ + + + Quelle + {dayHeaders.map((d) => ( + + {formatDayLabel(d)} + + ))} + + Ø 7 Tage + + + Status + + + + + {data.sources.map((s) => { + const isImap = s.source_type === "imap"; + return ( + + + {formatSourceLabel(s)} + + {!s.enough_data ? ( + + Noch nicht genug Daten + + ) : ( + <> + {s.points.map((p) => ( + + + + ))} + + {s.avg_7d != null + ? s.avg_7d.toLocaleString("de-DE", { + maximumFractionDigits: 1, + }) + : "—"} + + + )} + + {s.alert ? ( + Auffällig + ) : s.enough_data ? ( + OK + ) : ( + — + )} + + + ); + })} + +
+
+ )} + + + ); +} diff --git a/src/lib/api/index.ts b/src/lib/api/index.ts index 9c833e7..2a7cdc3 100644 --- a/src/lib/api/index.ts +++ b/src/lib/api/index.ts @@ -178,6 +178,16 @@ export { deleteArchivingRule, } from "./archiving_rules"; +export type { + ReconciliationPoint, + ReconciliationSource, + ReconciliationResponse, +} from "./reconciliation"; +export { + getReconciliation, + exportReconciliationCSV, +} from "./reconciliation"; + export type { SavedSearch } from "./saved_searches"; export { listSavedSearches, diff --git a/src/lib/api/reconciliation.ts b/src/lib/api/reconciliation.ts new file mode 100644 index 0000000..8fce53e --- /dev/null +++ b/src/lib/api/reconciliation.ts @@ -0,0 +1,51 @@ +import { API_BASE, request } from "./core"; + +// ── Types ──────────────────────────────────────────────────────────────────── + +export interface ReconciliationPoint { + date: string; // "2026-06-27" + archived_count: number | null; + expected_count: number | null; + delta: number | null; + missing: boolean; // true = kein Report-Datensatz (Cron nicht gelaufen) ≠ archived_count:0 +} + +export interface ReconciliationSource { + source_type: string; // "smtp" | "imap" | "pop3" | "import" + source_id: number | null; + source_key: string; // "smtp" | "import" | "imap:" | "pop3:" + tenant_id: number | null; + points: ReconciliationPoint[]; + avg_7d: number | null; + enough_data: boolean; + alert: boolean; +} + +export interface ReconciliationResponse { + days: number; + threshold_pct: number; + sources: ReconciliationSource[]; +} + +// ── API ────────────────────────────────────────────────────────────────────── + +export async function getReconciliation(days = 7): Promise { + return request(`/api/admin/reconciliation?days=${days}`); +} + +/** Lädt den Reconciliation-Report als CSV-Datei herunter. */ +export async function exportReconciliationCSV( + days = 30 +): Promise<{ blob: Blob; filename: string }> { + const res = await fetch( + `${API_BASE}/api/admin/reconciliation/export.csv?days=${days}`, + { credentials: "include" } + ); + if (!res.ok) throw new Error(`Export fehlgeschlagen: ${res.status}`); + const disposition = res.headers.get("Content-Disposition") || ""; + const match = disposition.match(/filename[^;=\n]*=((['"]).*?\2|[^;\n]*)/); + const filename = match + ? match[1].replace(/['"]/g, "") + : `reconciliation-${days}d.csv`; + return { blob: await res.blob(), filename }; +}