// litestream — off-box replication of the fleet's SQLite files, and the daily
// proof that the replicas restore. Two verbs, both run ON the box (titogo's
// operator-side SSH orchestration deliberately did not come along — units and
// credentials are provisioned by bootstrap; this is only the box-side half):
//
//	carlos ops litestream       (re)generate /etc/litestream.yml from the
//	                            registry + data dir; prints "changed" or
//	                            "unchanged" so the calling timer/unit knows
//	                            whether a replicator restart is due
//	carlos ops restore-verify   restore every configured database from S3 and
//	                            integrity-check it
//
// The config is always generated, never hand-maintained: `carlos add`/`remove`
// change the database set, and re-running the verb is the one way the
// replication set follows. Credentials live only in the unit's EnvironmentFile
// (0600) — this code never sees or prints them.
//
// Ported from titogo internal/ops/litestream.go (config generator, round-trip
// parser, restore-verify), plus Eleven deploy/fleet-backup.sh's hibernation
// exclusion.
package ops

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

	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
)

// litestreamVersion is the pinned release provisioned on the box when the
// binary is missing. Bump deliberately; the restore path is the thing a
// version change can silently break, and restore-verify is what catches it.
const litestreamVersion = "0.5.15"

// retryDelay is the pause before restore-verify's one retry; a var so tests
// can shrink it.
var retryDelay = 10 * time.Second

// SnapshotInterval and SnapshotRetention pin the fleet's point-in-time
// recovery window instead of inheriting litestream's 24h default. They are
// lease's constants, which hibernating instances' replicas use too, so the
// box-wide window and the per-instance one are one decision; lease.go
// records why a week.
const (
	SnapshotInterval  = lease.SnapshotInterval
	SnapshotRetention = lease.SnapshotRetention
)

// SnapshotPolicySupported reports whether `litestream version` output names a
// binary that HONOURS the snapshot stanza LitestreamConfig writes. The policy
// arrived with litestream 0.5; 0.3 and 0.4 parse the stanza and drop it on the
// floor.
//
// This exists because parsing is not evidence. Verified on the fleet
// 2026-08-15: the flagship's litestream v0.3.13 accepts the generated config,
// exits 0, lists the databases — and ignores the retention entirely. So does
// 0.5.15 when handed a misspelled key. A config that STATES a recovery window
// nothing enforces is worse than one that states none, because it reads as a
// decision someone made and checked.
//
// Unparseable output is UNSUPPORTED, never assumed-fine: the whole point is to
// refuse to claim a window we cannot show is real.
func SnapshotPolicySupported(versionOutput string) bool {
	f := strings.Fields(strings.TrimSpace(versionOutput))
	if len(f) == 0 {
		return false
	}
	v := strings.TrimPrefix(f[len(f)-1], "v") // "litestream v0.5.15" or "0.5.15"
	major, rest, ok := strings.Cut(v, ".")
	if !ok {
		return false
	}
	minor, _, _ := strings.Cut(rest, ".")
	maj, err := strconv.Atoi(major)
	if err != nil {
		return false
	}
	min, err := strconv.Atoi(minor)
	if err != nil {
		return false
	}
	if maj > 0 {
		return true // a future 1.x must not regress to "unsupported"
	}
	return min >= 5
}

// litestreamVersionOutput runs `<bin> version`. A binary that is missing or
// will not answer yields "", which SnapshotPolicySupported treats as
// unsupported.
func litestreamVersionOutput(bin string) string {
	out, err := exec.Command(bin, "version").CombinedOutput()
	if err != nil {
		return ""
	}
	return string(out)
}

