textmachine/platform/internal/httpapi/stream.go

package httpapi

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"time"

	"textmachine/platform/internal/pgstore"
)

// stream.go: the book's live events (canon §streamBookEvents).
//
// On the BOOK and not on a run: a book is received and cut into chapters before any run exists, and
// those minutes are what a user watches. The stream is what made the frontend poll every three
// seconds for the end of a parse that takes minutes (Ф-56), and the cure was not a new frame but
// this channel being on the right object.
//
// What travels is a delta a client can APPLY (`note`) or a counter plus the SCOPE of what changed
// (everything else). Never a bare "something changed", and never translated text.

const (
	// heartbeat keeps the connection and every proxy on it alive. An SSE COMMENT line, invisible to
	// a browser EventSource, which is what the contract specifies.
	heartbeat = 20 * time.Second
	// pollEvery is how often a live stream looks for new frames. The frames are minted by the
	// writers, in other transactions and other processes, so this side reads rather than subscribes:
	// two indexed queries per second per open CONNECTION — the frames and the book's own state — so
	// twelve tabs on one book are twenty-four, not two (register row PD-247). A push channel
	// (LISTEN/NOTIFY) is the optimisation, not the correctness: a frame is a poke, and a second of
	// latency on a poke is not observable next to a translation.
	pollEvery = time.Second
	// framesPerRead bounds one catch-up read. HALF the stored buffer (pgstore.bufferFrames): a full
	// batch is re-read at once rather than a tick later, so falling behind takes a writer sustaining
	// more than this many frames a second — and the miss is told, not silent.
	framesPerRead = 256
	// writeTimeout bounds ONE frame's write. It is not a limit on how long a stream may live — a
	// stream lives as long as the book is doing something — but on how long a peer that has stopped
	// reading may hold the goroutine behind it.
	writeTimeout = 10 * time.Second
)

// streamEvents serves `GET /books/{bookId}/events`.
func (h *v0) streamEvents(w http.ResponseWriter, r *http.Request) {
	user, ok := principal(w, r)
	if !ok {
		return
	}
	bookID := r.PathValue("bookId")
	state, err := h.lib.ReadStream(r.Context(), user, bookID)
	if err != nil {
		h.fail(w, r, err)
		return
	}
	last, resuming := lastEventID(r)
	if resuming && last >= state.Position && state.AtRest {
		// The presented id is at or past the last frame and the book is at rest. 204 is how SSE is
		// told to stop reconnecting (WHATWG), and it is what makes the sequence finite for a book
		// nothing is happening to: hello → end → close → one reconnect → 204.
		w.WriteHeader(http.StatusNoContent)
		return
	}
	if r.Method == http.MethodHead {
		// A `GET` pattern matches HEAD since Go 1.22, and a HEAD has no body to stream: same headers,
		// no pump (RFC 9110 §9.3.2). Without this it held a goroutine per call.
		w.Header().Set("Content-Type", "text/event-stream")
		w.WriteHeader(http.StatusOK)
		return
	}
	// ⚠ Through the ResponseController and NOT a type assertion on the writer. The handler sits
	// under the access log's wrapper, which implements Unwrap but not Flush, so an assertion answers
	// "this cannot be streamed" for every real request — the controller is what follows the chain
	// (net/http, since 1.20). It is exercised AFTER the head is written, because a flush is what
	// commits the response: called first, it would send a bare 200 with none of the headers below.
	rc := http.NewResponseController(w)
	head := w.Header()
	head.Set("Content-Type", "text/event-stream")
	// ⚠ NOT compressed, ever: compressing a stream buffers it. This route does not go through
	// writeJSON, which is where compression lives, so the rule holds structurally — and the header
	// below is the same instruction to the proxies that honour it, which is the mechanism behind
	// "the server MUST NOT buffer the stream" (deploy note, §7 of the companion).
	head.Set("X-Accel-Buffering", "no")
	w.WriteHeader(http.StatusOK)
	if err := rc.Flush(); err != nil {
		// A buffered stream is the one thing the contract forbids of this route, so a writer that
		// cannot flush is abandoned rather than served slowly. The head is already out; there is
		// nothing to say in a body.
		h.log.ErrorContext(r.Context(), "the response writer cannot flush: the event stream is not servable here", "err", err)
		return
	}

	s := &stream{w: w, rc: rc}
	// The first frame is ALWAYS hello, and it carries the id of the last history frame — `0` on a
	// book that has produced none. A hello without an id would leave the client with nothing to
	// present on the reconnect a browser makes by itself, and the stream would reopen forever.
	if !s.send("hello", state.Position, map[string]any{
		"contract": ContractVersion, "revision": state.Revision,
		"structure_version": state.StructureVersion,
	}) {
		return
	}
	// "If the server cannot resume from the id it sends `resync_required` rather than silently
	// starting from now" (canon §streamBookEvents). Two ways it cannot:
	//
	//   - the next frame the client needs is already pruned. `last+1 < Oldest` and not
	//     `last < Oldest`: `0` is a legitimate id, carried by every connection frame of a book with
	//     no history.
	//   - the id is ABOVE anything this server issued, which a database restored from a backup
	//     produces — `event_position` moves back and every watcher then holds a number from the
	//     future. Starting such a client afresh loses the frames between, silently.
	if resuming && ((state.Oldest > 0 && last+1 < state.Oldest) || last > state.Position) {
		s.send("resync_required", state.Position, base(state))
		return
	}
	h.pump(r.Context(), s, user, bookID, state, last, resuming)
}

