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 }