From 240b7e4388b1d4495b3c2bdd5a5924861ec48485 Mon Sep 17 00:00:00 2001 From: Hampus Date: Mon, 31 Aug 2026 19:06:31 +0200 Subject: [PATCH] feat(expressions): add an expression pack residue cleanup tool (#2265) --- fluxer_api/package.json | 3 +- fluxer_api/scripts/build.mjs | 6 +- .../src/ExpressionPackResidueEntrypoint.ts | 110 +++++ .../ExpressionPackResidueCleanup.test.ts | 451 ++++++++++++++++++ .../ExpressionPackResidueCleanup.ts | 206 ++++++++ 5 files changed, 774 insertions(+), 2 deletions(-) create mode 100644 fluxer_api/src/ExpressionPackResidueEntrypoint.ts create mode 100644 fluxer_api/src/api/maintenance/ExpressionPackResidueCleanup.test.ts create mode 100644 fluxer_api/src/api/maintenance/ExpressionPackResidueCleanup.ts diff --git a/fluxer_api/package.json b/fluxer_api/package.json index ade0d74e1..86821f455 100644 --- a/fluxer_api/package.json +++ b/fluxer_api/package.json @@ -8,7 +8,8 @@ "typecheck": "tsgo --noEmit", "dev": "tsx watch --clear-screen=false src/AppEntrypoint.ts", "start": "tsx src/AppEntrypoint.ts", - "start:worker": "tsx src/WorkerEntrypoint.ts" + "start:worker": "tsx src/WorkerEntrypoint.ts", + "cleanup:expression-pack-residue": "tsx src/ExpressionPackResidueEntrypoint.ts" }, "dependencies": { "@atproto/api": "catalog:", diff --git a/fluxer_api/scripts/build.mjs b/fluxer_api/scripts/build.mjs index ad5af6bd8..5247192e8 100644 --- a/fluxer_api/scripts/build.mjs +++ b/fluxer_api/scripts/build.mjs @@ -64,7 +64,11 @@ const workspacePlugin = { rmSync(OUT_DIR, {recursive: true, force: true}); await build({ - entryPoints: [join(API_ROOT, 'src/AppEntrypoint.ts'), join(API_ROOT, 'src/WorkerEntrypoint.ts')], + entryPoints: [ + join(API_ROOT, 'src/AppEntrypoint.ts'), + join(API_ROOT, 'src/WorkerEntrypoint.ts'), + join(API_ROOT, 'src/ExpressionPackResidueEntrypoint.ts'), + ], outdir: OUT_DIR, bundle: true, platform: 'node', diff --git a/fluxer_api/src/ExpressionPackResidueEntrypoint.ts b/fluxer_api/src/ExpressionPackResidueEntrypoint.ts new file mode 100644 index 000000000..4cbf2d751 --- /dev/null +++ b/fluxer_api/src/ExpressionPackResidueEntrypoint.ts @@ -0,0 +1,110 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import '@app/Instrument'; +import {initializeConfig} from '@app/api/Config'; +import {setDatabaseQueryExecutor} from '@app/api/database/CassandraQueryExecution'; +import {ensurePostgresKvSchema, PostgresKvQueryExecutor} from '@app/api/database/PostgresKvQueryExecutor'; +import {GuildRepository} from '@app/api/guild/repositories/GuildRepository'; +import {ExpressionAssetPurger} from '@app/api/guild/services/content/ExpressionAssetPurger'; +import {AssetDeletionQueue} from '@app/api/infrastructure/AssetDeletionQueue'; +import {initializeLogger} from '@app/api/Logger'; +import { + ExpressionPackResidueCleanup, + formatExpressionPackResidueReport, +} from '@app/api/maintenance/ExpressionPackResidueCleanup'; +import {getKVClient} from '@app/api/middleware/ServiceRegistry'; +import {Config} from '@app/Config'; +import {Logger} from '@app/Logger'; +import {BACKGROUND_READ_TIMEOUT_MS, initCassandra, shutdownCassandra} from '@pkgs/cassandra/src/Client'; +import {getDefaultPostgresClient, initPostgres, shutdownPostgres} from '@pkgs/postgres/src/Client'; + +const USAGE = `Usage: fluxer-api expression-pack-residue [--apply] + +Deletes the guild_emojis and guild_stickers rows that the removed expression +packs feature wrote with guild_id set to a pack id, and queues their CDN and S3 +assets for purge through the asset deletion queue. + +Rows are only ever read from the guild_emojis and guild_stickers partitions +named by a pack id in expression_packs, so a real guild's expressions are never +read and never deleted. A pack id that also names a live guild is reported and +skipped. Running it again after a successful run is a no-op. + +Options: + --apply Delete the rows and queue their assets. Without it the tool reports + what it would delete and changes nothing. + --help Show this message.`; + +interface Options { + apply: boolean; + help: boolean; +} + +function parseArgs(): Options { + const options: Options = {apply: false, help: false}; + for (const arg of process.argv.slice(2)) { + if (arg === '--apply') { + options.apply = true; + } else if (arg === '--help' || arg === '-h') { + options.help = true; + } else { + throw new Error(`Unknown argument "${arg}". Run with --help for usage.`); + } + } + return options; +} + +async function initializeDatabase(): Promise<() => Promise> { + if (Config.database.backend === 'postgres') { + await initPostgres(Config.postgres); + const postgres = getDefaultPostgresClient(); + await ensurePostgresKvSchema(postgres); + setDatabaseQueryExecutor(new PostgresKvQueryExecutor(postgres)); + return async () => { + setDatabaseQueryExecutor(null); + await shutdownPostgres(); + }; + } + await initCassandra({ + hosts: Config.cassandra.hosts.split(',').filter(Boolean), + port: Config.cassandra.port, + keyspace: Config.cassandra.keyspace, + localDc: Config.cassandra.localDc, + username: Config.cassandra.username || undefined, + password: Config.cassandra.password || undefined, + readTimeoutMs: BACKGROUND_READ_TIMEOUT_MS, + }); + return async () => { + await shutdownCassandra(); + }; +} + +async function main(): Promise { + const options = parseArgs(); + if (options.help) { + console.log(USAGE); + return; + } + initializeConfig(Config); + initializeLogger(Logger); + const shutdownDatabase = await initializeDatabase(); + const kvClient = getKVClient(); + try { + const cleanup = new ExpressionPackResidueCleanup( + new GuildRepository(), + new ExpressionAssetPurger(new AssetDeletionQueue(kvClient)), + ); + const report = await cleanup.run({apply: options.apply}); + console.log(formatExpressionPackResidueReport(report)); + } finally { + await shutdownDatabase(); + } +} + +main() + .then(() => { + process.exit(0); + }) + .catch((error) => { + Logger.fatal({error}, 'Expression pack residue cleanup failed'); + process.exit(1); + }); diff --git a/fluxer_api/src/api/maintenance/ExpressionPackResidueCleanup.test.ts b/fluxer_api/src/api/maintenance/ExpressionPackResidueCleanup.test.ts new file mode 100644 index 000000000..7ae933c0b --- /dev/null +++ b/fluxer_api/src/api/maintenance/ExpressionPackResidueCleanup.test.ts @@ -0,0 +1,451 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {spawnSync} from 'node:child_process'; +import {createServer} from 'node:net'; +import { + getDefaultPostgresClient, + type IPostgresClient, + initPostgres, + shutdownPostgres, +} from '@pkgs/postgres/src/Client'; +import {afterAll, beforeAll, beforeEach, describe, expect, it, vi} from 'vitest'; +import type {EmojiID, GuildID, StickerID, UserID} from '../BrandedTypes'; +import {createEmojiID, createGuildID, createStickerID, createUserID} from '../BrandedTypes'; +import {setCassandraQueryExecutorForTesting, upsertOne} from '../database/CassandraQueryExecution'; +import {defineTable} from '../database/CassandraTableDsl'; +import {ensurePostgresKvSchema, PostgresKvQueryExecutor} from '../database/PostgresKvQueryExecutor'; +import {GuildRepository} from '../guild/repositories/GuildRepository'; +import {ExpressionAssetPurger} from '../guild/services/content/ExpressionAssetPurger'; +import {AssetDeletionQueue} from '../infrastructure/AssetDeletionQueue'; +import {getKVClient} from '../middleware/ServiceRegistry'; +import {GuildEmoji} from '../models/GuildEmoji'; +import {GuildSticker} from '../models/GuildSticker'; +import {startDockerContainer} from '../test/DockerTestContainer'; +import {InMemoryCassandraQueryExecutor} from '../test/InMemoryCassandraQueryExecutor'; +import { + ExpressionPackResidueCleanup, + type ExpressionPackResidueRepository, + formatExpressionPackResidueReport, +} from './ExpressionPackResidueCleanup'; + +interface ExpressionPackSeedRow { + pack_id: GuildID; + pack_type: string; + creator_id: UserID; + name: string; + description: string | null; + created_at: Date; + updated_at: Date; + version: number; +} + +const ExpressionPacksSeed = defineTable({ + name: 'expression_packs', + columns: ['pack_id', 'pack_type', 'creator_id', 'name', 'description', 'created_at', 'updated_at', 'version'], + primaryKey: ['pack_id'], +}); + +const KV_TABLE = 'kv_pack_residue'; +const CONTAINER = `fluxer-packresidue-${process.pid.toString(36)}-${Date.now().toString(36)}`; +const dockerAvailable = spawnSync('docker', ['version'], {stdio: 'ignore'}).status === 0; + +const CREATOR_ID = createUserID(700000000000000001n); +const PACK_ID = createGuildID(900000000000000001n); +const SECOND_PACK_ID = createGuildID(900000000000000002n); +const GUILD_ID = createGuildID(100000000000000001n); +const PACK_ID_THAT_IS_A_GUILD = createGuildID(100000000000000002n); +const PACK_EMOJI_ID = createEmojiID(910000000000000001n); +const SECOND_PACK_EMOJI_ID = createEmojiID(910000000000000002n); +const PACK_STICKER_ID = createStickerID(920000000000000001n); +const GUILD_EMOJI_ID = createEmojiID(110000000000000001n); +const GUILD_STICKER_ID = createStickerID(120000000000000001n); + +async function sleep(ms: number): Promise { + await new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function freePort(): Promise { + return new Promise((resolve, reject) => { + const server = createServer(); + server.on('error', reject); + server.listen(0, '127.0.0.1', () => { + const address = server.address(); + if (typeof address === 'string' || address === null) { + reject(new Error('no port')); + return; + } + const port = address.port; + server.close(() => resolve(port)); + }); + }); +} + +function assetQueue(): AssetDeletionQueue { + return new AssetDeletionQueue(getKVClient()); +} + +function createCleanup(repository?: ExpressionPackResidueRepository): ExpressionPackResidueCleanup { + return new ExpressionPackResidueCleanup(repository ?? new GuildRepository(), new ExpressionAssetPurger(assetQueue())); +} + +async function seedPack(packId: GuildID): Promise { + await upsertOne( + ExpressionPacksSeed.insert({ + pack_id: packId, + pack_type: 'emoji', + creator_id: CREATOR_ID, + name: `pack ${packId}`, + description: null, + created_at: new Date('2026-01-01T00:00:00.000Z'), + updated_at: new Date('2026-01-01T00:00:00.000Z'), + version: 1, + }), + ); +} + +async function seedGuild(guildId: GuildID): Promise { + await new GuildRepository().upsertPartial(guildId, {name: `guild ${guildId}`, owner_id: CREATOR_ID}); +} + +async function seedEmoji(guildId: GuildID, emojiId: EmojiID, name: string): Promise { + await new GuildRepository().upsertEmoji({ + guild_id: guildId, + emoji_id: emojiId, + name, + creator_id: CREATOR_ID, + animated: false, + nsfw: null, + version: 1, + }); +} + +async function seedSticker(guildId: GuildID, stickerId: StickerID, name: string): Promise { + await new GuildRepository().upsertSticker({ + guild_id: guildId, + sticker_id: stickerId, + name, + description: null, + animated: false, + nsfw: null, + tags: null, + creator_id: CREATOR_ID, + version: 1, + }); +} + +async function emojiNames(guildId: GuildID): Promise> { + const emojis = await new GuildRepository().listEmojis(guildId); + return emojis.map((emoji) => emoji.name).sort(); +} + +async function stickerNames(guildId: GuildID): Promise> { + const stickers = await new GuildRepository().listStickers(guildId); + return stickers.map((sticker) => sticker.name).sort(); +} + +async function queuedS3Keys(): Promise> { + const queue = assetQueue(); + const items = await queue.getBatch(100); + return items.map((item) => item.s3Key).filter((key) => key.length > 0); +} + +function defineResidueSuite(resetData: () => Promise): void { + beforeEach(async () => { + await resetData(); + await assetQueue().clear(); + }); + + it('deletes every emoji and sticker owned by a pack id and queues their assets', async () => { + await seedPack(PACK_ID); + await seedEmoji(PACK_ID, PACK_EMOJI_ID, 'packblob'); + await seedEmoji(PACK_ID, SECOND_PACK_EMOJI_ID, 'packwave'); + await seedSticker(PACK_ID, PACK_STICKER_ID, 'packsticker'); + + const report = await createCleanup().run({apply: true}); + + expect(report.emojiCount).toBe(2); + expect(report.stickerCount).toBe(1); + expect(await emojiNames(PACK_ID)).toEqual([]); + expect(await stickerNames(PACK_ID)).toEqual([]); + expect(await new GuildRepository().getEmojiById(PACK_EMOJI_ID)).toBeNull(); + expect(await new GuildRepository().getStickerById(PACK_STICKER_ID)).toBeNull(); + expect(await queuedS3Keys()).toEqual([ + `emojis/${PACK_EMOJI_ID}`, + `emojis/${SECOND_PACK_EMOJI_ID}`, + `stickers/${PACK_STICKER_ID}`, + ]); + }); + + it('leaves a real guild emoji and sticker completely untouched', async () => { + await seedPack(PACK_ID); + await seedGuild(GUILD_ID); + await seedEmoji(GUILD_ID, GUILD_EMOJI_ID, 'guildblob'); + await seedSticker(GUILD_ID, GUILD_STICKER_ID, 'guildsticker'); + + const report = await createCleanup().run({apply: true}); + + expect(report.residues).toEqual([]); + expect(report.emojiCount).toBe(0); + expect(report.stickerCount).toBe(0); + expect(await emojiNames(GUILD_ID)).toEqual(['guildblob']); + expect(await stickerNames(GUILD_ID)).toEqual(['guildsticker']); + expect(await new GuildRepository().getEmojiById(GUILD_EMOJI_ID)).not.toBeNull(); + expect(await new GuildRepository().getStickerById(GUILD_STICKER_ID)).not.toBeNull(); + expect(await queuedS3Keys()).toEqual([]); + }); + + it('deletes only the pack rows out of a mixed dataset', async () => { + await seedPack(PACK_ID); + await seedPack(SECOND_PACK_ID); + await seedGuild(GUILD_ID); + await seedEmoji(PACK_ID, PACK_EMOJI_ID, 'packblob'); + await seedSticker(PACK_ID, PACK_STICKER_ID, 'packsticker'); + await seedEmoji(GUILD_ID, GUILD_EMOJI_ID, 'guildblob'); + await seedSticker(GUILD_ID, GUILD_STICKER_ID, 'guildsticker'); + + const report = await createCleanup().run({apply: true}); + + expect(report.packIds).toEqual([PACK_ID, SECOND_PACK_ID]); + expect(report.residues.map((residue) => residue.packId)).toEqual([PACK_ID]); + expect(await emojiNames(PACK_ID)).toEqual([]); + expect(await stickerNames(PACK_ID)).toEqual([]); + expect(await emojiNames(GUILD_ID)).toEqual(['guildblob']); + expect(await stickerNames(GUILD_ID)).toEqual(['guildsticker']); + expect(await queuedS3Keys()).toEqual([`emojis/${PACK_EMOJI_ID}`, `stickers/${PACK_STICKER_ID}`]); + }); + + it('is a clean no-op on an empty dataset', async () => { + const report = await createCleanup().run({apply: true}); + + expect(report).toEqual({ + applied: true, + packIds: [], + liveGuildPackIds: [], + residues: [], + emojiCount: 0, + stickerCount: 0, + }); + expect(await queuedS3Keys()).toEqual([]); + }); + + it('reports the rows it would delete without deleting anything by default', async () => { + await seedPack(PACK_ID); + await seedEmoji(PACK_ID, PACK_EMOJI_ID, 'packblob'); + await seedSticker(PACK_ID, PACK_STICKER_ID, 'packsticker'); + + const report = await createCleanup().run({apply: false}); + + expect(formatExpressionPackResidueReport(report)).toBe( + [ + 'expression pack residue cleanup: DRY RUN, nothing deleted and nothing queued', + 'pack ids read from expression_packs: 1', + `pack ${PACK_ID}: 1 emojis, 1 stickers`, + ` guild_emojis guild_id=${PACK_ID} emoji_id=${PACK_EMOJI_ID} name="packblob"`, + ` guild_stickers guild_id=${PACK_ID} sticker_id=${PACK_STICKER_ID} name="packsticker"`, + 'totals: 1 emojis, 1 stickers, 2 assets', + ].join('\n'), + ); + expect(await emojiNames(PACK_ID)).toEqual(['packblob']); + expect(await stickerNames(PACK_ID)).toEqual(['packsticker']); + expect(await queuedS3Keys()).toEqual([]); + }); + + it('is a no-op when run again after a real run', async () => { + await seedPack(PACK_ID); + await seedEmoji(PACK_ID, PACK_EMOJI_ID, 'packblob'); + await seedSticker(PACK_ID, PACK_STICKER_ID, 'packsticker'); + + await createCleanup().run({apply: true}); + await assetQueue().clear(); + const second = await createCleanup().run({apply: true}); + + expect(second.packIds).toEqual([PACK_ID]); + expect(second.residues).toEqual([]); + expect(second.emojiCount).toBe(0); + expect(second.stickerCount).toBe(0); + expect(await queuedS3Keys()).toEqual([]); + }); + + it('skips a pack id that also names a live guild and leaves its expressions alone', async () => { + await seedPack(PACK_ID_THAT_IS_A_GUILD); + await seedGuild(PACK_ID_THAT_IS_A_GUILD); + await seedEmoji(PACK_ID_THAT_IS_A_GUILD, GUILD_EMOJI_ID, 'guildblob'); + await seedSticker(PACK_ID_THAT_IS_A_GUILD, GUILD_STICKER_ID, 'guildsticker'); + + const report = await createCleanup().run({apply: true}); + + expect(report.liveGuildPackIds).toEqual([PACK_ID_THAT_IS_A_GUILD]); + expect(report.residues).toEqual([]); + expect(await emojiNames(PACK_ID_THAT_IS_A_GUILD)).toEqual(['guildblob']); + expect(await stickerNames(PACK_ID_THAT_IS_A_GUILD)).toEqual(['guildsticker']); + expect(await queuedS3Keys()).toEqual([]); + expect(formatExpressionPackResidueReport(report)).toContain( + `skipped pack ${PACK_ID_THAT_IS_A_GUILD}: that id is a live guild, its expressions were never read`, + ); + }); + + it('refuses to delete when a pack partition hands back a row owned by another guild', async () => { + await seedPack(PACK_ID); + const deleteEmoji = vi.fn(async () => {}); + const deleteSticker = vi.fn(async () => {}); + const repository: ExpressionPackResidueRepository = { + listGuilds: async () => [], + listEmojis: async () => [ + new GuildEmoji({ + guild_id: GUILD_ID, + emoji_id: GUILD_EMOJI_ID, + name: 'guildblob', + creator_id: CREATOR_ID, + animated: false, + nsfw: null, + version: 1, + }), + ], + listStickers: async () => [], + deleteEmoji, + deleteSticker, + }; + + await expect(createCleanup(repository).run({apply: true})).rejects.toThrow( + `guild_emojis partition ${PACK_ID} returned a row owned by guild ${GUILD_ID}`, + ); + expect(deleteEmoji).not.toHaveBeenCalled(); + expect(deleteSticker).not.toHaveBeenCalled(); + expect(await queuedS3Keys()).toEqual([]); + }); + + it('refuses to delete when a pack partition hands back a sticker owned by another guild', async () => { + await seedPack(PACK_ID); + const deleteSticker = vi.fn(async () => {}); + const repository: ExpressionPackResidueRepository = { + listGuilds: async () => [], + listEmojis: async () => [], + listStickers: async () => [ + new GuildSticker({ + guild_id: GUILD_ID, + sticker_id: GUILD_STICKER_ID, + name: 'guildsticker', + description: null, + animated: false, + nsfw: null, + tags: null, + creator_id: CREATOR_ID, + version: 1, + }), + ], + deleteEmoji: async () => {}, + deleteSticker, + }; + + await expect(createCleanup(repository).run({apply: true})).rejects.toThrow( + `guild_stickers partition ${PACK_ID} returned a row owned by guild ${GUILD_ID}`, + ); + expect(deleteSticker).not.toHaveBeenCalled(); + }); +} + +describe('ExpressionPackResidueCleanup against the cassandra query shapes', () => { + const executor = new InMemoryCassandraQueryExecutor(); + + beforeAll(() => { + setCassandraQueryExecutorForTesting(executor); + }); + + afterAll(() => { + setCassandraQueryExecutorForTesting(new InMemoryCassandraQueryExecutor()); + }); + + defineResidueSuite(async () => { + executor.reset(); + }); +}); + +describe.skipIf(!dockerAvailable)('ExpressionPackResidueCleanup against the postgres kv backend', () => { + let raw: IPostgresClient; + + beforeAll(async () => { + const port = await freePort(); + startDockerContainer([ + 'run', + '-d', + '--name', + CONTAINER, + '-e', + 'POSTGRES_USER=fluxer', + '-e', + 'POSTGRES_PASSWORD=fluxer', + '-e', + 'POSTGRES_DB=fluxer', + '-p', + `127.0.0.1:${port}:5432`, + 'postgres:16-alpine', + '-c', + 'fsync=off', + ]); + let ready = false; + for (let attempt = 0; attempt < 180 && !ready; attempt += 1) { + await sleep(500); + const probe = spawnSync('docker', ['exec', CONTAINER, 'pg_isready', '-U', 'fluxer', '-d', 'fluxer'], { + stdio: 'ignore', + }); + if (probe.status !== 0) continue; + try { + await initPostgres({ + url: `postgres://fluxer:fluxer@127.0.0.1:${port}/fluxer`, + maxConnections: 1, + kvTable: KV_TABLE, + }); + await getDefaultPostgresClient().query('SELECT 1'); + ready = true; + } catch { + await shutdownPostgres().catch(() => {}); + } + } + if (!ready) throw new Error('postgres never came up'); + raw = getDefaultPostgresClient(); + await ensurePostgresKvSchema(raw); + setCassandraQueryExecutorForTesting(new PostgresKvQueryExecutor(raw)); + }, 900_000); + + afterAll(async () => { + setCassandraQueryExecutorForTesting(new InMemoryCassandraQueryExecutor()); + await shutdownPostgres().catch(() => {}); + spawnSync('docker', ['rm', '-f', CONTAINER], {stdio: 'ignore'}); + }); + + defineResidueSuite(async () => { + await raw.query(`DELETE FROM ${KV_TABLE}`); + }); +}); + +describe('ExpressionPackResidueCleanup on a deployment without the pack tables', () => { + afterAll(() => { + setCassandraQueryExecutorForTesting(new InMemoryCassandraQueryExecutor()); + }); + + it('treats a missing expression_packs table as nothing to clean', async () => { + setCassandraQueryExecutorForTesting({ + executeQuery: async () => { + throw new Error('unconfigured table expression_packs'); + }, + executeBatch: async () => {}, + }); + + const report = await createCleanup().run({apply: true}); + + expect(report.packIds).toEqual([]); + expect(report.residues).toEqual([]); + expect(await queuedS3Keys()).toEqual([]); + }); + + it('rethrows any other failure while reading expression_packs', async () => { + setCassandraQueryExecutorForTesting({ + executeQuery: async () => { + throw new Error('Operation timed out'); + }, + executeBatch: async () => {}, + }); + + await expect(createCleanup().run({apply: true})).rejects.toThrow('Operation timed out'); + }); +}); diff --git a/fluxer_api/src/api/maintenance/ExpressionPackResidueCleanup.ts b/fluxer_api/src/api/maintenance/ExpressionPackResidueCleanup.ts new file mode 100644 index 000000000..6209472bc --- /dev/null +++ b/fluxer_api/src/api/maintenance/ExpressionPackResidueCleanup.ts @@ -0,0 +1,206 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import type {EmojiID, GuildID, StickerID} from '../BrandedTypes'; +import {fetchPage, type PagedQueryResult} from '../database/CassandraQueryExecution'; +import {defineTable} from '../database/CassandraTableDsl'; +import type {IGuildRepositoryAggregate} from '../guild/repositories/IGuildRepositoryAggregate'; +import type {ExpressionAssetPurger} from '../guild/services/content/ExpressionAssetPurger'; + +const PACK_ID_PAGE_SIZE = 500; +const LIVE_GUILD_CHUNK_SIZE = 100; +const EXPRESSION_PACKS_TABLE = 'expression_packs'; +const MISSING_TABLE_PATTERN = /unconfigured|undefined table|does not exist/i; + +interface ExpressionPackIdRow { + pack_id: GuildID; +} + +const ExpressionPacks = defineTable({ + name: EXPRESSION_PACKS_TABLE, + columns: ['pack_id'], + primaryKey: ['pack_id'], +}); + +const SCAN_EXPRESSION_PACK_IDS_QUERY = ExpressionPacks.selectCql(); + +export type ExpressionPackResidueRepository = Pick< + IGuildRepositoryAggregate, + 'listGuilds' | 'listEmojis' | 'listStickers' | 'deleteEmoji' | 'deleteSticker' +>; + +interface ExpressionPackResidueEmoji { + id: EmojiID; + name: string; +} + +interface ExpressionPackResidueSticker { + id: StickerID; + name: string; +} + +interface ExpressionPackResidue { + packId: GuildID; + emojis: Array; + stickers: Array; +} + +interface ExpressionPackResidueReport { + applied: boolean; + packIds: Array; + liveGuildPackIds: Array; + residues: Array; + emojiCount: number; + stickerCount: number; +} + +function compareIds(left: bigint, right: bigint): number { + if (left === right) return 0; + return left < right ? -1 : 1; +} + +function isMissingPackTableError(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error); + return message.includes(EXPRESSION_PACKS_TABLE) && MISSING_TABLE_PATTERN.test(message); +} + +export class ExpressionPackResidueCleanup { + constructor( + private readonly guildRepository: ExpressionPackResidueRepository, + private readonly assetPurger: ExpressionAssetPurger, + ) {} + + async run(options: {apply: boolean}): Promise { + const packIds = await this.scanPackIds(); + const report: ExpressionPackResidueReport = { + applied: options.apply, + packIds, + liveGuildPackIds: [], + residues: [], + emojiCount: 0, + stickerCount: 0, + }; + if (packIds.length === 0) { + return report; + } + report.liveGuildPackIds = await this.findLiveGuildIds(packIds); + const liveGuildIds = new Set(report.liveGuildPackIds); + for (const packId of packIds) { + if (liveGuildIds.has(packId)) { + continue; + } + const residue = await this.collectResidue(packId); + if (residue.emojis.length === 0 && residue.stickers.length === 0) { + continue; + } + report.residues.push(residue); + report.emojiCount += residue.emojis.length; + report.stickerCount += residue.stickers.length; + if (options.apply) { + await this.deleteResidue(residue); + } + } + return report; + } + + private async scanPackIds(): Promise> { + const packIds = new Set(); + let pageState: string | null = null; + try { + do { + const page: PagedQueryResult = await fetchPage( + SCAN_EXPRESSION_PACK_IDS_QUERY, + {}, + {pageSize: PACK_ID_PAGE_SIZE, pageState}, + ); + for (const row of page.rows) { + packIds.add(row.pack_id); + } + pageState = page.pageState; + } while (pageState !== null); + } catch (error) { + if (!isMissingPackTableError(error)) { + throw error; + } + return []; + } + return [...packIds].sort(compareIds); + } + + private async findLiveGuildIds(packIds: ReadonlyArray): Promise> { + const liveGuildIds: Array = []; + for (let index = 0; index < packIds.length; index += LIVE_GUILD_CHUNK_SIZE) { + const guilds = await this.guildRepository.listGuilds(packIds.slice(index, index + LIVE_GUILD_CHUNK_SIZE)); + for (const guild of guilds) { + liveGuildIds.push(guild.id); + } + } + return liveGuildIds.sort(compareIds); + } + + private async collectResidue(packId: GuildID): Promise { + const [emojis, stickers] = await Promise.all([ + this.guildRepository.listEmojis(packId), + this.guildRepository.listStickers(packId), + ]); + for (const emoji of emojis) { + if (emoji.guildId !== packId) { + throw new Error(`guild_emojis partition ${packId} returned a row owned by guild ${emoji.guildId}`); + } + } + for (const sticker of stickers) { + if (sticker.guildId !== packId) { + throw new Error(`guild_stickers partition ${packId} returned a row owned by guild ${sticker.guildId}`); + } + } + return { + packId, + emojis: emojis + .map((emoji) => ({id: emoji.id, name: emoji.name})) + .sort((left, right) => compareIds(left.id, right.id)), + stickers: stickers + .map((sticker) => ({id: sticker.id, name: sticker.name})) + .sort((left, right) => compareIds(left.id, right.id)), + }; + } + + private async deleteResidue(residue: ExpressionPackResidue): Promise { + for (const emoji of residue.emojis) { + await this.guildRepository.deleteEmoji(residue.packId, emoji.id); + await this.assetPurger.purgeEmoji(emoji.id.toString()); + } + for (const sticker of residue.stickers) { + await this.guildRepository.deleteSticker(residue.packId, sticker.id); + await this.assetPurger.purgeSticker(sticker.id.toString()); + } + } +} + +export function formatExpressionPackResidueReport(report: ExpressionPackResidueReport): string { + const lines: Array = [ + report.applied + ? 'expression pack residue cleanup: APPLIED, rows deleted and assets queued for purge' + : 'expression pack residue cleanup: DRY RUN, nothing deleted and nothing queued', + `pack ids read from expression_packs: ${report.packIds.length}`, + ]; + for (const packId of report.liveGuildPackIds) { + lines.push(`skipped pack ${packId}: that id is a live guild, its expressions were never read`); + } + if (report.residues.length === 0) { + lines.push('no guild_emojis or guild_stickers rows are owned by a pack id'); + } + for (const residue of report.residues) { + lines.push(`pack ${residue.packId}: ${residue.emojis.length} emojis, ${residue.stickers.length} stickers`); + for (const emoji of residue.emojis) { + lines.push(` guild_emojis guild_id=${residue.packId} emoji_id=${emoji.id} name=${JSON.stringify(emoji.name)}`); + } + for (const sticker of residue.stickers) { + lines.push( + ` guild_stickers guild_id=${residue.packId} sticker_id=${sticker.id} name=${JSON.stringify(sticker.name)}`, + ); + } + } + lines.push( + `totals: ${report.emojiCount} emojis, ${report.stickerCount} stickers, ${report.emojiCount + report.stickerCount} assets`, + ); + return lines.join('\n'); +}