package catalogpublication

import (
	"context"
	"errors"
	"math"
	"regexp"
	"strconv"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	catalogpublicationv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalogpublicationv1"
)

const (
	publicationQueryPattern = `(?s)SELECT PublicationDigest, BindingVersion, OrgId, CatalogRevision,.*FROM _System_CatalogPublication WHERE OrgId = \? AND PublicationDigest = \?`
	activeQueryPattern      = `(?s)SELECT OrgId, ActivePublicationPresent, ActivePublicationDigest,.*FROM _System_CatalogPublicationActive WHERE OrgId = \?`
	commandQueryPattern     = `(?s)SELECT OrgId, CommandId, RequestDigest, RuntimeAudience,.*FROM _System_CatalogPublicationCommand WHERE OrgId = \? AND CommandId = \?`
	outcomeQueryPattern     = `(?s)SELECT OrgId, CommandId, CandidatePublicationDigest, Result,.*FROM _System_CatalogPublicationOutcome WHERE OrgId = \? AND CommandId = \?`
	historyQueryPattern     = `(?s)SELECT SequenceNumber, OrgId, CommandId,.*FROM _System_CatalogPublicationHistory.*ORDER BY SequenceNumber ASC LIMIT \?`
)

func TestReadExactRecomputesBindingDigestAndParsesUint64WithoutSignedIntermediate(t *testing.T) {
	authority, mock, closeDB := newReadTestAuthority(t)
	defer closeDB()

	binding := readTestBinding()
	binding.CatalogRevision = math.MaxUint64
	binding.SecurityEpoch = uint64(math.MaxInt64) + 1
	digest := mustBindingDigest(t, binding)
	mock.ExpectQuery(publicationQueryPattern).
		WithArgs(binding.OrgID, digest).
		WillReturnRows(publicationRows().AddRow(
			digest, legacyPublicationBindingVersion, binding.OrgID,
			strconv.FormatUint(binding.CatalogRevision, 10), binding.CatalogSnapshotDigest,
			"2", "3", binding.LayoutValidationViewDigest, nil,
			binding.LayoutActivationSetDigest, strconv.FormatUint(binding.SecurityEpoch, 10),
			binding.LayoutRoutingSecurityViewDigest,
		))

	publication, err := authority.ReadExact(context.Background(), binding.OrgID, digest)
	if err != nil {
		t.Fatal(err)
	}
	if publication.Digest != digest || publication.Binding != binding {
		t.Fatalf("publication = %#v, want digest %q and binding %#v", publication, digest, binding)
	}
	assertReadExpectations(t, mock)
}

func TestReadExactMapsAbsenceAndRejectsDigestMismatchedBinding(t *testing.T) {
	for _, test := range []struct {
		name string
		rows *sqlmock.Rows
		want error
	}{
		{
			name: "absent",
			rows: publicationRows(),
			want: ErrNotFound,
		},
		{
			name: "digest does not bind row",
			rows: publicationRows().AddRow(
				testDigest("8"), legacyPublicationBindingVersion, "org-1", "7",
				testDigest("1"), "2", "3", testDigest("2"), nil,
				testDigest("3"), "11", testDigest("4"),
			),
			want: ErrCorrupt,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			authority, mock, closeDB := newReadTestAuthority(t)
			defer closeDB()
			mock.ExpectQuery(publicationQueryPattern).
				WithArgs("org-1", testDigest("8")).
				WillReturnRows(test.rows)
			_, err := authority.ReadExact(context.Background(), "org-1", testDigest("8"))
			if !errors.Is(err, test.want) {
				t.Fatalf("error = %v, want %v", err, test.want)
			}
			assertReadExpectations(t, mock)
		})
	}
}

func TestReadExactRejectsBindingVersionAndReceiptPresenceMismatch(t *testing.T) {
	binding := readTestBinding()
	digest := mustBindingDigest(t, binding)
	for _, test := range []struct {
		name           string
		bindingVersion any
		receiptDigest  any
	}{
		{name: "missing binding version", bindingVersion: nil, receiptDigest: nil},
		{name: "unknown binding version", bindingVersion: "nexuscrm.catalog-publication-binding/v2", receiptDigest: nil},
		{name: "v1 with empty receipt digest", bindingVersion: legacyPublicationBindingVersion, receiptDigest: ""},
		{name: "v1 with receipt digest", bindingVersion: legacyPublicationBindingVersion, receiptDigest: testDigest("9")},
	} {
		t.Run(test.name, func(t *testing.T) {
			authority, mock, closeDB := newReadTestAuthority(t)
			defer closeDB()
			mock.ExpectQuery(publicationQueryPattern).
				WithArgs(binding.OrgID, digest).
				WillReturnRows(publicationRows().AddRow(
					digest, test.bindingVersion, binding.OrgID, "7",
					binding.CatalogSnapshotDigest, "2", "3",
					binding.LayoutValidationViewDigest, test.receiptDigest,
					binding.LayoutActivationSetDigest, "11",
					binding.LayoutRoutingSecurityViewDigest,
				))

			_, err := authority.ReadExact(context.Background(), binding.OrgID, digest)
			if !errors.Is(err, ErrCorrupt) {
				t.Fatalf("error = %v, want ErrCorrupt", err)
			}
			assertReadExpectations(t, mock)
		})
	}
}

func TestReadActiveFollowsExactAuthorityPointerWithoutServingFallback(t *testing.T) {
	authority, mock, closeDB := newReadTestAuthority(t)
	defer closeDB()

	binding := readTestBinding()
	digest := mustBindingDigest(t, binding)
	commandID := "018f0000-0000-7000-9000-000000000001"
	mock.ExpectQuery(activeQueryPattern).
		WithArgs(binding.OrgID).
		WillReturnRows(activeRows().AddRow(binding.OrgID, true, digest, commandID, "9"))
	mock.ExpectQuery(publicationQueryPattern).
		WithArgs(binding.OrgID, digest).
		WillReturnRows(publicationRows().AddRow(
			digest, legacyPublicationBindingVersion, binding.OrgID, "7",
			binding.CatalogSnapshotDigest, "2", "3", binding.LayoutValidationViewDigest,
			nil, binding.LayoutActivationSetDigest, "11",
			binding.LayoutRoutingSecurityViewDigest,
		))

	publication, err := authority.ReadActive(context.Background(), binding.OrgID)
	if err != nil {
		t.Fatal(err)
	}
	if publication.Binding != binding {
		t.Fatalf("binding = %#v, want %#v", publication.Binding, binding)
	}
	assertReadExpectations(t, mock)
}

func TestReadActiveDistinguishesNoSelectionFromDanglingPointer(t *testing.T) {
	for _, test := range []struct {
		name        string
		activeRows  *sqlmock.Rows
		expectExact bool
		want        error
	}{
		{
			name:       "no active selection",
			activeRows: activeRows().AddRow("org-1", false, nil, nil, "4"),
			want:       ErrNotFound,
		},
		{
			name:        "dangling active pointer",
			activeRows:  activeRows().AddRow("org-1", true, testDigest("8"), "018f0000-0000-7000-9000-000000000001", "4"),
			expectExact: true,
			want:        ErrCorrupt,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			authority, mock, closeDB := newReadTestAuthority(t)
			defer closeDB()
			mock.ExpectQuery(activeQueryPattern).WithArgs("org-1").WillReturnRows(test.activeRows)
			if test.expectExact {
				mock.ExpectQuery(publicationQueryPattern).
					WithArgs("org-1", testDigest("8")).
					WillReturnRows(publicationRows())
			}
			_, err := authority.ReadActive(context.Background(), "org-1")
			if !errors.Is(err, test.want) {
				t.Fatalf("error = %v, want %v", err, test.want)
			}
			assertReadExpectations(t, mock)
		})
	}
}

