package main

// Tests for the mini-app server surface (miniapps.go): scheduled "future
// messages" and per-thread state snapshots. Delivery is exercised by calling
// sweepDueScheduled directly — the ticker only runs in main().

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

	"github.com/coder/websocket"
)

func postJSON(t *testing.T, url, body string) *http.Response {
	t.Helper()
	resp, err := http.Post(url, "application/json", strings.NewReader(body))
	if err != nil {
		t.Fatalf("POST %s: %v", url, err)
	}
	return resp
}

func doReq(t *testing.T, method, url, body string) *http.Response {
	t.Helper()
	var rd *strings.Reader
	if body == "" {
		rd = strings.NewReader("")
	} else {
		rd = strings.NewReader(body)
	}
	req, _ := http.NewRequest(method, url, rd)
	if body != "" {
		req.Header.Set("Content-Type", "application/json")
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("%s %s: %v", method, url, err)
	}
	return resp
}

func decodeJSON(t *testing.T, resp *http.Response, into any) {
	t.Helper()
	defer resp.Body.Close()
	if err := json.NewDecoder(resp.Body).Decode(into); err != nil {
		t.Fatalf("decode: %v", err)
	}
}

// Scheduling is member-gated and validates its inputs.
func TestScheduleAuthAndValidation(t *testing.T) {
	ts, srv := newTestServer(t)
	tok, _, tid := joinedMember(t, srv, "Sam")
	future := time.Now().Add(time.Hour).Unix()

	// Bad token → 403.
	if r := postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"bogus","thread_id":"%s","deliver_at":%d,"blob":"x"}`, tid, future)); r.StatusCode != http.StatusForbidden {
		t.Fatalf("bad token: want 403, got %d", r.StatusCode)
	}
	// Not a member of the thread → 403.
	tok2, _, _ := joinedMember(t, srv, "Rival")
	if r := postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"x"}`, tok2, tid, future)); r.StatusCode != http.StatusForbidden {
		t.Fatalf("non-member: want 403, got %d", r.StatusCode)
	}
	// Empty blob → 400.
	if r := postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":""}`, tok, tid, future)); r.StatusCode != http.StatusBadRequest {
		t.Fatalf("empty blob: want 400, got %d", r.StatusCode)
	}
	// deliver_at too far out (past the horizon) → 400.
	tooFar := time.Now().Add(scheduleHorizon + 24*time.Hour).Unix()
	if r := postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"x"}`, tok, tid, tooFar)); r.StatusCode != http.StatusBadRequest {
		t.Fatalf("past horizon: want 400, got %d", r.StatusCode)
	}
	// deliver_at deep in the past → 400 (a minute of skew is tolerated).
	if r := postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"x"}`, tok, tid, time.Now().Add(-time.Hour).Unix())); r.StatusCode != http.StatusBadRequest {
		t.Fatalf("deep past: want 400, got %d", r.StatusCode)
	}
	// Valid → 200 with an id.
	r := postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"ciphertext"}`, tok, tid, future))
	if r.StatusCode != http.StatusOK {
		t.Fatalf("valid schedule: want 200, got %d", r.StatusCode)
	}
	var out struct {
		ID string `json:"id"`
	}
	decodeJSON(t, r, &out)
	if out.ID == "" {
		t.Fatal("schedule returned no id")
	}

	// The pending schedule is listable by a member, not by a non-member.
	r, _ = http.Get(ts.URL + "/api/schedule?token=" + tok + "&thread=" + tid)
	var list struct {
		Schedules []ScheduledMessage `json:"schedules"`
	}
	decodeJSON(t, r, &list)
	if len(list.Schedules) != 1 || list.Schedules[0].ID != out.ID || list.Schedules[0].Blob != "ciphertext" {
		t.Fatalf("list: %+v", list.Schedules)
	}
	if r, _ = http.Get(ts.URL + "/api/schedule?token=" + tok2 + "&thread=" + tid); r.StatusCode != http.StatusForbidden {
		t.Fatalf("non-member list: want 403, got %d", r.StatusCode)
	}
}

// A due schedule moves into the message log exactly once; an undue one stays.
func TestScheduleDelivery(t *testing.T) {
	ts, srv := newTestServer(t)
	tok, _, tid := joinedMember(t, srv, "Sam")

	// One due now (grace window lets a just-past time in), one due in an hour.
	now := time.Now().Unix()
	r := postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"due-blob"}`, tok, tid, now-1))
	if r.StatusCode != http.StatusOK {
		t.Fatalf("schedule due: want 200, got %d", r.StatusCode)
	}
	postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"later-blob"}`, tok, tid, now+3600))

	if n := srv.sweepDueScheduled(); n != 1 {
		t.Fatalf("sweep: want 1 delivered, got %d", n)
	}
	// Delivered blob is now a normal message in the thread.
	msgs, err := srv.db.messagesAfter(tid, 0, 10)
	if err != nil || len(msgs) != 1 || msgs[0].Blob != "due-blob" {
		t.Fatalf("messages after sweep: %v %+v", err, msgs)
	}
	// Sweep again: nothing due — no duplicate delivery.
	if n := srv.sweepDueScheduled(); n != 0 {
		t.Fatalf("second sweep: want 0, got %d", n)
	}
	if msgs, _ = srv.db.messagesAfter(tid, 0, 10); len(msgs) != 1 {
		t.Fatalf("after second sweep: want 1 message, got %d", len(msgs))
	}
	// The undue schedule is still pending.
	pending, _ := srv.db.listScheduledMessages(tid)
	if len(pending) != 1 || pending[0].Blob != "later-blob" {
		t.Fatalf("pending after sweep: %+v", pending)
	}
}

// The activator's due-time probe: /api/next-due is admin-gated and reports the
// earliest pending deliver_at (0 when nothing is scheduled) so the router can
// wake a hibernating instance in time to fire it.
func TestNextDue(t *testing.T) {
	ts, srv := newTestServer(t)
	tok, _, tid := joinedMember(t, srv, "Sam")

	getDue := func(bearer string) (int, int64) {
		t.Helper()
		req, _ := http.NewRequest("GET", ts.URL+"/api/next-due", nil)
		if bearer != "" {
			req.Header.Set("Authorization", "Bearer "+bearer)
		}
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			t.Fatalf("GET /api/next-due: %v", err)
		}
		if resp.StatusCode != http.StatusOK {
			resp.Body.Close()
			return resp.StatusCode, 0
		}
		var body struct {
			Due int64 `json:"due"`
		}
		decodeJSON(t, resp, &body)
		return resp.StatusCode, body.Due
	}

	// No auth → 401; a plain member's device token isn't admin either.
	if code, _ := getDue(""); code != http.StatusUnauthorized {
		t.Fatalf("unauthenticated: want 401, got %d", code)
	}
	if code, _ := getDue(tok); code != http.StatusUnauthorized {
		t.Fatalf("member token: want 401, got %d", code)
	}

	// Nothing scheduled → 0.
	if code, due := getDue("admintok"); code != 200 || due != 0 {
		t.Fatalf("empty schedule: want 200/0, got %d/%d", code, due)
	}

	// The earliest of several pending schedules wins, regardless of post order.
	now := time.Now().Unix()
	later, sooner := now+7200, now+3600
	postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"later"}`, tok, tid, later))
	postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"sooner"}`, tok, tid, sooner))
	if _, due := getDue("admintok"); due != sooner {
		t.Fatalf("due = %d, want the earlier %d", due, sooner)
	}

	// Delivery consumes the row, so the next due time advances.
	pending, _ := srv.db.listScheduledMessages(tid)
	if err := srv.db.deleteScheduledMessage(pending[0].ID); err != nil { // pending is deliver_at-ordered: [0] = sooner
		t.Fatalf("delete: %v", err)
	}
	if _, due := getDue("admintok"); due != later {
		t.Fatalf("after consuming the sooner schedule: due = %d, want %d", due, later)
	}
}

// combineNextDue: the MIN composition behind /api/next-due, and the staleness
// rule that keeps a dead sidecar's report from wake-looping a hibernating
// instance.
func TestCombineNextDue(t *testing.T) {
	const now = int64(1_000_000)
	cases := []struct {
		name                 string
		sched, charter, want int64
	}{
		{"neither", 0, 0, 0},
		{"only messages", now + 100, 0, now + 100},
		{"only charter", 0, now + 200, now + 200},
		{"both, schedule sooner", now + 100, now + 200, now + 100},
		{"both, charter sooner", now + 300, now + 200, now + 200},
		// Staleness: a past charter report is ignored — it changes only when
		// the sidecar re-reports, so honoring it would re-wake the instance
		// after every sleep, forever (at-most-one wasted wake, never a loop).
		{"past charter ignored", 0, now - 10, 0},
		{"past charter never shadows the queue", now + 100, now - 10, now + 100},
		{"charter due exactly now still fires", 0, now, now},
		// The asymmetry: a past SCHEDULE deliver_at is honored — delivery
		// consumes the row, so waking for it is right and self-limiting.
		{"past schedule honored", now - 10, 0, now - 10},
	}
	for _, c := range cases {
		if got := combineNextDue(c.sched, c.charter, now); got != c.want {
			t.Errorf("%s: combineNextDue(%d, %d) = %d, want %d", c.name, c.sched, c.charter, got, c.want)
		}
	}
}

// /api/next-due folds the sidecar-reported charter due-time in with the
// schedule queue — the endpoint-level view of combineNextDue.
func TestNextDueFoldsCharterDue(t *testing.T) {
	ts, srv := newTestServer(t)
	tok, _, tid := joinedMember(t, srv, "Sam")

	getDue := func() int64 {
		t.Helper()
		req, _ := http.NewRequest("GET", ts.URL+"/api/next-due", nil)
		req.Header.Set("Authorization", "Bearer admintok")
		resp, err := http.DefaultClient.Do(req)
		if err != nil || resp.StatusCode != http.StatusOK {
			t.Fatalf("GET /api/next-due: %v %d", err, resp.StatusCode)
		}
		var body struct {
			Due int64 `json:"due"`
		}
		decodeJSON(t, resp, &body)
		return body.Due
	}

	now := time.Now().Unix()

	// Only a charter report → it is the answer.
	if err := srv.db.setCharterNextDue(now + 600); err != nil {
		t.Fatal(err)
	}
	if due := getDue(); due != now+600 {
		t.Fatalf("charter only: due = %d, want %d", due, now+600)
	}

	// A sooner scheduled message wins; a later one doesn't.
	postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"x"}`, tok, tid, now+300))
	if due := getDue(); due != now+300 {
		t.Fatalf("sooner schedule: due = %d, want %d", due, now+300)
	}
	if err := srv.db.setCharterNextDue(now + 60); err != nil {
		t.Fatal(err)
	}
	if due := getDue(); due != now+60 {
		t.Fatalf("sooner charter: due = %d, want %d", due, now+60)
	}

	// A stale (past) charter report is ignored — the queue's answer stands.
	if err := srv.db.setCharterNextDue(now - 60); err != nil {
		t.Fatal(err)
	}
	if due := getDue(); due != now+300 {
		t.Fatalf("stale charter: due = %d, want %d", due, now+300)
	}
}

