payments-dashboard/src/fetchBoaTransactions.js

import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { DynamoDBDocumentClient, GetCommand, PutCommand, ScanCommand, UpdateCommand } from "@aws-sdk/lib-dynamodb";
import { SecretsManagerClient, GetSecretValueCommand } from "@aws-sdk/client-secrets-manager";
import {
  applyEvent,
  buildEventUpdate,
  classifyTransaction,
  isValidISODate,
  logSafe,
  matchAchTransaction,
  matchCheckTransaction,
  matchElectronicReturn,
  resolveDateRange,
  sweepStalePayments,
} from "./boaRecon.js";

const ddb = DynamoDBDocumentClient.from(new DynamoDBClient());
const secrets = new SecretsManagerClient();
const TABLE_NAME = process.env.TABLE_NAME;
const BOA_BASE_URL = process.env.BOA_BASE_URL;

const UNMATCHED_LIST_CAP = 50;
const UNKNOWN_CODES_CAP = 20;

let cachedCreds;
async function getReportingCreds() {
  if (cachedCreds) return cachedCreds;
  const { SecretString } = await secrets.send(
    new GetSecretValueCommand({ SecretId: process.env.BOA_REPORTING_SECRET_NAME })
  );
  cachedCreds = JSON.parse(SecretString);
  return cachedCreds;
}

// Parse a BoA response body without ever surfacing a body snippet — a
// malformed 200 must not leak account data into logs/errors.
async function parseJsonResponse(res, label) {
  try {
    return await res.json();
  } catch {
    throw new Error(`BoA ${label} response not JSON: HTTP ${res.status}`);
  }
}

async function getAccessToken(applicationID, clientId, clientSecret) {
  const res = await fetch(`${BOA_BASE_URL}/authn/v1/client-authentication`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      applicationID,
      authn: { client_id: clientId, client_secret: clientSecret },
    }),
  });

  if (!res.ok) {
    throw new Error(`OAuth token exchange failed: HTTP ${res.status}`);
  }

  const data = await parseJsonResponse(res, "authn");
  return data.access_token;
}

const directionRank = (d) => (d === "debit" ? 0 : d === "credit" ? 1 : 2);

