package host

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

	"github.com/carlosframework/platform/internal/hostclaim"
	execctl "github.com/carlosframework/platform/internal/hostexec"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostrelocate"
	"github.com/carlosframework/platform/internal/release"
)

// RelocationIdentity is this box's verified actor and stop identity, as
// Plan 6b's assembly reports it: main's claimStopLocal, which
// localStopIdentity fills (actor, boot, controller, and this process's PID
// and start time). Release records the process with the flushed positions
// and decides whether a recorded flush is this process's own with main's
// claimLaunchOwner.local rule.
type RelocationIdentity = claimStopLocal

// ReleasePosition is one recorded release: the exact stop request, the
// process that flushed it (PID and start time, beside the request's boot
// and controller), and the positions its flush reached.
type ReleasePosition struct {
	Request      StopRequest `json:"request"`
	PID          int         `json:"pid"`
	ProcessStart string      `json:"process_start"`
	TXID         uint64      `json:"txid"`
	Seq          int64       `json:"seq"`
}

// owner is the recorded flush's launch tuple in main's form, so the
// decision whether it is this process's own is claimLaunchOwner.local's.
func (p ReleasePosition) owner() claimLaunchOwner {
	return claimLaunchOwner{BootID: p.Request.BootID, ControllerID: p.Request.ControllerID, DirectoryRevision: p.Request.ExpectedDirectoryRevision,
		PID: p.PID, ProcessStart: p.ProcessStart}
}

// ReleasePositions keeps the flushed positions of one release durably,
// bound to the exact stop request (host, incarnation, stamp, session,
// operation, phase, box, enrollment, boot, controller) and the flushing
// process, from before root completes its receipt. A replayed receipt (root's durable fence after a
// lost response or a crash before JournalRelease) is paired with them;
// without a record for that exact request the release is refused rather
// than guessed.
// Recorded matches the stop's installation and session (host, incarnation,
// stamp, session, operation, phase, obligation, box, enrollment), not its
// controller identity: a restarted agent asks with a new boot and
// controller id (or the same ids in a new process), and gets back the
// request the flush was made under, so it can ask root for that stop's
// durable receipt.
type ReleasePositions interface {
	Record(ctx context.Context, p ReleasePosition) error
	Recorded(ctx context.Context, r StopRequest) (recorded ReleasePosition, ok bool, err error)
}

// RelocationEffects is hostrelocate.Effects over providers the claim
// assembly supplies (rev 2). It owns only the order of §5.8's release
// sequence and the shape of each record; every provider acts under the
// host guard. The execution controller is main's: one participation owner
// per box (ruling 42), opened by productionClaimLocal for every non-serving
// operation that lists this box, whichever of the claim's participation
// loop, WorkSource.Prepare or this pass asks first.
type RelocationEffects struct {
	// Claims is the claim store: EnsureLocal replaces the journaled
	// receipt there after a repair (final review A3).
	Claims release.Store
	// Participation is main's participation owner for the observed
	// operation (productionClaimLocal.Acknowledged): it opens or reuses
	// the box's one hostexec controller for {host, op} and returns its
	// ready state. A lost lease stays main's denial (CC9).
	Participation func(context.Context, hostclaim.Observed) (execctl.State, error)
	Identity      func(context.Context) (RelocationIdentity, error)
	// Producers is the admitted session's drain (ClaimWakeAssembly.Producers).
	Producers func(context.Context, hostclaim.Observed) (StopDrainer, error)
	// Collect asks root for full stop evidence (CollectClaimStop over the
	// instance-control socket); a RecoveryReleaseDrain is submitted as a
	// recovery request.
	Collect func(context.Context, StopRequest, StopDrainer) (StopResult, error)
	// WritersStopped is release step 1, action and observation: main's
	// ClaimAssembly.WritersStopped for the claim's host.
	WritersStopped func(context.Context, hostclaim.Observed) (ClaimWritersDrained, WritersStoppedEvidence, error)
	// Releases is main's relocation-release obligation (claim Plan 11).
	Releases RelocationReleaseDischarge
	// FlushDB and FlushFiles are release steps 2 and 3.
	FlushDB    func(context.Context, hostclaim.Observed) (uint64, error)
	FlushFiles func(context.Context, hostclaim.Observed) (int64, error)
	// Positions is the durable record of the flushed positions.
	Positions ReleasePositions
	// BuildP, Install, VerifyLocal, Receipt, Admit: Tasks 8b to 8d.
	BuildP      func(context.Context, hostclaim.Observed) error
	Install     func(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, error)
	VerifyLocal func(context.Context, hostclaim.Observed, hostclaim.LocalInstall) (bool, error)
	Receipt     func(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, bool, error)
	Admit       func(context.Context, hostclaim.Observed, bool) (uint64, bool, error)
	// Passed is the probe this box already passed for the admitted session,
	// from its attempt journal (relocationPassedProbe; final review A2).
	Passed func(context.Context, hostclaim.Observed) (uint64, bool, error)
	// PreflightKind is root's isolated-launch capability for kind
	// relocating (ClaimRoleAssembly.Preflight then PreflightKind; ruling 48).
	PreflightKind func(context.Context, hostclaim.Observed) error
	// StopProbe is ruling 49's stop of the passed probe session (Task 9).
	StopProbe func(context.Context, hostclaim.Observed) (hostclaim.ProbeStop, error)
	// ReleaseHold releases this box's row hold for a completed operation it
	// serves, and lifts the activator fence StopProbe installed (Task 9).
	ReleaseHold func(ctx context.Context, host, op string) error
}

