`routes/plaid.js`

182 lines
const express = require('express');
const db = require('../lib/db');
const audit = require('../lib/audit');
const { id } = require('../lib/ids');
const { encrypt, decrypt } = require('../lib/crypto');
const plaidLib = require('../lib/plaid-client');

const router = express.Router();
const DEV_USER_ID = 'user_steve';

// 0) Is Plaid wired up? Lets the UI show "Connect a bank" vs a setup hint instead of 500ing.
router.get('/api/plaid/status', async (_req, res) => {
  const configured = plaidLib.isConfigured();
  const banks = await db.query(
    `SELECT id, external_subject_id, last_sync_at, created_at
       FROM connector_account WHERE user_id = $1 AND provider = 'plaid' ORDER BY created_at DESC`,
    [DEV_USER_ID]
  );
  res.json({ configured, env: process.env.PLAID_ENV || 'sandbox', banks: banks.rows });
});

// 1) Browser asks for a Link token; we hand one back, they open Plaid Link in their browser.
router.post('/api/plaid/link/token', async (req, res) => {
  try {
    const c = plaidLib.client();
    const r = await c.linkTokenCreate({
      user: { client_user_id: req.userId || DEV_USER_ID },
      client_name: 'AbramsOS',
      products: plaidLib.PRODUCTS,
      country_codes: plaidLib.COUNTRIES,
      language: 'en',
    });
    res.json({ link_token: r.data.link_token, expiration: r.data.expiration });
  } catch (err) {
    console.error('[plaid link/token]', err.response?.data || err);
    res.status(500).json({ error: err.response?.data?.error_message || err.message });
  }
});

// 2) Plaid Link returns a public_token; we exchange it for an access_token and store encrypted.
router.post('/api/plaid/exchange', async (req, res) => {
  const { public_token, institution_name } = req.body || {};
  if (!public_token) return res.status(400).json({ error: 'public_token required' });
  try {
    const c = plaidLib.client();
    const r = await c.itemPublicTokenExchange({ public_token });
    const accessToken = r.data.access_token;
    const itemId = r.data.item_id;
    const enc = encrypt(accessToken);

    const consentId = id('consent');
    const connectorId = id('connector');

    await db.tx(async (client) => {
      await client.query(
        `INSERT INTO consent_grant (id, user_id, connector_type, scopes_json) VALUES ($1, $2, $3, $4)`,
        [consentId, DEV_USER_ID, 'plaid', JSON.stringify({ products: ['transactions'], institution: institution_name || null })]
      );
      await client.query(
        `INSERT INTO connector_account
           (id, user_id, consent_grant_id, provider, external_subject_id,
            refresh_token_encrypted, refresh_token_iv, refresh_token_tag)
         VALUES ($1, $2, $3, 'plaid', $4, $5, $6, $7)
         ON CONFLICT (provider, external_subject_id) DO UPDATE SET
           refresh_token_encrypted = EXCLUDED.refresh_token_encrypted,
           refresh_token_iv = EXCLUDED.refresh_token_iv,
           refresh_token_tag = EXCLUDED.refresh_token_tag,
           consent_grant_id = EXCLUDED.consent_grant_id`,
        [connectorId, DEV_USER_ID, consentId, itemId, enc.ciphertext, enc.iv, enc.tag]
      );
    });

    await audit.log({
      actorType: 'user',
      actorId: DEV_USER_ID,
      objectType: 'connector_account',
      objectId: connectorId,
      eventType: 'plaid_link_exchange',
      metadata: { institution: institution_name || null, item_id: itemId },
    });

    res.json({ ok: true, connector_id: connectorId });
  } catch (err) {
    console.error('[plaid exchange]', err.response?.data || err);
    res.status(500).json({ error: err.response?.data?.error_message || err.message });
  }
});

// 3) Sync transactions for a Plaid connector → upsert into purchase rows.
router.post('/api/connectors/plaid/:id/sync', async (req, res) => {
  const connectorId = req.params.id;
  const r = await db.query(
    `SELECT refresh_token_encrypted, refresh_token_iv, refresh_token_tag, sync_cursor
       FROM connector_account WHERE id = $1 AND user_id = $2 AND provider = 'plaid'`,
    [connectorId, DEV_USER_ID]
  );
  if (!r.rows.length) return res.status(404).json({ error: 'plaid connector not found' });
  const accessToken = decrypt(r.rows[0].refresh_token_encrypted, r.rows[0].refresh_token_iv, r.rows[0].refresh_token_tag);

  await audit.log({
    actorType: 'user',
    actorId: DEV_USER_ID,
    objectType: 'connector_account',
    objectId: connectorId,
    eventType: 'sync_started',
  });

  let purchases = 0;
  try {
    const c = plaidLib.client();
    // /transactions/sync — resume from the saved cursor so we only pull what's NEW.
    let cursor = r.rows[0].sync_cursor || null;
    let added = [];
    let hasMore = true;
    let calls = 0;
    while (hasMore && calls < 20) {  // safety cap; cursor is persisted so the next sync resumes
      const resp = await c.transactionsSync({ access_token: accessToken, cursor: cursor || undefined });
      added = added.concat(resp.data.added || []);
      cursor = resp.data.next_cursor;
      hasMore = resp.data.has_more;
      calls += 1;
    }

    for (const tx of added) {
      // Plaid amounts are positive for outflow (purchase), negative for inflow (refund/deposit).
      if (tx.amount == null || tx.amount <= 0) continue;
      const purchaseId = id('purchase');
      const ins = await db.query(
        `INSERT INTO purchase
           (id, user_id, source_message_id, merchant_name, merchant_domain, order_number, purchase_date, total_amount, currency, confidence, raw_extract)
         VALUES ($1, $2, NULL, $3, $4, $5, $6, $7, $8, $9, $10)
         ON CONFLICT (user_id, order_number) WHERE order_number IS NOT NULL DO NOTHING
         RETURNING id`,
        [
          purchaseId,
          DEV_USER_ID,
          tx.merchant_name || tx.name || 'Unknown',
          tx.website || null,
          tx.transaction_id,                    // Plaid's stable id
          tx.date || tx.authorized_date || new Date(),
          tx.amount,
          tx.iso_currency_code || 'USD',
          0.95,                                  // bank-statement-derived purchases are high-confidence
          tx,
        ]
      );
      if (!ins.rows.length) continue;           // already had this transaction
      purchases += 1;
      await audit.log({
        actorType: 'system',
        objectType: 'purchase',
        objectId: purchaseId,
        eventType: 'purchase_extracted',
        metadata: { source: 'plaid', merchant: tx.merchant_name || tx.name, total: tx.amount, plaid_tx_id: tx.transaction_id },
      });
    }

    await db.query(`UPDATE connector_account SET last_sync_at = now(), sync_cursor = $2 WHERE id = $1`, [connectorId, cursor]);
    await audit.log({
      actorType: 'system',
      objectType: 'connector_account',
      objectId: connectorId,
      eventType: 'sync_completed',
      metadata: { source: 'plaid', added: added.length, purchases },
    });

    res.json({ ok: true, fetched: added.length, purchases });
  } catch (err) {
    console.error('[plaid sync]', err.response?.data || err);
    await audit.log({
      actorType: 'system',
      objectType: 'connector_account',
      objectId: connectorId,
      eventType: 'sync_failed',
      metadata: { source: 'plaid', error: err.message },
    });
    res.status(500).json({ error: err.response?.data?.error_message || err.message });
  }
});

module.exports = router;