package activator

// Unit-backed tenant tests: the same state machine (activator.go) driving a
// carlos-app@ unit through the Systemctl seam instead of an exec'd child —
// no systemd, no AWS. The end-to-end test at the bottom proves the edge
// shape: a request for a sleeping unit tenant is HELD while the "unit"
// starts and turns healthy, then forwarded.

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/registry"
)

// fakeSystemctl records verbs and optionally runs hooks — the Proc-style
// seam's test double. Hooks let the end-to-end test bind/close a real
// listener exactly when the "unit" would.
type fakeSystemctl struct {
	mu       sync.Mutex
	calls    []string
	startErr error
	onStart  func() // runs on "start", outside the lock
	onStop   func() // runs on "stop", outside the lock
}

func (f *fakeSystemctl) run(_ context.Context, verb, unit string) error {
	f.mu.Lock()
	f.calls = append(f.calls, verb+" "+unit)
	err := f.startErr
	onStart, onStop := f.onStart, f.onStop
	f.mu.Unlock()
	if verb == "start" {
		if err != nil {
			return err
		}
		if onStart != nil {
			onStart()
		}
	}
	if verb == "stop" && onStop != nil {
		onStop()
	}
	return nil
}

func (f *fakeSystemctl) recorded() []string {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]string(nil), f.calls...)
}

// testUnitRoute is testRoute's unit-backed sibling: a whole-app tenant
// (carlos-app@<host>) flagged for hibernation.
func testUnitRoute(host string) registry.Route {
	return registry.Route{Host: host, Kind: "instance", Socket: "/run/carlos/apps/" + host + ".sock",
		Hibernate: true, Backing: registry.BackingUnit}
}

// testUnitActivator wires a Service over a SplitBackend: the usual
// fakeBackend on the exec side, a UnitBackend with the fake systemctl and an
// instant health gate on the unit side.
func testUnitActivator(t *testing.T, idle time.Duration) (*Service, *fakeBackend, *fakeSystemctl, *fakeClock, *registry.Registry) {
	t.Helper()
	eb := &fakeBackend{}
	sc := &fakeSystemctl{}
	ub := &UnitBackend{
		Systemctl:   sc.run,
		OptDir:      t.TempDir(),
		WaitHealthy: func(context.Context, registry.Route) error { return nil },
	}
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatalf("open registry: %v", err)
	}
	a := New(Config{Bucket: "test-bucket", Idle: idle}, &SplitBackend{Exec: eb, Unit: ub}, reg)
	a.now = clk.now
	return a, eb, sc, clk, reg
}

// The whole-app hibernation cycle: a request starts the unit (systemctl
// start, health-gated); the idle window stops it (systemctl stop); the next
// request starts it again. The exec side is never touched.
func TestUnitTenantWakeIdleSleepCycle(t *testing.T) {
	t.Parallel()
	a, eb, sc, clk, _ := testUnitActivator(t, 10*time.Second)
	route := testUnitRoute("kass.example.com")
	unit := UnitName(route.Host)

	release, err := a.beginRequest(context.Background(), route)
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	if got := sc.recorded(); len(got) != 1 || got[0] != "start "+unit {
		t.Fatalf("wake systemctl calls = %v, want [start %s]", got, unit)
	}

	// Not yet idle: the unit stays up.
	clk.advance(9 * time.Second)
	a.sweepOnce()
	if a.entryCount() != 1 {
		t.Fatal("unit tenant hibernated before its idle window elapsed")
	}

	// Idle: stopped via systemctl, entry forgotten.
	clk.advance(2 * time.Second)
	a.sweepOnce()
	waitFor(t, "unit hibernation", func() bool { return a.entryCount() == 0 })
	if got := sc.recorded(); len(got) != 2 || got[1] != "stop "+unit {
		t.Fatalf("sleep systemctl calls = %v, want [... stop %s]", got, unit)
	}

	// Re-wake: a fresh start.
	release, err = a.beginRequest(context.Background(), route)
	if err != nil {
		t.Fatalf("beginRequest after sleep: %v", err)
	}
	release()
	if got := sc.recorded(); len(got) != 3 || got[2] != "start "+unit {
		t.Fatalf("re-wake systemctl calls = %v, want a second start", got)
	}

	// The exec backend was never involved: no lease claims, no spawns.
	if acquires, restores, starts := eb.counts(); acquires != 0 || restores != 0 || starts != 0 {
		t.Fatalf("unit tenant leaked into the exec backend: acquires=%d restores=%d starts=%d", acquires, restores, starts)
	}
}

