Analytics ETL Pipeline Design - MongoDB Edition

AI Chat System Usage Analytics with MongoDB Data Warehouse

Version: 1.0 (MongoDB Variant)
Date: November 10, 2025
Status: Design Proposal

---

Table of Contents

1. Overview
2. MongoDB vs PostgreSQL Decision
3. MongoDB Data Warehouse Schema
4. ETL Pipeline for MongoDB
5. API Implementation with MongoDB
6. Performance Optimization
7. Pros and Cons
8. Implementation Guide

---

Overview

This document adapts the analytics ETL design to use MongoDB as the data warehouse instead of PostgreSQL. This approach leverages your existing MongoDB infrastructure and avoids introducing a new database technology.

Key Differences from PostgreSQL Design

Aspect	PostgreSQL Design	MongoDB Design
Schema	Star schema (normalized)	Denormalized documents
Joins	SQL JOINs	Embedded documents or $lookup
Aggregations	GROUP BY, window functions	Aggregation pipeline
Indexes	B-tree, GIN, etc.	Single/compound/text indexes
Storage	Separate database	Separate MongoDB database
Query Language	SQL	MongoDB aggregation framework

---

MongoDB vs PostgreSQL Decision

Pros of Using MongoDB

✅ No New Infrastructure - Use existing MongoDB servers - No new database to manage - Familiar technology for your team - Same backup/monitoring tools

✅ Operational Simplicity - Single database technology - Unified security/access control - Easier deployment - Lower operational overhead

✅ Flexible Schema - Easy to add new fields - No schema migrations needed - Handles varying document structures well

✅ Good for Document-Heavy Data - Questions/answers are text-heavy (good fit) - Nested data structures natural - JSON-like documents easy to work with

✅ Fast Development - Your team already knows MongoDB - Faster initial implementation - Less learning curve

Cons of Using MongoDB

❌ Less Optimized for Analytics - Aggregation pipeline more verbose than SQL - No window functions (running totals, etc.) - Limited support for complex joins - GROUP BY operations less efficient than PostgreSQL

❌ Query Performance - Generally 2-5x slower than PostgreSQL for analytics queries - Aggregation pipeline less optimized - Index-only queries harder to achieve

❌ Storage Efficiency - Denormalized data = more storage - Repeated data across documents - Can be 2-3x larger than normalized SQL

❌ Limited Analytics Features - No materialized views - No CTEs (Common Table Expressions) - No PARTITION BY for window functions - Less sophisticated query optimizer

❌ Aggregation Complexity - Complex aggregations more difficult to write - Longer pipeline stages - Harder to debug

Performance Comparison

Query Type	PostgreSQL	MongoDB	Winner
Simple aggregations	50ms	150ms	PostgreSQL (3x)
Complex joins	100ms	500ms	PostgreSQL (5x)
Text search	200ms	150ms	MongoDB (1.3x)
Time-series aggregations	80ms	250ms	PostgreSQL (3x)
Document retrieval	10ms	5ms	MongoDB (2x)
Full-text search	300ms	100ms	MongoDB (3x)

Overall: PostgreSQL is 2-5x faster for analytics workloads, but MongoDB is faster for document retrieval and text search.

Recommendation

Use MongoDB if: - ✅ You want to avoid adding new infrastructure - ✅ Operational simplicity is priority #1 - ✅ Your team lacks PostgreSQL experience - ✅ 2-5x slower queries are acceptable (still 10-20x faster than current) - ✅ You need fast text search - ✅ Storage cost is not a concern

Use PostgreSQL if: - ✅ You need maximum query performance - ✅ You have complex analytical requirements - ✅ Storage efficiency matters - ✅ You need advanced SQL features (window functions, CTEs) - ✅ You're willing to manage another database

For your use case: MongoDB is a reasonable choice for MVP, with option to migrate to PostgreSQL later if performance becomes an issue.

---

MongoDB Data Warehouse Schema

Database Structure

MongoDB Server
├── <catname>_cat_data (operational - existing)
│   └── logs (raw chat logs - source data)
│
└── bast_analytics (new - analytics warehouse)
    ├── dim_cat (dimension)
    ├── dim_topic (dimension)
    ├── dim_model (dimension)
    ├── dim_question (dimension)
    ├── dim_answer (dimension)
    ├── dim_session (dimension)
    ├── dim_user (dimension)
    ├── fact_interaction (atomic facts)
    ├── fact_hourly_summary (pre-aggregated)
    ├── fact_daily_summary (pre-aggregated)
    ├── fact_weekly_summary (pre-aggregated)
    ├── fact_monthly_summary (pre-aggregated)
    ├── etl_log (metadata)
    ├── etl_checkpoint (metadata)
    └── data_quality_metrics (metadata)

---

Dimension Collections

1. dim_cat

{
  _id: ObjectId("..."),
  cat_name: "testcat",           // Indexed (unique)
  cat_prefix: "test",
  cat_description: "Test catalog",
  is_active: true,
  created_at: ISODate("2025-01-01T00:00:00Z"),
  updated_at: ISODate("2025-01-01T00:00:00Z")
}

// Indexes
db.dim_cat.createIndex({ cat_name: 1 }, { unique: true });

2. dim_topic

{
  _id: ObjectId("..."),
  topic_label: "advising",       // Indexed (unique)
  topic_category: "academic",    // Optional parent category
  topic_description: "Academic advising questions",
  is_active: true,
  created_at: ISODate("2025-01-01T00:00:00Z"),
  updated_at: ISODate("2025-01-01T00:00:00Z")
}

// Indexes
db.dim_topic.createIndex({ topic_label: 1 }, { unique: true });

3. dim_model

{
  _id: ObjectId("..."),
  model_name: "gpt-4o",          // Indexed (unique)
  model_provider: "OpenAI",
  model_version: "2024-05-13",
  model_type: "LLM",
  is_active: true,
  created_at: ISODate("2025-01-01T00:00:00Z"),
  updated_at: ISODate("2025-01-01T00:00:00Z")
}

// Indexes
db.dim_model.createIndex({ model_name: 1 }, { unique: true });

4. dim_persona (NEW)

{
  _id: ObjectId("..."),
  persona_name: "teacher",          // Indexed (unique)
  persona_category: "user_type",    // Optional category
  persona_description: "Teacher persona",
  is_active: true,
  created_at: ISODate("2025-01-01T00:00:00Z"),
  updated_at: ISODate("2025-01-01T00:00:00Z")
}

// Indexes
db.dim_persona.createIndex({ persona_name: 1 }, { unique: true });

5. dim_question (Deduplication)

{
  _id: ObjectId("..."),
  question_text: "How do I register for classes?",
  question_hash: "a3f2b8c...",   // SHA256 hash (indexed, unique)
  first_asked_at: ISODate("2025-01-01T10:00:00Z"),
  last_asked_at: ISODate("2025-01-10T15:30:00Z"),
  total_ask_count: 42,           // Updated on each occurrence
  created_at: ISODate("2025-01-01T10:00:00Z")
}

// Indexes
db.dim_question.createIndex({ question_hash: 1 }, { unique: true });
db.dim_question.createIndex({ question_text: "text" });  // Full-text search

6. dim_answer (Deduplication)

{
  _id: ObjectId("..."),
  answer_text: "To register for classes, go to...",
  answer_hash: "b4e3c9d...",     // SHA256 hash (indexed, unique)
  first_given_at: ISODate("2025-01-01T10:00:00Z"),
  last_given_at: ISODate("2025-01-10T15:30:00Z"),
  total_given_count: 42,
  created_at: ISODate("2025-01-01T10:00:00Z")
}

// Indexes
db.dim_answer.createIndex({ answer_hash: 1 }, { unique: true });
db.dim_answer.createIndex({ answer_text: "text" });  // Full-text search

7. dim_session

{
  _id: ObjectId("..."),
  session_id: "uuid-string",     // Indexed (unique)
  user_id: "user123",            // Optional
  session_start_time: ISODate("2025-01-01T10:00:00Z"),
  session_end_time: ISODate("2025-01-01T10:45:00Z"),
  total_interactions: 5,
  created_at: ISODate("2025-01-01T10:00:00Z"),
  updated_at: ISODate("2025-01-01T10:45:00Z")
}

// Indexes
db.dim_session.createIndex({ session_id: 1 }, { unique: true });
db.dim_session.createIndex({ user_id: 1 });

8. dim_user

{
  _id: ObjectId("..."),
  user_id: "user123",            // Indexed (unique)
  user_type: "student",          // student, faculty, staff
  user_segment: "power_user",    // power_user, occasional, new
  first_interaction_date: ISODate("2025-01-01"),
  last_interaction_date: ISODate("2025-01-10"),
  total_interactions: 25,
  is_active: true,
  created_at: ISODate("2025-01-01T00:00:00Z"),
  updated_at: ISODate("2025-01-10T15:30:00Z")
}