func TestReadCommandRecomputesRequestDigestAndPreservesPriorPresence(t *testing.T) {
	authority, mock, closeDB := newReadTestAuthority(t)
	defer closeDB()

	commandID := "018f0000-0000-7000-9000-000000000001"
	candidate := testDigest("5")
	prior := Reference{Present: true, Digest: testDigest("4")}
	requestDigest := publicationCommandDigest("runtime", prior, candidate)
	mock.ExpectQuery(commandQueryPattern).
		WithArgs("org-1", commandID).
		WillReturnRows(commandRows().AddRow(
			"org-1", commandID, requestDigest, "runtime", candidate,
			true, prior.Digest, string(CommandReserved),
		))

	command, err := authority.ReadCommand(context.Background(), "org-1", commandID)
	if err != nil {
		t.Fatal(err)
	}
	if command.ExpectedPrior != prior || command.RequestDigest != requestDigest || command.State != CommandReserved {
		t.Fatalf("command = %#v", command)
	}
	assertReadExpectations(t, mock)
}

func TestReadCommandRejectsDigestThatDoesNotBindDurableRequest(t *testing.T) {
	authority, mock, closeDB := newReadTestAuthority(t)
	defer closeDB()

	commandID := "018f0000-0000-7000-9000-000000000001"
	mock.ExpectQuery(commandQueryPattern).
		WithArgs("org-1", commandID).
		WillReturnRows(commandRows().AddRow(
			"org-1", commandID, testDigest("9"), "runtime", testDigest("5"),
			false, nil, string(CommandCompleted),
		))

	_, err := authority.ReadCommand(context.Background(), "org-1", commandID)
	if !errors.Is(err, ErrCorrupt) {
		t.Fatalf("error = %v, want ErrCorrupt", err)
	}
	assertReadExpectations(t, mock)
}

func TestReadOutcomeRecomputesDigestAndPreservesAbsentPrior(t *testing.T) {
	authority, mock, closeDB := newReadTestAuthority(t)
	defer closeDB()

	outcome := readTestSelectedOutcome(t)
	mock.ExpectQuery(outcomeQueryPattern).
		WithArgs(outcome.OrgID, outcome.CommandID).
		WillReturnRows(outcomeRows().AddRow(
			outcome.OrgID, outcome.CommandID, outcome.CandidatePublicationDigest,
			string(outcome.Result), false, nil, true, outcome.ResultingActive.Digest,
			outcome.OutcomeDigest,
		))

	got, err := authority.ReadOutcome(context.Background(), outcome.OrgID, outcome.CommandID)
	if err != nil {
		t.Fatal(err)
	}
	if got != outcome {
		t.Fatalf("outcome = %#v, want %#v", got, outcome)
	}
	assertReadExpectations(t, mock)
}

func TestReadOutcomeRejectsInconsistentReferenceOrDigest(t *testing.T) {
	outcome := readTestSelectedOutcome(t)
	for _, test := range []struct {
		name            string
		observedPresent bool
		observedDigest  any
		outcomeDigest   string
	}{
		{
			name:            "nullable reference",
			observedPresent: false,
			observedDigest:  testDigest("9"),
			outcomeDigest:   outcome.OutcomeDigest,
		},
		{
			name:            "tampered outcome digest",
			observedPresent: false,
			observedDigest:  nil,
			outcomeDigest:   testDigest("9"),
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			authority, mock, closeDB := newReadTestAuthority(t)
			defer closeDB()
			mock.ExpectQuery(outcomeQueryPattern).
				WithArgs(outcome.OrgID, outcome.CommandID).
				WillReturnRows(outcomeRows().AddRow(
					outcome.OrgID, outcome.CommandID, outcome.CandidatePublicationDigest,
					string(outcome.Result), test.observedPresent, test.observedDigest, true,
					outcome.ResultingActive.Digest, test.outcomeDigest,
				))
			_, err := authority.ReadOutcome(context.Background(), outcome.OrgID, outcome.CommandID)
			if !errors.Is(err, ErrCorrupt) {
				t.Fatalf("error = %v, want ErrCorrupt", err)
			}
			assertReadExpectations(t, mock)
		})
	}
}

