logger = new Logger("data-sync"); $stream_handler = new StreamHandler($this->log_file); $this->logger->pushHandler($stream_handler); $this->logger->info("initializing sync."); # database setup $this->db_name = getenv('DB_DATABASE'); $this->db_host = getenv('DB_HOST'); $this->db_user = getenv('DB_USERNAME'); $this->db_password = getenv('DB_PASSWORD'); $this->db = new PDO("mysql:host=". $this->db_host .";dbname=". $this->db_name, $this->db_user, $this->db_password); $this->db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); if(!$this->db) $this->logger->error("failed to connect to database."); $this->logger->debug("Connected to database."); # redis $this->redis_host = getenv("REDIS_HOST"); $this->redis_port = getenv("REDIS_PORT"); $this->redis_ttl = getenv("REDIS_TTL"); $this->redis = new Redis(); $this->redis->connect($this->redis_host, $this->redis_port, $this->redis_timeout); if(!$this->redis) $this->logger->error('Failed to connect to redis server.'); $this->logger->debug("Connected to redis server."); # remote server/endpoint $this->server = getenv("SERVER"); $this->server_status_url = getenv("SERVER_STATUS_URL"); $this->endpoint = getenv('SERVER_ENDPOINT'); $this->http = new Client([ 'base_uri' => $this->server, 'timeout' => 3.0, ]); # fetch data $db_data = $this->fetchData(); } function startSync(){ try{ if(!$this->endpoint) throw new Exception("No remote server-endpoint specified. sync terminated."); $this->logger->info("Checking internet connection.", ["url"=> $this->endpoint ]); $response = $this->http->get($this->server_status_url, [ 'timeout'=> 10, 'http_errors'=> true, 'headers' => [ 'Accept' => 'application/json', 'X-Requested-With' => 'XMLHttpRequest' ], 'verify'=> true, 'allow_redirects'=> false ]); $status_code = $response->getStatusCode(); $response_body = $response->getBody()->getContents(); if($status_code != 200) throw new Exception("Internet connection unavailable, saving to cache."); $this->logger->debug("Internet connection available.", ["status_code"=> $status_code, "response"=> $response_body]); $upload_cache = $this->uploadCache(); if(!$upload_cache) throw new Exception("Upload failed. [". $upload_cache."]"); $this->logger->info("Internet connection available. uploading data.", ["upload"=> $upload]); return true; }catch(Exception $ex){ $this->logger->error($ex->getMessage()); return false; } } public function fetchData(){ $this->logger->info("Querying database. ", ["db"=> $this->db_name]); try{ $previous = $this->db->query("SELECT * FROM signalytic_data")->fetch(PDO::FETCH_ASSOC)['last_fetch_date']; if(!$previous) $this->logger->error("last fetch date not found. "); $hospitalInformation = $this->db->query("SELECT * FROM hospital_information")->fetch(PDO::FETCH_ASSOC); $episodes = $this->db->query("SELECT triage_id, consultation_id, gender, clinic_id, name, slug FROM patient_episodes JOIN patients ON patient_episodes.patient_id = patients.id LEFT JOIN clinics ON patient_episodes.clinic_id = clinics.id WHERE patient_episodes.created_at > '{$previous}'"); $episodes_created_male = 0; $episodes_created_female = 0; $episodes_completed_with_outcome = []; $episodes_per_clinic = []; $episodes_triaged = 0; while ($row = $episodes->fetch(PDO::FETCH_ASSOC)) { if ($row['gender'] == 1) { $episodes_created_male++; } else if ($row['gender'] == 2) { $episodes_created_female++; } if (!is_null($row['triage_id'])) { $episodes_triaged++; } if (!is_null($row['clinic_id'])) { $episodes_per_clinic[$row['clinic_id']] = [ 'clinic_name' => $row['name'], 'clinic_type' => $row['slug'], 'clinic_count' => ($episodes_per_clinic[$row['clinic_id']]['clinic_count'] ?? 0) + 1, ]; } if (!is_null($row['consultation_id'])) { $cons = $this->db->query("SELECT consultations.outcome_id, outcomes.name FROM consultations JOIN outcomes ON consultations.outcome_id = outcomes.id WHERE consultations.id = '{$row['consultation_id']}'")->fetch(PDO::FETCH_ASSOC); $episodes_completed_with_outcome[$cons['outcome_id']] = [ 'outcome_name' => $cons['name'], 'outcome_count' => ($episodes_completed_with_outcome[$cons['outcome_id']]['outcome_count'] ?? 0) + 1, ]; } } $users_seen = $this->db->query("SELECT * FROM users WHERE last_seen > '{$previous}'")->rowCount(); $male_patients = $this->db->query("SELECT * FROM patients WHERE created_at > '{$previous}' AND gender = '1'")->rowCount(); $female_patients = $this->db->query("SELECT * FROM patients WHERE created_at > '{$previous}' AND gender = '2'")->rowCount(); $receipts = $this->db->query("SELECT * FROM track_receipts WHERE created_at > '{$previous}'")->rowCount(); $pharmacyRec = $this->db->query("SELECT * FROM pharmacy_stock_reconciliations WHERE created_at > '{$previous}' AND completion_status = 1")->rowCount(); $storeRec = $this->db->query("SELECT * FROM store_stock_reconciliations WHERE created_at > '{$previous}' AND completion_status = 1")->rowCount(); $quotations = $this->db->query("SELECT * FROM quotations WHERE created_at > '{$previous}' AND receive_date IS NOT NULL")->rowCount(); $inpatients = $this->db->query("SELECT inpatient_info.discharged, inpatient_info.died_on, patients.gender FROM inpatient_info JOIN patients ON inpatient_info.patient_id = patients.id WHERE inpatient_info.created_at > '{$previous}'"); $patients_admitted_male = 0; $patients_admitted_female = 0; $patients_discharged_male = 0; $patients_discharged_female = 0; while ($row = $inpatients->fetch(PDO::FETCH_ASSOC)) { if ($row['gender'] == 1) { $patients_admitted_male++; } else if ($row['gender'] == 2) { $patients_admitted_female++; } if ($row['discharged'] === 1 || !is_null($row['died_on'])) { if ($row['gender'] == 1) { $patients_discharged_male++; } else if ($row['gender'] == 2) { $patients_discharged_female++; } } } $investigations = $this->db->query("SELECT * FROM ordered_investigations WHERE created_at > '{$previous}'"); $investigation_requests = 0; $investigation_results = 0; while ($row = $investigations->fetch(PDO::FETCH_ASSOC)) { $investigation_requests++; // top_investigations if ($row['investigation_status'] === 1) { $investigation_results++; } } $treatments = $this->db->query("SELECT * FROM treatments WHERE created_at > '{$previous}'"); $prescriptions = 0; $prescriptions_dispensed = 0; while ($row = $treatments->fetch(PDO::FETCH_ASSOC)) { $prescriptions++; if ($row['dispense_status'] === 1) { $prescriptions_dispensed++; } } $top_diagnosis = $this->db->query("SELECT primary_diagnosis, COUNT(*) AS freq, name FROM consultations JOIN diagnoses ON consultations.primary_diagnosis = diagnoses.id WHERE consultations.created_at > '{$previous}' GROUP BY primary_diagnosis ORDER BY freq DESC LIMIT 5")->fetchAll(); $data = json_encode([ 'facility_name' => $hospitalInformation['name'], 'facility_email' => $hospitalInformation['email'], 'facility_address' => $hospitalInformation['address'], 'facility_unique_identifier' => $hospitalInformation['unique_hospital_identifier'] ?? '', 'time_period' => "{$previous} to " . date("Y-m-d H:i:s"), 'episodes_created_male' => $episodes_created_male, 'episodes_created_female' => $episodes_created_female, 'episodes_completed_with_outcome' => $episodes_completed_with_outcome, 'episodes_triaged' => $episodes_triaged, 'patients_admitted_male' => $patients_admitted_male, 'patients_admitted_female' => $patients_admitted_female, 'patients_discharged_male' => $patients_discharged_male, 'patients_discharged_female' => $patients_discharged_female, 'number_of_prescriptions' => $prescriptions, 'number_of_prescriptions_dispensed' => $prescriptions_dispensed, 'payment_receipts_generated' => $receipts, 'active_users' => $users_seen, 'male_patients_registered' => $male_patients, 'female_patients_registered' => $female_patients, 'investigation_requests' => $investigation_requests, 'investigation_requests_with_results' => $investigation_results, 'times_inventory_stock_received' => $quotations, 'times_stock_reconciled_pharmacy' => $pharmacyRec, 'times_stock_reconciled_store' => $storeRec, 'top_diagnosis' => $top_diagnosis, 'top_investigations' => [], 'top_prescribed_drugs_opd' => [], 'top_prescribed_drugs_ipd' => [], 'top_dispensed_drugs_opd' => [], 'top_dispensed_drugs_ipd' => [], 'episodes_per_clinic' => $episodes_per_clinic]); $this->logger->debug("Fetched: ". $data); $this->logger->info("Queries executed successfully. data saved.", ['size'=> strlen($data)]); return $this->saveToCache($data); }catch(Exception $ex){ $this->logger->error($ex->getMessage()); return false; } } public function uploadData($endpoint = null, $data = null) { $this->logger->info("Uploading payload.", ["url"=> $this->server. $this->endpoint ]); try{ if(!$endpoint) throw new Exception("No endpoint specified."); if(empty($data)) throw new Exception("Payload is empty."); $options = [ 'json' => $data, 'allow_redirects' => false, 'timeout' => 5, 'headers' => [ 'Accept' => 'application/json', 'X-Requested-With' => 'XMLHttpRequest' ] ]; $response = $this->http->post($endpoint, $options); $statusCode = $response->getStatusCode(); $response_body = $response->getBody()->getContents(); if($statusCode != 200) throw new Exception("Unexpected http-response: ". $response_body); $this->logger->info("Data uploaded.", ["remote"=> $this->server. $endpoint, "status_code"=> $statusCode, "response"=> $response_body ]); $this->updateLastFetchDate(); return true; }catch(Exception $ex){ $this->logger->error($ex->getMessage()); return false; } } function updateLastFetchDate(){ try { $update_date = $this->db->prepare("UPDATE signalytic_data SET last_fetch_date = ? WHERE id = ?"); $update_date->execute([date("Y-m-d H:i:s"), 1]); $this->logger->info("Last fetch updated."); return true; }catch(Exception $ex){ $this->logger->error("Last fetch failed. ", [ 'error_msg'=> $ex->getMessage()]); return false; } } function saveToCache($data = null){ try{ if(!$data) throw new Exception("No data fetched."); $cache_key = "data:". date("d_m_Y_H"); $result = $this->redis->setex($cache_key, $this->redis_ttl, $data); if (!$result) $this->logger->error("cache: $cache_key failed", ["error"=> $result]); $this->logger->info("Cached: $cache_key"); return true; }catch(Exception $ex){ $this->logger->error($ex->getMessage()); return false; } } function uploadCache(){ $this->logger->info("Checking cache."); try{ $cache_keys = $this->getCacheKeys(); if(empty($cache_keys)) throw new Exception("Cache is empty. "); foreach($cache_keys as $cache_key) { $cached_data = $this->redis->get($cache_key); $this->logger->debug("[key] $cache_key found.", ["key"=>$cache_key, "data"=> $cached_data]); $result = $this->uploadData($this->endpoint, $cached_data); $this->logger->debug("[key] $cache_key uploaded", ["upload"=> $result]); $this->clearCache($cache_key); # clear cache } }catch(Exception $ex){ $this->logger->warning($ex->getMessage()); return false; } } private function getCacheKeys(){ $keys = []; $iterator = null; do { //get all keys $result = $this->redis->scan($iterator, 'data:*'); if ($result !== false) $keys = array_merge($keys, $result); } while ($iterator > 0); return $keys; } function clearCache($cache_key = null){ if(!$cache_key){ $this->logger->error("No cache key provided. exiting ..."); return false; } try{ $clear = $this->redis->del($cache_key); if($clear === 0) throw new Exception("Invalid cache key. [ $cache_key ]"); $this->logger->info("Cache cleared", ["key"=> $cache_key, "result"=> $clear ]); return true; }catch(Exception $ex){ $this->logger->error($ex->getMessage()); return false; } } function resetCache(){ try{ $this->logger->warning("Flushing cache"); $clear = $this->redis->flushDB(); if(!$clear) throw new Exception("Clearing cache failed. ($clear) "); $this->logger->info("Cache cleared", ["result"=> $clear]); return true; }catch(Exception $ex){ $this->logger->error($ex->getMessage()); return false; } } } $sync = new Sync(); $sync->startSync();