var _ hostrelocate.Effects = (*RelocationEffects)(nil)

// Acknowledge returns the lease of main's participation controller for the
// operation (ruling 42). The claim keeps the first acknowledgement; a
// restarted agent's later controller is main's to open.
func (e *RelocationEffects) Acknowledge(ctx context.Context, obs hostclaim.Observed) (hostclaim.ExecutionAcknowledgement, error) {
	if e.Participation == nil || obs.Claim == nil || obs.Claim.Holder == nil {
		return hostclaim.ExecutionAcknowledgement{}, ErrClaimStopUnavailable
	}
	st, err := e.Participation(ctx, obs)
	if err != nil {
		return hostclaim.ExecutionAcknowledgement{}, err
	}
	if st.Op != obs.Claim.Op.ID || st.Incarnation != obs.Claim.Holder.Incarnation || st.Host != obs.Host {
		return hostclaim.ExecutionAcknowledgement{}, execctl.ErrLost
	}
	return hostclaim.ExecutionAcknowledgement{Op: obs.Claim.Op.ID, Enrollment: st.Enrollment, BootID: st.BootID, ControllerID: st.ControllerID, LeaseETag: st.LeaseETag, Expires: st.Expires}, nil
}

// Ready checks main's participation controller is active now. A lost one
// is main's retained denial, never reopened here.
func (e *RelocationEffects) Ready(ctx context.Context, obs hostclaim.Observed) error {
	_, err := e.Acknowledge(ctx, obs)
	return err
}

