feat(PROJ-52): Vollständigkeits-Reconciliation (Zähl-Report Mailserver vs. Archiv)
Täglicher Cron-Job (archivmail reconcile) berechnet pro Tenant/Quelle (SMTP-Journal, IMAP-Konto, POP3-Konto, Datei-Import) archivierte Mail-Zahlen, für IMAP zusätzlich einen Soll/Ist-Vergleich via UID-Tracking. Abweichungen über Schwellenwert erzeugen Audit-Log-Warnung. Neue Admin-Dashboard-Kachel "Vollständigkeits-Check" (letzte 7 Tage, Warn-Badge, CSV-Export). Schließt die "teilweise erfüllt"-Lücke bei Vollständigkeit im GoBD/DSGVO-Compliance-Check (VOI-Grundsatz 2). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
b286352d07
commit
be93614c9f
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
+18
-1
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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).
|
||||
|
||||
+1
-1
@@ -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 |
|
||||
|
||||
@@ -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:<account_id>`, `pop3:<account_id>`, `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 <Datum>" 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)
|
||||
|
||||
---
|
||||
<!-- Sections below are added by subsequent skills -->
|
||||
|
||||
## 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:<id>` | `pop3:<id>`. `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<T>`-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 <expected> (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_
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)))
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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:<account_id>). 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 {
|
||||
|
||||
@@ -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++
|
||||
|
||||
@@ -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:<account_id>). 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 {
|
||||
|
||||
@@ -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,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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:<account_id>", "pop3:<account_id>".
|
||||
func SourceKey(sourceType string, sourceID *int64) string {
|
||||
if sourceID != nil && (sourceType == "imap" || sourceType == "pop3") {
|
||||
return fmt.Sprintf("%s:%d", sourceType, *sourceID)
|
||||
}
|
||||
return sourceType
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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({
|
||||
</Card>
|
||||
)}
|
||||
|
||||
{/* Vollständigkeits-Check (PROJ-52) — tenant-gescoped, domain_admin+ */}
|
||||
<ReconciliationCard />
|
||||
|
||||
{/* Benutzerübersicht */}
|
||||
<Card>
|
||||
<CardContent className="pt-6 space-y-2">
|
||||
|
||||
@@ -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 (
|
||||
<span
|
||||
className="text-muted-foreground"
|
||||
title="Kein Report-Datensatz für diesen Tag (Cron nicht gelaufen)"
|
||||
>
|
||||
—
|
||||
</span>
|
||||
);
|
||||
}
|
||||
return (
|
||||
<span className="inline-flex flex-col items-end leading-tight">
|
||||
<span className="font-medium tabular-nums">
|
||||
{(archived ?? 0).toLocaleString("de-DE")}
|
||||
</span>
|
||||
{isImap && expected != null && (
|
||||
<span className="text-[10px] text-muted-foreground tabular-nums">
|
||||
Soll {expected.toLocaleString("de-DE")}
|
||||
{delta != null && (
|
||||
<span
|
||||
className={
|
||||
delta < 0 ? "ml-1 text-destructive" : "ml-1 text-green-600"
|
||||
}
|
||||
>
|
||||
({delta > 0 ? "+" : ""}
|
||||
{delta.toLocaleString("de-DE")})
|
||||
</span>
|
||||
)}
|
||||
</span>
|
||||
)}
|
||||
</span>
|
||||
);
|
||||
}
|
||||
|
||||
export function ReconciliationCard() {
|
||||
const [data, setData] = useState<ReconciliationResponse | null>(null);
|
||||
const [loading, setLoading] = useState(true);
|
||||
const [error, setError] = useState<string | null>(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 (
|
||||
<Card>
|
||||
<CardContent className="pt-6 space-y-3">
|
||||
<div className="flex items-center justify-between gap-3">
|
||||
<div className="flex items-center gap-2">
|
||||
<span className="text-sm font-medium text-muted-foreground">
|
||||
Vollständigkeits-Check
|
||||
</span>
|
||||
{data && (
|
||||
<span className="text-xs text-muted-foreground">
|
||||
letzte {data.days} Tage · Schwelle {data.threshold_pct}%
|
||||
</span>
|
||||
)}
|
||||
</div>
|
||||
<div className="flex items-center gap-2">
|
||||
<Button
|
||||
variant="outline"
|
||||
size="sm"
|
||||
onClick={load}
|
||||
disabled={loading}
|
||||
>
|
||||
{loading ? "..." : "Aktualisieren"}
|
||||
</Button>
|
||||
<Button
|
||||
variant="outline"
|
||||
size="sm"
|
||||
onClick={handleExport}
|
||||
disabled={exporting}
|
||||
>
|
||||
{exporting ? "Export..." : "CSV-Export (30 Tage)"}
|
||||
</Button>
|
||||
</div>
|
||||
</div>
|
||||
<Separator />
|
||||
|
||||
{loading ? (
|
||||
<div className="space-y-2">
|
||||
{Array.from({ length: 3 }).map((_, i) => (
|
||||
<Skeleton key={i} className="h-8 w-full" />
|
||||
))}
|
||||
</div>
|
||||
) : error ? (
|
||||
<Alert variant="destructive">
|
||||
<AlertDescription>
|
||||
Vollständigkeits-Report konnte nicht geladen werden: {error}
|
||||
</AlertDescription>
|
||||
</Alert>
|
||||
) : !data || data.sources.length === 0 ? (
|
||||
<p className="text-sm text-muted-foreground">
|
||||
Noch keine Reconciliation-Daten vorhanden. Der tägliche Zähl-Job
|
||||
(<code className="font-mono">archivmail reconcile</code>) hat noch
|
||||
keine Datensätze erzeugt.
|
||||
</p>
|
||||
) : (
|
||||
<div className="overflow-x-auto">
|
||||
<Table>
|
||||
<TableHeader>
|
||||
<TableRow>
|
||||
<TableHead className="whitespace-nowrap">Quelle</TableHead>
|
||||
{dayHeaders.map((d) => (
|
||||
<TableHead
|
||||
key={d}
|
||||
className="text-right whitespace-nowrap"
|
||||
>
|
||||
{formatDayLabel(d)}
|
||||
</TableHead>
|
||||
))}
|
||||
<TableHead className="text-right whitespace-nowrap">
|
||||
Ø 7 Tage
|
||||
</TableHead>
|
||||
<TableHead className="text-right whitespace-nowrap">
|
||||
Status
|
||||
</TableHead>
|
||||
</TableRow>
|
||||
</TableHeader>
|
||||
<TableBody>
|
||||
{data.sources.map((s) => {
|
||||
const isImap = s.source_type === "imap";
|
||||
return (
|
||||
<TableRow key={s.source_key}>
|
||||
<TableCell className="font-medium whitespace-nowrap">
|
||||
{formatSourceLabel(s)}
|
||||
</TableCell>
|
||||
{!s.enough_data ? (
|
||||
<TableCell
|
||||
colSpan={dayHeaders.length + 1}
|
||||
className="text-center text-sm text-muted-foreground"
|
||||
>
|
||||
Noch nicht genug Daten
|
||||
</TableCell>
|
||||
) : (
|
||||
<>
|
||||
{s.points.map((p) => (
|
||||
<TableCell key={p.date} className="text-right">
|
||||
<DayCell
|
||||
archived={p.archived_count}
|
||||
expected={p.expected_count}
|
||||
delta={p.delta}
|
||||
missing={p.missing}
|
||||
isImap={isImap}
|
||||
/>
|
||||
</TableCell>
|
||||
))}
|
||||
<TableCell className="text-right tabular-nums text-muted-foreground">
|
||||
{s.avg_7d != null
|
||||
? s.avg_7d.toLocaleString("de-DE", {
|
||||
maximumFractionDigits: 1,
|
||||
})
|
||||
: "—"}
|
||||
</TableCell>
|
||||
</>
|
||||
)}
|
||||
<TableCell className="text-right">
|
||||
{s.alert ? (
|
||||
<Badge variant="destructive">Auffällig</Badge>
|
||||
) : s.enough_data ? (
|
||||
<Badge variant="secondary">OK</Badge>
|
||||
) : (
|
||||
<Badge variant="outline">—</Badge>
|
||||
)}
|
||||
</TableCell>
|
||||
</TableRow>
|
||||
);
|
||||
})}
|
||||
</TableBody>
|
||||
</Table>
|
||||
</div>
|
||||
)}
|
||||
</CardContent>
|
||||
</Card>
|
||||
);
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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:<id>" | "pop3:<id>"
|
||||
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<ReconciliationResponse> {
|
||||
return request<ReconciliationResponse>(`/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 };
|
||||
}
|
||||
Reference in New Issue
Block a user