package connector_inbound

import (
	"context"
	"encoding/json"
	"testing"
	"time"

	events "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	"github.com/pingcap-inc/nexuscrm-connectors/testing/connectorfixture"
)

// The receipt, range, claim, adapter, payload assembly and completion are real.
// Only the external record read and Flow boundary are captured.
func TestProjectionRangeEnqueueThroughWorkerFlowInputsTiDB(t *testing.T) {
	db := connectorfixture.OpenTiDB(t)
	connectorfixture.CreateSystemTables(t, db, "_System_Migration", "_System_InstalledPack", "_System_ConnectorInboundRoute", "_System_ConnectorIngressControl", "_System_ConnectorEvent", "_System_ConnectorWorkItem", "_System_ConnectorWorkAttempt", "_System_ConnectorProjectionProtocol")
	if _, err := db.Exec(`INSERT INTO _System_Migration (Id,Version,Description) VALUES (?,?,?)`, events.WorkItemOnlyExecutionCutoverMigrationID, events.WorkItemOnlyExecutionCutoverMigrationVersion, events.WorkItemOnlyExecutionCutoverMigrationDescription); err != nil {
		t.Fatal(err)
	}
	const ns, pack, kind = "range-worker", "range-worker-install", "contact.propertychange"
	for _, q := range []string{
		`INSERT INTO _System_InstalledPack (Id,Name,PackInstallId,Category,Status) VALUES ('pack-row','range-worker','range-worker-install','connector-pack','active')`,
		`INSERT INTO _System_ConnectorProjectionProtocol (InstalledPackId,ProtocolVersion,LegacyMaxOrdinal) VALUES ('range-worker-install',1,0)`,
		`INSERT INTO _System_ConnectorInboundRoute (Id,Namespace,InstalledPackId,EventKind,Version,FlowDeveloperName,InvocationMode,QueueClass,IsCurrent,IsDeleted) VALUES ('route','range-worker','range-worker-install','contact.propertychange',1,'DeltaFlow','latest_by_source_full_record_preserve_payload_settled','default',1,0)`,
	} {
		if _, err := db.Exec(q); err != nil {
			t.Fatal(err)
		}
	}
	now := time.Now().UTC().Truncate(time.Second).Add(-time.Hour)
	svc := events.NewService(db, nil, func() time.Time { return now })
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()
	var ids []string
	for i, property := range []string{"firstname", "email"} {
		payload, _ := json.Marshal(map[string]any{"objectKind": "contact", "sourceID": "contact-123", "payload": map[string]any{"propertyName": property}})
		got, _, err := svc.EnqueueBatch(ctx, ns, pack, property, []events.BatchEvent{{EventKind: kind, IdempotencyKey: property, RouteVersion: 1, DeliverySequence: 9 - i, Payload: payload}})
		if err != nil {
			t.Fatal(err)
		}
		ids = append(ids, got[0])
		now = now.Add(-time.Minute) // clock and delivery order disagree with admission.
	}
	flow := &capturingFlowInvoker{}
	worker := newMandatoryWorkerWithDependenciesForTest(t, WorkerConfig{Namespace: ns, PackInstallID: pack, WorkerID: "range-worker", WorkItemSchedulerOwnerID: "range-owner", WorkItemSchedulerFencingToken: "range-fence"}, svc, &fakeSyncState{}, &returningBatchRead{result: map[string]json.RawMessage{"contact-123": json.RawMessage(`{"id":"contact-123"}`)}}, flow, &stubRouteResolver{specs: map[string]routing.RouteExecutionSpec{ns + "/" + kind: {FlowDeveloperName: "DeltaFlow", InvocationMode: routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled}}})
	claimed, err := worker.processOne(ctx)
	if err != nil || !claimed || len(flow.capturedArgs) != 1 {
		t.Fatalf("worker=%v/%v Flow calls=%d", claimed, err, len(flow.capturedArgs))
	}
	payloads, ok := flow.capturedArgs[0]["coalescedPayloads"].([]any)
	if !ok || len(payloads) != 2 {
		t.Fatalf("Flow inputs=%#v", flow.capturedArgs)
	}
	for i, id := range ids {
		if payloads[i].(map[string]any)["eventID"] != id {
			t.Fatalf("Flow input order=%#v want=%v", payloads, ids)
		}
	}
	var state string
	var attempt int
	if err := db.QueryRow(`SELECT State,Attempt FROM _System_ConnectorWorkItem WHERE EventId=?`, ids[1]).Scan(&state, &attempt); err != nil {
		t.Fatal(err)
	}
	if state != "processed" || attempt != 1 {
		t.Fatalf("durable completion=%s attempt=%d", state, attempt)
	}
	if claimed, err = worker.processOne(ctx); err != nil || claimed {
		t.Fatalf("unexpected second execution=%v/%v", claimed, err)
	}
}