major jobs refactor

Author: Anmol1696

DEVELOPMENT_JOBS.md

@@ -56,7 +56,7 @@ This ensures all subsequent `pgpm` and `psql` commands point at the same local d
 
 Make sure `pgpm` is installed and up to date.
 
-From the `constructive/` directory (with `pgenv` applied):
+From the `constructive-db/` directory (with `pgenv` applied):
 
 1. Bootstrap admin users:
 

Dockerfile

@@ -23,7 +23,7 @@ COPY . .
 
 # Install and build all workspaces
 RUN set -eux; \
-    pnpm install --frozen-lockfile; \
+    CI=true pnpm install --frozen-lockfile; \
     pnpm run build
 
 ################################################################################

docker-compose.jobs.yml

@@ -2,7 +2,10 @@ services:
   # LaunchQL API server (GraphQL)
   launchql-server:
     container_name: launchql-server
-    image: ghcr.io/constructive-io/launchql:b88e3d1
+    image: constructive-launchql:dev
+    build:
+      context: .
+      dockerfile: ./Dockerfile
     # The image entrypoint already runs the LaunchQL CLI (`lql`).
     # We only need to provide the subcommand and flags here.
     entrypoint: ["lql", "server", "--port", "3000", "--origin", "*", "--strictAuth", "false"]
@@ -34,7 +37,7 @@ services:
   # Simple email function (Knative-style HTTP function)
   simple-email:
     container_name: simple-email
-    image: ghcr.io/constructive-io/launchql:b88e3d1
+    image: constructive-launchql:dev
     # Override the image entrypoint (LaunchQL CLI) and run the Node function directly.
     entrypoint: ["node", "functions/simple-email/dist/index.js"]
     environment:
@@ -43,8 +46,10 @@ services:
       # Mailgun / email provider configuration for @launchql/postmaster
       # Replace with real credentials for local testing.
       MAILGUN_API_KEY: "change-me-mailgun-api-key"
+      MAILGUN_KEY: "change-me-mailgun-api-key"
       MAILGUN_DOMAIN: "mg.constructive.io"
       MAILGUN_FROM: "no-reply@mg.constructive.io"
+      MAILGUN_REPLY: "info@mg.constructive.io"
     ports:
       # Expose function locally (optional)
       - "8081:8080"
@@ -54,7 +59,7 @@ services:
   # Jobs runtime: callback server + worker + scheduler
   knative-job-service:
     container_name: knative-job-service
-    image: ghcr.io/constructive-io/launchql:b88e3d1
+    image: constructive-launchql:dev
     # Override the image entrypoint and run the jobs runtime directly.
     entrypoint: ["node", "jobs/knative-job-service/dist/run.js"]
     depends_on:
@@ -77,12 +82,11 @@ services:
 
       # Callback HTTP server (job completion callbacks)
       INTERNAL_JOBS_CALLBACK_PORT: "8080"
-      INTERNAL_JOBS_CALLBACK_URL: "http://knative-job-service:8080"
+      # Hostname used by job-utils.getCallbackBaseUrl for callbacks
+      JOBS_CALLBACK_HOST: "knative-job-service"
 
       # Function gateway base URL (used by worker when no dev map is present)
-      # Not used in dev when INTERNAL_GATEWAY_DEVELOPMENT_MAP is set, but required for validation.
       KNATIVE_SERVICE_URL: "dev.internal"
-      INTERNAL_GATEWAY_URL: "http://simple-email:8080"
 
       # Development-only map from task identifier -> function URL
       # Used by @launchql/knative-job-worker when NODE_ENV !== 'production'.

jobs/job-pg/package.json

@@ -28,7 +28,7 @@
     "url": "https://github.com/launchql/jobs/issues"
   },
   "dependencies": {
-    "envalid": "6.0.2",
+    "@launchql/job-utils": "workspace:^",
     "pg": "8.16.3"
   }
 }

jobs/job-pg/src/env.ts

@@ -1,13 +1,9 @@
-import { cleanEnv, str, port } from 'envalid';
-
-export default cleanEnv(
-  process.env,
-  {
-    PGUSER: str({ default: 'postgres' }),
-    PGHOST: str({ default: 'localhost' }),
-    PGPASSWORD: str({ default: 'password' }),
-    PGPORT: port({ default: 5432 }),
-    PGDATABASE: str({ default: 'jobs' })
-  },
-  { dotEnvPath: null }
-);
+// Legacy env module (no longer used in code).
+// PG configuration is now resolved via @launchql/job-utils.getJobPgConfig().
+export default {
+  PGUSER: process.env.PGUSER || 'postgres',
+  PGHOST: process.env.PGHOST || 'localhost',
+  PGPASSWORD: process.env.PGPASSWORD || 'password',
+  PGPORT: Number(process.env.PGPORT) || 5432,
+  PGDATABASE: process.env.PGDATABASE || 'jobs'
+};

