package medicationdispense import ( "bufio" "context" "encoding/json" "fmt" "io/fs" "net/http" "os" "path/filepath" "sort" "strconv" "strings" "time" "service/internal/infrastructure/database" extapi "service/internal/worker/interface" "service/pkg/logger" ) const ( trackerFile = "internal/worker/satusehat/medicationdispense/last_medicationdispense_id.txt" medReqMappingFile = "internal/worker/satusehat/medicationrequest/medicationrequest_mapping.txt" medReqMappingBaseDir = "internal/worker/satusehat/medicationrequest/mapperdata" medMappingFile = "internal/worker/satusehat/medication/medication_mapping.txt" medMappingBaseDir = "internal/worker/satusehat/medication/mapperdata" medDispenseMappingBaseDir = "internal/worker/satusehat/medicationdispense/mapperdata" rateLimitSleep = 60 * time.Second ) type Config struct { DBManager database.Service InternalBaseURL string OrganizationID string } type TokenManager interface { GetAccessToken() string ForceRefreshAndGetToken(ctx context.Context) (string, error) } type WorkerService interface { Run(ctx context.Context) } type mappedData struct { MedicationRequestID string MedicationID string EncounterID string } type mappingFileInfo struct { Path string Count int MinID int64 MaxID int64 } type worker struct { cfg Config repo Repository apiClient extapi.Client tokenManager TokenManager medReqMap map[int64]mappedData medFhirMap map[int64]string // fallback medicationID dari mapping medication (untuk entry legacy 2-part) minMedReqID int64 maxMedReqID int64 fileStats []mappingFileInfo // statistik per-file mapping medicationrequest (urut kronologis) } func NewWorker(cfg Config, tokenManager TokenManager) WorkerService { w := &worker{ cfg: cfg, repo: NewRepository(cfg.DBManager), apiClient: extapi.NewClient(15 * time.Second), tokenManager: tokenManager, medReqMap: make(map[int64]mappedData), medFhirMap: make(map[int64]string), } w.loadMedicationRequestMapping() w.loadMedicationFhirMapping() return w } func (w *worker) loadMedicationRequestMapping() { var minID, maxID int64 minSet := false var stats []mappingFileInfo loadFromFile := func(path string) { file, err := os.Open(path) if err != nil { return } defer file.Close() info := mappingFileInfo{Path: path} fileMinSet := false scanner := bufio.NewScanner(file) for scanner.Scan() { line := strings.TrimSpace(scanner.Text()) if line == "" { continue } parts := strings.Split(line, "|") // Format yang didukung: // 2 parts (legacy): id|medicationRequestFhirID // 3 parts : id|medicationRequestFhirID|medicationFhirID // 4 parts : id|medicationRequestFhirID|medicationFhirID|encounterID // Entry 2-part legacy tidak memiliki medicationID/encounterID β€” keduanya // di-fallback dari mapping medication & data DB pada saat runtime. if len(parts) >= 2 { id, err := strconv.ParseInt(parts[0], 10, 64) if err == nil { mData := mappedData{ MedicationRequestID: parts[1], } if len(parts) >= 3 { mData.MedicationID = parts[2] } if len(parts) >= 4 { mData.EncounterID = parts[3] } // Jika sudah ada entry sebelumnya yang lebih lengkap, jangan timpa // dengan entry yang lebih ringkas (mis. legacy 2-part menimpa daily 4-part). if existing, ok := w.medReqMap[id]; ok { if mData.MedicationID == "" { mData.MedicationID = existing.MedicationID } if mData.EncounterID == "" { mData.EncounterID = existing.EncounterID } } w.medReqMap[id] = mData info.Count++ if !fileMinSet || id < info.MinID { info.MinID = id fileMinSet = true } if id > info.MaxID { info.MaxID = id } if !minSet || id < minID { minID = id minSet = true } if id > maxID { maxID = id } } } } if info.Count > 0 { stats = append(stats, info) } } // Kumpulkan path mapping dalam urutan kronologis: legacy (paling tua) β†’ file harian // (mapperdata//.txt) di-sort ascending sehingga tanggal terkecil // dimuat lebih dulu. Worker akan menghabiskan ID tanggal terkecil sebelum lanjut ke // tanggal berikutnya (urutan iterasi tetap ditentukan oleh `lastID` dari DB query). var paths []string if _, err := os.Stat(medReqMappingFile); err == nil { paths = append(paths, medReqMappingFile) } if _, err := os.Stat(medReqMappingBaseDir); err == nil { var dailyPaths []string errWalk := filepath.WalkDir(medReqMappingBaseDir, func(path string, d fs.DirEntry, err error) error { if err != nil { return err } if d.IsDir() || !strings.HasSuffix(d.Name(), ".txt") { return nil } dailyPaths = append(dailyPaths, path) return nil }) if errWalk != nil { logger.Default().Warn("[MEDICATIONDISPENSE WORKER] Gagal walk direktori mapping medicationrequest", logger.ErrorField(errWalk)) } sort.Strings(dailyPaths) paths = append(paths, dailyPaths...) } for _, p := range paths { loadFromFile(p) } w.minMedReqID = minID w.maxMedReqID = maxID w.fileStats = stats } // logMappingProgress menampilkan ringkasan per-file mapping medicationrequest // pada saat startup, menunjukkan apakah data tanggal-tanggal sebelumnya sudah // DONE, IN_PROGRESS, atau masih PENDING relatif terhadap lastID. Tujuannya: // memastikan tidak ada hari yang terlewat sebelum lanjut ke hari berikutnya. func (w *worker) logMappingProgress(lastID int64) { if len(w.fileStats) == 0 { logger.Default().Warn("[MEDICATIONDISPENSE WORKER] ⚠️ Tidak ada file mapping medicationrequest yang dimuat") return } logger.Default().Info("[MEDICATIONDISPENSE WORKER] πŸ“‹ Status file mapping medicationrequest (urut kronologis)", logger.Int("file_count", len(w.fileStats)), logger.Int64("last_id", lastID), ) for _, fs := range w.fileStats { var status string switch { case fs.MaxID <= lastID: status = "DONE" case fs.MinID > lastID: status = "PENDING" default: status = "IN_PROGRESS" } logger.Default().Info("[MEDICATIONDISPENSE WORKER] πŸ“ "+filepath.Base(fs.Path), logger.String("status", status), logger.Int("count", fs.Count), logger.Int64("min_id", fs.MinID), logger.Int64("max_id", fs.MaxID), ) } } // loadMedicationFhirMapping memuat seluruh mapping medication (idxPesanObat β†’ medicationFhirID) // untuk fallback medicationID pada entry legacy medicationrequest yang hanya 2 parts. func (w *worker) loadMedicationFhirMapping() { loadFromFile := func(path string) { file, err := os.Open(path) if err != nil { return } defer file.Close() scanner := bufio.NewScanner(file) for scanner.Scan() { line := strings.TrimSpace(scanner.Text()) if line == "" { continue } parts := strings.Split(line, "|") if len(parts) >= 2 { id, err := strconv.ParseInt(parts[0], 10, 64) if err == nil && parts[1] != "" { w.medFhirMap[id] = parts[1] } } } } var paths []string if _, err := os.Stat(medMappingFile); err == nil { paths = append(paths, medMappingFile) } if _, err := os.Stat(medMappingBaseDir); err == nil { var dailyPaths []string errWalk := filepath.WalkDir(medMappingBaseDir, func(path string, d fs.DirEntry, err error) error { if err != nil { return err } if d.IsDir() || !strings.HasSuffix(d.Name(), ".txt") { return nil } dailyPaths = append(dailyPaths, path) return nil }) if errWalk != nil { logger.Default().Warn("[MEDICATIONDISPENSE WORKER] Gagal walk direktori mapping medication", logger.ErrorField(errWalk)) } sort.Strings(dailyPaths) paths = append(paths, dailyPaths...) } for _, p := range paths { loadFromFile(p) } } func (w *worker) appendMedicationDispenseMapping(idx int64, fhirID, medicationRequestID, medicationID, encounterID string) { now := logger.LocalNow() monthDir := filepath.Join(medDispenseMappingBaseDir, now.Format("2006-01")) if err := os.MkdirAll(monthDir, 0755); err != nil { logger.Default().Error("[MEDICATIONDISPENSE WORKER] Gagal membuat direktori mapping bulanan", logger.ErrorField(err)) return } mappingPath := filepath.Join(monthDir, now.Format("2006-01-02")+".txt") f, err := os.OpenFile(mappingPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644) if err != nil { logger.Default().Error("[MEDICATIONDISPENSE WORKER] Gagal membuka/membuat file mapping medicationdispense", logger.ErrorField(err)) return } defer f.Close() _, _ = f.WriteString(fmt.Sprintf("%d|%s|%s|%s|%s\n", idx, fhirID, medicationRequestID, medicationID, encounterID)) } func (w *worker) Run(ctx context.Context) { lastID := w.readLastID() // Jika lastID 0, mulai dari ID terkecil pada tanggal terkecil pada mapping // medicationrequest (file dimuat kronologis: legacy β†’ harian asc), sehingga semua // data tanggal terkecil dihabiskan dulu sebelum berlanjut ke tanggal berikutnya. if lastID == 0 && w.minMedReqID > 0 { lastID = w.minMedReqID - 1 } logger.Default().Info("[MEDICATIONDISPENSE WORKER] ⏳ Memulai proses sinkronisasi...", logger.Int64("lastID", lastID)) w.logMappingProgress(lastID) // testCount := 0 for { select { case <-ctx.Done(): logger.Default().Info("[MEDICATIONDISPENSE WORKER] πŸ›‘ Proses dihentikan.") return default: } md, err := w.repo.GetNextMedicationDispense(ctx, lastID) if err != nil { if !strings.Contains(err.Error(), "no rows") { logger.Default().Error("[MEDICATIONDISPENSE WORKER] ❌ Gagal query data dari DB", logger.ErrorField(err)) } time.Sleep(5 * time.Second) continue } // Cari MedicationRequest ID dan Medication ID dari mapping internal memory mData, ok := w.medReqMap[md.IdxPesanObat] if !ok { // Coba reload mapping file barangkali ada data baru w.loadMedicationRequestMapping() mData, ok = w.medReqMap[md.IdxPesanObat] } if ok { md.MedicationRequestID = mData.MedicationRequestID md.MedicationID = mData.MedicationID if mData.EncounterID != "" && md.EncounterID == "" { md.EncounterID = mData.EncounterID } // Fallback: medicationID dari mapping medication untuk entry legacy 2-part // yang tidak menyertakan medicationFhirID di file medicationrequest_mapping. if md.MedicationID == "" { medID, found := w.medFhirMap[md.IdxPesanObat] if !found { w.loadMedicationFhirMapping() medID, found = w.medFhirMap[md.IdxPesanObat] } if found { md.MedicationID = medID } } } else { if md.IdxPesanObat <= w.maxMedReqID { logger.Default().Warn(fmt.Sprintf("[MEDICATIONDISPENSE WORKER] ⏭️ Skip Dokumen: MedicationRequest ID tidak ditemukan di mapping untuk IdxPesanObat: %d (Di-skip/Error di MedicationRequest Worker)", md.IdxPesanObat)) lastID = md.IdxPesanObat os.MkdirAll("internal/worker/satusehat/medicationdispense", 0755) os.WriteFile(trackerFile, []byte(strconv.FormatInt(lastID, 10)), 0644) } else { logger.Default().Info(fmt.Sprintf("[MEDICATIONDISPENSE WORKER] ⏳ Menunggu data mapping baru dari MedicationRequest Worker untuk IdxPesanObat: %d...", md.IdxPesanObat)) time.Sleep(5 * time.Second) } continue } // Validasi data mandatory dari satudata if md.PatientID == "" || md.EncounterID == "" || md.PractitionerID == "" || md.MedicationRequestID == "" || md.MedicationID == "" { logger.Default().Warn("[MEDICATIONDISPENSE WORKER] ⏭️ Skip Dokumen: Data mandatory (PatientID/EncounterID/PractitionerID/MedicationRequestID/MedicationID) kosong", logger.Int64("idxpesanobat", md.IdxPesanObat), logger.String("patient_id", md.PatientID), logger.String("encounter_id", md.EncounterID), logger.String("practitioner_id", md.PractitionerID), logger.String("medication_request_id", md.MedicationRequestID), ) lastID = md.IdxPesanObat os.MkdirAll("internal/worker/satusehat/medicationdispense", 0755) os.WriteFile(trackerFile, []byte(strconv.FormatInt(lastID, 10)), 0644) continue } internalPayload := MapToInternalAPI(md, w.cfg.OrganizationID) reqURL := fmt.Sprintf("%s/satusehat/medicationdispense", strings.TrimRight(w.cfg.InternalBaseURL, "/")) logger.Default().Info("[MEDICATIONDISPENSE WORKER] πŸš€ Mengirim request ke API", logger.Int64("idxpesanobat", md.IdxPesanObat), logger.Any("payload", internalPayload)) respBody, statusCode, err := w.apiClient.PostJSON(ctx, reqURL, w.tokenManager.GetAccessToken(), internalPayload) if statusCode == http.StatusUnauthorized { logger.Default().Warn("[MEDICATIONDISPENSE WORKER] ⚠️ Token expired, refreshing...") if newToken, errRef := w.tokenManager.ForceRefreshAndGetToken(ctx); errRef == nil { respBody, statusCode, err = w.apiClient.PostJSON(ctx, reqURL, newToken, internalPayload) } } if isRateLimitResponse(statusCode, respBody) { logger.Default().Warn("[MEDICATIONDISPENSE WORKER] ⏸️ Rate limit dari Satu Sehat - jeda dan akan retry dokumen yang sama", logger.Int64("idxpesanobat", md.IdxPesanObat), logger.Int("status_code", statusCode), logger.String("response", truncateMD(string(respBody), 300)), logger.String("sleep", rateLimitSleep.String()), ) time.Sleep(rateLimitSleep) continue } var status, errMsg, fhirID string if err != nil || (statusCode != http.StatusOK && statusCode != http.StatusCreated) { status, errMsg = "FAILED", fmt.Sprintf("Err: %v, HTTP: %d", err, statusCode) logger.Default().Error("[MEDICATIONDISPENSE WORKER] ❌ Gagal mengirim data", logger.String("error", errMsg), logger.String("response", string(respBody))) } else { status = "SUCCESS" var res map[string]interface{} if json.Unmarshal(respBody, &res) == nil { if data, ok := res["data"].(map[string]interface{}); ok { fhirID, _ = data["id"].(string) } else { fhirID, _ = res["id"].(string) } } logger.Default().Info("[MEDICATIONDISPENSE WORKER] βœ… Berhasil mengirim data", logger.String("fhir_id", fhirID), logger.String("response", string(respBody))) } _ = w.repo.SaveSyncLog(ctx, MapToSyncLog(md.IdxPesanObat, fhirID, internalPayload, respBody, status, errMsg)) if status == "SUCCESS" && fhirID != "" { _ = w.repo.SaveSatuSehatID(ctx, md.IdxPesanObat, fhirID) w.appendMedicationDispenseMapping(md.IdxPesanObat, fhirID, md.MedicationRequestID, md.MedicationID, md.EncounterID) } lastID = md.IdxPesanObat os.MkdirAll("internal/worker/satusehat/medicationdispense", 0755) os.WriteFile(trackerFile, []byte(strconv.FormatInt(lastID, 10)), 0644) time.Sleep(500 * time.Millisecond) // testCount++ // if testCount >= 5 { // logger.Default().Info("[MEDICATIONDISPENSE WORKER] πŸ›‘ Uji coba 5 data selesai. Menghentikan worker.") // return // } } } func (w *worker) readLastID() int64 { data, err := os.ReadFile(trackerFile) if err != nil { return 0 } id, _ := strconv.ParseInt(strings.TrimSpace(string(data)), 10, 64) return id } // isRateLimitResponse mendeteksi respons rate-limit dari Satu Sehat // (HTTP 429 atau pesan body yang mengindikasikan limit pengiriman terlampaui). func isRateLimitResponse(statusCode int, respBody []byte) bool { if statusCode == http.StatusTooManyRequests { return true } body := strings.ToLower(string(respBody)) return strings.Contains(body, "too many requests") || strings.Contains(body, "rate limit") || strings.Contains(body, "limit exceeded") || strings.Contains(body, "limit pengiriman") || strings.Contains(body, "quota exceeded") } func truncateMD(s string, max int) string { if len(s) > max { return s[:max] + "..." } return s }