meta-core/internal/api/sse_events.go

package api

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

	"github.com/metazla/meta-core/internal/kv"
	"github.com/metazla/meta-core/internal/storage"
)

// SSE event stream endpoints for the file:events and meta:events streams.
//
// Wire contract (see docs/api-mediated-access.md):
//   - One SSE event per stream entry.
//   - `id:` is the opaque stream entry ID (`<ms>-<seq>`, the Redis Streams
//     form, kept across the move to SQLite so cursors consumers persisted to
//     disk still compare correctly); consumers MUST treat it as a black-box
//     string they echo back via Last-Event-ID.
//   - `event:` is the `type` field from the underlying entry.
//   - `data:` is the JSON-encoded rest of the entry (all fields except `type`).
//   - Heartbeats are SSE comment lines (`:keep-alive\n\n`) every 30s of silence.
//   - On reconnect with Last-Event-ID, resume from the next entry. If the
//     cursor has been trimmed by retention, emit one `event: gap` payload
//     before resuming from the oldest available entry.

const (
	sseHeartbeatInterval = 30 * time.Second

	// sseReadBlock controls how long each read waits for new entries.
	// Much shorter than sseHeartbeatInterval so the handler stays responsive
	// to client disconnects and the heartbeat check runs frequently enough
	// that the keep-alive fires within a few seconds of its nominal cadence.
	sseReadBlock = 2 * time.Second

	// sseReadCount caps how many entries one read returns. Small enough
	// to keep per-flush memory low, large enough to drain bursts cheaply.
	sseReadCount = 100
)

// handleEventsFiles streams the file:events stream as SSE.
func (s *Server) handleEventsFiles(w http.ResponseWriter, r *http.Request) {
	s.streamSSE(w, r, "file:events")
}

// handleEventsMeta streams the meta:events stream as SSE.
func (s *Server) handleEventsMeta(w http.ResponseWriter, r *http.Request) {
	s.streamSSE(w, r, "meta:events")
}

// streamSSE is the SSE pump shared by both endpoints. Each subscriber reads
// the stream itself and sleeps on the store's commit broadcast between reads —
// fine at our scale (≤20 subscribers; one per consumer service).
func (s *Server) streamSSE(w http.ResponseWriter, r *http.Request, stream string) {
	if !s.storage.IsConnected() {
		w.Header().Set("Retry-After", "5")
		writeError(w, http.StatusServiceUnavailable, "storage not connected")
		return
	}

	flusher, ok := w.(http.Flusher)
	if !ok {
		writeError(w, http.StatusInternalServerError, "streaming unsupported")
		return
	}

	// The server-wide WriteTimeout (30s on Server.server in server.go) is
	// fine for ordinary handlers but lethal for SSE — it would close every
	// connection at exactly 30s, racing the heartbeat. Disable it
	// per-request via http.NewResponseController. SSE endpoints never need
	// a write deadline; the client disconnect path already cleans up.
	rc := http.NewResponseController(w)
	if err := rc.SetWriteDeadline(time.Time{}); err != nil {
		// Older transports (or nested response writers) may not support
		// this. Log and continue — the 30s cap will still apply, but the
		// handler is still useful for short-lived subscriptions.
		log.Printf("[SSE %s] could not disable write deadline: %v", stream, err)
	}

	// SSE response headers. X-Accel-Buffering disables nginx response
	// buffering; the hash-lock sidecar uses nginx, so this matters.
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("X-Accel-Buffering", "no")
	w.WriteHeader(http.StatusOK)
	flusher.Flush()

	store := s.storage.Store()
	if store == nil {
		// Storage was just disconnected — close cleanly.
		return
	}

	ctx := r.Context()
	lastID := r.Header.Get("Last-Event-ID")
	var cursor string
	if stream == storage.MetaEventsStream && (lastID == "0" || lastID == "0-0") {
		// "From the beginning" on meta:events means "the whole current
		// state": stream it from the records themselves, then the live tail.
		head, err := streamSnapshot(ctx, store, w, flusher)
		if err != nil {
			if ctx.Err() == nil {
				log.Printf("[SSE %s] snapshot failed: %v", stream, err)
			}
			return
		}
		cursor = head
	} else {
		cursor = resolveStartCursor(ctx, store, stream, lastID, w, flusher)
	}

	lastWrite := time.Now()
	for {
		if err := ctx.Err(); err != nil {
			return
		}

		// Take the wake-up channel BEFORE reading, so an append landing
		// between the read and the wait still wakes us.
		wake := store.Wait(stream)
		entries, err := store.ReadAfter(ctx, stream, cursor, sseReadCount)
		if err != nil {
			if ctx.Err() != nil {
				return
			}
			log.Printf("[SSE %s] read error: %v", stream, err)
			return
		}

		for _, e := range entries {
			writeSSEEntry(w, flusher, e)
			cursor = e.ID
			lastWrite = time.Now()
		}

		if len(entries) == 0 {
			timer := time.NewTimer(sseReadBlock)
			select {
			case <-ctx.Done():
				timer.Stop()
				return
			case <-wake:
			case <-timer.C:
			}
			timer.Stop()
		}

		// Heartbeat if quiet for >= the interval. Comment line; SSE
		// clients ignore it per spec.
		if time.Since(lastWrite) >= sseHeartbeatInterval {
			if _, err := fmt.Fprint(w, ":keep-alive\n\n"); err != nil {
				return
			}
			flusher.Flush()
			lastWrite = time.Now()
		}
	}
}

