konnosDashboard
loop.ts605 lines · main
1// Runs inside the Deno isolate.
2//
3// Wire protocol:
4// stdin: newline-delimited JSON HostToIsolate messages
5// stdout: newline-delimited JSON IsolateToHost messages
6// stderr: newline-delimited JSON LogLine envelopes (best-effort)
7
8const encoder = new TextEncoder();
9const originalStderrWrite = Deno.stderr.write.bind(Deno.stderr);
10
11let currentRequestId: string | null = null;
12
13function writeStdout(obj: unknown): Promise<number> {
14 return Deno.stdout.write(encoder.encode(JSON.stringify(obj) + '\n'));
15}
16
17function emitLog(level: 'debug' | 'info' | 'warn' | 'error', args: unknown[]): void {
18 const msg = args.map(stringify).join(' ');
19 const line = JSON.stringify({
20 type: 'log',
21 requestId: currentRequestId,
22 level,
23 msg,
24 ts: Date.now(),
25 }) + '\n';
26 originalStderrWrite(encoder.encode(line));
27}
28
29function stringify(v: unknown): string {
30 if (typeof v === 'string') return v;
31 try {
32 return JSON.stringify(v);
33 } catch {
34 return String(v);
35 }
36}
37
38// Replace global console — must happen before customer code runs.
39(globalThis as { console: unknown }).console = {
40 log: (...a: unknown[]) => emitLog('info', a),
41 info: (...a: unknown[]) => emitLog('info', a),
42 warn: (...a: unknown[]) => emitLog('warn', a),
43 error: (...a: unknown[]) => emitLog('error', a),
44 debug: (...a: unknown[]) => emitLog('debug', a),
45};
46
47// Belt-and-braces v6 deny — Deno 2.x's --deny-net doesn't accept v6 CIDRs,
48// so we shim fetch to reject literal v6 addresses in ::1, fc00::/7, fe80::/10.
49// Hostnames that resolve to v6 are NOT covered here — the Phase 3 host-level
50// proxy will close that gap. This guards the obvious literal-IP case.
51const originalFetch = globalThis.fetch;
52function isBlockedV6(host: string): boolean {
53 // strip brackets if present
54 const h = host.startsWith('[') && host.endsWith(']') ? host.slice(1, -1) : host;
55 // ::1 (loopback)
56 if (h === '::1') return true;
57 // fc00::/7 — first 7 bits = 1111110_ → first hextet starts with fc or fd
58 if (/^fc[0-9a-f]{2}:/i.test(h) || /^fd[0-9a-f]{2}:/i.test(h)) return true;
59 // fe80::/10 — first 10 bits → first hextet starts with fe8, fe9, fea, feb
60 if (/^fe[89ab][0-9a-f]:/i.test(h)) return true;
61 return false;
62}
63(globalThis as { fetch: typeof fetch }).fetch = ((input: RequestInfo | URL, init?: RequestInit) => {
64 try {
65 const url = typeof input === 'string' || input instanceof URL ? new URL(input.toString()) : new URL(input.url);
66 if (isBlockedV6(url.hostname)) {
67 return Promise.reject(
68 Object.assign(new Error(`fetch blocked: IPv6 ${url.hostname} in deny range`), { name: 'PermissionDenied' }),
69 );
70 }
71 } catch { /* fall through */ }
72 return originalFetch(input as never, init);
73}) as typeof fetch;
74
75// Pending queries — qid → resolver. Multiple in flight per invocation
76// when customer uses Promise.all.
77const pendingQueries = new Map<
78 string,
79 { resolve: (rows: readonly unknown[]) => void; reject: (err: Error) => void }
80>();
81let qidCounter = 0;
82
83// ---------------------------------------------------------------------------
84// Ctx — structurally satisfies `Ctx` from @briven/schema. We can't import
85// from @briven/schema here because loop.ts is materialized into the isolate
86// and resolved by Deno (no node_modules access). The shape MUST stay in sync
87// with packages/schema/src/ctx.ts.
88// ---------------------------------------------------------------------------
89
90export interface AuthContext {
91 readonly userId: string;
92 readonly tokenType: 'session' | 'api_key';
93}
94
95export interface Logger {
96 debug(event: string, fields?: Record<string, unknown>): void;
97 info(event: string, fields?: Record<string, unknown>): void;
98 warn(event: string, fields?: Record<string, unknown>): void;
99 error(event: string, fields?: Record<string, unknown>): void;
100}
101
102export interface SelectQuery extends PromiseLike<unknown[]> {
103 where(predicate: Record<string, unknown>): SelectQuery;
104 orderBy(column: string, direction?: 'asc' | 'desc'): SelectQuery;
105 limit(n: number): SelectQuery;
106 offset(n: number): SelectQuery;
107}
108
109export interface InsertQuery extends PromiseLike<unknown[]> {
110 returning(columns?: readonly string[]): PromiseLike<unknown[]>;
111}
112
113export interface UpdateQuery extends PromiseLike<unknown[]> {
114 where(predicate: Record<string, unknown>): UpdateQuery;
115 returning(columns?: readonly string[]): PromiseLike<unknown[]>;
116}
117
118export interface DeleteQuery extends PromiseLike<unknown[]> {
119 where(predicate: Record<string, unknown>): DeleteQuery;
120 returning(columns?: readonly string[]): PromiseLike<unknown[]>;
121}
122
123export interface TableQuery {
124 select(columns?: readonly string[]): SelectQuery;
125 insert(values: Record<string, unknown> | readonly Record<string, unknown>[]): InsertQuery;
126 update(patch: Record<string, unknown>): UpdateQuery;
127 delete(): DeleteQuery;
128}
129
130export interface DbClient {
131 <TTable extends string>(table: TTable): TableQuery;
132 execute(sql: string, params?: readonly unknown[]): Promise<unknown[]>;
133}
134
135export interface Ctx {
136 readonly db: DbClient;
137 readonly requestId: string;
138 readonly log: Logger;
139 readonly env: Readonly<Record<string, string | undefined>>;
140 readonly auth: AuthContext | null;
141}
142
143// ---------------------------------------------------------------------------
144// Query plumbing
145// ---------------------------------------------------------------------------
146
147/**
148 * Allocate a qid, register a pending entry, send the query frame to the host,
149 * and return a promise that resolves with the rows the host sends back.
150 *
151 * Centralising this means every builder's `then()` and `db.execute()` shares
152 * the same null-guard, error-rejection, and stdout-failure handling.
153 */
154function sendQuery(req: { sql: string; params: readonly unknown[]; table: string }): Promise<unknown[]> {
155 if (currentRequestId === null) {
156 return Promise.reject(new Error('ctx.db(...) used outside an invocation'));
157 }
158 const requestId = currentRequestId;
159 const qid = `q${++qidCounter}`;
160 return new Promise<unknown[]>((resolve, reject) => {
161 pendingQueries.set(qid, {
162 resolve: (rows) => resolve(rows as unknown[]),
163 reject,
164 });
165 writeStdout({
166 type: 'query',
167 requestId,
168 qid,
169 sql: req.sql,
170 params: req.params,
171 table: req.table,
172 }).catch((err: unknown) => {
173 pendingQueries.delete(qid);
174 reject(err instanceof Error ? err : new Error(String(err)));
175 });
176 });
177}
178
179function makeCtx(auth: AuthContext | null, requestId: string): Ctx {
180 const db = ((table: string) => makeTableProxy(table)) as DbClient;
181 db.execute = (sql: string, params?: readonly unknown[]) =>
182 sendQuery({ sql, params: params ?? [], table: '' });
183
184 const log: Logger = {
185 debug: (event, fields) => emitLog('debug', fields ? [event, fields] : [event]),
186 info: (event, fields) => emitLog('info', fields ? [event, fields] : [event]),
187 warn: (event, fields) => emitLog('warn', fields ? [event, fields] : [event]),
188 error: (event, fields) => emitLog('error', fields ? [event, fields] : [event]),
189 };
190
191 const env = new Proxy({} as Record<string, string | undefined>, {
192 get(_, key: string) {
193 return Deno.env.get(key);
194 },
195 ownKeys() {
196 // Phase 1: customer iterates ctx.env at their own peril; allow-env
197 // gates which keys are even visible to Deno.env.
198 return [];
199 },
200 });
201
202 return {
203 db,
204 requestId,
205 log,
206 env,
207 auth,
208 };
209}
210
211function makeTableProxy(table: string): TableQuery {
212 return {
213 select: (columns?: readonly string[]) => makeSelectProxy(table, columns),
214 insert: (values) => makeInsertProxy(table, values),
215 update: (patch) => makeUpdateProxy(table, patch),
216 delete: () => makeDeleteProxy(table),
217 };
218}
219
220// ---------------------------------------------------------------------------
221// SELECT
222// ---------------------------------------------------------------------------
223
224interface SelectState {
225 where?: Record<string, unknown>;
226 orderBy?: { col: string; dir: 'asc' | 'desc' };
227 limit?: number;
228 offset?: number;
229}
230
231function makeSelectProxy(
232 table: string,
233 columns?: readonly string[],
234 state: SelectState = {},
235): SelectQuery {
236 const run = (): Promise<unknown[]> => {
237 const { sql, params } = buildSelectSql(table, columns, state);
238 return sendQuery({ sql, params, table });
239 };
240 return {
241 where(p) {
242 return makeSelectProxy(table, columns, { ...state, where: { ...state.where, ...p } });
243 },
244 orderBy(col, dir = 'asc') {
245 return makeSelectProxy(table, columns, { ...state, orderBy: { col, dir } });
246 },
247 limit(n) {
248 return makeSelectProxy(table, columns, { ...state, limit: n });
249 },
250 offset(n) {
251 return makeSelectProxy(table, columns, { ...state, offset: n });
252 },
253 then(onfulfilled, onrejected) {
254 return run().then(onfulfilled as never, onrejected as never);
255 },
256 };
257}
258
259// ---------------------------------------------------------------------------
260// INSERT
261// ---------------------------------------------------------------------------
262
263function makeInsertProxy(
264 table: string,
265 values: Record<string, unknown> | readonly Record<string, unknown>[],
266 returningCols: readonly string[] | null = null,
267): InsertQuery {
268 const run = (): Promise<unknown[]> => {
269 const { sql, params } = buildInsertSql(table, values, returningCols);
270 return sendQuery({ sql, params, table });
271 };
272 return {
273 returning(cols) {
274 const next = cols ?? [];
275 return makeInsertProxy(table, values, next);
276 },
277 then(onfulfilled, onrejected) {
278 return run().then(onfulfilled as never, onrejected as never);
279 },
280 };
281}
282
283// ---------------------------------------------------------------------------
284// UPDATE
285// ---------------------------------------------------------------------------
286
287interface UpdateState {
288 where?: Record<string, unknown>;
289}
290
291function makeUpdateProxy(
292 table: string,
293 patch: Record<string, unknown>,
294 state: UpdateState = {},
295 returningCols: readonly string[] | null = null,
296): UpdateQuery {
297 const run = (): Promise<unknown[]> => {
298 const { sql, params } = buildUpdateSql(table, patch, state, returningCols);
299 return sendQuery({ sql, params, table });
300 };
301 return {
302 where(p) {
303 return makeUpdateProxy(
304 table,
305 patch,
306 { ...state, where: { ...state.where, ...p } },
307 returningCols,
308 );
309 },
310 returning(cols) {
311 return makeUpdateProxy(table, patch, state, cols ?? []);
312 },
313 then(onfulfilled, onrejected) {
314 return run().then(onfulfilled as never, onrejected as never);
315 },
316 };
317}
318
319// ---------------------------------------------------------------------------
320// DELETE
321// ---------------------------------------------------------------------------
322
323interface DeleteState {
324 where?: Record<string, unknown>;
325}
326
327function makeDeleteProxy(
328 table: string,
329 state: DeleteState = {},
330 returningCols: readonly string[] | null = null,
331): DeleteQuery {
332 const run = (): Promise<unknown[]> => {
333 const { sql, params } = buildDeleteSql(table, state, returningCols);
334 return sendQuery({ sql, params, table });
335 };
336 return {
337 where(p) {
338 return makeDeleteProxy(table, { ...state, where: { ...state.where, ...p } }, returningCols);
339 },
340 returning(cols) {
341 return makeDeleteProxy(table, state, cols ?? []);
342 },
343 then(onfulfilled, onrejected) {
344 return run().then(onfulfilled as never, onrejected as never);
345 },
346 };
347}
348
349// ---------------------------------------------------------------------------
350// SQL builders — mirror apps/runtime/src/query-builder.ts identifier rules.
351// ---------------------------------------------------------------------------
352
353const IDENT = /^[a-zA-Z_][a-zA-Z0-9_]{0,62}$/;
354function quote(name: string): string {
355 if (!IDENT.test(name)) throw new Error(`invalid identifier: ${JSON.stringify(name)}`);
356 return `"${name}"`;
357}
358
359function quoteList(names: readonly string[]): string {
360 return names.map(quote).join(', ');
361}
362
363function buildSelectSql(
364 table: string,
365 columns: readonly string[] | undefined,
366 state: SelectState,
367): { sql: string; params: readonly unknown[] } {
368 const cols = columns ? quoteList(columns) : '*';
369 let sql = `SELECT ${cols} FROM ${quote(table)}`;
370 const params: unknown[] = [];
371 if (state.where && Object.keys(state.where).length > 0) {
372 const parts: string[] = [];
373 for (const [col, val] of Object.entries(state.where)) {
374 parts.push(`${quote(col)} = $${params.length + 1}`);
375 params.push(val);
376 }
377 sql += ' WHERE ' + parts.join(' AND ');
378 }
379 if (state.orderBy) sql += ` ORDER BY ${quote(state.orderBy.col)} ${state.orderBy.dir.toUpperCase()}`;
380 if (state.limit != null) sql += ` LIMIT ${Number(state.limit)}`;
381 if (state.offset != null) sql += ` OFFSET ${Number(state.offset)}`;
382 return { sql, params };
383}
384
385function buildInsertSql(
386 table: string,
387 values: Record<string, unknown> | readonly Record<string, unknown>[],
388 returningCols: readonly string[] | null,
389): { sql: string; params: readonly unknown[] } {
390 const rows = Array.isArray(values) ? values : [values as Record<string, unknown>];
391 if (rows.length === 0) {
392 // Postgres has no useful zero-row INSERT shape; surface explicitly so the
393 // host doesn't try to send malformed SQL.
394 throw new Error('insert() called with empty array');
395 }
396 const cols = Object.keys(rows[0]!);
397 const colSql = quoteList(cols);
398 const params: unknown[] = [];
399 const valueRows = rows.map((r) => {
400 const placeholders = cols.map((c) => {
401 params.push(r[c]);
402 return `$${params.length}`;
403 });
404 return `(${placeholders.join(', ')})`;
405 });
406 let sql = `INSERT INTO ${quote(table)} (${colSql}) VALUES ${valueRows.join(', ')}`;
407 if (returningCols !== null) {
408 const ret = returningCols.length === 0 ? '*' : quoteList(returningCols);
409 sql += ` RETURNING ${ret}`;
410 }
411 return { sql, params };
412}
413
414function buildUpdateSql(
415 table: string,
416 patch: Record<string, unknown>,
417 state: UpdateState,
418 returningCols: readonly string[] | null,
419): { sql: string; params: readonly unknown[] } {
420 const setParts: string[] = [];
421 const params: unknown[] = [];
422 for (const [col, value] of Object.entries(patch)) {
423 params.push(value);
424 setParts.push(`${quote(col)} = $${params.length}`);
425 }
426 if (setParts.length === 0) {
427 throw new Error('update() called with empty patch');
428 }
429 let sql = `UPDATE ${quote(table)} SET ${setParts.join(', ')}`;
430 if (state.where && Object.keys(state.where).length > 0) {
431 const whereParts: string[] = [];
432 for (const [col, val] of Object.entries(state.where)) {
433 params.push(val);
434 whereParts.push(`${quote(col)} = $${params.length}`);
435 }
436 sql += ' WHERE ' + whereParts.join(' AND ');
437 }
438 if (returningCols !== null) {
439 const ret = returningCols.length === 0 ? '*' : quoteList(returningCols);
440 sql += ` RETURNING ${ret}`;
441 }
442 return { sql, params };
443}
444
445function buildDeleteSql(
446 table: string,
447 state: DeleteState,
448 returningCols: readonly string[] | null,
449): { sql: string; params: readonly unknown[] } {
450 let sql = `DELETE FROM ${quote(table)}`;
451 const params: unknown[] = [];
452 if (state.where && Object.keys(state.where).length > 0) {
453 const whereParts: string[] = [];
454 for (const [col, val] of Object.entries(state.where)) {
455 params.push(val);
456 whereParts.push(`${quote(col)} = $${params.length}`);
457 }
458 sql += ' WHERE ' + whereParts.join(' AND ');
459 }
460 if (returningCols !== null) {
461 const ret = returningCols.length === 0 ? '*' : quoteList(returningCols);
462 sql += ` RETURNING ${ret}`;
463 }
464 return { sql, params };
465}
466
467// ---------------------------------------------------------------------------
468// Loop
469// ---------------------------------------------------------------------------
470
471export async function runQuery(): Promise<never> {
472 throw new Error('runQuery is reserved for future use');
473}
474
475export async function runIsolateLoop(
476 dispatch: Record<string, (ctx: Ctx, args: unknown) => Promise<unknown> | unknown>,
477 deploymentId: string,
478 // Per-function load failures captured by __entry.ts. A function whose file
479 // had a bad import (or no valid export) lands here instead of in `dispatch`;
480 // invoking it surfaces this message rather than a generic "not found". Empty
481 // in the healthy case.
482 loadErrors: Record<string, string> = {},
483): Promise<void> {
484 // Send ready handshake.
485 await writeStdout({ type: 'ready', deploymentId });
486
487 const decoder = new TextDecoder();
488 let buf = '';
489 for await (const chunk of Deno.stdin.readable) {
490 buf += decoder.decode(chunk, { stream: true });
491 let nl: number;
492 while ((nl = buf.indexOf('\n')) !== -1) {
493 const line = buf.slice(0, nl);
494 buf = buf.slice(nl + 1);
495 if (!line) continue;
496 let msg: { type: string } & Record<string, unknown>;
497 try {
498 msg = JSON.parse(line);
499 } catch {
500 // Malformed input — surface a structured error to the host (when we
501 // know the in-flight request) before exiting.
502 if (currentRequestId !== null) {
503 await writeStdout({
504 type: 'error',
505 requestId: currentRequestId,
506 code: 'isolate_protocol_error',
507 message: 'malformed input from host',
508 durationMs: 0,
509 }).catch(() => undefined);
510 }
511 Deno.exit(1);
512 }
513 if (msg.type === 'shutdown') {
514 Deno.exit(0);
515 } else if (msg.type === 'invoke') {
516 // Do NOT await — the invoked function may issue queries whose
517 // `query_result` replies arrive as *later* stdin messages handled by
518 // THIS same loop. Awaiting here blocks the reader, so the reply is
519 // never read, the query promise never resolves, and Deno aborts with
520 // "Top-level await promise never resolved" (a self-deadlock). Running
521 // it concurrently lets the loop keep reading; handleInvoke writes its
522 // own result/error frame to stdout when it finishes. The host
523 // serialises one invocation per isolate, so at most one is in flight.
524 void handleInvoke(msg as never, dispatch, loadErrors).catch(() => undefined);
525 } else if (msg.type === 'query_result') {
526 const m = msg as { qid: string; rows?: readonly unknown[]; error?: { code: string; message: string } };
527 const pending = pendingQueries.get(m.qid);
528 if (pending) {
529 pendingQueries.delete(m.qid);
530 if (m.error) pending.reject(new Error(m.error.message));
531 else pending.resolve(m.rows ?? []);
532 } else {
533 emitLog('warn', [`unknown qid: ${m.qid}`]);
534 }
535 }
536 }
537 }
538}
539
540async function handleInvoke(
541 msg: { requestId: string; functionName: string; args: unknown; auth: AuthContext | null },
542 dispatch: Record<string, (ctx: Ctx, args: unknown) => Promise<unknown> | unknown>,
543 loadErrors: Record<string, string> = {},
544): Promise<void> {
545 currentRequestId = msg.requestId;
546 const started = performance.now();
547 const fn = dispatch[msg.functionName];
548 // A function excluded from `dispatch` because its file failed to load gets a
549 // structured import error carrying the real cause, not a generic not-found.
550 const loadError = loadErrors[msg.functionName];
551 try {
552 if (!fn) {
553 if (loadError !== undefined) {
554 await writeStdout({
555 type: 'error',
556 requestId: msg.requestId,
557 code: 'import_blocked',
558 message: `function ${msg.functionName} failed to load: ${loadError}`,
559 durationMs: Math.round(performance.now() - started),
560 });
561 return;
562 }
563 throw new Error(`function not found: ${msg.functionName}`);
564 }
565 const ctx = makeCtx(msg.auth, msg.requestId);
566 const value = await fn(ctx, msg.args);
567 const durationMs = Math.round(performance.now() - started);
568 try {
569 await writeStdout({
570 type: 'result',
571 requestId: msg.requestId,
572 value,
573 durationMs,
574 });
575 } catch {
576 // Most likely the value contains something JSON.stringify can't
577 // serialise (cyclic ref, BigInt, etc.). Fall back to a structured
578 // error frame so the host doesn't hang waiting for a result.
579 await writeStdout({
580 type: 'error',
581 requestId: msg.requestId,
582 code: 'function_threw',
583 message: 'result not JSON-serializable',
584 durationMs,
585 }).catch(() => undefined);
586 }
587 } catch (err) {
588 const e = err as { name?: string; message?: string };
589 let code = 'function_threw';
590 if (e?.name === 'PermissionDenied') {
591 code = (e.message ?? '').includes('net') ? 'network_blocked'
592 : (e.message ?? '').includes('env') ? 'env_access_denied'
593 : 'fs_access_denied';
594 }
595 await writeStdout({
596 type: 'error',
597 requestId: msg.requestId,
598 code,
599 message: e?.message ?? String(err),
600 durationMs: Math.round(performance.now() - started),
601 });
602 } finally {
603 currentRequestId = null;
604 }
605}