package connector_events

import (
	"context"
	"database/sql"
	"testing"
	"time"

	connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
)

func TestService_LatestBySourceCoalescesAcrossUnrelatedDeliveryPredecessor_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "coalesce-blocked-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "coalesce-blocked-test")

	baseNow := time.Date(2026, 6, 21, 18, 20, 0, 0, time.UTC)
	nowSeq := []time.Time{
		baseNow,
		baseNow.Add(100 * time.Microsecond),
		baseNow.Add(200 * time.Microsecond),
		baseNow.Add(300 * time.Microsecond),
	}
	nowIdx := 0
	svc := NewService(db, deterministicID(
		"evt-coalesce-current",
		"evt-coalesce-delivery-head",
		"evt-coalesce-blocked-tail",
	), func() time.Time {
		if nowIdx >= len(nowSeq) {
			return nowSeq[len(nowSeq)-1]
		}
		v := nowSeq[nowIdx]
		nowIdx++
		return v
	})
	ctx := context.Background()
	const packInstall = "install-token-coalesce-blocked-test"
	const eventKind = "webhook.event"

	currentPayload := []byte(`{"objectKind":"workflow-actions","sourceID":"shared-source","payload":{"status":"old"}}`)
	blockerPayload := []byte(`{"objectKind":"workflow-actions","sourceID":"other-source","payload":{"status":"head"}}`)
	tailPayload := []byte(`{"objectKind":"workflow-actions","sourceID":"shared-source","payload":{"status":"new"}}`)

	currentID, _, err := svc.Enqueue(ctx, "coalesce-blocked-test", eventKind, "ikey-coalesce-current", currentPayload, packInstall, 1, "", "", 0, "dlv-coalesce-current")
	if err != nil {
		t.Fatalf("Enqueue current: %v", err)
	}
	ids, _, err := svc.EnqueueBatch(ctx, "coalesce-blocked-test", packInstall, "dlv-coalesce-blocked-tail", []BatchEvent{
		{EventKind: eventKind, IdempotencyKey: "ikey-coalesce-head", Payload: blockerPayload, RouteVersion: 1, DeliverySequence: 0},
		{EventKind: eventKind, IdempotencyKey: "ikey-coalesce-tail", Payload: tailPayload, RouteVersion: 1, DeliverySequence: 1},
	})
	if err != nil {
		t.Fatalf("Enqueue blocked delivery: %v", err)
	}
	blockerID, tailID := ids[0], ids[1]

	current, ok, err := svc.ClaimNext(ctx, packInstall, "worker-coalesce-current")
	if err != nil {
		t.Fatalf("ClaimNext current: %v", err)
	}
	if !ok || current == nil || current.ID != currentID {
		t.Fatalf("ClaimNext current = (%v, %v), want %s", current, ok, currentID)
	}

	next, found, err := svc.ClaimNextCoalescedSibling(ctx, current.ID, current.ClaimToken, "worker-coalesce-next")
	if err != nil {
		t.Fatalf("ClaimNextCoalescedSibling: %v", err)
	}
	if found || next != nil {
		t.Fatalf("legacy coalesced sibling = (%v, %v), want none while delivery head blocks tail", next, found)
	}

	next, found, err = svc.ClaimNextCoalescedSiblingSourceScoped(ctx, current.ID, current.ClaimToken, "worker-coalesce-next")
	if err != nil {
		t.Fatalf("ClaimNextCoalescedSiblingSourceScoped: %v", err)
	}
	if !found || next == nil || next.ID != tailID {
		t.Fatalf("source-scoped coalesced sibling = (%v, %v), want %s across unrelated delivery predecessor", next, found, tailID)
	}

	assertEventStatus(t, db, currentID, statusProcessed)
	assertEventStatus(t, db, blockerID, statusPending)
	assertEventStatus(t, db, tailID, statusProcessing)
}