// Cancellation is member-gated; cancelling an already-fired/unknown id is quiet.
func TestScheduleCancel(t *testing.T) {
	ts, srv := newTestServer(t)
	tok, _, tid := joinedMember(t, srv, "Sam")
	tok2, _, _ := joinedMember(t, srv, "Rival")
	future := time.Now().Add(time.Hour).Unix()

	r := postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"x"}`, tok, tid, future))
	var out struct {
		ID string `json:"id"`
	}
	decodeJSON(t, r, &out)

	// A non-member may not cancel someone else's schedule.
	if r := doReq(t, "DELETE", ts.URL+"/api/schedule/"+out.ID+"?token="+tok2, ""); r.StatusCode != http.StatusForbidden {
		t.Fatalf("non-member cancel: want 403, got %d", r.StatusCode)
	}
	// A member may.
	if r := doReq(t, "DELETE", ts.URL+"/api/schedule/"+out.ID+"?token="+tok, ""); r.StatusCode != http.StatusOK {
		t.Fatalf("member cancel: want 200, got %d", r.StatusCode)
	}
	if pending, _ := srv.db.listScheduledMessages(tid); len(pending) != 0 {
		t.Fatalf("after cancel: %+v", pending)
	}
	// Cancelled schedules never deliver.
	if n := srv.sweepDueScheduled(); n != 0 {
		t.Fatalf("sweep after cancel: want 0, got %d", n)
	}
	// Cancelling again (already gone) is quiet success — but still needs a token.
	if r := doReq(t, "DELETE", ts.URL+"/api/schedule/"+out.ID+"?token="+tok, ""); r.StatusCode != http.StatusOK {
		t.Fatalf("re-cancel: want 200, got %d", r.StatusCode)
	}
	if r := doReq(t, "DELETE", ts.URL+"/api/schedule/"+out.ID+"?token=bogus", ""); r.StatusCode != http.StatusForbidden {
		t.Fatalf("re-cancel bad token: want 403, got %d", r.StatusCode)
	}
}

// A real DB failure while resolving the schedule's thread must surface as 5xx,
// never a false 200 — otherwise the client believes the reminder was cancelled
// while it still fires. Dropping the table forces a non-ErrNoRows lookup error.
func TestScheduleCancelDBError(t *testing.T) {
	ts, srv := newTestServer(t)
	tok, _, tid := joinedMember(t, srv, "Sam")
	future := time.Now().Add(time.Hour).Unix()

	r := postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"x"}`, tok, tid, future))
	var out struct {
		ID string `json:"id"`
	}
	decodeJSON(t, r, &out)

	if _, err := srv.db.sql.Exec("DROP TABLE scheduled_messages"); err != nil {
		t.Fatalf("drop table: %v", err)
	}
	if r := doReq(t, "DELETE", ts.URL+"/api/schedule/"+out.ID+"?token="+tok, ""); r.StatusCode != http.StatusInternalServerError {
		t.Fatalf("cancel during db error: want 500, got %d", r.StatusCode)
	}
}

