package agentops

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"testing"
	"time"
	"unicode/utf8"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
)

func TestNewOperationWorkerRejectsMissingDependencies(t *testing.T) {
	t.Parallel()
	observer := &fakeOperationTerminalObserver{}
	worker, err := NewOperationWorker(nil, &fakeOperationExecutor{}, observer, 0, nil)
	require.Nil(t, worker)
	require.EqualError(t, err, "operation worker dependencies are incomplete")
	worker, err = NewOperationWorker(&Store{}, nil, observer, 0, nil)
	require.Nil(t, worker)
	require.EqualError(t, err, "operation worker dependencies are incomplete")
	worker, err = NewOperationWorker(&Store{}, &fakeOperationExecutor{}, nil, 0, nil)
	require.Nil(t, worker)
	require.EqualError(t, err, "operation worker dependencies are incomplete")
}

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

	worker, err := NewOperationWorker(
		&Store{},
		&fakeOperationExecutor{},
		&fakeOperationTerminalObserver{},
		0,
		nil,
		WithOperationWorkspaceGroupID(" group-a "),
	)
	require.NoError(t, err)

	require.Equal(t, "group-a", worker.workspaceGroupID)
}

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

	called := make(chan struct{}, 1)
	worker, err := NewOperationWorker(
		&Store{}, &fakeOperationExecutor{}, &fakeOperationTerminalObserver{}, time.Hour, nil,
	)
	require.NoError(t, err)
	worker.processOnce = func(context.Context) {
		called <- struct{}{}
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	go worker.Run(ctx)

	select {
	case <-called:
		cancel()
	case <-time.After(time.Second):
		t.Fatal("Run did not process immediately")
	}
}

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

	var calls int
	worker, err := NewOperationWorker(
		&Store{}, &fakeOperationExecutor{}, &fakeOperationTerminalObserver{}, time.Hour, nil,
	)
	require.NoError(t, err)
	worker.processOnce = func(context.Context) {
		calls++
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	worker.Run(ctx)

	require.Zero(t, calls)
}

type fakeOperationExecutor struct {
	results map[int]json.RawMessage
	errs    map[int]error
	calls   int
}

func (e *fakeOperationExecutor) ExecuteItem(_ context.Context, _ *Operation, item OperationItem) (json.RawMessage, error) {
	e.calls++
	if err, ok := e.errs[item.ItemIndex]; ok {
		return e.results[item.ItemIndex], err
	}
	return e.results[item.ItemIndex], nil
}

type fakeOperationTerminalObserver struct {
	inputs           []OperationTerminalInput
	errs             []error
	errByOperationID map[int64]error
}

func (observer *fakeOperationTerminalObserver) ObserveOperationTerminal(_ context.Context, input OperationTerminalInput) error {
	observer.inputs = append(observer.inputs, input)
	if err := observer.errByOperationID[input.Operation.ID]; err != nil {
		return err
	}
	if len(observer.errs) == 0 {
		return nil
	}
	err := observer.errs[0]
	observer.errs = observer.errs[1:]
	return err
}

