Maven/internal/store/delivery.go

package store

import (
	"context"
	"database/sql"
	"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).
// deliveryGroup is the exact persisted reminder occurrence or collapsed
// bundle; it is empty for nudges and legacy reminder attempts.
// 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, deliveryGroup, channel, bodyHash string, now time.Time) (int64, error) {
	if kind == "reminder" && deliveryGroup != "" {
		return s.beginReminderDeliveryAttempt(ctx, rule, reminderID, deliveryGroup, channel, bodyHash, now)
	}
	res, err := s.db.ExecContext(ctx,
		`INSERT INTO delivery_attempts (kind, rule, reminder_id, delivery_group, channel, body_hash, status, created_ts)
		 VALUES (?, ?, ?, ?, ?, ?, 'pending', ?)`,
		kind, rule, reminderID, deliveryGroup, 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
}

// beginReminderDeliveryAttempt serializes the last local cancellation point
// with the first externally ambiguous delivery point. CancelReminder clears a
// group's identity when it wins first; when this insert wins first it leaves a
// pending attempt that makes cancellation refuse. There is therefore no state
// in which both operations report success.
func (s *Store) beginReminderDeliveryAttempt(ctx context.Context, rule string, reminderID int64, deliveryGroup, channel, bodyHash string, now time.Time) (int64, error) {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return 0, fmt.Errorf("begin reminder delivery attempt: begin: %w", err)
	}
	defer tx.Rollback()

	var status, currentGroup string
	err = tx.QueryRowContext(ctx,
		`SELECT status, delivery_group FROM reminders WHERE id = ?`, reminderID,
	).Scan(&status, &currentGroup)
	if err != nil {
		if err == sql.ErrNoRows {
			return 0, ErrReminderNotFound
		}
		return 0, fmt.Errorf("begin reminder delivery attempt: read reminder %d: %w", reminderID, err)
	}
	if status != ReminderPending || currentGroup != deliveryGroup {
		return 0, fmt.Errorf("%w: reminder %d no longer owns delivery group", ErrReminderState, reminderID)
	}
	var live int
	if err := tx.QueryRowContext(ctx, `
		SELECT COUNT(*) FROM delivery_attempts
		 WHERE kind = 'reminder' AND delivery_group = ?
		   AND status IN ('pending', 'sent', 'unknown')`, deliveryGroup).Scan(&live); err != nil {
		return 0, fmt.Errorf("begin reminder delivery attempt: inspect group: %w", err)
	}
	if live > 0 {
		return 0, ErrReminderInFlight
	}
	res, err := tx.ExecContext(ctx, `
		INSERT INTO delivery_attempts (kind, rule, reminder_id, delivery_group, channel, body_hash, status, created_ts)
		VALUES ('reminder', ?, ?, ?, ?, ?, 'pending', ?)`,
		rule, reminderID, deliveryGroup, channel, bodyHash, now.UnixMilli())
	if err != nil {
		return 0, fmt.Errorf("begin reminder delivery attempt: %w", err)
	}
	id, err := res.LastInsertId()
	if err != nil {
		return 0, fmt.Errorf("begin reminder delivery attempt: last insert id: %w", err)
	}
	if err := tx.Commit(); err != nil {
		return 0, fmt.Errorf("begin reminder delivery attempt: commit: %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
}

// DeliveryAttempt — one row of the outbox, as a reader sees it.
type DeliveryAttempt struct {
	ID            int64
	Kind          string // nudge|reminder
	Rule          string // set for nudges
	ReminderID    int64  // set for reminders
	DeliveryGroup string // exact reminder occurrence/bundle; empty for nudges and legacy rows
	Channel       string
	Status        string // one of the Delivery* constants
	Created       time.Time
	Completed     time.Time // zero while pending
	HasComplete   bool
}

// ListDeliveryAttempts returns recent attempts, newest first. An empty status
// means every status; anything else filters on it.
//
// The table was write-only until 04-08-2026: rows were recorded and nothing
// could read them, so the tests for #368 and #370 had to reach past the store
// into store.DB, which is the tell (Vikunja #390). A durable record nobody can
// read answers no question, and "why did Maven go quiet" is supposed to be a
// query rather than a mystery.
//
// Status is the filter that earns its place, because the two questions actually
// asked are "what got dropped" and "what is still pending". Neither is
// answerable by reading the whole list on a busy day.
func (s *Store) ListDeliveryAttempts(ctx context.Context, status string, limit int) ([]DeliveryAttempt, error) {
	if limit <= 0 {
		limit = 50
	}
	q := `SELECT id, kind, rule, reminder_id, delivery_group, channel, status, created_ts, completed_ts
	        FROM delivery_attempts`
	args := []any{}
	if status != "" {
		q += ` WHERE status = ?`
		args = append(args, status)
	}
	q += ` ORDER BY created_ts DESC, id DESC LIMIT ?`
	args = append(args, limit)

	rows, err := s.db.QueryContext(ctx, q, args...)
	if err != nil {
		return nil, fmt.Errorf("list delivery attempts: %w", err)
	}
	defer rows.Close()

	var out []DeliveryAttempt
	for rows.Next() {
		var a DeliveryAttempt
		var created int64
		var completed *int64
		if err := rows.Scan(&a.ID, &a.Kind, &a.Rule, &a.ReminderID, &a.DeliveryGroup, &a.Channel, &a.Status, &created, &completed); err != nil {
			return nil, fmt.Errorf("list delivery attempts: scan: %w", err)
		}
		a.Created = time.UnixMilli(created)
		if completed != nil {
			a.Completed, a.HasComplete = time.UnixMilli(*completed), true
		}
		out = append(out, a)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list delivery attempts: %w", err)
	}
	return out, nil
}