sandbox_db.py

import sqlite3
import uuid
import os
from datetime import datetime, timezone

DB_FOLDER = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scratch")
DB_PATH = os.path.join(DB_FOLDER, "sandbox.db")

def init_db():
    os.makedirs(DB_FOLDER, exist_ok=True)
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    # Enable foreign keys
    cursor.execute("PRAGMA foreign_keys = ON;")
    
    # 1. Problems table
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS problems (
        id TEXT PRIMARY KEY,
        user_id TEXT NOT NULL,
        name TEXT NOT NULL,
        problem_number INTEGER,
        pattern TEXT NOT NULL,
        difficulty TEXT NOT NULL,
        google_doc_url TEXT,
        google_doc_id TEXT,
        optimal_time TEXT DEFAULT 'O(N)',
        optimal_space TEXT DEFAULT 'O(N)',
        description TEXT,
        reference_code TEXT,
        created_at TEXT DEFAULT CURRENT_TIMESTAMP,
        UNIQUE(user_id, name)
    );
    """)
    
    # 2. User Reviews table
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS user_reviews (
        id TEXT PRIMARY KEY,
        user_id TEXT NOT NULL,
        problem_id TEXT NOT NULL,
        box_level INTEGER DEFAULT 1,
        last_reviewed TEXT DEFAULT CURRENT_TIMESTAMP,
        next_review TEXT DEFAULT CURRENT_TIMESTAMP,
        times_correct INTEGER DEFAULT 0,
        total_attempts INTEGER DEFAULT 0,
        UNIQUE(user_id, problem_id),
        FOREIGN KEY(problem_id) REFERENCES problems(id) ON DELETE CASCADE
    );
    """)
    
    # 3. User Streaks table
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS user_streaks (
        user_id TEXT PRIMARY KEY,
        current_streak INTEGER DEFAULT 0,
        longest_streak INTEGER DEFAULT 0,
        last_active_date TEXT,
        updated_at TEXT DEFAULT CURRENT_TIMESTAMP
    );
    """)
    
    # 4. User Journey Progress table
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS user_journey_progress (
        id TEXT PRIMARY KEY,
        user_id TEXT NOT NULL,
        problem_number INTEGER NOT NULL,
        viewed_at TEXT DEFAULT CURRENT_TIMESTAMP,
        UNIQUE(user_id, problem_number)
    );
    """)
    
    # 5. User Scores table
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS user_scores (
        user_id TEXT PRIMARY KEY,
        total_score INTEGER DEFAULT 0,
        python_challenges_completed INTEGER DEFAULT 0,
        sql_challenges_completed INTEGER DEFAULT 0,
        updated_at TEXT DEFAULT CURRENT_TIMESTAMP
    );
    """)
    
    # 6. Challenge History table
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS challenge_history (
        id TEXT PRIMARY KEY,
        user_id TEXT NOT NULL,
        challenge_type TEXT NOT NULL,
        challenge_title TEXT NOT NULL,
        points_awarded INTEGER DEFAULT 0,
        completed_at TEXT DEFAULT CURRENT_TIMESTAMP
    );
    """)
    
    conn.commit()
    
    # Seed default problems and reviews if empty
    cursor.execute("SELECT COUNT(*) FROM problems WHERE user_id = 'sandbox-user-id'")
    if cursor.fetchone()[0] == 0:
        seed_default_problems(cursor)
        conn.commit()
        
    conn.close()

