// Package caps: the converge pass (spec §3c, §4b, §4d, §4e).
//
// Converge is the only thing in this repo that writes root-owned systemd
// unit text, so every rail it carries is here rather than spread across its
// callers: the three-state rule, the marker check, the box-wide claims, the
// DynamicUser-flip refusal, and the socket restart ordering. Its only
// contact with the outside world is through Ops, so its tests touch no
// systemd, no /etc, no /run and no /var/lib.
package caps

import (
	"bytes"
	"errors"
	"fmt"
	"os"
	"os/exec"
	"os/user"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
)

const (
	// DefaultEtcDir is where systemd reads unit configuration; DefaultStateRoot
	// is the parent of the template's StateDirectory=carlos-app/%i; DefaultRunDir
	// is the parent of RuntimeDirectory=<name>. All three are overridable ONLY
	// so tests can point them at a t.TempDir() — nothing in a manifest or a
	// registry row ever reaches them.
	DefaultEtcDir    = "/etc/systemd/system"
	DefaultStateRoot = "/var/lib/carlos-app"
	DefaultRunDir    = "/run"

	// DropInName is the one generated file per unit. `10-` sorts before an
	// operator's `90-operator.conf`, which this package never writes and never
	// reads (spec §3c).
	DropInName = "10-carlos-capabilities.conf"
)

// Ops is every piece of the outside world this package touches, injected so
// the converge tests need no systemd, no root, and no /etc — the same seam
// shape activator.UnitBackend.Systemctl uses.
type Ops struct {
	Authorize func() error

	CheckHost  func(string) error
	EtcDir     string                        // "" = DefaultEtcDir
	StateRoot  string                        // "" = DefaultStateRoot
	RunDir     string                        // "" = DefaultRunDir
	Systemctl  func(args ...string) error    // nil = the real systemctl binary
	Query      func(args ...string) bool     // nil = the real systemctl; exit 0 = yes
	EnsureUser func(name string) error       // nil = the real, idempotent useradd
	Logf       func(format string, a ...any) // nil = fmt.Printf
}

// Plan is one route's converge input.
type Plan struct {
	Host      string
	App       string
	Set       *Set // NIL means the manifest did not resolve — LEAVE THE DROP-INS ALONE
	Grants    Grants
	Hibernate bool

	// DeferRestart says the CALLER will start this route's service itself,
	// after this converge returns, so nothing here may bring it up.
	//
	// It exists for exactly one caller: release.adoptHost's version-change
	// converge, which runs BEFORE the live symlink moves. Without it a daemon
	// version change cycles the unit twice, and the first cycle runs the OLD
	// binary under the NEW unit shape — a boot of a combination nobody
	// shipped, on a mail server, for the width of a symlink swap. With it the
	// service stays down across the swap and comes up exactly once, on the
	// new binary, through Restart.
	//
	// It only ever affects a DAEMON plan's service verbs (step 8's start,
	// step 9's restart, and socketSequence's trailing start). A non-daemon
	// route's service is not started here with or without it, and the socket
	// unit's own verbs are never deferred — a listener the caller is not
	// going to re-bind must be re-bound here.
	//
	// The grant-drift pass (host.capabilityPass) leaves this FALSE and must:
	// restarting the live binary to pick up a new unit shape is the entire
	// point of that pass, and nothing there follows up with a Restart.
	DeferRestart bool
}

// Snapshot is what was on disk before Converge wrote, enough to put it back.
//
// A half whose file exists WITHOUT the marker is recorded as "not had": this
// pass never wrote it, so putting it back is not this pass's business — and
// Restore is marker-checked too, so a "not had" half can never turn into the
// deletion of somebody's hand-written drop-in.
//
// Daemon travels with the snapshot because putting the BYTES back is only
// half of putting the box back: a restored ListenStream that systemd never
// re-binds restores nothing, and re-binding it needs the same stop → restart
// → start ordering Converge uses, which needs to know whether the service is
// supposed to be running. See Restore.
type Snapshot struct {
	Host                  string
	Service, Socket       []byte
	ServiceHad, SocketHad bool
	Daemon                bool
}

// Result is what Converge did.
type Result struct {
	Prev    Snapshot
	Changed bool // any byte written or removed; the caller need not daemon-reload, Converge already did
}

// ServiceUnit and SocketUnit are the two per-instance units the template
// family instantiates for a route host.
func ServiceUnit(host string) string { return "carlos-app@" + host + ".service" }
func SocketUnit(host string) string  { return "carlos-app@" + host + ".socket" }

// DropInPaths returns the two generated file paths for a host. etcDir "" is
// DefaultEtcDir. The host is NOT validated here — every caller that joins
// these paths validates it first with validHost; this function is the naming
// rule alone, so tests and advisories can name a path without re-deriving it.
func DropInPaths(etcDir, host string) (service, socket string) {
	if etcDir == "" {
		etcDir = DefaultEtcDir
	}
	return filepath.Join(etcDir, ServiceUnit(host)+".d", DropInName),
		filepath.Join(etcDir, SocketUnit(host)+".d", DropInName)
}

