chore: failfast client tests if connectivity issues

jumski · jumski · commit 2c8e42e828bc · 2025-11-13T00:13:36.000+01:00
diff --git a/.claude/skills/restack/SKILL.md b/.claude/skills/restack/SKILL.md
@@ -1,7 +1,7 @@
 ---
 name: restack
 description: Use when user asks to "restack", "resolve conflicts", "gt restack", or mentions restack conflicts. Guides through gt restack conflicts with intelligent diagnostics and resolution.
-allowed-tools: Bash(gt restack:*), Bash(gt continue:*), Bash(git show:*), Bash(git diff:*), Bash(git log:*), Bash(git ls-tree:*), Bash(pnpm install:*), Bash(git status:*), Bash(gt log:*), Bash(rm pnpm-lock.yaml), Bash(gt add -A:*),
+allowed-tools: Bash(gt restack:*), Bash(gt continue:*), Bash(git show:*), Bash(git diff:*), Bash(git log:*), Bash(git ls-tree:*), Bash(pnpm install:*), Bash(git status:*), Bash(gt log:*), Bash(rm pnpm-lock.yaml), Bash(gt add -A:*), Bash(git checkout --theirs:*), Bash(git checkout --ours:*)
 ---
 
 <critical>
diff --git a/pkgs/client/__tests__/dummy.test.ts b/pkgs/client/__tests__/dummy.test.ts
@@ -0,0 +1,7 @@
+import { describe, it, expect } from 'vitest';
+
+describe('Dummy', () => {
+  it('passes', () => {
+    expect(true).toBe(true);
+  });
+});
diff --git a/pkgs/client/__tests__/integration/connectivity-check.test.ts b/pkgs/client/__tests__/integration/connectivity-check.test.ts
@@ -0,0 +1,262 @@
+/**
+ * Connectivity smoke test for Supabase services required by pgflow client.
+ * This test validates all the connectivity patterns used by the client:
+ * 1. Realtime channel subscription and message broadcasting
+ * 2. Database queries (from() method)
+ * 3. RPC calls
+ *
+ * Run this test first in CI to quickly identify connectivity issues
+ * before running the full test suite.
+ */
+
+import { describe, it, expect, beforeAll } from 'vitest';
+import { createClient, SupabaseClient, RealtimeChannel } from '@supabase/supabase-js';
+import { createTestSupabaseClient } from '../helpers/setup.js';
+import { withPgNoTransaction } from '../helpers/db.js';
+import { grantMinimalPgflowPermissions } from '../helpers/permissions.js';
+
+describe('Supabase Connectivity Check', () => {
+  let supabase: SupabaseClient;
+
+  beforeAll(async () => {
+    // Use the same setup as other integration tests - service_role key
+    supabase = createTestSupabaseClient();
+
+    console.log('🔍 Supabase connectivity check starting...');
+    console.log(`   URL: http://localhost:50521`);
+    console.log(`   Using: service_role key (same as other integration tests)`);
+
+    // Grant permissions like other integration tests do
+    await withPgNoTransaction(async (sql) => {
+      await grantMinimalPgflowPermissions(sql);
+    })();
+  });
+
+  it('has valid Supabase configuration', () => {
+    const url = process.env.SUPABASE_URL || 'http://localhost:50521';
+    const key = process.env.SUPABASE_ANON_KEY ||
+      'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6ImFub24iLCJleHAiOjE5ODM4MTI5OTZ9.CRXP1A7WOeoJeXxjNni43kdQwgnWNReilDMblYTn_I0';
+
+    expect(url).toBeTruthy();
+    expect(key).toBeTruthy();
+    expect(url).toMatch(/^https?:\/\//);
+    expect(key).toMatch(/^eyJ/);  // JWT tokens start with eyJ
+  });
+
+  it('can establish Realtime channel subscription', async () => {
+    const channelName = `connectivity-test-${Date.now()}`;
+    let channel: RealtimeChannel | null = null;
+
+    try {
+      channel = supabase.channel(channelName);
+
+      // Track connection status
+      const statusPromise = new Promise<string>((resolve, reject) => {
+        const timeout = setTimeout(() => {
+          const currentStatus = channel?.state || 'unknown';
+          reject(new Error(
+            `Realtime subscription timeout after 10s. ` +
+            `Final status: ${currentStatus}. ` +
+            `This suggests Supabase Realtime is not accessible in CI.`
+          ));
+        }, 10000);
+
+        channel!.subscribe((status) => {
+          console.log(`   Channel status: ${status}`);
+
+          if (status === 'SUBSCRIBED') {
+            clearTimeout(timeout);
+            resolve(status);
+          } else if (status === 'CHANNEL_ERROR' || status === 'TIMED_OUT') {
+            clearTimeout(timeout);
+            reject(new Error(
+              `Channel subscription failed with status: ${status}. ` +
+              `This indicates Supabase Realtime connection issues.`
+            ));
+          }
+        });
+      });
+
+      const status = await statusPromise;
+      expect(status).toBe('SUBSCRIBED');
+      console.log('   ✓ Realtime subscription established');
+    } finally {
+      if (channel) {
+        await supabase.removeChannel(channel);
+      }
+    }
+  }, 15000);
+
+  it('can send and receive broadcast messages', async () => {
+    const channelName = `broadcast-test-${Date.now()}`;
+    const testMessage = { test: 'connectivity', timestamp: Date.now() };
+    let channel: RealtimeChannel | null = null;
+
+    try {
+      channel = supabase.channel(channelName);
+
+      // Set up message reception
+      const messagePromise = new Promise<any>((resolve, reject) => {
+        const timeout = setTimeout(() => {
+          reject(new Error(
+            'Broadcast message timeout after 10s. ' +
+            'Channel subscribed but messages not being received.'
+          ));
+        }, 10000);
+
+        // Listen for broadcast messages (pgflow client pattern)
+        channel!.on('broadcast', { event: '*' }, (payload) => {
+          console.log('   Received broadcast:', payload);
+          if (payload.event === 'test-event') {
+            clearTimeout(timeout);
+            resolve(payload.payload);
+          }
+        });
+
+        // Subscribe and send message
+        channel!.subscribe(async (status) => {
+          if (status === 'SUBSCRIBED') {
+            console.log('   Sending test broadcast...');
+            // Send using the pattern from SupabaseBroadcastAdapter
+            await channel!.send({
+              type: 'broadcast',
+              event: 'test-event',
+              payload: testMessage
+            });
+          }
+        });
+      });
+
+      const received = await messagePromise;
+      expect(received).toEqual(testMessage);
+      console.log('   ✓ Broadcast messages working');
+    } finally {
+      if (channel) {
+        await supabase.removeChannel(channel);
+      }
+    }
+  }, 15000);
+
+  it('can query pgflow database tables', async () => {
+    // Test the same queries used by SupabaseBroadcastAdapter
+    const { data: flowsData, error: flowsError } = await supabase
+      .schema('pgflow')
+      .from('flows')
+      .select('flow_slug')
+      .limit(1);
+
+    if (flowsError) {
+      throw new Error(
+        `Failed to query pgflow.flows table: ${flowsError.message}. ` +
+        `This suggests database connectivity or schema issues.`
+      );
+    }
+
+    // We don't need any flows to exist, just that the query works
+    expect(flowsError).toBeNull();
+    console.log('   ✓ Database query to pgflow.flows working');
+
+    // Test steps table query
+    const { data: stepsData, error: stepsError } = await supabase
+      .schema('pgflow')
+      .from('steps')
+      .select('step_slug')
+      .limit(1);
+
+    if (stepsError) {
+      throw new Error(
+        `Failed to query pgflow.steps table: ${stepsError.message}. ` +
+        `This suggests database schema issues.`
+      );
+    }
+
+    expect(stepsError).toBeNull();
+    console.log('   ✓ Database query to pgflow.steps working');
+  });
+
+  it('can call pgflow RPC functions', async () => {
+    // Test calling the RPC used by PgflowClient (with a fake run_id that won't exist)
+    const testRunId = '00000000-0000-0000-0000-000000000000';
+
+    const { data, error } = await supabase
+      .schema('pgflow')
+      .rpc('get_run_with_states', { run_id: testRunId });
+
+    // We expect no data for a fake run_id, but the RPC should execute without error
+    if (error && !error.message.includes('not found')) {
+      throw new Error(
+        `Failed to call pgflow.get_run_with_states RPC: ${error.message}. ` +
+        `This suggests RPC connectivity or permission issues.`
+      );
+    }
+
+    // The RPC should return null/empty for non-existent run, not an error
+    expect(error).toBeNull();
+    console.log('   ✓ RPC call to pgflow.get_run_with_states working');
+  });
+
+  it('validates full PgflowClient connection pattern', async () => {
+    // This test simulates the exact pattern PgflowClient uses:
+    // 1. Create channel with specific naming pattern
+    // 2. Subscribe to channel
+    // 3. Listen for events with wildcard pattern
+
+    const runId = `test-${Date.now()}`;
+    const channelName = `pgflow:run:${runId}`;
+    let channel: RealtimeChannel | null = null;
+
+    try {
+      console.log(`   Testing PgflowClient pattern with channel: ${channelName}`);
+      channel = supabase.channel(channelName);
+
+      const eventReceived = new Promise<boolean>((resolve, reject) => {
+        const timeout = setTimeout(() => {
+          reject(new Error(
+            `PgflowClient pattern timeout. Channel: ${channelName}, ` +
+            `Status: ${channel?.state}. ` +
+            `This is the exact pattern failing in the test suite.`
+          ));
+        }, 10000);
+
+        // Use the exact event pattern from SupabaseBroadcastAdapter
+        channel!.on('broadcast', { event: '*' }, (payload) => {
+          clearTimeout(timeout);
+          resolve(true);
+        });
+
+        channel!.subscribe(async (status) => {
+          console.log(`   PgflowClient channel status: ${status}`);
+          if (status === 'SUBSCRIBED') {
+            // Send a test event matching pgflow patterns
+            await channel!.send({
+              type: 'broadcast',
+              event: 'run:started',
+              payload: { run_id: runId, status: 'Started' }
+            });
+          } else if (status === 'TIMED_OUT' || status === 'CHANNEL_ERROR') {
+            clearTimeout(timeout);
+            reject(new Error(`Channel failed with status: ${status}`));
+          }
+        });
+      });
+
+      await expect(eventReceived).resolves.toBe(true);
+      console.log('   ✓ PgflowClient connection pattern working');
+    } finally {
+      if (channel) {
+        await supabase.removeChannel(channel);
+      }
+    }
+  }, 15000);
+
+  // Summary test that provides actionable feedback
+  it('provides connectivity summary', () => {
+    console.log('\n✅ All Supabase connectivity checks passed!');
+    console.log('   The client can successfully:');
+    console.log('   - Subscribe to Realtime channels');
+    console.log('   - Send and receive broadcast messages');
+    console.log('   - Query database tables');
+    console.log('   - Call RPC functions');
+    console.log('   - Use PgflowClient connection patterns');
+  });
+});
diff --git a/pkgs/client/test-stability.sh b/pkgs/client/test-stability.sh
@@ -0,0 +1,51 @@
+#!/bin/bash
+
+# Script to test channel stabilization delay reliability
+# Usage: ./test-stability.sh <number_of_iterations> [delay_ms]
+
+iterations=${1:-10}
+delay_ms=${2:-"current"}
+
+if [ "$delay_ms" != "current" ]; then
+  echo "Note: To change delay, you need to manually edit vitest.global-setup.ts"
+  echo "This script will test with the current delay setting"
+fi
+
+echo "Testing channel stabilization with $iterations iterations"
+echo "Current delay: Check vitest.global-setup.ts for actual value"
+echo "================================================"
+
+success_count=0
+fail_count=0
+
+for i in $(seq 1 $iterations); do
+  echo -n "Run $i/$iterations: "
+
+  # Run test and capture output
+  output=$(pnpm vitest run __tests__/dummy.test.ts 2>&1)
+
+  # Check if test passed
+  if echo "$output" | grep -q "Test Files  1 passed"; then
+    echo "✓ PASS"
+    success_count=$((success_count + 1))
+  else
+    echo "✗ FAIL"
+    fail_count=$((fail_count + 1))
+
+    # Show error if failed
+    if echo "$output" | grep -q "Supabase check failed"; then
+      error_msg=$(echo "$output" | grep "Supabase check failed" | head -1)
+      echo "  Error: $error_msg"
+    fi
+  fi
+done
+
+echo "================================================"
+echo "Results:"
+echo "  Success: $success_count/$iterations ($(( success_count * 100 / iterations ))%)"
+echo "  Failed:  $fail_count/$iterations ($(( fail_count * 100 / iterations ))%)"
+
+# Exit with non-zero if any failures
+if [ $fail_count -gt 0 ]; then
+  exit 1
+fi
diff --git a/pkgs/client/vite.config.ts b/pkgs/client/vite.config.ts
@@ -51,6 +51,7 @@ export default defineConfig({
       '__tests__/**/*.{test,spec}.{js,mjs,cjs,ts,mts,cts,jsx,tsx}'
     ],
     setupFiles: ['__tests__/setup.ts'],
+    globalSetup: './vitest.global-setup.ts',
     reporters: ['default'],
     coverage: {
       reportsDirectory: '../../coverage/pkgs/client',
diff --git a/pkgs/client/vitest.global-setup.ts b/pkgs/client/vitest.global-setup.ts
@@ -0,0 +1,53 @@
+import { createClient } from '@supabase/supabase-js';
+import postgres from 'postgres';
+import { createHash } from 'crypto';
+
+export async function setup() {
+  // Create a hash-based channel name with only alphanumeric characters
+  const timestamp = Date.now().toString();
+  const random = Math.random().toString();
+  const hash = createHash('sha1').update(timestamp + random).digest('hex');
+  const channelName = `setup${hash.substring(0, 16)}`; // Use first 16 chars of hash
+  console.log(`[GLOBAL SETUP] Using random channel: ${channelName}`);
+
+  const supabase = createClient('http://localhost:50521', 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6InNlcnZpY2Vfcm9sZSIsImV4cCI6MTk4MzgxMjk5Nn0.EGIM96RAZx35lJzdJsyH-qQwv8Hdp7fsn3W0YpN81IU');
+  const sql = postgres('postgresql://postgres:postgres@localhost:50522/postgres', { prepare: false, onnotice: () => {} });
+  const channel = supabase.channel(channelName);
+  const events: any[] = [];
+
+  try {
+    await sql`SELECT pgflow_tests.create_realtime_partition()`;
+    channel.on('broadcast', { event: '*' }, (p) => events.push(p));
+
+    await new Promise<void>((ok, fail) => {
+      const t = setTimeout(() => fail(new Error('timeout')), 10000);
+      channel.subscribe((s) => {
+        if (s === 'SUBSCRIBED') { clearTimeout(t); ok(); }
+        if (s === 'TIMED_OUT' || s === 'CHANNEL_ERROR') { clearTimeout(t); fail(new Error(s)); }
+      });
+    });
+
+    // Add stabilization delay for cold channels to fully establish routing
+    console.log('[GLOBAL SETUP] Channel subscribed, waiting 75ms for stabilization...');
+    await new Promise(resolve => setTimeout(resolve, 75));
+
+    await sql`SELECT realtime.send('{}', 'e', ${channelName}, false)`;
+
+    const start = Date.now();
+    while (events.length === 0 && Date.now() - start < 10000) {
+      await new Promise((ok) => setTimeout(ok, 100));
+    }
+
+    if (events.length === 0) throw new Error('realtime.send() failed');
+  } catch (e) {
+    console.error('\n❌ Supabase check failed:', e instanceof Error ? e.message : e, '\n');
+    process.exit(1);
+  } finally {
+    await supabase.removeChannel(channel);
+    await sql.end();
+  }
+}
+
+export async function teardown() {
+  // Nothing to clean up globally
+}
