"""C6 tile identity + LRU + freshness rules — strictly additive on 0001_initial. Per ``_docs/02_tasks/todo/AZ-304_c6_postgres_schema.md`` and ``_docs/02_document/data_model.md`` §§ 6.1 / 6.3. The migration: - adds the deterministic-identity columns (``tile_uuid``, ``location_hash``); - adds the content-hash chain column (``content_sha256``); - adds the LRU + disk-budget bookkeeping columns (``disk_bytes``, ``accessed_at``, ``uploaded_at``); - adds the per-flight natural-key UNIQUE expression index (``idx_tiles_natural_key``) over the COALESCE-zero-uuid form so two ``onboard_ingest`` rows for the same cell from different flights coexist while two ``googlemaps`` rows (both flight_id NULL) cannot; - adds four new btree indices on ``tiles`` (location_hash, accessed_at, pending_upload partial, flight_captured partial); - widens the ``tiles.freshness_status`` CHECK to accept the UNION of the AZ-263 legacy vocabulary AND the AZ-303 ``FreshnessLabel`` vocabulary; - adds four NULLable bbox columns to ``sector_classifications``; - creates the new ``tile_freshness_rules`` table seeded with the per-classification thresholds (6 months active_conflict / reject, 12 months stable_rear / downgrade). The migration assumes ``tiles`` is empty at apply time (greenfield); NOT-NULL additive columns without server defaults rely on that. Reverse ``downgrade()`` drops every addition and restores the AZ-263 ``freshness_status`` CHECK to its original ``('fresh','stale_warn','stale_reject')`` vocabulary. Downgrade is destructive for any rows holding the new column values and is documented operator-only. Revision ID: 0002_c6_tile_identity_and_lru Revises: 0001_initial Create Date: 2026-05-12 """ from __future__ import annotations from collections.abc import Sequence from datetime import datetime, timezone import sqlalchemy as sa from alembic import op revision: str = "0002_c6_tile_identity_and_lru" down_revision: str | None = "0001_initial" branch_labels: str | Sequence[str] | None = None depends_on: str | Sequence[str] | None = None _NATURAL_KEY_INDEX_SQL = """ CREATE UNIQUE INDEX idx_tiles_natural_key ON tiles ( zoom_level, tile_x, tile_y, tile_size_meters, source, COALESCE(flight_id, '00000000-0000-0000-0000-000000000000'::uuid) ) """ _FRESHNESS_STATUS_LEGACY = "freshness_status IN ('fresh','stale_warn','stale_reject')" _FRESHNESS_STATUS_UNION = ( "freshness_status IN (" "'fresh','stale_warn','stale_reject'," "'stale_active_conflict','stale_rear','downgraded'" ")" ) def upgrade() -> None: # tiles -- additive columns ------------------------------------------------ op.add_column( "tiles", sa.Column( "tile_uuid", sa.dialects.postgresql.UUID(as_uuid=True), nullable=False, unique=True, ), ) op.add_column( "tiles", sa.Column( "location_hash", sa.dialects.postgresql.UUID(as_uuid=True), nullable=False, ), ) op.add_column( "tiles", sa.Column("content_sha256", sa.Text(), nullable=False), ) op.create_check_constraint( "ck_tiles_content_sha256_len", "tiles", "length(content_sha256) = 64", ) op.add_column( "tiles", sa.Column("disk_bytes", sa.BigInteger(), nullable=False), ) op.create_check_constraint( "ck_tiles_disk_bytes_nonneg", "tiles", "disk_bytes >= 0", ) op.add_column( "tiles", sa.Column( "accessed_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()"), ), ) op.add_column( "tiles", sa.Column("uploaded_at", sa.DateTime(timezone=True), nullable=True), ) # tiles -- freshness_status CHECK widening (drop + recreate) --------------- op.drop_constraint("ck_tiles_freshness_status", "tiles", type_="check") op.create_check_constraint( "ck_tiles_freshness_status", "tiles", _FRESHNESS_STATUS_UNION, ) # tiles -- new indices ----------------------------------------------------- op.execute(_NATURAL_KEY_INDEX_SQL) op.create_index("idx_tiles_location_hash", "tiles", ["location_hash"]) op.create_index("idx_tiles_accessed_at", "tiles", ["accessed_at"]) op.create_index( "idx_tiles_pending_upload", "tiles", ["uploaded_at"], postgresql_where=sa.text("source = 'onboard_ingest' AND uploaded_at IS NULL"), ) op.create_index( "idx_tiles_flight_captured", "tiles", ["flight_id", "capture_timestamp"], postgresql_where=sa.text("flight_id IS NOT NULL"), ) # sector_classifications -- nullable bbox additions ------------------------ op.add_column( "sector_classifications", sa.Column("min_lat", sa.Float(precision=53), nullable=True), ) op.add_column( "sector_classifications", sa.Column("min_lon", sa.Float(precision=53), nullable=True), ) op.add_column( "sector_classifications", sa.Column("max_lat", sa.Float(precision=53), nullable=True), ) op.add_column( "sector_classifications", sa.Column("max_lon", sa.Float(precision=53), nullable=True), ) # tile_freshness_rules -- new table with seed rows ------------------------- rules_table = op.create_table( "tile_freshness_rules", sa.Column("classification", sa.Text(), primary_key=True), sa.Column("max_age_seconds", sa.BigInteger(), nullable=False), sa.Column("action", sa.Text(), nullable=False), sa.Column( "set_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()"), ), sa.CheckConstraint("action IN ('reject','downgrade')", name="ck_tfr_action"), sa.CheckConstraint("max_age_seconds > 0", name="ck_tfr_max_age_pos"), ) seed_at = datetime(2026, 5, 12, tzinfo=timezone.utc) op.bulk_insert( rules_table, [ { "classification": "active_conflict", "max_age_seconds": 6 * 30 * 86400, # 6 months ≈ 15552000 s "action": "reject", "set_at": seed_at, }, { "classification": "stable_rear", "max_age_seconds": 12 * 30 * 86400, # 12 months ≈ 31104000 s "action": "downgrade", "set_at": seed_at, }, ], ) def downgrade() -> None: op.drop_table("tile_freshness_rules") op.drop_column("sector_classifications", "max_lon") op.drop_column("sector_classifications", "max_lat") op.drop_column("sector_classifications", "min_lon") op.drop_column("sector_classifications", "min_lat") op.drop_index("idx_tiles_flight_captured", table_name="tiles") op.drop_index("idx_tiles_pending_upload", table_name="tiles") op.drop_index("idx_tiles_accessed_at", table_name="tiles") op.drop_index("idx_tiles_location_hash", table_name="tiles") op.execute("DROP INDEX IF EXISTS idx_tiles_natural_key") op.drop_constraint("ck_tiles_freshness_status", "tiles", type_="check") op.create_check_constraint( "ck_tiles_freshness_status", "tiles", _FRESHNESS_STATUS_LEGACY, ) op.drop_column("tiles", "uploaded_at") op.drop_column("tiles", "accessed_at") op.drop_constraint("ck_tiles_disk_bytes_nonneg", "tiles", type_="check") op.drop_column("tiles", "disk_bytes") op.drop_constraint("ck_tiles_content_sha256_len", "tiles", type_="check") op.drop_column("tiles", "content_sha256") op.drop_column("tiles", "location_hash") op.drop_column("tiles", "tile_uuid")