From 1c390db493400ea31329f6efd4d0ad39b150ed27 Mon Sep 17 00:00:00 2001 From: pablohashescobar Date: Thu, 14 Dec 2023 16:34:54 +0530 Subject: [PATCH] dev: setup segway and django connection --- apiserver/plane/bgtasks/__init__.py | 2 +- apiserver/plane/bgtasks/import_create_task.py | 5 -- apiserver/plane/bgtasks/importer_task.py | 44 +++++------------ apiserver/plane/bgtasks/issue_sync_task.py | 5 ++ apiserver/plane/settings/common.py | 27 +++++++--- segway/src/controller/jira.controller.ts | 49 ++++++++++++++++--- segway/src/middleware/authkey.middleware.ts | 22 +++++++++ segway/src/queue/mq.singleton.ts | 25 ++++++---- segway/src/server.ts | 22 --------- segway/src/worker/base.worker.ts | 12 ++--- segway/src/worker/index.ts | 1 - segway/src/worker/jira.worker.ts | 24 --------- 12 files changed, 121 insertions(+), 117 deletions(-) delete mode 100644 apiserver/plane/bgtasks/import_create_task.py create mode 100644 apiserver/plane/bgtasks/issue_sync_task.py create mode 100644 segway/src/middleware/authkey.middleware.ts delete mode 100644 segway/src/worker/index.ts delete mode 100644 segway/src/worker/jira.worker.ts diff --git a/apiserver/plane/bgtasks/__init__.py b/apiserver/plane/bgtasks/__init__.py index 89cc2d9496..d956c99cc7 100644 --- a/apiserver/plane/bgtasks/__init__.py +++ b/apiserver/plane/bgtasks/__init__.py @@ -1 +1 @@ -from .import_create_task import issue_create_task \ No newline at end of file +from .issue_sync_task import issue_sync \ No newline at end of file diff --git a/apiserver/plane/bgtasks/import_create_task.py b/apiserver/plane/bgtasks/import_create_task.py deleted file mode 100644 index 83aa7fe26b..0000000000 --- a/apiserver/plane/bgtasks/import_create_task.py +++ /dev/null @@ -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}") \ No newline at end of file diff --git a/apiserver/plane/bgtasks/importer_task.py b/apiserver/plane/bgtasks/importer_task.py index 313561a8dc..e0e5dc9214 100644 --- a/apiserver/plane/bgtasks/importer_task.py +++ b/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) diff --git a/apiserver/plane/bgtasks/issue_sync_task.py b/apiserver/plane/bgtasks/issue_sync_task.py new file mode 100644 index 0000000000..dac38430cd --- /dev/null +++ b/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}") \ No newline at end of file diff --git a/apiserver/plane/settings/common.py b/apiserver/plane/settings/common.py index 3f04f36a81..8e505859c4 100644 --- a/apiserver/plane/settings/common.py +++ b/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) diff --git a/segway/src/controller/jira.controller.ts b/segway/src/controller/jira.controller.ts index b7ae48be98..c09e85adeb 100644 --- a/segway/src/controller/jira.controller.ts +++ b/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 - */ -} \ No newline at end of file +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" }); + } + } +} diff --git a/segway/src/middleware/authkey.middleware.ts b/segway/src/middleware/authkey.middleware.ts new file mode 100644 index 0000000000..cfa0d1af45 --- /dev/null +++ b/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; diff --git a/segway/src/queue/mq.singleton.ts b/segway/src/queue/mq.singleton.ts index 3d0f2e37b1..c1e47c5259 100644 --- a/segway/src/queue/mq.singleton.ts +++ b/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 { + public async publish(body: object, taskName: string): Promise { + + // 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, diff --git a/segway/src/server.ts b/segway/src/server.ts index 88cb396920..965489618f 100644 --- a/segway/src/server.ts +++ b/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 { 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) => { diff --git a/segway/src/worker/base.worker.ts b/segway/src/worker/base.worker.ts index aa894758c7..e298285eaa 100644 --- a/segway/src/worker/base.worker.ts +++ b/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 { try { this.mq?.consume(this.queueName, this.onMessage); @@ -27,9 +28,10 @@ export abstract class BaseWorker { } } - protected async publish(queueName: string, content: Buffer): Promise { + // Publish this to queue + protected async publish(body: object, taskName: string): Promise { 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; - } } diff --git a/segway/src/worker/index.ts b/segway/src/worker/index.ts deleted file mode 100644 index 54fc515df3..0000000000 --- a/segway/src/worker/index.ts +++ /dev/null @@ -1 +0,0 @@ -export * from "./jira.worker" \ No newline at end of file diff --git a/segway/src/worker/jira.worker.ts b/segway/src/worker/jira.worker.ts deleted file mode 100644 index 23297b0c79..0000000000 --- a/segway/src/worker/jira.worker.ts +++ /dev/null @@ -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); - } - } -}