// pump is the stream's life after the handshake.
func (h *v0) pump(ctx context.Context, s *stream, user, bookID string, state pgstore.StreamState, last int64, resuming bool) {
	from := state.Position
	if resuming {
		from = last
	}
	structure := state.StructureVersion
	beat := time.NewTicker(heartbeat)
	defer beat.Stop()
	poll := time.NewTicker(pollEvery)
	defer poll.Stop()
	for {
		frames, err := h.lib.ReadFrames(ctx, bookID, from, framesPerRead)
		if err != nil {
			// A client that closed the tab cancels the context, and that is not a fault to report:
			// logging it at ERROR made the ordinary end of every stream look like a defect.
			if ctx.Err() == nil {
				h.log.ErrorContext(ctx, "the event stream could not be read", "err", err)
			}
			return
		}
		// The buffer is pruned while the connection is open, and a `note` is delivered once, so a gap
		// is told rather than skipped. A prune BEFORE this read shows as a hole at the HEAD of the
		// batch (positions are contiguous — one writer, pgstore.emitFrame); a prune after it shows as
		// an Oldest past the watermark below. Checking only the second missed the first: the watermark
		// jumps over the hole and then sits ABOVE Oldest.
		gap := len(frames) > 0 && frames[0].Position > from+1
		if !gap {
			for _, f := range coalesce(frames) {
				if !s.raw(f.Event, f.Position, f.Data) {
					return
				}
			}
			if len(frames) > 0 {
				// The watermark moves past every frame READ, coalesced ones included: a frame replaced
				// by a later one of its kind has been superseded, not lost.
				from = frames[len(frames)-1].Position
			}
		}
		state, err = h.lib.ReadStream(ctx, user, bookID)
		if err != nil {
			// The book was deleted, or the database went away. Either way this stream is over; a
			// reconnect is answered by the ordinary handler, which says which of the two it was.
			if ctx.Err() == nil {
				h.log.ErrorContext(ctx, "the stream's book state could not be read", "err", err)
			}
			return
		}
		if gap || state.Oldest > from+1 {
			s.send("resync_required", state.Position, base(state))
			return
		}
		if state.StructureVersion != structure {
			// The book was cut again: every pair anchor and every cursor the client holds is invalid,
			// and a delta cannot express what happened. This is the one thing a client cannot be told
			// by a counter.
			s.send("resync_required", state.Position, base(state))
			return
		}
		if state.AtRest && from >= state.Position {
			// Nothing is live and nothing is buffered beyond what this client has seen.
			s.send("end", state.Position, base(state))
			return
		}
		if len(frames) == framesPerRead {
			continue // a full batch means there is more behind it; catching up must not cost a tick each
		}
		select {
		case <-ctx.Done():
			return
		case <-beat.C:
			if !s.comment() {
				return
			}
		case <-poll.C:
		}
	}
}

