mirror of
https://gitlab.com/signalytic/client-external/streamline/streamline-emr.git
synced 2026-09-11 18:51:31 +00:00
streamline-emr v2.3 custom sync service scripts, docker-compose updates
This commit is contained in:
+399
-182
@@ -1,228 +1,445 @@
|
||||
<?php
|
||||
/**
|
||||
* Data Synchronization Script
|
||||
*
|
||||
* @filename sync.php
|
||||
* @description Performs periodic data synchronization between local & remote database
|
||||
* @author conrad96
|
||||
* @date 2025-03-26
|
||||
* @license MIT
|
||||
*
|
||||
* @usage Executed via cronjob at configured intervals
|
||||
*
|
||||
* Overview:
|
||||
* - Check available services e.g Redis, MySQL
|
||||
* - Perform periodic query fetch
|
||||
* - Check Internet connectivity before uploading payload.
|
||||
* - Saving query result set to redis cache
|
||||
* - Upload cached data
|
||||
*
|
||||
* Cron Configuration Example:
|
||||
* execution of sync.php after every hour
|
||||
* 0 + + + php /app/sync.php >> /var/log/sync.log 2>&1
|
||||
*
|
||||
* Monitoring:
|
||||
* - script monitoring logs are saved in sync.log
|
||||
*/
|
||||
require __DIR__ . "/vendor/autoload.php";
|
||||
|
||||
use GuzzleHttp\Client;
|
||||
|
||||
use Monolog\Logger;
|
||||
use Monolog\Level;
|
||||
use Monolog\Handler\StreamHandler;
|
||||
|
||||
class sync{
|
||||
|
||||
protected $db_host;
|
||||
protected $db;
|
||||
protected $db_name;
|
||||
protected $db_user;
|
||||
protected $db_password;
|
||||
protected $server;
|
||||
|
||||
protected $redis;
|
||||
protected $redis_port;
|
||||
protected $redis_host;
|
||||
protected $redis_ttl;
|
||||
protected $redis_timeout = 5;
|
||||
|
||||
protected $db;
|
||||
protected $http;
|
||||
|
||||
protected $endpoint;
|
||||
|
||||
protected $logger;
|
||||
protected $log_file = __DIR__."/sync.log";
|
||||
|
||||
private $payload;
|
||||
protected $server_status_url;
|
||||
|
||||
function __construct(){
|
||||
# log service
|
||||
$this->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 = getenv('DB_DATABASE');
|
||||
$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_host = getenv("REDIS_PORT");
|
||||
$this->redis_port = getenv("REDIS_PORT");
|
||||
$this->redis_ttl = getenv("REDIS_TTL");
|
||||
|
||||
$this->redis = new Redis();
|
||||
$this->redis->connect($this->redis_host, $this->redis_host);
|
||||
$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,
|
||||
]);
|
||||
|
||||
# init fetch data
|
||||
$this->payload = $this->fetchData();
|
||||
|
||||
# check / upload cache
|
||||
$this->uploadCache();
|
||||
}
|
||||
|
||||
function databaseConnection(){
|
||||
$dsn = "mysql:host=". $this->db_host .";dbname=". $this->db;
|
||||
function startSync(){
|
||||
|
||||
$pdo = new PDO($dsn, $this->db_user, $this->db_password);
|
||||
$pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
|
||||
}
|
||||
try{
|
||||
if(!$this->endpoint) throw new Exception("No remote server-endpoint specified. sync terminated.");
|
||||
|
||||
function connectionFail(){
|
||||
$failedRequests = $this->redis->lrange("failed_requests", 0, -1);
|
||||
$this->logger->info("Checking internet connection.", ["url"=> $this->endpoint ]);
|
||||
|
||||
foreach ($failedRequests as $request) {
|
||||
$data = json_decode($request, true);
|
||||
|
||||
$response = sendData(true, $data['data']);
|
||||
|
||||
if ($response) {
|
||||
// success, remove from redis list
|
||||
$redis->lrem("failed_requests", $request, 1);
|
||||
}
|
||||
$response = $this->http->get($this->server_status_url, [
|
||||
'timeout'=> 10,
|
||||
'http_errors'=> true
|
||||
]);
|
||||
|
||||
$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 = $this->uploadData(); # upload payload
|
||||
|
||||
$this->logger->info("Internet connection available. uploading data.", ["upload"=> $upload]);
|
||||
|
||||
return true;
|
||||
}catch(Exception $ex){
|
||||
$this->logger->error($ex->getMessage());
|
||||
|
||||
$this->saveToCache($this->payload);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function fetchData(){
|
||||
$previous = $pdo->query("SELECT * FROM signalytic_data")->fetch(PDO::FETCH_ASSOC)['last_fetch_date'];
|
||||
public function fetchData(){
|
||||
$this->logger->info("Querying database. ", ["db"=> $this->db_name]);
|
||||
|
||||
$hospitalInformation = $pdo->query("SELECT * FROM hospital_information")->fetch(PDO::FETCH_ASSOC);
|
||||
$episodes = $pdo->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}'");
|
||||
try{
|
||||
$previous = $this->db->query("SELECT * FROM signalytic_data")->fetch(PDO::FETCH_ASSOC)['last_fetch_date'];
|
||||
|
||||
$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 = $pdo->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 = $pdo->query("SELECT * FROM users WHERE last_seen > '{$previous}'")->rowCount();
|
||||
$male_patients = $pdo->query("SELECT * FROM patients WHERE created_at > '{$previous}' AND gender = '1'")->rowCount();
|
||||
$female_patients = $pdo->query("SELECT * FROM patients WHERE created_at > '{$previous}' AND gender = '2'")->rowCount();
|
||||
$receipts = $pdo->query("SELECT * FROM track_receipts WHERE created_at > '{$previous}'")->rowCount();
|
||||
$pharmacyRec = $pdo->query("SELECT * FROM pharmacy_stock_reconciliations WHERE created_at > '{$previous}' AND completion_status = 1")->rowCount();
|
||||
$storeRec = $pdo->query("SELECT * FROM store_stock_reconciliations WHERE created_at > '{$previous}' AND completion_status = 1")->rowCount();
|
||||
$quotations = $pdo->query("SELECT * FROM quotations WHERE created_at > '{$previous}' AND receive_date IS NOT NULL")->rowCount();
|
||||
|
||||
$inpatients = $pdo->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(!$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) {
|
||||
$patients_discharged_male++;
|
||||
$episodes_created_male++;
|
||||
} else if ($row['gender'] == 2) {
|
||||
$patients_discharged_female++;
|
||||
$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,
|
||||
];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
$investigations = $pdo->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++;
|
||||
|
||||
$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++;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
$treatments = $pdo->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++;
|
||||
|
||||
$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 $data;
|
||||
}catch(Exception $ex){
|
||||
$this->logger->error($ex->getMessage());
|
||||
return false;
|
||||
}
|
||||
|
||||
$top_diagnosis = $pdo->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,
|
||||
]);
|
||||
return $data;
|
||||
}
|
||||
|
||||
function uploadData(){
|
||||
global $pdo;
|
||||
$server_endpoint = getenv("SERVER_URL");
|
||||
public function uploadData($endpoint = null, $data = null)
|
||||
{
|
||||
if(is_null($endpoint)) $endpoint= $this->endpoint;
|
||||
if(is_null($data)) $data= $this->payload;
|
||||
|
||||
$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);
|
||||
|
||||
$ch = curl_init($server_endpoint);
|
||||
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
|
||||
curl_setopt($ch, CURLOPT_POST, true);
|
||||
curl_setopt($ch, CURLOPT_POSTFIELDS, $data);
|
||||
curl_setopt($ch, CURLOPT_HTTPHEADER, ["Content-Type: application/json"]);
|
||||
|
||||
$response = curl_exec($ch);
|
||||
$httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
|
||||
$error = curl_error($ch);
|
||||
curl_close($ch);
|
||||
|
||||
if ($httpCode >= 200 && $httpCode < 300) {
|
||||
// success
|
||||
$stmt = $pdo->prepare("UPDATE signalytic_data SET last_fetch_date = ? WHERE id = ?");
|
||||
$stmt->execute([date("Y-m-d H:i:s"), 1]);
|
||||
$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;
|
||||
} else if (!$is_retry) {
|
||||
// request failed
|
||||
$failedRequest = json_encode([
|
||||
'timestamp' => date('Y-m-d H:i:s'),
|
||||
'data' => $data,
|
||||
'error' => $error ?: "HTTP Code $httpCode",
|
||||
]);
|
||||
|
||||
$redis = new Redis();
|
||||
$redis->connect(getenv('REDIS_HOST'), getenv('REDIS_PORT'));
|
||||
$redis->lpush("failed_requests", $failedRequest);
|
||||
}catch(Exception $ex){
|
||||
$this->logger->error($ex->getMessage());
|
||||
|
||||
$this->saveToCache($data);
|
||||
|
||||
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;
|
||||
} else {
|
||||
}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 with data: pattern
|
||||
$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();
|
||||
|
||||
Reference in New Issue
Block a user