package activator

import (
	"context"
	"fmt"
	"log"
	"maps"
	"reflect"
	"time"

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/registry"
)

// The source's recovery probe entry (claim Plan 10 J2; moves §7.8 source
// probing). The host's recovery probe adapter calls RecoveryProbe once the
// record and the claim say probing with this box executing.
//
// One call is one claim wake of rt under the caller's probe policy, through
// the same path as every claim wake: K6 (a foreign executor is stopped and
// refused before anything else), the queued incarnation, WakePhase with
// probe set, then ClaimRuntime's Prerequisites, Admit and Prepare and
// startAdmitted's start loop. The policy, its Probe and its exact Inputs,
// reaches Prerequisites, Admit and Prepare unchanged: the host's
// admitsOperation, the moves Inputs owner and the attempt journal decide
// what is admitted, rendered and counted. Only a reverting or restoring op
// at probing is probed here; a landing probe is not this entry's.
//
// Lifecycle. The probe owns the host's entry exactly as a wake does: one per
// host. An identical call joins the probe in flight (or the healthy one it
// left) and shares its outcome; any other entry, or a different policy, is
// refused with hostbox.ErrChanged. Requests and sidecar requests never join
// a probe and are refused with ErrFenced, as under a fence.
//
// Fences. The rollback holds its cut fence on this host through the whole
// recovery, and the claim's exact probing executor, not the fence, is the
// authority this entry starts under. The adapter supplies its own cut fence
// generation in RecoveryFence; an unrelated installed fence refuses the wake.
// The probe runs with its own fence installed
// and the fence keeps everything else out. Installing or replacing a fence
// stops the probe. If the moves worker already lifted its own fence, the
// claim must still confirm the exact opened session before the entry serves.
//
// A healthy start (OperationAttempts.Healthy accepted it) is retained: the
// entry is awake with the started process and its replication, held. While
// held it serves nothing, has no idle clock or convergence and ignores
// Bounce; a lost lease or an exit still stands it down. The sweep reads the
// claim for it (reviewRecoveryHold): once the source opened on the same
// session (or the op released to serving keeping it) the hold ends and the
// owning cut fence is released and the entry serves as any awake entry;
// once the claim names another session or
// executor, the entry is stopped.
func (a *Service) RecoveryProbe(ctx context.Context, rt registry.Route, p StartPolicy) error {
	if a == nil || ctx == nil {
		return hostwrite.ErrUnavailable
	}
	// Every dependency of the run is checked before the first read, stop or
	// write.
	if !a.RecoveryProbeReady() {
		return hostwrite.ErrUnavailable
	}
	if rt.Host == "" || !recoveryProbePolicy(p) {
		return errRecoveryProbePolicy
	}
	a.mu.Lock()
	if a.shuttingDown {
		a.mu.Unlock()
		return ErrFenced
	}
	if fence := a.fences[rt.Host]; fence != 0 && fence != p.RecoveryFence {
		a.mu.Unlock()
		return ErrFenced
	}
	if e := a.entries[rt.Host]; e != nil {
		a.mu.Unlock()
		return a.joinRecoveryProbe(ctx, e, p)
	}
	ctx, cancel := context.WithTimeout(ctx, recoveryProbeLimit)
	defer cancel()
	policy := p
	inputs := *p.Inputs
	inputs.BundleETags = maps.Clone(p.Inputs.BundleETags)
	policy.Inputs = &inputs
	e := &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{}), recoveryPolicy: &policy, probeHeld: true, wakeCancel: cancel}
	if a.entries == nil {
		a.entries = make(map[string]*actEntry)
	}
	a.entries[rt.Host] = e
	fence := a.fences[rt.Host]
	a.mu.Unlock()
	if a.lifeCtx != nil {
		stop := context.AfterFunc(a.lifeCtx, cancel)
		defer stop()
	}
	w, err := a.doClaimWake(ctx, e)
	return a.finishRecoveryProbe(e, fence, w, err)
}

