//go:build unix

package host

import (
	"context"
	"encoding/json"
	"errors"
	"slices"
	"testing"
	"time"

	"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/hostrelocate"
)

// probeRemote is the probe session's database as the region's bucket sees
// it: acked is the newest transaction the session acknowledged to a
// client, shipped the newest the replicator has put in the bucket.
// SyncAndWait ships everything acknowledged; once root's stop closes the
// replicator (closed) nothing more ships, and an unshipped write is gone
// with the session. A write is acknowledged only while the writers run.
type probeRemote struct {
	acked, shipped  uint64
	stopped, closed bool
	syncErr         error
}

func (r *probeRemote) ack(t *testing.T) uint64 {
	t.Helper()
	if r.stopped || r.closed {
		t.Fatal("a write acknowledged after the writers stopped")
	}
	r.acked++
	return r.acked
}

func (r *probeRemote) SyncAndWait(context.Context) error {
	if r.syncErr != nil {
		return r.syncErr
	}
	if r.closed {
		return errors.New("the replicator is closed")
	}
	r.shipped = r.acked
	return nil
}

// plan is the bucket's listing: what a cold restore reads.
func (r *probeRemote) plan(context.Context, genselect.Request) (*genselect.Plan, error) {
	return &genselect.Plan{MaxTXID: r.shipped, MaxLevel0TXID: r.shipped}, nil
}

// probeFlush is the probe stop as RelocationEffects.EndProbe runs it on the
// stop-probe fixture's target "dst" (the real relocationStopProbe, its
// activator and root's writer-session stop), with the release's own step
// providers: WritersStopped stops the session's writers, FlushDB is the
// release's releaseFinalTXID over the session's replicator, FlushFiles a
// recorded stand-in, Positions the real positions file. Root's stop
// closes the replicator. Every step lands in f.order.
type probeFlush struct {
	f         *stopProbeFixture
	e         *RelocationEffects
	remote    *probeRemote
	positions ReleasePositionsFile
	filesErr  error
	recordErr error
}

func newProbeFlush(t *testing.T, passed time.Time) *probeFlush {
	t.Helper()
	f := newStopProbeFixture(t)
	probe, _ := json.Marshal(hostclaim.Probe{Box: "dst", Session: 12, At: passed})
	f.obs.Claim.Op.Journal["probe"] = string(probe)
	p := &probeFlush{f: f, remote: &probeRemote{acked: 40, shipped: 40}, positions: ReleasePositionsFile{Dir: t.TempDir()}}
	stop := f.a.Wake.Stop
	f.a.Wake.Stop = func(ctx context.Context, o hostclaim.Observed) error {
		p.remote.closed = true // root's stop ends the session and its replicator
		return stop(ctx, o)
	}
	p.e = &RelocationEffects{
		Identity: func(ctx context.Context) (RelocationIdentity, error) { return f.a.localStopIdentity(ctx) },
		WritersStopped: func(_ context.Context, o hostclaim.Observed) (ClaimWritersDrained, WritersStoppedEvidence, error) {
			f.order = append(f.order, "writers-stopped")
			p.remote.stopped = true
			return probeWritersStopped(o)
		},
		FlushDB: func(ctx context.Context, o hostclaim.Observed) (uint64, error) {
			f.order = append(f.order, "db")
			return releaseFinalTXID(ctx, p.remote, p.remote.plan, genselect.Request{Host: o.Host, Holder: *o.Claim.Holder})
		},
		FlushFiles: func(context.Context, hostclaim.Observed) (int64, error) {
			f.order = append(f.order, "files")
			return 3, p.filesErr
		},
		Positions: probeFlushPositions{p},
		StopProbe: f.a.relocationStopProbe(f.cfg),
	}
	return p
}

// probeFlushPositions is the real positions file, its writes in f.order.
type probeFlushPositions struct{ p *probeFlush }

func (s probeFlushPositions) Record(ctx context.Context, r ReleasePosition) error {
	s.p.f.order = append(s.p.f.order, "record")
	if s.p.recordErr != nil {
		return s.p.recordErr
	}
	return s.p.positions.Record(ctx, r)
}

func (s probeFlushPositions) Recorded(ctx context.Context, r StopRequest) (ReleasePosition, bool, error) {
	return s.p.positions.Recorded(ctx, r)
}

