package persistence

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"testing"
	"time"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
)

type fakeParentLaneClock struct {
	mu  sync.Mutex
	now time.Time
}

func newFakeParentLaneClock() *fakeParentLaneClock {
	return &fakeParentLaneClock{now: time.Date(2026, 7, 8, 9, 0, 0, 0, time.UTC)}
}

func (c *fakeParentLaneClock) Now() time.Time {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.now
}

func (c *fakeParentLaneClock) Sleep(ctx context.Context, d time.Duration) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	c.Advance(d)
	return nil
}

func (c *fakeParentLaneClock) Advance(d time.Duration) {
	c.mu.Lock()
	c.now = c.now.Add(d)
	c.mu.Unlock()
}

type fakeParentLaneBackend struct {
	mu             sync.Mutex
	rows           map[string]parentLaneRow
	tryInsertCount int
	releaseCount   int
	heartbeatCount map[string]int
	heartbeatErr   error
	releaseErrFor  map[string]error
}

func newFakeParentLaneBackend() *fakeParentLaneBackend {
	return &fakeParentLaneBackend{
		rows:           map[string]parentLaneRow{},
		heartbeatCount: map[string]int{},
		releaseErrFor:  map[string]error{},
	}
}

func (b *fakeParentLaneBackend) tryInsert(_ context.Context, row parentLaneRow, _ time.Duration) (bool, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.tryInsertCount++
	if _, exists := b.rows[row.KeyHash]; exists {
		return false, nil
	}
	b.rows[row.KeyHash] = row
	return true, nil
}

func (b *fakeParentLaneBackend) tryTakeover(_ context.Context, row parentLaneRow, now time.Time, _ time.Duration) (bool, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	existing, exists := b.rows[row.KeyHash]
	if !exists || existing.Key != row.Key || existing.LeaseExpiresAt.After(now) {
		return false, nil
	}
	b.rows[row.KeyHash] = row
	return true, nil
}

func (b *fakeParentLaneBackend) heartbeat(_ context.Context, keyHash, ownerToken string, now, leaseExpiresAt time.Time, _ time.Duration) (bool, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.heartbeatCount[keyHash]++
	if b.heartbeatErr != nil {
		return false, b.heartbeatErr
	}
	row, exists := b.rows[keyHash]
	if !exists || row.OwnerToken != ownerToken || !row.LeaseExpiresAt.After(now) {
		return false, nil
	}
	row.HeartbeatAt = now
	row.LeaseExpiresAt = leaseExpiresAt
	b.rows[keyHash] = row
	return true, nil
}

func (b *fakeParentLaneBackend) release(_ context.Context, keyHash, ownerToken string) (bool, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.releaseCount++
	if err := b.releaseErrFor[keyHash]; err != nil {
		return false, err
	}
	row, exists := b.rows[keyHash]
	if !exists || row.OwnerToken != ownerToken {
		return false, nil
	}
	delete(b.rows, keyHash)
	return true, nil
}

func (b *fakeParentLaneBackend) releases() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.releaseCount
}

func (b *fakeParentLaneBackend) inserts() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.tryInsertCount
}

func (b *fakeParentLaneBackend) heartbeats(keyHash string) int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.heartbeatCount[keyHash]
}

func (b *fakeParentLaneBackend) replaceOwner(owner ParentLaneOwnership, ownerToken string, leaseExpiresAt time.Time) {
	b.mu.Lock()
	defer b.mu.Unlock()
	row := b.rows[owner.KeyHash]
	row.OwnerToken = ownerToken
	row.LeaseExpiresAt = leaseExpiresAt
	b.rows[owner.KeyHash] = row
}

type blockingReleaseBackend struct {
	started chan struct{}
	done    chan struct{}
}

type deadlineOnceReleaseBackend struct {
	*fakeParentLaneBackend

	mu       sync.Mutex
	attempts int
}

func newDeadlineOnceReleaseBackend() *deadlineOnceReleaseBackend {
	return &deadlineOnceReleaseBackend{fakeParentLaneBackend: newFakeParentLaneBackend()}
}