// State snapshots: member-gated read/write with optimistic versioning.
func TestThreadState(t *testing.T) {
	ts, srv := newTestServer(t)
	tok, _, tid := joinedMember(t, srv, "Sam")
	tok2, _, _ := joinedMember(t, srv, "Rival")
	stateURL := ts.URL + "/api/threads/" + tid + "/state"

	// Nothing saved yet: version 0, empty blob.
	r, _ := http.Get(stateURL + "?token=" + tok + "&slot=apps")
	var st struct {
		Blob    string `json:"blob"`
		Version int64  `json:"version"`
	}
	decodeJSON(t, r, &st)
	if st.Version != 0 || st.Blob != "" {
		t.Fatalf("fresh state: %+v", st)
	}

	// Create (version 0 → 1).
	r = doReq(t, "PUT", stateURL, fmt.Sprintf(`{"token":"%s","slot":"apps","blob":"snap1","version":0}`, tok))
	var put struct {
		Version int64 `json:"version"`
	}
	if r.StatusCode != http.StatusOK {
		t.Fatalf("create state: want 200, got %d", r.StatusCode)
	}
	decodeJSON(t, r, &put)
	if put.Version != 1 {
		t.Fatalf("create version: want 1, got %d", put.Version)
	}

	// A stale writer (still claiming version 0) conflicts.
	if r = doReq(t, "PUT", stateURL, fmt.Sprintf(`{"token":"%s","slot":"apps","blob":"snap-stale","version":0}`, tok)); r.StatusCode != http.StatusConflict {
		t.Fatalf("stale write: want 409, got %d", r.StatusCode)
	}
	// The current version updates fine (1 → 2).
	if r = doReq(t, "PUT", stateURL, fmt.Sprintf(`{"token":"%s","slot":"apps","blob":"snap2","version":1}`, tok)); r.StatusCode != http.StatusOK {
		t.Fatalf("update state: want 200, got %d", r.StatusCode)
	}

	// Read back: latest blob + version.
	r, _ = http.Get(stateURL + "?token=" + tok + "&slot=apps")
	decodeJSON(t, r, &st)
	if st.Blob != "snap2" || st.Version != 2 {
		t.Fatalf("read back: %+v", st)
	}

	// Non-members can neither read nor write.
	if r, _ = http.Get(stateURL + "?token=" + tok2 + "&slot=apps"); r.StatusCode != http.StatusForbidden {
		t.Fatalf("non-member read: want 403, got %d", r.StatusCode)
	}
	if r = doReq(t, "PUT", stateURL, fmt.Sprintf(`{"token":"%s","slot":"apps","blob":"evil","version":2}`, tok2)); r.StatusCode != http.StatusForbidden {
		t.Fatalf("non-member write: want 403, got %d", r.StatusCode)
	}
}