// probeWritersStopped is a complete StopWriters witness and root
// observation for o's writer session on "dst".
func probeWritersStopped(o hostclaim.Observed) (ClaimWritersDrained, WritersStoppedEvidence, error) {
	h := o.Claim.Holder
	at := time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC)
	target := hostguard.FenceTarget{Host: o.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Box: "dst", Enrollment: "enr", BootID: "boot", ControllerID: "ctl"}
	drained := ClaimWritersDrained{Target: target, Generation: h.Generation, WriterBox: h.WriterBox, FenceClosed: true, LaunchJoined: true, WorkDrained: true, SpawnsSettled: true,
		CgroupEmpty: true, Cgroup: "carlos/shop", DBPaths: []string{"/data/shop.db"}, CheckedAt: at}
	seen := WritersStoppedEvidence{Host: o.Host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, WriterSession: h.Session, Box: "dst", Enrollment: "enr",
		BootID: "boot", ControllerID: "ctl", ClaimETag: o.ETag, CheckedAt: at}
	return drained, seen, nil
}

func (p *probeFlush) end(t *testing.T) (hostclaim.ProbeStop, error) {
	t.Helper()
	return p.e.EndProbe(t.Context(), p.f.obs)
}

// firstServingWake is the serving CAS then main's cold serving admission
// on "dst", at the bucket's MaxTXID as a cold restore reads it: its
// writes_lost declares the probe session's writes after that position.
func (p *probeFlush) firstServingWake(t *testing.T) hostclaim.AdmissionRecord {
	t.Helper()
	c := *p.f.obs.Claim
	h := *c.Holder
	c.Holder = &h
	c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "dst", Participants: []string{"dst"}}
	plan, err := p.remote.plan(t.Context(), genselect.Request{})
	if err != nil {
		t.Fatal(err)
	}
	if err := hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: 12, ObservedGeneration: "P1", Box: "dst", Mode: "cold", NewGeneration: "S1", Region: "ie", RemoteMaxTXID: plan.MaxTXID})(&c); err != nil {
		t.Fatal(err)
	}
	return c.Holder.Admissions[len(c.Holder.Admissions)-1]
}

// Ruling 87: a write the probe session acknowledged just before its stop
// survives the first serving wake. The probe stop stops the writers,
// flushes the database (the release's releaseFinalTXID: the replicator
// ships every acknowledged write), flushes the files and records the
// positions, and only then fences and asks root's stop, which closes the
// replicator. The first serving wake is cold, as main's is, and its
// writes_lost names the probe session at the bucket's MaxTXID, which is
// now the acknowledged write's own: it declares nothing acknowledged lost.
func TestProbeStopFlushesSoAnAcknowledgedWriteSurvivesTheFirstServingWake(t *testing.T) {
	p := newProbeFlush(t, time.Now())
	acked := p.remote.ack(t) // acknowledged; the replicator has not shipped it
	if p.remote.shipped >= acked {
		t.Fatal("the write shipped before the stop")
	}
	s, err := p.end(t)
	if err != nil {
		t.Fatal(err)
	}
	if want := []string{"writers-stopped", "db", "files", "record", "fence-as", "wake-stop"}; !slices.Equal(p.f.order, want) {
		t.Fatalf("order %v, want %v", p.f.order, want)
	}
	if s.Box != "dst" || s.Session != 12 || s.Receipt != "fence-1" {
		t.Fatalf("stop %+v", s)
	}
	req := probeStopRequest(p.f.obs, mustIdentity(t, p.f))
	rec, ok, err := p.positions.Recorded(t.Context(), req)
	if err != nil || !ok || rec.TXID != acked || rec.Seq != 3 || rec.Request.ObligationID != hostguard.WriterStopObligation(12) || rec.Request.Phase != hostclaim.PhaseRelocatingProbing {
		t.Fatalf("recorded %+v %v %v", rec, ok, err)
	}
	a := p.firstServingWake(t)
	if a.Mode != "cold" || a.WritesLost == nil || a.WritesLost.Session != 12 || a.WritesLost.RemoteMaxTXID < acked {
		t.Fatalf("the first serving wake declares %+v, the probe acknowledged %d", a.WritesLost, acked)
	}
	// A repeat (the probe-stop CAS lost) does not flush again: the
	// replicator is closed, and the recorded positions are the flush.
	p.f.order = nil
	again, err := p.end(t)
	if err != nil || again != s {
		t.Fatalf("repeat: %+v %v", again, err)
	}
	if want := []string{"fence-as", "wake-stop"}; !slices.Equal(p.f.order, want) {
		t.Fatalf("repeat order %v, want %v", p.f.order, want)
	}
}

