//go:build linux && rootrunner

package host

import (
	"bufio"
	"bytes"
	"context"
	"crypto/ecdh"
	"crypto/ed25519"
	"crypto/rand"
	"crypto/sha256"
	"database/sql"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"os/exec"
	"os/user"
	"path/filepath"
	"strconv"
	"strings"
	"syscall"
	"testing"
	"time"

	lsfile "github.com/benbjohnson/litestream/file"
	"github.com/superfly/ltx"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
	"github.com/carlosframework/platform/internal/dbreads"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

const (
	dbreadRunnerHost = "shop.example.com"
	// dbreadReportPrefix marks the payload's one report line. m.Run prints
	// PASS after the payload test returns, so the report is never the last
	// line (plan review finding 3).
	dbreadReportPrefix = "DBREAD-REPORT "
	// dbreadServiceFile is where ensureSystemUnits("/etc", ...) installs the
	// worker's service (plan review finding 4).
	dbreadServiceFile = "/etc/systemd/system/carlos-dbread@.service"
)

// dbreadRunner is the production wiring on a throwaway root runner: the real
// units installed by the real ensureSystemUnits into /etc, as the agent's
// adopt pass installs them (agent.go), this test binary at /opt/carlos/carlos
// where the unit's ExecStart finds it, and the agent half run as carlos in a
// child process, because the worker serves only the carlos user's
// connections. /var/tmp is a tmpfs of tmpSize: the runner's small root
// volume would refuse every read's 5 GiB floor, and tmpfs only spends memory
// on what is written.
type dbreadRunner struct {
	t        *testing.T
	dir      string // owned by carlos: the store and the replica
	uid, gid int
	rootPub  ed25519.PublicKey
	rootPriv ed25519.PrivateKey
}

func runSystemctl(args ...string) (string, error) {
	out, err := exec.Command("systemctl", args...).CombinedOutput()
	return string(out), err
}

func mustSystemctl(t *testing.T, args ...string) {
	t.Helper()
	if out, err := runSystemctl(args...); err != nil {
		t.Fatalf("systemctl %s: %v: %s", strings.Join(args, " "), err, out)
	}
}

// sqliteFixture writes a real SQLite database at path, so an ATTACH of it
// fails only when the file cannot be reached (plan review finding 17).
func sqliteFixture(t *testing.T, path string) {
	t.Helper()
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS secret (v TEXT); INSERT INTO secret VALUES ('not for tenants')`); err != nil {
		t.Fatal(err)
	}
}

func newDBReadRunner(t *testing.T, tmpSize string) *dbreadRunner {
	t.Helper()
	if os.Geteuid() != 0 {
		t.Fatal("run on the root runner, as root")
	}
	r := &dbreadRunner{t: t}
	if out, err := exec.Command("mount", "-t", "tmpfs", "-o", "size="+tmpSize+",mode=1777", "tmpfs", "/var/tmp").CombinedOutput(); err != nil {
		t.Fatalf("mount /var/tmp: %v: %s", err, out)
	}
	t.Cleanup(func() { exec.Command("umount", "/var/tmp").Run() })
	u, err := user.Lookup("carlos")
	if err != nil {
		if out, err := exec.Command("useradd", "--system", "--user-group", "--no-create-home", "--shell", "/usr/sbin/nologin", "carlos").CombinedOutput(); err != nil {
			t.Fatalf("useradd: %v: %s", err, out)
		}
		t.Cleanup(func() { exec.Command("userdel", "carlos").Run() })
		if u, err = user.Lookup("carlos"); err != nil {
			t.Fatal(err)
		}
	}
	r.uid, _ = strconv.Atoi(u.Uid)
	r.gid, _ = strconv.Atoi(u.Gid)
	if _, err := os.Stat("/opt/carlos/carlos"); err == nil {
		t.Fatal("/opt/carlos/carlos exists: this test runs only on a fresh runner")
	}
	self, err := os.ReadFile(os.Args[0])
	if err != nil {
		t.Fatal(err)
	}
	for _, d := range []string{"/opt/carlos", "/etc/carlos", "/data/instances/other.example.com", "/run/carlos", dbreadTestMarkers} {
		if err := os.MkdirAll(d, 0o755); err != nil {
			t.Fatal(err)
		}
	}
	// Things the witness must not reach, present on the host: real SQLite
	// databases where the witness ATTACHes, a host.env it opens.
	sqliteFixture(t, "/data/registry.db")
	sqliteFixture(t, "/etc/carlos/witness.db")
	sqliteFixture(t, "/data/instances/other.example.com/other.db")
	// host.env arms db reads (dbReadsSettings), so the adopt pass's
	// ensureSystemUnits arms the socket: it stays dark otherwise.
	r.rootPub, r.rootPriv, _ = ed25519.GenerateKey(rand.Reader)
	if err := os.WriteFile("/etc/carlos/host.env", []byte(dbreadRunnerHostEnv(r.rootPub)), 0o640); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile("/opt/carlos/carlos", self, 0o755); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		for _, p := range []string{"/opt/carlos/carlos", "/data/registry.db", "/etc/carlos/witness.db", "/etc/carlos/host.env"} {
			os.Remove(p)
		}
		os.RemoveAll("/data/instances/other.example.com")
		os.RemoveAll(dbreadTestMarkers)
	})
	ensureSystemUnits("/etc", filepath.Join(t.TempDir(), "lock"), runSystemctl, systemctlProbe, t.Logf)
	t.Cleanup(func() {
		runSystemctl("stop", "carlos-dbread.socket", "carlos-instance-control.socket", "carlos-system-update.path")
		runSystemctl("disable", "carlos-dbread.socket", "carlos-instance-control.socket", "carlos-system-update.path", "carlos-system-recover.service")
		for _, u := range systemUnitFiles {
			os.Remove(filepath.Join("/etc/systemd/system", u))
		}
		runSystemctl("daemon-reload")
	})
	if _, err := os.Stat(dbreadServiceFile); err != nil {
		t.Fatalf("ensureSystemUnits installed no %s: %v", dbreadServiceFile, err)
	}
	if out, _ := systemctlProbe("is-active", dbReadSocketUnit); strings.TrimSpace(out) != "active" {
		t.Fatalf("carlos-dbread.socket is %q after ensureSystemUnits", out)
	}
	r.dir, err = os.MkdirTemp("/var/lib", "carlos-dbread-runner-")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(r.dir) })
	replica := filepath.Join(r.dir, "replica", "instances", dbreadRunnerHost)
	dbdoortest.Replicate(t, replica, 3)
	yes := true
	rec := provision.Record{Host: dbreadRunnerHost, App: "shop", Account: "bab", Region: "uk", Hibernate: &yes,
		CreatedAt: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)}
	body, _ := json.Marshal(rec)
	if _, err := r.store().Put(context.Background(), provision.RecordKey("bab", "shop", dbreadRunnerHost), body, ""); err != nil {
		t.Fatal(err)
	}
	r.chown()
	return r
}

// dbreadRunnerHostEnv is a host.env that arms db reads, plus extra lines.
func dbreadRunnerHostEnv(pub ed25519.PublicKey, extra ...string) string {
	lines := append([]string{"CARLOS_REGION=uk", "CARLOS_HIBERNATE_BUCKET=data-uk", "CARLOS_RELEASE_PUBKEY=" + base64.StdEncoding.EncodeToString(pub)}, extra...)
	return strings.Join(lines, "\n") + "\n"
}

func (r *dbreadRunner) store() release.Store {
	return release.NewDirStore(filepath.Join(r.dir, "store"))
}

func (r *dbreadRunner) chown() {
	filepath.Walk(r.dir, func(p string, _ os.FileInfo, _ error) error { return os.Lchown(p, r.uid, r.gid) })
}

func (r *dbreadRunner) marker(name string, on bool) {
	r.t.Helper()
	p := filepath.Join(dbreadTestMarkers, name)
	var err error
	if on {
		err = os.WriteFile(p, nil, 0o644)
	} else if err = os.Remove(p); os.IsNotExist(err) {
		err = nil
	}
	if err != nil {
		r.t.Fatal(err)
	}
}

// submit is the console's two creates for a read of op; it returns the
// read's id and reply key.
func (r *dbreadRunner) submit(op, sqlText string) (string, *ecdh.PrivateKey) {
	r.t.Helper()
	reply, _ := dbreads.NewReplyKey()
	id, _ := dbreads.MintID()
	inc, _ := hostclaim.Incarnation("bab", "shop", dbreadRunnerHost, time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC))
	req := dbreads.Request{ID: id, Account: "bab", App: "shop", Host: dbreadRunnerHost, Incarnation: inc, Region: "uk",
		Op: op, SQL: sqlText, ReplyKey: dbreads.EncodeReplyKey(reply.PublicKey()),
		By: "member@example.com", IssuedAt: time.Now().UTC().Format(time.RFC3339)}
	wire, err := dbreads.Sign(release.RootSigner(r.rootPriv), req)
	if err != nil {
		r.t.Fatal(err)
	}
	s := r.store()
	if _, err := s.Put(context.Background(), dbreads.RequestKey(id), wire, ""); err != nil {
		r.t.Fatal(err)
	}
	issued, _ := req.Issued()
	if _, err := s.Put(context.Background(), dbreads.QueueKey("uk", issued, id), nil, ""); err != nil {
		r.t.Fatal(err)
	}
	r.chown()
	return id, reply
}

// payload runs the agent half as carlos. With wait it returns the payload's
// report, the line marked dbreadReportPrefix; without, the running command.
// id, when set, is the read the scenario waits on or reports about.
func (r *dbreadRunner) payload(scenario string, wait bool, id string) (map[string]any, *exec.Cmd) {
	r.t.Helper()
	// The copy at /opt/carlos/carlos, not os.Args[0]: the runner keeps its
	// binary under /root, which carlos cannot enter.
	cmd := exec.Command("/opt/carlos/carlos")
	cmd.Env = append(os.Environ(), "DBREAD_RUNNER_PAYLOAD="+scenario, "DBREAD_RUNNER_DIR="+r.dir,
		"DBREAD_RUNNER_ID="+id, "DBREAD_RUNNER_PUB="+base64.StdEncoding.EncodeToString(r.rootPub))
	cmd.SysProcAttr = &syscall.SysProcAttr{Credential: &syscall.Credential{Uid: uint32(r.uid), Gid: uint32(r.gid)}}
	var out bytes.Buffer
	cmd.Stdout, cmd.Stderr = &out, os.Stderr
	if !wait {
		if err := cmd.Start(); err != nil {
			r.t.Fatal(err)
		}
		return nil, cmd
	}
	if err := cmd.Run(); err != nil {
		r.t.Fatalf("payload %s: %v: %s", scenario, err, out.String())
	}
	sc := bufio.NewScanner(&out)
	for sc.Scan() {
		if line, ok := strings.CutPrefix(sc.Text(), dbreadReportPrefix); ok {
			var rep map[string]any
			if err := json.Unmarshal([]byte(line), &rep); err != nil {
				r.t.Fatalf("payload %s: unreadable report %q", scenario, line)
			}
			return rep, cmd
		}
	}
	r.t.Fatalf("payload %s gave no report: %s", scenario, out.String())
	return nil, cmd
}

func (r *dbreadRunner) answer(id string, reply *ecdh.PrivateKey, stream string) ([]byte, dbreads.Result) {
	r.t.Helper()
	s := r.store()
	raw, _, err := s.Get(context.Background(), dbreads.ResultKey(id))
	if err != nil {
		r.t.Fatalf("no result: %v", err)
	}
	var res dbreads.Result
	json.Unmarshal(raw, &res)
	var out bytes.Buffer
	for seq := int64(1); seq <= res.LastSeq; seq++ {
		sealed, _, _ := s.Get(context.Background(), dbreads.ChunkKey(id, seq))
		plain, err := dbreads.OpenChunk(reply, id, seq, stream, seq == res.LastSeq, sealed)
		if err != nil {
			r.t.Fatal(err)
		}
		out.Write(plain)
	}
	return out.Bytes(), res
}

// privateTmpGone waits up to limit for every worker's private /var/tmp to
// be removed.
func privateTmpGone(t *testing.T, limit time.Duration) {
	t.Helper()
	deadline := time.Now().Add(limit)
	for {
		left, _ := filepath.Glob("/var/tmp/systemd-private-*-carlos-dbread@*")
		if len(left) == 0 {
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf("a worker's private tmp outlived it: %v", left)
		}
		time.Sleep(200 * time.Millisecond)
	}
}

func unitsActive(t *testing.T) int {
	n, err := dbReadUnits(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	return n
}

// dbreadWaitFor polls ok until it holds or limit passes. (waitFor is
// housekeeper_test.go's, with another signature: plan review finding 2.)
func dbreadWaitFor(t *testing.T, limit time.Duration, ok func() bool) {
	t.Helper()
	deadline := time.Now().Add(limit)
	for !ok() {
		if time.Now().After(deadline) {
			t.Fatal("timed out")
		}
		time.Sleep(200 * time.Millisecond)
	}
}

// T2: a forwarded query through the real socket and the real service,
// installed by the real ensureSystemUnits; the socket is the worker's stdin
// and stdout (codex 2). The probe that makes the box capable runs the
// default SelfView inside the unit (plan review finding 1).
func TestRootDBReadServesThroughTheInstalledUnits(t *testing.T) {
	r := newDBReadRunner(t, "40g")
	id, reply := r.submit(dbreads.OpQuery, "SELECT count(*) AS n FROM t")
	rep, _ := r.payload("serve", true, id)
	if rep["capable"] != true {
		t.Fatalf("the installed worker failed its probe: %v", rep)
	}
	rows, res := r.answer(id, reply, dbreads.StreamRows)
	if res.State != dbreads.StateDone || string(rows) != "n\n30\n" {
		t.Fatalf("result %+v rows %q", res, rows)
	}
	privateTmpGone(t, 10*time.Second)
}

// A forwarded pull through the real socket and the real service (plan 3
// Task 5): the installed worker verifies and holds the planned files and
// sends them back after check B; the answer, opened as the ltx stream, is
// the planned files byte for byte in plan order, and adds up to the sealed
// summary. The worker's private tmp is gone afterwards.
func TestRootDBReadPullThroughTheInstalledUnits(t *testing.T) {
	r := newDBReadRunner(t, "40g")
	id, reply := r.submit(dbreads.OpPull, "")
	rep, _ := r.payload("serve", true, id)
	if rep["capable"] != true {
		t.Fatalf("the installed worker failed its probe: %v", rep)
	}
	got, res := r.answer(id, reply, dbreads.StreamLTX)
	if res.State != dbreads.StateDone || res.Code != "" {
		t.Fatalf("result %+v", res)
	}
	sum, err := dbreads.OpenSummary(reply, id, res.Sealed)
	if err != nil {
		t.Fatal(err)
	}
	h := sha256.Sum256(got)
	if hex.EncodeToString(h[:]) != sum.SHA256 || int64(len(got)) != sum.Bytes {
		t.Fatalf("%d bytes do not add up to the summary %+v", len(got), sum)
	}
	rd := lsfile.NewReplicaClient(filepath.Join(r.dir, "replica", "instances", dbreadRunnerHost))
	plan, err := dbdoor.PlanFrom(context.Background(), rd, time.Time{})
	if err != nil {
		t.Fatal(err)
	}
	var want bytes.Buffer
	for _, f := range plan.Planned {
		rc, err := rd.OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0)
		if err != nil {
			t.Fatal(err)
		}
		io.Copy(&want, rc)
		rc.Close()
	}
	if !bytes.Equal(got, want.Bytes()) || sum.TXID != plan.TXID || len(sum.Files) != len(plan.Planned) {
		t.Fatalf("pulled %d bytes at %d in %d files; planned %d bytes at %d in %d files",
			len(got), sum.TXID, len(sum.Files), want.Len(), plan.TXID, len(plan.Planned))
	}
	privateTmpGone(t, 10*time.Second)
}

// T3: from inside the worker, nothing outside its scratch is reachable,
// including any network interface but loopback; the private tmp is gone
// after a normal end, a cancel and kill -9 of the agent. Mutation: remove
// PrivateNetwork=yes from the installed unit; "see an interface other than
// loopback" succeeds and this test is red (plan review finding 17).
func TestRootDBReadWorkerSandbox(t *testing.T) {
	r := newDBReadRunner(t, "40g")
	r.marker("witness", true)
	rep, _ := r.payload("witness", true, "")
	r.marker("witness", false)
	report, _ := rep["report"].(map[string]any)
	if len(report) != 14 {
		t.Fatalf("an incomplete witness report: %v", rep)
	}
	for name, v := range report {
		failed := v.(string) != ""
		if name == "write /var/tmp" {
			if failed {
				t.Errorf("the positive control failed: %s", v)
			}
			continue
		}
		if !failed {
			t.Errorf("%s succeeded inside the worker", name)
		}
	}
	// A cancel, then kill -9 of the agent, mid-read.
	r.marker("hang-restore", true)
	defer r.marker("hang-restore", false)
	id, _ := r.submit(dbreads.OpQuery, "SELECT 1")
	_, cmd := r.payload("serve-and-hang", false, "")
	dbreadWaitFor(t, 30*time.Second, func() bool { return unitsActive(t) >= 1 })
	if _, err := r.store().Put(context.Background(), dbreads.CancelKey(id), nil, ""); err != nil {
		t.Fatal(err)
	}
	dbreadWaitFor(t, 20*time.Second, func() bool { return unitsActive(t) == 0 })
	privateTmpGone(t, 10*time.Second)
	cmd.Process.Kill()
	cmd.Wait()
	r.submit(dbreads.OpQuery, "SELECT 1")
	_, cmd = r.payload("serve-and-hang", false, "")
	dbreadWaitFor(t, 30*time.Second, func() bool { return unitsActive(t) >= 1 })
	cmd.Process.Signal(syscall.SIGKILL)
	cmd.Wait()
	dbreadWaitFor(t, 20*time.Second, func() bool { return unitsActive(t) == 0 })
	privateTmpGone(t, 10*time.Second)
}

// T12 with real units. The disk floor: /var/tmp on a 12 GiB tmpfs fits one
// read's floor reservation (5 GiB and 10%) and not two. The first read is
// running (its unit observed) before the second is submitted, so which read
// waits never depends on queue order (plan review finding 16); the agent
// passes every second and still never picks the second while the first
// holds its floor. A fixture database small enough for the runner cannot
// make the plan's growth bind, so the no-room result itself is pinned by
// TestReadTooLargeAndNoRoom. Then the agent is killed mid-read: the
// restarted agent takes no read and no probe while the old unit is active,
// the old worker ends within 5 s of losing the connection though its
// restore ignores the context, and the restarted agent admits the held-back
// read once the unit and its scratch are gone. Mutations: drop the orphan
// rule in reconcileLocked (the restarted agent picks beside the old unit),
// or drop the worker's 5 s exit (the old unit outlives the 15 s bound).
func TestRootDBReadAdmissionWithRealUnits(t *testing.T) {
	r := newDBReadRunner(t, "12g")
	r.marker("hang-restore", true)
	r.submit(dbreads.OpQuery, "SELECT 1")
	_, hung := r.payload("serve-and-hang", false, "")
	dbreadWaitFor(t, 30*time.Second, func() bool { return unitsActive(t) == 1 })
	second, reply := r.submit(dbreads.OpQuery, "SELECT count(*) AS n FROM t")
	time.Sleep(5 * time.Second) // five passes
	if _, _, err := r.store().Get(context.Background(), dbreads.PickedKey(second)); err == nil {
		t.Fatal("a second read was picked with room for one")
	}
	hung.Process.Signal(syscall.SIGKILL)
	hung.Wait()
	killedAt := time.Now()
	rep, _ := r.payload("restarted", true, second)
	if rep["units_at_start"].(float64) < 1 {
		t.Fatalf("the old unit was gone before the restarted agent looked: %v", rep)
	}
	if rep["capable"] == true || rep["picked"] == true {
		t.Fatalf("a restarted agent admitted beside the old unit: %v", rep)
	}
	dbreadWaitFor(t, 20*time.Second, func() bool { return unitsActive(t) == 0 })
	if gone := time.Since(killedAt); gone > 15*time.Second {
		t.Fatalf("the old worker took %s to stop; its 5 s exit did not hold", gone)
	}
	privateTmpGone(t, 10*time.Second)
	r.marker("hang-restore", false)
	rep, _ = r.payload("serve", true, second)
	if rep["capable"] != true {
		t.Fatalf("the agent did not admit once the unit was gone: %v", rep)
	}
	if rows, res := r.answer(second, reply, dbreads.StreamRows); res.State != dbreads.StateDone || string(rows) != "n\n30\n" {
		t.Fatalf("the held-back read: %+v %q", res, rows)
	}
}

// T13: a socket whose service is missing, masked, failing to start or
// speaking another protocol version withdraws db-read-1 within one probe and
// edges pass; a read whose worker fails to start withdraws it at once
// through the real loop (plan review finding 15); a box running two reads
// keeps it (codex 9).
func TestRootDBReadReadiness(t *testing.T) {
	r := newDBReadRunner(t, "40g")
	capable := func() bool {
		rep, _ := r.payload("probe", true, "")
		return rep["capable"] == true && rep["published"] == true
	}
	socketState := func(when string) string {
		out, _ := runSystemctl("show", "-p", "ActiveState,SubState,Result", "carlos-dbread.socket")
		t.Logf("carlos-dbread.socket %s: %s", when, strings.Join(strings.Fields(out), " "))
		return out
	}
	// repair is the box's own way back once the service is restored: a
	// connection to a socket whose service cannot load fails the socket unit
	// (systemd stops listening), and the next adopt pass's ensureSystemUnits
	// sees it inactive and arms it again. Nothing else restarts it.
	repair := func(when string) {
		t.Helper()
		socketState(when + ", before the adopt pass")
		ensureSystemUnits("/etc", filepath.Join(t.TempDir(), "lock"), runSystemctl, systemctlProbe, t.Logf)
		if out, _ := systemctlProbe("is-active", dbReadSocketUnit); strings.TrimSpace(out) != "active" {
			t.Fatalf("carlos-dbread.socket is %q after the adopt pass", out)
		}
	}
	if !capable() {
		t.Fatal("baseline: not capable")
	}
	body, err := os.ReadFile(dbreadServiceFile)
	if err != nil {
		t.Fatal(err)
	}
	// Cleanup removes first: during the mask the path is a symlink to
	// /dev/null, and writing through it would write nothing.
	t.Cleanup(func() {
		os.Remove(dbreadServiceFile)
		os.WriteFile(dbreadServiceFile, body, 0o644)
		runSystemctl("daemon-reload")
	})
	if err := os.Remove(dbreadServiceFile); err != nil {
		t.Fatal(err)
	}
	mustSystemctl(t, "daemon-reload")
	if capable() {
		t.Error("capable with the service missing")
	}
	if err := os.WriteFile(dbreadServiceFile, body, 0o644); err != nil {
		t.Fatal(err)
	}
	mustSystemctl(t, "daemon-reload")
	repair("with the service missing")
	if !capable() {
		t.Fatal("not capable once the service file is back")
	}
	// Masked where the unit is installed (plan review finding 27): a mask is
	// a /dev/null symlink, and only one in /etc/systemd/system itself
	// outranks the unit file ensureSystemUnits writes there. `systemctl
	// mask` refuses while the file exists, and `--runtime` would land in
	// /run/systemd/system, which /etc outranks; so swap the file for the
	// symlink by hand, and prove the mask took before asking.
	if err := os.Remove(dbreadServiceFile); err != nil {
		t.Fatal(err)
	}
	if err := os.Symlink("/dev/null", dbreadServiceFile); err != nil {
		t.Fatal(err)
	}
	mustSystemctl(t, "daemon-reload")
	if out, _ := runSystemctl("show", "-p", "LoadState", "--value", "carlos-dbread@probe.service"); strings.TrimSpace(out) != "masked" {
		t.Fatalf("the mask did not take: LoadState %q", out)
	}
	if capable() {
		t.Error("capable with the service masked")
	}
	if err := os.Remove(dbreadServiceFile); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(dbreadServiceFile, body, 0o644); err != nil {
		t.Fatal(err)
	}
	mustSystemctl(t, "daemon-reload")
	repair("with the service masked")
	for _, m := range []string{"exit-1", "version-2"} {
		r.marker(m, true)
		if capable() {
			t.Errorf("capable with %s", m)
		}
		socketState("with " + m)
		r.marker(m, false)
	}
	if !capable() {
		t.Fatal("not capable again once restored")
	}

	// A read whose worker fails to start, through the real loop.
	_, loop := r.payload("fail-start", false, "")
	dbreadWaitFor(t, 30*time.Second, func() bool { _, err := os.Stat(filepath.Join(r.dir, "probed")); return err == nil })
	r.marker("exit-1", true)
	r.submit(dbreads.OpQuery, "SELECT 1")
	if err := os.WriteFile(filepath.Join(r.dir, "go"), nil, 0o644); err != nil {
		t.Fatal(err)
	}
	if err := loop.Wait(); err != nil {
		t.Fatalf("fail-start payload: %v", err)
	}
	if _, err := os.Stat(filepath.Join(r.dir, "withdrawn")); err != nil {
		t.Fatal("the loop kept db-read-1 after a worker failed to start")
	}
	r.marker("exit-1", false)

	r.marker("hang-restore", true)
	defer r.marker("hang-restore", false)
	r.submit(dbreads.OpQuery, "SELECT 1")
	r.submit(dbreads.OpQuery, "SELECT 1")
	rep, _ := r.payload("busy-probe", true, "")
	if rep["capable"] != true {
		t.Fatalf("a box running two reads lost the capability: %v", rep)
	}
}

// TestRootDBReadPayload is the agent half, run as carlos by payload();
// never run it by name.
func TestRootDBReadPayload(t *testing.T) {
	scenario := os.Getenv("DBREAD_RUNNER_PAYLOAD")
	if scenario == "" {
		t.Skip("run by the db-read root-runner tests as their carlos child")
	}
	dir := os.Getenv("DBREAD_RUNNER_DIR")
	pubRaw, _ := base64.StdEncoding.DecodeString(os.Getenv("DBREAD_RUNNER_PUB"))
	store := release.NewDirStore(filepath.Join(dir, "store"))
	a := dbreads.NewAgent(dbreads.AgentConfig{
		Store: store, Region: "uk", Box: "runner", Boot: newBootID(), Pub: ed25519.PublicKey(pubRaw),
		HibernatePrefix: "instances",
		OpenReplica: func(_ context.Context, prefix string) (dbdoor.ReplicaReader, error) {
			return lsfile.NewReplicaClient(filepath.Join(dir, "replica", prefix)), nil
		},
		DialWorker: dialDBReadWorker, Units: dbReadUnits, Free: varTmpFree, Logf: t.Logf,
	})
	ctx := context.Background()
	report := map[string]any{}
	say := func() {
		b, _ := json.Marshal(report)
		fmt.Println(dbreadReportPrefix + string(b))
	}
	switch scenario {
	case "probe":
		a.Probe(ctx)
		report["capable"] = a.Capable()
		report["published"] = len(withDBReadCapability(nil, a.Capable())) == 1
	case "serve":
		a.Probe(ctx)
		report["capable"] = a.Capable()
		id := os.Getenv("DBREAD_RUNNER_ID")
		deadline := time.Now().Add(2 * time.Minute)
		for time.Now().Before(deadline) {
			a.Pass(ctx)
			a.Wait()
			if _, _, err := store.Get(ctx, dbreads.ResultKey(id)); err == nil {
				break
			}
			time.Sleep(time.Second)
		}
	case "serve-and-hang":
		a.Probe(ctx)
		for { // the parent kills this process
			a.Pass(ctx)
			time.Sleep(time.Second)
		}
	case "restarted":
		units, _ := dbReadUnits(ctx)
		report["units_at_start"] = units
		a.Probe(ctx)
		a.Pass(ctx)
		report["capable"] = a.Capable()
		_, _, err := store.Get(ctx, dbreads.PickedKey(os.Getenv("DBREAD_RUNNER_ID")))
		report["picked"] = err == nil
	case "fail-start":
		// The production loop: probe at start, a pass on a nudge, a probe
		// as soon as a read's worker fails to start.
		nudge := make(chan struct{}, 1)
		rctx, cancel := context.WithCancel(ctx)
		defer cancel()
		go a.Run(rctx, nudge)
		for !a.Capable() {
			time.Sleep(100 * time.Millisecond)
		}
		os.WriteFile(filepath.Join(dir, "probed"), nil, 0o644)
		for {
			if _, err := os.Stat(filepath.Join(dir, "go")); err == nil {
				break
			}
			time.Sleep(100 * time.Millisecond)
		}
		nudge <- struct{}{}
		deadline := time.Now().Add(30 * time.Second)
		for a.Capable() && time.Now().Before(deadline) {
			time.Sleep(100 * time.Millisecond)
		}
		if !a.Capable() {
			os.WriteFile(filepath.Join(dir, "withdrawn"), nil, 0o644)
		}
	case "busy-probe":
		a.Probe(ctx)
		a.Pass(ctx)
		deadline := time.Now().Add(30 * time.Second)
		for n, _ := dbReadUnits(ctx); n < 2 && time.Now().Before(deadline); n, _ = dbReadUnits(ctx) {
			time.Sleep(200 * time.Millisecond)
		}
		a.Probe(ctx)
		report["capable"] = a.Capable()
	case "witness":
		c, err := dialDBReadWorker(ctx)
		if err != nil {
			t.Fatal(err)
		}
		var w map[string]string
		json.NewDecoder(io.LimitReader(c, 1<<20)).Decode(&w)
		c.Close()
		report["report"] = w
	}
	say()
}

// dbreadTenantUnit is the real tenant template, carlos-instance@.service,
// installed under its own name with only what a runner cannot provide taken
// out: the edge and mount dependencies, the per-instance slice, the stop
// hook and the log socket. Its sandbox (User=carlos, the hidden /run/carlos,
// /data, /opt/carlos and /etc/carlos, the address families) is the
// production one, line for line. ExecStart is this binary as a tenant that
// tries to reach the db-read worker.
const dbreadTenantUnit = "carlos-dbread-tenant@.service"

func installDBReadTenantUnit(t *testing.T, instance string) {
	t.Helper()
	raw, err := unitsFS.ReadFile("units/carlos-instance@.service")
	if err != nil {
		t.Fatal(err)
	}
	var out []string
	for _, line := range strings.Split(string(raw), "\n") {
		switch {
		case strings.HasPrefix(line, "BindsTo="), strings.HasPrefix(line, "Requires="), strings.HasPrefix(line, "After="),
			strings.HasPrefix(line, "ExecStopPost="), strings.HasPrefix(line, "Slice="):
			continue
		case strings.HasPrefix(line, "ExecStart="):
			line = "ExecStart=/opt/carlos/carlos dbread-tenant-witness"
		case line == "Type=simple":
			line = "Type=oneshot"
		case strings.HasPrefix(line, "StandardOutput=file:"):
			line = "StandardOutput=journal"
		}
		out = append(out, line)
	}
	unit := strings.Join(out, "\n")
	if !strings.Contains(unit, "TemporaryFileSystem=/run/carlos:ro") || !strings.Contains(unit, "User=carlos") {
		t.Fatal("the tenant template no longer hides /run/carlos as carlos; the witness would prove nothing")
	}
	path := filepath.Join("/etc/systemd/system", dbreadTenantUnit)
	if err := os.WriteFile(path, []byte(unit), 0o644); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.Remove(path); runSystemctl("daemon-reload") })
	mustSystemctl(t, "daemon-reload")
	uid, gid := carlosIDs(t)
	for _, d := range []string{"/data/instances/" + instance, "/run/carlos/instances/" + instance} {
		if err := os.MkdirAll(d, 0o750); err != nil {
			t.Fatal(err)
		}
		os.Chown(d, uid, gid)
	}
	for _, d := range []string{"/run/carlos-control/exe/" + instance + "/app", "/run/carlos-control/exe/" + instance + "/live"} {
		if err := os.MkdirAll(d, 0o755); err != nil {
			t.Fatal(err)
		}
	}
	t.Cleanup(func() {
		os.RemoveAll("/data/instances/" + instance)
		os.RemoveAll("/run/carlos/instances/" + instance)
		os.RemoveAll("/run/carlos-control/exe/" + instance)
	})
}

