package localapi

import (
	"context"
	"io"
	"log/slog"
	"os"
	"path/filepath"
	"strings"
	"testing"

	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
	"gitea.dooplex.hu/admin/felhom-agent/internal/storage"
)

// R-117 — BoundUnderParent must mean THE BIND ACTUALLY WORKS, not "a mount by that name exists".
//
// THE BUG THESE PIN. GuestSeesMount (intermediary.go) and isHostMountpoint both parse a mountinfo line
// and then test only fields[4], the mount POINT. Field 3 — major:minor — sits in the same parsed slice
// and was discarded. So after a drive is detached and returned, the raw host mount heals onto the NEW
// device via its fs-UUID-keyed unit while the bind still names the OLD one, and BOTH existing terms stay
// true. Measured live: raw on 8:32 /dev/sdc, bind on 8:16 /dev/sdb with `shutdown`, BoundUnderParent
// TRUE, EIO on every read and write, and the controller's gate taking its Return branch — restarting the
// customer's apps onto that namespace and emailing backup_target_restored, with no alarm on any channel
// (felhom.eu audits/SPIKE-r117-bind-liveness-2026-07-30.md §3.3, §5.2).
//
// AND THE HALF THAT EMITS NOTHING AT ALL (spike §9, filed R-117a): a device that fails WITHOUT
// disappearing leaves the raw mount active, the devnos EQUAL, and the drive never Disconnected — so the
// gate produces neither a Stop nor a Return action and nothing is emitted, indefinitely. A devno
// comparison alone reads stale-device=false there, which is why term 3 checks the filesystem's own abort
// flags too. TestDisks_BindLiveness_AbortedFilesystemReadsAbsent is that case; a fix that shipped only
// the devno comparison would pass every other test in this file.
//
// WHY THE FIXTURES ARE REAL. Each mountinfo body below is the captured output of the spike run, not a
// hand-written line. R-116's fix shipped green and inert because its fixture supplied a MountPath
// production never supplies. These tests redirect procSelfMountinfo at a fixture file, so the REAL
// parser (hostMountEntries), the REAL predicate (bindLiveness) and the REAL /disks handler all run —
// the data is injected, the logic is not.
//
// RED-PROOFS (each verified to land, see REPORT.md): dropping `&& s.bindUsable(...)` from either /disks
// construction site fails StaleBindReadsAbsent / UnionPath_StaleBindReadsAbsent and
// AbortedFilesystemReadsAbsent; dropping the abort check so only the device comparison remains fails
// AbortedFilesystemReadsAbsent and AbortedWins_WhenDevnosAgree ALONE — that is the P1-only fix, and it is
// the one worth fearing; dropping `emergency_ro` from abortTokensByFS fails only the emergency_ro subtest;
// and returning BindLive instead of BindUnknown for an unreadable table fails UnknownIsTreatedAsPresent.
//
// A THIRD ordering trap, caught by TestBindLiveness_Verdicts during development and worth naming because
// it reports correctly while breaking the repair: reading the abort flag BEFORE comparing devices
// classifies the real return state as BindAborted, since its stale bind carries `shutdown` as well as a
// different device. BoundUnderParent still reads absent — every test in Group A still passes — but
// AttachDrive then refuses the re-bind that actually repairs it, so the self-heal never runs. The verdict
// must answer "would a re-bind help", which means reading the abort flag of the RAW mount (the re-bind's
// target) in the stale case, and of the bind itself only when the devices already agree.

// ── fixtures, from the spike's captures ─────────────────────────────────────────────────────────
//
// Devices, super options, optional-field tags and root paths are verbatim. ONE substitution: the spike
// ran against a SCRATCH shared parent (/mnt/r117-drives) so it could not disturb the live
// /mnt/felhom-drives peer group, whereas the code under test derives the stable path itself, as
// StableParentDir + "/" + DriveNameFromRaw(raw). So /mnt/r117-drives/sd becomes
// /mnt/felhom-drives/r117sd. Substituting anything else would make these fixtures describe a path
// production never produces — which is precisely how R-116's fix shipped green and inert.

// mountinfoHealthy is the HEALTHY state (spike §3.3): raw and bind on the SAME device, no abort token.
const mountinfoHealthy = `748 33 8:16 / /mnt/r117sd rw,relatime shared:450 - ext4 /dev/sdb rw,stripe=512
755 118 8:16 /felhom-data /mnt/felhom-drives/r117sd rw,relatime shared:450 - ext4 /dev/sdb rw,stripe=512
`