func base(state pgstore.StreamState) map[string]any {
	return map[string]any{"revision": state.Revision, "structure_version": state.StructureVersion}
}

// coalesce applies the contract's rule: a STATE frame may be replaced by a later one of its kind,
// and `note` may not be — a lost note is lost silently and forever.
//
// ⚠ "Of its kind" is read as the SCOPE the frame describes, not merely the event name: a `chapter`
// frame is a counter plus the id it belongs to, so dropping chapter 5's frame because chapter 7's
// arrived in the same batch loses 5's counter with nothing to repair it inside the connection.
// Keying on (event, id) keeps one frame per chapter and still collapses the storm the rule exists
// for — a run resolving units in one chapter emits one frame per unit.
//
// The gap this leaves is legal and declared: a client MUST NOT read a skipped number as a lost frame.
func coalesce(frames []pgstore.Frame) []pgstore.Frame {
	newest := map[string]int64{}
	for _, f := range frames {
		if f.Event != pgstore.FrameNote {
			newest[coalesceKey(f)] = f.Position
		}
	}
	out := make([]pgstore.Frame, 0, len(frames))
	for _, f := range frames {
		if f.Event == pgstore.FrameNote || newest[coalesceKey(f)] == f.Position {
			out = append(out, f)
		}
	}
	return out
}

// coalesceKey is the frame's kind plus, where the frame is about one entity, that entity.
func coalesceKey(f pgstore.Frame) string {
	if f.Event != pgstore.FrameChapter {
		return f.Event
	}
	var scope struct {
		ID string `json:"id"`
	}
	if err := json.Unmarshal(f.Data, &scope); err != nil {
		return f.Event
	}
	return f.Event + "\x00" + scope.ID
}

// lastEventID reads the reconnect header. A value that is not a decimal integer is treated as no
// value at all: the id is the server's own and a client that mangled it gets a new stream rather
// than an error it cannot act on.
func lastEventID(r *http.Request) (int64, bool) {
	raw := strings.TrimSpace(r.Header.Get("Last-Event-ID"))
	if raw == "" {
		return 0, false
	}
	n, err := strconv.ParseInt(raw, 10, 64)
	if err != nil || n < 0 {
		return 0, false
	}
	return n, true
}

// stream writes SSE frames.
//
// Written out rather than taken from a library: the framing is four lines of the WHATWG spec, and
// what actually matters here — that `id` is the book's own position, that a heartbeat is a comment
// and that nothing is buffered — is exactly what a generic helper would hide.
type stream struct {
	w    http.ResponseWriter
	rc   *http.ResponseController
	dead bool
}

func (s *stream) send(event string, id int64, payload map[string]any) bool {
	data, err := json.Marshal(payload)
	if err != nil {
		return false
	}
	return s.raw(event, id, data)
}

func (s *stream) raw(event string, id int64, data []byte) bool {
	return s.write(func() error {
		_, err := fmt.Fprintf(s.w, "event: %s\nid: %d\ndata: %s\n\n", event, id, data)
		return err
	})
}

func (s *stream) comment() bool {
	return s.write(func() error {
		_, err := s.w.Write([]byte(":\n\n"))
		return err
	})
}

// write emits one frame under a deadline: a peer that stops reading fills the socket and blocks the
// write forever, and going quiet does not cancel `r.Context()`. Writers that take no deadline (a
// test recorder) are served without one.
//
// ⚠ The FLUSH decides, not the Fprintf: a small frame lands in the bufio and the write returns nil
// while the socket is stuck.
func (s *stream) write(emit func() error) bool {
	if s.dead {
		return false
	}
	_ = s.rc.SetWriteDeadline(time.Now().Add(writeTimeout))
	if err := emit(); err != nil {
		s.dead = true
		return false
	}
	if err := s.rc.Flush(); err != nil {
		s.dead = true
		return false
	}
	return true
}