export const handler = async (event) => {
  // Optional replay payload {fromDate, toDate}; default is the trailing
  // 3-day window so the Monday run covers Friday through Sunday.
  const { fromDate, toDate } = resolveDateRange(event ?? {});

  const { appId, clientId, token: clientSecret, accountNumber, bankId } = await getReportingCreds();

  const bearerToken = await getAccessToken(appId, clientId, clientSecret);

  // Call CashPro Previous Day Transaction Inquiry
  const res = await fetch(`${BOA_BASE_URL}/cashpro/reporting/v1/transaction-inquiries/previous-day`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${bearerToken}`,
    },
    body: JSON.stringify({
      fromDate,
      toDate,
      accounts: [{ accountNumber, bankId }],
    }),
  });

  if (!res.ok) {
    throw new Error(`BoA API error: HTTP ${res.status}`);
  }

  const data = await parseJsonResponse(res, "reporting");

  // Response shape: { accountTransactions: [{ accountNumber, bankId, currency, transactions: [...] }] }
  const allTransactions = (data.accountTransactions || []).flatMap(
    (acct) => acct.transactions || []
  );

  // Process in posting-date order so multi-day replays apply paid -> return
  // -> redeposit sequences in the order the bank did; within a day, debits
  // before credits (deterministic, and a same-day return follows its debit).
  const classifiedTxns = allTransactions.map((txn) => ({
    txn,
    classified: classifyTransaction(txn),
  }));
  classifiedTxns.sort(
    (a, b) =>
      String(a.txn.valueDate ?? "").localeCompare(String(b.txn.valueDate ?? "")) ||
      directionRank(a.classified.direction) - directionRank(b.classified.direction)
  );

  // Load all payment records (Check AND ACH — ACH is bank-confirmed here
  // too, #69) to match against.
  const payments = [];
  let lastKey;
  do {
    const result = await ddb.send(
      new ScanCommand({
        TableName: TABLE_NAME,
        FilterExpression: "begins_with(pk, :prefix)",
        ExpressionAttributeValues: { ":prefix": "payment#" },
        ExclusiveStartKey: lastKey,
      })
    );
    payments.push(...result.Items);
    lastKey = result.LastEvaluatedKey;
  } while (lastKey);

  const summary = {
    transactions_seen: allTransactions.length,
    classified: {},
    matched: 0,
    applied: 0,
    already_applied: 0,
    redeposits: 0,
    voided_and_bounced: 0,
    write_conflicts: 0,
    unmatched_count: 0,
    unmatched: [],
    unknown_count: 0,
    unknown_codes: {},
  };

  const recordUnmatched = (classified, reason) => {
    summary.unmatched_count++;
    if (summary.unmatched.length < UNMATCHED_LIST_CAP) {
      summary.unmatched.push({
        event: classified.event,
        check_number: classified.checkNumber || classified.embeddedPaymentNumber,
        amount: classified.amount,
        reason,
      });
    }
    console.error(
      `Unmatched ${classified.event}: check=${logSafe(classified.checkNumber || classified.embeddedPaymentNumber)}, ` +
        `amount=${classified.amount}, reason=${reason} ` +
        `(bankRef: ${logSafe(classified.bankReference)})`
    );
  };

  for (const { txn, classified } of classifiedTxns) {
    summary.classified[classified.event] = (summary.classified[classified.event] || 0) + 1;

    if (classified.event === "ignored") continue;

    if (classified.event === "unknown") {
      summary.unknown_count++;
      const codeKey = (classified.code || "?").slice(0, 16);
      if (
        codeKey in summary.unknown_codes ||
        Object.keys(summary.unknown_codes).length < UNKNOWN_CODES_CAP
      ) {
        summary.unknown_codes[codeKey] = (summary.unknown_codes[codeKey] || 0) + 1;
      }
      console.error(
        `Unknown check-shaped transaction: code=${logSafe(classified.code)}, ` +
          `description=${logSafe(classified.description)}, ` +
          `ref=${logSafe(classified.customerReference)}, amount=${classified.amount}`
      );
      continue;
    }

    // Never substitute a date: event identity and transition ordering both
    // key on the posting date, so a feed row without one is a human case.
    if (!isValidISODate(txn.valueDate)) {
      recordUnmatched(classified, "missing valueDate");
      continue;
    }
    const eventDate = txn.valueDate;

    let match;
    if (classified.event === "ach_debit" || classified.event === "ach_return") {
      match = matchAchTransaction(classified, payments, eventDate);
    } else if (classified.event === "electronic_return") {
      match = matchElectronicReturn(classified, payments, eventDate);
    } else {
      match = matchCheckTransaction(classified, payments, eventDate);
    }

    if (!match.payment) {
      recordUnmatched(classified, match.unmatched);
      continue;
    }

    if (match.matchedBy === "vendor+amount") {
      // Audit trail for the loosest ACH rung.
      console.log(
        `ACH vendor+amount match: payee=${logSafe(match.payment.payee)}, ` +
          `bankVendor=${logSafe(classified.vendorText)}, pmt=${logSafe(classified.pmtId)}`
      );
    }

    summary.matched++;
    const target = match.payment;
    let applied = applyEvent(target, classified, eventDate);
    if (applied.kind === "noop") {
      summary.already_applied++;
      continue;
    }

    // Conditioned write: assert the snapshot's status/clear_status are
    // still current; on conflict, re-read, re-derive, retry once.
    let outcome = "conflict";
    for (let attempt = 0; attempt < 2; attempt++) {
      try {
        await ddb.send(new UpdateCommand(buildEventUpdate(TABLE_NAME, target, applied)));
        outcome = "written";
        break;
      } catch (err) {
        if (err.name !== "ConditionalCheckFailedException") throw err;
        if (attempt === 1) break;
        const { Item: fresh } = await ddb.send(
          new GetCommand({ TableName: TABLE_NAME, Key: { pk: target.pk } })
        );
        if (!fresh) break;
        // Refresh the in-memory record in place (it is shared with the
        // payments array) and re-derive the transition.
        for (const k of Object.keys(target)) delete target[k];
        Object.assign(target, fresh);
        applied = applyEvent(target, classified, eventDate);
        if (applied.kind === "noop") {
          outcome = "noop";
          break;
        }
      }
    }

    if (outcome === "noop") {
      summary.already_applied++;
      continue;
    }
    if (outcome !== "written") {
      summary.write_conflicts++;
      console.error(
        `Write conflict (gave up after retry): pk=${logSafe(target.pk)}, event=${classified.event}`
      );
      continue;
    }

    summary.applied++;
    if (applied.kind === "redeposit") summary.redeposits++;
    if (applied.kind === "voided_and_bounced") summary.voided_and_bounced++;

    // Mirror the write onto the in-memory copy so later transactions in the
    // same run (return after paid, redeposit after return) see current state.
    Object.assign(target, applied.updates);
    target.history = [...(target.history || []), applied.historyEvent];

    console.log(
      `${applied.kind}: check ${logSafe(target.check_number)} via ${match.matchedBy} ` +
        `(bankRef: ${logSafe(classified.bankReference)})`
    );
  }

  // Staleness sweep: payments the bank has never confirmed (#69).
  const todayISO = new Date().toISOString().split("T")[0];
  const stale = sweepStalePayments(payments, todayISO);
  if (stale.staleAchCount) {
    console.error(
      `Stale ACH (no bank settlement, sent > 16 days ago): ${stale.staleAchCount} records, ` +
        `checks=${logSafe(stale.staleAch.map((s) => s.check_number).join(","))}`
    );
  }
  if (stale.staleChecksCount) {
    console.error(
      `Stale checks (no bank activity, issued > 60 days ago): ${stale.staleChecksCount} records`
    );
  }

  // Run summary record for auditing/alerting (Slack wiring is #71). The pk
  // is append-only (run timestamp suffix) so re-runs never overwrite a
  // prior run's record.
  const runAt = new Date().toISOString();
  const runKey = `boa_recon#${fromDate}_${toDate}#${runAt}`;
  await ddb.send(
    new PutCommand({
      TableName: TABLE_NAME,
      Item: {
        pk: runKey,
        run_at: runAt,
        from_date: fromDate,
        to_date: toDate,
        transactions_seen: summary.transactions_seen,
        classified: summary.classified,
        matched: summary.matched,
        applied: summary.applied,
        already_applied: summary.already_applied,
        redeposits: summary.redeposits,
        voided_and_bounced: summary.voided_and_bounced,
        write_conflicts: summary.write_conflicts,
        unmatched_count: summary.unmatched_count,
        unmatched: summary.unmatched,
        unknown_count: summary.unknown_count,
        unknown_codes: summary.unknown_codes,
        stale_ach: stale.staleAch,
        stale_ach_count: stale.staleAchCount,
        stale_checks_count: stale.staleChecksCount,
        ttl: Math.floor(Date.now() / 1000) + 90 * 24 * 60 * 60,
      },
    })
  );

  if (summary.unmatched_count || summary.unknown_count || summary.write_conflicts) {
    console.error(
      `Reconciliation ${fromDate}..${toDate}: ${summary.unmatched_count} unmatched, ` +
        `${summary.unknown_count} unknown-code transactions, ` +
        `${summary.write_conflicts} write conflicts (see ${logSafe(runKey)})`
    );
  }

  console.log(
    `Processed ${summary.transactions_seen} transactions ${fromDate}..${toDate}: ` +
      `${summary.matched} matched, ${summary.applied} applied, ` +
      `${summary.already_applied} already applied, ${summary.redeposits} redeposits, ` +
      `${summary.voided_and_bounced} voided-and-bounced, ${summary.unmatched_count} unmatched`
  );

  return {
    statusCode: 200,
    body:
      `${summary.matched} of ${summary.transactions_seen} transactions matched ` +
      `(${summary.applied} applied, ${summary.unmatched_count} unmatched)`,
  };
};