//go:build unix

package host

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

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// failingClaimPut is a control bucket whose next fail writes of one claim
// key fail: a lost or failed claim CAS.
type failingClaimPut struct {
	release.Store
	key  string
	fail int
}

func (s *failingClaimPut) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key == s.key && s.fail > 0 {
		s.fail--
		return "", errors.New("the claim CAS failed")
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

// finishWorld is a host that serves on dst after relocation rel-1
// completed there, undischarged: its row held for rel-1, the probe stop's
// owned fence on the activator, and the pass of dst with the production
// Finish (relocationReleaseHold).
type finishWorld struct {
	t          *testing.T
	store      *failingClaimPut
	reg        *registry.Registry
	act        *stopProbeActivator
	order      []string
	discharged int
	discharge  error
	runtime    *RelocationRuntime
	dataDir    string
	assembly   *ClaimAssembly
}

const finishHost = "shop.example"

func newFinishWorld(t *testing.T) *finishWorld {
	t.Helper()
	ctx := t.Context()
	w := &finishWorld{t: t, dataDir: t.TempDir()}
	created := time.Date(2026, 10, 7, 0, 0, 0, 0, time.UTC)
	inc, err := hostclaim.Incarnation("acct", "app", finishHost, created)
	if err != nil {
		t.Fatal(err)
	}
	done, _ := json.Marshal(hostclaim.CompletedRelocation{Op: "rel-1", Stop: hostclaim.ProbeStop{Box: "dst", Session: 12, Receipt: "fence-1", ReceiptSHA256: strings.Repeat("a", 64)}, Stamp: 9, Phase: hostclaim.PhaseRelocatingProbing, Generation: "P1"})
	c := &hostclaim.Claim{V: 2, Gen: 31, Host: finishHost,
		Holder: &hostclaim.Holder{Family: "instance", Key: "apps/acct/app/instances/" + finishHost + ".json", Account: "acct", App: "app", Incarnation: inc, CreatedAt: created,
			Location: "instances/" + finishHost, Stamp: 9, Session: 12, Generation: "P1", WriterBox: "dst", Admissions: []hostclaim.AdmissionRecord{{Session: 12, Box: "dst", Generation: "P1", Mode: "cold", Region: "ie"}}},
		Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "dst", Participants: []string{"dst"}, Journal: map[string]string{hostclaim.JournalCompleted: string(done)}}}
	key, _ := hostclaim.Key(finishHost)
	w.store = &failingClaimPut{Store: release.NewDirStore(t.TempDir()), key: key}
	raw, _ := json.Marshal(c)
	if _, err := w.store.Put(ctx, key, raw, ""); err != nil {
		t.Fatal(err)
	}
	if w.reg, err = registry.Open(filepath.Join(t.TempDir(), "registry.db")); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { w.reg.Close() })
	if err := w.reg.Upsert(registry.Route{Host: finishHost, Status: "relocating", HoldOp: "rel-1", Socket: "/run/shop.sock"}); err != nil {
		t.Fatal(err)
	}
	marker, err := relocationProbeFencePath(w.dataDir, "rel-1")
	if err != nil {
		t.Fatal(err)
	}
	owner, err := relocationProbeFenceOwner(marker, finishHost, 12)
	if err != nil {
		t.Fatal(err)
	}
	w.act = newStopProbeActivator(&w.order)
	fctx, cancel := context.WithTimeout(ctx, time.Second)
	defer cancel()
	if _, err := w.act.svc.FenceAndStopAs(fctx, finishHost, 12, owner); err != nil && !errors.Is(err, activator.ErrStopUnproven) {
		t.Fatal(err)
	}
	a := ClaimAssembly{Claims: w.store, Guard: hostguard.Guard{DataDir: w.dataDir}, stopLocal: func(context.Context) (claimStopLocal, error) {
		return claimStopLocal{Actor: hostbox.Actor{Box: "dst"}}, nil
	}}
	w.assembly = &a
	cfg := RelocationAssemblyConfig{Registry: w.reg, Activator: w.act, DataDir: w.dataDir, Discharge: func(context.Context, hostclaim.Observed) error {
		w.discharged++
		return w.discharge
	}}
	var steps []string
	w.runtime = &RelocationRuntime{Worker: stepRecorder{name: "worker", order: &steps}, Participant: stepRecorder{name: "participant", order: &steps}, Box: "dst",
		Finish: a.relocationReleaseHold(cfg), DataDir: w.dataDir}
	return w
}

func (w *finishWorld) pass() { w.runtime.once(w.t.Context(), w.store) }

func (w *finishWorld) held() bool {
	row, err := w.reg.Get(finishHost)
	return err == nil && row != nil && row.Status == "relocating" && row.HoldOp == "rel-1"
}

