//go:build integration

package store_test

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"errors"
	"os"
	"path/filepath"
	"runtime"
	"sort"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/rvbox/rvbox/internal/server/store"
)

const busyTimeout = 2 * time.Second

func TestRealSQLiteInitializationAndRestart_HP_STORE_01(t *testing.T) {
	t.Parallel()

	dataDir := filepath.Join(t.TempDir(), "state")
	opened := openStore(t, dataDir)

	for _, directory := range []string{dataDir, filepath.Join(dataDir, "segments"), filepath.Join(dataDir, "audit"), filepath.Join(dataDir, "deleting")} {
		info, err := os.Stat(directory)
		if err != nil {
			t.Fatal(err)
		}
		if runtime.GOOS != "windows" && info.Mode().Perm() != 0o700 {
			t.Fatalf("%s mode = %04o, want 0700", directory, info.Mode().Perm())
		}
	}
	for _, file := range []string{"rvbox.db", "server.lock"} {
		info, err := os.Stat(filepath.Join(dataDir, file))
		if err != nil {
			t.Fatal(err)
		}
		if runtime.GOOS != "windows" && info.Mode().Perm() != 0o600 {
			t.Fatalf("%s mode = %04o, want 0600", file, info.Mode().Perm())
		}
	}

	assertPragma(t, opened.DB(), "journal_mode", "wal")
	assertPragma(t, opened.DB(), "foreign_keys", "1")
	assertPragma(t, opened.DB(), "synchronous", "2")
	assertPragma(t, opened.DB(), "busy_timeout", "2000")

	rows, err := opened.DB().Query(`SELECT name FROM sqlite_schema WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name`)
	if err != nil {
		t.Fatal(err)
	}
	var names []string
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			t.Fatal(err)
		}
		names = append(names, name)
	}
	if err := rows.Close(); err != nil {
		t.Fatal(err)
	}
	want := []string{"audit_events", "clients", "command_events", "command_payloads", "command_tombstones", "commands", "control_mutations", "output_segments", "output_truncations", "schema_migrations", "sessions", "stdin_writes", "storage_counters", "storage_incidents", "takeover_authorizations"}
	sort.Strings(want)
	if strings.Join(names, ",") != strings.Join(want, ",") {
		t.Fatalf("tables = %v, want %v", names, want)
	}
	var migrationCount int
	if err := opened.DB().QueryRow(`SELECT count(*) FROM schema_migrations`).Scan(&migrationCount); err != nil || migrationCount != 1 {
		t.Fatalf("migration count = %d, err = %v", migrationCount, err)
	}
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}

	reopened := openStore(t, dataDir)
	if err := reopened.DB().QueryRow(`SELECT count(*) FROM schema_migrations`).Scan(&migrationCount); err != nil || migrationCount != 1 {
		t.Fatalf("reopened migration count = %d, err = %v", migrationCount, err)
	}
	if err := reopened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestSingleInstanceLockAndConcurrentClose_RACE_STORE_01(t *testing.T) {
	t.Parallel()

	dataDir := filepath.Join(t.TempDir(), "state")
	first := openStore(t, dataDir)
	if _, err := store.Open(context.Background(), store.Options{DataDir: dataDir, BusyTimeout: busyTimeout}); !errors.Is(err, store.ErrAlreadyOpen) {
		t.Fatalf("second Open error = %v, want ErrAlreadyOpen", err)
	}
	var wait sync.WaitGroup
	for range 8 {
		wait.Add(1)
		go func() {
			defer wait.Done()
			if err := first.Close(); err != nil && !errors.Is(err, sql.ErrConnDone) {
				t.Errorf("Close: %v", err)
			}
		}()
	}
	wait.Wait()
	reopened := openStore(t, dataDir)
	if err := reopened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestSchemaRejectsInvalidIDsAndForeignKeys_BH_STORE_01(t *testing.T) {
	t.Parallel()

	opened := openStore(t, filepath.Join(t.TempDir(), "state"))
	defer opened.Close()
	_, err := opened.DB().Exec(`INSERT INTO clients (
client_id, platform, architecture, daemon_version, daemon_cwd, supported_shells, capabilities, client_instance_id
) VALUES ('client-a', 3, 'amd64', 'test', 'C:\\work', x'', x'', x'01')`)
	if err == nil {
		t.Fatal("invalid client instance UUID was accepted")
	}
	_, err = opened.DB().Exec(`INSERT INTO sessions (
session_id, client_id, client_instance_id, generation, opened_at
) VALUES (?, 'missing-client', ?, 1, 1)`, bytesOf(16, 1), bytesOf(16, 2))
	if err == nil || !strings.Contains(strings.ToLower(err.Error()), "foreign key") {
		t.Fatalf("foreign-key insert error = %v", err)
	}
}

func TestUnsafePathsAndCancelledOpen_BH_STORE_02(t *testing.T) {
	t.Parallel()

	if _, err := store.Open(context.Background(), store.Options{DataDir: "relative", BusyTimeout: busyTimeout}); !errors.Is(err, store.ErrUnsafeDataDirectory) {
		t.Fatalf("relative path error = %v", err)
	}
	if _, err := store.Open(context.Background(), store.Options{DataDir: string(filepath.Separator), BusyTimeout: busyTimeout}); !errors.Is(err, store.ErrUnsafeDataDirectory) {
		t.Fatalf("root path error = %v", err)
	}
	permissive := filepath.Join(t.TempDir(), "permissive")
	if err := os.Mkdir(permissive, 0o755); err != nil {
		t.Fatal(err)
	}
	if _, err := store.Open(context.Background(), store.Options{DataDir: permissive, BusyTimeout: busyTimeout}); !errors.Is(err, store.ErrUnsafeDataDirectory) {
		t.Fatalf("permissive path error = %v", err)
	}
	if runtime.GOOS != "windows" {
		target := filepath.Join(t.TempDir(), "target")
		if err := os.Mkdir(target, 0o700); err != nil {
			t.Fatal(err)
		}
		link := filepath.Join(t.TempDir(), "linked")
		if err := os.Symlink(target, link); err != nil {
			t.Fatal(err)
		}
		if _, err := store.Open(context.Background(), store.Options{DataDir: link, BusyTimeout: busyTimeout}); !errors.Is(err, store.ErrUnsafeDataDirectory) {
			t.Fatalf("symlink path error = %v", err)
		}
	}

	cancelled, cancel := context.WithCancel(context.Background())
	cancel()
	dataDir := filepath.Join(t.TempDir(), "cancelled")
	if _, err := store.Open(cancelled, store.Options{DataDir: dataDir, BusyTimeout: busyTimeout}); !errors.Is(err, context.Canceled) {
		t.Fatalf("cancelled Open error = %v", err)
	}
	opened := openStore(t, dataDir)
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestMigrationChecksumMismatchPreventsOpen_REC_STORE_01(t *testing.T) {
	t.Parallel()

	dataDir := filepath.Join(t.TempDir(), "state")
	opened := openStore(t, dataDir)
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}
	raw, err := sql.Open("sqlite", filepath.Join(dataDir, "rvbox.db"))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := raw.Exec(`UPDATE schema_migrations SET checksum = 'tampered' WHERE version = 1`); err != nil {
		t.Fatal(err)
	}
	if err := raw.Close(); err != nil {
		t.Fatal(err)
	}
	if _, err := store.Open(context.Background(), store.Options{DataDir: dataDir, BusyTimeout: busyTimeout}); err == nil || !strings.Contains(err.Error(), "checksum mismatch") {
		t.Fatalf("Open error = %v, want checksum mismatch", err)
	}
	// A failed open must release the process lock for offline inspection/repair.
	raw, err = sql.Open("sqlite", filepath.Join(dataDir, "rvbox.db"))
	if err != nil {
		t.Fatal(err)
	}
	if err := raw.Ping(); err != nil {
		t.Fatal(err)
	}
	if err := raw.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestSegmentAppendTailRecoveryAndCommittedCorruption_REC_STORE_02(t *testing.T) {
	t.Parallel()

	directory := filepath.Join(t.TempDir(), "segments")
	if err := os.Mkdir(directory, 0o700); err != nil {
		t.Fatal(err)
	}
	owner := uuidBytes(10)
	segment, err := store.CreateSegment(directory, owner, 1, store.DefaultSegmentLimits())
	if err != nil {
		t.Fatal(err)
	}
	name := segment.Name()
	first := integrationRecord(owner, 1, "first")
	second := integrationRecord(owner, 2, "second")
	if _, err := segment.Append(first); err != nil {
		t.Fatal(err)
	}
	committed, err := segment.Append(second)
	if err != nil {
		t.Fatal(err)
	}
	if err := segment.Close(); err != nil {
		t.Fatal(err)
	}
	path := filepath.Join(directory, name)
	file, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := file.Write([]byte("unacknowledged-tail")); err != nil {
		t.Fatal(err)
	}
	if err := file.Sync(); err != nil {
		t.Fatal(err)
	}
	if err := file.Close(); err != nil {
		t.Fatal(err)
	}
	recovery, err := store.RecoverSegment(directory, name, committed, store.DefaultSegmentLimits())
	if err != nil {
		t.Fatal(err)
	}
	if recovery.TruncatedTail != uint64(len("unacknowledged-tail")) || len(recovery.Records) != 2 {
		t.Fatalf("recovery = %+v", recovery)
	}
	info, err := os.Stat(path)
	if err != nil {
		t.Fatal(err)
	}
	if uint64(info.Size()) != committed {
		t.Fatalf("size = %d, want %d", info.Size(), committed)
	}

	corrupt, err := os.OpenFile(path, os.O_RDWR, 0)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := corrupt.WriteAt([]byte{0xff}, 128); err != nil {
		t.Fatal(err)
	}
	if err := corrupt.Close(); err != nil {
		t.Fatal(err)
	}
	if _, err := store.RecoverSegment(directory, name, committed, store.DefaultSegmentLimits()); !errors.Is(err, store.ErrSegmentChecksum) {
		t.Fatalf("corrupt committed range error = %v", err)
	}
	info, err = os.Stat(path)
	if err != nil {
		t.Fatal(err)
	}
	if uint64(info.Size()) != committed {
		t.Fatalf("corruption changed committed file: size=%d", info.Size())
	}
}

func TestSegmentShortRangeUnsafeReferenceAndExclusiveCreate_BH_STORE_05(t *testing.T) {
	t.Parallel()

	directory := filepath.Join(t.TempDir(), "segments")
	if err := os.Mkdir(directory, 0o700); err != nil {
		t.Fatal(err)
	}
	owner := uuidBytes(20)
	segment, err := store.CreateSegment(directory, owner, 9, store.DefaultSegmentLimits())
	if err != nil {
		t.Fatal(err)
	}
	name := segment.Name()
	committed, err := segment.Append(integrationRecord(owner, 1, "payload"))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := store.CreateSegment(directory, owner, 9, store.DefaultSegmentLimits()); err == nil {
		t.Fatal("exclusive segment name collision succeeded")
	}
	if err := segment.Close(); err != nil {
		t.Fatal(err)
	}
	path := filepath.Join(directory, name)
	if err := os.Truncate(path, int64(committed-1)); err != nil {
		t.Fatal(err)
	}
	if _, err := store.RecoverSegment(directory, name, committed, store.DefaultSegmentLimits()); !errors.Is(err, store.ErrCommittedRangeMissing) {
		t.Fatalf("short committed range error = %v", err)
	}
	for _, unsafe := range []string{"../" + name, "not-a-segment", filepath.Join("nested", name)} {
		if _, err := store.RecoverSegment(directory, unsafe, 0, store.DefaultSegmentLimits()); !errors.Is(err, store.ErrUnsafeSegmentReference) {
			t.Errorf("unsafe reference %q error = %v", unsafe, err)
		}
	}
}

func TestSegmentConcurrentAppendSerializesRecords_RACE_STORE_02(t *testing.T) {
	t.Parallel()

	directory := filepath.Join(t.TempDir(), "segments")
	if err := os.Mkdir(directory, 0o700); err != nil {
		t.Fatal(err)
	}
	owner := uuidBytes(30)
	segment, err := store.CreateSegment(directory, owner, 1, store.DefaultSegmentLimits())
	if err != nil {
		t.Fatal(err)
	}
	const count = 16
	ends := make(chan uint64, count)
	errorsFound := make(chan error, count)
	var wait sync.WaitGroup
	for sequence := 1; sequence <= count; sequence++ {
		wait.Add(1)
		go func(sequence uint64) {
			defer wait.Done()
			end, err := segment.Append(integrationRecord(owner, sequence, "parallel"))
			if err != nil {
				errorsFound <- err
				return
			}
			ends <- end
		}(uint64(sequence))
	}
	wait.Wait()
	close(errorsFound)
	for err := range errorsFound {
		t.Fatal(err)
	}
	close(ends)
	var committed uint64
	seenEnds := map[uint64]bool{}
	for end := range ends {
		if seenEnds[end] {
			t.Fatalf("duplicate committed end %d", end)
		}
		seenEnds[end] = true
		if end > committed {
			committed = end
		}
	}
	name := segment.Name()
	if err := segment.Close(); err != nil {
		t.Fatal(err)
	}
	recovery, err := store.RecoverSegment(directory, name, committed, store.DefaultSegmentLimits())
	if err != nil {
		t.Fatal(err)
	}
	if len(recovery.Records) != count {
		t.Fatalf("record count = %d, want %d", len(recovery.Records), count)
	}
}

func TestEventCrashBeforeMetadataDropsUncommittedTail_CRASH_STORE_01(t *testing.T) {
	t.Parallel()

	dataDir := filepath.Join(t.TempDir(), "state")
	injected := errors.New("injected crash before metadata")
	opened := openStoreWithOptions(t, store.Options{
		DataDir: dataDir, BusyTimeout: busyTimeout,
		FaultInjector: &failOnceInjector{target: store.FaultAfterSegmentSyncBeforeMetadata, failure: injected},
	})
	issue := uuidBytes(40)
	seedCommand(t, opened.DB(), issue)
	event := appendEvent(issue, 1, "not-acknowledged")
	if result, err := opened.AppendCommandEvent(context.Background(), event); !errors.Is(err, injected) || result.ThroughEventSeq != 0 {
		t.Fatalf("AppendCommandEvent = (%+v, %v)", result, err)
	}
	assertCommandEventState(t, opened.DB(), issue, 0, 0)
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}

	reopened := openStore(t, dataDir)
	recovery, err := reopened.RecoverCommandSegments(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if recovery.OrphansRemoved != 1 || recovery.RecordsChecked != 0 {
		t.Fatalf("recovery = %+v", recovery)
	}
	result, err := reopened.AppendCommandEvent(context.Background(), event)
	if err != nil || result.ThroughEventSeq != 1 || result.Duplicate {
		t.Fatalf("retry = (%+v, %v)", result, err)
	}
	assertCommandEventState(t, reopened.DB(), issue, 1, 1)
	if err := reopened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestEventCrashAfterCommitReplaysWithoutDuplication_CRASH_STORE_02(t *testing.T) {
	t.Parallel()

	dataDir := filepath.Join(t.TempDir(), "state")
	injected := errors.New("injected lost acknowledgement")
	opened := openStoreWithOptions(t, store.Options{
		DataDir: dataDir, BusyTimeout: busyTimeout,
		FaultInjector: &failOnceInjector{target: store.FaultAfterMetadataCommit, failure: injected},
	})
	issue := uuidBytes(60)
	seedCommand(t, opened.DB(), issue)
	event := appendEvent(issue, 1, "committed")
	if result, err := opened.AppendCommandEvent(context.Background(), event); !errors.Is(err, injected) || result.ThroughEventSeq != 1 {
		t.Fatalf("AppendCommandEvent = (%+v, %v)", result, err)
	}
	assertCommandEventState(t, opened.DB(), issue, 1, 1)
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}

	reopened := openStore(t, dataDir)
	recovery, err := reopened.RecoverCommandSegments(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if recovery.SegmentsChecked != 1 || recovery.RecordsChecked != 1 || recovery.OrphansRemoved != 0 {
		t.Fatalf("recovery = %+v", recovery)
	}
	result, err := reopened.AppendCommandEvent(context.Background(), event)
	if err != nil || !result.Duplicate || result.ThroughEventSeq != 1 {
		t.Fatalf("duplicate retry = (%+v, %v)", result, err)
	}
	conflict := event
	conflict.ImmutableSHA256 = sha256.Sum256([]byte("different immutable event"))
	if _, err := reopened.AppendCommandEvent(context.Background(), conflict); !errors.Is(err, store.ErrEventConflict) {
		t.Fatalf("conflicting duplicate error = %v", err)
	}
	assertCommandEventState(t, reopened.DB(), issue, 1, 1)
	if err := reopened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestEventCrashOnExistingSegmentTruncatesOnlyTail_CRASH_STORE_03(t *testing.T) {
	t.Parallel()

	dataDir := filepath.Join(t.TempDir(), "state")
	injected := errors.New("injected second append crash")
	opened := openStoreWithOptions(t, store.Options{
		DataDir: dataDir, BusyTimeout: busyTimeout,
		FaultInjector: &failOnceInjector{target: store.FaultAfterSegmentSyncBeforeMetadata, failure: injected, skip: 1},
	})
	issue := uuidBytes(70)
	seedCommand(t, opened.DB(), issue)
	if _, err := opened.AppendCommandEvent(context.Background(), appendEvent(issue, 1, "committed-first")); err != nil {
		t.Fatal(err)
	}
	second := appendEvent(issue, 2, "uncommitted-second")
	if _, err := opened.AppendCommandEvent(context.Background(), second); !errors.Is(err, injected) {
		t.Fatalf("second append error = %v", err)
	}
	assertCommandEventState(t, opened.DB(), issue, 1, 1)
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}

	reopened := openStore(t, dataDir)
	recovery, err := reopened.RecoverCommandSegments(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if recovery.SegmentsChecked != 1 || recovery.RecordsChecked != 1 || recovery.TailBytesTruncated == 0 || recovery.OrphansRemoved != 0 {
		t.Fatalf("recovery = %+v", recovery)
	}
	if result, err := reopened.AppendCommandEvent(context.Background(), second); err != nil || result.ThroughEventSeq != 2 {
		t.Fatalf("second retry = (%+v, %v)", result, err)
	}
	assertCommandEventState(t, reopened.DB(), issue, 2, 2)
	if err := reopened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestEventGapRejectedBeforeFileWriteAndRotation_HP_STORE_03(t *testing.T) {
	t.Parallel()

	dataDir := filepath.Join(t.TempDir(), "state")
	opened := openStoreWithOptions(t, store.Options{DataDir: dataDir, BusyTimeout: busyTimeout, SegmentTargetSize: 200})
	issue := uuidBytes(80)
	seedCommand(t, opened.DB(), issue)
	if _, err := opened.AppendCommandEvent(context.Background(), appendEvent(issue, 2, "gap")); !errors.Is(err, store.ErrEventSequenceGap) {
		t.Fatalf("gap error = %v", err)
	}
	entries, err := os.ReadDir(filepath.Join(dataDir, "segments"))
	if err != nil || len(entries) != 0 {
		t.Fatalf("segments after gap = %v, err = %v", entries, err)
	}
	for sequence, payload := range []string{"one", "two"} {
		if _, err := opened.AppendCommandEvent(context.Background(), appendEvent(issue, uint64(sequence+1), payload)); err != nil {
			t.Fatal(err)
		}
	}
	var segmentCount int
	if err := opened.DB().QueryRow(`SELECT count(*) FROM output_segments WHERE issue_uuid = ?`, issue[:]).Scan(&segmentCount); err != nil || segmentCount != 2 {
		t.Fatalf("segment count = %d, err = %v", segmentCount, err)
	}
	if err := opened.QuickCheck(context.Background()); err != nil {
		t.Fatal(err)
	}
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestEventQuotaCountersAtomicAndCapacityRejectsBeforeWrite_HP_STORE_07(t *testing.T) {
	t.Parallel()

	dataDir := filepath.Join(t.TempDir(), "state")
	limits := store.QuotaLimits{
		HardAllocationBytes: 1000, CommandOutputBytes: 1000, CommandTotalBytes: 2000,
		ClientTotalBytes: 3000, ServerTotalBytes: 4000, CloseoutReserveBytes: 200, FilesystemFloorBytes: 50,
	}
	opened := openStoreWithOptions(t, store.Options{
		DataDir: dataDir, BusyTimeout: busyTimeout, QuotaLimits: limits,
		FreeSpaceProbe: store.FreeSpaceProbeFunc(func(string) (uint64, error) { return 10_000, nil }),
	})
	issue := uuidBytes(90)
	seedCommand(t, opened.DB(), issue)
	if _, err := opened.DB().Exec(`UPDATE commands SET closeout_remaining_bytes = ? WHERE issue_uuid = ?`, limits.CloseoutReserveBytes, issue[:]); err != nil {
		t.Fatal(err)
	}
	first := appendEvent(issue, 1, "first")
	if _, err := opened.AppendCommandEvent(context.Background(), first); err != nil {
		t.Fatal(err)
	}
	var commandOutput, commandTotal, closeout, clientTotal, serverTotal uint64
	if err := opened.DB().QueryRow(`SELECT output_charged_bytes, charged_bytes, closeout_remaining_bytes FROM commands WHERE issue_uuid = ?`, issue[:]).Scan(&commandOutput, &commandTotal, &closeout); err != nil {
		t.Fatal(err)
	}
	if err := opened.DB().QueryRow(`SELECT charged_bytes FROM clients WHERE client_id = 'client-a'`).Scan(&clientTotal); err != nil {
		t.Fatal(err)
	}
	if err := opened.DB().QueryRow(`SELECT command_charged_bytes FROM storage_counters WHERE singleton = 1`).Scan(&serverTotal); err != nil {
		t.Fatal(err)
	}
	if commandOutput == 0 || commandOutput != commandTotal || commandTotal != clientTotal || clientTotal != serverTotal || closeout != limits.CloseoutReserveBytes {
		t.Fatalf("quota counters = output:%d command:%d client:%d server:%d closeout:%d", commandOutput, commandTotal, clientTotal, serverTotal, closeout)
	}
	entries, err := os.ReadDir(filepath.Join(dataDir, "segments"))
	if err != nil || len(entries) != 1 {
		t.Fatalf("segment entries = %v, err = %v", entries, err)
	}
	segmentPath := filepath.Join(dataDir, "segments", entries[0].Name())
	before, err := os.Stat(segmentPath)
	if err != nil {
		t.Fatal(err)
	}
	if result, err := opened.AppendCommandEvent(context.Background(), first); err != nil || !result.Duplicate {
		t.Fatalf("duplicate = (%+v, %v)", result, err)
	}
	second := appendEvent(issue, 2, "second")
	if _, err := opened.AppendCommandEvent(context.Background(), second); err == nil {
		t.Fatal("over-quota event was accepted")
	} else {
		var capacity *store.CapacityError
		if !errors.As(err, &capacity) || capacity.Tier != store.CapacityTierCommandOut {
			t.Fatalf("capacity error = %v", err)
		}
	}
	after, err := os.Stat(segmentPath)
	if err != nil {
		t.Fatal(err)
	}
	if after.Size() != before.Size() {
		t.Fatalf("rejected event changed segment size from %d to %d", before.Size(), after.Size())
	}
	assertCommandEventState(t, opened.DB(), issue, 1, 1)
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestFilesystemFloorAndCounterMismatch_BH_STORE_08(t *testing.T) {
	t.Parallel()

	dataDir := filepath.Join(t.TempDir(), "state")
	opened := openStoreWithOptions(t, store.Options{
		DataDir: dataDir, BusyTimeout: busyTimeout,
		FreeSpaceProbe: store.FreeSpaceProbeFunc(func(string) (uint64, error) { return 0, nil }),
	})
	issue := uuidBytes(110)
	seedCommand(t, opened.DB(), issue)
	if _, err := opened.AppendCommandEvent(context.Background(), appendEvent(issue, 1, "floor")); err == nil {
		t.Fatal("filesystem-floor event was accepted")
	} else {
		var capacity *store.CapacityError
		if !errors.As(err, &capacity) || capacity.Tier != store.CapacityTierFilesystem {
			t.Fatalf("floor error = %v", err)
		}
	}
	entries, err := os.ReadDir(filepath.Join(dataDir, "segments"))
	if err != nil || len(entries) != 0 {
		t.Fatalf("segments after floor rejection = %v, err = %v", entries, err)
	}
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}

	reopened := openStore(t, dataDir)
	if _, err := reopened.DB().Exec(`UPDATE clients SET charged_bytes = 1 WHERE client_id = 'client-a'`); err != nil {
		t.Fatal(err)
	}
	if _, err := reopened.RecoverCommandSegments(context.Background()); !errors.Is(err, store.ErrQuotaCounterMismatch) {
		t.Fatalf("counter mismatch recovery error = %v", err)
	}
	if err := reopened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestTerminalAgeRetentionEvictsWholeCommand_HP_STORE_08(t *testing.T) {
	t.Parallel()

	dataDir := filepath.Join(t.TempDir(), "state")
	opened := openStore(t, dataDir)
	now := time.Unix(2_000_000, 0).UTC()
	old := uuidBytes(120)
	active := uuidBytes(140)
	seedCommand(t, opened.DB(), old)
	seedCommand(t, opened.DB(), active)
	if _, err := opened.AppendCommandEvent(context.Background(), appendEvent(old, 1, "retained history")); err != nil {
		t.Fatal(err)
	}
	markTerminal(t, opened.DB(), old, now.Add(-40*24*time.Hour), now.Add(-31*24*time.Hour))
	if _, err := opened.DB().Exec(`UPDATE commands SET issue_time = ? WHERE issue_uuid = ?`, now.Add(-100*24*time.Hour).UnixNano(), active[:]); err != nil {
		t.Fatal(err)
	}
	report, err := opened.RunRetention(context.Background(), now, store.RetentionPolicy{
		TerminalAge: 30 * 24 * time.Hour, TombstoneMaxEntries: 1_000_000,
	})
	if err != nil {
		t.Fatal(err)
	}
	if report.CommandsEvicted != 1 || report.FilesDeleted != 1 || report.BytesReleased == 0 {
		t.Fatalf("retention report = %+v", report)
	}
	var commands, tombstones int
	if err := opened.DB().QueryRow(`SELECT count(*) FROM commands WHERE issue_uuid = ?`, old[:]).Scan(&commands); err != nil || commands != 0 {
		t.Fatalf("old command count = %d, err = %v", commands, err)
	}
	if err := opened.DB().QueryRow(`SELECT count(*) FROM commands WHERE issue_uuid = ?`, active[:]).Scan(&commands); err != nil || commands != 1 {
		t.Fatalf("active command count = %d, err = %v", commands, err)
	}
	if err := opened.DB().QueryRow(`SELECT count(*) FROM command_tombstones WHERE issue_uuid = ?`, old[:]).Scan(&tombstones); err != nil || tombstones != 1 {
		t.Fatalf("tombstone count = %d, err = %v", tombstones, err)
	}
	for _, directory := range []string{"segments", "deleting"} {
		entries, err := os.ReadDir(filepath.Join(dataDir, directory))
		if err != nil || len(entries) != 0 {
			t.Fatalf("%s entries = %v, err = %v", directory, entries, err)
		}
	}
	if _, err := opened.RecoverCommandSegments(context.Background()); err != nil {
		t.Fatal(err)
	}
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestTombstoneFIFOIsCappedInEvictionTransaction_HP_STORE_09(t *testing.T) {
	t.Parallel()

	opened := openStore(t, filepath.Join(t.TempDir(), "state"))
	now := time.Unix(3_000_000, 0).UTC()
	first := uuidBytes(150)
	second := uuidBytes(170)
	seedCommand(t, opened.DB(), first)
	seedCommand(t, opened.DB(), second)
	markTerminal(t, opened.DB(), first, now.Add(-2*time.Hour), now.Add(-time.Hour))
	markTerminal(t, opened.DB(), second, now.Add(-time.Hour), now.Add(-time.Hour))
	report, err := opened.RunRetention(context.Background(), now, store.RetentionPolicy{PressureBytes: 1, TombstoneMaxEntries: 1})
	if err != nil {
		t.Fatal(err)
	}
	// Zero-charge commands cannot satisfy byte pressure, so every eligible terminal command is reclaimed.
	if report.CommandsEvicted != 2 || report.TombstonesTrimmed != 1 {
		t.Fatalf("retention report = %+v", report)
	}
	var count int
	var remaining []byte
	if err := opened.DB().QueryRow(`SELECT count(*), max(issue_uuid) FROM command_tombstones`).Scan(&count, &remaining); err != nil {
		t.Fatal(err)
	}
	if count != 1 || !bytes.Equal(remaining, second[:]) {
		t.Fatalf("tombstones = count %d remaining %x, want %x", count, remaining, second)
	}
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestRetentionCrashStagesRollForward_CRASH_STORE_04(t *testing.T) {
	t.Parallel()

	for index, checkpoint := range []string{
		store.FaultAfterEvictionMarked,
		store.FaultAfterEvictionFilesMoved,
		store.FaultAfterEvictionMetadataGone,
	} {
		t.Run(checkpoint, func(t *testing.T) {
			dataDir := filepath.Join(t.TempDir(), "state")
			injected := errors.New("injected retention crash")
			opened := openStoreWithOptions(t, store.Options{
				DataDir: dataDir, BusyTimeout: busyTimeout,
				FaultInjector: &failOnceInjector{target: checkpoint, failure: injected},
			})
			now := time.Unix(4_000_000+int64(index), 0).UTC()
			issue := uuidBytes(byte(190 + index*20))
			seedCommand(t, opened.DB(), issue)
			if _, err := opened.AppendCommandEvent(context.Background(), appendEvent(issue, 1, "evict me")); err != nil {
				t.Fatal(err)
			}
			markTerminal(t, opened.DB(), issue, now.Add(-2*time.Hour), now.Add(-time.Hour))
			if _, err := opened.RunRetention(context.Background(), now, store.RetentionPolicy{PressureBytes: 1, TombstoneMaxEntries: 10}); !errors.Is(err, injected) {
				t.Fatalf("RunRetention error = %v", err)
			}
			if err := opened.Close(); err != nil {
				t.Fatal(err)
			}

			reopened := openStore(t, dataDir)
			if _, err := reopened.RecoverEvictions(context.Background(), 10); err != nil {
				t.Fatal(err)
			}
			var commandCount, tombstoneCount int
			if err := reopened.DB().QueryRow(`SELECT count(*) FROM commands WHERE issue_uuid = ?`, issue[:]).Scan(&commandCount); err != nil {
				t.Fatal(err)
			}
			if err := reopened.DB().QueryRow(`SELECT count(*) FROM command_tombstones WHERE issue_uuid = ?`, issue[:]).Scan(&tombstoneCount); err != nil {
				t.Fatal(err)
			}
			if commandCount != 0 || tombstoneCount != 1 {
				t.Fatalf("recovered counts = commands %d tombstones %d", commandCount, tombstoneCount)
			}
			for _, directory := range []string{"segments", "deleting"} {
				entries, err := os.ReadDir(filepath.Join(dataDir, directory))
				if err != nil || len(entries) != 0 {
					t.Fatalf("%s entries = %v, err = %v", directory, entries, err)
				}
			}
			if _, err := reopened.RecoverCommandSegments(context.Background()); err != nil {
				t.Fatal(err)
			}
			if err := reopened.Close(); err != nil {
				t.Fatal(err)
			}
		})
	}
}

func TestIncidentDirtyResolutionIdempotencyAndRecurrence_HP_STORE_11(t *testing.T) {
	t.Parallel()

	opened := openStore(t, filepath.Join(t.TempDir(), "state"))
	now := time.Unix(5_000_000, 0).UTC()
	firstID := uuidBytes(10)
	input := store.IncidentInput{
		IncidentUUID: firstID, DetectedAt: now, Kind: store.IncidentCounterMismatch,
		Scope: store.IncidentScopeClient, ScopeKey: "client-a", ClientID: "client-a",
		Summary: "charged totals disagree", Evidence: []byte("counter-version=1"), AutomaticallyRepairable: true,
	}
	created, err := opened.RecordIncident(context.Background(), input)
	if err != nil || !created.Created || created.IncidentUUID != firstID {
		t.Fatalf("RecordIncident = (%+v, %v)", created, err)
	}
	duplicateInput := input
	duplicateInput.IncidentUUID = uuidBytes(30)
	duplicate, err := opened.RecordIncident(context.Background(), duplicateInput)
	if err != nil || duplicate.Created || duplicate.IncidentUUID != firstID {
		t.Fatalf("duplicate RecordIncident = (%+v, %v)", duplicate, err)
	}
	dirty, err := opened.HasDirtyIncidents(context.Background())
	if err != nil || !dirty {
		t.Fatalf("dirty = %v, err = %v", dirty, err)
	}
	requestID := uuidBytes(50)
	resolution := store.IncidentResolution{
		RequestUUID: requestID, IncidentUUID: firstID, State: store.IncidentRepaired,
		Note: "recomputed counters from command rows", ResolvedAt: now.Add(time.Minute),
	}
	resolved, err := opened.ResolveIncident(context.Background(), resolution)
	if err != nil || resolved.State != store.IncidentRepaired {
		t.Fatalf("ResolveIncident = (%+v, %v)", resolved, err)
	}
	replayed, err := opened.ResolveIncident(context.Background(), resolution)
	if err != nil || replayed != resolved {
		t.Fatalf("resolution replay = (%+v, %v)", replayed, err)
	}
	conflict := resolution
	conflict.Note = "different repair claim"
	if _, err := opened.ResolveIncident(context.Background(), conflict); !errors.Is(err, store.ErrMutationConflict) {
		t.Fatalf("resolution conflict error = %v", err)
	}
	dirty, err = opened.HasDirtyIncidents(context.Background())
	if err != nil || dirty {
		t.Fatalf("resolved dirty = %v, err = %v", dirty, err)
	}
	var auditCount int
	if err := opened.DB().QueryRow(`SELECT count(*) FROM audit_events WHERE action = 'repair_storage_incident'`).Scan(&auditCount); err != nil || auditCount != 1 {
		t.Fatalf("repair audit count = %d, err = %v", auditCount, err)
	}
	recurrence := input
	recurrence.IncidentUUID = uuidBytes(70)
	recurrence.DetectedAt = now.Add(2 * time.Minute)
	recorded, err := opened.RecordIncident(context.Background(), recurrence)
	if err != nil || !recorded.Created || recorded.IncidentUUID != recurrence.IncidentUUID {
		t.Fatalf("incident recurrence = (%+v, %v)", recorded, err)
	}
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestIrreparableIncidentRequiresExplicitAcknowledgement_BH_STORE_09(t *testing.T) {
	t.Parallel()

	opened := openStore(t, filepath.Join(t.TempDir(), "state"))
	now := time.Unix(6_000_000, 0).UTC()
	incidentID := uuidBytes(90)
	if _, err := opened.RecordIncident(context.Background(), store.IncidentInput{
		IncidentUUID: incidentID, DetectedAt: now, Kind: store.IncidentMissingCommittedBytes,
		Scope: store.IncidentScopeSegment, ScopeKey: "segment-redacted", Summary: "committed bytes are missing",
		Evidence: []byte("offset=128"), DataLoss: true,
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := opened.ResolveIncident(context.Background(), store.IncidentResolution{
		RequestUUID: uuidBytes(110), IncidentUUID: incidentID, State: store.IncidentRepaired,
		Note: "cannot really repair", ResolvedAt: now.Add(time.Minute),
	}); !errors.Is(err, store.ErrIncidentResolution) {
		t.Fatalf("unsafe repair error = %v", err)
	}
	if _, err := opened.ResolveIncident(context.Background(), store.IncidentResolution{
		RequestUUID: uuidBytes(130), IncidentUUID: incidentID, State: store.IncidentAcknowledged,
		ResolvedAt: now.Add(2 * time.Minute),
	}); !errors.Is(err, store.ErrIncidentResolution) {
		t.Fatalf("empty acknowledgement error = %v", err)
	}
	acknowledged, err := opened.ResolveIncident(context.Background(), store.IncidentResolution{
		RequestUUID: uuidBytes(150), IncidentUUID: incidentID, State: store.IncidentAcknowledged,
		Note: "operator accepts loss after external verification", ResolvedAt: now.Add(3 * time.Minute),
	})
	if err != nil || acknowledged.State != store.IncidentAcknowledged {
		t.Fatalf("acknowledgement = (%+v, %v)", acknowledged, err)
	}
	dirty, err := opened.HasDirtyIncidents(context.Background())
	if err != nil || dirty {
		t.Fatalf("acknowledged dirty = %v, err = %v", dirty, err)
	}
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestSessionRegistrationFencingAndExactTakeover_HP_SES_02(t *testing.T) {
	t.Parallel()

	opened := openStore(t, filepath.Join(t.TempDir(), "state"))
	now := time.Unix(7_000_000, 0).UTC()
	instanceA := uuidBytes(1)
	first := registerClient(t, opened, "client-a", instanceA, uuidBytes(20), now)
	if first.Generation != 1 {
		t.Fatalf("first generation = %d", first.Generation)
	}
	if clientID, err := opened.ValidateLiveSession(context.Background(), first.SessionID, first.Generation); err != nil || clientID != "client-a" {
		t.Fatalf("first session validation = (%q, %v)", clientID, err)
	}
	replacement := registerClient(t, opened, "client-a", instanceA, uuidBytes(40), now.Add(time.Second))
	if replacement.Generation != 2 {
		t.Fatalf("replacement generation = %d", replacement.Generation)
	}
	if _, err := opened.ValidateLiveSession(context.Background(), first.SessionID, first.Generation); !errors.Is(err, store.ErrSessionUnknown) {
		t.Fatalf("fenced first session error = %v", err)
	}

	instanceB := uuidBytes(60)
	_, err := opened.RegisterClientSession(context.Background(), store.ClientRegistration{
		ClientID: "client-a", Platform: 3, Architecture: "amd64", DaemonVersion: "test", DaemonCWD: `C:\work`,
		SupportedShells: []byte{1}, ClientInstanceID: instanceB, SessionID: uuidBytes(80), ConnectedAt: now.Add(2 * time.Second),
	})
	if !errors.Is(err, store.ErrTakeoverRequired) {
		t.Fatalf("different live instance error = %v", err)
	}
	var pending []byte
	if err := opened.DB().QueryRow(`SELECT pending_instance_id FROM clients WHERE client_id = 'client-a'`).Scan(&pending); err != nil || !bytes.Equal(pending, instanceB[:]) {
		t.Fatalf("pending instance = %x, err = %v", pending, err)
	}
	grantRequest := uuidBytes(100)
	expires := time.Now().UTC().Add(5 * time.Minute)
	granted, replay, err := opened.AuthorizeClientTakeover(context.Background(), store.TakeoverAuthorization{
		ClientID: "client-a", ClientInstanceID: instanceB, RequestID: grantRequest, ExpiresAt: expires,
	})
	if err != nil || replay || !granted.Equal(expires) {
		t.Fatalf("takeover grant = (%s, replay=%v, err=%v)", granted, replay, err)
	}
	granted, replay, err = opened.AuthorizeClientTakeover(context.Background(), store.TakeoverAuthorization{
		ClientID: "client-a", ClientInstanceID: instanceB, RequestID: grantRequest, ExpiresAt: expires,
	})
	if err != nil || !replay || !granted.Equal(expires) {
		t.Fatalf("takeover replay = (%s, replay=%v, err=%v)", granted, replay, err)
	}
	accepted := registerClient(t, opened, "client-a", instanceB, uuidBytes(120), now.Add(3*time.Second))
	if accepted.Generation != 3 {
		t.Fatalf("takeover generation = %d", accepted.Generation)
	}
	if _, err := opened.ValidateLiveSession(context.Background(), replacement.SessionID, replacement.Generation); !errors.Is(err, store.ErrSessionUnknown) {
		t.Fatalf("takeover did not fence prior session: %v", err)
	}
	if clientID, err := opened.ValidateLiveSession(context.Background(), accepted.SessionID, accepted.Generation); err != nil || clientID != "client-a" {
		t.Fatalf("accepted session validation = (%q, %v)", clientID, err)
	}
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestTakeoverBoundariesAndClosedSession_BH_SES_02(t *testing.T) {
	t.Parallel()

	opened := openStore(t, filepath.Join(t.TempDir(), "state"))
	now := time.Unix(8_000_000, 0).UTC()
	instanceA := uuidBytes(140)
	active := registerClient(t, opened, "client-b", instanceA, uuidBytes(160), now)
	instanceB := uuidBytes(180)
	_, err := opened.RegisterClientSession(context.Background(), store.ClientRegistration{
		ClientID: "client-b", Platform: 3, Architecture: "amd64", DaemonVersion: "test", DaemonCWD: `C:\work`,
		SupportedShells: []byte{1}, ClientInstanceID: instanceB, SessionID: uuidBytes(200), ConnectedAt: now.Add(time.Second),
	})
	if !errors.Is(err, store.ErrTakeoverRequired) {
		t.Fatalf("takeover-needed error = %v", err)
	}
	if _, _, err := opened.AuthorizeClientTakeover(context.Background(), store.TakeoverAuthorization{
		ClientID: "client-b", ClientInstanceID: uuidBytes(220), RequestID: uuidBytes(10), ExpiresAt: time.Now().UTC().Add(time.Minute),
	}); !errors.Is(err, store.ErrTakeoverMismatch) {
		t.Fatalf("mismatched grant error = %v", err)
	}
	request := uuidBytes(30)
	if _, _, err := opened.AuthorizeClientTakeover(context.Background(), store.TakeoverAuthorization{
		ClientID: "client-b", ClientInstanceID: instanceB, RequestID: request, ExpiresAt: time.Now().UTC().Add(time.Minute),
	}); err != nil {
		t.Fatal(err)
	}
	if _, _, err := opened.AuthorizeClientTakeover(context.Background(), store.TakeoverAuthorization{
		ClientID: "client-b", ClientInstanceID: instanceB, RequestID: uuidBytes(50), ExpiresAt: time.Now().UTC().Add(time.Minute),
	}); !errors.Is(err, store.ErrTakeoverAlreadyGranted) {
		t.Fatalf("second active grant error = %v", err)
	}
	if err := opened.CloseLiveSession(context.Background(), active.SessionID, active.Generation, "test close", now.Add(2*time.Second)); err != nil {
		t.Fatal(err)
	}
	if err := opened.CloseLiveSession(context.Background(), active.SessionID, active.Generation, "again", now.Add(3*time.Second)); !errors.Is(err, store.ErrSessionUnknown) {
		t.Fatalf("duplicate close error = %v", err)
	}
	// With no live session, a new instance is accepted without consuming the prior grant.
	accepted := registerClient(t, opened, "client-b", instanceB, uuidBytes(70), now.Add(4*time.Second))
	if accepted.Generation != 2 {
		t.Fatalf("closed-session registration generation = %d", accepted.Generation)
	}
	instanceC := uuidBytes(90)
	if _, err := opened.RegisterClientSession(context.Background(), store.ClientRegistration{
		ClientID: "client-b", Platform: 3, Architecture: "amd64", DaemonVersion: "test", DaemonCWD: `C:\work`,
		SupportedShells: []byte{1}, ClientInstanceID: instanceC, SessionID: uuidBytes(110), ConnectedAt: now.Add(5 * time.Second),
	}); !errors.Is(err, store.ErrTakeoverRequired) {
		t.Fatalf("post-cleanup takeover-needed error = %v", err)
	}
	if _, _, err := opened.AuthorizeClientTakeover(context.Background(), store.TakeoverAuthorization{
		ClientID: "client-b", ClientInstanceID: instanceC, RequestID: uuidBytes(130), ExpiresAt: time.Now().UTC().Add(time.Minute),
	}); err != nil {
		t.Fatalf("stale grant blocked fresh pending instance: %v", err)
	}
	if err := opened.Close(); err != nil {
		t.Fatal(err)
	}
}

func openStore(t *testing.T, dataDir string) *store.Store {
	t.Helper()
	return openStoreWithOptions(t, store.Options{DataDir: dataDir, BusyTimeout: busyTimeout})
}

func openStoreWithOptions(t *testing.T, options store.Options) *store.Store {
	t.Helper()
	opened, err := store.Open(context.Background(), options)
	if err != nil {
		t.Fatalf("Open(%s): %v", options.DataDir, err)
	}
	return opened
}

func assertPragma(t *testing.T, database *sql.DB, name, want string) {
	t.Helper()
	var got string
	if err := database.QueryRow(`PRAGMA ` + name).Scan(&got); err != nil {
		t.Fatalf("PRAGMA %s: %v", name, err)
	}
	if strings.ToLower(got) != want {
		t.Fatalf("PRAGMA %s = %q, want %q", name, got, want)
	}
}

func bytesOf(length int, value byte) []byte {
	result := make([]byte, length)
	for index := range result {
		result[index] = value
	}
	return result
}

func uuidBytes(start byte) [16]byte {
	var result [16]byte
	for index := range result {
		result[index] = start + byte(index)
	}
	return result
}

func integrationRecord(owner [16]byte, sequence uint64, payload string) store.SegmentRecord {
	return store.SegmentRecord{
		OwnerUUID: owner, Sequence: sequence, ObservedUnixNano: int64(sequence), ReceiptUnixNano: int64(sequence + 1),
		Kind: store.PayloadKindCommandEvent, Stream: 1, Compression: 1, RawLength: uint64(len(payload)), Payload: []byte(payload),
	}
}

type failOnceInjector struct {
	mu      sync.Mutex
	target  string
	failure error
	skip    int
	fired   bool
}

func (injector *failOnceInjector) Checkpoint(name string) error {
	injector.mu.Lock()
	defer injector.mu.Unlock()
	if name == injector.target && !injector.fired {
		if injector.skip > 0 {
			injector.skip--
			return nil
		}
		injector.fired = true
		return injector.failure
	}
	return nil
}

func seedCommand(t *testing.T, database *sql.DB, issue [16]byte) {
	t.Helper()
	clientInstance := uuidBytes(100)
	if _, err := database.Exec(`INSERT OR IGNORE INTO clients (
client_id, platform, architecture, daemon_version, daemon_cwd, supported_shells, capabilities, client_instance_id
) VALUES ('client-a', 3, 'amd64', 'test', 'C:\\work', x'01', x'', ?)`, clientInstance[:]); err != nil {
		t.Fatal(err)
	}
	requestHash := sha256.Sum256([]byte("request"))
	if _, err := database.Exec(`INSERT INTO commands (
issue_uuid, client_id, issue_time, server_receipt_time, lifecycle, revision, immutable_request_sha256,
execution_spec, execution_spec_raw_bytes, execution_spec_stored_bytes, execution_spec_compression
) VALUES (?, 'client-a', 1, 1, 1, 1, ?, x'01', 1, 1, 1)`, issue[:], requestHash[:]); err != nil {
		t.Fatal(err)
	}
}

func appendEvent(issue [16]byte, sequence uint64, payload string) store.EventAppend {
	immutableHash := sha256.Sum256(append([]byte{byte(sequence)}, []byte(payload)...))
	return store.EventAppend{
		IssueUUID: issue, EventSeq: sequence, ObservedUnixNano: int64(sequence), ReceiptUnixNano: int64(sequence + 10),
		EventType: 1, Stream: 1, Compression: 1, RawLength: uint64(len(payload)), Payload: []byte(payload),
		ImmutableSHA256: immutableHash,
	}
}

func assertCommandEventState(t *testing.T, database *sql.DB, issue [16]byte, wantLast, wantEvents int) {
	t.Helper()
	var last, events int
	if err := database.QueryRow(`SELECT last_event_seq FROM commands WHERE issue_uuid = ?`, issue[:]).Scan(&last); err != nil {
		t.Fatal(err)
	}
	if err := database.QueryRow(`SELECT count(*) FROM command_events WHERE issue_uuid = ?`, issue[:]).Scan(&events); err != nil {
		t.Fatal(err)
	}
	if last != wantLast || events != wantEvents {
		t.Fatalf("event state = (last=%d, rows=%d), want (%d, %d)", last, events, wantLast, wantEvents)
	}
}

func markTerminal(t *testing.T, database *sql.DB, issue [16]byte, issueTime, terminalTime time.Time) {
	t.Helper()
	if _, err := database.Exec(`UPDATE commands SET issue_time = ?, lifecycle = 5, terminal_time = ? WHERE issue_uuid = ?`, issueTime.UnixNano(), terminalTime.UnixNano(), issue[:]); err != nil {
		t.Fatal(err)
	}
}

func registerClient(t *testing.T, opened *store.Store, clientID string, instance, session [16]byte, at time.Time) store.SessionRegistration {
	t.Helper()
	registration, err := opened.RegisterClientSession(context.Background(), store.ClientRegistration{
		ClientID: clientID, Platform: 3, Architecture: "amd64", DaemonVersion: "test", DaemonCWD: `C:\work`,
		SupportedShells: []byte{1}, ClientInstanceID: instance, SessionID: session, ConnectedAt: at,
	})
	if err != nil {
		t.Fatalf("RegisterClientSession: %v", err)
	}
	return registration
}