Protect live presence intake and isolate retention cleanup

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,

src/components/admin/LiveVisitors.tsx

@@ -67,6 +67,7 @@ const lookupLabels: Record<string, string> = {
   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) {

src/db/schema.ts

@@ -3470,7 +3470,14 @@ export const liveSessions = pgTable(
     company:
       jsonb('company').$type<import('../utils/analytics/live').LiveCompany>(),
     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)],

src/server/runtime/host.server.ts

@@ -5,7 +5,13 @@ type StaticAssetService = {
   fetch: (input: RequestInfo | URL, init?: RequestInit) => Promise<Response>
 }
 
+export type HostRateLimiter = {
+  limit(options: { key: string }): Promise<{ success: boolean }>
+}
+
 type HostRuntimeEnv = Record<string, unknown> & {
+  LIVE_PRESENCE_RATE_LIMITER?: HostRateLimiter
+  LIVE_LOOKUP_RATE_LIMITER?: HostRateLimiter
   CLOUDFLARE_CACHE_PURGE_TOKEN?: string
   CLOUDFLARE_ZONE_ID?: string
   GITHUB_AUTH_TOKEN?: string

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),

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:

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)
   )
 }
 

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',

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<string, string> = {}) {
   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)
+})