475 lines
15 KiB
Go
475 lines
15 KiB
Go
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/<YYYY-MM>/<YYYY-MM-DD>.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
|
|
}
|