Ignore duplicate messages in SQLiteDBAdapter

The SQLiteDBAdapter now includes an 'onConflictDoNothing' statement during record insertion to handle duplicate messages. Additionally, a test behavior has been implemented to verify that duplicate messages are disregarded.
usherlabs · Jun 7, 2024 · 08bbda7 · 08bbda7
1 parent 5c21b7c
commit 08bbda7
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 1 deletion.
diff --git a/packages/core/src/plugins/logStore/database/SQLiteDBAdapter.ts b/packages/core/src/plugins/logStore/database/SQLiteDBAdapter.ts
@@ -297,7 +297,7 @@ export class SQLiteDBAdapter extends DatabaseAdapter {
 			payload: payload,
 			content_bytes: payload.length,
 		};
-		await this.dbClient.insert(streamDataTable).values(record);
+		await this.dbClient.insert(streamDataTable).values(record).onConflictDoNothing();
 
 		this.emit('write', record.payload);
 

diff --git a/packages/core/test/integration/plugins/logStore/database/SQLite.test.ts b/packages/core/test/integration/plugins/logStore/database/SQLite.test.ts
@@ -84,6 +84,17 @@ describe('SQLite', () => {
 		await testDb.store(createMockMessage(1));
 	});
 
+	test('duplicate messages are ignored', async () => {
+		const testDb = new SQLiteDBAdapter({ type: 'sqlite', dataPath: dbPath });
+		await testDb.store(createMockMessage(1));
+		await testDb.store(createMockMessage(1));
+		// try querying more than necessary
+		const stream = testDb.queryLast(MOCK_STREAM_ID, MOCK_PARTITION, 5);
+		const contentValues = await streamToContentValues(stream);
+		// there should be only one message
+		expect(contentValues).toEqual([1, 0]);
+	})
+
 	describe('methods test', () => {
 		let db: SQLiteDBAdapter;