TapTrack-Hub/backend/app/tasks/sms.py

"""Celery task: process pending SMS jobs via Semaphore (push path)."""
import logging
import os
from datetime import datetime, timezone, timedelta

import httpx
from sqlalchemy import create_engine, select, update, and_
from sqlalchemy.orm import sessionmaker

from app.worker import celery_app
from app.config import settings

logger = logging.getLogger(__name__)

def _make_sync_engine():
    db_url = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@db:5432/taptrack_hub")
    return create_engine(db_url.replace("postgresql+asyncpg://", "postgresql://"), pool_pre_ping=True, pool_size=2)

_engine = _make_sync_engine()
_Session = sessionmaker(bind=_engine)


@celery_app.task(name="sms.process_queue")
def process_sms_queue():
    """
    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

    db = _Session()
    try:
        jobs = db.execute(
            select(SmsJob).where(SmsJob.status == SmsJobStatus.pending).limit(20)
        ).scalars().all()

        for job in jobs:
            school = db.get(School, job.school_id)
            if not school or float(school.sms_credits) <= 0:
                job.status = SmsJobStatus.cancelled
                job.error_message = "Insufficient credits"
                db.commit()
                continue

            result = _send_semaphore(job.recipient_phone, job.message, job.sender_name)
            if result["success"]:
                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
                db.add(SmsCreditLedger(
                    school_id=school.id,
                    tx_type=SmsCreditTx.deduct,
                    amount=-1.0,
                    balance_after=float(school.sms_credits),
                    description=f"SMS sent via Semaphore to {job.recipient_phone}",
                    reference_id=job.id,
                ))
                # Low credit alert
                if float(school.sms_credits) <= school.sms_credit_low_threshold:
                    send_low_credit_alert.delay(school.id)
            else:
                job.retry_count += 1
                if job.retry_count >= 5:
                    job.status = SmsJobStatus.failed
                job.error_message = result.get("error")
            db.commit()
    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:
            resp = client.post(settings.SEMAPHORE_URL, data={
                "apikey": settings.SEMAPHORE_API_KEY,
                "number": phone,
                "message": message,
                "sendername": sender,
            })
        if resp.status_code == 200:
            data = resp.json()
            msg_id = str(data[0].get("message_id", "")) if isinstance(data, list) and data else None
            return {"success": True, "message_id": msg_id}
        return {"success": False, "error": f"HTTP {resp.status_code}"}
    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."""
    from app.services.email import send_email
    from app.models.school import School
    db = _Session()
    try:
        school = db.get(School, school_id)
        if school and school.billing_email:
            send_email(
                to=school.billing_email,
                subject=f"[TapTrack Hub] Low SMS Credits — {school.name}",
                body=f"Your SMS credit balance for {school.name} is low ({float(school.sms_credits):.0f} remaining). Please top up to continue sending SMS notifications.",
            )
    finally:
        db.close()