// mountinfoStaleBind is the R-117 RETURN state (spike §3.3 / §5.2): the drive came back as /dev/sdc
// (8:32) and the raw mount healed onto it, while the bind still names /dev/sdb (8:16) and carries
// `shutdown`. Both pre-R-117 terms read true here.
const mountinfoStaleBind = `755 118 8:16 /felhom-data /mnt/felhom-drives/r117sd rw,relatime shared:450 - ext4 /dev/sdb rw,stripe=512,shutdown
814 33 8:32 / /mnt/r117sd rw,relatime shared:485 - ext4 /dev/sdc rw,stripe=512
`

// mountinfoAborted is the STEADY-STATE state (spike §9): the device errored in place, so it NEVER LEFT.
// Raw and bind are the SAME device — the device comparison cannot see this — and ext4 has done an
// emergency remount-ro. Today this state emits nothing on any channel.
//
// SECOND substitution, and it is the one that nearly made this test decoration. The spike produced this
// state on a dm device (dm is the only mechanism that can make a device error WITHOUT disappearing), so
// the capture reads 252:11 /dev/mapper/r117cel. Transposed here onto the USB drive shape, because
// RoleForStorage derives role="system" for a /dev/mapper backing device — and a system-role row never
// enters the block that computes BoundUnderParent, so the field stays false by DEFAULT and the assertion
// below passes without term 3 ever running. It did exactly that until RP1 failed to fail (see REPORT.md).
// An in-place abort on a USB drive is the realistic customer case anyway (a link reset that recovers the
// link after ext4 has already given up); only the super options and the matching devnos carry the claim.
const mountinfoAborted = `748 33 8:16 / /mnt/r117sd rw,relatime shared:450 - ext4 /dev/sdb rw,stripe=512,emergency_ro
755 118 8:16 /felhom-data /mnt/felhom-drives/r117sd rw,relatime shared:450 - ext4 /dev/sdb rw,stripe=512,emergency_ro
`

// mountinfoAbortedShutdown is the same in-place shape with the OTHER ext4 abort token, the one a device
// removal sets. Both were measured; a check for only `shutdown` passes mountinfoAborted and a check for
// only `emergency_ro` passes this — which is why abortTokensByFS carries both, and why RP4 exists.
const mountinfoAbortedShutdown = `748 33 8:16 / /mnt/r117sd rw,relatime shared:450 - ext4 /dev/sdb rw,stripe=512,shutdown
755 118 8:16 /felhom-data /mnt/felhom-drives/r117sd rw,relatime shared:450 - ext4 /dev/sdb rw,stripe=512,shutdown
`

// mountinfoUnknownFS is healthy-looking but on a filesystem whose abort vocabulary we have not measured.
// The honest verdict is UNKNOWN — which must be treated as PRESENT, not as live and not as absent.
const mountinfoUnknownFS = `748 33 8:16 / /mnt/r117sd rw,relatime shared:450 - btrfs /dev/sdb rw
755 118 8:16 /felhom-data /mnt/felhom-drives/r117sd rw,relatime shared:450 - btrfs /dev/sdb rw
`

// useMountinfo points the REAL parsers at a fixture for the duration of one test.
func useMountinfo(t *testing.T, body string) {
	t.Helper()
	p := filepath.Join(t.TempDir(), "mountinfo")
	if err := os.WriteFile(p, []byte(body), 0o600); err != nil {
		t.Fatal(err)
	}
	prev := procSelfMountinfo
	procSelfMountinfo = p
	t.Cleanup(func() { procSelfMountinfo = prev })
}

// livenessServer builds a /disks server over one Observe target (or one registry drive) whose raw mount
// is `raw` and stable guest path derives from it. The two PRE-R-117 terms are forced TRUE — that is the
// whole point: they were both true in the measured defect, so term 3 is the only thing that can save us.
func livenessServer(t *testing.T, obs []hub.StorageTarget, known []storage.KnownTarget) *Server {
	t.Helper()
	opts := Options{
		ListenAddr: "127.0.0.1:0",
		Guests:     &fakeGuestsCfg{},
		Backups:    &fakeBackups{},
		Store:      &fakeStore{},
		Storage:    fakeStorage{targets: obs},
		Tokens:     staticTokens{"A": 8200},
		Disks:      &fakeDiskOps{probe: storage.DeviceProbe{Probed: true, HasFilesystem: true, FSType: "ext4"}},
		DiskGate:   &fakeGate{},
		HostReader: sysOnSDA(),
		Logger:     slog.New(slog.NewTextHandler(io.Discard, nil)),
	}
	if known != nil {
		opts.DriveTargets = fakeKnownTargets{drives: known}
	}
	srv, err := NewServer(opts)
	if err != nil {
		t.Fatal(err)
	}
	srv.baseCtx = context.Background()
	// Terms 1 and 2 TRUE — the measured defect's own conditions. livenessCheck is left nil so the real
	// bindLiveness runs against the fixture.
	srv.boundCheck = func(string) bool { return true }
	srv.deviceCheck = func(string) bool { return true }
	srv.resolveStorageDevice = func(string) (string, error) { return "/dev/sdb1", nil }
	return srv
}

