Income verification with paystubs and bank statements

import { readFile } from "node:fs/promises";
import { basename } from "node:path";
import { Constaia, InsufficientCreditsError } from "@constaia/sdk";

const constaia = new Constaia(); // reads CONSTAIA_API_KEY

// Adjust the maximum age to your policy.
const MAX_AGE_DAYS = { us_paystub: 45, us_bank_statement: 90 } as const;
type Kind = keyof typeof MAX_AGE_DAYS;

async function item(path: string, kind: Kind) {
  return {
    base64: (await readFile(path)).toString("base64"),
    filename: basename(path),
    options: { expect: kind, checks: { maxAgeDays: MAX_AGE_DAYS[kind] } },
  };
}

export async function submitApplication(applicationId: string, applicantName: string, paystubs: string[], statements: string[]) {
  try {
    const batch = await constaia.batches.create({
      items: [
        ...(await Promise.all(paystubs.map((p) => item(p, "us_paystub")))),
        ...(await Promise.all(statements.map((p) => item(p, "us_bank_statement")))),
      ],
      options: {
        checks: { holder: { fullName: applicantName } },
        storage: "none",
        metadata: { application_id: applicationId },
      },
    });
    return batch.id;
  } catch (err) {
    // In live mode a batch needs at least 1 credit per document up front.
    if (err instanceof InsufficientCreditsError) throw new Error("Top up credits before submitting applications");
    throw err;
  }
}

import type { Analysis } from "@constaia/sdk";

const PERIODS_PER_MONTH = { weekly: 52 / 12, biweekly: 26 / 12, semimonthly: 2, monthly: 1 } as const;
type Frequency = keyof typeof PERIODS_PER_MONTH;

const num = (v: unknown) => (typeof v === "number" && Number.isFinite(v) ? v : null);
const avg = (xs: number[]) => (xs.length ? Math.round((xs.reduce((a, b) => a + b, 0) / xs.length) * 100) / 100 : null);

// Verdict and warnings: type_mismatch, max_age_days, holder, id_number_checksum, edited_suspected…
function commonFlags(a: Analysis): string[] {
  if (a.status !== "completed") return [`${a.id}:failed`];
  const reasons = (a.verdict?.reasons ?? [])
    .filter((r) => r.severity !== "info")
    .map((r) => `${a.id}:${r.code}:${r.severity}`);
  return [...reasons, ...a.warnings.map((w) => `${a.id}:${w}`)];
}

export function monthlyFromPaystubs(analyses: Analysis[]) {
  const flags: string[] = [];
  const monthly: number[] = [];
  for (const a of analyses) {
    flags.push(...commonFlags(a));
    if (a.status !== "completed" || a.verdict?.status === "invalid") continue;
    const gross = num(a.fields.gross_pay?.value);
    const net = num(a.fields.net_pay?.value);
    const ytd = num(a.fields.ytd_gross?.value);
    const freq = a.fields.pay_frequency?.value as Frequency | undefined;
    if (gross === null || !freq || !(freq in PERIODS_PER_MONTH)) {
      flags.push(`${a.id}:incomplete`);
      continue;
    }
    if (net !== null && net > gross) flags.push(`${a.id}:net_above_gross`);
    if (ytd !== null && ytd < gross) flags.push(`${a.id}:ytd_below_gross`);
    monthly.push(gross * PERIODS_PER_MONTH[freq]);
  }
  return { monthlyGross: avg(monthly), documents: monthly.length, flags };
}

export function monthlyFromStatements(analyses: Analysis[]) {
  const flags: string[] = [];
  const monthly: number[] = [];
  for (const a of analyses) {
    flags.push(...commonFlags(a));
    if (a.status !== "completed" || a.verdict?.status === "invalid") continue;
    const transactions = (a.fields.transactions?.value as { amount?: number }[] | undefined) ?? [];
    const sum = transactions.reduce((s, t) => s + Math.max(num(t.amount) ?? 0, 0), 0); // negative = debit
    const total = num(a.fields.total_deposits?.value) ?? sum;
    if (transactions.length && Math.abs(total - sum) > 1) flags.push(`${a.id}:deposits_do_not_add_up`);
    monthly.push(total);
  }
  return { monthlyDeposits: avg(monthly), documents: monthly.length, flags };
}

