package activator

import (
	"context"
	"errors"
	"log"
	"net/http"
	"time"

	"github.com/carlosframework/platform/internal/controlname"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/registry"
)

var (
	ErrFenceNotInstalled = errors.New("activator: maintenance fence not installed")
	ErrFenced            = errors.New("activator: maintenance fence active")
	ErrFenceGeneration   = errors.New("activator: maintenance generation refused")
	ErrStopUnproven      = errors.New("activator: stop evidence unavailable")
	ErrFenceHeld         = errors.New("activator: another maintenance fence is held")
)

// StopEvidence records local process and replica closure. It grants no lease,
// cgroup, database or resume authority.
type StopEvidence struct {
	host        string
	generation  uint64
	observation *stopObservation
}

func (e *StopEvidence) Matches(host string, generation uint64) bool {
	return e != nil && e.host == host && e.generation == generation && e.observation != nil && e.observation.complete()
}

type stopObservation struct {
	leaseReleased              bool
	mainStop, mainExited       bool
	sidecarStop, sidecarExited bool
	replicaClosed              bool
	uncertain                  bool
	// owedBy is the one retained failed wake whose unproven legacy start
	// (#292) is this observation's only uncertainty, or nil. Only that
	// wake's proof discharges it (reapUnprovenWake); any other source of
	// uncertainty clears it, and then the uncertainty stays.
	owedBy *unprovenWake
	// noStart: a wake refused before anything of it started (notStarted).
	// It is evidence for a host with no stop history, and leaves any
	// history the host has exactly as it was (rememberStopLocked).
	noStart bool
}

func (o *stopObservation) complete() bool {
	return o != nil && !o.uncertain && o.mainStop && o.mainExited && o.sidecarStop && o.sidecarExited && o.replicaClosed
}

// wakeStartedNothing reports whether a failed wake holds no process,
// replica or sidecar of its own: the condition for notStarted.
func wakeStartedNothing(w *wokenInstance) bool {
	return w != nil && w.proc == nil && w.rep == nil && w.sidecar == nil
}

// notStarted records a wake whose launch is known not to have started: no
// process, no sidecar and no replica of it exist, so every stage holds and
// nothing is uncertain (seeded pre-land review, P2).
func (o *stopObservation) notStarted() {
	o.mainStop, o.mainExited = true, true
	o.sidecarStop, o.sidecarExited = true, true
	o.replicaClosed = true
	o.uncertain = false
	o.noStart = true
}

// FenceAndStop installs a local fence before collecting stop evidence. Once this
// call installs its fence, cancellation or failure leaves admission closed.
// ErrFenceNotInstalled means this call could not install or confirm its fence;
// it makes no claim about admission, including a fence from an earlier call.
// ErrStopUnproven guarantees this call installed a fence, but proves no stop.
// A missing entry is not evidence that a prior process or replica stopped.
func (a *Service) FenceAndStop(ctx context.Context, host string, generation uint64) (*StopEvidence, error) {
	return a.fenceAndStop(ctx, host, generation, false, "")
}

// FenceAndStopIfUnfenced is FenceAndStop for a caller that must never take
// over somebody else's fence. FenceAndStop replaces a fence at a lower
// generation, and a caller that then lifts its own generation reopens
// admission the other operation closed. Here the check and the install
// happen under the same lock: if any fence is held for host, of any
// generation, it answers ErrFenceHeld, installs nothing and stops nothing.
// Otherwise it is FenceAndStop, answers included.
func (a *Service) FenceAndStopIfUnfenced(ctx context.Context, host string, generation uint64) (*StopEvidence, error) {
	return a.fenceAndStop(ctx, host, generation, true, "")
}

// FenceAndStopAs is FenceAndStopIfUnfenced for a caller that must be able
// to meet its own fence again: a retry after a lost answer, or after the
// caller's own later step failed. owner is the caller's persisted name for
// this one fence. Under the lock that installs, a held fence is the
// caller's only when it was installed by FenceAndStopAs with exactly this
// owner at exactly this generation; then the call is FenceAndStop under it
// (it re-collects the stop evidence). Any other held fence, of any
// generation and however installed, answers ErrFenceHeld, installs nothing
// and stops nothing. Fences live only in this process, so after a restart
// no fence is the caller's, whatever it persisted. An empty owner is
// ErrFenceGeneration.
func (a *Service) FenceAndStopAs(ctx context.Context, host string, generation uint64, owner string) (*StopEvidence, error) {
	if owner == "" {
		return nil, ErrFenceGeneration
	}
	return a.fenceAndStop(ctx, host, generation, true, owner)
}

