package app

import (
	"context"
	"github.com/tidbcloud/AgentTeam/internal/store/app/registry"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/coredb/testkit"
	installationstore "github.com/tidbcloud/AgentTeam/internal/store/app/installation"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/serviceactor"
)

func TestAttributedChannelAppUsesManagerOrWorkspaceAdmin(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	admin := seedAppAdminFixture(t, pool, "bot_channel_attributed_admin", now)
	creator := seedWorkspaceMemberFixture(t, pool, admin.WorkspaceID, "bot_channel_attributed_creator", now)
	peer := seedWorkspaceMemberFixture(t, pool, admin.WorkspaceID, "bot_channel_attributed_peer", now)
	channelOwner := seedWorkspaceMemberFixture(t, pool, admin.WorkspaceID, "bot_channel_attributed_owner", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, admin, "channel-attributed", now)
	_, err := appStore.AddAppManagerForUser(
		ctx, admin.UserID, admin.WorkspaceID, installation.ID, creator.MemberID, now.Add(time.Second),
	)
	require.NoError(t, err)

	channelID := "chn_bot_channel_attributed"
	for _, actor := range []appAdminFixture{channelOwner, admin, creator, peer} {
		seedChannelParticipantProjectionForTests(t, pool, admin.WorkspaceID, channelID, channelID, "Attributed App channel", "private", channelOwner.MemberID, actor.MemberID, now)
	}

	requireChannelAddablePeopleContainsService(t, ctx, appStore, creator.UserID, admin.WorkspaceID, channelID, installation.ID)
	requireChannelAddablePeopleContainsService(t, ctx, appStore, admin.UserID, admin.WorkspaceID, channelID, installation.ID)
	requireChannelAddablePeopleOmitsService(t, ctx, appStore, peer.UserID, admin.WorkspaceID, channelID, installation.ID)
	_, err = appStore.AddChannelServiceForUserWithEvent(ctx, peer.UserID, admin.WorkspaceID, channelID, installation.MemberID)
	require.ErrorIs(t, err, ErrNotFound)

	_, err = appStore.AddChannelServiceForUserWithEvent(ctx, creator.UserID, admin.WorkspaceID, channelID, installation.MemberID)
	require.NoError(t, err)
	requireChannelServiceCanRemove(t, ctx, appStore, creator.UserID, admin.WorkspaceID, channelID, installation.MemberID, true)
	requireChannelServiceCanRemove(t, ctx, appStore, admin.UserID, admin.WorkspaceID, channelID, installation.MemberID, true)
	requireChannelServiceCanRemove(t, ctx, appStore, peer.UserID, admin.WorkspaceID, channelID, installation.MemberID, false)
	requireChannelServiceCanRemove(t, ctx, appStore, channelOwner.UserID, admin.WorkspaceID, channelID, installation.MemberID, false)
	_, err = appStore.RemoveChannelServiceForUserWithEvent(ctx, peer.UserID, admin.WorkspaceID, channelID, installation.MemberID)
	require.ErrorIs(t, err, ErrForbidden)
	_, err = appStore.RemoveChannelServiceForUserWithEvent(ctx, creator.UserID, admin.WorkspaceID, channelID, installation.MemberID)
	require.NoError(t, err)

	_, err = appStore.AddChannelServiceForUserWithEvent(ctx, admin.UserID, admin.WorkspaceID, channelID, installation.MemberID)
	require.NoError(t, err)
	_, err = appStore.RemoveChannelServiceForUserWithEvent(ctx, admin.UserID, admin.WorkspaceID, channelID, installation.MemberID)
	require.NoError(t, err)
}

