diff --git a/alembic_db/versions/0007_record_content_split.py b/alembic_db/versions/0007_record_content_split.py new file mode 100644 index 000000000..5ed1428c3 --- /dev/null +++ b/alembic_db/versions/0007_record_content_split.py @@ -0,0 +1,77 @@ +from alembic import op +import sqlalchemy as sa + + +revision = "0007_record_content_split" +down_revision = "0006_add_loader_path" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.drop_table("asset_reference_meta") + op.drop_table("asset_reference_tags") + op.drop_table("asset_references") + op.drop_table("assets") + op.execute("DELETE FROM tags") + op.create_table( + "asset_contents", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("hash", sa.String(256)), + sa.Column("size_bytes", sa.BigInteger(), nullable=False), + sa.Column("path", sa.Text(), nullable=False), + sa.Column("mtime_ns", sa.BigInteger()), + sa.Column("is_missing", sa.Boolean(), nullable=False, server_default="0"), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.CheckConstraint("size_bytes >= 0", name="ck_asset_contents_size_nonneg"), + sa.CheckConstraint("mtime_ns >= 0", name="ck_asset_contents_mtime_nonneg"), + ) + op.create_index("ix_asset_contents_hash", "asset_contents", ["hash"]) + op.create_index( + "uq_asset_contents_path_live", "asset_contents", ["path"], unique=True, + sqlite_where=sa.text("is_missing = 0"), + ) + op.create_table( + "assets", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("content_id", sa.String(36), sa.ForeignKey("asset_contents.id", ondelete="RESTRICT"), nullable=False), + sa.Column("name", sa.String(512), nullable=False), + sa.Column("mime_type", sa.String(255)), + sa.Column("system_metadata", sa.JSON()), + sa.Column("job_id", sa.String(36)), + sa.Column("user_metadata", sa.JSON()), + sa.Column("loader_path", sa.Text()), + sa.Column("preview_id", sa.String(36), sa.ForeignKey("assets.id", ondelete="SET NULL")), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=False), + sa.Column("last_access_time", sa.DateTime()), + ) + op.create_index("ix_assets_content_id", "assets", ["content_id"]) + op.create_index("ix_assets_name", "assets", ["name"]) + op.create_index("ix_assets_created_at", "assets", ["created_at"]) + op.create_index("ix_assets_preview_id", "assets", ["preview_id"]) + op.create_table( + "asset_meta", + sa.Column("asset_id", sa.String(36), sa.ForeignKey("assets.id", ondelete="CASCADE"), primary_key=True), + sa.Column("key", sa.String(256), primary_key=True), + sa.Column("ordinal", sa.Integer(), primary_key=True), + sa.Column("val_str", sa.String(2048)), sa.Column("val_num", sa.Numeric(38, 10)), + sa.Column("val_bool", sa.Boolean()), sa.Column("val_json", sa.JSON()), + sa.CheckConstraint("val_str IS NOT NULL OR val_num IS NOT NULL OR val_bool IS NOT NULL OR val_json IS NOT NULL", name="ck_asset_meta_has_value"), + ) + op.create_table("asset_tags", sa.Column("asset_id", sa.String(36), sa.ForeignKey("assets.id", ondelete="CASCADE"), primary_key=True), sa.Column("tag_name", sa.String(512), sa.ForeignKey("tags.name", ondelete="RESTRICT"), primary_key=True), sa.Column("origin", sa.String(32), nullable=False), sa.Column("added_at", sa.DateTime(), nullable=False)) + op.create_index("ix_asset_tags_tag_name", "asset_tags", ["tag_name"]) + op.create_index("ix_asset_tags_asset_id", "asset_tags", ["asset_id"]) + op.create_table("asset_system_state", sa.Column("key", sa.String(256), primary_key=True), sa.Column("value", sa.Text(), nullable=False)) + + +def downgrade() -> None: + op.drop_table("asset_system_state") + op.drop_table("asset_tags") + op.drop_table("asset_meta") + op.drop_table("assets") + op.drop_table("asset_contents") + op.create_table("assets", sa.Column("id", sa.String(36), primary_key=True), sa.Column("hash", sa.String(256)), sa.Column("size_bytes", sa.BigInteger(), nullable=False), sa.Column("mime_type", sa.String(255)), sa.Column("created_at", sa.DateTime(), nullable=False)) + op.create_table("asset_references", sa.Column("id", sa.String(36), primary_key=True), sa.Column("asset_id", sa.String(36), sa.ForeignKey("assets.id", ondelete="CASCADE"), nullable=False), sa.Column("file_path", sa.Text()), sa.Column("loader_path", sa.Text()), sa.Column("mtime_ns", sa.BigInteger()), sa.Column("needs_verify", sa.Boolean(), nullable=False), sa.Column("is_missing", sa.Boolean(), nullable=False), sa.Column("enrichment_level", sa.Integer(), nullable=False), sa.Column("owner_id", sa.String(128), nullable=False), sa.Column("name", sa.String(512), nullable=False), sa.Column("preview_id", sa.String(36), sa.ForeignKey("asset_references.id", ondelete="SET NULL")), sa.Column("user_metadata", sa.JSON()), sa.Column("system_metadata", sa.JSON()), sa.Column("job_id", sa.String(36)), sa.Column("created_at", sa.DateTime(), nullable=False), sa.Column("updated_at", sa.DateTime(), nullable=False), sa.Column("last_access_time", sa.DateTime(), nullable=False), sa.Column("deleted_at", sa.DateTime())) + op.create_table("asset_reference_meta", sa.Column("asset_reference_id", sa.String(36), sa.ForeignKey("asset_references.id", ondelete="CASCADE"), primary_key=True), sa.Column("key", sa.String(256), primary_key=True), sa.Column("ordinal", sa.Integer(), primary_key=True), sa.Column("val_str", sa.String(2048)), sa.Column("val_num", sa.Numeric(38, 10)), sa.Column("val_bool", sa.Boolean()), sa.Column("val_json", sa.JSON()), sa.CheckConstraint("val_str IS NOT NULL OR val_num IS NOT NULL OR val_bool IS NOT NULL OR val_json IS NOT NULL", name="ck_asset_reference_meta_has_value")) + op.create_table("asset_reference_tags", sa.Column("asset_reference_id", sa.String(36), sa.ForeignKey("asset_references.id", ondelete="CASCADE"), primary_key=True), sa.Column("tag_name", sa.String(512), sa.ForeignKey("tags.name", ondelete="RESTRICT"), primary_key=True), sa.Column("origin", sa.String(32), nullable=False), sa.Column("added_at", sa.DateTime(), nullable=False)) diff --git a/app/assets/database/models.py b/app/assets/database/models.py index 329cd483d..239b93d36 100644 --- a/app/assets/database/models.py +++ b/app/assets/database/models.py @@ -2,6 +2,7 @@ from __future__ import annotations import uuid from datetime import datetime +from decimal import Decimal from typing import Any from sqlalchemy import ( @@ -16,197 +17,133 @@ from sqlalchemy import ( Numeric, String, Text, + text, ) -from sqlalchemy.orm import Mapped, foreign, mapped_column, relationship +from sqlalchemy.orm import Mapped, mapped_column, relationship from app.assets.helpers import get_utc_now from app.database.models import Base +class AssetContent(Base): + __tablename__ = "asset_contents" + + id: Mapped[str] = mapped_column( + String(36), primary_key=True, default=lambda: str(uuid.uuid4()) + ) + hash: Mapped[str | None] = mapped_column(String(256), index=True) + size_bytes: Mapped[int] = mapped_column( + BigInteger, + CheckConstraint("size_bytes >= 0", name="ck_asset_contents_size_nonneg"), + nullable=False, + default=0, + ) + path: Mapped[str] = mapped_column(Text, nullable=False) + mtime_ns: Mapped[int | None] = mapped_column( + BigInteger, + CheckConstraint("mtime_ns >= 0", name="ck_asset_contents_mtime_nonneg"), + ) + is_missing: Mapped[bool] = mapped_column( + Boolean, nullable=False, server_default="0" + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=False), nullable=False, default=get_utc_now + ) + + records: Mapped[list[Asset]] = relationship(back_populates="content") + + __table_args__ = ( + Index( + "uq_asset_contents_path_live", + "path", + unique=True, + sqlite_where=text("is_missing = 0"), + ), + ) + + class Asset(Base): __tablename__ = "assets" id: Mapped[str] = mapped_column( String(36), primary_key=True, default=lambda: str(uuid.uuid4()) ) - hash: Mapped[str | None] = mapped_column(String(256), nullable=True) - size_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) - mime_type: Mapped[str | None] = mapped_column(String(255)) - created_at: Mapped[datetime] = mapped_column( - DateTime(timezone=False), nullable=False, default=get_utc_now + content_id: Mapped[str] = mapped_column( + String(36), + ForeignKey("asset_contents.id", ondelete="RESTRICT"), + nullable=False, ) - - references: Mapped[list[AssetReference]] = relationship( - "AssetReference", - back_populates="asset", - primaryjoin=lambda: Asset.id == foreign(AssetReference.asset_id), - foreign_keys=lambda: [AssetReference.asset_id], - cascade="all,delete-orphan", - passive_deletes=True, - ) - - # preview_id on AssetReference is a self-referential FK to asset_references.id - - __table_args__ = ( - Index("uq_assets_hash", "hash", unique=True), - Index("ix_assets_mime_type", "mime_type"), - CheckConstraint("size_bytes >= 0", name="ck_assets_size_nonneg"), - ) - - def __repr__(self) -> str: - return f"" - - -class AssetReference(Base): - """Unified model combining file cache state and user-facing metadata. - - Each row represents either: - - A filesystem reference (file_path is set) with cache state - - An API-created reference (file_path is NULL) without cache state - """ - - __tablename__ = "asset_references" - - id: Mapped[str] = mapped_column( - String(36), primary_key=True, default=lambda: str(uuid.uuid4()) - ) - asset_id: Mapped[str] = mapped_column( - String(36), ForeignKey("assets.id", ondelete="CASCADE"), nullable=False - ) - - # Cache state fields (from former AssetCacheState) - file_path: Mapped[str | None] = mapped_column(Text, nullable=True) - # In-root loader path derived from file_path at scan/ingest time. - loader_path: Mapped[str | None] = mapped_column(Text, nullable=True) - mtime_ns: Mapped[int | None] = mapped_column(BigInteger, nullable=True) - needs_verify: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) - is_missing: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) - enrichment_level: Mapped[int] = mapped_column(Integer, nullable=False, default=0) - - # Info fields (from former AssetInfo) - owner_id: Mapped[str] = mapped_column(String(128), nullable=False, default="") name: Mapped[str] = mapped_column(String(512), nullable=False) + mime_type: Mapped[str | None] = mapped_column(String(255)) + system_metadata: Mapped[dict[str, Any] | None] = mapped_column(JSON) + job_id: Mapped[str | None] = mapped_column(String(36)) + user_metadata: Mapped[dict[str, Any] | None] = mapped_column(JSON) + loader_path: Mapped[str | None] = mapped_column(Text) preview_id: Mapped[str | None] = mapped_column( - String(36), ForeignKey("asset_references.id", ondelete="SET NULL") + String(36), ForeignKey("assets.id", ondelete="SET NULL") ) - user_metadata: Mapped[dict[str, Any] | None] = mapped_column( - JSON(none_as_null=True) - ) - system_metadata: Mapped[dict[str, Any] | None] = mapped_column( - JSON(none_as_null=True), nullable=True, default=None - ) - job_id: Mapped[str | None] = mapped_column(String(36), nullable=True, default=None) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=False), nullable=False, default=get_utc_now ) updated_at: Mapped[datetime] = mapped_column( - DateTime(timezone=False), nullable=False, default=get_utc_now - ) - last_access_time: Mapped[datetime] = mapped_column( - DateTime(timezone=False), nullable=False, default=get_utc_now - ) - deleted_at: Mapped[datetime | None] = mapped_column( - DateTime(timezone=False), nullable=True, default=None + DateTime(timezone=False), nullable=False, default=get_utc_now, onupdate=get_utc_now ) + last_access_time: Mapped[datetime | None] = mapped_column(DateTime(timezone=False)) - asset: Mapped[Asset] = relationship( - "Asset", - back_populates="references", - foreign_keys=[asset_id], - lazy="selectin", + content: Mapped[AssetContent] = relationship(back_populates="records", lazy="selectin") + preview: Mapped[Asset | None] = relationship( + "Asset", foreign_keys=[preview_id], remote_side=lambda: [Asset.id] ) - preview_ref: Mapped[AssetReference | None] = relationship( - "AssetReference", - foreign_keys=[preview_id], - remote_side=lambda: [AssetReference.id], + metadata_entries: Mapped[list[AssetMeta]] = relationship( + back_populates="asset", cascade="all,delete-orphan", passive_deletes=True ) - - metadata_entries: Mapped[list[AssetReferenceMeta]] = relationship( - back_populates="asset_reference", - cascade="all,delete-orphan", - passive_deletes=True, + tag_links: Mapped[list[AssetTag]] = relationship( + back_populates="asset", cascade="all,delete-orphan", passive_deletes=True ) - - tag_links: Mapped[list[AssetReferenceTag]] = relationship( - back_populates="asset_reference", - cascade="all,delete-orphan", - passive_deletes=True, - overlaps="tags,asset_references", - ) - tags: Mapped[list[Tag]] = relationship( - secondary="asset_reference_tags", - back_populates="asset_references", - lazy="selectin", - viewonly=True, - overlaps="tag_links,asset_reference_links,asset_references,tag", + secondary="asset_tags", back_populates="assets", viewonly=True, lazy="selectin" ) __table_args__ = ( - Index("uq_asset_references_file_path", "file_path", unique=True), - Index("ix_asset_references_asset_id", "asset_id"), - Index("ix_asset_references_owner_id", "owner_id"), - Index("ix_asset_references_name", "name"), - Index("ix_asset_references_is_missing", "is_missing"), - Index("ix_asset_references_enrichment_level", "enrichment_level"), - Index("ix_asset_references_created_at", "created_at"), - Index("ix_asset_references_last_access_time", "last_access_time"), - Index("ix_asset_references_deleted_at", "deleted_at"), - Index("ix_asset_references_preview_id", "preview_id"), - Index("ix_asset_references_owner_name", "owner_id", "name"), - CheckConstraint( - "(mtime_ns IS NULL) OR (mtime_ns >= 0)", name="ck_ar_mtime_nonneg" - ), - CheckConstraint( - "enrichment_level >= 0 AND enrichment_level <= 2", - name="ck_ar_enrichment_level_range", - ), + Index("ix_assets_content_id", "content_id"), + Index("ix_assets_name", "name"), + Index("ix_assets_created_at", "created_at"), + Index("ix_assets_preview_id", "preview_id"), ) - def __repr__(self) -> str: - path_part = f" path={self.file_path!r}" if self.file_path else "" - return f"" +class AssetMeta(Base): + __tablename__ = "asset_meta" -class AssetReferenceMeta(Base): - __tablename__ = "asset_reference_meta" - - asset_reference_id: Mapped[str] = mapped_column( - String(36), - ForeignKey("asset_references.id", ondelete="CASCADE"), - primary_key=True, + asset_id: Mapped[str] = mapped_column( + String(36), ForeignKey("assets.id", ondelete="CASCADE"), primary_key=True ) key: Mapped[str] = mapped_column(String(256), primary_key=True) ordinal: Mapped[int] = mapped_column(Integer, primary_key=True, default=0) + val_str: Mapped[str | None] = mapped_column(String(2048)) + val_num: Mapped[Decimal | None] = mapped_column(Numeric(38, 10)) + val_bool: Mapped[bool | None] = mapped_column(Boolean) + val_json: Mapped[Any | None] = mapped_column(JSON) - val_str: Mapped[str | None] = mapped_column(String(2048), nullable=True) - val_num: Mapped[float | None] = mapped_column(Numeric(38, 10), nullable=True) - val_bool: Mapped[bool | None] = mapped_column(Boolean, nullable=True) - val_json: Mapped[Any | None] = mapped_column(JSON(none_as_null=True), nullable=True) - - asset_reference: Mapped[AssetReference] = relationship( - back_populates="metadata_entries" - ) + asset: Mapped[Asset] = relationship(back_populates="metadata_entries") __table_args__ = ( - Index("ix_asset_reference_meta_key", "key"), - Index("ix_asset_reference_meta_key_val_str", "key", "val_str"), - Index("ix_asset_reference_meta_key_val_num", "key", "val_num"), - Index("ix_asset_reference_meta_key_val_bool", "key", "val_bool"), + Index("ix_asset_meta_key", "key"), + Index("ix_asset_meta_key_val_str", "key", "val_str"), + Index("ix_asset_meta_key_val_num", "key", "val_num"), + Index("ix_asset_meta_key_val_bool", "key", "val_bool"), CheckConstraint( "val_str IS NOT NULL OR val_num IS NOT NULL OR val_bool IS NOT NULL OR val_json IS NOT NULL", - name="has_value", + name="ck_asset_meta_has_value", ), ) -class AssetReferenceTag(Base): - __tablename__ = "asset_reference_tags" +class AssetTag(Base): + __tablename__ = "asset_tags" - asset_reference_id: Mapped[str] = mapped_column( - String(36), - ForeignKey("asset_references.id", ondelete="CASCADE"), - primary_key=True, + asset_id: Mapped[str] = mapped_column( + String(36), ForeignKey("assets.id", ondelete="CASCADE"), primary_key=True ) tag_name: Mapped[str] = mapped_column( String(512), ForeignKey("tags.name", ondelete="RESTRICT"), primary_key=True @@ -216,12 +153,12 @@ class AssetReferenceTag(Base): DateTime(timezone=False), nullable=False, default=get_utc_now ) - asset_reference: Mapped[AssetReference] = relationship(back_populates="tag_links") - tag: Mapped[Tag] = relationship(back_populates="asset_reference_links") + asset: Mapped[Asset] = relationship(back_populates="tag_links") + tag: Mapped[Tag] = relationship(back_populates="asset_links") __table_args__ = ( - Index("ix_asset_reference_tags_tag_name", "tag_name"), - Index("ix_asset_reference_tags_asset_reference_id", "asset_reference_id"), + Index("ix_asset_tags_tag_name", "tag_name"), + Index("ix_asset_tags_asset_id", "asset_id"), ) @@ -229,17 +166,19 @@ class Tag(Base): __tablename__ = "tags" name: Mapped[str] = mapped_column(String(512), primary_key=True) - - asset_reference_links: Mapped[list[AssetReferenceTag]] = relationship( - back_populates="tag", - overlaps="asset_references,tags", - ) - asset_references: Mapped[list[AssetReference]] = relationship( - secondary="asset_reference_tags", - back_populates="tags", - viewonly=True, - overlaps="asset_reference_links,tag_links,tags,asset_reference", + asset_links: Mapped[list[AssetTag]] = relationship(back_populates="tag") + assets: Mapped[list[Asset]] = relationship( + secondary="asset_tags", back_populates="tags", viewonly=True ) - def __repr__(self) -> str: - return f"" + +class AssetSystemState(Base): + __tablename__ = "asset_system_state" + + key: Mapped[str] = mapped_column(String(256), primary_key=True) + value: Mapped[str] = mapped_column(Text, nullable=False) + + +AssetReference = None +AssetReferenceMeta = None +AssetReferenceTag = None diff --git a/app/assets/database/queries/__init__.py b/app/assets/database/queries/__init__.py index 38dd82d81..7f0883b2c 100644 --- a/app/assets/database/queries/__init__.py +++ b/app/assets/database/queries/__init__.py @@ -1,139 +1,32 @@ -from app.assets.database.queries.asset import ( - asset_exists_by_hash, - bulk_insert_assets, - create_stub_asset, - get_asset_by_hash, - get_existing_asset_ids, - reassign_asset_references, - update_asset_hash_and_mime, - upsert_asset, -) -from app.assets.database.queries.asset_reference import ( - CacheStateRow, - UnenrichedReferenceRow, - bulk_insert_references_ignore_conflicts, - bulk_update_enrichment_level, - count_active_siblings, - bulk_update_is_missing, - bulk_update_needs_verify, - convert_metadata_to_rows, - delete_assets_by_ids, - delete_orphaned_seed_asset, - delete_reference_by_id, - delete_references_by_ids, - fetch_reference_and_asset, - fetch_reference_asset_and_tags, - get_or_create_reference, - get_reference_by_file_path, - get_reference_by_id, - get_reference_with_owner_check, - get_reference_ids_by_ids, - get_reference_paths_by_ids, - get_references_by_paths_and_asset_ids, - get_references_for_prefixes, - get_unenriched_references, - get_unreferenced_unhashed_asset_ids, - insert_reference, - list_all_file_paths_by_asset_id, - list_references_by_asset_id, - list_references_page, - mark_references_missing_outside_prefixes, - rebuild_metadata_projection, - reference_exists, - reference_exists_for_asset_id, - restore_references_by_paths, - set_reference_metadata, - set_reference_preview, - set_reference_system_metadata, - soft_delete_reference_by_id, - update_reference_access_time, - update_reference_name, - update_is_missing_by_asset_id, - update_reference_timestamps, - update_reference_updated_at, - upsert_reference, -) -from app.assets.database.queries.tags import ( - AddTagsResult, - RemoveTagsResult, - SetTagsResult, - add_missing_tag_for_asset_id, - add_tags_to_reference, - bulk_insert_tags_and_meta, - ensure_tags_exist, - get_reference_tags, - list_tag_counts_for_filtered_assets, - list_tags_with_usage, - remove_missing_tag_for_asset_id, - remove_tags_from_reference, - set_reference_tags, - validate_tags_exist, +from app.assets.database.queries.records import ( + create_content, + create_record, + delete_record, + get_record_by_id, + list_records_page, + mark_content_missing, + rename_record, + unset_content_missing, ) __all__ = [ - "AddTagsResult", - "CacheStateRow", - "RemoveTagsResult", - "SetTagsResult", - "UnenrichedReferenceRow", - "add_missing_tag_for_asset_id", - "add_tags_to_reference", - "asset_exists_by_hash", - "bulk_insert_assets", - "bulk_insert_references_ignore_conflicts", - "bulk_insert_tags_and_meta", - "bulk_update_enrichment_level", - "count_active_siblings", - "create_stub_asset", - "bulk_update_is_missing", - "bulk_update_needs_verify", - "convert_metadata_to_rows", - "delete_assets_by_ids", - "delete_orphaned_seed_asset", - "delete_reference_by_id", - "delete_references_by_ids", - "ensure_tags_exist", - "fetch_reference_and_asset", - "fetch_reference_asset_and_tags", - "get_asset_by_hash", - "get_existing_asset_ids", - "get_or_create_reference", - "get_reference_by_file_path", - "get_reference_by_id", - "get_reference_with_owner_check", - "get_reference_ids_by_ids", - "get_reference_paths_by_ids", - "get_reference_tags", - "get_references_by_paths_and_asset_ids", - "get_references_for_prefixes", - "get_unenriched_references", - "get_unreferenced_unhashed_asset_ids", - "insert_reference", - "list_all_file_paths_by_asset_id", - "list_references_by_asset_id", - "list_references_page", - "list_tag_counts_for_filtered_assets", - "list_tags_with_usage", - "mark_references_missing_outside_prefixes", - "reassign_asset_references", - "rebuild_metadata_projection", - "reference_exists", - "reference_exists_for_asset_id", - "remove_missing_tag_for_asset_id", - "remove_tags_from_reference", - "restore_references_by_paths", - "set_reference_metadata", - "set_reference_preview", - "set_reference_system_metadata", - "soft_delete_reference_by_id", - "set_reference_tags", - "update_asset_hash_and_mime", - "update_is_missing_by_asset_id", - "update_reference_access_time", - "update_reference_name", - "update_reference_timestamps", - "update_reference_updated_at", - "upsert_asset", - "upsert_reference", - "validate_tags_exist", + "create_content", + "create_record", + "delete_record", + "get_record_by_id", + "list_records_page", + "mark_content_missing", + "rename_record", + "unset_content_missing", ] + + +def __getattr__(name: str): + from importlib import import_module + + for module_name in ("asset", "asset_reference", "tags"): + module = import_module(f"app.assets.database.queries.{module_name}") + candidate = getattr(module, name, None) + if candidate is not None: + return candidate + raise AttributeError(name) diff --git a/app/assets/database/queries/records.py b/app/assets/database/queries/records.py new file mode 100644 index 000000000..ce871fb70 --- /dev/null +++ b/app/assets/database/queries/records.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app.assets.database.models import Asset, AssetContent, AssetTag, Tag +from app.assets.helpers import get_utc_now + + +def create_content(session: Session, path: str, hash: str | None = None, size_bytes: int = 0, mtime_ns: int | None = None) -> AssetContent: + content = AssetContent(path=path, hash=hash, size_bytes=size_bytes, mtime_ns=mtime_ns) + try: + with session.begin_nested(): + session.add(content) + session.flush() + return content + except IntegrityError: + winner = session.execute(sa.select(AssetContent).where(AssetContent.path == path, AssetContent.is_missing.is_(False))).scalar_one() + return winner + + +def create_record(session: Session, content_id: str, name: str, mime_type: str | None = None, job_id: str | None = None, loader_path: str | None = None, tags: Sequence[str] | None = None) -> Asset: + record = Asset(content_id=content_id, name=name, mime_type=mime_type, job_id=job_id, loader_path=loader_path) + session.add(record) + session.flush() + for tag_name in tags or (): + if session.get(Tag, tag_name) is None: + session.add(Tag(name=tag_name)) + session.flush() + session.add(AssetTag(asset_id=record.id, tag_name=tag_name)) + session.flush() + return record + + +def get_record_by_id(session: Session, id: str) -> Asset | None: + return session.get(Asset, id) + + +def list_records_page(session: Session, cursor: str | None = None, limit: int = 50, include_tags: Sequence[str] | None = None, exclude_tags: Sequence[str] | None = None) -> tuple[list[Asset], str | None]: + statement = sa.select(Asset).order_by(Asset.created_at.asc(), Asset.id.asc()).limit(limit) + if cursor is not None: + statement = statement.where(Asset.id > cursor) + for tag_name in include_tags or (): + statement = statement.where(sa.exists(sa.select(AssetTag.asset_id).where(AssetTag.asset_id == Asset.id, AssetTag.tag_name == tag_name))) + for tag_name in exclude_tags or (): + statement = statement.where(~sa.exists(sa.select(AssetTag.asset_id).where(AssetTag.asset_id == Asset.id, AssetTag.tag_name == tag_name))) + records = list(session.execute(statement).scalars()) + return records, records[-1].id if len(records) == limit else None + + +def rename_record(session: Session, id: str, name: str) -> Asset: + record = session.get(Asset, id) + if record is None: + raise LookupError(id) + record.name = name + record.updated_at = get_utc_now() + session.flush() + return record + + +def delete_record(session: Session, id: str) -> None: + record = session.get(Asset, id) + if record is None: + return + preview_id = record.preview_id + session.delete(record) + session.flush() + if preview_id is not None and session.scalar(sa.select(sa.func.count()).select_from(Asset).where(Asset.preview_id == preview_id)) == 0: + preview = session.get(Asset, preview_id) + if preview is not None: + session.delete(preview) + session.flush() + + +def mark_content_missing(session: Session, content_id: str) -> None: + content = session.get(AssetContent, content_id) + if content is None: + raise LookupError(content_id) + content.is_missing = True + if session.get(Tag, "missing") is None: + session.add(Tag(name="missing")) + session.flush() + for record_id in session.scalars(sa.select(Asset.id).where(Asset.content_id == content_id)): + if session.get(AssetTag, {"asset_id": record_id, "tag_name": "missing"}) is None: + session.add(AssetTag(asset_id=record_id, tag_name="missing", origin="automatic")) + session.flush() + + +def unset_content_missing(session: Session, content_id: str) -> None: + content = session.get(AssetContent, content_id) + if content is None: + raise LookupError(content_id) + content.is_missing = False + session.execute(sa.delete(AssetTag).where(AssetTag.tag_name == "missing", AssetTag.asset_id.in_(sa.select(Asset.id).where(Asset.content_id == content_id)))) + session.flush() diff --git a/app/assets/mode.py b/app/assets/mode.py new file mode 100644 index 000000000..7ea2705c8 --- /dev/null +++ b/app/assets/mode.py @@ -0,0 +1,22 @@ +"""Runtime hash-mode accessor.""" + +from __future__ import annotations + +from typing import Protocol + + +class _HashingArguments(Protocol): + enable_asset_hashing: bool + + +_args: _HashingArguments | None = None + + +def init(args: _HashingArguments) -> None: + global _args + _args = args + + +def hashing_enabled() -> bool: + """Return whether startup enabled asset hashing.""" + return bool(getattr(_args, "enable_asset_hashing", False)) diff --git a/app/assets/services/__init__.py b/app/assets/services/__init__.py index 2747ad35b..cdcf010ba 100644 --- a/app/assets/services/__init__.py +++ b/app/assets/services/__init__.py @@ -1,93 +1,42 @@ -from app.assets.services.asset_management import ( - asset_exists, - delete_asset_reference, - get_asset_by_hash, - get_asset_detail, - list_assets_page, - get_preview_file_paths, - resolve_asset_for_download, - set_asset_preview, - update_asset_metadata, -) -from app.assets.services.bulk_ingest import ( - BulkInsertResult, - batch_insert_seed_assets, - cleanup_unreferenced_assets, -) -from app.assets.services.file_utils import ( - get_mtime_ns, - get_size_and_mtime_ns, - list_files_recursively, - verify_file_unchanged, -) +"""Re-export surface for app.assets.services — keeps routes.py and tests importable during the B-schema cutover.""" from app.assets.services.ingest import ( DependencyMissingError, HashMismatchError, - create_from_hash, - ingest_existing_file, - register_output_files, upload_from_temp_path, + create_from_hash, + register_file_in_place, ) -from app.assets.database.queries import ( - AddTagsResult, - RemoveTagsResult, -) -from app.assets.services.schemas import ( - AssetData, - AssetDetailResult, - AssetSummaryData, - DownloadResolutionResult, - IngestResult, - ListAssetsResult, - ReferenceData, - RegisterAssetResult, - TagUsage, - UploadResult, - UserMetadata, +from app.assets.services.asset_management import ( + get_asset_detail, + update_asset_metadata, + delete_asset_reference, + set_asset_preview, + asset_exists, + list_assets_page, + get_preview_file_paths, + resolve_asset_for_download, ) from app.assets.services.tagging import ( apply_tags, - list_tags, remove_tags, + list_tags, ) __all__ = [ - "AddTagsResult", - "AssetData", - "AssetDetailResult", - "AssetSummaryData", - "ReferenceData", - "BulkInsertResult", "DependencyMissingError", - "DownloadResolutionResult", "HashMismatchError", - "IngestResult", - "ListAssetsResult", - "RegisterAssetResult", - "RemoveTagsResult", - "TagUsage", - "UploadResult", - "UserMetadata", - "apply_tags", - "asset_exists", - "batch_insert_seed_assets", + "upload_from_temp_path", "create_from_hash", - "delete_asset_reference", - "get_asset_by_hash", + "register_file_in_place", "get_asset_detail", - "ingest_existing_file", - "register_output_files", - "get_mtime_ns", - "get_size_and_mtime_ns", + "update_asset_metadata", + "delete_asset_reference", + "set_asset_preview", + "asset_exists", "list_assets_page", - "list_files_recursively", - "list_tags", - "cleanup_unreferenced_assets", - "remove_tags", "get_preview_file_paths", "resolve_asset_for_download", - "set_asset_preview", - "update_asset_metadata", - "upload_from_temp_path", - "verify_file_unchanged", + "apply_tags", + "remove_tags", + "list_tags", ] diff --git a/app/assets/services/ingest.py b/app/assets/services/ingest.py index 1ffb3d634..2423c434f 100644 --- a/app/assets/services/ingest.py +++ b/app/assets/services/ingest.py @@ -25,7 +25,7 @@ from app.assets.database.queries import ( set_reference_tags, update_asset_hash_and_mime, upsert_asset, - upsert_reference, + upsert_reference as _legacy_upsert_reference, # wave-3-fixes: replaced by B-schema write paths in Wave 3 validate_tags_exist, ) from app.assets.helpers import get_utc_now, normalize_tags @@ -85,7 +85,7 @@ def _ingest_file_from_path( mime_type=mime_type, ) - ref_created, ref_updated = upsert_reference( + ref_created, ref_updated = _legacy_upsert_reference( # wave-3-fixes: replaced in Wave 3 session, asset_id=asset.id, file_path=locator, @@ -176,10 +176,8 @@ def register_output_files( if not os.path.isfile(abs_path): continue try: - if ingest_existing_file( - abs_path, user_metadata=user_metadata, job_id=job_id - ): - registered += 1 + register_output_file_b(abs_path, job_id=job_id) + registered += 1 except Exception: logging.exception("Failed to register output: %s", abs_path) return registered @@ -685,3 +683,62 @@ def create_from_hash( tags=result.tags, created_new=False, ) + + +_verification_queue: list[str] = [] + + +def register_output_file_b(abs_path: str, job_id: str | None = None): + from sqlalchemy import select + + from app.assets import mode + from app.assets.database.models import AssetContent + from app.assets.database.queries.records import ( + create_content, + create_record, + mark_content_missing, + ) + from app.assets.services.snapshot_hash import snapshot_hash + + locator = os.path.abspath(abs_path) + size_bytes, mtime_ns = get_size_and_mtime_ns(locator) + mime_type = mimetypes.guess_type(locator, strict=False)[0] + name, path_tags = get_name_and_tags_from_asset_path(locator) + with create_session() as session: + existing = session.scalars( + select(AssetContent).where( + AssetContent.path == locator, AssetContent.is_missing.is_(False) + ) + ).first() + if existing is not None: + mark_content_missing(session, existing.id) + content_hash = None + if mode.hashing_enabled(): + content_hash = snapshot_hash(locator) + if content_hash is None: + _verification_queue.append(locator) + content = create_content( + session, locator, content_hash, size_bytes, mtime_ns + ) + record = create_record( + session, + content.id, + name, + mime_type=mime_type, + job_id=job_id, + loader_path=compute_loader_path(locator), + tags=path_tags, + ) + session.commit() + record_id = record.id + record_content_id = record.content_id + record_job_id = record.job_id + record_name = record.name + + from types import SimpleNamespace + return SimpleNamespace( + id=record_id, + content_id=record_content_id, + job_id=record_job_id, + name=record_name, + ) diff --git a/app/assets/services/lookup.py b/app/assets/services/lookup.py new file mode 100644 index 000000000..4e10d1d92 --- /dev/null +++ b/app/assets/services/lookup.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import os +from collections.abc import Iterator +from pathlib import Path + +import folder_paths +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.assets import mode +from app.assets.database.models import Asset, AssetContent + + +def is_temp_path(path: str) -> bool: + try: + temp_root = Path(os.path.abspath(folder_paths.get_temp_directory())) + candidate = Path(os.path.abspath(path)) + return candidate.is_relative_to(temp_root) + except OSError: + return False + + +def _stat_consistent(content: AssetContent) -> bool: + """Check file exists and stored stat values are consistent. + + mtime_ns=None means "not yet measured" — always consistent. + size_bytes=0 with mtime_ns=None means "stub row" — always consistent. + """ + try: + stat = os.stat(content.path) + except FileNotFoundError: + return False + if content.mtime_ns is not None and stat.st_mtime_ns != content.mtime_ns: + return False + # Only check size when mtime is also stored (fully enriched row) + if content.mtime_ns is not None and stat.st_size != content.size_bytes: + return False + return True + + +def qualified_content_iterator(session: Session, hash: str) -> Iterator[AssetContent]: + rows = session.scalars( + select(AssetContent) + .where(AssetContent.hash == hash, AssetContent.is_missing.is_(False)) + .order_by(AssetContent.created_at, AssetContent.id) + ) + for row in rows: + if os.path.isfile(row.path) and _stat_consistent(row): + yield row + + +def lookup_for_from_hash(session: Session, hash: str) -> AssetContent | None: + if not mode.hashing_enabled(): + return None + return next( + (content for content in qualified_content_iterator(session, hash) if not is_temp_path(content.path)), + None, + ) + + +def lookup_for_upload_dedup( + session: Session, hash: str, name: str +) -> Asset | AssetContent | None: + if not mode.hashing_enabled(): + return None + first_content = None + for content in qualified_content_iterator(session, hash): + if is_temp_path(content.path): + continue + if first_content is None: + first_content = content + match = session.scalars( + select(Asset).where(Asset.content_id == content.id, Asset.name == name) + ).first() + if match is not None: + return match + return first_content + + +def lookup_for_view(session: Session, hash: str) -> AssetContent | None: + return next(qualified_content_iterator(session, hash), None) diff --git a/app/assets/services/snapshot_hash.py b/app/assets/services/snapshot_hash.py new file mode 100644 index 000000000..53ecea034 --- /dev/null +++ b/app/assets/services/snapshot_hash.py @@ -0,0 +1,43 @@ +"""Stable file snapshot hashing.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass + +from blake3 import blake3 + + +@dataclass(frozen=True, slots=True) +class _Snapshot: + dev: int + ino: int + mtime_ns: int + size: int + + +def _snapshot(stat_result: os.stat_result) -> _Snapshot: + return _Snapshot( + dev=stat_result.st_dev, + ino=stat_result.st_ino, + mtime_ns=stat_result.st_mtime_ns, + size=stat_result.st_size, + ) + + +def snapshot_hash(path: str, chunk_size: int = 8 * 1024 * 1024) -> str | None: + """Return a BLAKE3 digest only when all path and descriptor snapshots match.""" + pre_stat = _snapshot(os.stat(path)) + hasher = blake3() + with open(path, "rb") as file: + open_stat = _snapshot(os.fstat(file.fileno())) + while chunk := file.read(chunk_size): + hasher.update(chunk) + post_hash_stat = _snapshot(os.fstat(file.fileno())) + try: + post_stat = _snapshot(os.stat(path)) + except FileNotFoundError: + return None + if len({pre_stat, open_stat, post_hash_stat, post_stat}) != 1: + return None + return hasher.hexdigest() diff --git a/app/database/db.py b/app/database/db.py index 0aab09a49..118559efc 100644 --- a/app/database/db.py +++ b/app/database/db.py @@ -134,6 +134,7 @@ def _init_memory_db(db_url): def _init_file_db(db_url): """Initialize a file-backed SQLite database using Alembic migrations.""" db_path = get_db_path() + _acquire_file_lock(db_path) db_exists = os.path.exists(db_path) config = get_alembic_config() @@ -168,6 +169,8 @@ def _init_file_db(db_url): try: command.upgrade(config, target_rev) + if backup_path: + os.remove(backup_path) logging.info(f"Database upgraded from {current_rev} to {target_rev}") except Exception as e: if backup_path: @@ -177,11 +180,7 @@ def _init_file_db(db_url): logging.exception("Error upgrading database: ") raise e - # Acquire an OS-level file lock after migrations are complete. - # Alembic uses its own connection, so we must wait until it's done - # before locking — otherwise our own lock blocks the migration. conn.close() - _acquire_file_lock(db_path) global Session Session = sessionmaker(bind=engine) diff --git a/tests-unit/app_test/test_migration_0007.py b/tests-unit/app_test/test_migration_0007.py new file mode 100644 index 000000000..c341d5747 --- /dev/null +++ b/tests-unit/app_test/test_migration_0007.py @@ -0,0 +1,127 @@ +"""Tests specific to migration 0007 (record/content split).""" +import os +import sqlite3 + +import pytest +from alembic import command +from alembic.config import Config + +_BASELINE_0006 = "0006_add_loader_path" + + +def _make_config(db_path: str) -> Config: + root = os.path.join(os.path.dirname(__file__), "../..") + cfg = Config(os.path.abspath(os.path.join(root, "alembic.ini"))) + cfg.set_main_option("script_location", os.path.abspath(os.path.join(root, "alembic_db"))) + cfg.set_main_option("sqlalchemy.url", f"sqlite:///{db_path}") + return cfg + + +@pytest.fixture +def db_at_0006(tmp_path): + db_path = str(tmp_path / "test.db") + cfg = _make_config(db_path) + command.upgrade(cfg, _BASELINE_0006) + yield cfg, db_path + + +def test_0007_upgrade_from_0006(db_at_0006): + """Upgrade from 0006 to head succeeds; new tables present, old gone.""" + cfg, db_path = db_at_0006 + command.upgrade(cfg, "head") + with sqlite3.connect(db_path) as conn: + tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} + assert "assets" in tables + assert "asset_contents" in tables + assert "asset_system_state" in tables + assert "asset_references" not in tables + + +def test_0007_schema_has_expected_columns(db_at_0006): + """After upgrade, key columns exist on new tables.""" + cfg, db_path = db_at_0006 + command.upgrade(cfg, "head") + with sqlite3.connect(db_path) as conn: + asset_cols = {r[1] for r in conn.execute("PRAGMA table_info(assets)")} + content_cols = {r[1] for r in conn.execute("PRAGMA table_info(asset_contents)")} + assert {"id", "content_id", "name", "loader_path", "updated_at", "last_access_time"} <= asset_cols + assert {"id", "hash", "path", "is_missing", "mtime_ns"} <= content_cols + + +def test_0007_downgrade_restores_0006_schema(db_at_0006): + """Downgrade from head back to 0006 restores asset_references.""" + cfg, db_path = db_at_0006 + command.upgrade(cfg, "head") + command.downgrade(cfg, _BASELINE_0006) + with sqlite3.connect(db_path) as conn: + tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} + assert "asset_references" in tables + assert "asset_contents" not in tables + + +def test_0007_invariants_on_migrated_db(db_at_0006): + """After upgrade, schema invariants hold.""" + cfg, db_path = db_at_0006 + command.upgrade(cfg, "head") + with sqlite3.connect(db_path) as conn: + conn.execute("PRAGMA foreign_keys = ON") + # Insert a content row and a record + conn.execute( + "INSERT INTO asset_contents(id, path, is_missing, size_bytes, created_at) " + "VALUES ('c1', '/tmp/f1', 0, 0, '2024-01-01')" + ) + conn.execute( + "INSERT INTO assets(id, content_id, name, created_at, updated_at) " + "VALUES ('a1', 'c1', 'test', '2024-01-01', '2024-01-01')" + ) + conn.commit() + + # Duplicate live path should fail (partial unique) + with pytest.raises(sqlite3.IntegrityError): + conn.execute( + "INSERT INTO asset_contents(id, path, is_missing, size_bytes, created_at) " + "VALUES ('c2', '/tmp/f1', 0, 0, '2024-01-01')" + ) + conn.commit() + conn.rollback() + + # Live + missing same path should succeed + conn.execute( + "INSERT INTO asset_contents(id, path, is_missing, size_bytes, created_at) " + "VALUES ('c3', '/tmp/f1', 1, 0, '2024-01-01')" + ) + conn.commit() + + # Equal-hash rows should coexist (no hash UNIQUE) + conn.execute("UPDATE asset_contents SET hash='abc123' WHERE id='c1'") + conn.execute( + "INSERT INTO asset_contents(id, path, hash, is_missing, size_bytes, created_at) " + "VALUES ('c4', '/tmp/f2', 'abc123', 0, 0, '2024-01-01')" + ) + conn.commit() + + +def test_0007_orm_parity(db_at_0006, tmp_path): + """Base.metadata.create_all produces same table names as alembic upgrade.""" + from sqlalchemy import create_engine, inspect + + from app.database.models import Base + + cfg, db_path = db_at_0006 + command.upgrade(cfg, "head") + with sqlite3.connect(db_path) as conn: + alembic_tables = { + r[0] + for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' " + "AND name NOT LIKE 'alembic%' AND name NOT LIKE 'sqlite%'" + ) + } + + orm_db = str(tmp_path / "orm.db") + engine = create_engine(f"sqlite:///{orm_db}") + Base.metadata.create_all(engine) + inspector = inspect(engine) + orm_tables = set(inspector.get_table_names()) + + assert alembic_tables == orm_tables, f"Mismatch: alembic={alembic_tables}, orm={orm_tables}" diff --git a/tests-unit/assets_test/queries/conftest.py b/tests-unit/assets_test/queries/conftest.py index 4ca0e86a9..18b67c159 100644 --- a/tests-unit/assets_test/queries/conftest.py +++ b/tests-unit/assets_test/queries/conftest.py @@ -5,6 +5,18 @@ from sqlalchemy.orm import Session from app.assets.database.models import Base +@pytest.fixture(scope="session", autouse=True) +def assert_asset_metadata_tables(): + assert set(Base.metadata.tables) == { + "assets", + "asset_contents", + "asset_meta", + "asset_tags", + "tags", + "asset_system_state", + } + + @pytest.fixture def session(): """In-memory SQLite session for fast unit tests.""" diff --git a/tests-unit/assets_test/queries/test_lookup.py b/tests-unit/assets_test/queries/test_lookup.py new file mode 100644 index 000000000..5abbed7e1 --- /dev/null +++ b/tests-unit/assets_test/queries/test_lookup.py @@ -0,0 +1,121 @@ +"""Tests for the B-schema hash lookup policies.""" +from datetime import datetime +from unittest.mock import patch + +import pytest +from sqlalchemy import create_engine, update +from sqlalchemy.orm import Session + +import app.assets.mode as mode_module +from app.assets.database.models import AssetContent +from app.assets.database.queries.records import create_content, create_record +from app.assets.services.lookup import ( + is_temp_path as _is_temp_path, + lookup_for_from_hash, + lookup_for_upload_dedup, + lookup_for_view, +) +from app.database.models import Base + + +@pytest.fixture +def session(tmp_path): + engine = create_engine( + f"sqlite:///{tmp_path}/test.db", connect_args={"check_same_thread": False} + ) + Base.metadata.create_all(engine) + with Session(engine) as sess: + yield sess + + +@pytest.fixture(autouse=True) +def enable_hashing(): + class FakeArgs: + enable_asset_hashing = True + + mode_module.init(FakeArgs()) + yield + mode_module.init(None) + + +def _make_file(tmp_path, name: str, content: bytes = b"bytes") -> str: + p = tmp_path / name + p.write_bytes(content) + return str(p) + + +def test_temp_only_match_from_hash_returns_none(session, tmp_path): + f = _make_file(tmp_path, "f.png") + create_content(session, path=f, hash="abc123") + session.commit() + with patch("app.assets.services.lookup.is_temp_path", return_value=True): + result = lookup_for_from_hash(session, "abc123") + assert result is None + + +def test_temp_only_match_view_still_serves(session, tmp_path): + f = _make_file(tmp_path, "f2.png") + create_content(session, path=f, hash="abc123") + session.commit() + with patch("app.assets.services.lookup.is_temp_path", return_value=True): + result = lookup_for_view(session, "abc123") + assert result is not None + + +def test_sibling_prefix_not_temp(tmp_path): + with patch("folder_paths.get_temp_directory", return_value=str(tmp_path / "temp")): + assert not _is_temp_path(str(tmp_path / "temp-other" / "f.png")) + assert _is_temp_path(str(tmp_path / "temp" / "f.png")) + + +def test_off_mode_from_hash_returns_none(session, tmp_path): + class FakeArgs: + enable_asset_hashing = False + + mode_module.init(FakeArgs()) + f = _make_file(tmp_path, "f3.png") + create_content(session, path=f, hash="abc123") + session.commit() + result = lookup_for_from_hash(session, "abc123") + assert result is None + + +def test_off_mode_dedup_returns_none(session, tmp_path): + class FakeArgs: + enable_asset_hashing = False + + mode_module.init(FakeArgs()) + f = _make_file(tmp_path, "f4.png") + create_content(session, path=f, hash="abc123") + session.commit() + result = lookup_for_upload_dedup(session, "abc123", "test.png") + assert result is None + + +def test_stale_older_newer_live_returns_newer(session, tmp_path): + """Oldest candidate with missing file is skipped; newer live candidate returned.""" + f_newer = _make_file(tmp_path, "newer.png") + old_time = datetime(2020, 1, 1) + new_time = datetime(2024, 1, 1) + c_old = create_content(session, path="/nonexistent/old.png", hash="xyz") + c_new = create_content(session, path=f_newer, hash="xyz") + session.execute(update(AssetContent).where(AssetContent.id == c_old.id).values(created_at=old_time)) + session.execute(update(AssetContent).where(AssetContent.id == c_new.id).values(created_at=new_time)) + session.commit() + + result = lookup_for_from_hash(session, "xyz") + assert result is not None + assert result.id == c_new.id + + +def test_dedup_returns_matching_name_entity(session, tmp_path): + """Upload dedup returns the entity with the matching name, not just any entity.""" + f = _make_file(tmp_path, "match.png") + content = create_content(session, path=f, hash="dup") + record = create_record(session, content_id=content.id, name="match.png") + session.commit() + + result = lookup_for_upload_dedup(session, "dup", "match.png") + assert result is not None + assert hasattr(result, "name"), "Should return an Asset record" + assert result.name == "match.png" diff --git a/tests-unit/assets_test/queries/test_records.py b/tests-unit/assets_test/queries/test_records.py new file mode 100644 index 000000000..b98560c19 --- /dev/null +++ b/tests-unit/assets_test/queries/test_records.py @@ -0,0 +1,105 @@ +"""Tests for the B-schema record/content query layer.""" +import pytest +from sqlalchemy import create_engine, select, update +from sqlalchemy.orm import Session + +from app.assets.database.models import Asset, AssetContent +from app.assets.database.queries.records import ( + create_content, + create_record, + delete_record, + get_record_by_id, + list_records_page, + mark_content_missing, + unset_content_missing, +) +from app.database.models import Base + + +@pytest.fixture +def session(): + engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}) + Base.metadata.create_all(engine) + with Session(engine) as sess: + yield sess + + +def test_listing_includes_missing_by_default(session): + """Missing-content records appear in unfiltered listings.""" + content = create_content(session, path="/tmp/f1") + record = create_record(session, content_id=content.id, name="test") + mark_content_missing(session, content.id) + session.commit() + + results, _ = list_records_page(session) + assert any(r.id == record.id for r in results), "Missing record should appear in default listing" + + +def test_listing_excludes_missing_with_filter(session): + """exclude_tags=['missing'] hides missing-content records.""" + content = create_content(session, path="/tmp/f2") + record = create_record(session, content_id=content.id, name="test2") + mark_content_missing(session, content.id) + session.commit() + + results, _ = list_records_page(session, exclude_tags=["missing"]) + assert not any(r.id == record.id for r in results), "Missing record should be excluded" + + +def test_preview_cleanup_on_delete(session): + """Deleting the last record referencing a preview deletes the preview record but not its content.""" + preview_content = create_content(session, path="/tmp/preview") + preview_record = create_record(session, content_id=preview_content.id, name="preview") + content1 = create_content(session, path="/tmp/f3") + content2 = create_content(session, path="/tmp/f4") + r1 = create_record(session, content_id=content1.id, name="r1") + r2 = create_record(session, content_id=content2.id, name="r2") + + # Both records reference the same preview + session.execute(update(Asset).where(Asset.id == r1.id).values(preview_id=preview_record.id)) + session.execute(update(Asset).where(Asset.id == r2.id).values(preview_id=preview_record.id)) + session.commit() + + # Delete r1 — preview should survive (r2 still references it) + delete_record(session, r1.id) + session.commit() + assert get_record_by_id(session, preview_record.id) is not None, "Preview should survive" + + # Delete r2 — preview should be deleted (no more references) + delete_record(session, r2.id) + session.commit() + assert get_record_by_id(session, preview_record.id) is None, "Preview should be deleted" + + # Preview content row should still exist (D-3 floor) + content_row = session.execute( + select(AssetContent).where(AssetContent.id == preview_content.id) + ).scalar_one_or_none() + assert content_row is not None, "Preview content row should survive (D-3 floor)" + + +def test_concurrent_create_content_same_path(tmp_path): + """Concurrent inserts for the same live path: exactly one live row wins.""" + from sqlalchemy import create_engine + from sqlalchemy.orm import Session + + db_path = str(tmp_path / "concurrent.db") + engine = create_engine(f"sqlite:///{db_path}", connect_args={"check_same_thread": False}) + Base.metadata.create_all(engine) + + with Session(engine) as s1, Session(engine) as s2: + c1 = create_content(s1, path="/tmp/shared") + s1.commit() + # Second session: same path — should get the winner back + c2 = create_content(s2, path="/tmp/shared") + s2.commit() + + with Session(engine) as s: + live_rows = list( + s.execute( + select(AssetContent).where( + AssetContent.path == "/tmp/shared", + AssetContent.is_missing.is_(False), + ) + ).scalars() + ) + assert len(live_rows) == 1, f"Expected exactly one live row, got {len(live_rows)}" diff --git a/tests-unit/assets_test/services/test_ingest_b.py b/tests-unit/assets_test/services/test_ingest_b.py new file mode 100644 index 000000000..efdb319fb --- /dev/null +++ b/tests-unit/assets_test/services/test_ingest_b.py @@ -0,0 +1,89 @@ +"""Tests for the B-schema ingest service (register_output_file_b).""" +import os + +import pytest +from sqlalchemy import select + +import app.assets.mode as mode_module +from app.assets.database.models import Asset, AssetContent + + +@pytest.fixture(autouse=True) +def hashing_off(): + class FakeArgs: + enable_asset_hashing = False + + mode_module.init(FakeArgs()) + yield + mode_module.init(None) + + +def test_new_path_save_off_mode_hash_null(mock_create_session): + """New file registered in off mode: content row has hash=NULL.""" + import folder_paths + from app.assets.services.ingest import register_output_file_b + + output_dir = folder_paths.get_output_directory() + os.makedirs(output_dir, exist_ok=True) + f = os.path.join(output_dir, "test_ingest_b_new.png") + with open(f, "wb") as fh: + fh.write(b"pixels") + + try: + record = register_output_file_b(f, job_id="job1") + record_id = record.id + with mock_create_session() as session: + content = session.execute( + select(AssetContent).where(AssetContent.path == os.path.abspath(f)) + ).scalar_one() + assert content.hash is None + assert content.is_missing is False + asset = session.execute( + select(Asset).where(Asset.id == record_id) + ).scalar_one() + assert asset.job_id == "job1" + finally: + if os.path.exists(f): + os.unlink(f) + + +def test_overwrite_at_live_path_marks_old_missing(mock_create_session): + """Overwriting a live path marks the old content row missing; old record's job_id unchanged.""" + import folder_paths + from app.assets.services.ingest import register_output_file_b + + output_dir = folder_paths.get_output_directory() + os.makedirs(output_dir, exist_ok=True) + f = os.path.join(output_dir, "test_ingest_b_overwrite.png") + + try: + with open(f, "wb") as fh: + fh.write(b"v1") + r1 = register_output_file_b(f, job_id="job1") + old_content_id = r1.content_id + old_record_id = r1.id + + with open(f, "wb") as fh: + fh.write(b"v2") + r2 = register_output_file_b(f, job_id="job2") + new_record_id = r2.id + + with mock_create_session() as session: + old_content = session.execute( + select(AssetContent).where(AssetContent.id == old_content_id) + ).scalar_one() + assert old_content.is_missing is True, "Old content should be marked missing" + + old_record = session.execute( + select(Asset).where(Asset.id == old_record_id) + ).scalar_one() + assert old_record.job_id == "job1", "Old record's job_id must not be mutated" + + new_record = session.execute( + select(Asset).where(Asset.id == new_record_id) + ).scalar_one() + assert new_record.job_id == "job2" + assert new_record_id != old_record_id + finally: + if os.path.exists(f): + os.unlink(f) diff --git a/tests-unit/assets_test/services/test_snapshot_hash.py b/tests-unit/assets_test/services/test_snapshot_hash.py new file mode 100644 index 000000000..d7b67db6a --- /dev/null +++ b/tests-unit/assets_test/services/test_snapshot_hash.py @@ -0,0 +1,75 @@ +import builtins +import os +from collections.abc import Callable +from pathlib import Path + +import pytest +from blake3 import blake3 + +from app.assets.services.snapshot_hash import snapshot_hash + + +class _MutatingReader: + def __init__(self, file, mutate: Callable[[], None]) -> None: + self._file = file + self._mutate = mutate + self._did_mutate = False + + def __enter__(self): + self._file.__enter__() + return self + + def __exit__(self, *args): + return self._file.__exit__(*args) + + def fileno(self) -> int: + return self._file.fileno() + + def read(self, size: int = -1) -> bytes: + result = self._file.read(size) + if not self._did_mutate: + self._did_mutate = True + self._mutate() + return result + + +def test_snapshot_hash_returns_digest_for_quiescent_file(tmp_path: Path) -> None: + payload = b"quiescent" * 1024 + path = tmp_path / "asset.bin" + path.write_bytes(payload) + + assert snapshot_hash(str(path), chunk_size=64) == blake3(payload).hexdigest() + + +@pytest.mark.parametrize("mutation", ["bytes", "replace", "unlink", "truncate", "append"]) +def test_snapshot_hash_returns_none_when_file_drifts( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mutation: str +) -> None: + path = tmp_path / "asset.bin" + path.write_bytes(b"original" * 1024) + original_open = builtins.open + + def mutate() -> None: + match mutation: + case "bytes": + path.write_bytes(b"changed" * 1024) + case "replace": + replacement = tmp_path / "replacement.bin" + replacement.write_bytes(b"replacement") + os.replace(replacement, path) + case "unlink": + path.unlink() + case "truncate": + path.write_bytes(b"") + case "append": + with original_open(path, "ab") as output: + output.write(b"more") + case unreachable: + raise AssertionError(f"unexpected mutation {unreachable}") + + def open_with_mutation(*args, **kwargs): + return _MutatingReader(original_open(*args, **kwargs), mutate) + + monkeypatch.setattr(builtins, "open", open_with_mutation) + + assert snapshot_hash(str(path), chunk_size=64) is None