From 94ca9896db7e2a1f485a301c9acb75ddc8785280 Mon Sep 17 00:00:00 2001 From: Br0wnHammer Date: Fri, 27 Mar 2026 20:05:03 +0530 Subject: [PATCH 1/5] feat(dlq): add DLQItem type definitions for dead letter queue --- server/src/types/dlqItem.ts | 20 ++++++++++++++++++++ server/src/types/index.ts | 1 + 2 files changed, 21 insertions(+) create mode 100644 server/src/types/dlqItem.ts diff --git a/server/src/types/dlqItem.ts b/server/src/types/dlqItem.ts new file mode 100644 index 000000000..da882ace5 --- /dev/null +++ b/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; + monitorId: string; + teamId: string; + retryCount: number; + maxRetries: number; + lastError: string; + nextRetryAt: string; + createdAt: string; + updatedAt: string; +} diff --git a/server/src/types/index.ts b/server/src/types/index.ts index e5b7bf59d..89de348e2 100644 --- a/server/src/types/index.ts +++ b/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"; From 60b1d95b74bd8edec7c12b36f3bb1fd37c324236 Mon Sep 17 00:00:00 2001 From: Br0wnHammer Date: Fri, 27 Mar 2026 20:06:12 +0530 Subject: [PATCH 2/5] feat(dlq): add DLQItem Mongoose model with indexes for retry polling and cleanup --- server/src/db/models/DLQItem.ts | 73 +++++++++++++++++++++++++++++++++ server/src/db/models/index.ts | 3 ++ 2 files changed, 76 insertions(+) create mode 100644 server/src/db/models/DLQItem.ts diff --git a/server/src/db/models/DLQItem.ts b/server/src/db/models/DLQItem.ts new file mode 100644 index 000000000..9453519a1 --- /dev/null +++ b/server/src/db/models/DLQItem.ts @@ -0,0 +1,73 @@ +import { Schema, model, type Types } from "mongoose"; +import { DLQItemTypes, DLQItemStatuses, type DLQItem } from "@/types/dlqItem.js"; + +type DLQItemDocumentBase = Omit & { + monitorId: Types.ObjectId; + teamId: Types.ObjectId; + nextRetryAt: Date; + createdAt: Date; + updatedAt: Date; +}; + +export interface DLQItemDocument extends DLQItemDocumentBase { + _id: Types.ObjectId; +} + +const DLQItemSchema = new Schema( + { + 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("DLQItem", DLQItemSchema); + +export { DLQItemModel }; +export default DLQItemModel; diff --git a/server/src/db/models/index.ts b/server/src/db/models/index.ts index 1024b2b93..d4c8f1a3b 100644 --- a/server/src/db/models/index.ts +++ b/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"; From e6088fe06f53f08cae27d8035668f1c1d2e15602 Mon Sep 17 00:00:00 2001 From: Br0wnHammer Date: Fri, 27 Mar 2026 20:07:00 +0530 Subject: [PATCH 3/5] feat(dlq): add IDLQRepository interface for dead letter queue data access --- server/src/repositories/dlq/IDLQRepository.ts | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) create mode 100644 server/src/repositories/dlq/IDLQRepository.ts diff --git a/server/src/repositories/dlq/IDLQRepository.ts b/server/src/repositories/dlq/IDLQRepository.ts new file mode 100644 index 000000000..5e26e6096 --- /dev/null +++ b/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): Promise; + // fetch + findById(id: string): Promise; + findByTeamId(teamId: string, filters: DLQQueryFilters): Promise; + findRetryable(limit: number): Promise; + countByTeamId(teamId: string): Promise; + countByTeamIdGrouped(teamId: string): Promise; + // update + updateById(id: string, patch: Partial): Promise; + // delete + deleteById(id: string, teamId: string): Promise; + deleteOlderThan(date: Date): Promise; +} From 214fc4899c90268759e0612273d2d7e6cbddb38d Mon Sep 17 00:00:00 2001 From: Br0wnHammer Date: Fri, 27 Mar 2026 20:08:18 +0530 Subject: [PATCH 4/5] feat(dlq): add MongoDLQRepository with retry polling and grouped count queries --- .../repositories/dlq/MongoDLQRepository.ts | 139 ++++++++++++++++++ server/src/repositories/index.ts | 3 + 2 files changed, 142 insertions(+) create mode 100644 server/src/repositories/dlq/MongoDLQRepository.ts diff --git a/server/src/repositories/dlq/MongoDLQRepository.ts b/server/src/repositories/dlq/MongoDLQRepository.ts new file mode 100644 index 000000000..8c9da76ad --- /dev/null +++ b/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): Promise { + const newItem = await DLQItemModel.create(item); + return this.toEntity(newItem); + } + + findById = async (id: string): Promise => { + const item = await DLQItemModel.findById(new mongoose.Types.ObjectId(id)); + return item ? this.toEntity(item) : null; + }; + + findByTeamId = async (teamId: string, filters: DLQQueryFilters): Promise => { + const query: Record = { + 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 => { + 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 => { + return DLQItemModel.countDocuments({ + teamId: new mongoose.Types.ObjectId(teamId), + }); + }; + + countByTeamIdGrouped = async (teamId: string): Promise => { + 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): Promise => { + 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 => { + 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 => { + const result = await DLQItemModel.deleteMany({ + createdAt: { $lt: date }, + }); + return result.deletedCount || 0; + }; +} + +export default MongoDLQRepository; diff --git a/server/src/repositories/index.ts b/server/src/repositories/index.ts index c1d8c2ecf..2dc1a17c3 100644 --- a/server/src/repositories/index.ts +++ b/server/src/repositories/index.ts @@ -36,3 +36,6 @@ export { default as MongoMaintenanceWindowsRepository } from "@/repositories/mai 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"; From a895912b9901682e49330ead801efaccc42a0467 Mon Sep 17 00:00:00 2001 From: Br0wnHammer Date: Fri, 27 Mar 2026 23:23:01 +0530 Subject: [PATCH 5/5] refactor(dlq): inline DLQItemDocumentBase into DLQItemDocument --- server/src/db/models/DLQItem.ts | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/server/src/db/models/DLQItem.ts b/server/src/db/models/DLQItem.ts index 9453519a1..707846adb 100644 --- a/server/src/db/models/DLQItem.ts +++ b/server/src/db/models/DLQItem.ts @@ -1,16 +1,13 @@ import { Schema, model, type Types } from "mongoose"; import { DLQItemTypes, DLQItemStatuses, type DLQItem } from "@/types/dlqItem.js"; -type DLQItemDocumentBase = Omit & { +export interface DLQItemDocument extends Omit { + _id: Types.ObjectId; monitorId: Types.ObjectId; teamId: Types.ObjectId; nextRetryAt: Date; createdAt: Date; updatedAt: Date; -}; - -export interface DLQItemDocument extends DLQItemDocumentBase { - _id: Types.ObjectId; } const DLQItemSchema = new Schema(