feat(housekeeping): orchestrate partial providers

src/features/housekeeping/foundation/providers/orchestrate.ts

@@ -0,0 +1,55 @@
+import type {
+	HousekeepingCapabilityContext,
+	HousekeepingErrorCode,
+} from "../contracts";
+import { createCorrelationId } from "../correlation";
+import { type HousekeepingProvider, runProvider } from "./run-provider";
+
+export interface ProviderPolicy<T> {
+	timeoutMs: number;
+	perProviderLimit: number;
+	combinedLimit: number;
+	sort(items: readonly T[]): readonly T[];
+	dedupeKey(item: T): string;
+}
+
+export interface ProviderBatchResult<T> {
+	items: readonly T[];
+	errors: readonly { providerId: string; code: HousekeepingErrorCode }[];
+	correlationId: string;
+}
+
+export async function orchestrateProviders<T>(
+	providers: readonly HousekeepingProvider<T>[],
+	context: HousekeepingCapabilityContext,
+	policy: ProviderPolicy<T>,
+): Promise<ProviderBatchResult<T>> {
+	const providerResults = await Promise.all(
+		providers.map((provider) =>
+			runProvider(provider, context, policy.timeoutMs),
+		),
+	);
+	const items: T[] = [];
+	const seen = new Set<string>();
+	const errors: { providerId: string; code: HousekeepingErrorCode }[] = [];
+
+	for (const result of providerResults) {
+		if (result.error !== undefined) {
+			errors.push({ providerId: result.providerId, code: result.error });
+		}
+
+		for (const item of result.items.slice(0, policy.perProviderLimit)) {
+			if (items.length >= policy.combinedLimit) break;
+			const key = policy.dedupeKey(item);
+			if (seen.has(key)) continue;
+			seen.add(key);
+			items.push(item);
+		}
+	}
+
+	return {
+		items: policy.sort(items),
+		errors,
+		correlationId: createCorrelationId(),
+	};
+}

src/features/housekeeping/foundation/providers/run-provider.test.ts

@@ -0,0 +1,78 @@
+import { describe, expect, it, vi } from "vitest";
+import {
+	anyCapability,
+	type HousekeepingCapabilityContext,
+	ok,
+} from "../contracts";
+import { type HousekeepingProvider, runProvider } from "./run-provider";
+
+function context(granted: readonly string[]): HousekeepingCapabilityContext {
+	return {
+		actor: { id: 42, username: "operator", rank: 999 },
+		isSuperAdmin: false,
+		has: (slug) => granted.includes(slug),
+		hasAny: (...slugs) => slugs.some((slug) => granted.includes(slug)),
+		hasAll: (...slugs) => slugs.every((slug) => granted.includes(slug)),
+	};
+}
+
+describe("runProvider", () => {
+	it("aborts a slow provider at its 2,000 ms deadline and returns TIMEOUT", async () => {
+		vi.useFakeTimers();
+		let observedAbort = false;
+		const provider: HousekeepingProvider<string> = {
+			id: "slow",
+			capability: anyCapability("housekeeping.read"),
+			run: async (_context, signal) =>
+				await new Promise((_, reject) => {
+					signal.addEventListener("abort", () => {
+						observedAbort = true;
+						reject(new Error("aborted"));
+					});
+				}),
+		};
+
+		const result = runProvider(provider, context(["housekeeping.read"]), 2_000);
+		await vi.advanceTimersByTimeAsync(2_000);
+
+		expect(await result).toEqual({
+			providerId: "slow",
+			items: [],
+			error: "TIMEOUT",
+		});
+		expect(observedAbort).toBe(true);
+		vi.useRealTimers();
+	});
+
+	it("skips a provider whose required capability is absent", async () => {
+		let invoked = false;
+		const provider: HousekeepingProvider<string> = {
+			id: "restricted",
+			capability: anyCapability("housekeeping.read"),
+			run: async () => {
+				invoked = true;
+				return ok(["private"], "provider-correlation");
+			},
+		};
+
+		expect(await runProvider(provider, context([]), 2_000)).toEqual({
+			providerId: "restricted",
+			items: [],
+		});
+		expect(invoked).toBe(false);
+	});
+
+	it("maps one thrown provider failure to one INTERNAL partial error", async () => {
+		const provider: HousekeepingProvider<string> = {
+			id: "broken",
+			capability: anyCapability("housekeeping.read"),
+			run: async () => {
+				throw new Error("database unavailable");
+			},
+		};
+
+		expect(
+			await runProvider(provider, context(["housekeeping.read"]), 2_000),
+		).toEqual({ providerId: "broken", items: [], error: "INTERNAL" });
+	});
+});

src/features/housekeeping/foundation/providers/run-provider.ts

@@ -0,0 +1,62 @@
+import { satisfiesCapability } from "../capability-context";
+import type {
+	CapabilityRequirement,
+	HousekeepingCapabilityContext,
+	HousekeepingErrorCode,
+	HousekeepingResult,
+} from "../contracts";
+
+export interface HousekeepingProvider<T> {
+	id: string;
+	capability: CapabilityRequirement;
+	run(
+		context: HousekeepingCapabilityContext,
+		signal: AbortSignal,
+	): Promise<HousekeepingResult<readonly T[]>>;
+}
+
+export interface ProviderRunResult<T> {
+	providerId: string;
+	items: readonly T[];
+	error?: HousekeepingErrorCode;
+}
+
+export async function runProvider<T>(
+	provider: HousekeepingProvider<T>,
+	context: HousekeepingCapabilityContext,
+	timeoutMs: number,
+): Promise<ProviderRunResult<T>> {
+	if (!satisfiesCapability(context, provider.capability)) {
+		return { providerId: provider.id, items: [] };
+	}
+
+	const controller = new AbortController();
+	let timeout: ReturnType<typeof setTimeout> | undefined;
+	const run = provider
+		.run(context, controller.signal)
+		.then(
+			(result): ProviderRunResult<T> =>
+				result.ok
+					? { providerId: provider.id, items: result.data }
+					: { providerId: provider.id, items: [], error: result.error.code },
+		)
+		.catch(
+			(): ProviderRunResult<T> => ({
+				providerId: provider.id,
+				items: [],
+				error: "INTERNAL",
+			}),
+		);
+	const timeoutResult = new Promise<ProviderRunResult<T>>((resolve) => {
+		timeout = setTimeout(() => {
+			controller.abort();
+			resolve({ providerId: provider.id, items: [], error: "TIMEOUT" });
+		}, timeoutMs);
+	});
+
+	try {
+		return await Promise.race([run, timeoutResult]);
+	} finally {
+		if (timeout !== undefined) clearTimeout(timeout);
+	}
+}