func (a *Service) fenceAndStop(ctx context.Context, host string, generation uint64, exclusive bool, owner string) (*StopEvidence, error) {
	if a == nil || !controlname.ValidHostShape(host) || generation == 0 {
		return nil, ErrFenceGeneration
	}
	if ctx == nil {
		return nil, ErrFenceNotInstalled
	}
	deadline, ok := ctx.Deadline()
	if !ok || ctx.Err() != nil || !time.Now().Before(deadline) {
		return nil, ErrFenceNotInstalled
	}
	installed := false
	for {
		if err := a.lockFence(ctx); err != nil {
			if installed {
				return nil, ErrStopUnproven
			}
			return nil, ErrFenceNotInstalled
		}
		if a.seedExclusions[host] != nil {
			a.mu.Unlock()
			return nil, ErrFenceHeld
		}
		if preparation := a.tokenPreparations[host]; preparation != nil {
			a.mu.Unlock()
			select {
			case <-ctx.Done():
				if installed {
					return nil, ErrStopUnproven
				}
				return nil, ErrFenceNotInstalled
			case <-preparation:
				continue
			}
		}
		if a.fences == nil {
			a.fences = make(map[string]uint64)
		}
		prior := a.fences[host]
		own := owner != "" && prior == generation && a.fenceOwners[host] == owner
		// Only before this call installs: on a later round of the loop the
		// fence held is this call's own.
		if exclusive && !installed && prior != 0 && !own {
			a.mu.Unlock()
			return nil, ErrFenceHeld
		}
		if owner != "" && installed && !own {
			// An owned fence replaced while this call waited is no longer
			// this call's: it installed one, and proves nothing more.
			a.mu.Unlock()
			return nil, ErrStopUnproven
		}
		if prior > generation {
			a.mu.Unlock()
			return nil, ErrFenceGeneration
		}
		a.fences[host] = generation
		if owner != "" {
			if a.fenceOwners == nil {
				a.fenceOwners = make(map[string]string)
			}
			a.fenceOwners[host] = owner
		} else {
			delete(a.fenceOwners, host)
		}
		installed = true
		e := a.entries[host]
		if e == nil {
			outcome := a.stopHistory[host]
			a.mu.Unlock()
			if ctx.Err() != nil || !time.Now().Before(deadline) {
				return nil, ErrStopUnproven
			}
			if !outcome.complete() {
				return nil, ErrStopUnproven
			}
			return &StopEvidence{host: host, generation: generation, observation: outcome}, nil
		}
		var wait <-chan struct{}
		switch e.state {
		case actWaking:
			if e.wakeCancel != nil {
				e.wakeCancel()
			}
			wait = e.ready
		case actAwake:
			a.beginStopLocked(e, StopFenced)
			wait = e.stopped
		case actStopping:
			wait = e.stopped
		default:
			a.mu.Unlock()
			return nil, ErrStopUnproven
		}
		a.mu.Unlock()
		select {
		case <-ctx.Done():
			return nil, ErrStopUnproven
		case <-wait:
		}
	}
}