func (b *deadlineOnceReleaseBackend) release(ctx context.Context, keyHash, ownerToken string) (bool, error) {
	b.mu.Lock()
	b.attempts++
	attempt := b.attempts
	b.mu.Unlock()
	if attempt == 1 {
		<-ctx.Done()
		return false, ctx.Err()
	}
	return b.fakeParentLaneBackend.release(ctx, keyHash, ownerToken)
}

func (b *deadlineOnceReleaseBackend) releaseAttempts() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.attempts
}

type blockingAcquireBackend struct {
	*fakeParentLaneBackend

	mu        sync.Mutex
	unblock   chan struct{}
	active    int
	maxActive int
}

func newBlockingAcquireBackend() *blockingAcquireBackend {
	return &blockingAcquireBackend{
		fakeParentLaneBackend: newFakeParentLaneBackend(),
		unblock:               make(chan struct{}),
	}
}

func (b *blockingAcquireBackend) tryInsert(ctx context.Context, row parentLaneRow, lease time.Duration) (bool, error) {
	if row.OwnerToken == "owner-a" {
		return b.fakeParentLaneBackend.tryInsert(ctx, row, lease)
	}
	b.mu.Lock()
	b.active++
	if b.active > b.maxActive {
		b.maxActive = b.active
	}
	b.mu.Unlock()
	defer func() {
		b.mu.Lock()
		b.active--
		b.mu.Unlock()
	}()
	select {
	case <-ctx.Done():
		return false, ctx.Err()
	case <-b.unblock:
		return b.fakeParentLaneBackend.tryInsert(ctx, row, lease)
	}
}

func (b *blockingAcquireBackend) maxConcurrentAcquires() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.maxActive
}

func (b *blockingReleaseBackend) tryInsert(context.Context, parentLaneRow, time.Duration) (bool, error) {
	return false, nil
}

func (b *blockingReleaseBackend) tryTakeover(context.Context, parentLaneRow, time.Time, time.Duration) (bool, error) {
	return false, nil
}

func (b *blockingReleaseBackend) heartbeat(context.Context, string, string, time.Time, time.Time, time.Duration) (bool, error) {
	return false, nil
}

func (b *blockingReleaseBackend) release(ctx context.Context, _, _ string) (bool, error) {
	close(b.started)
	<-ctx.Done()
	close(b.done)
	return false, ctx.Err()
}

func (b *fakeParentLaneBackend) snapshot() map[string]parentLaneRow {
	b.mu.Lock()
	defer b.mu.Unlock()
	out := make(map[string]parentLaneRow, len(b.rows))
	for k, v := range b.rows {
		out[k] = v
	}
	return out
}

