package connector_events

import (
	"context"
	"encoding/json"
	"errors"
	"reflect"
	"sort"
	"testing"
	"time"

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

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

func TestQueryQueueStatsAggregatesAuthoritativeWorkItemRows(t *testing.T) {
	svc, mock := newMock(t, "id-1")
	now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	svc.now = fixedNow(now)
	rows := connectorQueueStatsRows().
		AddRow("pack-a", "install-a", "interactive",
			int64(6), int64(1), int64(1), int64(1), int64(1), int64(1), int64(1), int64(0), int64(3),
			now.Add(-10*time.Minute), now.Add(-15*time.Minute), now.Add(-7*time.Minute), now.Add(-4*time.Minute)).
		AddRow("pack-b", "install-b", "bulk",
			int64(2), int64(2), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0),
			now.Add(-3*time.Minute), now.Add(-3*time.Minute), nil, nil)
	mock.ExpectQuery(`(?s)WITH params AS.*authoritative_queue AS.*FROM authoritative_queue q`).
		WithArgs(now, now.Add(-StaleProcessingSeconds*time.Second)).
		WillReturnRows(rows)

	stats, err := svc.QueryQueueStats(context.Background())
	if err != nil {
		t.Fatalf("QueryQueueStats: %v", err)
	}
	if got, want := stats.Global.Total, int64(8); got != want {
		t.Fatalf("Global.Total = %d, want %d", got, want)
	}
	if got, want := stats.Global.Due, int64(5); got != want {
		t.Fatalf("Global.Due = %d, want %d", got, want)
	}
	if got, want := stats.Global.OldestDueAgeSeconds, int64(900); got != want {
		t.Fatalf("Global.OldestDueAgeSeconds = %d, want %d", got, want)
	}
	if len(stats.ByQueueClass) != 2 ||
		stats.ByQueueClass[0].QueueClass != "interactive" ||
		stats.ByQueueClass[1].QueueClass != "bulk" {
		t.Fatalf("ByQueueClass = %+v", stats.ByQueueClass)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestQueryLiveQueueStatsUsesExactWorkItemAuthoritySnapshot(t *testing.T) {
	svc, mock := newMock(t, "id-1")
	now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	svc.now = fixedNow(now)
	rows := connectorQueueStatsRows().
		AddRow("pack-a", "install-a", "default",
			int64(3), int64(1), int64(1), int64(0), int64(1), int64(1), int64(0), int64(0), int64(2),
			now.Add(-10*time.Minute), now.Add(-12*time.Minute), now.Add(-7*time.Minute), now.Add(-4*time.Minute))
	mock.ExpectQuery(`(?s)WITH params AS.*q\.Claimable.*WHERE q\.State IN`).
		WithArgs(now, now.Add(-StaleProcessingSeconds*time.Second)).
		WillReturnRows(rows)

	stats, err := svc.QueryLiveQueueStats(context.Background())
	if err != nil {
		t.Fatalf("QueryLiveQueueStats: %v", err)
	}
	if stats.Global.Total != 3 || stats.Global.Due != 3 {
		t.Fatalf("Global = %+v", stats.Global)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestQueryCheapLiveQueueStatsUsesWorkItemAuthoritySnapshot(t *testing.T) {
	svc, mock := newMock(t, "id-1")
	now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	svc.now = fixedNow(now)
	rows := connectorQueueStatsRows().
		AddRow("pack-a", "install-a", "default",
			int64(4), int64(1), int64(1), int64(1), int64(1), int64(1), int64(0), int64(0), int64(2),
			now.Add(-10*time.Minute), now.Add(-12*time.Minute), now.Add(-7*time.Minute), now.Add(-4*time.Minute))
	mock.ExpectQuery(`(?s)WITH params AS.*authoritative_queue AS.*q\.RawDue.*WHERE q\.State IN`).
		WithArgs(now, now.Add(-StaleProcessingSeconds*time.Second)).
		WillReturnRows(rows)

	stats, err := svc.QueryCheapLiveQueueStats(context.Background())
	if err != nil {
		t.Fatalf("QueryCheapLiveQueueStats: %v", err)
	}
	if stats.Global.Total != 4 || stats.Global.Due != 3 ||
		stats.Global.OldestDueAgeSeconds != 720 {
		t.Fatalf("Global = %+v", stats.Global)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestQueueStatsFailsClosedWhenWorkItemSchemaIsMissing(t *testing.T) {
	now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	for _, tc := range []struct {
		name  string
		exact bool
		call  func(*Service) (QueueStats, error)
	}{
		{name: "all", call: func(s *Service) (QueueStats, error) {
			return s.QueryQueueStats(context.Background())
		}},
		{name: "live", exact: true, call: func(s *Service) (QueueStats, error) {
			return s.QueryLiveQueueStats(context.Background())
		}},
		{name: "cheap_live", call: func(s *Service) (QueueStats, error) {
			return s.QueryCheapLiveQueueStats(context.Background())
		}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			svc, mock := newMock(t, "id-1")
			svc.now = fixedNow(now)
			authorityQuery := mock.ExpectQuery(`(?s)WITH params AS.*authoritative_queue AS`)
			authorityQuery.WithArgs(now, now.Add(-StaleProcessingSeconds*time.Second))
			authorityQuery.WillReturnError(errors.New("Error 1146 (42S02): Table '_System_ConnectorWorkItem' doesn't exist"))

			stats, err := tc.call(svc)
			if err == nil || stats.Global.Total != 0 || len(stats.ByQueueClass) != 0 {
				t.Fatalf("queue stats = %+v, %v; want mandatory WorkItem schema error", stats, err)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sql expectations: %v", err)
			}
		})
	}
}

func TestQueueStatsFailsClosedWhenAdditiveWorkItemColumnIsMissing(t *testing.T) {
	now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	for _, tc := range []struct {
		name string
		err  error
	}{
		{name: "queue class", err: errors.New("Error 1054 (42S22): Unknown column 'w.QueueClass' in 'field list'")},
		{name: "projection", err: errors.New("Error 1054 (42S22): Unknown column 'w.ProjectionActiveKey' in 'where clause'")},
		{name: "execution owner", err: errors.New("Error 1054 (42S22): Unknown column 'e.ExecutionOwner' in 'where clause'")},
	} {
		t.Run(tc.name, func(t *testing.T) {
			svc, mock := newMock(t, "id-1")
			svc.now = fixedNow(now)
			mock.ExpectQuery(`(?s)WITH params AS.*authoritative_queue AS`).
				WithArgs(now, now.Add(-StaleProcessingSeconds*time.Second)).
				WillReturnError(tc.err)

			stats, err := svc.QueryQueueStats(context.Background())
			if err == nil || stats.Global.Total != 0 || len(stats.ByQueueClass) != 0 {
				t.Fatalf("queue stats = %+v, %v; want mandatory WorkItem schema error", stats, err)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sql expectations: %v", err)
			}
		})
	}
}

func TestQueueStatsDoesNotHideUnrelatedAuthorityQueryError(t *testing.T) {
	now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	for _, tc := range []struct {
		name string
		err  error
	}{
		{name: "connection", err: errors.New("driver: bad connection")},
		{name: "unrecognized column", err: errors.New("Error 1054 (42S22): Unknown column 'w.DoesNotExist' in 'field list'")},
	} {
		t.Run(tc.name, func(t *testing.T) {
			svc, mock := newMock(t, "id-1")
			svc.now = fixedNow(now)
			mock.ExpectQuery(`(?s)WITH params AS.*authoritative_queue AS`).
				WithArgs(now, now.Add(-StaleProcessingSeconds*time.Second)).
				WillReturnError(tc.err)

			if _, err := svc.QueryQueueStats(context.Background()); err == nil {
				t.Fatal("QueryQueueStats error = nil, want unrelated authority query error")
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sql expectations: %v", err)
			}
		})
	}
}

func TestQueueStatsPublishedJSONShapeRemainsStable(t *testing.T) {
	stats := QueueStats{
		CollectedAt: time.Date(2026, 7, 17, 4, 0, 0, 0, time.UTC),
		Global: QueueBacklogStats{
			Total: 1, Pending: 2, Due: 3, FailedDue: 4, FailedNotDue: 5,
			Processing: 6, StaleProcessing: 7, Processed: 8, Dead: 9,
			RetryAttempts: 10, OldestPendingAgeSeconds: 11,
			OldestDueAgeSeconds: 12, OldestFailedAgeSeconds: 13,
			OldestProcessingSeconds: 14,
		},
		ByQueueClass: []QueueClassStats{{
			Namespace: "ns", InstalledPackID: "pack", QueueClass: "bulk",
		}},
		DrainThrottle: DrainThrottleState{State: "unwired", Reason: "test"},
	}
	payload, err := json.Marshal(stats)
	if err != nil {
		t.Fatalf("json.Marshal: %v", err)
	}
	var got map[string]any
	if err := json.Unmarshal(payload, &got); err != nil {
		t.Fatalf("json.Unmarshal: %v", err)
	}
	if want := []string{"by_queue_class", "collected_at", "drain_throttle", "global"}; !reflect.DeepEqual(sortedJSONKeys(got), want) {
		t.Fatalf("top-level JSON keys = %v, want %v", sortedJSONKeys(got), want)
	}
	global, ok := got["global"].(map[string]any)
	if !ok {
		t.Fatalf("global JSON = %#v", got["global"])
	}
	wantGlobal := []string{
		"dead", "due", "failed_due", "failed_not_due", "oldest_due_age_seconds",
		"oldest_failed_age_seconds", "oldest_pending_age_seconds",
		"oldest_processing_age_seconds", "pending", "processed", "processing",
		"retry_attempts", "stale_processing", "total",
	}
	if !reflect.DeepEqual(sortedJSONKeys(global), wantGlobal) {
		t.Fatalf("global JSON keys = %v, want %v", sortedJSONKeys(global), wantGlobal)
	}
}

func sortedJSONKeys(value map[string]any) []string {
	keys := make([]string, 0, len(value))
	for key := range value {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	return keys
}