func TestReadHistoryUsesUnsignedCursorAndValidatesEveryOutcome(t *testing.T) {
	authority, mock, closeDB := newReadTestAuthority(t)
	defer closeDB()

	after := uint64(math.MaxInt64) + 1
	outcome := readTestSelectedOutcome(t)
	sequence := after + 1
	mock.ExpectQuery(historyQueryPattern).
		WithArgs("org-1", strconv.FormatUint(after, 10), 2).
		WillReturnRows(historyRows().AddRow(
			strconv.FormatUint(sequence, 10), outcome.OrgID, outcome.CommandID,
			outcome.CandidatePublicationDigest, string(outcome.Result), false, nil,
			true, outcome.ResultingActive.Digest, outcome.OutcomeDigest,
		))

	entries, err := authority.ReadHistory(context.Background(), "org-1", after, 2)
	if err != nil {
		t.Fatal(err)
	}
	if len(entries) != 1 || entries[0].Sequence != sequence || entries[0].Outcome != outcome {
		t.Fatalf("entries = %#v", entries)
	}
	assertReadExpectations(t, mock)
}

func TestReadHistoryRejectsNonAscendingOrTamperedRows(t *testing.T) {
	for _, test := range []struct {
		name string
		rows *sqlmock.Rows
	}{
		{
			name: "non-ascending",
			rows: historyRows().
				AddRow("8", "org-1", "018f0000-0000-7000-9000-000000000001", testDigest("5"), string(ResultSelected), false, nil, true, testDigest("5"), mustOutcomeDigest(t, Outcome{
					OrgID: "org-1", CommandID: "018f0000-0000-7000-9000-000000000001", CandidatePublicationDigest: testDigest("5"), Result: ResultSelected, ResultingActive: Reference{Present: true, Digest: testDigest("5")},
				})).
				AddRow("7", "org-1", "018f0000-0000-7000-9000-000000000002", testDigest("6"), string(ResultSelected), false, nil, true, testDigest("6"), mustOutcomeDigest(t, Outcome{
					OrgID: "org-1", CommandID: "018f0000-0000-7000-9000-000000000002", CandidatePublicationDigest: testDigest("6"), Result: ResultSelected, ResultingActive: Reference{Present: true, Digest: testDigest("6")},
				})),
		},
		{
			name: "tampered digest",
			rows: historyRows().AddRow(
				"8", "org-1", "018f0000-0000-7000-9000-000000000001",
				testDigest("5"), string(ResultSelected), false, nil, true,
				testDigest("5"), testDigest("9"),
			),
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			authority, mock, closeDB := newReadTestAuthority(t)
			defer closeDB()
			mock.ExpectQuery(historyQueryPattern).WithArgs("org-1", "6", 10).WillReturnRows(test.rows)
			_, err := authority.ReadHistory(context.Background(), "org-1", 6, 10)
			if !errors.Is(err, ErrCorrupt) {
				t.Fatalf("error = %v, want ErrCorrupt", err)
			}
			assertReadExpectations(t, mock)
		})
	}
}

func TestReadHistoryRejectsOutOfContractLimitBeforeQuery(t *testing.T) {
	authority, mock, closeDB := newReadTestAuthority(t)
	defer closeDB()
	for _, limit := range []int{0, 1001} {
		if _, err := authority.ReadHistory(context.Background(), "org-1", 0, limit); err == nil {
			t.Fatalf("limit %d accepted", limit)
		}
	}
	assertReadExpectations(t, mock)
}

