diff --git a/fluxer_api/src/api/channel/repositories/message/MessageDataRepository.test.ts b/fluxer_api/src/api/channel/repositories/message/MessageDataRepository.test.ts index cf66e7643..6c2fc9e03 100644 --- a/fluxer_api/src/api/channel/repositories/message/MessageDataRepository.test.ts +++ b/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 = []; @@ -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 { return fetchOne(FETCH_CHANNEL_STATE.bind({channel_id: channelId})); } + +async function loadIndexedBuckets(channelId: ChannelID): Promise> { + const rows = await fetchMany>( + FETCH_MESSAGE_BUCKETS.bind({channel_id: channelId}), + ); + return rows.map((row) => row.bucket); +} + +async function loadEmptyBuckets(channelId: ChannelID): Promise> { + const rows = await fetchMany>( + FETCH_EMPTY_BUCKETS.bind({channel_id: channelId}), + ); + return rows.map((row) => row.bucket); +} diff --git a/fluxer_api/src/api/channel/repositories/message/MessageDataRepository.ts b/fluxer_api/src/api/channel/repositories/message/MessageDataRepository.ts index d087649ac..4d316104b 100644 --- a/fluxer_api/src/api/channel/repositories/message/MessageDataRepository.ts +++ b/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 { @@ -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 {