package catalogapp

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

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/go-sql-driver/mysql"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogadmissionfence"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogwriter"
)

func TestWriterGenerationExhaustionMapsToAppConflict(t *testing.T) {
	if err := catalogWriterError(catalogwriter.ErrLineageExhausted); !errors.Is(
		err, ErrConflict,
	) {
		t.Fatalf("writer generation exhaustion = %v", err)
	}
}

func TestAppOperationIDRejectsCandidateSourceCollision(t *testing.T) {
	const operationID = "018f0000-0000-7000-9000-000000000001"
	for _, test := range []struct {
		name      string
		queryErr  error
		sourceOrg any
		wantErr   error
	}{
		{
			name:     "no source journal",
			queryErr: sql.ErrNoRows,
		},
		{
			name:     "source schema not provisioned yet",
			queryErr: &mysql.MySQLError{Number: 1146, Message: "table does not exist"},
		},
		{
			name:      "foreign-org source journal",
			sourceOrg: "org-foreign",
			wantErr:   ErrConflict,
		},
		{
			name:     "storage failure",
			queryErr: errOperationIdentityStorage,
			wantErr:  errOperationIdentityStorage,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()
			mock.ExpectBegin()
			expectation := mock.ExpectQuery(
				"SELECT OrgId.*_System_CatalogCandidateSource",
			).WithArgs(operationID)
			if test.sourceOrg != nil {
				expectation.WillReturnRows(
					sqlmock.NewRows([]string{"OrgId"}).AddRow(test.sourceOrg),
				)
			} else {
				expectation.WillReturnError(test.queryErr)
			}
			mock.ExpectRollback()

			tx, err := db.BeginTx(context.Background(), nil)
			if err != nil {
				t.Fatal(err)
			}
			got := rejectCandidateSourceOperationIDCollisionLocked(
				context.Background(), tx, operationID, false,
			)
			switch {
			case test.wantErr != nil && !errors.Is(got, test.wantErr):
				t.Fatalf("error = %v, want %v", got, test.wantErr)
			case test.wantErr == nil && got != nil:
				t.Fatalf("error = %v", got)
			}
			_ = tx.Rollback()
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatal(err)
			}
		})
	}
}

var errOperationIdentityStorage = errors.New("storage failure")

