package connector_inbound

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"strings"
	"testing"
	"time"

	connector_events "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/constants"
)

type recordingRuntimeCommandHandoff struct {
	requests     []RuntimeCommandHandoffRequest
	resumeScopes []RuntimeCommandHandoffScope
	resumeFound  bool
	resumeErr    error
	err          error
	readyErr     error
}

func (r *recordingRuntimeCommandHandoff) VerifyReady(context.Context) error {
	return r.readyErr
}

func (r *recordingRuntimeCommandHandoff) ResumePreparedRuntimeCommand(
	_ context.Context,
	scope RuntimeCommandHandoffScope,
) (bool, error) {
	r.resumeScopes = append(r.resumeScopes, scope)
	return r.resumeFound, r.resumeErr
}

func (r *recordingRuntimeCommandHandoff) AcceptRuntimeCommand(
	_ context.Context,
	request RuntimeCommandHandoffRequest,
) error {
	r.requests = append(r.requests, request)
	return r.err
}

func TestRuntimeCommandHandoffScopeRequiresExactWorkItemRouteLease(t *testing.T) {
	scope := RuntimeCommandHandoffScope{
		WorkItemID: " work-1 ", EventID: " event-1 ",
		LeaseToken: " lease-1 ", Namespace: " connector-pack ",
		InstalledPackID: " install-1 ", EventKind: " CONTACT.CREATED ",
		SourcePartition: " source-1 ", RouteVersion: 7,
	}.normalized()
	if err := scope.validate(); err != nil {
		t.Fatalf("valid scope rejected: %v", err)
	}
	if scope.WorkItemID != "work-1" || scope.EventID != "event-1" ||
		scope.LeaseToken != "lease-1" || scope.EventKind != "contact.created" ||
		scope.SourcePartition != "source-1" || scope.RouteVersion != 7 {
		t.Fatalf("Runtime command handoff scope = %#v", scope)
	}

	scope.RouteVersion = 0
	if err := scope.validate(); err == nil {
		t.Fatal("scope without an immutable route version was accepted")
	}
}

func TestRuntimeCommandHandoffFailureKeepsRetryClassification(t *testing.T) {
	cause := errors.New("Runtime unavailable")
	err := &RuntimeCommandHandoffFailure{Cause: cause, Retriable: true}
	if !runtimeCommandHandoffIsRetriable(err) || !errors.Is(err, cause) {
		t.Fatalf("handoff failure classification lost: %v", err)
	}
}

func TestNewRuntimeCommandHandoffWorkerDoesNotInstallFlowInvoker(t *testing.T) {
	service := newCompleteWorkItemServiceFake(
		&fakeWorkItemClaims{},
		&fakeWorkItemEventLoader{},
		nil,
	)
	handoff := &recordingRuntimeCommandHandoff{resumeFound: true}
	cfg := mandatoryWorkerTestConfig()
	cfg.FlowDeadline = time.Minute
	worker, err := NewRuntimeCommandHandoffWorker(
		context.Background(),
		cfg,
		service,
		&fakeSyncState{},
		fakeBatchRead{},
		handoff,
		&stubRouteResolver{},
	)
	if err != nil {
		t.Fatalf("NewRuntimeCommandHandoffWorker: %v", err)
	}
	if worker.flow != nil || worker.runtimeCommands != handoff {
		t.Fatalf(
			"execution owners = flow:%T Runtime:%T",
			worker.flow,
			worker.runtimeCommands,
		)
	}
}

func TestNewRuntimeCommandHandoffWorkerRejectsMissingDeadlineBeforeClaim(t *testing.T) {
	service := newCompleteWorkItemServiceFake(
		&fakeWorkItemClaims{},
		&fakeWorkItemEventLoader{},
		nil,
	)
	_, err := NewRuntimeCommandHandoffWorker(
		context.Background(),
		mandatoryWorkerTestConfig(),
		service,
		&fakeSyncState{},
		fakeBatchRead{},
		&recordingRuntimeCommandHandoff{},
		&stubRouteResolver{},
	)
	if err == nil || !strings.Contains(err.Error(), "positive FlowDeadline") {
		t.Fatalf("NewRuntimeCommandHandoffWorker() error = %v", err)
	}
}

