package host

import (
	"context"
	"database/sql"
	"errors"
	"os"
	"strings"

	"github.com/carlosframework/platform/internal/controlsock"
	"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/park"
	"github.com/carlosframework/platform/internal/regionname"
	"github.com/carlosframework/platform/internal/registry"
)

// claimStopLocal is this controller's verified identity when it asks for a stop.
type claimStopLocal struct {
	Actor        hostbox.Actor
	BootID       string
	ControllerID string
	PID          int
	ProcessStart string
}

// claimLaunchOwner is the tuple that journaled a launch. PID and ProcessStart
// are zero when no record was found; root then refuses the stop.
type claimLaunchOwner struct {
	BootID, ControllerID string
	DirectoryRevision    string
	RosterRevision       uint64
	PID                  int
	ProcessStart         string
}

// local reports whether this very process wrote the record. The same IDs in a
// restarted process are not local: its producers ended with the old process.
func (o claimLaunchOwner) local(l claimStopLocal) bool {
	if o.BootID != l.BootID || o.ControllerID != l.ControllerID {
		return false
	}
	return (o.PID == 0 && o.ProcessStart == "") || (o.PID == l.PID && o.ProcessStart == l.ProcessStart)
}

func ownerOf(in hostbox.LaunchInventory) claimLaunchOwner {
	return claimLaunchOwner{BootID: in.Target.BootID, ControllerID: in.Target.ControllerID, DirectoryRevision: in.Target.DirectoryRevision, RosterRevision: in.Target.RosterRevision, PID: in.ControllerPID, ProcessStart: in.ProcessStart}
}

// recoveryStopDrain refuses. A recovered controller's producers ended with it,
// and this controller's producers are never offered as that controller's drain.
type recoveryStopDrain struct{}

func (recoveryStopDrain) DrainClaim(context.Context, StopRequest, string) (DrainAcknowledgement, error) {
	return DrainAcknowledgement{}, ErrClaimStopIdentity
}

func (a ClaimAssembly) bindClaimWakeStop() {
	a.Wake.Stop = func(ctx context.Context, captured hostclaim.Observed) error {
		if err := a.stopCapturedSession(ctx, captured); err != nil {
			return err
		}
		pending := a.Wake.get(captured.Host)
		if pending != nil {
			current, _ := pending.snapshot()
			// Root has already stopped this exact session. Its pending
			// checkpoint may have advanced since capture; either observation
			// order identifies the same stopped lease, never a new session.
			reaches := a.stopReaches()
			if sameStopWriter(current, captured, reaches) || sameStopWriter(captured, current, reaches) {
				return a.Wake.release(ctx, captured.Host, pending)
			}
		}
		return nil
	}
	a.Wake.Retries.StopAttempt = a.stopRetryAttempt
}

func (a ClaimAssembly) localStopIdentity(ctx context.Context) (claimStopLocal, error) {
	if a.stopLocal != nil {
		return a.stopLocal(ctx)
	}
	if a.Controllers == nil {
		return claimStopLocal{}, hostwrite.ErrUnavailable
	}
	actor, err := a.Identity.Verify(ctx)
	if err != nil {
		return claimStopLocal{}, err
	}
	controller, err := a.Controllers.VerifiedController(ctx, os.Getpid())
	if err != nil {
		return claimStopLocal{}, err
	}
	boot, err := os.ReadFile("/proc/sys/kernel/random/boot_id")
	if err != nil {
		return claimStopLocal{}, err
	}
	start, err := claimConstructionProcessStart()
	if err != nil {
		return claimStopLocal{}, err
	}
	local := claimStopLocal{Actor: actor, BootID: strings.TrimSpace(string(boot)), ControllerID: controller, PID: os.Getpid(), ProcessStart: start}
	if local.BootID == "" || local.ControllerID == "" || local.ProcessStart == "" {
		return claimStopLocal{}, hostwrite.ErrUnavailable
	}
	return local, nil
}

