From 8b2506cc3d8200d0481dddcf24cf4726060fdc7f Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:44:49 -0700 Subject: [PATCH] add queue worker repository --- server/src/repositories/index.ts | 3 ++ .../queue-workers/IQueueWorkersRepository.ts | 7 ++++ .../MongoQueueWorkersRepository.ts | 33 +++++++++++++++++++ 3 files changed, 43 insertions(+) create mode 100644 server/src/repositories/queue-workers/IQueueWorkersRepository.ts create mode 100644 server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts diff --git a/server/src/repositories/index.ts b/server/src/repositories/index.ts index f45768b02..d36d9d189 100644 --- a/server/src/repositories/index.ts +++ b/server/src/repositories/index.ts @@ -42,3 +42,6 @@ export { default as MongoGeoChecksRepository } from "@/repositories/geo-checks/M export * from "@/repositories/dlq/IDLQRepository.js"; export { default as MongoDLQRepository } from "@/repositories/dlq/MongoDLQRepository.js"; + +export * from "@/repositories/queue-workers/IQueueWorkersRepository.js"; +export { default as MongoQueueWorkersRepository } from "@/repositories/queue-workers/MongoQueueWorkersRepository.js"; diff --git a/server/src/repositories/queue-workers/IQueueWorkersRepository.ts b/server/src/repositories/queue-workers/IQueueWorkersRepository.ts new file mode 100644 index 000000000..b1f7e76b1 --- /dev/null +++ b/server/src/repositories/queue-workers/IQueueWorkersRepository.ts @@ -0,0 +1,7 @@ +import type { QueueMode, QueueWorker } from "@/types/settings.js"; + +export interface IQueueWorkersRepository { + upsert(workerId: string, mode: QueueMode): Promise; + findRecent(maxAgeMs: number): Promise; + deleteById(workerId: string): Promise; +} diff --git a/server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts b/server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts new file mode 100644 index 000000000..b8155ea1d --- /dev/null +++ b/server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts @@ -0,0 +1,33 @@ +import type { QueueMode, QueueWorker } from "@/types/settings.js"; +import type { IQueueWorkersRepository } from "./IQueueWorkersRepository.js"; +import type { QueueWorkerDocument } from "@/db/models/QueueWorker.js"; +import { QueueWorkerModel } from "@/db/models/QueueWorker.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 => { + await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, lastSeenAt: new Date() } }, { upsert: true }); + }; + + findRecent = async (maxAgeMs: number): Promise => { + 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 => { + await QueueWorkerModel.deleteOne({ _id: workerId }); + }; +} + +export default MongoQueueWorkersRepository;