package catalogapp

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

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

func TestVerifyStoppedSourceReleaseCandidateBindsExactCanonicalSourceTuple(t *testing.T) {
	release := testRegisteredRelease(t)
	input := testStoppedSourceInput(t, release)

	candidate, reference, err := VerifyStoppedSourceReleaseCandidate(input)
	if err != nil {
		t.Fatalf("verify stopped-source candidate: %v", err)
	}
	if candidate.AppID != release.AppID ||
		candidate.PublisherID != release.PublisherID ||
		candidate.SemanticVersion != release.SemanticVersion ||
		candidate.AppReleaseDigest != release.AppReleaseDigest ||
		candidate.MetadataBundleDigest != release.MetadataBundleDigest {
		t.Fatalf("candidate = %#v, want exact release %#v", candidate, release)
	}
	if reference.AppID != input.AppID ||
		reference.AppReleaseDigest != input.ExpectedAppReleaseDigest ||
		reference.PackageContentDigest != input.ExpectedPackageContentDigest ||
		reference.SourceRepository != input.SourceRepository ||
		reference.SourceCommit != input.SourceCommit ||
		reference.ReceiptDigest == "" {
		t.Fatalf("reference does not bind the exact source tuple: %#v", reference)
	}
	wantReference, err := newStoppedSourceReleaseReference(
		input.AppID,
		input.ExpectedAppReleaseDigest,
		input.ExpectedPackageContentDigest,
		input.SourceRepository,
		input.SourceCommit,
	)
	if err != nil {
		t.Fatal(err)
	}
	if reference != wantReference {
		t.Fatalf("reference = %#v, want %#v", reference, wantReference)
	}
}

func TestVerifyStoppedSourceReleaseCandidateRejectsUnprovenInput(t *testing.T) {
	valid := testStoppedSourceInput(t, testRegisteredRelease(t))
	otherRelease := testRegisteredReleaseVersion(t, "1.0.1", "Status")

	malformedComponentRaw, _ := testMalformedSlotImportArtifact(
		t,
		time.Unix(1_800_000_000, 0).UTC(),
	)
	_, malformedComponentPackage, err := releaseartifact.ParsePackage(malformedComponentRaw)
	if err != nil {
		t.Fatal(err)
	}
	malformedComponent := StoppedSourceReleaseInput{
		AppID:                        malformedComponentPackage.Manifest.AppID,
		ExpectedAppReleaseDigest:     malformedComponentPackage.AppReleaseDigest,
		ExpectedPackageContentDigest: exactPackageContentDigest(malformedComponentRaw),
		SourceRepository:             valid.SourceRepository,
		SourceCommit:                 valid.SourceCommit,
		PackageBytes:                 malformedComponentRaw,
		Compatibility: releaseartifact.HostCompatibility{
			Protocols:          []releaseartifact.SupportedProtocol{},
			EngineCapabilities: []string{"runtime.records.read.v1"},
			ComponentKinds: []releaseartifact.SupportedComponentKind{
				{ComponentKind: "metadata.object.v1", ComponentFormatVersion: 1},
				{ComponentKind: layoutbundleslotv1.ComponentKind, ComponentFormatVersion: layoutbundleslotv1.ComponentFormatVersion},
			},
		},
	}

	tests := []struct {
		name  string
		input StoppedSourceReleaseInput
	}{
		{
			name: "non-canonical package bytes",
			input: mutateStoppedSourceInput(valid, func(value *StoppedSourceReleaseInput) {
				value.PackageBytes = append(append([]byte(nil), value.PackageBytes...), '\n')
				value.ExpectedPackageContentDigest = exactPackageContentDigest(value.PackageBytes)
			}),
		},
		{
			name: "wrong AppID",
			input: mutateStoppedSourceInput(valid, func(value *StoppedSourceReleaseInput) {
				value.AppID = "io.example.other"
			}),
		},
		{
			name: "wrong AppReleaseDigest",
			input: mutateStoppedSourceInput(valid, func(value *StoppedSourceReleaseInput) {
				value.ExpectedAppReleaseDigest = otherRelease.AppReleaseDigest
			}),
		},
		{
			name: "wrong package content digest",
			input: mutateStoppedSourceInput(valid, func(value *StoppedSourceReleaseInput) {
				value.ExpectedPackageContentDigest = testDigest("different-package-bytes")
			}),
		},
		{
			name: "non-canonical repository",
			input: mutateStoppedSourceInput(valid, func(value *StoppedSourceReleaseInput) {
				value.SourceRepository = " github.com/pingcap-inc/nexuscrm-app-packs"
			}),
		},
		{
			name: "non-canonical commit",
			input: mutateStoppedSourceInput(valid, func(value *StoppedSourceReleaseInput) {
				value.SourceCommit = strings.Repeat("A", 40)
			}),
		},
		{
			name:  "recognized component grammar",
			input: malformedComponent,
		},
		{
			name: "host compatibility",
			input: mutateStoppedSourceInput(valid, func(value *StoppedSourceReleaseInput) {
				value.Compatibility.EngineCapabilities = []string{}
			}),
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			if _, _, err := VerifyStoppedSourceReleaseCandidate(test.input); err == nil {
				t.Fatal("unproven stopped-source input was accepted")
			}
		})
	}
}

func TestImportStoppedSourceReleaseRejectsExpectationMismatchBeforeDatabaseAccess(t *testing.T) {
	valid := testStoppedSourceInput(t, testRegisteredRelease(t))
	tests := []struct {
		name   string
		mutate func(*StoppedSourceReleaseInput)
	}{
		{name: "AppID", mutate: func(value *StoppedSourceReleaseInput) { value.AppID = "io.example.other" }},
		{name: "AppReleaseDigest", mutate: func(value *StoppedSourceReleaseInput) { value.ExpectedAppReleaseDigest = testDigest("wrong-release") }},
		{name: "package digest", mutate: func(value *StoppedSourceReleaseInput) {
			value.ExpectedPackageContentDigest = testDigest("wrong-package")
		}},
		{name: "repository", mutate: func(value *StoppedSourceReleaseInput) { value.SourceRepository += " " }},
		{name: "commit", mutate: func(value *StoppedSourceReleaseInput) { value.SourceCommit = strings.Repeat("b", 39) }},
		{name: "compatibility", mutate: func(value *StoppedSourceReleaseInput) {
			value.Compatibility.ComponentKinds = []releaseartifact.SupportedComponentKind{}
		}},
	}
	for _, test := range tests {
		t.Run(test.name, func(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)
			}
			input := mutateStoppedSourceInput(valid, test.mutate)
			if _, _, err := authority.ImportStoppedSourceRelease(context.Background(), input); err == nil {
				t.Fatal("unproven stopped-source input was imported")
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("rejected input accessed Catalog storage: %v", err)
			}
		})
	}
}