// Indexes
db.dim_user.createIndex({ user_id: 1 }, { unique: true });

---

Fact Collections

Design Choice: Embedded vs Referenced

Option 1: Embedded Dimensions (Recommended for MongoDB) - Denormalize dimension data directly into facts - Faster queries (no $lookup needed) - More storage (duplicated data) - Best for MongoDB performance

Option 2: Referenced Dimensions - Store ObjectId references - Normalized (less storage) - Requires $lookup (slower) - More like relational design

Recommendation: Use embedded dimensions for better MongoDB performance.

---

1. fact_interaction (Atomic Grain)

Grain: One document per chat interaction

Design: Embedded (denormalized for performance)

{
  _id: ObjectId("..."),

  // Time dimensions (denormalized for fast filtering)
  ts: 1704110400.123,            // Original epoch timestamp
  date: ISODate("2025-01-01T00:00:00Z"),
  year: 2025,
  month: 1,
  week: 1,
  day_of_month: 1,
  day_of_week: 1,                // 1=Monday, 7=Sunday
  hour: 14,
  minute: 30,

  // CRITICAL: Event field (must match f"{cat.prefix}_api")
  event: "test_api",             // Event type filter

  // Dimension data (embedded for performance)
  cat: {
    _id: ObjectId("..."),
    name: "testcat"
  },

  topic: {
    _id: ObjectId("..."),
    label: "advising"
  },

  model: {
    _id: ObjectId("..."),
    name: "gpt-4o"
  },

  // NEW: Persona dimension (embedded)
  persona: {
    _id: ObjectId("..."),
    name: "teacher"              // or "default_value" if missing
  },

  question: {
    _id: ObjectId("..."),
    text: "How do I register for classes?",
    hash: "a3f2b8c..."
  },

  answer: {
    _id: ObjectId("..."),
    text: "To register for classes...",
    hash: "b4e3c9d..."
  },

  session: {
    _id: ObjectId("..."),
    session_id: "uuid-string"
  },

  user: {
    _id: ObjectId("..."),
    user_id: "user123"
  },

  // Measures (facts)
  response_time_ms: 245,
  token_count_input: 25,
  token_count_output: 150,
  cost_usd: 0.0023,
  has_error: false,
  is_valid_response: true,

  // Source reference
  source_log_id: "64a1b2c3d4e5f6g7h8i9j0k1",  // Original MongoDB _id

  // Metadata
  created_at: ISODate("2025-01-01T14:30:00Z")
}

// Indexes (CRITICAL for performance)
db.fact_interaction.createIndex({ date: 1, hour: 1 });
db.fact_interaction.createIndex({ "cat.name": 1, date: 1 });
db.fact_interaction.createIndex({ "topic.label": 1, date: 1 });
db.fact_interaction.createIndex({ "model.name": 1, date: 1 });
db.fact_interaction.createIndex({ year: 1, month: 1 });
db.fact_interaction.createIndex({ ts: 1 });
db.fact_interaction.createIndex({ source_log_id: 1 }, { unique: true });

// Compound indexes for common queries
db.fact_interaction.createIndex({ "cat.name": 1, "topic.label": 1, date: 1 });
db.fact_interaction.createIndex({ "cat.name": 1, year: 1, month: 1 });

---

2. fact_hourly_summary

Grain: One document per hour, per cat, per topic, per model

{
  _id: ObjectId("..."),

  // Time dimension
  date: ISODate("2025-01-01T00:00:00Z"),
  year: 2025,
  month: 1,
  week: 1,
  hour: 14,                      // 0-23

  // CRITICAL: Event field (for filtering)
  event: "test_api",

  // Dimension keys (embedded)
  cat: {
    _id: ObjectId("..."),
    name: "testcat"
  },

  topic: {
    _id: ObjectId("..."),
    label: "advising"
  },

  model: {
    _id: ObjectId("..."),
    name: "gpt-4o"
  },

  persona: {
    _id: ObjectId("..."),
    name: "teacher"
  },

  // Aggregated measures
  total_interactions: 125,
  unique_users: 45,              // Count distinct user_ids
  unique_sessions: 67,           // Count distinct session_ids
  unique_questions: 89,          // Count distinct question_ids

  avg_response_time_ms: 245.3,
  total_tokens: 15000,
  total_cost_usd: 2.45,

  error_count: 2,
  error_rate: 1.6,               // Percentage

  // Metadata
  created_at: ISODate("2025-01-01T15:00:00Z"),
  updated_at: ISODate("2025-01-01T15:00:00Z")
}

// Indexes
db.fact_hourly_summary.createIndex({ date: 1, hour: 1 });
db.fact_hourly_summary.createIndex({ "cat.name": 1, date: 1 });
db.fact_hourly_summary.createIndex({ year: 1, month: 1, hour: 1 });

// Unique constraint (prevents duplicates) - includes event and persona
db.fact_hourly_summary.createIndex(
  { "cat.name": 1, event: 1, "topic.label": 1, "model.name": 1, "persona.name": 1, date: 1, hour: 1 },
  { unique: true }
);

---

3. fact_daily_summary

Grain: One document per day, per cat, per topic, per model, per persona

{
  _id: ObjectId("..."),

  // Time dimension
  date: ISODate("2025-01-01T00:00:00Z"),
  year: 2025,
  month: 1,
  week: 1,
  day_of_month: 1,
  day_of_week: 1,

  // CRITICAL: Event field (for filtering)
  event: "test_api",

  // Dimension keys (embedded)
  cat: {
    _id: ObjectId("..."),
    name: "testcat"
  },

  topic: {
    _id: ObjectId("..."),
    label: "advising"
  },

  model: {
    _id: ObjectId("..."),
    name: "gpt-4o"
  },

  persona: {
    _id: ObjectId("..."),
    name: "teacher"
  },

  // Aggregated measures
  total_interactions: 1250,
  unique_users: 345,
  unique_sessions: 567,
  unique_questions: 789,

  avg_response_time_ms: 245.3,
  total_tokens: 150000,
  total_cost_usd: 24.50,

  error_count: 15,
  error_rate: 1.2,

  // Additional daily metrics
  peak_hour: 14,                 // Hour with most traffic
  peak_hour_count: 250,          // Interactions in peak hour

  // Metadata
  created_at: ISODate("2025-01-02T02:00:00Z"),
  updated_at: ISODate("2025-01-02T02:00:00Z")
}

// Indexes
db.fact_daily_summary.createIndex({ date: 1 });
db.fact_daily_summary.createIndex({ "cat.name": 1, date: 1 });
db.fact_daily_summary.createIndex({ year: 1, month: 1 });

// Unique constraint - includes event and persona for correct filtering
db.fact_daily_summary.createIndex(
  { "cat.name": 1, event: 1, "topic.label": 1, "model.name": 1, "persona.name": 1, date: 1 },
  { unique: true }
);

---

4. fact_weekly_summary

{
  _id: ObjectId("..."),

  // Time dimension
  year: 2025,
  week: 1,                       // ISO week number (1-53)
  week_start_date: ISODate("2024-12-30T00:00:00Z"),  // Monday
  week_end_date: ISODate("2025-01-05T00:00:00Z"),    // Sunday

  // CRITICAL: Event field (for filtering)
  event: "test_api",

  // Dimension keys (embedded)
  cat: { _id: ObjectId("..."), name: "testcat" },
  topic: { _id: ObjectId("..."), label: "advising" },
  model: { _id: ObjectId("..."), name: "gpt-4o" },
  persona: { _id: ObjectId("..."), name: "teacher" },

  // Aggregated measures
  total_interactions: 8750,
  unique_users: 1234,
  unique_sessions: 2345,
  unique_questions: 3456,

  avg_response_time_ms: 245.3,
  total_tokens: 1050000,
  total_cost_usd: 171.50,

  error_count: 105,
  error_rate: 1.2,

  // Week-over-week growth
  wow_growth_pct: 5.3,           // % change from previous week

  // Metadata
  created_at: ISODate("2025-01-06T02:00:00Z"),
  updated_at: ISODate("2025-01-06T02:00:00Z")
}

// Indexes
db.fact_weekly_summary.createIndex({ year: 1, week: 1 });
db.fact_weekly_summary.createIndex({ "cat.name": 1, year: 1, week: 1 });

// Unique constraint - includes event and persona
db.fact_weekly_summary.createIndex(
  { "cat.name": 1, event: 1, "topic.label": 1, "model.name": 1, "persona.name": 1, year: 1, week: 1 },
  { unique: true }
);