// An ephemeral msg frame is relayed to the thread's members but never stored:
// presence traffic (typing, live cursors) must leave no trace in the log.
func TestEphemeralRelayNotStored(t *testing.T) {
	ts, srv := newTestServer(t)
	tokA, uidA, _ := joinedMember(t, srv, "Alice")
	tokB, uidB, _ := joinedMember(t, srv, "Bob")
	tid, err := srv.db.createThread("group", "", []ThreadMemberKey{{UserID: uidA, Wrapped: "wa"}, {UserID: uidB, Wrapped: "wb"}})
	if err != nil {
		t.Fatalf("createThread: %v", err)
	}

	wsURL := "ws" + strings.TrimPrefix(ts.URL, "http") + "/ws?token="
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	dial := func(tok string) *websocket.Conn {
		c, _, err := websocket.Dial(ctx, wsURL+tok, nil)
		if err != nil {
			t.Fatalf("ws dial: %v", err)
		}
		return c
	}
	connA := dial(tokA)
	defer connA.CloseNow()
	connB := dial(tokB)
	defer connB.CloseNow()
	// Registration races the first send (the hub processes register/broadcast on
	// one loop, but Dial returns before the server's register lands). A stored
	// message round-trip on A both proves the pipe and orders B's registration.
	if err := connA.Write(ctx, websocket.MessageText, []byte(`{"type":"msg","thread_id":"`+tid+`","blob":"warmup"}`)); err != nil {
		t.Fatalf("warmup write: %v", err)
	}
	if _, _, err := connA.Read(ctx); err != nil {
		t.Fatalf("warmup read: %v", err)
	}

	// A sends an ephemeral frame; B must receive it flagged, with no id.
	if err := connA.Write(ctx, websocket.MessageText, []byte(`{"type":"msg","thread_id":"`+tid+`","blob":"eph-blob","ephemeral":true}`)); err != nil {
		t.Fatalf("ephemeral write: %v", err)
	}
	// B first drains the warmup broadcast, then gets the ephemeral frame.
	var got wsOut
	for i := 0; i < 3; i++ {
		_, data, err := connB.Read(ctx)
		if err != nil {
			t.Fatalf("B read: %v", err)
		}
		if err := json.Unmarshal(data, &got); err != nil {
			t.Fatalf("B unmarshal: %v", err)
		}
		if got.Ephemeral {
			break
		}
	}
	if !got.Ephemeral || got.Blob != "eph-blob" || got.ID != 0 {
		t.Fatalf("ephemeral frame: %+v", got)
	}
	// Only the warmup message was stored — the ephemeral one left no trace.
	msgs, err := srv.db.messagesAfter(tid, 0, 10)
	if err != nil || len(msgs) != 1 || msgs[0].Blob != "warmup" {
		t.Fatalf("stored messages: %v %+v", err, msgs)
	}
}