package runtimeworker

import (
	"context"
	"strings"
	"testing"

	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/backend/internal/runtimeconfig"
	"github.com/nexuscrm/backend/internal/runtimepermissionstate"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/records"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
)

const testWorkerSourceRevision = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"

func TestAssemblyRequiresMetadataScopeForMetadataWorkBeforeDatabaseAccess(t *testing.T) {
	role := validWorkerRole("runtime.record-event-dispatch.v1")
	_, _, err := AssembleProduction(context.Background(), testWorkerSourceRevision, runtimeconfig.Worker{Role: role})
	if err == nil || !strings.Contains(err.Error(), "requires an immutable-view scope") {
		t.Fatalf("error=%v", err)
	}
}

func TestAssemblyRejectsMetadataSchemaForPayloadWorkBeforeDatabaseAccess(t *testing.T) {
	role := validWorkerRole("runtime.audit-history-fanout.v1")
	role.Schema = runtimeconfig.WorkerConfigSchemaV2
	setValidMetadataScope(&role)
	_, _, err := AssembleProduction(context.Background(), testWorkerSourceRevision, runtimeconfig.Worker{Role: role})
	if err == nil || !strings.Contains(err.Error(), "payload work family requires payload schema v1 or v3") {
		t.Fatalf("error=%v", err)
	}
}

func TestAssemblyRejectsMetadataScopeForV3PayloadWorkBeforeDatabaseAccess(t *testing.T) {
	role := validWorkerRole("runtime.audit-history-fanout.v1")
	role.Schema = runtimeconfig.WorkerConfigSchemaV3
	role.PermissionStateProtocol = runtimeconfig.PermissionStateProtocolV1
	setValidMetadataScope(&role)
	_, _, err := AssembleProduction(context.Background(), testWorkerSourceRevision, runtimeconfig.Worker{Role: role})
	if err == nil || !strings.Contains(err.Error(), "payload work family requires payload schema v1 or v3") {
		t.Fatalf("error=%v", err)
	}
}

func TestAssemblyRejectsUnknownWorkFamilyBeforeDatabaseAccess(t *testing.T) {
	role := validWorkerRole("runtime.unknown.v1")
	_, _, err := AssembleProduction(context.Background(), testWorkerSourceRevision, runtimeconfig.Worker{Role: role})
	if err == nil || !strings.Contains(err.Error(), "unsupported work family") {
		t.Fatalf("error=%v", err)
	}
}

func TestAssemblyRequiresCatalogSelectedFlowAuthorityBeforeDatabaseAccess(t *testing.T) {
	role := validWorkerRole(automation.RuntimeScheduledFlowWorkFamily)
	role.Schema = runtimeconfig.WorkerConfigSchemaV2
	setValidMetadataScope(&role)
	role.RuntimeCapabilityID = flowsemantic.CapabilityName
	_, _, err := AssembleProduction(
		context.Background(),
		testWorkerSourceRevision,
		runtimeconfig.Worker{Role: role},
	)
	if err == nil || !strings.Contains(err.Error(), "requires App capability selection") {
		t.Fatalf("missing App capability error = %v", err)
	}

	role.AppCapabilityID = "io.nexuscrm.test.flow-handler.v1"
	_, _, err = AssembleProduction(
		context.Background(),
		testWorkerSourceRevision,
		runtimeconfig.Worker{Role: role},
	)
	if err == nil || !strings.Contains(err.Error(), "exact compatibility receipt selection") {
		t.Fatalf("missing exact receipt error = %v", err)
	}

	role.ExpectedCompatibilityReceiptDigest = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	role.RuntimeCapabilityID = "runtime.records.mutate.v1"
	_, _, err = AssembleProduction(
		context.Background(),
		testWorkerSourceRevision,
		runtimeconfig.Worker{Role: role},
	)
	if err == nil || !strings.Contains(err.Error(), "neutral Flow metadata scope") {
		t.Fatalf("wrong Runtime capability error = %v", err)
	}
}

