// pathwayd is the pathway memory service. Wraps internal/pathway's // Store with HTTP routes for the Mem0-style operations defined in // ADR-004. // // Routes (all under /pathway): // POST /pathway/add — new trace with fresh UID // POST /pathway/add_idempotent — UID-keyed add or replay-bump // POST /pathway/update — replace content for an existing UID // POST /pathway/revise — new revision linked to predecessor // POST /pathway/retire — mark trace retired (excluded from search) // GET /pathway/get/{uid} — fetch one trace (incl. retired) // GET /pathway/history/{uid} — backward chain via predecessor links // POST /pathway/search — filter-based listing // GET /pathway/stats — total/active/retired counters // // Persistence: optional. Empty [pathwayd].persist_path = in-memory // only (matches vectord G1's pattern). Set a path for durable // per-trace JSONL append. package main import ( "encoding/json" "errors" "flag" "log/slog" "net/http" "os" "strings" "github.com/go-chi/chi/v5" "git.agentview.dev/profit/golangLAKEHOUSE/internal/pathway" "git.agentview.dev/profit/golangLAKEHOUSE/internal/shared" ) const maxRequestBytes = 4 << 20 // 4 MiB cap on request bodies func main() { configPath := flag.String("config", "lakehouse.toml", "path to TOML config") flag.Parse() cfg, err := shared.LoadConfig(*configPath) if err != nil { slog.Error("config", "err", err) os.Exit(1) } // Persistence is optional — empty path = in-memory ephemeral. var persistor *pathway.Persistor if cfg.Pathwayd.PersistPath != "" { persistor, err = pathway.NewPersistor(cfg.Pathwayd.PersistPath) if err != nil { slog.Error("pathway persistor", "err", err) os.Exit(1) } } store := pathway.NewStore(persistor) if persistor != nil { n, err := store.Load() if err != nil { slog.Warn("pathway load", "err", err, "loaded", n) } else { slog.Info("pathway loaded", "events", n, "path", cfg.Pathwayd.PersistPath) } } h := &handlers{store: store} if err := shared.Run("pathwayd", cfg.Pathwayd.Bind, h.register, cfg.Auth); err != nil { slog.Error("server", "err", err) os.Exit(1) } } type handlers struct { store *pathway.Store } func (h *handlers) register(r chi.Router) { r.Post("/pathway/add", h.handleAdd) r.Post("/pathway/add_idempotent", h.handleAddIdempotent) r.Post("/pathway/update", h.handleUpdate) r.Post("/pathway/revise", h.handleRevise) r.Post("/pathway/retire", h.handleRetire) r.Get("/pathway/get/{uid}", h.handleGet) r.Get("/pathway/history/{uid}", h.handleHistory) r.Post("/pathway/search", h.handleSearch) r.Get("/pathway/stats", h.handleStats) } // ── request shapes ─────────────────────────────────────────────── type addRequest struct { Content json.RawMessage `json:"content"` Tags []string `json:"tags,omitempty"` } type addIdempotentRequest struct { UID string `json:"uid"` Content json.RawMessage `json:"content"` Tags []string `json:"tags,omitempty"` } type updateRequest struct { UID string `json:"uid"` Content json.RawMessage `json:"content"` } type reviseRequest struct { PredecessorUID string `json:"predecessor_uid"` Content json.RawMessage `json:"content"` Tags []string `json:"tags,omitempty"` } type retireRequest struct { UID string `json:"uid"` } type searchRequest struct { Tag string `json:"tag,omitempty"` ContentContains string `json:"content_contains,omitempty"` CreatedAfterNs int64 `json:"created_after_ns,omitempty"` CreatedBeforeNs int64 `json:"created_before_ns,omitempty"` IncludeRetired bool `json:"include_retired,omitempty"` } // ── handlers ──────────────────────────────────────────────────── func (h *handlers) handleAdd(w http.ResponseWriter, r *http.Request) { var req addRequest if !decodeJSON(w, r, &req) { return } tr, err := h.store.Add(req.Content, req.Tags...) if writeStoreError(w, err) { return } writeJSON(w, http.StatusCreated, tr) } func (h *handlers) handleAddIdempotent(w http.ResponseWriter, r *http.Request) { var req addIdempotentRequest if !decodeJSON(w, r, &req) { return } tr, err := h.store.AddIdempotent(req.UID, req.Content, req.Tags...) if writeStoreError(w, err) { return } writeJSON(w, http.StatusOK, tr) } func (h *handlers) handleUpdate(w http.ResponseWriter, r *http.Request) { var req updateRequest if !decodeJSON(w, r, &req) { return } if err := h.store.Update(req.UID, req.Content); writeStoreError(w, err) { return } writeJSON(w, http.StatusOK, map[string]any{"status": "updated"}) } func (h *handlers) handleRevise(w http.ResponseWriter, r *http.Request) { var req reviseRequest if !decodeJSON(w, r, &req) { return } tr, err := h.store.Revise(req.PredecessorUID, req.Content, req.Tags...) if writeStoreError(w, err) { return } writeJSON(w, http.StatusCreated, tr) } func (h *handlers) handleRetire(w http.ResponseWriter, r *http.Request) { var req retireRequest if !decodeJSON(w, r, &req) { return } if err := h.store.Retire(req.UID); writeStoreError(w, err) { return } w.WriteHeader(http.StatusNoContent) } func (h *handlers) handleGet(w http.ResponseWriter, r *http.Request) { uid := chi.URLParam(r, "uid") tr, err := h.store.Get(uid) if writeStoreError(w, err) { return } writeJSON(w, http.StatusOK, tr) } func (h *handlers) handleHistory(w http.ResponseWriter, r *http.Request) { uid := chi.URLParam(r, "uid") chain, err := h.store.History(uid) if writeStoreError(w, err) { return } writeJSON(w, http.StatusOK, map[string]any{ "chain": chain, "length": len(chain), }) } func (h *handlers) handleSearch(w http.ResponseWriter, r *http.Request) { var req searchRequest if !decodeJSON(w, r, &req) { return } results := h.store.Search(pathway.SearchFilter{ Tag: req.Tag, ContentContains: req.ContentContains, CreatedAfterNs: req.CreatedAfterNs, CreatedBeforeNs: req.CreatedBeforeNs, IncludeRetired: req.IncludeRetired, }) writeJSON(w, http.StatusOK, map[string]any{ "results": results, "count": len(results), }) } func (h *handlers) handleStats(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, h.store.Stats()) } // ── helpers ──────────────────────────────────────────────────── func decodeJSON(w http.ResponseWriter, r *http.Request, v any) bool { defer r.Body.Close() r.Body = http.MaxBytesReader(w, r.Body, maxRequestBytes) if err := json.NewDecoder(r.Body).Decode(v); err != nil { var maxErr *http.MaxBytesError if errors.As(err, &maxErr) || strings.Contains(err.Error(), "http: request body too large") { http.Error(w, "body too large", http.StatusRequestEntityTooLarge) return false } http.Error(w, "decode body: "+err.Error(), http.StatusBadRequest) return false } return true } func writeJSON(w http.ResponseWriter, code int, v any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(code) if err := json.NewEncoder(w).Encode(v); err != nil { slog.Warn("pathway write json", "err", err) } } // writeStoreError maps internal/pathway sentinel errors to HTTP // status codes. Returns true if a response was written (caller // should return). Returns false on success (caller continues). func writeStoreError(w http.ResponseWriter, err error) bool { if err == nil { return false } switch { case errors.Is(err, pathway.ErrNotFound): http.Error(w, err.Error(), http.StatusNotFound) case errors.Is(err, pathway.ErrPredecessorMissing): http.Error(w, err.Error(), http.StatusNotFound) case errors.Is(err, pathway.ErrEmptyUID), errors.Is(err, pathway.ErrInvalidContent): http.Error(w, err.Error(), http.StatusBadRequest) case errors.Is(err, pathway.ErrCycle): http.Error(w, err.Error(), http.StatusConflict) default: slog.Error("pathway store", "err", err) http.Error(w, "internal", http.StatusInternalServerError) } return true }