Layer 1: Model

class DatasetModel(Dataset):
    storage: StoreModel | None = None
    """Set storage for external lakehouse"""
    public_url_prefix: HttpUrlStr | None = None
    """Public url prefix for resources"""
    shards: int = DEFAULT_SHARDS
    """Number of entity-id hash shards for the parquet store. ``0`` (default)
    means a single shard; huge datasets should configure ``8`` or more at
    creation for bounded per-partition working sets (e.g.
    ``ensure_dataset("big_leak", shards=8)``). Fixed once the store is
    written: setting it here only changes where readers *look*, so changing
    it after the fact means a full rewrite –
    [`ShardOperation`][ftm_lakehouse.operation.maintenance.ShardOperation]."""
    compression: CompressKind | None = None
    """Compress exported artifacts (statements.csv, entities.ftm.json, diffs...)"""

    def get_public_prefix(self) -> str | None:
        if self.public_url_prefix:
            return self.public_url_prefix
        if settings.public_url_prefix:
            # ``${dataset}`` placeholder; safe_substitute so literal ``$``
            # (and ``%``) in a prefix never breaks
            return Template(settings.public_url_prefix).safe_substitute(
                dataset=self.name
            )

class LakehouseStatement(LakeStatement):
    """A statement carrying the two columns the lakehouse adds to the schema.

    ``deleted_at`` is the tombstone marker – a storage fact about a statement
    rather than statement content, and lakehouse-only (ftmq's lake store
    deletes physically). ``role`` records *who* asserted the statement: an
    identifier a submitting application supplies, alongside ``origin``'s
    *where*. Both live here for the same reason ``fragment`` lives on
    `ftmq.store.lake.LakeStatement`: so the write path can pass statements
    around instead of ``(stmt, deleted_at, role)`` tuples.

    ``role`` joins ``origin`` and ``fragment`` in `dedupe_key`, so two roles
    asserting identical content stay two rows through
    [`merge`][ftm_lakehouse.storage.parquet.ParquetStore.merge] – full
    provenance, rather than one row whose role is whoever wrote last. The
    empty string collapses to ``None`` so "no role" has one representation.

    There is deliberately no ``shard`` attribute – a statement is content plus
    provenance, and which partition it lands in is
    [`append`][ftm_lakehouse.storage.parquet.ParquetStore.append]'s call.
    """

    __slots__ = ["deleted_at", "role"]

    def __init__(
        self,
        *args: Any,
        deleted_at: datetime | None = None,
        role: str | None = None,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)
        self.deleted_at = deleted_at
        self.role = role or None

    @property
    def dedupe_key(self) -> str:
        """Stable row identity: ``id``, ``origin``, ``fragment``, ``role``.

        Extends `ftmq.store.lake.LakeStatement.dedupe_key` with the
        lakehouse's fourth identity dimension, so the write buffers collapse
        re-emissions exactly where `merge` does.
        `ftm_lakehouse.helpers.statements.dedupe_key` keeps the same key shape
        for the packed-row paths that never build a statement object.
        """
        return f"{super().dedupe_key}\t{self.role or ''}"

    @classmethod
    def from_dict(cls, data: StatementDict) -> "LakehouseStatement":
        """Read a statement back from a row dict, keeping ``role``.

        ``deleted_at`` is deliberately not read back: every consumer of this
        (statement queries, the api NDJSON wire) reads the *live* view, where
        a surfaced row is by definition not a tombstone.
        """
        stmt = cast("LakehouseStatement", super().from_dict(data))
        stmt.role = cast(dict[str, Any], data).get("role") or None
        return stmt

    @classmethod
    def from_db_row(cls, row: Any) -> "LakehouseStatement":
        """Read a statement back from a SQL row, keeping ``role``."""
        stmt = cast("LakehouseStatement", super().from_db_row(row))
        stmt.role = getattr(row, "role", None) or None
        return stmt

class JobModel(BaseModel):
    """Status model for a (probably long running) job"""

    run_id: str
    started: datetime | None = None
    stopped: datetime | None = None
    last_updated: datetime | None = None
    pending: int = 0
    done: int = 0
    errors: int = 0
    running: bool = False
    exc: str | None = None
    took: timedelta = timedelta()

    @computed_field
    @property
    def name(self) -> str:
        return self.__class__.__name__

    @field_validator("run_id", mode="before")
    @classmethod
    def ensure_run_id(cls, value: str | None = None) -> str:
        """Give a manual run id or create one"""
        return value or ensure_uuid()

    def touch(self) -> None:
        self.last_updated = utc_now()

    def stop(self, exc: Exception | None = None) -> None:
        self.running = False
        self.stopped = utc_now()
        self.exc = str(exc)
        if self.started and self.stopped:
            self.took = self.stopped - self.started

    @classmethod
    def make(cls, **kwargs) -> Self:
        kwargs["run_id"] = cls.ensure_run_id(kwargs.get("run_id"))
        return cls(**kwargs)

    @cached_property
    def log(self) -> BoundLogger:
        return get_logger(__name__, run_id=self.run_id)

class DatasetJobModel(JobModel):
    """Status model for a (probably long running) job bound to a dataset"""

    dataset: str

    @cached_property
    def log(self) -> BoundLogger:
        return get_logger(
            f"{self.dataset}.{self.name}",
            run_id=self.run_id,
            dataset=self.dataset,
        )