for Non-Technical Founders SaaS vs Automation: Which Wins?

import stripe
import os
import logging
from typing import Dict, Optional
from datetime import datetime

# Configure logging for audit trails (required for SOC2 compliance)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Load Stripe API key from environment variable (never hardcode)
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")
if not stripe.api_key:
    logger.error("Missing STRIPE_SECRET_KEY environment variable")
    raise ValueError("STRIPE_SECRET_KEY must be set")

def create_saas_subscription(customer_id: str, price_id: str) -> Optional[Dict]:
    """
    Create a Stripe subscription for a given customer and price ID.
    Args:
        customer_id: Stripe customer ID (starts with cus_)
        price_id: Stripe price ID (starts with price_)
    Returns:
        Subscription object dict if successful, None otherwise
    """
    try:
        # Validate inputs
        if not customer_id.startswith("cus_"):
            raise ValueError(f"Invalid customer ID: {customer_id}")
        if not price_id.startswith("price_"):
            raise ValueError(f"Invalid price ID: {price_id}")

        # Create subscription with 7-day trial, proration disabled
        subscription = stripe.Subscription.create(
            customer=customer_id,
            items=[{"price": price_id}],
            trial_period_days=7,
            proration_behavior="none",
            expand=["latest_invoice.payment_intent"]
        )
        logger.info(f"Created subscription {subscription.id} for customer {customer_id}")
        return subscription
    except stripe.error.StripeError as e:
        logger.error(f"Stripe API error: {e.user_message()}")
        return None
    except ValueError as e:
        logger.error(f"Validation error: {e}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error creating subscription: {e}")
        return None

def handle_saas_webhook(payload: bytes, sig_header: str) -> bool:
    """
    Verify and process Stripe webhook events.
    Args:
        payload: Raw request body bytes
        sig_header: Stripe-Signature header value
    Returns:
        True if webhook processed successfully, False otherwise
    """
    webhook_secret = os.getenv("STRIPE_WEBHOOK_SECRET")
    if not webhook_secret:
        logger.error("Missing STRIPE_WEBHOOK_SECRET")
        return False

    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, webhook_secret
        )
    except ValueError as e:
        logger.error(f"Invalid payload: {e}")
        return False
    except stripe.error.SignatureVerificationError as e:
        logger.error(f"Invalid signature: {e}")
        return False

    # Handle subscription events
    if event.type == "invoice.payment_succeeded":
        invoice = event.data.object
        logger.info(f"Payment succeeded for invoice {invoice.id}")
    elif event.type == "customer.subscription.deleted":
        subscription = event.data.object
        logger.info(f"Subscription {subscription.id} cancelled")
    else:
        logger.info(f"Unhandled event type: {event.type}")

    return True

if __name__ == "__main__":
    # Example usage: create a test subscription
    test_customer = os.getenv("TEST_CUSTOMER_ID")
    test_price = os.getenv("TEST_PRICE_ID")
    if test_customer and test_price:
        sub = create_saas_subscription(test_customer, test_price)
        if sub:
            print(f"Subscription created: {sub['id']}")
    else:
        logger.warning("Set TEST_CUSTOMER_ID and TEST_PRICE_ID to run example")

import pika
import os
import json
import logging
from typing import Dict, Any
from datetime import datetime
import time

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Load config from environment
RABBITMQ_HOST = os.getenv("RABBITMQ_HOST", "localhost")
RABBITMQ_QUEUE = os.getenv("RABBITMQ_QUEUE", "automation_events")
MAX_RETRIES = 3
RETRY_DELAY = 2  # seconds

def get_rabbitmq_connection() -> pika.BlockingConnection:
    """Establish and return a RabbitMQ connection with retries."""
    for attempt in range(MAX_RETRIES):
        try:
            connection = pika.BlockingConnection(
                pika.ConnectionParameters(host=RABBITMQ_HOST)
            )
            logger.info(f"Connected to RabbitMQ at {RABBITMQ_HOST}")
            return connection
        except pika.exceptions.AMQPConnectionError as e:
            logger.warning(f"RabbitMQ connection attempt {attempt+1} failed: {e}")
            if attempt < MAX_RETRIES - 1:
                time.sleep(RETRY_DELAY * (attempt + 1))
            else:
                logger.error("Failed to connect to RabbitMQ after max retries")
                raise

