package rollout

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

type swapCall struct {
	a       activator.Attempt
	outcome activator.SwapOutcome
}

type replaceCall struct {
	a        activator.Attempt
	fallback *release.StagedIdentity
}

// fakeAct models plan 3's Swap faithfully on the one point this plan relies
// on: the re-check before the switch. A superseded attempt never switches.
type fakeAct struct {
	mu        sync.Mutex
	gens      map[string]activator.Gen
	owned     map[string]bool
	n         uint64
	gates     map[string]chan struct{}
	outcome   func(a activator.Attempt) activator.SwapOutcome
	started   []string
	calls     []swapCall
	inFlight  int
	maxFlight int
	replaced  []replaceCall
	worker    func(a activator.Attempt, fallback *release.StagedIdentity) activator.SwapResult
	// mode, when set, is the effective mode Swap reports (SwapResult.Mode):
	// plan 3 runs an overlap attempt on a foreign-layout row as a hold.
	mode activator.SwapMode
	// reason, when set, is a failed Swap's diagnostic text in place of
	// "health check failed".
	reason string
	// onHeldPublish, when set, runs on every PublishHeldServing: a test's
	// view of the moment the scheduler releases a held worker publication.
	onHeldPublish func(host string)
	// executed, when set, is Swap's execution evidence (SwapResult.Executed)
	// for an attempt that ended out. Unset, only an abandoned attempt ran
	// nothing.
	executed func(a activator.Attempt, out activator.SwapOutcome) bool
}

// PublishHeldServing runs onHeldPublish, when set.
func (f *fakeAct) PublishHeldServing(host string) {
	f.mu.Lock()
	hook := f.onHeldPublish
	f.mu.Unlock()
	if hook != nil {
		hook(host)
	}
}

// EffectiveMode is the mode Swap will run m in: mode when a test set it (the
// foreign-layout override it models), m otherwise.
func (f *fakeAct) EffectiveMode(_ string, m activator.SwapMode) activator.SwapMode {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.mode != "" {
		return f.mode
	}
	return m
}

func newFakeAct() *fakeAct {
	return &fakeAct{gens: map[string]activator.Gen{}, owned: map[string]bool{}, gates: map[string]chan struct{}{}}
}

// probeAct is the Activator the harness hands its manager: the fake, with
// a hook a test may install at any moment, run before every Owns and
// Current. It is wired in at harness build, before any scheduler starts,
// and the hook is swapped atomically (final review round 1, finding 7,
// Astra): a test that replaced Deps.Act after a Pass raced the scheduler
// goroutines that read it.
type probeAct struct {
	*fakeAct
	hook atomic.Pointer[func(method, host string)]
}

// onAsk installs f as the hook; nil removes it.
func (p *probeAct) onAsk(f func(method, host string)) {
	if f == nil {
		p.hook.Store(nil)
		return
	}
	p.hook.Store(&f)
}

func (p *probeAct) ask(method, host string) {
	if f := p.hook.Load(); f != nil {
		(*f)(method, host)
	}
}

func (p *probeAct) Owns(host string) bool {
	p.ask("Owns", host)
	return p.fakeAct.Owns(host)
}

func (p *probeAct) Current(host string) (activator.Gen, bool) {
	p.ask("Current", host)
	return p.fakeAct.Current(host)
}

func (f *fakeAct) Current(host string) (activator.Gen, bool) {
	f.mu.Lock()
	defer f.mu.Unlock()
	g, ok := f.gens[host]
	return g, ok
}

func (f *fakeAct) Owns(host string) bool {
	f.mu.Lock()
	defer f.mu.Unlock()
	_, ok := f.gens[host]
	return ok || f.owned[host]
}

func (f *fakeAct) serve(host, build string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.n++
	f.gens[host] = activator.Gen{N: f.n, Build: build, StagedPath: "/opt/" + build, Digest: "sha-" + build}
}

func (f *fakeAct) hold(host string) chan struct{} {
	f.mu.Lock()
	defer f.mu.Unlock()
	ch := make(chan struct{})
	f.gates[host] = ch
	return ch
}