func TestImportStoppedSourceReleasePersistsExactNonPromotableGraphAtomically(t *testing.T) {
	release := testRegisteredRelease(t)
	input := testStoppedSourceInput(t, release)
	_, reference, err := VerifyStoppedSourceReleaseCandidate(input)
	if err != nil {
		t.Fatal(err)
	}
	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)
	}

	expectStoppedSourceImportPrelude(mock, release)
	mock.ExpectQuery(stoppedSourceReplayQueryPattern()).
		WithArgs(release.AppID, release.SemanticVersion).
		WillReturnError(sql.ErrNoRows)
	expectStoppedSourceReleaseInsert(mock, release)
	expectStoppedSourceReceiptInsert(mock, reference, nil)
	expectStoppedSourceReleaseGraphInsert(mock, release)
	mock.ExpectCommit()
	expectStoredReleaseRead(mock, release, ReleaseAdmissionStoppedSource)
	expectStoppedSourceReceiptRead(mock, reference, ReleaseAdmissionStoppedSource, nil, nil, nil, false)
	expectStoppedSourceGraphRead(mock, release)

	got, gotReference, err := authority.ImportStoppedSourceRelease(context.Background(), input)
	if err != nil {
		t.Fatalf("import stopped-source release: %v", err)
	}
	if got.AdmissionKind != ReleaseAdmissionStoppedSource ||
		got.AppReleaseDigest != release.AppReleaseDigest ||
		gotReference != reference {
		t.Fatalf("import = %#v, %#v; want source release %#v, %#v", got, gotReference, release, reference)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestImportStoppedSourceReleaseExactReplayIsIdempotent(t *testing.T) {
	release := testRegisteredRelease(t)
	input := testStoppedSourceInput(t, release)
	_, reference, err := VerifyStoppedSourceReleaseCandidate(input)
	if err != nil {
		t.Fatal(err)
	}
	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)
	}

	expectStoppedSourceImportPrelude(mock, release)
	expectStoppedSourceExistingRelease(mock, release, ReleaseAdmissionStoppedSource, nil, nil, nil, false, true)
	expectStoppedSourceReceiptReadTx(mock, reference)
	expectEmptyReleaseLayoutBundleSlots(mock, release.AppID, release.AppReleaseDigest)
	mock.ExpectCommit()
	expectStoredReleaseRead(mock, release, ReleaseAdmissionStoppedSource)
	expectStoppedSourceReceiptRead(mock, reference, ReleaseAdmissionStoppedSource, nil, nil, nil, false)
	expectStoppedSourceGraphRead(mock, release)

	got, gotReference, err := authority.ImportStoppedSourceRelease(context.Background(), input)
	if err != nil {
		t.Fatalf("replay stopped-source release: %v", err)
	}
	if got.AdmissionKind != ReleaseAdmissionStoppedSource || gotReference != reference {
		t.Fatalf("replay = %#v, %#v", got, gotReference)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestImportStoppedSourceReleaseRejectsChangedReceiptAndSignedCollision(t *testing.T) {
	release := testRegisteredRelease(t)
	input := testStoppedSourceInput(t, release)
	_, reference, err := VerifyStoppedSourceReleaseCandidate(input)
	if err != nil {
		t.Fatal(err)
	}

	t.Run("changed source receipt", func(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)
		}
		existingReference, err := newStoppedSourceReleaseReference(
			reference.AppID,
			reference.AppReleaseDigest,
			reference.PackageContentDigest,
			reference.SourceRepository,
			strings.Repeat("b", 40),
		)
		if err != nil {
			t.Fatal(err)
		}
		expectStoppedSourceImportPrelude(mock, release)
		expectStoppedSourceExistingRelease(mock, release, ReleaseAdmissionStoppedSource, nil, nil, nil, false, true)
		expectStoppedSourceReceiptReadTx(mock, existingReference)
		mock.ExpectRollback()

		if _, _, err := authority.ImportStoppedSourceRelease(context.Background(), input); !errors.Is(err, ErrConflict) {
			t.Fatalf("changed receipt import = %v, want ErrConflict", err)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatal(err)
		}
	})

	t.Run("signed release at semantic version", func(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)
		}
		verifiedAt := time.Unix(1_800_000_000, 0).UTC()
		expectStoppedSourceImportPrelude(mock, release)
		expectStoppedSourceExistingRelease(
			mock,
			release,
			ReleaseAdmissionSignedPublisher,
			"key-1",
			testDigest("attestation"),
			verifiedAt,
			true,
			false,
		)
		mock.ExpectRollback()

		if _, _, err := authority.ImportStoppedSourceRelease(context.Background(), input); !errors.Is(err, ErrConflict) {
			t.Fatalf("source over signed release = %v, want ErrConflict", err)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatal(err)
		}
	})
}

