perf(message): unlog bucket index writes on the read path (#2209)

fluxer_api/src/api/channel/repositories/message/MessageDataRepository.test.ts

@@ -1,16 +1,37 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+import {generateSnowflake} from '@fluxer/snowflake/src/Snowflake';
 import * as BucketUtils from '@fluxer/snowflake/src/SnowflakeBuckets';
 import {afterEach, beforeEach, describe, expect, it} from 'vitest';
 import {type ChannelID, createChannelID, createMessageID, createUserID, type MessageID} from '../../../BrandedTypes';
-import {fetchOne, setCassandraQueryExecutorForTesting} from '../../../database/CassandraQueryExecution';
+import {
+	deleteOneOrMany,
+	fetchMany,
+	fetchOne,
+	setCassandraQueryExecutorForTesting,
+} from '../../../database/CassandraQueryExecution';
 import type {CassandraParams, KvQueryMeta, PreparedQuery} from '../../../database/CassandraTypes';
-import type {ChannelStateRow, MessageRow} from '../../../database/types/MessageTypes';
-import {ChannelState} from '../../../Tables';
+import type {
+	ChannelEmptyBucketRow,
+	ChannelMessageBucketRow,
+	ChannelStateRow,
+	MessageRow,
+} from '../../../database/types/MessageTypes';
+import {ChannelEmptyBuckets, ChannelMessageBuckets, ChannelState, Messages} from '../../../Tables';
 import {InMemoryCassandraQueryExecutor} from '../../../test/InMemoryCassandraQueryExecutor';
 import {MessageDataRepository} from './MessageDataRepository';
 
 const FETCH_CHANNEL_STATE = ChannelState.select({where: ChannelState.where.eq('channel_id'), limit: 1});
+const FETCH_MESSAGE_BUCKETS = ChannelMessageBuckets.select({
+	columns: ['bucket'],
+	where: ChannelMessageBuckets.where.eq('channel_id'),
+	orderBy: {col: 'bucket', direction: 'DESC'},
+});
+const FETCH_EMPTY_BUCKETS = ChannelEmptyBuckets.select({
+	columns: ['bucket'],
+	where: ChannelEmptyBuckets.where.eq('channel_id'),
+	orderBy: {col: 'bucket', direction: 'DESC'},
+});
 
 class RecordingCassandraQueryExecutor {
 	readonly statements: Array<string> = [];
@@ -123,6 +144,66 @@ describe('MessageDataRepository.upsertMessage round trips', () => {
 	});
 });
 
+describe('MessageDataRepository read scans maintain the bucket index', () => {
+	beforeEach(() => {
+		executor = new RecordingCassandraQueryExecutor();
+		setCassandraQueryExecutorForTesting(executor);
+	});
+	afterEach(() => {
+		setCassandraQueryExecutorForTesting(new InMemoryCassandraQueryExecutor());
+	});
+	it('re-indexes a bucket that still holds messages and keeps the index batch unlogged', async () => {
+		const channelId = createChannelID(generateSnowflake());
+		const messageId = createMessageID(generateSnowflake());
+		const bucket = BucketUtils.makeBucket(messageId);
+		const repository = new MessageDataRepository();
+		await repository.upsertMessage(makeMessageRow(channelId, messageId), null);
+		await deleteOneOrMany(ChannelMessageBuckets.deleteByPk({channel_id: channelId, bucket}));
+		expect(await loadIndexedBuckets(channelId)).toEqual([]);
+		executor.statements.length = 0;
+		executor.batchAtomicity.length = 0;
+
+		const messages = await repository.listMessages(channelId);
+
+		expect(messages.map((message) => message.id)).toEqual([messageId]);
+		expect(await loadIndexedBuckets(channelId)).toEqual([bucket]);
+		expect(await loadEmptyBuckets(channelId)).toEqual([]);
+		expect(executor.countStatements('upsert:channel_message_buckets')).toBe(1);
+		expect(executor.batchAtomicity).toEqual([false]);
+	});
+	it('drops a drained bucket from the index and keeps the empty-bucket batch unlogged', async () => {
+		const channelId = createChannelID(generateSnowflake());
+		const messageId = createMessageID(generateSnowflake());
+		const bucket = BucketUtils.makeBucket(messageId);
+		const repository = new MessageDataRepository();
+		await repository.upsertMessage(makeMessageRow(channelId, messageId), null);
+		await deleteOneOrMany(Messages.deleteByPk({channel_id: channelId, bucket, message_id: messageId}));
+		executor.statements.length = 0;
+		executor.batchAtomicity.length = 0;
+
+		const messages = await repository.listMessages(channelId);
+
+		expect(messages).toEqual([]);
+		expect(await loadIndexedBuckets(channelId)).toEqual([]);
+		expect(await loadEmptyBuckets(channelId)).toEqual([bucket]);
+		expect(executor.batchAtomicity).toEqual([false]);
+	});
+});
+
 async function loadChannelState(channelId: ChannelID): Promise<ChannelStateRow | null> {
 	return fetchOne<ChannelStateRow>(FETCH_CHANNEL_STATE.bind({channel_id: channelId}));
 }
+
+async function loadIndexedBuckets(channelId: ChannelID): Promise<Array<number>> {
+	const rows = await fetchMany<Pick<ChannelMessageBucketRow, 'bucket'>>(
+		FETCH_MESSAGE_BUCKETS.bind({channel_id: channelId}),
+	);
+	return rows.map((row) => row.bucket);
+}
+
+async function loadEmptyBuckets(channelId: ChannelID): Promise<Array<number>> {
+	const rows = await fetchMany<Pick<ChannelEmptyBucketRow, 'bucket'>>(
+		FETCH_EMPTY_BUCKETS.bind({channel_id: channelId}),
+	);
+	return rows.map((row) => row.bucket);
+}

fluxer_api/src/api/channel/repositories/message/MessageDataRepository.ts

@@ -490,7 +490,7 @@ export class MessageDataRepository {
 				bucket,
 			}),
 		);
-		await batch.execute(true);
+		await batch.execute(false);
 	}
 
 	private async markBucketEmpty(channelId: ChannelID, bucket: number): Promise<void> {
@@ -508,7 +508,7 @@ export class MessageDataRepository {
 				updated_at: new Date(),
 			}),
 		);
-		await batch.execute(true);
+		await batch.execute(false);
 	}
 
 	private async touchChannelHasMessages(channelId: ChannelID): Promise<void> {