func TestNewRuntimeCommandHandoffWorkerFailsBeforeClaimWhenReceiptSchemaIsMissing(t *testing.T) {
	service := newCompleteWorkItemServiceFake(
		&fakeWorkItemClaims{},
		&fakeWorkItemEventLoader{},
		nil,
	)
	cfg := mandatoryWorkerTestConfig()
	cfg.FlowDeadline = time.Minute
	_, err := NewRuntimeCommandHandoffWorker(
		context.Background(),
		cfg,
		service,
		&fakeSyncState{},
		fakeBatchRead{},
		&recordingRuntimeCommandHandoff{readyErr: errors.New("receipt table missing")},
		&stubRouteResolver{},
	)
	if err == nil || !strings.Contains(err.Error(), "receipt table missing") {
		t.Fatalf("NewRuntimeCommandHandoffWorker() error = %v", err)
	}
}

func TestRuntimeCommandHandoffTransfersOwnershipBeforeWorkItemCompletion(t *testing.T) {
	events := newFakeEvents()
	handoff := &recordingRuntimeCommandHandoff{}
	worker := &Worker{
		cfg: WorkerConfig{
			Namespace: "connector-pack", FinalizeTimeout: time.Second,
		},
		events: events, syncstate: &fakeSyncState{}, runtimeCommands: handoff,
	}
	event := runtimeCommandHandoffEvent()
	ctx := constants.WithConnectorInboundTrace(
		context.Background(),
		constants.ConnectorInboundTrace{
			WorkItemID: "work-1", EventID: event.ID,
			Namespace: event.Namespace,
		},
	)
	err := worker.executeRuntimeCommandHandoff(
		ctx,
		event,
		preparedEvent{objectKind: "contacts", sourceID: "external-1", flowName: "Companion_Flow"},
		map[string]any{"sourceID": "external-1"},
	)
	if err != nil {
		t.Fatalf("handoff error = %v", err)
	}
	if len(handoff.requests) != 1 || len(events.processed) != 1 ||
		events.processed[0] != event.ID {
		t.Fatalf(
			"handoff requests=%d processed=%v",
			len(handoff.requests),
			events.processed,
		)
	}
	request := handoff.requests[0]
	if request.Scope.WorkItemID != "work-1" ||
		request.Scope.LeaseToken != event.ClaimToken ||
		request.Scope.RouteVersion != 7 ||
		request.DefinitionRef != "Companion_Flow" ||
		request.DefinitionVersion != event.CompanionFlowVersionID ||
		request.DependencySource.ObjectKind != "contacts" ||
		request.DependencySource.SourceID != "external-1" {
		t.Fatalf("handoff request = %#v", request)
	}
}

func TestRuntimeCommandHandoffRetriableFailureKeepsWorkItemNonTerminal(t *testing.T) {
	events := newFakeEvents()
	handoff := &recordingRuntimeCommandHandoff{
		err: &RuntimeCommandHandoffFailure{
			Cause: errors.New("Runtime unavailable"), Retriable: true,
		},
	}
	worker := &Worker{
		cfg: WorkerConfig{
			Namespace: "connector-pack", FinalizeTimeout: time.Second,
		},
		events: events, syncstate: &fakeSyncState{}, runtimeCommands: handoff,
	}
	event := runtimeCommandHandoffEvent()
	ctx := constants.WithConnectorInboundTrace(
		context.Background(),
		constants.ConnectorInboundTrace{
			WorkItemID: "work-1", EventID: event.ID,
			Namespace: event.Namespace,
		},
	)
	err := worker.executeRuntimeCommandHandoff(
		ctx,
		event,
		preparedEvent{objectKind: "contacts", flowName: "Companion_Flow"},
		map[string]any{"sourceID": "external-1"},
	)
	if err == nil || !strings.Contains(err.Error(), "Runtime unavailable") {
		t.Fatalf("handoff error = %v, want surfaced retriable attempt", err)
	}
	if len(events.releasedForRetry) != 1 || len(events.processed) != 0 ||
		len(events.failed) != 0 {
		t.Fatalf(
			"released=%v processed=%v failed=%v",
			events.releasedForRetry,
			events.processed,
			events.failed,
		)
	}
}