func TestAssemblyRequiresCatalogSelectedBulkMutationAuthorityBeforeDatabaseAccess(t *testing.T) {
	role := validWorkerRole(records.RuntimeBulkIngestWorkFamily)
	role.Schema = runtimeconfig.WorkerConfigSchemaV2
	setValidMetadataScope(&role)
	_, _, err := AssembleProduction(
		context.Background(),
		testWorkerSourceRevision,
		runtimeconfig.Worker{Role: role},
	)
	if err == nil || !strings.Contains(err.Error(), "requires App capability selection") {
		t.Fatalf("missing App capability error = %v", err)
	}

	role.AppCapabilityID = "io.nexuscrm.test.record-mutation.v1"
	_, _, err = AssembleProduction(
		context.Background(),
		testWorkerSourceRevision,
		runtimeconfig.Worker{Role: role},
	)
	if err == nil || !strings.Contains(err.Error(), "exact compatibility receipt selection") {
		t.Fatalf("missing exact receipt error = %v", err)
	}

	role.ExpectedCompatibilityReceiptDigest = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	role.RuntimeCapabilityID = flowsemantic.CapabilityName
	_, _, err = AssembleProduction(
		context.Background(),
		testWorkerSourceRevision,
		runtimeconfig.Worker{Role: role},
	)
	if err == nil || !strings.Contains(err.Error(), "neutral record-mutation metadata scope") {
		t.Fatalf("wrong Runtime capability error = %v", err)
	}
}

func TestAssemblyRequiresInjectedPlatformCommandExecutorBeforeDatabaseAccess(t *testing.T) {
	role := validWorkerRole(persistence.RuntimePlatformCommandExecutionWorkFamily)
	role.Schema = runtimeconfig.WorkerConfigSchemaV2
	role.Parallelism = 2
	role.DBMaxOpenConns = 5
	setValidMetadataScope(&role)
	role.RuntimeCapabilityID = commandsemantic.CapabilityName
	_, _, err := assembleProductionWithDependencies(
		context.Background(),
		testWorkerSourceRevision,
		runtimeconfig.Worker{Role: role},
		productionAssemblyDependencies{
			permissionStateFactory: runtimepermissionstate.NewGate,
		},
	)
	if err == nil || !strings.Contains(err.Error(), "release-owned neutral executor registry") {
		t.Fatalf("missing Platform-command executor error = %v", err)
	}
}

func TestAssemblyRequiresProcessedRetentionOnlyForRecordEventWork(t *testing.T) {
	eventRole := validWorkerRole("runtime.record-event-dispatch.v1")
	eventRole.Schema = runtimeconfig.WorkerConfigSchemaV2
	setValidMetadataScope(&eventRole)
	if _, _, err := AssembleProduction(context.Background(), testWorkerSourceRevision, runtimeconfig.Worker{Role: eventRole}); err == nil ||
		!strings.Contains(err.Error(), "requires explicit processed retention and cleanup interval") {
		t.Fatalf("missing record-event retention error=%v", err)
	}

	rollupRole := validWorkerRole("runtime.rollup-delta.v1")
	rollupRole.Schema = runtimeconfig.WorkerConfigSchemaV2
	rollupRole.ProcessedRetention = "336h"
	rollupRole.ProcessedCleanupInterval = "1h"
	setValidMetadataScope(&rollupRole)
	if _, _, err := AssembleProduction(context.Background(), testWorkerSourceRevision, runtimeconfig.Worker{Role: rollupRole}); err == nil ||
		!strings.Contains(err.Error(), "does not own processed-row retention") {
		t.Fatalf("rollup retention error=%v", err)
	}
}

func TestTimedFlowOwnerRoleRequiresOneExactCandidateScope(t *testing.T) {
	recordEvent, scheduled := validTimedFlowHandoffRoles()
	scope, err := validateTimedFlowOwnerRole(recordEvent, scheduled)
	if err != nil {
		t.Fatal(err)
	}
	if scope.OrgID != recordEvent.OrgID ||
		scope.AppInstallationID != recordEvent.AppInstallationID ||
		scope.RuntimeCapabilityID != flowsemantic.CapabilityName {
		t.Fatalf("scheduled scope = %+v", scope)
	}

	for _, test := range []struct {
		name   string
		mutate func(*runtimeconfig.WorkerRole, *runtimeconfig.WorkerRole)
	}{
		{
			name: "different Worker artifact",
			mutate: func(_ *runtimeconfig.WorkerRole, scheduled *runtimeconfig.WorkerRole) {
				scheduled.WorkerReleaseDigest = "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"
			},
		},
		{
			name: "different AppInstallation",
			mutate: func(_ *runtimeconfig.WorkerRole, scheduled *runtimeconfig.WorkerRole) {
				scheduled.AppInstallationID = "installation-other"
			},
		},
		{
			name: "event role lacks mutation authority",
			mutate: func(recordEvent *runtimeconfig.WorkerRole, _ *runtimeconfig.WorkerRole) {
				recordEvent.RuntimeCapabilityID = flowsemantic.CapabilityName
			},
		},
		{
			name: "scheduled role lacks Flow authority",
			mutate: func(_ *runtimeconfig.WorkerRole, scheduled *runtimeconfig.WorkerRole) {
				scheduled.RuntimeCapabilityID = "runtime.records.mutate.v1"
			},
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			recordEvent, scheduled := validTimedFlowHandoffRoles()
			test.mutate(&recordEvent, &scheduled)
			if _, err := validateTimedFlowOwnerRole(recordEvent, scheduled); err == nil {
				t.Fatal("mismatched timed Flow roles were accepted")
			}
		})
	}
}

