feat(phase-5): on-prem sync poll hardening + stale job reclaim

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
2026-03-16 12:30:37 +08:00
parent 5cab1938b1
commit 132290957c
8 changed files with 303 additions and 51 deletions
--- 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  | —          |
--- 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

--- 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
+```
--- 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")

--- 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),
        },
    }
--- 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."""
--- 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",
--- a/backend/migrations/versions/001_phase5_sms_job_delivery_tracking.py
+++ 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")