package hostrelocate

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

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/release"
)

// Effects are a participant box's local effects, each under the host
// guard and bound to the claim observation it is given. Plan 6b's box
// assembly implements them (ClaimAssembly); this package only orders them
// and records each result in the claim.
type Effects interface {
	// Acknowledge writes this box's pending controller state, refuses while
	// root has an update begun (§5.10), takes the execution lease
	// (hostexec.Open under the agent's lifetime, then Ready) and returns
	// the acknowledgement to journal. After an agent restart it opens a
	// fresh controller for the same participation; the journaled
	// acknowledgement (the first lease, D6) is left as it is.
	Acknowledge(context.Context, hostclaim.Observed) (hostclaim.ExecutionAcknowledgement, error)
	// Ready reports whether this box's execution controller for the
	// operation is open and active now: ErrNoController when none is
	// retained (the agent restarted, or a controller that lost its lease
	// was joined and its scoped cleanup repeated), and the join's or the
	// cleanup's error while a lost one cannot be retired yet.
	// Every effect after the acknowledgement runs only after Ready.
	Ready(context.Context, hostclaim.Observed) error
	// Release runs §5.8's release sequence for this box's current writer
	// session: queue main's relocation-release obligation, stop every
	// writer (main's ClaimAssembly.WritersStopped, with root's own
	// observation), flush the database to its final TXID, publish every
	// pending file and wake-state mutation, close the replicator and
	// publisher, complete full stop evidence
	// (ClaimStopCoordinator.FenceAndStop), and discharge the obligation with
	// root's receipt. It returns the journal entry; it is idempotent for
	// one session.
	Release(context.Context, hostclaim.Observed) (hostclaim.Release, error)
	// Build builds the prepared generation P from the journaled pin with
	// moves' shared install routine (L16), under the replica lease.
	Build(context.Context, hostclaim.Observed) error
	// InstallLocal makes this box's local install from P's files manifest
	// and returns the receipt (§5.8 "The local install").
	InstallLocal(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, error)
	// EnsureLocal is called before P's first admission, each time: it
	// verifies that the local install the journaled receipt names is
	// complete on this box and, when it is not (a lost lease's scoped
	// cleanup retires an unopened P's local install), restores it under
	// install_stamp from P's files manifest. It is never called once P is
	// opened: an opened P is never reinstalled (rev 15, round 14). A
	// repair may write the claim: the typed inventory, and the receipt
	// itself when the repaired journal's digest differs (a configuration
	// changed since it was journaled; final review A3), replaced in one
	// CAS before any admission. The caller re-reads the same installation
	// after it.
	EnsureLocal(context.Context, hostclaim.Observed) error
	// VerifyReceipt is called before every admission after P is opened:
	// it reports whether this box's durable local install receipt is
	// still the one the claim journaled (box, enrollment, install_stamp,
	// journal digest). It checks the receipt and its binding only, never
	// the files, env or timers, which the admitted writers have changed
	// legitimately since. A missing or different receipt is false: the
	// local state cannot be rebuilt from P, so the operation is stuck
	// (§5.8, B2-5; rev 16, round 15).
	VerifyReceipt(context.Context, hostclaim.Observed) (bool, error)
	// Probe makes one cold probe admission (AdmitWriter, PredecessorCertified
	// as the caller names it, coverage "" as claim Plan 11's relocating
	// admission requires), runs the probe and waits for it to complete before
	// returning (R1: one attempt, its exact health). passed is the probe's
	// result; false is a failed attempt, never a probe still running. The
	// caller counts an attempt in the claim before each call, so a Probe
	// that returned early would spend the attempts without probing.
	Probe(ctx context.Context, obs hostclaim.Observed, certified bool) (session uint64, passed bool, err error)
	// PassedProbe reports the probe this box already made for the claim's
	// admitted session when its own durable attempt record holds it as
	// healthy (stage B final review A2): after Probe returned passed, the
	// claim read or the CAS that records the result may have failed,
	// leaving the live, passed session with no journaled probe. It is
	// asked once P is opened and no probe is journaled, before Preflight
	// or any attempt, so the result is recovered rather than a new attempt
	// launched (none can be: ruling 58, until claim ask C11). ok is false
	// when the record holds no healthy start for the holder's session; a
	// record that cannot be read is the error, never "no probe".
	PassedProbe(ctx context.Context, obs hostclaim.Observed) (session uint64, ok bool, err error)
	// Preflight is called in probing before an attempt is counted (region
	// HA ruling 48): root's isolated launch must speak the protocol and
	// install an attempt reader for relocating, or nothing is counted.
	Preflight(context.Context, hostclaim.Observed) error
	// EndProbe stops this box's passed probe session while the op is still
	// relocating/probing and returns root's receipt for it (ruling 49):
	// the activator's local fence and stop, then the claim's writer-session
	// stop through root. Idempotent for one session.
	EndProbe(context.Context, hostclaim.Observed) (hostclaim.ProbeStop, error)
	// Retained lists the participations whose execution controllers this
	// box still holds; Finish closes one and releases its lifetime lock.
	// A controller is finished once the claim no longer runs that
	// operation with this box as a participant (Sweep), so the next
	// relocation of the host can open its own (hostexec keeps the lock
	// until Close). Retained also lists every participation journaled on
	// disk whose durable record a restart still has to clear; an entry it
	// cannot read is the error, returned beside every one it can.
	Retained() ([]Participation, error)
	Finish(ctx context.Context, host, op string) error
}

