feat(workers): Add dual-transport preflight system

Workers now run both curl and http (Puppeteer) preflights on startup: - curl-preflight.ts: Tests axios + proxy via httpbin.org - puppeteer-preflight.ts: Tests browser + StealthPlugin via fingerprint.com (with amiunique.org fallback) - Migration 084: Adds preflight columns to worker_registry and method column to worker_tasks - Workers report preflight status, IP, fingerprint, and response time - Tasks can require specific transport method (curl/http) - Dashboard shows Transport column with preflight status badges 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-11 22:47:52 -07:00
parent a35976b9e9
commit cdab71a1ee
7 changed files with 1085 additions and 8 deletions
--- a/backend/migrations/084_dual_transport_preflight.sql
+++ b/backend/migrations/084_dual_transport_preflight.sql
@@ -0,0 +1,253 @@
+-- Migration 084: Dual Transport Preflight System
+-- Workers run both curl and http (Puppeteer) preflights on startup
+-- Tasks can require a specific transport method
+
+-- ===================================================================
+-- PART 1: Add preflight columns to worker_registry
+-- ===================================================================
+
+-- Preflight status for curl/axios transport (proxy-based)
+ALTER TABLE worker_registry
+ADD COLUMN IF NOT EXISTS preflight_curl_status VARCHAR(20) DEFAULT 'pending';
+
+-- Preflight status for http/Puppeteer transport (browser-based)
+ALTER TABLE worker_registry
+ADD COLUMN IF NOT EXISTS preflight_http_status VARCHAR(20) DEFAULT 'pending';
+
+-- Timestamps for when each preflight completed
+ALTER TABLE worker_registry
+ADD COLUMN IF NOT EXISTS preflight_curl_at TIMESTAMPTZ;
+
+ALTER TABLE worker_registry
+ADD COLUMN IF NOT EXISTS preflight_http_at TIMESTAMPTZ;
+
+-- Error messages for failed preflights
+ALTER TABLE worker_registry
+ADD COLUMN IF NOT EXISTS preflight_curl_error TEXT;
+
+ALTER TABLE worker_registry
+ADD COLUMN IF NOT EXISTS preflight_http_error TEXT;
+
+-- Response time for successful preflights (ms)
+ALTER TABLE worker_registry
+ADD COLUMN IF NOT EXISTS preflight_curl_ms INTEGER;
+
+ALTER TABLE worker_registry
+ADD COLUMN IF NOT EXISTS preflight_http_ms INTEGER;
+
+-- Constraints for preflight status values
+ALTER TABLE worker_registry
+DROP CONSTRAINT IF EXISTS valid_preflight_curl_status;
+
+ALTER TABLE worker_registry
+ADD CONSTRAINT valid_preflight_curl_status
+CHECK (preflight_curl_status IN ('pending', 'passed', 'failed', 'skipped'));
+
+ALTER TABLE worker_registry
+DROP CONSTRAINT IF EXISTS valid_preflight_http_status;
+
+ALTER TABLE worker_registry
+ADD CONSTRAINT valid_preflight_http_status
+CHECK (preflight_http_status IN ('pending', 'passed', 'failed', 'skipped'));
+
+-- ===================================================================
+-- PART 2: Add method column to worker_tasks
+-- ===================================================================
+
+-- Transport method requirement for the task
+-- NULL = no preference (any worker can claim)
+-- 'curl' = requires curl/axios transport (proxy-based, fast)
+-- 'http' = requires http/Puppeteer transport (browser-based, anti-detect)
+ALTER TABLE worker_tasks
+ADD COLUMN IF NOT EXISTS method VARCHAR(10);
+
+-- Constraint for valid method values
+ALTER TABLE worker_tasks
+DROP CONSTRAINT IF EXISTS valid_task_method;
+
+ALTER TABLE worker_tasks
+ADD CONSTRAINT valid_task_method
+CHECK (method IS NULL OR method IN ('curl', 'http'));
+
+-- Index for method-based task claiming
+CREATE INDEX IF NOT EXISTS idx_worker_tasks_method
+  ON worker_tasks(method)
+  WHERE status = 'pending';
+
+-- Set default method for all existing pending tasks to 'http'
+-- ALL current tasks require Puppeteer/browser-based transport
+UPDATE worker_tasks
+SET method = 'http'
+WHERE method IS NULL;
+
+-- ===================================================================
+-- PART 3: Update claim_task function for method compatibility
+-- ===================================================================
+
+CREATE OR REPLACE FUNCTION claim_task(
+  p_role VARCHAR(50),
+  p_worker_id VARCHAR(100),
+  p_curl_passed BOOLEAN DEFAULT TRUE,
+  p_http_passed BOOLEAN DEFAULT FALSE
+) RETURNS worker_tasks AS $$
+DECLARE
+  claimed_task worker_tasks;
+BEGIN
+  UPDATE worker_tasks
+  SET
+    status = 'claimed',
+    worker_id = p_worker_id,
+    claimed_at = NOW(),
+    updated_at = NOW()
+  WHERE id = (
+    SELECT id FROM worker_tasks
+    WHERE role = p_role
+      AND status = 'pending'
+      AND (scheduled_for IS NULL OR scheduled_for <= NOW())
+      -- Method compatibility: worker must have passed the required preflight
+      AND (
+        method IS NULL  -- No preference, any worker can claim
+        OR (method = 'curl' AND p_curl_passed = TRUE)
+        OR (method = 'http' AND p_http_passed = TRUE)
+      )
+      -- Exclude stores that already have an active task
+      AND (dispensary_id IS NULL OR dispensary_id NOT IN (
+        SELECT dispensary_id FROM worker_tasks
+        WHERE status IN ('claimed', 'running')
+        AND dispensary_id IS NOT NULL
+      ))
+    ORDER BY priority DESC, created_at ASC
+    LIMIT 1
+    FOR UPDATE SKIP LOCKED
+  )
+  RETURNING * INTO claimed_task;
+
+  RETURN claimed_task;
+END;
+$$ LANGUAGE plpgsql;
+
+-- ===================================================================
+-- PART 4: Update v_active_workers view
+-- ===================================================================
+
+DROP VIEW IF EXISTS v_active_workers;
+
+CREATE VIEW v_active_workers AS
+SELECT
+  wr.id,
+  wr.worker_id,
+  wr.friendly_name,
+  wr.role,
+  wr.status,
+  wr.pod_name,
+  wr.hostname,
+  wr.started_at,
+  wr.last_heartbeat_at,
+  wr.last_task_at,
+  wr.tasks_completed,
+  wr.tasks_failed,
+  wr.current_task_id,
+  -- Preflight status
+  wr.preflight_curl_status,
+  wr.preflight_http_status,
+  wr.preflight_curl_at,
+  wr.preflight_http_at,
+  wr.preflight_curl_error,
+  wr.preflight_http_error,
+  wr.preflight_curl_ms,
+  wr.preflight_http_ms,
+  -- Computed fields
+  EXTRACT(EPOCH FROM (NOW() - wr.last_heartbeat_at)) as seconds_since_heartbeat,
+  CASE
+    WHEN wr.status = 'offline' THEN 'offline'
+    WHEN wr.last_heartbeat_at < NOW() - INTERVAL '2 minutes' THEN 'stale'
+    WHEN wr.current_task_id IS NOT NULL THEN 'busy'
+    ELSE 'ready'
+  END as health_status,
+  -- Capability flags (can this worker handle curl/http tasks?)
+  (wr.preflight_curl_status = 'passed') as can_curl,
+  (wr.preflight_http_status = 'passed') as can_http
+FROM worker_registry wr
+WHERE wr.status != 'terminated'
+ORDER BY wr.status = 'active' DESC, wr.last_heartbeat_at DESC;
+
+-- ===================================================================
+-- PART 5: View for task queue with method info
+-- ===================================================================
+
+DROP VIEW IF EXISTS v_task_history;
+
+CREATE VIEW v_task_history AS
+SELECT
+  t.id,
+  t.role,
+  t.dispensary_id,
+  d.name as dispensary_name,
+  t.platform,
+  t.status,
+  t.priority,
+  t.method,
+  t.worker_id,
+  t.scheduled_for,
+  t.claimed_at,
+  t.started_at,
+  t.completed_at,
+  t.error_message,
+  t.retry_count,
+  t.created_at,
+  EXTRACT(EPOCH FROM (t.completed_at - t.started_at)) as duration_sec
+FROM worker_tasks t
+LEFT JOIN dispensaries d ON d.id = t.dispensary_id
+ORDER BY t.created_at DESC;
+
+-- ===================================================================
+-- PART 6: Helper function to update worker preflight status
+-- ===================================================================
+
+CREATE OR REPLACE FUNCTION update_worker_preflight(
+  p_worker_id VARCHAR(100),
+  p_transport VARCHAR(10),  -- 'curl' or 'http'
+  p_status VARCHAR(20),     -- 'passed', 'failed', 'skipped'
+  p_response_ms INTEGER DEFAULT NULL,
+  p_error TEXT DEFAULT NULL
+) RETURNS VOID AS $$
+BEGIN
+  IF p_transport = 'curl' THEN
+    UPDATE worker_registry
+    SET
+      preflight_curl_status = p_status,
+      preflight_curl_at = NOW(),
+      preflight_curl_ms = p_response_ms,
+      preflight_curl_error = p_error,
+      updated_at = NOW()
+    WHERE worker_id = p_worker_id;
+  ELSIF p_transport = 'http' THEN
+    UPDATE worker_registry
+    SET
+      preflight_http_status = p_status,
+      preflight_http_at = NOW(),
+      preflight_http_ms = p_response_ms,
+      preflight_http_error = p_error,
+      updated_at = NOW()
+    WHERE worker_id = p_worker_id;
+  END IF;
+END;
+$$ LANGUAGE plpgsql;
+
+-- ===================================================================
+-- Comments
+-- ===================================================================
+
+COMMENT ON COLUMN worker_registry.preflight_curl_status IS 'Status of curl/axios preflight: pending, passed, failed, skipped';
+COMMENT ON COLUMN worker_registry.preflight_http_status IS 'Status of http/Puppeteer preflight: pending, passed, failed, skipped';
+COMMENT ON COLUMN worker_registry.preflight_curl_at IS 'When curl preflight completed';
+COMMENT ON COLUMN worker_registry.preflight_http_at IS 'When http preflight completed';
+COMMENT ON COLUMN worker_registry.preflight_curl_error IS 'Error message if curl preflight failed';
+COMMENT ON COLUMN worker_registry.preflight_http_error IS 'Error message if http preflight failed';
+COMMENT ON COLUMN worker_registry.preflight_curl_ms IS 'Response time of successful curl preflight (ms)';
+COMMENT ON COLUMN worker_registry.preflight_http_ms IS 'Response time of successful http preflight (ms)';
+
+COMMENT ON COLUMN worker_tasks.method IS 'Transport method required: NULL=any, curl=proxy-based, http=browser-based';
+
+COMMENT ON FUNCTION claim_task IS 'Atomically claim a task, respecting method requirements and per-store locking';
+COMMENT ON FUNCTION update_worker_preflight IS 'Update a workers preflight status for a given transport';