Merge pull request #719 from constructive-io/fix/job-upgrade-v5

Zetazzz · web-flow · commit db546ecb450f · 2026-02-14T13:36:39.000+08:00
fixed job svc, work again after upgrade to v5
diff --git a/Dockerfile b/Dockerfile
@@ -1,5 +1,5 @@
 ARG BASE=node
-ARG BASE_VERSION=20-bookworm
+ARG BASE_VERSION=22-bookworm
 FROM ${BASE}:${BASE_VERSION} AS build
 
 LABEL org.opencontainers.image.source="https://github.com/constructive-io/constructive"
diff --git a/jobs/job-scheduler/package.json b/jobs/job-scheduler/package.json
@@ -39,6 +39,6 @@
     "@constructive-io/job-pg": "workspace:^",
     "@constructive-io/job-utils": "workspace:^",
     "@pgpmjs/logger": "workspace:^",
-    "node-schedule": "1.3.2"
+    "node-schedule": "^2.1.1"
   }
 }
diff --git a/jobs/job-scheduler/src/index.ts b/jobs/job-scheduler/src/index.ts
@@ -199,59 +199,55 @@ export default class Scheduler {
       }
     }
   }
-  listen() {
+  async listen(): Promise<void> {
     if (this.stopped) return;
-    const listenForChanges = (
-      err: Error | null,
-      client: PoolClient,
-      release: () => void
-    ) => {
-      if (err) {
-        log.error('Error connecting with notify listener', err);
-        if (err instanceof Error && err.stack) {
-          log.debug(err.stack);
-        }
-        // Try again in 5 seconds
-        // should this really be done in the node process?
-        if (!this.stopped) {
-          setTimeout(this.listen, 5000);
-        }
-        return;
+    let client: PoolClient;
+    let release: () => void;
+    try {
+      client = await this.pgPool.connect();
+      release = () => client.release();
+    } catch (err) {
+      log.error('Error connecting with notify listener', err);
+      if (err instanceof Error && err.stack) {
+        log.debug(err.stack);
+      }
+      if (!this.stopped) {
+        setTimeout(() => this.listen(), 5000);
       }
+      return;
+    }
+    if (this.stopped) {
+      release();
+      return;
+    }
+    this.listenClient = client;
+    this.listenRelease = release;
+    client.on('notification', () => {
+      log.info('a NEW scheduled JOB!');
+      if (this.doNextTimer) {
+        // Must be idle, do something!
+        this.doNext(client);
+      }
+    });
+    client.query('LISTEN "scheduled_jobs:insert"');
+    client.on('error', (e: unknown) => {
       if (this.stopped) {
         release();
         return;
       }
-      this.listenClient = client;
-      this.listenRelease = release;
-      client.on('notification', () => {
-        log.info('a NEW scheduled JOB!');
-        if (this.doNextTimer) {
-          // Must be idle, do something!
-          this.doNext(client);
-        }
-      });
-      client.query('LISTEN "scheduled_jobs:insert"');
-      client.on('error', (e: unknown) => {
-        if (this.stopped) {
-          release();
-          return;
-        }
-        log.error('Error with database notify listener', e);
-        if (e instanceof Error && e.stack) {
-          log.debug(e.stack);
-        }
-        release();
-        if (!this.stopped) {
-          this.listen();
-        }
-      });
-      log.info(
-        `${this.workerId} connected and looking for scheduled jobs...`
-      );
-      this.doNext(client);
-    };
-    this.pgPool.connect(listenForChanges);
+      log.error('Error with database notify listener', e);
+      if (e instanceof Error && e.stack) {
+        log.debug(e.stack);
+      }
+      release();
+      if (!this.stopped) {
+        this.listen();
+      }
+    });
+    log.info(
+      `${this.workerId} connected and looking for scheduled jobs...`
+    );
+    this.doNext(client);
   }
 
   async stop(): Promise<void> {
diff --git a/jobs/knative-job-fn/package.json b/jobs/knative-job-fn/package.json
@@ -37,7 +37,6 @@
   },
   "dependencies": {
     "@pgpmjs/logger": "workspace:^",
-    "body-parser": "1.19.0",
     "express": "5.2.1"
   }
 }
diff --git a/jobs/knative-job-fn/src/index.ts b/jobs/knative-job-fn/src/index.ts
@@ -1,5 +1,4 @@
 import express from 'express';
-import bodyParser from 'body-parser';
 import http from 'node:http';
 import https from 'node:https';
 import { URL } from 'node:url';
@@ -131,7 +130,7 @@ const logger = createLogger('knative-job-fn');
 const createJobApp = () => {
   const app: any = express();
 
-  app.use(bodyParser.json());
+  app.use(express.json());
 
   // Basic request logging for all incoming job invocations.
   app.use((req: any, res: any, next: any) => {
diff --git a/jobs/knative-job-server/package.json b/jobs/knative-job-server/package.json
@@ -42,7 +42,6 @@
     "@constructive-io/job-pg": "workspace:^",
     "@constructive-io/job-utils": "workspace:^",
     "@pgpmjs/logger": "workspace:^",
-    "body-parser": "1.19.0",
     "express": "5.2.1",
     "pg": "8.17.1"
   }
diff --git a/jobs/knative-job-server/src/index.ts b/jobs/knative-job-server/src/index.ts
@@ -1,5 +1,4 @@
 import express from 'express';
-import bodyParser from 'body-parser';
 import type { Pool, PoolClient } from 'pg';
 import * as jobs from '@constructive-io/job-utils';
 import poolManager from '@constructive-io/job-pg';
@@ -39,7 +38,7 @@ const logger = createLogger('knative-job-server');
 
 export default (pgPool: Pool = poolManager.getPool()) => {
   const app = express();
-  app.use(bodyParser.json());
+  app.use(express.json());
 
   const withClient =
     (cb: WithClientHandler) =>
diff --git a/jobs/knative-job-worker/package.json b/jobs/knative-job-worker/package.json
@@ -40,8 +40,7 @@
     "@constructive-io/job-pg": "workspace:^",
     "@constructive-io/job-utils": "workspace:^",
     "@pgpmjs/logger": "workspace:^",
-    "pg": "8.17.1",
-    "request": "2.88.2"
+    "pg": "8.17.1"
   },
   "devDependencies": {
     "@pgpm/database-jobs": "^0.16.0",
diff --git a/jobs/knative-job-worker/src/index.ts b/jobs/knative-job-worker/src/index.ts
@@ -183,56 +183,52 @@ export default class Worker {
       }
     }
   }
-  listen() {
+  async listen(): Promise<void> {
     if (this.stopped) return;
-    const listenForChanges = (
-      err: Error | null,
-      client: PoolClient,
-      release: () => void
-    ) => {
-      if (err) {
-        log.error('Error connecting with notify listener', err);
-        if (err instanceof Error && err.stack) {
-          log.debug(err.stack);
-        }
-        // Try again in 5 seconds
-        // should this really be done in the node process?
-        if (!this.stopped) {
-          setTimeout(this.listen, 5000);
-        }
-        return;
+    let client: PoolClient;
+    let release: () => void;
+    try {
+      client = await this.pgPool.connect();
+      release = () => client.release();
+    } catch (err) {
+      log.error('Error connecting with notify listener', err);
+      if (err instanceof Error && err.stack) {
+        log.debug(err.stack);
+      }
+      if (!this.stopped) {
+        setTimeout(() => this.listen(), 5000);
+      }
+      return;
+    }
+    if (this.stopped) {
+      release();
+      return;
+    }
+    this.listenClient = client;
+    this.listenRelease = release;
+    client.on('notification', () => {
+      if (this.doNextTimer) {
+        // Must be idle, do something!
+        this.doNext(client);
       }
+    });
+    client.query('LISTEN "jobs:insert"');
+    client.on('error', (e: unknown) => {
       if (this.stopped) {
         release();
         return;
       }
-      this.listenClient = client;
-      this.listenRelease = release;
-      client.on('notification', () => {
-        if (this.doNextTimer) {
-          // Must be idle, do something!
-          this.doNext(client);
-        }
-      });
-      client.query('LISTEN "jobs:insert"');
-      client.on('error', (e: unknown) => {
-        if (this.stopped) {
-          release();
-          return;
-        }
-        log.error('Error with database notify listener', e);
-        if (e instanceof Error && e.stack) {
-          log.debug(e.stack);
-        }
-        release();
-        if (!this.stopped) {
-          this.listen();
-        }
-      });
-      log.info(`${this.workerId} connected and looking for jobs...`);
-      this.doNext(client);
-    };
-    this.pgPool.connect(listenForChanges);
+      log.error('Error with database notify listener', e);
+      if (e instanceof Error && e.stack) {
+        log.debug(e.stack);
+      }
+      release();
+      if (!this.stopped) {
+        this.listen();
+      }
+    });
+    log.info(`${this.workerId} connected and looking for jobs...`);
+    this.doNext(client);
   }
 
   async stop(): Promise<void> {
diff --git a/jobs/knative-job-worker/src/req.ts b/jobs/knative-job-worker/src/req.ts
@@ -1,4 +1,6 @@
-import requestLib from 'request';
+import http from 'node:http';
+import https from 'node:https';
+import { URL } from 'node:url';
 import {
   getCallbackBaseUrl,
   getJobGatewayConfig,
@@ -47,35 +49,55 @@ const request = (
     databaseId
   });
   return new Promise<boolean>((resolve, reject) => {
-    requestLib.post(
+    let parsed: URL;
+    try {
+      parsed = new URL(url);
+    } catch (e) {
+      return reject(e);
+    }
+
+    const isHttps = parsed.protocol === 'https:';
+    const client = isHttps ? https : http;
+    const payload = JSON.stringify(body);
+
+    const req = client.request(
       {
+        hostname: parsed.hostname,
+        port: parsed.port || (isHttps ? 443 : 80),
+        path: parsed.pathname + parsed.search,
+        method: 'POST',
         headers: {
           'Content-Type': 'application/json',
+          'Content-Length': Buffer.byteLength(payload),
 
           // these are used by job-worker/job-fn
           'X-Worker-Id': workerId,
-          'X-Job-Id': jobId,
+          'X-Job-Id': String(jobId),
           'X-Database-Id': databaseId,
 
           // async HTTP completion callback
           'X-Callback-Url': completeUrl
-        },
-        url,
-        json: true,
-        body
-      },
-      function (error: unknown) {
-        if (error) {
-          log.error(`request error for job[${jobId}] fn[${fn}]`, error);
-          if (error instanceof Error && error.stack) {
-            log.debug(error.stack);
-          }
-          return reject(error);
         }
-        log.debug(`request success for job[${jobId}] fn[${fn}]`);
-        return resolve(true);
+      },
+      (res) => {
+        res.on('data', () => {});
+        res.on('end', () => {
+          log.debug(`request success for job[${jobId}] fn[${fn}]`);
+          resolve(true);
+        });
       }
     );
+
+    req.on('error', (error) => {
+      log.error(`request error for job[${jobId}] fn[${fn}]`, error);
+      if (error.stack) {
+        log.debug(error.stack);
+      }
+      reject(error);
+    });
+
+    req.write(payload);
+    req.end();
   });
 };
 
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml

Original file line number	Diff line number	Diff line change
`@@ -39,6 +39,6 @@`
`39`	`39`	`"@constructive-io/job-pg": "workspace:^",`
`40`	`40`	`"@constructive-io/job-utils": "workspace:^",`
`41`	`41`	`"@pgpmjs/logger": "workspace:^",`
`42`		`- "node-schedule": "1.3.2"`
	`42`	`+ "node-schedule": "^2.1.1"`
`43`	`43`	`}`
`44`	`44`	`}`
Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,6 @@`
`37`	`37`	`},`
`38`	`38`	`"dependencies": {`
`39`	`39`	`"@pgpmjs/logger": "workspace:^",`
`40`		`- "body-parser": "1.19.0",`
`41`	`40`	`"express": "5.2.1"`
`42`	`41`	`}`
`43`	`42`	`}`
Original file line number	Diff line number	Diff line change
`@@ -42,7 +42,6 @@`
`42`	`42`	`"@constructive-io/job-pg": "workspace:^",`
`43`	`43`	`"@constructive-io/job-utils": "workspace:^",`
`44`	`44`	`"@pgpmjs/logger": "workspace:^",`
`45`		`- "body-parser": "1.19.0",`
`46`	`45`	`"express": "5.2.1",`
`47`	`46`	`"pg": "8.17.1"`
`48`	`47`	`}`