Skip to content

Commit 5e9c0a9

Browse files
pgflow botjumski
pgflow bot
authored and
jumski
committed
feat: add process platform adapter
1 parent 30f8253 commit 5e9c0a9

8 files changed

Lines changed: 455 additions & 13 deletions

File tree

‎pkgs/edge-worker/project.json‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,13 @@
147147
"parallel": false
148148
}
149149
},
150+
"test:node": {
151+
"executor": "nx:run-commands",
152+
"options": {
153+
"command": "pnpm vitest run tests/unit/platform/ProcessPlatformAdapter.test.ts",
154+
"cwd": "pkgs/edge-worker"
155+
}
156+
},
150157
"test:integration": {
151158
"dependsOn": ["db:ensure", "^build"],
152159
"executor": "nx:run-commands",

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

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
import type postgres from 'postgres';
2+
import type { SupabaseClient } from '@supabase/supabase-js';
3+
import type { SupabaseResources } from '@pgflow/dsl/supabase';
4+
import type { CreateWorkerFn, Logger, PlatformAdapter } from './types.js';
5+
import type { Worker } from '../core/Worker.js';
6+
import { createServiceSupabaseClient } from '../core/supabase-utils.js';
7+
import { Queries } from '../core/Queries.js';
8+
import { isLocalSupabaseEnv } from '../shared/localDetection.js';
9+
import { createLoggingFactory } from './logging.js';
10+
import { resolveConnectionString, resolveSqlConnection } from './resolveConnection.js';
11+
import { getProcessDeps, type ProcessDeps, type ProcessSignal } from './processDeps.js';
12+
13+
interface ProcessEnv extends Record<string, string | undefined> {
14+
SUPABASE_URL: string;
15+
SUPABASE_SERVICE_ROLE_KEY: string;
16+
WORKER_NAME?: string;
17+
DATABASE_URL?: string;
18+
EDGE_WORKER_DB_URL?: string;
19+
EDGE_WORKER_LOG_LEVEL?: string;
20+
}
21+
22+
type ProcessAdapterOptions = {
23+
sql?: postgres.Sql;
24+
connectionString?: string;
25+
maxPgConnections?: number;
26+
};
27+
28+
export class ProcessPlatformAdapter implements PlatformAdapter<SupabaseResources> {
29+
private readonly deps: ProcessDeps;
30+
private readonly logger: Logger;
31+
private readonly loggingFactory: ReturnType<typeof createLoggingFactory>;
32+
private readonly abortController = new AbortController();
33+
private readonly validatedEnv: ProcessEnv;
34+
private readonly _connectionString: string | undefined;
35+
private readonly _platformResources: SupabaseResources;
36+
private readonly ownsSql: boolean;
37+
private readonly queries: Queries;
38+
private worker: Worker | null = null;
39+
private workerId: string | null = null;
40+
private shutdownStarted = false;
41+
42+
constructor(
43+
options?: ProcessAdapterOptions,
44+
deps: ProcessDeps = getProcessDeps()
45+
) {
46+
this.deps = deps;
47+
this.assertProcessEnv(deps.env);
48+
this.validatedEnv = deps.env;
49+
this._connectionString = resolveConnectionString(this.validatedEnv, {
50+
hasSql: !!options?.sql,
51+
connectionString: options?.connectionString,
52+
});
53+
this.ownsSql = !options?.sql;
54+
this.loggingFactory = createLoggingFactory(this.validatedEnv);
55+
this.logger = this.loggingFactory.createLogger('ProcessPlatformAdapter');
56+
this._platformResources = {
57+
sql: resolveSqlConnection(this.validatedEnv, options),
58+
supabase: createServiceSupabaseClient(this.validatedEnv),
59+
};
60+
this.queries = new Queries(this._platformResources.sql);
61+
}
62+
63+
async startWorker(createWorkerFn: CreateWorkerFn): Promise<void> {
64+
const workerName = this.validatedEnv.WORKER_NAME || 'pgflow-worker';
65+
const workerId = this.deps.randomUUID();
66+
67+
this.workerId = workerId;
68+
this.loggingFactory.setWorkerId(workerId);
69+
this.loggingFactory.setWorkerName(workerName);
70+
this.registerSignalHandlers();
71+
72+
this.worker = createWorkerFn(this.loggingFactory.createLogger);
73+
await this.worker.startOnlyOnce({
74+
edgeFunctionName: workerName,
75+
workerId,
76+
startMode: 'process',
77+
});
78+
}
79+
80+
async stopWorker(): Promise<void> {
81+
this.requestShutdown();
82+
83+
try {
84+
if (this.worker) {
85+
await this.worker.stop();
86+
}
87+
if (this.workerId) {
88+
await this.queries.markWorkerStopped(this.workerId);
89+
}
90+
} finally {
91+
if (this.ownsSql) {
92+
await this._platformResources.sql.end();
93+
}
94+
}
95+
}
96+
97+
requestShutdown(): void {
98+
this.abortController.abort();
99+
}
100+
101+
createLogger(module: string): Logger {
102+
return this.loggingFactory.createLogger(module);
103+
}
104+
105+
get connectionString(): string | undefined {
106+
return this._connectionString;
107+
}
108+
109+
get env(): Record<string, string | undefined> {
110+
return this.validatedEnv;
111+
}
112+
113+
get shutdownSignal(): AbortSignal {
114+
return this.abortController.signal;
115+
}
116+
117+
get platformResources(): SupabaseResources {
118+
return this._platformResources;
119+
}
120+
121+
get isLocalEnvironment(): boolean {
122+
return isLocalSupabaseEnv(this.validatedEnv);
123+
}
124+
125+
get sql(): postgres.Sql {
126+
return this._platformResources.sql;
127+
}
128+
129+
get supabase(): SupabaseClient {
130+
return this._platformResources.supabase;
131+
}
132+
133+
private registerSignalHandlers(): void {
134+
for (const signal of ['SIGTERM', 'SIGINT', 'SIGQUIT'] satisfies ProcessSignal[]) {
135+
this.deps.onSignal(signal, () => this.handleSignal());
136+
}
137+
}
138+
139+
private async handleSignal(): Promise<void> {
140+
if (this.shutdownStarted) {
141+
this.deps.exit(1);
142+
}
143+
144+
this.shutdownStarted = true;
145+
146+
try {
147+
await this.stopWorker();
148+
} catch (error) {
149+
this.logger.error('Process worker shutdown failed', error);
150+
this.deps.setExitCode(1);
151+
this.deps.exit(1);
152+
}
153+
154+
this.deps.setExitCode(0);
155+
this.deps.exit(0);
156+
}
157+
158+
private assertProcessEnv(env: Record<string, string | undefined>): asserts env is ProcessEnv {
159+
const required = ['SUPABASE_URL', 'SUPABASE_SERVICE_ROLE_KEY'];
160+
const missing = required.filter((key) => !env[key]);
161+
162+
if (missing.length > 0) {
163+
throw new Error(`Missing required environment variables: ${missing.join(', ')}`);
164+
}
165+
}
166+
}

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

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import type { PlatformAdapter } from './types.js';
2+
import { ProcessPlatformAdapter } from './ProcessPlatformAdapter.js';
23
import { SupabasePlatformAdapter } from './SupabasePlatformAdapter.js';
34
import type { SupabaseResources } from '@pgflow/dsl/supabase';
45
import type postgres from 'postgres';
@@ -21,11 +22,17 @@ export function createAdapter(options?: AdapterOptions): PlatformAdapter<Supabas
2122
return adapter;
2223
}
2324

