Merge pull request #3447 from bluewave-labs/feat/dlq

Feat(DLQ): Add DLQ data layer -> types, model, and repository

server/src/db/models/DLQItem.ts

@@ -0,0 +1,70 @@
+import { Schema, model, type Types } from "mongoose";
+import { DLQItemTypes, DLQItemStatuses, type DLQItem } from "@/types/dlqItem.js";
+
+export interface DLQItemDocument extends Omit<DLQItem, "id" | "monitorId" | "teamId" | "nextRetryAt" | "createdAt" | "updatedAt"> {
+	_id: Types.ObjectId;
+	monitorId: Types.ObjectId;
+	teamId: Types.ObjectId;
+	nextRetryAt: Date;
+	createdAt: Date;
+	updatedAt: Date;
+}
+
+const DLQItemSchema = new Schema<DLQItemDocument>(
+	{
+		type: {
+			type: String,
+			enum: DLQItemTypes,
+			required: true,
+		},
+		status: {
+			type: String,
+			enum: DLQItemStatuses,
+			default: "pending",
+		},
+		payload: {
+			type: Schema.Types.Mixed,
+			required: true,
+		},
+		monitorId: {
+			type: Schema.Types.ObjectId,
+			ref: "Monitor",
+			required: true,
+			immutable: true,
+			index: true,
+		},
+		teamId: {
+			type: Schema.Types.ObjectId,
+			ref: "Team",
+			required: true,
+			immutable: true,
+			index: true,
+		},
+		retryCount: {
+			type: Number,
+			default: 0,
+		},
+		maxRetries: {
+			type: Number,
+			default: 5,
+		},
+		lastError: {
+			type: String,
+			default: "",
+		},
+		nextRetryAt: {
+			type: Date,
+			required: true,
+		},
+	},
+	{ timestamps: true }
+);
+
+DLQItemSchema.index({ status: 1, nextRetryAt: 1 });
+DLQItemSchema.index({ teamId: 1, status: 1 });
+DLQItemSchema.index({ createdAt: 1 });
+
+const DLQItemModel = model<DLQItemDocument>("DLQItem", DLQItemSchema);
+
+export { DLQItemModel };
+export default DLQItemModel;

server/src/db/models/index.ts

@@ -36,3 +36,6 @@ export { default as MaintenanceWindowModel } from "@/db/models/MaintenanceWindow
 
 export * from "@/db/models/GeoCheck.js";
 export { default as GeoCheckModel } from "@/db/models/GeoCheck.js";
+
+export * from "@/db/models/DLQItem.js";
+export { default as DLQItemModel } from "@/db/models/DLQItem.js";

server/src/repositories/dlq/IDLQRepository.ts

@@ -0,0 +1,30 @@
+import type { DLQItem, DLQItemType, DLQItemStatus } from "@/types/index.js";
+
+export interface DLQQueryFilters {
+	type?: DLQItemType;
+	status?: DLQItemStatus;
+	page: number;
+	rowsPerPage: number;
+}
+
+export interface DLQStatusCount {
+	status: DLQItemStatus;
+	type: DLQItemType;
+	count: number;
+}
+
+export interface IDLQRepository {
+	// create
+	create(item: Partial<DLQItem>): Promise<DLQItem>;
+	// fetch
+	findById(id: string): Promise<DLQItem | null>;
+	findByTeamId(teamId: string, filters: DLQQueryFilters): Promise<DLQItem[]>;
+	findRetryable(limit: number): Promise<DLQItem[]>;
+	countByTeamId(teamId: string): Promise<number>;
+	countByTeamIdGrouped(teamId: string): Promise<DLQStatusCount[]>;
+	// update
+	updateById(id: string, patch: Partial<DLQItem>): Promise<DLQItem>;
+	// delete
+	deleteById(id: string, teamId: string): Promise<number>;
+	deleteOlderThan(date: Date): Promise<number>;
+}

server/src/repositories/dlq/MongoDLQRepository.ts