import express from "express";
import { Constaia, WebhookVerificationError, type Analysis, type Batch } from "@constaia/sdk";
import { monthlyFromPaystubs, monthlyFromStatements } from "./compute";

const constaia = new Constaia(); // reads CONSTAIA_API_KEY
const secret = process.env.CONSTAIA_WEBHOOK_SECRET!;
const processed = new Set<string>(); // use your database in production

const app = express();

app.post("/webhooks/constaia", express.raw({ type: "application/json" }), async (req, res) => {
  let event;
  try {
    event = await constaia.webhooks.verify(req.body, req.headers, secret);
  } catch (err) {
    if (err instanceof WebhookVerificationError) return res.status(400).send("invalid signature");
    throw err;
  }
  res.sendStatus(200);

  const deliveryId = req.header("webhook-id")!;
  if (processed.has(deliveryId)) return;
  processed.add(deliveryId);

  if (event.type === "batch.completed") {
    handleBatch(event.data as unknown as Batch).catch((e) => console.error("batch handling failed", e));
  }
});

async function handleBatch(batch: Batch) {
  const analyses: Analysis[] = [];
  for (const id of batch.analyses) analyses.push(await constaia.analyses.get(id));

  // verdict.expected keeps the type you asked for even when the document is something else.
  const expected = (a: Analysis) => a.verdict?.expected?.[0];
  const paystubs = monthlyFromPaystubs(analyses.filter((a) => expected(a) === "us_paystub"));
  const statements = monthlyFromStatements(analyses.filter((a) => expected(a) === "us_bank_statement"));
  const flags = [...paystubs.flags, ...statements.flags];
  const needsReview = flags.length > 0 || paystubs.documents === 0;

  await saveIncomeResult(batch.metadata?.application_id, { paystubs, statements }, needsReview);

  for (const a of analyses) await constaia.analyses.delete(a.id);
}

async function saveIncomeResult(applicationId: string | undefined, result: object, needsReview: boolean) {
  console.log({ applicationId, result, needsReview }); // replace with your database
}

app.listen(3000, () => console.log("Listening on :3000"));

import sys
import time

from constaia import Constaia, ConstaiaError, InsufficientCreditsError

client = Constaia()  # reads CONSTAIA_API_KEY

PERIODS_PER_MONTH = {"weekly": 52 / 12, "biweekly": 26 / 12, "semimonthly": 2, "monthly": 1}


def run(application_id: str, applicant_name: str, paths: list[str]) -> dict:
    batch = client.batches.create(
        files=paths,
        options={
            "expect": "us_paystub",
            "checks": {"max_age_days": 45, "holder": {"full_name": applicant_name}},
            "storage": "none",
            "metadata": {"application_id": application_id},
        },
    )
    while batch["status"] != "completed":
        time.sleep(5)
        batch = client.batches.get(batch["id"])

    monthly, flags = [], []
    for analysis_id in batch["analyses"]:
        a = client.analyses.get(analysis_id)
        if a["status"] != "completed":
            flags.append(f"{analysis_id}:failed")
        else:
            verdict = a.get("verdict") or {}
            flags += [f"{analysis_id}:{r['code']}:{r['severity']}" for r in verdict.get("reasons", []) if r["severity"] != "info"]
            flags += [f"{analysis_id}:{w}" for w in a["warnings"]]
            f = a["fields"]
            gross = (f.get("gross_pay") or {}).get("value")
            freq = (f.get("pay_frequency") or {}).get("value")
            if verdict.get("status") == "invalid":
                pass  # different document, too old or someone else's: kept in flags
            elif isinstance(gross, (int, float)) and freq in PERIODS_PER_MONTH:
                monthly.append(gross * PERIODS_PER_MONTH[freq])
            else:
                flags.append(f"{analysis_id}:incomplete")
        client.analyses.delete(analysis_id)

    avg = round(sum(monthly) / len(monthly), 2) if monthly else None
    return {"monthly_gross": avg, "flags": flags, "needs_review": bool(flags) or not monthly}


if __name__ == "__main__":
    try:
        print(run("app_123", "Jane Q Specimen", sys.argv[1:]))
    except InsufficientCreditsError:
        sys.exit("Top up credits before submitting applications")
    except ConstaiaError as err:
        sys.exit(f"Constaia error {err.status} {err.code} (request {err.request_id})")