func TestOperationWorkerProcessOnceAggregation(t *testing.T) {
	tests := []struct {
		name         string
		itemApplied  map[int]bool
		wantTerminal string
	}{
		{"all applied", map[int]bool{0: true, 1: true}, "applied"},
		{"mixed", map[int]bool{0: true, 1: false}, "partial"},
		{"all failed", map[int]bool{0: false, 1: false}, "failed"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctx := context.Background()
			store := startAgentOpsConcurrencyStore(t, ctx)

			items := make([]json.RawMessage, 0, len(tt.itemApplied))
			for i := 0; i < len(tt.itemApplied); i++ {
				items = append(items, json.RawMessage(`{}`))
			}
			var opID int64
			require.NoError(t, coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
				var err error
				opID, err = store.InTx(tx).EnqueueOperationInTx(ctx, EnqueueInput{
					WorkspaceID: "wsp-1", ApplicantAgentID: "agt-c", OwnerUserID: "usr-o",
					Action: "duplicate", Payload: json.RawMessage(`{}`),
					ApprovalID: "appr-" + tt.name, RequestID: "req-" + tt.name,
					Items: items,
				})
				return err
			}))

			executor := &fakeOperationExecutor{
				results: make(map[int]json.RawMessage),
				errs:    make(map[int]error),
			}
			for i, applied := range tt.itemApplied {
				if applied {
					executor.results[i] = json.RawMessage(`{"ok":true}`)
				} else {
					executor.errs[i] = errors.New("simulated failure")
				}
			}

			observer := &fakeOperationTerminalObserver{}
			worker, err := NewOperationWorker(store, executor, observer, 0, nil)
			require.NoError(t, err)
			worker.ProcessOnce(ctx)

			status := operationStatus(t, store, opID)
			require.Equal(t, tt.wantTerminal, status)
			require.Len(t, observer.inputs, 1)
			require.Equal(t, tt.wantTerminal, observer.inputs[0].TerminalStatus)
			require.Len(t, observer.inputs[0].Items, len(tt.itemApplied))
			for index, item := range observer.inputs[0].Items {
				require.Equal(t, index, item.ItemIndex)
			}

			// Operation is terminal → not re-claimable.
			op, err := store.ClaimNextOperation(ctx, "")
			require.NoError(t, err)
			require.Nil(t, op)
		})
	}
}

func TestOperationWorkerRetriesFinalizationWithoutExecutingItems(t *testing.T) {
	ctx := context.Background()
	store := startAgentOpsConcurrencyStore(t, ctx)
	opID := enqueueTestOperation(t, ctx, store, "wsp-finalization", "finalization")
	executor := &fakeOperationExecutor{
		results: map[int]json.RawMessage{0: json.RawMessage(`{"agentId":"agt-safe"}`)},
	}
	observer := &fakeOperationTerminalObserver{
		errs: []error{errors.New("injected observer failure")},
	}
	worker, err := NewOperationWorker(store, executor, observer, 0, nil)
	require.NoError(t, err)

	worker.ProcessOnce(ctx)
	require.Equal(t, "running", operationStatus(t, store, opID))
	require.Equal(t, 1, executor.calls)
	items, err := store.LoadItems(ctx, opID, false)
	require.NoError(t, err)
	require.Equal(t, "applied", items[0].Status)

	worker.ProcessOnce(ctx)
	require.Equal(t, "applied", operationStatus(t, store, opID))
	require.Equal(t, 1, executor.calls, "finalization retry must not execute items")
	require.Len(t, observer.inputs, 2)
}

func TestOperationWorkerFailedFinalizationDoesNotStarveQueuedWork(t *testing.T) {
	ctx := context.Background()
	store := startAgentOpsConcurrencyStore(t, ctx)
	poisonID := enqueueTestOperation(t, ctx, store, "wsp-final-poison", "final-poison")
	executor := &fakeOperationExecutor{
		results: map[int]json.RawMessage{
			0: json.RawMessage(`{"agentId":"agt-after-poison"}`),
		},
	}
	observer := &fakeOperationTerminalObserver{
		errByOperationID: map[int64]error{
			poisonID: errors.New("poison projection"),
		},
	}
	worker, err := NewOperationWorker(store, executor, observer, 0, nil)
	require.NoError(t, err)

	worker.ProcessOnce(ctx)
	require.Equal(t, "running", operationStatus(t, store, poisonID))
	require.Equal(t, 1, executor.calls)
	require.Len(t, observer.inputs, 1)
	require.Equal(t, poisonID, observer.inputs[0].Operation.ID)

	queuedID := enqueueTestOperation(t, ctx, store, "wsp-after-poison", "after-poison")
	worker.ProcessOnce(ctx)
	require.Equal(t, "applied", operationStatus(t, store, queuedID),
		"queued work executes despite the older poison finalization")
	require.Equal(t, 2, executor.calls)
	require.Len(t, observer.inputs, 2)
	require.Equal(t, queuedID, observer.inputs[1].Operation.ID)

	worker.ProcessOnce(ctx)
	require.Equal(t, "running", operationStatus(t, store, poisonID))
	require.Equal(t, 2, executor.calls,
		"poison finalization retry must not re-execute terminal items")
	require.Len(t, observer.inputs, 3)
	require.Equal(t, poisonID, observer.inputs[2].Operation.ID)

	secondQueuedID := enqueueTestOperation(
		t, ctx, store, "wsp-after-poison-2", "after-poison-2",
	)
	worker.ProcessOnce(ctx)
	require.Equal(t, "applied", operationStatus(t, store, secondQueuedID),
		"repeated poison projection failures must not starve later queued work")
	require.Equal(t, 3, executor.calls)
}

