feat(14-02): add migration 0007 and ORM models for cloud analysis cache schema

- 0007_cloud_analysis_cache.py: new Alembic migration adding 4 Phase 14 tables - cloud_byte_cache_entries: durable MinIO cache with version_key uniqueness, pin_count/active_job_count guards, LRU eviction indexes - cloud_analysis_jobs: batch job rows (scope_kind, status, aggregate counters, provider_bytes_estimate, failure_behavior) - cloud_analysis_job_items: per-item idempotency rows (status vocabulary, controlled error fields, cache_entry_id, retry_count, version_key fingerprint) - user_analysis_settings: per-user preferences (progress_detail, failure_behavior, cloud_cache_limit_bytes) - Reversible downgrade drops Phase 14 objects only - db/models.py: CloudByteCacheEntry, CloudAnalysisJob, CloudAnalysisJobItem, UserAnalysisSettings ORM models mirroring migration
2026-06-23 15:06:24 +02:00
parent 493d348ba6
commit fd76d06b3a
2 changed files with 588 additions and 0 deletions
@@ -0,0 +1,345 @@
+"""Add cloud_byte_cache_entries, cloud_analysis_jobs, cloud_analysis_job_items, and user analysis settings.
+
+Revision ID: 0007
+Revises: 0006
+Create Date: 2026-06-23
+
+Changes:
+  1. cloud_byte_cache_entries: durable per-item provider byte cache backed by MinIO.
+     Tracks version key, object key, content type, size, pin count, active job count,
+     and LRU access timestamps.  Entries are owner-scoped (user_id + connection_id +
+     cloud_item_id + version_key).  quota.used_bytes is incremented on retain and
+     decremented on eviction via the atomic UPDATE pattern.
+
+  2. cloud_analysis_jobs: user-visible analysis batch rows for file, selection, folder,
+     or connection scope.  Aggregate counts (total, queued, downloading, extracting,
+     classifying, indexed, already_current, skipped, cancelled, failed, unsupported) and
+     provider bytes estimate are updated as workers process items.
+
+  3. cloud_analysis_job_items: per-item rows that drive idempotent processing.
+     Status, error code, cache entry linkage, and retry counter.  Unique constraint on
+     (job_id, cloud_item_id) prevents duplicate item rows per job.
+
+  4. user_analysis_settings: one row per user with analysis preferences.
+     progress_detail, failure_behavior, and cloud_cache_limit_bytes.  Does NOT create
+     a second settings authority — this is user-scoped, separate from system_settings.
+
+Design notes:
+  - Ownership boundary: every row has user_id.  Eviction and status queries are always
+    scoped by user_id before any other predicate.
+  - version_key uniqueness: (user_id, connection_id, cloud_item_id, version_key) is
+    the idempotency key for cache entries.
+  - Downgrade drops only Phase 14 objects in reverse-creation order.
+  - D-18 compliance: cloud provider byte sizes flow only to cache entries and quota
+    accounting, never to quotas.used_bytes via cloud_items.provider_size.
+"""
+from __future__ import annotations
+
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import UUID as PG_UUID
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0007"
+down_revision = "0006"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ── 1. cloud_byte_cache_entries ────────────────────────────────────────────
+
+    op.create_table(
+        "cloud_byte_cache_entries",
+        sa.Column(
+            "id",
+            PG_UUID(as_uuid=True),
+            primary_key=True,
+            server_default=sa.text("gen_random_uuid()"),
+        ),
+        sa.Column(
+            "user_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "connection_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("cloud_connections.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "cloud_item_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("cloud_items.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        # Snapshot of the provider item id — for audit/debug without joins
+        sa.Column("provider_item_id", sa.Text, nullable=False),
+        # Derived from etag / version / metadata fingerprint / content hash (in precedence order)
+        sa.Column("version_key", sa.Text, nullable=False),
+        # Private MinIO object key (UUID-based) — never exposed in API responses
+        sa.Column("object_key", sa.Text, nullable=False),
+        sa.Column("content_type", sa.Text, nullable=True),
+        # Actual byte count stored in MinIO — flows to quota.used_bytes while retained
+        sa.Column("size_bytes", sa.BigInteger, nullable=False),
+        # Optional content hash computed ONLY after bytes are already hydrated
+        sa.Column("content_hash", sa.Text, nullable=True),
+        # pin_count > 0 means a preview or download lease is active — must not be evicted
+        sa.Column("pin_count", sa.Integer, nullable=False, server_default="0"),
+        # active_job_count > 0 means a Celery analysis task is using these bytes
+        sa.Column("active_job_count", sa.Integer, nullable=False, server_default="0"),
+        # LRU eviction timestamps
+        sa.Column("last_accessed_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column("evicted_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        # Idempotency key: same item + same version must reuse the same cache entry
+        sa.UniqueConstraint(
+            "user_id",
+            "connection_id",
+            "cloud_item_id",
+            "version_key",
+            name="uq_cache_entry_user_connection_item_version",
+        ),
+    )
+
+    # LRU eviction scan: scan only non-evicted entries for a user ordered by last_accessed_at
+    op.create_index(
+        "ix_cache_entries_user_evicted_accessed",
+        "cloud_byte_cache_entries",
+        ["user_id", "evicted_at", "last_accessed_at"],
+    )
+    # Eviction guard: quick check for active pins / jobs before eviction candidate selection
+    op.create_index(
+        "ix_cache_entries_user_active",
+        "cloud_byte_cache_entries",
+        ["user_id", "pin_count", "active_job_count", "last_accessed_at"],
+    )
+
+    # ── 2. cloud_analysis_jobs ─────────────────────────────────────────────────
+
+    op.create_table(
+        "cloud_analysis_jobs",
+        sa.Column(
+            "id",
+            PG_UUID(as_uuid=True),
+            primary_key=True,
+            server_default=sa.text("gen_random_uuid()"),
+        ),
+        sa.Column(
+            "user_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "connection_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("cloud_connections.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        # "file" | "selection" | "folder" | "connection"
+        sa.Column("scope_kind", sa.String(16), nullable=False),
+        # Provider item id of the root scope (null for connection-wide scope)
+        sa.Column("scope_ref", sa.Text, nullable=True),
+        sa.Column("recursive", sa.Boolean, nullable=False, server_default="false"),
+        # "queued" | "running" | "paused" | "completed" | "cancelled" | "failed"
+        sa.Column("status", sa.String(16), nullable=False, server_default="queued"),
+        # Aggregate item counters — kept in sync by Celery workers
+        sa.Column("total_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("queued_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("downloading_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("extracting_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("classifying_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("indexed_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("already_current_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("skipped_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("cancelled_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("failed_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("unsupported_count", sa.Integer, nullable=False, server_default="0"),
+        # Provider-reported byte estimate (never used for quota)
+        sa.Column("provider_bytes_estimate", sa.BigInteger, nullable=False, server_default="0"),
+        # "pause_batch" | "continue_item" (default is pause_batch per D-11)
+        sa.Column("failure_behavior", sa.String(16), nullable=False, server_default="pause_batch"),
+        sa.Column("started_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column("finished_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+    )
+
+    op.create_index("ix_analysis_jobs_user_id", "cloud_analysis_jobs", ["user_id"])
+    op.create_index(
+        "ix_analysis_jobs_user_status",
+        "cloud_analysis_jobs",
+        ["user_id", "status"],
+    )
+
+    # ── 3. cloud_analysis_job_items ────────────────────────────────────────────
+
+    op.create_table(
+        "cloud_analysis_job_items",
+        sa.Column(
+            "id",
+            PG_UUID(as_uuid=True),
+            primary_key=True,
+            server_default=sa.text("gen_random_uuid()"),
+        ),
+        sa.Column(
+            "job_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("cloud_analysis_jobs.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "user_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "connection_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("cloud_connections.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "cloud_item_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("cloud_items.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        # Snapshot for idempotency and debug — avoids joins in hot paths
+        sa.Column("provider_item_id", sa.Text, nullable=False),
+        # version_key at enqueue time — compared to re-resolved key at processing time
+        sa.Column("version_key", sa.Text, nullable=True),
+        # Metadata fingerprint at enqueue time — "item_id:size:modified_at:content_type"
+        sa.Column("fingerprint", sa.Text, nullable=True),
+        # "queued" | "downloading" | "extracting" | "classifying" | "indexed"
+        # | "already_current" | "cancelled" | "failed" | "unsupported" | "stale"
+        sa.Column("status", sa.String(16), nullable=False, server_default="queued"),
+        # Controlled error strings only — never raw provider/exception text
+        sa.Column("error_code", sa.String(64), nullable=True),
+        sa.Column("error_message", sa.Text, nullable=True),
+        # Link to the cache entry used during processing (null until bytes hydrated)
+        sa.Column(
+            "cache_entry_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("cloud_byte_cache_entries.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column("retry_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("started_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column("finished_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        # Each item appears once per job
+        sa.UniqueConstraint(
+            "job_id",
+            "cloud_item_id",
+            name="uq_job_items_job_cloud_item",
+        ),
+    )
+
+    # Duplicate-current check: fast lookup by (user_id, cloud_item_id, version_key)
+    # across all jobs to detect already-current items without byte download
+    op.create_index(
+        "ix_job_items_user_item_version",
+        "cloud_analysis_job_items",
+        ["user_id", "cloud_item_id", "version_key"],
+    )
+    op.create_index("ix_job_items_job_id", "cloud_analysis_job_items", ["job_id"])
+
+    # ── 4. user_analysis_settings ──────────────────────────────────────────────
+
+    op.create_table(
+        "user_analysis_settings",
+        sa.Column(
+            "user_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+        # "simple" (default) | "detailed"
+        sa.Column(
+            "analysis_progress_detail",
+            sa.String(16),
+            nullable=False,
+            server_default="simple",
+        ),
+        # "pause_batch" (default) | "continue_item"
+        sa.Column(
+            "analysis_failure_behavior",
+            sa.String(16),
+            nullable=False,
+            server_default="pause_batch",
+        ),
+        # User-preferred cache byte limit (capped by tier/admin maximum in service layer)
+        # Default: 512 MB
+        sa.Column(
+            "cloud_cache_limit_bytes",
+            sa.BigInteger,
+            nullable=False,
+            server_default=str(512 * 1024 * 1024),
+        ),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+    )
+
+
+def downgrade() -> None:
+    # Drop in reverse-creation order.  Phase 14 objects only — no Phase 12/13 changes.
+    op.drop_table("user_analysis_settings")
+
+    op.drop_index("ix_job_items_job_id", table_name="cloud_analysis_job_items")
+    op.drop_index("ix_job_items_user_item_version", table_name="cloud_analysis_job_items")
+    op.drop_table("cloud_analysis_job_items")
+
+    op.drop_index("ix_analysis_jobs_user_status", table_name="cloud_analysis_jobs")
+    op.drop_index("ix_analysis_jobs_user_id", table_name="cloud_analysis_jobs")
+    op.drop_table("cloud_analysis_jobs")
+
+    op.drop_index("ix_cache_entries_user_active", table_name="cloud_byte_cache_entries")
+    op.drop_index("ix_cache_entries_user_evicted_accessed", table_name="cloud_byte_cache_entries")
+    op.drop_table("cloud_byte_cache_entries")