so-many-crows/server.go

package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"math"
	"net"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"strconv"
	"sync/atomic"
	"time"
)

const (
	maxBody   = 64 << 10 // POST bodies
	heartbeat = 15 * time.Second
	webDir    = "web"
)

// Server is the web front door: the same pipeline as the command line,
// with web approvals and an SSE hub instead of stdin and stdout.
type Server struct {
	p          *Pipeline
	approver   *WebApprover
	hub        *Hub
	nests      *NestStore
	crows      *CrowStore
	msgs       *MessageStore
	ctx        context.Context // cancelled on shutdown; lists and streams end with it
	beat       time.Duration
	askTimeout atomic.Int64 // nanos: an ask_owner unanswered past this reads as rejected (#41, fail safe like WebApprover). Atomic because handlers and the watchdog emit path read it concurrently, and tests narrow the window while the server runs (#45).
}

func NewServer(ctx context.Context, p *Pipeline, approver *WebApprover, nests *NestStore) *Server {
	s := &Server{p: p, approver: approver, hub: NewHub(), nests: nests, ctx: ctx, beat: heartbeat}
	s.askTimeout.Store(int64(10 * time.Minute))
	return s
}

// WithMessages attaches the nest conversation log (chatroom path).
func (s *Server) WithMessages(m *MessageStore) *Server { s.msgs = m; return s }

// WithCrows attaches the spun-crow registry (the Boss writes through it) and
// lets registry crows be nest members (#43).
func (s *Server) WithCrows(c *CrowStore) *Server {
	s.crows = c
	if c != nil && s.nests != nil {
		s.nests.AllowCrows(c.Has)
	}
	return s
}

func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("POST /api/lists", s.postList)
	mux.HandleFunc("GET /api/events", s.events)
	mux.HandleFunc("POST /api/verdicts", s.postVerdict)
	mux.HandleFunc("GET /api/trust", s.trust)
	mux.HandleFunc("GET /api/personas", s.crowPersonas)
	mux.HandleFunc("GET /api/crows", s.getCrowRegistry)
	mux.HandleFunc("GET /api/nests", s.getNests)
	mux.HandleFunc("POST /api/nests", s.postNest)
	mux.HandleFunc("GET /api/nests/{nest_id}/messages", s.getNestMessages)
	mux.HandleFunc("POST /api/nests/{nest_id}/messages", s.postNestMessage)
	if st, err := os.Stat(webDir); err == nil && st.IsDir() {
		mux.Handle("GET /", http.FileServer(http.Dir(webDir)))
	} else {
		mux.HandleFunc("GET /{$}", func(w http.ResponseWriter, r *http.Request) {
			fmt.Fprintln(w, "So Many Crows is listening. The web front end lands in ./web later.")
		})
	}
	return mux
}

func cmdServe(args []string) {
	fs := flag.NewFlagSet("serve", flag.ExitOnError)
	addr := fs.String("addr", "127.0.0.1:8080", "address to listen on (localhost only by default: there is no login yet)")
	timeout := fs.Duration("approval-timeout", 10*time.Minute, "unanswered approvals count as rejected after this long")
	fs.Parse(args)

	g := DefaultGuardrails()
	ledger, err := LoadLedger(filepath.Join(dataDir, "ledger.json"))
	must(err)
	flog, err := OpenFlightLog(filepath.Join(dataDir, "flight.jsonl"))
	must(err)
	defer flog.Close()
	nests, err := LoadNests(filepath.Join(dataDir, "nests.json"))
	must(err)
	msgs, err := LoadMessageStore(filepath.Join(dataDir, "messages"))
	must(err)
	spun, err := LoadCrowStore(filepath.Join(dataDir, "crows.json"))
	must(err)

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
	defer stop()
	brain := NewBrain()
	approver := NewWebApprover(*timeout)
	p := &Pipeline{Flock: NewFlock(brain, g, flog), Ledger: ledger, Log: flog, G: g, Approver: approver}
	jev := NewJev()
	p.Flock.SetClassifier(jev)
	if jev != nil {
		fmt.Printf("Sentry triage: %s, LLM fallback\n", jev.Name())
	}
	s := NewServer(ctx, p, approver, nests).WithMessages(msgs).WithCrows(spun)

	srv := &http.Server{Addr: *addr, Handler: s.Handler(), ReadHeaderTimeout: 10 * time.Second}
	go func() {
		<-ctx.Done()
		shut, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		srv.Shutdown(shut)
	}()
	fmt.Printf("Brain: %s · listening on %s\n", brain.Name(), *addr)
	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
		must(err)
	}
	if err := ledger.Save(); err != nil {
		fmt.Fprintln(os.Stderr, "could not save ledger:", err)
	}
}

