Maven/internal/store/reminders.go

package store

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/robfig/cron/v3"
)

// Reminder — user-stated future intent. fires once or recurring (if cron set).
type Reminder struct {
	ID         int64
	CreatedTs  time.Time
	FireTs     time.Time
	NextFireTs time.Time // computed next fire (for recurring) or same as FireTs
	Payload    string    // raw json
	Status     string    // pending | fired | cancelled
	Cron       string    // cron expression, empty for one-shot

	// DeliveryGroup and Phrase* are the durable presentation for this exact
	// occurrence. A collapsed catch-up bundle stores the same group and phrase
	// on every original reminder, so a retry (including after restart) says the
	// same thing without asking the model again. Rescheduling a recurring
	// reminder clears them for its next occurrence.
	DeliveryGroup        string
	PhraseBody           string
	PhraseSummary        string
	PhraseMood           string
	DeliveryAttempts     int
	NextAttemptTs        time.Time
	DeliveryBlockedTs    time.Time
	DeliveryBlockedError string

	// Collapsed — set only on a synthetic digest reminder (ID=0): the original
	// due reminders it stands in for. Not persisted. The dispatcher completes
	// (marks fired / reschedules) each of these after the digest delivers.
	Collapsed []Reminder
}

// Text — what the user actually asked for, out of the raw-JSON payload.
//
// The router's reminder slot extraction owns the payload shape and the
// conventional field is "text". A payload that is not JSON, or that lacks the
// field, is returned as-is: he said it, so they are his words, and showing
// them beats showing nothing.
//
// Here rather than in a caller because there is more than one caller and they
// disagreed. The phraser unwrapped the payload; the day plan did not, so
// "какие у меня планы на сегодня" recited a reminder as the literal string
// {"text":"..."} on the deployed daemon, 01-08-2026.
func (r Reminder) Text() string { return ReminderText(r.Payload) }

// ReminderText — Reminder.Text for callers holding a bare payload string.
func ReminderText(payload string) string {
	var m map[string]any
	if err := json.Unmarshal([]byte(payload), &m); err == nil {
		if t, ok := m["text"].(string); ok && t != "" {
			return t
		}
		if t, ok := m["text"]; ok {
			return fmt.Sprintf("%v", t)
		}
	}
	return strings.TrimSpace(payload)
}

// Reminder lifecycle states. Named for the same reason DigestStatus is: a
// caller filtering on the string literal "pending" is one typo away from a
// filter that silently matches nothing.
const (
	ReminderPending   = "pending"
	ReminderFired     = "fired"
	ReminderCancelled = "cancelled"

	// ReminderRetryBase and ReminderRetryMax bound the retry cadence. Attempts
	// continue indefinitely because a reminder must not disappear during a
	// long transport outage; only the delay stops growing.
	ReminderRetryBase = time.Minute
	ReminderRetryMax  = time.Hour
)

var (
	ErrReminderNotFound = errors.New("store: reminder not found")
	ErrReminderState    = errors.New("store: reminder not in a mutable state")
	ErrReminderInFlight = errors.New("store: reminder delivery already started")
	ErrReminderPhrase   = errors.New("store: reminder delivery phrase invalid")
)

const reminderColumns = `id, created_ts, fire_ts, next_fire_ts, payload, status, cron,
	 delivery_group, phrase_body, phrase_summary, phrase_mood, delivery_attempts, next_attempt_ts,
	 delivery_blocked_ts, delivery_blocked_error`

func scanReminder(sc scanner) (Reminder, error) {
	var r Reminder
	var created, fire, nextFire int64
	var cron *string
	var nextAttempt *int64
	var blocked *int64
	if err := sc.Scan(
		&r.ID, &created, &fire, &nextFire, &r.Payload, &r.Status, &cron,
		&r.DeliveryGroup, &r.PhraseBody, &r.PhraseSummary, &r.PhraseMood,
		&r.DeliveryAttempts, &nextAttempt, &blocked, &r.DeliveryBlockedError,
	); err != nil {
		return Reminder{}, err
	}
	r.CreatedTs = time.UnixMilli(created).UTC()
	r.FireTs = time.UnixMilli(fire).UTC()
	r.NextFireTs = time.UnixMilli(nextFire).UTC()
	if cron != nil {
		r.Cron = *cron
	}
	if nextAttempt != nil {
		r.NextAttemptTs = time.UnixMilli(*nextAttempt).UTC()
	}
	if blocked != nil {
		r.DeliveryBlockedTs = time.UnixMilli(*blocked).UTC()
	}
	return r, nil
}

