package calls

import (
	"bytes"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
	"time"

	"github.com/go-chi/chi/v5"

	"amadan.net/rastrillo/rastrillo/sessions"

	"amadan.net/oficina/calls/internal/calls/hub"
)

const (
	// defaultHeartbeat and defaultStreamTTL are Docs' live-sync numbers
	// (spec §4.1): a comment line every fifteen seconds so a dead stream
	// is noticed, and a five-minute life so hibernation is never held
	// off by one. The browser resumes; nothing is lost at the TTL.
	defaultHeartbeat  = 15 * time.Second
	defaultStreamTTL  = 5 * time.Minute
	streamWriteWindow = 30 * time.Second
	// minStreamTiming is the floor under both, so a zero or negative
	// value from SetStreamTimings cannot panic a ticker in a request
	// goroutine.
	minStreamTiming = 10 * time.Millisecond
	// maxLastQuery bounds the resume's ?last= before it is parsed.
	maxLastQuery = 8 << 10
)

// sessionKey is the session's identity for the hub (spec §4.2): the
// session cookie this request presented, keyed through a salt this
// process chose at boot, so what the hub holds is neither the token nor
// the hash rastrillo stores. Two tabs of one browser share it, which is
// what one peer per room and its takeover need; two browsers of one
// person do not, which §4.4's four streams per person allows.
//
// The plan expected the subject plus the session's creation instant.
// That is not an identity: two sign-ins of one person within one second
// collide, and the second browser is then refused the room.
func (a *app) sessionKey(r *http.Request) (string, bool) {
	s, ok := sessions.Current(r)
	if !ok || s.Subject == "" {
		return "", false
	}
	c, err := r.Cookie(a.auth.SessionCookie())
	if err != nil || c.Value == "" {
		return "", false
	}
	return hubSessionKey(a.hubSalt, s.Subject, c.Value), true
}

// hubSessionKey is the derivation itself, kept apart from the request so
// its three properties can be gated (review W-4): the subject is bound
// in as well as the cookie, so the key cannot survive the person it
// belongs to whatever a cookie value might come to be reused for; the
// per-boot salt means it is meaningless across a restart; and the
// result is neither the session token nor the hash rastrillo stores, so
// a hub key is not a credential.
func hubSessionKey(salt, subject, cookie string) string {
	sum := sha256.Sum256([]byte(salt + "|" + subject + "|" + cookie))
	return hex.EncodeToString(sum[:])
}

// acceptsEventStream is ruling 15. The attaching stream is only ever
// opened by EventSource, which always asks for text/event-stream; a
// cross-site top-level navigation asks for HTML, and without this check
// it would take the session's one peer slot in the room and hold it for
// the resume window. Anything else is 406 with {"error":"accept"}. The
// probe forms are exempt: they attach nothing.
func acceptsEventStream(r *http.Request) bool {
	return strings.Contains(strings.ToLower(r.Header.Get("Accept")), "text/event-stream")
}

// testTTLHeader shortens one stream's life, and only in development.
// EventSource cannot set a header, so nothing a browser does on its own
// can reach this; the browser gate rewrites it on the way past, which is
// the only way to watch a renewal without waiting five minutes. Outside
// development the header is not read at all: a request may not shorten
// a stream on its own say-so.
const testTTLHeader = "X-Calls-Test-TTL"

// streamLifeFor is streamLife with that override applied, through the
// same floor atLeast puts under every timing — a header cannot panic a
// timer in a request goroutine any more than SetStreamTimings can.
func (a *app) streamLifeFor(r *http.Request) time.Duration {
	if !a.cfg.development() {
		return a.streamLife()
	}
	raw := r.Header.Get(testTTLHeader)
	if raw == "" {
		return a.streamLife()
	}
	d, err := time.ParseDuration(raw)
	if err != nil {
		return a.streamLife()
	}
	return atLeast(int64(d))
}

// streamJSON is every refusal this endpoint makes. The room's client is
// a script, not a page: spec §4.5 fixes the browser's behaviour per
// status and body, so a refusal answers JSON rather than the error page
// a screen would get.
func streamJSON(w http.ResponseWriter, status int, body map[string]any) {
	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Cache-Control", "no-store")
	w.WriteHeader(status)
	if body != nil {
		_ = json.NewEncoder(w).Encode(body)
	}
}

func (a *app) refuseResume(w http.ResponseWriter, re *hub.ResumeError) {
	body := map[string]any{}
	if re.Reason != "" {
		body["reason"] = string(re.Reason)
	}
	if re.Boot != "" {
		body["boot"] = re.Boot
	}
	streamJSON(w, re.Status, body)
}

