package connector_workqueue

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

	"github.com/DATA-DOG/go-sqlmock"
)

func queueStatsRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{
		"Namespace",
		"InstalledPackId",
		"QueueClass",
		"Total",
		"Pending",
		"FailedDue",
		"FailedNotDue",
		"Processing",
		"StaleProcessing",
		"Processed",
		"Dead",
		"RetryAttempts",
		"OldestPendingAt",
		"OldestDueAt",
		"OldestFailedDueAt",
		"OldestProcessingAt",
	})
}

func TestConnectorQueueAuthorityUsesWorkItemsWithoutEventHistory(t *testing.T) {
	query := connectorQueueStatsQuery(ConnectorQueueStatsAll)
	for _, required := range []string{
		"FROM _System_ConnectorWorkItem w",
		"w.WorkType IN ('connector_event', 'workflow_action_apply')",
		"AND 1 = 1",
		"w.State",
		"w.RunAfter",
		"w.LeaseUntil",
		"NOT (w.DependencyKind IS NOT NULL AND w.LeaseOwner IS NULL AND w.LeaseToken IS NULL AND w.LeaseUntil IS NULL)",
		"w.FailureCount",
		"JOIN _System_ConnectorEvent e",
	} {
		if !strings.Contains(query, required) {
			t.Fatalf("authoritative queue query missing %q", required)
		}
	}
	if got := connectorQueueWorkItemAuthorityPredicate("w", ConnectorQueueStatsAll); got != "1 = 1" {
		t.Fatalf("all-history WorkItem predicate = %q, want terminal rows included", got)
	}
	if live := connectorQueueWorkItemAuthorityPredicate("w", ConnectorQueueStatsLive); !strings.Contains(live, "w.State IN ('pending', 'retry', 'blocked', 'blocked_retry', 'processing')") {
		t.Fatalf("live WorkItem predicate does not exclude terminal rows: %s", live)
	} else if !strings.Contains(live, "w.ProjectionActiveKey") {
		t.Fatalf("live WorkItem predicate does not apply projection authority: %s", live)
	}
	for _, forbidden := range []string{"e.LastError", "e.Payload", "w.FailureSummary", "UNION ALL", "e.ExecutionOwner = 'event'"} {
		if strings.Contains(query, forbidden) {
			t.Fatalf("queue stats query reads diagnostic body %q", forbidden)
		}
	}
}

func TestConnectorQueueExactStatsMatchesWorkItemClaimFences(t *testing.T) {
	exact := connectorQueueStatsQuery(ConnectorQueueStatsLive)
	for _, required := range []string{
		"q.Claimable = 1",
		"w.LeaseUntil IS NOT NULL",
		"w.ProjectionActiveKey",
		"older.Generation = w.Generation",
		"older.State IN ('pending', 'retry', 'blocked', 'blocked_retry')",
		"older.LaneOrdinal < w.LaneOrdinal",
		"WHERE q.State IN",
	} {
		if !strings.Contains(exact, required) {
			t.Fatalf("exact queue stats query missing claim fence %q", required)
		}
	}
	if got := strings.Count(exact, "?"); got != 2 {
		t.Fatalf("exact query placeholders = %d, want WorkItem snapshot binds", got)
	}
	if got := connectorQueueStatsQueryArgs(ConnectorQueueStatsLive, time.Unix(1, 0), time.Unix(0, 0)); len(got) != 2 {
		t.Fatalf("exact query args = %d, want two", len(got))
	}
}

func TestConnectorQueueStatsModesSeparateExactAndCheapDue(t *testing.T) {
	exact := connectorQueueStatsQuery(ConnectorQueueStatsLive)
	cheap := connectorQueueStatsQuery(ConnectorQueueStatsCheapLive)
	if !strings.Contains(exact, "q.State = 'processing' AND q.Claimable = 1") {
		t.Fatalf("exact live stats do not apply claimability to stale processing")
	}
	if strings.Contains(cheap, "q.State = 'processing' AND q.Claimable = 1") ||
		!strings.Contains(cheap, "q.State = 'processing' AND q.RawDue = 1") ||
		!strings.Contains(cheap, "WHERE q.State IN") {
		t.Fatalf("cheap live query does not use raw WorkItem pressure")
	}
	for _, forbidden := range []string{"USE INDEX (idx_connector_work_item_lease_due)", "older.Generation = w.Generation", "q.Claimable"} {
		if strings.Contains(cheap, forbidden) {
			t.Fatalf("cheap live query evaluates exact claimability fragment %q", forbidden)
		}
	}
}

func TestRepositoryQueryConnectorEventQueueStatsScansWorkItemProjection(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()
	now := time.Date(2026, 7, 17, 4, 0, 0, 0, time.UTC)
	staleCutoff := now.Add(-5 * time.Minute)
	rows := queueStatsRows().AddRow(
		"hubspot", "pack-1", "bulk",
		int64(5), int64(1), int64(1), int64(1),
		int64(1), int64(1), int64(0), int64(1), int64(4),
		now.Add(-10*time.Minute), now.Add(-12*time.Minute),
		now.Add(-8*time.Minute), now.Add(-3*time.Minute),
	)
	mock.ExpectQuery(`(?s)WITH params AS.*FROM authoritative_queue q`).
		WithArgs(now, staleCutoff).
		WillReturnRows(rows)

	got, err := NewRepository(db, nil, nil).QueryConnectorEventQueueStats(
		context.Background(), now, staleCutoff, ConnectorQueueStatsAll,
	)
	if err != nil {
		t.Fatalf("QueryConnectorEventQueueStats: %v", err)
	}
	if len(got) != 1 || got[0].Namespace != "hubspot" ||
		got[0].QueueClass != "bulk" || got[0].Stats.Due != 3 ||
		got[0].Stats.OldestDueAgeSeconds != 720 {
		t.Fatalf("queue stats = %+v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestRepositoryQueryConnectorEventQueueStatsRejectsInvalidInputs(t *testing.T) {
	db, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()
	repo := NewRepository(db, nil, nil)
	now := time.Now().UTC()
	for _, tc := range []struct {
		name        string
		now         time.Time
		staleCutoff time.Time
		mode        ConnectorQueueStatsMode
	}{
		{name: "zero now", staleCutoff: now.Add(-time.Minute)},
		{name: "zero cutoff", now: now},
		{name: "cutoff after now", now: now, staleCutoff: now.Add(time.Minute)},
		{name: "unknown mode", now: now, staleCutoff: now.Add(-time.Minute), mode: 255},
	} {
		t.Run(tc.name, func(t *testing.T) {
			if _, err := repo.QueryConnectorEventQueueStats(context.Background(), tc.now, tc.staleCutoff, tc.mode); err == nil {
				t.Fatal("QueryConnectorEventQueueStats error = nil, want invalid input")
			}
		})
	}
}

func TestQueueAgeSecondsFutureClockSkewClampsToZero(t *testing.T) {
	now := time.Date(2026, 7, 17, 4, 0, 0, 0, time.UTC)
	if got := queueAgeSeconds(now, sql.NullTime{
		Time: now.Add(time.Minute), Valid: true,
	}); got != 0 {
		t.Fatalf("future age = %d, want 0", got)
	}
}