func TestBeginOperationRejectsMintedCandidateSourceIDBeforeAppMutation(t *testing.T) {
	const operationID = "018f0000-0000-7000-9000-000000000001"
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	authority, err := New(db, Options{
		NewID: func(time.Time) (string, error) { return operationID, nil },
	})
	if err != nil {
		t.Fatal(err)
	}
	request := BeginOperationRequest{
		Kind:            OperationPrepare,
		OrgID:           "org-1",
		AppID:           "io.example.app",
		IdempotencyKey:  "prepare-1",
		RuntimeAudience: "runtime-1",
	}
	const requestDigest = "sha256:request"

	mock.ExpectBegin()
	expectAppPublicationAdmissionFence(mock)
	mock.ExpectQuery(
		"SELECT OperationId, RequestDigest, InitiationKind FROM _System_CatalogAppOperation",
	).WithArgs(request.OrgID, request.AppID, request.IdempotencyKey).
		WillReturnError(sql.ErrNoRows)
	mock.ExpectQuery(
		"SELECT OrgId.*_System_CatalogCandidateSource",
	).WithArgs(operationID).WillReturnRows(
		sqlmock.NewRows([]string{"OrgId"}).AddRow("org-foreign"),
	)
	mock.ExpectRollback()

	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	_, got := authority.beginOperationTx(
		context.Background(),
		tx,
		request,
		RegisteredRelease{},
		requestDigest,
		initiationDirect,
	)
	if !errors.Is(got, ErrConflict) {
		t.Fatalf("BeginOperation collision = %v", got)
	}
	_ = tx.Rollback()
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestBeginOperationWriterWaitRecheckRejectsCandidateSourceID(t *testing.T) {
	const operationID = "018f0000-0000-7000-9000-000000000001"
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	authority, err := New(db, Options{
		NewID: func(time.Time) (string, error) { return operationID, nil },
	})
	if err != nil {
		t.Fatal(err)
	}
	request := BeginOperationRequest{
		Kind:            OperationPrepare,
		OrgID:           "org-1",
		AppID:           "io.example.app",
		IdempotencyKey:  "prepare-1",
		RuntimeAudience: "runtime-1",
	}

	mock.ExpectBegin()
	expectAppPublicationAdmissionFence(mock)
	mock.ExpectQuery(
		"SELECT OperationId, RequestDigest, InitiationKind FROM _System_CatalogAppOperation",
	).WithArgs(request.OrgID, request.AppID, request.IdempotencyKey).
		WillReturnError(sql.ErrNoRows)
	mock.ExpectQuery(
		"SELECT OrgId.*_System_CatalogCandidateSource",
	).WithArgs(operationID).WillReturnError(sql.ErrNoRows)
	mock.ExpectQuery(
		"SELECT OrgId, RuntimeAudience, CatalogRevision, LastSecurityEpoch, ActiveOperationId, WriterGeneration",
	).WithArgs(request.OrgID).WillReturnRows(sqlmock.NewRows([]string{
		"OrgId",
		"RuntimeAudience",
		"CatalogRevision",
		"LastSecurityEpoch",
		"ActiveOperationId",
		"WriterGeneration",
	}).AddRow(
		request.OrgID,
		request.RuntimeAudience,
		[]byte("0"),
		[]byte("0"),
		nil,
		[]byte("3"),
	))
	mock.ExpectQuery(
		"SELECT OrgId.*_System_CatalogCandidateSource",
	).WithArgs(operationID).WillReturnRows(
		sqlmock.NewRows([]string{"OrgId"}).AddRow("org-foreign"),
	)
	mock.ExpectRollback()

	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	_, got := authority.beginOperationTx(
		context.Background(),
		tx,
		request,
		RegisteredRelease{},
		"sha256:request",
		initiationDirect,
	)
	if !errors.Is(got, ErrConflict) {
		t.Fatalf("writer-wait collision = %v", got)
	}
	_ = tx.Rollback()
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestBeginOperationExistingRetryRejectsDualCandidateSourceJournal(t *testing.T) {
	const operationID = "018f0000-0000-7000-9000-000000000001"
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	authority, err := New(db, Options{})
	if err != nil {
		t.Fatal(err)
	}
	request := BeginOperationRequest{
		Kind:            OperationPrepare,
		OrgID:           "org-1",
		AppID:           "io.example.app",
		IdempotencyKey:  "prepare-1",
		RuntimeAudience: "runtime-1",
	}
	const requestDigest = "sha256:request"

	mock.ExpectBegin()
	expectAppPublicationAdmissionFence(mock)
	mock.ExpectQuery(
		"SELECT OperationId, RequestDigest, InitiationKind FROM _System_CatalogAppOperation",
	).WithArgs(request.OrgID, request.AppID, request.IdempotencyKey).
		WillReturnRows(sqlmock.NewRows([]string{
			"OperationId", "RequestDigest", "InitiationKind",
		}).AddRow(operationID, requestDigest, string(initiationDirect)))
	mock.ExpectQuery(
		"SELECT OrgId.*_System_CatalogCandidateSource",
	).WithArgs(operationID).WillReturnRows(
		sqlmock.NewRows([]string{"OrgId"}).AddRow("org-foreign"),
	)
	mock.ExpectRollback()

	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	_, got := authority.beginOperationTx(
		context.Background(),
		tx,
		request,
		RegisteredRelease{},
		requestDigest,
		initiationDirect,
	)
	if !errors.Is(got, ErrCorrupt) {
		t.Fatalf("existing retry collision = %v", got)
	}
	_ = tx.Rollback()
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func expectAppPublicationAdmissionFence(mock sqlmock.Sqlmock) {
	mock.ExpectQuery(`SELECT FenceKey, FenceGeneration
FROM _System_CatalogPublicationAdmissionFence
WHERE FenceKey = \? FOR UPDATE`).
		WithArgs(catalogadmissionfence.Key).
		WillReturnRows(sqlmock.NewRows([]string{
			"FenceKey",
			"FenceGeneration",
		}).AddRow(catalogadmissionfence.Key, []byte("0")))
}