func writeJSON(w http.ResponseWriter, code int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	json.NewEncoder(w).Encode(v)
}

func httpError(w http.ResponseWriter, code int, msg string) {
	writeJSON(w, code, map[string]string{"message": msg})
}

// decode reads a size-capped JSON body.
func decode(w http.ResponseWriter, r *http.Request, v any) bool {
	r.Body = http.MaxBytesReader(w, r.Body, maxBody)
	if err := json.NewDecoder(r.Body).Decode(v); err != nil {
		var tooBig *http.MaxBytesError
		if errors.As(err, &tooBig) {
			httpError(w, http.StatusRequestEntityTooLarge, fmt.Sprintf("body over %d bytes", maxBody))
		} else {
			httpError(w, http.StatusBadRequest, "body must be JSON: "+err.Error())
		}
		return false
	}
	return true
}

func (s *Server) postList(w http.ResponseWriter, r *http.Request) {
	var body struct {
		Text      string `json:"text"`
		Requester string `json:"requester"`
		NestID    string `json:"nest_id"`
	}
	if !decode(w, r, &body) {
		return
	}
	if body.NestID == "" {
		body.NestID = defaultNestID
	}
	nest, found := s.nests.Get(body.NestID)
	if !found {
		httpError(w, http.StatusNotFound, fmt.Sprintf("There is no nest called %q.", body.NestID))
		return
	}
	if body.Requester == "" {
		body.Requester, _, _ = net.SplitHostPort(r.RemoteAddr)
	}

	ok, msg, wait := s.p.Ledger.Admit(body.Requester, time.Now(), s.p.G)
	if !ok {
		cooldown := int(math.Ceil(wait.Seconds()))
		s.p.Log.Write(Event{Kind: "denied", Requester: body.Requester, Note: msg})
		s.p.save()
		// Only a list that is still flying can hear about it; otherwise the 429 says it all.
		for _, id := range s.hub.ActiveLists(body.Requester) {
			s.hub.Publish(id, "requester.pranked", map[string]any{"message": msg, "cooldown_s": cooldown})
		}
		writeJSON(w, http.StatusTooManyRequests, map[string]any{"message": msg, "cooldown_s": cooldown})
		return
	}
	s.p.save()

	items, err := parseTodo(body.Text, s.p.G)
	switch {
	case err != nil:
		httpError(w, http.StatusBadRequest, err.Error())
		return
	case len(items) == 0:
		httpError(w, http.StatusBadRequest, "That list is empty.")
		return
	case len(items) > s.p.G.MaxItemsPerList:
		httpError(w, http.StatusBadRequest, fmt.Sprintf("That list has %d items. The flock takes at most %d at a time.", len(items), s.p.G.MaxItemsPerList))
		return
	}

	id := newListID()
	s.hub.Open(id, body.Requester)
	l := List{ID: id, Requester: body.Requester, Source: "web", Items: items, Nest: nest, Parallel: true}
	go s.p.Run(s.ctx, l, s.hub.Emitter(id))
	writeJSON(w, http.StatusAccepted, map[string]string{"list_id": id})
}

func newListID() string {
	b := make([]byte, 6)
	rand.Read(b)
	return hex.EncodeToString(b)
}

