Merge pull request #3295 from bluewave-labs/feat/cleanup-job

feat: cleanup job

server/src/config/services.ts

@@ -202,6 +202,10 @@ export const initializeServices = async ({
 		incidentService,
 		maintenanceWindowsRepository,
 		monitorsRepository,
+		teamsRepository,
+		monitorStatsRepository,
+		checksRepository,
+		incidentsRepository,
 	});
 
 	const superSimpleQueue = await SuperSimpleQueue.create({

server/src/repositories/checks/IChecksRepository.ts

@@ -39,4 +39,5 @@ export interface IChecksRepository {
 	//delete
 	deleteByMonitorId(monitorId: string): Promise<number>;
 	deleteByTeamId(teamId: string): Promise<number>;
+	deleteByMonitorIdsNotIn(monitorIds: string[]): Promise<number>;
 }

server/src/repositories/checks/MongoChecksRepistory.ts

@@ -13,7 +13,7 @@ import type {
 	GotTimings,
 	MonitorType,
 } from "@/types/index.js";
-import { CheckModel, type CheckDocument } from "@/db/models/index.js";
+import { CheckModel, MonitorModel, type CheckDocument } from "@/db/models/index.js";
 import mongoose from "mongoose";
 
 const SERVICE_NAME = "StatusService";
@@ -210,19 +210,16 @@ class MongoChecksRepository implements IChecksRepository {
 	private toDocument = (check: Partial<Check>): CheckDocument => {
 		// Map id to _id for MongoDB storage
 		const { id, metadata, ...rest } = check;
+		if (!metadata || !metadata.monitorId || !metadata.teamId) {
+			throw new Error(`Check must have valid metadata with monitorId and teamId. Got: ${JSON.stringify({ id, metadata })}`);
+		}
 		return {
 			_id: id ? new mongoose.Types.ObjectId(id) : new mongoose.Types.ObjectId(),
-			metadata: metadata
-				? {
-						monitorId: new mongoose.Types.ObjectId(metadata.monitorId),
-						teamId: new mongoose.Types.ObjectId(metadata.teamId),
-						type: metadata.type,
-					}
-				: {
-						monitorId: new mongoose.Types.ObjectId(),
-						teamId: new mongoose.Types.ObjectId(),
-						type: "http",
-					},
+			metadata: {
+				monitorId: new mongoose.Types.ObjectId(metadata.monitorId),
+				teamId: new mongoose.Types.ObjectId(metadata.teamId),
+				type: metadata.type,
+			},
 			...rest,
 		} as unknown as CheckDocument;
 	};
@@ -417,6 +414,12 @@ class MongoChecksRepository implements IChecksRepository {
 		return deleteResult.deletedCount;
 	};
 
+	deleteByMonitorIdsNotIn = async (monitorIds: string[]): Promise<number> => {
+		const objectIds = monitorIds.map((id) => new mongoose.Types.ObjectId(id));
+		const result = await CheckModel.deleteMany({ "metadata.monitorId": { $nin: objectIds } });
+		return result.deletedCount ?? 0;
+	};
+
 	private findUptimeDateRangeChecks = async (
 		monitorType: Exclude<MonitorType, "hardware" | "pagespeed">,
 		monitorObjectId: mongoose.Types.ObjectId,

server/src/repositories/incidents/IIncidentsRepository.ts

@@ -24,5 +24,6 @@ export interface IIncidentsRepository {
 	updateById(incidentId: string, teamId: string, updateData: Partial<Incident>): Promise<Incident>;
 	// delete
 	deleteByMonitorId(monitorId: string, teamId: string): Promise<number>;
+	deleteByMonitorIdsNotIn(monitorIds: string[]): Promise<number>;
 	// other
 }

server/src/repositories/incidents/MongoIncidentRepository.ts

@@ -281,5 +281,11 @@ class MongoIncidentRepository implements IIncidentsRepository {
 		});
 		return result.deletedCount || 0;
 	};
+
+	deleteByMonitorIdsNotIn = async (monitorIds: string[]): Promise<number> => {
+		const objectIds = monitorIds.map((id) => new mongoose.Types.ObjectId(id));
+		const result = await IncidentModel.deleteMany({ monitorId: { $nin: objectIds } });
+		return result.deletedCount ?? 0;
+	};
 }
 export default MongoIncidentRepository;

server/src/repositories/monitor-stats/IMonitorStatsRepository.ts

@@ -7,5 +7,7 @@ export interface IMonitorStatsRepository {
 	// update
 	// delete
 	deleteByMonitorId(monitorId: string): Promise<MonitorStats>;
+	deleteByMonitorIds(monitorIds: string[]): Promise<number>;
+	deleteByMonitorIdsNotIn(monitorIds: string[]): Promise<number>;
 	// other
 }

server/src/repositories/monitor-stats/MongoMonitorStatsRepository.ts

@@ -52,6 +52,18 @@ class MongoMonitorStatsRepository implements IMonitorStatsRepository {
 		}
 		return this.toEntity(deleted);
 	};
