// Unit-backed tenants: hibernation for whole-app instances whose process
// manager stays systemd (carlos-app@<host>.service — DynamicUser,
// StateDirectory) rather than the agent. The activator still owns WHEN the
// process runs — stopped after the idle window, started when a request
// arrives, the edge holding that request while the unit boots and its
// upstream turns healthy — but the HOW is systemctl, behind the Systemctl
// seam so the state-machine tests need no systemd.
//
// The lease/replication seams are satisfied by inert implementations, on
// purpose — this is a deliberate division of labor, not a shortcut:
//
//   - A unit tenant's data lives in its DynamicUser-owned state dir
//     (/var/lib/carlos-app/<host>, 0700) which the agent, as user carlos,
//     cannot read — so the agent can neither restore it nor run litestream
//     over it. Its replication belongs to the box-wide carlos-litestream
//     unit (root), which streams changes continuously whether the tenant is
//     awake or asleep. That IS the snapshot-on-sleep story here: a stop only
//     happens after a full idle window, so streaming replication shipped the
//     final writes long before the unit goes down.
//
//     WHAT THIS ACTUALLY REQUIRES from the operator: box-wide litestream
//     (ops.FleetDBs) has no platform-wide convention for a unit tenant's
//     database filename inside its state dir — that filename is the APP's
//     own choice, unlike the dataDir/<host>.db convention exec children
//     get for free — so it can only replicate a unit tenant whose route
//     was added with an explicit `carlos add -backing unit -db <path>`.
//     `-hibernate` unit routes REQUIRE -db (cli.go's runAdd enforces it:
//     going unreplicated there is a genuine data-loss risk, the moment the
//     unit is ever stopped for idleness, not just a missed nicety).
//     Non-hibernating unit routes do NOT enforce it — -db is strongly
//     recommended but stays the operator's unchecked call, and a route
//     added without it is silently replicated by nobody. See
//     ops.FleetDBs's own doc comment for the mechanics.
//
//   - There is no lease to trade: the unit exists on exactly one box (its
//     state dir is that box's disk) and the box replicator never hands
//     ownership off, so wake/sleep is not a writer handoff. When unit
//     tenants go multi-box, a real lease slots in through this same seam.
//
// Pairing with carlos-app@.socket (see internal/host/units): with the
// tenant's socket unit enabled, systemd holds the listener bound across
// stops, so the edge's unix dial never hits the socket-absent window — and a
// connection while the service is stopped starts it (wake-on-connect). The
// activator's wake path here remains the primary one because it is
// health-gated: a broken boot becomes a clean 503, not a connect that hangs
// in the listen backlog.
package activator

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"os/exec"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// UnitName is the systemd unit backing a unit-backed tenant's route —
// carlos-app@<host>.service, the same instance name the adopt pass
// try-restarts for non-hibernating tenants.
func UnitName(host string) string { return "carlos-app@" + host + ".service" }

// ExecName is the metrics sample key for an exec-backed tenant's children —
// the delegated cgroup directory the activator spawns them into. Defined by
// internal/cgroup (which creates the directories); surfaced here next to
// UnitName so the console learns both of a host's legal metric keys from
// the same place.
func ExecName(host string) string { return cgroup.ExecName(host) }

// UnitBackend is the Backend for routes with Backing == registry.BackingUnit.
type UnitBackend struct {
	// Systemctl runs one systemctl verb against a unit — the Proc-style seam
	// wrapping systemctl that keeps hibernation tests systemd-free. Nil = the
	// real systemctl binary.
	Systemctl func(ctx context.Context, verb, unit string) error
	// OptDir is the on-box install root (the live/<host> symlinks) — same
	// layout as ExecBackend's, read for spawnVersion so build convergence
	// cycles awake unit tenants exactly like exec children.
	OptDir string
	// WaitHealthy blocks until the tenant answers HTTP on its upstream
	// (rt.Socket, else rt.Addr). Nil = the package's socket/TCP prober with
	// the standard spawnHealthTimeout. A seam so unit tests need no listener.
	WaitHealthy func(ctx context.Context, rt registry.Route) error
}