func TestParentLaneStoreAcquireReleaseHeartbeatStaleTakeover(t *testing.T) {
	clock := newFakeParentLaneClock()
	backend := newFakeParentLaneBackend()
	store := newParentLaneStoreForBackend(backend, clock)
	key := recordlocks.ParentKey{ObjectName: "Account", RecordID: "001"}

	ownerA, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-a", Lease: 10 * time.Second})
	if err != nil {
		t.Fatalf("Acquire owner-a: %v", err)
	}
	if ownerA.Key.ObjectName != "account" {
		t.Fatalf("canonical object = %q, want account", ownerA.Key.ObjectName)
	}

	clock.Advance(5 * time.Second)
	heartbeatA, ok, err := store.Heartbeat(context.Background(), ownerA, 10*time.Second)
	if err != nil || !ok {
		t.Fatalf("Heartbeat owner-a = ok %v err %v, want ok", ok, err)
	}
	if !heartbeatA.LeaseExpiresAt.Equal(clock.Now().Add(10 * time.Second)) {
		t.Fatalf("heartbeat lease = %s, want %s", heartbeatA.LeaseExpiresAt, clock.Now().Add(10*time.Second))
	}

	released, err := store.Release(context.Background(), heartbeatA)
	if err != nil || !released {
		t.Fatalf("Release owner-a = %v err %v, want released", released, err)
	}
	if rows := backend.snapshot(); len(rows) != 0 {
		t.Fatalf("rows after release = %#v, want empty", rows)
	}

	ownerA, err = store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-a", Lease: 10 * time.Second})
	if err != nil {
		t.Fatalf("Acquire owner-a again: %v", err)
	}
	clock.Advance(11 * time.Second)
	ownerB, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-b", Lease: 10 * time.Second})
	if err != nil {
		t.Fatalf("Acquire owner-b after stale lease: %v", err)
	}
	if ownerB.OwnerToken != "owner-b" {
		t.Fatalf("owner token = %q, want owner-b", ownerB.OwnerToken)
	}

	if _, ok, err := store.Heartbeat(context.Background(), ownerA, 10*time.Second); err != nil || ok {
		t.Fatalf("stale owner heartbeat = ok %v err %v, want false/nil", ok, err)
	}
	if released, err := store.Release(context.Background(), ownerA); err != nil || released {
		t.Fatalf("stale owner release = %v err %v, want false/nil", released, err)
	}
	rows := backend.snapshot()
	if len(rows) != 1 || rows[ownerB.KeyHash].OwnerToken != "owner-b" {
		t.Fatalf("rows after stale owner attempts = %#v, want owner-b live", rows)
	}
}

func TestParentLaneStoreDefaultWaitCoversOneLeaseForStaleTakeover(t *testing.T) {
	clock := newFakeParentLaneClock()
	backend := newFakeParentLaneBackend()
	store := newParentLaneStoreForBackend(backend, clock)
	store.pollInterval = time.Second
	key := recordlocks.ParentKey{ObjectName: "Parent__c", RecordID: "parent-001"}

	if _, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-a"}); err != nil {
		t.Fatalf("Acquire owner-a: %v", err)
	}
	started := clock.Now()
	ownerB, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-b"})
	if err != nil {
		t.Fatalf("Acquire owner-b through one default lease: %v", err)
	}
	if ownerB.OwnerToken != "owner-b" {
		t.Fatalf("owner token = %q, want owner-b", ownerB.OwnerToken)
	}
	if waited := clock.Now().Sub(started); waited < DefaultParentLaneLeaseDuration {
		t.Fatalf("waited %s, want at least one default lease %s", waited, DefaultParentLaneLeaseDuration)
	}
}

func TestParentLaneStoreDefaultWaitIncludesReleaseGrace(t *testing.T) {
	want := DefaultParentLaneLeaseDuration + DefaultParentLaneReleaseTimeout
	if DefaultParentLaneAcquireTimeout != want {
		t.Fatalf("default acquire timeout = %s, want lease plus release grace %s", DefaultParentLaneAcquireTimeout, want)
	}
}

func TestParentLaneStoreDefaultPollingAvoidsHotRowSpin(t *testing.T) {
	clock := newFakeParentLaneClock()
	backend := newFakeParentLaneBackend()
	store := newParentLaneStoreForBackend(backend, clock)
	key := recordlocks.ParentKey{ObjectName: "Parent__c", RecordID: "parent-001"}
	if _, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-a", Lease: time.Minute}); err != nil {
		t.Fatalf("Acquire owner-a: %v", err)
	}
	before := backend.inserts()
	_, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{
		OwnerToken:  "owner-b",
		Lease:       time.Minute,
		WaitTimeout: time.Second,
	})
	if err == nil || !errors.Is(err, ErrParentLaneContention) {
		t.Fatalf("Acquire owner-b error = %v, want parent-lane contention", err)
	}
	if attempts := backend.inserts() - before; attempts > 12 {
		t.Fatalf("durable insert polls in one second = %d, want at most 12", attempts)
	}
}

func TestParentLaneCompletionReleaseBudgetAllowsAnotherAttempt(t *testing.T) {
	if DefaultParentLaneCompletionReleaseTimeout <= DefaultParentLaneReleaseTimeout {
		t.Fatalf("completion release budget = %s, want greater than attempt timeout %s", DefaultParentLaneCompletionReleaseTimeout, DefaultParentLaneReleaseTimeout)
	}
}

