From a4ffb1196de4a22a2da0e0ad424407eea29ea8c6 Mon Sep 17 00:00:00 2001 From: Tom Scanlan Date: Sat, 27 Jun 2026 15:55:15 -0400 Subject: [PATCH] =?UTF-8?q?feat(geocode):=20in-Worker=20address=E2=86=92?= =?UTF-8?q?=5Fgeo=20drip=20as=20a=20cron=20module?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move geocoding into the Worker as a cron-driven drip: each run geocodes a bounded batch of address-only events and writes _geo back to Meili, giving steady-state near-me coverage without the external script. Adds config/db/process modules with tests and folds in two rounds of code-review fixes (om-16z0, om-xip5). --- apps/web/scripts/geocode-events.ts | 250 +++-------- apps/web/src/app.d.ts | 14 + apps/web/src/lib/geocode/config.ts | 20 + apps/web/src/lib/geocode/db.ts | 40 ++ apps/web/src/lib/geocode/process.test.ts | 160 +++++++ apps/web/src/lib/geocode/process.ts | 89 ++++ apps/web/src/lib/search/server/d1-native.ts | 16 + .../src/lib/search/server/geocode-job.test.ts | 420 ++++++++++++++++++ apps/web/src/lib/search/server/geocode-job.ts | 293 ++++++++++++ .../src/lib/search/server/geocoder.test.ts | 104 ++++- apps/web/src/lib/search/server/geocoder.ts | 63 ++- apps/web/src/lib/search/server/normalize.ts | 6 +- apps/web/src/routes/api/cron/+server.ts | 11 + 13 files changed, 1270 insertions(+), 216 deletions(-) create mode 100644 apps/web/src/lib/geocode/config.ts create mode 100644 apps/web/src/lib/geocode/db.ts create mode 100644 apps/web/src/lib/geocode/process.test.ts create mode 100644 apps/web/src/lib/geocode/process.ts create mode 100644 apps/web/src/lib/search/server/d1-native.ts create mode 100644 apps/web/src/lib/search/server/geocode-job.test.ts create mode 100644 apps/web/src/lib/search/server/geocode-job.ts diff --git a/apps/web/scripts/geocode-events.ts b/apps/web/scripts/geocode-events.ts index b3413f2..cb74fe8 100644 --- a/apps/web/scripts/geocode-events.ts +++ b/apps/web/scripts/geocode-events.ts @@ -1,31 +1,24 @@ // apps/web/scripts/geocode-events.ts -// External geocode job (runs OFF Cloudflare): finds address-only events in the -// openmeet-atmo D1 that lack coordinates, geocodes each unique address through -// the config-selected Nominatim/LocationIQ client, writes geocode_cache, and -// _geo-updates the affected Meili docs. ONE job: "find work" and "geocode" are -// sequential steps sharing the cache, not two passes. Rate limiting is the -// sleep between calls — the whole reason geocoding stays off Workers. +// CLI wrapper around the shared geocode core (lib/search/server/geocode-job.ts): +// runs it OFF Cloudflare against the openmeet-atmo D1 over the REST client, for +// the one-time bulk backfill and ad-hoc manual drips. The in-Worker cron drip +// (lib/geocode/process.ts) calls the same core over the native D1 binding. This +// wrapper owns the CLI surface: env/flag parsing, the public-Nominatim bulk +// guard, and client construction. // // Run (backfill MUST point the geocoder at LocationIQ, never public Nominatim): // CLOUDFLARE_API_TOKEN=… MEILI_URL=https://search.testnet.openmeet.net MEILI_KEY=… \ // GEOCODER_URL=https://us1.locationiq.com/v1/search GEOCODER_KEY=… \ // pnpm -C apps/web exec tsx scripts/geocode-events.ts --limit 50 -import { createD1Client, type D1Client } from '../src/lib/search/server/d1-http'; +import { createD1Client } from '../src/lib/search/server/d1-http'; import { createGeocoder, - addressToQuery, - requireGeocoderForBulk + requireGeocoderForBulk, + isPublicNominatimHost } from '../src/lib/search/server/geocoder'; -import { - isEligible, - groupEventsByNorm, - addressNeedingGeocode, - type GeocodeCacheRow, - type WorklistEvent -} from '../src/lib/search/server/geocode-cache'; -import { eventToSearchDoc } from '../src/lib/search/server/normalize'; -import { discoverableSql } from '../src/lib/search/server/discoverability'; -import { MeiliEventIndex, EVENT_COLLECTION } from '../src/lib/search/server/meili-sink'; +import { MeiliEventIndex } from '../src/lib/search/server/meili-sink'; +import { runGeocodeJob } from '../src/lib/search/server/geocode-job'; +import { DEFAULT_GEOCODE_SLEEP_MS } from '../src/lib/geocode/config'; const env = process.env; const argv = process.argv.slice(2); @@ -49,56 +42,33 @@ const parseLimit = (raw: string | undefined): number => { return n; }; +// Throttle between geocoder calls (ms). Unset → the safe default; set → must be a +// finite, non-negative number. Fail loud on a malformed value here (a human typed +// it): an unvalidated Number() would yield NaN → sleep(NaN) ≈ 0ms and silently +// hammer the provider on a bulk backfill. (The core also clamps defensively.) +const parseSleepMs = (raw: string | undefined): number => { + if (raw === undefined) return DEFAULT_GEOCODE_SLEEP_MS; + const n = Number(raw); + if (!Number.isFinite(n) || n < 0) { + throw new Error( + `GEOCODE_SLEEP_MS must be a non-negative number of milliseconds; got ${JSON.stringify(raw)}` + ); + } + return n; +}; + const retryNegative = flag('--retry-negative'); const dryRun = flag('--dry-run'); const allowPublicNominatim = flag('--allow-public-nominatim'); const limit = parseLimit(opt('--limit', '0')); // 0 = no cap -const sleepMs = Number(env.GEOCODE_SLEEP_MS ?? '1100'); - -const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); - -// Worklist: discoverable events carrying a .address location. We deliberately -// do NOT exclude coordinate locations in SQL — that filtered by $type presence, -// which diverges from the sink's actual _geo derivation (it ignored fsq, and -// excluded events whose only geo/hthree coords are out of range and so never -// get an in-index _geo). The precise "does this already resolve to coordinates?" -// decision is made in memory by addressNeedingGeocode (recordGeo), the same -// derivation the sink uses. json_each walks locations[]; the "$type" key is -// quoted because it starts with $. -const WORKLIST_SQL = ` -SELECT r.uri AS uri, r.did AS did, r.rkey AS rkey, r.record AS record -FROM records_event AS r -WHERE EXISTS ( - SELECT 1 FROM json_each(r.record, '$.locations') - WHERE json_extract(value, '$."$type"') = 'community.lexicon.location.address') - AND ${discoverableSql('r.record')} -`; - -// Re-read events from records_event by uri, keeping only those still present and -// discoverable. The parsed record drives a full-doc upsert, so it's read fresh -// (not from the job-start worklist) — an event deleted or unlisted while the slow -// geocode loop runs is skipped here instead of being resurrected as a stale doc. -async function fetchLiveDocs( - d1: D1Client, - uris: string[] -): Promise<{ uri: string; did: string; rkey: string; record: Record }[]> { - if (uris.length === 0) return []; - const placeholders = uris.map(() => '?').join(','); - const rows = await d1.query<{ uri: string; did: string; rkey: string; record: string }>( - `SELECT uri, did, rkey, record FROM records_event - WHERE uri IN (${placeholders}) AND ${discoverableSql('record')}`, - uris - ); - const out: { uri: string; did: string; rkey: string; record: Record }[] = []; - for (const r of rows) { - try { - out.push({ uri: r.uri, did: r.did, rkey: r.rkey, record: JSON.parse(r.record) }); - } catch { - // Unparseable record JSON — skip, same as the worklist parse. - } - } - return out; -} +const sleepMs = parseSleepMs(env.GEOCODE_SLEEP_MS); +// The effective geocoder is public OSM Nominatim when GEOCODER_URL is unset or +// points at the public host — EVEN if GEOCODER_KEY is set (createGeocoder ignores +// the key against Nominatim, the ?key= is dropped). Gate the bulk guard AND the +// cache provenance tag on the effective host, not key presence, so a +// key-without-URL run can't slip an uncapped backfill onto public Nominatim or +// mis-tag the cache rows as locationiq. +const onPublicNominatim = isPublicNominatimHost(env.GEOCODER_URL); async function main() { if (!env.CLOUDFLARE_API_TOKEN) throw new Error('CLOUDFLARE_API_TOKEN is required'); @@ -107,15 +77,21 @@ async function main() { if (!env.CLOUDFLARE_ACCOUNT_ID || !env.D1_DATABASE_ID) throw new Error('CLOUDFLARE_ACCOUNT_ID and D1_DATABASE_ID are required'); if (!env.MEILI_URL || !env.MEILI_KEY) throw new Error('MEILI_URL and MEILI_KEY are required'); - if (!env.GEOCODER_KEY) { + if (onPublicNominatim) { console.warn( - '[geocode] GEOCODER_KEY is unset → using PUBLIC Nominatim. OK for a small drip; ' + - 'a bulk backfill against public Nominatim risks a silent IP ban. Use LocationIQ for backfill.' + '[geocode] effective geocoder is PUBLIC Nominatim' + + (env.GEOCODER_KEY + ? ' (GEOCODER_KEY is set but GEOCODER_URL is unset/public, so the key is IGNORED)' + : '') + + '. OK for a small drip; a bulk backfill risks a silent IP ban. ' + + 'Set GEOCODER_URL to your LocationIQ endpoint for backfill.' ); } - // Hard-stop a bulk/uncapped keyless run before it can touch public Nominatim. + // Hard-stop a bulk/uncapped run before it can touch public Nominatim — keyed on + // the EFFECTIVE host so a GEOCODER_KEY with no URL (still public Nominatim) can't + // slip the gate the way `!!GEOCODER_KEY` did. requireGeocoderForBulk({ - hasKey: !!env.GEOCODER_KEY, + nonPublicEndpoint: !onPublicNominatim, dryRun, limit, allowPublic: allowPublicNominatim @@ -133,128 +109,18 @@ async function main() { indexUid: env.SEARCH_INDEX ?? 'events' }); - // Defensive: the table is created by geocode-cache.sql, but a fresh DB - // shouldn't make the job crash before it can self-heal. - await d1.query( - `CREATE TABLE IF NOT EXISTS geocode_cache ( - address_norm TEXT PRIMARY KEY, lat REAL, lng REAL, precision TEXT, - source TEXT NOT NULL, geocoded_at INTEGER NOT NULL, - fail_count INTEGER NOT NULL DEFAULT 0, last_error TEXT)` - ); - - // Load the whole cache once (small) and decide eligibility in memory. - const cacheRows = await d1.query(`SELECT * FROM geocode_cache`); - const cache = new Map(cacheRows.map((r) => [r.address_norm, r])); - - // Worklist → WorklistEvent[] (parse record JSON; keep only events that need - // geocoding — an address location AND no coordinates the index already - // derives, per addressNeedingGeocode). - const rawEvents = await d1.query<{ uri: string; did: string; rkey: string; record: string }>( - WORKLIST_SQL - ); - const events: WorklistEvent[] = []; - for (const e of rawEvents) { - let record: Record; - try { - record = JSON.parse(e.record); - } catch { - continue; - } - const loc = addressNeedingGeocode(record); - if (loc) events.push({ uri: e.uri, did: e.did, rkey: e.rkey, loc }); - } - - const byNorm = groupEventsByNorm(events); - const now = Date.now(); - const work = [...byNorm.entries()].filter(([norm]) => - isEligible(cache.get(norm), now, retryNegative) - ); - const capped = limit > 0 ? work.slice(0, limit) : work; - - console.log( - `[geocode] worklist events=${events.length} unique-addresses=${byNorm.size} ` + - `eligible=${work.length} processing=${capped.length}${dryRun ? ' (dry-run)' : ''}` - ); - - let resolved = 0; - let negative = 0; - let transient = 0; - let skippedGone = 0; - for (const [norm, group] of capped) { - const query = addressToQuery(group[0].loc); - if (dryRun) { - console.log(`[geocode] would geocode "${query}" -> ${group.length} event(s)`); - continue; - } - try { - const point = await geocoder.geocode(query); - if (point) { - await d1.query( - `INSERT INTO geocode_cache (address_norm, lat, lng, precision, source, geocoded_at, fail_count, last_error) - VALUES (?, ?, ?, ?, ?, ?, 0, NULL) - ON CONFLICT(address_norm) DO UPDATE SET - lat=excluded.lat, lng=excluded.lng, precision=excluded.precision, - source=excluded.source, geocoded_at=excluded.geocoded_at, fail_count=0, last_error=NULL`, - [ - norm, - point.lat, - point.lng, - point.precision ?? null, - env.GEOCODER_KEY ? 'locationiq' : 'nominatim', - now - ] - ); - // Re-read the events fresh and upsert the FULL doc with _geo attached, - // keeping only those still present AND discoverable. A full-doc upsert is - // idempotent and needs no "is it indexed?" snapshot: it merges if the - // event is already indexed, lands a complete doc if not (never a {id,_geo} - // stub), and is identical to what the sink writes — so a concurrent sink - // write converges instead of clobbering. - const live = await fetchLiveDocs( - d1, - group.map((e) => e.uri) - ); - skippedGone += group.length - live.length; - if (live.length) { - await meili.upsert( - live.map((r) => { - const doc = eventToSearchDoc({ - uri: r.uri, - did: r.did, - collection: EVENT_COLLECTION, - rkey: r.rkey, - record: r.record - }); - // Don't overwrite a coordinate _geo the record gained mid-run. - if (!doc._geo) doc._geo = { lat: point.lat, lng: point.lng }; - return doc; - }) - ); - } - resolved++; - } else { - // No-match: write/increment a negative row (backoff handled by isEligible). - await d1.query( - `INSERT INTO geocode_cache (address_norm, lat, lng, precision, source, geocoded_at, fail_count, last_error) - VALUES (?, NULL, NULL, NULL, ?, ?, 1, 'no match') - ON CONFLICT(address_norm) DO UPDATE SET - source=excluded.source, geocoded_at=excluded.geocoded_at, fail_count=geocode_cache.fail_count+1, last_error='no match'`, - [norm, env.GEOCODER_KEY ? 'locationiq' : 'nominatim', now] - ); - negative++; - } - } catch (err) { - // Transient (HTTP/network): DON'T write a negative row — retry next run. - transient++; - console.warn(`[geocode] transient error for "${query}": ${(err as Error).message}`); - } - await sleep(sleepMs); - } - - console.log( - `[geocode] done resolved=${resolved} negative=${negative} transient=${transient} ` + - `skipped-gone=${skippedGone}` - ); + await runGeocodeJob({ + d1, + geocoder, + meili, + source: onPublicNominatim ? 'nominatim' : 'locationiq', + limit, + sleepMs, + retryNegative, + dryRun, + log: (m) => console.log('[geocode]', m), + warn: (m) => console.warn('[geocode]', m) + }); } main().catch((e) => { diff --git a/apps/web/src/app.d.ts b/apps/web/src/app.d.ts index 827e42e..067b36b 100644 --- a/apps/web/src/app.d.ts +++ b/apps/web/src/app.d.ts @@ -101,6 +101,20 @@ declare global { /** Default Admin API Key for the write path (set via * `wrangler secret put`). Never the instance root key. */ SEARCH_SINK_API_KEY?: string; + /** Forward-geocoder endpoint for the address→_geo drip (the cron job + * that resolves coordinates for newly-ingested address-only events). + * Nominatim-compatible /search; LocationIQ = `https://us1.locationiq.com/v1/search`. + * Reuses the SEARCH_SINK_* Meili write creds + the DB binding, so this + * plus GEOCODER_KEY are the only drip-specific config. */ + GEOCODER_URL?: string; + /** Geocoder API key (LocationIQ), set via `wrangler secret put`. When + * unset, the drip no-ops (it won't fall back to public Nominatim). */ + GEOCODER_KEY?: string; + /** Optional User-Agent for geocoder requests. */ + GEOCODER_USER_AGENT?: string; + /** Min ms between geocoder calls in the drip — the rate limiter. Set to + * the ceiling the geocoder tier allows; defaults to DEFAULT_GEOCODE_SLEEP_MS. */ + GEOCODE_SLEEP_MS?: string; }; /** Cloudflare Worker execution context. Use `ctx.waitUntil(promise)` to * let the worker keep a fire-and-forget task alive after the response diff --git a/apps/web/src/lib/geocode/config.ts b/apps/web/src/lib/geocode/config.ts new file mode 100644 index 0000000..26ebb03 --- /dev/null +++ b/apps/web/src/lib/geocode/config.ts @@ -0,0 +1,20 @@ +// Address→_geo geocode drip — shared constants for the in-cron backfill of +// coordinates for newly-ingested address-only events. Mirrors lib/notify/config. + +/** Cadence gate: the drip body runs at most once per this interval. It rides the + * every-minute cron (routes/api/cron) via a D1 timestamp, so a new address-only + * event waits at most ~this long for its _geo. 30 min keeps geocoder usage and + * D1 reads low while bounding that latency. */ +export const GEOCODE_DRIP_INTERVAL_MS = 30 * 60 * 1000; + +/** Max unique addresses geocoded per drip run. Each costs ~one geocoder call + + * GEOCODE_SLEEP_MS, so this bounds the run's wall time far under the 15-min cron + * cap and bounds per-tier geocoder spend. Leftover eligible work is picked up on + * the next run; steady-state inflow is well under the cap. */ +export const MAX_GEOCODE_PER_TICK = 50; + +/** Default min delay between geocoder calls, in ms — the rate limiter. Operators + * set GEOCODE_SLEEP_MS to the ceiling their geocoder tier allows (e.g. LocationIQ + * paid can go faster; public Nominatim wants >=1000). Sized safe-by-default; a + * drip's low volume makes the exact value latency-irrelevant. */ +export const DEFAULT_GEOCODE_SLEEP_MS = 1100; diff --git a/apps/web/src/lib/geocode/db.ts b/apps/web/src/lib/geocode/db.ts new file mode 100644 index 0000000..b7e548a --- /dev/null +++ b/apps/web/src/lib/geocode/db.ts @@ -0,0 +1,40 @@ +// D1 state for the geocode drip's cadence gate. A tiny single-purpose key/value +// table so geocode_cache stays purely address rows. The job's own geocode_cache +// table is created/self-healed by the shared core (runGeocodeJob). + +/** Row key for the drip's last-run timestamp (epoch ms). */ +const LAST_DRIP_KEY = 'last_drip_at'; + +export async function ensureGeocodeDripSchema(db: D1Database): Promise { + await db + .prepare( + `CREATE TABLE IF NOT EXISTS geocode_drip_state ( + key TEXT PRIMARY KEY, + value INTEGER NOT NULL + )` + ) + .run(); +} + +/** Atomically claim the drip's cadence slot for `now`. A single conditional + * upsert: insert the first-ever timestamp, or advance it ONLY when a full + * `intervalMs` has elapsed since the stored one. Returns true iff THIS call won + * the slot (it wrote a row) — so two cron ticks racing inside the same interval + * can't both proceed, unlike the old read-then-write gate (a window where both + * read the stale timestamp before either wrote). Relies on D1 reporting + * meta.changes (1 = claimed, 0 = gated out by the WHERE on DO UPDATE). */ +export async function claimDripSlot( + db: D1Database, + now: number, + intervalMs: number +): Promise { + const res = await db + .prepare( + `INSERT INTO geocode_drip_state (key, value) VALUES (?, ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value + WHERE excluded.value - geocode_drip_state.value >= ?` + ) + .bind(LAST_DRIP_KEY, now, intervalMs) + .run(); + return (res.meta?.changes ?? 0) > 0; +} diff --git a/apps/web/src/lib/geocode/process.test.ts b/apps/web/src/lib/geocode/process.test.ts new file mode 100644 index 0000000..6f53daf --- /dev/null +++ b/apps/web/src/lib/geocode/process.test.ts @@ -0,0 +1,160 @@ +import { describe, it, expect } from 'vitest'; +import { runGeocodeDrip, geocodeDripConfigured, dripGeocoderPolicy } from './process'; +import { GEOCODE_DRIP_INTERVAL_MS, MAX_GEOCODE_PER_TICK, DEFAULT_GEOCODE_SLEEP_MS } from './config'; +import { + PUBLIC_NOMINATIM_DRIP_MAX, + PUBLIC_NOMINATIM_MIN_SLEEP_MS +} from '../search/server/geocoder'; + +type Env = Parameters[0]; + +/** The drip needs only the sink creds; the geocoder key is optional (keyless runs + * on public Nominatim at safe limits). Tests add GEOCODER_* / GEOCODE_SLEEP_MS. */ +const configuredEnv = (over: Partial = {}): Env => + ({ + SEARCH_SINK_URL: 'http://meili.test', + SEARCH_SINK_API_KEY: 'admin-key', + ...over + }) as Env; + +const locationiqEnv = (over: Partial = {}): Env => + configuredEnv({ + GEOCODER_KEY: 'loc-key', + GEOCODER_URL: 'https://us1.locationiq.com/v1/search', + ...over + }); + +/** Minimal stand-in for the native D1 binding. Implements the atomic cadence + * claim (INSERT ... ON CONFLICT DO UPDATE ... WHERE → meta.changes) against an + * in-memory value, and returns empty results for everything else so the core + * runs but does no network. Records every prepared SQL so tests count core runs. + * `failCore` makes the core's geocode_cache queries throw, to test that the + * cadence slot is claimed BEFORE the work (a failed run still advances it). */ +function fakeNativeDb(opts: { failCore?: boolean } = {}) { + const state = new Map(); + const sqls: string[] = []; + const db = { + prepare(sql: string) { + sqls.push(sql); + let bound: unknown[] = []; + const stmt = { + bind(...args: unknown[]) { + bound = args; + return stmt; + }, + async first() { + return null as T; + }, + async all() { + if (opts.failCore && sql.includes('geocode_cache')) { + throw new Error('core query failed'); + } + return { results: [] as T[] }; + }, + async run() { + if (sql.includes('INSERT INTO geocode_drip_state')) { + const [key, value, interval] = bound as [string, number, number]; + const prev = state.get(key); + const claimed = prev === undefined || value - prev >= interval; + if (claimed) state.set(key, value); + return { success: true, meta: { changes: claimed ? 1 : 0 } }; + } + return { success: true, meta: { changes: 0 } }; + } + }; + return stmt; + } + } as unknown as D1Database; + return { db, sqls, state }; +} + +const coreRuns = (sqls: string[]) => + sqls.filter((s) => s.includes('CREATE TABLE IF NOT EXISTS geocode_cache')).length; + +describe('geocodeDripConfigured', () => { + it('requires only the sink creds — the geocoder key is optional', () => { + expect(geocodeDripConfigured(configuredEnv())).toBe(true); + expect(geocodeDripConfigured(locationiqEnv())).toBe(true); + }); + + it('is false without the sink (nowhere to write _geo)', () => { + expect(geocodeDripConfigured(configuredEnv({ SEARCH_SINK_URL: undefined }))).toBe(false); + expect(geocodeDripConfigured(configuredEnv({ SEARCH_SINK_API_KEY: undefined }))).toBe(false); + }); +}); + +describe('dripGeocoderPolicy', () => { + it('keyless → public Nominatim safe defaults: small cap, ≥1 req/s, source nominatim', () => { + const p = dripGeocoderPolicy(configuredEnv()); + expect(p.source).toBe('nominatim'); + expect(p.limit).toBe(Math.min(MAX_GEOCODE_PER_TICK, PUBLIC_NOMINATIM_DRIP_MAX)); + expect(p.sleepMs).toBe(DEFAULT_GEOCODE_SLEEP_MS); + expect(p.sleepMs).toBeGreaterThanOrEqual(PUBLIC_NOMINATIM_MIN_SLEEP_MS); + }); + + it('floors a too-fast GEOCODE_SLEEP_MS on Nominatim to the policy minimum', () => { + const p = dripGeocoderPolicy(configuredEnv({ GEOCODE_SLEEP_MS: '100' })); + expect(p.sleepMs).toBe(PUBLIC_NOMINATIM_MIN_SLEEP_MS); + }); + + it('a key with no URL still hits Nominatim → safe defaults, key ignored, source nominatim', () => { + const p = dripGeocoderPolicy(configuredEnv({ GEOCODER_KEY: 'k' })); + expect(p.source).toBe('nominatim'); + expect(p.limit).toBe(Math.min(MAX_GEOCODE_PER_TICK, PUBLIC_NOMINATIM_DRIP_MAX)); + expect(p.sleepMs).toBeGreaterThanOrEqual(PUBLIC_NOMINATIM_MIN_SLEEP_MS); + }); + + it('keyed LocationIQ endpoint lifts the cap and honors the operator throttle', () => { + const p = dripGeocoderPolicy(locationiqEnv({ GEOCODE_SLEEP_MS: '300' })); + expect(p.source).toBe('locationiq'); + expect(p.limit).toBe(MAX_GEOCODE_PER_TICK); + expect(p.sleepMs).toBe(300); + }); + + it('falls back to the default throttle on a malformed GEOCODE_SLEEP_MS', () => { + expect(dripGeocoderPolicy(locationiqEnv({ GEOCODE_SLEEP_MS: 'abc' })).sleepMs).toBe( + DEFAULT_GEOCODE_SLEEP_MS + ); + // negative on the LocationIQ path also falls back (no floor there) + expect(dripGeocoderPolicy(locationiqEnv({ GEOCODE_SLEEP_MS: '-5' })).sleepMs).toBe( + DEFAULT_GEOCODE_SLEEP_MS + ); + }); +}); + +describe('runGeocodeDrip cadence gate', () => { + it('no-ops when unconfigured (touches no D1)', async () => { + const { db, sqls } = fakeNativeDb(); + await runGeocodeDrip(configuredEnv({ SEARCH_SINK_URL: undefined }), db, 1_000); + expect(sqls).toHaveLength(0); + }); + + it('runs on first call, skips within the interval, runs again after it', async () => { + const { db, sqls } = fakeNativeDb(); + const t0 = 1_700_000_000_000; + + await runGeocodeDrip(configuredEnv(), db, t0); + expect(coreRuns(sqls)).toBe(1); + + // 1 min later — inside the 30-min interval → gated out, core does not run. + await runGeocodeDrip(configuredEnv(), db, t0 + 60_000); + expect(coreRuns(sqls)).toBe(1); + + // Past the interval → runs again. + await runGeocodeDrip(configuredEnv(), db, t0 + GEOCODE_DRIP_INTERVAL_MS + 1); + expect(coreRuns(sqls)).toBe(2); + }); + + it('claim-before-work: a failed run still advances the slot, so the next tick gates out', async () => { + const { db, sqls } = fakeNativeDb({ failCore: true }); + const t0 = 1_700_000_000_000; + + // The core throws, but the slot was claimed before the work began. + await expect(runGeocodeDrip(configuredEnv(), db, t0)).rejects.toThrow(); + expect(coreRuns(sqls)).toBe(1); + + // Next tick within the interval is gated out — no retry-storm despite the failure. + await runGeocodeDrip(configuredEnv(), db, t0 + 60_000); + expect(coreRuns(sqls)).toBe(1); + }); +}); diff --git a/apps/web/src/lib/geocode/process.ts b/apps/web/src/lib/geocode/process.ts new file mode 100644 index 0000000..9500257 --- /dev/null +++ b/apps/web/src/lib/geocode/process.ts @@ -0,0 +1,89 @@ +import { + createGeocoder, + isPublicNominatimHost, + PUBLIC_NOMINATIM_DRIP_MAX, + PUBLIC_NOMINATIM_MIN_SLEEP_MS +} from '$lib/search/server/geocoder'; +import { MeiliEventIndex, meiliSinkBackendFromEnv } from '$lib/search/server/meili-sink'; +import { nativeD1Client } from '$lib/search/server/d1-native'; +import { runGeocodeJob } from '$lib/search/server/geocode-job'; +import { GEOCODE_DRIP_INTERVAL_MS, MAX_GEOCODE_PER_TICK, DEFAULT_GEOCODE_SLEEP_MS } from './config'; +import { ensureGeocodeDripSchema, claimDripSlot } from './db'; + +type Env = App.Platform['env']; + +/** Configured when the Meili write path (the sink creds, reused) is present — + * that's the only hard requirement, since the drip just needs somewhere to write + * the resolved _geo. The geocoder works keyless against public Nominatim, so a + * key is OPTIONAL: keyless still runs the drip, just at Nominatim-safe limits (a + * smaller per-tick cap and a >=1 req/s throttle floor — see dripGeocoderPolicy); + * set GEOCODER_KEY + a non-public GEOCODER_URL (LocationIQ) to lift them. No sink + * → no-op (dev, or before search is provisioned). */ +export function geocodeDripConfigured(env: Env): boolean { + return meiliSinkBackendFromEnv(env) !== null; +} + +/** Per-tick volume + throttle + cache-source tag for the drip. When the effective + * endpoint is public Nominatim (keyless, OR a key with an unset/public + * GEOCODER_URL — the ?key= is then ignored and we're really on Nominatim), both + * are clamped to Nominatim's usage policy: the per-tick cap drops to the small- + * drip ceiling and the throttle is floored to >=1 req/s. This keeps the keyless + * default safe for public use without refusing to run. A keyed, non-public + * endpoint (LocationIQ) keeps the full cap and honors the operator's throttle. + * Pure + exported so the safe-default policy can be unit-tested directly. */ +export function dripGeocoderPolicy(env: Env): { limit: number; sleepMs: number; source: string } { + const raw = Number(env.GEOCODE_SLEEP_MS); + const requested = Number.isFinite(raw) && raw >= 0 ? raw : DEFAULT_GEOCODE_SLEEP_MS; + if (isPublicNominatimHost(env.GEOCODER_URL)) { + return { + limit: Math.min(MAX_GEOCODE_PER_TICK, PUBLIC_NOMINATIM_DRIP_MAX), + sleepMs: Math.max(requested, PUBLIC_NOMINATIM_MIN_SLEEP_MS), + source: 'nominatim' + }; + } + return { limit: MAX_GEOCODE_PER_TICK, sleepMs: requested, source: 'locationiq' }; +} + +/** Entry point, called from the cron after firehose ingest. Resolves _geo for + * newly-ingested address-only events so they surface in /near-me. No-ops when + * unconfigured, and self-throttles to GEOCODE_DRIP_INTERVAL_MS via an atomic D1 + * claim so it runs ~every 30 min while riding the every-minute cron — and so two + * overlapping ticks can't both start a drip (the claim is one conditional write, + * not a read-then-write). Idempotent: cached addresses are skipped, so a run with + * no new addresses makes ~zero geocoder calls. `now` is injectable for tests. */ +export async function runGeocodeDrip(env: Env, db: D1Database, now = Date.now()): Promise { + if (!geocodeDripConfigured(env)) return; + await ensureGeocodeDripSchema(db); + + // Atomically claim the cadence slot: a single conditional upsert that advances + // the timestamp ONLY if a full interval has elapsed, reporting whether it won. + // This both gates to the interval and closes the overlap window the old + // read-then-write left open. A claimed-but-failed run waits one interval to + // retry, which is fine for an idempotent drip. + if (!(await claimDripSlot(db, now, GEOCODE_DRIP_INTERVAL_MS))) return; + + const backend = meiliSinkBackendFromEnv(env)!; + const { limit, sleepMs, source } = dripGeocoderPolicy(env); + // A key with no (or a public) URL silently hits Nominatim with the key ignored + // — surface that misconfig rather than letting the operator believe they're on + // LocationIQ. We still run (safely), just on Nominatim's reduced limits. + if (env.GEOCODER_KEY && source === 'nominatim') { + console.warn( + '[geocode-drip] GEOCODER_KEY is set but GEOCODER_URL is unset or public Nominatim — the key ' + + 'is ignored and the drip runs on public Nominatim at its reduced cap/throttle. Set ' + + 'GEOCODER_URL to your LocationIQ endpoint to use the key.' + ); + } + + await runGeocodeJob({ + d1: nativeD1Client(db), + geocoder: createGeocoder(env), + meili: new MeiliEventIndex(backend), + source, + limit, + sleepMs, + now, + log: (m) => console.log('[geocode-drip]', m), + warn: (m) => console.warn('[geocode-drip]', m) + }); +} diff --git a/apps/web/src/lib/search/server/d1-native.ts b/apps/web/src/lib/search/server/d1-native.ts new file mode 100644 index 0000000..eb15f69 --- /dev/null +++ b/apps/web/src/lib/search/server/d1-native.ts @@ -0,0 +1,16 @@ +// Adapt the native Worker D1 binding (platform.env.DB) to the D1Client interface +// the geocode job uses. Lets the same core run in-Worker (the cron drip) over the +// native binding — no REST endpoint, no CLOUDFLARE_API_TOKEN — while the off-box +// CLI keeps the HTTP client in d1-http.ts. One parameterized-query method, the +// only surface the job needs. +import type { D1Client } from './d1-http'; + +export function nativeD1Client(db: D1Database): D1Client { + return { + async query>(sql: string, params: unknown[] = []): Promise { + const stmt = params.length ? db.prepare(sql).bind(...params) : db.prepare(sql); + const { results } = await stmt.all(); + return results ?? []; + } + }; +} diff --git a/apps/web/src/lib/search/server/geocode-job.test.ts b/apps/web/src/lib/search/server/geocode-job.test.ts new file mode 100644 index 0000000..5d34f7e --- /dev/null +++ b/apps/web/src/lib/search/server/geocode-job.test.ts @@ -0,0 +1,420 @@ +import { describe, it, expect, vi } from 'vitest'; +import { runGeocodeJob } from './geocode-job'; +import type { D1Client } from './d1-http'; +import type { GeoPoint } from './geocoder'; +import type { GeocodeCacheRow } from './geocode-cache'; +import { ADDRESS_TYPE, normalizeAddress, addressLocation } from './address-norm'; +import type { MeiliEventIndex } from './meili-sink'; + +const NOW = 1_700_000_000_000; + +type RawEvent = { uri: string; did: string; rkey: string; record: string }; + +const addrRecord = (over: Record = {}) => + JSON.stringify({ + name: 'Test Event', + startsAt: '2026-07-01T18:00:00Z', + locations: [ + { $type: ADDRESS_TYPE, locality: 'Louisville', region: 'KY', country: 'US', ...over } + ] + }); + +/** The normalized cache key the job will compute for a given record — so a + * seeded cache row lines up with the worklist's grouping. */ +const normFor = (record: string) => + normalizeAddress(addressLocation(JSON.parse(record)) as Record); + +/** Minimal D1Client that routes by SQL fragment and captures geocode_cache + * writes. `live` (the fresh re-read for fetchLiveDocs) defaults to the worklist + * but can differ — to model an event deleted mid-run, or one that gained coords. */ +function fakeD1(opts: { cache?: GeocodeCacheRow[]; worklist: RawEvent[]; live?: RawEvent[] }) { + const live = opts.live ?? opts.worklist; + const inserts: { sql: string; params: unknown[] }[] = []; + const client: D1Client = { + async query>(sql: string, params: unknown[] = []): Promise { + if (sql.includes('INSERT INTO geocode_cache')) { + inserts.push({ sql, params }); + return [] as T[]; + } + if (sql.includes('CREATE TABLE')) return [] as T[]; + if (sql.includes('SELECT * FROM geocode_cache')) return (opts.cache ?? []) as T[]; + if (sql.includes('WHERE uri IN')) { + // fetchLiveDocs: only rows whose uri is in the bound params + return live.filter((e) => params.includes(e.uri)) as T[]; + } + if (sql.includes('json_each')) return opts.worklist as T[]; // WORKLIST_SQL + return [] as T[]; + } + }; + return { client, inserts }; +} + +const geocoderReturning = (point: GeoPoint | null) => ({ geocode: vi.fn(async () => point) }); + +function fakeMeili() { + const upsert = vi.fn(async () => {}); + const applySettings = vi.fn(async () => {}); + return { + meili: { upsert, applySettings } as unknown as MeiliEventIndex, + upsert, + applySettings + }; +} + +const resolved = (norm: string): GeocodeCacheRow => ({ + address_norm: norm, + lat: 1, + lng: 2, + precision: 'locality', + source: 'locationiq', + geocoded_at: NOW - 1000, + fail_count: 0, + last_error: null +}); + +describe('runGeocodeJob', () => { + it('geocodes a new address: writes cache and _geo-upserts the doc', async () => { + const worklist = [{ uri: 'at://did:x/c/1', did: 'did:x', rkey: '1', record: addrRecord() }]; + const { client, inserts } = fakeD1({ worklist }); + const { meili, upsert } = fakeMeili(); + const geocoder = geocoderReturning({ lat: 38.25, lng: -85.76, precision: 'locality' }); + + const res = await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + sleepMs: 0, + now: NOW + }); + + expect(geocoder.geocode).toHaveBeenCalledTimes(1); + expect(res.resolved).toBe(1); + // Resolved insert carries lat/lng at params [1],[2]. + const ins = inserts.find((i) => i.params[1] === 38.25); + expect(ins).toBeTruthy(); + expect(upsert).toHaveBeenCalledTimes(1); + const docs = ( + upsert.mock.calls[0] as unknown as [Array<{ _geo?: { lat: number; lng: number } }>] + )[0]; + expect(docs[0]._geo).toEqual({ lat: 38.25, lng: -85.76 }); + }); + + it('is idempotent: skips an already-resolved address (no geocoder call)', async () => { + const record = addrRecord(); + const worklist = [{ uri: 'at://did:x/c/1', did: 'did:x', rkey: '1', record }]; + const { client, inserts } = fakeD1({ worklist, cache: [resolved(normFor(record)!)] }); + const { meili, upsert } = fakeMeili(); + const geocoder = geocoderReturning({ lat: 1, lng: 2 }); + + const res = await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + sleepMs: 0, + now: NOW + }); + + expect(geocoder.geocode).not.toHaveBeenCalled(); + expect(res.processed).toBe(0); + expect(inserts).toHaveLength(0); + expect(upsert).not.toHaveBeenCalled(); + }); + + it('respects the limit cap across unique addresses', async () => { + const worklist = [ + { + uri: 'at://did:x/c/1', + did: 'did:x', + rkey: '1', + record: addrRecord({ locality: 'Berlin' }) + }, + { + uri: 'at://did:x/c/2', + did: 'did:x', + rkey: '2', + record: addrRecord({ locality: 'Brussels' }) + }, + { uri: 'at://did:x/c/3', did: 'did:x', rkey: '3', record: addrRecord({ locality: 'Paris' }) } + ]; + const { client } = fakeD1({ worklist }); + const { meili } = fakeMeili(); + const geocoder = geocoderReturning({ lat: 1, lng: 2 }); + + const res = await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + limit: 2, + sleepMs: 0, + now: NOW + }); + + expect(res.uniqueAddresses).toBe(3); + expect(res.processed).toBe(2); + expect(geocoder.geocode).toHaveBeenCalledTimes(2); + }); + + it('dry-run computes the worklist but makes no geocoder/cache/Meili writes', async () => { + const worklist = [{ uri: 'at://did:x/c/1', did: 'did:x', rkey: '1', record: addrRecord() }]; + const { client, inserts } = fakeD1({ worklist }); + const { meili, upsert } = fakeMeili(); + const geocoder = geocoderReturning({ lat: 1, lng: 2 }); + + const res = await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + dryRun: true, + sleepMs: 0, + now: NOW + }); + + expect(res.processed).toBe(1); + expect(res.resolved).toBe(0); + expect(geocoder.geocode).not.toHaveBeenCalled(); + expect(inserts).toHaveLength(0); + expect(upsert).not.toHaveBeenCalled(); + }); + + it('writes a negative cache row on no-match (and does not upsert)', async () => { + const worklist = [{ uri: 'at://did:x/c/1', did: 'did:x', rkey: '1', record: addrRecord() }]; + const { client, inserts } = fakeD1({ worklist }); + const { meili, upsert } = fakeMeili(); + const geocoder = geocoderReturning(null); + + const res = await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + sleepMs: 0, + now: NOW + }); + + expect(res.negative).toBe(1); + expect(res.resolved).toBe(0); + expect(inserts.some((i) => /no match/.test(i.sql))).toBe(true); + expect(upsert).not.toHaveBeenCalled(); + }); + + it('counts a thrown geocoder error as transient and writes no cache row', async () => { + const worklist = [{ uri: 'at://did:x/c/1', did: 'did:x', rkey: '1', record: addrRecord() }]; + const { client, inserts } = fakeD1({ worklist }); + const { meili } = fakeMeili(); + const geocoder = { + geocode: vi.fn(async () => { + throw new Error('429 rate limited'); + }) + }; + + const res = await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + sleepMs: 0, + now: NOW + }); + + expect(res.transient).toBe(1); + expect(res.resolved).toBe(0); + expect(res.negative).toBe(0); + expect(inserts).toHaveLength(0); + }); + + it('does NOT positive-cache when the Meili upsert fails (address retries, not skips forever)', async () => { + // Geocode succeeds and a live doc exists, but the upsert throws. The positive + // cache row must NOT be written — otherwise the address is cached "resolved" + // yet never indexed, and the idempotent skip locks in a permanent _geo gap. + // It must count transient and stay eligible for the next run. + const worklist = [{ uri: 'at://did:x/c/1', did: 'did:x', rkey: '1', record: addrRecord() }]; + const { client, inserts } = fakeD1({ worklist }); + const upsert = vi.fn(async () => { + throw new Error('Meilisearch PUT failed: 503'); + }); + const applySettings = vi.fn(async () => {}); + const meili = { upsert, applySettings } as unknown as MeiliEventIndex; + const geocoder = geocoderReturning({ lat: 38.25, lng: -85.76 }); + + const res = await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + sleepMs: 0, + now: NOW + }); + + expect(res.transient).toBe(1); + expect(res.resolved).toBe(0); + expect(inserts).toHaveLength(0); // no positive (or any) geocode_cache write + }); + + it('does NOT positive-cache when applySettings fails before the first upsert', async () => { + const worklist = [{ uri: 'at://did:x/c/1', did: 'did:x', rkey: '1', record: addrRecord() }]; + const { client, inserts } = fakeD1({ worklist }); + const upsert = vi.fn(async () => {}); + const applySettings = vi.fn(async () => { + throw new Error('Meilisearch PATCH failed: 500'); + }); + const meili = { upsert, applySettings } as unknown as MeiliEventIndex; + const geocoder = geocoderReturning({ lat: 38.25, lng: -85.76 }); + + const res = await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + sleepMs: 0, + now: NOW + }); + + expect(res.transient).toBe(1); + expect(res.resolved).toBe(0); + expect(upsert).not.toHaveBeenCalled(); + expect(inserts).toHaveLength(0); + }); + + it('applies index settings exactly once, before the first upsert (S3)', async () => { + const worklist = [ + { + uri: 'at://did:x/c/1', + did: 'did:x', + rkey: '1', + record: addrRecord({ locality: 'Berlin' }) + }, + { uri: 'at://did:x/c/2', did: 'did:x', rkey: '2', record: addrRecord({ locality: 'Paris' }) } + ]; + const { client } = fakeD1({ worklist }); + const { meili, upsert, applySettings } = fakeMeili(); + const geocoder = geocoderReturning({ lat: 38.25, lng: -85.76 }); + + await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + sleepMs: 0, + now: NOW + }); + + // Two distinct addresses → two upserts, but settings applied just once... + expect(applySettings).toHaveBeenCalledTimes(1); + expect(upsert).toHaveBeenCalledTimes(2); + // ...and that one application precedes the first upsert (a bare index 400s). + expect(applySettings.mock.invocationCallOrder[0]).toBeLessThan( + upsert.mock.invocationCallOrder[0] + ); + }); + + it('never touches Meili when nothing resolves (a no-match run skips settings + upsert)', async () => { + const worklist = [{ uri: 'at://did:x/c/1', did: 'did:x', rkey: '1', record: addrRecord() }]; + const { client } = fakeD1({ worklist }); + const { meili, upsert, applySettings } = fakeMeili(); + const geocoder = geocoderReturning(null); + + await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + sleepMs: 0, + now: NOW + }); + + expect(applySettings).not.toHaveBeenCalled(); + expect(upsert).not.toHaveBeenCalled(); + }); + + it('caches the address but skips the upsert when the event vanished mid-run (gap 1)', async () => { + const worklist = [{ uri: 'at://did:x/c/1', did: 'did:x', rkey: '1', record: addrRecord() }]; + // In the worklist, but gone (deleted/unlisted) by the fresh re-read. + const { client, inserts } = fakeD1({ worklist, live: [] }); + const { meili, upsert, applySettings } = fakeMeili(); + const geocoder = geocoderReturning({ lat: 38.25, lng: -85.76 }); + + const res = await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + sleepMs: 0, + now: NOW + }); + + expect(res.resolved).toBe(1); // the ADDRESS resolved + cached (won't re-geocode) + expect(res.skippedGone).toBe(1); // but its one event is gone + expect(upsert).not.toHaveBeenCalled(); + expect(applySettings).not.toHaveBeenCalled(); + expect(inserts.some((i) => i.params[1] === 38.25)).toBe(true); + }); + + it('geocodes a shared address once and upserts every event carrying it (gap 2)', async () => { + const record = addrRecord(); // identical address on two distinct events + const worklist = [ + { uri: 'at://did:x/c/1', did: 'did:x', rkey: '1', record }, + { uri: 'at://did:y/c/2', did: 'did:y', rkey: '2', record } + ]; + const { client } = fakeD1({ worklist }); + const { meili, upsert } = fakeMeili(); + const geocoder = geocoderReturning({ lat: 38.25, lng: -85.76 }); + + const res = await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + sleepMs: 0, + now: NOW + }); + + expect(geocoder.geocode).toHaveBeenCalledTimes(1); // one address → one geocoder call + expect(res.uniqueAddresses).toBe(1); + expect(res.resolved).toBe(1); + expect(upsert).toHaveBeenCalledTimes(1); + const docs = ( + upsert.mock.calls[0] as unknown as [Array<{ _geo?: { lat: number; lng: number } }>] + )[0]; + expect(docs).toHaveLength(2); // both events upserted in one batch + expect(docs.every((d) => d._geo?.lat === 38.25)).toBe(true); + }); + + it('does not overwrite a coordinate _geo the record gained mid-run (gap 3)', async () => { + // Worklist sees an address-only record (eligible to geocode), but the fresh + // re-read finds it now also carries an in-range geo location. + const uri = 'at://did:x/c/1'; + const liveRecord = JSON.stringify({ + name: 'Test Event', + startsAt: '2026-07-01T18:00:00Z', + locations: [ + { $type: ADDRESS_TYPE, locality: 'Louisville', region: 'KY', country: 'US' }, + { $type: 'community.lexicon.location.geo', latitude: '40.0', longitude: '-80.0' } + ] + }); + const { client } = fakeD1({ + worklist: [{ uri, did: 'did:x', rkey: '1', record: addrRecord() }], + live: [{ uri, did: 'did:x', rkey: '1', record: liveRecord }] + }); + const { meili, upsert } = fakeMeili(); + const geocoder = geocoderReturning({ lat: 38.25, lng: -85.76 }); + + await runGeocodeJob({ + d1: client, + geocoder, + meili, + source: 'locationiq', + sleepMs: 0, + now: NOW + }); + + const docs = ( + upsert.mock.calls[0] as unknown as [Array<{ _geo?: { lat: number; lng: number } }>] + )[0]; + // The record's own precise coords win — the geocoder point must not clobber them. + expect(docs[0]._geo).toEqual({ lat: 40, lng: -80 }); + }); +}); diff --git a/apps/web/src/lib/search/server/geocode-job.ts b/apps/web/src/lib/search/server/geocode-job.ts new file mode 100644 index 0000000..6f70140 --- /dev/null +++ b/apps/web/src/lib/search/server/geocode-job.ts @@ -0,0 +1,293 @@ +// Shared core of the address→_geo geocode job: find address-only events that +// lack coordinates, geocode each unique address through the injected geocoder, +// write geocode_cache, and _geo-upsert the affected Meili docs. ONE job: "find +// work" and "geocode" are sequential steps sharing the cache, not two passes. +// +// Deliberately I/O-agnostic — it takes an injected D1Client, Geocoder, and +// MeiliEventIndex rather than reading env or constructing clients. Two callers: +// • scripts/geocode-events.ts — off-box CLI, D1 over the REST client, one-shot +// backfill / manual drip (keeps the CLI flags + bulk guard). +// • lib/geocode/process.ts — in-Worker cron drip, D1 over the native binding +// (no CLOUDFLARE_API_TOKEN), self-throttled to a cadence. +// Rate limiting is the caller-supplied sleep between calls — the whole reason +// geocoding stays off the ingest hot path. +import { addressToQuery, type Geocoder } from './geocoder'; +import { + isEligible, + groupEventsByNorm, + addressNeedingGeocode, + type GeocodeCacheRow, + type WorklistEvent +} from './geocode-cache'; +import { eventToSearchDoc } from './normalize'; +import { discoverableSql } from './discoverability'; +import { MeiliEventIndex, EVENT_COLLECTION } from './meili-sink'; +import type { D1Client } from './d1-http'; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +// Worklist: discoverable events carrying a .address location. We deliberately +// do NOT exclude coordinate locations in SQL — that filtered by $type presence, +// which diverges from the sink's actual _geo derivation (it ignored fsq, and +// excluded events whose only geo/hthree coords are out of range and so never +// get an in-index _geo). The precise "does this already resolve to coordinates?" +// decision is made in memory by addressNeedingGeocode (recordGeo), the same +// derivation the sink uses. json_each walks locations[]; the "$type" key is +// quoted because it starts with $. +// +// The CASE WHEN json_valid(r.record) guard is REQUIRED, not belt-and-suspenders: +// json_each / json_extract raise "malformed JSON" and abort the ENTIRE query on a +// single poison record, which would fail the whole drip/backfill — so one bad row +// must not even reach them. We use CASE rather than `json_valid(...) AND ...` +// because SQLite guarantees lazy, per-branch evaluation for CASE, whereas WHERE +// conjuncts are optimizer-reorderable (their left-to-right short-circuit is not a +// contract): the THEN holding json_each/json_extract is evaluated ONLY when +// json_valid is true. (The in-memory JSON.parse skip below is a second layer, but +// it never runs if the SQL itself aborts.) +const WORKLIST_SQL = ` +SELECT r.uri AS uri, r.did AS did, r.rkey AS rkey, r.record AS record +FROM records_event AS r +WHERE CASE WHEN json_valid(r.record) THEN ( + EXISTS ( + SELECT 1 FROM json_each(r.record, '$.locations') + WHERE json_extract(value, '$."$type"') = 'community.lexicon.location.address') + AND ${discoverableSql('r.record')} + ) ELSE 0 END +`; + +// Re-read events from records_event by uri, keeping only those still present and +// discoverable. The parsed record drives a full-doc upsert, so it's read fresh +// (not from the job-start worklist) — an event deleted or unlisted while the slow +// geocode loop runs is skipped here instead of being resurrected as a stale doc. +async function fetchLiveDocs( + d1: D1Client, + uris: string[] +): Promise<{ uri: string; did: string; rkey: string; record: Record }[]> { + if (uris.length === 0) return []; + const placeholders = uris.map(() => '?').join(','); + // CASE-guard discoverableSql for the same reason as WORKLIST_SQL: json_extract + // aborts the query on a poison record. CASE guarantees the THEN (json_extract) + // runs ONLY when json_valid is true — robust against the WHERE-conjunct + // reordering a bare `... AND json_valid(record) AND json_extract(...)` leaves to + // the optimizer. uri IN (...) still narrows the scan to the candidate rows. + const rows = await d1.query<{ uri: string; did: string; rkey: string; record: string }>( + `SELECT uri, did, rkey, record FROM records_event + WHERE uri IN (${placeholders}) + AND CASE WHEN json_valid(record) THEN ${discoverableSql('record')} ELSE 0 END`, + uris + ); + const out: { uri: string; did: string; rkey: string; record: Record }[] = []; + for (const r of rows) { + try { + out.push({ uri: r.uri, did: r.did, rkey: r.rkey, record: JSON.parse(r.record) }); + } catch { + // Unparseable record JSON — skip, same as the worklist parse. + } + } + return out; +} + +export interface GeocodeJobOptions { + d1: D1Client; + geocoder: Geocoder; + meili: MeiliEventIndex; + /** Cache provenance tag written to geocode_cache.source ('locationiq'|'nominatim'). */ + source: string; + /** Max unique addresses to process this run; 0 = no cap. */ + limit?: number; + /** Delay between geocoder calls in ms — the rate limiter. */ + sleepMs?: number; + /** Force-retry negative-cached addresses (ignores backoff). */ + retryNegative?: boolean; + /** Compute the worklist and log it, but make no geocoder/cache/Meili writes. */ + dryRun?: boolean; + /** Injectable clock (eligibility backoff + geocoded_at); defaults to now. */ + now?: number; + /** Progress line sink; callers prefix as they like. Default: no-op. */ + log?: (msg: string) => void; + /** Transient-error sink. Default: no-op. */ + warn?: (msg: string) => void; +} + +export interface GeocodeJobResult { + worklistEvents: number; + uniqueAddresses: number; + eligible: number; + processed: number; + resolved: number; + negative: number; + transient: number; + skippedGone: number; +} + +/** Run one geocode pass over the eligible worklist. Pure orchestration over the + * injected clients; idempotent (cached addresses are skipped via isEligible), so + * a steady-state run with no new addresses makes ~zero geocoder calls. */ +export async function runGeocodeJob(opts: GeocodeJobOptions): Promise { + const { + d1, + geocoder, + meili, + source, + limit = 0, + sleepMs, + retryNegative = false, + dryRun = false, + now = Date.now(), + log = () => {}, + warn = () => {} + } = opts; + + // Defensive throttle: a caller passing NaN/negative/undefined (e.g. a malformed + // GEOCODE_SLEEP_MS that slipped past a caller's own parse) must not collapse the + // rate limiter into a busy loop. undefined → the safe 1100ms default; any + // non-finite or negative value → the same. This is the single clamp both the + // Worker drip and the CLI rely on, so the loop below never sleeps on a bad value. + const throttleMs = + typeof sleepMs === 'number' && Number.isFinite(sleepMs) && sleepMs >= 0 ? sleepMs : 1100; + + // Defensive: the table is created by geocode-cache.sql, but a fresh DB + // shouldn't make the job crash before it can self-heal. + await d1.query( + `CREATE TABLE IF NOT EXISTS geocode_cache ( + address_norm TEXT PRIMARY KEY, lat REAL, lng REAL, precision TEXT, + source TEXT NOT NULL, geocoded_at INTEGER NOT NULL, + fail_count INTEGER NOT NULL DEFAULT 0, last_error TEXT)` + ); + + // Load the whole cache once (small) and decide eligibility in memory. + const cacheRows = await d1.query(`SELECT * FROM geocode_cache`); + const cache = new Map(cacheRows.map((r) => [r.address_norm, r])); + + // Worklist → WorklistEvent[] (parse record JSON; keep only events that need + // geocoding — an address location AND no coordinates the index already + // derives, per addressNeedingGeocode). + const rawEvents = await d1.query<{ uri: string; did: string; rkey: string; record: string }>( + WORKLIST_SQL + ); + const events: WorklistEvent[] = []; + for (const e of rawEvents) { + let record: Record; + try { + record = JSON.parse(e.record); + } catch { + continue; + } + const loc = addressNeedingGeocode(record); + if (loc) events.push({ uri: e.uri, did: e.did, rkey: e.rkey, loc }); + } + + const byNorm = groupEventsByNorm(events); + const work = [...byNorm.entries()].filter(([norm]) => + isEligible(cache.get(norm), now, retryNegative) + ); + const capped = limit > 0 ? work.slice(0, limit) : work; + + log( + `worklist events=${events.length} unique-addresses=${byNorm.size} ` + + `eligible=${work.length} processing=${capped.length}${dryRun ? ' (dry-run)' : ''}` + ); + + let resolved = 0; + let negative = 0; + let transient = 0; + let skippedGone = 0; + // Apply the read-path's filterable/sortable settings once, lazily, before the + // FIRST upsert. A PUT to a bare auto-created index would 400 the read path's + // _geo/startsAt filters; the sink ensures this for live ingest, but a drip can + // be the first writer to a freshly (re)created index. Only fires when there's + // actually a doc to write, so a no-resolution run touches Meili zero times. + let settingsApplied = false; + for (let i = 0; i < capped.length; i++) { + const [norm, group] = capped[i]; + const query = addressToQuery(group[0].loc); + if (dryRun) { + log(`would geocode "${query}" -> ${group.length} event(s)`); + continue; + } + try { + const point = await geocoder.geocode(query); + if (point) { + // Re-read the events fresh and upsert the FULL doc with _geo attached, + // keeping only those still present AND discoverable. A full-doc upsert is + // idempotent and needs no "is it indexed?" snapshot: it merges if the + // event is already indexed, lands a complete doc if not (never a {id,_geo} + // stub), and is identical to what the sink writes — so a concurrent sink + // write converges instead of clobbering. + const live = await fetchLiveDocs( + d1, + group.map((e) => e.uri) + ); + skippedGone += group.length - live.length; + if (live.length) { + if (!settingsApplied) { + await meili.applySettings(); + settingsApplied = true; + } + await meili.upsert( + live.map((r) => { + const doc = eventToSearchDoc({ + uri: r.uri, + did: r.did, + collection: EVENT_COLLECTION, + rkey: r.rkey, + record: r.record + }); + // Don't overwrite a coordinate _geo the record gained mid-run. + if (!doc._geo) doc._geo = { lat: point.lat, lng: point.lng }; + return doc; + }) + ); + } + // Write the positive cache row ONLY after the Meili write succeeds (or + // when there's no live doc to write). If applySettings/upsert throws, the + // catch below counts it transient and we do NOT cache — so the address + // stays eligible and retries next run, instead of being cached "resolved" + // yet never indexed: a permanent _geo gap the idempotent skip would lock + // in. A redundant re-geocode on the rare Meili failure is the cheap price. + await d1.query( + `INSERT INTO geocode_cache (address_norm, lat, lng, precision, source, geocoded_at, fail_count, last_error) + VALUES (?, ?, ?, ?, ?, ?, 0, NULL) + ON CONFLICT(address_norm) DO UPDATE SET + lat=excluded.lat, lng=excluded.lng, precision=excluded.precision, + source=excluded.source, geocoded_at=excluded.geocoded_at, fail_count=0, last_error=NULL`, + [norm, point.lat, point.lng, point.precision ?? null, source, now] + ); + resolved++; + } else { + // No-match: write/increment a negative row (backoff handled by isEligible). + await d1.query( + `INSERT INTO geocode_cache (address_norm, lat, lng, precision, source, geocoded_at, fail_count, last_error) + VALUES (?, NULL, NULL, NULL, ?, ?, 1, 'no match') + ON CONFLICT(address_norm) DO UPDATE SET + source=excluded.source, geocoded_at=excluded.geocoded_at, fail_count=geocode_cache.fail_count+1, last_error='no match'`, + [norm, source, now] + ); + negative++; + } + } catch (err) { + // Transient (HTTP/network): DON'T write a negative row — retry next run. + transient++; + warn(`transient error for "${query}": ${(err as Error).message}`); + } + // Throttle BETWEEN geocoder calls only — skip the trailing sleep after the + // last one, which would just waste a sleepMs at the end of every run. + if (i < capped.length - 1) await sleep(throttleMs); + } + + log( + `done resolved=${resolved} negative=${negative} transient=${transient} ` + + `skipped-gone=${skippedGone}` + ); + + return { + worklistEvents: events.length, + uniqueAddresses: byNorm.size, + eligible: work.length, + processed: capped.length, + resolved, + negative, + transient, + skippedGone + }; +} diff --git a/apps/web/src/lib/search/server/geocoder.test.ts b/apps/web/src/lib/search/server/geocoder.test.ts index c11ba58..6163fd4 100644 --- a/apps/web/src/lib/search/server/geocoder.test.ts +++ b/apps/web/src/lib/search/server/geocoder.test.ts @@ -4,7 +4,8 @@ import { addressToQuery, derivePrecision, createGeocoder, - requireGeocoderForBulk + requireGeocoderForBulk, + isPublicNominatimHost } from './geocoder'; describe('addressToQuery', () => { @@ -82,6 +83,20 @@ describe('createGeocoder', () => { expect(await createGeocoder({}, fn).geocode('nowhere')).toBeNull(); }); + it('returns null on out-of-WGS84-range coords (so a bad hit negative-caches, not poisons Meili)', async () => { + // A finite-but-out-of-range result must not be cached as resolved: Meili + // silently fails the whole _geo batch on such a doc. Reject → null → negative. + const { fn } = fakeFetch([{ lat: '91.5', lon: '-85.76', addresstype: 'city' }]); + expect(await createGeocoder({}, fn).geocode('off the map')).toBeNull(); + const { fn: fn2 } = fakeFetch([{ lat: '38.25', lon: '-200', addresstype: 'city' }]); + expect(await createGeocoder({}, fn2).geocode('off the map')).toBeNull(); + }); + + it('returns null on non-finite coords', async () => { + const { fn } = fakeFetch([{ lat: 'not-a-number', lon: '4.36', addresstype: 'city' }]); + expect(await createGeocoder({}, fn).geocode('garbled')).toBeNull(); + }); + it('throws on a transient HTTP error (429/5xx) so the caller retries', async () => { const { fn } = fakeFetch({}, 429); await expect(createGeocoder({}, fn).geocode('x')).rejects.toThrow(/429/); @@ -89,52 +104,119 @@ describe('createGeocoder', () => { }); describe('requireGeocoderForBulk', () => { - it('allows any run when a geocoder key is set', () => { + it('allows any run against a non-public endpoint (keyed LocationIQ or self-hosted)', () => { expect(() => - requireGeocoderForBulk({ hasKey: true, dryRun: false, limit: 0, allowPublic: false }) + requireGeocoderForBulk({ + nonPublicEndpoint: true, + dryRun: false, + limit: 0, + allowPublic: false + }) ).not.toThrow(); expect(() => - requireGeocoderForBulk({ hasKey: true, dryRun: false, limit: 5000, allowPublic: false }) + requireGeocoderForBulk({ + nonPublicEndpoint: true, + dryRun: false, + limit: 5000, + allowPublic: false + }) ).not.toThrow(); }); it('allows a keyless small drip (1..25)', () => { expect(() => - requireGeocoderForBulk({ hasKey: false, dryRun: false, limit: 25, allowPublic: false }) + requireGeocoderForBulk({ + nonPublicEndpoint: false, + dryRun: false, + limit: 25, + allowPublic: false + }) ).not.toThrow(); expect(() => - requireGeocoderForBulk({ hasKey: false, dryRun: false, limit: 1, allowPublic: false }) + requireGeocoderForBulk({ + nonPublicEndpoint: false, + dryRun: false, + limit: 1, + allowPublic: false + }) ).not.toThrow(); }); it('blocks a keyless run over the drip ceiling', () => { expect(() => - requireGeocoderForBulk({ hasKey: false, dryRun: false, limit: 26, allowPublic: false }) + requireGeocoderForBulk({ + nonPublicEndpoint: false, + dryRun: false, + limit: 26, + allowPublic: false + }) ).toThrow(/LocationIQ|allow-public-nominatim/); }); it('blocks a keyless uncapped run (limit 0 = no cap, the worst case)', () => { expect(() => - requireGeocoderForBulk({ hasKey: false, dryRun: false, limit: 0, allowPublic: false }) + requireGeocoderForBulk({ + nonPublicEndpoint: false, + dryRun: false, + limit: 0, + allowPublic: false + }) ).toThrow(/LocationIQ|allow-public-nominatim/); }); it('blocks a keyless negative limit (uncapped, not a tiny drip)', () => { // A stray `--limit -1` must not masquerade as a 1-call drip and slip the gate. expect(() => - requireGeocoderForBulk({ hasKey: false, dryRun: false, limit: -1, allowPublic: false }) + requireGeocoderForBulk({ + nonPublicEndpoint: false, + dryRun: false, + limit: -1, + allowPublic: false + }) ).toThrow(/LocationIQ|allow-public-nominatim/); }); it('lets --allow-public-nominatim override a bulk keyless run', () => { expect(() => - requireGeocoderForBulk({ hasKey: false, dryRun: false, limit: 0, allowPublic: true }) + requireGeocoderForBulk({ + nonPublicEndpoint: false, + dryRun: false, + limit: 0, + allowPublic: true + }) ).not.toThrow(); }); it('never blocks a dry run (it makes no geocoder calls)', () => { expect(() => - requireGeocoderForBulk({ hasKey: false, dryRun: true, limit: 0, allowPublic: false }) + requireGeocoderForBulk({ + nonPublicEndpoint: false, + dryRun: true, + limit: 0, + allowPublic: false + }) ).not.toThrow(); }); }); + +describe('isPublicNominatimHost', () => { + it('flags the public OSM Nominatim host', () => { + expect(isPublicNominatimHost('https://nominatim.openstreetmap.org/search')).toBe(true); + }); + + it('does not flag a keyed LocationIQ endpoint', () => { + expect(isPublicNominatimHost('https://us1.locationiq.com/v1/search')).toBe(false); + }); + + it('treats unset or malformed URLs as public (fail safe toward Nominatim limits)', () => { + expect(isPublicNominatimHost(undefined)).toBe(true); + expect(isPublicNominatimHost('')).toBe(true); + expect(isPublicNominatimHost('not a url')).toBe(true); + }); + + it('is host-based, not substring-based (a path mentioning the host is not public)', () => { + expect(isPublicNominatimHost('https://geo.example.com/nominatim.openstreetmap.org')).toBe( + false + ); + }); +}); diff --git a/apps/web/src/lib/search/server/geocoder.ts b/apps/web/src/lib/search/server/geocoder.ts index c752e2b..595db97 100644 --- a/apps/web/src/lib/search/server/geocoder.ts +++ b/apps/web/src/lib/search/server/geocoder.ts @@ -6,6 +6,7 @@ // behind an interface. Default = public Nominatim (parity with atmo today); // set GEOCODER_KEY (+ GEOCODER_URL) to use LocationIQ. Used only by the // external geocode job — geocoding never runs on the Worker hot path. +import { inGeoRange } from './normalize'; export interface GeoPoint { lat: number; @@ -26,6 +27,25 @@ export interface GeocoderEnv { const DEFAULT_URL = 'https://nominatim.openstreetmap.org/search'; const DEFAULT_USER_AGENT = 'atmo-events (https://atmo.rsvp)'; +const PUBLIC_NOMINATIM_HOST = 'nominatim.openstreetmap.org'; + +/** True when the effective geocoder endpoint is public OSM Nominatim — i.e. the + * shared, usage-policy-bound host. The in-Worker drip uses this to pick SAFE + * defaults (a smaller per-tick cap + a slower throttle floor) when on Nominatim, + * rather than refusing to run: the drip works keyless out of the box, just + * policy-compliantly. `createGeocoder` falls back to DEFAULT_URL (this host) + * when GEOCODER_URL is unset — even with a key present, in which case the ?key= + * is ignored and we're really on Nominatim — so an unset URL counts as public. + * Malformed URL → treated as public (fail safe toward the slower limits). A + * keyed, non-public GEOCODER_URL (LocationIQ) is the only way to lift them. */ +export function isPublicNominatimHost(url: string | undefined): boolean { + if (!url) return true; + try { + return new URL(url).hostname.toLowerCase() === PUBLIC_NOMINATIM_HOST; + } catch { + return true; + } +} // Same fixed order as the cache key, but a human-readable freeform query (commas, // original case/diacritics) — both Nominatim and LocationIQ handle UTF-8 / non- @@ -34,28 +54,40 @@ const DEFAULT_USER_AGENT = 'atmo-events (https://atmo.rsvp)'; const QUERY_FIELDS = ['name', 'street', 'locality', 'region', 'postalCode', 'country']; /** Largest keyless run we treat as a sanctioned "small drip" against public - * Nominatim. Above this (or uncapped), a backfill must use LocationIQ. */ + * Nominatim. Above this (or uncapped), a backfill must use LocationIQ. The + * in-Worker drip also clamps its per-tick cap to this when on public Nominatim. */ export const PUBLIC_NOMINATIM_DRIP_MAX = 25; -/** Enforce the file's contract that a BULK backfill uses LocationIQ, not public - * Nominatim (whose usage policy a large unkeyed run would breach, risking a - * silent IP ban). Throws for a keyless, non-dry-run, non-overridden run that is - * uncapped (limit <= 0) or over the drip ceiling; a small keyless drip stays - * allowed. The hazard is request VOLUME, so the gate is on size, not on the - * mere use of Nominatim. limit <= 0 (not just === 0) counts as uncapped so a - * stray negative can't masquerade as a tiny drip and slip the gate. */ +/** Throttle floor (ms between calls) for public Nominatim — its usage policy is + * an absolute max of 1 request/second. The drip enforces this as a FLOOR when on + * Nominatim (a faster GEOCODE_SLEEP_MS override can't undercut the policy); a + * keyed LocationIQ endpoint honors the operator's own, possibly faster, value. */ +export const PUBLIC_NOMINATIM_MIN_SLEEP_MS = 1000; + +/** Enforce the file's contract that a BULK backfill runs against a NON-PUBLIC + * endpoint (a keyed LocationIQ URL, or a self-hosted host), not public OSM + * Nominatim — whose usage policy a large run would breach, risking a silent IP + * ban. The deciding input is the EFFECTIVE endpoint, not key presence: a + * GEOCODER_KEY with an unset/public GEOCODER_URL still hits public Nominatim (the + * ?key= is ignored), so callers pass `!isPublicNominatimHost(GEOCODER_URL)` here + * rather than `!!GEOCODER_KEY`. Throws for a public-Nominatim, non-dry-run, + * non-overridden run that is uncapped (limit <= 0) or over the drip ceiling; a + * small public drip stays allowed. The hazard is request VOLUME, so the gate is + * on size. limit <= 0 (not just === 0) counts as uncapped so a stray negative + * can't masquerade as a tiny drip and slip the gate. */ export function requireGeocoderForBulk(opts: { - hasKey: boolean; + nonPublicEndpoint: boolean; dryRun: boolean; limit: number; allowPublic: boolean; }): void { - if (opts.hasKey || opts.dryRun || opts.allowPublic) return; + if (opts.nonPublicEndpoint || opts.dryRun || opts.allowPublic) return; const bulk = opts.limit <= 0 || opts.limit > PUBLIC_NOMINATIM_DRIP_MAX; if (bulk) { throw new Error( `Keyless public Nominatim is only allowed for a small drip (--limit 1..${PUBLIC_NOMINATIM_DRIP_MAX}). ` + - 'Set GEOCODER_KEY (LocationIQ) for a bulk/uncapped backfill, or pass --allow-public-nominatim to override.' + 'Set GEOCODER_URL to your LocationIQ endpoint (with GEOCODER_KEY) for a bulk/uncapped backfill, ' + + 'or pass --allow-public-nominatim to override.' ); } } @@ -136,7 +168,14 @@ export function createGeocoder(env: GeocoderEnv = {}, fetchImpl: typeof fetch = const lat = Number(top.lat); const lng = Number(top.lon); - if (!Number.isFinite(lat) || !Number.isFinite(lng)) return null; + // Reject non-finite OR out-of-WGS84-range coordinates. Meilisearch + // silently fails the whole async index task on a bad _geo (losing the + // batch), and the sink guards its own path identically via the SAME + // inGeoRange (normalize.ts). An out-of-range hit therefore becomes a + // no-match (return null → negative cache w/ backoff) rather than a + // poisoned coordinate that's cached as "resolved" and never retried. + // inGeoRange returns false for NaN, so it subsumes the finite check. + if (!inGeoRange(lat, lng)) return null; return { lat, lng, precision: derivePrecision(top) }; } diff --git a/apps/web/src/lib/search/server/normalize.ts b/apps/web/src/lib/search/server/normalize.ts index ab9e610..0dd0e63 100644 --- a/apps/web/src/lib/search/server/normalize.ts +++ b/apps/web/src/lib/search/server/normalize.ts @@ -54,7 +54,11 @@ type Loc = Record; // finite-but-out-of-range value. Meilisearch rejects such a doc and can fail // the whole async indexing task (losing the rest of the batch), so we drop the // _geo here rather than index it. Bounds are WGS84: lat [-90, 90], lng [-180, 180]. -function inGeoRange(lat: number, lng: number): boolean { +// Exported so the geocoder (geocoder.ts) range-checks its results against the +// EXACT same bounds — a geocoder hit and a record's own coords must not be able +// to disagree on what Meili will accept. NaN fails every comparison, so this also +// rejects non-finite input. +export function inGeoRange(lat: number, lng: number): boolean { return lat >= -90 && lat <= 90 && lng >= -180 && lng <= 180; } diff --git a/apps/web/src/routes/api/cron/+server.ts b/apps/web/src/routes/api/cron/+server.ts index 9e65774..6f19117 100644 --- a/apps/web/src/routes/api/cron/+server.ts +++ b/apps/web/src/routes/api/cron/+server.ts @@ -1,6 +1,7 @@ import { contrail, ensureInit } from '$lib/contrail/index'; import { processBotMentions } from '$lib/bot/process-mentions'; import { runNotifications } from '$lib/notify/process'; +import { runGeocodeDrip } from '$lib/geocode/process'; import type { RequestHandler } from './$types'; export const POST: RequestHandler = async ({ request, platform }) => { @@ -60,5 +61,15 @@ export const POST: RequestHandler = async ({ request, platform }) => { console.error('[cron] runNotifications failed:', e); } + // Address→_geo geocode drip: resolves coordinates for newly-ingested + // address-only events so they appear in /near-me. Self-throttled to ~30 min + // via a D1 gate (rides this every-minute cron); no-ops when the geocoder/sink + // isn't configured. Isolated so a geocoder/Meili hiccup can't 500 the tick. + try { + await runGeocodeDrip(platform!.env, db); + } catch (e) { + console.error('[cron] runGeocodeDrip failed:', e); + } + return new Response('OK'); }; -- 2.51.2