package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/superfly/ltx"
)

// plan9D5Sync stands in for the session's replicator. run sees each call's
// ordinal and does what litestream's sync would have uploaded by then.
type plan9D5Sync struct {
	mu    sync.Mutex
	calls int
	run   func(n int) error
}

func (s *plan9D5Sync) SyncAndWait(context.Context) error {
	s.mu.Lock()
	s.calls++
	n, run := s.calls, s.run
	s.mu.Unlock()
	if run != nil {
		return run(n)
	}
	return nil
}

func (s *plan9D5Sync) count() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.calls
}

// plan9D5Pub is one admitted attempt's publisher: the real assembly
// publish, the claim's serving owners and a real producer registry.
type plan9D5Pub struct {
	f         *plan9D5Serving
	w         *ClaimWakeAssembly
	a         *claimWakeAttempt
	producers *ClaimProducers
	target    hostguard.FenceTarget
	syncer    *plan9D5Sync
	work      plan9D5Work
	mu        sync.Mutex
	current   hostclaim.Observed
	published []hostclaim.Observed
}

func plan9D5Publisher(t *testing.T, f *plan9D5Serving, admitted hostclaim.Observed) *plan9D5Pub {
	t.Helper()
	h := admitted.Claim.Holder
	target := hostguard.FenceTarget{Host: admitted.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Box: "box-1", Region: "ie", Enrollment: "enrollment-1", BootID: "boot-1", ControllerID: "controller-1", ClaimETag: admitted.ETag}
	producers, err := NewClaimProducers(target, "start-1", func(context.Context, hostguard.FenceTarget, string) error { return nil })
	if err != nil {
		t.Fatal(err)
	}
	live, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	p := &plan9D5Pub{f: f, producers: producers, target: target, syncer: &plan9D5Sync{}, work: plan9D5Work{identities: &[]hostwrite.Identity{}}, current: admitted}
	p.w = &ClaimWakeAssembly{Publication: ClaimOwnerDispatch{Serving: f.owners.owners()}, Producers: func(context.Context, hostclaim.Observed) (*ClaimProducers, error) { return producers, nil }}
	p.a = &claimWakeAttempt{observed: admitted, held: recoveryHeld{}, ctx: live, cancel: cancel, done: make(chan struct{}), launch: &activator.RoleLaunch{Target: target}, syncer: p.syncer}
	return p
}

func (p *plan9D5Pub) read(context.Context, string) (hostclaim.Observed, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.current, nil
}

func (p *plan9D5Pub) markPublished(_ context.Context, o hostclaim.Observed) (hostclaim.Observed, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.published = append(p.published, o)
	return o, nil
}

func (p *plan9D5Pub) publishedCount() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return len(p.published)
}

func (p *plan9D5Pub) publish(ctx context.Context, in ClaimLaunchInputs) (hostclaim.Observed, error) {
	selected := claimLeaseWork{Work: p.work, Alive: p.a.ctx, Check: p.a.leaseErr}
	admitted, _ := p.read(ctx, "")
	return p.w.publish(ctx, p.a, admitted, admitted, in, selected, p.read, p.markPublished)
}

// drain is root's fence of the attempt's launch tuple.
func (p *plan9D5Pub) drain(ctx context.Context) (DrainAcknowledgement, error) {
	t := p.target
	return p.producers.DrainClaim(ctx, StopRequest{Host: t.Host, Incarnation: t.Incarnation, InstallationStamp: t.Stamp, WriterSession: t.Session, Box: t.Box, Enrollment: t.Enrollment, BootID: t.BootID, ControllerID: t.ControllerID}, "fence-1")
}

func (p *plan9D5Pub) outcome() (*ClaimPublicationReceipt, error) {
	p.a.mu.Lock()
	defer p.a.mu.Unlock()
	return p.a.receipt, p.a.pubErr
}

// plan9D5RootBody is a level-0 [1, 1] file, the first upload's shape.
func plan9D5RootBody(t *testing.T) []byte {
	t.Helper()
	h := ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, MinTXID: 1, MaxTXID: 1}
	body, err := h.MarshalBinary()
	if err != nil {
		t.Fatal(err)
	}
	return body
}

func plan9D5RootKey(gen string) string {
	return fmt.Sprintf("instances/%s/g-%s/0000/%s", recoveryHost, gen, ltx.FormatFilename(1, 1))
}

