"""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 " f"({float(school.sms_credits):.0f} credits remaining). " f"Please top up to continue sending SMS notifications." ), template_name="email/low_credit.html", context={ "contact_name": school.contact_name or school.name, "school_name": school.name, "credits_remaining": int(float(school.sms_credits)), "threshold": school.sms_credit_low_threshold, }, email_type="low_credit", school_id=school.id, ) finally: db.close()