package store

import (
	"bufio"
	"errors"
	"fmt"
	"io"
	"math"
	"math/rand"
	"os"
	"os/exec"
	"path/filepath"
	"syscall"
	"testing"
	"time"
)

// kill9_test.go — Р6/Phase 0 acceptance test: kill -9 mid-work loses at most one in-flight call and
// NEVER desyncs the money from the checkpoints. A real SIGKILL against a subprocess (the standard
// library's helper-process pattern), not emulation.
//
// Invariant after any kill: committed_usd == SUM(checkpoints.cost_usd) — settle and the checkpoint are
// written in ONE transaction. Orphaned reserved_usd are cleared by the recovery pass on open.
//
// The helper announces its first committed checkpoint on stdout and the parent kills it only once that
// line has arrived. That handshake is what gives the assertions their subject: the kill is known to land
// on a process that has already committed, so a checkpoint the helper CONFIRMED before dying has to be
// on disk when the database is reopened. That is durability, not merely internal consistency, and it is
// the stronger of the two claims this test can make. Liveness is a fact the helper reports rather than
// an inference from a clock, so a helper that never commits is a loud failure instead of a round that
// quietly asserts nothing.
//
// The database is shared across rounds, so every stored figure is cumulative and only the INCREMENT
// speaks about the round that just ran: a total is non-zero for the rest of the run once any round has
// written, whatever the helper does afterwards. Each round therefore compares against the previous
// round's own reading and requires strict growth, which is also why no round needs an exemption.
//
// Two shapes this test deliberately does not take:
//   - a line per commit would fill the 64 KB pipe and stall the helper, leaving the test measuring its
//     own plumbing. The helper prints exactly once, and the parent drains to EOF regardless so nothing
//     the testing framework writes later can block either.
//   - polling the database while the helper runs would block on the exclusive flock store.Open takes.
//     Nothing polls: the previous round's reading is this round's baseline, and no second connection is
//     ever opened against a live helper.

const (
	killLoopEnv = "TM_STORE_KILL_LOOP_DB"
	// killLoopRoundEnv scopes the helper's request hashes to its round. The rest of a hash is the
	// helper's pid and a counter that restarts at zero every round, so two rounds can only collide when
	// the OS hands the second helper the pid the first one had. Checkpoints are idempotent by
	// request_hash (ON CONFLICT DO NOTHING), so that collision would settle once between the two rounds,
	// leave the count flat, and make the increment check below report a durability failure that never
	// happened. The round makes each round's writes unambiguously its own.
	killLoopRoundEnv = "TM_STORE_KILL_LOOP_ROUND"
	// killLoopReady is printed by the helper exactly once, after its FIRST settle+checkpoint has
	// committed. It is the whole liveness signal: everything the parent asserts afterwards is about work
	// that is known to be on disk.
	killLoopReady = "KILL9-HELPER-COMMITTED-FIRST-CHECKPOINT"
	// killLoopReadyWait bounds the wait for that signal. It is deliberately far larger than any plausible
	// first-commit latency (the race detector's ~15× on this loop included): it is not a performance
	// budget, it is the difference between "the helper is slow" and "the helper is not committing", and
	// only the second is a failure.
	killLoopReadyWait = 60 * time.Second
)

