package main

// 🤖 The live two-box mesh tests ("the kill-a-box demo", #550 Part 4 v2
// Phase 3): TWO in-process boxes — two activator+router pairs on different
// loopback ports, each spawning REAL instance processes — sharing one REAL S3
// bucket for leases, replicas, and mesh heartbeats. They prove the mesh's
// whole story:
//
//   - a request landing on the NON-owner box is reverse-proxied to the HRW
//     owner, which wakes the instance and serves;
//   - killing the owner box moves its instance to the survivor with data
//     intact — in BOTH kill flavors:
//       drain      (TestMeshLiveProxyAndDrainFailover): heartbeat deleted
//                  first, checkpoint + lease release — the survivor takes
//                  over on its next membership poll;
//       hard kill  (TestMeshLiveHardKillFailover): the box just dies — its
//                  heartbeat rots to staleness, its lease expires on TTL, and
//                  the survivor's wake is gated by the lease until then.
//
// Gated on LCHAT_TEST_S3_BUCKET like the other live tests. Run with:
//
//	eval "$(aws configure export-credentials --profile eleven-dev --format env)"
//	LCHAT_TEST_S3_BUCKET=<bucket> go test -run TestMeshLive -v .

import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"lchat/internal/activator"
)

// liveMeshBox is one box of the pair: its own activator (real s3Backend, real
// child processes), registry, mesh, and front door.
type liveMeshBox struct {
	name  string
	self  string // the front URL — this box's mesh identity
	act   *activatorSvc
	mesh  *meshSvc
	front *httptest.Server
	reg   *Registry
	route Route

	shutdownOnce sync.Once
}

// shutdown drains the box exactly once — tests that dismantle a box by hand
// (the hard kill) consume the Once so cleanup can't double-drain it.
func (b *liveMeshBox) shutdown(ctx context.Context) {
	b.shutdownOnce.Do(func() { b.act.shutdown(ctx) })
}

// setupMeshLive builds the shared world (S3 prefix, staged binary, env file)
// and the two boxes. Both boxes register the same host, each pointing at its
// OWN socket and DB path — a box must never see its peer's local files, so a
// failover's restore provably comes from S3.
func setupMeshLive(t *testing.T, host string) (boxA, boxB *liveMeshBox, adminToken string) {
	t.Helper()
	bucket := os.Getenv("LCHAT_TEST_S3_BUCKET")
	if bucket == "" {
		t.Skip("LCHAT_TEST_S3_BUCKET not set; skipping live two-box mesh test")
	}
	region := os.Getenv("LCHAT_TEST_S3_REGION")
	if region == "" {
		region = "eu-west-1"
	}
	rb := make([]byte, 4)
	_, _ = rand.Read(rb)
	prefix := fmt.Sprintf("mesh-live/%d-%s", time.Now().Unix(), hex.EncodeToString(rb))
	t.Cleanup(func() { cleanupS3Prefix(t, bucket, prefix, region) })

	// Sandbox the box-layout globals (shared by both in-process boxes: the
	// staged binary, live symlink, and the instance env file are box-agnostic).
	dataDir := t.TempDir()
	sockDir := fmt.Sprintf("/tmp/lchat-mesh-%d", os.Getpid())
	if err := os.MkdirAll(sockDir, 0o755); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(sockDir) })
	optTmp := t.TempDir()
	oldData, oldSock, oldOpt := instanceDataDir, instanceSocketDir, optDir
	instanceDataDir, instanceSocketDir, optDir = dataDir, sockDir, optTmp
	t.Cleanup(func() { instanceDataDir, instanceSocketDir, optDir = oldData, oldSock, oldOpt })

	bin := filepath.Join(optTmp, "lchat-built")
	build := exec.Command("go", "build", "-o", bin, ".")
	build.Env = append(os.Environ(), "GOTOOLCHAIN=auto")
	if out, err := build.CombinedOutput(); err != nil {
		t.Fatalf("go build: %v\n%s", err, out)
	}
	if err := stageVersion(bin, "meshlive"); err != nil {
		t.Fatalf("stage: %v", err)
	}
	if err := pointLive(host, "meshlive"); err != nil {
		t.Fatalf("point live: %v", err)
	}

	adminToken = "meshlive-" + hex.EncodeToString(rb)
	if err := writeEnvFile(filepath.Join(dataDir, host+".env"),
		map[string]string{"LCHAT_ADMIN_TOKEN": adminToken}); err != nil {
		t.Fatalf("env file: %v", err)
	}

	cfg := hibernateConfig{Bucket: bucket, Region: region, Prefix: prefix, Idle: 2 * time.Minute}
	newBox := func(name string) *liveMeshBox {
		box := &liveMeshBox{name: name}
		var err error
		box.reg, err = openRegistry(filepath.Join(t.TempDir(), "registry.db"))
		if err != nil {
			t.Fatalf("open registry: %v", err)
		}
		dbDir := filepath.Join(dataDir, name)
		if err := os.MkdirAll(dbDir, 0o755); err != nil {
			t.Fatal(err)
		}
		box.route = Route{Host: host, Kind: "instance",
			Socket:    filepath.Join(sockDir, name+".sock"),
			DBPath:    filepath.Join(dbDir, host+".db"),
			Hibernate: true}
		if err := box.reg.upsert(box.route); err != nil {
			t.Fatalf("upsert: %v", err)
		}
		box.act = newActivator(cfg, &s3Backend{cfg: cfg}, box.reg)
		rt, err := newRouter(box.reg, box.act)
		if err != nil {
			t.Fatalf("new router: %v", err)
		}
		box.front = httptest.NewServer(rt)
		t.Cleanup(box.front.Close)
		box.self = box.front.URL

		hb, err := activator.NewHeartbeatStore(context.Background(), bucket, cfg.meshPrefix(), region)
		if err != nil {
			t.Fatalf("heartbeat store: %v", err)
		}
		box.mesh = newMesh(box.self, hb, nil)
		box.mesh.interval = time.Second
		box.mesh.staleAfter = 3 * time.Second
		box.act.mesh = box.mesh
		box.mesh.start()
		box.act.start()
		t.Cleanup(func() {
			ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
			defer cancel()
			box.shutdown(ctx)
		})
		return box
	}
	boxA, boxB = newBox("box-a"), newBox("box-b")

	// Both boxes must see the full membership before the test routes anything.
	waitForLive(t, "both boxes to see each other", 15*time.Second, func() bool {
		return len(boxA.mesh.liveBoxes()) == 2 && len(boxB.mesh.liveBoxes()) == 2
	})
	return boxA, boxB, adminToken
}