func TestPublicAttributedChannelAppAllowsOrdinaryChannelInvite(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	admin := seedAppAdminFixture(t, pool, "bot_channel_public_admin", now)
	channelOwner := seedWorkspaceMemberFixture(t, pool, admin.WorkspaceID, "bot_channel_public_owner", now)
	peer := seedWorkspaceMemberFixture(t, pool, admin.WorkspaceID, "bot_channel_public_peer", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, admin, "channel-public", now)

	channelIDs := []string{
		"chn_bot_channel_public_single",
		"chn_bot_channel_public_batch",
		"chn_bot_channel_public_private_again",
	}
	for _, channelID := range channelIDs {
		seedChannelParticipantProjectionForTests(t, pool, admin.WorkspaceID, channelID, channelID, "Public App channel", "private", channelOwner.MemberID, channelOwner.MemberID, now)
		seedChannelParticipantProjectionForTests(t, pool, admin.WorkspaceID, channelID, channelID, "Public App channel", "private", channelOwner.MemberID, peer.MemberID, now)
	}

	requireChannelAddablePeopleOmitsService(t, ctx, appStore, peer.UserID, admin.WorkspaceID, channelIDs[0], installation.ID)
	_, err := appStore.AddChannelServiceForUserWithEvent(ctx, peer.UserID, admin.WorkspaceID, channelIDs[0], installation.MemberID)
	require.ErrorIs(t, err, ErrNotFound)

	mustExec(t, pool, `
UPDATE extension_catalog_attributions
SET catalog_visibility = $3, updated_at = $4
WHERE workspace_id = $1 AND service_definition_id = $2
`, admin.WorkspaceID, installation.AppID, "public", now.Add(time.Second))

	page, err := appStore.ListChannelAddablePeopleForUser(ctx, peer.UserID, admin.WorkspaceID, channelIDs[0], ChannelAddablePeopleOptions{Limit: 100})
	require.NoError(t, err)
	var candidate *AppInstallationView
	for idx := range page.Services {
		if strings.TrimSpace(page.Services[idx].ID) == strings.TrimSpace(installation.ID) {
			candidate = &page.Services[idx]
			break
		}
	}
	require.NotNil(t, candidate)
	require.Equal(t, "public", candidate.CatalogVisibility)
	require.False(t, candidate.CanManageApp)
	require.Nil(t, candidate.Config)
	require.NotNil(t, candidate.Connections)
	require.Empty(t, candidate.Connections)
	require.Empty(t, candidate.Grants)
	require.Nil(t, candidate.RuntimeCapability)

	_, err = appStore.AddChannelServiceForUserWithEvent(ctx, peer.UserID, admin.WorkspaceID, channelIDs[0], installation.MemberID)
	require.NoError(t, err)
	assertActiveAppTargetBinding(t, ctx, pool, admin.WorkspaceID, installation.ID, installation.MemberID, channelIDs[0], "channel", channelIDs[0], channelIDs[0])
	requireChannelServiceCanRemove(t, ctx, appStore, peer.UserID, admin.WorkspaceID, channelIDs[0], installation.MemberID, true)
	requireChannelServiceCanRemove(t, ctx, appStore, channelOwner.UserID, admin.WorkspaceID, channelIDs[0], installation.MemberID, true)
	_, err = appStore.RemoveChannelServiceForUserWithEvent(ctx, peer.UserID, admin.WorkspaceID, channelIDs[0], installation.MemberID)
	require.NoError(t, err)

	results, err := appStore.AddChannelMembersForUserWithEvents(ctx, peer.UserID, admin.WorkspaceID, channelIDs[1], []message.ChannelMemberAddRequest{{ServiceMemberID: installation.MemberID}})
	require.NoError(t, err)
	require.Len(t, results, 1)
	assertActiveAppTargetBinding(t, ctx, pool, admin.WorkspaceID, installation.ID, installation.MemberID, channelIDs[1], "channel", channelIDs[1], channelIDs[1])

	mustExec(t, pool, `
UPDATE extension_catalog_attributions
SET catalog_visibility = $3, updated_at = $4
WHERE workspace_id = $1 AND service_definition_id = $2
`, admin.WorkspaceID, installation.AppID, "private", now.Add(2*time.Second))
	requireChannelAddablePeopleOmitsService(t, ctx, appStore, peer.UserID, admin.WorkspaceID, channelIDs[2], installation.ID)
	_, err = appStore.AddChannelMembersForUserWithEvents(ctx, peer.UserID, admin.WorkspaceID, channelIDs[2], []message.ChannelMemberAddRequest{{ServiceMemberID: installation.MemberID}})
	require.ErrorIs(t, err, ErrNotFound)
	requireChannelServiceCanRemove(t, ctx, appStore, channelOwner.UserID, admin.WorkspaceID, channelIDs[1], installation.MemberID, false)
	_, err = appStore.RemoveChannelServiceForUserWithEvent(ctx, channelOwner.UserID, admin.WorkspaceID, channelIDs[1], installation.MemberID)
	require.ErrorIs(t, err, ErrForbidden)
}