// RecoveryProbeReady checks wiring only. The host calls it before recording
// inputs; admission still validates the exact claim and session on every run.
func (a *Service) RecoveryProbeReady() bool {
	if a == nil || !a.cfg.HostWrites.Enabled {
		return false
	}
	r := a.cfg.Claims
	return r != nil && r.RecoveryAttemptsReady != nil && r.RecoveryAttemptsReady() && r.Box != "" && r.Read != nil && r.StopForeign != nil && r.Prerequisites != nil && r.Admit != nil && r.Prepare != nil && r.FailedAdmission != nil
}

// recoveryProbeLimit bounds one probe call, as wake bounds a wake.
const recoveryProbeLimit = 5 * time.Minute

// errRecoveryProbePolicy is a call that is not a recovery probe: no host, an
// ordinary or prepared policy, no inputs, or starts and window outside the
// start loop's bounds. A caller bug, refused before anything runs.
var errRecoveryProbePolicy = fmt.Errorf("%w: not a recovery probe policy", hostbox.ErrUnsupported)

func recoveryProbePolicy(p StartPolicy) bool {
	return p.Probe && p.Inputs != nil && p.Prepared == nil && p.MaxStarts >= 1 && p.MaxStarts <= 5 && p.Window > 0 && p.Window <= 30*time.Second
}

// recoveryProbeKind reports whether kind is the source's recovery, the only
// op this entry probes.
func recoveryProbeKind(kind string) bool {
	return kind == hostclaim.KindRestoring || kind == hostclaim.KindReverting
}

func sameRecoveryProbe(x, y StartPolicy) bool {
	return x.RecoveryFence == y.RecoveryFence && x.MaxStarts == y.MaxStarts && x.Window == y.Window && x.Deadline.Equal(y.Deadline) && x.Probe == y.Probe &&
		x.Prepared == nil && y.Prepared == nil && x.Inputs != nil && y.Inputs != nil && reflect.DeepEqual(*x.Inputs, *y.Inputs)
}

// joinRecoveryProbe answers a call that found host's entry already present.
// Only a probe entry under the same policy is joined: a probe in flight
// shares its outcome, and a retained healthy probe answers once its process,
// its replication and the claim's session still hold. A failed or superseded
// held probe is retired so the next pass can spend its remaining attempts.
func (a *Service) joinRecoveryProbe(ctx context.Context, e *actEntry, p StartPolicy) error {
	for {
		a.mu.Lock()
		if a.entries[e.host] != e || e.recoveryPolicy == nil {
			a.mu.Unlock()
			return hostbox.ErrChanged
		}
		if !sameRecoveryProbe(*e.recoveryPolicy, p) {
			// A new input selection must not strand an older held probe.
			// Cancel a pending start, including a late healthy result, or
			// retire the held process before the next pass can start anew.
			if e.probeHeld {
				e.probeObsolete = true
				if e.state == actWaking && e.wakeCancel != nil {
					e.wakeCancel()
				}
				if e.state == actAwake {
					a.beginStopLocked(e, StopFenced)
				}
			}
			a.mu.Unlock()
			return hostbox.ErrChanged
		}
		switch e.state {
		case actWaking:
			ready := e.ready
			a.mu.Unlock()
			select {
			case <-ready:
				if e.err != nil {
					return e.err
				}
				continue
			case <-ctx.Done():
				return ctx.Err()
			}
		case actAwake:
			if a.retireRecoveryProbeLocked(e) {
				a.mu.Unlock()
				return hostbox.ErrChanged
			}
			admitted := e.probeAdmitted
			a.mu.Unlock()
			r := a.cfg.Claims
			fresh, err := r.Read(ctx, e.host)
			if err != nil {
				return err
			}
			a.mu.Lock()
			defer a.mu.Unlock()
			if a.entries[e.host] != e || e.state != actAwake || a.retireRecoveryProbeLocked(e) {
				return hostbox.ErrChanged
			}
			_, err = recoveryHoldVerdict(admitted, fresh, r.Box)
			if err != nil && e.probeHeld {
				a.beginStopLocked(e, StopFenced)
			}
			return err
		default:
			a.mu.Unlock()
			return hostbox.ErrChanged
		}
	}
}