// marked reads the record, and whether a successor (a move from dst) may
// start: the completion barrier.
func (w *finishWorld) marked() (bool, error) {
	obs, err := hostclaim.Read(w.t.Context(), w.store, finishHost)
	if err != nil {
		w.t.Fatal(err)
	}
	done, ok, err := hostclaim.ReadCompletedRelocation(obs.Claim)
	if err != nil {
		w.t.Fatal(err)
	}
	return ok && done.Discharged, hostclaim.StartGuard(obs.Claim, []string{"uk"}, []string{"spa"}, "")
}

// done asserts the finished state: discharged as many times in all as
// the passes asked (Task 9 review M2: once the hold is released, a replay
// discharges again rather than inferring the discharge ran), the hold
// released, no fence, the completion marked so a successor may start, and
// the op's journals swept by the same pass.
func (w *finishWorld) done(discharges int) {
	w.t.Helper()
	ok, barrier := w.marked()
	if w.discharged != discharges || w.held() || w.act.svc.Fenced(finishHost) || !ok || barrier != nil {
		w.t.Fatalf("discharged %d (want %d) held=%v fenced=%v marked=%v barrier=%v", w.discharged, discharges, w.held(), w.act.svc.Fenced(finishHost), ok, barrier)
	}
	if _, err := os.Stat(filepath.Join(w.dataDir, "relocate", "ops", "rel-1")); !errors.Is(err, os.ErrNotExist) {
		w.t.Fatalf("the finished op's journals survived: %v", err)
	}
}

// Astra round 7, finding 1 (ruling 80): a completion is finished from the
// claim's record on every pass, whatever the row says, and every step
// resumes after a failure or a crash at any point, through the pass.
func TestRelocatePassFinishesACompletionFromItsRecord(t *testing.T) {
	t.Parallel()
	t.Run("in one pass", func(t *testing.T) {
		w := newFinishWorld(t)
		w.pass()
		w.done(1)
		w.pass()
		w.done(1)
	})
	t.Run("a failed discharge keeps everything, the next pass finishes", func(t *testing.T) {
		w := newFinishWorld(t)
		w.discharge = errors.New("the participation slot is still open")
		w.pass()
		if ok, barrier := w.marked(); w.discharged != 1 || !w.held() || w.act.svc.FenceGeneration(finishHost) != 12 || ok || !errors.Is(barrier, hostclaim.ErrPendingChore) {
			t.Fatalf("discharged %d held=%v fence=%d marked=%v barrier=%v", w.discharged, w.held(), w.act.svc.FenceGeneration(finishHost), ok, barrier)
		}
		w.discharge = nil
		w.pass()
		w.done(2)
	})
	t.Run("the mark's CAS fails, the next pass marks", func(t *testing.T) {
		w := newFinishWorld(t)
		w.store.fail = 1
		w.pass()
		if ok, barrier := w.marked(); w.discharged != 1 || w.held() || w.act.svc.Fenced(finishHost) || ok || !errors.Is(barrier, hostclaim.ErrPendingChore) {
			t.Fatalf("discharged %d held=%v fenced=%v marked=%v barrier=%v", w.discharged, w.held(), w.act.svc.Fenced(finishHost), ok, barrier)
		}
		// The row is no longer held: the record alone brings it back, and
		// the released-hold branch discharges again (Task 9 review M2).
		w.pass()
		w.done(2)
	})
	t.Run("the lift never ran after the hold went, the next pass lifts", func(t *testing.T) {
		w := newFinishWorld(t)
		// An earlier call discharged and released the hold, then failed.
		w.discharged = 1
		if _, err := w.reg.ReleaseHold(finishHost, "rel-1"); err != nil {
			t.Fatal(err)
		}
		w.pass()
		w.done(2)
	})
	t.Run("a restart after the hold release", func(t *testing.T) {
		w := newFinishWorld(t)
		w.discharged = 1
		if _, err := w.reg.ReleaseHold(finishHost, "rel-1"); err != nil {
			t.Fatal(err)
		}
		w.act.svc = &activator.Service{} // the restart: the fence went with the process
		w.pass()
		w.done(2)
	})
}

// productionLocal is main's participation owner after the restart, bound
// to the world's assembly: DischargeCompleted is the production discharge
// (carry 6), its identity a verifying one for dst.
func (w *finishWorld) productionLocal() *productionClaimLocal {
	identity := handoverRecoveryIdentity(w.t, hostbox.Actor{Box: "dst", Enrollment: "enr", Region: "ie", Revision: 1, Binding: "1"})
	return &productionClaimLocal{legacyParticipation: true, Resources: ClaimRuntimeResources{Mode: "agent", DataDir: w.dataDir}, Identity: identity, Store: w.store,
		slots: map[string]*claimParticipation{}, assembly: w.assembly, lifetime: w.t.Context()}
}

