homelab-frontend/internal/sessions/handler.go

// Package sessions provides the internal handler for the sessions service adapter.
// 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.
// episode_id is sha256(session_id + milestone_id + attempt) — callers compute it the
// same way; the handler accepts it on write and returns it on read.
package sessions

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"time"

	"github.com/google/uuid"

	"forgejo.riotpiao.com/rock/homelab-frontend/internal/tracing"
)

const (
	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).
// Sandbox creates/deletes pi-session pods via the SandboxFleet controller
// (primary) with SandboxManager as direct-k8s fallback.
type Handler struct {
	graphqlURL string
	client     *http.Client
	sandbox    *SandboxManager
	fleet      *FleetClient
}

// NewHandler creates a Handler reading upstream URL from env var.
// GRAPHQL_UPSTREAM_URL — default: defaultGraphQLURL
func NewHandler() *Handler {
	gql := os.Getenv("GRAPHQL_UPSTREAM_URL")
	if gql == "" {
		gql = defaultGraphQLURL
	}
	h := &Handler{
		graphqlURL: gql,
		client:     &http.Client{Timeout: 30 * time.Second, Transport: tracing.NewTransport(nil)},
		sandbox:    NewSandboxManager(),
		fleet:      NewFleetClient(),
	}
	h.startIdleReaper()
	return h
}

// startIdleReaper launches a background goroutine that deletes exited pi-session
// pods (bridge.js idle-timeout triggered) while preserving their PVCs.
func (h *Handler) startIdleReaper() {
	intervalStr := os.Getenv("SANDBOX_REAPER_INTERVAL_MINUTES")
	interval := defaultReaperInterval
	if intervalStr != "" {
		var min int
		if n, _ := fmt.Sscanf(intervalStr, "%d", &min); n == 1 && min > 0 {
			interval = time.Duration(min) * time.Minute
		}
	}
	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for range ticker.C {
			h.sandbox.ReapIdlePods(context.Background())
		}
	}()
}

// ServeHTTP routes by X-Upstream-Path set by the dispatcher.
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	path := r.Header.Get("X-Upstream-Path")
	switch path {
	case "/v1/sessions/create":
		h.handleCreate(w, r)
	case "/v1/sessions/list":
		h.handleList(w, r)
	case "/v1/sessions/get":
		h.handleGet(w, r)
	case "/v1/sessions/patch":
		h.handlePatch(w, r)
	case "/v1/sessions/milestones/create":
		h.handleCreateMilestone(w, r)
	case "/v1/sessions/episodes/list":
		h.handleListEpisodes(w, r)
	case "/v1/sessions/episodes/get":
		h.handleGetEpisode(w, r)
	case "/v1/sessions/sandbox":
		h.handleSandbox(w, r)
	default:
		writeJSON(w, http.StatusNotFound, map[string]string{"error": "sessions endpoint not found"})
	}
}

// ── request / response types ──────────────────────────────────────────────────

type createSessionReq struct {
	Name string `json:"name"`
	Goal string `json:"goal,omitempty"`
	Cwd  string `json:"cwd,omitempty"`
}

type sessionResp struct {
	SessionID     string `json:"session_id"`
	Name          string `json:"name"`
	Goal          string `json:"goal,omitempty"`
	Status        string `json:"status"`
	MemoryProject string `json:"memory_project"`
	Cwd           string `json:"cwd,omitempty"`
	CreatedAt     string `json:"created_at"`
}

type patchSessionReq struct {
	Name   *string `json:"name,omitempty"`
	Goal   *string `json:"goal,omitempty"`
	Status *string `json:"status,omitempty"`
}

type createMilestoneReq struct {
	Title    string  `json:"title"`
	OrderIdx int     `json:"order_idx"`
	ParentID *string `json:"parent_id,omitempty"`
}

type milestoneResp struct {
	MilestoneID string  `json:"milestone_id"`
	SessionID   string  `json:"session_id"`
	Title       string  `json:"title"`
	Status      string  `json:"status"`
	OrderIdx    int     `json:"order_idx"`
	ParentID    *string `json:"parent_id,omitempty"`
	CreatedAt   string  `json:"created_at"`
}

// ── handlers ──────────────────────────────────────────────────────────────────