var _ Backend = (*UnitBackend)(nil)

func (b *UnitBackend) systemctl(ctx context.Context, verb, unit string) error {
	if b.Systemctl != nil {
		return b.Systemctl(ctx, verb, unit)
	}
	out, err := exec.CommandContext(ctx, "systemctl", verb, unit).CombinedOutput()
	if err != nil {
		return fmt.Errorf("systemctl %s %s: %v: %s", verb, unit, err, strings.TrimSpace(string(out)))
	}
	return nil
}

// LiveVersion mirrors ExecBackend.LiveVersion — the unit's ExecStart is the
// same live/<host> symlink the adopt pass repoints, so convergence reads the
// same layout, through the same one definition of which path segment is the
// version (see ExecBackend.LiveVersion for why that matters now).
func (b *UnitBackend) LiveVersion(host string) (string, error) {
	return release.LiveVersion(b.OptDir, host)
}

// Acquire returns the inert lease (see the package comment: replication
// ownership stays with the root carlos-litestream unit, and a single box's
// unit has no writer handoff to guard).
func (b *UnitBackend) Acquire(context.Context, registry.Route) (Lease, error) {
	return unitLease{}, nil
}

// Restore is a no-op, and NEVER fresh: the tenant's state dir persists on the
// box and the agent can't even read it (a DynamicUser's 0700 dir), so it
// cannot know whether a database exists in there — and "I can't tell" must
// read as "not a first boot", because the alternative is running an app's
// seed over its live data. Disaster recovery here is a litestream restore run
// by the operator as root. A unit tenant that wants first-boot initialization
// does it in its own unit (ExecStartPre), where the state dir is readable.
func (b *UnitBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil }

// Seed is unreachable for unit tenants (Restore never reports fresh) and is a
// no-op for the same reason it has to be: the agent can neither read the
// tenant's state dir nor run a process as its DynamicUser.
func (b *UnitBackend) Seed(context.Context, registry.Route) error { return nil }

// Start brings the unit up and holds until its upstream answers — the
// health gate that makes the edge-mediated wake strictly better than bare
// socket activation (a broken boot becomes a clean 503, not a connect
// parked in the listen backlog). A start against an already-running unit is
// systemd's no-op, which is what lets a fresh agent re-adopt tenants left
// awake by its previous incarnation. No live-symlink precheck: if the
// binary is missing the unit's exec fails, the health wait times out, and
// the stop below puts the unit back down.
func (b *UnitBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
	unit := UnitName(rt.Host)
	// Read the symlink before the start, same fail-safe direction as
	// ExecBackend: a concurrent repoint at worst records the older version
	// (one extra convergence cycle), never the newer (a stale child the
	// sweep would think converged).
	spawnVer, _ := b.LiveVersion(rt.Host)
	if err := b.systemctl(ctx, "start", unit); err != nil {
		return nil, "", err
	}
	if err := b.waitHealthy(ctx, rt); err != nil {
		stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
		defer cancel()
		_ = b.systemctl(stopCtx, "stop", unit)
		return nil, "", err
	}
	return &svcUnitProc{b: b, unit: unit}, spawnVer, nil
}

func (b *UnitBackend) waitHealthy(ctx context.Context, rt registry.Route) error {
	if b.WaitHealthy != nil {
		return b.WaitHealthy(ctx, rt)
	}
	switch {
	case rt.Socket != "":
		return waitHealthyCtx(ctx, rt.Socket, spawnHealthTimeout)
	case rt.Addr != "":
		return waitHealthyAddrCtx(ctx, rt.Addr, spawnHealthTimeout)
	}
	return fmt.Errorf("route %s has neither socket nor addr to health-check", rt.Host)
}

