mirror of
https://github.com/bilawalsidhu/gods-eye-view.git
synced 2026-11-13 06:42:23 +00:00
Merge branch 'main' into denis/gev-news-layer
Wildlife Sightings (#1591) published share token 9, so News takes the next free token, 00. The token tests read the ledger before both where they were written against it, and the README counts both layers.
This commit is contained in:
Denis Groshev committed
commit
2e87dd87eb
144 files changed
+26748
-489
No files matched your search
@@ -0,0 +1,181 @@
|
||||
import { normalizeAcartiaSightings } from '../../src/layers/wildlifeSightings/records.js';
|
||||
import { coalesceProxyRequest, readResponseJsonCapped } from './common/http.js';
|
||||
import { makeRateLimiter, clientKey } from './common/rate-limit.js';
|
||||
import { admitSameSiteKeyless } from './common/same-site.js';
|
||||
import { providerUserAgent } from './common/userAgent.js';
|
||||
|
||||
const API_URL = 'https://acartia.io/api/v1/sightings/current';
|
||||
// All views share the provider's seven-day snapshot.
|
||||
const TTL_MS = 600_000;
|
||||
const STALE_MAX_MS = 24 * 60 * 60_000;
|
||||
const MAX_RESPONSE_BYTES = 2 * 1024 * 1024;
|
||||
const TIMEOUT_MS = 15_000;
|
||||
const COOLDOWN_MS = 60_000;
|
||||
const COOLDOWN_MIN_MS = 5000;
|
||||
const COOLDOWN_MAX_MS = 60 * 60_000;
|
||||
const FAILURE_COOLDOWN_MS = 15_000;
|
||||
|
||||
/** Bound upstream Retry-After seconds or dates to five seconds–one hour. */
|
||||
export function acartiaRetryCooldownMs(raw, nowMs) {
|
||||
const value = typeof raw === 'string' ? raw.trim() : '';
|
||||
if (value) {
|
||||
const seconds = /^\d+(?:\.\d+)?$/.test(value) ? Number(value) : NaN;
|
||||
const deadline =
|
||||
!Number.isFinite(seconds) && /[a-z]/i.test(value)
|
||||
? Date.parse(value)
|
||||
: NaN;
|
||||
const delay = Number.isFinite(seconds) ? seconds * 1000 : deadline - nowMs;
|
||||
if (Number.isFinite(delay) && delay >= 0)
|
||||
return Math.min(COOLDOWN_MAX_MS, Math.max(COOLDOWN_MIN_MS, delay));
|
||||
}
|
||||
return COOLDOWN_MS;
|
||||
}
|
||||
|
||||
/** Salish Sea coverage. Precise coordinates and comments never enter cache. */
|
||||
export function acartiaProxy({
|
||||
fetchImpl = (...args) => globalThis.fetch(...args),
|
||||
now = () => Date.now(),
|
||||
cache = new Map(),
|
||||
warn = (...args) => console.warn(...args),
|
||||
} = {}) {
|
||||
const inFlight = new Map();
|
||||
const allow = makeRateLimiter({
|
||||
windowMs: 60_000,
|
||||
max: 60,
|
||||
globalMax: 600,
|
||||
now,
|
||||
});
|
||||
let cooldownUntil = 0;
|
||||
let cooldownStatus = 429;
|
||||
let warnedClamp = false;
|
||||
|
||||
const retrySeconds = () =>
|
||||
Math.max(1, Math.ceil((cooldownUntil - now()) / 1000));
|
||||
const unavailable = (status = 502) =>
|
||||
Object.assign(new Error('acartia_unavailable'), {
|
||||
status,
|
||||
retryAfterS: retrySeconds(),
|
||||
});
|
||||
|
||||
function coolDown(status, delay) {
|
||||
const until = now() + delay;
|
||||
if (until > cooldownUntil) {
|
||||
cooldownUntil = until;
|
||||
cooldownStatus = status;
|
||||
} else if (status === 429 && now() < cooldownUntil) cooldownStatus = 429;
|
||||
}
|
||||
|
||||
async function load() {
|
||||
if (now() < cooldownUntil) throw unavailable(cooldownStatus);
|
||||
const signal = AbortSignal.timeout(TIMEOUT_MS);
|
||||
try {
|
||||
const response = await fetchImpl(API_URL, {
|
||||
signal,
|
||||
redirect: 'error',
|
||||
headers: {
|
||||
Accept: 'application/json',
|
||||
'User-Agent': providerUserAgent('wildlife-sightings'),
|
||||
},
|
||||
});
|
||||
if (!response.ok) {
|
||||
await response.body?.cancel().catch(() => {});
|
||||
if (response.status === 429) {
|
||||
const delay = acartiaRetryCooldownMs(
|
||||
response.headers.get('retry-after'),
|
||||
now(),
|
||||
);
|
||||
if (delay === COOLDOWN_MAX_MS && !warnedClamp) {
|
||||
warnedClamp = true;
|
||||
warn('[acartia-proxy] upstream cooldown capped at one hour');
|
||||
}
|
||||
coolDown(429, delay);
|
||||
throw unavailable(429);
|
||||
}
|
||||
throw unavailable();
|
||||
}
|
||||
// This endpoint documents a JSON array, not a results/data envelope.
|
||||
const payload = await readResponseJsonCapped(
|
||||
response,
|
||||
MAX_RESPONSE_BYTES,
|
||||
signal,
|
||||
);
|
||||
const rows = normalizeAcartiaSightings(payload, now());
|
||||
// Only a non-array is malformed. A quiet feed, or one whose rows all fall
|
||||
// outside coverage or the species checks, is a valid empty snapshot.
|
||||
if (!rows) throw new Error('invalid_snapshot');
|
||||
return { fetchedAt: now(), rows };
|
||||
} catch (error) {
|
||||
if (error.status !== 429) coolDown(502, FAILURE_COOLDOWN_MS);
|
||||
warn('[acartia-proxy] upstream refresh failed');
|
||||
throw unavailable(error.status === 429 ? 429 : 502);
|
||||
}
|
||||
}
|
||||
|
||||
function age(entry) {
|
||||
if (!entry) return Infinity;
|
||||
const elapsed = now() - entry.savedAt;
|
||||
return elapsed < 0 ? Infinity : elapsed;
|
||||
}
|
||||
|
||||
async function acquire() {
|
||||
const previous = cache.get('sightings');
|
||||
if (age(previous) < TTL_MS) return { value: previous.value, stale: false };
|
||||
if (age(previous) > STALE_MAX_MS) cache.delete('sightings');
|
||||
try {
|
||||
const { promise } = coalesceProxyRequest(
|
||||
inFlight,
|
||||
'sightings',
|
||||
async () => {
|
||||
const value = await load();
|
||||
cache.set('sightings', { value, savedAt: now() });
|
||||
return value;
|
||||
},
|
||||
);
|
||||
return { value: await promise, stale: false };
|
||||
} catch (error) {
|
||||
if (age(previous) <= STALE_MAX_MS)
|
||||
return { value: previous.value, stale: true };
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function handler(req, res) {
|
||||
const json = (status, value, { stale = false, retryAfterS } = {}) => {
|
||||
if (res.destroyed) return;
|
||||
res.writeHead(status, {
|
||||
'Content-Type': 'application/json',
|
||||
'Cache-Control': 'no-store',
|
||||
...(status === 405 ? { Allow: 'GET' } : {}),
|
||||
...(retryAfterS ? { 'Retry-After': String(retryAfterS) } : {}),
|
||||
...(stale ? { 'X-Data-Stale': 'true' } : {}),
|
||||
});
|
||||
res.end(JSON.stringify(value));
|
||||
};
|
||||
if (admitSameSiteKeyless(req, res)) return;
|
||||
if (req.method !== 'GET') return json(405, { error: 'method_not_allowed' });
|
||||
const pathname = (req.url || '/').split('?')[0];
|
||||
if (pathname !== '/sightings') return json(404, { error: 'unknown_route' });
|
||||
if (!allow(clientKey(req)))
|
||||
return json(429, { error: 'rate_limited' }, { retryAfterS: 60 });
|
||||
try {
|
||||
const { value, stale } = await acquire();
|
||||
return json(200, stale ? { ...value, stale: true } : value, { stale });
|
||||
} catch (error) {
|
||||
return json(
|
||||
error.status === 429 ? 429 : 502,
|
||||
{ error: 'acartia_unavailable' },
|
||||
{ retryAfterS: error.retryAfterS },
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
name: 'acartia',
|
||||
configureServer({ middlewares }) {
|
||||
middlewares.use('/api/acartia', handler);
|
||||
},
|
||||
configurePreviewServer({ middlewares }) {
|
||||
middlewares.use('/api/acartia', handler);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,419 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { acartiaProxy, acartiaRetryCooldownMs } from './acartia.js';
|
||||
|
||||
const NOW = Date.parse('2026-11-04T18:05:00Z');
|
||||
const sighting = (id = 'synthetic-1') => ({
|
||||
entry_id: id,
|
||||
latitude: 48.12345,
|
||||
longitude: -123.12345,
|
||||
created: '2026-11-04 17:00:00',
|
||||
type: 'Orca',
|
||||
no_sighted: '5',
|
||||
photo_url: 'https://provider.example/photos/synthetic.jpg',
|
||||
data_source_name: 'Synthetic Whale Network',
|
||||
data_source_entity: 'Community reports',
|
||||
trusted: true,
|
||||
data_source_comments: '[email protected] +1-555-000-0000',
|
||||
});
|
||||
|
||||
function install(options = {}, hook = 'configureServer') {
|
||||
let handler;
|
||||
const plugin = acartiaProxy({ now: () => NOW, warn() {}, ...options });
|
||||
assert.equal(plugin.name, 'acartia');
|
||||
assert.equal(typeof plugin.configureServer, 'function');
|
||||
assert.equal(typeof plugin.configurePreviewServer, 'function');
|
||||
plugin[hook]({
|
||||
middlewares: {
|
||||
use(mount, callback) {
|
||||
assert.equal(mount, '/api/acartia');
|
||||
handler = callback;
|
||||
},
|
||||
},
|
||||
});
|
||||
return async (
|
||||
url = '/sightings',
|
||||
{ method = 'GET', peer = 'local', headers = {} } = {},
|
||||
) => {
|
||||
const res = {
|
||||
headers: {},
|
||||
writeHead(status, headers) {
|
||||
this.status = status;
|
||||
this.headers = headers;
|
||||
},
|
||||
setHeader(key, value) {
|
||||
this.headers[key] = value;
|
||||
},
|
||||
end(value) {
|
||||
this.status ??= this.statusCode;
|
||||
this.body = JSON.parse(value);
|
||||
},
|
||||
};
|
||||
await handler(
|
||||
{ url, method, headers, socket: { remoteAddress: peer } },
|
||||
res,
|
||||
);
|
||||
return res;
|
||||
};
|
||||
}
|
||||
|
||||
test('cache stores only supported marine mammals inside the coverage box', async () => {
|
||||
const cache = new Map();
|
||||
const request = install({
|
||||
cache,
|
||||
fetchImpl: async () =>
|
||||
Response.json([
|
||||
sighting('orca'),
|
||||
{ ...sighting('seal'), type: 'Harbor seal' },
|
||||
{ ...sighting('unknown'), type: 'Unknown' },
|
||||
{ ...sighting('fish'), type: 'Whale shark' },
|
||||
{ ...sighting('outside'), latitude: 37.8 },
|
||||
{ ...sighting('rounded-edge'), latitude: 45.49999 },
|
||||
]),
|
||||
});
|
||||
const result = await request();
|
||||
assert.equal(result.status, 200);
|
||||
for (const rows of [result.body.rows, cache.get('sightings').value.rows]) {
|
||||
assert.deepEqual(
|
||||
rows.map((row) => [row.stableId, row.cetacean]),
|
||||
[
|
||||
['acartia:orca', true],
|
||||
['acartia:seal', false],
|
||||
],
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
for (const hook of ['configureServer', 'configurePreviewServer']) {
|
||||
test(`${hook}: documented array is normalized and coarsened before caching`, async () => {
|
||||
const cache = new Map();
|
||||
let calls = 0;
|
||||
const request = install(
|
||||
{
|
||||
cache,
|
||||
fetchImpl: async (url, options) => {
|
||||
calls += 1;
|
||||
assert.equal(url, 'https://acartia.io/api/v1/sightings/current');
|
||||
assert.equal(options.redirect, 'error');
|
||||
assert.ok(options.signal instanceof AbortSignal);
|
||||
assert.equal(options.headers.Accept, 'application/json');
|
||||
assert.equal(
|
||||
options.headers['User-Agent'],
|
||||
'gods-eye-view-wildlife-sightings-proxy/1.0 (+https://github.com/bilawalsidhu/gods-eye-view)',
|
||||
);
|
||||
return Response.json([sighting()]);
|
||||
},
|
||||
},
|
||||
hook,
|
||||
);
|
||||
assert.equal(calls, 0);
|
||||
assert.equal(cache.size, 0);
|
||||
const result = await request(
|
||||
'/sightings?url=https://attacker.example&latitude=1',
|
||||
);
|
||||
assert.equal(result.status, 200);
|
||||
assert.equal(result.headers['Cache-Control'], 'no-store');
|
||||
assert.equal(result.body.fetchedAt, NOW);
|
||||
assert.equal(result.body.rows[0].stableId, 'acartia:synthetic-1');
|
||||
assert.equal(result.body.rows[0].lat, 48.12);
|
||||
assert.equal(result.body.rows[0].lon, -123.12);
|
||||
assert.equal(result.body.rows[0].provider.name, 'Synthetic Whale Network');
|
||||
const retained = cache.get('sightings');
|
||||
assert.equal(retained.value.rows[0].lat, 48.12);
|
||||
assert.equal(retained.value.rows[0].lon, -123.12);
|
||||
for (const value of [result.body, [...cache.values()]])
|
||||
assert.doesNotMatch(
|
||||
JSON.stringify(value),
|
||||
/private-observer|555|data_source_comments|48\.12345|123\.12345/,
|
||||
);
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
}
|
||||
|
||||
test('gate, method and fixed route run before cache or upstream', async () => {
|
||||
const cache = new Map();
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
cache,
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return Response.json([sighting()]);
|
||||
},
|
||||
});
|
||||
for (let i = 0; i < 65; i += 1) {
|
||||
assert.equal(
|
||||
(
|
||||
await request('/sightings', {
|
||||
headers: {
|
||||
host: 'localhost:5173',
|
||||
origin: 'https://foreign.example',
|
||||
'sec-fetch-site': 'cross-site',
|
||||
},
|
||||
})
|
||||
).status,
|
||||
403,
|
||||
);
|
||||
}
|
||||
const method = await request('/sightings', { method: 'POST' });
|
||||
assert.equal(method.status, 405);
|
||||
assert.equal(method.headers.Allow, 'GET');
|
||||
for (const route of ['/', '/unknown', '/sightings/extra']) {
|
||||
const result = await request(route);
|
||||
assert.equal(result.status, 404);
|
||||
assert.equal(result.body.error, 'unknown_route');
|
||||
}
|
||||
assert.equal(calls, 0);
|
||||
assert.equal(cache.size, 0);
|
||||
const sameOrigin = await request('/sightings', {
|
||||
headers: {
|
||||
host: 'gev.example',
|
||||
origin: 'https://gev.example',
|
||||
'sec-fetch-site': 'same-origin',
|
||||
'x-forwarded-proto': 'https',
|
||||
'x-forwarded-for': '203.0.113.9',
|
||||
},
|
||||
});
|
||||
assert.equal(sameOrigin.status, 200);
|
||||
});
|
||||
|
||||
test('request 61 is locally limited even with a fresh cache', async () => {
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return Response.json([sighting()]);
|
||||
},
|
||||
});
|
||||
for (let i = 0; i < 60; i += 1) assert.equal((await request()).status, 200);
|
||||
const limited = await request();
|
||||
assert.equal(limited.status, 429);
|
||||
assert.equal(limited.body.error, 'rate_limited');
|
||||
assert.equal(limited.headers['Retry-After'], '60');
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
|
||||
test('one global snapshot single-flights and stays fresh for ten minutes', async () => {
|
||||
let at = NOW;
|
||||
let calls = 0;
|
||||
let release;
|
||||
const request = install({
|
||||
now: () => at,
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
if (calls === 1)
|
||||
await new Promise((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
return Response.json([sighting()]);
|
||||
},
|
||||
});
|
||||
const first = request();
|
||||
const concurrent = request();
|
||||
await new Promise(setImmediate);
|
||||
assert.equal(calls, 1);
|
||||
release();
|
||||
assert.deepEqual((await first).body, (await concurrent).body);
|
||||
at += 599_999;
|
||||
await request();
|
||||
assert.equal(calls, 1);
|
||||
at += 1;
|
||||
await request();
|
||||
assert.equal(calls, 2);
|
||||
});
|
||||
|
||||
test('a valid empty or fully filtered snapshot is a fresh success that clears old rows', async () => {
|
||||
for (const quiet of [
|
||||
[],
|
||||
[null],
|
||||
[{ ...sighting(), data_source_name: '' }],
|
||||
[{ ...sighting(), latitude: 37.8 }],
|
||||
[{ ...sighting(), type: 'Unknown' }],
|
||||
]) {
|
||||
let at = NOW;
|
||||
let calls = 0;
|
||||
let empty = false;
|
||||
const cache = new Map();
|
||||
const request = install({
|
||||
cache,
|
||||
now: () => at,
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return Response.json(empty ? quiet : [sighting()]);
|
||||
},
|
||||
});
|
||||
assert.equal((await request()).body.rows.length, 1);
|
||||
empty = true;
|
||||
at += 600_001;
|
||||
const result = await request();
|
||||
assert.equal(result.status, 200);
|
||||
assert.deepEqual(result.body, { fetchedAt: at, rows: [] });
|
||||
assert.equal(result.headers['X-Data-Stale'], undefined);
|
||||
assert.equal(result.headers['Retry-After'], undefined);
|
||||
assert.deepEqual(cache.get('sightings').value.rows, []);
|
||||
// The quiet feed keeps the normal ten-minute cadence, not a failure retry.
|
||||
at += 599_999;
|
||||
assert.equal((await request()).status, 200);
|
||||
assert.equal(calls, 2);
|
||||
at += 1;
|
||||
await request();
|
||||
assert.equal(calls, 3);
|
||||
const cold = install({ fetchImpl: async () => Response.json(quiet) });
|
||||
const coldResult = await cold();
|
||||
assert.equal(coldResult.status, 200);
|
||||
assert.deepEqual(coldResult.body, { fetchedAt: NOW, rows: [] });
|
||||
}
|
||||
});
|
||||
|
||||
test('an invented wrapped snapshot never replaces the cache', async () => {
|
||||
const bad = { results: [sighting()] };
|
||||
let at = NOW;
|
||||
let fail = false;
|
||||
const cache = new Map();
|
||||
const request = install({
|
||||
cache,
|
||||
now: () => at,
|
||||
fetchImpl: async () => Response.json(fail ? bad : [sighting()]),
|
||||
});
|
||||
const fresh = await request();
|
||||
const retained = cache.get('sightings');
|
||||
fail = true;
|
||||
at += 600_001;
|
||||
const stale = await request();
|
||||
assert.equal(stale.status, 200);
|
||||
assert.equal(stale.headers['X-Data-Stale'], 'true');
|
||||
assert.equal(stale.body.stale, true);
|
||||
assert.equal(stale.body.fetchedAt, fresh.body.fetchedAt);
|
||||
assert.equal(cache.get('sightings'), retained);
|
||||
const coldCache = new Map();
|
||||
const cold = install({
|
||||
cache: coldCache,
|
||||
fetchImpl: async () => Response.json(bad),
|
||||
});
|
||||
const result = await cold();
|
||||
assert.equal(result.status, 502);
|
||||
assert.equal(result.body.error, 'acartia_unavailable');
|
||||
assert.equal(result.headers['Retry-After'], '15');
|
||||
assert.equal(coldCache.size, 0);
|
||||
});
|
||||
|
||||
test('failure serves stale for at most 24 hours and never reports a cold success', async () => {
|
||||
let at = NOW;
|
||||
let fail = false;
|
||||
const request = install({
|
||||
now: () => at,
|
||||
fetchImpl: async () => {
|
||||
if (fail) throw new Error('private upstream detail');
|
||||
return Response.json([sighting()]);
|
||||
},
|
||||
});
|
||||
const first = await request();
|
||||
fail = true;
|
||||
at += 23 * 60 * 60_000 + 59 * 60_000;
|
||||
const stale = await request();
|
||||
assert.equal(stale.status, 200);
|
||||
assert.equal(stale.body.stale, true);
|
||||
assert.equal(stale.body.fetchedAt, first.body.fetchedAt);
|
||||
at += 2 * 60_000;
|
||||
const expired = await request();
|
||||
assert.equal(expired.status, 502);
|
||||
assert.deepEqual(expired.body, { error: 'acartia_unavailable' });
|
||||
});
|
||||
|
||||
test('upstream 429 honors Retry-After, caps it, and spends no requests while cooling down', async () => {
|
||||
assert.equal(acartiaRetryCooldownMs('2', NOW), 5000);
|
||||
assert.equal(acartiaRetryCooldownMs('999999', NOW), 3_600_000);
|
||||
assert.equal(
|
||||
acartiaRetryCooldownMs(new Date(NOW + 120_000).toUTCString(), NOW),
|
||||
120_000,
|
||||
);
|
||||
assert.equal(acartiaRetryCooldownMs(null, NOW), 60_000);
|
||||
let at = NOW;
|
||||
let calls = 0;
|
||||
const warnings = [];
|
||||
const request = install({
|
||||
now: () => at,
|
||||
warn: (message) => warnings.push(message),
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return new Response('limit', {
|
||||
status: 429,
|
||||
headers: { 'Retry-After': '999999' },
|
||||
});
|
||||
},
|
||||
});
|
||||
const first = await request();
|
||||
assert.equal(first.status, 429);
|
||||
assert.equal(first.body.error, 'acartia_unavailable');
|
||||
assert.equal(first.headers['Retry-After'], '3600');
|
||||
at += 5000;
|
||||
const cooling = await request();
|
||||
assert.equal(cooling.headers['Retry-After'], '3595');
|
||||
assert.equal(calls, 1);
|
||||
assert.equal(
|
||||
warnings.filter((message) => message.includes('capped')).length,
|
||||
1,
|
||||
);
|
||||
at += 3_600_000;
|
||||
await request();
|
||||
assert.equal(calls, 2);
|
||||
assert.equal(
|
||||
warnings.filter((message) => message.includes('capped')).length,
|
||||
1,
|
||||
);
|
||||
});
|
||||
|
||||
test('upstream 429 also serves a prior snapshot stale', async () => {
|
||||
let at = NOW;
|
||||
let fail = false;
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
now: () => at,
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return fail
|
||||
? new Response('limit', { status: 429 })
|
||||
: Response.json([sighting()]);
|
||||
},
|
||||
});
|
||||
const first = await request();
|
||||
fail = true;
|
||||
at += 600_001;
|
||||
const stale = await request();
|
||||
assert.equal(stale.status, 200);
|
||||
assert.equal(stale.body.stale, true);
|
||||
assert.equal(stale.body.fetchedAt, first.body.fetchedAt);
|
||||
at += 1000;
|
||||
assert.equal((await request()).body.stale, true);
|
||||
assert.equal(calls, 2);
|
||||
});
|
||||
|
||||
test('bad JSON, 5xx and capped bodies start a 15 second failure cooldown', async () => {
|
||||
for (const bad of [
|
||||
() => new Response('{malformed'),
|
||||
() => new Response('private upstream message', { status: 503 }),
|
||||
() =>
|
||||
new Response('{}', {
|
||||
headers: { 'Content-Length': String(2 * 1024 * 1024 + 1) },
|
||||
}),
|
||||
]) {
|
||||
let at = NOW;
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
now: () => at,
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return calls === 1 ? bad() : Response.json([sighting()]);
|
||||
},
|
||||
});
|
||||
const first = await request();
|
||||
assert.equal(first.status, 502);
|
||||
assert.deepEqual(first.body, { error: 'acartia_unavailable' });
|
||||
assert.equal(first.headers['Retry-After'], '15');
|
||||
at += 14_999;
|
||||
assert.equal((await request()).headers['Retry-After'], '1');
|
||||
assert.equal(calls, 1);
|
||||
at += 1;
|
||||
assert.equal((await request()).status, 200);
|
||||
assert.equal(calls, 2);
|
||||
}
|
||||
});
|
||||
@@ -0,0 +1,4 @@
|
||||
/** Identify each provider while keeping the project contact URL consistent. */
|
||||
export function providerUserAgent(name) {
|
||||
return `gods-eye-view-${name}-proxy/1.0 (+https://github.com/bilawalsidhu/gods-eye-view)`;
|
||||
}
|
||||
@@ -0,0 +1,370 @@
|
||||
import path from 'node:path';
|
||||
import { promises as fs } from 'node:fs';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { normalizeInatObservations } from '../../../src/layers/wildlifeSightings/records.js';
|
||||
import { readResponseJsonCapped } from '../common/http.js';
|
||||
import { clampInt } from '../common/query.js';
|
||||
import { providerUserAgent } from '../common/userAgent.js';
|
||||
import { projectInatGrid, validInatGridTile } from './grid.js';
|
||||
import { createInatQueue } from './queue.js';
|
||||
|
||||
const DAILY_BUDGET = 8000;
|
||||
const DAY_MS = 24 * 60 * 60_000;
|
||||
// One page is capped independently of upstream Content-Length.
|
||||
const MAX_RESPONSE_BYTES = 8 * 1024 * 1024;
|
||||
const TIMEOUT_MS = 20_000;
|
||||
const API_URL = 'https://api.inaturalist.org/v1';
|
||||
// Request only the fields used by the public observation projection.
|
||||
const OBSERVATIONS_URL = 'https://api.inaturalist.org/v2/observations';
|
||||
const OBSERVATION_FIELDS =
|
||||
'(id:!t,uri:!t,location:!t,geojson:all,geoprivacy:!t,taxon_geoprivacy:!t,' +
|
||||
'obscured:!t,quality_grade:!t,observed_on:!t,' +
|
||||
'time_observed_at:!t,faves_count:!t,' +
|
||||
'num_identification_agreements:!t,' +
|
||||
'photos:(url:!t,attribution:!t,license_code:!t),' +
|
||||
'taxon:(name:!t,preferred_common_name:!t,iconic_taxon_name:!t,' +
|
||||
'threatened:!t))';
|
||||
const COOLDOWN_MS = 60_000;
|
||||
const FAILURE_COOLDOWN_MS = 15_000;
|
||||
|
||||
/** Seconds and HTTP dates are accepted; a missing value earns one minute. */
|
||||
export function inatRetryCooldownMs(raw, nowMs) {
|
||||
const value = typeof raw === 'string' ? raw.trim() : '';
|
||||
if (value) {
|
||||
const seconds = /^\d+(?:\.\d+)?$/.test(value) ? Number(value) : NaN;
|
||||
const deadline =
|
||||
!Number.isFinite(seconds) && /[a-z]/i.test(value)
|
||||
? Date.parse(value)
|
||||
: NaN;
|
||||
const delay = Number.isFinite(seconds) ? seconds * 1000 : deadline - nowMs;
|
||||
if (Number.isFinite(delay) && delay >= 0) return delay;
|
||||
}
|
||||
return COOLDOWN_MS;
|
||||
}
|
||||
|
||||
function upstreamError(code, status, retryAfterS, upstreamStatus) {
|
||||
return Object.assign(new Error(code), {
|
||||
code,
|
||||
status,
|
||||
retryAfterS,
|
||||
upstreamStatus,
|
||||
});
|
||||
}
|
||||
|
||||
function observationFilters(params) {
|
||||
const query = new URLSearchParams({
|
||||
verifiable: 'true',
|
||||
captive: 'false',
|
||||
// d1 is an observation DATE, unlike creation-time created_d1.
|
||||
d1: params.observedD1,
|
||||
});
|
||||
if (params.taxa !== 'whales') query.set('photos', 'true');
|
||||
if (params.quality === 'research') query.set('quality_grade', 'research');
|
||||
if (params.taxa === 'whales') query.set('taxon_id', '152871');
|
||||
else if (['Aves', 'Mammalia', 'Insecta', 'Plantae'].includes(params.taxa))
|
||||
query.set('iconic_taxa', params.taxa);
|
||||
return query;
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared iNaturalist governor. Only request counters are persisted; observation
|
||||
* payloads are projected immediately and are never written to disk.
|
||||
*/
|
||||
export function createInatClient({
|
||||
fetchImpl = (...args) => globalThis.fetch(...args),
|
||||
now = () => Date.now(),
|
||||
cacheDir = path.join(process.cwd(), '.gev-cache', 'inaturalist'),
|
||||
warn = (...args) => console.warn(...args),
|
||||
queueOptions,
|
||||
} = {}) {
|
||||
const enqueue = createInatQueue(queueOptions);
|
||||
let budget = null;
|
||||
let budgetLoaded = false;
|
||||
let budgetTail = Promise.resolve();
|
||||
let cooldownUntil = 0;
|
||||
// An unavailable UTFGrid tile must never place observation queries into a
|
||||
// 502 cooldown. Only an upstream 429 is a provider-wide restriction.
|
||||
const failures = new Map();
|
||||
|
||||
function dailyLimit() {
|
||||
return clampInt(
|
||||
process.env.INATURALIST_DAILY_BUDGET,
|
||||
1,
|
||||
Number.MAX_SAFE_INTEGER,
|
||||
DAILY_BUDGET,
|
||||
);
|
||||
}
|
||||
|
||||
function day() {
|
||||
return new Date(now()).toISOString().slice(0, 10);
|
||||
}
|
||||
|
||||
function currentBudget() {
|
||||
const date = day();
|
||||
if (budget?.date !== date) budget = { date, count: 0 };
|
||||
return budget;
|
||||
}
|
||||
|
||||
async function readBudgetOnce() {
|
||||
if (budgetLoaded) return;
|
||||
budgetLoaded = true;
|
||||
if (cacheDir === null) return;
|
||||
try {
|
||||
const value = JSON.parse(
|
||||
await fs.readFile(path.join(cacheDir, 'budget.json'), 'utf8'),
|
||||
);
|
||||
if (
|
||||
typeof value?.date === 'string' &&
|
||||
/^\d{4}-\d{2}-\d{2}$/.test(value.date) &&
|
||||
Number.isSafeInteger(value.count) &&
|
||||
value.count >= 0
|
||||
)
|
||||
budget = { date: value.date, count: value.count };
|
||||
} catch (error) {
|
||||
if (error.code !== 'ENOENT')
|
||||
warn('[inaturalist-proxy] budget read failed');
|
||||
}
|
||||
}
|
||||
|
||||
async function persistBudget(value) {
|
||||
if (cacheDir === null) return;
|
||||
const destination = path.join(cacheDir, 'budget.json');
|
||||
const temporary = `${destination}.${randomUUID()}.tmp`;
|
||||
try {
|
||||
await fs.mkdir(cacheDir, { recursive: true });
|
||||
await fs.writeFile(temporary, JSON.stringify(value), 'utf8');
|
||||
await fs.rename(temporary, destination);
|
||||
} catch {
|
||||
await fs.rm(temporary, { force: true }).catch(() => {});
|
||||
warn('[inaturalist-proxy] budget write failed');
|
||||
}
|
||||
}
|
||||
|
||||
function retrySeconds(until = cooldownUntil) {
|
||||
return Math.max(0, Math.ceil((until - now()) / 1000));
|
||||
}
|
||||
|
||||
function checkCooldown(resource) {
|
||||
if (now() < cooldownUntil)
|
||||
throw upstreamError('inaturalist_rate_limited', 429, retrySeconds(), 429);
|
||||
const failure = failures.get(resource);
|
||||
if (failure && now() < failure.until)
|
||||
throw upstreamError(
|
||||
'inaturalist_unavailable',
|
||||
502,
|
||||
retrySeconds(failure.until),
|
||||
failure.status,
|
||||
);
|
||||
failures.delete(resource);
|
||||
}
|
||||
|
||||
// Reserve and atomically persist one attempt before issuing it. A recovered
|
||||
// tail prevents one disk/reservation failure from poisoning future calls.
|
||||
function reserveAttempt(signal, resource) {
|
||||
const result = budgetTail.then(async () => {
|
||||
signal?.throwIfAborted();
|
||||
checkCooldown(resource);
|
||||
await readBudgetOnce();
|
||||
signal?.throwIfAborted();
|
||||
checkCooldown(resource);
|
||||
const value = currentBudget();
|
||||
if (value.count >= dailyLimit()) {
|
||||
const retry = Math.max(
|
||||
1,
|
||||
Math.ceil((Date.parse(`${day()}T00:00:00Z`) + DAY_MS - now()) / 1000),
|
||||
);
|
||||
throw upstreamError('inaturalist_budget_exhausted', 503, retry);
|
||||
}
|
||||
value.count += 1;
|
||||
await persistBudget({ ...value });
|
||||
});
|
||||
budgetTail = result.catch(() => {});
|
||||
return result;
|
||||
}
|
||||
|
||||
async function requestProjected(url, project, resource, { signal } = {}) {
|
||||
signal?.throwIfAborted();
|
||||
checkCooldown(resource);
|
||||
return enqueue(() => dispatch(url, project, resource, signal), {
|
||||
signal,
|
||||
priority: resource === 'observations' ? 0 : 1,
|
||||
});
|
||||
}
|
||||
|
||||
async function dispatch(url, project, resource, signal) {
|
||||
signal?.throwIfAborted();
|
||||
await reserveAttempt(signal, resource);
|
||||
signal?.throwIfAborted();
|
||||
checkCooldown(resource);
|
||||
const timeout = AbortSignal.timeout(TIMEOUT_MS);
|
||||
const requestSignal = signal ? AbortSignal.any([signal, timeout]) : timeout;
|
||||
let upstreamStatus;
|
||||
let retryDelay = FAILURE_COOLDOWN_MS;
|
||||
try {
|
||||
const response = await fetchImpl(url, {
|
||||
signal: requestSignal,
|
||||
redirect: 'error',
|
||||
headers: {
|
||||
Accept: 'application/json',
|
||||
'User-Agent': providerUserAgent('wildlife-sightings'),
|
||||
},
|
||||
});
|
||||
upstreamStatus = response.status;
|
||||
if (!response.ok) {
|
||||
await response.body?.cancel().catch(() => {});
|
||||
if (response.status === 429) {
|
||||
const delay = inatRetryCooldownMs(
|
||||
response.headers.get('retry-after'),
|
||||
now(),
|
||||
);
|
||||
cooldownUntil = Math.max(cooldownUntil, now() + delay);
|
||||
throw upstreamError(
|
||||
'inaturalist_rate_limited',
|
||||
429,
|
||||
retrySeconds(),
|
||||
429,
|
||||
);
|
||||
}
|
||||
if (response.headers.has('retry-after'))
|
||||
retryDelay = inatRetryCooldownMs(
|
||||
response.headers.get('retry-after'),
|
||||
now(),
|
||||
);
|
||||
throw upstreamError('inaturalist_unavailable', 502);
|
||||
}
|
||||
const payload = await readResponseJsonCapped(
|
||||
response,
|
||||
MAX_RESPONSE_BYTES,
|
||||
requestSignal,
|
||||
);
|
||||
return project(payload);
|
||||
} catch (error) {
|
||||
// Log only safe diagnostic categories, never a URL, body, status text or
|
||||
// exception message (each can contain request/private information).
|
||||
const reason = signal?.aborted
|
||||
? 'abort'
|
||||
: timeout.aborted || error.name === 'TimeoutError'
|
||||
? 'timeout'
|
||||
: upstreamStatus && upstreamStatus >= 400
|
||||
? 'http'
|
||||
: error.code === 'RESPONSE_TOO_LARGE'
|
||||
? 'response-too-large'
|
||||
: upstreamStatus
|
||||
? 'invalid-payload'
|
||||
: 'network';
|
||||
warn(
|
||||
`[inaturalist-proxy] upstream refresh failed resource=${resource} status=${upstreamStatus ?? 'none'} reason=${reason}`,
|
||||
);
|
||||
if (signal?.aborted) throw error;
|
||||
// Network, abort, schema and size failures are not upstream cooldown
|
||||
// signals. A later healthy request may recover immediately.
|
||||
if (upstreamStatus >= 500)
|
||||
failures.set(resource, {
|
||||
until: now() + retryDelay,
|
||||
status: upstreamStatus,
|
||||
});
|
||||
const failure = failures.get(resource);
|
||||
throw upstreamError(
|
||||
error.status === 429
|
||||
? 'inaturalist_rate_limited'
|
||||
: 'inaturalist_unavailable',
|
||||
error.status === 429 ? 429 : 502,
|
||||
error.status === 429
|
||||
? retrySeconds()
|
||||
: failure
|
||||
? retrySeconds(failure.until)
|
||||
: undefined,
|
||||
upstreamStatus === 429 || upstreamStatus >= 500
|
||||
? upstreamStatus
|
||||
: undefined,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/** Request one bounded observation sample and project its public fields. */
|
||||
async function getObservations(params, options = {}) {
|
||||
const query = observationFilters(params);
|
||||
// View samples stay stable in the route cache until their TTL expires.
|
||||
query.set('order_by', params.scope === 'view' ? 'random' : 'observed_on');
|
||||
if (params.scope !== 'view') query.set('order', 'desc');
|
||||
query.set('per_page', '200');
|
||||
query.set('fields', OBSERVATION_FIELDS);
|
||||
if (params.scope === 'view') {
|
||||
query.set('swlat', String(params.box.south));
|
||||
query.set('swlng', String(params.box.west));
|
||||
query.set('nelat', String(params.box.north));
|
||||
query.set('nelng', String(params.box.east));
|
||||
}
|
||||
return requestProjected(
|
||||
`${OBSERVATIONS_URL}?${query}`,
|
||||
(payload) => {
|
||||
if (
|
||||
!Array.isArray(payload?.results) ||
|
||||
!Number.isSafeInteger(payload.total_results) ||
|
||||
payload.total_results < 0
|
||||
)
|
||||
throw upstreamError('inaturalist_unavailable', 502);
|
||||
const rows = normalizeInatObservations(payload.results, {
|
||||
nowMs: now(),
|
||||
windowMs: (params.taxa === 'whales' ? 7 : 2) * DAY_MS,
|
||||
});
|
||||
// Empty views need total_results: 0. A wholly invalid page must not
|
||||
// replace good data. Valid rows excluded by the finer rolling-time
|
||||
// cutoff are a healthy empty result of the upstream DATE over-fetch.
|
||||
if (
|
||||
!rows ||
|
||||
(!rows.length &&
|
||||
(payload.results.length || payload.total_results) &&
|
||||
!normalizeInatObservations(payload.results)?.length)
|
||||
)
|
||||
throw upstreamError('inaturalist_unavailable', 502);
|
||||
return {
|
||||
rows,
|
||||
totalResults: payload.total_results,
|
||||
...(payload.total_results > payload.results.length
|
||||
? { partial: true, rawRowCount: payload.total_results }
|
||||
: {}),
|
||||
};
|
||||
},
|
||||
'observations',
|
||||
options,
|
||||
);
|
||||
}
|
||||
|
||||
/** Decode one validated density tile without observation identifiers. */
|
||||
async function getGrid(params, options = {}) {
|
||||
if (!validInatGridTile(params)) throw new TypeError('Invalid grid tile');
|
||||
const query = observationFilters(params);
|
||||
return requestProjected(
|
||||
`${API_URL}/grid/${params.z}/${params.x}/${params.y}.grid.json?${query}`,
|
||||
(payload) => projectInatGrid(payload, params),
|
||||
'grid',
|
||||
options,
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
getObservations,
|
||||
getGrid,
|
||||
/** Current provider-wide or resource-specific retry information. */
|
||||
cooldownState(resource) {
|
||||
if (now() < cooldownUntil)
|
||||
return {
|
||||
upstreamStatus: 429,
|
||||
retryAfterS: retrySeconds(),
|
||||
};
|
||||
const failure = failures.get(resource);
|
||||
if (failure && now() < failure.until)
|
||||
return {
|
||||
upstreamStatus: failure.status,
|
||||
retryAfterS: retrySeconds(failure.until),
|
||||
};
|
||||
return undefined;
|
||||
},
|
||||
/** The current UTC day's reserved upstream attempts and configured limit. */
|
||||
budgetState() {
|
||||
const value = currentBudget();
|
||||
return { ...value, budget: dailyLimit() };
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createInatClient } from './client.js';
|
||||
import { syntheticGrid } from './grid.fixture.mjs';
|
||||
|
||||
const NOW = Date.parse('2026-11-04T18:05:00Z');
|
||||
const observation = (id, time) => ({
|
||||
id,
|
||||
geojson: { coordinates: [-122.3, 47.6] },
|
||||
quality_grade: 'research',
|
||||
time_observed_at: time,
|
||||
observed_on: time.slice(0, 10),
|
||||
// Upload age does not determine whether an observation is recent.
|
||||
created_at: '2026-11-04T18:00:00Z',
|
||||
taxon: { name: 'Synthetic taxon', iconic_taxon_name: 'Aves' },
|
||||
});
|
||||
|
||||
function clientFor(results) {
|
||||
return createInatClient({
|
||||
now: () => NOW,
|
||||
cacheDir: null,
|
||||
queueOptions: { intervalMs: 0 },
|
||||
warn() {},
|
||||
fetchImpl: async () =>
|
||||
Response.json({ results, total_results: results.length }),
|
||||
});
|
||||
}
|
||||
|
||||
test('iNaturalist drops ancient and boundary-expired observations uploaded today', async () => {
|
||||
const client = clientFor([
|
||||
observation(1, '2010-11-04T18:00:00Z'),
|
||||
observation(2, '2026-11-02T18:04:59Z'),
|
||||
observation(3, '2026-11-02T18:05:00Z'),
|
||||
]);
|
||||
const result = await client.getObservations({
|
||||
scope: 'global',
|
||||
taxa: 'all',
|
||||
observedD1: '2026-11-02',
|
||||
});
|
||||
assert.deepEqual(
|
||||
result.rows.map((row) => row.stableId),
|
||||
['inat:3'],
|
||||
);
|
||||
});
|
||||
|
||||
test('a valid date-filtered page can be empty after the exact recency cutoff', async () => {
|
||||
const client = clientFor([observation(1, '2026-11-02T06:00:00Z')]);
|
||||
const result = await client.getObservations({
|
||||
scope: 'global',
|
||||
taxa: 'all',
|
||||
observedD1: '2026-11-02',
|
||||
});
|
||||
assert.deepEqual(result.rows, []);
|
||||
assert.equal(result.totalResults, 1);
|
||||
});
|
||||
|
||||
const params = {
|
||||
scope: 'view',
|
||||
box: { south: 47, west: -123, north: 48, east: -122 },
|
||||
taxa: 'all',
|
||||
observedD1: '2026-11-02',
|
||||
};
|
||||
|
||||
test('observations and tiles share pacing, and observation priority wins a cold burst', async () => {
|
||||
let at = 0;
|
||||
const calls = [];
|
||||
const client = createInatClient({
|
||||
cacheDir: null,
|
||||
now: () => NOW + at,
|
||||
queueOptions: {
|
||||
now: () => at,
|
||||
wait: async (ms) => {
|
||||
at += ms;
|
||||
},
|
||||
},
|
||||
fetchImpl: async (url) => {
|
||||
const isGrid = new URL(url).pathname.includes('/grid/');
|
||||
calls.push([isGrid ? 'grid' : 'observations', at]);
|
||||
return Response.json(
|
||||
isGrid ? syntheticGrid() : { results: [], total_results: 0 },
|
||||
);
|
||||
},
|
||||
});
|
||||
const tiles = Array.from({ length: 3 }, (_, x) =>
|
||||
client.getGrid({ ...params, z: 2, x, y: 1 }),
|
||||
);
|
||||
await Promise.all([...tiles, client.getObservations(params)]);
|
||||
assert.deepEqual(calls, [
|
||||
['observations', 0],
|
||||
['grid', 0],
|
||||
['grid', 0],
|
||||
['grid', 0],
|
||||
]);
|
||||
assert.equal(client.budgetState().count, 4);
|
||||
});
|
||||
|
||||
test('upstream status and safe failure reasons are logged without upstream messages or URLs', async () => {
|
||||
for (const [response, expected] of [
|
||||
[
|
||||
() =>
|
||||
new Response('request-secret', {
|
||||
status: 503,
|
||||
statusText: 'request-secret',
|
||||
}),
|
||||
/status=503 reason=http/,
|
||||
],
|
||||
[
|
||||
() => Promise.reject(new DOMException('request-secret', 'TimeoutError')),
|
||||
/status=none reason=timeout/,
|
||||
],
|
||||
[
|
||||
() => Response.json({ unexpected: 'request-secret' }),
|
||||
/status=200 reason=invalid-payload/,
|
||||
],
|
||||
]) {
|
||||
const warnings = [];
|
||||
const client = createInatClient({
|
||||
cacheDir: null,
|
||||
now: () => NOW,
|
||||
queueOptions: { intervalMs: 0 },
|
||||
warn: (message) => warnings.push(message),
|
||||
fetchImpl: response,
|
||||
});
|
||||
await assert.rejects(client.getObservations(params), { status: 502 });
|
||||
assert.equal(warnings.length, 1);
|
||||
assert.match(warnings[0], expected);
|
||||
assert.doesNotMatch(warnings[0], /request-secret|https?:|swlat|d1=/);
|
||||
}
|
||||
});
|
||||
|
||||
test('a dispatched abort is diagnosed, spends only its attempt, and leaves the next request healthy', async () => {
|
||||
const warnings = [];
|
||||
const controller = new AbortController();
|
||||
let started;
|
||||
const didStart = new Promise((resolve) => {
|
||||
started = resolve;
|
||||
});
|
||||
let calls = 0;
|
||||
const client = createInatClient({
|
||||
cacheDir: null,
|
||||
now: () => NOW,
|
||||
queueOptions: { intervalMs: 0 },
|
||||
warn: (message) => warnings.push(message),
|
||||
fetchImpl: async (url, { signal }) => {
|
||||
calls += 1;
|
||||
if (calls > 1) return Response.json({ results: [], total_results: 0 });
|
||||
started();
|
||||
return new Promise((resolve, reject) =>
|
||||
signal.addEventListener('abort', () => reject(signal.reason), {
|
||||
once: true,
|
||||
}),
|
||||
);
|
||||
},
|
||||
});
|
||||
const pending = client.getObservations(params, { signal: controller.signal });
|
||||
const failed = assert.rejects(pending, { name: 'AbortError' });
|
||||
await didStart;
|
||||
controller.abort();
|
||||
await failed;
|
||||
assert.match(warnings[0], /resource=observations status=none reason=abort/);
|
||||
await client.getObservations(params);
|
||||
assert.equal(calls, 2);
|
||||
});
|
||||
@@ -0,0 +1,51 @@
|
||||
/** Build a synthetic UTFGrid with logical cell counts and repeated hit pixels. */
|
||||
export function syntheticGrid({
|
||||
z = 2,
|
||||
x = 1,
|
||||
y = 1,
|
||||
cells = [
|
||||
{ row: 1, column: 51, count: 6, iconic: 3 },
|
||||
{ row: 3, column: 49, count: 12, iconic: 47170 },
|
||||
],
|
||||
} = {}) {
|
||||
const pixels = Array.from({ length: 64 }, () => Array(64).fill(' '));
|
||||
const keys = [''];
|
||||
const data = {};
|
||||
cells.forEach((cell, index) => {
|
||||
const key = String(900_000_000 + index);
|
||||
keys.push(key);
|
||||
const value = index + 1;
|
||||
let code = value + 32;
|
||||
if (code >= 34) code += 1;
|
||||
if (code >= 92) code += 1;
|
||||
const character = String.fromCharCode(code);
|
||||
for (const [row, column] of [
|
||||
[cell.row, cell.column],
|
||||
[cell.row - 1, cell.column],
|
||||
[cell.row + 1, cell.column],
|
||||
[cell.row, cell.column - 1],
|
||||
[cell.row, cell.column + 1],
|
||||
]) {
|
||||
if (row >= 0 && row < 64 && column >= 0 && column < 64)
|
||||
pixels[row][column] = character;
|
||||
}
|
||||
data[key] = {
|
||||
captive: null,
|
||||
cellCount: cell.count,
|
||||
geoprivacy: null,
|
||||
id: Number(key),
|
||||
latitude:
|
||||
(Math.atan(
|
||||
Math.sinh(Math.PI * (1 - 2 * ((y + cell.row / 64) / 2 ** z))),
|
||||
) *
|
||||
180) /
|
||||
Math.PI,
|
||||
longitude: ((x + cell.column / 64) / 2 ** z) * 360 - 180,
|
||||
private_location: null,
|
||||
quality_grade: 'needs_id',
|
||||
'taxon.iconic_taxon_id': cell.iconic,
|
||||
uuid: `00000000-0000-4000-8000-${String(index).padStart(12, '0')}`,
|
||||
};
|
||||
});
|
||||
return { grid: pixels.map((row) => row.join('')), keys, data };
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
// iNaturalist's operational /v1/grid/{z}/{x}/{y}.grid.json endpoint is not
|
||||
// covered by its v1 Swagger contract. Validate the entire UTFGrid shape before
|
||||
// accepting any cell; callers preserve the last good sample on schema drift.
|
||||
import {
|
||||
wildlifeGridCellId,
|
||||
WILDLIFE_GRID_MAX_ZOOM,
|
||||
} from '../../../src/layers/wildlifeSightings/gridViewport.js';
|
||||
|
||||
const INAT_GRID_SIZE = 64;
|
||||
export const INAT_GRID_DECODE_MEASURE = 'wildlife:server:tile-decode';
|
||||
const grades = new Set(['research', 'needs_id', 'casual']);
|
||||
const object = (value) =>
|
||||
!!value && typeof value === 'object' && !Array.isArray(value);
|
||||
|
||||
/** Accept only integer XYZ coordinates within the supported zoom range. */
|
||||
export function validInatGridTile({ z, x, y } = {}) {
|
||||
return (
|
||||
Number.isInteger(z) &&
|
||||
z >= 0 &&
|
||||
z <= WILDLIFE_GRID_MAX_ZOOM &&
|
||||
Number.isInteger(x) &&
|
||||
Number.isInteger(y) &&
|
||||
x >= 0 &&
|
||||
y >= 0 &&
|
||||
x < 2 ** z &&
|
||||
y < 2 ** z
|
||||
);
|
||||
}
|
||||
|
||||
/** Pure allow-list projection. A UTFGrid key is an observation identifier and
|
||||
* its repeated pixels form a dilated hit area, not additional occupied cells.
|
||||
* Neither those keys nor representative/private fields cross this boundary. */
|
||||
export function projectInatGrid(payload, tile) {
|
||||
const start = `${INAT_GRID_DECODE_MEASURE}:start`;
|
||||
const end = `${INAT_GRID_DECODE_MEASURE}:end`;
|
||||
performance.mark(start);
|
||||
try {
|
||||
return decodeInatGrid(payload, tile);
|
||||
} finally {
|
||||
performance.mark(end);
|
||||
// Keep the latest sample inspectable without retaining every tile forever.
|
||||
// PerformanceObservers still receive each individual decode measurement.
|
||||
performance.clearMeasures(INAT_GRID_DECODE_MEASURE);
|
||||
performance.measure(INAT_GRID_DECODE_MEASURE, start, end);
|
||||
performance.clearMarks(start);
|
||||
performance.clearMarks(end);
|
||||
}
|
||||
}
|
||||
|
||||
function decodeInatGrid(payload, tile) {
|
||||
const invalid = () => {
|
||||
throw new TypeError('Invalid iNaturalist UTFGrid');
|
||||
};
|
||||
if (
|
||||
!validInatGridTile(tile) ||
|
||||
!object(payload) ||
|
||||
!Array.isArray(payload.grid) ||
|
||||
payload.grid.length !== INAT_GRID_SIZE ||
|
||||
!payload.grid.every(
|
||||
(row) => typeof row === 'string' && row.length === INAT_GRID_SIZE,
|
||||
) ||
|
||||
!Array.isArray(payload.keys) ||
|
||||
!payload.keys.length ||
|
||||
payload.keys.length > INAT_GRID_SIZE ** 2 + 1 ||
|
||||
payload.keys[0] !== '' ||
|
||||
!payload.keys.every(
|
||||
(key) => typeof key === 'string' && key.length <= 128,
|
||||
) ||
|
||||
new Set(payload.keys).size !== payload.keys.length ||
|
||||
!object(payload.data)
|
||||
)
|
||||
invalid();
|
||||
const dataKeys = Object.keys(payload.data);
|
||||
if (dataKeys.length !== payload.keys.length - 1) invalid();
|
||||
for (const key of payload.keys.slice(1)) {
|
||||
if (!key || !Object.hasOwn(payload.data, key)) invalid();
|
||||
const value = payload.data[key];
|
||||
if (
|
||||
!object(value) ||
|
||||
!Number.isSafeInteger(value.cellCount) ||
|
||||
value.cellCount <= 0 ||
|
||||
!Number.isFinite(value.latitude) ||
|
||||
Math.abs(value.latitude) > 90 ||
|
||||
!Number.isFinite(value.longitude) ||
|
||||
Math.abs(value.longitude) > 180 ||
|
||||
!grades.has(value.quality_grade) ||
|
||||
!(
|
||||
value['taxon.iconic_taxon_id'] === null ||
|
||||
(Number.isSafeInteger(value['taxon.iconic_taxon_id']) &&
|
||||
value['taxon.iconic_taxon_id'] > 0)
|
||||
)
|
||||
)
|
||||
invalid();
|
||||
}
|
||||
const occupied = new Set();
|
||||
for (const line of payload.grid) {
|
||||
for (let column = 0; column < INAT_GRID_SIZE; column += 1) {
|
||||
let code = line.charCodeAt(column);
|
||||
if (code < 32 || code === 34 || code === 92) invalid();
|
||||
if (code >= 93) code -= 1;
|
||||
if (code >= 35) code -= 1;
|
||||
const index = code - 32;
|
||||
if (index < 0 || index >= payload.keys.length) invalid();
|
||||
if (index) occupied.add(payload.keys[index]);
|
||||
}
|
||||
}
|
||||
if (occupied.size !== dataKeys.length) invalid();
|
||||
const cells = new Map();
|
||||
for (const key of occupied) {
|
||||
const value = payload.data[key];
|
||||
const id = wildlifeGridCellId(value.latitude, value.longitude, tile.z);
|
||||
const cell = {
|
||||
id,
|
||||
cellCount: value.cellCount,
|
||||
lat: value.latitude,
|
||||
lon: value.longitude,
|
||||
};
|
||||
// Two different representatives claiming one logical cell are a contract
|
||||
// change; summing or choosing one silently would fabricate the count.
|
||||
if (cells.has(id)) invalid();
|
||||
cells.set(id, cell);
|
||||
}
|
||||
const rows = [...cells.values()];
|
||||
const totalResults = rows.reduce((total, cell) => total + cell.cellCount, 0);
|
||||
if (!Number.isSafeInteger(totalResults)) invalid();
|
||||
return { cells: rows, totalResults };
|
||||
}
|
||||
@@ -0,0 +1,374 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { projectInatGrid } from './grid.js';
|
||||
import { syntheticGrid } from './grid.fixture.mjs';
|
||||
import { installInatProxy, NOW } from './proxy.fixture.mjs';
|
||||
import { createInatClient } from './client.js';
|
||||
|
||||
const tile = { z: 2, x: 1, y: 1 };
|
||||
const gridUrl = '/grid?z=2&x=1&y=1&taxa=Aves&quality=research';
|
||||
const install = (options, hook) => installInatProxy(gridUrl, options, hook);
|
||||
|
||||
test('UTFGrid projection validates the full shape and counts each logical occupied cell once', () => {
|
||||
const raw = syntheticGrid();
|
||||
const projected = projectInatGrid(raw, tile);
|
||||
assert.equal(raw.grid.length, 64);
|
||||
assert.ok(raw.grid.every((row) => row.length === 64));
|
||||
assert.equal(
|
||||
raw.grid.join('').split('!').length - 1,
|
||||
5,
|
||||
'one key is repeated by its hit area',
|
||||
);
|
||||
assert.equal(projected.cells.length, 2);
|
||||
assert.equal(projected.totalResults, 18);
|
||||
assert.deepEqual(projected.cells[0], {
|
||||
id: 'grid:2/1/1:1:51',
|
||||
cellCount: 6,
|
||||
lat: raw.data[raw.keys[1]].latitude,
|
||||
lon: raw.data[raw.keys[1]].longitude,
|
||||
});
|
||||
for (const cell of projected.cells)
|
||||
assert.deepEqual(Object.keys(cell).sort(), [
|
||||
'cellCount',
|
||||
'id',
|
||||
'lat',
|
||||
'lon',
|
||||
]);
|
||||
for (const privateField of [
|
||||
'900000000',
|
||||
'uuid',
|
||||
'geoprivacy',
|
||||
'private_location',
|
||||
'captive',
|
||||
'quality_grade',
|
||||
])
|
||||
assert.ok(!JSON.stringify(projected).includes(privateField));
|
||||
});
|
||||
|
||||
test('legitimate blank grid and null/unknown iconic ids remain safe', () => {
|
||||
assert.deepEqual(projectInatGrid(syntheticGrid({ cells: [] }), tile), {
|
||||
cells: [],
|
||||
totalResults: 0,
|
||||
});
|
||||
const raw = syntheticGrid({
|
||||
cells: [{ row: 1, column: 1, count: 1, iconic: null }],
|
||||
});
|
||||
const projected = projectInatGrid(raw, tile);
|
||||
assert.equal(projected.cells[0].cellCount, 1);
|
||||
raw.data[raw.keys[1]]['taxon.iconic_taxon_id'] = 999999;
|
||||
assert.deepEqual(projectInatGrid(raw, tile), projected);
|
||||
});
|
||||
|
||||
test('schema drift anywhere fails closed, including unused, malformed and orphan cells', () => {
|
||||
const changes = [
|
||||
(raw) => {
|
||||
raw.grid.pop();
|
||||
},
|
||||
(raw) => {
|
||||
raw.grid[0] = '';
|
||||
},
|
||||
(raw) => {
|
||||
raw.grid[0] = '~'.repeat(64);
|
||||
},
|
||||
(raw) => {
|
||||
raw.keys[0] = 'not-empty';
|
||||
},
|
||||
(raw) => {
|
||||
raw.keys.push(raw.keys[1]);
|
||||
},
|
||||
(raw) => {
|
||||
delete raw.data[raw.keys[1]];
|
||||
},
|
||||
(raw) => {
|
||||
raw.data.extra = {};
|
||||
},
|
||||
(raw) => {
|
||||
raw.data[raw.keys[1]].cellCount = -1;
|
||||
},
|
||||
(raw) => {
|
||||
raw.data[raw.keys[1]].latitude = '47';
|
||||
},
|
||||
(raw) => {
|
||||
raw.data[raw.keys[1]].longitude = 181;
|
||||
},
|
||||
(raw) => {
|
||||
raw.data[raw.keys[1]].quality_grade = 'new-grade';
|
||||
},
|
||||
(raw) => {
|
||||
delete raw.data[raw.keys[1]]['taxon.iconic_taxon_id'];
|
||||
},
|
||||
(raw) => {
|
||||
raw.data[raw.keys[1]]['taxon.iconic_taxon_id'] = 'Aves';
|
||||
},
|
||||
(raw) => {
|
||||
raw.grid = Array(64).fill(' '.repeat(64));
|
||||
},
|
||||
];
|
||||
for (const change of changes) {
|
||||
const raw = syntheticGrid();
|
||||
change(raw);
|
||||
assert.throws(
|
||||
() => projectInatGrid(raw, tile),
|
||||
/Invalid iNaturalist UTFGrid/,
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
for (const hook of ['configureServer', 'configurePreviewServer']) {
|
||||
test(`${hook}: fixed grid URL shares filters, caps and allow-list cache`, async () => {
|
||||
const calls = [];
|
||||
const cache = new Map();
|
||||
const request = install(
|
||||
{
|
||||
cache,
|
||||
fetchImpl: async (url, options) => {
|
||||
calls.push(new URL(url));
|
||||
assert.equal(options.redirect, 'error');
|
||||
assert.ok(options.signal instanceof AbortSignal);
|
||||
assert.match(
|
||||
options.headers['User-Agent'],
|
||||
/^gods-eye-view-wildlife-sightings-proxy/,
|
||||
);
|
||||
return Response.json(syntheticGrid());
|
||||
},
|
||||
},
|
||||
hook,
|
||||
);
|
||||
assert.equal(calls.length, 0);
|
||||
const response = await request(
|
||||
`${gridUrl}&url=https://attacker.test&per_page=999`,
|
||||
);
|
||||
assert.equal(response.status, 200);
|
||||
assert.equal(response.body.cells.length, 2);
|
||||
assert.equal(response.body.fetchedAt, NOW);
|
||||
assert.equal(response.headers['Cache-Control'], 'no-store');
|
||||
assert.equal(
|
||||
calls[0].origin + calls[0].pathname,
|
||||
'https://api.inaturalist.org/v1/grid/2/1/1.grid.json',
|
||||
);
|
||||
assert.deepEqual(Object.fromEntries(calls[0].searchParams), {
|
||||
verifiable: 'true',
|
||||
captive: 'false',
|
||||
d1: '2026-11-02',
|
||||
photos: 'true',
|
||||
quality_grade: 'research',
|
||||
iconic_taxa: 'Aves',
|
||||
});
|
||||
await request(gridUrl);
|
||||
assert.equal(calls.length, 1, 'fresh stable tile/filter key is cached');
|
||||
assert.ok(!JSON.stringify([...cache]).includes('private_location'));
|
||||
assert.ok(!JSON.stringify([...cache]).includes('900000000'));
|
||||
const whale = await request('/grid?z=2&x=1&y=1&taxa=whales&confirmed=true');
|
||||
assert.equal(whale.status, 200);
|
||||
assert.equal(calls[1].searchParams.get('taxon_id'), '152871');
|
||||
assert.equal(calls[1].searchParams.get('d1'), '2026-10-28');
|
||||
assert.equal(calls[1].searchParams.has('photos'), false);
|
||||
});
|
||||
}
|
||||
|
||||
test('grid admission rejects cross-site/method/path/tile before acquisition', async () => {
|
||||
const request = install({
|
||||
fetchImpl() {
|
||||
assert.fail('invalid grid request acquired');
|
||||
},
|
||||
});
|
||||
assert.equal(
|
||||
(
|
||||
await request(gridUrl, {
|
||||
headers: {
|
||||
host: 'localhost:5173',
|
||||
origin: 'https://other.test',
|
||||
'sec-fetch-site': 'cross-site',
|
||||
},
|
||||
})
|
||||
).status,
|
||||
403,
|
||||
);
|
||||
assert.equal((await request('/grid', { method: 'POST' })).status, 405);
|
||||
assert.equal((await request('/grid/extra')).status, 404);
|
||||
for (const query of [
|
||||
'',
|
||||
'z=2&x=4&y=0',
|
||||
'z=-1&x=0&y=0',
|
||||
'z=23&x=0&y=0',
|
||||
'z=2&x=1.5&y=0',
|
||||
'z=2&x=&y=0',
|
||||
'z=2&x=0&y=4',
|
||||
]) {
|
||||
const result = await request(`/grid?${query}`);
|
||||
assert.equal(result.status, 400, query);
|
||||
assert.equal(result.body.error, 'invalid_tile');
|
||||
}
|
||||
});
|
||||
|
||||
test('grid refresh single-flight, stale fallback and hour-key expiry keep the original age', async () => {
|
||||
let time = NOW;
|
||||
let calls = 0;
|
||||
let resolve;
|
||||
let mode = 'pending';
|
||||
const request = install({
|
||||
now: () => time,
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
if (mode === 'pending')
|
||||
return new Promise((done) => {
|
||||
resolve = done;
|
||||
});
|
||||
if (mode === 'bad') return Response.json({ grid: [] });
|
||||
return Response.json(syntheticGrid());
|
||||
},
|
||||
});
|
||||
const first = request();
|
||||
const second = request();
|
||||
for (let i = 0; i < 100 && !resolve; i += 1) await new Promise(setImmediate);
|
||||
assert.equal(calls, 1);
|
||||
resolve(Response.json(syntheticGrid()));
|
||||
assert.equal((await first).status, 200);
|
||||
assert.equal((await second).status, 200);
|
||||
time += 900_001;
|
||||
mode = 'bad';
|
||||
const stale = await request();
|
||||
assert.equal(stale.status, 200);
|
||||
assert.equal(stale.body.stale, true);
|
||||
assert.equal(stale.body.fetchedAt, NOW);
|
||||
assert.equal(stale.headers['X-Data-Stale'], 'true');
|
||||
time = NOW + 3_600_001;
|
||||
const expired = await request();
|
||||
assert.equal(expired.status, 502);
|
||||
assert.equal(expired.body.error, 'inaturalist_unavailable');
|
||||
});
|
||||
|
||||
test('grid and observations share the upstream cooldown and daily reservations', async () => {
|
||||
let calls = 0;
|
||||
const client = createInatClient({
|
||||
now: () => NOW,
|
||||
cacheDir: null,
|
||||
queueOptions: { intervalMs: 0 },
|
||||
warn() {},
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return new Response('', {
|
||||
status: 429,
|
||||
headers: { 'Retry-After': '120' },
|
||||
});
|
||||
},
|
||||
});
|
||||
await assert.rejects(
|
||||
client.getGrid({ ...tile, taxa: 'all', observedD1: '2026-11-02' }),
|
||||
(error) => error.status === 429 && error.retryAfterS === 120,
|
||||
);
|
||||
await assert.rejects(
|
||||
client.getObservations({
|
||||
scope: 'global',
|
||||
taxa: 'all',
|
||||
observedD1: '2026-11-02',
|
||||
}),
|
||||
(error) => error.status === 429,
|
||||
);
|
||||
assert.equal(calls, 1);
|
||||
assert.equal(client.budgetState().count, 1);
|
||||
});
|
||||
|
||||
test('grid body cap and upstream failures return fixed errors; only real 5xx cools down', async () => {
|
||||
for (const [response, retryAfter] of [
|
||||
[() => new Response('x'.repeat(8 * 1024 * 1024 + 1)), undefined],
|
||||
[() => new Response('', { status: 503 }), '15'],
|
||||
]) {
|
||||
const request = install({ fetchImpl: async () => response() });
|
||||
const result = await request();
|
||||
assert.equal(result.status, 502);
|
||||
assert.deepEqual(result.body, { error: 'inaturalist_unavailable' });
|
||||
assert.equal(result.headers['Retry-After'], retryAfter);
|
||||
}
|
||||
});
|
||||
|
||||
test('a failed tile preserves last good cells and cannot send unrelated observations into 502', async () => {
|
||||
let at = NOW;
|
||||
let failGrid = false;
|
||||
const calls = [];
|
||||
const warnings = [];
|
||||
const request = install({
|
||||
now: () => at,
|
||||
warn: (message) => warnings.push(message),
|
||||
fetchImpl: async (url) => {
|
||||
const isGrid = new URL(url).pathname.includes('/grid/');
|
||||
calls.push(isGrid ? 'grid' : 'observations');
|
||||
if (isGrid && failGrid)
|
||||
return new Response('private upstream text', {
|
||||
status: 503,
|
||||
headers: { 'Retry-After': '45' },
|
||||
});
|
||||
return Response.json(
|
||||
isGrid ? syntheticGrid() : { results: [], total_results: 0 },
|
||||
);
|
||||
},
|
||||
});
|
||||
const first = await request();
|
||||
at += 900_001;
|
||||
failGrid = true;
|
||||
const failed = await request();
|
||||
assert.equal(failed.status, 200);
|
||||
assert.equal(failed.headers['X-Data-Stale'], 'true');
|
||||
assert.equal(failed.body.stale, true);
|
||||
assert.equal(failed.body.fetchedAt, first.body.fetchedAt);
|
||||
assert.deepEqual(failed.body.cells, first.body.cells);
|
||||
assert.equal(
|
||||
(await request('/observations?scope=global&taxa=whales')).status,
|
||||
200,
|
||||
);
|
||||
assert.deepEqual(calls, ['grid', 'grid', 'observations']);
|
||||
assert.match(warnings[0], /resource=grid status=503 reason=http/);
|
||||
assert.doesNotMatch(warnings[0], /private upstream text/);
|
||||
// The tile cooldown also preserves its data without another upstream try.
|
||||
const cooling = await request();
|
||||
assert.equal(cooling.body.stale, true);
|
||||
assert.equal(cooling.headers['Retry-After'], '45');
|
||||
assert.equal(cooling.headers['X-Upstream-Status'], '503');
|
||||
assert.equal(calls.length, 3);
|
||||
at += 45_000;
|
||||
failGrid = false;
|
||||
const refreshing = await request();
|
||||
assert.equal(refreshing.body.stale, true);
|
||||
await new Promise(setImmediate);
|
||||
const recovered = await request();
|
||||
assert.equal(recovered.status, 200);
|
||||
assert.equal(recovered.body.stale, undefined);
|
||||
assert.equal(
|
||||
calls.length,
|
||||
4,
|
||||
'a failed queue task does not poison later tile work',
|
||||
);
|
||||
});
|
||||
|
||||
test('twelve queued tiles leave observation capacity and give the next dispatch to observations', async () => {
|
||||
const calls = [];
|
||||
let release;
|
||||
const request = install({
|
||||
fetchImpl: (url) => {
|
||||
const path = new URL(url).pathname;
|
||||
calls.push(path);
|
||||
if (calls.length === 1)
|
||||
return new Promise((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
return Response.json(
|
||||
path.includes('/grid/')
|
||||
? syntheticGrid()
|
||||
: { results: [], total_results: 0 },
|
||||
);
|
||||
},
|
||||
});
|
||||
const first = request('/grid?z=4&x=0&y=1');
|
||||
for (let i = 0; i < 100 && !release; i += 1) await new Promise(setImmediate);
|
||||
assert.ok(release);
|
||||
const tiles = Array.from({ length: 11 }, (_, i) =>
|
||||
request(`/grid?z=4&x=${i + 1}&y=1`),
|
||||
);
|
||||
const observation = request('/observations?scope=global');
|
||||
const queuedTile = request('/grid?z=4&x=12&y=1');
|
||||
release(Response.json(syntheticGrid()));
|
||||
const results = await Promise.all([first, observation, ...tiles, queuedTile]);
|
||||
assert.ok(results.every((result) => result.status === 200));
|
||||
assert.equal(calls.length, 14);
|
||||
assert.equal(calls[1], '/v2/observations');
|
||||
});
|
||||
@@ -0,0 +1,343 @@
|
||||
import { createInatClient } from './client.js';
|
||||
import { coalesceProxyRequest } from '../common/http.js';
|
||||
import { admitSameSiteKeyless } from '../common/same-site.js';
|
||||
import { requiredFiniteQueryNumber } from '../common/query.js';
|
||||
import { snapWildlifeBox } from '../../../src/layers/wildlifeSightings/bounds.js';
|
||||
import { validInatGridTile } from './grid.js';
|
||||
|
||||
// The upstream advertises five-minute freshness; global samples cost less often.
|
||||
const TTL_MS = 300_000;
|
||||
const GLOBAL_TTL_MS = 600_000;
|
||||
const INAT_GRID_CACHE_TTL_MS = 15 * 60_000;
|
||||
const STALE_MAX_MS = 3_600_000;
|
||||
const INAT_CACHE_MAX = 400;
|
||||
const HOUR_MS = 3_600_000;
|
||||
const TAXA = Object.freeze([
|
||||
'all',
|
||||
'Aves',
|
||||
'Mammalia',
|
||||
'Insecta',
|
||||
'Plantae',
|
||||
'whales',
|
||||
]);
|
||||
|
||||
function requestParams(pathname, query, nowMs) {
|
||||
const grid = pathname === '/grid';
|
||||
const scope = grid ? 'grid' : (query.get('scope') ?? 'view');
|
||||
if (!grid && scope !== 'view' && scope !== 'global')
|
||||
return { error: 'invalid_scope' };
|
||||
const quality =
|
||||
query.get('quality') === 'research' ||
|
||||
(grid && query.get('confirmed') === 'true')
|
||||
? 'research'
|
||||
: 'all';
|
||||
const taxa = TAXA.includes(query.get('taxa')) ? query.get('taxa') : 'all';
|
||||
let box = null;
|
||||
let tile;
|
||||
if (grid) {
|
||||
tile = Object.fromEntries(
|
||||
['z', 'x', 'y'].map((name) => [
|
||||
name,
|
||||
requiredFiniteQueryNumber(query, name),
|
||||
]),
|
||||
);
|
||||
if (!validInatGridTile(tile)) return { error: 'invalid_tile' };
|
||||
} else if (scope === 'view') {
|
||||
const south = requiredFiniteQueryNumber(query, 'south');
|
||||
const west = requiredFiniteQueryNumber(query, 'west');
|
||||
const north = requiredFiniteQueryNumber(query, 'north');
|
||||
const east = requiredFiniteQueryNumber(query, 'east');
|
||||
box = snapWildlifeBox({ south, west, north, east });
|
||||
if (!box) return { error: 'invalid_bbox' };
|
||||
}
|
||||
const windowMs = (taxa === 'whales' ? 7 * 24 : 48) * HOUR_MS;
|
||||
const windowKey = new Date(Math.floor((nowMs - windowMs) / HOUR_MS) * HOUR_MS)
|
||||
.toISOString()
|
||||
.replace('.000Z', 'Z');
|
||||
const bounds = box ? `${box.south},${box.west},${box.north},${box.east}` : '';
|
||||
const baseKey = grid
|
||||
? `grid:v1:${tile.z}/${tile.x}/${tile.y}:${quality}:${taxa}`
|
||||
: `v1:${scope}:${bounds}:${quality}:${taxa}`;
|
||||
return {
|
||||
scope,
|
||||
...(grid
|
||||
? { ...tile, waitForFresh: query.get('refresh') === 'wait' }
|
||||
: { box }),
|
||||
quality,
|
||||
taxa,
|
||||
observedD1: windowKey.slice(0, 10),
|
||||
baseKey,
|
||||
key: `${baseKey}:${windowKey}`,
|
||||
};
|
||||
}
|
||||
|
||||
/** A shared-provider route for dev and preview; acquisition starts on demand. */
|
||||
export function inaturalistProxy({
|
||||
fetchImpl = (...args) => globalThis.fetch(...args),
|
||||
now = () => Date.now(),
|
||||
cacheDir,
|
||||
cache = new Map(),
|
||||
warn = (...args) => console.warn(...args),
|
||||
queueOptions,
|
||||
} = {}) {
|
||||
const inFlight = new Map();
|
||||
const consumers = new Map();
|
||||
let client = null;
|
||||
|
||||
function age(entry) {
|
||||
if (!entry) return Infinity;
|
||||
const elapsed = now() - entry.savedAt;
|
||||
return elapsed < 0 ? Infinity : elapsed;
|
||||
}
|
||||
|
||||
function joinFlight(promise, group, signal) {
|
||||
group.count += 1;
|
||||
return new Promise((resolve, reject) => {
|
||||
let released = false;
|
||||
function release() {
|
||||
if (released) return;
|
||||
released = true;
|
||||
signal?.removeEventListener('abort', abort);
|
||||
group.count -= 1;
|
||||
if (!group.count && !group.background && !group.settled) {
|
||||
group.controller.abort();
|
||||
// A fresh consumer must not join an abandoned queued generation.
|
||||
if (inFlight.get(group.key) === promise) inFlight.delete(group.key);
|
||||
if (consumers.get(group.key) === group) consumers.delete(group.key);
|
||||
}
|
||||
}
|
||||
function abort() {
|
||||
release();
|
||||
reject(signal.reason);
|
||||
}
|
||||
signal?.addEventListener('abort', abort, { once: true });
|
||||
if (signal?.aborted) abort();
|
||||
promise.then(
|
||||
(value) => {
|
||||
release();
|
||||
resolve(value);
|
||||
},
|
||||
(error) => {
|
||||
release();
|
||||
reject(error);
|
||||
},
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
function refresh(params, background = false) {
|
||||
// Global samples share pending refreshes across hourly window changes.
|
||||
const flightKey = params.scope === 'global' ? params.baseKey : params.key;
|
||||
let group = consumers.get(flightKey);
|
||||
if (!group) {
|
||||
group = {
|
||||
key: flightKey,
|
||||
controller: new AbortController(),
|
||||
count: 0,
|
||||
settled: false,
|
||||
background,
|
||||
};
|
||||
consumers.set(flightKey, group);
|
||||
}
|
||||
group.background ||= background;
|
||||
const { promise, shared } = coalesceProxyRequest(
|
||||
inFlight,
|
||||
flightKey,
|
||||
async () => {
|
||||
client ??= createInatClient({
|
||||
fetchImpl,
|
||||
now,
|
||||
cacheDir,
|
||||
warn,
|
||||
queueOptions,
|
||||
});
|
||||
const options = { signal: group.controller.signal };
|
||||
const result = await (params.scope === 'grid'
|
||||
? client.getGrid(params, options)
|
||||
: client.getObservations(params, options));
|
||||
const value = {
|
||||
fetchedAt: now(),
|
||||
...(params.scope === 'grid'
|
||||
? {}
|
||||
: { scope: params.scope, box: params.box }),
|
||||
...result,
|
||||
};
|
||||
cache.delete(params.key);
|
||||
cache.set(params.key, {
|
||||
value,
|
||||
savedAt: now(),
|
||||
baseKey: params.baseKey,
|
||||
});
|
||||
while (cache.size > INAT_CACHE_MAX)
|
||||
cache.delete(cache.keys().next().value);
|
||||
return value;
|
||||
},
|
||||
);
|
||||
if (!shared) {
|
||||
const settle = () => {
|
||||
group.settled = true;
|
||||
if (consumers.get(flightKey) === group) consumers.delete(flightKey);
|
||||
};
|
||||
void promise.then(settle, settle);
|
||||
}
|
||||
return { promise, group };
|
||||
}
|
||||
|
||||
async function acquire(params, signal) {
|
||||
signal?.throwIfAborted();
|
||||
for (const [key, entry] of cache) {
|
||||
if (age(entry) > STALE_MAX_MS) cache.delete(key);
|
||||
}
|
||||
const current = cache.get(params.key);
|
||||
const ttl =
|
||||
params.scope === 'grid'
|
||||
? INAT_GRID_CACHE_TTL_MS
|
||||
: params.scope === 'global'
|
||||
? GLOBAL_TTL_MS
|
||||
: TTL_MS;
|
||||
if (current && age(current) < ttl) {
|
||||
cache.delete(params.key);
|
||||
cache.set(params.key, current);
|
||||
return { value: current.value, stale: false };
|
||||
}
|
||||
// Window-hour rollover can reuse prior cells without claiming a new age.
|
||||
let previous = current;
|
||||
let previousKey = params.key;
|
||||
if (!previous) {
|
||||
for (const [key, entry] of cache) {
|
||||
if (
|
||||
entry.baseKey === params.baseKey &&
|
||||
(!previous || entry.savedAt > previous.savedAt)
|
||||
) {
|
||||
previous = entry;
|
||||
previousKey = key;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (params.scope === 'global' && age(previous) < GLOBAL_TTL_MS)
|
||||
return { value: previous.value, stale: false };
|
||||
if (
|
||||
params.scope === 'grid' &&
|
||||
!params.waitForFresh &&
|
||||
age(previous) <= STALE_MAX_MS
|
||||
) {
|
||||
cache.delete(previousKey);
|
||||
cache.set(previousKey, previous);
|
||||
// Retain one background consumer after this HTTP response closes. A stale
|
||||
// tile is useful immediately; waiting for the token queue defeats SWR.
|
||||
const cooling = client?.cooldownState('grid');
|
||||
if (!cooling) {
|
||||
const { promise } = refresh(params, true);
|
||||
void promise.catch(() => {});
|
||||
}
|
||||
return {
|
||||
value: previous.value,
|
||||
stale: true,
|
||||
...cooling,
|
||||
};
|
||||
}
|
||||
try {
|
||||
const { promise, group } = refresh(params);
|
||||
return { value: await joinFlight(promise, group, signal), stale: false };
|
||||
} catch (error) {
|
||||
if (signal?.aborted) throw error;
|
||||
if (age(previous) <= STALE_MAX_MS)
|
||||
return {
|
||||
value: previous.value,
|
||||
stale: true,
|
||||
retryAfterS: error.retryAfterS,
|
||||
upstreamStatus: error.upstreamStatus,
|
||||
};
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function handler(req, res) {
|
||||
const json = (
|
||||
status,
|
||||
value,
|
||||
{ stale = false, retryAfterS, upstreamStatus } = {},
|
||||
) => {
|
||||
if (res.destroyed) return;
|
||||
res.writeHead(status, {
|
||||
'Content-Type': 'application/json',
|
||||
'Cache-Control': 'no-store',
|
||||
...(status === 405 ? { Allow: 'GET' } : {}),
|
||||
...(Number.isFinite(retryAfterS)
|
||||
? { 'Retry-After': String(retryAfterS) }
|
||||
: {}),
|
||||
...(upstreamStatus
|
||||
? { 'X-Upstream-Status': String(upstreamStatus) }
|
||||
: {}),
|
||||
...(stale ? { 'X-Data-Stale': 'true' } : {}),
|
||||
});
|
||||
res.end(JSON.stringify(value));
|
||||
};
|
||||
if (admitSameSiteKeyless(req, res)) return;
|
||||
if (req.method !== 'GET') return json(405, { error: 'method_not_allowed' });
|
||||
const url = String(req.url || '/');
|
||||
const queryAt = url.indexOf('?');
|
||||
const pathname = queryAt < 0 ? url : url.slice(0, queryAt);
|
||||
if (pathname !== '/observations' && pathname !== '/grid')
|
||||
return json(404, { error: 'unknown_route' });
|
||||
const params = requestParams(
|
||||
pathname,
|
||||
new URLSearchParams(queryAt < 0 ? '' : url.slice(queryAt + 1)),
|
||||
now(),
|
||||
);
|
||||
if (params.error) return json(400, { error: params.error });
|
||||
const controller = new AbortController();
|
||||
const abort = () => controller.abort();
|
||||
// IncomingMessage.close also fires after a complete request body. The
|
||||
// response close identifies the abandoned browser fetch, while aborted
|
||||
// covers a request cut short before it reached the response stage.
|
||||
req.once?.('aborted', abort);
|
||||
res.once?.('close', abort);
|
||||
try {
|
||||
const { value, stale, retryAfterS, upstreamStatus } = await acquire(
|
||||
params,
|
||||
controller.signal,
|
||||
);
|
||||
return json(200, stale ? { ...value, stale: true } : value, {
|
||||
stale,
|
||||
retryAfterS,
|
||||
upstreamStatus,
|
||||
});
|
||||
} catch (error) {
|
||||
if (controller.signal.aborted) return;
|
||||
const status =
|
||||
error.code === 'inaturalist_budget_exhausted'
|
||||
? 503
|
||||
: error.status === 429
|
||||
? 429
|
||||
: 502;
|
||||
const code =
|
||||
error.code === 'inaturalist_budget_exhausted'
|
||||
? error.code
|
||||
: status === 429
|
||||
? 'inaturalist_rate_limited'
|
||||
: 'inaturalist_unavailable';
|
||||
return json(
|
||||
status,
|
||||
{ error: code },
|
||||
{
|
||||
retryAfterS: error.retryAfterS,
|
||||
upstreamStatus: error.upstreamStatus,
|
||||
},
|
||||
);
|
||||
} finally {
|
||||
req.removeListener?.('aborted', abort);
|
||||
res.removeListener?.('close', abort);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
name: 'inaturalist',
|
||||
configureServer({ middlewares }) {
|
||||
middlewares.use('/api/inaturalist', handler);
|
||||
},
|
||||
configurePreviewServer({ middlewares }) {
|
||||
middlewares.use('/api/inaturalist', handler);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,893 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { mkdtemp, readFile, readdir, rm } from 'node:fs/promises';
|
||||
import os from 'node:os';
|
||||
import path from 'node:path';
|
||||
import { installInatProxy, NOW } from './proxy.fixture.mjs';
|
||||
import { createInatClient, inatRetryCooldownMs } from './client.js';
|
||||
|
||||
const observation = (id = 1) => ({
|
||||
id,
|
||||
uri: `https://www.inaturalist.org/observations/${id}`,
|
||||
geojson: { coordinates: [-122.351, 47.613] },
|
||||
time_observed_at: '2026-11-04T17:00:00Z',
|
||||
observed_on: '2026-11-04',
|
||||
created_at: '2026-11-04T17:01:00Z',
|
||||
quality_grade: 'research',
|
||||
license_code: 'cc-by',
|
||||
positional_accuracy: 12,
|
||||
faves_count: 2,
|
||||
num_identification_agreements: 3,
|
||||
taxon: {
|
||||
name: 'Calypte anna',
|
||||
preferred_common_name: "Anna's hummingbird",
|
||||
iconic_taxon_name: 'Aves',
|
||||
},
|
||||
photos: [
|
||||
{
|
||||
url: 'https://inaturalist-open-data.s3.amazonaws.com/photos/123/square.jpeg',
|
||||
license_code: 'cc-by',
|
||||
attribution: 'Synthetic photographer, CC BY',
|
||||
},
|
||||
],
|
||||
user: { login: 'must-never-leave-server' },
|
||||
});
|
||||
const envelope = (rows = [observation()], total = rows.length) => ({
|
||||
total_results: total,
|
||||
page: 1,
|
||||
per_page: 200,
|
||||
results: rows,
|
||||
});
|
||||
const boxUrl = (index = 0, extra = '') =>
|
||||
`/observations?south=47.51&west=${-122.49 + index * 0.2}&north=47.69&east=${-122.31 + index * 0.2}${extra}`;
|
||||
|
||||
function install(options = {}, hook = 'configureServer') {
|
||||
return installInatProxy(
|
||||
boxUrl(),
|
||||
{
|
||||
// Serial dispatch makes queued cancellation and coalescing observable.
|
||||
queueOptions: { intervalMs: 0, windowMs: 0, maxInFlight: 1 },
|
||||
...options,
|
||||
},
|
||||
hook,
|
||||
);
|
||||
}
|
||||
|
||||
async function until(check) {
|
||||
for (let i = 0; i < 100; i += 1) {
|
||||
if (check()) return;
|
||||
await new Promise(setImmediate);
|
||||
}
|
||||
assert.fail('expected fake upstream calls were not dispatched');
|
||||
}
|
||||
|
||||
for (const hook of ['configureServer', 'configurePreviewServer']) {
|
||||
test(`${hook}: actual v2 envelope, fixed query, projection and privacy`, async () => {
|
||||
const calls = [];
|
||||
const cache = new Map();
|
||||
const request = install(
|
||||
{
|
||||
cache,
|
||||
fetchImpl: async (url, options) => {
|
||||
calls.push(new URL(url));
|
||||
assert.equal(options.redirect, 'error');
|
||||
assert.ok(options.signal instanceof AbortSignal);
|
||||
assert.equal(options.headers.Accept, 'application/json');
|
||||
assert.equal(
|
||||
options.headers['User-Agent'],
|
||||
'gods-eye-view-wildlife-sightings-proxy/1.0 (+https://github.com/bilawalsidhu/gods-eye-view)',
|
||||
);
|
||||
return Response.json(envelope());
|
||||
},
|
||||
},
|
||||
hook,
|
||||
);
|
||||
assert.equal(calls.length, 0, 'construction and mounting do not acquire');
|
||||
const result = await request(
|
||||
boxUrl(0, '&url=https://attacker.example&per_page=999'),
|
||||
);
|
||||
assert.equal(result.status, 200);
|
||||
assert.equal(result.headers['Cache-Control'], 'no-store');
|
||||
assert.equal(result.body.rows[0].stableId, 'inat:1');
|
||||
assert.equal(result.body.fetchedAt, NOW);
|
||||
assert.equal(result.body.scope, 'view');
|
||||
assert.deepEqual(result.body.box, {
|
||||
south: 47.5,
|
||||
west: -122.5,
|
||||
north: 47.7,
|
||||
east: -122.3,
|
||||
});
|
||||
const query = calls[0].searchParams;
|
||||
assert.equal(
|
||||
calls[0].origin + calls[0].pathname,
|
||||
'https://api.inaturalist.org/v2/observations',
|
||||
);
|
||||
// Only the projected fields are requested, never user or identification data.
|
||||
assert.match(query.get('fields'), /^\(id:!t,uri:!t,.*photos:\(url:!t/);
|
||||
assert.doesNotMatch(query.get('fields'), /user|identifications/);
|
||||
assert.equal(query.get('swlat'), '47.5');
|
||||
assert.equal(query.get('swlng'), '-122.5');
|
||||
assert.equal(query.get('nelat'), '47.7');
|
||||
assert.equal(query.get('nelng'), '-122.3');
|
||||
assert.equal(query.get('d1'), '2026-11-02');
|
||||
assert.equal(query.has('created_d1'), false);
|
||||
assert.equal(query.get('per_page'), '200');
|
||||
assert.equal(query.get('verifiable'), 'true');
|
||||
assert.equal(query.get('photos'), 'true');
|
||||
assert.equal(query.get('captive'), 'false');
|
||||
assert.equal(query.get('order_by'), 'random');
|
||||
assert.equal(query.has('order'), false);
|
||||
assert.doesNotMatch(
|
||||
JSON.stringify(result.body),
|
||||
/must-never-leave-server|"user"/,
|
||||
);
|
||||
assert.doesNotMatch(
|
||||
JSON.stringify([...cache.values()]),
|
||||
/must-never-leave-server|"user"/,
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
test('gate, method, route and query validation run before quota, cache or upstream', async () => {
|
||||
const cache = new Map();
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
cache,
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return Response.json(envelope());
|
||||
},
|
||||
});
|
||||
for (let i = 0; i < 32; i += 1) {
|
||||
const response = await request('/observations?scope=global', {
|
||||
headers: {
|
||||
host: 'localhost:5173',
|
||||
origin: 'https://foreign.example',
|
||||
'sec-fetch-site': 'cross-site',
|
||||
},
|
||||
});
|
||||
assert.equal(response.status, 403);
|
||||
}
|
||||
const method = await request('/observations', { method: 'POST' });
|
||||
assert.equal(method.status, 405);
|
||||
assert.equal(method.headers.Allow, 'GET');
|
||||
for (const route of ['/', '/unknown', '/observations/extra', '//[invalid'])
|
||||
assert.equal((await request(route)).body.error, 'unknown_route');
|
||||
assert.equal(
|
||||
(await request('/observations?scope=planet')).body.error,
|
||||
'invalid_scope',
|
||||
);
|
||||
for (const query of [
|
||||
'',
|
||||
'south=NaN&west=0&north=1&east=1',
|
||||
'south=&west=0&north=1&east=1',
|
||||
'south=0&west=0&north=0&east=1',
|
||||
'south=0&west=1&north=1&east=0',
|
||||
'south=-91&west=0&north=1&east=1',
|
||||
'south=0&west=-181&north=1&east=1',
|
||||
'south=0&west=0&north=41&east=1',
|
||||
'south=0&west=0&north=1&east=41',
|
||||
]) {
|
||||
const result = await request(`/observations?${query}`);
|
||||
assert.equal(result.status, 400, query);
|
||||
assert.equal(result.body.error, 'invalid_bbox', query);
|
||||
}
|
||||
assert.equal(cache.size, 0);
|
||||
assert.equal(calls, 0);
|
||||
const allowed = await request('/observations?scope=global', {
|
||||
headers: {
|
||||
host: 'gev.example',
|
||||
origin: 'https://gev.example',
|
||||
'sec-fetch-site': 'same-origin',
|
||||
'x-forwarded-proto': 'https',
|
||||
'x-forwarded-for': '203.0.113.9',
|
||||
},
|
||||
});
|
||||
assert.equal(allowed.status, 200);
|
||||
});
|
||||
|
||||
test('cached requests never exhaust a local request admission limiter', async () => {
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return Response.json(envelope());
|
||||
},
|
||||
});
|
||||
for (let i = 0; i < 30; i += 1) assert.equal((await request()).status, 200);
|
||||
const cached = await request();
|
||||
assert.equal(cached.status, 200);
|
||||
assert.equal(cached.headers['Retry-After'], undefined);
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
|
||||
test('proxy rejects oversized snapped spans before acquisition and preserves canonical client boxes', async () => {
|
||||
const urls = [];
|
||||
const request = install({
|
||||
fetchImpl: async (url) => {
|
||||
urls.push(new URL(url));
|
||||
return Response.json(envelope());
|
||||
},
|
||||
});
|
||||
for (const query of [
|
||||
'south=10.2&west=0&north=50.1&east=1',
|
||||
'south=0&west=10.2&north=1&east=50.1',
|
||||
]) {
|
||||
const result = await request(`/observations?${query}`);
|
||||
assert.equal(result.status, 400);
|
||||
assert.equal(result.body.error, 'invalid_bbox');
|
||||
}
|
||||
assert.equal(urls.length, 0);
|
||||
const canonical =
|
||||
'/observations?south=47.4&west=-122.3&north=47.6&east=-122.1';
|
||||
const first = await request(canonical);
|
||||
assert.equal(first.status, 200);
|
||||
assert.deepEqual(first.body.box, {
|
||||
south: 47.4,
|
||||
west: -122.3,
|
||||
north: 47.6,
|
||||
east: -122.1,
|
||||
});
|
||||
assert.equal(urls[0].searchParams.get('swlat'), '47.4');
|
||||
assert.equal(urls[0].searchParams.get('swlng'), '-122.3');
|
||||
const second = await request(canonical);
|
||||
assert.deepEqual(second.body, first.body);
|
||||
assert.equal(urls.length, 1, 'canonical boxes share their cache key');
|
||||
});
|
||||
|
||||
test('view snapping, whales, research and global queries obey the requested policy', async () => {
|
||||
const urls = [];
|
||||
const request = install({
|
||||
fetchImpl: async (url) => {
|
||||
urls.push(new URL(url));
|
||||
return Response.json(envelope());
|
||||
},
|
||||
});
|
||||
await request(
|
||||
'/observations?south=40.2&west=-123.7&north=46.1&east=-121.4&quality=research&taxa=Aves',
|
||||
);
|
||||
assert.equal(urls[0].searchParams.get('swlat'), '40');
|
||||
assert.equal(urls[0].searchParams.get('nelat'), '47');
|
||||
assert.equal(urls[0].searchParams.get('per_page'), '200');
|
||||
assert.equal(urls[0].searchParams.get('quality_grade'), 'research');
|
||||
assert.equal(urls[0].searchParams.get('iconic_taxa'), 'Aves');
|
||||
await request(boxUrl(0, '&taxa=whales'));
|
||||
const whales = urls[1].searchParams;
|
||||
assert.equal(whales.get('taxon_id'), '152871');
|
||||
assert.equal(whales.get('d1'), '2026-10-28');
|
||||
assert.equal(whales.has('created_d1'), false);
|
||||
assert.equal(whales.has('photos'), false);
|
||||
assert.equal(whales.has('iconic_taxa'), false);
|
||||
await request('/observations?scope=global&quality=typo&taxa=typo&south=bad');
|
||||
const global = urls[2].searchParams;
|
||||
for (const field of [
|
||||
'swlat',
|
||||
'swlng',
|
||||
'nelat',
|
||||
'nelng',
|
||||
'quality_grade',
|
||||
'iconic_taxa',
|
||||
])
|
||||
assert.equal(global.has(field), false, field);
|
||||
assert.equal(global.get('per_page'), '200');
|
||||
assert.equal(global.get('d1'), '2026-11-02');
|
||||
assert.equal(global.has('created_d1'), false);
|
||||
assert.equal(global.get('order_by'), 'observed_on');
|
||||
assert.equal(global.get('order'), 'desc');
|
||||
await request('/observations?scope=global&taxa=whales');
|
||||
const globalWhales = urls[3].searchParams;
|
||||
assert.equal(globalWhales.get('d1'), '2026-10-28');
|
||||
assert.equal(globalWhales.has('created_d1'), false);
|
||||
assert.equal(globalWhales.get('order_by'), 'observed_on');
|
||||
assert.equal(globalWhales.get('order'), 'desc');
|
||||
});
|
||||
|
||||
test('fresh cache hits and global ten-minute TTL spend no upstream quota', async () => {
|
||||
let at = NOW;
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
now: () => at,
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return Response.json(envelope());
|
||||
},
|
||||
});
|
||||
const first = await request();
|
||||
at += 299_999;
|
||||
assert.deepEqual((await request()).body, first.body);
|
||||
at += 1;
|
||||
await request();
|
||||
assert.equal(calls, 2);
|
||||
await request('/observations?scope=global');
|
||||
at += 599_999;
|
||||
await request('/observations?scope=global');
|
||||
assert.equal(calls, 3);
|
||||
at += 1;
|
||||
await request('/observations?scope=global');
|
||||
assert.equal(calls, 4);
|
||||
});
|
||||
|
||||
test('the same snapped view and window hour returns one cached random sample until expiry', async () => {
|
||||
let at = NOW;
|
||||
let draws = 0;
|
||||
const cache = new Map();
|
||||
const request = install({
|
||||
cache,
|
||||
now: () => at,
|
||||
fetchImpl: async (url) => {
|
||||
const query = new URL(url).searchParams;
|
||||
assert.equal(query.get('order_by'), 'random');
|
||||
assert.equal(query.get('per_page'), '200');
|
||||
return Response.json(envelope([observation(++draws)]));
|
||||
},
|
||||
});
|
||||
const first = await request();
|
||||
for (const elapsed of [1, 1000, 60_000, 299_999]) {
|
||||
at = NOW + elapsed;
|
||||
const next = await request();
|
||||
assert.deepEqual(next.body, first.body);
|
||||
}
|
||||
assert.equal(draws, 1);
|
||||
assert.match(cache.keys().next().value, /2026-11-02T18:00:00Z$/);
|
||||
at = NOW + 300_000;
|
||||
const refresh = await request();
|
||||
assert.equal(refresh.body.rows[0].stableId, 'inat:2');
|
||||
assert.equal(draws, 2);
|
||||
});
|
||||
|
||||
test('disconnected queued views are dropped while a remaining coalesced consumer keeps its request', async () => {
|
||||
const calls = [];
|
||||
const releases = [];
|
||||
const cache = new Map();
|
||||
const request = install({
|
||||
cache,
|
||||
fetchImpl: (url, { signal }) => {
|
||||
calls.push({ url: new URL(url), signal });
|
||||
return new Promise((resolve) => releases.push(resolve));
|
||||
},
|
||||
});
|
||||
const first = request(boxUrl(0));
|
||||
await until(() => releases.length === 1);
|
||||
const superseded = new AbortController();
|
||||
const oldView = request(boxUrl(1), { signal: superseded.signal });
|
||||
const oneConsumer = new AbortController();
|
||||
const one = request(boxUrl(2), { signal: oneConsumer.signal });
|
||||
const other = request(boxUrl(2));
|
||||
superseded.abort();
|
||||
oneConsumer.abort();
|
||||
await Promise.all([oldView, one]);
|
||||
// Rejoining the abandoned key creates a healthy generation; the old
|
||||
// coalescing promise must not later delete or cancel this replacement.
|
||||
const replacement = request(boxUrl(1));
|
||||
releases[0](Response.json(envelope()));
|
||||
await until(() => releases.length === 2);
|
||||
assert.equal(calls[1].url.searchParams.get('swlng'), '-122.1');
|
||||
assert.equal(calls[1].signal.aborted, false);
|
||||
releases[1](Response.json(envelope([observation(2)])));
|
||||
await until(() => releases.length === 3);
|
||||
assert.equal(calls[2].url.searchParams.get('swlng'), '-122.3');
|
||||
releases[2](Response.json(envelope([observation(3)])));
|
||||
const results = await Promise.all([first, other, replacement]);
|
||||
assert.ok(results.every((result) => result.status === 200));
|
||||
assert.equal(
|
||||
calls.length,
|
||||
3,
|
||||
'the abandoned queued generation never dispatched',
|
||||
);
|
||||
assert.equal(cache.size, 3);
|
||||
assert.equal(results[1].body.rows[0].stableId, 'inat:2');
|
||||
assert.equal(results[2].body.rows[0].stableId, 'inat:3');
|
||||
});
|
||||
|
||||
test('observation windows reject old sightings uploaded today and keep recent observations', async () => {
|
||||
for (const [taxa, recentTime, expectedD1] of [
|
||||
['all', '2026-11-04T17:00:00Z', '2026-11-02'],
|
||||
['whales', '2026-10-30T17:00:00Z', '2026-10-28'],
|
||||
]) {
|
||||
let upstreamQuery;
|
||||
const request = install({
|
||||
fetchImpl: async (url) => {
|
||||
upstreamQuery = new URL(url).searchParams;
|
||||
return Response.json(
|
||||
envelope([
|
||||
{
|
||||
...observation(80),
|
||||
time_observed_at: '2010-06-01T12:00:00Z',
|
||||
observed_on: '2010-06-01',
|
||||
},
|
||||
{
|
||||
...observation(81),
|
||||
time_observed_at: recentTime,
|
||||
observed_on: recentTime.slice(0, 10),
|
||||
},
|
||||
]),
|
||||
);
|
||||
},
|
||||
});
|
||||
const result = await request(boxUrl(0, `&taxa=${taxa}`));
|
||||
assert.equal(result.status, 200, taxa);
|
||||
assert.equal(upstreamQuery.get('d1'), expectedD1);
|
||||
assert.equal(upstreamQuery.has('created_d1'), false);
|
||||
assert.equal(upstreamQuery.get('order_by'), 'random');
|
||||
assert.deepEqual(
|
||||
result.body.rows.map((row) => row.stableId),
|
||||
['inat:81'],
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
test('same-key requests single-flight and a seventeenth distinct in-flight key queues', async () => {
|
||||
const releases = [];
|
||||
const request = install({
|
||||
fetchImpl: () => new Promise((resolve) => releases.push(resolve)),
|
||||
});
|
||||
const a = request();
|
||||
const b = request();
|
||||
await until(() => releases.length === 1);
|
||||
releases.shift()(Response.json(envelope()));
|
||||
assert.deepEqual((await a).body, (await b).body);
|
||||
const pending = Array.from({ length: 16 }, (_, i) => request(boxUrl(i + 1)));
|
||||
await until(() => releases.length === 1);
|
||||
pending.push(request(boxUrl(18)));
|
||||
for (let i = 0; i < 17; i += 1) {
|
||||
await until(() => releases.length > i);
|
||||
releases[i](Response.json(envelope()));
|
||||
}
|
||||
assert.ok((await Promise.all(pending)).every(({ status }) => status === 200));
|
||||
});
|
||||
|
||||
test('a fresh global sample survives the hourly observation-window rollover without another fetch', async () => {
|
||||
let at = Date.parse('2026-11-04T18:59:00Z');
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
now: () => at,
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return Response.json(envelope());
|
||||
},
|
||||
});
|
||||
const first = await request('/observations?scope=global');
|
||||
at += 60_000;
|
||||
const nextHour = await request('/observations?scope=global');
|
||||
assert.deepEqual(nextHour.body, first.body);
|
||||
assert.equal(nextHour.body.stale, undefined);
|
||||
assert.equal(calls, 1);
|
||||
at += 9 * 60_000;
|
||||
await request('/observations?scope=global');
|
||||
assert.equal(calls, 2);
|
||||
});
|
||||
|
||||
test('pending global refresh coalesces across the hourly observation-window rollover with a busy queue', async () => {
|
||||
let at = Date.parse('2026-11-04T18:59:59Z');
|
||||
const releases = [];
|
||||
const cache = new Map();
|
||||
const request = install({
|
||||
cache,
|
||||
now: () => at,
|
||||
fetchImpl: (url) =>
|
||||
new Promise((resolve) =>
|
||||
releases.push({
|
||||
resolve,
|
||||
url: new URL(url),
|
||||
}),
|
||||
),
|
||||
});
|
||||
const first = request(
|
||||
'/observations?scope=global&quality=research&taxa=Aves',
|
||||
);
|
||||
const views = Array.from({ length: 15 }, (_, i) => request(boxUrl(i)));
|
||||
await until(() => releases.length === 1);
|
||||
assert.equal(releases[0].url.searchParams.get('d1'), '2026-11-02');
|
||||
at = Date.parse('2026-11-04T19:00:01Z');
|
||||
const nextHour = request(
|
||||
'/observations?scope=global&quality=research&taxa=Aves',
|
||||
);
|
||||
await new Promise(setImmediate);
|
||||
assert.equal(
|
||||
releases.length,
|
||||
1,
|
||||
'the same global query joins the pending refresh',
|
||||
);
|
||||
const distinct = request(
|
||||
'/observations?scope=global&quality=research&taxa=Mammalia',
|
||||
);
|
||||
for (let i = 0; i < 17; i += 1) {
|
||||
await until(() => releases.length > i);
|
||||
releases[i].resolve(Response.json(envelope()));
|
||||
}
|
||||
const [original, joined] = await Promise.all([first, nextHour]);
|
||||
assert.equal(original.status, 200);
|
||||
assert.equal((await distinct).status, 200);
|
||||
assert.deepEqual(joined.body, original.body);
|
||||
assert.ok((await Promise.all(views)).every(({ status }) => status === 200));
|
||||
const globalEntries = [...cache.entries()].filter(
|
||||
([, entry]) => entry.value.scope === 'global',
|
||||
);
|
||||
assert.equal(globalEntries.length, 2);
|
||||
assert.match(globalEntries[0][0], /2026-11-02T18:00:00Z$/);
|
||||
assert.equal(
|
||||
(await request('/observations?scope=global&quality=research&taxa=Aves'))
|
||||
.status,
|
||||
200,
|
||||
);
|
||||
assert.equal(
|
||||
releases.length,
|
||||
17,
|
||||
'the accepted global answer keeps its ten-minute freshness',
|
||||
);
|
||||
});
|
||||
|
||||
test('view refreshes retain distinct hourly keys while a previous-hour request is pending', async () => {
|
||||
let at = Date.parse('2026-11-04T18:59:59Z');
|
||||
const releases = [];
|
||||
const request = install({
|
||||
now: () => at,
|
||||
fetchImpl: () => new Promise((resolve) => releases.push(resolve)),
|
||||
});
|
||||
const first = request();
|
||||
await until(() => releases.length === 1);
|
||||
at = Date.parse('2026-11-04T19:00:01Z');
|
||||
const nextHour = request();
|
||||
await new Promise(setImmediate);
|
||||
assert.equal(
|
||||
releases.length,
|
||||
1,
|
||||
'the next view queues behind the current call',
|
||||
);
|
||||
releases[0](Response.json(envelope()));
|
||||
await until(() => releases.length === 2);
|
||||
releases[1](Response.json(envelope()));
|
||||
assert.ok(
|
||||
(await Promise.all([first, nextHour])).every(
|
||||
({ status }) => status === 200,
|
||||
),
|
||||
);
|
||||
});
|
||||
|
||||
test('upstream minute ceiling waits for request 61 across client addresses', async () => {
|
||||
let at = NOW;
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
now: () => at,
|
||||
queueOptions: {
|
||||
now: () => at,
|
||||
intervalMs: 0,
|
||||
wait: async (ms) => {
|
||||
at += ms;
|
||||
},
|
||||
},
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return Response.json(envelope());
|
||||
},
|
||||
});
|
||||
for (let i = 0; i < 60; i += 1)
|
||||
assert.equal(
|
||||
(await request(boxUrl(i), { peer: `10.0.0.${i}` })).status,
|
||||
200,
|
||||
);
|
||||
const queued = await request(boxUrl(60), { peer: '10.0.1.1' });
|
||||
assert.equal(queued.status, 200);
|
||||
assert.equal(queued.headers['Retry-After'], undefined);
|
||||
assert.equal(at, NOW + 60_000);
|
||||
assert.equal(calls, 61);
|
||||
});
|
||||
|
||||
test('views refresh after many prior requests without fabricating an upstream limit', async () => {
|
||||
let at = NOW;
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
now: () => at,
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return Response.json(envelope());
|
||||
},
|
||||
});
|
||||
const first = await request();
|
||||
at += 300_001;
|
||||
for (let i = 1; i <= 60; i += 1)
|
||||
await request(boxUrl(i), { peer: `10.0.0.${i}` });
|
||||
const stale = await request();
|
||||
assert.equal(stale.status, 200);
|
||||
assert.equal(stale.body.stale, undefined);
|
||||
assert.notEqual(stale.body.fetchedAt, first.body.fetchedAt);
|
||||
assert.equal(calls, 62);
|
||||
});
|
||||
|
||||
test('bounded samples preserve partial metadata, and well-formed empty views are valid', async () => {
|
||||
let at = NOW;
|
||||
let fail = false;
|
||||
const request = install({
|
||||
now: () => at,
|
||||
fetchImpl: async () =>
|
||||
fail
|
||||
? new Response('unavailable', { status: 503 })
|
||||
: Response.json(envelope([observation()], 500)),
|
||||
});
|
||||
const first = await request();
|
||||
assert.equal(first.body.partial, true);
|
||||
assert.equal(first.body.rawRowCount, 500);
|
||||
fail = true;
|
||||
at += 59 * 60_000;
|
||||
const stale = await request();
|
||||
assert.equal(
|
||||
stale.status,
|
||||
200,
|
||||
'hourly window changes still permit stale fallback',
|
||||
);
|
||||
assert.equal(stale.headers['X-Data-Stale'], 'true');
|
||||
assert.equal(stale.body.stale, true);
|
||||
assert.equal(stale.body.fetchedAt, first.body.fetchedAt);
|
||||
assert.equal(stale.body.partial, true);
|
||||
at += 2 * 60_000;
|
||||
const expired = await request();
|
||||
assert.equal(expired.status, 502);
|
||||
assert.equal(expired.body.error, 'inaturalist_unavailable');
|
||||
const empty = install({
|
||||
fetchImpl: async () => Response.json(envelope([], 0)),
|
||||
});
|
||||
const result = await empty();
|
||||
assert.equal(result.status, 200);
|
||||
assert.deepEqual(result.body.rows, []);
|
||||
});
|
||||
|
||||
test('bad envelopes, JSON and capped bodies fail closed without inventing a cooldown', async () => {
|
||||
for (const bad of [
|
||||
() => Response.json({ total_results: 1, results: {} }),
|
||||
() => Response.json({ total_results: 1, observations: [observation()] }),
|
||||
() => Response.json({ total_results: '1', results: [observation()] }),
|
||||
() => Response.json(envelope([], 1)),
|
||||
() => Response.json(envelope([{ ...observation(), id: null }], 1)),
|
||||
() => new Response('{malformed'),
|
||||
() =>
|
||||
new Response('{}', {
|
||||
headers: { 'Content-Length': String(8 * 1024 * 1024 + 1) },
|
||||
}),
|
||||
]) {
|
||||
let at = NOW;
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
now: () => at,
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return calls === 1 ? bad() : Response.json(envelope());
|
||||
},
|
||||
});
|
||||
const first = await request();
|
||||
assert.equal(first.status, 502);
|
||||
assert.equal(first.body.error, 'inaturalist_unavailable');
|
||||
assert.equal(first.headers['Retry-After'], undefined);
|
||||
assert.equal((await request(boxUrl(1))).status, 200);
|
||||
assert.equal(calls, 2);
|
||||
}
|
||||
});
|
||||
|
||||
test('a wholly invalid iNat page cannot replace a previously valid snapshot', async () => {
|
||||
let at = NOW;
|
||||
let fail = false;
|
||||
const cache = new Map();
|
||||
const request = install({
|
||||
cache,
|
||||
now: () => at,
|
||||
fetchImpl: async () =>
|
||||
Response.json(
|
||||
fail
|
||||
? envelope([{ ...observation(), quality_grade: 'casual' }])
|
||||
: envelope(),
|
||||
),
|
||||
});
|
||||
const first = await request();
|
||||
const original = cache.values().next().value;
|
||||
fail = true;
|
||||
at += 300_001;
|
||||
const stale = await request();
|
||||
assert.equal(stale.status, 200);
|
||||
assert.equal(stale.body.stale, true);
|
||||
assert.equal(stale.body.fetchedAt, first.body.fetchedAt);
|
||||
assert.equal(cache.values().next().value, original);
|
||||
});
|
||||
|
||||
test('429 cooldown honors seconds, HTTP dates, and long upstream deadlines', async () => {
|
||||
assert.equal(inatRetryCooldownMs(null, NOW), 60_000);
|
||||
assert.equal(inatRetryCooldownMs('0', NOW), 0);
|
||||
assert.equal(inatRetryCooldownMs('2', NOW), 2000);
|
||||
assert.equal(inatRetryCooldownMs('999999', NOW), 999_999_000);
|
||||
assert.equal(inatRetryCooldownMs('7200', NOW), 7_200_000);
|
||||
assert.equal(
|
||||
inatRetryCooldownMs(new Date(NOW + 7_200_000).toUTCString(), NOW),
|
||||
7_200_000,
|
||||
);
|
||||
let at = NOW;
|
||||
const releases = [];
|
||||
const warnings = [];
|
||||
const request = install({
|
||||
now: () => at,
|
||||
warn: (message) => warnings.push(message),
|
||||
fetchImpl: () => new Promise((resolve) => releases.push(resolve)),
|
||||
});
|
||||
const first = request();
|
||||
const later = request(boxUrl(1));
|
||||
await until(() => releases.length === 1);
|
||||
releases[0](
|
||||
new Response('limit', {
|
||||
status: 429,
|
||||
headers: { 'Retry-After': '999999' },
|
||||
}),
|
||||
);
|
||||
assert.equal((await first).headers['Retry-After'], '999999');
|
||||
assert.equal((await later).headers['Retry-After'], '999999');
|
||||
at += 1000;
|
||||
assert.equal((await request(boxUrl(1))).headers['Retry-After'], '999998');
|
||||
at += 5000;
|
||||
const cooling = await request(boxUrl(2));
|
||||
assert.equal(cooling.status, 429);
|
||||
assert.equal(cooling.headers['Retry-After'], '999993');
|
||||
assert.equal(
|
||||
releases.length,
|
||||
1,
|
||||
'queued calls do not dispatch during cooldown',
|
||||
);
|
||||
assert.equal(
|
||||
warnings.filter((message) => message.includes('capped')).length,
|
||||
0,
|
||||
);
|
||||
});
|
||||
|
||||
for (const [upstreamStatus, expectedStatus, code, retryAfterS] of [
|
||||
[429, 429, 'inaturalist_rate_limited', 120],
|
||||
[503, 502, 'inaturalist_unavailable', 120],
|
||||
]) {
|
||||
test(`concurrent reservations honor an immediate upstream ${upstreamStatus} before dispatch`, async () => {
|
||||
let calls = 0;
|
||||
const warnings = [];
|
||||
const client = createInatClient({
|
||||
now: () => 0,
|
||||
cacheDir: null,
|
||||
queueOptions: { intervalMs: 0 },
|
||||
warn: (message) => warnings.push(message),
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return new Response('synthetic failure', {
|
||||
status: upstreamStatus,
|
||||
headers: { 'Retry-After': '120' },
|
||||
});
|
||||
},
|
||||
});
|
||||
const params = {
|
||||
scope: 'global',
|
||||
quality: 'all',
|
||||
taxa: 'all',
|
||||
observedD1: '1969-12-30',
|
||||
};
|
||||
const results = await Promise.allSettled(
|
||||
Array.from({ length: 3 }, () => client.getObservations(params)),
|
||||
);
|
||||
assert.equal(
|
||||
calls,
|
||||
1,
|
||||
'pending reservations must not dispatch during cooldown',
|
||||
);
|
||||
for (const result of results) {
|
||||
assert.equal(result.status, 'rejected');
|
||||
assert.equal(result.reason.status, expectedStatus);
|
||||
assert.equal(result.reason.code, code);
|
||||
assert.equal(result.reason.retryAfterS, retryAfterS);
|
||||
}
|
||||
assert.equal(
|
||||
warnings.length,
|
||||
1,
|
||||
'only the dispatched refresh logs a failure',
|
||||
);
|
||||
await assert.rejects(client.getObservations(params), {
|
||||
status: expectedStatus,
|
||||
code,
|
||||
retryAfterS,
|
||||
});
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
}
|
||||
|
||||
test('budget is read lazily, persisted as counters only, enforced across restart and UTC rollover', async (t) => {
|
||||
const parent = await mkdtemp(path.join(os.tmpdir(), 'gev-inat-budget-'));
|
||||
t.after(() => rm(parent, { recursive: true, force: true }));
|
||||
const cacheDir = path.join(parent, 'not-created');
|
||||
const before = process.env.INATURALIST_DAILY_BUDGET;
|
||||
process.env.INATURALIST_DAILY_BUDGET = '2';
|
||||
t.after(() =>
|
||||
before === undefined
|
||||
? delete process.env.INATURALIST_DAILY_BUDGET
|
||||
: (process.env.INATURALIST_DAILY_BUDGET = before),
|
||||
);
|
||||
let at = NOW;
|
||||
let calls = 0;
|
||||
const options = {
|
||||
now: () => at,
|
||||
cacheDir,
|
||||
warn() {},
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
return Response.json(envelope());
|
||||
},
|
||||
};
|
||||
const client = createInatClient(options);
|
||||
assert.deepEqual(
|
||||
await readdir(parent),
|
||||
[],
|
||||
'construction performs no directory write',
|
||||
);
|
||||
const params = {
|
||||
scope: 'global',
|
||||
quality: 'all',
|
||||
taxa: 'all',
|
||||
observedD1: '2026-11-02',
|
||||
};
|
||||
await Promise.all([
|
||||
client.getObservations(params),
|
||||
client.getObservations(params),
|
||||
]);
|
||||
assert.deepEqual(
|
||||
JSON.parse(await readFile(path.join(cacheDir, 'budget.json'), 'utf8')),
|
||||
{ date: '2026-11-04', count: 2 },
|
||||
);
|
||||
assert.deepEqual(
|
||||
await readdir(cacheDir),
|
||||
['budget.json'],
|
||||
'atomic temporary files are retired',
|
||||
);
|
||||
const restarted = createInatClient(options);
|
||||
await assert.rejects(restarted.getObservations(params), {
|
||||
status: 503,
|
||||
code: 'inaturalist_budget_exhausted',
|
||||
});
|
||||
assert.equal(calls, 2);
|
||||
at = Date.parse('2026-11-05T00:00:00Z');
|
||||
await restarted.getObservations(params);
|
||||
assert.deepEqual(restarted.budgetState(), {
|
||||
date: '2026-11-05',
|
||||
count: 1,
|
||||
budget: 2,
|
||||
});
|
||||
assert.deepEqual(
|
||||
JSON.parse(await readFile(path.join(cacheDir, 'budget.json'), 'utf8')),
|
||||
{ date: '2026-11-05', count: 1 },
|
||||
);
|
||||
assert.equal(calls, 3);
|
||||
});
|
||||
|
||||
test('memory cache is an LRU bounded to 400 projected entries', async () => {
|
||||
let at = NOW;
|
||||
const cache = new Map();
|
||||
const request = install({
|
||||
cache,
|
||||
now: () => at,
|
||||
fetchImpl: async () => Response.json(envelope()),
|
||||
});
|
||||
for (let i = 0; i < 400; i += 1) {
|
||||
if (i && i % 20 === 0) at += 61_000;
|
||||
await request(boxUrl(i), { peer: String(i) });
|
||||
}
|
||||
assert.equal(cache.size, 400);
|
||||
const firstKey = cache.keys().next().value;
|
||||
const secondKey = [...cache.keys()][1];
|
||||
await request(boxUrl(0));
|
||||
await request(boxUrl(400));
|
||||
assert.equal(cache.size, 400);
|
||||
assert.equal(cache.has(firstKey), true);
|
||||
assert.equal(cache.has(secondKey), false);
|
||||
});
|
||||
|
||||
test('aborted client calls spend no budget and never fetch', async () => {
|
||||
const controller = new AbortController();
|
||||
controller.abort();
|
||||
const client = createInatClient({
|
||||
now: () => NOW,
|
||||
cacheDir: null,
|
||||
queueOptions: { intervalMs: 0 },
|
||||
fetchImpl: () => assert.fail('aborted call fetched'),
|
||||
});
|
||||
await assert.rejects(
|
||||
client.getObservations({}, { signal: controller.signal }),
|
||||
{ name: 'AbortError' },
|
||||
);
|
||||
assert.equal(client.budgetState().count, 0);
|
||||
});
|
||||
@@ -0,0 +1,70 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import { EventEmitter } from 'node:events';
|
||||
import { inaturalistProxy } from './observations.js';
|
||||
|
||||
export const NOW = Date.parse('2026-11-04T18:05:00Z');
|
||||
|
||||
/** Mount the proxy over synthetic HTTP objects, without opening a socket. */
|
||||
export function installInatProxy(
|
||||
defaultUrl,
|
||||
options = {},
|
||||
hook = 'configureServer',
|
||||
) {
|
||||
let handler;
|
||||
const plugin = inaturalistProxy({
|
||||
now: () => NOW,
|
||||
cacheDir: null,
|
||||
queueOptions: { intervalMs: 0, windowMs: 0 },
|
||||
warn() {},
|
||||
...options,
|
||||
});
|
||||
assert.equal(plugin.name, 'inaturalist');
|
||||
assert.equal(typeof plugin.configureServer, 'function');
|
||||
assert.equal(typeof plugin.configurePreviewServer, 'function');
|
||||
plugin[hook]({
|
||||
middlewares: {
|
||||
use(mount, callback) {
|
||||
assert.equal(mount, '/api/inaturalist');
|
||||
handler = callback;
|
||||
},
|
||||
},
|
||||
});
|
||||
return async (
|
||||
url = defaultUrl,
|
||||
{ method = 'GET', peer = 'local', headers = {}, signal } = {},
|
||||
) => {
|
||||
const res = Object.assign(new EventEmitter(), {
|
||||
headers: {},
|
||||
writeHead(status, headers) {
|
||||
this.status = status;
|
||||
Object.assign(this.headers, headers);
|
||||
},
|
||||
setHeader(key, value) {
|
||||
this.headers[key] = value;
|
||||
},
|
||||
end(value) {
|
||||
this.status ??= this.statusCode;
|
||||
this.body = JSON.parse(value);
|
||||
},
|
||||
});
|
||||
const abort = () => {
|
||||
res.destroyed = true;
|
||||
res.emit('close');
|
||||
};
|
||||
signal?.addEventListener('abort', abort, { once: true });
|
||||
try {
|
||||
await handler(
|
||||
Object.assign(new EventEmitter(), {
|
||||
url,
|
||||
method,
|
||||
headers,
|
||||
socket: { remoteAddress: peer },
|
||||
}),
|
||||
res,
|
||||
);
|
||||
} finally {
|
||||
signal?.removeEventListener('abort', abort);
|
||||
}
|
||||
return res;
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
import { setTimeout as delay } from 'node:timers/promises';
|
||||
|
||||
// A four-request burst keeps a cold view responsive; thereafter every provider
|
||||
// call shares the same one-per-second token supply. The rolling minute ceiling
|
||||
// queues the few requests a token bucket alone could allow above 60/min.
|
||||
const INAT_INTERVAL_MS = 1000;
|
||||
const INAT_BURST = 4;
|
||||
const INAT_WINDOW_MS = 60_000;
|
||||
const INAT_MAX_PER_WINDOW = 60;
|
||||
const INAT_MAX_IN_FLIGHT = 4;
|
||||
|
||||
/** Observation queries outrank tiles waiting for a token. Aborted consumers
|
||||
* leave the queue before an attempt or daily-budget reservation is spent. */
|
||||
export function createInatQueue({
|
||||
now = () => performance.now(),
|
||||
wait = (ms) => delay(ms),
|
||||
intervalMs = INAT_INTERVAL_MS,
|
||||
burst = INAT_BURST,
|
||||
windowMs = INAT_WINDOW_MS,
|
||||
maxInFlight = INAT_MAX_IN_FLIGHT,
|
||||
} = {}) {
|
||||
const pending = [];
|
||||
const dispatchedAt = [];
|
||||
let tokens = burst;
|
||||
let refilledAt = null;
|
||||
let running = false;
|
||||
let inFlight = 0;
|
||||
let slotFreed = null;
|
||||
|
||||
function refill() {
|
||||
const at = now();
|
||||
if (refilledAt !== null && intervalMs > 0)
|
||||
tokens = Math.min(
|
||||
burst,
|
||||
tokens + Math.max(0, at - refilledAt) / intervalMs,
|
||||
);
|
||||
if (intervalMs <= 0) tokens = burst;
|
||||
refilledAt = at;
|
||||
}
|
||||
|
||||
function schedule() {
|
||||
if (running) return;
|
||||
running = true;
|
||||
// A microtask also lets same-turn observation queries take priority.
|
||||
void Promise.resolve().then(drain);
|
||||
}
|
||||
|
||||
async function drain() {
|
||||
try {
|
||||
while (pending.length) {
|
||||
if (inFlight >= maxInFlight) {
|
||||
await new Promise((resolve) => (slotFreed = resolve));
|
||||
continue;
|
||||
}
|
||||
refill();
|
||||
const at = now();
|
||||
while (dispatchedAt.length && dispatchedAt[0] <= at - windowMs)
|
||||
dispatchedAt.shift();
|
||||
const tokenWait = tokens < 1 ? (1 - tokens) * intervalMs : 0;
|
||||
const windowWait =
|
||||
windowMs > 0 && dispatchedAt.length >= INAT_MAX_PER_WINDOW
|
||||
? dispatchedAt[0] + windowMs - at
|
||||
: 0;
|
||||
if (tokenWait > 0 || windowWait > 0) {
|
||||
await wait(Math.ceil(Math.max(tokenWait, windowWait)));
|
||||
continue;
|
||||
}
|
||||
// Stable sorting preserves FIFO order within each priority.
|
||||
pending.sort((a, b) => a.priority - b.priority);
|
||||
const job = pending.shift();
|
||||
job.signal?.removeEventListener('abort', job.abort);
|
||||
if (job.signal?.aborted) {
|
||||
job.reject(job.signal.reason);
|
||||
continue;
|
||||
}
|
||||
tokens -= 1;
|
||||
if (windowMs > 0) dispatchedAt.push(at);
|
||||
inFlight += 1;
|
||||
// A rejected task releases its slot without stopping the queue.
|
||||
void new Promise((resolve) => resolve(job.task()))
|
||||
.then(job.resolve, job.reject)
|
||||
.finally(() => {
|
||||
inFlight -= 1;
|
||||
const wake = slotFreed;
|
||||
slotFreed = null;
|
||||
wake?.();
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
running = false;
|
||||
}
|
||||
}
|
||||
|
||||
return function enqueue(task, { signal, priority = 1 } = {}) {
|
||||
if (signal?.aborted) return Promise.reject(signal.reason);
|
||||
return new Promise((resolve, reject) => {
|
||||
const job = { task, signal, priority, resolve, reject };
|
||||
job.abort = () => {
|
||||
const index = pending.indexOf(job);
|
||||
if (index >= 0) pending.splice(index, 1);
|
||||
signal.removeEventListener('abort', job.abort);
|
||||
reject(signal.reason);
|
||||
};
|
||||
signal?.addEventListener('abort', job.abort, { once: true });
|
||||
pending.push(job);
|
||||
schedule();
|
||||
});
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createInatQueue } from './queue.js';
|
||||
|
||||
async function until(check) {
|
||||
for (let i = 0; i < 100; i += 1) {
|
||||
if (check()) return;
|
||||
await new Promise(setImmediate);
|
||||
}
|
||||
assert.fail('queue did not reach its expected state');
|
||||
}
|
||||
|
||||
test('shared governor paces a four-call burst within the rolling minute ceiling', async () => {
|
||||
let at = 0;
|
||||
const waits = [];
|
||||
const queue = createInatQueue({
|
||||
now: () => at,
|
||||
wait: async (ms) => {
|
||||
waits.push(ms);
|
||||
at += ms;
|
||||
},
|
||||
});
|
||||
const dispatched = [];
|
||||
await Promise.all(
|
||||
Array.from({ length: 65 }, () => queue(() => dispatched.push(at))),
|
||||
);
|
||||
assert.equal(dispatched.length, 65);
|
||||
assert.deepEqual(dispatched.slice(0, 7), [0, 0, 0, 0, 1000, 2000, 3000]);
|
||||
assert.deepEqual(waits.slice(0, 3), [1000, 1000, 1000]);
|
||||
for (const time of dispatched)
|
||||
assert.ok(
|
||||
dispatched.filter((other) => other > time - 60_000 && other <= time)
|
||||
.length <= 60,
|
||||
);
|
||||
assert.equal(dispatched[60], 60_000);
|
||||
});
|
||||
|
||||
test('an observation arriving during the pacing wait overtakes queued tiles', async () => {
|
||||
let at = 0;
|
||||
const waits = [];
|
||||
const queue = createInatQueue({
|
||||
burst: 1,
|
||||
now: () => at,
|
||||
wait: (ms) => new Promise((resolve) => waits.push({ ms, resolve })),
|
||||
});
|
||||
const dispatched = [];
|
||||
await queue(() => dispatched.push('first tile'));
|
||||
const tile = queue(() => dispatched.push('waiting tile'));
|
||||
await until(() => waits.length === 1);
|
||||
const observation = queue(() => dispatched.push('observation'), {
|
||||
priority: 0,
|
||||
});
|
||||
at += waits[0].ms;
|
||||
waits[0].resolve();
|
||||
await observation;
|
||||
assert.deepEqual(dispatched, ['first tile', 'observation']);
|
||||
await until(() => waits.length === 2);
|
||||
at += waits[1].ms;
|
||||
waits[1].resolve();
|
||||
await tile;
|
||||
assert.deepEqual(dispatched, ['first tile', 'observation', 'waiting tile']);
|
||||
});
|
||||
|
||||
test('superseded tasks leave before dispatch, and a rejection cannot poison the tail', async () => {
|
||||
const queue = createInatQueue({ intervalMs: 0 });
|
||||
const dispatched = [];
|
||||
let release;
|
||||
const first = queue(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
release = resolve;
|
||||
}),
|
||||
);
|
||||
await until(() => release);
|
||||
const controller = new AbortController();
|
||||
const superseded = queue(() => assert.fail('superseded tile dispatched'), {
|
||||
signal: controller.signal,
|
||||
});
|
||||
const dropped = assert.rejects(superseded, { name: 'AbortError' });
|
||||
controller.abort();
|
||||
await dropped;
|
||||
const failed = queue(() => {
|
||||
dispatched.push('failed');
|
||||
throw new Error('synthetic');
|
||||
});
|
||||
const failure = assert.rejects(failed, /synthetic/);
|
||||
const succeeding = queue(() => dispatched.push('recovered'));
|
||||
release();
|
||||
await Promise.all([first, failure, succeeding]);
|
||||
assert.deepEqual(dispatched, ['failed', 'recovered']);
|
||||
});
|
||||
|
||||
test('a slow call does not hold later calls; at most four run at once', async () => {
|
||||
const queue = createInatQueue({ intervalMs: 0, windowMs: 0 });
|
||||
const releases = [];
|
||||
const started = [];
|
||||
const slow = (id) => () =>
|
||||
new Promise((resolve) => {
|
||||
started.push(id);
|
||||
releases.push(() => resolve(id));
|
||||
});
|
||||
const jobs = [1, 2, 3, 4, 5].map((id) => queue(slow(id)));
|
||||
await until(() => started.length === 4);
|
||||
assert.deepEqual(started, [1, 2, 3, 4], 'four start without waiting');
|
||||
releases[1]();
|
||||
assert.equal(await jobs[1], 2, 'a later call finishes before the first');
|
||||
await until(() => started.length === 5);
|
||||
assert.deepEqual(started, [1, 2, 3, 4, 5], 'a freed slot starts the next');
|
||||
for (const release of releases) release();
|
||||
assert.deepEqual(await Promise.all(jobs), [1, 2, 3, 4, 5]);
|
||||
});
|
||||
@@ -0,0 +1,218 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createInatClient } from './client.js';
|
||||
import { installInatProxy, NOW } from './proxy.fixture.mjs';
|
||||
import { syntheticGrid } from './grid.fixture.mjs';
|
||||
import { INAT_GRID_DECODE_MEASURE } from './grid.js';
|
||||
|
||||
const gridUrl = '/grid?z=2&x=1&y=1&taxa=Aves';
|
||||
const params = {
|
||||
scope: 'global',
|
||||
taxa: 'all',
|
||||
observedD1: '2026-11-02',
|
||||
};
|
||||
const tick = () => new Promise(setImmediate);
|
||||
|
||||
const install = (options, hook) => installInatProxy(gridUrl, options, hook);
|
||||
|
||||
test('tile cache lasts fifteen minutes, then returns stale before revalidation finishes', async () => {
|
||||
let at = NOW;
|
||||
let calls = 0;
|
||||
let release;
|
||||
const request = install({
|
||||
now: () => at,
|
||||
fetchImpl: () => {
|
||||
calls += 1;
|
||||
return calls === 1
|
||||
? Response.json(syntheticGrid())
|
||||
: new Promise((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
},
|
||||
});
|
||||
const first = await request();
|
||||
at += 899_999;
|
||||
const freshPromise = request();
|
||||
await tick();
|
||||
assert.equal(calls, 1, 'a fresh tile never spends another upstream attempt');
|
||||
assert.deepEqual((await freshPromise).body, first.body);
|
||||
at += 1;
|
||||
const stalePromise = request();
|
||||
let resolved = false;
|
||||
void stalePromise.then(() => {
|
||||
resolved = true;
|
||||
});
|
||||
await tick();
|
||||
try {
|
||||
assert.equal(
|
||||
resolved,
|
||||
true,
|
||||
'stale cells publish while refresh is pending',
|
||||
);
|
||||
const stale = await stalePromise;
|
||||
assert.equal(stale.body.stale, true);
|
||||
assert.equal(stale.body.fetchedAt, NOW);
|
||||
assert.deepEqual(stale.body.cells, first.body.cells);
|
||||
await request();
|
||||
assert.equal(calls, 2, 'all stale consumers share one background refresh');
|
||||
const refreshPromise = request(`${gridUrl}&refresh=wait`);
|
||||
let refreshed = false;
|
||||
void refreshPromise.then(() => {
|
||||
refreshed = true;
|
||||
});
|
||||
await tick();
|
||||
assert.equal(
|
||||
refreshed,
|
||||
false,
|
||||
'a background consumer can await the same refresh',
|
||||
);
|
||||
release(Response.json(syntheticGrid()));
|
||||
const fresh = await refreshPromise;
|
||||
assert.equal(fresh.body.stale, undefined);
|
||||
assert.equal(fresh.body.fetchedAt, at);
|
||||
assert.equal(
|
||||
calls,
|
||||
2,
|
||||
'waiting for fresh cells never duplicates upstream work',
|
||||
);
|
||||
release = null;
|
||||
} finally {
|
||||
release?.(Response.json(syntheticGrid()));
|
||||
}
|
||||
});
|
||||
|
||||
test('more than sixteen queued views and thirty local requests never report local rate limiting', async () => {
|
||||
let release;
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
fetchImpl: async () => {
|
||||
calls += 1;
|
||||
if (calls === 1)
|
||||
await new Promise((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
return Response.json(syntheticGrid());
|
||||
},
|
||||
});
|
||||
const pending = Array.from({ length: 40 }, (_, x) =>
|
||||
request(`/grid?z=6&x=${x}&y=1`),
|
||||
);
|
||||
await tick();
|
||||
release();
|
||||
const results = await Promise.all(pending);
|
||||
assert.ok(results.every((result) => result.status === 200));
|
||||
assert.equal(calls, 40);
|
||||
});
|
||||
|
||||
test('only real upstream 429 or 5xx failures start cooldown', async () => {
|
||||
for (const bad of [
|
||||
() => Promise.reject(new Error('network')),
|
||||
() => Response.json({ malformed: true }),
|
||||
() => new Response('', { status: 404 }),
|
||||
]) {
|
||||
let calls = 0;
|
||||
const client = createInatClient({
|
||||
now: () => NOW,
|
||||
cacheDir: null,
|
||||
queueOptions: { intervalMs: 0, windowMs: 0 },
|
||||
warn() {},
|
||||
fetchImpl: () => {
|
||||
calls += 1;
|
||||
return calls === 1
|
||||
? bad()
|
||||
: Response.json({ results: [], total_results: 0 });
|
||||
},
|
||||
});
|
||||
await assert.rejects(client.getObservations(params), { status: 502 });
|
||||
await client.getObservations(params);
|
||||
assert.equal(
|
||||
calls,
|
||||
2,
|
||||
'a non-cooldown failure must not hold the next attempt',
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
test('actual upstream cooldown is marked separately from a local budget stop', async () => {
|
||||
for (const upstream of [429, 503]) {
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
fetchImpl: () => {
|
||||
calls += 1;
|
||||
return new Response('', {
|
||||
status: upstream,
|
||||
headers: { 'Retry-After': '120' },
|
||||
});
|
||||
},
|
||||
});
|
||||
const first = await request();
|
||||
assert.equal(first.headers['X-Upstream-Status'], String(upstream));
|
||||
assert.equal(first.headers['Retry-After'], '120');
|
||||
await request();
|
||||
assert.equal(calls, 1);
|
||||
}
|
||||
});
|
||||
|
||||
test('upstream Retry-After zero survives the HTTP response and permits an immediate next request', async () => {
|
||||
for (const upstream of [429, 503]) {
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
fetchImpl: () => {
|
||||
calls += 1;
|
||||
return calls === 1
|
||||
? new Response('', {
|
||||
status: upstream,
|
||||
headers: { 'Retry-After': '0' },
|
||||
})
|
||||
: Response.json(syntheticGrid());
|
||||
},
|
||||
});
|
||||
const first = await request();
|
||||
assert.equal(first.headers['X-Upstream-Status'], String(upstream));
|
||||
assert.equal(first.headers['Retry-After'], '0');
|
||||
assert.equal((await request()).status, 200);
|
||||
assert.equal(calls, 2);
|
||||
}
|
||||
});
|
||||
|
||||
test('a local daily budget stop has a distinct code and no upstream cooldown marker', async (t) => {
|
||||
const before = process.env.INATURALIST_DAILY_BUDGET;
|
||||
process.env.INATURALIST_DAILY_BUDGET = '1';
|
||||
t.after(() => {
|
||||
if (before === undefined) delete process.env.INATURALIST_DAILY_BUDGET;
|
||||
else process.env.INATURALIST_DAILY_BUDGET = before;
|
||||
});
|
||||
let calls = 0;
|
||||
const request = install({
|
||||
fetchImpl: () => {
|
||||
calls += 1;
|
||||
return Response.json(syntheticGrid());
|
||||
},
|
||||
});
|
||||
assert.equal((await request()).status, 200);
|
||||
const stopped = await request('/grid?z=2&x=0&y=1');
|
||||
assert.equal(stopped.status, 503);
|
||||
assert.equal(stopped.body.error, 'inaturalist_budget_exhausted');
|
||||
assert.equal(stopped.headers['X-Upstream-Status'], undefined);
|
||||
assert.equal((await request()).status, 200, 'cached data remains available');
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
|
||||
test('cached tiles avoid UTFGrid decode and decode timing entries stay bounded', async () => {
|
||||
const request = install({
|
||||
fetchImpl: () => Response.json(syntheticGrid()),
|
||||
});
|
||||
await request();
|
||||
const sample = performance.getEntriesByName(INAT_GRID_DECODE_MEASURE)[0];
|
||||
assert.ok(sample.duration >= 0);
|
||||
for (let i = 0; i < 10; i += 1) await request();
|
||||
assert.equal(
|
||||
performance.getEntriesByName(INAT_GRID_DECODE_MEASURE)[0],
|
||||
sample,
|
||||
);
|
||||
for (let x = 0; x < 4; x += 1) await request(`/grid?z=3&x=${x}&y=1`);
|
||||
assert.equal(
|
||||
performance.getEntriesByName(INAT_GRID_DECODE_MEASURE).length,
|
||||
1,
|
||||
);
|
||||
});
|
||||
@@ -30,6 +30,8 @@ import { gnssIntegrityProxy } from './gnssIntegrity.js';
|
||||
import { ghostMapsProxy } from './ghostMaps.js';
|
||||
import { droneAirspaceProxy } from './droneAirspace.js';
|
||||
import { newsProxy } from './news.js';
|
||||
import { inaturalistProxy } from './inaturalist/observations.js';
|
||||
import { acartiaProxy } from './acartia.js';
|
||||
import { cycloneProxy } from './cyclones.js';
|
||||
import { windProxy } from './wind.js';
|
||||
|
||||
@@ -83,6 +85,8 @@ function localProviderPlugins({
|
||||
ghostMapsProxy(),
|
||||
droneAirspaceProxy(),
|
||||
newsProxy(),
|
||||
inaturalistProxy(),
|
||||
acartiaProxy(),
|
||||
keySetupEndpoint(),
|
||||
];
|
||||
}
|
||||
|
||||
@@ -523,7 +523,7 @@ export const ACTION_DESCRIPTIONS = {
|
||||
properties: {
|
||||
direction: {
|
||||
description:
|
||||
'Required for pan/tilt/rotate; orbit/rotate_globe default right.',
|
||||
'Pan/rotate/tilt steps; orbit/rotate_globe default right.',
|
||||
$position: 2,
|
||||
},
|
||||
mode: {
|
||||
@@ -531,6 +531,10 @@ export const ACTION_DESCRIPTIONS = {
|
||||
'once = bounded eased nudge; continuous = until stop/manual input. rotate_globe defaults continuous; other motions default once.',
|
||||
$position: 2,
|
||||
},
|
||||
pitchDeg: {
|
||||
description: 'Absolute; -45 default, -89 top-down.',
|
||||
$position: 1,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
|
||||
Reference in new issue
Block a user