nngo/Personal-Finance-Management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

06/05 Optimize app: Add Plaid web hook, healthcheck

Browse Source
This commit is contained in:
nngo
2026-06-05 13:57:36 -04:00
parent 452374365c
commit 2f62417c9f
9 changed files with 449 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/__init__.py
+2
View File
@@ -99,6 +99,7 @@ def create_app(config_name=None):
)
logging.getLogger('app').info('Sentry initialised')
from app.routes.health import health_bp
from app.routes.auth import auth_bp
from app.routes.dashboard import dashboard_bp
from app.routes.accounts import accounts_bp
@@ -116,6 +117,7 @@ def create_app(config_name=None):
from app.routes.logs import logs_bp
from app.routes.bank_import import bank_import_bp
app.register_blueprint(health_bp)
app.register_blueprint(auth_bp)
app.register_blueprint(dashboard_bp)
app.register_blueprint(accounts_bp)
app/config.py
+4 -1
View File
@@ -34,7 +34,10 @@ class Config:
PLAID_CLIENT_ID = os.environ.get('PLAID_CLIENT_ID', '')
PLAID_SECRET = os.environ.get('PLAID_SECRET', '')
# Valid values: sandbox, production (Plaid retired the 'development' environment)
PLAID_ENV = os.environ.get('PLAID_ENV', 'sandbox')
PLAID_ENV = os.environ.get('PLAID_ENV', 'sandbox')
# Full public URL Plaid will POST transaction webhooks to (e.g. https://pfm.ngodanguyen.tech/plaid/webhook)
# Leave blank to disable webhook registration during Link token creation
PLAID_WEBHOOK_URL = os.environ.get('PLAID_WEBHOOK_URL', '')
# Schwab Developer API (OAuth 2.0)
SCHWAB_CLIENT_ID = os.environ.get('SCHWAB_CLIENT_ID', '')
app/routes/health.py
+87
View File
@@ -0,0 +1,87 @@
"""
Health check endpoint — no authentication required.
Used by uptime monitors (UptimeRobot, etc.) to verify the app and subsystems.
"""
import logging
import time
from datetime import datetime
from flask import Blueprint, jsonify
from sqlalchemy import func, text
from app.extensions import db
health_bp = Blueprint('health', __name__)
log = logging.getLogger('app.health')
@health_bp.route('/health')
def health_check():
db_result = _check_db()
cron_result = _check_crons()
status = 'ok'
if db_result['status'] != 'ok':
status = 'degraded'
if any(v.get('overdue') for v in cron_result.values() if isinstance(v, dict)):
status = 'degraded'
return jsonify({
'status': status,
'timestamp': datetime.utcnow().isoformat() + 'Z',
'db': db_result,
'crons': cron_result,
}), 200 if status == 'ok' else 503
def _check_db():
t0 = time.monotonic()
try:
db.session.execute(text('SELECT 1'))
ms = round((time.monotonic() - t0) * 1000, 1)
return {'status': 'ok', 'latency_ms': ms}
except Exception as e:
log.error('[health] DB check failed: %s', e)
return {'status': 'error', 'error': str(e)}
def _check_crons():
from app.models.fx_rate import FxRate
from app.models.ai_insight import AiInsight
from app.models.net_worth_snapshot import NetWorthSnapshot
from app.models.investment import Investment
return {
# expected daily — overdue after 48 h
'fetch_fx_rate': _stat(
db.session.query(func.max(FxRate.fetched_at)).scalar(),
max_hours=48,
),
# expected daily — overdue after 48 h
'daily_ai_insight': _stat(
db.session.query(func.max(AiInsight.created_at))
.filter(AiInsight.insight_type == 'daily_summary').scalar(),
max_hours=48,
),
# expected monthly — overdue after 35 days
'daily_snapshot': _stat(
db.session.query(func.max(NetWorthSnapshot.created_at)).scalar(),
max_hours=35 * 24,
),
# expected weekdays — overdue after 4 days
'fetch_prices': _stat(
db.session.query(func.max(Investment.last_price_update)).scalar(),
max_hours=4 * 24,
),
}
def _stat(last_run_dt, max_hours):
if last_run_dt is None:
return {'last_run': None, 'age_hours': None, 'overdue': False}
age = round((datetime.utcnow() - last_run_dt).total_seconds() / 3600, 1)
return {
'last_run': last_run_dt.isoformat() + 'Z',
'age_hours': age,
'overdue': age > max_hours,
}
app/routes/plaid.py
+95 -1
View File
@@ -6,13 +6,14 @@ from flask import (Blueprint, render_template, redirect, url_for, flash,
request, jsonify, session, current_app)
from flask_login import login_required
from app.extensions import db
from app.extensions import db, csrf as _csrf
from app.models.account import Account
from app.models.plaid_item import PlaidItem, PlaidAccount, PlaidSyncPreview
from app.services.plaid_service import (
create_link_token, exchange_public_token,
get_accounts, get_balances,
refresh_liabilities, sync_preview, import_transactions,
verify_webhook_token, auto_sync_item, update_item_webhook,
)
plaid_bp = Blueprint('plaid', __name__, url_prefix='/plaid')
@@ -388,6 +389,99 @@ def full_resync(item_db_id):
return redirect(url_for('plaid.index'))
# ── Webhook Receiver ─────────────────────────────────────────────────────────
@plaid_bp.route('/webhook', methods=['POST'])
@_csrf.exempt
def webhook():
"""
Receive Plaid transaction webhooks.
Plaid signs every request with a JWT in the Plaid-Verification header (ES256).
On TRANSACTIONS events: auto-import new transactions without requiring user confirmation.
On ITEM errors: log for operator visibility.
"""
token = request.headers.get('Plaid-Verification', '')
if not token:
log.warning('[plaid] webhook received without Plaid-Verification header')
return jsonify({'error': 'Missing verification token'}), 400
try:
verify_webhook_token(token)
except Exception as e:
log.warning('[plaid] webhook JWT verification failed: %s', e)
return jsonify({'error': 'Verification failed'}), 401
payload = request.get_json(silent=True) or {}
wh_type = payload.get('webhook_type', '')
wh_code = payload.get('webhook_code', '')
item_id = payload.get('item_id', '')
log.info('[plaid] webhook %s/%s item_id=%s', wh_type, wh_code, item_id)
if wh_type == 'TRANSACTIONS' and wh_code in (
'SYNC_UPDATES_AVAILABLE', 'DEFAULT_UPDATE',
'HISTORICAL_UPDATE', 'INITIAL_UPDATE',
):
item = PlaidItem.query.filter_by(item_id=item_id, is_active=True).first()
if not item:
log.warning('[plaid] webhook: no active item for item_id=%s', item_id)
return jsonify({'ok': True}) # 200 so Plaid doesn't keep retrying
try:
imported, skipped, removed = auto_sync_item(item)
log.info('[plaid] webhook auto-sync done: +%d skipped=%d removed=%d',
imported, skipped, removed)
except Exception as e:
log.error('[plaid] webhook auto-sync failed for item_id=%s: %s',
item_id, e, exc_info=True)
# Still return 200 — error is logged; retrying won't help an app-level error
elif wh_type == 'ITEM':
error = payload.get('error') or {}
if wh_code == 'ERROR':
log.error('[plaid] ITEM/ERROR item_id=%s code=%s msg=%s',
item_id, error.get('error_code'), error.get('error_message'))
elif wh_code == 'PENDING_EXPIRATION':
log.warning('[plaid] ITEM/PENDING_EXPIRATION item_id=%s — re-auth required soon', item_id)
elif wh_code == 'USER_PERMISSION_REVOKED':
log.warning('[plaid] ITEM/USER_PERMISSION_REVOKED item_id=%s', item_id)
return jsonify({'ok': True})
# ── Update Webhook URL for Existing Items ─────────────────────────────────────
@plaid_bp.route('/update-webhook', methods=['POST'])
@login_required
def update_webhook():
"""
Tell Plaid to use the currently configured PLAID_WEBHOOK_URL for all active items.
Call this once after adding/changing PLAID_WEBHOOK_URL in .env for items that were
connected before the webhook was configured.
"""
webhook_url = current_app.config.get('PLAID_WEBHOOK_URL', '').strip()
if not webhook_url:
flash('PLAID_WEBHOOK_URL is not set in .env — nothing to update.', 'warning')
return redirect(url_for('plaid.index'))
items = PlaidItem.query.filter_by(is_active=True).all()
updated = 0
errors = 0
for item in items:
try:
update_item_webhook(item, webhook_url)
updated += 1
except Exception as e:
log.error('[plaid] update_item_webhook failed for item %s: %s', item.item_id, e)
errors += 1
if updated:
flash(f'Webhook URL updated for {updated} item(s).', 'success')
if errors:
flash(f'{errors} item(s) failed — check app logs.', 'warning')
return redirect(url_for('plaid.index'))
# ── Disconnect ────────────────────────────────────────────────────────────────
@plaid_bp.route('/disconnect/<int:item_db_id>', methods=['POST'])
app/services/plaid_service.py
+103 -2
View File
@@ -88,14 +88,19 @@ def _post(path, payload):
def create_link_token():
"""Create a Link token for the frontend Plaid Link widget."""
data = _post('/link/token/create', {
from flask import current_app
payload = {
'user': {'client_user_id': 'pfm-user'},
'client_name': 'Personal Finance Manager',
'products': ['transactions'],
'additional_consented_products': ['liabilities'],
'country_codes': ['US'],
'language': 'en',
})
}
webhook_url = current_app.config.get('PLAID_WEBHOOK_URL', '').strip()
if webhook_url:
payload['webhook'] = webhook_url
data = _post('/link/token/create', payload)
return data['link_token']
@@ -381,3 +386,99 @@ def import_transactions(parsed_txns, next_cursor, item):
calc_balance(acct_id)
return imported, skipped
# ── Webhook Verification ──────────────────────────────────────────────────────
def verify_webhook_token(token):
"""
Verify a Plaid webhook JWT from the Plaid-Verification header.
Plaid signs webhooks with a rotating EC key (ES256).
Raises ValueError / jwt.exceptions.* on invalid tokens.
"""
import json
import jwt as pyjwt
from datetime import timezone
# Decode header without verification to extract key ID
header = pyjwt.get_unverified_header(token)
kid = header.get('kid')
if not kid:
raise ValueError('Missing kid in Plaid webhook JWT header')
# Fetch the matching public key from Plaid
data = _post('/webhook_verification_key/get', {'key_id': kid})
jwk = data.get('key', {})
if not jwk:
raise ValueError('Plaid returned empty JWK')
pub_key = pyjwt.algorithms.ECAlgorithm.from_jwk(json.dumps(jwk))
# Verify signature and standard claims
decoded = pyjwt.decode(token, pub_key, algorithms=['ES256'])
# Reject tokens issued more than 5 minutes ago (replay protection)
import time
age = time.time() - decoded.get('iat', 0)
if age > 300:
raise ValueError(f'Plaid webhook JWT is too old ({age:.0f}s)')
return decoded
# ── Auto Sync (used by webhook handler — no preview step) ────────────────────
def auto_sync_item(item):
"""
Silently fetch and import new transactions for a Plaid item.
Called from the webhook handler — skips the preview/confirm UI flow.
Also handles removed transactions by deleting them from the DB.
Returns (imported, skipped, removed_count).
"""
from app.extensions import db
from app.models.transaction import Transaction
from app.models.plaid_item import PlaidAccount
added, modified, removed, next_cursor = sync_transactions(item)
plaid_accounts = PlaidAccount.query.filter_by(item_id=item.id, is_active=True).all()
plaid_account_map = {pa.plaid_account_id: pa.pfm_account_id for pa in plaid_accounts}
cat_map = build_category_map()
parsed = []
for txn in added + modified:
if txn.get('pending', False):
continue
p = parse_transaction(txn, plaid_account_map, cat_map)
if p['account_id'] is None:
continue
parsed.append(p)
imported, skipped = import_transactions(parsed, next_cursor, item)
# Remove transactions that Plaid says are gone (e.g. pending dropped)
removed_count = 0
if removed:
for r in removed:
tid = r.get('transaction_id', '')
txn = Transaction.query.filter(
Transaction.notes.like(f'%Plaid:{tid}%')
).first()
if txn:
db.session.delete(txn)
removed_count += 1
if removed_count:
db.session.commit()
log.info('[plaid] auto_sync removed %d transaction(s) for item %s',
removed_count, item.item_id)
return imported, skipped, removed_count
def update_item_webhook(item, webhook_url):
"""Tell Plaid to send future webhooks for this item to a new URL."""
_post('/item/webhook/update', {
'access_token': item.access_token,
'webhook': webhook_url,
})
log.info('[plaid] webhook URL updated for item %s → %s', item.item_id, webhook_url)
app/templates/plaid/index.html
+33
View File
@@ -39,6 +39,39 @@
</div>
</div>
<!-- Webhook status card -->
<div class="pcard mb-4" style="border-left:3px solid {% if config.get('PLAID_WEBHOOK_URL') %}#7c3aed{% else %}#94a3b8{% endif %};">
<div class="d-flex align-items-center justify-content-between flex-wrap gap-3">
<div>
<div style="font-size:13px;font-weight:600;margin-bottom:2px;">
<i class="bi bi-bell{% if config.get('PLAID_WEBHOOK_URL') %}-fill text-primary{% else %}{% endif %} me-1"></i>
Automatic Transaction Sync (Webhooks)
</div>
{% if config.get('PLAID_WEBHOOK_URL') %}
<div style="font-size:11px;color:var(--muted);">
Webhook URL: <code style="font-size:11px;">{{ config.get('PLAID_WEBHOOK_URL') }}</code>
— Plaid will push transaction updates automatically.
</div>
{% else %}
<div style="font-size:11px;color:var(--muted);">
Not configured. Set <code>PLAID_WEBHOOK_URL=https://pfm.ngodanguyen.tech/plaid/webhook</code>
in <code>.env</code> and restart. New connections will register it automatically.
For existing items, click "Apply to Existing" after setting the URL.
</div>
{% endif %}
</div>
{% if config.get('PLAID_WEBHOOK_URL') and items %}
<form method="POST" action="{{ url_for('plaid.update_webhook') }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="btn btn-sm btn-outline-primary" style="font-size:12px;"
title="Register the configured PLAID_WEBHOOK_URL with all connected Plaid items">
<i class="bi bi-arrow-repeat me-1"></i>Apply to Existing Items
</button>
</form>
{% endif %}
</div>
</div>
{% if items %}
{% for item in items %}
<div class="pcard mb-3">