diff --git a/backend/alembic/versions/0001_initial_control_plane.py b/backend/alembic/versions/0001_initial_control_plane.py index cc14ce39b..31b187f55 100644 --- a/backend/alembic/versions/0001_initial_control_plane.py +++ b/backend/alembic/versions/0001_initial_control_plane.py @@ -9,7 +9,7 @@ from sqlalchemy import text from db.models import Base -from scripts.bootstrap_db import schema_backfill_sql +from scripts.bootstrap_db import execute_schema_backfill revision = "0001_initial_control_plane" down_revision = None @@ -19,8 +19,7 @@ def upgrade() -> None: connection = op.get_bind() connection.execute(text("CREATE EXTENSION IF NOT EXISTS vector")) Base.metadata.create_all(connection) - for statement in schema_backfill_sql(): - connection.execute(statement) + execute_schema_backfill(connection) def downgrade() -> None: diff --git a/backend/alembic/versions/0011_email_read_state.py b/backend/alembic/versions/0011_email_read_state.py index 716590cd1..04fa068ce 100644 --- a/backend/alembic/versions/0011_email_read_state.py +++ b/backend/alembic/versions/0011_email_read_state.py @@ -1,6 +1,14 @@ -"""Add is_read to emails (IMAP \\Seen read state). +"""Add is_read to email_records (IMAP \\Seen read state). Existing rows default to read so historical/file imports do not surface as unread. + +Checks both "email_records" (the real, current table -- a database whose own +0001_initial_control_plane ran before ``is_read`` was added to the ``Email`` +model has this table without the column, and needs it added) and "emails" +(a legacy name that, per 0011_email_model_reconciliation's docstring, no +migration in this repo's history ever actually created for a real managed +database, but is checked defensively in case one somehow exists). Guarded by +column existence, not just table existence, so it is safely idempotent. """ from alembic import op @@ -12,18 +20,40 @@ branch_labels = None depends_on = None +_CANDIDATE_TABLES = ("email_records", "emails") + def upgrade() -> None: - op.add_column( - "emails", - sa.Column( - "is_read", - sa.Boolean(), - nullable=False, - server_default=sa.text("true"), - ), - ) + inspector = sa.inspect(op.get_bind()) + for table_name in _CANDIDATE_TABLES: + if inspector.has_table(table_name) and not _has_column( + inspector, table_name, "is_read" + ): + op.add_column( + table_name, + sa.Column( + "is_read", + sa.Boolean(), + nullable=False, + server_default=sa.text("true"), + ), + ) def downgrade() -> None: - op.drop_column("emails", "is_read") + # Same ownership-ambiguity problem as 0018_workspace_registry's downgrade: + # a fresh database gets email_records.is_read from 0001's live + # Base.metadata.create_all, not from this revision, so there is no way to + # tell "this revision added the column" apart from "the baseline already + # had it" -- and is_read holds real per-message read/unread state, not + # rebuildable derived data. As with 0001_initial_control_plane and + # 0018_workspace_registry: production rollbacks should restore from + # backup or a later explicit down revision rather than dropping + # customer-owned data. + return None + + +def _has_column(inspector, table_name: str, column_name: str) -> bool: + return any( + column["name"] == column_name for column in inspector.get_columns(table_name) + ) diff --git a/backend/alembic/versions/0016_document_org_scope.py b/backend/alembic/versions/0016_document_org_scope.py index 0a5cd0035..0c823754e 100644 --- a/backend/alembic/versions/0016_document_org_scope.py +++ b/backend/alembic/versions/0016_document_org_scope.py @@ -8,6 +8,13 @@ recognition worker can resolve the owning organization's provider without joining through the (organization-less) workspace entity. Nullable and additive so existing rows and personal-scope documents are unaffected. + +A database that has never had ``workspace_documents`` at all (one that ran +``0001_initial_control_plane`` before ``Workspace``/``Document`` existed in +``db/models.py``, and has only applied incremental migrations since) has no +table for this revision to alter. This revision is a no-op for that case; +``0018_workspace_registry`` creates the table later in the chain, already +including this column. """ from alembic import op @@ -24,6 +31,8 @@ def upgrade() -> None: connection = op.get_bind() inspector = sa.inspect(connection) + if not inspector.has_table(_DOCUMENTS_TABLE): + return columns = {column["name"] for column in inspector.get_columns(_DOCUMENTS_TABLE)} if _ORG_COLUMN not in columns: op.add_column( @@ -39,9 +48,7 @@ def upgrade() -> None: def downgrade() -> None: - connection = op.get_bind() - inspector = sa.inspect(connection) - columns = {column["name"] for column in inspector.get_columns(_DOCUMENTS_TABLE)} - op.drop_index(_ORG_INDEX, table_name=_DOCUMENTS_TABLE, if_exists=True) - if _ORG_COLUMN in columns: - op.drop_column(_DOCUMENTS_TABLE, _ORG_COLUMN) + # 0018 can create this table and column after 0016 was a no-op. Alembic + # cannot distinguish that case from a table altered by this revision, so + # dropping the column here could destroy later organization assignments. + return None diff --git a/backend/alembic/versions/0018_workspace_registry.py b/backend/alembic/versions/0018_workspace_registry.py new file mode 100644 index 000000000..aaef5f446 --- /dev/null +++ b/backend/alembic/versions/0018_workspace_registry.py @@ -0,0 +1,81 @@ +"""create workspace registry and workspace document tables + +Revision ID: 0018_workspace_registry +Revises: 0017_merge_newsdom_carddav_heads +Create Date: 2026-09-01 00:00:00.000000 + +``Workspace``/``Document`` (``workspace_entities``/``workspace_documents``) have +been declared in ``db/models.py`` since before this repository's incremental +migration history begins tracking them explicitly. A database that ran +``0001_initial_control_plane``'s ``Base.metadata.create_all`` after these +models existed already has both tables; a database that ran ``0001`` earlier +and has only applied incremental migrations since never got them, so +``/api/data/documents`` fails with an undefined-relation error the first time +it is hit. This revision is idempotent (``has_table`` guarded) so it is a +no-op for a database that already has the tables and a real fix for one that +does not. +""" + +from alembic import op +import sqlalchemy as sa + +revision = "0018_workspace_registry" +down_revision = "0017_merge_newsdom_carddav_heads" + +_ENTITIES_TABLE = "workspace_entities" +_DOCUMENTS_TABLE = "workspace_documents" + + +def upgrade() -> None: + connection = op.get_bind() + inspector = sa.inspect(connection) + + if not inspector.has_table(_ENTITIES_TABLE): + op.create_table( + _ENTITIES_TABLE, + sa.Column("workspace_id", sa.String(), nullable=False), + sa.Column("workspace_name", sa.String(), nullable=False), + sa.Column("workspace_domain", sa.String(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("workspace_id"), + ) + + if not inspector.has_table(_DOCUMENTS_TABLE): + op.create_table( + _DOCUMENTS_TABLE, + sa.Column("document_id", sa.String(), nullable=False), + sa.Column("workspace_id", sa.String(), nullable=False), + sa.Column("organization_id", sa.String(), nullable=True), + sa.Column("document_name", sa.String(), nullable=False), + sa.Column("document_type", sa.String(), nullable=False), + sa.Column("document_content", sa.Text(), nullable=True), + sa.Column("document_status", sa.String(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["workspace_id"], [f"{_ENTITIES_TABLE}.workspace_id"] + ), + sa.PrimaryKeyConstraint("document_id"), + ) + + for index_name, column_names in ( + ("ix_workspace_documents_workspace_id", ["workspace_id"]), + ("ix_workspace_documents_organization_id", ["organization_id"]), + ): + op.create_index( + index_name, + _DOCUMENTS_TABLE, + column_names, + if_not_exists=True, + ) + + +def downgrade() -> None: + # This revision's upgrade is a no-op whenever the tables already exist + # (e.g. created by 0001's create_all), so a downgrade cannot tell "this + # revision created these tables" apart from "they predate it" -- and + # workspace_documents.document_content holds real uploaded content, not + # rebuildable derived state. Unconditionally dropping it risks destroying + # data this revision never created. As with 0001_initial_control_plane: + # production rollbacks should restore from backup or a later explicit + # down revision rather than dropping customer-owned data. + return None diff --git a/backend/alembic/versions/0019_email_read_state_repair.py b/backend/alembic/versions/0019_email_read_state_repair.py new file mode 100644 index 000000000..814fd0c25 --- /dev/null +++ b/backend/alembic/versions/0019_email_read_state_repair.py @@ -0,0 +1,57 @@ +"""idempotently ensure email_records has is_read + +Revision ID: 0019_email_read_state_repair +Revises: 0018_workspace_registry +Create Date: 2026-09-01 00:00:00.000000 + +Alembic never re-runs a revision's ``upgrade()`` once that revision id is +recorded as applied for a database -- editing ``0011_email_read_state.py``'s +content cannot repair a database that already has "0011_email_read_state" +in its ``alembic_version`` history but never actually got +``email_records.is_read`` (whatever the reason: an earlier version of that +revision that targeted the wrong table, a partial/interrupted apply, manual +intervention). This revision is the real repair path for such a database: +appended after the current head, so it runs regardless of what 0011 already +did or didn't do. Idempotent (has_table/has_column guarded) so it is a +no-op for every database that already has the column, from any path. +""" + +from alembic import op +import sqlalchemy as sa + +revision = "0019_email_read_state_repair" +down_revision = "0018_workspace_registry" + +_EMAIL_TABLE = "email_records" + + +def upgrade() -> None: + inspector = sa.inspect(op.get_bind()) + if inspector.has_table(_EMAIL_TABLE) and not _has_column( + inspector, _EMAIL_TABLE, "is_read" + ): + op.add_column( + _EMAIL_TABLE, + sa.Column( + "is_read", + sa.Boolean(), + nullable=False, + server_default=sa.text("true"), + ), + ) + + +def downgrade() -> None: + # Same ownership-ambiguity reasoning as 0011_email_read_state and + # 0018_workspace_registry: this revision cannot tell whether it was the + # one that added the column (repairing a stamped-but-incomplete + # database) or the column already existed from another path, and + # is_read holds real read/unread state, not rebuildable derived data. + # No-op; production rollbacks should restore from backup. + return None + + +def _has_column(inspector, table_name: str, column_name: str) -> bool: + return any( + column["name"] == column_name for column in inspector.get_columns(table_name) + ) diff --git a/backend/api/data.py b/backend/api/data.py index dccd85890..db4681083 100644 --- a/backend/api/data.py +++ b/backend/api/data.py @@ -36,6 +36,7 @@ ) from services.ontology_service import ontology_service from services.webdav_service import webdav_service +from services.workspace_scope import get_or_create_workspace router = APIRouter(prefix="/api/data", tags=["data"]) @@ -2484,6 +2485,36 @@ async def _count_scalar(db: AsyncSession, statement) -> int: return int(result.scalar_one() or 0) +def _auth_context_owns_its_workspace(auth_context: AuthContext) -> bool: + """Whether ``auth_context.workspace_id`` is the canonical + ``workspace-``/``workspace-`` derivation for + this session, rather than an internally inconsistent claim pairing.""" + expected_workspace_id = ( + f"workspace-{auth_context.organization_id}" + if auth_context.organization_id + else f"workspace-{auth_context.user_id}" + ) + return auth_context.workspace_id == expected_workspace_id + + +def _document_organization_filter(auth_context: AuthContext) -> ColumnElement: + """Scope a ``Document`` query to this session's organization. + + Revision 0016_document_org_scope added ``organization_id`` without + backfilling existing rows, so a legacy document can have it unset. Such a + row is trusted to belong to whichever single organization the document's + (already-filtered) ``workspace_id`` canonically maps to -- but only when + this session's own workspace/organization pairing is that canonical + derivation, not an internally inconsistent claim. + """ + organization_filter = Document.organization_id == auth_context.organization_id + if _auth_context_owns_its_workspace(auth_context): + organization_filter = or_( + organization_filter, Document.organization_id.is_(None) + ) + return organization_filter + + async def _get_workspace_document( db: AsyncSession, auth_context: AuthContext, @@ -2493,6 +2524,7 @@ async def _get_workspace_document( select(Document).where( Document.document_id == document_id, Document.workspace_id == auth_context.workspace_id, + _document_organization_filter(auth_context), ) ) document = result.scalar_one_or_none() @@ -3166,6 +3198,7 @@ async def upload_data_document( auth_context: AuthContext = Depends(get_auth_context), db: AsyncSession = Depends(get_db), ) -> DataDocumentActionResponse: + await get_or_create_workspace(db, auth_context.workspace_id) document = Document( workspace_id=auth_context.workspace_id, organization_id=auth_context.organization_id, @@ -3301,6 +3334,7 @@ async def upload_document_for_pdf_dom_recognition( status_code=415, detail="Only application/pdf uploads are supported for DOM recognition.", ) + await get_or_create_workspace(db, auth_context.workspace_id) document = Document( workspace_id=auth_context.workspace_id, organization_id=auth_context.organization_id, @@ -3931,7 +3965,10 @@ async def get_data_quality_surface( documents = await _scoped_rows( db, select(Document) - .where(Document.workspace_id == auth_context.workspace_id) + .where( + Document.workspace_id == auth_context.workspace_id, + _document_organization_filter(auth_context), + ) .order_by(Document.created_at.desc(), Document.document_id.asc()) .limit(8), ) diff --git a/backend/scripts/bootstrap_db.py b/backend/scripts/bootstrap_db.py index 1047103e8..3a1211181 100644 --- a/backend/scripts/bootstrap_db.py +++ b/backend/scripts/bootstrap_db.py @@ -1,16 +1,12 @@ import asyncio import os -from collections.abc import Sequence - -from sqlalchemy import Executable, text +from sqlalchemy import Executable, Index, MetaData, Table, inspect, text from sqlalchemy.engine import Connection from db.models import Base from db.session import engine INVALID_EMAIL_BACKFILL_OWNER_IDS = {None, "", "default"} - - def _static_bootstrap_sql(statement: str) -> Executable: # ponytail: repo-authored static bootstrap SQL only; bind params before runtime input. return text(statement) @@ -186,10 +182,6 @@ def _get_create_indexes_statements() -> list[Executable]: "CREATE INDEX IF NOT EXISTS ix_email_records_owner_date " "ON email_records (user_id, organization_id, date)" ), - text( - "CREATE INDEX IF NOT EXISTS ix_emails_owner_date " - "ON emails (user_id, organization_id, date)" - ), text( "CREATE INDEX IF NOT EXISTS ix_sender_relationships_owner_source " "ON sender_relationships " @@ -527,16 +519,24 @@ def schema_backfill_sql() -> list[Executable]: return statements -def _execute_statements(conn: Connection, statements: Sequence[Executable]) -> None: - for statement in statements: +def execute_schema_backfill(conn: Connection) -> None: + for statement in schema_backfill_sql(): conn.execute(statement) + if inspect(conn).has_table("emails"): + emails = Table("emails", MetaData(), autoload_with=conn) + Index( + "ix_emails_owner_date", + emails.c.user_id, + emails.c.organization_id, + emails.c.date, + ).create(conn, checkfirst=True) async def bootstrap_db() -> None: async with engine.begin() as conn: await conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector")) await conn.run_sync(Base.metadata.create_all) - await conn.run_sync(_execute_statements, schema_backfill_sql()) + await conn.run_sync(execute_schema_backfill) if __name__ == "__main__": diff --git a/backend/services/workspace_scope.py b/backend/services/workspace_scope.py new file mode 100644 index 000000000..5fe076a97 --- /dev/null +++ b/backend/services/workspace_scope.py @@ -0,0 +1,42 @@ +import datetime + +from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert +from sqlalchemy.ext.asyncio import AsyncSession + +from db.models import Workspace + + +async def get_or_create_workspace( + session: AsyncSession, + workspace_id: str, +) -> Workspace: + """Return the ``Workspace`` row for ``workspace_id``, creating it first if needed. + + ``workspace_id`` is the signed session's ``workspace`` claim + (``workspace-`` or ``workspace-`` for + personal scope, per ``api/auth.py``/``_derive_workspace_id``), not the + model's own opaque default. Rows created here always use that claim as + the primary key so ``Document.workspace_id``'s foreign key resolves. + """ + # The first two requests for a signed workspace may arrive concurrently. + # A SELECT-then-INSERT races on the workspace_id primary key, so let + # PostgreSQL serialize creation and return the row when this transaction + # won the insert. + result = await session.execute( + insert(Workspace) + .values( + workspace_id=workspace_id, + workspace_name=workspace_id, + created_at=datetime.datetime.now(datetime.timezone.utc), + ) + .on_conflict_do_nothing(index_elements=[Workspace.workspace_id]) + .returning(Workspace) + ) + workspace = result.scalar_one_or_none() + if workspace is None: + result = await session.execute( + select(Workspace).where(Workspace.workspace_id == workspace_id) + ) + workspace = result.scalar_one() + return workspace diff --git a/backend/tests/test_alembic_migrations.py b/backend/tests/test_alembic_migrations.py index f8f3ffeae..19b4384b7 100644 --- a/backend/tests/test_alembic_migrations.py +++ b/backend/tests/test_alembic_migrations.py @@ -32,7 +32,40 @@ def test_initial_alembic_revision_records_current_schema_path(): assert "down_revision = None" in revision_text assert "CREATE EXTENSION IF NOT EXISTS vector" in revision_text assert "Base.metadata.create_all" in revision_text - assert "schema_backfill_sql" in revision_text + assert "execute_schema_backfill" in revision_text + + +def test_email_read_state_guards_both_legacy_and_current_table_names(): + """0011_email_read_state must add is_read to a genuinely historical + email_records table missing it (a database whose own 0001 ran before + is_read was added to the Email model), not just a legacy "emails" table + that, per 0011_email_model_reconciliation's docstring, no migration in + this repo's history ever actually created for a real managed database. + The upgrade check must guard on column existence, not just table + existence, so it stays idempotent against a table that already has the + column. downgrade is a no-op: a fresh database's email_records.is_read + comes from 0001's live create_all, not from this revision, so there is + no way to tell "this revision added it" apart from "the baseline already + had it" -- and is_read holds real read/unread state, not rebuildable + derived data (same ownership-ambiguity reasoning as + 0018_workspace_registry's downgrade).""" + revision_path = BACKEND_ROOT / "alembic" / "versions" / "0011_email_read_state.py" + revision_text = revision_path.read_text() + + assert '"email_records"' in revision_text + assert '"emails"' in revision_text + assert "has_table" in revision_text + assert "_has_column" in revision_text + assert "op.add_column(" in revision_text + assert "op.drop_column(" not in revision_text + + +def test_document_org_scope_downgrade_preserves_later_assignments(): + revision_path = ( + BACKEND_ROOT / "alembic" / "versions" / "0016_document_org_scope.py" + ) + + assert "op.drop_column(" not in revision_path.read_text() def test_provider_writeback_retry_queue_has_incremental_revision(): diff --git a/backend/tests/test_bootstrap_db.py b/backend/tests/test_bootstrap_db.py index 5af0540f0..95a4e72d9 100644 --- a/backend/tests/test_bootstrap_db.py +++ b/backend/tests/test_bootstrap_db.py @@ -6,7 +6,7 @@ from core.config import settings from db.models import Base -from scripts.bootstrap_db import schema_backfill_sql +from scripts.bootstrap_db import execute_schema_backfill, schema_backfill_sql from db.models import ( AgentRunRecord, CalendarWritebackSource, @@ -30,8 +30,7 @@ def _get_schema_statements(monkeypatch): def _execute_schema_backfill(sync_conn): - for statement in schema_backfill_sql(): - sync_conn.execute(statement) + execute_schema_backfill(sync_conn) def test_schema_backfill_adds_email_columns(monkeypatch): @@ -770,11 +769,11 @@ async def test_connector_signal_events_real_postgres_bootstrap_smoke(): text(""" INSERT INTO email_records ( user_id, organization_id, message_id, sender, recipients, - subject, "date", body + subject, "date", body, is_read ) VALUES ( :user_id, :organization_id, :message_id, :sender, - :recipients, :subject, now(), :body + :recipients, :subject, now(), :body, true ) RETURNING id """), diff --git a/backend/tests/test_data_api.py b/backend/tests/test_data_api.py index cd0b7bf37..39df79883 100644 --- a/backend/tests/test_data_api.py +++ b/backend/tests/test_data_api.py @@ -28,6 +28,7 @@ Email, ProjectFolder, WebdavAccount, + Workspace, ) from db.session import get_db from main import app @@ -59,6 +60,7 @@ class MockAsyncSession: def __init__(self, results): self.results = results self.documents: list[Document] = [] + self.workspaces: list[Workspace] = [] self.queries = [] self.execute_calls = 0 @@ -66,6 +68,31 @@ async def execute(self, query): self.queries.append(query) rendered_query = str(query) rendered_query_lower = rendered_query.lower() + if "insert into workspace_entities" in rendered_query_lower: + compiled = query.compile() + params = compiled.params + workspace_id = next( + value + for key, value in params.items() + if key.startswith("workspace_id") + ) + workspace = next( + ( + workspace + for workspace in self.workspaces + if workspace.workspace_id == workspace_id + ), + None, + ) + if workspace is None: + workspace = Workspace( + workspace_id=workspace_id, + workspace_name=workspace_id, + created_at=_now(), + ) + self.workspaces.append(workspace) + return MockResult(workspace) + return MockResult(None) if ( "webdav_accounts.source_uid" in rendered_query_lower and "webdav_accounts.account_id" not in rendered_query_lower @@ -82,6 +109,26 @@ async def execute(self, query): for account in result ] ) + if "from workspace_entities" in rendered_query_lower: + compiled = query.compile() + params = compiled.params + workspace_id = next( + ( + value + for key, value in params.items() + if key.startswith("workspace_id") + ), + None, + ) + workspace = next( + ( + workspace + for workspace in self.workspaces + if workspace.workspace_id == workspace_id + ), + None, + ) + return MockResult(workspace) if "from workspace_documents" in rendered_query_lower: compiled = query.compile() params = compiled.params @@ -101,11 +148,25 @@ async def execute(self, query): ), None, ) + organization_id = next( + ( + value + for key, value in params.items() + if key.startswith("organization_id") + ), + None, + ) rows = [ document for document in self.documents if (document_id is None or document.document_id == document_id) and (workspace_id is None or document.workspace_id == workspace_id) + and ( + organization_id is None + # Older fixtures predate Document.organization_id; keep + # them usable while enforcing any explicit organization. + or document.organization_id in (None, organization_id) + ) ] if "order by" in rendered_query_lower: return MockResult(rows) @@ -121,10 +182,17 @@ def add(self, obj): if not obj.created_at: obj.created_at = _now() self.documents.append(obj) + elif isinstance(obj, Workspace): + if not obj.created_at: + obj.created_at = _now() + self.workspaces.append(obj) async def commit(self): pass + async def flush(self): + pass + async def refresh(self, obj): pass @@ -2484,6 +2552,7 @@ def test_data_quality_surface_includes_workspace_document_assets(mock_db): Document( document_id="doc_owned", workspace_id="workspace-org-acme", + organization_id="org-acme", document_name="roadmap.md", document_type="text/markdown", document_content="# Roadmap", @@ -2499,6 +2568,16 @@ def test_data_quality_surface_includes_workspace_document_assets(mock_db): document_status="uploaded", created_at=_now(), ), + Document( + document_id="doc_other_org", + workspace_id="workspace-org-acme", + organization_id="org-rival", + document_name="other-org.md", + document_type="text/markdown", + document_content="other organization", + document_status="uploaded", + created_at=_now(), + ), ] ) token = _signed_session_token(_valid_session_payload()) @@ -2544,6 +2623,7 @@ def test_data_quality_surface_includes_workspace_document_assets(mock_db): } ] assert "doc_rival" not in response.text + assert "doc_other_org" not in response.text def test_data_document_upload_creates_workspace_scoped_document(mock_db): @@ -2601,7 +2681,17 @@ def test_data_document_actions_are_workspace_scoped_and_intent_only(mock_db): document_status="uploaded", created_at=_now(), ) - mock_db.documents.extend([document, rival_document]) + other_organization_document = Document( + document_id="doc_other_org", + workspace_id="workspace-org-acme", + organization_id="org-rival", + document_name="other-org.md", + document_type="text/markdown", + document_content="other organization", + document_status="uploaded", + created_at=_now(), + ) + mock_db.documents.extend([document, rival_document, other_organization_document]) token = _signed_session_token(_valid_session_payload()) client, previous_secret, original_overrides = _with_signed_auth(mock_db, token) try: @@ -2613,6 +2703,9 @@ def test_data_document_actions_are_workspace_scoped_and_intent_only(mock_db): "/api/data/documents/doc_owned/hwp-conversion-intent" ) rival_response = client.post("/api/data/documents/doc_rival/reparse") + other_organization_response = client.post( + "/api/data/documents/doc_other_org/reparse" + ) finally: client.close() _restore_overrides(previous_secret, original_overrides) @@ -2638,6 +2731,8 @@ def test_data_document_actions_are_workspace_scoped_and_intent_only(mock_db): assert rival_response.status_code == 404 assert "doc_rival" not in rival_response.text + assert other_organization_response.status_code == 404 + assert "doc_other_org" not in other_organization_response.text def test_data_document_webdav_materialization_executes_source_backed_write( @@ -2951,11 +3046,11 @@ async def _seed_smoke_test_data(conn, ids: dict): """ INSERT INTO email_records ( user_id, organization_id, message_id, thread_id, - fingerprint, sender, recipients, subject, "date", body + fingerprint, sender, recipients, subject, "date", body, is_read ) VALUES ( :user_id, :organization_id, :message_id, :thread_id, - :fingerprint, :sender, :recipients, :subject, now(), :body + :fingerprint, :sender, :recipients, :subject, now(), :body, true ) RETURNING id """ @@ -2977,11 +3072,11 @@ async def _seed_smoke_test_data(conn, ids: dict): """ INSERT INTO email_records ( user_id, organization_id, message_id, sender, recipients, - subject, "date", body + subject, "date", body, is_read ) VALUES ( :user_id, :organization_id, :message_id, :sender, - :recipients, :subject, now(), :body + :recipients, :subject, now(), :body, true ) RETURNING id """ @@ -3001,11 +3096,11 @@ async def _seed_smoke_test_data(conn, ids: dict): """ INSERT INTO email_records ( user_id, organization_id, message_id, thread_id, - fingerprint, sender, recipients, subject, "date", body + fingerprint, sender, recipients, subject, "date", body, is_read ) VALUES ( :user_id, :organization_id, :message_id, :thread_id, - :fingerprint, :sender, :recipients, :subject, now(), :body + :fingerprint, :sender, :recipients, :subject, now(), :body, true ) RETURNING id """ diff --git a/backend/tests/test_email_read_state_migration_postgres.py b/backend/tests/test_email_read_state_migration_postgres.py new file mode 100644 index 000000000..d6294e69c --- /dev/null +++ b/backend/tests/test_email_read_state_migration_postgres.py @@ -0,0 +1,227 @@ +"""PostgreSQL regression coverage for 0011_email_read_state. + +String-matching the revision file's source (test_alembic_migrations.py) +cannot detect a destructive downgrade or prove the upgrade is actually +idempotent -- both require running the real migration against a real +database in each of the shapes it must handle. +""" + +import subprocess +import secrets +import sys +import uuid +from pathlib import Path + +import asyncpg +import pytest +from asyncpg.exceptions import InvalidAuthorizationSpecificationError, InvalidPasswordError +from sqlalchemy import text +from sqlalchemy.engine import make_url +from sqlalchemy.ext.asyncio import create_async_engine + +from core.config import settings + +pytestmark = pytest.mark.postgres + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] +_PRE_READ_STATE_REVISION = "0009_project_graph_projection" + + +def _run_migrations(database_url: str, revision: str = "head") -> None: + result = subprocess.run( + [sys.executable, str(_BACKEND_ROOT / "scripts" / "migrate_db.py"), revision], + cwd=_BACKEND_ROOT, + env={ + "DATABASE_URL": database_url, + "AUTH_SESSION_HMAC_SECRET": secrets.token_urlsafe(48), + }, + capture_output=True, + text=True, + timeout=180, + ) + assert result.returncode == 0, ( + f"scripts/migrate_db.py {revision} failed " + f"(exit {result.returncode}):\nstdout:\n{result.stdout}\nstderr:\n{result.stderr}" + ) + assert not any( + status in result.stdout + result.stderr + for status in ("Timeout", "Fatal", "Warn", "Denied") + ) + + +def _run_downgrade(database_url: str, revision: str) -> None: + # scripts/migrate_db.py only wraps alembic's upgrade command; reuse its + # alembic_config() but call command.downgrade() directly for this test. + script = ( + "from scripts.migrate_db import alembic_config\n" + "from alembic import command\n" + f"command.downgrade(alembic_config(), {revision!r})\n" + ) + result = subprocess.run( + [sys.executable, "-c", script], + cwd=_BACKEND_ROOT, + env={ + "DATABASE_URL": database_url, + "AUTH_SESSION_HMAC_SECRET": secrets.token_urlsafe(48), + }, + capture_output=True, + text=True, + timeout=180, + ) + assert result.returncode == 0, ( + f"alembic downgrade {revision} failed " + f"(exit {result.returncode}):\nstdout:\n{result.stdout}\nstderr:\n{result.stderr}" + ) + assert not any( + status in result.stdout + result.stderr + for status in ("Timeout", "Fatal", "Warn", "Denied") + ) + + +@pytest.fixture +def fresh_database_url(): + base_url = make_url(settings.DATABASE_URL) + database_name = f"test_email_read_state_{uuid.uuid4().hex[:12]}" + + async def admin(sql: str) -> None: + connection = await asyncpg.connect( + host=base_url.host, + port=base_url.port, + user=base_url.username, + password=base_url.password, + database="postgres", + ) + try: + await connection.execute(sql) + finally: + await connection.close() + + import asyncio + + try: + asyncio.run(admin(f'CREATE DATABASE "{database_name}"')) + except ( + InvalidAuthorizationSpecificationError, + InvalidPasswordError, + OSError, + ConnectionError, + ) as exc: + pytest.skip(f"PostgreSQL smoke database unavailable: {exc}") + + try: + yield base_url.set(database=database_name).render_as_string(hide_password=False) + finally: + asyncio.run(admin(f'DROP DATABASE IF EXISTS "{database_name}" WITH (FORCE)')) + + +async def _column_exists(database_url: str, table_name: str, column_name: str) -> bool: + engine = create_async_engine(database_url) + try: + async with engine.connect() as connection: + result = await connection.execute( + text( + "SELECT 1 FROM information_schema.columns " + "WHERE table_name = :table_name AND column_name = :column_name" + ), + {"table_name": table_name, "column_name": column_name}, + ) + return result.first() is not None + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_upgrade_adds_is_read_to_a_historical_email_records_table( + fresh_database_url, +): + """A database whose own 0001 ran before is_read existed in the Email + model has email_records without the column; upgrading to head must add + it, not silently leave it missing.""" + _run_migrations(fresh_database_url, revision=_PRE_READ_STATE_REVISION) + + engine = create_async_engine(fresh_database_url) + try: + async with engine.begin() as connection: + await connection.execute( + text("ALTER TABLE email_records DROP COLUMN IF EXISTS is_read") + ) + finally: + await engine.dispose() + + assert await _column_exists(fresh_database_url, "email_records", "is_read") is False + + _run_migrations(fresh_database_url) + + assert await _column_exists(fresh_database_url, "email_records", "is_read") is True + + +@pytest.mark.asyncio +async def test_upgrade_is_idempotent_against_a_legacy_emails_table_with_is_read( + fresh_database_url, +): + """A legacy "emails" table that already has is_read (e.g. from a partial + earlier application) must not make upgrade() crash with a + duplicate-column error.""" + _run_migrations(fresh_database_url, revision=_PRE_READ_STATE_REVISION) + + engine = create_async_engine(fresh_database_url) + try: + async with engine.begin() as connection: + await connection.execute( + text( + "CREATE TABLE emails " + "(id serial primary key, is_read boolean not null default true)" + ) + ) + finally: + await engine.dispose() + + _run_migrations(fresh_database_url) + + +@pytest.mark.asyncio +async def test_downgrade_does_not_destroy_read_state_on_a_fresh_database( + fresh_database_url, +): + """A fresh database's email_records.is_read comes from 0001's live + create_all, not from 0011_email_read_state -- downgrading past 0011 must + not drop it (and, if it did, would destroy real per-message read/unread + state along with it).""" + _run_migrations(fresh_database_url) + assert await _column_exists(fresh_database_url, "email_records", "is_read") is True + + _run_downgrade(fresh_database_url, _PRE_READ_STATE_REVISION) + + assert await _column_exists(fresh_database_url, "email_records", "is_read") is True + + +@pytest.mark.asyncio +async def test_upgrade_head_repairs_a_database_already_stamped_past_0011( + fresh_database_url, +): + """Alembic never re-runs a revision's upgrade() once that revision id is + recorded as applied -- editing 0011_email_read_state.py cannot repair a + database whose alembic_version history already includes it but is + missing is_read regardless (e.g. an earlier broken version of that + revision, a partial apply, manual intervention). 0019_email_read_state_ + repair is the real fix: it must add the column even though the database + is already stamped through 0018 with 0011 long since applied, so only + 0019 itself -- not a re-run of 0011 -- is what's left to bring it to + head.""" + _run_migrations(fresh_database_url, revision="0018_workspace_registry") + assert await _column_exists(fresh_database_url, "email_records", "is_read") is True + + engine = create_async_engine(fresh_database_url) + try: + async with engine.begin() as connection: + await connection.execute( + text("ALTER TABLE email_records DROP COLUMN IF EXISTS is_read") + ) + finally: + await engine.dispose() + + assert await _column_exists(fresh_database_url, "email_records", "is_read") is False + + _run_migrations(fresh_database_url) + + assert await _column_exists(fresh_database_url, "email_records", "is_read") is True diff --git a/backend/tests/test_legacy_document_scope_postgres.py b/backend/tests/test_legacy_document_scope_postgres.py new file mode 100644 index 000000000..952036d9d --- /dev/null +++ b/backend/tests/test_legacy_document_scope_postgres.py @@ -0,0 +1,184 @@ +"""PostgreSQL regression for legacy workspace documents with no organization id. + +Revision 0016 intentionally left pre-existing ``workspace_documents.organization_id`` +values NULL. Organization-scoped sessions still need to reach those rows when the +signed workspace claim is the canonical ``workspace-`` value, +without making the same NULL row visible to a different organization that presents +the same workspace string. +""" + +import subprocess +import secrets +import sys +import uuid +from pathlib import Path + +import asyncpg +import httpx +import pytest +from asyncpg.exceptions import InvalidAuthorizationSpecificationError, InvalidPasswordError +from sqlalchemy import text +from sqlalchemy.engine import make_url +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from api.auth import AuthContext, get_auth_context +from core.config import settings +from db.session import get_db, get_readonly_db +from main import app + +pytestmark = pytest.mark.postgres + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] +_ORGANIZATION_ID = "legacy-document-org" +_WORKSPACE_ID = f"workspace-{_ORGANIZATION_ID}" +_DOCUMENT_ID = "document_legacy_org_scope" + + +def _run_migrations(database_url: str) -> None: + result = subprocess.run( + [sys.executable, str(_BACKEND_ROOT / "scripts" / "migrate_db.py"), "head"], + cwd=_BACKEND_ROOT, + env={ + "DATABASE_URL": database_url, + "AUTH_SESSION_HMAC_SECRET": secrets.token_urlsafe(48), + }, + capture_output=True, + text=True, + timeout=180, + ) + output = result.stdout + result.stderr + assert result.returncode == 0, output + assert not any( + status in output for status in ("Timeout", "Fatal", "Warn", "Denied") + ), output + + +@pytest.mark.asyncio +async def test_legacy_null_org_document_is_visible_only_to_matching_signed_org() -> None: + base_url = make_url(settings.DATABASE_URL) + database_name = f"test_legacy_doc_scope_{uuid.uuid4().hex[:12]}" + + async def admin(sql: str) -> None: + connection = await asyncpg.connect( + host=base_url.host, + port=base_url.port, + user=base_url.username, + password=base_url.password, + database="postgres", + ) + try: + await connection.execute(sql) + finally: + await connection.close() + + try: + await admin(f'CREATE DATABASE "{database_name}"') + except ( + InvalidAuthorizationSpecificationError, + InvalidPasswordError, + OSError, + ConnectionError, + ) as exc: + pytest.skip(f"PostgreSQL smoke database unavailable: {exc}") + + database_url = base_url.set(database=database_name).render_as_string( + hide_password=False + ) + engine = create_async_engine(database_url) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + + async def override_db(): + async with session_factory() as session: + yield session + + async def matching_auth() -> AuthContext: + return AuthContext( + user_id="legacy-document-user", + role="member", + organization_id=_ORGANIZATION_ID, + group_ids=(), + workspace_id=_WORKSPACE_ID, + ) + + async def other_org_same_workspace_auth() -> AuthContext: + return AuthContext( + user_id="other-user", + role="member", + organization_id="other-organization", + group_ids=(), + workspace_id=_WORKSPACE_ID, + ) + + try: + _run_migrations(database_url) + async with engine.begin() as connection: + await connection.execute( + text( + """ + INSERT INTO workspace_entities + (workspace_id, workspace_name, workspace_domain, created_at) + VALUES (:workspace_id, :workspace_name, NULL, now()) + ON CONFLICT (workspace_id) DO NOTHING + """ + ), + {"workspace_id": _WORKSPACE_ID, "workspace_name": _WORKSPACE_ID}, + ) + await connection.execute( + text( + """ + INSERT INTO workspace_documents + (document_id, workspace_id, organization_id, document_name, + document_type, document_content, document_status, created_at) + VALUES + (:document_id, :workspace_id, NULL, :document_name, + 'text/markdown', '# Legacy', 'uploaded', now()) + """ + ), + { + "document_id": _DOCUMENT_ID, + "workspace_id": _WORKSPACE_ID, + "document_name": "legacy.md", + }, + ) + + app.dependency_overrides[get_db] = override_db + app.dependency_overrides[get_readonly_db] = override_db + app.dependency_overrides[get_auth_context] = matching_auth + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://testserver" + ) as client: + reparse_response = await client.post( + f"/api/data/documents/{_DOCUMENT_ID}/reparse" + ) + assert reparse_response.status_code == 200, reparse_response.text + + quality_response = await client.get("/api/data/quality-surface") + assert quality_response.status_code == 200, quality_response.text + assert _DOCUMENT_ID in { + asset["asset_key"] for asset in quality_response.json()["repository_assets"] + } + + app.dependency_overrides[get_auth_context] = other_org_same_workspace_auth + denied_response = await client.post( + f"/api/data/documents/{_DOCUMENT_ID}/reparse" + ) + assert denied_response.status_code == 404 + + other_quality_response = await client.get("/api/data/quality-surface") + assert other_quality_response.status_code == 200, other_quality_response.text + assert _DOCUMENT_ID not in { + asset["asset_key"] + for asset in other_quality_response.json()["repository_assets"] + } + finally: + app.dependency_overrides.pop(get_db, None) + app.dependency_overrides.pop(get_readonly_db, None) + app.dependency_overrides.pop(get_auth_context, None) + await engine.dispose() + try: + await admin(f'DROP DATABASE IF EXISTS "{database_name}" WITH (FORCE)') + except (OSError, ConnectionError): + # Best-effort teardown: a transient connectivity error here must not + # mask the test's actual assertions. + pass diff --git a/backend/tests/test_workspace_document_migration.py b/backend/tests/test_workspace_document_migration.py new file mode 100644 index 000000000..f685675b7 --- /dev/null +++ b/backend/tests/test_workspace_document_migration.py @@ -0,0 +1,259 @@ +"""Regression coverage for the missing ``workspace_entities``/``workspace_documents`` +Alembic migration. + +``Workspace``/``Document`` have been declared in ``db/models.py`` since before +this repository's incremental migration history tracked them explicitly (see +``alembic/versions/0018_workspace_registry.py``). No production code path ever +inserted a ``Workspace`` row for a real signed session either, so +``Document.workspace_id``'s foreign key could never be satisfied by a real +``/api/data/documents`` upload. A database missing these tables also used to +crash on ``0016_document_org_scope`` (``NoSuchTableError``) before ever +reaching ``0018_workspace_registry``; ``0016`` is now ``has_table``-guarded. + +These tests exercise the actual documented production path +(``scripts/migrate_db.py`` -> ``alembic upgrade head``, never +``Base.metadata.create_all``) against a real, disposable PostgreSQL database, +then call the real ``/api/data/documents`` endpoints through the real FastAPI +app with only the database session swapped for one bound to that database. +""" + +import asyncio +import subprocess +import secrets +import sys +import uuid +from pathlib import Path + +import asyncpg +import httpx +import pytest +import pytest_asyncio +from asyncpg.exceptions import ( + InvalidAuthorizationSpecificationError, + InvalidPasswordError, +) +from sqlalchemy import text +from sqlalchemy.engine import make_url +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from api.auth import AuthContext, get_auth_context +from core.config import settings +from db.session import get_db, get_readonly_db +from main import app + +pytestmark = pytest.mark.postgres + +BACKEND_ROOT = Path(__file__).resolve().parents[1] +# The revision immediately before 0016_document_org_scope, which -- for a +# database missing workspace_documents -- is the first migration in the +# chain that touches the table at all. Stopping here (rather than at 0017, +# after 0016 has already run) is what actually exercises the real historical +# gap: 0016 must not crash before 0018_workspace_registry ever gets to run. +_PRE_REGISTRY_REVISION = "0015_merge_newsdom_email_heads" +_SMOKE_WORKSPACE_ID = "workspace-workspace-migration-smoke-org" + + +def _run_migrations(database_url: str, revision: str = "head") -> None: + result = subprocess.run( + [sys.executable, str(BACKEND_ROOT / "scripts" / "migrate_db.py"), revision], + cwd=BACKEND_ROOT, + env={ + "DATABASE_URL": database_url, + "AUTH_SESSION_HMAC_SECRET": secrets.token_urlsafe(48), + }, + capture_output=True, + text=True, + timeout=180, + ) + assert result.returncode == 0, ( + f"scripts/migrate_db.py {revision} failed " + f"(exit {result.returncode}):\n" + f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}" + ) + assert not any( + status in result.stdout + result.stderr + for status in ("Timeout", "Fatal", "Warn", "Denied") + ) + + +@pytest_asyncio.fixture +async def fresh_database_url(): + base_url = make_url(settings.DATABASE_URL) + test_db_name = f"test_workspace_doc_{uuid.uuid4().hex[:16]}" + + async def _admin(sql: str) -> None: + conn = await asyncpg.connect( + host=base_url.host, + port=base_url.port, + user=base_url.username, + password=base_url.password, + database="postgres", + ) + try: + await conn.execute(sql) + finally: + await conn.close() + + try: + await _admin(f'CREATE DATABASE "{test_db_name}"') + except ( + InvalidAuthorizationSpecificationError, + InvalidPasswordError, + OSError, + ConnectionError, + ) as exc: + pytest.skip(f"PostgreSQL smoke database unavailable: {exc}") + + try: + yield base_url.set(database=test_db_name).render_as_string( + hide_password=False + ) + finally: + await _admin(f'DROP DATABASE IF EXISTS "{test_db_name}" WITH (FORCE)') + + +@pytest_asyncio.fixture +async def migrated_client(fresh_database_url): + engine = create_async_engine(fresh_database_url) + sessionmaker = async_sessionmaker(engine, expire_on_commit=False) + + async def override_db(): + async with sessionmaker() as session: + yield session + + async def override_auth_context() -> AuthContext: + return AuthContext( + user_id="workspace_migration_smoke_user", + role="member", + organization_id="workspace-migration-smoke-org", + group_ids=(), + workspace_id=_SMOKE_WORKSPACE_ID, + ) + + app.dependency_overrides[get_db] = override_db + app.dependency_overrides[get_readonly_db] = override_db + app.dependency_overrides[get_auth_context] = override_auth_context + try: + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://testserver" + ) as client: + yield client + finally: + app.dependency_overrides.pop(get_db, None) + app.dependency_overrides.pop(get_readonly_db, None) + app.dependency_overrides.pop(get_auth_context, None) + await engine.dispose() + + +async def _table_exists(database_url: str, name: str) -> bool: + engine = create_async_engine(database_url) + try: + async with engine.connect() as conn: + result = await conn.execute( + text("SELECT to_regclass(:name) IS NOT NULL"), {"name": name} + ) + return bool(result.scalar()) + finally: + await engine.dispose() + + +async def _drop_workspace_registry_tables(database_url: str) -> None: + engine = create_async_engine(database_url) + try: + async with engine.begin() as conn: + await conn.execute(text("DROP TABLE IF EXISTS workspace_documents")) + await conn.execute(text("DROP TABLE IF EXISTS workspace_entities")) + finally: + await engine.dispose() + + +async def _assert_document_upload_serves_cleanly(client: httpx.AsyncClient) -> None: + response = await client.post( + "/api/data/documents", + json={ + "document_name": "roadmap.md", + "document_type": "text/markdown", + "document_content": "# Roadmap", + }, + ) + assert response.status_code == 200, response.text + body = response.json() + assert body["workspace_id"] == _SMOKE_WORKSPACE_ID + assert body["document_id"] + + # A second upload from the same signed workspace must not fail trying to + # re-insert the already-provisioned Workspace row (workspace_id is its + # primary key). + second_response = await client.post( + "/api/data/documents", + json={ + "document_name": "notes.md", + "document_type": "text/markdown", + "document_content": "# Notes", + }, + ) + assert second_response.status_code == 200, second_response.text + + +@pytest.mark.asyncio +async def test_document_upload_serves_after_full_head_migration_from_empty( + fresh_database_url, migrated_client +): + """A brand-new database migrated straight to head must be able to serve + /api/data/documents without a missing-relation or FK-violation error.""" + _run_migrations(fresh_database_url) + await _assert_document_upload_serves_cleanly(migrated_client) + + +@pytest.mark.asyncio +async def test_concurrent_first_uploads_provision_one_workspace( + fresh_database_url, migrated_client +): + """Concurrent first requests must not race on the workspace primary key.""" + _run_migrations(fresh_database_url) + + responses = await asyncio.gather( + *( + migrated_client.post( + "/api/data/documents", + json={ + "document_name": f"concurrent-{index}.md", + "document_type": "text/markdown", + "document_content": "# Concurrent", + }, + ) + for index in range(16) + ) + ) + + assert [response.status_code for response in responses] == [200] * 16 + + +@pytest.mark.asyncio +async def test_document_upload_serves_after_upgrading_a_pre_registry_database( + fresh_database_url, migrated_client +): + """Reproduce the exact reported gap: a real, already-incrementally-migrated + production database that was provisioned before ``Workspace``/``Document`` + existed in ``db/models.py`` never gets ``workspace_entities``/ + ``workspace_documents`` created by any migration prior to + ``0018_workspace_registry`` (``0001_initial_control_plane``'s + ``Base.metadata.create_all`` only reflects *today's* model metadata, so it + cannot recreate that historical, pre-model-addition state on its own). + Force that end state directly -- dropping the tables a stopped-at-0015 + database would never have had -- then migrate straight to head in one + call, crossing 0016_document_org_scope (which used to crash with + NoSuchTableError on a database in exactly this state) before + 0018_workspace_registry ever runs. Prove that both tables end up correct + and /api/data/documents serves cleanly.""" + _run_migrations(fresh_database_url, revision=_PRE_REGISTRY_REVISION) + + await _drop_workspace_registry_tables(fresh_database_url) + assert await _table_exists(fresh_database_url, "workspace_entities") is False + assert await _table_exists(fresh_database_url, "workspace_documents") is False + + _run_migrations(fresh_database_url) + + assert await _table_exists(fresh_database_url, "workspace_entities") is True + assert await _table_exists(fresh_database_url, "workspace_documents") is True + await _assert_document_upload_serves_cleanly(migrated_client)