From 249711005d7846e31ea9bdce029cabc0966f0d7e Mon Sep 17 00:00:00 2001 From: jckail Date: Thu, 9 Jul 2026 12:50:32 -0700 Subject: [PATCH] Add a real aggregator gateway behind TravelProviderGateway (Phase 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First real provider integration: replace the simulated-only sync path with an HTTP loyalty-data aggregator adapter, activated by config. - core: HttpAggregatorTravelProviderGateway — POSTs {providerId, membershipNumber, credential?} to {baseUrl}/v1/balance with bearer auth, validates the response, forwards transient credentials for one-time use, optionally scoped to supportedProviderIds. Injectable fetch; 5 unit tests (incl. composite precedence over the simulated gateway). - refactor: shared buildTravelProviderGateway() factory composes aggregator (when configured) ahead of the simulated fallback; web + worker containers now use it instead of hand-composing (dropped their direct Composite/Simulated imports). - env: AGGREGATOR_API_URL / AGGREGATOR_API_KEY (web + worker). - infra: opt-in AGGREGATOR_API_KEY secret (`-c enableAggregator=true`) + AGGREGATOR_API_URL context on the web service and the sync task; secret ARN output. Default deploys unchanged. - docs: integrations.md (aggregator section + factory), .env-example. Verification: 128 tests pass (97 core incl. 5 new + 20 bot + 11 extension); typecheck + lint clean; cdk synth clean default + `-c enableAggregator=true`. Co-Authored-By: Claude Opus 4.8 (1M context) --- .env-example | 5 + apps/web/src/env.ts | 5 + apps/web/src/server/container.ts | 12 +- apps/worker/src/container.ts | 12 +- apps/worker/src/env.ts | 4 + docs/integrations.md | 35 ++++- infra/lib/app-stack.ts | 28 +++- packages/core/src/index.ts | 2 + .../infrastructure/providers/build-gateway.ts | 39 ++++++ ...http-aggregator-travel-provider-gateway.ts | 107 +++++++++++++++ .../core/test/http-aggregator-gateway.test.ts | 122 ++++++++++++++++++ 11 files changed, 359 insertions(+), 12 deletions(-) create mode 100644 packages/core/src/infrastructure/providers/build-gateway.ts create mode 100644 packages/core/src/infrastructure/providers/http-aggregator-travel-provider-gateway.ts create mode 100644 packages/core/test/http-aggregator-gateway.test.ts diff --git a/.env-example b/.env-example index f2f1626..f539f08 100644 --- a/.env-example +++ b/.env-example @@ -38,6 +38,11 @@ CLERK_SECRET_KEY="" # FIRECRAWL_API_KEY="" # FIRECRAWL_BASE_URL="https://api.firecrawl.dev" +# Optional: loyalty-data aggregator for real balance syncs (falls back to the +# simulated gateway). See docs/integrations.md. +# AGGREGATOR_API_URL="https://api.vendor.example" +# AGGREGATOR_API_KEY="" + # Background worker (apps/worker) - scheduled syncs and email digests. # MAILER: ses | smtp | console (default console = log instead of sending) # MAILER="smtp" diff --git a/apps/web/src/env.ts b/apps/web/src/env.ts index a85364f..a38a85b 100644 --- a/apps/web/src/env.ts +++ b/apps/web/src/env.ts @@ -50,6 +50,9 @@ export const env = createEnv({ // Optional Firecrawl for deal / award-chart scraping. FIRECRAWL_API_KEY: z.string().min(1).optional(), FIRECRAWL_BASE_URL: z.url().optional(), + // Optional loyalty-data aggregator for real balance syncs. + AGGREGATOR_API_URL: z.url().optional(), + AGGREGATOR_API_KEY: z.string().min(1).optional(), }, client: { NEXT_PUBLIC_CLERK_PUBLISHABLE_KEY: z.string().min(1), @@ -70,6 +73,8 @@ export const env = createEnv({ AWS_REGION: process.env.AWS_REGION, FIRECRAWL_API_KEY: process.env.FIRECRAWL_API_KEY, FIRECRAWL_BASE_URL: process.env.FIRECRAWL_BASE_URL, + AGGREGATOR_API_URL: process.env.AGGREGATOR_API_URL, + AGGREGATOR_API_KEY: process.env.AGGREGATOR_API_KEY, }, skipValidation: !!process.env.SKIP_ENV_VALIDATION, emptyStringAsUndefined: true, diff --git a/apps/web/src/server/container.ts b/apps/web/src/server/container.ts index e3815ed..a7bf501 100644 --- a/apps/web/src/server/container.ts +++ b/apps/web/src/server/container.ts @@ -1,6 +1,6 @@ import { ChatWithAssistant, - CompositeTravelProviderGateway, + buildTravelProviderGateway, createDb, type Database, CreatePortfolioShare, @@ -42,7 +42,6 @@ import { RestoreLoyaltyAccount, RevokePortfolioShare, SeedDemoPortfolio, - SimulatedTravelProviderGateway, StubPageScraper, SyncAllLoyaltyAccounts, SyncLoyaltyAccount, @@ -151,9 +150,12 @@ function buildContainer(): Container { const shares = new DrizzlePortfolioShareRepository(db); const customValuations = new DrizzleCustomValuationRepository(db); const vault = buildVault(); - const gateway = new CompositeTravelProviderGateway([ - new SimulatedTravelProviderGateway(), - ]); + const gateway = buildTravelProviderGateway({ + aggregator: + env.AGGREGATOR_API_URL && env.AGGREGATOR_API_KEY + ? { baseUrl: env.AGGREGATOR_API_URL, apiKey: env.AGGREGATOR_API_KEY } + : undefined, + }); const syncLoyaltyAccount = new SyncLoyaltyAccount( loyaltyAccounts, diff --git a/apps/worker/src/container.ts b/apps/worker/src/container.ts index a3442ab..b1c0839 100644 --- a/apps/worker/src/container.ts +++ b/apps/worker/src/container.ts @@ -1,6 +1,6 @@ import { BuildPortfolioDigest, - CompositeTravelProviderGateway, + buildTravelProviderGateway, createDb, DrizzleBalanceSnapshotRepository, DrizzleLoyaltyAccountRepository, @@ -9,7 +9,6 @@ import { ListTripGoals, NullCredentialVault, OnePasswordConnectVault, - SimulatedTravelProviderGateway, SyncAllLoyaltyAccounts, SyncLoyaltyAccount, type CredentialVault, @@ -41,9 +40,12 @@ export function createContainer(env: WorkerEnv): WorkerContainer { }) : new NullCredentialVault(); - const gateway = new CompositeTravelProviderGateway([ - new SimulatedTravelProviderGateway(), - ]); + const gateway = buildTravelProviderGateway({ + aggregator: + env.AGGREGATOR_API_URL && env.AGGREGATOR_API_KEY + ? { baseUrl: env.AGGREGATOR_API_URL, apiKey: env.AGGREGATOR_API_KEY } + : undefined, + }); const syncOne = new SyncLoyaltyAccount( accounts, diff --git a/apps/worker/src/env.ts b/apps/worker/src/env.ts index 36fc9bd..72d7bb0 100644 --- a/apps/worker/src/env.ts +++ b/apps/worker/src/env.ts @@ -47,6 +47,10 @@ const envSchema = z.object({ OP_CONNECT_HOST: z.url().optional(), OP_CONNECT_TOKEN: z.string().min(1).optional(), + /** Optional loyalty-data aggregator for real balance syncs. */ + AGGREGATOR_API_URL: z.url().optional(), + AGGREGATOR_API_KEY: z.string().min(1).optional(), + /** Optional chat digest delivery — Slack / Discord incoming webhooks. */ SLACK_WEBHOOK_URL: z.url().optional(), DISCORD_WEBHOOK_URL: z.url().optional(), diff --git a/docs/integrations.md b/docs/integrations.md index fc8f407..4a9fb04 100644 --- a/docs/integrations.md +++ b/docs/integrations.md @@ -25,7 +25,7 @@ interface TravelProviderGateway { 1. Add the program to the catalog if it isn't there (one line). 2. Write an adapter in `packages/core/src/infrastructure/providers/`, e.g. `UnitedGateway implements TravelProviderGateway`, calling the airline's API (or an aggregator such as an NDC/loyalty API vendor). -3. Register it **ahead of the simulated fallback** in the web composition root (`apps/web/src/server/container.ts`): +3. Register it **ahead of the simulated fallback** — either add it to the shared `buildTravelProviderGateway()` factory (`packages/core/src/infrastructure/providers/build-gateway.ts`, used by both web and worker) or compose directly for a one-off: ```ts const gateway = new CompositeTravelProviderGateway([ @@ -36,6 +36,39 @@ const gateway = new CompositeTravelProviderGateway([ No use case, route, or schema changes are required — this is the open/closed principle at work. Balance history accumulates automatically because syncs append `balance_snapshot` rows. +### Shipped: the aggregator gateway + +`HttpAggregatorTravelProviderGateway` is a real adapter for a loyalty-data +aggregator's HTTP API — one integration to cover the long tail of programs. It's +composed by the shared `buildTravelProviderGateway()` factory (used by both the +web app and the worker), which registers the aggregator **ahead of** the +simulated fallback when it's configured: + +```ts +const gateway = buildTravelProviderGateway({ + aggregator: + env.AGGREGATOR_API_URL && env.AGGREGATOR_API_KEY + ? { baseUrl: env.AGGREGATOR_API_URL, apiKey: env.AGGREGATOR_API_KEY } + : undefined, +}); +``` + +Set `AGGREGATOR_API_URL` + `AGGREGATOR_API_KEY` (locally, or via CDK context +`-c enableAggregator=true -c aggregatorApiUrl=…` in AWS — the key becomes a +Secrets Manager placeholder) and real syncs route through it; everything else +falls back to the simulation. The vendor wire contract is: + +``` +POST {baseUrl}/v1/balance +Authorization: Bearer {apiKey} +{ "providerId", "membershipNumber", "credential"?: { "username", "secret" } } +→ 200 { "points": number } +``` + +Transient credentials the calling surface supplies are forwarded for one-time +use and never persisted. `supportedProviderIds` scopes the adapter to the +programs a given vendor actually covers. + ## Credential vaults PointUp **never stores raw provider passwords**. A loyalty account carries at most a `credentialRef` — an opaque pointer into a vault the user controls. At sync time the credential is resolved, used once in memory, and discarded. diff --git a/infra/lib/app-stack.ts b/infra/lib/app-stack.ts index 4801b67..0da29f0 100644 --- a/infra/lib/app-stack.ts +++ b/infra/lib/app-stack.ts @@ -148,6 +148,23 @@ export class AppStack extends cdk.Stack { : {}), }; + // Optional loyalty-data aggregator for real balance syncs. API key becomes + // a Secrets Manager placeholder gated by `-c enableAggregator=true`; the + // base URL is plain config: + // -c enableAggregator=true -c aggregatorApiUrl=https://api.vendor.example + const aggregatorSecret = this.node.tryGetContext("enableAggregator") + ? placeholderSecret( + "AggregatorApiKey", + "Loyalty-data aggregator API key (set the real value after deploy)", + ) + : undefined; + const aggregatorEnvironment: Record = ctx("aggregatorApiUrl") + ? { AGGREGATOR_API_URL: ctx("aggregatorApiUrl")! } + : {}; + const aggregatorSecrets: Record = aggregatorSecret + ? { AGGREGATOR_API_KEY: ecs.Secret.fromSecretsManager(aggregatorSecret) } + : {}; + const image = new ecrAssets.DockerImageAsset(this, "AppImage", { directory: path.join(__dirname, "..", ".."), platform: ecrAssets.Platform.LINUX_AMD64, @@ -185,6 +202,7 @@ export class AppStack extends cdk.Stack { // src/env.ts composes DATABASE_URL from the DB_* variables below. // Assistant (Bedrock/OpenAI) + Firecrawl config, when configured. ...assistantEnvironment, + ...aggregatorEnvironment, }, secrets: { DB_HOST: ecs.Secret.fromSecretsManager(dbSecret, "host"), @@ -194,6 +212,7 @@ export class AppStack extends cdk.Stack { DB_NAME: ecs.Secret.fromSecretsManager(dbSecret, "dbname"), CLERK_SECRET_KEY: ecs.Secret.fromSecretsManager(clerkSecret), ...assistantSecrets, + ...aggregatorSecrets, }, logDriver: ecs.LogDrivers.awsLogs({ streamPrefix: "app", @@ -380,6 +399,7 @@ export class AppStack extends cdk.Stack { DB_PASSWORD: ecs.Secret.fromSecretsManager(dbSecret, "password"), DB_NAME: ecs.Secret.fromSecretsManager(dbSecret, "dbname"), CLERK_SECRET_KEY: ecs.Secret.fromSecretsManager(clerkSecret), + ...aggregatorSecrets, }; const syncTask = new ecsPatterns.ScheduledFargateTask(this, "SyncTask", { @@ -391,7 +411,7 @@ export class AppStack extends cdk.Stack { command: ["sync"], cpu: 256, memoryLimitMiB: 512, - environment: { NODE_ENV: "production" }, + environment: { NODE_ENV: "production", ...aggregatorEnvironment }, secrets: workerSecrets, logDriver: ecs.LogDrivers.awsLogs({ streamPrefix: "worker-sync", @@ -573,6 +593,12 @@ export class AppStack extends cdk.Stack { description: "Set the real Firecrawl API key (fc-...) in this secret", }); } + if (aggregatorSecret) { + new cdk.CfnOutput(this, "AggregatorSecretArn", { + value: aggregatorSecret.secretArn, + description: "Set the real loyalty-data aggregator API key in this secret", + }); + } // Consumed by .github/workflows/deploy.yml to run migrations post-deploy. new cdk.CfnOutput(this, "ClusterArn", { value: cluster.clusterArn }); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index b39b09d..2612ecc 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -54,6 +54,8 @@ export * from "./infrastructure/repositories/drizzle-loyalty-account-repository" export * from "./infrastructure/repositories/drizzle-custom-valuation-repository"; export * from "./infrastructure/providers/composite-travel-provider-gateway"; export * from "./infrastructure/providers/simulated-travel-provider-gateway"; +export * from "./infrastructure/providers/http-aggregator-travel-provider-gateway"; +export * from "./infrastructure/providers/build-gateway"; export * from "./infrastructure/vault/one-password-connect-vault"; export * from "./infrastructure/vault/null-credential-vault"; export * from "./infrastructure/llm/openai-compatible-assistant"; diff --git a/packages/core/src/infrastructure/providers/build-gateway.ts b/packages/core/src/infrastructure/providers/build-gateway.ts new file mode 100644 index 0000000..b490368 --- /dev/null +++ b/packages/core/src/infrastructure/providers/build-gateway.ts @@ -0,0 +1,39 @@ +import type { TravelProviderGateway } from "../../application/ports"; +import { CompositeTravelProviderGateway } from "./composite-travel-provider-gateway"; +import { + HttpAggregatorTravelProviderGateway, + type HttpAggregatorConfig, +} from "./http-aggregator-travel-provider-gateway"; +import { SimulatedTravelProviderGateway } from "./simulated-travel-provider-gateway"; + +export interface BuildGatewayOptions { + /** When set (baseUrl + apiKey), a real aggregator gateway is added first. */ + readonly aggregator?: Partial; +} + +/** + * Composition helper shared by every host (web, worker) so they build the same + * provider gateway: the real aggregator adapter first (when configured), then + * the simulated gateway as the fallback for everything else. + */ +export function buildTravelProviderGateway( + options: BuildGatewayOptions = {}, +): CompositeTravelProviderGateway { + const gateways: TravelProviderGateway[] = []; + + const aggregator = options.aggregator; + if (aggregator?.baseUrl && aggregator.apiKey) { + gateways.push( + new HttpAggregatorTravelProviderGateway({ + baseUrl: aggregator.baseUrl, + apiKey: aggregator.apiKey, + ...(aggregator.supportedProviderIds + ? { supportedProviderIds: aggregator.supportedProviderIds } + : {}), + }), + ); + } + + gateways.push(new SimulatedTravelProviderGateway()); + return new CompositeTravelProviderGateway(gateways); +} diff --git a/packages/core/src/infrastructure/providers/http-aggregator-travel-provider-gateway.ts b/packages/core/src/infrastructure/providers/http-aggregator-travel-provider-gateway.ts new file mode 100644 index 0000000..7cac4a6 --- /dev/null +++ b/packages/core/src/infrastructure/providers/http-aggregator-travel-provider-gateway.ts @@ -0,0 +1,107 @@ +import type { LoyaltyAccount } from "../../domain/loyalty/loyalty-account"; +import type { + ProviderBalance, + ProviderCredential, + TravelProviderGateway, +} from "../../application/ports"; + +/** Minimal fetch shape so the adapter is unit-testable without a network. */ +export type AggregatorFetch = ( + url: string, + init: { + method: string; + headers: Record; + body: string; + signal?: AbortSignal; + }, +) => Promise<{ ok: boolean; status: number; text(): Promise }>; + +const defaultFetch: AggregatorFetch = (url, init) => + fetch(url, init as RequestInit); + +export interface HttpAggregatorConfig { + /** Aggregator API base URL, e.g. https://api.aggregator.example */ + readonly baseUrl: string; + readonly apiKey: string; + /** + * Provider ids this aggregator serves. Omit to let it attempt any provider + * (useful for a broad aggregator); set it to scope the adapter to the + * programs the vendor actually supports. + */ + readonly supportedProviderIds?: readonly string[]; + readonly fetchImpl?: AggregatorFetch; + readonly timeoutMs?: number; +} + +/** + * Real balance gateway backed by a loyalty-data aggregator's HTTP API — one + * adapter to cover the long tail of programs (roadmap Phase 3). Activated by + * configuration and registered ahead of the simulated gateway in the composite, + * so real providers win and everything else falls back to the simulation. + * + * Aggregator wire contract (vendor side): + * POST {baseUrl}/v1/balance + * Authorization: Bearer {apiKey} + * { "providerId", "membershipNumber", "credential"?: { username, secret } } + * → 200 { "points": number } + * + * Credentials, when supplied by the calling surface, are forwarded for + * one-time use and never persisted here. + */ +export class HttpAggregatorTravelProviderGateway + implements TravelProviderGateway +{ + private readonly baseUrl: string; + private readonly fetchImpl: AggregatorFetch; + private readonly timeoutMs: number; + + constructor(private readonly config: HttpAggregatorConfig) { + this.baseUrl = config.baseUrl.replace(/\/$/, ""); + this.fetchImpl = config.fetchImpl ?? defaultFetch; + this.timeoutMs = config.timeoutMs ?? 20_000; + } + + supports(providerId: string): boolean { + return ( + !this.config.supportedProviderIds || + this.config.supportedProviderIds.includes(providerId) + ); + } + + async fetchBalance( + account: LoyaltyAccount, + credential: ProviderCredential | null, + ): Promise { + const response = await this.fetchImpl(`${this.baseUrl}/v1/balance`, { + method: "POST", + headers: { + Authorization: `Bearer ${this.config.apiKey}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + providerId: account.providerId, + membershipNumber: account.membershipNumber, + ...(credential + ? { credential: { username: credential.username, secret: credential.secret } } + : {}), + }), + signal: AbortSignal.timeout(this.timeoutMs), + }); + + if (!response.ok) { + const body = await response.text().catch(() => ""); + throw new Error( + `Aggregator balance fetch failed for "${account.providerId}" (${response.status}): ${body.slice(0, 200)}`, + ); + } + + const payload = JSON.parse(await response.text()) as { points?: unknown }; + const points = payload.points; + if (typeof points !== "number" || !Number.isFinite(points) || points < 0) { + throw new Error( + `Aggregator returned an invalid balance for "${account.providerId}"`, + ); + } + return { points: Math.round(points) }; + } +} diff --git a/packages/core/test/http-aggregator-gateway.test.ts b/packages/core/test/http-aggregator-gateway.test.ts new file mode 100644 index 0000000..456bd51 --- /dev/null +++ b/packages/core/test/http-aggregator-gateway.test.ts @@ -0,0 +1,122 @@ +import { describe, expect, it } from "vitest"; + +import { + HttpAggregatorTravelProviderGateway, + type AggregatorFetch, +} from "../src/infrastructure/providers/http-aggregator-travel-provider-gateway"; +import { CompositeTravelProviderGateway } from "../src/infrastructure/providers/composite-travel-provider-gateway"; +import { SimulatedTravelProviderGateway } from "../src/infrastructure/providers/simulated-travel-provider-gateway"; +import { createLoyaltyAccount } from "../src/domain/loyalty/loyalty-account"; + +function account(providerId = "united") { + return createLoyaltyAccount({ + userId: "u", + providerId, + membershipNumber: "MP1", + }); +} + +function stubFetch( + handler: (url: string, body: unknown) => { ok: boolean; status?: number; text: string }, +): { fetchImpl: AggregatorFetch; calls: Array<{ url: string; body: unknown }> } { + const calls: Array<{ url: string; body: unknown }> = []; + const fetchImpl: AggregatorFetch = async (url, init) => { + const body = JSON.parse(init.body); + calls.push({ url, body }); + const r = handler(url, body); + return { ok: r.ok, status: r.status ?? (r.ok ? 200 : 500), text: async () => r.text }; + }; + return { fetchImpl, calls }; +} + +describe("HttpAggregatorTravelProviderGateway", () => { + it("posts to /v1/balance with auth and returns the rounded points", async () => { + const { fetchImpl, calls } = stubFetch(() => ({ + ok: true, + text: JSON.stringify({ points: 124_300.6 }), + })); + const gw = new HttpAggregatorTravelProviderGateway({ + baseUrl: "https://api.aggr.test/", + apiKey: "k", + fetchImpl, + }); + + const balance = await gw.fetchBalance(account(), null); + expect(balance).toEqual({ points: 124_301 }); + expect(calls[0]?.url).toBe("https://api.aggr.test/v1/balance"); + expect(calls[0]?.body).toMatchObject({ + providerId: "united", + membershipNumber: "MP1", + }); + }); + + it("forwards a transient credential when provided", async () => { + const { fetchImpl, calls } = stubFetch(() => ({ + ok: true, + text: JSON.stringify({ points: 10 }), + })); + const gw = new HttpAggregatorTravelProviderGateway({ + baseUrl: "https://a", + apiKey: "k", + fetchImpl, + }); + await gw.fetchBalance(account(), { username: "user", secret: "pw" }); + expect(calls[0]?.body).toMatchObject({ + credential: { username: "user", secret: "pw" }, + }); + }); + + it("throws on a non-2xx response and on an invalid balance", async () => { + const failing = stubFetch(() => ({ ok: false, status: 502, text: "upstream down" })); + await expect( + new HttpAggregatorTravelProviderGateway({ + baseUrl: "https://a", + apiKey: "k", + fetchImpl: failing.fetchImpl, + }).fetchBalance(account(), null), + ).rejects.toThrow(/502/); + + const bad = stubFetch(() => ({ ok: true, text: JSON.stringify({ points: -5 }) })); + await expect( + new HttpAggregatorTravelProviderGateway({ + baseUrl: "https://a", + apiKey: "k", + fetchImpl: bad.fetchImpl, + }).fetchBalance(account(), null), + ).rejects.toThrow(/invalid balance/); + }); + + it("scopes support to configured providers", () => { + const gw = new HttpAggregatorTravelProviderGateway({ + baseUrl: "https://a", + apiKey: "k", + supportedProviderIds: ["united", "delta"], + }); + expect(gw.supports("united")).toBe(true); + expect(gw.supports("hyatt")).toBe(false); + }); + + it("takes precedence over the simulated gateway in the composite", async () => { + const { fetchImpl } = stubFetch(() => ({ + ok: true, + text: JSON.stringify({ points: 999 }), + })); + const aggregator = new HttpAggregatorTravelProviderGateway({ + baseUrl: "https://a", + apiKey: "k", + supportedProviderIds: ["united"], + fetchImpl, + }); + const composite = new CompositeTravelProviderGateway([ + aggregator, + new SimulatedTravelProviderGateway(), + ]); + + // united → real aggregator (999); hyatt → simulated (deterministic hash). + expect(await composite.fetchBalance(account("united"), null)).toEqual({ + points: 999, + }); + const sim = await composite.fetchBalance(account("hyatt"), null); + expect(sim.points).not.toBe(999); + }); +});