"""
TTG Scratchpad MCP Server - Database Index Creation Script
Run this script to create/update database indexes after deployment.
Usage:
python scripts/create_indexes.py
Environment Variables Required:
MONGODB_URI - MongoDB Atlas connection string
"""
import asyncio
import os
import sys
from datetime import datetime
# Add parent directory to path for imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from dotenv import load_dotenv
from motor.motor_asyncio import AsyncIOMotorClient
# Load environment variables
load_dotenv()
async def create_indexes():
"""Create all database indexes for the scratchpad MCP server."""
mongo_uri = os.environ.get("MONGODB_URI")
if not mongo_uri:
print("ERROR: MONGODB_URI environment variable not set")
print("Set it to your MongoDB Atlas connection string")
sys.exit(1)
print(f"[{datetime.now().isoformat()}] Connecting to MongoDB...")
client = AsyncIOMotorClient(
mongo_uri,
serverSelectionTimeoutMS=10000,
connectTimeoutMS=10000,
)
db = client["ttg-workspaces"]
# Verify connection
try:
await client.admin.command('ping')
print(f"[{datetime.now().isoformat()}] Connected to MongoDB (ttg-workspaces)")
except Exception as e:
print(f"ERROR: Failed to connect to MongoDB: {e}")
sys.exit(1)
print(f"\n[{datetime.now().isoformat()}] Creating indexes...")
try:
# ============================================================
# WORKSPACES COLLECTION
# ============================================================
print("\n--- Workspaces Collection ---")
# User + status index (for finding active workspaces)
idx = await db.workspaces.create_index(
[("user_id", 1), ("status", 1)],
name="user_status_idx"
)
print(f" Created index: {idx}")
# User + updated_at index (for sorting by recent)
idx = await db.workspaces.create_index(
[("user_id", 1), ("updated_at", -1)],
name="user_updated_idx"
)
print(f" Created index: {idx}")
# User + conversation_id index (for conversation scoping)
idx = await db.workspaces.create_index(
[("user_id", 1), ("conversation_id", 1)],
name="user_conversation_idx"
)
print(f" Created index: {idx}")
# TTL index - auto-delete completed workspaces after 7 days
idx = await db.workspaces.create_index(
[("updated_at", 1)],
expireAfterSeconds=7 * 24 * 60 * 60, # 7 days
name="workspace_ttl_idx",
partialFilterExpression={"status": "complete"} # Only delete completed workspaces
)
print(f" Created TTL index: {idx}")
# ============================================================
# FILES COLLECTION
# ============================================================
print("\n--- Files Collection ---")
# User + workspace index (for listing files)
idx = await db.files.create_index(
[("user_id", 1), ("workspace_id", 1)],
name="user_workspace_idx"
)
print(f" Created index: {idx}")
# Workspace + path unique index (prevent duplicate paths)
idx = await db.files.create_index(
[("workspace_id", 1), ("path", 1)],
unique=True,
name="workspace_path_unique_idx"
)
print(f" Created index: {idx}")
# TTL index - auto-delete files after 7 days
idx = await db.files.create_index(
[("updated_at", 1)],
expireAfterSeconds=7 * 24 * 60 * 60, # 7 days
name="files_ttl_idx"
)
print(f" Created TTL index: {idx}")
# ============================================================
# ACTIVITY LOGS COLLECTION
# ============================================================
print("\n--- Activity Logs Collection ---")
# User + timestamp index (for user activity history)
idx = await db.activity_logs.create_index(
[("user_id", 1), ("timestamp", -1)],
name="user_timestamp_idx"
)
print(f" Created index: {idx}")
# Workspace + timestamp index (for workspace activity)
idx = await db.activity_logs.create_index(
[("workspace_id", 1), ("timestamp", -1)],
name="workspace_timestamp_idx"
)
print(f" Created index: {idx}")
# TTL index - auto-delete activity logs after 7 days
idx = await db.activity_logs.create_index(
[("timestamp", 1)],
expireAfterSeconds=7 * 24 * 60 * 60, # 7 days
name="activity_ttl_idx"
)
print(f" Created TTL index: {idx}")
print(f"\n[{datetime.now().isoformat()}] All indexes created successfully!")
# ============================================================
# VERIFY INDEXES
# ============================================================
print(f"\n[{datetime.now().isoformat()}] Verifying indexes...")
print("\n--- Workspaces Indexes ---")
async for idx in db.workspaces.list_indexes():
print(f" {idx['name']}: {idx['key']}")
print("\n--- Files Indexes ---")
async for idx in db.files.list_indexes():
print(f" {idx['name']}: {idx['key']}")
print("\n--- Activity Logs Indexes ---")
async for idx in db.activity_logs.list_indexes():
print(f" {idx['name']}: {idx['key']}")
except Exception as e:
print(f"ERROR creating indexes: {e}")
sys.exit(1)
finally:
client.close()
print(f"\n[{datetime.now().isoformat()}] Done!")
if __name__ == "__main__":
asyncio.run(create_indexes())
