-- DEPosit: Pipeline artifact tables
-- Applied automatically by collect_pipeline_artifacts.py on first run.
-- Manual init: sqlite3 Data/DE_pipeline_artifacts.db < Scripts/pipeline_schema.sql

-- ── Collection progress (including repos with 0 matched artifacts) ─────────
CREATE TABLE IF NOT EXISTS repo_scan_status (
    repo_full_name       TEXT PRIMARY KEY,
    artifact_file_count  INTEGER NOT NULL,
    tree_status          TEXT,               -- ok | 404 | empty | truncated | http_*
    scanned_at           TEXT NOT NULL
);

-- ── Discovery ──────────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS pipeline_files (
    id                  INTEGER PRIMARY KEY AUTOINCREMENT,
    repo_full_name      TEXT    NOT NULL,
    file_path           TEXT    NOT NULL,
    artifact_type       TEXT    NOT NULL,   -- airflow_dag | dbt_model | dbt_schema |
                                            -- prefect_flow | dagster | luigi |
                                            -- kedro | beam | dlt | mage_block
    file_sha            TEXT,               -- git blob SHA
    file_size_bytes     INTEGER,
    collected_at        TEXT    NOT NULL,
    parse_error         TEXT,               -- NULL when parsing succeeded
    UNIQUE(repo_full_name, file_path)
);

-- ── Airflow ─────────────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS airflow_dag_features (
    id                          INTEGER PRIMARY KEY AUTOINCREMENT,
    repo_full_name              TEXT    NOT NULL,
    file_path                   TEXT    NOT NULL,
    dag_id                      TEXT,
    schedule_interval           TEXT,       -- cron expr, timedelta repr, preset, or NULL
    schedule_type               TEXT,       -- cron | timedelta | preset | none | dynamic
    catchup                     INTEGER,    -- 0/1/NULL (NULL = not set, inherits default)
    max_active_runs             INTEGER,
    default_retries             INTEGER,
    default_retry_delay_seconds INTEGER,
    task_count                  INTEGER,
    edge_count                  INTEGER,    -- explicit dependency edges
    unique_operator_count       INTEGER,
    operator_types              TEXT,       -- JSON array of operator class names
    has_python_operator         INTEGER,    -- 0/1
    has_bash_operator           INTEGER,
    has_external_task_sensor    INTEGER,    -- cross-DAG coupling
    has_branch_operator         INTEGER,
    has_trigger_rule            INTEGER,    -- non-default trigger rules
    uses_xcom                   INTEGER,    -- xcom_push/xcom_pull detected
    uses_pools                  INTEGER,
    uses_connections            INTEGER,    -- conn_id parameter present
    has_sla                     INTEGER,    -- sla= or sla_miss_callback
    has_on_failure_callback     INTEGER,
    has_on_success_callback     INTEGER,
    dag_tags                    TEXT,       -- JSON array
    -- Anti-pattern flags (directly usable as SE research features)
    antipattern_catchup_no_maxruns  INTEGER, -- catchup=True with no max_active_runs
    antipattern_no_retries          INTEGER, -- no retry config anywhere in DAG
    antipattern_no_failure_cb       INTEGER, -- no failure callback
    antipattern_no_schedule         INTEGER, -- schedule_interval=None (orphaned DAG)
    antipattern_bare_xcom           INTEGER, -- xcom used for large data transfer
    file_lines                  INTEGER,
    FOREIGN KEY(repo_full_name, file_path) REFERENCES pipeline_files(repo_full_name, file_path)
);

-- ── dbt ─────────────────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS dbt_model_features (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT,
    repo_full_name          TEXT    NOT NULL,
    file_path               TEXT    NOT NULL,
    model_name              TEXT    NOT NULL,
    model_layer             TEXT,           -- staging | intermediate | marts | other
    materialization         TEXT,           -- view | table | incremental | ephemeral | NULL
    ref_count               INTEGER,        -- upstream dbt model dependencies
    source_count            INTEGER,        -- raw source references
    cte_count               INTEGER,        -- WITH clause count
    has_where_clause        INTEGER,
    has_join                INTEGER,
    has_window_function     INTEGER,
    uses_incremental        INTEGER,
    has_unique_key          INTEGER,        -- required for safe incremental models
    hardcoded_limit         INTEGER,        -- LIMIT with literal integer (smell)
    select_star             INTEGER,        -- SELECT * (smell)
    has_tests               INTEGER,        -- set from schema.yml via extract_schema
    file_lines              INTEGER,
    -- Anti-pattern flags
    antipattern_incremental_no_unique_key   INTEGER,
    antipattern_select_star                 INTEGER,
    antipattern_no_source_no_ref            INTEGER, -- raw table name in FROM (hardcoded)
    FOREIGN KEY(repo_full_name, file_path) REFERENCES pipeline_files(repo_full_name, file_path)
);

CREATE TABLE IF NOT EXISTS dbt_project_summary (
    repo_full_name          TEXT    PRIMARY KEY,
    dbt_version_required    TEXT,
    model_paths             TEXT,           -- JSON array from dbt_project.yml
    has_tests_dir           INTEGER,
    has_snapshots           INTEGER,
    has_seeds               INTEGER,
    has_analyses            INTEGER,
    profile_outputs         TEXT,           -- JSON: target names from profiles.yml if present
    total_model_files       INTEGER,
    total_schema_files      INTEGER,
    FOREIGN KEY(repo_full_name) REFERENCES pipeline_files(repo_full_name)
);

