unibus/pkg/membership/server.go

package membership

import (
	"bytes"
	"context"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"strconv"
	"strings"
	"time"

	cs "fn-registry/functions/cybersecurity"

	"golang.org/x/time/rate"

	"github.com/enmanuel/unibus/pkg/blobstore"
	"github.com/enmanuel/unibus/pkg/frame"
	"github.com/nats-io/nats.go/jetstream"
)

// Body-size ceilings for the control plane. They bound how much an unauthenticated
// peer can make the server buffer in RAM before the request is even authenticated
// (the signature is verified over the full body, so the body must be read — but
// not unboundedly). maxControlBodyBytes covers JSON metadata requests; /blobs gets
// a separate, larger ceiling because media ciphertext is legitimately bigger. A
// request whose declared Content-Length already exceeds its ceiling is rejected
// before a single byte is buffered.
const (
	maxControlBodyBytes = 1 << 20  // 1 MiB for JSON control-plane requests
	maxBlobBytes        = 16 << 20 // 16 MiB for a single media blob upload
	// MaxHeaderBytes caps request header size; wired into the http.Server by the
	// command. Exported so the bound lives next to its body-size siblings.
	MaxHeaderBytes = 1 << 20 // 1 MiB
	// maxInflightBytes is the GLOBAL cap on request-body bytes buffered across all
	// concurrent requests (audit N2). The per-request ceilings above bound one
	// request; this bounds the sum, so a concurrent (even multi-IP) flood of
	// max-size uploads cannot drive the resident set without limit. 128 MiB allows
	// ~8 concurrent 16 MiB blob uploads or ~128 concurrent control requests before
	// further POSTs are shed with 503 — generous for an interactive bus, bounded
	// for an attacker.
	maxInflightBytes = 128 << 20 // 128 MiB
)

// Per-IP rate-limit defaults for the control plane. Tuned for an interactive
// human/agent bus rather than a high-QPS API: a steady ~20 req/s with a burst of
// 40 absorbs a chat client's bursty polling while throttling a flood. Loopback
// dev stacks pass r<=0 to disable limiting entirely.
const (
	defaultRatePerSec = rate.Limit(20)
	defaultRateBurst  = 40
	rateBucketTTL     = 10 * time.Minute
)

// Server is the HTTP control plane: the authoritative source of room metadata,
// membership, and per-epoch sealed keys. The data plane (messages) is NATS.
//
// Auth model (v1): mutating endpoints require an Ed25519 signature from the
// room owner over the canonical bytes of the request (the request body with the
// "sig" field cleared). v1 trusts the internal network: there is no TLS, no
// rate limiting, and read endpoints (GET) are unauthenticated. Hardening
// (mTLS, capabilities, rate limits) is a later phase.
type Server struct {
	store    Store
	blobs    blobstore.Store
	mux      *http.ServeMux
	authMode AuthMode
	nonces   nonceStore
	limiter  *ipRateLimiter
	inflight *inflightLimiter

	// RequireEncryptedRooms, when true, refuses to create cleartext (ModeNATS)
	// rooms. It is the minimum-defensive control for the data plane (audit H4):
	// the embedded NATS has no per-subject ACL, so a cleartext room is readable by
	// any registered peer that knows (or guesses) its subject. Forcing every room
	// to be end-to-end encrypted keeps message CONTENT confidential even when the
	// transport offers no subject isolation. The command sets this on a public
	// (non-loopback) bind. See dev/0004d-dataplane-acl.md for the full rationale
	// and the residual metadata exposure this does NOT close.
	RequireEncryptedRooms bool

	// Posture is the node's security posture, surfaced on /healthz so an operator
	// or a peer can detect a node NOT running the homogeneous enforce+ACL+TLS
	// posture a secure cluster requires (audit 0008 N1). It is set by the command;
	// the zero value (all false) reflects an unsecured dev node.
	Posture Posture

	// AllowedOrigins is the CORS allowlist of browser Origin headers permitted to
	// call the control plane cross-origin. It exists so a browser-native client
	// (uniweb) can talk to membershipd directly, the way the Go/Kotlin clients
	// already do over a non-browser transport (issue uniweb/0001). Native clients
	// send no Origin header and are unaffected. The zero value (empty) keeps CORS
	// OFF — no Access-Control headers are emitted and the server behaves exactly as
	// before — so this is opt-in per deployment. Entries are matched exactly (scheme
	// + host + port); never use "*" with credentials. Set by the command from a flag.
	AllowedOrigins []string

	// trustedProxies names the reverse proxies whose forwarding headers
	// (X-Forwarded-For / X-Real-IP) the rate limiter is allowed to believe. It
	// exists for the same-origin deployment where a single proxy (Caddy) fronts
	// the control plane: without it every proxied request would share the proxy's
	// one IP and collapse the per-IP rate limit into a single bucket for the whole
	// world. Only when the immediate peer is one of these addresses is the
	// forwarded client IP trusted; the zero value (nil) trusts nobody, preserving
	// the RemoteAddr-only behavior that predates the flag. Set by the command via
	// SetTrustedProxies. See clientIP.
	trustedProxies trustedProxyMatcher

	// js is the privileged JetStream context the server uses to own the durable
	// per-room streams of persisted rooms: it ensures a room's stream on creation
	// so the room's subject is captured from the first message — even from a
	// JetStream-less browser client (uniweb) that speaks only core NATS — and reads
	// it back for GET /rooms/{id}/history. It is wired by the command via
	// SetJetStream whenever a JetStream-capable data plane is available (always for
	// the embedded server). nil leaves history empty and stream-ensure a no-op,
	// preserving the pre-feature behavior for a deployment without JetStream.
	js jetstream.JetStream
	// streamReplicas is the replication factor for the room streams the server
	// creates, matched to the cluster's control-plane (KV) replication — 1 for a
	// standalone node, up to 3 in an HA cluster — so a persisted room's history is
	// as available as its metadata. Used only when js != nil. See SetJetStream.
	streamReplicas int
}