type runtimeCommandHandoffEvents struct {
	*fakeEventsService
}

func (e *runtimeCommandHandoffEvents) WorkItemIDForEvent(eventID string) (string, bool) {
	if eventID == "" {
		return "", false
	}
	return "work:" + eventID, true
}

type sourceProjectionRuntimeCommandHandoffEvents struct {
	*runtimeCommandHandoffEvents
	workItemID string
}

func (e *sourceProjectionRuntimeCommandHandoffEvents) WorkItemIDForEvent(eventID string) (string, bool) {
	if e == nil || eventID == "" || e.workItemID == "" {
		return "", false
	}
	return e.workItemID, true
}

func TestRuntimeCommandHandoffResumesReceiptBeforeProviderRead(t *testing.T) {
	event := noopEvent("event-1", "contact.created")
	event.SourceLaneKey = "source-1"
	// A stored receipt is sufficient authority to retry exact bytes. Deliberately
	// make the legacy preparation inputs unusable so this test proves resume does
	// not reopen payload, route, companion, Catalog, or provider state.
	event.Payload = json.RawMessage(`not-json`)
	events := &runtimeCommandHandoffEvents{newFakeEvents(event)}
	batchRead := &countingBatchRead{}
	handoff := &recordingRuntimeCommandHandoff{resumeFound: true}
	worker := &Worker{
		cfg: WorkerConfig{
			Namespace: workerTestNS, FinalizeTimeout: time.Second,
		},
		events: events, syncstate: &fakeSyncState{}, batchRead: batchRead,
		runtimeCommands: handoff,
		resolver:        nil,
	}

	claimed, err := worker.processClaimedWithDrain(
		context.Background(), nil, event, 1,
	)
	if err != nil || claimed != 1 {
		t.Fatalf("process claimed = %d, %v", claimed, err)
	}
	if batchRead.calls != 0 || len(handoff.requests) != 0 ||
		len(handoff.resumeScopes) != 1 || len(events.processed) != 1 {
		t.Fatalf(
			"batch reads=%d accepts=%d resumes=%d processed=%v",
			batchRead.calls,
			len(handoff.requests),
			len(handoff.resumeScopes),
			events.processed,
		)
	}
	if handoff.resumeScopes[0].WorkItemID != "work:event-1" {
		t.Fatalf("resume scope = %#v", handoff.resumeScopes[0])
	}
}

func TestRuntimeCommandHandoffRejectsCollectionModeBeforeOwnership(t *testing.T) {
	event := noopEvent("event-1", "contact.created")
	events := &runtimeCommandHandoffEvents{newFakeEvents(event)}
	batchRead := &countingBatchRead{}
	handoff := &recordingRuntimeCommandHandoff{}
	worker := &Worker{
		cfg: WorkerConfig{
			Namespace: workerTestNS, FinalizeTimeout: time.Second,
		},
		events: events, syncstate: &fakeSyncState{}, batchRead: batchRead,
		runtimeCommands: handoff,
		resolver: &stubRouteResolver{specs: map[string]connector_inbound_routing.RouteExecutionSpec{
			workerTestNS + "/contact.created": {
				FlowDeveloperName:     "Companion_Flow",
				InvocationMode:        connector_inbound_routing.InvocationModeBatchCollection,
				BatchInputName:        "records",
				BatchResultOutputName: "results",
				MaxBatchSize:          10,
			},
		}},
	}

	claimed, err := worker.processClaimedWithDrain(
		context.Background(), nil, event, 1,
	)
	if err == nil || claimed != 1 ||
		!strings.Contains(err.Error(), "not supported by the Runtime command handoff") {
		t.Fatalf("process claimed = %d, %v", claimed, err)
	}
	if batchRead.calls != 0 || len(handoff.resumeScopes) != 1 ||
		len(handoff.requests) != 0 || len(events.failed) != 1 {
		t.Fatalf(
			"batch reads=%d resumes=%d accepts=%d failed=%v",
			batchRead.calls,
			len(handoff.resumeScopes),
			len(handoff.requests),
			events.failed,
		)
	}
}