// stopRetryAttempt stops the exact tuple persisted in a serving retry attempt.
// After a controller restart or reboot that tuple is not this controller's.
// It is requested in recovery mode, and root proves its controller dead.
func (a ClaimAssembly) stopRetryAttempt(ctx context.Context, o hostclaim.Observed, attempt hostclaim.WakeRetryAttempt) error {
	if o.Claim == nil || o.Claim.Holder == nil || attempt.ID == "" || attempt.BootID == "" || attempt.ControllerID == "" {
		return hostbox.ErrChanged
	}
	if a.Wake == nil || a.Launches == nil {
		return hostwrite.ErrUnavailable
	}
	if attempt.Session == 0 {
		// An unbound attempt cannot reach Prepare, so it never launched. Its
		// holder session is stopped from that session's own launch records.
		if a.Wake.Stop == nil {
			return hostwrite.ErrUnavailable
		}
		return a.Wake.Stop(ctx, o)
	}
	h := o.Claim.Holder
	if h.Session != attempt.Session {
		return hostbox.ErrChanged
	}
	local, err := a.localStopIdentity(ctx)
	if err != nil {
		return err
	}
	if attempt.Box != local.Actor.Box || attempt.Enrollment != local.Actor.Enrollment {
		return hostbox.ErrChanged
	}
	owner := claimLaunchOwner{BootID: attempt.BootID, ControllerID: attempt.ControllerID}
	recorded, err := a.Launches.Read(ctx, hostguard.FenceTarget{Host: o.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: attempt.Session, Box: attempt.Box, Enrollment: attempt.Enrollment, BootID: attempt.BootID, ControllerID: attempt.ControllerID})
	switch {
	case err == nil:
		owner = ownerOf(recorded)
		// This process already stopped and settled the attempt's own launch
		// before failing it; fencing the session would end its later retries.
		if owner.local(local) && attemptSettled(recorded, attempt) {
			return nil
		}
	case errors.Is(err, sql.ErrNoRows):
		// Keep the persisted tuple. Root refuses it without a launch record;
		// this controller's tuple is never substituted.
	default:
		return err
	}
	_, err = a.collectCapturedStop(ctx, o, local, owner, false)
	return err
}

func attemptSettled(in hostbox.LaunchInventory, attempt hostclaim.WakeRetryAttempt) bool {
	for _, r := range in.Reservations {
		if r.ID == attempt.ID+":main" && r.InvocationID == attempt.InvocationID && r.Inputs == attempt.LaunchInputs && r.State == "settled" {
			return true
		}
	}
	return false
}

// stopCapturedSession stops each journaled launch of the captured session. A
// record left by an earlier controller or boot keeps its original tuple.
func (a ClaimAssembly) stopCapturedSession(ctx context.Context, captured hostclaim.Observed) error {
	if captured.Claim == nil || captured.Claim.Holder == nil {
		return hostbox.ErrChanged
	}
	h := captured.Claim.Holder
	local, err := a.localStopIdentity(ctx)
	if err != nil {
		return err
	}
	if a.stopReaches() != nil && claimInitialSourceRecovery(captured, local.Actor.Box) {
		if err := (claimCrossBoxHandover{Guard: a.Guard, Evidence: a.Claims, Box: local.Actor.Box}).sourceRecoveryEvidence(ctx, captured); err != nil {
			return err
		}
	}
	if h.Session == 0 {
		// No writer was ever admitted, so nothing was launched or journaled.
		// Root's positive no-process witness (R4) is the stop: this
		// controller's tuple, no drain.
		_, err = a.collectCapturedStop(ctx, captured, local, claimLaunchOwner{BootID: local.BootID, ControllerID: local.ControllerID}, false)
		return err
	}
	if a.Launches == nil {
		return hostwrite.ErrUnavailable
	}
	launches, err := a.Launches.SessionLaunches(ctx, captured.Host, h.Incarnation, h.Stamp, h.Session, local.Actor.Box, local.Actor.Enrollment)
	if err != nil {
		return err
	}
	if len(launches) == 0 {
		// Nothing is journaled. Request this controller's tuple; root refuses it
		// unless its own records prove the stop. Absence is never emptiness.
		_, err = a.collectCapturedStop(ctx, captured, local, claimLaunchOwner{BootID: local.BootID, ControllerID: local.ControllerID}, true)
		return err
	}
	for _, in := range launches {
		if _, err := a.collectCapturedStop(ctx, captured, local, ownerOf(in), true); err != nil {
			return err
		}
	}
	return nil
}