const plan9D5MarkerKey = "instances/" + recoveryHost + "/generation.json"

// plan9D5Fresh is a serving session admitted cold into g1 from g0's baseline.
func plan9D5Fresh(t *testing.T) (*plan9D5Serving, hostclaim.Observed, ClaimLaunchInputs) {
	t.Helper()
	f := plan9D5Fixture(t)
	f.file(t, "g0", 9, 1, 11)
	f.marker(t, "g0")
	admitted := plan9D5Observed(hostclaim.Holder{Stamp: 7, Session: 7, Generation: "g1", WriterBox: "box-1",
		Admissions: []hostclaim.AdmissionRecord{{Session: 7, Box: "box-1", Generation: "g1", Predecessor: "g0", Region: "ie", PredecessorRegion: "ie", Mode: "cold"}}})
	return f, admitted, ClaimLaunchInputs{Identity: f.launch.Identity, Admission: hostclaim.Admission{NewGeneration: "g1"}}
}

// The review's P1-3: replication opens before the session's first upload
// lands. Publication syncs the current replicator first and names only the
// root that sync acknowledged.
func TestPlan9D5PublisherSyncsTheSessionRootBeforeTheMarker(t *testing.T) {
	f, admitted, in := plan9D5Fresh(t)
	ctx := t.Context()
	root := plan9D5RootBody(t)
	// Without the sync the generation has no root yet: the marker is refused.
	if _, err := f.owners.PublishFenced(ctx, admitted, in, ClaimPublication{Sync: func(context.Context) error { return nil }, Check: func(context.Context) error { return nil }}); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("a marker before the session's first upload: %v", err)
	}
	p := plan9D5Publisher(t, f, admitted)
	p.syncer.run = func(n int) error {
		if n == 1 {
			f.bucket.Put("data", plan9D5RootKey("g1"), root)
		}
		return nil
	}
	if _, err := p.publish(ctx, in); err != nil {
		t.Fatal(err)
	}
	receipt, err := p.outcome()
	if err != nil || receipt == nil || receipt.Root.Key != plan9D5RootKey("g1") || receipt.Root.VersionID != f.bucket.LatestID("data", plan9D5RootKey("g1")) || receipt.MarkerVersion != f.bucket.LatestID("data", plan9D5MarkerKey) {
		t.Fatalf("receipt %+v %v", receipt, err)
	}
	if p.syncer.count() != 1 || p.publishedCount() != 1 {
		t.Fatalf("syncs=%d published=%d", p.syncer.count(), p.publishedCount())
	}
	body, _ := f.bucket.Current("data", plan9D5MarkerKey)
	var m genselect.Marker
	if json.Unmarshal(body, &m) != nil || m.Op != "g1" || m.Box != "box-1" || m.SnapshotTXID != 0 {
		t.Fatalf("marker %s", body)
	}
	if err := f.owners.publicationEvidence(ctx, admitted); err != nil {
		t.Fatal(err)
	}
	select {
	case <-p.a.publisher:
	default:
		t.Fatal("the publisher did not join")
	}
	// The publisher is a producer of the launch: root's drain names it.
	ack, err := p.drain(ctx)
	if err != nil || !slices.Contains(ack.ProducerIDs, "publisher") {
		t.Fatalf("drain %+v %v", ack, err)
	}
	// The evidence is bound to the exact root version the marker named: a
	// late duplicate of the upload does not replace it, a removed root does.
	f.bucket.Put("data", plan9D5RootKey("g1"), root)
	if err := f.owners.publicationEvidence(ctx, admitted); err != nil {
		t.Fatalf("a late duplicate root: %v", err)
	}
	f.bucket.BareDelete("data", plan9D5RootKey("g1"))
	if err := f.owners.publicationEvidence(ctx, admitted); err == nil {
		t.Fatal("evidence for a root that is gone")
	}
}