func TestChannelAppMissingAttributionFailsClosedExceptExactProducts(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	admin := seedAppAdminFixture(t, pool, "bot_channel_unattributed_admin", now)
	peer := seedWorkspaceMemberFixture(t, pool, admin.WorkspaceID, "bot_channel_unattributed_peer", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, admin, "channel-unattributed", now)
	channelID := "chn_bot_channel_unattributed"
	for _, actor := range []appAdminFixture{admin, peer} {
		seedChannelParticipantProjectionForTests(
			t, pool, admin.WorkspaceID, channelID, channelID, "Unattributed App channel", "private", admin.MemberID, actor.MemberID, now,
		)
	}

	mustExec(t, pool, `
DELETE FROM extension_catalog_attributions
WHERE workspace_id = $1 AND service_definition_id = $2
`, admin.WorkspaceID, installation.AppID)
	requireChannelAddablePeopleOmitsService(t, ctx, appStore, peer.UserID, admin.WorkspaceID, channelID, installation.ID)
	requireChannelAddablePeopleOmitsService(t, ctx, appStore, admin.UserID, admin.WorkspaceID, channelID, installation.ID)
	_, err := appStore.AddChannelServiceForUserWithEvent(
		ctx, peer.UserID, admin.WorkspaceID, channelID, installation.MemberID,
	)
	require.ErrorIs(t, err, ErrNotFound)
	_, err = appStore.AddChannelServiceForUserWithEvent(
		ctx, admin.UserID, admin.WorkspaceID, channelID, installation.MemberID,
	)
	require.ErrorIs(t, err, ErrNotFound)

	adminMember, err := appStore.requireWorkspaceAccess(ctx, admin.UserID, admin.WorkspaceID)
	require.NoError(t, err)
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		authorization, lockErr := lockChannelAppManagerAuthorizationForUpdate(
			ctx, tx, admin.WorkspaceID, registry.GitHubAppID, adminMember,
		)
		if lockErr != nil {
			return lockErr
		}
		require.False(t, authorization.HasAttribution)
		require.True(t, authorization.IsProduct, "the exact GitHub pair retains the product exception")
		if _, updateErr := tx.Exec(ctx, `
UPDATE service_definitions
SET slug = 'github-spoof'
WHERE id = $1 AND service_kind = $2
`, registry.GitHubAppID, serviceactor.KindApp); updateErr != nil {
			return updateErr
		}
		authorization, lockErr = lockChannelAppManagerAuthorizationForUpdate(
			ctx, tx, admin.WorkspaceID, registry.GitHubAppID, adminMember,
		)
		if lockErr != nil {
			return lockErr
		}
		require.False(t, authorization.IsProduct, "a product ID with the wrong slug must fail closed")
		return nil
	}))
}

func TestRemoveAppFromAllChannelsForVisibilityChangeCleansDependentMentionTargets(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_visibility_channel_cleanup", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, fixture, "visibility-cleanup", now)
	channelZ := "chn_bot_visibility_cleanup_z"
	channelA := "chn_bot_visibility_cleanup_a"
	for _, channelID := range []string{channelZ, channelA} {
		seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, channelID, channelID, "App visibility cleanup", "private", fixture.MemberID, fixture.MemberID, now)
		_, err := appStore.AddChannelServiceForUserWithEvent(ctx, fixture.UserID, fixture.WorkspaceID, channelID, installation.MemberID)
		require.NoError(t, err)
	}
	// An archived membership is still joined and can be unarchived later. The
	// destructive Private transition must remove it too, rather than leaving a
	// dormant channel membership that can reappear.
	mustExec(t, pool, `
UPDATE scope_memberships
SET archived_at = $3, updated_at = $3
WHERE scope_id = $1
  AND member_id = $2
  AND visibility = 'public'
  AND removed_at IS NULL
`, channelA, installation.MemberID, now.Add(time.Millisecond))

	threadID := "thr_bot_visibility_cleanup"
	mustExec(t, pool, `
INSERT INTO scopes (
    id, workspace_id, type, parent_scope_id, root_message_id, root_message_short_id,
    identity_key, status, created_by_member_id, created_at, updated_at
) VALUES (
    $1, $2, 'thread', $3, $4, $5,
    CONCAT('thread/', $3, '/', $4), 'active', $6, $7, $7
)
`, threadID, fixture.WorkspaceID, channelZ, "msg_bot_visibility_cleanup_root", "rootvis1", fixture.MemberID, now)
	_, err := appStore.bindAppInstallationTargetForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, threadID, AppMentionCreatedViaTarget)
	require.NoError(t, err)

	taskID := "tsk_bot_visibility_cleanup"
	seedTaskTargetForAppBindingTestsOnChannel(t, pool, fixture, channelZ, taskID, now)
	_, err = appStore.bindAppInstallationTargetForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, "task:"+taskID, AppMentionCreatedViaTarget)
	require.NoError(t, err)

	peer := seedWorkspaceMemberFixture(t, pool, fixture.WorkspaceID, "bot_visibility_cleanup_dm_peer", now)
	var dm message.DMChannel
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		var openErr error
		dm, openErr = message.OpenDMChannelForMember(ctx, tx, message.MemberActor{WorkspaceID: fixture.WorkspaceID, MemberID: fixture.MemberID}, peer.MemberID)
		return openErr
	}))
	_, err = appStore.bindAppInstallationTargetForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, dm.ID, AppMentionCreatedViaTarget)
	require.NoError(t, err)
	actionSubscriptionID := seedVisibilityCleanupActionSubscription(
		t, ctx, pool, appStore, fixture, installation, channelZ, now,
	)
	requireAppTargetSingleInt(t, ctx, pool, 1, `
SELECT COUNT(*) FROM service_event_subscriptions
WHERE id = $1 AND subscription_kind = $2 AND status = 'active' AND removed_at IS NULL
`, actionSubscriptionID, serviceactor.SubscriptionKindAppNotificationAction)

	for _, scopeID := range []string{channelZ, threadID, taskID, dm.ID} {
		mustExec(t, pool, `
INSERT INTO scope_notification_settings (scope_id, member_id, state, created_at, updated_at)
VALUES ($1, $2, 'no-wake', $3, $3)
ON DUPLICATE KEY UPDATE state = VALUES(state), updated_at = VALUES(updated_at)
`, scopeID, installation.MemberID, now)
	}
	// Simulate a partially completed historical channel removal: the parent
	// membership is already gone, but its direct/thread/task runtime and child
	// memberships remain. A private -> private retry must repair that residue
	// from installation/binding provenance rather than returning early.
	mustExec(t, pool, `
UPDATE scope_memberships
SET removed_at = $3, updated_at = $3
WHERE scope_id = $1
  AND member_id = $2
  AND visibility = 'public'
  AND removed_at IS NULL
`, channelZ, installation.MemberID, now.Add(2*time.Millisecond))

	var messagesBefore int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*) FROM scope_messages WHERE scope_id IN ($1, $2)
