diff --git a/deployments/api/alembic/versions/b1f2c3d4e5a6_add_resource_state_read_model.py b/deployments/api/alembic/versions/b1f2c3d4e5a6_add_resource_state_read_model.py new file mode 100644 index 00000000..517acc92 --- /dev/null +++ b/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}") diff --git a/deployments/api/alembic/versions/c2d3e4f5a6b7_single_attribute_priority_table.py b/deployments/api/alembic/versions/c2d3e4f5a6b7_single_attribute_priority_table.py new file mode 100644 index 00000000..d426bc58 --- /dev/null +++ b/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}") diff --git a/deployments/api/alembic/versions/d4e5f6a7b8c9_membership_uniqueness_and_priority_fk.py b/deployments/api/alembic/versions/d4e5f6a7b8c9_membership_uniqueness_and_priority_fk.py new file mode 100644 index 00000000..75554519 --- /dev/null +++ b/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}") diff --git a/deployments/api/alembic/versions/e6f7a8b9c0d1_add_membership_resource_index.py b/deployments/api/alembic/versions/e6f7a8b9c0d1_add_membership_resource_index.py new file mode 100644 index 00000000..70481599 --- /dev/null +++ b/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}") diff --git a/deployments/api/src/stitch/api/db/merge_candidate_actions.py b/deployments/api/src/stitch/api/db/merge_candidate_actions.py index b93a712c..e9dc7ebd 100644 --- a/deployments/api/src/stitch/api/db/merge_candidate_actions.py +++ b/deployments/api/src/stitch/api/db/merge_candidate_actions.py @@ -24,14 +24,13 @@ MergeCandidateView, ) from stitch.api.observability.context import named_query -from stitch.ogsi.model import OGFieldSource +from stitch.ogsi.model import SOURCE_PRIORITY, OGFieldSource from stitch.ogsi.model.og_field import OilGasFieldBase from stitch.ogsi.model.types import OGSISrcKey from .model import ( MergeCandidateItemModel, MergeCandidateModel, - OGFieldSourcePriority, ResourceModel, ) from .og_field_resource_actions import apply_resource_merge @@ -161,15 +160,14 @@ def _build_comparison( async def _default_source_priority(session: AsyncSession) -> dict[str, int]: - """Global default source ordering (``source`` key -> priority, lower wins). + """Global default source ordering (``source`` key -> rank, lower wins). A merge resets the merged resource to this default order, so the comparison - ranks sources by it rather than by any per-resource override. + ranks sources by it rather than by any per-resource curation. Derived from the + canonical ``SOURCE_PRIORITY`` (the priority lookup table is gone); ``session`` + is kept for call-site symmetry. """ - rows = await session.execute( - select(OGFieldSourcePriority.source, OGFieldSourcePriority.priority) - ) - return {source: priority for source, priority in rows.all()} + return {source: rank for rank, source in enumerate(SOURCE_PRIORITY)} def _candidate_to_detail_view( @@ -240,8 +238,7 @@ async def get_merge_candidate( # `status` compares the resources' coalesced values; `values` lists every # contributing source tagged with the resource it's attached to, ranked by # the default source order (winner-first). - with named_query("merge_candidates.detail.default_priority"): - default_priority = await _default_source_priority(session) + default_priority = await _default_source_priority(session) fallback_priority = max(default_priority.values(), default=0) + 1 sources_with_priority = [ (rid, source, default_priority.get(source.source, fallback_priority)) diff --git a/deployments/api/src/stitch/api/db/model/__init__.py b/deployments/api/src/stitch/api/db/model/__init__.py index 55cd212a..941ebbd8 100644 --- a/deployments/api/src/stitch/api/db/model/__init__.py +++ b/deployments/api/src/stitch/api/db/model/__init__.py @@ -1,9 +1,9 @@ from .common import Base as StitchBase -from .og_field_source_priority import OGFieldSourcePriority -from .og_field_resource_source_priority import OGFieldResourceSourcePriority +from .og_field_resource_attribute_priority import OGFieldResourceAttributePriority from .oil_gas_field_source import OilGasFieldSourceModel from .oil_gas_field_source_value import OilGasFieldSourceValueModel from .membership import MembershipModel, MembershipStatus +from .og_field_resource_state import OGFieldResourceState from .resource import ResourceModel from .merge_candidate import MergeCandidateItemModel, MergeCandidateModel from .user import User as UserModel @@ -11,8 +11,8 @@ __all__ = [ "MembershipModel", "MembershipStatus", - "OGFieldSourcePriority", - "OGFieldResourceSourcePriority", + "OGFieldResourceAttributePriority", + "OGFieldResourceState", "OilGasFieldSourceModel", "OilGasFieldSourceValueModel", "MergeCandidateItemModel", diff --git a/deployments/api/src/stitch/api/db/model/membership.py b/deployments/api/src/stitch/api/db/model/membership.py index a31f73a6..0070eaff 100644 --- a/deployments/api/src/stitch/api/db/model/membership.py +++ b/deployments/api/src/stitch/api/db/model/membership.py @@ -2,7 +2,7 @@ from enum import StrEnum -from sqlalchemy import ForeignKey, String +from sqlalchemy import ForeignKey, Index, String, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column from stitch.ogsi.model.types import OGSISrcKey @@ -28,11 +28,10 @@ class MembershipModel(TimestampMixin, UserAuditMixin, Base): resource_id: Mapped[int] = mapped_column( ForeignKey("og_field_resources.id"), nullable=False ) - source: Mapped[OGSISrcKey] = mapped_column( - String(10), - ForeignKey("og_field_source_priority.source"), - nullable=False, - ) + # Source key of the membership. Validity is guaranteed by the OGSISrcKey enum + # and the write path; there is no longer an og_field_source_priority table to + # FK against (source ordering moved to og_field_resource_attribute_priority). + source: Mapped[OGSISrcKey] = mapped_column(String(10), nullable=False) source_pk: Mapped[int] = mapped_column( ForeignKey("oil_gas_field_sources.id"), nullable=False ) @@ -40,6 +39,20 @@ class MembershipModel(TimestampMixin, UserAuditMixin, Base): default=MembershipStatus.ACTIVE, nullable=False ) + __table_args__ = ( + # A source record is attached to a resource at most once. The write paths + # dedup (attach + merge) to uphold this; it also lets + # og_field_resource_attribute_priority FK against (resource_id, source_pk), + # guaranteeing a priority row's source is a real member of its resource. + UniqueConstraint( + "resource_id", "source_pk", name="uq_membership_resource_source" + ), + # Reads constantly scan a resource's active memberships (the source_data + # join and the coalescing universe filter on resource_id + status == ACTIVE); + # without this only the PK is indexed, forcing a full-table seq scan. + Index("ix_membership_resource_active", "resource_id", "status"), + ) + @classmethod def create( cls, diff --git a/deployments/api/src/stitch/api/db/model/og_field_resource_attribute_priority.py b/deployments/api/src/stitch/api/db/model/og_field_resource_attribute_priority.py new file mode 100644 index 00000000..21aef80f --- /dev/null +++ b/deployments/api/src/stitch/api/db/model/og_field_resource_attribute_priority.py @@ -0,0 +1,118 @@ +"""Single per-attribute source priority store (STIT-766). + +Replaces the two-table split of global defaults (``og_field_source_priority``) and +per-field overrides (``og_field_resource_source_priority``) with one table at the +value grain ``(resource_id, colname, source_pk)`` -- one row per source record that +carries a value for a field of a resource. + +Ranking collapses to a single column: ``priority`` (0-based, lower wins), unique +within each ``(resource, colname)``. The old two-tier order is *linearized* into +it -- curated rows (``is_curated``) occupy the low, contiguous positions and always +outrank the default rows that follow, which are ordered by the global +``SOURCE_PRIORITY`` rank (see ``db.source_priority``) then ``source_pk``. So the +coalescing query is just ``ORDER BY priority`` with no NULLS-LAST tiering. + +``is_curated`` marks rows a user explicitly re-ranked; it powers the ``is_override`` +flag the API exposes and lets the default-seeding path (``db.priorities``) rebuild +default rows without disturbing curation. Rows are maintained inside the writing +transaction and are fully rebuildable from memberships + values. +""" + +from sqlalchemy import ( + Boolean, + CheckConstraint, + ForeignKey, + ForeignKeyConstraint, + Integer, + String, + UniqueConstraint, +) +from sqlalchemy.orm import Mapped, mapped_column +from stitch.ogsi.model.types import OGSISrcKey + +from stitch.api.entities import User as UserEntity + +from .common import Base +from .mixins import TimestampMixin, UserAuditMixin +from .oil_gas_field_source_value import ATTRIBUTE_NAMES +from .types import PORTABLE_BIGINT + + +class OGFieldResourceAttributePriority(TimestampMixin, UserAuditMixin, Base): + __tablename__ = "og_field_resource_attribute_priority" + + resource_id: Mapped[int] = mapped_column( + PORTABLE_BIGINT, + ForeignKey("og_field_resources.id", ondelete="CASCADE"), + primary_key=True, + ) + colname: Mapped[str] = mapped_column(String(50), primary_key=True) + # The specific source record ranked (two records of one source key rank + # independently), part of the grain. + source_pk: Mapped[int] = mapped_column( + PORTABLE_BIGINT, + primary_key=True, + ) + # Kept as a ranking input (global default order) and a known-source-key value. + source: Mapped[OGSISrcKey] = mapped_column(String(10), nullable=False) + # 0-based, lower wins; unique within a (resource, colname). Curated rows take + # the low contiguous positions, defaults follow. + priority: Mapped[int] = mapped_column(Integer, nullable=False) + # True when a user explicitly ranked this row (surfaced as ``is_override``). + is_curated: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + + __table_args__ = ( + UniqueConstraint( + "resource_id", "colname", "priority", name="uq_attr_priority_position" + ), + CheckConstraint("priority >= 0", name="ck_attr_priority_nonnegative"), + CheckConstraint( + "colname IN (" + ", ".join(f"'{n}'" for n in ATTRIBUTE_NAMES) + ")", + name="ck_attr_priority_colname", + ), + # The row must reference a populated source value for the field. + 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", + ), + # The source must actually be attached to this resource (a real membership), + # not merely exist and carry a value. Backed by + # uq_membership_resource_source on og_field_memberships. + ForeignKeyConstraint( + ["resource_id", "source_pk"], + [ + "og_field_memberships.resource_id", + "og_field_memberships.source_pk", + ], + name="fk_attr_priority_membership", + ondelete="CASCADE", + ), + ) + + @classmethod + def create( + cls, + *, + created_by: UserEntity, + resource_id: int, + colname: str, + source: OGSISrcKey, + source_pk: int, + priority: int, + is_curated: bool, + ): + return cls( + resource_id=resource_id, + colname=colname, + source=source, + source_pk=source_pk, + priority=priority, + is_curated=is_curated, + created_by_id=created_by.id, + last_updated_by_id=created_by.id, + ) diff --git a/deployments/api/src/stitch/api/db/model/og_field_resource_source_priority.py b/deployments/api/src/stitch/api/db/model/og_field_resource_source_priority.py deleted file mode 100644 index b4ebeb85..00000000 --- a/deployments/api/src/stitch/api/db/model/og_field_resource_source_priority.py +++ /dev/null @@ -1,89 +0,0 @@ -"""Per-resource, per-field, per-source-record priority overrides. - -Lets a curator re-rank the sources for a *single field* of a resource, overriding -the global defaults in ``og_field_source_priority``. The override is keyed on the -value grain ``(resource_id, source_pk, colname)`` -- a specific source *record* -(``source_pk``) for a specific field (``colname``) -- so two records from the same -source can be ranked independently, and reordering one field never touches another. - -Ranking is **tiered**: a record with an override row for a field ("curated") always -ranks above one without ("default"). See ``queries.add_ranking``: -``override_priority ASC NULLS LAST, default_priority ASC, source, source_pk``. The -``NULLS LAST`` split is why coalescing carries ``override_priority`` and -``default_priority`` as two columns rather than a single ``COALESCE``. Absent any -override row for a ``(resource, field)``, every record is in the default tier and -behaviour is identical to having no overrides at all. - -Audit is lightweight (``created``/``updated``/``*_by_id`` via the mixins): it records -who last set the current ordering, not a history of prior winners. -""" - -from sqlalchemy import CheckConstraint, ForeignKey, Integer, String -from sqlalchemy.orm import Mapped, mapped_column -from stitch.ogsi.model.types import OGSISrcKey - -from stitch.api.entities import User as UserEntity - -from .common import Base -from .mixins import TimestampMixin, UserAuditMixin -from .oil_gas_field_source_value import ATTRIBUTE_NAMES -from .types import PORTABLE_BIGINT - - -class OGFieldResourceSourcePriority(TimestampMixin, UserAuditMixin, Base): - __tablename__ = "og_field_resource_source_priority" - - resource_id: Mapped[int] = mapped_column( - PORTABLE_BIGINT, - ForeignKey("og_field_resources.id", ondelete="CASCADE"), - primary_key=True, - ) - # The specific source record ranked, not just its source class -- two records - # from the same source (e.g. two WoodMac rows) can be ranked independently. - source_pk: Mapped[int] = mapped_column( - PORTABLE_BIGINT, - ForeignKey("oil_gas_field_sources.id", ondelete="CASCADE"), - primary_key=True, - ) - # The field this override applies to; overrides are per-field. - colname: Mapped[str] = mapped_column(String(50), primary_key=True) - # Kept (not part of the grain) as a ranking tiebreak and a known-source-key - # guarantee via the FK to the default-priority table. - source: Mapped[OGSISrcKey] = mapped_column( - String(10), - ForeignKey("og_field_source_priority.source"), - nullable=False, - ) - # Not globally unique (unlike the default table): different resources/fields may - # legitimately reuse the same priority value. 0-based, lower = higher priority; - # only ever compared within the curated tier of a single (resource, field). - priority: Mapped[int] = mapped_column(Integer, nullable=False) - - __table_args__ = ( - # colname is the same closed, code-defined set as the value table. - CheckConstraint( - "colname IN (" + ", ".join(f"'{n}'" for n in ATTRIBUTE_NAMES) + ")", - name="ck_field_priority_colname", - ), - ) - - @classmethod - def create( - cls, - *, - created_by: UserEntity, - resource_id: int, - source: OGSISrcKey, - source_pk: int, - colname: str, - priority: int, - ): - return cls( - resource_id=resource_id, - source=source, - source_pk=source_pk, - colname=colname, - priority=priority, - created_by_id=created_by.id, - last_updated_by_id=created_by.id, - ) diff --git a/deployments/api/src/stitch/api/db/model/og_field_resource_state.py b/deployments/api/src/stitch/api/db/model/og_field_resource_state.py new file mode 100644 index 00000000..0f120b2f --- /dev/null +++ b/deployments/api/src/stitch/api/db/model/og_field_resource_state.py @@ -0,0 +1,90 @@ +"""Precomputed, permission-scoped current-state read model for OG field resources. + +One row per ``(resource, permission_mask)`` holding the fully coalesced value of +every ``OilGasFieldBase`` field, so ``list`` and ``filter-options`` read a stored +answer instead of re-running the coalescing window on every request. This is +derived data: rebuildable at any time from memberships + source values + +priorities (see ``db.read_model.state``) and kept current inside the writing +transaction. + +``permission_mask`` encodes the only visibility axes that vary between users -- +``wm`` (bit 2) and ``ccr`` (bit 1); every other source is a shared public tier +(see ``db.read_model.permissions`` and the STIT-766 design). A resource therefore +has at most four rows, one per mask. Only unmerged resources are stored; a merge +deletes the merged-away resource's rows. + +Value columns mirror ``OilGasFieldBase`` one-for-one (typed, so ``filter-options`` +is a plain indexed ``DISTINCT`` and list sort/filter are plain column predicates). +``provenance`` carries the field -> winning source key map the list view exposes. +""" + +from __future__ import annotations + +from typing import Any + +from sqlalchemy import ForeignKey, Index, Integer, SmallInteger, String +from sqlalchemy.orm import Mapped, mapped_column + +from stitch.api.entities import FILTER_OPTION_FIELDS + +from .common import Base +from .oil_gas_field_source_value import ATTRIBUTE_NAMES +from .types import PORTABLE_BIGINT, PORTABLE_FLOAT, PORTABLE_JSON, PORTABLE_JSON_NULL + + +class OGFieldResourceState(Base): + __tablename__ = "og_field_resource_state" + + resource_id: Mapped[int] = mapped_column( + PORTABLE_BIGINT, + ForeignKey("og_field_resources.id", ondelete="CASCADE"), + primary_key=True, + ) + # Two-bit visibility profile: wm (2) | ccr (1). See read_model.permissions. + permission_mask: Mapped[int] = mapped_column(SmallInteger, primary_key=True) + + # --- Coalesced value columns, one per OilGasFieldBase field --- + name: Mapped[str | None] = mapped_column(String, nullable=True) + country: Mapped[str | None] = mapped_column(String, nullable=True) + name_local: Mapped[str | None] = mapped_column(String, nullable=True) + state_province: Mapped[str | None] = mapped_column(String, nullable=True) + region: Mapped[str | None] = mapped_column(String, nullable=True) + basin: Mapped[str | None] = mapped_column(String, nullable=True) + reservoir_formation: Mapped[str | None] = mapped_column(String, nullable=True) + location_type: Mapped[str | None] = mapped_column(String, nullable=True) + production_conventionality: Mapped[str | None] = mapped_column( + String, nullable=True + ) + primary_hydrocarbon_group: Mapped[str | None] = mapped_column(String, nullable=True) + field_status: Mapped[str | None] = mapped_column(String, nullable=True) + latitude: Mapped[float | None] = mapped_column(PORTABLE_FLOAT, nullable=True) + longitude: Mapped[float | None] = mapped_column(PORTABLE_FLOAT, nullable=True) + discovery_year: Mapped[int | None] = mapped_column(Integer, nullable=True) + production_start_year: Mapped[int | None] = mapped_column(Integer, nullable=True) + fid_year: Mapped[int | None] = mapped_column(Integer, nullable=True) + owners: Mapped[Any | None] = mapped_column(PORTABLE_JSON_NULL, nullable=True) + operators: Mapped[Any | None] = mapped_column(PORTABLE_JSON_NULL, nullable=True) + + # field -> winning source key (or None); powers the list `provenance` map. + provenance: Mapped[dict[str, Any]] = mapped_column(PORTABLE_JSON, nullable=False) + + __table_args__ = tuple( + # filter-options does one DISTINCT per field within a mask. + Index(f"ix_og_field_resource_state_{field}", "permission_mask", field) + for field in FILTER_OPTION_FIELDS + ) + + +# Fail fast if the value columns drift from the coalesced entity they mirror +# (mirrors the ATTRIBUTE_KINDS guard against OilGasFieldBase). Use an explicit +# raise (not assert) so the check survives `python -O`. +_VALUE_COLUMNS = frozenset(OGFieldResourceState.__table__.columns.keys()) - { + "resource_id", + "permission_mask", + "provenance", +} +if _VALUE_COLUMNS != frozenset(ATTRIBUTE_NAMES): + raise RuntimeError( + "OGFieldResourceState value columns out of sync with ATTRIBUTE_NAMES: " + f"{_VALUE_COLUMNS ^ frozenset(ATTRIBUTE_NAMES)}" + ) diff --git a/deployments/api/src/stitch/api/db/model/og_field_source_priority.py b/deployments/api/src/stitch/api/db/model/og_field_source_priority.py deleted file mode 100644 index f080f2fa..00000000 --- a/deployments/api/src/stitch/api/db/model/og_field_source_priority.py +++ /dev/null @@ -1,20 +0,0 @@ -"""Source priority lookup table for coalescing field values. - -The canonical order lives in ``stitch.ogsi.model.SOURCE_PRIORITY``. Production -databases are seeded by the Alembic migrations; the integration tests seed this -table from ``SOURCE_PRIORITY`` in a conftest fixture (they build the schema with -``create_all`` rather than running migrations). -""" - -from sqlalchemy import Integer, String -from sqlalchemy.orm import Mapped, mapped_column -from stitch.ogsi.model import OGSISrcKey - -from .common import Base - - -class OGFieldSourcePriority(Base): - __tablename__ = "og_field_source_priority" - - source: Mapped[OGSISrcKey] = mapped_column(String(10), primary_key=True) - priority: Mapped[int] = mapped_column(Integer, nullable=False, unique=True) diff --git a/deployments/api/src/stitch/api/db/model/resource.py b/deployments/api/src/stitch/api/db/model/resource.py index 5fa03e71..3977347a 100644 --- a/deployments/api/src/stitch/api/db/model/resource.py +++ b/deployments/api/src/stitch/api/db/model/resource.py @@ -17,7 +17,7 @@ from .membership import MembershipModel, MembershipStatus from .oil_gas_field_source import OilGasFieldSourceModel -from .og_field_source_priority import OGFieldSourcePriority +from ..source_priority import source_priority_case from stitch.api.entities import User as UserEntity from .common import Base @@ -73,12 +73,12 @@ async def source_data_by_resource_id( return by_id m, r, s = MembershipModel, cls, OilGasFieldSourceModel - p = OGFieldSourcePriority + # Global default rank now comes from SOURCE_PRIORITY, not a lookup table. + priority = source_priority_case(m.source) stmt = ( - select(m.resource_id, s, p.priority.label("priority")) + select(m.resource_id, s, priority.label("priority")) .select_from(m) .join(r, r.id == m.resource_id) - .join(p, p.source == m.source) # dual-key: membership.source is not FK-tied to the header's source, # so matching on source_pk alone could admit a mismatched row. .join(s, and_(s.id == m.source_pk, s.source == m.source)) @@ -91,7 +91,7 @@ async def source_data_by_resource_id( if licensed_sources is not None: stmt = stmt.where(m.source.in_(list(dict.fromkeys(licensed_sources)))) - stmt = stmt.order_by(p.priority, m.source, s.id) + stmt = stmt.order_by(priority, m.source, s.id) for resource_id, src_model, prio in (await session.execute(stmt)).all(): by_id[resource_id].append((src_model.as_entity(), prio)) diff --git a/deployments/api/src/stitch/api/db/og_field_resource_actions.py b/deployments/api/src/stitch/api/db/og_field_resource_actions.py index 7b4dda3a..280f4ae0 100644 --- a/deployments/api/src/stitch/api/db/og_field_resource_actions.py +++ b/deployments/api/src/stitch/api/db/og_field_resource_actions.py @@ -2,7 +2,7 @@ from typing import get_args from fastapi import HTTPException -from sqlalchemy import delete, func, select +from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from starlette.status import HTTP_404_NOT_FOUND @@ -27,23 +27,29 @@ OGFieldResource, OGFieldSourceValueView, ) +from stitch.ogsi.model.og_field import OilGasFieldBase from stitch.ogsi.model.types import OGSISrcKey from .model import ( MembershipModel, MembershipStatus, - OGFieldResourceSourcePriority, + OGFieldResourceState, ResourceModel, ) +from .priorities import lock_resource, seed_or_refresh_defaults, set_curated from .model.oil_gas_field_source_value import ( ATTRIBUTE_NAMES, materialize_value, ) from .queries import ( + _build_field_conditions, + _build_sort_clauses, base_resource_query, field_source_candidates, filter_option_rows, ) +from .read_model.permissions import read_model_mask +from .read_model.state import refresh_resource_state, remove_resource_state from .utils import ( coalesce_resources, resource_model_to_entity, @@ -61,10 +67,12 @@ async def query( ) -> tuple[list[OGFieldListItemView], int]: """Query coalesced resource list items, restricted to licensed sources. - A narrowed id-query (+ count) over the participating fields selects and - orders the page, then one SQL coalesce (``coalesce_resources``) hydrates - those ids -- values and provenance come straight from the query, with no - second coalesce pass. + Callers with a canonical visibility profile (all public sources plus any + combination of wm/ccr) are served from the precomputed + ``og_field_resource_state`` read model; every other profile (unscoped, or a + partial public grant) falls back to live coalescing, which is always correct. + An unpopulated read model (e.g. before the post-migration rebuild has run) also + falls back to live, so a canonical caller is never served an empty page. """ if params.sort_by == "source": raise HTTPException( @@ -72,6 +80,86 @@ async def query( detail="sort_by=source is not supported for resource list queries.", ) + mask = read_model_mask(licensed_sources) + if mask is not None: + items, total = await _query_read_model(session, params, mask) + # An empty page from a canonical caller is only trustworthy when the read + # model is actually populated; otherwise fall back to the live path. The + # populated check runs only on an empty result, so the common (non-empty) + # path pays nothing extra. + if total or await _read_model_populated(session): + return items, total + return await _query_live(session, params, licensed_sources) + + +async def _read_model_populated(session: AsyncSession) -> bool: + """True when the read model has any rows (a cheap ``LIMIT 1`` probe). + + Used to fall back to live coalescing when the model has not been rebuilt yet. + Any row implies every mask is present for every listable resource, since the + rebuild/refresh paths always write all masks together. + """ + return ( + await session.scalar(select(OGFieldResourceState.resource_id).limit(1)) + ) is not None + + +async def _query_read_model( + session: AsyncSession, + params: OGFieldQueryParams, + mask: int, +) -> tuple[list[OGFieldListItemView], int]: + """List page served from the precomputed read model for one permission mask. + + Filtering/sorting/paging reuse the live path's shared clause builders over the + read model's typed columns, so query semantics stay identical; the winning + values and provenance are read straight from the stored row. + """ + state = OGFieldResourceState + scoped = select(state).where(state.permission_mask == mask).cte("resource_state") + + ids_stmt = select(scoped.c.resource_id) + for cond in _build_field_conditions(scoped, params): + ids_stmt = ids_stmt.where(cond) + if params.id is not None: + ids_stmt = ids_stmt.where(scoped.c.resource_id == params.id) + ids_stmt = ids_stmt.order_by(*_build_sort_clauses(scoped, params, "resource_id")) + + count_stmt = select(func.count()).select_from(ids_stmt.subquery()) + with named_query("resources.count"): + total = (await session.scalar(count_stmt)) or 0 + ids_stmt = ids_stmt.limit(params.limit).offset(params.offset) + with named_query("resources.list_ids"): + ids = list((await session.scalars(ids_stmt)).all()) + + if not ids: + return [], total + + with named_query("resources.list_hydrate"): + rows = ( + await session.scalars( + select(state).where( + state.resource_id.in_(ids), state.permission_mask == mask + ) + ) + ).all() + by_id = {row.resource_id: row for row in rows} + items = [_state_row_to_list_item_view(by_id[rid]) for rid in ids] + return items, total + + +async def _query_live( + session: AsyncSession, + params: OGFieldQueryParams, + licensed_sources: Collection[OGSISrcKey] | None = None, +) -> tuple[list[OGFieldListItemView], int]: + """Live coalescing list path (used when the read model cannot serve a caller). + + A narrowed id-query (+ count) over the participating fields selects and orders + the page, then one SQL coalesce (``coalesce_resources``) hydrates those ids -- + values and provenance come straight from the query, with no second coalesce + pass. + """ ids_stmt = base_resource_query(params, licensed_sources) count_stmt = select(func.count()).select_from(ids_stmt.subquery()) with named_query("resources.count"): @@ -91,11 +179,45 @@ async def query( return items, total +def _state_row_to_list_item_view(row: OGFieldResourceState) -> OGFieldListItemView: + """Project one stored read-model row into the list item wire model.""" + data = OilGasFieldBase(**{field: getattr(row, field) for field in ATTRIBUTE_NAMES}) + return OGFieldListItemView(id=row.resource_id, data=data, provenance=row.provenance) + + async def filter_options( session: AsyncSession, licensed_sources: Collection[OGSISrcKey] | None = None, ) -> dict[str, list[str]]: - """Distinct coalesced values for every filterable field, in one query.""" + """Distinct coalesced values for every filterable field. + + Served from the read model (a plain indexed ``DISTINCT`` per field) for + canonical visibility profiles; falls back to live coalescing otherwise. + """ + mask = read_model_mask(licensed_sources) + if mask is None or not await _read_model_populated(session): + return await _filter_options_live(session, licensed_sources) + + state = OGFieldResourceState + options: dict[str, list[str]] = {field: [] for field in FILTER_OPTION_FIELDS} + with named_query("resources.filter_options"): + for field in FILTER_OPTION_FIELDS: + col = getattr(state, field) + stmt = ( + select(col) + .where(state.permission_mask == mask, col.is_not(None)) + .distinct() + .order_by(col) + ) + options[field] = list((await session.scalars(stmt)).all()) + return options + + +async def _filter_options_live( + session: AsyncSession, + licensed_sources: Collection[OGSISrcKey] | None = None, +) -> dict[str, list[str]]: + """Live-coalescing distinct filter values (read-model fallback).""" options: dict[str, list[str]] = {field: [] for field in FILTER_OPTION_FIELDS} with named_query("resources.filter_options"): for colname, value in await session.execute( @@ -268,6 +390,11 @@ async def set_field_source_priority( if len(set(ordered_source_pks)) != len(ordered_source_pks): raise InvalidActionError("ordered_source_pks contains duplicate ids.") + # Serialize with concurrent mutations of this resource before we read the + # candidate snapshot and rewrite its priority/state rows. + with named_query("resources.set_field_source_priority.load"): + await lock_resource(session, id) + # Belt-and-suspenders behind the route's all-source-read requirement: this # write replaces the field's entire override set, so a caller who cannot read # every source could clobber rankings for sources they can't see. Require the @@ -300,26 +427,12 @@ async def set_field_source_priority( if list(ordered_source_pks) == current_order: return await field_source_values(session, id, field, licensed_sources) - source_by_pk = {row.source_pk: row.source for row in rows} with named_query("resources.set_field_source_priority.persist"): - await session.execute( - delete(OGFieldResourceSourcePriority).where( - OGFieldResourceSourcePriority.resource_id == id, - OGFieldResourceSourcePriority.colname == field, - ) - ) - for priority, source_pk in enumerate(ordered_source_pks): - session.add( - OGFieldResourceSourcePriority.create( - created_by=user, - resource_id=id, - source=source_by_pk[source_pk], - source_pk=source_pk, - colname=field, - priority=priority, - ) - ) - await session.flush() + # The listed sources become the curated tier (winner-first); the rest are + # re-derived as defaults. Rewrites the field's priority rows. + await set_curated(session, user, id, field, ordered_source_pks) + # Re-prioritizing changes the coalesced winners; refresh the read model. + await refresh_resource_state(session, id) return await field_source_values(session, id, field, licensed_sources) @@ -392,13 +505,19 @@ async def apply_resource_merge( # all ids exist, none have already been repointed # - # The merge target is a brand-new resource with no rows in - # og_field_resource_source_priority, so it resolves fields in the default - # global source order. Any per-field/per-resource priority overrides on the - # originals are intentionally NOT carried over -- merging resets ordering to - # default. (No-op reset today since the target is fresh; a later PR handles - # an explicit reset if merge semantics ever preserve an existing resource.) + # The merge target is a brand-new resource; it is seeded with default priority + # rows only (no curated rows), so it resolves fields in the default global + # source order. Any per-field curation on the originals is intentionally NOT + # carried over -- merging resets ordering to default. (No-op reset today since + # the target is fresh; a later PR handles an explicit reset if merge semantics + # ever preserve an existing resource.) with named_query("resources.merge.apply"): + # Lock the originals (sorted, to avoid deadlock between overlapping merges) + # before repointing + rebuilding derived state, serializing with concurrent + # attaches/merges on them. + for original_id in sorted(unique_ids): + await lock_resource(session, original_id) + new_resource = ResourceModel.create(created_by=user) session.add(new_resource) await session.flush() @@ -410,6 +529,17 @@ async def apply_resource_merge( _ = await _repoint_memberships(session, user, new_resource.id, unique_ids) + # Seed default priority rows for the new canonical resource. Per-resource + # curation is intentionally not carried over (merge resets to the default + # global order), so the fresh target gets defaults only. + await seed_or_refresh_defaults(session, user, new_resource.id) + + # Keep the read model in sync: the merged-away originals are now repointed + # (drop their rows) and the new canonical resource needs fresh rows. + for original_id in unique_ids: + await remove_resource_state(session, original_id) + await refresh_resource_state(session, new_resource.id) + # Return the canonical resource entity await session.refresh(new_resource, ["memberships"]) return await resource_model_to_entity(session, new_resource) @@ -447,20 +577,32 @@ async def _repoint_memberships( ) ).all() - # create new memberships pointing to the new resource - new_memberships: list[MembershipModel] = [] + # Dedup by source_pk: the merged resource gets at most one membership per source + # record (uphold uq_membership_resource_source, since the same source may be a + # member of more than one of the merged resources). The target membership is + # ACTIVE if any contributing membership was active. Every old ACTIVE membership + # is still flipped INACTIVE. + by_source_pk: dict[int, tuple[str, bool]] = {} for mem in existing_memberships: - # set status on - new_memberships.append( - MembershipModel.create( - created_by=user, - resource_id=res.id, - source=mem.source, - source_pk=mem.source_pk, - status=mem.status, - ) + source, any_active = by_source_pk.get(mem.source_pk, (mem.source, False)) + by_source_pk[mem.source_pk] = ( + source, + any_active or mem.status == MembershipStatus.ACTIVE, ) if mem.status == MembershipStatus.ACTIVE: mem.status = MembershipStatus.INACTIVE + + new_memberships = [ + MembershipModel.create( + created_by=user, + resource_id=res.id, + source=source, + source_pk=source_pk, + status=( + MembershipStatus.ACTIVE if any_active else MembershipStatus.INACTIVE + ), + ) + for source_pk, (source, any_active) in by_source_pk.items() + ] session.add_all(new_memberships) return new_memberships diff --git a/deployments/api/src/stitch/api/db/og_field_source_actions.py b/deployments/api/src/stitch/api/db/og_field_source_actions.py index 2abd5eec..2f839cb8 100644 --- a/deployments/api/src/stitch/api/db/og_field_source_actions.py +++ b/deployments/api/src/stitch/api/db/og_field_source_actions.py @@ -23,6 +23,8 @@ from .queries import ( base_source_query, ) +from .priorities import lock_resource, seed_or_refresh_defaults +from .read_model.state import refresh_resource_state from .utils import resource_model_to_entity @@ -182,18 +184,48 @@ async def _attach_source_models( src_models: Sequence[OilGasFieldSourceModel], user: User, ) -> None: - """Create ACTIVE memberships linking each source model to ``resource``.""" - memberships = [ - MembershipModel.create( - created_by=user, - resource_id=resource.id, - source=src.source, - source_pk=src.id, + """Create ACTIVE memberships linking each source model to ``resource``. + + A source is attached to a resource at most once: sources already attached (or + repeated within this batch) are skipped, upholding + ``uq_membership_resource_source``. + """ + # Serialize with any concurrent mutation of this resource before we read the + # membership snapshot and rebuild its derived priority/state rows below. + await lock_resource(session, resource.id) + already_attached = set( + ( + await session.scalars( + select(MembershipModel.source_pk).where( + MembershipModel.resource_id == resource.id + ) + ) + ).all() + ) + memberships: list[MembershipModel] = [] + seen: set[int] = set() + for src in src_models: + if src.id in already_attached or src.id in seen: + continue + seen.add(src.id) + memberships.append( + MembershipModel.create( + created_by=user, + resource_id=resource.id, + source=src.source, + source_pk=src.id, + ) ) - for src in src_models - ] session.add_all(memberships) await session.flush() + # Single choke point for membership creation on the attach path + # (create/attach/create-and-attach): seed default priority rows, scoped to just + # the fields the newly attached sources carry (no churn on unaffected fields), + # preserving existing curation; then refresh the read model. Priority rows must + # exist before the read model coalesces, and both stay in this transaction. + attached_pks = {mem.source_pk for mem in memberships} + await seed_or_refresh_defaults(session, user, resource.id, source_pks=attached_pks) + await refresh_resource_state(session, resource.id) async def attach_sources_to_resource( diff --git a/deployments/api/src/stitch/api/db/priorities.py b/deployments/api/src/stitch/api/db/priorities.py new file mode 100644 index 00000000..0ce6332e --- /dev/null +++ b/deployments/api/src/stitch/api/db/priorities.py @@ -0,0 +1,214 @@ +"""Maintenance of the single per-attribute priority store. + +Every ``(resource, colname, source_pk)`` that has a value gets exactly one row in +``og_field_resource_attribute_priority``; coalescing then ranks by the single +``priority`` column. Two operations keep that table correct, both inside the +writing transaction: + +* :func:`seed_or_refresh_defaults` -- (re)build the *default* rows for a resource + from the global ``SOURCE_PRIORITY`` order, preserving any curated rows. Called on + create / attach / merge. +* :func:`set_curated` -- replace one field's ordering, marking the listed sources + curated (they take the low, winning positions) and re-deriving the rest as + defaults. Called by the re-prioritize action. + +Curated rows occupy the low priority positions and always outrank defaults, which +reproduces the previous two-tier (override-then-default) coalescing exactly. +""" + +from __future__ import annotations + +from collections.abc import Collection, Sequence + +from sqlalchemy import delete, select +from sqlalchemy.ext.asyncio import AsyncSession + +from stitch.api.entities import User + +from .errors import InvalidActionError +from .model import ( + MembershipModel, + MembershipStatus, + OGFieldResourceAttributePriority, + OilGasFieldSourceValueModel, + ResourceModel, +) +from .source_priority import source_priority_rank + + +async def lock_resource(session: AsyncSession, resource_id: int) -> None: + """Serialize derived-state rebuilds for one resource via a row lock. + + Every path that rebuilds a resource's priority/state rows (attach, reprioritize, + merge) takes this lock first, so two concurrent mutations of the same resource + cannot interleave their delete+reinsert and collide on a unique key or leave + stale derived state. ``FOR UPDATE`` is a no-op on SQLite (unsupported), which is + fine for the single-connection test suite; it serializes under PostgreSQL. + """ + await session.execute( + select(ResourceModel.id) + .where(ResourceModel.id == resource_id) + .with_for_update() + ) + + +async def _valued_members( + session: AsyncSession, resource_id: int +) -> dict[str, dict[int, str]]: + """``colname -> {source_pk: source_key}`` for every value the resource has. + + Drawn from the resource's ACTIVE memberships; license-agnostic (priority rows + exist for every value, and licensing is applied only at read time). + """ + m = MembershipModel + v = OilGasFieldSourceValueModel + rows = await session.execute( + select(v.colname, m.source_pk, m.source) + .select_from(m) + .join(v, v.source_pk == m.source_pk) + .where( + m.resource_id == resource_id, + m.status == MembershipStatus.ACTIVE, + ) + ) + by_colname: dict[str, dict[int, str]] = {} + for colname, source_pk, source in rows.all(): + by_colname.setdefault(colname, {})[source_pk] = source + return by_colname + + +def _default_order(members: dict[int, str]) -> list[int]: + """Source pks in global default order: SOURCE_PRIORITY rank, then source_pk.""" + return sorted(members, key=lambda pk: (source_priority_rank(members[pk]), pk)) + + +async def seed_or_refresh_defaults( + session: AsyncSession, + user: User, + resource_id: int, + source_pks: Collection[int] | None = None, +) -> None: + """Rebuild the default priority rows for a resource, preserving curation. + + For each field the resource has a value for, curated rows are left untouched + and the remaining valued sources are (re)written as default rows in global + order, positioned after the curated block. Idempotent. + + ``source_pks`` scopes the work: when given (e.g. the sources just attached), + only fields those sources carry a value for are rebuilt, so an attach doesn't + churn every field's default rows. ``None`` rebuilds all fields (bootstrap/merge). + """ + valued = await _valued_members(session, resource_id) + if source_pks is not None: + scope = set(source_pks) + valued = { + colname: members + for colname, members in valued.items() + if scope.intersection(members) + } + if not valued: + return + + existing = ( + await session.scalars( + select(OGFieldResourceAttributePriority).where( + OGFieldResourceAttributePriority.resource_id == resource_id + ) + ) + ).all() + curated_by_colname: dict[str, list[OGFieldResourceAttributePriority]] = {} + for row in existing: + if row.is_curated: + curated_by_colname.setdefault(row.colname, []).append(row) + + for colname, members in valued.items(): + curated_rows = curated_by_colname.get(colname, []) + curated_pks = {row.source_pk for row in curated_rows} + # Defaults start after the highest curated position (which may be sparse + # if a curated source's value was removed), so positions never collide. + start = max((row.priority for row in curated_rows), default=-1) + 1 + + # Replace only the default rows for this field; curated rows stay put. + await session.execute( + delete(OGFieldResourceAttributePriority).where( + OGFieldResourceAttributePriority.resource_id == resource_id, + OGFieldResourceAttributePriority.colname == colname, + OGFieldResourceAttributePriority.is_curated.is_(False), + ) + ) + default_pks = [pk for pk in _default_order(members) if pk not in curated_pks] + session.add_all( + OGFieldResourceAttributePriority.create( + created_by=user, + resource_id=resource_id, + colname=colname, + source=members[pk], + source_pk=pk, + priority=start + offset, + is_curated=False, + ) + for offset, pk in enumerate(default_pks) + ) + await session.flush() + + +async def set_curated( + session: AsyncSession, + user: User, + resource_id: int, + colname: str, + ordered_source_pks: Sequence[int], +) -> None: + """Replace one field's ordering: ``ordered_source_pks`` become the curated tier. + + The listed sources take positions ``0..k-1`` (curated, winner-first); every + other valued source for the field follows as a default in global order. Fully + rewrites the field's rows for the resource. + """ + members = (await _valued_members(session, resource_id)).get(colname, {}) + curated = list(ordered_source_pks) + # Every curated source must actually carry a value for this field; guard here so + # a bad id is a clean client error, not a KeyError below. (Callers such as + # set_field_source_priority validate this too; this keeps the helper safe on its + # own.) + unknown = [pk for pk in curated if pk not in members] + if unknown: + raise InvalidActionError( + f"Cannot curate sources with no value for field {colname!r} on " + f"resource {resource_id}: {unknown}" + ) + curated_set = set(curated) + + await session.execute( + delete(OGFieldResourceAttributePriority).where( + OGFieldResourceAttributePriority.resource_id == resource_id, + OGFieldResourceAttributePriority.colname == colname, + ) + ) + rows = [ + OGFieldResourceAttributePriority.create( + created_by=user, + resource_id=resource_id, + colname=colname, + source=members[pk], + source_pk=pk, + priority=position, + is_curated=True, + ) + for position, pk in enumerate(curated) + ] + defaults = [pk for pk in _default_order(members) if pk not in curated_set] + rows.extend( + OGFieldResourceAttributePriority.create( + created_by=user, + resource_id=resource_id, + colname=colname, + source=members[pk], + source_pk=pk, + priority=len(curated) + offset, + is_curated=False, + ) + for offset, pk in enumerate(defaults) + ) + session.add_all(rows) + await session.flush() diff --git a/deployments/api/src/stitch/api/db/queries.py b/deployments/api/src/stitch/api/db/queries.py index 02ba5d5c..aaba12eb 100644 --- a/deployments/api/src/stitch/api/db/queries.py +++ b/deployments/api/src/stitch/api/db/queries.py @@ -31,13 +31,13 @@ from stitch.api.db.model import ( MembershipModel, MembershipStatus, - OGFieldResourceSourcePriority, - OGFieldSourcePriority, + OGFieldResourceAttributePriority, OilGasFieldSourceModel, OilGasFieldSourceValueModel, ResourceModel, ) from stitch.api.db.model.oil_gas_field_source_value import value_attr_for +from stitch.api.db.source_priority import source_priority_case from stitch.api.entities import FILTER_OPTION_FIELDS, OGFieldQueryParams from stitch.ogsi.model.types import OGSISrcKey @@ -102,19 +102,6 @@ def _participating_columns(params: OGFieldQueryParams) -> list[str]: return list(dict.fromkeys(participating)) -def _override_join(resource_id: Any) -> ColumnElement[bool]: - """Join condition matching an override row to its value row.""" - m = MembershipModel - v = OilGasFieldSourceValueModel - o = OGFieldResourceSourcePriority - return and_( - o.resource_id == resource_id, - o.source_pk == m.source_pk, - o.source == m.source, - o.colname == v.colname, - ) - - def construct_base_query_statement( licensed_sources: Collection[OGSISrcKey] | None = None, resource_ids: Collection[int] | None = None, @@ -123,20 +110,17 @@ def construct_base_query_statement( v = OilGasFieldSourceValueModel m = MembershipModel r = ResourceModel - p = OGFieldSourcePriority - o = OGFieldResourceSourcePriority + ap = OGFieldResourceAttributePriority active_src = ( select( r.id.label("resource_id"), m.source.label("source"), m.source_pk.label("source_pk"), - # Two columns, not a collapsed COALESCE: tiering (curated records above - # default ones) needs override_priority NULLS LAST as a distinct sort - # key. Absent an override row for the value, override_priority is NULL - # and only default_priority ranks the row (identical to no overrides). - o.priority.label("override_priority"), - p.priority.label("default_priority"), + # Single ranking key: curated rows take the low positions and outrank + # defaults, so coalescing is a plain ORDER BY priority (no tiering). + ap.priority.label("priority"), + ap.is_curated.label("is_curated"), v.colname.label("colname"), v.value_text, v.value_num, @@ -144,10 +128,23 @@ def construct_base_query_statement( ) .select_from(m) .join(r, r.id == m.resource_id) - .join(p, p.source == m.source) .join(s, and_(s.id == m.source_pk, s.source == m.source)) .join(v, v.source_pk == m.source_pk) - .outerjoin(o, _override_join(r.id)) + # LEFT join: every value has exactly one priority row once seeded, so this + # neither drops nor duplicates candidate rows in practice. Kept as an outer + # join so the source-listing path (which ignores priority) never loses a + # value row, and a missing priority row degrades to "ranks last" rather than + # silently vanishing. That completeness invariant (every active value has a + # priority row) is what keeps NULLS LAST from ever hiding a real winner; it + # is guarded by test_seeding_completeness in test_resource_state_read_model. + .outerjoin( + ap, + and_( + ap.resource_id == r.id, + ap.source_pk == m.source_pk, + ap.colname == v.colname, + ), + ) .where( r.repointed_id.is_(None), m.status == MembershipStatus.ACTIVE, @@ -194,7 +191,7 @@ def base_source_query( def _ranked(base_cte: CTE) -> CTE: - """Attach the tiered coalesce rank ``rn`` to every candidate row (no cut). + """Attach the coalesce rank ``rn`` to every candidate row (no cut). Single source of truth for "who beats whom", shared by the winner cut (``add_ranking`` -> ``rn == 1``), the per-field listing @@ -202,12 +199,13 @@ def _ranked(base_cte: CTE) -> CTE: (``coalesced_candidate_rows``) so they can't drift. ``rn == 1`` is the winner within each ``(resource_id, colname)`` partition. - The order is tiered: curated records (an override row exists for the value, so - ``override_priority`` is NOT NULL) rank above default ones (``NULLS LAST`` is - the tier split), then global default priority, then source/source_pk - tiebreaks. No empty-string handling is needed here: empty text can't be - persisted (write-path skip + DB CHECK ``ck_source_value_text_nonempty``, see - ``model.oil_gas_field_source_value``). + Order is the single ``priority`` column from + ``og_field_resource_attribute_priority`` (unique within a ``(resource, colname)``, + lower wins): curated rows hold the low positions and so outrank defaults, which + are ordered by global ``SOURCE_PRIORITY`` rank then ``source_pk`` -- exactly the + old override-then-default tiering, now linearized into one column. No + empty-string handling is needed: empty text can't be persisted (write-path skip + + DB CHECK ``ck_source_value_text_nonempty``). """ cols = base_cte.c return ( @@ -216,15 +214,10 @@ def _ranked(base_cte: CTE) -> CTE: func.row_number() .over( partition_by=(cols.resource_id, cols.colname), - # Tiered: curated records (override_priority NOT NULL) rank above - # default ones (NULLS LAST is the tier split), then global default - # priority, then source/source_pk tiebreaks. - order_by=( - cols.override_priority.asc().nulls_last(), - cols.default_priority.asc(), - cols.source.asc(), - cols.source_pk.asc(), - ), + # priority is unique within a (resource, colname) once seeded, so + # the source_pk tiebreak only matters for the defensive NULLS-LAST + # case (a value without a seeded priority row). + order_by=(cols.priority.asc().nulls_last(), cols.source_pk.asc()), ) .label("rn") ) @@ -269,23 +262,28 @@ def filter_option_rows( """Distinct winning ``(colname, value)`` pairs for every filterable field.""" m = MembershipModel v = OilGasFieldSourceValueModel - p = OGFieldSourcePriority - o = OGFieldResourceSourcePriority + ap = OGFieldResourceAttributePriority base = ( select( m.resource_id.label("resource_id"), m.source.label("source"), m.source_pk.label("source_pk"), - o.priority.label("override_priority"), - p.priority.label("default_priority"), + ap.priority.label("priority"), + ap.is_curated.label("is_curated"), v.colname.label("colname"), v.value_text, ) .select_from(m) .join(v, v.source_pk == m.source_pk) - .join(p, p.source == m.source) - .outerjoin(o, _override_join(m.resource_id)) + .outerjoin( + ap, + and_( + ap.resource_id == m.resource_id, + ap.source_pk == m.source_pk, + ap.colname == v.colname, + ), + ) .where( m.status == MembershipStatus.ACTIVE, v.colname.in_(FILTER_OPTION_FIELDS), @@ -329,7 +327,7 @@ def field_source_candidates( c.value_text, c.value_num, c.value_json, - c.override_priority.is_not(None).label("is_override"), + c.is_curated.label("is_override"), ) .where(c.colname == field) .order_by(c.rn) @@ -349,7 +347,7 @@ def coalesced_candidate_rows( grouped by ``source_pk``) from a SINGLE query, replacing the former winner-query + ``source_data_by_resource_id`` pair (see ``utils.resource_model_to_entity``). Rows are ordered - ``(default_priority, source, source_pk, colname)`` so grouping by source + ``(global source rank, source, source_pk, colname)`` so grouping by source yields sources best-priority-first, matching the old source ordering. ``source_record`` repeats per value row -- bounded (one resource's handful of @@ -377,10 +375,10 @@ def coalesced_candidate_rows( .join(s, s.id == ranked.c.source_pk) # source_data grouping (utils._source_data_from_rows) relies on each # source's rows being contiguous, so order by the source's global default - # priority -- per-field override_priority varies within a source and would - # fragment the grouping. rn (not this order) picks the per-field winner. + # rank -- per-field priority varies within a source and would fragment the + # grouping. rn (not this order) picks the per-field winner. .order_by( - ranked.c.default_priority, + source_priority_case(ranked.c.source), ranked.c.source, ranked.c.source_pk, ranked.c.colname, diff --git a/deployments/api/src/stitch/api/db/read_model/__init__.py b/deployments/api/src/stitch/api/db/read_model/__init__.py new file mode 100644 index 00000000..6f8b3b8f --- /dev/null +++ b/deployments/api/src/stitch/api/db/read_model/__init__.py @@ -0,0 +1,6 @@ +"""Precomputed current-state read model for OG field resources (STIT-766). + +``state`` builds and maintains the ``og_field_resource_state`` table; ``permissions`` +encodes the ``permission_mask`` used to key it. The table is derived data -- +rebuildable at any time from memberships, source values, and priorities. +""" diff --git a/deployments/api/src/stitch/api/db/read_model/permissions.py b/deployments/api/src/stitch/api/db/read_model/permissions.py new file mode 100644 index 00000000..88bae3bd --- /dev/null +++ b/deployments/api/src/stitch/api/db/read_model/permissions.py @@ -0,0 +1,83 @@ +"""Permission-mask encoding for the current-state read model (STIT-766). + +The precomputed read model stores one coalesced variant of each resource per +*visibility profile*. Only two source keys vary between users -- ``wm`` and +``ccr`` -- while every other source is a shared **public** tier that all users are +expected to hold read on (see the STIT-766 design). A resource therefore has at +most four variants, indexed by a two-bit ``permission_mask``: + + bit 1 (``CCR_BIT``) -> ``ccr`` visible + bit 2 (``WM_BIT``) -> ``wm`` visible + + 0 = public only · 1 = public+ccr · 2 = public+wm · 3 = public+wm+ccr + +A caller may only be served from the read model when their licensed set is one of +these four canonical profiles -- i.e. they hold read on *every* public source and +any combination of ``wm``/``ccr``. Anything else (a partial public grant) falls +back to live coalescing, so correctness never depends on the assumption holding. +""" + +from __future__ import annotations + +from collections.abc import Collection +from typing import Final, get_args + +from stitch.ogsi.model import CCR_SRC, WM_SRC +from stitch.ogsi.model.types import OGSISrcKey + +ALL_SOURCES: Final[frozenset[OGSISrcKey]] = frozenset(get_args(OGSISrcKey)) + +# The only source keys whose visibility varies between users. +RESTRICTED_SOURCES: Final[frozenset[OGSISrcKey]] = frozenset({WM_SRC, CCR_SRC}) + +# Sources every user is expected to be able to read (the "public" tier). +PUBLIC_SOURCES: Final[frozenset[OGSISrcKey]] = ALL_SOURCES - RESTRICTED_SOURCES + +CCR_BIT: Final[int] = 1 +WM_BIT: Final[int] = 2 + +# Every permission_mask a resource may be materialized for. +ALL_MASKS: Final[tuple[int, ...]] = (0, 1, 2, 3) + + +def mask_for_sources(sources: Collection[OGSISrcKey]) -> int: + """The ``permission_mask`` for a visibility profile (only wm/ccr count).""" + mask = 0 + if WM_SRC in sources: + mask |= WM_BIT + if CCR_SRC in sources: + mask |= CCR_BIT + return mask + + +def visible_sources_for_mask(mask: int) -> frozenset[OGSISrcKey]: + """The source keys visible at ``mask``: the public tier plus wm/ccr per bit.""" + visible = set(PUBLIC_SOURCES) + if mask & WM_BIT: + visible.add(WM_SRC) + if mask & CCR_BIT: + visible.add(CCR_SRC) + return frozenset(visible) + + +def is_canonical_profile(sources: Collection[OGSISrcKey]) -> bool: + """True when ``sources`` is one of the four read-model visibility profiles. + + Requires read on every public source; ``wm``/``ccr`` may be present or not. + A caller whose grants do not satisfy this must use live coalescing. + """ + return PUBLIC_SOURCES <= frozenset(sources) + + +def read_model_mask(sources: Collection[OGSISrcKey] | None) -> int | None: + """The mask to read for ``sources``, or ``None`` to fall back to live coalescing. + + ``None`` sources means "unscoped" (no licensing filter, e.g. internal calls); + that intentionally uses the live path rather than the cache. A non-canonical + grant (partial public tier) also returns ``None``. + """ + if sources is None: + return None + if not is_canonical_profile(sources): + return None + return mask_for_sources(sources) diff --git a/deployments/api/src/stitch/api/db/read_model/rebuild.py b/deployments/api/src/stitch/api/db/read_model/rebuild.py new file mode 100644 index 00000000..265ee6cc --- /dev/null +++ b/deployments/api/src/stitch/api/db/read_model/rebuild.py @@ -0,0 +1,74 @@ +"""Offline (re)build of the current-state read model. + +Run after an Alembic upgrade that introduces or changes the read model, and after +any data restore:: + + python -m stitch.api.db.read_model.rebuild # full rebuild + python -m stitch.api.db.read_model.rebuild --ids 1,2,3 # only these resources + python -m stitch.api.db.read_model.rebuild --since 2026-09-01T00:00:00+00:00 + +Ongoing writes keep the model current in-band (see ``state``); this is the +"rebuildable at any time" path for bootstrapping, recovery, and targeted refreshes. +A full rebuild fully replaces the table's contents inside one transaction; the +targeted forms touch only the affected resources. +""" + +from __future__ import annotations + +import argparse +import asyncio +from datetime import UTC, datetime + +from ..config import UnitOfWork, dispose_engine, get_session_factory +from .state import ( + rebuild_all_resource_state, + refresh_changed_since, + refresh_resource_states, +) + + +def _parse_ids(raw: str) -> list[int]: + return [int(part) for part in raw.split(",") if part.strip()] + + +def _parse_since(raw: str) -> datetime: + parsed = datetime.fromisoformat(raw) + # Compare against timezone-aware ``updated`` columns; assume UTC if naive. + return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC) + + +async def _run(ids: list[int] | None, since: datetime | None) -> None: + try: + async with UnitOfWork(get_session_factory()) as uow: + if ids is not None: + await refresh_resource_states(uow.session, ids) + elif since is not None: + changed = list(await refresh_changed_since(uow.session, since)) + print(f"Refreshed {len(changed)} changed resource(s).") + else: + await rebuild_all_resource_state(uow.session) + finally: + await dispose_engine() + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Rebuild the resource-state read model." + ) + group = parser.add_mutually_exclusive_group() + group.add_argument( + "--ids", + type=_parse_ids, + help="Comma-separated resource ids to refresh (targeted).", + ) + group.add_argument( + "--since", + type=_parse_since, + help="ISO-8601 timestamp; refresh only resources changed after it.", + ) + args = parser.parse_args() + asyncio.run(_run(args.ids, args.since)) + + +if __name__ == "__main__": + main() diff --git a/deployments/api/src/stitch/api/db/read_model/state.py b/deployments/api/src/stitch/api/db/read_model/state.py new file mode 100644 index 00000000..417f10ac --- /dev/null +++ b/deployments/api/src/stitch/api/db/read_model/state.py @@ -0,0 +1,187 @@ +"""Build and maintain the ``og_field_resource_state`` current-state read model. + +The read model is derived data: every row is the coalesced value of a resource for +one visibility profile (``permission_mask``). It is defined by the *same* live +coalescer the fallback read path uses (``utils.coalesce_resources``), restricted +to each mask's visible sources -- there is no second coalescing implementation, so +the cache cannot drift from live behavior. + +Callers keep it current with the app-controlled hooks: + +* ``refresh_resource_state`` / ``refresh_resource_states`` -- recompute one or many + resources' rows (create/attach/reprioritize/merge). The steady-state mechanism: + only the touched resources are rewritten. +* ``remove_resource_state`` -- drop a resource's rows (merged-away / repointed). +* ``refresh_changed_since`` -- recompute only resources whose inputs changed after a + timestamp (targeted drift recovery). +* ``rebuild_all_resource_state`` -- full rebuild from scratch (bootstrap after the + migration, or a coalescing-logic change). Proves the "rebuildable at any time" + invariant; reserved for those cases since ongoing writes maintain rows in-band. + +All run inside the caller's unit of work, so the cache commits atomically with the +data change that triggered it. +""" + +from __future__ import annotations + +from collections.abc import Collection, Iterable, Sequence +from datetime import datetime + +from sqlalchemy import delete, select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from stitch.ogsi.model import OGFieldResource + +from ..model import ( + MembershipModel, + OGFieldResourceAttributePriority, + OGFieldResourceState, +) +from ..model.oil_gas_field_source_value import ATTRIBUTE_NAMES +from ..queries import _resource_universe +from ..utils import coalesce_resources +from .permissions import ALL_MASKS, visible_sources_for_mask + + +def _state_row( + resource_id: int, mask: int, resource: OGFieldResource +) -> OGFieldResourceState: + """One read-model row from an already-coalesced resource entity.""" + view = resource.view + data = {} if view is None else view.model_dump(mode="json") + provenance = { + field: (None if prov is None else prov[1]) + for field, prov in resource.provenance.items() + } + return OGFieldResourceState( + resource_id=resource_id, + permission_mask=mask, + provenance=provenance, + **{field: data.get(field) for field in ATTRIBUTE_NAMES}, + ) + + +async def _compute_rows( + session: AsyncSession, resource_ids: Sequence[int] +) -> list[OGFieldResourceState]: + """Coalesced rows for every ``(resource, mask)`` pair, one coalesce per mask.""" + rows: list[OGFieldResourceState] = [] + for mask in ALL_MASKS: + visible = visible_sources_for_mask(mask) + coalesced = await coalesce_resources(session, resource_ids, visible) + rows.extend(_state_row(rid, mask, coalesced[rid]) for rid in resource_ids) + return rows + + +async def _listable_ids( + session: AsyncSession, resource_ids: Collection[int] +) -> list[int]: + """Subset of ``resource_ids`` that would appear in a list (universe members). + + Uses the same universe definition as the list query so stored rows exactly + match the set of listable resources. + """ + if not resource_ids: + return [] + universe = _resource_universe().subquery() + rows = await session.scalars( + select(universe.c.resource_id).where( + universe.c.resource_id.in_(list(resource_ids)) + ) + ) + return list(rows.all()) + + +async def remove_resource_state(session: AsyncSession, resource_id: int) -> None: + """Delete a resource's read-model rows (e.g. after it is merged away).""" + await session.execute( + delete(OGFieldResourceState).where( + OGFieldResourceState.resource_id == resource_id + ) + ) + + +async def refresh_resource_states( + session: AsyncSession, resource_ids: Collection[int] +) -> None: + """Recompute read-model rows for the given resources to match current data. + + Only these resources are touched: their rows are deleted, then the listable + ones (non-repointed, active membership) get one row per permission mask. + Non-listable resources are left with no rows, matching the list universe. + """ + ids = list(dict.fromkeys(resource_ids)) + if not ids: + return + await session.execute( + delete(OGFieldResourceState).where(OGFieldResourceState.resource_id.in_(ids)) + ) + listable = await _listable_ids(session, ids) + if listable: + session.add_all(await _compute_rows(session, listable)) + await session.flush() + + +async def refresh_resource_state(session: AsyncSession, resource_id: int) -> None: + """Recompute a single resource's read-model rows (see ``refresh_resource_states``).""" + await refresh_resource_states(session, [resource_id]) + + +async def _resources_changed_since(session: AsyncSession, since: datetime) -> set[int]: + """Resource ids whose coalescing inputs changed after ``since``. + + A resource's coalesced state changes only when its memberships or its + per-attribute priorities change (values are only written alongside a new + membership). Both tables carry an ``updated`` timestamp, so this captures + attaches, merges (memberships flip), and re-prioritizations. + """ + ids: set[int] = set() + ids.update( + ( + await session.scalars( + select(MembershipModel.resource_id).where( + MembershipModel.updated > since + ) + ) + ).all() + ) + ids.update( + ( + await session.scalars( + select(OGFieldResourceAttributePriority.resource_id).where( + OGFieldResourceAttributePriority.updated > since + ) + ) + ).all() + ) + return ids + + +async def refresh_changed_since( + session: AsyncSession, since: datetime +) -> Iterable[int]: + """Recompute only resources whose inputs changed after ``since``; return their ids. + + Targeted drift recovery -- does not catch a coalescing-*logic* change (no input + row is dirty then); use ``rebuild_all_resource_state`` for that. + """ + ids = await _resources_changed_since(session, since) + await refresh_resource_states(session, ids) + return ids + + +async def rebuild_all_resource_state(session: AsyncSession) -> None: + """Rebuild the entire read model from scratch (bootstrap / logic change). + + Wipes the table (``TRUNCATE`` on PostgreSQL to avoid dead-tuple churn) and + recomputes every listable resource. Reserved for bootstrap and coalescing-logic + changes; steady-state drift is handled incrementally by the refresh hooks. + """ + if session.bind is not None and session.bind.dialect.name == "postgresql": + await session.execute(text("TRUNCATE TABLE og_field_resource_state")) + else: + await session.execute(delete(OGFieldResourceState)) + ids: Collection[int] = list((await session.scalars(_resource_universe())).all()) + if ids: + session.add_all(await _compute_rows(session, list(ids))) + await session.flush() diff --git a/deployments/api/src/stitch/api/db/source_priority.py b/deployments/api/src/stitch/api/db/source_priority.py new file mode 100644 index 00000000..bdc95a13 --- /dev/null +++ b/deployments/api/src/stitch/api/db/source_priority.py @@ -0,0 +1,32 @@ +"""Global default source ordering, derived from the canonical ``SOURCE_PRIORITY``. + +With the per-attribute priority table (``og_field_resource_attribute_priority``) +replacing the old ``og_field_source_priority`` lookup table, the *global* default +rank of a source key now comes straight from the ``SOURCE_PRIORITY`` constant +rather than a DB row. This module is the single place that maps a source key to +its rank, for both SQL (``source_priority_case``) and Python (``source_priority_rank``). + +Lower rank wins, matching the coalescing convention. +""" + +from __future__ import annotations + +from typing import Final + +from sqlalchemy import Case, ColumnElement, case +from stitch.ogsi.model import SOURCE_PRIORITY + +_RANK: Final[dict[str, int]] = {src: rank for rank, src in enumerate(SOURCE_PRIORITY)} + +# Rank assigned to an unknown source key: after every known source. +FALLBACK_RANK: Final[int] = len(SOURCE_PRIORITY) + + +def source_priority_rank(source: str) -> int: + """The global default rank of a source key (lower wins).""" + return _RANK.get(source, FALLBACK_RANK) + + +def source_priority_case(source_col: ColumnElement[str]) -> Case[int]: + """SQL expression mapping a ``source`` column to its global default rank.""" + return case(_RANK, value=source_col, else_=FALLBACK_RANK) diff --git a/deployments/api/tests/conftest.py b/deployments/api/tests/conftest.py index 209053aa..09691086 100644 --- a/deployments/api/tests/conftest.py +++ b/deployments/api/tests/conftest.py @@ -7,16 +7,13 @@ import pytest from httpx import ASGITransport, AsyncClient from polyfactory.pytest_plugin import register_fixture -from sqlalchemy import insert from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from stitch.auth import TokenClaims from stitch.auth.permissions import ALL_PERMISSIONS -from stitch.ogsi.model import SOURCE_PRIORITY from stitch.api.db.config import UnitOfWork, get_uow from stitch.api.db.model import ( - OGFieldSourcePriority, StitchBase, UserModel, ) @@ -149,16 +146,6 @@ async def integration_engine(): ) async with engine.begin() as conn: await conn.run_sync(StitchBase.metadata.create_all, tables=non_view_tables) - # Production seeds og_field_source_priority via Alembic; the test schema - # is built with create_all, so seed the priority order here from the - # canonical SOURCE_PRIORITY constant. - await conn.execute( - insert(OGFieldSourcePriority), - [ - {"source": source, "priority": i + 1} - for i, source in enumerate(SOURCE_PRIORITY) - ], - ) yield engine await engine.dispose() diff --git a/deployments/api/tests/db/test_resource_actions.py b/deployments/api/tests/db/test_resource_actions.py index 01ea3dad..317de8fc 100644 --- a/deployments/api/tests/db/test_resource_actions.py +++ b/deployments/api/tests/db/test_resource_actions.py @@ -17,10 +17,11 @@ from stitch.api.db.model import ( MembershipModel, MembershipStatus, - OGFieldResourceSourcePriority, + OGFieldResourceAttributePriority, OilGasFieldSourceValueModel, ResourceModel, ) +from stitch.api.db.priorities import seed_or_refresh_defaults, set_curated from stitch.api.db.queries import filter_option_rows from stitch.api.entities import ( FILTER_OPTION_FIELDS, @@ -64,6 +65,9 @@ async def _create_resource_with_sources( ) await session.flush() + # Direct seeding bypasses the action layer, so seed the per-attribute default + # priority rows the coalescing path now requires (the write actions do this). + await seed_or_refresh_defaults(session, user, resource.id) return resource.id @@ -84,6 +88,7 @@ async def _add_source(session, user, rid: int, **attrs) -> int: ) ) await session.flush() + await seed_or_refresh_defaults(session, user, rid) return source.id @@ -103,19 +108,15 @@ async def _source_pks(session, rid: int, source_key: str) -> list[int]: async def _override( session, user, rid: int, source_key: str, field: str, priority: int ): - """Insert one per-field override row for the (single) record of a source.""" + """Curate the (single) record of a source as the winner for a field. + + The two-table override split is gone; curation now lives in the single + per-attribute priority table. This makes ``source_key``'s record the curated + top source for ``field`` (others become defaults) via ``set_curated``. + """ + assert priority == 0, "test helper only curates a single top source" (pk,) = await _source_pks(session, rid, source_key) - session.add( - OGFieldResourceSourcePriority.create( - created_by=user, - resource_id=rid, - source=source_key, - source_pk=pk, - colname=field, - priority=priority, - ) - ) - await session.flush() + await set_curated(session, user, rid, field, [pk]) return pk @@ -1587,15 +1588,19 @@ async def test_noop_when_order_unchanged_writes_nothing( (gem_pk,) = await _source_pks(session, rid, "gem") (wm_pk,) = await _source_pks(session, rid, "wm") - # wm, gem is already the default order -> no-op, no rows written. + # wm, gem is already the default order -> no-op: nothing is promoted to the + # curated tier (rows stay default, so a later-added source still ranks by + # global default priority). await resource_actions.set_field_source_priority( session, test_user, rid, "name", [wm_pk, gem_pk] ) - count = await session.scalar( - select(func.count()).select_from(OGFieldResourceSourcePriority) + curated_count = await session.scalar( + select(func.count()) + .select_from(OGFieldResourceAttributePriority) + .where(OGFieldResourceAttributePriority.is_curated.is_(True)) ) - assert count == 0 + assert curated_count == 0 @pytest.mark.anyio async def test_duplicate_pks_rejected( @@ -1999,20 +2004,11 @@ async def test_list_and_detail_agree_on_coalesced_winner( {"source": "wm", "name": "WM Second", "country": "CAN"}, {"source": "gem", "name": "GEM Name", "country": "BRA"}, ) - # Override: wm becomes top priority for the NAME field, both wm records - # curated in source_pk order (WM First < WM Second), so WM First wins. - for priority, pk in enumerate(await _source_pks(session, rid, "wm")): - session.add( - OGFieldResourceSourcePriority.create( - created_by=test_user, - resource_id=rid, - source="wm", - source_pk=pk, - colname="name", - priority=priority, - ) - ) - await session.flush() + # Curate wm to the top for the NAME field, both wm records in source_pk + # order (WM First < WM Second), so WM First wins. + await set_curated( + session, test_user, rid, "name", await _source_pks(session, rid, "wm") + ) # rmi unlicensed -> falls through; among licensed sources wm (override) # wins, and the lowest source_pk among the duplicate wm records wins. @@ -2092,3 +2088,170 @@ async def test_repointed_resource_yields_nullshell( res = out[rid] assert res.view.name is None assert res.provenance["name"] is None + + +class TestPriorityCollapseBehaviour: + """Single per-attribute priority store reproduces the old two-tier ranking.""" + + @pytest.mark.anyio + async def test_source_attached_after_curation_ranks_last( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + rid = await _create_resource_with_sources( + session, + test_user, + {"source": "gem", "name": "GEM Name"}, + {"source": "wm", "name": "WM Name"}, + ) + # Curate gem to the top for name (gem curated, wm default). + await _override(session, test_user, rid, "gem", "name", priority=0) + + # Attach a new rmi source (top global default) for name AFTER curation. + await _add_source(session, test_user, rid, source="rmi", name="RMI Name") + + rows = await resource_actions.field_source_values(session, rid, "name") + # Curated gem still wins; the newly attached rmi joins the default tier and + # ranks by global default order (rmi < wm), never above the curated gem. + assert [(r.source, r.is_override) for r in rows] == [ + ("gem", True), + ("rmi", False), + ("wm", False), + ] + + +class TestMembershipUniqueness: + """(resource, source) is unique; merge dedups a source shared across parents.""" + + @pytest.mark.anyio + async def test_duplicate_membership_rejected( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + rid = await _create_resource_with_sources( + session, test_user, {"source": "gem", "name": "X", "country": "USA"} + ) + (pk,) = await _source_pks(session, rid, "gem") + session.add( + MembershipModel.create( + created_by=test_user, resource_id=rid, source="gem", source_pk=pk + ) + ) + with pytest.raises(IntegrityError): + await session.flush() + + @pytest.mark.anyio + async def test_merge_dedups_source_shared_by_parents( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + # One source record, attached to two resources (allowed across resources). + rid_a = ResourceModel.create(created_by=test_user) + rid_b = ResourceModel.create(created_by=test_user) + session.add_all([rid_a, rid_b]) + await session.flush() + source = make_source_model( + source="gem", created_by_id=test_user.id, name="Shared", country="USA" + ) + session.add(source) + await session.flush() + for rid in (rid_a.id, rid_b.id): + session.add( + MembershipModel.create( + created_by=test_user, + resource_id=rid, + source="gem", + source_pk=source.id, + ) + ) + await session.flush() + + merged = await resource_actions.apply_resource_merge( + session, test_user, [rid_a.id, rid_b.id] + ) + + # The merged resource has exactly one membership for the shared source + # (no uq_membership_resource_source violation). + count = await session.scalar( + select(func.count()) + .select_from(MembershipModel) + .where( + MembershipModel.resource_id == merged.id, + MembershipModel.source_pk == source.id, + ) + ) + assert count == 1 + + +class TestPrioritySeeding: + """set_curated guards bad ids; attach-time default seeding is field-scoped.""" + + @pytest.mark.anyio + async def test_set_curated_rejects_source_without_value( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + rid = await _create_resource_with_sources( + session, test_user, {"source": "gem", "name": "N", "country": "USA"} + ) + (gem_pk,) = await _source_pks(session, rid, "gem") + # gem has no value for `basin`, so it cannot be curated there -> clean error. + with pytest.raises(InvalidActionError): + await set_curated(session, test_user, rid, "basin", [gem_pk]) + + @pytest.mark.anyio + async def test_seed_defaults_scoped_to_given_sources( + self, seeded_integration_session: AsyncSession, test_user: User + ): + from datetime import datetime + + session = seeded_integration_session + rid = await _create_resource_with_sources( + session, test_user, {"source": "gem", "name": "N", "basin": "B"} + ) + # Attach an rmi source carrying only `region`, WITHOUT seeding priorities. + rmi = make_source_model(source="rmi", created_by_id=test_user.id, region="R") + session.add(rmi) + await session.flush() + session.add( + MembershipModel.create( + created_by=test_user, + resource_id=rid, + source="rmi", + source_pk=rmi.id, + status=MembershipStatus.ACTIVE, + ) + ) + await session.flush() + + # Stamp the existing (name/basin) rows so any rewrite is detectable. + sentinel = datetime(2000, 1, 1) + for row in ( + await session.scalars( + select(OGFieldResourceAttributePriority).where( + OGFieldResourceAttributePriority.resource_id == rid + ) + ) + ).all(): + row.created = sentinel + await session.flush() + + # Scope the seed to the rmi source -> only `region` should be (re)built. + await seed_or_refresh_defaults(session, test_user, rid, source_pks=[rmi.id]) + session.expire_all() + + by_colname: dict[str, list] = {} + for row in ( + await session.scalars( + select(OGFieldResourceAttributePriority).where( + OGFieldResourceAttributePriority.resource_id == rid + ) + ) + ).all(): + by_colname.setdefault(row.colname, []).append(row) + + # region got a fresh row; name/basin were untouched (sentinel preserved). + assert "region" in by_colname + assert by_colname["region"][0].created != sentinel + assert all(r.created == sentinel for r in by_colname["name"]) + assert all(r.created == sentinel for r in by_colname["basin"]) diff --git a/deployments/api/tests/db/test_resource_state_read_model.py b/deployments/api/tests/db/test_resource_state_read_model.py new file mode 100644 index 00000000..9ddddf53 --- /dev/null +++ b/deployments/api/tests/db/test_resource_state_read_model.py @@ -0,0 +1,500 @@ +"""Tests for the precomputed current-state read model (STIT-766). + +Three concerns: + +* **Parity** -- for every canonical visibility profile, the read-model-backed + ``query``/``filter_options`` return exactly what live coalescing returns. +* **Permission masking** -- a restricted source's value only appears at masks that + can see it. +* **Sync** -- after each mutating action the incrementally-maintained table equals + a full rebuild from scratch. +""" + +from __future__ import annotations + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from stitch.api.db import og_field_resource_actions as resource_actions +from stitch.api.db.model import ( + MembershipModel, + MembershipStatus, + OGFieldResourceAttributePriority, + OGFieldResourceState, + OilGasFieldSourceValueModel, + ResourceModel, +) +from stitch.api.db.model.oil_gas_field_source_value import ATTRIBUTE_NAMES +from stitch.api.db.og_field_resource_actions import ( + _filter_options_live, + _query_live, +) +from stitch.api.db.read_model.permissions import ( + ALL_SOURCES, + PUBLIC_SOURCES, +) +from stitch.api.db.priorities import seed_or_refresh_defaults +from stitch.api.db.read_model.state import ( + rebuild_all_resource_state, + refresh_changed_since, + refresh_resource_states, +) +from stitch.api.entities import OGFieldQueryParams, User +from tests.utils import make_source_model + +pytestmark = pytest.mark.anyio + +# The four canonical visibility profiles, one per permission_mask. +_PUBLIC = frozenset(PUBLIC_SOURCES) +_PROFILES = { + "public": _PUBLIC, + "public+ccr": _PUBLIC | {"ccr"}, + "public+wm": _PUBLIC | {"wm"}, + "all": frozenset(ALL_SOURCES), +} + + +async def _new_resource(session: AsyncSession, user: User) -> int: + resource = ResourceModel.create(created_by=user) + session.add(resource) + await session.flush() + return resource.id + + +async def _add( + session: AsyncSession, user: User, rid: int, source: str, **attrs +) -> int: + model = make_source_model(source=source, created_by_id=user.id, **attrs) + session.add(model) + await session.flush() + session.add( + MembershipModel.create( + created_by=user, + resource_id=rid, + source=model.source, + source_pk=model.id, + status=MembershipStatus.ACTIVE, + ) + ) + await session.flush() + await seed_or_refresh_defaults(session, user, rid) + return model.id + + +async def _seed_dataset(session: AsyncSession, user: User) -> list[int]: + """A small dataset spanning public, wm and ccr sources with varied fields.""" + ids: list[int] = [] + + # Public-only resource. + r1 = await _new_resource(session, user) + await _add(session, user, r1, "gem", name="Alpha", country="USA", basin="Permian") + ids.append(r1) + + # wm outranks public for name; public still supplies basin. + r2 = await _new_resource(session, user) + await _add(session, user, r2, "gem", name="Beta GEM", country="SAU", basin="Ghawar") + await _add(session, user, r2, "wm", name="Beta WM", country="SAU") + ids.append(r2) + + # ccr supplies a value no public source has (region). + r3 = await _new_resource(session, user) + await _add(session, user, r3, "rmi", name="Gamma", country="NOR") + await _add(session, user, r3, "ccr", region="North Sea", basin="Viking") + ids.append(r3) + + # Resource whose only source is wm -> a null-shell for public viewers. + r4 = await _new_resource(session, user) + await _add(session, user, r4, "wm", name="Delta WM", country="GBR") + ids.append(r4) + + await session.flush() + return ids + + +def _row_key(row: OGFieldResourceState) -> tuple[int, int]: + return (row.resource_id, row.permission_mask) + + +def _row_values(row: OGFieldResourceState) -> dict: + values = {name: getattr(row, name) for name in ATTRIBUTE_NAMES} + values["provenance"] = row.provenance + return values + + +async def _dump_state(session: AsyncSession) -> dict[tuple[int, int], dict]: + rows = (await session.scalars(select(OGFieldResourceState))).all() + return {_row_key(row): _row_values(row) for row in rows} + + +_PARAM_SCENARIOS = { + "default": OGFieldQueryParams(), + "q_search": OGFieldQueryParams(q="Beta"), + "filter_basin": OGFieldQueryParams(basin=["Ghawar", "Permian"]), + "filter_country": OGFieldQueryParams(country=["SAU"]), + "sort_country_desc": OGFieldQueryParams(sort_by="country", sort_order="desc"), + "paged": OGFieldQueryParams(page=1, page_size=2, sort_by="name"), +} + + +class TestReadModelParity: + @pytest.mark.parametrize("profile", list(_PROFILES)) + @pytest.mark.parametrize("scenario", list(_PARAM_SCENARIOS)) + async def test_query_matches_live( + self, + seeded_integration_session: AsyncSession, + test_user: User, + profile: str, + scenario: str, + ): + session = seeded_integration_session + await _seed_dataset(session, test_user) + await rebuild_all_resource_state(session) + + licensed = _PROFILES[profile] + params = _PARAM_SCENARIOS[scenario] + + model_items, model_total = await resource_actions.query( + session, params, licensed_sources=licensed + ) + live_items, live_total = await _query_live( + session, params, licensed_sources=licensed + ) + + assert model_total == live_total + assert [i.id for i in model_items] == [i.id for i in live_items] + assert model_items == live_items + + @pytest.mark.parametrize("profile", list(_PROFILES)) + async def test_filter_options_match_live( + self, + seeded_integration_session: AsyncSession, + test_user: User, + profile: str, + ): + session = seeded_integration_session + await _seed_dataset(session, test_user) + await rebuild_all_resource_state(session) + + licensed = _PROFILES[profile] + assert await resource_actions.filter_options( + session, licensed_sources=licensed + ) == await _filter_options_live(session, licensed_sources=licensed) + + +class TestPermissionMasking: + async def test_restricted_value_hidden_below_its_mask( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + rid = await _new_resource(session, test_user) + # Public source has no basin; only wm supplies it. + await _add(session, test_user, rid, "gem", name="Shell", country="USA") + await _add(session, test_user, rid, "wm", basin="WM Basin") + await rebuild_all_resource_state(session) + + rows = { + row.permission_mask: row + for row in ( + await session.scalars( + select(OGFieldResourceState).where( + OGFieldResourceState.resource_id == rid + ) + ) + ).all() + } + # Four variants always stored. + assert set(rows) == {0, 1, 2, 3} + # basin only visible where wm is (masks 2 and 3). + assert rows[0].basin is None + assert rows[1].basin is None + assert rows[2].basin == "WM Basin" + assert rows[3].basin == "WM Basin" + assert rows[2].provenance["basin"] == "wm" + assert rows[0].provenance["basin"] is None + + +class TestReadModelSync: + """After each mutating action, the maintained table equals a full rebuild.""" + + async def _assert_synced(self, session: AsyncSession) -> None: + before = await _dump_state(session) + await rebuild_all_resource_state(session) + after = await _dump_state(session) + assert before == after + + async def test_synced_after_create( + self, + seeded_integration_session: AsyncSession, + test_user: User, + og_create_res_fact, + ): + session = seeded_integration_session + await resource_actions.create( + session, test_user, og_create_res_fact(name="Created") + ) + await self._assert_synced(session) + + async def test_synced_after_attach( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + ids = await _seed_dataset(session, test_user) + await rebuild_all_resource_state(session) + # Attach a new source through the action path (refresh hook fires). + from stitch.api.db.og_field_source_actions import attach_sources_to_resource + from tests.utils import make_source + from tests.factories import OGFieldBaseFactory + + base_fact = OGFieldBaseFactory + await attach_sources_to_resource( + session=session, + resource_id=ids[0], + source_rows=[make_source(base_fact, managed=False, source="bc")], + user=test_user, + ) + await self._assert_synced(session) + + async def test_synced_after_reprioritize( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + rid = await _new_resource(session, test_user) + await _add(session, test_user, rid, "gem", name="GEM Name", country="USA") + await _add(session, test_user, rid, "rmi", name="RMI Name", country="USA") + await rebuild_all_resource_state(session) + + current = await resource_actions.field_source_values(session, rid, "name") + reordered = [v.source_id for v in reversed(current)] + await resource_actions.set_field_source_priority( + session, test_user, rid, "name", reordered + ) + await self._assert_synced(session) + + async def test_merge_removes_originals_and_syncs( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + a = await _new_resource(session, test_user) + await _add(session, test_user, a, "gem", name="A", country="USA") + b = await _new_resource(session, test_user) + await _add(session, test_user, b, "rmi", name="B", country="USA") + await rebuild_all_resource_state(session) + + merged = await resource_actions.apply_resource_merge(session, test_user, [a, b]) + + # Merged-away originals have no rows; the new target does. + for original in (a, b): + assert ( + await session.scalar( + select(OGFieldResourceState.resource_id).where( + OGFieldResourceState.resource_id == original + ) + ) + is None + ) + assert ( + await session.scalar( + select(OGFieldResourceState.resource_id).where( + OGFieldResourceState.resource_id == merged.id + ) + ) + == merged.id + ) + await self._assert_synced(session) + + +class TestReadModelFallback: + """An unpopulated read model falls back to live coalescing (never empty/wrong).""" + + async def test_query_falls_back_when_unpopulated( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + await _seed_dataset(session, test_user) + # Deliberately do NOT rebuild -> og_field_resource_state is empty. + licensed = _PROFILES["all"] + params = OGFieldQueryParams() + + model_items, model_total = await resource_actions.query( + session, params, licensed_sources=licensed + ) + live_items, live_total = await _query_live( + session, params, licensed_sources=licensed + ) + assert model_total == live_total + assert model_total > 0 # dataset is non-empty, so this proves fallback ran + assert model_items == live_items + + async def test_filter_options_falls_back_when_unpopulated( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + await _seed_dataset(session, test_user) + licensed = _PROFILES["all"] + + opts = await resource_actions.filter_options(session, licensed_sources=licensed) + assert opts == await _filter_options_live(session, licensed_sources=licensed) + assert any(opts.values()) # non-empty -> came from live, not an empty table + + +class TestTargetedRebuild: + async def test_refresh_resource_states_only_touches_given_ids( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + ids = await _seed_dataset(session, test_user) + + await refresh_resource_states(session, [ids[0]]) + present = set( + (await session.scalars(select(OGFieldResourceState.resource_id))).all() + ) + assert present == {ids[0]} + + await rebuild_all_resource_state(session) + present_all = set( + (await session.scalars(select(OGFieldResourceState.resource_id))).all() + ) + assert set(ids) <= present_all + + async def test_refresh_changed_since_bounds( + self, seeded_integration_session: AsyncSession, test_user: User + ): + from datetime import datetime + + session = seeded_integration_session + ids = await _seed_dataset(session, test_user) + + # Naive bounds to match SQLite's naive timestamp storage in tests. + past = datetime(2000, 1, 1) + future = datetime(2100, 1, 1) + + changed = set(await refresh_changed_since(session, past)) + assert set(ids) <= changed # everything changed after the distant past + present = set( + (await session.scalars(select(OGFieldResourceState.resource_id))).all() + ) + assert set(ids) <= present + + # Nothing changed after the distant future. + assert set(await refresh_changed_since(session, future)) == set() + + +class TestReadModelRoundTrip: + """JSON (owners/operators) and numeric fields survive the typed-column store.""" + + async def test_json_and_numeric_fields_round_trip( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + rid = await _new_resource(session, test_user) + await _add( + session, + test_user, + rid, + "gem", + name="Round Trip", + country="USA", + latitude=12.5, + longitude=-7.25, + discovery_year=1990, + owners=[{"name": "Acme", "stake": 50}], + operators=[{"name": "OpCo", "stake": 100}], + ) + await rebuild_all_resource_state(session) + + licensed = _PROFILES["all"] + items, _ = await resource_actions.query( + session, OGFieldQueryParams(), licensed_sources=licensed + ) + item = next(i for i in items if i.id == rid) + + # Numeric columns keep their type/value through the read model. + assert item.data.latitude == 12.5 + assert item.data.longitude == -7.25 + assert item.data.discovery_year == 1990 + # JSON columns rebuild into the nested models. + assert [(o.name, o.stake) for o in item.data.owners] == [("Acme", 50)] + assert [(o.name, o.stake) for o in item.data.operators] == [("OpCo", 100)] + + # And the whole item is identical to the live coalescing path. + live_items, _ = await _query_live( + session, OGFieldQueryParams(), licensed_sources=licensed + ) + assert item == next(i for i in live_items if i.id == rid) + + +class TestSeedingCompleteness: + """Every active value has a priority row -- the invariant the LEFT join relies on. + + If it holds, coalescing's ``priority ASC NULLS LAST`` can never hide a real + winner behind a NULL, so keeping the priority join defensive (LEFT) is safe. + """ + + async def _assert_complete(self, session: AsyncSession) -> None: + valued = { + tuple(row) + for row in ( + await session.execute( + select( + MembershipModel.resource_id, + OilGasFieldSourceValueModel.colname, + MembershipModel.source_pk, + ) + .join( + OilGasFieldSourceValueModel, + OilGasFieldSourceValueModel.source_pk + == MembershipModel.source_pk, + ) + .join( + ResourceModel, ResourceModel.id == MembershipModel.resource_id + ) + .where( + MembershipModel.status == MembershipStatus.ACTIVE, + ResourceModel.repointed_id.is_(None), + ) + ) + ).all() + } + priority = { + tuple(row) + for row in ( + await session.execute( + select( + OGFieldResourceAttributePriority.resource_id, + OGFieldResourceAttributePriority.colname, + OGFieldResourceAttributePriority.source_pk, + ) + ) + ).all() + } + missing = valued - priority + assert not missing, f"active values without a priority row: {sorted(missing)}" + + async def test_complete_after_seed_and_reprioritize( + self, seeded_integration_session: AsyncSession, test_user: User + ): + session = seeded_integration_session + ids = await _seed_dataset(session, test_user) + await self._assert_complete(session) + + # Reprioritize a 2-source field (r2 has gem+wm for name) -> set_curated + # rewrites its priority rows; completeness must still hold. + rows = await resource_actions.field_source_values(session, ids[1], "name") + reordered = [r.source_id for r in reversed(rows)] + await resource_actions.set_field_source_priority( + session, test_user, ids[1], "name", reordered + ) + await self._assert_complete(session) + + async def test_complete_after_create_action( + self, + seeded_integration_session: AsyncSession, + test_user: User, + og_create_res_fact, + ): + session = seeded_integration_session + await resource_actions.create( + session, test_user, og_create_res_fact(name="Created") + ) + await self._assert_complete(session) diff --git a/deployments/api/tests/db/test_source_priority_seed.py b/deployments/api/tests/db/test_source_priority_seed.py index e462a598..b4aa4dff 100644 --- a/deployments/api/tests/db/test_source_priority_seed.py +++ b/deployments/api/tests/db/test_source_priority_seed.py @@ -1,30 +1,23 @@ -"""Guard against drift between the seeded priority table and SOURCE_PRIORITY. +"""Guard the global default source ordering against drift from SOURCE_PRIORITY. -The integration test DB seeds ``og_field_source_priority`` from the canonical -``SOURCE_PRIORITY`` constant in the ``integration_engine`` conftest fixture. -Asserting the seeded rows equal the constant keeps that seed path aligned with -the single source of truth. +The old ``og_field_source_priority`` lookup table is gone; the global default rank +of a source key now comes from ``db.source_priority`` (derived from the canonical +``SOURCE_PRIORITY`` constant). This asserts that derivation stays aligned with the +single source of truth. """ from __future__ import annotations -import pytest -from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession - -from stitch.api.db.model import OGFieldSourcePriority +from stitch.api.db.source_priority import FALLBACK_RANK, source_priority_rank from stitch.ogsi.model import SOURCE_PRIORITY -@pytest.mark.anyio -async def test_seeded_priorities_match_source_priority( - integration_session: AsyncSession, -): - result = await integration_session.execute( - select(OGFieldSourcePriority).order_by(OGFieldSourcePriority.priority) +def test_source_priority_rank_matches_source_priority(): + assert [source_priority_rank(src) for src in SOURCE_PRIORITY] == list( + range(len(SOURCE_PRIORITY)) ) - rows = result.scalars().all() - assert [(row.source, row.priority) for row in rows] == [ - (source, i + 1) for i, source in enumerate(SOURCE_PRIORITY) - ] + +def test_unknown_source_ranks_last(): + assert source_priority_rank("not-a-source") == FALLBACK_RANK + assert FALLBACK_RANK >= len(SOURCE_PRIORITY) diff --git a/deployments/api/tests/observability/test_query_name_actions.py b/deployments/api/tests/observability/test_query_name_actions.py index 619d2af3..fbbb9760 100644 --- a/deployments/api/tests/observability/test_query_name_actions.py +++ b/deployments/api/tests/observability/test_query_name_actions.py @@ -25,6 +25,7 @@ from stitch.api.db import og_field_source_actions as source_actions from stitch.api.db.model import MembershipModel, MembershipStatus, ResourceModel from stitch.api.db.model.oil_gas_field_source_value import ATTRIBUTE_NAMES +from stitch.api.db.priorities import seed_or_refresh_defaults from stitch.api.entities import ( MergeCandidateCreateRequest, MergeCandidateReviewRequest, @@ -138,6 +139,9 @@ async def _attach_source( ) ) await session.flush() + # Direct seeding bypasses the action layer; seed the default priority rows the + # coalescing path requires (done before any captured_query_events.clear()). + await seed_or_refresh_defaults(session, user, resource_id) return model.id @@ -248,8 +252,8 @@ async def test_set_field_source_priority_labels( session.add(resource) await session.flush() rid = resource.id - await _attach_source(session, test_user, rid, "gem", country="Gemland") - await _attach_source(session, test_user, rid, "rmi", country="Rmiland") + await _attach_source(session, test_user, rid, "gem", country="SAU") + await _attach_source(session, test_user, rid, "rmi", country="USA") await session.commit() # Both sources carry `country`, so both are eligible; reverse the current @@ -379,7 +383,6 @@ async def test_merge_candidate_action_labels( { "merge_candidates.detail.load", "merge_candidates.detail.coalesce", - "merge_candidates.detail.default_priority", }, ) diff --git a/deployments/api/tests/routers/test_licensed_sources_routes.py b/deployments/api/tests/routers/test_licensed_sources_routes.py index 3122fa7c..9e9276b2 100644 --- a/deployments/api/tests/routers/test_licensed_sources_routes.py +++ b/deployments/api/tests/routers/test_licensed_sources_routes.py @@ -18,6 +18,7 @@ ResourceModel, UserModel, ) +from stitch.api.db.priorities import seed_or_refresh_defaults from stitch.api.entities import User from stitch.api.main import app from tests.utils import make_source_model, make_source_record @@ -133,6 +134,10 @@ async def _seed_resource_with_sources( status=MembershipStatus.ACTIVE, ) ) + await session.flush() + # Direct seeding bypasses the action layer; seed the per-attribute default + # priority rows the coalescing path requires. + await seed_or_refresh_defaults(session, user, resource.id) await session.commit() return resource.id diff --git a/deployments/api/tests/routers/test_merge_candidates_integration.py b/deployments/api/tests/routers/test_merge_candidates_integration.py index b1d9ace7..5a74cf19 100644 --- a/deployments/api/tests/routers/test_merge_candidates_integration.py +++ b/deployments/api/tests/routers/test_merge_candidates_integration.py @@ -8,7 +8,9 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from tests.factories import ResourceCreateFactory -from stitch.api.db.model import MembershipModel, OGFieldResourceSourcePriority +from stitch.api.db.model import MembershipModel, OGFieldResourceAttributePriority +from stitch.api.db.priorities import set_curated +from stitch.api.entities import User from stitch.ogsi.model import OGFieldResource, OGFieldSource @@ -162,17 +164,7 @@ async def test_per_resource_override_is_reverted_by_merge( MembershipModel.source == "gem", ) ) - session.add( - OGFieldResourceSourcePriority( - resource_id=id_a, - source="gem", - source_pk=gem_pk, - colname="name", - priority=0, - created_by_id=1, - last_updated_by_id=1, - ) - ) + await set_curated(session, User(id=1, sub="test"), id_a, "name", [gem_pk]) await session.commit() # A's coalesced value reflects the override: its name resolves to GEM's. @@ -200,7 +192,7 @@ async def test_per_resource_override_is_reverted_by_merge( } assert {("rmi", "RMI-name"), ("gem", "GEM-name")} <= a_values - # Approving materializes the reset: the merged resource has no override + # Approving materializes the reset: the merged resource has no curated # rows and resolves `name` in default order (RMI), dropping the override. approve = await integration_client.post( f"/oil-gas-fields/merge-candidates/{candidate_id}/approve", @@ -214,14 +206,15 @@ async def test_per_resource_override_is_reverted_by_merge( assert merged.json()["data"]["name"] == "RMI-name" async with integration_session_factory() as session: - overrides = ( + curated = ( await session.execute( - select(OGFieldResourceSourcePriority).where( - OGFieldResourceSourcePriority.resource_id == merged_id + select(OGFieldResourceAttributePriority).where( + OGFieldResourceAttributePriority.resource_id == merged_id, + OGFieldResourceAttributePriority.is_curated.is_(True), ) ) ).all() - assert overrides == [] + assert curated == [] @pytest.mark.anyio async def test_overlapping_approval_reroutes_pending_candidate(