var obsSD = []hub.StorageTarget{
	{Name: "sd", Type: hub.StorageTypeUSB, BackingDevice: "/dev/sdb", MountPath: "/mnt/r117sd", State: hub.StorageStateAttached},
}
var knownSD = []storage.KnownTarget{
	{Name: "sd", Type: hub.StorageTypeUSB, MountPath: "/mnt/r117sd", DurableID: "uuid:71e1", UUID: "71e1"},
}

// ── Group A — the consequence: a dead namespace reads ABSENT ────────────────────────────────────

// TestDisks_BindLiveness_StaleBindReadsAbsent is R-117 case (a), through the real /disks handler.
// It asserts the CONSEQUENCE — what the controller reads off the wire — not that a comparison happened.
func TestDisks_BindLiveness_StaleBindReadsAbsent(t *testing.T) {
	useMountinfo(t, mountinfoStaleBind)
	di := diskByMount(t, livenessServer(t, obsSD, nil), "/mnt/r117sd")
	if di.BoundUnderParent {
		t.Error("BoundUnderParent reports PRESENT over a stale bind (R-117). The bind names 8:16 /dev/sdb " +
			"while the raw mount is 8:32 /dev/sdc; every access through it returns EIO. The controller's " +
			"gate would take its Return branch (controller intermediary.go:258,:299) and restart the " +
			"customer's apps onto a dead namespace, then email backup_target_restored.")
	}
}

// TestDisks_BindLiveness_AbortedFilesystemReadsAbsent is R-117a, the steady-state half — and the test a
// devno-only fix would fail. The device NEVER LEFT, so raw and bind agree on 252:11.
func TestDisks_BindLiveness_AbortedFilesystemReadsAbsent(t *testing.T) {
	for _, c := range []struct{ name, body string }{
		{"emergency_ro (errors=remount-ro fired in place)", mountinfoAborted},
		{"shutdown (forced abort)", mountinfoAbortedShutdown},
	} {
		t.Run(c.name, func(t *testing.T) {
			useMountinfo(t, c.body)
			// GUARD, earned: assert the row is the shape production emits BEFORE asserting the field.
			// A system-role row has no GuestPath, never runs the conjunction, and reports
			// BoundUnderParent=false by default — passing this test while proving nothing.
			di := diskByMount(t, livenessServer(t, obsSD, nil), "/mnt/r117sd")
			if di.Role != "user-data" || di.GuestPath == "" {
				t.Fatalf("fixture does not reproduce the production row shape: role=%q guest_path=%q — "+
					"the conjunction never runs on such a row, so any assertion below is vacuous",
					di.Role, di.GuestPath)
			}
			if di.BoundUnderParent {
				t.Error("BoundUnderParent reports PRESENT over an ABORTED filesystem (R-117a). The devnos " +
					"MATCH (the device never disappeared), so the device comparison cannot see this — only " +
					"the filesystem's own abort token can. Today this state emits NOTHING on any channel: " +
					"the drive is never Disconnected, so the gate produces neither a Stop nor a Return.")
			}
		})
	}
}

// TestDisks_BindLiveness_UnionPath_AbortedReadsAbsent — the union path for the steady-state half. It
// matters more than the Observe one here: this row's Role is hardcoded user-data and its State hardcoded
// attached, so the conjunction is the ONLY thing on the row that can report the abort.
func TestDisks_BindLiveness_UnionPath_AbortedReadsAbsent(t *testing.T) {
	useMountinfo(t, mountinfoAborted)
	di := diskByMount(t, livenessServer(t, nil, knownSD), "/mnt/r117sd")
	if di.BoundUnderParent {
		t.Error("union-path drive reports PRESENT over an ABORTED filesystem (R-117a) — and its Role and " +
			"State are both hardcoded on this row, so nothing else can contradict it")
	}
}

// The union path carries no PVE dir-storage and hardcodes State:"attached", so these terms are the only
// device truth on the row — R-113's reasoning, and it applies to term 3 identically.
func TestDisks_BindLiveness_UnionPath_StaleBindReadsAbsent(t *testing.T) {
	useMountinfo(t, mountinfoStaleBind)
	di := diskByMount(t, livenessServer(t, nil, knownSD), "/mnt/r117sd")
	if di.BoundUnderParent {
		t.Error("union-path drive reports PRESENT over a stale bind (R-117) — and its State is hardcoded " +
			"attached, so nothing else on the row can contradict it")
	}
}