// NextDue reports "nothing scheduled": the due-time probe is admin-gated by
// the instance's CARLOS_ADMIN_TOKEN, and a unit tenant's env file
// (/etc/carlos/apps/<host>.env) is root-0600 — unreadable to the agent by
// design. Unit tenants therefore sleep until a request; the due-time index
// stays exec-only until the contract grows an agent-readable channel.
func (b *UnitBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil }

// HasSidecar: sidecar co-spawn is exec-child machinery (the sidecar is the
// agent's own child, CARLOS_*-scrubbed); a unit tenant that wants a sidecar
// runs it inside its own unit.
func (b *UnitBackend) HasSidecar(registry.Route) bool { return false }

func (b *UnitBackend) StartSidecar(context.Context, registry.Route) (Proc, error) {
	return nil, errors.New("unit-backed tenants have no agent-managed sidecars")
}

// StartWorker: worker spawn is exec-child machinery exactly like sidecar
// co-spawn — a unit tenant that wants a worker runs it inside its own unit.
func (b *UnitBackend) StartWorker(context.Context, registry.Route, map[string]string) (Proc, string, error) {
	return nil, "", errors.New("unit-backed tenants have no agent-managed workers")
}

// svcUnitProc adapts a UnitBackend-managed systemd unit (carlos-app@<host>.service,
// a whole-app tenant whose process manager stays systemd) to the Proc seam.
// Named apart from the isolated-exec unitProc in unitproc.go, which is a
// different thing entirely: this one wraps a tenant unit the operator
// authored, that one is a nested exec instance the activator itself spawns
// through the root instance-control channel.
type svcUnitProc struct {
	b    *UnitBackend
	unit string

	mu      sync.Mutex
	stopped bool
}

// Stop is `systemctl stop` — systemd runs the unit's own SIGTERM drain and
// the ctx bounds how long we wait for the job.
func (p *svcUnitProc) Stop(ctx context.Context) error {
	err := p.b.systemctl(ctx, "stop", p.unit)
	p.mu.Lock()
	p.stopped = true
	p.mu.Unlock()
	return err
}

// Exited reports false until Stop: crash supervision belongs to systemd
// (Restart=always on the unit), so a crashed tenant restarts under its unit
// without the activator noticing — deliberately, because Exited is polled
// under a.mu on every request and sweep tick, and an exec'd `systemctl
// is-active` there would put a fork on the hot path. The trade-off: a unit
// wedged past its start rate limit looks awake until its idle window stops
// it (the edge's dials 502 meanwhile) — operator territory, exactly as it
// is for a non-hibernating unit today.
func (p *svcUnitProc) Exited() bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.stopped
}

// unitLease / unitRep: the inert lease (rationale in the package comment).
type unitLease struct{}

func (unitLease) Replicate(context.Context, string, time.Duration) (Rep, error) {
	return unitRep{}, nil
}
func (unitLease) Release(context.Context) error { return nil }

type unitRep struct{}

func (unitRep) Err() error                  { return nil }
func (unitRep) Close(context.Context) error { return nil }

// waitHealthyAddr is waitHealthy for TCP upstreams (rt.Addr — unit tenants
// that can't accept a unix listener yet; see the carlos-app@.socket unit's
// fallback note).
func waitHealthyAddr(addr string, timeout time.Duration) error {
	deadline := time.Now().Add(timeout)
	client := &http.Client{Timeout: 3 * time.Second}
	var last error
	var poll time.Duration
	for time.Now().Before(deadline) {
		resp, err := client.Get("http://" + addr + "/")
		if err == nil {
			resp.Body.Close()
			if resp.StatusCode < 500 {
				return nil
			}
			last = fmt.Errorf("unhealthy: HTTP %d", resp.StatusCode)
		} else {
			last = err
		}
		poll = nextHealthPoll(poll)
		time.Sleep(poll)
	}
	return fmt.Errorf("instance not healthy within %s: %v", timeout, last)
}