// Posture describes the security posture a membershipd node runs with. It is
// non-secret operational metadata (booleans + the store backend name), published
// on /healthz so a monitor can flag a cluster member that is not enforce+ACL+TLS
// — the weak node that would let an unauthenticated peer harvest the cluster's
// forwarded traffic (audit 0008 N1).
type Posture struct {
	Enforce bool   `json:"enforce"`
	ACL     bool   `json:"acl"`
	TLS     bool   `json:"tls"`
	Cluster bool   `json:"cluster"`
	Store   string `json:"store"` // "sqlite" | "kv"
}

// NewServer wires the membership store and blob store into an http.Handler. The
// authMode selects the control-plane auth rollout state (AuthOff for callers and
// tests that have not migrated to signed requests yet). It installs a per-IP
// rate limiter with the package defaults; loopback dev behavior is unchanged
// because the burst comfortably exceeds any single client's request rate.
func NewServer(store Store, blobs blobstore.Store, authMode AuthMode) *Server {
	s := &Server{
		store:    store,
		blobs:    blobs,
		mux:      http.NewServeMux(),
		authMode: authMode,
		nonces:   newMemNonceCache(nonceTTL, maxNonceCacheEntries),
		limiter:  newIPRateLimiter(defaultRatePerSec, defaultRateBurst, rateBucketTTL),
		inflight: newInflightLimiter(maxInflightBytes),
	}
	s.routes()
	return s
}

// SetJetStream wires the privileged JetStream context (and the room-stream
// replication factor) the server uses to ensure and read the durable streams of
// persisted rooms. replicas below 1 is clamped to 1. It must be called once at
// startup, before the server begins serving; leaving it unset keeps history empty
// and stream-ensure a no-op, the behavior for a deployment without JetStream.
func (s *Server) SetJetStream(js jetstream.JetStream, replicas int) {
	if replicas < 1 {
		replicas = 1
	}
	s.js = js
	s.streamReplicas = replicas
}

// UseReplicatedNonces switches the server's anti-replay store from the
// per-process in-memory cache to a JetStream KV bucket shared across the cluster
// (issue 0003e). It MUST be called on every node of a multi-node deployment:
// otherwise a request captured on one node can be replayed to another whose
// local cache never saw the nonce. replicas is the bucket's replication factor
// (R1..R3). The TTL matches the in-memory cache (nonceTTL = 2*clockSkew), so a
// replay can never outlive its memory.
func (s *Server) UseReplicatedNonces(js jetstream.JetStream, replicas int) error {
	ns, err := newKVNonceStore(js, nonceTTL, replicas, 0)
	if err != nil {
		return err
	}
	s.nonces = ns
	return nil
}

// ServeHTTP satisfies http.Handler. It runs the control-plane auth middleware
// (signature verification + anti-replay + allowlist) ahead of the router
// according to authMode, then dispatches to the matched handler.
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	now := time.Now()

	// CORS runs before everything else so a browser preflight never pays the rate
	// limit or auth cost. When the request carries an allowed Origin we echo the
	// Access-Control headers; a preflight (OPTIONS) is answered here and short-
	// circuits the pipeline. With an empty allowlist this is a no-op, so non-browser
	// clients and untouched deployments behave exactly as before (issue uniweb/0001).
	if s.applyCORS(w, r) {
		return // preflight handled
	}

	// Per-IP rate limit runs first, ahead of auth and body reads, so a flood is
	// shed at the cheapest possible point. The health probe is exempt so liveness
	// checks are never throttled.
	if !isAuthExempt(r) && !s.limiter.allow(s.clientIP(r), now) {
		writeErr(w, http.StatusTooManyRequests, "rate limit exceeded")
		return
	}

	// Cap how much body we will buffer, BEFORE reading a single byte. The ceiling
	// is per-route: /blobs may legitimately carry a media ciphertext, everything
	// else is small JSON. A declared Content-Length over the ceiling is rejected
	// outright (no buffering); MaxBytesReader then guards against a lying or
	// chunked sender by failing the read once the limit is crossed. This is the
	// fix for the pre-auth DoS: without it an unauthenticated peer could make the
	// server buffer an unbounded body in RAM before authenticate() ever ran.
	limit := int64(maxControlBodyBytes)
	if r.Method == http.MethodPost && r.URL.Path == "/blobs" {
		limit = int64(maxBlobBytes)
	}
	if r.ContentLength > limit {
		writeErr(w, http.StatusRequestEntityTooLarge, "request body too large")
		return
	}
	r.Body = http.MaxBytesReader(w, r.Body, limit)

	// Aggregate memory bound (audit N2): the per-request ceiling above and the
	// per-IP rate limit do not cap the TOTAL bytes buffered across concurrent
	// requests. A POST reserves its worst-case buffered size (its route ceiling)
	// from a global limiter before the body is read, and is shed with 503 when the
	// cap is reached, so the resident set stays bounded under a concurrent (even
	// multi-IP) upload flood instead of growing linearly with the number of
	// connections. Reservation is released when the request finishes. Only POSTs
	// buffer a body; GETs carry none, so they do not consume the budget.
	if r.Method == http.MethodPost {
		if !s.inflight.tryAcquire(limit) {
			writeErr(w, http.StatusServiceUnavailable, "server busy: too many concurrent uploads in flight")
			return
		}
		defer s.inflight.release(limit)
	}

	if s.authMode == AuthOff || isAuthExempt(r) {
		s.mux.ServeHTTP(w, r)
		return
	}

	// Buffer the (now bounded) body so the signature can be verified over it and
	// the handler still reads it.
	body, err := io.ReadAll(r.Body)
	if err != nil {
		if isBodyTooLarge(err) {
			writeErr(w, http.StatusRequestEntityTooLarge, "request body too large")
			return
		}
		writeErr(w, http.StatusBadRequest, "read body")
		return
	}
	_ = r.Body.Close()
	r.Body = io.NopCloser(bytes.NewReader(body))

	res, err := s.authenticate(r, body, now)
	if err != nil {
		if s.authMode == AuthSoft {
			log.Printf("[auth] soft: would reject %s %s: %v", r.Method, r.URL.Path, err)
			s.mux.ServeHTTP(w, r)
			return
		}
		writeErr(w, http.StatusUnauthorized, "unauthorized: "+err.Error())
		return
	}
	// Carry the authenticated signer's endpoint AND signing key into the handler.
	// Room handlers authorize by membership via the endpoint (audit H3); the
	// user-management handlers authorize by role via the signing key (the endpoint
	// id is a one-way hash of the key, so it cannot be reversed to look the signer
	// up in the user allowlist). Both are set only on a verified identity.
	s.mux.ServeHTTP(w, r.WithContext(withSigner(r.Context(), res.endpoint, res.pubHex)))
}