func TestSignedImporterCannotUpgradeStoppedSourceAdmission(t *testing.T) {
	now := time.Unix(1_800_000_000, 0).UTC()
	packageBytes, attestationBytes := testImportArtifact(t, now.Add(-time.Second))
	_, verified, err := releaseartifact.ParsePackage(packageBytes)
	if err != nil {
		t.Fatal(err)
	}
	release := registeredRelease(verified)
	attestation, err := releaseartifact.ParseAttestation(attestationBytes)
	if err != nil {
		t.Fatal(err)
	}
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	authority, err := New(db, Options{Now: func() time.Time { return now }})
	if err != nil {
		t.Fatal(err)
	}
	validFrom, validUntil := now.Add(-time.Hour), now.Add(time.Hour)

	mock.ExpectQuery(`(?s)SELECT p\.State, k\.State, k\.Algorithm, k\.PublicKey, k\.ValidFrom, k\.ValidUntil.*WHERE p\.PublisherId = \? AND k\.KeyId = \?$`).
		WithArgs(release.PublisherID, attestation.KeyID).
		WillReturnRows(sqlmock.NewRows([]string{
			"PublisherState", "KeyState", "Algorithm", "PublicKey", "ValidFrom", "ValidUntil",
		}).AddRow("ACTIVE", "ACTIVE", releaseartifact.SignatureAlgorithm, testImportPublicKey(), validFrom, validUntil))
	mock.ExpectBegin()
	expectAppPublicationAdmissionFence(mock)
	mock.ExpectQuery(`(?s)SELECT PublisherId FROM _System_CatalogAppIdentity WHERE AppId = \? FOR UPDATE`).
		WithArgs(release.AppID).
		WillReturnRows(sqlmock.NewRows([]string{"PublisherId"}).AddRow(release.PublisherID))
	mock.ExpectQuery(`(?s)SELECT p\.State, k\.State, k\.Algorithm, k\.PublicKey, k\.ValidFrom, k\.ValidUntil.*WHERE p\.PublisherId = \? AND k\.KeyId = \? FOR UPDATE`).
		WithArgs(release.PublisherID, attestation.KeyID).
		WillReturnRows(sqlmock.NewRows([]string{
			"PublisherState", "KeyState", "Algorithm", "PublicKey", "ValidFrom", "ValidUntil",
		}).AddRow("ACTIVE", "ACTIVE", releaseartifact.SignatureAlgorithm, testImportPublicKey(), validFrom, validUntil))
	mock.ExpectQuery(`(?s)SELECT r\.AppReleaseDigest, r\.PublisherId, r\.CanonicalizationProfile, r\.ArtifactMediaType,.*WHERE r\.AppId = \? AND r\.SemanticVersion = \? FOR UPDATE`).
		WithArgs(release.AppID, release.SemanticVersion).
		WillReturnRows(sqlmock.NewRows([]string{
			"AppReleaseDigest",
			"PublisherId",
			"CanonicalizationProfile",
			"ArtifactMediaType",
			"AdmissionKind",
			"AttestationKeyId",
			"AttestationDigest",
			"AttestationVerifiedAt",
			"HasArtifactReceipt",
			"HasSourceReceipt",
		}).AddRow(
			release.AppReleaseDigest,
			release.PublisherID,
			releaseartifact.CanonicalizationProfile,
			releaseartifact.ArtifactMediaType,
			ReleaseAdmissionStoppedSource,
			nil,
			nil,
			nil,
			false,
			true,
		))
	mock.ExpectRollback()

	if _, err := authority.ImportReleaseArtifact(context.Background(), packageBytes, attestationBytes, testImportHost()); !errors.Is(err, ErrConflict) {
		t.Fatalf("signed import over stopped source = %v, want ErrConflict", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("signed importer mutated stopped-source admission: %v", err)
	}
}

func TestImportStoppedSourceReleaseNeverMintsOrChangesAppIdentity(t *testing.T) {
	release := testRegisteredRelease(t)
	input := testStoppedSourceInput(t, release)
	for _, test := range []struct {
		name        string
		identityErr error
		publisherID string
		want        error
	}{
		{name: "identity absent", identityErr: sql.ErrNoRows, want: ErrNotReady},
		{name: "publisher mismatch", publisherID: "io.other", want: ErrConflict},
	} {
		t.Run(test.name, func(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)
			}
			mock.ExpectBegin()
			expectAppPublicationAdmissionFence(mock)
			identity := mock.ExpectQuery(`(?s)SELECT PublisherId.*FROM _System_CatalogAppIdentity.*WHERE AppId = \? FOR UPDATE`).
				WithArgs(release.AppID)
			if test.identityErr != nil {
				identity.WillReturnError(test.identityErr)
			} else {
				identity.WillReturnRows(sqlmock.NewRows([]string{"PublisherId"}).AddRow(test.publisherID))
			}
			mock.ExpectRollback()

			if _, _, err := authority.ImportStoppedSourceRelease(context.Background(), input); !errors.Is(err, test.want) {
				t.Fatalf("import = %v, want %v", err, test.want)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("identity rejection reached a release or identity write: %v", err)
			}
		})
	}
}

func TestImportStoppedSourceReleaseRollsBackParentWhenReceiptWriteFails(t *testing.T) {
	release := testRegisteredRelease(t)
	input := testStoppedSourceInput(t, release)
	_, reference, err := VerifyStoppedSourceReleaseCandidate(input)
	if err != nil {
		t.Fatal(err)
	}
	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)
	}
	writeErr := errors.New("receipt storage failed")

	expectStoppedSourceImportPrelude(mock, release)
	mock.ExpectQuery(stoppedSourceReplayQueryPattern()).
		WithArgs(release.AppID, release.SemanticVersion).
		WillReturnError(sql.ErrNoRows)
	expectStoppedSourceReleaseInsert(mock, release)
	expectStoppedSourceReceiptInsert(mock, reference, writeErr)
	mock.ExpectRollback()

	if _, _, err := authority.ImportStoppedSourceRelease(context.Background(), input); !errors.Is(err, writeErr) {
		t.Fatalf("receipt write failure = %v, want %v", err, writeErr)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("failed receipt write leaked a child write or commit: %v", err)
	}
}

