package serviceaccount

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/store/app"
	"github.com/tidbcloud/AgentTeam/internal/store/attachment"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	serviceaccountstore "github.com/tidbcloud/AgentTeam/internal/store/serviceaccount"
	"github.com/tidbcloud/AgentTeam/internal/workspace/owner"
)

func TestPlatformServiceAccountProvisioningBroadcastAndReplay(t *testing.T) {
	ctx := context.Background()
	db := materializerTiDB.SerializedRollbackDB(t, ctx)
	now := time.Date(2026, 7, 11, 6, 0, 0, 0, time.UTC)
	seedMaterializerWorkspace(t, ctx, db, "ws_platform_one", now, map[string]string{
		"mem_platform_one_admin": "active",
		"mem_platform_one_user":  "active",
	})
	seedMaterializerWorkspace(t, ctx, db, "ws_platform_two", now, map[string]string{
		"mem_platform_two_admin": "active",
		"mem_platform_two_user":  "active",
	})
	operatorUserID := "usr_ws_platform_one_mem_platform_one_admin"
	_, err := db.Exec(ctx, `UPDATE users SET is_admin = TRUE WHERE id = $1`, operatorUserID)
	require.NoError(t, err)

	clock := now
	deletionObserver := &recordingLocalAccountDeletionObserver{}
	messageFanout := &recordingPlatformBroadcastMessageFanout{queryer: db}
	messageStore := app.New(db, app.Options{})
	messageStore.SetCommittedMessageFanout(messageFanout)
	platform := NewPlatformService(db, PlatformServiceOptions{
		ControlPlane: true, ManagedWorkspaces: owner.NewAllSet(),
		MessageWriter:                messageStore,
		LocalAccountDeletionObserver: deletionObserver,
		Now:                          func() time.Time { return clock },
	})
	created, err := platform.CreatePlatformAccount(ctx, PlatformAccountCreate{
		ID: "psacct_platform_news", Name: "Platform News",
		Description: "Deployment announcements", OperatorUserID: operatorUserID,
		CorrelationID: "corr-platform-create", Now: now,
	})
	require.NoError(t, err)
	secondAccount, err := platform.CreatePlatformAccount(ctx, PlatformAccountCreate{
		ID: "psacct_platform_status", Name: "Platform Status",
		OperatorUserID: operatorUserID, Now: now,
	})
	require.NoError(t, err)
	require.NotEqual(t, created.ID, secondAccount.ID)
	accounts, err := platform.ListPlatformAccounts(ctx, "", "", 10)
	require.NoError(t, err)
	require.Len(t, accounts, 2)
	matchingAccounts, err := platform.ListPlatformAccounts(ctx, "", "deployment announcements", 10)
	require.NoError(t, err)
	require.Len(t, matchingAccounts, 1)
	require.Equal(t, created.ID, matchingAccounts[0].ID)
	require.Equal(t, int64(2), countRows(t, ctx, db, `
SELECT COUNT(*) FROM platform_service_account_provisioning_inputs
WHERE platform_service_account_id = $1
`, created.ID))

	drainPlatformWork(t, ctx, platform, 40)
	ready, err := platform.GetPlatformAccount(ctx, created.ID)
	require.NoError(t, err)
	require.Equal(t, "ready", ready.ProvisioningState)
	require.Equal(t, int64(1), ready.AppliedProvisioningGeneration)
	require.Equal(t, int64(2), countRows(t, ctx, db, `
SELECT COUNT(*) FROM platform_service_account_workspace_instances
WHERE platform_service_account_id = $1 AND provisioning_state = 'ready'
`, created.ID))
	for _, workspaceID := range []string{"ws_platform_one", "ws_platform_two"} {
		require.Equal(t, int64(1), countRows(t, ctx, db, `
	SELECT COUNT(*) FROM service_accounts
	WHERE workspace_id = $1 AND platform_service_account_id = $2
	  AND management_kind = 'platform'
  AND audience_state = 'ready' AND effective_audience_version = audience_policy_version
`, workspaceID, created.ID))
		require.Equal(t, int64(2), countRows(t, ctx, db, `
SELECT COUNT(*)
FROM scope_memberships membership
JOIN service_accounts account ON account.message_scope_id = membership.scope_id
JOIN members member ON member.id = membership.member_id
WHERE account.workspace_id = $1 AND account.platform_service_account_id = $2
  AND member.type = 'human' AND member.status = 'active'
  AND membership.effective_to_seq IS NULL AND membership.removed_at IS NULL
`, workspaceID, created.ID))
	}
	avatarAssetID, err := attachment.New(db, nil, nil, 0, attachment.Options{}).UploadResourceAvatar(
		ctx, "ws_platform_one", "service_account", ready.ID,
		attachment.CreateAttachmentParams{
			Filename: "platform-news.png", ContentType: "image/png", Blob: []byte("platform-news-avatar"),
		},
	)
	require.NoError(t, err)
	updatedProfile, err := platform.UpdatePlatformAccountAvatar(ctx, PlatformAccountAvatarMutation{
		PlatformServiceAccountID: ready.ID,
		ExpectedVersion:          ready.Version,
		AvatarAssetID:            avatarAssetID,
		OperatorUserID:           operatorUserID,
		CorrelationID:            "corr-platform-avatar",
		Now:                      now.Add(time.Minute),
	})
	require.NoError(t, err)
	require.Equal(t, avatarAssetID, updatedProfile.AvatarAssetID)
	require.Equal(t, ready.ProfileVersion+1, updatedProfile.ProfileVersion)
	require.Equal(t, int64(2), countRows(t, ctx, db, `
SELECT COUNT(*)
FROM service_members member
JOIN service_accounts account ON account.service_member_id = member.id
WHERE account.platform_service_account_id = $1
  AND member.avatar_url = $2
  AND account.avatar_asset_id = $3
  AND account.applied_platform_profile_version = $4
`, ready.ID, attachment.AvatarAssetURL(avatarAssetID), avatarAssetID, updatedProfile.ProfileVersion))
	assetPayload := []byte("platform broadcast asset")
	assetDigest := sha256.Sum256(assetPayload)
	var sourceScopeID, sourceAuthorMemberID string
	require.NoError(t, db.QueryRow(ctx, `
SELECT message_scope_id, author_member_id
FROM service_accounts
WHERE workspace_id = 'ws_platform_one' AND platform_service_account_id = $1
`, created.ID).Scan(&sourceScopeID, &sourceAuthorMemberID))
	_, err = db.Exec(ctx, `
INSERT INTO attachments (
  id, workspace_id, scope_id, uploaded_by_member_id, message_id,
  filename, content_type, size_bytes, storage_driver, object_key, payload, created_at
) VALUES (
  'att_platform_broadcast_source', 'ws_platform_one', $1, $2, NULL,
  'source.txt', 'text/plain', $3, '', NULL, $4, $5
)
`, sourceScopeID, sourceAuthorMemberID, len(assetPayload), assetPayload, now)
	require.NoError(t, err)

	broadcast, err := platform.SubmitPlatformBroadcast(ctx, PlatformBroadcastSubmit{
		PlatformServiceAccountID: created.ID, OperatorUserID: operatorUserID,
		ClientBroadcastID: "broadcast-client-1", CorrelationID: "corr-broadcast-1",
		Plan: PlatformBroadcastPlan{
			Title: "Deployment announcement", Content: "A deployment-wide announcement", Metadata: map[string]any{"clientFact": "retained"},
			Assets: []PlatformBroadcastAsset{{
				SourceAssetRef: "att_platform_broadcast_source",
				ContentHash:    hex.EncodeToString(assetDigest[:]), MediaType: "text/plain",
				SizeBytes: int64(len(assetPayload)), ClientLabel: "announcement.txt",
			}},
		},
		Now: clock,
	})
	require.NoError(t, err)
	require.Equal(t, serviceaccountstore.PlatformBroadcastStateWaitingProvisioning, broadcast.State)
	replay, err := platform.SubmitPlatformBroadcast(ctx, PlatformBroadcastSubmit{
		PlatformServiceAccountID: created.ID, OperatorUserID: operatorUserID,
		ClientBroadcastID: "broadcast-client-1", CorrelationID: "different-correlation-does-not-change-payload",
		Plan: PlatformBroadcastPlan{
			Title: "Deployment announcement", Content: "A deployment-wide announcement", Metadata: map[string]any{"clientFact": "retained"},
			Assets: []PlatformBroadcastAsset{{
				SourceAssetRef: "att_platform_broadcast_source",
				ContentHash:    hex.EncodeToString(assetDigest[:]), MediaType: "text/plain",
				SizeBytes: int64(len(assetPayload)), ClientLabel: "announcement.txt",
			}},
		},
		Now: clock,
	})
	require.NoError(t, err)
	require.Equal(t, broadcast.ID, replay.ID)
	_, err = platform.SubmitPlatformBroadcast(ctx, PlatformBroadcastSubmit{
		PlatformServiceAccountID: created.ID, OperatorUserID: operatorUserID,
		ClientBroadcastID: "broadcast-client-1",
		Plan:              PlatformBroadcastPlan{Title: "Different", Content: "different payload"}, Now: clock,
	})
	require.ErrorIs(t, err, serviceaccountstore.ErrConflict)

	drainPlatformWork(t, ctx, platform, 40)
	completed, err := platform.GetPlatformBroadcast(ctx, created.ID, broadcast.ID)
	require.NoError(t, err)
	require.NotNil(t, completed.BroadcastSeq)
	require.Equal(t, int64(1), *completed.BroadcastSeq)
	require.Equal(t, serviceaccountstore.PlatformBroadcastStateCompleted, completed.State)
	require.Equal(t, int64(2), completed.TargetCount)
	require.Equal(t, int64(2), completed.DeliveredCount)
	messageFanout.mu.Lock()
	committedFanouts := append([]recordedPlatformBroadcastMessageFanout(nil), messageFanout.calls...)
	messageFanout.mu.Unlock()
	require.Len(t, committedFanouts, 2)
	require.ElementsMatch(t, []string{"ws_platform_one", "ws_platform_two"}, []string{
		committedFanouts[0].workspaceID,
		committedFanouts[1].workspaceID,
	})
	for _, fanout := range committedFanouts {
		require.NoError(t, fanout.queryErr)
		require.Equal(t, serviceaccountstore.PlatformDeliveryStateDelivered, fanout.deliveryState)
		require.NotEmpty(t, fanout.delivery.MessageID)
		require.Equal(t, fanout.delivery.MessageID, fanout.appMessage.ID)
		require.NotNil(t, fanout.delivery.AppMessage)
		require.Equal(t, fanout.appMessage.ID, fanout.delivery.AppMessage.ID)
		require.False(t, fanout.delivery.IdempotentReplay)
		require.True(t, fanout.delivery.RecoverableStreamProjection)
		require.Empty(t, fanout.delivery.WakeRecipients)
	}

	for _, workspaceID := range []string{"ws_platform_one", "ws_platform_two"} {
		var messageID, authorMemberID, accountAuthorMemberID, authorType, notifyPrincipalID, idempotencySubjectID, clientMessageID string
		var globalSeq int64
		require.NoError(t, db.QueryRow(ctx, `
SELECT message.id, message.author_member_id, account.author_member_id,
       message.author_type, message.notify_principal_member_id,
       message.idempotency_subject_member_id, message.client_message_id,
       message.workspace_global_seq
FROM scope_messages message
JOIN service_accounts account ON account.message_scope_id = message.scope_id
WHERE message.workspace_id = $1 AND account.platform_service_account_id = $2
`, workspaceID, created.ID).Scan(
			&messageID, &authorMemberID, &accountAuthorMemberID, &authorType,
			&notifyPrincipalID, &idempotencySubjectID, &clientMessageID, &globalSeq,
		))
		require.Equal(t, accountAuthorMemberID, authorMemberID)
		require.Equal(t, "service", authorType)
		require.Equal(t, accountAuthorMemberID, notifyPrincipalID)
		require.Equal(t, accountAuthorMemberID, idempotencySubjectID)
		require.Equal(t, "platform-broadcast:"+broadcast.ID, clientMessageID)
		require.Positive(t, globalSeq)
		require.Equal(t, int64(1), countRows(t, ctx, db, `
SELECT COUNT(*) FROM scope_fanout_outbox
WHERE workspace_id = $1 AND message_id = $2 AND payload IS NOT NULL
`, workspaceID, messageID))
		var fanoutPayloadJSON []byte
		require.NoError(t, db.QueryRow(ctx, `
SELECT payload FROM scope_fanout_outbox
WHERE workspace_id = $1 AND message_id = $2
`, workspaceID, messageID).Scan(&fanoutPayloadJSON))
		fanoutPayload, err := message.UnmarshalScopeFanoutPayload(fanoutPayloadJSON)
		require.NoError(t, err)
		require.True(t, fanoutPayload.Delivery.RecoverableStreamProjection)
		require.Equal(t, int64(1), countRows(t, ctx, db, `
SELECT COUNT(*)
FROM platform_service_account_delivery_assets asset
JOIN platform_service_account_broadcast_deliveries delivery ON delivery.id = asset.delivery_id
JOIN attachments attachment ON attachment.id = asset.local_attachment_id
WHERE delivery.broadcast_id = $1 AND delivery.workspace_id = $2
  AND asset.state = 'materialized' AND attachment.message_id = $3
  AND attachment.filename = 'announcement.txt'
`, broadcast.ID, workspaceID, messageID))
		require.Equal(t, int64(1), countRows(t, ctx, db, `
SELECT COUNT(*) FROM service_account_audit_events
WHERE workspace_id = $1 AND platform_service_account_id = $2
		  AND subject_id = $3 AND caused_by_actor_type = 'platform_operator'
		  AND caused_by_actor_id = $4 AND execution_actor_type = 'internal_service'
`, workspaceID, created.ID, messageID, operatorUserID))
	}
	deletionTarget, err := platform.GetPlatformAccount(ctx, created.ID)
	require.NoError(t, err)
	_, err = platform.DeletePlatformAccount(ctx, PlatformAccountDelete{
		PlatformServiceAccountID: created.ID,
		ExpectedVersion:          deletionTarget.Version,
		OperatorUserID:           operatorUserID,
		CorrelationID:            "corr-platform-delete",
		Now:                      clock.Add(time.Minute),
	})
	require.NoError(t, err)
	clock = clock.Add(time.Minute)
	require.Len(t, deletionObserver.accounts, 2)
	require.ElementsMatch(t, []string{"ws_platform_one", "ws_platform_two"}, []string{
		deletionObserver.accounts[0].WorkspaceID,
		deletionObserver.accounts[1].WorkspaceID,
	})
	require.Equal(t, int64(2), countRows(t, ctx, db, `
SELECT COUNT(*) FROM platform_service_account_deletion_outbox
`), "the all-profile fast path must leave the durable recovery rows for idempotent worker acknowledgement")
	drainPlatformWork(t, ctx, platform, 10)
	require.Len(t, deletionObserver.accounts, 4, "the durable worker may replay the already-applied all-profile fast path")
	require.Equal(t, int64(0), countRows(t, ctx, db, `
SELECT COUNT(*) FROM platform_service_account_deletion_outbox
`))
}

