routstr-core/docs/contributing/database.md

Database Guide

This guide covers database design, migrations, and best practices for Routstr Core.

Database Overview

Routstr uses:

• SQLite for local development and single-node deployments
• PostgreSQL (optional) for production scale
• SQLModel for ORM with type safety
• Alembic for schema migrations
• Async SQLAlchemy for non-blocking I/O

Schema Design

Core Tables

APIKey Table

class APIKey(SQLModel, table=True):
    """API key with balance tracking"""
    
    # Primary key
    id: int | None = Field(default=None, primary_key=True)
    
    # Key data (indexed for fast lookups)
    key_hash: str = Field(index=True, unique=True)
    
    # Balance in millisatoshis (1 sat = 1000 msats)
    balance: int = Field(default=0)
    total_deposited: int = Field(default=0)
    total_spent: int = Field(default=0)
    
    # Timestamps
    created_at: datetime = Field(default_factory=datetime.utcnow)
    last_used_at: datetime | None = None
    expires_at: datetime | None = None
    
    # Metadata (JSON field)
    metadata: dict = Field(
        default_factory=dict, 
        sa_column=Column(JSON)
    )
    
    # Relationships
    transactions: list["Transaction"] = Relationship(back_populates="api_key")

Transaction Table

class Transaction(SQLModel, table=True):
    """Transaction log for audit trail"""
    
    # Primary key
    id: int | None = Field(default=None, primary_key=True)
    
    # Foreign key to API key
    api_key_id: int = Field(foreign_key="apikey.id", index=True)
    
    # Transaction details
    amount: int  # Can be negative for deductions
    balance_after: int  # Balance snapshot
    type: TransactionType  # Enum: deposit, usage, withdrawal
    description: str
    
    # Timestamp (indexed for range queries)
    timestamp: datetime = Field(
        default_factory=datetime.utcnow,
        index=True
    )
    
    # Request details (optional)
    request_data: dict | None = Field(
        default=None,
        sa_column=Column(JSON)
    )
    
    # Relationships
    api_key: APIKey = Relationship(back_populates="transactions")

Withdrawal Table

class Withdrawal(SQLModel, table=True):
    """Track withdrawal requests"""
    
    id: int | None = Field(default=None, primary_key=True)
    api_key_id: int = Field(foreign_key="apikey.id")
    
    # Withdrawal details
    amount: int
    token: str  # Encrypted eCash token
    mint_url: str
    
    # Status tracking
    status: WithdrawalStatus  # pending, completed, failed
    created_at: datetime = Field(default_factory=datetime.utcnow)
    completed_at: datetime | None = None
    
    # Error handling
    error_message: str | None = None
    retry_count: int = Field(default=0)

Indexes

Critical indexes for performance:

# In models
key_hash: str = Field(index=True, unique=True)  # Fast key lookup
timestamp: datetime = Field(index=True)  # Range queries

# Composite indexes (in migrations)
Index('idx_transactions_key_time', 'api_key_id', 'timestamp')
Index('idx_apikey_expires', 'expires_at').where(expires_at.isnot(None))

Migrations

Creating Migrations

Auto-generate from Model Changes

# After modifying SQLModel classes
make db-migrate

# Enter descriptive message
> Add withdrawal status field

Manual Migration

# Create empty migration
alembic revision -m "custom migration"

# Edit the generated file

Migration Template

"""Add withdrawal status field

Revision ID: abc123
Revises: def456
Create Date: 2024-01-01 12:00:00

"""
from alembic import op
import sqlalchemy as sa
import sqlmodel

# revision identifiers
revision = 'abc123'
down_revision = 'def456'

def upgrade() -> None:
    """Apply migration"""
    op.add_column(
        'withdrawal',
        sa.Column(
            'status',
            sa.String(),
            nullable=False,
            server_default='pending'
        )
    )
    
    # Add index
    op.create_index(
        'idx_withdrawal_status',
        'withdrawal',
        ['status']
    )

def downgrade() -> None:
    """Revert migration"""
    op.drop_index('idx_withdrawal_status', 'withdrawal')
    op.drop_column('withdrawal', 'status')

Running Migrations

Development

# Apply all migrations
make db-upgrade

# Check current version
make db-current

# Rollback one version
make db-downgrade

# View history
make db-history

Production

Migrations run automatically on startup:

# In routstr/core/main.py
def run_migrations():
    """Run database migrations on startup"""
    from alembic import command
    from alembic.config import Config
    
    alembic_cfg = Config("alembic.ini")
    command.upgrade(alembic_cfg, "head")

Migration Best Practices

1. Always Review Generated Migrations

   • Check for data loss
   • Verify index creation
   • Test rollback