// Participation names one retained execution controller.
type Participation struct {
	Host, Op string
}

// Participant is one box's half of relocation. Step makes at most one
// participant transition for one host, chosen from the claim alone.
type Participant struct {
	Claims     release.Store
	Box        string
	Enrollment string
	Effects    Effects
	Now        func() time.Time
}

// Step acts for host if this box has work in its relocation, and reports
// ErrNotYet when it has none.
func (p Participant) Step(ctx context.Context, host string) (hostclaim.Observed, error) {
	if p.Claims == nil || p.Box == "" || p.Enrollment == "" || p.Effects == nil || p.Now == nil {
		return hostclaim.Observed{}, ErrUnsupported
	}
	obs, err := hostclaim.Read(ctx, p.Claims, host)
	if err != nil || obs.Absent {
		return obs, err
	}
	c := obs.Claim
	if c.Op.Kind != hostclaim.KindRelocating || c.Op.Journal["stuck"] != "" || !slices.Contains(c.Op.Participants, p.Box) || slices.Contains(c.Lost, p.Box) {
		return obs, ErrNotYet
	}
	id := c.Op.ID
	if c.Op.Journal["exec_ack."+p.Box] == "" {
		ack, err := p.Effects.Acknowledge(ctx, obs)
		if err != nil {
			return obs, err
		}
		return p.update(ctx, obs, hostclaim.AcknowledgeExecution(p.Box, ack))
	}
	// The controller is checked before every effect, never inferred from
	// the historical acknowledgement: a restarted agent reopens one (the
	// claim keeps the first lease's record) and acts on the next pass.
	if err := p.Effects.Ready(ctx, obs); errors.Is(err, ErrNoController) {
		_, err := p.Effects.Acknowledge(ctx, obs)
		return obs, errors.Join(ErrNotYet, err)
	} else if err != nil {
		return obs, err
	}
	executor := c.Op.Exec == p.Box
	switch {
	case executor && (c.Op.Phase == hostclaim.PhaseReleasing || c.Op.Phase == hostclaim.PhaseProbeFailed) && c.Op.Journal["release"] == "":
		r, err := p.Effects.Release(ctx, obs)
		if err != nil {
			return obs, err
		}
		// The release wrote the claim (main's relocation-release obligation,
		// queued before root's stop and discharged by its receipt): journal
		// it on a fresh read.
		fresh, err := hostclaim.Read(ctx, p.Claims, host)
		if err != nil || fresh.Absent {
			return obs, errors.Join(ErrNotYet, err)
		}
		return p.update(ctx, fresh, hostclaim.JournalRelease(id, r))
	case executor && c.Op.Phase == hostclaim.PhaseInstalling:
		return obs, errors.Join(ErrNotYet, p.Effects.Build(ctx, obs))
	case executor && c.Op.Phase == hostclaim.PhaseInstalled:
		if c.Op.Journal["local_install"] != "" {
			// The receipt is a past fact; the files, env and timers it
			// names are checked (and restored) before probing.
			if err := p.Effects.EnsureLocal(ctx, obs); err != nil {
				return obs, err
			}
			// A repairing EnsureLocal may have written the claim (the
			// typed inventory): the transition is made on a fresh read of
			// the same installation (rev 3, Astra round 2 finding 2).
			fresh, err := p.sameInstallation(ctx, obs, hostclaim.PhaseInstalled)
			if err != nil {
				return obs, err
			}
			// The target enters probing itself (§5.8 phase table).
			return p.update(ctx, fresh, hostclaim.ToProbing(id, p.Now()))
		}
		l, err := p.Effects.InstallLocal(ctx, obs)
		if err != nil {
			return obs, err
		}
		if l.Box != p.Box || l.Enrollment != p.Enrollment {
			return obs, ErrRefused
		}
		// The install recorded its typed inventory in the claim (park's
		// RecordInventory): record the receipt on a fresh read of the
		// same installation, never on the stale observation.
		fresh, err := p.sameInstallation(ctx, obs, hostclaim.PhaseInstalled)
		if err != nil {
			return obs, err
		}
		return p.update(ctx, fresh, hostclaim.RecordLocalInstall(id, l))
	case executor && c.Op.Phase == hostclaim.PhaseRelocatingProbing && c.Op.Journal["probe"] == "":
		if !hostclaim.PreparedOpened(c) {
			// Before P's first admission the local install must be whole.
			if err := p.Effects.EnsureLocal(ctx, obs); err != nil {
				return obs, err
			}
		} else {
			// After it, the install is never remade; its receipt must
			// still be this box's, or the operation is stuck.
			ok, err := p.Effects.VerifyReceipt(ctx, obs)
			if err != nil {
				return obs, err
			}
			if !ok {
				return p.update(ctx, obs, hostclaim.MarkStuck(id, "the local install receipt is missing after P was admitted (§5.8, B2-5)"))
			}
			// A probe this box already passed whose record never reached
			// the claim (the read or the CAS after Probe failed) is
			// recovered from the box's durable attempt record, keyed by
			// the admitted session, before Preflight can turn the live,
			// passed session into probe-failed (final review A2).
			session, passed, err := p.Effects.PassedProbe(ctx, obs)
			if err != nil {
				return obs, err
			}
			if passed {
				return p.recordProbe(ctx, obs, session)
			}
		}
		// EnsureLocal may have written the claim: count on a fresh read
		// of the same installation.
		if !hostclaim.PreparedOpened(c) {
			fresh, err := p.sameInstallation(ctx, obs, hostclaim.PhaseRelocatingProbing)
			if err != nil {
				return obs, err
			}
			obs, c = fresh, fresh.Claim
		}
		// The release the admission would certify is read before any
		// attempt is counted (review I3, M1): a journaled release that is
		// unreadable, or names another writer than the predecessor, is a
		// claim defect no attempt can mend, so the operation is stuck for
		// an operator and no attempt is spent on it.
		certified, err := CertifiedFor(c)
		if err != nil {
			return p.update(ctx, obs, hostclaim.MarkStuck(id, "the journaled release cannot certify P's opening (§5.8, C4): "+err.Error()))
		}
		// Root must be able to authorize a relocating launch before an
		// attempt is spent (ruling 48): a refusal counts nothing. A later
		// attempt root cannot launch (ruling 58, until claim ask C11) is
		// probe-failed now, not three uncountable refusals.
		if err := p.Effects.Preflight(ctx, obs); errors.Is(err, ErrLaterAttemptUnlaunchable) {
			return p.update(ctx, obs, hostclaim.ToProbeFailed(id, p.Now()))
		} else if err != nil {
			return obs, errors.Join(ErrNotYet, err)
		}
		// The attempt is counted in the claim before it is made (§5.8:
		// at most three); the fourth is probe-failed, by this CAS.
		counted, err := hostclaim.Update(ctx, p.Claims, obs, hostclaim.BeginProbeAttempt(id, p.Box))
		if errors.Is(err, hostclaim.ErrProbeAttemptsExhausted) {
			return p.update(ctx, obs, hostclaim.ToProbeFailed(id, p.Now()))
		}
		if err != nil {
			return obs, err
		}
		obs, c = counted, counted.Claim
		session, passed, err := p.Effects.Probe(ctx, obs, certified)
		if err != nil || !passed {
			return obs, errors.Join(ErrNotYet, err)
		}
		return p.recordProbe(ctx, obs, session)
	case executor && c.Op.Phase == hostclaim.PhaseRelocatingProbing && c.Op.Journal["probe_stop"] == "":
		// Ruling 49: the passed probe's session is stopped, with root's
		// receipt, before the serving CAS ends it.
		s, err := p.Effects.EndProbe(ctx, obs)
		if err != nil {
			return obs, err
		}
		fresh, err := hostclaim.Read(ctx, p.Claims, host)
		if err != nil || fresh.Absent {
			return obs, errors.Join(ErrNotYet, err)
		}
		return p.update(ctx, fresh, hostclaim.RecordProbeStop(id, s))
	}
	return obs, ErrNotYet
}