func TestAssemblyRejectsTimedFlowOwnerForUnrelatedWorkBeforeDatabaseAccess(t *testing.T) {
	role := validWorkerRole("runtime.audit-history-fanout.v1")
	_, scheduled := validTimedFlowHandoffRoles()
	_, _, err := AssembleProduction(
		context.Background(),
		testWorkerSourceRevision,
		runtimeconfig.Worker{Role: role, TimedFlowOwner: &scheduled},
	)
	if err == nil || !strings.Contains(err.Error(), "belongs only to record-event work") {
		t.Fatalf("error=%v", err)
	}
}

func TestAssemblyRequiresStartupContextAndMatchingVersion(t *testing.T) {
	if _, _, err := AssembleProduction(nil, testWorkerSourceRevision, runtimeconfig.Worker{}); err == nil {
		t.Fatal("nil startup context was accepted")
	}
	role := validWorkerRole("runtime.audit-history-fanout.v1")
	if _, _, err := AssembleProduction(context.Background(), "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", runtimeconfig.Worker{Role: role}); err == nil || !strings.Contains(err.Error(), "binary source revision") {
		t.Fatalf("mismatched binary source revision error = %v", err)
	}
}

func TestRepairRejectsMismatchedBinarySourceRevisionBeforeDatabaseAccess(t *testing.T) {
	role := validWorkerRole("runtime.audit-history-fanout.v1")
	err := RepairProduction(context.Background(), "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", runtimeconfig.Worker{Role: role}, RepairRequest{})
	if err == nil || !strings.Contains(err.Error(), "binary source revision") {
		t.Fatalf("mismatched repair binary source revision error = %v", err)
	}
}

func validWorkerRole(workFamily string) runtimeconfig.WorkerRole {
	return runtimeconfig.WorkerRole{
		Schema: runtimeconfig.WorkerConfigSchemaV1, WorkFamily: workFamily,
		DeploymentID: "runtime-worker-test", WorkerReleaseDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		WorkerSourceRevision: testWorkerSourceRevision, OwnershipEpoch: 1, BatchSize: 10, PollInterval: "100ms",
		LeaseDuration: "30s", RetryJitterMax: "10ms", MaxAttempts: 5, DBMaxOpenConns: 4, DBMaxIdleConns: 2,
		ShutdownTimeout: "10s",
	}
}

func setValidMetadataScope(role *runtimeconfig.WorkerRole) {
	role.OrgID = "org-test"
	role.AppInstallationID = "installation-test"
	role.RuntimeCapabilityID = "runtime.records.mutate.v1"
	role.CatalogSnapshotStoreRoot = "/tmp/catalog"
	role.SchemaGeneration = 1
	role.EngineGeneration = 1
}

func validTimedFlowHandoffRoles() (runtimeconfig.WorkerRole, runtimeconfig.WorkerRole) {
	recordEvent := validWorkerRole("runtime.record-event-dispatch.v1")
	recordEvent.Schema = runtimeconfig.WorkerConfigSchemaV2
	recordEvent.ProcessedRetention = "336h"
	recordEvent.ProcessedCleanupInterval = "1h"
	setValidMetadataScope(&recordEvent)

	scheduled := validWorkerRole(automation.RuntimeScheduledFlowWorkFamily)
	scheduled.Schema = runtimeconfig.WorkerConfigSchemaV2
	scheduled.DeploymentID = "runtime-worker-scheduled"
	scheduled.OwnershipEpoch = 2
	setValidMetadataScope(&scheduled)
	scheduled.AppCapabilityID = "io.nexuscrm.sfcrm.automation.execute-flow.v1"
	scheduled.RuntimeCapabilityID = flowsemantic.CapabilityName
	scheduled.ExpectedCompatibilityReceiptDigest = "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"
	return recordEvent, scheduled
}