// Release runs §5.8's release in claim Plan 11 item 4's order (ruling 28):
// writers stopped and observed, the database and files flushed and the
// positions recorded at the claim as observed; then the obligation queued;
// then root's full stop with the producers' own drain; then Complete.
func (e *RelocationEffects) Release(ctx context.Context, obs hostclaim.Observed) (hostclaim.Release, error) {
	if e.Identity == nil || e.Producers == nil || e.Collect == nil || e.WritersStopped == nil || e.Releases == nil || e.FlushDB == nil || e.FlushFiles == nil || e.Positions == nil || obs.Claim == nil || obs.Claim.Holder == nil {
		return hostclaim.Release{}, ErrClaimStopUnavailable
	}
	id, err := e.Identity(ctx)
	if err != nil {
		return hostclaim.Release{}, err
	}
	if id.PID <= 0 || id.ProcessStart == "" {
		return hostclaim.Release{}, ErrClaimStopUnavailable
	}
	if key, digest := obs.Claim.Op.Journal["release_stop_receipt"], obs.Claim.Op.Journal["release_stop_sha256"]; key != "" || digest != "" {
		return e.replayDischarged(ctx, obs, id, key, digest)
	}
	h := obs.Claim.Holder
	req := releaseStopRequest(obs, id)
	recorded, err := e.flushStopped(ctx, obs, id, req)
	if err != nil {
		return hostclaim.Release{}, err
	}
	// Root fences a relocation-release stop only while the claim owes it
	// (hostguard.StopAuthorityMatches): queued now, at the flushed claim.
	if obs, err = e.Releases.Queue(ctx, obs); err != nil {
		return hostclaim.Release{}, err
	}
	req = releaseStopRequest(obs, id)
	req.ExpectedDirectoryRevision = recorded.Request.ExpectedDirectoryRevision
	var d StopDrainer = RecoveryReleaseDrain{}
	if recorded.owner().local(id) {
		inner, err := e.Producers(ctx, obs)
		if err != nil {
			return hostclaim.Release{}, err
		}
		d = &releaseDrain{inner: inner}
	} else {
		req.BootID, req.ControllerID = recorded.Request.BootID, recorded.Request.ControllerID
	}
	result, err := e.Collect(ctx, req, d)
	if err != nil {
		return hostclaim.Release{}, err
	}
	if _, err := e.Releases.Complete(ctx, obs, result.Key, result.Digest); err != nil {
		return hostclaim.Release{}, err
	}
	return hostclaim.Release{ReleasedBy: id.Actor.Box, Enrollment: id.Actor.Enrollment, Generation: h.Generation, Session: h.Session,
		FinalTXID: recorded.TXID, ManifestSeq: recorded.Seq, Receipt: result.Key, ReceiptSHA256: result.Digest}, nil
}

// flushStopped is release steps 1 to 3 and the record (ruling 28), the
// one implementation the release and the probe stop share (ruling 87):
// writers stopped and observed for the claim as observed, then the
// database flushed (FlushDB), then the files (FlushFiles), then the
// positions recorded durably for req. A position already recorded for req
// (a retry after a later step failed, or after root's stop closed the
// session's replicator) is returned with nothing done again. Any failure
// returns its error with nothing recorded, so the stop that follows is
// never asked and the next pass flushes again: a flush is never taken as
// done without its record. A context already done refuses before the
// writers are touched.
func (e *RelocationEffects) flushStopped(ctx context.Context, obs hostclaim.Observed, id RelocationIdentity, req StopRequest) (ReleasePosition, error) {
	recorded, flushed, err := e.Positions.Recorded(ctx, req)
	if err != nil {
		return ReleasePosition{}, err
	}
	if flushed {
		return recorded, nil
	}
	if err := ctx.Err(); err != nil {
		return ReleasePosition{}, err
	}
	h := obs.Claim.Holder
	drained, seen, err := e.WritersStopped(ctx, obs)
	if err != nil {
		return ReleasePosition{}, err
	}
	if err := drained.For(obs); err != nil {
		return ReleasePosition{}, err
	}
	if seen.Host != obs.Host || seen.Incarnation != h.Incarnation || seen.InstallationStamp != h.Stamp || seen.WriterSession != h.Session ||
		seen.Box != id.Actor.Box || seen.Enrollment != id.Actor.Enrollment || seen.CheckedAt.IsZero() {
		return ReleasePosition{}, fmt.Errorf("%w: writers-stopped evidence names another session", ErrClaimStopIdentity)
	}
	txid, err := e.FlushDB(ctx, obs)
	if err != nil {
		return ReleasePosition{}, err
	}
	seq, err := e.FlushFiles(ctx, obs)
	if err != nil {
		return ReleasePosition{}, err
	}
	recorded = ReleasePosition{Request: req, PID: id.PID, ProcessStart: id.ProcessStart, TXID: txid, Seq: seq}
	if err := e.Positions.Record(ctx, recorded); err != nil {
		return ReleasePosition{}, err
	}
	return recorded, nil
}

