fix: UNIQUE constraint failed on threads.id

app/lib/methods/getThreadName.test.ts

@@ -160,9 +160,10 @@ describe('getThreadName', () => {
 		const tmsg = await getThreadName('ROOM_ID', 'THREAD_ID', 'MESSAGE_ID');
 
 		expect(tmsg).toBe('thread name');
-		expect((database.active as any).write).not.toHaveBeenCalled();
 		expect(batch).not.toHaveBeenCalled();
 		expect(threadCollection.prepareCreate).not.toHaveBeenCalled();
+		expect(record.update).toHaveBeenCalledTimes(1);
+		expect(record.tmsg).toBe('thread name');
 	});
 
 	it('logs and resolves undefined when fetching the remote thread fails', async () => {
@@ -177,4 +178,56 @@ describe('getThreadName', () => {
 		expect(mockedLog).toHaveBeenCalledWith(error);
 		expect(batch).not.toHaveBeenCalled();
 	});
+
+	it('creates the thread once when two callers race for the same tmid', async () => {
+		const storedThreadIds = new Set<string>();
+		let writeQueue: Promise<unknown> = Promise.resolve();
+		const threadsCollection = {
+			schema: {},
+			prepareCreate: jest.fn((cb: (t: any) => void) => {
+				const raw: any = {};
+				cb({
+					set _raw(value: any) {
+						Object.assign(raw, value);
+					},
+					get _raw() {
+						return raw;
+					}
+				});
+				return { id: raw.id, table: 'threads' };
+			})
+		};
+		(database as any).active = {
+			get: jest.fn(() => threadsCollection),
+			write: jest.fn((fn: () => Promise<void>) => {
+				const run = writeQueue.then(fn);
+				writeQueue = run.catch(() => {});
+				return run;
+			}),
+			batch: jest.fn((...records: any[]) => {
+				records.forEach(record => {
+					if (record?.table !== 'threads') return;
+					if (storedThreadIds.has(record.id)) {
+						throw new Error('UNIQUE constraint failed: threads.id');
+					}
+					storedThreadIds.add(record.id);
+				});
+				return Promise.resolve();
+			})
+		};
+
+		const messageRecords = ['MESSAGE_A', 'MESSAGE_B'].map(buildMessageRecord);
+		mockedGetMessageById.mockImplementation(id => Promise.resolve(messageRecords.find(message => message.id === id) as any));
+		mockedGetThreadById.mockImplementation(() =>
+			Promise.resolve(storedThreadIds.has('THREAD_ID') ? ({ msg: 'thread name' } as any) : null)
+		);
+		mockedGetSingleMessage.mockResolvedValue({ _id: 'THREAD_ID', msg: 'thread name' } as any);
+		mockedDecryptMessage.mockImplementation((message: any) => Promise.resolve(message));
+
+		await Promise.all([getThreadName('ROOM_ID', 'THREAD_ID', 'MESSAGE_A'), getThreadName('ROOM_ID', 'THREAD_ID', 'MESSAGE_B')]);
+
+		expect(mockedLog).not.toHaveBeenCalled();
+		expect(threadsCollection.prepareCreate).toHaveBeenCalledTimes(1);
+		expect(messageRecords.map(message => message.tmsg)).toEqual(['thread name', 'thread name']);
+	});
 });

app/lib/methods/getThreadName.ts

@@ -16,7 +16,7 @@ const getThreadName = async (rid: string, tmid: string, messageId: string): Prom
 		const db = database.active;
 		const threadCollection = db.get('threads');
 		let messageRecord = await getMessageById(messageId);
-		let threadRecord = await getThreadById(tmid);
+		const threadRecord = await getThreadById(tmid);
 		if (threadRecord) {
 			tmsg = buildThreadName(threadRecord);
 			if (tmsg !== messageRecord?.tmsg) {
@@ -30,23 +30,28 @@ const getThreadName = async (rid: string, tmid: string, messageId: string): Prom
 			const thread = await getSingleMessage(tmid);
 			const decryptedThread = await Encryption.decryptMessage(thread);
 			tmsg = buildThreadName(decryptedThread as IMessage);
-			// check it again to avoid race condition
-			threadRecord = await getThreadById(tmid);
-			if (!threadRecord) {
-				await db.write(async () => {
-					messageRecord = await getMessageById(messageId);
-					await db.batch(
-						threadCollection?.prepareCreate((t: TThreadModel) => {
-							t._raw = sanitizedRaw({ id: thread._id }, threadCollection.schema);
-							if (t.subscription) t.subscription.id = rid;
-							Object.assign(t, { ...thread, ...decryptedThread });
-						}),
-						messageRecord?.prepareUpdate(m => {
+			await db.write(async () => {
+				const createdMeanwhile = await getThreadById(tmid);
+				messageRecord = await getMessageById(messageId);
+				if (createdMeanwhile) {
+					if (tmsg !== messageRecord?.tmsg) {
+						await messageRecord?.update(m => {
 							m.tmsg = tmsg;
-						})
-					);
-				});
-			}
+						});
+					}
+					return;
+				}
+				await db.batch(
+					threadCollection?.prepareCreate((t: TThreadModel) => {
+						t._raw = sanitizedRaw({ id: thread._id }, threadCollection.schema);
+						if (t.subscription) t.subscription.id = rid;
+						Object.assign(t, { ...thread, ...decryptedThread });
+					}),
+					messageRecord?.prepareUpdate(m => {
+						m.tmsg = tmsg;
+					})
+				);
+			});
 		}
 	} catch (e) {
 		log(e);