// applyCORS handles cross-origin requests for the control plane. When the request
// carries an Origin in the allowlist it sets the Access-Control-Allow-* response
// headers so the browser accepts the eventual response; when the request is a CORS
// preflight (OPTIONS) it writes the preflight reply and returns true so ServeHTTP
// short-circuits before the rate limiter and auth ever run. It returns false for
// every non-preflight request — including same-origin and native clients that send
// no Origin header — leaving the normal pipeline to run unchanged. With an empty
// AllowedOrigins it never sets a header (CORS is off): the opt-in default.
func (s *Server) applyCORS(w http.ResponseWriter, r *http.Request) (preflight bool) {
	origin := r.Header.Get("Origin")
	allowed := origin != "" && s.originAllowed(origin)
	if allowed {
		h := w.Header()
		h.Set("Access-Control-Allow-Origin", origin)
		// Vary: Origin so a cache never serves an allow-listed response to another
		// origin. Add (not Set) to preserve any Vary the handler may add later.
		h.Add("Vary", "Origin")
		h.Set("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
		// Allow the control-plane request-auth headers a browser client signs every
		// request with (busauth.signedHeaders), or the browser's CORS preflight blocks
		// the real request. Content-Type/Authorization stay for JSON bodies.
		h.Set("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Unibus-Pub, X-Unibus-Ts, X-Unibus-Nonce, X-Unibus-Sig")
		h.Set("Access-Control-Max-Age", "600")
	}
	if r.Method == http.MethodOptions {
		// Answer the preflight here so it never reaches the rate limiter or auth. An
		// allowed origin gets 204 with the headers above; a disallowed or missing
		// origin gets 403 with no Access-Control headers, so the browser blocks the
		// real cross-origin request.
		if allowed {
			w.WriteHeader(http.StatusNoContent)
		} else {
			w.WriteHeader(http.StatusForbidden)
		}
		return true
	}
	return false
}

// originAllowed reports whether origin is in the CORS allowlist. Matching is exact
// (scheme + host + port): a browser Origin is an opaque string, so an exact compare
// is both correct and the safest policy (no wildcard, no suffix tricks).
func (s *Server) originAllowed(origin string) bool {
	for _, o := range s.AllowedOrigins {
		if o == origin {
			return true
		}
	}
	return false
}

// isBodyTooLarge reports whether err is the sentinel returned by MaxBytesReader
// when the body exceeds its limit, so the middleware can map it to 413.
func isBodyTooLarge(err error) bool {
	var maxErr *http.MaxBytesError
	return errors.As(err, &maxErr)
}

// ctxKey is the unexported type for this package's request-context keys, so the
// values cannot collide with keys set by other packages.
type ctxKey int

const (
	ctxSignerEndpoint ctxKey = iota
	ctxSignerPub
)

// withSigner returns a context carrying the authenticated signer's endpoint id
// and signing public key (lowercase hex). The endpoint authorizes room
// membership; the signing key authorizes user-management by role, because the
// endpoint id is a one-way hash of the key (base64url(sha256(signPub))) and so
// cannot be reversed to look the signer up in the user allowlist.
func withSigner(ctx context.Context, endpoint, pubHex string) context.Context {
	ctx = context.WithValue(ctx, ctxSignerEndpoint, endpoint)
	return context.WithValue(ctx, ctxSignerPub, pubHex)
}

// signerEndpoint returns the authenticated signer's endpoint id and whether one
// is present. It is absent under AuthOff (no verification) and when a soft-mode
// request was let through unauthenticated — in both cases membership
// authorization is skipped, preserving dev/legacy behavior.
func signerEndpoint(r *http.Request) (string, bool) {
	v, ok := r.Context().Value(ctxSignerEndpoint).(string)
	return v, ok && v != ""
}

// signerPubHex returns the authenticated signer's signing public key (lowercase
// hex) and whether one is present. Like signerEndpoint it is absent under
// AuthOff and on a soft-mode pass-through; the user-management handlers treat
// that absence as "no admin identity" and deny (default-deny), since a
// privilege-granting operation must never run without a verified admin.
func signerPubHex(r *http.Request) (string, bool) {
	v, ok := r.Context().Value(ctxSignerPub).(string)
	return v, ok && v != ""
}

// requireMember authorizes a room request by membership (audit H3): it returns
// the signer endpoint and true when the request may proceed, or writes 403 and
// returns false when an authenticated signer is not a member of roomID. When no
// authenticated signer is present (AuthOff/dev, or soft pass-through) it allows
// the request — membership is only enforced once the caller's identity is known.
func (s *Server) requireMember(w http.ResponseWriter, r *http.Request, roomID string) (string, bool) {
	signer, ok := signerEndpoint(r)
	if !ok {
		return "", true
	}
	if _, err := s.store.GetMember(roomID, signer); err != nil {
		writeErr(w, http.StatusForbidden, "forbidden: not a member of this room")
		return signer, false
	}
	return signer, true
}

// requireAdmin authorizes a user-management request: it returns the signer's
// signing-key hex and true ONLY when the authenticated signer is a user with
// role admin and active status; otherwise it writes 403 and returns false.
//
// Default-deny, with no dev relaxation: unlike requireMember (which allows a
// request when no authenticated signer is present, preserving AuthOff/dev
// behavior for room reads), this denies whenever the signer is absent or is not
// a verified active admin. The user-management endpoints grant and revoke bus
// access, so they must never be reachable without a verified admin identity —
// the store is consulted on every call so a just-revoked admin is denied
// immediately, and any store error fails closed.
func (s *Server) requireAdmin(w http.ResponseWriter, r *http.Request) (string, bool) {
	pubHex, ok := signerPubHex(r)
	if !ok {
		writeErr(w, http.StatusForbidden, "forbidden: admin role required")
		return "", false
	}
	u, err := s.store.GetUser(pubHex)
	if err != nil || u.Role != RoleAdmin || u.Status != StatusActive {
		writeErr(w, http.StatusForbidden, "forbidden: admin role required")
		return "", false
	}
	return pubHex, true
}

// isAuthExempt lists requests that bypass control-plane auth even under enforce.
// Only the unauthenticated health probe qualifies: it carries no data and is
// needed by load balancers / smoke checks / systemd before any identity exists.
func isAuthExempt(r *http.Request) bool {
	return r.Method == http.MethodGet && r.URL.Path == "/healthz"
}

func (s *Server) routes() {
	s.mux.HandleFunc("GET /healthz", s.handleHealth)
	s.mux.HandleFunc("POST /rooms", s.handleCreateRoom)
	s.mux.HandleFunc("POST /rooms/{id}/invite", s.handleInvite)
	s.mux.HandleFunc("GET /rooms/{id}/key", s.handleGetKey)
	s.mux.HandleFunc("GET /rooms/{id}/members", s.handleListMembers)
	// Durable message history for a persisted room, read server-side from the room's
	// JetStream stream so a client without JetStream (the browser client uniweb) can
	// load the backlog over plain HTTP. Member-only, like /key and /members.
	// Registered without the /api prefix like every other control-plane route: Caddy
	// strips /api via handle_path /api/* before forwarding, so the SPA's
	// GET /api/rooms/{id}/history arrives here as GET /rooms/{id}/history.
	s.mux.HandleFunc("GET /rooms/{id}/history", s.handleRoomHistory)
	s.mux.HandleFunc("GET /members/{endpoint}/rooms", s.handleListMemberRooms)
	s.mux.HandleFunc("POST /rooms/{id}/rekey", s.handleRekey)
	s.mux.HandleFunc("GET /rooms/{id}", s.handleGetRoom)
	s.mux.HandleFunc("POST /blobs", s.handlePutBlob)
	s.mux.HandleFunc("GET /blobs/{hash}", s.handleGetBlob)
	// User-management (admin-only) — the HTTP-signed equivalent of the local
	// `membershipd user` CLI, so the admin panel manages the bus allowlist by
	// signing as an admin instead of needing direct store/KV access. All three
	// pass through requireAdmin; they hit the same store the room handlers do.
	s.mux.HandleFunc("GET /users", s.handleListUsers)
	s.mux.HandleFunc("POST /users", s.handleAddUser)
	s.mux.HandleFunc("POST /users/{signpub}/revoke", s.handleRevokeUser)
	// Member directory — any authenticated bus user (member or admin) may map an
	// endpoint id back to its human handle, so clients can render readable sender
	// names instead of raw endpoint hashes. Unlike /users it is NOT admin-only and
	// returns only active users; under enforce the auth middleware already rejects
	// an unauthenticated caller with 401 before this handler runs (uniweb/0002).
	// Registered without the /api prefix like every other control-plane route:
	// Caddy strips /api via handle_path /api/* before forwarding to membershipd,
	// so the SPA's GET /api/directory arrives here as GET /directory.
	s.mux.HandleFunc("GET /directory", s.handleDirectory)
}

// ---- wire types -----------------------------------------------------------

type policyJSON struct {
	Encrypt  bool `json:"encrypt"`
	Persist  bool `json:"persist"`
	SignMsgs bool `json:"sign_msgs"`
}

type endpointJSON struct {
	Endpoint string `json:"endpoint"`
	SignPub  []byte `json:"sign_pub"`
	KexPub   []byte `json:"kex_pub"`
}

type createRoomReq struct {
	Subject       string       `json:"subject"`
	Policy        policyJSON   `json:"policy"`
	Owner         endpointJSON `json:"owner"`
	SealedKeySelf []byte       `json:"sealed_key_self"`
}

type createRoomResp struct {
	RoomID string `json:"room_id"`
}

type inviteReq struct {
	By        string       `json:"by"`  // owner endpoint id
	Sig       []byte       `json:"sig"` // Ed25519 over canonical(request with sig cleared)
	Member    endpointJSON `json:"member"`
	SealedKey []byte       `json:"sealed_key"`
}

type keyResp struct {
	Epoch     int    `json:"epoch"`
	SealedKey []byte `json:"sealed_key"`
}

type memberJSON struct {
	Endpoint string `json:"endpoint"`
	Role     string `json:"role"`
	SignPub  []byte `json:"sign_pub"`
	KexPub   []byte `json:"kex_pub"`
}

type roomResp struct {
	Subject string     `json:"subject"`
	Epoch   int        `json:"epoch"`
	Policy  policyJSON `json:"policy"`
}

type memberRoomJSON struct {
	RoomID  string     `json:"room_id"`
	Subject string     `json:"subject"`
	Epoch   int        `json:"epoch"`
	Policy  policyJSON `json:"policy"`
	Role    string     `json:"role"`
}

type rekeyKey struct {
	Endpoint  string `json:"endpoint"`
	SealedKey []byte `json:"sealed_key"`
}

type rekeyReq struct {
	By       string     `json:"by"`
	Sig      []byte     `json:"sig"`
	NewEpoch int        `json:"new_epoch"`
	Keys     []rekeyKey `json:"keys"`
	Remove   []string   `json:"remove"`
}

type blobResp struct {
	Hash string `json:"hash"`
}

// userJSON is the wire representation of a bus user on the admin endpoints. It
// carries the full record the panel needs to render the allowlist, including
// status (so revoked users are visible) and the timestamps. revoked_at is
// omitted for an active user.
type userJSON struct {
	SignPub   string `json:"sign_pub"`
	Handle    string `json:"handle"`
	Role      string `json:"role"`
	Status    string `json:"status"`
	CreatedAt string `json:"created_at"`
	RevokedAt string `json:"revoked_at,omitempty"`
}

// addUserReq is the POST /users body: the new user's Ed25519 signing key
// (64-hex), human handle, and role. role is optional and defaults to member.
type addUserReq struct {
	SignPub string `json:"sign_pub"`
	Handle  string `json:"handle"`
	Role    string `json:"role"`
}

// directoryMember is one entry of the GET /directory response: enough for a
// client to map a message's endpoint id (which the bus stamps on every frame)
// back to a readable handle. endpoint is derived server-side from sign_pub with
// the SAME construction the bus uses (frame.EndpointID = base64url(sha256(signPub)),
// unpadded), so it matches the sender id a client already has byte-for-byte.
type directoryMember struct {
	SignPub  string `json:"sign_pub"`
	Endpoint string `json:"endpoint"`
	Handle   string `json:"handle"`
	Role     string `json:"role"`
}

// directoryResp is the GET /directory response envelope. The members key is a
// stable contract consumed by the browser client; do not rename it.
type directoryResp struct {
	Members []directoryMember `json:"members"`
}

// ---- helpers --------------------------------------------------------------

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 writeErr(w http.ResponseWriter, code int, msg string) {
	writeJSON(w, code, map[string]string{"error": msg})
}

// writeServerErr logs the internal error detail and returns ONLY a generic
// message to the client (audit H12): raw store/blob errors embed SQL fragments
// and filesystem paths, which must not leak to a caller. Use it for any error
// that originates inside the server (5xx, or a not-found wrapping a store error).
func writeServerErr(w http.ResponseWriter, r *http.Request, code int, publicMsg string, err error) {
	log.Printf("[handler] %s %s -> %d: %v", r.Method, r.URL.Path, code, err)
	writeErr(w, code, publicMsg)
}

// canonicalSig returns the bytes to verify for a request: the request struct
// re-marshaled with its Sig field cleared. The caller passes a copy with Sig
// already zeroed. This is symmetric with how the client signs.
func canonicalSig(v any) []byte {
	b, _ := json.Marshal(v)
	return b
}

// verifyOwnerSig checks that sig is a valid Ed25519 signature by the room owner
// over canonical(reqWithSigCleared). It returns the owner Member on success.
func (s *Server) verifyOwnerSig(roomID, by string, sig, canonical []byte) (Member, error) {
	info, err := s.store.GetRoom(roomID)
	if err != nil {
		return Member{}, fmt.Errorf("room not found")
	}
	if by != info.OwnerEndpoint {
		return Member{}, fmt.Errorf("requester %q is not the room owner", by)
	}
	owner, err := s.store.GetMember(roomID, by)
	if err != nil {
		return Member{}, fmt.Errorf("owner member not found")
	}
	if !cs.VerifyEd25519(owner.SignPub, canonical, sig) {
		return Member{}, fmt.Errorf("invalid owner signature")
	}
	return owner, nil
}

// ---- handlers -------------------------------------------------------------

func (s *Server) handleHealth(w http.ResponseWriter, _ *http.Request) {
	writeJSON(w, http.StatusOK, map[string]any{"status": "ok", "posture": s.Posture})
}

func (s *Server) handleCreateRoom(w http.ResponseWriter, r *http.Request) {
	var req createRoomReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeErr(w, http.StatusBadRequest, "bad json: "+err.Error())
		return
	}
	if req.Subject == "" || req.Owner.Endpoint == "" {
		writeErr(w, http.StatusBadRequest, "subject and owner.endpoint required")
		return
	}
	// Data-plane minimum defense (audit H4): on a public deployment cleartext
	// rooms are disabled, so no message ever rides the un-ACL'd NATS subject in
	// the clear for another registered peer to sniff.
	if s.RequireEncryptedRooms && !req.Policy.Encrypt {
		writeErr(w, http.StatusForbidden,
			"cleartext rooms are disabled on this deployment; create an encrypted (Matrix-policy) room")
		return
	}
	// Owner binding (audit H6): the declared owner must BE the authenticated
	// signer — both the endpoint id and the signing key. Otherwise a registered
	// peer could create rooms in another identity's name. Enforced only when an
	// authenticated signer is present (AuthOff/dev trusts the caller).
	if signer, ok := signerEndpoint(r); ok {
		if req.Owner.Endpoint != signer || frame.EndpointID(req.Owner.SignPub) != signer {
			writeErr(w, http.StatusForbidden, "forbidden: room owner must be the authenticated signer")
			return
		}
	}
	roomID := newULID()
	info := RoomInfo{
		RoomID:        roomID,
		Subject:       req.Subject,
		Encrypt:       req.Policy.Encrypt,
		Persist:       req.Policy.Persist,
		SignMsgs:      req.Policy.SignMsgs,
		OwnerEndpoint: req.Owner.Endpoint,
	}
	// Own the durable stream for a persisted room (issue room-history): ensure it
	// BEFORE the room row is written so the subject is captured from the very first
	// message whoever publishes it — a Go client OR a JetStream-less browser client.
	// Done first so a stream failure aborts cleanly with no orphan room row (the
	// rare orphan empty stream it can leave is harmless and idempotently reused).
	// Skipped when no JetStream is wired: the room still works, just without history.
	if info.Persist && s.js != nil {
		ctx, cancel := context.WithTimeout(r.Context(), historyOpTimeout)
		err := ensureRoomStream(ctx, s.js, roomID, info.Subject, s.streamReplicas)
		cancel()
		if err != nil {
			writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
			return
		}
	}
	if err := s.store.CreateRoom(info, req.Owner.SignPub, req.Owner.KexPub, req.SealedKeySelf); err != nil {
		writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
		return
	}
	writeJSON(w, http.StatusCreated, createRoomResp{RoomID: roomID})
}

