diff --git a/deploy/self-hosting/.env.example b/deploy/self-hosting/.env.example index 58a30d487..13cc82030 100644 --- a/deploy/self-hosting/.env.example +++ b/deploy/self-hosting/.env.example @@ -620,6 +620,9 @@ FLUXER_DISCOVERY_ENABLED=true #FLUXER_CACHE_PURGE_HTTP_ENDPOINT= #FLUXER_CACHE_PURGE_HTTP_TOKEN= #FLUXER_CACHE_PURGE_HTTP_TIMEOUT_MS=10000 +#FLUXER_CACHE_PURGE_CONTROLLER_ENDPOINT= +#FLUXER_CACHE_PURGE_CONTROLLER_TOKEN= +#FLUXER_CACHE_PURGE_CONTROLLER_TIMEOUT_MS=10000 # media-proxy limits and timeouts. #FLUXER_MEDIA_PROXY_READ_ONLY=false diff --git a/deploy/self-hosting/docker-compose.yml b/deploy/self-hosting/docker-compose.yml index 9c185311b..11f535b7a 100644 --- a/deploy/self-hosting/docker-compose.yml +++ b/deploy/self-hosting/docker-compose.yml @@ -97,6 +97,9 @@ x-fluxer-env: &fluxer-env FLUXER_CACHE_PURGE_HTTP_ENDPOINT: ${FLUXER_CACHE_PURGE_HTTP_ENDPOINT:-} FLUXER_CACHE_PURGE_HTTP_TOKEN: ${FLUXER_CACHE_PURGE_HTTP_TOKEN:-} FLUXER_CACHE_PURGE_HTTP_TIMEOUT_MS: ${FLUXER_CACHE_PURGE_HTTP_TIMEOUT_MS:-} + FLUXER_CACHE_PURGE_CONTROLLER_ENDPOINT: ${FLUXER_CACHE_PURGE_CONTROLLER_ENDPOINT:-} + FLUXER_CACHE_PURGE_CONTROLLER_TOKEN: ${FLUXER_CACHE_PURGE_CONTROLLER_TOKEN:-} + FLUXER_CACHE_PURGE_CONTROLLER_TIMEOUT_MS: ${FLUXER_CACHE_PURGE_CONTROLLER_TIMEOUT_MS:-} FLUXER_LIVEKIT_ENABLED: "${FLUXER_LIVEKIT_ENABLED:-true}" FLUXER_LIVEKIT_API_KEY: ${LIVEKIT_API_KEY:?set LIVEKIT_API_KEY in .env} diff --git a/fluxer_api/src/api/Config.ts b/fluxer_api/src/api/Config.ts index 3eef9596b..5c0ec81bd 100644 --- a/fluxer_api/src/api/Config.ts +++ b/fluxer_api/src/api/Config.ts @@ -374,6 +374,11 @@ export function buildAPIConfigFromMaster(master: MasterConfig): APIConfig { token: master.integrations.cache_purge.http.token, timeoutMs: master.integrations.cache_purge.http.timeout_ms, }, + controller: { + endpoint: master.integrations.cache_purge.controller.endpoint, + token: master.integrations.cache_purge.controller.token, + timeoutMs: master.integrations.cache_purge.controller.timeout_ms, + }, }, clamav: { enabled: master.integrations.clamav.enabled, diff --git a/fluxer_api/src/api/config/APIConfig.ts b/fluxer_api/src/api/config/APIConfig.ts index 2bdf8f044..e4f0099c5 100644 --- a/fluxer_api/src/api/config/APIConfig.ts +++ b/fluxer_api/src/api/config/APIConfig.ts @@ -19,13 +19,16 @@ export interface TrustedCallerConfig { buckets: Array; } +interface APICachePurgeTarget { + endpoint: string; + token: string; + timeoutMs: number; +} + export interface APICachePurgeConfig { adapter: CachePurgeAdapterName; - http: { - endpoint: string; - token: string; - timeoutMs: number; - }; + http: APICachePurgeTarget; + controller: APICachePurgeTarget; } interface APIGeoipFilesystemConfig { diff --git a/fluxer_api/src/api/infrastructure/CachePurgeAdapter.ts b/fluxer_api/src/api/infrastructure/CachePurgeAdapter.ts index 7e881da59..1cc225ad8 100644 --- a/fluxer_api/src/api/infrastructure/CachePurgeAdapter.ts +++ b/fluxer_api/src/api/infrastructure/CachePurgeAdapter.ts @@ -1,6 +1,7 @@ // SPDX-License-Identifier: AGPL-3.0-or-later import type {APICachePurgeConfig} from '@app/api/config/APIConfig'; +import {createControllerCachePurgeAdapter} from '@app/api/infrastructure/ControllerCachePurgeAdapter'; import {createHttpCachePurgeAdapter} from '@app/api/infrastructure/HttpCachePurgeAdapter'; import {createNoneCachePurgeAdapter} from '@app/api/infrastructure/NoneCachePurgeAdapter'; import type {CachePurgeAdapterName} from '@fluxer/config/src/MasterConfig'; @@ -17,6 +18,7 @@ export interface CachePurgeAdapter { const CACHE_PURGE_ADAPTERS = { none: createNoneCachePurgeAdapter, http: createHttpCachePurgeAdapter, + controller: createControllerCachePurgeAdapter, } satisfies Record CachePurgeAdapter>; export function createCachePurgeAdapter(config: APICachePurgeConfig): CachePurgeAdapter { diff --git a/fluxer_api/src/api/infrastructure/ControllerCachePurgeAdapter.test.ts b/fluxer_api/src/api/infrastructure/ControllerCachePurgeAdapter.test.ts new file mode 100644 index 000000000..0b70270a5 --- /dev/null +++ b/fluxer_api/src/api/infrastructure/ControllerCachePurgeAdapter.test.ts @@ -0,0 +1,82 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import type {APICachePurgeConfig} from '@app/api/config/APIConfig'; +import {createControllerCachePurgeAdapter} from '@app/api/infrastructure/ControllerCachePurgeAdapter'; +import {afterEach, beforeEach, describe, expect, it, type MockInstance, vi} from 'vitest'; + +const ENDPOINT = 'https://controller.test/v1/orgs/org-a/sites/site-b/purge'; + +const CONFIG: APICachePurgeConfig = { + adapter: 'controller', + http: {endpoint: '', token: '', timeoutMs: 10_000}, + controller: {endpoint: ENDPOINT, token: 'pak_test', timeoutMs: 10_000}, +}; + +function prefixes(count: number): Array { + return Array.from({length: count}, (_, index) => `media.test/attachments/1/${index}/photo`); +} + +function sentPatterns(spy: MockInstance, call: number): Array { + const init = spy.mock.calls[call]![1]!; + return JSON.parse(String(init.body)).patterns; +} + +describe('createControllerCachePurgeAdapter', () => { + let fetchSpy: MockInstance; + + beforeEach(() => { + fetchSpy = vi.spyOn(globalThis, 'fetch'); + }); + + afterEach(() => { + fetchSpy.mockRestore(); + }); + + it('sends each prefix as a wildcard pattern with the bearer token', async () => { + fetchSpy.mockResolvedValue(new Response(null, {status: 202})); + const outcome = await createControllerCachePurgeAdapter(CONFIG).purge(['media.test/emojis/1']); + expect(outcome).toEqual({kind: 'purged'}); + const [url, init] = fetchSpy.mock.calls[0]!; + expect(url).toBe(ENDPOINT); + expect(init!.method).toBe('POST'); + expect(new Headers(init!.headers).get('Authorization')).toBe('Bearer pak_test'); + expect(JSON.parse(String(init!.body))).toEqual({patterns: ['media.test/emojis/1*'], soft: false}); + }); + + it('splits a batch into requests of at most 30 patterns', async () => { + fetchSpy.mockImplementation(async () => new Response(null, {status: 202})); + const outcome = await createControllerCachePurgeAdapter(CONFIG).purge(prefixes(65)); + expect(outcome).toEqual({kind: 'purged'}); + expect(fetchSpy).toHaveBeenCalledTimes(3); + expect(sentPatterns(fetchSpy, 0)).toHaveLength(30); + expect(sentPatterns(fetchSpy, 1)).toHaveLength(30); + expect(sentPatterns(fetchSpy, 2)).toHaveLength(5); + }); + + it('stops at the first chunk that fails so the whole batch is retried', async () => { + fetchSpy + .mockResolvedValueOnce(new Response(null, {status: 202})) + .mockResolvedValueOnce(new Response(null, {status: 503})); + const outcome = await createControllerCachePurgeAdapter(CONFIG).purge(prefixes(90)); + expect(outcome).toEqual({kind: 'failed', status: 503, error: null}); + expect(fetchSpy).toHaveBeenCalledTimes(2); + }); + + it('reports a 400 as rejected', async () => { + fetchSpy.mockResolvedValue(new Response(null, {status: 400})); + expect(await createControllerCachePurgeAdapter(CONFIG).purge(prefixes(1))).toEqual({ + kind: 'rejected', + status: 400, + }); + }); + + it('reports a network error as failed', async () => { + const error = new TypeError('fetch failed'); + fetchSpy.mockRejectedValue(error); + expect(await createControllerCachePurgeAdapter(CONFIG).purge(prefixes(1))).toEqual({ + kind: 'failed', + status: null, + error, + }); + }); +}); diff --git a/fluxer_api/src/api/infrastructure/ControllerCachePurgeAdapter.ts b/fluxer_api/src/api/infrastructure/ControllerCachePurgeAdapter.ts new file mode 100644 index 000000000..c21832e56 --- /dev/null +++ b/fluxer_api/src/api/infrastructure/ControllerCachePurgeAdapter.ts @@ -0,0 +1,45 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import type {APICachePurgeConfig} from '@app/api/config/APIConfig'; +import type {CachePurgeAdapter, CachePurgeOutcome} from '@app/api/infrastructure/CachePurgeAdapter'; +import * as FetchUtils from '@app/api/utils/FetchUtils'; + +const MAX_PATTERNS_PER_REQUEST = 30; + +export function createControllerCachePurgeAdapter(config: APICachePurgeConfig): CachePurgeAdapter { + async function purgeChunk(patterns: ReadonlyArray): Promise { + let response: Response; + try { + response = await fetch(config.controller.endpoint, { + method: 'POST', + headers: {'Content-Type': 'application/json', Authorization: `Bearer ${config.controller.token}`}, + body: JSON.stringify({patterns, soft: false}), + redirect: 'manual', + signal: AbortSignal.timeout(config.controller.timeoutMs), + }); + } catch (error) { + return {kind: 'failed', status: null, error}; + } + FetchUtils.discardResponseBody(response.body, response.status); + if (response.ok) { + return {kind: 'purged'}; + } + if (response.status === 400 || response.status === 422) { + return {kind: 'rejected', status: response.status}; + } + return {kind: 'failed', status: response.status, error: null}; + } + + return { + async purge(prefixes: ReadonlyArray): Promise { + const patterns = prefixes.map((prefix) => `${prefix}*`); + for (let start = 0; start < patterns.length; start += MAX_PATTERNS_PER_REQUEST) { + const outcome = await purgeChunk(patterns.slice(start, start + MAX_PATTERNS_PER_REQUEST)); + if (outcome.kind !== 'purged') { + return outcome; + } + } + return {kind: 'purged'}; + }, + }; +} diff --git a/fluxer_api/src/api/worker/tests/ProcessCachePurgeQueue.test.ts b/fluxer_api/src/api/worker/tests/ProcessCachePurgeQueue.test.ts index 2623296ae..7ce7a7325 100644 --- a/fluxer_api/src/api/worker/tests/ProcessCachePurgeQueue.test.ts +++ b/fluxer_api/src/api/worker/tests/ProcessCachePurgeQueue.test.ts @@ -64,7 +64,11 @@ describe('processCachePurgeQueue', () => { let previousMedia: string; beforeEach(() => { - previousCachePurge = {adapter: Config.cachePurge.adapter, http: Config.cachePurge.http}; + previousCachePurge = { + adapter: Config.cachePurge.adapter, + http: Config.cachePurge.http, + controller: Config.cachePurge.controller, + }; previousMedia = Config.endpoints.media; Config.endpoints.media = MEDIA; Config.cachePurge.adapter = 'http'; diff --git a/fluxer_docs/src/content/docs/operator/configuration.mdx b/fluxer_docs/src/content/docs/operator/configuration.mdx index 6a7b3facc..b988441d0 100644 --- a/fluxer_docs/src/content/docs/operator/configuration.mdx +++ b/fluxer_docs/src/content/docs/operator/configuration.mdx @@ -1218,11 +1218,11 @@ Default empty. The YouTube Data API key. Also read by `unfurl`, which accepts `Y #### `FLUXER_CACHE_PURGE_ADAPTER` -Default `none`. How cached copies of deleted or replaced media are purged. `none` or `http`. Anything else fails startup. With `none`, nothing is queued or sent. +Default `none`. How cached copies of deleted or replaced media are purged. `none`, `http` or `controller`. Anything else fails startup. With `none`, nothing is queued or sent. #### `FLUXER_CACHE_PURGE_HTTP_ENDPOINT` -Default empty. Required when the adapter is `http`. Must be an absolute `http` or `https` URL without credentials, or startup fails. While purges are queued, the worker posts JSON `{"exact": [...], "prefix": [...]}` here every 10 seconds. Redirects are not followed. Any 2xx marks the batch done. Your endpoint then purges those URLs from every proxy that caches media. A `400` or `422` makes the worker resend each URL alone, and a URL refused alone moves to the `cache_purge:rejected` key. Any other failure is retried on a later run. +Default empty. Required when the adapter is `http`. Must be an absolute `http` or `https` URL without credentials, or startup fails. While purges are queued, the worker posts JSON `{"prefixes": [...]}` here every 10 seconds, each a media URL prefix without its scheme. Redirects are not followed. Any 2xx marks the batch done. Your endpoint then purges every cached URL under those prefixes from each proxy that caches media. A `400` or `422` moves the batch to the `cache_purge:rejected` key. Any other failure is retried on a later run. #### `FLUXER_CACHE_PURGE_HTTP_TOKEN` @@ -1232,6 +1232,18 @@ Default empty. The bearer token sent in the `Authorization` header of each purge Default `10000`. Purge request timeout. Accepts 1000 to 10000. The range is checked only when the adapter is `http`. +#### `FLUXER_CACHE_PURGE_CONTROLLER_ENDPOINT` + +Default empty. Required when the adapter is `controller`. The purge URL of the Controller edge site that serves your media, in the form `https:///v1/orgs//sites//purge`. Must be an absolute `https` URL without credentials, or startup fails. While purges are queued, the worker posts JSON `{"patterns": [...], "soft": false}` here every 10 seconds, at most 30 patterns per request, each a media URL prefix without its scheme followed by `*`. Any 2xx marks the patterns done. A `400` or `422` moves the batch to the `cache_purge:rejected` key. Any other failure retries the whole batch on a later run. + +#### `FLUXER_CACHE_PURGE_CONTROLLER_TOKEN` + +Default empty. Required when the adapter is `controller`. The Controller API key, sent as a bearer token. A token with spaces, line breaks or non-ASCII characters fails startup. + +#### `FLUXER_CACHE_PURGE_CONTROLLER_TIMEOUT_MS` + +Default `10000`. Timeout of each purge request. Accepts 1000 to 10000. The range is checked only when the adapter is `controller`. + #### `FLUXER_GEOIP_DB_PATH` Default empty. The GeoIP database. A filesystem path, or an `s3://bucket/key` URL whose `download_path` query parameter is mandatory and must be absolute. Read by `api` and `worker`. diff --git a/packages/config/src/ConfigLoader.ts b/packages/config/src/ConfigLoader.ts index 07d480b6d..8890d850c 100644 --- a/packages/config/src/ConfigLoader.ts +++ b/packages/config/src/ConfigLoader.ts @@ -218,6 +218,11 @@ function defaultConfig(): MasterConfig { token: '', timeout_ms: 10_000, }, + controller: { + endpoint: '', + token: '', + timeout_ms: 10_000, + }, }, blocklist_feeds: {}, breached_password_check: {}, @@ -484,26 +489,37 @@ function normalizeAppOriginAliases(config: MasterConfig): void { ]; } -function validateCachePurgeConfig(config: MasterConfig): void { - const cachePurge = config.integrations.cache_purge; - if (cachePurge.adapter !== 'http') { - return; - } - requireString(cachePurge.http.endpoint, 'FLUXER_CACHE_PURGE_HTTP_ENDPOINT'); - const endpoint = URL.parse(cachePurge.http.endpoint); +function validateCachePurgeTarget( + target: {endpoint: string; token: string; timeout_ms: number}, + envPrefix: string, + protocols: ReadonlyArray, +): void { + requireString(target.endpoint, `${envPrefix}_ENDPOINT`); + const endpoint = URL.parse(target.endpoint); if ( endpoint === null || - (endpoint.protocol !== 'http:' && endpoint.protocol !== 'https:') || + !protocols.includes(endpoint.protocol) || endpoint.username !== '' || endpoint.password !== '' ) { - throw new Error('FLUXER_CACHE_PURGE_HTTP_ENDPOINT must be an absolute http or https URL without credentials'); + const schemes = protocols.map((protocol) => protocol.slice(0, -1)).join(' or '); + throw new Error(`${envPrefix}_ENDPOINT must be an absolute ${schemes} URL without credentials`); } - requireString(cachePurge.http.token, 'FLUXER_CACHE_PURGE_HTTP_TOKEN'); - if (!/^[\x21-\x7e]+$/u.test(cachePurge.http.token)) { - throw new Error('FLUXER_CACHE_PURGE_HTTP_TOKEN must contain only visible ASCII characters'); + requireString(target.token, `${envPrefix}_TOKEN`); + if (!/^[\x21-\x7e]+$/u.test(target.token)) { + throw new Error(`${envPrefix}_TOKEN must contain only visible ASCII characters`); + } + assertIntegerInRange(target.timeout_ms, `${envPrefix}_TIMEOUT_MS`, 1_000, 10_000); +} + +function validateCachePurgeConfig(config: MasterConfig): void { + const cachePurge = config.integrations.cache_purge; + if (cachePurge.adapter === 'http') { + validateCachePurgeTarget(cachePurge.http, 'FLUXER_CACHE_PURGE_HTTP', ['http:', 'https:']); + } + if (cachePurge.adapter === 'controller') { + validateCachePurgeTarget(cachePurge.controller, 'FLUXER_CACHE_PURGE_CONTROLLER', ['https:']); } - assertIntegerInRange(cachePurge.http.timeout_ms, 'FLUXER_CACHE_PURGE_HTTP_TIMEOUT_MS', 1_000, 10_000); } function validateStoreProductSlots( diff --git a/packages/config/src/MasterConfig.ts b/packages/config/src/MasterConfig.ts index ea241b5d5..e76b0568e 100644 --- a/packages/config/src/MasterConfig.ts +++ b/packages/config/src/MasterConfig.ts @@ -5,7 +5,7 @@ import type {DerivedEndpoints} from '@fluxer/config/src/EndpointDerivation'; type RuntimeEnv = 'development' | 'production' | 'test'; type DatabaseBackend = 'postgres' | 'cassandra'; type PublicScheme = 'http' | 'https'; -export const CACHE_PURGE_ADAPTER_NAMES = ['none', 'http'] as const; +export const CACHE_PURGE_ADAPTER_NAMES = ['none', 'http', 'controller'] as const; export type CachePurgeAdapterName = (typeof CACHE_PURGE_ADAPTER_NAMES)[number]; export const STORE_PRODUCT_SLOT_NAMES = ['monthly', 'yearly', 'gift_1_month', 'gift_1_year'] as const; export type StoreProductSlotName = (typeof STORE_PRODUCT_SLOT_NAMES)[number]; @@ -243,6 +243,11 @@ export interface MasterConfig { token: string; timeout_ms: number; }; + controller: { + endpoint: string; + token: string; + timeout_ms: number; + }; }; blocklist_feeds: { enabled?: boolean; diff --git a/packages/config/src/__tests__/ConfigLoader.test.ts b/packages/config/src/__tests__/ConfigLoader.test.ts index a70288460..cc3fd82ab 100644 --- a/packages/config/src/__tests__/ConfigLoader.test.ts +++ b/packages/config/src/__tests__/ConfigLoader.test.ts @@ -603,6 +603,46 @@ describe('ConfigLoader', () => { } }); + test('rejects the controller cache purge adapter without an endpoint', async () => { + stubMinimalEnv({FLUXER_CACHE_PURGE_ADAPTER: 'controller'}); + await expect(loadConfig()).rejects.toThrow('FLUXER_CACHE_PURGE_CONTROLLER_ENDPOINT is required'); + }); + + test('rejects a controller cache purge endpoint that is not https', async () => { + stubMinimalEnv({ + FLUXER_CACHE_PURGE_ADAPTER: 'controller', + FLUXER_CACHE_PURGE_CONTROLLER_ENDPOINT: 'http://controller.test/v1/orgs/a/sites/b/purge', + FLUXER_CACHE_PURGE_CONTROLLER_TOKEN: 'pak_test', + }); + await expect(loadConfig()).rejects.toThrow( + 'FLUXER_CACHE_PURGE_CONTROLLER_ENDPOINT must be an absolute https URL without credentials', + ); + }); + + test('rejects the controller cache purge adapter without a token', async () => { + stubMinimalEnv({ + FLUXER_CACHE_PURGE_ADAPTER: 'controller', + FLUXER_CACHE_PURGE_CONTROLLER_ENDPOINT: 'https://controller.test/v1/orgs/a/sites/b/purge', + }); + await expect(loadConfig()).rejects.toThrow('FLUXER_CACHE_PURGE_CONTROLLER_TOKEN is required'); + }); + + test('accepts a complete controller cache purge configuration', async () => { + stubMinimalEnv({ + FLUXER_CACHE_PURGE_ADAPTER: 'controller', + FLUXER_CACHE_PURGE_CONTROLLER_ENDPOINT: 'https://controller.test/v1/orgs/a/sites/b/purge', + FLUXER_CACHE_PURGE_CONTROLLER_TOKEN: 'pak_test', + FLUXER_CACHE_PURGE_CONTROLLER_TIMEOUT_MS: '5000', + }); + const cachePurge = (await loadConfig()).integrations.cache_purge; + expect(cachePurge.adapter).toBe('controller'); + expect(cachePurge.controller).toEqual({ + endpoint: 'https://controller.test/v1/orgs/a/sites/b/purge', + token: 'pak_test', + timeout_ms: 5000, + }); + }); + test('leaves the cache purge settings unvalidated when the adapter is none', async () => { stubMinimalEnv({ FLUXER_CACHE_PURGE_HTTP_ENDPOINT: 'ftp://purge.internal/purge', diff --git a/packages/config/src/config_loader/EnvironmentOverrides.ts b/packages/config/src/config_loader/EnvironmentOverrides.ts index 4a22e5eed..30bd48d34 100644 --- a/packages/config/src/config_loader/EnvironmentOverrides.ts +++ b/packages/config/src/config_loader/EnvironmentOverrides.ts @@ -239,6 +239,12 @@ const NAMED_FLUXER_ENV_OVERRIDES: Record = { path: ['integrations', 'cache_purge', 'http', 'timeout_ms'], parse: parseInteger, }, + FLUXER_CACHE_PURGE_CONTROLLER_ENDPOINT: {path: ['integrations', 'cache_purge', 'controller', 'endpoint']}, + FLUXER_CACHE_PURGE_CONTROLLER_TOKEN: {path: ['integrations', 'cache_purge', 'controller', 'token']}, + FLUXER_CACHE_PURGE_CONTROLLER_TIMEOUT_MS: { + path: ['integrations', 'cache_purge', 'controller', 'timeout_ms'], + parse: parseInteger, + }, FLUXER_BLOCKLIST_FEEDS_ENABLED: {path: ['integrations', 'blocklist_feeds', 'enabled'], parse: parseBoolean}, FLUXER_BREACHED_PASSWORD_CHECK_ENABLED: { path: ['integrations', 'breached_password_check', 'enabled'],