func TestReadHelpersAppendForUpdateForMutationCallers(t *testing.T) {
	authority, mock, closeDB := newReadTestAuthority(t)
	defer closeDB()

	commandID := "018f0000-0000-7000-9000-000000000001"
	candidate := testDigest("5")
	requestDigest := publicationCommandDigest("runtime", Reference{}, candidate)
	mock.ExpectQuery(commandQueryPattern+regexp.QuoteMeta(" FOR UPDATE")).
		WithArgs("org-1", commandID).
		WillReturnRows(commandRows().AddRow(
			"org-1", commandID, requestDigest, "runtime", candidate,
			false, nil, string(CommandReserved),
		))

	if _, err := readCommandDB(context.Background(), authority.db, "org-1", commandID, true); err != nil {
		t.Fatal(err)
	}
	assertReadExpectations(t, mock)
}

func newReadTestAuthority(t *testing.T) (*Authority, sqlmock.Sqlmock, func()) {
	t.Helper()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	authority, err := New(db)
	if err != nil {
		db.Close()
		t.Fatal(err)
	}
	return authority, mock, func() {
		_ = db.Close()
	}
}

func readTestBinding() catalogpublicationv1.Binding {
	return catalogpublicationv1.Binding{
		OrgID:                           "org-1",
		CatalogRevision:                 7,
		CatalogSnapshotDigest:           testDigest("1"),
		SchemaGeneration:                2,
		EngineGeneration:                3,
		LayoutValidationViewDigest:      testDigest("2"),
		LayoutActivationSetDigest:       testDigest("3"),
		SecurityEpoch:                   11,
		LayoutRoutingSecurityViewDigest: testDigest("4"),
	}
}

func readTestSelectedOutcome(t *testing.T) Outcome {
	t.Helper()
	outcome := Outcome{
		OrgID:                      "org-1",
		CommandID:                  "018f0000-0000-7000-9000-000000000001",
		CandidatePublicationDigest: testDigest("5"),
		Result:                     ResultSelected,
		ObservedPrior:              Reference{},
		ResultingActive:            Reference{Present: true, Digest: testDigest("5")},
	}
	outcome.OutcomeDigest = mustOutcomeDigest(t, outcome)
	return outcome
}

func mustBindingDigest(t *testing.T, binding catalogpublicationv1.Binding) string {
	t.Helper()
	digest, err := catalogpublicationv1.BindingDigest(binding)
	if err != nil {
		t.Fatal(err)
	}
	return digest
}

func mustOutcomeDigest(t *testing.T, outcome Outcome) string {
	t.Helper()
	digest, err := canonicalOutcomeDigest(outcome)
	if err != nil {
		t.Fatal(err)
	}
	return digest
}

func publicationRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{
		"PublicationDigest", "BindingVersion", "OrgId", "CatalogRevision",
		"CatalogSnapshotDigest", "SchemaGeneration", "EngineGeneration",
		"LayoutValidationViewDigest", "LayoutBundleOwnerVerificationReceiptSetDigest",
		"LayoutActivationSetDigest", "SecurityEpoch", "LayoutRoutingSecurityViewDigest",
	})
}

func activeRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{
		"OrgId", "ActivePublicationPresent", "ActivePublicationDigest",
		"SelectedByCommandId", "HistorySequence",
	})
}

func commandRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{
		"OrgId", "CommandId", "RequestDigest", "RuntimeAudience",
		"CandidatePublicationDigest", "ExpectedPriorPresent", "ExpectedPriorDigest", "State",
	})
}

func outcomeRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{
		"OrgId", "CommandId", "CandidatePublicationDigest", "Result",
		"ObservedPriorPresent", "ObservedPriorDigest", "ResultingActivePresent",
		"ResultingActiveDigest", "OutcomeDigest",
	})
}

func historyRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{
		"SequenceNumber", "OrgId", "CommandId", "CandidatePublicationDigest", "Result",
		"ObservedPriorPresent", "ObservedPriorDigest", "ResultingActivePresent",
		"ResultingActiveDigest", "OutcomeDigest",
	})
}

func assertReadExpectations(t *testing.T, mock sqlmock.Sqlmock) {
	t.Helper()
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}