// streamSnapshot sends one `set` event per existing record key, then returns
// the stream position the live tail continues from.
//
// This is how a consumer that rebuilds its state from events (meta-fuse sends
// Last-Event-ID 0-0 on every start) gets ALL of it. The stream itself only
// retains ~100k entries, so replaying it never could; meta-core used to
// republish every field into it on boot instead, which took minutes and still
// lost everything past the retention window.
//
// The head is captured BEFORE the snapshot is read, so a change racing the
// snapshot appears in the tail too — re-applying a `set` is idempotent. Only
// the final snapshot event carries an `id:`, so a consumer that disconnects
// mid-snapshot still holds its old cursor and gets a fresh snapshot.
func streamSnapshot(ctx context.Context, store *kv.Store, w http.ResponseWriter, flusher http.Flusher) (string, error) {
	head, err := store.LastID(ctx, storage.MetaEventsStream)
	if err != nil {
		return "", err
	}
	ts := strconv.FormatInt(time.Now().UnixMilli(), 10)
	pending := ""
	err = store.ScanPrefix(ctx, "file:", 5000, func(rows []kv.KV) error {
		for _, row := range rows {
			if pending != "" {
				writeSnapshotEvent(w, pending, ts, "")
			}
			pending = row.K
		}
		flusher.Flush()
		return ctx.Err()
	})
	if err != nil {
		return "", err
	}
	if pending != "" {
		writeSnapshotEvent(w, pending, ts, head)
		flusher.Flush()
	}
	return head, nil
}

func writeSnapshotEvent(w http.ResponseWriter, key, ts, id string) {
	payload, _ := json.Marshal(map[string]string{"key": key, "ts": ts})
	if id != "" {
		fmt.Fprintf(w, "id: %s\n", id)
	}
	fmt.Fprintf(w, "event: set\ndata: %s\n\n", payload)
}

// resolveStartCursor decides where to start reading.
//   - No Last-Event-ID → the stream's last issued ID, resolved ONCE here (only
//     entries appended from now on). Redis' "$" was re-resolved on every
//     XREAD, which dropped anything appended between two reads.
//   - Last-Event-ID provided AND still in stream → resume from it.
//   - Last-Event-ID provided BUT trimmed → emit one synthetic `gap` event
//     and resume from the oldest available entry's predecessor "0", so the
//     next read picks up everything still in the stream.
//
// The "still in stream" check is cheap: one read of the first entry.
func resolveStartCursor(ctx context.Context, store *kv.Store, stream, lastID string, w http.ResponseWriter, flusher http.Flusher) string {
	xCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()

	if lastID == "" {
		head, err := store.LastID(xCtx, stream)
		if err != nil {
			log.Printf("[SSE %s] head lookup failed: %v (starting from the beginning)", stream, err)
			return "0"
		}
		return head
	}

	oldestEntry, ok, err := store.Oldest(xCtx, stream)
	if err != nil {
		// Treat any error as "stream not available right now" and resume
		// from the cursor — the read either finds it or waits for new entries.
		log.Printf("[SSE %s] oldest-entry probe failed: %v (resuming from cursor)", stream, err)
		return lastID
	}

	if !ok {
		// Stream is empty. Cursor may be from a previous incarnation; safest
		// is to resume from the cursor — reading an empty stream returns
		// nothing and waits until new entries arrive.
		return lastID
	}

	oldest := oldestEntry.ID
	if compareStreamIDs(lastID, oldest) >= 0 {
		// Cursor is at or past the oldest — fine to resume.
		return lastID
	}

	// Cursor was trimmed. Emit a gap event so consumers that care can react.
	gap := map[string]string{
		"requested":  lastID,
		"resumeFrom": oldest,
		"reason":     "retention",
	}
	payload, _ := json.Marshal(gap)
	fmt.Fprintf(w, "event: gap\ndata: %s\n\n", string(payload))
	flusher.Flush()

	// Return "0" so the next read picks up entries starting from the
	// stream's true minimum — including `oldest` itself.
	return "0"
}

// writeSSEEntry serialises one stream entry as an SSE event.
//
// `event:` comes from the entry's `type` field (no prefix munging). `data:`
// is a JSON object containing all OTHER fields.
func writeSSEEntry(w http.ResponseWriter, flusher http.Flusher, e kv.Entry) {
	eventType := "message"
	data := make(map[string]string, len(e.Values))
	for k, v := range e.Values {
		if k == "type" {
			eventType = v
			continue
		}
		data[k] = v
	}

	payload, err := json.Marshal(data)
	if err != nil {
		log.Printf("[SSE] failed to marshal entry %s: %v", e.ID, err)
		return
	}

	// Spec: each event ends with a blank line. id/event/data lines are
	// terminated with single \n.
	fmt.Fprintf(w, "id: %s\n", e.ID)
	fmt.Fprintf(w, "event: %s\n", eventType)
	fmt.Fprintf(w, "data: %s\n\n", string(payload))
	flusher.Flush()
}

// compareStreamIDs compares two Redis Stream IDs of the form "<ms>-<seq>".
// Returns -1/0/+1. Missing or malformed IDs sort as zero parts.
func compareStreamIDs(a, b string) int {
	aMs, aSeq := parseStreamID(a)
	bMs, bSeq := parseStreamID(b)
	switch {
	case aMs < bMs:
		return -1
	case aMs > bMs:
		return 1
	case aSeq < bSeq:
		return -1
	case aSeq > bSeq:
		return 1
	default:
		return 0
	}
}

func parseStreamID(id string) (int64, int64) {
	parts := strings.SplitN(id, "-", 2)
	ms, _ := strconv.ParseInt(parts[0], 10, 64)
	var seq int64
	if len(parts) == 2 {
		seq, _ = strconv.ParseInt(parts[1], 10, 64)
	}
	return ms, seq
}