fix: preserve portable worker compatibility

pkgs/edge-worker/src/core/Queries.ts

@@ -82,6 +82,13 @@ export class Queries {
    * pinging during startup (debounce).
    */
   async trackWorkerFunction(functionName: string, startMode: WorkerStartMode = 'http'): Promise<void> {
+    if (startMode === 'http') {
+      await this.sql`
+        SELECT pgflow.track_worker_function(${functionName})
+      `;
+      return;
+    }
+
     await this.sql`
       SELECT pgflow.track_worker_function(${functionName}, ${startMode})
     `;

pkgs/edge-worker/src/core/Worker.ts

@@ -67,7 +67,7 @@ export class Worker {
 
         if (!this.isMainLoopActive) {
           this.logDeprecation();
-          if (this.lifecycle.isDeprecated) {
+          if (this.isDeprecated) {
             this.deprecationHandler?.();
           }
           break;
@@ -142,15 +142,15 @@ export class Worker {
   }
 
   get isStarting() {
-    return this.lifecycle.isStarting;
+    return this.lifecycle.isStarting ?? false;
   }
 
   get isRunning() {
     return this.lifecycle.isRunning;
   }
 
   get isDeprecated() {
-    return this.lifecycle.isDeprecated;
+    return this.lifecycle.isDeprecated ?? false;
   }
 
   get isStopped() {

pkgs/edge-worker/src/core/WorkerLifecycle.ts

@@ -1,6 +1,6 @@
 import type { Queries } from './Queries.js';
 import type { Queue } from '../queue/Queue.js';
-import type { ILifecycle, Json, WorkerBootstrap, WorkerRow } from './types.js';
+import type { InternalLifecycle, Json, WorkerBootstrap, WorkerRow } from './types.js';
 import { States, WorkerState } from './WorkerState.js';
 import type { Logger } from '../platform/types.js';
 
@@ -9,7 +9,7 @@ export interface LifecycleConfig {
   heartbeatInterval?: number;
 }
 
-export class WorkerLifecycle<IMessage extends Json> implements ILifecycle {
+export class WorkerLifecycle<IMessage extends Json> implements InternalLifecycle {
   private workerState: WorkerState;
   private logger: Logger;
   private queries: Queries;

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

@@ -27,15 +27,20 @@ export interface ILifecycle {
   get edgeFunctionName(): string | undefined;
   get queueName(): string;
   get isCreated(): boolean;
-  get isStarting(): boolean;
+  readonly isStarting?: boolean;
   get isRunning(): boolean;
-  get isDeprecated(): boolean;
+  readonly isDeprecated?: boolean;
   get isStopping(): boolean;
   get isStopped(): boolean;
 
   transitionToStopping(): void;
 }
 
+export interface InternalLifecycle extends ILifecycle {
+  readonly isStarting: boolean;
+  readonly isDeprecated: boolean;
+}
+
 export interface IBatchProcessor {
   processBatch(): Promise<void>;
   awaitCompletion(): Promise<void>;

pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts

@@ -1,5 +1,5 @@
 import type { Queries } from '../core/Queries.js';
-import type { ILifecycle, WorkerBootstrap, WorkerRow } from '../core/types.js';
+import type { InternalLifecycle, WorkerBootstrap, WorkerRow } from '../core/types.js';
 import type { Logger, StartupContext } from '../platform/types.js';
 import { States, WorkerState } from '../core/WorkerState.js';
 import type { AnyFlow } from '@pgflow/dsl';
@@ -21,7 +21,7 @@ type CompilationStatus = 'compiled' | 'verified' | 'recompiled' | 'mismatch';
 /**
  * A specialized WorkerLifecycle for Flow-based workers that is aware of the Flow's step types
  */
-export class FlowWorkerLifecycle<TFlow extends AnyFlow> implements ILifecycle {
+export class FlowWorkerLifecycle<TFlow extends AnyFlow> implements InternalLifecycle {
   private workerState: WorkerState;
   private logger: Logger;
   private queries: Queries;

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

@@ -37,9 +37,13 @@ export class ProcessPlatformAdapter implements PlatformAdapter<SupabaseResources
   private readonly queries: Queries;
   private worker: Worker | null = null;
   private workerId: string | null = null;
+  private startupPromise: Promise<void> | null = null;
   private stopPromise: Promise<void> | null = null;
   private gracefulExitPromise: Promise<void> | null = null;
+  private cleanupPromise: Promise<void> | null = null;
+  private signalHandlersRegistered = false;
   private signalCount = 0;
+  private readonly signalHandler = () => this.handleSignal();
 
   constructor(
     options?: ProcessAdapterOptions,
@@ -48,41 +52,34 @@ export class ProcessPlatformAdapter implements PlatformAdapter<SupabaseResources
     this.deps = deps;
     this.assertProcessEnv(deps.env);
     this.validatedEnv = deps.env;
-    this._connectionString = resolveConnectionString(this.validatedEnv, {
+    const connectionOptions = {
+      ...options,
       hasSql: !!options?.sql,
-      connectionString: options?.connectionString,
-    });
+      allowDatabaseUrl: true,
+    };
+    this._connectionString = resolveConnectionString(
+      this.validatedEnv,
+      connectionOptions
+    );
     this.ownsSql = !options?.sql;
     this.loggingFactory = createLoggingFactory(this.validatedEnv);
     this.logger = this.loggingFactory.createLogger('ProcessPlatformAdapter');
     this._platformResources = {
-      sql: resolveSqlConnection(this.validatedEnv, options),
+      sql: resolveSqlConnection(this.validatedEnv, connectionOptions),
       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.worker = createWorkerFn(this.loggingFactory.createLogger);
-
-    this.worker.onDeprecated(() => {
-      this.handleDeprecation().catch(() => undefined);
-    });
-
-    await this.worker.startOnlyOnce({
-      edgeFunctionName: workerName,
-      workerId,
-      startMode: 'process',
-    });
-
+  startWorker(createWorkerFn: CreateWorkerFn): Promise<void> {
     this.registerSignalHandlers();
+    this.startupPromise ??= this.performStartWorker(createWorkerFn).catch(
+      async (error) => {
+        await this.cleanup();
+        throw error;
+      }
+    );
+    return this.startupPromise;
   }
 
   stopWorker(): Promise<void> {
@@ -126,27 +123,68 @@ export class ProcessPlatformAdapter implements PlatformAdapter<SupabaseResources
     return this._platformResources.supabase;
   }
 
+  private async performStartWorker(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.worker = createWorkerFn(this.loggingFactory.createLogger);
+    this.worker.onDeprecated(() => {
+      this.handleDeprecation().catch(() => undefined);
+    });
+
+    await this.worker.startOnlyOnce({
+      edgeFunctionName: workerName,
+      workerId,
+      startMode: 'process',
+    });
+  }
+
   private registerSignalHandlers(): void {
+    if (this.signalHandlersRegistered) return;
+
+    this.signalHandlersRegistered = true;
     for (const signal of ['SIGTERM', 'SIGINT', 'SIGQUIT'] satisfies ProcessSignal[]) {
-      this.deps.onSignal(signal, () => this.handleSignal());
+      this.deps.onSignal(signal, this.signalHandler);
+    }
+  }
+
+  private removeSignalHandlers(): void {
+    if (!this.signalHandlersRegistered) return;
+
+    this.signalHandlersRegistered = false;
+    for (const signal of ['SIGTERM', 'SIGINT', 'SIGQUIT'] satisfies ProcessSignal[]) {
+      this.deps.offSignal?.(signal, this.signalHandler);
     }
   }
 
   private async performStopWorker(): Promise<void> {
     this.requestShutdown();
 
     try {
+      await this.startupPromise;
       if (this.worker) {
         await this.worker.stop();
       }
       if (this.workerId) {
         await this.queries.markWorkerStopped(this.workerId);
       }
     } finally {
+      await this.cleanup();
+    }
+  }
+
+  private cleanup(): Promise<void> {
+    this.cleanupPromise ??= (async () => {
+      this.removeSignalHandlers();
       if (this.ownsSql) {
         await this._platformResources.sql.end();
       }
-    }
+    })();
+    return this.cleanupPromise;
   }
 
   private gracefulExit(): Promise<void> {

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

@@ -222,15 +222,20 @@ export class SupabasePlatformAdapter implements PlatformAdapter<SupabaseResource
 
       this.logger.debug(`HTTP Request: ${this.edgeFunctionName}`);
 
-      const wasStarted = await this.ensureWorkerStarted(req, createWorkerFn);
-
-      return new Response(JSON.stringify({
-        status: wasStarted ? 'started' : 'running',
-        workerId: this.workerId,
-        functionName: this.edgeFunctionName,
-      }), {
-        headers: { 'Content-Type': 'application/json' },
-      });
+      try {
+        const wasStarted = await this.ensureWorkerStarted(req, createWorkerFn);
+
+        return new Response(JSON.stringify({
+          status: wasStarted ? 'started' : 'running',
+          workerId: this.workerId,
+          functionName: this.edgeFunctionName,
+        }), {
+          headers: { 'Content-Type': 'application/json' },
+        });
+      } catch (error) {
+        this.logger.error('Worker startup failed', error);
+        return createServerErrorResponse();
+      }
     });
   }
 
@@ -279,12 +284,21 @@ export class SupabasePlatformAdapter implements PlatformAdapter<SupabaseResource
     this.loggingFactory.setWorkerName(this.edgeFunctionName);
 
     // Create the worker using the factory function and the logger
-    this.worker = createWorkerFn(this.loggingFactory.createLogger);
-    void this.worker.startOnlyOnce({
-      edgeFunctionName: this.edgeFunctionName,
-      workerId,
-      startMode: 'http',
-    });
+    const worker = createWorkerFn(this.loggingFactory.createLogger);
+    this.worker = worker;
+
+    try {
+      await worker.startOnlyOnce({
+        edgeFunctionName: this.edgeFunctionName,
+        workerId,
+        startMode: 'http',
+      });
+    } catch (error) {
+      if (this.worker === worker) {
+        this.worker = null;
+      }
+      throw error;
+    }
   }
 
   /**

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

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