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:
sysops
co-authored by
Claude Sonnet 5
Claude Sonnet 5
parent
b286352d07
commit
be93614c9f
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user