"""BigQuery adapter: register the Iceberg table as an external table,
then materialise daily_counts as a view."""

from __future__ import annotations

import os

REQUIRED_ENV = ("GOOGLE_CLOUD_PROJECT", "BIGQUERY_DATASET",
                "ICEBERG_GCS_PATH")


def _check_env() -> None:
    missing = [k for k in REQUIRED_ENV if not os.environ.get(k)]
    if missing:
        raise NotImplementedError(
            "BigQuery adapter requires env vars: " + ", ".join(missing) +
            ". For the DuckDB-only fallback, use "
            "warehouse.duckdb_local.materialize_daily_counts() instead."
        )


def materialize_daily_counts() -> None:
    _check_env()
    from google.cloud import bigquery

    client = bigquery.Client(project=os.environ["GOOGLE_CLOUD_PROJECT"])
    dataset = os.environ["BIGQUERY_DATASET"]
    base = os.environ["ICEBERG_GCS_PATH"]
    # BigQuery's standard Jobs API rejects multi-statement DDL in one query call;
    # run the external-table registration and the view as two separate jobs.
    external_table_ddl = f"""
    CREATE OR REPLACE EXTERNAL TABLE `{dataset}.events`
    WITH CONNECTION `{os.environ.get("BIGQUERY_CONNECTION", "default")}`
    OPTIONS (
        format = 'ICEBERG',
        uris = ['{base}/metadata/v*.metadata.json']
    )
    """
    view_ddl = f"""
    CREATE OR REPLACE VIEW `{dataset}.daily_counts` AS
    SELECT event_date, type, COUNT(*) AS n
    FROM `{dataset}.events`
    GROUP BY event_date, type
    """
    client.query(external_table_ddl).result()
    client.query(view_ddl).result()
    print("bigquery: daily_counts view materialised.")


if __name__ == "__main__":
    materialize_daily_counts()