type recordingLocalAccountDeletionObserver struct {
	accounts []serviceaccountstore.Account
}

func (observer *recordingLocalAccountDeletionObserver) LocalAccountsDeleted(_ context.Context, accounts []serviceaccountstore.Account) error {
	observer.accounts = append(observer.accounts, accounts...)
	return nil
}

func TestPlatformBroadcastGenerationFenceExcludesLateWorkspaceAndIncludesItNext(t *testing.T) {
	ctx := context.Background()
	db := materializerTiDB.SerializedRollbackDB(t, ctx)
	now := time.Date(2026, 7, 11, 7, 0, 0, 0, time.UTC)
	seedMaterializerWorkspace(t, ctx, db, "ws_platform_early", now, map[string]string{
		"mem_platform_early_admin": "active",
	})
	operatorUserID := "usr_ws_platform_early_mem_platform_early_admin"
	_, err := db.Exec(ctx, `UPDATE users SET is_admin = TRUE WHERE id = $1`, operatorUserID)
	require.NoError(t, err)
	clock := now
	platform := NewPlatformService(db, PlatformServiceOptions{
		ControlPlane: true, ManagedWorkspaces: owner.NewAllSet(),
		MessageWriter: app.New(db, app.Options{}), Now: func() time.Time { return clock },
	})
	account, err := platform.CreatePlatformAccount(ctx, PlatformAccountCreate{
		ID: "psacct_generation_fence", Name: "Generation News",
		OperatorUserID: operatorUserID, Now: now,
	})
	require.NoError(t, err)
	drainPlatformWork(t, ctx, platform, 20)

	first, err := platform.SubmitPlatformBroadcast(ctx, PlatformBroadcastSubmit{
		PlatformServiceAccountID: account.ID, OperatorUserID: operatorUserID,
		ClientBroadcastID: "generation-first", Plan: PlatformBroadcastPlan{Title: "First", Content: "First"}, Now: clock,
	})
	require.NoError(t, err)
	require.Equal(t, int64(1), first.RequiredProvisioningGeneration)

	seedMaterializerWorkspace(t, ctx, db, "ws_platform_late", now.Add(time.Minute), map[string]string{
		"mem_platform_late_admin": "active",
	})
	drainPlatformWork(t, ctx, platform, 40)
	first, err = platform.GetPlatformBroadcast(ctx, account.ID, first.ID)
	require.NoError(t, err)
	require.Equal(t, serviceaccountstore.PlatformBroadcastStateCompleted, first.State)
	require.Equal(t, int64(1), first.TargetCount)
	require.Equal(t, int64(1), countRows(t, ctx, db, `
SELECT COUNT(*)
FROM platform_service_account_broadcast_deliveries
WHERE broadcast_id = $1 AND workspace_id = 'ws_platform_late'
  AND recipient_eligible = FALSE AND state = 'closed' AND outcome = 'generation_excluded'
`, first.ID))
	require.Equal(t, int64(0), countRows(t, ctx, db, `
SELECT COUNT(*)
FROM scope_messages message
JOIN service_accounts account ON account.message_scope_id = message.scope_id
WHERE account.workspace_id = 'ws_platform_late'
  AND account.platform_service_account_id = $1
`, account.ID))

	second, err := platform.SubmitPlatformBroadcast(ctx, PlatformBroadcastSubmit{
		PlatformServiceAccountID: account.ID, OperatorUserID: operatorUserID,
		ClientBroadcastID: "generation-second", Plan: PlatformBroadcastPlan{Title: "Second", Content: "Second"}, Now: clock,
	})
	require.NoError(t, err)
	require.Equal(t, int64(2), second.RequiredProvisioningGeneration)
	drainPlatformWork(t, ctx, platform, 30)
	second, err = platform.GetPlatformBroadcast(ctx, account.ID, second.ID)
	require.NoError(t, err)
	require.Equal(t, serviceaccountstore.PlatformBroadcastStateCompleted, second.State)
	require.Equal(t, int64(2), second.TargetCount)
	require.Equal(t, int64(2), second.DeliveredCount)
	require.Equal(t, int64(1), countRows(t, ctx, db, `
SELECT COUNT(*)
FROM scope_messages message
JOIN service_accounts account ON account.message_scope_id = message.scope_id
WHERE account.workspace_id = 'ws_platform_late'
  AND account.platform_service_account_id = $1
  AND message.client_message_id = $2
`, account.ID, "platform-broadcast:"+second.ID))
	var lateCursor int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT last_applied_broadcast_seq
FROM platform_service_account_workspace_instances
WHERE platform_service_account_id = $1 AND workspace_id = 'ws_platform_late'
`, account.ID).Scan(&lateCursor))
	require.Equal(t, int64(2), lateCursor)
}

func TestPlatformBroadcastDeliveryHonorsWorkspacePlacementOwnership(t *testing.T) {
	ctx := context.Background()
	db := materializerTiDB.SerializedRollbackDB(t, ctx)
	now := time.Date(2026, 7, 11, 9, 0, 0, 0, time.UTC)
	for _, fixture := range []struct {
		workspaceID string
		memberID    string
	}{
		{workspaceID: "ws_placement_a", memberID: "mem_placement_a_admin"},
		{workspaceID: "ws_placement_b", memberID: "mem_placement_b_admin"},
	} {
		seedMaterializerWorkspace(t, ctx, db, fixture.workspaceID, now, map[string]string{fixture.memberID: "active"})
	}
	operatorUserID := "usr_ws_placement_a_mem_placement_a_admin"
	_, err := db.Exec(ctx, `UPDATE users SET is_admin = TRUE WHERE id = $1`, operatorUserID)
	require.NoError(t, err)
	_, err = db.Exec(ctx, `
INSERT INTO workspace_placements (workspace_id, workspace_group_id, state, placement_epoch, created_at, updated_at)
VALUES
  ('ws_placement_a', 'workspace-a', 'active', 3, $1, $1),
  ('ws_placement_b', 'workspace-b', 'active', 3, $1, $1)
`, now)
	require.NoError(t, err)
	appStore := app.New(db, app.Options{})
	bootstrap := NewPlatformService(db, PlatformServiceOptions{
		ControlPlane: true, ManagedWorkspaces: owner.NewAllSet(),
		MessageWriter: appStore, Now: func() time.Time { return now },
	})
	account, err := bootstrap.CreatePlatformAccount(ctx, PlatformAccountCreate{
		ID: "psacct_placement", Name: "Placement News",
		OperatorUserID: operatorUserID, Now: now,
	})
	require.NoError(t, err)
	drainPlatformWork(t, ctx, bootstrap, 30)

	central := NewPlatformService(db, PlatformServiceOptions{
		ControlPlane: true, ManagedWorkspaces: owner.NewEmptySet(),
		MessageWriter: appStore, Now: func() time.Time { return now },
	})
	broadcast, err := central.SubmitPlatformBroadcast(ctx, PlatformBroadcastSubmit{
		PlatformServiceAccountID: account.ID, OperatorUserID: operatorUserID,
		ClientBroadcastID: "placement-broadcast", Plan: PlatformBroadcastPlan{Title: "Owned execution", Content: "Owned execution"}, Now: now,
	})
	require.NoError(t, err)
	for round := 0; round < 5; round++ {
		_, err := central.ProcessOnce(ctx)
		require.NoError(t, err)
	}
	require.Equal(t, int64(2), countRows(t, ctx, db, `
SELECT COUNT(*) FROM platform_service_account_broadcast_deliveries
WHERE broadcast_id = $1 AND state = 'queued'
`, broadcast.ID))
	pending, err := central.GetPlatformBroadcast(ctx, account.ID, broadcast.ID)
	require.NoError(t, err)
	_, err = central.ProcessOnce(ctx)
	require.NoError(t, err)
	stillPending, err := central.GetPlatformBroadcast(ctx, account.ID, broadcast.ID)
	require.NoError(t, err)
	require.Equal(t, pending.Version, stillPending.Version, "an idle aggregate pass must not churn the broadcast version")

	workerA := NewPlatformService(db, PlatformServiceOptions{
		ManagedWorkspaces: owner.NewStaticSet([]string{"ws_placement_a"}),
		MessageWriter:     appStore, Now: func() time.Time { return now },
	})
	result, err := workerA.ProcessOnce(ctx)
	require.NoError(t, err)
	require.Equal(t, 1, result.DeliveredMessages)
	require.Equal(t, int64(1), countRows(t, ctx, db, `
SELECT COUNT(*) FROM platform_service_account_broadcast_deliveries
WHERE broadcast_id = $1 AND workspace_id = 'ws_placement_a' AND state = 'delivered'
`, broadcast.ID))
	require.Equal(t, int64(1), countRows(t, ctx, db, `
SELECT COUNT(*) FROM platform_service_account_broadcast_deliveries
WHERE broadcast_id = $1 AND workspace_id = 'ws_placement_b' AND state = 'queued'
`, broadcast.ID))
	_, err = db.Exec(ctx, `
UPDATE workspace_placements
SET workspace_group_id = 'workspace-b-next', placement_epoch = 4, updated_at = $2
WHERE workspace_id = $1
`, "ws_placement_b", now.Add(time.Second))
	require.NoError(t, err)

	workerB := NewPlatformService(db, PlatformServiceOptions{
		ManagedWorkspaces: owner.NewStaticSet([]string{"ws_placement_b"}),
		MessageWriter:     appStore, Now: func() time.Time { return now },
	})
	result, err = workerB.ProcessOnce(ctx)
	require.NoError(t, err)
	require.Equal(t, 1, result.DeliveredMessages)
	var settledPlacementEpoch int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT placement_epoch
FROM platform_service_account_broadcast_deliveries
WHERE broadcast_id = $1 AND workspace_id = 'ws_placement_b'
`, broadcast.ID).Scan(&settledPlacementEpoch))
	require.Equal(t, int64(4), settledPlacementEpoch)
	_, err = central.ProcessOnce(ctx)
	require.NoError(t, err)
	broadcast, err = central.GetPlatformBroadcast(ctx, account.ID, broadcast.ID)
	require.NoError(t, err)
	require.Equal(t, serviceaccountstore.PlatformBroadcastStateCompleted, broadcast.State)
}