// LitestreamConfig renders the generated /etc/litestream.yml. Deliberately
// minimal — path + replica URL + region, plus the snapshot policy above — so
// it can't rot against litestream's config surface; other tuning belongs in
// flags on the unit, not here. The snapshot stanza is the considered
// exception to that minimalism: a recovery window is a decision, and leaving
// it to a library default meant nobody had made it.
// Paths are sorted so re-runs with an unchanged fleet produce a byte-identical
// file (the `litestream` verb's changed/unchanged report depends on it).
func LitestreamConfig(dbPaths []string, bucket, prefix, region string) string {
	sorted := append([]string(nil), dbPaths...)
	sort.Strings(sorted)
	var b strings.Builder
	b.WriteString("# GENERATED by `carlos ops litestream` — do not hand-edit; re-run the verb\n")
	b.WriteString("# after `carlos add`/`remove` so the replication set follows the fleet.\n")
	// Global snapshot policy: applies to every database below.
	b.WriteString("snapshot:\n")
	fmt.Fprintf(&b, "  interval: %s\n", SnapshotInterval)
	fmt.Fprintf(&b, "  retention: %s\n", SnapshotRetention)
	b.WriteString("dbs:\n")
	for _, p := range sorted {
		fmt.Fprintf(&b, "  - path: %s\n", p)
		b.WriteString("    replicas:\n")
		fmt.Fprintf(&b, "      - url: s3://%s/%s/%s\n", bucket, prefix, replicaKey(p))
		if region != "" {
			fmt.Fprintf(&b, "        region: %s\n", region)
		}
	}
	return b.String()
}

// replicaKey is a database's place in the replica bucket: its full path with
// the leading separator dropped, so /data/registry.db becomes
// data/registry.db.
//
// It is the full path and not the basename because the basename is NOT
// unique, and the consequence of pretending otherwise is severe. Every
// SQLite-backed tenant on this fleet calls its database registry.db or
// data.db, and the box's own control DB is /data/registry.db; a basename key
// collapses all of them onto one S3 prefix, where two litestream replicators
// interleave their WAL into a replica that restores as neither database.
// Tree discovery (treeDBs) turns that from a curiosity into the normal case —
// every tenant tree contains a registry.db.
//
// Keeping the path readable rather than hashing it is deliberate: the person
// reading this bucket is doing a restore, probably at speed, and needs to see
// which database each key is.
//
// Changing this scheme moves every database to a new replica lineage, so it
// is a one-time freedom: no box has ever run a config generated by this verb
// (the flagship's is hand-written, tito's comes from the legacy `titogo ops
// litestream`), so there is no live lineage to break. There will be after the
// first cutover — treat it as fixed then.
func replicaKey(dbPath string) string {
	return strings.TrimPrefix(filepath.Clean(dbPath), "/")
}

// ParseLitestreamDBs recovers the database paths from a generated config.
// It reads only the exact shape LitestreamConfig writes (the round-trip is
// pinned by TestConfigRoundTrip) — it is not a YAML parser.
func ParseLitestreamDBs(cfg string) []string {
	var dbs []string
	for _, line := range strings.Split(cfg, "\n") {
		if p, ok := strings.CutPrefix(line, "  - path: "); ok {
			dbs = append(dbs, strings.TrimSpace(p))
		}
	}
	return dbs
}

// CoverageShortfall names every database the EXISTING config replicates that
// the proposed set does not — the databases a write would stop replicating.
//
// This is the guard for the one genuinely dangerous moment in this verb's
// life: cutting a box over from a hand-maintained config to a generated one.
// The generated list comes from the registry, so anything the registry does
// not know about is simply absent from it — and nothing downstream would say
// so. Replication of the dropped database stops, and the drill keeps passing
// because the drill only checks what the config names. The failure is
// invisible in both directions at once.
//
// Growing the set is always fine and needs no ceremony; shrinking it is a
// decision someone has to make on purpose.
func CoverageShortfall(existingConfig string, proposed []string) []string {
	have := map[string]bool{}
	for _, p := range proposed {
		have[filepath.Clean(p)] = true
	}
	var missing []string
	for _, p := range ParseLitestreamDBs(existingConfig) {
		if !have[filepath.Clean(p)] {
			missing = append(missing, p)
		}
	}
	sort.Strings(missing)
	return missing
}

