Maven/cmd/mavend/intake.go

// 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"
	"encoding/json"
	"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 {
		// No config at all is a test, not an operator decision. Saying "off"
		// here was noise in every suite that passes nil.
		return nil
	}
	if 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,
				NoticedAt:  e.NoticedAt,
			})
		}
		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
	}
	if selfWrite(req) {
		// Maven's own bookkeeping is not something that arrived. The feed
		// watermark, the crawl hash, the praxis trace of an act she performed
		// and a quiet-hours toggle he pressed all used to sit on a page headed
		// "everything that arrived", and on a cold start a handful of feeds
		// could evict real intake behind their marks.
		return id, nil
	}
	// 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.
	title := req.Key
	if req.VoidsID != nil {
		// A retraction is not a reading. Without this it published an envelope
		// indistinguishable from a fresh value for the same key, on a page
		// whose whole job is "what came in".
		title = "отмена: " + req.Key
	}
	a.bus.Publish(event.Event{
		Source:     req.Source,
		Kind:       event.SourceKind(req.Source, event.KindFact),
		Title:      title,
		Body:       req.Value,
		Priority:   factPriority(req),
		OccurredAt: req.Ts,
		EntityIDs:  entityIDs(req.Subject),
		Payload:    factPayload(req),
	}, 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,
	}
}

// selfWrite reports whether a fact write is Maven describing her own state
// rather than something arriving from outside. The store's fact kinds are
// 'self', 'env' and 'config'; 'config' is where every watermark and toggle
// lands, and the praxis trace is an audit record of an act she performed, which
// is the same class of thing under an 'env' kind.
func selfWrite(req ipc.WriteFactReq) bool {
	switch req.Kind {
	case "config", "system":
		return true
	}
	return strings.HasPrefix(req.Source, "praxis:trace")
}

// 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 a
// retraction is a correction rather than news.
//
// Confidence is NOT recoverable from this, which is why the number itself goes
// into Payload: three display buckets must not be the only surviving trace of
// the distinction internal/calendar went out of its way to keep.
func factPriority(req ipc.WriteFactReq) string {
	if req.VoidsID != nil {
		return event.PriorityLow
	}
	if req.Confidence > 0 && req.Confidence < 1.0 {
		return event.PriorityLow
	}
	return event.PriorityNormal
}

// factDetail is the fact-shaped Payload: the fields the envelope's own flat
// shape cannot carry, kept so a reader can tell an inference from a
// credentialled read, and "nobody said" from "certain".
type factDetail struct {
	// FactKind — the fact's own kind ('self', 'env', 'config'), a different
	// taxonomy from Event.Kind.
	FactKind string `json:"fact_kind,omitempty"`
	// Confidence — the number itself, so an inference stays distinguishable
	// from a credentialled read. nil when the writer set none, which the ipc
	// layer rejects today; the pointer keeps "nobody said" and "certain" from
	// collapsing into each other the way the priority bucket does.
	Confidence *float64 `json:"confidence,omitempty"`
	// VoidsID — the fact this one retracts.
	VoidsID *int64 `json:"voids_id,omitempty"`
}

func factPayload(req ipc.WriteFactReq) json.RawMessage {
	d := factDetail{FactKind: req.Kind, VoidsID: req.VoidsID}
	if req.Confidence != 0 {
		c := req.Confidence
		d.Confidence = &c
	}
	if d.FactKind == "" && d.Confidence == nil && d.VoidsID == nil {
		return nil
	}
	b, err := json.Marshal(d)
	if err != nil {
		return nil
	}
	return b
}

// 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, ""
}