package catalogpublicationv2store

import (
	"context"
	"errors"
	"reflect"
	"regexp"
	"strings"
	"testing"

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

func TestPutAndReadExactRoundTripAllTenV2Fields(t *testing.T) {
	binding := testBinding()
	digest, err := catalogpublicationv2.BindingDigest(binding)
	if err != nil {
		t.Fatal(err)
	}
	table, mock, closeDB := newVerifiedMockTable(t, shadowTableIdentity)
	defer closeDB()

	mock.ExpectBegin()
	expectBindingMissing(mock, shadowTableName, binding.OrgID, digest, true)
	mock.ExpectExec(regexp.QuoteMeta(insertStatement(shadowTableName))).
		WithArgs(
			digest,
			binding.OrgID,
			"41",
			binding.CatalogSnapshotDigest,
			"43",
			"47",
			binding.LayoutValidationViewDigest,
			binding.LayoutBundleOwnerVerificationReceiptSetDigest,
			binding.LayoutActivationSetDigest,
			"53",
			binding.LayoutRoutingSecurityViewDigest,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	expectBindingRow(mock, shadowTableName, binding, digest, true)
	mock.ExpectCommit()

	gotDigest, err := PutExact(context.Background(), table, binding)
	if err != nil {
		t.Fatal(err)
	}
	if gotDigest != digest {
		t.Fatalf("PutExact digest = %q, want %q", gotDigest, digest)
	}

	mock.ExpectBegin()
	expectBindingRow(mock, shadowTableName, binding, digest, false)
	mock.ExpectCommit()
	got, err := ReadExact(context.Background(), table, binding.OrgID, digest)
	if err != nil {
		t.Fatal(err)
	}
	if got != binding {
		t.Fatalf("ReadExact = %#v, want %#v", got, binding)
	}
	assertStoreExpectations(t, mock)
}

func TestPutExactReplayIsImmutableAndDoesNotInsert(t *testing.T) {
	binding := testBinding()
	digest, err := catalogpublicationv2.BindingDigest(binding)
	if err != nil {
		t.Fatal(err)
	}
	table, mock, closeDB := newVerifiedMockTable(t, liveTableIdentity)
	defer closeDB()
	mock.ExpectBegin()
	expectBindingRow(mock, liveTableName, binding, digest, true)
	mock.ExpectCommit()

	got, err := PutExact(context.Background(), table, binding)
	if err != nil || got != digest {
		t.Fatalf("PutExact replay = (%q, %v), want (%q, nil)", got, err, digest)
	}
	assertStoreExpectations(t, mock)
}

func TestPutExactRequiresACompleteOwnerReceiptDigest(t *testing.T) {
	table, mock, closeDB := newVerifiedMockTable(t, shadowTableIdentity)
	defer closeDB()
	binding := testBinding()
	binding.LayoutBundleOwnerVerificationReceiptSetDigest = ""

	if digest, err := PutExact(context.Background(), table, binding); err == nil || digest != "" {
		t.Fatalf("PutExact incomplete receipt = (%q, %v), want error", digest, err)
	}
	assertStoreExpectations(t, mock)
}

func TestReadExactRejectsEveryCanonicalDigestMutation(t *testing.T) {
	binding := testBinding()
	digest, err := catalogpublicationv2.BindingDigest(binding)
	if err != nil {
		t.Fatal(err)
	}
	mutations := []struct {
		name   string
		mutate func(*catalogpublicationv2.Binding)
	}{
		{"org", func(value *catalogpublicationv2.Binding) { value.OrgID = "other-org" }},
		{"catalog revision", func(value *catalogpublicationv2.Binding) { value.CatalogRevision++ }},
		{"catalog snapshot", func(value *catalogpublicationv2.Binding) { value.CatalogSnapshotDigest = testSHA256("a") }},
		{"schema generation", func(value *catalogpublicationv2.Binding) { value.SchemaGeneration++ }},
		{"engine generation", func(value *catalogpublicationv2.Binding) { value.EngineGeneration++ }},
		{"layout validation", func(value *catalogpublicationv2.Binding) { value.LayoutValidationViewDigest = testSHA256("b") }},
		{"owner receipt", func(value *catalogpublicationv2.Binding) {
			value.LayoutBundleOwnerVerificationReceiptSetDigest = testSHA256("c")
		}},
		{"layout activation", func(value *catalogpublicationv2.Binding) { value.LayoutActivationSetDigest = testSHA256("d") }},
		{"security epoch", func(value *catalogpublicationv2.Binding) { value.SecurityEpoch++ }},
		{"routing security", func(value *catalogpublicationv2.Binding) { value.LayoutRoutingSecurityViewDigest = testSHA256("e") }},
	}
	for _, mutation := range mutations {
		mutation := mutation
		t.Run(mutation.name, func(t *testing.T) {
			stored := binding
			mutation.mutate(&stored)
			table, mock, closeDB := newVerifiedMockTable(t, liveTableIdentity)
			defer closeDB()
			mock.ExpectBegin()
			expectBindingRowForLookup(mock, liveTableName, stored, binding.OrgID, digest, false)
			mock.ExpectRollback()

			got, err := ReadExact(context.Background(), table, binding.OrgID, digest)
			if !errors.Is(err, ErrCorrupt) || got != (catalogpublicationv2.Binding{}) {
				t.Fatalf("ReadExact mutated row = (%#v, %v), want zero ErrCorrupt", got, err)
			}
			assertStoreExpectations(t, mock)
		})
	}
}

func TestReadExactMapsOnlyAbsenceToNotFound(t *testing.T) {
	binding := testBinding()
	digest, err := catalogpublicationv2.BindingDigest(binding)
	if err != nil {
		t.Fatal(err)
	}
	table, mock, closeDB := newVerifiedMockTable(t, liveTableIdentity)
	defer closeDB()
	mock.ExpectBegin()
	expectBindingMissing(mock, liveTableName, binding.OrgID, digest, false)
	mock.ExpectRollback()

	got, err := ReadExact(context.Background(), table, binding.OrgID, digest)
	if !errors.Is(err, ErrNotFound) || got != (catalogpublicationv2.Binding{}) {
		t.Fatalf("ReadExact absent = (%#v, %v), want zero ErrNotFound", got, err)
	}
	assertStoreExpectations(t, mock)
}

func TestPutAndReadRejectUnverifiedHandles(t *testing.T) {
	binding := testBinding()
	for _, table := range []*VerifiedTable{nil, {}} {
		if digest, err := PutExact(context.Background(), table, binding); !errors.Is(err, ErrSchema) || digest != "" {
			t.Fatalf("PutExact unverified = (%q, %v)", digest, err)
		}
		if got, err := ReadExact(context.Background(), table, binding.OrgID, testSHA256("f")); !errors.Is(err, ErrSchema) || got != (catalogpublicationv2.Binding{}) {
			t.Fatalf("ReadExact unverified = (%#v, %v)", got, err)
		}
	}
}

func TestVerifiedTableSurfaceDoesNotExposeDatabaseOrIdentity(t *testing.T) {
	typeOfTable := reflect.TypeOf(VerifiedTable{})
	if typeOfTable.NumField() != 2 {
		t.Fatalf("VerifiedTable fields = %d, want 2 private capability members", typeOfTable.NumField())
	}
	for index := 0; index < typeOfTable.NumField(); index++ {
		if typeOfTable.Field(index).PkgPath == "" {
			t.Fatalf("VerifiedTable field %s is exported", typeOfTable.Field(index).Name)
		}
	}
}

func testBinding() catalogpublicationv2.Binding {
	return catalogpublicationv2.Binding{
		OrgID:                      "org-conformance",
		CatalogRevision:            41,
		CatalogSnapshotDigest:      testSHA256("1"),
		SchemaGeneration:           43,
		EngineGeneration:           47,
		LayoutValidationViewDigest: testSHA256("2"),
		LayoutBundleOwnerVerificationReceiptSetDigest: testSHA256("5"),
		LayoutActivationSetDigest:                     testSHA256("3"),
		SecurityEpoch:                                 53,
		LayoutRoutingSecurityViewDigest:               testSHA256("4"),
	}
}

func testSHA256(character string) string {
	return "sha256:" + strings.Repeat(character, 64)
}

func newVerifiedMockTable(
	t *testing.T,
	identity *tableIdentity,
) (*VerifiedTable, sqlmock.Sqlmock, func()) {
	t.Helper()
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	if err != nil {
		t.Fatal(err)
	}
	return &VerifiedTable{db: db, identity: identity}, mock, func() {
		_ = db.Close()
	}
}

func selectStatement(tableName string, forUpdate bool) string {
	statement := `SELECT PublicationDigest, OrgId, CatalogRevision,
CatalogSnapshotDigest, SchemaGeneration, EngineGeneration,
LayoutValidationViewDigest, LayoutBundleOwnerVerificationReceiptSetDigest,
LayoutActivationSetDigest, SecurityEpoch, LayoutRoutingSecurityViewDigest
FROM ` + tableName + ` WHERE OrgId = ? AND PublicationDigest = ?`
	if forUpdate {
		statement += " FOR UPDATE"
	}
	return statement
}

func insertStatement(tableName string) string {
	return `INSERT INTO ` + tableName + `
(PublicationDigest, OrgId, CatalogRevision, CatalogSnapshotDigest,
SchemaGeneration, EngineGeneration, LayoutValidationViewDigest,
LayoutBundleOwnerVerificationReceiptSetDigest, LayoutActivationSetDigest,
SecurityEpoch, LayoutRoutingSecurityViewDigest, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`
}

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

func expectBindingMissing(
	mock sqlmock.Sqlmock,
	tableName string,
	orgID string,
	digest string,
	forUpdate bool,
) {
	mock.ExpectQuery(regexp.QuoteMeta(selectStatement(tableName, forUpdate))).
		WithArgs(orgID, digest).
		WillReturnRows(bindingRows())
}

func expectBindingRow(
	mock sqlmock.Sqlmock,
	tableName string,
	binding catalogpublicationv2.Binding,
	digest string,
	forUpdate bool,
) {
	expectBindingRowForLookup(mock, tableName, binding, binding.OrgID, digest, forUpdate)
}

func expectBindingRowForLookup(
	mock sqlmock.Sqlmock,
	tableName string,
	stored catalogpublicationv2.Binding,
	lookupOrgID string,
	digest string,
	forUpdate bool,
) {
	rows := bindingRows().AddRow(
		digest,
		stored.OrgID,
		decimalText(stored.CatalogRevision),
		stored.CatalogSnapshotDigest,
		decimalText(stored.SchemaGeneration),
		decimalText(stored.EngineGeneration),
		stored.LayoutValidationViewDigest,
		stored.LayoutBundleOwnerVerificationReceiptSetDigest,
		stored.LayoutActivationSetDigest,
		decimalText(stored.SecurityEpoch),
		stored.LayoutRoutingSecurityViewDigest,
	)
	mock.ExpectQuery(regexp.QuoteMeta(selectStatement(tableName, forUpdate))).
		WithArgs(lookupOrgID, digest).
		WillReturnRows(rows)
}

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