// Converge materializes p and returns what it replaced.
//
// The three states are NOT interchangeable, and this is the same rule
// appconfig.Sync uses for exactly the same reason: a bucket hiccup must not
// strip :25 off a running mail server.
//   - p.Set == nil (absent or unreadable staged manifest): leave every drop-in
//     exactly as it is, log one line, return Changed=false, nil error.
//   - p.Set non-nil but INVALID for this binary (a capability name or a "v"
//     it does not know — reachable via a platform-binary rollback onto an
//     already-staged version, since ensureStaged short-circuits on a staged
//     artifact and never re-reads its manifest): same leave-alone posture,
//     plus an error, so a version-change caller refuses the version instead
//     of pointing live at something it cannot materialize.
//   - p.Set non-nil, renders bytes: write them if they differ.
//   - p.Set non-nil, renders nil: REMOVE the generated file (marker-checked).
//
// Two different kinds of error come back, and they are not the same claim:
//
//   - a REFUSAL (every check above, plus the claims and state-dir guards) is
//     Changed=false with nothing written, having already logged one line
//     naming the operator's fix. Callers that merely survey the box (the
//     hourly grant-drift pass) can log-and-continue; callers on the
//     version-change path let it refuse the version.
//   - a mid-pass IO or systemd FAILURE (a write, a daemon-reload, the socket
//     sequence) rolls the bytes back to Prev so that "the bytes on disk say
//     converged" stays a sound invariant — if systemd never loaded them, they
//     must not be on disk claiming it did, or the next pass would find no
//     diff and never retry. Changed is then false, because the disk is back
//     where it started. It is true only when even the roll-back failed, which
//     is the honest answer: bytes moved and this pass could not undo it.
//
// Failures AFTER a successful reload — an `enable`, a `start`, a daemon's
// `restart` — deliberately do NOT roll the bytes back. systemd has the new
// configuration and it is correct; the failure is the app's. Rolling back
// there would flap the unit text between two shapes on every pass, and the
// version-change caller already owns that case (it rolls back the symlink and
// calls Restore).
//
// That hand-off has a signal, and a caller that ignores it leaves the NEW
// version's unit shape on disk under the OLD binary forever — the next pass
// finds no diff, snapshots those bytes as Prev, and "restores" them:
//
//	Changed == true together with a non-nil error means BYTES ARE STILL MOVED
//	and this function is not going to move them back.
//
// A version-change caller must answer that with Restore(o, res.Prev). Changed
// == false with an error is a refusal (or a roll-back this function already
// performed) and needs nothing: Prev is then either empty — because the
// refusal came before the drop-ins were even read — or already on disk. Do
// not call Restore unconditionally on a refusal; an empty Prev says "this
// route had no generated drop-ins", and Restore honours that by REMOVING the
// ones it finds.
//
// Concurrency, stated rather than solved: two converges in one process (the
// hourly RunAdopt and the version-change converge inside Pass) are
// sequential. `carlos adopt` run by hand while carlos-adopt.timer fires is
// two processes, and there the atomic rename makes the FILE safe — a reader
// sees old or new bytes, never a torn file — but NOT the systemctl sequence:
// two passes could interleave a stop and a start. The repo already accepts
// exactly this for appconfig.Sync, and the failure mode is a unit that ends
// the minute running the right unit text, so it is accepted here too.
func Converge(o Ops, p Plan) (Result, error) {
	var res Result
	if o.CheckHost != nil {
		if err := o.CheckHost(p.Host); err != nil {
			return res, err
		}
	}
	res.Prev.Host = p.Host
	if err := o.authorizeMutation(); err != nil {
		return res, err
	}

	// The host becomes a filename under /etc/systemd/system, written as
	// root. Same guard, same reason, same trust level as appconfig.syncHost's
	// at :293 — a registry row is trusted, but not to place a root-owned file
	// wherever it likes.
	if err := validHost(p.Host); err != nil {
		o.logf("  %q: %v", p.Host, err)
		return res, err
	}
	svcPath, sockPath := DropInPaths(o.etcDir(), p.Host)

	if p.Set == nil {
		// State two of three. NOT a removal: a capability is only ever
		// removed because a manifest that ACTUALLY RESOLVED no longer
		// declares it, or because a grant was revoked.
		o.logf("  %s: no capability manifest resolved this pass — leaving its drop-ins exactly as they are", p.Host)
		return res, nil
	}

	// The staged manifest is re-validated here even though ship and adopt
	// already gate it: ensureStaged returns early when the artifact is
	// already on disk, so a ROLLED-BACK platform binary can meet a staged
	// version whose capability block it does not understand. This is the
	// backstop, and it falls the same way as an unreadable manifest.
	if err := ValidCapabilities(p.App, p.Set); err != nil {
		o.logf("  %s: %v; leaving its drop-ins exactly as they are", p.Host, err)
		return res, err
	}

	// A daemon the platform is asked to keep running and to stop when idle is
	// a contradiction, and hibernation for these apps means an MX that sleeps.
	if p.Set.Daemon && p.Hibernate {
		err := fmt.Errorf("capability daemon and hibernation contradict")
		o.logf("  %s: %v — nothing written and no unit state changed; fix it with: carlos route --host %s --hibernate=false", p.Host, err, p.Host)
		return res, err
	}

	hostRole := p.Set.HostRole && p.Grants.HostRole
	res.Prev.Daemon = p.Set.Daemon

	// Not a refusal — a granted port on a hibernating route is a legitimate
	// shape (socket activation is the wake-on-connect half of hibernation) —
	// but it is a wake path that does NOT go through the activator, so the
	// edge's health gate does not cover it: a broken boot answers a connection
	// on :25 by parking it in the backlog rather than by returning a clean
	// 503. One line per pass, for as long as the combination exists, because
	// the operator who granted the port did not necessarily choose this.
	if p.Hibernate {
		for _, tok := range grantedPortTokens(p) {
			o.logf("  %s: %s is granted on a hibernating route — a connection to that port starts the unit directly, bypassing the activator's health-gated wake; the edge's clean 503 on a broken boot does not cover this path", p.Host, tok)
		}
	}

	// Read both halves BEFORE any refusal that needs to know what this route
	// already claims, and before anything is written. Nothing here changes
	// state, so an early read costs a refusal nothing.
	curSvc, svcExists, svcGenerated, err := readDropIn(svcPath)
	if err != nil {
		o.logf("  %s: read %s: %v", p.Host, svcPath, err)
		return res, err
	}
	curSock, sockExists, sockGenerated, err := readDropIn(sockPath)
	if err != nil {
		o.logf("  %s: read %s: %v", p.Host, sockPath, err)
		return res, err
	}
	if svcExists && svcGenerated {
		res.Prev.Service, res.Prev.ServiceHad = curSvc, true
	}
	if sockExists && sockGenerated {
		res.Prev.Socket, res.Prev.SocketHad = curSock, true
	}

	if err := o.checkStateDirShape(p, hostRole, curSvc, svcExists && svcGenerated); err != nil {
		return res, err
	}
	if err := o.checkClaimsOnDisk(p, curSvc, svcExists && svcGenerated); err != nil {
		return res, err
	}

	// The user must exist before a User= line naming it does: the unit would
	// fail to start and Restart=always would spin it.
	if hostRole {
		user := AppUser(p.App)
		if err := o.ensureUser(user); err != nil {
			o.logf("  %s: create system user %s: %v — refusing the whole plan rather than write a User= line for a user that does not exist", p.Host, user, err)
			return res, err
		}
	}

	svcBody, err := RenderService(p.App, p.Set, p.Grants)
	if err != nil {
		o.logf("  %s: render service drop-in: %v; leaving its drop-ins exactly as they are", p.Host, err)
		return res, err
	}
	sockBody, err := RenderSocket(p.Set, p.Grants)
	if err != nil {
		o.logf("  %s: render socket drop-in: %v; leaving its drop-ins exactly as they are", p.Host, err)
		return res, err
	}

	svcChanged, svcErr := o.materialize(p.Host, svcPath, svcBody, curSvc, svcExists, svcGenerated)
	var sockChanged bool
	var sockErr error
	if svcErr == nil {
		sockChanged, sockErr = o.materialize(p.Host, sockPath, sockBody, curSock, sockExists, sockGenerated)
	}
	res.Changed = svcChanged || sockChanged
	if err := errors.Join(svcErr, sockErr); err != nil {
		// A half-written pass is the divergence this whole rollback exists to
		// prevent: one unit converged, the other not, and the bytes on disk
		// claiming both are done. Put them back so the next pass sees a diff
		// and retries the WHOLE change.
		res.Changed = o.rollbackBytes(res.Prev, res.Changed)
		return res, err
	}

	// systemd reads nothing from disk until it is told to. One reload, and
	// only when something actually changed — this is what makes an unchanged
	// pass free. It happens BEFORE any restart or start below, without
	// exception: a restart against configuration systemd has not re-read is a
	// pass that looks like it worked and changed nothing.
	if res.Changed {
		if err := o.systemctl("daemon-reload"); err != nil {
			o.logf("  %s: daemon-reload: %v", p.Host, err)
			res.Changed = o.rollbackBytes(res.Prev, res.Changed)
			return res, err
		}
	}

	sockLive := sockBody != nil
	st := o.applyUnitState(p, sockLive, svcChanged, sockChanged)
	if st.err != nil {
		if st.diverged {
			// The socket sequence is the one place where systemd's running
			// state can end up disagreeing with bytes that read "correct".
			// Roll them back so the next pass retries rather than concluding
			// there is nothing to do.
			res.Changed = o.rollbackBytes(res.Prev, res.Changed)
		}
		return res, st.err
	}
	if res.Changed {
		what := "capability drop-ins converged"
		if st.restarted {
			what += " and the unit restarted"
		}
		o.logf("  %s: %s", p.Host, what)
	}
	return res, nil
}