// collectRetirementStop stops the one launch tuple represented by a retiring
// route. Retirement can carry only one fence receipt, so multiple journal
// owners are ambiguous and must be refused before any of them is affected.
func (a ClaimAssembly) collectRetirementStop(ctx context.Context, row registry.Route) (hostguard.FenceTarget, error) {
	if a.Launches == nil {
		return hostguard.FenceTarget{}, hostwrite.ErrUnavailable
	}
	local, err := a.localStopIdentity(ctx)
	if err != nil {
		return hostguard.FenceTarget{}, err
	}
	captured, err := hostclaim.Read(ctx, a.Claims, row.Host)
	if err != nil {
		return hostguard.FenceTarget{}, err
	}
	if captured.Claim == nil || captured.Claim.Holder == nil {
		return hostguard.FenceTarget{}, hostbox.ErrChanged
	}
	claim, holder := captured.Claim, captured.Claim.Holder
	if holder.Incarnation != row.ClaimIncarnation || holder.Stamp != row.ClaimStamp || claim.Op.Exec != local.Actor.Box {
		return hostguard.FenceTarget{}, hostbox.ErrChanged
	}
	obligation := ""
	for _, ob := range claim.Obligations {
		if ob.Kind == "park-local" && ob.Box == local.Actor.Box && ob.Owner == "box:"+local.Actor.Box && ob.Op == claim.Op.ID {
			if obligation != "" {
				return hostguard.FenceTarget{}, hostbox.ErrChanged
			}
			obligation = ob.ID
		}
	}
	if obligation == "" {
		return hostguard.FenceTarget{}, hostbox.ErrChanged
	}
	launches, err := a.Launches.SessionLaunches(ctx, captured.Host, holder.Incarnation, holder.Stamp, holder.Session, local.Actor.Box, local.Actor.Enrollment)
	if err != nil {
		return hostguard.FenceTarget{}, err
	}
	if len(launches) != 1 {
		return hostguard.FenceTarget{}, hostbox.ErrChanged
	}
	result, err := a.collectCapturedStop(ctx, captured, local, ownerOf(launches[0]), false)
	if err != nil {
		return hostguard.FenceTarget{}, err
	}
	if verifyRetirementReceipt(ctx, hostguard.FenceReceipt{Target: result.Target, Bytes: result.Bytes, EvidenceAt: result.EvidenceAt}) != nil {
		return hostguard.FenceTarget{}, ErrClaimStopReceipt
	}
	return result.Target, nil
}

func (a ClaimAssembly) collectCapturedStop(ctx context.Context, captured hostclaim.Observed, local claimStopLocal, owner claimLaunchOwner, writerSession bool) (StopResult, error) {
	if captured.Claim == nil || captured.Claim.Holder == nil {
		return StopResult{}, hostbox.ErrChanged
	}
	zero := captured.Claim.Holder.Session == 0
	if zero && !owner.local(local) {
		// A session-zero witness is only ever this controller's own request.
		return StopResult{}, ErrClaimStopIdentity
	}
	if owner.BootID == "" || owner.ControllerID == "" {
		return StopResult{}, hostwrite.ErrUnavailable
	}
	observed, err := hostclaim.Read(ctx, a.Claims, captured.Host)
	if err != nil {
		return StopResult{}, err
	}
	if !sameStopWriter(captured, observed, a.stopReaches()) {
		return StopResult{}, hostbox.ErrChanged
	}
	actor := local.Actor
	claim, holder := observed.Claim, observed.Claim.Holder
	home, err := park.VerifiedHome(actor.FleetHomeRegion, actor.HomeRegion, actor.Region, claim.Op.Intent["home_region"])
	if err != nil {
		return StopResult{}, hostbox.ErrChanged
	}
	obligation := ""
	if id, ok := a.operationWriterStop(claim, actor); (writerSession || zero) && ok {
		obligation = id
	} else if claim.Op.Kind != hostclaim.KindServing {
		for _, ob := range claim.Obligations {
			park := ob.Kind == "park-local" && ob.Owner == "box:"+actor.Box
			release := ob.Kind == "relocation-release" && ob.ID == "relocation-release" && ob.Park == nil && regionname.SameRegion(home, ob.Region, actor.Region) && strings.HasPrefix(ob.Owner, "region:") && regionname.SameRegion(home, strings.TrimPrefix(ob.Owner, "region:"), actor.Region)
			if (park || release) && ob.Box == actor.Box && ob.Op == claim.Op.ID {
				if obligation != "" {
					return StopResult{}, hostbox.ErrChanged
				}
				obligation = ob.ID
			}
		}
		if obligation == "" {
			return StopResult{}, hostbox.ErrChanged
		}
	}
	recovering := !owner.local(local)
	directoryRevision := owner.DirectoryRevision
	if directoryRevision == "" {
		// No record was found. Root still requires exact journal evidence.
		directoryRevision = actor.Binding
	}
	// The receipt numeral: the launch record's roster counter, or this
	// actor's when no record (or a record from before roster binding) names one.
	rosterRevision := owner.RosterRevision
	if rosterRevision == 0 {
		rosterRevision = actor.Revision
	}
	req := StopRequest{Region: actor.Region, Recover: recovering, Host: observed.Host, Incarnation: holder.Incarnation, InstallationStamp: holder.Stamp, WriterSession: holder.Session, OperationID: claim.Op.ID, Phase: claim.Op.Phase, ObligationID: obligation, Box: actor.Box, Enrollment: actor.Enrollment, BootID: owner.BootID, ControllerID: owner.ControllerID, ExpectedClaimETag: observed.ETag, ExpectedDirectoryRevision: directoryRevision, RosterRevision: rosterRevision}
	target := hostguard.FenceTarget{Region: req.Region, Host: req.Host, Incarnation: req.Incarnation, Stamp: req.InstallationStamp, Session: req.WriterSession, Op: req.OperationID, Phase: req.Phase, ObligationID: req.ObligationID, Box: req.Box}
	if zero && claim.Op.Kind == hostclaim.KindServing {
		// A never-admitted holder may be a new creator or an adopted legacy
		// writer on this box. Root independently proves that no process remains.
		if claim.Op.Exec != actor.Box || (holder.WriterBox != "" && holder.WriterBox != actor.Box) || len(holder.Admissions) != 0 {
			return StopResult{}, hostbox.ErrChanged
		}
	} else if !hostguard.StopAuthorityMatches(claim, target) {
		return StopResult{}, hostbox.ErrChanged
	}
	var drain StopDrainer = recoveryStopDrain{}
	if !recovering && !zero {
		producers, err := a.Wake.Producers(ctx, observed)
		if err != nil {
			return StopResult{}, err
		}
		drain = producers
	}
	var result StopResult
	if a.collectStop != nil {
		result, err = a.collectStop(ctx, req, drain)
	} else {
		result, err = CollectClaimStop(ctx, controlsock.Path, req, drain)
	}
	if err != nil {
		return StopResult{}, err
	}
	return validateTransportStop(req, result)
}

