package activator

import (
	"context"
	"errors"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
)

// The import rollback (seeded-instances spec §5.6) against this package's
// real Service: a failed wake of the import's row must not leave the
// rollback stuck once its stop is known (seeded pre-land review, P2).
//
// The rollback's fence hook is internal/host's fenceForRollback, which
// this package cannot import (host imports it). rollbackFence below is
// that rule, step for step, with the startup sweep taken as done.

type importRollback struct {
	a        *Service
	b        *fakeBackend
	reg      *registry.Registry
	store    *release.DirStore
	cfg      provision.ReconcileConfig
	rec      provision.Record
	original registry.Route

	mu       sync.Mutex
	removed  int // RemoveLocal calls: the import's local database
	replicas int // DeleteReplica calls: the host's replica
	// beforeFence, when set, runs once as the rollback's fence hook is
	// entered: after step 1 closed the gate, before the fence.
	beforeFence func()
}

func newImportRollback(t *testing.T) *importRollback {
	t.Helper()
	ctx := context.Background()
	a, b, _, reg := testActivator(t, time.Hour)
	a.Routes = reg
	x := &importRollback{a: a, b: b, reg: reg, store: release.NewDirStore(t.TempDir())}

	sqid := domains.Sqid(2)
	manifest := `{"id":2,"sqid":"` + sqid + `","name":"Shop","created_at":"2026-08-02T09:00:00Z"}`
	if _, err := x.store.Put(ctx, "accounts/"+sqid+"/account.json", []byte(manifest), ""); err != nil {
		t.Fatal(err)
	}
	if err := provision.PutManifest(ctx, x.store, sqid, "shop", provision.Manifest{Schema: 1, App: "shop", Domain: "shop.example.com"}); err != nil {
		t.Fatal(err)
	}
	id, err := seeds.NewID()
	if err != nil {
		t.Fatal(err)
	}
	x.rec = provision.Record{Schema: 1, Host: "a.shop.example.com", Slug: "a", App: "shop", Account: sqid,
		Region: "ie", Channel: "stable", CreatedAt: time.Now(),
		Seed: &provision.SeedSource{ID: id, Kind: seeds.KindImport, State: seeds.StateReady,
			SHA256: strings.Repeat("a", 64), Bytes: 10, Replaces: "127.0.0.1:8401",
			VerifyHash: strings.Repeat("b", 64), By: "t", At: time.Now()}}
	if err := provision.CreateRecord(ctx, x.store, x.rec); err != nil {
		t.Fatal(err)
	}
	// The legacy row the import replaces: an addr route to the app's own
	// process, which the rollback must put back exactly.
	if err := reg.Upsert(registry.Route{Host: x.rec.Host, Kind: "service", Addr: "127.0.0.1:8401", App: "shop",
		Account: sqid, Region: "ie", Channel: "stable", Status: "ready", Unit: "shop@a.service"}); err != nil {
		t.Fatal(err)
	}
	x.original = *mustRoute(t, reg, x.rec.Host)

	data := release.NewDirStore(t.TempDir())
	if _, err := data.Put(ctx, seeds.SnapshotKey(x.rec.Host, id), []byte("db"), ""); err != nil {
		t.Fatal(err)
	}
	var gens uint64
	x.cfg = provision.ReconcileConfig{Region: "ie", DataDir: t.TempDir(),
		Layout: func(context.Context, release.Store) (string, error) { return release.LayoutModeNested, nil },
		Seeds: &provision.SeedHooks{
			Data: data,
			LegacyOwners: func(context.Context) (*seeds.LegacyOwners, error) {
				return &seeds.LegacyOwners{V: 1}, nil
			},
			EnvBefore:        func(string, string) (string, error) { return registry.EnvBeforeAbsent, nil },
			FenceForRollback: x.rollbackFence,
			LiftFence:        func(host string, gen uint64) { a.LiftFence(host, gen) },
			RemoveLocal: func(registry.Route) error {
				x.mu.Lock()
				x.removed++
				x.mu.Unlock()
				return nil
			},
			DeleteReplica: func(context.Context, string) error {
				x.mu.Lock()
				x.replicas++
				x.mu.Unlock()
				return nil
			},
			NextFenceGen: func() (uint64, error) {
				x.mu.Lock()
				defer x.mu.Unlock()
				gens++
				return gens, nil
			},
			LocalSeedID: func(string) (string, error) { return id, nil },
			RetireKey:   func(string, string, string) error { return nil },
		}}
	x.pass(t) // the swap
	if row := x.row(t); !registry.IsImportRow(*row, registry.ImportJournal{Host: x.rec.Host, SeedID: id}) {
		t.Fatalf("the swap did not happen: %+v", row)
	}
	return x
}