func (s *Server) handleInvite(w http.ResponseWriter, r *http.Request) {
	roomID := r.PathValue("id")
	var req inviteReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeErr(w, http.StatusBadRequest, "bad json: "+err.Error())
		return
	}
	// Canonical bytes = the request with Sig cleared.
	sig := req.Sig
	req.Sig = nil
	if _, err := s.verifyOwnerSig(roomID, req.By, sig, canonicalSig(req)); err != nil {
		writeErr(w, http.StatusForbidden, err.Error())
		return
	}
	info, err := s.store.GetRoom(roomID)
	if err != nil {
		writeServerErr(w, r, http.StatusNotFound, "room not found", err)
		return
	}
	m := Member{
		Endpoint: req.Member.Endpoint,
		Role:     "member",
		SignPub:  req.Member.SignPub,
		KexPub:   req.Member.KexPub,
	}
	if err := s.store.AddMember(roomID, m, info.Epoch, req.SealedKey); err != nil {
		writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
		return
	}
	writeJSON(w, http.StatusOK, map[string]string{"status": "invited"})
}

func (s *Server) handleGetKey(w http.ResponseWriter, r *http.Request) {
	roomID := r.PathValue("id")
	endpoint := r.URL.Query().Get("endpoint")
	if endpoint == "" {
		writeErr(w, http.StatusBadRequest, "endpoint query param required")
		return
	}
	// A sealed room key is sealed to one identity's X25519 key. Serving it only to
	// that identity (the signer) stops a registered peer from harvesting another
	// member's sealed key (audit H3). Membership is implied by owning a sealed key,
	// but we also require the signer to be a member for defense in depth.
	if signer, ok := signerEndpoint(r); ok {
		if endpoint != signer {
			writeErr(w, http.StatusForbidden, "forbidden: may only fetch your own sealed key")
			return
		}
		if _, err := s.store.GetMember(roomID, signer); err != nil {
			writeErr(w, http.StatusForbidden, "forbidden: not a member of this room")
			return
		}
	}
	epoch := 0
	if e := r.URL.Query().Get("epoch"); e != "" {
		if n, err := strconv.Atoi(e); err == nil {
			epoch = n
		}
	}
	ep, sealed, err := s.store.GetSealedKey(roomID, endpoint, epoch)
	if err != nil {
		if errors.Is(err, ErrNotFound) {
			writeErr(w, http.StatusForbidden,
				"not invited to this encrypted room: no key has been sealed for your identity. Ask the room owner to invite you before joining.")
			return
		}
		writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
		return
	}
	writeJSON(w, http.StatusOK, keyResp{Epoch: ep, SealedKey: sealed})
}

