497dfac1: crm activity fails with workspace_source_too_large on prod - read org deals via a paged scoped scan

backend/src/crmWorkspace/activity.ts

@@ -1,5 +1,5 @@
 import { QueryCommand, type DynamoDBDocumentClient } from '@aws-sdk/lib-dynamodb';
-import { getCrmRecord, listCrmRecords } from '../db/sponsorCrm';
+import { getCrmRecord, listOrganizationDeals } from '../db/sponsorCrm';
 import { listTasksByStatus } from '../db/tasks';
 import type { Task } from '../types';
 import { berlinToday } from './berlin';
@@ -228,7 +228,10 @@ export async function readActivity(
     }
 
     if (wanted.has('deal-history') && target === 'organization') {
-      const deals = (await listCrmRecords(client, 'deal')).filter((deal) => String(deal.organizationId) === id);
+      // Org-scoped DEAL# scan, never the whole-workspace snapshot: the snapshot
+      // scan trips the storage bounds on prod-sized sources and fails the whole
+      // read with workspace_source_too_large.
+      const deals = await listOrganizationDeals(client, id);
       for (const deal of deals) {
         const events = await readSource(`HISTORY#${deal.id}`);
         historyReads.push({ booking: deal.id, itemsRead: events.length });
@@ -349,7 +352,10 @@ export async function readActivity(
       relatedWork,
       tasksScope: 'open',
       reads: {
-        snapshot: ['ORGANIZATION#', 'CONTACT#', 'DEAL#', 'DELIVERABLE#'],
+        // What this read actually scanned: keyed queries plus, for organization
+        // deal history, one DEAL#-prefix scan scoped to the record. No
+        // whole-workspace snapshot scan takes place.
+        snapshot: wanted.has('deal-history') && target === 'organization' ? ['DEAL#'] : [],
         queries,
         historyReads,
       },

backend/src/db/sponsorCrm.ts

@@ -147,6 +147,28 @@ export async function listCrmRecords(
     a.id.localeCompare(b.id));
 }
 
+/**
+ * Deals of one organization for single-record reads. A DEAL#-only bounded scan
+ * filtered to the organization: it never populates or reads the workspace
+ * snapshot, and only this organization's deals accumulate inside the page loop,
+ * so prod-sized workspaces stay within the storage bounds.
+ */
+export async function listOrganizationDeals(client: DynamoDBDocumentClient, organizationId: string) {
+  const rows = await collectBoundedPages<Record<string, unknown>>({
+    loadPage: (ExclusiveStartKey, abortSignal) => client.send(
+      new ScanCommand({
+        TableName: TABLE_SPONSOR_CRM,
+        ConsistentRead: true,
+        FilterExpression: 'begins_with(PK, :deal) AND organizationId = :organizationId',
+        ExpressionAttributeValues: { ':deal': 'DEAL#', ':organizationId': organizationId },
+        ...(ExclusiveStartKey ? { ExclusiveStartKey } : {}),
+      }),
+      { abortSignal },
+    ),
+  });
+  return rows.map(item => clean(item)!).sort((a, b) => a.id.localeCompare(b.id));
+}
+
 function messageTargetWrites(message: Record<string, unknown>) {
   const messageId = String(message.id);
   const sentAt = String(message.sentAt || message.createdAt || new Date().toISOString());

backend/tests/crm-workspace-activity.test.ts

@@ -3,7 +3,7 @@ import { setDefaultAssignee } from '../src/db/workspaceSettings';
 import { before, after, describe, it } from 'node:test';
 import assert from 'node:assert';
 import type { DynamoDBDocumentClient } from '@aws-sdk/lib-dynamodb';
-import { BatchWriteCommand, QueryCommand } from '@aws-sdk/lib-dynamodb';
+import { BatchWriteCommand, QueryCommand, ScanCommand } from '@aws-sdk/lib-dynamodb';
 import { TABLE_SPONSOR_CRM } from '../src/db/tableNames';
 import { readActivity } from '../src/crmWorkspace/activity';
 import { handler } from '../src/handler';
@@ -110,7 +110,7 @@ describe('crm workspace activity', () => {
     assert.ok(body.reads.queries.some((scan: any) => scan.kind === 'message'));
     assert.ok(body.reads.queries.some((scan: any) => scan.kind === 'note'));
     assert.ok(body.reads.historyReads.length >= 1);
-    assert.ok(body.reads.snapshot.includes('ORGANIZATION#'));
+    assert.ok(body.reads.snapshot.includes('DEAL#'), 'org deal history comes from the scoped DEAL# scan');
     void contact;
   });
 
@@ -259,6 +259,58 @@ describe('crm workspace activity', () => {
     }
   });
 
+  it('assembles deal history from a paged organization-scoped scan, never the workspace snapshot', async () => {
+    const suffix = String(seedCount++);
+    const org = data(await invoke('POST', '/api/sponsor-crm/organizations', {
+      displayName: `Many Deals Org ${suffix}`, domain: `manydeals-${suffix}.example`, sourceKey: `ws-manydeals-org-${suffix}`,
+    }));
+    const dealIds: string[] = [];
+    for (let index = 0; index < 5; index++) {
+      const deal = data(await invoke('POST', '/api/sponsor-crm/deals', {
+        organizationId: org.id, status: 'open', sourceKey: `ws-manydeals-deal-${suffix}-${index}`,
+      }));
+      const fresh = data(await invoke('GET', `/api/sponsor-crm/deals/${deal.id}`));
+      await invoke('PUT', `/api/sponsor-crm/deals/${deal.id}`, { status: 'agreed', version: fresh.version, historyNote: `Phase ${index}` });
+      dealIds.push(deal.id);
+    }
+    // Every scan page is capped at 25 evaluated items so the organization's
+    // deals arrive across many storage pages, like on a prod-sized source.
+    const scans: any[] = [];
+    const observed = { send: async (command: any) => {
+      if (command instanceof ScanCommand) {
+        scans.push(command.input);
+        return client.send(new ScanCommand({ ...command.input, Limit: 25 }));
+      }
+      return client.send(command);
+    } } as unknown as DynamoDBDocumentClient;
+    const seen = new Set<string>();
+    const historyDeals = new Set<string>();
+    let cursor: string | undefined;
+    let pages = 0;
+    do {
+      const response = await readActivity(observed, 'organization', org.id, { limit: '2', cursor });
+      assert.equal(response.status, 200);
+      const body = response.body as any;
+      for (const item of body.items) {
+        const key = `${item.source}:${item.id}`;
+        assert.ok(!seen.has(key), `duplicate ${key}`);
+        seen.add(key);
+        if (item.source === 'deal-history') historyDeals.add(String(item.provenance.dealId));
+      }
+      cursor = body.nextCursor || undefined;
+      assert.ok(++pages < 200, 'pagination must advance to the end');
+    } while (cursor);
+    assert.ok(scans.length > 10, 'the deals scan must paginate across many storage pages');
+    // The too-large request is the whole-workspace snapshot scan (every prefix
+    // ORed together). Only the org-scoped DEAL# scan may ever fire.
+    assert.ok(scans.every((scan) =>
+      scan.FilterExpression === 'begins_with(PK, :deal) AND organizationId = :organizationId' &&
+      scan.ExpressionAttributeValues?.[':deal'] === 'DEAL#' &&
+      scan.ExpressionAttributeValues?.[':organizationId'] === org.id,
+    ), 'activity must scan only this organization\'s deals, never the whole workspace source');
+    assert.deepEqual([...historyDeals].sort(), [...dealIds].sort());
+  });
+
   it('merges all sources without losing events on one-item pages', async () => {
     const { org, contact, deal } = await seed(String(seedCount++));
     for (const [target, id] of [['organization', org.id], ['contact', contact.id], ['deal', deal.id]]) {