func TestReadStoppedSourceReleaseFailsClosedOnMissingOrWrongReceiptGraph(t *testing.T) {
	release := testRegisteredRelease(t)
	input := testStoppedSourceInput(t, release)
	_, reference, err := VerifyStoppedSourceReleaseCandidate(input)
	if err != nil {
		t.Fatal(err)
	}

	t.Run("orphan receipt is never consulted without its release", func(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)
		}
		mock.ExpectQuery(`(?s)SELECT PublisherId, SemanticVersion, MetadataBundleDigest, AdmissionKind, CanonicalManifest, CanonicalBundle.*FROM _System_CatalogAppRelease`).
			WithArgs(reference.AppID, reference.AppReleaseDigest).
			WillReturnError(sql.ErrNoRows)
		if _, err := authority.ReadStoppedSourceRelease(context.Background(), reference); !errors.Is(err, ErrNotFound) {
			t.Fatalf("orphan receipt read = %v, want ErrNotFound", err)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatalf("orphan receipt was queried before ReadRelease: %v", err)
		}
	})

	t.Run("missing receipt", func(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)
		}
		expectStoredReleaseRead(mock, release, ReleaseAdmissionStoppedSource)
		mock.ExpectQuery(stoppedSourceReceiptReadQueryPattern()).
			WithArgs(reference.AppID, reference.AppReleaseDigest).
			WillReturnError(sql.ErrNoRows)
		if _, err := authority.ReadStoppedSourceRelease(context.Background(), reference); !errors.Is(err, ErrCorrupt) {
			t.Fatalf("missing receipt read = %v, want ErrCorrupt", err)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatal(err)
		}
	})

	t.Run("different exact source receipt", func(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)
		}
		other, err := newStoppedSourceReleaseReference(
			reference.AppID,
			reference.AppReleaseDigest,
			reference.PackageContentDigest,
			reference.SourceRepository,
			strings.Repeat("b", 40),
		)
		if err != nil {
			t.Fatal(err)
		}
		expectStoredReleaseRead(mock, release, ReleaseAdmissionStoppedSource)
		expectStoppedSourceReceiptRead(mock, other, ReleaseAdmissionStoppedSource, nil, nil, nil, false)
		if _, err := authority.ReadStoppedSourceRelease(context.Background(), reference); !errors.Is(err, ErrConflict) {
			t.Fatalf("different receipt read = %v, want ErrConflict", err)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatal(err)
		}
	})

	t.Run("receipt attached to conflicting parent graph", func(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)
		}
		expectStoredReleaseRead(mock, release, ReleaseAdmissionStoppedSource)
		expectStoppedSourceReceiptRead(mock, reference, ReleaseAdmissionStoppedSource, nil, nil, nil, true)
		if _, err := authority.ReadStoppedSourceRelease(context.Background(), reference); !errors.Is(err, ErrCorrupt) {
			t.Fatalf("conflicting receipt parent = %v, want ErrCorrupt", err)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatal(err)
		}
	})

	t.Run("receipt package digest differs from stored package bytes", func(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)
		}
		other, err := newStoppedSourceReleaseReference(
			reference.AppID,
			reference.AppReleaseDigest,
			testDigest("different-package-content"),
			reference.SourceRepository,
			reference.SourceCommit,
		)
		if err != nil {
			t.Fatal(err)
		}
		expectStoredReleaseRead(mock, release, ReleaseAdmissionStoppedSource)
		expectStoppedSourceReceiptRead(mock, other, ReleaseAdmissionStoppedSource, nil, nil, nil, false)
		if _, err := authority.ReadStoppedSourceRelease(context.Background(), other); !errors.Is(err, ErrCorrupt) {
			t.Fatalf("wrong package receipt = %v, want ErrCorrupt", err)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatal(err)
		}
	})

	t.Run("derived release graph is incomplete", func(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)
		}
		expectStoredReleaseRead(mock, release, ReleaseAdmissionStoppedSource)
		expectStoppedSourceReceiptRead(mock, reference, ReleaseAdmissionStoppedSource, nil, nil, nil, false)
		mock.ExpectQuery(`(?s)SELECT Alias, DependencyAppId, MinimumVersionInclusive,.*FROM _System_CatalogAppReleaseDependency.*ORDER BY Alias`).
			WithArgs(release.AppID, release.AppReleaseDigest).
			WillReturnRows(sqlmock.NewRows([]string{
				"Alias",
				"DependencyAppId",
				"MinimumVersionInclusive",
				"MaximumVersionExclusive",
				"RequiredCapabilities",
			}))
		mock.ExpectQuery(`(?s)SELECT NamespaceId.*FROM _System_CatalogAppReleaseNamespace.*ORDER BY NamespaceId`).
			WithArgs(release.AppID, release.AppReleaseDigest).
			WillReturnRows(sqlmock.NewRows([]string{"NamespaceId"}))
		if _, err := authority.ReadStoppedSourceRelease(context.Background(), reference); !errors.Is(err, ErrCorrupt) {
			t.Fatalf("incomplete derived graph = %v, want ErrCorrupt", err)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatal(err)
		}
	})
}

func TestStoppedSourceReleaseCannotUseOrdinaryOperationOrSealedRead(t *testing.T) {
	release := testRegisteredRelease(t)
	input := testStoppedSourceInput(t, release)
	_, reference, err := VerifyStoppedSourceReleaseCandidate(input)
	if err != nil {
		t.Fatal(err)
	}
	request := testStoppedSourcePrepareRequest(release, "prepare-ordinary")

	t.Run("ordinary operation", func(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)
		}
		expectStoredReleaseRead(mock, release, ReleaseAdmissionStoppedSource)
		if _, err := authority.BeginOperation(context.Background(), request); !errors.Is(err, ErrNotReady) {
			t.Fatalf("ordinary source operation = %v, want ErrNotReady", err)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatalf("ordinary operation entered a lifecycle transaction: %v", err)
		}
	})

	t.Run("sealed release", func(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)
		}
		expectStoredReleaseRead(mock, release, ReleaseAdmissionStoppedSource)
		mock.ExpectQuery(`(?s)SELECT PublisherId, SemanticVersion,.*AdmissionKind.*FROM _System_CatalogAppRelease.*WHERE AppId = \? AND AppReleaseDigest = \?`).
			WithArgs(reference.AppID, reference.AppReleaseDigest).
			WillReturnRows(sqlmock.NewRows([]string{
				"PublisherId",
				"SemanticVersion",
				"CanonicalizationProfile",
				"ArtifactMediaType",
				"AdmissionKind",
				"AttestationKeyId",
				"AttestationDigest",
				"AttestationVerifiedAt",
			}).AddRow(
				release.PublisherID,
				release.SemanticVersion,
				releaseartifact.CanonicalizationProfile,
				releaseartifact.ArtifactMediaType,
				ReleaseAdmissionStoppedSource,
				nil,
				nil,
				nil,
			))
		if _, err := authority.ReadSealedRelease(context.Background(), reference.AppID, reference.AppReleaseDigest); !errors.Is(err, ErrUnproven) {
			t.Fatalf("sealed source read = %v, want ErrUnproven", err)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatalf("sealed read consulted signed receipt for source admission: %v", err)
		}
	})
}

