reorg to domain and api

server/src/domain/queue-workers/queue-worker.model.ts

@@ -0,0 +1,18 @@
+import { Schema, model } from "mongoose";
+import { QueueModes } from "@/domain/app-settings/app-settings.type.js";
+import { QueueWorkerDocument } from "@/domain/queue-workers/queue-worker.type.js";
+
+const QueueWorkerSchema = new Schema<QueueWorkerDocument>(
+	{
+		_id: { type: String, required: true },
+		mode: { type: String, enum: QueueModes, required: true },
+		lastSeenAt: { type: Date, default: Date.now, expires: 30 },
+	},
+	{ timestamps: true }
+);
+
+const QueueWorkerModel = model<QueueWorkerDocument>("QueueWorker", QueueWorkerSchema);
+
+export type { QueueWorkerDocument };
+export { QueueWorkerModel };
+export default QueueWorkerModel;

server/src/domain/queue-workers/queue-worker.repository.interface.ts

@@ -0,0 +1,8 @@
+import type { QueueMode } from "@/domain/app-settings/app-settings.type.js";
+import type { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js";
+
+export interface IQueueWorkersRepository {
+	upsert(workerId: string, mode: QueueMode): Promise<void>;
+	findRecent(maxAgeMs: number): Promise<QueueWorker[]>;
+	deleteById(workerId: string): Promise<void>;
+}

server/src/domain/queue-workers/queue-worker.repository.mongo.ts

@@ -0,0 +1,33 @@
+import type { QueueMode } from "@/domain/app-settings/app-settings.type.js";
+import type { IQueueWorkersRepository } from "./queue-worker.repository.interface.js";
+import type { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js";
+import { QueueWorkerModel, type QueueWorkerDocument } from "@/domain/queue-workers/queue-worker.model.js";
+const SERVICE_NAME = "MongoQueueWorkersRepository";
+
+class MongoQueueWorkersRepository implements IQueueWorkersRepository {
+	static SERVICE_NAME = SERVICE_NAME;
+
+	protected toEntity = (doc: QueueWorkerDocument): QueueWorker => {
+		return {
+			workerId: doc._id,
+			mode: doc.mode,
+			lastSeenAt: doc.lastSeenAt.getTime(),
+		};
+	};
+
+	upsert = async (workerId: string, mode: QueueMode): Promise<void> => {
+		await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, lastSeenAt: new Date() } }, { upsert: true });
+	};
+
+	findRecent = async (maxAgeMs: number): Promise<QueueWorker[]> => {
+		const cutoff = new Date(Date.now() - maxAgeMs);
+		const docs = await QueueWorkerModel.find({ lastSeenAt: { $gte: cutoff } });
+		return docs.map(this.toEntity);
+	};
+
+	deleteById = async (workerId: string): Promise<void> => {
+		await QueueWorkerModel.deleteOne({ _id: workerId });
+	};
+}
+
+export default MongoQueueWorkersRepository;

server/src/domain/queue-workers/queue-worker.type.ts

@@ -0,0 +1,18 @@
+import { type QueueMode } from "@/domain/app-settings/app-settings.type.js";
+
+// Heartbeat presence record for a scheduler instance. _id is the scheduler's
+// workerId (hostname:pid:uuid). The TTL index on lastSeenAt garbage-collects
+// records for instances that have stopped heartbeating.
+export interface QueueWorkerDocument {
+	_id: string;
+	mode: QueueMode;
+	lastSeenAt: Date;
+	createdAt: Date;
+	updatedAt: Date;
+}
+
+export type QueueWorker = {
+	workerId: string; // hostname:pid:uuid
+	mode: QueueMode;
+	lastSeenAt: number; // epoch ms of the last heartbeat
+};