replace findAll with findAllForScheduling

This commit is contained in:
Alex Holliday committed 2026-08-04 15:16:03 -07:00
1 parent 4c8342f379
commit 5fa74bd883
7 files changed
+47 -23

No files matched your search

@@ -1,4 +1,4 @@
import type { MonitorType, Monitor, MonitorStatus, MonitorsSummary, CheckSnapshot } from "@/domain/monitors/monitor.type.js";
import type { MonitorType, Monitor, MonitorStatus, MonitorsSummary, CheckSnapshot, MonitorScheduleFields } from "@/domain/monitors/monitor.type.js";
export interface TeamQueryConfig {
limit?: number;
@@ -26,7 +26,7 @@ export interface IMonitorsRepository {
findByIdLean(monitorId: string): Promise<Monitor | null>;
// collection fetch
findAll(): Promise<Monitor[]>;
findAllForScheduling(): Promise<MonitorScheduleFields[]>;
findByTeamId(teamId: string, config: TeamQueryConfig): Promise<Monitor[]>;
findByTeamIdWithStats(teamId: string, config: TeamQueryConfig): Promise<Monitor[]>;
findByIds(monitorIds: string[]): Promise<Monitor[]>;
@@ -1,7 +1,7 @@
import { MonitorModel } from "@/domain/monitors/monitor.model.js";
import type { MonitorDocument, CheckSnapshotDocument } from "@/domain/monitors/monitor.model.js";
import type { CheckSnapshot } from "@/domain/checks/check.type.js";
import type { Monitor, MonitorStatus, MonitorsSummary } from "@/domain/monitors/monitor.type.js";
import type { Monitor, MonitorScheduleFields, MonitorStatus, MonitorsSummary } from "@/domain/monitors/monitor.type.js";
import mongoose, { type FilterQuery, type PipelineStage } from "mongoose";
import { MongoBulkWriteError } from "mongodb";
import { AppError } from "@/utils/AppError.js";
@@ -41,13 +41,20 @@ class MongoMonitorsRepository implements IMonitorsRepository {
};
findByIdLean = async (monitorId: string): Promise<Monitor | null> => {
const monitor = await MonitorModel.findOne({ _id: monitorId });
const monitor = await MonitorModel.findOne({ _id: monitorId }).select({ recentChecks: 0 });
return monitor ? this.toEntity(monitor) : null;
};
findAll = async (): Promise<Monitor[]> => {
const monitors = await MonitorModel.find();
return this.mapDocuments(monitors);
findAllForScheduling = async (): Promise<MonitorScheduleFields[]> => {
const docs = await MonitorModel.find({}, { type: 1, isActive: 1, interval: 1, geoCheckEnabled: 1, geoCheckInterval: 1 }).lean();
return docs.map((doc) => ({
id: doc._id.toString(),
type: doc.type,
isActive: doc.isActive,
interval: doc.interval,
geoCheckEnabled: doc.geoCheckEnabled ?? false,
geoCheckInterval: doc.geoCheckInterval ?? 300000,
}));
};
private queryBuilder = (config: TeamQueryConfig, teamId: string): FilterQuery<MonitorDocument> => {
@@ -182,7 +189,7 @@ class MongoMonitorsRepository implements IMonitorsRepository {
},
...(statusPatch && { $set: statusPatch }),
},
{ returnDocument: "after" }
{ returnDocument: "after", projection: { recentChecks: 0 } }
);
if (!updatedMonitor) {
@@ -178,3 +178,5 @@ export interface Game {
}
export type GamesMap = Record<string, Game>;
export type MonitorScheduleFields = Pick<Monitor, "id" | "type" | "isActive" | "interval" | "geoCheckEnabled" | "geoCheckInterval">;
+4 -4
View File
@@ -1,7 +1,7 @@
import { IJobScheduler } from "@/worker/worker.interface.js";
import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js";
import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.type.js";
import { Monitor, MonitorScheduleFields, supportsGeoCheck } from "@/domain/monitors/monitor.type.js";
import { type Job, type JobSeed, type JobType, jobId } from "@/domain/jobs/job.type.js";
import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js";
import { WorkerJobsPagination, WorkerJobSummary, WorkerMetrics } from "@/worker/worker.interface.js";
@@ -43,7 +43,7 @@ export class JobScheduler implements IJobScheduler {
"cleanup-retention": POLL_MS,
};
// Helpers, builds job seeds. immediate=true runs on the next tick; otherwise jitter spreads the herd.
protected toCheckJob = (monitor: Monitor, now: number, immediate = false): JobSeed => ({
protected toCheckJob = (monitor: MonitorScheduleFields, now: number, immediate = false): JobSeed => ({
id: jobId("check", monitor.id),
type: "check",
refId: monitor.id,
@@ -52,7 +52,7 @@ export class JobScheduler implements IJobScheduler {
intervalMs: monitor.interval,
});
protected toGeoCheckJob = (monitor: Monitor, now: number, immediate = false): JobSeed => ({
protected toGeoCheckJob = (monitor: MonitorScheduleFields, now: number, immediate = false): JobSeed => ({
...this.toCheckJob(monitor, now, immediate),
id: jobId("geo-check", monitor.id),
type: "geo-check",
@@ -187,7 +187,7 @@ export class JobScheduler implements IJobScheduler {
// Seed queue
protected reconcile = async () => {
const now = Date.now();
const monitors = await this.monitorsRepository.findAll();
const monitors = await this.monitorsRepository.findAllForScheduling();
for (const monitor of monitors) {
await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now));
if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) {
@@ -1,6 +1,6 @@
import type { IMonitorsRepository, TeamQueryConfig, SummaryConfig } from "../../src/domain/monitors/monitor.repository.interface.ts";
import type { CheckSnapshot } from "../../src/domain/checks/check.type.ts";
import type { Monitor, MonitorsSummary } from "../../src/domain/monitors/monitor.type.ts";
import type { Monitor, MonitorScheduleFields, MonitorsSummary } from "../../src/domain/monitors/monitor.type.ts";
export class InMemoryMonitorsRepository implements IMonitorsRepository {
private monitors: Monitor[] = [];
@@ -24,8 +24,15 @@ export class InMemoryMonitorsRepository implements IMonitorsRepository {
return { ...monitor };
}
async findAll(): Promise<Monitor[]> {
return this.monitors.map((m) => ({ ...m }));
async findAllForScheduling(): Promise<MonitorScheduleFields[]> {
return this.monitors.map((m) => ({
id: m.id,
type: m.type,
isActive: m.isActive,
interval: m.interval,
geoCheckEnabled: m.geoCheckEnabled ?? false,
geoCheckInterval: m.geoCheckInterval ?? 300000,
}));
}
async findByTeamId(_teamId: string, _config: TeamQueryConfig): Promise<Monitor[]> {
@@ -84,7 +84,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses
const monitorsRepository = {
findByIds: jest.fn<any>().mockResolvedValue([makeMonitor()]),
findByIdLean: jest.fn<any>().mockResolvedValue(makeMonitor()),
findAll: jest.fn<any>().mockResolvedValue([]),
findAllForScheduling: jest.fn<any>().mockResolvedValue([]),
updateById: jest.fn<any>().mockResolvedValue({}),
};
const checksRepository = {
@@ -183,7 +183,11 @@ describe("DBQueueWorker", () => {
queueMode: "primary",
queuePrimaryProcesses: false, // isolate reconcile from the polling loops
mocks: {
monitorsRepository: { findByIds: jest.fn<any>(), findAll: jest.fn<any>().mockResolvedValue([makeMonitor()]), updateById: jest.fn<any>() },
monitorsRepository: {
findByIds: jest.fn<any>(),
findAllForScheduling: jest.fn<any>().mockResolvedValue([makeMonitor()]),
updateById: jest.fn<any>(),
},
},
});
@@ -200,7 +204,11 @@ describe("DBQueueWorker", () => {
queueMode: "primary",
queuePrimaryProcesses: false,
mocks: {
monitorsRepository: { findByIds: jest.fn<any>(), findAll: jest.fn<any>().mockResolvedValue([geoMonitor]), updateById: jest.fn<any>() },
monitorsRepository: {
findByIds: jest.fn<any>(),
findAllForScheduling: jest.fn<any>().mockResolvedValue([geoMonitor]),
updateById: jest.fn<any>(),
},
},
});
@@ -38,7 +38,7 @@ const createScheduler = (overrides?: { queueMode?: QueueMode; mocks?: Record<str
deleteById: jest.fn<any>().mockResolvedValue(undefined),
};
const monitorsRepository = {
findAll: jest.fn<any>().mockResolvedValue([]),
findAllForScheduling: jest.fn<any>().mockResolvedValue([]),
};
const logger = createMockLogger();
const mocks = { jobsRepository, queueWorkersRepository, monitorsRepository, logger, ...overrides?.mocks };
@@ -73,7 +73,7 @@ describe("JobScheduler", () => {
it("primary mode reconciles: seeds a check row per monitor plus the two cleanup rows", async () => {
const { scheduler, mocks } = createScheduler({
queueMode: "primary",
mocks: { monitorsRepository: { findAll: jest.fn<any>().mockResolvedValue([makeMonitor()]) } },
mocks: { monitorsRepository: { findAllForScheduling: jest.fn<any>().mockResolvedValue([makeMonitor()]) } },
});
await scheduler.init();
@@ -87,7 +87,7 @@ describe("JobScheduler", () => {
it("primary mode also seeds a geo-check row for geo-enabled monitors", async () => {
const { scheduler, mocks } = createScheduler({
queueMode: "primary",
mocks: { monitorsRepository: { findAll: jest.fn<any>().mockResolvedValue([makeMonitor({ geoCheckEnabled: true })]) } },
mocks: { monitorsRepository: { findAllForScheduling: jest.fn<any>().mockResolvedValue([makeMonitor({ geoCheckEnabled: true })]) } },
});
await scheduler.init();
@@ -98,12 +98,12 @@ describe("JobScheduler", () => {
it("worker mode heartbeats but does not reconcile (no monitor reads, no seeds)", async () => {
const { scheduler, mocks } = createScheduler({
queueMode: "worker",
mocks: { monitorsRepository: { findAll: jest.fn<any>().mockResolvedValue([makeMonitor()]) } },
mocks: { monitorsRepository: { findAllForScheduling: jest.fn<any>().mockResolvedValue([makeMonitor()]) } },
});
await scheduler.init();
expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", false);
expect(mocks.monitorsRepository.findAll).not.toHaveBeenCalled();
expect(mocks.monitorsRepository.findAllForScheduling).not.toHaveBeenCalled();
expect(mocks.jobsRepository.upsertJob).not.toHaveBeenCalled();
expect(mocks.jobsRepository.upsertCleanupJob).not.toHaveBeenCalled();
});