diff --git a/server/src/domain/jobs/job.model.ts b/server/src/domain/jobs/job.model.ts new file mode 100644 index 000000000..f64ef53b8 --- /dev/null +++ b/server/src/domain/jobs/job.model.ts @@ -0,0 +1,79 @@ +import { Schema, model } from "mongoose"; +import { JOB_TYPES, type Job } from "@/domain/jobs/job.type.js"; + +export interface JobDocument extends Omit { + _id: string; +} + +const JobSchema = new Schema( + { + _id: { + type: String, + required: true, + }, + type: { + type: String, + enum: JOB_TYPES, + required: true, + }, + // monitorId for monitor-bound jobs; null for global cleanup jobs. Kept as a String pointer (no populate). + refId: { + type: String, + default: null, + }, + // paused monitor => false; the claim scan skips inactive rows (replaces pauseJob/resumeJob) + isActive: { + type: Boolean, + default: true, + }, + + // Scheduling — epoch ms, so the claim/$min queries operate on plain numbers + nextScheduledAt: { + type: Number, + required: true, + }, + intervalMs: { + type: Number, + default: null, + }, + + // Lock / lease + lockedBy: { + type: String, + default: null, + }, + lockedUntil: { + type: Number, + default: null, + }, + + // Observability + runCount: { + type: Number, + default: 0, + }, + failCount: { + type: Number, + default: 0, + }, + lastFinishedAt: { + type: Number, + default: null, + }, + lastFailReason: { + type: String, + default: null, + }, + }, + { timestamps: true } +); + +// Used for scanning for claims +JobSchema.index({ type: 1, isActive: 1, nextScheduledAt: 1 }); +// for findById, upsertEvaluate, deleteById +JobSchema.index({ refId: 1, type: 1 }); + +const JobModel = model("Job", JobSchema); + +export { JobModel }; +export default JobModel; diff --git a/server/src/domain/jobs/job.repository.interface.ts b/server/src/domain/jobs/job.repository.interface.ts new file mode 100644 index 000000000..044ea709e --- /dev/null +++ b/server/src/domain/jobs/job.repository.interface.ts @@ -0,0 +1,55 @@ +import { Job, JobType } from "@/domain/jobs/job.type.js"; + +export type JobPageQuery = { + page?: number; + rowsPerPage?: number; +}; +export type JobPage = { + jobs: Job[]; + count: number; +}; + +export interface IJobsRepository { + // ******************** + // Claims + // ******************** + + // Atomically claim a job + claimDue(type: JobType, now: number): Promise; + + // Completes a repeating job + recordSuccess(id: string, nextScheduledAt: number, intervalMs: number, now: number): Promise; + + // Record failure, release lease, bump fail count, record reason, reset nextScheduledAt for retry + recordFailure(id: string, error: unknown, now: number): Promise; + + // ******************** + // Hand off to evaluator // This is accomplished by creating an "evaluation" type job + // ******************** + upsertEvaluate(monitorId: string, now: number): Promise; + + // ******************** + // Job CRUD + // ******************** + + // Create jobs + upsertJob(job: Job): Promise; + + // Pause/Resume + setActiveById(refId: string, isActive: boolean): Promise; + + // Edit job schedule + updateScheduleById(refId: string, type: JobType, intervalMs: number | null): Promise; + + // Delete, drop all rows + deleteById(refId: string): Promise; + + // Get job + findById(refId: string): Promise; + + // ******************** + // Observability + // ******************** + findPage(pagination: JobPageQuery): Promise; + findAll(): Promise; +} diff --git a/server/src/domain/jobs/job.repository.mongo.ts b/server/src/domain/jobs/job.repository.mongo.ts new file mode 100644 index 000000000..1e0fb9d35 --- /dev/null +++ b/server/src/domain/jobs/job.repository.mongo.ts @@ -0,0 +1,194 @@ +import { IJobsRepository, JobPageQuery } from "@/domain/jobs/job.repository.interface.js"; +import JobModel, { JobDocument } from "@/domain/jobs/job.model.js"; +import { JobType, Job, BACKOFF_MS, LOCK_MS } from "@/domain/jobs/job.type.js"; +import { hostname } from "node:os"; +import { randomUUID } from "node:crypto"; + +class MongoJobsRepository implements IJobsRepository { + private readonly workerId = `${hostname()}:${process.pid}:${randomUUID()}`; + + private toEntity(doc: JobDocument): Job { + return { + id: doc._id, + type: doc.type, + refId: doc.refId, + isActive: doc.isActive, + nextScheduledAt: doc.nextScheduledAt, + intervalMs: doc.intervalMs, + lockedBy: doc.lockedBy, + lockedUntil: doc.lockedUntil, + runCount: doc.runCount, + failCount: doc.failCount, + lastFinishedAt: doc.lastFinishedAt, + lastFailReason: doc.lastFailReason, + }; + } + + claimDue = async (type: JobType, now: number) => { + const doc = await JobModel.findOneAndUpdate( + { + type, + isActive: true, + nextScheduledAt: { $lte: now }, // Find due jobs + $or: [ + { lockedUntil: null }, // Not locked + { lockedUntil: { $lt: now } }, // Free or expired lock + ], + }, + { + $set: { + lockedBy: this.workerId, + lockedUntil: now + LOCK_MS, + }, + }, + { + new: true, + sort: { nextScheduledAt: 1 }, // Get the oldest due job + } + ).lean(); + return doc ? this.toEntity(doc) : null; + }; + + recordSuccess = async (id: string, scheduledRunAt: number, intervalMs: number, now: number) => { + // Fixed rate job scheduling + + let nextScheduledAt = scheduledRunAt + intervalMs; + + // If the job fell behind, skipped missed runs + if (nextScheduledAt <= now) { + nextScheduledAt = now + intervalMs; + } + + const res = await JobModel.updateOne( + { + _id: id, + }, + { + $set: { + nextScheduledAt, + lockedBy: null, // Remove lock + lockedUntil: null, + lastFinishedAt: now, + }, + $inc: { + runCount: 1, // Increment run count + }, + } + ); + return res.modifiedCount === 1; + }; + + recordFailure = async (id: string, error: unknown, now: number) => { + const res = await JobModel.updateOne( + { + _id: id, + }, + { + $set: { + nextScheduledAt: now + BACKOFF_MS, + lockedBy: null, // Remove lock + lockedUntil: null, + lastFailReason: error instanceof Error ? error.message : String(error), + }, + $inc: { + failCount: 1, // Increment fail count + }, + } + ); + return res.modifiedCount === 1; + }; + + upsertEvaluate = async (monitorId: string, now: number) => { + const filter = { type: "evaluate", refId: monitorId }; + const update = { + $setOnInsert: { + _id: `evaluate:${monitorId}`, + // refId and Type inferred from filter + isActive: true, + intervalMs: null, + lockedBy: null, + lockedUntil: null, + runCount: 0, + failCount: 0, + lastFinishedAt: null, + lastFailReason: null, + }, + $min: { + nextScheduledAt: now, // no op if there is already a pending evaluation that is older + }, + }; + try { + const res = await JobModel.updateOne(filter, update, { upsert: true }); + return res.acknowledged; + } catch (error) { + // Two callers can race the first insert for the same monitor, so the + // loser hits a duplicate _id. Retry once, the row now exists, so this becomes the $min. + if (error && typeof error === "object" && (error as { code?: number }).code === 11000) { + const res = await JobModel.updateOne(filter, update, { upsert: true }); + return res.acknowledged; + } + throw error; + } + }; + upsertJob = async (job: Job) => { + const res = await JobModel.updateOne( + { _id: job.id }, + { + $set: { + // These fields are owned by the monitor, always safe to set + type: job.type, + refId: job.refId, + isActive: job.isActive, + intervalMs: job.intervalMs, + }, + $setOnInsert: { + // These fields are scheduling related, only set on insert so they aren't clobbered + nextScheduledAt: job.nextScheduledAt, + lockedBy: null, + lockedUntil: null, + runCount: 0, + failCount: 0, + lastFinishedAt: null, + lastFailReason: null, + }, + }, + { upsert: true } + ); + return res.acknowledged; + }; + + setActiveById = async (refId: string, isActive: boolean) => { + const res = await JobModel.updateMany({ refId }, { $set: { isActive } }); // inverts check and geo check rows together + return res.modifiedCount > 0; + }; + + updateScheduleById = async (refId: string, type: JobType, intervalMs: number | null) => { + const res = await JobModel.updateOne({ refId, type }, { $set: { intervalMs } }); + return res.modifiedCount === 1; + }; + + deleteById = async (refId: string) => { + const res = await JobModel.deleteMany({ refId }); // Delete all jobs for monitor + return res.deletedCount > 0; + }; + + findById = async (refId: string) => { + const docs = await JobModel.find({ refId }).lean(); + return docs.map(this.toEntity); + }; + + findPage = async ({ page = 0, rowsPerPage = 0 }: JobPageQuery) => { + const count = await JobModel.countDocuments(); + const query = JobModel.find().sort({ nextScheduledAt: 1 }); + if (rowsPerPage > 0) query.skip(Math.max(page, 0) * rowsPerPage).limit(rowsPerPage); + const docs = await query.lean(); + return { jobs: docs.map(this.toEntity), count }; + }; + + findAll = async () => { + const docs = await JobModel.find().lean(); + return docs.map(this.toEntity); + }; +} + +export default MongoJobsRepository; diff --git a/server/src/domain/jobs/job.type.ts b/server/src/domain/jobs/job.type.ts new file mode 100644 index 000000000..1a9098160 --- /dev/null +++ b/server/src/domain/jobs/job.type.ts @@ -0,0 +1,25 @@ +export const JOB_TYPES = ["check", "geo-check", "evaluate", "cleanup-orphaned", "cleanup-retention"] as const; +export type JobType = (typeof JOB_TYPES)[number]; +export const LOCK_MS = 30_000; +export const BACKOFF_MS = 5_000; + +export type Job = { + id: string; + type: JobType; + refId: string | null; + isActive: boolean; + + // Scheduling fields + nextScheduledAt: number; // epoch ms; due when nextScheduledAt <= now + intervalMs: number | null; // repeat period in ms; null = one time job + + // Lock fields + lockedBy: string | null; // workerId holding the lease + lockedUntil: number | null; // epoch ms; lease expiry => reclaimable after a crash + + // observability (replaces what getMetrics/getJobs) + runCount: number; + failCount: number; + lastFinishedAt: number | null; + lastFailReason: string | null; +};