func (f *fakeAct) Swap(_ context.Context, a activator.Attempt, newest func(string) uint64) activator.SwapResult {
	f.mu.Lock()
	f.inFlight++
	f.maxFlight = max(f.maxFlight, f.inFlight)
	f.started = append(f.started, a.Host+":"+a.Build)
	gate := f.gates[a.Host]
	delete(f.gates, a.Host)
	f.mu.Unlock()
	if gate != nil {
		<-gate
	}
	out := activator.SwapDone
	if newest(a.Host) != a.SchedRev {
		out = activator.SwapAbandoned
	} else if f.outcome != nil {
		out = f.outcome(a)
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	f.inFlight--
	f.calls = append(f.calls, swapCall{a: a, outcome: out})
	executed := out != activator.SwapAbandoned
	if f.executed != nil {
		executed = f.executed(a, out)
	}
	if out != activator.SwapDone {
		reason := "health check failed"
		if f.reason != "" {
			reason = f.reason
		}
		return activator.SwapResult{Outcome: out, Reason: reason, Mode: f.mode, Executed: executed}
	}
	f.n++
	g := activator.Gen{N: f.n, Build: a.Build, StagedPath: a.StagedPath, Digest: a.Digest, Demand: a.Demand}
	f.gens[a.Host] = g
	return activator.SwapResult{Outcome: out, Gen: g, Executed: executed}
}

// ReplaceWorker models plan 5's same-entry socketless attempt (Task 11):
// worker decides the outcome, and every call is recorded with the fallback
// it was handed. The default is a clean replacement onto the attempt's build.
func (f *fakeAct) ReplaceWorker(_ context.Context, a activator.Attempt, fallback *release.StagedIdentity, newest func(string) uint64) activator.SwapResult {
	f.mu.Lock()
	f.replaced = append(f.replaced, replaceCall{a: a, fallback: fallback})
	decide := f.worker
	f.mu.Unlock()
	if newest(a.Host) != a.SchedRev {
		return activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: "superseded"}
	}
	res := activator.SwapResult{Outcome: activator.SwapDone}
	if decide != nil {
		res = decide(a, fallback)
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	switch {
	case res.Outcome == activator.SwapDone && res.CleanExit:
		// A one-shot's child exited: the result names its generation, but
		// the activator serves nothing for the host afterwards.
		f.n++
		res.Gen = activator.Gen{N: f.n, Build: a.Build, StagedPath: a.StagedPath, Digest: a.Digest, Demand: a.Demand}
		delete(f.gens, a.Host)
	case res.Outcome == activator.SwapDone:
		f.n++
		res.Gen = activator.Gen{N: f.n, Build: a.Build, StagedPath: a.StagedPath, Digest: a.Digest, Demand: a.Demand}
		f.gens[a.Host] = res.Gen
	case res.Outcome == activator.SwapFailed && fallback != nil:
		f.n++
		res.Gen = activator.Gen{N: f.n, Build: fallback.Version, StagedPath: fallback.StagedPath, Digest: fallback.Digest, Demand: fallback.Demand}
		f.gens[a.Host] = res.Gen
	case res.Outcome == activator.SwapDown:
		delete(f.gens, a.Host)
	}
	return res
}

func (f *fakeAct) replacements() []replaceCall {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]replaceCall(nil), f.replaced...)
}

func (f *fakeAct) snapshot() ([]swapCall, []string, int) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]swapCall(nil), f.calls...), append([]string(nil), f.started...), f.maxFlight
}

type fakeRoutes struct {
	mu   sync.Mutex
	rows map[string]registry.Route
	// getHook, when set, runs at the start of every Get, with no lock held:
	// a test's window into the moment the scheduler reads a row. getErr,
	// when set and answering an error, fails that Get.
	getHook func(host string)
	getErr  func(host string) error
	// setGoodHook, when set, runs at the start of every SetGoodBuild, with
	// no lock held. setServingHook does the same for SetServingVersion.
	setGoodHook    func(host string)
	setServingHook func(host string)
}

