Maven/internal/store/delivery.go

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.
// dropped = the routing table deliberately suppressed this one (a care nudge
// while you're away). Nothing was sent and nothing went wrong; the row exists
// so "she dropped it" and "the rule never fired" don't look the same later.
const (
	DeliveryPending = "pending"
	DeliverySent    = "sent"
	DeliveryFailed  = "failed"
	DeliveryUnknown = "unknown"
	DeliveryDropped = "dropped"
)

// 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", "failed" or "dropped" — 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 && status != DeliveryDropped {
		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
}