GCW-3681 Create upsertReadReceipt atomic sql query

lib/services/activities.ts

@@ -3,7 +3,60 @@ import { throwForbidden }                                from "../utils/errors"
 import { ActivitiesApi }                                 from "sunshine-conversations-client"
 import { abilities }                                     from "./abilities"
 import config                                            from "../config"
-import db                                                from "../db"
+import { sql }                                           from "../db"
+import { isFreshRecord, pubsub }                         from "./events"
+
+const transformToPrismaDAO = (obj: any): ReadReceipt => {
+  return {
+    ...obj,
+    createdAt: new Date(obj.createdAt),
+    updatedAt: new Date(obj.updatedAt)
+  }
+}
+
+/**
+ * this atomic query upserts ReadReceipt with the latest message of a conversation
+ * 
+ * If there are no messages for a particular conversationId, 
+ * the insert clause would simply insert nothing
+ */
+async function createOrUpdateReadReceipt(
+  conversationId: number, 
+  staffId: number, 
+  authorType: string
+): Promise<ReadReceipt|null> {
+  const result = await sql`
+    insert into "ReadReceipt" ("userId", "userType", "conversationId", "lastReadMessageId", "updatedAt") 
+    select ${staffId}, ${authorType}, ${conversationId}, "Message"."id", now()
+    from "Message"
+    where "conversationId" = ${conversationId} 
+    order by "createdAt" desc 
+    limit 1
+    on conflict ("userId", "userType", "conversationId") 
+    do 
+      update
+      set "lastReadMessageId" = (
+        select "Message".id 
+        from "Message"
+        where "Message"."conversationId" = ${conversationId}
+        order by "createdAt" desc 
+        limit 1
+        ),
+        "updatedAt" = now() 
+    returning *
+  `
+
+  if(result.length !== 0) {
+    const readReceiptObj = transformToPrismaDAO(result[0])
+    await pubsub.publish('read_receipt', {
+      ['readReceipt']: readReceiptObj,
+      action: isFreshRecord(readReceiptObj) ? 'create' : 'update'
+    })
+  }
+
+  return result[0]
+}
+
 
 export function activities(staff: Staff) {
 
@@ -60,32 +113,7 @@ export function activities(staff: Staff) {
       await triggerSunshineActivity(conv.sunshineConversationId, 'conversation:read');
     }
 
-    const [lastMessage]= await crud.getMessages({
-      limit: 1,
-      order: 'desc',
-      conversationId: conversationId
-    });
-
-    if (!lastMessage) return null;
-
-    return db.readReceipt.upsert({
-      where: {
-        userId_userType_conversationId: {
-          userId: staff.id,
-          userType: AuthorType.STAFF,
-          conversationId: conversationId
-        }
-      },
-      update: {
-        lastReadMessageId: lastMessage.id
-      },
-      create: {
-        lastReadMessageId: lastMessage.id,
-        userId: staff.id,
-        userType: AuthorType.STAFF,
-        conversationId: conversationId
-      }
-    })
+    return createOrUpdateReadReceipt(conversationId, staff.id, AuthorType.STAFF)
   }
 
   return {

lib/services/events/pubsub.ts

@@ -90,7 +90,7 @@ const SUPPORTED_MODELS = _.keys(EVENT_PER_MODEL);
  * @param {*} obj
  * @returns {boolean}
  */
-function isFreshRecord(obj?: any) : boolean {
+export function isFreshRecord(obj?: any) : boolean {
   const createdAt = obj?.createdAt;
   const updatedAt = obj?.updatedAt;