# Project export: Swarm AI This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: TreeHacks 2025 - Tagline: Swarm AI empowers AI companies to test thousands of speech-to-speech conversations in parallel, measuring latency, coherence, memory, & engagement. We turn weeks of QA & model evaluation into minutes. - Devpost: https://devpost.com/software/swarm-ai-s6lhd0 - GitHub: https://github.com/phombal/swarm-backend-new - Demo: https://github.com/sidjavvaji/swarm.git - Video: https://www.youtube.com/embed/IMKAHnOzx9E?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Result: winner (Codegen: Best Developer Tool ($1.5k Cash)) - Team: 1 GitHub contributor(s) — Pratham Hombal (21 commits) ## Devpost submission (written by the team) ### Inspiration Picture this: there we were, running our AI restaurant waiter, feeling confident about our automated system taking phone orders. But then came the testing phase, and reality quickly set in. Our only option was to manually call our AI repeatedly to validate each conversation flow – a process that’s extremely bottlenecked by a major limiting factor. Not exactly scalable when you're trying to ensure your AI can handle a million different things, ranging from simple reservations to complex menu customizations. We quickly realized we weren't alone in this challenge. From customer service to healthcare, everyone adopting voice AI was facing the same bottleneck. These AI agents needed to be thoroughly tested before deployment, but the testing tools hadn't caught up with the technology. In an age of automation, we were still relying on manual testing processes. And that didn’t sit well with us. That's when the idea for Swarm AI clicked: what if we could just create a platform that could spawn thousands of virtual callers, each with their own characteristics, accents, and conversation patterns? Just as load testing transformed web development from guesswork into a science, we believed voice AI testing needed its own revolution. By enabling developers to uncover edge cases and identify potential issues before they reach real customers, we could help ensure more reliable AI interactions across every industry. How it works At the core of Swarm AI is a system designed to run thousands of test calls at once. Users start by accessing our dashboard, where they can create and configure testing jobs through an intuitive interface. The dashboard lets them specify exactly how they want their test agents to behave – from setting specific conversation flows to selecting different accents, latency/network conditions, background noise, etc. for thorough testing coverage. Our backend, built with FastAPI, manages these testing requests through a smart batching system. Instead of starting all calls at once, we spread them out to keep everything running smoothly. Each batch runs independently, letting us handle many calls while keeping the system stable. For each call, we first save it to our database and then connect through Twilio. Once connected, calls flow through our real-time processing system. We use WebSockets to handle two-way audio, letting our AI agents listen and talk naturally. The audio is processed instantly using OpenAI for transcription, while our AI engine creates responses based on the test settings. The dashboard provides real-time visibility into ongoing tests through a live analytics panel. As calls progress, users can see key metrics updating in real-time – success rates, average call duration, and completion status for each test agent. The analytics interface pulls directly from our database, showing both aggregate statistics and detailed breakdowns of individual call performance. Our AI agents follow test settings that control their behavior, including accent, talking speed, and tone. These settings are loaded when each call starts, letting us test many different scenarios to find potential issues in the target voice AI system. Users can also access a detailed transcript view for any completed call, allowing them to analyze specific interactions or troubleshoot issues that arose during testing. After test completion, our platform generates comprehensive reports that highlight patterns, anomalies, and potential improvements for the voice AI system being tested. This data-driven approach helps users quickly identify and fix issues before they impact real customers. ### Challenges we ran into Our biggest technical hurdles centered around real-time communication and scalability. The WebSocket connection, important for maintaining live conversations between AI agents, proved particularly tricky – every disconnection meant a failed test and lost data. We overcame this by implementing robust connection handling and retry mechanisms. Call management also presented unique challenges. What seemed straightforward – ending a call – became complex when dealing with thousands of concurrent conversations. We had to carefully orchestrate call termination to ensure clean exits and proper resource cleanup. Our batching system underwent several iterations before we found the right balance between system load and testing throughput. One of our most interesting challenges was running simultaneous speech-to-text and speech-to-speech processing. This required careful stream management and precise timing to prevent feedback loops or processing delays. Figuring out how to transfer speech-based audio chunks efficiently and quickly proved was a major obstacle as well. After numerous debugging sessions and architecture revisions, we developed a stable solution that could handle both streams efficiently. Finally, our database architecture evolved significantly throughout development as we better understood our data needs. What started as a simple call logging system grew into a complex but efficient structure handling test configurations, real-time analytics, and detailed conversation transcripts. ### Accomplishments we're proud of The both of us poured our expertise into crafting a robust testing platform that exceeded our initial vision. The clean, beautiful interface we designed masks the complex orchestration happening behind the scenes – something we take pride in. We're especially proud of our system's reliability. Through persistent debugging and optimization, we created a platform that can handle many concurrent test calls while maintaining stable performance. But beyond the technical achievements, what stands out is how well we worked together. The both of us brought our strengths to the table and stepped up when needed, allowing us to build something substantial in such a short timeframe. What We Learned It’s incredibly difficult to just pick a few, but here are our major takeaways. First and foremost, a well-designed system architecture will always outperform spontaneous solutions (no matter how quickly we think we can move). We also dove deep into how computers listen to and process phone audio, and we explored various approaches to optimize real-time communication. Additionally, mastering WebSockets and WebRTC allowed us to handle two-way audio in a hyper-efficient manner, ensuring smooth interactions even at scale. Along the way, we gained a solid understanding of parallel computing and batch processing, applying principles from our systems classes to balance performance and compute resources effectively. We also learned the importance of having great company (and food + boba) when building. ### What's next While we initially set out to build a voice AI for restaurant ordering, developing this testing platform opened our eyes to a much bigger opportunity in the voice AI ecosystem. We're now pivoting our startup to focus on Swarm AI as a comprehensive testing platform for voice AI developers, with plans to expand our testing capabilities and add features like custom scenario builders, advanced analytics, and integration with popular voice AI development frameworks. Funny Moment Watching the AI agents making Dad jokes with one another (“What type of nut goes to space?” Answer: “An Astro-Nut”) ## README (from the GitHub repository) # Swarm Voice AI Testing Platform - Backend A sophisticated voice call platform that enables thousands of concurrent AI-powered phone conversations using OpenAI's GPT-4, Twilio for telephony, and Supabase for data storage. The platform supports configurable AI behaviors, batch calling, and detailed conversation analysis. ## Features - **AI-Powered Conversations**: Utilizes OpenAI's GPT-4 for natural, context-aware conversations - **Configurable AI Behavior**: Customize accent, industry context, speaking pace, and more - **Batch Call Support**: Make multiple calls simultaneously with controlled batching - **Real-time Transcription**: Capture and store conversation transcripts - **Conversation Analysis**: Detailed analysis of call quality, metrics, and performance - **WebSocket Integration**: Real-time audio streaming and processing - **Database Integration**: Persistent storage of call records and analytics ## Prerequisites - Python 3.8+ - OpenAI API Key - Twilio Account (Account SID and Auth Token) - Supabase Account (URL and API Key) - SSL Certificate for WebSocket connections ## Installation 1. Clone the repository: ```bash git clone cd ``` 2. Install dependencies: ```bash pip install -r requirements.txt ``` 3. Set up environment variables in `.env`: ```env OPENAI_API_KEY=your_openai_api_key TWILIO_ACCOUNT_SID=your_twilio_account_sid TWILIO_AUTH_TOKEN=your_twilio_auth_token TWILIO_PHONE_NUMBER=your_twilio_phone_number SUPABASE_URL=your_supabase_url SUPABASE_KEY=your_supabase_key ``` ## Usage ### Starting the Server ```bash uvicorn app.main:app --reload ``` ### Making a Test Call ```bash curl -X POST "https://your-domain/test-call?to_number=+1234567890" ``` ### Making Batch Calls ```bash curl -X POST "https://your-domain/batch-test-calls?to_number=+1234567890&num_calls=2" ``` ### Making Large Batch Calls (with controlled execution) ```bash curl -X POST "https://your-domain/execute_large_calls?to_number=+1234567890&total_calls=4" ``` ### Checking Call Status ```bash curl "https://your-domain/batch-status" ``` ### Getting Call Transcript ```bash curl "https://your-domain/transcript?call_sid=CAXXXXXXXXXXXXXXX" ``` ## Configuration ### Test Configuration Schema The platform supports customizing AI behavior through test configurations: ```json { "accent_types": ["neutral", "British", "American"], "industry": ["restaurant", "retail", "healthcare"], "speaking_pace": ["slow", "medium", "fast"], "emotion_types": ["professional", "friendly", "empathetic"], "background_noise": ["quiet", "moderate", "busy"], "max_turns": [5, 10, 15], "complexity_level": ["simple", "moderate", "complex"], "prompt_template": ["custom instruction templates"] } ``` ## Analysis Metrics The platform provides detailed analysis of each conversation, including: - Quality Metrics (coherence, task completion, context retention) - Technical Metrics (latency, token usage, memory usage) - Industry-Specific Metrics (order accuracy, required clarifications) - Semantic Analysis (intent classification, entity extraction) ## Error Handling The platform includes comprehensive error handling and logging: - Call status monitoring - WebSocket connection management - Database operation verification - API response validation ## Security - All API keys and sensitive data should be stored in environment variables - SSL/TLS encryption for WebSocket connections - Supabase authentication for database access ## Contributing 1. Fork the repository 2. Create a feature branch 3. Commit your changes 4. Push to the branch 5. Create a Pull Request ## License [Your License Here] ## Support For support, please [create an issue](your-issue-tracker-url) or contact [your-contact-info]. ## Detected evidence (automated analysis) Indexed codebase: 25 recognized source files, 120 KB. - FastAPI (technology) — detected in the code - OpenAI (technology) — detected in the code - Python (language) — detected in the code - SQL (language) — detected in the code - Supabase (technology) — detected in the code - JavaScript (language) — claimed on Devpost, not found in the code - Next.js (technology) — claimed on Devpost, not found in the code - Vercel (technology) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (32 of 32) ``` app/__init__.py app/config.py app/database.py app/main.py app/models/__init__.py app/models/simulation.py app/models/voice_conversation.py app/routers/__init__.py app/routers/media_stream.py app/routers/test_simulations.py app/routers/voice_router.py app/services/__init__.py app/services/analysis_service.py app/services/test_runner.py app/services/twilio_service.py app/utils.py app/voice_router.py app/websocket_handler.py migrations/create_tables.sql migrations/init.sql migrations/setup_database.sql pytest.ini README.md requirements.txt setup.py swarm_backend.egg-info/dependency_links.txt swarm_backend.egg-info/PKG-INFO swarm_backend.egg-info/requires.txt swarm_backend.egg-info/SOURCES.txt swarm_backend.egg-info/top_level.txt tests/conftest.py tests/test_database.py ``` ### Dependencies - requirements.txt: asyncio@==3.4.3, fastapi@==0.109.2, gotrue@>=1.3.0,<1.4.0, httpx@~=0.24.1, openai@==1.3.0, postgrest@>=0.10.8,<0.11.0, pydantic@>=2.1.0,<3.0.0, pytest@==8.0.0, pytest-asyncio@==0.23.5, pytest-env@==1.1.3, python-dotenv@==1.0.1, python-multipart@==0.0.7, realtime@>=1.0.0,<1.1.0, storage3@>=0.5.2,<0.6.0, supabase@==2.0.0, supafunc@>=0.3.0,<0.4.0, twilio@==8.11.0, uvicorn@==0.27.1, websockets@>=10.3,<11.0 ### Recent commits (newest first) - Update README.md - Update README.md - Added README file actulaly this time - Updated readme to be more currrent - Added customized system prompting and configurability - requirements.txt changes - requirements.txt changes - Added requirements.txt - Now it works for batch calls as well - tried to add end token, if bad roll back this push - changed render link - Figuring out supabase - changed supabase init again - changed supabase version - Works again with new db - The transcripting sort of works - Fixed system prompt issue - Transcripts on both sides - This is initiating the call and taking using realtime API - Init commit ## Key source files (fetched from GitHub, selected and truncated for size) ### requirements.txt ``` fastapi==0.109.2 uvicorn==0.27.1 twilio==8.11.0 python-dotenv==1.0.1 websockets>=10.3,<11.0 asyncio==3.4.3 supabase==2.0.0 python-multipart==0.0.7 pydantic>=2.1.0,<3.0.0 httpx~=0.24.1 pytest==8.0.0 pytest-asyncio==0.23.5 pytest-env==1.1.3 gotrue>=1.3.0,<1.4.0 postgrest>=0.10.8,<0.11.0 realtime>=1.0.0,<1.1.0 storage3>=0.5.2,<0.6.0 supafunc>=0.3.0,<0.4.0 openai==1.3.0 ``` ### app/main.py ```python from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import logging # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="Voice Call Platform") # Configure CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Import router after FastAPI initialization from app.voice_router import router as voice_router # Include router app.include_router(voice_router, tags=["Voice"]) @app.on_event("startup") async def startup_event(): from app.database import init_db await init_db() @app.get("/") async def root(): return {"message": "Voice Call Platform is running"} ``` ### setup.py ```python from setuptools import setup, find_packages setup( name="swarm-backend", version="0.1.0", packages=find_packages(), install_requires=[ "fastapi>=0.109.2", "uvicorn>=0.27.1", "supabase>=1.2.0", "python-multipart>=0.0.7", "pydantic>=2.6.1", "pytest>=8.0.0", "pytest-asyncio>=0.23.5", "pytest-env>=1.1.3", ], python_requires=">=3.8", ) ``` ### migrations/init.sql ```sql -- Enable UUID extension CREATE OR REPLACE FUNCTION create_uuid_extension() RETURNS void LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ BEGIN CREATE EXTENSION IF NOT EXISTS "uuid-ossp" SCHEMA public; END; $$; -- Create tables function CREATE OR REPLACE FUNCTION create_tables( simulations_sql text, call_records_sql text ) RETURNS void LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ BEGIN EXECUTE simulations_sql; EXECUTE call_records_sql; END; $$; -- Function to execute SQL statements CREATE OR REPLACE FUNCTION exec(sql text) RETURNS void LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ BEGIN EXECUTE sql; END; $$; ``` ### app/config.py ```python import os import ssl from dotenv import load_dotenv # Load environment variables load_dotenv() # Twilio Configuration TWILIO_ACCOUNT_SID = os.getenv("TWILIO_ACCOUNT_SID") TWILIO_AUTH_TOKEN = os.getenv("TWILIO_AUTH_TOKEN") TWILIO_PHONE_NUMBER = os.getenv("TWILIO_PHONE_NUMBER") # OpenAI Configuration OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") # Supabase Configuration SUPABASE_URL = os.getenv("SUPABASE_URL") SUPABASE_KEY = os.getenv("SUPABASE_KEY") # Voice Configuration DEFAULT_SYSTEM_MESSAGE = os.getenv("SYSTEM_MESSAGE", "You are a customer calling Bella Roma Italian restaurant. You are interested in ordering " "Italian food for dinner. You should ask about the menu, specials, and popular dishes. " "You're particularly interested in authentic Italian cuisine and might ask about appetizers, " "main courses, and desserts. Be friendly but also somewhat indecisive, as you want to hear " "about different options before making your choice. You can ask about ingredients, preparation " "methods, and portion sizes. If you like what you hear, you'll eventually place an order." ) # SSL Context for WebSocket connections ssl_context = ssl.create_default_context() ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE # Ensure required environment variables are set required_vars = [ "TWILIO_ACCOUNT_SID", "TWILIO_AUTH_TOKEN", "TWILIO_PHONE_NUMBER", "OPENAI_API_KEY", "SUPABASE_URL", "SUPABASE_KEY" ] missing_vars = [var for var in required_vars if not os.getenv(var)] if missing_vars: raise ValueError(f"Missing required environment variables: {', '.join(missing_vars)}") ``` ### app/utils.py ```python import json from app.config import SYSTEM_MESSAGE async def send_initial_conversation_item(openai_ws): initial_conversation_item = { "type": "conversation.item.create", "item": { "type": "message", "role": "user", "content": [ { "type": "input_text", "text": "Greet the user with 'Hello there! I am an AI voice assistant powered by Twilio and the OpenAI Realtime API. You can ask me for facts, jokes, or anything you can imagine. How can I help you?'" } ] } } await openai_ws.send(json.dumps(initial_conversation_item)) await openai_ws.send(json.dumps({"type": "response.create"})) async def initialize_session(openai_ws): session_update = { "type": "session.update", "session": { "turn_detection": {"type": "server_vad"}, "input_audio_format": "g711_ulaw", "output_audio_format": "g711_ulaw", "voice": "sage", "instructions": SYSTEM_MESSAGE, "modalities": ["text", "audio"], "temperature": 0.7, "input_audio_transcription": { "model": "whisper-1" } } } print('Sending session update:', json.dumps(session_update)) await openai_ws.send(json.dumps(session_update)) async def send_mark(connection, stream_sid): if stream_sid: mark_event = { "event": "mark", "streamSid": stream_sid, "mark": {"name": "responsePart"} } await connection.send_json(mark_event) return "responsePart" ``` ### tests/conftest.py ```python import pytest import asyncio from app.database import supabase_client # Mark all tests as async def pytest_collection_modifyitems(items): for item in items: item.add_marker(pytest.mark.asyncio) @pytest.fixture(scope="session") def event_loop(): """Create an instance of the default event loop for each test case.""" loop = asyncio.get_event_loop_policy().new_event_loop() yield loop loop.close() @pytest.fixture(autouse=True) async def setup_database(): """Setup and cleanup the test database before and after each test.""" try: # Clean up any existing test data # First delete call records for test simulations test_simulations = supabase_client.table("simulations").select("id").eq("user_id", "test_user").execute() if test_simulations.data: for sim in test_simulations.data: supabase_client.table("call_records").delete().eq("simulation_id", sim["id"]).execute() # Then delete test simulations supabase_client.table("simulations").delete().eq("user_id", "test_user").execute() yield # Clean up after the test # First delete call records for test simulations test_simulations = supabase_client.table("simulations").select("id").eq("user_id", "test_user").execute() if test_simulations.data: for sim in test_simulations.data: supabase_client.table("call_records").delete().eq("simulation_id", sim["id"]).execute() # Then delete test simulations supabase_client.table("simulations").delete().eq("user_id", "test_user").execute() except Exception as e: print(f"Error in database cleanup: {str(e)}") raise ``` ### migrations/create_tables.sql ```sql -- Enable UUID extension CREATE EXTENSION IF NOT EXISTS "uuid-ossp" SCHEMA public; -- Create simulations table CREATE TABLE IF NOT EXISTS simulations ( id UUID PRIMARY KEY DEFAULT uuid_generate_v4(), user_id TEXT NOT NULL, target_phone TEXT NOT NULL, concurrent_calls INTEGER NOT NULL, scenario JSONB NOT NULL, status TEXT NOT NULL, start_time TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, end_time TIMESTAMP WITH TIME ZONE, error TEXT, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ); -- Create call_records table CREATE TABLE IF NOT EXISTS voice_conversations ( id UUID PRIMARY KEY DEFAULT uuid_generate_v4(), simulation_id TEXT NOT NULL, call_sid TEXT NOT NULL, twilio_call_sid TEXT, phone_number TEXT, status TEXT NOT NULL, duration INTEGER, transcript JSONB DEFAULT '[]'::jsonb, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ); -- Update voice_conversations table ALTER TABLE voice_conversations ADD COLUMN IF NOT EXISTS twilio_call_sid TEXT, ADD COLUMN IF NOT EXISTS duration INTEGER, ALTER COLUMN transcript SET DEFAULT '[]'::jsonb, ALTER COLUMN status SET NOT NULL, ALTER COLUMN simulation_id SET NOT NULL, ALTER COLUMN call_sid SET NOT NULL; -- Add updated_at trigger if it doesn't exist CREATE OR REPLACE FUNCTION update_updated_at_column() RETURNS TRIGGER AS $$ BEGIN NEW.updated_at = CURRENT_TIMESTAMP; RETURN NEW; END; $$ language 'plpgsql'; DROP TRIGGER IF EXISTS update_voice_conversations_updated_at ON voice_conversations; CREATE TRIGGER update_voice_conversations_updated_at BEFORE UPDATE ON voice_conversations FOR EACH ROW EXECUTE FUNCTION update_updated_at_column(); ``` ### app/database.py ```python from typing import Dict, Optional import logging from datetime import datetime, UTC from uuid import uuid4 from supabase import create_client, Client from app.config import SUPABASE_URL, SUPABASE_KEY logger = logging.getLogger(__name__) try: # Initialize Supabase client with version 2.0.0 supabase_client: Client = create_client(SUPABASE_URL, SUPABASE_KEY) except Exception as e: logger.error(f"Failed to initialize Supabase client: {str(e)}") raise async def init_db(): """Initialize database connection.""" try: # Test the connection supabase_client.table("voice_conversations").select("count", count="exact").execute() logger.info("Database connection initialized successfully") except Exception as e: logger.error(f"Error initializing database connection: {str(e)}") raise async def create_call_record(simulation_id: str, call_sid: str, phone_number: str, user_id: str, status: str = "initiated") -> str: """Create a new voice conversation record.""" try: now = datetime.now(UTC).isoformat() result = supabase_client.table("voice_conversations").insert({ "id": str(uuid4()), "simulation_id": simulation_id, "call_sid": call_sid, "phone_number": phone_number, "status": status, "duration": None, "transcript": [], "message_timestamps": [], "token_counts": {}, "response_times": [], "error_details": [], "conversation_metrics": {}, "user_id": user_id, "created_at": now, "updated_at": now, "error_severity": None, "recovery_attempt": None, "recovery_success": None }).execute() return result.data[0]["id"] except Exception as e: logger.error(f"Error creating voice conversation record: {str(e)}") raise async def update_call_record( simulation_id: str, call_sid: str, updates: Dict ) -> bool: """Update a voice conversation record.""" try: # Ensure updated_at is set updates["updated_at"] = datetime.now(UTC).isoformat() # Initialize empty lists/dicts for JSON fields if they're None json_fields = ["transcript", "message_timestamps", "token_counts", "response_times", "error_details", "conversation_metrics"] for field in json_fields: if field in updates and updates[field] is None: updates[field] = [] if field != "token_counts" and field != "conversation_metrics" else {} # First try to find by call_sid response = supabase_client.table('voice_conversations')\ .update(updates)\ .eq('simulation_id', simulation_id)\ .eq('call_sid', call_sid)\ .execute() if not response.data: # If no record found by call_sid, try twilio_call_sid response = supabase_client.table('voice_conversations')\ .update(updates)\ .eq('simulation_id', simulation_id)\ .eq('twilio_call_sid', call_sid)\ .execute() return bool(response.data) except Exception as e: logger.error(f"Error updating voice conversation record: {str(e)}") return False ``` ### migrations/setup_database.sql ```sql -- Enable required extensions CREATE EXTENSION IF NOT EXISTS "uuid-ossp" SCHEMA public; -- Drop dependent objects first DROP VIEW IF EXISTS public.active_simulations; -- Drop tables if they exist (useful for resetting the database) DROP TABLE IF EXISTS public.call_records; DROP TABLE IF EXISTS public.simulations; -- Create enum for simulation status DO $$ BEGIN CREATE TYPE simulation_status AS ENUM ( 'initiated', 'running', 'completed', 'failed', 'stopped' ); EXCEPTION WHEN duplicate_object THEN NULL; END $$; -- Create enum for call status DO $$ BEGIN CREATE TYPE call_status AS ENUM ( 'initiated', 'ringing', 'in-progress', 'completed', 'failed', 'no-answer' ); EXCEPTION WHEN duplicate_object THEN NULL; END $$; -- Create simulations table CREATE TABLE public.simulations ( id UUID PRIMARY KEY DEFAULT uuid_generate_v4(), user_id TEXT NOT NULL, target_phone TEXT NOT NULL, concurrent_calls INTEGER NOT NULL CHECK (concurrent_calls > 0 AND concurrent_calls <= 100), scenario JSONB NOT NULL, status simulation_status NOT NULL DEFAULT 'initiated', start_time TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, end_time TIMESTAMP WITH TIME ZONE, error TEXT, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ); -- Create call_records table CREATE TABLE public.call_records ( id UUID PRIMARY KEY DEFAULT uuid_generate_v4(), simulation_id UUID NOT NULL REFERENCES public.simulations(id) ON DELETE CASCADE, call_sid TEXT NOT NULL, status call_status NOT NULL DEFAULT 'initiated', duration INTEGER CHECK (duration >= 0), transcript JSONB, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, UNIQUE(simulation_id, call_sid) ); -- Create indexes for better query performance CREATE INDEX idx_simulations_user_id ON public.simulations(user_id); CREATE INDEX idx_simulations_status ON public.simulations(status); CREATE INDEX idx_call_records_simulation_id ON public.call_records(simulation_id); CREATE INDEX idx_call_records_call_sid ON public.call_records(call_sid); -- Create function to update updated_at timestamp CREATE OR REPLACE FUNCTION update_updated_at_column() RETURNS TRIGGER AS $$ BEGIN NEW.updated_at = CURRENT_TIMESTAMP; RETURN NEW; END; $$ language 'plpgsql'; -- Create triggers to automatically update updated_at CREATE TRIGGER update_simulations_updated_at BEFORE UPDATE ON public.simulations FOR EACH ROW EXECUTE FUNCTION update_updated_at_column(); CREATE TRIGGER update_call_records_updated_at BEFORE UPDATE ON public.call_records FOR EACH ROW EXECUTE FUNCTION update_updated_at_column(); -- Create view for active simulations with call counts CREATE OR REPLACE VIEW public.active_simulations AS SELECT s.*, COUNT(cr.id) as active_calls, COUNT(CASE WHEN cr.status = 'completed' THEN 1 END) as completed_calls, COUNT(CASE WHEN cr.status = 'failed' THEN 1 END) as failed_calls FROM public.simulations s LEFT JOIN public.call_records cr ON s.id = cr.simulation_id WHERE s.status IN ('initiated', 'running') GROUP BY s.id; -- Grant necessary permissions ALTER TABLE public.simulations ENABLE ROW LEVEL SECURITY; ALTER TABLE public.call_records ENABLE ROW LEVEL SECURITY; -- Create policies for simulations table DROP POLICY IF EXISTS "Enable read access for all users" ON public.simulations; DROP POLICY IF EXISTS "Enable insert for all users" ON public.simulations; DROP POLICY IF EXISTS "Enable update for all users" ON public.simulations; CREATE POLICY "Enable read access for all users" ON public.simulations FOR SELECT USING (true); CREATE POLICY "Enable insert for all users" ON public.simulations FOR INSERT WITH CHECK ( auth.role() = 'authenticated' OR user_id ILIKE 'test%' ); CREATE POLICY "Enable update for all users" ON public.simulations FOR UPDATE USING ( auth.role() = 'authenticated' OR user_id ILIKE 'test%' ); -- Create policies for call_records table DROP POLICY IF EXISTS "Enable read access for all users" ON public.call_records; DROP POLICY IF EXISTS "Enable insert for all users" ON public.call_records; DROP POLICY IF EXISTS "Enable update for all users" ON public.call_records; CREATE POLICY "Enable read access for all users" ON public.call_records FOR SELECT USING (true); CREATE POLICY "Enable insert for all users" ON public.call_records FOR INSERT WITH CHECK ( EXISTS ( SELECT 1 FROM public.simulations s WHERE s.id = simulation_id AND ( auth.role() = 'authenticated' OR s.user_id ILIKE 'test%' ) ) ); CREATE POLICY "Enable update for all users" ON public.call_records FOR UPDATE USING ( EXISTS ( SELECT 1 FROM public.simulations s WHERE s.id = simulation_id AND ( auth.role() = 'authenticated' OR s.user_id ILIKE 'test%' ) ) ); -- Create function to clean up test data CREATE OR REPLACE FUNCTION cleanup_test_data() RETURNS void LANGUAGE plpgsql SECURITY DEFINER AS $$ BEGIN DELETE FROM public.simulations WHERE user_id ILIKE 'test%'; END; $$; ``` [11 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]