From cdb276d5676600588d1813d150987c4e7243d520 Mon Sep 17 00:00:00 2001
From: Alex Holliday <ajhollid@gmail.com>
Date: Wed, 5 Feb 2025 12:40:52 -0800
Subject: [PATCH 1/7] update DU check

---
 Server/db/models/DistributedUptimeCheck.js | 84 ++++++++++++++++++++++
 1 file changed, 84 insertions(+)

diff --git a/Server/db/models/DistributedUptimeCheck.js b/Server/db/models/DistributedUptimeCheck.js
index 0c7c25de3..d09d9b47e 100644
--- a/Server/db/models/DistributedUptimeCheck.js
+++ b/Server/db/models/DistributedUptimeCheck.js
@@ -1,13 +1,97 @@
 import mongoose from "mongoose";
+
 import { BaseCheckSchema } from "./Check.js";
 
+// {
+// 	"id": "12123",
+// 	"result": {
+// 	  "task_arrived": "2025-01-13T19:21:37.463466602Z",
+// 	  "dns_start": "2025-01-14T00:21:33.1801319+05:00",
+// 	  "dns_end": "2025-01-14T00:21:33.4582552+05:00",
+// 	  "conn_start": "2025-01-14T00:21:33.1801319+05:00",
+// 	  "conn_end": "2025-01-14T00:21:33.7076318+05:00",
+// 	  "connect_start": "2025-01-14T00:21:33.4582552+05:00",
+// 	  "connect_end": "2025-01-14T00:21:33.541899+05:00",
+// 	  "tls_hand_shake_start": "2025-01-14T00:21:33.541899+05:00",
+// 	  "tls_hand_shake_end": "2025-01-14T00:21:33.7076318+05:00",
+// 	  "body_read_start": "2025-01-14T00:21:34.1894707+05:00",
+// 	  "body_read_end": "2025-01-14T00:21:34.1894707+05:00",
+// 	  "wrote_request": "2025-01-14T00:21:33.7076318+05:00",
+// 	  "got_first_response_byte": "2025-01-14T00:21:34.1327652+05:00",
+// 	  "first_byte_took": 425133400,
+// 	  "body_read_took": 56030000,
+// 	  "dns_took": 278123300,
+// 	  "conn_took": 527499900,
+// 	  "connect_took": 83643800,
+// 	  "tls_took": 165732800,
+// 	  "sni_name": "uprock.com",
+// 	  "status_code": 200,
+// 	  "body_size": 19320,
+// 	  "request_header_size": 95,
+// 	  "response_header_size": 246,
+// 	  "response_headers": "X-Vercel-Id: bom1::iad1::sm87v-1736796096856-aec270c01f23\nDate: Mon, 13 Jan 2025 19:21:37 GMT\nServer: Vercel\nStrict-Transport-Security: max-age=63072000\nVary: RSC, Next-Router-State-Tree, Next-Router-Prefetch, Next-Url\nX-Matched-Path: /\nX-Powered-By: Next.js\nX-Vercel-Cache: MISS\nAge: 0\nCache-Control: private, no-cache, no-store, max-age=0, must-revalidate\nContent-Type: text/html; charset=utf-8",
+// 	  "error": "",
+// 	  "device_id": "d5f578e143a2cd603dd6bf5f846a86a538bde4a8fbe2ad1fca284ad9f033daf8",
+// 	  "ip_address": "223.123.19.0",
+// 	  "proof": "",
+// 	  "created_at": "2025-01-13T19:21:37.463466912Z",
+// 	  "continent": "AS",
+// 	  "country_code": "PK",
+// 	  "city": "Muzaffargarh",
+// 	  "upt_burnt" : "0.01",
+// 	  "location": {
+// 		"lat": 71.0968,
+// 		"lng": 30.0208
+// 	  },
+// 	  "payload": {
+// 		"callback": "https://webhook.site/2a15b0af-545a-4ac2-b913-153b97592d7a",
+// 		"x": "y"
+// 	  }
+// 	}
+//   }
+
+const LocationSchema = new mongoose.Schema(
+	{
+		lat: { type: Number, required: true },
+		lng: { type: Number, required: true },
+	},
+	{ _id: false }
+);
+
 const DistributedUptimeCheckSchema = mongoose.Schema(
 	{
 		...BaseCheckSchema.obj,
+		location: {
+			type: LocationSchema,
+			required: false,
+		},
+		continent: {
+			type: String,
+			required: false,
+		},
+		countryCode: {
+			type: String,
+			required: false,
+		},
+		city: {
+			type: String,
+			required: false,
+		},
+		uptBurnt: {
+			type: mongoose.Schema.Types.Decimal128,
+			required: false,
+		},
 	},
 	{ timestamps: true }
 );
 