func TestPlan9D5PublisherSyncFailureWritesNoMarker(t *testing.T) {
	f, admitted, in := plan9D5Fresh(t)
	ctx := t.Context()
	f.bucket.Put("data", plan9D5RootKey("g1"), plan9D5RootBody(t))
	p := plan9D5Publisher(t, f, admitted)
	failure := errors.New("replica sync failed")
	p.syncer.run = func(int) error { return failure }
	before := f.bucket.LatestID("data", plan9D5MarkerKey)
	if _, err := p.publish(ctx, in); !errors.Is(err, failure) {
		t.Fatalf("publish after a failed sync: %v", err)
	}
	if f.bucket.LatestID("data", plan9D5MarkerKey) != before || p.publishedCount() != 0 {
		t.Fatal("a marker or publication without an acknowledged sync")
	}
	if receipt, err := p.outcome(); receipt != nil || !errors.Is(err, failure) {
		t.Fatalf("outcome %+v %v", receipt, err)
	}
	// One publisher per attempt: a second call never re-registers.
	if _, err := p.publish(ctx, in); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("second publisher: %v", err)
	}
}

// plan9D5HeldMarkers holds a marker request in flight until released.
type plan9D5HeldMarkers struct {
	plan9D5Markers
	entered, release chan struct{}
}

func (m plan9D5HeldMarkers) PutMarker(ctx context.Context, bucket, key string, body []byte, expected string, check func(context.Context) error) (string, error) {
	close(m.entered)
	<-m.release
	return m.plan9D5Markers.PutMarker(ctx, bucket, key, body, expected, check)
}

// The review's P2-1 (spec model 29): a fenced session whose publication
// listed the marker before its successor wrote one must not overwrite it.
// Root's drain cancels the publisher, waits for its request to return, and
// the store's check refuses the write.
func TestPlan9D5PublisherCancelledByRootFenceWritesNoStaleMarker(t *testing.T) {
	t.Parallel()
	f, admitted, in := plan9D5Fresh(t)
	ctx := t.Context()
	f.bucket.Put("data", plan9D5RootKey("g1"), plan9D5RootBody(t))
	held := plan9D5HeldMarkers{plan9D5Markers{f.bucket}, make(chan struct{}), make(chan struct{})}
	f.owners.Sources.Markers = held
	p := plan9D5Publisher(t, f, admitted)
	published := make(chan error, 1)
	go func() {
		_, err := p.publish(ctx, in)
		published <- err
	}()
	<-held.entered
	drained := make(chan error, 1)
	var ack DrainAcknowledgement
	go func() {
		var err error
		ack, err = p.drain(ctx)
		drained <- err
	}()
	select {
	case err := <-drained:
		t.Fatalf("the drain acknowledged a publisher with a request in flight: %v", err)
	case <-time.After(100 * time.Millisecond):
	}
	// The successor's marker lands after the fence.
	f.marker(t, "g2")
	successor := f.bucket.LatestID("data", plan9D5MarkerKey)
	close(held.release)
	if err := <-published; !errors.Is(err, context.Canceled) {
		t.Fatalf("fenced publication: %v", err)
	}
	if err := <-drained; err != nil || !slices.Contains(ack.ProducerIDs, "publisher") {
		t.Fatalf("drain %+v %v", ack, err)
	}
	if f.bucket.LatestID("data", plan9D5MarkerKey) != successor || p.publishedCount() != 0 {
		t.Fatal("the fenced publisher overwrote its successor's marker")
	}
	if receipt, _ := p.outcome(); receipt != nil {
		t.Fatalf("receipt for a fenced publication: %+v", receipt)
	}
	if err := f.owners.publicationEvidence(ctx, admitted); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("evidence for a fenced publication: %v", err)
	}
}

// A drained producer set refuses a publisher that has not started yet.
func TestPlan9D5PublisherRefusedAfterTheFence(t *testing.T) {
	f, admitted, in := plan9D5Fresh(t)
	ctx := t.Context()
	f.bucket.Put("data", plan9D5RootKey("g1"), plan9D5RootBody(t))
	p := plan9D5Publisher(t, f, admitted)
	if _, err := p.drain(ctx); err != nil {
		t.Fatal(err)
	}
	before := f.bucket.LatestID("data", plan9D5MarkerKey)
	if _, err := p.publish(ctx, in); !errors.Is(err, ErrClaimStopIdentity) || p.syncer.count() != 0 {
		t.Fatalf("publisher after the fence: %v syncs=%d", err, p.syncer.count())
	}
	if f.bucket.LatestID("data", plan9D5MarkerKey) != before {
		t.Fatal("a marker after the fence")
	}
}

// The selection check runs immediately before the marker request: a claim
// that no longer names the session writes nothing.
func TestPlan9D5PublisherChecksTheSelectionBeforeTheMarker(t *testing.T) {
	f, admitted, in := plan9D5Fresh(t)
	ctx := t.Context()
	f.bucket.Put("data", plan9D5RootKey("g1"), plan9D5RootBody(t))
	p := plan9D5Publisher(t, f, admitted)
	p.work.err = hostbox.ErrChanged
	before := f.bucket.LatestID("data", plan9D5MarkerKey)
	if _, err := p.publish(ctx, in); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("publication under a changed selection: %v", err)
	}
	if f.bucket.LatestID("data", plan9D5MarkerKey) != before || len(*p.work.identities) == 0 {
		t.Fatal("the marker moved, or the selection was never checked")
	}
}

func plan9D5ZeroPosition(t *testing.T) (*plan9D5Serving, hostclaim.Observed, ClaimLaunchInputs) {
	t.Helper()
	f := plan9D5Fixture(t)
	admitted := plan9D5Observed(hostclaim.Holder{Stamp: 5, Session: 5, Generation: "fresh-1", WriterBox: "box-1",
		Admissions: []hostclaim.AdmissionRecord{{Session: 5, Box: "box-1", Generation: "fresh-1", Region: "ie", Mode: "cold"}}})
	return f, admitted, ClaimLaunchInputs{Identity: f.launch.Identity, Empty: true, Admission: hostclaim.Admission{NewGeneration: "fresh-1"}}
}

// R8-3: an empty bootstrap's app may not have committed anything when it is
// ready. No marker names a generation without a root; the publisher waits
// in the background and publishes after the first durable root.
func TestPlan9D5ZeroPositionBootstrapPublishesAfterTheFirstRoot(t *testing.T) {
	t.Parallel()
	f, admitted, in := plan9D5ZeroPosition(t)
	ctx := t.Context()
	root := plan9D5RootBody(t)
	var mu sync.Mutex
	committed := false
	p := plan9D5Publisher(t, f, admitted)
	p.syncer.run = func(int) error {
		mu.Lock()
		defer mu.Unlock()
		if committed && f.bucket.LatestID("data", plan9D5RootKey("fresh-1")) == "" {
			f.bucket.Put("data", plan9D5RootKey("fresh-1"), root)
		}
		return nil
	}
	if _, err := p.publish(ctx, in); err != nil {
		t.Fatal(err)
	}
	if _, ok := f.bucket.Current("data", plan9D5MarkerKey); ok || p.publishedCount() != 0 {
		t.Fatal("a marker or publication before the generation's root")
	}
	if receipt, _ := p.outcome(); receipt != nil {
		t.Fatal("a receipt before the generation's root")
	}
	// A non-empty admission never defers: its restored root must exist.
	if _, err := f.owners.PublishFenced(ctx, admitted, ClaimLaunchInputs{Admission: in.Admission}, f.publication()); !errors.Is(err, genselect.ErrEmpty) {
		t.Fatalf("a non-empty admission without a root: %v", err)
	}
	mu.Lock()
	committed = true
	mu.Unlock()
	select {
	case <-p.a.publisher:
	case <-time.After(20 * time.Second):
		t.Fatal("the publisher never published the first root")
	}
	receipt, err := p.outcome()
	if err != nil || receipt == nil || receipt.Root.Key != plan9D5RootKey("fresh-1") || p.publishedCount() != 1 {
		t.Fatalf("receipt %+v %v published=%d", receipt, err, p.publishedCount())
	}
	body, _ := f.bucket.Current("data", plan9D5MarkerKey)
	var m genselect.Marker
	if json.Unmarshal(body, &m) != nil || m.Op != "fresh-1" || m.SnapshotTXID != 0 || !strings.Contains(string(body), `"snapshot_txid":0`) {
		t.Fatalf("marker %s", body)
	}
	if err := f.owners.publicationEvidence(ctx, admitted); err != nil {
		t.Fatal(err)
	}
}

