package activator

import (
	"context"
	"errors"
	"log"
	"time"

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

// The landing's probe wake (moves Phase 1 Plan 3 Task 5a; spec §7.7, §7.14;
// ruling H2). ProbeWake is glue over the claim's probe admission path (claim
// Plan 9 Group D): the same entry, wake and doClaimWake a request's wake
// runs, with the probe's start policy and the probe's two refusals (the
// registry row held by HoldOp, the claim this box's landing op at probing).
// The claim runtime does everything else (Prerequisites, Admit, Prepare and
// startAdmitted): the moves owners it dispatches to choose and evidence the
// cold admission into the prepared generation, render and materialize the
// held inputs, run the start check and publish; the moves attempts journal
// the window and the starts; the claim's probe readiness checks the peer.
// ProbeWake contains no admission, restore or readiness logic of its own.
//
// Production: unarmed. Nothing calls ProbeWake until the landing's probing
// phase (Plan 3 Task 6) is armed with Plan 5 Task 9's parks, and on main
// every claim start is refused (ClaimRoleAssembly.Preflight and
// RoleLaunch.Start answer ErrClaimIsolationUnavailable) until the claim's
// isolated root launcher lands.

var (
	// ErrProbeRefused is a probe refused before any admission: the row is not
	// held by the probe's hold op, the claim does not name this box the
	// executor of the probe's op at probing, or the host already has a live
	// entry. Nothing was admitted and nothing ran.
	ErrProbeRefused = errors.New("activator: the probe was refused before admission")
	// ErrProbeFailed is an admitted probe attempt that did not pass: the
	// claim settled the admitted tree (FailedAdmission) and the attempt is
	// over. The cause is joined to it (ErrAttemptOver, a refusal from the
	// moves owners, a readiness failure).
	ErrProbeFailed = errors.New("activator: the probe attempt failed")
)

// The probe's start policy (§7.7): at most five starts, the 30 s window
// counted from the first app start.
const (
	probeMaxStarts = 5
	probeWindow30  = time.Duration(moves.ProbeSeconds * float64(time.Second))
)

// ProbeConfig is one probe attempt.
type ProbeConfig struct {
	Host string
	// Op is the claim's op id: the move id.
	Op string
	// HoldOp is the registry row's hold_op: the move id for a landing.
	HoldOp string
	// Deadline bounds the whole attempt: the landing's deadline, the earlier
	// of cut_deadline and landing_deadline (ruling H8).
	Deadline time.Time
	// Prepared and Inputs are the build's install manifest and the move's
	// pins; the moves owners check them against the journal and the record.
	Prepared *moves.InstallManifest
	Inputs   *provision.MoveInputs
}

// ProbeResult names the admitted session that passed: the process the entry
// now holds, which serves after the open on the same session (H3).
type ProbeResult struct {
	Session    uint64
	Stamp      uint64
	Generation string
}

// LiveProbe reports the awake entry ProbeWake left for host: the probe's op
// and its admitted session. ok is false for no entry, a waking or stopping
// one, or an entry that is not a passed probe.
func (a *Service) LiveProbe(host string) (op string, res ProbeResult, ok bool) {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	if e == nil || e.state != actAwake || e.probe == nil || e.probe.admitted == nil || e.probe.admitted.Claim == nil || e.probe.admitted.Claim.Holder == nil {
		return "", ProbeResult{}, false
	}
	return e.probe.op, probeResultOf(e.probe.admitted), true
}

func probeResultOf(o *hostclaim.Observed) ProbeResult {
	h := o.Claim.Holder
	return ProbeResult{Session: h.Session, Stamp: h.Stamp, Generation: h.Generation}
}

// Test seams: retireBetweenLocks runs in RetireExcept between the probe
// snapshot and the retirement pass; probeWakePublished runs in ProbeWake
// after the entry is published and before its wake starts.
var retireBetweenLocks, probeWakePublished, retireBeforeRegistryRead func()

// probeEntry marks an actEntry ProbeWake made.
type probeEntry struct {
	policy     startPolicy
	op, holdOp string
	admitted   *hostclaim.Observed
	// parent is ProbeWake's context: the wake's own context derives from
	// it, so a cancel is in force before the entry is published or its
	// goroutine starts (fix round 1, finding 7).
	parent context.Context
	// phase is the probe's open (final review, ruling 3): held, opening
	// once the target's open marks it (OpenProbeEntry, before it releases
	// the hold), opened once the release is confirmed (ConfirmProbeOpened).
	// Retirement protection holds from opening on; idle protection ends only
	// at opened.
	phase probePhase
	// managed is set when the edge manages the opened probe's ready row: the
	// entry is an ordinary serving one from then on. The probe metadata is
	// kept, never cleared, so an older overlapping retirement pass still
	// validates against it (ruling 2).
	managed bool
	// seq counts every change of phase or managed. A retirement pass keeps
	// an entry whose seq moved since its snapshot: its registry read is
	// older than the change.
	seq uint64
}

// probePhase is a probe entry's open state.
type probePhase int

const (
	probeHeldPhase probePhase = iota
	probeOpening
	probeOpened
)

// probeSnap is the immutable probe metadata a retirement pass snapshots
// under a.mu and reads outside it (ruling 2).
type probeSnap struct {
	holdOp string
	phase  probePhase
	seq    uint64
}

// ConfirmProbeOpened records that the target's open released host's hold
// for op (final review, ruling 3): the entry, opening, is opened, and its
// idle protection ends. false when host has no opening probe entry of op.
func (a *Service) ConfirmProbeOpened(host, op string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	if e == nil || e.state != actAwake || e.probe == nil || e.probe.op != op || e.probe.managed || e.probe.phase < probeOpening {
		return false
	}
	if e.probe.phase != probeOpened {
		e.probe.phase = probeOpened
		e.probe.seq++
	}
	return true
}

// OpenProbeEntry marks host's awake probe entry for op opening, before the
// target's open releases the row's hold (Plan 3 Task 7; final review,
// ruling 3). RetireExcept keeps it while its row is held or, once released,
// ready, until the edge manages it (the open race); it stays exempt from
// idle stop until ConfirmProbeOpened. A retried open marks it again. false
// when host has no awake probe entry of op: the next request then wakes the
// host under the commit.
func (a *Service) OpenProbeEntry(host, op string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	if e == nil || e.state != actAwake || e.probe == nil || e.probe.op != op || e.probe.admitted == nil || e.probe.managed {
		return false
	}
	if e.probe.phase < probeOpening {
		e.probe.phase = probeOpening
		e.probe.seq++
	}
	return true
}

// ProbeWake runs one probe attempt for cfg.Host and waits for it. On
// success the host's entry is awake, holding the admitted session's process
// and replication, and stays held while the row is (RetireExcept).
func (a *Service) ProbeWake(ctx context.Context, cfg ProbeConfig) (ProbeResult, error) {
	if ctx == nil || cfg.Host == "" || cfg.Op == "" || cfg.HoldOp == "" || cfg.Deadline.IsZero() {
		return ProbeResult{}, hostwrite.ErrUnavailable
	}
	if !a.cfg.HostWrites.Enabled || a.cfg.Claims == nil || a.Routes == nil {
		return ProbeResult{}, hostwrite.ErrUnavailable
	}
	// The cut deadline alone, before any admission: the window opens only
	// at the first app start (brief step 4).
	if !a.now().Before(cfg.Deadline) {
		return ProbeResult{}, errors.Join(ErrProbeRefused, ErrAttemptOver)
	}
	rt, err := a.Routes.Get(cfg.Host)
	if err != nil {
		return ProbeResult{}, err
	}
	if rt == nil || !probeHeld(*rt, cfg.HoldOp) {
		return ProbeResult{}, errors.Join(ErrProbeRefused, errProbeRow)
	}
	e := &actEntry{host: cfg.Host, route: *rt, state: actWaking, ready: make(chan struct{}),
		probe: &probeEntry{op: cfg.Op, holdOp: cfg.HoldOp, parent: ctx,
			policy: startPolicy{MaxStarts: probeMaxStarts, Window: probeWindow30, Deadline: cfg.Deadline, Probe: true, Prepared: cfg.Prepared, Inputs: cfg.Inputs}}}
	a.mu.Lock()
	switch {
	case a.fences[cfg.Host] != 0:
		a.mu.Unlock()
		return ProbeResult{}, errors.Join(ErrProbeRefused, ErrFenced)
	case a.entries[cfg.Host] != nil:
		a.mu.Unlock()
		return ProbeResult{}, errors.Join(ErrProbeRefused, errProbeLive)
	}
	a.entries[cfg.Host] = e
	a.mu.Unlock()
	if probeWakePublished != nil {
		probeWakePublished()
	}
	go a.wake(e)
	<-e.ready
	// e.err and e.probe are written before e.ready closes.
	if e.err != nil {
		return ProbeResult{}, e.err
	}
	return probeResultOf(e.probe.admitted), nil
}

var (
	errProbeRow  = errors.New("activator: the registry row is not held by the probe's hold op")
	errProbeOp   = errors.New("activator: the claim is not the probe's op")
	errProbeLive = errors.New("activator: the host already has a live entry")
)

func probeHeld(rt registry.Route, holdOp string) bool {
	return registry.IsHold(rt.Status) && rt.HoldOp == holdOp
}

// probeRefused marks a probe's refusal before admission; other wakes'
// errors are unchanged.
func probeRefused(e *actEntry, err error) error {
	if e.probe == nil || err == nil {
		return err
	}
	return errors.Join(ErrProbeRefused, err)
}

// heldProbes snapshots, under a.mu, the immutable metadata of every probe
// entry RetireExcept may retire (seen), then reads their registry rows
// outside the lock from the snapshot alone, never dereferencing e.probe
// there (final review, ruling 2). held is the entries it keeps beyond keep:
// those whose row is still held by the probe's hold op, and opening or
// opened ones whose row the open made ready, which the edge's next reload
// manages (the open race). The edge manages only ready rows (edge.Reload),
// so a held probe is never in keep. A row that cannot be read keeps its
// entry for this reload. RetireExcept re-validates each entry's seq under
// its final lock: one that moved since the snapshot is kept.
func (a *Service) heldProbes(keep map[string]bool) (seen map[*actEntry]probeSnap, held map[*actEntry]bool) {
	seen, held = map[*actEntry]probeSnap{}, map[*actEntry]bool{}
	a.mu.Lock()
	for _, e := range a.entries {
		if e.probe != nil && !e.probe.managed && !keep[e.host] {
			seen[e] = probeSnap{holdOp: e.probe.holdOp, phase: e.probe.phase, seq: e.probe.seq}
		}
	}
	a.mu.Unlock()
	if retireBeforeRegistryRead != nil {
		retireBeforeRegistryRead()
	}
	if a.Routes == nil {
		return seen, held
	}
	for e, snap := range seen {
		rt, err := a.Routes.Get(e.host)
		if err != nil {
			log.Printf("activator: %s: the probe's row could not be read (%v); kept for this reload", e.host, err)
			held[e] = true
			continue
		}
		switch {
		case rt == nil:
		case probeHeld(*rt, snap.holdOp):
			held[e] = true
		case snap.phase >= probeOpening && rt.Status == "ready" && rt.HoldOp == "":
			held[e] = true
		}
	}
	return seen, held
}