Merge pull request #1258 from mykonos-ibiza/move-to-supabase-cron

refactor: replace QStash with Supabase Cron for background job processing

Author: escapade-mckv

.cursor/rules/suna-project.mdc

@@ -70,7 +70,6 @@ project/
 - **LLM Integration**: LiteLLM for multi-provider support, structured prompts
 - **Tool System**: Dual schema decorators (OpenAPI + XML), consistent ToolResult
 - **Real-time**: Supabase subscriptions for live updates
-- **Background Jobs**: Dramatiq for async processing, QStash for scheduling
 
 ## Key Technologies
 

CONTRIBUTING.md

@@ -41,7 +41,6 @@ Before contributing, ensure you have access to:
 - Daytona account (for agent execution)
 - Tavily API key (for search)
 - Firecrawl API key (for web scraping)
-- QStash account (for background jobs)
 
 **Optional:**
 

README.md

@@ -99,7 +99,6 @@ The setup process includes:
 - Setting up Daytona for secure agent execution
 - Integrating with LLM providers (Anthropic, OpenAI, OpenRouter, etc.)
 - Configuring web search and scraping capabilities (Tavily, Firecrawl)
-- Setting up QStash for background job processing and workflows
 - Configuring webhook handling for automated tasks
 - Optional integrations (RapidAPI for data providers)
 