func TestPlan9D5ZeroPositionPublisherEndsAtTheFence(t *testing.T) {
	f, admitted, in := plan9D5ZeroPosition(t)
	ctx := t.Context()
	p := plan9D5Publisher(t, f, admitted)
	if _, err := p.publish(ctx, in); err != nil {
		t.Fatal(err)
	}
	ack, err := p.drain(ctx)
	if err != nil || !slices.Contains(ack.ProducerIDs, "publisher") {
		t.Fatalf("drain %+v %v", ack, err)
	}
	select {
	case <-p.a.publisher:
	default:
		t.Fatal("the drain acknowledged before the waiting publisher joined")
	}
	if receipt, err := p.outcome(); receipt != nil || !errors.Is(err, context.Canceled) {
		t.Fatalf("outcome %+v %v", receipt, err)
	}
	if _, ok := f.bucket.Current("data", plan9D5MarkerKey); ok || p.publishedCount() != 0 {
		t.Fatal("a fenced zero-position publisher published")
	}
}

// The review's P1-4: no C8 provider, no serving wake. Select refuses, typed,
// before a generation is minted, any lease is taken or anything launches.
func TestPlan9D5ServingWithoutFilesProviderRefusesBeforeLaunch(t *testing.T) {
	f := plan9D5Fixture(t)
	f.owners.Sources.Files = nil
	created := plan9D5Observed(hostclaim.Holder{Stamp: 3})
	var refusal *ClaimDependencyError
	_, err := f.owners.Select(t.Context(), plan9D5Route, created, activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second})
	if !errors.As(err, &refusal) || !errors.Is(err, hostwrite.ErrUnavailable) || len(refusal.Contracts) != 1 || !strings.Contains(refusal.Contracts[0], "C8") || f.seq != 0 {
		t.Fatalf("select without a files provider: %v seq=%d", err, f.seq)
	}
}

// The C8 provider gets the start's captured work, after the identity proof.
func TestPlan9D5FilesProviderRunsUnderTheCapturedWork(t *testing.T) {
	f := plan9D5Fixture(t)
	admitted := plan9D5Observed(hostclaim.Holder{Stamp: 7, Session: 7, Generation: "g1", WriterBox: "box-1"})
	in := ClaimLaunchInputs{Identity: f.launch.Identity, Admission: hostclaim.Admission{NewGeneration: "g1"}}
	var seen []hostwrite.Identity
	if err := f.owners.Materialize(t.Context(), admitted, in, plan9D5Work{identities: &seen}); err != nil {
		t.Fatal(err)
	}
	if len(seen) != 2 || len(f.files) != 1 {
		t.Fatalf("work uses %v, files %v", seen, f.files)
	}
}

// A moves owner set without its own fenced publisher keeps a typed refusal,
// never the unfenced Publish.
func TestPlan9D5FencedPublicationRoutesByKind(t *testing.T) {
	var calls []string
	moves := plan9D5Owner{"moves", &calls}.owners()
	d := ClaimOwnerDispatch{Moves: &moves}
	var refusal *ClaimDependencyError
	_, err := d.PublishFenced(t.Context(), plan9D5Kind(hostclaim.KindLanding), ClaimLaunchInputs{}, ClaimPublication{})
	if !errors.As(err, &refusal) || len(calls) != 0 {
		t.Fatalf("moves without a fenced publisher: %v %v", err, calls)
	}
	if _, err := d.PublishFenced(t.Context(), plan9D5Kind(hostclaim.KindServing), ClaimLaunchInputs{}, ClaimPublication{}); !errors.Is(err, hostwrite.ErrUnavailable) {
		t.Fatalf("serving without owners: %v", err)
	}
}

// Probe readiness uses the exact attempt the start passed to Start.
func TestPlan9D5ProbeStartCapturesItsAttempt(t *testing.T) {
	probe := activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}
	_, c, ops, _, admitted := plan9D2Landing(t, hostclaim.PhaseLandingProbing, probe, true)
	s, err := c.runtime.Prepare(t.Context(), registry.Route{Host: recoveryHost}, admitted, probe)
	if err != nil {
		t.Fatal(err)
	}
	a := c.wake.get(recoveryHost)
	a.mu.Lock()
	before := a.started
	a.mu.Unlock()
	if before != nil {
		t.Fatal("an attempt captured before Start")
	}
	attempt := s.CurrentAttempt()
	a.mu.Lock()
	started := a.started
	a.mu.Unlock()
	if started == nil || *started != attempt || attempt.ID != "attempt" || slices.Index(ops.calls, "current") < 0 {
		t.Fatalf("captured %+v for %+v", started, attempt)
	}
}