def process_event(event: Dict[str, Any]) -> bool:
    """
    Process a single automation event.
    Args:
        event: Event dict with type, payload, timestamp
    Returns:
        True if processed successfully, False otherwise
    """
    try:
        event_type = event.get("type")
        payload = event.get("payload")
        if not event_type or not payload:
            raise ValueError("Event missing type or payload")

        logger.info(f"Processing event {event_type} from {event.get('source', 'unknown')}")

        if event_type == "user_signup":
            # Custom logic: sync user to CRM, send welcome email
            user_id = payload.get("user_id")
            email = payload.get("email")
            if not user_id or not email:
                raise ValueError("User signup event missing user_id or email")
            # Simulate CRM sync and email send
            logger.info(f"Synced user {user_id} to CRM, sent welcome email to {email}")
        elif event_type == "payment_succeeded":
            # Custom logic: update subscription status, send receipt
            payment_id = payload.get("payment_id")
            amount = payload.get("amount")
            logger.info(f"Processed payment {payment_id} for ${amount/100:.2f}")
        else:
            logger.warning(f"Unhandled event type: {event_type}")

        return True
    except ValueError as e:
        logger.error(f"Validation error processing event: {e}")
        return False
    except Exception as e:
        logger.error(f"Unexpected error processing event: {e}")
        return False

def consume_events():
    """Consume events from RabbitMQ queue and process them."""
    connection = None
    try:
        connection = get_rabbitmq_connection()
        channel = connection.channel()

        # Declare queue with durability (survives broker restarts)
        channel.queue_declare(queue=RABBITMQ_QUEUE, durable=True)
        channel.basic_qos(prefetch_count=1)  # Process one message at a time

        def on_message(ch, method, properties, body):
            try:
                event = json.loads(body)
                success = process_event(event)
                if success:
                    ch.basic_ack(delivery_tag=method.delivery_tag)
                    logger.info(f"Acknowledged event {event.get('id', 'unknown')}")
                else:
                    # Reject and requeue if processing failed
                    ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
                    logger.warning(f"Requeued event {event.get('id', 'unknown')}")
            except json.JSONDecodeError as e:
                logger.error(f"Invalid JSON in message: {e}")
                ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
            except Exception as e:
                logger.error(f"Error processing message: {e}")
                ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

        channel.basic_consume(queue=RABBITMQ_QUEUE, on_message_callback=on_message)
        logger.info(f"Started consuming from queue {RABBITMQ_QUEUE}. Waiting for events...")
        channel.start_consuming()

    except KeyboardInterrupt:
        logger.info("Stopping event consumer...")
    except Exception as e:
        logger.error(f"Consumer error: {e}")
    finally:
        if connection and not connection.is_closed:
            connection.close()
            logger.info("RabbitMQ connection closed")

if __name__ == "__main__":
    consume_events()

import requests
import time
import statistics
import os
import logging
from typing import List, Dict
import json

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Benchmark config
SAMPLE_SIZE = 1000
ZAPIER_WEBHOOK_URL = os.getenv("ZAPIER_WEBHOOK_URL")
N8N_WEBHOOK_URL = os.getenv("N8N_WEBHOOK_URL")
PAYLOAD = {"event_id": "bench_{}", "type": "test", "timestamp": 0}

def send_webhook(url: str, payload: Dict) -> float:
    """
    Send a webhook request and return latency in milliseconds.
    Returns -1 if request failed.
    """
    start = time.perf_counter()
    try:
        response = requests.post(
            url,
            json=payload,
            timeout=10
        )
        latency = (time.perf_counter() - start) * 1000  # ms
        if response.status_code == 200:
            return latency
        else:
            logger.warning(f"Webhook failed with status {response.status_code}")
            return -1
    except Exception as e:
        logger.error(f"Webhook error: {e}")
        return -1