// LiftFence removes the fence FenceAndStop installed at generation, and only
// that one: a fence at any other generation stays, so a caller holding a stale
// generation cannot reopen admission somebody else closed. It reports whether
// it lifted anything. The caller owns the proof that nothing on the far side
// of the fence still needs it closed; the move worker lifts only once this
// box no longer holds the host's route (instance-moves design §6.1, G12).
func (a *Service) LiftFence(host string, generation uint64) bool {
	if a == nil || generation == 0 {
		return false
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.seedExclusions[host] != nil || a.fences[host] != generation {
		return false
	}
	delete(a.fences, host)
	delete(a.fenceOwners, host)
	return true
}

// LiftFenceAs is LiftFence for an owned fence: it lifts host's fence only
// when it is exactly the one FenceAndStopAs installed with owner at
// generation, and reports whether it lifted anything. A fence another
// operation installed at the same generation stays.
func (a *Service) LiftFenceAs(host string, generation uint64, owner string) bool {
	if a == nil || generation == 0 || owner == "" {
		return false
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.seedExclusions[host] != nil || a.fences[host] != generation || a.fenceOwners[host] != owner {
		return false
	}
	delete(a.fences, host)
	delete(a.fenceOwners, host)
	return true
}

func (a *Service) lockFence(ctx context.Context) error {
	timer := time.NewTicker(time.Millisecond)
	defer timer.Stop()
	for {
		if ctx.Err() != nil {
			return ErrStopUnproven
		}
		if a.mu.TryLock() {
			if ctx.Err() != nil {
				a.mu.Unlock()
				return ErrStopUnproven
			}
			return nil
		}
		select {
		case <-ctx.Done():
			return ErrStopUnproven
		case <-timer.C:
		}
	}
}
func (a *Service) wakeAllowed(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.fences[host] == 0
}

// Fenced reports whether a maintenance fence is installed for host right
// now. It is the layout pass's own affirmative read of the same map
// wakeAllowed checks, so a pass deciding whether it is safe to install its
// own fence can ask the question the way it is phrased in its own logic,
// rather than negating an admission check meant for something else.
func (a *Service) Fenced(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.fences[host] != 0
}

// FenceGeneration is the generation of the fence installed for host right
// now, or 0 when there is none. The layout pass uses it to tell a fence it
// kept itself from another operation's, since both answer ErrFenceHeld to
// its exclusive install.
func (a *Service) FenceGeneration(host string) uint64 {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.fences[host]
}

// StopUncertain reports whether host's process state cannot yet be trusted
// as stopped: either an entry for it is still live right now, in any state
// (waking, awake or stopping), or the last recorded stop observation never
// finished every check FenceAndStop requires for proof. A pass that meets
// ErrStopUnproven for an entry-less host still needs this before it can tell
// "nothing was ever running" apart from "something was running and its
// closure was never confirmed"; only the first of those is safe to treat as
// a completed stop.
func (a *Service) StopUncertain(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.entries[host] != nil {
		return true
	}
	outcome, ok := a.stopHistory[host]
	return ok && !outcome.complete()
}
func (a *Service) rememberStopLocked(host string, observation *stopObservation) {
	if a.stopHistory == nil {
		a.stopHistory = make(map[string]*stopObservation)
	}
	prior := a.stopHistory[host]
	if observation.noStart && prior != nil {
		// A wake that started nothing says nothing about what an earlier
		// one left: that record, its obligations and their owner, stay.
		return
	}
	if prior != nil && prior.uncertain {
		retained := *observation
		retained.uncertain = true
		// The prior's uncertainty stays one retained wake's alone only
		// while that same wake is the one recording.
		retained.owedBy = nil
		if observation.uncertain && observation.owedBy == prior.owedBy {
			retained.owedBy = prior.owedBy
		}
		observation = &retained
	}
	a.stopHistory[host] = observation
}
func maintenanceResponse(w http.ResponseWriter, r *http.Request) {
	edge.MarkAnswered(r)
	w.Header().Set("Cache-Control", "no-store")
	w.Header().Set("Retry-After", "5")
	http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
}

// cleanupFailedWake stops a failed wake's process and gives its lease back,
// recording the outcome on w.stopOutcome. rt is the route the wake spawned
// against, which the termination proof needs.
//
// The lease is given back only over a generation proven terminated (spec
// §7.4) whenever that proof is asked for: with generations on, when the
// start's own error said its kill was unproven (w.genUnproven, which only
// GenContainment produces and which can be on with generations off), or,
// whatever the switch, when the wake left a process (below). Unproven,
// ownership and the lease are kept in unprovenWakes, the sweep retries the
// stop and the proof (reapUnprovenWakesLocked), and no new wake of the host
// starts until it holds.
func (a *Service) cleanupFailedWake(ctx context.Context, rt registry.Route, w *wokenInstance, lease Lease) {
	stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
	defer cancel()
	// A wake that left a process is proven stopped before its lease goes,
	// whatever the switch says: two fail-closed exceptions to dormancy.
	// One is a leader that did not stop when asked (ErrStillRunning) or is
	// still alive (zero-downtime plan 3, task 2 fix): giving the lease back
	// would let another box wake the host while it still writes. The other
	// is an exec child whose leader did stop: with the flag off it ran in
	// carlos-exec@<host> with no g<n>, and what it forked may still run
	// there, so that cgroup is proven empty too (failedWakeStopped's
	// uncontainedHostEmpty; Task 15 fix round 1, ruling b; Astra IMPL-3
	// F6). A nested or unit-backed row's own proof is unchanged.
	if w.proc != nil {
		err := w.proc.Stop(stopCtx)
		w.stopOutcome.mainStop, w.stopOutcome.mainExited = err == nil, w.proc.Exited()
	}
	// An aborted or failed start of the host's original run unit that root
	// may still carry out (#292, Astra P1): cancelling the client's /start
	// leaves a start job systemd queued, and released over it, the lease
	// would let that job launch a writer no lease covers. The unit is
	// stopped (a stop job replaces the queued start) and proven gone on a
	// budget of its own first. Unproven, the lease is KEPT: retained below
	// and retried by the reaper, and at worst left to its TTL. A stall is
	// the safe outcome; two writers are not.
	legacy := w.proc == nil && w.legacyUncertain && !w.genUnproven
	if legacy {
		if err := a.legacyRunGone(rt); err != nil {
			log.Printf("activator: %s: the aborted start's unit is not proven stopped (%v); keeping the lease until it is", rt.Host, err)
			w.stopOutcome.mainExited = false
			w.stopOutcome.uncertain = true
			u := &unprovenWake{rt: rt, gen: w.gen, lease: lease, rep: w.rep, build: w.spawnIdentity,
				layout: w.launchLayout, legacy: true}
			// This wake's uncertainty is this retained start's, and its
			// proof discharges it (reapUnprovenWake; seeded pre-land
			// review, P2).
			w.stopOutcome.owedBy = u
			a.mu.Lock()
			if a.unprovenWakes == nil {
				a.unprovenWakes = make(map[string]*unprovenWake)
			}
			a.unprovenWakes[rt.Host] = u
			a.mu.Unlock()
			return
		}
		// #292's whole proof held: the start's unit is stopped and its
		// cgroup empty. That was this wake's only uncertainty; it reached
		// no sidecar, and no replica unless it holds one, which is closed
		// below (seeded pre-land review, P2).
		w.stopOutcome.mainStop, w.stopOutcome.mainExited = true, true
		w.stopOutcome.uncertain = false
		if w.sidecar == nil {
			w.stopOutcome.sidecarStop, w.stopOutcome.sidecarExited = true, true
		}
		if w.rep == nil {
			w.stopOutcome.replicaClosed = true
		}
	}
	if !legacy && (a.gens || w.genUnproven || w.proc != nil) && !a.failedWakeStopped(rt, w.gen, w.proc, w.launchLayout) {
		// Its leader did not exit (ErrStillRunning, a failed kill) or its
		// child is not proven empty.
		w.stopOutcome.mainExited = false
		w.stopOutcome.uncertain = true
		log.Printf("activator: %s: failed wake's generation %d is not proven stopped; keeping the lease until it is", rt.Host, w.gen.N)
		a.mu.Lock()
		if a.unprovenWakes == nil {
			a.unprovenWakes = make(map[string]*unprovenWake)
		}
		// A worker's staged identity moves with it (final review F12): the
		// failed wake's entry is about to go, and the generations file must
		// keep naming the build its retained process may still run.
		a.unprovenWakes[rt.Host] = &unprovenWake{rt: rt, gen: w.gen, proc: w.proc, lease: lease, rep: w.rep, build: w.spawnIdentity,
			layout: w.launchLayout}
		a.mu.Unlock()
		return
	}
	// A backend failure may conceal resources beyond the returned handle.
	// Successful cleanup of that handle does not erase the uncertainty.
	if w.rep != nil {
		closeCtx, closeCancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer closeCancel()
		w.stopOutcome.replicaClosed = w.rep.Close(closeCtx) == nil
		w.stopOutcome.leaseReleased = w.stopOutcome.replicaClosed
		return
	}
	w.stopOutcome.leaseReleased = a.giveBack(ctx, rt.Host, lease)
}