118 lines
3.1 KiB
Go
118 lines
3.1 KiB
Go
package condition
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"net/http"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"service/internal/infrastructure/database"
|
|
extapi "service/internal/worker/interface"
|
|
)
|
|
|
|
const trackerFile = "last_migrated_condition_id.txt"
|
|
|
|
type Config struct {
|
|
DBManager database.Service
|
|
InternalBaseURL string
|
|
InternalToken string
|
|
OrganizationID string
|
|
}
|
|
|
|
type WorkerService interface {
|
|
Run(ctx context.Context)
|
|
}
|
|
|
|
type worker struct {
|
|
cfg Config
|
|
repo Repository
|
|
apiClient extapi.Client
|
|
}
|
|
|
|
func NewWorker(cfg Config) WorkerService {
|
|
return &worker{
|
|
cfg: cfg,
|
|
repo: NewRepository(cfg.DBManager),
|
|
apiClient: extapi.NewClient(15 * time.Second),
|
|
}
|
|
}
|
|
|
|
func (w *worker) Run(ctx context.Context) {
|
|
log.Println("[CONDITION WORKER] Memulai proses sinkronisasi ke API Satu Sehat...")
|
|
lastID := w.readLastID()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
log.Println("[CONDITION WORKER] Proses dihentikan oleh sistem.")
|
|
return
|
|
default:
|
|
}
|
|
|
|
cond, err := w.repo.GetNextCondition(ctx, lastID)
|
|
if err != nil {
|
|
if err == sql.ErrNoRows {
|
|
time.Sleep(5 * time.Second)
|
|
continue
|
|
}
|
|
log.Printf("[CONDITION WORKER] Gagal mengambil data database: %v\n", err)
|
|
time.Sleep(5 * time.Second)
|
|
continue
|
|
}
|
|
|
|
log.Printf("[CONDITION WORKER] Memproses Diagnosa ID: %d\n", cond.IdxDiagnosa)
|
|
internalPayload := MapToInternalAPI(cond, w.cfg.OrganizationID)
|
|
reqURL := fmt.Sprintf("%s/satusehat/condition", strings.TrimRight(w.cfg.InternalBaseURL, "/"))
|
|
|
|
respBody, statusCode, err := w.apiClient.PostJSON(ctx, reqURL, w.cfg.InternalToken, internalPayload)
|
|
|
|
var status, errMsg, fhirID string
|
|
if err != nil {
|
|
status, errMsg = "FAILED", err.Error()
|
|
log.Printf("[CONDITION WORKER] Gagal HTTP Request (ID %d): %v\n", cond.IdxDiagnosa, err)
|
|
} else if statusCode != http.StatusOK && statusCode != http.StatusCreated {
|
|
status, errMsg = "FAILED", fmt.Sprintf("HTTP %d", statusCode)
|
|
log.Printf("[CONDITION WORKER] Gagal Response ID %d. HTTP: %d, Body: %s\n", cond.IdxDiagnosa, statusCode, string(respBody))
|
|
} else {
|
|
status = "SUCCESS"
|
|
var result map[string]interface{}
|
|
if errUnmarshal := json.Unmarshal(respBody, &result); errUnmarshal == nil {
|
|
if data, ok := result["data"].(map[string]interface{}); ok {
|
|
if id, ok := data["id"].(string); ok {
|
|
fhirID = id
|
|
}
|
|
} else if id, ok := result["id"].(string); ok {
|
|
fhirID = id
|
|
}
|
|
}
|
|
}
|
|
|
|
syncLog := MapToSyncLog(cond.IdxDiagnosa, fhirID, internalPayload, respBody, status, errMsg)
|
|
_ = w.repo.SaveSyncLog(ctx, syncLog)
|
|
|
|
if status == "SUCCESS" && fhirID != "" {
|
|
_ = w.repo.SaveSatuSehatID(ctx, cond.IdxDiagnosa, fhirID)
|
|
}
|
|
|
|
w.updateTracker(&lastID, cond.IdxDiagnosa)
|
|
time.Sleep(500 * time.Millisecond) // Rate limiting
|
|
}
|
|
}
|
|
|
|
func (w *worker) updateTracker(lastID *int64, currentID int64) {
|
|
*lastID = currentID
|
|
os.WriteFile(trackerFile, []byte(strconv.FormatInt(*lastID, 10)), 0644)
|
|
}
|
|
|
|
func (w *worker) readLastID() int64 {
|
|
data, _ := os.ReadFile(trackerFile)
|
|
id, _ := strconv.ParseInt(strings.TrimSpace(string(data)), 10, 64)
|
|
return id
|
|
}
|