Cloud-Tour-to-Libo/app/data_platform/schema.py

from __future__ import annotations

import hashlib
import re
from typing import Any

from psycopg import sql

from app.config import settings
from app.db import get_conn
from app.data_platform.registry import TABLE_DEFINITIONS, TableDefinition


def _slug(value: str) -> str:
    normalized = re.sub(r"[^a-z0-9]+", "_", value.lower()).strip("_")
    return normalized[:42] or "project"


def project_schema_name(project_id: str) -> str:
    digest = hashlib.sha1(project_id.encode("utf-8")).hexdigest()[:6]
    return f"biz_{_slug(project_id)}_{digest}"


def project_database_name(project_id: str) -> str:
    return f"{_slug(project_id)}_db"


def _column_sql(table: TableDefinition) -> sql.Composed:
    columns: list[sql.Composable] = [
        sql.SQL("id UUID PRIMARY KEY"),
        sql.SQL("tenant_id TEXT NOT NULL"),
        sql.SQL("project_id TEXT NOT NULL"),
    ]
    for item in table.fields:
        parts: list[sql.Composable] = [
            sql.Identifier(item.code),
            sql.SQL(item.sql_type),
        ]
        if item.required:
            parts.append(sql.SQL("NOT NULL"))
        if item.default_sql:
            parts.extend((sql.SQL("DEFAULT"), sql.SQL(item.default_sql)))
        columns.append(sql.SQL(" ").join(parts))
    columns.extend(
        (
            sql.SQL("created_at TIMESTAMPTZ NOT NULL DEFAULT now()"),
            sql.SQL("updated_at TIMESTAMPTZ NOT NULL DEFAULT now()"),
            sql.SQL("deleted_at TIMESTAMPTZ"),
            sql.SQL("deleted_by TEXT"),
        )
    )
    return sql.SQL(", ").join(columns)


async def ensure_platform_registry() -> None:
    admin_schema = sql.Identifier(settings.db_schema)
    async with get_conn() as conn:
        async with conn.cursor() as cur:
            await cur.execute(
                sql.SQL(
                    """
                    CREATE TABLE IF NOT EXISTS {}.project_databases (
                        project_id TEXT PRIMARY KEY,
                        tenant_id TEXT NOT NULL,
                        display_name TEXT NOT NULL,
                        database_name TEXT NOT NULL UNIQUE,
                        schema_name TEXT NOT NULL UNIQUE,
                        status TEXT NOT NULL DEFAULT 'ready',
                        created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
                        updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
                    )
                    """
                ).format(admin_schema)
            )
        await conn.commit()


async def ensure_project_database(
    project_id: str,
    tenant_id: str,
    display_name: str,
) -> dict[str, Any]:
    await ensure_platform_registry()
    schema_name = project_schema_name(project_id)
    database_name = project_database_name(project_id)
    admin_schema = sql.Identifier(settings.db_schema)
    project_schema = sql.Identifier(schema_name)

    async with get_conn() as conn:
        async with conn.cursor() as cur:
            await cur.execute(
                sql.SQL(
                    """
                    INSERT INTO {}.project_databases (
                        project_id, tenant_id, display_name, database_name, schema_name, status, updated_at
                    )
                    VALUES (%s, %s, %s, %s, %s, 'ready', now())
                    ON CONFLICT (project_id) DO UPDATE
                    SET tenant_id=EXCLUDED.tenant_id,
                        display_name=EXCLUDED.display_name,
                        database_name=EXCLUDED.database_name,
                        schema_name=EXCLUDED.schema_name,
                        status='ready',
                        updated_at=now()
                    RETURNING *
                    """
                ).format(admin_schema),
                (project_id, tenant_id, display_name, database_name, schema_name),
            )
            database_row = await cur.fetchone()
            await cur.execute(sql.SQL("CREATE SCHEMA IF NOT EXISTS {}").format(project_schema))

            for table in TABLE_DEFINITIONS:
                await cur.execute(
                    sql.SQL("CREATE TABLE IF NOT EXISTS {}.{} ({})").format(
                        project_schema,
                        sql.Identifier(table.code),
                        _column_sql(table),
                    )
                )
                await cur.execute(
                    sql.SQL(
                        "CREATE INDEX IF NOT EXISTS {} ON {}.{} (updated_at DESC)"
                    ).format(
                        sql.Identifier(f"{table.code}_updated_idx"),
                        project_schema,
                        sql.Identifier(table.code),
                    )
                )
                await cur.execute(
                    sql.SQL(
                        "CREATE INDEX IF NOT EXISTS {} ON {}.{} (project_id) WHERE deleted_at IS NULL"
                    ).format(
                        sql.Identifier(f"{table.code}_project_idx"),
                        project_schema,
                        sql.Identifier(table.code),
                    )
                )
            await cur.execute(
                sql.SQL(
                    """
                    CREATE TABLE IF NOT EXISTS {}.data_change_logs (
                        id BIGSERIAL PRIMARY KEY,
                        tenant_id TEXT NOT NULL,
                        project_id TEXT NOT NULL,
                        table_code TEXT NOT NULL,
                        record_id UUID NOT NULL,
                        operation TEXT NOT NULL,
                        before_data JSONB,
                        after_data JSONB,
                        actor TEXT NOT NULL,
                        created_at TIMESTAMPTZ NOT NULL DEFAULT now()
                    )
                    """
                ).format(project_schema)
            )
            await cur.execute(
                sql.SQL(
                    """
                    CREATE TABLE IF NOT EXISTS {}.graph_sync_queue (
                        id BIGSERIAL PRIMARY KEY,
                        tenant_id TEXT NOT NULL,
                        project_id TEXT NOT NULL,
                        table_code TEXT NOT NULL,
                        record_id UUID NOT NULL,
                        operation TEXT NOT NULL,
                        record_version INTEGER NOT NULL DEFAULT 1,
                        status TEXT NOT NULL DEFAULT 'pending',
                        retry_count INTEGER NOT NULL DEFAULT 0,
                        error_message TEXT,
                        created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
                        completed_at TIMESTAMPTZ
                    )
                    """
                ).format(project_schema)
            )
        await conn.commit()
    return dict(database_row)


async def ensure_all_project_databases() -> None:
    await ensure_platform_registry()
    async with get_conn() as conn:
        async with conn.cursor() as cur:
            await cur.execute(
                sql.SQL(
                    """
                    SELECT project_id, tenant_id, display_name
                    FROM {}.projects
                    WHERE status <> 'archived'
                    ORDER BY created_at
                    """
                ).format(sql.Identifier(settings.db_schema))
            )
            projects = await cur.fetchall()
    for project in projects:
        await ensure_project_database(
            str(project["project_id"]),
            str(project["tenant_id"]),
            str(project["display_name"]),
        )


async def get_project_database(project_id: str) -> dict[str, Any] | None:
    await ensure_platform_registry()
    async with get_conn() as conn:
        async with conn.cursor() as cur:
            await cur.execute(
                sql.SQL("SELECT * FROM {}.project_databases WHERE project_id=%s").format(
                    sql.Identifier(settings.db_schema)
                ),
                (project_id,),
            )
            row = await cur.fetchone()
            return dict(row) if row else None