func TestPlatformAccountDeletionOutboxRoutesTombstonesToWorkspaceOwners(t *testing.T) {
	ctx := context.Background()
	db := materializerTiDB.SerializedRollbackDB(t, ctx)
	now := time.Date(2026, 7, 11, 10, 0, 0, 0, time.UTC)
	for _, fixture := range []struct {
		workspaceID string
		memberID    string
	}{
		{workspaceID: "ws_delete_owner_a", memberID: "mem_delete_owner_a_admin"},
		{workspaceID: "ws_delete_owner_b", memberID: "mem_delete_owner_b_admin"},
	} {
		seedMaterializerWorkspace(t, ctx, db, fixture.workspaceID, now, map[string]string{fixture.memberID: "active"})
	}
	operatorUserID := "usr_ws_delete_owner_a_mem_delete_owner_a_admin"
	_, err := db.Exec(ctx, `UPDATE users SET is_admin = TRUE WHERE id = $1`, operatorUserID)
	require.NoError(t, err)

	bootstrap := NewPlatformService(db, PlatformServiceOptions{
		ControlPlane: true, ManagedWorkspaces: owner.NewAllSet(),
		MessageWriter: app.New(db, app.Options{}), Now: func() time.Time { return now },
	})
	account, err := bootstrap.CreatePlatformAccount(ctx, PlatformAccountCreate{
		ID: "psacct_delete_owner", Name: "Delete Owner News",
		OperatorUserID: operatorUserID, Now: now,
	})
	require.NoError(t, err)
	drainPlatformWork(t, ctx, bootstrap, 30)

	expected := make(map[string]serviceaccountstore.Account, 2)
	for _, workspaceID := range []string{"ws_delete_owner_a", "ws_delete_owner_b"} {
		var local serviceaccountstore.Account
		require.NoError(t, db.QueryRow(ctx, `
SELECT id, principal_id, workspace_id, platform_service_account_id, message_scope_id
FROM service_accounts
WHERE workspace_id = $1 AND platform_service_account_id = $2
`, workspaceID, account.ID).Scan(
			&local.ID,
			&local.PrincipalID,
			&local.WorkspaceID,
			&local.PlatformServiceAccountID,
			&local.MessageScopeID,
		))
		expected[workspaceID] = local
	}

	centralObserver := &recordingLocalAccountDeletionObserver{}
	central := NewPlatformService(db, PlatformServiceOptions{
		ControlPlane: true, ManagedWorkspaces: owner.NewEmptySet(),
		LocalAccountDeletionObserver: centralObserver,
		Now:                          func() time.Time { return now },
	})
	current, err := central.GetPlatformAccount(ctx, account.ID)
	require.NoError(t, err)
	_, err = central.DeletePlatformAccount(ctx, PlatformAccountDelete{
		PlatformServiceAccountID: account.ID,
		ExpectedVersion:          current.Version,
		OperatorUserID:           operatorUserID,
		Now:                      now,
	})
	require.NoError(t, err)
	require.Empty(t, centralObserver.accounts, "the central profile must not run owner-local post-commit effects even when an observer is accidentally present")
	require.Equal(t, int64(2), countRows(t, ctx, db, `
SELECT COUNT(*)
FROM platform_service_account_deletion_outbox
WHERE platform_service_account_id = $1
`, account.ID))

	result, err := central.ProcessOnce(ctx)
	require.NoError(t, err)
	require.Zero(t, result.DeliveredPlatformDeletions)
	require.Equal(t, int64(2), countRows(t, ctx, db, `
SELECT COUNT(*) FROM platform_service_account_deletion_outbox
`), "a central process without workspace ownership must not acknowledge owner-local effects")

	observerA := &recordingLocalAccountDeletionObserver{}
	workerA := NewPlatformService(db, PlatformServiceOptions{
		ManagedWorkspaces:            owner.NewStaticSet([]string{"ws_delete_owner_a"}),
		LocalAccountDeletionObserver: observerA,
		Now:                          func() time.Time { return now },
	})
	result, err = workerA.ProcessOnce(ctx)
	require.NoError(t, err)
	require.Equal(t, 1, result.DeliveredPlatformDeletions)
	require.Len(t, observerA.accounts, 1)
	requireDeletionTombstoneAccount(t, expected["ws_delete_owner_a"], observerA.accounts[0])
	require.Equal(t, int64(1), countRows(t, ctx, db, `
SELECT COUNT(*) FROM platform_service_account_deletion_outbox
WHERE workspace_id = 'ws_delete_owner_b'
`))

	observerB := &recordingLocalAccountDeletionObserver{}
	workerB := NewPlatformService(db, PlatformServiceOptions{
		ManagedWorkspaces:            owner.NewStaticSet([]string{"ws_delete_owner_b"}),
		LocalAccountDeletionObserver: observerB,
		Now:                          func() time.Time { return now },
	})
	result, err = workerB.ProcessOnce(ctx)
	require.NoError(t, err)
	require.Equal(t, 1, result.DeliveredPlatformDeletions)
	require.Len(t, observerB.accounts, 1)
	requireDeletionTombstoneAccount(t, expected["ws_delete_owner_b"], observerB.accounts[0])
	require.Equal(t, int64(0), countRows(t, ctx, db, `
SELECT COUNT(*) FROM platform_service_account_deletion_outbox
`))
}