func (s *Server) events(w http.ResponseWriter, r *http.Request) {
	listID := r.URL.Query().Get("list_id")
	nestID := r.URL.Query().Get("nest_id")
	if listID == "" && nestID != "" {
		if _, found := s.nests.Get(nestID); !found {
			httpError(w, http.StatusNotFound, "no such nest_id")
			return
		}
		s.hub.OpenNest(nestID)
		listID = nestID
	}
	lastID, _ := strconv.Atoi(r.Header.Get("Last-Event-ID"))
	past, ch, ok := s.hub.Subscribe(listID, lastID)
	if !ok {
		httpError(w, http.StatusNotFound, "no such list_id")
		return
	}
	defer s.hub.Unsubscribe(listID, ch)
	flusher, canFlush := w.(http.Flusher)
	if !canFlush {
		httpError(w, http.StatusInternalServerError, "streaming unsupported")
		return
	}

	h := w.Header()
	h.Set("Content-Type", "text/event-stream")
	h.Set("Cache-Control", "no-cache")
	h.Set("Connection", "keep-alive")
	h.Set("X-Accel-Buffering", "no")
	w.WriteHeader(http.StatusOK)

	send := func(m sseMsg) bool {
		_, err := fmt.Fprintf(w, "id: %d\nevent: %s\ndata: %s\n\n", m.ID, m.Name, m.Data)
		flusher.Flush()
		return err == nil
	}
	io.WriteString(w, ": caw\n\n")
	flusher.Flush()
	for _, m := range past {
		if !send(m) {
			return
		}
	}

	tick := time.NewTicker(s.beat)
	defer tick.Stop()
	for {
		select {
		case m, open := <-ch:
			if !open || !send(m) {
				return
			}
		case <-tick.C:
			if _, err := io.WriteString(w, ": heartbeat\n\n"); err != nil {
				return
			}
			flusher.Flush()
		case <-r.Context().Done():
			return
		case <-s.ctx.Done():
			return
		}
	}
}

func (s *Server) postVerdict(w http.ResponseWriter, r *http.Request) {
	var body struct {
		ItemID  string `json:"item_id"`
		Target  string `json:"target"`
		Approve *bool  `json:"approve"`
	}
	if !decode(w, r, &body) {
		return
	}
	if body.ItemID == "" || (body.Target != "triage" && body.Target != "output") || body.Approve == nil {
		httpError(w, http.StatusBadRequest, `need item_id, target ("triage" or "output") and approve (bool)`)
		return
	}
	if !s.approver.Resolve(body.ItemID, body.Target, *body.Approve) {
		httpError(w, http.StatusNotFound, "nothing is waiting on that verdict")
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

func (s *Server) trust(w http.ResponseWriter, r *http.Request) {
	data, err := s.p.Ledger.MarshalSnapshot()
	if err != nil {
		httpError(w, http.StatusInternalServerError, err.Error())
		return
	}
	w.Header().Set("Content-Type", "application/json")
	w.Write(data)
}

func (s *Server) getNests(w http.ResponseWriter, r *http.Request) {
	writeJSON(w, http.StatusOK, map[string]any{"roster": roster, "nests": s.nests.All()})
}

func (s *Server) postNest(w http.ResponseWriter, r *http.Request) {
	var body struct {
		Name  string   `json:"name"`
		Crows []string `json:"crows"`
	}
	if !decode(w, r, &body) {
		return
	}
	n, err := s.nests.Create(body.Name, body.Crows)
	var bad *NestError
	switch {
	case errors.As(err, &bad) && bad.Clash:
		httpError(w, http.StatusConflict, bad.Msg)
	case errors.As(err, &bad):
		httpError(w, http.StatusBadRequest, bad.Msg)
	case err != nil:
		httpError(w, http.StatusInternalServerError, "could not save the nest: "+err.Error())
	default:
		writeJSON(w, http.StatusCreated, n)
	}
}

// crowPersonas lists each persona crow for the UI (name, epithet, strength,
// backstory). The spun-crow registry lives at GET /api/crows (crowreg.go).
func (s *Server) crowPersonas(w http.ResponseWriter, r *http.Request) {
	out := make([]map[string]string, 0, len(rosterOrder))
	for _, n := range rosterOrder {
		p := personas[n]
		out = append(out, map[string]string{"name": p.Name, "epithet": p.Epithet, "strength": p.Strength, "backstory": p.Backstory})
	}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(out)
}