// Astra round 8 (ruling 82): A to B completes, the mark's CAS fails, and B
// restarts with CARLOS_REGION_HA off. Claim mode alone builds the
// completion-only runtime, which steps nothing, and its pass still lifts
// and marks, so the barrier opens. The restarted runtime's discharge is
// main's own (productionClaimLocal.DischargeCompleted, carry 6), which
// finds no slot and no .hostexec record.
func TestCompletionPassFinishesWithRegionHAOff(t *testing.T) {
	w := newFinishWorld(t)
	w.store.fail = 1
	w.pass()
	if ok, barrier := w.marked(); ok || w.held() || !errors.Is(barrier, hostclaim.ErrPendingChore) {
		t.Fatalf("the first pass marked or kept the hold: marked=%v held=%v barrier=%v", ok, w.held(), barrier)
	}
	w.act.svc = &activator.Service{} // the restart
	claimOnly := func(k string) string {
		if k == "CARLOS_HOST_CLAIM" {
			return "1"
		}
		return ""
	}
	r, err := NewRelocationRuntime(t.Context(), claimOnly, w.store, w.productionLocal(), ClaimProviders{}, w.reg, w.act, relocationRuntimeInputs{})
	if err != nil || r == nil || r.Worker != nil || r.Participant != nil || r.Convert != nil || r.Finish == nil {
		t.Fatalf("not a completion-only runtime: %+v %v", r, err)
	}
	w.runtime = r
	w.pass()
	w.done(1)
}

// Carry 6: a probe-fence owner the hold release cannot find is ErrRefused,
// which the pass alerts and retries; it never skips the completion and
// never marks it. Nothing is released while the owner is missing; once it
// is back, the next pass finishes.
func TestRelocatePassRetriesAMissingProbeFenceOwner(t *testing.T) {
	w := newFinishWorld(t)
	marker, err := relocationProbeFencePath(w.dataDir, "rel-1")
	if err != nil {
		t.Fatal(err)
	}
	raw, err := os.ReadFile(marker)
	if err != nil {
		t.Fatal(err)
	}
	if err := os.Remove(marker); err != nil {
		t.Fatal(err)
	}
	buf := captureRacyLog(t)
	w.pass()
	w.pass()
	if ok, barrier := w.marked(); ok || !w.held() || w.discharged != 0 || w.act.svc.FenceGeneration(finishHost) != 12 || !errors.Is(barrier, hostclaim.ErrPendingChore) {
		t.Fatalf("over a missing owner: marked=%v held=%v discharged=%d fence=%d barrier=%v", ok, w.held(), w.discharged, w.act.svc.FenceGeneration(finishHost), barrier)
	}
	if n := strings.Count(buf.String(), "has no probe-fence owner"); n != 1 {
		t.Fatalf("alerted %d times, want once:\n%s", n, buf.String())
	}
	// An operator puts the owner back. (The sweep has removed the op's
	// directory meanwhile: with the marker gone it held no journal.)
	if err := os.MkdirAll(filepath.Dir(marker), 0o700); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(marker, raw, 0o600); err != nil {
		t.Fatal(err)
	}
	w.pass()
	w.done(1)
}

// Review M1: a completion whose row is gone is still visited by the
// completion-only pass (its probe-fence journal names the host), and the
// hold release's refusal is said once; nothing is released or marked.
func TestCompletionPassSaysACompletionWhoseRowIsGone(t *testing.T) {
	w := newFinishWorld(t)
	if err := w.reg.Delete(finishHost); err != nil {
		t.Fatal(err)
	}
	claimOnly := func(k string) string {
		if k == "CARLOS_HOST_CLAIM" {
			return "1"
		}
		return ""
	}
	r, err := NewRelocationRuntime(t.Context(), claimOnly, w.store, w.productionLocal(), ClaimProviders{}, w.reg, w.act, relocationRuntimeInputs{})
	if err != nil {
		t.Fatal(err)
	}
	w.runtime = r
	buf := captureRacyLog(t)
	w.pass()
	w.pass()
	if n := strings.Count(buf.String(), "has no row on this box"); n != 1 {
		t.Fatalf("said %d times, want once:\n%s", n, buf.String())
	}
	if ok, _ := w.marked(); ok || !w.act.svc.Fenced(finishHost) {
		t.Fatalf("finished over a missing row: marked=%v fenced=%v", ok, w.act.svc.Fenced(finishHost))
	}
}