+
+	deleteByMonitorIds = async (monitorIds: string[]): Promise<number> => {
+		const objectIds = monitorIds.map((id) => new mongoose.Types.ObjectId(id));
+		const result = await MonitorStatsModel.deleteMany({ monitorId: { $in: objectIds } });
+		return result.deletedCount ?? 0;
+	};
+
+	deleteByMonitorIdsNotIn = async (monitorIds: string[]): Promise<number> => {
+		const objectIds = monitorIds.map((id) => new mongoose.Types.ObjectId(id));
+		const result = await MonitorStatsModel.deleteMany({ monitorId: { $nin: objectIds } });
+		return result.deletedCount ?? 0;
+	};
 }
 
 export default MongoMonitorStatsRepository;

server/src/repositories/monitors/IMonitorsRepository.ts

@@ -41,4 +41,6 @@ export interface IMonitorsRepository {
 	findMonitorsSummaryByTeamId(teamId: string, config?: SummaryConfig): Promise<MonitorsSummary>;
 	findGroupsByTeamId(teamId: string): Promise<string[]>;
 	removeNotificationFromMonitors(notificationId: string): Promise<void>;
+	deleteByTeamIdsNotIn(teamIds: string[]): Promise<number>;
+	findAllMonitorIds(): Promise<string[]>;
 }

server/src/repositories/monitors/MongoMonitorsRepository.ts

@@ -431,6 +431,17 @@ class MongoMonitorsRepository implements IMonitorsRepository {
 			createdAt: toDateString(doc.createdAt),
 		};
 	};
+
+	deleteByTeamIdsNotIn = async (teamIds: string[]): Promise<number> => {
+		const objectIds = teamIds.map((id) => new mongoose.Types.ObjectId(id));
+		const result = await MonitorModel.deleteMany({ teamId: { $nin: objectIds } });
+		return result.deletedCount ?? 0;
+	};
+
+	findAllMonitorIds = async (): Promise<string[]> => {
+		const monitors = await MonitorModel.find({}, { _id: 1 }).lean();
+		return monitors.map((doc) => doc._id.toString());
+	};
 }
 
 export default MongoMonitorsRepository;

server/src/repositories/teams/ITeamsRepository.ts

@@ -6,4 +6,5 @@ export interface ITeamsRepository {
 	// update
 	// delete
 	// other
+	findAllTeamIds(): Promise<string[]>;
 }

server/src/repositories/teams/MongoTeamsRepository.ts

@@ -31,6 +31,11 @@ class MongoTeamsRepository implements ITeamsRepository {
 		const team = await TeamModel.create({ email });
 		return this.toEntity(team);
 	};
+
+	findAllTeamIds = async (): Promise<string[]> => {
+		const teams = await TeamModel.find({}, { _id: 1 }).lean();
+		return teams.map((team) => this.toStringId(team._id));
+	};
 }
 
 export default MongoTeamsRepository;