func TestRuntimeCommandHandoffSettledSourceProjectionTransfersCompleteGeneration(t *testing.T) {
	first := propertyChangeEvent("property-first", "firstname")
	latest := propertyChangeEvent("property-latest", "lastname")
	for sequence, event := range []*connector_events.ConnectorEvent{first, latest} {
		event.DeliveryId = "delivery-1"
		event.DeliverySequence = sequence
		event.SourceLaneKey = "source-lane-contact-123"
	}

	storedEvents := newFakeEvents(latest)
	storedEvents.settledProjectionPayloads = map[string][]*connector_events.ConnectorEvent{
		latest.ID: {first, latest},
	}
	events := &sourceProjectionRuntimeCommandHandoffEvents{
		runtimeCommandHandoffEvents: &runtimeCommandHandoffEvents{storedEvents},
		workItemID:                  connector_workqueue.SourceProjectionWorkIDPrefix + "contact-123-generation-1",
	}
	handoff := &recordingRuntimeCommandHandoff{}
	worker := &Worker{
		cfg: WorkerConfig{
			Namespace:       workerTestNS,
			ClaimPath:       ClaimPathWorkItem,
			FinalizeTimeout: time.Second,
			FlowDeadline:    time.Minute,
		},
		events:    events,
		syncstate: &fakeSyncState{},
		batchRead: &returningBatchRead{result: map[string]json.RawMessage{
			"contact-123": json.RawMessage(`{"id":"contact-123","firstname":"Grace","lastname":"Hopper"}`),
		}},
		runtimeCommands: handoff,
		resolver: &stubRouteResolver{specs: map[string]connector_inbound_routing.RouteExecutionSpec{
			workerTestNS + "/contact.propertychange": {
				FlowDeveloperName: "HubSpot_Contact_Lead_Upsert",
				InvocationMode: connector_inbound_routing.
					InvocationModeLatestBySourceFullRecordPreservePayloadSettled,
			},
		}},
	}

	claimed, err := worker.processClaimedWithDrain(
		context.Background(), nil, latest, 1,
	)
	if err != nil || claimed != 1 {
		t.Fatalf("process claimed = %d, %v", claimed, err)
	}
	if len(handoff.requests) != 1 || len(events.processed) != 1 || events.processed[0] != latest.ID {
		t.Fatalf("accepts=%d processed=%v", len(handoff.requests), events.processed)
	}
	request := handoff.requests[0]
	if request.Scope.WorkItemID != events.workItemID || request.DefinitionRef != "HubSpot_Contact_Lead_Upsert" {
		t.Fatalf("handoff request = %#v", request)
	}
	record, ok := request.Inputs["record"].(map[string]any)
	if !ok || record["firstname"] != "Grace" || record["lastname"] != "Hopper" {
		t.Fatalf("record = %#v", request.Inputs["record"])
	}
	coalesced, ok := request.Inputs["coalescedPayloads"].([]any)
	if !ok || len(coalesced) != 2 {
		t.Fatalf("coalescedPayloads = %#v", request.Inputs["coalescedPayloads"])
	}
	for i, want := range []string{"firstname", "lastname"} {
		item, ok := coalesced[i].(map[string]any)
		if !ok {
			t.Fatalf("coalescedPayloads[%d] = %#v", i, coalesced[i])
		}
		payload, ok := item["payload"].(map[string]any)
		if !ok || payload["propertyName"] != want {
			t.Fatalf("coalescedPayloads[%d].payload = %#v, want propertyName %q", i, item["payload"], want)
		}
	}
}