// FleetDBs lists the non-empty *.db files this box-wide replicator owns:
// everything found by globbing dataDir (the agent's own exec-child
// databases, plus the registry itself), PLUS — separately, not via the glob
// — every BackingUnit route's DBPath when the operator set one.
//
// WHAT IS COVERED: a unit tenant (carlos-app@<host>: DynamicUser,
// StateDirectory=/var/lib/carlos-app/<host>) whose route was added with an
// explicit `-db <path>`, hibernating or not — its wake/sleep is systemctl
// and the agent can't even read a DynamicUser's 0700 state dir, so this
// box-wide replicator (running as root) is the only thing that can stream
// its writes off-box.
//
// WHAT IS NOT COVERED, and why this function does not try to guess it: a
// unit tenant's database filename inside its state dir is the APP's own
// choice — the platform enforces no naming convention there the way it does
// the dataDir/<host>.db convention for exec children — so there is no
// honest default path to assume. A unit route added without -db is
// silently invisible here. cli.go's runAdd requires -db for
// `-backing unit -hibernate` routes for exactly this reason (going
// unreplicated there is a genuine data-loss risk, not just a missed
// nicety); a non-hibernating unit tenant without -db is still the
// operator's to hand-configure. See internal/activator/backend_unit.go's
// package comment for the fuller operator-facing story.
//
// Hibernating EXEC-backed routes' databases (matched by DBPath, found via
// the dataDir glob) are excluded from the results below: the activator's
// embedded Litestream owns their replication (to the instances bucket,
// under the lease) — a second replicator on the same WAL would fight it for
// checkpointing and corrupt the replica. (Eleven's fleet-backup.sh lesson,
// carried verbatim.) That exclusion never applies to BackingUnit routes —
// their replication has no other owner at all, so they are added here
// rather than filtered.
// treeDBWalkDepth bounds the descent. It is a loop guard, not a policy: os.Stat
// follows symlinks (which it must — systemd's StateDirectory is a symlink into
// /var/lib/private), and a symlink pointing at an ancestor would otherwise
// recurse forever. No real tenant layout is anywhere near this deep.
const treeDBWalkDepth = 8

// treeDBs lists the *.db files beneath root.
//
// It exists because a unit tenant's "database" is sometimes a TREE. kass on
// the flagship keeps registry.db plus an instances/ directory that grows as the
// tenant creates instances; a single -db path cannot say that, which is why
// that box's config was still hand-maintained — and why it had silently
// drifted, missing an instance database created after the list was written.
//
// Two things are deliberately skipped:
//
//   - dot-prefixed directories, which is where litestream keeps its own shadow
//     WAL (".<db>-litestream"). Replicating a replicator's bookkeeping would be
//     absurd, and worse, some of those files are named *.db.
//   - anything that is not a non-empty *.db. Sidecars (-wal, -shm) travel with
//     their database and are not replicated separately; an empty file is a
//     placeholder, not a database.
//
// Discovery requires the .db convention, unlike a single-file DBPath which is
// taken verbatim: when the operator names one file we believe them, but when
// we go looking we only claim what we can recognise.
func treeDBs(root string) []string {
	var out []string
	var walk func(dir string, depth int)
	walk = func(dir string, depth int) {
		if depth > treeDBWalkDepth {
			return
		}
		entries, err := os.ReadDir(dir)
		if err != nil {
			return // unreadable subtree: nothing to claim, and not this pass's business to fail
		}
		for _, e := range entries {
			name := e.Name()
			if strings.HasPrefix(name, ".") {
				continue
			}
			p := filepath.Join(dir, name)
			// Stat, not e.Info(): entries may be symlinks and the target is
			// what matters.
			fi, err := os.Stat(p)
			if err != nil {
				continue
			}
			if fi.IsDir() {
				walk(p, depth+1)
				continue
			}
			if !strings.HasSuffix(name, ".db") || fi.Size() == 0 {
				continue
			}
			out = append(out, p)
		}
	}
	walk(root, 0)
	return out
}

