Merge pull request #746 from SprocketBot/codex/postgres-queue-state

Author: gankoji

clients/discord-bot/src/main.ts

@@ -1,6 +1,5 @@
 import {HttpAdapterHost, NestFactory} from "@nestjs/core";
-import {Transport} from "@nestjs/microservices";
-import {AllExceptionsFilter, config} from "@sprocketbot/common";
+import {AllExceptionsFilter, config, PostgresServer} from "@sprocketbot/common";
 import fetch from "node-fetch";
 
 import {AppModule} from "./app.module";
@@ -10,16 +9,8 @@ global.fetch = fetch as any;
 
 async function bootstrap(): Promise<void> {
     const app = await NestFactory.createMicroservice(AppModule, {
-        transport: Transport.RMQ,
         logger: config.logger.levels,
-        options: {
-            urls: [config.transport.url],
-            queue: config.transport.bot_queue,
-            queueOptions: {
-                durable: true,
-            },
-            heartbeat: 120,
-        },
+        strategy: new PostgresServer({queue: config.transport.bot_queue}),
     });
 
     const httpAdapter = app.get(HttpAdapterHost);

clients/image-generation-frontend/package.json

@@ -14,7 +14,6 @@
     "@commitlint/cli": "^12.1.4",
     "@commitlint/config-conventional": "^12.1.4",
     "@sveltejs/kit": "^1.0.0-next.156",
-    "@types/amqplib": "^0.8.2",
     "@typescript-eslint/eslint-plugin": "^5.10.1",
     "@typescript-eslint/parser": "^5.10.1",
     "autoprefixer": "^10.3.3",
@@ -35,7 +34,6 @@
     "@sveltejs/adapter-node": "^1.0.0-next.43",
     "@types/knex": "^0.16.1",
     "@types/minio": "^7.0.10",
-    "amqplib": "^0.8.0",
     "knex": "^0.95.10",
     "minio": "^7.0.19",
     "nats": "^2.2.0",

clients/image-generation-frontend/src/routes/api/exec/[id].ts

@@ -1,14 +1,14 @@
 // import type {EndpointOutput, Request} from "@sveltejs/kit";
 import { ReportTemplateDAO } from '$utils/server/database/ReportTemplate.dao';
-import { rmqRequest } from '$utils/rabbitmq';
+import { postgresRpcRequest } from '$utils/postgres-rpc';
 
 export async function POST({ request, params }) {
   const data = await request.json();
   // const data = JSON.parse(body.toString());
   const results = await ReportTemplateDAO.runReport(params.id, data.filterValues);
 
   try {
-    const res: any = await rmqRequest('media-gen.img.create', {
+    const res: any = await postgresRpcRequest('media-gen.img.create', {
       inputFile: data.inputFile,
       outputFile: data.outputFile,
       template: results,

clients/image-generation-frontend/src/routes/api/run/[imageType]/[name].ts

@@ -1,4 +1,4 @@
-import { rmqRequest } from '$src/utils/rabbitmq';
+import { postgresRpcRequest } from '$src/utils/postgres-rpc';
 import { ReportTemplateDAO } from '$src/utils/server/database/ReportTemplate.dao';
 import type { EndpointOutput, Request } from '@sveltejs/kit';
 
@@ -40,7 +40,7 @@ export async function POST({ body, params }: Request): Promise<EndpointOutput> {
   const results = await ReportTemplateDAO.runReport(params.id, data.filterValues);
 
   try {
-    const res: any = await rmqRequest('media-gen.img.create', {
+    const res: any = await postgresRpcRequest('media-gen.img.create', {
       inputFile: data.inputFile,
       outputFile: data.outputFile,
       filterValues: results,

clients/image-generation-frontend/src/utils/postgres-rpc.ts

@@ -0,0 +1,44 @@
+import pg from 'pg';
+import { v4 } from 'uuid';
+import config from '../config';
+
+const { Pool } = pg;
+
+const pool = new Pool({
+  ...config.knex,
+  ssl: false,
+});
+
+export async function postgresRpcRequest(
+  pattern: string,
+  data: Record<string, unknown> = {},
+): Promise<unknown> {
+  const id = v4();
+  await pool.query('CREATE SCHEMA IF NOT EXISTS sprocket');
+  await pool.query(
+    `
+      INSERT INTO sprocket.platform_rpc_queue (id, queue, pattern, payload)
+      VALUES ($1, $2, $3, $4)
+    `,
+    [id, config.transport.queue, pattern, data],
+  );
+
+  const deadline = Date.now() + 30000;
+  while (Date.now() < deadline) {
+    const result = await pool.query(
+      'SELECT status, response, error FROM sprocket.platform_rpc_queue WHERE id = $1',
+      [id],
+    );
+    const row = result.rows[0];
+    if (row?.status === 'completed') {
+      await pool.query('DELETE FROM sprocket.platform_rpc_queue WHERE id = $1', [id]);
+      return row.response;
+    }
+    if (row?.status === 'failed') {
+      throw new Error(row.error?.message ?? 'Request failed');
+    }
+    await new Promise(resolve => setTimeout(resolve, 250));
+  }
+
+  throw new Error('Request timed out.');
+}

clients/image-generation-frontend/src/utils/rabbitmq.ts

@@ -27,6 +27,7 @@
     "@commitlint/config-conventional": "^15.0.0",
     "@types/config": "^0.0.40",
     "@types/minio": "^7.0.12",
+    "@types/pg": "^8.20.0",
     "@types/uuid": "^8.3.4",
     "husky": "^7.0.4",
     "rimraf": "^3.0.2",
@@ -39,19 +40,16 @@
     "@nestjs/common": "^8.2.3",
     "@nestjs/microservices": "^8.2.3",
     "@sprocketbot/gql-client": "^0.0.3",
-    "@types/amqplib": "^0.8.2",
     "@types/ioredis": "^4.28.5",
     "@types/jest": "^29.0.3",
     "@types/node": "^17.0.13",
-    "amqp-connection-manager": "^3.7.0",
-    "amqplib": "^0.8.0",
-    "celery-node": "^0.5.8",
     "config": "^3.3.6",
     "dotenv": "^16.0.3",
     "eslint-plugin-simple-import-sort": "^7.0.0",
     "ioredis": "^4.28.2",
     "minio": "^7.0.26",
     "nanoid": "^3.3.4",
+    "pg": "^8.7.1",
     "redis": "^4.0.4",
     "rxjs": "^7.4.0",
     "uuid": "^8.3.2",

common/package.json

@@ -1,8 +1,10 @@
 import {Module} from "@nestjs/common";
 
+import {PostgresModule} from "../postgres";
 import {CeleryService} from "./celery.service";
 
 @Module({
+    imports: [PostgresModule],
     providers: [CeleryService],
     exports: [CeleryService],
 })