func TestRuntimeCommandHandoffInvocationModeAuthority(t *testing.T) {
	settled := connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled
	tests := []struct {
		name       string
		claimPath  ClaimPath
		workItemID string
		mode       string
		wantMode   bool
		wantRow    bool
	}{
		{name: "per event", mode: connector_inbound_routing.InvocationModePerEvent, wantMode: true, wantRow: true},
		{
			name:       "settled source projection",
			claimPath:  ClaimPathWorkItem,
			workItemID: connector_workqueue.SourceProjectionWorkIDPrefix + "generation-1",
			mode:       settled,
			wantMode:   true,
			wantRow:    true,
		},
		{name: "settled ordinary WorkItem", claimPath: ClaimPathWorkItem, workItemID: "work-1", mode: settled, wantMode: true},
		{name: "settled legacy event claim", claimPath: ClaimPathLegacyEvent, workItemID: connector_workqueue.SourceProjectionWorkIDPrefix + "generation-1", mode: settled, wantMode: true},
		{name: "batch collection", claimPath: ClaimPathWorkItem, workItemID: "work-1", mode: connector_inbound_routing.InvocationModeBatchCollection},
		{name: "replaceable latest by source", claimPath: ClaimPathWorkItem, workItemID: "work-1", mode: connector_inbound_routing.InvocationModeLatestBySource},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := RuntimeCommandHandoffSupportsInvocationMode(tt.mode); got != tt.wantMode {
				t.Fatalf("mode supported = %v, want %v", got, tt.wantMode)
			}
			if got := runtimeCommandHandoffSupportsClaimedWorkItem(tt.claimPath, tt.workItemID, tt.mode); got != tt.wantRow {
				t.Fatalf("claimed row supported = %v, want %v", got, tt.wantRow)
			}
		})
	}
}

func TestRuntimeCommandHandoffRechecksCompanionBeforeOwnershipTransfer(t *testing.T) {
	event := makeCompanionEvent(
		"event-1",
		"connector-pack",
		"install-1",
		"companion-install-1",
		"flow-version-1",
	)
	event.RouteVersion = sql.NullInt64{Int64: 7, Valid: true}
	event.SourceLaneKey = "source-1"
	work := liveWorkItem(
		"work-event-1",
		event.ID,
		connector_workqueue.WorkTypeConnectorEvent,
		-1,
		"seed-lease",
	)
	work.InstalledPackID = event.InstalledPackID
	work.State = connector_workqueue.WorkStatePending
	resolvedGeneration := companionClaimGeneration
	claims := &fakeWorkItemClaims{
		work: work, resolveGeneration: &resolvedGeneration,
	}
	service := &companionTerminalIdentityService{
		completeWorkItemServiceFake: newCompleteWorkItemServiceFake(
			claims,
			&fakeWorkItemEventLoader{event: event},
			nil,
		),
	}
	handoff := &recordingRuntimeCommandHandoff{}
	cfg := mandatoryWorkerTestConfig()
	cfg.Namespace = event.Namespace
	cfg.PackInstallID = event.InstalledPackID
	cfg.FlowDeadline = time.Minute
	worker, err := NewRuntimeCommandHandoffWorker(
		context.Background(),
		cfg,
		service,
		&fakeSyncState{},
		fakeBatchRead{},
		handoff,
		&fakeRouteResolver{flowName: "companion-flow"},
	)
	if err != nil {
		t.Fatalf("NewRuntimeCommandHandoffWorker: %v", err)
	}
	checker := &scriptedCompanionChecker{results: []companionCheckResult{
		{present: true},  // prepareEvent
		{present: false}, // immediately before AcceptCommand
	}}
	worker.SetCompanionInstalledChecker(checker.check)

	claimed, err := worker.processOne(context.Background())
	if !claimed || err == nil || !strings.Contains(err.Error(), "no longer eligible") {
		t.Fatalf("processOne = %v, %v, want claimed fail-closed", claimed, err)
	}
	checker.assertCalls(t, 2)
	if len(handoff.requests) != 0 {
		t.Fatalf("AcceptCommand requests = %d, want none after companion rotation", len(handoff.requests))
	}
	if len(claims.completeCalls) != 0 || len(claims.failInputs) != 1 {
		t.Fatalf("WorkItem completions=%v failures=%v", claims.completeCalls, claims.failInputs)
	}
}

func runtimeCommandHandoffEvent() *connector_events.ConnectorEvent {
	return &connector_events.ConnectorEvent{
		ID: "event-1", Namespace: "connector-pack",
		InstalledPackID: "install-1", EventKind: "contact.created",
		SourceLaneKey: "source-1", ClaimToken: "lease-1",
		RouteVersion:           sql.NullInt64{Int64: 7, Valid: true},
		CompanionFlowVersionID: "flow-version-1",
	}
}