// finishRecoveryProbe resolves the probe entry: a failed probe is forgotten
// as a failed wake is; a healthy one is retained, held, unless the fence it
// started under was replaced, when it stops and answers ErrFenced. An owner
// lift is checked against the exact claim session by the hold review.
func (a *Service) finishRecoveryProbe(e *actEntry, fence uint64, w *wokenInstance, err error) error {
	a.mu.Lock()
	if err == nil && (w == nil || w.proc == nil || w.rep == nil) {
		err = hostwrite.ErrUnavailable
	}
	if err != nil {
		e.err = err
		outcome := &stopObservation{uncertain: true}
		if w != nil && w.stopOutcome != nil {
			outcome = w.stopOutcome
		}
		a.rememberStopLocked(e.host, outcome)
		if a.entries[e.host] == e {
			delete(a.entries, e.host)
		}
		a.mu.Unlock()
		close(e.ready)
		return err
	}
	now := a.now()
	e.state = actAwake
	e.probeFence = fence
	e.rep, e.proc = w.rep, w.proc
	e.spawnVersion = w.spawnVersion
	// Retain the launcher's socket; legacy claim launchers use the same
	// canonical fallback as an ordinary wake. Publish only when opened.
	e.gen = w.gen
	if e.gen.N == 0 && !e.worker {
		e.gen = canonicalGen(e.route, nextGen(), w.spawnVersion)
	}
	e.lastActive, e.spawnedAt = now, now
	e.bounceAfterWake = false
	if e.probeObsolete {
		e.err = hostbox.ErrChanged
		a.beginStopLocked(e, StopFenced)
	} else if a.shuttingDown || a.recoveryFenceChangedLocked(e) {
		log.Printf("activator: %s recovery probe healthy, but its fence changed; stopping it", e.host)
		e.err = ErrFenced
		why := StopFenced
		if a.shuttingDown {
			why = StopShutdown
		}
		a.beginStopLocked(e, why)
	} else {
		log.Printf("activator: %s recovery probe healthy; held until the source opens", e.host)
	}
	err = e.err
	a.mu.Unlock()
	close(e.ready)
	return err
}

// recoveryHoldVerdict compares the claim read now with the admission a held
// probe started under. The same writer (incarnation, stamp, session,
// generation) on the same op and executor at probing or probed keeps the
// hold; at opened, or released to serving with this box still the writer,
// the hold ends (open). Anything else is hostbox.ErrChanged: the probe's
// session is over.
func recoveryHoldVerdict(admitted, fresh hostclaim.Observed, box string) (open bool, err error) {
	x, y := admitted.Claim, fresh.Claim
	if fresh.Absent || fresh.Host != admitted.Host || box == "" || x == nil || y == nil || x.Holder == nil || y.Holder == nil {
		return false, hostbox.ErrChanged
	}
	for _, lost := range y.Lost {
		if lost == box {
			return false, hostbox.ErrChanged
		}
	}
	h, g := x.Holder, y.Holder
	if h.Incarnation != g.Incarnation || h.Stamp != g.Stamp || h.Session != g.Session || h.Generation != g.Generation {
		return false, hostbox.ErrChanged
	}
	if y.Op.Kind == hostclaim.KindServing && y.Op.ID == "" && y.Op.Phase == "" && g.WriterBox == box && (y.Op.Exec == "" || y.Op.Exec == box) {
		return true, nil
	}
	if y.Op.ID != x.Op.ID || y.Op.Kind != x.Op.Kind || x.Op.Exec != box || y.Op.Exec != box || !hostclaim.SessionReaches(y.Op.Kind, x.Op.Phase, y.Op.Phase) {
		return false, hostbox.ErrChanged
	}
	return y.Op.Phase == hostclaim.PhaseOpened, nil
}