// consoleDBPath resolves where this box's console keeps its database:
// CONSOLE_DB when set (the same env var cmd/console reads, so a relocated
// console cannot silently drop out of replication), else the production
// layout <data>/console/console.db — expressed against dataDir rather than
// hardcoded to /data so a test box's console is found under its own root and
// never the real one.
func consoleDBPath(dataDir string) string {
	if p := strings.TrimSpace(os.Getenv("CONSOLE_DB")); p != "" {
		return filepath.Clean(p)
	}
	return filepath.Join(dataDir, "console", "console.db")
}

func FleetDBs(reg *registry.Registry, dataDir string) ([]string, error) {
	dbs, _, err := FleetDBsWithSkipped(reg, dataDir, nil)
	return dbs, err
}

// LayoutJournalPresent is the layout pass's "held" predicate for dataDir: a
// host has a migration journal unless its journal path is definitely absent.
// A journal that cannot be read is not proof there is none. It is the same
// test internal/layout's JournalPresent makes, spelled out here so this
// package does not depend on the pass.
func LayoutJournalPresent(dataDir string) func(host string) bool {
	return func(host string) bool {
		_, err := os.Lstat(provision.LayoutJournalPath(dataDir, host))
		return !errors.Is(err, os.ErrNotExist)
	}
}

// FleetDBsWithSkipped is FleetDBs, plus the files under dataDir it declined to
// claim — see the "glob proposes, registry disposes" note inside. The verb
// prints them; nothing here is dropped without being named.
//
// fenced, when non-nil, names hosts the layout pass holds mid-move. Such a
// host's row does not claim its database, so the file is named among the
// skipped instead of being replicated from a directory it is leaving.
func FleetDBsWithSkipped(reg *registry.Registry, dataDir string, fenced func(host string) bool) (dbs []string, skipped []string, err error) {
	routes, err := reg.List()
	if err != nil {
		return nil, nil, err
	}
	hibernating := map[string]bool{}
	unitPaths := map[string]bool{}
	for _, r := range routes {
		if r.Backing == registry.BackingUnit {
			// Unit tenants' data lives outside dataDir (their systemd
			// StateDirectory), so the glob below can never find them on its
			// own — DBPath, when the operator set it, is the only way this
			// replicator learns where they are.
			if r.DBPath != "" {
				unitPaths[filepath.Clean(r.DBPath)] = true
			}
			continue
		}
		if r.Hibernate && r.DBPath != "" {
			hibernating[filepath.Clean(r.DBPath)] = true
		}
	}

	matches, err := filepath.Glob(filepath.Join(dataDir, "*.db"))
	if err != nil {
		return nil, nil, err
	}
	// A nested row's DB lives one level down, under
	// <dataDir>/instances/<id>/<host>.db (isolated-exec design §7): the flat
	// glob above never walks into that directory, so its matches are folded
	// in here before the claimed-files gate below, which already knows how
	// to accept or skip each one. A nested row's DBPath was added to
	// claimed by the very same loop that adds a flat row's.
	nested, err := filepath.Glob(filepath.Join(dataDir, "instances", "*", "*.db"))
	if err != nil {
		return nil, nil, err
	}
	matches = append(matches, nested...)
	// THE GLOB PROPOSES, THE REGISTRY DISPOSES. A *.db under dataDir is
	// replicated when a route claims it, or when it is the box's own registry
	// or console. Anything else is somebody's leftover, and streaming leftovers
	// off-box costs storage forever: a box with a hand-written config can
	// accumulate a large, root-owned, unrelated *.db file in /data that
	// matches this glob exactly and would otherwise escape replication
	// silently.
	//
	// Skipped files are NAMED to the caller, never dropped in silence — a
	// database that someone did want, quietly left out, is the precise failure
	// this area keeps producing. The shrink guard (CoverageShortfall) is the
	// second net: if a skipped file is one the live config replicates today,
	// the write is refused outright.
	claimed := map[string]bool{
		filepath.Clean(filepath.Join(dataDir, "registry.db")): true,
		filepath.Clean(consoleDBPath(dataDir)):                true,
	}
	for _, r := range routes {
		if r.DBPath == "" || (fenced != nil && fenced(r.Host)) {
			continue
		}
		claimed[filepath.Clean(r.DBPath)] = true
	}
	var globbed []string
	for _, p := range matches {
		if claimed[filepath.Clean(p)] {
			globbed = append(globbed, p)
			continue
		}
		skipped = append(skipped, filepath.Clean(p))
	}
	matches = globbed

	// Unit-tenant DBPaths are added on top of the glob: they typically live
	// entirely outside dataDir, at their StateDirectory path. A DBPath naming
	// a DIRECTORY is a tree — see treeDBs. These are claimed by construction:
	// a route named them.
	for p := range unitPaths {
		if fi, err := os.Stat(p); err == nil && fi.IsDir() {
			matches = append(matches, treeDBs(p)...)
			continue
		}
		matches = append(matches, p)
	}
	// And the console's own database, for the same reason in miniature: it
	// sits at <data>/console/console.db — ONE DIRECTORY BELOW the flat glob,
	// which is exactly why it went unreplicated. It holds no platform state
	// (routes and channels are read live) but it holds every account,
	// ownership row, session and the deployment's Keymail credentials: losing
	// it loses who everyone is. Boxes without a console simply have no file
	// here and the loop below drops it.
	matches = append(matches, consoleDBPath(dataDir))

	seen := map[string]bool{}
	for _, p := range matches {
		p = filepath.Clean(p)
		if seen[p] {
			continue // a unit DBPath that happens to also sit under dataDir
		}
		fi, err := os.Stat(p)
		if err != nil || !fi.Mode().IsRegular() || fi.Size() == 0 {
			continue // gone, odd, or an empty placeholder — nothing to replicate
		}
		if hibernating[p] {
			continue
		}
		seen[p] = true
		dbs = append(dbs, p)
	}
	sort.Strings(skipped)
	return dbs, skipped, nil
}

