package connector_events

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

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

func TestService_SourceProjectionDuplicateSeesCommittedGenerationAfterWaiting(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	const ns, pack, kind = "projection-snapshot", "install-token-projection-snapshot", "contact.propertychange"
	createInstalledPackTable(t, db, ns)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, ns)
	makeSettledProjectionRoute(t, db, ns, kind)
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()
	now := time.Now().UTC()
	writer := NewService(db, nil, func() time.Time { return now })
	event := func(key string) BatchEvent {
		return BatchEvent{EventKind: kind, IdempotencyKey: key, RouteVersion: 1,
			Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"snapshot-source"}`)}
	}
	oldIDs, _, err := writer.EnqueueBatch(ctx, ns, pack, "delivery-A", []BatchEvent{event("A")})
	if err != nil {
		t.Fatal(err)
	}
	started, resume, done := make(chan struct{}), make(chan struct{}), make(chan error, 1)
	var releaseOnce sync.Once
	release := func() { releaseOnce.Do(func() { close(resume) }) }
	waiter := NewService(db, nil, func() time.Time { return now.Add(2 * time.Minute) })
	waiter.afterEnqueueTransactionBegin = func() {
		close(started)
		select {
		case <-resume:
		case <-ctx.Done():
		}
	}
	go func() {
		ids, dups, err := waiter.EnqueueBatch(ctx, ns, pack, "delivery-A", []BatchEvent{event("A")})
		if err == nil && (len(ids) != 1 || ids[0] != oldIDs[0] || len(dups) != 1 || !dups[0]) {
			err = wq.ErrInvalidWorkItem
		}
		done <- err
	}()
	joined := false
	defer func() {
		release()
		if !joined {
			cancel()
			select {
			case <-done:
			case <-time.After(5 * time.Second):
				t.Error("waiter did not stop")
			}
		}
	}()
	select {
	case <-started:
	case <-ctx.Done():
		t.Fatal(ctx.Err())
	}
	// Commit a new selected Event and terminal generation AFTER the duplicate
	// transaction began, before allowing it to read/lock admission state.
	now = now.Add(time.Minute)
	newIDs, _, err := writer.EnqueueBatch(ctx, ns, pack, "delivery-B", []BatchEvent{event("B")})
	if err != nil {
		t.Fatal(err)
	}
	work, claimed, err := writer.WorkItemRepository().ClaimNextWorkItem(wq.WithDistributedClaim(ctx), wq.ClaimWorkItemInput{
		InstalledPackID: pack, OwnerID: "snapshot-worker", FencingToken: "snapshot-fence",
		LeaseToken: "snapshot-lease", Now: now.Add(time.Minute), LeaseUntil: now.Add(3 * time.Minute),
	})
	if err != nil || !claimed || work == nil || work.EventID != newIDs[0] {
		t.Fatalf("claim: work=%+v claimed=%v err=%v", work, claimed, err)
	}
	completed, err := writer.CompleteWorkItem(ctx, work.WorkID, "snapshot-lease", wq.WorkStateProcessed, now.Add(time.Minute))
	if err != nil || !completed {
		t.Fatalf("complete: %v %v", completed, err)
	}
	release()
	select {
	case err = <-done:
		joined = true
	case <-ctx.Done():
		t.Fatal(ctx.Err())
	}
	if err != nil {
		t.Fatal(err)
	}
	var total, unfinished int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*), SUM(State NOT IN ('processed','skipped','dead')) FROM _System_ConnectorWorkItem WHERE InstalledPackId=?`, pack).Scan(&total, &unfinished); err != nil {
		t.Fatal(err)
	}
	if total != 1 || unfinished != 0 {
		t.Fatalf("duplicate resurrected covered history: total=%d unfinished=%d", total, unfinished)
	}
}