+DistributedUptimeCheckSchema.pre("save", function (next) {
+	if (this.isModified("uptBurnt") && typeof this.uptBurnt === "string") {
+		this.uptBurnt = mongoose.Types.Decimal128.fromString(this.uptBurnt);
+	}
+	next();
+});
+
 DistributedUptimeCheckSchema.index({ createdAt: 1 });
 DistributedUptimeCheckSchema.index({ monitorId: 1, createdAt: 1 });
 DistributedUptimeCheckSchema.index({ monitorId: 1, createdAt: -1 });

From 3d731137dfa9bd16e9197db08483dc6ea52a68f8 Mon Sep 17 00:00:00 2001
From: Alex Holliday <ajhollid@gmail.com>
Date: Wed, 5 Feb 2025 12:45:14 -0800
Subject: [PATCH 2/7] add DU details query

---
 Server/db/mongo/modules/monitorModule.js      |  45 +++++
 .../db/mongo/modules/monitorModuleQueries.js  | 186 +++++++++++++++++-
 2 files changed, 230 insertions(+), 1 deletion(-)

diff --git a/Server/db/mongo/modules/monitorModule.js b/Server/db/mongo/modules/monitorModule.js
index af4ea7e9f..c3d1a38be 100644
--- a/Server/db/mongo/modules/monitorModule.js
+++ b/Server/db/mongo/modules/monitorModule.js
@@ -2,6 +2,7 @@ import Monitor from "../../models/Monitor.js";
 import Check from "../../models/Check.js";
 import PageSpeedCheck from "../../models/PageSpeedCheck.js";
 import HardwareCheck from "../../models/HardwareCheck.js";
+import DistributedUptimeCheck from "../../models/DistributedUptimeCheck.js";
 import { errorMessages } from "../../../utils/messages.js";
 import Notification from "../../models/Notification.js";
 import { NormalizeData, NormalizeDataUptimeDetails } from "../../../utils/dataUtils.js";
@@ -11,6 +12,7 @@ import { fileURLToPath } from "url";
 import {
 	buildUptimeDetailsPipeline,
 	buildHardwareDetailsPipeline,
+	buildDistributedUptimeDetailsPipeline,
 } from "./monitorModuleQueries.js";
 import { ObjectId } from "mongodb";
 const __filename = fileURLToPath(import.meta.url);
@@ -366,6 +368,48 @@ const getUptimeDetailsById = async (req) => {
 	}
 };
 
