package host

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/runs"
)

const runHost = "acme.example.com"

func runRecord() provision.Record {
	return provision.Record{Schema: 1, Host: runHost, Slug: "acme", App: "shop", Account: "bab", Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()}
}

func runOwnedRoute(host string) func(string) (*registry.Route, error) {
	return func(h string) (*registry.Route, error) {
		if h != host {
			return nil, nil
		}
		return &registry.Route{Host: host, Kind: "instance", Hibernate: true, Socket: "/run/x.sock", Provisioned: true, App: "shop", Account: "bab"}, nil
	}
}

func noRoutes(string) (*registry.Route, error) { return nil, nil }

// queueRun writes a record whose run is queued, with its request.json.
func queueRun(t *testing.T, s release.Store, deadline time.Time) (provision.Record, string) {
	t.Helper()
	rec := runRecord()
	if err := provision.CreateRecord(context.Background(), s, rec); err != nil {
		t.Fatal(err)
	}
	id := "run_abcdefghijklmnopqrstuvwxyz"
	inc, _ := hostclaimIncarnation(rec)
	req := runs.Request{Schema: 1, ID: id, Account: rec.Account, App: rec.App, Host: rec.Host, Incarnation: inc, Argv: []string{"import"}, TimeoutS: 60, RequestedAt: time.Now()}
	putJSONObject(t, s, runs.RequestKey(rec.Account, rec.App, id), req)
	got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunAdmit(provision.RunAdmission{ID: id, By: "alice@keymail.dev", Deadline: deadline}))
	if err != nil {
		t.Fatal(err)
	}
	return *got, id
}

type execLog struct {
	mu  sync.Mutex
	ids []string
}

func (l *execLog) exec(_ context.Context, _ provision.Record, req runs.Request, _ registry.Route, _ *runHandle) {
	l.mu.Lock()
	l.ids = append(l.ids, req.ID)
	l.mu.Unlock()
}
func (l *execLog) count() int { l.mu.Lock(); defer l.mu.Unlock(); return len(l.ids) }

// testWorker execs inline, so a test counts launches as soon as consider
// returns.
func testWorker(s release.Store, routes func(string) (*registry.Route, error), box, boot string, log *execLog) *runsWorker {
	w := newRunsWorker(s, routes, nil, "/data", box, boot)
	w.exec = log.exec
	w.start = func(f func()) { f() }
	w.booted.Store(true) // a test of the boot gate unsets it
	return w
}

func currentRun(t *testing.T, s release.Store) *provision.RunState {
	t.Helper()
	rec, err := provision.GetRecord(context.Background(), s, "bab", "shop", runHost)
	if err != nil {
		t.Fatal(err)
	}
	return rec.ActiveRun
}

func TestRunsWorkerPicksOnlyRunsOnItsOwnRows(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, id := queueRun(t, s, time.Now().Add(time.Minute))
	var other, mine execLog
	testWorker(s, noRoutes, "box-2", "boot-2", &other).consider(context.Background(), rec)
	if other.count() != 0 || currentRun(t, s).State != provision.RunQueued {
		t.Fatal("a box that does not own the row acted on the run")
	}
	testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &mine).consider(context.Background(), rec)
	r := currentRun(t, s)
	if mine.count() != 1 || r.State != provision.RunPicked || r.Box != "box-1" || r.BootID != "boot-1" || r.ID != id {
		t.Fatalf("pickup: execs %d, run %+v", mine.count(), r)
	}
	if d := time.Until(r.Deadline); d < 50*time.Second || d > 61*time.Second {
		t.Errorf("the picked deadline is the run's timeout (60 s), got %s", d)
	}
}

func TestRunsWorkerPickupRaceLaunchesOnce(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, _ := queueRun(t, s, time.Now().Add(time.Minute))
	var a, b execLog
	wa := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &a)
	wb := testWorker(s, runOwnedRoute(runHost), "box-2", "boot-2", &b)
	var wg sync.WaitGroup
	for _, w := range []*runsWorker{wa, wb} {
		wg.Add(1)
		go func() { defer wg.Done(); w.consider(context.Background(), rec) }()
	}
	wg.Wait()
	if a.count()+b.count() != 1 {
		t.Fatalf("launched %d times, want once", a.count()+b.count())
	}
}

