// mavend/intake.go — the unified event intake envelope, wired (Vikunja #283). // // internal/event defines the envelope and the bounded in-memory journal. This // file is the one place that FILLS it, and the reason it is one place is worth // stating, because the alternative was eight patches: // // Every intake path in Maven already converges on three writes, and all three // are ipc.CoreAPI methods — // // WriteFact ← POST /api/ambient, mavcaldav, mavpoll's zenmoney + wg reads, // /api/signal presence probes, the RSS/crawl watermarks // WriteNote ← the RSS poller, the page crawler, meeting transcripts, // image descriptions // CaptureTask ← the voice path, the web form, and the mail reader // // — so decorating that ONE interface with a publish covers the lot without a // caller knowing about events at all. cmd/mavmaild, cmd/mavcaldav, cmd/mavpoll, // cmd/mavweb and the in-core feed/crawl/capture/vision workers are unchanged: // they call the same interface they always called, and it now also narrates. // // The exception is cmd/mavend/mail.go, which reaches past the interface to // st.CaptureTask directly. It publishes explicitly; see mailIntake.ingest. // // # Production behaviour when nobody is watching // // A nil *event.Bus makes Publish a no-op, and newIntakeAPI with a nil bus // returns the wrapped API unchanged, so there is not even a decorator on the // call path. The journal is memory-only and is never consulted by the tick // loop, the router, or delivery — nothing Maven says depends on it. It is a // read surface (`/events`, `recent_events`) and an observation seam for the // simulator. // // # What is deliberately NOT here // // No dispatch. An event is a report that something arrived, never an // instruction to speak: "a feed item appeared" becoming a notification is the // nag this repo refuses. Digestion may one day read the journal; it will still // go through internal/loop's rules and the severity/presence routing table. package main import ( "context" "log" "strings" "time" "github.com/kami/maven/internal/config" "github.com/kami/maven/internal/event" "github.com/kami/maven/internal/ipc" "github.com/kami/maven/internal/store" ) // newEventBus builds the journal, or returns nil when the operator turned it // off (a negative config.intake_journal). nil is the "behave exactly as before" // value all the way down: no decorator, no ring, no /events rows. func newEventBus(cfg *config.Config) *event.Bus { if cfg == nil || cfg.IntakeJournal < 0 { log.Printf("intake journal: off (intake_journal < 0)") return nil } n := cfg.IntakeJournal if n == 0 { n = config.DefaultIntakeJournal } log.Printf("intake journal: keeping the last %d intake events in memory", n) return event.NewBus(n) } // intakeEventsFn is the daemonAPI.getEvents closure: the bus's ring rendered as // the wire type. Returns nil for a nil bus, which the daemonAPI reports as an // empty journal rather than an error. func intakeEventsFn(bus *event.Bus) func(n int) []ipc.IntakeEvent { if bus == nil { return nil } return func(n int) []ipc.IntakeEvent { evs := bus.Recent(n) out := make([]ipc.IntakeEvent, 0, len(evs)) for _, e := range evs { out = append(out, ipc.IntakeEvent{ Source: e.Source, Kind: e.Kind, EntityIDs: e.EntityIDs, Title: e.Title, Body: e.Body, Priority: e.Priority, OccurredAt: e.OccurredAt, }) } return out } } // intakeAPI decorates a CoreAPI, publishing one envelope per successful // intake write. Embedding the interface means every other method passes // through untouched, and a new CoreAPI method is inherited rather than // silently dropped. type intakeAPI struct { ipc.CoreAPI bus *event.Bus now func() time.Time } // newIntakeAPI wraps api so its intake writes are journalled. A nil bus // returns api itself — no decorator, no allocation, no behaviour change. func newIntakeAPI(api ipc.CoreAPI, bus *event.Bus, now func() time.Time) ipc.CoreAPI { if bus == nil || api == nil { return api } if now == nil { now = time.Now } return &intakeAPI{CoreAPI: api, bus: bus, now: now} } // WriteFact journals the fact after it lands. Order matters: an event is a // report of something that HAPPENED, so a failed write publishes nothing. func (a *intakeAPI) WriteFact(ctx context.Context, req ipc.WriteFactReq) (int64, error) { id, err := a.CoreAPI.WriteFact(ctx, req) if err != nil { return id, err } // OccurredAt is req.Ts, not now: mavpoll's wg read carries the handshake // instant and the ambient path carries the meeting's start. Flattening // those to notice-time would make the journal lie about when things // happened, which is the one thing it is for. a.bus.Publish(event.Event{ Source: req.Source, Kind: event.SourceKind(req.Source, event.KindFact), Title: req.Key, Body: req.Value, Priority: factPriority(req), OccurredAt: req.Ts, EntityIDs: entityIDs(req.Subject), }, a.now()) return id, nil } // WriteNote journals a note. This is the RSS and crawler path, and also the // meeting transcript and image description paths, which write their derived // text as ordinary notes. func (a *intakeAPI) WriteNote(ctx context.Context, ts time.Time, text string, embedding []float32, source string) (int64, error) { id, err := a.CoreAPI.WriteNote(ctx, ts, text, embedding, source) if err != nil { return id, err } title, body := splitFirstLine(text) a.bus.Publish(event.Event{ Source: source, Kind: event.SourceKind(source, event.KindNote), Title: title, Body: body, Priority: event.PriorityLow, OccurredAt: ts, }, a.now()) return id, nil } // CaptureTask journals a captured task, but only when a row was actually // created. CaptureTask dedupes on normalised text among live rows, so a // mailbox re-read after a restart must not refill the journal with tasks that // were already there. func (a *intakeAPI) CaptureTask(ctx context.Context, req ipc.CaptureTaskReq) (ipc.CaptureTaskResp, error) { resp, err := a.CoreAPI.CaptureTask(ctx, req) if err != nil || !resp.Created { return resp, err } a.bus.Publish(publishableTask(store.Task{ CreatedTs: req.Ts, Text: req.Text, Source: req.Source, Evidence: req.Evidence, Status: req.Status, Due: req.Due, }, a.now()), a.now()) return resp, nil } // publishableTask is the task→envelope shape, shared with mail.go, which // captures through the store directly rather than through the interface. // // Priority is high for a candidate with a due date and normal otherwise. That // is the only place this file makes a judgement, and it is a display hint on a // review page — nothing routes on it. func publishableTask(t store.Task, now time.Time) event.Event { occurred := t.CreatedTs if occurred.IsZero() { occurred = now } prio := event.PriorityNormal if t.Due != nil { prio = event.PriorityHigh } return event.Event{ Source: t.Source, Kind: event.KindTask, Title: t.Text, Body: t.Evidence, Priority: prio, OccurredAt: occurred, } } // factPriority is the attention hint for a fact write. Deliberately crude: // a low-confidence inference (the ambient notification path writes below 1.0) // is worth less attention than a read he or a credentialled poller made, and // nothing else is distinguishable from here. func factPriority(req ipc.WriteFactReq) string { if req.Confidence > 0 && req.Confidence < 1.0 { return event.PriorityLow } return event.PriorityNormal } // entityIDs turns a fact's free-text Subject into the EntityIDs slot when it // already looks resolved. Intake runs BEFORE the fact enrichment worker // resolves a subject against Nexus, so this is almost always empty — the slot // exists for the paths that do know (the ecosystem acts), not for guessing. func entityIDs(subject string) []string { subject = strings.TrimSpace(subject) if subject == "" || !strings.HasPrefix(subject, "entity:") { return nil } return []string{strings.TrimPrefix(subject, "entity:")} } // splitFirstLine renders a note as title + body. Feed and crawl notes are // written "headline\nsummary\nlink", so the first line is already the title. func splitFirstLine(text string) (title, body string) { text = strings.TrimSpace(text) if i := strings.IndexByte(text, '\n'); i >= 0 { return strings.TrimSpace(text[:i]), strings.TrimSpace(text[i+1:]) } return text, "" }