+const getDistributedUptimeDetailsById = async (req) => {
+	try {
+		const { monitorId } = req.params;
+		const monitor = await Monitor.findById(monitorId);
+		if (monitor === null || monitor === undefined) {
+			throw new Error(errorMessages.DB_FIND_MONITOR_BY_ID(monitorId));
+		}
+
+		const { dateRange, normalize } = req.query;
+		const dates = getDateRange(dateRange);
+		const formatLookup = {
+			day: "%Y-%m-%dT%H:%M:00Z",
+			week: "%Y-%m-%dT%H:00:00Z",
+			month: "%Y-%m-%dT00:00:00Z",
+		};
+
+		const dateString = formatLookup[dateRange];
+		const results = await DistributedUptimeCheck.aggregate(
+			buildDistributedUptimeDetailsPipeline(monitor, dates, dateString)
+		);
+
+		const monitorData = results[0];
+		const normalizedGroupChecks = NormalizeDataUptimeDetails(
+			monitorData.groupedChecks,
+			10,
+			100
+		);
+
+		const monitorStats = {
+			...monitor.toObject(),
+			...monitorData,
+			groupedChecks: normalizedGroupChecks,
+		};
+
+		return monitorStats;
+	} catch (error) {
+		error.service = SERVICE_NAME;
+		error.method = "getUptimeDetailsById";
+		throw error;
+	}
+};
+
 /**
  * Get stats by monitor ID
  * @async
@@ -815,6 +859,7 @@ export {
 	getMonitorById,
 	getMonitorsByTeamId,
 	getUptimeDetailsById,
+	getDistributedUptimeDetailsById,
 	createMonitor,
 	deleteMonitor,
 	deleteAllMonitors,
diff --git a/Server/db/mongo/modules/monitorModuleQueries.js b/Server/db/mongo/modules/monitorModuleQueries.js
index 094e1f932..489017f51 100644
--- a/Server/db/mongo/modules/monitorModuleQueries.js
+++ b/Server/db/mongo/modules/monitorModuleQueries.js
@@ -534,4 +534,188 @@ const buildHardwareDetailsPipeline = (monitor, dates, dateString) => {
 	];
 };
 
-export { buildUptimeDetailsPipeline, buildHardwareDetailsPipeline };
+const buildDistributedUptimeDetailsPipeline = (monitor, dates, dateString) => {
+	return [
+		{
+			$match: {
+				monitorId: monitor._id,
+			},
+		},
+		{
+			$sort: {
+				createdAt: 1,
+			},
+		},
+		{
+			$facet: {
+				aggregateData: [
+					{
+						$group: {
+							_id: null,
+							avgResponseTime: {
+								$avg: "$responseTime",
+							},
+							lastCheck: {
+								$last: "$$ROOT",
+							},
+							totalChecks: {
+								$sum: 1,
+							},
+							uptBurnt: {
+								$sum: "$uptBurnt",
+							},
+						},
+					},
+				],
+				// For the response time chart, should return checks for date window
+				// Grouped by: {day: hour}, {week: day}, {month: day}
+				groupedMapChecks: [
+					{
+						$match: {
+							createdAt: { $gte: dates.start, $lte: dates.end },
+						},
+					},
+					{
+						$group: {
+							_id: {
+								date: {
+									$dateToString: {
+										format: dateString,
+										date: "$createdAt",
+									},
+								},
+								city: "$city",
+								lat: "$location.lat",
+								lng: "$location.lng",
+							},
+							city: { $first: "$city" }, // Add this line to include city in output
+							lat: { $first: "$location.lat" },
+							lng: { $first: "$location.lng" },
+							avgResponseTime: {
+								$avg: "$responseTime",
+							},
+							totalChecks: {
+								$sum: 1,
+							},
+						},
+					},
+					{
+						$sort: {
+							"_id.date": 1,
+						},
+					},
+				],
+				groupedChecks: [
+					{
+						$match: {
+							createdAt: { $gte: dates.start, $lte: dates.end },
+						},
+					},
+					{
+						$group: {
+							_id: {
+								$dateToString: {
+									format: dateString,
+									date: "$createdAt",
+								},
+							},
+							avgResponseTime: {
+								$avg: "$responseTime",
+							},
+							totalChecks: {
+								$sum: 1,
+							},
+						},
+					},
+					{
+						$sort: {
+							_id: 1,
+						},
+					},
+				],
+				// Average response time for the date window
+				groupAvgResponseTime: [
+					{
+						$match: {
+							createdAt: { $gte: dates.start, $lte: dates.end },
+						},
+					},
+					{
+						$group: {
+							_id: null,
+							avgResponseTime: {
+								$avg: "$responseTime",
+							},
+						},
+					},
+				],
+				latestChecks: [
+					{
+						$sort: { createdAt: -1 }, // Sort by newest first
+					},
+					{
+						$limit: 5, // Get only the first 5 documents
+					},
+					{
+						$project: {
+							responseTime: 1,
+							city: 1,
+							countryCode: 1,
+							uptBurnt: { $toString: "$uptBurnt" },
+						},
+					},
+				],
+			},
+		},
+		{
+			$project: {
+				totalUptBurnt: {
+					$toString: {
+						$arrayElemAt: ["$aggregateData.uptBurnt", 0],
+					},
+				},
+				avgResponseTime: {
+					$arrayElemAt: ["$aggregateData.avgResponseTime", 0],
+				},
+				totalChecks: {
+					$arrayElemAt: ["$aggregateData.totalChecks", 0],
+				},
+				latestResponseTime: {
+					$arrayElemAt: ["$aggregateData.lastCheck.responseTime", 0],
+				},
+				timeSinceLastCheck: {
+					$let: {
+						vars: {
+							lastCheck: {
+								$arrayElemAt: ["$aggregateData.lastCheck", 0],
+							},
+						},
+						in: {
+							$cond: [
+								{
+									$ifNull: ["$$lastCheck", false],
+								},
+								{
+									$subtract: [new Date(), "$$lastCheck.createdAt"],
+								},
+								0,
+							],
+						},
+					},
+				},
+				groupedMapChecks: "$groupedMapChecks",
+				groupedChecks: "$groupedChecks",
+				groupedAvgResponseTime: {
+					$arrayElemAt: ["$groupAvgResponseTime", 0],
+				},
+				latestChecks: "$latestChecks",
+			},
+		},
+	];
+};
+
+export {
+	buildUptimeDetailsPipeline,
+	buildHardwareDetailsPipeline,
+	buildDistributedUptimeDetailsPipeline,
+};

From 8bfde01bfa6bdf849336f1ca886c2c284262384b Mon Sep 17 00:00:00 2001
From: Alex Holliday <ajhollid@gmail.com>
Date: Wed, 5 Feb 2025 12:47:17 -0800
Subject: [PATCH 3/7] add DU to jobqueue

---
 Server/service/jobQueue.js | 52 ++++----------------------------------
 1 file changed, 5 insertions(+), 47 deletions(-)

diff --git a/Server/service/jobQueue.js b/Server/service/jobQueue.js
index 419785540..f7c70f0a7 100644
--- a/Server/service/jobQueue.js
+++ b/Server/service/jobQueue.js
@@ -1,4 +1,4 @@
-const QUEUE_NAMES = ["uptime", "pagespeed", "hardware"];
+const QUEUE_NAMES = ["uptime", "pagespeed", "hardware", "distributed"];
 const SERVICE_NAME = "JobQueue";
 const JOBS_PER_WORKER = 5;
 const QUEUE_LOOKUP = {
@@ -8,6 +8,7 @@ const QUEUE_LOOKUP = {
 	port: "uptime",
 	docker: "uptime",
 	pagespeed: "pagespeed",
+	distributed_http: "distributed",
 };
 const getSchedulerId = (monitor) => `scheduler:${monitor.type}:${monitor._id}`;
 
@@ -163,6 +164,9 @@ class NewJobQueue {
 				// Get the current status
 				await job.updateProgress(30);
 				const networkResponse = await this.networkService.getStatus(job);
+				if (job.data.type === "distributed_http") {
+					return;
+				}
 				// Handle status change
 				await job.updateProgress(60);
 				const { monitor, statusChanged, prevStatus } =
@@ -219,51 +223,6 @@ class NewJobQueue {
 			connection: this.connection,
 			concurrency: 5,
 		});
-
-		// worker.on("active", (job) => {
-		// 	this.logger.info({
-		// 		message: `Worker started processing job: ${job.id}`,
-		// 		service: SERVICE_NAME,
-		// 		method: "createWorker",
-		// 	});
-		// });
-
-		// worker.on("completed", (job) => {
-		// 	this.logger.info({
-		// 		message: `Worker completed job: ${job.id}`,
-		// 		service: SERVICE_NAME,
-		// 		method: "createWorker",
-		// 	});
-		// });
-
-		// // Log job progress updates
-		// worker.on("progress", (job, progress) => {
-		// 	this.logger.info({
-		// 		message: `Job progress: ${job.id}`,
-		// 		service: SERVICE_NAME,
-		// 		method: "createWorker",
-		// 		details: `Progress: ${progress}%`,
-		// 	});
-		// });
-
-		// // Log when a job fails
-		// worker.on("failed", (job, err) => {
-		// 	this.logger.error({
-		// 		message: `Worker failed job: ${job.id}`,
-		// 		service: SERVICE_NAME,
-		// 		method: "createWorker",
-		// 		details: `Error: ${err.message}`,
-		// 		stack: err.stack,
-		// 	});
-		// });
-
-		// worker.on("stalled", (jobId) => {
-		// 	this.logger.warn({
-		// 		message: `Worker stalled job: ${jobId}`,
-		// 		service: SERVICE_NAME,
-		// 		method: "createWorker",
-		// 	});
-		// });
 		return worker;
 	}
 
@@ -392,7 +351,6 @@ class NewJobQueue {
 					const jobs = await queue.getJobs();
 					const ret = await Promise.all(
 						jobs.map(async (job) => {
-							console.log(job);
 							const state = await job.getState();
 							return { url: job.data.url, state, progress: job.progress };
 						})

From f4b87daa7f71c0380ce0351e2e1f83778d6ebd2f Mon Sep 17 00:00:00 2001
From: Alex Holliday <ajhollid@gmail.com>
Date: Wed, 5 Feb 2025 12:48:57 -0800
Subject: [PATCH 4/7] Add DU to network service

---
 Server/service/networkService.js | 33 ++++++++++++++++++++++++++++++++
 1 file changed, 33 insertions(+)

diff --git a/Server/service/networkService.js b/Server/service/networkService.js
index 54edf39f9..8a58d4b88 100644
--- a/Server/service/networkService.js
+++ b/Server/service/networkService.js
@@ -1,5 +1,6 @@
 import { errorMessages, successMessages } from "../utils/messages.js";
 const SERVICE_NAME = "NetworkService";
+const UPROCK_ENDPOINT = "https://api.uprock.com/checkmate/push";
 
 /**
  * Constructs a new NetworkService instance.
@@ -19,6 +20,7 @@ class NetworkService {
 		this.TYPE_HARDWARE = "hardware";
 		this.TYPE_DOCKER = "docker";
 		this.TYPE_PORT = "port";
+		this.TYPE_DISTRIBUTED_HTTP = "distributed_http";
 		this.SERVICE_NAME = SERVICE_NAME;
 		this.NETWORK_ERROR = 5000;
 		this.PING_ERROR = 5001;
@@ -323,6 +325,35 @@ class NetworkService {
 		}
 	}
 
+	async requestDistributedHttp(job) {
+		try {
+			const monitor = job.data;
+			const CALLBACK_URL = process.env.NGROK_URL
+				? process.env.NGROK_URL
+				: process.env.CALLBACK_URL;
+
+			const response = await this.axios.post(
+				UPROCK_ENDPOINT,
+				{
+					id: monitor._id,
+					url: monitor.url,
+					callback: `${CALLBACK_URL}/api/v1/distributed-uptime/callback`,
+				},
+				{
+					headers: {
+						"Content-Type": "application/json",
+						"x-checkmate-key": process.env.UPROCK_API_KEY,
+					},
+				}
+			);
+		} catch (error) {
+			console.log(error.message);
+			error.service = this.SERVICE_NAME;
+			error.method = "requestDistributedHttp";
+			throw error;
+		}
+	}
+
 	/**
 	 * Handles unsupported job types by throwing an error with details.
 	 *
@@ -360,6 +391,8 @@ class NetworkService {
 				return await this.requestDocker(job);
 			case this.TYPE_PORT:
 				return await this.requestPort(job);
+			case this.TYPE_DISTRIBUTED_HTTP:
+				return await this.requestDistributedHttp(job);
 
 			default:
 				return this.handleUnsupportedType(type);

From 254079be0361ececafd43a5412ae0017d6f8312f Mon Sep 17 00:00:00 2001
From: Alex Holliday <ajhollid@gmail.com>
Date: Wed, 5 Feb 2025 12:50:15 -0800
Subject: [PATCH 5/7] add DU controller

---
 .../distributedUptimeController.js            | 203 +++++++++++++++++-
 1 file changed, 200 insertions(+), 3 deletions(-)

diff --git a/Server/controllers/distributedUptimeController.js b/Server/controllers/distributedUptimeController.js
index 5215c2cce..ee8f5398e 100644
--- a/Server/controllers/distributedUptimeController.js
+++ b/Server/controllers/distributedUptimeController.js
@@ -1,16 +1,213 @@
 import { handleError } from "./controllerUtils.js";
-import { successMessages } from "../utils/messages.js";
+import Monitor from "../db/models/Monitor.js";
+import DistributedUptimeCheck from "../db/models/DistributedUptimeCheck.js";
 
 const SERVICE_NAME = "DistributedUptimeQueueController";
 
 class DistributedUptimeController {
-	constructor() {}
+	constructor(db, http, statusService) {
+		this.db = db;
+		this.http = http;
+		this.statusService = statusService;
+		this.resultsCallback = this.resultsCallback.bind(this);
+		this.getDistributedUptimeMonitors = this.getDistributedUptimeMonitors.bind(this);
+		this.getDistributedUptimeMonitorDetails =
+			this.getDistributedUptimeMonitorDetails.bind(this);
+	}
 
 	async resultsCallback(req, res, next) {
 		try {
+			const { id, result } = req.body;
+			// Calculate response time
+			const {
+				first_byte_took,
+				body_read_took,
+				dns_took,
+				conn_took,
+				connect_took,
+				tls_took,
+				status_code,
+				error,
+			} = result;
+
+			// Calculate response time
+			const responseTime =
+				(first_byte_took +
+					body_read_took +
+					dns_took +
+					conn_took +
+					connect_took +
+					tls_took) /
+				1_000_000;
+
+			// Calculate if server is up or down
+			const isErrorStatus = status_code >= 400;
+			const hasError = error !== "";
+
+			const status = isErrorStatus || hasError ? false : true;
+
+			// Build response
+			const distributedUptimeResponse = {
+				monitorId: id,
+				type: "distributed_http",
+				payload: result,
+				status,
+				code: status_code,
+				responseTime,
+			};
+
+			if (error) {
+				const code = status_code || this.NETWORK_ERROR;
+				distributedUptimeResponse.code = code;
+				distributedUptimeResponse.message =
+					this.http.STATUS_CODES[code] || "Network Error";
+			} else {
+				distributedUptimeResponse.message = this.http.STATUS_CODES[status_code];
+			}
+
+			await this.statusService.updateStatus(distributedUptimeResponse);
+
 			res.status(200).json({ message: "OK" });
 		} catch (error) {
-			throw handleError(error, SERVICE_NAME, "resultsCallback");
+			next(handleError(error, SERVICE_NAME, "resultsCallback"));
+		}
+	}
+
+	async getDistributedUptimeMonitors(req, res, next) {
+		try {
+			res.setHeader("Content-Type", "text/event-stream");
+			res.setHeader("Cache-Control", "no-cache");
+			res.setHeader("Connection", "keep-alive");
+			res.setHeader("Access-Control-Allow-Origin", "*");
+
+			const BATCH_DELAY = 1000;
+			let batchTimeout = null;
+			let opInProgress = false;
+
+			// Do things here
+			const notifyChange = async () => {
+				if (opInProgress) {
+					// Get data
+					const monitors = await this.db.getMonitorsByTeamId(req);
+					res.write(`data: ${JSON.stringify({ monitors })}\n\n`);
+					opInProgress = false;
+				}
+				batchTimeout = null;
+			};
+
+			const handleChange = () => {
+				opInProgress = true;
+				if (batchTimeout) clearTimeout(batchTimeout);
+				batchTimeout = setTimeout(notifyChange, BATCH_DELAY);
+			};
+
+			const monitorStream = Monitor.watch(
+				[{ $match: { operationType: { $in: ["insert", "update", "delete"] } } }],
+				{ fullDocument: "updateLookup" }
+			);
+
+			const checksStream = DistributedUptimeCheck.watch(
+				[{ $match: { operationType: { $in: ["insert", "update", "delete"] } } }],
+				{ fullDocument: "updateLookup" }
+			);
+
+			monitorStream.on("change", handleChange);
+			checksStream.on("change", handleChange);
+
+			// Send initial data
+			const monitors = await this.db.getMonitorsByTeamId(req);
+			res.write(`data: ${JSON.stringify({ monitors })}\n\n`);
+
+			// Handle client disconnect
+			req.on("close", () => {
+				if (batchTimeout) {
+					clearTimeout(batchTimeout);
+				}
+				monitorStream.close();
+				checksStream.close();
+				clearInterval(keepAlive);
+			});
+
+			// Keep connection alive
+			const keepAlive = setInterval(() => {
+				res.write(": keepalive\n\n");
+			}, 30000);
+
+			// Clean up on close
+			req.on("close", () => {
+				clearInterval(keepAlive);
+			});
+		} catch (error) {
+			next(handleError(error, SERVICE_NAME, "getDistributedUptimeMonitors"));
+		}
+	}
+
+	async getDistributedUptimeMonitorDetails(req, res, next) {
+		try {
+			res.setHeader("Content-Type", "text/event-stream");
+			res.setHeader("Cache-Control", "no-cache");
+			res.setHeader("Connection", "keep-alive");
+			res.setHeader("Access-Control-Allow-Origin", "*");
+
+			const BATCH_DELAY = 1000;
+			let batchTimeout = null;
+			let opInProgress = false;
+
+			// Do things here
+			const notifyChange = async () => {
+				try {
+					if (opInProgress) {
+						// Get data
+						const monitor = await this.db.getDistributedUptimeDetailsById(req);
+						res.write(`data: ${JSON.stringify({ monitor })}\n\n`);
+						opInProgress = false;
+					}
+					batchTimeout = null;
+				} catch (error) {
+					console.error("Error in notifyChange:", error);
+					opInProgress = false;
+					batchTimeout = null;
+					next(handleError(error, SERVICE_NAME, "getDistributedUptimeMonitorDetails"));
+				}
+			};
+
+			const handleChange = () => {
+				opInProgress = true;
+				if (batchTimeout) clearTimeout(batchTimeout);
+				batchTimeout = setTimeout(notifyChange, BATCH_DELAY);
+			};
+
+			const checksStream = DistributedUptimeCheck.watch(
+				[{ $match: { operationType: { $in: ["insert", "update", "delete"] } } }],
+				{ fullDocument: "updateLookup" }
+			);
+
+			checksStream.on("change", handleChange);
+
+			// Send initial data
+			const monitor = await this.db.getDistributedUptimeDetailsById(req);
+			res.write(`data: ${JSON.stringify({ monitor })}\n\n`);
+
+			// Handle client disconnect
+			req.on("close", () => {
+				if (batchTimeout) {
+					clearTimeout(batchTimeout);
+				}
+				checksStream.close();
+				clearInterval(keepAlive);
+			});
+
+			// Keep connection alive
+			const keepAlive = setInterval(() => {
+				res.write(": keepalive\n\n");
+			}, 30000);
+
+			// Clean up on close
+			req.on("close", () => {
+				clearInterval(keepAlive);
+			});
+		} catch (error) {
+			next(handleError(error, SERVICE_NAME, "getDistributedUptimeMonitorDetails"));
 		}
 	}
 }

From 72460b15d10c9537a12272cd0eb20622ea97e4b5 Mon Sep 17 00:00:00 2001
From: Alex Holliday <ajhollid@gmail.com>
Date: Wed, 5 Feb 2025 12:53:47 -0800
Subject: [PATCH 6/7] add DU routes

---
 Server/routes/distributedUptimeRoute.js | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/Server/routes/distributedUptimeRoute.js b/Server/routes/distributedUptimeRoute.js
index eac4cb675..3fbd88d3a 100644
--- a/Server/routes/distributedUptimeRoute.js
+++ b/Server/routes/distributedUptimeRoute.js
@@ -8,6 +8,14 @@ class DistributedUptimeRoutes {
 	}
 	initRoutes() {
 		this.router.post("/callback", this.distributedUptimeController.resultsCallback);
+		this.router.get(
+			"/monitors/:teamId",
+			this.distributedUptimeController.getDistributedUptimeMonitors
+		);
+		this.router.get(
+			"/monitors/details/:monitorId",
+			this.distributedUptimeController.getDistributedUptimeMonitorDetails
+		);
 	}
 
 	getRouter() {

From e361ded4f178b71f975d396a17b80a5349ea3eaa Mon Sep 17 00:00:00 2001
From: Alex Holliday <ajhollid@gmail.com>
Date: Wed, 5 Feb 2025 12:59:06 -0800
Subject: [PATCH 7/7] add DU to status service

---
 Server/service/statusService.js | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git a/Server/service/statusService.js b/Server/service/statusService.js
index cca443e9e..04f654d59 100644
--- a/Server/service/statusService.js
+++ b/Server/service/statusService.js
@@ -25,7 +25,7 @@ class StatusService {
 	 * @param {Object} networkResponse - The network response containing monitorId and status.
 	 * @param {string} networkResponse.monitorId - The ID of the monitor.
 	 * @param {string} networkResponse.status - The new status of the monitor.
-	 * @returns {Promise<Object>} - A promise that resolves to an object containing the monitor, statusChanged flag, and previous status if the status changed, or false if an error occurred.
+	 * @returns {Promise<Object>} - A promise that resolves to an object containinfg the monitor, statusChanged flag, and previous status if the status changed, or false if an error occurred.
 	 * @returns {Promise<Object>} returnObject - The object returned by the function.
 	 * @returns {Object} returnObject.monitor - The monitor object.
 	 * @returns {boolean} returnObject.statusChanged - Flag indicating if the status has changed.
@@ -85,6 +85,7 @@ class StatusService {
 	buildCheck = (networkResponse) => {
 		const { monitorId, teamId, type, status, responseTime, code, message, payload } =
 			networkResponse;
+
 		const check = {
 			monitorId,
 			teamId,
@@ -94,6 +95,14 @@ class StatusService {
 			message,
 		};
 
+		if (type === "distributed_http") {
+			check.continent = payload.continent;
+			check.countryCode = payload.country_code;
+			check.city = payload.city;
+			check.location = payload.location;
+			check.uptBurnt = payload.upt_burnt;
+		}
+
 		if (type === "pagespeed") {
 			const categories = payload.lighthouseResult?.categories;
 			const audits = payload.lighthouseResult?.audits;
@@ -145,6 +154,7 @@ class StatusService {
 				hardware: this.db.createHardwareCheck,
 				docker: this.db.createCheck,
 				port: this.db.createCheck,
+				distributed_http: this.db.createDistributedCheck,
 			};
 			const operation = operationMap[networkResponse.type];