// rollbackFence is internal/host's fenceForRollback over this Service.
func (x *importRollback) rollbackFence(ctx context.Context, host string, gen uint64) (bool, error) {
	if f := x.beforeFence; f != nil {
		x.beforeFence = nil
		f()
	}
	if gen == 0 {
		return false, ErrFenceGeneration
	}
	if x.a.FenceGeneration(host) == gen {
		return true, nil
	}
	fctx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	_, err := x.a.FenceAndStopIfUnfenced(fctx, host, gen)
	switch {
	case err == nil:
		return true, nil
	case errors.Is(err, ErrFenceHeld):
		return false, err
	case errors.Is(err, ErrStopUnproven) && x.a.Quiesced(host) && !x.a.StopUncertain(host):
		return true, nil
	}
	x.a.LiftFence(host, gen)
	return false, err
}

func (x *importRollback) pass(t *testing.T) {
	t.Helper()
	if err := provision.Reconcile(context.Background(), x.store, x.reg, x.cfg); err != nil {
		t.Fatal(err)
	}
}

func (x *importRollback) row(t *testing.T) *registry.Route {
	t.Helper()
	return mustRoute(t, x.reg, x.rec.Host)
}

func (x *importRollback) abort(t *testing.T) {
	t.Helper()
	if _, err := provision.SetRecordSeed(context.Background(), x.store, x.rec.Account, x.rec.App, x.rec.Host,
		provision.SeedAbort(x.rec.Seed.ID)); err != nil {
		t.Fatal(err)
	}
}

// wake starts a request's wake of the import's row in the background and
// returns its outcome.
func (x *importRollback) wake(t *testing.T) <-chan error {
	t.Helper()
	rt := *x.row(t)
	done := make(chan error, 1)
	go func() {
		release, err := x.a.beginRequest(context.Background(), rt)
		if err == nil {
			release()
		}
		done <- err
	}()
	return done
}

// wantRestored: the original route back exactly, the import's local
// database and replica removed once each, and no fence left.
func (x *importRollback) wantRestored(t *testing.T) {
	t.Helper()
	if got := x.row(t); !reflect.DeepEqual(*got, x.original) {
		t.Fatalf("row = %+v, want the original %+v", got, x.original)
	}
	if j, err := x.reg.ImportJournal(x.rec.Host); err != nil || (j != nil && j.State != registry.JournalCleanupPending) {
		t.Fatalf("journal = %+v, %v", j, err)
	}
	x.mu.Lock()
	removed := x.removed
	x.mu.Unlock()
	if removed != 1 {
		t.Fatalf("the import's local database was removed %d times, want once", removed)
	}
	if x.a.Fenced(x.rec.Host) {
		t.Fatal("the rollback left its fence")
	}
}

// wantBlocked: nothing deleted, the import's row still in place with its
// gate closed, and the journal still rolling back.
func (x *importRollback) wantBlocked(t *testing.T) {
	t.Helper()
	row := x.row(t)
	if row.ImportGate != registry.ImportGateClosed {
		t.Fatalf("row = %+v, want the import's row with its gate closed", row)
	}
	if j, err := x.reg.ImportJournal(x.rec.Host); err != nil || j == nil || j.State != registry.JournalRollingBack {
		t.Fatalf("journal = %+v, %v; want rolling-back", j, err)
	}
	x.mu.Lock()
	removed, replicas := x.removed, x.replicas
	x.mu.Unlock()
	if removed != 0 || replicas != 0 {
		t.Fatalf("deleted under an unproven stop: local %d, replica %d", removed, replicas)
	}
}

func (x *importRollback) retained() bool {
	x.a.mu.Lock()
	defer x.a.mu.Unlock()
	_, ok := x.a.unprovenWakes[x.rec.Host]
	return ok
}

// reap runs the sweep's retry of retained failed wakes once and waits
// for it to finish.
func (x *importRollback) reap(t *testing.T) {
	t.Helper()
	x.a.mu.Lock()
	u := x.a.unprovenWakes[x.rec.Host]
	x.a.reapUnprovenWakesLocked()
	x.a.mu.Unlock()
	waitFor(t, "the reaper's retry to finish", func() bool {
		x.a.mu.Lock()
		defer x.a.mu.Unlock()
		return u == nil || !u.busy || x.a.unprovenWakes[x.rec.Host] != u
	})
}

