Checkmate/server/src/service/bufferService.ts

import type { Check } from "@/domain/checks/check.type.js";
import type { GeoCheck } from "@/domain/geo-checks/geo-check.type.js";
import type { IGeoChecksService } from "../domain/geo-checks/geo-check.service.js";
import type { ILogger } from "@/utils/logger.js";
import type { ISettingsService } from "@/domain/app-settings/app-settings.service.js";
import type { IJobsRepository } from "@/domain/jobs/job.repository.interface.js";
import { ICheckService } from "../domain/checks/check.service.js";
import { DockerLog } from "@/domain/docker/docker-log.type.js";
import { IDockerLogsService } from "@/domain/docker/docker-log.service.js";
import { PendingCheck } from "@/domain/jobs/job.type.js";
const SERVICE_NAME = "BufferService";

export interface IBufferService {
	addToBuffer(check: Check): void;
	addGeoCheckToBuffer(geoCheck: GeoCheck): void;
	addDockerLogToBuffer(dockerLog: DockerLog): void;
	ingestChecks(checks: Check[]): Promise<void>;
	scheduleNextFlush(): void;
	flushBuffer(): Promise<void>;
	flushGeoBuffer(): Promise<void>;
	flushDockerLogsBuffer(): Promise<void>;
	shutdown(): Promise<void>;
}

export class BufferService implements IBufferService {
	static SERVICE_NAME = SERVICE_NAME;
	private BUFFER_TIMEOUT: number;
	private logger: ILogger;
	private SERVICE_NAME: string;
	private buffer: Check[];
	private geoBuffer: GeoCheck[];
	private dockerLogBuffer: DockerLog[];
	private bufferTimer: NodeJS.Timeout | null = null;
	private checksService: ICheckService;
	private geoChecksService: IGeoChecksService;
	private dockerLogsService: IDockerLogsService;
	private jobsRepository: IJobsRepository;

	// Checks that are stored but whose evaluate row could not be armed. Retried on the next flush so they are never unevaluated.
	private unarmed: Map<string, PendingCheck[]> = new Map();

	constructor(
		logger: ILogger,
		checkService: ICheckService,
		geoChecksService: IGeoChecksService,
		dockerLogsService: IDockerLogsService,
		settingsService: ISettingsService,
		jobsRepository: IJobsRepository
	) {
		this.BUFFER_TIMEOUT = settingsService.getSettings().nodeEnv === "development" ? 1000 : 1000 * 60 * 1; // 1 minute
		this.logger = logger;
		this.checksService = checkService;
		this.geoChecksService = geoChecksService;
		this.dockerLogsService = dockerLogsService;
		this.jobsRepository = jobsRepository;
		this.SERVICE_NAME = SERVICE_NAME;
		this.buffer = [];
		this.geoBuffer = [];
		this.dockerLogBuffer = [];
		this.scheduleNextFlush();
		this.logger.info({
			message: `Buffer service initialized, flushing every ${this.BUFFER_TIMEOUT / 1000}s`,
			service: this.SERVICE_NAME,
			method: "constructor",
		});
	}

	addToBuffer(check: Check) {
		try {
			this.buffer.push(check);
		} catch (error: unknown) {
			this.logger.error({
				message: error instanceof Error ? error.message : "Unknown error",
				service: this.SERVICE_NAME,
				method: "addToBuffer",
				stack: error instanceof Error ? error.stack : undefined,
			});
		}
	}

	addGeoCheckToBuffer(geoCheck: GeoCheck) {
		try {
			this.geoBuffer.push(geoCheck);
		} catch (error: unknown) {
			this.logger.error({
				message: error instanceof Error ? error.message : "Unknown error",
				service: this.SERVICE_NAME,
				method: "addGeoCheckToBuffer",
				stack: error instanceof Error ? error.stack : undefined,
			});
		}
	}

	addDockerLogToBuffer(dockerLog: DockerLog): void {
		try {
			this.dockerLogBuffer.push(dockerLog);
		} catch (error: unknown) {
			this.logger.error({
				message: error instanceof Error ? error.message : "Unknown error",
				service: this.SERVICE_NAME,
				method: "addDockerLogToBuffer",
				stack: error instanceof Error ? error.stack : undefined,
			});
		}
	}