`, channelA, channelZ).Scan(&messagesBefore))

	var removedChannelIDs []string
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		if err := installationstore.LockAppInstallationMutationScope(ctx, tx, installation.AppID); err != nil {
			return err
		}
		var visibility string
		if err := tx.QueryRow(ctx, `
SELECT catalog_visibility
FROM extension_catalog_attributions
WHERE workspace_id = $1 AND service_definition_id = $2
FOR UPDATE
`, fixture.WorkspaceID, installation.AppID).Scan(&visibility); err != nil {
			return err
		}
		require.Equal(t, "private", visibility, "target-private cleanup must repair residual channel state even when visibility is already private")
		currentInstallation, _, found, isApp, err := loadActiveAppServiceInstallationByServiceMemberID(ctx, tx, fixture.WorkspaceID, installation.MemberID, true)
		if err != nil {
			return err
		}
		if !found || !isApp {
			return ErrNotFound
		}
		removedChannelIDs, err = removeAppFromAllChannelsForVisibilityChangeInTx(
			ctx, tx, fixture.WorkspaceID, fixture.MemberID, currentInstallation, installation.MemberID, now.Add(time.Second),
		)
		return err
	}))
	require.Equal(t, []string{channelA}, removedChannelIDs)

	requireAppTargetSingleInt(t, ctx, pool, 0, `
SELECT COUNT(*)
FROM scope_memberships
WHERE member_id = $1
  AND scope_id IN ($2, $3)
  AND visibility = 'public'
  AND removed_at IS NULL
  AND effective_to_seq IS NULL
`, installation.MemberID, channelA, channelZ)
	requireAppTargetSingleInt(t, ctx, pool, 0, `
SELECT COUNT(*)
FROM scope_memberships
WHERE member_id = $1
  AND scope_id IN ($2, $3)
  AND visibility = 'public'
  AND removed_at IS NULL
  AND effective_to_seq IS NULL
`, installation.MemberID, threadID, taskID)
	requireAppTargetSingleInt(t, ctx, pool, 1, `
SELECT COUNT(*)
FROM scope_memberships
WHERE member_id = $1
  AND scope_id = $2
  AND visibility = 'public'
  AND removed_at IS NULL
  AND effective_to_seq IS NULL
`, installation.MemberID, dm.ID)
	requireAppTargetSingleInt(t, ctx, pool, 0, `
SELECT COUNT(*)
FROM service_event_subscriptions
WHERE workspace_id = $1
  AND service_installation_id = $2
  AND permission_check_scope_id IN ($3, $4)
  AND removed_at IS NULL
`, fixture.WorkspaceID, installation.ID, channelA, channelZ)
	requireAppTargetSingleInt(t, ctx, pool, 0, `
SELECT COUNT(*) FROM service_event_subscriptions
WHERE id = $1 AND subscription_kind = $2 AND status = 'active' AND removed_at IS NULL
`, actionSubscriptionID, serviceactor.SubscriptionKindAppNotificationAction)
	requireAppTargetSingleInt(t, ctx, pool, 4, `
