feat(cohorts): anon-inclusive event cohorts — property→v2, name-only→raw events

Event-based cohorts silently undercounted anonymous users: both cohort MVs
bake `WHERE profile_id != device_id` at write time, so pre-identify events
never entered them.

- Property criteria ("did X where flag=Y") → profile_event_property_summary_v2
  (anon-inclusive), gated on timeframe start >= COHORTS_V2_START_DATE (default
  2026-07-01, env-overridable); older timeframes fall back to the v1 MV. Pure
  table-name swap — same countMerge(event_count)/event_date aggregates.
- Name-only criteria ("did X", any N×) → raw events (GROUP BY profile_id
  HAVING count() <op> / SELECT DISTINCT). Exact for any frequency,
  anon-inclusive, faster than v2 (whose ARRAY JOIN explodes name-only reads).
  Removes the only cohort read of cohort_events_mv.
- Extract operatorClause into shared filter-operators.ts; use in the cohort
  property branch + profile.service. Fixes doesNotContain/startsWith/gt/regex
  silently falling through to `IN`.
- Drop dead filterClause chain in buildEventCriteriaQuery.

Validated on shortreels: property cohort 13,862 (v1) → 110,651 (v2) in 0.3s.

cohort_events_mv table kept — retention (trpc/chart.ts) still reads it.

Author: ayushjhanwar-png

.env.example

@@ -67,3 +67,11 @@ PROFILES_BEHAVIORAL_V2_PROJECTS=""
 # docs/v2-migration-progress.md). Drop to 2026-07-01 once the 07-10 → 07-12
 # recovery backfill lands.
 PROFILES_BEHAVIORAL_V2_START_DATE="2026-07-12"
+
+# Cohorts: property criteria ("did X where flag=Y") route to the anon-inclusive
+# v2 property MV when the criterion's timeframe STARTS on/after this date (all
+# projects — no allowlist); older timeframes fall back to the v1 anon-excluded
+# MV. Name-only criteria always use raw events (source of truth, no gate).
+# NOTE: the 07-11 v2 gap (~80% coverage) still sits inside this window — bump to
+# 2026-07-12 to exclude it until the 07-11 re-backfill lands.
+COHORTS_V2_START_DATE="2026-07-01"

packages/db/src/services/cohort.service.test.ts

@@ -16,16 +16,23 @@ import {
   updateCohortMembership,
 } from './cohort.service';
 
-// Mock the ch and db modules
+// Mock the ch and db modules. mockChQuery is created via vi.hoisted so the
+// hoisted vi.mock factory can reference it (a plain top-level const would be
+// initialised AFTER the hoisted factory runs → ReferenceError).
+const { mockChQuery } = vi.hoisted(() => ({ mockChQuery: vi.fn() }));
+
 vi.mock('../clickhouse/client', () => ({
   ch: vi.fn(),
-  chQuery: vi.fn(),
+  chQuery: mockChQuery,
   TABLE_NAMES: {
     events: 'events',
     profiles: 'profiles',
     cohort_members: 'cohort_members',
     cohort_metadata: 'cohort_metadata',
     profile_event_summary_mv: 'profile_event_summary_mv',
+    profile_event_property_summary_mv: 'profile_event_property_summary_mv',
+    profile_event_property_summary_v2: 'profile_event_property_summary_v2',
+    cohort_events_mv: 'cohort_events_mv',
   },
 }));
 
@@ -38,15 +45,6 @@ vi.mock('../prisma-client', () => ({
   },
 }));
 
