593 lines
20 KiB
TypeScript
593 lines
20 KiB
TypeScript
import { BadRequestException, Injectable, Logger } from "@nestjs/common";
|
|
import {
|
|
Configuration,
|
|
CountryCode,
|
|
PlaidApi,
|
|
PlaidEnvironments,
|
|
Products,
|
|
} from "plaid";
|
|
import * as crypto from "crypto";
|
|
import { Prisma } from "@prisma/client";
|
|
import { PrismaService } from "../prisma/prisma.service";
|
|
import { EncryptionService } from "../common/encryption.service";
|
|
import { RawPayloadEncryptionService } from "../common/raw-payload-encryption.service";
|
|
import { PlanLimitsService } from "../stripe/plan-limits.service";
|
|
|
|
@Injectable()
|
|
export class PlaidService {
|
|
private readonly logger = new Logger(PlaidService.name);
|
|
private readonly client: PlaidApi;
|
|
private readonly webhookKeys = new Map<string, JsonWebKey>();
|
|
|
|
constructor(
|
|
private readonly prisma: PrismaService,
|
|
private readonly encryption: EncryptionService,
|
|
private readonly planLimits: PlanLimitsService,
|
|
private readonly rawPayloads: RawPayloadEncryptionService,
|
|
) {
|
|
const env = (process.env.PLAID_ENV ?? "sandbox") as keyof typeof PlaidEnvironments;
|
|
const clientId = this.requireEnv("PLAID_CLIENT_ID");
|
|
const secret = this.requireEnv("PLAID_SECRET");
|
|
|
|
const config = new Configuration({
|
|
basePath: PlaidEnvironments[env] ?? PlaidEnvironments.sandbox,
|
|
baseOptions: {
|
|
headers: {
|
|
"PLAID-CLIENT-ID": clientId,
|
|
"PLAID-SECRET": secret,
|
|
"Plaid-Version": "2020-09-14",
|
|
},
|
|
},
|
|
});
|
|
this.client = new PlaidApi(config);
|
|
}
|
|
|
|
async createLinkToken(userId: string) {
|
|
const products = (process.env.PLAID_PRODUCTS ?? "transactions")
|
|
.split(",")
|
|
.map((item) => item.trim())
|
|
.filter(Boolean) as Products[];
|
|
const countryCodes = (process.env.PLAID_COUNTRY_CODES ?? "US")
|
|
.split(",")
|
|
.map((item) => item.trim())
|
|
.filter(Boolean) as CountryCode[];
|
|
const redirectUri = process.env.PLAID_REDIRECT_URI?.trim();
|
|
|
|
try {
|
|
const response = await this.client.linkTokenCreate({
|
|
user: { client_user_id: userId },
|
|
client_name: "LedgerOne",
|
|
products,
|
|
country_codes: countryCodes,
|
|
language: "en",
|
|
redirect_uri: redirectUri || undefined,
|
|
webhook: process.env.PLAID_WEBHOOK_URL?.trim() || undefined,
|
|
});
|
|
|
|
return {
|
|
linkToken: response.data.link_token,
|
|
expiration: response.data.expiration,
|
|
};
|
|
} catch (error: unknown) {
|
|
const err = error as { response?: { data?: { error_message?: string } } };
|
|
const message =
|
|
err.response?.data?.error_message ?? "Plaid link token request failed.";
|
|
throw new BadRequestException(message);
|
|
}
|
|
}
|
|
|
|
async exchangePublicTokenForUser(userId: string, publicToken: string) {
|
|
const exchange = await this.client.itemPublicTokenExchange({
|
|
public_token: publicToken,
|
|
});
|
|
const rawAccessToken = exchange.data.access_token;
|
|
const itemId = exchange.data.item_id;
|
|
|
|
// Encrypt before storing
|
|
const encryptedToken = this.encryption.encrypt(rawAccessToken);
|
|
|
|
const accountsResponse = await this.client.accountsGet({
|
|
access_token: rawAccessToken,
|
|
});
|
|
const institutionId = accountsResponse.data.item?.institution_id;
|
|
const institutionName = institutionId
|
|
? await this.getInstitutionName(institutionId)
|
|
: "Plaid institution";
|
|
const incomingAccountIds = accountsResponse.data.accounts.map((account) => account.account_id);
|
|
const existingAccounts = await this.prisma.account.findMany({
|
|
where: { userId, plaidAccountId: { in: incomingAccountIds } },
|
|
select: { plaidAccountId: true },
|
|
});
|
|
const existingIds = new Set(existingAccounts.map((account) => account.plaidAccountId));
|
|
const newAccountCount = incomingAccountIds.filter((id) => !existingIds.has(id)).length;
|
|
await this.planLimits.assertCanAddAccounts(userId, newAccountCount);
|
|
|
|
for (const account of accountsResponse.data.accounts) {
|
|
await this.prisma.account.upsert({
|
|
where: { plaidAccountId: account.account_id },
|
|
update: {
|
|
institutionName,
|
|
accountType: account.subtype ?? account.type,
|
|
mask: account.mask ?? null,
|
|
plaidAccessToken: encryptedToken,
|
|
plaidItemId: itemId,
|
|
currentBalance: account.balances.current ?? null,
|
|
availableBalance: account.balances.available ?? null,
|
|
isoCurrencyCode: account.balances.iso_currency_code ?? null,
|
|
lastBalanceSync: new Date(),
|
|
userId,
|
|
},
|
|
create: {
|
|
userId,
|
|
institutionName,
|
|
accountType: account.subtype ?? account.type,
|
|
mask: account.mask ?? null,
|
|
plaidAccessToken: encryptedToken,
|
|
plaidItemId: itemId,
|
|
plaidAccountId: account.account_id,
|
|
currentBalance: account.balances.current ?? null,
|
|
availableBalance: account.balances.available ?? null,
|
|
isoCurrencyCode: account.balances.iso_currency_code ?? null,
|
|
lastBalanceSync: new Date(),
|
|
isActive: true,
|
|
},
|
|
});
|
|
}
|
|
|
|
return {
|
|
itemId,
|
|
accountCount: accountsResponse.data.accounts.length,
|
|
};
|
|
}
|
|
|
|
async syncBalancesForUser(userId: string) {
|
|
const accounts = await this.prisma.account.findMany({
|
|
where: { userId, plaidAccessToken: { not: null }, plaidAccountId: { not: null } },
|
|
});
|
|
|
|
// Deduplicate by decrypted token
|
|
const tokenMap = new Map<string, string>();
|
|
for (const acct of accounts) {
|
|
if (!acct.plaidAccessToken) continue;
|
|
const raw = this.encryption.decrypt(acct.plaidAccessToken);
|
|
tokenMap.set(acct.plaidAccessToken, raw);
|
|
}
|
|
|
|
let updated = 0;
|
|
for (const [, rawToken] of tokenMap) {
|
|
const response = await this.client.accountsBalanceGet({ access_token: rawToken });
|
|
for (const account of response.data.accounts) {
|
|
const record = await this.prisma.account.updateMany({
|
|
where: { plaidAccountId: account.account_id, userId },
|
|
data: {
|
|
currentBalance: account.balances.current ?? null,
|
|
availableBalance: account.balances.available ?? null,
|
|
isoCurrencyCode: account.balances.iso_currency_code ?? null,
|
|
lastBalanceSync: new Date(),
|
|
},
|
|
});
|
|
updated += record.count;
|
|
}
|
|
}
|
|
return { updated };
|
|
}
|
|
|
|
async syncTransactionsForUser(userId: string, startDate: string, endDate: string) {
|
|
const accounts = await this.prisma.account.findMany({
|
|
where: { userId, plaidAccessToken: { not: null }, plaidAccountId: { not: null } },
|
|
});
|
|
|
|
// Build map: raw decrypted token → Plaid account ids for one item.
|
|
const tokenMap = new Map<string, string[]>();
|
|
for (const account of accounts) {
|
|
if (!account.plaidAccessToken || !account.plaidAccountId) continue;
|
|
const raw = this.encryption.decrypt(account.plaidAccessToken);
|
|
const list = tokenMap.get(raw) ?? [];
|
|
list.push(account.plaidAccountId);
|
|
tokenMap.set(raw, list);
|
|
}
|
|
|
|
const now = new Date();
|
|
await this.prisma.account.updateMany({
|
|
where: { userId, plaidAccessToken: { not: null }, plaidAccountId: { not: null } },
|
|
data: { syncStatus: "syncing", lastSyncAttemptAt: now, lastSyncError: null },
|
|
});
|
|
|
|
let created = 0;
|
|
for (const [rawToken, plaidAccountIds] of tokenMap) {
|
|
try {
|
|
const response = await this.client.transactionsGet({
|
|
access_token: rawToken,
|
|
start_date: startDate,
|
|
end_date: endDate,
|
|
options: { count: 500, offset: 0 },
|
|
});
|
|
|
|
for (const tx of response.data.transactions) {
|
|
const account = accounts.find((acct) => acct.plaidAccountId === tx.account_id);
|
|
if (!account) continue;
|
|
const rawPayload = this.rawPayloads.encrypt(tx);
|
|
await this.prisma.transactionRaw.upsert({
|
|
where: { bankTransactionId: tx.transaction_id },
|
|
update: {
|
|
accountId: account.id,
|
|
date: new Date(tx.date),
|
|
amount: tx.amount,
|
|
description: tx.name ?? "Plaid transaction",
|
|
rawPayload,
|
|
source: "plaid",
|
|
ingestedAt: new Date(),
|
|
},
|
|
create: {
|
|
accountId: account.id,
|
|
bankTransactionId: tx.transaction_id,
|
|
date: new Date(tx.date),
|
|
amount: tx.amount,
|
|
description: tx.name ?? "Plaid transaction",
|
|
rawPayload,
|
|
ingestedAt: new Date(),
|
|
source: "plaid",
|
|
},
|
|
});
|
|
created += 1;
|
|
}
|
|
|
|
await this.prisma.account.updateMany({
|
|
where: { userId, plaidAccountId: { in: plaidAccountIds } },
|
|
data: {
|
|
syncStatus: "idle",
|
|
lastTransactionSync: new Date(),
|
|
lastSyncError: null,
|
|
syncConsecutiveFailures: 0,
|
|
},
|
|
});
|
|
} catch (error: unknown) {
|
|
const err = error as { response?: { data?: { error_message?: string; error_code?: string } }; message?: string };
|
|
const message = err.response?.data?.error_message ?? err.message ?? "Plaid transaction sync failed.";
|
|
await this.prisma.account.updateMany({
|
|
where: { userId, plaidAccountId: { in: plaidAccountIds } },
|
|
data: {
|
|
syncStatus: "error",
|
|
lastSyncError: message.slice(0, 500),
|
|
syncConsecutiveFailures: { increment: 1 },
|
|
},
|
|
});
|
|
}
|
|
}
|
|
|
|
return { created };
|
|
}
|
|
|
|
async createUpdateModeLinkToken(userId: string, accountId: string) {
|
|
const account = await this.prisma.account.findFirst({
|
|
where: {
|
|
id: accountId,
|
|
userId,
|
|
plaidAccessToken: { not: null },
|
|
plaidItemId: { not: null },
|
|
isActive: true,
|
|
},
|
|
select: {
|
|
plaidAccessToken: true,
|
|
},
|
|
});
|
|
|
|
if (!account?.plaidAccessToken) {
|
|
throw new BadRequestException("Plaid account not found for update mode.");
|
|
}
|
|
|
|
const rawAccessToken = this.encryption.decrypt(account.plaidAccessToken);
|
|
try {
|
|
const response = await this.client.linkTokenCreate({
|
|
user: { client_user_id: userId },
|
|
client_name: "LedgerOne",
|
|
country_codes: this.getCountryCodes(),
|
|
language: "en",
|
|
access_token: rawAccessToken,
|
|
redirect_uri: process.env.PLAID_REDIRECT_URI?.trim() || undefined,
|
|
webhook: process.env.PLAID_WEBHOOK_URL?.trim() || undefined,
|
|
});
|
|
|
|
return {
|
|
linkToken: response.data.link_token,
|
|
expiration: response.data.expiration,
|
|
};
|
|
} catch (error: unknown) {
|
|
const err = error as { response?: { data?: { error_message?: string } } };
|
|
const message =
|
|
err.response?.data?.error_message ?? "Plaid update-mode link token request failed.";
|
|
throw new BadRequestException(message);
|
|
}
|
|
}
|
|
|
|
async markItemRepairComplete(userId: string, accountId: string) {
|
|
const account = await this.prisma.account.findFirst({
|
|
where: {
|
|
id: accountId,
|
|
userId,
|
|
plaidItemId: { not: null },
|
|
isActive: true,
|
|
},
|
|
select: {
|
|
plaidItemId: true,
|
|
},
|
|
});
|
|
|
|
if (!account?.plaidItemId) {
|
|
throw new BadRequestException("Plaid account not found for repair completion.");
|
|
}
|
|
|
|
const updated = await this.prisma.account.updateMany({
|
|
where: {
|
|
userId,
|
|
plaidItemId: account.plaidItemId,
|
|
},
|
|
data: {
|
|
syncStatus: "idle",
|
|
lastSyncError: null,
|
|
syncConsecutiveFailures: 0,
|
|
plaidWebhookCode: "UPDATE_MODE_COMPLETED",
|
|
plaidWebhookAt: new Date(),
|
|
},
|
|
});
|
|
|
|
return { updated: updated.count };
|
|
}
|
|
|
|
async handleWebhook(payload: PlaidWebhookPayload, verificationHeader?: string, rawBody?: Buffer) {
|
|
if (this.shouldVerifyWebhooks()) {
|
|
const valid = await this.verifyWebhook(verificationHeader, rawBody);
|
|
if (!valid) {
|
|
throw new BadRequestException("Invalid Plaid webhook signature.");
|
|
}
|
|
}
|
|
|
|
const webhookType = payload.webhook_type ?? "UNKNOWN";
|
|
const webhookCode = payload.webhook_code ?? "UNKNOWN";
|
|
const itemId = payload.item_id ?? null;
|
|
const event = await this.prisma.plaidWebhookEvent.create({
|
|
data: {
|
|
itemId,
|
|
webhookType,
|
|
webhookCode,
|
|
payload: payload as Prisma.InputJsonValue,
|
|
},
|
|
});
|
|
|
|
try {
|
|
const result = await this.processWebhook(payload);
|
|
await this.prisma.plaidWebhookEvent.update({
|
|
where: { id: event.id },
|
|
data: {
|
|
status: "processed",
|
|
processedAt: new Date(),
|
|
},
|
|
});
|
|
return { received: true, processed: true, webhookType, webhookCode, ...result };
|
|
} catch (error: unknown) {
|
|
const message = error instanceof Error ? error.message : "Plaid webhook processing failed.";
|
|
this.logger.error(`Plaid webhook ${webhookType}:${webhookCode} failed: ${message}`);
|
|
await this.prisma.plaidWebhookEvent.update({
|
|
where: { id: event.id },
|
|
data: {
|
|
status: "error",
|
|
error: message.slice(0, 500),
|
|
processedAt: new Date(),
|
|
},
|
|
});
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
private async processWebhook(payload: PlaidWebhookPayload) {
|
|
if (!payload.item_id) {
|
|
return { skipped: true, reason: "missing_item_id" };
|
|
}
|
|
|
|
const accounts = await this.prisma.account.findMany({
|
|
where: { plaidItemId: payload.item_id },
|
|
select: { userId: true },
|
|
distinct: ["userId"],
|
|
});
|
|
const userIds = accounts.map((account) => account.userId);
|
|
if (userIds.length === 0) {
|
|
return { skipped: true, reason: "item_not_found" };
|
|
}
|
|
|
|
const webhookType = payload.webhook_type ?? "UNKNOWN";
|
|
const webhookCode = payload.webhook_code ?? "UNKNOWN";
|
|
await this.prisma.account.updateMany({
|
|
where: { plaidItemId: payload.item_id },
|
|
data: {
|
|
plaidWebhookCode: webhookCode,
|
|
plaidWebhookAt: new Date(),
|
|
},
|
|
});
|
|
|
|
if (
|
|
webhookType === "TRANSACTIONS" &&
|
|
["SYNC_UPDATES_AVAILABLE", "INITIAL_UPDATE", "HISTORICAL_UPDATE", "DEFAULT_UPDATE"].includes(webhookCode)
|
|
) {
|
|
const days = Number(process.env.PLAID_WEBHOOK_LOOKBACK_DAYS ?? process.env.AUTO_SYNC_LOOKBACK_DAYS ?? 30);
|
|
const endDate = new Date().toISOString().slice(0, 10);
|
|
const startDate = new Date(Date.now() - days * 24 * 60 * 60 * 1000).toISOString().slice(0, 10);
|
|
let created = 0;
|
|
for (const userId of userIds) {
|
|
const result = await this.syncTransactionsForUser(userId, startDate, endDate);
|
|
created += result.created;
|
|
}
|
|
return { usersSynced: userIds.length, created };
|
|
}
|
|
|
|
if (webhookType === "ITEM") {
|
|
return this.handleItemWebhook(payload, userIds);
|
|
}
|
|
|
|
return { skipped: true, reason: "unsupported_webhook" };
|
|
}
|
|
|
|
private async handleItemWebhook(payload: PlaidWebhookPayload, userIds: string[]) {
|
|
if (!payload.item_id) return { skipped: true, reason: "missing_item_id" };
|
|
const webhookCode = payload.webhook_code ?? "UNKNOWN";
|
|
const errorCode = payload.error?.error_code ?? webhookCode;
|
|
const errorMessage = payload.error?.error_message ?? webhookCode;
|
|
|
|
if (webhookCode === "ERROR" || errorCode === "ITEM_LOGIN_REQUIRED") {
|
|
await this.prisma.account.updateMany({
|
|
where: { plaidItemId: payload.item_id },
|
|
data: {
|
|
syncStatus: "needs_reauth",
|
|
lastSyncError: errorMessage.slice(0, 500),
|
|
syncConsecutiveFailures: { increment: 1 },
|
|
},
|
|
});
|
|
return { usersMarked: userIds.length, status: "needs_reauth" };
|
|
}
|
|
|
|
if (["PENDING_EXPIRATION", "PENDING_DISCONNECT", "NEW_ACCOUNTS_AVAILABLE"].includes(webhookCode)) {
|
|
await this.prisma.account.updateMany({
|
|
where: { plaidItemId: payload.item_id },
|
|
data: {
|
|
syncStatus: "attention_required",
|
|
lastSyncError: webhookCode,
|
|
},
|
|
});
|
|
return { usersMarked: userIds.length, status: "attention_required" };
|
|
}
|
|
|
|
if (["USER_PERMISSION_REVOKED", "USER_ACCOUNT_REVOKED"].includes(webhookCode)) {
|
|
await this.prisma.account.updateMany({
|
|
where: { plaidItemId: payload.item_id },
|
|
data: {
|
|
isActive: false,
|
|
syncStatus: "revoked",
|
|
lastSyncError: webhookCode,
|
|
},
|
|
});
|
|
return { usersMarked: userIds.length, status: "revoked" };
|
|
}
|
|
|
|
if (webhookCode === "LOGIN_REPAIRED") {
|
|
await this.prisma.account.updateMany({
|
|
where: { plaidItemId: payload.item_id },
|
|
data: {
|
|
syncStatus: "idle",
|
|
lastSyncError: null,
|
|
syncConsecutiveFailures: 0,
|
|
},
|
|
});
|
|
return { usersMarked: userIds.length, status: "repaired" };
|
|
}
|
|
|
|
return { skipped: true, reason: "unsupported_item_webhook" };
|
|
}
|
|
|
|
private shouldVerifyWebhooks() {
|
|
if (process.env.NODE_ENV === "test") return false;
|
|
return process.env.PLAID_VERIFY_WEBHOOKS !== "false";
|
|
}
|
|
|
|
private async verifyWebhook(verificationHeader?: string, rawBody?: Buffer) {
|
|
if (!verificationHeader || !rawBody) return false;
|
|
const parts = verificationHeader.split(".");
|
|
if (parts.length !== 3) return false;
|
|
|
|
const header = this.decodeJwtPart<{ alg?: string; kid?: string }>(parts[0]);
|
|
if (header.alg !== "ES256" || !header.kid) return false;
|
|
|
|
const key = await this.getWebhookKey(header.kid);
|
|
const publicKey = crypto.createPublicKey({ key, format: "jwk" } as crypto.JsonWebKeyInput);
|
|
const verifier = crypto.createVerify("SHA256");
|
|
verifier.update(`${parts[0]}.${parts[1]}`);
|
|
verifier.end();
|
|
|
|
const signature = this.ecJoseSignatureToDer(Buffer.from(parts[2], "base64url"));
|
|
if (!verifier.verify(publicKey, signature)) return false;
|
|
|
|
const claims = this.decodeJwtPart<{ iat?: number; request_body_sha256?: string }>(parts[1]);
|
|
if (!claims.iat || Math.abs(Date.now() / 1000 - claims.iat) > 300) return false;
|
|
if (!claims.request_body_sha256) return false;
|
|
|
|
const actualHash = crypto.createHash("sha256").update(rawBody).digest("hex");
|
|
return this.timingSafeEqual(actualHash, claims.request_body_sha256);
|
|
}
|
|
|
|
private async getWebhookKey(keyId: string) {
|
|
const cached = this.webhookKeys.get(keyId);
|
|
if (cached) return cached;
|
|
const response = await (this.client as unknown as {
|
|
webhookVerificationKeyGet(request: { key_id: string }): Promise<{ data: { key: JsonWebKey } }>;
|
|
}).webhookVerificationKeyGet({ key_id: keyId });
|
|
this.webhookKeys.set(keyId, response.data.key);
|
|
return response.data.key;
|
|
}
|
|
|
|
private decodeJwtPart<T>(part: string): T {
|
|
return JSON.parse(Buffer.from(part, "base64url").toString("utf8")) as T;
|
|
}
|
|
|
|
private timingSafeEqual(left: string, right: string) {
|
|
const leftBuffer = Buffer.from(left);
|
|
const rightBuffer = Buffer.from(right);
|
|
return leftBuffer.length === rightBuffer.length && crypto.timingSafeEqual(leftBuffer, rightBuffer);
|
|
}
|
|
|
|
private ecJoseSignatureToDer(signature: Buffer) {
|
|
if (signature.length !== 64) return signature;
|
|
const r = this.derInteger(signature.subarray(0, 32));
|
|
const s = this.derInteger(signature.subarray(32));
|
|
const length = r.length + s.length;
|
|
return Buffer.concat([Buffer.from([0x30, length]), r, s]);
|
|
}
|
|
|
|
private derInteger(bytes: Buffer) {
|
|
let value = bytes;
|
|
while (value.length > 1 && value[0] === 0) {
|
|
value = value.subarray(1);
|
|
}
|
|
if (value[0] & 0x80) {
|
|
value = Buffer.concat([Buffer.from([0]), value]);
|
|
}
|
|
return Buffer.concat([Buffer.from([0x02, value.length]), value]);
|
|
}
|
|
|
|
private requireEnv(name: string) {
|
|
const value = process.env[name];
|
|
if (!value) {
|
|
throw new Error(`Missing ${name} environment variable.`);
|
|
}
|
|
return value;
|
|
}
|
|
|
|
private getCountryCodes() {
|
|
return (process.env.PLAID_COUNTRY_CODES ?? "US")
|
|
.split(",")
|
|
.map((item) => item.trim())
|
|
.filter(Boolean) as CountryCode[];
|
|
}
|
|
|
|
private async getInstitutionName(institutionId: string) {
|
|
try {
|
|
const response = await this.client.institutionsGetById({
|
|
institution_id: institutionId,
|
|
country_codes: ["US" as CountryCode],
|
|
});
|
|
return response.data.institution.name ?? "Plaid institution";
|
|
} catch {
|
|
return "Plaid institution";
|
|
}
|
|
}
|
|
}
|
|
|
|
type PlaidWebhookPayload = {
|
|
webhook_type?: string;
|
|
webhook_code?: string;
|
|
item_id?: string;
|
|
environment?: string;
|
|
error?: {
|
|
error_code?: string;
|
|
error_message?: string;
|
|
};
|
|
[key: string]: unknown;
|
|
};
|