SELECT COUNT(*)
FROM service_target_bindings binding
WHERE binding.workspace_id = $1
  AND binding.service_installation_id = $2
  AND binding.permission_check_scope_id IN ($3, $4)
  AND binding.status = 'disabled'
  AND binding.active_binding_key IS NULL
  AND binding.removed_at IS NULL
`, fixture.WorkspaceID, installation.ID, channelA, channelZ)
	requireAppTargetSingleInt(t, ctx, pool, 1, `
SELECT COUNT(*)
FROM service_event_subscriptions
WHERE workspace_id = $1
  AND service_installation_id = $2
  AND subscription_kind = $3
  AND target_kind = 'dm'
  AND target_ref = $4
  AND removed_at IS NULL
`, fixture.WorkspaceID, installation.ID, serviceactor.SubscriptionKindAppMention, dm.ID)
	requireAppTargetSingleInt(t, ctx, pool, 1, `
SELECT COUNT(*)
FROM service_target_bindings
WHERE workspace_id = $1
  AND service_installation_id = $2
  AND target_kind = 'dm'
  AND target_ref = $3
  AND status = 'active'
  AND active_binding_key IS NOT NULL
  AND removed_at IS NULL
`, fixture.WorkspaceID, installation.ID, dm.ID)
	requireAppTargetSingleInt(t, ctx, pool, 0, `
SELECT COUNT(*)
FROM scope_notification_settings
WHERE member_id = $1 AND scope_id IN ($2, $3, $4)
`, installation.MemberID, channelZ, threadID, taskID)
	requireAppTargetSingleInt(t, ctx, pool, 1, `
SELECT COUNT(*)
FROM scope_notification_settings
WHERE member_id = $1 AND scope_id = $2
`, installation.MemberID, dm.ID)

	var messagesAfter int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*) FROM scope_messages WHERE scope_id IN ($1, $2)
`, channelA, channelZ).Scan(&messagesAfter))
	require.Equal(t, messagesBefore, messagesAfter, "visibility cleanup must not append channel system messages under a reversed scope lock")

	var secondRemovedChannelIDs []string
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		if err := installationstore.LockAppInstallationMutationScope(ctx, tx, installation.AppID); err != nil {
			return err
		}
		currentInstallation, _, found, isApp, err := loadActiveAppServiceInstallationByServiceMemberID(ctx, tx, fixture.WorkspaceID, installation.MemberID, true)
		if err != nil {
			return err
		}
		if !found || !isApp {
			return ErrNotFound
		}
		secondRemovedChannelIDs, err = removeAppFromAllChannelsForVisibilityChangeInTx(
			ctx, tx, fixture.WorkspaceID, fixture.MemberID, currentInstallation, installation.MemberID, now.Add(2*time.Second),
		)
		return err
	}))
	require.Empty(t, secondRemovedChannelIDs)
}

func seedVisibilityCleanupActionSubscription(
	t *testing.T,
	ctx context.Context,
	pool coredb.DB,
	appStore *Store,
	fixture appAdminFixture,
	installation appTargetInstallationFixture,
	channelID string,
	now time.Time,
) string {
	t.Helper()
	mentionSubscriptions, err := appStore.serviceActorStore().ListEventSubscriptions(ctx, serviceactor.EventSubscriptionListQuery{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: installation.ID,
		SubscriptionKind: serviceactor.SubscriptionKindAppMention,
		TargetKind:       "channel", TargetRef: channelID,
	})
	require.NoError(t, err)
	require.Len(t, mentionSubscriptions, 1)
	mention := mentionSubscriptions[0]
	var action serviceactor.EventSubscription
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		var upsertErr error
		action, _, upsertErr = appStore.serviceActorStore().InTx(tx).UpsertEventSubscription(ctx, serviceactor.EventSubscriptionUpsert{
			ID: "svcesub_visibility_cleanup_action", WorkspaceID: fixture.WorkspaceID,
			ServiceDefinitionID: mention.ServiceDefinitionID, ServiceInstallationID: installation.ID,
			ServiceMemberID: mention.ServiceMemberID, ServiceConnectionRef: "app_runtime",
			ServiceTargetBindingID: mention.ServiceTargetBindingID, ServiceKind: serviceactor.KindApp,
			SubscriptionKind: serviceactor.SubscriptionKindAppNotificationAction,
			TargetKind:       mention.TargetKind, TargetRef: mention.TargetRef,
			DeliveryScopeID: mention.DeliveryScopeID, PermissionCheckScopeID: mention.PermissionCheckScopeID,
			OwnerMemberID: fixture.MemberID, AuthorizerMemberID: fixture.MemberID,
			CreatedByActorType: "human", CreatedByActorID: fixture.UserID, CreatedVia: "test",
			ExternalResourceType: "app_notification_action", ExternalResourceKey: "visibility-cleanup-action",
			ExternalResourceRef: "visibility-cleanup-action", EventFilterKey: registry.AppEventNotificationAction,
			EventFilters: []byte(`{"events":["agentteam.app.notification.action"]}`), DeliveryPolicy: "app_runtime_delivery",
			Status: serviceactor.EventSubscriptionStatusActive, ActiveSubscriptionKey: "visibility-cleanup-action:" + installation.ID,
			Now: now,
		})
		return upsertErr
	}))
	return action.ID
}