func TestRunsWorkerLeavesARunPastItsPickupDeadline(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, _ := queueRun(t, s, time.Now().Add(-time.Second))
	var l execLog
	testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), rec)
	if l.count() != 0 || currentRun(t, s).State != provision.RunQueued {
		t.Fatal("a box picked a run past its pickup deadline: that run is the expiry's")
	}
}

func TestRunsWorkerNeverPicksACancelledOrExpiredRun(t *testing.T) {
	for _, mut := range []func(string) func(*provision.Record) error{provision.RunCancelTransition, provision.RunExpire} {
		s := release.NewDirStore(t.TempDir())
		rec, id := queueRun(t, s, time.Now().Add(time.Minute))
		if _, err := provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, mut(id)); err != nil {
			t.Fatal(err)
		}
		var l execLog
		// rec is the stale read the pass noted, still queued: the CAS refuses.
		testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), rec)
		if l.count() != 0 {
			t.Fatal("a cancelled or expired run was launched (codex R2-6)")
		}
	}
}

func TestRunsWorkerClosesOutARunPickedUnderAnEarlierBoot(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	_, id := queueRun(t, s, time.Now().Add(time.Minute))
	got, err := provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunPick(id, "box-1", "boot-old", time.Now().Add(time.Hour)))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Put(context.Background(), runs.ChunkKey("bab", "shop", id, 4, runs.StreamOut), []byte("x"), ""); err != nil {
		t.Fatal(err)
	}
	var l execLog
	testWorker(s, runOwnedRoute(runHost), "box-1", "boot-new", &l).consider(context.Background(), *got)
	if l.count() != 0 {
		t.Fatal("a run picked before an agent restart was run again")
	}
	res, err := runs.ReadResult(context.Background(), s, "bab", "shop", id)
	if err != nil || res == nil || res.Interrupted != runs.InterruptedAgentRestart || res.LastSeq != 4 || res.Box != "box-1" {
		t.Fatalf("result %+v, %v", res, err)
	}
	if r := currentRun(t, s); r.State != provision.RunDone {
		t.Fatalf("run %+v, want done", r)
	}
}

func TestRunsWorkerForwardsACancelToItsRun(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	_, id := queueRun(t, s, time.Now().Add(time.Minute))
	var l execLog
	w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
	h := newRunHandle()
	w.active[id] = h
	got, err := provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
	if err != nil {
		t.Fatal(err)
	}
	if got, err = provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunCancelTransition(id)); err != nil {
		t.Fatal(err)
	}
	w.consider(context.Background(), *got)
	select {
	case <-h.cancel:
	default:
		t.Fatal("the cancel did not reach the run")
	}
}

func TestBootPassPicksARunQueuedBeforeStart(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	queueRun(t, s, time.Now().Add(time.Minute))
	var l execLog
	w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
	w.dataDir = t.TempDir()
	w.bootPass(context.Background())
	if l.count() != 1 {
		t.Fatal("the boot pass did not pick a run queued before agent start")
	}
}

func TestRunsWorkerNotesAndPasses(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, _ := queueRun(t, s, time.Now().Add(time.Minute))
	var l execLog
	w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
	w.note(rec)
	w.note(rec)
	w.pass(context.Background())
	w.pass(context.Background())
	if l.count() != 1 {
		t.Fatalf("execs %d, want 1: a noted record is acted on once", l.count())
	}
}

func TestRunDirNamesAgree(t *testing.T) {
	if runs.DirName != activator.CmdRunDirName {
		t.Fatalf("runs.DirName %q, activator.CmdRunDirName %q", runs.DirName, activator.CmdRunDirName)
	}
}

func hostclaimIncarnation(rec provision.Record) (string, error) {
	return hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
}

func putJSONObject(t *testing.T, s release.Store, key string, v any) {
	t.Helper()
	raw, err := json.Marshal(v)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Put(context.Background(), key, raw, ""); err != nil {
		t.Fatal(err)
	}
}

