Maven/internal/delivery/dispatcher.go

package delivery

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"log"
	"time"

	"github.com/kami/maven/internal/loop"
	"github.com/kami/maven/internal/store"
)

// NudgeRecorder — the seam the store implements. the dispatcher records one
// nudge row per channel-sent (the nudges table IS the restraint memory + the
// feedback loop's only input). recording happens AFTER a successful send, so
// a failed send doesn't pollute the feedback signal with a phantom nudge —
// ignored_rate would drift on a row that never reached anyone.
type NudgeRecorder interface {
	RecordNudge(ctx context.Context, rule, channel, message string, ts time.Time) (int64, error)
}

// ReminderCompleter — the seam the store implements. Every original represented
// by one external delivery is completed in one transaction. That matters for a
// collapsed catch-up bundle: partially firing the originals would make the
// next tick repeat a presentation that the user already received.
type ReminderCompleter interface {
	CompleteReminderDelivery(ctx context.Context, originals []store.Reminder, now time.Time) error
}

// DurableReminderCompleter closes the successful outbox attempt and advances
// every reminder occurrence in one local transaction. The external send and
// local commit cannot be one transaction, but the local half must be: a crash
// between `attempt=sent` and `reminder=fired` otherwise strands the reminder in
// a permanently suppressed state.
type DurableReminderCompleter interface {
	CompleteSuccessfulReminderAttempt(ctx context.Context, attemptID int64, originals []store.Reminder, now time.Time) error
}

type ReminderBlocker interface {
	BlockReminderDelivery(ctx context.Context, originals []store.Reminder, now time.Time, reason string) error
}

// Outbox — the durable delivery ledger. Begin is recorded BEFORE the external
// send, so a crash between "the sink accepted it" and "we recorded that" (the
// window that causes duplicate sends on the next tick — the audit finding
// this closes) leaves a durable "pending" row instead of silence. Complete
// records the sink's actual outcome once Send returns. nil Outbox = tracking
// disabled (existing send/record behavior, unchanged — test scenarios that
// don't care about crash recovery).
type Outbox interface {
	BeginDeliveryAttempt(ctx context.Context, kind, rule string, reminderID int64, deliveryGroup, channel, bodyHash string, now time.Time) (int64, error)
	CompleteDeliveryAttempt(ctx context.Context, id int64, status string, now time.Time) error
}

// bodyHash is an opaque dedup/triage key for a delivery attempt row — not a
// uniqueness constraint (the same rule/reminder legitimately re-sends across
// ticks), just something an operator can compare across attempts after a
// crash to tell "same message resent" from "different message".
func bodyHash(channel Channel, body string) string {
	sum := sha256.Sum256([]byte(string(channel) + "\x00" + body))
	return hex.EncodeToString(sum[:8])
}

// beginOutbox records intent to send, if an Outbox is wired. A failure to
// record intent is not fatal to the send itself — losing outbox visibility
// on one attempt shouldn't block a nudge/reminder actually reaching the user
// — but it does mean this attempt can't be reconciled after a crash, so it's
// logged. Returns 0 (no-op id) when unrecorded.
func (d *Dispatcher) beginOutbox(ctx context.Context, kind, rule string, reminderID int64, deliveryGroup string, channel Channel, body string, now time.Time) int64 {
	if d.cfg.Outbox == nil {
		return 0
	}
	id, err := d.cfg.Outbox.BeginDeliveryAttempt(ctx, kind, rule, reminderID, deliveryGroup, string(channel), bodyHash(channel, body), now)
	if err != nil {
		log.Printf("dispatcher: outbox begin failed (send proceeds untracked): %v", err)
		return 0
	}
	return id
}

// beginReminderOutbox is stricter than the nudge helper above. A reminder may
// be retried indefinitely, so sending it without the durable attempt row would
// reopen an unobservable duplicate window after a crash. A configured but
// unhealthy outbox therefore blocks this transport attempt; a deliberately nil
// outbox still supports small isolated test/development wiring.
func (d *Dispatcher) beginReminderOutbox(ctx context.Context, reminderID int64, deliveryGroup string, channel Channel, body string, now time.Time) (int64, error) {
	if d.cfg.Outbox == nil {
		return 0, nil
	}
	id, err := d.cfg.Outbox.BeginDeliveryAttempt(
		ctx, "reminder", "", reminderID, deliveryGroup,
		string(channel), bodyHash(channel, body), now,
	)
	if err != nil {
		return 0, fmt.Errorf("begin reminder delivery attempt: %w", err)
	}
	return id, nil
}