func TestHelperKillLoop(t *testing.T) {
	path := os.Getenv(killLoopEnv)
	if path == "" {
		t.Skip("helper process only")
	}
	s, err := Open(path)
	if err != nil {
		fmt.Fprintln(os.Stderr, "helper open:", err)
		os.Exit(1)
	}
	if err := s.UpsertSnapshot("snap", "brief", `{}`); err != nil {
		os.Exit(1)
	}
	job, err := s.EnsureJob("book", 1, "draft", "snap")
	if err != nil {
		os.Exit(1)
	}
	round := os.Getenv(killLoopRoundEnv)
	// Infinite reserve → settle+checkpoint loop; the parent will SIGKILL us
	// at a random moment. No Release — recovery will clear the "crashed" reservations.
	for i := 0; ; i++ {
		res, verdict, err := s.Reserve("book", 0.002, Ceilings{BookUSD: 10_000})
		if err != nil || verdict != ReserveOK {
			os.Exit(1)
		}
		cp := Checkpoint{
			RequestHash: fmt.Sprintf("hash-%s-%d-%d", round, os.Getpid(), i),
			JobID:       job.ID, ChunkIdx: i, Stage: "draft", Role: "translator",
			ModelRequested: "m", ModelActual: "m",
			ResponseText: "ответ", UsageJSON: "{}", CostUSD: 0.001,
		}
		if err := s.SettleWithCheckpoint(res, 0.001, cp, nil); err != nil {
			os.Exit(1)
		}
		if i == 0 {
			// ONCE, and only after the write is durable. Announcing before the settle would put the
			// parent's whole assertion back on a guess.
			fmt.Println(killLoopReady)
		}
	}
}

// killHelper is a running helper plus the one piece of bookkeeping its lifecycle needs: os/exec is
// explicit that "it is incorrect to call Wait before all reads from the pipe have completed", because
// Wait closes the pipe out from under the reader. drained is closed when the reader has reached EOF,
// and killAndReap is the only place Wait is called.
type killHelper struct {
	cmd     *exec.Cmd
	drained <-chan struct{}
}

// killAndReap SIGKILLs a helper that is expected to be running and collects it in the order os/exec
// requires. A helper that has already exited means the round lost its subject before the kill landed,
// so that is named rather than absorbed.
func (h *killHelper) killAndReap(t *testing.T, round int) {
	t.Helper()
	if err := h.cmd.Process.Signal(syscall.SIGKILL); err != nil {
		t.Fatalf("round %d: SIGKILL: %v", round, err)
	}
	h.reap()
}

// reap leaves nothing of the helper behind. The signal is best-effort because reap also serves the
// paths where the helper may already have exited on its own; what it guarantees is that a live helper
// is stopped BEFORE the reader is waited on. The reader drains until EOF, and a process still running
// never produces one — so waiting for it first would hang the test where it means to fail it.
func (h *killHelper) reap() {
	_ = h.cmd.Process.Signal(syscall.SIGKILL)
	<-h.drained      // every read from the pipe has completed…
	_ = h.cmd.Wait() // …and only now may Wait close it
}

// startKillHelper launches the helper and returns once it has COMMITTED its first checkpoint, so the
// caller knows the kill it is about to send will land on a process that has already done real work.
// Everything it can go wrong with is a loud failure, never a skip and never a silent pass.
func startKillHelper(t *testing.T, round int, dbPath string) *killHelper {
	t.Helper()
	exe, err := os.Executable()
	if err != nil {
		t.Fatal(err)
	}
	cmd := exec.Command(exe, "-test.run", "^TestHelperKillLoop$", "-test.v")
	cmd.Env = append(os.Environ(), killLoopEnv+"="+dbPath, fmt.Sprintf("%s=%d", killLoopRoundEnv, round))
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		t.Fatalf("round %d: helper stdout pipe: %v", round, err)
	}
	if err := cmd.Start(); err != nil {
		t.Fatalf("round %d: start helper: %v", round, err)
	}
	// One reader for the whole life of the process: it reports the marker and then keeps draining, so the
	// pipe can never fill and stall the helper mid-loop. It reports reaching EOF without the marker as an
	// error rather than as silence, because "the helper never committed" and "the helper died" are
	// different failures and a test that confuses them sends the next reader down the wrong path.
	ready := make(chan error, 1)
	drained := make(chan struct{})
	go func() {
		defer close(drained)
		sc := bufio.NewScanner(stdout)
		seen := false
		for sc.Scan() {
			if !seen && sc.Text() == killLoopReady {
				seen = true
				ready <- nil
			}
		}
		if !seen {
			// A scanner failure and a clean EOF are different facts about the helper — one is a broken
			// read, the other a process that ran and said nothing — and the reader keeps them apart so
			// the parent's diagnosis names which one happened.
			if err := sc.Err(); err != nil {
				ready <- fmt.Errorf("the helper's stdout could not be read to the end: %w", err)
			} else {
				ready <- errors.New("the helper's stdout reached a clean EOF without the commit marker")
			}
		}
		// An over-long line would have stopped the scanner above; keep draining regardless so a stalled
		// pipe can never be mistaken for a stalled store.
		_, _ = io.Copy(io.Discard, stdout)
	}()
	h := &killHelper{cmd: cmd, drained: drained}
	select {
	case err := <-ready:
		if err != nil {
			h.reap()
			t.Fatalf("round %d: %v — there was nothing for SIGKILL to land on, so everything asserted afterwards would be vacuous", round, err)
		}
	case <-time.After(killLoopReadyWait):
		h.reap()
		t.Fatalf("round %d: the helper did not commit a first checkpoint within %s — the kill would have landed on a process that had done nothing, and everything asserted afterwards would be vacuous",
			round, killLoopReadyWait)
	}
	return h
}