func TestService_LatestBySourceCoalesceDefersFuturePendingSibling_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "coalesce-not-before-test")
	createConnectorEventTable(t, db)

	ctx := context.Background()
	const packInstall = "install-token-coalesce-not-before-test"
	const eventKind = "webhook.event"
	base := time.Date(2026, 7, 11, 0, 0, 0, 0, time.UTC)
	future := base.Add(5 * time.Second)
	payload := []byte(`{"objectKind":"workflow-actions","sourceID":"settled-source","payload":{"status":"new"}}`)

	for _, row := range []struct {
		id, status, token string
		ordinal           int
		nextAttempt       interface{}
	}{
		{id: "evt-settled-current", status: statusProcessing, token: "current-token", ordinal: 1},
		{id: "evt-settled-future", status: statusPending, ordinal: 2, nextAttempt: future},
	} {
		if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
			(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, Payload, RetryCount,
			 NextAttemptAt, ClaimToken, RouteVersion, DeliveryId, DeliverySequence, EnqueueOrdinal, CreatedDate, LastModifiedDate)
			VALUES (?, ?, ?, ?, ?, ?, ?, 0, ?, ?, 1, ?, 0, ?, ?, ?)`,
			row.id, "coalesce-not-before-test", packInstall, eventKind, row.id, row.status, payload,
			row.nextAttempt, row.token, row.id+"-delivery", row.ordinal, base, base); err != nil {
			t.Fatalf("insert %s: %v", row.id, err)
		}
	}

	svc := NewService(db, deterministicID("unused-token"), func() time.Time { return base })
	next, found, err := svc.ClaimNextCoalescedSibling(ctx, "evt-settled-current", "current-token", "worker-settled")
	if err != nil {
		t.Fatalf("ClaimNextCoalescedSibling: %v", err)
	}
	if found || next != nil {
		t.Fatalf("coalesced sibling = (%v, %v), want none before NextAttemptAt", next, found)
	}

	assertEventStatus(t, db, "evt-settled-current", statusProcessing)
	assertEventStatus(t, db, "evt-settled-future", statusPending)
}

func TestService_SettledRouteReusesOpenSourceDeadlineAcrossDeliveries_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const namespace = "coalesce-settle-window-test"
	const installID = "install-token-" + namespace
	const eventKind = "webhook.event"
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	if _, err := db.Exec(`UPDATE _System_ConnectorInboundRoute
		SET InvocationMode = ?
		WHERE InstalledPackId = ? AND EventKind = ? AND Version = 1`,
		connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled,
		installID, eventKind); err != nil {
		t.Fatalf("configure settled route: %v", err)
	}

	base := time.Date(2026, 7, 11, 0, 0, 0, 0, time.UTC)
	now := base
	svc := NewService(db, deterministicID("evt-settled-first", "projection-generation", "evt-settled-second"), func() time.Time { return now })
	payload := []byte(`{"objectKind":"contact","sourceID":"contact-settled","payload":{"property":"email"}}`)

	firstID, firstDup, err := svc.Enqueue(context.Background(), namespace, eventKind, "settled-first", payload, installID, 1, "", "", 0, "delivery-first")
	if err != nil || firstDup {
		t.Fatalf("first settled enqueue = (%q, %v, %v)", firstID, firstDup, err)
	}
	now = base.Add(4 * time.Second)
	secondID, secondDup, err := svc.Enqueue(context.Background(), namespace, eventKind, "settled-second", payload, installID, 1, "", "", 0, "delivery-second")
	if err != nil || secondDup {
		t.Fatalf("second settled enqueue = (%q, %v, %v)", secondID, secondDup, err)
	}

	wantDeadline := base.Add(connector_inbound_routing.DefaultSourceSettleDelay)
	for _, id := range []string{firstID, secondID} {
		var got time.Time
		if err := db.QueryRow(`SELECT NextAttemptAt FROM _System_ConnectorEvent WHERE Id = ?`, id).Scan(&got); err != nil {
			t.Fatalf("read NextAttemptAt for %s: %v", id, err)
		}
		if !got.Equal(wantDeadline) {
			t.Fatalf("NextAttemptAt for %s = %s, want shared %s", id, got, wantDeadline)
		}
	}
}

func TestService_SourceLaneOverrideDoesNotCoalesceDifferentSourceObjects_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "coalesce-override-lane-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "coalesce-override-lane-test")
	if _, err := db.Exec(`
UPDATE _System_ConnectorInboundRoute
   SET EventKind = 'workflow.action.execute'
 WHERE Namespace = 'coalesce-override-lane-test'`); err != nil {
		t.Fatalf("configure workflow action route: %v", err)
	}

	baseNow := time.Date(2026, 7, 1, 10, 0, 0, 0, time.UTC)
	nowSeq := []time.Time{
		baseNow,
		baseNow.Add(100 * time.Microsecond),
		baseNow.Add(200 * time.Microsecond),
	}
	nowIdx := 0
	svc := NewService(db, deterministicID(
		"evt-override-current",
		"evt-override-next",
		"claim-token-override-current",
	), func() time.Time {
		if nowIdx >= len(nowSeq) {
			return nowSeq[len(nowSeq)-1]
		}
		v := nowSeq[nowIdx]
		nowIdx++
		return v
	})
	ctx := context.Background()
	const packInstall = "install-token-coalesce-override-lane-test"
	const eventKind = "workflow.action.execute"
	const parentID = "parent-001"

	currentPayload := []byte(`{"objectKind":"workflow-actions","sourceID":"contact-a","payload":{"parent_id":"` + parentID + `"}}`)
	nextPayload := []byte(`{"objectKind":"workflow-actions","sourceID":"contact-b","payload":{"parent_id":"` + parentID + `"}}`)

	ids, _, err := svc.EnqueueBatch(ctx, "coalesce-override-lane-test", packInstall, "dlv-override-lane", []BatchEvent{
		{
			EventKind:            eventKind,
			IdempotencyKey:       "ikey-override-current",
			Payload:              currentPayload,
			RouteVersion:         1,
			SourceLaneObjectKind: "shared-parent",
			SourceLaneID:         parentID,
			DeliverySequence:     0,
		},
		{
			EventKind:            eventKind,
			IdempotencyKey:       "ikey-override-next",
			Payload:              nextPayload,
			RouteVersion:         1,
			SourceLaneObjectKind: "shared-parent",
			SourceLaneID:         parentID,
			DeliverySequence:     1,
		},
	})
	if err != nil {
		t.Fatalf("EnqueueBatch: %v", err)
	}
	currentID, nextID := ids[0], ids[1]

	current, ok, err := svc.ClaimNext(ctx, packInstall, "worker-override-current")
	if err != nil {
		t.Fatalf("ClaimNext current: %v", err)
	}
	if !ok || current == nil || current.ID != currentID {
		t.Fatalf("ClaimNext current = (%v, %v), want %s", current, ok, currentID)
	}

	next, found, err := svc.ClaimNextCoalescedSibling(ctx, current.ID, current.ClaimToken, "worker-override-next")
	if err != nil {
		t.Fatalf("ClaimNextCoalescedSibling: %v", err)
	}
	if found || next != nil {
		t.Fatalf("coalesced sibling = (%v, %v), want none for different source objects sharing an override lane", next, found)
	}

	assertEventStatus(t, db, currentID, statusProcessing)
	assertEventStatus(t, db, nextID, statusPending)
}

func TestService_LatestBySourceCoalesceZeroOrdinalUsesHistoricalOrder_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "coalesce-zero-ordinal-test")
	createConnectorEventTable(t, db)

	ctx := context.Background()
	const packInstall = "install-token-coalesce-zero-ordinal-test"
	const eventKind = "webhook.event"
	payload := []byte(`{"objectKind":"workflow-actions","sourceID":"legacy-coalesce-source","payload":{"status":"Created"}}`)
	currentCreated := time.Date(2026, 6, 21, 19, 40, 0, 0, time.UTC)
	nextCreated := currentCreated.Add(time.Minute)

	for _, row := range []struct {
		id          string
		idempotency string
		created     time.Time
	}{
		// The current row's Id sorts after the later row. EnqueueOrdinal=0 rows
		// must still use historical queue order, not Id order, while coalescing.
		{id: "evt-zero-z-current", idempotency: "zero-current", created: currentCreated},
		{id: "evt-zero-a-next", idempotency: "zero-next", created: nextCreated},
	} {
		if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
			(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, Payload, RetryCount,
			 RouteVersion, DeliverySequence, EnqueueOrdinal, CreatedDate, LastModifiedDate)
			VALUES (?, ?, ?, ?, ?, 'pending', ?, 0, 1, 0, 0, ?, ?)`,
			row.id, "coalesce-zero-ordinal-test", packInstall, eventKind, row.idempotency, payload, row.created, row.created); err != nil {
			t.Fatalf("insert zero ordinal row %s: %v", row.id, err)
		}
	}

	svc := NewService(db, deterministicID("claim-token-zero-current", "claim-token-zero-next"), func() time.Time {
		return nextCreated.Add(time.Hour)
	})
	current, ok, err := svc.ClaimNext(ctx, packInstall, "worker-zero-current")
	if err != nil {
		t.Fatalf("ClaimNext current: %v", err)
	}
	if !ok || current == nil || current.ID != "evt-zero-z-current" {
		t.Fatalf("ClaimNext current = (%v, %v), want evt-zero-z-current", current, ok)
	}

	next, found, err := svc.ClaimNextCoalescedSibling(ctx, current.ID, current.ClaimToken, "worker-zero-next")
	if err != nil {
		t.Fatalf("ClaimNextCoalescedSibling: %v", err)
	}
	if !found || next == nil || next.ID != "evt-zero-a-next" {
		t.Fatalf("coalesced sibling = (%v, %v), want evt-zero-a-next", next, found)
	}

	assertEventStatus(t, db, "evt-zero-z-current", statusProcessed)
	assertEventStatus(t, db, "evt-zero-a-next", statusProcessing)
}