// completeOutbox records the sink's outcome for a prior beginOutbox call.
// id==0 means either tracking is disabled or the begin failed — nothing to
// complete either way.
func (d *Dispatcher) completeOutbox(ctx context.Context, id int64, status string, now time.Time) error {
	if id == 0 || d.cfg.Outbox == nil {
		return nil
	}
	if err := d.cfg.Outbox.CompleteDeliveryAttempt(ctx, id, status, now); err != nil {
		log.Printf("dispatcher: outbox complete failed: %v", err)
		return err
	}
	return nil
}

// PhrasedNudge — the phraser module's output for a nudge. the phraser (the
// resident model, Qwen3-1.7B) takes (rule, severity, context) and produces
// Body (full message for voice) + Summary (minimal body for away channels).
// the phraser is a separate module; the dispatcher only consumes its output.
type PhrasedNudge struct {
	Candidate loop.Candidate
	Body      string
	Summary   string
	Mood      string
}

// PhrasedReminder — the phraser's output for a reminder.
type PhrasedReminder struct {
	Decision loop.ReminderDecision
	Body     string
	Summary  string
	Mood     string
}

// Dispatch — record of one successful send. returned to the daemon for
// logging, ack wiring, and nudge-id tracking. NudgeID is set for nudges
// (the feedback loop's key); 0 for reminders.
type Dispatch struct {
	Sendable Sendable
	NudgeID  int64
}

// Config — wires the dispatcher. nil sinks = that channel not wired. nil
// AckTracker = repeat-til-ack disabled (the daemon doesn't wire it until the
// telegram module lands). nil Nudges/Reminders = recording disabled (test
// scenarios that only exercise routing).
type Config struct {
	Voice     Sink
	Ntfy      Sink
	Telegram  Sink
	Ack       AckTracker
	Nudges    NudgeRecorder
	Reminders ReminderCompleter
	Outbox    Outbox
}

// Dispatcher — holds one sink per channel + the recorder seams. the daemon
// wires it once; per-tick it calls DispatchNudge / DispatchReminder. the
// routing table (channel.go) is pure; this struct is the impure orchestration.
type Dispatcher struct {
	cfg Config
}

func NewDispatcher(cfg Config) *Dispatcher {
	return &Dispatcher{cfg: cfg}
}