// rollbackBytes puts both halves back to what this pass found, WITHOUT
// touching unit state — the caller is already in a failure path and the point
// is only to stop the disk from claiming a change systemd never took. It
// reports whether any byte is still moved afterwards, which is what Converge
// returns as Changed: false when the disk is back where it started, true when
// even the roll-back failed.
func (o Ops) rollbackBytes(prev Snapshot, changed bool) bool {
	if !changed {
		return false
	}
	svcChanged, sockChanged, errs := o.applyHalves(prev)
	if len(errs) > 0 {
		o.logf("  %s: could not roll the capability drop-ins back after a failed pass (%v) — the next pass will find them mid-change", prev.Host, errors.Join(errs...))
		return true
	}
	if svcChanged || sockChanged {
		if err := o.systemctl("daemon-reload"); err != nil {
			o.logf("  %s: daemon-reload after roll-back: %v", prev.Host, err)
		}
		o.logf("  %s: rolled the capability drop-ins back to what this pass found, so the next pass retries the whole change", prev.Host)
	}
	return false
}

// applyHalves writes a Snapshot's bytes over whatever is on disk now — the
// shared body of Restore and rollbackBytes. Marker-checked in both
// directions, exactly as Converge is: a half that was not ours when the
// snapshot was taken is not ours to write over or delete now either.
func (o Ops) applyHalves(want Snapshot) (svcChanged, sockChanged bool, errs []error) {
	svcPath, sockPath := DropInPaths(o.etcDir(), want.Host)
	halves := []struct {
		path string
		body []byte
		had  bool
		out  *bool
	}{
		{svcPath, want.Service, want.ServiceHad, &svcChanged},
		{sockPath, want.Socket, want.SocketHad, &sockChanged},
	}
	for _, half := range halves {
		body := half.body
		if !half.had {
			body = nil
		}
		cur, exists, generated, err := readDropIn(half.path)
		if err != nil {
			o.logf("  %s: read %s: %v", want.Host, half.path, err)
			errs = append(errs, err)
			continue
		}
		did, err := o.materialize(want.Host, half.path, body, cur, exists, generated)
		if err != nil {
			errs = append(errs, err)
		}
		*half.out = did
	}
	return svcChanged, sockChanged, errs
}

