perf(workspace): use indexed lookups for channel previews

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')

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/%'")),
     )
 
 

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 = {}

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)

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) == {}