// DispatchNudge — routes a phrased nudge to the channels the routing table
// picks for (severity, presence), sends via the matching sink, and records
// one nudge row per successful send. returns the dispatches (one per channel).
//
// a Drop channel = no send (the nudge was suppressed by routing, not by a
// failure — "a missed water nudge is noise"), but it does leave a 'dropped'
// outbox row so the suppression is visible. a nil sink = channel not wired,
// skip silently. a send error stops the dispatch and returns what
// got through — the daemon decides whether to retry.
func (d *Dispatcher) DispatchNudge(ctx context.Context, pn PhrasedNudge, now time.Time) ([]Dispatch, error) {
	c := pn.Candidate
	channels := ChannelsFor(c.Severity, c.State.Presence)
	var out []Dispatch
	for i := 0; i < len(channels); i++ {
		ch := channels[i]
		if ch == ChannelDrop {
			// the routing table suppressed this nudge on purpose (a care nudge
			// while you're away is noise). that stays — but it must not be
			// invisible, or "she dropped it" and "the rule never fired" look
			// the same afterwards. no nudges row: that table feeds the
			// ignored_rate signal, and a nudge nobody could see must not
			// count as ignored.
			id := d.beginOutbox(ctx, "nudge", c.Rule.Name, 0, "", ch, pn.Summary, now)
			d.completeOutbox(ctx, id, store.DeliveryDropped, now)
			log.Printf("dispatcher: dropped %s (sev%d, presence=%s) — routing table suppressed it",
				c.Rule.Name, c.Severity, c.State.Presence)
			continue
		}
		s := Sendable{
			Channel:        ch,
			Kind:           KindNudge,
			Severity:       c.Severity,
			RuleName:       c.Rule.Name,
			Body:           pn.Body,
			Summary:        pn.Summary,
			RepeatUntilAck: ch == ChannelTelegram && c.Severity >= loop.Sev4,
			Ts:             now,
		}
		s = minimalForAway(s)
		sink := d.sinkFor(ch)
		if sink == nil {
			continue
		}
		attemptID := d.beginOutbox(ctx, "nudge", c.Rule.Name, 0, "", ch, messageForChannel(s), now)
		if err := safeSend(ctx, sink, s); err != nil {
			if errors.Is(err, ErrSinkPanicked) {
				// one broken sink must not eat the other channels for this
				// nudge (sev4 present is voice + ntfy). the attempt is closed
				// as failed and we move on.
				d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now)
				continue
			}
			if errors.Is(err, ErrVoiceNoSession) {
				// voice was assumed reachable (presence=present) but no live
				// session exists — the presence guess was wrong. reroute through
				// the AWAY table per § away-channel fallthrough: sev3→ntfy,
				// sev4→telegram-repeat-til-ack, sev≤2→drop. voice is always the
				// first present channel, so nothing has been sent yet; replace
				// the remaining list wholesale. away channels never include
				// voice, so this can't re-trigger.
				d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now)
				log.Printf("dispatcher: no live voice session for %s, rerouting to away channels", c.Rule.Name)
				channels = ChannelsFor(c.Severity, store.Away)
				i = -1
				continue
			}
			d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now)
			return out, fmt.Errorf("send %s: %w", ch, err)
		}
		d.completeOutbox(ctx, attemptID, store.DeliverySent, now)
		// record AFTER successful send — a failed send must not pollute the
		// feedback loop with a phantom nudge (ignored_rate would drift).
		var nudgeID int64
		if d.cfg.Nudges != nil {
			id, err := d.cfg.Nudges.RecordNudge(ctx, c.Rule.Name, string(ch), messageForChannel(s), now)
			if err != nil {
				return out, fmt.Errorf("record nudge %s: %w", ch, err)
			}
			nudgeID = id
		}
		// for repeat-til-ack telegram sends, mark the initial send in the ack
		// tracker so ShouldRepeat's clock starts now.
		if s.RepeatUntilAck && d.cfg.Ack != nil {
			if err := d.cfg.Ack.MarkSent(ctx, c.Rule.Name, now); err != nil {
				return out, fmt.Errorf("ack mark-sent: %w", err)
			}
		}
		out = append(out, Dispatch{Sendable: s, NudgeID: nudgeID})
	}
	return out, nil
}

