dev: setup segway and django connection

apiserver/plane/bgtasks/__init__.py

@@ -1 +1 @@
-from .import_create_task import issue_create_task
+from .issue_sync_task import issue_sync

apiserver/plane/bgtasks/import_create_task.py

@@ -1,5 +0,0 @@
-from celery import shared_task
-
-@shared_task(queue="node_to_celery_queue")
-def issue_create_task(x,y):
-    print(f"Received data from Node.js: {x,y}")

apiserver/plane/bgtasks/importer_task.py

@@ -2,7 +2,6 @@
 import json
 import requests
 import uuid
-from kombu import Connection, Exchange, Queue, Producer
 
 # Django imports
 from django.conf import settings
@@ -177,40 +176,21 @@ def service_importer(service, importer_id):
                 project_id=importer.project_id,
             )
 
-        import_data_json = json.dumps(
-            ImporterSerializer(importer).data,
-            cls=DjangoJSONEncoder,
-        )
+        import_data = ImporterSerializer(importer).data
 
-        rabbit_url = settings.RABBITMQ_URL
+        import_data_json = json.dumps(import_data, cls=DjangoJSONEncoder)
 
-        # Establish a connection and send a message
-        with Connection(rabbit_url) as conn:
-            queue_name = "django_to_node_queue"
-            routing_key = "django.node"
-            exchange = Exchange("django_exchange", type="direct")
-            queue = Queue(name=queue_name, exchange=exchange, routing_key=routing_key)
-            queue.maybe_bind(conn)
-            queue.declare()
-            producer = Producer(conn)
-            producer.publish(
-                import_data_json,
-                exchange=exchange,
-                routing_key=routing_key,
-                declare=[queue],
+        if settings.SEGWAY_BASE_URL:
+            headers = {
+                "Content-Type": "application/json",
+                "x-api-key": settings.SEGWAY_KEY,
+            }
+            res = requests.post(
+                f"{settings.SEGWAY_BASE_URL}/api/jira",
+                data=import_data_json,
+                headers=headers,
             )
-        # if settings.PROXY_BASE_URL:
-        #     headers = {"Content-Type": "application/json"}
-        #     import_data_json = json.dumps(
-        #         ImporterSerializer(importer).data,
-        #         cls=DjangoJSONEncoder,
-        #     )
-        #     _ = requests.post(
-        #         f"{settings.PROXY_BASE_URL}/hooks/workspaces/{str(importer.workspace_id)}/projects/{str(importer.project_id)}/importers/{str(service)}/",
-        #         json=import_data_json,
-        #         headers=headers,
-        #     )
-
+            print(res.json())
         return
     except Exception as e:
         print(e)

apiserver/plane/bgtasks/issue_sync_task.py

@@ -0,0 +1,5 @@
+from celery import shared_task
+
+@shared_task(queue="segway_tasks")
+def issue_sync(data):
+    print(f"Received data from Segway: {data}")

apiserver/plane/settings/common.py

@@ -280,9 +280,21 @@ CELERY_BROKER_URL = RABBITMQ_URL
 CELERY_RESULT_BACKEND = REDIS_URL
 
 CELERY_QUEUES = (
-    Queue("internal_tasks", Exchange("internal_exchange"), routing_key="internal"),
-    Queue("external_tasks", Exchange("external_exchange"), routing_key="external"),
-    Queue('node_to_celery_queue', Exchange('node_exchange', type='direct'), routing_key='node.celery'),
+    Queue(
+        "internal_tasks",
+        Exchange("internal_exchange", type="direct"),
+        routing_key="internal",
+    ),
+    Queue(
+        "external_tasks",
+        Exchange("external_exchange", type="direct"),
+        routing_key="external",
+    ),
+    Queue(
+        "segway_tasks",
+        Exchange("segway_exchange", type="direct"),
+        routing_key="segway",
+    ),
 )
 
 CELERY_IMPORTS = (
@@ -331,10 +343,9 @@ USE_MINIO = int(os.environ.get("USE_MINIO", 0)) == 1
 POSTHOG_API_KEY = os.environ.get("POSTHOG_API_KEY", False)
 POSTHOG_HOST = os.environ.get("POSTHOG_HOST", False)
 
-# instance key
-INSTANCE_KEY = os.environ.get(
-    "INSTANCE_KEY", "ae6517d563dfc13d8270bd45cf17b08f70b37d989128a9dab46ff687603333c3"
-)
-
 # Skip environment variable configuration
 SKIP_ENV_VAR = os.environ.get("SKIP_ENV_VAR", "1") == "1"
+
+# Segway
+SEGWAY_BASE_URL = os.environ.get("SEGWAY_BASE_URL", "http://localhost:9000")
+SEGWAY_KEY = os.environ.get("SEGWAY_KEY", False)

segway/src/controller/jira.controller.ts

@@ -1,8 +1,45 @@
+// overnight js
 import { Request, Response } from "express";
-import { Controller, Get } from "@overnightjs/core";
+import { Controller, Post, Middleware } from "@overnightjs/core";
+// mq
+import { MQSingleton } from "../queue/mq.singleton";
+// middleware
+import AuthKeyMiddlware from "../middleware/authkey.middleware";
+
 @Controller("api/jira")
-export class JiraController  {
-    /**
-     * This controller houses all routes for the Jira Importer
-     */
-}
+export class JiraController {
+  /**
+   * This controller houses all routes for the Jira Importer
+   */
+  mq: MQSingleton;
+  constructor(mq: MQSingleton) {
+    this.mq = mq;
+  }
+
+  @Post("")
+  @Middleware([AuthKeyMiddlware])
+  private home(req: Request, res: Response) {
+    try {
+      res.status(200).json({ message: "Hello, Plane Users" });
+
+      // Process Jira message
+      const body = {
+        args: [], // args
+        kwargs: {
+          data: {
+            type: "issue.create",
+            data: {
+              message: "Segway say's Hi",
+            },
+          },
+        }, // kwargs
+        other_data: {}, // other data
+      };
+
+      this.mq?.publish(body, "plane.bgtasks.issue_sync_task.issue_sync");
+      return;
+    } catch (error) {
+      return res.json({ message: "Server error" });
+    }
+  }
+}

segway/src/middleware/authkey.middleware.ts

@@ -0,0 +1,22 @@
+import { RequestHandler } from "express";
+import { logger } from "../utils/logger";
+
+const AuthKeyMiddlware: RequestHandler = (req, res, next) => {
+  // Retrieve the API key from the request header
+  const apiKey = req.headers["x-api-key"];
+
+  // Define the expected API key
+  const expectedApiKey = process.env.SEGWAY_KEY;
+
+  // Check if the API key is present and matches the expected key
+  if (apiKey === expectedApiKey) {
+    // If the key is valid, proceed with the next middleware or route handler
+    next();
+  } else {
+    // If the key is invalid, log the error and send an appropriate response
+    logger.error("Invalid API key");
+    res.status(401).json({ message: "Invalid API key" });
+  }
+};
+
+export default AuthKeyMiddlware;

segway/src/queue/mq.singleton.ts

@@ -1,3 +1,5 @@
+//uuid
+import { v4 as uuidv4 } from "uuid"
 // mq
 import { Connection, Channel, connect, ConsumeMessage } from "amqplib";
 // utils
@@ -38,32 +40,35 @@ export class MQSingleton {
   }
 
   // Send the message to the given queue
-  public async sendToQueue(queue: string, content: Buffer): Promise<void> {
+  public async publish(body: object, taskName: string): Promise<void> {
+
+    // Check if the channel exists
     if (!this.channel) {
       throw new Error("Channel not initialized");
     }
-    const exchange = "node_exchange";
-    const routingKey = "node.celery";
 
-    const body = {
-      args: ["Petr", 30], // args
-      kwargs: {}, // kwargs
-      other_data: {}, // other data
-    };
+    // Initialize the queue variables
+    const queue = "segway_tasks";
+    const exchange = "segway_exchange";
+    const routingKey = "segway";
 
+    // Create this message
     const msg = {
       contentType: "application/json",
       contentEncoding: "utf-8",
       headers: {
-        id: "3149beef-be66-4b0e-ba47-2fc46e4edac3",
-        task: "plane.bgtasks.import_create_task.issue_create_task",
+        id: uuidv4(),
+        task: taskName,
       },
       body: JSON.stringify(body),
     };
 
+    // Assert the queue
     await this.channel.assertExchange(exchange, "direct", { durable: true });
     await this.channel.assertQueue(queue, { durable: true });
     await this.channel.bindQueue(queue, exchange, routingKey);
+
+    // Try publishing the message
     try {
       this.channel.publish(exchange, routingKey, Buffer.from(msg.body), {
         contentType: msg.contentType,

segway/src/server.ts

@@ -7,9 +7,6 @@ import * as Sentry from "@sentry/node";
 import * as Tracing from "@sentry/tracing";
 // controllers
 import * as controllers from "./controller";
-// workers
-import * as workers from "./worker";
-import { BaseWorker } from "./worker/base.worker";
 // middlewares
 import loggerMiddleware from "./middleware/logger.middleware";
 // utils
@@ -17,10 +14,6 @@ import { logger } from "./utils/logger";
 // mq
 import { MQSingleton } from "./queue/mq.singleton";
 
-type WorkerClasses = {
-  [key: string]: { new (): BaseWorker }; // Type assertion that each key is a constructor for a BaseWorker
-};
-
 class ApiServer extends Server {
   private readonly SERVER_STARTED = "🚀 Api server started on port: ";
   SERVER_PORT: number;
@@ -95,26 +88,11 @@ class ApiServer extends Server {
   private async startMQAndWorkers(): Promise<void> {
     try {
       await this.mq?.initialize();
-      this.startWorkers();
     } catch (error) {
       logger.error("Failed to initialize MQ:", error);
     }
   }
 
-  // start all the workers to listen
-  private startWorkers(): void {
-    const workerClasses: WorkerClasses = workers;
-    for (const key in workerClasses) {
-      if (Object.prototype.hasOwnProperty.call(workerClasses, key)) {
-        const WorkerClass = workerClasses[key];
-        const worker = new WorkerClass();
-        worker
-          .start()
-          .catch((err) => logger.error(`Error starting ${key}:`, err));
-      }
-    }
-  }
-
   // This controller will return 404 for not found pages
   private setupNotFoundHandler(): void {
     this.app.use((req, res) => {

segway/src/worker/base.worker.ts

@@ -19,6 +19,7 @@ export abstract class BaseWorker {
     this.onMessage = this.onMessage.bind(this);
   }
 
+  // Start the consumer
   public async start(): Promise<void> {
     try {
       this.mq?.consume(this.queueName, this.onMessage);
@@ -27,9 +28,10 @@ export abstract class BaseWorker {
     }
   }
 
-  protected async publish(queueName: string, content: Buffer): Promise<void> {
+  // Publish this to queue
+  protected async publish(body: object, taskName: string): Promise<void> {
     try {
-      this.mq?.sendToQueue(queueName, content);
+      this.mq?.publish(body, taskName);
     } catch (error) {
       logger.error("Error sending to queue");
     }
@@ -37,10 +39,4 @@ export abstract class BaseWorker {
 
   protected abstract onMessage(msg: ConsumeMessage | null): void;
 
-  protected isRelevantMessage(msg: ConsumeMessage): boolean {
-    console.log(msg)
-    // Check if the message's routing key matches this worker's routing key
-    const messageRoutingKey = msg.properties.headers["routingKey"];
-    return messageRoutingKey === this.routingKey;
-  }
 }

segway/src/worker/index.ts

@@ -1 +0,0 @@
-export * from "./jira.worker"

segway/src/worker/jira.worker.ts

@@ -1,24 +0,0 @@
-// mq
-import { ConsumeMessage } from "amqplib";
-// base worker
-import { BaseWorker } from "./base.worker";
-
-export class JiraImportWorker extends BaseWorker {
-  constructor() {
-    super("django_to_node_queue", "django.node");
-  }
-
-  protected onMessage(msg: ConsumeMessage | null): void {
-    try {
-      // Process Jira message
-      console.log(msg);
-
-      this.publish(
-        "node_to_celery_queue",
-        Buffer.from(JSON.stringify(""))
-      );
-    } catch (error) {
-      console.log(error);
-    }
-  }
-}