24-
// For now, only support Deno
25-
// Later add NodeAdapter, BrowserAdapter, etc.
25+
if (isProcessEnvironment()) {
26+
return new ProcessPlatformAdapter(options);
27+
}
28+
2629
throw new Error('Unsupported environment');
2730
}
2831

2932
function isDenoEnvironment(): boolean {
3033
return typeof Deno !== 'undefined';
3134
}
35+
36+
function isProcessEnvironment(): boolean {
37+
return typeof (globalThis as { process?: unknown }).process !== 'undefined';
38+
}

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,3 +2,5 @@ export * from './types.js';
22
export { createAdapter } from './createAdapter.js';
33
export { SupabasePlatformAdapter } from './SupabasePlatformAdapter.js';
44
export type { SupabasePlatformDeps } from './deps.js';
5+
export { ProcessPlatformAdapter } from './ProcessPlatformAdapter.js';
6+
export type { ProcessDeps, ProcessSignal } from './processDeps.js';

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

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
export type ProcessSignal = 'SIGTERM' | 'SIGINT' | 'SIGQUIT';
2+
3+
export type ProcessDeps = {
4+
env: Record<string, string | undefined>;
5+
onSignal: (signal: ProcessSignal, handler: () => void | Promise<void>) => void;
6+
exit: (code: number) => never;
7+
setExitCode: (code: number) => void;
8+
randomUUID: () => string;
9+
};
10+
11+
type ProcessLike = {
12+
env?: Record<string, string | undefined>;
13+
on?: (signal: ProcessSignal, handler: () => void | Promise<void>) => void;
14+
exit?: (code: number) => never;
15+
exitCode?: number;
16+
};
17+
18+
type CryptoLike = {
19+
randomUUID?: () => string;
20+
};
21+
22+
export function getProcessDeps(): ProcessDeps {
23+
const processLike = (globalThis as { process?: ProcessLike }).process;
24+
const cryptoLike = globalThis.crypto as CryptoLike | undefined;
25+
26+
if (!processLike?.env || !processLike.on || !processLike.exit || !cryptoLike?.randomUUID) {
27+
throw new Error('Process runtime is not available');
28+
}
29+
30+
return {
31+
env: processLike.env,
32+
onSignal: (signal, handler) => processLike.on?.(signal, handler),
33+
exit: (code) => processLike.exit!(code),
34+
setExitCode: (code) => {
35+
processLike.exitCode = code;
36+
},
37+
randomUUID: () => cryptoLike.randomUUID!(),
38+
};
39+
}

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