// pickOwner orders the pair as (owner, survivor) for host by the same HRW
// hash the boxes themselves use.
func pickOwner(host string, a, b *liveMeshBox) (owner, survivor *liveMeshBox) {
	if hrwOwner(host, []string{a.self, b.self}) == b.self {
		return b, a
	}
	return a, b
}

func meshGet(t *testing.T, front, host, path string) (int, string) {
	t.Helper()
	req, _ := http.NewRequest("GET", front+path, nil)
	req.Host = host
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("GET %s: %v", path, err)
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)
	return resp.StatusCode, string(body)
}

func meshSetGroupName(t *testing.T, front, host, token, name string) {
	t.Helper()
	body, _ := json.Marshal(map[string]string{"name": name})
	req, _ := http.NewRequest("POST", front+"/api/admin/group", bytes.NewReader(body))
	req.Host = host
	req.Header.Set("Authorization", "Bearer "+token)
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("set group name: %v", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != 200 {
		b, _ := io.ReadAll(resp.Body)
		t.Fatalf("set group name: %d %s", resp.StatusCode, b)
	}
}

// pollUntilServed hammers the survivor's front door until the instance
// answers 200, logging every distinct (status, body) transition — the test's
// narrative of the failover: proxy-to-dead-peer 503s, then lease-held 503s,
// then the wake. Returns the winning body.
func pollUntilServed(t *testing.T, front, host string, deadline time.Duration) string {
	t.Helper()
	start := time.Now()
	last := ""
	for time.Now().Before(start.Add(deadline)) {
		code, body := meshGet(t, front, host, "/api/group")
		if code == 200 {
			t.Logf("  +%s: 200 — served", time.Since(start).Truncate(time.Millisecond))
			return body
		}
		if cur := fmt.Sprintf("%d %s", code, strings.TrimSpace(body)); cur != last {
			t.Logf("  +%s: %s", time.Since(start).Truncate(time.Millisecond), cur)
			last = cur
		}
		time.Sleep(time.Second)
	}
	t.Fatalf("instance never served on the survivor within %s", deadline)
	return ""
}

// The drain variant: proxy-to-owner, then an orderly box exit (heartbeat
// deleted FIRST, checkpoint, lease release) and the survivor taking over on
// its next membership poll — the same code path a deploy or scale-down runs.
func TestMeshLiveProxyAndDrainFailover(t *testing.T) {
	const host = "mesh-drain.test"
	boxA, boxB, adminToken := setupMeshLive(t, host)
	owner, survivor := pickOwner(host, boxA, boxB)
	t.Logf("HRW owner of %s is %s (%s); survivor is %s (%s)", host, owner.name, owner.self, survivor.name, survivor.self)

	// --- Act 1: a request to the NON-owner proxies to the owner and serves. ---
	start := time.Now()
	code, _ := meshGet(t, survivor.front.URL, host, "/api/group")
	if code != 200 {
		t.Fatalf("request via non-owner: %d, want 200", code)
	}
	t.Logf("request to the non-owner proxied + woke the instance on %s in %s", owner.name, time.Since(start).Truncate(time.Millisecond))
	if owner.act.entryCount() != 1 || !owner.act.entryAwake(host) {
		t.Fatalf("instance is not awake on the owner (entries=%d)", owner.act.entryCount())
	}
	if survivor.act.entryCount() != 0 {
		t.Fatal("the non-owner woke a local copy instead of proxying")
	}

	// Write real data through the proxied path.
	groupName := "Mesh failover " + hex.EncodeToString([]byte(host))
	meshSetGroupName(t, survivor.front.URL, host, adminToken, groupName)
	t.Logf("wrote group name %q through the non-owner's front door", groupName)

	// --- Act 2: drain the owner. Front door dark first (runRouter shuts the
	// HTTP servers down before the activator — a draining box accepts no new
	// work), then the drain proper: heartbeat delete, checkpoint, release. ---
	start = time.Now()
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	owner.front.Close()
	owner.shutdown(ctx)
	t.Logf("owner %s drained in %s (heartbeat deleted, instance checkpointed, lease released)", owner.name, time.Since(start).Truncate(time.Millisecond))

	// The heartbeat really is gone from the bucket — only the survivor beats.
	hbs, err := survivor.mesh.store.List(context.Background())
	if err != nil {
		t.Fatalf("list heartbeats: %v", err)
	}
	for _, hb := range hbs {
		if hb.Peer == owner.self {
			t.Fatal("drained owner's heartbeat is still in the bucket")
		}
	}

	// --- Act 3: the survivor takes over — wake local, restore from S3. ---
	body := pollUntilServed(t, survivor.front.URL, host, 60*time.Second)
	if !strings.Contains(body, groupName) {
		t.Fatalf("survivor lost the data: %s does not contain %q", body, groupName)
	}
	if !survivor.act.entryAwake(host) {
		t.Fatal("instance is not awake on the survivor")
	}
	t.Logf("survivor %s restored the instance from S3 with data intact: %s", survivor.name, body)
}

// The hard-kill variant: the owner box dies with no goodbye — its heartbeat
// is left to rot (stale in staleAfter) and its lease to expire (TTL). The
// survivor's requests walk the whole failure narrative: proxy to the corpse
// (503), local wake gated by the still-held lease (503 + Retry-After), then
// the lease expires and the instance wakes on the survivor with every byte
// that had replicated.
func TestMeshLiveHardKillFailover(t *testing.T) {
	const host = "mesh-kill.test"
	boxA, boxB, adminToken := setupMeshLive(t, host)
	owner, survivor := pickOwner(host, boxA, boxB)
	t.Logf("HRW owner of %s is %s (%s); survivor is %s (%s)", host, owner.name, owner.self, survivor.name, survivor.self)

	// --- Act 1: wake on the owner and write data. ---
	if code, _ := meshGet(t, owner.front.URL, host, "/api/group"); code != 200 {
		t.Fatalf("wake on owner: %d, want 200", code)
	}
	groupName := "Hard kill survivor " + hex.EncodeToString([]byte(host))
	meshSetGroupName(t, owner.front.URL, host, adminToken, groupName)

	// Make the write durable in S3 before the "crash" — a real hard kill
	// loses only the last seconds of unshipped WAL (that's the design's RPO);
	// the test pins the data so the assertion is about failover, not timing.
	owner.act.mu.Lock()
	entry := owner.act.entries[host]
	owner.act.mu.Unlock()
	if entry == nil {
		t.Fatal("no live entry on the owner")
	}
	rep := entry.rep.(*activator.LeasedReplicator)
	syncCtx, syncCancel := context.WithTimeout(context.Background(), 30*time.Second)
	if err := rep.SyncAndWait(syncCtx); err != nil {
		syncCancel()
		t.Fatalf("sync before kill: %v", err)
	}
	syncCancel()
	t.Logf("wrote group name %q on the owner and confirmed it replicated to S3", groupName)

	// --- Act 2: hard-kill the owner box. Everything at once, nothing orderly:
	// front door gone, sweeper gone, heartbeat loop gone (object NOT deleted),
	// instance child SIGKILLed, lease renewal abandoned (NOT released). ---
	owner.front.Close()
	close(owner.act.sweepStop)
	<-owner.act.sweepDone
	owner.mesh.halt() // the heartbeat object stays — it must rot to staleness
	owner.act.mu.Lock()
	delete(owner.act.entries, host)
	owner.act.mu.Unlock()
	proc := entry.proc.(*childProc)
	if err := proc.cmd.Process.Kill(); err != nil {
		t.Fatalf("SIGKILL instance child: %v", err)
	}
	<-proc.done
	abandonCtx, abandonCancel := context.WithTimeout(context.Background(), 30*time.Second)
	rep.Abandon(abandonCtx)
	abandonCancel()
	owner.shutdownOnce.Do(func() {}) // consume: cleanup must not re-drain a corpse
	t.Logf("owner %s hard-killed: heartbeat left in S3 (stale after %s), lease left to expire (TTL %s)",
		owner.name, survivor.mesh.staleAfter, activator.LeaseTTL)

	// --- Act 3: the survivor takes over once staleness + lease expiry pass. ---
	start := time.Now()
	body := pollUntilServed(t, survivor.front.URL, host, 120*time.Second)
	if !strings.Contains(body, groupName) {
		t.Fatalf("survivor lost the data: %s does not contain %q", body, groupName)
	}
	if !survivor.act.entryAwake(host) {
		t.Fatal("instance is not awake on the survivor")
	}
	t.Logf("survivor %s took over %s after the owner's death with data intact: %s",
		survivor.name, time.Since(start).Truncate(time.Millisecond), body)
}