feat: resilient background job retry & monitoring

README.md

@@ -65,6 +65,7 @@ OpenAPI: `backend/app/openapi.yaml`
 - Expenses: CRUD `/expenses`
 - Bills: CRUD `/bills`, pay/mark `/bills/{id}/pay`
 - Reminders: CRUD `/reminders`, trigger `/reminders/run`
+- Jobs: status `/jobs/status`, manual trigger `/jobs/trigger/{job_id}`
 - Insights: `/insights/monthly`, `/insights/budget-suggestion`
 
 ## MVP UI/UX Plan
@@ -173,6 +174,7 @@ finmind/
   - request count by endpoint/status
   - request duration histograms (latency, including dashboard p95 KPI)
   - reminder event counters (engagement KPI)
+- Background job monitoring is available at `/jobs/status`; scheduled jobs can be triggered manually with `/jobs/trigger/{job_id}`.
 - Logs are emitted as JSON with `request_id` and shipped to Loki via Promtail.
 - Pre-provisioned Grafana dashboard: `FinMind Operations and KPI`.
 

packages/backend/app/__init__.py

@@ -52,6 +52,11 @@ def create_app(settings: Settings | None = None) -> Flask:
     # Blueprint routes
     register_routes(app)
 
+    # Background job scheduler (retry + monitoring) — skipped in test/CI mode
+    if not os.environ.get("TESTING") and not os.environ.get("DISABLE_SCHEDULER"):
+        from .services.job_retry import init_job_scheduler
+        init_job_scheduler(app)
+
     # Backward-compatible schema patch for existing databases.
     with app.app_context():
         _ensure_schema_compatibility(app)

packages/backend/app/openapi.yaml

@@ -11,6 +11,7 @@ tags:
   - name: Expenses
   - name: Bills
   - name: Reminders
+  - name: Jobs
   - name: Insights
 paths:
   /auth/register:
@@ -385,6 +386,73 @@ paths:
             application/json:
               schema: { $ref: '#/components/schemas/Error' }
 
+  /jobs/status:
+    get:
+      summary: Get background job status
+      tags: [Jobs]
+      security: [{ bearerAuth: [] }]
+      responses:
+        '200':
+          description: Current execution state for registered background jobs
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/JobStatus'
+              example:
+                job_count: 1
+                jobs:
+                  - name: reminder_dispatch
+                    status: idle
+                    run_count: 0
+                    success_count: 0
+                    failure_count: 0
+                    last_run: null
+                    last_success: null
+                    last_failure: null
+                    last_error: null
+        '401':
+          description: Unauthorized
+          content:
+            application/json:
+              schema: { $ref: '#/components/schemas/Error' }
+
+  /jobs/trigger/{jobId}:
+    post:
+      summary: Trigger a scheduled background job
+      tags: [Jobs]
+      security: [{ bearerAuth: [] }]
+      parameters:
+        - in: path
+          name: jobId
+          required: true
+          schema: { type: string }
+      responses:
+        '200':
+          description: Job trigger scheduled
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  message: { type: string }
+              example:
+                message: "job 'reminder_dispatch' triggered"
+        '401':
+          description: Unauthorized
+          content:
+            application/json:
+              schema: { $ref: '#/components/schemas/Error' }
+        '404':
+          description: Job not found
+          content:
+            application/json:
+              schema: { $ref: '#/components/schemas/Error' }
+        '503':
+          description: Scheduler not running
+          content:
+            application/json:
+              schema: { $ref: '#/components/schemas/Error' }
+
   /reminders/bills/{billId}/schedule:
     post:
       summary: Create bill reminders with default/override offsets
@@ -587,3 +655,23 @@ components:
         message: { type: string }
         send_at: { type: string, format: date-time }
         channel: { type: string, enum: [email, whatsapp], default: email }
+    JobRecord:
+      type: object
+      properties:
+        name: { type: string }
+        status: { type: string, enum: [idle, running, success, failed] }
+        run_count: { type: integer }
+        success_count: { type: integer }
+        failure_count: { type: integer }
+        last_run: { type: string, format: date-time, nullable: true }
+        last_success: { type: string, format: date-time, nullable: true }
+        last_failure: { type: string, format: date-time, nullable: true }
+        last_error: { type: string, nullable: true }
+    JobStatus:
+      type: object
+      properties:
+        job_count: { type: integer }
+        jobs:
+          type: array
+          items:
+            $ref: '#/components/schemas/JobRecord'

packages/backend/app/routes/__init__.py

@@ -7,6 +7,7 @@
 from .categories import bp as categories_bp
 from .docs import bp as docs_bp
 from .dashboard import bp as dashboard_bp
+from .jobs import bp as jobs_bp
 
 
 def register_routes(app: Flask):
@@ -18,3 +19,4 @@ 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(jobs_bp, url_prefix="/jobs")

packages/backend/app/routes/jobs.py

@@ -0,0 +1,45 @@
+"""Job monitoring route.
+
+Endpoints
+---------
+GET /jobs/status   — returns current execution state for all registered jobs
+                     (JWT required)
+POST /jobs/trigger/<job_id>  — manually trigger a scheduled job immediately
+                               (JWT required)
+"""
+from flask import Blueprint, jsonify, current_app
+from flask_jwt_extended import jwt_required
+import logging
+
+from ..services.job_retry import job_monitor
+
+bp = Blueprint("jobs", __name__)
+logger = logging.getLogger("finmind.jobs.routes")
+
+
+@bp.get("/status")
+@jwt_required()
+def job_status():
+    """Return execution health of all registered background jobs."""
+    records = job_monitor.all_records()
+    return jsonify({
+        "job_count": len(records),
+        "jobs": [r.to_dict() for r in records],
+    })
+
+
+@bp.post("/trigger/<job_id>")
+@jwt_required()
+def trigger_job(job_id: str):
+    """Manually fire a scheduled job by its APScheduler id."""
+    scheduler = current_app.extensions.get("job_scheduler")
+    if not scheduler:
+        return jsonify(error="scheduler not running"), 503
+
+    job = scheduler.get_job(job_id)
+    if not job:
+        return jsonify(error=f"job '{job_id}' not found"), 404
+
+    job.modify(next_run_time=__import__("datetime").datetime.utcnow())
+    logger.info("Manually triggered job id=%s", job_id)
+    return jsonify(message=f"job '{job_id}' triggered"), 200