func (f *fakeRoutes) Get(host string) (*registry.Route, error) {
	f.mu.Lock()
	hook, fail := f.getHook, f.getErr
	f.mu.Unlock()
	if hook != nil {
		hook(host)
	}
	if fail != nil {
		if err := fail(host); err != nil {
			return nil, err
		}
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	r, ok := f.rows[host]
	if !ok {
		return nil, nil
	}
	return &r, nil
}

func (f *fakeRoutes) List() ([]registry.Route, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	out := make([]registry.Route, 0, len(f.rows))
	for _, r := range f.rows {
		out = append(out, r)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Host < out[j].Host })
	return out, nil
}

func (f *fakeRoutes) SetGoodBuild(host string, expect, next registry.GoodBuild) error {
	f.mu.Lock()
	hook := f.setGoodHook
	f.mu.Unlock()
	if hook != nil {
		hook(host)
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	r := f.rows[host]
	if r.Good() != expect {
		return registry.ErrGoodBuildChanged
	}
	r.GoodVersion, r.GoodDigest, r.GoodDemand = next.Version, next.Digest, next.Demand
	f.rows[host] = r
	return nil
}

func (f *fakeRoutes) SetServingVersion(host, ver string) error {
	f.mu.Lock()
	hook := f.setServingHook
	f.mu.Unlock()
	if hook != nil {
		hook(host)
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	r := f.rows[host]
	r.ServingVersion = ver
	f.rows[host] = r
	return nil
}

func (f *fakeRoutes) put(r registry.Route) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.rows[r.Host] = r
}

func (f *fakeRoutes) get(host string) registry.Route {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.rows[host]
}

type eventLog struct {
	mu    sync.Mutex
	lines []string
}

func (e *eventLog) App(string, string, string, string) {}

func (e *eventLog) Event(_, _, host, _, msg string) {
	e.mu.Lock()
	defer e.mu.Unlock()
	e.lines = append(e.lines, host+": "+msg)
}

func (e *eventLog) has(substr string) bool {
	e.mu.Lock()
	defer e.mu.Unlock()
	for _, l := range e.lines {
		if strings.Contains(l, substr) {
			return true
		}
	}
	return false
}

// count is how many logged lines contain substr.
func (e *eventLog) count(substr string) int {
	e.mu.Lock()
	defer e.mu.Unlock()
	n := 0
	for _, l := range e.lines {
		if strings.Contains(l, substr) {
			n++
		}
	}
	return n
}

// harness is a manager over the fakes, for app "blog" of account "bab",
// every host awake on v1, its good build.
type harness struct {
	t           *testing.T
	act         *fakeAct
	probe       *probeAct // the manager's Activator: act behind a test hook
	rows        *fakeRoutes
	book        *FailureBook
	m           *Manager
	opt, data   string
	mu          sync.Mutex
	now         time.Time
	concurrency int
	swap        string
	demands     map[string]string
	mismatch    map[string]bool
	demandErr   map[string]error
	noDigest    map[string]bool
	pointers    map[string]release.ChannelPointer // channel -> its signed pointer
	failing     atomic.Bool
	touched     atomic.Int32
	events      *eventLog
	// pointerHook, when set, runs at the start of every pointer read.
	pointerHook func()
	// demandHook, when set, runs at the start of every demand read, with no
	// lock held: a test's window into the planner's slow store read.
	demandHook func(ver string)
	// policyHook, when set, runs at the start of every policy load, with no
	// lock held: a test's window into the planner's policy read.
	policyHook func()
}

func newHarness(t *testing.T, hosts ...string) *harness {
	t.Helper()
	h := &harness{t: t, act: newFakeAct(), rows: &fakeRoutes{rows: map[string]registry.Route{}}, book: &FailureBook{},
		opt: t.TempDir(), data: t.TempDir(), now: t0, concurrency: 1,
		demands: map[string]string{}, mismatch: map[string]bool{}, demandErr: map[string]error{}, noDigest: map[string]bool{},
		pointers: map[string]release.ChannelPointer{},
		events:   &eventLog{}}
	h.book.now = h.clock // a marker's At follows the harness clock
	h.probe = &probeAct{fakeAct: h.act}
	h.stage("v1")
	for _, host := range hosts {
		h.rows.put(registry.Route{Host: host, Kind: "instance", App: "blog", Account: "bab", Status: "ready",
			Hibernate: true, Version: "v1", GoodVersion: "v1", GoodDigest: "sha-v1"})
		h.act.serve(host, "v1")
	}
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	h.m = NewManager(ctx, Deps{
		Act: h.probe, Routes: h.rows, Book: h.book, OptDir: h.opt, DataDir: h.data, Box: "box-1",
		Demand: h.demand, Policy: h.policy, Pointer: h.pointerFor, Events: h.events,
		Touch: func() { h.touched.Add(1) }, Now: h.clock, Logf: func(string, ...any) {},
	})
	return h
}

// pointer signs channel's pointer onto ver at promoted, as `carlos promote`
// (with or without --retry) would.
func (h *harness) pointer(channel, ver string, promoted time.Time) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.pointers[channel] = release.ChannelPointer{Schema: 1, Account: "bab", App: "blog", Channel: channel, Version: ver, PromotedAt: promoted}
}