// recordProbe journals the passed probe of session on a fresh read: the
// probe's admission wrote the claim, and RecordProbe binds the result to
// the admitted session. At is the time of this record; for a probe
// recovered from the box's attempt record (PassedProbe) that is later than
// the pass, never earlier, so the worker's release deadline (ruling 67)
// is counted from no sooner than the probe passed.
func (p Participant) recordProbe(ctx context.Context, obs hostclaim.Observed, session uint64) (hostclaim.Observed, error) {
	fresh, err := hostclaim.Read(ctx, p.Claims, obs.Host)
	if err != nil || fresh.Absent {
		return obs, errors.Join(ErrNotYet, err)
	}
	return p.update(ctx, fresh, hostclaim.RecordProbe(obs.Claim.Op.ID, hostclaim.Probe{Box: p.Box, Session: session, At: p.Now()}))
}

// ErrLaterAttemptUnlaunchable is Preflight's answer for a cold attempt
// after P's opening while root pins every relocating launch to
// install_stamp (claimLaunchAuthority, claim ask C11; ruling 58).
var ErrLaterAttemptUnlaunchable = errors.New("hostrelocate: root cannot launch a relocation attempt after P's opening")

// sameInstallation re-reads the claim after an effect that may have written
// it and requires the same operation, executor, phase, prepared P and
// install stamp, unstuck: only the effect's own bookkeeping may differ.
func (p Participant) sameInstallation(ctx context.Context, obs hostclaim.Observed, phase string) (hostclaim.Observed, error) {
	fresh, err := hostclaim.Read(ctx, p.Claims, obs.Host)
	if err != nil {
		return obs, err
	}
	a, b := obs.Claim, fresh.Claim
	if fresh.Absent || b == nil || b.Holder == nil || b.Op.Kind != hostclaim.KindRelocating || b.Op.ID != a.Op.ID || b.Op.Exec != p.Box || b.Op.Phase != phase ||
		b.Op.Journal["prepared"] != a.Op.Journal["prepared"] || b.Op.Journal["install_stamp"] != a.Op.Journal["install_stamp"] || b.Op.Journal["stuck"] != "" ||
		b.Holder.Incarnation != a.Holder.Incarnation || b.Holder.Session != a.Holder.Session || b.Holder.Stamp != a.Holder.Stamp {
		return obs, errors.Join(ErrNotYet, fmt.Errorf("%w: the installation changed under the effect", ErrRefused))
	}
	return fresh, nil
}