// SplitBackend dispatch, the other direction: an exec-backed route wakes
// through the fake exec backend and never touches systemctl.
func TestSplitBackendExecRouteBypassesUnit(t *testing.T) {
	t.Parallel()
	a, eb, sc, _, _ := testUnitActivator(t, time.Minute)

	release, err := a.beginRequest(context.Background(), testRoute("plain.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	if acquires, restores, starts := eb.counts(); acquires != 1 || restores != 1 || starts != 1 {
		t.Fatalf("exec route skipped the exec backend: acquires=%d restores=%d starts=%d", acquires, restores, starts)
	}
	if got := sc.recorded(); len(got) != 0 {
		t.Fatalf("exec route touched systemctl: %v", got)
	}
}

// Restore and Seed dispatch by Backing like every other SplitBackend method —
// and a unit tenant, whose Restore is never fresh, never reaches a seed even
// with one declared on its route.
func TestSplitBackendDispatchesFreshAndSeedByBacking(t *testing.T) {
	eb := &fakeBackend{restoreFresh: true}
	ub := &UnitBackend{Systemctl: (&fakeSystemctl{}).run, OptDir: t.TempDir()}
	sb := &SplitBackend{Exec: eb, Unit: ub}
	ctx := context.Background()

	if fresh, err := sb.Restore(ctx, testRoute("plain.example.com")); err != nil || !fresh {
		t.Fatalf("exec route Restore = %v, %v; want true, nil", fresh, err)
	}
	unitRoute := testUnitRoute("kass.example.com")
	unitRoute.Seed = `["seed"]`
	if fresh, err := sb.Restore(ctx, unitRoute); err != nil || fresh {
		t.Fatalf("unit route Restore = %v, %v; want false, nil", fresh, err)
	}

	if err := sb.Seed(ctx, testRoute("plain.example.com")); err != nil {
		t.Fatalf("exec route Seed: %v", err)
	}
	if got := eb.seedCount(); got != 1 {
		t.Fatalf("exec-side seeds = %d, want 1", got)
	}
	if err := sb.Seed(ctx, unitRoute); err != nil {
		t.Fatalf("unit route Seed: %v", err)
	}
	if got := eb.seedCount(); got != 1 {
		t.Fatalf("a unit tenant's seed leaked into the exec backend (seeds=%d)", got)
	}
}

// A unit that starts but never turns healthy fails the wake cleanly: the
// unit is put back down (no half-up tenant lingering behind a 503) and no
// entry is left, so the next request retries from scratch.
func TestUnitTenantFailedHealthStopsUnit(t *testing.T) {
	route := testUnitRoute("broken.example.com")
	unit := UnitName(route.Host)

	sc := &fakeSystemctl{}
	ub := &UnitBackend{Systemctl: sc.run, OptDir: t.TempDir(),
		WaitHealthy: func(context.Context, registry.Route) error { return errors.New("never healthy") }}
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatalf("open registry: %v", err)
	}
	a := New(Config{Bucket: "test-bucket", Idle: time.Minute}, &SplitBackend{Exec: &fakeBackend{}, Unit: ub}, reg)

	if _, err := a.beginRequest(context.Background(), route); err == nil {
		t.Fatal("wake succeeded despite the health gate failing")
	}
	if got := sc.recorded(); len(got) != 2 || got[0] != "start "+unit || got[1] != "stop "+unit {
		t.Fatalf("systemctl calls = %v, want [start %s, stop %s]", got, unit, unit)
	}
	if a.entryCount() != 0 {
		t.Fatal("failed unit wake left an entry behind")
	}
}

// A start that fails outright (unit file missing, systemd unreachable)
// fails the wake with no stop attempt and no entry.
func TestUnitTenantStartFailureFailsWake(t *testing.T) {
	a, _, sc, _, _ := testUnitActivator(t, time.Minute)
	sc.startErr = errors.New("Unit carlos-app@gone.example.com.service not found")

	if _, err := a.beginRequest(context.Background(), testUnitRoute("gone.example.com")); err == nil {
		t.Fatal("wake succeeded despite systemctl start failing")
	}
	if got := sc.recorded(); len(got) != 1 {
		t.Fatalf("systemctl calls = %v, want just the failed start", got)
	}
	if a.entryCount() != 0 {
		t.Fatal("failed start left an entry behind")
	}
}