func TestParentLaneStoreSerializesSameKeyAcquisitionPollingInProcess(t *testing.T) {
	backend := newBlockingAcquireBackend()
	store := newParentLaneStoreForBackend(backend, realParentLaneClock{})
	key := recordlocks.ParentKey{ObjectName: "Parent__c", RecordID: "parent-001"}

	if _, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-a"}); err != nil {
		t.Fatalf("Acquire owner-a: %v", err)
	}

	ctx, cancel := context.WithCancel(context.Background())
	const waiters = 16
	var wg sync.WaitGroup
	wg.Add(waiters)
	for i := 0; i < waiters; i++ {
		go func(i int) {
			defer wg.Done()
			_, _ = store.Acquire(ctx, key, ParentLaneAcquireOptions{OwnerToken: fmt.Sprintf("owner-%d", i+1)})
		}(i)
	}

	deadline := time.Now().Add(100 * time.Millisecond)
	for backend.maxConcurrentAcquires() < 2 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	cancel()
	wg.Wait()

	if got := backend.maxConcurrentAcquires(); got != 1 {
		t.Fatalf("concurrent durable acquisition pollers = %d, want 1 per parent key", got)
	}
	store.acquireGatesMu.Lock()
	remainingGates := len(store.acquireGates)
	store.acquireGatesMu.Unlock()
	if remainingGates != 0 {
		t.Fatalf("remaining acquisition gates = %d, want 0", remainingGates)
	}
}

func TestParentLaneStoreLocalQueueDoesNotConsumeDurableWaitBudget(t *testing.T) {
	backend := newFakeParentLaneBackend()
	store := newParentLaneStoreForBackend(backend, realParentLaneClock{})
	key := recordlocks.ParentKey{ObjectName: "Parent__c", RecordID: "parent-001"}
	canonical, ok := recordlocks.CanonicalParentLaneKey(key)
	if !ok {
		t.Fatal("CanonicalParentLaneKey = false, want true")
	}
	keyHash, ok := recordlocks.ParentLaneKeyHash(canonical)
	if !ok {
		t.Fatal("ParentLaneKeyHash = false, want true")
	}

	releaseGate, err := store.waitForAcquireGate(context.Background(), keyHash)
	if err != nil {
		t.Fatalf("hold local acquisition gate: %v", err)
	}
	gateReleased := false
	defer func() {
		if !gateReleased {
			releaseGate()
		}
	}()

	result := make(chan error, 1)
	go func() {
		_, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{
			OwnerToken:   "queued-owner",
			WaitTimeout:  20 * time.Millisecond,
			PollInterval: time.Millisecond,
		})
		result <- err
	}()

	time.Sleep(60 * time.Millisecond)
	select {
	case err := <-result:
		t.Fatalf("Acquire returned while waiting only on the local queue: %v", err)
	default:
	}

	releaseGate()
	gateReleased = true
	select {
	case err := <-result:
		if err != nil {
			t.Fatalf("Acquire after local queue handoff: %v", err)
		}
	case <-time.After(time.Second):
		t.Fatal("Acquire after local queue handoff timed out")
	}
}

func TestParentLaneStoreSameKeySingleLiveOwner(t *testing.T) {
	clock := newFakeParentLaneClock()
	backend := newFakeParentLaneBackend()
	store := newParentLaneStoreForBackend(backend, clock)
	key := recordlocks.ParentKey{ObjectName: "Account", RecordID: "001"}

	ownerA, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-a", Lease: time.Minute})
	if err != nil {
		t.Fatalf("Acquire owner-a: %v", err)
	}
	_, err = store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-b", Lease: time.Minute, WaitTimeout: time.Nanosecond, PollInterval: time.Nanosecond})
	if err == nil {
		t.Fatal("Acquire owner-b while owner-a live = nil, want contention")
	}
	var contention *apperrors.LockContentionError
	if !errors.As(err, &contention) {
		t.Fatalf("Acquire owner-b error = %T(%v), want LockContentionError", err, err)
	}
	if contention.HTTPStatus() != 409 {
		t.Fatalf("contention HTTPStatus = %d, want 409", contention.HTTPStatus())
	}

	released, err := store.Release(context.Background(), ownerA)
	if err != nil || !released {
		t.Fatalf("Release owner-a = %v err %v, want released", released, err)
	}
	if _, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-b", Lease: time.Minute}); err != nil {
		t.Fatalf("Acquire owner-b after release: %v", err)
	}
}

