queue and buffer refactor

server/src/service/infrastructure/bufferService.ts

@@ -0,0 +1,192 @@
+import { config } from "@/config/index.js";
+import type { Check } from "@/types/index.js";
+
+const SERVICE_NAME = "BufferService";
+
+class BufferService {
+	static SERVICE_NAME = SERVICE_NAME;
+	private BUFFER_TIMEOUT: number;
+	private db: any;
+	private logger: any;
+	private incidentService: any;
+	private SERVICE_NAME: string;
+	private buffer: any[];
+	private incidentBuffer: any[];
+	private bufferTimer: NodeJS.Timeout;
+	private checksService: any;
+
+	constructor({
+		db,
+		logger,
+		envSettings,
+		incidentService,
+		checkService,
+	}: {
+		db: any;
+		logger: any;
+		envSettings: any;
+		incidentService: any;
+		checkService: any;
+	}) {
+		this.BUFFER_TIMEOUT = config.NODE_ENV === "development" ? 10 : 1000 * 60 * 1; // 1 minute
+		this.db = db;
+		this.logger = logger;
+		this.incidentService = incidentService;
+		this.checksService = checkService;
+		this.SERVICE_NAME = SERVICE_NAME;
+		this.buffer = [];
+		this.incidentBuffer = [];
+		this.scheduleNextFlush();
+		this.logger.info({
+			message: `Buffer service initialized, flushing every ${this.BUFFER_TIMEOUT / 1000}s`,
+			service: this.SERVICE_NAME,
+			method: "constructor",
+		});
+	}
+
+	get serviceName() {
+		return BufferService.SERVICE_NAME;
+	}
+
+	addToBuffer({ check }: { check: Check }) {
+		try {
+			this.buffer.push(check);
+		} catch (error: any) {
+			this.logger.error({
+				message: error.message,
+				service: this.SERVICE_NAME,
+				method: "addToBuffer",
+				stack: error.stack,
+			});
+		}
+	}
+
+	addIncidentToBuffer({ monitor, check, action = "create" }: { monitor: any; check: Check; action?: string }) {
+		try {
+			if (!monitor || !check) {
+				this.logger.warn({
+					message: "Skipping incident buffer item: missing monitor or check",
+					service: this.SERVICE_NAME,
+					method: "addIncidentToBuffer",
+				});
+				return;
+			}
+
+			this.incidentBuffer.push({ monitor, check, action });
+		} catch (error: any) {
+			this.logger.error({
+				message: error.message,
+				service: this.SERVICE_NAME,
+				method: "addIncidentToBuffer",
+				stack: error.stack,
+			});
+		}
+	}
+
+	removeCheckFromBuffer(checkToRemove: Check) {
+		try {
+			if (!checkToRemove) {
+				return false;
+			}
+
+			const index = this.buffer.findIndex((check) => {
+				if (checkToRemove.id && check.id) {
+					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.status === checkToRemove.status &&
+					check.statusCode === checkToRemove.statusCode &&
+					check.responseTime === checkToRemove.responseTime &&
+					check.message === checkToRemove.message
+				);
+			});
+
+			if (index !== -1) {
+				this.buffer.splice(index, 1);
+				return true;
+			}
+
+			return false;
+		} catch (error: any) {
+			this.logger.error({
+				message: error.message,
+				service: this.SERVICE_NAME,
+				method: "removeCheckFromBuffer",
+				stack: error.stack,
+			});
+			return false;
+		}
+	}
+
+	scheduleNextFlush() {
+		this.bufferTimer = setTimeout(async () => {
+			try {
+				await this.flushBuffer();
+			} catch (error: any) {
+				this.logger.error({
+					message: `Error in flush cycle: ${error.message}`,
+					service: this.SERVICE_NAME,
+					method: "scheduleNextFlush",
+					stack: error.stack,
+				});
+			} finally {
+				// Schedule the next flush only after the current one completes
+				this.scheduleNextFlush();
+			}
+		}, this.BUFFER_TIMEOUT);
+	}
+	async flushBuffer() {
+		try {
+			if (this.buffer.length > 0) {
+				await this.checksService.createChecks(this.buffer);
+			}
+		} catch (error: any) {
+			this.logger.error({
+				message: error.message,
+				service: this.SERVICE_NAME,
+				method: "flushBuffer",
+				stack: error.stack,
+			});
+		}
+
+		try {
+			if (this.incidentBuffer.length > 0 && this.incidentService) {
+				await this.flushIncidentBuffer();
+			}
+		} catch (error: any) {
+			this.logger.error({
+				message: error.message,
+				service: this.SERVICE_NAME,
+				method: "flushBuffer",
+				stack: error.stack,
+			});
+		}
+
+		this.buffer = [];
+		this.incidentBuffer = [];
+	}
+
+	async flushIncidentBuffer() {
+		if (!this.incidentService || this.incidentBuffer.length === 0) {
+			return;
+		}
+
+		try {
+			const itemsToProcess = [...this.incidentBuffer];
+			await this.incidentService.processIncidentsFromBuffer(itemsToProcess);
+		} catch (error: any) {
+			this.logger.error({
+				message: `Error flushing incident buffer: ${error.message}`,
+				service: this.SERVICE_NAME,
+				method: "flushIncidentBuffer",
+				stack: error.stack,
+			});
+			throw error;
+		}
+	}
+}
+
+export default BufferService;