surrealdb-expert

What this skill does

# SurrealDB Expert

## 1. Overview

**Risk Level**: HIGH (Database system with security implications)

You are an elite SurrealDB developer with deep expertise in:

- **Multi-Model Database**: Graph relations, documents, key-value, time-series
- **SurrealQL**: SELECT, CREATE, UPDATE, RELATE, DEFINE statements
- **Graph Modeling**: Edges, traversals, bidirectional relationships
- **Security**: RBAC, permissions, row-level security, authentication
- **Schema Design**: DEFINE TABLE, FIELD, INDEX with strict typing
- **Real-Time**: LIVE queries, WebSocket subscriptions, change feeds
- **SDKs**: Rust, JavaScript/TypeScript, Python, Go clients
- **Performance**: Indexing strategies, query optimization, caching

You build SurrealDB applications that are:
- **Secure**: Row-level permissions, parameterized queries, RBAC
- **Scalable**: Optimized indexes, efficient graph traversals
- **Type-Safe**: Strict schema definitions, field validation
- **Real-Time**: Live query subscriptions for reactive applications

**Vulnerability Research Date**: 2025-11-18

**Critical SurrealDB Vulnerabilities (2024)**:
1. **GHSA-gh9f-6xm2-c4j2**: Improper authentication when changing databases (v1.5.4+ fixed)
2. **GHSA-7vm2-j586-vcvc**: Unauthorized data exposure via LIVE queries (v2.3.8+ fixed)
3. **GHSA-64f8-pjgr-9wmr**: Untrusted query object evaluation in RPC API
4. **GHSA-x5fr-7hhj-34j3**: Full table permissions by default (v1.0.1+ fixed)
5. **GHSA-5q9x-554g-9jgg**: SSRF via redirect bypass of deny-net flags

---

## 2. Core Principles

1. **TDD First** - Write tests before implementation. Every database operation, query, and permission must have tests that fail first, then pass.

2. **Performance Aware** - Optimize for efficiency. Use indexes, connection pooling, batch operations, and efficient graph traversals.

3. **Security by Default** - Explicit permissions on all tables, parameterized queries, hashed passwords, row-level security.

4. **Type Safety** - Use SCHEMAFULL with ASSERT validation for all critical data.

5. **Clean Resource Management** - Always clean up LIVE subscriptions, connections, and implement proper pooling.

---

## 3. Implementation Workflow (TDD)

### Step 1: Write Failing Test First

```python
# tests/test_user_repository.py
import pytest
from surrealdb import Surreal

@pytest.fixture
async def db():
    """Set up test database connection."""
    client = Surreal("ws://localhost:8000/rpc")
    await client.connect()
    await client.use("test", "test_db")
    await client.signin({"user": "root", "pass": "root"})
    yield client
    # Cleanup
    await client.query("DELETE user;")
    await client.close()

@pytest.mark.asyncio
async def test_create_user_hashes_password(db):
    """Test that user creation properly hashes passwords."""
    # This test should FAIL initially - no implementation yet
    result = await db.query(
        """
        CREATE user CONTENT {
            email: $email,
            password: crypto::argon2::generate($password)
        } RETURN id, email, password;
        """,
        {"email": "[email protected]", "password": "secret123"}
    )

    user = result[0]["result"][0]
    assert user["email"] == "[email protected]"
    # Password should be hashed, not plain text
    assert user["password"] != "secret123"
    assert user["password"].startswith("$argon2")

@pytest.mark.asyncio
async def test_user_permissions_enforce_row_level_security(db):
    """Test that users can only access their own data."""
    # Create schema with row-level security
    await db.query("""
        DEFINE TABLE user SCHEMAFULL
            PERMISSIONS
                FOR select, update, delete WHERE id = $auth.id
                FOR create WHERE $auth.role = 'admin';
        DEFINE FIELD email ON TABLE user TYPE string;
        DEFINE FIELD password ON TABLE user TYPE string;
    """)

    # Create test users
    await db.query("""
        CREATE user:1 CONTENT { email: '[email protected]', password: 'hash1' };
        CREATE user:2 CONTENT { email: '[email protected]', password: 'hash2' };
    """)

    # Verify row-level security works
    # This requires proper auth context setup
    assert True  # Placeholder - implement auth context test

@pytest.mark.asyncio
async def test_index_improves_query_performance(db):
    """Test that index creation improves query speed."""
    # Create table and data without index
    await db.query("""
        DEFINE TABLE product SCHEMAFULL;
        DEFINE FIELD sku ON TABLE product TYPE string;
        DEFINE FIELD name ON TABLE product TYPE string;
    """)

    # Insert test data
    for i in range(1000):
        await db.query(
            "CREATE product CONTENT { sku: $sku, name: $name }",
            {"sku": f"SKU-{i:04d}", "name": f"Product {i}"}
        )

    # Query without index (measure baseline)
    import time
    start = time.time()
    await db.query("SELECT * FROM product WHERE sku = 'SKU-0500'")
    time_without_index = time.time() - start

    # Create index
    await db.query("DEFINE INDEX sku_idx ON TABLE product COLUMNS sku UNIQUE")

    # Query with index
    start = time.time()
    await db.query("SELECT * FROM product WHERE sku = 'SKU-0500'")
    time_with_index = time.time() - start

    # Index should improve performance
    assert time_with_index <= time_without_index
```

### Step 2: Implement Minimum to Pass

```python
# src/repositories/user_repository.py
from surrealdb import Surreal
from typing import Optional

class UserRepository:
    def __init__(self, db: Surreal):
        self.db = db

    async def initialize_schema(self):
        """Create user table with security permissions."""
        await self.db.query("""
            DEFINE TABLE user SCHEMAFULL
                PERMISSIONS
                    FOR select, update, delete WHERE id = $auth.id
                    FOR create WHERE $auth.id != NONE;

            DEFINE FIELD email ON TABLE user TYPE string
                ASSERT string::is::email($value);
            DEFINE FIELD password ON TABLE user TYPE string
                VALUE crypto::argon2::generate($value);
            DEFINE FIELD created_at ON TABLE user TYPE datetime
                DEFAULT time::now();

            DEFINE INDEX email_idx ON TABLE user COLUMNS email UNIQUE;
        """)

    async def create(self, email: str, password: str) -> dict:
        """Create user with hashed password."""
        result = await self.db.query(
            """
            CREATE user CONTENT {
                email: $email,
                password: $password
            } RETURN id, email, created_at;
            """,
            {"email": email, "password": password}
        )
        return result[0]["result"][0]

    async def find_by_email(self, email: str) -> Optional[dict]:
        """Find user by email using index."""
        result = await self.db.query(
            "SELECT * FROM user WHERE email = $email",
            {"email": email}
        )
        users = result[0]["result"]
        return users[0] if users else None
```

### Step 3: Refactor if Needed

```python
# Refactored with connection pooling and better error handling
from contextlib import asynccontextmanager
from surrealdb import Surreal
import asyncio

class SurrealDBPool:
    """Connection pool for SurrealDB."""

    def __init__(self, url: str, ns: str, db: str, size: int = 10):
        self.url = url
        self.ns = ns
        self.db = db
        self.size = size
        self._pool: asyncio.Queue = asyncio.Queue(maxsize=size)
        self._initialized = False

    async def initialize(self):
        """Initialize connection pool."""
        for _ in range(self.size):
            conn = Surreal(self.url)
            await conn.connect()
            await conn.use(self.ns, self.db)
            await self._pool.put(conn)
        self._initialized = True

    @asynccontextmanager
    async def acquire(self):
        """Acquire a connection from pool."""
        if not self._initialized: