GEO-Butler Backend Architecture: A Deep Dive into Serverless Search Analytics

Introduction

GEO-Butler is a serverless SEO analytics platform that tracks website visibility across AI-powered search engines like ChatGPT, Gemini, and Perplexity. This post explores the backend architecture that powers real-time search operations, subscription billing, and AI-driven insights.

Tech Stack:

• Backend: Python 3.12 Cloud Functions (90+ endpoints)
• Database: Firebase Firestore (20 collections)
• APIs: DataForSEO, OpenAI, Gemini, Perplexity, Stripe
• Async Processing: Cloud Pub/Sub
• Region: europe-west2

System Architecture Overview

The platform is built entirely on serverless infrastructure, with Firebase Cloud Functions handling all API logic. Here's how the pieces fit together:

Key Architectural Decisions

1. Serverless-First: All backend logic runs in Cloud Functions that scale to zero when idle
2. Event-Driven: Pub/Sub decouples long-running operations from API responses
3. Multi-Engine: Abstract search interface supports 5 different search engines
4. Atomic Credits: Credit system uses atomic operations (not transactions) to prevent contention
5. Firestore-Native: NoSQL structure optimized for real-time updates and concurrent writes

Backend Service Breakdown

Our Python backend is organized into 12 core modules powering 90+ API endpoints:

Module Responsibilities

• billing_functions.py: Atomic credit operations with exponential backoff
• search_functions.py: Multi-engine abstraction (Google, GPT, Perplexity, Gemini)
• report_functions_v2.py: Lightweight reports with separate query_status for concurrency
• schema_functions.py: AI-powered schema.org generation with Gemini vision
• email_functions.py: Queue-based email delivery with template support

Data Flow #1: Single Search Operation

Let's trace a single search request from user input to final results:

Key Implementation Details

Credit Check Before Execution:

# Check balance BEFORE running expensive operation
balance = get_wallet_balance(user_id)
if balance < SEARCH_WEIGHTS[search_engine]:
    return Response("Insufficient credits", status=402)

# Execute search
results = execute_search(query, engine)

# Deduct credits AFTER successful execution
deduct_tokens_for_search(user_id, query_id, search_engine)

Atomic Wallet Updates:

def update_token_balance_atomic(user_id, amount, operation_type):
    """
    Atomic operation (NOT transaction) to prevent contention.
    Allows concurrent credit operations with retry logic.
    """
    max_retries = 5
    for attempt in range(max_retries):
        try:
            wallet_doc = get_wallet(user_id)
            new_balance = wallet_doc['current_balance'] + amount

            # Allow small negative balance (up to -100)
            if new_balance < -100:
                return False

            # Atomic update (no transaction lock)
            wallet_doc.reference.update({
                'current_balance': new_balance,
                'last_updated': SERVER_TIMESTAMP
            })

            # Log transaction
            log_transaction(user_id, amount, operation_type, new_balance)
            return True
        except Exception:
            # Exponential backoff with jitter
            time.sleep(0.1 * (2 ** attempt) + random.uniform(0, 0.1))

    return False

Why Atomic (Not Transactional)?

• Firestore transactions lock documents, causing contention under load
• Atomic operations allow concurrent credit operations
• Small negative balances (-100) act as buffer for race conditions
• Exponential backoff handles rare collision cases

Data Flow #2: Template Report Generation (Async)

Templates allow users to run multiple queries across multiple search engines. This generates 10s or 100s of individual searches that must run asynchronously:

Architectural Patterns

CQRS (Command Query Responsibility Segregation):

• Separate query_status collection prevents report document contention
• 30 concurrent workers can update individual query_status docs
• Report doc only tracks counts, not full results

Pub/Sub for Scale:

def search_template(template_id, user_id):
    """
    Publish search tasks to Pub/Sub instead of processing synchronously.
    Prevents function timeout and enables parallel execution.
    """
    template = get_template(template_id)
    report_id = create_report(template_id)

    # Create query_status docs (lightweight, no contention)
    for query in template['queries']:
        for engine in template['search_engines']:
            create_query_status(report_id, query, engine)

    # Publish to Pub/Sub for async processing
    for query in template['queries']:
        for engine in template['search_engines']:
            publish_message('search-tasks', {
                'report_id': report_id,
                'query': query,
                'engine': engine,
                'user_id': user_id
            })

    return {'report_id': report_id, 'status': 'in_progress'}

Why This Works:

• 540-second function timeout allows ~60 seconds per search
• 30 queries × 3 engines = 90 individual tasks
• Sequential would take 90 × 5s = 450s (near timeout)
• Pub/Sub allows parallel execution: 30 functions × 5s = 5s elapsed
• Frontend polls status every 5 seconds for real-time updates

Data Flow #3: Subscription & Billing with Stripe

Payment processing is entirely webhook-driven, ensuring reliability even if the user closes their browser:

Webhook Implementation

Signature Verification (Critical):

@https_fn.on_request()
def stripe_webhook(req):
    """Handle all Stripe webhook events with signature verification."""
    payload = req.get_data()
    sig_header = req.headers.get('Stripe-Signature')

    try:
        # Verify webhook signature (prevents spoofing)
        event = stripe.Webhook.construct_event(
            payload, sig_header, STRIPE_WEBHOOK_SECRET
        )
    except ValueError:
        return Response("Invalid payload", status=400)
    except stripe.error.SignatureVerificationError:
        return Response("Invalid signature", status=400)

    # Route event to appropriate handler
    if event['type'] == 'checkout.session.completed':
        handle_checkout_completed(event['data']['object'])
    elif event['type'] == 'invoice.paid':
        handle_invoice_paid(event['data']['object'])
    elif event['type'] == 'customer.subscription.updated':
        handle_subscription_updated(event['data']['object'])

    return Response("Success", status=200)

Idempotent Credit Allocation:

def handle_invoice_paid(invoice):
    """
    Add credits when subscription invoice is paid.
    Handles both initial subscription and monthly renewals.
    """
    subscription_id = invoice['subscription']
    billing_reason = invoice['billing_reason']

    # Get subscription from Firestore
    subscription = get_subscription_by_stripe_id(subscription_id)
    user_id = subscription['user_id']
    credits = subscription['credits_per_cycle']

    # Determine operation type
    if billing_reason == 'subscription_create':
        operation_type = 'subscription_credit'
    elif billing_reason == 'subscription_cycle':
        operation_type = 'renewal_credit'
    else:
        operation_type = 'subscription_credit'

    # Add credits atomically
    success = update_token_balance_atomic(
        user_id=user_id,
        amount=credits,
        operation_type=operation_type,
        description=f"{operation_type}: {credits} credits"
    )

    if success:
        # Update subscription metadata
        update_subscription_payment_details(
            subscription_id,
            invoice['amount_paid'],
            invoice['status']
        )

Credit Cost Structure:

Data Flow #4: AI Schema Generation with Gemini

One of our most complex features is AI-powered schema.org markup generation using multimodal analysis:

Multimodal AI Implementation

Gemini Vision Prompt:

def generate_schema_with_gemini(domain, schema_type, screenshot_base64, html_content):
    """
    Generate schema.org JSON-LD using Gemini's multimodal capabilities.
    Analyzes both visual appearance and HTML structure.
    """
    model = genai.GenerativeModel('gemini-2.5-flash')

    # Prepare multimodal input
    image_part = {
        'mime_type': 'image/png',
        'data': screenshot_base64
    }

    prompt = f"""
    Analyze this website and generate comprehensive schema.org JSON-LD markup.

    Domain: {domain}
    Requested Schema Type: {schema_type}

    Use the screenshot to understand:
    - Visual branding and design
    - Layout and structure
    - Key content placement

    Use the HTML to extract:
    - Structured data
    - Metadata
    - Content hierarchy

    Generate verbose, detailed schema with all relevant properties.
    Follow schema.org vocabulary strictly.

    HTML Content:
    {html_content[:10000]}  # First 10KB
    """

    # Call Gemini with multimodal input
    response = model.generate_content([prompt, image_part])
    schema_json = extract_json_from_response(response.text)

    # Validate against schema.org vocabulary
    validation_results = validate_schema(schema_json)

    return {
        'schema_json': schema_json,
        'validation_score': validation_results['score'],
        'validation_errors': validation_results['errors']
    }

