feat: add ProcessPlatformAdapter for Node.js process environments

pkgs/edge-worker/project.json

@@ -147,6 +147,13 @@
         "parallel": false
       }
     },
+    "test:node": {
+      "executor": "nx:run-commands",
+      "options": {
+        "command": "pnpm vitest run tests/unit/platform/ProcessPlatformAdapter.test.ts",
+        "cwd": "pkgs/edge-worker"
+      }
+    },
     "test:integration": {
       "dependsOn": ["db:ensure", "^build"],
       "executor": "nx:run-commands",

pkgs/edge-worker/src/platform/ProcessPlatformAdapter.ts

@@ -0,0 +1,166 @@
+import type postgres from 'postgres';
+import type { SupabaseClient } from '@supabase/supabase-js';
+import type { SupabaseResources } from '@pgflow/dsl/supabase';
+import type { CreateWorkerFn, Logger, PlatformAdapter } from './types.js';
+import type { Worker } from '../core/Worker.js';
+import { createServiceSupabaseClient } from '../core/supabase-utils.js';
+import { Queries } from '../core/Queries.js';
+import { isLocalSupabaseEnv } from '../shared/localDetection.js';
+import { createLoggingFactory } from './logging.js';
+import { resolveConnectionString, resolveSqlConnection } from './resolveConnection.js';
+import { getProcessDeps, type ProcessDeps, type ProcessSignal } from './processDeps.js';
+
+interface ProcessEnv extends Record<string, string | undefined> {
+  SUPABASE_URL: string;
+  SUPABASE_SERVICE_ROLE_KEY: string;
+  WORKER_NAME?: string;
+  DATABASE_URL?: string;
+  EDGE_WORKER_DB_URL?: string;
+  EDGE_WORKER_LOG_LEVEL?: string;
+}
+
+type ProcessAdapterOptions = {
+  sql?: postgres.Sql;
+  connectionString?: string;
+  maxPgConnections?: number;
+};
+
+export class ProcessPlatformAdapter implements PlatformAdapter<SupabaseResources> {
+  private readonly deps: ProcessDeps;
+  private readonly logger: Logger;
+  private readonly loggingFactory: ReturnType<typeof createLoggingFactory>;
+  private readonly abortController = new AbortController();
+  private readonly validatedEnv: ProcessEnv;
+  private readonly _connectionString: string | undefined;
+  private readonly _platformResources: SupabaseResources;
+  private readonly ownsSql: boolean;
+  private readonly queries: Queries;
+  private worker: Worker | null = null;
+  private workerId: string | null = null;
+  private shutdownStarted = false;
+
+  constructor(
+    options?: ProcessAdapterOptions,
+    deps: ProcessDeps = getProcessDeps()
+  ) {
+    this.deps = deps;
+    this.assertProcessEnv(deps.env);
+    this.validatedEnv = deps.env;
+    this._connectionString = resolveConnectionString(this.validatedEnv, {
+      hasSql: !!options?.sql,
+      connectionString: options?.connectionString,
+    });
+    this.ownsSql = !options?.sql;
+    this.loggingFactory = createLoggingFactory(this.validatedEnv);
+    this.logger = this.loggingFactory.createLogger('ProcessPlatformAdapter');
+    this._platformResources = {
+      sql: resolveSqlConnection(this.validatedEnv, options),
+      supabase: createServiceSupabaseClient(this.validatedEnv),
+    };
+    this.queries = new Queries(this._platformResources.sql);
+  }
+
+  async startWorker(createWorkerFn: CreateWorkerFn): Promise<void> {
+    const workerName = this.validatedEnv.WORKER_NAME || 'pgflow-worker';
+    const workerId = this.deps.randomUUID();
+
+    this.workerId = workerId;
+    this.loggingFactory.setWorkerId(workerId);
+    this.loggingFactory.setWorkerName(workerName);
+    this.registerSignalHandlers();
+
+    this.worker = createWorkerFn(this.loggingFactory.createLogger);
+    await this.worker.startOnlyOnce({
+      edgeFunctionName: workerName,
+      workerId,
+      startMode: 'process',
+    });
+  }
+
+  async stopWorker(): Promise<void> {
+    this.requestShutdown();
+
+    try {
+      if (this.worker) {
+        await this.worker.stop();
+      }
+      if (this.workerId) {
+        await this.queries.markWorkerStopped(this.workerId);
+      }
+    } finally {
+      if (this.ownsSql) {
+        await this._platformResources.sql.end();
+      }
+    }
+  }
+
+  requestShutdown(): void {
+    this.abortController.abort();
+  }
+
+  createLogger(module: string): Logger {
+    return this.loggingFactory.createLogger(module);
+  }
+
+  get connectionString(): string | undefined {
+    return this._connectionString;
+  }
+
+  get env(): Record<string, string | undefined> {
+    return this.validatedEnv;
+  }
+
+  get shutdownSignal(): AbortSignal {
+    return this.abortController.signal;
+  }
+
+  get platformResources(): SupabaseResources {
+    return this._platformResources;
+  }
+
+  get isLocalEnvironment(): boolean {
+    return isLocalSupabaseEnv(this.validatedEnv);
+  }
+
+  get sql(): postgres.Sql {
+    return this._platformResources.sql;
+  }
+
+  get supabase(): SupabaseClient {
+    return this._platformResources.supabase;
+  }
+
+  private registerSignalHandlers(): void {
+    for (const signal of ['SIGTERM', 'SIGINT', 'SIGQUIT'] satisfies ProcessSignal[]) {
+      this.deps.onSignal(signal, () => this.handleSignal());
+    }
+  }
+
+  private async handleSignal(): Promise<void> {
+    if (this.shutdownStarted) {
+      this.deps.exit(1);
+    }
+
+    this.shutdownStarted = true;
+
+    try {
+      await this.stopWorker();
+    } catch (error) {
+      this.logger.error('Process worker shutdown failed', error);
+      this.deps.setExitCode(1);
+      this.deps.exit(1);
+    }
+
+    this.deps.setExitCode(0);
+    this.deps.exit(0);
+  }
+
+  private assertProcessEnv(env: Record<string, string | undefined>): asserts env is ProcessEnv {
+    const required = ['SUPABASE_URL', 'SUPABASE_SERVICE_ROLE_KEY'];
+    const missing = required.filter((key) => !env[key]);
+
+    if (missing.length > 0) {
+      throw new Error(`Missing required environment variables: ${missing.join(', ')}`);
+    }
+  }
+}

pkgs/edge-worker/src/platform/createAdapter.ts

@@ -1,4 +1,5 @@
 import type { PlatformAdapter } from './types.js';
+import { ProcessPlatformAdapter } from './ProcessPlatformAdapter.js';
 import { SupabasePlatformAdapter } from './SupabasePlatformAdapter.js';
 import type { SupabaseResources } from '@pgflow/dsl/supabase';
 import type postgres from 'postgres';
@@ -21,11 +22,17 @@ export function createAdapter(options?: AdapterOptions): PlatformAdapter<Supabas
     return adapter;
   }
 