// CreateReminder persists a reminder with a resolved absolute fire time.
// The caller (router/capture path) MUST have already converted "in 4h" → now+4h.
// We do not accept strings here. cron is a cron expression for recurring
// reminders; empty for one-shot.
func (s *Store) CreateReminder(ctx context.Context, fire time.Time, payload, cron string) (int64, error) {
	now := time.Now().UTC()
	var cronPtr *string
	if cron != "" {
		cronPtr = &cron
	}
	res, err := s.db.ExecContext(ctx,
		`INSERT INTO reminders (created_ts, fire_ts, next_fire_ts, payload, status, cron) VALUES (?,?,?,?, 'pending', ?)`,
		now.UnixMilli(), fire.UnixMilli(), fire.UnixMilli(), payload, cronPtr)
	if err != nil {
		return 0, fmt.Errorf("create reminder: %w", err)
	}
	id, err := res.LastInsertId()
	if err != nil {
		return 0, fmt.Errorf("create reminder: last insert id: %w", err)
	}
	return id, nil
}

// DueReminders returns pending reminders with next_fire_ts <= now, oldest first.
// This is the predicate input from the loop side: `next_fire_ts <= now AND status='pending'`.
func (s *Store) DueReminders(ctx context.Context, now time.Time) ([]Reminder, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT `+reminderColumns+`
		  FROM reminders
		 WHERE status = 'pending' AND next_fire_ts <= ?
		   AND (next_attempt_ts IS NULL OR next_attempt_ts <= ?)
		   AND delivery_blocked_ts IS NULL
		   AND NOT EXISTS (
		       SELECT 1
		         FROM delivery_attempts AS attempt
		        WHERE attempt.kind = 'reminder'
		          AND attempt.delivery_group = reminders.delivery_group
		          AND reminders.delivery_group <> ''
		          AND attempt.status IN ('pending', 'sent', 'unknown')
		   )
		 ORDER BY next_fire_ts ASC, id ASC`, now.UnixMilli(), now.UnixMilli())
	if err != nil {
		return nil, fmt.Errorf("due reminders: %w", err)
	}
	defer rows.Close()
	var out []Reminder
	for rows.Next() {
		r, err := scanReminder(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

// PendingReminders returns the pending reminders whose next fire time falls in
// [from, to), earliest first.
//
// The day plan used to take the newest 500 rows out of ListReminders, which
// orders by creation, and then filter them by day. A reminder stated long ago
// for today fell off the end of that scan while a reminder stated this morning
// for next year stayed on it. Bounding by fire time drops what is out of range
// instead of what is old.
func (s *Store) PendingReminders(ctx context.Context, from, to time.Time) ([]Reminder, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT `+reminderColumns+`
		  FROM reminders
		 WHERE status = ? AND next_fire_ts >= ? AND next_fire_ts < ?
		 ORDER BY next_fire_ts ASC, id ASC`,
		ReminderPending, from.UnixMilli(), to.UnixMilli())
	if err != nil {
		return nil, fmt.Errorf("pending reminders: %w", err)
	}
	defer rows.Close()
	var out []Reminder
	for rows.Next() {
		r, err := scanReminder(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

// MarkReminder records successful completion of a reminder delivery. The only
// valid transition is pending→fired. Cancellation has stronger outbox and
// collapsed-group invariants and must go through CancelReminder; accepting the
// same status here would leave a legacy bypass around those invariants.
func (s *Store) MarkReminder(ctx context.Context, id int64, status string) error {
	if status != ReminderFired {
		return fmt.Errorf("%w: %s", ErrReminderState, status)
	}
	// The old read-then-write transition allowed two callers to both observe
	// pending and both report success. Keeping the source state in the UPDATE
	// predicate makes pending → fired one atomic contest (V-678).
	res, err := s.db.ExecContext(ctx,
		"UPDATE reminders SET status = ? WHERE id = ? AND status = ?",
		status, id, ReminderPending)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("mark reminder: rows affected: %w", err)
	}
	if n == 1 {
		return nil
	}

	// Preserve the public distinction between a missing id and a completed
	// state without weakening the atomic transition above.
	var current string
	err = s.db.QueryRowContext(ctx, "SELECT status FROM reminders WHERE id = ?", id).Scan(&current)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrReminderNotFound
	}
	if err != nil {
		return err
	}
	return fmt.Errorf("%w: currently %s", ErrReminderState, current)
}