func TestParentLaneStoreDifferentKeyConcurrency(t *testing.T) {
	backend := newFakeParentLaneBackend()
	store := newParentLaneStoreForBackend(backend, realParentLaneClock{})
	keys := []recordlocks.ParentKey{
		{ObjectName: "Account", RecordID: "001"},
		{ObjectName: "Account", RecordID: "002"},
	}

	var wg sync.WaitGroup
	errs := make(chan error, len(keys))
	for i, key := range keys {
		wg.Add(1)
		go func(i int, key recordlocks.ParentKey) {
			defer wg.Done()
			_, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner", Lease: time.Minute})
			if err != nil {
				errs <- err
			}
		}(i, key)
	}
	wg.Wait()
	close(errs)
	for err := range errs {
		t.Fatalf("Acquire different key: %v", err)
	}
	if rows := backend.snapshot(); len(rows) != 2 {
		t.Fatalf("live rows = %d (%#v), want 2", len(rows), rows)
	}
}

func TestParentLaneStoreTimeoutReturnsRetryableContention(t *testing.T) {
	clock := newFakeParentLaneClock()
	backend := newFakeParentLaneBackend()
	store := newParentLaneStoreForBackend(backend, clock)
	key := recordlocks.ParentKey{ObjectName: "Account", RecordID: "001"}

	if _, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-a", Lease: time.Minute}); err != nil {
		t.Fatalf("Acquire owner-a: %v", err)
	}
	_, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-b", Lease: time.Minute, WaitTimeout: 5 * time.Millisecond, PollInterval: time.Millisecond})
	if err == nil {
		t.Fatal("Acquire owner-b = nil, want contention")
	}
	var internal *apperrors.InternalError
	if errors.As(err, &internal) {
		t.Fatalf("Acquire owner-b error = %v, want non-internal contention", err)
	}
	var contention *apperrors.LockContentionError
	if !errors.As(err, &contention) {
		t.Fatalf("Acquire owner-b error = %T(%v), want LockContentionError", err, err)
	}
}

func TestParentLaneStoreObservesAcquireOutcomeWithoutRawRecordIdentity(t *testing.T) {
	clock := newFakeParentLaneClock()
	backend := newFakeParentLaneBackend()
	store := newParentLaneStoreForBackend(backend, clock)
	var events []parentLaneAcquireEvent
	store.observeAcquire = func(_ context.Context, event parentLaneAcquireEvent) {
		events = append(events, event)
	}
	key := recordlocks.ParentKey{ObjectName: "Account", RecordID: "001-sensitive-record-id"}

	if _, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{
		OwnerToken: "owner-a",
		Lease:      time.Minute,
	}); err != nil {
		t.Fatalf("Acquire owner-a: %v", err)
	}
	_, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{
		OwnerToken:   "owner-b",
		Lease:        time.Minute,
		WaitTimeout:  5 * time.Millisecond,
		PollInterval: time.Millisecond,
	})
	if err == nil || !errors.Is(err, ErrParentLaneContention) {
		t.Fatalf("Acquire owner-b error = %v, want contention", err)
	}

	if len(events) != 2 {
		t.Fatalf("acquire events = %d, want 2", len(events))
	}
	if events[0].result != "inserted" || events[0].polls != 0 || events[0].err != nil {
		t.Fatalf("first acquire event = %#v, want immediate insert", events[0])
	}
	if events[1].result != "contention_timeout" || events[1].polls == 0 || !errors.Is(events[1].err, ErrParentLaneContention) {
		t.Fatalf("second acquire event = %#v, want polled contention", events[1])
	}
	for _, event := range events {
		if event.objectName != "account" {
			t.Fatalf("event object = %q, want canonical account", event.objectName)
		}
		if len(event.keyHash) != 12 || event.keyHash == key.RecordID {
			t.Fatalf("event key hash = %q, want short non-raw identity", event.keyHash)
		}
	}
}