func TestBeginStoppedSourceOperationRejectsWrongAdmissionReferenceAndDependencies(t *testing.T) {
	release := testRegisteredRelease(t)
	input := testStoppedSourceInput(t, release)
	_, reference, err := VerifyStoppedSourceReleaseCandidate(input)
	if err != nil {
		t.Fatal(err)
	}
	request := testStoppedSourcePrepareRequest(release, "prepare-rejections")

	for _, test := range []struct {
		name   string
		mutate func(*BeginOperationRequest, *StoppedSourceReleaseReference)
		want   error
	}{
		{
			name: "wrong App coordinate",
			mutate: func(request *BeginOperationRequest, _ *StoppedSourceReleaseReference) {
				request.AppID = "io.example.other"
			},
			want: ErrConflict,
		},
		{
			name: "wrong release coordinate",
			mutate: func(request *BeginOperationRequest, _ *StoppedSourceReleaseReference) {
				request.TargetAppReleaseDigest = testDigest("wrong-target")
			},
			want: ErrConflict,
		},
		{
			name: "remove has no exact target",
			mutate: func(request *BeginOperationRequest, _ *StoppedSourceReleaseReference) {
				request.Kind = OperationRemove
				request.TargetAppReleaseDigest = ""
			},
			want: ErrInvalidTransition,
		},
	} {
		t.Run(test.name, func(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)
			}
			candidateRequest, candidateReference := request, reference
			test.mutate(&candidateRequest, &candidateReference)
			if _, err := authority.BeginStoppedSourceOperation(context.Background(), candidateRequest, candidateReference); !errors.Is(err, test.want) {
				t.Fatalf("begin = %v, want %v", err, test.want)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("coordinate rejection accessed Catalog: %v", err)
			}
		})
	}

	t.Run("signed target", func(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)
		}
		expectStoredReleaseRead(mock, release, ReleaseAdmissionSignedPublisher)
		if _, err := authority.BeginStoppedSourceOperation(context.Background(), request, reference); !errors.Is(err, ErrNotReady) {
			t.Fatalf("signed stopped-source target = %v, want ErrNotReady", err)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatal(err)
		}
	})

	t.Run("tampered source coordinate", func(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)
		}
		tampered := reference
		tampered.SourceCommit = strings.Repeat("b", 40)
		expectStoredReleaseRead(mock, release, ReleaseAdmissionStoppedSource)
		if _, err := authority.BeginStoppedSourceOperation(context.Background(), request, tampered); err == nil {
			t.Fatal("tampered source reference was accepted")
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatalf("tampered reference reached receipt or lifecycle mutation: %v", err)
		}
	})

	t.Run("dependency set", func(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)
		}
		withDependency := request
		withDependency.ResolvedDependencies = []ResolvedDependency{{
			Alias: "base", AppID: "io.example.base", AppReleaseDigest: testDigest("base-release"),
		}}
		expectStoredReleaseRead(mock, release, ReleaseAdmissionStoppedSource)
		expectStoppedSourceReceiptRead(mock, reference, ReleaseAdmissionStoppedSource, nil, nil, nil, false)
		expectStoppedSourceGraphRead(mock, release)
		if _, err := authority.BeginStoppedSourceOperation(context.Background(), withDependency, reference); !errors.Is(err, ErrNotReady) {
			t.Fatalf("dependency-bearing stopped-source operation = %v, want ErrNotReady", err)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatalf("dependency rejection entered lifecycle mutation: %v", err)
		}
	})
}

func TestBeginStoppedSourceOperationDelegatesValidPrepareAndUpgrade(t *testing.T) {
	release := testRegisteredRelease(t)
	input := testStoppedSourceInput(t, release)
	_, reference, err := VerifyStoppedSourceReleaseCandidate(input)
	if err != nil {
		t.Fatal(err)
	}
	sharedLifecycleErr := errors.New("shared lifecycle boundary reached")
	for _, test := range []struct {
		name    string
		request BeginOperationRequest
	}{
		{name: "prepare", request: testStoppedSourcePrepareRequest(release, "prepare-valid")},
		{name: "upgrade", request: func() BeginOperationRequest {
			request := testStoppedSourcePrepareRequest(release, "upgrade-valid")
			request.Kind = OperationUpgrade
			request.AppInstallationID = "018f0000-0000-7000-9000-000000000002"
			return request
		}()},
	} {
		t.Run(test.name, func(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)
			}
			expectExactStoppedSourceRead(mock, release, reference)
			mock.ExpectBegin().WillReturnError(sharedLifecycleErr)
			if _, err := authority.BeginStoppedSourceOperation(context.Background(), test.request, reference); !errors.Is(err, sharedLifecycleErr) {
				t.Fatalf("valid %s = %v, want shared lifecycle delegation", test.name, err)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatal(err)
			}
		})
	}
}