// CancelReminder atomically wins against the beginning of external delivery.
// A pending/sent/unknown outbox row means the presentation may already be
// outside the process, so claiming cancellation would be false. Definite
// failures do not block cancellation.
//
// A collapsed catch-up bundle shares one cached phrase. Cancelling any member
// invalidates that presentation on every still-pending sibling; otherwise the
// next retry could continue saying "three reminders" after one was removed.
func (s *Store) CancelReminder(ctx context.Context, id int64) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("cancel reminder: begin: %w", err)
	}
	defer tx.Rollback()

	var status, group string
	var nextFire int64
	err = tx.QueryRowContext(ctx,
		`SELECT status, delivery_group, next_fire_ts FROM reminders WHERE id = ?`, id,
	).Scan(&status, &group, &nextFire)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrReminderNotFound
	}
	if err != nil {
		return fmt.Errorf("cancel reminder %d: read: %w", id, err)
	}
	if status != ReminderPending {
		return fmt.Errorf("%w: currently %s", ErrReminderState, status)
	}

	// New attempts carry an occurrence-scoped delivery group. Migration 25
	// assigned older rows the empty group, so those can only be related to this
	// occurrence when they began at or after its next-fire boundary. Without
	// that bound, one successful delivery from a recurring reminder's history
	// would make the whole series permanently uncancellable after an upgrade.
	var live int
	err = tx.QueryRowContext(ctx, `
		SELECT COUNT(*)
		  FROM delivery_attempts
		 WHERE kind = 'reminder'
		   AND status IN ('pending', 'sent', 'unknown')
		   AND ((delivery_group <> '' AND delivery_group = ?)
		        OR (delivery_group = '' AND reminder_id = ? AND created_ts >= ?))`,
		group, id, nextFire).Scan(&live)
	if err != nil {
		return fmt.Errorf("cancel reminder %d: inspect delivery: %w", id, err)
	}
	if live > 0 {
		return ErrReminderInFlight
	}

	if group != "" {
		if _, err := tx.ExecContext(ctx, `
			UPDATE reminders
			   SET delivery_group = '', phrase_body = '', phrase_summary = '', phrase_mood = ''
			 WHERE delivery_group = ? AND status = ?`, group, ReminderPending); err != nil {
			return fmt.Errorf("cancel reminder %d: invalidate group: %w", id, err)
		}
	}
	res, err := tx.ExecContext(ctx, `
		UPDATE reminders
		   SET status = ?, delivery_group = '', phrase_body = '', phrase_summary = '', phrase_mood = '',
		       next_attempt_ts = NULL, delivery_blocked_ts = NULL, delivery_blocked_error = ''
		 WHERE id = ? AND status = ?`, ReminderCancelled, id, ReminderPending)
	if err != nil {
		return fmt.Errorf("cancel reminder %d: %w", id, err)
	}
	if err := requireOneReminderRow(res, id); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("cancel reminder %d: commit: %w", id, err)
	}
	return nil
}