def seed_default_problems(cursor):
    default_problems = [
        {
            "id": "prob-contains-duplicate",
            "user_id": "sandbox-user-id",
            "name": "Contains Duplicate",
            "problem_number": 217,
            "pattern": "Arrays & Hashing",
            "difficulty": "Easy",
            "google_doc_url": "https://leetcode.com/problems/contains-duplicate/",
            "google_doc_id": "doc-contains-duplicate",
            "optimal_time": "O(N)",
            "optimal_space": "O(N)",
            "description": "Given an integer array nums, return true if any value appears at least twice in the array, and return false if every element is distinct.",
            "reference_code": "def solve(nums: list[int]) -> bool:\n    # Return True if any value appears at least twice\n    return len(nums) != len(set(nums))"
        },
        {
            "id": "prob-valid-anagram",
            "user_id": "sandbox-user-id",
            "name": "Valid Anagram",
            "problem_number": 242,
            "pattern": "Arrays & Hashing",
            "difficulty": "Easy",
            "google_doc_url": "https://leetcode.com/problems/valid-anagram/",
            "google_doc_id": "doc-valid-anagram",
            "optimal_time": "O(N)",
            "optimal_space": "O(1)",
            "description": "Given two strings s and t, return true if t is an anagram of s, and false otherwise.",
            "reference_code": "def solve(s: str, t: str) -> bool:\n    # Return True if t is an anagram of s\n    if len(s) != len(t):\n        return False\n    return sorted(s) == sorted(t)"
        },
        {
            "id": "prob-two-sum",
            "user_id": "sandbox-user-id",
            "name": "Two Sum",
            "problem_number": 1,
            "pattern": "Arrays & Hashing",
            "difficulty": "Easy",
            "google_doc_url": "https://leetcode.com/problems/two-sum/",
            "google_doc_id": "doc-two-sum",
            "optimal_time": "O(N)",
            "optimal_space": "O(N)",
            "description": "Given an array of integers nums and an integer target, return indices of the two numbers such that they add up to target.",
            "reference_code": "def solve(nums: list[int], target: int) -> list[int]:\n    # Return indices of the two numbers that add up to target\n    seen = {}\n    for i, n in enumerate(nums):\n        diff = target - n\n        if diff in seen:\n            return [seen[diff], i]\n        seen[n] = i\n    return []"
        },
        {
            "id": "prob-group-anagrams",
            "user_id": "sandbox-user-id",
            "name": "Group Anagrams",
            "problem_number": 49,
            "pattern": "Arrays & Hashing",
            "difficulty": "Medium",
            "google_doc_url": "https://leetcode.com/problems/group-anagrams/",
            "google_doc_id": "doc-group-anagrams",
            "optimal_time": "O(N * K log K)",
            "optimal_space": "O(N * K)",
            "description": "Given an array of strings strs, group the anagrams together. You can return the answer in any order.",
            "reference_code": "def solve(strs: list[str]) -> list[list[str]]:\n    # Group anagrams together\n    ans = {}\n    for s in strs:\n        key = ''.join(sorted(s))\n        if key not in ans:\n            ans[key] = []\n        ans[key].append(s)\n    return list(ans.values())"
        },
        {
            "id": "prob-valid-parentheses",
            "user_id": "sandbox-user-id",
            "name": "Valid Parentheses",
            "problem_number": 20,
            "pattern": "Stack",
            "difficulty": "Easy",
            "google_doc_url": "https://leetcode.com/problems/valid-parentheses/",
            "google_doc_id": "doc-valid-parentheses",
            "optimal_time": "O(N)",
            "optimal_space": "O(N)",
            "description": "Given a string s containing just the characters '(', ')', '{', '}', '[' and ']', determine if the input string is valid.",
            "reference_code": "def solve(s: str) -> bool:\n    # Determine if the parentheses input string is valid\n    stack = []\n    mapping = {')': '(', '}': '{', ']': '['}\n    for char in s:\n        if char in mapping:\n            top_element = stack.pop() if stack else '#'\n            if mapping[char] != top_element:\n                return False\n        else:\n            stack.append(char)\n    return not stack"
        }
    ]
    
    for p in default_problems:
        columns = list(p.keys())
        placeholders = ", ".join(["?"] * len(columns))
        query = f"INSERT OR IGNORE INTO problems ({', '.join(columns)}) VALUES ({placeholders})"
        cursor.execute(query, list(p.values()))
        
    now_iso = datetime.now(timezone.utc).isoformat()
    default_reviews = [
        {
            "id": "rev-contains-duplicate",
            "user_id": "sandbox-user-id",
            "problem_id": "prob-contains-duplicate",
            "box_level": 1,
            "last_reviewed": now_iso,
            "next_review": now_iso,
            "times_correct": 0,
            "total_attempts": 0
        },
        {
            "id": "rev-valid-anagram",
            "user_id": "sandbox-user-id",
            "problem_id": "prob-valid-anagram",
            "box_level": 1,
            "last_reviewed": now_iso,
            "next_review": now_iso,
            "times_correct": 0,
            "total_attempts": 0
        },
        {
            "id": "rev-two-sum",
            "user_id": "sandbox-user-id",
            "problem_id": "prob-two-sum",
            "box_level": 1,
            "last_reviewed": now_iso,
            "next_review": now_iso,
            "times_correct": 0,
            "total_attempts": 0
        }
    ]
    
    for r in default_reviews:
        cols = list(r.keys())
        places = ", ".join(["?"] * len(cols))
        query = f"INSERT OR IGNORE INTO user_reviews ({', '.join(cols)}) VALUES ({places})"
        cursor.execute(query, list(r.values()))

class MockResponse:
    def __init__(self, data):
        self.data = data

