1- import { MongoClient , ObjectId } from "mongodb" ;
1+ import { EventEmitter } from "node:events" ;
2+ import { MongoClient , ObjectId , type ChangeStream } from "mongodb" ;
23import type {
34 TxOBEvent ,
45 TxOBProcessorClient ,
56 TxOBProcessorClientOpts ,
67 TxOBTransactionProcessorClient ,
8+ WakeupEmitter ,
79} from "../processor.js" ;
810import { getDate } from "../date.js" ;
911
@@ -20,12 +22,17 @@ const createReadyToProcessFilter = (maxErrors: number) => ({
2022 errors : { $lt : maxErrors } ,
2123} ) ;
2224
25+ export type CreateProcessorClientOpts < EventType extends string > = {
26+ mongo : MongoClient ;
27+ db : string ;
28+ collection ?: string ;
29+ limit ?: number ;
30+ } ;
31+
2332export const createProcessorClient = < EventType extends string > (
24- mongo : MongoClient ,
25- db : string ,
26- collection : string = "events" ,
27- limit : number = 100 ,
33+ opts : CreateProcessorClientOpts < EventType > ,
2834) : TxOBProcessorClient < EventType > => {
35+ const { mongo, db, collection = "events" , limit = 100 } = opts ;
2936 const getEventsToProcess = async (
3037 opts : TxOBProcessorClientOpts ,
3138 ) : Promise < Pick < TxOBEvent < EventType > , "id" | "errors" > [ ] > => {
@@ -142,3 +149,87 @@ export const createProcessorClient = <EventType extends string>(
142149 transaction,
143150 } ;
144151} ;
152+
153+ type CreateWakeupEmitterOpts = {
154+ mongo : MongoClient ;
155+ db : string ;
156+ collection ?: string ;
157+ } ;
158+
159+ /**
160+ * Creates a MongoDB Change Stream-based wakeup emitter for reducing polling frequency.
161+ * This watches for INSERT operations on the events collection and emits wakeup signals.
162+ *
163+ * **Important**: MongoDB Change Streams require a replica set or sharded cluster.
164+ * If your MongoDB instance is a standalone server, you must convert it to a single-node
165+ * replica set by running `rs.initiate()` in the mongo shell.
166+ *
167+ * If the database is not configured for Change Streams, an error will be emitted via
168+ * the 'error' event on the returned WakeupEmitter. The error typically occurs when
169+ * the change stream attempts to connect.
170+ *
171+ * See: https://www.mongodb.com/docs/manual/changeStreams/
172+ *
173+ * @param opts - Options for the wakeup emitter
174+ * @returns A WakeupEmitter that emits 'wakeup' events when new events are inserted.
175+ * Errors (including replica set requirement failures) are emitted via the 'error' event.
176+ * @throws Does not throw synchronously. Errors are emitted via the 'error' event.
177+ */
178+ export const createWakeupEmitter = async (
179+ opts : CreateWakeupEmitterOpts ,
180+ ) : Promise < WakeupEmitter & { close : ( ) => Promise < void > } > => {
181+ const { mongo, db, collection = "events" } = opts ;
182+ const emitter = new EventEmitter ( ) ;
183+
184+ // Get the collection to watch
185+ const eventsCollection = mongo . db ( db ) . collection ( collection ) ;
186+
187+ // Create a change stream that watches for insert operations
188+ // We only care about inserts - retries after backoff are handled by fallback polling
189+ // Note: watch() may not error immediately - errors typically occur when the change
190+ // stream attempts to connect, which happens asynchronously
191+ const changeStream : ChangeStream = eventsCollection . watch (
192+ [
193+ {
194+ $match : {
195+ operationType : "insert" ,
196+ } ,
197+ } ,
198+ ] ,
199+ {
200+ fullDocument : "default" ,
201+ } ,
202+ ) ;
203+
204+ // Handle change stream events
205+ changeStream . on ( "change" , ( ) => {
206+ emitter . emit ( "wakeup" ) ;
207+ } ) ;
208+
209+ // Handle change stream errors
210+ // Common errors include:
211+ // - "Change streams are only supported on replica sets" (standalone MongoDB)
212+ // - Connection errors
213+ // - Permission errors
214+ changeStream . on ( "error" , ( err ) => {
215+ emitter . emit ( "error" , err ) ;
216+ } ) ;
217+
218+ // Handle change stream close
219+ changeStream . on ( "close" , ( ) => {
220+ emitter . emit ( "error" , new Error ( "MongoDB Change Stream closed" ) ) ;
221+ } ) ;
222+
223+ // Return a WakeupEmitter that wraps the EventEmitter
224+ return {
225+ on : ( event : "wakeup" , listener : ( ) => void ) => {
226+ emitter . on ( event , listener ) ;
227+ } ,
228+ off : ( event : "wakeup" , listener : ( ) => void ) => {
229+ emitter . off ( event , listener ) ;
230+ } ,
231+ close : async ( ) => {
232+ await changeStream . close ( ) ;
233+ } ,
234+ } ;
235+ } ;
0 commit comments