func TestPrivateVisibilityCleanupFencesDynamicThreadMaterialization(t *testing.T) {
	ctx := context.Background()
	rootDSN := appTiDB.RootDSN(t)
	dsn := testkit.StartInitializedTiDBWithRootDSN(t, rootDSN)
	sqlDB, err := testkit.OpenTiDBContext(ctx, dsn)
	require.NoError(t, err)
	t.Cleanup(func() { _ = sqlDB.Close() })
	sqlDB.SetMaxOpenConns(8)
	sqlDB.SetMaxIdleConns(8)
	pool := coredb.WrapSQLDB(sqlDB)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_visibility_thread_race", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, fixture, "visibility-thread-race", now)
	channelID := "chn_bot_visibility_thread_race"
	seedChannelParticipantProjectionForTests(
		t, pool, fixture.WorkspaceID, channelID, channelID, "Visibility thread race", "private", fixture.MemberID, fixture.MemberID, now,
	)
	_, err = appStore.AddChannelServiceForUserWithEvent(
		ctx, fixture.UserID, fixture.WorkspaceID, channelID, installation.MemberID,
	)
	require.NoError(t, err)
	mustExec(t, pool, `
UPDATE extension_catalog_attributions
SET catalog_visibility = $3, updated_at = $4
WHERE workspace_id = $1 AND service_definition_id = $2
`, fixture.WorkspaceID, installation.AppID, "public", now.Add(time.Second))

	threadID := "thr_bot_visibility_thread_race"
	mustExec(t, pool, `
INSERT INTO scopes (
    id, workspace_id, type, parent_scope_id, root_message_id, root_message_short_id,
    identity_key, status, created_by_member_id, created_at, updated_at
) VALUES (
    $1, $2, 'thread', $3, $4, $5,
    CONCAT('thread/', $3, '/', $4), 'active', $6, $7, $7
)
`, threadID, fixture.WorkspaceID, channelID, "msg_bot_visibility_thread_race", "rootrace", fixture.MemberID, now)
	var threadTarget ServiceEventTarget
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		var resolveErr error
		threadTarget, resolveErr = ResolveServiceEventTargetForScope(
			ctx, tx, fixture.WorkspaceID, fixture.MemberID, threadID,
		)
		return resolveErr
	}))
	_, err = appStore.bindAppInstallationTargetForUser(
		ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, threadID, AppMentionCreatedViaTarget,
	)
	require.NoError(t, err)

	visibilityTx, err := pool.BeginTx(ctx, coredb.TxOptions{})
	require.NoError(t, err)
	committed := false
	t.Cleanup(func() {
		if !committed {
			_ = visibilityTx.Rollback(context.Background())
		}
	})
	require.NoError(t, installationstore.LockAppInstallationMutationScope(ctx, visibilityTx, installation.AppID))
	_, attributed, err := loadAppManagementAttribution(
		ctx, visibilityTx, fixture.WorkspaceID, installation.AppID, true,
	)
	require.NoError(t, err)
	require.True(t, attributed)
	currentInstallation, err := appStore.serviceActorStore().InTx(visibilityTx).LoadInstallationForUpdate(ctx, installation.ID)
	require.NoError(t, err)
	removedChannelIDs, err := removeAppFromAllChannelsForVisibilityChangeInTx(
		ctx, visibilityTx, fixture.WorkspaceID, fixture.MemberID, currentInstallation, installation.MemberID, now.Add(2*time.Second),
	)
	require.NoError(t, err)
	require.Equal(t, []string{channelID}, removedChannelIDs)
	require.NoError(t, updateWorkspaceAppCatalogVisibilityInTx(
		ctx, visibilityTx, fixture.WorkspaceID, installation.AppID, "private", now.Add(2*time.Second),
	))

	type bindOutcome struct {
		err error
	}
	bindDone := make(chan bindOutcome, 1)
	go func() {
		bindErr := coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
			keys := appMentionLookupKeys{MemberIDs: []string{installation.MemberID}}
			eligibleMemberIDs, innerErr := appStore.ensureAppMentionThreadTargetBindingsFromParent(
				ctx,
				tx,
				fixture.WorkspaceID,
				fixture.MemberID,
				"human",
				threadTarget,
				keys,
				now.Add(3*time.Second),
			)
			if innerErr != nil {
				return innerErr
			}
			return ensureAppMentionThreadTargetMemberships(
				ctx, tx, threadTarget, eligibleMemberIDs, now.Add(3*time.Second),
			)
		})
		bindDone <- bindOutcome{err: bindErr}
	}()

	select {
	case outcome := <-bindDone:
		require.Failf(t, "dynamic thread binding bypassed installation fence", "err=%v", outcome.err)
	case <-time.After(150 * time.Millisecond):
	}
	require.NoError(t, visibilityTx.Commit(ctx))
	committed = true
	select {
	case outcome := <-bindDone:
		require.NoError(t, outcome.err, "a stale App candidate must not fail the user's thread message transaction")
	case <-time.After(5 * time.Second):
		require.Fail(t, "dynamic thread binding did not finish after visibility cleanup committed")
	}

	requireAppTargetSingleInt(t, ctx, pool, 0, `
SELECT COUNT(*) FROM scope_memberships
WHERE scope_id = $1 AND member_id = $2 AND removed_at IS NULL
`, threadID, installation.MemberID)
	requireAppTargetSingleInt(t, ctx, pool, 0, `
SELECT COUNT(*) FROM service_target_bindings
WHERE workspace_id = $1 AND service_installation_id = $2
  AND delivery_scope_id = $3 AND status = 'active' AND removed_at IS NULL
`, fixture.WorkspaceID, installation.ID, threadID)
	requireAppTargetSingleInt(t, ctx, pool, 0, `
SELECT COUNT(*) FROM service_event_subscriptions
WHERE workspace_id = $1 AND service_installation_id = $2
  AND delivery_scope_id = $3 AND status = 'active' AND removed_at IS NULL
	`, fixture.WorkspaceID, installation.ID, threadID)
}

