DataOps-platform/migrations/versions/20260716_01_baseline.py

"""Create the non-destructive DataOps schema baseline."""

from alembic import op


revision = "20260716_01"
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE IF NOT EXISTS public.users (
            id VARCHAR(100) PRIMARY KEY,
            username VARCHAR(50) UNIQUE NOT NULL,
            password VARCHAR(100) NOT NULL,
            created_at DOUBLE PRECISION NOT NULL,
            last_login DOUBLE PRECISION,
            is_admin BOOLEAN DEFAULT FALSE
        );
        CREATE INDEX IF NOT EXISTS idx_users_username
            ON public.users(username);

        CREATE TABLE IF NOT EXISTS public.data_products (
            id SERIAL PRIMARY KEY,
            product_name VARCHAR(200) NOT NULL,
            product_name_en VARCHAR(200) NOT NULL,
            description TEXT,
            source_dataflow_id INTEGER,
            source_dataflow_name VARCHAR(200),
            target_table VARCHAR(200) NOT NULL,
            target_schema VARCHAR(100) DEFAULT 'public',
            record_count BIGINT DEFAULT 0,
            column_count INTEGER DEFAULT 0,
            last_updated_at TIMESTAMP,
            last_viewed_at TIMESTAMP,
            status VARCHAR(50) DEFAULT 'active',
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            created_by VARCHAR(100) DEFAULT 'system',
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );
        CREATE UNIQUE INDEX IF NOT EXISTS idx_data_products_unique_target
            ON public.data_products(target_schema, target_table);
        CREATE INDEX IF NOT EXISTS idx_data_products_source_dataflow_id
            ON public.data_products(source_dataflow_id);
        CREATE INDEX IF NOT EXISTS idx_data_products_status
            ON public.data_products(status);

        CREATE TABLE IF NOT EXISTS public.data_orders (
            id SERIAL PRIMARY KEY,
            order_no VARCHAR(50) NOT NULL UNIQUE,
            title VARCHAR(200) NOT NULL,
            description TEXT NOT NULL,
            extracted_domains JSONB,
            extracted_fields JSONB,
            extraction_purpose TEXT,
            graph_analysis JSONB,
            can_connect BOOLEAN,
            connection_path JSONB,
            status VARCHAR(50) NOT NULL DEFAULT 'pending',
            reject_reason TEXT,
            result_product_id INTEGER,
            result_dataflow_id INTEGER,
            data_source INTEGER,
            created_by VARCHAR(100) NOT NULL DEFAULT 'user',
            created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
            processed_by VARCHAR(100),
            processed_at TIMESTAMP
        );
        CREATE INDEX IF NOT EXISTS idx_data_orders_status
            ON public.data_orders(status);
        CREATE INDEX IF NOT EXISTS idx_data_orders_created_at
            ON public.data_orders(created_at DESC);

        CREATE TABLE IF NOT EXISTS public.metadata_review_records (
            id BIGSERIAL PRIMARY KEY,
            record_type VARCHAR(20) NOT NULL,
            source VARCHAR(50) NOT NULL DEFAULT 'ddl',
            business_domain_id BIGINT,
            new_meta JSONB NOT NULL,
            candidates JSONB NOT NULL DEFAULT '[]'::jsonb,
            old_meta JSONB,
            status VARCHAR(20) NOT NULL DEFAULT 'pending',
            resolution_action VARCHAR(30),
            resolution_payload JSONB,
            notes TEXT,
            created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
            resolved_at TIMESTAMP,
            resolved_by VARCHAR(100)
        );
        CREATE INDEX IF NOT EXISTS idx_metadata_review_records_status
            ON public.metadata_review_records(status);
        CREATE INDEX IF NOT EXISTS idx_metadata_review_records_created_at
            ON public.metadata_review_records(created_at DESC);

        CREATE TABLE IF NOT EXISTS public.metadata_version_history (
            id BIGSERIAL PRIMARY KEY,
            meta_id BIGINT NOT NULL,
            change_source VARCHAR(50) NOT NULL DEFAULT 'ddl',
            before_snapshot JSONB NOT NULL,
            after_snapshot JSONB NOT NULL,
            created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
            created_by VARCHAR(100)
        );
        CREATE INDEX IF NOT EXISTS idx_metadata_version_history_meta_id
            ON public.metadata_version_history(meta_id);

        CREATE TABLE IF NOT EXISTS public.task_list (
            task_id SERIAL PRIMARY KEY,
            task_name VARCHAR(255) NOT NULL,
            task_description TEXT NOT NULL,
            status VARCHAR(20) NOT NULL DEFAULT 'pending',
            code_name VARCHAR(255),
            code_path VARCHAR(500),
            create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
            update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
            create_by VARCHAR(100) NOT NULL
        );
        CREATE INDEX IF NOT EXISTS idx_task_list_status
            ON public.task_list(status);
        """
    )


def downgrade() -> None:
    # Baseline rollback deliberately preserves tables that may predate Alembic.
    pass