import asyncio
import json
import time
from fastapi import APIRouter, HTTPException
from fastapi.responses import HTMLResponse, StreamingResponse
from app.models import GenericJSONResponse
debug_router = APIRouter(
prefix="/api/1rx/v1/scheduler/debug",
tags=["Debug / Inspection"],
)
def _state():
"""Lazy import to avoid circular dependency with main."""
from app import main
return {
"redis": main.redis,
"vendor_keys": main.vendor_keys,
"doctors_data": main.doctors_data,
"VENDOR_DOCTOR_MAPPING": main.VENDOR_DOCTOR_MAPPING,
"ASSIGNED_DOCTORS_KEY": main.ASSIGNED_DOCTORS_KEY,
"PRESCRIPTION_QUEUE": main.PRESCRIPTION_QUEUE,
"PRESCRIPTION_VENDOR_MAP_KEY": main.PRESCRIPTION_VENDOR_MAP_KEY,
"RX_ASSIGNMENT_MAPPING_KEY": main.RX_ASSIGNMENT_MAPPING_KEY,
"RX_PRESCRIPTION_MAPPING_KEY": main.RX_PRESCRIPTION_MAPPING_KEY,
}
@debug_router.get("/rx_assignment_mapping/{prescription_id}")
async def get_rx_assignment_mapping(prescription_id: str):
"""Get assignment mapping details for a specific prescription."""
s = _state()
data = await s["redis"].hget(s["RX_ASSIGNMENT_MAPPING_KEY"], prescription_id)
if not data:
raise HTTPException(
status_code=404,
detail=f"No assignment mapping found for prescription {prescription_id}.",
)
return GenericJSONResponse(data=json.loads(data))
@debug_router.get("/rx_assignment_mapping")
async def get_all_rx_assignment_mappings():
"""Get all assignment mappings (assignment_time, vendor_id, is_blocked)."""
s = _state()
raw = await s["redis"].hgetall(s["RX_ASSIGNMENT_MAPPING_KEY"])
if not raw:
return GenericJSONResponse(data={})
result = {}
for prescription_id, value in raw.items():
result[prescription_id] = json.loads(value)
return GenericJSONResponse(data=result)
@debug_router.get("/rx_prescription_mapping/{prescription_id}")
async def get_rx_prescription_mapping(prescription_id: str):
"""Get prescription mapping (create_time, vendor_id, tried_doctors) for a prescription."""
s = _state()
data = await s["redis"].hget(s["RX_PRESCRIPTION_MAPPING_KEY"], prescription_id)
if not data:
raise HTTPException(
status_code=404,
detail=f"No prescription mapping found for {prescription_id}.",
)
return GenericJSONResponse(data=json.loads(data))
@debug_router.get("/rx_prescription_mapping")
async def get_all_rx_prescription_mappings():
"""Get all prescription mappings."""
s = _state()
raw = await s["redis"].hgetall(s["RX_PRESCRIPTION_MAPPING_KEY"])
if not raw:
return GenericJSONResponse(data={})
result = {}
for prescription_id, value in raw.items():
result[prescription_id] = json.loads(value)
return GenericJSONResponse(data=result)
@debug_router.get("/vendor_cleanup/{vendor_id}")
async def vendor_cleanup_check(vendor_id: str):
"""
Check for inconsistencies in a vendor's Redis state:
- Doctors in pool that are also in assigned_doctors
- Duplicate doctors in pool
- Orphaned assignments (prescription assigned but no mapping)
"""
s = _state()
redis = s["redis"]
vendor_keys = s["vendor_keys"]
doctors_data = s["doctors_data"]
redis_key = vendor_keys.get(vendor_id)
if not redis_key:
raise HTTPException(status_code=404, detail=f"Vendor {vendor_id} not found.")
pool = await redis.lrange(redis_key, 0, -1)
pool = [str(d) for d in pool]
assigned_raw = await redis.hgetall(s["ASSIGNED_DOCTORS_KEY"])
assignment_mappings = await redis.hgetall(s["RX_ASSIGNMENT_MAPPING_KEY"])
issues = []
# Check for duplicate doctors in pool
seen = set()
duplicates = []
for doc in pool:
if doc in seen:
duplicates.append(doc)
seen.add(doc)
if duplicates:
issues.append(
{
"type": "duplicate_in_pool",
"message": f"Duplicate doctors in pool: {duplicates}",
"doctors": duplicates,
}
)
# Check for doctors in pool AND assigned
for prescription_id, doctor_str in assigned_raw.items():
mapping_data = assignment_mappings.get(prescription_id)
if mapping_data:
mapping = json.loads(mapping_data)
if mapping.get("vendor_id") != vendor_id:
continue
doctor_ids = doctor_str.split(",")
in_both = [d for d in doctor_ids if d in pool]
if in_both:
issues.append(
{
"type": "assigned_and_in_pool",
"message": f"Doctors {in_both} are assigned to {prescription_id} but also in pool",
"prescription_id": prescription_id,
"doctors": in_both,
}
)
# Check for orphaned assignments (in assigned_doctors but no assignment mapping)
for prescription_id, doctor_str in assigned_raw.items():
mapping_data = assignment_mappings.get(prescription_id)
if mapping_data:
mapping = json.loads(mapping_data)
if mapping.get("vendor_id") != vendor_id:
continue
else:
doctor_ids = doctor_str.split(",")
vendor_doctors = []
if doctors_data and len(doctors_data.data) > 0:
vendor_doctors = next(
(
entry.doctors
for entry in doctors_data.data
if entry.vendor == vendor_id
),
[],
)
if any(d in vendor_doctors for d in doctor_ids):
issues.append(
{
"type": "orphaned_assignment",
"message": f"Prescription {prescription_id} has assigned doctors but no assignment mapping",
"prescription_id": prescription_id,
"doctors": doctor_ids,
}
)
return GenericJSONResponse(
data={
"vendor_id": vendor_id,
"pool_size": len(pool),
"pool_unique_size": len(seen),
"pool_doctors": pool,
"issues_count": len(issues),
"issues": issues,
}
)
@debug_router.get("/state")
async def get_full_debug_state():
"""
Single endpoint showing full Redis state:
all vendor pools, all assignments, all queues, all mappings.
"""
s = _state()
redis = s["redis"]
vendor_keys = s["vendor_keys"]
state = {}
# Vendor pools
vendor_pools = {}
for vid, rkey in vendor_keys.items():
pool = await redis.lrange(rkey, 0, -1)
vendor_pools[vid] = [str(d) for d in pool]
state["vendor_pools"] = vendor_pools
# Assigned doctors
assigned_raw = await redis.hgetall(s["ASSIGNED_DOCTORS_KEY"])
state["assigned_doctors"] = {
pid: val.split(",") for pid, val in assigned_raw.items()
}
# Assignment mappings
assignment_mappings_raw = await redis.hgetall(s["RX_ASSIGNMENT_MAPPING_KEY"])
state["assignment_mappings"] = {
pid: json.loads(val) for pid, val in assignment_mappings_raw.items()
}
# Prescription mappings
prescription_mappings_raw = await redis.hgetall(s["RX_PRESCRIPTION_MAPPING_KEY"])
state["prescription_mappings"] = {
pid: json.loads(val) for pid, val in prescription_mappings_raw.items()
}
# Prescription queue
queue = await redis.lrange(s["PRESCRIPTION_QUEUE"], 0, -1)
state["prescription_queue"] = [str(p) for p in queue]
# Vendor prescription map
vendor_map_raw = await redis.hgetall(s["PRESCRIPTION_VENDOR_MAP_KEY"])
state["vendor_prescription_map"] = {
vid: json.loads(val) for vid, val in vendor_map_raw.items()
}
# Vendor config
state["vendor_config"] = s["VENDOR_DOCTOR_MAPPING"]
return GenericJSONResponse(data=state)
@debug_router.get("/live/stream")
async def live_state_stream(interval: int = 5):
"""SSE data stream endpoint — used by the dashboard."""
async def event_generator():
while True:
try:
s = _state()
redis = s["redis"]
vendor_keys = s["vendor_keys"]
state = {}
vendor_pools = {}
for vid, rkey in vendor_keys.items():
pool = await redis.lrange(rkey, 0, -1)
vendor_pools[vid] = [str(d) for d in pool]
state["vendor_pools"] = vendor_pools
assigned_raw = await redis.hgetall(s["ASSIGNED_DOCTORS_KEY"])
state["assigned_doctors"] = {
pid: val.split(",") for pid, val in assigned_raw.items()
}
assignment_mappings_raw = await redis.hgetall(
s["RX_ASSIGNMENT_MAPPING_KEY"]
)
state["assignment_mappings"] = {
pid: json.loads(val)
for pid, val in assignment_mappings_raw.items()
}
prescription_mappings_raw = await redis.hgetall(
s["RX_PRESCRIPTION_MAPPING_KEY"]
)
state["prescription_mappings"] = {
pid: json.loads(val)
for pid, val in prescription_mappings_raw.items()
}
queue = await redis.lrange(s["PRESCRIPTION_QUEUE"], 0, -1)
state["prescription_queue"] = [str(p) for p in queue]
vendor_map_raw = await redis.hgetall(s["PRESCRIPTION_VENDOR_MAP_KEY"])
state["vendor_prescription_map"] = {
vid: json.loads(val) for vid, val in vendor_map_raw.items()
}
state["vendor_config"] = s["VENDOR_DOCTOR_MAPPING"]
state["timestamp"] = int(time.time())
yield f"data: {json.dumps(state)}\n\n"
except Exception as e:
yield f"data: {json.dumps({'error': str(e)})}\n\n"
await asyncio.sleep(interval)
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
},
)
@debug_router.get("/live", response_class=HTMLResponse)
async def live_dashboard(interval: int = 5):
"""
Live HTML dashboard showing all Redis cache state in tables.
Auto-refreshes via SSE. Open in browser:
/api/1rx/v1/scheduler/debug/live?interval=5
"""
html = f"""
Scheduler - Redis Live Dashboard