Production connector readiness, finance query filters/pagination, and observability

README.md

@@ -56,7 +56,9 @@ The server listens on port `4000` by default (override with the `PORT`
 environment variable):
 
 - GraphQL endpoint: <http://localhost:4000/graphql>
-- Health check: <http://localhost:4000/health>
+- Liveness check: <http://localhost:4000/health>
+- Readiness check (per-upstream): <http://localhost:4000/ready>
+- Metrics (Prometheus text): <http://localhost:4000/metrics>
 
 Opening the GraphQL endpoint in a browser loads the Apollo Sandbox, where you can
 explore the schema and run the operations below.
@@ -131,6 +133,17 @@ The running server loads connector settings from the environment via
 | `TAX_BREAK_ENDPOINT` | tax-break endpoint placeholder | `mock://tax-break` |
 | `TAX_BREAK_API_KEY` | tax-break credential placeholder | empty |
 | `FINANCE_CACHE_TTL_MS` | Minimal resolver cache TTL | `1000` |
+| `FINANCE_HTTP_TIMEOUT_MS` | Per-request upstream timeout | `5000` |
+| `FINANCE_HTTP_MAX_RETRIES` | Retries for timeouts, 429s, and 5xx responses | `2` |
+| `FINANCE_DEFAULT_PAGE_SIZE` | Default page size when `limit` is omitted | `25` |
+| `FINANCE_MAX_PAGE_SIZE` | Upper bound applied to any requested `limit` | `100` |
+| `LOG_LEVEL` | Structured log level (`debug`/`info`/`warn`/`error`) | `info` |
+
+Each connector endpoint that is **not** a `mock://` URL is served by the
+production HTTP client in `src/connectors/httpClient.js`, which adds bearer
+authentication, request timeouts, bounded retries with exponential backoff, and
+per-call metrics. Mock adapters remain the default so the service still runs
+from a clean checkout.
 
 Do not commit real credentials. Production connectors should read credentials
 from environment variables or a secret manager and keep the same method names as
