package delivery import ( "context" "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. a reminder fires once: // pending → fired after successful delivery. a failed send does NOT mark the // reminder fired (it stays pending; the next tick re-delivers). type ReminderCompleter interface { MarkReminder(ctx context.Context, id int64, status string) error } // PhrasedNudge — the phraser module's output for a nudge. the phraser (LFM // sub-1b, prompted not trained) 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 } // PhrasedReminder — the phraser's output for a reminder. type PhrasedReminder struct { Decision loop.ReminderDecision Body string Summary 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 } // 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, no record (the nudge was suppressed by routing, // not by a failure — "a missed water nudge is noise"). 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 { 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, } sink := d.sinkFor(ch) if sink == nil { continue } if err := sink.Send(ctx, s); err != nil { 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. 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 } return out, fmt.Errorf("send %s: %w", ch, err) } // 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 when present, // ntfy when away. no repeat (reminders fire once). marks the reminder fired // only if at least one channel succeeded — a failed send leaves it pending // for the next tick to re-deliver. func (d *Dispatcher) DispatchReminder(ctx context.Context, pr PhrasedReminder, now time.Time) ([]Dispatch, error) { rd := pr.Decision channels := ChannelsForReminder(rd.State.Presence) var out []Dispatch 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, } sink := d.sinkFor(ch) if sink == nil { continue } if err := sink.Send(ctx, s); err != nil { if errors.Is(err, ErrVoiceNoSession) { // presence guess was wrong — reroute reminder to the away // channel (ntfy). voice is the only present channel, so nothing // has been sent yet. log.Printf("dispatcher: no live voice session for reminder %d, rerouting to away channels", rd.Reminder.ID) channels = ChannelsForReminder(store.Away) i = -1 continue } return out, fmt.Errorf("send %s: %w", ch, err) } out = append(out, Dispatch{Sendable: s}) } if d.cfg.Reminders != nil && len(out) > 0 { // ID=0 is a synthetic digest reminder; it's not in the DB so // MarkReminder would fail with ErrReminderNotFound. The originals // were already marked fired by collapseReminders in gather.go. if rd.Reminder.ID != 0 { if err := d.cfg.Reminders.MarkReminder(ctx, rd.Reminder.ID, "fired"); err != nil { return out, fmt.Errorf("mark reminder fired: %w", err) } } } return out, nil } // 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, } if err := d.cfg.Telegram.Send(ctx, s); err != nil { return out, fmt.Errorf("repeat send telegram %s: %w", key, err) } 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 } 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 } } // messageForChannel — away channels get the minimal summary (no shoulder-surf // exfil — "disk low on homesrv," not detail); voice gets the full body (local). // a missing summary falls back to body — a terse full message is better than // no message, and the phraser should have produced a summary for away-bound // severities. this is the "minimal body" rule from the spec, enforced at the // last mile so a phraser bug can't accidentally exfil via the relay. func messageForChannel(s Sendable) string { switch s.Channel { case ChannelNtfy, ChannelTelegram: if s.Summary != "" { return s.Summary } return s.Body default: return s.Body } }