// DispatchReminder — routes a phrased reminder. reminders bypass the gate and
// fire once (pending → fired after successful delivery). Voice is preferred
// when present; if it has no live session, delivery falls back to the ordered
// away alternatives. Away delivery tries ntfy, then telegram, and stops after
// the first success. A failed or unwired alternative falls through to the next
// one. If every selected alternative fails, the reminder stays pending and an
// error is returned for the tick's retry path.
func (d *Dispatcher) DispatchReminder(ctx context.Context, pr PhrasedReminder, now time.Time) ([]Dispatch, error) {
	rd := pr.Decision
	channels := ChannelsForReminder(rd.State.Presence)
	var failures []error
	allPermanent := true
	for i := 0; i < len(channels); i++ {
		ch := channels[i]
		s := Sendable{
			Channel:    ch,
			Kind:       KindReminder,
			ReminderID: rd.Reminder.ID,
			Body:       pr.Body,
			Summary:    pr.Summary,
			Ts:         now,
		}
		s = minimalForAway(s)
		reminderID, deliveryGroup := reminderDeliveryIdentity(rd.Reminder)
		attemptID, err := d.beginReminderOutbox(ctx, reminderID, deliveryGroup, ch, messageForChannel(s), now)
		if err != nil {
			allPermanent = false
			failures = append(failures, err)
			log.Printf("dispatcher: reminder %d delivery via %s withheld: %v", rd.Reminder.ID, ch, err)
			continue
		}
		sink := d.sinkFor(ch)
		if sink == nil {
			d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now)
			err := fmt.Errorf("%s sink is not configured", ch)
			failures = append(failures, err)
			log.Printf("dispatcher: reminder %d delivery via %s failed: %v", rd.Reminder.ID, ch, err)
			if ch == ChannelVoice {
				channels = ChannelsForReminder(store.Away)
				failures = nil
				allPermanent = true
				i = -1
			}
			continue
		}
		if err := safeSend(ctx, sink, s); err != nil {
			d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now)
			failures = append(failures, fmt.Errorf("send %s: %w", ch, err))
			if !errors.Is(err, ErrPermanent) {
				allPermanent = false
			}
			if errors.Is(err, ErrVoiceNoSession) {
				// Presence was stale. Voice is the only present alternative, so
				// nothing has been sent and it is safe to start the away chain.
				log.Printf("dispatcher: no live voice session for reminder %d, rerouting to away channels", rd.Reminder.ID)
				channels = ChannelsForReminder(store.Away)
				failures = nil
				allPermanent = true
				i = -1
				continue
			}
			log.Printf("dispatcher: reminder %d delivery via %s failed: %v", rd.Reminder.ID, ch, err)
			continue
		}
		out := []Dispatch{{Sendable: s}}
		originals := []store.Reminder{rd.Reminder}
		if rd.Reminder.ID == 0 {
			originals = rd.Reminder.Collapsed
		}
		if durable, ok := d.cfg.Reminders.(DurableReminderCompleter); ok && attemptID != 0 {
			if err := durable.CompleteSuccessfulReminderAttempt(ctx, attemptID, originals, now); err != nil {
				return out, fmt.Errorf("commit successful reminder delivery: %w", err)
			}
			return out, nil
		}
		if err := d.completeOutbox(ctx, attemptID, store.DeliverySent, now); err != nil {
			// The external sink accepted the reminder, but its durable outcome is
			// ambiguous. Do not complete the reminder row: startup reconciliation
			// will mark the attempt unknown and DueReminders will hold the exact
			// occurrence for operator resolution rather than sending a duplicate.
			return out, fmt.Errorf("record successful reminder delivery: %w", err)
		}
		if d.cfg.Reminders != nil {
			if err := d.cfg.Reminders.CompleteReminderDelivery(ctx, originals, now); err != nil {
				return out, fmt.Errorf("complete reminder delivery: %w", err)
			}
		}
		return out, nil
	}
	if len(failures) == 0 {
		failures = append(failures, errors.New("no delivery alternatives selected"))
		allPermanent = false
	}
	joined := errors.Join(failures...)
	if allPermanent && d.cfg.Reminders != nil {
		originals := []store.Reminder{rd.Reminder}
		if rd.Reminder.ID == 0 {
			originals = rd.Reminder.Collapsed
		}
		if blocker, ok := d.cfg.Reminders.(ReminderBlocker); ok {
			if err := blocker.BlockReminderDelivery(ctx, originals, now, joined.Error()); err != nil {
				return nil, fmt.Errorf("block permanently undeliverable reminder %d: %w", rd.Reminder.ID, err)
			}
		}
	}
	return nil, fmt.Errorf("deliver reminder %d: %w", rd.Reminder.ID, joined)
}

// reminderDeliveryIdentity gives the outbox both a human-readable real row id
// and the exact occurrence key used for crash suppression. A collapsed digest
// has synthetic ID zero, so its first original is the representative; the
// shared delivery group still identifies every original atomically.
func reminderDeliveryIdentity(r store.Reminder) (int64, string) {
	if r.ID != 0 {
		return r.ID, r.DeliveryGroup
	}
	if len(r.Collapsed) == 0 {
		return 0, ""
	}
	return r.Collapsed[0].ID, r.Collapsed[0].DeliveryGroup
}

