db = db_connect(); $this->queueModel = new BridgeSyncQueueModel(); } /** * @param array $payload */ public function enqueue(string $eventType, array $payload, string $entityType = '', string $entityId = '', int $maxAttempts = 10): ?int { if (! $this->db->tableExists('bridge_sync_queue')) { return null; } if (! $this->canEnqueueEventForPatient($eventType, $payload)) { return null; } $payloadJson = (string) json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES); if ($payloadJson === '' || $payloadJson === 'null') { return null; } $this->queueModel->insert([ 'channel' => 'bridge', 'event_type' => $eventType, 'entity_type' => $entityType !== '' ? $entityType : null, 'entity_id' => $entityId !== '' ? $entityId : null, 'payload_json' => $payloadJson, 'payload_hash' => hash('sha256', $eventType . '|' . $entityType . '|' . $entityId . '|' . $payloadJson), 'status' => 'pending', 'attempts' => 0, 'max_attempts' => max(1, $maxAttempts), 'next_attempt_at' => null, ]); $insertId = $this->queueModel->getInsertID(); return $insertId > 0 ? (int) $insertId : null; } /** * @param array $payload */ private function canEnqueueEventForPatient(string $eventType, array $payload): bool { if (! in_array($eventType, self::ABHA_VERIFIED_ONLY_EVENT_TYPES, true)) { return true; } $patientId = $this->resolvePatientIdForVerifiedOnlyEvent($eventType, $payload); if ($patientId <= 0) { return false; } return $this->isPatientAbhaVerified($patientId); } /** * @param array $payload */ private function resolvePatientIdForVerifiedOnlyEvent(string $eventType, array $payload): int { $patientId = (int) ($payload['patient_id'] ?? 0); if ($patientId > 0) { return $patientId; } if ($eventType === 'opd.fhir.generated') { $opdId = (int) ($payload['opd_id'] ?? 0); if ($opdId <= 0 && isset($payload['bundle_json']) && is_array($payload['bundle_json'])) { $opdId = (int) ($payload['bundle_json']['opd_id'] ?? 0); } if ($opdId > 0 && $this->db->tableExists('opd_master')) { $row = $this->db->table('opd_master') ->select('p_id') ->where('id', $opdId) ->get(1) ->getRowArray(); return (int) ($row['p_id'] ?? 0); } } return 0; } private function isPatientAbhaVerified(int $patientId): bool { if ($patientId <= 0 || ! $this->db->tableExists('patient_master')) { return false; } $allFields = $this->db->getFieldNames('patient_master') ?? []; $selectFields = []; foreach (['abha_verified_status', 'abha_kyc_verified', 'abha_mobile_verified', 'abha_id', 'abha_no', 'abha', 'abha_address'] as $field) { if (in_array($field, $allFields, true)) { $selectFields[] = $field; } } if (empty($selectFields)) { return false; } $row = $this->db->table('patient_master') ->select(implode(',', $selectFields)) ->where('id', $patientId) ->get(1) ->getRowArray(); if (! is_array($row) || $row === []) { return false; } $abhaLinked = false; foreach (['abha_id', 'abha_no', 'abha', 'abha_address'] as $field) { if (array_key_exists($field, $row) && trim((string) $row[$field]) !== '') { $abhaLinked = true; break; } } if (! $abhaLinked) { return false; } $verifiedStatus = strtoupper(trim((string) ($row['abha_verified_status'] ?? ''))); if (in_array($verifiedStatus, ['1', 'VERIFIED', 'YES', 'Y', 'TRUE'], true)) { return true; } $kycVerified = (int) ($row['abha_kyc_verified'] ?? 0) === 1; $mobileVerified = (int) ($row['abha_mobile_verified'] ?? 0) === 1; if ($kycVerified && $mobileVerified) { return true; } // Backward-compat fallback for legacy schemas without verification meta columns. $hasVerificationMeta = in_array('abha_verified_status', $allFields, true) || in_array('abha_kyc_verified', $allFields, true) || in_array('abha_mobile_verified', $allFields, true); return ! $hasVerificationMeta && $abhaLinked; } /** * @return array */ public function processPending(int $limit = 50, string $workerName = 'spark-bridge-sync'): array { $summary = [ 'processed' => 0, 'sent' => 0, 'failed' => 0, 'skipped' => 0, 'message' => '', ]; if (! $this->db->tableExists('bridge_sync_queue')) { $summary['message'] = 'bridge_sync_queue table not found'; return $summary; } $rows = $this->db->table('bridge_sync_queue') ->whereIn('status', ['pending', 'retry']) ->groupStart() ->where('next_attempt_at IS NULL', null, false) ->orWhere('next_attempt_at <=', Time::now('Asia/Kolkata')->toDateTimeString()) ->groupEnd() ->orderBy('id', 'ASC') ->limit(max(1, $limit)) ->get() ->getResultArray(); if (empty($rows)) { $summary['message'] = 'no pending queue records'; return $summary; } $clientOptions = [ 'timeout' => 20, 'http_errors' => false, ]; if (defined('ENVIRONMENT') && in_array((string) ENVIRONMENT, ['development', 'testing'], true)) { $clientOptions['verify'] = false; } $client = service('curlrequest', $clientOptions); foreach ($rows as $row) { $queueId = (int) ($row['id'] ?? 0); if ($queueId <= 0) { $summary['skipped']++; continue; } $locked = $this->db->table('bridge_sync_queue') ->where('id', $queueId) ->whereIn('status', ['pending', 'retry']) ->update([ 'status' => 'processing', 'locked_at' => Time::now('Asia/Kolkata')->toDateTimeString(), 'locked_by' => $workerName, ]); if (! $locked) { $summary['skipped']++; continue; } $summary['processed']++; $attempts = (int) ($row['attempts'] ?? 0); $maxAttempts = max(1, (int) ($row['max_attempts'] ?? 10)); $payload = json_decode((string) ($row['payload_json'] ?? '{}'), true); if (! is_array($payload)) { $payload = ['raw' => (string) ($row['payload_json'] ?? '')]; } $dispatch = $this->buildDispatchContext($row, $payload); if (($dispatch['ok'] ?? false) !== true) { $reason = trim((string) ($dispatch['error'] ?? 'No dispatch configuration')); if ($dispatch['terminal'] ?? false) { // No real destination for this event type — skip permanently, don't retry/error. $this->db->table('bridge_sync_queue')->where('id', $queueId)->update([ 'status' => 'skipped', 'next_attempt_at' => null, 'last_error' => mb_substr($reason, 0, 500), 'locked_at' => null, 'locked_by' => null, ]); $summary['skipped']++; continue; } $nextAttempts = $attempts + 1; $isFinalFailure = $nextAttempts >= $maxAttempts; $delaySeconds = min(3600, (int) pow(2, min(10, $nextAttempts)) * 30); $this->db->table('bridge_sync_queue')->where('id', $queueId)->update([ 'status' => $isFinalFailure ? 'failed' : 'retry', 'attempts' => $nextAttempts, 'next_attempt_at' => $isFinalFailure ? null : date('Y-m-d H:i:s', time() + $delaySeconds), 'last_error' => mb_substr($reason, 0, 500), 'locked_at' => null, 'locked_by' => null, ]); $summary['failed']++; continue; } $channel = (string) ($dispatch['channel'] ?? 'bridge'); $endpoint = (string) ($dispatch['endpoint'] ?? ''); $method = strtoupper((string) ($dispatch['method'] ?? 'POST')); $headers = (array) ($dispatch['headers'] ?? ['Content-Type' => 'application/json']); $postBody = (array) ($dispatch['body'] ?? []); try { $response = $client->request($method, $endpoint, [ 'headers' => $headers, 'json' => $postBody, ]); $status = $response->getStatusCode(); $body = (string) $response->getBody(); $this->logApiCall( $channel, (string) ($row['event_type'] ?? ''), $endpoint, $method, (string) ($row['entity_type'] ?? ''), (string) ($row['entity_id'] ?? ''), $postBody, $status, $body, $status >= 200 && $status < 300 ? 'success' : 'error', $status >= 200 && $status < 300 ? '' : ('HTTP ' . $status) ); if ($status >= 200 && $status < 300) { $this->db->table('bridge_sync_queue')->where('id', $queueId)->update([ 'status' => 'sent', 'sent_at' => Time::now('Asia/Kolkata')->toDateTimeString(), 'last_error' => null, 'locked_at' => null, 'locked_by' => null, ]); $summary['sent']++; continue; } $nextAttempts = $attempts + 1; $isFinalFailure = $nextAttempts >= $maxAttempts; $delaySeconds = min(3600, (int) pow(2, min(10, $nextAttempts)) * 30); $this->db->table('bridge_sync_queue')->where('id', $queueId)->update([ 'status' => $isFinalFailure ? 'failed' : 'retry', 'attempts' => $nextAttempts, 'next_attempt_at' => $isFinalFailure ? null : date('Y-m-d H:i:s', time() + $delaySeconds), 'last_error' => 'HTTP ' . $status . ' ' . mb_substr(trim($body), 0, 500), 'locked_at' => null, 'locked_by' => null, ]); $summary['failed']++; } catch (\Throwable $e) { $this->logApiCall( $channel, (string) ($row['event_type'] ?? ''), $endpoint, $method, (string) ($row['entity_type'] ?? ''), (string) ($row['entity_id'] ?? ''), $postBody, null, '', 'error', $e->getMessage() ); $nextAttempts = $attempts + 1; $isFinalFailure = $nextAttempts >= $maxAttempts; $delaySeconds = min(3600, (int) pow(2, min(10, $nextAttempts)) * 30); $this->db->table('bridge_sync_queue')->where('id', $queueId)->update([ 'status' => $isFinalFailure ? 'failed' : 'retry', 'attempts' => $nextAttempts, 'next_attempt_at' => $isFinalFailure ? null : date('Y-m-d H:i:s', time() + $delaySeconds), 'last_error' => mb_substr($e->getMessage(), 0, 500), 'locked_at' => null, 'locked_by' => null, ]); $summary['failed']++; } } $summary['message'] = 'queue processing completed'; return $summary; } /** * Cleanup helper for historical rows created before ABHA verification gating. * * @return array */ public function cleanupAbhaVerifiedOnlyQueue(int $limit = 500, bool $dryRun = false): array { $summary = [ 'scanned' => 0, 'marked_skipped' => 0, 'kept' => 0, 'decode_errors' => 0, ]; if (! $this->db->tableExists('bridge_sync_queue')) { return $summary; } $rows = $this->db->table('bridge_sync_queue') ->select('id,event_type,payload_json,status') ->whereIn('event_type', self::ABHA_VERIFIED_ONLY_EVENT_TYPES) ->whereIn('status', ['pending', 'retry', 'failed']) ->orderBy('id', 'ASC') ->limit(max(1, $limit)) ->get() ->getResultArray(); foreach ($rows as $row) { $summary['scanned']++; $eventType = (string) ($row['event_type'] ?? ''); $payload = json_decode((string) ($row['payload_json'] ?? ''), true); if (! is_array($payload)) { $summary['decode_errors']++; continue; } if ($this->canEnqueueEventForPatient($eventType, $payload)) { $summary['kept']++; continue; } if (! $dryRun) { $this->db->table('bridge_sync_queue') ->where('id', (int) ($row['id'] ?? 0)) ->update([ 'status' => 'skipped', 'next_attempt_at' => null, 'locked_at' => null, 'locked_by' => null, 'last_error' => 'Skipped by ABHA verified-only cleanup', ]); } $summary['marked_skipped']++; } return $summary; } /** * @param array $requestPayload */ private function logApiCall( string $channel, string $eventType, string $endpoint, string $method, string $entityType, string $entityId, array $requestPayload, ?int $responseCode, string $responseBody, string $status, string $errorMessage ): void { if (! $this->db->tableExists('abdm_api_logs')) { return; } $requestJson = (string) json_encode($requestPayload, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES); $decodedBody = json_decode($responseBody, true); $responseJson = is_array($decodedBody) ? (string) json_encode($decodedBody, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) : trim($responseBody); $this->db->table('abdm_api_logs')->insert([ 'channel' => $channel !== '' ? $channel : 'bridge', 'event_type' => $eventType, 'endpoint' => $endpoint, 'http_method' => strtoupper($method), 'entity_type' => $entityType !== '' ? $entityType : null, 'entity_id' => $entityId !== '' ? $entityId : null, 'request_json' => $requestJson, 'response_code' => $responseCode, 'response_json' => $responseJson !== '' ? $responseJson : null, 'status' => $status, 'error_message' => $errorMessage !== '' ? mb_substr($errorMessage, 0, 1000) : null, 'created_at' => Time::now('Asia/Kolkata')->toDateTimeString(), ]); } private function readSetting(string $name): string { // 1. Environment variable (highest priority — .env or server env) $envValue = getenv($name); if ($envValue !== false) { $value = trim((string) $envValue); if ($value !== '') { return $value; } } // 2. PHP constant if (defined($name)) { $value = trim((string) constant($name)); if ($value !== '') { return $value; } } // 3. Config\AbdmConnector PHP config file (editable without DB access) static $dreamsoftConfigMap = null; if ($dreamsoftConfigMap === null) { try { $cfg = config('AbdmConnector'); $dreamsoftConfigMap = [ 'BRIDGE_SYNC_URL' => trim((string) ($cfg->dreamsoftBridgeUrl ?? '')), 'BRIDGE_SYNC_TOKEN' => trim((string) ($cfg->dreamsoftBridgeToken ?? '')), 'ABDM_BRIDGE_URL' => trim((string) ($cfg->abdmBridgeUrl ?? '')), 'ABDM_BRIDGE_TOKEN' => trim((string) ($cfg->abdmBridgeToken ?? '')), 'BRIDGE_SOURCE_CODE' => trim((string) ($cfg->dreamsoftSourceCode ?? '')), ]; } catch (\Throwable $e) { $dreamsoftConfigMap = []; } } if (isset($dreamsoftConfigMap[$name]) && $dreamsoftConfigMap[$name] !== '') { return $dreamsoftConfigMap[$name]; } // 4. hospital_setting DB table (lowest priority — runtime admin panel) if (! $this->db->tableExists('hospital_setting')) { return ''; } $row = $this->db->table('hospital_setting') ->select('s_value') ->where('s_name', $name) ->get(1) ->getRowArray(); return trim((string) ($row['s_value'] ?? '')); } /** * @param array $row * @param array $payload * @return array */ private function buildDispatchContext(array $row, array $payload): array { $eventType = (string) ($row['event_type'] ?? ''); if ($eventType === '') { return ['ok' => false, 'error' => 'Missing event_type']; } if (str_starts_with($eventType, 'snomed.')) { return $this->buildBridgeDispatchContext($row, $payload, 'csnotk'); } $isAbdmOrNhcx = str_starts_with($eventType, 'abdm.') || str_starts_with($eventType, 'nhcx.'); $provider = strtolower($this->readSetting('ABDM_SYNC_PROVIDER')); if ($provider === '' || ! $isAbdmOrNhcx) { $provider = strtolower($this->readSetting('BRIDGE_SYNC_PROVIDER')); } if ($provider === '') { $provider = 'bridge'; } if ($provider === 'eka' && $isAbdmOrNhcx) { return $this->buildEkaDispatchContext($row, $payload); } if ($isAbdmOrNhcx) { return $this->buildAbdmBridgeDispatchContext($row, $payload); } // Only snomed.*/abdm.*/nhcx. events have a real bridge destination configured // (CSNOtk only implements the SNOMED terminology endpoint on /api/bridge and // 404s on generic app events like ipd.*/patient.*/lab.* — see bridge debug logs). // Mark these as a terminal skip instead of retrying/erroring forever. return [ 'ok' => false, 'terminal' => true, 'error' => 'No bridge destination configured for event type "' . $eventType . '" (only snomed.*, abdm.*, nhcx.* events are routed)', ]; } /** * @param array $row * @param array $payload * @return array */ private function buildBridgeDispatchContext(array $row, array $payload, string $channel = 'bridge'): array { $endpoint = $this->readSetting('BRIDGE_SYNC_URL'); if ($endpoint === '') { return ['ok' => false, 'error' => 'BRIDGE_SYNC_URL is not configured']; } $token = $this->readSetting('BRIDGE_SYNC_TOKEN'); $source = $this->readSetting('BRIDGE_SOURCE_CODE'); if ($source === '') { $source = (string) ($this->readSetting('HOSPITAL_CODE') ?: 'hms-local'); } $headers = ['Content-Type' => 'application/json']; if ($token !== '') { $headers['Authorization'] = 'Bearer ' . $token; } return [ 'ok' => true, 'channel' => $channel, 'endpoint' => $endpoint, 'method' => 'POST', 'headers' => $headers, 'body' => [ 'queue_id' => (int) ($row['id'] ?? 0), 'source' => $source, 'event_type' => (string) ($row['event_type'] ?? ''), 'entity_type' => (string) ($row['entity_type'] ?? ''), 'entity_id' => (string) ($row['entity_id'] ?? ''), 'payload' => $payload, 'occurred_at' => (string) ($row['created_at'] ?? Time::now('Asia/Kolkata')->toDateTimeString()), ], ]; } /** * @param array $row * @param array $payload * @return array */ private function buildAbdmBridgeDispatchContext(array $row, array $payload): array { $endpoint = $this->readSetting('ABDM_BRIDGE_URL'); if ($endpoint === '') { // Backward compatibility fallback $endpoint = $this->readSetting('BRIDGE_SYNC_URL'); } if ($endpoint === '') { return ['ok' => false, 'error' => 'ABDM_BRIDGE_URL is not configured']; } $token = $this->readSetting('ABDM_BRIDGE_TOKEN'); if ($token === '') { // Backward compatibility fallback $token = $this->readSetting('BRIDGE_SYNC_TOKEN'); } $source = $this->readSetting('BRIDGE_SOURCE_CODE'); if ($source === '') { $source = (string) ($this->readSetting('HOSPITAL_CODE') ?: 'hms-local'); } $headers = ['Content-Type' => 'application/json']; if ($token !== '') { $headers['Authorization'] = 'Bearer ' . $token; } return [ 'ok' => true, 'channel' => 'abdm', 'endpoint' => $endpoint, 'method' => 'POST', 'headers' => $headers, 'body' => [ 'queue_id' => (int) ($row['id'] ?? 0), 'source' => $source, 'event_type' => (string) ($row['event_type'] ?? ''), 'entity_type' => (string) ($row['entity_type'] ?? ''), 'entity_id' => (string) ($row['entity_id'] ?? ''), 'payload' => $payload, 'occurred_at' => (string) ($row['created_at'] ?? Time::now('Asia/Kolkata')->toDateTimeString()), ], ]; } /** * @param array $row * @param array $payload * @return array */ private function buildEkaDispatchContext(array $row, array $payload): array { $baseUrl = rtrim($this->readSetting('EKA_BASE_URL'), '/'); if ($baseUrl === '') { return ['ok' => false, 'error' => 'EKA_BASE_URL is not configured']; } $eventType = (string) ($row['event_type'] ?? ''); $eventPath = $this->resolveEkaEventPath($eventType); if ($eventPath === '') { return ['ok' => false, 'error' => 'No Eka endpoint mapping found for event ' . $eventType]; } $token = $this->readSetting('EKA_BEARER_TOKEN'); $apiKey = $this->readSetting('EKA_API_KEY'); $clientId = $this->readSetting('EKA_CLIENT_ID'); $headers = ['Content-Type' => 'application/json']; if ($token !== '') { $headers['Authorization'] = 'Bearer ' . $token; } elseif ($apiKey !== '') { $headers['x-api-key'] = $apiKey; } if ($clientId !== '') { $headers['x-client-id'] = $clientId; } $source = (string) ($this->readSetting('HOSPITAL_CODE') ?: 'hms-local'); return [ 'ok' => true, 'channel' => 'eka', 'endpoint' => $baseUrl . $eventPath, 'method' => 'POST', 'headers' => $headers, 'body' => [ 'queue_id' => (int) ($row['id'] ?? 0), 'source' => $source, 'event_type' => $eventType, 'entity_type' => (string) ($row['entity_type'] ?? ''), 'entity_id' => (string) ($row['entity_id'] ?? ''), 'occurred_at' => (string) ($row['created_at'] ?? Time::now('Asia/Kolkata')->toDateTimeString()), 'payload' => $payload, ], ]; } private function resolveEkaEventPath(string $eventType): string { $mapping = [ 'abdm.abha.validate' => '/api/v1/abdm/abha/validate', 'abdm.abha.profile.fetch.requested' => '/api/v1/abdm/abha/profile', 'abdm.abha.create.requested' => '/api/v1/abdm/abha/create', 'abdm.abha.update.requested' => '/api/v1/abdm/abha/update', 'abdm.scan_share.lookup' => '/api/v1/abdm/scan-share/lookup', 'abdm.consent.requested' => '/api/v1/abdm/consents/request', 'abdm.fhir.share.requested' => '/api/v1/abdm/health-information/share', 'abdm.opd.prescription.share.requested' => '/api/v1/abdm/health-information/share', 'abdm.ipd.admission.share.requested' => '/api/v1/abdm/health-information/ipd-admission/share', 'abdm.ipd.discharge.share.requested' => '/api/v1/abdm/health-information/ipd-discharge/share', 'abdm.diagnosis.report.share.requested' => '/api/v1/abdm/health-information/diagnosis-report/share', 'nhcx.claim.created' => '/api/v1/nhcx/claims/create', 'nhcx.claim.status.requested' => '/api/v1/nhcx/claims/status', ]; $jsonMap = trim($this->readSetting('EKA_EVENT_ENDPOINTS_JSON')); if ($jsonMap !== '') { $decoded = json_decode($jsonMap, true); if (is_array($decoded)) { foreach ($decoded as $key => $value) { if (is_string($key) && is_string($value) && $key !== '' && $value !== '') { $mapping[$key] = $value; } } } } $path = trim((string) ($mapping[$eventType] ?? '')); if ($path === '') { return ''; } return '/' . ltrim($path, '/'); } }