package connector_events

import (
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"testing"
	"time"

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

func TestIngressBatchProjectionLaterChunkRollsBackEarlierWrites(t *testing.T) {
	f := newProjectionAdmissionFixture(t)
	batch := func(round int) []BatchEvent {
		out := make([]BatchEvent, 130)
		for i := range out {
			out[i] = projectionAdmissionEvent(fmt.Sprintf("atomic-%d-%d", round, i))
			out[i].Payload = json.RawMessage(fmt.Sprintf(`{"objectKind":"contact","sourceID":"atomic-source-%d"}`, i))
		}
		return out
	}
	if _, _, err := f.svc.EnqueueBatch(f.ctx, projectionAdmissionNamespace, projectionAdmissionPack, "initial", batch(0)); err != nil {
		t.Fatal(err)
	}
	var last string
	var first int64
	if err := f.db.QueryRow(`SELECT WorkId,InputStartOrdinal FROM _System_ConnectorWorkItem ORDER BY ProjectionKey DESC LIMIT 1`).Scan(&last, &first); err != nil {
		t.Fatal(err)
	}
	// Admission orders by projection key; this invalid range is deliberately in
	// the second chunk, after the first chunk has performed its UPDATEs.
	if _, err := f.db.Exec(`UPDATE _System_ConnectorWorkItem SET InputStartOrdinal=0 WHERE WorkId=?`, last); err != nil {
		t.Fatal(err)
	}
	before := f.snapshot()
	if _, _, err := f.svc.EnqueueBatch(f.ctx, projectionAdmissionNamespace, projectionAdmissionPack, "invalid", batch(1)); !errors.Is(err, wq.ErrInvalidWorkItem) {
		t.Fatalf("expected invalid range, got %v", err)
	}
	if !reflect.DeepEqual(before, f.snapshot()) {
		t.Fatal("partial projection writes escaped rollback")
	}
	var events int
	if err := f.db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorEvent`).Scan(&events); err != nil || events != 130 {
		t.Fatalf("partial immutable inputs escaped rollback: %d/%v", events, err)
	}
	if _, err := f.db.Exec(`UPDATE _System_ConnectorWorkItem SET InputStartOrdinal=? WHERE WorkId=?`, first, last); err != nil {
		t.Fatal(err)
	}
	if _, dups, err := f.svc.EnqueueBatch(f.ctx, projectionAdmissionNamespace, projectionAdmissionPack, "retry", batch(1)); err != nil {
		t.Fatal(err)
	} else {
		for _, dup := range dups {
			if dup {
				t.Fatal("rolled-back input incorrectly treated as duplicate")
			}
		}
	}
	for _, work := range f.snapshot() {
		if ids := f.payloadIDs(work.EventID); len(ids) != 2 {
			t.Fatalf("retry coverage=%d want=2", len(ids))
		}
	}
}

func TestIngressBatchProjectionWalkingAndFrozenGenerations(t *testing.T) {
	f := newProjectionAdmissionFixture(t)
	batch := func(round int) []BatchEvent {
		var events []BatchEvent
		for delta := 0; delta < 3; delta++ {
			for source := 0; source < 4; source++ {
				e := projectionAdmissionEvent(fmt.Sprintf("round-%d-source-%d-delta-%d", round, source, delta))
				e.Payload = json.RawMessage(fmt.Sprintf(`{"objectKind":"contact","sourceID":"batch-source-%d"}`, source))
				events = append(events, e)
			}
		}
		return events
	}
	expected := map[string]bool{}
	admit := func(round int) {
		t.Helper()
		events := batch(round)
		ids, dups, err := f.svc.EnqueueBatch(f.ctx, projectionAdmissionNamespace, projectionAdmissionPack, "batch", events)
		if err != nil {
			t.Fatal(err)
		}
		for i, id := range ids {
			if dups[i] {
				t.Fatal("fresh duplicate")
			}
			expected[id] = true
		}
	}
	admit(0)
	active := rangeTestClaim(f)
	frozen := rangeTestIDs(f, active)
	admit(1) // Extend three active ranges; append a successor for the claimed one.
	admit(2) // Extend that successor without changing its lower bound.
	if got := rangeTestIDs(f, active); !reflect.DeepEqual(got, frozen) {
		t.Fatal("active changed", got)
	}
	consumed := map[string]bool{}
	consume := func(work *wq.WorkItem) {
		t.Helper()
		for _, id := range rangeTestIDs(f, work) {
			if consumed[id] {
				t.Fatal("executed twice", id)
			}
			consumed[id] = true
		}
		rangeTestComplete(f, work, wq.WorkStateProcessed)
	}
	consume(active)
	for i := 0; i < 4; i++ {
		consume(rangeTestClaim(f))
	}
	if work := rangeTestClaim(f); work != nil {
		t.Fatal("extra execution", work.WorkID)
	}
	if !reflect.DeepEqual(expected, consumed) {
		t.Fatalf("accepted=%d consumed=%d", len(expected), len(consumed))
	}
	before := f.snapshot()
	if _, dups, err := f.svc.EnqueueBatch(f.ctx, projectionAdmissionNamespace, projectionAdmissionPack, "duplicate", batch(0)); err != nil || len(dups) != 12 {
		t.Fatalf("duplicate:%v/%v", dups, err)
	}
	if after := f.snapshot(); !reflect.DeepEqual(before, after) {
		t.Fatal("duplicate changed terminal work")
	}
}

func TestIngressBatchProjectionConcurrentCompletion(t *testing.T) {
	f := newProjectionAdmissionFixture(t)
	makeBatch := func(round int) []BatchEvent {
		out := make([]BatchEvent, 4)
		for i := range out {
			out[i] = projectionAdmissionEvent(fmt.Sprintf("race-%d-%d", round, i))
			out[i].Payload = json.RawMessage(fmt.Sprintf(`{"objectKind":"contact","sourceID":"race-source-%d"}`, i))
		}
		return out
	}
	expected, consumed := map[string]bool{}, map[string]bool{}
	for round := 0; round < 6; round++ {
		ids, _, err := f.svc.EnqueueBatch(f.ctx, projectionAdmissionNamespace, projectionAdmissionPack, "initial", makeBatch(round*2))
		if err != nil {
			t.Fatal(err)
		}
		for _, id := range ids {
			expected[id] = true
		}
		active := rangeTestClaim(f)
		for _, id := range rangeTestIDs(f, active) {
			if consumed[id] {
				t.Fatal("duplicate", id)
			}
			consumed[id] = true
		}
		done := make(chan error, 1)
		go func() {
			ok, err := f.svc.CompleteWorkItem(f.ctx, active.WorkID, active.LeaseToken, wq.WorkStateProcessed, f.now.Add(10*time.Minute))
			if err == nil && !ok {
				err = fmt.Errorf("completion lost")
			}
			done <- err
		}()
		ids, _, err = f.svc.EnqueueBatch(f.ctx, projectionAdmissionNamespace, projectionAdmissionPack, "racing", makeBatch(round*2+1))
		completionErr := <-done
		if err != nil || completionErr != nil {
			t.Fatalf("enqueue=%v complete=%v", err, completionErr)
		}
		for _, id := range ids {
			expected[id] = true
		}
		for {
			work := rangeTestClaim(f)
			if work == nil {
				break
			}
			for _, id := range rangeTestIDs(f, work) {
				if consumed[id] {
					t.Fatal("duplicate", id)
				}
				consumed[id] = true
			}
			rangeTestComplete(f, work, wq.WorkStateProcessed)
		}
	}
	if !reflect.DeepEqual(expected, consumed) {
		t.Fatalf("accepted=%d consumed=%d", len(expected), len(consumed))
	}
}

func TestIngressBatchSettleMatchesTypedSourceOracle(t *testing.T) {
	f := newProjectionAdmissionFixture(t)
	// Include a case-insensitive source column: SQL comparison, not a Go map
	// keyed by returned spelling, must decide which requested keys match.
	if _, err := f.db.Exec(`ALTER TABLE _System_ConnectorEvent MODIFY SourceObjectID LONGTEXT COLLATE utf8mb4_general_ci`); err != nil {
		t.Fatal(err)
	}
	now := f.now.UTC().Truncate(time.Second)
	for i, seed := range []struct {
		source, kind, companion, flow, lane string
		version                             int
		seconds                             int
	}{
		{"Case-ID", "contact.propertychange", "", "", "override-a", 1, 2},
		{"case-id", "contact.propertychange", "", "", "override-b", 1, 4},
		{"Case-ID", "contact.propertychange", "", "", "override-a", 2, 1},
		{"other", "contact.propertychange", "", "", "override-a", 1, 3},
		{"Case-ID", "contact.propertychange", "companion", "flow", "override-a", 1, 1},
		{"Case-ID", "contact.creation", "", "", "override-a", 1, 1},
	} {
		_, err := f.db.Exec(`INSERT INTO _System_ConnectorEvent (Id,Namespace,InstalledPackId,EventKind,IdempotencyKey,Status,
 SourceObjectKind,SourceObjectID,SourceLaneKey,RouteVersion,CompanionPackInstallId,CompanionFlowVersionId,NextAttemptAt,RetryCount,CreatedDate,LastModifiedDate)
 VALUES (?,?,?,?,?,'pending','contact',?,?,?,?,?,?,0,?,?)`, fmt.Sprint("settle-", i), projectionAdmissionNamespace, projectionAdmissionPack, seed.kind, fmt.Sprint("settle-", i), seed.source, seed.lane, seed.version, seed.companion, seed.flow, now.Add(time.Duration(seed.seconds)*time.Second), now, now)
		if err != nil {
			t.Fatal(err)
		}
	}
	keys := []sourceSettleKey{
		{projectionAdmissionPack, "contact.propertychange", 1, "", "", "contact", "CASE-ID"},
		{projectionAdmissionPack, "contact.propertychange", 1, "", "", "contact", "other"},
		{projectionAdmissionPack, "contact.propertychange", 2, "", "", "contact", "Case-ID"},
		{projectionAdmissionPack, "contact.propertychange", 1, "companion", "flow", "contact", "Case-ID"},
		{projectionAdmissionPack, "contact.creation", 1, "", "", "contact", "Case-ID"},
	}
	for i := 0; i < 260; i++ {
		keys = append(keys, sourceSettleKey{projectionAdmissionPack, "contact.propertychange", 1, "", "", "contact", fmt.Sprint("absent-", i)})
	}
	tx, err := f.db.BeginTx(f.ctx, nil)
	if err != nil {
		t.Fatal(err)
	}
	defer tx.Rollback()
	deadlines := map[sourceSettleKey]sql.NullTime{}
	for _, key := range keys {
		deadlines[key] = sql.NullTime{Time: now.Add(5 * time.Second), Valid: true}
	}
	if err := loadSourceSettleDeadlines(f.ctx, tx, now, deadlines); err != nil {
		t.Fatal(err)
	}
	for _, key := range keys[:5] {
		var expected sql.NullTime
		err := tx.QueryRowContext(f.ctx, `SELECT MIN(NextAttemptAt) FROM _System_ConnectorEvent
 WHERE InstalledPackId=? AND EventKind=? AND COALESCE(RouteVersion,0)=?
 AND COALESCE(CompanionPackInstallId,'')=? AND COALESCE(CompanionFlowVersionId,'')=?
 AND SourceObjectKind=? AND SourceObjectID=? AND Status='pending' AND RetryCount=0 AND NextAttemptAt>?`, key.installedPackID, key.eventKind, key.routeVersion, key.companionPack, key.companionFlow, key.sourceKind, key.sourceID, now).Scan(&expected)
		if err != nil {
			t.Fatal(err)
		}
		if !deadlines[key].Time.Equal(expected.Time) {
			t.Fatalf("%+v: got%v want%v", key, deadlines[key], expected)
		}
	}
	for _, key := range keys[5:] {
		if !deadlines[key].Time.Equal(now.Add(5 * time.Second)) {
			t.Fatal("missing default", key)
		}
	}
	// TestIngressBatchSettleReadScope checks physical work on skewed data;
	// using a particular index alone does not establish a bounded read scope.
}