refactor: consolidate flow registration in index.ts and remove flows.ts (#437)

# Update Flow Registration to Use Single File

This PR simplifies the pgflow architecture by consolidating flow registration into a single file. Instead of using a separate `flows.ts` file, flows are now registered directly in the `index.ts` file of the pgflow edge function.

Key changes:
- Removed `flows.ts` and moved flow registration directly into `index.ts`
- Updated error messages and documentation to reference `index.ts` instead of `flows.ts`
- Added a new documentation page explaining the auto-compilation architecture
- Updated the getting started guide to reflect the new registration pattern
- Modified tests to match the new file structure

This change makes flow registration more straightforward for users and reduces the number of files they need to manage. The documentation now clearly explains how to register flows and how the auto-compilation process works.

Author: jumski

pkgs/cli/__tests__/commands/compile/index.test.ts

@@ -72,7 +72,7 @@ describe('fetchFlowSQL', () => {
       status: 404,
       json: async () => ({
         error: 'Flow Not Found',
-        message: "Flow 'unknown_flow' not found. Did you add it to flows.ts?",
+        message: "Flow 'unknown_flow' not found. Did you add it to supabase/functions/pgflow/index.ts?",
       }),
     };
 
@@ -83,7 +83,7 @@ describe('fetchFlowSQL', () => {
     ).rejects.toThrow("Flow 'unknown_flow' not found");
     await expect(
       fetchFlowSQL('unknown_flow', 'http://127.0.0.1:50621/functions/v1/pgflow', 'test-publishable-key')
-    ).rejects.toThrow('Add your flow to supabase/functions/pgflow/flows.ts');
+    ).rejects.toThrow('Add your flow to supabase/functions/pgflow/index.ts');
   });
 
   it('should handle ECONNREFUSED with startup instructions', async () => {
@@ -167,7 +167,7 @@ describe('fetchFlowSQL', () => {
     ).rejects.toThrow("Flow 'unknown_flow' not found");
     await expect(
       fetchFlowSQL('unknown_flow', 'http://127.0.0.1:50621/functions/v1/pgflow', 'test-publishable-key')
-    ).rejects.toThrow('Did you add it to flows.ts');
+    ).rejects.toThrow('Did you add it to supabase/functions/pgflow/index.ts');
   });
 
   it('should construct correct URL with flow slug', async () => {

pkgs/cli/__tests__/commands/install/create-edge-function.test.ts

@@ -22,7 +22,7 @@ describe('createEdgeFunction', () => {
     fs.rmSync(tempDir, { recursive: true, force: true });
   });
 
-  it('should create all three files when none exist', async () => {
+  it('should create both files when none exist', async () => {
     const result = await createEdgeFunction({
       supabasePath,
       autoConfirm: true,
@@ -36,23 +36,16 @@ describe('createEdgeFunction', () => {
 
     // Verify all files exist
     const indexPath = path.join(pgflowFunctionDir, 'index.ts');
-    const flowsPath = path.join(pgflowFunctionDir, 'flows.ts');
     const denoJsonPath = path.join(pgflowFunctionDir, 'deno.json');
 
     expect(fs.existsSync(indexPath)).toBe(true);
-    expect(fs.existsSync(flowsPath)).toBe(true);
     expect(fs.existsSync(denoJsonPath)).toBe(true);
 
-    // Verify index.ts content
+    // Verify index.ts content (inline flow registration, no flows.ts)
     const indexContent = fs.readFileSync(indexPath, 'utf8');
     expect(indexContent).toContain("import { ControlPlane } from '@pgflow/edge-worker'");
-    expect(indexContent).toContain("import { flows } from './flows.ts'");
-    expect(indexContent).toContain('ControlPlane.serve(flows)');
-
-    // Verify flows.ts content
-    const flowsContent = fs.readFileSync(flowsPath, 'utf8');
-    expect(flowsContent).toContain('export const flows = [');
-    expect(flowsContent).toContain('// Import your flows here');
+    expect(indexContent).toContain('ControlPlane.serve([');
+    expect(indexContent).toContain('// Import your flows here');
 
     // Verify deno.json content
     const denoJsonContent = fs.readFileSync(denoJsonPath, 'utf8');
@@ -66,11 +59,9 @@ describe('createEdgeFunction', () => {
     fs.mkdirSync(pgflowFunctionDir, { recursive: true });
 
     const indexPath = path.join(pgflowFunctionDir, 'index.ts');
-    const flowsPath = path.join(pgflowFunctionDir, 'flows.ts');
     const denoJsonPath = path.join(pgflowFunctionDir, 'deno.json');
 
     fs.writeFileSync(indexPath, '// existing content');
-    fs.writeFileSync(flowsPath, '// existing content');
     fs.writeFileSync(denoJsonPath, '// existing content');
 
     const result = await createEdgeFunction({
@@ -83,7 +74,6 @@ describe('createEdgeFunction', () => {
 
     // Verify files still exist with original content
     expect(fs.readFileSync(indexPath, 'utf8')).toBe('// existing content');
-    expect(fs.readFileSync(flowsPath, 'utf8')).toBe('// existing content');
     expect(fs.readFileSync(denoJsonPath, 'utf8')).toBe('// existing content');
   });
 
@@ -92,7 +82,6 @@ describe('createEdgeFunction', () => {
     fs.mkdirSync(pgflowFunctionDir, { recursive: true });
 
     const indexPath = path.join(pgflowFunctionDir, 'index.ts');
-    const flowsPath = path.join(pgflowFunctionDir, 'flows.ts');
     const denoJsonPath = path.join(pgflowFunctionDir, 'deno.json');
 
     // Only create index.ts
@@ -103,19 +92,15 @@ describe('createEdgeFunction', () => {
       autoConfirm: true,
     });
 
-    // Should return true because some files were created
+    // Should return true because deno.json was created
     expect(result).toBe(true);
 
     // Verify index.ts was not modified
     expect(fs.readFileSync(indexPath, 'utf8')).toBe('// existing content');
 
-    // Verify flows.ts and deno.json were created
-    expect(fs.existsSync(flowsPath)).toBe(true);
+    // Verify deno.json was created
     expect(fs.existsSync(denoJsonPath)).toBe(true);
 
-    const flowsContent = fs.readFileSync(flowsPath, 'utf8');
-    expect(flowsContent).toContain('export const flows = [');
-
     const denoJsonContent = fs.readFileSync(denoJsonPath, 'utf8');
     expect(denoJsonContent).toContain('"imports"');
   });
@@ -138,7 +123,6 @@ describe('createEdgeFunction', () => {
 
     // Verify files exist
     expect(fs.existsSync(path.join(pgflowFunctionDir, 'index.ts'))).toBe(true);
-    expect(fs.existsSync(path.join(pgflowFunctionDir, 'flows.ts'))).toBe(true);
     expect(fs.existsSync(path.join(pgflowFunctionDir, 'deno.json'))).toBe(true);
   });
 

pkgs/cli/src/commands/compile/index.ts

@@ -30,9 +30,9 @@ export async function fetchFlowSQL(
       const errorData = await response.json();
       throw new Error(
         `Flow '${flowSlug}' not found.\n\n` +
-          `${errorData.message || 'Did you add it to flows.ts?'}\n\n` +
+          `${errorData.message || 'Did you add it to supabase/functions/pgflow/index.ts?'}\n\n` +
           `Fix:\n` +
-          `1. Add your flow to supabase/functions/pgflow/flows.ts\n` +
+          `1. Add your flow to supabase/functions/pgflow/index.ts\n` +
           `2. Restart edge functions: supabase functions serve`
       );
     }
@@ -45,11 +45,6 @@ export async function fetchFlowSQL(
     return await response.json();
   } catch (error) {
     if (error instanceof Error) {
-      // Debug: show actual error and URL
-      console.error(`[DEBUG] Fetch failed for URL: ${url}`);
-      console.error(`[DEBUG] Error message: ${error.message}`);
-      console.error(`[DEBUG] Error cause:`, (error as any).cause);
-
       // Check for connection refused errors
       if (
         error.message.includes('ECONNREFUSED') ||

pkgs/cli/src/commands/install/create-edge-function.ts

@@ -5,18 +5,13 @@ import chalk from 'chalk';
 import { getVersion } from '../../utils/get-version.js';
 
 const INDEX_TS_TEMPLATE = `import { ControlPlane } from '@pgflow/edge-worker';
-import { flows } from './flows.ts';
-
-ControlPlane.serve(flows);
-`;
-
-const FLOWS_TS_TEMPLATE = `// Import your flows here
+// Import your flows here:
 // import { MyFlow } from '../_flows/my_flow.ts';
 
-// Export flows array for ControlPlane
-export const flows = [
+ControlPlane.serve([
+  // Add your flows here:
   // MyFlow,
-];
+]);
 `;
 
 const DENO_JSON_TEMPLATE = (version: string) => `{
@@ -44,7 +39,6 @@ export async function createEdgeFunction({
   const pgflowFunctionDir = path.join(functionsDir, 'pgflow');
 
   const indexPath = path.join(pgflowFunctionDir, 'index.ts');
-  const flowsPath = path.join(pgflowFunctionDir, 'flows.ts');
   const denoJsonPath = path.join(pgflowFunctionDir, 'deno.json');
 
   // Check what needs to be created
@@ -54,10 +48,6 @@ export async function createEdgeFunction({
     filesToCreate.push({ path: indexPath, name: 'index.ts' });
   }
 
-  if (!fs.existsSync(flowsPath)) {
-    filesToCreate.push({ path: flowsPath, name: 'flows.ts' });
-  }
-
   if (!fs.existsSync(denoJsonPath)) {
     filesToCreate.push({ path: denoJsonPath, name: 'deno.json' });
   }
@@ -69,7 +59,6 @@ export async function createEdgeFunction({
     const detailedMsg = [
       'Existing files:',
       `  ${chalk.dim('•')} ${chalk.bold('supabase/functions/pgflow/index.ts')}`,
-      `  ${chalk.dim('•')} ${chalk.bold('supabase/functions/pgflow/flows.ts')}`,
       `  ${chalk.dim('•')} ${chalk.bold('supabase/functions/pgflow/deno.json')}`,
     ].join('\n');
 
@@ -113,10 +102,6 @@ export async function createEdgeFunction({
     fs.writeFileSync(indexPath, INDEX_TS_TEMPLATE);
   }
 
-  if (filesToCreate.some((f) => f.path === flowsPath)) {
-    fs.writeFileSync(flowsPath, FLOWS_TS_TEMPLATE);
-  }
-
   if (filesToCreate.some((f) => f.path === denoJsonPath)) {
     fs.writeFileSync(denoJsonPath, DENO_JSON_TEMPLATE(getVersion()));
   }

pkgs/cli/supabase/functions/pgflow/flows.ts

@@ -1,4 +1,4 @@
 import { ControlPlane } from '@pgflow/edge-worker';
-import { flows } from './flows.ts';
+import { TestFlowE2E } from '../_flows/test_flow_e2e.ts';
 
-ControlPlane.serve(flows);
+ControlPlane.serve([TestFlowE2E]);

pkgs/cli/supabase/functions/pgflow/index.ts

@@ -7,9 +7,9 @@
  * @example
  * ```typescript
  * import { ControlPlane } from '@pgflow/edge-worker';
- * import { flows } from './flows.ts';
+ * import { MyFlow } from '../_flows/my_flow.ts';
  *
- * ControlPlane.serve(flows);
+ * ControlPlane.serve([MyFlow]);
  * ```
  */
 

pkgs/edge-worker/src/control-plane/index.ts

@@ -85,18 +85,15 @@ export function serveControlPlane(flows: AnyFlow[]): void {
 /**
  * Handles GET /flows/:slug requests
  */
-function handleGetFlow(
-  registry: Map<string, AnyFlow>,
-  slug: string
-): Response {
+function handleGetFlow(registry: Map<string, AnyFlow>, slug: string): Response {
   try {
     const flow = registry.get(slug);
 
     if (!flow) {
       return jsonResponse(
         {
           error: 'Flow Not Found',
-          message: `Flow '${slug}' not found. Did you add it to flows.ts?`,
+          message: `Flow '${slug}' not found. Did you add it to supabase/functions/pgflow/index.ts?`,
         },
         404
       );
@@ -134,17 +131,3 @@ function jsonResponse(data: unknown, status: number): Response {
     },
   });
 }
-
-/**
- * ControlPlane class for serving flow compilation HTTP API
- */
-export class ControlPlane {
-  /**
-   * Serves the ControlPlane HTTP API for flow compilation
-   * @param flows Array of flow definitions to register
-   */
-  static serve(flows: AnyFlow[]): void {
-    const handler = createControlPlaneHandler(flows);
-    Deno.serve({}, handler);
-  }
-}

pkgs/edge-worker/src/control-plane/server.ts

@@ -321,6 +321,10 @@ export default defineConfig({
                       link: '/concepts/three-layer-architecture/',
                     },
                     { label: 'Data model', link: '/concepts/data-model/' },
+                    {
+                      label: 'Compilation',
+                      link: '/concepts/compilation/',
+                    },
                   ],
                 },
                 {
@@ -366,6 +370,10 @@ export default defineConfig({
                     },
                     { label: 'Context API', link: '/reference/context/' },
                     { label: 'Compile API', link: '/reference/compile-api/' },
+                    {
+                      label: 'ControlPlane API',
+                      link: '/reference/control-plane-api/',
+                    },
                     {
                       label: 'Manual installation',
                       link: '/reference/manual-installation/',

pkgs/website/astro.config.mjs

@@ -35,6 +35,11 @@ Now that you've created your first flow, learn how to structure your code, integ
     href="/build/process-arrays-in-parallel/"
     description="Process arrays of data in parallel using map steps"
   />
+  <LinkCard
+    title="Validation steps"
+    href="/build/validation-steps/"
+    description="Use explicit validation steps to fail fast on invalid input"
+  />
 </CardGrid>
 
 ## Starting Flows