From 132290957c7130d09350585030a7972d71760445 Mon Sep 17 00:00:00 2001 From: kevin-asprec Date: Mon, 16 Mar 2026 12:30:37 +0800 Subject: [PATCH] feat(phase-5): on-prem sync poll hardening + stale job reclaim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Harden the Hub-side polling protocol for on-prem TapTrack agents: - sync/poll: accept job_failed_ids (retry/fail on-prem delivery failures) - sync/poll: deduct credits + write ledger when on-prem reports sent jobs - sync/poll: return feature_flags (tier-based) + suspended flag in config - sync/poll: skip job dispatch for suspended/expired schools - sms_jobs: add delivered_via (pull|push) + processing_started_at columns - tasks/sms: new sms.reclaim_stale_jobs task resets processing→pending if on-prem goes offline (jobs stuck >5 min), enabling Celery push fallback - tasks/sms: tag Celery-sent jobs as delivered_via='push' - worker: schedule reclaim_stale_jobs every 5 minutes - migration: 001_phase5 adds delivered_via + processing_started_at to sms_jobs --- .paul/ROADMAP.md | 6 +- .paul/STATE.md | 16 +- .paul/phases/05-onprem-sms-agent/README.md | 75 ++++++++- backend/app/models/sms.py | 2 + backend/app/routers/sync.py | 155 ++++++++++++++---- backend/app/tasks/sms.py | 58 ++++++- backend/app/worker.py | 7 +- .../001_phase5_sms_job_delivery_tracking.py | 35 ++++ 8 files changed, 303 insertions(+), 51 deletions(-) create mode 100644 backend/migrations/versions/001_phase5_sms_job_delivery_tracking.py diff --git a/.paul/ROADMAP.md b/.paul/ROADMAP.md index 85efd9f..16c0420 100644 --- a/.paul/ROADMAP.md +++ b/.paul/ROADMAP.md @@ -19,8 +19,8 @@ SMS flow: On-prem TapTrack polls Hub every 30s → Hub queues SMS jobs → Hub s ## Current Milestone **v1.0 — Foundation & Core Services** -Status: Phase 4 complete — Phase 5 next -Phases: 4 of 15 complete +Status: Phase 5 complete — Phase 6 next +Phases: 5 of 15 complete --- @@ -32,7 +32,7 @@ Phases: 4 of 15 complete | 2 | School Registry + License Mgmt | 2 | ✅ Complete | 2026-03-15 | | 3 | On-Prem License Validation | 1 | ✅ Complete | 2026-03-15 | | 4 | SMS Gateway (credits + queue) | 2 | ✅ Complete | 2026-03-15 | -| 5 | On-Prem SMS Polling Agent | TBD | Not started | — | +| 5 | On-Prem SMS Polling Agent | 1 | ✅ Complete | 2026-03-16 | | 6 | Super Admin Dashboard UI | TBD | Not started | — | | 7 | School Admin Portal UI | TBD | Not started | — | | 8 | Billing Engine + Invoice PDF | TBD | Not started | — | diff --git a/.paul/STATE.md b/.paul/STATE.md index eafcb07..e41d556 100644 --- a/.paul/STATE.md +++ b/.paul/STATE.md @@ -3,16 +3,16 @@ ## Current Position Milestone: v1.0 — Foundation & Core Services -Phase: 4 of 15 (SMS Gateway — complete) -Plan: Phase 4 complete — Phase 5 next -Status: **Phase 4 applied — ready to begin Phase 5** -Last activity: 2026-03-15 — Phase 4 complete (SMS stats/health/retry/trigger endpoints + full SmsPage dashboard with chart, KPIs, school breakdown) +Phase: 5 of 15 (On-Prem SMS Polling Agent — complete) +Plan: Phase 5 complete — Phase 6 next +Status: **Phase 5 applied — ready to begin Phase 6** +Last activity: 2026-03-16 — Phase 5 complete (sync/poll hardening: credit deduction, failed job reporting, feature flags, suspension flag, stale job reclaim) ## Loop Position ``` PLAN ──▶ APPLY ──▶ UNIFY - · · · [No active plan — Phase 2 planning next] + · · · [No active plan — Phase 6 planning next] ``` ## Progress @@ -23,7 +23,7 @@ PLAN ──▶ APPLY ──▶ UNIFY - Phase 2 (School Registry + License Mgmt): [██████████] 100% ✓ - Phase 3 (On-Prem License Validation): [██████████] 100% ✓ - Phase 4 (SMS Gateway): [██████████] 100% ✓ -- Phase 5 (On-Prem SMS Polling Agent): [░░░░░░░░░░] 0% +- Phase 5 (On-Prem SMS Polling Agent): [██████████] 100% ✓ - Phase 6 (Super Admin Dashboard UI): [░░░░░░░░░░] 0% - Phase 7 (School Admin Portal UI): [░░░░░░░░░░] 0% - Phase 8 (Billing Engine + Invoice PDF): [░░░░░░░░░░] 0% @@ -37,8 +37,8 @@ PLAN ──▶ APPLY ──▶ UNIFY ## Next Action -Run: `/paul:plan` for Phase 5 — On-Prem SMS Polling Agent -Resume file: .paul/ROADMAP.md → Phase 5 +Run: `/paul:plan` for Phase 6 — Super Admin Dashboard UI +Resume file: .paul/ROADMAP.md → Phase 6 ## Repo diff --git a/.paul/phases/05-onprem-sms-agent/README.md b/.paul/phases/05-onprem-sms-agent/README.md index 94d7ae0..0cde6a3 100644 --- a/.paul/phases/05-onprem-sms-agent/README.md +++ b/.paul/phases/05-onprem-sms-agent/README.md @@ -1,9 +1,74 @@ -# Phase 05: On-Prem SMS Polling Agent +# Phase 05: On-Prem SMS Polling Agent (Hub Side) -**Status:** Not started +**Status:** Complete +**Completed:** 2026-03-16 ## Goal -TapTrack polls Hub every 30s for pending SMS jobs; sends them; reports back completion. -## Plans -- [ ] TBD — run /paul:plan when Phase 4 is complete +Harden the Hub's side of the on-prem polling protocol: +- Credit deduction when on-prem reports delivered jobs +- Failed job reporting (on-prem couldn't send → Hub handles retry) +- Feature flags returned with every poll +- Suspension flag in config response +- Graceful degradation: stale `processing` jobs reclaimed back to `pending` if on-prem goes offline + +## Plan + +### 5-01: sync/poll enhancements + credit deduction (Hub side) + +**Changes to `backend/app/routers/sync.py`:** +- Accept `job_failed_ids: list[str]` in poll body — increment retry_count, mark failed at 5 +- Deduct 1 SMS credit per job in `report_sent_ids`, write SmsCreditLedger entries +- Fire `send_low_credit_alert` if credits fall below threshold after deductions +- Add `feature_flags` dict to config response (tier-based, reusing `_tier_features()` from licenses.py) +- Add `suspended: bool` to config response +- Return `403` with `reason` field if license is expired (not just revoked) + +**Changes to `backend/app/models/sms.py`:** +- Add `delivered_via: Mapped[str | None]` column (`"pull"` | `"push"` | None) + +**Changes to `backend/app/tasks/sms.py`:** +- Add `reclaim_stale_jobs()` task — reset `processing` jobs older than 5 minutes back to `pending` + (handles on-prem going offline mid-cycle) +- Mark jobs sent by Celery push path as `delivered_via = "push"` + +**Changes to `backend/app/worker.py`:** +- Schedule `sms.reclaim_stale_jobs` every 5 minutes + +### 5-02: Alembic migration for delivered_via column + +Add `delivered_via VARCHAR(10)` nullable to `sms_jobs` table. + +## Architecture + +``` +On-prem TapTrack (every 30s): + POST /api/sync/poll + Body: { + license_key: "TTUB-XXXXX", + report_sent_ids: ["uuid1", "uuid2"], ← jobs on-prem successfully sent + job_failed_ids: ["uuid3"] ← jobs on-prem could NOT send + } + +Hub response: + { + sms_jobs: [...], ← up to 50 pending jobs to send + config: { + sms_sender_name: "...", + sms_credits: 47.0, + school_status: "active", + suspended: false, + feature_flags: { sms: true, reports: true, ... } + } + } + +Credit flow (pull path): + On-prem sends → reports sent_ids next poll → Hub deducts 1 credit per job + (NOT deducted when job is dispatched — only when confirmed sent) + +Graceful degradation: + sms.reclaim_stale_jobs (every 5min): + UPDATE sms_jobs SET status='pending' + WHERE status='processing' AND updated_at < now() - 5min + → If on-prem dies mid-poll, jobs return to Celery push queue +``` diff --git a/backend/app/models/sms.py b/backend/app/models/sms.py index ab55b41..dfa8e0a 100644 --- a/backend/app/models/sms.py +++ b/backend/app/models/sms.py @@ -33,6 +33,8 @@ class SmsJob(Base): created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) retry_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + delivered_via: Mapped[str | None] = mapped_column(String(10), nullable=True) # "pull" | "push" + processing_started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) school: Mapped["School"] = relationship("School", back_populates="sms_jobs") diff --git a/backend/app/routers/sync.py b/backend/app/routers/sync.py index f394b95..b385908 100644 --- a/backend/app/routers/sync.py +++ b/backend/app/routers/sync.py @@ -1,4 +1,5 @@ """On-prem sync endpoint — polled by TapTrack every 30s to get SMS jobs and config.""" +import uuid from datetime import datetime, timezone from fastapi import APIRouter, HTTPException, Request from sqlalchemy.ext.asyncio import AsyncSession @@ -7,11 +8,20 @@ from fastapi import Depends from app.database import get_db from app.models.license import License, LicenseStatus -from app.models.school import School -from app.models.sms import SmsJob, SmsJobStatus +from app.models.school import School, SchoolStatus +from app.models.sms import SmsJob, SmsJobStatus, SmsCreditLedger, SmsCreditTx router = APIRouter(prefix="/api/sync", tags=["sync"]) +# Feature flags by tier — mirrors licenses.py +def _tier_features(tier: str) -> dict: + base = {"sms": True, "reports": True, "websocket": True, "multi_terminal": True} + if tier == "premium": + base.update({"api_keys": True, "webhooks": True, "bulk_enrollment": True}) + elif tier == "basic": + base.update({"multi_terminal": False, "api_keys": False, "webhooks": False}) + return base + @router.post("/poll") async def sync_poll( request: Request, @@ -20,58 +30,143 @@ async def sync_poll( """ Called by on-prem TapTrack every 30s. Returns pending SMS jobs and current config (sender_name, credits, feature flags). - Body: { license_key: str, report_sent_ids: [str] } (completed job IDs to mark as sent) + + Body: + { + "license_key": "TTUB-XXXXX-XXXXX-XXXXX", + "report_sent_ids": ["uuid1", "uuid2"], // confirmed-sent job IDs → deduct credits + "job_failed_ids": ["uuid3"] // jobs on-prem could NOT deliver → retry/fail + } """ body = await request.json() key = body.get("license_key", "") - sent_ids = body.get("report_sent_ids", []) + sent_ids: list[str] = body.get("report_sent_ids", []) + failed_ids: list[str] = body.get("job_failed_ids", []) + # Validate license lic = (await db.execute(select(License).where(License.key == key))).scalar_one_or_none() if not lic or lic.status == LicenseStatus.revoked: - raise HTTPException(403, "Invalid or revoked license") + raise HTTPException(403, detail={"reason": "revoked", "message": "Invalid or revoked license"}) + if lic.status == LicenseStatus.expired: + raise HTTPException(403, detail={"reason": "expired", "message": "License has expired"}) school = (await db.execute(select(School).where(School.id == lic.school_id))).scalar_one_or_none() if not school: - raise HTTPException(404) + raise HTTPException(404, detail="School not found") - # Mark completed jobs + now = datetime.now(timezone.utc) + + # ── Mark completed jobs (on-prem confirmed delivery) ────────────────────── if sent_ids: - await db.execute( - update(SmsJob) - .where(and_(SmsJob.id.in_(sent_ids), SmsJob.school_id == school.id)) - .values(status=SmsJobStatus.sent, sent_at=datetime.now(timezone.utc)) - ) + # Fetch jobs to deduct credits (only for this school, only processing/pending) + jobs_sent = (await db.execute( + select(SmsJob).where( + and_( + SmsJob.id.in_(sent_ids), + SmsJob.school_id == school.id, + SmsJob.status.in_([SmsJobStatus.processing, SmsJobStatus.pending]), + ) + ) + )).scalars().all() - # Get pending jobs (max 50 per poll) - pending_jobs = (await db.execute( - select(SmsJob) - .where(and_(SmsJob.school_id == school.id, SmsJob.status == SmsJobStatus.pending)) - .limit(50) - )).scalars().all() + if jobs_sent: + credit_deductions = len(jobs_sent) + new_balance = float(school.sms_credits) - credit_deductions + new_balance = max(new_balance, 0.0) # floor at 0 - # Mark as processing - job_ids = [j.id for j in pending_jobs] - if job_ids: - await db.execute( - update(SmsJob) - .where(SmsJob.id.in_(job_ids)) - .values(status=SmsJobStatus.processing) - ) + # Deduct credits from school + school.sms_credits = new_balance - # Update last seen - lic.last_validated_at = datetime.now(timezone.utc) + # Write a single consolidated ledger entry for this batch + db.add(SmsCreditLedger( + id=str(uuid.uuid4()), + school_id=school.id, + tx_type=SmsCreditTx.deduct, + amount=-float(credit_deductions), + balance_after=new_balance, + description=f"SMS delivered via on-prem agent ({credit_deductions} messages)", + reference_id=sent_ids[0] if len(sent_ids) == 1 else None, + )) + + # Mark jobs as sent, tag delivered_via=pull + await db.execute( + update(SmsJob) + .where(and_(SmsJob.id.in_([j.id for j in jobs_sent]), SmsJob.school_id == school.id)) + .values(status=SmsJobStatus.sent, sent_at=now, delivered_via="pull") + ) + + # Low credit alert — fire if below threshold + if new_balance <= float(school.sms_credit_low_threshold): + try: + from app.tasks.sms import send_low_credit_alert + send_low_credit_alert.delay(school.id) + except Exception: + pass # don't fail the poll if alert fails to queue + + # ── Handle jobs on-prem failed to send ──────────────────────────────────── + if failed_ids: + failed_jobs = (await db.execute( + select(SmsJob).where( + and_( + SmsJob.id.in_(failed_ids), + SmsJob.school_id == school.id, + ) + ) + )).scalars().all() + + for job in failed_jobs: + job.retry_count = (job.retry_count or 0) + 1 + if job.retry_count >= 5: + job.status = SmsJobStatus.failed + job.error_message = "Max retries reached — on-prem delivery failed" + else: + # Return to pending so it can be re-dispatched (pull or push) + job.status = SmsJobStatus.pending + job.error_message = f"On-prem delivery failed (attempt {job.retry_count})" + job.processing_started_at = None + + # ── Fetch pending jobs for this school ──────────────────────────────────── + # Suspended schools get an empty job list — on-prem should show banner + pending_jobs: list[SmsJob] = [] + if school.status == SchoolStatus.active: + pending_jobs = (await db.execute( + select(SmsJob) + .where(and_(SmsJob.school_id == school.id, SmsJob.status == SmsJobStatus.pending)) + .limit(50) + )).scalars().all() + + # Mark dispatched jobs as processing + if pending_jobs: + job_ids = [j.id for j in pending_jobs] + await db.execute( + update(SmsJob) + .where(SmsJob.id.in_(job_ids)) + .values(status=SmsJobStatus.processing, processing_started_at=now) + ) + + # ── Update license heartbeat ─────────────────────────────────────────────── + lic.last_validated_at = now lic.last_seen_ip = request.client.host if request.client else None + await db.commit() + suspended = school.status in (SchoolStatus.suspended, SchoolStatus.expired) + return { "sms_jobs": [ - {"id": j.id, "recipient_phone": j.recipient_phone, - "message": j.message, "sender_name": j.sender_name} + { + "id": j.id, + "recipient_phone": j.recipient_phone, + "message": j.message, + "sender_name": j.sender_name, + } for j in pending_jobs ], "config": { "sms_sender_name": school.sms_sender_name, "sms_credits": float(school.sms_credits), "school_status": school.status.value, + "suspended": suspended, + "feature_flags": _tier_features(lic.tier), }, } diff --git a/backend/app/tasks/sms.py b/backend/app/tasks/sms.py index 1d11864..80230f1 100644 --- a/backend/app/tasks/sms.py +++ b/backend/app/tasks/sms.py @@ -1,7 +1,7 @@ -"""Celery task: process pending SMS jobs via Semaphore.""" +"""Celery task: process pending SMS jobs via Semaphore (push path).""" import logging import os -from datetime import datetime, timezone +from datetime import datetime, timezone, timedelta import httpx from sqlalchemy import create_engine, select, update, and_ @@ -19,9 +19,17 @@ def _make_sync_engine(): _engine = _make_sync_engine() _Session = sessionmaker(bind=_engine) + @celery_app.task(name="sms.process_queue") def process_sms_queue(): - """Process up to 20 pending SMS jobs per run via Semaphore API.""" + """ + Push path: process up to 20 pending SMS jobs per run via Semaphore API. + + Only picks up `pending` jobs. Jobs in `processing` are being handled + by the on-prem pull agent via /api/sync/poll — do not touch them here. + The reclaim_stale_jobs task will reset stale `processing` jobs back to + `pending` if the on-prem agent stops polling. + """ from app.models.sms import SmsJob, SmsJobStatus from app.models.school import School @@ -44,6 +52,7 @@ def process_sms_queue(): job.status = SmsJobStatus.sent job.sent_at = datetime.now(timezone.utc) job.semaphore_message_id = result.get("message_id") + job.delivered_via = "push" # Deduct credit from app.models.sms import SmsCreditLedger, SmsCreditTx school.sms_credits = float(school.sms_credits) - 1.0 @@ -52,7 +61,7 @@ def process_sms_queue(): tx_type=SmsCreditTx.deduct, amount=-1.0, balance_after=float(school.sms_credits), - description=f"SMS sent to {job.recipient_phone}", + description=f"SMS sent via Semaphore to {job.recipient_phone}", reference_id=job.id, )) # Low credit alert @@ -67,6 +76,46 @@ def process_sms_queue(): finally: db.close() + +@celery_app.task(name="sms.reclaim_stale_jobs") +def reclaim_stale_jobs(): + """ + Graceful degradation: reset `processing` jobs that have been stuck for + more than 5 minutes back to `pending`. + + This handles the case where an on-prem TapTrack instance went offline + after receiving jobs from /api/sync/poll but before reporting them back. + Once reclaimed, the jobs are eligible to be picked up by sms.process_queue + (push path via Semaphore) or by the next on-prem poll. + """ + from app.models.sms import SmsJob, SmsJobStatus + + stale_cutoff = datetime.now(timezone.utc) - timedelta(minutes=5) + db = _Session() + try: + result = db.execute( + update(SmsJob) + .where( + and_( + SmsJob.status == SmsJobStatus.processing, + SmsJob.processing_started_at != None, # noqa: E711 + SmsJob.processing_started_at < stale_cutoff, + ) + ) + .values( + status=SmsJobStatus.pending, + processing_started_at=None, + error_message="Reclaimed: on-prem agent did not report back within 5 minutes", + ) + ) + reclaimed = result.rowcount + db.commit() + if reclaimed: + logger.info("sms.reclaim_stale_jobs: reclaimed %d stale processing jobs → pending", reclaimed) + finally: + db.close() + + def _send_semaphore(phone: str, message: str, sender: str) -> dict: try: with httpx.Client(timeout=15) as client: @@ -84,6 +133,7 @@ def _send_semaphore(phone: str, message: str, sender: str) -> dict: except Exception as e: return {"success": False, "error": str(e)} + @celery_app.task(name="sms.send_low_credit_alert") def send_low_credit_alert(school_id: str): """Send low credit warning email to school billing contact.""" diff --git a/backend/app/worker.py b/backend/app/worker.py index 0afbcea..48a102c 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -24,11 +24,16 @@ celery_app.conf.update( timezone="Asia/Manila", enable_utc=True, beat_schedule={ - # Process pending SMS jobs every 30 seconds + # Process pending SMS jobs every 30 seconds (push path via Semaphore) "process-sms-queue": { "task": "sms.process_queue", "schedule": 30.0, }, + # Reclaim processing jobs where on-prem went offline (every 5 minutes) + "reclaim-stale-sms-jobs": { + "task": "sms.reclaim_stale_jobs", + "schedule": 300.0, + }, # Check license expiry every day at 8am "check-license-expiry": { "task": "license.check_expiry", diff --git a/backend/migrations/versions/001_phase5_sms_job_delivery_tracking.py b/backend/migrations/versions/001_phase5_sms_job_delivery_tracking.py new file mode 100644 index 0000000..7c646e2 --- /dev/null +++ b/backend/migrations/versions/001_phase5_sms_job_delivery_tracking.py @@ -0,0 +1,35 @@ +"""phase5: add delivered_via and processing_started_at to sms_jobs + +Revision ID: 001_phase5 +Revises: +Create Date: 2026-03-16 + +Adds two columns to sms_jobs to support the on-prem polling protocol: +- delivered_via: tracks whether the job was sent via the on-prem pull agent ("pull") + or directly via the Hub's Celery push path ("push") +- processing_started_at: timestamp when the job was marked `processing` (dispatched to on-prem). + Used by sms.reclaim_stale_jobs to detect and reclaim stuck jobs. +""" +from alembic import op +import sqlalchemy as sa + +revision = "001_phase5" +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "sms_jobs", + sa.Column("delivered_via", sa.String(10), nullable=True), + ) + op.add_column( + "sms_jobs", + sa.Column("processing_started_at", sa.DateTime(timezone=True), nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("sms_jobs", "processing_started_at") + op.drop_column("sms_jobs", "delivered_via")