// restoreAndCheckFn is the per-database restore behind a package var, so the
// one-at-a-time property below is testable without S3 or a litestream binary.
var restoreAndCheckFn = restoreAndCheck

// RestoreVerify restores every database in the generated config from S3 into
// tmpDir and proves each opens clean. Returns an error on any failure so the
// caller (a oneshot unit) can exit non-zero where the fleet sweeps see it.
//
// Exactly ONE restored copy exists at a time: each is deleted as soon as it
// has been checked. This is a memory constraint, not housekeeping —
// carlos-restore-verify.service runs PrivateTmp=yes, so tmpDir is a tmpfs and
// every byte written there is charged to the unit's own cgroup against
// MemoryMax. Keeping all of them meant the drill's peak was the SUM of the
// fleet's databases: measured on carlos-tito-ie 2026-08-15, 54 databases
// totalling 247M under MemoryMax=256M, OOM-killed at a "256.2M memory peak"
// after ~40 clean restores — a drill that grew its own failure as the fleet
// grew. The peak is now the largest single database.
func RestoreVerify(litestreamBin, cfgPath, tmpDir string) error {
	raw, err := os.ReadFile(cfgPath)
	if err != nil {
		return fmt.Errorf("restore-verify: %v", err)
	}
	dbs := ParseLitestreamDBs(string(raw))
	if len(dbs) == 0 {
		return fmt.Errorf("restore-verify: no databases in %s — regenerate it: carlos ops litestream", cfgPath)
	}

	failures, stale := 0, 0
	for _, db := range dbs {
		// Sweep this database's copy however the check ends — a failure must
		// not leave the wreckage behind for the next restore to be charged for.
		out := filepath.Join(tmpDir, filepath.Base(db))
		err := restoreAndCheckFn(litestreamBin, cfgPath, db, tmpDir)
		if err != nil {
			removeRestored(out)
			// One retry after a pause: a database created moments ago may not
			// have finished its first replication pass yet.
			time.Sleep(retryDelay)
			err = restoreAndCheckFn(litestreamBin, cfgPath, db, tmpDir)
		}
		// Freshness is judged BEFORE the copy is swept, and only once the
		// restore itself passed: "restores clean but stopped tracking its
		// source" is a different finding from "does not restore", and
		// conflating them would hide whichever came second.
		var fresh freshness
		var freshDetail string
		if err == nil {
			fresh, freshDetail = replicaFreshness(db, out, sourceModTime(db))
		}
		removeRestored(out) // every path, success included
		if err != nil {
			failures++
			fmt.Printf("FAIL  %s: %v\n", db, err)
			continue
		}
		switch fresh {
		case freshnessStale:
			stale++
			fmt.Printf("STALE %s: %s\n", db, freshDetail)
		case freshnessNoSource:
			stale++
			fmt.Printf("STALE %s: %s\n", db, freshDetail)
		case freshnessUnknown:
			// Never a pass and never an alarm — see the freshness doc.
			fmt.Printf("ok    %s (freshness unknown: %s)\n", db, freshDetail)
		default:
			fmt.Printf("ok    %s\n", db)
		}
	}
	switch {
	case failures > 0 && stale > 0:
		return fmt.Errorf("restore-verify: %d of %d replicas failed to restore clean, and %d more restore but have stopped tracking their source",
			failures, len(dbs), stale)
	case failures > 0:
		return fmt.Errorf("restore-verify: %d of %d replicas failed to restore clean", failures, len(dbs))
	case stale > 0:
		// A stale replica restores perfectly and is still a durability
		// failure: what comes back is the database as it was when replication
		// stopped. It fails the run for the same reason an unrestorable one
		// does.
		return fmt.Errorf("restore-verify: %d of %d replicas restore clean but have STOPPED TRACKING their source — replication is broken for them",
			stale, len(dbs))
	}
	fmt.Printf("restore-verify: all %d replicas restore clean and are current\n", len(dbs))
	return nil
}