server/src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueue.ts

@@ -97,6 +97,7 @@ class SuperSimpleQueue implements ISuperSimpleQueue {
 			this.scheduler.start();
 
 			this.scheduler.addTemplate("monitor-job", this.helper.getMonitorJob());
+			this.scheduler.addTemplate("cleanup-orphaned", this.helper.getCleanupOrphanedJob());
 			const monitors = await this.monitorsRepository.findAll();
 			if (!monitors) {
 				return true;
@@ -108,6 +109,8 @@ class SuperSimpleQueue implements ISuperSimpleQueue {
 				}, randomOffset);
 			}
 
+			this.scheduler.addJob({ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true });
+
 			return true;
 		} catch (error) {
 			this.logger.error({

server/src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.ts

@@ -4,7 +4,14 @@ import { AppError } from "@/utils/AppError.js";
 import { INetworkService, INotificationsService, IStatusService } from "@/service/index.js";
 import type { StatusChangeResult, MonitorStatusResponse, HardwareStatusPayload, MonitorStatus } from "@/types/index.js";
 import IncidentService from "@/service/business/incidentService.js";
-import { IMaintenanceWindowsRepository, IMonitorsRepository } from "@/repositories/index.js";
+import {
+	IMaintenanceWindowsRepository,
+	IMonitorsRepository,
+	ITeamsRepository,
+	IMonitorStatsRepository,
+	IChecksRepository,
+	IIncidentsRepository,
+} from "@/repositories/index.js";
 
 export interface MonitorActionDecision {
 	shouldCreateIncident: boolean;
@@ -32,6 +39,10 @@ class SuperSimpleQueueHelper {
 	private incidentService: IncidentService;
 	private maintenanceWindowsRepository: IMaintenanceWindowsRepository;
 	private monitorsRepository: IMonitorsRepository;
+	private teamsRepository: ITeamsRepository;
+	private monitorStatsRepository: IMonitorStatsRepository;
+	private checksRepository: IChecksRepository;
+	private incidentsRepository: IIncidentsRepository;
 
 	constructor({
 		logger,
@@ -43,6 +54,10 @@ class SuperSimpleQueueHelper {
 		incidentService,
 		maintenanceWindowsRepository,
 		monitorsRepository,
+		teamsRepository,
+		monitorStatsRepository,
+		checksRepository,
+		incidentsRepository,
 	}: {
 		logger: any;
 		networkService: INetworkService;
@@ -53,6 +68,10 @@ class SuperSimpleQueueHelper {
 		incidentService: IncidentService;
 		maintenanceWindowsRepository: IMaintenanceWindowsRepository;
 		monitorsRepository: IMonitorsRepository;
+		teamsRepository: ITeamsRepository;
+		monitorStatsRepository: IMonitorStatsRepository;
+		checksRepository: IChecksRepository;
+		incidentsRepository: IIncidentsRepository;
 	}) {
 		this.logger = logger;
 		this.networkService = networkService;
@@ -63,6 +82,10 @@ class SuperSimpleQueueHelper {
 		this.incidentService = incidentService;
 		this.maintenanceWindowsRepository = maintenanceWindowsRepository;
 		this.monitorsRepository = monitorsRepository;
+		this.teamsRepository = teamsRepository;
+		this.monitorStatsRepository = monitorStatsRepository;
+		this.checksRepository = checksRepository;
+		this.incidentsRepository = incidentsRepository;
 	}
 
 	get serviceName() {
@@ -111,7 +134,7 @@ class SuperSimpleQueueHelper {
 					return;
 				}
 				// Step 4 Add check to buffer
-				this.buffer.addToBuffer({ check });
+				this.buffer.addToBuffer(check);
 				// Step 4.  Update monitor status
 				const statusChangeResult = await this.statusService.updateMonitorStatus(status, check);
 
@@ -156,10 +179,72 @@ class SuperSimpleQueueHelper {
 	getCleanupOrphanedJob = () => {
 		return async () => {
 			try {
-				// Remove orphaned monitors
-				// remove orphaned monitorStats
+				this.logger.info({
+					message: "Starting cleanup of orphaned data",
+					service: SERVICE_NAME,
+					method: "getCleanupOrphanedJob",
+				});
+
+				// Get all valid team IDs
+				const validTeamIds = await this.teamsRepository.findAllTeamIds();
+				this.logger.debug({
+					message: `Found ${validTeamIds.length} valid teams`,
+					service: SERVICE_NAME,
+					method: "getCleanupOrphanedJob",
+				});
+
+				// Remove orphaned monitors (monitors without a valid team)
+				const deletedMonitorCount = await this.monitorsRepository.deleteByTeamIdsNotIn(validTeamIds);
+				if (deletedMonitorCount > 0) {
+					this.logger.info({
+						message: `Deleted ${deletedMonitorCount} orphaned monitors`,
+						service: SERVICE_NAME,
+						method: "getCleanupOrphanedJob",
+					});
+				}
+
+				// Remove orphaned monitorStats (stats without a valid monitor)
+				const allMonitorIds = await this.monitorsRepository.findAllMonitorIds();
+				this.logger.debug({
+					message: `Found ${allMonitorIds.length} valid monitors`,
+					service: SERVICE_NAME,
+					method: "getCleanupOrphanedJob",
+				});
+
+				const deletedStatsCount = await this.monitorStatsRepository.deleteByMonitorIdsNotIn(allMonitorIds);
+				if (deletedStatsCount > 0) {
+					this.logger.info({
+						message: `Deleted ${deletedStatsCount} orphaned monitor stats`,
+						service: SERVICE_NAME,
+						method: "getCleanupOrphanedJob",
+					});
+				}
+
 				// Remove orphaned checks
+				const deletedChecksCount = await this.checksRepository.deleteByMonitorIdsNotIn(allMonitorIds);
+				if (deletedChecksCount > 0) {
+					this.logger.info({
+						message: `Deleted ${deletedChecksCount} orphaned checks`,
+						service: SERVICE_NAME,
+						method: "getCleanupOrphanedJob",
+					});
+				}
+
 				// Remove orphaned incidents
+				const deletedIncidentsCount = await this.incidentsRepository.deleteByMonitorIdsNotIn(allMonitorIds);
+				if (deletedIncidentsCount > 0) {
+					this.logger.info({
+						message: `Deleted ${deletedIncidentsCount} orphaned incidents`,
+						service: SERVICE_NAME,
+						method: "getCleanupOrphanedJob",
+					});
+				}
+
+				this.logger.info({
+					message: "Cleanup of orphaned data completed",
+					service: SERVICE_NAME,
+					method: "getCleanupOrphanedJob",
+				});
 			} catch (error: any) {
 				this.logger.warn({
 					message: error.message,

server/src/service/infrastructure/bufferService.ts

@@ -60,9 +60,9 @@ class BufferService implements IBufferService {
 					return check.id.toString() === checkToRemove.id.toString();
 				}
 				return (
-					check.monitorId?.toString() === checkToRemove.metadata.monitorId &&
-					check.teamId?.toString() === checkToRemove.metadata.teamId &&
-					check.type === checkToRemove.metadata.type &&
+					check.metadata.monitorId?.toString() === checkToRemove.metadata.monitorId &&
+					check.metadata.teamId?.toString() === checkToRemove.metadata.teamId &&
+					check.metadata.type === checkToRemove.metadata.type &&
 					check.status === checkToRemove.status &&
 					check.statusCode === checkToRemove.statusCode &&
 					check.responseTime === checkToRemove.responseTime &&

server/src/service/infrastructure/statusService.ts

@@ -362,7 +362,7 @@ export class StatusService implements IStatusService {
 				});
 				return false;
 			}
-			this.buffer.addToBuffer({ check });
+			this.buffer.addToBuffer(check);
 			return true;
 		} catch (error: any) {
 			this.logger.error({