"""Schema and create-or-load helper for the events Iceberg table.

Iceberg-internal schema uses the canonical column names for this pipeline:
  - ``actor``      (source column is also ``actor`` in the L02 Parquet)
  - ``event_date`` (source column ``date`` is renamed during ingestion in
                    write_from_parquet.py; the warehouse data dirs will
                    therefore use ``event_date=...`` Hive-style directories)

This is distinct from the raw source Parquet layout at
``data/parquet/date=*/`` which uses a ``date`` partition column.
"""

from __future__ import annotations

from pyiceberg.schema import Schema
from pyiceberg.types import (
    DateType,
    LongType,
    NestedField,
    StringType,
    TimestampType,
)
from pyiceberg.partitioning import PartitionSpec, PartitionField
from pyiceberg.transforms import IdentityTransform

from .catalog import open_catalog

NAMESPACE = "default"
TABLE_NAME = "events"
FULL_NAME = f"{NAMESPACE}.{TABLE_NAME}"

# Field IDs are stable — renaming a column preserves the ID so older
# snapshots continue to resolve correctly.
SCHEMA = Schema(
    NestedField(1, "event_id",   LongType(),       required=True),
    NestedField(2, "type",       StringType(),     required=True),
    NestedField(3, "actor",      StringType(),     required=False),
    NestedField(4, "repo",       StringType(),     required=False),
    NestedField(5, "created_at", TimestampType(),  required=True),
    NestedField(6, "event_date", DateType(),       required=True),
)

# Identity-transform partition on event_date — same logical partitioning
# as the source Parquet, but expressed as Iceberg metadata.
PARTITION_SPEC = PartitionSpec(
    PartitionField(
        source_id=6,
        field_id=1000,
        transform=IdentityTransform(),
        name="event_date",
    ),
)


def create_or_load_events():
    """Create namespace + table if missing; return the loaded table."""
    cat = open_catalog()
    if NAMESPACE not in [n[0] for n in cat.list_namespaces()]:
        cat.create_namespace(NAMESPACE)
    if not cat.table_exists(FULL_NAME):
        cat.create_table(
            identifier=FULL_NAME,
            schema=SCHEMA,
            partition_spec=PARTITION_SPEC,
        )
    return cat.load_table(FULL_NAME)