// Sweep finishes every retained controller whose participation is over:
// the claim no longer runs that operation, or this box is no longer a
// participant of it. It runs once per pass, for hosts the pass itself
// skips (a completed relocation is a serving claim).
func (p Participant) Sweep(ctx context.Context) error {
	if p.Claims == nil || p.Effects == nil {
		return ErrUnsupported
	}
	retained, err := p.Effects.Retained()
	errs := []error{err} // reported, while every readable participation is still finished
	for _, r := range retained {
		obs, err := hostclaim.Read(ctx, p.Claims, r.Host)
		if err != nil {
			errs = append(errs, err)
			continue
		}
		if !obs.Absent && obs.Claim.Op.Kind == hostclaim.KindRelocating && obs.Claim.Op.ID == r.Op && slices.Contains(obs.Claim.Op.Participants, p.Box) && !slices.Contains(obs.Claim.Lost, p.Box) {
			continue
		}
		if err := p.Effects.Finish(ctx, r.Host, r.Op); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

// ErrRelease is a journaled release that is present but unreadable, or
// that names another writer than the holder's predecessor. Either is a
// claim defect, never "no release": an uncertified cold opening declares
// the predecessor's writes lost, which only a failover's evidence (the
// source lost and fenced, with no release at all) may do.
var ErrRelease = errors.New("hostrelocate: the journaled release is unreadable or does not name the predecessor")

// CertifiedFor is whether the probe admission certifies its predecessor
// (§5.8, L9): only the admission that opens P, when this operation
// journaled a release whose session is the predecessor writer's. Claim
// Plan 11 gives a relocation no coverage value of its own, so the release
// is the generic PredecessorCertified, which AdmitWriter accepts on P's
// opening only with that release and a pin reaching it
// (relocationProbeOpening). No release at all (a failover, or a host that
// never had a writer) is an uncertified cold admission; after P's opening
// every later attempt is one too. A release that is present but
// unreadable, or that names another writer than the predecessor, is
// ErrRelease: absence is never evidence, and an unreadable or mismatched
// entry is not absence (fix round 1, I3).
func CertifiedFor(c *hostclaim.Claim) (bool, error) {
	raw, ok := c.Op.Journal["release"]
	if !ok {
		return false, nil
	}
	var r hostclaim.Release
	if json.Unmarshal([]byte(raw), &r) != nil || r.ReleasedBy == "" {
		return false, fmt.Errorf("%w: unreadable", ErrRelease)
	}
	if hostclaim.PreparedOpened(c) {
		return false, nil
	}
	if c.Holder == nil {
		return false, fmt.Errorf("%w: no holder", ErrRelease)
	}
	pred := hostclaim.AdmissionRecord{Session: c.Holder.Session, Box: c.Holder.WriterBox, Generation: c.Holder.Generation}
	if n := len(c.Holder.Admissions); n > 0 {
		pred = c.Holder.Admissions[n-1]
	}
	if pred.Box != r.ReleasedBy || pred.Session != r.Session || pred.Generation != r.Generation {
		return false, fmt.Errorf("%w: released by %s (session %d, generation %q), the predecessor is %s (session %d, generation %q)", ErrRelease, r.ReleasedBy, r.Session, r.Generation, pred.Box, pred.Session, pred.Generation)
	}
	return true, nil
}

func (p Participant) update(ctx context.Context, obs hostclaim.Observed, tr func(*hostclaim.Claim) error) (hostclaim.Observed, error) {
	next, err := hostclaim.Update(ctx, p.Claims, obs, tr)
	if errors.Is(err, hostclaim.ErrAlreadyRecorded) || errors.Is(err, hostclaim.ErrAckAlreadyApplied) {
		return obs, nil
	}
	return next, err
}