package connector_events

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

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

func TestService_SourceProjectionDuplicatePreservesCoverage(t *testing.T) {
	for _, scenario := range []struct {
		name               string
		oldClock, newClock int64
	}{
		{"older_provider_clock", 1788343844748, 1788343864304},
		{"equal_provider_clock", 1788343864304, 1788343864304},
		{"missing_provider_clock", -1, -1},
	} {
		t.Run(scenario.name, func(t *testing.T) {
			testSourceProjectionDuplicatePreservesCoverage(t, scenario.oldClock, scenario.newClock)
		})
	}
}

func testSourceProjectionDuplicatePreservesCoverage(t *testing.T, oldClock, newClock int64) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	const namespace = "duplicate-head-diagnosis"
	const pack = "install-token-duplicate-head-diagnosis"
	const kind = "contact.propertychange"
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, kind)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	now := time.Date(2026, 9, 2, 10, 11, 16, 0, time.UTC)
	svc := NewService(db, nil, func() time.Time { return now })
	event := func(key string, occurred int64) BatchEvent {
		if occurred < 0 {
			return BatchEvent{EventKind: kind, IdempotencyKey: key, RouteVersion: 1,
				Payload: json.RawMessage(fmt.Sprintf(`{"objectKind":"contact","sourceID":"synthetic-contact","payload":{"propertyName":%q}}`, key))}
		}
		return BatchEvent{EventKind: kind, IdempotencyKey: key, RouteVersion: 1,
			Payload: json.RawMessage(fmt.Sprintf(`{"objectKind":"contact","sourceID":"synthetic-contact","payload":{"occurredAt":%d,"propertyName":%q}}`, occurred, key))}
	}
	older := event("earlier-notification", oldClock)
	oldIDs, _, err := svc.EnqueueBatch(ctx, namespace, pack, "earlier-delivery", []BatchEvent{older})
	if err != nil {
		t.Fatal(err)
	}
	now = now.Add(46 * time.Second)
	newer := []BatchEvent{event("new-field-1", newClock), event("new-field-2", newClock), event("new-field-3", newClock)}
	for i := range newer {
		newer[i].DeliverySequence = i
	}
	newIDs, _, err := svc.EnqueueBatch(ctx, namespace, pack, "newer-delivery", newer)
	if err != nil {
		t.Fatal(err)
	}
	var before string
	if err = db.QueryRowContext(ctx, `SELECT EventId FROM _System_ConnectorWorkItem WHERE ProjectionActiveKey IS NOT NULL`).Scan(&before); err != nil {
		t.Fatal(err)
	}
	if before != newIDs[2] {
		t.Fatalf("new batch did not own head: got %s want %s", before, newIDs[2])
	}
	now = now.Add(118 * time.Second)
	dupIDs, dups, err := svc.EnqueueBatch(ctx, namespace, pack, "earlier-delivery", []BatchEvent{older})
	if err != nil {
		t.Fatal(err)
	}
	if len(dups) != 1 || !dups[0] || dupIDs[0] != oldIDs[0] {
		t.Fatalf("did not exercise duplicate admission: %v %v", dupIDs, dups)
	}
	var workID, after string
	if err = db.QueryRowContext(ctx, `SELECT WorkId,EventId FROM _System_ConnectorWorkItem WHERE ProjectionActiveKey IS NOT NULL`).Scan(&workID, &after); err != nil {
		t.Fatal(err)
	}
	gap, err := svc.WorkItemRepository().HasNonTerminalLegacyEventWithoutWorkItem(ctx, workqueue.LegacyEventMirrorReadinessInput{InstalledPackID: pack})
	if err != nil {
		t.Fatal(err)
	}
	t.Logf("old=%s newer_head=%s selected_after_duplicate=%s readiness_gap=%v", oldIDs[0], newIDs[2], after, gap)
	// Use the real claim path: no hand-written lease or event-state repair.
	claimedWork, claimed, err := svc.WorkItemRepository().ClaimNextWorkItem(workqueue.WithDistributedClaim(ctx), workqueue.ClaimWorkItemInput{
		InstalledPackID: pack, Generation: 0, OwnerID: "diagnosis-worker", FencingToken: "diagnosis-fence",
		LeaseToken: "diagnosis-token", LeaseUntil: now.Add(2 * time.Minute), Now: now.Add(time.Minute),
	})
	if err != nil || !claimed || claimedWork == nil || claimedWork.WorkID != workID || claimedWork.EventID != after {
		t.Fatalf("real claim: work=%+v claimed=%v err=%v", claimedWork, claimed, err)
	}
	t.Logf("real claimant selected=%s (expected newer=%s)", claimedWork.EventID, newIDs[2])
	completed, err := svc.CompleteWorkItem(ctx, workID, "diagnosis-token", workqueue.WorkStateProcessed, now.Add(time.Minute))
	if err != nil || !completed {
		t.Fatalf("complete work: %v %v", completed, err)
	}
	var unfinished, uncovered int
	if err = db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE State NOT IN ('processed','skipped','dead')`).Scan(&unfinished); err != nil {
		t.Fatal(err)
	}
	if err = db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorEvent e WHERE e.Status='pending' AND e.EnqueueOrdinal>(SELECT MAX(s.EnqueueOrdinal) FROM _System_ConnectorWorkItem w JOIN _System_ConnectorEvent s ON s.Id=w.EventId)`).Scan(&uncovered); err != nil {
		t.Fatal(err)
	}
	t.Logf("after real completion: unfinished_workitems=%d uncovered_new_events=%d", unfinished, uncovered)
	if after != newIDs[2] || gap || uncovered != 0 {
		t.Fatalf("duplicate receipt rewound durable projection: newer events lost their execution owner")
	}
}