package syncalert import ( "context" _ "embed" "errors" "fmt" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) //go:embed migrations/0001_mail_sync_alert_state.sql var schemaMigration string const ( // NotificationChannel/AdminRecipient sind bewusst statisch (kleinste // Lösung) — eine konfigurierbare Empfängerverwaltung ist Sache einer // späteren Kachel, nicht Bestandteil von IMP-08. NotificationChannel = "mail-sync-failure" AdminRecipient = "mail-admins" ) // Monitor verfolgt Sync-Fehlschläge je Mandant/Postfach und löst bei // Überschreiten der Schwelle GENAU EINE Benachrichtigung aus // (Akzeptanzkriterium 1). type Monitor struct { pool *pgxpool.Pool dispatcher NotificationDispatcher threshold int now func() time.Time } // DefaultThreshold ist die Vorgabe-Eskalationsschwelle (konsekutive // Fehlschläge), überschreibbar über WithThreshold. const DefaultThreshold = 3 func NewMonitor(pool *pgxpool.Pool, dispatcher NotificationDispatcher) *Monitor { return &Monitor{pool: pool, dispatcher: dispatcher, threshold: DefaultThreshold, now: time.Now} } // WithThreshold setzt eine abweichende Eskalationsschwelle. func (m *Monitor) WithThreshold(threshold int) *Monitor { m.threshold = threshold return m } // EnsureSchema legt die Tabelle an, falls sie noch nicht existiert. func (m *Monitor) EnsureSchema(ctx context.Context) error { if _, err := m.pool.Exec(ctx, schemaMigration); err != nil { return fmt.Errorf("syncalert: schema anlegen: %w", err) } return nil } type alertState struct { consecutiveFailures int alerted bool lastSuccessAt *time.Time } func (m *Monitor) getOrCreate(ctx context.Context, tenantSlug, mailboxName string) (alertState, error) { if _, err := m.pool.Exec(ctx, ` INSERT INTO mail_sync_alert_state (tenant_slug, mailbox_name) VALUES ($1, $2) ON CONFLICT (tenant_slug, mailbox_name) DO NOTHING `, tenantSlug, mailboxName); err != nil { return alertState{}, fmt.Errorf("syncalert: zustand anlegen: %w", err) } var st alertState err := m.pool.QueryRow(ctx, ` SELECT consecutive_failures, alerted, last_success_at FROM mail_sync_alert_state WHERE tenant_slug = $1 AND mailbox_name = $2 `, tenantSlug, mailboxName).Scan(&st.consecutiveFailures, &st.alerted, &st.lastSuccessAt) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return alertState{}, fmt.Errorf("syncalert: gerade angelegten zustand nicht gefunden") } return alertState{}, fmt.Errorf("syncalert: zustand lesen: %w", err) } return st, nil } // RecordFailure verzeichnet einen fehlgeschlagenen Sync-Versuch. Erst // wenn consecutive_failures die konfigurierte Schwelle ERSTMALIG // erreicht (noch nicht "alerted"), wird GENAU EINE Benachrichtigung an // CFG-02 ausgelöst (Akzeptanzkriterium 1) — weitere Fehlschläge danach // lösen KEINE zusätzliche Benachrichtigung aus, solange der Alarmzustand // nicht durch einen erfolgreichen Sync zurückgesetzt wurde (kein // Einzel-Alarm pro Fehlversuch, keine Spam-Flut). func (m *Monitor) RecordFailure(ctx context.Context, tenantSlug, mailboxName, reason string) error { now := m.now() st, err := m.getOrCreate(ctx, tenantSlug, mailboxName) if err != nil { return err } newFailures := st.consecutiveFailures + 1 if _, err := m.pool.Exec(ctx, ` UPDATE mail_sync_alert_state SET consecutive_failures = $3, last_failure_reason = $4, last_failure_at = $5, updated_at = now() WHERE tenant_slug = $1 AND mailbox_name = $2 `, tenantSlug, mailboxName, newFailures, reason, now); err != nil { return fmt.Errorf("syncalert: fehlschlag erfassen: %w", err) } if newFailures < m.threshold || st.alerted { return nil } // Akzeptanzkriterium 2: Benachrichtigung enthält Postfach, // Fehlerursache und Zeitpunkt des letzten erfolgreichen Abrufs. payload := map[string]any{ "tenant_slug": tenantSlug, "mailbox": mailboxName, "reason": reason, "consecutive_failures": newFailures, "last_successful_sync": formatOptionalTime(st.lastSuccessAt), } if _, err := m.dispatcher.Enqueue(ctx, NotificationChannel, AdminRecipient, payload); err != nil { return fmt.Errorf("syncalert: benachrichtigung auslösen: %w", err) } if _, err := m.pool.Exec(ctx, ` UPDATE mail_sync_alert_state SET alerted = true, updated_at = now() WHERE tenant_slug = $1 AND mailbox_name = $2 `, tenantSlug, mailboxName); err != nil { return fmt.Errorf("syncalert: alarmzustand markieren: %w", err) } return nil } // RecordSuccess verzeichnet einen erfolgreichen Sync und setzt den // Alarmzustand zurück (Akzeptanzkriterium 3) — der nächste Fehlschlag // nach einem Erfolg beginnt wieder bei 0 konsekutiven Fehlschlägen. func (m *Monitor) RecordSuccess(ctx context.Context, tenantSlug, mailboxName string) error { now := m.now() if _, err := m.pool.Exec(ctx, ` INSERT INTO mail_sync_alert_state (tenant_slug, mailbox_name, consecutive_failures, alerted, last_success_at) VALUES ($1, $2, 0, false, $3) ON CONFLICT (tenant_slug, mailbox_name) DO UPDATE SET consecutive_failures = 0, alerted = false, last_success_at = $3, updated_at = now() `, tenantSlug, mailboxName, now); err != nil { return fmt.Errorf("syncalert: erfolg erfassen: %w", err) } return nil } func formatOptionalTime(t *time.Time) string { if t == nil { return "" } return t.UTC().Format(time.RFC3339) }