import asyncio import json import logging import os import signal import time from contextlib import asynccontextmanager from datetime import datetime, timedelta from logging.handlers import TimedRotatingFileHandler from pathlib import Path from threading import Thread from typing import Optional import aiohttp try: import aioredis # type: ignore except Exception: import redis.asyncio as aioredis import httpx from apscheduler.schedulers.background import BackgroundScheduler from fastapi import APIRouter, BackgroundTasks, FastAPI, HTTPException, Request from fastapi.responses import JSONResponse from pydantic import ValidationError from app.config import RX_MAX_BLOCK_TIME, RX_MAX_HOLD_TIME, RX_NOTIFICATION_INTERVAL from app.models import ( AddDoctorResponse, AddDoctorResponseData, AddVendorResponse, AddVendorResponseData, AllAvailableDoctorsResponse, AssignDoctorRequest, AssignDoctorResponse, AssignDoctorResponseData, AssignedDoctorPrescriptionResponse, AssignedDoctorPrescriptionResponseData, BlockDoctorRequest, BlockDoctorResponse, BlockDoctorResponseData, CachedDoctorsResponse, CachedDoctorsResponseData, DeleteDoctorResponse, DeleteDoctorResponseData, DoctorRequest, DoctorResponseData, GenericJSONResponse, HoldPrescriptionRequest, HoldPrescriptionResponse, HoldPrescriptionResponseData, PrescriptionsInQueueResponse, PrescriptionsInQueueResponseData, ReassignDoctorResponse, ReassignDoctorResponseData, ReleaseDoctorRequest, ReleaseDoctorResponse, ReleaseDoctorResponseData, UpdateVendorRequest, UpdateVendorResponse, UpdateVendorResponseData, ) log_dir = Path(__file__).parent.parent / "logs" log_dir.mkdir(exist_ok=True) log_file_name = log_dir / "scheduler_log.log" # Clear all handlers to prevent duplication root_logger = logging.getLogger() root_logger.handlers.clear() file_handler = TimedRotatingFileHandler( log_file_name, when="H", interval=1, backupCount=24, ) file_handler.setLevel(logging.INFO) console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) formatter = logging.Formatter( fmt="%(asctime)s [%(levelname)s:%(process)d] - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) # Attach formatter to handlers file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) root_logger.addHandler(file_handler) root_logger.addHandler(console_handler) root_logger.setLevel(logging.INFO) logging.getLogger("apscheduler").setLevel(logging.WARNING) logger = logging.getLogger(__name__) def cleanup_old_logs(log_directory: Path, days_to_keep: int = 30): current_time = time.time() for log_file in log_directory.iterdir(): if log_file.is_file(): file_age = current_time - log_file.stat().st_mtime if file_age > days_to_keep * 86400: os.remove(log_file) logger.info(f"Deleted old log file: {log_file}") cleanup_old_logs(log_dir, days_to_keep=10) REDIS_HOST = os.getenv("REDIS_HOST", "localhost") REDIS_PORT = os.getenv("REDIS_PORT", "6379") REDIS_DB = os.getenv("REDIS_DB", "0") ONERX_SERVER = os.getenv("ONERX_SERVER", "http://192.168.1.31:8080") VENDOR_DOCTOR_MAPPING = {} @asynccontextmanager async def lifespan(app: FastAPI): global redis_client, main_loop redis_client = aioredis.from_url( f"redis://{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}", encoding="utf-8", decode_responses=True, ) global doctors_data global VENDOR_DOCTOR_MAPPING doctors_data = await get_all_available_doctors() for vendor_info in doctors_data.data: vendor_id = vendor_info.vendor doctor_pool_key = f"doctor_pool_{vendor_id}" vendor_keys[vendor_id] = doctor_pool_key VENDOR_DOCTOR_MAPPING[vendor_id] = { "concurrent_doctors": vendor_info.concurrent_doctors, "immediate_unblock": vendor_info.immediate_unblock, "rx_max_hold_time": vendor_info.rx_max_hold_time, "rx_max_block_time": vendor_info.rx_max_block_time, "rx_notification_interval": vendor_info.rx_notification_interval, "delay_after_hold": vendor_info.delay_after_hold, } await initialize_doctor_pool_for_vendor(redis, vendor_id) main_loop = asyncio.get_event_loop() # Capture the main event loop thread = Thread(target=start_scheduler) thread.daemon = True thread.start() yield await redis_client.close() logger.info("Cleanup completed.") app = FastAPI( title="Doctor Assignment API", description="This API allows clients to assign and block doctors for specific prescriptions.", version="1.0.0", docs_url="/api/docs", # Custom URL for the Swagger docs redoc_url="/api/redoc", # Custom URL for ReDoc docs lifespan=lifespan, ) api_router = APIRouter( prefix="/api/1rx/v1/scheduler", tags=["Scheduler Service"], ) redis = aioredis.from_url( f"redis://{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}", encoding="utf-8", decode_responses=True, ) doctors_data = None vendor_keys = {} # Initialize the scheduler scheduler = BackgroundScheduler() async def get_all_available_doctors() -> AllAvailableDoctorsResponse: base_url = ONERX_SERVER endpoint = f"{base_url}/api/1rx/v1/scheduler/get-all-available-doctors" try: async with httpx.AsyncClient() as client: response = await client.get(endpoint) response.raise_for_status() data = response.json() # Attempt to validate the response against the Pydantic model return AllAvailableDoctorsResponse.model_validate(data) except (httpx.HTTPStatusError, httpx.RequestError) as e: logger.error(f"Failed to retrieve doctors data: {e}") return AllAvailableDoctorsResponse(data=[], error="Failed to fetch doctor data") except ValidationError as e: logger.error(f"Validation error in doctors data: {e}") return AllAvailableDoctorsResponse( data=[], error="Invalid data format from API" ) except Exception as e: logger.error(f"Unexpected error: {e}") return AllAvailableDoctorsResponse(data=[], error="Unexpected error occurred") # Redis key for assigned doctors ASSIGNED_DOCTORS_KEY = "assigned_doctors" # Redis key for prescription queue PRESCRIPTION_QUEUE = "prescription_queue" RX_PRESCRIPTION_MAPPING_KEY = "rx_prescription_mapping" RX_ASSIGNMENT_MAPPING_KEY = "rx_assignment_mapping" DOCTORS_POOL = "doctor_pool" # Per-vendor Redis SET of currently logged-in doctors. The pool list alone cannot # represent "logged in", because a doctor is temporarily removed from the pool # while a prescription is being offered to them. This set is the source of truth # for activity so the re-add paths never resurrect a doctor who has logged off. ACTIVE_DOCTORS_POOL = "active_doctors" # Redis SET of vendor_ids for which activity tracking is enabled. This is the # durable "are we tracking this vendor" signal, and it is deliberately NOT # derived from whether the per-vendor active-doctors set exists: Redis deletes a # set when its last member is removed, so an active set that is merely empty # (every doctor logged off) is indistinguishable via exists() from a vendor we # never tracked. Without this durable flag, removing the last active doctor would # make exists(active_key) read as "not tracked" and wrongly resurrect that doctor # into the pool on their next release/expiry. ACTIVE_DOCTORS_TRACKED_KEY = "active_doctors_tracked" # External endpoints (assuming URLs for demonstration purposes) CANCEL_PX_URL = f"{ONERX_SERVER}/api/1rx/v1/scheduler/cancel-rx" ASSIGN_PX_URL = f"{ONERX_SERVER}/api/1rx/v1/scheduler/assign-rx" # Tells onerx-service that these doctors were offered an order and let it time out, so it can move # them to Away. Responds with the ids it actually moved. DOCTOR_NOT_RESPONDING_URL = f"{ONERX_SERVER}/api/1rx/v1/scheduler/doctor-not-responding" # Constants and logging ASSIGMENT_EXPIRY_DURATION = 30 # expiry in seconds PRESCRIPTION_VENDOR_MAP_KEY = "prescription_vendor_map" async def safe_rpush_doctor(redis_conn, redis_key: str, doctor_id: str) -> bool: """ Safely add a doctor back to the pool only if they are not already present. Prevents duplicate entries in the doctor pool list. Returns True if the doctor was added, False if already in the pool. Uses an atomic Lua script to prevent race conditions. """ # Never resurrect a doctor who has logged off. The active-doctors set is the # source of truth for who is logged in: if tracking is enabled for this vendor # and the doctor is not a member, skip the re-add. (If the vendor is not # tracked we are not managing activity for it, so fall back to old behaviour.) # # Tracking is gated on the durable ACTIVE_DOCTORS_TRACKED_KEY set, NOT on # exists(active_key): Redis deletes a set when its last member is removed, so # removing the last active doctor deletes the per-vendor active set. An # exists() check would then read as "not tracked" and wrongly re-add that # just-removed doctor on the following release/expiry. active_key = f"{ACTIVE_DOCTORS_POOL}{redis_key[len(DOCTORS_POOL):]}" vendor_id = redis_key[len(DOCTORS_POOL) + 1 :] if await redis_conn.sismember( ACTIVE_DOCTORS_TRACKED_KEY, vendor_id ) and not await redis_conn.sismember(active_key, doctor_id): logger.info( f"Doctor {doctor_id} is not active (logged off); not re-adding to pool {redis_key}." ) return False lua_script = """ local exists = redis.call('LPOS', KEYS[1], ARGV[1]) if exists then return 0 end redis.call('RPUSH', KEYS[1], ARGV[1]) return 1 """ result = await redis_conn.eval(lua_script, 1, redis_key, doctor_id) if result == 0: logger.info( f"Doctor {doctor_id} already in pool {redis_key}, skipping duplicate add." ) return False return True async def initialize_doctor_pool_for_vendor(redis, vendor_id: str): """ Initialize the doctor pool for the given vendor ID. Clears the existing Redis key and populates it with the doctor IDs for the given vendor. """ logger.info(f"Initializing doctor pool for {vendor_id}") doctors = None global doctors_data # Get doctors for the specific vendor if doctors_data and len(doctors_data.data) > 0: doctors = next( (entry.doctors for entry in doctors_data.data if entry.vendor == vendor_id) ) # Determine Redis key for the vendor redis_key = f"{DOCTORS_POOL}_{vendor_id}" # Clear the existing doctor pool in Redis for the vendor await redis.delete(redis_key) # Reset the active-doctors set (source of truth for logged-in doctors). active_key = f"{ACTIVE_DOCTORS_POOL}_{vendor_id}" await redis.delete(active_key) # Mark this vendor as tracked. This durable flag — not the existence of the # active set above — is what gates the re-add guard in safe_rpush_doctor, so # it must be set even when the vendor currently has zero doctors. await redis.sadd(ACTIVE_DOCTORS_TRACKED_KEY, vendor_id) logger.info(f"ACTIVE-TRACKING enabled for vendor {vendor_id}.") # Add each doctor to the vendor's Redis pool (deduplicated) if doctors: unique_doctors = list(dict.fromkeys(doctors)) await redis.rpush(redis_key, *unique_doctors) await redis.sadd(active_key, *unique_doctors) logger.info( f"ACTIVE-SET seeded for vendor {vendor_id} ({active_key}): {unique_doctors}" ) logger.info( f"Doctor ID's: {unique_doctors} added to the pool for vendor {vendor_id}" ) @app.on_event("startup") async def startup_event(): """ This function runs when the server starts and initializes the doctor pools for all vendors. """ logger.info("Server starting, initializing doctor pools for all vendors.") try: global vendor_keys global doctors_data doctors_data = await get_all_available_doctors() for vendor_info in doctors_data.data: vendor_id = vendor_info.vendor await initialize_doctor_pool_for_vendor(redis, vendor_id) finally: await redis.close() logger.info("Doctor pools initialized for all vendors.") thread = Thread(target=start_scheduler) thread.daemon = True # Ensure the thread is killed when the app stops thread.start() @app.exception_handler(HTTPException) async def custom_http_exception_handler(request: Request, exc: HTTPException): return GenericJSONResponse( data=None, error={"message": exc.detail}, status_code=exc.status_code ) async def add_prescription_to_queue( prescription_id, vendor_id, hold_time=0, tried_doctors=None ): current_time = int(time.time()) await redis.rpush(PRESCRIPTION_QUEUE, prescription_id) await add_prescription_to_vendor_map( vendor_id=vendor_id, prescription_id=prescription_id ) prescription_mapping_data = await redis.hget( RX_PRESCRIPTION_MAPPING_KEY, prescription_id ) if not prescription_mapping_data: mapping = { "prescription_create_time": current_time + hold_time, "vendor_id": vendor_id, } if tried_doctors: mapping["tried_doctors"] = tried_doctors value = json.dumps(mapping) await redis.hset(RX_PRESCRIPTION_MAPPING_KEY, prescription_id, value) else: # Update existing mapping with tried_doctors only. # prescription_create_time is intentionally NOT updated here — preserving the # original value so the expiry clock keeps ticking and the cancel API can fire. # (This matches pre-PR behaviour: no else branch existed before, so the # existing create_time was always left unchanged on re-queue.) data = json.loads(prescription_mapping_data) if tried_doctors: existing_tried = data.get("tried_doctors", []) existing_tried.extend(tried_doctors) data["tried_doctors"] = list(set(existing_tried)) await redis.hset(RX_PRESCRIPTION_MAPPING_KEY, prescription_id, json.dumps(data)) logger.info( f"Prescription {prescription_id} mapped with create_time {current_time} in rx_prescription_mapping." ) ### Assign Multiple Doctors to a Prescription ### @api_router.post( "/assign_doctor", operation_id="assign_doctor", ) async def assign_doctor( assign_request: AssignDoctorRequest, ) -> AssignDoctorResponse: """ API to assign up to n doctors from the vendor's pool to a prescription. If no doctors are available, add the prescription to a queue. """ prescription_id = assign_request.prescription_id vendor_id = assign_request.vendor_id global vendor_keys redis_key = vendor_keys.get(vendor_id) if not redis_key: logger.info(f"Vendor {vendor_id} not found.") raise HTTPException(status_code=404, detail=f"Vendor {vendor_id} not found.") # Check if the prescription is already assigned to doctors assigned_doctors = await redis.hget(ASSIGNED_DOCTORS_KEY, prescription_id) if assigned_doctors: logger.error(f"Prescription {prescription_id} already has doctors assigned.") raise HTTPException( status_code=400, detail=f"Prescription {prescription_id} already has doctors assigned.", ) # Try to pop up to n doctors from the pool assigned_doctor_ids = [] global doctors_data concurrent_doctors = None immediate_unblock = None if doctors_data and len(doctors_data.data) > 0: result = next( ( (entry.concurrent_doctors, entry.immediate_unblock) for entry in doctors_data.data if entry.vendor == vendor_id ), None, # Default value if the vendor_id is not found ) if result is not None: concurrent_doctors, immediate_unblock = result else: logger.error(f"No data found for vendor_id: {vendor_id}") raise HTTPException( status_code=404, detail=f"No data found for vendor_id: {vendor_id}" ) if concurrent_doctors: # Get tried_doctors for this prescription to implement round-robin tried_doctors = [] prescription_mapping_data = await redis.hget( RX_PRESCRIPTION_MAPPING_KEY, prescription_id ) if prescription_mapping_data: data = json.loads(prescription_mapping_data) tried_doctors = data.get("tried_doctors", []) # Use lrange + lrem to skip previously tried doctors pool_doctors = await redis.lrange(redis_key, 0, -1) pool_doctors = [str(d) for d in pool_doctors] # Separate untried and tried doctors untried = [d for d in pool_doctors if d not in tried_doctors] tried_available = [d for d in pool_doctors if d in tried_doctors] # If all doctors have been tried, reset (give everyone another chance) if not untried and tried_available: logger.info( f"All doctors tried for prescription {prescription_id}, resetting tried list." ) untried = tried_available tried_doctors = [] # Pick from untried doctors first. Only claim a doctor if lrem actually # removed them from the pool (returns >= 1); otherwise a concurrent # assign_doctor call already took them and appending here would offer the # same doctor to two prescriptions. for doctor_id in untried: if len(assigned_doctor_ids) >= concurrent_doctors: break removed = await redis.lrem(redis_key, 1, doctor_id) if removed: assigned_doctor_ids.append(doctor_id) # If still need more, use tried doctors if len(assigned_doctor_ids) < concurrent_doctors: for doctor_id in tried_available: if len(assigned_doctor_ids) >= concurrent_doctors: break if doctor_id not in assigned_doctor_ids: removed = await redis.lrem(redis_key, 1, doctor_id) if removed: assigned_doctor_ids.append(doctor_id) else: # If concurrent_doctors is None, return the list as is (deduplicated) doctor_ids = await redis.lrange(redis_key, 0, -1) seen = set() for id in doctor_ids: doc_str = str(id) if doc_str not in seen: seen.add(doc_str) assigned_doctor_ids.append(doc_str) if not assigned_doctor_ids: # Check if prescription exists in `rx_prescription_mapping` prescription_mapping_data = await redis.hget( RX_PRESCRIPTION_MAPPING_KEY, prescription_id ) if prescription_mapping_data is None: # If no doctors are available, add the prescription to the queue logger.info( f"No doctors available. Prescription {prescription_id} added to the queue." ) await add_prescription_to_queue( prescription_id=prescription_id, vendor_id=vendor_id ) response_data = AssignDoctorResponseData( prescription_id=prescription_id, doctor_ids=[], ) return AssignDoctorResponse(data=response_data) if not immediate_unblock: # Mark the doctors as assigned to the prescription in Redis await redis.hset( ASSIGNED_DOCTORS_KEY, prescription_id, ",".join(map(str, assigned_doctor_ids)), ) # Update rx_assignment_mapping with prescription_id and creation time assignment_time = int(time.time()) value = json.dumps( { "assignment_time": assignment_time, "vendor_id": vendor_id, "is_blocked": False, } ) await redis.hset(RX_ASSIGNMENT_MAPPING_KEY, prescription_id, value) logger.info( f"Prescription {prescription_id} mapped with create_time {assignment_time} in rx_assignment_mapping." ) else: # Remove the blocked doctor from the pool if present and re-add to end await redis.lrem(redis_key, 0, assigned_doctor_ids[0]) await safe_rpush_doctor(redis, redis_key, assigned_doctor_ids[0]) logger.info( f"Doctors {assigned_doctor_ids} assigned to prescription {prescription_id} for vendor {vendor_id}." ) response_data = AssignDoctorResponseData( prescription_id=prescription_id, doctor_ids=assigned_doctor_ids ) return AssignDoctorResponse(data=response_data) async def acquire_redis_lock( lock_key: str, lock_timeout: int = 5, wait_timeout: int = 10 ): """ Tries to acquire a Redis lock with a given lock_key. lock_timeout: time (in seconds) before the lock automatically expires. wait_timeout: maximum time (in seconds) to wait for the lock. """ start_time = time.time() lock_value = str(time.time()) while True: was_set = await redis.set(lock_key, lock_value, ex=lock_timeout, nx=True) if was_set: return lock_value if time.time() - start_time > wait_timeout: raise asyncio.TimeoutError(f"Could not acquire lock on {lock_key}") await asyncio.sleep(0.1) async def release_redis_lock(lock_key: str, lock_value: str): """ Releases the Redis lock if the lock_key matches. """ await redis.delete(lock_key) ### Block and Unblock Doctors ### @api_router.post("/block_doctor", operation_id="block_doctor") async def block_doctor(block_request: BlockDoctorRequest) -> BlockDoctorResponse: """ API to block one doctor and unblock the remaining doctors assigned to the same prescription. If the prescription has a single doctor assigned, prevent blocking/unblocking and send a message indicating that the doctor is already assigned. """ block_doctor_id = block_request.block_doctor_id vendor_id = block_request.vendor_id prescription_id = ( block_request.prescription_id ) # Get the prescription_id from the request global vendor_keys redis_key = vendor_keys.get(vendor_id) if not redis_key: logger.error(f"Vendor {vendor_id} not found.") raise HTTPException(status_code=404, detail=f"Vendor {vendor_id} not found.") # Serialise every pick made by this doctor. The per-prescription lock taken # below only guards a single prescription, so without a doctor-scoped lock # one doctor firing block_doctor for several prescriptions at once passes the # "already assigned" check in every call before any of them commits, and ends # up blocked on all of them (see test_block_doctor_race). Holding # lock:doctor: across the check-then-commit makes those picks atomic so # at most one can win. doctor_lock_key = f"lock:doctor:{block_doctor_id}" try: doctor_lock_value = await acquire_redis_lock( doctor_lock_key, lock_timeout=5, wait_timeout=10 ) except asyncio.TimeoutError: raise HTTPException( status_code=503, detail=f"Could not acquire lock for doctor {block_doctor_id}. Please try again.", ) try: # Check if this doctor is assigned to any prescription assigned_prescriptions = await redis.hgetall(ASSIGNED_DOCTORS_KEY) for assigned_prescription_id, assigned_doc_id in assigned_prescriptions.items(): if assigned_doc_id == str(block_doctor_id): assigned_doc_stored_values = await redis.hget( RX_ASSIGNMENT_MAPPING_KEY, assigned_prescription_id ) if assigned_doc_stored_values: assigned_doc_stored_values = json.loads(assigned_doc_stored_values) if ( assigned_doc_stored_values and assigned_doc_stored_values["is_blocked"] ): logger.info( f"Doctor {block_doctor_id} is already assigned to prescription {assigned_prescription_id}." ) raise HTTPException( status_code=400, detail=f"Doctor {block_doctor_id} is already assigned to prescription {assigned_prescription_id}.", ) lock_key = f"lock:prescription:{prescription_id}" # Acquire the Redis lock try: lock_value = await acquire_redis_lock( lock_key, lock_timeout=5, wait_timeout=10 ) except asyncio.TimeoutError: raise HTTPException( status_code=503, detail=f"Could not acquire lock for prescription {prescription_id}. Please try again.", ) try: # Fetch the assigned doctors for the given prescription assigned_doctors = await redis.hget(ASSIGNED_DOCTORS_KEY, prescription_id) prescription_mapping_data = await redis.hget( RX_PRESCRIPTION_MAPPING_KEY, prescription_id ) if not assigned_doctors and prescription_mapping_data: assigned_doctors = block_doctor_id await redis.lrem(PRESCRIPTION_QUEUE, 0, prescription_id) await redis.hdel(RX_PRESCRIPTION_MAPPING_KEY, prescription_id) await remove_prescription_from_vendor_map( vendor_id=vendor_id, prescription_id=prescription_id ) logger.info( "Removed prescription from prescription queue and adding it back to assignment queue." ) if assigned_doctors: # Split the string of doctor IDs into a list doctor_list = assigned_doctors.split(",") assigned_doc_stored_values = await redis.hget( RX_ASSIGNMENT_MAPPING_KEY, prescription_id ) if assigned_doc_stored_values: assigned_doc_stored_values = json.loads(assigned_doc_stored_values) # If the prescription has only one doctor assigned, block/unblock should not happen if ( assigned_doc_stored_values and assigned_doc_stored_values["is_blocked"] ): # Send a message that the prescription already has a doctor assigned logger.error( f"Prescription {prescription_id} already has Doctor ID {doctor_list[0]} assigned. Blocking is not allowed." ) raise HTTPException( status_code=400, detail=f"Prescription {prescription_id} already has Doctor ID {doctor_list[0]} assigned. Blocking is not allowed.", ) # Remove the blocked doctor from the list global doctors_data immediate_unblock = None if doctors_data and len(doctors_data.data) > 0: immediate_unblock = next( ( entry.immediate_unblock for entry in doctors_data.data if entry.vendor == vendor_id ), None, # Default value if the vendor_id is not found ) if not immediate_unblock and str(block_doctor_id) in doctor_list: doctor_list.remove(str(block_doctor_id)) # Return the non-accepting doctors to the end of the pool so they # stay available for other prescriptions. They were removed from # the pool when this prescription was assigned; without returning # them here they are stranded (neither consulting nor in the pool), # which starves later prescriptions. Skip any doctor genuinely # blocked on another prescription. blocked_doctors = await get_blocked_doctors_for_vendor( vendor_id, exclude_prescription_id=prescription_id ) for remaining_doctor_id in doctor_list: if remaining_doctor_id in blocked_doctors: continue await safe_rpush_doctor(redis, redis_key, remaining_doctor_id) # Mark the blocked doctor in Redis as the only doctor assigned await redis.hset( ASSIGNED_DOCTORS_KEY, prescription_id, block_doctor_id ) # Update the assignment time assignment_time = int(time.time()) value = json.dumps( { "assignment_time": assignment_time, "vendor_id": vendor_id, "is_blocked": True, } ) await redis.hset(RX_ASSIGNMENT_MAPPING_KEY, prescription_id, value) # Remove the blocked doctor from the pool if present await redis.lrem(redis_key, 0, block_doctor_id) prescription_mapping_data = await redis.hget( RX_PRESCRIPTION_MAPPING_KEY, prescription_id ) if prescription_mapping_data: await redis.hdel(RX_PRESCRIPTION_MAPPING_KEY, prescription_id) logger.info( f"Doctor {block_doctor_id} blocked and remaining doctors unblocked for prescription {prescription_id}." ) response_data = BlockDoctorResponseData( message=f"Doctor {block_doctor_id} blocked and remaining doctors unblocked for prescription {prescription_id}." ) return BlockDoctorResponse(data=response_data) else: logger.error("No doctor's were assigned to the prescription.") raise HTTPException( status_code=404, detail=f"No doctors blocked for Prescription {prescription_id}.", ) finally: await release_redis_lock(lock_key, lock_value) finally: await release_redis_lock(doctor_lock_key, doctor_lock_value) async def process_prescription(prescription_id, vendor_id): """Process a prescription based on queue and expiration status.""" # Check if prescription exists in `rx_prescription_mapping` prescription_mapping_data = await redis.hget( RX_PRESCRIPTION_MAPPING_KEY, prescription_id ) if prescription_mapping_data is not None: data = json.loads(prescription_mapping_data) prescription_create_time = data["prescription_create_time"] create_time = int(prescription_create_time) current_time = int(time.time()) # Check if prescription is expired rx_max_hold_time = VENDOR_DOCTOR_MAPPING.get(vendor_id, {}).get( "rx_max_hold_time", RX_MAX_HOLD_TIME ) if current_time - create_time > rx_max_hold_time: async with aiohttp.ClientSession() as session: async with session.post( CANCEL_PX_URL, json={"prescription_id": prescription_id}, ) as response: if response.status == 200: await redis.lrem(PRESCRIPTION_QUEUE, 0, prescription_id) await redis.hdel(RX_PRESCRIPTION_MAPPING_KEY, prescription_id) await remove_prescription_from_vendor_map( vendor_id=vendor_id, prescription_id=prescription_id ) logger.info( f"Prescription {prescription_id} expired and canceled." ) else: logger.error(f"Failed to cancel prescription {prescription_id}") elif create_time > current_time: logger.info( f"Prescription {prescription_id} is still in hold delay period." ) else: # Get doctors list by calling `assign_doctor` function assigned_doctor_response = await assign_doctor( AssignDoctorRequest( prescription_id=prescription_id, vendor_id=vendor_id ) ) doctors_list = assigned_doctor_response.data.model_dump()["doctor_ids"] if doctors_list: await redis.lrem(PRESCRIPTION_QUEUE, 0, prescription_id) await remove_prescription_from_vendor_map( vendor_id=vendor_id, prescription_id=prescription_id ) async with aiohttp.ClientSession() as session: async with session.post( ASSIGN_PX_URL, json={ "prescription_id": prescription_id, "doctors": doctors_list, }, ) as response: if response.status == 200: logger.info( f"Prescription {prescription_id} assigned to doctors." ) else: logger.error( f"Failed to assign doctors for prescription {prescription_id}" ) else: logger.warning( f"No available doctors for prescription {prescription_id}." ) else: logger.info( f"Prescription {prescription_id} is not present in both required queues." ) # API endpoint to add vendor @api_router.post("/add_vendor", operation_id="add_vendor") async def add_vendor(doctor_request: DoctorResponseData) -> AddVendorResponse: """ API to add a doctors to the Redis pool for a specific vendor. """ global doctors_data global vendor_keys global VENDOR_DOCTOR_MAPPING vendor_id = doctor_request.vendor # Check if the vendor already exists existing_vendor = next( (entry for entry in doctors_data.data if entry.vendor == vendor_id), None ) if existing_vendor: logger.info(f"Vendor {vendor_id} already exists.") response_data = AddVendorResponseData( message=f"Vendor {vendor_id} already exists." ) return AddVendorResponse(data=response_data) # Add the new vendor to doctors_data new_vendor_entry = DoctorResponseData( vendor=vendor_id, doctors=doctor_request.doctors, concurrent_doctors=doctor_request.concurrent_doctors, immediate_unblock=doctor_request.immediate_unblock, rx_max_hold_time=doctor_request.rx_max_hold_time, rx_max_block_time=doctor_request.rx_max_block_time, rx_notification_interval=doctor_request.rx_notification_interval, delay_after_hold=doctor_request.delay_after_hold, ) doctors_data.data.append(new_vendor_entry) # Add doctors to the Redis pool redis_key = f"{DOCTORS_POOL}_{vendor_id}" vendor_keys[vendor_id] = redis_key # Keep vendor-level runtime settings in sync for queue/block flows. VENDOR_DOCTOR_MAPPING[vendor_id] = { "concurrent_doctors": doctor_request.concurrent_doctors, "immediate_unblock": doctor_request.immediate_unblock, "rx_max_hold_time": doctor_request.rx_max_hold_time, "rx_max_block_time": doctor_request.rx_max_block_time, "rx_notification_interval": doctor_request.rx_notification_interval, "delay_after_hold": doctor_request.delay_after_hold, } if doctor_request.doctors: for doctor_id in doctor_request.doctors: await redis.rpush(redis_key, doctor_id) logger.info(f"Doctor's data for vendor {vendor_id} added to the pool.") response_data = AddVendorResponseData( message=f"Doctor's data for vendor {vendor_id} added to the pool." ) return AddVendorResponse(data=response_data) # API endpoint to update vendor settings @api_router.patch("/update_vendor", operation_id="update_vendor") async def update_vendor(vendor_request: UpdateVendorRequest) -> UpdateVendorResponse: """ API to update vendor settings and refresh the Redis cache. Updates any of the following fields if provided: - concurrent_doctors - immediate_unblock - rx_max_hold_time - rx_max_block_time - rx_notification_interval - delay_after_hold """ global doctors_data global VENDOR_DOCTOR_MAPPING vendor_id = vendor_request.vendor_id # Check if the vendor exists existing_vendor = next( (entry for entry in doctors_data.data if entry.vendor == vendor_id), None ) if not existing_vendor: logger.error(f"Vendor {vendor_id} not found.") raise HTTPException(status_code=404, detail=f"Vendor {vendor_id} not found.") # Track which fields were updated updated_fields = [] # Update the fields in doctors_data if provided if vendor_request.concurrent_doctors is not None: existing_vendor.concurrent_doctors = vendor_request.concurrent_doctors updated_fields.append("concurrent_doctors") if vendor_request.immediate_unblock is not None: existing_vendor.immediate_unblock = vendor_request.immediate_unblock updated_fields.append("immediate_unblock") if vendor_request.rx_max_hold_time is not None: existing_vendor.rx_max_hold_time = vendor_request.rx_max_hold_time updated_fields.append("rx_max_hold_time") if vendor_request.rx_max_block_time is not None: existing_vendor.rx_max_block_time = vendor_request.rx_max_block_time updated_fields.append("rx_max_block_time") if vendor_request.rx_notification_interval is not None: existing_vendor.rx_notification_interval = ( vendor_request.rx_notification_interval ) updated_fields.append("rx_notification_interval") if vendor_request.delay_after_hold is not None: existing_vendor.delay_after_hold = vendor_request.delay_after_hold updated_fields.append("delay_after_hold") # Update VENDOR_DOCTOR_MAPPING if vendor_id in VENDOR_DOCTOR_MAPPING: if vendor_request.concurrent_doctors is not None: VENDOR_DOCTOR_MAPPING[vendor_id][ "concurrent_doctors" ] = vendor_request.concurrent_doctors if vendor_request.immediate_unblock is not None: VENDOR_DOCTOR_MAPPING[vendor_id][ "immediate_unblock" ] = vendor_request.immediate_unblock if vendor_request.rx_max_hold_time is not None: VENDOR_DOCTOR_MAPPING[vendor_id][ "rx_max_hold_time" ] = vendor_request.rx_max_hold_time if vendor_request.rx_max_block_time is not None: VENDOR_DOCTOR_MAPPING[vendor_id][ "rx_max_block_time" ] = vendor_request.rx_max_block_time if vendor_request.rx_notification_interval is not None: VENDOR_DOCTOR_MAPPING[vendor_id][ "rx_notification_interval" ] = vendor_request.rx_notification_interval if vendor_request.delay_after_hold is not None: VENDOR_DOCTOR_MAPPING[vendor_id][ "delay_after_hold" ] = vendor_request.delay_after_hold # Refresh the Redis cache for this vendor await initialize_doctor_pool_for_vendor(redis, vendor_id) logger.info( f"Vendor {vendor_id} settings updated. Fields: {', '.join(updated_fields)}" ) logger.info(f"Redis cache refreshed for vendor {vendor_id}") response_data = UpdateVendorResponseData( message=f"Vendor {vendor_id} settings updated successfully. Updated fields: {', '.join(updated_fields)}" ) return UpdateVendorResponse(data=response_data) async def acquire_vendor_map_lock( vendor_id: str, lock_timeout: int = 5, wait_timeout: int = 10 ): lock_key = f"lock:vendor_map:{vendor_id}" start_time = time.time() lock_value = str(time.time()) while True: was_set = await redis.set(lock_key, lock_value, ex=lock_timeout, nx=True) if was_set: return lock_key, lock_value if time.time() - start_time > wait_timeout: raise asyncio.TimeoutError(f"Could not acquire lock on {lock_key}") await asyncio.sleep(0.1) async def release_vendor_map_lock(lock_key: str, lock_value: str): await redis.delete(lock_key) async def get_blocked_doctors_for_vendor( vendor_id: str, exclude_prescription_id: Optional[str] = None ) -> set: blocked_doctors = set() assigned_prescriptions = await redis.hgetall(RX_ASSIGNMENT_MAPPING_KEY) if not assigned_prescriptions: return blocked_doctors for prescription_id, stored_value in assigned_prescriptions.items(): try: data = json.loads(stored_value) except (TypeError, json.JSONDecodeError): continue if not data.get("is_blocked"): continue if data.get("vendor_id") != vendor_id: continue if exclude_prescription_id and prescription_id == exclude_prescription_id: continue assigned_doctors = await redis.hget(ASSIGNED_DOCTORS_KEY, prescription_id) if not assigned_doctors: continue if isinstance(assigned_doctors, (bytes, bytearray)): assigned_doctors = assigned_doctors.decode() for doctor_id in assigned_doctors.split(","): blocked_doctors.add(doctor_id) return blocked_doctors async def any_doctor_inactive(vendor_id: str, doctor_ids: list) -> bool: """ Whether any of these doctors has gone inactive and so is not returning to the pool. Covers the doctor whose own popup countdown ran out first: their browser reports it, onerx-service moves them away and calls /remove_doctor, and by the time this expiry runs they are already Away — so nothing is reported as newly moved, yet the order still needs a different doctor now rather than at the next tick. Gated on the durable tracked flag for the same reason safe_rpush_doctor is: a vendor whose activity we do not track has no meaningful active set to read, and every doctor would look inactive. """ if not doctor_ids: return False if not await redis.sismember(ACTIVE_DOCTORS_TRACKED_KEY, vendor_id): return False active_key = f"{ACTIVE_DOCTORS_POOL}_{vendor_id}" for doctor_id in doctor_ids: if not await redis.sismember(active_key, doctor_id): return True return False async def mark_doctors_away_for_missed_notification( vendor_id: str, prescription_id: str, doctor_ids: list ) -> set: """ Move doctors who were offered an order and never accepted it to Away. Only applies to vendors with a configured concurrency: those doctors were each sent a notification and ignored it. With no concurrency set, assign_doctor offers the order to the entire pool rather than to chosen few, so a timeout says nothing about any one doctor and marking them all away would empty the pool on a single unaccepted order. Returns the ids that were actually moved, which the caller must keep out of the pool. onerx-service decides that — a doctor who logged off, or who is already away, is not counted — so a doctor missing from the returned set is deliberately still available. On any failure the set is empty and every doctor goes back to the pool, keeping the old behaviour: a service that is unreachable must not silently drain the pool. """ # Read the concurrency the same way assign_doctor does, so this triggers exactly when the # doctors were individually picked and notified rather than offered the whole pool. concurrent_doctors = None if doctors_data and len(doctors_data.data) > 0: concurrent_doctors = next( ( entry.concurrent_doctors for entry in doctors_data.data if entry.vendor == vendor_id ), None, ) if not concurrent_doctors or not doctor_ids: return set() try: async with aiohttp.ClientSession() as session: async with session.post( DOCTOR_NOT_RESPONDING_URL, json={ "prescription_id": prescription_id, "vendor": vendor_id, "doctors": doctor_ids, }, ) as response: if response.status != 200: logger.error( f"Failed to mark doctors {doctor_ids} away for prescription " f"{prescription_id} (status={response.status}); returning them to the pool." ) return set() body = await response.json() except Exception as e: logger.error( f"Error while marking doctors {doctor_ids} away for prescription " f"{prescription_id}: {e}. Returning them to the pool." ) return set() marked_away = {str(doctor_id) for doctor_id in (body.get("data") or [])} if not marked_away: logger.info( f"No doctors were moved away for prescription {prescription_id}; " f"all of {doctor_ids} return to the pool." ) return marked_away # Drop them from the active set, which is what safe_rpush_doctor consults. This is the piece # that keeps them out until they set themselves Active again — that call re-adds them via # /add_doctor. They are already out of the pool list (assign_doctor removed them when the # order was offered), but lrem defensively in case another path put them back. active_key = f"{ACTIVE_DOCTORS_POOL}_{vendor_id}" redis_key = f"{DOCTORS_POOL}_{vendor_id}" for doctor_id in marked_away: await redis.srem(active_key, doctor_id) await redis.lrem(redis_key, 0, doctor_id) logger.info( f"Doctors {sorted(marked_away)} moved to away for not accepting prescription " f"{prescription_id} (vendor {vendor_id}) and removed from the active set." ) return marked_away async def add_prescription_to_vendor_map(vendor_id: str, prescription_id: str): """Adds a prescription to the ordered list under the given vendor_id in Redis Hash.""" lock_key, lock_value = await acquire_vendor_map_lock(vendor_id) try: existing_data = await redis.hget(PRESCRIPTION_VENDOR_MAP_KEY, vendor_id) prescription_list = json.loads(existing_data) if existing_data else [] prescription_list.append(prescription_id) await redis.hset( PRESCRIPTION_VENDOR_MAP_KEY, vendor_id, json.dumps(prescription_list) ) logger.info(f"Added {prescription_id} to vendor {vendor_id}'s map") finally: await release_vendor_map_lock(lock_key, lock_value) async def remove_prescription_from_vendor_map(vendor_id: str, prescription_id: str): """Removes a specific prescription from the vendor's map while maintaining order.""" lock_key, lock_value = await acquire_vendor_map_lock(vendor_id) try: existing_data = await redis.hget(PRESCRIPTION_VENDOR_MAP_KEY, vendor_id) if not existing_data: logger.info(f"No prescriptions found for vendor {vendor_id}") return prescription_list = json.loads(existing_data) if prescription_id in prescription_list: prescription_list.remove(prescription_id) await redis.hset( PRESCRIPTION_VENDOR_MAP_KEY, vendor_id, json.dumps(prescription_list) ) logger.info(f"Removed {prescription_id} from vendor {vendor_id}'s map") else: logger.info( f"Prescription {prescription_id} not found for vendor {vendor_id}" ) finally: await release_vendor_map_lock(lock_key, lock_value) async def pop_prescription_from_vendor_map(vendor_id: str) -> Optional[str]: """Returns the first ready-to-process prescription for a vendor (skipping those on hold).""" lock_key, lock_value = await acquire_vendor_map_lock(vendor_id) try: # Get all prescriptions for the vendor existing_data = await redis.hget(PRESCRIPTION_VENDOR_MAP_KEY, vendor_id) if not existing_data: logger.info(f"No prescriptions left for vendor {vendor_id}") return None prescription_list = json.loads(existing_data) if not prescription_list: logger.info(f"Empty prescription list for vendor {vendor_id}") return None current_time = int(time.time()) # Iterate FIFO and find the first ready prescription for idx, prescription_id in enumerate(prescription_list): prescription_mapping_data = await redis.hget( RX_PRESCRIPTION_MAPPING_KEY, prescription_id ) if not prescription_mapping_data: logger.warning( f"No mapping data found for {prescription_id}, removing it." ) prescription_list.pop(idx) continue data = json.loads(prescription_mapping_data) create_time = int(data.get("prescription_create_time", 0)) # Check if hold period has passed if create_time <= current_time: # Prescription is ready for processing prescription_list.pop(idx) if prescription_list: await redis.hset( PRESCRIPTION_VENDOR_MAP_KEY, vendor_id, json.dumps(prescription_list), ) else: await redis.hdel(PRESCRIPTION_VENDOR_MAP_KEY, vendor_id) logger.info( f"Prescription {prescription_id} ready for processing for vendor {vendor_id}" ) return prescription_id else: logger.info( f"Prescription {prescription_id} for vendor {vendor_id} is still on hold " f"(create_time={create_time}, current_time={current_time})" ) # If loop finishes, no ready prescriptions logger.info(f"No ready prescriptions for vendor {vendor_id}") return None finally: await release_vendor_map_lock(lock_key, lock_value) @api_router.post("/add_doctor", operation_id="add_doctor") async def add_doctor( doctor_request: DoctorRequest, background_tasks: BackgroundTasks = None ) -> AddDoctorResponse: """ API to add a doctor to the Redis pool for a specific vendor. """ vendor_id = doctor_request.vendor_id redis_key = f"{DOCTORS_POOL}_{vendor_id}" doctor_id = doctor_request.id logger.info(f"Adding doctor {doctor_id} to the pool for vendor {vendor_id}") # Mark the doctor active (logged in) regardless of which branch below runs, # so the active-doctors set stays the source of truth for re-add gating. # Also ensure the vendor is tracked (covers vendors first seen at runtime that # did not go through initialize_doctor_pool_for_vendor). active_key = f"{ACTIVE_DOCTORS_POOL}_{vendor_id}" await redis.sadd(active_key, doctor_id) await redis.sadd(ACTIVE_DOCTORS_TRACKED_KEY, vendor_id) logger.info( f"ACTIVE-SET add: doctor {doctor_id} marked active for vendor {vendor_id} ({active_key})." ) # Check if the doctor is already in the pool doctor_ids = await redis.lrange(redis_key, 0, -1) if str(doctor_id) in doctor_ids: logger.info(f"Doctor {doctor_id} already exists in vendor {vendor_id}'s pool.") response_data = AddDoctorResponseData( message=f"Doctor {doctor_id} already exists in vendor {vendor_id}'s pool." ) return AddDoctorResponse(data=response_data) # Check if this doctor is assigned to any prescription assigned_prescriptions = await redis.hgetall(ASSIGNED_DOCTORS_KEY) if any( str(doctor_id) in assigned_doc_ids.split(",") for assigned_doc_ids in assigned_prescriptions.values() ): response_data = AddDoctorResponseData( message=f"Doctor {doctor_id} is currently assigned to a prescription and cannot be added to the pool." ) logger.info( f"Doctor {doctor_id} is currently assigned to a prescription and cannot be added to the pool." ) return AddDoctorResponse(data=response_data) # Add doctor to the Redis list await safe_rpush_doctor(redis, redis_key, doctor_id) logger.info(f"Doctor {doctor_id} added to the pool for vendor {vendor_id}") # Check if there are any pending prescriptions in the queue prescription_in_queue = await pop_prescription_from_vendor_map(vendor_id) if prescription_in_queue: await redis.lrem(PRESCRIPTION_QUEUE, 0, prescription_in_queue) await process_prescription(prescription_in_queue, vendor_id) logger.info(f"Doctor {doctor_id} added to vendor {vendor_id}'s pool.") response_data = AddDoctorResponseData( message=f"Doctor {doctor_id} added to vendor {vendor_id}'s pool." ) return AddDoctorResponse(data=response_data) ### Reassign New Doctors to a Prescription ### @api_router.post( "/reassign_doctor", operation_id="reassign_doctor", ) async def reassign_doctor( reassign_request: AssignDoctorRequest, ) -> ReassignDoctorResponse: """ API to reassign up to n new doctors from the vendor's pool to a prescription. The previously assigned doctors are returned to the pool. """ prescription_id = reassign_request.prescription_id vendor_id = reassign_request.vendor_id global vendor_keys redis_key = vendor_keys.get(vendor_id) if not redis_key: logger.error(f"Vendor {vendor_id} not found.") raise HTTPException(status_code=404, detail=f"Vendor {vendor_id} not found.") # Fetch the current assigned doctors for this prescription assigned_doctors = await redis.hget(ASSIGNED_DOCTORS_KEY, prescription_id) if not assigned_doctors: logger.error( f"No doctors are currently assigned to prescription {prescription_id}." ) raise HTTPException( status_code=404, detail=f"No doctors are currently assigned to prescription {prescription_id}.", ) # Get the list of currently assigned doctors and return them to the pool previous_doctor_list = assigned_doctors.split(",") if len(previous_doctor_list) == 1: blocked_doctors = await get_blocked_doctors_for_vendor( vendor_id, exclude_prescription_id=prescription_id ) if previous_doctor_list[0] in blocked_doctors: logger.info( f"Doctor {previous_doctor_list[0]} remains blocked and was not returned to pool for vendor {vendor_id}." ) else: await safe_rpush_doctor(redis, redis_key, previous_doctor_list[0]) logger.info( f"Doctor {previous_doctor_list[0]} returned to pool for vendor {vendor_id}." ) new_assigned_doctors = [] global doctors_data concurrent_doctors = None if doctors_data and len(doctors_data.data) > 0: concurrent_doctors = next( ( entry.concurrent_doctors for entry in doctors_data.data if entry.vendor == vendor_id ), None, # Default value if the vendor_id is not found ) if concurrent_doctors: for _ in range(concurrent_doctors): doctor_id = await redis.lpop(redis_key) if not doctor_id: break new_assigned_doctors.append(str(doctor_id)) else: # If concurrent_doctors is None, return the list as is doctor_ids = await redis.lrange(redis_key, 0, -1) new_assigned_doctors = [str(id) for id in doctor_ids] if not new_assigned_doctors: logger.error("No doctors available in the pool for reassignment.") raise HTTPException( status_code=404, detail="No doctors available in the pool for reassignment." ) # Update the assigned doctors in Redis await redis.hset( ASSIGNED_DOCTORS_KEY, prescription_id, ",".join(new_assigned_doctors) ) # Log the new assignments logger.info( f"New doctors {new_assigned_doctors} assigned to prescription {prescription_id}." ) response_data = ReassignDoctorResponseData( prescription_id=prescription_id, doctor_ids=new_assigned_doctors ) return ReassignDoctorResponse(data=response_data) ### Release Doctor API ### @api_router.post("/release_doctor", operation_id="release_doctor") async def release_doctor( release_request: ReleaseDoctorRequest, background_tasks: BackgroundTasks = None ) -> ReleaseDoctorResponse: """ API to release a doctor from a prescription and push the doctor back to the pool. If the doctor was previously blocked, they will be unblocked and returned to the pool. """ doctor_id = str( release_request.doctor_id ) # Ensure doctor_id is a string for Redis operations prescription_id = release_request.prescription_id vendor_id = release_request.vendor_id global vendor_keys redis_key = vendor_keys.get(vendor_id) if not redis_key: logger.error(f"Vendor {vendor_id} not found.") raise HTTPException(status_code=404, detail=f"Vendor {vendor_id} not found.") # Fetch the assigned doctor for the given prescription assigned_doctor = await redis.hget(ASSIGNED_DOCTORS_KEY, prescription_id) if not assigned_doctor: logger.error(f"No doctor assigned to prescription {prescription_id}.") raise HTTPException( status_code=404, detail=f"No doctor assigned to prescription {prescription_id}.", ) # Check if the doctor matches the one assigned if assigned_doctor != doctor_id: logger.error( f"Doctor {doctor_id} is not assigned to prescription {prescription_id}." ) raise HTTPException( status_code=404, detail=f"Doctor {doctor_id} is not assigned to prescription {prescription_id}.", ) # Remove the doctor from the assigned doctors in Redis await redis.hdel(ASSIGNED_DOCTORS_KEY, prescription_id) await redis.hdel(RX_ASSIGNMENT_MAPPING_KEY, prescription_id) # Push the doctor back to the pool blocked_doctors = await get_blocked_doctors_for_vendor( vendor_id, exclude_prescription_id=prescription_id ) if doctor_id in blocked_doctors: logger.info( f"Doctor {doctor_id} remains blocked and was not returned to pool for vendor {vendor_id}." ) else: added = await safe_rpush_doctor(redis, redis_key, doctor_id) if added: logger.info( f"Doctor {doctor_id} released from prescription {prescription_id} and added back to pool for vendor {vendor_id}." ) # Check if there are any pending prescriptions in the queue prescription_in_queue = await pop_prescription_from_vendor_map(vendor_id) if prescription_in_queue: await redis.lrem(PRESCRIPTION_QUEUE, 0, prescription_in_queue) await process_prescription(prescription_in_queue, vendor_id) response_data = ReleaseDoctorResponseData( message=f"Doctor {doctor_id} released from prescription {prescription_id} and added back to pool." ) return ReleaseDoctorResponse(data=response_data) async def assign_doctor_from_queue(doctor_id: str, prescription_id: str, redis): """ Assign a doctor to a prescription from the queue and mark the doctor as assigned. """ # Assign the doctor from the queue await redis.hset(ASSIGNED_DOCTORS_KEY, prescription_id, doctor_id) logger.info( f"Doctor {doctor_id} assigned to prescription {prescription_id} from the queue." ) return {"prescription_id": prescription_id, "doctor_id": str(doctor_id)} @api_router.post("/remove_doctor", operation_id="remove_doctor") async def remove_doctor(doctor_request: DoctorRequest) -> DeleteDoctorResponse: """ API to remove a doctor from the Redis pool for a specific vendor. """ vendor_id = doctor_request.vendor_id redis_key = f"{DOCTORS_POOL}_{vendor_id}" active_key = f"{ACTIVE_DOCTORS_POOL}_{vendor_id}" doctor_id = doctor_request.id logger.info(f"Remove doctor {doctor_id} from the pool for vendor {vendor_id}") # Mark the doctor inactive first. This is the source of truth used by # safe_rpush_doctor, so even when the doctor is currently out of the pool # (because a prescription is being offered to them right now), the # prescription lifecycle (expiry / release / reassign / block) will NOT # resurrect them into the pool afterwards. await redis.srem(active_key, doctor_id) logger.info( f"ACTIVE-SET remove: doctor {doctor_id} marked inactive for vendor {vendor_id} ({active_key})." ) # Remove the doctor from the pool if they are currently in it. removed = await redis.lrem(redis_key, 0, doctor_id) if removed: logger.info(f"Doctor {doctor_id} removed from the pool for vendor {vendor_id}") message = f"Doctor {doctor_id} removed from vendor {vendor_id}'s pool." else: logger.info( f"Doctor {doctor_id} was not in vendor {vendor_id}'s pool (e.g. mid-assignment); " f"marked inactive so they will not be re-added." ) message = f"Doctor {doctor_id} marked inactive for vendor {vendor_id}." response_data = DeleteDoctorResponseData(message=message) return DeleteDoctorResponse(data=response_data) @api_router.get( "/list_all_cached_doctors", operation_id="all_doctors", ) async def list_doctors(vendor_id: str) -> CachedDoctorsResponse: """ API to list all doctors from the Redis pool. """ redis_key = f"{DOCTORS_POOL}_{vendor_id}" # Get all doctor IDs from the Redis pool doctor_ids = await redis.lrange(redis_key, 0, -1) if not doctor_ids: logger.info("No doctors found in the pool.") doctor_ids_list = [str(doc_id) for doc_id in doctor_ids] logger.debug(f"Doctors in the pool: {doctor_ids_list}") response_data = CachedDoctorsResponseData(doctor_ids=doctor_ids_list) return CachedDoctorsResponse(data=response_data) @api_router.get( "/get_prescriptions_in_queue", operation_id="prescriptions", ) async def get_prescriptions_in_queue( vendor_id: str, ) -> PrescriptionsInQueueResponse: """ API to get all data from the Redis prescription queue. """ # Retrieve all items in the prescription queue prescriptions = await redis.lrange(PRESCRIPTION_QUEUE, 0, -1) if not prescriptions: raise HTTPException( status_code=404, detail="No prescriptions found in the queue." ) # Return the list of prescriptions response_data = PrescriptionsInQueueResponseData(prescription_ids=prescriptions) return PrescriptionsInQueueResponse(data=response_data) @api_router.get( "/assigned_doctors", operation_id="assigned_doctors", ) async def get_all_assigned_doctors() -> AssignedDoctorPrescriptionResponse: """ API to get all data from the Redis 'assigned_doctors' set. """ # Retrieve all prescription entries from the hash assigned_doctors_raw = await redis.hgetall(ASSIGNED_DOCTORS_KEY) if not assigned_doctors_raw: raise HTTPException(status_code=404, detail="No assigned doctors found.") # Parse the comma-separated doctor IDs for each prescription assigned_doctors = { prescription_id: value.split(",") for prescription_id, value in assigned_doctors_raw.items() } # Return the parsed assigned doctors response_data = AssignedDoctorPrescriptionResponseData( prescription_doctor=assigned_doctors ) return AssignedDoctorPrescriptionResponse(data=response_data) @api_router.post( "/hold", operation_id="hold", ) async def hold_prescription( hold_prescription_request: HoldPrescriptionRequest, ) -> HoldPrescriptionResponse: """ Processes a hold request for a prescription, releasing doctors back to the pool and updating necessary queues. """ try: vendor_id = hold_prescription_request.vendor_id prescription_id = hold_prescription_request.prescription_id doctors_pool_key = f"{DOCTORS_POOL}_{vendor_id}" # Fetch the doctor list for the prescription ID in assigned_doctors doctor_list = await redis.hget(ASSIGNED_DOCTORS_KEY, prescription_id) if not doctor_list: logger.error("Prescription not found in assigned doctors queue.") raise HTTPException( status_code=404, detail="Prescription not found in assigned doctors queue.", ) # Release each doctor back to the doctor's pool doctor_ids = doctor_list.split(",") blocked_doctors = await get_blocked_doctors_for_vendor( vendor_id, exclude_prescription_id=prescription_id ) for doctor_id in doctor_ids: if doctor_id in blocked_doctors: logger.info( f"Doctor {doctor_id} remains blocked and was not returned to pool for vendor {vendor_id}." ) continue added = await safe_rpush_doctor(redis, doctors_pool_key, doctor_id) if added: logger.info( f"Doctor {doctor_id} returned to pool for vendor {vendor_id}." ) # Remove the prescription from assigned_doctors await redis.hdel(ASSIGNED_DOCTORS_KEY, prescription_id) await redis.hdel(RX_ASSIGNMENT_MAPPING_KEY, prescription_id) logger.info( f"Prescription {prescription_id} removed from assigned_doctors for vendor {vendor_id}." ) delay_after_hold = VENDOR_DOCTOR_MAPPING.get(vendor_id, {}).get( "delay_after_hold", 0 ) # Add the prescription to prescription_queue await add_prescription_to_queue( prescription_id=prescription_id, vendor_id=vendor_id, hold_time=delay_after_hold, ) logger.info(f"Prescription {prescription_id} added to prescription_queue.") logger.info(f"Hold processed for prescription {prescription_id}.") response_data = HoldPrescriptionResponseData( message=f"Hold processed for prescription {prescription_id}" ) return HoldPrescriptionResponse(data=response_data) except Exception as e: logger.error( f"Error in processing hold for prescription {prescription_id}: {e}" ) raise HTTPException(status_code=500, detail="Error processing hold request.") # The task function that will run every 10 seconds async def check_prescription_queue(): """Check the prescription queue and process items based on their expiry.""" try: logger.debug("Running prescription queue check...") prescription_queue = await redis.lrange(PRESCRIPTION_QUEUE, 0, -1) if not prescription_queue: logger.debug("No prescriptions in queue to check.") return prescriptions = await redis_client.hgetall(RX_PRESCRIPTION_MAPPING_KEY) for prescription_id, stored_value in prescriptions.items(): if prescription_id in prescription_queue: data = json.loads(stored_value) vendor_id = data["vendor_id"] logger.info( f"Processing prescription {prescription_id} for vendor {vendor_id}..." ) await process_prescription( prescription_id=prescription_id, vendor_id=vendor_id ) except Exception as e: logger.error(f"Error in check_prescription_queue: {e}") def start_scheduler(): logger.info("Starting the scheduler...") scheduler.add_job(schedule_check_expired_prescriptions, "interval", seconds=10) scheduler.add_job( schedule_check_prescriptions_queue, "interval", seconds=10, start_date=datetime.now() + timedelta(seconds=1), ) scheduler.start() logger.info("Scheduler started.") def schedule_check_expired_prescriptions(): asyncio.run_coroutine_threadsafe(check_expired_prescriptions_async(), main_loop) def schedule_check_prescriptions_queue(): asyncio.run_coroutine_threadsafe(check_prescription_queue(), main_loop) async def handle_expired_prescription(prescription_id, stored_value, current_time): start_ts = time.time() logger.debug(f"Processing expiry for prescription {prescription_id}.") data = json.loads(stored_value) vendor_id = data["vendor_id"] rx_notification_interval = VENDOR_DOCTOR_MAPPING.get(vendor_id, {}).get( "rx_notification_interval", RX_NOTIFICATION_INTERVAL ) cutoff_time = current_time - rx_notification_interval assigned_time = int(data["assignment_time"]) blocked = data["is_blocked"] redis_key = vendor_keys.get(vendor_id) assigned_doctors = await redis.hget(ASSIGNED_DOCTORS_KEY, prescription_id) doctor_list = assigned_doctors.split(",") if assigned_doctors else [] lock_key = f"lock:{RX_ASSIGNMENT_MAPPING_KEY}:{prescription_id}" try: logger.debug(f"Acquiring lock for prescription {prescription_id} expiry check.") lock_value = await acquire_redis_lock(lock_key, lock_timeout=5, wait_timeout=10) logger.debug(f"Lock acquired for prescription {prescription_id}.") except asyncio.TimeoutError: logger.warning( f"Could not acquire lock for prescription {prescription_id}, skipping expiry check." ) return try: assigned_doc_stored_values = await redis.hget( RX_ASSIGNMENT_MAPPING_KEY, prescription_id ) if assigned_doc_stored_values: assigned_doc_stored_values = json.loads(assigned_doc_stored_values) blocked = assigned_doc_stored_values.get("is_blocked", blocked) finally: await release_redis_lock(lock_key, lock_value) logger.debug(f"Lock released for prescription {prescription_id}.") if blocked: rx_max_block_time = VENDOR_DOCTOR_MAPPING.get(vendor_id, {}).get( "rx_max_block_time", RX_MAX_BLOCK_TIME ) cutoff_time = current_time - rx_max_block_time if assigned_time < cutoff_time: if blocked: logger.info( f"Blocked prescription {prescription_id} expired, sending cancel request to vendor {vendor_id}." ) async with aiohttp.ClientSession() as session: async with session.post( CANCEL_PX_URL, json={"prescription_id": prescription_id} ) as response: if response.status == 200: blocked_doctors = await get_blocked_doctors_for_vendor( vendor_id, exclude_prescription_id=prescription_id ) for doctor_id in doctor_list: if doctor_id in blocked_doctors: logger.info( f"Doctor {doctor_id} remains blocked and was not returned to pool for vendor {vendor_id}." ) continue await safe_rpush_doctor(redis, redis_key, doctor_id) await redis_client.hdel(ASSIGNED_DOCTORS_KEY, prescription_id) await redis_client.hdel( RX_ASSIGNMENT_MAPPING_KEY, prescription_id ) else: logger.error( f"Failed to cancel blocked prescription {prescription_id} for vendor {vendor_id}." ) else: logger.info( f"Prescription {prescription_id} expired, requeuing for vendor {vendor_id}." ) # Nobody accepted inside the notification window. For a vendor with a concurrency # these doctors each ignored a popup, so they go Away instead of back to the pool. marked_away = await mark_doctors_away_for_missed_notification( vendor_id, prescription_id, doctor_list ) blocked_doctors = await get_blocked_doctors_for_vendor(vendor_id) for doctor_id in doctor_list: if doctor_id in marked_away: continue if doctor_id in blocked_doctors: logger.info( f"Doctor {doctor_id} remains blocked and was not returned to pool for vendor {vendor_id}." ) continue await safe_rpush_doctor(redis, redis_key, doctor_id) # Pass expired doctors as tried_doctors for round-robin await add_prescription_to_queue( prescription_id, vendor_id, tried_doctors=doctor_list ) await redis_client.hdel(ASSIGNED_DOCTORS_KEY, prescription_id) await redis_client.hdel(RX_ASSIGNMENT_MAPPING_KEY, prescription_id) # Only when doctors actually left the pool — either moved away just now, or by their # own browser reporting the timeout first. Otherwise they went straight back in and the # ordinary queue tick already reassigns with round-robin. When they did leave, waiting # up to a full tick would add to the window the patient has already spent. # # Safe here: both assignment keys are cleared above, so assign_doctor does not still # see the order as assigned. Errors are contained — the order is already queued, so the # worst case is it waits for the next tick after all. if marked_away or await any_doctor_inactive(vendor_id, doctor_list): try: await process_prescription(prescription_id, vendor_id) except Exception as e: logger.error( f"Error while immediately reassigning prescription {prescription_id} " f"for vendor {vendor_id}: {e}. It stays queued for the next tick." ) else: logger.debug( f"Prescription {prescription_id} not expired for vendor {vendor_id}." ) end_ts = time.time() logger.debug( f"Finished expiry check for prescription {prescription_id} in {end_ts - start_ts:.3f}s." ) # The job function to check for expired prescriptions async def check_expired_prescriptions_async(): logger.info("Checking expired prescriptions...") current_time = int(time.time()) assigned_prescriptions = await redis_client.hgetall(RX_ASSIGNMENT_MAPPING_KEY) if not assigned_prescriptions: logger.debug("No assigned prescriptions to check for expiry.") return start_ts = time.time() logger.info(f"Launching {len(assigned_prescriptions)} tasks concurrently") # Limit concurrency to 5 tasks at a time semaphore = asyncio.Semaphore(5) async def sem_task(prescription_id, stored_value): async with semaphore: await handle_expired_prescription( prescription_id, stored_value, current_time ) tasks = [ sem_task(prescription_id, stored_value) for prescription_id, stored_value in assigned_prescriptions.items() ] await asyncio.gather(*tasks) logger.info( f"All prescription expiry checks completed in {time.time() - start_ts:.3f}s" ) # Define the shutdown endpoint @api_router.get("/shutdown") async def shutdown(): # Send SIGTERM to the current process os.kill(os.getpid(), signal.SIGTERM) return JSONResponse(content={"message": "Server shutting down..."}) app.include_router(api_router) from app.debug_routes import debug_router # noqa: E402 app.include_router(debug_router) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8010)