RFC: Optimize Room & Subscription Stream Updates via Atomic Multi-Record Database Batching

[Shevilll]

Optimize Room & Subscription Stream Updates via Atomic Multi-Record Database Batching

• Type: Performance & Architecture RFC
• Target Subsystem: Database Layer (WatermelonDB) & Subscription Synchronization (app/lib/methods/subscriptions/rooms.ts)
• Status: Proposed

1. Problem Statement & Root Cause Analysis

When a user logs in, reconnects, or experiences high workspace activity, the client receives a torrent of real-time events for room and subscription updates via the DDP WebSocket streaming API (channel stream-notify-user, events subscriptions and rooms).

In the current implementation of app/lib/methods/subscriptions/rooms.ts, these updates are routed through debouncedUpdate with a WINDOW_TIME of 500ms:

const debouncedUpdate = (subscription: ISubscription) => {
	if (!subTimer) {
		subTimer = setTimeout(() => {
			const batch = queue;
			queue = {};
			subTimer = null;
			Object.keys(batch).forEach(key => {
				InteractionManager.runAfterInteractions(() => {
					// ...
					createOrUpdateSubscription(sub, room);
				});
			});
		}, WINDOW_TIME);
	}
	queue[subscription.rid ? getSubQueueId(subscription.rid) : getRoomQueueId(subscription._id)] = subscription;
};

The Bottleneck

1. Transaction Overhead: While debouncedUpdate accumulates notifications for 500ms, when the timer fires, it loops over each key in the queue and schedules an InteractionManager.runAfterInteractions call for each item individually.
2. Individual db.write Calls: Inside each scheduled task, createOrUpdateSubscription is invoked. Inside this function, a new independent database transaction is executed:

   await db.write(async () => {
       await db.batch(batch); // Typically holds only 1 subscription and/or 1 room + lastMessage
   });

3. Lock Contention & UI Thread Block: SQLite (used under the hood by WatermelonDB) uses filesystem-level locks. Running 20, 50, or 100 consecutive asynchronous db.write transactions in parallel or rapid succession forces SQLite to open and commit transactions repeatedly. This consumes enormous CPU cycles, drives up JS-to-native bridge traffic, causes severe SQLite thread lock contention, and blocks the React Native JavaScript thread—leading to noticeable UI freezes (lag) and dropped frames.

---

2. Proposed Architectural Solution

Instead of executing a separate database transaction for every single room and subscription update in the batch, we should collect all prepared database operations across all items in the batch, and write them to the database in a single, atomic transaction using one db.write call.

Architectural Comparison

• Current Architecture:
  DDP Stream Notifications -> Queue in debouncedUpdate -> After 500ms: Loop over N queued items -> Individual InteractionManager scheduling -> Individual db.write Transactions (N times)
• Proposed Architecture (Atomic Batching):
  DDP Stream Notifications -> Queue in debouncedUpdate -> After 500ms: Collect all N items in memory -> Bulk Fetch Existing Models (Q.oneOf) -> Prepare updates & creates in memory -> Single db.write Transaction (1 time)

Performance Impact Analysis

• I/O Operations: Reduces from $O(N)$ filesystem transactions to $O(1)$ atomic transaction.
• UI Thread Latency: Eliminates consecutive React Native bridge crossings, reducing UI block time from several seconds (during a cold sync/reconnect of 100 rooms) to less than 50 milliseconds.

---

3. Proposed Schema & Blueprint

The proposed implementation aggregates all queued changes, fetches existing records in bulk using Q.oneOf, prepares all updates/creations synchronously, and executes them in one batch.

Refactored Blueprint of app/lib/methods/subscriptions/rooms.ts

// Proposed Refactoring for app/lib/methods/subscriptions/rooms.ts

import { sanitizedRaw } from '@nozbe/watermelondb/RawRecord';
import { InteractionManager } from 'react-native';
import { Model, Q } from '@nozbe/watermelondb';
import database from '../../database';
import { merge } from '../helpers/mergeSubscriptionsRooms';
import buildMessage from '../helpers/buildMessage';
import { getSubscriptionByRoomId } from '../../database/services/Subscription';
import { getMessageById } from '../../database/services/Message';
import { Encryption } from '../../encryption';
import { store } from '../../store/auxStore';
import { type ISubscription, type IRoom, SubscriptionType } from '../../../definitions';

let queue: { [key: string]: ISubscription | IRoom } = {};
let subTimer: ReturnType<typeof setTimeout> | null | false = null;
const WINDOW_TIME = 500;

export const debouncedUpdate = (subscription: ISubscription) => {
	if (!subTimer) {
		subTimer = setTimeout(() => {
			const batch = { ...queue };
			queue = {};
			subTimer = null;

			InteractionManager.runAfterInteractions(async () => {
				try {
					await processSubscriptionBatch(batch);
				} catch (error) {
					console.error('[Database] Failed to write subscription batch:', error);
				}
			});
		}, WINDOW_TIME);
	}
	queue[subscription.rid ? getSubQueueId(subscription.rid) : getRoomQueueId(subscription._id)] = subscription;
};

/**
 * Core Batch Processor: Integrates all updates into a single database write.
 */
