Enhance UpdateManager for object event handling and subscription management

- Refactored constructor to utilize socketClient for improved clarity.
- Added methods to subscribe and unsubscribe from object creation and update events.
- Implemented functionality to retrieve and set object updates in Etcd.
- Improved logging for event handling and error management.
This commit is contained in:
Tom Butcher 2025-08-18 01:09:00 +01:00
parent d695772a3a
commit 1b86d256ca

View File

@ -1,7 +1,9 @@
import { etcdServer } from '../database/etcd.js';
import log4js from 'log4js';
import { loadConfig } from '../config.js';
import NodeCache from 'node-cache';
import { etcdServer } from '../database/etcd.js';
import { updateObjectCache } from '../database/database.js';
const config = loadConfig();
// Setup logger
@ -12,40 +14,88 @@ logger.level = config.server.logLevel;
* UpdateManager handles tracking object updates using Etcd and broadcasts update events via websockets.
*/
export class UpdateManager {
constructor(socketManager) {
this.socketManager = socketManager;
this.setupUpdatesListeners();
constructor(socketClient) {
this.socketClient = socketClient;
}
async updateObject(object) {
// Add an 'update' event to the 'updates' stream
logger.debug('Updating object:', object._id);
try {
const updateData = {
_id: object._id,
type: object.type,
updatedAt: new Date().toISOString()
};
async subscribeToObjectNew(objectType) {
await etcdServer.onKeyPutEvent(
`/${objectType}s/new`,
this.socketClient.socketId,
(key, value) => {
logger.trace('Object new event:', value);
this.socketClient.socket.emit('objectNew', {
_id: value,
objectType: objectType
});
}
);
return { success: true };
}
await etcdServer.set(
`/updates/${object.type}s/${object._id}`,
updateData
async subscribeToObjectUpdate(id, objectType) {
await etcdServer.onKeyPutEvent(
`/${objectType}s/${id}/object`,
this.socketClient.socketId,
(key, value) => {
logger.trace('Object update event:', id);
this.socketClient.socket.emit('objectUpdate', {
_id: id,
objectType: objectType,
object: { ...value }
});
}
);
return { success: true };
}
async removeObjectNewListener(objectType) {
await etcdServer.removeKeyWatcher(
`/${objectType}s/new`,
this.socketClient.socketId,
'put'
);
return { success: true };
}
async removeObjectUpdateListener(id, objectType) {
await etcdServer.removeKeyWatcher(
`/${objectType}s/${id}/object`,
this.socketClient.socketId,
'put'
);
return { success: true };
}
async getObjectUpdate(id, objectType) {
try {
const objectUpdate = {
_id: id,
objectType: objectType,
object: await etcdServer.get(`/${objectType}s/${id}/object`)
};
logger.trace(`Returning path: /${objectType}s/${id}/object`);
return objectUpdate;
} catch (error) {
logger.error(
`UpdateManager: Failed to get current value for /${objectType}s/${id}/object:`,
error
);
logger.info(`Update event for id: ${object._id}`);
} catch (err) {
logger.error(`Error adding update event to: ${object._id}:`, err);
throw err;
return { error: 'Not found' };
}
}
setupUpdatesListeners() {
etcdServer.onPrefixPut('/updates', (key, value) => {
const id = key.split('/').pop();
logger.debug('Update object event:', id);
this.socketManager.broadcast('notify_object_update', {
...value
});
});
logger.info('Subscribed to Etcd stream for update changes.');
async setObjectUpdate(id, objectType, value) {
try {
await etcdServer.set(`/${objectType}s/${id}/object`, value);
logger.trace(`Set value for path: /${objectType}s/${id}/object`);
return true;
} catch (error) {
logger.error(
`Failed to set value for /${objectType}s/${id}/object:`,
error
);
return false;
}
}
}