Add teller and plaid webhooks

apps/dashboard/.env-example

@@ -39,6 +39,7 @@ TELLER_CERTIFICATE=
 TELLER_CERTIFICATE_PRIVATE_KEY=
 NEXT_PUBLIC_TELLER_APPLICATION_ID=
 NEXT_PUBLIC_TELLER_ENVIRONMENT=
+TELLER_SIGNING_SECRET=
 
 # Plaid
 PLAID_CLIENT_ID=

apps/dashboard/jobs/tasks/bank/transactions/upsert.ts

@@ -47,6 +47,7 @@ export const upsertTransactions = schemaTask({
         .from("transactions")
         .upsert(formattedTransactions, {
           onConflict: "internal_id",
+          ignoreDuplicates: true,
         })
         .throwOnError();
     } catch (error) {

apps/dashboard/src/actions/institutions/exchange-public-token.ts

@@ -13,5 +13,5 @@ export const exchangePublicToken = async (token: string) => {
 
   const { data } = await plaidResponse.json();
 
-  return data.access_token;
+  return data;
 };

apps/dashboard/src/app/api/webhook/plaid/route.ts

@@ -0,0 +1,93 @@
+import { createClient } from "@midday/supabase/server";
+import { isAfter, subDays } from "date-fns";
+import { syncConnection } from "jobs/tasks/bank/sync/connection";
+import { type NextRequest, NextResponse } from "next/server";
+import { z } from "zod";
+
+// https://plaid.com/docs/api/webhooks/#configuring-webhooks
+const ALLOWED_IPS = [
+  "52.21.26.131",
+  "52.21.47.157",
+  "52.41.247.19",
+  "52.88.82.239",
+];
+
+const webhookSchema = z.object({
+  webhook_type: z.enum([
+    "TRANSACTIONS",
+    "HISTORICAL_UPDATE",
+    "INITIAL_UPDATE",
+    "TRANSACTIONS_REMOVED",
+  ]),
+  webhook_code: z.enum(["SYNC_UPDATES_AVAILABLE"]),
+  item_id: z.string(),
+  error: z
+    .object({
+      error_type: z.string(),
+      error_code: z.string(),
+      error_code_reason: z.string(),
+      error_message: z.string(),
+      display_message: z.string(),
+      request_id: z.string(),
+      causes: z.array(z.string()),
+      status: z.number(),
+    })
+    .nullable(),
+  initial_update_complete: z.boolean(),
+  historical_update_complete: z.boolean(),
+  environment: z.enum(["sandbox", "production"]),
+});
+
+export async function POST(req: NextRequest) {
+  const clientIp = req.headers.get("x-forwarded-for") || "";
+
+  if (!ALLOWED_IPS.includes(clientIp)) {
+    return NextResponse.json(
+      { error: "Unauthorized IP address" },
+      { status: 403 },
+    );
+  }
+
+  const body = await req.json();
+  const result = webhookSchema.safeParse(body);
+
+  if (!result.success) {
+    return NextResponse.json(
+      { error: "Invalid webhook payload", details: result.error.issues },
+      { status: 400 },
+    );
+  }
+
+  const supabase = createClient({ admin: true });
+
+  const { data: connectionData } = await supabase
+    .from("bank_connections")
+    .select("id, created_at")
+    .eq("reference_id", result.data.item_id)
+    .single();
+
+  if (!connectionData) {
+    return NextResponse.json(
+      { error: "Connection not found" },
+      { status: 404 },
+    );
+  }
+
+  switch (result.data.webhook_type) {
+    case "TRANSACTIONS": {
+      // Only run manual sync if the historical update is complete and the connection was created in the last 24 hours
+      const manualSync =
+        result.data.historical_update_complete &&
+        isAfter(new Date(connectionData.created_at), subDays(new Date(), 1));
+
+      await syncConnection.trigger({
+        connectionId: connectionData.id,
+        manualSync,
+      });
+
+      break;
+    }
+  }
+
+  return NextResponse.json({ success: true });
+}

apps/dashboard/src/app/api/webhook/teller/route.ts

@@ -0,0 +1,95 @@
+import { validateTellerSignature } from "@/utils/teller";
+import { createClient } from "@midday/supabase/server";
+import { isAfter, subDays } from "date-fns";
+import { syncConnection } from "jobs/tasks/bank/sync/connection";
+import { type NextRequest, NextResponse } from "next/server";
+import { z } from "zod";
+
+const webhookSchema = z.object({
+  id: z.string(),
+  payload: z.object({
+    enrollment_id: z.string().optional(),
+    reason: z.string().optional(),
+  }),
+  timestamp: z.string(),
+  type: z.enum([
+    "enrollment.disconnected",
+    "transactions.processed",
+    "account.number_verification.processed",
+    "webhook.test",
+  ]),
+});
+
+export async function POST(req: NextRequest) {
+  const text = await req.clone().text();
+  const body = await req.json();
+
+  const signatureValid = validateTellerSignature({
+    signatureHeader: req.headers.get("teller-signature"),
+    text,
+  });
+
+  if (!signatureValid) {
+    return NextResponse.json(
+      { error: "Invalid webhook signature" },
+      { status: 401 },
+    );
+  }
+
+  // Parse and validate webhook body
+  const result = webhookSchema.safeParse(body);
+
+  if (!result.success) {
+    return NextResponse.json(
+      { error: "Invalid webhook payload", details: result.error.issues },
+      { status: 400 },
+    );
+  }
+
+  const { type, payload } = result.data;
+
+  if (type === "webhook.test") {
+    return NextResponse.json({ success: true });
+  }
+
+  if (!payload.enrollment_id) {
+    return NextResponse.json(
+      { error: "Missing enrollment_id" },
+      { status: 400 },
+    );
+  }
+
+  const supabase = createClient({ admin: true });
+
+  const { data: connectionData } = await supabase
+    .from("bank_connections")
+    .select("id, created_at")
+    .eq("enrollment_id", payload.enrollment_id)
+    .single();
+
+  if (!connectionData) {
+    return NextResponse.json(
+      { error: "Connection not found" },
+      { status: 404 },
+    );
+  }
+
+  switch (type) {
+    case "transactions.processed":
+      {
+        // Only run manual sync if the connection was created in the last 24 hours
+        const manualSync = isAfter(
+          new Date(connectionData.created_at),
+          subDays(new Date(), 1),
+        );
+
+        await syncConnection.trigger({
+          connectionId: connectionData.id,
+          manualSync,
+        });
+      }
+      break;
+  }
+
+  return NextResponse.json({ success: true });
+}

apps/dashboard/src/components/modals/connect-transactions-modal.tsx

@@ -118,12 +118,13 @@ export function ConnectTransactionsModal({
     clientName: "Midday",
     product: ["transactions"],
     onSuccess: async (public_token, metadata) => {
-      const accessToken = await exchangePublicToken(public_token);
+      const { access_token, item_id } = await exchangePublicToken(public_token);
 
       setParams({
         step: "account",
         provider: "plaid",
-        token: accessToken,
+        token: access_token,
+        ref: item_id,
         institution_id: metadata.institution?.institution_id,
       });
       track({

apps/dashboard/src/components/modals/select-bank-accounts.tsx

@@ -228,7 +228,7 @@ export function SelectBankAccountsModal() {
           provider: provider as "gocardless" | "plaid" | "teller",
           accessToken: token ?? undefined,
           enrollmentId: enrollment_id ?? undefined,
-          // TODO: GoCardLess Requestion ID or Plaid Item ID
+          // GoCardLess Requestion ID or Plaid Item ID
           referenceId: ref ?? undefined,
           accounts: data.map((account) => ({
             name: account.name,

apps/dashboard/src/env.mjs

@@ -31,6 +31,7 @@ export const env = createEnv({
     WEBHOOK_SECRET_KEY: z.string(),
     AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT: z.string(),
     AZURE_DOCUMENT_INTELLIGENCE_KEY: z.string(),
+    TELLER_SIGNING_SECRET: z.string(),
   },
   /**
    * Specify your client-side environment variables schema here.
@@ -85,6 +86,7 @@ export const env = createEnv({
       process.env.AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT,
     AZURE_DOCUMENT_INTELLIGENCE_KEY:
       process.env.AZURE_DOCUMENT_INTELLIGENCE_KEY,
+    TELLER_SIGNING_SECRET: process.env.TELLER_SIGNING_SECRET,
   },
   skipValidation: !!process.env.CI || !!process.env.SKIP_ENV_VALIDATION,
 });

apps/dashboard/src/utils/teller.ts

@@ -0,0 +1,54 @@
+import crypto from "node:crypto";
+
+// https://teller.io/docs/api/webhooks#verifying-messages
+export const validateTellerSignature = (params: {
+  signatureHeader: string | null;
+  text: string;
+}): boolean => {
+  if (!params.signatureHeader) {
+    return false;
+  }
+
+  const { timestamp, signatures } = parseTellerSignatureHeader(
+    params.signatureHeader,
+  );
+
+  const threeMinutesAgo = Math.floor(Date.now() / 1000) - 3 * 60;
+
+  if (Number.parseInt(timestamp) < threeMinutesAgo) {
+    return false;
+  }
+
+  // Ensure the text is used as a raw string
+  const signedMessage = `${timestamp}.${params.text}`;
+  const calculatedSignature = crypto
+    .createHmac("sha256", process.env.TELLER_SIGNING_SECRET!)
+    .update(signedMessage)
+    .digest("hex");
+
+  // Compare calculated signature with provided signatures
+  return signatures.includes(calculatedSignature);
+};
+
+export const parseTellerSignatureHeader = (
+  header: string,
+): { timestamp: string; signatures: string[] } => {
+  const parts = header.split(",");
+  const timestampPart = parts.find((p) => p.startsWith("t="));
+  const signatureParts = parts.filter((p) => p.startsWith("v1="));
+
+  if (!timestampPart) {
+    throw new Error("No timestamp in Teller-Signature header");
+  }
+
+  const timestamp = timestampPart.split("=")[1];
+  const signatures = signatureParts
+    .map((p) => p.split("=")[1])
+    .filter((sig): sig is string => sig !== undefined);
+
+  if (!timestamp || signatures.some((sig) => !sig)) {
+    throw new Error("Invalid Teller-Signature header format");
+  }
+
+  return { timestamp, signatures };
+};

apps/engine/.dev.vars-example

@@ -3,6 +3,7 @@ GOCARDLESS_SECRET_ID=
 GOCARDLESS_SECRET_KEY=
 PLAID_CLIENT_ID=
 PLAID_SECRET=
+PLAID_ENVIRONMENT=sandbox
 TYPESENSE_API_KEY=
 TYPESENSE_ENDPOINT=
 TYPESENSE_ENDPOINT_US=

apps/engine/src/routes/auth/schema.ts

@@ -42,6 +42,9 @@ export const PlaidExchangeSchema = z
       access_token: z.string().openapi({
         example: "access_9293961c",
       }),
+      item_id: z.string().openapi({
+        example: "item_9293961c",
+      }),
     }),
   })
   .openapi("PlaidExchangeSchema");