package controlplane

import (
	"context"
	"regexp"
	"strings"
	"testing"

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

func TestBackfillConnectorEventSourceLanesMaterializesAndStamps(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectMigrationApplied(mock, migrationIdBackfillConnectorEventSourceLanes, false)
	expectConnectorEventSourceLaneColumnsReady(mock, 4)
	mock.ExpectExec(regexp.QuoteMeta(connectorEventSourceLaneBackfillUpdateSQL)).
		WillReturnResult(sqlmock.NewResult(0, 2))
	expectConnectorEventSourceLanePayloadOnlyCount(mock, 0)
	mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO _System_Migration (Id, Version, Description, AppliedAt, CreatedDate, LastModifiedDate)
			VALUES (?, ?, ?, NOW(), NOW(), NOW())
			ON DUPLICATE KEY UPDATE LastModifiedDate = NOW()`)).
		WithArgs(migrationIdBackfillConnectorEventSourceLanes, 1, sqlmock.AnyArg()).
		WillReturnResult(sqlmock.NewResult(0, 1))

	touched, err := BackfillConnectorEventSourceLanes(context.Background(), db)
	if err != nil {
		t.Fatalf("BackfillConnectorEventSourceLanes: %v", err)
	}
	if touched != 2 {
		t.Fatalf("BackfillConnectorEventSourceLanes touched %d rows, want 2", touched)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestBackfillConnectorEventSourceLanesSkipsWhenStamped(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectMigrationApplied(mock, migrationIdBackfillConnectorEventSourceLanes, true)
	expectConnectorEventSourceLanePayloadOnlyCount(mock, 0)

	touched, err := BackfillConnectorEventSourceLanes(context.Background(), db)
	if err != nil {
		t.Fatalf("BackfillConnectorEventSourceLanes: %v", err)
	}
	if touched != 0 {
		t.Fatalf("BackfillConnectorEventSourceLanes touched %d rows, want 0", touched)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestBackfillConnectorEventSourceLanesSQLShape(t *testing.T) {
	for _, want := range []string{
		"SourceLaneKey = SHA2",
		"SourceObjectKind =",
		"SourceObjectID =",
		"JSON_EXTRACT(Payload, '$.objectKind')",
		"JSON_EXTRACT(Payload, '$.sourceID')",
	} {
		if !strings.Contains(connectorEventSourceLaneBackfillUpdateSQL, want) {
			t.Fatalf("backfill SQL missing %q:\n%s", want, connectorEventSourceLaneBackfillUpdateSQL)
		}
	}
}

func expectConnectorEventSourceLaneColumnsReady(mock sqlmock.Sqlmock, count int) {
	mock.ExpectQuery(regexp.QuoteMeta(`
SELECT COUNT(*)
  FROM INFORMATION_SCHEMA.COLUMNS
 WHERE TABLE_SCHEMA = DATABASE()
   AND TABLE_NAME = '_System_ConnectorEvent'
   AND COLUMN_NAME IN ('Payload', 'SourceObjectKind', 'SourceObjectID', 'SourceLaneKey')`)).
		WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(count))
}

func expectConnectorEventSourceLanePayloadOnlyCount(mock sqlmock.Sqlmock, count int) {
	mock.ExpectQuery(regexp.QuoteMeta(`
SELECT COUNT(*)
  FROM _System_ConnectorEvent
 WHERE Payload IS NOT NULL
   AND NULLIF(JSON_UNQUOTE(JSON_EXTRACT(Payload, '$.objectKind')), '') IS NOT NULL
   AND NULLIF(JSON_UNQUOTE(JSON_EXTRACT(Payload, '$.sourceID')), '') IS NOT NULL
   AND (
        SourceObjectKind IS NULL OR SourceObjectKind = ''
        OR SourceObjectID IS NULL OR SourceObjectID = ''
        OR SourceLaneKey IS NULL OR SourceLaneKey = ''
   )`)).
		WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(count))
}

func expectMigrationApplied(mock sqlmock.Sqlmock, migrationID string, applied bool) {
	q := mock.ExpectQuery(regexp.QuoteMeta(`SELECT 1 FROM _System_Migration WHERE Id = ? LIMIT 1`)).
		WithArgs(migrationID)
	if applied {
		q.WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
	} else {
		q.WillReturnRows(sqlmock.NewRows([]string{"1"}))
	}
}