// ── Group B — no false negatives ────────────────────────────────────────────────────────────────

func TestDisks_BindLiveness_HealthyReadsPresent(t *testing.T) {
	useMountinfo(t, mountinfoHealthy)
	for _, c := range []struct {
		name  string
		obs   []hub.StorageTarget
		known []storage.KnownTarget
	}{
		{"observe", obsSD, nil},
		{"union", nil, knownSD},
	} {
		t.Run(c.name, func(t *testing.T) {
			di := diskByMount(t, livenessServer(t, c.obs, c.known), "/mnt/r117sd")
			if !di.BoundUnderParent {
				t.Error("a healthy drive reads ABSENT — a false absent STOPS a working customer's apps, " +
					"which is strictly worse than the bug being fixed")
			}
		})
	}
}

// ── Group C — cannot tell must never mean absent ────────────────────────────────────────────────

// TestDisks_BindLiveness_UnknownIsTreatedAsPresent pins the rule in every way it can be reached. The
// workspace's false-invariant table records newestArchiveOn promising exactly this over a signature that
// could not express it; Usable() is the one place it lives, so this is the test that keeps it honest.
func TestDisks_BindLiveness_UnknownIsTreatedAsPresent(t *testing.T) {
	t.Run("unreadable mount table", func(t *testing.T) {
		prev := procSelfMountinfo
		procSelfMountinfo = filepath.Join(t.TempDir(), "does-not-exist")
		t.Cleanup(func() { procSelfMountinfo = prev })
		if got := bindLiveness("/mnt/felhom-drives/r117sd", "/mnt/r117sd"); got != BindUnknown {
			t.Errorf("unreadable /proc gave %v, want BindUnknown", got)
		}
		di := diskByMount(t, livenessServer(t, obsSD, nil), "/mnt/r117sd")
		if !di.BoundUnderParent {
			t.Error("an unreadable mount table made the drive read ABSENT — cannot-tell must never stop apps")
		}
	})

	t.Run("no raw mount entry to compare against", func(t *testing.T) {
		// Only the bind is in the table. devicePresent is the term that answers device absence; this one
		// must abstain rather than double-count it.
		useMountinfo(t, `755 118 8:16 /felhom-data /mnt/felhom-drives/r117sd rw,relatime shared:450 - ext4 /dev/sdb rw
`)
		if got := bindLiveness("/mnt/felhom-drives/r117sd", "/mnt/r117sd"); got != BindUnknown {
			t.Errorf("missing raw entry gave %v, want BindUnknown", got)
		}
	})

	t.Run("empty paths", func(t *testing.T) {
		if got := bindLiveness("", "/mnt/r117sd"); got != BindUnknown {
			t.Errorf("empty stable gave %v, want BindUnknown", got)
		}
		if got := bindLiveness("/mnt/felhom-drives/r117sd", ""); got != BindUnknown {
			t.Errorf("empty raw gave %v, want BindUnknown", got)
		}
	})

	t.Run("filesystem whose abort vocabulary is unmeasured", func(t *testing.T) {
		useMountinfo(t, mountinfoUnknownFS)
		if got := bindLiveness("/mnt/felhom-drives/r117sd", "/mnt/r117sd"); got != BindUnknown {
			t.Errorf("btrfs bind gave %v, want BindUnknown — we cannot read its abort state, so we must "+
				"not claim LIVE either", got)
		}
		di := diskByMount(t, livenessServer(t, obsSD, nil), "/mnt/r117sd")
		if !di.BoundUnderParent {
			t.Error("an unmeasured filesystem read ABSENT — that would stop apps on every non-ext4 drive")
		}
	})

	t.Run("Usable is the single place the rule lives", func(t *testing.T) {
		for _, c := range []struct {
			l    BindLiveness
			want bool
		}{
			{BindLive, true},
			{BindUnknown, true}, // the rule
			{BindStaleDevice, false},
			{BindAborted, false},
		} {
			if got := c.l.Usable(); got != c.want {
				t.Errorf("%v.Usable() = %v, want %v", c.l, got, c.want)
			}
		}
	})
}

// ── Group D — the verdict itself, including the ordering that matters ───────────────────────────

