Apply Nginx as a caching proxy to serve SSE requests (buerokratt#380)

* Add nginx service

* Handle SSE connection in nginx

* Add node server to server SSEs

* Rename opensearch index

* Add notification service

* Update opensearch index

* Create notification backend

* Serve sse and connect to OpenSearch

* Remove console.log

* Remove timestamp filtering

* Solve OpenSearch update conflict

* Only send payload back to notification listeners

* Change pyload type to obect

* Refactor & clean up

* Remove console.log

* Add nginx mapping for OpenSearch

Author: baha-a

.gitignore

@@ -3,3 +3,4 @@
 /tim-db
 .DS_Store
 /loba
+node_modules

DSL.OpenSearch/fieldMappings/sse-events.json DSL.OpenSearch/fieldMappings/notifications.json

@@ -1,17 +1,17 @@
 {
   "mappings": {
     "properties": {
-      "event_id": {
+      "channelId": {
         "type": "keyword"
       },
-      "event_type": {
-        "type": "keyword"
-      },
-      "data": {
-        "type": "text"
+      "payload": {
+        "type": "object"
       },
       "timestamp": {
-        "type": "date"
+        "type": "long"
+      },
+      "sentTo": {
+        "type": "keyword"
       }
     }
   }

DSL.OpenSearch/init.sh

@@ -1,7 +1,6 @@
 #clear
 curl -XDELETE 'http://localhost:9200/*' -u admin:admin --insecure
 
-#SSE events
-curl -H "Content-Type: application/x-ndjson" -X PUT "http://localhost:9200/sse-events" -ku admin:admin --data-binary "@fieldMappings/sse-events.json"
-curl -H "Content-Type: application/x-ndjson" -X PUT "http://localhost:9200/sse-events/_bulk" -ku admin:admin --data-binary "@mock/sse-events.json"
-curl -L -X POST 'http://localhost:9200/_scripts/sse-events-by-type' -H 'Content-Type: application/json' -H 'Cookie: customJwtCookie=test' --data-binary "@templates/sse-events-by-type.json"
+#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"

DSL.OpenSearch/mock/notifications.json

@@ -0,0 +1,8 @@
+{"index":{"_id":"1"}}
+{"channelId": "b794c417-7f90-4e4e-b7e5-16f41614c4fe","payload": {"content":"This is a notification for user 1"},"timestamp": 1801371325497, "sentTo": []}
+{"index":{"_id":"2"}}
+{"channelId": "9f6d91f8-ee5a-4d68-9c79-f71f5cda2c57","payload": {"content":"Notification for user 2"},"timestamp": 1801371325597, "sentTo": []}
+{"index":{"_id":"3"}}
+{"channelId": "b794c417-7f90-4e4e-b7e5-16f41614c4fe","payload": {"content":"Another notification for user 1"},"timestamp": 1801371325697, "sentTo": []}
+{"index":{"_id":"4"}}
+{"channelId": "11111111-7f90-4e4e-b7e5op-111111111111","payload": {"content":"Again, another notification for user 1"},"timestamp": 1801371325797, "sentTo": []}

DSL.OpenSearch/mock/sse-events.json

@@ -18,3 +18,7 @@ CHATBOT_TIM=http://tim
 CHATBOT_TIM_PORT=8085
 CHATBOT_OPENSEARCH=http://opensearch-node
 CHATBOT_OPENSEARCH_PORT=9200
+CHATBOT_NOTIFICATIONS=http://notifications-node
+CHATBOT_NOTIFICATIONS_PORT=4040
+CHATBOT_NGINX=http://nginx-node
+CHATBOT_NGINX_PORT=80

DSL.OpenSearch/templates/sse-events-by-type.json

@@ -273,6 +273,42 @@ services:
     networks:
       - bykstack
 
+  notifications-node:
+    container_name: notifications-node
+    build:
+      context: ./notification-server
+      dockerfile: Dockerfile
+    ports:
+      - 4040:4040
+    depends_on:
+      - opensearch-node
+    environment:
+      OPENSEARCH_PROTOCOL: http
+      OPENSEARCH_HOST: opensearch-node
+      OPENSEARCH_PORT: 9200
+      OPENSEARCH_USERNAME: admin
+      OPENSEARCH_PASSWORD: admin
+      PORT: 4040
+      REFRESH_INTERVAL: 1000
+    volumes:
+      - ./notification-server:/app
+      - ./notification-server/node_modules:/app/node_modules
+    command: ["npm", "run", "dev"]
+    networks:
+      - bykstack
+
+  nginx:
+    image: nginx:latest
+    container_name: nginx-node
+    ports:
+      - "80:80"
+    volumes:
+      - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
+    depends_on:
+      - notifications-node
+    networks:
+      - bykstack
+  
 volumes:
   opensearch-data:
 networks:

constants.ini

@@ -0,0 +1,48 @@
+
+worker_processes auto;
+
+events {
+    worker_connections 1024;
+}
+
+http {
+    include mime.types;
+    default_type application/octet-stream;
+    sendfile on;
+
+    proxy_buffering off;
+    proxy_request_buffering off;
+    proxy_http_version 1.1;
+    keepalive_timeout 65;
+    keepalive_requests 1000;
+
+    server {
+        listen 80;
+
+        location /sse {
+            proxy_pass http://notifications-node:4040/sse;
+
+            proxy_http_version 1.1;
+            proxy_set_header Upgrade $http_upgrade;
+            proxy_set_header Connection "upgrade";
+            proxy_set_header Host $host;
+            proxy_set_header X-Real-IP $remote_addr;
+            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+            proxy_set_header X-Forwarded-Proto $scheme;
+
+            proxy_buffering off;
+            proxy_cache off;
+            proxy_redirect off;
+            proxy_read_timeout 24h;
+        }
+        
+        location /notifications {
+            proxy_pass http://opensearch-node:9200/notifications;
+            
+            proxy_set_header Host $host;
+            proxy_set_header X-Real-IP $remote_addr;
+            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+            proxy_set_header X-Forwarded-Proto $scheme;
+        }
+    }
+}

docker-compose.yml

@@ -0,0 +1,7 @@
+OPENSEARCH_PROTOCOL=http
+OPENSEARCH_HOST=opensearch-node
+OPENSEARCH_PORT=9200
+OPENSEARCH_USERNAME=admin
+OPENSEARCH_PASSWORD=admin
+PORT=4040
+REFRESH_INTERVAL=1000

nginx/nginx.conf

@@ -0,0 +1,13 @@
+FROM node:lts-slim
+
+WORKDIR /app
+
+COPY package.json package-lock.json /app/
+
+RUN npm install
+
+COPY . /app/
+
+EXPOSE 4040
+
+CMD ["npm", "start"]

notification-server/.env

@@ -0,0 +1,17 @@
+require('dotenv').config();
+const { client } = require('./src/openSearch');
+
+(async () => {
+  try {
+    await client.indices.putSettings({
+      index: 'notifications',
+      body: {
+        refresh_interval: '5s',
+      },
+    });
+
+    require('./src/server');
+  } catch (error) {
+    console.error('Error:', error);
+  }
+})();