Notification Schema

models/Notification.ts

@@ -0,0 +1,20 @@
+import mongoose from 'mongoose';
+const { Schema } = mongoose;
+
+const notificationSchema = new Schema({
+    labId: { type: String, required: true },
+    type: { type: String, required: true },
+    resourceId: { type: String, required: true },
+    recipients: [
+        {
+            role: { type: String },
+            required: true
+        }
+    ],
+    createdAt: { type: Date, required: true, default: Date.now }
+});
+
+const Notification = mongoose.models.Notification || 
+    mongoose.model('Notification', notificationSchema);
+
+export default Notification;

services/notifications/index.ts

@@ -0,0 +1,4 @@
+import { startNotificationWatcher } from "./watchUpdates";
+
+// Entry point for notification service
+startNotificationWatcher();

services/notifications/watchUpdates.ts

@@ -0,0 +1,38 @@
+import mongoose from "mongoose";
+import { connectToDatabase } from "@/lib/mongoose";
+import Notification from "@/models/Notification";
+
+/**
+ * Starts a Change Stream watcher on the "products" collection.
+ * Inserts a DB_UPDATE notification whenever a product is updated.
+ */
+export async function startNotificationWatcher() {
+  await connectToDatabase();
+
+  const collection = mongoose.connection.collection("items");
+
+  const changeStream = collection.watch([], {
+    fullDocument: "updateLookup",
+  });
+
+  for await (const change of changeStream) {
+    if (change.operationType !== "update") continue;
+
+    // Get the updated document from the change stream
+    const updatedDoc = change.fullDocument;
+
+    // If there’s no document, skip this iteration
+    if (!updatedDoc) continue;
+
+    // TODO: Check the quantity in updatedDoc
+    // If quantity is below the threshold, continue
+    // Otherwise, send a notification
+    await Notification.create({
+      _id: `notif_${Date.now()}`,
+      type: "DB_UPDATE",
+      labId: updatedDoc.labId ?? "unknown",
+      resourceId: String(updatedDoc._id),
+      recipients: [],
+    });
+  }
+}