Merge pull request #95 from Plant-for-the-Planet-org/feature/stop-dup…

…licate-alerts

Feature/stop duplicate alerts in sms

apps/server/prisma/migrations/20231027000000_stopAlertUntil_and_lastSMSCreated/migration.sql

@@ -0,0 +1,7 @@
+-- Add stopAlertUntil field to Site table
+ALTER TABLE "Site" 
+ADD COLUMN "stopAlertUntil" TIMESTAMP(3);
+
+-- Add lastMessageCreated field to Site table
+ALTER TABLE "Site" 
+ADD COLUMN "lastMessageCreated" TIMESTAMP(3);

apps/server/prisma/schema.prisma

@@ -44,7 +44,7 @@ model VerificationRequest {
 
 model AlertMethod {
     id                  String               @id @default(cuid())
-    method              String  
+    method              String
     destination         String
     isVerified          Boolean              @default(false)
     isEnabled           Boolean              @default(false)
@@ -62,25 +62,27 @@ model AlertMethod {
 }
 
 model Site {
-    id                String                   @id @default(cuid())
-    remoteId          String?
-    name              String?
-    origin            String                   @default("firealert")
-    type              SiteType
-    geometry          Json
-    radius            Int                      @default(0)
-    isMonitored       Boolean                  @default(true)
-    deletedAt         DateTime?
-    projectId         String?
-    lastUpdated       DateTime?
-    userId            String
-    slices            Json? // Will be something like ["1","2"] or ["3"] or ["1"] or ["7","8"]
-    detectionGeometry Unsupported("geometry")?
-    originalGeometry  Unsupported("geometry")?
-    detectionArea     Float?
-    alerts            SiteAlert[]
-    project           Project?                 @relation(fields: [projectId], references: [id])
-    user              User                     @relation(fields: [userId], references: [id], onDelete: Cascade)
+    id                 String                   @id @default(cuid())
+    remoteId           String?
+    name               String?
+    origin             String                   @default("firealert")
+    type               SiteType
+    geometry           Json
+    radius             Int                      @default(0)
+    isMonitored        Boolean                  @default(true)
+    deletedAt          DateTime?
+    projectId          String?
+    lastUpdated        DateTime?
+    stopAlertUntil     DateTime?
+    lastMessageCreated DateTime?
+    userId             String
+    slices             Json? // Will be something like ["1","2"] or ["3"] or ["1"] or ["7","8"]
+    detectionGeometry  Unsupported("geometry")?
+    originalGeometry   Unsupported("geometry")?
+    detectionArea      Float?
+    alerts             SiteAlert[]
+    project            Project?                 @relation(fields: [projectId], references: [id])
+    user               User                     @relation(fields: [userId], references: [id], onDelete: Cascade)
 }
 
 model Project {

apps/server/src/Services/Notifications/CreateNotifications.ts

@@ -1,32 +1,85 @@
 import {Prisma} from '@prisma/client';
 import {prisma} from '../../server/db';
 
+// Logic
+// Execute SQL
+
+// 1. Join SiteAlert with alertMethod using site and userId: 
+//     1. **get the oldest unique unprocessed alert for all sites**
+//         1. Create notifications for alertMethods 
+//             1. Create notifications for ‘device’ and ‘webhook’
+//             2. if `site.lastMessageCreated = NULL` or `site.lastMessageCreated < 2 hours ago`create notifications for ‘sms’, ‘whatsapp’, and ‘email’
+//                 1. set `site.lastMessageCreated = now`
+// 2. For all processed siteAlerts , set `siteAlert.isProcessed` to true
+
+// Repeat until nothing left to process
+
 const createNotifications = async () => {
-  let notificationsCreated = 0;
+  let totalSiteAlertProcessed = 0;
   try {
-    // In this query, the subquery retrieves all enabled and verified AlertMethods (m) for the user associated with the site.
-    // Then, a cross join is performed between the SiteAlert table (a) and the AlertMethod subquery (m), ensuring that each siteAlert is paired with all relevant alertMethods.
-    const notificationCreationQuery = Prisma.sql`
-        INSERT INTO "Notification" (id, "siteAlertId", "alertMethod", destination, "isDelivered") 
-        SELECT gen_random_uuid(), a.id, m.method, m.destination, false 
-            FROM "SiteAlert" a 
-                INNER JOIN "Site" s ON a."siteId" = s.id 
-                INNER JOIN "AlertMethod" m ON m."userId" = s."userId" 
-                    WHERE a."isProcessed" = false AND a."deletedAt" IS NULL AND m."deletedAt" IS NULL AND m."isEnabled" = true AND m."isVerified" = true AND a."eventDate" > CURRENT_TIMESTAMP - INTERVAL '24 hours'`;
-
-    const updateSiteAlertIsProcessedToTrue = Prisma.sql`UPDATE "SiteAlert" SET "isProcessed" = true WHERE "isProcessed" = false AND "deletedAt" IS NULL`;
-
-    // Create Notifications for all unprocessed SiteAlerts and Set all SiteAlert as processed
-    const results = await prisma.$transaction([
-      prisma.$executeRaw(notificationCreationQuery),
-      prisma.$executeRaw(updateSiteAlertIsProcessedToTrue),
-    ]);
-    // Since $executeRaw() returns the number of rows affected, the first result of the transaction would be notificationsCreated
-    notificationsCreated = results[0];
+      let moreAlertsToProcess = true;
+
+      while(moreAlertsToProcess){
+        const notificationCreationAndUpdate = Prisma.sql`
+        WITH NotificationsForInsert AS (
+          SELECT 
+            a.id AS "siteAlertId", 
+            m.method AS "alertMethod", 
+            m.destination,
+            a."siteId"
+          FROM "AlertMethod" m
+            INNER JOIN "Site" s ON s."userId" = m."userId" 
+            INNER JOIN (
+              SELECT DISTINCT ON ("siteId") *
+              FROM "SiteAlert"
+              WHERE "isProcessed" = false AND "deletedAt" IS NULL
+              ORDER BY "siteId", "eventDate"
+            ) a ON a."siteId" = s.id 
+          WHERE 
+            m."deletedAt" IS NULL 
+            AND m."isEnabled" = true 
+            AND m."isVerified" = true 
+            AND (
+              ((s."lastMessageCreated" IS NULL OR s."lastMessageCreated" < (CURRENT_TIMESTAMP - INTERVAL '2 hours')) AND m."method" IN ('sms', 'whatsapp', 'email')) 
+              OR m."method" IN ('device', 'webhook')
+            )
+        ),
+        InsertedNotifications AS (
+          INSERT INTO "Notification" (id, "siteAlertId", "alertMethod", destination, "isDelivered")
+          SELECT 
+            gen_random_uuid(), 
+            "siteAlertId", 
+            "alertMethod", 
+            destination, 
+            false 
+          FROM NotificationsForInsert
+          RETURNING "siteAlertId", "alertMethod"
+        ),
+        UpdatedSites AS (
+          UPDATE "Site"
+          SET "lastMessageCreated" = CURRENT_TIMESTAMP
+          WHERE "id" IN (
+            SELECT "siteId" FROM NotificationsForInsert WHERE "alertMethod" IN ('sms', 'whatsapp', 'email')
+          )
+          RETURNING "id"
+        )
+        UPDATE "SiteAlert"
+        SET "isProcessed" = true
+        WHERE "id" IN (SELECT "siteAlertId" FROM InsertedNotifications);`;
+
+        const siteAlertProcessed = await prisma.$executeRaw(notificationCreationAndUpdate);
+        totalSiteAlertProcessed += siteAlertProcessed
+
+        if(siteAlertProcessed === 0){
+          moreAlertsToProcess = false; // No more alerts to process, exit the loop
+        }else {
+          await new Promise(resolve => setTimeout(resolve, 200)); // Delay of 1/5 second
+        }
+      }
   } catch (error) {
     console.log(error);
   }
-  return notificationsCreated;
+  return totalSiteAlertProcessed;
 };
 
 export default createNotifications;

apps/server/src/Services/SiteAlert/CreateSiteAlert.ts

@@ -29,6 +29,10 @@ const createSiteAlerts = async (
                 INNER JOIN "Site" s ON ST_Within(ST_SetSRID (e.geometry, 4326), s. "detectionGeometry")
                     AND s. "deletedAt" IS NULL
                     AND s. "isMonitored" = TRUE
+                    AND (
+                      s."stopAlertUntil" IS NULL OR 
+                      s."stopAlertUntil" < CURRENT_TIMESTAMP
+                    )
             WHERE
                 e. "isProcessed" = FALSE
                 AND e. "geoEventProviderId" = ${geoEventProviderId}

apps/server/src/server/api/routers/site.ts

@@ -1,12 +1,12 @@
 import {TRPCError} from "@trpc/server";
-import {createSiteSchema, getSitesWithProjectIdParams, params, updateSiteSchema} from '../zodSchemas/site.schema'
+import {createSiteSchema, getSitesWithProjectIdParams, params, pauseAlertInputSchema, updateSiteSchema} from '../zodSchemas/site.schema'
 import {
     createTRPCRouter,
     protectedProcedure,
 } from "../trpc";
 import {checkUserHasSitePermission, checkIfPlanetROSite, triggerTestAlert} from '../../../utils/routers/site'
 import {Prisma, SiteAlert} from "@prisma/client";
-import {UserPlan} from "../../../Interfaces/AlertMethod";
+// import {UserPlan} from "../../../Interfaces/AlertMethod";
 
 export const siteRouter = createTRPCRouter({
 
@@ -341,6 +341,50 @@ export const siteRouter = createTRPCRouter({
             }
         }),
 
+        pauseAlertForSite: protectedProcedure
+            .input(pauseAlertInputSchema)
+            .mutation(async ({ctx, input}) => {
+                try {
+                    // Destructure input parameters, including siteId
+                    const {siteId, duration, unit} = input;
+
+                    // Calculate the time for the stopAlertUntil field based on unit
+                    const additionFactor = {
+                    minutes: 1000 * 60,
+                    hours: 1000 * 60 * 60,
+                    days: 1000 * 60 * 60 * 24,
+                    };
+
+                    // Calculate future date based on current time, duration, and unit
+                    const futureDate = new Date(Date.now() + duration * additionFactor[unit]);
+
+                    // Update specific site's stopAlertUntil field in the database
+                    await ctx.prisma.site.update({
+                        where: { 
+                            id: siteId 
+                        },
+                        data: { 
+                            stopAlertUntil: futureDate 
+                        },
+                    });
+                    // Constructing a readable duration message
+                    const durationUnit = unit === 'minutes' && duration === 1 ? 'minute' :
+                                        unit === 'hours' && duration === 1 ? 'hour' :
+                                        unit === 'days' && duration === 1 ? 'day' : unit;
+
+                    // Respond with a success message including pause duration details
+                    return { 
+                        status: 'success', 
+                        message: `Alert has been successfully paused for the site for ${duration} ${durationUnit}.` 
+                    };
+                } catch (error) {
+                    throw new TRPCError({
+                    code: 'INTERNAL_SERVER_ERROR',
+                    message: 'An error occurred while pausing the alert for the site',
+                    });
+                }
+            }),
+
     deleteSite: protectedProcedure
         .input(params)
         .mutation(async ({ctx, input}) => {

apps/server/src/server/api/zodSchemas/site.schema.ts

@@ -1,5 +1,5 @@
-import { z } from 'zod';
-import { nameSchema } from './user.schema';
+import {z} from 'zod';
+import {nameSchema} from './user.schema';
 
 const PointSchema = z.object({
     type: z.literal("Point"),
@@ -31,7 +31,7 @@ export const createSiteSchema = z.object({
 
 
 export const params = z.object({
-    siteId: z.string().cuid({ message: "Invalid CUID" }),
+    siteId: z.string().cuid({message: "Invalid CUID"}),
 })
 
 export const getSitesWithProjectIdParams = z.object({
@@ -58,5 +58,11 @@ export const updateSiteSchema = z.object({
     body: bodySchema,
 });
 
+export const pauseAlertInputSchema = z.object({
+    siteId: z.string().cuid({message: "Invalid CUID"}),
+    duration: z.number().min(1),
+    unit: z.enum(['minutes', 'hours', 'days']),
+});
+
 
 export type Geometry = z.TypeOf<typeof PointSchema | typeof PolygonSchema | typeof MultiPolygonSchema>;