func TestDynamicThreadMaterializationLocksDefinitionBeforeVisibility(t *testing.T) {
	ctx := context.Background()
	rootDSN := appTiDB.RootDSN(t)
	dsn := testkit.StartInitializedTiDBWithRootDSN(t, rootDSN)
	sqlDB, err := testkit.OpenTiDBContext(ctx, dsn)
	require.NoError(t, err)
	t.Cleanup(func() { _ = sqlDB.Close() })
	sqlDB.SetMaxOpenConns(8)
	sqlDB.SetMaxIdleConns(8)
	pool := coredb.WrapSQLDB(sqlDB)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_visibility_definition_fence", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, fixture, "visibility-definition-fence", now)
	channelID := "chn_bot_visibility_definition_fence"
	seedChannelParticipantProjectionForTests(
		t, pool, fixture.WorkspaceID, channelID, channelID, "Visibility definition fence", "private", fixture.MemberID, fixture.MemberID, now,
	)
	_, err = appStore.AddChannelServiceForUserWithEvent(
		ctx, fixture.UserID, fixture.WorkspaceID, channelID, installation.MemberID,
	)
	require.NoError(t, err)
	mustExec(t, pool, `
UPDATE extension_catalog_attributions
SET catalog_visibility = $3, updated_at = $4
WHERE workspace_id = $1 AND service_definition_id = $2
`, fixture.WorkspaceID, installation.AppID, "public", now.Add(time.Second))

	threadID := "thr_bot_visibility_definition_fence"
	mustExec(t, pool, `
INSERT INTO scopes (
    id, workspace_id, type, parent_scope_id, root_message_id, root_message_short_id,
    identity_key, status, created_by_member_id, created_at, updated_at
) VALUES (
    $1, $2, 'thread', $3, $4, $5,
    CONCAT('thread/', $3, '/', $4), 'active', $6, $7, $7
)
`, threadID, fixture.WorkspaceID, channelID, "msg_bot_visibility_definition_fence", "rootdfnc", fixture.MemberID, now)
	var threadTarget ServiceEventTarget
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		var resolveErr error
		threadTarget, resolveErr = ResolveServiceEventTargetForScope(
			ctx, tx, fixture.WorkspaceID, fixture.MemberID, threadID,
		)
		return resolveErr
	}))

	materializationTx, err := pool.BeginTx(ctx, coredb.TxOptions{})
	require.NoError(t, err)
	defer func() { _ = materializationTx.Rollback(context.Background()) }()
	installationLockReached := make(chan struct{})
	allowInstallationLock := make(chan struct{})
	var allowInstallationLockOnce sync.Once
	releaseInstallationLock := func() {
		allowInstallationLockOnce.Do(func() { close(allowInstallationLock) })
	}
	defer releaseInstallationLock()
	observedTx := &appVisibilityInstallationLockObserverTx{
		Tx: materializationTx, Reached: installationLockReached, Release: allowInstallationLock,
	}
	type materializationOutcome struct {
		eligible int
		err      error
	}
	materializationDone := make(chan materializationOutcome, 1)
	go func() {
		keys := appMentionLookupKeys{MemberIDs: []string{installation.MemberID}}
		eligibleMemberIDs, materializationErr := appStore.ensureAppMentionThreadTargetBindingsFromParent(
			ctx, observedTx, fixture.WorkspaceID, fixture.MemberID, "human", threadTarget, keys, now.Add(3*time.Second),
		)
		if materializationErr == nil {
			materializationErr = ensureAppMentionThreadTargetMemberships(
				ctx, observedTx, threadTarget, eligibleMemberIDs, now.Add(3*time.Second),
			)
		}
		if materializationErr == nil {
			materializationErr = materializationTx.Commit(ctx)
		} else {
			_ = materializationTx.Rollback(context.Background())
		}
		materializationDone <- materializationOutcome{eligible: len(eligibleMemberIDs), err: materializationErr}
	}()
	select {
	case <-installationLockReached:
		// The production path is paused immediately before its installation
		// SELECT ... FOR UPDATE. It must already hold definition/attribution.
	case outcome := <-materializationDone:
		require.Failf(t, "materialization did not reach the installation lock", "err=%v", outcome.err)
	case <-time.After(5 * time.Second):
		require.Fail(t, "materialization did not reach the installation lock")
	}

	type visibilityOutcome struct {
		result AppVisibilityUpdateResult
		err    error
	}
	visibilityDone := make(chan visibilityOutcome, 1)
	go func() {
		result, updateErr := appStore.UpdateAppVisibilityForUser(
			ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, "private", now.Add(2*time.Second),
		)
		visibilityDone <- visibilityOutcome{result: result, err: updateErr}
	}()
	select {
	case outcome := <-visibilityDone:
		releaseInstallationLock()
		select {
		case <-materializationDone:
		case <-time.After(5 * time.Second):
		}
		require.Failf(t, "visibility bypassed materialization definition fence", "err=%v", outcome.err)
	case <-time.After(150 * time.Millisecond):
	}
	releaseInstallationLock()
	select {
	case outcome := <-materializationDone:
		require.NoError(t, outcome.err, "the concurrent visibility request must not fail the user's thread message")
		require.Equal(t, 1, outcome.eligible)
	case <-time.After(5 * time.Second):
		require.Fail(t, "thread materialization did not finish after its installation lock was released")
	}

	select {
	case outcome := <-visibilityDone:
		require.NoError(t, outcome.err)
		require.Equal(t, "private", outcome.result.CatalogVisibility)
		require.Equal(t, []string{channelID}, outcome.result.RemovedChannelIDs)
	case <-time.After(5 * time.Second):
		require.Fail(t, "visibility did not finish after thread materialization committed")
	}
	requireAppTargetSingleInt(t, ctx, pool, 0, `
SELECT COUNT(*) FROM scope_memberships
WHERE scope_id IN ($1, $2) AND member_id = $3 AND removed_at IS NULL
`, channelID, threadID, installation.MemberID)
	requireAppTargetSingleInt(t, ctx, pool, 0, `
SELECT COUNT(*) FROM service_target_bindings
WHERE workspace_id = $1 AND service_installation_id = $2
  AND permission_check_scope_id = $3 AND status = 'active' AND removed_at IS NULL
`, fixture.WorkspaceID, installation.ID, channelID)
	requireAppTargetSingleInt(t, ctx, pool, 0, `
SELECT COUNT(*) FROM service_event_subscriptions
WHERE workspace_id = $1 AND service_installation_id = $2
  AND permission_check_scope_id = $3 AND removed_at IS NULL
	`, fixture.WorkspaceID, installation.ID, channelID)
}

type appVisibilityInstallationLockObserverTx struct {
	coredb.Tx
	Reached chan<- struct{}
	Release <-chan struct{}
	once    sync.Once
}

func (tx *appVisibilityInstallationLockObserverTx) QueryRow(ctx context.Context, query string, args ...any) coredb.Row {
	if strings.Contains(query, "FROM service_installations WHERE id = $1 FOR UPDATE") {
		tx.once.Do(func() {
			close(tx.Reached)
			<-tx.Release
		})
	}
	return tx.Tx.QueryRow(ctx, query, args...)
}

type appTargetInstallationFixture struct {
	ID       string
	AppID    string
	MemberID string
	Grants   []string
}