func (s *Server) handleListMembers(w http.ResponseWriter, r *http.Request) {
	roomID := r.PathValue("id")
	// Membership authorization (audit H3): the member list exposes every member's
	// sign_pub + kex_pub, so it must not be served to a non-member.
	if _, ok := s.requireMember(w, r, roomID); !ok {
		return
	}
	members, err := s.store.ListMembers(roomID)
	if err != nil {
		writeErr(w, http.StatusInternalServerError, "internal error")
		return
	}
	out := make([]memberJSON, 0, len(members))
	for _, m := range members {
		out = append(out, memberJSON{Endpoint: m.Endpoint, Role: m.Role, SignPub: m.SignPub, KexPub: m.KexPub})
	}
	writeJSON(w, http.StatusOK, out)
}

func (s *Server) handleListMemberRooms(w http.ResponseWriter, r *http.Request) {
	endpoint := r.PathValue("endpoint")
	if endpoint == "" {
		writeErr(w, http.StatusBadRequest, "endpoint required")
		return
	}
	// A peer may only enumerate its OWN room directory (audit H3): otherwise any
	// registered identity could map another's entire social graph of rooms.
	if signer, ok := signerEndpoint(r); ok && endpoint != signer {
		writeErr(w, http.StatusForbidden, "forbidden: may only list your own rooms")
		return
	}
	rooms, err := s.store.ListRoomsForEndpoint(endpoint)
	if err != nil {
		writeErr(w, http.StatusInternalServerError, "internal error")
		return
	}
	out := make([]memberRoomJSON, 0, len(rooms))
	for _, rm := range rooms {
		out = append(out, memberRoomJSON{
			RoomID:  rm.RoomID,
			Subject: rm.Subject,
			Epoch:   rm.Epoch,
			Policy:  policyJSON{Encrypt: rm.Encrypt, Persist: rm.Persist, SignMsgs: rm.SignMsgs},
			Role:    rm.Role,
		})
	}
	writeJSON(w, http.StatusOK, out)
}

