StarPunk/starpunk/database.py

"""
Database initialization and operations for StarPunk
SQLite database for metadata, sessions, and tokens
"""

import sqlite3
from pathlib import Path


# Database schema
SCHEMA_SQL = """
-- Notes metadata (content is in files)
CREATE TABLE IF NOT EXISTS notes (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    slug TEXT UNIQUE NOT NULL,
    file_path TEXT UNIQUE NOT NULL,
    published BOOLEAN DEFAULT 0,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    deleted_at TIMESTAMP,
    content_hash TEXT
);

CREATE INDEX IF NOT EXISTS idx_notes_created_at ON notes(created_at DESC);
CREATE INDEX IF NOT EXISTS idx_notes_published ON notes(published);
CREATE INDEX IF NOT EXISTS idx_notes_slug ON notes(slug);
CREATE INDEX IF NOT EXISTS idx_notes_deleted_at ON notes(deleted_at);

-- Authentication sessions (IndieLogin)
CREATE TABLE IF NOT EXISTS sessions (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_token_hash TEXT UNIQUE NOT NULL,
    me TEXT NOT NULL,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    expires_at TIMESTAMP NOT NULL,
    last_used_at TIMESTAMP,
    user_agent TEXT,
    ip_address TEXT
);

CREATE INDEX IF NOT EXISTS idx_sessions_token_hash ON sessions(session_token_hash);
CREATE INDEX IF NOT EXISTS idx_sessions_expires ON sessions(expires_at);
CREATE INDEX IF NOT EXISTS idx_sessions_me ON sessions(me);

-- Micropub access tokens (secure storage with hashed tokens)
CREATE TABLE IF NOT EXISTS tokens (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    token_hash TEXT UNIQUE NOT NULL,
    me TEXT NOT NULL,
    client_id TEXT,
    scope TEXT DEFAULT 'create',
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    expires_at TIMESTAMP NOT NULL,
    last_used_at TIMESTAMP,
    revoked_at TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_tokens_hash ON tokens(token_hash);
CREATE INDEX IF NOT EXISTS idx_tokens_me ON tokens(me);
CREATE INDEX IF NOT EXISTS idx_tokens_expires ON tokens(expires_at);

-- Authorization codes for IndieAuth token exchange
CREATE TABLE IF NOT EXISTS authorization_codes (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    code_hash TEXT UNIQUE NOT NULL,
    me TEXT NOT NULL,
    client_id TEXT NOT NULL,
    redirect_uri TEXT NOT NULL,
    scope TEXT,
    state TEXT,
    code_challenge TEXT,
    code_challenge_method TEXT,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    expires_at TIMESTAMP NOT NULL,
    used_at TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_auth_codes_hash ON authorization_codes(code_hash);
CREATE INDEX IF NOT EXISTS idx_auth_codes_expires ON authorization_codes(expires_at);

-- CSRF state tokens (for IndieAuth flow)
CREATE TABLE IF NOT EXISTS auth_state (
    state TEXT PRIMARY KEY,
    code_verifier TEXT NOT NULL DEFAULT '',
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    expires_at TIMESTAMP NOT NULL,
    redirect_uri TEXT
);

CREATE INDEX IF NOT EXISTS idx_auth_state_expires ON auth_state(expires_at);
"""


def init_db(app=None):
    """
    Initialize database schema and run migrations

    Args:
        app: Flask application instance (optional, for config access)
    """
    if app:
        db_path = app.config["DATABASE_PATH"]
        logger = app.logger
    else:
        # Fallback to default path
        db_path = Path("./data/starpunk.db")
        logger = None

    # Ensure parent directory exists
    db_path.parent.mkdir(parents=True, exist_ok=True)

    # Create database and initial schema
    conn = sqlite3.connect(db_path)
    try:
        conn.executescript(SCHEMA_SQL)
        conn.commit()
        if logger:
            logger.info(f"Database initialized: {db_path}")
        else:
            print(f"Database initialized: {db_path}")
    finally:
        conn.close()

    # Run migrations
    from starpunk.migrations import run_migrations
    run_migrations(db_path, logger=logger)


def get_db(app):
    """
    Get database connection

    Args:
        app: Flask application instance

    Returns:
        sqlite3.Connection
    """
    db_path = app.config["DATABASE_PATH"]
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row  # Return rows as dictionaries
    return conn