Fix daily alert

Do not send an alert again for someone who modified something in their profile that's unrelated to the alert

backend/api/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@compass/api",
-  "version": "1.45.0",
+  "version": "1.45.1",
   "private": true,
   "description": "Backend API endpoints",
   "main": "src/serve.ts",

backend/api/src/get-profiles.ts

@@ -18,7 +18,7 @@ import {OptionTableKey} from 'common/profiles/constants'
 import {compact} from 'lodash'
 import {log} from 'shared/monitoring/log'
 import {convertRow} from 'shared/profiles/supabase'
-import {createSupabaseDirectClient, pgp} from 'shared/supabase/init'
+import {createSupabaseDirectClient, pgp, SupabaseDirectClient} from 'shared/supabase/init'
 import {
   from,
   join,
@@ -102,6 +102,10 @@ export type profileQueryType = {
   skipId?: string | undefined
   orderBy?: string | undefined
   lastModificationWithin?: string | undefined
+  /** Restrict the results to these user ids. An empty array matches nothing. */
+  userIds?: string[] | undefined
+  /** Skip the total-count query when the caller only needs the rows */
+  skipCount?: boolean | undefined
   last_active?: string | undefined
   locale?: string | undefined
 } & {
@@ -156,8 +160,9 @@ let profileCols: any
 export const getProfileCols = async () => {
   if (profileCols) return profileCols
   const pg = createSupabaseDirectClient()
+  // table_schema matters: the search-alert snapshot schemas hold a `profiles` table too.
   const rows = await pg.manyOrNone<{column_name: string}>(
-    `SELECT column_name FROM information_schema.columns WHERE table_name = 'profiles' ORDER BY ordinal_position`,
+    `SELECT column_name FROM information_schema.columns WHERE table_schema = 'public' AND table_name = 'profiles' ORDER BY ordinal_position`,
   )
   profileCols = rows
     .map((r) => r.column_name)
@@ -168,9 +173,12 @@ export const getProfileCols = async () => {
   return profileCols
 }
 
-export const loadProfiles = async (props: profileQueryType) => {
-  const pg = createSupabaseDirectClient()
-  log('get-profiles', props)
+/**
+ * @param db pass a client bound to a snapshot schema (see `withSchema` in `profile-snapshot.ts`) to
+ *   evaluate the same filters against the profiles as they were at an earlier point in time.
+ */
+export const loadProfiles = async (props: profileQueryType, db?: SupabaseDirectClient) => {
+  const pg = db ?? createSupabaseDirectClient()
   const {
     limit: limitParam,
     after,
@@ -226,11 +234,15 @@ export const loadProfiles = async (props: profileQueryType) => {
     compatibleWithUserId,
     orderBy: orderByParam = 'created_time',
     lastModificationWithin,
+    userIds,
+    skipCount,
     skipId,
     locale = 'en',
     last_active,
   } = props
 
+  log('get-profiles', {...props, userIds: userIds && `${userIds.length} ids`})
+
   const filterLocation = lat && lon && radius
   const filterRaisedInLocation = raised_in_lat && raised_in_lon && raised_in_radius
 
@@ -600,6 +612,8 @@ export const loadProfiles = async (props: profileQueryType) => {
 
     skipId && where(`profiles.user_id != $(skipId)`, {skipId}),
 
+    userIds && where(`profiles.user_id = any($(userIds))`, {userIds}),
+
     !shortBio &&
       where(
         `bio_length >= ${100}
@@ -679,7 +693,9 @@ export const loadProfiles = async (props: profileQueryType) => {
 
   const countQuery = renderSql(select(`count(*) as count`), ...tableSelection, ...filters)
 
-  const count = await pg.one<number>(countQuery, [], (r) => Number(r.count))
+  const count = skipCount
+    ? profiles.length
+    : await pg.one<number>(countQuery, [], (r) => Number(r.count))
 
   return {profiles, count}
 }

backend/api/src/profile-snapshot.ts

@@ -0,0 +1,200 @@
+import {debug} from 'common/logger'
+import {isEqual} from 'lodash'
+import {SupabaseDirectClient} from 'shared/supabase/init'
+
+/**
+ * A copy of the profile-side tables as they were at the end of the last successful search-alert run.
+ *
+ * Bookmarked-search alerts are a set diff: a profile is worth an email on the run where it *starts*
+ * matching a search, not on every later edit. Running the very same filter query against this
+ * snapshot answers "did it already match yesterday?" without duplicating any of the filter logic in
+ * `get-profiles.ts` — the tables here carry the same names as their `public` counterparts, so the
+ * generated SQL resolves to them through the `search_path` (see {@link withSchema}).
+ */
+export const SNAPSHOT_SCHEMA = 'profile_snapshot'
+
+/**
+ * Where the next snapshot is built, at the *start* of a run. Both sides of the diff then read from a
+ * snapshot rather than from `public`, so profiles edited while the run is in flight belong to
+ * neither side and are picked up by the next run instead of being silently absorbed.
+ *
+ * It is promoted to {@link SNAPSHOT_SCHEMA} only once every search has been processed. A run that
+ * dies halfway therefore leaves it behind, and the next run resumes against it.
+ */
+export const STAGING_SCHEMA = 'profile_snapshot_staging'
+
+/**
+ * The tables `loadProfiles` filters on that a profile owner can actually change.
+ *
+ * Everything else it touches keeps resolving to `public` through the search_path, which is what we
+ * want: the option tables (`interests`, `causes`, `work`, `*_translations`) are static, and
+ * `hidden_profiles` / `compatibility_scores` describe the searcher rather than the profile being
+ * matched.
+ *
+ * `user_activity` is deliberately absent. It churns on every page load, so snapshotting it would
+ * cost more storage than everything else combined. Leaving it in `public` also means a `last_active`
+ * filter sees identical rows on both sides of the diff, so merely coming back online can never look
+ * like a new match.
+ */
+const SNAPSHOT_TABLES = ['profiles', 'users', 'profile_interests', 'profile_causes', 'profile_work']
+
+/** Indexes to recreate on the copies — CTAS keeps no index from the source table. */
+const SNAPSHOT_INDEXES: Record<string, string[]> = {
+  profiles: ['id', 'user_id'],
+  users: ['id'],
+  profile_interests: ['profile_id'],
+  profile_causes: ['profile_id'],
+  profile_work: ['profile_id'],
+}
+
+const MANY_TO_MANY_LABELS = ['interests', 'causes', 'work']
+
+/** Runs `fn` against a client whose unqualified table names resolve to `schema` first. */
+export const withSchema = <T>(
+  pg: SupabaseDirectClient,
+  schema: string,
+  fn: (db: SupabaseDirectClient) => Promise<T>,
+) =>
+  pg.tx(async (t) => {
+    // `set local` is scoped to the transaction, so pooled connections keep their own search_path.
+    await t.none('set local search_path to $(schema:name), public', {schema})
+    const result = await fn(t)
+    // Releasing a savepoint hands its `set local` up to the enclosing transaction, so when `pg` is
+    // itself a transaction the search_path would outlive this call and quietly point later queries
+    // at the snapshot. Restoring is a no-op at the top level, where the commit reverts it anyway.
+    await t.none('set local search_path to default')
+    return result
+  })
+
+export const hasStagingSnapshot = async (pg: SupabaseDirectClient) =>
+  await pg.one<boolean>(
+    `select to_regclass($(table)) is not null as exists`,
+    {table: `${STAGING_SCHEMA}.meta`},
+    (r) => r.exists,
+  )
+
+export const getStagingTakenAt = async (pg: SupabaseDirectClient) =>
+  await pg.one<Date>(`select taken_at from ${STAGING_SCHEMA}.meta`, [], (r) => r.taken_at)
+
+/**
+ * Rebuilds the staging snapshot from `public`. Recreating the tables from scratch on every run
+ * (rather than truncating fixed ones) is what keeps this maintenance-free: a migration that adds a
+ * column to `profiles` needs no matching change here.
+ */
+export const buildStagingSnapshot = async (pg: SupabaseDirectClient) => {
+  await pg.tx(async (t) => {
+    await t.none('drop schema if exists $(schema:name) cascade', {schema: STAGING_SCHEMA})
+    await t.none('create schema $(schema:name)', {schema: STAGING_SCHEMA})
+
+    for (const table of SNAPSHOT_TABLES) {
+      await t.none('create table $(schema:name).$(table:name) as table public.$(table:name)', {
+        schema: STAGING_SCHEMA,
+        table,
+      })
+      for (const column of SNAPSHOT_INDEXES[table]) {
+        await t.none('create index on $(schema:name).$(table:name) ($(column:name))', {
+          schema: STAGING_SCHEMA,
+          table,
+          column,
+        })
+      }
+      await t.none('analyze $(schema:name).$(table:name)', {schema: STAGING_SCHEMA, table})
+    }
+
+    await t.none('create table $(schema:name).meta as select now() as taken_at', {
+      schema: STAGING_SCHEMA,
+    })
+  })
+  debug(`built ${STAGING_SCHEMA}`)
+}
+
+/**
+ * True when the committed snapshot exists and has the same columns as `public`.
+ *
+ * A migration that changes a snapshotted table leaves the old snapshot unable to answer the filter
+ * query (a new filterable column simply would not exist there), so the caller rebuilds instead and
+ * skips one run of alerts. Failing closed costs at most a day of alerts; failing open would send
+ * every user an email about every profile that already matched.
+ */
+export const isSnapshotUsable = async (pg: SupabaseDirectClient) => {
+  const rows = await pg.manyOrNone<{live: string[] | null; snapshot: string[] | null}>(
+    `select
+       (select array_agg(c.column_name::text order by c.column_name)
+          from information_schema.columns c
+         where c.table_schema = 'public' and c.table_name = t.table_name) as live,
+       (select array_agg(c.column_name::text order by c.column_name)
+          from information_schema.columns c
+         where c.table_schema = $(schema) and c.table_name = t.table_name) as snapshot
+     from unnest($(tables)::text[]) as t(table_name)`,
+    {schema: SNAPSHOT_SCHEMA, tables: SNAPSHOT_TABLES},
+  )
+  return rows.every((r) => r.snapshot != null && isEqual(r.live, r.snapshot))
+}
+
+/** Swaps the staging snapshot in as the committed one. Only safe once every search is processed. */
+export const promoteStagingSnapshot = async (pg: SupabaseDirectClient) => {
+  await pg.tx(async (t) => {
+    await t.none('drop schema if exists $(schema:name) cascade', {schema: SNAPSHOT_SCHEMA})
+    await t.none('alter schema $(staging:name) rename to $(snapshot:name)', {
+      staging: STAGING_SCHEMA,
+      snapshot: SNAPSHOT_SCHEMA,
+    })
+  })
+  debug(`promoted ${STAGING_SCHEMA} to ${SNAPSHOT_SCHEMA}`)
+}
+
+const manyToManyCte = (label: string) => `
+    changed_${label} as (
+      select coalesce(staging.profile_id, snapshot.profile_id) as profile_id
+      from (
+        select profile_id, md5(array_agg(option_id order by option_id)::text) as hash
+        from ${STAGING_SCHEMA}.profile_${label}
+        group by profile_id
+      ) staging
+      full outer join (
+        select profile_id, md5(array_agg(option_id order by option_id)::text) as hash
+        from ${SNAPSHOT_SCHEMA}.profile_${label}
+        group by profile_id
+      ) snapshot on snapshot.profile_id = staging.profile_id
+      where staging.hash is distinct from snapshot.hash
+    )`
+
+/**
+ * The users whose profile differs between the staging snapshot and the committed one.
+ *
+ * Hashing the whole row rather than trusting `profiles.last_modification_time` matters: that
+ * column's trigger only fires on `profiles` updates, so it never sees a change to a profile's
+ * interests, causes or work.
+ */
+export const getChangedUserIds = async (pg: SupabaseDirectClient) => {
+  const query = `
+    with staging_profiles as (
+      select p.id, p.user_id, md5(p::text) as hash from ${STAGING_SCHEMA}.profiles p
+    ),
+    changed_profiles as (
+      select coalesce(staging.id, snapshot.id) as id
+      from staging_profiles staging
+      full outer join (
+        select p.id, md5(p::text) as hash from ${SNAPSHOT_SCHEMA}.profiles p
+      ) snapshot on snapshot.id = staging.id
+      where staging.hash is distinct from snapshot.hash
+    ),
+    changed_users as (
+      select coalesce(staging.id, snapshot.id) as id
+      from (select u.id, md5(u::text) as hash from ${STAGING_SCHEMA}.users u) staging
+      full outer join (
+        select u.id, md5(u::text) as hash from ${SNAPSHOT_SCHEMA}.users u
+      ) snapshot on snapshot.id = staging.id
+      where staging.hash is distinct from snapshot.hash
+    ),
+    ${MANY_TO_MANY_LABELS.map(manyToManyCte).join(',')}
+    select distinct staging_profiles.user_id
+    from staging_profiles
+    where staging_profiles.id in (select id from changed_profiles)
+       or staging_profiles.user_id in (select id from changed_users)
+       ${MANY_TO_MANY_LABELS.map(
+         (label) => `or staging_profiles.id in (select profile_id from changed_${label})`,
+       ).join('\n       ')}`
+
+  return await pg.map(query, [], (r) => r.user_id as string)
+}

backend/api/src/send-search-notifications.ts

@@ -1,80 +1,215 @@
 import {loadProfiles, profileQueryType} from 'api/get-profiles'
+import {
+  buildStagingSnapshot,
+  getChangedUserIds,
+  getStagingTakenAt,
+  hasStagingSnapshot,
+  isSnapshotUsable,
+  promoteStagingSnapshot,
+  SNAPSHOT_SCHEMA,
+  STAGING_SCHEMA,
+  withSchema,
+} from 'api/profile-snapshot'
+import {sendDiscordMessage} from 'common/discord/core'
 import {debug} from 'common/logger'
-import {MatchesByUserType} from 'common/profiles/bookmarked_searches'
+import {MatchesType} from 'common/profiles/bookmarked_searches'
 import {Row} from 'common/supabase/utils'
+import {DAY_MS} from 'common/util/time'
 import {sendSearchAlertsEmail} from 'email/functions/helpers'
-import {keyBy} from 'lodash'
-import {createSupabaseDirectClient} from 'shared/supabase/init'
-import {from, renderSql, select} from 'shared/supabase/sql-builder'
+import {groupBy, keyBy, uniq} from 'lodash'
+import {log} from 'shared/monitoring/log'
+import {createSupabaseDirectClient, SupabaseDirectClient} from 'shared/supabase/init'
 
-export function convertSearchRow(row: any): any {
-  return row
+/**
+ * A staging snapshot is only promoted once every search has been processed, so a search whose email
+ * keeps failing would otherwise pin the snapshot in place forever and no new profile would ever be
+ * diffed again. Past this age we promote anyway and lose that user's pending alerts.
+ */
+const MAX_STAGING_AGE_MS = 3 * DAY_MS
+
+// `last_checked_at` lands in the generated types on the next `regen-types-dev`.
+type SearchRow = Row<'bookmarked_searches'> & {last_checked_at: string | null}
+
+type CreatorAlert = {
+  user: Row<'users'>
+  privateUser: any
+  matches: MatchesType[]
+  /** Every pending search of this creator, matched or not — all of them get a fresh watermark. */
+  searchIds: number[]
+  matchedSearchIds: number[]
 }
 
-export const notifyBookmarkedSearch = async (matches: MatchesByUserType) => {
-  for (const [_, value] of Object.entries(matches)) {
-    await sendSearchAlertsEmail(value.user, value.privateUser, value.matches)
+const searchProps = (row: SearchRow, userIds: string[]): profileQueryType => {
+  // orderBy is meaningless here, and 'compatibility_score' would throw for lack of a target user.
+  const {orderBy: _, ...filters} = (row.search_filters ?? {}) as Record<string, any>
+  return {
+    ...filters,
+    skipId: row.creator_id,
+    userId: row.creator_id,
+    shortBio: true,
+    skipCount: true,
+    lastModificationWithin: '24 hours',
+    userIds,
   }
 }
 
+/**
+ * The profiles that match `row` now but did not match it at the last run.
+ *
+ * Both sides run the identical filter query; only the schema they read from differs. Restricting the
+ * first side to the profiles that actually changed keeps a broad search ("any woman") from scanning
+ * everyone, and lets the second side look at a handful of rows.
+ */
+const findNewMatches = async (
+  pg: SupabaseDirectClient,
+  row: SearchRow,
+  changedUserIds: string[],
+) => {
+  const {profiles: candidates} = await withSchema(pg, STAGING_SCHEMA, (db) =>
+    loadProfiles(searchProps(row, changedUserIds), db),
+  )
+  if (!candidates.length) return []
+
+  const {profiles: alreadyMatched} = await withSchema(pg, SNAPSHOT_SCHEMA, (db) =>
+    loadProfiles(
+      searchProps(
+        row,
+        candidates.map((profile: any) => profile.user_id),
+      ),
+      db,
+    ),
+  )
+  const previously = new Set(alreadyMatched.map((profile: any) => profile.user_id))
+
+  return candidates.filter((profile: any) => !previously.has(profile.user_id))
+}
+
+/** Emails each creator, then stamps their searches so a crash never re-sends what already went out. */
+export const notifyBookmarkedSearch = async (
+  pg: SupabaseDirectClient,
+  alerts: Record<string, CreatorAlert>,
+) => {
+  let failed = 0
+  let notified = 0
+
+  for (const [creatorId, alert] of Object.entries(alerts)) {
+    try {
+      if (alert.matches.length) {
+        await sendSearchAlertsEmail(alert.user as any, alert.privateUser, alert.matches)
+        notified++
+      }
+      await pg.none(
+        `update bookmarked_searches
+         set last_checked_at = now(),
+             last_notified_at = case when id = any($(matchedSearchIds)) then now()
+                                     else last_notified_at end
+         where id = any($(searchIds))`,
+        {searchIds: alert.searchIds, matchedSearchIds: alert.matchedSearchIds},
+      )
+    } catch (error) {
+      failed++
+      log.error(`Failed to send search alerts to ${creatorId}`, {error})
+    }
+  }
+
+  return {failed, notified}
+}
+
 export const sendSearchNotifications = async () => {
   const pg = createSupabaseDirectClient()
 
-  const search_query = renderSql(select('bookmarked_searches.*'), from('bookmarked_searches'))
-  const searches = (await pg.map(
-    search_query,
-    [],
-    convertSearchRow,
-  )) as Row<'bookmarked_searches'>[]
+  // A crashed run leaves its staging snapshot behind on purpose. Reusing it means the searches it
+  // already emailed are skipped by their watermark, and the rest see exactly the diff they would
+  // have seen had the run completed.
+  const resuming = await hasStagingSnapshot(pg)
+  if (resuming) log.info('Resuming search notifications against the existing staging snapshot')
+  else await buildStagingSnapshot(pg)
+
+  if (!(await isSnapshotUsable(pg))) {
+    await promoteStagingSnapshot(pg)
+    log.info('Profile snapshot initialised (or rebuilt after a schema change); alerts skipped')
+    return {status: 'success', notified: 0, skipped: true}
+  }
+
+  const changedUserIds = await getChangedUserIds(pg)
+  debug(`${changedUserIds.length} profiles changed since the last run`)
+  if (!changedUserIds.length) {
+    await promoteStagingSnapshot(pg)
+    return {status: 'success', notified: 0}
+  }
+
+  const stagingTakenAt = await getStagingTakenAt(pg)
+  const searches = await pg.manyOrNone<SearchRow>(
+    `select * from bookmarked_searches
+     where last_checked_at is null or last_checked_at < $(stagingTakenAt)`,
+    {stagingTakenAt},
+  )
   debug(`Running ${searches.length} bookmarked searches`)
 
-  const _users = (await pg.map(
-    renderSql(select('users.*'), from('users')),
-    [],
-    convertSearchRow,
-  )) as Row<'users'>[]
-  const users = keyBy(_users, 'id')
-  debug('users', users)
+  const creatorIds = uniq(searches.map((row) => row.creator_id))
+  const users = keyBy(
+    await pg.manyOrNone<Row<'users'>>(`select * from users where id = any($(creatorIds))`, {
+      creatorIds,
+    }),
+    'id',
+  )
+  const privateUsers = keyBy(
+    await pg.manyOrNone<Row<'private_users'>>(
+      `select * from private_users where id = any($(creatorIds))`,
+      {creatorIds},
+    ),
+    'id',
+  )
 
-  const _privateUsers = (await pg.map(
-    renderSql(select('private_users.*'), from('private_users')),
-    [],
-    convertSearchRow,
-  )) as Row<'private_users'>[]
-  const privateUsers = keyBy(_privateUsers, 'id')
-  debug('privateUsers', privateUsers)
+  const alerts: Record<string, CreatorAlert> = {}
 
-  const matches: MatchesByUserType = {}
+  for (const [creatorId, creatorSearches] of Object.entries(groupBy(searches, 'creator_id'))) {
+    if (!users[creatorId] || !privateUsers[creatorId]) continue
 
-  for (const row of searches) {
-    if (typeof row.search_filters !== 'object') continue
-    const {orderBy: _, ...filters} = (row.search_filters ?? {}) as Record<string, any>
-    const props = {
-      ...filters,
-      skipId: row.creator_id,
-      userId: row.creator_id,
-      lastModificationWithin: '24 hours',
-      shortBio: true,
+    const alert: CreatorAlert = {
+      user: users[creatorId],
+      privateUser: privateUsers[creatorId]['data'],
+      matches: [],
+      searchIds: creatorSearches.map((row) => row.id),
+      matchedSearchIds: [],
     }
-    const {profiles} = await loadProfiles(props as profileQueryType)
-    debug(profiles.map((item: any) => item.name))
-    if (!profiles.length) continue
-    if (!(row.creator_id in matches)) {
-      if (!privateUsers[row.creator_id]) continue
-      matches[row.creator_id] = {
-        user: users[row.creator_id],
-        privateUser: privateUsers[row.creator_id]['data'],
-        matches: [],
-      }
+
+    for (const row of creatorSearches) {
+      if (typeof row.search_filters !== 'object') continue
+      const profiles = await findNewMatches(pg, row, changedUserIds)
+      debug(
+        row.id,
+        profiles.map((profile: any) => profile.name),
+      )
+      if (!profiles.length) continue
+
+      alert.matchedSearchIds.push(row.id)
+      alert.matches.push({
+        id: creatorId,
+        description: {filters: row.search_filters, location: row.location},
+        matches: profiles.map((profile: any) => profile.user),
+      })
     }
-    matches[row.creator_id].matches.push({
-      id: row.creator_id,
-      description: {filters: row.search_filters, location: row.location},
-      matches: profiles.map((profile: any) => profile.user),
-    })
+
+    alerts[creatorId] = alert
   }
-  debug('matches:', JSON.stringify(matches, null, 2))
-  await notifyBookmarkedSearch(matches)
 
-  return {status: 'success'}
+  const {failed, notified} = await notifyBookmarkedSearch(pg, alerts)
+
+  const stagingAgeMs = Date.now() - new Date(stagingTakenAt).getTime()
+  if (!failed) {
+    await promoteStagingSnapshot(pg)
+  } else if (stagingAgeMs > MAX_STAGING_AGE_MS) {
+    await promoteStagingSnapshot(pg)
+    await sendDiscordMessage(
+      `Search alerts failed for ${failed} user(s) ${Math.round(stagingAgeMs / DAY_MS)} days running; ` +
+        `promoting the profile snapshot anyway. Their pending alerts are lost.`,
+      'health',
+    )
+  } else {
+    // Keep the staging snapshot so the next run resumes and retries only the unstamped searches.
+    log.error(`Search alerts failed for ${failed} user(s); snapshot not promoted`)
+  }
+
+  return {status: 'success', notified, failed}
 }

backend/api/tests/unit/send-search-notifications.unit.test.ts

@@ -1,221 +1,200 @@
 jest.mock('shared/supabase/init')
-jest.mock('shared/supabase/sql-builder')
 jest.mock('api/get-profiles')
+jest.mock('api/profile-snapshot')
 jest.mock('email/functions/helpers')
+jest.mock('common/discord/core')
 
 import * as profileModules from 'api/get-profiles'
+import * as snapshotModules from 'api/profile-snapshot'
 import * as searchNotificationModules from 'api/send-search-notifications'
 import * as helperModules from 'email/functions/helpers'
 import * as supabaseInit from 'shared/supabase/init'
-import * as sqlBuilderModules from 'shared/supabase/sql-builder'
 
-describe('sendSearchNotification', () => {
+const CREATOR_ID = 'creator-1'
+
+const search = (id: number, creator_id = CREATOR_ID) => ({
+  id,
+  created_time: '2026-07-01T00:00:00Z',
+  creator_id,
+  search_filters: {genders: ['female']},
+  location: null,
+  last_notified_at: null,
+  last_checked_at: null,
+  search_name: null,
+})
+
+const profile = (user_id: string) => ({
+  user_id,
+  name: `name-${user_id}`,
+  user: {id: user_id, name: `name-${user_id}`, username: user_id},
+})
+
+describe('sendSearchNotifications', () => {
   let mockPg = {} as any
+  let takenAt: Date
+
   beforeEach(() => {
     jest.resetAllMocks()
+    takenAt = new Date()
+
     mockPg = {
-      map: jest.fn(),
+      manyOrNone: jest.fn(),
+      none: jest.fn().mockResolvedValue(undefined),
     }
     ;(supabaseInit.createSupabaseDirectClient as jest.Mock).mockReturnValue(mockPg)
+
+    // A healthy, freshly built snapshot with one changed profile in it.
+    ;(snapshotModules.hasStagingSnapshot as jest.Mock).mockResolvedValue(false)
+    ;(snapshotModules.buildStagingSnapshot as jest.Mock).mockResolvedValue(undefined)
+    ;(snapshotModules.isSnapshotUsable as jest.Mock).mockResolvedValue(true)
+    ;(snapshotModules.promoteStagingSnapshot as jest.Mock).mockResolvedValue(undefined)
+    ;(snapshotModules.getStagingTakenAt as jest.Mock).mockResolvedValue(takenAt)
+    ;(snapshotModules.getChangedUserIds as jest.Mock).mockResolvedValue(['woman-1'])
+    ;(snapshotModules.withSchema as jest.Mock).mockImplementation(
+      (_pg: any, _schema: string, fn: any) => fn(mockPg),
+    )
+    ;(helperModules.sendSearchAlertsEmail as jest.Mock).mockResolvedValue(null)
   })
+
   afterEach(() => {
     jest.restoreAllMocks()
   })
 
-  describe('when given valid input', () => {
-    it('should send search notification emails', async () => {
-      const mockSearchQuery = 'mockSqlQuery'
-      const mockSearches = [
-        {
-          created_time: 'mockSearchCreatedTime',
-          creator_id: 'mockCreatorId',
-          id: 123,
-          last_notified_at: null,
-          location: {mockLocation: 'mockLocationValue'},
-          search_filters: null,
-          search_name: null,
-        },
-        {
-          created_time: 'mockCreatedTime1',
-          creator_id: 'mockCreatorId1',
-          id: 1234,
-          last_notified_at: null,
-          location: {mockLocation1: 'mockLocationValue1'},
-          search_filters: null,
-          search_name: null,
-        },
-      ]
-      const _mockUsers = [
-        {
-          created_time: 'mockUserCreatedTime',
-          data: {mockData: 'mockDataValue'},
-          id: 'mockId',
-          name: 'mockName',
-          name_username_vector: 'mockNameUsernameVector',
-          username: 'mockUsername',
-        },
-        {
-          created_time: 'mockUserCreatedTime1',
-          data: {mockData1: 'mockDataValue1'},
-          id: 'mockId1',
-          name: 'mockName1',
-          name_username_vector: 'mockNameUsernameVector1',
-          username: 'mockUsername1',
-        },
-      ]
-      const _mockPrivateUsers = [
-        {
-          data: {mockData: 'mockDataValue'},
-          id: 'mockId',
-        },
-        {
-          data: {mockData1: 'mockDataValue1'},
-          id: 'mockId1',
-        },
-      ]
-      const mockProfiles = [
-        {
-          name: 'mockProfileName',
-          username: 'mockProfileUsername',
-        },
-        {
-          name: 'mockProfileName1',
-          username: 'mockProfileUsername1',
-        },
-      ]
-      const mockProps = [
-        {
-          skipId: 'mockCreatorId',
-          userId: 'mockCreatorId',
-          lastModificationWithin: '24 hours',
-          shortBio: true,
-        },
-        {
-          skipId: 'mockCreatorId1',
-          userId: 'mockCreatorId1',
-          lastModificationWithin: '24 hours',
-          shortBio: true,
-        },
-      ]
-      ;(sqlBuilderModules.renderSql as jest.Mock)
-        .mockReturnValueOnce(mockSearchQuery)
-        .mockReturnValueOnce('usersRenderSql')
-        .mockReturnValueOnce('privateUsersRenderSql')
-      ;(sqlBuilderModules.select as jest.Mock).mockReturnValue('Select')
-      ;(sqlBuilderModules.from as jest.Mock).mockReturnValue('From')
-      ;(mockPg.map as jest.Mock)
-        .mockResolvedValueOnce(mockSearches)
-        .mockResolvedValueOnce(_mockUsers)
-        .mockResolvedValueOnce(_mockPrivateUsers)
-      ;(profileModules.loadProfiles as jest.Mock)
-        .mockResolvedValueOnce({profiles: mockProfiles})
-        .mockResolvedValueOnce({profiles: mockProfiles})
-      jest.spyOn(searchNotificationModules, 'notifyBookmarkedSearch')
-      ;(helperModules.sendSearchAlertsEmail as jest.Mock).mockResolvedValue(null)
+  /** searches, then users, then private_users */
+  const mockQueries = (searches: any[]) => {
+    mockPg.manyOrNone
+      .mockResolvedValueOnce(searches)
+      .mockResolvedValueOnce([{id: CREATOR_ID, name: 'Creator'}])
+      .mockResolvedValueOnce([{id: CREATOR_ID, data: {email: 'creator@example.com'}}])
+  }
 
-      const result = await searchNotificationModules.sendSearchNotifications()
+  it('notifies about a profile that did not match before it was modified', async () => {
+    mockQueries([search(1)])
+    ;(profileModules.loadProfiles as jest.Mock)
+      .mockResolvedValueOnce({profiles: [profile('woman-1')]}) // matches now
+      .mockResolvedValueOnce({profiles: []}) // matched nothing in the snapshot
 
-      expect(result.status).toBe('success')
-      expect(sqlBuilderModules.renderSql).toBeCalledTimes(3)
-      expect(sqlBuilderModules.renderSql).toHaveBeenNthCalledWith(1, 'Select', 'From')
-      expect(sqlBuilderModules.renderSql).toHaveBeenNthCalledWith(2, 'Select', 'From')
-      expect(sqlBuilderModules.renderSql).toHaveBeenNthCalledWith(3, 'Select', 'From')
-      expect(mockPg.map).toBeCalledTimes(3)
-      expect(mockPg.map).toHaveBeenNthCalledWith(1, mockSearchQuery, [], expect.any(Function))
-      expect(mockPg.map).toHaveBeenNthCalledWith(2, 'usersRenderSql', [], expect.any(Function))
-      expect(mockPg.map).toHaveBeenNthCalledWith(
-        3,
-        'privateUsersRenderSql',
-        [],
-        expect.any(Function),
-      )
-      expect(profileModules.loadProfiles).toBeCalledTimes(2)
-      expect(profileModules.loadProfiles).toHaveBeenNthCalledWith(1, mockProps[0])
-      expect(profileModules.loadProfiles).toHaveBeenNthCalledWith(2, mockProps[1])
-      expect(searchNotificationModules.notifyBookmarkedSearch).toBeCalledTimes(1)
-      expect(searchNotificationModules.notifyBookmarkedSearch).toBeCalledWith({})
+    const result = await searchNotificationModules.sendSearchNotifications()
+
+    expect(result).toEqual({status: 'success', notified: 1, failed: 0})
+    expect(helperModules.sendSearchAlertsEmail).toBeCalledTimes(1)
+    expect(helperModules.sendSearchAlertsEmail).toBeCalledWith(
+      {id: CREATOR_ID, name: 'Creator'},
+      {email: 'creator@example.com'},
+      [
+        {
+          id: CREATOR_ID,
+          description: {filters: {genders: ['female']}, location: null},
+          matches: [profile('woman-1').user],
+        },
+      ],
+    )
+    expect(snapshotModules.promoteStagingSnapshot).toBeCalledTimes(1)
+  })
+
+  it('does not notify when the modified profile already matched the search', async () => {
+    mockQueries([search(1)])
+    ;(profileModules.loadProfiles as jest.Mock)
+      .mockResolvedValueOnce({profiles: [profile('woman-1')]}) // still matches
+      .mockResolvedValueOnce({profiles: [profile('woman-1')]}) // and already did yesterday
+
+    const result = await searchNotificationModules.sendSearchNotifications()
+
+    expect(result).toEqual({status: 'success', notified: 0, failed: 0})
+    expect(helperModules.sendSearchAlertsEmail).not.toBeCalled()
+    // The search is still watermarked, so a resumed run skips it.
+    expect(mockPg.none).toBeCalledWith(expect.stringContaining('set last_checked_at = now()'), {
+      searchIds: [1],
+      matchedSearchIds: [],
     })
+    expect(snapshotModules.promoteStagingSnapshot).toBeCalledTimes(1)
+  })
 
-    it('should send search notification emails when there is a matching creator_id entry in private users', async () => {
-      const mockSearchQuery = 'mockSqlQuery'
-      const mockSearches = [
-        {
-          created_time: 'mockSearchCreatedTime',
-          creator_id: 'mockCreatorId',
-          id: 123,
-          last_notified_at: null,
-          location: {mockLocation: 'mockLocationValue'},
-          search_filters: null,
-          search_name: null,
-        },
-        {
-          created_time: 'mockCreatedTime1',
-          creator_id: 'mockCreatorId1',
-          id: 1234,
-          last_notified_at: null,
-          location: {mockLocation1: 'mockLocationValue1'},
-          search_filters: null,
-          search_name: null,
-        },
-      ]
-      const _mockUsers = [
-        {
-          created_time: 'mockUserCreatedTime',
-          data: {mockCreatorId: 'mockDataValue'},
-          id: 'mockId',
-          name: 'mockName',
-          name_username_vector: 'mockNameUsernameVector',
-          username: 'mockUsername',
-        },
-        {
-          created_time: 'mockUserCreatedTime1',
-          data: {mockData1: 'mockDataValue1'},
-          id: 'mockId1',
-          name: 'mockName1',
-          name_username_vector: 'mockNameUsernameVector1',
-          username: 'mockUsername1',
-        },
-      ]
-      const _mockPrivateUsers = [
-        {
-          data: {mockData: 'mockDataValue'},
-          id: 'mockCreatorId',
-        },
-        {
-          data: {mockData1: 'mockDataValue1'},
-          id: 'mockId1',
-        },
-      ]
-      const mockProfiles = [
-        {
-          name: 'mockProfileName',
-          username: 'mockProfileUsername',
-        },
-        {
-          name: 'mockProfileName1',
-          username: 'mockProfileUsername1',
-        },
-      ]
-      ;(sqlBuilderModules.renderSql as jest.Mock)
-        .mockReturnValueOnce(mockSearchQuery)
-        .mockReturnValueOnce('usersRenderSql')
-        .mockReturnValueOnce('privateUsersRenderSql')
-      ;(sqlBuilderModules.select as jest.Mock).mockReturnValue('Select')
-      ;(sqlBuilderModules.from as jest.Mock).mockReturnValue('From')
-      ;(mockPg.map as jest.Mock)
-        .mockResolvedValueOnce(mockSearches)
-        .mockResolvedValueOnce(_mockUsers)
-        .mockResolvedValueOnce(_mockPrivateUsers)
-      ;(profileModules.loadProfiles as jest.Mock)
-        .mockResolvedValueOnce({profiles: mockProfiles})
-        .mockResolvedValueOnce({profiles: mockProfiles})
-      jest.spyOn(searchNotificationModules, 'notifyBookmarkedSearch')
-      ;(helperModules.sendSearchAlertsEmail as jest.Mock).mockResolvedValue(null)
+  it('only queries the snapshot for the profiles that currently match', async () => {
+    mockQueries([search(1)])
+    ;(profileModules.loadProfiles as jest.Mock)
+      .mockResolvedValueOnce({profiles: [profile('woman-1')]})
+      .mockResolvedValueOnce({profiles: []})
 
-      await searchNotificationModules.sendSearchNotifications()
+    await searchNotificationModules.sendSearchNotifications()
 
-      expect(searchNotificationModules.notifyBookmarkedSearch).toBeCalledTimes(1)
-      expect(searchNotificationModules.notifyBookmarkedSearch).not.toBeCalledWith({})
-    })
+    const [liveProps] = (profileModules.loadProfiles as jest.Mock).mock.calls[0]
+    const [snapshotProps] = (profileModules.loadProfiles as jest.Mock).mock.calls[1]
+    expect(liveProps).toMatchObject({genders: ['female'], skipId: CREATOR_ID, userIds: ['woman-1']})
+    expect(snapshotProps).toMatchObject({userIds: ['woman-1']})
+    expect(snapshotModules.withSchema).toHaveBeenNthCalledWith(
+      1,
+      mockPg,
+      snapshotModules.STAGING_SCHEMA,
+      expect.any(Function),
+    )
+    expect(snapshotModules.withSchema).toHaveBeenNthCalledWith(
+      2,
+      mockPg,
+      snapshotModules.SNAPSHOT_SCHEMA,
+      expect.any(Function),
+    )
+  })
+
+  it('skips a search whose profiles are all unchanged', async () => {
+    ;(snapshotModules.getChangedUserIds as jest.Mock).mockResolvedValue([])
+
+    const result = await searchNotificationModules.sendSearchNotifications()
+
+    expect(result).toEqual({status: 'success', notified: 0})
+    expect(profileModules.loadProfiles).not.toBeCalled()
+    expect(snapshotModules.promoteStagingSnapshot).toBeCalledTimes(1)
+  })
+
+  it('seeds the snapshot without alerting when it is missing or stale', async () => {
+    ;(snapshotModules.isSnapshotUsable as jest.Mock).mockResolvedValue(false)
+
+    const result = await searchNotificationModules.sendSearchNotifications()
+
+    expect(result).toEqual({status: 'success', notified: 0, skipped: true})
+    expect(snapshotModules.getChangedUserIds).not.toBeCalled()
+    expect(helperModules.sendSearchAlertsEmail).not.toBeCalled()
+    expect(snapshotModules.promoteStagingSnapshot).toBeCalledTimes(1)
+  })
+
+  it('reuses the staging snapshot of a crashed run instead of rebuilding it', async () => {
+    ;(snapshotModules.hasStagingSnapshot as jest.Mock).mockResolvedValue(true)
+    mockQueries([search(1)])
+    ;(profileModules.loadProfiles as jest.Mock)
+      .mockResolvedValueOnce({profiles: []})
+      .mockResolvedValueOnce({profiles: []})
+
+    await searchNotificationModules.sendSearchNotifications()
+
+    expect(snapshotModules.buildStagingSnapshot).not.toBeCalled()
+  })
+
+  it('keeps the staging snapshot when an email fails, so the next run retries it', async () => {
+    mockQueries([search(1)])
+    ;(profileModules.loadProfiles as jest.Mock)
+      .mockResolvedValueOnce({profiles: [profile('woman-1')]})
+      .mockResolvedValueOnce({profiles: []})
+    ;(helperModules.sendSearchAlertsEmail as jest.Mock).mockRejectedValue(new Error('smtp down'))
+
+    const result = await searchNotificationModules.sendSearchNotifications()
+
+    expect(result).toEqual({status: 'success', notified: 0, failed: 1})
+    expect(mockPg.none).not.toBeCalled() // no watermark, so the search is retried
+    expect(snapshotModules.promoteStagingSnapshot).not.toBeCalled()
+  })
+
+  it('promotes anyway once a failing staging snapshot gets too old', async () => {
+    ;(snapshotModules.getStagingTakenAt as jest.Mock).mockResolvedValue(
+      new Date(Date.now() - 5 * 24 * 60 * 60 * 1000),
+    )
+    mockQueries([search(1)])
+    ;(profileModules.loadProfiles as jest.Mock)
+      .mockResolvedValueOnce({profiles: [profile('woman-1')]})
+      .mockResolvedValueOnce({profiles: []})
+    ;(helperModules.sendSearchAlertsEmail as jest.Mock).mockRejectedValue(new Error('smtp down'))
+
+    await searchNotificationModules.sendSearchNotifications()
+
+    expect(snapshotModules.promoteStagingSnapshot).toBeCalledTimes(1)
   })
 })

backend/supabase/migration.sql

@@ -57,4 +57,5 @@ BEGIN;
 \i backend/supabase/migrations/20260330_add_substance_fields_to_profiles.sql
 \i backend/supabase/email_unsubscribe_tokens.sql
 \i backend/supabase/migrations/20260524_add_orientation_to_profiles.sql
+\i backend/supabase/migrations/20260709_add_last_checked_at_to_bookmarked_searches.sql
 COMMIT;

backend/supabase/migrations/20260709_add_last_checked_at_to_bookmarked_searches.sql

@@ -0,0 +1,17 @@
+-- Add the alert-run watermark to bookmarked searches
+-- Created: 2026-07-09
+
+-- Search alerts diff the current profiles against a snapshot of the profiles as they were at the end
+-- of the last successful notification run, so a profile only alerts on the run where it *starts*
+-- matching a search rather than on every later edit to it.
+--
+-- The snapshot schemas (profile_snapshot, profile_snapshot_staging) are created and swapped by the
+-- notification job itself (backend/api/src/profile-snapshot.ts), which rebuilds them from `public` on
+-- every run. They are deliberately not declared here: a later migration on `profiles` then needs no
+-- matching change, and the job detects the drift and reseeds itself.
+
+ALTER TABLE bookmarked_searches
+    ADD COLUMN IF NOT EXISTS last_checked_at TIMESTAMPTZ DEFAULT NULL;
+
+COMMENT ON COLUMN bookmarked_searches.last_checked_at IS
+    'Last time the alert job evaluated this search. Watermark that lets a crashed run resume without re-sending emails.';

web/lib/supabase/searches.ts

@@ -19,7 +19,7 @@ export const getUserBookmarkedSearches = async (userId: string) => {
 
 export type BookmarkedSearchSubmitType = Omit<
   Row<'bookmarked_searches'>,
-  'created_time' | 'id' | 'last_notified_at' | 'creator_id'
+  'created_time' | 'id' | 'last_notified_at' | 'last_checked_at' | 'creator_id'
 >
 
 export const submitBookmarkedSearch = async (
@@ -37,7 +37,8 @@ export const submitBookmarkedSearch = async (
   const props = {
     ...filterKeys(
       row,
-      (key, _) => !['id', 'created_time', 'last_notified_at', 'creator_id'].includes(key),
+      (key, _) =>
+        !['id', 'created_time', 'last_notified_at', 'last_checked_at', 'creator_id'].includes(key),
     ),
   } as BookmarkedSearchSubmitType