From 5dd82c568839e6918576b1abe919da6b6ceefb4a Mon Sep 17 00:00:00 2001 From: John Yeo Date: Wed, 1 Apr 2026 12:34:53 +0100 Subject: [PATCH] latest --- .agents/skills/edge-config/SKILL.md | 171 ++++++++ .../honoMiddlewares/requestBlockMiddleware.ts | 2 - server/src/init.ts | 13 +- .../admin/handleGetAdminOrgRequestBlock.ts | 2 +- .../misc/edgeConfig/edgeConfigRegistry.ts | 28 ++ .../misc/edgeConfig/edgeConfigStore.ts | 187 +++++++++ .../misc/requestBlocks/requestBlockStore.ts | 206 +--------- .../edge-config/edge-config-store.test.ts | 389 ++++++++++++++++++ 8 files changed, 806 insertions(+), 192 deletions(-) create mode 100644 .agents/skills/edge-config/SKILL.md create mode 100644 server/src/internal/misc/edgeConfig/edgeConfigRegistry.ts create mode 100644 server/src/internal/misc/edgeConfig/edgeConfigStore.ts create mode 100644 server/tests/unit/edge-config/edge-config-store.test.ts diff --git a/.agents/skills/edge-config/SKILL.md b/.agents/skills/edge-config/SKILL.md new file mode 100644 index 000000000..7f8a241f5 --- /dev/null +++ b/.agents/skills/edge-config/SKILL.md @@ -0,0 +1,171 @@ +--- +name: edge-config +description: Understand and create S3-backed edge configs with poll-based caching. Use when adding new runtime configs (feature rollouts, request blocking, operational toggles), debugging edge config polling, or working with the EdgeConfigStore factory. +--- + +# Edge Configs + +Edge configs are JSON files stored in S3 that are polled into server memory at a regular cadence. They let us change server behavior at runtime without deploys -- load shedding, feature rollouts, operational toggles, etc. + +## Architecture + +Each edge config has its own S3 file under the `admin/` prefix in the admin S3 bucket (`autumn-dev-server` / `autumn-prod-server`). Configs are independent -- a parse error in one does not affect another. + +### Key files + +| File | Purpose | +|------|---------| +| `server/src/internal/misc/edgeConfig/edgeConfigStore.ts` | `createEdgeConfigStore` factory -- generic S3 read/write, polling, status tracking | +| `server/src/internal/misc/edgeConfig/edgeConfigRegistry.ts` | Registry -- `registerEdgeConfig`, `startAllEdgeConfigPolling`, `stopAllEdgeConfigPolling` | +| `server/src/init.ts` | Calls `startAllEdgeConfigPolling` on boot, `stopAllEdgeConfigPolling` on shutdown | +| `server/src/external/aws/s3/adminS3Config.ts` | Bucket/region resolution (dev vs prod) | + +### Existing configs + +| Config | S3 Key | Module | +|--------|--------|--------| +| Request blocks | `admin/request-block-config.json` | `server/src/internal/misc/requestBlocks/requestBlockStore.ts` | + +## Lifecycle + +``` +Boot: + 1. Config module is imported (top-level side effect) -> calls registerEdgeConfig() + 2. init.ts calls startAllEdgeConfigPolling({ logger }) + 3. Each store: await refresh() (initial S3 fetch, blocks until done) + 4. Each store: setInterval(refresh, pollIntervalMs) + 5. Server starts accepting traffic + +Runtime: + - get() reads from in-memory cache (zero I/O, sync) + - refresh() runs every pollIntervalMs (fire-and-forget) + - writeToSource() writes to S3 + updates local cache immediately + +Shutdown: + - stopAllEdgeConfigPolling() clears all intervals +``` + +## Fail-Open Guarantee + +On **any** S3 failure (network error, auth error, bad JSON, schema mismatch), `runtimeConfig` resets to `defaultValue()`. This means: + +- No request is ever blocked due to infrastructure failure +- If S3 goes down, configs degrade to their default (empty/safe) state +- The store does NOT hold onto stale data -- it resets to default + +`NoSuchKey` (file doesn't exist yet) is treated as a normal empty state, not an error. + +## How to Add a New Edge Config + +### 1. Define the schema + +Create a schemas file with a Zod schema: + +```typescript +// server/src/internal/misc/featureRollouts/featureRolloutSchemas.ts +import { z } from "zod/v4"; + +export const FeatureRolloutConfigSchema = z.object({ + features: z.record(z.string(), z.object({ + percentage: z.number().min(0).max(100).default(0), + enabled: z.boolean().default(false), + })).default({}), +}); + +export type FeatureRolloutConfig = z.infer; +``` + +### 2. Create the store module + +```typescript +// server/src/internal/misc/featureRollouts/featureRolloutStore.ts +import { createEdgeConfigStore } from "@/internal/misc/edgeConfig/edgeConfigStore.js"; +import { registerEdgeConfig } from "@/internal/misc/edgeConfig/edgeConfigRegistry.js"; +import { + type FeatureRolloutConfig, + FeatureRolloutConfigSchema, +} from "./featureRolloutSchemas.js"; + +const store = createEdgeConfigStore({ + s3Key: "admin/feature-rollout-config.json", + schema: FeatureRolloutConfigSchema, + defaultValue: () => ({ features: {} }), + pollIntervalMs: 300_000, // 5 minutes +}); + +registerEdgeConfig({ store }); + +export const getFeatureRolloutConfig = () => store.get(); + +export const isFeatureEnabled = ({ + featureId, +}: { + featureId: string; +}): boolean => { + const feature = store.get().features[featureId]; + return feature?.enabled ?? false; +}; +``` + +### 3. Register the import in init.ts + +Add a side-effect import so the module runs `registerEdgeConfig` at boot: + +```typescript +// server/src/init.ts +import "./internal/misc/featureRollouts/featureRolloutStore.js"; +``` + +### 4. (Optional) Add the S3 key constant + +```typescript +// server/src/external/aws/s3/adminS3Config.ts +export const ADMIN_FEATURE_ROLLOUT_CONFIG_KEY = "admin/feature-rollout-config.json"; +``` + +### 5. Use it + +```typescript +import { isFeatureEnabled } from "@/internal/misc/featureRollouts/featureRolloutStore.js"; + +if (isFeatureEnabled({ featureId: "new-billing-engine" })) { + // new path +} +``` + +## Factory API Reference + +```typescript +const store = createEdgeConfigStore({ + s3Key: string, // S3 object key under admin bucket + schema: z.ZodType, // Zod schema for validation + defaultValue: () => T, // Factory for empty/safe default + pollIntervalMs?: number, // Poll interval (default: 60_000) + s3Client?: S3Client, // Optional DI for testing +}); + +store.get() // T -- in-memory cached config (sync, no I/O) +store.getStatus() // EdgeConfigStatus -- health/timing info +store.refresh({ logger? }) // Re-fetch from S3, update cache +store.startPolling({ logger? }) // Initial fetch + start interval +store.stopPolling() // Clear interval +store.readFromSource() // Direct S3 read (bypasses cache) +store.writeToSource({ config }) // S3 write + update cache immediately +``` + +## Testing + +Unit tests use the `s3Client` DI parameter to inject a mock: + +```typescript +const mockClient = { send: jest.fn(async () => ({ Body: ... })) } as unknown as S3Client; + +const store = createEdgeConfigStore({ + s3Key: "admin/test.json", + schema: MyConfigSchema, + defaultValue: () => ({ ... }), + s3Client: mockClient, +}); +``` + +Existing tests: `server/tests/unit/edge-config/edge-config-store.test.ts` diff --git a/server/src/honoMiddlewares/requestBlockMiddleware.ts b/server/src/honoMiddlewares/requestBlockMiddleware.ts index 09af1288d..e23bb0e2c 100644 --- a/server/src/honoMiddlewares/requestBlockMiddleware.ts +++ b/server/src/honoMiddlewares/requestBlockMiddleware.ts @@ -22,8 +22,6 @@ export const requestBlockMiddleware = async ( return; } - console.log("Entry:", entry); - if (entry.blockAll) { ctx.logger.warn("Rejecting blocked /v1 request (block all)"); diff --git a/server/src/init.ts b/server/src/init.ts index edd9cd9d7..215301ba9 100644 --- a/server/src/init.ts +++ b/server/src/init.ts @@ -12,9 +12,12 @@ import { } from "./db/pgHealthMonitor.js"; import { logger } from "./external/logtail/logtailUtils.js"; import { - startRequestBlockPolling, - stopRequestBlockPolling, -} from "./internal/misc/requestBlocks/requestBlockStore.js"; + startAllEdgeConfigPolling, + stopAllEdgeConfigPolling, +} from "./internal/misc/edgeConfig/edgeConfigRegistry.js"; + +// Edge config modules self-register on import +import "./internal/misc/requestBlocks/requestBlockStore.js"; import { warmupRegionalRedis } from "./external/redis/initRedis.js"; import { createHonoApp } from "./initHono.js"; import { otelSdk } from "./instrumentation.js"; @@ -28,7 +31,7 @@ const init = async () => { initPgHealthMonitor({ client: clientCritical }); await Promise.all([warmupRegionalRedis()]); - await startRequestBlockPolling({ logger }); + await startAllEdgeConfigPolling({ logger }); const PORT = process.env.SERVER_PORT ? Number.parseInt(process.env.SERVER_PORT) @@ -88,7 +91,7 @@ async function gracefulShutdown() { await otelSdk.shutdown(); } shutdownPgHealthMonitor(); - stopRequestBlockPolling(); + stopAllEdgeConfigPolling(); await Promise.all([ client.end(), clientCritical.end(), diff --git a/server/src/internal/admin/handleGetAdminOrgRequestBlock.ts b/server/src/internal/admin/handleGetAdminOrgRequestBlock.ts index fd272c740..ca6a2962d 100644 --- a/server/src/internal/admin/handleGetAdminOrgRequestBlock.ts +++ b/server/src/internal/admin/handleGetAdminOrgRequestBlock.ts @@ -12,7 +12,7 @@ export const handleGetAdminOrgRequestBlock = createRoute({ handler: async (c) => { const { org_id: orgId } = c.req.valid("param"); const status = getRuntimeRequestBlockStatus(); - const entry = await getOrgRequestBlockFromSource(orgId); + const entry = await getOrgRequestBlockFromSource({ orgId }); return c.json({ blockAll: entry?.blockAll ?? false, diff --git a/server/src/internal/misc/edgeConfig/edgeConfigRegistry.ts b/server/src/internal/misc/edgeConfig/edgeConfigRegistry.ts new file mode 100644 index 000000000..42f0d20cd --- /dev/null +++ b/server/src/internal/misc/edgeConfig/edgeConfigRegistry.ts @@ -0,0 +1,28 @@ +import type { Logger } from "@/external/logtail/logtailUtils.js"; + +type EdgeConfigLifecycle = { + startPolling: (options?: { logger?: Logger }) => Promise; + stopPolling: () => void; +}; + +const stores: EdgeConfigLifecycle[] = []; + +export const registerEdgeConfig = ({ + store, +}: { + store: EdgeConfigLifecycle; +}) => { + stores.push(store); +}; + +export const startAllEdgeConfigPolling = async ({ + logger, +}: { + logger?: Logger; +} = {}) => { + await Promise.all(stores.map((store) => store.startPolling({ logger }))); +}; + +export const stopAllEdgeConfigPolling = () => { + for (const store of stores) store.stopPolling(); +}; diff --git a/server/src/internal/misc/edgeConfig/edgeConfigStore.ts b/server/src/internal/misc/edgeConfig/edgeConfigStore.ts new file mode 100644 index 000000000..9e9616645 --- /dev/null +++ b/server/src/internal/misc/edgeConfig/edgeConfigStore.ts @@ -0,0 +1,187 @@ +import { ErrCode, ms } from "@autumn/shared"; +import type { S3Client } from "@aws-sdk/client-s3"; +import { GetObjectCommand, PutObjectCommand } from "@aws-sdk/client-s3"; +import type { z } from "zod/v4"; +import { getAdminS3Config } from "@/external/aws/s3/adminS3Config.js"; +import { getS3Client } from "@/external/aws/s3/initS3.js"; +import { getS3BodyAsString } from "@/external/aws/s3/s3Utils.js"; +import type { Logger } from "@/external/logtail/logtailUtils.js"; +import RecaseError from "@/utils/errorUtils.js"; + +export type EdgeConfigStatus = { + configured: boolean; + healthy: boolean; + lastFetchAt?: string; + lastSuccessAt?: string; + error?: string; +}; + +const nowIso = () => new Date().toISOString(); + +/** + * Factory that creates a typed, poll-based edge config backed by S3. + * Fail-open: any S3 error resets the in-memory config to `defaultValue()`. + */ +export const createEdgeConfigStore = ({ + s3Key, + schema, + defaultValue, + pollIntervalMs = ms.seconds(30), + s3Client: injectedS3Client, +}: { + s3Key: string; + schema: z.ZodType; + defaultValue: () => T; + pollIntervalMs?: number; + s3Client?: S3Client; +}) => { + let runtimeConfig: T = defaultValue(); + let runtimeStatus: EdgeConfigStatus = { + configured: false, + healthy: false, + error: "Edge config not yet initialized", + }; + let pollTimer: ReturnType | null = null; + + const getConfigLocation = () => { + const { bucket, region } = getAdminS3Config(); + return { + bucket, + region, + key: s3Key, + configured: Boolean(bucket && s3Key), + }; + }; + + const resolveClient = () => { + if (injectedS3Client) return injectedS3Client; + const { region } = getConfigLocation(); + return getS3Client({ region }); + }; + + const readFromSource = async (): Promise => { + const { bucket, key, configured } = getConfigLocation(); + + if (!configured || !bucket || !key) return defaultValue(); + + const client = resolveClient(); + try { + const response = await client.send( + new GetObjectCommand({ Bucket: bucket, Key: key }), + ); + + if (!response.Body) return defaultValue(); + + const raw = (await getS3BodyAsString({ body: response.Body })).trim(); + if (!raw) return defaultValue(); + + return schema.parse(JSON.parse(raw)); + } catch (error) { + const name = error instanceof Error ? error.name : ""; + if (name === "NoSuchKey") return defaultValue(); + throw error; + } + }; + + const writeToSource = async ({ config }: { config: T }) => { + const { bucket, key, configured } = getConfigLocation(); + + if (!configured || !bucket || !key) { + throw new RecaseError({ + message: "Edge config S3 is not configured", + code: ErrCode.InvalidRequest, + statusCode: 503, + }); + } + + const client = resolveClient(); + await client.send( + new PutObjectCommand({ + Bucket: bucket, + Key: key, + Body: JSON.stringify(config, null, 2), + ContentType: "application/json", + }), + ); + + runtimeConfig = config; + runtimeStatus = { + configured: true, + healthy: true, + lastFetchAt: nowIso(), + lastSuccessAt: nowIso(), + }; + }; + + const refresh = async ({ logger }: { logger?: Logger } = {}) => { + const { configured } = getConfigLocation(); + runtimeStatus = { + ...runtimeStatus, + configured, + lastFetchAt: nowIso(), + }; + + if (!configured) { + runtimeConfig = defaultValue(); + runtimeStatus = { + configured: false, + healthy: false, + lastFetchAt: runtimeStatus.lastFetchAt, + lastSuccessAt: runtimeStatus.lastSuccessAt, + error: "Edge config S3 is not configured", + }; + return; + } + + try { + const config = await readFromSource(); + runtimeConfig = config; + runtimeStatus = { + configured: true, + healthy: true, + lastFetchAt: runtimeStatus.lastFetchAt, + lastSuccessAt: nowIso(), + }; + } catch (error) { + runtimeConfig = defaultValue(); + runtimeStatus = { + configured: true, + healthy: false, + lastFetchAt: runtimeStatus.lastFetchAt, + lastSuccessAt: runtimeStatus.lastSuccessAt, + error: error instanceof Error ? error.message : "Failed to load config", + }; + logger?.warn(`Failed to refresh edge config "${s3Key}": ${error}`); + } + }; + + const startPolling = async ({ logger }: { logger?: Logger } = {}) => { + if (pollTimer) return; + + await refresh({ logger }); + pollTimer = setInterval(() => { + void refresh({ logger }); + }, pollIntervalMs); + }; + + const stopPolling = () => { + if (pollTimer) { + clearInterval(pollTimer); + pollTimer = null; + } + }; + + return { + get: () => runtimeConfig, + getStatus: () => runtimeStatus, + refresh, + startPolling, + stopPolling, + readFromSource, + writeToSource, + }; +}; + +export type EdgeConfigStore = ReturnType< + typeof createEdgeConfigStore +>; diff --git a/server/src/internal/misc/requestBlocks/requestBlockStore.ts b/server/src/internal/misc/requestBlocks/requestBlockStore.ts index 6c10683d3..30a3680f8 100644 --- a/server/src/internal/misc/requestBlocks/requestBlockStore.ts +++ b/server/src/internal/misc/requestBlocks/requestBlockStore.ts @@ -1,13 +1,6 @@ -import { ErrCode } from "@autumn/shared"; -import { GetObjectCommand, PutObjectCommand } from "@aws-sdk/client-s3"; -import { - ADMIN_REQUEST_BLOCK_CONFIG_KEY, - getAdminS3Config, -} from "@/external/aws/s3/adminS3Config.js"; -import { getS3Client } from "@/external/aws/s3/initS3.js"; -import { getS3BodyAsString } from "@/external/aws/s3/s3Utils.js"; -import type { Logger } from "@/external/logtail/logtailUtils.js"; -import RecaseError from "@/utils/errorUtils.js"; +import { ADMIN_REQUEST_BLOCK_CONFIG_KEY } from "@/external/aws/s3/adminS3Config.js"; +import { registerEdgeConfig } from "@/internal/misc/edgeConfig/edgeConfigRegistry.js"; +import { createEdgeConfigStore } from "@/internal/misc/edgeConfig/edgeConfigStore.js"; import { type RequestBlockConfig, RequestBlockConfigSchema, @@ -15,179 +8,32 @@ import { type RequestBlockUpdate, } from "./requestBlockSchemas.js"; -const POLL_INTERVAL_MS = 60_000; - -type RequestBlockStatus = { - configured: boolean; - healthy: boolean; - lastFetchAt?: string; - lastSuccessAt?: string; - error?: string; -}; - -const emptyConfig = (): RequestBlockConfig => ({ orgs: {} }); - -let runtimeConfig: RequestBlockConfig = emptyConfig(); -let runtimeStatus: RequestBlockStatus = { - configured: false, - healthy: false, - error: "Request block config is not configured", -}; -let pollTimer: ReturnType | null = null; - -const getConfigLocation = () => { - const { bucket, region } = getAdminS3Config(); - const key = ADMIN_REQUEST_BLOCK_CONFIG_KEY; - - return { - bucket, - key, - region, - configured: Boolean(bucket && key), - }; -}; - -const readConfigFromS3 = async (): Promise => { - const { bucket, key, configured, region } = getConfigLocation(); - - if (!configured || !bucket || !key) { - return emptyConfig(); - } - - const client = getS3Client({ region }); - try { - const response = await client.send( - new GetObjectCommand({ - Bucket: bucket, - Key: key, - }), - ); - - if (!response.Body) { - return emptyConfig(); - } - - const raw = (await getS3BodyAsString({ body: response.Body })).trim(); - if (!raw) { - return emptyConfig(); - } - - return RequestBlockConfigSchema.parse(JSON.parse(raw)); - } catch (error) { - const name = error instanceof Error ? error.name : ""; - if (name === "NoSuchKey") { - return emptyConfig(); - } - throw error; - } -}; - -const writeConfigToS3 = async (config: RequestBlockConfig) => { - const { bucket, key, configured, region } = getConfigLocation(); - - if (!configured || !bucket || !key) { - throw new RecaseError({ - message: "Request block config is not configured", - code: ErrCode.InvalidRequest, - statusCode: 503, - }); - } - - const client = getS3Client({ region }); - await client.send( - new PutObjectCommand({ - Bucket: bucket, - Key: key, - Body: JSON.stringify(config, null, 2), - ContentType: "application/json", - }), - ); -}; - const nowIso = () => new Date().toISOString(); -export const getRuntimeRequestBlockStatus = () => runtimeStatus; +const store = createEdgeConfigStore({ + s3Key: ADMIN_REQUEST_BLOCK_CONFIG_KEY, + schema: RequestBlockConfigSchema, + defaultValue: () => ({ orgs: {} }), +}); + +registerEdgeConfig({ store }); + +export const getRuntimeRequestBlockStatus = () => store.getStatus(); export const getRuntimeRequestBlockEntry = ( orgId: string, -): RequestBlockEntry | undefined => runtimeConfig.orgs[orgId]; - -export const getRuntimeRequestBlockConfig = () => runtimeConfig; - -export const refreshRequestBlockConfig = async ({ - logger, -}: { - logger?: Logger; -} = {}) => { - const { configured } = getConfigLocation(); - runtimeStatus = { - ...runtimeStatus, - configured, - lastFetchAt: nowIso(), - }; - - if (!configured) { - runtimeConfig = emptyConfig(); - runtimeStatus = { - configured: false, - healthy: false, - lastFetchAt: runtimeStatus.lastFetchAt, - lastSuccessAt: runtimeStatus.lastSuccessAt, - error: "Request block config is not configured", - }; - return; - } - - try { - const config = await readConfigFromS3(); - runtimeConfig = config; - runtimeStatus = { - configured: true, - healthy: true, - lastFetchAt: runtimeStatus.lastFetchAt, - lastSuccessAt: nowIso(), - }; - } catch (error) { - runtimeConfig = emptyConfig(); - runtimeStatus = { - configured: true, - healthy: false, - lastFetchAt: runtimeStatus.lastFetchAt, - lastSuccessAt: runtimeStatus.lastSuccessAt, - error: error instanceof Error ? error.message : "Failed to load config", - }; - logger?.warn(`Failed to refresh request block config: ${error}`); - } -}; - -export const startRequestBlockPolling = async ({ - logger, -}: { - logger?: Logger; -} = {}) => { - if (pollTimer) { - return; - } - - await refreshRequestBlockConfig({ logger }); - pollTimer = setInterval(() => { - void refreshRequestBlockConfig({ logger }); - }, POLL_INTERVAL_MS); -}; - -export const stopRequestBlockPolling = () => { - if (pollTimer) { - clearInterval(pollTimer); - pollTimer = null; - } -}; +): RequestBlockEntry | undefined => store.get().orgs[orgId]; export const getRequestBlockConfigFromSource = async () => { - return await readConfigFromS3(); + return await store.readFromSource(); }; -export const getOrgRequestBlockFromSource = async (orgId: string) => { - const config = await readConfigFromS3(); +export const getOrgRequestBlockFromSource = async ({ + orgId, +}: { + orgId: string; +}) => { + const config = await store.readFromSource(); return config.orgs[orgId]; }; @@ -200,7 +46,7 @@ export const updateOrgRequestBlockInSource = async ({ update: RequestBlockUpdate; updatedBy?: string; }) => { - const config = await readConfigFromS3(); + const config = await store.readFromSource(); const shouldDelete = !update.blockAll && update.blockedEndpoints.length === 0; if (shouldDelete) { @@ -214,15 +60,7 @@ export const updateOrgRequestBlockInSource = async ({ }; } - await writeConfigToS3(config); - - runtimeConfig = config; - runtimeStatus = { - configured: true, - healthy: true, - lastFetchAt: nowIso(), - lastSuccessAt: nowIso(), - }; + await store.writeToSource({ config }); return config.orgs[orgId]; }; diff --git a/server/tests/unit/edge-config/edge-config-store.test.ts b/server/tests/unit/edge-config/edge-config-store.test.ts new file mode 100644 index 000000000..846068093 --- /dev/null +++ b/server/tests/unit/edge-config/edge-config-store.test.ts @@ -0,0 +1,389 @@ +import { afterEach, describe, expect, jest, test } from "bun:test"; +import type { S3Client } from "@aws-sdk/client-s3"; +import { z } from "zod/v4"; +import { createEdgeConfigStore } from "@/internal/misc/edgeConfig/edgeConfigStore.js"; + +const TestConfigSchema = z.object({ + enabled: z.boolean().default(false), + message: z.string().default("hello"), +}); + +type TestConfig = z.infer; + +const defaultConfig = (): TestConfig => ({ enabled: false, message: "hello" }); + +const createMockS3Client = ({ + getResponse, +}: { + getResponse: () => { + Body?: { transformToString: () => Promise } | null; + }; +}): S3Client => { + const sendFn = jest.fn(async (command: unknown) => { + const commandName = + command?.constructor?.name ?? (command as { name?: string })?.name; + + if (commandName === "GetObjectCommand") { + return getResponse(); + } + + if (commandName === "PutObjectCommand") { + return {}; + } + + throw new Error(`Unexpected command: ${commandName}`); + }); + + return { send: sendFn } as unknown as S3Client; +}; + +const makeBody = (data: unknown) => ({ + Body: { + transformToString: async () => JSON.stringify(data), + }, +}); + +const makeNoSuchKeyError = () => { + const error = new Error("NoSuchKey"); + error.name = "NoSuchKey"; + return error; +}; + +describe("createEdgeConfigStore", () => { + let store: ReturnType>; + + afterEach(() => { + store?.stopPolling(); + }); + + describe("initial fetch via startPolling", () => { + test("populates get() with parsed config from S3", async () => { + const mockClient = createMockS3Client({ + getResponse: () => makeBody({ enabled: true, message: "from-s3" }), + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + s3Client: mockClient, + }); + + await store.startPolling(); + + expect(store.get()).toEqual({ enabled: true, message: "from-s3" }); + expect(store.getStatus().healthy).toBe(true); + expect(store.getStatus().configured).toBe(true); + expect(store.getStatus().lastSuccessAt).toBeDefined(); + }); + + test("uses defaultValue before startPolling is called", () => { + const mockClient = createMockS3Client({ + getResponse: () => makeBody({ enabled: true, message: "from-s3" }), + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + s3Client: mockClient, + }); + + expect(store.get()).toEqual(defaultConfig()); + }); + }); + + describe("fail-open behavior", () => { + test("returns defaultValue when S3 throws a network error", async () => { + const mockClient = createMockS3Client({ + getResponse: () => { + throw new Error("NetworkingError: socket hang up"); + }, + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + s3Client: mockClient, + }); + + await store.startPolling(); + + expect(store.get()).toEqual(defaultConfig()); + expect(store.getStatus().healthy).toBe(false); + expect(store.getStatus().error).toContain("NetworkingError"); + }); + + test("returns defaultValue when S3 file does not exist (NoSuchKey)", async () => { + const mockClient = createMockS3Client({ + getResponse: () => { + throw makeNoSuchKeyError(); + }, + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + s3Client: mockClient, + }); + + await store.startPolling(); + + expect(store.get()).toEqual(defaultConfig()); + expect(store.getStatus().healthy).toBe(true); + }); + + test("returns defaultValue when S3 body is malformed JSON", async () => { + const mockClient = createMockS3Client({ + getResponse: () => ({ + Body: { + transformToString: async () => "not-valid-json{{{", + }, + }), + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + s3Client: mockClient, + }); + + await store.startPolling(); + + expect(store.get()).toEqual(defaultConfig()); + expect(store.getStatus().healthy).toBe(false); + }); + + test("returns defaultValue when S3 body is null", async () => { + const mockClient = createMockS3Client({ + getResponse: () => ({ Body: null }), + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + s3Client: mockClient, + }); + + await store.startPolling(); + + expect(store.get()).toEqual(defaultConfig()); + expect(store.getStatus().healthy).toBe(true); + }); + + test("returns defaultValue when S3 body is empty string", async () => { + const mockClient = createMockS3Client({ + getResponse: () => ({ + Body: { transformToString: async () => " " }, + }), + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + s3Client: mockClient, + }); + + await store.startPolling(); + + expect(store.get()).toEqual(defaultConfig()); + expect(store.getStatus().healthy).toBe(true); + }); + + test("returns defaultValue when schema validation fails", async () => { + const strictSchema = z.object({ + enabled: z.boolean(), + message: z.string(), + requiredField: z.string(), + }); + + const mockClient = createMockS3Client({ + getResponse: () => + makeBody({ enabled: true, message: "no-required-field" }), + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: strictSchema as unknown as z.ZodType, + defaultValue: defaultConfig, + s3Client: mockClient, + }); + + await store.startPolling(); + + expect(store.get()).toEqual(defaultConfig()); + expect(store.getStatus().healthy).toBe(false); + }); + }); + + describe("refresh", () => { + test("updates cached config when S3 content changes", async () => { + let callCount = 0; + const mockClient = createMockS3Client({ + getResponse: () => { + callCount++; + if (callCount === 1) { + return makeBody({ enabled: false, message: "first" }); + } + return makeBody({ enabled: true, message: "second" }); + }, + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + s3Client: mockClient, + }); + + await store.startPolling(); + expect(store.get().message).toBe("first"); + + await store.refresh(); + expect(store.get().message).toBe("second"); + expect(store.get().enabled).toBe(true); + }); + }); + + describe("writeToSource", () => { + test("updates local cache immediately after write", async () => { + const mockClient = createMockS3Client({ + getResponse: () => makeBody(defaultConfig()), + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + s3Client: mockClient, + }); + + await store.startPolling(); + expect(store.get().enabled).toBe(false); + + await store.writeToSource({ + config: { enabled: true, message: "written" }, + }); + + expect(store.get()).toEqual({ enabled: true, message: "written" }); + expect(store.getStatus().healthy).toBe(true); + }); + + test("calls S3 PutObject with correct payload", async () => { + const mockClient = createMockS3Client({ + getResponse: () => makeBody(defaultConfig()), + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + s3Client: mockClient, + }); + + await store.writeToSource({ + config: { enabled: true, message: "test-write" }, + }); + + const sendFn = ( + mockClient as unknown as { send: ReturnType } + ).send; + const calls = sendFn.mock.calls; + const lastCall = calls[calls.length - 1]?.[0]; + expect(lastCall?.constructor?.name).toBe("PutObjectCommand"); + }); + }); + + describe("readFromSource", () => { + test("returns fresh data from S3 without updating cache", async () => { + let callCount = 0; + const mockClient = createMockS3Client({ + getResponse: () => { + callCount++; + return makeBody({ + enabled: callCount > 1, + message: `call-${callCount}`, + }); + }, + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + s3Client: mockClient, + }); + + await store.startPolling(); + expect(store.get().message).toBe("call-1"); + + const fresh = await store.readFromSource(); + expect(fresh.message).toBe("call-2"); + expect(store.get().message).toBe("call-1"); + }); + }); + + describe("polling lifecycle", () => { + test("double startPolling does not create a second interval", async () => { + let callCount = 0; + const mockClient = createMockS3Client({ + getResponse: () => { + callCount++; + return makeBody(defaultConfig()); + }, + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + pollIntervalMs: 50, + s3Client: mockClient, + }); + + await store.startPolling(); + await store.startPolling(); + + const countAfterStart = callCount; + + await new Promise((resolve) => setTimeout(resolve, 130)); + + const countAfterWait = callCount; + const intervalFetchCount = countAfterWait - countAfterStart; + + expect(intervalFetchCount).toBeGreaterThanOrEqual(1); + expect(intervalFetchCount).toBeLessThanOrEqual(3); + }); + + test("stopPolling prevents further refreshes", async () => { + let callCount = 0; + const mockClient = createMockS3Client({ + getResponse: () => { + callCount++; + return makeBody(defaultConfig()); + }, + }); + + store = createEdgeConfigStore({ + s3Key: "admin/test-config.json", + schema: TestConfigSchema, + defaultValue: defaultConfig, + pollIntervalMs: 50, + s3Client: mockClient, + }); + + await store.startPolling(); + store.stopPolling(); + + const countAfterStop = callCount; + await new Promise((resolve) => setTimeout(resolve, 150)); + + expect(callCount).toBe(countAfterStop); + }); + }); +});