feat: add background job retry and monitoring system

packages/backend/app/models.py

@@ -127,9 +127,36 @@ class UserSubscription(db.Model):
     started_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
 
 
+class FinancialAccount(db.Model):
+    __tablename__ = "financial_accounts"
+    id = db.Column(db.Integer, primary_key=True)
+    user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
+    name = db.Column(db.String(100), nullable=False)
+    account_type = db.Column(db.String(30), default="checking")  # checking, savings, credit, investment, cash
+    balance = db.Column(db.Numeric(12, 2), default=0, nullable=False)
+    currency = db.Column(db.String(10), default="INR", nullable=False)
+    active = db.Column(db.Boolean, default=True, nullable=False)
+    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
+    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
+
+
 class AuditLog(db.Model):
     __tablename__ = "audit_logs"
     id = db.Column(db.Integer, primary_key=True)
     user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
     action = db.Column(db.String(100), nullable=False)
     created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
+
+
+class JobRecord(db.Model):
+    __tablename__ = "job_records"
+    id = db.Column(db.Integer, primary_key=True)
+    user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
+    job_type = db.Column(db.String(100), nullable=False)
+    args = db.Column(db.Text, default="{}", nullable=False)
+    status = db.Column(db.String(20), default="pending", nullable=False, server_default="pending")
+    retry_count = db.Column(db.Integer, default=0, nullable=False)
+    max_retries = db.Column(db.Integer, default=3, nullable=False)
+    last_error = db.Column(db.Text, nullable=True)
+    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
+    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)

packages/backend/app/routes/__init__.py

@@ -7,6 +7,9 @@
 from .categories import bp as categories_bp
 from .docs import bp as docs_bp
 from .dashboard import bp as dashboard_bp
+from .accounts import bp as accounts_bp
+from .digest import bp as digest_bp
+from .jobs import bp as jobs_bp
 
 
 def register_routes(app: Flask):
@@ -18,3 +21,6 @@ def register_routes(app: Flask):
     app.register_blueprint(categories_bp, url_prefix="/categories")
     app.register_blueprint(docs_bp, url_prefix="/docs")
     app.register_blueprint(dashboard_bp, url_prefix="/dashboard")
+    app.register_blueprint(accounts_bp, url_prefix="/accounts")
+    app.register_blueprint(digest_bp, url_prefix="/digest")
+    app.register_blueprint(jobs_bp, url_prefix="/jobs")

packages/backend/app/routes/accounts.py

@@ -0,0 +1,78 @@
+"""Financial accounts API."""
+from flask import Blueprint, jsonify, request
+from flask_jwt_extended import jwt_required, get_jwt_identity
+from ..extensions import db
+from ..models import FinancialAccount
+from datetime import datetime
+import logging
+
+bp = Blueprint("accounts", __name__)
+logger = logging.getLogger("finmind.accounts")
+
+
+@bp.get("")
+@jwt_required()
+def list_accounts():
+    uid = int(get_jwt_identity())
+    accounts = FinancialAccount.query.filter_by(user_id=uid, active=True).all()
+    return jsonify({
+        "accounts": [{
+            "id": a.id,
+            "name": a.name,
+            "account_type": a.account_type,
+            "balance": float(a.balance),
+            "currency": a.currency,
+        } for a in accounts]
+    })
+
+
+@bp.post("")
+@jwt_required()
+def create_account():
+    uid = int(get_jwt_identity())
+    data = request.get_json(force=True, silent=True) or {}
+    name = data.get("name", "").strip()
+    if not name:
+        return jsonify({"error": "Account name is required"}), 400
+    account = FinancialAccount(
+        user_id=uid, name=name,
+        account_type=data.get("account_type", "checking"),
+        balance=float(data.get("balance", 0)),
+        currency=data.get("currency", "INR"),
+    )
+    db.session.add(account)
+    db.session.commit()
+    return jsonify({"id": account.id, "name": account.name}), 201
+
+
+@bp.get("/consolidated")
+@jwt_required()
+def consolidated_view():
+    uid = int(get_jwt_identity())
+    accounts = FinancialAccount.query.filter_by(user_id=uid, active=True).all()
+    total = sum(float(a.balance) for a in accounts)
+    by_type = {}
+    for a in accounts:
+        t = a.account_type
+        by_type[t] = by_type.get(t, 0) + float(a.balance)
+    return jsonify({
+        "total_balance": round(total, 2),
+        "account_count": len(accounts),
+        "by_type": {k: round(v, 2) for k, v in by_type.items()},
+        "accounts": [{
+            "id": a.id, "name": a.name, "type": a.account_type,
+            "balance": float(a.balance), "currency": a.currency,
+        } for a in accounts],
+    })
+
+
+@bp.delete("/<int:account_id>")
+@jwt_required()
+def delete_account(account_id):
+    uid = int(get_jwt_identity())
+    account = FinancialAccount.query.filter_by(id=account_id, user_id=uid).first()
+    if not account:
+        return jsonify({"error": "Not found"}), 404
+    account.active = False
+    db.session.commit()
+    return jsonify({"ok": True})