-const mockChQuery = vi.fn();
-vi.mock('../clickhouse/client', async () => {
-  const actual = await vi.importActual('../clickhouse/client');
-  return {
-    ...actual,
-    chQuery: mockChQuery,
-  };
-});
-
 describe('Cohort Service', () => {
   beforeEach(() => {
     vi.clearAllMocks();
@@ -62,7 +60,7 @@ describe('Cohort Service', () => {
               name: 'page_view',
               filters: [],
               timeframe: { type: 'relative', value: '30d' },
-              frequency: { operator: 'gte', value: 1 },
+              frequency: { operator: 'at_least', count: 1 },
             },
           ],
           operator: 'or',
@@ -94,7 +92,7 @@ describe('Cohort Service', () => {
               name: 'purchase',
               filters: [],
               timeframe: { type: 'relative', value: '7d' },
-              frequency: { operator: 'gte', value: 1 },
+              frequency: { operator: 'at_least', count: 1 },
             },
           ],
           operator: 'and',
@@ -111,7 +109,7 @@ describe('Cohort Service', () => {
       expect(result).toEqual(['user1', 'user2']);
       expect(mockChQuery).toHaveBeenCalledTimes(1);
       // Verify INTERSECT is used for AND
-      const query = mockChQuery.mock.calls[0][0];
+      const query = mockChQuery.mock.calls[0]![0];
       expect(query).toContain('INTERSECT');
     });
 
@@ -131,7 +129,7 @@ describe('Cohort Service', () => {
                 },
               ],
               timeframe: { type: 'relative', value: '7d' },
-              frequency: { operator: 'gte', value: 3 },
+              frequency: { operator: 'at_least', count: 3 },
             },
           ],
           operator: 'or',
@@ -154,7 +152,7 @@ describe('Cohort Service', () => {
             {
               name: 'signup',
               filters: [],
-              timeframe: { type: 'absolute', value: '2024-01-01' },
+              timeframe: { type: 'absolute', start: '2024-01-01' },
             },
           ],
           operator: 'or',
@@ -166,7 +164,7 @@ describe('Cohort Service', () => {
       const result = await computeEventBasedCohort('project-123', definition);
 
       expect(result).toEqual(['user1']);
-      const query = mockChQuery.mock.calls[0][0];
+      const query = mockChQuery.mock.calls[0]![0];
       expect(query).toContain('2024-01-01');
     });
   });
@@ -180,7 +178,7 @@ describe('Cohort Service', () => {
             {
               id: 'p1',
               name: 'email',
-              operator: 'isSet',
+              operator: 'isNotNull',
               value: [],
             },
           ],
@@ -267,7 +265,7 @@ describe('Cohort Service', () => {
             {
               id: 'p1',
               name: 'email',
-              operator: 'isSet',
+              operator: 'isNotNull',
               value: [],
             },
           ],
@@ -288,13 +286,14 @@ describe('Cohort Service', () => {
       mockChQuery.mockResolvedValueOnce(undefined);
 
       await storeCohortMembership(
-        'cohort-123',
         'project-123',
+        'cohort-123',
         ['user1', 'user2', 'user3'],
+        1,
       );
 
       expect(mockChQuery).toHaveBeenCalledTimes(1);
-      const query = mockChQuery.mock.calls[0][0];
+      const query = mockChQuery.mock.calls[0]![0];
       expect(query).toContain('cohort_members');
       expect(query).toContain('cohort-123');
       expect(query).toContain('project-123');
@@ -303,7 +302,7 @@ describe('Cohort Service', () => {
     it('should handle empty profile list', async () => {
       mockChQuery.mockResolvedValueOnce(undefined);
 
-      await storeCohortMembership('cohort-123', 'project-123', []);
+      await storeCohortMembership('project-123', 'cohort-123', [], 1);
 
       // Should still execute to clear old members
       expect(mockChQuery).toHaveBeenCalledTimes(1);

packages/db/src/services/cohort.service.ts

@@ -11,7 +11,13 @@ import type { IChartEventFilter } from '@openpanel/validation';
 
 import { ch, chQuery, TABLE_NAMES } from '../clickhouse/client';
 import { db } from '../prisma-client';
-import { getEventFiltersWhereClause } from './chart.service';
+import { operatorClause } from './filter-operators';
+
+// v2 property MV (profile_event_property_summary_v2) is anon-inclusive but only
+// backfilled from this date forward; June/pre-July is dirty/partial. Property
+// cohorts whose timeframe STARTS on/after this route to v2 (all projects); older
+// ones fall back to the anon-excluded v1 MV. Env-overridable coverage date.
+const COHORTS_V2_START_DATE = process.env.COHORTS_V2_START_DATE || '2026-07-01';
 
 /**
  * Build time constraint SQL from timeframe
@@ -39,6 +45,32 @@ function buildTimeConstraint(timeframe: Timeframe): string {
   }
 }
 
+/**
+ * The UTC start day of a criterion's timeframe. Relative "Nd" resolves to
+ * N days before today; absolute uses its `start`. Used only to gate v1 vs v2.
+ */
+function criteriaTimeframeStart(timeframe: Timeframe): Date {
+  if (timeframe.type === 'relative') {
+    const match = timeframe.value.match(/^(\d+)d$/);
+    const days = match ? Number.parseInt(match[1]!, 10) : 0;
+    const d = new Date();
+    d.setUTCDate(d.getUTCDate() - days);
+    return d;
+  }
+  return new Date(`${timeframe.start}T00:00:00Z`);
+}
+
+/**
+ * Property cohorts route to the anon-inclusive v2 MV only when the whole
+ * timeframe sits within v2's coverage window (start >= COHORTS_V2_START_DATE);
+ * otherwise fall back to v1 so we never read a partial/absent v2 range.
+ */
+function canRouteCohortToV2(criteria: EventCriteria): boolean {
+  const start = criteriaTimeframeStart(criteria.timeframe);
+  const gate = new Date(`${COHORTS_V2_START_DATE}T00:00:00Z`);
+  return start.getTime() >= gate.getTime();
+}
+
 /**
  * Convert frequency operator to SQL comparison
  */
@@ -74,56 +106,39 @@ export function buildEventCriteriaQuery(
     ? `AND profile_id IN (${profileIdPrefilter})`
     : '';
 
-  // Build event filters
-  const filterWhere = filters.length > 0
-    ? getEventFiltersWhereClause(filters)
-    : {};
-  const filterClauses = Object.values(filterWhere);
-  const filterClause = filterClauses.length > 0
-    ? `AND ${filterClauses.join(' AND ')}`
-    : '';
-
   // Check if there are event property filters
   const hasEventPropertyFilters = filters.some(
     (f) => f.name.startsWith('properties.') && !f.name.startsWith('profile.properties.')
   );
 
-  // Use property-aware MV for event property filters
+  // PROPERTY criteria ("did X where flag=Y") → property summary MV. Anon-inclusive
+  // v2 when the timeframe sits in v2's coverage window, else the v1 fallback. Both
+  // MVs share the same schema (name/property_key/property_value/event_date/
+  // countMerge(event_count)/profile_id) so this is a pure table-name swap.
   if (hasEventPropertyFilters) {
+    const propertyTable = canRouteCohortToV2(criteria)
+      ? TABLE_NAMES.profile_event_property_summary_v2
+      : TABLE_NAMES.profile_event_property_summary_mv;
+
     const propertyFilters = filters.filter(
       (f) => f.name.startsWith('properties.')
     );
 
-    // Build WHERE conditions for property filters
+    // One `(property_key = k AND <value predicate>)` per filter, OR'd together.
+    // operatorClause handles ALL operators (contains/doesNotContain/gt/regex/…);
+    // the old inline switch fell through to `IN` for anything but is/isNot/
+    // contains/doesNotContain — silently inverting doesNotContain etc.
     const propertyConditions = propertyFilters.map((filter) => {
       const propertyKey = filter.name.replace('properties.', '');
       const { value, operator } = filter;
-
-      switch (operator) {
-        case 'is':
-          if (value.length === 1) {
-            return `(property_key = ${sqlstring.escape(propertyKey)} AND property_value = ${sqlstring.escape(String(value[0]).trim())})`;
-          }
-          return `(property_key = ${sqlstring.escape(propertyKey)} AND property_value IN (${value.map((val) => sqlstring.escape(String(val).trim())).join(', ')}))`;
-        case 'isNot':
-          if (value.length === 1) {
-            return `(property_key = ${sqlstring.escape(propertyKey)} AND property_value != ${sqlstring.escape(String(value[0]).trim())})`;
-          }
-          return `(property_key = ${sqlstring.escape(propertyKey)} AND property_value NOT IN (${value.map((val) => sqlstring.escape(String(val).trim())).join(', ')}))`;
-        case 'contains':
-          return `(property_key = ${sqlstring.escape(propertyKey)} AND (${value.map((val) => `property_value LIKE ${sqlstring.escape(`%${String(val).trim()}%`)}`).join(' OR ')}))`;
-        case 'doesNotContain':
-          return `(property_key = ${sqlstring.escape(propertyKey)} AND (${value.map((val) => `property_value NOT LIKE ${sqlstring.escape(`%${String(val).trim()}%`)}`).join(' AND ')}))`;
-        default:
-          return `(property_key = ${sqlstring.escape(propertyKey)} AND property_value IN (${value.map((val) => sqlstring.escape(String(val).trim())).join(', ')}))`;
-      }
+      return `(property_key = ${sqlstring.escape(propertyKey)} AND ${operatorClause('property_value', operator, value)})`;
     }).join(' OR ');
 
     if (frequency) {
       const frequencyOp = getFrequencyOperator(frequency);
       return `
         SELECT profile_id
-        FROM ${TABLE_NAMES.profile_event_property_summary_mv}
+        FROM ${propertyTable}
         WHERE project_id = ${sqlstring.escape(projectId)}
           AND name = ${sqlstring.escape(name)}
           AND ${timeConstraint.replace('created_at', 'event_date')}
@@ -136,7 +151,7 @@ export function buildEventCriteriaQuery(
 
     return `
       SELECT DISTINCT profile_id
-      FROM ${TABLE_NAMES.profile_event_property_summary_mv}
+      FROM ${propertyTable}
       WHERE project_id = ${sqlstring.escape(projectId)}
         AND name = ${sqlstring.escape(name)}
         AND ${timeConstraint.replace('created_at', 'event_date')}
@@ -145,27 +160,31 @@ export function buildEventCriteriaQuery(
     `;
   }
 
-  // cohort_events_mv: name is 2nd in sort key (vs profile_id 2nd in
-  // profile_event_summary_mv) — much better prefix match for these filters.
+  // NAME-ONLY criteria ("did X", any N×) → raw events. The property MV explodes
+  // each event into one row per property (ARRAY JOIN), so countMerge over it
+  // massively overcounts frequency; and cohort_events_mv is anon-excluded. Raw
+  // events is exact for any frequency, anon-inclusive, and faster than v2 here
+  // (name is an effective sort-key prefix via the proj_funnel projection). Uses
+  // created_at + plain count() — no event_date/countMerge rewrite.
   if (frequency) {
     const frequencyOp = getFrequencyOperator(frequency);
 
     return `
       SELECT profile_id
-      FROM ${TABLE_NAMES.cohort_events_mv}
+      FROM ${TABLE_NAMES.events}
       WHERE project_id = ${sqlstring.escape(projectId)}
         AND name = ${sqlstring.escape(name)}
         AND ${timeConstraint}
         ${prefilterClause}
       GROUP BY profile_id
-      HAVING sum(event_count) ${frequencyOp}
+      HAVING count() ${frequencyOp}
     `;
   }
 
   // For simple "did event" queries
   return `
     SELECT DISTINCT profile_id
-    FROM ${TABLE_NAMES.cohort_events_mv}
+    FROM ${TABLE_NAMES.events}
     WHERE project_id = ${sqlstring.escape(projectId)}
       AND name = ${sqlstring.escape(name)}
       AND ${timeConstraint}

packages/db/src/services/filter-operators.ts

@@ -0,0 +1,47 @@
+import sqlstring from 'sqlstring';
+
+// Full operator support for a value filter, shared by the profiles behavioural
+// filter and cohort event-property criteria. `col` is the SQL expression the
+// operator compares (e.g. `property_value`, `properties['x']`, a bare column).
+//
+// Fixes the old "every unhandled operator falls through to IN" bug — e.g.
+// `doesNotContain` used to emit `col IN (...)`, the exact OPPOSITE set.
+export function operatorClause(
+  col: string,
+  operator: string | undefined,
+  values: (string | number | boolean | null)[],
+): string {
+  const trimmed = values.map((v) => String(v).trim());
+  const esc = (v: string) => sqlstring.escape(v);
+  const inList = trimmed.map(esc).join(', ');
+  const anyLike = (pat: (v: string) => string) =>
+    `(${trimmed.map((v) => `${col} LIKE ${esc(pat(v))}`).join(' OR ')})`;
+  switch (operator) {
+    case 'isNot':
+      return `${col} NOT IN (${inList})`;
+    case 'contains':
+      return anyLike((v) => `%${v}%`);
+    case 'doesNotContain':
+      return `(${trimmed.map((v) => `${col} NOT LIKE ${esc(`%${v}%`)}`).join(' OR ')})`;
+    case 'startsWith':
+      return anyLike((v) => `${v}%`);
+    case 'endsWith':
+      return anyLike((v) => `%${v}`);
+    case 'regex':
+      return `(${trimmed.map((v) => `match(${col}, ${esc(v)})`).join(' OR ')})`;
+    case 'isNull':
+      return `(${col} = '' OR ${col} IS NULL)`;
+    case 'isNotNull':
+      return `(${col} != '' AND ${col} IS NOT NULL)`;
+    case 'gt':
+      return `(${trimmed.map((v) => `toFloat64OrZero(${col}) > toFloat64(${esc(v)})`).join(' OR ')})`;
+    case 'lt':
+      return `(${trimmed.map((v) => `toFloat64OrZero(${col}) < toFloat64(${esc(v)})`).join(' OR ')})`;
+    case 'gte':
+      return `(${trimmed.map((v) => `toFloat64OrZero(${col}) >= toFloat64(${esc(v)})`).join(' OR ')})`;
+    case 'lte':
+      return `(${trimmed.map((v) => `toFloat64OrZero(${col}) <= toFloat64(${esc(v)})`).join(' OR ')})`;
+    default: // 'is' and any unknown operator
+      return `${col} IN (${inList})`;
+  }
+}