// POST /v1/sessions  (X-Resource: sessions)
func (h *Handler) handleCreate(w http.ResponseWriter, r *http.Request) {
	var req createSessionReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"})
		return
	}
	if req.Name == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "name is required"})
		return
	}

	sessionID := uuid.New().String()
	project := "session-" + sessionID
	now := time.Now().UTC().Format(time.RFC3339)

	summaryBytes, err := json.Marshal(map[string]any{
		"id":             sessionID,
		"name":           req.Name,
		"goal":           req.Goal,
		"status":         "active",
		"cwd":            req.Cwd,
		"memory_project": project,
		"created_at":     now,
	})
	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
	}

	writeJSON(w, http.StatusCreated, sessionResp{
		SessionID:     sessionID,
		Name:          req.Name,
		Goal:          req.Goal,
		Status:        "active",
		MemoryProject: project,
		Cwd:           req.Cwd,
		CreatedAt:     now,
	})
}

// GET /v1/sessions  (X-Resource: sessions)  ?status=active&cwd=/path
func (h *Handler) handleList(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	status := q.Get("status")
	cwd := q.Get("cwd")
	limitStr := q.Get("limit")
	limit := 50
	if limitStr != "" {
		fmt.Sscanf(limitStr, "%d", &limit) //nolint:errcheck
	}
	if limit > 200 {
		limit = 200
	}

	searchQuery := "session"
	if status != "" {
		searchQuery = fmt.Sprintf("session status:%s", status)
	}
	if cwd != "" {
		searchQuery = fmt.Sprintf("%s cwd:%s", searchQuery, cwd)
	}

	// entities query (not hybridSearch) — project arg is optional, so it queries
	// across ALL projects. Required because each session lives in its own
	// session-<uuid> project; there is no single project to hybridSearch against.
	gqlQuery := `query($q:String!,$limit:Int!){
		entities(query:$q,limit:$limit,entityType:SESSION){
			entities{id name summary projectId tCreated}
			searchMode
		}
	}`
	vars := map[string]any{"q": searchQuery, "limit": limit}
	result, err := h.gql(r, gqlQuery, vars)
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "graphql query failed: " + err.Error()})
		return
	}
	writeJSON(w, http.StatusOK, result)
}

// GET /v1/sessions/:id  (X-Resource: session)  ?session_id=<uuid>
func (h *Handler) handleGet(w http.ResponseWriter, r *http.Request) {
	sessionID := r.URL.Query().Get("session_id")
	if sessionID == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "session_id query param required"})
		return
	}

	project := "session-" + sessionID
	gqlQuery := `query($id:String!,$project:String!){
		hybridSearch(query:$id,project:$project,limit:1,entityType:SESSION){
			entities{id name summary tCreated}
		}
	}`
	vars := map[string]any{"id": sessionID, "project": project}
	result, err := h.gql(r, gqlQuery, vars)
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "graphql query failed: " + err.Error()})
		return
	}
	writeJSON(w, http.StatusOK, result)
}

// PATCH /v1/sessions/:id  (X-Resource: session)  ?session_id=<uuid>
func (h *Handler) handlePatch(w http.ResponseWriter, r *http.Request) {
	sessionID := r.URL.Query().Get("session_id")
	if sessionID == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "session_id query param required"})
		return
	}

	var req patchSessionReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"})
		return
	}

	now := time.Now().UTC().Format(time.RFC3339)
	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 {
		current["name"] = *req.Name
	}
	if req.Goal != nil {
		current["goal"] = *req.Goal
	}
	if req.Status != nil {
		current["status"] = *req.Status
		if *req.Status == "archived" {
			current["archived_at"] = now
		}
	}
	summaryBytes, err := json.Marshal(current)
	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
	}
	writeJSON(w, http.StatusOK, map[string]string{"session_id": sessionID, "status": "updated"})
}

// POST /v1/sessions/:id/milestones  (X-Resource: milestones)  ?session_id=<uuid>
func (h *Handler) handleCreateMilestone(w http.ResponseWriter, r *http.Request) {
	sessionID := r.URL.Query().Get("session_id")
	if sessionID == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "session_id query param required"})
		return
	}

	var req createMilestoneReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"})
		return
	}
	if req.Title == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "title is required"})
		return
	}

	milestoneID := uuid.New().String()
	now := time.Now().UTC().Format(time.RFC3339)
	project := "session-" + sessionID

	msContent := map[string]any{
		"type":       "MILESTONE",
		"id":         milestoneID,
		"session_id": sessionID,
		"title":      req.Title,
		"status":     "pending",
		"order_idx":  req.OrderIdx,
		"created_at": now,
	}
	if req.ParentID != nil {
		msContent["parent_id"] = *req.ParentID
	}
	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
	}

	writeJSON(w, http.StatusCreated, milestoneResp{
		MilestoneID: milestoneID,
		SessionID:   sessionID,
		Title:       req.Title,
		Status:      "pending",
		OrderIdx:    req.OrderIdx,
		ParentID:    req.ParentID,
		CreatedAt:   now,
	})
}