packages/backend/app/routes/digest.py

@@ -0,0 +1,18 @@
+"""Weekly digest API endpoint."""
+from datetime import date
+from flask import Blueprint, jsonify, request
+from flask_jwt_extended import jwt_required, get_jwt_identity
+from ..services.digest import generate_weekly_digest
+import logging
+
+bp = Blueprint("digest", __name__)
+logger = logging.getLogger("finmind.digest")
+
+
+@bp.get("/weekly")
+@jwt_required()
+def weekly_digest():
+    uid = int(get_jwt_identity())
+    week_param = request.args.get("week")
+    target_date = date.fromisoformat(week_param) if week_param else None
+    return jsonify(generate_weekly_digest(uid, target_date))

packages/backend/app/routes/jobs.py

@@ -0,0 +1,86 @@
+"""Job monitoring API endpoints."""
+from flask import Blueprint, jsonify, request
+from flask_jwt_extended import jwt_required, get_jwt_identity
+from ..extensions import db
+from ..models import JobRecord
+import logging
+
+bp = Blueprint("jobs", __name__)
+logger = logging.getLogger("finmind.jobs")
+
+
+@bp.get("")
+@jwt_required()
+def list_jobs():
+    """List job records for the authenticated user."""
+    uid = get_jwt_identity()
+    page = request.args.get("page", 1, type=int)
+    per_page = request.args.get("per_page", 20, type=int)
+    status_filter = request.args.get("status", None)
+
+    query = JobRecord.query.filter_by(user_id=uid)
+    if status_filter:
+        query = query.filter_by(status=status_filter)
+
+    pagination = query.order_by(JobRecord.created_at.desc()).paginate(
+        page=page, per_page=per_page, error_out=False
+    )
+
+    return jsonify({
+        "jobs": [
+            {
+                "id": j.id,
+                "job_type": j.job_type,
+                "status": j.status,
+                "retry_count": j.retry_count,
+                "max_retries": j.max_retries,
+                "last_error": j.last_error,
+                "created_at": j.created_at.isoformat() if j.created_at else None,
+                "updated_at": j.updated_at.isoformat() if j.updated_at else None,
+            }
+            for j in pagination.items
+        ],
+        "total": pagination.total,
+        "page": page,
+        "per_page": per_page,
+    })
+
+
+@bp.get("/<int:job_id>")
+@jwt_required()
+def get_job(job_id: int):
+    """Get a specific job record."""
+    uid = get_jwt_identity()
+    job = JobRecord.query.filter_by(id=job_id, user_id=uid).first()
+    if not job:
+        return jsonify({"error": "Job not found"}), 404
+
+    return jsonify({
+        "id": job.id,
+        "job_type": job.job_type,
+        "args": job.args,
+        "status": job.status,
+        "retry_count": job.retry_count,
+        "max_retries": job.max_retries,
+        "last_error": job.last_error,
+        "created_at": job.created_at.isoformat() if job.created_at else None,
+        "updated_at": job.updated_at.isoformat() if job.updated_at else None,
+    })
+
+
+@bp.get("/stats")
+@jwt_required()
+def job_stats():
+    """Get job execution statistics."""
+    uid = get_jwt_identity()
+    total = JobRecord.query.filter_by(user_id=uid).count()
+    by_status = {}
+    for status in ["pending", "running", "completed", "failed", "dead_letter"]:
+        count = JobRecord.query.filter_by(user_id=uid, status=status).count()
+        if count > 0:
+            by_status[status] = count
+
+    return jsonify({
+        "total": total,
+        "by_status": by_status,
+    })