// ListReminders returns the n most recent reminders, newest first.
func (s *Store) ListReminders(ctx context.Context, n int) ([]Reminder, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT `+reminderColumns+`
		  FROM reminders ORDER BY created_ts DESC, id DESC LIMIT ?`, n)
	if err != nil {
		return nil, fmt.Errorf("list reminders: %w", err)
	}
	defer rows.Close()
	var out []Reminder
	for rows.Next() {
		r, err := scanReminder(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

// ListPendingReminders returns cancellable reminders in firing order. n <= 0
// means all pending rows: spoken resolution must not miss an old reminder just
// because newer fired history filled ListReminders' window.
func (s *Store) ListPendingReminders(ctx context.Context, n int) ([]Reminder, error) {
	query := `SELECT ` + reminderColumns + `
		  FROM reminders
		 WHERE status = ?
		 ORDER BY next_fire_ts ASC, id ASC`
	args := []any{ReminderPending}
	if n > 0 {
		query += ` LIMIT ?`
		args = append(args, n)
	}
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("list pending reminders: %w", err)
	}
	defer rows.Close()
	var out []Reminder
	for rows.Next() {
		r, err := scanReminder(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

// HasDeliveryPhrase reports whether this occurrence already has a durable
// presentation. Summary may intentionally be empty (the delivery boundary has
// a generic privacy-preserving fallback), so Body is the readiness marker.
func (r Reminder) HasDeliveryPhrase() bool {
	return r.DeliveryGroup != "" && r.PhraseBody != ""
}

// ReminderRetryDelay returns the delay after attempt (one-based). It grows
// exponentially from one minute and stays at one hour; retries do not stop.
func ReminderRetryDelay(attempt int) time.Duration {
	if attempt <= 1 {
		return ReminderRetryBase
	}
	delay := ReminderRetryBase
	for i := 1; i < attempt && delay < ReminderRetryMax; i++ {
		if delay >= ReminderRetryMax/2 {
			return ReminderRetryMax
		}
		delay *= 2
	}
	if delay > ReminderRetryMax {
		return ReminderRetryMax
	}
	return delay
}

// CacheReminderPhrase stores one presentation on every original represented
// by a reminder delivery. For a collapsed bundle, originals contains every
// row in Reminder.Collapsed and group is shared across all of them.
//
// The occurrence timestamp and empty-body predicates keep this from attaching
// an old phrase to a recurring reminder's next occurrence or overwriting a
// phrase another delivery already claimed. The transaction prevents a partial
// bundle cache: after a crash either every original can reconstruct the bundle
// or none can.
func (s *Store) CacheReminderPhrase(
	ctx context.Context,
	originals []Reminder,
	group, body, summary, mood string,
) error {
	if len(originals) == 0 || group == "" || body == "" {
		return ErrReminderPhrase
	}
	if mood == "" {
		mood = "neutral"
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("cache reminder phrase: begin: %w", err)
	}
	defer tx.Rollback()

	seen := make(map[int64]struct{}, len(originals))
	for _, r := range originals {
		if r.ID <= 0 || r.NextFireTs.IsZero() {
			return ErrReminderPhrase
		}
		if _, ok := seen[r.ID]; ok {
			return ErrReminderPhrase
		}
		seen[r.ID] = struct{}{}
		res, err := tx.ExecContext(ctx, `
			UPDATE reminders
			   SET delivery_group = ?, phrase_body = ?, phrase_summary = ?, phrase_mood = ?
			 WHERE id = ? AND status = ? AND next_fire_ts = ? AND phrase_body = ''`,
			group, body, summary, mood, r.ID, ReminderPending, r.NextFireTs.UnixMilli())
		if err != nil {
			return fmt.Errorf("cache reminder phrase %d: %w", r.ID, err)
		}
		n, err := res.RowsAffected()
		if err != nil {
			return fmt.Errorf("cache reminder phrase %d: rows affected: %w", r.ID, err)
		}
		if n != 1 {
			return fmt.Errorf("%w: reminder %d is no longer an unphrased pending occurrence", ErrReminderState, r.ID)
		}
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("cache reminder phrase: commit: %w", err)
	}
	return nil
}

// ScheduleReminderRetry moves every still-pending original into a retry wait.
// Terminal rows are skipped: cancellation winning while a send was in flight
// must not be resurrected. A recurring row whose occurrence changed is skipped
// for the same reason. All remaining originals get their own persisted attempt
// count; normally a collapsed bundle keeps those counts in lockstep.
func (s *Store) ScheduleReminderRetry(ctx context.Context, originals []Reminder, now time.Time) error {
	if len(originals) == 0 {
		return ErrReminderNotFound
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("schedule reminder retry: begin: %w", err)
	}
	defer tx.Rollback()

	seen := make(map[int64]struct{}, len(originals))
	for _, expected := range originals {
		if expected.ID <= 0 || expected.NextFireTs.IsZero() {
			return ErrReminderNotFound
		}
		if _, ok := seen[expected.ID]; ok {
			continue
		}
		seen[expected.ID] = struct{}{}

		var status string
		var nextFire int64
		var attempts int
		var blocked *int64
		err := tx.QueryRowContext(ctx,
			`SELECT status, next_fire_ts, delivery_attempts, delivery_blocked_ts FROM reminders WHERE id = ?`,
			expected.ID).Scan(&status, &nextFire, &attempts, &blocked)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrReminderNotFound
		}
		if err != nil {
			return fmt.Errorf("schedule reminder retry %d: read: %w", expected.ID, err)
		}
		if status != ReminderPending || nextFire != expected.NextFireTs.UnixMilli() || blocked != nil {
			continue
		}
		if expected.DeliveryGroup != "" {
			var ambiguous int
			if err := tx.QueryRowContext(ctx, `
				SELECT EXISTS (
					SELECT 1 FROM delivery_attempts
					 WHERE kind = 'reminder' AND delivery_group = ?
					   AND status IN ('pending', 'sent', 'unknown')
				)`, expected.DeliveryGroup).Scan(&ambiguous); err != nil {
				return fmt.Errorf("schedule reminder retry %d: inspect outbox: %w", expected.ID, err)
			}
			if ambiguous != 0 {
				// The sink may have accepted this presentation. Leave the reminder
				// pending but ineligible for automatic retry; operator resolution
				// must not be replaced with a guessed duplicate.
				continue
			}
		}

		attempts++
		nextAttempt := now.Add(ReminderRetryDelay(attempts)).UnixMilli()
		res, err := tx.ExecContext(ctx, `
			UPDATE reminders
			   SET delivery_attempts = ?, next_attempt_ts = ?
			 WHERE id = ? AND status = ? AND next_fire_ts = ? AND delivery_attempts = ?`,
			attempts, nextAttempt, expected.ID, ReminderPending, nextFire, attempts-1)
		if err != nil {
			return fmt.Errorf("schedule reminder retry %d: %w", expected.ID, err)
		}
		n, err := res.RowsAffected()
		if err != nil {
			return fmt.Errorf("schedule reminder retry %d: rows affected: %w", expected.ID, err)
		}
		if n != 1 {
			return fmt.Errorf("%w: reminder %d changed concurrently", ErrReminderState, expected.ID)
		}
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("schedule reminder retry: commit: %w", err)
	}
	return nil
}

// BlockReminderDelivery records a permanent transport/configuration refusal on
// every original represented by one presentation. Blocked rows stay pending and
// visible, but never retry automatically. UnblockReminderDelivery is the
// deliberate recovery path after credentials or policy are repaired.
func (s *Store) BlockReminderDelivery(ctx context.Context, originals []Reminder, now time.Time, reason string) error {
	if len(originals) == 0 || strings.TrimSpace(reason) == "" {
		return ErrReminderState
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("block reminder delivery: begin: %w", err)
	}
	defer tx.Rollback()
	seen := make(map[int64]struct{}, len(originals))
	for _, expected := range originals {
		if expected.ID <= 0 || expected.NextFireTs.IsZero() {
			return ErrReminderNotFound
		}
		if _, ok := seen[expected.ID]; ok {
			continue
		}
		seen[expected.ID] = struct{}{}
		res, err := tx.ExecContext(ctx, `
			UPDATE reminders
			   SET delivery_blocked_ts = ?, delivery_blocked_error = ?, next_attempt_ts = NULL
			 WHERE id = ? AND status = ? AND next_fire_ts = ?`,
			now.UnixMilli(), reason, expected.ID, ReminderPending, expected.NextFireTs.UnixMilli())
		if err != nil {
			return fmt.Errorf("block reminder %d: %w", expected.ID, err)
		}
		if err := requireOneReminderRow(res, expected.ID); err != nil {
			return err
		}
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("block reminder delivery: commit: %w", err)
	}
	return nil
}

func (s *Store) UnblockReminderDelivery(ctx context.Context, id int64) error {
	res, err := s.db.ExecContext(ctx, `
		UPDATE reminders
		   SET delivery_blocked_ts = NULL, delivery_blocked_error = '', next_attempt_ts = NULL
		 WHERE id = ? AND status = ? AND delivery_blocked_ts IS NOT NULL`, id, ReminderPending)
	if err != nil {
		return fmt.Errorf("unblock reminder %d: %w", id, err)
	}
	return requireOneReminderRow(res, id)
}

// CompleteReminderDelivery applies the bookkeeping for every original covered
// by one successful external send in a single transaction. One-shot reminders
// become fired; recurring reminders advance to their next occurrence and shed
// the old presentation/retry state. The all-or-nothing boundary prevents a
// collapsed digest from becoming half-fired and then being repeated.
func (s *Store) CompleteReminderDelivery(ctx context.Context, originals []Reminder, now time.Time) error {
	return s.completeReminderDeliveryIn(ctx, 0, originals, now, time.Local)
}

// CompleteSuccessfulReminderAttempt atomically closes a definitely successful
// external attempt and advances every reminder occurrence it represented. If
// this transaction cannot commit, the attempt remains pending; startup then
// reconciles it to unknown, and the occurrence is held from automatic replay.
func (s *Store) CompleteSuccessfulReminderAttempt(ctx context.Context, attemptID int64, originals []Reminder, now time.Time) error {
	if attemptID <= 0 {
		return fmt.Errorf("complete reminder delivery: invalid attempt id %d", attemptID)
	}
	return s.completeReminderDeliveryIn(ctx, attemptID, originals, now, time.Local)
}

func (s *Store) completeReminderDeliveryIn(ctx context.Context, attemptID int64, originals []Reminder, now time.Time, loc *time.Location) error {
	if len(originals) == 0 {
		return ErrReminderNotFound
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("complete reminder delivery: begin: %w", err)
	}
	defer tx.Rollback()
	if attemptID != 0 {
		res, err := tx.ExecContext(ctx,
			`UPDATE delivery_attempts SET status = ?, completed_ts = ? WHERE id = ? AND status = ?`,
			DeliverySent, now.UnixMilli(), attemptID, DeliveryPending)
		if err != nil {
			return fmt.Errorf("complete reminder delivery attempt %d: %w", attemptID, err)
		}
		n, err := res.RowsAffected()
		if err != nil {
			return fmt.Errorf("complete reminder delivery attempt %d: rows affected: %w", attemptID, err)
		}
		if n != 1 {
			return fmt.Errorf("complete reminder delivery attempt %d: not pending", attemptID)
		}
	}

	seen := make(map[int64]struct{}, len(originals))
	for _, expected := range originals {
		if expected.ID <= 0 || expected.NextFireTs.IsZero() {
			return ErrReminderNotFound
		}
		if _, ok := seen[expected.ID]; ok {
			return fmt.Errorf("%w: duplicate reminder %d in one delivery", ErrReminderState, expected.ID)
		}
		seen[expected.ID] = struct{}{}

		row := tx.QueryRowContext(ctx, `SELECT `+reminderColumns+` FROM reminders WHERE id = ?`, expected.ID)
		current, err := scanReminder(row)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrReminderNotFound
		}
		if err != nil {
			return fmt.Errorf("complete reminder %d: read: %w", expected.ID, err)
		}
		if current.Status != ReminderPending || !current.NextFireTs.Equal(expected.NextFireTs) {
			return fmt.Errorf("%w: reminder %d occurrence changed", ErrReminderState, expected.ID)
		}
		if expected.DeliveryGroup != "" && current.DeliveryGroup != expected.DeliveryGroup {
			return fmt.Errorf("%w: reminder %d delivery group changed", ErrReminderState, expected.ID)
		}

		if current.Cron == "" {
			if err := completeOneShotReminderTx(ctx, tx, current); err != nil {
				return err
			}
			continue
		}
		next, err := nextReminderOccurrence(current, now, loc)
		if err != nil {
			return err
		}
		if next.IsZero() {
			if err := completeOneShotReminderTx(ctx, tx, current); err != nil {
				return err
			}
			continue
		}
		res, err := tx.ExecContext(ctx, `
			UPDATE reminders
			   SET next_fire_ts = ?, delivery_group = '', phrase_body = '',
			       phrase_summary = '', phrase_mood = '', delivery_attempts = 0,
			       next_attempt_ts = NULL, delivery_blocked_ts = NULL,
			       delivery_blocked_error = ''
			 WHERE id = ? AND status = ? AND next_fire_ts = ?`,
			next.UnixMilli(), current.ID, ReminderPending, current.NextFireTs.UnixMilli())
		if err != nil {
			return fmt.Errorf("complete recurring reminder %d: %w", current.ID, err)
		}
		if err := requireOneReminderRow(res, current.ID); err != nil {
			return err
		}
	}

	if err := tx.Commit(); err != nil {
		return fmt.Errorf("complete reminder delivery: commit: %w", err)
	}
	return nil
}

func completeOneShotReminderTx(ctx context.Context, tx *sql.Tx, r Reminder) error {
	res, err := tx.ExecContext(ctx,
		`UPDATE reminders SET status = ? WHERE id = ? AND status = ? AND next_fire_ts = ?`,
		ReminderFired, r.ID, ReminderPending, r.NextFireTs.UnixMilli())
	if err != nil {
		return fmt.Errorf("complete reminder %d: %w", r.ID, err)
	}
	return requireOneReminderRow(res, r.ID)
}

func requireOneReminderRow(res sql.Result, id int64) error {
	n, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("complete reminder %d: rows affected: %w", id, err)
	}
	if n != 1 {
		return fmt.Errorf("%w: reminder %d changed concurrently", ErrReminderState, id)
	}
	return nil
}

// RescheduleReminder computes the next fire time for a recurring reminder and
// updates next_fire_ts. Returns ErrReminderState if the reminder is not
// recurring or not pending. If the schedule yields no further fire time at all,
// marks it fired.
//
// Two things the first version got wrong, both fixed 06-08-2026 (V-616).
//
// The cron expression is a WALL CLOCK statement — "0 9 * * *" is nine in the
// morning where the owner stands — but scanReminder hands back instants in UTC,
// and robfig's Next walks the calendar in the location of the time it is given.
// Computing from a UTC instant therefore produced the next 09:00 UTC, so the
// second occurrence of a daily reminder landed one UTC offset late and stayed
// there: 12:00 for a Moscow owner. Everything is converted to loc first, which
// also makes the walk DST-correct — the schedule keeps its wall-clock hour
// across a changeover instead of drifting an hour with the offset.
//
// And a missed occurrence used to KILL the reminder: any next fire earlier than
// now marked it fired, so a daemon down overnight ended a daily standup forever.
// Occurrences in the past are skipped instead, so the reminder rolls forward to
// the first one strictly after now. Skipping and not replaying is deliberate:
// the same no-backlog rule routine.DueAccepted follows.
func (s *Store) RescheduleReminder(ctx context.Context, id int64, now time.Time) error {
	return s.rescheduleReminderIn(ctx, id, now, time.Local)
}

func (s *Store) rescheduleReminderIn(ctx context.Context, id int64, now time.Time, loc *time.Location) error {
	row := s.db.QueryRowContext(ctx, `SELECT `+reminderColumns+`
		  FROM reminders WHERE id = ?`, id)
	r, err := scanReminder(row)
	if err != nil {
		return err
	}
	if r.Cron == "" {
		return fmt.Errorf("%w: not a recurring reminder", ErrReminderState)
	}
	if r.Status != "pending" {
		return fmt.Errorf("%w: currently %s", ErrReminderState, r.Status)
	}

	next, err := nextReminderOccurrence(r, now, loc)
	if err != nil {
		return err
	}
	if next.IsZero() {
		return s.MarkReminder(ctx, id, ReminderFired)
	}
	res, err := s.db.ExecContext(ctx, `
		UPDATE reminders
		   SET next_fire_ts = ?, delivery_group = '', phrase_body = '',
		       phrase_summary = '', phrase_mood = '', delivery_attempts = 0,
		       next_attempt_ts = NULL, delivery_blocked_ts = NULL,
		       delivery_blocked_error = ''
		 WHERE id = ? AND status = ? AND next_fire_ts = ?`,
		next.UnixMilli(), id, ReminderPending, r.NextFireTs.UnixMilli())
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("reschedule reminder: rows affected: %w", err)
	}
	if n != 1 {
		return fmt.Errorf("%w: reminder %d changed concurrently", ErrReminderState, id)
	}
	return nil
}

func nextReminderOccurrence(r Reminder, now time.Time, loc *time.Location) (time.Time, error) {
	sched, err := cron.ParseStandard(r.Cron)
	if err != nil {
		return time.Time{}, fmt.Errorf("parse cron %q: %w", r.Cron, err)
	}
	// Next is strictly after the time it is given, so the last fire cannot be
	// returned again and no fudge minute is needed. The bound stops a schedule
	// that somehow yields a non-advancing time from spinning here.
	next := sched.Next(r.NextFireTs.In(loc))
	for i := 0; i < 4096 && !next.IsZero() && !next.After(now); i++ {
		next = sched.Next(next)
	}
	if next.IsZero() || !next.After(now) {
		return time.Time{}, nil
	}
	return next, nil
}