package catalogapp

import (
	"context"
	"encoding/json"
	"errors"
	"testing"

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

type servingTestProjectionWriter struct{}

func (servingTestProjectionWriter) ActivateProjection(context.Context, uint64, RuntimeSecurityProjection) error {
	return nil
}

func TestProjectionCannotBypassServingArtifactCommitment(t *testing.T) {
	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)
	}
	operationID := "01900000-0000-7000-8000-000000000001"
	mock.ExpectQuery(`(?s)SELECT OperationId, Kind, AppInstallationId.*FROM _System_CatalogAppOperation WHERE OperationId = \?`).
		WithArgs(operationID).
		WillReturnRows(sqlmock.NewRows([]string{
			"OperationId", "Kind", "AppInstallationId", "OrgId", "AppId", "State", "Step",
			"TargetReleaseDigest", "ExpectedCatalogRevision", "ExpectedSecurityEpoch", "ExpectedProjectionDigest",
			"AcknowledgedSecurityEpoch", "AcknowledgedProjectionDigest", "LastError",
		}).AddRow(operationID, OperationPrepare, "01900000-0000-7000-8000-000000000002", "org-test", "io.example.app",
			StatePreparing, "CANDIDATE_COMMITTED", testDigest("release"), "7", "11", testDigest("projection"), nil, nil, ""))
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), operationID, servingTestProjectionWriter{}); !errors.Is(err, ErrNotReady) {
		t.Fatalf("projection without serving commitment = %v, want not ready", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestCommitServingArtifactsCommitsCompleteBindingSet(t *testing.T) {
	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)
	}
	operationID := "01900000-0000-7000-8000-000000000001"
	installationID := "01900000-0000-7000-8000-000000000002"
	appID := "io.example.app"
	releaseDigest := testDigest("release")
	release := testRegisteredReleaseVersionForApp(t, appID, "1.0.0", "Subject")
	capability := release.Package.Manifest.ProvidedCapabilities[0]
	requiredCapabilitiesJSON, err := json.Marshal(capability.RequiredRuntimeCapabilities)
	if err != nil {
		t.Fatal(err)
	}
	selection := servingSelection{
		Schema: "nexuscrm.catalog-serving-selection/v1",
		Handlers: []servingHandlerSelection{{
			OrgID: "org-test", AppInstallationID: installationID, AppReleaseDigest: releaseDigest,
			CapabilityID: capability.CapabilityID, HandlerRef: capability.HandlerRef, WorkloadID: "workload-1",
			ServiceReleaseDigest: testDigest("service"), OwnershipEpoch: 1,
			CompatibilityReceiptDigest: testDigest("receipt"),
		}},
	}
	if release.Package.Manifest.FrontendCompatibility.Mode == appauthority.FrontendModeRequired {
		selection.Frontend = &servingFrontendSelection{
			OrgID: "org-test", AppInstallationID: installationID, AppReleaseDigest: releaseDigest,
			FrontendReleaseDigest: testDigest("frontend"), CompatibilityReceiptDigest: testDigest("frontend-receipt"),
		}
	}
	encoded, err := json.Marshal(selection)
	if err != nil {
		t.Fatal(err)
	}
	wantDigest := digestParts(string(encoded))

	expectServingCommitOperation(t, mock, operationID, installationID, appID, releaseDigest, "CANDIDATE_COMMITTED")
	mock.ExpectQuery(`(?s)SELECT SelectedHandlerCapabilities, WorkloadHandlerAdmissions FROM _System_CatalogAppOperation.*FOR UPDATE`).
		WithArgs(operationID).
		WillReturnRows(sqlmock.NewRows([]string{"SelectedHandlerCapabilities", "WorkloadHandlerAdmissions"}).AddRow(nil, nil))
	mock.ExpectQuery(`(?s)SELECT CapabilityId, RequiredRuntimeCapabilities FROM _System_CatalogAppReleaseCapability.*ORDER BY CapabilityId$`).
		WithArgs(appID, releaseDigest, appauthority.CapabilityKindServiceHandler).
		WillReturnRows(sqlmock.NewRows([]string{"CapabilityId", "RequiredRuntimeCapabilities"}).AddRow(capability.CapabilityID, requiredCapabilitiesJSON))
	mock.ExpectQuery(`(?s)SELECT CapabilityId, OrgId, AppInstallationId, AppReleaseDigest FROM _System_CatalogHandlerBinding.*State = 'STAGED'.*FOR UPDATE`).
		WithArgs(operationID).WillReturnRows(sqlmock.NewRows([]string{"CapabilityId", "OrgId", "AppInstallationId", "AppReleaseDigest"}).
		AddRow(capability.CapabilityID, "org-test", installationID, releaseDigest))
	mock.ExpectQuery(`(?s)SELECT CanonicalManifest FROM _System_CatalogAppRelease.*FOR UPDATE`).
		WithArgs(appID, releaseDigest).WillReturnRows(sqlmock.NewRows([]string{"CanonicalManifest"}).AddRow(release.Package.CanonicalManifest))
	frontendIDs := sqlmock.NewRows([]string{"FrontendBindingId", "OrgId", "AppInstallationId", "AppReleaseDigest"})
	if selection.Frontend != nil {
		frontendIDs.AddRow("frontend-binding", "org-test", installationID, releaseDigest)
	}
	mock.ExpectQuery(`(?s)SELECT FrontendBindingId, OrgId, AppInstallationId, AppReleaseDigest FROM _System_CatalogFrontendBinding.*State = 'STAGED' FOR UPDATE`).
		WithArgs(operationID).WillReturnRows(frontendIDs)
	expectServingSelection(mock, operationID, selection)
	mock.ExpectExec(`(?s)UPDATE _System_CatalogAppOperation.*SET Step = \?,.*WHERE OperationId = \? AND Step = 'CANDIDATE_COMMITTED' AND State = \?`).
		WithArgs(OperationStepServingArtifactsCommitted, operationID, StatePreparing).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`(?s)SELECT SequenceNumber FROM _System_CatalogAppOperationJournal.*ORDER BY SequenceNumber DESC LIMIT 1 FOR UPDATE`).
		WithArgs(operationID).WillReturnRows(sqlmock.NewRows([]string{"SequenceNumber"}).AddRow("1"))
	mock.ExpectExec(`(?s)INSERT INTO _System_CatalogAppOperationJournal.*VALUES \(\?, \?, \?, \?, \?, CURRENT_TIMESTAMP\(6\)\)`).
		WithArgs(operationID, "2", OperationStepServingArtifactsCommitted, string(StatePreparing), wantDigest).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	got, err := authority.CommitServingArtifacts(context.Background(), operationID)
	if err != nil || got != wantDigest {
		t.Fatalf("commit serving artifacts = %q, %v; want %q", got, err, wantDigest)
	}

	expectServingCommitOperation(t, mock, operationID, installationID, appID, releaseDigest, OperationStepServingArtifactsCommitted)
	expectServingReceiptHistory(
		mock,
		operationID,
		wantDigest,
		testDigest("failure-event"),
		testDigest("resume-event"),
	)
	expectServingSelection(mock, operationID, selection)
	mock.ExpectCommit()
	got, err = authority.CommitServingArtifacts(context.Background(), operationID)
	if err != nil || got != wantDigest {
		t.Fatalf("retry serving artifact commit = %q, %v", got, err)
	}

	drifted := selection
	drifted.Handlers = append([]servingHandlerSelection(nil), selection.Handlers...)
	drifted.Handlers[0].CompatibilityReceiptDigest = testDigest("changed")
	expectServingCommitOperation(t, mock, operationID, installationID, appID, releaseDigest, OperationStepServingArtifactsCommitted)
	expectServingReceipt(mock, operationID, wantDigest)
	expectServingSelection(mock, operationID, drifted)
	mock.ExpectRollback()
	if _, err := authority.CommitServingArtifacts(context.Background(), operationID); !errors.Is(err, ErrConflict) {
		t.Fatalf("changed serving selection error = %v, want conflict", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestCommitServingArtifactsUsesDurableHandlerSelection(t *testing.T) {
	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)
	}
	operationID := "01900000-0000-7000-8000-000000000001"
	installationID := "01900000-0000-7000-8000-000000000002"
	appID := "io.example.app"
	releaseDigest := testDigest("release")
	release := testRegisteredReleaseVersionForApp(t, appID, "1.0.0", "Subject")
	capability := release.Package.Manifest.ProvidedCapabilities[0]
	requiredCapabilitiesJSON, err := json.Marshal(capability.RequiredRuntimeCapabilities)
	if err != nil {
		t.Fatal(err)
	}
	workloadCapability := "io.example.workload-only.v1"
	workloadRuntimeCapability := "runtime.identity.session-lifecycle.v1"
	admissionsJSON, err := json.Marshal([]WorkloadHandlerAdmission{{
		CapabilityID: workloadCapability, AuthorizedWorkloadID: "workload-1",
		RequiredRuntimeCapabilities: []string{workloadRuntimeCapability},
	}})
	if err != nil {
		t.Fatal(err)
	}
	selection := servingSelection{
		Schema: "nexuscrm.catalog-serving-selection/v1",
		Handlers: []servingHandlerSelection{{
			OrgID: "org-test", AppInstallationID: installationID, AppReleaseDigest: releaseDigest,
			CapabilityID: capability.CapabilityID, HandlerRef: capability.HandlerRef, WorkloadID: "workload-1",
			ServiceReleaseDigest: testDigest("service"), OwnershipEpoch: 1,
			CompatibilityReceiptDigest: testDigest("receipt"),
		}},
	}
	encoded, err := json.Marshal(selection)
	if err != nil {
		t.Fatal(err)
	}
	wantDigest := digestParts(string(encoded))

	expectServingCommitOperation(t, mock, operationID, installationID, appID, releaseDigest, "CANDIDATE_COMMITTED")
	mock.ExpectQuery(`(?s)SELECT SelectedHandlerCapabilities, WorkloadHandlerAdmissions FROM _System_CatalogAppOperation.*FOR UPDATE`).
		WithArgs(operationID).
		WillReturnRows(sqlmock.NewRows([]string{"SelectedHandlerCapabilities", "WorkloadHandlerAdmissions"}).
			AddRow(`["`+capability.CapabilityID+`"]`, admissionsJSON))
	mock.ExpectQuery(`(?s)SELECT CapabilityId, RequiredRuntimeCapabilities FROM _System_CatalogAppReleaseCapability.*ORDER BY CapabilityId$`).
		WithArgs(appID, releaseDigest, appauthority.CapabilityKindServiceHandler).
		WillReturnRows(sqlmock.NewRows([]string{"CapabilityId", "RequiredRuntimeCapabilities"}).
			AddRow(capability.CapabilityID, requiredCapabilitiesJSON).
			AddRow(workloadCapability, `["`+workloadRuntimeCapability+`"]`))
	mock.ExpectQuery(`(?s)SELECT CandidateAuthorizedWorkloads, CandidateGrantedCapabilities FROM _System_CatalogAppInstallation.*FOR UPDATE`).
		WithArgs(installationID).
		WillReturnRows(sqlmock.NewRows([]string{"CandidateAuthorizedWorkloads", "CandidateGrantedCapabilities"}).
			AddRow(`["workload-1"]`, `["`+capability.RequiredRuntimeCapabilities[0]+`","`+workloadRuntimeCapability+`"]`))
	mock.ExpectQuery(`(?s)SELECT CapabilityId, OrgId, AppInstallationId, AppReleaseDigest FROM _System_CatalogHandlerBinding.*State = 'STAGED'.*FOR UPDATE`).
		WithArgs(operationID).WillReturnRows(sqlmock.NewRows([]string{"CapabilityId", "OrgId", "AppInstallationId", "AppReleaseDigest"}).
		AddRow(capability.CapabilityID, "org-test", installationID, releaseDigest))
	mock.ExpectQuery(`(?s)SELECT CanonicalManifest FROM _System_CatalogAppRelease.*FOR UPDATE`).
		WithArgs(appID, releaseDigest).WillReturnRows(sqlmock.NewRows([]string{"CanonicalManifest"}).AddRow(release.Package.CanonicalManifest))
	mock.ExpectQuery(`(?s)SELECT FrontendBindingId, OrgId, AppInstallationId, AppReleaseDigest FROM _System_CatalogFrontendBinding.*State = 'STAGED' FOR UPDATE`).
		WithArgs(operationID).WillReturnRows(sqlmock.NewRows([]string{"FrontendBindingId", "OrgId", "AppInstallationId", "AppReleaseDigest"}))
	expectServingSelection(mock, operationID, selection)
	mock.ExpectExec(`(?s)UPDATE _System_CatalogAppOperation.*SET Step = \?,.*WHERE OperationId = \? AND Step = 'CANDIDATE_COMMITTED' AND State = \?`).
		WithArgs(OperationStepServingArtifactsCommitted, operationID, StatePreparing).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`(?s)SELECT SequenceNumber FROM _System_CatalogAppOperationJournal.*ORDER BY SequenceNumber DESC LIMIT 1 FOR UPDATE`).
		WithArgs(operationID).WillReturnRows(sqlmock.NewRows([]string{"SequenceNumber"}).AddRow("1"))
	mock.ExpectExec(`(?s)INSERT INTO _System_CatalogAppOperationJournal.*VALUES \(\?, \?, \?, \?, \?, CURRENT_TIMESTAMP\(6\)\)`).
		WithArgs(operationID, "2", OperationStepServingArtifactsCommitted, string(StatePreparing), wantDigest).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	got, err := authority.CommitServingArtifacts(context.Background(), operationID)
	if err != nil || got != wantDigest {
		t.Fatalf("commit selected serving artifacts = %q, %v; want %q", got, err, wantDigest)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestCommitServingArtifactsRejectsIncompleteDurableHandlerSelection(t *testing.T) {
	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)
	}
	operationID := "01900000-0000-7000-8000-000000000001"
	installationID := "01900000-0000-7000-8000-000000000002"
	appID := "io.example.app"
	releaseDigest := testDigest("release")
	selected := []string{
		"io.example.issues.read.v1",
		"io.example.issues.write.v1",
	}
	selectedJSON, err := json.Marshal(selected)
	if err != nil {
		t.Fatal(err)
	}
	workloadCapability := "io.example.issues.workload.v1"
	workloadRuntimeCapability := "runtime.identity.session-lifecycle.v1"
	admissionsJSON, err := json.Marshal([]WorkloadHandlerAdmission{{
		CapabilityID: workloadCapability, AuthorizedWorkloadID: "workload-1",
		RequiredRuntimeCapabilities: []string{workloadRuntimeCapability},
	}})
	if err != nil {
		t.Fatal(err)
	}

	expectServingCommitOperation(t, mock, operationID, installationID, appID, releaseDigest, "CANDIDATE_COMMITTED")
	mock.ExpectQuery(`(?s)SELECT SelectedHandlerCapabilities, WorkloadHandlerAdmissions FROM _System_CatalogAppOperation.*FOR UPDATE`).
		WithArgs(operationID).
		WillReturnRows(sqlmock.NewRows([]string{"SelectedHandlerCapabilities", "WorkloadHandlerAdmissions"}).
			AddRow(selectedJSON, admissionsJSON))
	mock.ExpectQuery(`(?s)SELECT CapabilityId, RequiredRuntimeCapabilities FROM _System_CatalogAppReleaseCapability.*ORDER BY CapabilityId$`).
		WithArgs(appID, releaseDigest, appauthority.CapabilityKindServiceHandler).
		WillReturnRows(sqlmock.NewRows([]string{"CapabilityId", "RequiredRuntimeCapabilities"}).
			AddRow(selected[0], `["runtime.records.read.v1"]`).
			AddRow(selected[1], `["runtime.records.write.v1"]`).
			AddRow(workloadCapability, `["`+workloadRuntimeCapability+`"]`))
	mock.ExpectQuery(`(?s)SELECT CandidateAuthorizedWorkloads, CandidateGrantedCapabilities FROM _System_CatalogAppInstallation.*FOR UPDATE`).
		WithArgs(installationID).
		WillReturnRows(sqlmock.NewRows([]string{"CandidateAuthorizedWorkloads", "CandidateGrantedCapabilities"}).
			AddRow(`["workload-1"]`, `["runtime.identity.session-lifecycle.v1","runtime.records.read.v1","runtime.records.write.v1"]`))
	mock.ExpectQuery(`(?s)SELECT CapabilityId, OrgId, AppInstallationId, AppReleaseDigest FROM _System_CatalogHandlerBinding.*State = 'STAGED'.*FOR UPDATE`).
		WithArgs(operationID).WillReturnRows(sqlmock.NewRows([]string{"CapabilityId", "OrgId", "AppInstallationId", "AppReleaseDigest"}).
		AddRow(selected[0], "org-test", installationID, releaseDigest))
	mock.ExpectRollback()
	if _, err := authority.CommitServingArtifacts(
		context.Background(), operationID,
	); !errors.Is(err, ErrNotReady) {
		t.Fatalf("incomplete selected serving artifacts = %v, want not ready", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestLegacyProjectionAcknowledgementBackfillsServingCommitment(t *testing.T) {
	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)
	}
	operationID := "01900000-0000-7000-8000-000000000001"
	installationID := "01900000-0000-7000-8000-000000000002"
	appID := "io.example.app"
	releaseDigest := testDigest("release")
	release := testRegisteredReleaseVersionForApp(t, appID, "1.0.0", "Subject")
	capability := release.Package.Manifest.ProvidedCapabilities[0]
	selection := servingSelection{Schema: "nexuscrm.catalog-serving-selection/v1", Handlers: []servingHandlerSelection{{
		OrgID: "org-test", AppInstallationID: installationID, AppReleaseDigest: releaseDigest,
		CapabilityID: capability.CapabilityID, HandlerRef: capability.HandlerRef, WorkloadID: "workload-1",
		ServiceReleaseDigest: testDigest("service"), OwnershipEpoch: 1,
		CompatibilityReceiptDigest: testDigest("receipt"),
	}}}
	encoded, err := json.Marshal(selection)
	if err != nil {
		t.Fatal(err)
	}
	wantDigest := digestParts(string(encoded))

	expectServingCommitOperation(t, mock, operationID, installationID, appID, releaseDigest, "PROJECTION_ACKNOWLEDGED")
	expectMissingServingReceipt(mock, operationID)
	expectServingCoverage(mock, operationID, installationID, appID, releaseDigest, release, selection)
	expectServingSelection(mock, operationID, selection)
	mock.ExpectQuery(`(?s)SELECT SequenceNumber FROM _System_CatalogAppOperationJournal.*ORDER BY SequenceNumber DESC LIMIT 1 FOR UPDATE`).
		WithArgs(operationID).WillReturnRows(sqlmock.NewRows([]string{"SequenceNumber"}).AddRow("2"))
	mock.ExpectExec(`(?s)INSERT INTO _System_CatalogAppOperationJournal.*VALUES \(\?, \?, \?, \?, \?, CURRENT_TIMESTAMP\(6\)\)`).
		WithArgs(operationID, "3", operationJournalStepLegacyServingArtifactsBackfilled, string(StatePreparing), wantDigest).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	got, err := authority.CommitServingArtifacts(context.Background(), operationID)
	if err != nil || got != wantDigest {
		t.Fatalf("legacy serving artifact backfill = %q, %v; want %q", got, err, wantDigest)
	}

	expectServingCommitOperation(t, mock, operationID, installationID, appID, releaseDigest, "PROJECTION_ACKNOWLEDGED")
	expectServingReceipt(mock, operationID, wantDigest)
	expectServingSelection(mock, operationID, selection)
	mock.ExpectCommit()
	got, err = authority.CommitServingArtifacts(context.Background(), operationID)
	if err != nil || got != wantDigest {
		t.Fatalf("legacy serving artifact backfill retry = %q, %v", got, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func expectServingOperation(t *testing.T, mock sqlmock.Sqlmock, operationID, installationID, appID, releaseDigest, step string) {
	t.Helper()
	mock.ExpectBegin()
	mock.ExpectQuery(`(?s)SELECT OperationId, Kind, AppInstallationId.*FROM _System_CatalogAppOperation WHERE OperationId = \? FOR UPDATE`).
		WithArgs(operationID).
		WillReturnRows(sqlmock.NewRows([]string{
			"OperationId", "Kind", "AppInstallationId", "OrgId", "AppId", "State", "Step",
			"TargetReleaseDigest", "ExpectedCatalogRevision", "ExpectedSecurityEpoch", "ExpectedProjectionDigest",
			"AcknowledgedSecurityEpoch", "AcknowledgedProjectionDigest", "LastError",
		}).AddRow(operationID, OperationPrepare, installationID, "org-test", appID,
			StatePreparing, step, releaseDigest, "7", "11", testDigest("projection"), nil, nil, ""))

}

func expectServingReceipt(mock sqlmock.Sqlmock, operationID, receipt string) {
	expectServingReceiptHistory(mock, operationID, receipt)
}

func expectServingReceiptHistory(
	mock sqlmock.Sqlmock,
	operationID string,
	evidence ...string,
) {
	rows := sqlmock.NewRows([]string{"EvidenceDigest"})
	for _, digest := range evidence {
		rows.AddRow(digest)
	}
	mock.ExpectQuery(`(?s)SELECT EvidenceDigest FROM _System_CatalogAppOperationJournal.*WHERE OperationId = \? AND Step IN \(\?, \?\) ORDER BY SequenceNumber`).
		WithArgs(operationID, OperationStepServingArtifactsCommitted, operationJournalStepLegacyServingArtifactsBackfilled).
		WillReturnRows(rows)
}

func expectMissingServingReceipt(mock sqlmock.Sqlmock, operationID string) {
	mock.ExpectQuery(`(?s)SELECT EvidenceDigest FROM _System_CatalogAppOperationJournal.*WHERE OperationId = \? AND Step IN \(\?, \?\) ORDER BY SequenceNumber`).
		WithArgs(operationID, OperationStepServingArtifactsCommitted, operationJournalStepLegacyServingArtifactsBackfilled).
		WillReturnRows(sqlmock.NewRows([]string{"EvidenceDigest"}))
}

func expectServingCoverage(mock sqlmock.Sqlmock, operationID, installationID, appID, releaseDigest string, release RegisteredRelease, selection servingSelection) {
	capability := release.Package.Manifest.ProvidedCapabilities[0]
	requiredCapabilitiesJSON, _ := json.Marshal(capability.RequiredRuntimeCapabilities)
	mock.ExpectQuery(`(?s)SELECT SelectedHandlerCapabilities, WorkloadHandlerAdmissions FROM _System_CatalogAppOperation.*FOR UPDATE`).
		WithArgs(operationID).
		WillReturnRows(sqlmock.NewRows([]string{"SelectedHandlerCapabilities", "WorkloadHandlerAdmissions"}).AddRow(nil, nil))
	mock.ExpectQuery(`(?s)SELECT CapabilityId, RequiredRuntimeCapabilities FROM _System_CatalogAppReleaseCapability.*ORDER BY CapabilityId$`).
		WithArgs(appID, releaseDigest, appauthority.CapabilityKindServiceHandler).
		WillReturnRows(sqlmock.NewRows([]string{"CapabilityId", "RequiredRuntimeCapabilities"}).AddRow(capability.CapabilityID, requiredCapabilitiesJSON))
	mock.ExpectQuery(`(?s)SELECT CapabilityId, OrgId, AppInstallationId, AppReleaseDigest FROM _System_CatalogHandlerBinding.*State = 'STAGED'.*FOR UPDATE`).
		WithArgs(operationID).WillReturnRows(sqlmock.NewRows([]string{"CapabilityId", "OrgId", "AppInstallationId", "AppReleaseDigest"}).
		AddRow(capability.CapabilityID, "org-test", installationID, releaseDigest))
	mock.ExpectQuery(`(?s)SELECT CanonicalManifest FROM _System_CatalogAppRelease.*FOR UPDATE`).
		WithArgs(appID, releaseDigest).WillReturnRows(sqlmock.NewRows([]string{"CanonicalManifest"}).AddRow(release.Package.CanonicalManifest))
	frontendIDs := sqlmock.NewRows([]string{"FrontendBindingId", "OrgId", "AppInstallationId", "AppReleaseDigest"})
	if selection.Frontend != nil {
		frontendIDs.AddRow("frontend-binding", "org-test", installationID, releaseDigest)
	}
	mock.ExpectQuery(`(?s)SELECT FrontendBindingId, OrgId, AppInstallationId, AppReleaseDigest FROM _System_CatalogFrontendBinding.*State = 'STAGED' FOR UPDATE`).
		WithArgs(operationID).WillReturnRows(frontendIDs)
}

func expectServingSelection(mock sqlmock.Sqlmock, operationID string, selection servingSelection) {
	handlers := sqlmock.NewRows([]string{
		"OrgId", "AppInstallationId", "AppReleaseDigest", "CapabilityId", "HandlerRef", "WorkloadId", "ServiceReleaseDigest", "OwnershipEpoch", "CompatibilityReceiptDigest",
	})
	for _, binding := range selection.Handlers {
		handlers.AddRow(binding.OrgID, binding.AppInstallationID, binding.AppReleaseDigest,
			binding.CapabilityID, binding.HandlerRef, binding.WorkloadID, binding.ServiceReleaseDigest,
			epochText(binding.OwnershipEpoch), binding.CompatibilityReceiptDigest)
	}
	mock.ExpectQuery(`(?s)SELECT OrgId, AppInstallationId, AppReleaseDigest, CapabilityId, HandlerRef, WorkloadId, ServiceReleaseDigest,.*FROM _System_CatalogHandlerBinding.*ORDER BY CapabilityId`).
		WithArgs(operationID).WillReturnRows(handlers)
	frontend := sqlmock.NewRows([]string{"OrgId", "AppInstallationId", "AppReleaseDigest", "FrontendReleaseDigest", "CompatibilityReceiptDigest"})
	if selection.Frontend != nil {
		frontend.AddRow(selection.Frontend.OrgID, selection.Frontend.AppInstallationID, selection.Frontend.AppReleaseDigest,
			selection.Frontend.FrontendReleaseDigest, selection.Frontend.CompatibilityReceiptDigest)
	}
	mock.ExpectQuery(`(?s)SELECT OrgId, AppInstallationId, AppReleaseDigest, FrontendReleaseDigest, CompatibilityReceiptDigest.*FROM _System_CatalogFrontendBinding.*ORDER BY FrontendBindingId`).
		WithArgs(operationID).WillReturnRows(frontend)
}

func expectServingCommitOperation(t *testing.T, mock sqlmock.Sqlmock, operationID, installationID, appID, releaseDigest, step string) {
	t.Helper()
	expectServingOperation(t, mock, operationID, installationID, appID, releaseDigest, step)
	if step == "CANDIDATE_COMMITTED" {
		expectNoOperationConsumerGrants(mock, operationID)
	}
}