const processSubscriptionBatch = async (batchQueue: { [key: string]: ISubscription | IRoom }) => {
	const db = database.active;
	const subCollection = db.get('subscriptions');
	const roomsCollection = db.get('rooms');
	const messagesCollection = db.get('messages');

	const rids: string[] = [];
	const roomIds: string[] = [];
	const lastMessageIds: string[] = [];

	const processedPairs: { sub: ISubscription; room: IRoom; rid: string }[] = [];

	// 1. Group subscriptions and rooms together in-memory
	Object.keys(batchQueue).forEach(key => {
		if (/SUB/.test(key)) {
			const sub = batchQueue[key] as ISubscription;
			if (!sub.tunread) sub.tunread = [];
			const roomQueueId = getRoomQueueId(sub.rid);
			const room = batchQueue[roomQueueId] as IRoom;
			delete batchQueue[roomQueueId];

			rids.push(sub.rid);
			processedPairs.push({ sub, room, rid: sub.rid });
		} else {
			const room = batchQueue[key] as IRoom;
			if (room.t === SubscriptionType.OMNICHANNEL && room.onHold && room.waitingResponse) {
				return;
			}
			const subQueueId = getSubQueueId(room._id);
			const sub = batchQueue[subQueueId] as ISubscription;
			delete batchQueue[subQueueId];

			rids.push(room._id);
			processedPairs.push({ sub, room, rid: room._id });
		}
	});

	if (processedPairs.length === 0) return;

	// 2. Bulk Fetch Existing Subscriptions and Messages from DB to minimize database roundtrips
	const existingSubs = await subCollection.query(Q.where('id', Q.oneOf(rids))).fetch();
	const subMap = new Map(existingSubs.map(s => [s.id, s]));

	// Collect any message IDs we might need to find
	processedPairs.forEach(({ sub, room }) => {
		const tmp = merge(sub, room);
		if (tmp.lastMessage) {
			lastMessageIds.push(tmp.lastMessage._id);
		}
	});

	const existingMsgs = lastMessageIds.length > 0 
		? await messagesCollection.query(Q.where('id', Q.oneOf(lastMessageIds))).fetch() 
		: [];
	const msgMap = new Map(existingMsgs.map(m => [m.id, m]));

	const operations: Model[] = [];
	const decryptedRids = new Set<string>();

	const { subscribedRoom } = store.getState().room;

	// 3. Process pairs sequentially in memory to generate models
	for (const { sub, room, rid } of processedPairs) {
		const tmp = merge(sub, room);
		const dbSub = subMap.get(tmp.rid);

		if (dbSub) {
			try {
				const update = dbSub.prepareUpdate(s => {
					Object.assign(s, tmp);
					if (sub?.announcement && sub.announcement !== dbSub.announcement) {
						s.bannerClosed = false;
					}
					if (dbSub.hideUnreadStatus && sub?.hasOwnProperty('hideUnreadStatus')) {
						if (dbSub.hideUnreadStatus !== sub.hideUnreadStatus) {
							s.hideUnreadStatus = !!sub.hideUnreadStatus;
						}
					}
				});
				operations.push(update);
			} catch (e) {
				console.error(e);
			}
		} else {
			try {
				const create = subCollection.prepareCreate(s => {
					s._raw = sanitizedRaw({ id: tmp.rid }, subCollection.schema);
					Object.assign(s, tmp);
					if (s.roomUpdatedAt) {
						s.roomUpdatedAt = new Date();
					}
				});
				operations.push(create);
			} catch (e) {
				console.error(e);
			}
		}

		// 4. Handle lastMessage updates inside the same batch
		if (tmp.lastMessage && subscribedRoom !== tmp.rid) {
			const lastMessage = buildMessage(tmp.lastMessage);
			if (lastMessage) {
				const msgRecord = msgMap.get(lastMessage._id);
				if (msgRecord) {
					operations.push(
						msgRecord.prepareUpdate(() => {
							Object.assign(msgRecord, lastMessage);
						})
					);
				} else {
					operations.push(
						messagesCollection.prepareCreate(m => {
							m._raw = sanitizedRaw({ id: lastMessage._id }, messagesCollection.schema);
							if (m.subscription) {
								m.subscription.id = lastMessage.rid;
							}
							return Object.assign(m, lastMessage);
						})
					);
				}
			}
		}
		decryptedRids.add(tmp.rid);
	}

	// 5. Single atomic write to database
	if (operations.length > 0) {
		await db.write(async () => {
			await db.batch(operations);
		});
	}

	// 6. Post-write decryption processes
	Encryption.decryptPendingSubscriptions();
	decryptedRids.forEach(rid => {
		Encryption.decryptPendingMessages(rid);
		Encryption.getRoomInstance(rid);
	});
};

4. Implementation Steps

• Step 1: Implement processSubscriptionBatch in app/lib/methods/subscriptions/rooms.ts.
• Step 2: Replace debouncedUpdate with the refactored version.
• Step 3: Verify offline support by disconnecting the network, generating a series of message/room changes, re-establishing connection, and confirming that the entire burst of updates is committed in exactly 1 transaction with no UI freeze.

[coderabbitai]

🔗 Related PRs

#7395 - chore: add Drizzle schema and generated migrations for default and per-server databases [merged]
#7396 - feat: add expo-sqlite driver adapter, key service, and live-query hooks [merged]
#7407 - refactor: cut all database call sites over to the facade [merged]
#7408 - feat: port client data into the encrypted database at first boot [merged]
#7426 - fix: re-subscribe room streams on DDP reconnect [merged]

---

📝 Issue Planner

_{Check the box below or use the @coderabbitai plan command to generate an implementation plan and prompts that you can use with your favorite coding assistant.}

• Create Plan

---

🧪 Issue enrichment is currently in open beta.

You can configure auto-planning by selecting labels in the issue_enrichment configuration.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord!