diff --git a/src/backend/services/team/TeamService.ts b/src/backend/services/team/TeamService.ts index d32e8e105..818fc42f9 100644 --- a/src/backend/services/team/TeamService.ts +++ b/src/backend/services/team/TeamService.ts @@ -638,22 +638,12 @@ export class TeamService extends PuterService { // Otherwise they keep working, unreachable through a deleted team. let page = await this.stores.team.listMembers(teamUid, { limit: 200 }); for (;;) { - // Bounded: a seat costs a mail send, and a full team is 40 of them. - const outcomes = await runWithConcurrencyLimitSettled( + disabled += await this.#disableSeatsForDelete( page.items.filter((m) => Number(m.org_owned) === 1), - TEARDOWN_CONCURRENCY, - (member) => - this.#disableSeatForDelete( - member, - team, - actorUserId, - billing, - ), + team, + actorUserId, + billing, ); - disabled += outcomes.filter((o) => o.status === 'fulfilled').length; - // Settled: a rejection cannot cancel its peers, so finish the page. - const failed = outcomes.find((o) => o.status === 'rejected'); - if (failed) throw failed.reason; if (!page.cursor) break; page = await this.stores.team.listMembers(teamUid, { limit: 200, @@ -675,41 +665,72 @@ export class TeamService extends PuterService { }); } - /** As `disableMember`; a suspended seat is skipped so a retry resumes. */ - async #disableSeatForDelete( - member: TeamMemberRow, + /** + * One page of seats, as `disableMember` does one member. Everything that + * answers for the whole page -- who is already off, what they hold, the + * audit -- goes out once; what is genuinely per account stays per account. + * Returns how many of them end up disabled. + */ + async #disableSeatsForDelete( + seats: TeamMemberRow[], team: TeamRow, actorUserId: number, billing: TeamBillingContext, - ): Promise { - const current = await this.stores.user.getByProperty( - 'id', - member.user_id, - { force: true }, + ): Promise { + if (seats.length === 0) return 0; + + // From the primary: a cached `suspended` would re-charge on a retry. + const suspended = await this.stores.user.getSuspendedByIds( + seats.map((m) => m.user_id), ); - if (current?.suspended) return; + const todo = seats.filter((m) => !suspended.get(m.user_id)); + // Already off, and still accounts the team is losing. + if (todo.length === 0) return seats.length; - await this.stores.team.appendAudit({ - teamId: team.id, - userId: member.user_id, - actorUserId, - action: 'disable', - reason: 'team_deleted', - }); - const held = await this.#heldBytes(member.user_id); - await this.#suspend(member.user_id); + const ids = todo.map((m) => m.user_id); + const held = await this.stores.fsEntry.getHeldBytesForUsers(ids); + const rows = await this.stores.user.getByIds(ids); + // Recorded first, so a failed append cannot leave an unlogged suspension. + await this.stores.team.appendAuditMany( + todo.map((m) => ({ + teamId: team.id, + userId: m.user_id, + actorUserId, + action: 'disable', + reason: 'team_deleted', + })), + ); - // Per seat, not one bulk event: the byte charge is per account. - this.#emitBilling('team.account.disabled', { - ...billing, - user_id: member.user_id, - user_uuid: member.uuid, - username: member.username, - held_bytes: held, - }); + // Bounded: what is left per seat is a write and a mail send. + const outcomes = await runWithConcurrencyLimitSettled( + todo, + TEARDOWN_CONCURRENCY, + async (member) => { + await this.#suspend(member.user_id); - // The team notice covers the disabling, so a member is told once. - await this.#notifyMember(member.user_id, 'team_closed', team); + // Per seat, not one bulk event: the byte charge is per account. + this.#emitBilling('team.account.disabled', { + ...billing, + user_id: member.user_id, + user_uuid: member.uuid, + username: member.username, + held_bytes: held.get(member.user_id) ?? 0, + }); + + // The team notice covers the disabling, so a member is told once. + await this.#notifyUser( + rows.get(member.user_id), + 'team_closed', + team, + ); + }, + ); + + const done = outcomes.filter((o) => o.status === 'fulfilled').length; + // Settled: a rejection cannot cancel its peers, so finish the page. + const failed = outcomes.find((o) => o.status === 'rejected'); + if (failed) throw failed.reason; + return seats.length - todo.length + done; } /** Team owner only. Readable after deletion -- that is the point of it. */ diff --git a/src/backend/stores/fs/FSEntryStore.ts b/src/backend/stores/fs/FSEntryStore.ts index 84a81a783..d33985a1c 100644 --- a/src/backend/stores/fs/FSEntryStore.ts +++ b/src/backend/stores/fs/FSEntryStore.ts @@ -3344,6 +3344,38 @@ export class FSEntryStore extends PuterStore { return Number(rows[0]?.totalUsage ?? 0); } + /** + * Held bytes for several accounts in one grouped read, as `getHeldBytes` + * does for one. + */ + async getHeldBytesForUsers( + userIds: number[], + ): Promise> { + const out = new Map(); + const unique = [ + ...new Set( + (Array.isArray(userIds) ? userIds : []).filter( + (id): id is number => typeof id === 'number', + ), + ), + ]; + for (const id of unique) out.set(id, 0); + + for (let offset = 0; offset < unique.length; offset += 200) { + const chunk = unique.slice(offset, offset + 200); + const placeholders = chunk.map(() => '?').join(', '); + const rows = (await this.clients.db.read( + `SELECT user_id, COALESCE(SUM(size), 0) AS ${this.clients.db.quoteIdentifier('totalUsage')} ` + + `FROM fsentries WHERE user_id IN (${placeholders}) GROUP BY user_id`, + chunk, + )) as { user_id: number; totalUsage: number }[]; + for (const row of rows) { + out.set(Number(row.user_id), Number(row.totalUsage ?? 0)); + } + } + return out; + } + /** * `consistentRead` reads both queries from the primary instead of a replica * — used for the exact re-check when a replica-backed read would reject a diff --git a/src/backend/stores/team/TeamStore.ts b/src/backend/stores/team/TeamStore.ts index 4e08f4c67..1cbd46cab 100644 --- a/src/backend/stores/team/TeamStore.ts +++ b/src/backend/stores/team/TeamStore.ts @@ -820,6 +820,38 @@ export class TeamStore extends PuterStore { ); } + /** The same rows as `appendAudit`, in one statement per chunk. */ + async appendAuditMany( + entries: Array<{ + teamId: number; + userId: number; + actorUserId: number; + action: string; + reason?: string | null; + }>, + ): Promise { + if (entries.length === 0) return; + for (let offset = 0; offset < entries.length; offset += 200) { + const chunk = entries.slice(offset, offset + 200); + const values = chunk.map(() => '(?, ?, ?, ?, ?, ?, ?)').join(', '); + const params = chunk.flatMap((e) => [ + e.teamId, + e.teamId, + e.userId, + e.userId, + e.actorUserId, + e.action, + e.reason ?? null, + ]); + await this.clients.db.write( + 'INSERT INTO `audit_team_membership` ' + + '(`group_id`, `group_id_keep`, `user_id`, `user_id_keep`, ' + + `\`actor_user_id\`, \`action\`, \`reason\`) VALUES ${values}`, + params, + ); + } + } + /** The whole team's audit, newest first, keyset-paginated on `id`. */ async listAudit( teamId: number, diff --git a/src/backend/stores/user/UserStore.ts b/src/backend/stores/user/UserStore.ts index 176ff4c9a..c077b25c5 100644 --- a/src/backend/stores/user/UserStore.ts +++ b/src/backend/stores/user/UserStore.ts @@ -248,6 +248,40 @@ export class UserStore extends PuterStore { return this.getByProperty('referral_code', normalized, opts); } + /** + * Which of these accounts are suspended, read from the primary. + * + * Deliberately not `getByIds`, which answers from cache: a caller deciding + * whether to suspend an account needs the row as it is now, not as it was + * when something last cached it. + */ + async getSuspendedByIds(ids: number[]): Promise> { + const out = new Map(); + const unique = [ + ...new Set( + (Array.isArray(ids) ? ids : []).filter( + (id): id is number => typeof id === 'number', + ), + ), + ]; + for ( + let offset = 0; + offset < unique.length; + offset += BULK_QUERY_CHUNK_SIZE + ) { + const chunk = unique.slice(offset, offset + BULK_QUERY_CHUNK_SIZE); + const placeholders = chunk.map(() => '?').join(', '); + const rows = (await this.clients.db.pread( + `SELECT \`id\`, \`suspended\` FROM \`user\` WHERE \`id\` IN (${placeholders})`, + chunk, + )) as Array<{ id: number; suspended: unknown }>; + for (const row of rows) { + out.set(Number(row.id), Boolean(row.suspended)); + } + } + return out; + } + /** * Batched lookup by id. Dedupes input ids, reads cache via a pipelined * MGET, and resolves remaining misses with a single `SELECT … WHERE id IN