// probeStopRequest is the probe stop's request for obs's passed probe
// session (ruling 87): the release's request shape at relocating/probing,
// under root's writer-session stop obligation for that session, so its
// recorded positions are never a release's.
func probeStopRequest(obs hostclaim.Observed, id RelocationIdentity) StopRequest {
	r := releaseStopRequest(obs, id)
	r.ObligationID = hostguard.WriterStopObligation(obs.Claim.Holder.Session)
	return r
}

// probeStopDeadline is ruling 67's bound on the probe stop, as the worker
// counts it: ReleaseDeadline after the probe passed (Probe.At; for an
// untimed probe, probing_since plus ProbeDeadline). A probe or a
// probing_since that cannot be read is an error, never "no bound".
func probeStopDeadline(c *hostclaim.Claim) (time.Time, error) {
	var p hostclaim.Probe
	if err := json.Unmarshal([]byte(c.Op.Journal["probe"]), &p); err != nil {
		return time.Time{}, fmt.Errorf("%w: the passed probe cannot be read: %v", hostrelocate.ErrRefused, err)
	}
	passed := p.At
	if passed.IsZero() {
		since, err := time.Parse(time.RFC3339Nano, c.Op.Journal["probing_since"])
		if err != nil {
			return time.Time{}, fmt.Errorf("%w: probing_since cannot be read: %v", hostrelocate.ErrRefused, err)
		}
		passed = since.Add(hostrelocate.ProbeDeadline)
	}
	return passed.Add(hostrelocate.ReleaseDeadline), nil
}

// releaseStopRequest is the release's stop request for obs's session, as
// this box's identity id asks it.
func releaseStopRequest(obs hostclaim.Observed, id RelocationIdentity) StopRequest {
	c, h := obs.Claim, obs.Claim.Holder
	return StopRequest{Region: id.Actor.Region, Host: c.Host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, WriterSession: h.Session,
		OperationID: c.Op.ID, Phase: c.Op.Phase, ObligationID: RelocationReleaseObligation, Box: id.Actor.Box, Enrollment: id.Actor.Enrollment,
		BootID: id.BootID, ControllerID: id.ControllerID, ExpectedClaimETag: obs.ETag, ExpectedDirectoryRevision: id.Actor.Binding, RosterRevision: id.Actor.Revision}
}

// replayDischarged returns the release whose receipt already discharged
// the obligation: the claim journals key and digest and owes the operation
// nothing, the receipt verifies again (main's Complete in its replay form),
// and this box recorded the positions its flush reached for this stop.
// Without a recorded flush it refuses rather than guess, as a replayed
// receipt always does.
func (e *RelocationEffects) replayDischarged(ctx context.Context, obs hostclaim.Observed, id RelocationIdentity, key, digest string) (hostclaim.Release, error) {
	c, h := obs.Claim, obs.Claim.Holder
	if err := releaseDischarged(c, key, digest); err != nil {
		return hostclaim.Release{}, err
	}
	if err := e.Releases.Verify(ctx, obs, key, digest); err != nil {
		return hostclaim.Release{}, err
	}
	recorded, flushed, err := e.Positions.Recorded(ctx, releaseStopRequest(obs, id))
	if err != nil {
		return hostclaim.Release{}, err
	}
	if !flushed {
		return hostclaim.Release{}, fmt.Errorf("%w: a discharged receipt has no recorded flushed positions for this stop", ErrClaimStopReceipt)
	}
	return hostclaim.Release{ReleasedBy: id.Actor.Box, Enrollment: id.Actor.Enrollment, Generation: h.Generation, Session: h.Session,
		FinalTXID: recorded.TXID, ManifestSeq: recorded.Seq, Receipt: key, ReceiptSHA256: digest}, nil
}

// RecoveryReleaseDrain refuses: a restarted agent's producers are not the
// stop's, and root must replay the durable receipt or nothing. It is
// main's recoveryStopDrain, under the name the Collect provider matches.
type RecoveryReleaseDrain = recoveryStopDrain

// releaseDrain is the producers' own drain: steps 1 to 3 ran before the
// obligation was queued (ruling 28), so root's collection closes the
// producers and acknowledges, nothing more.
type releaseDrain struct{ inner StopDrainer }

