diff --git a/apps/sim/lib/billing/organizations/membership.ts b/apps/sim/lib/billing/organizations/membership.ts index 127a9b26e0c..a76ec3243e2 100644 --- a/apps/sim/lib/billing/organizations/membership.ts +++ b/apps/sim/lib/billing/organizations/membership.ts @@ -61,6 +61,7 @@ import { isRetryableTransactionError } from '@/lib/db/transaction' import type { DbOrTx, DbTransaction } from '@/lib/db/types' import { acquireInvitationMutationLocks } from '@/lib/invitations/locks' import { type StoppedRun, stopLiveRunsTx } from '@/lib/mothership/async-runs/repository' +import { enqueueMemoryCleanup } from '@/lib/mothership/memories/cleanup-outbox' import { requireMemberManagementAuthority } from '@/lib/organizations/members/authority' import { revokePersonalApiKeysTx, @@ -1192,6 +1193,13 @@ export async function transferUserBetweenOrganizations( if (deleted.length === 0) { throw new Error('Member could not be transferred because their role changed') } + await enqueueMemoryCleanup(tx, [ + { + kind: 'organization_member', + organizationId: params.sourceOrganizationId, + userId: params.userId, + }, + ]) const cancelledInvitations = invitationIds.length ? await tx @@ -1407,6 +1415,7 @@ export async function removeUserFromOrganization( 'The membership changed before removal. Refresh and try again.' ) } + await enqueueMemoryCleanup(tx, [{ kind: 'organization_member', organizationId, userId }]) if (!skipBillingLogic) { await enqueueOutboxEvent(tx, MEMBER_BILLING_RECONCILIATION_EVENT_TYPE, { @@ -1679,6 +1688,7 @@ export async function removeExternalUserFromOrganizationWorkspaces(params: { ) await setOrgMemberUsageLimit(organizationId, userId, null, undefined, tx) + await enqueueMemoryCleanup(tx, [{ kind: 'organization_member', organizationId, userId }]) const cancelledInvitations = invitationIds.length ? await tx diff --git a/apps/sim/lib/mothership/generated/memory-notes-access.ts b/apps/sim/lib/mothership/generated/memory-notes-access.ts index b1c414e4986..94361b84b4f 100644 --- a/apps/sim/lib/mothership/generated/memory-notes-access.ts +++ b/apps/sim/lib/mothership/generated/memory-notes-access.ts @@ -53,6 +53,14 @@ export const MemoryCleanupInput = z.strictObject({ z.strictObject({ kind: z.literal("private_owner"), userId: z.string().min(1).max(200) }), z.strictObject({ kind: z.literal("workspace"), workspaceId: z.uuid() }), z.strictObject({ kind: z.literal("project"), projectId: z.uuid() }), + // Erase private memory held in an organization: all of it when the organization is + // deleted, one person's when they leave it (D106). + z.strictObject({ kind: z.literal("organization"), organizationId: z.string().min(1).max(200) }), + z.strictObject({ + kind: z.literal("organization_member"), + organizationId: z.string().min(1).max(200), + userId: z.string().min(1).max(200), + }), ]), }); export type MemoryCleanupInput = z.infer; diff --git a/apps/sim/lib/mothership/memories/offboarding.integration.ts b/apps/sim/lib/mothership/memories/offboarding.integration.ts new file mode 100644 index 00000000000..8b5bf8a8f5e --- /dev/null +++ b/apps/sim/lib/mothership/memories/offboarding.integration.ts @@ -0,0 +1,145 @@ +import { db } from '@sim/db' +import { member, organization, outboxEvent, permissions, user, workspace } from '@sim/db/schema' +import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure' +import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' +import { generateId } from '@sim/utils/id' +import { and, eq, inArray, sql } from 'drizzle-orm' +import { afterAll, beforeAll, expect, it } from 'vitest' + +readTestDatabaseUrl() + +const ownerId = generateId() +const memberId = generateId() +const externalId = generateId() +const organizationId = generateId() +const membershipId = generateId() +const workspaceId = generateId() +let runtime: { + membership: typeof import('@/lib/billing/organizations/membership') + resources: typeof import('@/lib/organizations/resource-cleanup') + event: typeof import('@/lib/mothership/memories/cleanup-event') +} + +beforeAll(async () => { + // The cleanup outbox reads its worker credential and destination when it loads. + process.env.SIM_AGENT_API_URL = 'http://127.0.0.1:9' + process.env.COPILOT_API_KEY = 'memory-offboarding-fixture-key' + const [membership, resources, event] = await Promise.all([ + import('@/lib/billing/organizations/membership'), + import('@/lib/organizations/resource-cleanup'), + import('@/lib/mothership/memories/cleanup-event'), + ]) + runtime = { membership, resources, event } + await db.insert(user).values( + [ownerId, memberId, externalId].map((id) => ({ + id, + name: 'Offboarding fixture', + email: `${id}@memory.test`, + emailVerified: true, + createdAt: new Date(), + updatedAt: new Date(), + })) + ) + await db + .insert(organization) + .values({ id: organizationId, name: 'Offboarding', slug: organizationId }) + await db.insert(member).values([ + { id: generateId(), organizationId, userId: ownerId, role: 'owner', createdAt: new Date() }, + { id: membershipId, organizationId, userId: memberId, role: 'member', createdAt: new Date() }, + ]) + // The external collaborator works in one of the organization's workspaces without membership. + await insertWorkspaceFixture(db, { + id: workspaceId, + name: 'Offboarding', + ownerId, + billedAccountUserId: ownerId, + organizationId, + workspaceMode: 'organization', + }) + await db.insert(permissions).values({ + id: generateId(), + userId: externalId, + entityType: 'workspace', + entityId: workspaceId, + permissionType: 'write', + }) +}, 60_000) + +afterAll(async () => { + await db + .delete(outboxEvent) + .where( + and( + eq(outboxEvent.eventType, runtime.event.MEMORY_CLEANUP_EVENT), + sql`${outboxEvent.payload}->'request'->'target'->>'organizationId' = ${organizationId}` + ) + ) + await deleteWorkspaceFixture(db, eq(workspace.id, workspaceId)) + await db.delete(organization).where(eq(organization.id, organizationId)) + await db.delete(user).where(inArray(user.id, [ownerId, memberId, externalId])) +}) + +/** The memory erasures queued for this organization, as `kind:userId` (or `kind`). */ +async function queued(): Promise { + const rows = await db + .select({ payload: outboxEvent.payload }) + .from(outboxEvent) + .where( + and( + eq(outboxEvent.eventType, runtime.event.MEMORY_CLEANUP_EVENT), + sql`${outboxEvent.payload}->'request'->'target'->>'organizationId' = ${organizationId}` + ) + ) + return rows + .map(({ payload }) => { + const target = (payload as { request: { target: { kind: string; userId?: string } } }).request + .target + return target.userId ? `${target.kind}:${target.userId}` : target.kind + }) + .sort() +} + +it('queues private memory erasure with each committed departure and organization deletion, never on rollback', async () => { + // A removal whose actor lacks authority rolls back with nothing queued. + await expect( + runtime.membership.removeUserFromOrganization({ + organizationId, + userId: memberId, + memberId: membershipId, + actorUserId: externalId, + skipBillingLogic: true, + onError: 'throw', + }) + ).rejects.toThrow() + expect(await queued()).toEqual([]) + + await runtime.membership.removeUserFromOrganization({ + organizationId, + userId: memberId, + memberId: membershipId, + skipBillingLogic: true, + onError: 'throw', + }) + expect( + await runtime.membership.removeExternalUserFromOrganizationWorkspaces({ + organizationId, + userId: externalId, + }) + ).toMatchObject({ success: true, workspaceAccessRevoked: 1 }) + expect(await queued()).toEqual( + [`organization_member:${externalId}`, `organization_member:${memberId}`].sort() + ) + + // Organization deletion queues the erasure in the delete transaction: a rollback drops it. + await expect( + db.transaction(async (tx) => { + await runtime.resources.enqueueOrganizationResourceCleanup(tx, organizationId) + throw new Error('rollback fixture') + }) + ).rejects.toThrow('rollback fixture') + expect(await queued()).not.toContain('organization') + await db.transaction(async (tx) => { + await runtime.resources.enqueueOrganizationResourceCleanup(tx, organizationId) + }) + expect(await queued()).toContain('organization') +}, 30_000) diff --git a/apps/sim/lib/organizations/resource-cleanup.ts b/apps/sim/lib/organizations/resource-cleanup.ts index d6eb271cc10..0ff0ee39d4a 100644 --- a/apps/sim/lib/organizations/resource-cleanup.ts +++ b/apps/sim/lib/organizations/resource-cleanup.ts @@ -8,6 +8,7 @@ import { env } from '@/lib/core/config/env' import { enqueueOutboxEvent, type OutboxHandlerRegistry } from '@/lib/core/outbox/service' import { mapWithConcurrency } from '@/lib/core/utils/concurrency' import type { DbTransaction } from '@/lib/db/types' +import { enqueueMemoryCleanup } from '@/lib/mothership/memories/cleanup-outbox' import { ORGANIZATION_RESOURCE_CLEANUP_EVENT } from '@/lib/organizations/resource-cleanup-event' import { deleteFile } from '@/lib/uploads/core/storage-service' @@ -35,7 +36,8 @@ const organizationResourceCleanupSchema = z.discriminatedUnion('kind', [ * Captures canonical bindings before the org FK cascade removes them. The org * row lock excludes new direct resources; a rebound resource is checked again * by the worker. Each keyset page becomes its own bounded outbox event in the - * caller's delete transaction, so rollback cannot leave cleanup work behind. + * caller's delete transaction, so rollback cannot leave cleanup work behind. The + * members' private Mothership memory in the organization is erased the same way. */ export async function enqueueOrganizationResourceCleanup( tx: DbTransaction, @@ -105,6 +107,8 @@ export async function enqueueOrganizationResourceCleanup( chatCursor = chats[chats.length - 1].id if (chats.length < CLEANUP_BATCH_SIZE) break } + + await enqueueMemoryCleanup(tx, [{ kind: 'organization', organizationId }]) } export const organizationResourceCleanupOutboxHandlers = {