777 lines
23 KiB
TypeScript
777 lines
23 KiB
TypeScript
import * as plugins from '../plugins.js';
|
|
import type { TsView } from '../tsview.classes.tsview.js';
|
|
import type * as interfaces from './interfaces.streaming.js';
|
|
import type { IS3ChangeEvent } from '@push.rocks/smartbucket';
|
|
|
|
/**
|
|
* Subscription entry tracking a client's subscription to a resource
|
|
*/
|
|
interface ISubscriptionEntry {
|
|
connectionId: string;
|
|
subscriptionId: string;
|
|
createdAt: Date;
|
|
}
|
|
|
|
/**
|
|
* MongoDB watcher entry
|
|
*/
|
|
interface IMongoWatcherEntry {
|
|
watcher: plugins.mongodb.ChangeStream;
|
|
subscriptions: Map<string, ISubscriptionEntry>; // connectionId -> subscription
|
|
}
|
|
|
|
/**
|
|
* S3 watcher entry
|
|
*/
|
|
interface IS3WatcherEntry {
|
|
watcher: plugins.smartbucket.BucketWatcher;
|
|
subscriptions: Map<string, ISubscriptionEntry>; // connectionId -> subscription
|
|
}
|
|
|
|
/**
|
|
* ChangeStreamManager manages real-time change streaming for both MongoDB and S3.
|
|
*
|
|
* Features:
|
|
* - MongoDB Change Streams for real-time database updates
|
|
* - S3 BucketWatcher for polling-based S3 change detection
|
|
* - Subscription management per WebSocket client
|
|
* - Activity stream with ring buffer for recent events
|
|
* - Automatic cleanup on client disconnect
|
|
*/
|
|
export class ChangeStreamManager {
|
|
private tsview: TsView;
|
|
private typedSocket: plugins.typedserver.TypedServer['typedsocket'] | null = null;
|
|
|
|
// MongoDB watchers: "db/collection" -> watcher entry
|
|
private mongoWatchers: Map<string, IMongoWatcherEntry> = new Map();
|
|
|
|
// S3 watchers: "bucket/prefix" -> watcher entry
|
|
private s3Watchers: Map<string, IS3WatcherEntry> = new Map();
|
|
|
|
// Activity subscribers: connectionId -> subscription entry
|
|
private activitySubscribers: Map<string, ISubscriptionEntry> = new Map();
|
|
|
|
// Activity ring buffer (max 1000 events)
|
|
private activityBuffer: interfaces.IActivityEvent[] = [];
|
|
private readonly ACTIVITY_BUFFER_SIZE = 1000;
|
|
|
|
// Global watchers for the activity stream (started lazily on first subscriber)
|
|
private globalMongoWatcher: plugins.mongodb.ChangeStream | null = null;
|
|
private globalS3Watchers: Map<string, plugins.smartbucket.BucketWatcher> = new Map();
|
|
private globalWatchersActive: boolean = false;
|
|
|
|
// Counter for generating unique subscription IDs
|
|
private subscriptionCounter = 0;
|
|
|
|
constructor(tsview: TsView) {
|
|
this.tsview = tsview;
|
|
}
|
|
|
|
/**
|
|
* Initialize the manager with a TypedSocket instance
|
|
*/
|
|
public setTypedSocket(typedSocket: plugins.typedserver.TypedServer['typedsocket']): void {
|
|
this.typedSocket = typedSocket;
|
|
}
|
|
|
|
/**
|
|
* Generate a unique subscription ID
|
|
*/
|
|
private generateSubscriptionId(): string {
|
|
return `sub_${Date.now()}_${++this.subscriptionCounter}`;
|
|
}
|
|
|
|
/**
|
|
* Get the MongoDB key for a database/collection pair
|
|
*/
|
|
private getMongoKey(database: string, collection: string): string {
|
|
return `${database}/${collection}`;
|
|
}
|
|
|
|
/**
|
|
* Get the S3 key for a bucket/prefix pair
|
|
*/
|
|
private getS3Key(bucket: string, prefix?: string): string {
|
|
return prefix ? `${bucket}/${prefix}` : bucket;
|
|
}
|
|
|
|
// ===========================================
|
|
// MongoDB Change Streams
|
|
// ===========================================
|
|
|
|
/**
|
|
* Subscribe a client to MongoDB collection changes
|
|
*/
|
|
public async subscribeToMongo(
|
|
connectionId: string,
|
|
database: string,
|
|
collection: string
|
|
): Promise<{ success: boolean; subscriptionId: string }> {
|
|
const key = this.getMongoKey(database, collection);
|
|
|
|
let entry = this.mongoWatchers.get(key);
|
|
|
|
// Create watcher if it doesn't exist
|
|
if (!entry) {
|
|
const watcher = await this.createMongoWatcher(database, collection);
|
|
if (!watcher) {
|
|
return { success: false, subscriptionId: '' };
|
|
}
|
|
|
|
entry = {
|
|
watcher,
|
|
subscriptions: new Map(),
|
|
};
|
|
this.mongoWatchers.set(key, entry);
|
|
}
|
|
|
|
// Add subscription
|
|
const subscriptionId = this.generateSubscriptionId();
|
|
entry.subscriptions.set(connectionId, {
|
|
connectionId,
|
|
subscriptionId,
|
|
createdAt: new Date(),
|
|
});
|
|
|
|
console.log(`[ChangeStream] MongoDB subscription added: ${key} for connection ${connectionId}`);
|
|
return { success: true, subscriptionId };
|
|
}
|
|
|
|
/**
|
|
* Unsubscribe a client from MongoDB collection changes
|
|
*/
|
|
public async unsubscribeFromMongo(
|
|
connectionId: string,
|
|
database: string,
|
|
collection: string
|
|
): Promise<boolean> {
|
|
const key = this.getMongoKey(database, collection);
|
|
const entry = this.mongoWatchers.get(key);
|
|
|
|
if (!entry) {
|
|
return false;
|
|
}
|
|
|
|
entry.subscriptions.delete(connectionId);
|
|
console.log(`[ChangeStream] MongoDB subscription removed: ${key} for connection ${connectionId}`);
|
|
|
|
// Close watcher if no more subscribers
|
|
if (entry.subscriptions.size === 0) {
|
|
await this.closeMongoWatcher(key);
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* Create a MongoDB change stream for a collection
|
|
*/
|
|
private async createMongoWatcher(
|
|
database: string,
|
|
collection: string
|
|
): Promise<plugins.mongodb.ChangeStream | null> {
|
|
try {
|
|
const db = await this.tsview.getMongoDb();
|
|
if (!db) {
|
|
console.error('[ChangeStream] MongoDB not configured');
|
|
return null;
|
|
}
|
|
|
|
const client = (db as any).mongoDbClient;
|
|
const mongoDb = client.db(database);
|
|
const mongoCollection = mongoDb.collection(collection);
|
|
|
|
// Create change stream
|
|
const changeStream = mongoCollection.watch([], {
|
|
fullDocument: 'updateLookup',
|
|
});
|
|
|
|
// Handle change events
|
|
changeStream.on('change', (change: any) => {
|
|
this.handleMongoChange(database, collection, change);
|
|
});
|
|
|
|
changeStream.on('error', (error: Error) => {
|
|
console.error(`[ChangeStream] MongoDB error for ${database}/${collection}:`, error);
|
|
});
|
|
|
|
console.log(`[ChangeStream] MongoDB watcher created for ${database}/${collection}`);
|
|
return changeStream;
|
|
} catch (error) {
|
|
console.error(`[ChangeStream] Failed to create MongoDB watcher for ${database}/${collection}:`, error);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Handle a MongoDB change event
|
|
*/
|
|
private handleMongoChange(database: string, collection: string, change: any): void {
|
|
const key = this.getMongoKey(database, collection);
|
|
const entry = this.mongoWatchers.get(key);
|
|
|
|
if (!entry) return;
|
|
|
|
// Convert MongoDB change event to our interface
|
|
const event: interfaces.IMongoChangeEvent = {
|
|
type: change.operationType as interfaces.IMongoChangeEvent['type'],
|
|
database,
|
|
collection,
|
|
documentId: change.documentKey?._id?.toString(),
|
|
document: change.fullDocument,
|
|
updateDescription: change.updateDescription,
|
|
timestamp: new Date().toISOString(),
|
|
};
|
|
|
|
// Only add to activity buffer if global watchers are NOT active.
|
|
// When active, the global MongoDB watcher already feeds the activity stream.
|
|
if (!this.globalWatchersActive) {
|
|
this.addToActivityBuffer('mongodb', event);
|
|
}
|
|
|
|
// Push to all subscribed clients
|
|
this.pushMongoChangeToClients(key, event);
|
|
}
|
|
|
|
/**
|
|
* Push MongoDB change to subscribed clients
|
|
*/
|
|
private async pushMongoChangeToClients(
|
|
key: string,
|
|
event: interfaces.IMongoChangeEvent
|
|
): Promise<void> {
|
|
const entry = this.mongoWatchers.get(key);
|
|
if (!entry || !this.typedSocket) return;
|
|
|
|
for (const [connectionId, _sub] of entry.subscriptions) {
|
|
try {
|
|
// Find the connection and push the event
|
|
const connection = await this.typedSocket.findTargetConnection(async (conn: any) => {
|
|
return conn.peer?.id === connectionId;
|
|
});
|
|
|
|
if (connection) {
|
|
const request = this.typedSocket.createTypedRequest<interfaces.IReq_PushMongoChange>(
|
|
'pushMongoChange',
|
|
connection
|
|
);
|
|
await request.fire({ event });
|
|
}
|
|
} catch (error) {
|
|
console.error(`[ChangeStream] Failed to push MongoDB change to ${connectionId}:`, error);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Close a MongoDB change stream
|
|
*/
|
|
private async closeMongoWatcher(key: string): Promise<void> {
|
|
const entry = this.mongoWatchers.get(key);
|
|
if (!entry) return;
|
|
|
|
try {
|
|
await entry.watcher.close();
|
|
this.mongoWatchers.delete(key);
|
|
console.log(`[ChangeStream] MongoDB watcher closed for ${key}`);
|
|
} catch (error) {
|
|
console.error(`[ChangeStream] Error closing MongoDB watcher for ${key}:`, error);
|
|
}
|
|
}
|
|
|
|
// ===========================================
|
|
// S3 Change Watching
|
|
// ===========================================
|
|
|
|
/**
|
|
* Subscribe a client to S3 bucket/prefix changes
|
|
*/
|
|
public async subscribeToS3(
|
|
connectionId: string,
|
|
bucket: string,
|
|
prefix?: string
|
|
): Promise<{ success: boolean; subscriptionId: string }> {
|
|
const key = this.getS3Key(bucket, prefix);
|
|
|
|
let entry = this.s3Watchers.get(key);
|
|
|
|
// Create watcher if it doesn't exist
|
|
if (!entry) {
|
|
const watcher = await this.createS3Watcher(bucket, prefix);
|
|
if (!watcher) {
|
|
return { success: false, subscriptionId: '' };
|
|
}
|
|
|
|
entry = {
|
|
watcher,
|
|
subscriptions: new Map(),
|
|
};
|
|
this.s3Watchers.set(key, entry);
|
|
}
|
|
|
|
// Add subscription
|
|
const subscriptionId = this.generateSubscriptionId();
|
|
entry.subscriptions.set(connectionId, {
|
|
connectionId,
|
|
subscriptionId,
|
|
createdAt: new Date(),
|
|
});
|
|
|
|
console.log(`[ChangeStream] S3 subscription added: ${key} for connection ${connectionId}`);
|
|
return { success: true, subscriptionId };
|
|
}
|
|
|
|
/**
|
|
* Unsubscribe a client from S3 bucket/prefix changes
|
|
*/
|
|
public async unsubscribeFromS3(
|
|
connectionId: string,
|
|
bucket: string,
|
|
prefix?: string
|
|
): Promise<boolean> {
|
|
const key = this.getS3Key(bucket, prefix);
|
|
const entry = this.s3Watchers.get(key);
|
|
|
|
if (!entry) {
|
|
return false;
|
|
}
|
|
|
|
entry.subscriptions.delete(connectionId);
|
|
console.log(`[ChangeStream] S3 subscription removed: ${key} for connection ${connectionId}`);
|
|
|
|
// Close watcher if no more subscribers
|
|
if (entry.subscriptions.size === 0) {
|
|
await this.closeS3Watcher(key);
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* Create an S3 bucket watcher
|
|
*/
|
|
private async createS3Watcher(
|
|
bucket: string,
|
|
prefix?: string
|
|
): Promise<plugins.smartbucket.BucketWatcher | null> {
|
|
try {
|
|
const smartbucket = await this.tsview.getSmartBucket();
|
|
if (!smartbucket) {
|
|
console.error('[ChangeStream] S3 not configured');
|
|
return null;
|
|
}
|
|
|
|
const bucketInstance = await smartbucket.getBucketByName(bucket);
|
|
|
|
// Create watcher using smartbucket's BucketWatcher
|
|
const watcher = bucketInstance.createWatcher({
|
|
prefix: prefix || '',
|
|
pollIntervalMs: 5000,
|
|
bufferTimeMs: 500,
|
|
});
|
|
|
|
// Subscribe to change events
|
|
watcher.changeSubject.subscribe((eventOrEvents: IS3ChangeEvent | IS3ChangeEvent[]) => {
|
|
const events = Array.isArray(eventOrEvents) ? eventOrEvents : [eventOrEvents];
|
|
for (const event of events) {
|
|
this.handleS3Change(bucket, prefix, event);
|
|
}
|
|
});
|
|
|
|
// Start the watcher
|
|
await watcher.start();
|
|
await watcher.readyDeferred.promise;
|
|
|
|
console.log(`[ChangeStream] S3 watcher created for ${bucket}${prefix ? '/' + prefix : ''}`);
|
|
return watcher;
|
|
} catch (error) {
|
|
console.error(`[ChangeStream] Failed to create S3 watcher for ${bucket}:`, error);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Handle an S3 change event
|
|
*/
|
|
private handleS3Change(bucket: string, prefix: string | undefined, event: IS3ChangeEvent): void {
|
|
const key = this.getS3Key(bucket, prefix);
|
|
const entry = this.s3Watchers.get(key);
|
|
|
|
if (!entry) return;
|
|
|
|
// Only add to activity buffer if global watchers are NOT active.
|
|
// When active, the global S3 watchers already feed the activity stream.
|
|
if (!this.globalWatchersActive) {
|
|
this.addToActivityBuffer('s3', event);
|
|
}
|
|
|
|
// Push to all subscribed clients
|
|
this.pushS3ChangeToClients(key, event);
|
|
}
|
|
|
|
/**
|
|
* Push S3 change to subscribed clients
|
|
*/
|
|
private async pushS3ChangeToClients(
|
|
key: string,
|
|
event: IS3ChangeEvent
|
|
): Promise<void> {
|
|
const entry = this.s3Watchers.get(key);
|
|
if (!entry || !this.typedSocket) return;
|
|
|
|
for (const [connectionId, _sub] of entry.subscriptions) {
|
|
try {
|
|
const connection = await this.typedSocket.findTargetConnection(async (conn: any) => {
|
|
return conn.peer?.id === connectionId;
|
|
});
|
|
|
|
if (connection) {
|
|
const request = this.typedSocket.createTypedRequest<interfaces.IReq_PushS3Change>(
|
|
'pushS3Change',
|
|
connection
|
|
);
|
|
await request.fire({ event });
|
|
}
|
|
} catch (error) {
|
|
console.error(`[ChangeStream] Failed to push S3 change to ${connectionId}:`, error);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Close an S3 bucket watcher
|
|
*/
|
|
private async closeS3Watcher(key: string): Promise<void> {
|
|
const entry = this.s3Watchers.get(key);
|
|
if (!entry) return;
|
|
|
|
try {
|
|
await entry.watcher.stop();
|
|
this.s3Watchers.delete(key);
|
|
console.log(`[ChangeStream] S3 watcher closed for ${key}`);
|
|
} catch (error) {
|
|
console.error(`[ChangeStream] Error closing S3 watcher for ${key}:`, error);
|
|
}
|
|
}
|
|
|
|
// ===========================================
|
|
// Activity Stream
|
|
// ===========================================
|
|
|
|
/**
|
|
* Subscribe a client to the activity stream
|
|
*/
|
|
public async subscribeToActivity(connectionId: string): Promise<{ success: boolean; subscriptionId: string }> {
|
|
const subscriptionId = this.generateSubscriptionId();
|
|
|
|
this.activitySubscribers.set(connectionId, {
|
|
connectionId,
|
|
subscriptionId,
|
|
createdAt: new Date(),
|
|
});
|
|
|
|
console.log(`[ChangeStream] Activity subscription added for connection ${connectionId}`);
|
|
|
|
// Start global watchers when the first activity subscriber connects
|
|
if (this.activitySubscribers.size === 1) {
|
|
await this.startGlobalWatchers();
|
|
}
|
|
|
|
return { success: true, subscriptionId };
|
|
}
|
|
|
|
/**
|
|
* Unsubscribe a client from the activity stream
|
|
*/
|
|
public async unsubscribeFromActivity(connectionId: string): Promise<boolean> {
|
|
const result = this.activitySubscribers.delete(connectionId);
|
|
if (result) {
|
|
console.log(`[ChangeStream] Activity subscription removed for connection ${connectionId}`);
|
|
|
|
// Stop global watchers when no activity subscribers remain
|
|
if (this.activitySubscribers.size === 0) {
|
|
await this.stopGlobalWatchers();
|
|
}
|
|
}
|
|
return result;
|
|
}
|
|
|
|
/**
|
|
* Get recent activity events
|
|
*/
|
|
public getRecentActivity(limit: number = 100): interfaces.IActivityEvent[] {
|
|
const count = Math.min(limit, this.activityBuffer.length);
|
|
return this.activityBuffer.slice(-count);
|
|
}
|
|
|
|
/**
|
|
* Emit a MongoDB activity event from an API handler (no change stream required).
|
|
*/
|
|
public emitMongoActivityEvent(event: interfaces.IMongoChangeEvent): void {
|
|
this.addToActivityBuffer('mongodb', event);
|
|
}
|
|
|
|
/**
|
|
* Add an event to the activity buffer
|
|
*/
|
|
private addToActivityBuffer(
|
|
source: 'mongodb' | 's3',
|
|
event: interfaces.IMongoChangeEvent | IS3ChangeEvent
|
|
): void {
|
|
const activityEvent: interfaces.IActivityEvent = {
|
|
id: `evt_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`,
|
|
source,
|
|
event,
|
|
timestamp: new Date().toISOString(),
|
|
};
|
|
|
|
this.activityBuffer.push(activityEvent);
|
|
|
|
// Trim buffer if it exceeds max size
|
|
if (this.activityBuffer.length > this.ACTIVITY_BUFFER_SIZE) {
|
|
this.activityBuffer = this.activityBuffer.slice(-this.ACTIVITY_BUFFER_SIZE);
|
|
}
|
|
|
|
// Push to activity subscribers
|
|
this.pushActivityToClients(activityEvent);
|
|
}
|
|
|
|
/**
|
|
* Push activity event to subscribed clients
|
|
*/
|
|
private async pushActivityToClients(event: interfaces.IActivityEvent): Promise<void> {
|
|
if (!this.typedSocket || this.activitySubscribers.size === 0) return;
|
|
|
|
for (const [connectionId, _sub] of this.activitySubscribers) {
|
|
try {
|
|
const connection = await this.typedSocket.findTargetConnection(async (conn: any) => {
|
|
return conn.peer?.id === connectionId;
|
|
});
|
|
|
|
if (connection) {
|
|
const request = this.typedSocket.createTypedRequest<interfaces.IReq_PushActivityEvent>(
|
|
'pushActivityEvent',
|
|
connection
|
|
);
|
|
await request.fire({ event });
|
|
}
|
|
} catch (error) {
|
|
console.error(`[ChangeStream] Failed to push activity to ${connectionId}:`, error);
|
|
}
|
|
}
|
|
}
|
|
|
|
// ===========================================
|
|
// Global Watchers for Activity Stream
|
|
// ===========================================
|
|
|
|
/**
|
|
* Start global watchers when the first activity subscriber connects.
|
|
* These watch all MongoDB and S3 activity and feed into the activity buffer.
|
|
*/
|
|
private async startGlobalWatchers(): Promise<void> {
|
|
if (this.globalWatchersActive) return;
|
|
this.globalWatchersActive = true;
|
|
|
|
console.log('[ChangeStream] Starting global watchers for activity stream...');
|
|
|
|
await Promise.all([
|
|
this.startGlobalMongoWatcher(),
|
|
this.startGlobalS3Watchers(),
|
|
]);
|
|
}
|
|
|
|
/**
|
|
* Start a deployment-level MongoDB change stream that watches ALL databases/collections.
|
|
*/
|
|
private async startGlobalMongoWatcher(): Promise<void> {
|
|
try {
|
|
const db = await this.tsview.getMongoDb();
|
|
if (!db) {
|
|
console.log('[ChangeStream] MongoDB not configured, skipping global MongoDB watcher');
|
|
return;
|
|
}
|
|
|
|
const client = (db as any).mongoDbClient as plugins.mongodb.MongoClient;
|
|
|
|
// Deployment-level watch — one stream for everything
|
|
const changeStream = client.watch([], {
|
|
fullDocument: 'updateLookup',
|
|
});
|
|
|
|
changeStream.on('change', (change: any) => {
|
|
const database = change.ns?.db || 'unknown';
|
|
const collection = change.ns?.coll || 'unknown';
|
|
|
|
const event: interfaces.IMongoChangeEvent = {
|
|
type: change.operationType as interfaces.IMongoChangeEvent['type'],
|
|
database,
|
|
collection,
|
|
documentId: change.documentKey?._id?.toString(),
|
|
document: change.fullDocument,
|
|
updateDescription: change.updateDescription,
|
|
timestamp: new Date().toISOString(),
|
|
};
|
|
|
|
this.addToActivityBuffer('mongodb', event);
|
|
});
|
|
|
|
changeStream.on('error', (error: Error) => {
|
|
console.error('[ChangeStream] Global MongoDB watcher error:', error);
|
|
});
|
|
|
|
this.globalMongoWatcher = changeStream;
|
|
console.log('[ChangeStream] Global MongoDB watcher started');
|
|
} catch (error) {
|
|
console.warn('[ChangeStream] MongoDB change streams unavailable (requires replica set). MongoDB activity events will come from API operations only.');
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Start S3 bucket watchers — one BucketWatcher per bucket.
|
|
*/
|
|
private async startGlobalS3Watchers(): Promise<void> {
|
|
try {
|
|
const smartbucket = await this.tsview.getSmartBucket();
|
|
if (!smartbucket) {
|
|
console.log('[ChangeStream] S3 not configured, skipping global S3 watchers');
|
|
return;
|
|
}
|
|
|
|
// List all buckets
|
|
const command = new plugins.s3.ListBucketsCommand({});
|
|
const response = await smartbucket.s3Client.send(command) as plugins.s3.ListBucketsCommandOutput;
|
|
const bucketNames = response.Buckets?.map(b => b.Name).filter((name): name is string => !!name) || [];
|
|
|
|
for (const bucketName of bucketNames) {
|
|
try {
|
|
const bucketInstance = await smartbucket.getBucketByName(bucketName);
|
|
const watcher = bucketInstance.createWatcher({
|
|
prefix: '',
|
|
pollIntervalMs: 5000,
|
|
bufferTimeMs: 500,
|
|
});
|
|
|
|
watcher.changeSubject.subscribe((eventOrEvents: IS3ChangeEvent | IS3ChangeEvent[]) => {
|
|
const events = Array.isArray(eventOrEvents) ? eventOrEvents : [eventOrEvents];
|
|
for (const event of events) {
|
|
this.addToActivityBuffer('s3', event);
|
|
}
|
|
});
|
|
|
|
await watcher.start();
|
|
await watcher.readyDeferred.promise;
|
|
|
|
this.globalS3Watchers.set(bucketName, watcher);
|
|
console.log(`[ChangeStream] Global S3 watcher started for bucket: ${bucketName}`);
|
|
} catch (bucketError) {
|
|
console.error(`[ChangeStream] Failed to start global S3 watcher for bucket ${bucketName}:`, bucketError);
|
|
}
|
|
}
|
|
|
|
console.log(`[ChangeStream] Global S3 watchers started (${this.globalS3Watchers.size}/${bucketNames.length} buckets)`);
|
|
} catch (error) {
|
|
console.error('[ChangeStream] Failed to start global S3 watchers:', error);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Stop all global watchers when no activity subscribers remain.
|
|
*/
|
|
private async stopGlobalWatchers(): Promise<void> {
|
|
if (!this.globalWatchersActive) return;
|
|
|
|
console.log('[ChangeStream] Stopping global watchers...');
|
|
|
|
// Close global MongoDB watcher
|
|
if (this.globalMongoWatcher) {
|
|
try {
|
|
await this.globalMongoWatcher.close();
|
|
console.log('[ChangeStream] Global MongoDB watcher stopped');
|
|
} catch (error) {
|
|
console.error('[ChangeStream] Error closing global MongoDB watcher:', error);
|
|
}
|
|
this.globalMongoWatcher = null;
|
|
}
|
|
|
|
// Close all global S3 watchers
|
|
for (const [bucketName, watcher] of this.globalS3Watchers) {
|
|
try {
|
|
await watcher.stop();
|
|
console.log(`[ChangeStream] Global S3 watcher stopped for bucket: ${bucketName}`);
|
|
} catch (error) {
|
|
console.error(`[ChangeStream] Error closing global S3 watcher for ${bucketName}:`, error);
|
|
}
|
|
}
|
|
this.globalS3Watchers.clear();
|
|
|
|
this.globalWatchersActive = false;
|
|
console.log('[ChangeStream] Global watchers stopped');
|
|
}
|
|
|
|
// ===========================================
|
|
// Connection Management
|
|
// ===========================================
|
|
|
|
/**
|
|
* Handle client disconnect - clean up all subscriptions
|
|
*/
|
|
public async handleDisconnect(connectionId: string): Promise<void> {
|
|
console.log(`[ChangeStream] Cleaning up subscriptions for disconnected connection ${connectionId}`);
|
|
|
|
// Clean up MongoDB subscriptions
|
|
for (const [key, entry] of this.mongoWatchers) {
|
|
if (entry.subscriptions.has(connectionId)) {
|
|
entry.subscriptions.delete(connectionId);
|
|
if (entry.subscriptions.size === 0) {
|
|
await this.closeMongoWatcher(key);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Clean up S3 subscriptions
|
|
for (const [key, entry] of this.s3Watchers) {
|
|
if (entry.subscriptions.has(connectionId)) {
|
|
entry.subscriptions.delete(connectionId);
|
|
if (entry.subscriptions.size === 0) {
|
|
await this.closeS3Watcher(key);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Clean up activity subscription
|
|
this.activitySubscribers.delete(connectionId);
|
|
|
|
// Stop global watchers if no activity subscribers remain
|
|
if (this.activitySubscribers.size === 0) {
|
|
await this.stopGlobalWatchers();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Stop all watchers and clean up
|
|
*/
|
|
public async stop(): Promise<void> {
|
|
console.log('[ChangeStream] Stopping all watchers...');
|
|
|
|
// Stop global watchers first
|
|
await this.stopGlobalWatchers();
|
|
|
|
// Close all MongoDB watchers
|
|
for (const key of this.mongoWatchers.keys()) {
|
|
await this.closeMongoWatcher(key);
|
|
}
|
|
|
|
// Close all S3 watchers
|
|
for (const key of this.s3Watchers.keys()) {
|
|
await this.closeS3Watcher(key);
|
|
}
|
|
|
|
// Clear activity buffer and subscribers
|
|
this.activityBuffer = [];
|
|
this.activitySubscribers.clear();
|
|
|
|
console.log('[ChangeStream] All watchers stopped');
|
|
}
|
|
}
|