Refactor/db mv

deployments/api/alembic/versions/b1f2c3d4e5a6_add_resource_state_read_model.py

@@ -0,0 +1,93 @@
+"""add og_field_resource_state read model
+
+Revision ID: b1f2c3d4e5a6
+Revises: 71aa7ef8150b
+Create Date: 2026-09-25 00:00:00.000000
+
+Create the precomputed current-state read model ``og_field_resource_state`` (STIT-766):
+one row per ``(resource, permission_mask)`` holding the coalesced value of every
+``OilGasFieldBase`` field plus a ``provenance`` map.
+
+Schema only. The table is *derived* data and is populated by the application's
+rebuild routine (``stitch.api.db.read_model.state.rebuild_all_resource_state``,
+runnable via ``python -m stitch.api.db.read_model.rebuild``). Run that once after
+upgrading (and after any data restore) to get the performance benefit; ongoing
+writes keep it current in-band thereafter. The read path falls back to live
+coalescing whenever the model can't serve a caller -- including a canonical
+(all-public) caller while the table is unpopulated -- so an un-rebuilt table
+degrades to today's behavior (correct, just not accelerated), never empty or wrong.
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql, sqlite
+
+# revision identifiers, used by Alembic.
+revision = "b1f2c3d4e5a6"
+down_revision = "71aa7ef8150b"
+branch_labels = None
+depends_on = None
+
+_BIGINT = (
+    sa.BigInteger()
+    .with_variant(postgresql.BIGINT(), "postgresql")
+    .with_variant(sqlite.INTEGER(), "sqlite")
+)
+_FLOAT = (
+    sa.Float()
+    .with_variant(postgresql.DOUBLE_PRECISION(), "postgresql")
+    .with_variant(sqlite.REAL(), "sqlite")
+)
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
+
+_FILTER_OPTION_FIELDS = (
+    "basin",
+    "country",
+    "field_status",
+    "primary_hydrocarbon_group",
+    "region",
+    "state_province",
+)
+
+
+def upgrade() -> None:
+    op.create_table(
+        "og_field_resource_state",
+        sa.Column("resource_id", _BIGINT, nullable=False),
+        sa.Column("permission_mask", sa.SmallInteger(), nullable=False),
+        sa.Column("name", sa.String(), nullable=True),
+        sa.Column("country", sa.String(), nullable=True),
+        sa.Column("name_local", sa.String(), nullable=True),
+        sa.Column("state_province", sa.String(), nullable=True),
+        sa.Column("region", sa.String(), nullable=True),
+        sa.Column("basin", sa.String(), nullable=True),
+        sa.Column("reservoir_formation", sa.String(), nullable=True),
+        sa.Column("location_type", sa.String(), nullable=True),
+        sa.Column("production_conventionality", sa.String(), nullable=True),
+        sa.Column("primary_hydrocarbon_group", sa.String(), nullable=True),
+        sa.Column("field_status", sa.String(), nullable=True),
+        sa.Column("latitude", _FLOAT, nullable=True),
+        sa.Column("longitude", _FLOAT, nullable=True),
+        sa.Column("discovery_year", sa.Integer(), nullable=True),
+        sa.Column("production_start_year", sa.Integer(), nullable=True),
+        sa.Column("fid_year", sa.Integer(), nullable=True),
+        sa.Column("owners", _JSON, nullable=True),
+        sa.Column("operators", _JSON, nullable=True),
+        sa.Column("provenance", _JSON, nullable=False),
+        sa.ForeignKeyConstraint(
+            ["resource_id"], ["og_field_resources.id"], ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("resource_id", "permission_mask"),
+    )
+    for field in _FILTER_OPTION_FIELDS:
+        op.create_index(
+            f"ix_og_field_resource_state_{field}",
+            "og_field_resource_state",
+            ["permission_mask", field],
+        )
+
+
+def downgrade() -> None:
+    raise RuntimeError(f"Irreversible migration: {revision}")

deployments/api/alembic/versions/c2d3e4f5a6b7_single_attribute_priority_table.py

@@ -0,0 +1,179 @@
+"""collapse priority tables into og_field_resource_attribute_priority
+
+Revision ID: c2d3e4f5a6b7
+Revises: b1f2c3d4e5a6
+Create Date: 2026-09-25 00:00:00.000000
+
+Replace the two-table priority split -- ``og_field_source_priority`` (global
+defaults) and ``og_field_resource_source_priority`` (per-field overrides) -- with a
+single per-attribute store ``og_field_resource_attribute_priority`` at the value
+grain ``(resource_id, colname, source_pk)`` (STIT-766).
+
+Backfill linearizes the old two-tier order into the single ``priority`` column:
+per ``(resource, colname)``, curated rows (an override existed) take the low
+contiguous positions ordered by override priority, then defaults follow ordered by
+the global default priority and ``source_pk``. ``is_curated`` records which rows
+were curated. Only ACTIVE memberships are backfilled (repointed resources' memberships
+are INACTIVE), matching the coalescing universe.
+
+Then drops the membership FK to ``og_field_source_priority`` and both old tables.
+Irreversible, per repo convention.
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql, sqlite
+
+# revision identifiers, used by Alembic.
+revision = "c2d3e4f5a6b7"
+down_revision = "b1f2c3d4e5a6"
+branch_labels = None
+depends_on = None
+
+_BIGINT = (
+    sa.BigInteger()
+    .with_variant(postgresql.BIGINT(), "postgresql")
+    .with_variant(sqlite.INTEGER(), "sqlite")
+)
+
+_ATTRIBUTE_NAMES = (
+    "name",
+    "country",
+    "name_local",
+    "state_province",
+    "region",
+    "basin",
+    "reservoir_formation",
+    "location_type",
+    "production_conventionality",
+    "primary_hydrocarbon_group",
+    "field_status",
+    "latitude",
+    "longitude",
+    "discovery_year",
+    "production_start_year",
+    "fid_year",
+    "owners",
+    "operators",
+)
+
+_BACKFILL_SQL = """
+INSERT INTO og_field_resource_attribute_priority
+    (resource_id, colname, source_pk, source, priority, is_curated,
+     created, updated, created_by_id, last_updated_by_id)
+SELECT
+    resource_id,
+    colname,
+    source_pk,
+    source,
+    ROW_NUMBER() OVER (
+        PARTITION BY resource_id, colname
+        ORDER BY (override_priority IS NULL), override_priority,
+                 default_priority, source_pk
+    ) - 1 AS priority,
+    (override_priority IS NOT NULL) AS is_curated,
+    CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, created_by_id, last_updated_by_id
+FROM (
+    SELECT
+        m.resource_id,
+        v.colname,
+        m.source_pk,
+        m.source,
+        o.priority AS override_priority,
+        p.priority AS default_priority,
+        m.created_by_id,
+        m.last_updated_by_id
+    FROM og_field_memberships m
+    JOIN oil_gas_field_source_values v ON v.source_pk = m.source_pk
+    JOIN og_field_source_priority p ON p.source = m.source
+    LEFT JOIN og_field_resource_source_priority o
+        ON o.resource_id = m.resource_id
+       AND o.source_pk = m.source_pk
+       AND o.colname = v.colname
+    WHERE m.status = 'ACTIVE'
+) sub
+"""
+
+# Every foreign key (from any table) that references og_field_source_priority, so
+# they can be dropped before the table itself. Matching on the *referenced* table
+# (confrelid) is robust to DB-generated constraint names. og_field_memberships is
+# the only expected referencer once og_field_resource_source_priority is dropped.
+_DISCOVER_FKS_TO_PRIORITY_SQL = """
+SELECT c.conname, t.relname AS table_name
+FROM pg_constraint c
+JOIN pg_class t ON t.oid = c.conrelid
+JOIN pg_class rt ON rt.oid = c.confrelid
+WHERE rt.relname = 'og_field_source_priority'
+  AND c.contype = 'f'
+"""
+
+
+def upgrade() -> None:
+    op.create_table(
+        "og_field_resource_attribute_priority",
+        sa.Column("resource_id", _BIGINT, nullable=False),
+        sa.Column("colname", sa.String(length=50), nullable=False),
+        sa.Column("source_pk", _BIGINT, nullable=False),
+        sa.Column("source", sa.String(length=10), nullable=False),
+        sa.Column("priority", sa.Integer(), nullable=False),
+        sa.Column("is_curated", sa.Boolean(), nullable=False),
+        sa.Column(
+            "created",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+            nullable=False,
+        ),
+        sa.Column("created_by_id", sa.Integer(), nullable=False),
+        sa.Column("last_updated_by_id", sa.Integer(), nullable=False),
+        sa.CheckConstraint("priority >= 0", name="ck_attr_priority_nonnegative"),
+        sa.CheckConstraint(
+            "colname IN (" + ", ".join(f"'{n}'" for n in _ATTRIBUTE_NAMES) + ")",
+            name="ck_attr_priority_colname",
+        ),
+        sa.ForeignKeyConstraint(
+            ["resource_id"], ["og_field_resources.id"], ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(
+            ["source_pk", "colname"],
+            [
+                "oil_gas_field_source_values.source_pk",
+                "oil_gas_field_source_values.colname",
+            ],
+            name="fk_attr_priority_source_value",
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(["created_by_id"], ["users.id"]),
+        sa.ForeignKeyConstraint(["last_updated_by_id"], ["users.id"]),
+        sa.PrimaryKeyConstraint("resource_id", "colname", "source_pk"),
+        sa.UniqueConstraint(
+            "resource_id", "colname", "priority", name="uq_attr_priority_position"
+        ),
+    )
+
+    bind = op.get_bind()
+    bind.execute(sa.text(_BACKFILL_SQL))
+
+    # Override table first (it FKs the default table), then the default table.
+    op.drop_table("og_field_resource_source_priority")
+
+    # Drop any remaining FK referencing og_field_source_priority (DB-generated
+    # names, so discover them) before dropping the table itself.
+    if bind.dialect.name == "postgresql":
+        for conname, table_name in bind.execute(
+            sa.text(_DISCOVER_FKS_TO_PRIORITY_SQL)
+        ).all():
+            op.drop_constraint(conname, table_name, type_="foreignkey")
+
+    op.drop_table("og_field_source_priority")
+
+
+def downgrade() -> None:
+    raise RuntimeError(f"Irreversible migration: {revision}")

deployments/api/alembic/versions/d4e5f6a7b8c9_membership_uniqueness_and_priority_fk.py

@@ -0,0 +1,70 @@
+"""enforce membership uniqueness and priority->membership FK
+
+Revision ID: d4e5f6a7b8c9
+Revises: c2d3e4f5a6b7
+Create Date: 2026-09-28 00:00:00.000000
+
+Close the integrity gap on og_field_resource_attribute_priority: its rows proved a
+source has a value for a field, but not that the source is actually attached to the
+resource. Enforce it structurally (STIT-766, Copilot review):
+
+1. Dedup og_field_memberships to one row per (resource_id, source_pk), keeping the
+   ACTIVE row when present, else the lowest id. (The write paths now dedup on
+   attach and merge; this cleans any pre-existing duplicates.)
+2. Add UNIQUE (resource_id, source_pk) on og_field_memberships.
+3. Add composite FK og_field_resource_attribute_priority (resource_id, source_pk)
+   -> og_field_memberships (resource_id, source_pk), so a priority row can only
+   reference a source that is a real member of its resource.
+
+Irreversible, per repo convention.
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = "d4e5f6a7b8c9"
+down_revision = "c2d3e4f5a6b7"
+branch_labels = None
+depends_on = None
+
+# Keep exactly one membership per (resource_id, source_pk): a row is removed when a
+# strictly-preferred sibling exists (ACTIVE beats non-ACTIVE; within the same
+# active-ness, the lowest id wins).
+_DEDUP_MEMBERSHIPS_SQL = """
+DELETE FROM og_field_memberships m
+WHERE EXISTS (
+    SELECT 1
+    FROM og_field_memberships k
+    WHERE k.resource_id = m.resource_id
+      AND k.source_pk = m.source_pk
+      AND k.id <> m.id
+      AND (
+          (k.status = 'ACTIVE' AND m.status <> 'ACTIVE')
+          OR ((k.status = 'ACTIVE') = (m.status = 'ACTIVE') AND k.id < m.id)
+      )
+)
+"""
+
+
+def upgrade() -> None:
+    op.execute(sa.text(_DEDUP_MEMBERSHIPS_SQL))
+    op.create_unique_constraint(
+        "uq_membership_resource_source",
+        "og_field_memberships",
+        ["resource_id", "source_pk"],
+    )
+    op.create_foreign_key(
+        "fk_attr_priority_membership",
+        "og_field_resource_attribute_priority",
+        "og_field_memberships",
+        ["resource_id", "source_pk"],
+        ["resource_id", "source_pk"],
+        ondelete="CASCADE",
+    )
+
+
+def downgrade() -> None:
+    raise RuntimeError(f"Irreversible migration: {revision}")

deployments/api/alembic/versions/e6f7a8b9c0d1_add_membership_resource_index.py

@@ -0,0 +1,36 @@
+"""add og_field_memberships (resource_id, status) index
+
+Revision ID: e6f7a8b9c0d1
+Revises: d4e5f6a7b8c9
+Create Date: 2026-10-08 00:00:00.000000
+
+`og_field_memberships` had only its primary-key index, so every read that scans a
+resource's active memberships (the single-resource `source_data` join and the
+coalescing universe filter, both keyed on `resource_id` + `status = ACTIVE`) fell
+back to a full-table seq scan. Add a composite index on `(resource_id, status)`.
+
+A general read win independent of the read-model work (it speeds the live
+detail/source_data path too). Irreversible, per repo convention.
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "e6f7a8b9c0d1"
+down_revision = "d4e5f6a7b8c9"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_index(
+        "ix_membership_resource_active",
+        "og_field_memberships",
+        ["resource_id", "status"],
+    )
+
+
+def downgrade() -> None:
+    raise RuntimeError(f"Irreversible migration: {revision}")