86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from datetime import datetime, timedelta
|
|
from app.db.base import get_db
|
|
from app.models.job import Job
|
|
from app.models.campaign import Campaign, CampaignStatus
|
|
from app.services.messaging import MessagingService
|
|
import logging
|
|
|
|
router = APIRouter()
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@router.post("/tick")
|
|
def worker_tick(db: Session = Depends(get_db)):
|
|
"""
|
|
Worker endpoint to process pending jobs.
|
|
In production, this would be called by a cron job or replaced with Celery.
|
|
For development, you can call this manually or set up a simple scheduler.
|
|
"""
|
|
# Get pending jobs
|
|
now = datetime.utcnow()
|
|
jobs = db.query(Job).filter(
|
|
Job.status == "pending",
|
|
Job.run_after <= now,
|
|
Job.attempts < 3
|
|
).limit(10).all()
|
|
|
|
processed = 0
|
|
|
|
for job in jobs:
|
|
logger.info(f"Processing job {job.id} of type {job.type}")
|
|
|
|
job.status = "running"
|
|
job.attempts += 1
|
|
db.commit()
|
|
|
|
try:
|
|
if job.type == "send_campaign":
|
|
campaign_id = job.payload_json.get("campaign_id")
|
|
user_id = job.user_id
|
|
|
|
messaging_service = MessagingService(db)
|
|
result = messaging_service.send_campaign_batch(campaign_id, user_id)
|
|
|
|
if result["status"] == "done" or result.get("remaining", 0) == 0:
|
|
# Campaign done
|
|
job.status = "completed"
|
|
|
|
campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first()
|
|
if campaign:
|
|
campaign.status = CampaignStatus.DONE
|
|
else:
|
|
# More batches to process
|
|
job.status = "pending"
|
|
job.run_after = datetime.utcnow() + timedelta(minutes=1)
|
|
|
|
processed += 1
|
|
|
|
else:
|
|
job.status = "failed"
|
|
job.last_error = f"Unknown job type: {job.type}"
|
|
|
|
except Exception as e:
|
|
logger.error(f"Job {job.id} failed: {str(e)}")
|
|
job.last_error = str(e)
|
|
|
|
if job.attempts >= 3:
|
|
job.status = "failed"
|
|
|
|
# Mark campaign as failed
|
|
if job.type == "send_campaign":
|
|
campaign_id = job.payload_json.get("campaign_id")
|
|
campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first()
|
|
if campaign:
|
|
campaign.status = CampaignStatus.FAILED
|
|
else:
|
|
job.status = "pending"
|
|
job.run_after = datetime.utcnow() + timedelta(minutes=5)
|
|
|
|
db.commit()
|
|
|
|
return {
|
|
"status": "ok",
|
|
"processed": processed
|
|
}
|