jobs/job-pg/src/index.ts

@@ -1,7 +1,7 @@
 /* eslint-disable no-console */
 
 import { Pool, PoolConfig } from 'pg';
-import env from './env';
+import { getJobPgConfig } from '@launchql/job-utils';
 
 // k8s only does SIGINT
 // other events are bad for babel-watch
@@ -31,12 +31,7 @@ function once<T extends (...args: unknown[]) => unknown>(
   };
 }
 
-const getDbString = (): string =>
-  `postgres://${env.PGUSER}:${env.PGPASSWORD}@${env.PGHOST}:${env.PGPORT}/${env.PGDATABASE}`;
-
-const pgPoolConfig: PoolConfig = {
-  connectionString: getDbString()
-};
+const pgPoolConfig: PoolConfig = getJobPgConfig() as PoolConfig;
 
 const end = (pool: Pool): void => {
   try {

jobs/job-scheduler/package.json

@@ -30,7 +30,7 @@
   "dependencies": {
     "@launchql/job-pg": "workspace:^",
     "@launchql/job-utils": "workspace:^",
-    "envalid": "6.0.2",
+    "@pgpmjs/logger": "workspace:^",
     "node-schedule": "1.3.2"
   }
 }

jobs/job-scheduler/src/env.ts

@@ -1,21 +1,16 @@
-import { cleanEnv, str, bool, port, makeValidator } from 'envalid';
-
-const array = makeValidator<string[]>((x) => x.split(',').filter((i) => i), '');
-
-export default cleanEnv(
-  process.env,
-  {
-    PGUSER: str({ default: 'postgres' }),
-    PGHOST: str({ default: 'localhost' }),
-    PGPASSWORD: str({ default: 'password' }),
-    PGPORT: port({ default: 5432 }),
-    PGDATABASE: str({ default: 'jobs' }),
-    JOBS_SCHEMA: str({ default: 'app_jobs' }),
-    JOBS_SUPPORT_ANY: bool({ default: true }),
-    JOBS_SUPPORTED: array({ default: '' as unknown as string[] }),
-    HOSTNAME: str({
-      default: 'scheduler-0'
-    })
-  },
-  { dotEnvPath: null }
-);
+// Legacy env module kept for compatibility.
+// Scheduler configuration now flows through @launchql/job-utils runtime helpers.
+export default {
+  PGUSER: process.env.PGUSER || 'postgres',
+  PGHOST: process.env.PGHOST || 'localhost',
+  PGPASSWORD: process.env.PGPASSWORD || 'password',
+  PGPORT: Number(process.env.PGPORT) || 5432,
+  PGDATABASE: process.env.PGDATABASE || 'jobs',
+  JOBS_SCHEMA: process.env.JOBS_SCHEMA || 'app_jobs',
+  JOBS_SUPPORT_ANY: process.env.JOBS_SUPPORT_ANY !== 'false',
+  JOBS_SUPPORTED: (process.env.JOBS_SUPPORTED || '')
+    .split(',')
+    .map(s => s.trim())
+    .filter(Boolean),
+  HOSTNAME: process.env.HOSTNAME || 'scheduler-0'
+};

jobs/job-scheduler/src/index.ts

@@ -1,11 +1,9 @@
-import env from './env';
 import * as jobs from '@launchql/job-utils';
 import type { PgClientLike } from '@launchql/job-utils';
 import schedule from 'node-schedule';
 import poolManager from '@launchql/job-pg';
 import type { Pool, PoolClient } from 'pg';
-
-/* eslint-disable no-console */
+import { Logger } from '@pgpmjs/logger';
 
 export interface ScheduledJobRow {
   id: number | string;
@@ -17,6 +15,8 @@ interface SchedulerJobHandle {
   cancel(): void;
 }
 
+const log = new Logger('jobs:scheduler');
+
 export default class Scheduler {
   idleDelay: number;
   supportedTaskNames: string[];
@@ -77,10 +77,8 @@ export default class Scheduler {
     }: { err?: Error; fatalError: unknown; jobId: ScheduledJobRow['id'] }
   ) {
     const when = err ? `after failure '${err.message}'` : 'after success';
-    console.error(
-      `Failed to release job '${jobId}' ${when}; committing seppuku`
-    );
-    console.error(fatalError);
+    log.error(`Failed to release job '${jobId}' ${when}; committing seppuku`);
+    log.error(String(fatalError));
     await poolManager.close();
     process.exit(1);
   }
@@ -92,9 +90,8 @@ export default class Scheduler {
       duration
     }: { err: Error; job: ScheduledJobRow; duration: string }
   ) {
-    console.error(
-      `scheduler: Failed to initialize scheduler for ${job.id} (${job.task_identifier}) with error ${err.message} (${duration}ms)`,
-      { err, stack: err.stack }
+    log.error(
+      `Failed to initialize scheduler for ${job.id} (${job.task_identifier}) with error ${err.message} (${duration}ms)`
     );
     const j = this.jobs[job.id];
     if (j) j.cancel();
@@ -107,8 +104,8 @@ export default class Scheduler {
     client: PgClientLike,
     { job, duration }: { job: ScheduledJobRow; duration: string }
   ) {
-    console.log(
-      `scheduler: initialized ${job.id} (${job.task_identifier}) with success (${duration}ms)`
+    log.info(
+      `initialized ${job.id} (${job.task_identifier}) with success (${duration}ms)`
     );
   }
   async scheduleJob(client: PgClientLike, job: ScheduledJobRow) {
@@ -120,18 +117,18 @@ export default class Scheduler {
 
       if (newjob) {
         if (newjob.id) {
-          console.log(`spinning up job[${newjob.task_identifier}]`);
+          log.info(`spinning up job[${newjob.task_identifier}]`);
         } else {
           // this means the scheduled_job has been deleted from db, so cancel it
           console.log(
-            `scheduler: attempted job[${job.task_identifier}] but it's probably non existent, unscheduling...`
+            `attempted job[${job.task_identifier}] but it's probably non existent, unscheduling...`
           );
           const scheduledJob = this.jobs[job.id];
           if (scheduledJob) scheduledJob.cancel();
         }
       } else {
-        console.log(
-          `scheduler: job already scheduled but not yet run or complete: [${job.task_identifier}]`
+        log.info(
+          `job already scheduled but not yet run or complete: [${job.task_identifier}]`
         );
       }
     });
@@ -149,7 +146,7 @@ export default class Scheduler {
     try {
       const job = await jobs.getScheduledJob<ScheduledJobRow>(client, {
         workerId: this.workerId,
-        supportedTaskNames: env.JOBS_SUPPORT_ANY
+        supportedTaskNames: jobs.getJobSupportAny()
           ? null
           : this.supportedTaskNames
       });
@@ -189,33 +186,33 @@ export default class Scheduler {
     }
   }
   listen() {
-    const listenForChanges = (
+      const listenForChanges = (
       err: Error | null,
       client: PoolClient,
       release: () => void
     ) => {
       if (err) {
-        console.error('scheduler: Error connecting with notify listener', err);
+        log.error('Error connecting with notify listener');
         // Try again in 5 seconds
         // should this really be done in the node process?
         setTimeout(this.listen, 5000);
         return;
       }
       client.on('notification', () => {
-        console.log('scheduler: a NEW scheduled JOB!');
+        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) => {
-        console.error('scheduler: Error with database notify listener', e);
+        log.error('Error with database notify listener');
         release();
         this.listen();
       });
-      console.log(
-        `scheduler: ${this.workerId} connected and looking for scheduled jobs...`
+      log.info(
+        `${this.workerId} connected and looking for scheduled jobs...`
       );
       this.doNext(client);
     };

jobs/job-scheduler/src/run.ts

@@ -1,15 +1,18 @@
 #!/usr/bin/env node
 
 import Scheduler from './index';
-import env from './env';
 import poolManager from '@launchql/job-pg';
+import {
+  getSchedulerHostname,
+  getJobSupported
+} from '@launchql/job-utils';
 
 const pgPool = poolManager.getPool();
 
 const scheduler = new Scheduler({
   pgPool,
-  workerId: env.HOSTNAME,
-  tasks: env.JOBS_SUPPORTED
+  workerId: getSchedulerHostname(),
+  tasks: getJobSupported()
 });
 
 scheduler.listen();