PM-1551 - integrate the AI workflow events

package.json

@@ -33,6 +33,7 @@
     "@platformatic/kafka": "^1.12.0",
     "@prisma/client": "^6.3.1",
     "@types/jsonwebtoken": "^9.0.9",
+    "archiver": "^6.0.2",
     "axios": "^1.9.0",
     "class-transformer": "^0.5.1",
     "class-validator": "^0.14.1",
@@ -42,10 +43,10 @@
     "lodash": "^4.17.21",
     "multer": "^2.0.1",
     "nanoid": "~5.1.2",
+    "pg-boss": "^11.0.2",
     "reflect-metadata": "^0.2.2",
     "rxjs": "^7.8.1",
-    "tc-core-library-js": "appirio-tech/tc-core-library-js.git#v3.0.1",
-    "archiver": "^6.0.2"
+    "tc-core-library-js": "appirio-tech/tc-core-library-js.git#v3.0.1"
   },
   "devDependencies": {
     "@eslint/eslintrc": "^3.2.0",

prisma/migrations/20251003103752_rename_workflow_error_to_scheduledjobid/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "aiWorkflowRun" RENAME COLUMN "error" TO "scheduledJobId";

prisma/schema.prisma

@@ -629,7 +629,7 @@ model aiWorkflowRun {
   score        Float?    @db.DoublePrecision
   status       String    @db.VarChar
   usage        Json?     @db.JsonB
-  error        String?   @db.Text
+  scheduledJobId String?   @db.Text
   completedJobs Int? @default(0)
   jobsCount     Int? @default(0)
 

src/api/webhook/webhook.service.ts

@@ -6,6 +6,7 @@ import {
   WebhookEventDto,
   WebhookResponseDto,
 } from '../../dto/webhook-event.dto';
+import { WorkflowQueueHandler } from 'src/shared/modules/global/workflow-queue.handler';
 
 @Injectable()
 export class WebhookService {
@@ -14,6 +15,7 @@ export class WebhookService {
   constructor(
     private readonly prisma: PrismaService,
     private readonly prismaErrorService: PrismaErrorService,
+    private readonly workflowQueueHandler: WorkflowQueueHandler,
   ) {}
 
   async processWebhook(
@@ -45,7 +47,7 @@ export class WebhookService {
       });
 
       // Future extensibility: Add event-specific handlers here
-      this.handleEventSpecificProcessing(
+      await this.handleEventSpecificProcessing(
         webhookEvent.event,
         webhookEvent.eventPayload,
       );
@@ -76,27 +78,33 @@ export class WebhookService {
    * Placeholder for future event-specific processing logic
    * This method can be extended to handle different GitHub events differently
    */
-  private handleEventSpecificProcessing(event: string, payload: any): void {
+  private async handleEventSpecificProcessing(
+    event: string,
+    payload: any,
+  ): Promise<void> {
     this.logger.log({
       message: 'Event-specific processing placeholder',
       event,
       payloadSize: JSON.stringify(payload).length,
     });
 
     // Future implementation examples:
-    // switch (event) {
-    //   case 'push':
-    //     await this.handlePushEvent(payload);
-    //     break;
-    //   case 'pull_request':
-    //     await this.handlePullRequestEvent(payload);
-    //     break;
-    //   case 'issues':
-    //     await this.handleIssuesEvent(payload);
-    //     break;
-    //   default:
-    //     this.logger.log(`No specific handler for event type: ${event}`);
-    // }
+    switch (event) {
+      case 'workflow_job':
+        await this.workflowQueueHandler.handleWorkflowRunEvents(payload);
+        break;
+      // case 'push':
+      //   await this.handlePushEvent(payload);
+      //   break;
+      // case 'pull_request':
+      //   await this.handlePullRequestEvent(payload);
+      //   break;
+      // case 'issues':
+      //   await this.handleIssuesEvent(payload);
+      //   break;
+      default:
+        this.logger.log(`No specific handler for event type: ${event}`);
+    }
   }
 
   /**

src/shared/clients/gitea/gitea.client.ts

@@ -5734,9 +5734,9 @@ export class Api<
       jobId: number,
       params: RequestParams = {},
     ) =>
-      this.request<void, any>({
+      this.request<string, any>({
         path: `/repos/${owner}/${repo}/actions/jobs/${jobId}/logs`,
-        method: "GET",
+        method: 'GET',
         secure: true,
         ...params,
       }),