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 File(Stats):
    """File metadata model. Arbitrary data can be stored in `extra`, including
    ftm properties that should be added to the generated Entity"""

    model_config = ConfigDict(extra="allow")

    id: str

    dataset: str
    """Dataset name"""
    checksum: str
    """SHA256 checksum (often referred to as `content_hash`)"""
    extra: dict[str, Any] = {}
    """Arbitrary extra data"""
    origin: str | None = None
    """Origin stage of this file"""

    @field_validator("checksum")
    @classmethod
    def check_checksum(cls, v: str) -> str:
        return validate_checksum(v)

    @field_validator("store")
    @classmethod
    def hide_store(cls, v: str) -> str:
        # always don't include original store uri
        return "lakehouse://"

    @model_validator(mode="before")
    @classmethod
    def collect_extra_fields(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        known_fields = set(cls.model_fields.keys())
        known_fields.update(["path", "parent"])  # computed_field
        extra = data.get("extra", {})
        extra_fields = {k: v for k, v in data.items() if k not in known_fields}
        if extra_fields:
            data = {k: v for k, v in data.items() if k in known_fields}
            data["extra"] = {**extra, **extra_fields}
        # an explicitly given id may arrive as a non-string (e.g. a numeric
        # source id from a crawler payload) – pydantic won't coerce that
        data["id"] = str(data["id"]) if data.get("id") else cls.make_id(data)
        return data

    def to_entity(self) -> StatementEntity:
        """Make an entity for this File"""
        schema = mime_to_schema(self.mimetype)
        entity = make_entity(
            {"id": self.id, "schema": schema},
            entity_type=StatementEntity,
            default_dataset=self.dataset,
        )
        entity.add("contentHash", self.checksum)
        entity.add("fileName", self.name)
        entity.add("fileSize", self.size)
        entity.add("mimeType", self.mimetype)
        entity.add("parent", self.parent)
        for prop in schema.properties:
            if prop in self.extra:
                entity.add(prop, self.extra[prop])
        return entity

    def make_parents(self) -> StatementEntities:
        """Make parent `Folder` entities"""
        parent = Path(self.key).parent
        if parent.name:
            yield from make_folders(parent, dataset=self.dataset)

    def make_entities(self) -> StatementEntities:
        yield from self.make_parents()
        yield self.to_entity()

    @staticmethod
    def make_id(data: SDict) -> str:
        """The entity id is generated by a hash of the file path and the
        checksum. Uses just the checksum as id if that's the key"""
        if data["key"] == data["checksum"]:
            return data["key"]
        return make_file_id(data["key"], data["checksum"])

    @computed_field
    @property
    def path(self) -> str:
        # "key" can be misleading in the codebase so this is an alias
        return self.key

    @computed_field
    @property
    def parent(self) -> str | None:
        for parent in reversed(list(self.make_parents())):
            return parent.id

    @property
    def blob_path(self) -> str:
        """Relative path to blob in dataset archive"""
        return path.ArchiveKey(self.checksum).blob

    @property
    def meta_path(self) -> str:
        """Relative path for this file's metadata json in dataset archive"""
        return path.ArchiveKey(self.checksum).meta(self.id)

    @classmethod
    def from_info(cls, info: Stats, checksum: str, **data) -> Self:
        data["dataset"] = data.get("dataset", DEFAULT_DATASET)
        data["checksum"] = checksum
        return cls(**{**info.model_dump(), **data})

    def to_document(self) -> Document:
        return Document(
            id=self.id,
            checksum=self.checksum,
            name=self.name,
            path=self.path,
            size=self.size,
            mimetype=self.mimetype,
            updated_at=self.updated_at,
        )

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,
        )