func TestKillMinus9LosesAtMostOneCall(t *testing.T) {
	if testing.Short() {
		t.Skip("subprocess test")
	}
	path := filepath.Join(t.TempDir(), "kill9.db")

	// The database is shared across rounds, so every stored figure is cumulative. The previous round's own
	// reading is this round's baseline — which is also why no second connection is ever opened while the
	// helper holds the project lock.
	var prevCheckpoints int
	var prevCommitted float64

	for round := 0; round < 5; round++ { // D23.4: 3→5 rounds — more SIGKILL windows per run
		helper := startKillHelper(t, round, path)
		// The kill lands at a random point of a reserve→settle cycle. This delay varies WHERE it lands;
		// whether there is anything for it to land on was settled by the readiness signal.
		time.Sleep(time.Duration(rand.Intn(200)) * time.Millisecond)
		helper.killAndReap(t, round)

		s, err := Open(path) // the recovery pass clears the stuck reservations
		if err != nil {
			t.Fatalf("round %d: reopen after kill: %v", round, err)
		}

		var checkpointSum float64
		var checkpointCount int
		if err := s.r.QueryRow(`SELECT COALESCE(SUM(cost_usd),0), COUNT(*) FROM checkpoints`).Scan(&checkpointSum, &checkpointCount); err != nil {
			t.Fatal(err)
		}
		committed, reserved, err := s.SpentUSD("book")
		if err != nil {
			t.Fatal(err)
		}
		s.Close()

		if reserved != 0 {
			t.Fatalf("round %d: recovery must clear reservations, reserved=%v", round, reserved)
		}
		// Acceptance core: money and checkpoints are atomic — a mismatch would mean
		// a paid-but-lost (or free-but-recorded) call.
		if math.Abs(committed-checkpointSum) > 1e-9 {
			t.Fatalf("round %d: committed=%v != sum(checkpoints)=%v over %d checkpoints — settle+checkpoint not atomic",
				round, committed, checkpointSum, checkpointCount)
		}
		// DURABILITY: the helper reported a committed checkpoint before it was killed, so this round must
		// have added at least one, in every round including the first.
		if checkpointCount <= prevCheckpoints {
			t.Fatalf("round %d: %d checkpoints before the kill, %d after reopening — a checkpoint the helper confirmed as committed did not survive SIGKILL, or the round did no work the previous rounds had not already done",
				round, prevCheckpoints, checkpointCount)
		}
		// The money must have moved with them, by the same increment logic.
		if committed <= prevCommitted {
			t.Fatalf("round %d: committed went from $%.6f to $%.6f across a round that added %d checkpoint(s)",
				round, prevCommitted, committed, checkpointCount-prevCheckpoints)
		}
		t.Logf("round %d: +%d checkpoints (%d total), committed=$%.6f — a confirmed commit survived SIGKILL and the ledger matches",
			round, checkpointCount-prevCheckpoints, checkpointCount, committed)
		prevCheckpoints, prevCommitted = checkpointCount, committed
	}
}