2. Handle Data Migrations

   def upgrade():
       # Schema change
       op.add_column('apikey', sa.Column('status', sa.String()))
   
       # Data migration
       connection = op.get_bind()
       connection.execute(
           "UPDATE apikey SET status = 'active' WHERE expires_at IS NULL"
       )
   

3. Make Migrations Idempotent

   def upgrade():
       # Check if column exists
       inspector = sa.inspect(op.get_bind())
       columns = [col['name'] for col in inspector.get_columns('apikey')]
   
       if 'new_field' not in columns:
           op.add_column('apikey', sa.Column('new_field', sa.String()))
   

Database Operations

Connection Management

# Database session factory
async_session = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# Dependency injection
async def get_db() -> AsyncSession:
    async with async_session() as session:
        yield session

Query Patterns

Basic Queries

# Get by primary key
api_key = await session.get(APIKey, key_id)

# Get by unique field
result = await session.execute(
    select(APIKey).where(APIKey.key_hash == hash_value)
)
api_key = result.scalar_one_or_none()

# Get multiple with filter
result = await session.execute(
    select(APIKey)
    .where(APIKey.balance > 0)
    .where(APIKey.expires_at > datetime.utcnow())
)
active_keys = result.scalars().all()

Joins and Relationships

# Eager loading
result = await session.execute(
    select(APIKey)
    .options(selectinload(APIKey.transactions))
    .where(APIKey.id == key_id)
)
api_key = result.scalar_one()

# Join query
result = await session.execute(
    select(Transaction)
    .join(APIKey)
    .where(APIKey.key_hash == hash_value)
    .order_by(Transaction.timestamp.desc())
    .limit(10)
)
recent_transactions = result.scalars().all()

Aggregations

# Sum total spent
result = await session.execute(
    select(func.sum(Transaction.amount))
    .where(Transaction.api_key_id == key_id)
    .where(Transaction.type == TransactionType.USAGE)
)
total_spent = result.scalar() or 0

# Count active keys
result = await session.execute(
    select(func.count(APIKey.id))
    .where(APIKey.balance > 0)
)
active_count = result.scalar()

Transactions

Atomic Operations

async def transfer_balance(
    session: AsyncSession,
    from_key: int,
    to_key: int,
    amount: int
):
    """Atomic balance transfer"""
    async with session.begin():
        # Lock rows to prevent race conditions
        from_api_key = await session.execute(
            select(APIKey)
            .where(APIKey.id == from_key)
            .with_for_update()
        )
        from_api_key = from_api_key.scalar_one()
        
        to_api_key = await session.execute(
            select(APIKey)
            .where(APIKey.id == to_key)
            .with_for_update()
        )
        to_api_key = to_api_key.scalar_one()
        
        # Check balance
        if from_api_key.balance < amount:
            raise InsufficientBalanceError()
        
        # Update balances
        from_api_key.balance -= amount
        to_api_key.balance += amount
        
        # Log transactions
        session.add(Transaction(
            api_key_id=from_key,
            amount=-amount,
            balance_after=from_api_key.balance,
            type=TransactionType.TRANSFER_OUT
        ))
        
        session.add(Transaction(
            api_key_id=to_key,
            amount=amount,
            balance_after=to_api_key.balance,
            type=TransactionType.TRANSFER_IN
        ))
        
        # Commit happens automatically

Optimistic Locking

class APIKey(SQLModel, table=True):
    # Add version field
    version: int = Field(default=1)

async def update_with_version_check(
    session: AsyncSession,
    api_key: APIKey,
    new_balance: int
):
    """Update with optimistic locking"""
    result = await session.execute(
        update(APIKey)
        .where(APIKey.id == api_key.id)
        .where(APIKey.version == api_key.version)
        .values(
            balance=new_balance,
            version=APIKey.version + 1
        )
    )
    
    if result.rowcount == 0:
        raise ConcurrentModificationError()

Performance Optimization

Query Optimization

1. Use Indexes Effectively

   # Good: Uses index
   where(APIKey.key_hash == value)
   
   # Bad: Function prevents index use
   where(func.lower(APIKey.key_hash) == value.lower())
   

2. Limit Results

   # Always limit when possible
   query.limit(100)
   
   # Use pagination
   query.offset(page * page_size).limit(page_size)
   

3. Select Only Needed Columns

   # Select specific columns
   result = await session.execute(
       select(APIKey.id, APIKey.balance)
       .where(APIKey.key_hash == hash_value)
   )
   

Connection Pooling

# Configure connection pool
engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,          # Number of connections
    max_overflow=10,       # Extra connections when needed
    pool_timeout=30,       # Wait time for connection
    pool_recycle=3600,     # Recycle connections after 1 hour
    pool_pre_ping=True,    # Check connection health
)

Batch Operations