func TestOperationWorkerWorkspaceGroupFailedFinalizationDoesNotStarveLocalQueuedWork(t *testing.T) {
	ctx := context.Background()
	store := startAgentOpsConcurrencyStore(t, ctx)
	seedOperationWorkspacePlacements(t, ctx, store, []operationWorkspacePlacement{
		{workspaceID: "wsp-worker-poison", groupID: "group-a", state: "active"},
		{workspaceID: "wsp-worker-foreign", groupID: "group-b", state: "active"},
		{workspaceID: "wsp-worker-local", groupID: "group-a", state: "active"},
	})
	poisonID := enqueueTestOperation(t, ctx, store, "wsp-worker-poison", "worker-poison")
	executor := &fakeOperationExecutor{
		results: map[int]json.RawMessage{
			0: json.RawMessage(`{"agentId":"agt-worker-local"}`),
		},
	}
	observer := &fakeOperationTerminalObserver{
		errByOperationID: map[int64]error{
			poisonID: errors.New("poison projection"),
		},
	}
	worker, err := NewOperationWorker(
		store, executor, observer, 0, nil,
		WithOperationWorkspaceGroupID("group-a"),
	)
	require.NoError(t, err)

	worker.ProcessOnce(ctx)
	require.Equal(t, "running", operationStatus(t, store, poisonID))

	foreignID := enqueueTestOperation(t, ctx, store, "wsp-worker-foreign", "worker-foreign")
	localID := enqueueTestOperation(t, ctx, store, "wsp-worker-local", "worker-local")
	worker.ProcessOnce(ctx)
	require.Equal(t, "applied", operationStatus(t, store, localID),
		"local queued work executes despite an older local poison finalization")
	require.Equal(t, "queued", operationStatus(t, store, foreignID),
		"workspace-group worker must not claim foreign queued work")
	require.Len(t, observer.inputs, 2)
	require.Equal(t, localID, observer.inputs[1].Operation.ID)

	worker.ProcessOnce(ctx)
	require.Len(t, observer.inputs, 3)
	require.Equal(t, poisonID, observer.inputs[2].Operation.ID)
	require.Equal(t, "queued", operationStatus(t, store, foreignID))

	secondLocalID := enqueueTestOperation(
		t, ctx, store, "wsp-worker-local", "worker-local-2",
	)
	worker.ProcessOnce(ctx)
	require.Equal(t, "applied", operationStatus(t, store, secondLocalID),
		"repeated local poison failures must not starve later local queued work")
	require.Equal(t, "queued", operationStatus(t, store, foreignID))
}

func TestOperationWorkerRetriesAfterCompleteOperationFailure(t *testing.T) {
	ctx := context.Background()
	store := startAgentOpsConcurrencyStore(t, ctx)
	opID := enqueueTestOperation(t, ctx, store, "wsp-complete-retry", "complete-retry")
	executor := &fakeOperationExecutor{
		results: map[int]json.RawMessage{0: json.RawMessage(`{"agentId":"agt-safe"}`)},
	}
	observer := &fakeOperationTerminalObserver{}
	worker, err := NewOperationWorker(store, executor, observer, 0, nil)
	require.NoError(t, err)
	realComplete := worker.completeOperation
	failures := 1
	worker.completeOperation = func(ctx context.Context, id int64, status string) error {
		if failures > 0 {
			failures--
			return errors.New("injected complete failure")
		}
		return realComplete(ctx, id, status)
	}

	worker.ProcessOnce(ctx)
	require.Equal(t, "running", operationStatus(t, store, opID))
	require.Equal(t, 1, executor.calls)
	require.Len(t, observer.inputs, 1)

	worker.ProcessOnce(ctx)
	require.Equal(t, "applied", operationStatus(t, store, opID))
	require.Equal(t, 1, executor.calls)
	require.Len(t, observer.inputs, 2, "observer idempotently converges again")
}

