From 0900bcac252b1392bea8b39ebaaaa5a77dd0a5dd Mon Sep 17 00:00:00 2001 From: Devin Zuczek Date: Mon, 5 Oct 2026 13:10:05 -0400 Subject: [PATCH] [www] token drop performance improvement --- apps/econ/src/balance-db.ts | 135 ++++++++++++++++ apps/econ/src/discord-role-gift.ts | 159 ++++++++++++++----- apps/econ/src/test/integration/api.test.ts | 43 +++++ apps/econ/vitest.config.ts | 8 + apps/notify/src/notifications-hub.ts | 49 ++++++ apps/notify/src/test/integration/api.test.ts | 64 ++++++++ apps/www/src/staff.ts | 130 +++++++++++---- apps/www/src/test/integration/api.test.ts | 129 +++++++++++++++ apps/www/vitest.config.ts | 6 + 9 files changed, 645 insertions(+), 78 deletions(-) diff --git a/apps/econ/src/balance-db.ts b/apps/econ/src/balance-db.ts index bcd8536..6901ce7 100644 --- a/apps/econ/src/balance-db.ts +++ b/apps/econ/src/balance-db.ts @@ -1,6 +1,8 @@ import { BalancePlatform } from '../../notify/src/notification-payloads' import { CurrencyType } from './currency' +import type { GiftContent } from '@repo/domain' + // The currency vocabulary and the Plus reload SQL live in the import-free `currency.ts` so // the Node CLI in @repo/tools can share them; re-exported here so this stays the module // everything else reads balances from. @@ -199,3 +201,136 @@ export async function spendCurrency( .run() return meta.changes > 0 } + +/** One token box to hand over: whose, how many, and the stored box. See {@link payTokenGifts}. */ +export interface TokenGift { + accountId: number + /** Signed: positive credits, negative debits (guarded — see below), zero only boxes. */ + amount: number + /** The box as stored in `received_gift`, serialized as-is. */ + content: GiftContent +} + +/** One box handed over by {@link payTokenGifts}: the resulting balance and the box's id. */ +export interface PaidTokenGift extends TokenGift { + balance: number + giftId: number +} + +/** + * How many accounts one round of {@link payTokenGifts} writes at once. Every statement + * binds three values per account, and D1 allows 100 bound parameters a statement. + */ +export const TOKEN_GIFT_CHUNK = 30 + +/** + * Hand RecCenterTokens in a gift box to MANY accounts at once — the staff drops, the room + * gift and the Discord supporter cron, which all pay a crowd. Paid one account at a time + * (seed, credit, re-read, box: five D1 round trips each) a full server was ~1,250 queries + * in series, past D1's 1,000-queries-an-invocation cap somewhere around 200 players, and + * died partway with the first players paid and the rest not. This pays + * {@link TOKEN_GIFT_CHUNK} accounts a round in two batches of multi-row statements. + * + * Per account it is exactly the one-player grant: the signup grant seeded first (a + * never-touched balance must not start from the gift alone), the balance moved, then a box + * stored. A credit is an upsert; a debit is `amount >= ?` guarded in the UPDATE itself, so + * nobody is overdrawn, and an account that can't cover it is SKIPPED — no box — rather than + * failing the rest. `paid` keeps the order of `gifts`; `skipped` is the rest. A duplicate + * account among `gifts` is paid once, the first entry's amount: a multi-row upsert may not + * name one key twice. + * + * Sends NOTHING: the caller builds the balance frame (the RESULTING total, `balance`, into + * the -2 bucket) and the box's announcement from the rows returned, and should push them in + * one hub call. Nothing here is transactional across rounds: a round that throws leaves the + * earlier ones paid, which the caller reports. + */ +export async function payTokenGifts( + db: D1Database, + gifts: readonly TokenGift[], + startingTokens: number +): Promise<{ paid: PaidTokenGift[]; skipped: number[] }> { + const grants = startingBalances(startingTokens) + const paid: PaidTokenGift[] = [] + const skipped: number[] = [] + + const seen = new Set() + const unique = gifts.filter((g) => !seen.has(g.accountId) && seen.add(g.accountId)) + for (let i = 0; i < unique.length; i += TOKEN_GIFT_CHUNK) { + const chunk = unique.slice(i, i + TOKEN_GIFT_CHUNK) + const credits = chunk.filter((g) => g.amount > 0) + const debits = chunk.filter((g) => g.amount < 0) + const holds = chunk.filter((g) => g.amount === 0) + + // Seed the signup grant, then move the money: one batch, one transaction. A debit's + // guard is per row, so a chunk of debits is one UPDATE per distinct amount. + const statements: D1PreparedStatement[] = [ + db + .prepare( + `INSERT OR IGNORE INTO balance (account_id, currency_type, amount) VALUES ${chunk + .flatMap(() => grants.map(() => '(?, ?, ?)')) + .join(', ')}` + ) + .bind( + ...chunk.flatMap((g) => grants.flatMap((b) => [g.accountId, b.currencyType, b.amount])) + ), + ] + if (credits.length > 0) { + statements.push( + db + .prepare( + `INSERT INTO balance (account_id, currency_type, amount) VALUES ${credits + .map(() => '(?, ?, ?)') + .join(', ')} + ON CONFLICT (account_id, currency_type) DO UPDATE SET amount = amount + excluded.amount + RETURNING account_id, amount` + ) + .bind(...credits.flatMap((g) => [g.accountId, CurrencyType.RecCenterTokens, g.amount])) + ) + } + for (const amount of new Set(debits.map((g) => g.amount))) { + const ids = debits.filter((g) => g.amount === amount).map((g) => g.accountId) + statements.push( + db + .prepare( + `UPDATE balance SET amount = amount - ?1 + WHERE currency_type = ?2 AND amount >= ?1 + AND account_id IN (${ids.map(() => '?').join(', ')}) + RETURNING account_id, amount` + ) + .bind(-amount, CurrencyType.RecCenterTokens, ...ids) + ) + } + if (holds.length > 0) { + statements.push( + db + .prepare( + `SELECT account_id, amount FROM balance + WHERE currency_type = ?1 AND account_id IN (${holds.map(() => '?').join(', ')})` + ) + .bind(CurrencyType.RecCenterTokens, ...holds.map((g) => g.accountId)) + ) + } + const [, ...moved] = await db.batch<{ account_id: number; amount: number }>(statements) + // RETURNING comes back in no promised order; key it, then walk the chunk in its order. + const balances = new Map(moved.flatMap((r) => r.results).map((r) => [r.account_id, r.amount])) + const debited = chunk.filter((g) => balances.has(g.accountId)) + for (const g of chunk) if (!balances.has(g.accountId)) skipped.push(g.accountId) + if (debited.length === 0) continue + + const createdAt = new Date().toISOString() + const { results: boxes } = await db + .prepare( + `INSERT INTO received_gift (account_id, data, created_at) VALUES ${debited + .map(() => '(?, ?, ?)') + .join(', ')} RETURNING id, account_id` + ) + .bind(...debited.flatMap((g) => [g.accountId, JSON.stringify(g.content), createdAt])) + .all<{ id: number; account_id: number }>() + const giftIds = new Map(boxes.map((r) => [r.account_id, r.id])) + + for (const g of debited) { + paid.push({ ...g, balance: balances.get(g.accountId)!, giftId: giftIds.get(g.accountId)! }) + } + } + return { paid, skipped } +} diff --git a/apps/econ/src/discord-role-gift.ts b/apps/econ/src/discord-role-gift.ts index e5bd3a5..b888df2 100644 --- a/apps/econ/src/discord-role-gift.ts +++ b/apps/econ/src/discord-role-gift.ts @@ -4,8 +4,9 @@ * An operator maps Discord ROLE ids to a token amount — `DISCORD_ROLE_TOKENS`, * `=,=` — and every time the cron fires, every account * holding one of those roles is handed that many RecCenterTokens in a gift box. ONE box per - * account per run: an account holding several mapped roles is paid the HIGHEST amount among - * them, not the sum, so the map reads as tiers however Discord stacks the roles. + * account per run: an account holding several mapped roles — across every Discord link it + * has — is paid the HIGHEST amount among them, not the sum, so the map reads as tiers + * however Discord stacks the roles. * * HOW OFTEN is the cron schedule's business, not this module's: `triggers.crons` in * wrangler.jsonc says weekly, and an operator who wants a daily gift changes that line. There @@ -42,13 +43,21 @@ import { PlatformType } from '@repo/domain/src/enums' import { getLinksForPlatform } from '../../auth/src/platform-db' import { NotificationType } from '../../notify/src/notification-types' -import { ALL_PLATFORMS, creditCurrency, CurrencyType, ensureStartingBalances } from './balance-db' +import { + ALL_PLATFORMS, + creditCurrency, + CurrencyType, + ensureStartingBalances, + payTokenGifts, + TOKEN_GIFT_CHUNK, +} from './balance-db' import type { GiftContent } from '@repo/domain' import type { BalanceResponsePayload, GiftPackagePayload, } from '../../notify/src/notification-payloads' +import type { PlayerNotification } from '../../notify/src/notifications-hub' import type { Env } from './context' /** The system "Coach" account — who a box the server hands over is from. */ @@ -178,19 +187,24 @@ export interface RoleGiftSummary { granted: number /** Tokens handed out this run, across every box. */ tokens: number - /** Grants that failed partway — logged with the account and role to sort out by hand. */ + /** Accounts whose grant failed — logged with the account and role to sort out by hand. */ failed: number } /** - * One run of the gift: every Discord link holding a mapped role is paid its best role's - * tokens. Sequential: each grant is a handful of D1 writes and two hub calls, and a cron - * invocation has a fixed subrequest budget (50 free, 1000 paid) — a community whose - * supporters outgrow that is the point to batch it, not before. + * One run of the gift: every account whose Discord links hold a mapped role is paid its + * best role's tokens. Batched, as the staff drops are: `payTokenGifts` banks the boxes a + * few dozen accounts a round, and the frames go out in ONE hub call — grant by grant this + * was five D1 round trips and two hub calls an account, in series, against a cron + * invocation's fixed subrequest budget (50 free, 1,000 paid). * - * `startingTokens` is the signup grant to seed before the credit — `creditCurrency` upserts - * the balance row, so a never-touched balance would otherwise start from this gift instead - * of the grant plus this gift. + * A round that throws fails every account in it: they are logged, with their roles, and + * counted in `failed` rather than retried — a retry that ALSO failed partway is how a + * balance gets credited twice — and the later rounds still run. + * + * `startingTokens` is the signup grant to seed before the credit — the credit upserts the + * balance row, so a never-touched balance would otherwise start from this gift instead of + * the grant plus this gift. */ export async function grantDiscordRoleGifts( env: Env, @@ -215,19 +229,53 @@ export async function grantDiscordRoleGifts( const links = await getLinksForPlatform(env.DB, PlatformType.Discord) summary.links = links.length + // One box per ACCOUNT: an account with two Discord links is paid the best role across + // both, once. Kept in link order (oldest first) so the frames go out in a stable order. + const best = new Map() for (const link of links) { const role = bestRoleGift(link.roles, roles) if (role === null) continue + const held = best.get(link.accountId) + if (held === undefined || role.tokens > held.tokens) best.set(link.accountId, role) + } + + const frames: PlayerNotification[] = [] + const due = [...best] + for (let i = 0; i < due.length; i += TOKEN_GIFT_CHUNK) { + const round = due.slice(i, i + TOKEN_GIFT_CHUNK) try { - await grantOne(env, link.accountId, role, startingTokens) - summary.granted++ - summary.tokens += role.tokens + const { paid } = await payTokenGifts( + env.DB, + round.map(([accountId, role]) => ({ + accountId, + amount: role.tokens, + content: roleGiftContent(role.tokens), + })), + startingTokens + ) + for (const p of paid) { + summary.granted++ + summary.tokens += p.amount + frames.push(...giftFrames(p.accountId, p.balance, p.giftId, p.content)) + } } catch (err) { - // Logged with everything needed to check what did land rather than retried: a - // retry that ALSO failed partway is how a balance gets credited twice. - summary.failed++ + summary.failed += round.length console.error( - `discord role gift: account ${link.accountId}'s grant for role ${role.roleId} (${role.tokens} tokens) failed: ${err instanceof Error ? err.message : String(err)}` + `discord role gift: a round of ${round.length} grant(s) failed (${round + .map( + ([accountId, role]) => `account ${accountId} role ${role.roleId} ${role.tokens} tokens` + ) + .join('; ')}): ${err instanceof Error ? err.message : String(err)}` + ) + } + } + + if (frames.length > 0) { + try { + await env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).notifyPlayers(frames) + } catch (err) { + console.error( + `discord role gift: could not notify ${summary.granted} account(s) of their box: ${err instanceof Error ? err.message : String(err)}` ) } } @@ -283,36 +331,59 @@ async function grantOne( const content = roleGiftContent(role.tokens) const gift = await createGift(env.DB, accountId, content) - const hub = env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE) try { - // The balance first, so the box's announcement lands on a total that includes it. - // `Balance` is the RESULTING total, into the one -2 bucket: the frame SETS it. - await hub.notifyPlayer(accountId, NotificationType.StorefrontBalanceUpdate, { - Balance: balance, - CurrencyType: CurrencyType.RecCenterTokens, - Platform: ALL_PLATFORMS, - } satisfies BalanceResponsePayload) - await hub.notifyPlayer(accountId, NotificationType.GiftPackageReceivedImmediate, { - Id: gift.id, - FromPlayerId: COACH_ACCOUNT_ID, - ConsumableItemDesc: content.ConsumableItemDesc, - AvatarItemType: content.AvatarItemType, - AvatarItemDesc: content.AvatarItemDesc, - EquipmentPrefabName: content.EquipmentPrefabName, - EquipmentModificationGuid: content.EquipmentModificationGuid, - CurrencyType: content.CurrencyType, - Currency: content.Currency, - Xp: content.Xp, - GiftContext: content.GiftContext ?? 0, - GiftRarity: content.GiftRarity, - Message: content.Message, - Platform: -1, - PlatformsToSpawnOn: -1, - BalanceType: ALL_PLATFORMS, - } satisfies GiftPackagePayload) + await env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).notifyPlayers( + giftFrames(accountId, balance, gift.id, content) + ) } catch (err) { console.error( `discord role gift: could not notify account ${accountId} of box ${gift.id}: ${err instanceof Error ? err.message : String(err)}` ) } } + +/** + * The two frames announcing a stored box, in order: the balance first — the RESULTING + * total, into the one -2 bucket; the frame SETS it — so the box's announcement lands on a + * total that already includes it. + */ +function giftFrames( + accountId: number, + balance: number, + giftId: number, + content: GiftContent +): PlayerNotification[] { + return [ + { + playerId: accountId, + notificationType: NotificationType.StorefrontBalanceUpdate, + data: { + Balance: balance, + CurrencyType: CurrencyType.RecCenterTokens, + Platform: ALL_PLATFORMS, + } satisfies BalanceResponsePayload, + }, + { + playerId: accountId, + notificationType: NotificationType.GiftPackageReceivedImmediate, + data: { + Id: giftId, + FromPlayerId: COACH_ACCOUNT_ID, + ConsumableItemDesc: content.ConsumableItemDesc, + AvatarItemType: content.AvatarItemType, + AvatarItemDesc: content.AvatarItemDesc, + EquipmentPrefabName: content.EquipmentPrefabName, + EquipmentModificationGuid: content.EquipmentModificationGuid, + CurrencyType: content.CurrencyType, + Currency: content.Currency, + Xp: content.Xp, + GiftContext: content.GiftContext ?? 0, + GiftRarity: content.GiftRarity, + Message: content.Message, + Platform: -1, + PlatformsToSpawnOn: -1, + BalanceType: ALL_PLATFORMS, + } satisfies GiftPackagePayload, + }, + ] +} diff --git a/apps/econ/src/test/integration/api.test.ts b/apps/econ/src/test/integration/api.test.ts index c9164d1..03076aa 100644 --- a/apps/econ/src/test/integration/api.test.ts +++ b/apps/econ/src/test/integration/api.test.ts @@ -7670,6 +7670,49 @@ describe('discord role gift', () => { ]) }) + // One box per ACCOUNT, not per link: a player who linked two Discord identities is paid + // once, the best role across both. And more accounts than one round pays (30) are all + // paid once, each with their own box, with two frames apiece in link order. + test('pays an account with two links once, and more accounts than one round holds', async () => { + const many = Array.from({ length: 40 }, (_, i) => 9400 + i) + await onlyDiscordLinks([ + [9341, '900000000000000341', [ROLE_A]], + [9341, '900000000000000342', [ROLE_B]], + ...many.map((id): [number, string, string[]] => [id, `9000000000000${id}`, [ROLE_A]]), + ]) + await drainFrames() + + const summary = await grantDiscordRoleGifts(giftEnv(MAP), DEFAULT_STARTING_TOKENS) + expect(summary).toEqual({ + skipped: false, + roles: 2, + links: 42, + granted: 41, + tokens: 10000 + 40 * 2500, + failed: 0, + }) + expect(await tokens(9341)).toBe(DEFAULT_STARTING_TOKENS + 10000) + expect(await getPendingGifts(env.DB, 9341)).toMatchObject([{ Currency: 10000 }]) + + const frames = await drainFrames() + expect(frames.map((f) => f.accountId)).toEqual([9341, 9341, ...many.flatMap((id) => [id, id])]) + const boxIds = new Set() + for (const id of many) { + expect(await tokens(id)).toBe(DEFAULT_STARTING_TOKENS + 2500) + const boxes = await getPendingGifts(env.DB, id) + expect(boxes).toHaveLength(1) + boxIds.add(boxes[0]!.Id) + const own = frames.filter((f) => f.accountId === id) + expect(own[0]!.payload).toEqual({ + Balance: DEFAULT_STARTING_TOKENS + 2500, + CurrencyType: CurrencyType.RecCenterTokens, + Platform: -2, + }) + expect(own[1]!.payload).toMatchObject({ Id: boxes[0]!.Id, Currency: 2500 }) + } + expect(boxIds.size).toBe(many.length) + }) + test('every run pays again — the schedule is the cadence — and a lapsed role is not paid', async () => { await onlyDiscordLinks([ [9321, '900000000000000321', [ROLE_A]], diff --git a/apps/econ/vitest.config.ts b/apps/econ/vitest.config.ts index 0cd3b70..daca42d 100644 --- a/apps/econ/vitest.config.ts +++ b/apps/econ/vitest.config.ts @@ -35,6 +35,14 @@ export default defineConfig({ this.frames.push({ accountId, notificationType, payload }) return { delivered: 0, queued: true } } + // One recorded frame per entry, the shape notifyPlayer records, so a test + // asserts the same thing whichever call sent it. + async notifyPlayers(notifications) { + for (const { playerId, notificationType, data } of notifications) { + this.frames.push({ accountId: playerId, notificationType, payload: data }) + } + return { delivered: 0, queued: notifications.length } + } async broadcast() { return { delivered: 0 } } /** Everything pushed since the last call, then forget it. */ async drainFrames() { diff --git a/apps/notify/src/notifications-hub.ts b/apps/notify/src/notifications-hub.ts index 73abd4f..d36ac40 100644 --- a/apps/notify/src/notifications-hub.ts +++ b/apps/notify/src/notifications-hub.ts @@ -99,6 +99,13 @@ export interface HubState { pending: Array<{ playerId: number; count: number; latest: string }> } +/** One entry of a {@link NotificationsHub.notifyPlayers} call: a `notifyPlayer`'s arguments. */ +export interface PlayerNotification { + playerId: number + notificationType: string | number + data?: Record +} + /** * How many queued notifications one offline player may accumulate. Past this the OLDEST * are dropped, newest kept. @@ -464,6 +471,48 @@ export class NotificationsHub extends DurableObject { return { delivered, queued: false } } + /** + * {@link notifyPlayer} for MANY notifications in ONE RPC, each with its own player and + * payload, delivered (or queued, when that player is offline) in the order given. The + * server-wide token drop pays a few hundred players at once, and each needs two frames of + * their own — a balance that differs per player, then a box whose id does — so sent + * through `notifyPlayer` it was two round trips per player, in series, from a request + * already short of its subrequest budget. One call carries the lot. + * + * `notifyPlayersEphemeral` is the other bulk send: one identical payload to many, never + * queued. This one keeps `notifyPlayer`'s durability, player by player. The "nobody to + * deliver to" warning is logged ONCE for the whole call, naming the players, rather than + * once per frame: a drop reaches everyone with an unexpired presence row, and a good + * share of them are between sessions at any moment. + */ + async notifyPlayers( + notifications: readonly PlayerNotification[] + ): Promise<{ delivered: number; queued: number }> { + let delivered = 0 + let queued = 0 + const unreachable = new Set() + for (const { playerId, notificationType, data } of notifications) { + const payload = this.buildNotificationPayload(notificationType, data) + const sent = this.deliverToPlayer(playerId, payload) + if (sent === 0) { + this.queuePending(playerId, payload) + queued++ + unreachable.add(playerId) + } else { + delivered += sent + } + } + if (queued > 0) { + console.warn('hub: notifications queued, nobody to deliver to', { + playerIds: [...unreachable], + queued, + notifications: notifications.length, + liveSockets: this.ctx.getWebSockets().length, + }) + } + return { delivered, queued } + } + /** * Queue a notification for a player who wasn't reachable, keeping the queue to * {@link MAX_PENDING_PER_PLAYER}. Trimming is oldest-first and happens on the write, so diff --git a/apps/notify/src/test/integration/api.test.ts b/apps/notify/src/test/integration/api.test.ts index ef5e8c4..baa3820 100644 --- a/apps/notify/src/test/integration/api.test.ts +++ b/apps/notify/src/test/integration/api.test.ts @@ -311,6 +311,70 @@ describe('notification delivery', () => { ws.close() }) + // The bulk send is notifyPlayer many times over in one RPC: each entry its own player and + // payload, delivered live or queued per player, in the order given. What the token drop + // sends — two frames a player, a few hundred players. + test('delivers a batch to each player, live or queued, in order', async () => { + const live = 9005 + const offline = 9006 + const { ws, waitFor, records } = await connect('conn-bulk') + send(ws, { + type: 1, + invocationId: 's', + target: 'SubscribeToPlayers', + arguments: [{ playerIds: [live] }], + }) + await waitFor((r) => r.type === 3 && r.invocationId === 's') + + const stub = env.RECFLARE_NOTIFICATIONS_HUB.getByName('global') + const result = await runInDurableObject(stub, (instance) => + instance.notifyPlayers([ + { playerId: live, notificationType: 61, data: { Balance: 10 } }, + { playerId: offline, notificationType: 61, data: { Balance: 20 } }, + { playerId: live, notificationType: 31, data: { Id: 1 } }, + { playerId: offline, notificationType: 31, data: { Id: 2 } }, + ]) + ) + expect(result).toEqual({ delivered: 2, queued: 2 }) + + await waitFor( + (r) => + r.type === 1 && + r.target === 'Notification' && + (JSON.parse((r.arguments as string[])[0]) as { Id: string }).Id === '31' + ) + const notes = records + .filter((r) => r.type === 1 && r.target === 'Notification') + .map((r) => JSON.parse((r.arguments as string[])[0]) as unknown) + expect(notes).toEqual([ + { Id: '61', Msg: { Balance: 10 } }, + { Id: '31', Msg: { Id: 1 } }, + ]) + ws.close() + + // The offline player's two are queued, and flush in order when they subscribe. + const late = await connect('conn-bulk-late') + send(late.ws, { + type: 1, + target: 'SubscribeToPlayers', + arguments: [{ playerIds: [offline] }], + }) + await late.waitFor( + (r) => + r.type === 1 && + r.target === 'Notification' && + (JSON.parse((r.arguments as string[])[0]) as { Id: string }).Id === '31' + ) + const flushed = late.records + .filter((r) => r.type === 1 && r.target === 'Notification') + .map((r) => JSON.parse((r.arguments as string[])[0]) as unknown) + expect(flushed).toEqual([ + { Id: '61', Msg: { Balance: 20 } }, + { Id: '31', Msg: { Id: 2 } }, + ]) + late.ws.close() + }) + test('broadcast reaches connected clients', async () => { const { ws, waitFor } = await connect('conn-broadcast') const res = await post('/internal/broadcast', { diff --git a/apps/www/src/staff.ts b/apps/www/src/staff.ts index da388ff..d2f0b96 100644 --- a/apps/www/src/staff.ts +++ b/apps/www/src/staff.ts @@ -44,6 +44,7 @@ import { DEFAULT_STARTING_TOKENS, ensureStartingBalances, getBalance, + payTokenGifts, spendCurrency, } from '../../econ/src/balance-db' // The item catalog and the two inventories a skin or a consumable gift writes, all `econ`'s. @@ -71,6 +72,7 @@ import type { ModerationKickPayload, PlayerProgressionLevelPayload, } from '../../notify/src/notification-payloads' +import type { PlayerNotification } from '../../notify/src/notifications-hub' import type { App, Env } from './context' /** @@ -658,12 +660,11 @@ function staffGiftContent(fields: Partial): GiftContent { * drops unknown ones, so it costs nothing if unread. Best-effort: the box is already stored, * and an offline player meets it on their next read of their gifts. */ -async function announceGift( - c: Context, - playerId: number, +/** The `GiftPackageReceivedImmediate` frame announcing a stored box — see {@link announceGift}. */ +function giftFrame( giftId: number, content: GiftContent -): Promise { +): GiftPackagePayload & { CustomAvatarItemId?: string } { const frame: GiftPackagePayload = { Id: giftId, FromPlayerId: content.FromPlayerId ?? COACH_ACCOUNT_ID, @@ -682,13 +683,22 @@ async function announceGift( PlatformsToSpawnOn: -1, BalanceType: ALL_PLATFORMS, } + return content.CustomAvatarItemId + ? { ...frame, CustomAvatarItemId: content.CustomAvatarItemId } + : frame +} + +async function announceGift( + c: Context, + playerId: number, + giftId: number, + content: GiftContent +): Promise { try { await c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).notifyPlayer( playerId, NotificationType.GiftPackageReceivedImmediate, - content.CustomAvatarItemId - ? { ...frame, CustomAvatarItemId: content.CustomAvatarItemId } - : { ...frame } + { ...giftFrame(giftId, content) } ) } catch (err) { logger.error('failed to announce a staff gift', { @@ -852,6 +862,74 @@ async function sendTokens( return { balance, giftId: gift.id } } +/** + * {@link sendTokens} for a crowd — the room gift, the online drop and the role drop, which + * pay the same amount in the same box to everyone on a list. The one-player path is five + * D1 round trips and two hub calls; done in series for a full server that was a request of + * ~1,750 subrequests, a minute or more long, and past D1's 1,000-queries-an-invocation cap + * somewhere around 200 players — the drop then died partway, with the first players paid + * and the rest not, and no audit row. This banks the boxes through econ's batched + * `payTokenGifts` and announces the lot in ONE hub RPC. + * + * Per player it is exactly the one-player gift: the signup grant seeded first, the balance + * moved, a box stored, then a balance frame (the RESULTING total, into the -2 bucket) and + * the box announced. A negative amount is a debit guarded per row — `amount >= ?` in the + * UPDATE, so nobody is overdrawn — and a player who can't afford it is SKIPPED (no box, no + * frame) rather than failing the rest, as the one-player gift returns null. `paid` and + * `skipped` keep the order of `playerIds`. + * + * The frames are best-effort, as everywhere: the tokens and boxes are stored before the hub + * is called, and a player who was offline meets both on their next sign-in. + */ +async function sendTokensToMany( + c: Context, + playerIds: readonly number[], + amount: number, + startingTokens: number, + message: string = STAFF_GIFT_MESSAGE +): Promise<{ paid: number[]; skipped: number[] }> { + const content = staffGiftContent({ + CurrencyType: CurrencyType.RecCenterTokens, + Currency: amount, + Message: message, + }) + const { paid, skipped } = await payTokenGifts( + c.env.DB, + playerIds.map((accountId) => ({ accountId, amount, content })), + startingTokens + ) + + // The balance first, so the box's announcement lands on a total that already includes + // it — the same two frames, in the same order, as sendTokens. + const frames: PlayerNotification[] = paid.flatMap((p) => [ + { + playerId: p.accountId, + notificationType: NotificationType.StorefrontBalanceUpdate, + data: { + Balance: p.balance, + CurrencyType: CurrencyType.RecCenterTokens, + Platform: ALL_PLATFORMS, + } satisfies BalanceResponsePayload, + }, + { + playerId: p.accountId, + notificationType: NotificationType.GiftPackageReceivedImmediate, + data: { ...giftFrame(p.giftId, p.content) }, + }, + ]) + if (frames.length > 0) { + try { + await c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).notifyPlayers(frames) + } catch (err) { + logger.error('failed to announce a bulk staff token gift', { + paidCount: paid.length, + error: err instanceof Error ? err.message : String(err), + }) + } + } + return { paid: paid.map((p) => p.accountId), skipped } +} + /** * Send tokens to EVERYONE standing in a room right now — every instance of it at once. * @@ -861,10 +939,9 @@ async function sendTokens( * has lapsed is already gone from it. * * Each player is paid exactly as the one-player gift pays them — the same box, with the same - * optional `message` on it — one after another rather than at once: this is a handful of writes per player, and a busy room would otherwise open a - * hundred at a time. A player a negative amount can't be taken from is SKIPPED rather than - * failing the room — the ones who could afford it have already been debited by then — and the - * response says who was missed. + * optional `message` on it — in batches, through {@link sendTokensToMany}. A player a + * negative amount can't be taken from is SKIPPED rather than failing the room — the ones who + * could afford it are debited regardless — and the response says who was missed. */ export async function giftRoomTokensHandler(c: Context) { const roomId = Number(c.req.param('roomId')) @@ -885,13 +962,7 @@ export async function giftRoomTokensHandler(c: Context) { } const startingTokens = intVar(c.env.STARTING_TOKENS, DEFAULT_STARTING_TOKENS) - const paid: number[] = [] - const skipped: number[] = [] - for (const playerId of playerIds) { - const sent = await sendTokens(c, playerId, amount, startingTokens, message) - if (sent === null) skipped.push(playerId) - else paid.push(playerId) - } + const { paid, skipped } = await sendTokensToMany(c, playerIds, amount, startingTokens, message) await recordPlayerAudit(c, 'gift_tokens_room', { roomId, amount, message, paid, skipped }) logger.info('staff gifted tokens to a room', { @@ -911,8 +982,9 @@ export async function giftRoomTokensHandler(c: Context) { * The audience is every unexpired `presence` row, lobby included (see `getOnlinePlayerIds`), * read when the button is pressed: the same population the coach broadcast reaches and the * status page counts. Each player is paid exactly as the one-player and room gifts pay them - * (banked, then a balance frame and an announced box), one after another for the same reason - * the room gift is — a full server is a lot of writes, and they should not all open at once. + * (banked, then a balance frame and an announced box), in batches through + * {@link sendTokensToMany}: a full server is a lot of writes, and paid one player at a time + * the drop outran D1's per-invocation query cap and died partway. * * Narrower than the other gifts on purpose: the amount is POSITIVE and capped at * {@link MAX_TOKEN_DROP} per player, whatever `MAX_TOKEN_GIFT` says — a stray zero here is @@ -943,13 +1015,8 @@ export async function giftOnlineTokensHandler(c: Context) { if (playerIds.length === 0) return c.json({ error: 'Nobody is online right now' }, 404) const startingTokens = intVar(c.env.STARTING_TOKENS, DEFAULT_STARTING_TOKENS) - const paid: number[] = [] - for (const playerId of playerIds) { - // A positive amount never comes back null: nothing to skip. - if ((await sendTokens(c, playerId, amount, startingTokens, message)) !== null) { - paid.push(playerId) - } - } + // A positive amount skips nobody. + const { paid } = await sendTokensToMany(c, playerIds, amount, startingTokens, message) await recordPlayerAudit(c, 'gift_tokens_online', { amount, message, paid }) logger.info('staff dropped tokens on everyone online', { @@ -1015,13 +1082,8 @@ export async function giftRoleTokensHandler(c: Context) { } const startingTokens = intVar(c.env.STARTING_TOKENS, DEFAULT_STARTING_TOKENS) - const paid: number[] = [] - for (const playerId of playerIds) { - // A positive amount never comes back null: nothing to skip. - if ((await sendTokens(c, playerId, amount, startingTokens, message)) !== null) { - paid.push(playerId) - } - } + // A positive amount skips nobody. + const { paid } = await sendTokensToMany(c, playerIds, amount, startingTokens, message) await recordPlayerAudit(c, 'gift_tokens_role', { roleId, amount, message, paid }) logger.info('staff dropped tokens on a discord role', { diff --git a/apps/www/src/test/integration/api.test.ts b/apps/www/src/test/integration/api.test.ts index 2cca8b6..ef160f4 100644 --- a/apps/www/src/test/integration/api.test.ts +++ b/apps/www/src/test/integration/api.test.ts @@ -52,7 +52,9 @@ import { BALANCE_SCHEMA_DDL, CurrencyType, DEFAULT_STARTING_TOKENS, + ensureStartingBalances, getBalance, + spendCurrency, } from '../../../../econ/src/balance-db' import { CATALOG_SCHEMA_DDL } from '../../../../econ/src/catalog-db' import { CONSUMABLE_SCHEMA_DDL, getConsumables } from '../../../../econ/src/consumables-db' @@ -1770,6 +1772,7 @@ it('drops tokens on everyone online, lobby included, with the message on the box `UPDATE presence SET data = json_set(data, '$.expiresAt', 1) WHERE account_id = 8383` ).run() + await clearHub() const res = await devPost('/api/staff/online/gift-tokens', 8110, { amount: 100, message: 'Thanks for playing!', @@ -1781,6 +1784,7 @@ it('drops tokens on everyone online, lobby included, with the message on the box expect(body.paid).toEqual(expect.arrayContaining([8380, 8381, 8382])) expect(body.paid).not.toContain(8383) + const frames = await hubFrames() for (const playerId of [8380, 8381, 8382]) { await expect( getBalance(env.DB, playerId, CurrencyType.RecCenterTokens, DEFAULT_STARTING_TOKENS) @@ -1788,7 +1792,23 @@ it('drops tokens on everyone online, lobby included, with the message on the box const gifts = await getPendingGifts(env.DB, playerId) expect(gifts).toHaveLength(1) expect(gifts[0]).toMatchObject({ Currency: 100, Message: 'Thanks for playing!' }) + // The same two frames the one-player gift sends, in the same order: the resulting + // total into the -2 bucket, then the box it came in — every player's own. + const own = frames.filter((f) => f.playerId === playerId) + expect(own.map((f) => f.notificationType)).toEqual([61, 31]) + expect(own[0].data).toEqual({ + Balance: DEFAULT_STARTING_TOKENS + 100, + CurrencyType: CurrencyType.RecCenterTokens, + Platform: -2, + }) + expect(own[1].data).toMatchObject({ + Id: gifts[0]!.Id, + Currency: 100, + Message: 'Thanks for playing!', + BalanceType: -2, + }) } + expect(frames.filter((f) => f.playerId === 8383)).toEqual([]) expect(await getPendingGifts(env.DB, 8383)).toEqual([]) const { results } = await env.DB.prepare( @@ -1813,6 +1833,115 @@ it('refuses a token drop over the cap, without a message, or with nobody online' expect((await drop({ amount: 1_000, message: 'hi' })).status).toBe(404) }) +// The drop pays in rounds of 30 (three bound values a player, 100 a statement) and announces +// the lot in one hub call. More players than one round holds, with a balance row already +// present for some and none for the rest: everyone is paid once, in presence order, each +// box has its own id, and the hub gets exactly two frames a player — nobody doubled by a +// round boundary and nobody dropped past it. +it('drops tokens on more players than one round pays, each exactly once', async () => { + await env.DB.prepare('DELETE FROM presence').run() + const players = Array.from({ length: 70 }, (_, i) => 8600 + i) + for (const accountId of players) { + await setPresence(env.DB, { + accountId, + roomInstance: null, + statusVisibility: 0, + deviceClass: 0, + vrMovementMode: 0, + platform: 4, + appVersion: 'test', + }) + } + // Half already hold a balance row (one touched by a prior read); the rest get the signup + // grant seeded by the drop itself. Both must land on grant + drop. + for (const accountId of players.filter((id) => id % 2 === 0)) { + await getBalance(env.DB, accountId, CurrencyType.RecCenterTokens, DEFAULT_STARTING_TOKENS) + } + + await clearHub() + const res = await devPost('/api/staff/online/gift-tokens', 8110, { + amount: 7, + message: 'Round and round', + }) + expect(res.status).toBe(200) + const body = (await res.json()) as { paid: number[] } + expect(body.paid).toEqual(players) + + const { results } = await env.DB.prepare( + `SELECT account_id AS accountId, amount FROM balance + WHERE currency_type = ?1 AND account_id BETWEEN 8600 AND 8669 ORDER BY account_id` + ) + .bind(CurrencyType.RecCenterTokens) + .all<{ accountId: number; amount: number }>() + expect(results.map((r) => r.accountId)).toEqual(players) + expect(new Set(results.map((r) => r.amount))).toEqual(new Set([DEFAULT_STARTING_TOKENS + 7])) + + const frames = await hubFrames() + expect(frames).toHaveLength(players.length * 2) + const boxIds = new Set() + for (const playerId of players) { + const own = frames.filter((f) => f.playerId === playerId) + expect(own.map((f) => f.notificationType)).toEqual([61, 31]) + expect(own[0].data.Balance).toBe(DEFAULT_STARTING_TOKENS + 7) + const gifts = await getPendingGifts(env.DB, playerId) + expect(gifts).toHaveLength(1) + expect(own[1].data.Id).toBe(gifts[0]!.Id) + boxIds.add(gifts[0]!.Id) + } + expect(boxIds.size).toBe(players.length) +}) + +// A negative room gift is a debit guarded per player: whoever can't cover it is skipped — +// no debit, no box, no frame — and the ones who can are still charged. +it('skips the players a room debit would overdraw, and charges the rest', async () => { + const inRoom = async (accountId: number) => + setPresence(env.DB, { + accountId, + roomInstance: { roomId: 7730, roomInstanceId: 77301 }, + statusVisibility: 0, + deviceClass: 0, + vrMovementMode: 0, + platform: 4, + appVersion: 'test', + }) + await inRoom(8480) + await inRoom(8481) + await inRoom(8482) + // 8481 has already spent most of their grant. + await ensureStartingBalances(env.DB, 8481, DEFAULT_STARTING_TOKENS) + expect( + await spendCurrency( + env.DB, + 8481, + CurrencyType.RecCenterTokens, + DEFAULT_STARTING_TOKENS - 10, + DEFAULT_STARTING_TOKENS + ) + ).toBe(true) + + await clearHub() + const res = await devPost('/api/staff/rooms/7730/gift-tokens', 8110, { amount: -100 }) + expect(res.status).toBe(200) + expect(await res.json()).toEqual({ + roomId: 7730, + amount: -100, + paid: [8480, 8482], + skipped: [8481], + }) + for (const playerId of [8480, 8482]) { + await expect( + getBalance(env.DB, playerId, CurrencyType.RecCenterTokens, DEFAULT_STARTING_TOKENS) + ).resolves.toBe(DEFAULT_STARTING_TOKENS - 100) + expect(await getPendingGifts(env.DB, playerId)).toHaveLength(1) + } + await expect( + getBalance(env.DB, 8481, CurrencyType.RecCenterTokens, DEFAULT_STARTING_TOKENS) + ).resolves.toBe(10) + expect(await getPendingGifts(env.DB, 8481)).toEqual([]) + const frames = await hubFrames() + expect(frames.map((f) => f.playerId)).toEqual([8480, 8480, 8482, 8482]) +}) + // The role drop's audience is the Discord links that record the role — offline included, // since that is the point — and nobody else: a link holding other roles, a link holding none, // a player with no link at all. The same Discord identity on two accounts pays both. diff --git a/apps/www/vitest.config.ts b/apps/www/vitest.config.ts index 9e883b0..91288af 100644 --- a/apps/www/vitest.config.ts +++ b/apps/www/vitest.config.ts @@ -61,6 +61,12 @@ export default defineConfig({ this.sent.push({ playerIds, ephemeral: true, notificationType, data }) return { delivered: 0 } } + // Recorded one entry per frame, as notifyPlayer records one, so a test + // asserts the same shape whichever path a gift took. + async notifyPlayers(notifications) { + for (const n of notifications) this.sent.push(n) + return { delivered: 0, queued: notifications.length } + } async broadcast() { return { delivered: 0 } } async fetch(request) { if (request.method === 'DELETE') {