-  // For now, only support Deno
-  // Later add NodeAdapter, BrowserAdapter, etc.
+  if (isProcessEnvironment()) {
+    return new ProcessPlatformAdapter(options);
+  }
+
   throw new Error('Unsupported environment');
 }
 
 function isDenoEnvironment(): boolean {
   return typeof Deno !== 'undefined';
 }
+
+function isProcessEnvironment(): boolean {
+  return typeof (globalThis as { process?: unknown }).process !== 'undefined';
+}

pkgs/edge-worker/src/platform/index.ts

@@ -2,3 +2,5 @@ export * from './types.js';
 export { createAdapter } from './createAdapter.js';
 export { SupabasePlatformAdapter } from './SupabasePlatformAdapter.js';
 export type { SupabasePlatformDeps } from './deps.js';
+export { ProcessPlatformAdapter } from './ProcessPlatformAdapter.js';
+export type { ProcessDeps, ProcessSignal } from './processDeps.js';

pkgs/edge-worker/src/platform/processDeps.ts

@@ -0,0 +1,39 @@
+export type ProcessSignal = 'SIGTERM' | 'SIGINT' | 'SIGQUIT';
+
+export type ProcessDeps = {
+  env: Record<string, string | undefined>;
+  onSignal: (signal: ProcessSignal, handler: () => void | Promise<void>) => void;
+  exit: (code: number) => never;
+  setExitCode: (code: number) => void;
+  randomUUID: () => string;
+};
+
+type ProcessLike = {
+  env?: Record<string, string | undefined>;
+  on?: (signal: ProcessSignal, handler: () => void | Promise<void>) => void;
+  exit?: (code: number) => never;
+  exitCode?: number;
+};
+
+type CryptoLike = {
+  randomUUID?: () => string;
+};
+
+export function getProcessDeps(): ProcessDeps {
+  const processLike = (globalThis as { process?: ProcessLike }).process;
+  const cryptoLike = globalThis.crypto as CryptoLike | undefined;
+
+  if (!processLike?.env || !processLike.on || !processLike.exit || !cryptoLike?.randomUUID) {
+    throw new Error('Process runtime is not available');
+  }
+
+  return {
+    env: processLike.env,
+    onSignal: (signal, handler) => processLike.on?.(signal, handler),
+    exit: (code) => processLike.exit!(code),
+    setExitCode: (code) => {
+      processLike.exitCode = code;
+    },
+    randomUUID: () => cryptoLike.randomUUID!(),
+  };
+}

