diff --git a/apps/sim/app/api/organizations/[id]/transfer-ownership/route.ts b/apps/sim/app/api/organizations/[id]/transfer-ownership/route.ts index 02975118faa..53ef7dab812 100644 --- a/apps/sim/app/api/organizations/[id]/transfer-ownership/route.ts +++ b/apps/sim/app/api/organizations/[id]/transfer-ownership/route.ts @@ -151,7 +151,7 @@ export const POST = withRouteHandler( memberId: selfMember.id, spareSessionToken: session.session.token, }) - deliverRunStops(removeResult.stoppedRuns ?? []) + deliverRunStops(removeResult.stoppedRuns) if (!removeResult.success) { logger.error('Transfer succeeded but self-removal failed', { diff --git a/apps/sim/app/api/v1/admin/organizations/[id]/members/[memberId]/route.ts b/apps/sim/app/api/v1/admin/organizations/[id]/members/[memberId]/route.ts index 2d54f37ae2c..759f1b3bb6b 100644 --- a/apps/sim/app/api/v1/admin/organizations/[id]/members/[memberId]/route.ts +++ b/apps/sim/app/api/v1/admin/organizations/[id]/members/[memberId]/route.ts @@ -272,7 +272,7 @@ export const DELETE = withRouteHandler( memberId, skipBillingLogic, }) - deliverRunStops(result.stoppedRuns ?? []) + deliverRunStops(result.stoppedRuns) if (!result.success) { if (result.error === 'Cannot remove organization owner') { diff --git a/apps/sim/ee/scim/lib/application/groups/manage-groups.test.ts b/apps/sim/ee/scim/lib/application/groups/manage-groups.test.ts index 982391eb525..7b469c7a426 100644 --- a/apps/sim/ee/scim/lib/application/groups/manage-groups.test.ts +++ b/apps/sim/ee/scim/lib/application/groups/manage-groups.test.ts @@ -94,6 +94,7 @@ describe('replaceScimGroup', () => { memberIds.delete(scimUserId) ) mocks.countMembers.mockImplementation(async () => memberIds.size) + mocks.reconcile.mockResolvedValue({ added: [], removed: [], raised: [], stoppedRuns: [] }) mocks.touch.mockImplementation(async () => { group = { ...group, updatedAt: new Date('2026-01-02T00:00:00.000Z') } }) diff --git a/apps/sim/ee/scim/lib/application/groups/manage-groups.ts b/apps/sim/ee/scim/lib/application/groups/manage-groups.ts index 16a303b2871..2ad3f480e59 100644 --- a/apps/sim/ee/scim/lib/application/groups/manage-groups.ts +++ b/apps/sim/ee/scim/lib/application/groups/manage-groups.ts @@ -5,6 +5,8 @@ import { and, eq, ne } from 'drizzle-orm' import type { ScimPatchOperation } from '@/lib/api/contracts/scim' import { acquireOrganizationMutationLock } from '@/lib/billing/organizations/membership' import type { DbOrTx, DbTransaction } from '@/lib/db/types' +import type { StoppedRun } from '@/lib/mothership/async-runs/repository' +import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery' import { defineAuthorizedScimUseCase, type ScimUseCaseArgs, @@ -50,16 +52,19 @@ import { /** * Every group write runs in one transaction under the organization lock, which * heads the documented lock order, so two full-membership PATCHes cannot both - * compute from the same stale membership and keep members from both. + * compute from the same stale membership and keep members from both. The runs + * its projection stopped are delivered once that transaction commits. */ -function withGroupWrite( +async function withGroupWrite( context: ScimUseCaseContext, work: (tx: DbTransaction) => Promise -): Promise { - return db.transaction(async (tx) => { +): Promise> { + const { stoppedRuns, ...result } = await db.transaction(async (tx) => { await acquireOrganizationMutationLock(tx, context.organizationId) return work(tx) }) + deliverRunStops(stoppedRuns) + return result } async function assertDisplayNameAvailable( @@ -205,7 +210,7 @@ export const createScimGroup = defineAuthorizedScimUseCase({ }) : 'no-match' - await reconcileUsersProjection(tx, { + const projection = await reconcileUsersProjection(tx, { connectionId: context.connection.id, organizationId: context.organizationId, scimUserIds: memberIds, @@ -225,6 +230,7 @@ export const createScimGroup = defineAuthorizedScimUseCase({ resource: toGroupResource({ ...created, members }, context.baseUrl), touchedUserIds: memberIds, renamed: true, + stoppedRuns: projection.stoppedRuns, } }) }, @@ -295,7 +301,7 @@ export const replaceScimGroup = defineAuthorizedScimUseCase({ await assertMemberCount(tx, current.id) if (touched.size > 0) await touchScimGroup(tx, current.id) - await reconcileUsersProjection(tx, { + const projection = await reconcileUsersProjection(tx, { connectionId: context.connection.id, organizationId: context.organizationId, scimUserIds: [...touched], @@ -317,6 +323,7 @@ export const replaceScimGroup = defineAuthorizedScimUseCase({ resource: toGroupResource({ ...refreshed, members }, context.baseUrl), touchedUserIds: [...touched], renamed: renamed || (current.externalId ?? null) !== (refreshed.externalId ?? null), + stoppedRuns: projection.stoppedRuns, } }) }, @@ -428,7 +435,7 @@ export const patchScimGroup = defineAuthorizedScimUseCase({ await assertMemberCount(tx, current.id) if (touched.size > 0) await touchScimGroup(tx, current.id) - await reconcileUsersProjection(tx, { + const projection = await reconcileUsersProjection(tx, { connectionId: context.connection.id, organizationId: context.organizationId, scimUserIds: [...touched], @@ -447,6 +454,7 @@ export const patchScimGroup = defineAuthorizedScimUseCase({ patch.kind === 'full' ? (patch.displayName ?? current.displayName) : current.displayName, touchedUserIds: [...touched], renamed, + stoppedRuns: projection.stoppedRuns, } }) }, @@ -481,7 +489,7 @@ export const deleteScimGroup = defineAuthorizedScimUseCase({ */ const memberIds = await loadGroupMemberIds(tx, current.id) await deleteScimGroupRow(tx, current.id) - await reconcileUsersProjection(tx, { + const projection = await reconcileUsersProjection(tx, { connectionId: context.connection.id, organizationId: context.organizationId, scimUserIds: memberIds, @@ -491,6 +499,7 @@ export const deleteScimGroup = defineAuthorizedScimUseCase({ groupId: current.id, displayName: current.displayName, memberCount: memberIds.length, + stoppedRuns: projection.stoppedRuns, } }) }, diff --git a/apps/sim/ee/scim/lib/application/users/deprovision-user.ts b/apps/sim/ee/scim/lib/application/users/deprovision-user.ts index d08d030c839..6af018da930 100644 --- a/apps/sim/ee/scim/lib/application/users/deprovision-user.ts +++ b/apps/sim/ee/scim/lib/application/users/deprovision-user.ts @@ -75,7 +75,7 @@ export const deprovisionScimUser = defineAuthorizedScimUseCase({ memberId: membership.id, revokePersonalApiKeys: true, }) - deliverRunStops(removal.stoppedRuns ?? []) + deliverRunStops(removal.stoppedRuns) if (!removal.success) { throw new ScimError(409, undefined, removal.error ?? 'The member could not be removed') } diff --git a/apps/sim/ee/scim/lib/application/users/provision-user.ts b/apps/sim/ee/scim/lib/application/users/provision-user.ts index 08523eee734..17fc69a3b63 100644 --- a/apps/sim/ee/scim/lib/application/users/provision-user.ts +++ b/apps/sim/ee/scim/lib/application/users/provision-user.ts @@ -215,6 +215,7 @@ export const provisionScimUser = defineAuthorizedScimUseCase({ connectionId: context.connection.id, externalId: attributes.externalId, }) + /** The new directory record has no provenance to withdraw, so this stops no runs. */ await reconcileUserProjection(tx, { connectionId: context.connection.id, organizationId: context.organizationId, diff --git a/apps/sim/ee/scim/lib/application/users/update-user.test.ts b/apps/sim/ee/scim/lib/application/users/update-user.test.ts index 43d9ce514a2..b809b8a8495 100644 --- a/apps/sim/ee/scim/lib/application/users/update-user.test.ts +++ b/apps/sim/ee/scim/lib/application/users/update-user.test.ts @@ -133,7 +133,7 @@ describe('user updates', () => { beforeEach(() => { resetDbChainMock() mocks.loadGroups.mockResolvedValue(new Map()) - mocks.reconcile.mockResolvedValue({ added: [], removed: [], raised: [] }) + mocks.reconcile.mockResolvedValue({ added: [], removed: [], raised: [], stoppedRuns: [] }) }) it('serializes on the organization and user locks before computing the update', async () => { diff --git a/apps/sim/ee/scim/lib/application/users/update-user.ts b/apps/sim/ee/scim/lib/application/users/update-user.ts index 900f814a5be..471e93b2deb 100644 --- a/apps/sim/ee/scim/lib/application/users/update-user.ts +++ b/apps/sim/ee/scim/lib/application/users/update-user.ts @@ -5,6 +5,8 @@ import { normalizeEmail } from '@sim/utils/string' import type { ScimPatchOperation } from '@/lib/api/contracts/scim' import { acquireOrganizationUserMutationLocks } from '@/lib/billing/organizations/membership' import type { DbOrTx, DbTransaction } from '@/lib/db/types' +import type { StoppedRun } from '@/lib/mothership/async-runs/repository' +import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery' import { suspendMemberTx, unsuspendMemberTx } from '@/lib/organizations/members/lifecycle' import { invalidateAfterSessionRevocation, @@ -42,18 +44,24 @@ import { * deactivation actually does. */ -export interface UpdateOutcome { +interface UpdateOutcome { emailChanged: boolean deactivated: boolean reactivated: boolean } +/** An applied update and the runs its projection stopped, delivered once the write commits. */ +interface AppliedUserUpdate { + outcome: UpdateOutcome + stoppedRuns: StoppedRun[] +} + async function applyUserUpdate( tx: DbTransaction, context: ScimUseCaseContext, current: ScimUserRecord, next: ScimUserAttributes -): Promise { +): Promise { const nextEmail = primaryEmail(next) const nextName = accountName(next) const emailChanged = accountEmailDiverged(current, next) @@ -108,14 +116,17 @@ async function applyUserUpdate( active: next.active, }) - await reconcileUserProjection(tx, { + const projection = await reconcileUserProjection(tx, { connectionId: context.connection.id, organizationId: context.organizationId, scimUserId: current.id, settings: context.connection.settings, }) - return { emailChanged, deactivated, reactivated } + return { + outcome: { emailChanged, deactivated, reactivated }, + stoppedRuns: projection.stoppedRuns, + } } export interface UpdateScimUserResult { @@ -125,6 +136,27 @@ export interface UpdateScimUserResult { resource: ReturnType } +/** Runs a user write in one transaction and delivers the runs it stopped once that commits. */ +async function writeUser( + context: ScimUseCaseContext, + scimUserId: string, + update: (tx: DbTransaction, current: ScimUserRecord) => Promise +): Promise { + const { stoppedRuns, ...result } = await db.transaction(async (tx) => { + const current = await loadUserForUpdate(tx, context, scimUserId) + const applied = await update(tx, current) + return { + scimUserId: current.id, + userId: current.userId, + outcome: applied?.outcome ?? null, + resource: await renderUpdated(tx, context.connection.id, current.id, context.baseUrl), + stoppedRuns: applied?.stoppedRuns ?? [], + } + }) + deliverRunStops(stoppedRuns) + return result +} + /** * Loads the stored resource under the organization and user advisory locks. * @@ -229,9 +261,7 @@ export const replaceScimUser = defineAuthorizedScimUseCase({ input, context, }: ScimUseCaseArgs): Promise { - return db.transaction(async (tx) => { - const current = await loadUserForUpdate(tx, context, input.scimUserId) - + return writeUser(context, input.scimUserId, async (tx, current) => { /** * A replace keeps attributes Sim does not model that the directory sent on * a previous write but omitted now, so a partial mapping does not erase @@ -249,16 +279,10 @@ export const replaceScimUser = defineAuthorizedScimUseCase({ * changes nothing must not write, audit, or re-project, or a 2,000-user * organization produces 2,000 spurious audit rows per sync. */ - const outcome = - userAttributesEqual(current.attributes, next) && !accountIdentityDiverged(current, next) - ? null - : await applyUserUpdate(tx, context, current, next) - return { - scimUserId: current.id, - userId: current.userId, - outcome, - resource: await renderUpdated(tx, context.connection.id, current.id, context.baseUrl), - } + return userAttributesEqual(current.attributes, next) && + !accountIdentityDiverged(current, next) + ? null + : applyUserUpdate(tx, context, current, next) }) }, projectAudit: ({ result }) => auditEntries(result), @@ -277,9 +301,7 @@ export const patchScimUser = defineAuthorizedScimUseCase({ input, context, }: ScimUseCaseArgs): Promise { - return db.transaction(async (tx) => { - const current = await loadUserForUpdate(tx, context, input.scimUserId) - + return writeUser(context, input.scimUserId, async (tx, current) => { const { next, changed } = applyUserPatch(current.attributes, input.operations) /** @@ -289,16 +311,9 @@ export const patchScimUser = defineAuthorizedScimUseCase({ * projection pass, and a `lastModified` bump for a request that meant * nothing. */ - const outcome = - changed || accountIdentityDiverged(current, next) - ? await applyUserUpdate(tx, context, current, next) - : null - return { - scimUserId: current.id, - userId: current.userId, - outcome, - resource: await renderUpdated(tx, context.connection.id, current.id, context.baseUrl), - } + return changed || accountIdentityDiverged(current, next) + ? applyUserUpdate(tx, context, current, next) + : null }) }, projectAudit: ({ result }) => auditEntries(result), diff --git a/apps/sim/ee/scim/lib/projection/reconcile-user.test.ts b/apps/sim/ee/scim/lib/projection/reconcile-user.test.ts index 161dacf2451..2a03b1d107e 100644 --- a/apps/sim/ee/scim/lib/projection/reconcile-user.test.ts +++ b/apps/sim/ee/scim/lib/projection/reconcile-user.test.ts @@ -23,6 +23,7 @@ const hoistedMocks = vi.hoisted(() => ({ lowerWorkspace: vi.fn(), readPermission: vi.fn(), revokeWorkspace: vi.fn(), + stopRuns: vi.fn(), })) vi.mock('@/lib/billing/organizations/membership', () => organizationMembershipMock) @@ -42,6 +43,7 @@ vi.mock('@/lib/workspaces/access/workspace-access', () => ({ lowerWorkspaceAccessTx: hoistedMocks.lowerWorkspace, readWorkspacePermission: hoistedMocks.readPermission, revokeWorkspaceAccessTx: hoistedMocks.revokeWorkspace, + stopRunsWithoutWorkspaceAccessTx: hoistedMocks.stopRuns, permissionRank: (permission: string) => ({ read: 1, write: 2, admin: 3 })[permission] ?? 0, })) @@ -122,6 +124,7 @@ describe('reconcileUserProjection', () => { ownershipTransferred: false, stoppedRuns: [], }) + mocks.stopRuns.mockResolvedValue([]) mocks.addMember.mockResolvedValue('added') mocks.removeMember.mockResolvedValue('removed') mocks.changeMemberRole.mockResolvedValue({ @@ -207,6 +210,7 @@ describe('reconcileUserProjection', () => { organizationId: 'org-1', userId: 'u-1', role: 'admin', + deferRunStops: true, }) }) diff --git a/apps/sim/ee/scim/lib/projection/reconcile-user.ts b/apps/sim/ee/scim/lib/projection/reconcile-user.ts index 696f4290945..15be39369c5 100644 --- a/apps/sim/ee/scim/lib/projection/reconcile-user.ts +++ b/apps/sim/ee/scim/lib/projection/reconcile-user.ts @@ -17,6 +17,8 @@ import { and, eq, inArray } from 'drizzle-orm' import { acquireOrganizationUserMutationLocks } from '@/lib/billing/organizations/membership' import { OrchestrationError } from '@/lib/core/orchestration/types' import type { DbOrTx, DbTransaction } from '@/lib/db/types' +import type { StoppedRun } from '@/lib/mothership/async-runs/repository' +import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery' import { changeMemberRoleTx } from '@/lib/organizations/members/lifecycle' import { addPermissionGroupMemberTx, @@ -31,6 +33,7 @@ import { permissionRank, readWorkspacePermission, revokeWorkspaceAccessTx, + stopRunsWithoutWorkspaceAccessTx, } from '@/lib/workspaces/access/workspace-access' import { type MappingRow, @@ -58,9 +61,24 @@ export interface ProjectionDelta { removed: ProjectionGrant[] /** Workspace grants whose level changed in either direction. */ raised: ProjectionGrant[] + /** + * Chat runs the withdrawals stopped in the caller's transaction. Whoever owns that transaction + * delivers their Stop with `deliverRunStops` once it commits. + */ + stoppedRuns: StoppedRun[] +} + +function emptyDelta(): ProjectionDelta { + return { added: [], removed: [], raised: [], stoppedRuns: [] } } -const EMPTY_DELTA: ProjectionDelta = { added: [], removed: [], raised: [] } +type ProjectionChanges = Omit + +function addChanges(total: ProjectionChanges, changes: ProjectionChanges): void { + total.added.push(...changes.added) + total.removed.push(...changes.removed) + total.raised.push(...changes.raised) +} /** Users per transaction when projecting outside a request's own transaction; the organization lock is held for the batch. */ export const PROJECTION_BATCH_SIZE = 25 @@ -265,7 +283,12 @@ async function setOrganizationRole( role: 'admin' | 'member' ): Promise { try { - const change = await changeMemberRoleTx(tx, { organizationId, userId, role }) + const change = await changeMemberRoleTx(tx, { + organizationId, + userId, + role, + deferRunStops: true, + }) return change.changed ? 'applied' : 'unchanged' } catch (error) { if (error instanceof OrchestrationError && error.code === 'conflict') { @@ -284,9 +307,10 @@ async function setOrganizationRole( } /** - * Withdraws one grant. Returns false when the grant must stay in place — the - * access could not be handed on — so the provenance row survives and the next - * pass retries instead of forgetting. + * Withdraws one grant, returning the workspaces where it may have ended the + * user's access. Returns null when the grant must stay in place — the access + * could not be handed on — so the provenance row survives and the next pass + * retries instead of forgetting. * * Adopted access — held by hand before any mapping covered it — is the person's * own, so the directory only forgets its record of it, unless the organization @@ -301,13 +325,13 @@ async function withdrawGrant( lockManualMembership: boolean foreignWorkspaceIds: Set } -): Promise { +): Promise { const { grant } = params - if (grant.origin === 'adopted' && !params.lockManualMembership) return true + if (grant.origin === 'adopted' && !params.lockManualMembership) return [] switch (grant.targetKind) { case 'workspace': { /** The workspace belongs to another tenant now; its access is theirs to manage. Forget the grant. */ - if (params.foreignWorkspaceIds.has(grant.targetId)) return true + if (params.foreignWorkspaceIds.has(grant.targetId)) return [] /** * A grant raised by hand above what the directory set is left alone. The * directory said "at least write"; someone deliberately made it admin, and @@ -318,13 +342,13 @@ async function withdrawGrant( workspaceId: grant.targetId, userId: params.userId, }) - if (!current) return true + if (!current) return [] if ( !params.lockManualMembership && grant.permissionType && permissionRank(current) > permissionRank(grant.permissionType) ) { - return true + return [] } if (!params.lockManualMembership && grant.baselinePermission) { if (permissionRank(current) > permissionRank(grant.baselinePermission)) { @@ -335,11 +359,12 @@ async function withdrawGrant( to: grant.baselinePermission, }) } - return true + return [] } const outcome = await revokeWorkspaceAccessTx(tx, { workspaceId: grant.targetId, userId: params.userId, + deferRunStops: true, }) if (!outcome.revoked) { logger.warn('Left workspace access in place: ownership could not be handed on', { @@ -347,14 +372,21 @@ async function withdrawGrant( userId: params.userId, reason: outcome.reason, }) - return false + return null } - return true + return [grant.targetId] + } + case 'org_role': { + if (grant.targetId !== 'admin') return [] + const demoted = await setOrganizationRole(tx, params.organizationId, params.userId, 'member') + if (demoted !== 'applied') return [] + /** An organization admin reaches every organization workspace; the demotion may end any of them. */ + const workspaces = await tx + .select({ id: workspace.id }) + .from(workspace) + .where(eq(workspace.organizationId, params.organizationId)) + return workspaces.map(({ id }) => id) } - case 'org_role': - if (grant.targetId !== 'admin') return true - await setOrganizationRole(tx, params.organizationId, params.userId, 'member') - return true case 'permission_group': try { await removePermissionGroupMemberTx(tx, { @@ -368,7 +400,7 @@ async function withdrawGrant( * target column carries no foreign key. Withdrawing from nothing is * complete, not a failure. */ - if (error instanceof PermissionGroupNotFoundError) return true + if (error instanceof PermissionGroupNotFoundError) return [] /** * An administrator moved the group back to governing everyone, and its * last member cannot leave without emptying it. The grant stays on @@ -381,11 +413,11 @@ async function withdrawGrant( userId: params.userId, conflict: error.message, }) - return false + return null } throw error } - return true + return [] } } @@ -412,7 +444,7 @@ export async function reconcileUserProjection( .where(and(eq(scimUser.id, params.scimUserId), eq(scimUser.connectionId, params.connectionId))) .limit(1) - if (!record) return EMPTY_DELTA + if (!record) return emptyDelta() /** * Taken before any target is touched. The organization lock is the root of @@ -450,19 +482,20 @@ export async function reconcileUserProjection( }) const plan = planGrantChanges(desired, current, actual) - const delta: ProjectionDelta = { added: [], removed: [], raised: [] } + const delta = emptyDelta() + const accessEndedIn = new Set() const lockManualMembership = params.settings.lockManualMembership === true /** Withdrawals first, so a move between groups frees its workspace slot. */ for (const grant of plan.withdraw) { - const withdrawn = await withdrawGrant(tx, { + const accessEnded = await withdrawGrant(tx, { organizationId: params.organizationId, userId: record.userId, grant, lockManualMembership, foreignWorkspaceIds, }) - if (!withdrawn) continue + if (!accessEnded) continue await tx .delete(scimProjectionGrant) .where( @@ -473,6 +506,7 @@ export async function reconcileUserProjection( ) ) delta.removed.push(grant) + for (const workspaceId of accessEnded) accessEndedIn.add(workspaceId) } for (const { grant, previousGrant } of plan.apply) { @@ -571,6 +605,12 @@ export async function reconcileUserProjection( else delta.added.push(grant) } + /** + * Runs are judged once every withdrawal and grant above is in place: a run in + * a workspace this pass takes away and gives back, by grant or by role, keeps + * going. + */ + delta.stoppedRuns = await stopRunsWithoutWorkspaceAccessTx(tx, record.userId, [...accessEndedIn]) return delta } @@ -583,15 +623,19 @@ export async function reconcileUsersProjection( scimUserIds: string[] settings: ScimConnectionSettings } -): Promise { +): Promise { + const total = emptyDelta() for (const scimUserId of [...new Set(params.scimUserIds)].sort()) { - await reconcileUserProjection(tx, { + const delta = await reconcileUserProjection(tx, { connectionId: params.connectionId, organizationId: params.organizationId, scimUserId, settings: params.settings, }) + addChanges(total, delta) + total.stoppedRuns.push(...delta.stoppedRuns) } + return total } /** @@ -601,30 +645,27 @@ export async function reconcileUsersProjection( * and an administrator changing a mapping on a large group. One transaction over * thousands of users would hold the organization's advisory locks for its whole * duration, blocking every invitation and role change in the tenant meanwhile. + * + * Each batch's stopped runs are delivered once that batch commits, so a later + * batch that fails neither loses an earlier batch's Stops nor delivers its own. */ export async function reconcileUsersProjectionInBatches(params: { connectionId: string organizationId: string scimUserIds: string[] settings: ScimConnectionSettings -}): Promise { - const total: ProjectionDelta = { added: [], removed: [], raised: [] } +}): Promise { + const changes: ProjectionChanges = { added: [], removed: [], raised: [] } const ids = [...new Set(params.scimUserIds)].sort() for (let start = 0; start < ids.length; start += PROJECTION_BATCH_SIZE) { - const batch = ids.slice(start, start + PROJECTION_BATCH_SIZE) - await db.transaction(async (tx) => { - for (const scimUserId of batch) { - const delta = await reconcileUserProjection(tx, { - connectionId: params.connectionId, - organizationId: params.organizationId, - scimUserId, - settings: params.settings, - }) - total.added.push(...delta.added) - total.removed.push(...delta.removed) - total.raised.push(...delta.raised) - } - }) + const { stoppedRuns, ...batch } = await db.transaction((tx) => + reconcileUsersProjection(tx, { + ...params, + scimUserIds: ids.slice(start, start + PROJECTION_BATCH_SIZE), + }) + ) + deliverRunStops(stoppedRuns) + addChanges(changes, batch) } - return total + return changes } diff --git a/apps/sim/lib/admin/dashboard.ts b/apps/sim/lib/admin/dashboard.ts index bf803aa99b4..6865a6be4ff 100644 --- a/apps/sim/lib/admin/dashboard.ts +++ b/apps/sim/lib/admin/dashboard.ts @@ -1782,7 +1782,7 @@ export async function removeDashboardOrganizationMember( organizationId, memberId, }) - deliverRunStops(result.stoppedRuns ?? []) + deliverRunStops(result.stoppedRuns) if (!result.success) throw new Error(result.error ?? 'Failed to remove member') try { await reconcileOrganizationSeats({ diff --git a/apps/sim/lib/admin/member-operation.ts b/apps/sim/lib/admin/member-operation.ts index fd987246c8e..e692c00d8e0 100644 --- a/apps/sim/lib/admin/member-operation.ts +++ b/apps/sim/lib/admin/member-operation.ts @@ -515,7 +515,7 @@ async function applyMembership( usageLimitDollars: request.usageLimitDollars, setBy: request.actor.id ?? undefined, }) - deliverRunStops(transferred.stoppedRuns ?? []) + deliverRunStops(transferred.stoppedRuns) if (!transferred.success || !transferred.memberId) { throw new Error(transferred.error ?? 'Failed to transfer organization member') } diff --git a/apps/sim/lib/billing/organizations/membership.ts b/apps/sim/lib/billing/organizations/membership.ts index 157f1a7ad36..6af43569466 100644 --- a/apps/sim/lib/billing/organizations/membership.ts +++ b/apps/sim/lib/billing/organizations/membership.ts @@ -548,7 +548,7 @@ export interface RemoveMemberResult { * Chat runs the change stopped in its transaction. The caller delivers their Stop with * `deliverRunStops`: this module sits in page graphs that must not load that delivery. */ - stoppedRuns?: StoppedRun[] + stoppedRuns: StoppedRun[] } export interface RemoveExternalWorkspaceAccessResult { @@ -562,7 +562,7 @@ export interface RemoveExternalWorkspaceAccessResult { * Chat runs the change stopped in its transaction. The caller delivers their Stop with * `deliverRunStops`: this module sits in page graphs that must not load that delivery. */ - stoppedRuns?: StoppedRun[] + stoppedRuns: StoppedRun[] } export type MembershipAdditionFailureCode = @@ -1110,7 +1110,7 @@ export interface TransferOrganizationMemberResult { * Chat runs the change stopped in its transaction. The caller delivers their Stop with * `deliverRunStops`: this module sits in page graphs that must not load that delivery. */ - stoppedRuns?: StoppedRun[] + stoppedRuns: StoppedRun[] } /** @@ -1127,6 +1127,7 @@ export async function transferUserBetweenOrganizations( credentialMembershipsRevoked: 0, pendingInvitationsCancelled: 0, usageCaptured: 0, + stoppedRuns: [], } if (params.sourceOrganizationId === params.destinationOrganizationId) { return { @@ -1136,8 +1137,6 @@ export async function transferUserBetweenOrganizations( } } - /** The source organization's runs this transfer ends. */ - let stoppedRuns: StoppedRun[] = [] try { const transferResult = await withInvitationSafeOrganizationAccessMutation( { @@ -1260,7 +1259,8 @@ export async function transferUserBetweenOrganizations( ) await removeWorkspaceSkillMembershipsTx(tx, workspaceIds, params.userId) } - stoppedRuns = [ + /** The source organization's runs this transfer ends. */ + const stoppedRuns = [ ...(await stopLiveRunsTx(tx, { userId: params.userId, organizationId: params.sourceOrganizationId, @@ -1296,6 +1296,7 @@ export async function transferUserBetweenOrganizations( // Nothing to capture: the member's ledger rows stay stamped to the // source org's period and are billed at its cycle close. usageCaptured: 0, + stoppedRuns, } } ) @@ -1303,7 +1304,7 @@ export async function transferUserBetweenOrganizations( // immediately, and their sessions clamp to its policy — same treatment as // invite acceptance. Best-effort. await applySessionPolicyToNewMember(params.userId, params.destinationOrganizationId) - return { ...transferResult, stoppedRuns } + return transferResult } catch (error) { logger.error('Failed to transfer organization member', { ...params, error }) return { success: false, error: getErrorMessage(error), ...emptyResult } @@ -1359,11 +1360,16 @@ export async function removeUserFromOrganization( .limit(1) if (!existingMember) { - return { success: false, error: 'Member not found', billingActions } + return { success: false, error: 'Member not found', billingActions, stoppedRuns: [] } } if (existingMember.role === 'owner') { - return { success: false, error: 'Cannot remove organization owner', billingActions } + return { + success: false, + error: 'Cannot remove organization owner', + billingActions, + stoppedRuns: [], + } } const result = await withInvitationSafeOrganizationAccessMutation( @@ -1520,7 +1526,7 @@ export async function removeUserFromOrganization( userId, memberId, }) - return { success: true, removed: false, billingActions } + return { success: true, removed: false, billingActions, stoppedRuns: [] } } billingActions.usageCaptured = result.usageCaptured @@ -1586,7 +1592,7 @@ export async function removeUserFromOrganization( return { success: true, removed: true, billingActions, stoppedRuns: result.stoppedRuns } } catch (error) { if (error instanceof WorkspaceBillingAccountRemovalError) { - return { success: false, error: error.message, billingActions } + return { success: false, error: error.message, billingActions, stoppedRuns: [] } } if (onError === 'throw') throw error @@ -1596,7 +1602,12 @@ export async function removeUserFromOrganization( memberId, error, }) - return { success: false, error: 'Failed to remove user from organization', billingActions } + return { + success: false, + error: 'Failed to remove user from organization', + billingActions, + stoppedRuns: [], + } } } @@ -1610,8 +1621,13 @@ export async function removeExternalUserFromOrganizationWorkspaces(params: { actorUserId?: string }): Promise { const { userId, organizationId } = params - /** The member's runs in the workspaces they lose. */ - let stoppedRuns: StoppedRun[] = [] + const nothingRemoved = { + workspaceAccessRevoked: 0, + permissionGroupsRevoked: 0, + credentialMembershipsRevoked: 0, + pendingInvitationsCancelled: 0, + stoppedRuns: [], + } try { const [existingMember] = await db @@ -1624,10 +1640,7 @@ export async function removeExternalUserFromOrganizationWorkspaces(params: { return { success: false, error: 'User is an organization member', - workspaceAccessRevoked: 0, - permissionGroupsRevoked: 0, - credentialMembershipsRevoked: 0, - pendingInvitationsCancelled: 0, + ...nothingRemoved, } } @@ -1636,6 +1649,7 @@ export async function removeExternalUserFromOrganizationWorkspaces(params: { permissionGroupsRevoked, credentialMembershipsRevoked, pendingInvitationsCancelled, + stoppedRuns, } = await withInvitationSafeOrganizationAccessMutation( { userId, organizationId, scope: 'external' }, async (tx, { workspaceIds, invitationIds }) => { @@ -1692,6 +1706,7 @@ export async function removeExternalUserFromOrganizationWorkspaces(params: { permissionGroupsRevoked: deletedPermissionGroups.length, credentialMembershipsRevoked: 0, pendingInvitationsCancelled: cancelledInvitations.length, + stoppedRuns: [], } } @@ -1722,13 +1737,14 @@ export async function removeExternalUserFromOrganizationWorkspaces(params: { userId ) await removeWorkspaceSkillMembershipsTx(tx, workspaceIds, userId) - stoppedRuns = await stopLiveRunsTx(tx, { userId, workspaceIds }) return { workspaceAccessRevoked: deletedPermissions.length, permissionGroupsRevoked: deletedPermissionGroups.length, credentialMembershipsRevoked, pendingInvitationsCancelled: cancelledInvitations.length, + /** The member's runs in the workspaces they lose. */ + stoppedRuns: await stopLiveRunsTx(tx, { userId, workspaceIds }), } } ) @@ -1773,10 +1789,7 @@ export async function removeExternalUserFromOrganizationWorkspaces(params: { return { success: false, error: error.message, - workspaceAccessRevoked: 0, - permissionGroupsRevoked: 0, - credentialMembershipsRevoked: 0, - pendingInvitationsCancelled: 0, + ...nothingRemoved, } } @@ -1788,10 +1801,7 @@ export async function removeExternalUserFromOrganizationWorkspaces(params: { return { success: false, error: 'Failed to remove external workspace member', - workspaceAccessRevoked: 0, - permissionGroupsRevoked: 0, - credentialMembershipsRevoked: 0, - pendingInvitationsCancelled: 0, + ...nothingRemoved, } } } diff --git a/apps/sim/lib/mothership/request/application/removal-stops-runs.integration.ts b/apps/sim/lib/mothership/request/application/removal-stops-runs.integration.ts index 6d49149df9c..0aeac64fa6f 100644 --- a/apps/sim/lib/mothership/request/application/removal-stops-runs.integration.ts +++ b/apps/sim/lib/mothership/request/application/removal-stops-runs.integration.ts @@ -4,6 +4,7 @@ * paths and the worker's run-control read. A local HTTP server stands in for the worker and * records each Stop pushed to it. */ +import type { Principal } from '@sim/auth/principal' import { createSessionPrincipal } from '@sim/testing/factories/principal.factory' import { createDeferred } from '@sim/testing/helpers/deferred' import { authMock, authMockFns } from '@sim/testing/mocks/auth.mock' @@ -45,6 +46,20 @@ const { redisUrl, inheritedEnv, worker } = await vi.hoisted(async () => { vi.mock('@/lib/auth', () => authMock) +/** Every stream a Stop delivery was started for, recorded when delivery is requested. */ +const requestedDeliveries = vi.hoisted(() => [] as string[]) +vi.mock('@/lib/mothership/request/application/stop-delivery', async (importOriginal) => { + const actual = + await importOriginal() + return { + ...actual, + deliverRunStops: (...args: Parameters) => { + requestedDeliveries.push(...args[0].map((run) => run.streamId)) + actual.deliverRunStops(...args) + }, + } +}) + import { db } from '@sim/db' import { copilotAsyncToolCalls, @@ -55,6 +70,9 @@ import { member, organization, permissions, + scimConnection, + scimGroupMapping, + scimGroupMember, user, workspace, } from '@sim/db/schema' @@ -87,6 +105,13 @@ import { AccountDeletionBlockedError, deleteUserAccount } from '@/lib/users/acco import { revokeWorkspaceAccessTx } from '@/lib/workspaces/access/workspace-access' import { removeWorkspaceMemberRecord } from '@/lib/workspaces/permissions/member-removal-store' import { DELETE as deleteWorkspace } from '@/app/api/workspaces/[id]/route' +import { replaceScimGroup } from '@/ee/scim/lib/application/groups/manage-groups' +import { + PROJECTION_BATCH_SIZE, + reconcileUsersProjectionInBatches, +} from '@/ee/scim/lib/projection/reconcile-user' +import { insertScimGroup } from '@/ee/scim/lib/repository/groups' +import { insertScimUser } from '@/ee/scim/lib/repository/users' const userIds: string[] = [] const workspaceIds: string[] = [] @@ -326,6 +351,109 @@ async function whileHolding(chatId: string, run: (holder: number) => Promise< } } +type DirectoryTarget = { workspaceId: string } | { role: 'admin' } + +/** Points the group's only mapping at `target`, as an administrator's mapping edit does. */ +async function mapGroup(groupId: string, target: DirectoryTarget) { + await db.delete(scimGroupMapping).where(eq(scimGroupMapping.groupId, groupId)) + await db.insert(scimGroupMapping).values({ + id: generateId(), + groupId, + ...('workspaceId' in target + ? { + targetKind: 'workspace', + workspaceId: target.workspaceId, + permissionType: 'write' as const, + } + : { targetKind: 'org_role', role: target.role }), + }) +} + +/** A directory connection for the organization, with one group mapped to `target`. */ +async function seedDirectory(organizationId: string, target: DirectoryTarget) { + const connectionId = generateId() + await db.insert(scimConnection).values({ id: connectionId, organizationId }) + const group = await insertScimGroup(db, { connectionId, displayName: generateId() }) + await mapGroup(group.id, target) + const principal: Principal = { + kind: 'scim_connection', + organizationId, + connectionId, + credentialId: generateId(), + scopes: ['groups:write'], + } + /** Sets the group's members, as a directory's full-group PUT does. */ + const setGroupMembers = (memberIds: string[]) => + replaceScimGroup.execute({ + principal, + input: { groupId: group.id, group: { displayName: group.displayName, memberIds } }, + }) + /** Re-projects these users in bounded transactions, as the scheduled pass and mapping edits do. */ + const reconcile = (scimUserIds: string[]) => + reconcileUsersProjectionInBatches({ connectionId, organizationId, scimUserIds, settings: {} }) + return { + connectionId, + groupId: group.id, + setGroupMembers, + reconcile, + remap: (next: DirectoryTarget) => mapGroup(group.id, next), + } +} + +/** An organization member the directory provisions, returning their directory user id. */ +async function seedDirectoryMember(connectionId: string, organizationId: string) { + const userId = await seedUser() + await db.insert(member).values({ id: generateId(), organizationId, userId, role: 'member' }) + const { id } = await insertScimUser(db, { + connectionId, + userId, + attributes: { + userName: `${userId}@run-removal.test`, + active: true, + name: { formatted: 'Run removal fixture' }, + emails: [{ value: `${userId}@run-removal.test`, primary: true }], + }, + active: true, + }) + return { userId, scimUserId: id } +} + +/** Fails, inside its transaction, any projection that records a grant for this directory user. */ +async function refuseGrantsFor(scimUserId: string) { + const name = `refuse_grant_${scimUserId.replaceAll('-', '_')}` + await db.execute( + sql.raw(`CREATE FUNCTION ${name}() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'projection refused'; END $$`) + ) + await db.execute( + sql.raw(`CREATE TRIGGER ${name} BEFORE INSERT ON scim_projection_grant FOR EACH ROW + WHEN (NEW.scim_user_id = '${scimUserId}') EXECUTE FUNCTION ${name}()`) + ) + return async () => { + await db.execute(sql.raw(`DROP TRIGGER ${name} ON scim_projection_grant`)) + await db.execute(sql.raw(`DROP FUNCTION ${name}()`)) + } +} + +const REFUSED = { cause: { message: 'projection refused' } } + +/** Directory members, ordered as a projection visits them. */ +async function seedDirectoryMembersInOrder( + connectionId: string, + organizationId: string, + count: number +) { + const members = [] + for (let index = 0; index < count; index += 1) { + members.push(await seedDirectoryMember(connectionId, organizationId)) + } + return members.sort((left, right) => (left.scimUserId < right.scimUserId ? -1 : 1)) +} + +/** Stop deliveries started for this test's runs; one is started before its write returns. */ +const requestedStops = () => + requestedDeliveries.filter((streamId) => testStreams.has(streamId)).sort() + /** Queues admission and deletion behind a held chat row in the order `first` names. */ async function queueOnChat( chatId: string, @@ -621,6 +749,130 @@ describe.runIf(Boolean(redisUrl))('removing a run’s chat, owner or access stop expect(await runControl(run)).toEqual(STOPPED) }) + it('pushes Stop for a directory group removal once its transaction commits', async () => { + const ownerId = await seedUser() + const { organizationId, workspaceId } = await seedOrganization(ownerId) + const directory = await seedDirectory(organizationId, { workspaceId }) + const removed = await seedDirectoryMember(directory.connectionId, organizationId) + await directory.setGroupMembers([removed.scimUserId]) + const run = await seedRun({ userId: removed.userId, workspaceId }) + const ownerRun = await seedRun({ userId: ownerId, workspaceId }) + + await directory.setGroupMembers([]) + + await expect.poll(pushedStops).toEqual([run.streamId]) + expect(await runControl(run)).toEqual(STOPPED) + expect(await runControl(ownerRun)).toEqual(LIVE) + }) + + it('pushes Stop for a directory demotion that ends an admin’s derived access', async () => { + const ownerId = await seedUser() + const { organizationId, workspaceId } = await seedOrganization(ownerId) + const directory = await seedDirectory(organizationId, { role: 'admin' }) + const demoted = await seedDirectoryMember(directory.connectionId, organizationId) + await directory.setGroupMembers([demoted.scimUserId]) + const derived = await seedRun({ userId: demoted.userId, workspaceId }) + + await directory.setGroupMembers([]) + + await expect.poll(pushedStops).toEqual([derived.streamId]) + expect(await runControl(derived)).toEqual(STOPPED) + }) + + it('pushes no Stop when the directory group write that stopped the run rolls back', async () => { + const ownerId = await seedUser() + const { organizationId, workspaceId } = await seedOrganization(ownerId) + const directory = await seedDirectory(organizationId, { workspaceId }) + const [removed, added] = await seedDirectoryMembersInOrder( + directory.connectionId, + organizationId, + 2 + ) + await directory.setGroupMembers([removed.scimUserId]) + const run = await seedRun({ userId: removed.userId, workspaceId }) + const allowGrants = await refuseGrantsFor(added.scimUserId) + + try { + /** The removal is projected first and stops the run; the addition then fails the write. */ + await expect(directory.setGroupMembers([added.scimUserId])).rejects.toMatchObject(REFUSED) + } finally { + await allowGrants() + } + + expect(requestedStops()).toEqual([]) + expect(await hasStop(run)).toBe(false) + expect(await runControl(run)).toEqual(LIVE) + }) + + it('pushes Stop for a committed batch when a later batch rolls back, and none for that batch', async () => { + const ownerId = await seedUser() + const { organizationId, workspaceId } = await seedOrganization(ownerId) + const directory = await seedDirectory(organizationId, { workspaceId }) + const members = await seedDirectoryMembersInOrder( + directory.connectionId, + organizationId, + PROJECTION_BATCH_SIZE + 2 + ) + const committed = members[0] + const rolledBack = members[PROJECTION_BATCH_SIZE] + const refused = members[PROJECTION_BATCH_SIZE + 1] + await directory.setGroupMembers([committed.scimUserId, rolledBack.scimUserId]) + const committedRun = await seedRun({ userId: committed.userId, workspaceId }) + const rolledBackRun = await seedRun({ userId: rolledBack.userId, workspaceId }) + /** Membership drifts behind the projection's back, as the scheduled pass repairs it. */ + await db.delete(scimGroupMember).where(eq(scimGroupMember.groupId, directory.groupId)) + await db + .insert(scimGroupMember) + .values({ id: generateId(), groupId: directory.groupId, scimUserId: refused.scimUserId }) + const allowGrants = await refuseGrantsFor(refused.scimUserId) + + try { + await expect( + directory.reconcile(members.map((entry) => entry.scimUserId)) + ).rejects.toMatchObject(REFUSED) + } finally { + await allowGrants() + } + + expect(requestedStops()).toEqual([committedRun.streamId]) + await expect.poll(pushedStops).toEqual([committedRun.streamId]) + expect(await runControl(committedRun)).toEqual(STOPPED) + expect(await hasStop(rolledBackRun)).toBe(false) + expect(await runControl(rolledBackRun)).toEqual(LIVE) + }) + + it('keeps a demoted admin’s run when the same pass grants its workspace', async () => { + const ownerId = await seedUser() + const { organizationId, workspaceId } = await seedOrganization(ownerId) + const directory = await seedDirectory(organizationId, { role: 'admin' }) + const demoted = await seedDirectoryMember(directory.connectionId, organizationId) + await directory.setGroupMembers([demoted.scimUserId]) + const run = await seedRun({ userId: demoted.userId, workspaceId }) + + await directory.remap({ workspaceId }) + await directory.reconcile([demoted.scimUserId]) + + expect(requestedStops()).toEqual([]) + expect(await hasStop(run)).toBe(false) + expect(await runControl(run)).toEqual(LIVE) + }) + + it('keeps a run when the same pass trades its workspace grant for the admin role', async () => { + const ownerId = await seedUser() + const { organizationId, workspaceId } = await seedOrganization(ownerId) + const directory = await seedDirectory(organizationId, { workspaceId }) + const promoted = await seedDirectoryMember(directory.connectionId, organizationId) + await directory.setGroupMembers([promoted.scimUserId]) + const run = await seedRun({ userId: promoted.userId, workspaceId }) + + await directory.remap({ role: 'admin' }) + await directory.reconcile([promoted.scimUserId]) + + expect(requestedStops()).toEqual([]) + expect(await hasStop(run)).toBe(false) + expect(await runControl(run)).toEqual(LIVE) + }) + it('stops a removed organization member’s Assistant runs', async () => { const ownerId = await seedUser() const memberId = await seedUser() diff --git a/apps/sim/lib/organizations/application/member-removal.test.ts b/apps/sim/lib/organizations/application/member-removal.test.ts index ba6816ac84a..f89aa23366b 100644 --- a/apps/sim/lib/organizations/application/member-removal.test.ts +++ b/apps/sim/lib/organizations/application/member-removal.test.ts @@ -56,7 +56,7 @@ const principal = { const input = { organizationId: 'org', userId: 'target' } beforeEach(() => { resetDbChainMock() - mocks.remove.mockResolvedValue({ success: true, billingActions: {} }) + mocks.remove.mockResolvedValue({ success: true, billingActions: {}, stoppedRuns: [] }) mocks.seats.mockResolvedValue({ changed: false }) mocks.external.mockResolvedValue({ success: true, @@ -64,6 +64,7 @@ beforeEach(() => { permissionGroupsRevoked: 1, credentialMembershipsRevoked: 1, pendingInvitationsCancelled: 1, + stoppedRuns: [], }) }) function target(role = 'admin') { @@ -138,7 +139,7 @@ describe('organization member removal', () => { 'Cannot remove the workspace billing account. Please reassign billing first.', ])('preserves lifecycle refusal %s', async (error) => { target() - mocks.remove.mockResolvedValueOnce({ success: false, error }) + mocks.remove.mockResolvedValueOnce({ success: false, error, stoppedRuns: [] }) await expect(removeOrganizationMember.execute({ principal, input })).rejects.toMatchObject({ code: 'validation', }) diff --git a/apps/sim/lib/organizations/member-manager.test.ts b/apps/sim/lib/organizations/member-manager.test.ts index bac96c57fcf..864f7581e2f 100644 --- a/apps/sim/lib/organizations/member-manager.test.ts +++ b/apps/sim/lib/organizations/member-manager.test.ts @@ -51,7 +51,7 @@ const target = { beforeEach(() => { vi.resetAllMocks() resetDbChainMock() - mocks.remove.mockResolvedValue({ success: true }) + mocks.remove.mockResolvedValue({ success: true, stoppedRuns: [] }) mocks.seats.mockResolvedValue({ changed: false }) }) describe('organization member managers', () => { @@ -125,7 +125,11 @@ describe('organization member managers', () => { it('reports membership added before external-removal preflight as a conflict', async () => { queueTableRows(member, []) queueTableRows(user, [{ id: 'target', name: 'Person', email: 'person@example.com' }]) - mocks.external.mockResolvedValue({ success: false, error: 'User is an organization member' }) + mocks.external.mockResolvedValue({ + success: false, + error: 'User is an organization member', + stoppedRuns: [], + }) await expect(removeOrganizationMemberRecord(input)).rejects.toMatchObject({ code: 'conflict', diff --git a/apps/sim/lib/organizations/member-manager.ts b/apps/sim/lib/organizations/member-manager.ts index 230911c81f5..45e6e2252e6 100644 --- a/apps/sim/lib/organizations/member-manager.ts +++ b/apps/sim/lib/organizations/member-manager.ts @@ -73,7 +73,7 @@ export async function removeOrganizationMemberRecord(input: { if (!external) throw new OrchestrationError('not_found', 'Member not found') /** The stopped runs are delivered, never returned to the response or audit. */ const { stoppedRuns, ...removal } = await removeExternalUserFromOrganizationWorkspaces(input) - deliverRunStops(stoppedRuns ?? []) + deliverRunStops(stoppedRuns) if (!removal.success) { const message = removal.error || 'External workspace member not found' const code = @@ -98,7 +98,7 @@ export async function removeOrganizationMemberRecord(input: { memberId: target.id, onError: 'throw', }) - deliverRunStops(stoppedRuns ?? []) + deliverRunStops(stoppedRuns) if (!removal.success) { const message = removal.error || 'Failed to remove user from organization' const code = diff --git a/apps/sim/lib/organizations/members/lifecycle.ts b/apps/sim/lib/organizations/members/lifecycle.ts index c5c7004ce19..af7fe445c35 100644 --- a/apps/sim/lib/organizations/members/lifecycle.ts +++ b/apps/sim/lib/organizations/members/lifecycle.ts @@ -110,7 +110,16 @@ export type ChangeMemberRoleResult = */ export async function changeMemberRoleTx( tx: DbTransaction, - params: { organizationId: string; userId: string; role: OrganizationMemberRole } + params: { + organizationId: string + userId: string + role: OrganizationMemberRole + /** + * Leaves the member's runs to a caller that changes more access in this transaction and + * stops runs once that access is final; `stoppedRuns` is then empty. + */ + deferRunStops?: boolean + } ): Promise { await acquireOrganizationUserMutationLocks(tx, { userId: params.userId, @@ -132,7 +141,7 @@ export async function changeMemberRoleTx( await tx.update(member).set({ role: params.role }).where(eq(member.id, current.id)) /** An organization admin holds derived access to every organization workspace; a demotion ends it. */ const stoppedRuns = - isOrgAdminRole(current.role) && !isOrgAdminRole(params.role) + !params.deferRunStops && isOrgAdminRole(current.role) && !isOrgAdminRole(params.role) ? await stopRunsWithoutWorkspaceAccessTx( tx, params.userId, diff --git a/apps/sim/lib/workspaces/access/workspace-access.ts b/apps/sim/lib/workspaces/access/workspace-access.ts index a23e7582f0d..9bae43cf9fb 100644 --- a/apps/sim/lib/workspaces/access/workspace-access.ts +++ b/apps/sim/lib/workspaces/access/workspace-access.ts @@ -136,7 +136,16 @@ export type RevokeWorkspaceAccessResult = /** Reassigns shared resources and workspace ownership before revoking access in the same transaction. */ export async function revokeWorkspaceAccessTx( tx: DbTransaction, - params: { workspaceId: string; userId: string; expectedPermissionId?: string } + params: { + workspaceId: string + userId: string + expectedPermissionId?: string + /** + * Leaves the user's runs to a caller that changes more access in this transaction and + * stops runs once that access is final; `stoppedRuns` is then empty. + */ + deferRunStops?: boolean + } ): Promise { await lockProjectsForResourceDepartureTx(tx, [params.workspaceId]) if (params.expectedPermissionId !== undefined) { @@ -196,9 +205,9 @@ export async function revokeWorkspaceAccessTx( await revokeWorkspaceCredentialMembershipsTx(tx, params.workspaceId, params.userId) await removeWorkspaceSkillMembershipsTx(tx, params.workspaceId, params.userId) - const stoppedRuns = await stopRunsWithoutWorkspaceAccessTx(tx, params.userId, [ - params.workspaceId, - ]) + const stoppedRuns = params.deferRunStops + ? [] + : await stopRunsWithoutWorkspaceAccessTx(tx, params.userId, [params.workspaceId]) return { revoked: true, ownershipTransferred, stoppedRuns } } diff --git a/apps/sim/lib/workspaces/permissions/member-removal-store.ts b/apps/sim/lib/workspaces/permissions/member-removal-store.ts index 2f42eb129c8..8f37f458230 100644 --- a/apps/sim/lib/workspaces/permissions/member-removal-store.ts +++ b/apps/sim/lib/workspaces/permissions/member-removal-store.ts @@ -154,7 +154,7 @@ export async function removeWorkspaceMemberRecord( }) /** The workspace's runs were delivered with the revocation above. */ const delivered = new Set(revocation.stoppedRuns.map((run) => run.streamId)) - deliverRunStops((removal.stoppedRuns ?? []).filter((run) => !delivered.has(run.streamId))) + deliverRunStops(removal.stoppedRuns.filter((run) => !delivered.has(run.streamId))) if (removal.success && removal.removed) { organizationRemoval = true