// GET /v1/sessions/:id/episodes  (X-Resource: episodes)  ?session_id=<uuid>
func (h *Handler) handleListEpisodes(w http.ResponseWriter, r *http.Request) {
	sessionID := r.URL.Query().Get("session_id")
	if sessionID == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "session_id query param required"})
		return
	}

	limitStr := r.URL.Query().Get("limit")
	limit := 50
	if limitStr != "" {
		fmt.Sscanf(limitStr, "%d", &limit) //nolint:errcheck
	}
	if limit > 200 {
		limit = 200
	}

	project := "session-" + sessionID
	gqlQuery := `query($project:String!,$limit:Int!){
		hybridSearch(query:"episode",project:$project,limit:$limit,entityType:EPISODE){
			entities{id name summary tCreated}
			searchMode
		}
	}`
	vars := map[string]any{"project": project, "limit": limit}
	result, err := h.gql(r, gqlQuery, vars)
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "graphql query failed: " + err.Error()})
		return
	}
	writeJSON(w, http.StatusOK, result)
}

// GET /v1/sessions/:id/episodes/:eid  (X-Resource: episode)
// ?session_id=<uuid>&episode_id=<sha256>
func (h *Handler) handleGetEpisode(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	sessionID := q.Get("session_id")
	episodeID := q.Get("episode_id")
	if sessionID == "" || episodeID == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{
			"error": "session_id and episode_id query params required",
		})
		return
	}

	gqlQuery := `query($id:ID!){
		entity(id:$id){
			id name entityType summary tCreated
			edges{
				relationType fact confidence
				target{name entityType}
			}
		}
	}`
	vars := map[string]any{"id": episodeID}
	result, err := h.gql(r, gqlQuery, vars)
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "graphql query failed: " + err.Error()})
		return
	}
	writeJSON(w, http.StatusOK, result)
}

// ── helpers ───────────────────────────────────────────────────────────────────

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

// 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 nil, err
	}
	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
	}
	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
		}
	}
	if summary == nil {
		summary = map[string]any{}
	}
	return summary, nil
}

// gql executes a GraphQL query against poimen-graphql, forwarding the caller's JWT.
func (h *Handler) gql(r *http.Request, query string, variables map[string]any) (any, error) {
	payload, err := json.Marshal(map[string]any{
		"query":     query,
		"variables": variables,
	})
	if err != nil {
		return nil, fmt.Errorf("marshal: %w", err)
	}

	req, err := http.NewRequestWithContext(r.Context(), http.MethodPost,
		h.graphqlURL, bytes.NewReader(payload))
	if err != nil {
		return nil, fmt.Errorf("build request: %w", err)
	}
	req.Header.Set("Content-Type", "application/json")
	if auth := r.Header.Get("Authorization"); auth != "" {
		req.Header.Set("Authorization", auth)
	}

	resp, err := h.client.Do(req)
	if err != nil {
		return nil, fmt.Errorf("http: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode >= 300 {
		return nil, fmt.Errorf("graphql returned %d", resp.StatusCode)
	}

	var result map[string]any
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return nil, fmt.Errorf("decode: %w", err)
	}
	if errs, ok := result["errors"]; ok && errs != nil {
		b, _ := json.Marshal(errs)
		return nil, fmt.Errorf("graphql errors: %s", b)
	}
	return result, nil
}

// handleSandbox dispatches POST (create/resume) and DELETE (teardown).
// Both take ?session_id= as a query param.
func (h *Handler) handleSandbox(w http.ResponseWriter, r *http.Request) {
	sessionID := r.URL.Query().Get("session_id")
	if sessionID == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "session_id query param required"})
		return
	}

	switch r.Method {
	case http.MethodPost:
		h.handleSandboxCreate(w, r, sessionID)
	case http.MethodDelete:
		h.handleSandboxDelete(w, r, sessionID)
	default:
		writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"})
	}
}

