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"
|
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
|
var attachNames []string
|
||||||
for _, a := range pm.Attachments {
|
for _, a := range pm.Attachments {
|
||||||
attachNames = append(attachNames, a.Filename)
|
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/mailer"
|
||||||
"archivmail/internal/ocr"
|
"archivmail/internal/ocr"
|
||||||
pop3store "archivmail/internal/pop3"
|
pop3store "archivmail/internal/pop3"
|
||||||
|
"archivmail/internal/reconciliation"
|
||||||
"archivmail/internal/smtpoutconfig"
|
"archivmail/internal/smtpoutconfig"
|
||||||
"archivmail/internal/smtpd"
|
"archivmail/internal/smtpd"
|
||||||
"archivmail/internal/storage"
|
"archivmail/internal/storage"
|
||||||
@@ -73,6 +74,9 @@ func main() {
|
|||||||
case "index-pending":
|
case "index-pending":
|
||||||
runIndexPending(os.Args[2:])
|
runIndexPending(os.Args[2:])
|
||||||
return
|
return
|
||||||
|
case "reconcile":
|
||||||
|
runReconcile(os.Args[2:])
|
||||||
|
return
|
||||||
case "update":
|
case "update":
|
||||||
runUpdate(os.Args[2:])
|
runUpdate(os.Args[2:])
|
||||||
return
|
return
|
||||||
@@ -313,11 +317,24 @@ func main() {
|
|||||||
srv.SetGlobalRetentionDays(cfg.Storage.RetentionDays)
|
srv.SetGlobalRetentionDays(cfg.Storage.RetentionDays)
|
||||||
srv.SetMetrics(cfg.Metrics)
|
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
|
// PROJ-28: Self-Service Onboarding — mailer + token store + FQDN
|
||||||
mlr := mailer.New(cfg.SMTPOut)
|
mlr := mailer.New(cfg.SMTPOut)
|
||||||
|
|
||||||
// SMTP-Out config store — load from DB, overrides config.yml if present
|
// 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 {
|
if err != nil {
|
||||||
logger.Error("smtp-out config store init failed", "err", err)
|
logger.Error("smtp-out config store init failed", "err", err)
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
|
|||||||
@@ -50,3 +50,12 @@ audit:
|
|||||||
# Default falls leer: /var/log/archivmail/audit.log
|
# Default falls leer: /var/log/archivmail/audit.log
|
||||||
log_path: /tmp/archivmail-test/audit.log
|
log_path: /tmp/archivmail-test/audit.log
|
||||||
retention_days: 0
|
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
|
log_path: /var/archivmail/audit.log
|
||||||
retention_days: 365
|
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:
|
logging:
|
||||||
level: info
|
level: info
|
||||||
|
|||||||
@@ -42,6 +42,39 @@ type Config struct {
|
|||||||
// PROJ-56: load-spreading for background jobs.
|
// PROJ-56: load-spreading for background jobs.
|
||||||
OCR OCRConfig `yaml:"ocr"`
|
OCR OCRConfig `yaml:"ocr"`
|
||||||
IMAPScheduler IMAPSchedulerConfig `yaml:"imap_scheduler"`
|
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).
|
// 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-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-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-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-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-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 |
|
| 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"
|
ldapcfg "archivmail/internal/ldapconfig"
|
||||||
"archivmail/internal/mailer"
|
"archivmail/internal/mailer"
|
||||||
pop3store "archivmail/internal/pop3"
|
pop3store "archivmail/internal/pop3"
|
||||||
|
"archivmail/internal/reconciliation"
|
||||||
"archivmail/internal/smtpoutconfig"
|
"archivmail/internal/smtpoutconfig"
|
||||||
"archivmail/internal/smtpd"
|
"archivmail/internal/smtpd"
|
||||||
"archivmail/internal/storage"
|
"archivmail/internal/storage"
|
||||||
@@ -90,6 +91,8 @@ type Server struct {
|
|||||||
fqdn string // from server.fqdn config (PROJ-28)
|
fqdn string // from server.fqdn config (PROJ-28)
|
||||||
smtpOutStore *smtpoutconfig.Store
|
smtpOutStore *smtpoutconfig.Store
|
||||||
apiKeyMw *auth.APIKeyMiddleware // PROJ-13: external API auth
|
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.
|
// 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
|
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.
|
// New creates and wires up a new API server.
|
||||||
func New(
|
func New(
|
||||||
cfg config.APIConfig,
|
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/system/stats", s.authAdmin(s.handleSystemStats))
|
||||||
s.mux.HandleFunc("GET /api/admin/stats/timeseries", s.authAdmin(s.handleMailTimeseries))
|
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))
|
s.mux.HandleFunc("GET /api/admin/security/audit", s.authAdmin(s.handleSecurityAudit))
|
||||||
// SEC-17: Security fix actions require superadmin, not just domain_admin.
|
// 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)))
|
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"
|
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.
|
// Check dedup: storage.Save returns same id for duplicate content.
|
||||||
// If already indexed, skip indexing.
|
// If already indexed, skip indexing.
|
||||||
if already, _ := s.store.IsIndexed(ctx, id); already {
|
if already, _ := s.store.IsIndexed(ctx, id); already {
|
||||||
|
|||||||
@@ -24,6 +24,10 @@ const (
|
|||||||
EventUserMgmt = "user_mgmt"
|
EventUserMgmt = "user_mgmt"
|
||||||
EventOCRDownload = "mail:ocr_download" // PROJ-44: extracted OCR text downloaded
|
EventOCRDownload = "mail:ocr_download" // PROJ-44: extracted OCR text downloaded
|
||||||
EventDSGVORequest = "dsgvo_request" // PROJ-50: DSGVO Löschersuchen erfasst/bearbeitet
|
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.
|
// 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.
|
// Set per-batch deadline to prevent indefinite blocking on stalled connections.
|
||||||
c.SetFetchDeadline()
|
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()
|
c.ClearDeadline()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Error("batch fetch error — aborting import", "folder", folder, "offset", i, "err", err)
|
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.
|
// 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 {
|
if len(uids) == 0 {
|
||||||
return 0, nil
|
return 0, nil
|
||||||
}
|
}
|
||||||
@@ -223,7 +223,7 @@ func (imp *Importer) fetchBatch(ctx context.Context, c *imapclient.Client, uids
|
|||||||
continue
|
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)
|
log.Warn("failed to store/index message", "err", err)
|
||||||
continue
|
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.
|
// 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()
|
ctx := context.Background()
|
||||||
// Save to file storage (deduplicates by SHA256 automatically)
|
// Save to file storage (deduplicates by SHA256 automatically)
|
||||||
id, err := imp.mailStore.Save(ctx, raw, time.Now(), tenantID)
|
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)
|
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
|
// Parse for indexing
|
||||||
pm, err := mailparser.Parse(raw)
|
pm, err := mailparser.Parse(raw)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -438,7 +438,7 @@ func (s *Scheduler) syncFolder(
|
|||||||
batch := uids[i:end]
|
batch := uids[i:end]
|
||||||
|
|
||||||
c.SetFetchDeadline()
|
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()
|
c.ClearDeadline()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Warn("imap scheduler: batch error, continuing",
|
log.Warn("imap scheduler: batch error, continuing",
|
||||||
@@ -468,6 +468,7 @@ func (s *Scheduler) fetchSyncBatch(
|
|||||||
c *imapclient.Client,
|
c *imapclient.Client,
|
||||||
uids []imapv2.UID,
|
uids []imapv2.UID,
|
||||||
tenantID *int64,
|
tenantID *int64,
|
||||||
|
accountID int64,
|
||||||
log *slog.Logger,
|
log *slog.Logger,
|
||||||
) (int, uint32, error) {
|
) (int, uint32, error) {
|
||||||
if len(uids) == 0 {
|
if len(uids) == 0 {
|
||||||
@@ -515,7 +516,7 @@ func (s *Scheduler) fetchSyncBatch(
|
|||||||
}
|
}
|
||||||
|
|
||||||
if len(raw) > 0 {
|
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)
|
log.Warn("imap scheduler: store/index failed", "err", err)
|
||||||
} else {
|
} else {
|
||||||
imported++
|
imported++
|
||||||
|
|||||||
@@ -119,7 +119,7 @@ func (imp *Importer) doImport(ctx context.Context, acc *Account, password string
|
|||||||
continue
|
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)
|
log.Warn("failed to store/index message, skipping", "msg_num", num, "err", err)
|
||||||
} else {
|
} else {
|
||||||
imported++
|
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.
|
// 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()
|
ctx := context.Background()
|
||||||
// Save to file storage (deduplicates by SHA256 automatically)
|
// Save to file storage (deduplicates by SHA256 automatically)
|
||||||
id, err := imp.mailStore.Save(ctx, raw, time.Now(), imp.TenantID)
|
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)
|
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
|
// Parse for indexing
|
||||||
pm, err := mailparser.Parse(raw)
|
pm, err := mailparser.Parse(raw)
|
||||||
if err != nil {
|
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.Received.Add(1)
|
||||||
s.daemon.stats.LastMailAt.Store(time.Now())
|
s.daemon.stats.LastMailAt.Store(time.Now())
|
||||||
s.daemon.logger.Info("SMTP: mail stored", "id", id, "from", s.from,
|
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;
|
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;
|
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
|
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 ───────────────────────────────────────────────────────
|
// ── Core operations ───────────────────────────────────────────────────────
|
||||||
|
|
||||||
// Save writes raw email bytes to storage. The ID is the hex-encoded SHA256 of
|
// 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 { Skeleton } from "@/components/ui/skeleton";
|
||||||
import { Separator } from "@/components/ui/separator";
|
import { Separator } from "@/components/ui/separator";
|
||||||
import { Alert, AlertDescription } from "@/components/ui/alert";
|
import { Alert, AlertDescription } from "@/components/ui/alert";
|
||||||
|
import { ReconciliationCard } from "@/components/admin/tabs/ReconciliationCard";
|
||||||
|
|
||||||
function formatBytes(bytes: number): string {
|
function formatBytes(bytes: number): string {
|
||||||
if (bytes < 1024) return `${bytes} B`;
|
if (bytes < 1024) return `${bytes} B`;
|
||||||
@@ -491,6 +492,9 @@ export function DashboardTab({
|
|||||||
</Card>
|
</Card>
|
||||||
)}
|
)}
|
||||||
|
|
||||||
|
{/* Vollständigkeits-Check (PROJ-52) — tenant-gescoped, domain_admin+ */}
|
||||||
|
<ReconciliationCard />
|
||||||
|
|
||||||
{/* Benutzerübersicht */}
|
{/* Benutzerübersicht */}
|
||||||
<Card>
|
<Card>
|
||||||
<CardContent className="pt-6 space-y-2">
|
<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,
|
deleteArchivingRule,
|
||||||
} from "./archiving_rules";
|
} from "./archiving_rules";
|
||||||
|
|
||||||
|
export type {
|
||||||
|
ReconciliationPoint,
|
||||||
|
ReconciliationSource,
|
||||||
|
ReconciliationResponse,
|
||||||
|
} from "./reconciliation";
|
||||||
|
export {
|
||||||
|
getReconciliation,
|
||||||
|
exportReconciliationCSV,
|
||||||
|
} from "./reconciliation";
|
||||||
|
|
||||||
export type { SavedSearch } from "./saved_searches";
|
export type { SavedSearch } from "./saved_searches";
|
||||||
export {
|
export {
|
||||||
listSavedSearches,
|
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