// queueStart makes the next run-unit start one that systemd has queued:
// it blocks until its context ends and comes back marked as a start root
// may still carry out (#292).
func (x *importRollback) queueStart(t *testing.T) {
	t.Helper()
	queued := make(chan struct{})
	x.b.mu.Lock()
	x.b.startQueued = queued
	x.b.mu.Unlock()
	t.Cleanup(func() { close(queued) })
}

func mustRoute(t *testing.T, reg *registry.Registry, host string) *registry.Route {
	t.Helper()
	rt, err := reg.Get(host)
	if err != nil || rt == nil {
		t.Fatalf("route %s: %+v, %v", host, rt, err)
	}
	return rt
}

// An import aborted while its verify wake's run-unit start is queued:
// the rollback's fence cancels the start, #292's proof stops the unit at
// once, and the same pass restores the original route.
func TestImportRollbackAfterAnAbortedStartProvenAtOnce(t *testing.T) {
	t.Parallel()
	x := newImportRollback(t)
	x.queueStart(t)
	woke := x.wake(t)
	waitFor(t, "the verify wake to reach its queued start", func() bool { return x.b.startCount() == 1 })

	x.abort(t)
	x.pass(t)
	x.wantRestored(t)
	if err := <-woke; err == nil {
		t.Fatal("the aborted wake was served")
	}
	x.b.mu.Lock()
	stops := x.b.legacyStops
	x.b.mu.Unlock()
	if stops == 0 {
		t.Fatal("the queued start's unit was never stopped")
	}
}

// The same, with #292's proof failing at first: the rollback waits, and
// once the sweep's retry proves the unit gone, the next pass restores the
// original route, with no agent restart in between.
func TestImportRollbackAfterAnAbortedStartProvenLater(t *testing.T) {
	t.Parallel()
	x := newImportRollback(t)
	x.b.mu.Lock()
	x.b.legacyStopErr = errors.New("stop carlos-instance@a.shop.example.com: instance control socket: connection refused")
	x.b.mu.Unlock()
	x.queueStart(t)
	woke := x.wake(t)
	waitFor(t, "the verify wake to reach its queued start", func() bool { return x.b.startCount() == 1 })

	x.abort(t)
	x.pass(t)
	if err := <-woke; err == nil {
		t.Fatal("the aborted wake was served")
	}
	if !x.retained() {
		t.Fatal("the unproven start was not retained")
	}
	x.wantBlocked(t)

	x.b.mu.Lock()
	x.b.legacyStopErr = nil
	x.b.mu.Unlock()
	x.reap(t)
	if x.retained() {
		t.Fatal("the proven start is still retained")
	}
	x.pass(t)
	x.wantRestored(t)
}

// While #292's proof keeps failing, the rollback deletes nothing, however
// many passes and retries run.
func TestImportRollbackKeepsEverythingWhileTheStopIsUnproven(t *testing.T) {
	t.Parallel()
	x := newImportRollback(t)
	x.b.mu.Lock()
	x.b.legacyStopErr = errors.New("stop carlos-instance@a.shop.example.com: unit still active")
	x.b.mu.Unlock()
	x.queueStart(t)
	woke := x.wake(t)
	waitFor(t, "the verify wake to reach its queued start", func() bool { return x.b.startCount() == 1 })

	x.abort(t)
	x.pass(t)
	<-woke
	for range 3 {
		x.reap(t)
		x.pass(t)
	}
	if !x.retained() {
		t.Fatal("an unproven start was let go")
	}
	x.wantBlocked(t)
	if !x.a.StopUncertain(x.rec.Host) {
		t.Fatal("the unproven stop reads as certain")
	}
}

// The rollback's fence lands while the verify wake's build record is
// being written (amadan #272): the recheck after the write refuses the
// spawn before root is asked, so nothing started, and the same pass
// restores the original route.
func TestImportRollbackFencedDuringThePendingRecord(t *testing.T) {
	x := newImportRollback(t)
	x.b.setIdentity("v1")
	x.abort(t)
	rolled := make(chan error, 1)
	var once sync.Once
	x.a.BeforeExec = func(ctx context.Context, host, build string) error {
		once.Do(func() {
			// The rollback's pass runs now, while this write is in flight,
			// and fences the host before the write returns.
			go func() { rolled <- provision.Reconcile(context.Background(), x.store, x.reg, x.cfg) }()
			for deadline := time.Now().Add(5 * time.Second); !x.a.Fenced(host) && time.Now().Before(deadline); {
				time.Sleep(time.Millisecond)
			}
		})
		return nil
	}
	woke := x.wake(t)
	if err := <-woke; !errors.Is(err, ErrFenced) {
		t.Fatalf("wake = %v, want ErrFenced", err)
	}
	if err := <-rolled; err != nil {
		t.Fatal(err)
	}
	if x.b.startCount() != 0 {
		t.Fatalf("started %d after the fence", x.b.startCount())
	}
	x.wantRestored(t)
}

// A launch its pre-exec gate refused is known not started; a start that
// may have run is not, and an unrelated earlier uncertainty is kept.
func TestALaunchRefusedBeforeExecIsNotUncertain(t *testing.T) {
	refused := runExecGate(withExecGate(context.Background(), func() error { return ErrFenced }))
	if !errors.Is(refused, ErrFenced) || !launchNotStarted(refused) {
		t.Fatalf("a gate refusal = %v, not started %v", refused, launchNotStarted(refused))
	}
	if launchNotStarted(markLegacyStartUncertain(refused)) || launchNotStarted(markStartUncertain(refused)) {
		t.Fatal("a start that may have run reads as not started")
	}
	if launchNotStarted(errors.New("start unit: root did not answer")) {
		t.Fatal("an unmarked failure reads as not started")
	}

	a, _, _, reg := testActivator(t, time.Hour)
	defer reg.Close()
	var known stopObservation
	known.notStarted()
	a.mu.Lock()
	a.rememberStopLocked("fresh.example.com", &known)
	a.rememberStopLocked("other.example.com", &stopObservation{uncertain: true})
	a.rememberStopLocked("other.example.com", &known)
	a.mu.Unlock()
	if a.StopUncertain("fresh.example.com") {
		t.Fatal("a launch known not started reads as uncertain")
	}
	if !a.StopUncertain("other.example.com") {
		t.Fatal("a later known stop erased an unrelated uncertainty")
	}
}

// A retained start's proof discharges only its own uncertainty: one that
// another source added to stays.
func TestTheLegacyProofDischargesOnlyItsOwnUncertainty(t *testing.T) {
	t.Parallel()
	for _, other := range []bool{false, true} {
		a, b, _, reg := testActivator(t, time.Hour)
		host := "own.example.com"
		u := &unprovenWake{rt: testRoute(host), legacy: true}
		a.mu.Lock()
		if other {
			a.rememberStopLocked(host, &stopObservation{uncertain: true})
		}
		a.rememberStopLocked(host, &stopObservation{uncertain: true, owedBy: u})
		var refused stopObservation
		refused.notStarted() // a wake refused before it started anything
		a.rememberStopLocked(host, &refused)
		a.unprovenWakes = map[string]*unprovenWake{host: u}
		a.mu.Unlock()
		a.reapUnprovenWake(host, u)
		b.mu.Lock()
		stops := b.legacyStops
		b.mu.Unlock()
		if stops != 1 {
			t.Fatalf("other %v: %d proofs, want 1", other, stops)
		}
		if got := a.StopUncertain(host); got != other {
			t.Fatalf("other %v: uncertain after the proof = %v, want %v", other, got, other)
		}
		reg.Close()
	}
}

// The not-started record is only for a wake that started nothing: one
// that holds a process, a replica or a sidecar of its own does not get it.
func TestOnlyAWakeThatStartedNothingIsRecordedNotStarted(t *testing.T) {
	if !wakeStartedNothing(&wokenInstance{}) {
		t.Fatal("an empty wake reads as having started something")
	}
	for name, w := range map[string]*wokenInstance{
		"process": {proc: &maintenanceProc{}},
		"replica": {rep: &maintenanceRep{}},
		"sidecar": {sidecar: &maintenanceProc{}},
	} {
		if wakeStartedNothing(w) {
			t.Fatalf("a wake holding a %s reads as having started nothing", name)
		}
	}
}