// Shutdown puts unit tenants down like exec children — deliberately: with
// the tenant's carlos-app@.socket unit enabled, systemd's wake-on-connect
// brings it back on the next connection even while the agent is away, and
// the next agent's health-gated wake re-adopts it (start on a running unit
// is a no-op).
func TestUnitTenantShutdownStopsUnit(t *testing.T) {
	a, _, sc, _, _ := testUnitActivator(t, time.Hour)
	route := testUnitRoute("kass.example.com")

	release, err := a.beginRequest(context.Background(), route)
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	a.Shutdown(ctx)
	if a.entryCount() != 0 {
		t.Fatal("shutdown left the unit entry behind")
	}
	got := sc.recorded()
	if len(got) != 2 || got[1] != "stop "+UnitName(route.Host) {
		t.Fatalf("systemctl calls = %v, want the shutdown stop", got)
	}
}

// The inert lease/replication seams: everything the state machine calls on
// them must succeed and report healthy, so the machine treats a unit tenant
// exactly like a replicating exec child. (Replication ownership really
// lives with the root carlos-litestream unit — see backend_unit.go.)
func TestUnitBackendInertSeams(t *testing.T) {
	b := &UnitBackend{Systemctl: (&fakeSystemctl{}).run, OptDir: t.TempDir()}
	ctx := context.Background()
	rt := testUnitRoute("inert.example.com")

	lease, err := b.Acquire(ctx, rt)
	if err != nil {
		t.Fatalf("Acquire: %v", err)
	}
	rep, err := lease.Replicate(ctx, rt.DBPath, 0)
	if err != nil {
		t.Fatalf("Replicate: %v", err)
	}
	if err := rep.Err(); err != nil {
		t.Fatalf("Rep.Err: %v", err)
	}
	if err := rep.Close(ctx); err != nil {
		t.Fatalf("Rep.Close: %v", err)
	}
	if err := lease.Release(ctx); err != nil {
		t.Fatalf("Release: %v", err)
	}
	// Never fresh: the agent cannot see inside a DynamicUser's state dir, so
	// it must never claim a unit tenant is on its first boot — that claim is
	// what would run an app's seed over live data.
	if fresh, err := b.Restore(ctx, rt); err != nil || fresh {
		t.Fatalf("Restore = %v, %v; want false, nil", fresh, err)
	}
	if err := b.Seed(ctx, rt); err != nil {
		t.Fatalf("Seed: %v", err)
	}
	due, err := b.NextDue(ctx, rt)
	if err != nil || due != 0 {
		t.Fatalf("NextDue = %d, %v; want 0, nil (unit tenants sleep until a request)", due, err)
	}
	if b.HasSidecar(rt) {
		t.Fatal("unit tenants must not report agent-managed sidecars")
	}
}

// Worker spawn is exec-child machinery, exactly like sidecar co-spawn: a
// unit tenant that wants a worker runs it inside its own unit.
func TestUnitBackendRefusesWorkers(t *testing.T) {
	var b UnitBackend
	if _, _, err := b.StartWorker(context.Background(), registry.Route{Host: "x"}, nil); err == nil {
		t.Fatal("unit-backed tenants have no agent-managed workers, exactly as they have no agent-managed sidecars")
	}
}