Lines changed: 16 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ export const DOCKER_TRANSACTION_POOLER_URL =
1313
export interface ConnectionEnv extends Record<string, string | undefined> {
1414
SUPABASE_ANON_KEY?: string;
1515
SUPABASE_SERVICE_ROLE_KEY?: string;
16+
DATABASE_URL?: string;
1617
EDGE_WORKER_DB_URL?: string;
1718
}
1819

@@ -29,7 +30,7 @@ export interface SqlConnectionOptions {
2930

3031
/**
3132
* Resolves the connection string based on priority:
32-
* config.sql -> config.connectionString -> EDGE_WORKER_DB_URL -> local fallback
33+
* config.sql -> config.connectionString -> DATABASE_URL -> EDGE_WORKER_DB_URL -> local fallback
3334
*/
3435
export function resolveConnectionString(
3536
env: ConnectionEnv,
@@ -42,12 +43,13 @@ export function resolveConnectionString(
4243
isLocal &&
4344
!options?.hasSql &&
4445
!options?.connectionString &&
46+
!env.DATABASE_URL &&
4547
!env.EDGE_WORKER_DB_URL
4648
) {
4749
return DOCKER_TRANSACTION_POOLER_URL;
4850
}
4951

50-
return options?.connectionString || env.EDGE_WORKER_DB_URL;
52+
return options?.connectionString || env.DATABASE_URL || env.EDGE_WORKER_DB_URL;
5153
}
5254

5355
/**
@@ -60,7 +62,7 @@ export function assertConnectionAvailable(
6062
if (!hasSql && !connectionString) {
6163
throw new Error(
6264
'No database connection available. Provide one of: ' +
63-
'config.sql, config.connectionString, or EDGE_WORKER_DB_URL environment variable.'
65+
'config.sql, config.connectionString, DATABASE_URL, or EDGE_WORKER_DB_URL environment variable.'
6466
);
6567
}
6668
}
@@ -69,8 +71,9 @@ export function assertConnectionAvailable(
6971
* Resolves and creates the SQL connection based on priority:
7072
* 1. config.sql - User-provided SQL client (highest priority)
7173
* 2. config.connectionString - User-provided connection string
72-
* 3. EDGE_WORKER_DB_URL - Environment variable
73-
* 4. Local Supabase detection + Docker URL (lowest priority)
74+
* 3. DATABASE_URL - Environment variable
75+
* 4. EDGE_WORKER_DB_URL - Environment variable
76+
* 5. Local Supabase detection + Docker URL (lowest priority)
7477
*
7578
* @throws Error if no connection source is available
7679
*/
@@ -90,18 +93,23 @@ export function resolveSqlConnection(
9093
return postgres(options.connectionString, { prepare: false, max });
9194
}
9295

93-
// 3. EDGE_WORKER_DB_URL
96+
// 3. DATABASE_URL
97+
if (env.DATABASE_URL) {
98+
return postgres(env.DATABASE_URL, { prepare: false, max });
99+
}
100+
101+
// 4. EDGE_WORKER_DB_URL
94102
if (env.EDGE_WORKER_DB_URL) {
95103
return postgres(env.EDGE_WORKER_DB_URL, { prepare: false, max });
96104
}
97105

98-
// 4. Local Supabase detection + docker URL
106+
// 5. Local Supabase detection + docker URL
99107
if (isLocalSupabaseEnv(env)) {
100108
return postgres(DOCKER_TRANSACTION_POOLER_URL, { prepare: false, max });
101109
}
102110

103111
throw new Error(
104112
'No database connection available. Provide one of: ' +
105-
'config.sql, config.connectionString, or EDGE_WORKER_DB_URL environment variable.'
113+
'config.sql, config.connectionString, DATABASE_URL, or EDGE_WORKER_DB_URL environment variable.'
106114
);
107115
}

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
* made configurable or determined by build-time configuration.
66
*/
77

8-
import type { SupabaseResources, SupabaseEnv } from '@pgflow/dsl/supabase';
8+
import type { SupabaseResources } from '@pgflow/dsl/supabase';
99
import type { BaseContext } from '@pgflow/dsl';
1010

1111
/**
@@ -18,9 +18,9 @@ export type CurrentPlatformResources = SupabaseResources;
1818
* The environment type for the current platform.
1919
* This is hardcoded to Supabase for MVP but can be made configurable later.
2020
*/
21-
export type CurrentPlatformEnv = SupabaseEnv;
21+
export type CurrentPlatformEnv = Record<string, string | undefined>;
2222

2323
/**
2424
* All resources available to flows (base context + platform resources)
2525
*/
26-
export type AvailableResources = BaseContext<CurrentPlatformEnv> & CurrentPlatformResources;
26+
export type AvailableResources = BaseContext<CurrentPlatformEnv> & CurrentPlatformResources;

0 commit comments

Comments
 (0)