Fix Apollo Server 5 Express integration and harden finance cluster service

README.md

@@ -10,8 +10,8 @@ clean checkout without any database or other external dependency.
 
 ## Stack
 
-- [Node.js](https://nodejs.org/) 18+ (ES modules)
-- [Apollo Server 4](https://www.apollographql.com/docs/apollo-server/) on [Express](https://expressjs.com/)
+- [Node.js](https://nodejs.org/) 20+ (ES modules)
+- [Apollo Server 5](https://www.apollographql.com/docs/apollo-server/) on [Express 4](https://expressjs.com/) via `@as-integrations/express4`
 - [graphql-js](https://github.com/graphql/graphql-js)
 - Tests with the built-in `node:test` runner
 
@@ -147,6 +147,9 @@ the mock adapters.
 5. Connector failures are captured as `FinanceUpstreamError` objects so clients
    receive actionable source/code/message details while still getting any
    partial data from healthy upstreams.
+6. Upstream reads go through a short-lived TTL cache
+   (`FINANCE_CACHE_TTL_MS`) that also de-duplicates concurrent requests, so
+   overlapping resolvers share a single connector call.
 
 ### Finance queries
 

package.json

@@ -5,7 +5,7 @@
   "type": "module",
   "main": "src/index.js",
   "engines": {
-    "node": ">=18"
+    "node": ">=20"
   },
   "scripts": {
     "start": "node src/index.js",
@@ -21,6 +21,7 @@
   "license": "Apache-2.0",
   "dependencies": {
     "@apollo/server": "^5.5.1",
+    "@as-integrations/express4": "^1.1.2",
     "cors": "^2.8.5",
     "express": "^4.21.2",
     "graphql": "^16.10.0"

src/index.js

@@ -1,4 +1,4 @@
-import { expressMiddleware } from '@apollo/server/express4';
+import { expressMiddleware } from '@as-integrations/express4';
 import cors from 'cors';
 import express from 'express';
 

src/services/financeService.js

@@ -38,15 +38,30 @@ export function createFinanceService({ config = loadFinanceConfig(), connectors,
     taxBreak: createTaxBreakConnector(config.taxBreak),
   };
   const cache = new Map();
+  // Concurrent resolvers asking for the same upstream data share a single
+  // in-flight request instead of fanning out duplicate connector calls.
+  const inFlight = new Map();
 
   async function cached(key, load) {
     const now = Date.now();
     const hit = cache.get(key);
     if (hit && hit.expiresAt > now) return hit.value;
 
-    const value = await load();
-    cache.set(key, { value, expiresAt: now + cacheTtlMs });
-    return value;
+    const pending = inFlight.get(key);
+    if (pending) return pending;
+
+    const request = (async () => {
+      try {
+        const value = await load();
+        cache.set(key, { value, expiresAt: Date.now() + cacheTtlMs });
+        return value;
+      } finally {
+        inFlight.delete(key);
+      }
+    })();
+
+    inFlight.set(key, request);
+    return request;
   }
 
   async function getTradingData() {

test/finance.test.js

@@ -0,0 +1,136 @@
+import assert from 'node:assert/strict';
+import { describe, it } from 'node:test';
+
+import {
+  aggregatePortfolio,
+  estimateTaxFromEvents,
+  filterTrades,
+  normalizeAccount,
+  normalizePosition,
+  normalizeTrade,
+  tradeToTaxEvent,
+} from '../src/domain/finance.js';
+import { loadFinanceConfig } from '../src/config/finance.js';
+import { createFinanceService } from '../src/services/financeService.js';
+
+describe('finance domain normalization', () => {
+  it('normalizes inconsistent upstream account and position fields', () => {
+    const account = normalizeAccount({ acct_id: 7, display_name: 'IRA', account_type: 'RETIREMENT', base_currency: 'EUR', source: 'OpenTrading' });
+    assert.deepEqual(account, { id: '7', name: 'IRA', type: 'RETIREMENT', currency: 'EUR', provider: 'OpenTrading' });
+
+    const position = normalizePosition({ account_id: 'acct-1', ticker: 'msft', qty: 5, cost_basis_per_share: 100, last_price: 120 });
+    assert.equal(position.id, 'acct-1:msft');
+    assert.equal(position.symbol, 'MSFT');
+    assert.equal(position.marketValue, 600);
+    assert.equal(position.unrealizedPnL, 100);
+  });
+
+  it('normalizes trades and derives tax events for sells and buys', () => {
+    const trade = normalizeTrade({
+      trade_id: 't-1',
+      acct_id: 'acct-1',
+      order_ref: 'o-1',
+      ticker: 'aapl',
+      action: 'sell',
+      qty: 4,
+      avg_px: 180,
+      trade_time: '2026-04-05T15:45:00.000Z',
+      fills: [{ fill_id: 'f-1', fill_qty: 4, fill_px: 180, filled_at: '2026-04-05T15:45:00.000Z' }],
+    });
+
+    assert.equal(trade.side, 'SELL');
+    assert.equal(trade.symbol, 'AAPL');
+    assert.equal(trade.fills.length, 1);
+
+    const sellEvent = tradeToTaxEvent(trade);
+    assert.equal(sellEvent.tradeId, 't-1');
+    assert.equal(sellEvent.proceeds, 720);
+    assert.equal(sellEvent.realizedGain, 129.6);
+
+    const buyEvent = tradeToTaxEvent({ ...trade, id: 't-2', side: 'BUY' });
+    assert.equal(buyEvent.proceeds, 0);
+    assert.equal(buyEvent.realizedGain, -720);
+  });
+});
+
+describe('finance aggregation', () => {
+  it('sums market value and unrealized P/L across positions', () => {
+    const overview = aggregatePortfolio({
+      accounts: [{ id: 'acct-1', currency: 'USD' }],
+      positions: [
+        { marketValue: 1000, unrealizedPnL: 100 },
+        { marketValue: 145.5, unrealizedPnL: -5.5 },
+      ],
+    });
+
+    assert.equal(overview.currency, 'USD');
+    assert.equal(overview.totalMarketValue, 1145.5);
+    assert.equal(overview.totalUnrealizedPnL, 94.5);
+    assert.deepEqual(overview.errors, []);
+  });
+
+  it('filters trades by account and symbol and estimates tax for a single year', () => {
+    const trades = [
+      { accountId: 'acct-1', symbol: 'AAPL' },
+      { accountId: 'acct-2', symbol: 'AAPL' },
+      { accountId: 'acct-1', symbol: 'MSFT' },
+    ];
+
+    assert.equal(filterTrades(trades, { accountId: 'acct-1' }).length, 2);
+    assert.equal(filterTrades(trades, { accountId: 'acct-1', symbol: 'aapl' }).length, 1);
+
+    const estimate = estimateTaxFromEvents(
+      [
+        { proceeds: 720, costBasis: 590.4, realizedGain: 129.6, occurredAt: '2026-04-05T15:45:00.000Z' },
+        { proceeds: 100, costBasis: 50, realizedGain: 50, occurredAt: '2025-04-05T15:45:00.000Z' },
+      ],
+      2026
+    );
+
+    assert.equal(estimate.events.length, 1);
+    assert.equal(estimate.realizedGain, 129.6);
+    assert.equal(estimate.estimatedTax, 28.51);
+  });
+});
+
+describe('finance service configuration and batching', () => {
+  it('reads endpoints and cache TTL from the environment with mock defaults', () => {
+    const defaults = loadFinanceConfig({});
+    assert.equal(defaults.openTrading.endpoint, 'mock://opentrading');
+    assert.equal(defaults.cacheTtlMs, 1000);
+
+    const configured = loadFinanceConfig({ OPENTRADING_ENDPOINT: 'https://trading.example', FINANCE_CACHE_TTL_MS: '5000' });
+    assert.equal(configured.openTrading.endpoint, 'https://trading.example');
+    assert.equal(configured.cacheTtlMs, 5000);
+  });
+
+  it('shares a single upstream call between concurrent and cached requests', async () => {
+    let accountCalls = 0;
+    const finance = createFinanceService({
+      cacheTtlMs: 60_000,
+      connectors: {
+        openTrading: {
+          listAccounts: async () => {
+            accountCalls += 1;
+            return [{ acct_id: 'acct-1', display_name: 'Primary', source: 'OpenTrading' }];
+          },
+          listTrades: async () => [],
+          listOrders: async () => [],
+        },
+        portfolioWatcher: {
+          listPositions: async () => [],
+          listPerformanceSnapshots: async () => [],
+        },
+        taxBreak: {
+          mapTradesToTaxEvents: async () => [],
+          estimateTax: async () => ({}),
+        },
+      },
+    });
+
+    await Promise.all([finance.portfolioOverview(), finance.portfolioOverview()]);
+    await finance.portfolioOverview();
+
+    assert.equal(accountCalls, 1);
+  });
+});