|
| 1 | +import {debug} from 'common/logger' |
| 2 | +import {isEqual} from 'lodash' |
| 3 | +import {SupabaseDirectClient} from 'shared/supabase/init' |
| 4 | + |
| 5 | +/** |
| 6 | + * A copy of the profile-side tables as they were at the end of the last successful search-alert run. |
| 7 | + * |
| 8 | + * Bookmarked-search alerts are a set diff: a profile is worth an email on the run where it *starts* |
| 9 | + * matching a search, not on every later edit. Running the very same filter query against this |
| 10 | + * snapshot answers "did it already match yesterday?" without duplicating any of the filter logic in |
| 11 | + * `get-profiles.ts` — the tables here carry the same names as their `public` counterparts, so the |
| 12 | + * generated SQL resolves to them through the `search_path` (see {@link withSchema}). |
| 13 | + */ |
| 14 | +export const SNAPSHOT_SCHEMA = 'profile_snapshot' |
| 15 | + |
| 16 | +/** |
| 17 | + * Where the next snapshot is built, at the *start* of a run. Both sides of the diff then read from a |
| 18 | + * snapshot rather than from `public`, so profiles edited while the run is in flight belong to |
| 19 | + * neither side and are picked up by the next run instead of being silently absorbed. |
| 20 | + * |
| 21 | + * It is promoted to {@link SNAPSHOT_SCHEMA} only once every search has been processed. A run that |
| 22 | + * dies halfway therefore leaves it behind, and the next run resumes against it. |
| 23 | + */ |
| 24 | +export const STAGING_SCHEMA = 'profile_snapshot_staging' |
| 25 | + |
| 26 | +/** |
| 27 | + * The tables `loadProfiles` filters on that a profile owner can actually change. |
| 28 | + * |
| 29 | + * Everything else it touches keeps resolving to `public` through the search_path, which is what we |
| 30 | + * want: the option tables (`interests`, `causes`, `work`, `*_translations`) are static, and |
| 31 | + * `hidden_profiles` / `compatibility_scores` describe the searcher rather than the profile being |
| 32 | + * matched. |
| 33 | + * |
| 34 | + * `user_activity` is deliberately absent. It churns on every page load, so snapshotting it would |
| 35 | + * cost more storage than everything else combined. Leaving it in `public` also means a `last_active` |
| 36 | + * filter sees identical rows on both sides of the diff, so merely coming back online can never look |
| 37 | + * like a new match. |
| 38 | + */ |
| 39 | +const SNAPSHOT_TABLES = ['profiles', 'users', 'profile_interests', 'profile_causes', 'profile_work'] |
| 40 | + |
| 41 | +/** Indexes to recreate on the copies — CTAS keeps no index from the source table. */ |
| 42 | +const SNAPSHOT_INDEXES: Record<string, string[]> = { |
| 43 | + profiles: ['id', 'user_id'], |
| 44 | + users: ['id'], |
| 45 | + profile_interests: ['profile_id'], |
| 46 | + profile_causes: ['profile_id'], |
| 47 | + profile_work: ['profile_id'], |
| 48 | +} |
| 49 | + |
| 50 | +const MANY_TO_MANY_LABELS = ['interests', 'causes', 'work'] |
| 51 | + |
| 52 | +/** Runs `fn` against a client whose unqualified table names resolve to `schema` first. */ |
| 53 | +export const withSchema = <T>( |
| 54 | + pg: SupabaseDirectClient, |
| 55 | + schema: string, |
| 56 | + fn: (db: SupabaseDirectClient) => Promise<T>, |
| 57 | +) => |
| 58 | + pg.tx(async (t) => { |
| 59 | + // `set local` is scoped to the transaction, so pooled connections keep their own search_path. |
| 60 | + await t.none('set local search_path to $(schema:name), public', {schema}) |
| 61 | + const result = await fn(t) |
| 62 | + // Releasing a savepoint hands its `set local` up to the enclosing transaction, so when `pg` is |
| 63 | + // itself a transaction the search_path would outlive this call and quietly point later queries |
| 64 | + // at the snapshot. Restoring is a no-op at the top level, where the commit reverts it anyway. |
| 65 | + await t.none('set local search_path to default') |
| 66 | + return result |
| 67 | + }) |
| 68 | + |
| 69 | +export const hasStagingSnapshot = async (pg: SupabaseDirectClient) => |
| 70 | + await pg.one<boolean>( |
| 71 | + `select to_regclass($(table)) is not null as exists`, |
| 72 | + {table: `${STAGING_SCHEMA}.meta`}, |
| 73 | + (r) => r.exists, |
| 74 | + ) |
| 75 | + |
| 76 | +export const getStagingTakenAt = async (pg: SupabaseDirectClient) => |
| 77 | + await pg.one<Date>(`select taken_at from ${STAGING_SCHEMA}.meta`, [], (r) => r.taken_at) |
| 78 | + |
| 79 | +/** |
| 80 | + * Rebuilds the staging snapshot from `public`. Recreating the tables from scratch on every run |
| 81 | + * (rather than truncating fixed ones) is what keeps this maintenance-free: a migration that adds a |
| 82 | + * column to `profiles` needs no matching change here. |
| 83 | + */ |
| 84 | +export const buildStagingSnapshot = async (pg: SupabaseDirectClient) => { |
| 85 | + await pg.tx(async (t) => { |
| 86 | + await t.none('drop schema if exists $(schema:name) cascade', {schema: STAGING_SCHEMA}) |
| 87 | + await t.none('create schema $(schema:name)', {schema: STAGING_SCHEMA}) |
| 88 | + |
| 89 | + for (const table of SNAPSHOT_TABLES) { |
| 90 | + await t.none('create table $(schema:name).$(table:name) as table public.$(table:name)', { |
| 91 | + schema: STAGING_SCHEMA, |
| 92 | + table, |
| 93 | + }) |
| 94 | + for (const column of SNAPSHOT_INDEXES[table]) { |
| 95 | + await t.none('create index on $(schema:name).$(table:name) ($(column:name))', { |
| 96 | + schema: STAGING_SCHEMA, |
| 97 | + table, |
| 98 | + column, |
| 99 | + }) |
| 100 | + } |
| 101 | + await t.none('analyze $(schema:name).$(table:name)', {schema: STAGING_SCHEMA, table}) |
| 102 | + } |
| 103 | + |
| 104 | + await t.none('create table $(schema:name).meta as select now() as taken_at', { |
| 105 | + schema: STAGING_SCHEMA, |
| 106 | + }) |
| 107 | + }) |
| 108 | + debug(`built ${STAGING_SCHEMA}`) |
| 109 | +} |
| 110 | + |
| 111 | +/** |
| 112 | + * True when the committed snapshot exists and has the same columns as `public`. |
| 113 | + * |
| 114 | + * A migration that changes a snapshotted table leaves the old snapshot unable to answer the filter |
| 115 | + * query (a new filterable column simply would not exist there), so the caller rebuilds instead and |
| 116 | + * skips one run of alerts. Failing closed costs at most a day of alerts; failing open would send |
| 117 | + * every user an email about every profile that already matched. |
| 118 | + */ |
| 119 | +export const isSnapshotUsable = async (pg: SupabaseDirectClient) => { |
| 120 | + const rows = await pg.manyOrNone<{live: string[] | null; snapshot: string[] | null}>( |
| 121 | + `select |
| 122 | + (select array_agg(c.column_name::text order by c.column_name) |
| 123 | + from information_schema.columns c |
| 124 | + where c.table_schema = 'public' and c.table_name = t.table_name) as live, |
| 125 | + (select array_agg(c.column_name::text order by c.column_name) |
| 126 | + from information_schema.columns c |
| 127 | + where c.table_schema = $(schema) and c.table_name = t.table_name) as snapshot |
| 128 | + from unnest($(tables)::text[]) as t(table_name)`, |
| 129 | + {schema: SNAPSHOT_SCHEMA, tables: SNAPSHOT_TABLES}, |
| 130 | + ) |
| 131 | + return rows.every((r) => r.snapshot != null && isEqual(r.live, r.snapshot)) |
| 132 | +} |
| 133 | + |
| 134 | +/** Swaps the staging snapshot in as the committed one. Only safe once every search is processed. */ |
| 135 | +export const promoteStagingSnapshot = async (pg: SupabaseDirectClient) => { |
| 136 | + await pg.tx(async (t) => { |
| 137 | + await t.none('drop schema if exists $(schema:name) cascade', {schema: SNAPSHOT_SCHEMA}) |
| 138 | + await t.none('alter schema $(staging:name) rename to $(snapshot:name)', { |
| 139 | + staging: STAGING_SCHEMA, |
| 140 | + snapshot: SNAPSHOT_SCHEMA, |
| 141 | + }) |
| 142 | + }) |
| 143 | + debug(`promoted ${STAGING_SCHEMA} to ${SNAPSHOT_SCHEMA}`) |
| 144 | +} |
| 145 | + |
| 146 | +const manyToManyCte = (label: string) => ` |
| 147 | + changed_${label} as ( |
| 148 | + select coalesce(staging.profile_id, snapshot.profile_id) as profile_id |
| 149 | + from ( |
| 150 | + select profile_id, md5(array_agg(option_id order by option_id)::text) as hash |
| 151 | + from ${STAGING_SCHEMA}.profile_${label} |
| 152 | + group by profile_id |
| 153 | + ) staging |
| 154 | + full outer join ( |
| 155 | + select profile_id, md5(array_agg(option_id order by option_id)::text) as hash |
| 156 | + from ${SNAPSHOT_SCHEMA}.profile_${label} |
| 157 | + group by profile_id |
| 158 | + ) snapshot on snapshot.profile_id = staging.profile_id |
| 159 | + where staging.hash is distinct from snapshot.hash |
| 160 | + )` |
| 161 | + |
| 162 | +/** |
| 163 | + * The users whose profile differs between the staging snapshot and the committed one. |
| 164 | + * |
| 165 | + * Hashing the whole row rather than trusting `profiles.last_modification_time` matters: that |
| 166 | + * column's trigger only fires on `profiles` updates, so it never sees a change to a profile's |
| 167 | + * interests, causes or work. |
| 168 | + */ |
| 169 | +export const getChangedUserIds = async (pg: SupabaseDirectClient) => { |
| 170 | + const query = ` |
| 171 | + with staging_profiles as ( |
| 172 | + select p.id, p.user_id, md5(p::text) as hash from ${STAGING_SCHEMA}.profiles p |
| 173 | + ), |
| 174 | + changed_profiles as ( |
| 175 | + select coalesce(staging.id, snapshot.id) as id |
| 176 | + from staging_profiles staging |
| 177 | + full outer join ( |
| 178 | + select p.id, md5(p::text) as hash from ${SNAPSHOT_SCHEMA}.profiles p |
| 179 | + ) snapshot on snapshot.id = staging.id |
| 180 | + where staging.hash is distinct from snapshot.hash |
| 181 | + ), |
| 182 | + changed_users as ( |
| 183 | + select coalesce(staging.id, snapshot.id) as id |
| 184 | + from (select u.id, md5(u::text) as hash from ${STAGING_SCHEMA}.users u) staging |
| 185 | + full outer join ( |
| 186 | + select u.id, md5(u::text) as hash from ${SNAPSHOT_SCHEMA}.users u |
| 187 | + ) snapshot on snapshot.id = staging.id |
| 188 | + where staging.hash is distinct from snapshot.hash |
| 189 | + ), |
| 190 | + ${MANY_TO_MANY_LABELS.map(manyToManyCte).join(',')} |
| 191 | + select distinct staging_profiles.user_id |
| 192 | + from staging_profiles |
| 193 | + where staging_profiles.id in (select id from changed_profiles) |
| 194 | + or staging_profiles.user_id in (select id from changed_users) |
| 195 | + ${MANY_TO_MANY_LABELS.map( |
| 196 | + (label) => `or staging_profiles.id in (select profile_id from changed_${label})`, |
| 197 | + ).join('\n ')}` |
| 198 | + |
| 199 | + return await pg.map(query, [], (r) => r.user_id as string) |
| 200 | +} |
0 commit comments