pkgs/edge-worker/src/platform/resolveConnection.ts

@@ -13,6 +13,7 @@ export const DOCKER_TRANSACTION_POOLER_URL =
 export interface ConnectionEnv extends Record<string, string | undefined> {
   SUPABASE_ANON_KEY?: string;
   SUPABASE_SERVICE_ROLE_KEY?: string;
+  DATABASE_URL?: string;
   EDGE_WORKER_DB_URL?: string;
 }
 
@@ -29,7 +30,7 @@ export interface SqlConnectionOptions {
 
 /**
  * Resolves the connection string based on priority:
- * config.sql -> config.connectionString -> EDGE_WORKER_DB_URL -> local fallback
+ * config.sql -> config.connectionString -> DATABASE_URL -> EDGE_WORKER_DB_URL -> local fallback
  */
 export function resolveConnectionString(
   env: ConnectionEnv,
@@ -42,12 +43,13 @@ export function resolveConnectionString(
     isLocal &&
     !options?.hasSql &&
     !options?.connectionString &&
+    !env.DATABASE_URL &&
     !env.EDGE_WORKER_DB_URL
   ) {
     return DOCKER_TRANSACTION_POOLER_URL;
   }
 
-  return options?.connectionString || env.EDGE_WORKER_DB_URL;
+  return options?.connectionString || env.DATABASE_URL || env.EDGE_WORKER_DB_URL;
 }
 
 /**
@@ -60,7 +62,7 @@ export function assertConnectionAvailable(
   if (!hasSql && !connectionString) {
     throw new Error(
       'No database connection available. Provide one of: ' +
-        'config.sql, config.connectionString, or EDGE_WORKER_DB_URL environment variable.'
+        'config.sql, config.connectionString, DATABASE_URL, or EDGE_WORKER_DB_URL environment variable.'
     );
   }
 }
@@ -69,8 +71,9 @@ export function assertConnectionAvailable(
  * Resolves and creates the SQL connection based on priority:
  * 1. config.sql - User-provided SQL client (highest priority)
  * 2. config.connectionString - User-provided connection string
- * 3. EDGE_WORKER_DB_URL - Environment variable
- * 4. Local Supabase detection + Docker URL (lowest priority)
+ * 3. DATABASE_URL - Environment variable
+ * 4. EDGE_WORKER_DB_URL - Environment variable
+ * 5. Local Supabase detection + Docker URL (lowest priority)
  *
  * @throws Error if no connection source is available
  */
@@ -90,18 +93,23 @@ export function resolveSqlConnection(
     return postgres(options.connectionString, { prepare: false, max });
   }
 
-  // 3. EDGE_WORKER_DB_URL
+  // 3. DATABASE_URL
+  if (env.DATABASE_URL) {
+    return postgres(env.DATABASE_URL, { prepare: false, max });
+  }
+
+  // 4. EDGE_WORKER_DB_URL
   if (env.EDGE_WORKER_DB_URL) {
     return postgres(env.EDGE_WORKER_DB_URL, { prepare: false, max });
   }
 
-  // 4. Local Supabase detection + docker URL
+  // 5. Local Supabase detection + docker URL
   if (isLocalSupabaseEnv(env)) {
     return postgres(DOCKER_TRANSACTION_POOLER_URL, { prepare: false, max });
   }
 
   throw new Error(
     'No database connection available. Provide one of: ' +
-      'config.sql, config.connectionString, or EDGE_WORKER_DB_URL environment variable.'
+      'config.sql, config.connectionString, DATABASE_URL, or EDGE_WORKER_DB_URL environment variable.'
   );
 }

pkgs/edge-worker/src/types/currentPlatform.ts

@@ -5,7 +5,7 @@
  * made configurable or determined by build-time configuration.
  */
 
-import type { SupabaseResources, SupabaseEnv } from '@pgflow/dsl/supabase';
+import type { SupabaseResources } from '@pgflow/dsl/supabase';
 import type { BaseContext } from '@pgflow/dsl';
 
 /**
@@ -18,9 +18,9 @@ export type CurrentPlatformResources = SupabaseResources;
  * The environment type for the current platform.
  * This is hardcoded to Supabase for MVP but can be made configurable later.
  */
-export type CurrentPlatformEnv = SupabaseEnv;
+export type CurrentPlatformEnv = Record<string, string | undefined>;
 
 /**
  * All resources available to flows (base context + platform resources)
  */
-export type AvailableResources = BaseContext<CurrentPlatformEnv> & CurrentPlatformResources;
+export type AvailableResources = BaseContext<CurrentPlatformEnv> & CurrentPlatformResources;