// sameStopWriter reports whether b, observed at or after a, names a's writer
// session for a stop or a lease release. reaches compares the phases; nil
// requires the same phase, as main did.
func sameStopWriter(a, b hostclaim.Observed, reaches func(kind, from, to string) bool) bool {
	return claimSameWriter(a, b, reaches) && a.Claim.Holder.WriterBox == b.Claim.Holder.WriterBox && a.Claim.Holder.Location == b.Claim.Holder.Location
}

// stopReaches is the teardown's phase predicate once the moves C1 session
// table is bound (Wake.Continue). Stopping a session grants it nothing, so it
// is not M9's adjacency: a probe admitted at probing whose op reached
// committed unseen is still stopped and released as itself (R1). A step back
// never matches. Write authority keeps Wake.Continue.
func (a ClaimAssembly) stopReaches() func(kind, from, to string) bool {
	if a.Wake == nil || a.Wake.Continue == nil {
		return nil
	}
	return hostclaim.SessionReaches
}

// operationWriterStop is the stop authority for this box's admitted operation
// writer, at probing, probed or committed (claim Plan 9 D4): before a
// restarted probe or a cold wake after the open is admitted, the predecessor
// is that writer, and no park-local chore exists or is needed for it. The
// request keeps the op and phase and names the writer session instead of a
// chore; root's full stop and receipt are unchanged, and the cold handover
// then finds that receipt for the predecessor session. Bound only with the
// moves C1 integration (Wake.Continue: CARLOS_HOST_CLAIM and
// CARLOS_MOVES_PHASE1); otherwise main's park-local requirement stands.
func (a ClaimAssembly) operationWriterStop(c *hostclaim.Claim, actor hostbox.Actor) (string, bool) {
	if a.stopReaches() == nil {
		return "", false
	}
	if c != nil && c.Op.Kind == hostclaim.KindRelocating {
		return hostguard.RelocationWriterStop(c, actor.Box, actor.Region)
	}
	if c != nil && claimRecoveryKind(c.Op.Kind) {
		return hostguard.RecoveryWriterStop(c, actor.Box, actor.Region)
	}
	return hostguard.LandingWriterStop(c, actor.Box, actor.Region)
}