@@ -0,0 +1,139 @@
+import { DLQItemModel } from "@/db/models/index.js";
+import type { DLQItemDocument } from "@/db/models/DLQItem.js";
+import type { DLQItem } from "@/types/index.js";
+import type { IDLQRepository, DLQQueryFilters, DLQStatusCount } from "@/repositories/dlq/IDLQRepository.js";
+import mongoose from "mongoose";
+import { AppError } from "@/utils/AppError.js";
+
+class MongoDLQRepository implements IDLQRepository {
+	private toStringId = (value?: mongoose.Types.ObjectId | string | null): string => {
+		if (!value) {
+			return "";
+		}
+		return value instanceof mongoose.Types.ObjectId ? value.toString() : String(value);
+	};
+
+	private toDateString = (value?: Date | string | null): string => {
+		if (!value) {
+			return new Date(0).toISOString();
+		}
+		return value instanceof Date ? value.toISOString() : new Date(value).toISOString();
+	};
+
+	protected toEntity = (doc: DLQItemDocument): DLQItem => {
+		return {
+			id: this.toStringId(doc._id),
+			type: doc.type,
+			status: doc.status,
+			payload: doc.payload,
+			monitorId: this.toStringId(doc.monitorId),
+			teamId: this.toStringId(doc.teamId),
+			retryCount: doc.retryCount,
+			maxRetries: doc.maxRetries,
+			lastError: doc.lastError,
+			nextRetryAt: this.toDateString(doc.nextRetryAt),
+			createdAt: this.toDateString(doc.createdAt),
+			updatedAt: this.toDateString(doc.updatedAt),
+		};
+	};
+
+	protected mapDocuments = (documents: DLQItemDocument[] | DLQItemDocument | null): DLQItem[] => {
+		if (!documents) {
+			return [];
+		}
+		if (Array.isArray(documents)) {
+			return documents.map((doc) => this.toEntity(doc));
+		}
+		return [this.toEntity(documents)];
+	};
+
+	async create(item: Partial<DLQItem>): Promise<DLQItem> {
+		const newItem = await DLQItemModel.create(item);
+		return this.toEntity(newItem);
+	}
+
+	findById = async (id: string): Promise<DLQItem | null> => {
+		const item = await DLQItemModel.findById(new mongoose.Types.ObjectId(id));
+		return item ? this.toEntity(item) : null;
+	};
+
+	findByTeamId = async (teamId: string, filters: DLQQueryFilters): Promise<DLQItem[]> => {
+		const query: Record<string, unknown> = {
+			teamId: new mongoose.Types.ObjectId(teamId),
+		};
+		if (filters.type) {
+			query.type = filters.type;
+		}
+		if (filters.status) {
+			query.status = filters.status;
+		}
+
+		const items = await DLQItemModel.find(query)
+			.sort({ createdAt: -1 })
+			.skip(filters.page * filters.rowsPerPage)
+			.limit(filters.rowsPerPage);
+		return this.mapDocuments(items);
+	};
+
+	findRetryable = async (limit: number): Promise<DLQItem[]> => {
+		const items = await DLQItemModel.find({
+			status: { $in: ["pending", "retrying"] },
+			nextRetryAt: { $lte: new Date() },
+		})
+			.sort({ nextRetryAt: 1 })
+			.limit(limit);
+		return this.mapDocuments(items);
+	};
+
+	countByTeamId = async (teamId: string): Promise<number> => {
+		return DLQItemModel.countDocuments({
+			teamId: new mongoose.Types.ObjectId(teamId),
+		});
+	};
+
+	countByTeamIdGrouped = async (teamId: string): Promise<DLQStatusCount[]> => {
+		const results = await DLQItemModel.aggregate([
+			{ $match: { teamId: new mongoose.Types.ObjectId(teamId) } },
+			{
+				$group: {
+					_id: { status: "$status", type: "$type" },
+					count: { $sum: 1 },
+				},
+			},
+		]);
+		return results.map((r) => ({
+			status: r._id.status,
+			type: r._id.type,
+			count: r.count,
+		}));
+	};
+
+	updateById = async (id: string, patch: Partial<DLQItem>): Promise<DLQItem> => {
+		const updated = await DLQItemModel.findOneAndUpdate(
+			{ _id: new mongoose.Types.ObjectId(id) },
+			{ $set: { ...patch } },
+			{ new: true, runValidators: true }
+		);
+		if (!updated) {
+			throw new AppError({ message: `DLQ item with id ${id} not found`, status: 404 });
+		}
+		return this.toEntity(updated);
+	};
+
+	deleteById = async (id: string, teamId: string): Promise<number> => {
+		const result = await DLQItemModel.deleteOne({
+			_id: new mongoose.Types.ObjectId(id),
+			teamId: new mongoose.Types.ObjectId(teamId),
+		});
+		return result.deletedCount || 0;
+	};
+
+	deleteOlderThan = async (date: Date): Promise<number> => {
+		const result = await DLQItemModel.deleteMany({
+			createdAt: { $lt: date },
+		});
+		return result.deletedCount || 0;
+	};
+}
+
+export default MongoDLQRepository;

server/src/repositories/index.ts

@@ -48,4 +48,7 @@ export { TimescaleMaintenanceWindowsRepository } from "@/repositories/maintenanc
 
 export * from "@/repositories/geo-checks/IGeoChecksRepository.js";
 export { default as MongoGeoChecksRepository } from "@/repositories/geo-checks/MongoGeoChecksRepository.js";
+
+export * from "@/repositories/dlq/IDLQRepository.js";
+export { default as MongoDLQRepository } from "@/repositories/dlq/MongoDLQRepository.js";
 export { TimescaleGeoChecksRepository } from "@/repositories/geo-checks/TimescaleGeoChecksRepository.js";

server/src/types/dlqItem.ts

@@ -0,0 +1,20 @@
+export const DLQItemTypes = ["notification", "incident_create", "incident_resolve"] as const;
+export type DLQItemType = (typeof DLQItemTypes)[number];
+
+export const DLQItemStatuses = ["pending", "retrying", "failed"] as const;
+export type DLQItemStatus = (typeof DLQItemStatuses)[number];
+
+export interface DLQItem {
+	id: string;
+	type: DLQItemType;
+	status: DLQItemStatus;
+	payload: Record<string, unknown>;
+	monitorId: string;
+	teamId: string;
+	retryCount: number;
+	maxRetries: number;
+	lastError: string;
+	nextRetryAt: string;
+	createdAt: string;
+	updatedAt: string;
+}

server/src/types/index.ts

@@ -14,3 +14,4 @@ export * from "@/types/incident.js";
 export * from "@/types/email.js";
 export * from "@/types/team.js";
 export * from "@/types/maintenanceWindow.js";
+export * from "@/types/dlqItem.js";