@@ -147,14 +146,13 @@ We welcome contributions from the community! Please see our [Contributing Guide]
 ### Technologies
 
 - [Daytona](https://daytona.io/) - Secure agent execution environment
-- [Supabase](https://supabase.com/) - Database and authentication
+- [Supabase](https://supabase.com/) - Database, Cron, and Authentication
 - [Playwright](https://playwright.dev/) - Browser automation
 - [OpenAI](https://openai.com/) - LLM provider
 - [Anthropic](https://www.anthropic.com/) - LLM provider
 - [Morph](https://morphllm.com/) - For AI-powered code editing
 - [Tavily](https://tavily.com/) - Search capabilities
 - [Firecrawl](https://firecrawl.dev/) - Web scraping capabilities
-- [QStash](https://upstash.com/qstash) - Background job processing and workflows
 - [RapidAPI](https://rapidapi.com/) - API services
 - Custom MCP servers - Extend functionality with custom tools
 

backend/.env.example

@@ -54,11 +54,6 @@ SMITHERY_API_KEY=
 
 MCP_CREDENTIAL_ENCRYPTION_KEY=
 
-QSTASH_URL="https://qstash.upstash.io"
-QSTASH_TOKEN=""
-QSTASH_CURRENT_SIGNING_KEY=""
-QSTASH_NEXT_SIGNING_KEY=""
-
 WEBHOOK_BASE_URL=""
 
 # Optional

backend/README.md

@@ -96,11 +96,6 @@ DAYTONA_API_KEY=your-daytona-key
 DAYTONA_SERVER_URL=https://app.daytona.io/api
 DAYTONA_TARGET=us
 
-# Background Job Processing (Required)
-QSTASH_URL=https://qstash.upstash.io
-QSTASH_TOKEN=your-qstash-token
-QSTASH_CURRENT_SIGNING_KEY=your-current-signing-key
-QSTASH_NEXT_SIGNING_KEY=your-next-signing-key
 WEBHOOK_BASE_URL=https://yourdomain.com
 
 # MCP Configuration

backend/pyproject.toml

@@ -58,7 +58,6 @@ dependencies = [
   "cryptography>=41.0.0",
   "apscheduler>=3.10.0",
   "croniter>=1.4.0",
-  "qstash>=2.0.0",
   "structlog==25.4.0",
   "PyPDF2==3.0.1",
   "python-docx==1.1.0",

backend/supabase/migrations/20250808120000_supabase_cron.sql

@@ -0,0 +1,96 @@
+-- Enable Supabase Cron and HTTP extensions and provide helper RPCs
+-- This migration replaces QStash-based scheduling with Supabase Cron
+
+BEGIN;
+
+-- Enable required extensions if not already enabled
+CREATE EXTENSION IF NOT EXISTS pg_cron;
+CREATE EXTENSION IF NOT EXISTS pg_net;
+
+-- Helper function to schedule an HTTP POST via Supabase Cron
+-- Overwrites existing job with the same name
+CREATE OR REPLACE FUNCTION public.schedule_trigger_http(
+    job_name text,
+    schedule text,
+    url text,
+    headers jsonb DEFAULT '{}'::jsonb,
+    body jsonb DEFAULT '{}'::jsonb,
+    timeout_ms integer DEFAULT 8000
+) RETURNS bigint
+LANGUAGE plpgsql
+SECURITY DEFINER
+AS $$
+DECLARE
+    job_id bigint;
+    sql_text text;
+    headers_fixed jsonb;
+    body_fixed jsonb;
+BEGIN
+    -- Unschedule any existing jobs with the same name
+    PERFORM cron.unschedule(j.jobid)
+    FROM cron.job j
+    WHERE j.jobname = job_name;
+
+    -- Normalize headers/body in case callers pass JSON strings instead of objects
+    headers_fixed := COALESCE(
+        CASE
+            WHEN headers IS NULL THEN '{}'::jsonb
+            WHEN jsonb_typeof(headers) = 'object' THEN headers
+            WHEN jsonb_typeof(headers) = 'string' THEN (
+                -- Remove surrounding quotes then unescape to get raw JSON text, finally cast to jsonb
+                replace(replace(trim(both '"' from headers::text), '\\"', '"'), '\\\\', '\\')
+            )::jsonb
+            ELSE '{}'::jsonb
+        END,
+        '{}'::jsonb
+    );
+
+    body_fixed := COALESCE(
+        CASE
+            WHEN body IS NULL THEN '{}'::jsonb
+            WHEN jsonb_typeof(body) = 'object' THEN body
+            WHEN jsonb_typeof(body) = 'string' THEN (
+                replace(replace(trim(both '"' from body::text), '\\"', '"'), '\\\\', '\\')
+            )::jsonb
+            ELSE body
+        END,
+        '{}'::jsonb
+    );
+
+    -- Build the SQL snippet to be executed by pg_cron
+    sql_text := format(
+        $sql$select net.http_post(
+            url := %L,
+            headers := %L::jsonb,
+            body := %L::jsonb,
+            timeout_milliseconds := %s
+        );$sql$,
+        url,
+        headers_fixed::text,
+        body_fixed::text,
+        timeout_ms
+    );
+
+    job_id := cron.schedule(job_name, schedule, sql_text);
+    RETURN job_id;
+END;
+$$;
+
+-- Helper to unschedule by job name
+CREATE OR REPLACE FUNCTION public.unschedule_job_by_name(job_name text)
+RETURNS void
+LANGUAGE plpgsql
+SECURITY DEFINER
+AS $$
+BEGIN
+    PERFORM cron.unschedule(j.jobid)
+    FROM cron.job j
+    WHERE j.jobname = job_name;
+END;
+$$;
+
+-- Grant execute to service role (backend uses service role key)
+GRANT EXECUTE ON FUNCTION public.schedule_trigger_http(text, text, text, jsonb, jsonb, integer) TO service_role;
+GRANT EXECUTE ON FUNCTION public.unschedule_job_by_name(text) TO service_role;
+
+COMMIT;

backend/triggers/api.py

@@ -6,6 +6,7 @@
 import uuid
 from datetime import datetime, timezone
 import json
+import hmac
 
 from services.supabase import DBConnection
 from utils.auth_utils import get_current_user_id_from_jwt
@@ -452,6 +453,18 @@ async def trigger_webhook(
         raise HTTPException(status_code=403, detail="Agent triggers are not enabled")
 
     try:
+        # Simple header-based auth using a shared secret
+        # Configure the secret via environment variable: TRIGGER_WEBHOOK_SECRET
+        secret = os.getenv("TRIGGER_WEBHOOK_SECRET")
+        if not secret:
+            logger.error("TRIGGER_WEBHOOK_SECRET is not configured")
+            raise HTTPException(status_code=500, detail="Webhook secret not configured")
+
+        incoming_secret = request.headers.get("x-trigger-secret", "")
+        if not hmac.compare_digest(incoming_secret, secret):
+            logger.warning(f"Invalid webhook secret for trigger {trigger_id}")
+            raise HTTPException(status_code=401, detail="Unauthorized")
+
         # Get raw data from request
         raw_data = {}
         try:

backend/triggers/provider_service.py

@@ -7,7 +7,7 @@
 
 import croniter
 import pytz
-from qstash.client import QStash
+from services.supabase import DBConnection
 
 from services.supabase import DBConnection
 from utils.logger import logger
@@ -41,19 +41,11 @@ async def process_event(self, trigger: Trigger, event: TriggerEvent) -> TriggerR
 class ScheduleProvider(TriggerProvider):
     def __init__(self):
         super().__init__("schedule", TriggerType.SCHEDULE)
-        self._qstash_token = os.getenv("QSTASH_TOKEN")
-        self._webhook_base_url = os.getenv("WEBHOOK_BASE_URL", "http://localhost:3000")
-        
-        if not self._qstash_token:
-            logger.warning("QSTASH_TOKEN not found. Schedule provider will not work without it.")
-            self._qstash = None
-        else:
-            self._qstash = QStash(token=self._qstash_token)
+        # This should point to your backend base URL since Supabase Cron will POST to backend
+        self._webhook_base_url = os.getenv("WEBHOOK_BASE_URL", "http://localhost:8000")
+        self._db = DBConnection()
 
     async def validate_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
-        if not self._qstash:
-            raise ValueError("QSTASH_TOKEN environment variable is required for scheduled triggers")
-        
         if 'cron_expression' not in config:
             raise ValueError("cron_expression is required for scheduled triggers")
 
@@ -81,10 +73,6 @@ async def validate_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
         return config
 
     async def setup_trigger(self, trigger: Trigger) -> bool:
-        if not self._qstash:
-            logger.error("QStash client not available")
-            return False
-        
         try:
             webhook_url = f"{self._webhook_base_url}/api/triggers/{trigger.trigger_id}/webhook"
             cron_expression = trigger.config['cron_expression']
@@ -104,63 +92,71 @@ async def setup_trigger(self, trigger: Trigger) -> bool:
                 "timestamp": datetime.now(timezone.utc).isoformat()
             }
 
-            headers = {
+            headers: Dict[str, Any] = {
                 "Content-Type": "application/json",
                 "X-Trigger-Source": "schedule"
             }
-            
+
+            # Include simple shared secret header for backend auth
+            secret = os.getenv("TRIGGER_WEBHOOK_SECRET")
+            if secret:
+                headers["X-Trigger-Secret"] = secret
             if config.ENV_MODE == EnvMode.STAGING:
                 vercel_bypass_key = os.getenv("VERCEL_PROTECTION_BYPASS_KEY", "")
                 if vercel_bypass_key:
                     headers["X-Vercel-Protection-Bypass"] = vercel_bypass_key
-            
-            schedule_id = await asyncio.to_thread(
-                self._qstash.schedule.create,
-                destination=webhook_url,
-                cron=cron_expression,
-                body=json.dumps(payload),
-                headers=headers,
-                retries=3,
-                delay="5s"
-            )
-            
-            trigger.config['qstash_schedule_id'] = schedule_id
-            logger.info(f"Created QStash schedule {schedule_id} for trigger {trigger.trigger_id}")
+
+            # Supabase Cron job names are case-sensitive; we keep a stable name per trigger
+            job_name = f"trigger_{trigger.trigger_id}"
+
+            # Schedule via Supabase Cron RPC helper
+            client = await self._db.client
+            try:
+                result = await client.rpc(
+                    "schedule_trigger_http",
+                    {
+                        "job_name": job_name,
+                        "schedule": cron_expression,
+                        "url": webhook_url,
+                        "headers": headers,
+                        "body": payload,
+                        "timeout_ms": 8000,
+                    },
+                ).execute()
+            except Exception as rpc_err:
+                logger.error(f"Failed to schedule Supabase Cron job via RPC: {rpc_err}")
+                return False
+
+            trigger.config['cron_job_name'] = job_name
+            try:
+                trigger.config['cron_job_id'] = result.data
+            except Exception:
+                trigger.config['cron_job_id'] = None
+            logger.info(f"Created Supabase Cron job '{job_name}' for trigger {trigger.trigger_id}")
             return True
 
         except Exception as e:
-            logger.error(f"Failed to setup QStash schedule for trigger {trigger.trigger_id}: {e}")
+            logger.error(f"Failed to setup Supabase Cron schedule for trigger {trigger.trigger_id}: {e}")
             return False
 
     async def teardown_trigger(self, trigger: Trigger) -> bool:
-        if not self._qstash:
-            logger.warning("QStash client not available, skipping teardown")
-            return True
-        
         try:
-            schedule_id = trigger.config.get('qstash_schedule_id')
-            if schedule_id:
-                try:
-                    await asyncio.to_thread(self._qstash.schedule.delete, schedule_id)
-                    logger.info(f"Deleted QStash schedule {schedule_id} for trigger {trigger.trigger_id}")
-                    return True
-                except Exception as e:
-                    logger.warning(f"Failed to delete QStash schedule {schedule_id}: {e}")
-            
-            schedules = await asyncio.to_thread(self._qstash.schedule.list)
-            webhook_url = f"{self._webhook_base_url}/api/triggers/{trigger.trigger_id}/webhook"
-            
-            for schedule in schedules:
-                if schedule.get('destination') == webhook_url:
-                    await asyncio.to_thread(self._qstash.schedule.delete, schedule['scheduleId'])
-                    logger.info(f"Deleted QStash schedule {schedule['scheduleId']} for trigger {trigger.trigger_id}")
-                    return True
-            
-            logger.warning(f"No QStash schedule found for trigger {trigger.trigger_id}")
-            return True
+            job_name = trigger.config.get('cron_job_name') or f"trigger_{trigger.trigger_id}"
+            client = await self._db.client
+
+            try:
+                await client.rpc(
+                    "unschedule_job_by_name",
+                    {"job_name": job_name},
+                ).execute()
+                logger.info(f"Unschedule requested for Supabase Cron job '{job_name}' (trigger {trigger.trigger_id})")
+                return True
+            except Exception as rpc_err:
+                logger.warning(f"Failed to unschedule job '{job_name}' via RPC: {rpc_err}")
+                return False
 
         except Exception as e:
-            logger.error(f"Failed to teardown QStash schedule for trigger {trigger.trigger_id}: {e}")
+            logger.error(f"Failed to teardown Supabase Cron schedule for trigger {trigger.trigger_id}: {e}")
             return False
 
     async def process_event(self, trigger: Trigger, event: TriggerEvent) -> TriggerResult:

backend/triggers/trigger_service.py

@@ -135,7 +135,8 @@ async def update_trigger(
 
         trigger.updated_at = datetime.now(timezone.utc)
 
-        if config is not None or (is_active is True and not trigger.is_active):
+        # Reconcile provider scheduling when config changes or activation state toggles
+        if (config is not None) or (is_active is not None):
             from .provider_service import get_provider_service
             provider_service = get_provider_service(self._db)