.deploy/deploy-dev.yaml

@@ -8,8 +8,7 @@ services:
     environment:
       STAGE: "development"
       TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
-      - queues
     command: poll
     deploy:
       mode: replicated
@@ -26,8 +25,7 @@ services:
       STAGE: "development"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
-      - queues
     command: worker
     deploy:
       mode: replicated
@@ -42,8 +40,7 @@ services:
     environment:
       STAGE: "development"
       TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
-      - queues
     command: mailbox
     deploy:
       mode: replicated
@@ -60,8 +57,7 @@ services:
       STAGE: "development"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
-      - queues
     command: fetch
     deploy:
       mode: replicated
@@ -78,8 +74,7 @@ services:
       STAGE: "development"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
-      - queues
     command: notify
     deploy:
       mode: replicated
@@ -98,8 +93,7 @@ services:
       STAGE: "development"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
-      - queues
     command: api
     deploy:
       mode: replicated
@@ -112,5 +106,3 @@ services:
 networks:
   common-infra-nginx:
     external: true
-  queues:
-    external: true

.deploy/deploy-prod.yaml

@@ -8,8 +8,7 @@ services:
     environment:
       STAGE: "production"
       TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
-      - queues
     command: poll
     deploy:
       mode: replicated
@@ -26,8 +25,7 @@ services:
       STAGE: "production"
       MONGO_PASSWORD: $MONGO_PASSWORD_PROD
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
-      - queues
     command: worker
     deploy:
       mode: replicated
@@ -42,8 +40,7 @@ services:
     environment:
       STAGE: "production"
       TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
-      - queues
     command: mailbox
     deploy:
       mode: replicated
@@ -61,8 +58,7 @@ services:
       MONGO_PASSWORD: $MONGO_PASSWORD_PROD
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
       DEBUG: "false"
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
-      - queues
     command: fetch
     deploy:
       mode: replicated
@@ -84,8 +80,7 @@ services:
       MONGO_PASSWORD: $MONGO_PASSWORD_PROD
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
       DEBUG: "false"
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
-      - queues
     command: notify
     deploy:
       mode: replicated
@@ -125,5 +120,3 @@ services:
 networks:
   common-infra-nginx:
     external: true
-  queues:
-    external: true

.gitea/workflows/deploy-dev.yaml

@@ -42,4 +42,5 @@ jobs:
           TELEGRAM_TOKEN_DEV: ${{ secrets.TELEGRAM_TOKEN_DEV }}
           MONGO_PASSWORD_DEV: ${{ secrets.MONGO_PASSWORD_DEV }}
           PLATFORM_SECURITY_TOKEN: ${{ secrets.PLATFORM_SECURITY_TOKEN }}
+          QUEUES_TOKEN_DEV: ${{ secrets.QUEUES_TOKEN_DEV }}
         run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-dev.yaml ruz-bot

.gitea/workflows/deploy-prod.yaml

@@ -42,4 +42,5 @@ jobs:
           TELEGRAM_TOKEN_PROD: ${{ secrets.TELEGRAM_TOKEN_PROD }}
           MONGO_PASSWORD_PROD: ${{ secrets.MONGO_PASSWORD_PROD }}
           PLATFORM_SECURITY_TOKEN: ${{ secrets.PLATFORM_SECURITY_TOKEN }}
+          QUEUES_TOKEN_PROD: ${{ secrets.QUEUES_TOKEN_PROD }}
         run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-prod.yaml ruz-bot

utils/queues.py

@@ -1,13 +1,18 @@
+import json
 import os
 import requests
 import time
 
 
 stage = os.getenv("STAGE", 'local')
-if stage == 'local':
+if stage == 'development':
-    QUEUES_URL = 'http://localhost:1239'
+    QUEUES_URL = 'https://queues.develop.sprinthub.ru'
+elif stage == 'production':
+    QUEUES_URL = 'https://queues.sprinthub.ru'
 else:
-    QUEUES_URL = 'http://queues:1239'
+    QUEUES_URL = None
+
+token = os.getenv('QUEUES_TOKEN')
 
 
 class QueuesException(Exception):
@@ -17,16 +22,23 @@ class QueuesException(Exception):
 class TasksHandlerMixin:
     def poll(self):
         while True:
-            response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name})
+            if QUEUES_URL is None:
-            if response.status_code == 404:
+                data = {'payload': json.loads(input('Input message: '))}
-                time.sleep(0.2)
+            else:
-                continue
+                response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name, 'X-Queues-Token': token})
-            data = response.json()
+                if response.status_code == 404:
+                    time.sleep(0.2)
+                    continue
+                if response.status_code == 403:
+                    raise NotImplemented('QUEUE_TOKEN is incorrect')
+                data = response.json()
             try:
                 self.process(data['payload'])
             except Exception as exc:
                 print(f'Error processing message id={data["id"]}, payload={data["payload"]}, exc={exc}')
                 continue
+            if QUEUES_URL is None:
+                continue
             try:
                 resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': data['id']}, headers={'X-Queues-Token': token})
                 if resp.status_code != 202:
@@ -43,7 +55,7 @@ class TasksHandlerMixin:
 
 
 def set_task(queue_name: str, payload: dict, seconds_to_execute: int, delay: int|None = None):
-    resp = requests.post(f'{QUEUES_URL}/api/v1/put', headers={'queue': queue_name}, json={
+    resp = requests.post(f'{QUEUES_URL}/api/v1/put', headers={'queue': queue_name, 'X-Queues-Token': token}, json={
         'payload': payload,
         'seconds_to_execute': seconds_to_execute,
         'delay': delay,