func TestPlatformAccountDeletionOutboxRetriesObserverBeforeAck(t *testing.T) {
	ctx := context.Background()
	db := materializerTiDB.SerializedRollbackDB(t, ctx)
	now := time.Date(2026, 7, 11, 11, 0, 0, 0, time.UTC)
	clock := now
	seedMaterializerWorkspace(t, ctx, db, "ws_delete_retry", now, map[string]string{
		"mem_delete_retry_admin": "active",
	})
	operatorUserID := "usr_ws_delete_retry_mem_delete_retry_admin"
	_, err := db.Exec(ctx, `UPDATE users SET is_admin = TRUE WHERE id = $1`, operatorUserID)
	require.NoError(t, err)

	bootstrap := NewPlatformService(db, PlatformServiceOptions{
		ControlPlane: true, ManagedWorkspaces: owner.NewAllSet(),
		MessageWriter: app.New(db, app.Options{}), Now: func() time.Time { return clock },
	})
	account, err := bootstrap.CreatePlatformAccount(ctx, PlatformAccountCreate{
		ID: "psacct_delete_retry", Name: "Delete Retry News",
		OperatorUserID: operatorUserID, Now: now,
	})
	require.NoError(t, err)
	drainPlatformWork(t, ctx, bootstrap, 20)

	var expected serviceaccountstore.Account
	require.NoError(t, db.QueryRow(ctx, `
SELECT id, principal_id, workspace_id, platform_service_account_id, message_scope_id
FROM service_accounts
WHERE workspace_id = 'ws_delete_retry' AND platform_service_account_id = $1
`, account.ID).Scan(
		&expected.ID,
		&expected.PrincipalID,
		&expected.WorkspaceID,
		&expected.PlatformServiceAccountID,
		&expected.MessageScopeID,
	))
	central := NewPlatformService(db, PlatformServiceOptions{
		ControlPlane: true, ManagedWorkspaces: owner.NewEmptySet(),
		Now: func() time.Time { return clock },
	})
	current, err := central.GetPlatformAccount(ctx, account.ID)
	require.NoError(t, err)
	_, err = central.DeletePlatformAccount(ctx, PlatformAccountDelete{
		PlatformServiceAccountID: account.ID,
		ExpectedVersion:          current.Version,
		OperatorUserID:           operatorUserID,
		Now:                      clock,
	})
	require.NoError(t, err)

	observer := &failOnceLocalAccountDeletionObserver{}
	worker := NewPlatformService(db, PlatformServiceOptions{
		ManagedWorkspaces:            owner.NewStaticSet([]string{"ws_delete_retry"}),
		LocalAccountDeletionObserver: observer,
		Now:                          func() time.Time { return clock },
	})
	result, err := worker.ProcessOnce(ctx)
	require.NoError(t, err)
	require.Equal(t, 1, result.RetriedPlatformDeletions)
	require.Zero(t, result.DeliveredPlatformDeletions)
	require.Equal(t, int64(1), countRows(t, ctx, db, `
SELECT COUNT(*)
FROM platform_service_account_deletion_outbox
WHERE platform_service_account_id = $1
  AND attempt_count = 1
  AND lease_token IS NULL
  AND lease_expires_at IS NULL
  AND last_error <> ''
  AND next_attempt_at > $2
`, account.ID, now))

	clock = clock.Add(2 * time.Second)
	result, err = worker.ProcessOnce(ctx)
	require.NoError(t, err)
	require.Equal(t, 1, result.DeliveredPlatformDeletions)
	require.Len(t, observer.accounts, 2)
	for _, observed := range observer.accounts {
		requireDeletionTombstoneAccount(t, expected, observed)
	}
	require.Equal(t, int64(0), countRows(t, ctx, db, `
SELECT COUNT(*) FROM platform_service_account_deletion_outbox
`))
}