// LiveVersion reads the same live/<host> symlink layout as ExecBackend, so
// build convergence cycles awake unit tenants onto shipped builds.
func TestUnitBackendLiveVersion(t *testing.T) {
	opt := t.TempDir()
	b := &UnitBackend{Systemctl: (&fakeSystemctl{}).run, OptDir: opt}

	if v, err := b.LiveVersion("new.example.com"); err != nil || v != "" {
		t.Fatalf("LiveVersion(no symlink) = %q, %v; want empty, nil", v, err)
	}
	verDir := filepath.Join(opt, "versions", "abc1234")
	if err := os.MkdirAll(verDir, 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.MkdirAll(filepath.Join(opt, "live"), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.Symlink(filepath.Join(verDir, "app"), filepath.Join(opt, "live", "kass.example.com")); err != nil {
		t.Fatal(err)
	}
	if v, err := b.LiveVersion("kass.example.com"); err != nil || v != "abc1234" {
		t.Fatalf("LiveVersion = %q, %v; want abc1234", v, err)
	}

	// A migrated tenant's link has the account sqid between the version and
	// the binary; reading "the parent directory" would return the sqid and
	// convince ConvergeDecision this tenant is converged forever.
	stageLiveUnderAccount(t, opt, "moved.example.com", "kqmvexample", "def5678", "kass")
	if v, err := b.LiveVersion("moved.example.com"); err != nil || v != "def5678" {
		t.Fatalf("LiveVersion (account-scoped) = %q, %v; want def5678", v, err)
	}
}

// The end-to-end shape the feature promises: a request for a SLEEPING unit
// tenant is held at the edge while the unit starts and its upstream turns
// healthy (the real TCP prober — no injected health seam here), then
// forwarded; after the idle window the unit is stopped for real (listener
// closed); the next request repeats the cycle. TCP Addr keeps the test
// sandbox-safe; the unix-socket path differs only in the dialer.
func TestUnitTenantWakeServesHeldRequest(t *testing.T) {
	// Pre-pick the tenant's port so the route can exist before the "unit" runs.
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("pick port: %v", err)
	}
	addr := ln.Addr().String()
	ln.Close()

	// The "unit": the fake systemctl binds/closes a real HTTP listener on
	// start/stop, with a deliberate boot delay so the request provably waits
	// on the health gate.
	var srvMu sync.Mutex
	var srv *http.Server
	sc := &fakeSystemctl{}
	sc.onStart = func() {
		time.Sleep(200 * time.Millisecond) // boot time — the edge must hold the request
		l, err := net.Listen("tcp", addr)
		if err != nil {
			t.Errorf("unit listen: %v", err)
			return
		}
		s := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			io.WriteString(w, "kass up")
		})}
		srvMu.Lock()
		srv = s
		srvMu.Unlock()
		go s.Serve(l)
	}
	sc.onStop = func() {
		srvMu.Lock()
		s := srv
		srv = nil
		srvMu.Unlock()
		if s != nil {
			s.Close()
		}
	}

	old := spawnHealthTimeout
	spawnHealthTimeout = 5 * time.Second
	t.Cleanup(func() { spawnHealthTimeout = old })

	eb := &fakeBackend{}
	ub := &UnitBackend{Systemctl: sc.run, OptDir: t.TempDir()} // nil WaitHealthy: the real prober
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatalf("open registry: %v", err)
	}
	a := New(Config{Bucket: "test-bucket", Idle: 10 * time.Second}, &SplitBackend{Exec: eb, Unit: ub}, reg)
	a.now = clk.now

	route := registry.Route{Host: "kass.example.com", Kind: "instance", Addr: addr,
		Hibernate: true, Backing: registry.BackingUnit}
	if err := reg.Upsert(route); err != nil {
		t.Fatal(err)
	}
	rt, err := newTestRouter(reg, a)
	if err != nil {
		t.Fatalf("new router: %v", err)
	}
	front := httptest.NewServer(rt)
	t.Cleanup(front.Close)
	t.Cleanup(func() { sc.onStop() }) // in case the test dies mid-cycle

	// Asleep → the request wakes the unit and is answered by it.
	if body, code := proxyGet(t, front.URL, "kass.example.com"); code != 200 || body != "kass up" {
		t.Fatalf("held request = %d %q, want 200 kass up", code, body)
	}

	// Idle out → the unit is stopped for real: its listener is gone.
	clk.advance(11 * time.Second)
	a.sweepOnce()
	waitFor(t, "unit hibernation", func() bool { return a.entryCount() == 0 })
	waitFor(t, "listener closed", func() bool {
		conn, err := net.DialTimeout("tcp", addr, 100*time.Millisecond)
		if err == nil {
			conn.Close()
			return false
		}
		return true
	})

	// The next request wakes it again.
	if body, code := proxyGet(t, front.URL, "kass.example.com"); code != 200 || body != "kass up" {
		t.Fatalf("post-sleep request = %d %q, want 200 kass up", code, body)
	}
	want := []string{
		"start " + UnitName(route.Host),
		"stop " + UnitName(route.Host),
		"start " + UnitName(route.Host),
	}
	if got := sc.recorded(); fmt.Sprint(got) != fmt.Sprint(want) {
		t.Fatalf("systemctl calls = %v, want %v", got, want)
	}
}