diff --git a/src/govoplan_poll/backend/db/models.py b/src/govoplan_poll/backend/db/models.py index bc0ffa4..5f1e2f4 100644 --- a/src/govoplan_poll/backend/db/models.py +++ b/src/govoplan_poll/backend/db/models.py @@ -99,6 +99,18 @@ class PollResponse(Base, TimestampMixin): __table_args__ = ( Index("ix_poll_responses_poll_submitted", "poll_id", "submitted_at"), Index("ix_poll_responses_poll_respondent", "poll_id", "respondent_id"), + Index( + "uq_poll_responses_active_respondent", + "poll_id", + "respondent_id", + unique=True, + sqlite_where=text( + "deleted_at IS NULL AND respondent_id IS NOT NULL" + ), + postgresql_where=text( + "deleted_at IS NULL AND respondent_id IS NOT NULL" + ), + ), Index("ix_poll_responses_tenant_poll", "tenant_id", "poll_id"), ) diff --git a/src/govoplan_poll/backend/migrations/versions/6e7f8a9b0c1d_v0110_poll_response_uniqueness.py b/src/govoplan_poll/backend/migrations/versions/6e7f8a9b0c1d_v0110_poll_response_uniqueness.py new file mode 100644 index 0000000..b0db6dd --- /dev/null +++ b/src/govoplan_poll/backend/migrations/versions/6e7f8a9b0c1d_v0110_poll_response_uniqueness.py @@ -0,0 +1,99 @@ +"""v0.1.10 active identified Poll response uniqueness + +Revision ID: 6e7f8a9b0c1d +Revises: 5d6e7f8a9b0c +Create Date: 2026-07-22 00:00:00.000000 + +This migration deduplicates active identified responses before creating the +partial unique index. Run it while Poll response writes are quiesced or while +the platform maintenance lock is held. PostgreSQL additionally takes a table +lock so an unexpected writer cannot enter between cleanup and index creation. +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "6e7f8a9b0c1d" +down_revision = "5d6e7f8a9b0c" +branch_labels = None +depends_on = None + + +_INDEX_NAME = "uq_poll_responses_active_respondent" +_ACTIVE_IDENTIFIED_PREDICATE = ( + "deleted_at IS NULL AND respondent_id IS NOT NULL" +) + + +def _tombstone_duplicate_active_responses() -> None: + responses = sa.table( + "poll_responses", + sa.column("id", sa.String(length=36)), + sa.column("poll_id", sa.String(length=36)), + sa.column("respondent_id", sa.String(length=255)), + sa.column("submitted_at", sa.DateTime(timezone=True)), + sa.column("deleted_at", sa.DateTime(timezone=True)), + sa.column("updated_at", sa.DateTime(timezone=True)), + ) + ranked = ( + sa.select( + responses.c.id.label("id"), + sa.func.row_number() + .over( + partition_by=( + responses.c.poll_id, + responses.c.respondent_id, + ), + order_by=( + responses.c.submitted_at.desc(), + responses.c.id.desc(), + ), + ) + .label("response_rank"), + ) + .where( + responses.c.deleted_at.is_(None), + responses.c.respondent_id.is_not(None), + ) + .subquery("ranked_active_poll_responses") + ) + duplicate_ids = sa.select(ranked.c.id).where(ranked.c.response_rank > 1) + migration_timestamp = sa.func.current_timestamp() + op.get_bind().execute( + responses.update() + .where( + responses.c.id.in_(duplicate_ids), + responses.c.deleted_at.is_(None), + ) + .values( + deleted_at=migration_timestamp, + updated_at=migration_timestamp, + ) + ) + + +def upgrade() -> None: + connection = op.get_bind() + if connection.dialect.name == "postgresql": + connection.execute( + sa.text( + "LOCK TABLE poll_responses IN SHARE ROW EXCLUSIVE MODE" + ) + ) + _tombstone_duplicate_active_responses() + op.create_index( + _INDEX_NAME, + "poll_responses", + ["poll_id", "respondent_id"], + unique=True, + sqlite_where=sa.text(_ACTIVE_IDENTIFIED_PREDICATE), + postgresql_where=sa.text(_ACTIVE_IDENTIFIED_PREDICATE), + ) + + +def downgrade() -> None: + # Deduplicated rows remain tombstoned: a downgrade must not silently revive + # responses that were no longer part of the active result set. + op.drop_index(_INDEX_NAME, table_name="poll_responses") diff --git a/tests/test_migrations.py b/tests/test_migrations.py new file mode 100644 index 0000000..75bb979 --- /dev/null +++ b/tests/test_migrations.py @@ -0,0 +1,210 @@ +from __future__ import annotations + +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from alembic import command +from alembic.runtime.migration import MigrationContext +from sqlalchemy import create_engine, inspect +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from govoplan_core.db.migrations import alembic_config +from govoplan_poll.backend.db.models import PollResponse +from govoplan_poll.backend.manifest import get_manifest +from govoplan_poll.backend.schemas import ( + PollCreateRequest, + PollOptionInput, +) +from govoplan_poll.backend.service import _existing_response, create_poll + + +_PREVIOUS_POLL_HEAD = "5d6e7f8a9b0c" +_POLL_HEAD = "6e7f8a9b0c1d" +_INDEX_NAME = "uq_poll_responses_active_respondent" + + +class PollMigrationTests(unittest.TestCase): + @staticmethod + def _config(url: str): + return alembic_config( + database_url=url, + enabled_modules=("poll",), + manifest_factories=(get_manifest,), + ) + + def test_deduplicates_deterministically_and_enforces_partial_index(self) -> None: + with tempfile.TemporaryDirectory( + prefix="govoplan-poll-migration-" + ) as directory: + url = f"sqlite:///{Path(directory) / 'poll.db'}" + config = self._config(url) + command.upgrade(config, _PREVIOUS_POLL_HEAD) + engine = create_engine(url) + latest = datetime(2026, 7, 22, 10, 0, tzinfo=timezone.utc) + historical_deleted_at = latest - timedelta(days=2) + historical_updated_at = latest - timedelta(days=1) + try: + with Session(engine) as session: + poll = create_poll( + session, + tenant_id="tenant-1", + user_id="owner-1", + payload=PollCreateRequest( + title="Migration probe", + kind="single_choice", + status="open", + allow_anonymous=True, + options=[ + PollOptionInput(key="yes", label="Yes"), + PollOptionInput(key="no", label="No"), + ], + ), + ) + poll_id = poll.id + session.add_all( + [ + PollResponse( + id="identified-oldest", + tenant_id="tenant-1", + poll_id=poll_id, + respondent_id="person-1", + answers=[], + submitted_at=latest - timedelta(hours=1), + ), + PollResponse( + id="identified-latest-a", + tenant_id="tenant-1", + poll_id=poll_id, + respondent_id="person-1", + answers=[], + submitted_at=latest, + ), + PollResponse( + id="identified-latest-b", + tenant_id="tenant-1", + poll_id=poll_id, + respondent_id="person-1", + answers=[], + submitted_at=latest, + ), + PollResponse( + id="anonymous-a", + tenant_id="tenant-1", + poll_id=poll_id, + respondent_id=None, + answers=[], + submitted_at=latest, + ), + PollResponse( + id="anonymous-b", + tenant_id="tenant-1", + poll_id=poll_id, + respondent_id=None, + answers=[], + submitted_at=latest, + ), + PollResponse( + id="already-deleted", + tenant_id="tenant-1", + poll_id=poll_id, + respondent_id="person-1", + answers=[], + submitted_at=latest + timedelta(hours=1), + deleted_at=historical_deleted_at, + updated_at=historical_updated_at, + ), + ] + ) + session.flush() + self.assertEqual( + _existing_response( + session, + poll=poll, + respondent_id="person-1", + ).id, + "identified-latest-b", + ) + session.commit() + + command.upgrade(config, "heads") + + with Session(engine) as session: + rows = { + response.id: response + for response in session.query(PollResponse).all() + } + self.assertIsNone(rows["identified-latest-b"].deleted_at) + tombstoned = ( + rows["identified-oldest"], + rows["identified-latest-a"], + ) + self.assertTrue( + all(response.deleted_at is not None for response in tombstoned) + ) + self.assertTrue( + all( + response.deleted_at == response.updated_at + for response in tombstoned + ) + ) + self.assertEqual( + len({response.deleted_at for response in tombstoned}), + 1, + ) + self.assertIsNone(rows["anonymous-a"].deleted_at) + self.assertIsNone(rows["anonymous-b"].deleted_at) + self.assertEqual( + rows["already-deleted"].deleted_at, + historical_deleted_at.replace(tzinfo=None), + ) + self.assertEqual( + rows["already-deleted"].updated_at, + historical_updated_at.replace(tzinfo=None), + ) + + session.add( + PollResponse( + tenant_id="tenant-1", + poll_id=poll_id, + respondent_id="person-1", + answers=[], + submitted_at=latest + timedelta(hours=2), + ) + ) + with self.assertRaises(IntegrityError): + session.flush() + session.rollback() + + session.add( + PollResponse( + tenant_id="tenant-1", + poll_id=poll_id, + respondent_id=None, + answers=[], + submitted_at=latest + timedelta(hours=2), + ) + ) + session.commit() + + with engine.connect() as connection: + heads = set( + MigrationContext.configure(connection).get_current_heads() + ) + indexes = { + item["name"]: item + for item in inspect(connection).get_indexes( + "poll_responses" + ) + } + self.assertIn(_POLL_HEAD, heads) + self.assertIn(_INDEX_NAME, indexes) + self.assertTrue(indexes[_INDEX_NAME]["unique"]) + finally: + engine.dispose() + + +if __name__ == "__main__": + unittest.main()