func TestBeginStoppedSourceOperationExactRetryReturnsExistingPrepareAndUpgrade(t *testing.T) {
	release := testRegisteredRelease(t)
	input := testStoppedSourceInput(t, release)
	_, reference, err := VerifyStoppedSourceReleaseCandidate(input)
	if err != nil {
		t.Fatal(err)
	}
	const (
		operationID  = "018f0000-0000-7000-9000-000000000001"
		installation = "018f0000-0000-7000-9000-000000000002"
	)
	for _, test := range []struct {
		name    string
		kind    OperationKind
		state   InstallationState
		request BeginOperationRequest
	}{
		{
			name: "prepare retry",
			kind: OperationPrepare, state: StatePreparing,
			request: testStoppedSourcePrepareRequest(release, "prepare-retry"),
		},
		{
			name: "upgrade retry",
			kind: OperationUpgrade, state: StateUpgrading,
			request: func() BeginOperationRequest {
				request := testStoppedSourcePrepareRequest(release, "upgrade-retry")
				request.Kind = OperationUpgrade
				request.AppInstallationID = installation
				return request
			}(),
		},
	} {
		t.Run(test.name, func(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)
			}
			sourceRelease := registeredReleaseWithAdmission(release.Package, ReleaseAdmissionStoppedSource)
			normalized, _, requestDigest, err := authority.validateBeginOperationForTarget(
				context.Background(), test.request, &sourceRelease, true,
			)
			if err != nil {
				t.Fatalf("normalize stopped-source request: %v", err)
			}
			expectExactStoppedSourceRead(mock, release, reference)
			mock.ExpectBegin()
			expectAppPublicationAdmissionFence(mock)
			mock.ExpectQuery(`(?s)SELECT OperationId, RequestDigest, InitiationKind FROM _System_CatalogAppOperation.*FOR UPDATE`).
				WithArgs(normalized.OrgID, normalized.AppID, normalized.IdempotencyKey).
				WillReturnRows(sqlmock.NewRows([]string{
					"OperationId", "RequestDigest", "InitiationKind",
				}).AddRow(operationID, requestDigest, initiationDirect))
			mock.ExpectQuery(`(?s)SELECT OrgId.*FROM _System_CatalogCandidateSource.*WHERE WriterCommandId = \? LIMIT 1 FOR UPDATE`).
				WithArgs(operationID).
				WillReturnError(sql.ErrNoRows)
			projectionDigest := testDigest(test.name + "-projection")
			mock.ExpectQuery(`(?s)SELECT OperationId, Kind, AppInstallationId, OrgId, AppId, State, Step,.*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, test.kind, installation, normalized.OrgID, normalized.AppID, test.state, "CANDIDATE_COMMITTED",
					normalized.TargetAppReleaseDigest, []byte("1"), []byte("1"), projectionDigest,
					nil, nil, "",
				))
			grantDigest, err := ComputeConsumerGrantSetDigest(normalized.ConsumerGrants)
			if err != nil {
				t.Fatal(err)
			}
			mock.ExpectQuery(`(?s)SELECT.*AppInstallationId, GrantSetDigest, GrantCount.*FROM _System_CatalogConsumerGrantSet.*FOR UPDATE`).
				WithArgs(operationID).
				WillReturnRows(sqlmock.NewRows([]string{
					"AppInstallationId", "GrantSetDigest", "GrantCount",
				}).AddRow(installation, grantDigest, []byte("0")))
			bindingDigest, err := ComputeIntegrationBindingSetDigest(normalized.IntegrationBindings)
			if err != nil {
				t.Fatal(err)
			}
			mock.ExpectQuery(`(?s)SELECT.*AppInstallationId, BindingSetDigest, BindingCount.*FROM _System_CatalogIntegrationBindingSet.*FOR UPDATE`).
				WithArgs(operationID).
				WillReturnRows(sqlmock.NewRows([]string{
					"AppInstallationId", "BindingSetDigest", "BindingCount",
				}).AddRow(installation, bindingDigest, []byte("0")))
			mock.ExpectCommit()

			operation, err := authority.BeginStoppedSourceOperation(context.Background(), test.request, reference)
			if err != nil {
				t.Fatalf("exact %s: %v", test.name, err)
			}
			if operation.OperationID != operationID || operation.Kind != test.kind || operation.State != test.state ||
				operation.AppInstallationID != installation || operation.TargetAppReleaseDigest != release.AppReleaseDigest {
				t.Fatalf("retry operation = %#v", operation)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatal(err)
			}
		})
	}
}

func expectStoppedSourceImportPrelude(mock sqlmock.Sqlmock, release RegisteredRelease) {
	mock.ExpectBegin()
	expectAppPublicationAdmissionFence(mock)
	mock.ExpectQuery(`(?s)SELECT PublisherId.*FROM _System_CatalogAppIdentity.*WHERE AppId = \? FOR UPDATE`).
		WithArgs(release.AppID).
		WillReturnRows(sqlmock.NewRows([]string{"PublisherId"}).AddRow(release.PublisherID))
}

func stoppedSourceReplayQueryPattern() string {
	return `(?s)SELECT r\.AppReleaseDigest, r\.PublisherId, r\.MetadataBundleDigest,.*WHERE r\.AppId = \? AND r\.SemanticVersion = \? FOR UPDATE`
}

func stoppedSourceReceiptReadQueryPattern() string {
	return `(?s)SELECT s\.ReceiptSchema, s\.SourceRepository, s\.SourceCommit,.*FROM _System_CatalogAppRelease AS r.*JOIN _System_CatalogAppReleaseSourceReceipt AS s.*WHERE r\.AppId = \? AND r\.AppReleaseDigest = \?`
}

func expectStoppedSourceExistingRelease(
	mock sqlmock.Sqlmock,
	release RegisteredRelease,
	kind ReleaseAdmissionKind,
	attestationKeyID any,
	attestationDigest any,
	attestationVerifiedAt any,
	hasArtifactReceipt bool,
	hasSourceReceipt bool,
) {
	mock.ExpectQuery(stoppedSourceReplayQueryPattern()).
		WithArgs(release.AppID, release.SemanticVersion).
		WillReturnRows(sqlmock.NewRows([]string{
			"AppReleaseDigest",
			"PublisherId",
			"MetadataBundleDigest",
			"CanonicalizationProfile",
			"ArtifactMediaType",
			"AdmissionKind",
			"AttestationKeyId",
			"AttestationDigest",
			"AttestationVerifiedAt",
			"HasArtifactReceipt",
			"HasSourceReceipt",
		}).AddRow(
			release.AppReleaseDigest,
			release.PublisherID,
			release.MetadataBundleDigest,
			releaseartifact.CanonicalizationProfile,
			releaseartifact.ArtifactMediaType,
			kind,
			attestationKeyID,
			attestationDigest,
			attestationVerifiedAt,
			hasArtifactReceipt,
			hasSourceReceipt,
		))
}

func expectStoppedSourceReleaseInsert(mock sqlmock.Sqlmock, release RegisteredRelease) {
	mock.ExpectExec(`(?s)INSERT INTO _System_CatalogAppRelease.*NULL, NULL, NULL, CURRENT_TIMESTAMP\(6\)\)`).
		WithArgs(
			release.AppID,
			release.AppReleaseDigest,
			release.PublisherID,
			release.SemanticVersion,
			release.MetadataBundleDigest,
			release.Package.CanonicalManifest,
			release.Package.CanonicalBundle,
			releaseartifact.CanonicalizationProfile,
			releaseartifact.ArtifactMediaType,
			ReleaseAdmissionStoppedSource,
		).
		WillReturnResult(sqlmock.NewResult(1, 1))
}

func expectStoppedSourceReceiptInsert(
	mock sqlmock.Sqlmock,
	reference StoppedSourceReleaseReference,
	resultErr error,
) {
	expectation := mock.ExpectExec(`(?s)INSERT INTO _System_CatalogAppReleaseSourceReceipt.*VALUES \(\?, \?, \?, \?, \?, \?, \?, \?, CURRENT_TIMESTAMP\(6\)\)`).
		WithArgs(
			reference.AppID,
			reference.AppReleaseDigest,
			releaseSourceReceiptSchema,
			reference.SourceRepository,
			reference.SourceCommit,
			reference.PackageContentDigest,
			releaseSourceNonPromotable,
			reference.ReceiptDigest,
		)
	if resultErr != nil {
		expectation.WillReturnError(resultErr)
		return
	}
	expectation.WillReturnResult(sqlmock.NewResult(1, 1))
}

func expectStoppedSourceReceiptReadTx(
	mock sqlmock.Sqlmock,
	reference StoppedSourceReleaseReference,
) {
	mock.ExpectQuery(`(?s)SELECT ReceiptSchema, SourceRepository, SourceCommit,.*FROM _System_CatalogAppReleaseSourceReceipt.*FOR UPDATE`).
		WithArgs(reference.AppID, reference.AppReleaseDigest).
		WillReturnRows(sqlmock.NewRows([]string{
			"ReceiptSchema",
			"SourceRepository",
			"SourceCommit",
			"PackageContentDigest",
			"PromotionDisposition",
			"ReceiptDigest",
		}).AddRow(
			releaseSourceReceiptSchema,
			reference.SourceRepository,
			reference.SourceCommit,
			reference.PackageContentDigest,
			releaseSourceNonPromotable,
			reference.ReceiptDigest,
		))
}

func expectStoppedSourceReleaseGraphInsert(mock sqlmock.Sqlmock, release RegisteredRelease) {
	manifest := release.Package.Manifest
	for _, dependency := range manifest.Dependencies {
		required, err := marshalJSON(dependency.RequiredCapabilities)
		if err != nil {
			panic(err)
		}
		mock.ExpectExec(`(?s)INSERT INTO _System_CatalogAppReleaseDependency`).
			WithArgs(
				manifest.AppID,
				release.AppReleaseDigest,
				dependency.Alias,
				dependency.AppID,
				dependency.MinimumVersionInclusive,
				dependency.MaximumVersionExclusive,
				required,
			).
			WillReturnResult(sqlmock.NewResult(1, 1))
	}
	for _, namespace := range manifest.OwnedNamespaces {
		mock.ExpectExec(`(?s)INSERT INTO _System_CatalogAppReleaseNamespace`).
			WithArgs(manifest.AppID, release.AppReleaseDigest, namespace).
			WillReturnResult(sqlmock.NewResult(1, 1))
	}
	for _, capability := range manifest.ProvidedCapabilities {
		required, err := marshalJSON(capability.RequiredRuntimeCapabilities)
		if err != nil {
			panic(err)
		}
		mock.ExpectExec(`(?s)INSERT INTO _System_CatalogAppReleaseCapability`).
			WithArgs(
				manifest.AppID,
				release.AppReleaseDigest,
				capability.CapabilityID,
				capability.Kind,
				nullableString(capability.HandlerRef),
				required,
			).
			WillReturnResult(sqlmock.NewResult(1, 1))
	}
	for index, component := range release.Package.Bundle.Components {
		mock.ExpectExec(`(?s)INSERT INTO _System_CatalogAppReleaseComponent`).
			WithArgs(
				manifest.AppID,
				release.AppReleaseDigest,
				component.ComponentID,
				component.NamespaceID,
				component.ComponentKind,
				component.BusinessKey,
				component.DefinitionVersion,
				component.ContentDigest,
				release.Package.ComponentBlobs[index].CanonicalBehavior,
			).
			WillReturnResult(sqlmock.NewResult(1, 1))
	}
}

func expectEmptyReleaseLayoutBundleSlots(
	mock sqlmock.Sqlmock,
	appID, appReleaseDigest string,
) {
	mock.ExpectQuery(`(?s)FROM _System_CatalogAppReleaseLayoutBundleSlot b.*ORDER BY b.LocalComponentKey`).
		WithArgs(appID, appReleaseDigest).
		WillReturnRows(sqlmock.NewRows(releaseLayoutBundleSlotColumns()))
}

func expectStoredReleaseRead(
	mock sqlmock.Sqlmock,
	release RegisteredRelease,
	kind ReleaseAdmissionKind,
) {
	mock.ExpectQuery(`(?s)SELECT PublisherId, SemanticVersion, MetadataBundleDigest, AdmissionKind, CanonicalManifest, CanonicalBundle.*FROM _System_CatalogAppRelease WHERE AppId = \? AND AppReleaseDigest = \?`).
		WithArgs(release.AppID, release.AppReleaseDigest).
		WillReturnRows(sqlmock.NewRows([]string{
			"PublisherId",
			"SemanticVersion",
			"MetadataBundleDigest",
			"AdmissionKind",
			"CanonicalManifest",
			"CanonicalBundle",
		}).AddRow(
			release.PublisherID,
			release.SemanticVersion,
			release.MetadataBundleDigest,
			kind,
			release.Package.CanonicalManifest,
			release.Package.CanonicalBundle,
		))
	componentRows := sqlmock.NewRows([]string{"ComponentId", "CanonicalBehavior"})
	for index, component := range release.Package.Bundle.Components {
		componentRows.AddRow(component.ComponentID, release.Package.ComponentBlobs[index].CanonicalBehavior)
	}
	mock.ExpectQuery(`(?s)SELECT ComponentId, CanonicalBehavior FROM _System_CatalogAppReleaseComponent.*ORDER BY ComponentId`).
		WithArgs(release.AppID, release.AppReleaseDigest).
		WillReturnRows(componentRows)
	mock.ExpectBegin()
	mock.ExpectQuery(`(?s)SELECT AppId, AppReleaseDigest FROM _System_CatalogAppRelease.*WHERE AppId = \? AND AppReleaseDigest = \?`).
		WithArgs(release.AppID, release.AppReleaseDigest).
		WillReturnRows(sqlmock.NewRows([]string{"AppId", "AppReleaseDigest"}).AddRow(release.AppID, release.AppReleaseDigest))
	expectEmptyReleaseLayoutBundleSlots(mock, release.AppID, release.AppReleaseDigest)
	mock.ExpectCommit()
}

func expectStoppedSourceReceiptRead(
	mock sqlmock.Sqlmock,
	reference StoppedSourceReleaseReference,
	kind ReleaseAdmissionKind,
	attestationKeyID any,
	attestationDigest any,
	attestationVerifiedAt any,
	hasArtifactReceipt bool,
) {
	mock.ExpectQuery(stoppedSourceReceiptReadQueryPattern()).
		WithArgs(reference.AppID, reference.AppReleaseDigest).
		WillReturnRows(sqlmock.NewRows([]string{
			"ReceiptSchema",
			"SourceRepository",
			"SourceCommit",
			"PackageContentDigest",
			"PromotionDisposition",
			"ReceiptDigest",
			"CanonicalizationProfile",
			"ArtifactMediaType",
			"AdmissionKind",
			"AttestationKeyId",
			"AttestationDigest",
			"AttestationVerifiedAt",
			"HasArtifactReceipt",
		}).AddRow(
			releaseSourceReceiptSchema,
			reference.SourceRepository,
			reference.SourceCommit,
			reference.PackageContentDigest,
			releaseSourceNonPromotable,
			reference.ReceiptDigest,
			releaseartifact.CanonicalizationProfile,
			releaseartifact.ArtifactMediaType,
			kind,
			attestationKeyID,
			attestationDigest,
			attestationVerifiedAt,
			hasArtifactReceipt,
		))
}

func expectStoppedSourceGraphRead(mock sqlmock.Sqlmock, release RegisteredRelease) {
	dependencyRows := sqlmock.NewRows([]string{
		"Alias",
		"DependencyAppId",
		"MinimumVersionInclusive",
		"MaximumVersionExclusive",
		"RequiredCapabilities",
	})
	for _, dependency := range release.Package.Manifest.Dependencies {
		required, err := marshalJSON(dependency.RequiredCapabilities)
		if err != nil {
			panic(err)
		}
		dependencyRows.AddRow(
			dependency.Alias,
			dependency.AppID,
			dependency.MinimumVersionInclusive,
			dependency.MaximumVersionExclusive,
			required,
		)
	}
	mock.ExpectQuery(`(?s)SELECT Alias, DependencyAppId, MinimumVersionInclusive,.*FROM _System_CatalogAppReleaseDependency.*ORDER BY Alias`).
		WithArgs(release.AppID, release.AppReleaseDigest).
		WillReturnRows(dependencyRows)

	namespaceRows := sqlmock.NewRows([]string{"NamespaceId"})
	for _, namespaceID := range release.Package.Manifest.OwnedNamespaces {
		namespaceRows.AddRow(namespaceID)
	}
	mock.ExpectQuery(`(?s)SELECT NamespaceId.*FROM _System_CatalogAppReleaseNamespace.*ORDER BY NamespaceId`).
		WithArgs(release.AppID, release.AppReleaseDigest).
		WillReturnRows(namespaceRows)

	capabilityRows := sqlmock.NewRows([]string{
		"CapabilityId",
		"Kind",
		"HandlerRef",
		"RequiredRuntimeCapabilities",
	})
	for _, capability := range release.Package.Manifest.ProvidedCapabilities {
		required, err := marshalJSON(capability.RequiredRuntimeCapabilities)
		if err != nil {
			panic(err)
		}
		capabilityRows.AddRow(
			capability.CapabilityID,
			capability.Kind,
			nullableString(capability.HandlerRef),
			required,
		)
	}
	mock.ExpectQuery(`(?s)SELECT CapabilityId, Kind, HandlerRef, RequiredRuntimeCapabilities.*FROM _System_CatalogAppReleaseCapability.*ORDER BY CapabilityId`).
		WithArgs(release.AppID, release.AppReleaseDigest).
		WillReturnRows(capabilityRows)

	componentRows := sqlmock.NewRows([]string{
		"ComponentId",
		"NamespaceId",
		"ComponentKind",
		"BusinessKey",
		"DefinitionVersion",
		"ContentDigest",
	})
	for _, component := range release.Package.Bundle.Components {
		componentRows.AddRow(
			component.ComponentID,
			component.NamespaceID,
			component.ComponentKind,
			component.BusinessKey,
			component.DefinitionVersion,
			component.ContentDigest,
		)
	}
	mock.ExpectQuery(`(?s)SELECT ComponentId, NamespaceId, ComponentKind,.*FROM _System_CatalogAppReleaseComponent.*ORDER BY ComponentId`).
		WithArgs(release.AppID, release.AppReleaseDigest).
		WillReturnRows(componentRows)
}

func expectExactStoppedSourceRead(
	mock sqlmock.Sqlmock,
	release RegisteredRelease,
	reference StoppedSourceReleaseReference,
) {
	expectStoredReleaseRead(mock, release, ReleaseAdmissionStoppedSource)
	expectStoppedSourceReceiptRead(
		mock,
		reference,
		ReleaseAdmissionStoppedSource,
		nil,
		nil,
		nil,
		false,
	)
	expectStoppedSourceGraphRead(mock, release)
}

func testStoppedSourceInput(t *testing.T, release RegisteredRelease) StoppedSourceReleaseInput {
	t.Helper()
	_, packageBytes, err := releaseartifact.BuildPackage(release.Package.Package)
	if err != nil {
		t.Fatal(err)
	}
	return StoppedSourceReleaseInput{
		AppID:                        release.AppID,
		ExpectedAppReleaseDigest:     release.AppReleaseDigest,
		ExpectedPackageContentDigest: exactPackageContentDigest(packageBytes),
		SourceRepository:             "github.com/pingcap-inc/nexuscrm-app-packs",
		SourceCommit:                 strings.Repeat("a", 40),
		PackageBytes:                 packageBytes,
		Compatibility:                testImportHost(),
	}
}

func testStoppedSourcePrepareRequest(
	release RegisteredRelease,
	idempotencyKey string,
) BeginOperationRequest {
	return BeginOperationRequest{
		Kind:                   OperationPrepare,
		OrgID:                  "org-stopped-source",
		AppID:                  release.AppID,
		TargetAppReleaseDigest: release.AppReleaseDigest,
		IdempotencyKey:         idempotencyKey,
		RuntimeAudience:        "runtime-stopped-source",
		AuthorizedWorkloads:    []string{"workload-stopped-source"},
		GrantedCapabilities:    append([]string(nil), release.Package.Manifest.RequiredEngineCapabilities...),
		HandlerWorkloadID:      "workload-stopped-source",
		ResolvedDependencies:   nil,
	}
}

func mutateStoppedSourceInput(
	input StoppedSourceReleaseInput,
	mutate func(*StoppedSourceReleaseInput),
) StoppedSourceReleaseInput {
	input.PackageBytes = append([]byte(nil), input.PackageBytes...)
	mutate(&input)
	return input
}