type failOnceLocalAccountDeletionObserver struct {
	accounts []serviceaccountstore.Account
	failed   bool
}

func (observer *failOnceLocalAccountDeletionObserver) LocalAccountsDeleted(_ context.Context, accounts []serviceaccountstore.Account) error {
	observer.accounts = append(observer.accounts, accounts...)
	if !observer.failed {
		observer.failed = true
		return errors.New("injected transient service account deletion observer failure")
	}
	return nil
}

func requireDeletionTombstoneAccount(t testing.TB, expected serviceaccountstore.Account, actual serviceaccountstore.Account) {
	t.Helper()
	require.Equal(t, expected.ID, actual.ID)
	require.Equal(t, expected.PrincipalID, actual.PrincipalID)
	require.Equal(t, expected.WorkspaceID, actual.WorkspaceID)
	require.Equal(t, expected.PlatformServiceAccountID, actual.PlatformServiceAccountID)
	require.Equal(t, expected.MessageScopeID, actual.MessageScopeID)
}

func drainPlatformWork(t testing.TB, ctx context.Context, platform *PlatformService, maxRounds int) {
	t.Helper()
	idleRounds := 0
	for round := 0; round < maxRounds; round++ {
		result, err := platform.ProcessOnce(ctx)
		require.NoError(t, err)
		if result.hasWork() {
			idleRounds = 0
			continue
		}
		idleRounds++
		if idleRounds >= 2 {
			return
		}
	}
	t.Fatalf("platform Service Account worker did not become idle within %d rounds", maxRounds)
}