@@ -202,11 +215,48 @@ npm start
 npm test
 ```
 
+### Filtering and pagination
+
+Finance queries accept optional filters and offset pagination:
+
+- `portfolioOverview(accountId, from, to, limit, offset)` — `from`/`to` bound
+  performance snapshots (inclusive ISO-8601); `limit`/`offset` page positions.
+- `tradeHistory(accountId, symbol, side, status, from, to, limit, offset)` —
+  filters trades and orders, then pages them. Returned tax events always match
+  the trades on the current page.
+- `taxEstimate(taxYear, accountId, symbol, from, to, limit, offset)` — totals are
+  always computed over every matching event; `limit`/`offset` only page the
+  returned `events`.
+
+Every finance payload includes `pageInfo { totalCount limit offset hasNextPage
+hasPreviousPage }`. Requested limits are clamped to `FINANCE_MAX_PAGE_SIZE`.
+
+```graphql
+query RecentSells {
+  tradeHistory(side: "SELL", from: "2026-01-01T00:00:00.000Z", limit: 10) {
+    trades { id symbol quantity price executedAt }
+    pageInfo { totalCount hasNextPage }
+  }
+}
+```
+
+### Observability
+
+- **Structured logs**: JSON lines from `src/observability/logger.js`, with
+  credential-like fields redacted. One line per GraphQL operation includes the
+  operation name, duration, outcome, and error codes.
+- **Metrics**: `src/observability/metrics.js` records GraphQL operation
+  counts/latency, connector call counts/latency per source and operation,
+  upstream retry failures, and cache hit/miss/coalesced counters. Scrape them at
+  `GET /metrics`.
+- **Health**: `GET /health` is a liveness probe; `GET /ready` calls each
+  connector's health check and returns `503` when any upstream is degraded.
+
 Follow-up production tasks:
 
-- Replace mock connectors with authenticated clients for each upstream domain.
-- Add pagination/date filters once live trade and snapshot volumes grow.
 - Add persisted caching/batching if upstream latency becomes significant.
+- Move from offset pagination to cursor pagination if upstream APIs expose
+  stable cursors.
 
 ## API
 
@@ -216,9 +266,9 @@ Follow-up production tasks:
 | `user(id: ID!)` | Fetch a single user, `null` when unknown |
 | `posts` | List all posts |
 | `post(id: ID!)` | Fetch a single post, `null` when unknown |
-| `portfolioOverview(accountId)` | Fetch finance accounts, positions, snapshots, and P/L |
-| `tradeHistory(accountId, symbol)` | Fetch trades/orders enriched with tax events |
-| `taxEstimate(taxYear, accountId)` | Estimate tax from tax-relevant trading activity |
+| `portfolioOverview(accountId, from, to, limit, offset)` | Fetch finance accounts, positions, snapshots, and P/L |
+| `tradeHistory(accountId, symbol, side, status, from, to, limit, offset)` | Fetch trades/orders enriched with tax events |
+| `taxEstimate(taxYear, accountId, symbol, from, to, limit, offset)` | Estimate tax from tax-relevant trading activity |
 | `createUser(name, email)` | Create a user |
 | `createPost(title, content, authorId)` | Create a post for an existing user |
 

src/config/finance.js

@@ -1,18 +1,33 @@
 /** Environment-driven configuration for finance-cluster connectors. */
 export function loadFinanceConfig(env = process.env) {
+  const parseNonNegativeInt = (value, fallback) => {
+    const parsed = Number.parseInt(value, 10);
+    return Number.isNaN(parsed) ? fallback : Math.max(0, parsed);
+  };
+  const timeoutMs = parseNonNegativeInt(env.FINANCE_HTTP_TIMEOUT_MS, 5000);
+  const maxRetries = parseNonNegativeInt(env.FINANCE_HTTP_MAX_RETRIES, 2);
+
   return {
     openTrading: {
       endpoint: env.OPENTRADING_ENDPOINT ?? 'mock://opentrading',
       apiKey: env.OPENTRADING_API_KEY ?? '',
+      timeoutMs,
+      maxRetries,
     },
     portfolioWatcher: {
       endpoint: env.PORTFOLIO_WATCHER_ENDPOINT ?? 'mock://portfolio-watcher',
       apiKey: env.PORTFOLIO_WATCHER_API_KEY ?? '',
+      timeoutMs,
+      maxRetries,
     },
     taxBreak: {
       endpoint: env.TAX_BREAK_ENDPOINT ?? 'mock://tax-break',
       apiKey: env.TAX_BREAK_API_KEY ?? '',
+      timeoutMs,
+      maxRetries,
     },
-    cacheTtlMs: Number(env.FINANCE_CACHE_TTL_MS ?? 1000),
+    cacheTtlMs: parseNonNegativeInt(env.FINANCE_CACHE_TTL_MS, 1000),
+    defaultPageSize: Number(env.FINANCE_DEFAULT_PAGE_SIZE ?? 25),
+    maxPageSize: Number(env.FINANCE_MAX_PAGE_SIZE ?? 100),
   };
 }

src/connectors/httpClient.js

@@ -0,0 +1,116 @@
+import { logger as defaultLogger } from '../observability/logger.js';
+import { metrics as defaultMetrics } from '../observability/metrics.js';
+
+/** Error carrying the upstream source and HTTP status for actionable messages. */
+export class UpstreamHttpError extends Error {
+  constructor(source, message, { status = null, retryable = false } = {}) {
+    super(message);
+    this.name = 'UpstreamHttpError';
+    this.source = source;
+    this.status = status;
+    this.retryable = retryable;
+  }
+}
+
+const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
+
+/** A 5xx or 429 response is worth retrying; client errors are not. */
+function isRetryableStatus(status) {
+  return status === 429 || status >= 500;
+}
+
+/**
+ * Shared HTTP client for production connectors.
+ *
+ * Adds the pieces a real upstream call needs but a mock does not: bearer
+ * authentication, request timeouts, bounded retries with backoff, structured
+ * logging, and latency/outcome metrics.
+ */
+export function createHttpClient({
+  source,
+  endpoint,
+  apiKey = '',
+  timeoutMs = 5000,
+  maxRetries = 2,
+  retryBackoffMs = 100,
+  fetchImpl = globalThis.fetch,
+  logger = defaultLogger,
+  metrics = defaultMetrics,
+} = {}) {
+  if (!endpoint) throw new Error(`${source} connector requires an endpoint`);
+  if (typeof fetchImpl !== 'function') throw new Error(`${source} connector requires a fetch implementation`);
+
+  const baseUrl = endpoint.endsWith('/') ? endpoint.slice(0, -1) : endpoint;
+
+  async function once(path, { method = 'GET', body, signal } = {}) {
+    const headers = { accept: 'application/json' };
+    if (apiKey) headers.authorization = ['Bearer', apiKey].join(' ');
+    if (body !== undefined) headers['content-type'] = 'application/json';
+
+    const response = await fetchImpl(`${baseUrl}${path}`, {
+      method,
+      headers,
+      signal,
+      body: body === undefined ? undefined : JSON.stringify(body),
+    });
+
+    if (!response.ok) {
+      throw new UpstreamHttpError(source, `HTTP ${response.status} from ${path}`, {
+        status: response.status,
+        retryable: isRetryableStatus(response.status),
+      });
+    }
+
+    return response.json();
+  }
+
+  async function request(path, options = {}) {
+    return metrics.time('finance_upstream_request', { source, path }, async () => {
+      let lastError;
+
+      for (let attempt = 0; attempt <= maxRetries; attempt += 1) {
+        const controller = new AbortController();
+        const timer = setTimeout(() => controller.abort(), timeoutMs);
+
+        try {
+          return await once(path, { ...options, signal: controller.signal });
+        } catch (error) {
+          lastError = error instanceof UpstreamHttpError
+            ? error
+            : new UpstreamHttpError(source, error?.name === 'AbortError' ? `request to ${path} timed out after ${timeoutMs}ms` : String(error?.message ?? error), { retryable: true });
+
+          metrics.increment('finance_upstream_attempt_failures_total', { source, path });
+          logger.warn('upstream request failed', {
+            source,
+            path,
+            attempt: attempt + 1,
+            status: lastError.status,
+            error: lastError.message,
+          });
+
+          if (!lastError.retryable || attempt === maxRetries) break;
+          await sleep(retryBackoffMs * 2 ** attempt);
+        } finally {
+          clearTimeout(timer);
+        }
+      }
+
+      throw lastError;
+    });
+  }
+
+  return {
+    source,
+    endpoint: baseUrl,
+    request,
+    /** Lightweight readiness probe used by the /ready endpoint. */
+    async health() {
+      try {
+        await request('/health');
+        return { source, status: 'ok', endpoint: baseUrl };
+      } catch (error) {
+        return { source, status: 'degraded', endpoint: baseUrl, error: error.message };
+      }
+    },
+  };
+}

src/connectors/index.js

@@ -0,0 +1,50 @@
+import { loadFinanceConfig } from '../config/finance.js';
+import { logger as defaultLogger } from '../observability/logger.js';
+import { metrics as defaultMetrics } from '../observability/metrics.js';
+import { createOpenTradingConnector } from './opentrading/mockConnector.js';
+import { createOpenTradingHttpConnector } from './opentrading/httpConnector.js';
+import { createPortfolioWatcherConnector } from './portfolio-watcher/mockConnector.js';
+import { createPortfolioWatcherHttpConnector } from './portfolio-watcher/httpConnector.js';
+import { createTaxBreakConnector } from './tax-break/mockConnector.js';
+import { createTaxBreakHttpConnector } from './tax-break/httpConnector.js';
+
+/** Mock endpoints keep the service runnable from a clean checkout. */
+export function isMockEndpoint(endpoint = '') {
+  return endpoint.startsWith('mock://');
+}
+
+const FACTORIES = {
+  openTrading: { mock: createOpenTradingConnector, http: createOpenTradingHttpConnector },
+  portfolioWatcher: { mock: createPortfolioWatcherConnector, http: createPortfolioWatcherHttpConnector },
+  taxBreak: { mock: createTaxBreakConnector, http: createTaxBreakHttpConnector },
+};
+
+/**
+ * Builds one connector per upstream domain, choosing the production HTTP client
+ * whenever a real endpoint is configured and falling back to the mock adapter
+ * for `mock://` endpoints.
+ */
+export function createConnectors({
+  config = loadFinanceConfig(),
+  logger = defaultLogger,
+  metrics = defaultMetrics,
+  fetchImpl,
+} = {}) {
+  return Object.fromEntries(
+    Object.entries(FACTORIES).map(([name, factory]) => {
+      const settings = config[name] ?? {};
+      const mode = isMockEndpoint(settings.endpoint) ? 'mock' : 'http';
+
+      // Live wiring is worth an info line; mock wiring stays at debug so test
+      // runs and local demos are not noisy.
+      logger[mode === 'http' ? 'info' : 'debug']('finance connector configured', { connector: name, mode, endpoint: settings.endpoint });
+
+      return [
+        name,
+        mode === 'mock'
+          ? factory.mock(settings)
+          : factory.http({ ...settings, logger, metrics, ...(fetchImpl ? { fetchImpl } : {}) }),
+      ];
+    })
+  );
+}