func (s *Server) handleGetRoom(w http.ResponseWriter, r *http.Request) {
	roomID := r.PathValue("id")
	if _, ok := s.requireMember(w, r, roomID); !ok {
		return
	}
	info, err := s.store.GetRoom(roomID)
	if err != nil {
		writeErr(w, http.StatusNotFound, "room not found")
		return
	}
	writeJSON(w, http.StatusOK, roomResp{
		Subject: info.Subject,
		Epoch:   info.Epoch,
		Policy:  policyJSON{Encrypt: info.Encrypt, Persist: info.Persist, SignMsgs: info.SignMsgs},
	})
}

func (s *Server) handleRekey(w http.ResponseWriter, r *http.Request) {
	roomID := r.PathValue("id")
	var req rekeyReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeErr(w, http.StatusBadRequest, "bad json: "+err.Error())
		return
	}
	sig := req.Sig
	req.Sig = nil
	if _, err := s.verifyOwnerSig(roomID, req.By, sig, canonicalSig(req)); err != nil {
		writeErr(w, http.StatusForbidden, err.Error())
		return
	}
	if req.NewEpoch <= 0 {
		writeErr(w, http.StatusBadRequest, "new_epoch must be > 0")
		return
	}
	// Bump epoch, then store the fresh sealed keys for the remaining members,
	// then remove the kicked/left members.
	if err := s.store.BumpEpoch(roomID, req.NewEpoch); err != nil {
		writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
		return
	}
	keys := make(map[string][]byte, len(req.Keys))
	for _, k := range req.Keys {
		keys[k.Endpoint] = k.SealedKey
	}
	if len(keys) > 0 {
		if err := s.store.PutSealedKeys(roomID, req.NewEpoch, keys); err != nil {
			writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
			return
		}
	}
	for _, ep := range req.Remove {
		if err := s.store.RemoveMember(roomID, ep); err != nil {
			writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
			return
		}
	}
	writeJSON(w, http.StatusOK, map[string]any{"status": "rekeyed", "epoch": req.NewEpoch})
}

