package fanout

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

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/store/message"
	unreadservice "github.com/tidbcloud/AgentTeam/internal/unread"
)

type leaseTestCoordinator struct {
	mu sync.Mutex

	record        message.ScopeFanoutOutboxRecord
	available     bool
	missing       bool
	receipts      map[string]struct{}
	claims        int
	renews        int
	completions   int
	retries       int
	lastRetryErr  string
	currentOwner  string
	receiptWrites []string
}

func (coordinator *leaseTestCoordinator) MarkScopeFanoutDeliveredForMessage(context.Context, string) error {
	return nil
}

func (coordinator *leaseTestCoordinator) ClaimScopeFanoutForMessage(_ context.Context, _ string, owner string, leaseUntil time.Time) (message.ScopeFanoutOutboxRecord, bool, error) {
	coordinator.mu.Lock()
	defer coordinator.mu.Unlock()
	coordinator.claims++
	if coordinator.missing {
		return message.ScopeFanoutOutboxRecord{}, false, message.ErrNotFound
	}
	if !coordinator.available {
		return message.ScopeFanoutOutboxRecord{}, false, nil
	}
	coordinator.available = false
	coordinator.currentOwner = owner
	record := coordinator.record
	if record.Status == "pending_v2" || record.Status == "leased_v2" {
		record.Status = "leased_v2"
	} else {
		record.Status = "leased"
	}
	record.Attempts++
	record.LeaseOwner = owner
	record.LeaseExpiresAt = &leaseUntil
	coordinator.record = record
	return record, true, nil
}

func (coordinator *leaseTestCoordinator) RenewScopeFanoutLease(_ context.Context, outboxID string, owner string, _ time.Time) error {
	coordinator.mu.Lock()
	defer coordinator.mu.Unlock()
	if outboxID != coordinator.record.ID || owner != coordinator.currentOwner {
		return message.ErrNotFound
	}
	coordinator.renews++
	return nil
}

func (coordinator *leaseTestCoordinator) ScopeFanoutAgentDelivered(_ context.Context, eventType string, eventID string, agentID string) (bool, error) {
	coordinator.mu.Lock()
	defer coordinator.mu.Unlock()
	_, ok := coordinator.receipts[eventType+"\x00"+eventID+"\x00"+agentID]
	return ok, nil
}

func (coordinator *leaseTestCoordinator) RecordScopeFanoutAgentDelivered(_ context.Context, outboxID string, owner string, eventType string, eventID string, agentID string) error {
	coordinator.mu.Lock()
	defer coordinator.mu.Unlock()
	if outboxID != coordinator.record.ID || owner != coordinator.currentOwner {
		return message.ErrNotFound
	}
	key := eventType + "\x00" + eventID + "\x00" + agentID
	coordinator.receipts[key] = struct{}{}
	coordinator.receiptWrites = append(coordinator.receiptWrites, key)
	return nil
}

func (coordinator *leaseTestCoordinator) CompleteScopeFanoutLease(_ context.Context, outboxID string, owner string) error {
	coordinator.mu.Lock()
	defer coordinator.mu.Unlock()
	if outboxID != coordinator.record.ID || owner != coordinator.currentOwner {
		return message.ErrNotFound
	}
	coordinator.completions++
	coordinator.currentOwner = ""
	coordinator.record.Status = "delivered"
	coordinator.record.LeaseOwner = ""
	coordinator.record.LeaseExpiresAt = nil
	return nil
}

func (coordinator *leaseTestCoordinator) RetryScopeFanoutLease(_ context.Context, outboxID string, owner string, _ time.Time, lastError string) error {
	coordinator.mu.Lock()
	defer coordinator.mu.Unlock()
	if outboxID != coordinator.record.ID || owner != coordinator.currentOwner {
		return message.ErrNotFound
	}
	coordinator.retries++
	coordinator.lastRetryErr = lastError
	coordinator.available = true
	coordinator.currentOwner = ""
	if coordinator.record.Status == "leased_v2" {
		coordinator.record.Status = "pending_v2"
	} else {
		coordinator.record.Status = "pending"
	}
	coordinator.record.LeaseOwner = ""
	coordinator.record.LeaseExpiresAt = nil
	return nil
}