func (d *releaseDrain) DrainClaim(ctx context.Context, r StopRequest, fence string) (DrainAcknowledgement, error) {
	if d.inner == nil {
		return DrainAcknowledgement{}, ErrClaimStopUnavailable
	}
	return d.inner.DrainClaim(ctx, r, fence)
}

// Retained is empty: main's participation owner keeps and finishes every
// controller (ruling 42), so the participant's Sweep has nothing to do.
func (e *RelocationEffects) Retained() ([]hostrelocate.Participation, error) { return nil, nil }

// Finish releases the completed operation's row hold on the serving box
// and lifts the probe's activator fence (ruling 49 step 5). The controller
// is main's to close.
func (e *RelocationEffects) Finish(ctx context.Context, host, op string) error {
	if e.ReleaseHold == nil {
		return ErrClaimStopUnavailable
	}
	return e.ReleaseHold(ctx, host, op)
}

func (e *RelocationEffects) Build(ctx context.Context, obs hostclaim.Observed) error {
	if e.BuildP == nil {
		return ErrClaimStopUnavailable
	}
	return e.BuildP(ctx, obs)
}

func (e *RelocationEffects) InstallLocal(ctx context.Context, obs hostclaim.Observed) (hostclaim.LocalInstall, error) {
	if e.Install == nil {
		return hostclaim.LocalInstall{}, ErrClaimStopUnavailable
	}
	return e.Install(ctx, obs)
}

// EnsureLocal is hostrelocate.Effects' check before P's first admission
// (ruling 52): the journaled receipt's install is verified and, when it is
// damaged while its install stamp is still the active one, made whole by
// the same Install provider under that stamp. A retired stamp is never
// reactivated: Install refuses it, and the probe deadline replaces the
// target. An opened P is never reinstalled.
//
// A repair re-renders the held configuration, so its journal, and the
// digest the receipt carries, differ from the journaled receipt's when
// the configuration changed since (final review A3). The receipt is then
// replaced in the claim, in one CAS on a fresh read of the same
// installation (hostclaim.ReplaceLocalInstall), before any admission, so
// this verification, VerifyReceipt and relocationAfterAdmission all
// compare the one digest the box holds with the one the claim journals. A
// CAS that fails leaves the old receipt, which the next pass finds not
// verifiable and repairs again to the same journal.
func (e *RelocationEffects) EnsureLocal(ctx context.Context, obs hostclaim.Observed) error {
	if e.VerifyLocal == nil || e.Install == nil || e.Claims == nil || obs.Claim == nil {
		return ErrClaimStopUnavailable
	}
	c := obs.Claim
	if hostclaim.PreparedOpened(c) {
		return nil
	}
	var l hostclaim.LocalInstall
	if err := json.Unmarshal([]byte(c.Op.Journal["local_install"]), &l); err != nil || l.InstallStamp == 0 {
		return fmt.Errorf("%w: no local install receipt to verify", ErrClaimStopReceipt)
	}
	if ok, err := e.VerifyLocal(ctx, obs, l); err != nil || ok {
		return err
	}
	got, err := e.Install(ctx, obs)
	if err != nil {
		return err
	}
	if got.Box != l.Box || got.Enrollment != l.Enrollment || got.InstallStamp != l.InstallStamp {
		return fmt.Errorf("%w: the restored install is not the receipt's", ErrClaimStopReceipt)
	}
	if got != l {
		fresh, err := hostclaim.Read(ctx, e.Claims, obs.Host)
		if err != nil {
			return err
		}
		if fresh.Absent || fresh.Claim.Op.ID != c.Op.ID || fresh.Claim.Op.Journal["prepared"] != c.Op.Journal["prepared"] {
			return fmt.Errorf("%w: the installation changed under the repair", hostrelocate.ErrRefused)
		}
		if obs, err = hostclaim.Update(ctx, e.Claims, fresh, hostclaim.ReplaceLocalInstall(c.Op.ID, l, got)); err != nil {
			return err
		}
		l = got
	}
	ok, err := e.VerifyLocal(ctx, obs, l)
	if err == nil && !ok {
		err = fmt.Errorf("%w: the restored install did not verify", ErrClaimStopReceipt)
	}
	return err
}