// removeRestored deletes one restored copy and the sidecars an open may have
// attached beside it (CheckRestoredDB's read-only DSN does not stop the driver
// creating a -wal). Errors are dropped: this is reclamation, and a file that
// is already gone is the outcome wanted.
func removeRestored(path string) {
	for _, p := range []string{path, path + "-wal", path + "-shm"} {
		_ = os.Remove(p)
	}
}

// restoreAndCheck pulls one database back from its replica and runs the same
// checks a disaster recovery would lean on: the file opens, integrity_check
// says ok, and the schema is non-empty.
func restoreAndCheck(lsBin, cfgPath, dbPath, tmpDir string) error {
	out := filepath.Join(tmpDir, filepath.Base(dbPath))
	_ = os.Remove(out) // a retry must not trip over the failed attempt's file
	cmd := exec.Command(lsBin, "restore", "-config", cfgPath, "-o", out, dbPath)
	if b, err := cmd.CombinedOutput(); err != nil {
		return fmt.Errorf("litestream restore: %v: %s", err, strings.TrimSpace(string(b)))
	}
	return CheckRestoredDB(out)
}

// CheckRestoredDB opens a restored file read-only and proves it is a sane
// SQLite database: integrity_check == ok and at least one table present.
//
// The implementation moved to internal/lease so the WAKE path can run the
// same check without importing this package — one definition of "restores
// clean", used by both the daily sweep and the moment an instance is about to
// serve a restored file. The name stays here because the ops verb and its
// tests are its documented door.
func CheckRestoredDB(path string) error { return lease.CheckDB(path) }

// writeIfChanged writes content to path only when it differs, and reports
// whether it wrote. The byte-identical no-op is what lets a timer re-run the
// generator freely without a restart-worthy mtime/content churn.
func writeIfChanged(path, content string) (bool, error) {
	if old, err := os.ReadFile(path); err == nil && bytes.Equal(old, []byte(content)) {
		return false, nil
	}
	if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
		return false, err
	}
	return true, nil
}