func TestOperationWorkerZeroItemsKeepsRunning(t *testing.T) {
	ctx := context.Background()
	store := startAgentOpsConcurrencyStore(t, ctx)

	// Enqueue operation with 0 items via raw SQL (bypasses validation).
	var opID int64
	require.NoError(t, coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		_, err := tx.Exec(ctx, `INSERT INTO agent_operations (workspace_id, applicant_agent_id, owner_user_id, action, payload, approval_id, request_id, status) VALUES ($1,$2,$3,$4,$5,$6,$7,'queued')`,
			"wsp-1", "agt-c", "usr-o", "create", []byte(`{}`), "appr-zero", "req-zero")
		if err != nil {
			return err
		}
		return tx.QueryRow(ctx, `SELECT LAST_INSERT_ID()`).Scan(&opID)
	}))

	executor := &fakeOperationExecutor{}
	worker, err := NewOperationWorker(
		store, executor, &fakeOperationTerminalObserver{}, 0, nil,
	)
	require.NoError(t, err)
	worker.ProcessOnce(ctx)

	// Anomalous 0-items → stays running (not fabricated applied).
	require.Equal(t, "running", operationStatus(t, store, opID))
}

func operationStatus(t *testing.T, store *Store, opID int64) string {
	t.Helper()
	var status string
	err := store.db.QueryRow(context.Background(), `SELECT status FROM agent_operations WHERE id = $1`, opID).Scan(&status)
	require.NoError(t, err)
	return status
}

type cancelOnFirstExecutor struct {
	cancel context.CancelFunc
	count  *int
}

func (e *cancelOnFirstExecutor) ExecuteItem(_ context.Context, _ *Operation, item OperationItem) (json.RawMessage, error) {
	*e.count++
	if item.ItemIndex == 0 {
		e.cancel()
	}
	return json.RawMessage(`{}`), nil
}

func TestOperationWorkerContextCancelKeepsRunning(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	store := startAgentOpsConcurrencyStore(t, ctx)

	var opID int64
	require.NoError(t, coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		var err error
		opID, err = store.InTx(tx).EnqueueOperationInTx(ctx, EnqueueInput{
			WorkspaceID: "wsp-1", ApplicantAgentID: "agt-c", OwnerUserID: "usr-o",
			Action: "duplicate", Payload: json.RawMessage(`{}`),
			ApprovalID: "appr-cancel", RequestID: "req-cancel",
			Items: []json.RawMessage{json.RawMessage(`{}`), json.RawMessage(`{}`)},
		})
		return err
	}))

	var callCount int
	executor := &cancelOnFirstExecutor{cancel: cancel, count: &callCount}
	worker, err := NewOperationWorker(
		store, executor, &fakeOperationTerminalObserver{}, 0, nil,
	)
	require.NoError(t, err)
	worker.ProcessOnce(ctx)

	require.Equal(t, 1, callCount, "only the first item was called before cancel")
	require.Equal(t, "running", operationStatus(t, store, opID), "operation stays running on cancel")

	// Both items stay pending after cancel (no CompleteItem was called). Use
	// context.Background() because the test ctx is already cancelled.
	pendingItems, err := store.LoadItems(context.Background(), opID, true)
	require.NoError(t, err)
	require.Len(t, pendingItems, 2, "both items remain pending after cancel")
}

func TestTruncateItemErrorUTF8Safe(t *testing.T) {
	t.Parallel()
	// 1366 Chinese chars (3 bytes each) = 4098 > 4096. Must cut at a rune boundary.
	s := strings.Repeat("世", 1366)
	result := truncateItemError(s)
	require.LessOrEqual(t, len(result), maxItemErrorLen)
	require.True(t, utf8.ValidString(result), "truncated string must be valid UTF-8")
	// Short string unchanged.
	require.Equal(t, "short", truncateItemError("short"))
}