// VerifyReceipt is hostrelocate.Effects' check before every admission
// after P is opened: the box's durable receipt must be the one the claim
// journaled, exactly. It reads the receipt alone: the files, env and
// timers are the admitted writers' now, and nothing is restored.
func (e *RelocationEffects) VerifyReceipt(ctx context.Context, obs hostclaim.Observed) (bool, error) {
	if e.Receipt == nil || obs.Claim == nil {
		return false, ErrClaimStopUnavailable
	}
	var want hostclaim.LocalInstall
	if err := json.Unmarshal([]byte(obs.Claim.Op.Journal["local_install"]), &want); err != nil || want.InstallStamp == 0 {
		return false, nil // no journaled receipt to match: nothing vouches for the local state
	}
	got, ok, err := e.Receipt(ctx, obs)
	if err != nil {
		return false, err
	}
	return ok && got == want, nil
}

func (e *RelocationEffects) Probe(ctx context.Context, obs hostclaim.Observed, certified bool) (uint64, bool, error) {
	if e.Admit == nil {
		return 0, false, ErrClaimStopUnavailable
	}
	return e.Admit(ctx, obs, certified)
}

// PassedProbe recovers a passed probe whose record never reached the claim
// (final review A2) from the box's durable attempt journal, keyed by the
// admitted session.
func (e *RelocationEffects) PassedProbe(ctx context.Context, obs hostclaim.Observed) (uint64, bool, error) {
	if e.Passed == nil {
		return 0, false, ErrClaimStopUnavailable
	}
	return e.Passed(ctx, obs)
}

// Preflight refuses before an attempt is counted while root cannot launch
// a relocating probe (ruling 48).
func (e *RelocationEffects) Preflight(ctx context.Context, obs hostclaim.Observed) error {
	if e.PreflightKind == nil || obs.Claim == nil || obs.Claim.Op.Kind != hostclaim.KindRelocating {
		return ErrClaimStopUnavailable
	}
	return e.PreflightKind(ctx, obs)
}

// EndProbe is ruling 49's stop of the passed probe session, flushed first
// (ruling 87): the session's writers are stopped, its database and files
// flushed and the positions recorded (flushStopped, the release's own
// steps), and only then the probe stop (StopProbe: the activator's owned
// fence and stop, which closes the session's replicator, then root's
// writer-session stop). A write the probe session acknowledged is then
// shipped before the session ends, so the first serving wake's cold
// admission, whose writes_lost is bounded by the remote's MaxTXID, loses
// none of it. The flush is bounded by the stop's own deadline (ruling 67,
// probeStopDeadline): past it the flush refuses and the worker marks the
// op stuck. A failed flush refuses the stop with nothing asked of the
// activator or root, and the next pass flushes again.
func (e *RelocationEffects) EndProbe(ctx context.Context, obs hostclaim.Observed) (hostclaim.ProbeStop, error) {
	if e.StopProbe == nil || e.Identity == nil || e.WritersStopped == nil || e.FlushDB == nil || e.FlushFiles == nil || e.Positions == nil || obs.Claim == nil || obs.Claim.Holder == nil {
		return hostclaim.ProbeStop{}, ErrClaimStopUnavailable
	}
	id, err := e.Identity(ctx)
	if err != nil {
		return hostclaim.ProbeStop{}, err
	}
	if id.PID <= 0 || id.ProcessStart == "" {
		return hostclaim.ProbeStop{}, ErrClaimStopUnavailable
	}
	// Only this box's passed probe session is flushed: its writers are
	// stopped below, so another session is never touched.
	if err := passedProbeSession(obs.Claim, id.Actor.Box); err != nil {
		return hostclaim.ProbeStop{}, err
	}
	deadline, err := probeStopDeadline(obs.Claim)
	if err != nil {
		return hostclaim.ProbeStop{}, err
	}
	bounded, cancel := context.WithDeadline(ctx, deadline)
	_, err = e.flushStopped(bounded, obs, id, probeStopRequest(obs, id))
	cancel()
	if err != nil {
		return hostclaim.ProbeStop{}, err
	}
	return e.StopProbe(ctx, obs)
}