func (s *Server) handlePutBlob(w http.ResponseWriter, r *http.Request) {
	// The body arrives already bounded: ServeHTTP wraps it in a MaxBytesReader
	// (maxBlobBytes) and rejects an over-declared Content-Length before this
	// handler runs, in every auth mode. Reading here therefore cannot buffer
	// more than the ceiling; a sender that lies about its length (e.g. chunked)
	// trips MaxBytesReader and we map that to 413 rather than a generic 400.
	data, err := io.ReadAll(r.Body)
	if err != nil {
		if isBodyTooLarge(err) {
			writeErr(w, http.StatusRequestEntityTooLarge, "request body too large")
			return
		}
		writeErr(w, http.StatusBadRequest, "read body")
		return
	}
	hash, err := s.blobs.Put(data)
	if err != nil {
		writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
		return
	}
	writeJSON(w, http.StatusOK, blobResp{Hash: hash})
}

func (s *Server) handleGetBlob(w http.ResponseWriter, r *http.Request) {
	hash := r.PathValue("hash")
	if strings.ContainsAny(hash, "/\\.") {
		writeErr(w, http.StatusBadRequest, "invalid hash")
		return
	}
	data, err := s.blobs.Get(hash)
	if err != nil {
		writeServerErr(w, r, http.StatusNotFound, "not found", err)
		return
	}
	w.Header().Set("Content-Type", "application/octet-stream")
	w.WriteHeader(http.StatusOK)
	_, _ = w.Write(data)
}

