feat(api): purge media through the controller edge api (#3454)

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,

fluxer_api/src/api/config/APIConfig.ts

@@ -19,13 +19,16 @@ export interface TrustedCallerConfig {
 	buckets: Array<string>;
 }
 
+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 {

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<CachePurgeAdapterName, (config: APICachePurgeConfig) => CachePurgeAdapter>;
 
 export function createCachePurgeAdapter(config: APICachePurgeConfig): CachePurgeAdapter {

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<string> {
+	return Array.from({length: count}, (_, index) => `media.test/attachments/1/${index}/photo`);
+}
+
+function sentPatterns(spy: MockInstance<typeof fetch>, call: number): Array<string> {
+	const init = spy.mock.calls[call]![1]!;
+	return JSON.parse(String(init.body)).patterns;
+}
+
+describe('createControllerCachePurgeAdapter', () => {
+	let fetchSpy: MockInstance<typeof fetch>;
+
+	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,
+		});
+	});
+});

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<string>): Promise<CachePurgeOutcome> {
+		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<string>): Promise<CachePurgeOutcome> {
+			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'};
+		},
+	};
+}

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