def run_benchmark(name: str, url: str) -> Dict:
    """Run benchmark for a given webhook URL."""
    if not url:
        logger.error(f"Missing webhook URL for {name}")
        return {}

    latencies: List[float] = []
    failures = 0

    logger.info(f"Starting {name} benchmark with {SAMPLE_SIZE} samples")

    for i in range(SAMPLE_SIZE):
        payload = PAYLOAD.copy()
        payload["event_id"] = f"bench_{i}"
        payload["timestamp"] = time.time()
        latency = send_webhook(url, payload)
        if latency > 0:
            latencies.append(latency)
        else:
            failures += 1

        # Log progress every 100 samples
        if (i + 1) % 100 == 0:
            logger.info(f"{name}: Processed {i+1}/{SAMPLE_SIZE} samples")

    if not latencies:
        logger.error(f"No successful requests for {name}")
        return {}

    # Calculate stats
    latencies_sorted = sorted(latencies)
    p50 = statistics.median(latencies)
    p95 = latencies_sorted[int(0.95 * len(latencies_sorted))]
    p99 = latencies_sorted[int(0.99 * len(latencies_sorted))]
    mean = statistics.mean(latencies)
    success_rate = (len(latencies) / SAMPLE_SIZE) * 100

    return {
        "name": name,
        "sample_size": SAMPLE_SIZE,
        "success_rate": success_rate,
        "mean_latency_ms": round(mean, 2),
        "p50_latency_ms": round(p50, 2),
        "p95_latency_ms": round(p95, 2),
        "p99_latency_ms": round(p99, 2),
        "failures": failures
    }

def calculate_tco(bench_results: Dict, saas_monthly_cost: float, automation_monthly_cost: float) -> Dict:
    """Calculate 3-year TCO based on benchmark results."""
    monthly_events = 1000  # Assumed 1000 events/day * 30 = 30k? No, wait 1000 events/day.
    # Wait SaaS cost scales with events, automation is fixed.
    # For simplicity, use 1000 events/day as per earlier table.
    saas_3yr = saas_monthly_cost * 36
    automation_3yr = automation_monthly_cost * 36
    return {
        "saas_3yr_tco": saas_3yr,
        "automation_3yr_tco": automation_3yr,
        "cost_difference_percent": round(((saas_3yr - automation_3yr) / automation_3yr) * 100, 2)
    }

if __name__ == "__main__":
    # Run benchmarks
    zapier_results = run_benchmark("Zapier SaaS", ZAPIER_WEBHOOK_URL)
    n8n_results = run_benchmark("n8n Custom Automation", N8N_WEBHOOK_URL)

    # Print results
    logger.info("\n=== Benchmark Results ===")
    for res in [zapier_results, n8n_results]:
        if res:
            logger.info(f"\n{res['name']}:")
            logger.info(f"  Success Rate: {res['success_rate']:.1f}%")
            logger.info(f"  Mean Latency: {res['mean_latency_ms']}ms")
            logger.info(f"  P99 Latency: {res['p99_latency_ms']}ms")
            logger.info(f"  Failures: {res['failures']}")

    # Calculate TCO
    if zapier_results and n8n_results:
        tco = calculate_tco(
            zapier_results,
            saas_monthly_cost=49.0,  # Zapier Starter
            automation_monthly_cost=16.8  # AWS t4g.medium
        )
        logger.info("\n=== 3-Year TCO (1000 events/day) ===")
        logger.info(f"  SaaS (Zapier): ${tco['saas_3yr_tco']:.2f}")
        logger.info(f"  Custom Automation (n8n): ${tco['automation_3yr_tco']:.2f}")
        logger.info(f"  SaaS is {tco['cost_difference_percent']}% more expensive than automation")