brain/internal/handlers/signals.go

package handlers

import (
	"context"
	"encoding/json"
	"github.com/andrew/go-computer/internal/db"
	"github.com/andrew/go-computer/internal/outbound"
	"github.com/andrew/go-computer/internal/signals"
	"net/http"
	"strings"
	"time"
)

func HandleSignalsList(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	if _, e := bookmark(r, id, false); e != nil {
		respond(w, 0, nil, e)
		return
	}
	sources, e := db.Many(r.Context(), database(r), `SELECT to_jsonb(s) FROM "SignalSource" s WHERE "bookmarkId"=$1 AND "ownerId"=$2 ORDER BY "createdAt"`, id, user(r))
	if e != nil {
		respond(w, 0, nil, e)
		return
	}
	events, e := db.Many(r.Context(), database(r), `SELECT to_jsonb(s) FROM "Signal" s WHERE "bookmarkId"=$1 AND "ownerId"=$2 ORDER BY "occurredAt" DESC LIMIT 50`, id, user(r))
	respond(w, 200, object{"sources": sources, "signals": events, "types": signals.PublicSourceTypes()}, e)
}
func HandleSignalsCreate(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	if _, e := owned(r, database(r), "Bookmark", id, "ownerId"); e != nil {
		respond(w, 0, nil, e)
		return
	}
	b, ok := read(w, r)
	if !ok {
		return
	}
	if str(b, "action") == "check" {
		// Review COR-4: a manual "Check now" must not race with the scheduler
		// or another click. Take the SAME advisory lease the scheduler uses —
		// identical key, held on a dedicated connection because session-level
		// advisory locks are connection-bound (a pooled lock/unlock could land
		// on different connections and leak the lock) — and answer 409 while
		// it is held. The 45-second cap keeps the pass inside the server's
		// WriteTimeout so it cannot be cut off mid-mutation.
		// Review M-2: a manual check is a 45-second outbound pass; it draws
		// from the same budget as previews/IMAP/CardDAV so repeated clicks
		// cannot multiply concurrent mailbox connections.
		release, retry, ok := outbound.Reserve("signals:"+user(r), 10, time.Minute)
		if !ok {
			outboundBusy(w, retry)
			return
		}
		defer release()
		ctx := r.Context()
		leaseKey := "signal-poll:" + user(r)
		conn, e := database(r).Conn(ctx)
		if e != nil {
			respond(w, 0, nil, e)
			return
		}
		defer conn.Close()
		var locked bool
		if e := conn.QueryRowContext(ctx, `SELECT pg_try_advisory_lock(hashtextextended($1,0))`, leaseKey).Scan(&locked); e != nil {
			respond(w, 0, nil, e)
			return
		}
		if !locked {
			respond(w, 0, nil, conflict("A check is already running"))
			return
		}
		defer conn.ExecContext(context.WithoutCancel(ctx), `SELECT pg_advisory_unlock(hashtextextended($1,0))`, leaseKey)
		job, cancel := context.WithTimeout(ctx, 45*time.Second)
		defer cancel()
		res, e := signals.RunSignalChecks(job, database(r), user(r), id)
		if e != nil {
			respond(w, 0, nil, e)
			return
		}
		// Review CR-16: a manual check answers 200 with the structured
		// per-source outcome (detail lives in each source's lastStatus);
		// failed sources are counted, not folded into a blanket 500.
		respond(w, 200, object{"checked": res.Sources, "created": res.Created, "failed": res.Failed}, nil)
		return
	}
	if str(b, "type") != "email" {
		respond(w, 0, nil, bad("Unknown signal type"))
		return
	}
	cfg, ok := b["config"].(map[string]any)
	if !ok {
		respond(w, 0, nil, bad("Config must be an object"))
		return
	}
	clean := object{}
	for _, key := range []string{"from", "subject"} {
		if v, ok := cfg[key]; ok {
			s, ok := v.(string)
			if !ok {
				respond(w, 0, nil, bad("Config values must be text"))
				return
			}
			rs := []rune(strings.TrimSpace(s))
			if len(rs) > 200 {
				rs = rs[:200]
			}
			clean[key] = string(rs)
		}
	}
	raw, _ := json.Marshal(clean)
	v, e := db.Insert(r.Context(), database(r), "SignalSource", object{"id": db.NewID(), "bookmarkId": id, "ownerId": user(r), "type": "email", "config": string(raw)})
	respond(w, 201, v, e)
}
func HandleSignalSourcesUpdate(w http.ResponseWriter, r *http.Request) {
	b, ok := read(w, r)
	if !ok {
		return
	}
	if _, ok = b["enabled"].(bool); !ok {
		respond(w, 0, nil, bad("Enabled must be boolean"))
		return
	}
	v, e := db.Update(r.Context(), database(r), "SignalSource", r.PathValue("id"), "ownerId", user(r), object{"enabled": b["enabled"], "updatedAt": time.Now().UTC()})
	respond(w, 200, v, e)
}
func HandleSignalSourcesDelete(w http.ResponseWriter, r *http.Request) {
	respond(w, 200, OK{true}, remove(r, "SignalSource", r.PathValue("id"), "ownerId"))
}