src/connectors/opentrading/httpConnector.js

@@ -0,0 +1,18 @@
+import { createHttpClient } from '../httpClient.js';
+
+/**
+ * Production OpenTrading connector.
+ * Exposes exactly the same contract as the mock adapter so the finance service
+ * and GraphQL schema are unchanged when switching to a live endpoint.
+ */
+export function createOpenTradingHttpConnector(config = {}) {
+  const client = createHttpClient({ source: 'OpenTrading', ...config });
+
+  return {
+    source: 'OpenTrading',
+    health: client.health,
+    listAccounts: async () => (await client.request('/accounts')).accounts ?? [],
+    listTrades: async () => (await client.request('/trades')).trades ?? [],
+    listOrders: async () => (await client.request('/orders')).orders ?? [],
+  };
+}

src/connectors/opentrading/mockConnector.js

@@ -38,6 +38,7 @@ export function createOpenTradingConnector(_config = {}) {
 
   return {
     source: 'OpenTrading',
+    health: async () => ({ source: 'OpenTrading', status: 'ok', endpoint: 'mock://opentrading' }),
     listAccounts: async () => accounts,
     listTrades: async () => trades,
     listOrders: async () => orders,

src/connectors/portfolio-watcher/httpConnector.js

@@ -0,0 +1,13 @@
+import { createHttpClient } from '../httpClient.js';
+
+/** Production Portfolio-Watcher connector mirroring the mock adapter contract. */
+export function createPortfolioWatcherHttpConnector(config = {}) {
+  const client = createHttpClient({ source: 'Portfolio-Watcher', ...config });
+
+  return {
+    source: 'Portfolio-Watcher',
+    health: client.health,
+    listPositions: async () => (await client.request('/positions')).positions ?? [],
+    listPerformanceSnapshots: async () => (await client.request('/performance')).snapshots ?? [],
+  };
+}

src/connectors/portfolio-watcher/mockConnector.js

@@ -11,6 +11,7 @@ export function createPortfolioWatcherConnector(_config = {}) {
 
   return {
     source: 'Portfolio-Watcher',
+    health: async () => ({ source: 'Portfolio-Watcher', status: 'ok', endpoint: 'mock://portfolio-watcher' }),
     listPositions: async () => positions,
     listPerformanceSnapshots: async () => snapshots,
   };

src/connectors/tax-break/httpConnector.js

@@ -0,0 +1,13 @@
+import { createHttpClient } from '../httpClient.js';
+
+/** Production tax-break connector mirroring the mock adapter contract. */
+export function createTaxBreakHttpConnector(config = {}) {
+  const client = createHttpClient({ source: 'tax-break', ...config });
+
+  return {
+    source: 'tax-break',
+    health: client.health,
+    mapTradesToTaxEvents: async (trades) => (await client.request('/tax-events', { method: 'POST', body: { trades } })).events ?? [],
+    estimateTax: async ({ events, taxYear }) => client.request('/tax-estimate', { method: 'POST', body: { events, taxYear } }),
+  };
+}

src/connectors/tax-break/mockConnector.js

@@ -4,6 +4,7 @@ import { estimateTaxFromEvents, tradeToTaxEvent } from '../../domain/finance.js'
 export function createTaxBreakConnector(_config = {}) {
   return {
     source: 'tax-break',
+    health: async () => ({ source: 'tax-break', status: 'ok', endpoint: 'mock://tax-break' }),
     mapTradesToTaxEvents: async (trades) => trades.filter((trade) => trade.side === 'SELL').map(tradeToTaxEvent),
     estimateTax: async ({ events, taxYear }) => estimateTaxFromEvents(events, taxYear, 0.22),
   };