Review PR50: split internal/ipc/server.go (882 lines) into storeapi.go plus the dispatcher #151
@@ -2,9 +2,7 @@ package ipc
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
"net"
|
"net"
|
||||||
@@ -13,449 +11,9 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/kami/maven/internal/netaddr"
|
"github.com/kami/maven/internal/netaddr"
|
||||||
"github.com/kami/maven/internal/store"
|
|
||||||
"golang.org/x/sys/unix"
|
"golang.org/x/sys/unix"
|
||||||
)
|
)
|
||||||
|
|
||||||
// storeAPI — adapts *store.Store to CoreAPI. The daemon constructs one of
|
|
||||||
// these inside the core process; the socket Server calls it through the
|
|
||||||
// CoreAPI interface, so over-the-wire and in-process callers behave
|
|
||||||
// identically. The translation here is the only place store sentinels cross
|
|
||||||
// the wire: store.ErrNoFact becomes ipc.ErrNoFact, etc. — keeping the module
|
|
||||||
// view of errors stable regardless of transport.
|
|
||||||
type storeAPI struct {
|
|
||||||
s *store.Store
|
|
||||||
}
|
|
||||||
|
|
||||||
// NewStoreAPI wraps a *store.Store as a CoreAPI. The store is the sqlcipher-
|
|
||||||
// unlocked handle held ONLY in core's address space; this adapter never
|
|
||||||
// returns it to a caller — core mediates.
|
|
||||||
func NewStoreAPI(s *store.Store) CoreAPI { return &storeAPI{s: s} }
|
|
||||||
|
|
||||||
func (a *storeAPI) WriteFact(ctx context.Context, req WriteFactReq) (int64, error) {
|
|
||||||
var voids sql.NullInt64
|
|
||||||
if req.VoidsID != nil {
|
|
||||||
voids = sql.NullInt64{Int64: *req.VoidsID, Valid: true}
|
|
||||||
}
|
|
||||||
id, err := a.s.WriteFactAboutSubject(ctx, req.Ts, store.FactKind(req.Kind), req.Key, req.Subject, req.Value, req.Source, req.Confidence, voids)
|
|
||||||
return id, mapErr(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) LatestFact(ctx context.Context, key string) (Fact, error) {
|
|
||||||
f, err := a.s.LatestFact(ctx, key)
|
|
||||||
if err != nil {
|
|
||||||
return Fact{}, mapErr(err)
|
|
||||||
}
|
|
||||||
return toFact(f), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) LatestFactBySource(ctx context.Context, key, source string) (Fact, error) {
|
|
||||||
f, err := a.s.LatestFactBySource(ctx, key, source)
|
|
||||||
if err != nil {
|
|
||||||
return Fact{}, mapErr(err)
|
|
||||||
}
|
|
||||||
return toFact(f), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) Since(ctx context.Context, key string, now time.Time) (time.Duration, error) {
|
|
||||||
d, err := a.s.Since(ctx, key, now)
|
|
||||||
return d, mapErr(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) Presence(ctx context.Context) (Presence, error) {
|
|
||||||
b, score, upd, err := a.s.LoadPresenceState(ctx)
|
|
||||||
if err != nil {
|
|
||||||
return Presence{}, fmt.Errorf("ipc: load presence: %w", err)
|
|
||||||
}
|
|
||||||
return Presence{Bucket: Bucket(b), Score: score, Updated: upd}, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) CreateReminder(ctx context.Context, fire time.Time, payload, cron string) (int64, error) {
|
|
||||||
id, err := a.s.CreateReminder(ctx, fire, payload, cron)
|
|
||||||
return id, mapErr(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) MarkReminder(ctx context.Context, id int64, status string) error {
|
|
||||||
return mapErr(a.s.MarkReminder(ctx, id, status))
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) ListReminders(ctx context.Context, n int) ([]Reminder, error) {
|
|
||||||
rs, err := a.s.ListReminders(ctx, n)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]Reminder, len(rs))
|
|
||||||
for i, r := range rs {
|
|
||||||
out[i] = toReminder(r)
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) RescheduleReminder(ctx context.Context, id int64, now time.Time) error {
|
|
||||||
return mapErr(a.s.RescheduleReminder(ctx, id, now))
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) RecordNudge(ctx context.Context, rule, channel, message string, ts time.Time) (int64, error) {
|
|
||||||
id, err := a.s.RecordNudge(ctx, rule, channel, message, ts)
|
|
||||||
return id, mapErr(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) ResolveNudge(ctx context.Context, id int64, outcome string, ts time.Time) error {
|
|
||||||
return mapErr(a.s.ResolveNudge(ctx, id, outcome, ts))
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) RecentOutcomes(ctx context.Context, rule string, n int) ([]string, error) {
|
|
||||||
out, err := a.s.RecentOutcomes(ctx, rule, n)
|
|
||||||
return out, mapErr(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) RecentFacts(ctx context.Context, n int) ([]Fact, error) {
|
|
||||||
fs, err := a.s.RecentFacts(ctx, n)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]Fact, len(fs))
|
|
||||||
for i, f := range fs {
|
|
||||||
out[i] = toFact(f)
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) RecentActiveFactsByKind(ctx context.Context, kind string, n int) ([]Fact, error) {
|
|
||||||
fs, err := a.s.RecentActiveFactsByKind(ctx, store.FactKind(kind), n)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]Fact, len(fs))
|
|
||||||
for i, f := range fs {
|
|
||||||
out[i] = toFact(f)
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) CalendarEvents(ctx context.Context, from, to time.Time) ([]Fact, error) {
|
|
||||||
fs, err := a.s.CalendarEvents(ctx, from, to)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]Fact, len(fs))
|
|
||||||
for i, f := range fs {
|
|
||||||
out[i] = toFact(f)
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) RecentEcosystemTraces(ctx context.Context, n int) ([]EcosystemTrace, error) {
|
|
||||||
trs, err := a.s.RecentEcosystemTraces(ctx, n)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]EcosystemTrace, len(trs))
|
|
||||||
for i, tr := range trs {
|
|
||||||
out[i] = EcosystemTrace{
|
|
||||||
ID: tr.ID, Ts: tr.Ts, Service: tr.Service, Operation: tr.Operation,
|
|
||||||
Status: tr.Status, DurationMs: tr.DurationMs, CorrelationID: tr.CorrelationID,
|
|
||||||
CausationID: tr.CausationID, HTTPStatus: tr.HTTPStatus, Fields: tr.Fields,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) RecentNudges(ctx context.Context, n int) ([]Nudge, error) {
|
|
||||||
ns, err := a.s.RecentNudges(ctx, n)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]Nudge, len(ns))
|
|
||||||
for i, ng := range ns {
|
|
||||||
out[i] = toNudge(ng)
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) DeliveryAttempts(ctx context.Context, status string, n int) ([]DeliveryAttempt, error) {
|
|
||||||
as, err := a.s.ListDeliveryAttempts(ctx, status, n)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]DeliveryAttempt, len(as))
|
|
||||||
for i, at := range as {
|
|
||||||
out[i] = DeliveryAttempt{
|
|
||||||
ID: at.ID, Kind: at.Kind, Rule: at.Rule, ReminderID: at.ReminderID,
|
|
||||||
Channel: at.Channel, Status: at.Status, Created: at.Created,
|
|
||||||
}
|
|
||||||
if at.HasComplete {
|
|
||||||
t := at.Completed
|
|
||||||
out[i].Completed = &t
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) WriteNote(ctx context.Context, ts time.Time, text string, embedding []float32, source string) (int64, error) {
|
|
||||||
id, err := a.s.WriteNote(ctx, ts, text, embedding, source)
|
|
||||||
return id, mapErr(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) QueryNotes(ctx context.Context, embedding []float32, k int) ([]Note, error) {
|
|
||||||
ns, err := a.s.QueryNotes(ctx, embedding, k)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]Note, len(ns))
|
|
||||||
for i, n := range ns {
|
|
||||||
out[i] = toNote(n)
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) RecentNotesFromSource(ctx context.Context, prefix string, n int) ([]Note, error) {
|
|
||||||
ns, err := a.s.RecentNotesFromSource(ctx, prefix, n)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]Note, len(ns))
|
|
||||||
for i, note := range ns {
|
|
||||||
out[i] = toNote(note)
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) RecentNotes(ctx context.Context, n int) ([]Note, error) {
|
|
||||||
ns, err := a.s.RecentNotes(ctx, n)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]Note, len(ns))
|
|
||||||
for i, note := range ns {
|
|
||||||
out[i] = toNote(note)
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) ProposeTool(ctx context.Context, name, utterance, scope string, ts time.Time) (bool, error) {
|
|
||||||
ok, err := a.s.ProposeTool(ctx, name, utterance, scope, ts)
|
|
||||||
return ok, mapErr(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) EnableTool(ctx context.Context, name string, cmd []string, destructive bool, scope string, ts time.Time) error {
|
|
||||||
return mapErr(a.s.EnableTool(ctx, name, cmd, destructive, scope, ts))
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) DisableTool(ctx context.Context, name string) error {
|
|
||||||
return mapErr(a.s.DisableTool(ctx, name))
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) LookupTool(ctx context.Context, name string) (Tool, error) {
|
|
||||||
t, err := a.s.LookupTool(ctx, name)
|
|
||||||
if err != nil {
|
|
||||||
return Tool{}, mapErr(err)
|
|
||||||
}
|
|
||||||
return toTool(t), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) RevertFact(ctx context.Context, key string) (int64, error) {
|
|
||||||
_, newID, err := a.s.VoidLatestFact(ctx, key, "feedback", time.Now())
|
|
||||||
return newID, mapErr(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) Chat(ctx context.Context, conversation, text string) (string, error) {
|
|
||||||
return "", errors.New("store: chat not available via direct store API")
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) TickTrace(ctx context.Context) (TickTrace, error) {
|
|
||||||
return TickTrace{}, errors.New("store: tick trace not available via direct store API")
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) MorningStatus(ctx context.Context) ([]MorningRoutineStatus, error) {
|
|
||||||
return nil, errors.New("store: morning status not available via direct store API")
|
|
||||||
}
|
|
||||||
|
|
||||||
// RecentEvents — same shape as TickTrace: the intake journal is a bounded ring
|
|
||||||
// in the daemon's memory, not a table, so a bare store cannot serve it.
|
|
||||||
func (a *storeAPI) RecentEvents(ctx context.Context, n int) ([]IntakeEvent, error) {
|
|
||||||
return nil, errors.New("store: intake events not available via direct store API")
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) MCPServers(ctx context.Context) ([]MCPServerStatus, error) {
|
|
||||||
return nil, nil // no manager behind a bare store: nothing configured
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) DayPlan(ctx context.Context) (DayPlan, error) {
|
|
||||||
return DayPlan{}, errors.New("store: day plan not available via direct store API")
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) ListTools(ctx context.Context, status string) ([]Tool, error) {
|
|
||||||
ts, err := a.s.ListTools(ctx, status)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]Tool, len(ts))
|
|
||||||
for i, t := range ts {
|
|
||||||
out[i] = toTool(t)
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) DeleteTool(ctx context.Context, name string) error {
|
|
||||||
return mapErr(a.s.DeleteTool(ctx, name))
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) CaptureTask(ctx context.Context, req CaptureTaskReq) (CaptureTaskResp, error) {
|
|
||||||
res, err := a.s.CaptureTask(ctx, store.Task{
|
|
||||||
CreatedTs: req.Ts,
|
|
||||||
Text: req.Text,
|
|
||||||
Source: req.Source,
|
|
||||||
Evidence: req.Evidence,
|
|
||||||
ExternalID: req.ExternalID,
|
|
||||||
Status: req.Status,
|
|
||||||
Due: req.Due,
|
|
||||||
Weight: req.Weight,
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return CaptureTaskResp{}, mapErr(err)
|
|
||||||
}
|
|
||||||
return CaptureTaskResp{ID: res.ID, Created: res.Created, Promoted: res.Promoted}, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) ListTasks(ctx context.Context, status string) ([]Task, error) {
|
|
||||||
ts, err := a.s.ListTasks(ctx, status)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]Task, len(ts))
|
|
||||||
for i, t := range ts {
|
|
||||||
out[i] = Task{
|
|
||||||
ID: t.ID,
|
|
||||||
CreatedTs: t.CreatedTs,
|
|
||||||
Text: t.Text,
|
|
||||||
Source: t.Source,
|
|
||||||
Evidence: t.Evidence,
|
|
||||||
ExternalID: t.ExternalID,
|
|
||||||
Status: t.Status,
|
|
||||||
Due: t.Due,
|
|
||||||
Weight: t.Weight,
|
|
||||||
Resolved: t.ResolvedTs,
|
|
||||||
ResolvedBy: t.ResolvedBy,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) SetTaskStatus(ctx context.Context, id int64, status string, ts time.Time, by string) error {
|
|
||||||
return mapErr(a.s.SetTaskStatus(ctx, id, status, ts, by))
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) ListProposedRoutines(ctx context.Context) ([]ProposedRoutine, error) {
|
|
||||||
rs, err := a.s.ListProposedRoutines(ctx)
|
|
||||||
if err != nil {
|
|
||||||
return nil, mapErr(err)
|
|
||||||
}
|
|
||||||
out := make([]ProposedRoutine, len(rs))
|
|
||||||
for i, r := range rs {
|
|
||||||
out[i] = ProposedRoutine{
|
|
||||||
ID: r.ID,
|
|
||||||
Action: r.Action,
|
|
||||||
Object: r.Object,
|
|
||||||
IntervalDays: r.IntervalDays,
|
|
||||||
Status: r.Status,
|
|
||||||
CreatedTs: r.CreatedTs.UnixMilli(),
|
|
||||||
}
|
|
||||||
if r.ReminderID != nil {
|
|
||||||
out[i].ReminderID = r.ReminderID
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) DismissProposedRoutine(ctx context.Context, id int64) error {
|
|
||||||
return mapErr(a.s.DismissProposedRoutine(ctx, id))
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *storeAPI) AcceptProposedRoutine(ctx context.Context, id int64) error {
|
|
||||||
return mapErr(a.s.AcceptProposedRoutine(ctx, id, time.Now().UTC()))
|
|
||||||
}
|
|
||||||
|
|
||||||
func toTool(t store.Tool) Tool {
|
|
||||||
return Tool{
|
|
||||||
Name: t.Name, Scope: t.Scope, Cmd: t.Cmd, Destructive: t.Destructive,
|
|
||||||
Status: t.Status, Utterance: t.Utterance, Created: t.CreatedTs, Updated: t.UpdatedTs,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func toReminder(r store.Reminder) Reminder {
|
|
||||||
return Reminder{
|
|
||||||
ID: r.ID,
|
|
||||||
CreatedTs: r.CreatedTs,
|
|
||||||
FireTs: r.FireTs,
|
|
||||||
NextFireTs: r.NextFireTs,
|
|
||||||
Payload: r.Payload,
|
|
||||||
Status: r.Status,
|
|
||||||
Cron: r.Cron,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func toNote(n store.Note) Note {
|
|
||||||
return Note{ID: n.ID, Ts: n.Ts, Text: n.Text, Source: n.Source, Score: n.Score}
|
|
||||||
}
|
|
||||||
|
|
||||||
func toNudge(n store.Nudge) Nudge {
|
|
||||||
out := Nudge{
|
|
||||||
ID: n.ID, Ts: n.Ts, Rule: n.Rule, Channel: n.Channel,
|
|
||||||
Message: n.Message, Outcome: n.Outcome,
|
|
||||||
}
|
|
||||||
if n.OutcomeTs.Valid {
|
|
||||||
v := n.OutcomeTs.Int64
|
|
||||||
out.OutcomeTs = &v
|
|
||||||
}
|
|
||||||
return out
|
|
||||||
}
|
|
||||||
|
|
||||||
func toFact(f store.Fact) Fact {
|
|
||||||
out := Fact{
|
|
||||||
ID: f.ID,
|
|
||||||
Ts: f.Ts,
|
|
||||||
Kind: string(f.Kind),
|
|
||||||
Key: f.Key,
|
|
||||||
Value: f.Value,
|
|
||||||
Source: f.Source,
|
|
||||||
Confidence: f.Confidence,
|
|
||||||
}
|
|
||||||
if f.VoidsID.Valid {
|
|
||||||
v := f.VoidsID.Int64
|
|
||||||
out.VoidsID = &v
|
|
||||||
}
|
|
||||||
return out
|
|
||||||
}
|
|
||||||
|
|
||||||
// mapErr — store sentinel ↔ ipc sentinel. An unrecognized store error is
|
|
||||||
// wrapped but not mapped (server-side dispatch surfaces it as codeInternal,
|
|
||||||
// keeping internal text off the wire except to the daemon log).
|
|
||||||
func mapErr(err error) error {
|
|
||||||
if err == nil {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
switch {
|
|
||||||
case errors.Is(err, store.ErrNoFact):
|
|
||||||
return ErrNoFact
|
|
||||||
case errors.Is(err, store.ErrConfidence):
|
|
||||||
return ErrConfidence
|
|
||||||
case errors.Is(err, store.ErrVoidsMissing):
|
|
||||||
return ErrVoidsMissing
|
|
||||||
case errors.Is(err, store.ErrNudgeNotFound):
|
|
||||||
return ErrNudgeNotFound
|
|
||||||
case errors.Is(err, store.ErrNudgeOutcome):
|
|
||||||
return ErrNudgeOutcome
|
|
||||||
case errors.Is(err, store.ErrReminderNotFound):
|
|
||||||
return ErrReminderNotFound
|
|
||||||
case errors.Is(err, store.ErrReminderState):
|
|
||||||
return ErrReminderState
|
|
||||||
case errors.Is(err, store.ErrToolNotFound):
|
|
||||||
return ErrToolNotFound
|
|
||||||
}
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Server — the core side of the boundary. Listens on a unix domain socket,
|
// Server — the core side of the boundary. Listens on a unix domain socket,
|
||||||
// accepts module connections, frames requests to a CoreAPI and responses back.
|
// accepts module connections, frames requests to a CoreAPI and responses back.
|
||||||
// One Server per daemon process; concurrent connections are handled in their
|
// One Server per daemon process; concurrent connections are handled in their
|
||||||
|
|||||||
@@ -0,0 +1,455 @@
|
|||||||
|
// ipc/storeapi.go — the sqlite-backed CoreAPI.
|
||||||
|
//
|
||||||
|
// Split out of server.go, move-only (Vikunja #423). server.go was two
|
||||||
|
// unrelated things: this adapter, and the dispatcher that calls it over the
|
||||||
|
// socket. Nothing here knows there is a wire.
|
||||||
|
package ipc
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"database/sql"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/kami/maven/internal/store"
|
||||||
|
)
|
||||||
|
|
||||||
|
// storeAPI — adapts *store.Store to CoreAPI. The daemon constructs one of
|
||||||
|
// these inside the core process; the socket Server calls it through the
|
||||||
|
// CoreAPI interface, so over-the-wire and in-process callers behave
|
||||||
|
// identically. The translation here is the only place store sentinels cross
|
||||||
|
// the wire: store.ErrNoFact becomes ipc.ErrNoFact, etc. — keeping the module
|
||||||
|
// view of errors stable regardless of transport.
|
||||||
|
type storeAPI struct {
|
||||||
|
s *store.Store
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewStoreAPI wraps a *store.Store as a CoreAPI. The store is the sqlcipher-
|
||||||
|
// unlocked handle held ONLY in core's address space; this adapter never
|
||||||
|
// returns it to a caller — core mediates.
|
||||||
|
func NewStoreAPI(s *store.Store) CoreAPI { return &storeAPI{s: s} }
|
||||||
|
|
||||||
|
func (a *storeAPI) WriteFact(ctx context.Context, req WriteFactReq) (int64, error) {
|
||||||
|
var voids sql.NullInt64
|
||||||
|
if req.VoidsID != nil {
|
||||||
|
voids = sql.NullInt64{Int64: *req.VoidsID, Valid: true}
|
||||||
|
}
|
||||||
|
id, err := a.s.WriteFactAboutSubject(ctx, req.Ts, store.FactKind(req.Kind), req.Key, req.Subject, req.Value, req.Source, req.Confidence, voids)
|
||||||
|
return id, mapErr(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) LatestFact(ctx context.Context, key string) (Fact, error) {
|
||||||
|
f, err := a.s.LatestFact(ctx, key)
|
||||||
|
if err != nil {
|
||||||
|
return Fact{}, mapErr(err)
|
||||||
|
}
|
||||||
|
return toFact(f), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) LatestFactBySource(ctx context.Context, key, source string) (Fact, error) {
|
||||||
|
f, err := a.s.LatestFactBySource(ctx, key, source)
|
||||||
|
if err != nil {
|
||||||
|
return Fact{}, mapErr(err)
|
||||||
|
}
|
||||||
|
return toFact(f), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) Since(ctx context.Context, key string, now time.Time) (time.Duration, error) {
|
||||||
|
d, err := a.s.Since(ctx, key, now)
|
||||||
|
return d, mapErr(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) Presence(ctx context.Context) (Presence, error) {
|
||||||
|
b, score, upd, err := a.s.LoadPresenceState(ctx)
|
||||||
|
if err != nil {
|
||||||
|
return Presence{}, fmt.Errorf("ipc: load presence: %w", err)
|
||||||
|
}
|
||||||
|
return Presence{Bucket: Bucket(b), Score: score, Updated: upd}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) CreateReminder(ctx context.Context, fire time.Time, payload, cron string) (int64, error) {
|
||||||
|
id, err := a.s.CreateReminder(ctx, fire, payload, cron)
|
||||||
|
return id, mapErr(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) MarkReminder(ctx context.Context, id int64, status string) error {
|
||||||
|
return mapErr(a.s.MarkReminder(ctx, id, status))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) ListReminders(ctx context.Context, n int) ([]Reminder, error) {
|
||||||
|
rs, err := a.s.ListReminders(ctx, n)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]Reminder, len(rs))
|
||||||
|
for i, r := range rs {
|
||||||
|
out[i] = toReminder(r)
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) RescheduleReminder(ctx context.Context, id int64, now time.Time) error {
|
||||||
|
return mapErr(a.s.RescheduleReminder(ctx, id, now))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) RecordNudge(ctx context.Context, rule, channel, message string, ts time.Time) (int64, error) {
|
||||||
|
id, err := a.s.RecordNudge(ctx, rule, channel, message, ts)
|
||||||
|
return id, mapErr(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) ResolveNudge(ctx context.Context, id int64, outcome string, ts time.Time) error {
|
||||||
|
return mapErr(a.s.ResolveNudge(ctx, id, outcome, ts))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) RecentOutcomes(ctx context.Context, rule string, n int) ([]string, error) {
|
||||||
|
out, err := a.s.RecentOutcomes(ctx, rule, n)
|
||||||
|
return out, mapErr(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) RecentFacts(ctx context.Context, n int) ([]Fact, error) {
|
||||||
|
fs, err := a.s.RecentFacts(ctx, n)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]Fact, len(fs))
|
||||||
|
for i, f := range fs {
|
||||||
|
out[i] = toFact(f)
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) RecentActiveFactsByKind(ctx context.Context, kind string, n int) ([]Fact, error) {
|
||||||
|
fs, err := a.s.RecentActiveFactsByKind(ctx, store.FactKind(kind), n)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]Fact, len(fs))
|
||||||
|
for i, f := range fs {
|
||||||
|
out[i] = toFact(f)
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) CalendarEvents(ctx context.Context, from, to time.Time) ([]Fact, error) {
|
||||||
|
fs, err := a.s.CalendarEvents(ctx, from, to)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]Fact, len(fs))
|
||||||
|
for i, f := range fs {
|
||||||
|
out[i] = toFact(f)
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) RecentEcosystemTraces(ctx context.Context, n int) ([]EcosystemTrace, error) {
|
||||||
|
trs, err := a.s.RecentEcosystemTraces(ctx, n)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]EcosystemTrace, len(trs))
|
||||||
|
for i, tr := range trs {
|
||||||
|
out[i] = EcosystemTrace{
|
||||||
|
ID: tr.ID, Ts: tr.Ts, Service: tr.Service, Operation: tr.Operation,
|
||||||
|
Status: tr.Status, DurationMs: tr.DurationMs, CorrelationID: tr.CorrelationID,
|
||||||
|
CausationID: tr.CausationID, HTTPStatus: tr.HTTPStatus, Fields: tr.Fields,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) RecentNudges(ctx context.Context, n int) ([]Nudge, error) {
|
||||||
|
ns, err := a.s.RecentNudges(ctx, n)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]Nudge, len(ns))
|
||||||
|
for i, ng := range ns {
|
||||||
|
out[i] = toNudge(ng)
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) DeliveryAttempts(ctx context.Context, status string, n int) ([]DeliveryAttempt, error) {
|
||||||
|
as, err := a.s.ListDeliveryAttempts(ctx, status, n)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]DeliveryAttempt, len(as))
|
||||||
|
for i, at := range as {
|
||||||
|
out[i] = DeliveryAttempt{
|
||||||
|
ID: at.ID, Kind: at.Kind, Rule: at.Rule, ReminderID: at.ReminderID,
|
||||||
|
Channel: at.Channel, Status: at.Status, Created: at.Created,
|
||||||
|
}
|
||||||
|
if at.HasComplete {
|
||||||
|
t := at.Completed
|
||||||
|
out[i].Completed = &t
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) WriteNote(ctx context.Context, ts time.Time, text string, embedding []float32, source string) (int64, error) {
|
||||||
|
id, err := a.s.WriteNote(ctx, ts, text, embedding, source)
|
||||||
|
return id, mapErr(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) QueryNotes(ctx context.Context, embedding []float32, k int) ([]Note, error) {
|
||||||
|
ns, err := a.s.QueryNotes(ctx, embedding, k)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]Note, len(ns))
|
||||||
|
for i, n := range ns {
|
||||||
|
out[i] = toNote(n)
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) RecentNotesFromSource(ctx context.Context, prefix string, n int) ([]Note, error) {
|
||||||
|
ns, err := a.s.RecentNotesFromSource(ctx, prefix, n)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]Note, len(ns))
|
||||||
|
for i, note := range ns {
|
||||||
|
out[i] = toNote(note)
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) RecentNotes(ctx context.Context, n int) ([]Note, error) {
|
||||||
|
ns, err := a.s.RecentNotes(ctx, n)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]Note, len(ns))
|
||||||
|
for i, note := range ns {
|
||||||
|
out[i] = toNote(note)
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) ProposeTool(ctx context.Context, name, utterance, scope string, ts time.Time) (bool, error) {
|
||||||
|
ok, err := a.s.ProposeTool(ctx, name, utterance, scope, ts)
|
||||||
|
return ok, mapErr(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) EnableTool(ctx context.Context, name string, cmd []string, destructive bool, scope string, ts time.Time) error {
|
||||||
|
return mapErr(a.s.EnableTool(ctx, name, cmd, destructive, scope, ts))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) DisableTool(ctx context.Context, name string) error {
|
||||||
|
return mapErr(a.s.DisableTool(ctx, name))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) LookupTool(ctx context.Context, name string) (Tool, error) {
|
||||||
|
t, err := a.s.LookupTool(ctx, name)
|
||||||
|
if err != nil {
|
||||||
|
return Tool{}, mapErr(err)
|
||||||
|
}
|
||||||
|
return toTool(t), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) RevertFact(ctx context.Context, key string) (int64, error) {
|
||||||
|
_, newID, err := a.s.VoidLatestFact(ctx, key, "feedback", time.Now())
|
||||||
|
return newID, mapErr(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) Chat(ctx context.Context, conversation, text string) (string, error) {
|
||||||
|
return "", errors.New("store: chat not available via direct store API")
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) TickTrace(ctx context.Context) (TickTrace, error) {
|
||||||
|
return TickTrace{}, errors.New("store: tick trace not available via direct store API")
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) MorningStatus(ctx context.Context) ([]MorningRoutineStatus, error) {
|
||||||
|
return nil, errors.New("store: morning status not available via direct store API")
|
||||||
|
}
|
||||||
|
|
||||||
|
// RecentEvents — same shape as TickTrace: the intake journal is a bounded ring
|
||||||
|
// in the daemon's memory, not a table, so a bare store cannot serve it.
|
||||||
|
func (a *storeAPI) RecentEvents(ctx context.Context, n int) ([]IntakeEvent, error) {
|
||||||
|
return nil, errors.New("store: intake events not available via direct store API")
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) MCPServers(ctx context.Context) ([]MCPServerStatus, error) {
|
||||||
|
return nil, nil // no manager behind a bare store: nothing configured
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) DayPlan(ctx context.Context) (DayPlan, error) {
|
||||||
|
return DayPlan{}, errors.New("store: day plan not available via direct store API")
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) ListTools(ctx context.Context, status string) ([]Tool, error) {
|
||||||
|
ts, err := a.s.ListTools(ctx, status)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]Tool, len(ts))
|
||||||
|
for i, t := range ts {
|
||||||
|
out[i] = toTool(t)
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) DeleteTool(ctx context.Context, name string) error {
|
||||||
|
return mapErr(a.s.DeleteTool(ctx, name))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) CaptureTask(ctx context.Context, req CaptureTaskReq) (CaptureTaskResp, error) {
|
||||||
|
res, err := a.s.CaptureTask(ctx, store.Task{
|
||||||
|
CreatedTs: req.Ts,
|
||||||
|
Text: req.Text,
|
||||||
|
Source: req.Source,
|
||||||
|
Evidence: req.Evidence,
|
||||||
|
ExternalID: req.ExternalID,
|
||||||
|
Status: req.Status,
|
||||||
|
Due: req.Due,
|
||||||
|
Weight: req.Weight,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return CaptureTaskResp{}, mapErr(err)
|
||||||
|
}
|
||||||
|
return CaptureTaskResp{ID: res.ID, Created: res.Created, Promoted: res.Promoted}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) ListTasks(ctx context.Context, status string) ([]Task, error) {
|
||||||
|
ts, err := a.s.ListTasks(ctx, status)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]Task, len(ts))
|
||||||
|
for i, t := range ts {
|
||||||
|
out[i] = Task{
|
||||||
|
ID: t.ID,
|
||||||
|
CreatedTs: t.CreatedTs,
|
||||||
|
Text: t.Text,
|
||||||
|
Source: t.Source,
|
||||||
|
Evidence: t.Evidence,
|
||||||
|
ExternalID: t.ExternalID,
|
||||||
|
Status: t.Status,
|
||||||
|
Due: t.Due,
|
||||||
|
Weight: t.Weight,
|
||||||
|
Resolved: t.ResolvedTs,
|
||||||
|
ResolvedBy: t.ResolvedBy,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) SetTaskStatus(ctx context.Context, id int64, status string, ts time.Time, by string) error {
|
||||||
|
return mapErr(a.s.SetTaskStatus(ctx, id, status, ts, by))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) ListProposedRoutines(ctx context.Context) ([]ProposedRoutine, error) {
|
||||||
|
rs, err := a.s.ListProposedRoutines(ctx)
|
||||||
|
if err != nil {
|
||||||
|
return nil, mapErr(err)
|
||||||
|
}
|
||||||
|
out := make([]ProposedRoutine, len(rs))
|
||||||
|
for i, r := range rs {
|
||||||
|
out[i] = ProposedRoutine{
|
||||||
|
ID: r.ID,
|
||||||
|
Action: r.Action,
|
||||||
|
Object: r.Object,
|
||||||
|
IntervalDays: r.IntervalDays,
|
||||||
|
Status: r.Status,
|
||||||
|
CreatedTs: r.CreatedTs.UnixMilli(),
|
||||||
|
}
|
||||||
|
if r.ReminderID != nil {
|
||||||
|
out[i].ReminderID = r.ReminderID
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) DismissProposedRoutine(ctx context.Context, id int64) error {
|
||||||
|
return mapErr(a.s.DismissProposedRoutine(ctx, id))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *storeAPI) AcceptProposedRoutine(ctx context.Context, id int64) error {
|
||||||
|
return mapErr(a.s.AcceptProposedRoutine(ctx, id, time.Now().UTC()))
|
||||||
|
}
|
||||||
|
|
||||||
|
func toTool(t store.Tool) Tool {
|
||||||
|
return Tool{
|
||||||
|
Name: t.Name, Scope: t.Scope, Cmd: t.Cmd, Destructive: t.Destructive,
|
||||||
|
Status: t.Status, Utterance: t.Utterance, Created: t.CreatedTs, Updated: t.UpdatedTs,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func toReminder(r store.Reminder) Reminder {
|
||||||
|
return Reminder{
|
||||||
|
ID: r.ID,
|
||||||
|
CreatedTs: r.CreatedTs,
|
||||||
|
FireTs: r.FireTs,
|
||||||
|
NextFireTs: r.NextFireTs,
|
||||||
|
Payload: r.Payload,
|
||||||
|
Status: r.Status,
|
||||||
|
Cron: r.Cron,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func toNote(n store.Note) Note {
|
||||||
|
return Note{ID: n.ID, Ts: n.Ts, Text: n.Text, Source: n.Source, Score: n.Score}
|
||||||
|
}
|
||||||
|
|
||||||
|
func toNudge(n store.Nudge) Nudge {
|
||||||
|
out := Nudge{
|
||||||
|
ID: n.ID, Ts: n.Ts, Rule: n.Rule, Channel: n.Channel,
|
||||||
|
Message: n.Message, Outcome: n.Outcome,
|
||||||
|
}
|
||||||
|
if n.OutcomeTs.Valid {
|
||||||
|
v := n.OutcomeTs.Int64
|
||||||
|
out.OutcomeTs = &v
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
func toFact(f store.Fact) Fact {
|
||||||
|
out := Fact{
|
||||||
|
ID: f.ID,
|
||||||
|
Ts: f.Ts,
|
||||||
|
Kind: string(f.Kind),
|
||||||
|
Key: f.Key,
|
||||||
|
Value: f.Value,
|
||||||
|
Source: f.Source,
|
||||||
|
Confidence: f.Confidence,
|
||||||
|
}
|
||||||
|
if f.VoidsID.Valid {
|
||||||
|
v := f.VoidsID.Int64
|
||||||
|
out.VoidsID = &v
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// mapErr — store sentinel ↔ ipc sentinel. An unrecognized store error is
|
||||||
|
// wrapped but not mapped (server-side dispatch surfaces it as codeInternal,
|
||||||
|
// keeping internal text off the wire except to the daemon log).
|
||||||
|
func mapErr(err error) error {
|
||||||
|
if err == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
switch {
|
||||||
|
case errors.Is(err, store.ErrNoFact):
|
||||||
|
return ErrNoFact
|
||||||
|
case errors.Is(err, store.ErrConfidence):
|
||||||
|
return ErrConfidence
|
||||||
|
case errors.Is(err, store.ErrVoidsMissing):
|
||||||
|
return ErrVoidsMissing
|
||||||
|
case errors.Is(err, store.ErrNudgeNotFound):
|
||||||
|
return ErrNudgeNotFound
|
||||||
|
case errors.Is(err, store.ErrNudgeOutcome):
|
||||||
|
return ErrNudgeOutcome
|
||||||
|
case errors.Is(err, store.ErrReminderNotFound):
|
||||||
|
return ErrReminderNotFound
|
||||||
|
case errors.Is(err, store.ErrReminderState):
|
||||||
|
return ErrReminderState
|
||||||
|
case errors.Is(err, store.ErrToolNotFound):
|
||||||
|
return ErrToolNotFound
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user