feat: phone call worker

apps/server/models/chat.py

@@ -29,6 +29,8 @@ class ChatModel(BaseModel):
     agent_id = Column(UUID, ForeignKey("agent.id", ondelete="CASCADE"), index=True)
     team_id = Column(UUID, ForeignKey("team.id", ondelete="CASCADE"), index=True)
 
+    campaign_id = Column(UUID, nullable=True, index=True)
+
     voice_url = Column(String, nullable=True)
 
     parent_id = Column(UUID, ForeignKey("chat.id", ondelete="CASCADE"), index=True)

apps/worker/.env.example

@@ -4,4 +4,6 @@ REDIS_URL=redis://redis:6379/0
 DEV_SERVER_URL=https://api-dev.l3vels.xyz
 PROD_SERVER_URL=https://api.l3vels.xyz
 
+PR_DEV_SERVER_URL=https://api-pr-dev.l3agi.com
+
 SERVER_AUTH_TOKEN=secret

apps/worker/config.py

@@ -12,4 +12,6 @@ class Config:
 
     SERVER_URL = os.environ.get("SERVER_URL")
 
+    PR_DEV_SERVER_URL = os.environ.get("PR_DEV_SERVER_URL")
+
     SERVER_AUTH_TOKEN = os.environ.get("SERVER_AUTH_TOKEN")

apps/worker/main.py

@@ -1,3 +1,4 @@
+import time
 from datetime import timedelta
 
 import requests
@@ -21,6 +22,10 @@
         "task": "register-fine-tuning-tasks",
         "schedule": timedelta(minutes=2),
     },
+    "register-campaign-phone-call-tasks": {
+        "task": "register-campaign-phone-call-tasks",
+        "schedule": timedelta(minutes=2),
+    },
 }
 
 
@@ -98,5 +103,69 @@ def check_single_fine_tuning_task(fine_tuning_id: str):
     return res.json()
 
 
+@app.task(
+    name="register-campaign-phone-call-tasks",
+    autoretry_for=(Exception,),
+    retry_backoff=2,
+    max_retries=5,
+)
+def register_campaign_phone_call_tasks():
+    res = requests.get(
+        f"{Config.PR_DEV_SERVER_URL}/campaign/due",
+        headers={"Authorization": f"Bearer {Config.SERVER_AUTH_TOKEN}"},
+    )
+
+    campaigns = res.json()
+    campaign_ids = [campaign["id"] for campaign in campaigns]
+
+    for campaign in campaigns:
+        res = requests.post(
+            f"{Config.PR_DEV_SERVER_URL}/campaign/{campaign['id']}/start",
+            headers={"Authorization": f"Bearer {Config.SERVER_AUTH_TOKEN}"},
+        )
+
+        contact_ids = res.json()
+
+        for contact_id in contact_ids:
+            make_phone_call.apply_async(args=[campaign["id"], contact_id])
+
+    return campaign_ids
+
+
+@app.task(
+    name="make-phone-call",
+    autoretry_for=(Exception,),
+    retry_backoff=2,
+    max_retries=5,
+)
+def make_phone_call(campaign_id: str, contact_id: str):
+    res = requests.post(
+        f"{Config.PR_DEV_SERVER_URL}/call/campaign",
+        headers={"Authorization": f"Bearer {Config.SERVER_AUTH_TOKEN}"},
+        data={
+            "campaign_id": campaign_id,
+            "contact_id": contact_id,
+        },
+    )
+
+    call_id = res.json().get("call_id")
+
+    # Checks for the status of the phone call to make sure phone call is finished
+    while True:
+        status_res = requests.get(
+            f"{Config.PR_DEV_SERVER_URL}/call/{call_id}",
+            headers={"Authorization": f"Bearer {Config.SERVER_AUTH_TOKEN}"},
+        )
+
+        status = status_res.json().get("status")
+
+        if status is not None:
+            break
+
+        time.sleep(15)
+
+    return "Phone call finished"
+
+
 if __name__ == "__main__":
     app.start()