---

5. fact_monthly_summary

{
  _id: ObjectId("..."),

  // Time dimension
  year: 2025,
  month: 1,
  month_name: "January",

  // CRITICAL: Event field (for filtering)
  event: "test_api",

  // Dimension keys (embedded)
  cat: { _id: ObjectId("..."), name: "testcat" },
  topic: { _id: ObjectId("..."), label: "advising" },
  model: { _id: ObjectId("..."), name: "gpt-4o" },
  persona: { _id: ObjectId("..."), name: "teacher" },

  // Aggregated measures
  total_interactions: 38750,
  unique_users: 4567,
  unique_sessions: 7890,
  unique_questions: 9876,

  avg_response_time_ms: 245.3,
  total_tokens: 4650000,
  total_cost_usd: 758.50,

  error_count: 465,
  error_rate: 1.2,

  // Month-over-month growth
  mom_growth_pct: 12.5,          // % change from previous month

  // Metadata
  created_at: ISODate("2025-02-01T02:00:00Z"),
  updated_at: ISODate("2025-02-01T02:00:00Z")
}

// Indexes
db.fact_monthly_summary.createIndex({ year: 1, month: 1 });
db.fact_monthly_summary.createIndex({ "cat.name": 1, year: 1, month: 1 });

// Unique constraint - includes event and persona
db.fact_monthly_summary.createIndex(
  { "cat.name": 1, event: 1, "topic.label": 1, "model.name": 1, "persona.name": 1, year: 1, month: 1 },
  { unique: true }
);

---

Metadata Collections

1. etl_log

{
  _id: ObjectId("..."),
  job_name: "incremental_etl_testcat",
  job_start_time: ISODate("2025-01-01T14:00:00Z"),
  job_end_time: ISODate("2025-01-01T14:05:23Z"),
  status: "success",             // running, success, failed
  records_extracted: 1250,
  records_loaded: 1245,
  error_message: null,
  created_at: ISODate("2025-01-01T14:00:00Z")
}

// Indexes
db.etl_log.createIndex({ job_name: 1, job_start_time: -1 });
db.etl_log.createIndex({ status: 1 });

2. etl_checkpoint

{
  _id: ObjectId("..."),
  cat_name: "testcat",
  last_processed_ts: 1704110400.123,
  last_processed_log_id: "64a1b2c3d4e5f6g7h8i9j0k1",
  last_run_time: ISODate("2025-01-01T14:05:23Z")
}

// Indexes
db.etl_checkpoint.createIndex({ cat_name: 1 }, { unique: true });

3. data_quality_metrics

{
  _id: ObjectId("..."),
  check_date: ISODate("2025-01-01T00:00:00Z"),
  check_name: "missing_topic_labels",
  check_description: "Count of logs without topic_label",
  total_records: 10000,
  failed_records: 125,
  failure_rate: 1.25,
  severity: "medium",            // low, medium, high
  created_at: ISODate("2025-01-02T02:00:00Z")
}

// Indexes
db.data_quality_metrics.createIndex({ check_date: -1 });
db.data_quality_metrics.createIndex({ severity: 1, check_date: -1 });

---

ETL Pipeline for MongoDB

Extract Phase (extract.py)

from datetime import datetime
from typing import Dict, List, Any
from bast_ai_mongo import getNativePyMongo
from bast_ai_config import get_bast_config
from bast_ai.base import Stopwatch
import logging

logger = logging.getLogger(__name__)

class MongoLogExtractor:
    """Extracts new logs from MongoDB operational database."""

    def __init__(self, cat_name: str, batch_size: int = 10000):
        self.cat_name = cat_name
        self.batch_size = batch_size

        # Source database (operational logs)
        bast_config = get_bast_config()
        self.cat = bast_config.get_cat(catname=cat_name)
        source_mongodb = getNativePyMongo(self.cat.mongodb)
        db_name = f"{self.cat.prefix}_cat_data"
        self.source_collection = source_mongodb[db_name]['logs']

        # Warehouse database (for checkpoint)
        # Assume warehouse is on same MongoDB server for simplicity
        self.warehouse_db = source_mongodb['bast_analytics']
        self.checkpoint_collection = self.warehouse_db['etl_checkpoint']

    def get_last_checkpoint(self) -> float:
        """Get last processed timestamp from warehouse."""
        checkpoint = self.checkpoint_collection.find_one({'cat_name': self.cat_name})
        if checkpoint:
            return checkpoint['last_processed_ts']
        return 0.0  # Start from beginning if no checkpoint

    def extract_new_logs(self, checkpoint_ts: float) -> List[Dict[str, Any]]:
        """
        Extract logs newer than checkpoint timestamp.

        Args:
            checkpoint_ts: Last processed timestamp (epoch seconds)

        Returns:
            List of log documents
        """
        sw = Stopwatch()
        sw.start()

        # Query for logs newer than checkpoint
        # CRITICAL: Must filter by event field (only API events)
        event_name = f"{self.cat.prefix}_api"
        query = {
            'ts': {'$gt': checkpoint_ts},
            'event': event_name  # Filter by event type
        }

        # Use cursor with limit and sort
        cursor = self.source_collection.find(query).sort('ts', 1).limit(self.batch_size)

        logs = list(cursor)

        sw.stop()
        logger.info(f"Extracted {len(logs)} logs in {sw.duration:.2f}s")

        return logs

    def extract_incremental(self) -> List[Dict[str, Any]]:
        """Extract logs since last checkpoint."""
        checkpoint_ts = self.get_last_checkpoint()
        logger.info(f"Extracting logs since timestamp: {checkpoint_ts}")
        return self.extract_new_logs(checkpoint_ts)

---

Transform Phase (transform.py)

from typing import Dict, List, Any, Optional
from datetime import datetime
import hashlib
import logging

logger = logging.getLogger(__name__)