// A refusal at a gate later in the wake than the main spawn's, with the
// main process and its replica already running: those stages still go
// through the checked stop and are never marked held by the refusal.
// The sidecar's start (step 6) is the wake's last gate; a replica that
// opens and then fails is the stage before it.
func TestALateRefusalMarksNothingHeld(t *testing.T) {
	t.Parallel()
	refused := launchRefused{ErrFenced}

	t.Run("the sidecar's gate", func(t *testing.T) {
		a, base, _, reg := testActivator(t, time.Hour)
		defer reg.Close()
		base.sidecarStaged = true
		main, partial, rep := &maintenanceProc{}, &maintenanceProc{}, &maintenanceRep{}
		a.backend = &maintenanceBackend{fakeBackend: base, main: main, side: partial, rep: rep, sideFailure: refused}
		rt := testRoute("late-sidecar.example")
		release, err := a.beginRequest(t.Context(), rt)
		if err != nil {
			t.Fatal(err)
		}
		release()
		if !a.StopUncertain(rt.Host) {
			t.Fatal("a running wake reads as stopped")
		}
		if main.calls.Load() != 0 || rep.calls.Load() != 0 {
			t.Fatal("the refusal stopped the main process or closed the replica")
		}
		if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), rt.Host, 1); err != ErrStopUnproven || proof != nil {
			t.Fatal("a refused sidecar's returned handle yielded trusted evidence", err)
		}
		if main.calls.Load() != 1 || rep.calls.Load() != 1 || partial.calls.Load() != 1 {
			t.Fatalf("the stop skipped a stage: main %d, replica %d, sidecar %d", main.calls.Load(), rep.calls.Load(), partial.calls.Load())
		}
	})

	t.Run("after the replica opened", func(t *testing.T) {
		a, base, _, reg := testActivator(t, time.Hour)
		defer reg.Close()
		main, rep := &maintenanceProc{}, &maintenanceRep{}
		a.backend = &maintenanceBackend{fakeBackend: base, main: main, rep: rep, replicateFailure: refused}
		rt := testRoute("late-replica.example")
		if _, err := a.beginRequest(t.Context(), rt); !errors.Is(err, ErrFenced) {
			t.Fatalf("wake = %v, want the refusal", err)
		}
		if main.calls.Load() != 1 || !main.Exited() || rep.calls.Load() != 1 {
			t.Fatalf("the failed wake's cleanup skipped a stage: main %d, replica %d", main.calls.Load(), rep.calls.Load())
		}
		if !a.StopUncertain(rt.Host) {
			t.Fatal("a refusal after the spawn marked the wake's stages held")
		}
		if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), rt.Host, 1); err != ErrStopUnproven || proof != nil {
			t.Fatal("a late refusal yielded trusted evidence", err)
		}
	})
}

// Astra's re-review, P2: the rollback has closed the import's gate but not
// yet fenced the host, and a verify request the edge admitted on its
// cached row reaches the activator. The wake re-reads the row, finds the
// gate changed and refuses before its lease. That refusal started nothing:
// it must leave the host's stop history as it was, and the rollback
// restores the original route in the same pass, with no restart.
func TestImportRollbackAfterACachedRowWakeRefusedBeforeItsLease(t *testing.T) {
	t.Parallel()
	for _, history := range []string{"none", "a clean stop"} {
		t.Run(history, func(t *testing.T) {
			x := newImportRollback(t)
			cached := *x.row(t) // the edge's copy, gated by the verify hash
			if history == "a clean stop" {
				arrangeAwake(t, x.a, cached)
				x.a.mu.Lock()
				e := x.a.entries[cached.Host]
				x.a.beginStopLocked(e, StopIdle)
				stopped := e.stopped
				x.a.mu.Unlock()
				<-stopped
				if x.a.StopUncertain(cached.Host) {
					t.Fatal("the verify wake's clean stop reads as uncertain")
				}
			}
			starts := x.b.startCount()
			x.abort(t)
			var refused error
			x.beforeFence = func() {
				release, err := x.a.beginRequest(context.Background(), cached)
				if err == nil {
					release()
				}
				refused = err
			}
			x.pass(t)
			if refused == nil {
				t.Fatal("a wake admitted on the cached row was served after the gate closed")
			}
			if x.b.startCount() != starts {
				t.Fatal("the refused wake started the instance")
			}
			x.wantRestored(t)
		})
	}
}

// A refusal before anything of a wake started leaves every earlier record
// exactly as it was: an uncertain one with its owner, and an incomplete
// one.
func TestAWakeThatStartedNothingLeavesTheHistoryAsItWas(t *testing.T) {
	a, _, _, reg := testActivator(t, time.Hour)
	defer reg.Close()
	u := &unprovenWake{legacy: true}
	priors := map[string]*stopObservation{
		"owed.example.com":       {uncertain: true, owedBy: u},
		"incomplete.example.com": {mainStop: true, mainExited: true, sidecarStop: true},
	}
	var refused stopObservation
	refused.notStarted()
	a.mu.Lock()
	for host, prior := range priors {
		a.rememberStopLocked(host, prior)
		a.rememberStopLocked(host, &refused)
	}
	a.rememberStopLocked("new.example.com", &refused)
	for host, prior := range priors {
		if a.stopHistory[host] != prior {
			t.Fatalf("%s: the refusal replaced %+v with %+v", host, prior, a.stopHistory[host])
		}
	}
	a.mu.Unlock()
	if a.StopUncertain("new.example.com") {
		t.Fatal("a host whose only wake started nothing reads as uncertain")
	}
}