Real provider integration: aggregator gateway behind TravelProviderGateway (Phase 3)

.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"

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,

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,

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,

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(),

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.

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<string, string> = ctx("aggregatorApiUrl")
+      ? { AGGREGATOR_API_URL: ctx("aggregatorApiUrl")! }
+      : {};
+    const aggregatorSecrets: Record<string, ecs.Secret> = 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 });

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

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<HttpAggregatorConfig>;
+}
+
+/**
+ * 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);
+}

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<string, string>;
+    body: string;
+    signal?: AbortSignal;
+  },
+) => Promise<{ ok: boolean; status: number; text(): Promise<string> }>;
+
+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<ProviderBalance> {
+    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) };
+  }
+}