diff --git a/cmd/mavend/main.go b/cmd/mavend/main.go index 747da94..d6c6ea2 100644 --- a/cmd/mavend/main.go +++ b/cmd/mavend/main.go @@ -319,6 +319,13 @@ func run(args []string) error { if voiceW != nil { voiceSink = voiceW.voiceSink } + // A crashed prior run may have left "pending" delivery attempts (send + // may have landed externally, then the process died before recording + // it) — reconcile them to "unknown" before the tick loop resumes + // sending, so nothing auto-resends into that ambiguity. + if _, err := st.ReconcileStaleDeliveryAttempts(context.Background(), time.Now()); err != nil { + log.Printf("delivery outbox reconcile: %v", err) + } dispatcher = delivery.NewDispatcher(delivery.Config{ Ntfy: ntfy, Telegram: telegram, @@ -326,6 +333,7 @@ func run(args []string) error { Ack: st, Nudges: st, Reminders: st, + Outbox: st, }) // tick loop @@ -475,6 +483,9 @@ func run(args []string) error { if voiceW != nil { voiceSink = voiceW.voiceSink } + if _, err := st.ReconcileStaleDeliveryAttempts(context.Background(), time.Now()); err != nil { + log.Printf("delivery outbox reconcile: %v", err) + } dispatcher = delivery.NewDispatcher(delivery.Config{ Ntfy: ntfy, Telegram: telegram, @@ -482,6 +493,7 @@ func run(args []string) error { Ack: st, Nudges: st, Reminders: st, + Outbox: st, }) tickInterval := time.Duration(cfg.TickInterval) diff --git a/internal/store/delivery.go b/internal/store/delivery.go new file mode 100644 index 0000000..8c37b14 --- /dev/null +++ b/internal/store/delivery.go @@ -0,0 +1,84 @@ +package store + +import ( + "context" + "fmt" + "log" + "time" +) + +// DeliveryAttempt statuses. pending = Begin recorded, no Complete yet — either +// still in flight or the process died mid-send (crash window the outbox +// exists to close). sent/failed = Complete recorded the sink's outcome. +// unknown = a pending row found stale at startup: the process that started it +// is gone, and the send may or may not have reached the external channel. +// Never auto-resolved into sent or failed — that would be guessing. +const ( + DeliveryPending = "pending" + DeliverySent = "sent" + DeliveryFailed = "failed" + DeliveryUnknown = "unknown" +) + +// BeginDeliveryAttempt durably records intent to send BEFORE the external +// send happens, so a crash between "sent externally" and "recorded" leaves a +// trace instead of silence. kind is "nudge" or "reminder"; rule is set for +// nudges, reminderID for reminders (the other left at its zero value). +// bodyHash is an opaque caller-computed key (e.g. sha256 of channel+body) — +// stored for post-crash operator triage, not enforced as a uniqueness +// constraint (a rule/reminder legitimately re-sends across ticks). +func (s *Store) BeginDeliveryAttempt(ctx context.Context, kind, rule string, reminderID int64, channel, bodyHash string, now time.Time) (int64, error) { + res, err := s.db.ExecContext(ctx, + `INSERT INTO delivery_attempts (kind, rule, reminder_id, channel, body_hash, status, created_ts) + VALUES (?, ?, ?, ?, ?, 'pending', ?)`, + kind, rule, reminderID, channel, bodyHash, now.UnixMilli()) + if err != nil { + return 0, fmt.Errorf("begin delivery attempt: %w", err) + } + id, err := res.LastInsertId() + if err != nil { + return 0, fmt.Errorf("begin delivery attempt: last insert id: %w", err) + } + return id, nil +} + +// CompleteDeliveryAttempt records the sink's outcome for a prior +// BeginDeliveryAttempt. status is "sent" or "failed" — never "pending" or +// "unknown" (those are set only by Begin and reconciliation respectively). +func (s *Store) CompleteDeliveryAttempt(ctx context.Context, id int64, status string, now time.Time) error { + if status != DeliverySent && status != DeliveryFailed { + return fmt.Errorf("store: invalid delivery completion status %q", status) + } + _, err := s.db.ExecContext(ctx, + `UPDATE delivery_attempts SET status = ?, completed_ts = ? WHERE id = ? AND status = 'pending'`, + status, now.UnixMilli(), id) + if err != nil { + return fmt.Errorf("complete delivery attempt %d: %w", id, err) + } + return nil +} + +// ReconcileStaleDeliveryAttempts runs once at daemon startup, before the tick +// loop resumes sending. Any attempt still "pending" from a previous process +// life is the exact crash window the outbox exists to close: the external +// send may have landed and the process died before recording the outcome. +// Marking it "unknown" (rather than silently resending, and rather than +// silently dropping it) preserves the same never-guess-an-ambiguous-outcome +// rule as the IPC client and Hexis execution engine. Returns the count +// reconciled, for startup logging. +func (s *Store) ReconcileStaleDeliveryAttempts(ctx context.Context, now time.Time) (int, error) { + res, err := s.db.ExecContext(ctx, + `UPDATE delivery_attempts SET status = 'unknown', completed_ts = ? WHERE status = 'pending'`, + now.UnixMilli()) + if err != nil { + return 0, fmt.Errorf("reconcile stale delivery attempts: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return 0, fmt.Errorf("reconcile stale delivery attempts: rows affected: %w", err) + } + if n > 0 { + log.Printf("store: reconciled %d stale delivery attempt(s) from a prior run as outcome=unknown", n) + } + return int(n), nil +} diff --git a/internal/store/migrations.go b/internal/store/migrations.go index bd8c8b2..8221303 100644 --- a/internal/store/migrations.go +++ b/internal/store/migrations.go @@ -50,6 +50,19 @@ ALTER TABLE reminders ADD COLUMN next_fire_ts INTEGER;`, // #2 sent_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_ack_sends_rule ON ack_sends (rule, sent_at DESC);`, // #5 — sev4 telegram repeat-til-ack tracking + + `CREATE TABLE IF NOT EXISTS delivery_attempts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kind TEXT NOT NULL CHECK (kind IN ('nudge','reminder')), + rule TEXT NOT NULL DEFAULT '', + reminder_id INTEGER NOT NULL DEFAULT 0, + channel TEXT NOT NULL, + body_hash TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','sent','failed','unknown')), + created_ts INTEGER NOT NULL, + completed_ts INTEGER + ); + CREATE INDEX IF NOT EXISTS idx_delivery_attempts_status ON delivery_attempts (status);`, // #6 — durable delivery outbox } // migrate applies every migration with a number greater than the DB's current