Fixed Queue Conflicts

Author: 1AhmedYasser

DSL/Liquibase/changelog/1711445550-add-hstore.sql

@@ -0,0 +1,3 @@
+-- liquibase formatted sql
+-- changeset ahmedyasser:1711445550
+CREATE EXTENSION IF NOT EXISTS hstore;

DSL/OpenSearch/fieldMappings/chatqueue.json

@@ -0,0 +1,12 @@
+{
+  "mappings": {
+    "properties": {
+      "chatId": {
+        "type": "keyword"
+      },
+      "timestamp": {
+        "type": "keyword"
+      }
+    }
+  }
+}

DSL/OpenSearch/init.sh

@@ -4,3 +4,7 @@ curl -XDELETE 'http://localhost:9200/*' -u admin:admin --insecure
 #notifications
 curl -H "Content-Type: application/x-ndjson" -X PUT "http://localhost:9200/notifications" -ku admin:admin --data-binary "@fieldMappings/notifications.json"
 curl -H "Content-Type: application/x-ndjson" -X PUT "http://localhost:9200/notifications/_bulk" -ku admin:admin --data-binary "@mock/notifications.json"
+
+#chatqueue
+curl -H "Content-Type: application/x-ndjson" -X PUT "http://localhost:9200/chatqueue" -ku admin:admin --data-binary "@fieldMappings/chatqueue.json"
+curl -H "Content-Type: application/x-ndjson" -X PUT "http://localhost:9200/chatqueue/_bulk" -ku admin:admin --data-binary "@mock/chatqueue.json"

DSL/OpenSearch/mock/chatqueue.json

@@ -0,0 +1,2 @@
+{"index":{"_id":"1"}}
+{"chatId": "c307cb64-f2c0-4465-879c-762c26666122","timestamp": "1801371325497"}

DSL/Ruuter.public/DSL/GET/mock-tara-cookie.yml

@@ -0,0 +1,26 @@
+generate_cookie:
+  call: http.post
+  args:
+    url: "[#CHATBOT_TIM]/jwt/custom-jwt-generate"
+    body:
+      JWTName: "JWTTOKEN"
+      expirationInMinutes: 120
+      content: {
+        "personalCode": "EE30303039917",
+        "idCode": "EE30303039917",
+        "login": "EE30303039917",
+        "displayName": "David",
+        "firstName": "David",
+        "lastName": "Shawn",
+        "csaEmail": "d.shawn@gmail.com",
+        "csaTitle": "David User",
+        "authorities": [
+            "ROLE_ADMINISTRATOR"
+        ],
+        "authMethod": "smart-id",
+        "fullName": "David Shawn"
+      }
+  result: cookie_result
+
+return_result:
+  return: ${cookie_result.response.body.token}

migrate.sh

@@ -1,2 +1,2 @@
 #!/bin/bash
-docker run --rm --network bykstack -v `pwd`/DSL/Liquibase/changelog:/liquibase/changelog -v `pwd`/DSL/Liquibase/master.yml:/liquibase/master.yml -v `pwd`/DSL/Liquibase/data:/liquibase/data liquibase/liquibase --defaultsFile=/liquibase/changelog/liquibase.properties --changelog-file=master.yml --url=jdbc:postgresql://users_db:5432/byk?user=byk --password=01234 update
+docker run --rm --network bykstack -v `pwd`/DSL/Liquibase/changelog:/liquibase/changelog -v `pwd`/DSL/Liquibase/master.yml:/liquibase/master.yml -v `pwd`/DSL/Liquibase/data:/liquibase/data liquibase/liquibase --defaultsFile=/liquibase/changelog/liquibase.properties --changelog-file=master.yml --url=jdbc:postgresql://171.22.247.13:5433/byk?user=byk --password=2nH09n6Gly update

notification-server/src/openSearch.js

@@ -51,9 +51,9 @@ async function markAsSent({ _index, _id }, connectionId) {
 }
 
 async function enqueueChatId(chatId) {
-  if(await findChatId(chatId)) return;
-  
-  await client.index({ 
+  if (await findChatId(chatId)) return;
+
+  await client.index({
     index: openSearchConfig.chatQueueIndex,
     body: {
       chatId,
@@ -76,13 +76,13 @@ async function dequeueChatId(chatId) {
       },
     },
     refresh: true,
+    conflicts: "proceed",
   });
 }
 
 async function findChatId(chatId) {
   const found = await isQueueIndexExists();
-  if(!found)
-    return null;
+  if (!found) return null;
 
   const response = await client.search({
     index: openSearchConfig.chatQueueIndex,
@@ -97,24 +97,23 @@ async function findChatId(chatId) {
     },
   });
 
-  if(response.body.hits.hits.length == 0)
-    return null;
+  if (response.body.hits.hits.length == 0) return null;
 
   return response.body.hits.hits[0]._source;
-};
+}
 
 async function isQueueIndexExists() {
   const res = await client.indices.exists({
-    index: openSearchConfig.chatQueueIndex
+    index: openSearchConfig.chatQueueIndex,
   });
 
   return res.body;
 }
 
 async function findChatIdOrder(chatId) {
   const found = await findChatId(chatId);
-  if(!found) return 0;
-  
+  if (!found) return 0;
+
   const response = await client.search({
     index: openSearchConfig.chatQueueIndex,
     body: {
@@ -130,7 +129,7 @@ async function findChatIdOrder(chatId) {
   });
 
   return response.body.hits.total.value + 1;
-};
+}
 
 module.exports = {
   searchNotification,

notification-server/src/server.js

@@ -2,8 +2,11 @@ const express = require("express");
 const cors = require("cors");
 const { buildSSEResponse } = require("./sseUtil");
 const { serverConfig } = require("./config");
-const { buildNotificationSearchInterval, buildQueueCounter } = require("./addOns");
-const { enqueueChatId, dequeueChatId } = require('./openSearch');
+const {
+  buildNotificationSearchInterval,
+  buildQueueCounter,
+} = require("./addOns");
+const { enqueueChatId, dequeueChatId } = require("./openSearch");
 
 const app = express();
 
@@ -21,21 +24,21 @@ app.get("/sse/notifications/:channelId", (req, res) => {
 
 app.get("/sse/queue/:id", (req, res) => {
   const { id } = req.params;
-  buildSSEResponse({ 
+  buildSSEResponse({
     req,
     res,
     buildCallbackFunction: buildQueueCounter({ id }),
-   });
+  });
 });
 
-app.post("/enqueue", (req, res) => {
-  enqueueChatId(req.body.id);
-  res.sendStatus(200);
+app.post("/enqueue", async (req, res) => {
+  await enqueueChatId(req.body.id);
+  res.status(200).json({ response: `enqueued successfully` });
 });
 
-app.post("/dequeue", (req, res) => {
-  dequeueChatId(req.body.id);
-  res.sendStatus(200);
+app.post("/dequeue", async (req, res) => {
+  await dequeueChatId(req.body.id);
+  res.status(200).json({ response: `dequeued successfully` });
 });
 
 const server = app.listen(serverConfig.port, () => {