// grantedPortTokens is the declared-AND-granted port set, as canonical grant
// tokens, sorted — the ports that actually materialize.
func grantedPortTokens(p Plan) []string {
	var toks []string
	for _, decl := range p.Set.Ports {
		if p.Grants.Ports[decl.Port] {
			toks = append(toks, PortToken(decl.Port))
		}
	}
	sort.Strings(toks)
	return toks
}

// unitState is what applyUnitState did and how badly it failed. diverged is
// the distinction that matters: a failed SOCKET SEQUENCE means systemd's
// running listener no longer matches the bytes on disk, and only the caller
// rolling those bytes back makes the next pass retry.
type unitState struct {
	restarted bool
	diverged  bool
	err       error
}

// repaired announces a drift repair — a verb this pass issued because
// systemd's RUNNING state disagreed with what the route declares, not because
// any byte changed.
//
// It is deliberately not folded into Converge's trailing "converged" line,
// which is gated on res.Changed: the repairs below fire on passes where
// nothing was written at all, and reporting them there made the most
// surprising thing this pass does — starting a unit an operator stopped by
// hand ten minutes ago to debug it — completely silent. Whoever stopped that
// unit needs the hourly log to say which verb ran and why.
//
// Only a repair that SUCCEEDED is announced; a failed one is already reported
// by applyUnitState's error line, and "started it" is not a claim to make
// about a start that did not.
func (o Ops) announceRepair(host string, err error, format string, a ...any) {
	if err != nil {
		return
	}
	o.logf("  %s: %s", host, fmt.Sprintf(format, a...))
}

// applyUnitState is steps 7-9: the socket restart ordering, the enable/start
// a daemon route's drift repair needs, and the service restart. Every verb
// here runs AFTER Converge's daemon-reload, never before it.
//
// p.DeferRestart suppresses every verb that would BRING A DAEMON UP — step
// 8's start, step 9's restart, and socketSequence's trailing start — and
// nothing else. The socket unit's own verbs, the enables, and the service
// STOP still happen: a listener has to be re-bound here because the caller
// will not do it, and leaving the service down across the caller's symlink
// swap is the entire point. See Plan.DeferRestart.
func (o Ops) applyUnitState(p Plan, sockLive, svcChanged, sockChanged bool) unitState {
	svcUnit, sockUnit := ServiceUnit(p.Host), SocketUnit(p.Host)
	var st unitState
	note := func(err error) {
		if err != nil && st.err == nil {
			st.err = err
		}
	}

	restartedService := false

	// Step 7. A running service holds the bound fd, so restarting the socket
	// unit under it races into EADDRINUSE and the MX comes back unbound.
	//
	// This runs when the socket drop-in was REMOVED as well as when it was
	// rewritten: systemd is still holding the listener the removed file asked
	// for, and "a grant is not a promise" (spec §7) means revoking port:25
	// has to actually drop :25 — a rewrite nobody reloads into a running
	// socket unit is a revoke that did nothing.
	if sockChanged {
		err := o.socketSequence(p.Host, p.Set.Daemon, p.DeferRestart, "restart")
		note(err)
		if err != nil {
			st.diverged = true
		}
		restartedService = true
	}

	// Step 8. `systemctl enable` writes the boot-time .wants symlink and
	// bootstrap does not enable tenant socket units — without this the
	// ListenStream binds nothing after a reboot. Both verbs are gated on a
	// query rather than run unconditionally, so a converged box's hourly pass
	// changes no state at all (which is what makes "an unchanged pass does
	// nothing" true for a daemon route, not just for a quiet one).
	if sockLive {
		if !o.query("is-enabled", sockUnit) {
			err := o.systemctl("enable", sockUnit)
			note(err)
			o.announceRepair(p.Host, err, "%s was not enabled — enabled it, or its ListenStream binds nothing after a reboot", sockUnit)
		}
		if !o.query("is-active", sockUnit) && !restartedService {
			// Bringing an inactive socket up goes through the same ordering:
			// the service may be holding the port itself today (keymail binds
			// :25 until its LISTEN_FDS change lands), and a bare socket start
			// under it is the EADDRINUSE race again.
			err := o.socketSequence(p.Host, p.Set.Daemon, p.DeferRestart, "start")
			note(err)
			o.announceRepair(p.Host, err, "%s was not active — started it through the stop → socket → start ordering, so nothing races the service for the port", sockUnit)
			restartedService = true
		}
	}
	if p.Set.Daemon {
		if !o.query("is-enabled", svcUnit) {
			err := o.systemctl("enable", svcUnit)
			note(err)
			o.announceRepair(p.Host, err, "%s was not enabled — enabled it, so it comes back after a reboot", svcUnit)
		}
		if !o.query("is-active", svcUnit) && !p.DeferRestart {
			// enable+start, not try-restart: a daemon that crashed into a
			// stopped state comes back rather than being left down by a
			// try-restart no-op (spec §4d).
			//
			// Deferred, this is the start that would run the OLD binary: the
			// service is inactive here precisely because step 7 just stopped
			// it, and the caller's Restart is one symlink swap away.
			err := o.systemctl("start", svcUnit)
			note(err)
			o.announceRepair(p.Host, err, "the service was not running and this route declares daemon — started it")
			restartedService = true
		}
	}

	// Step 9. A non-daemon route's restart is not this function's business:
	// the caller's existing try-restart owns it, byte-identically to today.
	// Nor is a DEFERRED daemon's: restarting it here would cycle the old
	// binary under the new unit text, and the caller restarts it once, after
	// the symlink moves.
	if svcChanged && p.Set.Daemon && !restartedService && !p.DeferRestart {
		note(o.systemctl("restart", svcUnit))
		restartedService = true
	}
	st.restarted = restartedService
	if st.err != nil {
		o.logf("  %s: unit state: %v", p.Host, st.err)
	}
	return st
}

