"""Snowflake adapter: register the Iceberg table as an external table,
then materialise daily_counts."""

from __future__ import annotations

import os

REQUIRED_ENV = ("SNOWFLAKE_ACCOUNT", "SNOWFLAKE_USER",
                "SNOWFLAKE_PASSWORD", "SNOWFLAKE_WAREHOUSE",
                "SNOWFLAKE_DATABASE", "SNOWFLAKE_SCHEMA",
                "ICEBERG_S3_PATH")


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


def materialize_daily_counts() -> None:
    """Create (or refresh) the daily_counts view on top of the Iceberg
    external table.

    Pre-reqs (one-time, run by the operator outside this script):
      1. The Iceberg table's metadata + data live at s3://.../warehouse/events/.
      2. ``CREATE EXTERNAL VOLUME`` for that bucket has been done.
      3. ``CREATE CATALOG INTEGRATION`` for the Iceberg catalog exists.
    """
    _check_env()
    import snowflake.connector  # imported lazily to keep this module
                                # importable even without the SDK.

    con = snowflake.connector.connect(
        account=os.environ["SNOWFLAKE_ACCOUNT"],
        user=os.environ["SNOWFLAKE_USER"],
        password=os.environ["SNOWFLAKE_PASSWORD"],
        warehouse=os.environ["SNOWFLAKE_WAREHOUSE"],
        database=os.environ["SNOWFLAKE_DATABASE"],
        schema=os.environ["SNOWFLAKE_SCHEMA"],
    )
    cur = con.cursor()
    cur.execute("""
        CREATE OR REPLACE ICEBERG TABLE events
        EXTERNAL_VOLUME = 'iceberg_vol'
        CATALOG = 'iceberg_cat'
        BASE_LOCATION = %(base)s
    """, {"base": os.environ["ICEBERG_S3_PATH"]})
    cur.execute("""
        CREATE OR REPLACE VIEW daily_counts AS
        SELECT event_date, type, count(*) AS n
        FROM events
        GROUP BY event_date, type
    """)
    print("snowflake: daily_counts view materialised.")


if __name__ == "__main__":
    materialize_daily_counts()