// ---- user-management handlers (admin-only) --------------------------------

// handleListUsers returns the full bus allowlist, including revoked users, so an
// admin sees the complete picture (a revoked identity stays auditable). Admin-only.
func (s *Server) handleListUsers(w http.ResponseWriter, r *http.Request) {
	if _, ok := s.requireAdmin(w, r); !ok {
		return
	}
	users, err := s.store.ListUsers()
	if err != nil {
		writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
		return
	}
	out := make([]userJSON, 0, len(users))
	for _, u := range users {
		out = append(out, userJSON{
			SignPub:   u.SignPub,
			Handle:    u.Handle,
			Role:      u.Role,
			Status:    u.Status,
			CreatedAt: u.CreatedAt,
			RevokedAt: u.RevokedAt,
		})
	}
	writeJSON(w, http.StatusOK, out)
}

// handleDirectory returns the active bus user directory so a client can resolve a
// sender's endpoint id to a readable handle. Unlike handleListUsers it is NOT
// admin-only: every authenticated bus user may read it (the auth middleware has
// already verified the caller is an active user under enforce, and rejected an
// unauthenticated one with 401). Only active users are listed, and each endpoint
// is computed server-side from the user's sign_pub with frame.EndpointID — the
// exact derivation the bus stamps on every frame, so the returned endpoint matches
// the sender id a client already holds. A user with a malformed sign_pub (which
// the add path rejects, so this is defensive) is skipped rather than failing the
// whole listing.
func (s *Server) handleDirectory(w http.ResponseWriter, r *http.Request) {
	users, err := s.store.ListUsers()
	if err != nil {
		writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
		return
	}
	out := make([]directoryMember, 0, len(users))
	for _, u := range users {
		if u.Status != StatusActive {
			continue
		}
		signPub, err := hex.DecodeString(u.SignPub)
		if err != nil || len(signPub) != 32 {
			continue
		}
		out = append(out, directoryMember{
			SignPub:  u.SignPub,
			Endpoint: frame.EndpointID(signPub),
			Handle:   u.Handle,
			Role:     u.Role,
		})
	}
	writeJSON(w, http.StatusOK, directoryResp{Members: out})
}

// handleAddUser registers a new bus user from an admin-supplied Ed25519 signing
// key. It mirrors the `membershipd user add` CLI: the key must be 64-hex, the
// role must be admin or member (empty defaults to member), and re-adding an
// already-registered key is a 409 that leaves the existing row untouched — no
// silent upsert that could flip a role or clobber status. Admin-only.
func (s *Server) handleAddUser(w http.ResponseWriter, r *http.Request) {
	if _, ok := s.requireAdmin(w, r); !ok {
		return
	}
	var req addUserReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeErr(w, http.StatusBadRequest, "bad json: "+err.Error())
		return
	}
	if req.SignPub == "" || req.Handle == "" {
		writeErr(w, http.StatusBadRequest, "sign_pub and handle required")
		return
	}
	if err := ValidateSignPubHex(req.SignPub); err != nil {
		writeErr(w, http.StatusBadRequest, err.Error())
		return
	}
	role := req.Role
	if role == "" {
		role = RoleMember
	}
	if role != RoleAdmin && role != RoleMember {
		writeErr(w, http.StatusBadRequest,
			fmt.Sprintf("invalid role %q (want %q or %q)", role, RoleAdmin, RoleMember))
		return
	}
	if err := s.store.AddUser(req.SignPub, req.Handle, role); err != nil {
		if errors.Is(err, ErrUserExists) {
			// Idempotency contract (mirrors the CLI): re-adding a key is an explicit,
			// non-destructive conflict. To replace a user, revoke then add again.
			writeErr(w, http.StatusConflict,
				"user already registered (unchanged); revoke it first to replace")
			return
		}
		writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
		return
	}
	writeJSON(w, http.StatusCreated, map[string]string{"status": "added"})
}

// handleRevokeUser revokes a bus user by signing key. Revocation is a status
// flip (no hard delete) so the identity stays auditable and IsAuthorized denies
// it on both planes immediately. Revoking an unknown or already-revoked key is a
// 404. Admin-only.
func (s *Server) handleRevokeUser(w http.ResponseWriter, r *http.Request) {
	if _, ok := s.requireAdmin(w, r); !ok {
		return
	}
	signPub := r.PathValue("signpub")
	if err := ValidateSignPubHex(signPub); err != nil {
		writeErr(w, http.StatusBadRequest, err.Error())
		return
	}
	if err := s.store.RevokeUser(signPub); err != nil {
		writeServerErr(w, r, http.StatusNotFound, "no active user with that key", err)
		return
	}
	writeJSON(w, http.StatusOK, map[string]string{"status": "revoked"})
}