diff --git a/workspace/backend/alembic/versions/082_channel_preview_index.py b/workspace/backend/alembic/versions/082_channel_preview_index.py new file mode 100644 index 000000000..d1e2db103 --- /dev/null +++ b/workspace/backend/alembic/versions/082_channel_preview_index.py @@ -0,0 +1,26 @@ +"""Index newest channel events without blocking writes during construction.""" +from alembic import op +import sqlalchemy as sa + +revision = '082' +down_revision = '081' +branch_labels = None +depends_on = None + + +def upgrade(): + with op.get_context().autocommit_block(): + # IF NOT EXISTS allows deploying after an operational pre-build. Fail + # explicitly if an interrupted concurrent build left an invalid index. + op.execute("""CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_events_channel_latest + ON events (network_id, target, timestamp DESC, id DESC) + WHERE target LIKE 'channel/%'""") + valid = op.get_bind().execute(sa.text("""SELECT indisvalid FROM pg_index + WHERE indexrelid = 'idx_events_channel_latest'::regclass""")).scalar_one() + if not valid: + raise RuntimeError('Invalid idx_events_channel_latest: drop concurrently and retry') + + +def downgrade(): + with op.get_context().autocommit_block(): + op.execute('DROP INDEX CONCURRENTLY IF EXISTS idx_events_channel_latest') diff --git a/workspace/backend/app/models.py b/workspace/backend/app/models.py index 0856df562..15c7de7be 100644 --- a/workspace/backend/app/models.py +++ b/workspace/backend/app/models.py @@ -68,6 +68,8 @@ class EventRecord(Base): Index("idx_events_network_target", "network_id", "target"), Index("idx_events_network_timestamp", "network_id", "timestamp"), Index("idx_events_network_type_target_ts", "network_id", "type", "target", "timestamp"), + Index("idx_events_channel_latest", "network_id", "target", timestamp.desc(), id.desc(), + postgresql_where=text("target LIKE 'channel/%'")), ) diff --git a/workspace/backend/app/routers/events.py b/workspace/backend/app/routers/events.py index 3b572202c..f1781c6f2 100644 --- a/workspace/backend/app/routers/events.py +++ b/workspace/backend/app/routers/events.py @@ -1258,7 +1258,7 @@ def latest_per_channel( Return the most recent event per channel in a single query. Replaces N separate pollEvents calls for thread list previews. - Uses a SQL window function to efficiently pick the latest event per target. + Uses indexed per-target lookups on PostgreSQL, with a SQLite fallback. """ workspace = db.execute( select(Workspace).where(_workspace_filter(network)) @@ -1270,38 +1270,21 @@ def latest_per_channel( if not _verify_workspace_access(workspace, x_workspace_token, authorization): return json_response(ResponseCode.UNAUTHORIZED, "Invalid workspace credentials") - # Window function: ROW_NUMBER() OVER (PARTITION BY target ORDER BY timestamp DESC) - row_num = func.row_number().over( - partition_by=EventRecord.target, - order_by=EventRecord.timestamp.desc(), - ).label("rn") - - inner = ( - select(EventRecord, row_num) - .where( - EventRecord.network_id == workspace.id, - EventRecord.target.startswith("channel/"), - ) - ) - - if type: - inner = inner.where(EventRecord.type.startswith(type)) - # Visibility: a person gets no preview of a private thread they cannot # see, nor of a DM they are not a side of (same rules as GET /v1/events). # Machines / identified agents keep every channel. No response cache here. from app.services.visibility import hidden_channel_targets, resolve_viewer + from app.services.channel_previews import latest_channel_query + visibility_filters = [] viewer = resolve_viewer(db, workspace, x_workspace_token, authorization) if getattr(viewer, "kind", None) == "human": hidden_targets = hidden_channel_targets(db, str(workspace.id), viewer) if hidden_targets: - inner = inner.where(EventRecord.target.notin_(list(hidden_targets))) - inner = inner.where(_dm_privacy_clause(viewer.email)) - - inner = inner.subquery() + visibility_filters.append(EventRecord.target.notin_(list(hidden_targets))) + visibility_filters.append(_dm_privacy_clause(viewer.email)) - # Select only the first row per partition - query = select(inner).where(inner.c.rn == 1) + query = latest_channel_query(workspace.id, type, visibility_filters, + postgres=db.get_bind().dialect.name == 'postgresql') rows = db.execute(query).all() channels = {} diff --git a/workspace/backend/app/services/channel_previews.py b/workspace/backend/app/services/channel_previews.py new file mode 100644 index 000000000..f3ee78334 --- /dev/null +++ b/workspace/backend/app/services/channel_previews.py @@ -0,0 +1,43 @@ +"""Latest visible channel events without ranking full message histories.""" +from sqlalchemy import func, literal_column, select, true + +from app.models import EventRecord + + +def latest_channel_query(workspace_id, event_type, visibility_filters=(), *, + postgres=False, events=None): + events = events if events is not None else EventRecord.__table__ + e = events.c + # Keep this literal predicate identical to the partial index predicate. + # No user input is interpolated into SQL. + channel_filter = e.target.like(literal_column("'channel/%'")) + scope = (e.network_id == workspace_id, channel_filter) + filters = (*scope, *visibility_filters) + if event_type: + filters += (e.type.startswith(event_type),) + + if postgres: + # A loose index scan jumps to the next target instead of DISTINCT over + # every historical event. Deriving targets from events also preserves + # previews for legacy channels missing from the channels projection. + targets = (select(e.target).where(*scope).order_by(e.target).limit(1) + .cte('preview_targets', recursive=True)) + following = (select(e.target).where(*scope, e.target > targets.c.target) + .order_by(e.target).limit(1).correlate(targets).lateral()) + targets = targets.union_all( + select(following.c.target).select_from(targets.join(following, true()))) + latest = (select(e.id).where(*filters, e.target == targets.c.target) + .order_by(e.timestamp.desc(), e.id.desc()).limit(1) + .correlate(targets).lateral()) + winners = (select(latest.c.id).select_from(targets.join(latest, true())) + .cte('preview_winners')) + else: + # SQLite has no LATERAL. Rank only narrow IDs, then fetch the payloads + # of winning rows. Use the same deterministic tie-break as PostgreSQL. + ranked = select(e.id, func.row_number().over( + partition_by=e.target, + order_by=(e.timestamp.desc(), e.id.desc()), + ).label('rn')).where(*filters).subquery() + winners = select(ranked.c.id).where(ranked.c.rn == 1).cte('preview_winners') + + return select(events).join(winners, e.id == winners.c.id) diff --git a/workspace/backend/tests/test_channel_previews.py b/workspace/backend/tests/test_channel_previews.py new file mode 100644 index 000000000..425a05fc7 --- /dev/null +++ b/workspace/backend/tests/test_channel_previews.py @@ -0,0 +1,104 @@ +"""Identical preview semantics on SQLite and read-only PostgreSQL VALUES data.""" +import os + +import pytest +from sqlalchemy import bindparam, cast, create_engine, select, text, union_all +from sqlalchemy.sql.util import ClauseAdapter + +from app.models import EventRecord +from app.routers.events import _dm_privacy_clause +from app.services.channel_previews import latest_channel_query + +WID = '11111111-1111-4111-8111-111111111111' +OTHER = '22222222-2222-4222-8222-222222222222' +ME = 'me@example.test' + + +@pytest.fixture(params=['sqlite', 'postgresql']) +def query_db(request, db): + if request.param == 'sqlite': + yield db, False + return + if os.getenv('OPENAGENTS_TEST_POSTGRES_READONLY') != '1': + pytest.skip('requires opt-in read-only PostgreSQL connection') + from app.config import config + engine = create_engine(config.DATABASE_URL) + with engine.connect() as conn: + conn.execute(text('SET TRANSACTION READ ONLY')) + conn.execute(text("SET LOCAL statement_timeout='5s'")) + yield conn, True + conn.rollback() + engine.dispose() + + +def previews(query_db, rows, prefix='workspace.message', human=False): + conn, postgres = query_db + table = EventRecord.__table__ + # Synthetic rows never touch production tables, including in the PG path. + columns = [c for c in table.c if c.name != 'created_at'] + def value(i, c, row): + param = bindparam(f'r{i}_{c.name}', row[c.name], type_=c.type) + return (cast(param, c.type) if postgres else param).label(c.name) + selects = [select(*(value(i, c, row) for c in columns)) for i, row in enumerate(rows)] + events = (union_all(*selects) if len(selects) > 1 else selects[0]).cte('fixture_events') + filters = [] + if human: + filters = [events.c.target != 'channel/private', + ClauseAdapter(events, adapt_on_names=True).traverse(_dm_privacy_clause(ME))] + query = latest_channel_query(WID, prefix, filters, postgres=postgres, events=events) + return {r.target: dict(r._mapping) for r in conn.execute(query)} + + +def message(id, target='channel/legacy', timestamp=1, **kw): + return dict(id=id, network_id=WID, target=target, timestamp=timestamp, + type=kw.pop('type', 'workspace.message.posted'), + source=kw.pop('source', 'openagents:bot'), + visibility=kw.pop('visibility', 'channel'), + payload={'content': id, 'large': 'x' * 1000}, metadata={'id': id}, **kw) + + +@pytest.mark.parametrize('prefix', ['workspace.message', 'workspace.message.posted', '']) +@pytest.mark.parametrize('human', [False, True]) +def test_latest_visible_event_old_legacy_channels_type_and_workspace(query_db, prefix, human): + rows = [message('old'), message('new', timestamp=2), + message('private', 'channel/private'), + message('control', timestamp=10, type='workspace.control'), + message('foreign', timestamp=20), + message('nonchannel', 'human:someone@example.test', 30), + message('private-new', 'channel/private', 40), + message('visible', 'channel/filter', 1, visibility=None), + message('hidden', 'channel/filter', 2, visibility='direct', source='human:other@example.test'), + message('mine', 'channel/own', 2, visibility='direct', source='human:' + ME), + message('unicode', 'channel/你好/child', 1), + message('prefix-child', 'channel/subtype', 1, type='workspace.message.posted.extra')] + rows[4]['network_id'] = OTHER + got = previews(query_db, rows, prefix, human) + expected = {'channel/legacy': 'new' if prefix else 'control', + 'channel/filter': 'visible' if human else 'hidden', + 'channel/own': 'mine', 'channel/你好/child': 'unicode', + 'channel/subtype': 'prefix-child'} + if not human: + expected['channel/private'] = 'private-new' + assert {k: r['id'] for k, r in got.items()} == expected + for row in got.values(): + assert row['payload']['content'] == row['id'] + assert row['metadata'] == {'id': row['id']} + + +def test_equal_timestamps_use_stable_id_tiebreak(query_db): + got = previews(query_db, [message('a'), message('z'), message('b')]) + assert got['channel/legacy']['id'] == 'z' + + +@pytest.mark.parametrize('kind', ['other-workspace', 'no-channels', 'no-matching-type', 'hidden-only']) +def test_empty_result(query_db, kind): + row = message('one') + if kind == 'other-workspace': + row['network_id'] = OTHER + elif kind == 'no-channels': + row['target'] = 'agent/one' + elif kind == 'no-matching-type': + row['type'] = 'control' + else: + row['target'] = 'channel/private' + assert previews(query_db, [row], human=True) == {}