package connector_events

import (
	"context"
	"database/sql"
	"fmt"
	"reflect"
	"strconv"
	"strings"
	"testing"
	"time"
)

func TestIngressBatchSettleReadScope(t *testing.T) {
	for _, shape := range []string{"other_pack_future", "same_pack_expired"} {
		t.Run(shape, func(t *testing.T) {
			const otherNamespace = "settle-unrelated"
			f := newProjectionAdmissionFixture(t, otherNamespace)
			ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
			defer cancel()
			f.ctx = ctx
			otherPack := "install-token-" + otherNamespace
			namespace, pack := otherNamespace, otherPack
			if shape == "same_pack_expired" {
				namespace, pack = projectionAdmissionNamespace, projectionAdmissionPack
			}
			for _, p := range []string{projectionAdmissionPack, otherPack} {
				if err := f.svc.RequireSourceProjectionProtocol(f.ctx, p); err != nil {
					t.Fatal(err)
				}
			}
			if err := RequireSourceProjectionProtocol(f.ctx, f.db); err != nil {
				t.Fatal(err)
			}

			// Native producer with the real five-second delay. Advance the fixture's
			// logical clock, not wall-clock sleeps or modified schema/deadline columns.
			backlog := make([]BatchEvent, 20000)
			for i := range backlog {
				backlog[i] = projectionAdmissionEvent(fmt.Sprintf("unrelated-%d", i))
			}
			backlogIDs, _, err := f.svc.EnqueueBatch(f.ctx, namespace, pack, "backlog", backlog)
			if err != nil {
				t.Fatal(err)
			}
			if shape == "same_pack_expired" {
				f.now = f.now.Add(6 * time.Second)
			}
			selected := successorBoundBatch(0, 256)
			selectedIDs, _, err := f.svc.EnqueueBatch(f.ctx, projectionAdmissionNamespace, projectionAdmissionPack, "selected", selected)
			if err != nil {
				t.Fatal(err)
			}
			var future int
			if err := f.db.QueryRowContext(f.ctx, `SELECT COUNT(*) FROM _System_ConnectorEvent WHERE InstalledPackId=? AND SourceObjectID='historical-contact' AND Status='pending' AND RetryCount=0 AND NextAttemptAt>?`, pack, f.now).Scan(&future); err != nil {
				t.Fatal(err)
			}
			wantFuture := 20000
			if shape == "same_pack_expired" {
				wantFuture = 0
			}
			if future != wantFuture {
				t.Fatalf("fixture future backlog=%d want=%d", future, wantFuture)
			}
			if _, err := f.db.ExecContext(f.ctx, `ANALYZE TABLE _System_ConnectorEvent`); err != nil {
				t.Fatal(err)
			}

			keys := make([]sourceSettleKey, len(selected))
			for i := range keys {
				keys[i] = sourceSettleKey{projectionAdmissionPack, projectionAdmissionKind, 1, "", "", "contact", fmt.Sprintf("cap-source-%d", i)}
			}
			query, args := sourceSettleBatchQuery(keys, f.now)
			values := make([]sql.NullTime, len(keys))
			dest := make([]any, len(keys))
			for i := range values {
				dest[i] = &values[i]
			}
			if err := f.db.QueryRowContext(f.ctx, query, args...).Scan(dest...); err != nil {
				t.Fatal(err)
			}
			for _, value := range values {
				if !value.Valid || !value.Time.Equal(f.now.Add(5*time.Second)) {
					t.Fatalf("deadline=%v", value)
				}
			}
			requireSettleScanBound(t, f, query, args, 256)

			// Both callers of the shared lookup preserve every fresh input; the fallback
			// also acknowledges its duplicate without executing it again.
			fresh := []BatchEvent{selected[0], selected[0]}
			fresh[0].IdempotencyKey = "fresh-a"
			fresh[1].IdempotencyKey = "fresh-b"
			freshIDs, dups, err := f.svc.EnqueueBatch(f.ctx, projectionAdmissionNamespace, projectionAdmissionPack, "fresh", fresh)
			if err != nil || !reflect.DeepEqual(dups, []bool{false, false}) {
				t.Fatalf("fresh dups=%v err=%v", dups, err)
			}
			fallback := []BatchEvent{selected[0], selected[0]}
			fallback[0].IdempotencyKey = "fallback-fresh"
			fallbackIDs, dups, err := f.svc.EnqueueBatch(f.ctx, projectionAdmissionNamespace, projectionAdmissionPack, "fallback", fallback)
			if err != nil || !reflect.DeepEqual(dups, []bool{false, true}) {
				t.Fatalf("fallback dups=%v err=%v", dups, err)
			}
			want := []string{selectedIDs[0], freshIDs[0], freshIDs[1], fallbackIDs[0]}
			if got := f.payloadIDs(fallbackIDs[0]); !reflect.DeepEqual(got, want) {
				t.Fatalf("selected inputs=%v want=%v", got, want)
			}
			if got := f.payloadIDs(backlogIDs[len(backlogIDs)-1]); !reflect.DeepEqual(got, backlogIDs) {
				t.Fatalf("unrelated inputs changed: count=%d want=%d", len(got), len(backlogIDs))
			}
		})
	}
}

// Check the physical work class, not a particular SQL string or index name.
// Reject full scans: their actRows can already reflect pushed-down filtering.
func requireSettleScanBound(t *testing.T, f *projectionAdmissionFixture, query string, args []any, bound int64) {
	t.Helper()
	rows, err := f.db.QueryContext(f.ctx, "EXPLAIN ANALYZE "+query, args...)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	columns, err := rows.Columns()
	if err != nil {
		t.Fatal(err)
	}
	var indexRows, tableRows int64
	for rows.Next() {
		raw := make([]sql.RawBytes, len(columns))
		dest := make([]any, len(raw))
		for i := range raw {
			dest[i] = &raw[i]
		}
		if err := rows.Scan(dest...); err != nil {
			t.Fatal(err)
		}
		if len(raw) < 3 {
			t.Fatal("EXPLAIN ANALYZE did not return actual rows")
		}
		id := string(raw[0])
		if strings.Contains(id, "FullScan") {
			t.Fatalf("unbounded settle scan: %s", id)
		}
		index := strings.Contains(id, "IndexRangeScan")
		table := strings.Contains(id, "TableRangeScan") || strings.Contains(id, "TableRowIDScan")
		if !index && !table {
			continue
		}
		count, err := strconv.ParseInt(string(raw[2]), 10, 64)
		if err != nil {
			t.Fatal(err)
		}
		if index {
			indexRows += count
		} else {
			tableRows += count
		}
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	if indexRows+tableRows == 0 {
		t.Fatal("no physical scan evidence")
	}
	if indexRows > bound || tableRows > bound {
		t.Fatalf("unrelated rows scanned: index=%d table=%d bound=%d", indexRows, tableRows, bound)
	}
	t.Logf("settle actual index_rows=%d table_rows=%d bound=%d", indexRows, tableRows, bound)
}