fix(audit-trail): match search and fields against served values, not withheld ones [PRD-1295]

packages/agent/src/audit-trail/README.md

@@ -214,6 +214,13 @@ That test only runs when the snapshot can actually answer it. The capture keeps
 a field stored redacted — has no honest answer in the snapshot and the values are withheld rather
 than matched against a missing key: absent is not the same as passing.
 
+A captured `null` is tested the way the database tests a `NULL`: it answers only a condition asking
+for the null itself (`Blank`, `Missing`, `Equal` null, an `In` list holding null), never a negated or
+ordered one. In memory `status != 'private'` holds for a null status and `null < 5` coerces to
+`0 < 5`, while the scoped read that guarded the live record left that record out — without this, a
+record the caller could never read alive would become readable once deleted. On a datasource whose
+own `!=` keeps a NULL (Mongo's `$ne`), this is stricter than the live read, never looser.
+
 Primary keys are the exception, read back from the row's own id — but only for the side that id
 speaks for. A row is filed under the identity the record ended up with, so its id answers for a
 `create`, a `delete` and the new side of an `update`, never for what an update moved away from. A
@@ -292,9 +299,13 @@ ever written, so the real value was never in the database to find.
 Nor can a search confirm a value the scope withholding hides. On a record gone for good under a
 caller's permission scope, `search` and `fields` are matched against the values as served, never as
 captured — in SQL, which rows come back, `meta.count` and `availableUsers` would each say whether a
-withheld value holds the term. The rows are read without those two filters, in batches of 500, and
-matched and paged in memory, keeping only the requested page and the authors. That holds too for a
-record deleted while the request was in flight, whose SQL-matched answer is discarded and re-read.
+withheld value holds the term. The rows are read without those two filters, in batches of 500 that
+each continue past the last row read, never at an offset, and bounded at the instant the scan
+starts. They are matched and paged as they go, keeping only the page asked for. Only a gone
+record's history pays that scan, and it is one record's history — the same rows the SQL search
+would have scanned without an index. That holds too for a record deleted while the request was in flight:
+the second read of the record decides the withholding, so the SQL-matched answer is discarded and
+the history scanned the same way.
 
 Matching the *serialized* text rather than a structural walk of the parsed value is cheap and still
 correct for "keys and scalar values" — but two things follow from it. A punctuation-only term (`,`,

packages/agent/src/audit-trail/sql-store.ts

@@ -199,6 +199,8 @@ function buildHistoryWhereClause(
     endTimestamp,
     fields,
     search,
+    order,
+    after,
   }: AuditHistoryQuery,
   sequelize: Sequelize,
 ): Record<string | symbol, unknown> {
@@ -215,6 +217,16 @@ function buildHistoryWhereClause(
   const andConditions = [];
   if (fields?.length) andConditions.push(fieldsChangedCondition(sequelize, fields));
   if (search) andConditions.push(searchCondition(sequelize, search));
+
+  if (after) {
+    const past = order === 'desc' ? Op.lt : Op.gt;
+    const at = new Date(after.timestamp);
+
+    andConditions.push({
+      [Op.or]: [{ timestamp: { [past]: at } }, { timestamp: at, id: { [past]: after.id } }],
+    });
+  }
+
   if (andConditions.length) where[Op.and] = andConditions;
 
   return where;

packages/agent/src/audit-trail/types.ts

@@ -71,6 +71,11 @@ export type AuditHistoryQuery = {
   search?: string;
   /** Sort direction on `timestamp` (ties broken by insertion order). Defaults to `'asc'`. */
   order?: 'asc' | 'desc';
+  /**
+   * Keep only the rows strictly past this one in `order`, so a scan keyed on the last row it read
+   * neither repeats nor skips one when entries are written between its reads, as `skip` would.
+   */
+  after?: Pick<AuditRecord, 'timestamp' | 'id'>;
 };
 
 export type AuditCorrelationQuery = {

packages/agent/src/audit-trail/withhold.ts

@@ -1,5 +1,11 @@
 import type { AuditRecord } from './types';
-import type { Collection, ConditionTree, Logger } from '@forestadmin/datasource-toolkit';
+import type {
+  Collection,
+  ConditionTree,
+  ConditionTreeBranch,
+  ConditionTreeLeaf,
+  Logger,
+} from '@forestadmin/datasource-toolkit';
 
 import { SchemaUtils } from '@forestadmin/datasource-toolkit';
 
@@ -23,6 +29,50 @@ export type Withholding = {
  */
 type IdAnswersForSide = boolean;
 
+function asksForNull({ operator, value }: ConditionTreeLeaf): boolean {
+  switch (operator) {
+    case 'Blank':
+    case 'Missing':
+      return true;
+    case 'Equal':
+      return value === null || value === undefined;
+    case 'In':
+      return Array.isArray(value) && value.some(item => item === null || item === undefined);
+    default:
+      return false;
+  }
+}
+
+// `ConditionTree.match`, except that a null answers only a condition asking for the null itself, as
+// the database tests it. In memory `status != 'private'` holds for a null status and `null < 5`
+// coerces to `0 < 5`, while the scoped read that guarded the live record left that NULL out: a record
+// the caller could never read alive would become readable once deleted. Stricter than a datasource
+// that matches NULL there (Mongo's `$ne`), never looser.
+function matchesAsStored(
+  tree: ConditionTree,
+  values: Record<string, unknown>,
+  collection: Collection,
+  timezone: string,
+): boolean {
+  // By shape, not `instanceof`: a scope can be built by another copy of the toolkit.
+  if ('aggregator' in tree) {
+    const branch = tree as ConditionTreeBranch;
+    const evaluate = (condition: ConditionTree) =>
+      matchesAsStored(condition, values, collection, timezone);
+
+    return branch.aggregator === 'And'
+      ? branch.conditions.every(evaluate)
+      : branch.conditions.some(evaluate);
+  }
+
+  const leaf = tree as ConditionTreeLeaf;
+  const value = values[leaf.field];
+
+  return value === null || value === undefined
+    ? asksForNull(leaf)
+    : leaf.match(values, collection, timezone);
+}
+
 // Only a snapshot that answers every field the permission scope asks about is worth matching. The
 // capture keeps the writable columns, so a permission scope reaching for anything else — a
 // read-only column, a relation — reads `undefined` there and would answer for a value the row never
@@ -43,7 +93,7 @@ export function permissionScopeAccepts(
     Object.prototype.hasOwnProperty.call(values, field),
   );
 
-  return answered && permissionScope.match(values, collection, timezone);
+  return answered && matchesAsStored(permissionScope, values, collection, timezone);
 }
 
 /**

packages/agent/src/routes/access/audit-trail.ts

@@ -38,7 +38,7 @@ const ISO_INSTANT = /[Zz]$|[+-]\d{2}:?\d{2}$/;
 
 const DEFAULT_PAGE_SIZE = 20;
 const MAX_PAGE_SIZE = 100;
-const SERVED_MATCH_BATCH_SIZE = 500;
+const SCAN_BATCH_SIZE = 500;
 
 const AUDIT_OPERATIONS: readonly AuditOperation[] = [
   'create',
@@ -57,14 +57,6 @@ type AuditHistoryFilters = {
   search?: string;
 };
 
-type ServedMatchQuery = Pick<AuditHistoryFilters, 'fields' | 'search'> & {
-  rowFilters: AuditHistoryQuery;
-  order: 'asc' | 'desc';
-  skip: number;
-  limit: number;
-  isFirstFetch: boolean;
-};
-
 export default class AuditTrailRoute extends CollectionRoute {
   setupRoutes(router: Router): void {
     router.get(`/_audit-trail/${this.collectionUrlSlug}/:id`, this.handleHistory.bind(this));
@@ -110,18 +102,28 @@ export default class AuditTrailRoute extends CollectionRoute {
     const isFirstFetch =
       (context.request.query as Record<string, unknown>)['page[number]'] === undefined;
 
-    // Matched in SQL, `search` and `fields` test the values as captured, so which rows come back,
-    // the count and the authors would still say what the withholding below hides — one probe per
-    // character. For a gone record they are matched against the values served instead.
     const filtersOnValues = Boolean(fields || search);
-    const servedMatchQuery = { rowFilters, order, fields, search, skip, limit, isFirstFetch };
 
-    if (permissionScope && filtersOnValues && goneEntirely) {
-      context.response.body = await this.listServedMatches(
-        servedMatchQuery,
-        permissionScope,
-        context,
-      );
+    const serveMatchedValues = async (scope: ConditionTree) => {
+      const matched = await this.scanServedValues(context, scope, rowFilters, {
+        fields,
+        search,
+        order,
+        skip,
+        limit,
+      });
+
+      context.response.body = {
+        data: matched.page,
+        meta: {
+          count: matched.count,
+          ...(isFirstFetch && { availableUsers: [...matched.authors.values()] }),
+        },
+      };
+    };
+
+    if (permissionScope && goneEntirely && filtersOnValues) {
+      await serveMatchedValues(permissionScope);
 
       return;
     }
@@ -154,13 +156,10 @@ export default class AuditTrailRoute extends CollectionRoute {
 
     const gone = after ? after.goneEntirely : goneEntirely;
 
-    // Gone between the check and the read: the rows, count and authors above were matched in SQL.
-    if (permissionScope && filtersOnValues && gone) {
-      context.response.body = await this.listServedMatches(
-        servedMatchQuery,
-        permissionScope,
-        context,
-      );
+    // Gone between the check and the read: this answer withholds, so the rows, count and authors
+    // matched in SQL above must not decide what is served either.
+    if (permissionScope && gone && filtersOnValues) {
+      await serveMatchedValues(permissionScope);
 
       return;
     }
@@ -179,56 +178,67 @@ export default class AuditTrailRoute extends CollectionRoute {
     };
   }
 
-  // Read in batches so a gone record's history never sits in memory whole: only the requested
-  // page and the distinct authors are kept while the count runs over every match.
-  private async listServedMatches(
-    { rowFilters, order, fields, search, skip, limit, isFirstFetch }: ServedMatchQuery,
-    permissionScope: ConditionTree,
+  // Matched in SQL, `search` and `fields` test the values as captured, so which rows come back, the
+  // count and the authors would still say what the withholding hides — one probe per character. For
+  // a gone record they are matched against the values served instead, which means
+  // scanning the whole history: in batches, keeping only the page asked for, each continuing past
+  // the last row read rather than at an offset that entries written in between would shift, and
+  // bounded at the instant the scan starts so an id taken since cannot keep it chasing new rows.
+  private async scanServedValues(
     context: Context,
-  ): Promise<{
-    data: AuditRecord[];
-    meta: { count: number; availableUsers?: AuditUserSummary[] };
-  }> {
+    permissionScope: ConditionTree,
+    rowFilters: Omit<AuditHistoryQuery, 'fields' | 'search' | 'skip' | 'limit' | 'order' | 'after'>,
+    {
+      fields,
+      search,
+      order,
+      skip,
+      limit,
+    }: AuditHistoryFilters & { order: 'asc' | 'desc'; skip: number; limit: number },
+  ): Promise<{ page: AuditRecord[]; count: number; authors: Map<number, AuditUserSummary> }> {
     const { store } = this.options.auditTrail;
+    const now = new Date().toISOString();
+    const endTimestamp =
+      rowFilters.endTimestamp && rowFilters.endTimestamp < now ? rowFilters.endTimestamp : now;
     const page: AuditRecord[] = [];
     const authors = new Map<number, AuditUserSummary>();
     let count = 0;
+    let after: AuditRecord | undefined;
+    let rows: AuditRecord[];
 
-    for (let offset = 0; ; offset += SERVED_MATCH_BATCH_SIZE) {
-      // Sequential on purpose: batches are held one at a time.
+    do {
       // eslint-disable-next-line no-await-in-loop
-      const batch = await store.listByRecord({
+      rows = await store.listByRecord({
         ...rowFilters,
+        endTimestamp,
         order,
-        skip: offset,
-        limit: SERVED_MATCH_BATCH_SIZE,
+        limit: SCAN_BATCH_SIZE,
+        ...(after && { after: { timestamp: after.timestamp, id: after.id } }),
       });
 
-      const matched = this.withhold(batch, permissionScope, context).filter(entry =>
+      const matched = this.withhold(rows, permissionScope, context).filter(entry =>
         AuditTrailRoute.matchesServedValues(entry, fields, search),
       );
-      const pageStart = Math.max(0, skip - count);
-
-      page.push(...matched.slice(pageStart, pageStart + limit - page.length));
-      matched.forEach(entry => {
-        if (authors.has(entry.userId)) return;
-
-        authors.set(entry.userId, {
-          id: entry.userId,
-          firstName: entry.userFirstName,
-          lastName: entry.userLastName,
-          email: entry.userEmail,
-        });
-      });
-      count += matched.length;
 
-      if (batch.length < SERVED_MATCH_BATCH_SIZE) break;
-    }
+      for (const entry of matched) {
+        if (count >= skip && page.length < limit) page.push(entry);
+        count += 1;
+
+        // An author reads as their latest identity whichever way the history is sorted.
+        if (order === 'asc' || !authors.has(entry.userId)) {
+          authors.set(entry.userId, {
+            id: entry.userId,
+            firstName: entry.userFirstName,
+            lastName: entry.userLastName,
+            email: entry.userEmail,
+          });
+        }
+      }
 
-    return {
-      data: page,
-      meta: { count, ...(isFirstFetch && { availableUsers: [...authors.values()] }) },
-    };
+      after = rows[rows.length - 1];
+    } while (rows.length === SCAN_BATCH_SIZE);
+
+    return { page, count, authors };
   }
 
   // The SQL store's `fieldsChangedCondition` and `searchCondition`, run on the values as served.

packages/agent/test/audit-trail/in-memory-store.test.ts

@@ -153,6 +153,32 @@ describe('InMemoryAuditStore', () => {
     expect(page2.map(r => r.newValues)).toEqual([{ n: 3 }]);
   });
 
+  describe('listByRecord after a cursor', () => {
+    const history = { collection: 'accounts', recordId: '1' };
+    const seedTied = (store: InMemoryAuditStore) =>
+      [1, 2, 3].map(id =>
+        store.seed(record({ id, ...(id === 3 && { timestamp: '2026-01-02T00:00:00.000Z' }) })),
+      );
+
+    test('continues past the cursor oldest first, breaking a timestamp tie by id', () => {
+      const store = new InMemoryAuditStore();
+      const [first] = seedTied(store);
+
+      const rows = store.listByRecord({ ...history, order: 'asc', after: first });
+
+      expect(rows.map(row => row.id)).toEqual([2, 3]);
+    });
+
+    test('continues past the cursor newest first, breaking a timestamp tie by id', () => {
+      const store = new InMemoryAuditStore();
+      const [, second] = seedTied(store);
+
+      const rows = store.listByRecord({ ...history, order: 'desc', after: second });
+
+      expect(rows.map(row => row.id)).toEqual([1]);
+    });
+  });
+
   describe('countByRecord', () => {
     it('counts all matching entries, ignoring skip and limit', () => {
       const store = new InMemoryAuditStore();

packages/agent/test/audit-trail/in-memory-store.ts

@@ -115,17 +115,19 @@ export default class InMemoryAuditStore implements AuditStore {
     startTimestamp,
     endTimestamp,
     order = 'asc',
+    after,
   }: AuditHistoryQuery): AuditRecord[] {
     const direction = order === 'desc' ? -1 : 1;
+    const compare = (a: Pick<AuditRecord, 'timestamp' | 'id'>, b: typeof a) =>
+      direction * (a.timestamp.localeCompare(b.timestamp) || a.id - b.id);
 
-    // Ties on equal timestamps fall back to insertion order (the stable sort keeps it), which is
-    // the in-memory equivalent of the SQL store's auto-increment id — deterministic and chronological.
     return this.records
       .filter(record => record.collection === collection && record.recordId === recordId)
       .filter(record => !userIds || userIds.includes(record.userId))
       .filter(record => !operations?.length || operations.includes(record.operation))
       .filter(record => !startTimestamp || record.timestamp >= startTimestamp)
       .filter(record => !endTimestamp || record.timestamp <= endTimestamp)
-      .sort((a, b) => direction * a.timestamp.localeCompare(b.timestamp));
+      .filter(record => !after || compare(record, after) > 0)
+      .sort(compare);
   }
 }