func TestService_LatestBySourceCoalesceSkipsBlockedZeroOrdinalSibling_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "coalesce-mixed-ordinal-test")
	createConnectorEventTable(t, db)

	ctx := context.Background()
	const packInstall = "install-token-coalesce-mixed-ordinal-test"
	const eventKind = "webhook.event"
	sharedPayload := []byte(`{"objectKind":"workflow-actions","sourceID":"mixed-coalesce-source","payload":{"status":"value"}}`)
	blockerPayload := []byte(`{"objectKind":"workflow-actions","sourceID":"other-source","payload":{"status":"blocker"}}`)
	base := time.Date(2026, 6, 21, 20, 45, 0, 0, time.UTC)

	for _, row := range []struct {
		id          string
		idempotency string
		payload     []byte
		deliveryID  string
		sequence    int
		ordinal     int
		created     time.Time
	}{
		{id: "evt-mixed-current", idempotency: "mixed-current", payload: sharedPayload, deliveryID: "dlv-mixed-current", sequence: 0, ordinal: 1, created: base},
		{id: "evt-mixed-positive-next", idempotency: "mixed-positive-next", payload: sharedPayload, deliveryID: "dlv-mixed-positive", sequence: 0, ordinal: 2, created: base.Add(time.Second)},
		{id: "evt-mixed-blocker-head", idempotency: "mixed-blocker-head", payload: blockerPayload, deliveryID: "dlv-mixed-blocked-zero", sequence: 0, ordinal: 3, created: base.Add(2 * time.Second)},
		{id: "evt-mixed-zero-blocked-tail", idempotency: "mixed-zero-blocked-tail", payload: sharedPayload, deliveryID: "dlv-mixed-blocked-zero", sequence: 1, ordinal: 0, created: base.Add(3 * time.Second)},
	} {
		if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
			(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, Payload, RetryCount,
			 RouteVersion, DeliveryId, DeliverySequence, EnqueueOrdinal, CreatedDate, LastModifiedDate)
			VALUES (?, ?, ?, ?, ?, 'pending', ?, 0, 1, ?, ?, ?, ?, ?)`,
			row.id, "coalesce-mixed-ordinal-test", packInstall, eventKind, row.idempotency, row.payload,
			row.deliveryID, row.sequence, row.ordinal, row.created, row.created); err != nil {
			t.Fatalf("insert mixed ordinal row %s: %v", row.id, err)
		}
	}

	svc := NewService(db, deterministicID("claim-token-mixed-current", "claim-token-mixed-next"), func() time.Time {
		return base.Add(time.Hour)
	})
	current, ok, err := svc.ClaimNext(ctx, packInstall, "worker-mixed-current")
	if err != nil {
		t.Fatalf("ClaimNext current: %v", err)
	}
	if !ok || current == nil || current.ID != "evt-mixed-current" {
		t.Fatalf("ClaimNext current = (%v, %v), want evt-mixed-current", current, ok)
	}

	next, found, err := svc.ClaimNextCoalescedSibling(ctx, current.ID, current.ClaimToken, "worker-mixed-next")
	if err != nil {
		t.Fatalf("ClaimNextCoalescedSibling: %v", err)
	}
	if !found || next == nil || next.ID != "evt-mixed-positive-next" {
		t.Fatalf("coalesced sibling = (%v, %v), want evt-mixed-positive-next", next, found)
	}

	assertEventStatus(t, db, "evt-mixed-current", statusProcessed)
	assertEventStatus(t, db, "evt-mixed-positive-next", statusProcessing)
	assertEventStatus(t, db, "evt-mixed-blocker-head", statusPending)
	assertEventStatus(t, db, "evt-mixed-zero-blocked-tail", statusPending)
}

func assertEventStatus(t *testing.T, db *sql.DB, eventID, want string) {
	t.Helper()
	var got string
	if err := db.QueryRow(`SELECT Status FROM _System_ConnectorEvent WHERE Id = ?`, eventID).Scan(&got); err != nil {
		t.Fatalf("query event %s status: %v", eventID, err)
	}
	if got != want {
		t.Fatalf("event %s status = %s, want %s", eventID, got, want)
	}
}