fix: re-fetch message inside the write in getThreadName

app/lib/methods/getThreadName.test.ts

@@ -0,0 +1,180 @@
+import getThreadName from './getThreadName';
+import database from '../database';
+import { getMessageById } from '../database/services/Message';
+import { getThreadById } from '../database/services/Thread';
+import getSingleMessage from './getSingleMessage';
+import { Encryption } from '../encryption';
+import log from './helpers/log';
+
+jest.mock('../database', () => ({
+	__esModule: true,
+	default: { active: {} }
+}));
+
+jest.mock('../database/services/Message', () => ({
+	getMessageById: jest.fn()
+}));
+
+jest.mock('../database/services/Thread', () => ({
+	getThreadById: jest.fn()
+}));
+
+jest.mock('./getSingleMessage', () => ({
+	__esModule: true,
+	default: jest.fn()
+}));
+
+jest.mock('../encryption', () => ({
+	Encryption: { decryptMessage: jest.fn() }
+}));
+
+jest.mock('./helpers/log', () => ({
+	__esModule: true,
+	default: jest.fn()
+}));
+
+jest.mock('@nozbe/watermelondb/RawRecord', () => ({
+	sanitizedRaw: jest.fn((raw: any) => raw)
+}));
+
+const mockedGetMessageById = getMessageById as jest.MockedFunction<typeof getMessageById>;
+const mockedGetThreadById = getThreadById as jest.MockedFunction<typeof getThreadById>;
+const mockedGetSingleMessage = getSingleMessage as jest.MockedFunction<typeof getSingleMessage>;
+const mockedDecryptMessage = Encryption.decryptMessage as jest.MockedFunction<typeof Encryption.decryptMessage>;
+const mockedLog = log as jest.MockedFunction<typeof log>;
+
+// mimics watermelon rejecting an update prepared on a record another writer already touched
+const buildMessageRecord = (id: string) => {
+	const record: {
+		id: string;
+		tmsg: string | undefined;
+		stale: boolean;
+		prepareUpdate: jest.Mock;
+		update: jest.Mock;
+	} = {
+		id,
+		tmsg: undefined,
+		stale: false,
+		update: jest.fn((updater: (m: any) => void) => {
+			updater(record);
+			return record;
+		}),
+		prepareUpdate: jest.fn((updater: (m: any) => void) => {
+			if (record.stale) {
+				throw new Error('Cannot update a record with pending changes');
+			}
+			updater(record);
+			return record;
+		})
+	};
+	return record;
+};
+
+describe('getThreadName', () => {
+	const batch = jest.fn();
+	const threadCollection = { schema: {}, prepareCreate: jest.fn((cb: (t: any) => void) => cb({}) ?? { type: 'thread' }) };
+
+	beforeEach(() => {
+		jest.clearAllMocks();
+		batch.mockResolvedValue(undefined);
+		(database as any).active = {
+			get: jest.fn(() => threadCollection),
+			write: jest.fn((fn: () => Promise<void>) => fn()),
+			batch
+		};
+	});
+
+	it('re-fetches the message inside the write so a concurrent writer during the network gap does not break the update', async () => {
+		const staleRecord = buildMessageRecord('MESSAGE_ID');
+		const freshRecord = buildMessageRecord('MESSAGE_ID');
+		mockedGetMessageById.mockResolvedValue(staleRecord as any);
+
+		mockedGetThreadById.mockResolvedValue(null as any);
+
+		mockedGetSingleMessage.mockImplementation(() => {
+			// a sync write lands while we are off the lock: the old record is superseded
+			staleRecord.stale = true;
+			mockedGetMessageById.mockResolvedValue(freshRecord as any);
+			return Promise.resolve({ _id: 'THREAD_ID', msg: 'thread name' } as any);
+		});
+		mockedDecryptMessage.mockImplementation((message: any) => Promise.resolve(message));
+
+		const tmsg = await getThreadName('ROOM_ID', 'THREAD_ID', 'MESSAGE_ID');
+
+		expect(tmsg).toBe('thread name');
+		expect(mockedLog).not.toHaveBeenCalled();
+		expect(staleRecord.prepareUpdate).not.toHaveBeenCalled();
+		expect(freshRecord.prepareUpdate).toHaveBeenCalledTimes(1);
+		expect(freshRecord.tmsg).toBe('thread name');
+		expect(batch).toHaveBeenCalledTimes(1);
+		// the message is read again only after the network and decryption work
+		expect(mockedGetMessageById).toHaveBeenCalledTimes(2);
+	});
+
+	it('updates the message when the local thread name differs from the cached tmsg', async () => {
+		const record = buildMessageRecord('MESSAGE_ID');
+		record.tmsg = 'old name';
+		mockedGetMessageById.mockResolvedValue(record as any);
+		mockedGetThreadById.mockResolvedValue({ msg: 'new name' } as any);
+
+		const tmsg = await getThreadName('ROOM_ID', 'THREAD_ID', 'MESSAGE_ID');
+
+		expect(tmsg).toBe('new name');
+		expect(record.update).toHaveBeenCalledTimes(1);
+		expect(record.tmsg).toBe('new name');
+		expect(mockedGetSingleMessage).not.toHaveBeenCalled();
+		expect(batch).not.toHaveBeenCalled();
+	});
+
+	it('does not write when the cached tmsg already matches the local thread', async () => {
+		const record = buildMessageRecord('MESSAGE_ID');
+		record.tmsg = 'same name';
+		mockedGetMessageById.mockResolvedValue(record as any);
+		mockedGetThreadById.mockResolvedValue({ msg: 'same name' } as any);
+
+		const tmsg = await getThreadName('ROOM_ID', 'THREAD_ID', 'MESSAGE_ID');
+
+		expect(tmsg).toBe('same name');
+		expect((database.active as any).write).not.toHaveBeenCalled();
+		expect(record.update).not.toHaveBeenCalled();
+	});
+
+	it('falls back to the first attachment title when the local thread has no msg', async () => {
+		const record = buildMessageRecord('MESSAGE_ID');
+		mockedGetMessageById.mockResolvedValue(record as any);
+		mockedGetThreadById.mockResolvedValue({ msg: undefined, attachments: [{ title: 'attachment title' }] } as any);
+
+		const tmsg = await getThreadName('ROOM_ID', 'THREAD_ID', 'MESSAGE_ID');
+
+		expect(tmsg).toBe('attachment title');
+		expect(record.tmsg).toBe('attachment title');
+	});
+
+	it('skips creating the thread when another writer created it during the network gap', async () => {
+		const record = buildMessageRecord('MESSAGE_ID');
+		mockedGetMessageById.mockResolvedValue(record as any);
+		mockedGetThreadById.mockResolvedValueOnce(null as any).mockResolvedValueOnce({ msg: 'thread name' } as any);
+		mockedGetSingleMessage.mockResolvedValue({ _id: 'THREAD_ID', msg: 'thread name' } as any);
+		mockedDecryptMessage.mockImplementation((message: any) => Promise.resolve(message));
+
+		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();
+	});
+
+	it('logs and resolves undefined when fetching the remote thread fails', async () => {
+		const error = new Error('network down');
+		mockedGetMessageById.mockResolvedValue(buildMessageRecord('MESSAGE_ID') as any);
+		mockedGetThreadById.mockResolvedValue(null as any);
+		mockedGetSingleMessage.mockRejectedValue(error);
+
+		const tmsg = await getThreadName('ROOM_ID', 'THREAD_ID', 'MESSAGE_ID');
+
+		expect(tmsg).toBeUndefined();
+		expect(mockedLog).toHaveBeenCalledWith(error);
+		expect(batch).not.toHaveBeenCalled();
+	});
+});

app/lib/methods/getThreadName.ts

@@ -15,7 +15,7 @@ const getThreadName = async (rid: string, tmid: string, messageId: string): Prom
 	try {
 		const db = database.active;
 		const threadCollection = db.get('threads');
-		const messageRecord = await getMessageById(messageId);
+		let messageRecord = await getMessageById(messageId);
 		let threadRecord = await getThreadById(tmid);
 		if (threadRecord) {
 			tmsg = buildThreadName(threadRecord);
@@ -34,6 +34,7 @@ const getThreadName = async (rid: string, tmid: string, messageId: string): Prom
 			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);