func waitHealthyAddrCtx(ctx context.Context, addr string, timeout time.Duration) error {
	done := make(chan error, 1)
	go func() { done <- waitHealthyAddr(addr, timeout) }()
	select {
	case err := <-done:
		return err
	case <-ctx.Done():
		return ctx.Err()
	}
}

// SplitBackend composes the two production backends behind the one Backend
// seam the state machine sees: each call dispatches on the route's Backing,
// so exec children and unit tenants share the singleflight wakes, idle
// clock, due index, and convergence sweep without forking any of it.
type SplitBackend struct {
	Exec Backend // rt.Backing == "" — agent-spawned children (ExecBackend)
	Unit Backend // rt.Backing == registry.BackingUnit (UnitBackend)
}

var _ Backend = (*SplitBackend)(nil)

func (b *SplitBackend) pick(rt registry.Route) Backend {
	if rt.Backing == registry.BackingUnit {
		return b.Unit
	}
	return b.Exec
}

func (b *SplitBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, error) {
	return b.pick(rt).Acquire(ctx, rt)
}

// AdmitConfig dispatches like every other call: on Backing alone. A unit
// tenant's config is the root adopt pass's, so only the exec half is gated.
func (b *SplitBackend) AdmitConfig(ctx context.Context, rt registry.Route) error {
	if g, ok := b.pick(rt).(configAdmitter); ok {
		return g.AdmitConfig(ctx, rt)
	}
	return nil
}

// SeedPlaced dispatches on Backing, like every other call. A unit tenant
// never carries a seed.
func (b *SplitBackend) SeedPlaced(ctx context.Context, rt registry.Route, rep Rep) error {
	if p, ok := b.pick(rt).(seedPlacer); ok {
		return p.SeedPlaced(ctx, rt, rep)
	}
	return nil
}
func (b *SplitBackend) Restore(ctx context.Context, rt registry.Route) (bool, error) {
	return b.pick(rt).Restore(ctx, rt)
}
func (b *SplitBackend) Seed(ctx context.Context, rt registry.Route) error {
	return b.pick(rt).Seed(ctx, rt)
}
func (b *SplitBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
	return b.pick(rt).Start(ctx, rt)
}
func (b *SplitBackend) NextDue(ctx context.Context, rt registry.Route) (int64, error) {
	return b.pick(rt).NextDue(ctx, rt)
}
func (b *SplitBackend) HasSidecar(rt registry.Route) bool { return b.pick(rt).HasSidecar(rt) }

// StartGen dispatches like Start. A side without generation support (the
// unit side: systemd binds a unit tenant's canonical socket) is started
// with Start and gets the canonical handle.
func (b *SplitBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) {
	if g, ok := b.pick(rt).(GenBackend); ok {
		return g.StartGen(ctx, rt, n, perGen)
	}
	proc, ver, err := b.pick(rt).Start(ctx, rt)
	// The handle goes back with an error too, as ExecBackend.StartGen's does
	// (P2-13).
	return proc, canonicalGen(rt, n, ver), err
}

// NextDueVia dispatches like NextDue; a side without generation support
// answers on rt's own socket.
func (b *SplitBackend) NextDueVia(ctx context.Context, rt registry.Route, socket string) (int64, error) {
	if g, ok := b.pick(rt).(GenBackend); ok {
		return g.NextDueVia(ctx, rt, socket)
	}
	return b.pick(rt).NextDue(ctx, rt)
}