func mustIdentity(t *testing.T, f *stopProbeFixture) RelocationIdentity {
	t.Helper()
	id, err := f.a.localStopIdentity(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	return id
}

// Ruling 87's guards. A failed database flush, a failed files flush and a
// failed record each refuse the stop: nothing recorded, no fence, root
// not asked; the next pass flushes again and completes. Past the stop's
// deadline (ruling 67: ReleaseDeadline after the probe passed) the flush
// refuses before the writers are touched, yet a flush already recorded
// still lets the stop finish. Another session than the passed probe's is
// never flushed or stopped.
func TestProbeStopRefusesWithoutAFlush(t *testing.T) {
	failure := errors.New("flush failed")
	for name, fail := range map[string]func(*probeFlush){
		"the database": func(p *probeFlush) { p.remote.syncErr = failure },
		"the files":    func(p *probeFlush) { p.filesErr = failure },
		"the record":   func(p *probeFlush) { p.recordErr = failure },
	} {
		t.Run(name, func(t *testing.T) {
			p := newProbeFlush(t, time.Now())
			acked := p.remote.ack(t)
			fail(p)
			if _, err := p.end(t); !errors.Is(err, failure) {
				t.Fatalf("a failed flush of %s: %v", name, err)
			}
			if slices.Contains(p.f.order, "fence-as") || slices.Contains(p.f.order, "wake-stop") || p.remote.closed || p.f.act.svc.Fenced("shop.example") {
				t.Fatalf("the stop went ahead: %v", p.f.order)
			}
			if _, ok, err := p.positions.Recorded(t.Context(), probeStopRequest(p.f.obs, mustIdentity(t, p.f))); ok || err != nil {
				t.Fatalf("a failed flush was recorded: %v", err)
			}
			// Retried once the flush works.
			p.remote.syncErr, p.filesErr, p.recordErr = nil, nil, nil
			p.f.order = nil
			if _, err := p.end(t); err != nil {
				t.Fatal(err)
			}
			if want := []string{"writers-stopped", "db", "files", "record", "fence-as", "wake-stop"}; !slices.Equal(p.f.order, want) || p.remote.shipped != acked {
				t.Fatalf("retry order %v shipped %d", p.f.order, p.remote.shipped)
			}
		})
	}
	t.Run("past the stop's deadline", func(t *testing.T) {
		p := newProbeFlush(t, time.Now().Add(-hostrelocate.ReleaseDeadline-time.Minute))
		if _, err := p.end(t); !errors.Is(err, context.DeadlineExceeded) || len(p.f.order) != 0 {
			t.Fatalf("past the deadline: %v %v", err, p.f.order)
		}
	})
	t.Run("past the deadline, already flushed", func(t *testing.T) {
		p := newProbeFlush(t, time.Now())
		id := mustIdentity(t, p.f)
		if err := p.positions.Record(t.Context(), ReleasePosition{Request: probeStopRequest(p.f.obs, id), PID: id.PID, ProcessStart: id.ProcessStart, TXID: 40, Seq: 3}); err != nil {
			t.Fatal(err)
		}
		probe, _ := json.Marshal(hostclaim.Probe{Box: "dst", Session: 12, At: time.Now().Add(-hostrelocate.ReleaseDeadline - time.Minute)})
		p.f.obs.Claim.Op.Journal["probe"] = string(probe)
		if _, err := p.end(t); err != nil {
			t.Fatal(err)
		}
		if want := []string{"fence-as", "wake-stop"}; !slices.Equal(p.f.order, want) {
			t.Fatalf("order %v, want %v", p.f.order, want)
		}
	})
	t.Run("not the probe's session", func(t *testing.T) {
		p := newProbeFlush(t, time.Now())
		p.f.obs.Claim.Holder.Session = 13
		if _, err := p.end(t); !errors.Is(err, hostbox.ErrChanged) || len(p.f.order) != 0 || p.remote.stopped {
			t.Fatalf("another session: %v %v", err, p.f.order)
		}
	})
	t.Run("an unreadable probe", func(t *testing.T) {
		p := newProbeFlush(t, time.Now())
		p.f.obs.Claim.Op.Journal["probe"] = "{"
		if _, err := p.end(t); err == nil || len(p.f.order) != 0 {
			t.Fatalf("an unreadable probe: %v %v", err, p.f.order)
		}
	})
}