Schema Validation:

def validate_schema_against_vocabulary(schema_json):
    """
    Validate generated schema against official schema.org vocabulary.
    Loaded from 1.5MB schemaorg-current.jsonld file.
    """
    vocabulary = load_schema_org_vocabulary()  # Cached in memory

    errors = []
    warnings = []
    score = 100

    # Check type exists
    if schema_json['@type'] not in vocabulary['types']:
        errors.append(f"Invalid type: {schema_json['@type']}")
        score -= 50

    # Check properties
    for prop, value in schema_json.items():
        if prop.startswith('@'):
            continue

        if prop not in vocabulary['properties']:
            warnings.append(f"Non-standard property: {prop}")
            score -= 5

    return {
        'score': max(0, score),
        'errors': errors,
        'warnings': warnings
    }

Database Architecture

Our Firestore database uses 20 collections optimized for real-time updates and concurrent access:

Key Design Decisions

Separate query_status Collection:

• Prevents report document contention during concurrent updates
• Each query×engine combination gets its own lightweight document
• Workers update individual status docs in parallel
• Report doc only tracks aggregate counts

User-Scoped user_data Maps:

interface Site {
  domain: string;
  public_data: {
    title: string;
    screenshot_base64: string;
    // Shared across all users
  };
  user_data: {
    [user_id: string]: {
      private_data: {
        competitors: string[];
        custom_notes: string;
        // Private to each user
      }
    }
  }
}

Atomic Wallet Updates:

• No transactions = no lock contention
• Small negative balance buffer (-100 credits) handles race conditions
• Every operation logged in token_transactions for audit trail

External API Integration

GEO-Butler integrates with 8 external services:

Multi-Engine Search Abstraction

class SearchEngine:
    """Abstract base class for all search engines."""

    @abstractmethod
    def search(self, query: str, location: str, language: str) -> dict:
        pass

    @abstractmethod
    def get_cost(self) -> int:
        pass

class GoogleSearchEngine(SearchEngine):
    """Google Search via DataForSEO API."""
    def search(self, query, location, language):
        response = requests.post(
            'https://api.dataforseo.com/v3/serp/google/organic/live/advanced',
            auth=(DATAFORSEO_LOGIN, DATAFORSEO_PASSWORD),
            json=[{
                'keyword': query,
                'location_code': location,
                'language_code': language
            }]
        )
        return self.parse_results(response.json())

    def get_cost(self):
        return 2  # credits

class GPTSearchEngine(SearchEngine):
    """GPT-powered search (Google + AI analysis)."""
    def search(self, query, location, language):
        # First get Google results
        google_results = GoogleSearchEngine().search(query, location, language)

        # Then analyze with GPT
        analysis = openai.chat.completions.create(
            model='gpt-5-nano',
            messages=[{
                'role': 'system',
                'content': 'Analyze these search results and provide insights.'
            }, {
                'role': 'user',
                'content': f"Query: {query}\n\nResults: {google_results}"
            }]
        )

        return {
            'google_results': google_results,
            'ai_analysis': analysis.choices[0].message.content
        }

    def get_cost(self):
        return 6  # credits (2 for Google + 4 for GPT)

Security Architecture

Four layers of security protect every API call:

Token Verification Pattern

Every Cloud Function starts with this:

def verify_user(request):
    """Verify Firebase ID token and extract user_id."""
    user_id = request.headers.get('user-id')
    auth_header = request.headers.get('Authorization')

    if not user_id or not auth_header:
        raise ValueError("Missing authentication")

    # Extract token from "Bearer <token>"
    token = auth_header.split('Bearer ')[1]

    try:
        # Verify with Firebase Admin SDK
        decoded_token = firebase_auth.verify_id_token(token)

        # Ensure user_id matches token
        if decoded_token['uid'] != user_id:
            raise ValueError("User ID mismatch")

        return user_id, decoded_token
    except Exception as e:
        raise ValueError(f"Invalid token: {e}")

@https_fn.on_request()
def my_endpoint(req):
    try:
        user_id, token = verify_user(req)

        # Process request for verified user
        data = get_user_data(user_id)
        return Response(json.dumps(data), status=200)

    except ValueError as e:
        return Response(str(e), status=401)

Performance & Scalability

Current Configuration

Cloud Functions:

• Runtime: Python 3.12
• Memory: 256MB - 2048MB (function-specific)
• Timeout: 30s - 540s
• Concurrency: 20 per instance
• Max Instances: 50
• Min Instances: 0 (scale to zero)

Firestore:

• Write Rate: 10,000/sec per database
• Read Rate: 100,000/sec per database
• Document Limit: 1 MB
• Collections: 20

Scaling Strategies

1. Horizontal Scaling: Functions auto-scale based on load
2. Async Processing: Pub/Sub prevents function timeouts
3. CQRS Pattern: Separate query_status reduces contention
4. Atomic Operations: No transaction locks = better concurrency
5. Efficient Indexing: Composite indexes for common queries
6. Caching: Schema.org vocabulary cached in memory (1.5MB)

Cost Optimization

• Scale to zero when idle (no minimum instances)
• Efficient Firestore queries (indexed properly)
• Minimal data in report documents (references, not full results)
• Batch API calls where possible

Lessons Learned

1. Transactions vs. Atomic Operations

Problem: Initial implementation used Firestore transactions for credit updates, causing contention under load (concurrent template reports).

Solution: Switched to atomic operations with exponential backoff. Small negative balance buffer (-100 credits) handles rare race conditions.

Result: 10x improvement in concurrent report execution.

2. Report Document Contention

Problem: Storing all query results in report document caused timeouts when 30+ workers tried to update simultaneously.

Solution: Created separate query_status collection. Report doc only tracks counts, not full results.

Result: Reports with 100+ queries now complete reliably.

3. Function Timeouts on Large Templates

Problem: Processing 50 queries × 3 engines = 150 searches sequentially would timeout at 540 seconds.

Solution: Immediate Pub/Sub publish with 202 Accepted response. Frontend polls status.

Result: Templates of any size complete successfully.

4. Webhook Reliability

Problem: User closing browser during payment caused missed subscription credits.

Solution: 100% webhook-driven billing. Credits added asynchronously after payment confirmation.

Result: Zero missed credit allocations.

Conclusion

Building GEO-Butler taught us that serverless architecture requires different thinking than traditional backends:

• Embrace async: Don't try to complete everything in one request
• Avoid locks: Atomic operations > transactions for high concurrency
• Separate concerns: CQRS pattern prevents document contention
• Trust webhooks: Don't wait for synchronous responses from payment providers
• Monitor everything: Firestore operations, function execution times, error rates

The result is a platform that scales effortlessly from zero to thousands of concurrent searches, with sub-second API responses and reliable billing.

Project: GEO-Butler
Stack: Next.js + Firebase + Python
Architecture: Serverless Microservices
Database: Firestore (20 collections)
API Endpoints: 90+
External APIs: 8
Region: europe-west2

Documentation Repository: GitHub

Want to learn more? Check out the full architecture documentation for detailed API references, database schemas, and component architecture.