// POST ?session_id= — assign session to SandboxFleet pod (primary) or
// create a dedicated pod via SandboxManager (fallback).
func (h *Handler) handleSandboxCreate(w http.ResponseWriter, r *http.Request, sessionID string) {
	// Primary: fleet controller assigns session to a pre-warmed pod.
	if assign, err := h.fleet.Assign(r.Context(), sessionID); err == nil {
		svcDns := assign.PodSvcDns + fmt.Sprintf(":%d", sandboxPort)
		log.Printf("sandbox: fleet assigned session=%s pod=%s dns=%s", sessionID, assign.PodName, svcDns)
		go h.updateSandboxSummary(r, sessionID, assign.PodName, "", svcDns)
		writeJSON(w, http.StatusOK, map[string]any{
			"session_id": sessionID,
			"pod_name":   assign.PodName,
			"svc_dns":    svcDns,
			"resumed":    false,
			"fleet":      true,
		})
		return
	} else {
		log.Printf("sandbox: fleet unavailable for session=%s (%v), falling back to direct k8s", sessionID, err)
	}

	// Fallback: create a dedicated pod via SandboxManager.
	result, err := h.sandbox.CreateOrResume(r.Context(), sessionID)
	if err != nil {
		log.Printf("sandbox create/resume session=%s: %v", sessionID, err)
		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "sandbox error: " + err.Error()})
		return
	}
	go h.updateSandboxSummary(r, sessionID, result.PodName, result.PVCName, result.SvcDNS)
	log.Printf("sandbox: direct pod session=%s pod=%s resumed=%v dns=%s", sessionID, result.PodName, result.Resumed, result.SvcDNS)
	writeJSON(w, http.StatusOK, result)
}

// updateSandboxSummary merges pod/svc metadata into the session memory_entity (fire-and-forget).
// Uses context.Background() because r.Context() is cancelled once the HTTP handler returns.
func (h *Handler) updateSandboxSummary(r *http.Request, sessionID, podName, pvcName, svcDns string) {
	// Build a detached request carrying just the auth header.
	detached, _ := http.NewRequestWithContext(context.Background(), r.Method, r.URL.String(), nil)
	if auth := r.Header.Get("Authorization"); auth != "" {
		detached.Header.Set("Authorization", auth)
	}
	current, _ := h.fetchEntitySummary(detached, "session-"+sessionID, sessionID, "SESSION")
	if current == nil {
		current = map[string]any{}
	}
	current["pod_name"] = podName
	current["svc_dns"] = svcDns
	current["last_active_at"] = time.Now().UTC().Format(time.RFC3339)
	if pvcName != "" {
		current["pvc_name"] = pvcName
	}
	if b, err := json.Marshal(current); err == nil {
		h.upsertEntity(r, "session-"+sessionID, sessionID, "SESSION", string(b)) //nolint:errcheck
	}
}

// DELETE ?session_id= — release from fleet + delete dedicated pod+svc+pvc if any.
func (h *Handler) handleSandboxDelete(w http.ResponseWriter, r *http.Request, sessionID string) {
	// Release from fleet (cleans up per-session SA; non-fatal if session was fleet-based).
	if err := h.fleet.Release(r.Context(), sessionID); err != nil {
		log.Printf("sandbox: fleet release session=%s: %v (continuing)", sessionID, err)
	}
	// Also attempt direct k8s cleanup (no-ops if resources don't exist).
	if err := h.sandbox.Delete(r.Context(), sessionID); err != nil {
		log.Printf("sandbox delete session=%s: %v", sessionID, err)
	}
	log.Printf("sandbox: released session=%s", sessionID)
	writeJSON(w, http.StatusOK, map[string]string{"session_id": sessionID, "status": "deleted"})
}

// EpisodeID computes the deterministic episode ID per §4.14 Q6:
// sha256(session_id + milestone_id + attempt).
// Exported so callers can verify the value the server would produce.
func EpisodeID(sessionID, milestoneID string, attempt int) string {
	h := sha256.New()
	h.Write([]byte(fmt.Sprintf("%s%s%d", sessionID, milestoneID, attempt)))
	return fmt.Sprintf("%x", h.Sum(nil))
}

func writeJSON(w http.ResponseWriter, status int, data any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	json.NewEncoder(w).Encode(data) //nolint:errcheck
}