func carlosIDs(t *testing.T) (int, int) {
	t.Helper()
	u, err := user.Lookup("carlos")
	if err != nil {
		t.Fatal(err)
	}
	uid, _ := strconv.Atoi(u.Uid)
	gid, _ := strconv.Atoi(u.Gid)
	return uid, gid
}

// tenantReach runs the installed tenant once and returns its report, after
// checking that no connection reached the socket and no worker started.
func tenantReach(t *testing.T, instance string) map[string]string {
	t.Helper()
	accepted := func() string {
		out, _ := runSystemctl("show", "-p", "NAccepted", "--value", dbReadSocketUnit)
		return strings.TrimSpace(out)
	}
	before := accepted()
	report := filepath.Join("/data/instances", instance, "dbread-witness.json")
	os.Remove(report)
	if out, err := runSystemctl("start", strings.Replace(dbreadTenantUnit, "@", "@"+instance, 1)); err != nil {
		jout, _ := exec.Command("journalctl", "-u", strings.Replace(dbreadTenantUnit, "@", "@"+instance, 1), "-n", "30", "--no-pager").CombinedOutput()
		t.Fatalf("the tenant unit did not run: %v: %s\n%s", err, out, jout)
	}
	raw, err := os.ReadFile(report)
	if err != nil {
		t.Fatalf("the tenant wrote no report: %v", err)
	}
	var rep map[string]string
	if err := json.Unmarshal(raw, &rep); err != nil {
		t.Fatal(err)
	}
	t.Logf("tenant report: %v", rep)
	if rep["uid"] == "" || rep["uid"] == "0" {
		t.Fatalf("the witness did not run as the tenant user: %v", rep)
	}
	for _, k := range []string{"connect " + dbReadSocketPath, "connect /run/carlos-dbread/worker.sock"} {
		if v, ok := rep[k]; !ok || v == "" {
			t.Errorf("a tenant could %s: %v", k, rep)
		}
	}
	if after := accepted(); after != before {
		t.Errorf("the socket accepted a tenant connection: NAccepted %s -> %s", before, after)
	}
	if n := unitsActive(t); n != 0 {
		t.Errorf("%d db-read workers running after the tenant's attempt", n)
	}
	return rep
}

