diff --git a/app/Commands/AbdmPushSync.php b/app/Commands/AbdmPushSync.php index b682352..a9d28e7 100644 --- a/app/Commands/AbdmPushSync.php +++ b/app/Commands/AbdmPushSync.php @@ -3,6 +3,7 @@ namespace App\Commands; use App\Libraries\Abdm\Sync\AbdmSyncWorkerService; +use App\Libraries\Abdm\Sync\AbdmTaskBoardSyncService; use CodeIgniter\CLI\BaseCommand; use CodeIgniter\CLI\CLI; @@ -10,12 +11,13 @@ class AbdmPushSync extends BaseCommand { protected $group = 'ABDM'; protected $name = 'abdm:push-sync'; - protected $description = 'Process ABDM M2 sync outbox and push records to gateway /api/v3/records/push.'; - protected $usage = 'abdm:push-sync [--limit 20] [--worker worker-name]'; + protected $description = 'Process ABDM M2 sync outbox and ABDM Work Task Board care-context linking.'; + protected $usage = 'abdm:push-sync [--limit 20] [--worker worker-name] [--taskboard-only]'; protected $arguments = []; protected $options = [ - '--limit' => 'Maximum outbox rows to process in this run.', - '--worker' => 'Worker identifier for lock tracking.', + '--limit' => 'Maximum outbox and taskboard rows to process in this run.', + '--worker' => 'Worker identifier for lock tracking.', + '--taskboard-only' => 'Skip outbox queue and only sync ABDM Work Task Board records.', ]; public function run(array $params) @@ -30,14 +32,27 @@ class AbdmPushSync extends BaseCommand $worker = 'spark-abdm-push-sync'; } - $service = new AbdmSyncWorkerService(); - $summary = $service->process($limit, $worker); + $taskboardOnly = CLI::getOption('taskboard-only') !== null; - CLI::write('ABDM Push Sync', 'yellow'); - CLI::write('Processed: ' . (int) ($summary['processed'] ?? 0)); - CLI::write('Success: ' . (int) ($summary['success'] ?? 0), 'green'); - CLI::write('Failed: ' . (int) ($summary['failed'] ?? 0), ((int) ($summary['failed'] ?? 0) > 0 ? 'red' : 'green')); - CLI::write('Dead: ' . (int) ($summary['dead'] ?? 0), ((int) ($summary['dead'] ?? 0) > 0 ? 'red' : 'green')); - CLI::write('Skipped: ' . (int) ($summary['skipped'] ?? 0)); + if (! $taskboardOnly) { + $service = new AbdmSyncWorkerService(); + $summary = $service->process($limit, $worker); + + CLI::write('ABDM Push Sync (Outbox)', 'yellow'); + CLI::write('Processed: ' . (int) ($summary['processed'] ?? 0)); + CLI::write('Success: ' . (int) ($summary['success'] ?? 0), 'green'); + CLI::write('Failed: ' . (int) ($summary['failed'] ?? 0), ((int) ($summary['failed'] ?? 0) > 0 ? 'red' : 'green')); + CLI::write('Dead: ' . (int) ($summary['dead'] ?? 0), ((int) ($summary['dead'] ?? 0) > 0 ? 'red' : 'green')); + CLI::write('Skipped: ' . (int) ($summary['skipped'] ?? 0)); + } + + // Also sync ABDM Work Task Board records (OPD Consults & Work Queue items) + CLI::newLine(); + CLI::write('ABDM Work Task Board Sync', 'cyan'); + $tbService = new AbdmTaskBoardSyncService(); + $tbSummary = $tbService->syncAll($limit); + + CLI::write('OPD Consults -> Eligible: ' . ($tbSummary['opd']['eligible'] ?? 0) . ' | Linked: ' . ($tbSummary['opd']['linked'] ?? 0) . ' | Failed: ' . ($tbSummary['opd']['failed'] ?? 0), 'green'); + CLI::write('Work Tasks -> Eligible: ' . ($tbSummary['tasks']['eligible'] ?? 0) . ' | Linked: ' . ($tbSummary['tasks']['linked'] ?? 0) . ' | Failed: ' . ($tbSummary['tasks']['failed'] ?? 0), 'green'); } } diff --git a/app/Commands/AbdmTaskBoardSync.php b/app/Commands/AbdmTaskBoardSync.php new file mode 100644 index 0000000..1f3a69e --- /dev/null +++ b/app/Commands/AbdmTaskBoardSync.php @@ -0,0 +1,94 @@ + 'Maximum records to process per category (default: 20).', + '--type' => 'Which records to process: all, opd, or tasks (default: all).', + '--dry-run' => 'Inspect eligible records without linking/pushing.', + ]; + + public function run(array $params) + { + $limit = (int) (CLI::getOption('limit') ?? 20); + if ($limit <= 0) { + $limit = 20; + } + + $type = strtolower(trim((string) (CLI::getOption('type') ?? 'all'))); + if (! in_array($type, ['all', 'opd', 'tasks'], true)) { + $type = 'all'; + } + + $dryRun = CLI::getOption('dry-run') !== null; + + CLI::write('========================================================', 'cyan'); + CLI::write(' ABDM Work Task Board - Automated Care Context Sync' . ($dryRun ? ' (DRY RUN)' : ''), 'yellow'); + CLI::write('========================================================', 'cyan'); + CLI::write('Time: ' . date('Y-m-d H:i:s') . ' | Limit: ' . $limit . ' | Type: ' . $type); + + $service = new AbdmTaskBoardSyncService(); + + // 1. Process OPD Consult / Prescription Records + if ($type === 'all' || $type === 'opd') { + CLI::newLine(); + CLI::write('--- OPD Consult / Prescription Records ---', 'white'); + $opdSummary = $service->syncOpdConsultRecords($limit, $dryRun); + + CLI::write('Eligible : ' . $opdSummary['eligible']); + CLI::write('Linked : ' . $opdSummary['linked'], 'green'); + CLI::write('Failed : ' . $opdSummary['failed'], $opdSummary['failed'] > 0 ? 'red' : 'green'); + CLI::write('Skipped : ' . $opdSummary['skipped']); + + foreach ($opdSummary['details'] as $item) { + $statusColor = match ($item['status'] ?? '') { + 'linked' => 'green', + 'failed' => 'red', + default => 'yellow', + }; + $extra = isset($item['queue_id']) && $item['queue_id'] !== '' + ? ' [Queue: ' . $item['queue_id'] . (isset($item['bridge_record_id']) ? ' Bridge: #' . $item['bridge_record_id'] : '') . ']' + : (isset($item['error']) ? ' [Error: ' . $item['error'] . ']' : ''); + CLI::write(' OPD #' . ($item['opd_id'] ?? 0) . ' (' . ($item['patient'] ?? '') . ') -> ' . ($item['status'] ?? '') . $extra, $statusColor); + } + } + + // 2. Process Open Work Tasks + if ($type === 'all' || $type === 'tasks') { + CLI::newLine(); + CLI::write('--- ABDM Work Task Queue Items ---', 'white'); + $taskSummary = $service->syncOpenWorkTasks($limit, $dryRun); + + CLI::write('Eligible : ' . $taskSummary['eligible']); + CLI::write('Linked : ' . $taskSummary['linked'], 'green'); + CLI::write('Failed : ' . $taskSummary['failed'], $taskSummary['failed'] > 0 ? 'red' : 'green'); + CLI::write('Skipped : ' . $taskSummary['skipped']); + + foreach ($taskSummary['details'] as $item) { + $statusColor = match ($item['status'] ?? '') { + 'linked' => 'green', + 'failed' => 'red', + default => 'yellow', + }; + $extra = isset($item['queue_id']) && $item['queue_id'] !== '' + ? ' [Queue: ' . $item['queue_id'] . ']' + : (isset($item['error']) ? ' [Error: ' . $item['error'] . ']' : ''); + CLI::write(' Task #' . ($item['task_id'] ?? 0) . ' (' . ($item['task_type'] ?? '') . ') -> ' . ($item['status'] ?? '') . $extra, $statusColor); + } + } + + CLI::newLine(); + CLI::write('ABDM Task Board sync run completed.', 'cyan'); + } +} diff --git a/app/Controllers/Opd_prescription.php b/app/Controllers/Opd_prescription.php index 39e23b5..ce638fb 100644 --- a/app/Controllers/Opd_prescription.php +++ b/app/Controllers/Opd_prescription.php @@ -11781,7 +11781,7 @@ class Opd_prescription extends BaseController 'abha_id' => $abhaAddress !== '' ? $abhaAddress : null, 'hi_type' => 'OPConsultRecord', 'entity_type' => 'opd', - 'entity_id' => (string) $sessionId, + 'entity_id' => (string) $opdId, 'record_data' => $bundleJson, 'care_context_reference' => $careContextRef, 'push_status' => 'local_discovery_ready', diff --git a/app/Libraries/Abdm/Sync/AbdmTaskBoardSyncService.php b/app/Libraries/Abdm/Sync/AbdmTaskBoardSyncService.php new file mode 100644 index 0000000..95cada1 --- /dev/null +++ b/app/Libraries/Abdm/Sync/AbdmTaskBoardSyncService.php @@ -0,0 +1,811 @@ +db = $db ?? db_connect(); + $this->connector = $connector ?? AbdmConnectorFactory::make(); + $this->taskService = new AbdmWorkTaskService(); + } + + /** + * Run full ABDM Work Task Board sync (both OPD Consult and Work Queue tasks). + * + * @return array{ + * opd: array{eligible: int, linked: int, failed: int, skipped: int, details: array>}, + * tasks: array{eligible: int, linked: int, failed: int, skipped: int, details: array>} + * } + */ + public function syncAll(int $limit = 20, bool $dryRun = false): array + { + $opdSummary = $this->syncOpdConsultRecords($limit, $dryRun); + $tasksSummary = $this->syncOpenWorkTasks($limit, $dryRun); + + return [ + 'opd' => $opdSummary, + 'tasks' => $tasksSummary, + ]; + } + + /** + * Scan and link Done OPD records that have an ABHA identity and a generated FHIR bundle. + * + * @return array{ + * eligible: int, + * linked: int, + * failed: int, + * skipped: int, + * details: array> + * } + */ + public function syncOpdConsultRecords(int $limit = 20, bool $dryRun = false): array + { + $summary = [ + 'eligible' => 0, + 'linked' => 0, + 'failed' => 0, + 'skipped' => 0, + 'details' => [], + ]; + + if (! $this->db->tableExists('opd_master') || ! $this->db->tableExists('patient_master')) { + return $summary; + } + + $patientFields = $this->db->getFieldNames('patient_master') ?? []; + $abhaCol = $this->resolveFirstExistingColumn($patientFields, ['abha_id', 'abha_no', 'abha_address', 'abha']); + if ($abhaCol === null) { + return $summary; + } + + // 1. Fetch done appointments from the last 30 days where patient has ABHA + $rows = $this->db->table('opd_master o') + ->select('o.opd_id, o.p_id, o.P_name, o.apointment_date, o.opd_date, o.opd_status, o.doc_name, o.doc_id, p.' . $abhaCol . ' as abha_id, p.p_fname, p.gender, p.dob', false) + ->join('patient_master p', 'p.id = o.p_id', 'left') + ->where('o.opd_status', 2) + ->where('DATE(o.apointment_date) >=', date('Y-m-d', strtotime('-30 days')), false) + ->where('p.' . $abhaCol . ' !=', '') + ->orderBy('o.opd_id', 'DESC') + ->limit(max(50, $limit * 3)) + ->get() + ->getResultArray(); + + if (empty($rows)) { + return $summary; + } + + // Filter for valid ABHA (either 14-digit number or contains @) + $validRows = []; + foreach ($rows as $r) { + $rawAbha = trim((string) ($r['abha_id'] ?? '')); + if (preg_match('/^\d{14}$/', $rawAbha) === 1 || str_contains($rawAbha, '@')) { + $validRows[] = $r; + } + } + + if (empty($validRows)) { + return $summary; + } + + $opdIds = array_values(array_unique(array_map(static fn ($r) => (int) ($r['opd_id'] ?? 0), $validRows))); + + // 2. Fetch latest FHIR documents for these OPD records + $docByOpd = []; + if (! empty($opdIds) && $this->db->tableExists('opd_fhir_documents')) { + $docRows = $this->db->table('opd_fhir_documents') + ->select('id, opd_id, opd_session_id, bundle_type, bundle_json, generated_at') + ->whereIn('opd_id', $opdIds) + ->whereIn('bundle_type', ['OPConsultRecord', 'MedicationRequestBundle', 'PrescriptionRecord']) + ->orderBy('id', 'DESC') + ->get() + ->getResultArray(); + + foreach ($docRows as $doc) { + $oid = (int) ($doc['opd_id'] ?? 0); + if ($oid > 0 && ! isset($docByOpd[$oid])) { + $docByOpd[$oid] = $doc; + } + } + } + + // 3. Fetch existing health_records + $hrByOpd = []; + if (! empty($opdIds) && $this->db->tableExists('health_records')) { + $sessionIds = []; + foreach ($docByOpd as $d) { + $sId = (int) ($d['opd_session_id'] ?? 0); + if ($sId > 0) { + $sessionIds[] = (string) $sId; + } + } + + $searchEntityIds = array_merge( + array_map(static fn ($v) => (string) $v, $opdIds), + $sessionIds + ); + + $hrRows = $this->db->table('health_records') + ->select('id, entity_id, push_status, abdm_txn_id, bridge_record_id, care_context_reference, push_at, linked_at, updated_at') + ->where('entity_type', 'opd') + ->whereIn('entity_id', $searchEntityIds) + ->orderBy('id', 'DESC') + ->get() + ->getResultArray(); + + foreach ($hrRows as $hr) { + $eid = (int) ($hr['entity_id'] ?? 0); + if ($eid <= 0) { + continue; + } + if (! isset($hrByOpd[$eid])) { + $hrByOpd[$eid] = $hr; + } + } + } + + // 4. Process each eligible row + $processedCount = 0; + foreach ($validRows as $row) { + if ($processedCount >= $limit) { + break; + } + + $opdId = (int) ($row['opd_id'] ?? 0); + if ($opdId <= 0) { + continue; + } + + $doc = $docByOpd[$opdId] ?? null; + if ($doc === null) { + // No FHIR bundle generated yet for this OPD + continue; + } + + $sessionId = (int) ($doc['opd_session_id'] ?? 0); + + // Reconcile: check if health_records row was stored under opd_id or session_id + $hr = $hrByOpd[$opdId] ?? ($sessionId > 0 ? ($hrByOpd[$sessionId] ?? null) : null); + + // If stored under session_id, reconcile entity_id to opdId + if ($hr !== null && (string) ($hr['entity_id'] ?? '') === (string) $sessionId && (string) $sessionId !== (string) $opdId) { + try { + $this->db->table('health_records') + ->where('id', (int) $hr['id']) + ->update(['entity_id' => (string) $opdId]); + $hr['entity_id'] = (string) $opdId; + } catch (\Throwable) { + } + } + + $pushStatus = strtolower(trim((string) ($hr['push_status'] ?? ''))); + + // Already pushed/submitted/linked successfully? + if (in_array($pushStatus, ['queued', 'linked', 'pushed'], true)) { + $summary['skipped']++; + continue; + } + + $summary['eligible']++; + + $consultDate = (string) ($row['apointment_date'] ?? $row['opd_date'] ?? ''); + $visitDate = $consultDate !== '' ? date('Y-m-d', strtotime($consultDate)) : date('Y-m-d'); + $derivedCcRef = 'OPD-' . $opdId . '-S' . ($sessionId > 0 ? $sessionId : 0) . '-' . $visitDate; + + $careContextRef = trim((string) ($hr['care_context_reference'] ?? '')); + if ($careContextRef === '') { + $careContextRef = $derivedCcRef; + } + + if ($dryRun) { + $summary['details'][] = [ + 'opd_id' => $opdId, + 'patient' => trim((string) ($row['P_name'] ?? '')), + 'abha' => trim((string) ($row['abha_id'] ?? '')), + 'care_context' => $careContextRef, + 'status' => 'dry_run_eligible', + ]; + $processedCount++; + continue; + } + + // Perform care context linking and push to ABDM bridge + $pushResult = $this->linkAndPushOpdRecord($row, $doc, $careContextRef, $visitDate, $hr); + $processedCount++; + + if ($pushResult['ok'] === 1) { + $summary['linked']++; + $summary['details'][] = [ + 'opd_id' => $opdId, + 'patient' => trim((string) ($row['P_name'] ?? '')), + 'care_context' => $careContextRef, + 'queue_id' => $pushResult['queue_id'] ?? '', + 'bridge_record_id' => $pushResult['bridge_record_id'] ?? null, + 'status' => 'linked', + ]; + } else { + $summary['failed']++; + $summary['details'][] = [ + 'opd_id' => $opdId, + 'patient' => trim((string) ($row['P_name'] ?? '')), + 'care_context' => $careContextRef, + 'error' => $pushResult['error'] ?? 'Unknown push error', + 'status' => 'failed', + ]; + } + } + + return $summary; + } + + /** + * Link and push a single OPD record to ABDM gateway/bridge. + * + * @param array $opdRow + * @param array $docRow + * @param array|null $existingHr + * @return array{ok: int, queue_id?: string|null, bridge_record_id?: int|null, error?: string} + */ + private function linkAndPushOpdRecord( + array $opdRow, + array $docRow, + string $careContextRef, + string $visitDate, + ?array $existingHr + ): array { + $opdId = (int) ($opdRow['opd_id'] ?? 0); + $patientId = (int) ($opdRow['p_id'] ?? 0); + $patientName = trim((string) ($opdRow['p_fname'] ?? $opdRow['P_name'] ?? '')); + $rawAbha = trim((string) ($opdRow['abha_id'] ?? '')); + $genderRaw = (string) ($opdRow['gender'] ?? ''); + $gender = match ((string) $genderRaw) { + '1', 'M', 'm', 'Male' => 'M', + '2', 'F', 'f', 'Female' => 'F', + default => 'O', + }; + + $yearOfBirth = ''; + $dob = trim((string) ($opdRow['dob'] ?? '')); + if ($dob !== '' && $dob !== '0000-00-00') { + $yearOfBirth = substr($dob, 0, 4); + } + + $abhaNumber = preg_match('/^\d{14}$/', $rawAbha) === 1 ? $rawAbha : ''; + $abhaAddress = str_contains($rawAbha, '@') ? $rawAbha : ''; + + // If patient_master has abha_address separately, resolve it + if ($abhaAddress === '' && $this->db->tableExists('patient_master')) { + $pm = $this->db->table('patient_master') + ->select('abha_address, abha_id, abha_no') + ->where('id', $patientId) + ->get(1) + ->getRowArray(); + if (! empty($pm)) { + if (! empty($pm['abha_address'])) { + $abhaAddress = trim((string) $pm['abha_address']); + } + if ($abhaNumber === '' && ! empty($pm['abha_id']) && preg_match('/^\d{14}$/', trim((string) $pm['abha_id']))) { + $abhaNumber = trim((string) $pm['abha_id']); + } + if ($abhaNumber === '' && ! empty($pm['abha_no']) && preg_match('/^\d{14}$/', trim((string) $pm['abha_no']))) { + $abhaNumber = trim((string) $pm['abha_no']); + } + } + } + + $effectiveAbha = $abhaAddress !== '' ? $abhaAddress : $abhaNumber; + if ($effectiveAbha === '') { + return ['ok' => 0, 'error' => 'No valid ABHA found for patient #' . $patientId]; + } + + $bundleJson = (string) ($docRow['bundle_json'] ?? '{}'); + $bundle = json_decode($bundleJson, true); + if (! is_array($bundle)) { + $bundle = ['raw' => $bundleJson]; + } + + $bundleType = trim((string) ($docRow['bundle_type'] ?? 'PrescriptionRecord')); + $hiType = match ($bundleType) { + 'OPConsultRecord' => 'OPConsultRecord', + default => 'PrescriptionRecord', + }; + $careContextDisplay = $hiType === 'PrescriptionRecord' + ? 'Prescription - ' . $visitDate + : 'OPD Visit - ' . $visitDate; + + $now = Time::now('Asia/Kolkata')->toDateTimeString(); + + // 1. Upsert into health_records + $healthRecordId = (int) ($existingHr['id'] ?? 0); + $hrData = [ + 'patient_id' => $patientId, + 'abha_id' => $effectiveAbha, + 'hi_type' => $hiType, + 'entity_type' => 'opd', + 'entity_id' => (string) $opdId, + 'care_context_reference' => $careContextRef, + 'record_data' => $bundleJson, + 'push_status' => 'pending', + 'updated_at' => $now, + ]; + + try { + if ($healthRecordId > 0) { + $this->db->table('health_records')->where('id', $healthRecordId)->update($hrData); + } else { + $hrData['created_at'] = $now; + $this->db->table('health_records')->insert($hrData); + $healthRecordId = (int) $this->db->insertID(); + } + } catch (\Throwable $e) { + log_message('error', "[AbdmTaskBoardSync] health_records upsert error: " . $e->getMessage()); + } + + // 2. Push to ABDM Bridge Gateway + $pushData = [ + 'patient_id' => (string) $patientId, + 'patient_name' => $patientName !== '' ? $patientName : ('PATIENT-' . $patientId), + 'abha_id' => $abhaNumber, + 'abha_address' => $abhaAddress, + 'year_of_birth' => $yearOfBirth, + 'gender' => $gender, + 'hi_type' => $hiType, + 'record_type' => $hiType, + 'visit_date' => $visitDate, + 'care_context_reference' => $careContextRef, + 'care_context_display' => $careContextDisplay, + 'notes' => $careContextDisplay, + 'queue_id' => $careContextRef, + 'record_data' => $bundle, + 'fhir_bundle' => $bundle, + ]; + + $queueId = null; + $bridgeRecordId = 0; + $connectorError = null; + $isSuccess = false; + + try { + $result = $this->connector->pushRecord($pushData); + + $okVal = (int) ($result['ok'] ?? 0); + $httpCode = (int) ($result['http_code'] ?? 0); + $statusVal = strtolower((string) ($result['status'] ?? '')); + + if ($okVal === 1 || in_array($httpCode, [200, 201, 202, 409], true) || in_array($statusVal, ['queued', 'pushed', 'linked', 'duplicate'], true)) { + $isSuccess = true; + $queueId = (string) ($result['queue_id'] ?? $result['request_id'] ?? $careContextRef); + $bridgeRecordId = (int) ($result['record_id'] ?? $result['id'] ?? 0); + } else { + $connectorError = trim((string) ($result['message'] ?? $result['error_text'] ?? $result['error'] ?? 'Bridge push failed')); + } + } catch (\Throwable $e) { + $connectorError = $e->getMessage(); + log_message('warning', "[AbdmTaskBoardSync] Network exception pushing OPD #{$opdId}: " . $connectorError); + } + + // 3. Update health_records with final push status + if ($healthRecordId > 0 && $this->db->tableExists('health_records')) { + $hrUpdate = [ + 'push_status' => $isSuccess ? 'queued' : 'failed', + 'updated_at' => $now, + ]; + if ($queueId !== null && $queueId !== '') { + $hrUpdate['abdm_txn_id'] = $queueId; + } + if ($bridgeRecordId > 0) { + $hrFields = $this->db->getFieldNames('health_records') ?? []; + if (in_array('bridge_record_id', $hrFields, true)) { + $hrUpdate['bridge_record_id'] = $bridgeRecordId; + } + } + if ($isSuccess) { + $hrFields = $this->db->getFieldNames('health_records') ?? []; + if (in_array('push_at', $hrFields, true)) { + $hrUpdate['push_at'] = $now; + } + } + + try { + $this->db->table('health_records')->where('id', $healthRecordId)->update($hrUpdate); + } catch (\Throwable) { + } + } + + // 4. Update record_links + if ($this->db->tableExists('record_links')) { + try { + $existingLink = $this->db->table('record_links') + ->where('care_context_reference', $careContextRef) + ->get(1) + ->getRowArray(); + + $linkData = [ + 'abha_id' => $effectiveAbha, + 'care_context_reference' => $careContextRef, + 'health_record_id' => $healthRecordId > 0 ? $healthRecordId : null, + 'abdm_txn_id' => $queueId !== '' ? $queueId : null, + 'link_status' => $isSuccess ? 'pending' : 'failed', + 'updated_at' => $now, + ]; + + if (! empty($existingLink)) { + $this->db->table('record_links')->where('id', (int) $existingLink['id'])->update($linkData); + } else { + $linkData['created_at'] = $now; + $this->db->table('record_links')->insert($linkData); + } + } catch (\Throwable) { + } + } + + // 5. Complete open work task if any exists + if ($isSuccess && $this->db->tableExists('abdm_work_tasks')) { + try { + $this->db->table('abdm_work_tasks') + ->where('task_type', 'opd_prescription_publish') + ->where('entity_id', (string) $opdId) + ->whereIn('status', ['pending', 'in_progress']) + ->update([ + 'status' => 'completed', + 'last_action_result' => 'Linked by cron. Queue: ' . ($queueId ?? '-') . ' Bridge: #' . $bridgeRecordId, + 'completed_at' => $now, + 'updated_at' => $now, + ]); + } catch (\Throwable) { + } + } + + // 6. Log API activity + if ($this->db->tableExists('abdm_api_logs')) { + try { + $this->db->table('abdm_api_logs')->insert([ + 'channel' => 'bridge', + 'event_type' => 'abdm.opd.prescription.share.result', + 'endpoint' => '/api/v3/records/push', + 'http_method' => 'POST', + 'entity_type' => 'opd', + 'entity_id' => (string) $opdId, + 'status' => $isSuccess ? 'success' : 'error', + 'error_message' => $connectorError !== null ? mb_substr($connectorError, 0, 1000) : null, + 'created_at' => $now, + ]); + } catch (\Throwable) { + } + } + + if ($isSuccess) { + return [ + 'ok' => 1, + 'queue_id' => $queueId, + 'bridge_record_id' => $bridgeRecordId > 0 ? $bridgeRecordId : null, + ]; + } + + return [ + 'ok' => 0, + 'error' => $connectorError ?? 'Bridge push failed', + ]; + } + + /** + * Process open tasks on the ABDM Work Task Board (Lab, Radiology, Immunization, Wellness, Health Docs, Discharge). + * + * @return array{ + * eligible: int, + * linked: int, + * failed: int, + * skipped: int, + * details: array> + * } + */ + public function syncOpenWorkTasks(int $limit = 20, bool $dryRun = false): array + { + $summary = [ + 'eligible' => 0, + 'linked' => 0, + 'failed' => 0, + 'skipped' => 0, + 'details' => [], + ]; + + if (! $this->db->tableExists('abdm_work_tasks')) { + return $summary; + } + + $supportedTaskTypes = [ + 'opd_prescription_publish', + 'lab_report_publish', + 'radiology_report_publish', + 'immunization_record_publish', + 'wellness_record_publish', + 'health_document_publish', + 'ipd_discharge_publish', + ]; + + $tasks = $this->db->table('abdm_work_tasks') + ->whereIn('task_type', $supportedTaskTypes) + ->whereIn('status', ['pending', 'in_progress']) + ->orderBy('id', 'ASC') + ->limit(max(1, $limit)) + ->get() + ->getResultArray(); + + if (empty($tasks)) { + return $summary; + } + + $summary['eligible'] = count($tasks); + + if ($dryRun) { + foreach ($tasks as $t) { + $summary['details'][] = [ + 'task_id' => (int) $t['id'], + 'task_type' => $t['task_type'], + 'entity_id' => $t['entity_id'], + 'patient' => $t['patient_name'] ?? '', + 'status' => 'dry_run_eligible', + ]; + } + return $summary; + } + + foreach ($tasks as $task) { + $taskId = (int) $task['id']; + $taskType = (string) $task['task_type']; + $entityId = (int) ($task['entity_id'] ?? 0); + $patientId = (int) ($task['patient_id'] ?? 0); + $abhaId = trim((string) ($task['abha_id'] ?? '')); + + $result = $this->processIndividualWorkTask($task); + + if ($result['ok'] === 1) { + $summary['linked']++; + $this->taskService->markTaskStatus( + $taskId, + 'completed', + 'Linked by cron: ' . ($result['queue_id'] ?? 'done') + ); + $summary['details'][] = [ + 'task_id' => $taskId, + 'task_type' => $taskType, + 'entity_id' => $entityId, + 'status' => 'linked', + 'queue_id' => $result['queue_id'] ?? '', + ]; + } else { + $summary['failed']++; + $this->taskService->markTaskStatus( + $taskId, + 'pending', // remain pending for retry on next cron tick + 'Cron retry pending: ' . ($result['error'] ?? 'Sync failed') + ); + $summary['details'][] = [ + 'task_id' => $taskId, + 'task_type' => $taskType, + 'entity_id' => $entityId, + 'status' => 'failed', + 'error' => $result['error'] ?? 'Sync failed', + ]; + } + } + + return $summary; + } + + /** + * @param array $task + * @return array{ok: int, queue_id?: string|null, error?: string} + */ + private function processIndividualWorkTask(array $task): array + { + $taskType = (string) ($task['task_type'] ?? ''); + $entityId = (int) ($task['entity_id'] ?? 0); + $patientId = (int) ($task['patient_id'] ?? 0); + $abhaId = trim((string) ($task['abha_id'] ?? '')); + + if ($entityId <= 0 || $patientId <= 0) { + return ['ok' => 0, 'error' => 'Invalid task entity or patient']; + } + + // If it's an OPD prescription publish task, delegate to OPD consult push + if ($taskType === 'opd_prescription_publish') { + $docRow = null; + if ($this->db->tableExists('opd_fhir_documents')) { + $docRow = $this->db->table('opd_fhir_documents') + ->where('opd_id', $entityId) + ->orderBy('id', 'DESC') + ->get(1) + ->getRowArray(); + } + if (empty($docRow)) { + return ['ok' => 0, 'error' => 'No FHIR bundle for OPD #' . $entityId]; + } + + $opdRow = $this->db->table('opd_master')->where('opd_id', $entityId)->get(1)->getRowArray(); + if (empty($opdRow)) { + return ['ok' => 0, 'error' => 'OPD record not found']; + } + + $opdRow['abha_id'] = $abhaId; + $visitDate = ! empty($opdRow['apointment_date']) ? date('Y-m-d', strtotime((string) $opdRow['apointment_date'])) : date('Y-m-d'); + $ccRef = 'OPD-' . $entityId . '-S' . ($docRow['opd_session_id'] ?? 0) . '-' . $visitDate; + + return $this->linkAndPushOpdRecord($opdRow, $docRow, $ccRef, $visitDate, null); + } + + // Other record types: map and push care context + $hiType = match ($taskType) { + 'lab_report_publish', 'radiology_report_publish' => 'DiagnosticReportRecord', + 'immunization_record_publish' => 'ImmunizationRecord', + 'wellness_record_publish' => 'WellnessRecord', + 'health_document_publish' => 'HealthDocumentRecord', + 'ipd_discharge_publish' => 'DischargeSummaryRecord', + default => 'HealthDocumentRecord', + }; + + $patientRow = $this->db->table('patient_master')->where('id', $patientId)->get(1)->getRowArray(); + $patientName = trim((string) ($patientRow['p_fname'] ?? $task['patient_name'] ?? '')); + $gender = match ((string) ($patientRow['gender'] ?? '')) { + '1', 'M', 'm', 'Male' => 'M', + '2', 'F', 'f', 'Female' => 'F', + default => 'O', + }; + $dob = trim((string) ($patientRow['dob'] ?? '')); + $yearOfBirth = ($dob !== '' && $dob !== '0000-00-00') ? substr($dob, 0, 4) : ''; + + $abhaNumber = preg_match('/^\d{14}$/', $abhaId) === 1 ? $abhaId : ''; + $abhaAddress = str_contains($abhaId, '@') ? $abhaId : ''; + if ($abhaAddress === '' && ! empty($patientRow['abha_address'])) { + $abhaAddress = trim((string) $patientRow['abha_address']); + } + if ($abhaNumber === '' && ! empty($patientRow['abha_id']) && preg_match('/^\d{14}$/', trim((string) $patientRow['abha_id']))) { + $abhaNumber = trim((string) $patientRow['abha_id']); + } + + $visitDate = date('Y-m-d'); + $prefix = match ($taskType) { + 'lab_report_publish', 'radiology_report_publish' => 'LAB-', + 'immunization_record_publish' => 'IMM-', + 'wellness_record_publish' => 'WELLNESS-', + 'health_document_publish' => 'DOC-', + 'ipd_discharge_publish' => 'IPD-', + default => 'REC-', + }; + $careContextRef = $prefix . $entityId . '-' . $visitDate; + $careContextDisplay = $hiType . ' ' . $visitDate; + + $pushData = [ + 'patient_id' => (string) $patientId, + 'patient_name' => $patientName !== '' ? $patientName : ('PATIENT-' . $patientId), + 'abha_id' => $abhaNumber, + 'abha_address' => $abhaAddress, + 'year_of_birth' => $yearOfBirth, + 'gender' => $gender, + 'hi_type' => $hiType, + 'record_type' => $hiType, + 'visit_date' => $visitDate, + 'care_context_reference' => $careContextRef, + 'care_context_display' => $careContextDisplay, + 'notes' => $careContextDisplay, + 'queue_id' => $careContextRef, + ]; + + try { + $result = $this->connector->pushRecord($pushData); + $ok = (int) ($result['ok'] ?? 0); + $httpCode = (int) ($result['http_code'] ?? 0); + $statusVal = strtolower((string) ($result['status'] ?? '')); + + if ($ok === 1 || in_array($httpCode, [200, 201, 202, 409], true) || in_array($statusVal, ['queued', 'pushed', 'linked', 'duplicate'], true)) { + $queueId = (string) ($result['queue_id'] ?? $careContextRef); + + // Upsert health_records & record_links + $this->recordLinkedState( + $patientId, + $abhaAddress !== '' ? $abhaAddress : $abhaNumber, + $hiType, + $taskType, + (string) $entityId, + $careContextRef, + $queueId, + (int) ($result['record_id'] ?? 0) + ); + + return ['ok' => 1, 'queue_id' => $queueId]; + } + + return ['ok' => 0, 'error' => (string) ($result['message'] ?? $result['error_text'] ?? 'Push failed')]; + } catch (\Throwable $e) { + return ['ok' => 0, 'error' => $e->getMessage()]; + } + } + + private function recordLinkedState( + int $patientId, + string $abhaId, + string $hiType, + string $entityType, + string $entityId, + string $careContextRef, + string $queueId, + int $bridgeRecordId + ): void { + $now = Time::now('Asia/Kolkata')->toDateTimeString(); + + if ($this->db->tableExists('health_records')) { + try { + $existing = $this->db->table('health_records') + ->where('care_context_reference', $careContextRef) + ->get(1) + ->getRowArray(); + + $hrData = [ + 'patient_id' => $patientId, + 'abha_id' => $abhaId, + 'hi_type' => $hiType, + 'entity_type' => $entityType, + 'entity_id' => $entityId, + 'care_context_reference' => $careContextRef, + 'push_status' => 'queued', + 'abdm_txn_id' => $queueId, + 'updated_at' => $now, + ]; + if ($bridgeRecordId > 0 && in_array('bridge_record_id', $this->db->getFieldNames('health_records') ?? [], true)) { + $hrData['bridge_record_id'] = $bridgeRecordId; + } + + if (! empty($existing)) { + $this->db->table('health_records')->where('id', (int) $existing['id'])->update($hrData); + $hrId = (int) $existing['id']; + } else { + $hrData['created_at'] = $now; + $this->db->table('health_records')->insert($hrData); + $hrId = (int) $this->db->insertID(); + } + + if ($this->db->tableExists('record_links')) { + $this->db->table('record_links')->insert([ + 'health_record_id' => $hrId, + 'care_context_reference' => $careContextRef, + 'abdm_txn_id' => $queueId, + 'abha_id' => $abhaId, + 'link_status' => 'pending', + 'created_at' => $now, + 'updated_at' => $now, + ]); + } + } catch (\Throwable) { + } + } + } + + /** + * @param string[] $fields + * @param string[] $candidates + */ + private function resolveFirstExistingColumn(array $fields, array $candidates): ?string + { + foreach ($candidates as $candidate) { + if (in_array($candidate, $fields, true)) { + return $candidate; + } + } + + return null; + } +}