// A run another box picked is that box's: this box neither closes it out
// nor forwards anything to it, whatever boot picked it.
func TestRunsWorkerLeavesARunAnotherBoxPicked(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	_, id := queueRun(t, s, time.Now().Add(time.Minute))
	got, err := provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunPick(id, "box-2", "boot-old", time.Now().Add(time.Hour)))
	if err != nil {
		t.Fatal(err)
	}
	var l execLog
	testWorker(s, runOwnedRoute(runHost), "box-1", "boot-new", &l).consider(context.Background(), *got)
	if res, err := runs.ReadResult(context.Background(), s, "bab", "shop", id); err != nil || res != nil {
		t.Fatalf("a box wrote a result for a run another box picked: %+v, %v", res, err)
	}
	if r := currentRun(t, s); l.count() != 0 || r.State != provision.RunPicked || r.Box != "box-2" {
		t.Fatalf("execs %d, run %+v", l.count(), r)
	}
}

// A run this box picked under its own boot is not an agent restart, even
// when the worker holds no handle for it (a stale record noted after the
// run's goroutine let go).
func TestRunsWorkerLeavesItsOwnRunUnderThisBoot(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	_, id := queueRun(t, s, time.Now().Add(time.Minute))
	got, err := provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
	if err != nil {
		t.Fatal(err)
	}
	var l execLog
	testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), *got)
	if res, err := runs.ReadResult(context.Background(), s, "bab", "shop", id); err != nil || res != nil {
		t.Fatalf("a run of this boot was closed out as interrupted: %+v, %v", res, err)
	}
	if r := currentRun(t, s); l.count() != 0 || r.State != provision.RunPicked {
		t.Fatalf("execs %d, run %+v", l.count(), r)
	}
}

// A run of this boot that is no longer executing and has its result, but is
// still picked (its done CAS failed past the finish bound), is CASed to done
// on the next pass, so it does not hold the instance until the lost sweep.
func TestRunsWorkerFinishesItsOwnStrandedRun(t *testing.T) {
	for _, state := range []string{provision.RunPicked, provision.RunCancel} {
		s := release.NewDirStore(t.TempDir())
		_, id := queueRun(t, s, time.Now().Add(time.Minute))
		got, err := provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
		if err != nil {
			t.Fatal(err)
		}
		if state == provision.RunCancel {
			if got, err = provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunCancelTransition(id)); err != nil {
				t.Fatal(err)
			}
		}
		if err := runs.WriteResult(context.Background(), s, "bab", "shop", id, runs.Result{Exit: 3, Box: "box-1"}); err != nil {
			t.Fatal(err)
		}
		var l execLog
		testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), *got)
		if r := currentRun(t, s); l.count() != 0 || r.State != provision.RunDone {
			t.Fatalf("%s: execs %d, run %+v, want done", state, l.count(), r)
		}
		if res, _ := runs.ReadResult(context.Background(), s, "bab", "shop", id); res == nil || res.Exit != 3 {
			t.Fatalf("%s: the box's result changed: %+v", state, res)
		}
	}
}

// Only a provisioned row for the record's account and app that the
// activator manages is this box's to pick for.
func TestRunsWorkerPicksOnlyForActivatorManagedRowsOfTheRecord(t *testing.T) {
	for name, mut := range map[string]func(*registry.Route){
		"unprovisioned":       func(rt *registry.Route) { rt.Provisioned = false },
		"another account":     func(rt *registry.Route) { rt.Account = "other" },
		"another app":         func(rt *registry.Route) { rt.App = "other" },
		"not hibernating":     func(rt *registry.Route) { rt.Hibernate = false },
		"unit-backed":         func(rt *registry.Route) { rt.Backing = "unit" },
		"not an instance row": func(rt *registry.Route) { rt.Kind = "service" },
	} {
		s := release.NewDirStore(t.TempDir())
		rec, _ := queueRun(t, s, time.Now().Add(time.Minute))
		routes := func(h string) (*registry.Route, error) {
			rt, err := runOwnedRoute(runHost)(h)
			if rt != nil {
				mut(rt)
			}
			return rt, err
		}
		var l execLog
		testWorker(s, routes, "box-1", "boot-1", &l).consider(context.Background(), rec)
		if l.count() != 0 || currentRun(t, s).State != provision.RunQueued {
			t.Errorf("%s: the box picked a run for a row it does not run", name)
		}
	}
}