// KillGen dispatches like Start, for an exec row only (zero-downtime spec
// §8.2 step 1). A unit-backed row has no generation child to kill: systemd
// owns the unit tenant's one process tree. An exec half without KillGen (a
// test double) cannot prove a kill either. Both refuse, wrapping
// errNoGenContainment, so a caller never believes a kill it could not make.
func (b *SplitBackend) KillGen(ctx context.Context, rt registry.Route, n uint64) error {
	if rt.Backing == registry.BackingUnit {
		return fmt.Errorf("%s: a unit-backed row has no generation %d to kill: %w", rt.Host, n, errNoGenContainment)
	}
	k, ok := b.Exec.(interface {
		KillGen(ctx context.Context, rt registry.Route, n uint64) error
	})
	if !ok {
		return fmt.Errorf("%s: this backend cannot kill generation %d: %w", rt.Host, n, errNoGenContainment)
	}
	return k.KillGen(ctx, rt, n)
}

// LegacyRunGone is legacyRunProver for the composed backend (#292): an
// exec row's nested run unit is the exec side's to stop and prove. A
// unit-backed row never starts a carlos-instance@ unit, so it refuses
// rather than claiming a proof.
func (b *SplitBackend) LegacyRunGone(ctx context.Context, rt registry.Route) error {
	if rt.Backing == registry.BackingUnit {
		return fmt.Errorf("%s: a unit-backed row has no instance run unit: %w", rt.Host, errNoLegacyProof)
	}
	p, ok := b.Exec.(legacyRunProver)
	if !ok {
		return fmt.Errorf("%s: %w", rt.Host, errNoLegacyProof)
	}
	return p.LegacyRunGone(ctx, rt)
}

// Layout implements Classifier for the composed backend: a unit-backed row
// is always today's flat unit shape (the exec-only layouts do not apply to
// it), and an exec row defers to the exec side's own classification.
func (b *SplitBackend) Layout(rt registry.Route) (Layout, string) {
	if rt.Backing == registry.BackingUnit {
		return LayoutFlat, ""
	}
	if c, ok := b.Exec.(Classifier); ok {
		return c.Layout(rt)
	}
	return LayoutFlat, ""
}

func (b *SplitBackend) StartSidecar(ctx context.Context, rt registry.Route) (Proc, error) {
	return b.pick(rt).StartSidecar(ctx, rt)
}
func (b *SplitBackend) StartWorker(ctx context.Context, rt registry.Route, env map[string]string) (Proc, string, error) {
	return b.pick(rt).StartWorker(ctx, rt, env)
}

// StartCmd dispatches like every other call; a unit-backed row has no
// nested instance to run a command in.
func (b *SplitBackend) StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error) {
	if c, ok := b.pick(rt).(CmdBackend); ok {
		return c.StartCmd(ctx, rt, spec)
	}
	return nil, fmt.Errorf("%w: %s is unit-backed", ErrCmdNotNested, rt.Host)
}

// StartWorkerStaged forwards spec §7.9's staged spawn to the backend the
// route picks. One without the capability (UnitBackend, which refuses
// workers anyway) gets today's StartWorker, and its identity is the version
// alone, which no recycle can be verified against. That one cannot ask ctx's
// exec gate at its start, so it is asked here, immediately before
// StartWorker, as startWorker asks it for a backend without the capability
// (Astra R3 §5 item 4).
func (b *SplitBackend) StartWorkerStaged(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) {
	if s, ok := b.pick(rt).(stagedWorkerStarter); ok {
		return s.StartWorkerStaged(ctx, rt, env, onto)
	}
	if err := runExecGate(ctx); err != nil {
		return nil, release.StagedIdentity{}, err
	}
	p, ver, err := b.pick(rt).StartWorker(ctx, rt, env)
	return p, release.StagedIdentity{Version: ver}, err
}

// LiveVersion forwards to the exec side's reader when it has one — both
// backings share the live/<host> symlink layout, so one reader serves the
// convergence sweep for every host. (New picks this up via the optional
// interface, exactly as it does for a bare ExecBackend.)
func (b *SplitBackend) LiveVersion(host string) (string, error) {
	if lv, ok := b.Exec.(interface {
		LiveVersion(host string) (string, error)
	}); ok {
		return lv.LiveVersion(host)
	}
	return "", nil
}