func (h *harness) pointerFor(_ context.Context, _, _, channel string) (*release.ChannelPointer, error) {
	h.mu.Lock()
	hook := h.pointerHook
	h.mu.Unlock()
	if hook != nil {
		hook()
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	p, ok := h.pointers[channel]
	if !ok {
		return nil, nil
	}
	return &p, nil
}

func (h *harness) clock() time.Time {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.now
}

func (h *harness) advance(d time.Duration) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.now = h.now.Add(d)
}

func (h *harness) stage(vers ...string) {
	for _, v := range vers {
		if err := release.StageArtifact(h.opt, "bab", v, "blog", []byte(v)); err != nil {
			h.t.Fatal(err)
		}
	}
}

// demand is release.CanonicalDemand's contract: a mismatch wraps
// release.ErrStagedDigestMismatch (a staging refusal); demandErr is any other
// error, and noDigest is a store that could not compare, both of which must
// end as a hold attempt.
func (h *harness) demand(_ context.Context, _, _, ver, _ string) (string, string, error) {
	h.mu.Lock()
	hook := h.demandHook
	h.mu.Unlock()
	if hook != nil {
		hook(ver)
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.mismatch[ver] {
		return "", "", fmt.Errorf("%w: the staged executable's digest differs from the canonical manifest", release.ErrStagedDigestMismatch)
	}
	if err := h.demandErr[ver]; err != nil {
		return "", "", err
	}
	if h.noDigest[ver] {
		return "", "", nil
	}
	return h.demands[ver], "sha-" + ver, nil
}

func (h *harness) policy(context.Context, string, string) (*appconfig.Manifest, error) {
	h.mu.Lock()
	hook := h.policyHook
	h.mu.Unlock()
	if hook != nil {
		hook()
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	return &appconfig.Manifest{Schema: 1, App: "blog", Swap: h.swap, SwapConcurrency: h.concurrency}, nil
}

func (h *harness) install(ver string, hosts ...string) {
	for _, host := range hosts {
		r := h.rows.get(host)
		r.Version = ver
		h.rows.put(r)
	}
}

func (h *harness) pass() {
	rs, _ := h.rows.List()
	h.m.Pass(rs)
}

func (h *harness) servingAll(build string, hosts ...string) bool {
	for _, host := range hosts {
		if g, ok := h.act.Current(host); !ok || g.Build != build {
			return false
		}
	}
	return true
}

// flushAcks waits until the manager's ack writer has written every ack
// handed to it so far. Acks are written off the path that releases them
// (Task 11), so a test reads what was written only after this.
func (h *harness) flushAcks() {
	h.t.Helper()
	waitFor(h.t, "the ack writer to finish", h.m.acks.idle)
}

func waitFor(t *testing.T, what string, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		if cond() {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("timed out waiting for %s", what)
}