# Batch insert
async def bulk_create_transactions(
    session: AsyncSession,
    transactions: list[dict]
):
    """Efficient bulk insert"""
    await session.execute(
        insert(Transaction),
        transactions
    )
    await session.commit()

# Batch update
await session.execute(
    update(APIKey)
    .where(APIKey.expires_at < datetime.utcnow())
    .values(active=False)
)

Testing Database Code

Test Database Setup

@pytest.fixture
async def test_engine():
    """Create test database engine"""
    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        echo=True  # Log SQL for debugging
    )
    
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    
    yield engine
    
    await engine.dispose()

Testing Queries

async def test_balance_update(test_session):
    """Test atomic balance update"""
    # Create test data
    api_key = APIKey(key_hash="test", balance=1000)
    test_session.add(api_key)
    await test_session.commit()
    
    # Test update
    await deduct_balance(test_session, api_key.id, 100)
    
    # Verify
    await test_session.refresh(api_key)
    assert api_key.balance == 900
    
    # Verify transaction log
    result = await test_session.execute(
        select(Transaction)
        .where(Transaction.api_key_id == api_key.id)
    )
    transactions = result.scalars().all()
    assert len(transactions) == 1
    assert transactions[0].amount == -100

Testing Migrations

def test_migration_upgrade():
    """Test migration applies correctly"""
    # Create database at previous version
    alembic_cfg = Config("alembic.ini")
    command.downgrade(alembic_cfg, "-1")
    
    # Apply migration
    command.upgrade(alembic_cfg, "+1")
    
    # Verify schema changes
    engine = create_engine(DATABASE_URL)
    inspector = inspect(engine)
    columns = [col['name'] for col in inspector.get_columns('apikey')]
    assert 'new_column' in columns

Database Maintenance

Monitoring Queries

-- Slow queries (SQLite)
EXPLAIN QUERY PLAN
SELECT * FROM apikey WHERE balance > 0;

-- Table sizes
SELECT 
    name,
    COUNT(*) as row_count
FROM sqlite_master
WHERE type='table'
GROUP BY name;

-- Index usage
SELECT * FROM sqlite_stat1;

Cleanup Tasks

async def cleanup_expired_keys(session: AsyncSession):
    """Remove expired API keys"""
    result = await session.execute(
        delete(APIKey)
        .where(APIKey.expires_at < datetime.utcnow())
        .where(APIKey.balance == 0)
    )
    
    logger.info(f"Cleaned up {result.rowcount} expired keys")
    await session.commit()

async def vacuum_database(session: AsyncSession):
    """Optimize database file size (SQLite)"""
    await session.execute(text("VACUUM"))

Backup Strategies

async def backup_database(source_url: str, backup_path: str):
    """Create database backup"""
    if "sqlite" in source_url:
        # SQLite backup
        import shutil
        db_path = source_url.split("///")[1]
        shutil.copy2(db_path, backup_path)
    else:
        # PostgreSQL backup
        import subprocess
        subprocess.run([
            "pg_dump",
            source_url,
            "-f", backup_path
        ])

PostgreSQL Migration

Configuration

# PostgreSQL connection
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/routstr"

# Additional PostgreSQL-specific settings
engine = create_async_engine(
    DATABASE_URL,
    server_settings={
        "jit": "off",
        "statement_timeout": "30s"
    }
)

PostgreSQL-Specific Features

# Use PostgreSQL arrays
from sqlalchemy.dialects.postgresql import ARRAY

class APIKey(SQLModel, table=True):
    allowed_models: list[str] = Field(
        default_factory=list,
        sa_column=Column(ARRAY(String))
    )

# Use JSONB for better performance
metadata: dict = Field(
    default_factory=dict,
    sa_column=Column(JSONB)
)

# Full-text search
from sqlalchemy.dialects.postgresql import TSVECTOR

search_vector = Column(TSVECTOR)

Security Considerations

SQL Injection Prevention

# Always use parameterized queries
# Good
await session.execute(
    select(APIKey).where(APIKey.key_hash == key_hash)
)

# Bad - SQL injection risk
await session.execute(
    text(f"SELECT * FROM apikey WHERE key_hash = '{key_hash}'")
)

Data Encryption

from cryptography.fernet import Fernet

class EncryptedField(TypeDecorator):
    """Encrypt sensitive data at rest"""
    impl = String
    
    def __init__(self, key: bytes, *args, **kwargs):
        self.cipher = Fernet(key)
        super().__init__(*args, **kwargs)
    
    def process_bind_param(self, value, dialect):
        if value is not None:
            return self.cipher.encrypt(value.encode()).decode()
        return value
    
    def process_result_value(self, value, dialect):
        if value is not None:
            return self.cipher.decrypt(value.encode()).decode()
        return value

Next Steps

• Review Testing Guide for database testing
• Check Guidelines for code standards
• See Architecture for system design
• Read Setup Guide for development setup