// stream is GET /m/{sqid}/stream: one SSE stream per peer (spec §4.1),
// opened fresh on standing or resumed on a participation bound to this
// session (§2.5). ?probe=1 answers the same statuses and stops there.
func (a *app) stream(w http.ResponseWriter, r *http.Request) {
	sqid := chi.URLParam(r, "sqid")
	person, ok := currentPerson(r)
	if !ok {
		a.logger.Error("calls: stream with no person in context", "path", r.URL.Path)
		streamJSON(w, http.StatusInternalServerError, nil)
		return
	}
	session, ok := a.sessionKey(r)
	if !ok {
		a.logger.Error("calls: stream with no session subject", "path", r.URL.Path)
		streamJSON(w, http.StatusInternalServerError, nil)
		return
	}
	probe := r.URL.Query().Get("probe") == "1"
	if !probe && !acceptsEventStream(r) {
		// Not a stream client, so nothing is looked up and nothing is
		// allocated (ruling 15). This runs before any authorization, so
		// the answer is identical for every sqid and every person and
		// discloses nothing — which is why it can afford to be the
		// truthful 406 rather than the room's 404, and why a
		// mis-headered client is told to fix its header instead of
		// being sent round the fresh-stream loop.
		streamJSON(w, http.StatusNotAcceptable, map[string]any{"error": "accept"})
		return
	}

	var sub *hub.Sub
	var peer hub.PeerID
	var first []hub.Message

	if resume := hub.PeerID(r.URL.Query().Get("peer")); resume != "" {
		// A resume is authorized by the participation, not by standing
		// (§2.5), but the meeting itself must still be live: End writes
		// no headstone for a room nobody opened, so ended_at is the only
		// authority a fresh process has.
		m, found, err := a.store.meetingBySqid(r.Context(), sqid)
		if err != nil {
			a.logger.Error("calls: stream: load meeting", "err", err)
			streamJSON(w, http.StatusInternalServerError, nil)
			return
		}
		if !found || m.EndedAt != nil {
			// The hub answers first when it holds a tombstone, so a peer
			// that was in the room when it ended is told why (410 ended)
			// rather than sent to open a fresh stream.
			if re := a.hub.CanResume(r.Context(), sqid, session, resume); re != nil {
				a.refuseResume(w, re)
				return
			}
			a.refuseResume(w, &hub.ResumeError{Status: http.StatusNotFound, Boot: a.hub.Boot()})
			return
		}
		if probe {
			if re := a.hub.CanResume(r.Context(), sqid, session, resume); re != nil {
				a.refuseResume(w, re)
				return
			}
			streamJSON(w, http.StatusNoContent, nil)
			return
		}
		var last map[hub.PeerID]uint64
		if raw := r.URL.Query().Get("last"); raw != "" {
			// One seq per peer, sixteen peers to a room: a document this
			// far past that is not a browser's, and is refused before it
			// is parsed.
			if len(raw) > maxLastQuery {
				streamJSON(w, http.StatusBadRequest, map[string]any{"error": "bad last"})
				return
			}
			if err := json.Unmarshal([]byte(raw), &last); err != nil {
				streamJSON(w, http.StatusBadRequest, map[string]any{"error": "bad last"})
				return
			}
		}
		res, err := a.hub.Resume(r.Context(), sqid, session, resume, last)
		if err != nil {
			re, ok := err.(*hub.ResumeError)
			if !ok {
				a.logger.Error("calls: stream: resume", "err", err)
				streamJSON(w, http.StatusInternalServerError, nil)
				return
			}
			a.refuseResume(w, re)
			return
		}
		sub, peer = res.Sub, resume
		// Order matters: the roster, then the replay, then the gaps, all
		// written before anything the live subscriber carries. The
		// subscriber is live from the moment Resume returned, and the
		// browser discards what is at or below the seq it already has,
		// so a fresh candidate written ahead of the replay would make
		// the replay look stale.
		first = append(first, hub.Message{Type: hub.TypeHello, Peer: peer, Boot: a.hub.Boot(), Report: res.ReportToken})
		if res.Admitted {
			first = append(first, hub.Message{Type: hub.TypeRoster, Roster: res.Roster})
			first = append(first, res.Replay...)
			for _, from := range res.Gap {
				first = append(first, hub.Message{Type: hub.TypeGap, From: from})
			}
		} else {
			// Waiting or unjoined: nothing about the room (§4.5).
			first = append(first, hub.Message{Type: hub.TypeState, Waiting: true})
		}
	} else {
		m, found, err := a.store.meetingBySqid(r.Context(), sqid)
		if err != nil {
			a.logger.Error("calls: stream: load meeting", "err", err)
			streamJSON(w, http.StatusInternalServerError, nil)
			return
		}
		if !found || m.EndedAt != nil {
			streamJSON(w, http.StatusNotFound, map[string]any{"boot": a.hub.Boot()})
			return
		}
		st, err := a.standingFor(r.Context(), person, m)
		if err != nil {
			a.logger.Error("calls: stream: standing", "err", err)
			streamJSON(w, http.StatusInternalServerError, nil)
			return
		}
		if st == StandingNone {
			// Not this person's meeting to enter: 404, never 403 (§2.5).
			streamJSON(w, http.StatusNotFound, map[string]any{"boot": a.hub.Boot()})
			return
		}
		if probe {
			if existing, live := a.hub.LiveForSession(sqid, session); live {
				streamJSON(w, http.StatusConflict, map[string]any{"peer": string(existing)})
				return
			}
			streamJSON(w, http.StatusNoContent, nil)
			return
		}
		kind := hub.KindGuest
		if st == StandingMember {
			kind = hub.KindMember
		}
		res, err := a.hub.Open(sqid, session, person.ID, person.Name, kind)
		switch err {
		case nil:
		case hub.ErrSessionHasPeer:
			streamJSON(w, http.StatusConflict, map[string]any{"peer": string(res.Existing)})
			return
		case hub.ErrRoomFull, hub.ErrPersonStreams:
			streamJSON(w, http.StatusServiceUnavailable, map[string]any{"error": "full"})
			return
		default:
			// ErrUnknownPeer here is the room's headstone: the meeting
			// ended between the read above and the hub lock.
			streamJSON(w, http.StatusNotFound, map[string]any{"boot": a.hub.Boot()})
			return
		}
		sub, peer = res.Sub, res.Peer
		first = append(first, hub.Message{Type: hub.TypeHello, Peer: peer, Boot: a.hub.Boot(), Report: res.ReportToken})
	}

	// From here the participation exists and is this handler's to hold
	// open. Every exit detaches it — this handler's own subscriber, so a
	// stream displaced by a second one does not detach its replacement —
	// which starts the thirty-second resume window (§4.5).
	defer a.hub.Detach(sqid, peer, sub)
	a.serveStream(w, r, sub, first)
}

// serveStream writes the events already decided, then the subscriber's,
// until the client goes, the TTL expires, or the stream is lost.
func (a *app) serveStream(w http.ResponseWriter, r *http.Request, sub *hub.Sub, first []hub.Message) {
	rc := http.NewResponseController(w)
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-store")
	// Nothing between this process and the browser should buffer an
	// event stream; nginx and its imitators read this header.
	w.Header().Set("X-Accel-Buffering", "no")
	w.WriteHeader(http.StatusOK)
	if err := rc.Flush(); err != nil {
		return
	}
	for _, m := range first {
		if !writeEvent(w, rc, m) {
			return
		}
	}
	heartbeat := time.NewTicker(a.heartbeatEvery())
	defer heartbeat.Stop()
	ttl := time.NewTimer(a.streamLifeFor(r))
	defer ttl.Stop()
	for {
		select {
		case <-r.Context().Done():
			return
		case <-a.done:
			// The process is going down. Shutdown waits for handlers and
			// does not cancel their contexts, so a live stream would
			// hold the whole ten-second deadline and turn a clean stop
			// into a failed one. The browser resumes, as it does at the
			// TTL.
			return
		case <-ttl.C:
			return
		case <-heartbeat.C:
			if !writeRaw(w, rc, ": hb\n\n") {
				return
			}
		case m, ok := <-sub.C():
			if !ok {
				return
			}
			if !writeEvent(w, rc, m) {
				return
			}
			if m.Type == hub.TypeLost || m.Type == hub.TypeEnded {
				// Terminal for this stream either way: lost{buffer} is
				// followed by a resume, the rest by a screen that says
				// why (§4.5).
				return
			}
		}
	}
}

// writeEvent puts one message on the wire as an SSE frame.
//
// The encoder, rather than json.Marshal, for one reason: Marshal
// HTML-escapes `<`, `>` and `&` into <, > and &, and it
// does so inside a json.RawMessage too. A chat of 2048 literal `<` —
// validated at 2059 raw bytes, inside §4.4's cap — left this process
// as 12,299 payload bytes, six times what was measured (Codex S4-2).
// SSE data is not HTML and no reader of this stream treats it as
// markup: room.js parses the JSON and puts every string it contains on
// the page with textContent. Nothing here depends on the escaping.
//
// What the escaping never protected against, and what actually matters
// on a line-oriented wire, is a payload ending its own frame: JSON
// escapes a newline inside a string as \n rather than emitting one, so
// no value can produce the blank line that dispatches an event.
// callstest's TestChatCannotForgeASecondSSEFrame is the gate on that.
func writeEvent(w http.ResponseWriter, rc *http.ResponseController, m hub.Message) bool {
	var buf bytes.Buffer
	enc := json.NewEncoder(&buf)
	enc.SetEscapeHTML(false)
	if err := enc.Encode(m); err != nil {
		return false
	}
	// Encode writes its own trailing newline; the frame below supplies
	// the two this wire format needs.
	b := bytes.TrimRight(buf.Bytes(), "\n")
	return writeRaw(w, rc, fmt.Sprintf("event: %s\ndata: %s\n\n", m.Type, b))
}

// writeRaw writes one SSE frame under a deadline, so a client that has
// stopped reading cannot hold a goroutine and a participation open.
func writeRaw(w http.ResponseWriter, rc *http.ResponseController, s string) bool {
	_ = rc.SetWriteDeadline(time.Now().Add(streamWriteWindow))
	if _, err := fmt.Fprint(w, s); err != nil {
		return false
	}
	return rc.Flush() == nil
}