// A request.json that is not the run the record names, or whose timeout is
// outside 1 s to 6 h, is never picked: the run is left for the expiry.
func TestRunsWorkerLeavesARequestThatDoesNotMatchItsRecord(t *testing.T) {
	for name, mut := range map[string]func(*runs.Request){
		"no timeout":      func(r *runs.Request) { r.TimeoutS = 0 },
		"timeout over 6h": func(r *runs.Request) { r.TimeoutS = 6*3600 + 1 },
		"another id":      func(r *runs.Request) { r.ID = "run_bbbbbbbbbbbbbbbbbbbbbbbbbb" },
		"another account": func(r *runs.Request) { r.Account = "other" },
		"another app":     func(r *runs.Request) { r.App = "other" },
		"another host":    func(r *runs.Request) { r.Host = "other.example.com" },
	} {
		s := release.NewDirStore(t.TempDir())
		rec, id := queueRun(t, s, time.Now().Add(time.Minute))
		req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
		if err != nil {
			t.Fatal(err)
		}
		mut(req)
		key := runs.RequestKey(rec.Account, rec.App, id)
		if err := s.Delete(context.Background(), key); err != nil {
			t.Fatal(err)
		}
		putJSONObject(t, s, key, req)
		var l execLog
		w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
		w.consider(context.Background(), rec)
		if l.count() != 0 || currentRun(t, s).State != provision.RunQueued || len(w.active) != 0 {
			t.Errorf("%s: picked (execs %d, run %+v)", name, l.count(), currentRun(t, s))
		}
	}
}

// A refused request.json is logged once, not on every pass that sees the
// run still queued.
func TestRunsWorkerLogsAnUnfitRequestOnce(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, id := queueRun(t, s, time.Now().Add(time.Minute))
	req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
	if err != nil {
		t.Fatal(err)
	}
	req.TimeoutS = 0
	key := runs.RequestKey(rec.Account, rec.App, id)
	if err := s.Delete(context.Background(), key); err != nil {
		t.Fatal(err)
	}
	putJSONObject(t, s, key, req)
	var buf bytes.Buffer
	log.SetOutput(&buf)
	defer log.SetOutput(os.Stderr)
	var l execLog
	w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
	for range 3 {
		w.consider(context.Background(), rec)
	}
	if n := strings.Count(buf.String(), "not picked"); n != 1 {
		t.Fatalf("logged %d times, want once:\n%s", n, buf.String())
	}
}

// lostAnswer answers every Put of the record with an error; when land is
// set the write lands first, as a request whose answer was lost does.
type lostAnswer struct {
	release.Store
	land bool
}

func (l *lostAnswer) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key != provision.RecordKey("bab", "shop", runHost) {
		return l.Store.Put(ctx, key, body, ifMatch)
	}
	if l.land {
		if _, err := l.Store.Put(ctx, key, body, ifMatch); err != nil {
			return "", err
		}
	}
	return "", errors.New("connection reset by peer")
}

// A pickup CAS whose answer was lost is read back: if the record names
// this run as this box's under this boot, the box launches it, so the run
// is not stranded until the sweep calls it lost.
func TestRunsWorkerLaunchesAPickupWhoseAnswerWasLost(t *testing.T) {
	dir := release.NewDirStore(t.TempDir())
	rec, id := queueRun(t, dir, time.Now().Add(time.Minute))
	var l execLog
	w := testWorker(&lostAnswer{Store: dir, land: true}, runOwnedRoute(runHost), "box-1", "boot-1", &l)
	w.consider(context.Background(), rec)
	if r := currentRun(t, dir); l.count() != 1 || r.State != provision.RunPicked || r.Box != "box-1" || r.ID != id {
		t.Fatalf("execs %d, run %+v", l.count(), r)
	}
}

// A pickup CAS that failed and did not land is forgotten: the run stays
// queued and nothing is launched.
func TestRunsWorkerForgetsAPickupThatDidNotLand(t *testing.T) {
	dir := release.NewDirStore(t.TempDir())
	rec, _ := queueRun(t, dir, time.Now().Add(time.Minute))
	var l execLog
	w := testWorker(&lostAnswer{Store: dir}, runOwnedRoute(runHost), "box-1", "boot-1", &l)
	w.consider(context.Background(), rec)
	if l.count() != 0 || currentRun(t, dir).State != provision.RunQueued || len(w.active) != 0 {
		t.Fatalf("execs %d, run %+v, active %d", l.count(), currentRun(t, dir), len(w.active))
	}
}

