feat(sessions): switch writes to GraphQL upsertEntity mutation (#53)
CI / CI (push) Successful in 3m17s

Closes #52

## Summary

Migrates all session/milestone writes from REST `/memory/ingest` to GraphQL `upsertEntity` mutation — direct, typed, no LLM extraction lag.

## Changes

- **Removed**: `memoryURL` field, `defaultMemoryURL` const, `ingest()` REST helper, `"io"` import
- **Added**: `upsertEntity()` GraphQL mutation helper
- **Added**: `fetchEntitySummary()` helper — reads current entity before patch for read-modify-write
- **handlePatch**: now fetches current summary, merges patch fields, re-upserts full blob — no data loss on partial updates
- **handleCreate / handleCreateMilestone**: `json.Marshal` error now returns 500 instead of silently ignoring
- 29 tests pass, 85.4% coverage

## Mutation used

```graphql
mutation {
  upsertEntity(input: {
    project: "session-<uuid>"
    name: "<sessionID | milestoneID>"
    entityType: SESSION | MILESTONE
    summary: "{...json blob...}"
  }) { id created }
}
```

## Single upstream

All reads and writes now go through `graphqlURL` (poimen-graphql:8090). `MEMORY_UPSTREAM_URL` env var no longer read by this adapter.

---------

Co-authored-by: Admin Bot <[email protected]>
Reviewed-on: #53
This commit was merged in pull request #53.
This commit is contained in:
2026-09-28 02:24:11 +00:00
co-authored by Admin Bot
parent a24e4f7174
commit 50bcd19634
2 changed files with 241 additions and 241 deletions
+94 -59
View File
@@ -1,7 +1,8 @@
// Package sessions provides the internal handler for the sessions service adapter.
// It bridges the api-gw REST surface to:
// - poimen-memory /memory/ingest (writes — SESSION, MILESTONE, EPISODE entities)
// - poimen-graphql /graphql (reads — hybridSearch, entity queries)
// It bridges the api-gw REST surface to poimen-graphql for all reads and writes:
//
// - Reads: hybridSearch / entity queries
// - Writes: upsertEntity mutation (direct, no LLM extraction)
//
// All IDs passed by the caller (session_id, episode_id) go as query params because
// the ServiceAdapter dispatcher does not substitute URL path parameters.
@@ -14,7 +15,6 @@ import (
"crypto/sha256"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"time"
@@ -25,31 +25,24 @@ import (
)
const (
defaultMemoryURL = "http://poimen-memory.poimen.svc.cluster.local:8080"
defaultGraphQLURL = "http://poimen-graphql.poimen.svc.cluster.local:8090/graphql"
)
// Handler handles session REST endpoints via internal service calls.
// All traffic (reads and writes) goes to graphqlURL (poimen-graphql).
type Handler struct {
memoryURL string
graphqlURL string
client *http.Client
}
// NewHandler creates a Handler reading upstream URLs from env vars.
// MEMORY_UPSTREAM_URL — default: defaultMemoryURL
// NewHandler creates a Handler reading upstream URL from env var.
// GRAPHQL_UPSTREAM_URL — default: defaultGraphQLURL
func NewHandler() *Handler {
mem := os.Getenv("MEMORY_UPSTREAM_URL")
if mem == "" {
mem = defaultMemoryURL
}
gql := os.Getenv("GRAPHQL_UPSTREAM_URL")
if gql == "" {
gql = defaultGraphQLURL
}
return &Handler{
memoryURL: mem,
graphqlURL: gql,
client: &http.Client{Timeout: 30 * time.Second, Transport: tracing.NewTransport(nil)},
}
@@ -136,8 +129,7 @@ func (h *Handler) handleCreate(w http.ResponseWriter, r *http.Request) {
project := "session-" + sessionID
now := time.Now().UTC().Format(time.RFC3339)
contentBytes, _ := json.Marshal(map[string]any{
"type": "SESSION",
summaryBytes, err := json.Marshal(map[string]any{
"id": sessionID,
"name": req.Name,
"goal": req.Goal,
@@ -146,9 +138,12 @@ func (h *Handler) handleCreate(w http.ResponseWriter, r *http.Request) {
"memory_project": project,
"created_at": now,
})
content := string(contentBytes)
if err := h.ingest(r, project, sessionID, content); err != nil {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "ingest failed: " + err.Error()})
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "marshal failed: " + err.Error()})
return
}
if _, err := h.upsertEntity(r, project, sessionID, "SESSION", string(summaryBytes)); err != nil {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "upsert failed: " + err.Error()})
return
}
@@ -238,28 +233,35 @@ func (h *Handler) handlePatch(w http.ResponseWriter, r *http.Request) {
}
now := time.Now().UTC().Format(time.RFC3339)
patch := map[string]any{
"type": "SESSION",
"id": sessionID,
"updated_at": now,
project := "session-" + sessionID
// Read current summary so the upsert carries all existing fields.
current, err := h.fetchEntitySummary(r, project, sessionID, "SESSION")
if err != nil {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "fetch failed: " + err.Error()})
return
}
current["updated_at"] = now
if req.Name != nil {
patch["name"] = *req.Name
current["name"] = *req.Name
}
if req.Goal != nil {
patch["goal"] = *req.Goal
current["goal"] = *req.Goal
}
if req.Status != nil {
patch["status"] = *req.Status
current["status"] = *req.Status
if *req.Status == "archived" {
patch["archived_at"] = now
current["archived_at"] = now
}
}
contentBytes, _ := json.Marshal(patch)
summaryBytes, err := json.Marshal(current)
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "marshal failed: " + err.Error()})
return
}
project := "session-" + sessionID
if err := h.ingest(r, project, sessionID+"-patch", string(contentBytes)); err != nil {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "ingest failed: " + err.Error()})
if _, err := h.upsertEntity(r, project, sessionID, "SESSION", string(summaryBytes)); err != nil {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "upsert failed: " + err.Error()})
return
}
writeJSON(w, http.StatusOK, map[string]string{"session_id": sessionID, "status": "updated"})
@@ -299,10 +301,13 @@ func (h *Handler) handleCreateMilestone(w http.ResponseWriter, r *http.Request)
if req.ParentID != nil {
msContent["parent_id"] = *req.ParentID
}
contentBytes, _ := json.Marshal(msContent)
content := string(contentBytes)
if err := h.ingest(r, project, milestoneID, content); err != nil {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "ingest failed: " + err.Error()})
summaryBytes, err := json.Marshal(msContent)
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "marshal failed: " + err.Error()})
return
}
if _, err := h.upsertEntity(r, project, milestoneID, "MILESTONE", string(summaryBytes)); err != nil {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "upsert failed: " + err.Error()})
return
}
@@ -383,40 +388,70 @@ func (h *Handler) handleGetEpisode(w http.ResponseWriter, r *http.Request) {
// ── helpers ───────────────────────────────────────────────────────────────────
// ingest calls POST /memory/ingest on the memory service, forwarding the caller's JWT.
func (h *Handler) ingest(r *http.Request, project, recordID, content string) error {
body, err := json.Marshal(map[string]any{
"project": project,
"source": "sessions-api",
"records": []map[string]string{
{"id": recordID, "content": content},
// upsertEntity runs a GraphQL upsertEntity mutation on poimen-graphql.
// Writes are direct (no LLM extraction). Returns the entity id assigned by the server.
// The name field is used as the stable upsert key within (project, entityType).
func (h *Handler) upsertEntity(r *http.Request, project, name, entityType, summary string) (string, error) {
mutation := `mutation($input:UpsertEntityInput!){
upsertEntity(input:$input){ id created }
}`
vars := map[string]any{
"input": map[string]any{
"project": project,
"name": name,
"entityType": entityType,
"summary": summary,
},
})
}
result, err := h.gql(r, mutation, vars)
if err != nil {
return fmt.Errorf("marshal: %w", err)
return "", err
}
data, _ := result.(map[string]any)
d, _ := data["data"].(map[string]any)
entity, _ := d["upsertEntity"].(map[string]any)
id, _ := entity["id"].(string)
return id, nil
}
req, err := http.NewRequestWithContext(r.Context(), http.MethodPost,
h.memoryURL+"/memory/ingest", bytes.NewReader(body))
// fetchEntitySummary fetches the current summary blob of an entity by (project, name, entityType).
// Returns a parsed map so callers can merge patch fields before re-upserting.
// Returns empty map (not error) if no entity found.
func (h *Handler) fetchEntitySummary(r *http.Request, project, name, entityType string) (map[string]any, error) {
gqlQuery := `query($q:String!,$project:String!,$entityType:GqlEntityType!,$limit:Int!){
hybridSearch(query:$q,project:$project,limit:$limit,entityType:$entityType){
entities{id name summary tCreated}
}
}`
vars := map[string]any{
"q": name,
"project": project,
"entityType": entityType,
"limit": 1,
}
result, err := h.gql(r, gqlQuery, vars)
if err != nil {
return fmt.Errorf("build request: %w", err)
return nil, err
}
req.Header.Set("Content-Type", "application/json")
if auth := r.Header.Get("Authorization"); auth != "" {
req.Header.Set("Authorization", auth)
data, _ := result.(map[string]any)
d, _ := data["data"].(map[string]any)
search, _ := d["hybridSearch"].(map[string]any)
entities, _ := search["entities"].([]any)
if len(entities) == 0 {
return map[string]any{}, nil
}
resp, err := h.client.Do(req)
if err != nil {
return fmt.Errorf("http: %w", err)
entity, _ := entities[0].(map[string]any)
summaryStr, _ := entity["summary"].(string)
var summary map[string]any
if summaryStr != "" {
if err := json.Unmarshal([]byte(summaryStr), &summary); err != nil {
return map[string]any{}, nil // corrupt summary — start fresh
}
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
return fmt.Errorf("memory ingest returned %d: %s", resp.StatusCode, body)
if summary == nil {
summary = map[string]any{}
}
return nil
return summary, nil
}
// gql executes a GraphQL query against poimen-graphql, forwarding the caller's JWT.