diff --git a/docs/live-visitors.md b/docs/live-visitors.md index 8c61d9046..f3016aa60 100644 --- a/docs/live-visitors.md +++ b/docs/live-visitors.md @@ -82,17 +82,35 @@ DNT and GPC prevent collection in both the browser and server. Browser collectio is disabled in development, staging, and outside `tanstack.com`. `isLivePage` allows public library content and selected public site routes. +Showcase submission, editing, and individual detail pages are excluded because +detail pages can include owner-only drafts. Existing records with excluded paths +are filtered out of admin responses and expire through normal retention. Account, admin, auth, chat, private builder projects, shop, and internal endpoints are excluded. Unknown paths are excluded until added to the public allowlist. Queries, fragments, clipboard text, form values, event properties, account IDs, emails, and user agents are not stored. Only explicitly named actions from the existing analytics hooks are collected; there is no arbitrary click-text capture. -Visitor IPs exist only during the server-side lookup and are not written to the -presence tables or admin responses. Lookup tracing and Sentry breadcrumbs are +Visitor IPs are used transiently for the server-side lookup and Cloudflare edge +rate-limit counters, and are not written to the application database or admin +responses. Lookup tracing and Sentry breadcrumbs are suppressed to keep IP query parameters out of telemetry. The lookup key stays server-side. There is no joining sessions by IP, fingerprint, or login. +## Intake limits + +Cloudflare edge bindings limit intake to 600 requests per IP per minute and new +company lookups to 30 per IP per minute, per Cloudflare location. These are +approximate abuse limits, not a global billing quota. Intake returns 429 before +reading a body or writing presence when limited, and 503 if the limiter is +unavailable. A lookup limit leaves presence intact with a separate rate-limited +status. Corporate networks share an IP, so these limits can affect shared +networks during large bursts. Raw IPs are not added to the database-backed MCP +rate-limit table. Presence cleanup runs independently of workflow processing, so +a cleanup failure does not prevent scheduled workflow work. + +[Cloudflare rate-limit bindings](https://developers.cloudflare.com/workers/runtime-apis/bindings/rate-limit/) + ## Validation Focused tests cover privacy signals, intake validation, public-route boundaries, diff --git a/src/components/admin/LiveVisitors.tsx b/src/components/admin/LiveVisitors.tsx index 47eb27b0b..f71d0a06b 100644 --- a/src/components/admin/LiveVisitors.tsx +++ b/src/components/admin/LiveVisitors.tsx @@ -67,6 +67,7 @@ const lookupLabels: Record = { pending: 'Looking up company', unknown: 'No cached company match', failed: 'Company lookup failed', + rate_limited: 'Company lookup rate limited', unconfigured: 'Company lookup not configured', } function age(time: Date | string | number, now: string) { diff --git a/src/db/schema.ts b/src/db/schema.ts index 54a0fd536..bba2b3d84 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -3470,7 +3470,14 @@ export const liveSessions = pgTable( company: jsonb('company').$type(), lookupStatus: text('lookup_status', { - enum: ['pending', 'matched', 'unknown', 'failed', 'unconfigured'], + enum: [ + 'pending', + 'matched', + 'unknown', + 'failed', + 'unconfigured', + 'rate_limited', + ], }).notNull(), }, (table) => [index('live_sessions_created_at_idx').on(table.createdAt)], diff --git a/src/server/runtime/host.server.ts b/src/server/runtime/host.server.ts index 6d5990be3..e22b3338e 100644 --- a/src/server/runtime/host.server.ts +++ b/src/server/runtime/host.server.ts @@ -5,7 +5,13 @@ type StaticAssetService = { fetch: (input: RequestInfo | URL, init?: RequestInit) => Promise } +export type HostRateLimiter = { + limit(options: { key: string }): Promise<{ success: boolean }> +} + type HostRuntimeEnv = Record & { + LIVE_PRESENCE_RATE_LIMITER?: HostRateLimiter + LIVE_LOOKUP_RATE_LIMITER?: HostRateLimiter CLOUDFLARE_CACHE_PURGE_TOKEN?: string CLOUDFLARE_ZONE_ID?: string GITHUB_AUTH_TOKEN?: string diff --git a/src/server/scheduled.server.ts b/src/server/scheduled.server.ts index 54e60d31d..c1b55aa49 100644 --- a/src/server/scheduled.server.ts +++ b/src/server/scheduled.server.ts @@ -34,19 +34,30 @@ export async function runScheduledTasks(cron: string, scheduledTime: number) { ]) return case WORKFLOW_SWEEP_CRON: - await runWorkflowSweep(cron, scheduledTime) + await Promise.all([ + runWorkflowSweep(cron, scheduledTime), + runLivePresencePrune(scheduledTime), + ]) return default: console.warn(`[scheduled] No task registered for cron: ${cron}`) } } +async function runLivePresencePrune(scheduledTime: number) { + const startTime = Date.now() + try { + await pruneLiveVisitors(scheduledTime) + } catch (error) { + logScheduledError('live-presence-prune', startTime, error) + } +} + async function runWorkflowSweep(cron: string, scheduledTime: number) { const startTime = Date.now() console.log('[workflow-sweep] Starting workflow sweep...') try { - await pruneLiveVisitors(scheduledTime) const interruptedBuilderRuns = await interruptExpiredBuilderProjectRunLeases({ occurredAt: new Date(scheduledTime), diff --git a/src/utils/analytics/live.server.ts b/src/utils/analytics/live.server.ts index eda58e50f..0bd95c553 100644 --- a/src/utils/analytics/live.server.ts +++ b/src/utils/analytics/live.server.ts @@ -3,9 +3,15 @@ import { isIP } from 'node:net' import { db } from '~/db/client' import { liveSessions, liveTabs } from '~/db/schema' import { requireAdmin } from '~/utils/roles.server' -import { scheduleHostRuntimeTask } from '~/server/runtime/host.server' +import { + scheduleHostRuntimeTask, + type HostRateLimiter, +} from '~/server/runtime/host.server' import { getLiveConfig, lookupScarfCompany } from './scarf-lookup.server' import { + isLivePage, + livePublicPagePattern, + liveForbiddenPagePattern, liveEventSchema, LIVE_RETENTION_MS, LIVE_TIMEOUT_MS, @@ -33,6 +39,18 @@ export async function receiveLiveEvent(request: Request) { if (!ip || !isIP(ip)) return respond(204) const config = await getLiveConfig() if (!config.enabled) return respond(204) + // Edge counters bound abuse without writing visitor IPs to the database. + if (!config.requestLimiter || (config.key && !config.lookupLimiter)) + return respond(503) + try { + if (!(await config.requestLimiter.limit({ key: ip })).success) + return new Response(null, { + status: 429, + headers: { ...responseHeaders, 'Retry-After': '60' }, + }) + } catch { + return respond(503) + } const reader = request.body?.getReader() if (!reader) return respond(400) const decoder = new TextDecoder() @@ -58,7 +76,7 @@ export async function receiveLiveEvent(request: Request) { const parsed = liveEventSchema.safeParse(value) if (!parsed.success) return respond(400) try { - await recordLiveEvent(parsed.data, ip, config.key) + await recordLiveEvent(parsed.data, ip, config.key, config.lookupLimiter) return respond(204) } catch { // Never log a request, body, IP, or third-party exception here. @@ -71,6 +89,7 @@ export async function recordLiveEvent( event: LiveEvent, ip: string, key: string | undefined, + lookupLimiter: HostRateLimiter | undefined, ) { const now = new Date() if (event.kind === 'leave') { @@ -127,6 +146,16 @@ export async function recordLiveEvent( if (created && key) { const enrich = async () => { try { + if ( + !lookupLimiter || + !(await lookupLimiter.limit({ key: ip })).success + ) { + await db + .update(liveSessions) + .set({ lookupStatus: 'rate_limited' }) + .where(eq(liveSessions.id, event.sessionId)) + return + } const company = await lookupScarfCompany(ip, key) await db .update(liveSessions) @@ -174,6 +203,9 @@ export async function listLiveVisitors() { and( gt(liveTabs.lastSeen, new Date(now - 5 * 60_000)), gt(liveSessions.createdAt, new Date(now - LIVE_RETENTION_MS)), + sql`length(${liveTabs.page}) <= 400`, + sql`${liveTabs.page} ~ ${livePublicPagePattern.source}`, + sql`${liveTabs.page} !~ ${liveForbiddenPagePattern.source}`, ), ) .orderBy(desc(liveTabs.lastSeen)) @@ -185,6 +217,7 @@ export async function listLiveVisitors() { truncated: rows.length > 500, tabs: rows.slice(0, 500).map((row) => ({ ...row, + activity: row.activity.filter((event) => isLivePage(event.page)), active: row.leftAt === null && now - row.lastSeen.getTime() < LIVE_TIMEOUT_MS, lookupStatus: diff --git a/src/utils/analytics/live.ts b/src/utils/analytics/live.ts index 8fe61dbb7..b34f1b897 100644 --- a/src/utils/analytics/live.ts +++ b/src/utils/analytics/live.ts @@ -27,14 +27,15 @@ export const liveActionSchema = z.enum([ ]) // Only public content paths. Never store a query, fragment, or private route. +export const livePublicPagePattern = + /^\/(?:$|(?:query|router|start|table|db|form|store|virtual|pacer|hotkeys|ai|charts|intent|devtools|config|cli|highlight|markdown|ranger|redact|mcp|workflow)(?:\/|$)|(?:blog|partners|application-starter|libraries|stack|stats)(?:\/|$)|(?:brand-guide|community-projects|ethos|explore|feedback-leaderboard|learn|maintainers|paid-support|privacy|showcase|showcases|support|tenets|terms|workshops)\/?$)/ +export const liveForbiddenPagePattern = /[?#\\\s%]|\.\./ + export function isLivePage(page: string) { return ( page.length <= 400 && - /^\/(?:$|(?:query|router|start|table|db|form|store|virtual|pacer|hotkeys|ai|charts|intent|devtools|config|cli|highlight|markdown|ranger|redact|mcp|workflow)(?:\/|$)|(?:blog|partners|application-starter|libraries|stack|stats|showcase)(?:\/|$)|(?:brand-guide|community-projects|ethos|explore|feedback-leaderboard|learn|maintainers|paid-support|privacy|showcases|support|tenets|terms|workshops)\/?$)/.test( - page, - ) && - !/[?#\\\s%]/.test(page) && - !page.includes('..') + livePublicPagePattern.test(page) && + !liveForbiddenPagePattern.test(page) ) } diff --git a/src/utils/analytics/scarf-lookup.server.ts b/src/utils/analytics/scarf-lookup.server.ts index 4e6c43fb9..50a04e2a7 100644 --- a/src/utils/analytics/scarf-lookup.server.ts +++ b/src/utils/analytics/scarf-lookup.server.ts @@ -7,6 +7,8 @@ export async function getLiveConfig() { const host = await getHostRuntimeEnv() const key = host?.SCARF_LOOKUP_KEY ?? process.env.SCARF_LOOKUP_KEY return { + requestLimiter: host?.LIVE_PRESENCE_RATE_LIMITER, + lookupLimiter: host?.LIVE_LOOKUP_RATE_LIMITER, enabled: (host?.LIVE_VISITORS_ENABLED ?? process.env.LIVE_VISITORS_ENABLED) === 'true', diff --git a/tests/live-intake.test.ts b/tests/live-intake.test.ts index 6832c3b37..976daa96e 100644 --- a/tests/live-intake.test.ts +++ b/tests/live-intake.test.ts @@ -1,6 +1,7 @@ import assert from 'node:assert/strict' import { test } from 'node:test' import { receiveLiveEvent } from '../src/utils/analytics/live.server.ts' +import { runWithHostRuntimeEnv } from '../src/server/runtime/host.server.ts' function request(body = '{}', headers: Record = {}) { return new Request('https://tanstack.com/_a/live', { @@ -42,16 +43,81 @@ test('enabled intake bounds and validates data before using the database', async if (previous === undefined) delete process.env.LIVE_VISITORS_ENABLED else process.env.LIVE_VISITORS_ENABLED = previous }) - assert.equal((await receiveLiveEvent(request('x'.repeat(1025)))).status, 413) - assert.equal((await receiveLiveEvent(request('invalid'))).status, 400) + assert.equal((await allowedIntake(request('x'.repeat(1025)))).status, 413) + assert.equal((await allowedIntake(request('invalid'))).status, 400) assert.equal( - (await receiveLiveEvent(request(JSON.stringify({ page: '/admin' })))) - .status, + (await allowedIntake(request(JSON.stringify({ page: '/admin' })))).status, 400, ) assert.equal( - (await receiveLiveEvent(request('{}', { 'cf-connecting-ip': 'not-an-ip' }))) + (await allowedIntake(request('{}', { 'cf-connecting-ip': 'not-an-ip' }))) .status, 204, ) }) + +function allowedIntake(incoming: Request) { + return runWithHostRuntimeEnv( + { + LIVE_PRESENCE_RATE_LIMITER: { limit: async () => ({ success: true }) }, + LIVE_LOOKUP_RATE_LIMITER: { limit: async () => ({ success: true }) }, + }, + () => receiveLiveEvent(incoming), + ) +} + +test('edge throttling rejects requests before reading bodies or using the database', async (t) => { + const previous = process.env.LIVE_VISITORS_ENABLED + process.env.LIVE_VISITORS_ENABLED = 'true' + t.after(() => { + if (previous === undefined) delete process.env.LIVE_VISITORS_ENABLED + else process.env.LIVE_VISITORS_ENABLED = previous + }) + const incoming = request('not JSON') + const response = await runWithHostRuntimeEnv( + { + LIVE_PRESENCE_RATE_LIMITER: { + limit: async ({ key }: { key: string }) => { + assert.equal(key, '192.0.2.1') + return { success: false } + }, + }, + LIVE_LOOKUP_RATE_LIMITER: { limit: async () => ({ success: true }) }, + }, + () => receiveLiveEvent(incoming), + ) + assert.equal(response.status, 429) + assert.equal(response.headers.get('Retry-After'), '60') + assert.equal(incoming.bodyUsed, false) +}) + +test('enabled intake fails closed when its edge limiter is unavailable', async (t) => { + const previous = process.env.LIVE_VISITORS_ENABLED + process.env.LIVE_VISITORS_ENABLED = 'true' + t.after(() => { + if (previous === undefined) delete process.env.LIVE_VISITORS_ENABLED + else process.env.LIVE_VISITORS_ENABLED = previous + }) + assert.equal( + (await runWithHostRuntimeEnv({}, () => receiveLiveEvent(request()))).status, + 503, + ) + const incoming = request('not JSON') + assert.equal( + ( + await runWithHostRuntimeEnv( + { + LIVE_PRESENCE_RATE_LIMITER: { + limit: async () => { + throw new Error('Unavailable') + }, + }, + LIVE_LOOKUP_RATE_LIMITER: { limit: async () => ({ success: true }) }, + }, + () => receiveLiveEvent(incoming), + ) + ).status, + 503, + ) + assert.equal(incoming.bodyUsed, false) +}) diff --git a/tests/live-lookup-limit.test.ts b/tests/live-lookup-limit.test.ts new file mode 100644 index 000000000..4ac08c246 --- /dev/null +++ b/tests/live-lookup-limit.test.ts @@ -0,0 +1,87 @@ +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { runInNewContext } from 'node:vm' +import { build } from 'esbuild' +import { createRequire } from 'node:module' +const require = createRequire(import.meta.url) + +const bundle = build({ + entryPoints: ['src/utils/analytics/live.server.ts'], + bundle: true, + write: false, + platform: 'node', + format: 'iife', + globalName: 'live', + plugins: [ + { + name: 'lookup-dependencies', + setup(build) { + build.onResolve( + { + filter: + /^(~\/db\/client|~\/utils\/roles.server|~\/server\/runtime\/host.server|\.\/scarf-lookup.server)$/, + }, + (args) => ({ path: args.path, namespace: 'live-test' }), + ) + build.onLoad({ filter: /.*/, namespace: 'live-test' }, (args) => ({ + contents: + args.path === '~/db/client' + ? ` + export const db = { + insert() { return { values() { return { onConflictDoNothing() { return { async returning() { return [{ id: 'session' }] } } }, async onConflictDoUpdate() {} } } } }, + update() { return { set(value) { updates.push(value); return { async where() {} } } } }, + } + ` + : args.path === './scarf-lookup.server' + ? ` + export async function getLiveConfig() { return {} } + export async function lookupScarfCompany() { calls.push('lookup'); return null } + ` + : args.path === '~/utils/roles.server' + ? 'export async function requireAdmin() {}' + : 'export function scheduleHostRuntimeTask() { return false }', + })) + }, + }, + ], +}) + +test('new sessions do not call Scarf when the separate lookup budget is exhausted', async () => { + const calls: string[] = [] + const updates: unknown[] = [] + const result = await bundle + const api = runInNewContext(`${result.outputFiles[0].text}\nlive`, { + require, + calls, + updates, + console, + TextEncoder, + TextDecoder, + URL, + setTimeout, + clearTimeout, + }) + await api.recordLiveEvent( + { + sessionId: 'session', + tabId: 'tab', + sequence: 1, + page: '/query/latest', + visible: true, + kind: 'update', + }, + '192.0.2.1', + 'test-key', + { + async limit({ key }: { key: string }) { + assert.equal(key, '192.0.2.1') + return { success: false } + }, + }, + ) + assert.deepEqual(calls, []) + assert.equal( + JSON.stringify(updates), + JSON.stringify([{ lookupStatus: 'rate_limited' }]), + ) +}) diff --git a/tests/live-presence.test.ts b/tests/live-presence.test.ts index e4ecfb6ab..e66f89d7b 100644 --- a/tests/live-presence.test.ts +++ b/tests/live-presence.test.ts @@ -197,3 +197,17 @@ test('an expired session rotates its tab too, and restored pages rejoin', async assert.notEqual(last?.tabId, first.tabId) visitor.stop() }) + +test('showcase collection excludes drafts, submission, and editing paths', () => { + assert.equal(isLivePage('/showcase'), true) + assert.equal(isLivePage('/showcase/'), true) + assert.equal(isLivePage('/showcases'), true) + for (const page of [ + '/showcase/submit', + '/showcase/submit/', + '/showcase/edit/123', + '/showcase/491fa786-94d5-4d42-8c54-df45c7a7c066', + ]) { + assert.equal(isLivePage(page), false, page) + } +}) diff --git a/tests/live-scheduled.test.ts b/tests/live-scheduled.test.ts new file mode 100644 index 000000000..2f91ef710 --- /dev/null +++ b/tests/live-scheduled.test.ts @@ -0,0 +1,51 @@ +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { runInNewContext } from 'node:vm' +import { build } from 'esbuild' + +test('failed presence cleanup does not interrupt scheduled workflow processing', async () => { + const calls: string[] = [] + const bundle = await build({ + entryPoints: ['src/server/scheduled.server.ts'], + bundle: true, + write: false, + format: 'iife', + globalName: 'scheduled', + plugins: [ + { + name: 'scheduled-dependencies', + setup(build) { + build.onResolve({ filter: /.*/ }, (args) => + args.importer ? { path: args.path, namespace: 'stub' } : undefined, + ) + build.onLoad({ filter: /.*/, namespace: 'stub' }, () => ({ + contents: ` + export async function pruneLiveVisitors() { calls.push('presence-prune'); throw new Error('Database unavailable') } + export async function interruptExpiredBuilderProjectRunLeases() { calls.push('builder-leases'); return 0 } + export async function reconcileBuilderProjectSnapshotReservations() { return {} } + export async function pruneBuilderProjectSnapshotStorage() { return {} } + export async function pruneDeletedBuilderProjects() { return 0 } + export async function reconcileWorkflowRuntimeStore() { return {} } + export async function materializeWorkflowSchedules() { calls.push('materialize'); return [] } + export const workflowRuntime = { async sweep() { calls.push('workflow-sweep'); return { summary: {} } } } + export const WORKFLOW_RUNTIME_MAX_DURATION_MS = 1000 + export const WORKFLOW_RUNTIME_MIN_REMAINING_MS = 100 + export async function pruneStaleCacheRows() {} + export async function refreshHomepageNpmStatsSummary() {} + export async function refreshGitHubOrgStats() {} + `, + })) + }, + }, + ], + }) + const api = runInNewContext(`${bundle.outputFiles[0].text}\nscheduled`, { + calls, + console: { log() {}, warn() {}, error() {} }, + }) + await api.runScheduledTasks('* * * * *', Date.now()) + assert.ok(calls.includes('presence-prune')) + assert.ok(calls.includes('builder-leases')) + assert.ok(calls.includes('materialize')) + assert.ok(calls.includes('workflow-sweep')) +}) diff --git a/wrangler.jsonc b/wrangler.jsonc index 424946cfc..b0fcb35e3 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -18,6 +18,18 @@ "KODY_ORIGIN": "https://kody.codes", "INCLUDED_MODEL": "@cf/moonshotai/kimi-k2.6", }, + "ratelimits": [ + { + "name": "LIVE_PRESENCE_RATE_LIMITER", + "namespace_id": "144201", + "simple": { "limit": 600, "period": 60 }, + }, + { + "name": "LIVE_LOOKUP_RATE_LIMITER", + "namespace_id": "144202", + "simple": { "limit": 30, "period": 60 }, + }, + ], "limits": { "cpu_ms": 300000, },