func TestParentLaneReleaseOnTransactionCompletionUsesBoundedContext(t *testing.T) {
	backend := &blockingReleaseBackend{
		started: make(chan struct{}),
		done:    make(chan struct{}),
	}
	store := newParentLaneStoreForBackend(backend, newFakeParentLaneClock())
	store.releaseTimeout = time.Nanosecond
	txCtx := ports.WithTransactionCompletionHooks(context.Background())
	if !store.ReleaseOnTransactionCompletion(txCtx, ParentLaneOwnership{KeyHash: "hash", OwnerToken: "owner"}) {
		t.Fatal("ReleaseOnTransactionCompletion returned false")
	}

	ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRolledBack)
	select {
	case <-backend.started:
	default:
		t.Fatal("release was not attempted")
	}
	select {
	case <-backend.done:
	case <-time.After(time.Second):
		t.Fatal("bounded release did not return after context timeout")
	}
}

func TestParentLaneReleaseOnTransactionCompletionRetriesTimedOutAttempt(t *testing.T) {
	backend := newDeadlineOnceReleaseBackend()
	store := newParentLaneStoreForBackend(backend, realParentLaneClock{})
	store.releaseTimeout = 50 * time.Millisecond
	store.releaseAttemptTimeout = 5 * time.Millisecond
	store.releaseRetryInterval = time.Millisecond
	owner, err := store.Acquire(context.Background(), recordlocks.ParentKey{ObjectName: "Account", RecordID: "001"}, ParentLaneAcquireOptions{OwnerToken: "owner", Lease: time.Minute})
	if err != nil {
		t.Fatalf("Acquire: %v", err)
	}
	txCtx := ports.WithTransactionCompletionHooks(context.Background())
	if !store.ReleaseOnTransactionCompletion(txCtx, owner) {
		t.Fatal("ReleaseOnTransactionCompletion returned false")
	}

	ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionCommitted)
	if rows := backend.snapshot(); len(rows) != 0 {
		t.Fatalf("rows after retried completion release = %#v, want empty", rows)
	}
	if got := backend.releaseAttempts(); got != 2 {
		t.Fatalf("release attempts = %d, want 2", got)
	}
}

func TestParentLaneReleaseOnTransactionCompletion(t *testing.T) {
	for _, tc := range []struct {
		name      string
		committed bool
	}{
		{name: "commit", committed: true},
		{name: "rollback", committed: false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			clock := newFakeParentLaneClock()
			backend := newFakeParentLaneBackend()
			store := newParentLaneStoreForBackend(backend, clock)
			owner, err := store.Acquire(context.Background(), recordlocks.ParentKey{ObjectName: "Account", RecordID: "001"}, ParentLaneAcquireOptions{OwnerToken: "owner", Lease: time.Minute})
			if err != nil {
				t.Fatalf("Acquire: %v", err)
			}
			txCtx := ports.WithTransactionCompletionHooks(context.Background())
			if !store.ReleaseOnTransactionCompletion(txCtx, owner) {
				t.Fatal("ReleaseOnTransactionCompletion returned false")
			}
			outcome := ports.TransactionCompletionRolledBack
			if tc.committed {
				outcome = ports.TransactionCompletionCommitted
			}
			ports.RunTransactionCompletionHooks(txCtx, outcome)
			if rows := backend.snapshot(); len(rows) != 0 {
				t.Fatalf("rows after completion hook = %#v, want empty", rows)
			}
			if got := backend.releases(); got != 1 {
				t.Fatalf("release count = %d, want 1", got)
			}
		})
	}
}