// reviewRecoveryHold reads the claim for a held probe entry (off a.mu) and
// opens or stops it per recoveryHoldVerdict. An unreadable claim changes
// nothing; the next sweep reads again.
func (a *Service) reviewRecoveryHold(e *actEntry) {
	r := a.cfg.Claims
	a.mu.Lock()
	admitted := e.probeAdmitted
	a.mu.Unlock()
	var fresh hostclaim.Observed
	err := hostwrite.ErrUnavailable
	if r != nil && r.Read != nil {
		parent := a.lifeCtx
		if parent == nil {
			parent = context.Background()
		}
		ctx, cancel := context.WithTimeout(parent, 10*time.Second)
		fresh, err = r.Read(ctx, e.host)
		cancel()
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	e.probeReviewing = false
	if a.entries[e.host] != e || e.state != actAwake || !e.probeHeld {
		return
	}
	// Recheck after the off-lock read: a failed child, lost lease or a
	// replacement fence must not be opened by a late claim answer.
	if a.retireRecoveryProbeLocked(e) || err != nil {
		return
	}
	open, err := recoveryHoldVerdict(admitted, fresh, r.Box)
	switch {
	case err != nil:
		log.Printf("activator: %s recovery probe's session is over (%v); stopping it", e.host, err)
		a.beginStopLocked(e, StopFenced)
	case open:
		// The entry retains both the observed fence and the adapter's
		// owning generation. Check and remove under the same lock as the
		// hold transition; never lift a replacement operation's fence.
		// A lift clears an owned fence's owner with it (FenceAndStopAs),
		// so no owner outlives its fence.
		if e.recoveryPolicy.RecoveryFence != 0 && a.fences[e.host] == e.recoveryPolicy.RecoveryFence {
			delete(a.fences, e.host)
			delete(a.fenceOwners, e.host)
		}
		log.Printf("activator: %s source opened on the recovery probe's session; serving", e.host)
		if e.gen.N != 0 {
			a.noteCurrent(e.host, e.gen.N)
		}
		if a.gens && linksCanonical(e.route, e.gen) {
			if err := swapCanonicalLink(e.route.Socket, e.gen.Socket); err != nil {
				log.Printf("activator: %s: link %s to %s: %v (processes dialling the canonical path will not reach it)", e.host, e.route.Socket, e.gen.Socket, err)
			}
		}
		e.probeHeld = false
		e.lastActive = a.now()
	}
}

// retireRecoveryProbeLocked is shared by the real sweep and callers joining
// a retained probe. Retirement leaves the cut fence in place for the retry.
// The caller holds a.mu and has verified that e is the current awake entry.
func (a *Service) retireRecoveryProbeLocked(e *actEntry) bool {
	fenceChanged := e.probeHeld && a.recoveryFenceChangedLocked(e)
	if e.probeObsolete || fenceChanged || e.proc == nil || e.proc.Exited() || e.rep == nil || e.rep.Err() != nil {
		why := StopFenced
		switch {
		case e.probeObsolete || fenceChanged:
		case e.rep == nil || e.rep.Err() != nil:
			why = StopLeaseLost
		default:
			why = StopExited
		}
		a.beginStopLocked(e, why)
		return true
	}
	return false
}

// A moves worker may have closed the rollback and lifted its own fence
// before the activator reads opened/serving. Defer that case to the claim
// verdict, keeping probeHeld in force while the read is outstanding.
func (a *Service) recoveryFenceChangedLocked(e *actEntry) bool {
	fence := a.fences[e.host]
	if fence == 0 && e.recoveryPolicy != nil && e.probeFence == e.recoveryPolicy.RecoveryFence {
		return false
	}
	return fence != e.probeFence || (fence != 0 && (e.recoveryPolicy == nil || fence != e.recoveryPolicy.RecoveryFence))
}