class LogTransformer:
    """Transforms raw MongoDB logs into warehouse format."""

    def __init__(self):
        pass

    def validate_log(self, log: Dict[str, Any]) -> bool:
        """Validate log entry has required fields."""
        required_fields = ['_id', 'ts', 'event']  # Added 'event' as required
        for field in required_fields:
            if field not in log:
                logger.warning(f"Missing required field: {field}")
                return False

        if not isinstance(log['ts'], (int, float)):
            logger.warning(f"Invalid ts type: {type(log['ts'])}")
            return False

        # Validate event matches expected pattern
        catname = log.get('catname', '')
        expected_event = f"{catname}_api"
        if log.get('event') != expected_event:
            logger.warning(f"Unexpected event: {log.get('event')}, expected: {expected_event}")
            # Still process but log warning

        return True

    def clean_text(self, text: Optional[str]) -> Optional[str]:
        """Clean and normalize text fields."""
        if text is None:
            return None

        text = text.strip()
        if not text:
            return None

        # Normalize whitespace
        text = ' '.join(text.split())

        return text

    def hash_text(self, text: str) -> str:
        """Generate SHA256 hash for deduplication."""
        return hashlib.sha256(text.encode('utf-8')).hexdigest()

    def extract_date_parts(self, timestamp: float) -> Dict[str, Any]:
        """Extract date parts from epoch timestamp."""
        dt = datetime.fromtimestamp(timestamp)

        return {
            'ts': timestamp,
            'date': datetime(dt.year, dt.month, dt.day),  # ISODate at day level
            'year': dt.year,
            'month': dt.month,
            'week': dt.isocalendar()[1],
            'day_of_month': dt.day,
            'day_of_week': dt.isoweekday(),
            'hour': dt.hour,
            'minute': dt.minute
        }

    def transform_log(self, log: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """
        Transform a single log entry.

        Args:
            log: Raw log document

        Returns:
            Transformed log dict or None if invalid
        """
        if not self.validate_log(log):
            return None

        # Extract date/time parts
        date_parts = self.extract_date_parts(log['ts'])

        # Clean text fields
        input_text = self.clean_text(log.get('input_text'))
        output_text = self.clean_text(log.get('output'))
        topic_label = self.clean_text(log.get('topic_label'))

        # Build transformed record (matches fact_interaction schema)
        transformed = {
            # Time dimensions
            'ts': date_parts['ts'],
            'date': date_parts['date'],
            'year': date_parts['year'],
            'month': date_parts['month'],
            'week': date_parts['week'],
            'day_of_month': date_parts['day_of_month'],
            'day_of_week': date_parts['day_of_week'],
            'hour': date_parts['hour'],
            'minute': date_parts['minute'],

            # Raw dimension data (will be looked up in load phase)
            'event': log.get('event'),  # CRITICAL: Capture event field
            'cat_name': log.get('catname') or 'unknown',
            'topic_label': topic_label,
            'model_name': log.get('membership') or 'unknown',
            'persona_name': log.get('persona') or 'default_value',  # NEW: Persona field
            'session_id': log.get('session_id'),
            'user_id': log.get('user_id'),

            # Question/Answer
            'question_text': input_text,
            'question_hash': self.hash_text(input_text) if input_text else None,
            'answer_text': output_text,
            'answer_hash': self.hash_text(output_text) if output_text else None,

            # Measures
            'response_time_ms': log.get('response_time_ms'),
            'token_count_input': log.get('token_count_input'),
            'token_count_output': log.get('token_count_output'),
            'cost_usd': log.get('cost_usd'),
            'has_error': log.get('has_error', False),

            # Source reference
            'source_log_id': str(log['_id']),
        }

        return transformed

    def transform_batch(self, logs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Transform a batch of logs."""
        transformed = []

        for log in logs:
            t_log = self.transform_log(log)
            if t_log:
                transformed.append(t_log)

        logger.info(f"Transformed {len(transformed)}/{len(logs)} logs")
        return transformed

---

Load Phase (load.py)

from typing import Dict, List, Any, Optional
from bson import ObjectId
import logging
from bast_ai.base import Stopwatch

logger = logging.getLogger(__name__)

class MongoWarehouseLoader:
    """Loads transformed data into MongoDB warehouse."""

    def __init__(self, warehouse_db):
        """
        Args:
            warehouse_db: MongoDB database connection to 'bast_analytics'
        """
        self.db = warehouse_db

        # Dimension collections
        self.dim_cat = self.db['dim_cat']
        self.dim_topic = self.db['dim_topic']
        self.dim_model = self.db['dim_model']
        self.dim_question = self.db['dim_question']
        self.dim_answer = self.db['dim_answer']
        self.dim_session = self.db['dim_session']
        self.dim_user = self.db['dim_user']

        # Fact collections
        self.fact_interaction = self.db['fact_interaction']

        # Metadata
        self.etl_checkpoint = self.db['etl_checkpoint']

    def upsert_dimension(self, collection, unique_field: str, 
                        data: Dict[str, Any]) -> ObjectId:
        """
        Insert or update dimension record using MongoDB upsert.

        Args:
            collection: MongoDB collection
            unique_field: Field name that must be unique
            data: Record data

        Returns:
            ObjectId of the record
        """
        from datetime import datetime

        # Add timestamps
        data['updated_at'] = datetime.utcnow()
        if 'created_at' not in data:
            data['created_at'] = datetime.utcnow()

        # Upsert (update if exists, insert if not)
        result = collection.update_one(
            {unique_field: data[unique_field]},
            {'$set': data, '$setOnInsert': {'created_at': datetime.utcnow()}},
            upsert=True
        )

        if result.upserted_id:
            return result.upserted_id
        else:
            # Find the existing document
            doc = collection.find_one({unique_field: data[unique_field]})
            return doc['_id']

    def load_dimensions(self, transformed_log: Dict[str, Any]) -> Dict[str, Any]:
        """
        Load all dimension records for a log entry.
        Returns embedded dimension objects.

        Args:
            transformed_log: Transformed log dictionary

        Returns:
            Dict with embedded dimension objects
        """
        dims = {}

        # Load cat dimension
        cat_id = self.upsert_dimension(
            self.dim_cat,
            'cat_name',
            {'cat_name': transformed_log['cat_name']}
        )
        dims['cat'] = {'_id': cat_id, 'name': transformed_log['cat_name']}

        # Load topic dimension (if present)
        if transformed_log.get('topic_label'):
            topic_id = self.upsert_dimension(
                self.dim_topic,
                'topic_label',
                {'topic_label': transformed_log['topic_label']}
            )
            dims['topic'] = {'_id': topic_id, 'label': transformed_log['topic_label']}
        else:
            dims['topic'] = None

        # Load model dimension
        model_id = self.upsert_dimension(
            self.dim_model,
            'model_name',
            {'model_name': transformed_log['model_name']}
        )
        dims['model'] = {'_id': model_id, 'name': transformed_log['model_name']}

        # Load persona dimension (always present, defaults to 'default_value')
        persona_name = transformed_log.get('persona_name') or 'default_value'
        persona_id = self.upsert_dimension(
            self.dim_persona,
            'persona_name',
            {'persona_name': persona_name}
        )
        dims['persona'] = {'_id': persona_id, 'name': persona_name}

        # Load question dimension (if present)
        if transformed_log.get('question_text'):
            question_id = self.upsert_dimension(
                self.dim_question,
                'question_hash',
                {
                    'question_text': transformed_log['question_text'],
                    'question_hash': transformed_log['question_hash']
                }
            )
            dims['question'] = {
                '_id': question_id,
                'text': transformed_log['question_text'],
                'hash': transformed_log['question_hash']
            }
        else:
            dims['question'] = None

        # Load answer dimension (if present)
        if transformed_log.get('answer_text'):
            answer_id = self.upsert_dimension(
                self.dim_answer,
                'answer_hash',
                {
                    'answer_text': transformed_log['answer_text'],
                    'answer_hash': transformed_log['answer_hash']
                }
            )
            dims['answer'] = {
                '_id': answer_id,
                'text': transformed_log['answer_text'],
                'hash': transformed_log['answer_hash']
            }
        else:
            dims['answer'] = None

        # Load session dimension (if present)
        if transformed_log.get('session_id'):
            session_id = self.upsert_dimension(
                self.dim_session,
                'session_id',
                {'session_id': transformed_log['session_id']}
            )
            dims['session'] = {
                '_id': session_id,
                'session_id': transformed_log['session_id']
            }
        else:
            dims['session'] = None

        # Load user dimension (if present)
        if transformed_log.get('user_id'):
            user_id = self.upsert_dimension(
                self.dim_user,
                'user_id',
                {'user_id': transformed_log['user_id']}
            )
            dims['user'] = {
                '_id': user_id,
                'user_id': transformed_log['user_id']
            }
        else:
            dims['user'] = None

        return dims

    def insert_fact_interaction(self, transformed_log: Dict[str, Any], 
                               dims: Dict[str, Any]) -> ObjectId:
        """
        Insert fact_interaction record with embedded dimensions.

        Args:
            transformed_log: Transformed log dictionary
            dims: Embedded dimension objects from load_dimensions()

        Returns:
            ObjectId of inserted record
        """
        from datetime import datetime

        # Build fact document (matches schema)
        fact = {
            # Time dimensions
            'ts': transformed_log['ts'],
            'date': transformed_log['date'],
            'year': transformed_log['year'],
            'month': transformed_log['month'],
            'week': transformed_log['week'],
            'day_of_month': transformed_log['day_of_month'],
            'day_of_week': transformed_log['day_of_week'],
            'hour': transformed_log['hour'],
            'minute': transformed_log['minute'],

            # CRITICAL: Event field (for filtering)
            'event': transformed_log['event'],

            # Embedded dimensions
            'cat': dims['cat'],
            'topic': dims['topic'],
            'model': dims['model'],
            'persona': dims['persona'],  # NEW: Persona dimension
            'question': dims['question'],
            'answer': dims['answer'],
            'session': dims['session'],
            'user': dims['user'],

            # Measures
            'response_time_ms': transformed_log.get('response_time_ms'),
            'token_count_input': transformed_log.get('token_count_input'),
            'token_count_output': transformed_log.get('token_count_output'),
            'cost_usd': transformed_log.get('cost_usd'),
            'has_error': transformed_log.get('has_error', False),
            'is_valid_response': True,

            # Source reference
            'source_log_id': transformed_log['source_log_id'],

            # Metadata
            'created_at': datetime.utcnow()
        }

        # Insert
        result = self.fact_interaction.insert_one(fact)
        return result.inserted_id

    def load_batch(self, transformed_logs: List[Dict[str, Any]]) -> int:
        """
        Load a batch of transformed logs.

        Args:
            transformed_logs: List of transformed log dictionaries

        Returns:
            Number of records loaded
        """
        sw = Stopwatch()
        sw.start()

        loaded_count = 0

        for t_log in transformed_logs:
            try:
                # Load dimensions
                dims = self.load_dimensions(t_log)

                # Insert fact
                self.insert_fact_interaction(t_log, dims)

                loaded_count += 1
            except Exception as e:
                logger.error(f"Failed to load log {t_log.get('source_log_id')}: {e}")

        sw.stop()
        logger.info(f"Loaded {loaded_count} records in {sw.duration:.2f}s")

        return loaded_count

    def update_checkpoint(self, cat_name: str, last_ts: float, 
                         last_log_id: str):
        """Update ETL checkpoint."""
        from datetime import datetime

        self.etl_checkpoint.update_one(
            {'cat_name': cat_name},
            {
                '$set': {
                    'last_processed_ts': last_ts,
                    'last_processed_log_id': last_log_id,
                    'last_run_time': datetime.utcnow()
                }
            },
            upsert=True
        )

---

Aggregation Phase (aggregate.py)

from datetime import datetime, date
from typing import Optional
import logging

logger = logging.getLogger(__name__)

class MongoFactAggregator:
    """Aggregates fact_interaction into summary tables using MongoDB aggregation."""

    def __init__(self, warehouse_db):
        self.db = warehouse_db
        self.fact_interaction = self.db['fact_interaction']
        self.fact_hourly_summary = self.db['fact_hourly_summary']
        self.fact_daily_summary = self.db['fact_daily_summary']

    def aggregate_hourly(self, target_date: Optional[date] = None):
        """
        Aggregate fact_interaction into fact_hourly_summary.

        Args:
            target_date: Date to aggregate (default: today)
        """
        if target_date is None:
            target_date = date.today()

        start_dt = datetime(target_date.year, target_date.month, target_date.day)
        end_dt = datetime(target_date.year, target_date.month, target_date.day, 23, 59, 59)

        # MongoDB aggregation pipeline
        pipeline = [
            # Match date range
            {
                '$match': {
                    'date': {'$gte': start_dt, '$lt': end_dt}
                }
            },
            # Group by hour, cat, topic, model
            {
                '$group': {
                    '_id': {
                        'date': '$date',
                        'hour': '$hour',
                        'cat_name': '$cat.name',
                        'topic_label': '$topic.label',
                        'model_name': '$model.name'
                    },
                    'total_interactions': {'$sum': 1},
                    'unique_users': {'$addToSet': '$user.user_id'},
                    'unique_sessions': {'$addToSet': '$session.session_id'},
                    'unique_questions': {'$addToSet': '$question._id'},
                    'avg_response_time_ms': {'$avg': '$response_time_ms'},
                    'total_tokens': {
                        '$sum': {
                            '$add': ['$token_count_input', '$token_count_output']
                        }
                    },
                    'total_cost_usd': {'$sum': '$cost_usd'},
                    'error_count': {
                        '$sum': {'$cond': ['$has_error', 1, 0]}
                    }
                }
            },
            # Project to clean up
            {
                '$project': {
                    'date': '$_id.date',
                    'year': {'$year': '$_id.date'},
                    'month': {'$month': '$_id.date'},
                    'week': {'$isoWeek': '$_id.date'},
                    'hour': '$_id.hour',
                    'cat': {
                        'name': '$_id.cat_name'
                    },
                    'topic': {
                        'label': '$_id.topic_label'
                    },
                    'model': {
                        'name': '$_id.model_name'
                    },
                    'total_interactions': 1,
                    'unique_users': {'$size': '$unique_users'},
                    'unique_sessions': {'$size': '$unique_sessions'},
                    'unique_questions': {'$size': '$unique_questions'},
                    'avg_response_time_ms': 1,
                    'total_tokens': 1,
                    'total_cost_usd': 1,
                    'error_count': 1,
                    'error_rate': {
                        '$multiply': [
                            {'$divide': ['$error_count', '$total_interactions']},
                            100
                        ]
                    },
                    'created_at': datetime.utcnow(),
                    'updated_at': datetime.utcnow()
                }
            }
        ]

        # Run aggregation
        results = list(self.fact_interaction.aggregate(pipeline))

        # Upsert results into fact_hourly_summary
        for result in results:
            self.fact_hourly_summary.update_one(
                {
                    'cat.name': result['cat']['name'],
                    'topic.label': result['topic']['label'],
                    'model.name': result['model']['name'],
                    'date': result['date'],
                    'hour': result['hour']
                },
                {'$set': result},
                upsert=True
            )

        logger.info(f"Aggregated {len(results)} hourly records for {target_date}")

    def aggregate_daily(self, target_date: Optional[date] = None):
        """
        Aggregate fact_interaction into fact_daily_summary.

        Args:
            target_date: Date to aggregate (default: today)
        """
        if target_date is None:
            target_date = date.today()

        start_dt = datetime(target_date.year, target_date.month, target_date.day)
        end_dt = datetime(target_date.year, target_date.month, target_date.day, 23, 59, 59)

        # Similar pipeline to hourly, but group by day instead of hour
        pipeline = [
            {
                '$match': {
                    'date': {'$gte': start_dt, '$lt': end_dt}
                }
            },
            {
                '$group': {
                    '_id': {
                        'date': '$date',
                        'cat_name': '$cat.name',
                        'topic_label': '$topic.label',
                        'model_name': '$model.name'
                    },
                    'total_interactions': {'$sum': 1},
                    'unique_users': {'$addToSet': '$user.user_id'},
                    'unique_sessions': {'$addToSet': '$session.session_id'},
                    'unique_questions': {'$addToSet': '$question._id'},
                    'avg_response_time_ms': {'$avg': '$response_time_ms'},
                    'total_tokens': {
                        '$sum': {
                            '$add': ['$token_count_input', '$token_count_output']
                        }
                    },
                    'total_cost_usd': {'$sum': '$cost_usd'},
                    'error_count': {
                        '$sum': {'$cond': ['$has_error', 1, 0]}
                    },
                    # For peak hour calculation
                    'hourly_counts': {
                        '$push': {
                            'hour': '$hour',
                            'count': 1
                        }
                    }
                }
            },
            {
                '$project': {
                    'date': '$_id.date',
                    'year': {'$year': '$_id.date'},
                    'month': {'$month': '$_id.date'},
                    'week': {'$isoWeek': '$_id.date'},
                    'day_of_month': {'$dayOfMonth': '$_id.date'},
                    'day_of_week': {'$isoDayOfWeek': '$_id.date'},
                    'cat': {'name': '$_id.cat_name'},
                    'topic': {'label': '$_id.topic_label'},
                    'model': {'name': '$_id.model_name'},
                    'total_interactions': 1,
                    'unique_users': {'$size': '$unique_users'},
                    'unique_sessions': {'$size': '$unique_sessions'},
                    'unique_questions': {'$size': '$unique_questions'},
                    'avg_response_time_ms': 1,
                    'total_tokens': 1,
                    'total_cost_usd': 1,
                    'error_count': 1,
                    'error_rate': {
                        '$multiply': [
                            {'$divide': ['$error_count', '$total_interactions']},
                            100
                        ]
                    },
                    # Note: peak_hour calculation would require additional processing
                    'created_at': datetime.utcnow(),
                    'updated_at': datetime.utcnow()
                }
            }
        ]

        results = list(self.fact_interaction.aggregate(pipeline))

        # Upsert results
        for result in results:
            self.fact_daily_summary.update_one(
                {
                    'cat.name': result['cat']['name'],
                    'topic.label': result['topic']['label'],
                    'model.name': result['model']['name'],
                    'date': result['date']
                },
                {'$set': result},
                upsert=True
            )

        logger.info(f"Aggregated {len(results)} daily records for {target_date}")

---

API Implementation with MongoDB

Updated Analytics Endpoints

1. GET /analytics/summary

@router.get('/analytics/summary')
async def get_analytics_summary(
    request: Request,
    catname: str,
    startDate: Optional[str] = None,
    endDate: Optional[str] = None,
    granularity: str = 'day'
) -> JSONResponse:
    """Get summary analytics from MongoDB warehouse."""
    try:
        # Get warehouse connection
        from bast_ai_mongo import getNativePyMongo
        warehouse_db = getNativePyMongo()['bast_analytics']  # Warehouse database

        # Choose collection based on granularity
        collection_map = {
            'hour': 'fact_hourly_summary',
            'day': 'fact_daily_summary',
            'week': 'fact_weekly_summary',
            'month': 'fact_monthly_summary'
        }
        collection_name = collection_map.get(granularity, 'fact_daily_summary')
        collection = warehouse_db[collection_name]

        # Build aggregation pipeline
        pipeline = []

        # Match filters
        match_conditions = {'cat.name': catname}

        if startDate:
            start_dt = datetime.strptime(startDate, '%Y-%m-%d')
            match_conditions['date'] = {'$gte': start_dt}

        if endDate:
            end_dt = datetime.strptime(endDate, '%Y-%m-%d')
            if 'date' in match_conditions:
                match_conditions['date']['$lt'] = end_dt
            else:
                match_conditions['date'] = {'$lt': end_dt}

        pipeline.append({'$match': match_conditions})

        # Group and aggregate
        pipeline.append({
            '$group': {
                '_id': None,
                'total_interactions': {'$sum': '$total_interactions'},
                'unique_users': {'$max': '$unique_users'},
                'unique_sessions': {'$max': '$unique_sessions'},
                'avg_response_time_ms': {'$avg': '$avg_response_time_ms'},
                'total_cost_usd': {'$sum': '$total_cost_usd'},
                'error_rate': {'$avg': '$error_rate'}
            }
        })

        # Execute
        results = list(collection.aggregate(pipeline))

        if not results:
            return JSONResponse(
                status_code=status.HTTP_200_OK,
                content={
                    'total_interactions': 0,
                    'unique_users': 0,
                    'unique_sessions': 0,
                    'avg_response_time_ms': 0,
                    'total_cost_usd': 0,
                    'error_rate': 0,
                    'period': {
                        'start_date': startDate,
                        'end_date': endDate,
                        'granularity': granularity
                    }
                }
            )

        result = results[0]

        return JSONResponse(
            status_code=status.HTTP_200_OK,
            content={
                'total_interactions': int(result.get('total_interactions', 0)),
                'unique_users': int(result.get('unique_users', 0)),
                'unique_sessions': int(result.get('unique_sessions', 0)),
                'avg_response_time_ms': round(float(result.get('avg_response_time_ms', 0)), 2),
                'total_cost_usd': round(float(result.get('total_cost_usd', 0)), 4),
                'error_rate': round(float(result.get('error_rate', 0)), 2),
                'period': {
                    'start_date': startDate,
                    'end_date': endDate,
                    'granularity': granularity
                }
            }
        )

    except Exception as err:
        return JSONResponse(
            status_code=status.HTTP_400_BAD_REQUEST,
            content={
                'ERROR_CODE': 'ANALYTICS-ERR-100',
                'ERROR_MSG': f'Failed to retrieve summary: {str(err)}'
            }
        )

2. GET /analytics/topics

@router.get('/analytics/topics')
async def get_topic_analytics(
    request: Request,
    catname: str,
    startDate: Optional[str] = None,
    endDate: Optional[str] = None,
    granularity: str = 'day'
) -> JSONResponse:
    """Get topic breakdown from MongoDB warehouse."""
    try:
        from bast_ai_mongo import getNativePyMongo
        warehouse_db = getNativePyMongo()['bast_analytics']

        collection_map = {
            'hour': 'fact_hourly_summary',
            'day': 'fact_daily_summary',
            'week': 'fact_weekly_summary',
            'month': 'fact_monthly_summary'
        }
        collection = warehouse_db[collection_map.get(granularity, 'fact_daily_summary')]

        # Build pipeline
        pipeline = []

        # Match filters
        match_conditions = {'cat.name': catname}

        if startDate:
            start_dt = datetime.strptime(startDate, '%Y-%m-%d')
            match_conditions['date'] = {'$gte': start_dt}

        if endDate:
            end_dt = datetime.strptime(endDate, '%Y-%m-%d')
            if 'date' in match_conditions:
                match_conditions['date']['$lt'] = end_dt
            else:
                match_conditions['date'] = {'$lt': end_dt}

        pipeline.append({'$match': match_conditions})

        # Calculate total interactions for percentage
        pipeline.append({
            '$facet': {
                'totals': [
                    {'$group': {'_id': None, 'grand_total': {'$sum': '$total_interactions'}}}
                ],
                'by_topic': [
                    {
                        '$group': {
                            '_id': '$topic.label',
                            'count': {'$sum': '$total_interactions'}
                        }
                    },
                    {'$sort': {'count': -1}}
                ]
            }
        })

        # Execute
        results = list(collection.aggregate(pipeline))

        if not results or not results[0]['by_topic']:
            return JSONResponse(status_code=status.HTTP_200_OK, content=[])

        grand_total = results[0]['totals'][0]['grand_total'] if results[0]['totals'] else 1
        topic_data = results[0]['by_topic']

        # Format response
        topics = [
            {
                'topic': item['_id'] if item['_id'] else '(unlabeled)',
                'count': int(item['count']),
                'percentage': round((item['count'] / grand_total) * 100, 2)
            }
            for item in topic_data
        ]

        return JSONResponse(status_code=status.HTTP_200_OK, content=topics)

    except Exception as err:
        return JSONResponse(
            status_code=status.HTTP_400_BAD_REQUEST,
            content={
                'ERROR_CODE': 'ANALYTICS-ERR-101',
                'ERROR_MSG': f'Failed to retrieve topics: {str(err)}'
            }
        )

3. GET /analytics/all (Combined Analytics with Selective Filtering)

CRITICAL: This endpoint implements selective filtering where each facet excludes its own dimension from filters. This allows users to see "breakdown by X" while filtering by Y.

Example: If user filters by topic=advising: - total_interactions = only advising interactions (all filters applied) - topic_counts = ALL topics with counts (topic filter excluded, but model/persona filters apply) - model_counts = ALL models with counts (model filter excluded, but topic/persona filters apply) - persona_counts = ALL personas with counts (persona filter excluded, but topic/model filters apply) - questions = ALL questions (questionid filter excluded, but other filters apply) - answers = ALL answers (answerid filter excluded, but other filters apply)

@router.get('/all')
async def get_all_analytics(
    request: Request,
    catname: str,
    startDate: Optional[str] = None,
    endDate: Optional[str] = None,
    model: Optional[Union[str, List[str]]] = Query(default=None),
    persona: Optional[Union[str, List[str]]] = Query(default=None),
    topic: Optional[Union[str, List[str]]] = Query(default=None),
    questionid: Optional[Union[str, List[str]]] = Query(default=None),
    answerid: Optional[Union[str, List[str]]] = Query(default=None)
) -> JSONResponse:
    """
    Get all analytics data in a single request with selective filtering.

    This endpoint uses selective filtering where each facet excludes its own dimension:
    - topic_counts excludes topic filters (shows all topics)
    - model_counts excludes model filters (shows all models)
    - persona_counts excludes persona filters (shows all personas)
    - questions excludes questionid filters (shows all questions)
    - answers excludes answerid filters (shows all answers)
    - total_interactions includes ALL filters
    """
    try:
        from bast_ai_mongo import getNativePyMongo
        from bast_ai.base import get_bast_config

        # Get cat object to access prefix
        bast_config = get_bast_config()
        cat = bast_config.get_cat(catname=catname)

        warehouse_db = getNativePyMongo()['bast_analytics']

        # Get fact_interaction collection (or summary collections if appropriate)
        fact_collection = warehouse_db['fact_interaction']
        event_name = f"{cat.prefix}_api"

        # Build base match (event and date only)
        base_match_conditions = {'event': event_name, 'cat.name': catname}

        if startDate:
            start_dt = datetime.strptime(startDate, '%Y-%m-%d')
            base_match_conditions['date'] = {'$gte': start_dt}

        if endDate:
            end_dt = datetime.strptime(endDate, '%Y-%m-%d')
            if 'date' in base_match_conditions:
                base_match_conditions['date']['$lt'] = end_dt
            else:
                base_match_conditions['date'] = {'$lt': end_dt}

        # Initialize separate filters for each filter type
        model_filter = None
        topic_filter = None
        persona_filter = None
        questionid_input_text_filter = None
        answerid_output_filter = None

        # Process model filtering
        if model is not None:
            model_list = model if isinstance(model, list) else [model]
            if len(model_list) > 0:
                model_filter = {'$in': model_list}

        # Process topic filtering
        if topic is not None:
            topic_list = topic if isinstance(topic, list) else [topic]
            if len(topic_list) > 0:
                topic_filter = {'$in': topic_list}

        # Process persona filtering
        if persona is not None:
            persona_list = persona if isinstance(persona, list) else [persona]
            if len(persona_list) > 0:
                persona_filter = {'$in': persona_list}

        # Process questionid filtering (resolve IDs to input_text values)
        if questionid is not None:
            try:
                questionid_list = questionid if isinstance(questionid, list) else [questionid]

                input_texts = []
                for qid in questionid_list:
                    # Convert questionid string to ObjectId
                    from bson import ObjectId
                    question_object_id = ObjectId(qid)
                    # Find the document with this ID and get its input_text (from source logs)
                    # Note: In warehouse implementation, we'd query fact_interaction
                    question_doc = fact_collection.find_one({'_id': question_object_id})
                    if question_doc is None:
                        raise ValueError(f"Question with ID '{qid}' not found")

                    # Access nested question.text field in warehouse schema
                    input_text = question_doc.get('question', {}).get('text')
                    if input_text is None or input_text == '':
                        raise ValueError(f"Question with ID '{qid}' has no input_text")

                    input_texts.append(input_text)

                # Store the filter separately
                if len(input_texts) == 1:
                    questionid_input_text_filter = input_texts[0]
                elif len(input_texts) > 1:
                    questionid_input_text_filter = {'$in': input_texts}
            except Exception as e:
                if isinstance(e, ValueError):
                    raise
                raise ValueError(f"Invalid questionid format: {str(e)}")

        # Process answerid filtering (resolve IDs to output values)
        if answerid is not None:
            try:
                answerid_list = answerid if isinstance(answerid, list) else [answerid]

                outputs = []
                for aid in answerid_list:
                    # Convert answerid string to ObjectId
                    from bson import ObjectId
                    answer_object_id = ObjectId(aid)
                    # Find the document with this ID and get its output (from source logs)
                    # Note: In warehouse implementation, we'd query fact_interaction
                    answer_doc = fact_collection.find_one({'_id': answer_object_id})
                    if answer_doc is None:
                        raise ValueError(f"Answer with ID '{aid}' not found")

                    # Access nested answer.text field in warehouse schema
                    output = answer_doc.get('answer', {}).get('text')
                    if output is None or output == '':
                        raise ValueError(f"Answer with ID '{aid}' has no output")

                    outputs.append(output)

                # Store the filter separately
                if len(outputs) == 1:
                    answerid_output_filter = outputs[0]
                elif len(outputs) > 1:
                    answerid_output_filter = {'$in': outputs}
            except Exception as e:
                if isinstance(e, ValueError):
                    raise
                raise ValueError(f"Invalid answerid format: {str(e)}")

        # Helper function to build facet-specific filters
        def get_facet_match(exclude_model=False, exclude_topic=False, exclude_persona=False,
                           exclude_questionid=False, exclude_answerid=False):
            """
            Build match conditions for a facet, excluding specified dimensions.

            This is the KEY pattern for selective filtering:
            - Each facet excludes its own dimension
            - All other filters are still applied
            """
            match = {}
            if not exclude_model and model_filter:
                match['model.name'] = model_filter
            if not exclude_topic and topic_filter:
                match['topic.label'] = topic_filter
            if not exclude_persona and persona_filter:
                match['persona.name'] = persona_filter
            if not exclude_questionid and questionid_input_text_filter:
                match['question.text'] = questionid_input_text_filter
            if not exclude_answerid and answerid_output_filter:
                match['answer.text'] = answerid_output_filter
            return match if match else None

        # Build pipeline with $facet for parallel aggregations
        pipeline = []

        # Add base match (event, cat, date only)
        pipeline.append({'$match': base_match_conditions})

        # Use $facet to run multiple aggregations in parallel
        pipeline.append({
            '$facet': {
                # Total interactions - ALL filters (no exclusions)
                'total_interactions': [
                    *([{'$match': m}] if (m := get_facet_match()) else []),
                    {'$count': 'total'}
                ],
                # Topic counts - exclude topic filter (show all topics)
                'topic_counts': [
                    *([{'$match': m}] if (m := get_facet_match(exclude_topic=True)) else []),
                    {
                        '$group': {
                            '_id': {'$ifNull': ['$topic.label', '(unlabeled)']},
                            'count': {'$sum': 1}
                        }
                    },
                    {'$project': {'_id': 0, 'topic': '$_id', 'count': 1}},
                    {'$sort': {'count': -1, 'topic': 1}}
                ],
                # Model counts - exclude model filter (show all models)
                'model_counts': [
                    *([{'$match': m}] if (m := get_facet_match(exclude_model=True)) else []),
                    {
                        '$group': {
                            '_id': {'$ifNull': ['$model.name', '(unlabeled)']},
                            'count': {'$sum': 1}
                        }
                    },
                    {'$project': {'_id': 0, 'model': '$_id', 'count': 1}},
                    {'$sort': {'count': -1, 'model': 1}}
                ],
                # Persona counts - exclude persona filter (show all personas)
                'persona_counts': [
                    *([{'$match': m}] if (m := get_facet_match(exclude_persona=True)) else []),
                    {
                        '$group': {
                            '_id': {'$ifNull': ['$persona.name', '(unlabeled)']},
                            'count': {'$sum': 1}
                        }
                    },
                    {'$project': {'_id': 0, 'persona': '$_id', 'count': 1}},
                    {'$sort': {'count': -1, 'persona': 1}}
                ],
                # Questions - exclude questionid filter (show all questions)
                'questions': [
                    *([{'$match': m}] if (m := get_facet_match(exclude_questionid=True)) else []),
                    {'$match': {'question.text': {'$exists': True, '$ne': None, '$ne': ''}}},
                    {'$sort': {'_id': -1}},
                    {
                        '$group': {
                            '_id': '$question.text',
                            'count': {'$sum': 1},
                            'id': {'$last': '$$ROOT.question._id'}
                        }
                    },
                    {'$project': {'_id': 0, 'question': '$_id', 'count': 1, 'id': 1}},
                    {'$sort': {'count': -1, 'question': 1}}
                ],
                # Answers - exclude answerid filter (show all answers)
                'answers': [
                    *([{'$match': m}] if (m := get_facet_match(exclude_answerid=True)) else []),
                    {'$match': {'answer.text': {'$exists': True, '$ne': None, '$ne': ''}}},
                    {'$sort': {'_id': -1}},
                    {
                        '$group': {
                            '_id': '$answer.text',
                            'count': {'$sum': 1},
                            'id': {'$last': '$$ROOT.answer._id'}
                        }
                    },
                    {'$project': {'_id': 0, 'answers': '$_id', 'count': 1, 'id': 1}},
                    {'$sort': {'count': -1, 'answers': 1}}
                ]
            }
        })

        result = list(fact_collection.aggregate(pipeline))

        if not result:
            return JSONResponse(
                status_code=status.HTTP_200_OK,
                content={
                    'total_interactions': 0,
                    'topic_counts': [],
                    'model_counts': [],
                    'persona_counts': [],
                    'questions': [],
                    'answers': []
                }
            )

        facet_result = result[0]

        # Extract and format results (same as current implementation)
        # ... (extract logic)

        return JSONResponse(
            status_code=status.HTTP_200_OK,
            content={
                'total_interactions': int(total_interactions),
                'topic_counts': topic_counts,
                'model_counts': model_counts,
                'persona_counts': persona_counts,
                'questions': questions,
                'answers': answers
            }
        )

    except Exception as err:
        return JSONResponse(
            status_code=status.HTTP_400_BAD_REQUEST,
            content={
                'ERROR_CODE': 'ANALYTICS-ERR-0',
                'ERROR_MSG': f'Failed to retrieve analytics: {str(err)}'
            }
        )

Key Implementation Pattern: - Use helper function get_facet_match() to build facet-specific filters - Each facet calls get_facet_match() with its dimension excluded - total_interactions applies all filters (no exclusions) - This pattern enables users to filter by one dimension while seeing breakdowns by others

Example Use Case:

GET /analytics/all?catname=testcat&topic=advising&model=gpt-4o

Results: - total_interactions: Count of advising interactions using gpt-4o - topic_counts: ALL topics (but counts reflect gpt-4o model filter) - model_counts: ALL models (but counts reflect advising topic filter) - persona_counts: ALL personas (but counts reflect advising + gpt-4o filters)

---

4. GET /analytics/trends

@router.get('/analytics/trends')
async def get_analytics_trends(
    request: Request,
    catname: str,
    metric: str = 'interactions',
    startDate: Optional[str] = None,
    endDate: Optional[str] = None,
    granularity: str = 'day'
) -> JSONResponse:
    """Get time-series data for charts."""
    try:
        from bast_ai_mongo import getNativePyMongo
        warehouse_db = getNativePyMongo()['bast_analytics']

        # Metric mapping
        metric_map = {
            'interactions': 'total_interactions',
            'users': 'unique_users',
            'sessions': 'unique_sessions',
            'response_time': 'avg_response_time_ms',
            'cost': 'total_cost_usd',
            'errors': 'error_rate'
        }

        if metric not in metric_map:
            raise ValueError(f"Invalid metric: {metric}")

        metric_field = metric_map[metric]

        # Choose collection
        collection_map = {
            'hour': 'fact_hourly_summary',
            'day': 'fact_daily_summary',
            'week': 'fact_weekly_summary',
            'month': 'fact_monthly_summary'
        }
        collection = warehouse_db[collection_map.get(granularity, 'fact_daily_summary')]

        # Build pipeline
        pipeline = []

        # Match filters
        match_conditions = {'cat.name': catname}

        if startDate:
            start_dt = datetime.strptime(startDate, '%Y-%m-%d')
            match_conditions['date'] = {'$gte': start_dt}

        if endDate:
            end_dt = datetime.strptime(endDate, '%Y-%m-%d')
            if 'date' in match_conditions:
                match_conditions['date']['$lt'] = end_dt
            else:
                match_conditions['date'] = {'$lt': end_dt}

        pipeline.append({'$match': match_conditions})

        # Group by date
        if granularity == 'hour':
            pipeline.append({
                '$group': {
                    '_id': {'date': '$date', 'hour': '$hour'},
                    'value': {'$sum': f'${metric_field}'}
                }
            })
            pipeline.append({'$sort': {'_id.date': 1, '_id.hour': 1}})
        else:
            pipeline.append({
                '$group': {
                    '_id': '$date',
                    'value': {'$sum': f'${metric_field}'}
                }
            })
            pipeline.append({'$sort': {'_id': 1}})

        # Execute
        results = list(collection.aggregate(pipeline))

        # Format response
        data = []
        for item in results:
            if granularity == 'hour':
                date_str = f"{item['_id']['date'].strftime('%Y-%m-%d')}-{item['_id']['hour']:02d}"
            else:
                date_str = item['_id'].strftime('%Y-%m-%d')

            data.append({
                'date': date_str,
                'value': float(item['value']) if item['value'] else 0
            })

        return JSONResponse(
            status_code=status.HTTP_200_OK,
            content={
                'metric': metric,
                'granularity': granularity,
                'data': data
            }
        )

    except Exception as err:
        return JSONResponse(
            status_code=status.HTTP_400_BAD_REQUEST,
            content={
                'ERROR_CODE': 'ANALYTICS-ERR-103',
                'ERROR_MSG': f'Failed to retrieve trends: {str(err)}'
            }
        )

---

Performance Optimization

Critical Indexes

// fact_interaction indexes (MOST IMPORTANT)
// CRITICAL: Event filtering index (all analytics queries filter by event)
db.fact_interaction.createIndex({ "cat.name": 1, event: 1, date: 1 });
db.fact_interaction.createIndex({ event: 1, date: 1 });

// Filter-specific indexes for selective filtering pattern
db.fact_interaction.createIndex({ "cat.name": 1, event: 1, "model.name": 1, date: 1 });
db.fact_interaction.createIndex({ "cat.name": 1, event: 1, "topic.label": 1, date: 1 });
db.fact_interaction.createIndex({ "cat.name": 1, event: 1, "persona.name": 1, date: 1 });

// Compound indexes for multi-filter queries (selective filtering uses these)
db.fact_interaction.createIndex({ "cat.name": 1, event: 1, "model.name": 1, "topic.label": 1, date: 1 });
db.fact_interaction.createIndex({ "cat.name": 1, event: 1, "model.name": 1, "persona.name": 1, date: 1 });
db.fact_interaction.createIndex({ "cat.name": 1, event: 1, "topic.label": 1, "persona.name": 1, date: 1 });

// Time-based indexes
db.fact_interaction.createIndex({ date: 1, hour: 1 });
db.fact_interaction.createIndex({ year: 1, month: 1 });
db.fact_interaction.createIndex({ ts: 1 });

// Question/answer text search (for questionid/answerid filtering)
db.fact_interaction.createIndex({ "question.text": 1 });
db.fact_interaction.createIndex({ "answer.text": 1 });

// Deduplication (prevent duplicate loads)
db.fact_interaction.createIndex({ source_log_id: 1 }, { unique: true });

// fact_daily_summary indexes (same pattern with event field)
db.fact_daily_summary.createIndex({ "cat.name": 1, event: 1, date: 1 });
db.fact_daily_summary.createIndex({ "cat.name": 1, event: 1, "topic.label": 1, date: 1 });
db.fact_daily_summary.createIndex({ "cat.name": 1, event: 1, "model.name": 1, date: 1 });
db.fact_daily_summary.createIndex({ "cat.name": 1, event: 1, "persona.name": 1, date: 1 });
db.fact_daily_summary.createIndex({ date: 1 });
db.fact_daily_summary.createIndex(
  { "cat.name": 1, event: 1, "topic.label": 1, "model.name": 1, "persona.name": 1, date: 1 },
  { unique: true }
);

// Dimension indexes
db.dim_cat.createIndex({ cat_name: 1 }, { unique: true });
db.dim_topic.createIndex({ topic_label: 1 }, { unique: true });
db.dim_model.createIndex({ model_name: 1 }, { unique: true });
db.dim_persona.createIndex({ persona_name: 1 }, { unique: true });  // NEW: Persona dimension
db.dim_question.createIndex({ question_hash: 1 }, { unique: true });
db.dim_answer.createIndex({ answer_hash: 1 }, { unique: true });

// Text indexes for search
db.dim_question.createIndex({ question_text: "text" });
db.dim_answer.createIndex({ answer_text: "text" });

Sharding Strategy (for large scale)

// Enable sharding on warehouse database
sh.enableSharding("bast_analytics");

// Shard fact_interaction by cat.name + date (for multi-tenant scalability)
sh.shardCollection(
  "bast_analytics.fact_interaction",
  { "cat.name": 1, date: 1 }
);

// Shard fact_daily_summary by cat.name + date
sh.shardCollection(
  "bast_analytics.fact_daily_summary",
  { "cat.name": 1, date: 1 }
);

---

Pros and Cons Summary

MongoDB Warehouse: Pros ✅

1. No New Infrastructure: Leverage existing MongoDB
2. Operational Simplicity: Single database technology
3. Team Familiarity: No learning curve
4. Flexible Schema: Easy to add fields
5. Good Text Search: Better for searching questions/answers
6. Fast Document Retrieval: Get specific documents quickly
7. Easier Deployment: No new servers to manage

MongoDB Warehouse: Cons ❌

1. 2-5x Slower Than PostgreSQL: For analytics queries
2. More Storage Required: Denormalized data (2-3x larger)
3. Limited Analytics Features: No window functions, CTEs
4. Complex Aggregations: Verbose pipelines
5. Less Efficient Grouping: GROUP BY slower than PostgreSQL

When to Choose MongoDB

✅ Choose MongoDB if: - Operational simplicity is highest priority - You want to avoid adding new infrastructure - Team lacks SQL/PostgreSQL experience - 2-5x slower queries acceptable (still 10-20x faster than current) - You need strong text search - Storage cost is not a concern

When to Consider PostgreSQL

✅ Consider PostgreSQL if: - Maximum query performance critical - Complex analytical requirements - Storage efficiency matters - Need advanced SQL features - Willing to manage another database

---

Implementation Guide

Step 1: Create Warehouse Database

# Connect to MongoDB
mongosh

# Create warehouse database and collections
use bast_analytics

# Create dimension collections
db.createCollection("dim_cat")
db.createCollection("dim_topic")
db.createCollection("dim_model")
db.createCollection("dim_question")
db.createCollection("dim_answer")
db.createCollection("dim_session")
db.createCollection("dim_user")

# Create fact collections
db.createCollection("fact_interaction")
db.createCollection("fact_hourly_summary")
db.createCollection("fact_daily_summary")
db.createCollection("fact_weekly_summary")
db.createCollection("fact_monthly_summary")

# Create metadata collections
db.createCollection("etl_log")
db.createCollection("etl_checkpoint")
db.createCollection("data_quality_metrics")

Step 2: Create Indexes

# Run index creation script
mongosh bast_analytics < scripts/create_indexes.js

Step 3: Implement ETL

Use the Python code provided above in: - etl/extract.py - etl/transform.py - etl/load.py - etl/aggregate.py - etl/orchestrator.py

Step 4: Schedule ETL

# Cron job (every 15 minutes)
*/15 * * * * cd /path/to/bast-ai && pants run src/apis/analytics/etl/run_etl.py -- --job incremental --cat testcat

Step 5: Update API

Use the API code provided above to update your analytics endpoints.

---

Recommendation

For your use case, I recommend using MongoDB as the warehouse for the following reasons:

1. ✅ Operational simplicity - You already have MongoDB infrastructure
2. ✅ Faster MVP - No new database to set up
3. ✅ Team familiarity - Your team knows MongoDB
4. ✅ Good enough performance - Still 10-20x faster than current approach
5. ✅ Migration path - Can migrate to PostgreSQL later if needed

Performance Trade-off: - PostgreSQL: 50-200ms queries - MongoDB: 150-500ms queries - Current (direct MongoDB): 5,000-30,000ms queries

Verdict: MongoDB warehouse gives you 10-20x speedup, which is excellent for MVP. If you need another 3-5x improvement later, migrate to PostgreSQL.

---

Next Steps: Start implementing with MongoDB warehouse using the schemas and code provided above! 🚀