Rewrite automatic event handling system

2025-12-06 11:57:08 +00:00 · 2025-08-15 14:45:46 +02:00
parent 2fab06d340
commit 387d20a4f0
9 changed files with 173 additions and 87 deletions
--- a/lib/www/server/events/handlers/detect-project-configuration-change.js
+++ b/lib/www/server/events/handlers/detect-project-configuration-change.js
@@ -1,4 +1,3 @@
-const project = require('../../lib/db/project');
 const { ALERT, ERROR, WARNING, NOTICE, INFO, DEBUG } = require('DOUGAL_ROOT/debug')(__filename);

 class DetectProjectConfigurationChange {
@@ -10,7 +9,7 @@ class DetectProjectConfigurationChange {

 		// Grab project configurations.
 		// NOTE that this will run asynchronously
-		this.run({channel: "project"}, ctx);
+		//this.run({channel: "project"}, ctx);
 	}

 	async run (data, ctx) {
@@ -28,13 +27,13 @@ class DetectProjectConfigurationChange {
 		try {
 			DEBUG("Project configuration change detected")

-			const projects = await project.get();
-			project.organisations.setCache(projects);
+			const projects = await ctx.db.project.get();
+			ctx.db.project.organisations.setCache(projects);

 			const _ctx_data = {};
 			for (let pid of projects.map(i => i.pid)) {
 				DEBUG("Retrieving configuration for", pid);
-				const cfg = await project.configuration.get(pid);
+				const cfg = await ctx.db.project.configuration.get(pid);
 				if (cfg?.archived === true) {
 					DEBUG(pid, "is archived. Ignoring");
 					continue;
--- a/lib/www/server/events/handlers/detect-soft-start.js
+++ b/lib/www/server/events/handlers/detect-soft-start.js
@@ -1,5 +1,3 @@
-const { schema2pid } = require('../../lib/db/connection');
-const { event } = require('../../lib/db');
 const { ALERT, ERROR, WARNING, NOTICE, INFO, DEBUG } = require('DOUGAL_ROOT/debug')(__filename);

 class DetectSoftStart {
@@ -33,6 +31,11 @@ class DetectSoftStart {
 			const prev = this.prev?.payload?.new?.meta;
 			// DEBUG("%j", prev);
 			// DEBUG("%j", cur);
+			if (cur.lineStatus == "online" || prev.lineStatus == "online") {
+				DEBUG("lineStatus is online, assuming not in a soft start situation");
+				return;
+			}
+
 			DEBUG("cur.num_guns: %d\ncur.num_active: %d\nprv.num_active: %d\ntest passed: %j", cur.num_guns, cur.num_active, prev.num_active, cur.num_active >= 1 && !prev.num_active && cur.num_active < cur.num_guns);


@@ -40,7 +43,7 @@ class DetectSoftStart {
 				INFO("Soft start detected @", cur.tstamp);

 				// FIXME Shouldn't need to use schema2pid as pid already present in payload.
-				const projectId = await schema2pid(cur._schema ?? prev._schema);
+				const projectId = await ctx.schema2pid(cur._schema ?? prev._schema);

 				// TODO: Try and grab the corresponding comment from the configuration?
 				const payload = {
@@ -50,12 +53,16 @@ class DetectSoftStart {
 					meta: {auto: true, author: `*${this.constructor.name}*`}
 				};
 				DEBUG("Posting event", projectId, payload);
-				await event.post(projectId, payload);
-
 			} else if (cur.num_active == cur.num_guns && prev.num_active < cur.num_active) {
+				if (ctx.dryRun) {
+					DEBUG(`DRY RUN: await ctx.db.event.post(${projectId}, ${payload});`);
+				} else {
+					await ctx.db.event.post(projectId, payload);
+				}
+
 				INFO("Full volume detected @", cur.tstamp);

-				const projectId = await schema2pid(cur._schema ?? prev._schema);
+				const projectId = await ctx.schema2pid(cur._schema ?? prev._schema);

 				// TODO: Try and grab the corresponding comment from the configuration?
 				const payload = {
@@ -65,7 +72,11 @@ class DetectSoftStart {
 					meta: {auto: true, author: `*${this.constructor.name}*`}
 				};
 				DEBUG("Posting event", projectId, payload);
-				await event.post(projectId, payload);
+				if (ctx.dryRun) {
+					DEBUG(`DRY RUN: await ctx.db.event.post(${projectId}, ${payload});`);
+				} else {
+					await ctx.db.event.post(projectId, payload);
+				}
 			}

 		} catch (err) {
--- a/lib/www/server/events/handlers/detect-soleol.js
+++ b/lib/www/server/events/handlers/detect-soleol.js
@@ -1,5 +1,3 @@
-const { schema2pid } = require('../../lib/db/connection');
-const { event } = require('../../lib/db');
 const { ALERT, ERROR, WARNING, NOTICE, INFO, DEBUG } = require('DOUGAL_ROOT/debug')(__filename);

 class DetectSOLEOL {
@@ -43,7 +41,7 @@ class DetectSOLEOL {

 				// We must use schema2pid because the pid may not have been
 				// populated for this event.
-				const projectId = await schema2pid(cur._schema ?? prev._schema);
+				const projectId = await ctx.schema2pid(cur._schema ?? prev._schema);
 				const labels = ["FSP", "FGSP"];
 				const remarks = `SEQ ${cur._sequence}, SOL ${cur.lineName}, BSP: ${(cur.speed*3.6/1.852).toFixed(1)} kt, Water depth: ${Number(cur.waterDepth).toFixed(0)} m.`;
 				const payload = {
@@ -55,14 +53,18 @@ class DetectSOLEOL {
 					meta: {auto: true, author: `*${this.constructor.name}*`}
 				}
 				INFO("Posting event", projectId, payload);
-				await event.post(projectId, payload);
+				if (ctx.dryRun) {
+					DEBUG(`DRY RUN: await ctx.db.event.post(${projectId}, ${payload});`);
+				} else {
+					await ctx.db.event.post(projectId, payload);
+				}
 			} else if (prev.lineName == cur.lineName && prev._sequence == cur._sequence &&
 				prev.lineStatus == "online" && cur.lineStatus != "online" && sequence) {
 				INFO("Transition to OFFLINE detected");

-				const projectId = await schema2pid(prev._schema ?? cur._schema);
+				const projectId = await ctx.schema2pid(prev._schema ?? cur._schema);
 				const labels = ["LSP", "LGSP"];
-				const remarks = `SEQ ${cur._sequence}, EOL ${cur.lineName}, BSP: ${(cur.speed*3.6/1.852).toFixed(1)} kt, Water depth: ${Number(cur.waterDepth).toFixed(0)} m.`;
+				const remarks = `SEQ ${prev._sequence}, EOL ${prev.lineName}, BSP: ${(prev.speed*3.6/1.852).toFixed(1)} kt, Water depth: ${Number(prev.waterDepth).toFixed(0)} m.`;
 				const payload = {
 					type: "sequence",
 					sequence,
@@ -72,7 +74,11 @@ class DetectSOLEOL {
 					meta: {auto: true, author: `*${this.constructor.name}*`}
 				}
 				INFO("Posting event", projectId, payload);
-				await event.post(projectId, payload);
+				if (ctx.dryRun) {
+					DEBUG(`DRY RUN: await ctx.db.event.post(${projectId}, ${payload});`);
+				} else {
+					await ctx.db.event.post(projectId, payload);
+				}
 			}

 		} catch (err) {
--- a/lib/www/server/events/handlers/index.js
+++ b/lib/www/server/events/handlers/index.js
@@ -8,37 +8,6 @@ const Handlers = [
 	require('./detect-fdsp')
 ];

-function init (ctx) {
-
-	const instances = Handlers.map(Handler => new Handler(ctx));
-
-	function prepare (data, ctx) {
-		const promises = [];
-		for (let instance of instances) {
-			const promise = new Promise(async (resolve, reject) => {
-				try {
-					DEBUG("Run", instance.author);
-					const result = await instance.run(data, ctx);
-					DEBUG("%s result: %O", instance.author, result);
-					resolve(result);
-				} catch (err) {
-					ERROR("%s error:\n%O", instance.author, err);
-					reject(err);
-				}
-			});
-			promises.push(promise);
-		}
-		return promises;
-	}
-
-	function despatch (data, ctx) {
-		return Promise.allSettled(prepare(data, ctx));
-	}
-
-	return { instances, prepare, despatch };
-}
-
 module.exports = {
 	Handlers,
-	init
 };
--- a/lib/www/server/events/handlers/report-line-change-time.js
+++ b/lib/www/server/events/handlers/report-line-change-time.js
@@ -1,6 +1,3 @@
-const { event, project } = require('../../lib/db');
-const { withinValidity } = require('../../lib/utils/ranges');
-const unique = require('../../lib/utils/unique');
 const { ALERT, ERROR, WARNING, NOTICE, INFO, DEBUG } = require('DOUGAL_ROOT/debug')(__filename);

 class ReportLineChangeTime {
@@ -44,7 +41,7 @@ class ReportLineChangeTime {

 			async function getLineChangeTime (data, forward = false) {
 				if (forward) {
-					const ospEvents = await event.list(projectId, {label: "FGSP"});
+					const ospEvents = await ctx.db.event.list(projectId, {label: "FGSP"});
 					// DEBUG("ospEvents", ospEvents);
 					const osp = ospEvents.filter(i => i.tstamp > data.tstamp).pop();
 					DEBUG("fsp", osp);
@@ -55,7 +52,7 @@ class ReportLineChangeTime {
 						return { lineChangeTime: osp.tstamp - data.tstamp, osp };
 					}
 				} else {
-					const ospEvents = await event.list(projectId, {label: "LGSP"});
+					const ospEvents = await ctx.db.event.list(projectId, {label: "LGSP"});
 					// DEBUG("ospEvents", ospEvents);
 					const osp = ospEvents.filter(i => i.tstamp < data.tstamp).shift();
 					DEBUG("lsp", osp);
@@ -96,16 +93,20 @@ class ReportLineChangeTime {
 					const opts = {jpq};

 					if (Array.isArray(seq)) {
-						opts.sequences = unique(seq).filter(i => !!i);
+						opts.sequences = ctx.unique(seq).filter(i => !!i);
 					} else {
 						opts.sequence = seq;
 					}

-					const staleEvents = await event.list(projectId, opts);
+					const staleEvents = await ctx.db.event.list(projectId, opts);
 					DEBUG(staleEvents.length ?? 0, "events to delete");
 					for (let staleEvent of staleEvents) {
 						DEBUG(`Deleting event id ${staleEvent.id} (seq = ${staleEvent.sequence}, point = ${staleEvent.point})`);
-						await event.del(projectId, staleEvent.id);
+						if (ctx.dryRun) {
+							DEBUG(`await ctx.db.event.del(${projectId}, ${staleEvent.id});`);
+						} else {
+							await ctx.db.event.del(projectId, staleEvent.id);
+						}
 					}
 				}
 			}
@@ -180,7 +181,11 @@ class ReportLineChangeTime {

 			const maybePostEvent = async (projectId, payload) => {
 				DEBUG("Posting event", projectId, payload);
-				await event.post(projectId, payload);
+				if (ctx.dryRun) {
+					DEBUG(`await ctx.db.event.post(${projectId}, ${payload});`);
+				} else {
+					await ctx.db.event.post(projectId, payload);
+				}
 			}


@@ -192,7 +197,7 @@ class ReportLineChangeTime {
 				const data = n;
 				DEBUG("INSERT seen: will add lct events related to ", data.id);

-				if (withinValidity(data.validity)) {
+				if (ctx.withinValidity(data.validity)) {
 					DEBUG("Event within validity period", data.validity, new Date());

 					data.tstamp = new Date(data.tstamp);
--- a/lib/www/server/events/index.js
+++ b/lib/www/server/events/index.js
@@ -1,29 +1,101 @@
+const nodeAsync = require('async'); // npm install async
 const { listen } = require('../lib/db/notify');
+const db = require('../lib/db'); // Adjust paths; include all needed DB utils
+const { schema2pid } = require('../lib/db/connection');
+const unique = require('../lib/utils/unique'); // If needed by handlers
+const withinValidity = require('../lib/utils/ranges').withinValidity; // If needed
+const { ALERT, ERROR, DEBUG } = require('DOUGAL_ROOT/debug')(__filename);
+
+// List of handler classes (add more as needed)
+const handlerClasses = require('./handlers').Handlers;
+
+// Channels to listen to (hardcoded for simplicity; could scan handlers for mentions)
 const channels = require('../lib/db/channels');
-const handlers = require('./handlers');
-const { ActionsQueue } = require('../lib/queue');
-const { ERROR, INFO, DEBUG } = require('DOUGAL_ROOT/debug')(__filename);

-function start () {
+// Queue config: Process one at a time for order; max retries=3
+const eventQueue = nodeAsync.queue(async (task, callback) => {
+	const { data, ctx } = task;
+	DEBUG(`Processing event on channel ${data.channel} with timestamp ${data._received ?? 'unknown'}`);

-	const queue = new ActionsQueue();
-	const ctx = {}; // Context object
+	for (const handler of ctx.handlers) {
+		try {
+			await handler.run(data, ctx);
+		} catch (err) {
+			ERROR(`Error in handler ${handler.constructor.name}:`, err);
+			// Retry logic: Could add task.retries++, re-enqueue if < max
+		}
+	}

-	const { prepare, despatch } = handlers.init(ctx);
+	if (typeof callback === 'function') {
+		// async v3.2.6+ does not use callsbacks with AsyncFunctions, but anyway
+		callback();
+	}
+}, 1); // Concurrency=1 for strict order

-	listen(channels, function (data) {
-		DEBUG("Incoming data", data);
+eventQueue.error((err, task) => {
+	ALERT(`Queue error processing task:`, err, task);
+});

-		// We don't bother awaiting
-		queue.enqueue(() => despatch(data, ctx));
-		DEBUG("Queue size", queue.length());
+// Main setup function (call from server init)
+async function setupEventHandlers(projectsConfig) {
+	// Shared context
+	const ctx = {
+		dryRun: Boolean(process.env.DOUGAL_HANDLERS_DRY_RUN) ?? false, // If true, don't commit changes
+		projects: { configuration: projectsConfig }, // From user config
+		handlers: handlerClasses.map(Cls => new Cls()), // Instances
+		// DB utils (add more as needed)
+		db,
+		schema2pid,
+		unique,
+		withinValidity
+		// Add other utils, e.g., ctx.logger = DEBUG;
+	};
+
+	// Optional: Replay recent events on startup to rebuild state
+	// await replayRecentEvents(ctx);
+
+	// Setup listener
+	const subscriber = await listen(channels, (rawData) => {
+		const data = {
+			...rawData,
+			enqueuedAt: new Date() // For monitoring
+		};
+		eventQueue.push({ data, ctx });
 	});

-	INFO("Events manager started");
+	DEBUG('Event handler system initialized with channels:', channels);
+	if (ctx.dryRun) {
+		DEBUG('DRY RUNNING');
+	}
+
+	// Return for cleanup if needed
+	return {
+		close: () => {
+			subscriber.events.removeAllListeners();
+			subscriber.close();
+			eventQueue.kill();
+		}
+	};
 }

-module.exports = { start }
+// Optional: Replay last N events to rebuild handler state (e.g., this.prev)
+// async function replayRecentEvents(ctx) {
+// 	try {
+// 		// Example: Fetch last 10 realtime events, sorted by tstamp
+// 		const recentRealtime = await event.listAllProjects({ channel: 'realtime', limit: 10, sort: 'tstamp DESC' });
+// 		// Assume event.listAllProjects is a custom DB method; implement if needed
+//
+// 		// Enqueue in original order (reverse sort)
+// 		recentRealtime.reverse().forEach((evt) => {
+// 			const data = { channel: 'realtime', payload: { new: evt } };
+// 			eventQueue.push({ data, ctx });
+// 		});
+//
+// 		// Similarly for 'event' channel if needed
+// 		DEBUG('Replayed recent events for state rebuild');
+// 	} catch (err) {
+// 		ERROR('Error replaying events:', err);
+// 	}
+// }

-if (require.main === module) {
-	start();
-}
+module.exports = { setupEventHandlers };