mirror of
https://github.com/fluxerapp/fluxer.git
synced 2026-09-02 21:04:06 +03:00
chore(expressions): remove the pack residue cleanup tool (#2288)
This commit is contained in:
Hampus
@@ -8,8 +8,7 @@
|
||||
"typecheck": "tsgo --noEmit",
|
||||
"dev": "tsx watch --clear-screen=false src/AppEntrypoint.ts",
|
||||
"start": "tsx src/AppEntrypoint.ts",
|
||||
"start:worker": "tsx src/WorkerEntrypoint.ts",
|
||||
"cleanup:expression-pack-residue": "tsx src/ExpressionPackResidueEntrypoint.ts"
|
||||
"start:worker": "tsx src/WorkerEntrypoint.ts"
|
||||
},
|
||||
"dependencies": {
|
||||
"@atproto/api": "catalog:",
|
||||
|
||||
@@ -64,11 +64,7 @@ const workspacePlugin = {
|
||||
rmSync(OUT_DIR, {recursive: true, force: true});
|
||||
|
||||
await build({
|
||||
entryPoints: [
|
||||
join(API_ROOT, 'src/AppEntrypoint.ts'),
|
||||
join(API_ROOT, 'src/WorkerEntrypoint.ts'),
|
||||
join(API_ROOT, 'src/ExpressionPackResidueEntrypoint.ts'),
|
||||
],
|
||||
entryPoints: [join(API_ROOT, 'src/AppEntrypoint.ts'), join(API_ROOT, 'src/WorkerEntrypoint.ts')],
|
||||
outdir: OUT_DIR,
|
||||
bundle: true,
|
||||
platform: 'node',
|
||||
|
||||
@@ -1,110 +0,0 @@
|
||||
// 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<void>> {
|
||||
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<void> {
|
||||
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);
|
||||
});
|
||||
@@ -1,451 +0,0 @@
|
||||
// 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<ExpressionPackSeedRow, 'pack_id'>({
|
||||
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<void> {
|
||||
await new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
async function freePort(): Promise<number> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
await new GuildRepository().upsertPartial(guildId, {name: `guild ${guildId}`, owner_id: CREATOR_ID});
|
||||
}
|
||||
|
||||
async function seedEmoji(guildId: GuildID, emojiId: EmojiID, name: string): Promise<void> {
|
||||
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<void> {
|
||||
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<Array<string>> {
|
||||
const emojis = await new GuildRepository().listEmojis(guildId);
|
||||
return emojis.map((emoji) => emoji.name).sort();
|
||||
}
|
||||
|
||||
async function stickerNames(guildId: GuildID): Promise<Array<string>> {
|
||||
const stickers = await new GuildRepository().listStickers(guildId);
|
||||
return stickers.map((sticker) => sticker.name).sort();
|
||||
}
|
||||
|
||||
async function queuedS3Keys(): Promise<Array<string>> {
|
||||
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>): 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');
|
||||
});
|
||||
});
|
||||
@@ -1,206 +0,0 @@
|
||||
// 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<ExpressionPackIdRow, 'pack_id'>({
|
||||
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<ExpressionPackResidueEmoji>;
|
||||
stickers: Array<ExpressionPackResidueSticker>;
|
||||
}
|
||||
|
||||
interface ExpressionPackResidueReport {
|
||||
applied: boolean;
|
||||
packIds: Array<GuildID>;
|
||||
liveGuildPackIds: Array<GuildID>;
|
||||
residues: Array<ExpressionPackResidue>;
|
||||
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<ExpressionPackResidueReport> {
|
||||
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<Array<GuildID>> {
|
||||
const packIds = new Set<GuildID>();
|
||||
let pageState: string | null = null;
|
||||
try {
|
||||
do {
|
||||
const page: PagedQueryResult<ExpressionPackIdRow> = await fetchPage<ExpressionPackIdRow>(
|
||||
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<GuildID>): Promise<Array<GuildID>> {
|
||||
const liveGuildIds: Array<GuildID> = [];
|
||||
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<ExpressionPackResidue> {
|
||||
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<void> {
|
||||
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<string> = [
|
||||
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');
|
||||
}
|
||||
Reference in New Issue
Block a user