packages/backend/app/services/digest.py

@@ -0,0 +1,73 @@
+"""Weekly financial digest service."""
+from datetime import date, timedelta
+from sqlalchemy import extract, func
+from ..extensions import db
+from ..models import Expense, Category
+import logging
+
+logger = logging.getLogger("finmind.digest")
+
+
+def get_week_range(d=None):
+    d = d or date.today()
+    monday = d - timedelta(days=d.weekday())
+    sunday = monday + timedelta(days=6)
+    return monday, sunday
+
+
+def _spend_in_range(uid, start, end):
+    result = db.session.query(func.coalesce(func.sum(Expense.amount), 0)).filter(
+        Expense.user_id == uid, Expense.spent_at >= start,
+        Expense.spent_at <= end, Expense.expense_type != "INCOME").scalar()
+    return float(result)
+
+
+def _income_in_range(uid, start, end):
+    result = db.session.query(func.coalesce(func.sum(Expense.amount), 0)).filter(
+        Expense.user_id == uid, Expense.spent_at >= start,
+        Expense.spent_at <= end, Expense.expense_type == "INCOME").scalar()
+    return float(result)
+
+
+def _category_breakdown(uid, start, end):
+    rows = db.session.query(Expense.category_id, func.coalesce(func.sum(Expense.amount), 0)).filter(
+        Expense.user_id == uid, Expense.spent_at >= start,
+        Expense.spent_at <= end, Expense.expense_type != "INCOME").group_by(Expense.category_id).all()
+    cats = {c.id: c.name for c in Category.query.filter_by(user_id=uid).all()}
+    return {cats.get(cid, "Uncategorized"): round(float(v), 2) for cid, v in rows}
+
+
+def generate_weekly_digest(uid, target_date=None):
+    monday, sunday = get_week_range(target_date)
+    prev_monday = monday - timedelta(days=7)
+    prev_sunday = prev_monday + timedelta(days=6)
+
+    spend = _spend_in_range(uid, monday, sunday)
+    income = _income_in_range(uid, monday, sunday)
+    prev_spend = _spend_in_range(uid, prev_monday, prev_sunday)
+    breakdown = _category_breakdown(uid, monday, sunday)
+    tx_count = Expense.query.filter(
+        Expense.user_id == uid, Expense.spent_at >= monday,
+        Expense.spent_at <= sunday).count()
+
+    change = None
+    change_pct = None
+    if prev_spend > 0:
+        change = round(spend - prev_spend, 2)
+        change_pct = round(((spend - prev_spend) / prev_spend) * 100, 1)
+
+    insights = []
+    if spend > income:
+        insights.append("Spending exceeded income this week.")
+    elif income > 0 and spend / income < 0.5:
+        insights.append("Good savings rate! Under 50% of income spent.")
+    if change_pct and abs(change_pct) > 20:
+        insights.append(f"Spending {'increased' if change_pct > 0 else 'decreased'} {abs(change_pct)}% vs last week.")
+
+    return {
+        "week_range": {"start": monday.isoformat(), "end": sunday.isoformat()},
+        "summary": {"total_spent": round(spend, 2), "total_income": round(income, 2),
+                     "transaction_count": tx_count, "category_breakdown": breakdown},
+        "comparison": {"previous_week_spent": round(prev_spend, 2), "change": change, "change_pct": change_pct},
+        "insights": insights,
+    }