	ingestChecks = async (checks: Check[]) => {
		if (checks.length > 0) await this.checksService.createChecks(checks);

		// Group by monitor, starting from any entries a previous flush failed to arm
		const byMonitor = this.unarmed;
		this.unarmed = new Map();
		for (const check of checks) {
			const pending = byMonitor.get(check.metadata.monitorId) ?? [];
			pending.push({ checkId: check.id, createdAt: new Date(check.createdAt).getTime() });
			byMonitor.set(check.metadata.monitorId, pending);
		}
		if (byMonitor.size === 0) return;

		const now = Date.now();
		const ops: Promise<boolean>[] = [];
		for (const [monitorId, pendingChecks] of byMonitor) {
			ops.push(
				this.jobsRepository.upsertEvaluate(monitorId, pendingChecks, now).catch((error: unknown) => {
					// The checks are already stored, so a failed arm is retried on the next flush rather than dropped
					this.unarmed.set(monitorId, pendingChecks);
					this.logger.error({
						message: `Could not arm evaluation for monitor ${monitorId}, retrying on next flush: ${error instanceof Error ? error.message : String(error)}`,
						service: this.SERVICE_NAME,
						method: "ingestChecks",
					});
					return false;
				})
			);
		}
		await Promise.all(ops);
	};

	scheduleNextFlush() {
		if (this.bufferTimer) {
			clearTimeout(this.bufferTimer);
		}
		this.bufferTimer = setTimeout(async () => {
			try {
				await this.flushBuffer();
				await this.flushGeoBuffer();
				await this.flushDockerLogsBuffer();
			} catch (error: unknown) {
				this.logger.error({
					message: error instanceof Error ? error.message : "Unknown error",
					service: this.SERVICE_NAME,
					method: "scheduleNextFlush",
					stack: error instanceof Error ? error.stack : undefined,
				});
			} finally {
				// Schedule the next flush only after the current one completes
				this.scheduleNextFlush();
			}
		}, this.BUFFER_TIMEOUT);
	}

	async flushBuffer() {
		if (this.buffer.length === 0 && this.unarmed.size === 0) {
			return;
		}
		// Take the batch first so a write that drains it can't be appended to mid-flush
		const batch = this.buffer;
		this.buffer = [];
		try {
			this.logger.debug({
				message: `Flushing ${batch.length} checks to database`,
				service: this.SERVICE_NAME,
				method: "flushBuffer",
			});
			await this.ingestChecks(batch);
		} catch (error: unknown) {
			this.logger.error({
				message: error instanceof Error ? error.message : "Unknown error",
				service: this.SERVICE_NAME,
				method: "flushBuffer",
				stack: error instanceof Error ? error.stack : undefined,
			});
			// Batch is already cleared from the buffer; dropping it prevents infinite retry loops
		}
	}

	async flushGeoBuffer() {
		try {
			if (this.geoBuffer.length > 0) {
				this.logger.debug({
					message: `Flushing ${this.geoBuffer.length} geo checks to database`,
					service: this.SERVICE_NAME,
					method: "flushGeoBuffer",
				});
				await this.geoChecksService.createGeoChecks(this.geoBuffer);
				this.geoBuffer = [];
			}
		} catch (error: unknown) {
			this.logger.error({
				message: error instanceof Error ? error.message : "Unknown error",
				service: this.SERVICE_NAME,
				method: "flushGeoBuffer",
				stack: error instanceof Error ? error.stack : undefined,
			});
			// Clear buffer even on error to prevent infinite retry loops
			this.geoBuffer = [];
		}
	}

	async flushDockerLogsBuffer() {
		if (this.dockerLogBuffer.length === 0) {
			return;
		}
		// Take the batch first so a write that drains it can't be appended to mid-flush
		const batch = this.dockerLogBuffer;
		this.dockerLogBuffer = [];
		try {
			this.logger.debug({
				message: `Flushing ${batch.length} docker logs to database`,
				service: this.SERVICE_NAME,
				method: "flushDockerLogsBuffer",
			});
			await this.dockerLogsService.createDockerLogs(batch);
		} catch (error: unknown) {
			this.logger.error({
				message: error instanceof Error ? error.message : "Unknown error",
				service: this.SERVICE_NAME,
				method: "flushDockerLogsBuffer",
				stack: error instanceof Error ? error.stack : undefined,
			});
		}
	}

	async shutdown() {
		if (this.bufferTimer) {
			clearTimeout(this.bufferTimer);
			this.bufferTimer = null;
		}
		await this.flushBuffer();
		await this.flushGeoBuffer();
		await this.flushDockerLogsBuffer();
	}
}