Add Twilio call queue service with channel allocation
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
100
backend/services/twilio_service.py
Normal file
100
backend/services/twilio_service.py
Normal file
@@ -0,0 +1,100 @@
|
||||
"""Twilio call queue and media stream service"""
|
||||
|
||||
import time
|
||||
import threading
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class TwilioService:
|
||||
"""Manages Twilio call queue, channel allocation, and media streams"""
|
||||
|
||||
FIRST_REAL_CHANNEL = 3
|
||||
|
||||
def __init__(self):
|
||||
self._queue: list[dict] = []
|
||||
self.active_calls: dict[str, dict] = {}
|
||||
self._allocated_channels: set[int] = set()
|
||||
self._caller_counter: int = 0
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def add_to_queue(self, call_sid: str, phone: str):
|
||||
with self._lock:
|
||||
self._queue.append({
|
||||
"call_sid": call_sid,
|
||||
"phone": phone,
|
||||
"queued_at": time.time(),
|
||||
})
|
||||
print(f"[Twilio] Caller {phone} added to queue (SID: {call_sid})")
|
||||
|
||||
def remove_from_queue(self, call_sid: str):
|
||||
with self._lock:
|
||||
self._queue = [c for c in self._queue if c["call_sid"] != call_sid]
|
||||
print(f"[Twilio] Caller {call_sid} removed from queue")
|
||||
|
||||
def get_queue(self) -> list[dict]:
|
||||
now = time.time()
|
||||
with self._lock:
|
||||
return [
|
||||
{
|
||||
"call_sid": c["call_sid"],
|
||||
"phone": c["phone"],
|
||||
"wait_time": int(now - c["queued_at"]),
|
||||
}
|
||||
for c in self._queue
|
||||
]
|
||||
|
||||
def allocate_channel(self) -> int:
|
||||
with self._lock:
|
||||
ch = self.FIRST_REAL_CHANNEL
|
||||
while ch in self._allocated_channels:
|
||||
ch += 1
|
||||
self._allocated_channels.add(ch)
|
||||
return ch
|
||||
|
||||
def release_channel(self, channel: int):
|
||||
with self._lock:
|
||||
self._allocated_channels.discard(channel)
|
||||
|
||||
def take_call(self, call_sid: str) -> dict:
|
||||
caller = None
|
||||
with self._lock:
|
||||
for c in self._queue:
|
||||
if c["call_sid"] == call_sid:
|
||||
caller = c
|
||||
break
|
||||
if caller:
|
||||
self._queue = [c for c in self._queue if c["call_sid"] != call_sid]
|
||||
|
||||
if not caller:
|
||||
raise ValueError(f"Call {call_sid} not in queue")
|
||||
|
||||
channel = self.allocate_channel()
|
||||
self._caller_counter += 1
|
||||
name = f"Caller #{self._caller_counter}"
|
||||
|
||||
call_info = {
|
||||
"call_sid": call_sid,
|
||||
"phone": caller["phone"],
|
||||
"channel": channel,
|
||||
"name": name,
|
||||
"started_at": time.time(),
|
||||
}
|
||||
self.active_calls[call_sid] = call_info
|
||||
print(f"[Twilio] {name} ({caller['phone']}) taken on air — channel {channel}")
|
||||
return call_info
|
||||
|
||||
def hangup(self, call_sid: str):
|
||||
call_info = self.active_calls.pop(call_sid, None)
|
||||
if call_info:
|
||||
self.release_channel(call_info["channel"])
|
||||
print(f"[Twilio] {call_info['name']} hung up — channel {call_info['channel']} released")
|
||||
|
||||
def reset(self):
|
||||
with self._lock:
|
||||
for call_info in self.active_calls.values():
|
||||
self._allocated_channels.discard(call_info["channel"])
|
||||
self._queue.clear()
|
||||
self.active_calls.clear()
|
||||
self._allocated_channels.clear()
|
||||
self._caller_counter = 0
|
||||
print("[Twilio] Service reset")
|
||||
67
tests/test_twilio_service.py
Normal file
67
tests/test_twilio_service.py
Normal file
@@ -0,0 +1,67 @@
|
||||
import sys
|
||||
sys.path.insert(0, "/Users/lukemacneil/ai-podcast")
|
||||
|
||||
from backend.services.twilio_service import TwilioService
|
||||
|
||||
|
||||
def test_queue_starts_empty():
|
||||
svc = TwilioService()
|
||||
assert svc.get_queue() == []
|
||||
|
||||
|
||||
def test_add_caller_to_queue():
|
||||
svc = TwilioService()
|
||||
svc.add_to_queue("CA123", "+15125550142")
|
||||
q = svc.get_queue()
|
||||
assert len(q) == 1
|
||||
assert q[0]["call_sid"] == "CA123"
|
||||
assert q[0]["phone"] == "+15125550142"
|
||||
assert "wait_time" in q[0]
|
||||
|
||||
|
||||
def test_remove_caller_from_queue():
|
||||
svc = TwilioService()
|
||||
svc.add_to_queue("CA123", "+15125550142")
|
||||
svc.remove_from_queue("CA123")
|
||||
assert svc.get_queue() == []
|
||||
|
||||
|
||||
def test_allocate_channel():
|
||||
svc = TwilioService()
|
||||
ch1 = svc.allocate_channel()
|
||||
ch2 = svc.allocate_channel()
|
||||
assert ch1 == 3
|
||||
assert ch2 == 4
|
||||
svc.release_channel(ch1)
|
||||
ch3 = svc.allocate_channel()
|
||||
assert ch3 == 3
|
||||
|
||||
|
||||
def test_take_call():
|
||||
svc = TwilioService()
|
||||
svc.add_to_queue("CA123", "+15125550142")
|
||||
result = svc.take_call("CA123")
|
||||
assert result["call_sid"] == "CA123"
|
||||
assert result["channel"] >= 3
|
||||
assert svc.get_queue() == []
|
||||
assert svc.active_calls["CA123"]["channel"] == result["channel"]
|
||||
|
||||
|
||||
def test_hangup_real_caller():
|
||||
svc = TwilioService()
|
||||
svc.add_to_queue("CA123", "+15125550142")
|
||||
svc.take_call("CA123")
|
||||
ch = svc.active_calls["CA123"]["channel"]
|
||||
svc.hangup("CA123")
|
||||
assert "CA123" not in svc.active_calls
|
||||
assert ch not in svc._allocated_channels
|
||||
|
||||
|
||||
def test_caller_counter_increments():
|
||||
svc = TwilioService()
|
||||
svc.add_to_queue("CA1", "+15125550001")
|
||||
svc.add_to_queue("CA2", "+15125550002")
|
||||
r1 = svc.take_call("CA1")
|
||||
r2 = svc.take_call("CA2")
|
||||
assert r1["name"] == "Caller #1"
|
||||
assert r2["name"] == "Caller #2"
|
||||
Reference in New Issue
Block a user