func TestRunsHookIsOnlyThereWhenArmed(t *testing.T) {
	armedRuns.Store(nil)
	if cfg := withRunsHook(provision.ReconcileConfig{}); cfg.RunSeen != nil {
		t.Fatal("an unarmed box must not hook reconcile")
	}
	runRunsPass(context.Background())     // no-op, no panic
	runRunsBootPass(context.Background()) // ditto
	var l execLog
	w := testWorker(release.NewDirStore(t.TempDir()), noRoutes, "box-1", "boot-1", &l)
	armedRuns.Store(w)
	t.Cleanup(func() { armedRuns.Store(nil) })
	cfg := withRunsHook(provision.ReconcileConfig{})
	if cfg.RunSeen == nil {
		t.Fatal("an armed box hooks reconcile")
	}
	cfg.RunSeen(runRecord())
	if len(w.noted) != 1 {
		t.Fatal("the hook does not reach the worker")
	}
}

// listGate holds every List until released, so a test can act while the
// boot pass is part-way through.
type listGate struct {
	release.Store
	entered chan struct{}
	proceed chan struct{}
	once    sync.Once
}

func (g *listGate) List(ctx context.Context, prefix string) ([]string, error) {
	g.once.Do(func() { close(g.entered) })
	<-g.proceed
	return g.Store.List(ctx, prefix)
}

// The boot pass deletes every .carlos-run directory (run spec §5.8), so no
// reconcile-driven pass may pick or launch anything before it has finished:
// a run it launched would have its directory swept from under it.
func TestRunsPassDoesNothingUntilTheBootPassHasFinished(t *testing.T) {
	dir := release.NewDirStore(t.TempDir())
	rec, _ := queueRun(t, dir, time.Now().Add(time.Minute))
	gate := &listGate{Store: dir, entered: make(chan struct{}), proceed: make(chan struct{})}
	var l execLog
	w := testWorker(gate, runOwnedRoute(runHost), "box-1", "boot-1", &l)
	w.booted.Store(false)
	w.dataDir = t.TempDir()
	stale := filepath.Join(provision.InstanceDir(w.dataDir, runHost), runs.DirName, "run_old")
	if err := os.MkdirAll(stale, 0o700); err != nil {
		t.Fatal(err)
	}
	armedRuns.Store(w)
	t.Cleanup(func() { armedRuns.Store(nil) })

	// Reconcile ran before the boot pass: the hook notes, the pass waits.
	withRunsHook(provision.ReconcileConfig{}).RunSeen(rec)
	runRunsPass(context.Background())
	if l.count() != 0 || currentRun(t, dir).State != provision.RunQueued {
		t.Fatal("a pass picked a run before the boot pass ran")
	}

	done := make(chan struct{})
	go func() { runRunsBootPass(context.Background()); close(done) }()
	<-gate.entered // the sweep is done, the boot pass is reading records
	if _, err := os.Stat(stale); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the boot pass did not sweep first: %v", err)
	}
	runRunsPass(context.Background())
	if l.count() != 0 || currentRun(t, dir).State != provision.RunQueued {
		t.Fatal("a pass picked a run while the boot pass was still running")
	}
	close(gate.proceed)
	<-done
	if l.count() != 1 {
		t.Fatalf("execs %d, want 1: the boot pass picks the run queued before start", l.count())
	}
}

// A boot pass that cannot read the records still opens the gate: the runs
// it could not see are the next passes' to pick.
func TestRunsPassActsAfterABootPassThatFailed(t *testing.T) {
	dir := release.NewDirStore(t.TempDir())
	rec, _ := queueRun(t, dir, time.Now().Add(time.Minute))
	var l execLog
	w := testWorker(listFails{dir}, runOwnedRoute(runHost), "box-1", "boot-1", &l)
	w.booted.Store(false)
	w.dataDir = t.TempDir()
	w.note(rec)
	w.pass(context.Background())
	if l.count() != 0 {
		t.Fatal("a pass acted before the boot pass")
	}
	w.bootPass(context.Background())
	w.pass(context.Background())
	if l.count() != 1 {
		t.Fatalf("execs %d, want 1", l.count())
	}
}

type listFails struct{ release.Store }

func (listFails) List(context.Context, string) ([]string, error) {
	return nil, errors.New("store down")
}