// socketSequence is the pinned ordering: stop the service, restart (or
// start) the socket, start the service.
//
// What the ordering buys, stated precisely so nobody reads more into it than
// it gives: while the SERVICE is down and the socket unit is up, inbound
// connections queue in the kernel backlog and are answered when the service
// comes back — that is the phase spec §4c's "an MX restart never refuses a
// connection" is about, and it is strictly better than today, where a relay
// restart drops :25 entirely. During the socket unit's OWN restart the
// listener does not exist, and connections in that window are refused, not
// queued. The window is one systemctl verb wide and only opens when the
// rendered ListenStream set actually changed.
//
// The service start is ATTEMPTED even when the socket verb failed. Returning
// early there would leave the service stopped — trading a socket that did not
// rebind for an app that is not running at all — so the error is remembered
// and returned, and the start happens anyway.
//
// Accepted, and theoretical until keymail's LISTEN_FDS change lands: between
// the stop and the socket verb the socket unit may still be up, so a
// connection can socket-ACTIVATE the service back before this function starts
// it — after which the start is systemd's no-op. That is a correct end state
// (the service is running against the reloaded configuration), just not one
// this function drove; it is not worth an isolate/mask dance to prevent.
//
// A route that is neither a daemon nor currently running is neither stopped
// nor started: a hibernating tenant's drop-ins are unit text, not a wake.
//
// deferStart drops the trailing service start for a DAEMON whose caller is
// about to start it itself (Plan.DeferRestart) — the service is then stopped
// here and stays stopped until the caller's Restart, which is what keeps the
// old binary from ever running under the new unit shape. It is deliberately
// ignored for a non-daemon route: there the caller's follow-up is a
// try-restart, which is a no-op on a stopped unit, so skipping the start
// would leave the app down.
func (o Ops) socketSequence(host string, daemon, deferStart bool, verb string) error {
	svcUnit, sockUnit := ServiceUnit(host), SocketUnit(host)
	shouldRun := daemon || o.query("is-active", svcUnit)

	var errs []error
	if shouldRun {
		if err := o.systemctl("stop", svcUnit); err != nil {
			errs = append(errs, err)
		}
	}
	if err := o.systemctl(verb, sockUnit); err != nil {
		errs = append(errs, err)
	}
	if shouldRun && !(daemon && deferStart) {
		if err := o.systemctl("start", svcUnit); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

// materialize writes body (or removes the file when body is nil) and reports
// whether any byte moved. A file at a generated path WITHOUT the marker on
// its first line is never overwritten and never deleted: it is reported once
// per pass and this half is skipped entirely, while the other half still
// converges.
func (o Ops) materialize(host, path string, body, cur []byte, exists, generated bool) (bool, error) {
	if exists && !generated {
		o.logf("  %s: %s exists but does not carry the %q marker — hand-edited or written by something else; never overwritten, never removed, skipping this unit's drop-in", host, path, Marker)
		return false, nil
	}
	if body == nil {
		if !exists {
			return false, nil
		}
		if err := os.Remove(path); err != nil {
			o.logf("  %s: remove %s: %v", host, path, err)
			return false, err
		}
		o.logf("  %s: no capability materializes here any more — removed %s", host, path)
		return true, nil
	}
	if exists && bytes.Equal(cur, body) {
		return false, nil // up to date: no write, no reload, no restart
	}
	if err := writeFileAtomic(path, body); err != nil {
		o.logf("  %s: write %s: %v", host, path, err)
		return false, err
	}
	return true, nil
}

// writeFileAtomic creates the .d directory and renames a temp file into
// place, so a concurrent reader (systemd, or a second adopt process) sees
// either the old bytes or the new ones and never a torn file. Root-owned
// 0644, set explicitly rather than left to the process umask, because these
// bytes are read by systemd and their mode is part of the contract.
func writeFileAtomic(path string, body []byte) error {
	dir := filepath.Dir(path)
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return err
	}
	tmp := path + ".tmp"
	if err := os.WriteFile(tmp, body, 0o644); err != nil {
		return err
	}
	if err := os.Chmod(tmp, 0o644); err != nil {
		os.Remove(tmp)
		return err
	}
	if err := os.Rename(tmp, path); err != nil {
		os.Remove(tmp)
		return err
	}
	return nil
}

// readDropIn returns a generated file's bytes, whether it exists at all, and
// whether it carries the marker.
func readDropIn(path string) (b []byte, exists, generated bool, err error) {
	b, err = os.ReadFile(path)
	if err != nil {
		if os.IsNotExist(err) {
			return nil, false, false, nil
		}
		return nil, false, false, err
	}
	return b, true, bytes.HasPrefix(b, []byte(Marker+"\n")), nil
}

// checkStateDirShape is spec §4b's pinned sharp edge. With DynamicUser=yes
// systemd materialises StateDirectory under /var/lib/private/ with a SYMLINK
// at /var/lib/carlos-app/<host>; with a static User= it expects a real
// directory. Flipping between the two under an existing state dir is refused
// in place — the operator does the stop → move → start, because moving a
// live app's data is not a thing an hourly timer should do unattended.
//
// The OFF direction is deliberately narrow: it refuses only when THIS
// package's own current drop-in is what put the unit on a static user. A
// pre-existing real directory that no drop-in of ours created (a legacy
// hand-provisioned console, for example) is not a flip and must not freeze an
// unrelated route's capabilities forever.
func (o Ops) checkStateDirShape(p Plan, hostRole bool, curSvc []byte, generated bool) error {
	dir := filepath.Join(o.stateRoot(), p.Host)
	fi, err := os.Lstat(dir)
	if err != nil {
		return nil // absent is fine in both directions
	}
	wasStatic := generated && dropInHas(curSvc, "DynamicUser=no")
	switch {
	case hostRole && fi.Mode()&os.ModeSymlink != 0:
		e := fmt.Errorf("host_role wants a static user but %s is a symlink (the DynamicUser shape, real data under /var/lib/private)", dir)
		o.logf("  %s: %v — refused in place, nothing written; the operator's move is: systemctl stop %s && mv /var/lib/private/carlos-app/%s %s && rm the symlink && chown -R %s %s && systemctl start %s",
			p.Host, e, ServiceUnit(p.Host), p.Host, dir, AppUser(p.App), dir, ServiceUnit(p.Host))
		return e
	case !hostRole && wasStatic && fi.IsDir():
		e := fmt.Errorf("dropping host_role returns the unit to DynamicUser, which cannot use the existing real directory %s", dir)
		o.logf("  %s: %v — refused in place, nothing written; the operator's move is: systemctl stop %s, move the data under /var/lib/private/carlos-app/%s, then start it",
			p.Host, e, ServiceUnit(p.Host), p.Host)
		return e
	}
	return nil
}

// checkClaimsOnDisk resolves the two box-wide exclusivities against what is
// ALREADY WRITTEN on this box, before anything is written here. CheckClaims
// covers a whole pass's plans at once; this covers the single-route
// version-change path, which never sees the other routes — and it is the only
// one of the two that notices a drop-in left behind by a route that has since
// been removed from the registry.
//
// The /run existence check is the second guard on runtime_dir. The static
// reservedRuntimeDir list is necessarily finite — a box-specific
// /run/tailscale is not on it and never will be — so a name that already
// exists under /run and is not this route's own prior claim is refused,
// whatever the list says.
func (o Ops) checkClaimsOnDisk(p Plan, curSvc []byte, generated bool) error {
	ports, dirs := scanClaims(o.etcDir(), p.Host)

	for _, decl := range p.Set.Ports {
		if !p.Grants.Ports[decl.Port] {
			continue // an ungranted port materializes nothing and claims nothing
		}
		if other, taken := ports[decl.Port]; taken {
			e := fmt.Errorf("port %d is already claimed by %s's drop-in on this box", decl.Port, other)
			o.logf("  %s: %v — refusing before anything is written; :%d can be held by exactly one process on one machine", p.Host, e, decl.Port)
			return e
		}
	}

	if p.Set.RuntimeDir == "" {
		return nil
	}
	if other, taken := dirs[p.Set.RuntimeDir]; taken {
		e := fmt.Errorf("runtime_dir %q is already claimed by %s's drop-in on this box", p.Set.RuntimeDir, other)
		o.logf("  %s: %v — refusing before anything is written; a shared /run directory is another app's sockets", p.Host, e)
		return e
	}
	ownClaim := generated && dropInHas(curSvc, "RuntimeDirectory="+p.Set.RuntimeDir)
	if ownClaim {
		return nil
	}
	runPath := filepath.Join(o.runDir(), p.Set.RuntimeDir)
	if _, err := os.Lstat(runPath); err == nil {
		e := fmt.Errorf("runtime_dir %q: %s already exists and is not this route's own claim", p.Set.RuntimeDir, runPath)
		o.logf("  %s: %v — refusing before anything is written; systemd would hand this app a directory something else on the box already owns", p.Host, e)
		return e
	}
	return nil
}

// scanClaims reads every OTHER host's drop-in at the generated path on this
// box and returns which privileged ports and which runtime directories they
// claim, mapped to the host that claims them.
//
// A file at that path WITHOUT the marker counts too, and deliberately: it is
// not ours to rewrite, so systemd will apply it, so the claim it makes is
// real. Refusing is the fail-closed direction — the alternative is two units
// racing for :25, which is the exact failure this check exists to prevent.
// (Only the 10- path is read; an operator's 90-operator.conf is theirs, and
// this package neither reads nor reasons about it.)
func scanClaims(etcDir, exceptHost string) (map[int]string, map[string]string) {
	ports := map[int]string{}
	dirs := map[string]string{}
	ents, err := os.ReadDir(etcDir)
	if err != nil {
		return ports, dirs
	}
	for _, e := range ents {
		if !e.IsDir() {
			continue
		}
		name := e.Name()
		host, kind, ok := parseUnitDir(name)
		if !ok || host == exceptHost {
			continue
		}
		b, exists, _, err := readDropIn(filepath.Join(etcDir, name, DropInName))
		if err != nil || !exists {
			continue
		}
		for _, line := range strings.Split(string(b), "\n") {
			switch kind {
			case "socket":
				if v, ok := strings.CutPrefix(line, "ListenStream="); ok {
					if n, ok := listenStreamPort(v); ok {
						ports[n] = host
					}
				}
			case "service":
				if v, ok := strings.CutPrefix(line, "RuntimeDirectory="); ok && v != "" {
					dirs[v] = host
				}
			}
		}
	}
	return ports, dirs
}

// listenStreamPort reads the port out of a ListenStream= value. This package
// only ever WRITES the bare-number form, but the claim scan also reads files
// this package did not write, and systemd accepts several spellings of the
// same bind: "25", "0.0.0.0:25", "[::]:25", "1.2.3.4:25". Taking the segment
// after the last colon covers all of them, so an address form cannot dodge
// the box-wide port claim — the check would be worth little if it could.
//
// A value containing a path separator is a unix socket (the template's own
// ListenStream=/run/carlos/apps/%i.sock) and claims no TCP port at all.
func listenStreamPort(v string) (int, bool) {
	v = strings.TrimSpace(v)
	if v == "" || strings.Contains(v, "/") {
		return 0, false
	}
	if i := strings.LastIndex(v, ":"); i >= 0 {
		v = v[i+1:]
	}
	n, err := strconv.Atoi(v)
	if err != nil || n < 1 || n > 65535 {
		return 0, false
	}
	return n, true
}

// parseUnitDir splits "carlos-app@<host>.service.d" into its host and kind.
func parseUnitDir(name string) (host, kind string, ok bool) {
	rest, ok := strings.CutPrefix(name, "carlos-app@")
	if !ok {
		return "", "", false
	}
	for _, k := range []string{"service", "socket"} {
		if h, ok := strings.CutSuffix(rest, "."+k+".d"); ok && h != "" {
			return h, k, true
		}
	}
	return "", "", false
}

// dropInHas reports whether body carries the exact directive line.
func dropInHas(body []byte, line string) bool {
	for _, l := range strings.Split(string(body), "\n") {
		if l == line {
			return true
		}
	}
	return false
}

// CheckClaims resolves box-wide exclusivity BEFORE anything is written,
// across every plan on this box at once: a privileged port claimed by two
// routes, and a runtime_dir claimed by two routes. Returns host -> refusal; a
// host absent from the map cleared.
//
// Deterministic on ties: the route whose Host sorts first keeps the claim and
// the other is refused, so a conflict does not flap between passes — two
// routes taking turns owning :25 would restart a mail server every hour
// forever.
//
// A plan whose Set is nil claims nothing: it is the leave-alone state, and
// leaving a drop-in alone is not a new claim. An ungranted port likewise
// claims nothing, because nothing materializes for it.
//
// Known limit, named rather than papered over: this tie-break is decided
// against the PLANS, and Converge's own check is decided against what is
// already on DISK. When the lexicographically-LATER host is the one already
// materialized, the two disagree — CheckClaims refuses the later host, and
// Converge then refuses the earlier host too, because the later host's
// drop-in really is holding the port. The result is a stable both-refused
// stall, not a flap: neither route materializes and both log every pass until
// an operator revokes one grant. That is the fail-closed direction (nobody
// silently takes :25 off a running mail server), and the alternative —
// letting the winner evict the incumbent's drop-in — is a pass that stops
// somebody else's listener to satisfy a sort order.
func CheckClaims(plans []Plan) map[string]error {
	sorted := make([]Plan, len(plans))
	copy(sorted, plans)
	sort.Slice(sorted, func(i, j int) bool { return sorted[i].Host < sorted[j].Host })

	refused := map[string]error{}
	ports := map[int]string{}
	dirs := map[string]string{}
	for _, p := range sorted {
		if p.Set == nil || refused[p.Host] != nil {
			continue
		}
		for _, decl := range p.Set.Ports {
			if !p.Grants.Ports[decl.Port] {
				continue
			}
			if other, taken := ports[decl.Port]; taken {
				refused[p.Host] = fmt.Errorf("port %d is already claimed on this box by %s — :%d can be held by exactly one process on one machine; revoke it from one of the two (carlos route --host %s --revoke %s)",
					decl.Port, other, decl.Port, p.Host, PortToken(decl.Port))
				break
			}
			ports[decl.Port] = p.Host
		}
		if refused[p.Host] != nil {
			continue
		}
		if p.Set.RuntimeDir == "" {
			continue
		}
		if other, taken := dirs[p.Set.RuntimeDir]; taken {
			refused[p.Host] = fmt.Errorf("runtime_dir %q is already claimed on this box by %s — a shared /run directory is another app's sockets; one of the two must declare a different name",
				p.Set.RuntimeDir, other)
			continue
		}
		dirs[p.Set.RuntimeDir] = p.Host
	}
	return refused
}

// Restore puts prev back — the drop-in half of adoptHost's restart-failure
// rollback (the symlink half is release.PointLive's).
//
// The SERVICE half's restart is the caller's: it rolled the symlink back and
// it owns bringing the app up on the old version. The SOCKET half's is NOT.
// Bytes are not a binding: a restored ListenStream that systemd never re-binds
// restores nothing — the MX stays off :25 after a rollback — and a rolled-back
// ADDED port stays publicly bound to a binary that never asked for it. So when
// the socket half's bytes move, Restore runs the same stop → restart → start
// sequence Converge does, which is why Snapshot carries Daemon.
//
// Marker-checked in both directions, exactly as Converge is: a half that was
// not ours when the snapshot was taken is not ours to write over or delete
// now either.
func Restore(o Ops, prev Snapshot) error {
	if err := o.authorizeMutation(); err != nil {
		return err

	}
	if o.CheckHost != nil {
		if err := o.CheckHost(prev.Host); err != nil {
			return err
		}
	}
	if err := validHost(prev.Host); err != nil {
		o.logf("  %q: %v", prev.Host, err)
		return err
	}
	svcChanged, sockChanged, errs := o.applyHalves(prev)
	if svcChanged || sockChanged {
		// Before any unit verb, without exception.
		if err := o.systemctl("daemon-reload"); err != nil {
			o.logf("  %s: daemon-reload: %v", prev.Host, err)
			errs = append(errs, err)
		}
		o.logf("  %s: capability drop-ins rolled back to the previous version's", prev.Host)
	}
	if sockChanged {
		// Never deferred: a restore is the end of the line, and its caller
		// has already rolled the symlink back to the binary this shape
		// belongs to.
		if err := o.socketSequence(prev.Host, prev.Daemon, false, "restart"); err != nil {
			o.logf("  %s: rebinding the previous version's listeners: %v", prev.Host, err)
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

// Restart is the daemon route's restart: `systemctl restart`, and its error
// is REAL — it is what makes adoptHost's rollback fire. Non-daemon routes are
// not this function's business: rootRestartRoute's try-restart keeps owning them,
// byte-identically.
func Restart(o Ops, p Plan) error {
	if err := o.authorizeMutation(); err != nil {
		return err

	}
	if o.CheckHost != nil {
		if err := o.CheckHost(p.Host); err != nil {
			return err
		}
	}
	if err := validHost(p.Host); err != nil {
		o.logf("  %q: %v", p.Host, err)
		return err
	}
	return o.systemctl("restart", ServiceUnit(p.Host))
}

// validHost refuses anything that is not a bare hostname before it is joined
// into a root-written /etc/systemd/system path — the same guard, in the same
// direction, as appconfig.syncHost's.
func validHost(host string) error {
	if host == "" || host == "." || host == ".." || host != filepath.Base(host) {
		return fmt.Errorf("not a bare hostname, refusing to derive a /etc/systemd/system path from it")
	}
	return nil
}

func (o Ops) etcDir() string {
	if o.EtcDir == "" {
		return DefaultEtcDir
	}
	return o.EtcDir
}

func (o Ops) stateRoot() string {
	if o.StateRoot == "" {
		return DefaultStateRoot
	}
	return o.StateRoot
}

func (o Ops) runDir() string {
	if o.RunDir == "" {
		return DefaultRunDir
	}
	return o.RunDir
}

func (o Ops) logf(format string, a ...any) {
	if o.Logf != nil {
		o.Logf(format, a...)
		return
	}
	fmt.Printf(format+"\n", a...)
}

// systemctl runs one state-CHANGING verb. Every call here is something a
// converged box does not need: an unchanged pass makes none of them, which is
// why the tests can assert "zero calls" and mean it.
func (o Ops) systemctl(args ...string) error {
	if o.Systemctl != nil {
		return o.Systemctl(args...)
	}
	out, err := exec.Command("systemctl", args...).CombinedOutput()
	if err != nil {
		return fmt.Errorf("systemctl %s: %v: %s", strings.Join(args, " "), err, strings.TrimSpace(string(out)))
	}
	return nil
}

// query runs a systemctl QUERY verb (`is-active`, `is-enabled`) and reports
// whether it answered yes. It is a separate seam from Systemctl on purpose: a
// query changes nothing, and the properties this package promises — an
// unchanged pass writes no bytes, reloads nothing and restarts nothing — are
// about state CHANGES. Folding the two together would force a choice between
// repairing drift (an unenabled socket unit, a daemon that is not running,
// both named in spec §4d as what this pass fixes) and being quiet when there
// is no drift; with the seams split, both hold.
//
// A systemctl that cannot run at all answers "no", which errs toward
// attempting an idempotent enable/start rather than silently skipping the
// repair.
func (o Ops) query(args ...string) bool {
	if o.Query != nil {
		return o.Query(args...)
	}
	return exec.Command("systemctl", args...).Run() == nil
}

// ensureUser creates the locked, home-less, nologin system user a host_role
// app runs as, idempotently — exactly as host.ensureServiceUser creates
// `carlos`, and with the name DERIVED by AppUser, never taken from a manifest.
func (o Ops) ensureUser(name string) error {
	if o.EnsureUser != nil {
		return o.EnsureUser(name)
	}
	if _, err := user.Lookup(name); err == nil {
		return nil
	}
	out, err := exec.Command("useradd", "--system", "--user-group",
		"--no-create-home", "--shell", "/usr/sbin/nologin", name).CombinedOutput()
	if err != nil {
		return fmt.Errorf("create user %s: %v: %s", name, err, strings.TrimSpace(string(out)))
	}
	// useradd is not the only way this can fail: a NSS backend that accepted
	// the create but does not yet resolve the name would leave a User= line
	// pointing at nothing. Prove the account is resolvable before saying yes.
	if _, err := user.Lookup(name); err != nil {
		return fmt.Errorf("created user %s but it does not resolve: %v", name, err)
	}
	return nil
}