package persistence

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

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

type capturedParentLaneHeartbeatEvents struct {
	mu     sync.Mutex
	events []parentLaneHeartbeatEvent
}

type completionCancelledHeartbeatBackend struct {
	*fakeParentLaneBackend

	mu      sync.Mutex
	calls   int
	started chan struct{}
	once    sync.Once
}

func newCompletionCancelledHeartbeatBackend() *completionCancelledHeartbeatBackend {
	return &completionCancelledHeartbeatBackend{
		fakeParentLaneBackend: newFakeParentLaneBackend(),
		started:               make(chan struct{}),
	}
}

func (b *completionCancelledHeartbeatBackend) heartbeat(ctx context.Context, keyHash, ownerToken string, now, leaseExpiresAt time.Time, lease time.Duration) (bool, error) {
	b.mu.Lock()
	b.calls++
	call := b.calls
	b.mu.Unlock()
	if call == 1 {
		return b.fakeParentLaneBackend.heartbeat(ctx, keyHash, ownerToken, now, leaseExpiresAt, lease)
	}
	b.once.Do(func() { close(b.started) })
	<-ctx.Done()
	return false, ctx.Err()
}

func (c *capturedParentLaneHeartbeatEvents) observe(_ context.Context, event parentLaneHeartbeatEvent) {
	c.mu.Lock()
	c.events = append(c.events, event)
	c.mu.Unlock()
}

func (c *capturedParentLaneHeartbeatEvents) has(action, result string, laneCount int) bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	for _, event := range c.events {
		if event.action == action && event.result == result && event.laneCount == laneCount {
			return true
		}
	}
	return false
}

func (c *capturedParentLaneHeartbeatEvents) hasKey(action, result, keyHash string) bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	for _, event := range c.events {
		if event.action == action && event.result == result && event.keyHash == keyHash {
			return true
		}
	}
	return false
}

func TestParentLaneHeartbeatExtendsLongTransactionOwnership(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)
	}
	txCtx := ports.WithTransactionAbort(context.Background())
	observed := &capturedParentLaneHeartbeatEvents{}
	controller, err := newParentLaneHeartbeatController(txCtx, store, []ParentLaneOwnership{ownerA}, ParentLaneHeartbeatOptions{Lease: 10 * time.Second, observer: observed.observe})
	if err != nil {
		t.Fatalf("newParentLaneHeartbeatController: %v", err)
	}

	clock.Advance(6 * time.Second)
	if !controller.heartbeatOnce(txCtx) {
		t.Fatal("heartbeatOnce returned false, want success")
	}
	clock.Advance(6 * time.Second)
	_, err = store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-b", Lease: 10 * time.Second, WaitTimeout: time.Nanosecond, PollInterval: time.Nanosecond})
	if err == nil {
		t.Fatal("Acquire owner-b while heartbeat-extended owner-a is live = nil, want contention")
	}
	if !errors.Is(err, ErrParentLaneContention) {
		t.Fatalf("Acquire owner-b error = %T(%v), want ErrParentLaneContention", err, err)
	}
	if got := backend.heartbeats(ownerA.KeyHash); got != 1 {
		t.Fatalf("heartbeats(owner-a) = %d, want 1", got)
	}
	if !observed.has("extend", "success", 1) {
		t.Fatalf("heartbeat events = %#v, want extend success", observed.events)
	}

	controller.releaseAll(ports.TransactionCompletionCommitted)
	if rows := backend.snapshot(); len(rows) != 0 {
		t.Fatalf("rows after completion release = %#v, want empty", rows)
	}
	if _, err := store.Acquire(context.Background(), key, ParentLaneAcquireOptions{OwnerToken: "owner-b", Lease: 10 * time.Second}); err != nil {
		t.Fatalf("Acquire owner-b after release: %v", err)
	}
}

func TestParentLaneHeartbeatAddsOwnerToRunningController(t *testing.T) {
	backend := newFakeParentLaneBackend()
	store := newParentLaneStoreForBackend(backend, realParentLaneClock{})
	ownerA, err := store.Acquire(context.Background(), recordlocks.ParentKey{ObjectName: "Account", RecordID: "001"}, ParentLaneAcquireOptions{OwnerToken: "owner-a", Lease: time.Minute})
	if err != nil {
		t.Fatalf("Acquire owner-a: %v", err)
	}
	ownerB, err := store.Acquire(context.Background(), recordlocks.ParentKey{ObjectName: "Contact", RecordID: "003"}, ParentLaneAcquireOptions{OwnerToken: "owner-b", Lease: time.Minute})
	if err != nil {
		t.Fatalf("Acquire owner-b: %v", err)
	}

	controller, err := newParentLaneHeartbeatController(ports.WithTransactionAbort(context.Background()), store, []ParentLaneOwnership{ownerA}, ParentLaneHeartbeatOptions{Lease: time.Minute})
	if err != nil {
		t.Fatalf("newParentLaneHeartbeatController: %v", err)
	}
	if err := controller.AddOwner(ownerB); err != nil {
		t.Fatalf("AddOwner: %v", err)
	}
	if !controller.heartbeatOnce(context.Background()) {
		t.Fatal("heartbeatOnce = false, want success")
	}
	if got := backend.heartbeats(ownerA.KeyHash); got != 1 {
		t.Fatalf("heartbeats(owner-a) = %d, want 1", got)
	}
	if got := backend.heartbeats(ownerB.KeyHash); got != 1 {
		t.Fatalf("heartbeats(owner-b) = %d, want 1", got)
	}
}