-- ── Prefect ──────────────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS prefect_flow_features (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT,
    repo_full_name          TEXT    NOT NULL,
    file_path               TEXT    NOT NULL,
    prefect_version         TEXT,           -- 1.x | 2.x inferred from imports
    flow_count              INTEGER,
    task_count              INTEGER,
    has_retries             INTEGER,
    max_retries             INTEGER,
    has_schedule            INTEGER,
    has_timeout             INTEGER,
    uses_result_caching     INTEGER,
    uses_state_handler      INTEGER,
    uses_mapped_tasks       INTEGER,        -- .map() or unmapped() pattern
    has_deployment_block    INTEGER,        -- Prefect 2.x deployment definition
    -- Anti-patterns
    antipattern_no_retries          INTEGER,
    antipattern_no_timeout          INTEGER,
    file_lines              INTEGER,
    FOREIGN KEY(repo_full_name, file_path) REFERENCES pipeline_files(repo_full_name, file_path)
);

-- ── Dagster ──────────────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS dagster_features (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT,
    repo_full_name          TEXT    NOT NULL,
    file_path               TEXT    NOT NULL,
    asset_count             INTEGER,
    op_count                INTEGER,
    job_count               INTEGER,
    graph_count             INTEGER,
    resource_count          INTEGER,
    has_partitions          INTEGER,
    has_sensors             INTEGER,
    has_schedules           INTEGER,
    uses_io_manager         INTEGER,
    uses_config_schema      INTEGER,
    uses_metadata           INTEGER,
    has_freshness_policy    INTEGER,
    -- Anti-patterns
    antipattern_op_without_io_manager   INTEGER, -- op returning data without IOManager
    antipattern_no_metadata             INTEGER,
    file_lines              INTEGER,
    FOREIGN KEY(repo_full_name, file_path) REFERENCES pipeline_files(repo_full_name, file_path)
);

-- ── Luigi ────────────────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS luigi_task_features (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT,
    repo_full_name          TEXT    NOT NULL,
    file_path               TEXT    NOT NULL,
    task_count              INTEGER,
    max_requires_depth      INTEGER,        -- longest dependency chain in file
    uses_local_target       INTEGER,
    uses_s3_target          INTEGER,
    uses_gcs_target         INTEGER,
    uses_hdfs_target        INTEGER,
    uses_external_task      INTEGER,        -- ExternalTask dependency
    has_dynamic_requires    INTEGER,        -- requires() returns variable list
    has_run_method          INTEGER,
    has_output_method       INTEGER,
    -- Anti-patterns
    antipattern_missing_output          INTEGER, -- run() without output()
    antipattern_hardcoded_path          INTEGER, -- string literal in LocalTarget
    file_lines              INTEGER,
    FOREIGN KEY(repo_full_name, file_path) REFERENCES pipeline_files(repo_full_name, file_path)
);

-- ── Kedro ────────────────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS kedro_pipeline_features (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT,
    repo_full_name          TEXT    NOT NULL,
    file_path               TEXT    NOT NULL,
    node_count              INTEGER,
    pipeline_count          INTEGER,
    dataset_types           TEXT,           -- JSON array of dataset type strings
    has_parameters_file     INTEGER,
    has_catalog_file        INTEGER,
    uses_versioning         INTEGER,        -- versioned datasets
    uses_partitioned_ds     INTEGER,
    file_lines              INTEGER,
    FOREIGN KEY(repo_full_name, file_path) REFERENCES pipeline_files(repo_full_name, file_path)
);

-- ── Apache Beam ──────────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS beam_pipeline_features (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT,
    repo_full_name          TEXT    NOT NULL,
    file_path               TEXT    NOT NULL,
    ptransform_count        INTEGER,        -- custom PTransform subclasses
    pardo_count             INTEGER,
    combinefn_count         INTEGER,
    runner_type             TEXT,           -- DirectRunner | DataflowRunner | SparkRunner
    uses_windowing          INTEGER,
    uses_side_inputs        INTEGER,
    uses_state_and_timers   INTEGER,
    file_lines              INTEGER,
    FOREIGN KEY(repo_full_name, file_path) REFERENCES pipeline_files(repo_full_name, file_path)
);

-- ── dlt (data load tool) ────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS dlt_features (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT,
    repo_full_name          TEXT    NOT NULL,
    file_path               TEXT    NOT NULL,
    source_count            INTEGER,        -- @dlt.source decorated functions
    resource_count          INTEGER,        -- @dlt.resource decorated functions
    destination_types       TEXT,           -- JSON array: bigquery, snowflake, etc.
    uses_incremental        INTEGER,
    uses_schema_contracts   INTEGER,
    uses_transformers       INTEGER,        -- @dlt.transformer
    file_lines              INTEGER,
    FOREIGN KEY(repo_full_name, file_path) REFERENCES pipeline_files(repo_full_name, file_path)
);

-- ── Indexes ──────────────────────────────────────────────────────────────────
CREATE INDEX IF NOT EXISTS idx_pf_repo   ON pipeline_files(repo_full_name);
CREATE INDEX IF NOT EXISTS idx_pf_type   ON pipeline_files(artifact_type);
CREATE INDEX IF NOT EXISTS idx_adf_repo  ON airflow_dag_features(repo_full_name);
CREATE INDEX IF NOT EXISTS idx_dbt_repo  ON dbt_model_features(repo_full_name);
CREATE INDEX IF NOT EXISTS idx_prf_repo  ON prefect_flow_features(repo_full_name);
CREATE INDEX IF NOT EXISTS idx_dag_repo  ON dagster_features(repo_full_name);
CREATE INDEX IF NOT EXISTS idx_lui_repo  ON luigi_task_features(repo_full_name);
CREATE INDEX IF NOT EXISTS idx_ked_repo  ON kedro_pipeline_features(repo_full_name);
