Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions cloudflare_workers/api/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ import { app as replication } from '../../supabase/functions/_backend/public/rep
import { app as statistics } from '../../supabase/functions/_backend/public/statistics/index.ts'
import { app as translation } from '../../supabase/functions/_backend/public/translation.ts'
import { app as webhooks } from '../../supabase/functions/_backend/public/webhooks/index.ts'
import { app as cache_invalidate } from '../../supabase/functions/_backend/triggers/cache_invalidate.ts'
import { app as credit_usage_alerts } from '../../supabase/functions/_backend/triggers/credit_usage_alerts.ts'
import { app as cron_clean_orphan_images } from '../../supabase/functions/_backend/triggers/cron_clean_orphan_images.ts'
import { app as cron_clear_versions } from '../../supabase/functions/_backend/triggers/cron_clear_versions.ts'
Expand Down Expand Up @@ -151,6 +152,7 @@ appPrivate.route('/sso/verify-dns', sso_verify_dns)
const functionNameTriggers = 'triggers'
const appTriggers = createHono(functionNameTriggers, version)
appTriggers.route('/ok', ok)
appTriggers.route('/cache_invalidate', cache_invalidate)
appTriggers.route('/cron_email', cron_email)
appTriggers.route('/cron_clear_versions', cron_clear_versions)
appTriggers.route('/cron_clean_orphan_images', cron_clean_orphan_images)
Expand Down
5 changes: 4 additions & 1 deletion cloudflare_workers/api/wrangler.jsonc
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,10 @@
"binding": "AI"
},
"vars": {
"ENV_NAME": "capgo_api-prod"
"ENV_NAME": "capgo_api-prod",
// Regional plugin workers for /updates colo-cache invalidation
// fan-out (authenticated with the shared API_SECRET).
"PLUGIN_INVALIDATE_URLS": "https://plugin.eu.capgo.app,https://plugin.na.capgo.app,https://plugin.sa.capgo.app,https://plugin.oc.capgo.app,https://plugin.jp.capgo.app,https://plugin.as.capgo.app,https://plugin.hk.capgo.app,https://plugin.me.capgo.app,https://plugin.af.capgo.app"
},
"routes": [
{
Expand Down
2 changes: 2 additions & 0 deletions cloudflare_workers/plugin/index.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import { app as channel_self } from '../../supabase/functions/_backend/plugins/channel_self.ts'
import { app as stats } from '../../supabase/functions/_backend/plugins/stats.ts'
import { app as updates } from '../../supabase/functions/_backend/plugins/updates.ts'
import { app as cache_invalidate } from '../../supabase/functions/_backend/private/cache_invalidate.ts'
import { app as latency } from '../../supabase/functions/_backend/private/latency.ts'
import { app as ok } from '../../supabase/functions/_backend/public/ok.ts'
import { createAllCatch, createHono, useCors } from '../../supabase/functions/_backend/utils/hono.ts'
Expand Down Expand Up @@ -32,6 +33,7 @@ app.route('/updates', updates)
app.route('/stats', stats)
app.route('/ok', ok)
app.route('/latency', latency)
app.route('/cache_invalidate', cache_invalidate)

createAllCatch(app, functionName)

Expand Down
24 changes: 24 additions & 0 deletions cloudflare_workers/plugin/wrangler.jsonc
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@
"prod_eu": {
"name": "capgo_plugin-eu-prod",
"vars": {
// Colo cache for /updates: flip to "on" per region to enable.
"UPDATES_CACHE_MODE": "off",
"ENV_NAME": "capgo_plugin-eu-prod"
},
"observability": {
Expand Down Expand Up @@ -143,6 +145,8 @@
"prod_me": {
"name": "capgo_plugin-me-prod",
"vars": {
// Colo cache for /updates: flip to "on" per region to enable.
"UPDATES_CACHE_MODE": "off",
"ENV_NAME": "capgo_plugin-me-prod"
},
"observability": {
Expand Down Expand Up @@ -238,6 +242,8 @@
"prod_hk": {
"name": "capgo_plugin-hk-prod",
"vars": {
// Colo cache for /updates: flip to "on" per region to enable.
"UPDATES_CACHE_MODE": "off",
"ENV_NAME": "capgo_plugin-hk-prod"
},
"observability": {
Expand Down Expand Up @@ -333,6 +339,8 @@
"prod_jp": {
"name": "capgo_plugin-jp-prod",
"vars": {
// Colo cache for /updates: flip to "on" per region to enable.
"UPDATES_CACHE_MODE": "off",
"ENV_NAME": "capgo_plugin-jp-prod"
},
"observability": {
Expand Down Expand Up @@ -428,6 +436,8 @@
"prod_as": {
"name": "capgo_plugin-as-prod",
"vars": {
// Colo cache for /updates: flip to "on" per region to enable.
"UPDATES_CACHE_MODE": "off",
"ENV_NAME": "capgo_plugin-as-prod"
},
"observability": {
Expand Down Expand Up @@ -523,6 +533,8 @@
"prod_na": {
"name": "capgo_plugin-na-prod",
"vars": {
// Colo cache for /updates: flip to "on" per region to enable.
"UPDATES_CACHE_MODE": "off",
"ENV_NAME": "capgo_plugin-na-prod"
},
"observability": {
Expand Down Expand Up @@ -618,6 +630,8 @@
"prod_af": {
"name": "capgo_plugin-af-prod",
"vars": {
// Colo cache for /updates: flip to "on" per region to enable.
"UPDATES_CACHE_MODE": "off",
"ENV_NAME": "capgo_plugin-af-prod"
},
"placement": {
Expand Down Expand Up @@ -713,6 +727,8 @@
"prod_oc": {
"name": "capgo_plugin-oc-prod",
"vars": {
// Colo cache for /updates: flip to "on" per region to enable.
"UPDATES_CACHE_MODE": "off",
"ENV_NAME": "capgo_plugin-oc-prod"
},
"observability": {
Expand Down Expand Up @@ -808,6 +824,8 @@
"prod_sa": {
"name": "capgo_plugin-sa-prod",
"vars": {
// Colo cache for /updates: flip to "on" per region to enable.
"UPDATES_CACHE_MODE": "off",
"ENV_NAME": "capgo_plugin-sa-prod"
},
"observability": {
Expand Down Expand Up @@ -891,6 +909,8 @@
"preprod": {
"name": "capgo_plugin-eu-preprod",
"vars": {
// Colo cache for /updates: flip to "on" per region to enable.
"UPDATES_CACHE_MODE": "off",
"ENV_NAME": "capgo_plugin-eu-preprod"
},
"observability": {
Expand Down Expand Up @@ -982,6 +1002,8 @@
"alpha": {
"name": "capgo_plugin-alpha",
"vars": {
// Colo cache for /updates: flip to "on" per region to enable.
"UPDATES_CACHE_MODE": "off",
"ENV_NAME": "capgo_plugin-alpha"
},
"observability": {
Expand Down Expand Up @@ -1077,6 +1099,8 @@
"local": {
"name": "capgo_plugin-local",
"vars": {
// Colo cache for /updates: flip to "on" per region to enable.
"UPDATES_CACHE_MODE": "off",
"ENV_NAME": "capgo_plugin-local",
"CAPGO_PREVENT_BACKGROUND_FUNCTIONS": "true"
},
Expand Down
38 changes: 38 additions & 0 deletions supabase/functions/_backend/private/cache_invalidate.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
// Colo cache invalidation endpoint, mounted on every regional plugin worker.
//
// The plugin workers are placement-pinned, so a request to each regional
// domain lands in the colo whose cache serves that region's devices. The
// cache_invalidate trigger handler fans a token bump out to all regions;
// each bump makes every cached /updates payload of the app unreachable
// (see updates_colo_cache.ts). Auth is a dedicated shared secret so this
// stays independent from the API-secret used by Supabase triggers.

import type { MiddlewareKeyVariables } from '../utils/hono.ts'
import { Hono } from 'hono/tiny'
import { BRES, middlewareAPISecret, parseBody, quickError } from '../utils/hono.ts'
import { cloudlog } from '../utils/logging.ts'
import { bumpAppCacheToken, isUpdatesCacheEnabled } from '../utils/updates_colo_cache.ts'

export const MAX_INVALIDATE_APPS = 100

export const app = new Hono<MiddlewareKeyVariables>()

app.post('/', middlewareAPISecret, async (c) => {
Comment thread
riderx marked this conversation as resolved.
Comment thread
riderx marked this conversation as resolved.
const body = await parseBody<{ app_ids?: unknown }>(c)
const appIds = Array.isArray(body.app_ids)
? body.app_ids.filter((appId): appId is string => typeof appId === 'string' && appId.length > 0)
: []
if (appIds.length === 0)
throw quickError(400, 'missing_app_ids', 'app_ids must be a non-empty array of strings')
// Loud rejection instead of silent truncation: the fan-out chunks to this
// size, so anything larger is a caller bug that would leave caches stale.
if (appIds.length > MAX_INVALIDATE_APPS)
throw quickError(400, 'too_many_app_ids', `app_ids is limited to ${MAX_INVALIDATE_APPS} per request`)

// Bump even when the cache mode is off so entries from a prior "on"
// window can never be served stale after a toggle.
const results = await Promise.all(appIds.map(appId => bumpAppCacheToken(c, appId)))
const bumped = results.filter(Boolean).length
cloudlog({ requestId: c.get('requestId'), message: 'updates cache invalidated', appIds, bumped, enabled: isUpdatesCacheEnabled(c) })
return c.json({ ...BRES, bumped })
})
87 changes: 87 additions & 0 deletions supabase/functions/_backend/triggers/cache_invalidate.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
// Fan-out of /updates colo-cache invalidations to every regional plugin
// worker.
//
// Called by the invalidate_updates_cache() database trigger (via pg_net)
// whenever a row that feeds the update hot path changes (channels,
// channel_devices, apps, app_versions, orgs, stripe_info). The plugin
// workers are placement-pinned, so one POST per regional domain reaches
// every colo that caches this app — end-to-end reaction is ~1s from the
// database commit, with the cache TTL as backstop when a call is lost.

import type { MiddlewareKeyVariables } from '../utils/hono.ts'
import { Hono } from 'hono/tiny'
import { MAX_INVALIDATE_APPS } from '../private/cache_invalidate.ts'
import { BRES, middlewareAPISecret, parseBody } from '../utils/hono.ts'
import { cloudlog, cloudlogErr, serializeError } from '../utils/logging.ts'
import { existInEnv, getEnv } from '../utils/utils.ts'

const FANOUT_TIMEOUT_MS = 5000
const FANOUT_CHUNK_SIZE = MAX_INVALIDATE_APPS

export function chunkAppIds(appIds: string[], size = FANOUT_CHUNK_SIZE): string[][] {
const unique = [...new Set(appIds)]
const chunks: string[][] = []
for (let i = 0; i < unique.length; i += size)
chunks.push(unique.slice(i, i + size))
return chunks
}

export function parsePluginInvalidateUrls(raw: string): string[] {
return raw
.split(',')
.map(url => url.trim().replace(/\/$/, ''))
.filter(Boolean)
}

export const app = new Hono<MiddlewareKeyVariables>()

app.post('/', middlewareAPISecret, async (c) => {
Comment thread
riderx marked this conversation as resolved.
const body = await parseBody<{ app_ids?: unknown }>(c)
const appIds = Array.isArray(body.app_ids)
? body.app_ids.filter((appId): appId is string => typeof appId === 'string' && appId.length > 0)
: []
if (appIds.length === 0) {
cloudlog({ requestId: c.get('requestId'), message: 'cache invalidate fanout skipped (no app_ids)' })
return c.json(BRES)
}

if (!existInEnv(c, 'PLUGIN_INVALIDATE_URLS')) {
// Soft-skip: invalidation is an accelerator, the cache TTL is the backstop.
cloudlog({ requestId: c.get('requestId'), message: 'cache invalidate fanout skipped (missing env)', appIds })
return c.json(BRES)
}

const urls = parsePluginInvalidateUrls(getEnv(c, 'PLUGIN_INVALIDATE_URLS'))
// Same intercommunication secret that authenticated this request.
const secret = getEnv(c, 'API_SECRET')
const chunks = chunkAppIds(appIds)
const results = await Promise.all(urls.map(async (url) => {
try {
// Chunked to the route's per-request cap: nothing is ever silently
// dropped, a large org just becomes a few sequential calls per region.
for (const chunk of chunks) {
const response = await fetch(`${url}/cache_invalidate`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'apisecret': secret,
},
body: JSON.stringify({ app_ids: chunk }),
signal: AbortSignal.timeout(FANOUT_TIMEOUT_MS),
})
if (!response.ok) {
cloudlogErr({ requestId: c.get('requestId'), message: 'cache invalidate fanout failed', url, status: response.status })
return false
}
}
return true
}
catch (e) {
cloudlogErr({ requestId: c.get('requestId'), message: 'cache invalidate fanout error', url, error: serializeError(e) })
return false
}
}))
const succeeded = results.filter(Boolean).length
cloudlog({ requestId: c.get('requestId'), message: 'cache invalidate fanout done', appIds, regions: urls.length, succeeded })
return c.json({ ...BRES, regions: urls.length, succeeded })
})
5 changes: 5 additions & 0 deletions supabase/functions/_backend/utils/cloudflare.ts
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,11 @@ export type Bindings = {
NOTIFICATION_EVENTS?: AnalyticsEngineDataset
NOTIFICATION_QUEUE?: Queue
DB_STOREAPPS: D1Database
// /updates colo cache + targeted invalidation (see utils/updates_colo_cache.ts)
UPDATES_CACHE_MODE?: string
UPDATES_CACHE_TTL_SECONDS?: string
UPDATES_CACHE_NEGATIVE_TTL_SECONDS?: string
PLUGIN_INVALIDATE_URLS?: string
CHANNEL_SELF_STORE?: KVNamespace
PLUGIN_NOTIFICATION_QUEUE?: KVNamespace
LOCAL_READ_REPLICA_SUPABASE_DB_URL?: string
Expand Down
Loading
Loading