From 75cbcee919f81ce24e7ea71648249b647910517d Mon Sep 17 00:00:00 2001 From: MartinBraquet Date: Thu, 9 Jul 2026 23:16:38 +0200 Subject: [PATCH] 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 | 2 +- backend/api/src/get-profiles.ts | 28 +- backend/api/src/profile-snapshot.ts | 200 ++++++++++ backend/api/src/send-search-notifications.ts | 249 +++++++++--- .../send-search-notifications.unit.test.ts | 365 +++++++++--------- backend/supabase/migration.sql | 1 + ...last_checked_at_to_bookmarked_searches.sql | 17 + web/lib/supabase/searches.ts | 5 +- 8 files changed, 608 insertions(+), 259 deletions(-) create mode 100644 backend/api/src/profile-snapshot.ts create mode 100644 backend/supabase/migrations/20260709_add_last_checked_at_to_bookmarked_searches.sql diff --git a/backend/api/package.json b/backend/api/package.json index 01beb5c4..ceb98a08 100644 --- a/backend/api/package.json +++ b/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", diff --git a/backend/api/src/get-profiles.ts b/backend/api/src/get-profiles.ts index e219197a..1c06bc70 100644 --- a/backend/api/src/get-profiles.ts +++ b/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(countQuery, [], (r) => Number(r.count)) + const count = skipCount + ? profiles.length + : await pg.one(countQuery, [], (r) => Number(r.count)) return {profiles, count} } diff --git a/backend/api/src/profile-snapshot.ts b/backend/api/src/profile-snapshot.ts new file mode 100644 index 00000000..934d9591 --- /dev/null +++ b/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 = { + 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 = ( + pg: SupabaseDirectClient, + schema: string, + fn: (db: SupabaseDirectClient) => Promise, +) => + 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( + `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(`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) +} diff --git a/backend/api/src/send-search-notifications.ts b/backend/api/src/send-search-notifications.ts index d4a3616d..7cdc8060 100644 --- a/backend/api/src/send-search-notifications.ts +++ b/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 + 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, +) => { + 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( + `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>(`select * from users where id = any($(creatorIds))`, { + creatorIds, + }), + 'id', + ) + const privateUsers = keyBy( + await pg.manyOrNone>( + `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 = {} - 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 - 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} } diff --git a/backend/api/tests/unit/send-search-notifications.unit.test.ts b/backend/api/tests/unit/send-search-notifications.unit.test.ts index ee8f604f..bc0f963c 100644 --- a/backend/api/tests/unit/send-search-notifications.unit.test.ts +++ b/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) }) }) diff --git a/backend/supabase/migration.sql b/backend/supabase/migration.sql index fc3d4d56..9d5716bb 100644 --- a/backend/supabase/migration.sql +++ b/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; diff --git a/backend/supabase/migrations/20260709_add_last_checked_at_to_bookmarked_searches.sql b/backend/supabase/migrations/20260709_add_last_checked_at_to_bookmarked_searches.sql new file mode 100644 index 00000000..9617d3a3 --- /dev/null +++ b/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.'; diff --git a/web/lib/supabase/searches.ts b/web/lib/supabase/searches.ts index 5a186999..5e33383f 100644 --- a/web/lib/supabase/searches.ts +++ b/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