// Astra round 1, finding 1, on real units: a real tenant unit (the
// production carlos-instance@ sandbox, running as carlos) can neither
// connect to nor activate a db-read worker, with db reads armed on the box
// and with them off; and with them off nothing listens, for the agent
// either. Mutations: put the socket back at /run/carlos-dbread (the tenant
// connects, NAccepted moves); drop the gate in ensureSystemUnits (the
// socket listens with CARLOS_DB_READS=0).
func TestRootDBReadTenantCannotReachTheWorker(t *testing.T) {
	r := newDBReadRunner(t, "40g")
	const instance = "tenant.example.com"
	installDBReadTenantUnit(t, instance)

	// Armed: the socket listens, the agent's probe passes, the tenant
	// reaches nothing.
	if out, _ := systemctlProbe("is-active", dbReadSocketUnit); strings.TrimSpace(out) != "active" {
		t.Fatalf("armed box: carlos-dbread.socket is %q", out)
	}
	if rep, _ := r.payload("probe", true, ""); rep["capable"] != true {
		t.Fatalf("armed box: the agent's probe failed: %v", rep)
	}
	tenantReach(t, instance)

	for _, off := range [][]string{{"CARLOS_DB_READS=0"}, {"CARLOS_HIBERNATE_BUCKET="}} {
		if err := os.WriteFile("/etc/carlos/host.env", []byte(dbreadRunnerHostEnv(r.rootPub, off...)), 0o640); err != nil {
			t.Fatal(err)
		}
		ensureSystemUnits("/etc", filepath.Join(t.TempDir(), "lock"), runSystemctl, systemctlProbe, t.Logf)
		active, _ := systemctlProbe("is-active", dbReadSocketUnit)
		enabled, _ := systemctlProbe("is-enabled", dbReadSocketUnit)
		if strings.TrimSpace(active) == "active" || strings.TrimSpace(enabled) == "enabled" {
			t.Fatalf("%v: carlos-dbread.socket is %s and %s", off, strings.TrimSpace(active), strings.TrimSpace(enabled))
		}
		if rep, _ := r.payload("probe", true, ""); rep["capable"] == true {
			t.Fatalf("%v: something still answers on the worker socket: %v", off, rep)
		}
		tenantReach(t, instance)
	}
}