func TestBindLiveness_Verdicts(t *testing.T) {
	for _, c := range []struct {
		name, body  string
		stable, raw string
		want        BindLiveness
	}{
		{"healthy", mountinfoHealthy, "/mnt/felhom-drives/r117sd", "/mnt/r117sd", BindLive},
		{"stale device (case a)", mountinfoStaleBind, "/mnt/felhom-drives/r117sd", "/mnt/r117sd", BindStaleDevice},
		{"aborted in place, emergency_ro (case b)", mountinfoAborted, "/mnt/felhom-drives/r117sd", "/mnt/r117sd", BindAborted},
		{"aborted in place, shutdown", mountinfoAbortedShutdown, "/mnt/felhom-drives/r117sd", "/mnt/r117sd", BindAborted},
	} {
		t.Run(c.name, func(t *testing.T) {
			useMountinfo(t, c.body)
			if got := bindLiveness(c.stable, c.raw); got != c.want {
				t.Errorf("bindLiveness = %v, want %v", got, c.want)
			}
		})
	}
}

// TestBindLiveness_AbortedWins pins the ORDERING, which is load-bearing and not obvious: in the measured
// stale-bind state the filesystem ALSO carries `shutdown`, so both P1 and P2 apply. The verdict must be
// BindAborted-or-BindStaleDevice — never live — but more importantly the in-place state, where ONLY P2
// applies, must not fall through to a devno comparison that reads equal. This test fails if P1 is checked
// before P2 and returns early.
func TestBindLiveness_AbortedWins_WhenDevnosAgree(t *testing.T) {
	useMountinfo(t, mountinfoAborted)
	got := bindLiveness("/mnt/felhom-drives/r117sd", "/mnt/r117sd")
	if got.Usable() {
		t.Fatalf("bindLiveness = %v (usable) — the devnos agree because the device never left, so a "+
			"P1-first implementation reads this as LIVE and ships R-117's silent half intact", got)
	}
	if got != BindAborted {
		t.Errorf("bindLiveness = %v, want BindAborted (the abort token is the only signal here)", got)
	}
}

// ── Group E — the parser, on a real captured table ──────────────────────────────────────────────

// TestHostMountEntries_ParsesDevnoAndSuperOpts pins the field extraction R-117 turned on. The optional
// fields run (shared:NNN master:NNN) is variable-length, so the " - " separator — not a fixed index — is
// what locates fstype and the super options.
func TestHostMountEntries_ParsesDevnoAndSuperOpts(t *testing.T) {
	// A guest-side line with BOTH optional-field tags, the longest real shape (spike §5.2).
	useMountinfo(t, `759 1176 8:16 /felhom-data /mnt/felhom-drives/r117sd rw,relatime shared:459 master:450 - ext4 /dev/sdb rw,stripe=512,shutdown
`)
	got := hostMountEntries("/mnt/felhom-drives/r117sd")
	if len(got) != 1 {
		t.Fatalf("got %d entries, want 1", len(got))
	}
	e := got[0]
	if e.Devno != "8:16" {
		t.Errorf("Devno = %q, want 8:16 — this is the field R-117 was lost for want of reading", e.Devno)
	}
	if e.Root != "/felhom-data" {
		t.Errorf("Root = %q, want /felhom-data", e.Root)
	}
	if e.FSType != "ext4" {
		t.Errorf("FSType = %q, want ext4 (located via the ' - ' separator, not a fixed index)", e.FSType)
	}
	if !strings.Contains(e.SuperOpts, "shutdown") {
		t.Errorf("SuperOpts = %q, want it to carry `shutdown`", e.SuperOpts)
	}
}

// countHostMounts and isHostMountpoint were rewritten onto hostMountEntries; the double-bind convergence
// AttachDrive depends on must survive that (REUSE.md: a boolean could not converge stacked binds).
func TestHostMountEntries_CountsStackedBinds(t *testing.T) {
	useMountinfo(t, `755 118 8:16 /felhom-data /mnt/felhom-drives/r117sd rw,relatime - ext4 /dev/sdb rw
756 118 8:16 /felhom-data /mnt/felhom-drives/r117sd rw,relatime - ext4 /dev/sdb rw
`)
	if n := countHostMounts("/mnt/felhom-drives/r117sd"); n != 2 {
		t.Errorf("countHostMounts = %d, want 2 — AttachDrive's normalize leg needs the count, not a bool", n)
	}
	if !isHostMountpoint("/mnt/felhom-drives/r117sd") {
		t.Error("isHostMountpoint = false over two stacked binds")
	}
	if isHostMountpoint("/mnt/nope") {
		t.Error("isHostMountpoint = true for a path with no entry")
	}
	if n := countHostMounts("/mnt/nope"); n != 0 {
		t.Errorf("countHostMounts = %d for an absent path, want 0", n)
	}
}