type leaseTestDeferringCoordinator struct {
	*leaseTestCoordinator
	deferrals int
}

func (coordinator *leaseTestDeferringCoordinator) DeferScopeFanoutLease(
	_ context.Context,
	outboxID string,
	owner string,
	_ time.Time,
	lastError string,
) error {
	coordinator.mu.Lock()
	defer coordinator.mu.Unlock()
	if outboxID != coordinator.record.ID ||
		owner != coordinator.currentOwner ||
		coordinator.record.Status != "leased_v2" {
		return message.ErrNotFound
	}
	coordinator.deferrals++
	coordinator.lastRetryErr = lastError
	coordinator.available = true
	coordinator.currentOwner = ""
	coordinator.record.Status = "pending_v2"
	if coordinator.record.Attempts > 0 {
		coordinator.record.Attempts--
	}
	coordinator.record.LeaseOwner = ""
	coordinator.record.LeaseExpiresAt = nil
	return nil
}

type perAgentFailureNotifier struct {
	mu        sync.Mutex
	failCount map[string]int
	calls     map[string]int
}

func (notifier *perAgentFailureNotifier) NotifyAgentMessage(_ context.Context, recipient message.AgentRecipient, _ int64, _ message.ReceiveMessage) error {
	notifier.mu.Lock()
	defer notifier.mu.Unlock()
	notifier.calls[recipient.AgentID]++
	if notifier.failCount[recipient.AgentID] > 0 {
		notifier.failCount[recipient.AgentID]--
		return errors.New("temporary daemon handoff failure")
	}
	return nil
}

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

	coordinator := &leaseTestCoordinator{
		available: true,
		receipts:  map[string]struct{}{},
		record: message.ScopeFanoutOutboxRecord{
			ID:           "out_task_partial",
			WorkspaceID:  "wsp_task_partial",
			ScopeID:      "tsk_task_partial",
			MessageID:    "msg_task_partial",
			EventRefType: message.DeliverySourceRefMessage,
			EventRefID:   "msg_task_partial",
			FanoutType:   message.ScopeOutboxFanoutScope,
		},
	}
	notifier := &perAgentFailureNotifier{
		failCount: map[string]int{"agt_b": 1},
		calls:     map[string]int{},
	}
	realtime := &fakeRealtime{}
	fanout := NewDeliveryFanout(notifier, realtime, nil).WithScopeFanoutMarker(coordinator)
	appMessage := message.AppMessage{
		ID:          "msg_task_partial",
		ChannelID:   "tsk_task_partial",
		ChannelType: "task",
		MessageType: message.MessageTypeTaskDiscussion,
		Content:     "please continue",
	}
	delivery := message.MessageDelivery{
		ThreadID:  "tsk_task_partial",
		Seq:       11,
		MessageID: appMessage.ID,
		Message: message.ReceiveMessage{
			MessageID:   appMessage.ID,
			TargetKey:   appMessage.ChannelID,
			TargetSeq:   11,
			ChannelType: "task",
			Content:     appMessage.Content,
		},
		WakeRecipients: []message.AgentRecipient{
			{AgentID: "agt_a", MachineID: "mac_a", AgentMemberID: "mem_a", AutoWakeEnabled: true},
			{AgentID: "agt_b", MachineID: "mac_b", AgentMemberID: "mem_b", AutoWakeEnabled: true},
		},
	}

	err := fanout.DeliverCommittedMessage(t.Context(), coordinator.record.WorkspaceID, appMessage, delivery)
	require.ErrorContains(t, err, "temporary daemon handoff failure")
	require.Equal(t, 1, coordinator.retries)
	require.Equal(t, 0, coordinator.completions)

	require.NoError(t, fanout.DeliverCommittedMessage(t.Context(), coordinator.record.WorkspaceID, appMessage, delivery))
	require.Equal(t, map[string]int{"agt_a": 1, "agt_b": 2}, notifier.calls)
	require.Equal(t, 2, coordinator.claims)
	require.Equal(t, 1, coordinator.completions)
	require.Len(t, coordinator.receipts, 2)
	require.GreaterOrEqual(t, coordinator.renews, 5)
	// Browser delivery is idempotent by message event id even when the first
	// attempt reached the stream before one Agent failed.
	require.Equal(t, []string{"tsk_task_partial"}, realtime.messageNew)
}

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

	const (
		workspaceID = "wsp_task_inline_deferred"
		scopeID     = "tsk_task_inline_deferred"
		messageID   = "msg_task_inline_deferred"
	)
	baseCoordinator := &leaseTestCoordinator{
		available: true,
		receipts:  map[string]struct{}{},
		record: message.ScopeFanoutOutboxRecord{
			ID:           "out_task_inline_deferred",
			WorkspaceID:  workspaceID,
			ScopeID:      scopeID,
			MessageID:    messageID,
			EventRefType: message.DeliverySourceRefMessage,
			EventRefID:   messageID,
			FanoutType:   message.ScopeOutboxFanoutScope,
			Status:       "pending_v2",
			MaxAttempts:  3,
		},
	}
	coordinator := &leaseTestDeferringCoordinator{leaseTestCoordinator: baseCoordinator}
	runtime, err := unreadservice.New(unreadservice.Config{})
	require.NoError(t, err)
	unread := &capturingUnreadRuntime{runtime: runtime}
	realtime := &fakeRealtime{}
	notifier := &capturingNotifier{}
	fanout := NewDeliveryFanout(notifier, realtime, nil).
		WithScopeFanoutMarker(coordinator).
		WithUnreadRuntime(unread).
		WithUnreadDeltaSink(realtime)
	appMessage := taskLifecycleAppMessage(scopeID)
	appMessage.ID = messageID
	delivery := taskLifecycleDelivery(scopeID)
	delivery.MessageID = messageID
	delivery.Message.MessageID = messageID

	for attempt := 0; attempt < 4; attempt++ {
		err = fanout.DeliverCommittedMessage(t.Context(), workspaceID, appMessage, delivery)
		require.ErrorIs(t, err, errTargetedTaskUnreadPending)
		require.ErrorIs(t, err, message.ErrScopeFanoutDeferred)
		require.Equal(t, "pending_v2", coordinator.record.Status)
		require.Zero(t, coordinator.record.Attempts, "defer must refund the inline lease attempt")
		require.Empty(t, notifier.messages)
		require.Zero(t, coordinator.completions)
		require.Zero(t, coordinator.retries, "durable defer must not use ordinary failure retry")
	}
	require.Equal(t, 4, coordinator.deferrals)
	require.Len(t, realtime.messageNew, 1, "stable stream ids must dedupe each deferred replay")

	require.NoError(t, runtime.WarmScopeRoster(workspaceID, scopeID, unreadservice.ScopeTypeTask, []unreadservice.RosterMember{
		{MemberID: "mem_author", UserID: "usr_author", Visibility: "public", NotificationState: "mute"},
		{MemberID: "mem_reader", UserID: "usr_reader", Visibility: "public", NotificationState: "mute"},
	}))
	require.NoError(t, fanout.DeliverCommittedMessage(t.Context(), workspaceID, appMessage, delivery))
	require.Equal(t, "delivered", coordinator.record.Status)
	require.Equal(t, 1, coordinator.record.Attempts)
	require.Equal(t, 1, coordinator.completions)
	require.Len(t, notifier.messages, 1)
	require.Equal(t, int64(1), runtime.ScopeUnreadCounts(workspaceID, "mem_reader", nil)[scopeID])
	require.Len(t, realtime.unreadUpdated, 1)
}

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

	const (
		workspaceID = "wsp_task_inline_snapshot_barrier"
		scopeID     = "tsk_task_inline_snapshot_barrier"
		messageID   = "msg_task_inline_snapshot_barrier"
	)
	baseCoordinator := &leaseTestCoordinator{
		available: true,
		receipts:  map[string]struct{}{},
		record: message.ScopeFanoutOutboxRecord{
			ID:           "out_task_inline_snapshot_barrier",
			WorkspaceID:  workspaceID,
			ScopeID:      scopeID,
			MessageID:    messageID,
			EventRefType: message.DeliverySourceRefMessage,
			EventRefID:   messageID,
			FanoutType:   message.ScopeOutboxFanoutScope,
			Status:       "pending_v2",
			MaxAttempts:  3,
		},
	}
	coordinator := &leaseTestDeferringCoordinator{leaseTestCoordinator: baseCoordinator}
	runtime, err := unreadservice.New(unreadservice.Config{})
	require.NoError(t, err)
	require.NoError(t, runtime.WarmScopeRoster(workspaceID, scopeID, unreadservice.ScopeTypeTask, []unreadservice.RosterMember{
		{MemberID: "mem_author", UserID: "usr_author", Visibility: "public", NotificationState: "mute"},
		{MemberID: "mem_reader", UserID: "usr_reader", Visibility: "public", NotificationState: "mute"},
	}))
	snapshotFailure := errors.New("snapshot store unavailable")
	unread := &capturingUnreadRuntime{runtime: runtime, persistErr: snapshotFailure}
	realtime := &fakeRealtime{}
	notifier := &capturingNotifier{}
	fanout := NewDeliveryFanout(notifier, realtime, nil).
		WithScopeFanoutMarker(coordinator).
		WithUnreadRuntime(unread).
		WithUnreadDeltaSink(realtime)
	appMessage := taskLifecycleAppMessage(scopeID)
	appMessage.ID = messageID
	delivery := taskLifecycleDelivery(scopeID)
	delivery.MessageID = messageID
	delivery.Message.MessageID = messageID

	err = fanout.DeliverCommittedMessage(t.Context(), workspaceID, appMessage, delivery)
	require.ErrorIs(t, err, message.ErrScopeFanoutDeferred)
	require.ErrorIs(t, err, snapshotFailure)
	require.Equal(t, "pending_v2", coordinator.record.Status)
	require.Zero(t, coordinator.record.Attempts)
	require.Equal(t, 1, coordinator.deferrals)
	require.Zero(t, coordinator.completions)
	require.Empty(t, notifier.messages, "Agent wake must wait for durable Human unread")
	require.Equal(t, []taskUnreadPersistCall{{
		workspaceID: workspaceID,
		scopeID:     scopeID,
		memberIDs:   []string{"mem_reader"},
	}}, unread.persistCalls)
	require.Len(t, realtime.unreadUpdated, 1, "the already-applied live unread delta must not be stranded")

	unread.persistErr = nil
	require.NoError(t, fanout.DeliverCommittedMessage(t.Context(), workspaceID, appMessage, delivery))
	require.Equal(t, "delivered", coordinator.record.Status)
	require.Equal(t, 1, coordinator.record.Attempts)
	require.Equal(t, 1, coordinator.completions)
	require.Len(t, unread.persistCalls, 2, "idempotent replay must retry the durability barrier")
	require.Len(t, notifier.messages, 1)
	require.Len(t, realtime.unreadUpdated, 1, "the replayed accounted event must not double emit unread")
}

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

	newCoordinator := func(status string) *leaseTestDeferringCoordinator {
		return &leaseTestDeferringCoordinator{leaseTestCoordinator: &leaseTestCoordinator{
			available: true,
			receipts:  map[string]struct{}{},
			record: message.ScopeFanoutOutboxRecord{
				ID:           "out_task_inline_missing_unread_" + status,
				WorkspaceID:  "wsp_task_inline_missing_unread",
				ScopeID:      "tsk_task_inline_missing_unread",
				MessageID:    "msg_task_inline_missing_unread",
				EventRefType: message.DeliverySourceRefMessage,
				EventRefID:   "msg_task_inline_missing_unread",
				FanoutType:   message.ScopeOutboxFanoutScope,
				Status:       status,
				MaxAttempts:  3,
			},
		}}
	}
	appMessage := taskLifecycleAppMessage("tsk_task_inline_missing_unread")
	appMessage.ID = "msg_task_inline_missing_unread"
	delivery := taskLifecycleDelivery("tsk_task_inline_missing_unread")
	delivery.MessageID = appMessage.ID
	delivery.Message.MessageID = appMessage.ID

	t.Run("targeted_v2_fails_closed", func(t *testing.T) {
		coordinator := newCoordinator("pending_v2")
		realtime := &fakeRealtime{}
		notifier := &capturingNotifier{}
		fanout := NewDeliveryFanout(notifier, realtime, nil).
			WithScopeFanoutMarker(coordinator)

		err := fanout.DeliverCommittedMessage(t.Context(), coordinator.record.WorkspaceID, appMessage, delivery)
		require.ErrorIs(t, err, message.ErrScopeFanoutDeferred)
		require.ErrorIs(t, err, errTargetedTaskUnreadPending)
		require.Equal(t, "pending_v2", coordinator.record.Status)
		require.Zero(t, coordinator.record.Attempts)
		require.Equal(t, 1, coordinator.deferrals)
		require.Empty(t, notifier.messages)
		require.Zero(t, coordinator.completions)
	})

	t.Run("legacy_v1_keeps_best_effort_delivery", func(t *testing.T) {
		coordinator := newCoordinator("pending")
		realtime := &fakeRealtime{}
		notifier := &capturingNotifier{}
		fanout := NewDeliveryFanout(notifier, realtime, nil).
			WithScopeFanoutMarker(coordinator)
		legacyDelivery := delivery
		legacyDelivery.HumanRecipientsTargeted = false
		legacyDelivery.HumanRecipientMemberIDs = nil

		require.NoError(t, fanout.DeliverCommittedMessage(
			t.Context(),
			coordinator.record.WorkspaceID,
			appMessage,
			legacyDelivery,
		))
		require.Equal(t, "delivered", coordinator.record.Status)
		require.Equal(t, 1, coordinator.record.Attempts)
		require.Len(t, notifier.messages, 1)
		require.Equal(t, 1, coordinator.completions)
		require.Zero(t, coordinator.deferrals)
	})
}

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

	coordinator := &leaseTestCoordinator{
		available: false,
		receipts:  map[string]struct{}{},
		record: message.ScopeFanoutOutboxRecord{
			ID:           "out_task_owned",
			WorkspaceID:  "wsp_task_owned",
			ScopeID:      "tsk_task_owned",
			MessageID:    "msg_task_owned",
			EventRefType: message.DeliverySourceRefMessage,
			EventRefID:   "msg_task_owned",
			FanoutType:   message.ScopeOutboxFanoutScope,
		},
	}
	notifier := &perAgentFailureNotifier{failCount: map[string]int{}, calls: map[string]int{}}
	realtime := &fakeRealtime{}
	fanout := NewDeliveryFanout(notifier, realtime, nil).WithScopeFanoutMarker(coordinator)

	require.NoError(t, fanout.DeliverCommittedMessage(t.Context(), coordinator.record.WorkspaceID, message.AppMessage{
		ID: "msg_task_owned", ChannelID: "tsk_task_owned", ChannelType: "task", MessageType: message.MessageTypeTaskDiscussion,
	}, message.MessageDelivery{
		ThreadID: "tsk_task_owned", MessageID: "msg_task_owned",
		Message:        message.ReceiveMessage{MessageID: "msg_task_owned", TargetKey: "tsk_task_owned", ChannelType: "task"},
		WakeRecipients: []message.AgentRecipient{{AgentID: "agt_owned", MachineID: "mac_owned", AgentMemberID: "mem_owned", AutoWakeEnabled: true}},
	}))

	require.Equal(t, 1, coordinator.claims)
	require.Empty(t, notifier.calls)
	require.Empty(t, realtime.messageNew)
	require.Zero(t, coordinator.completions)
	require.Zero(t, coordinator.retries)
}