// RepeatUnacked — the daemon calls this each tick to re-send un-acked sev4
// telegram nudges. `keys` = rule names with un-acked telegram sends (the
// daemon queries the nudges table for pending-outcome sev4+telegram rows —
// a store helper that's deferred). for each key: if not acked and the repeat
// interval has elapsed since the last send, re-send + update last-sent.
//
// the body/summary are passed in because the phraser output from the original
// dispatch isn't retained — the daemon re-phrases (or reuses a cached phrase).
// a sev4 alarm repeating with the same terse body is correct; it's an alarm.
func (d *Dispatcher) RepeatUnacked(ctx context.Context, keys []string, now time.Time, interval time.Duration, body, summary string) ([]Dispatch, error) {
	if d.cfg.Telegram == nil || d.cfg.Ack == nil {
		return nil, nil
	}
	var out []Dispatch
	for _, key := range keys {
		acked, err := d.cfg.Ack.WasAcked(ctx, key)
		if err != nil {
			return out, fmt.Errorf("ack was-acked %s: %w", key, err)
		}
		last, err := d.cfg.Ack.LastSent(ctx, key)
		if err != nil {
			return out, fmt.Errorf("ack last-sent %s: %w", key, err)
		}
		if !ShouldRepeat(last, acked, now, interval) {
			continue
		}
		s := Sendable{
			Channel:        ChannelTelegram,
			Kind:           KindNudge,
			Severity:       loop.Sev4,
			RuleName:       key,
			Body:           body,
			Summary:        summary,
			RepeatUntilAck: true,
			Ts:             now,
		}
		s = minimalForAway(s)
		attemptID := d.beginOutbox(ctx, "nudge", key, 0, "", ChannelTelegram, messageForChannel(s), now)
		if err := safeSend(ctx, d.cfg.Telegram, s); err != nil {
			d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now)
			if errors.Is(err, ErrSinkPanicked) {
				continue
			}
			return out, fmt.Errorf("repeat send telegram %s: %w", key, err)
		}
		d.completeOutbox(ctx, attemptID, store.DeliverySent, now)
		if err := d.cfg.Ack.MarkSent(ctx, key, now); err != nil {
			return out, fmt.Errorf("ack mark-sent %s: %w", key, err)
		}
		out = append(out, Dispatch{Sendable: s})
	}
	return out, nil
}

// ErrSinkPanicked — a sink panicked mid-send. the send did not happen, so the
// attempt is recorded failed and never silently retried as if it had.
var ErrSinkPanicked = errors.New("delivery: sink panicked mid-send")

// safeSend calls a sink and turns a panic into an error. without this a
// panicking sink unwinds past completeOutbox and leaves the delivery_attempts
// row pending forever — reconciliation only runs at daemon startup, and core
// is long-lived, so the row would sit there for weeks.
func safeSend(ctx context.Context, sink Sink, s Sendable) (err error) {
	defer func() {
		if r := recover(); r != nil {
			log.Printf("dispatcher: PANIC in %s sink (this is a bug, fix the sink): %v", s.Channel, r)
			err = fmt.Errorf("%w: %s: %v", ErrSinkPanicked, s.Channel, r)
		}
	}()
	return sink.Send(ctx, s)
}

func (d *Dispatcher) sinkFor(ch Channel) Sink {
	switch ch {
	case ChannelVoice:
		return d.cfg.Voice
	case ChannelNtfy:
		return d.cfg.Ntfy
	case ChannelTelegram:
		return d.cfg.Telegram
	default:
		return nil
	}
}

// GenericAwayMessage — what an away channel gets when the phraser gave us no
// summary. no gendered forms, so it stays right whoever reads it.
const GenericAwayMessage = "что-то требует внимания"

// isAway — this channel leaves the box, so it only ever gets a minimal body.
func isAway(ch Channel) bool {
	return ch == ChannelNtfy || ch == ChannelTelegram
}

// messageForChannel — away channels get the minimal summary (no shoulder-surf
// exfil — "disk low on homesrv," not detail); voice gets the full body (local).
// an empty summary must NOT fall back to the body: the resident model is small
// and drops fields often, and the away path crosses the "never phones home"
// boundary. so we send a fixed generic line plus the rule name instead. voice
// is local, so it keeps the full body.
func messageForChannel(s Sendable) string {
	if !isAway(s.Channel) {
		return s.Body
	}
	return AwayMessage(s)
}

// AwayMessage — the only text an off-box channel may ever carry. Exported so
// the away sinks share this one rule instead of each inventing a fallback: the
// summary if we have one, otherwise a fixed generic line. Never the body.
func AwayMessage(s Sendable) string {
	if s.Summary != "" {
		return s.Summary
	}
	if s.RuleName != "" {
		return GenericAwayMessage + ": " + s.RuleName
	}
	return GenericAwayMessage
}

// minimalForAway — strips detail from a Sendable bound for an away channel
// before any sink sees it. the sinks pick Summary themselves too, but this is
// where the boundary actually is: a sink added later must not be able to leak
// the full body just by reading the wrong field.
func minimalForAway(s Sendable) Sendable {
	if !isAway(s.Channel) {
		return s
	}
	msg := messageForChannel(s)
	s.Body = msg
	s.Summary = msg
	return s
}