class QueryBuilder:
    def __init__(self, table_name, real_client=None):
        self.table_name = table_name
        self.real_client = real_client
        self.filters = []
        self.order_by = []
        self.limit_val = None
        self.op = None
        self.op_data = None
        self.on_conflict = None

    def select(self, columns="*"):
        self.op = 'select'
        self.op_data = columns
        return self

    def insert(self, data):
        self.op = 'insert'
        self.op_data = data
        return self

    def update(self, data):
        self.op = 'update'
        self.op_data = data
        return self

    def upsert(self, data, on_conflict=None):
        self.op = 'upsert'
        self.op_data = data
        self.on_conflict = on_conflict
        return self

    def delete(self):
        self.op = 'delete'
        return self

    def eq(self, column, value):
        self.filters.append(('eq', column, value))
        return self

    def lte(self, column, value):
        self.filters.append(('lte', column, value))
        return self

    def order(self, column, desc=False):
        self.order_by.append((column, desc))
        return self

    def limit(self, count):
        self.limit_val = count
        return self

    def execute(self):
        conn = sqlite3.connect(DB_PATH)
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        
        try:
            if self.op == 'select':
                # Build SELECT query
                # If table is 'problems' and real client is online, try fetching from Supabase first
                if self.table_name == 'problems' and self.real_client:
                    try:
                        # Build public Supabase query
                        supabase_query = self.real_client.table('problems').select('*')
                        for op, col, val in self.filters:
                            if op == 'eq':
                                supabase_query = supabase_query.eq(col, val)
                            elif op == 'lte':
                                supabase_query = supabase_query.lte(col, val)
                        for col, desc in self.order_by:
                            supabase_query = supabase_query.order(col, desc=desc)
                        if self.limit_val is not None:
                            supabase_query = supabase_query.limit(self.limit_val)
                            
                        res = supabase_query.execute()
                        data = res.data if res.data else []
                        
                        # Post-process for nested table relations problems -> user_reviews
                        if self.op_data and 'user_reviews(' in self.op_data:
                            for row in data:
                                prob_id = row.get('id')
                                if prob_id:
                                    cursor.execute("SELECT * FROM user_reviews WHERE problem_id = ? AND user_id = 'sandbox-user-id'", (prob_id,))
                                    rev_rows = cursor.fetchall()
                                    row['user_reviews'] = [dict(r) for r in rev_rows]
                        return MockResponse(data)
                    except Exception as e:
                        # Fallback silently to SQLite
                        pass

                query = f"SELECT * FROM {self.table_name}"
                params = []
                where_clauses = []
                for op, col, val in self.filters:
                    if op == 'eq':
                        where_clauses.append(f"{col} = ?")
                        params.append(val)
                    elif op == 'lte':
                        where_clauses.append(f"{col} <= ?")
                        params.append(val)
                
                if where_clauses:
                    query += " WHERE " + " AND ".join(where_clauses)
                
                # Order by
                if self.order_by:
                    order_clauses = []
                    for col, desc in self.order_by:
                        direction = "DESC" if desc else "ASC"
                        order_clauses.append(f"{col} {direction}")
                    query += " ORDER BY " + ", ".join(order_clauses)
                
                # Limit
                if self.limit_val is not None:
                    query += f" LIMIT {self.limit_val}"
                
                cursor.execute(query, params)
                rows = cursor.fetchall()
                data = [dict(row) for row in rows]
                
                # Post-process for nested table relations
                # 1. user_reviews -> problems
                if self.table_name == 'user_reviews' and self.op_data and 'problems(*)' in self.op_data:
                    for row in data:
                        prob_id = row.get('problem_id')
                        if prob_id:
                            cursor.execute("SELECT * FROM problems WHERE id = ?", (prob_id,))
                            prob_row = cursor.fetchone()
                            if prob_row:
                                row['problems'] = dict(prob_row)
                            else:
                                # Fallback to real Supabase client for public problems
                                if self.real_client:
                                    try:
                                        res_p = self.real_client.table('problems').select('*').eq('id', prob_id).execute()
                                        row['problems'] = res_p.data[0] if res_p.data else None
                                    except Exception:
                                        row['problems'] = None
                                else:
                                    row['problems'] = None
                
                # 2. problems -> user_reviews
                elif self.table_name == 'problems' and self.op_data and 'user_reviews(' in self.op_data:
                    for row in data:
                        prob_id = row.get('id')
                        if prob_id:
                            cursor.execute("SELECT * FROM user_reviews WHERE problem_id = ? AND user_id = 'sandbox-user-id'", (prob_id,))
                            rev_rows = cursor.fetchall()
                            row['user_reviews'] = [dict(r) for r in rev_rows]
                            
                return MockResponse(data)
                
            elif self.op == 'insert':
                data_list = self.op_data if isinstance(self.op_data, list) else [self.op_data]
                inserted_rows = []
                
                for row_data in data_list:
                    # Auto generate UUID if not exists or None
                    if 'id' in row_data and row_data['id'] is None:
                        row_data['id'] = str(uuid.uuid4())
                    elif self.table_name in ('problems', 'user_reviews', 'user_journey_progress', 'challenge_history') and 'id' not in row_data:
                        row_data['id'] = str(uuid.uuid4())
                    
                    columns = list(row_data.keys())
                    placeholders = ", ".join(["?"] * len(columns))
                    query = f"INSERT INTO {self.table_name} ({', '.join(columns)}) VALUES ({placeholders})"
                    cursor.execute(query, list(row_data.values()))
                    inserted_rows.append(row_data)
                
                conn.commit()
                return MockResponse(inserted_rows)
                
            elif self.op == 'update':
                columns = list(self.op_data.keys())
                set_clause = ", ".join([f"{col} = ?" for col in columns])
                params = list(self.op_data.values())
                
                query = f"UPDATE {self.table_name} SET {set_clause}"
                
                where_clauses = []
                where_params = []
                for op, col, val in self.filters:
                    if op == 'eq':
                        where_clauses.append(f"{col} = ?")
                        where_params.append(val)
                    elif op == 'lte':
                        where_clauses.append(f"{col} <= ?")
                        where_params.append(val)
                
                if where_clauses:
                    query += " WHERE " + " AND ".join(where_clauses)
                
                cursor.execute(query, params + where_params)
                conn.commit()
                
                # Fetch and return updated rows
                select_query = f"SELECT * FROM {self.table_name}"
                if where_clauses:
                    select_query += " WHERE " + " AND ".join(where_clauses)
                cursor.execute(select_query, where_params)
                rows = cursor.fetchall()
                return MockResponse([dict(row) for row in rows])
                
            elif self.op == 'upsert':
                row_data = self.op_data if not isinstance(self.op_data, list) else self.op_data[0]
                if self.table_name == 'user_journey_progress':
                    if 'id' not in row_data:
                        row_data['id'] = str(uuid.uuid4())
                    columns = list(row_data.keys())
                    placeholders = ", ".join(["?"] * len(columns))
                    query = f"""
                    INSERT INTO user_journey_progress ({', '.join(columns)}) 
                    VALUES ({placeholders})
                    ON CONFLICT(user_id, problem_number) 
                    DO UPDATE SET viewed_at = excluded.viewed_at
                    """
                    cursor.execute(query, list(row_data.values()))
                else:
                    columns = list(row_data.keys())
                    placeholders = ", ".join(["?"] * len(columns))
                    query = f"INSERT OR REPLACE INTO {self.table_name} ({', '.join(columns)}) VALUES ({placeholders})"
                    cursor.execute(query, list(row_data.values()))
                
                conn.commit()
                return MockResponse([row_data])
                
            elif self.op == 'delete':
                query = f"DELETE FROM {self.table_name}"
                params = []
                where_clauses = []
                for op, col, val in self.filters:
                    if op == 'eq':
                        where_clauses.append(f"{col} = ?")
                        params.append(val)
                
                if where_clauses:
                    query += " WHERE " + " AND ".join(where_clauses)
                
                cursor.execute(query, params)
                conn.commit()
                return MockResponse([])
        finally:
            conn.close()

class MockSupabaseClient:
    def __init__(self):
        self.current_user_id = 'sandbox-user-id'
        # Ensure database and tables exist
        init_db()
        
        # Initialize a real Supabase client for reading public data
        from supabase import create_client
        import os
        import toml
        
        url = os.environ.get("SUPABASE_URL")
        key = os.environ.get("SUPABASE_KEY")
        if not url or not key:
            try:
                secrets_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".streamlit", "secrets.toml")
                if os.path.exists(secrets_path):
                    secrets = toml.load(secrets_path)
                    url = url or secrets.get("SUPABASE_URL")
                    key = key or secrets.get("SUPABASE_KEY")
            except Exception:
                pass
                
        self.real_client = None
        if url and key:
            try:
                self.real_client = create_client(url, key)
            except Exception as e:
                print(f"Error creating real client in sandbox: {e}")
        
    def table(self, table_name):
        return QueryBuilder(table_name, real_client=self.real_client)

def get_sandbox_client():
    return MockSupabaseClient()