func TestParentLaneHeartbeatDoesNotAddOwnerAfterLoopCancellation(t *testing.T) {
	backend := newFakeParentLaneBackend()
	store := newParentLaneStoreForBackend(backend, realParentLaneClock{})
	ownerA, err := store.Acquire(context.Background(), recordlocks.ParentKey{ObjectName: "Account", RecordID: "001"}, ParentLaneAcquireOptions{OwnerToken: "owner-a", Lease: time.Minute})
	if err != nil {
		t.Fatalf("Acquire owner-a: %v", err)
	}
	ownerB, err := store.Acquire(context.Background(), recordlocks.ParentKey{ObjectName: "Contact", RecordID: "003"}, ParentLaneAcquireOptions{OwnerToken: "owner-b", Lease: time.Minute})
	if err != nil {
		t.Fatalf("Acquire owner-b: %v", err)
	}

	controller, err := newParentLaneHeartbeatController(ports.WithTransactionAbort(context.Background()), store, []ParentLaneOwnership{ownerA}, ParentLaneHeartbeatOptions{Lease: time.Minute})
	if err != nil {
		t.Fatalf("newParentLaneHeartbeatController: %v", err)
	}
	controller.cancel()
	if err := controller.AddOwner(ownerB); !errors.Is(err, ErrParentLaneContention) {
		t.Fatalf("AddOwner after stop error = %T(%v), want ErrParentLaneContention", err, err)
	}
	if owners := controller.snapshotOwners(); len(owners) != 1 || owners[0] != ownerA {
		t.Fatalf("owners after stopped AddOwner = %#v, want only %#v", owners, ownerA)
	}
}

func TestParentLaneHeartbeatReleaseRetriesTimedOutAttempt(t *testing.T) {
	backend := newDeadlineOnceReleaseBackend()
	store := newParentLaneStoreForBackend(backend, realParentLaneClock{})
	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)
	}
	observed := &capturedParentLaneHeartbeatEvents{}
	controller, err := newParentLaneHeartbeatController(
		ports.WithTransactionAbort(context.Background()),
		store,
		[]ParentLaneOwnership{owner},
		ParentLaneHeartbeatOptions{ReleaseTimeout: 50 * time.Millisecond, observer: observed.observe},
	)
	if err != nil {
		t.Fatalf("newParentLaneHeartbeatController: %v", err)
	}

	controller.releaseAll(ports.TransactionCompletionCommitted)
	if rows := backend.snapshot(); len(rows) != 0 {
		t.Fatalf("rows after retried heartbeat release = %#v, want empty", rows)
	}
	if got := backend.releaseAttempts(); got != 2 {
		t.Fatalf("release attempts = %d, want 2", got)
	}
	if !observed.has("release", "success", 1) {
		t.Fatalf("events = %#v, want release success", observed.events)
	}
	if observed.has("release", "failure", 1) {
		t.Fatalf("events = %#v, do not want a failure after successful retry", observed.events)
	}
}

func TestParentLaneHeartbeatMultiKeyReleasePartialFailure(t *testing.T) {
	backend := newFakeParentLaneBackend()
	store := newParentLaneStoreForBackend(backend, realParentLaneClock{})
	ownerA, err := store.Acquire(context.Background(), recordlocks.ParentKey{ObjectName: "Account", RecordID: "001"}, ParentLaneAcquireOptions{OwnerToken: "owner-a", Lease: time.Minute})
	if err != nil {
		t.Fatalf("Acquire owner-a: %v", err)
	}
	ownerB, err := store.Acquire(context.Background(), recordlocks.ParentKey{ObjectName: "Account", RecordID: "002"}, ParentLaneAcquireOptions{OwnerToken: "owner-b", Lease: time.Minute})
	if err != nil {
		t.Fatalf("Acquire owner-b: %v", err)
	}
	backend.releaseErrFor[ownerA.KeyHash] = errors.New("release failed")
	observed := &capturedParentLaneHeartbeatEvents{}
	txCtx := ports.WithTransactionAbort(ports.WithTransactionCompletionHooks(context.Background()))
	controller, err := store.StartHeartbeat(txCtx, []ParentLaneOwnership{ownerA, ownerB}, ParentLaneHeartbeatOptions{Interval: time.Hour, Lease: time.Minute, observer: observed.observe})
	if err != nil {
		t.Fatalf("StartHeartbeat: %v", err)
	}
	ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRolledBack)
	select {
	case <-controller.Done():
	case <-time.After(time.Second):
		t.Fatal("heartbeat controller did not stop after completion")
	}
	if got := backend.heartbeats(ownerA.KeyHash); got != 1 {
		t.Fatalf("heartbeats(owner-a) = %d, want 1", got)
	}
	if got := backend.heartbeats(ownerB.KeyHash); got != 1 {
		t.Fatalf("heartbeats(owner-b) = %d, want 1", got)
	}
	if got := backend.releases(); got != 2 {
		t.Fatalf("release count = %d, want both lanes attempted", got)
	}
	rows := backend.snapshot()
	if len(rows) != 1 || rows[ownerA.KeyHash].OwnerToken != "owner-a" {
		t.Fatalf("rows after partial release = %#v, want only failed owner-a release", rows)
	}
	if !observed.has("extend", "success", 2) {
		t.Fatalf("events = %#v, want multi-key extend success", observed.events)
	}
	if !observed.hasKey("release", "failure", shortParentLaneHash(ownerA.KeyHash)) {
		t.Fatalf("events = %#v, want release failure with short hash", observed.events)
	}
	if observed.hasKey("release", "failure", ownerA.KeyHash) {
		t.Fatalf("events = %#v, raw key hash leaked instead of short hash", observed.events)
	}
	if !observed.has("release", "success", 1) {
		t.Fatalf("events = %#v, want release success for remaining lane", observed.events)
	}
}