1 line
No EOL
55 KiB
Markdown
1 line
No EOL
55 KiB
Markdown
# Integration Automation Scripts\n\n## Rust Meetup 2025 - Complete Data Pipeline & Automation\n\nThis document provides complete automation scripts for integrating Typebot, Formbricks, Claper, and Baserow for seamless event management.\n\n---\n\n## 🏗️ **Architecture Overview**\n\n```plaintext\n┌─────────────┐ ┌─────────────┐ ┌─────────────┐\n│ Typebot │ │ Formbricks │ │ Claper │\n│(Conversational)│ │ (Surveys) │ │(Live Polls) │\n└──────┬──────┘ └──────┬──────┘ └──────┬──────┘\n │ │ │\n ▼ ▼ ▼\n┌─────────────────────────────────────────────────────┐\n│ Webhook Processor │\n│ (Python FastAPI Service) │\n└─────────────────────┬───────────────────────────────┘\n │\n ▼\n┌─────────────────────────────────────────────────────┐\n│ Baserow │\n│ (Central Database) │\n└─────────────────────────────────────────────────────┘\n```\n\n---\n\n## 🐍 **Webhook Processor Service**\n\n### **webhook-processor.py**\n\n```python\n#!/usr/bin/env python3\n"""\nWebhook Processor for Rust Meetup 2025\nHandles data integration between Typebot, Formbricks, Claper, and Baserow\n"""\n\nfrom fastapi import FastAPI, HTTPException, BackgroundTasks, Depends\nfrom fastapi.security import HTTPBearer, HTTPAuthorizationCredentials\nfrom pydantic import BaseModel, Field, validator\nfrom typing import Dict, List, Optional, Any\nimport httpx\nimport json\nimport logging\nimport os\nfrom datetime import datetime, timedelta\nimport asyncio\nimport hashlib\nimport hmac\nfrom enum import Enum\n\n# Configure logging\nlogging.basicConfig(\n level=logging.INFO,\n format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'\n)\nlogger = logging.getLogger(__name__)\n\n# Initialize FastAPI app\napp = FastAPI(\n title="Rust Meetup 2025 - Webhook Processor",\n description="Central webhook processor for event management data pipeline",\n version="1.0.0"\n)\n\n# Security\nsecurity = HTTPBearer()\n\n# Configuration\nclass Config:\n BASEROW_API_URL = os.getenv("BASEROW_API_URL", "https://baserow.rustmeetup.com/api")\n BASEROW_API_TOKEN = os.getenv("BASEROW_API_TOKEN")\n WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")\n\n # Table IDs in Baserow\n ATTENDEES_TABLE_ID = int(os.getenv("ATTENDEES_TABLE_ID", "1"))\n PRE_EVENT_TABLE_ID = int(os.getenv("PRE_EVENT_TABLE_ID", "2"))\n LIVE_ENGAGEMENT_TABLE_ID = int(os.getenv("LIVE_ENGAGEMENT_TABLE_ID", "3"))\n POST_EVENT_TABLE_ID = int(os.getenv("POST_EVENT_TABLE_ID", "4"))\n FOLLOW_UP_TABLE_ID = int(os.getenv("FOLLOW_UP_TABLE_ID", "5"))\n\n # Email configuration\n SMTP_HOST = os.getenv("SMTP_HOST", "smtp.gmail.com")\n SMTP_PORT = int(os.getenv("SMTP_PORT", "587"))\n SMTP_USER = os.getenv("SMTP_USER")\n SMTP_PASSWORD = os.getenv("SMTP_PASSWORD")\n\n # External service URLs\n DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL")\n SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")\n\nconfig = Config()\n\n# Pydantic Models\nclass SourceSystem(str, Enum):\n TYPEBOT = "typebot"\n FORMBRICKS = "formbricks"\n CLAPER = "claper"\n MANUAL = "manual"\n\nclass WebhookPayload(BaseModel):\n source: SourceSystem\n event_type: str\n data: Dict[str, Any]\n timestamp: Optional[datetime] = None\n\n @validator('timestamp', pre=True, always=True)\n def set_timestamp(cls, v):\n return v or datetime.utcnow()\n\nclass TypebotResponse(BaseModel):\n typebot_id: str\n result_id: str\n answers: Dict[str, Any]\n created_at: datetime\n\nclass FormbricksResponse(BaseModel):\n survey_id: str\n response_id: str\n person_id: Optional[str]\n answers: Dict[str, Any]\n created_at: datetime\n\nclass ClaperEngagement(BaseModel):\n session_id: str\n participant_id: str\n poll_responses: List[Dict[str, Any]]\n questions_submitted: List[str]\n session_duration: int\n engagement_metrics: Dict[str, Any]\n\n# Utility Functions\ndef verify_webhook_signature(payload: bytes, signature: str) -> bool:\n """Verify webhook signature for security"""\n if not config.WEBHOOK_SECRET:\n return True # Skip verification if no secret is set\n\n expected_signature = hmac.new(\n config.WEBHOOK_SECRET.encode(),\n payload,\n hashlib.sha256\n ).hexdigest()\n\n return hmac.compare_digest(f"sha256={expected_signature}", signature)\n\nasync def get_baserow_client():\n """Get authenticated Baserow HTTP client"""\n return httpx.AsyncClient(\n headers={\n "Authorization": f"Token {config.BASEROW_API_TOKEN}",\n "Content-Type": "application/json"\n },\n timeout=30.0\n )\n\ndef calculate_engagement_score(\n polls_participated: int,\n questions_asked: int,\n session_duration: int\n) -> float:\n """Calculate engagement score based on activity metrics"""\n if session_duration == 0:\n return 0.0\n\n # Weighted scoring: polls * 2 + questions * 3 + duration factor\n duration_factor = min(session_duration / 60, 1.5) # Cap at 1.5 hours\n score = (polls_participated * 2 + questions_asked * 3 + duration_factor) / 3\n\n return round(min(score, 10), 1) # Cap at 10\n\n# Baserow API Functions\nasync def find_attendee_by_email(email: str) -> Optional[Dict]:\n """Find attendee record by email address"""\n async with await get_baserow_client() as client:\n try:\n response = await client.get(\n f"{config.BASEROW_API_URL}/database/rows/table/{config.ATTENDEES_TABLE_ID}/",\n params={"filter__email__equal": email}\n )\n response.raise_for_status()\n\n results = response.json().get("results", [])\n return results[0] if results else None\n\n except Exception as e:\n logger.error(f"Error finding attendee by email {email}: {e}")\n return None\n\nasync def create_attendee_record(attendee_data: Dict) -> Optional[Dict]:\n """Create new attendee record in Baserow"""\n async with await get_baserow_client() as client:\n try:\n response = await client.post(\n f"{config.BASEROW_API_URL}/database/rows/table/{config.ATTENDEES_TABLE_ID}/",\n json=attendee_data\n )\n response.raise_for_status()\n return response.json()\n\n except Exception as e:\n logger.error(f"Error creating attendee record: {e}")\n return None\n\nasync def update_attendee_record(attendee_id: int, update_data: Dict) -> bool:\n """Update existing attendee record"""\n async with await get_baserow_client() as client:\n try:\n response = await client.patch(\n f"{config.BASEROW_API_URL}/database/rows/table/{config.ATTENDEES_TABLE_ID}/{attendee_id}/",\n json=update_data\n )\n response.raise_for_status()\n return True\n\n except Exception as e:\n logger.error(f"Error updating attendee {attendee_id}: {e}")\n return False\n\nasync def create_table_record(table_id: int, record_data: Dict) -> Optional[Dict]:\n """Create record in any Baserow table"""\n async with await get_baserow_client() as client:\n try:\n response = await client.post(\n f"{config.BASEROW_API_URL}/database/rows/table/{table_id}/",\n json=record_data\n )\n response.raise_for_status()\n return response.json()\n\n except Exception as e:\n logger.error(f"Error creating record in table {table_id}: {e}")\n return None\n\n# Data Processing Functions\nasync def process_typebot_registration(data: TypebotResponse) -> bool:\n """Process Typebot registration data"""\n try:\n answers = data.answers\n\n # Extract attendee information\n email = answers.get("email_input", "")\n name = answers.get("name_input", "")\n\n if not email or not name:\n logger.warning("Missing required fields in Typebot response")\n return False\n\n # Check if attendee already exists\n existing_attendee = await find_attendee_by_email(email)\n\n if existing_attendee:\n # Update existing attendee\n update_data = {\n "pre_survey_completed": True,\n "last_activity": datetime.utcnow().isoformat()\n }\n await update_attendee_record(existing_attendee["id"], update_data)\n attendee_id = existing_attendee["id"]\n else:\n # Create new attendee\n attendee_data = {\n "email": email,\n "full_name": name,\n "professional_role": answers.get("role", ""),\n "experience_years": answers.get("experience_years", ""),\n "rust_experience_level": answers.get("rust_experience", ""),\n "company_size": answers.get("company_size", ""),\n "attendance_status": "registered",\n "pre_survey_completed": True,\n "communication_preferences": answers.get("communication_preferences", [])\n }\n\n created_attendee = await create_attendee_record(attendee_data)\n if not created_attendee:\n return False\n attendee_id = created_attendee["id"]\n\n # Create pre-event response record\n pre_event_data = {\n "attendee": [attendee_id],\n "current_infrastructure_tools": answers.get("current_tools", []),\n "deployment_contexts": answers.get("deployment_contexts", []),\n "top_infrastructure_challenges": answers.get("challenges", []),\n "learning_expectations": answers.get("learning_expectations", []),\n "implementation_likelihood": answers.get("implementation_likelihood", ""),\n "specific_questions": answers.get("specific_questions", ""),\n "manual_tasks_time_weekly": answers.get("manual_time", ""),\n "infrastructure_budget_range": answers.get("budget", ""),\n "survey_source": "typebot"\n }\n\n await create_table_record(config.PRE_EVENT_TABLE_ID, pre_event_data)\n\n # Trigger follow-up actions\n await schedule_follow_up_actions(attendee_id, "registration_completed")\n\n logger.info(f"Successfully processed Typebot registration for {email}")\n return True\n\n except Exception as e:\n logger.error(f"Error processing Typebot registration: {e}")\n return False\n\nasync def process_formbricks_survey(data: FormbricksResponse) -> bool:\n """Process Formbricks survey data"""\n try:\n answers = data.answers\n\n # Determine if this is pre-event or post-event survey\n survey_type = "pre_event" if "rust_experience" in answers else "post_event"\n\n if survey_type == "pre_event":\n return await process_formbricks_pre_event(data)\n else:\n return await process_formbricks_post_event(data)\n\n except Exception as e:\n logger.error(f"Error processing Formbricks survey: {e}")\n return False\n\nasync def process_formbricks_pre_event(data: FormbricksResponse) -> bool:\n """Process Formbricks pre-event survey"""\n try:\n answers = data.answers\n email = answers.get("email_input", "")\n\n if not email:\n logger.warning("Missing email in Formbricks pre-event response")\n return False\n\n # Similar processing to Typebot but with Formbricks data structure\n existing_attendee = await find_attendee_by_email(email)\n\n if existing_attendee:\n await update_attendee_record(existing_attendee["id"], {\n "pre_survey_completed": True\n })\n attendee_id = existing_attendee["id"]\n else:\n # Create new attendee from Formbricks data\n attendee_data = {\n "email": email,\n "full_name": answers.get("name_input", ""),\n "professional_role": answers.get("role", ""),\n "experience_years": answers.get("experience_years", ""),\n "rust_experience_level": answers.get("rust_experience", ""),\n "company_size": answers.get("company_size", ""),\n "attendance_status": "registered",\n "pre_survey_completed": True\n }\n\n created_attendee = await create_attendee_record(attendee_data)\n if not created_attendee:\n return False\n attendee_id = created_attendee["id"]\n\n # Create pre-event response record\n pre_event_data = {\n "attendee": [attendee_id],\n "current_infrastructure_tools": answers.get("current_tools", []),\n "deployment_contexts": answers.get("deployment_contexts", []),\n "top_infrastructure_challenges": answers.get("infrastructure_challenges", []),\n "learning_expectations": answers.get("learning_goals", []),\n "implementation_likelihood": answers.get("implementation_likelihood", ""),\n "specific_questions": answers.get("specific_questions", ""),\n "manual_tasks_time_weekly": answers.get("manual_tasks_time", ""),\n "infrastructure_budget_range": answers.get("infrastructure_budget", ""),\n "survey_source": "formbricks"\n }\n\n await create_table_record(config.PRE_EVENT_TABLE_ID, pre_event_data)\n\n logger.info(f"Successfully processed Formbricks pre-event survey for {email}")\n return True\n\n except Exception as e:\n logger.error(f"Error processing Formbricks pre-event survey: {e}")\n return False\n\nasync def process_formbricks_post_event(data: FormbricksResponse) -> bool:\n """Process Formbricks post-event survey"""\n try:\n answers = data.answers\n\n # For post-event, we need to identify the attendee\n # This could be done via person_id or email lookup\n attendee = None\n\n if data.person_id:\n # Try to find attendee by person_id (if stored during pre-event)\n # Implementation depends on how you store the person_id\n pass\n\n # Fallback to email lookup if available in answers\n email = answers.get("email", "")\n if email:\n attendee = await find_attendee_by_email(email)\n\n if not attendee:\n logger.warning("Could not identify attendee for post-event survey")\n return False\n\n attendee_id = attendee["id"]\n\n # Update attendee record\n await update_attendee_record(attendee_id, {\n "post_survey_completed": True,\n "attendance_status": "attended"\n })\n\n # Create post-event response record\n post_event_data = {\n "attendee": [attendee_id],\n "overall_presentation_rating": answers.get("overall_rating", 0),\n "most_valuable_aspect": answers.get("most_valuable_aspect", ""),\n "concepts_learned": answers.get("learning_outcomes", []),\n "implementation_confidence_level": answers.get("implementation_confidence", ""),\n "evaluation_timeline": answers.get("implementation_timeline", ""),\n "immediate_next_steps": answers.get("next_steps", ""),\n "adoption_barriers": answers.get("adoption_barriers", []),\n "expected_business_benefits": answers.get("business_benefits", []),\n "nps_recommendation_score": answers.get("recommendation_likelihood", 0),\n "community_interests": answers.get("community_interest", []),\n "event_improvement_suggestions": answers.get("event_improvements", ""),\n "additional_feedback": answers.get("additional_feedback", ""),\n "survey_source": "formbricks"\n }\n\n await create_table_record(config.POST_EVENT_TABLE_ID, post_event_data)\n\n # Trigger follow-up actions based on responses\n await schedule_post_event_follow_ups(attendee_id, answers)\n\n logger.info(f"Successfully processed Formbricks post-event survey for attendee {attendee_id}")\n return True\n\n except Exception as e:\n logger.error(f"Error processing Formbricks post-event survey: {e}")\n return False\n\nasync def process_claper_engagement(data: ClaperEngagement) -> bool:\n """Process Claper live engagement data"""\n try:\n # Map participant_id to attendee (implementation depends on your setup)\n # This could be done via session tracking or email mapping\n attendee_id = await map_participant_to_attendee(data.participant_id)\n\n if not attendee_id:\n logger.warning(f"Could not map participant {data.participant_id} to attendee")\n return False\n\n # Calculate engagement metrics\n polls_participated = len(data.poll_responses)\n questions_asked = len(data.questions_submitted)\n engagement_score = calculate_engagement_score(\n polls_participated,\n questions_asked,\n data.session_duration\n )\n\n # Create live engagement record\n engagement_data = {\n "attendee": [attendee_id],\n "session_duration_minutes": data.session_duration,\n "polls_participated_count": polls_participated,\n "questions_submitted_count": questions_asked,\n "poll_responses_json": json.dumps(data.poll_responses),\n "questions_submitted_text": "\n".join(data.questions_submitted),\n "device_type": data.engagement_metrics.get("device_type", "desktop"),\n "interaction_quality_score": data.engagement_metrics.get("quality_score", 3)\n }\n\n await create_table_record(config.LIVE_ENGAGEMENT_TABLE_ID, engagement_data)\n\n # Update attendee record with engagement score\n await update_attendee_record(attendee_id, {\n "engagement_score": engagement_score,\n "attendance_status": "attended"\n })\n\n # Trigger high-engagement follow-up if score is high\n if engagement_score >= 8:\n await schedule_follow_up_actions(attendee_id, "high_engagement")\n\n logger.info(f"Successfully processed Claper engagement for attendee {attendee_id}")\n return True\n\n except Exception as e:\n logger.error(f"Error processing Claper engagement: {e}")\n return False\n\nasync def map_participant_to_attendee(participant_id: str) -> Optional[int]:\n """Map Claper participant ID to Baserow attendee ID"""\n # Implementation depends on how you track participants\n # Could be via session cookies, email lookup, or participant registration\n\n # Placeholder implementation - you'd implement actual mapping logic\n # For now, assume participant_id contains email or some identifier\n\n try:\n # Try to extract email from participant_id or use other mapping logic\n # This is a simplified example\n if "@" in participant_id:\n attendee = await find_attendee_by_email(participant_id)\n return attendee["id"] if attendee else None\n\n # Could also query a session mapping table\n # Or use participant registration data\n\n return None\n\n except Exception as e:\n logger.error(f"Error mapping participant {participant_id}: {e}")\n return None\n\n# Follow-up Action Functions\nasync def schedule_follow_up_actions(attendee_id: int, trigger_event: str):\n """Schedule appropriate follow-up actions based on trigger event"""\n try:\n actions = []\n\n if trigger_event == "registration_completed":\n actions = [\n {\n "attendee": [attendee_id],\n "action_type": "community_invite",\n "action_status": "pending",\n "priority_level": "medium",\n "scheduled_date": (datetime.utcnow() + timedelta(days=1)).date().isoformat(),\n "assigned_to": "automated_system",\n "action_details": "Send welcome email with community invite"\n }\n ]\n\n elif trigger_event == "high_engagement":\n actions = [\n {\n "attendee": [attendee_id],\n "action_type": "personal_followup",\n "action_status": "pending",\n "priority_level": "high",\n "scheduled_date": (datetime.utcnow() + timedelta(days=1)).date().isoformat(),\n "assigned_to": "organizer_1",\n "action_details": "Personal thank you for high engagement"\n }\n ]\n\n # Create follow-up action records\n for action in actions:\n await create_table_record(config.FOLLOW_UP_TABLE_ID, action)\n\n logger.info(f"Scheduled {len(actions)} follow-up actions for attendee {attendee_id}")\n\n except Exception as e:\n logger.error(f"Error scheduling follow-up actions: {e}")\n\nasync def schedule_post_event_follow_ups(attendee_id: int, survey_answers: Dict):\n """Schedule follow-up actions based on post-event survey responses"""\n try:\n actions = []\n\n # High NPS score - invite to be case study\n nps_score = survey_answers.get("recommendation_likelihood", 0)\n if nps_score >= 9:\n actions.append({\n "attendee": [attendee_id],\n "action_type": "personal_followup",\n "action_status": "pending",\n "priority_level": "medium",\n "scheduled_date": (datetime.utcnow() + timedelta(days=3)).date().isoformat(),\n "assigned_to": "community_manager",\n "action_details": "Invite to be case study or testimonial"\n })\n\n # Early implementation timeline - provide resources\n timeline = survey_answers.get("implementation_timeline", "")\n if timeline in ["this_week", "one_month"]:\n actions.append({\n "attendee": [attendee_id],\n "action_type": "resource_sharing",\n "action_status": "pending",\n "priority_level": "high",\n "scheduled_date": (datetime.utcnow() + timedelta(days=1)).date().isoformat(),\n "assigned_to": "technical_lead",\n "action_details": "Send implementation resources and offer consultation"\n })\n\n # Community interest - send invite\n community_interests = survey_answers.get("community_interest", [])\n if community_interests:\n actions.append({\n "attendee": [attendee_id],\n "action_type": "community_invite",\n "action_status": "pending",\n "priority_level": "medium",\n "scheduled_date": (datetime.utcnow() + timedelta(days=1)).date().isoformat(),\n "assigned_to": "community_manager",\n "action_details": f"Send community invites for: {', '.join(community_interests)}"\n })\n\n # Create follow-up action records\n for action in actions:\n await create_table_record(config.FOLLOW_UP_TABLE_ID, action)\n\n logger.info(f"Scheduled {len(actions)} post-event follow-up actions for attendee {attendee_id}")\n\n except Exception as e:\n logger.error(f"Error scheduling post-event follow-ups: {e}")\n\n# Notification Functions\nasync def send_discord_notification(message: str):\n """Send notification to Discord webhook"""\n if not config.DISCORD_WEBHOOK_URL:\n return\n\n try:\n async with httpx.AsyncClient() as client:\n await client.post(\n config.DISCORD_WEBHOOK_URL,\n json={"content": message}\n )\n except Exception as e:\n logger.error(f"Error sending Discord notification: {e}")\n\nasync def send_slack_notification(message: str):\n """Send notification to Slack webhook"""\n if not config.SLACK_WEBHOOK_URL:\n return\n\n try:\n async with httpx.AsyncClient() as client:\n await client.post(\n config.SLACK_WEBHOOK_URL,\n json={"text": message}\n )\n except Exception as e:\n logger.error(f"Error sending Slack notification: {e}")\n\n# Security dependency\nasync def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):\n """Verify webhook authentication token"""\n if credentials.credentials != config.WEBHOOK_SECRET:\n raise HTTPException(status_code=401, detail="Invalid authentication token")\n return credentials\n\n# API Endpoints\n@app.post("/webhooks/typebot/registration")\nasync def handle_typebot_registration(\n payload: TypebotResponse,\n background_tasks: BackgroundTasks,\n credentials: HTTPAuthorizationCredentials = Depends(verify_token)\n):\n """Handle Typebot registration webhook"""\n try:\n background_tasks.add_task(process_typebot_registration, payload)\n background_tasks.add_task(\n send_discord_notification,\n f"🦀 New registration from Typebot: {payload.answers.get('name_input', 'Unknown')}"\n )\n\n return {"status": "accepted", "message": "Registration webhook processed"}\n\n except Exception as e:\n logger.error(f"Error in Typebot registration webhook: {e}")\n raise HTTPException(status_code=500, detail="Internal server error")\n\n@app.post("/webhooks/formbricks/survey")\nasync def handle_formbricks_survey(\n payload: FormbricksResponse,\n background_tasks: BackgroundTasks,\n credentials: HTTPAuthorizationCredentials = Depends(verify_token)\n):\n """Handle Formbricks survey webhook"""\n try:\n background_tasks.add_task(process_formbricks_survey, payload)\n\n survey_type = "pre-event" if "rust_experience" in payload.answers else "post-event"\n background_tasks.add_task(\n send_discord_notification,\n f"📝 New {survey_type} survey response from Formbricks"\n )\n\n return {"status": "accepted", "message": "Survey webhook processed"}\n\n except Exception as e:\n logger.error(f"Error in Formbricks survey webhook: {e}")\n raise HTTPException(status_code=500, detail="Internal server error")\n\n@app.post("/webhooks/claper/engagement")\nasync def handle_claper_engagement(\n payload: ClaperEngagement,\n background_tasks: BackgroundTasks,\n credentials: HTTPAuthorizationCredentials = Depends(verify_token)\n):\n """Handle Claper engagement webhook"""\n try:\n background_tasks.add_task(process_claper_engagement, payload)\n\n engagement_score = calculate_engagement_score(\n len(payload.poll_responses),\n len(payload.questions_submitted),\n payload.session_duration\n )\n\n if engagement_score >= 8:\n background_tasks.add_task(\n send_discord_notification,\n f"🌟 High engagement detected! Score: {engagement_score}"\n )\n\n return {"status": "accepted", "message": "Engagement webhook processed"}\n\n except Exception as e:\n logger.error(f"Error in Claper engagement webhook: {e}")\n raise HTTPException(status_code=500, detail="Internal server error")\n\n@app.get("/health")\nasync def health_check():\n """Health check endpoint"""\n return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}\n\n@app.get("/metrics")\nasync def get_metrics():\n """Get system metrics"""\n try:\n async with await get_baserow_client() as client:\n # Get basic counts from each table\n attendees_response = await client.get(\n f"{config.BASEROW_API_URL}/database/rows/table/{config.ATTENDEES_TABLE_ID}/",\n params={"size": 1}\n )\n\n metrics = {\n "total_attendees": attendees_response.json().get("count", 0),\n "webhook_processor_status": "running",\n "last_update": datetime.utcnow().isoformat()\n }\n\n return metrics\n\n except Exception as e:\n logger.error(f"Error getting metrics: {e}")\n return {"error": "Could not fetch metrics"}\n\nif __name__ == "__main__":\n import uvicorn\n uvicorn.run(\n app,\n host="0.0.0.0",\n port=8000,\n log_level="info"\n )\n```\n\n---\n\n## 🔄 **Data Pipeline Scripts**\n\n### **data-pipeline.sh**\n\n```bash\n#!/bin/bash\n# data-pipeline.sh - ETL pipeline for event data processing\n\nset -e\n\n# Configuration\nNAMESPACE="rust-meetup"\nBACKUP_DIR="/tmp/event-data-$(date +%Y%m%d-%H%M%S)"\nLOG_FILE="/var/log/rust-meetup/data-pipeline.log"\n\n# Colors for output\nRED='\033[0;31m'\nGREEN='\033[0;32m'\nYELLOW='\033[1;33m'\nNC='\033[0m' # No Color\n\n# Logging function\nlog() {\n echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" | tee -a ${LOG_FILE}\n}\n\nlog_info() {\n echo -e "${GREEN}[INFO]${NC} $1"\n log "INFO: $1"\n}\n\nlog_warn() {\n echo -e "${YELLOW}[WARN]${NC} $1"\n log "WARN: $1"\n}\n\nlog_error() {\n echo -e "${RED}[ERROR]${NC} $1"\n log "ERROR: $1"\n}\n\n# Environment check\ncheck_environment() {\n log_info "Checking environment..."\n\n # Check if required tools are available\n for tool in kubectl curl jq python3; do\n if ! command -v $tool &> /dev/null; then\n log_error "$tool is not installed or not in PATH"\n exit 1\n fi\n done\n\n # Check if namespace exists\n if ! kubectl get namespace $NAMESPACE &> /dev/null; then\n log_error "Namespace $NAMESPACE does not exist"\n exit 1\n fi\n\n log_info "Environment check passed"\n}\n\n# Data extraction functions\nextract_typebot_data() {\n log_info "Extracting Typebot data..."\n\n # Get Typebot pod\n TYPEBOT_POD=$(kubectl get pods -n $NAMESPACE -l app=typebot -o jsonpath='{.items[0].metadata.name}')\n\n if [ -z "$TYPEBOT_POD" ]; then\n log_error "No Typebot pod found"\n return 1\n fi\n\n # Extract data via API or database dump\n kubectl exec -n $NAMESPACE $TYPEBOT_POD -- \n curl -s "http://localhost:3000/api/typebots/export" \n -H "Authorization: Bearer $TYPEBOT_API_TOKEN" > $BACKUP_DIR/typebot-data.json\n\n log_info "Typebot data extracted to $BACKUP_DIR/typebot-data.json"\n}\n\nextract_formbricks_data() {\n log_info "Extracting Formbricks data..."\n\n FORMBRICKS_POD=$(kubectl get pods -n $NAMESPACE -l app=formbricks -o jsonpath='{.items[0].metadata.name}')\n\n if [ -z "$FORMBRICKS_POD" ]; then\n log_error "No Formbricks pod found"\n return 1\n fi\n\n # Extract survey responses\n kubectl exec -n $NAMESPACE $FORMBRICKS_POD -- \n curl -s "http://localhost:3000/api/v1/responses" \n -H "Authorization: Bearer $FORMBRICKS_API_TOKEN" > $BACKUP_DIR/formbricks-data.json\n\n log_info "Formbricks data extracted to $BACKUP_DIR/formbricks-data.json"\n}\n\nextract_claper_data() {\n log_info "Extracting Claper data..."\n\n CLAPER_POD=$(kubectl get pods -n $NAMESPACE -l app=claper -o jsonpath='{.items[0].metadata.name}')\n\n if [ -z "$CLAPER_POD" ]; then\n log_error "No Claper pod found"\n return 1\n fi\n\n # Extract engagement data\n kubectl exec -n $NAMESPACE $CLAPER_POD -- \n curl -s "http://localhost:4000/api/presentations/export" \n -H "Authorization: Bearer $CLAPER_API_TOKEN" > $BACKUP_DIR/claper-data.json\n\n log_info "Claper data extracted to $BACKUP_DIR/claper-data.json"\n}\n\nextract_baserow_data() {\n log_info "Extracting Baserow data..."\n\n # Extract data from all tables\n TABLES=("1" "2" "3" "4" "5" "6") # Table IDs\n TABLE_NAMES=("attendees" "pre_event" "live_engagement" "post_event" "follow_up" "analytics")\n\n for i in "${!TABLES[@]}"; do\n TABLE_ID=${TABLES[$i]}\n TABLE_NAME=${TABLE_NAMES[$i]}\n\n curl -s "https://baserow.rustmeetup.com/api/database/rows/table/$TABLE_ID/" \n -H "Authorization: Token $BASEROW_API_TOKEN" \n > $BACKUP_DIR/baserow-$TABLE_NAME.json\n\n log_info "Exported Baserow table $TABLE_NAME"\n done\n}\n\n# Data transformation functions\ntransform_data() {\n log_info "Transforming extracted data..."\n\n # Create transformation directory\n mkdir -p $BACKUP_DIR/transformed\n\n # Run Python transformation script\n python3 << 'EOF'\nimport json\nimport os\nimport sys\nfrom datetime import datetime\n\nbackup_dir = os.environ.get('BACKUP_DIR')\nif not backup_dir:\n print("BACKUP_DIR environment variable not set")\n sys.exit(1)\n\ndef load_json_file(filename):\n try:\n with open(f"{backup_dir}/{filename}", 'r') as f:\n return json.load(f)\n except FileNotFoundError:\n print(f"File {filename} not found, skipping...")\n return {}\n except json.JSONDecodeError:\n print(f"Invalid JSON in {filename}, skipping...")\n return {}\n\ndef save_json_file(data, filename):\n with open(f"{backup_dir}/transformed/{filename}", 'w') as f:\n json.dump(data, f, indent=2, default=str)\n\n# Load all data sources\ntypebot_data = load_json_file('typebot-data.json')\nformbricks_data = load_json_file('formbricks-data.json')\nclaper_data = load_json_file('claper-data.json')\n\n# Transform to unified format\nunified_data = {\n "registrations": [],\n "surveys": [],\n "engagement": [],\n "metadata": {\n "extraction_timestamp": datetime.utcnow().isoformat(),\n "sources": ["typebot", "formbricks", "claper", "baserow"]\n }\n}\n\n# Process Typebot registrations\nif typebot_data and 'results' in typebot_data:\n for result in typebot_data['results']:\n registration = {\n "source": "typebot",\n "timestamp": result.get('createdAt'),\n "answers": result.get('answers', {}),\n "email": result.get('answers', {}).get('email_input'),\n "name": result.get('answers', {}).get('name_input')\n }\n unified_data["registrations"].append(registration)\n\n# Process Formbricks surveys\nif formbricks_data and 'data' in formbricks_data:\n for response in formbricks_data['data']:\n survey = {\n "source": "formbricks",\n "timestamp": response.get('createdAt'),\n "survey_id": response.get('surveyId'),\n "answers": response.get('data', {}),\n "person_id": response.get('personId')\n }\n unified_data["surveys"].append(survey)\n\n# Process Claper engagement\nif claper_data and 'sessions' in claper_data:\n for session in claper_data['sessions']:\n engagement = {\n "source": "claper",\n "timestamp": session.get('created_at'),\n "session_id": session.get('id'),\n "participant_count": session.get('participant_count', 0),\n "polls": session.get('polls', []),\n "questions": session.get('questions', [])\n }\n unified_data["engagement"].append(engagement)\n\n# Save unified data\nsave_json_file(unified_data, 'unified_data.json')\n\n# Generate summary statistics\nstats = {\n "total_registrations": len(unified_data["registrations"]),\n "total_surveys": len(unified_data["surveys"]),\n "total_engagement_sessions": len(unified_data["engagement"]),\n "extraction_timestamp": unified_data["metadata"]["extraction_timestamp"]\n}\n\nsave_json_file(stats, 'summary_stats.json')\n\nprint(f"Transformation complete. Processed {stats['total_registrations']} registrations, {stats['total_surveys']} surveys, {stats['total_engagement_sessions']} engagement sessions.")\nEOF\n\n if [ $? -eq 0 ]; then\n log_info "Data transformation completed successfully"\n else\n log_error "Data transformation failed"\n return 1\n fi\n}\n\n# Data validation functions\nvalidate_data() {\n log_info "Validating transformed data..."\n\n python3 << 'EOF'\nimport json\nimport os\nimport sys\nfrom jsonschema import validate, ValidationError\n\nbackup_dir = os.environ.get('BACKUP_DIR')\n\n# Define validation schema\nschema = {\n "type": "object",\n "properties": {\n "registrations": {\n "type": "array",\n "items": {\n "type": "object",\n "required": ["source", "timestamp"],\n "properties": {\n "source": {"type": "string"},\n "timestamp": {"type": "string"},\n "email": {"type": ["string", "null"]},\n "name": {"type": ["string", "null"]}\n }\n }\n },\n "surveys": {\n "type": "array",\n "items": {\n "type": "object",\n "required": ["source", "timestamp"],\n "properties": {\n "source": {"type": "string"},\n "timestamp": {"type": "string"}\n }\n }\n },\n "engagement": {\n "type": "array",\n "items": {\n "type": "object",\n "required": ["source", "timestamp"],\n "properties": {\n "source": {"type": "string"},\n "timestamp": {"type": "string"}\n }\n }\n }\n },\n "required": ["registrations", "surveys", "engagement"]\n}\n\ntry:\n with open(f"{backup_dir}/transformed/unified_data.json", 'r') as f:\n data = json.load(f)\n\n validate(instance=data, schema=schema)\n print("Data validation passed")\n\n # Additional checks\n total_records = len(data["registrations"]) + len(data["surveys"]) + len(data["engagement"])\n if total_records == 0:\n print("WARNING: No data records found")\n sys.exit(1)\n\n print(f"Validated {total_records} total records")\n\nexcept ValidationError as e:\n print(f"Data validation failed: {e}")\n sys.exit(1)\nexcept Exception as e:\n print(f"Validation error: {e}")\n sys.exit(1)\nEOF\n\n if [ $? -eq 0 ]; then\n log_info "Data validation passed"\n else\n log_error "Data validation failed"\n return 1\n fi\n}\n\n# Data loading functions\nload_to_analytics() {\n log_info "Loading data to analytics system..."\n\n # Send to analytics endpoint\n curl -X POST "https://analytics.rustmeetup.com/api/data-import" \n -H "Content-Type: application/json" \n -H "Authorization: Bearer $ANALYTICS_API_TOKEN" \n -d @$BACKUP_DIR/transformed/unified_data.json\n\n if [ $? -eq 0 ]; then\n log_info "Data loaded to analytics system successfully"\n else\n log_error "Failed to load data to analytics system"\n return 1\n fi\n}\n\nbackup_to_s3() {\n log_info "Backing up data to S3..."\n\n # Create archive\n tar -czf $BACKUP_DIR.tar.gz -C $(dirname $BACKUP_DIR) $(basename $BACKUP_DIR)\n\n # Upload to S3\n aws s3 cp $BACKUP_DIR.tar.gz s3://rust-meetup-backups/data-pipeline/\n\n if [ $? -eq 0 ]; then\n log_info "Data backed up to S3 successfully"\n else\n log_error "Failed to backup data to S3"\n return 1\n fi\n}\n\n# Notification functions\nsend_success_notification() {\n local stats_file="$BACKUP_DIR/transformed/summary_stats.json"\n\n if [ -f "$stats_file" ]; then\n local stats=$(cat $stats_file | jq -r '. | "Registrations: \(.total_registrations), Surveys: \(.total_surveys), Engagement: \(.total_engagement_sessions)"')\n local message="✅ Data pipeline completed successfully! $stats"\n else\n local message="✅ Data pipeline completed successfully!"\n fi\n\n # Send to Discord\n if [ ! -z "$DISCORD_WEBHOOK_URL" ]; then\n curl -X POST "$DISCORD_WEBHOOK_URL" \n -H "Content-Type: application/json" \n -d "{\"content\": \"$message\"}"\n fi\n\n # Send to Slack\n if [ ! -z "$SLACK_WEBHOOK_URL" ]; then\n curl -X POST "$SLACK_WEBHOOK_URL" \n -H "Content-Type: application/json" \n -d "{\"text\": \"$message\"}"\n fi\n\n log_info "Success notification sent"\n}\n\nsend_failure_notification() {\n local error_message="$1"\n local message="❌ Data pipeline failed: $error_message"\n\n # Send to Discord\n if [ ! -z "$DISCORD_WEBHOOK_URL" ]; then\n curl -X POST "$DISCORD_WEBHOOK_URL" \n -H "Content-Type: application/json" \n -d "{\"content\": \"$message\"}"\n fi\n\n # Send to Slack\n if [ ! -z "$SLACK_WEBHOOK_URL" ]; then\n curl -X POST "$SLACK_WEBHOOK_URL" \n -H "Content-Type: application/json" \n -d "{\"text\": \"$message\"}"\n fi\n\n log_error "Failure notification sent: $error_message"\n}\n\n# Cleanup function\ncleanup() {\n log_info "Cleaning up temporary files..."\n\n # Keep backups for 7 days\n find /tmp -name "event-data-*" -type d -mtime +7 -exec rm -rf {} \;\n\n # Clean up current backup directory\n if [ -d "$BACKUP_DIR" ]; then\n rm -rf "$BACKUP_DIR"\n fi\n\n log_info "Cleanup completed"\n}\n\n# Main execution function\nmain() {\n log_info "Starting data pipeline execution..."\n\n # Create backup directory\n mkdir -p $BACKUP_DIR\n export BACKUP_DIR # Make available to subprocesses\n\n # Set error handling\n trap 'send_failure_notification "Pipeline interrupted"; cleanup; exit 1' INT TERM\n\n # Execute pipeline steps\n check_environment || {\n send_failure_notification "Environment check failed"\n cleanup\n exit 1\n }\n\n # Data extraction\n extract_typebot_data || log_warn "Typebot extraction failed, continuing..."\n extract_formbricks_data || log_warn "Formbricks extraction failed, continuing..."\n extract_claper_data || log_warn "Claper extraction failed, continuing..."\n extract_baserow_data || {\n send_failure_notification "Baserow extraction failed"\n cleanup\n exit 1\n }\n\n # Data transformation and validation\n transform_data || {\n send_failure_notification "Data transformation failed"\n cleanup\n exit 1\n }\n\n validate_data || {\n send_failure_notification "Data validation failed"\n cleanup\n exit 1\n }\n\n # Data loading\n load_to_analytics || log_warn "Analytics loading failed, continuing..."\n backup_to_s3 || log_warn "S3 backup failed, continuing..."\n\n # Success notification\n send_success_notification\n\n # Cleanup\n cleanup\n\n log_info "Data pipeline completed successfully!"\n}\n\n# Script execution\nif [ "${BASH_SOURCE[0]}" == "${0}" ]; then\n main "$@"\nfi\n```\n\n---\n\n## 🚀 **Deployment Configuration**\n\n### **docker-compose.yml** (for local development)\n\n```yaml\nversion: '3.8'\nservices:\n webhook-processor:\n build: .\n ports:\n - "8000:8000"\n environment:\n - BASEROW_API_URL=https://baserow.rustmeetup.com/api\n - BASEROW_API_TOKEN=${BASEROW_API_TOKEN}\n - WEBHOOK_SECRET=${WEBHOOK_SECRET}\n - ATTENDEES_TABLE_ID=1\n - PRE_EVENT_TABLE_ID=2\n - LIVE_ENGAGEMENT_TABLE_ID=3\n - POST_EVENT_TABLE_ID=4\n - FOLLOW_UP_TABLE_ID=5\n - DISCORD_WEBHOOK_URL=${DISCORD_WEBHOOK_URL}\n - SLACK_WEBHOOK_URL=${SLACK_WEBHOOK_URL}\n depends_on:\n - redis\n volumes:\n - ./logs:/var/log/rust-meetup\n restart: unless-stopped\n\n redis:\n image: redis:7-alpine\n ports:\n - "6379:6379"\n command: redis-server --requirepass ${REDIS_PASSWORD}\n volumes:\n - redis-data:/data\n restart: unless-stopped\n\n data-pipeline:\n build: .\n environment:\n - TYPEBOT_API_TOKEN=${TYPEBOT_API_TOKEN}\n - FORMBRICKS_API_TOKEN=${FORMBRICKS_API_TOKEN}\n - CLAPER_API_TOKEN=${CLAPER_API_TOKEN}\n - BASEROW_API_TOKEN=${BASEROW_API_TOKEN}\n - ANALYTICS_API_TOKEN=${ANALYTICS_API_TOKEN}\n - AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}\n - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}\n - DISCORD_WEBHOOK_URL=${DISCORD_WEBHOOK_URL}\n - SLACK_WEBHOOK_URL=${SLACK_WEBHOOK_URL}\n volumes:\n - ./scripts:/scripts\n - ./logs:/var/log/rust-meetup\n command: /scripts/data-pipeline.sh\n profiles:\n - pipeline\n\nvolumes:\n redis-data:\n```\n\n### **Kubernetes Deployment**\n\n```yaml\n# webhook-processor-deployment.yaml\napiVersion: apps/v1\nkind: Deployment\nmetadata:\n name: webhook-processor\n namespace: rust-meetup\nspec:\n replicas: 2\n selector:\n matchLabels:\n app: webhook-processor\n template:\n metadata:\n labels:\n app: webhook-processor\n spec:\n containers:\n - name: webhook-processor\n image: rustmeetup/webhook-processor:latest\n ports:\n - containerPort: 8000\n env:\n - name: BASEROW_API_URL\n value: "https://baserow.rustmeetup.com/api"\n - name: BASEROW_API_TOKEN\n valueFrom:\n secretKeyRef:\n name: api-secrets\n key: baserow-token\n - name: WEBHOOK_SECRET\n valueFrom:\n secretKeyRef:\n name: api-secrets\n key: webhook-secret\n resources:\n requests:\n memory: "256Mi"\n cpu: "250m"\n limits:\n memory: "512Mi"\n cpu: "500m"\n livenessProbe:\n httpGet:\n path: /health\n port: 8000\n initialDelaySeconds: 30\n periodSeconds: 10\n readinessProbe:\n httpGet:\n path: /health\n port: 8000\n initialDelaySeconds: 5\n periodSeconds: 5\n---\napiVersion: v1\nkind: Service\nmetadata:\n name: webhook-processor-service\n namespace: rust-meetup\nspec:\n selector:\n app: webhook-processor\n ports:\n - port: 80\n targetPort: 8000\n type: ClusterIP\n---\n# CronJob for data pipeline\napiVersion: batch/v1\nkind: CronJob\nmetadata:\n name: data-pipeline\n namespace: rust-meetup\nspec:\n schedule: "0 2 * * *" # Daily at 2 AM\n jobTemplate:\n spec:\n template:\n spec:\n containers:\n - name: data-pipeline\n image: rustmeetup/data-pipeline:latest\n command: ["/scripts/data-pipeline.sh"]\n env:\n - name: NAMESPACE\n value: "rust-meetup"\n - name: BASEROW_API_TOKEN\n valueFrom:\n secretKeyRef:\n name: api-secrets\n key: baserow-token\n volumeMounts:\n - name: scripts\n mountPath: /scripts\n - name: logs\n mountPath: /var/log/rust-meetup\n volumes:\n - name: scripts\n configMap:\n name: data-pipeline-scripts\n defaultMode: 0755\n - name: logs\n persistentVolumeClaim:\n claimName: logs-pvc\n restartPolicy: OnFailure\n```\n\n---\n\n## 📊 **Monitoring and Alerting**\n\n### **monitoring-config.yaml**\n\n```yaml\napiVersion: v1\nkind: ConfigMap\nmetadata:\n name: webhook-processor-monitoring\n namespace: rust-meetup\ndata:\n prometheus.yml: |\n global:\n scrape_interval: 30s\n\n scrape_configs:\n - job_name: 'webhook-processor'\n static_configs:\n - targets: ['webhook-processor-service:80']\n metrics_path: '/metrics'\n scrape_interval: 30s\n\n - job_name: 'baserow'\n static_configs:\n - targets: ['baserow-service:80']\n metrics_path: '/_health/'\n scrape_interval: 60s\n\n alerts.yml: |\n groups:\n - name: webhook-processor\n rules:\n - alert: WebhookProcessorDown\n expr: up{job="webhook-processor"} == 0\n for: 5m\n labels:\n severity: critical\n annotations:\n summary: "Webhook processor is down"\n description: "Webhook processor has been down for more than 5 minutes"\n\n - alert: HighWebhookErrorRate\n expr: rate(webhook_errors_total[5m]) > 0.1\n for: 2m\n labels:\n severity: warning\n annotations:\n summary: "High webhook error rate"\n description: "Webhook error rate is {{ $value }} per second"\n\n - alert: BaserowConnectionFailed\n expr: baserow_connection_failures > 5\n for: 1m\n labels:\n severity: critical\n annotations:\n summary: "Baserow connection failures"\n description: "Multiple connection failures to Baserow detected"\n```\n\n---\n\n## 🔍 **Testing Scripts**\n\n### **test-integration.py**\n\n```python\n#!/usr/bin/env python3\n"""\nIntegration tests for webhook processor\n"""\n\nimport asyncio\nimport httpx\nimport json\nimport pytest\nfrom datetime import datetime\n\nBASE_URL = "http://localhost:8000"\nTEST_TOKEN = "test-webhook-secret"\n\nclass TestWebhookProcessor:\n\n @pytest.fixture\n async def client(self):\n async with httpx.AsyncClient(base_url=BASE_URL) as client:\n yield client\n\n async def test_health_endpoint(self, client):\n """Test health check endpoint"""\n response = await client.get("/health")\n assert response.status_code == 200\n assert response.json()["status"] == "healthy"\n\n async def test_typebot_webhook(self, client):\n """Test Typebot registration webhook"""\n payload = {\n "typebot_id": "test-typebot",\n "result_id": "test-result",\n "answers": {\n "email_input": "test@example.com",\n "name_input": "Test User",\n "role": "developer",\n "rust_experience": "intermediate"\n },\n "created_at": datetime.utcnow().isoformat()\n }\n\n response = await client.post(\n "/webhooks/typebot/registration",\n json=payload,\n headers={"Authorization": f"Bearer {TEST_TOKEN}"}\n )\n\n assert response.status_code == 200\n assert response.json()["status"] == "accepted"\n\n async def test_formbricks_webhook(self, client):\n """Test Formbricks survey webhook"""\n payload = {\n "survey_id": "test-survey",\n "response_id": "test-response",\n "person_id": "test-person",\n "answers": {\n "overall_rating": 5,\n "most_valuable_aspect": "technical_depth",\n "nps_recommendation_score": 9\n },\n "created_at": datetime.utcnow().isoformat()\n }\n\n response = await client.post(\n "/webhooks/formbricks/survey",\n json=payload,\n headers={"Authorization": f"Bearer {TEST_TOKEN}"}\n )\n\n assert response.status_code == 200\n assert response.json()["status"] == "accepted"\n\n async def test_claper_webhook(self, client):\n """Test Claper engagement webhook"""\n payload = {\n "session_id": "test-session",\n "participant_id": "test@example.com",\n "poll_responses": [\n {"poll_id": "welcome", "answer": "home"},\n {"poll_id": "rust_level", "answer": "intermediate"}\n ],\n "questions_submitted": [\n "How does this compare to Terraform?",\n "What about migration strategies?"\n ],\n "session_duration": 75,\n "engagement_metrics": {\n "device_type": "desktop",\n "quality_score": 4\n }\n }\n\n response = await client.post(\n "/webhooks/claper/engagement",\n json=payload,\n headers={"Authorization": f"Bearer {TEST_TOKEN}"}\n )\n\n assert response.status_code == 200\n assert response.json()["status"] == "accepted"\n\n async def test_unauthorized_request(self, client):\n """Test request without proper authentication"""\n payload = {"test": "data"}\n\n response = await client.post(\n "/webhooks/typebot/registration",\n json=payload\n )\n\n assert response.status_code == 401\n\nif __name__ == "__main__":\n pytest.main([__file__])\n```\n\n### **load-test.sh**\n\n```bash\n#!/bin/bash\n# load-test.sh - Load testing for webhook processor\n\n# Configuration\nWEBHOOK_URL="http://localhost:8000"\nCONCURRENT_USERS=10\nDURATION=60\nTOKEN="test-webhook-secret"\n\necho "Starting load test for $DURATION seconds with $CONCURRENT_USERS concurrent users..."\n\n# Create test payload\ncat > test_payload.json << EOF\n{\n "typebot_id": "load-test",\n "result_id": "test-\$(date +%s)",\n "answers": {\n "email_input": "loadtest\$(shuf -i 1-1000 -n 1)@example.com",\n "name_input": "Load Test User",\n "role": "developer"\n },\n "created_at": "\$(date -u +%Y-%m-%dT%H:%M:%S.%3NZ)"\n}\nEOF\n\n# Run load test with Apache Bench\nab -n 1000 -c $CONCURRENT_USERS -T application/json -H "Authorization: Bearer $TOKEN" \n -p test_payload.json "$WEBHOOK_URL/webhooks/typebot/registration"\n\n# Cleanup\nrm test_payload.json\n\necho "Load test completed!"\n```\n\n---\n\n## 📝 **Usage Instructions**\n\n### **1. Deployment**\n\n```bash\n# Deploy webhook processor\nkubectl apply -f webhook-processor-deployment.yaml\n\n# Create secrets\nkubectl create secret generic api-secrets \n --from-literal=baserow-token="your-baserow-token" \n --from-literal=webhook-secret="your-webhook-secret" \n -n rust-meetup\n\n# Deploy monitoring\nkubectl apply -f monitoring-config.yaml\n```\n\n### **2. Configuration**\n\n```bash\n# Set environment variables\nexport BASEROW_API_TOKEN="your-token"\nexport WEBHOOK_SECRET="your-secret"\nexport DISCORD_WEBHOOK_URL="your-discord-webhook"\n\n# Test configuration\ncurl -X GET "http://webhook-processor-service/health" \n -H "Authorization: Bearer $WEBHOOK_SECRET"\n```\n\n### **3. Testing**\n\n```bash\n# Run integration tests\npython3 test-integration.py\n\n# Run load tests\n./load-test.sh\n\n# Test data pipeline\n./data-pipeline.sh\n```\n\nThis comprehensive automation system provides seamless integration between all components of your event management stack, ensuring reliable data flow and automated follow-up actions for the Rust Meetup 2025. |