ryan/satusehat-worker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Medication Update

Browse Source
This commit is contained in:
meninjar
2026-04-26 08:06:57 +00:00
parent 9a15d74283
commit 24d25aa14a
37 changed files with 82527 additions and 2172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/auth/repository.go
+10 -8
View File
@@ -106,7 +106,7 @@ func (r *repository) FindUserByEmail(ctx context.Context, email string) (*User,
var user User
q := query.DynamicQuery{
From: "users", // GORM default table name for User struct
From: "public.users", // Gunakan nama tabel yang eksplisit
Filters: []query.FilterGroup{{
Filters: []query.DynamicFilter{
query.CreateEqualFilter("Email", email),
@@ -117,7 +117,8 @@ func (r *repository) FindUserByEmail(ctx context.Context, email string) (*User,
if err := r.qb.ExecuteQueryRow(ctx, sqlxDB, q, &user); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, nil // Not found is not an error, service layer will handle it
// Not found is not an error, service layer will handle it
return nil, nil
}
return nil, fmt.Errorf("failed to find user by email: %w", err)
}
@@ -159,7 +160,7 @@ func (r *repository) ChangePasswordInTx(ctx context.Context, userID int64, newPa
}}
// Berikan objek transaksi 'tx' ke Query Builder
if _, err = r.qb.ExecuteUpdate(ctx, tx, "users", updateData, filters); err != nil {
if _, err = r.qb.ExecuteUpdate(ctx, tx, "public.users", updateData, filters); err != nil {
return fmt.Errorf("failed to update password in tx: %w", err)
}
@@ -170,7 +171,7 @@ func (r *repository) ChangePasswordInTx(ctx context.Context, userID int64, newPa
}
// Gunakan objek transaksi 'tx' yang sama
if _, err = r.qb.ExecuteInsert(ctx, tx, "user_activity_logs", logData); err != nil {
if _, err = r.qb.ExecuteInsert(ctx, tx, "public.user_activity_logs", logData); err != nil {
// Asumsikan tabel 'user_activity_logs' ada
return fmt.Errorf("failed to log activity in tx: %w", err)
}
@@ -187,7 +188,7 @@ func (r *repository) FindUserByID(ctx context.Context, id int64) (*User, error)
var user User
q := query.DynamicQuery{
From: "users", // GORM default table name for User struct
From: "public.users",
Filters: []query.FilterGroup{{
Filters: []query.DynamicFilter{
query.CreateEqualFilter("Id", id),
@@ -198,7 +199,8 @@ func (r *repository) FindUserByID(ctx context.Context, id int64) (*User, error)
if err := r.qb.ExecuteQueryRow(ctx, sqlxDB, q, &user); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, nil // Not found is not an error, service layer will handle it
// Not found is not an error, service layer will handle it
return nil, nil
}
return nil, fmt.Errorf("failed to find user by id: %w", err)
}
@@ -213,7 +215,7 @@ func (r *repository) FindRefreshToken(ctx context.Context, token string) (*Store
var stored StoredToken
q := query.DynamicQuery{
From: "refresh_tokens",
From: "public.refresh_tokens",
Filters: []query.FilterGroup{{
Filters: []query.DynamicFilter{query.CreateEqualFilter("Token", token)},
}},
@@ -238,5 +240,5 @@ func (r *repository) RevokeAndSaveTokens(ctx context.Context, oldToken string, n
// TODO: Sesuaikan dengan struktur tabel Refresh Token yang Anda miliki di database
// Ini adalah stub untuk memenuhi kontrak interface
return writeDB.WithContext(ctx).Exec("UPDATE refresh_tokens SET is_revoked = ? WHERE token = ?", true, oldToken).Error
return writeDB.WithContext(ctx).Exec("UPDATE public.refresh_tokens SET is_revoked = ? WHERE token = ?", true, oldToken).Error
}
internal/auth/service.go
+50 -21
View File
@@ -14,6 +14,7 @@ import (
"time"
"github.com/golang-jwt/jwt/v5"
"golang.org/x/crypto/bcrypt"
)
type StoredToken struct {
@@ -53,25 +54,39 @@ func NewService(cmdRepo CommandRepository, queryRepo QueryRepository, cache *cac
// Login mengimplementasikan otentikasi menggunakan data dummy untuk keperluan testing.
func (s *service) Login(ctx context.Context, req LoginRequest) (*LoginResponse, error) {
// Simulasi pengecekan kredensial dengan beberapa akun dummy
type dummyUser struct {
Password string
UserID string
RoleID string
Name string
// Prioritaskan pengecekan dummy users untuk development, terutama untuk akun sistem seperti worker.
if dummyUser, exists := dummyUsers[req.Email]; exists {
if dummyUser.Password == req.Password {
// Jika kredensial cocok dengan dummy user, langsung generate token.
return s.generateTokens(dummyUser.UserID, req.Email, dummyUser.RoleID, dummyUser.Name)
}
}
dummyUsers := map[string]dummyUser{
"[email protected]": {"password123", "1001", "admin", "Admin Dummy"},
"[email protected]": {"password123", "1002", "user", "User Dummy"},
"[email protected]": {"password123", "1003", "manager", "Manager Dummy"},
// Jika tidak ditemukan di dummy users atau password salah, lanjutkan pengecekan ke database.
user, err := s.queryRepo.FindUserByEmail(ctx, req.Email)
if err != nil {
// Jika terjadi error saat query, kembalikan internal error
return nil, errors.InternalError().Cause(err).Message("Gagal mencari pengguna").Build()
}
user, exists := dummyUsers[req.Email]
if !exists || user.Password != req.Password {
return nil, errors.UnauthorizedError().Message("Email atau password salah. Coba: [email protected], [email protected], [email protected] (password123)").Build()
// Jika pengguna tidak ditemukan di mana pun (dummy atau DB).
if user == nil {
return nil, errors.UnauthorizedError().Message("Email atau password salah.").Build()
}
// Jika pengguna ditemukan di database, validasi password
// Asumsi password di database di-hash menggunakan bcrypt
err = bcrypt.CompareHashAndPassword([]byte(user.Password), []byte(req.Password))
if err != nil {
// Jika password tidak cocok
return nil, errors.UnauthorizedError().Message("Email atau password salah.").Build()
}
// Jika login berhasil, generate token
return s.generateTokens(fmt.Sprintf("%d", user.Id), user.Email, fmt.Sprintf("%d", *user.RoleID), user.Email)
}
func (s *service) generateTokens(userID, email, roleID, name string) (*LoginResponse, error) {
secret := os.Getenv("JWT_SECRET")
if secret == "" {
secret = "fallback_secret_key_change_in_production"
@@ -79,20 +94,20 @@ func (s *service) Login(ctx context.Context, req LoginRequest) (*LoginResponse,
// Buat Access Token baru (Umur pendek, misal 15 Menit)
accessToken := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{
"user_id": user.UserID,
"email": req.Email,
"role_id": user.RoleID,
"name": user.Name,
"user_id": userID,
"email": email,
"role_id": roleID,
"name": name,
"exp": time.Now().Add(60 * time.Minute).Unix(),
})
newAccessTokenStr, _ := accessToken.SignedString([]byte(secret))
// Buat Refresh Token baru (Umur panjang, misal 7 Hari)
refreshToken := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{
"user_id": user.UserID,
"email": req.Email,
"role_id": user.RoleID,
"name": user.Name,
"user_id": userID,
"email": email,
"role_id": roleID,
"name": name,
"exp": time.Now().Add(7 * 24 * time.Hour).Unix(),
})
newRefreshTokenStr, _ := refreshToken.SignedString([]byte(secret))
@@ -106,6 +121,20 @@ func (s *service) Login(ctx context.Context, req LoginRequest) (*LoginResponse,
}, nil
}
type dummyUser struct {
Password string
UserID string
RoleID string
Name string
}
var dummyUsers = map[string]dummyUser{
"[email protected]": {"password123", "1001", "admin", "Admin Dummy"},
"[email protected]": {"password123", "1002", "user", "User Dummy"},
"[email protected]": {"password123", "1003", "manager", "Manager Dummy"},
"[email protected]": {"password123", "1004", "worker", "Worker Dummy"},
}
// Register mengimplementasikan pendaftaran user dengan respons data json dummy.
func (s *service) Register(ctx context.Context, req RegisterRequest) (interface{}, error) {
// Mengembalikan response dummy seolah-olah user berhasil dibuat di DB
internal/infrastructure/config/config.go
+22 -20
View File
@@ -124,16 +124,17 @@ type BpjsConfig struct {
}
type SatuSehatConfig struct {
Enabled bool `mapstructure:"enabled"`
OrgID string `mapstructure:"org_id"`
FasyakesID string `mapstructure:"fasyakes_id"`
ClientID string `mapstructure:"client_id"`
ClientSecret string `mapstructure:"client_secret"`
AuthURL string `mapstructure:"auth_url"`
BaseURL string `mapstructure:"base_url"`
ConsentURL string `mapstructure:"consent_url"`
KFAURL string `mapstructure:"kfa_url"`
Timeout time.Duration `mapstructure:"timeout"`
Enabled bool `mapstructure:"enabled"`
OrgID string `mapstructure:"org_id"`
FasyakesID string `mapstructure:"fasyakes_id"`
ClientID string `mapstructure:"client_id"`
ClientSecret string `mapstructure:"client_secret"`
AuthURL string `mapstructure:"auth_url"`
BaseURL string `mapstructure:"base_url"`
ConsentURL string `mapstructure:"consent_url"`
InternalFHIRServerURL string `mapstructure:"internal_fhir_server_url"`
KFAURL string `mapstructure:"kfa_url"`
Timeout time.Duration `mapstructure:"timeout"`
}
type SwaggerConfig struct {
@@ -271,16 +272,17 @@ func LoadConfig() *Config {
Timeout: parseDuration(getEnv("BPJS_TIMEOUT", "30s")),
},
SatuSehat: SatuSehatConfig{
Enabled: getEnvAsBool("SATUSEHAT_ENABLED", getEnvAsBool("SATU_SEHAT_ENABLED", false)),
OrgID: getEnv("SATUSEHAT_ORG_ID", getEnv("BRIDGING_SATUSEHAT_ORG_ID", "")),
FasyakesID: getEnv("SATUSEHAT_FASYAKES_ID", getEnv("BRIDGING_SATUSEHAT_FASYAKES_ID", "")),
ClientID: getEnv("SATUSEHAT_CLIENT_ID", getEnv("BRIDGING_SATUSEHAT_CLIENT_ID", "")),
ClientSecret: getEnv("SATUSEHAT_CLIENT_SECRET", getEnv("BRIDGING_SATUSEHAT_CLIENT_SECRET", "")),
AuthURL: getEnv("SATUSEHAT_AUTH_URL", getEnv("BRIDGING_SATUSEHAT_AUTH_URL", "https://api-satusehat.kemkes.go.id/oauth2/v1")),
BaseURL: getEnv("SATUSEHAT_BASE_URL", getEnv("BRIDGING_SATUSEHAT_BASE_URL", "https://api-satusehat.kemkes.go.id/fhir-r4/v1")),
ConsentURL: getEnv("SATUSEHAT_CONSENT_URL", getEnv("BRIDGING_SATUSEHAT_CONSENT_URL", "https://api-satusehat.dto.kemkes.go.id/consent/v1")),
KFAURL: getEnv("SATUSEHAT_KFA_URL", getEnv("BRIDGING_SATUSEHAT_KFA_URL", "https://api-satusehat.kemkes.go.id/kfa-v2")),
Timeout: parseDuration(getEnv("SATUSEHAT_TIMEOUT", getEnv("BRIDGING_SATUSEHAT_TIMEOUT", "30s"))),
Enabled: getEnvAsBool("SATUSEHAT_ENABLED", getEnvAsBool("SATU_SEHAT_ENABLED", false)),
OrgID: getEnv("SATUSEHAT_ORG_ID", getEnv("BRIDGING_SATUSEHAT_ORG_ID", "")),
FasyakesID: getEnv("SATUSEHAT_FASYAKES_ID", getEnv("BRIDGING_SATUSEHAT_FASYAKES_ID", "")),
ClientID: getEnv("SATUSEHAT_CLIENT_ID", getEnv("BRIDGING_SATUSEHAT_CLIENT_ID", "")),
ClientSecret: getEnv("SATUSEHAT_CLIENT_SECRET", getEnv("BRIDGING_SATUSEHAT_CLIENT_SECRET", "")),
AuthURL: getEnv("SATUSEHAT_AUTH_URL", getEnv("BRIDGING_SATUSEHAT_AUTH_URL", "https://api-satusehat.kemkes.go.id/oauth2/v1")),
BaseURL: getEnv("SATUSEHAT_BASE_URL", getEnv("BRIDGING_SATUSEHAT_BASE_URL", "https://api-satusehat.kemkes.go.id/fhir-r4/v1")),
ConsentURL: getEnv("SATUSEHAT_CONSENT_URL", getEnv("BRIDGING_SATUSEHAT_CONSENT_URL", "https://api-satusehat.dto.kemkes.go.id/consent/v1")),
InternalFHIRServerURL: getEnv("SATU_SEHAT_INTERNAL_FHIR_SERVER_URL", "http://localhost:8096/api/v1/"),
KFAURL: getEnv("SATUSEHAT_KFA_URL", getEnv("BRIDGING_SATUSEHAT_KFA_URL", "https://api-satusehat.kemkes.go.id/kfa-v2")),
Timeout: parseDuration(getEnv("SATUSEHAT_TIMEOUT", getEnv("BRIDGING_SATUSEHAT_TIMEOUT", "30s"))),
},
Swagger: SwaggerConfig{
Title: getEnv("SWAGGER_TITLE", "SERVICE API"),
internal/worker/auth.go
+150
View File
@@ -0,0 +1,150 @@
package worker
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"service/internal/auth"
"service/pkg/logger"
"strings"
"time"
)
// apiLoginResponse is a helper struct to unmarshal the nested API response.
type apiLoginResponse struct {
Data auth.LoginResponse `json:"data"`
}
// login performs the initial login to get the first set of tokens.
func (m *Manager) login(ctx context.Context) error {
loginURL := fmt.Sprintf("%s/auth/login", strings.TrimRight(m.cfg.SatuSehat.InternalFHIRServerURL, "/"))
reqBody := map[string]string{
"email": "[email protected]",
"password": "password123",
}
payloadBytes, _ := json.Marshal(reqBody)
req, err := http.NewRequestWithContext(ctx, "POST", loginURL, bytes.NewBuffer(payloadBytes))
if err != nil {
return fmt.Errorf("gagal membuat login request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
logger.Default().Info("[WORKER AUTH] Melakukan login untuk mendapatkan token awal...", logger.String("url", loginURL))
client := &http.Client{Timeout: 15 * time.Second}
resp, err := client.Do(req)
if err != nil {
return fmt.Errorf("gagal melakukan login request: %w", err)
}
defer resp.Body.Close()
// Baca body untuk logging jika terjadi error
bodyBytes, readErr := io.ReadAll(resp.Body)
if readErr != nil {
return fmt.Errorf("gagal membaca response body login: %w", readErr)
}
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("login gagal dengan status: %d, response: %s", resp.StatusCode, string(bodyBytes))
}
var apiResp apiLoginResponse
if err := json.Unmarshal(bodyBytes, &apiResp); err != nil {
return fmt.Errorf("gagal decode response login: %w, body: %s", err, string(bodyBytes))
}
loginResp := apiResp.Data
if loginResp.AccessToken == "" || loginResp.RefreshToken == "" {
logger.Default().Error("[WORKER AUTH] Token kosong diterima dari server", logger.Any("response", loginResp))
return fmt.Errorf("access token atau refresh token kosong diterima dari server")
}
m.accessToken = loginResp.AccessToken
m.refreshToken = loginResp.RefreshToken
logger.Default().Info("[WORKER AUTH] Berhasil mendapatkan token awal.")
return nil
}
// refresh uses the refresh_token to get a new access_token.
func (m *Manager) refresh(ctx context.Context) error {
refreshURL := fmt.Sprintf("%s/auth/refresh", strings.TrimRight(m.cfg.SatuSehat.InternalFHIRServerURL, "/"))
if m.refreshToken == "" {
return fmt.Errorf("tidak ada refresh token untuk digunakan, akan mencoba login ulang")
}
reqBody := auth.RefreshTokenRequest{
RefreshToken: m.refreshToken,
Provider: "jwt",
}
payloadBytes, _ := json.Marshal(reqBody)
req, err := http.NewRequestWithContext(ctx, "POST", refreshURL, bytes.NewBuffer(payloadBytes))
if err != nil {
return fmt.Errorf("gagal membuat refresh request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
logger.Default().Info("[WORKER AUTH] Melakukan refresh token...", logger.String("url", refreshURL))
client := &http.Client{Timeout: 15 * time.Second}
resp, err := client.Do(req)
if err != nil {
return fmt.Errorf("gagal melakukan refresh request: %w", err)
}
defer resp.Body.Close()
bodyBytes, readErr := io.ReadAll(resp.Body)
if readErr != nil {
return fmt.Errorf("gagal membaca response body refresh: %w", readErr)
}
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("refresh token gagal dengan status: %d, response: %s", resp.StatusCode, string(bodyBytes))
}
var apiResp apiLoginResponse
if err := json.Unmarshal(bodyBytes, &apiResp); err != nil {
return fmt.Errorf("gagal decode response refresh: %w, body: %s", err, string(bodyBytes))
}
refreshResp := apiResp.Data
if refreshResp.AccessToken == "" {
logger.Default().Error("[WORKER AUTH] Access token kosong diterima saat refresh", logger.Any("response", refreshResp))
return fmt.Errorf("access token kosong diterima dari server saat refresh")
}
m.accessToken = refreshResp.AccessToken
// Update refresh token if a new one is provided
if refreshResp.RefreshToken != "" {
m.refreshToken = refreshResp.RefreshToken
}
logger.Default().Info("[WORKER AUTH] Berhasil me-refresh token.")
return nil
}
// GetAccessToken safely retrieves the current access token.
func (m *Manager) GetAccessToken() string {
m.tokenMutex.RLock()
defer m.tokenMutex.RUnlock()
return m.accessToken
}
// ForceRefreshAndGetToken tries to refresh the token, falls back to login, and returns the new token.
func (m *Manager) ForceRefreshAndGetToken(ctx context.Context) (string, error) {
m.tokenMutex.Lock()
defer m.tokenMutex.Unlock()
if err := m.refresh(ctx); err != nil {
logger.Default().Warn("[WORKER AUTH] Refresh token gagal, mencoba login ulang...", logger.ErrorField(err))
if loginErr := m.login(ctx); loginErr != nil {
return "", fmt.Errorf("fallback login juga gagal: %w", loginErr)
}
}
return m.accessToken, nil
}
internal/worker/satusehat/imagingstudy/dto.go
+22 -19
View File
@@ -2,27 +2,30 @@ package imagingstudy
import (
"database/sql"
"time"
)
// ImagingStudyData merepresentasikan gabungan data dari Postgres (Satu Data) dan MySQL (RIS)
type ImagingStudyData struct {
RequestID string `db:"id"`
ServiceRequestID sql.NullString `db:"Entity_Id"`
EncounterID sql.NullString `db:"Encounter_Id"`
NoRegister sql.NullString `db:"No_register"`
PatientID sql.NullString `db:"Nomor_satusehat_pasien"`
PatientName sql.NullString `db:"Nama_lengkap"`
// Data dari MySQL (RIS)
Modality sql.NullString
StartedDate sql.NullTime
// ImagingStudyDB merepresentasikan struktur tabel dari SIMRS
type ImagingStudyDB struct {
ID int64 `db:"id"`
SourceID string // Holds the UUID from the source table (data_service_requests)
CreatedAt time.Time // Holds the Created_at from the source table for tracking
NoMR sql.NullString `db:"nomr"`
PatientIHS sql.NullString `db:"patient_ihs"`
PractitionerIHS sql.NullString `db:"practitioner_ihs"`
EncounterIHS sql.NullString `db:"encounter_ihs"`
AccessionNumber sql.NullString `db:"accession_number"`
ServiceRequestID sql.NullString `db:"service_request_id"`
Modality sql.NullString `db:"modality"`
StartedDate sql.NullTime `db:"started_date"`
}
type ImagingStudySyncLog struct {
RequestID string `db:"id"`
ImagingStudyID string `db:"imagingstudy_id"`
RequestPayload string `db:"request_payload"`
ResponsePayload string `db:"response_payload"`
Status string `db:"status"`
ErrorMessage string `db:"error_message"`
// SyncLog menyimpan log sinkronisasi untuk diaudit di database
type SyncLog struct {
ImagingStudyID string
SatuSehatID string
RequestPayload string
ResponsePayload string
Status string
ErrorMessage string
}
internal/worker/satusehat/imagingstudy/last_imagingstudy_id.txt
+1
View File
@@ -0,0 +1 @@
2026-04-24T12:46:09Z
internal/worker/satusehat/imagingstudy/mapper.go
+20 -13
View File
@@ -6,28 +6,35 @@ import (
)
// MapToInternalAPI memetakan data gabungan ke payload API Internal SatuSehat
func MapToInternalAPI(dbData *ImagingStudyData, orgID string) map[string]interface{} {
func MapToInternalAPI(dbData *ImagingStudyDB, orgID string) map[string]interface{} {
waktu := time.Now()
if dbData.StartedDate.Valid {
waktu = dbData.StartedDate.Time
}
procedureCode := dbData.Modality.String
if procedureCode == "" {
procedureCode = "DX" // Default ke "DX" jika modality kosong
}
payload := map[string]interface{}{
"request_id": dbData.RequestID,
"servicerequest_id": dbData.ServiceRequestID.String,
"encounter_id": dbData.EncounterID.String,
"patient_id": dbData.PatientID.String,
"accession_number": dbData.NoRegister.String, // Digunakan sebagai Identifier SATUSEHAT
"modality": dbData.Modality.String,
"started_date": waktu.Format(time.RFC3339),
"organization_id": orgID,
"accession_number": dbData.AccessionNumber.String,
"service_request_id": dbData.ServiceRequestID.String,
"patient_id": dbData.PatientIHS.String,
"encounter_id": dbData.EncounterIHS.String,
"status": "available",
"started": waktu.Format(time.RFC3339),
"procedure_code": procedureCode,
}
return payload
}
func MapToSyncLog(reqID string, fhirID string, reqPayload interface{}, respBody []byte, status string, errMsg string) ImagingStudySyncLog {
reqBytes, _ := json.MarshalIndent(reqPayload, "", " ")
return ImagingStudySyncLog{
RequestID: reqID,
ImagingStudyID: fhirID,
func MapToSyncLog(id string, fhirID string, reqPayload interface{}, respBody []byte, status string, errMsg string) SyncLog {
reqBytes, _ := json.Marshal(reqPayload)
return SyncLog{
ImagingStudyID: id,
SatuSehatID: fhirID,
RequestPayload: string(reqBytes),
ResponsePayload: string(respBody),
Status: status,
internal/worker/satusehat/imagingstudy/repository.go
+155 -47
View File
@@ -3,13 +3,16 @@ package imagingstudy
import (
"context"
"database/sql"
"service/internal/infrastructure/database"
"service/pkg/logger"
"service/pkg/utils/query"
"time"
)
type Repository interface {
GetNextPendingRequest(ctx context.Context) (*ImagingStudyData, error)
UpdateSyncStatus(ctx context.Context, reqID string, status string, responsePayload string) error
GetNextImagingStudy(ctx context.Context, lastCreatedAt time.Time) (*ImagingStudyDB, error)
SaveSatuSehatID(ctx context.Context, sourceID string, fhirID string) error
SaveSyncLog(ctx context.Context, logData SyncLog) error
}
type repository struct {
@@ -20,68 +23,173 @@ func NewRepository(dbManager database.Service) Repository {
return &repository{dbManager: dbManager}
}
// GetNextPendingRequest mengambil 1 data service request radiologi yang belum terkirim
func (r *repository) GetNextPendingRequest(ctx context.Context) (*ImagingStudyData, error) {
var data ImagingStudyData
// GetNextImagingStudy mengambil data imaging study berikutnya untuk diproses.
// Tracker `lastID` adalah primary key integer (`id`) dari tabel `data_service_requests`.
// Data akan diambil mulai dari tanggal 2026-04-15 dan melompati `lastCreatedAt` yang sudah diproses.
func (r *repository) GetNextImagingStudy(ctx context.Context, lastCreatedAt time.Time) (*ImagingStudyDB, error) {
var data ImagingStudyDB
// 1. Ambil koneksi Postgres (Satu Data) - Sesuaikan nama dengan konfigurasi Anda
pgDB, err := r.dbManager.GetDB("satudata")
// 1. Inisialisasi Koneksi ke Database yang diperlukan
satudataDB, err := r.dbManager.GetSQLXDB("satudata")
if err != nil {
return nil, err
}
risDB, err := r.dbManager.GetSQLXDB("ris")
if err != nil {
return nil, err
}
// Ambil dari Postgres
queryPG := `
SELECT
sr.id, sr."Entity_Id", sr."Encounter_Id", sr."No_register",
kp."Nomor_satusehat_pasien", kp."Nama_lengkap"
FROM public.data_service_requests sr
JOIN public.data_kunjungan_pasien kp ON sr."Encounter_Id" = kp."IDXDAFTAR_satusehat"
WHERE LOWER(sr."Order_from") = 'radiologi'
AND (sr."Send_status" IS NULL OR sr."Send_status" = 'FAILED')
ORDER BY sr."Created_at" ASC
LIMIT 1
`
err = pgDB.QueryRowContext(ctx, queryPG).Scan(
&data.RequestID, &data.ServiceRequestID, &data.EncounterID, &data.NoRegister,
&data.PatientID, &data.PatientName,
)
if err != nil { // Bisa berupa sql.ErrNoRows jika data habis
return nil, err
// --- Langkah 1: Ambil data pemicu dari `satudata` ---
qbSatuData := query.NewSQLQueryBuilder(query.DBTypePostgreSQL).SetSecurityOptions(false, 0)
// Struct penampung sementara untuk mendapat foreign key ke db lain
type BaseSatuData struct {
ID string `db:"id"` // Perbaikan: Ubah ke string untuk menampung UUID
ServiceRequestID sql.NullString `db:"service_request_id"`
EncounterIHS sql.NullString `db:"encounter_ihs"`
PatientIHS sql.NullString `db:"patient_ihs"`
NoRegister sql.NullString `db:"No_register"`
TanggalOrder sql.NullTime `db:"Tanggal_order"`
NoMR sql.NullString `db:"nomr"`
CreatedAt time.Time `db:"Created_at"`
PractitionerIHS sql.NullString `db:"practitioner_ihs"`
StartedDate sql.NullTime `db:"started_date"`
}
var baseData BaseSatuData
// Hardcode tanggal mulai
startDate, _ := time.Parse("2006-01-02", "2026-01-01")
qSatuData := query.DynamicQuery{
From: "public.data_service_requests",
Aliases: "dsr",
Fields: []query.SelectField{
{Expression: `dsr.id`},
{Expression: `dsr."Entity_Id"`, Alias: "service_request_id"},
{Expression: `dsr."Encounter_Id"`, Alias: "encounter_ihs"},
{Expression: `dsr."No_register"`},
{Expression: `dsr."Tanggal_order"`},
{Expression: `dsr."Occurrence_date_time"`, Alias: "started_date"},
{Expression: `dsr."Requester_reference"`, Alias: "practitioner_ihs"},
{Expression: `dsr."Created_at"`},
{Expression: `dkp."Nomor_satusehat_pasien"`, Alias: "patient_ihs"},
{Expression: `dkp."NOMR"`, Alias: "nomr"},
},
Joins: []query.Join{
{
Type: "LEFT",
Table: "public.data_kunjungan_pasien",
Alias: "dkp",
OnConditions: query.CreateAndFilterGroup([]query.DynamicFilter{
query.CreateFilter(`dkp."IDXDAFTAR_satusehat"`, query.OpEqual, `dsr."Encounter_Id"`),
}),
},
},
Filters: []query.FilterGroup{
{Filters: []query.DynamicFilter{
query.CreateFilter(`dsr."Created_at"`, query.OpGreaterThan, lastCreatedAt),
query.CreateFilter(`dsr."Tanggal_order"`, query.OpGreaterThanEqual, startDate),
query.CreateFilter(`dsr."Entity_Id"`, query.OpNotNull, nil),
}},
},
Sort: []query.SortField{query.CreateAscSort(`dsr."Created_at"`)},
Limit: 1,
}
// 2. Ambil koneksi MySQL (RIS) - Sesuaikan nama dengan konfigurasi Anda
mySQLDB, err := r.dbManager.GetDB("simrs")
if err == nil && data.NoRegister.Valid {
queryMySQL := `SELECT modality, foto FROM periksa WHERE noregister = ? LIMIT 1`
err = qbSatuData.ExecuteQueryRow(ctx, satudataDB, qSatuData, &baseData)
if err != nil {
return nil, err // Akan melempar sql.ErrNoRows jika tidak ada data baru
}
var modality sql.NullString
var foto sql.NullTime
// Map data dari `satudata` ke struct utama
// SourceID (UUID) digunakan untuk tracking, sedangkan ID (int64) adalah dari tabel log,
// yang akan diisi saat menyimpan log.
data.SourceID = baseData.ID
data.CreatedAt = baseData.CreatedAt
data.ID = 0 // ID dari tabel log akan diisi nanti, bukan dari query ini.
data.ServiceRequestID = baseData.ServiceRequestID
data.EncounterIHS = baseData.EncounterIHS
data.PatientIHS = baseData.PatientIHS
data.NoMR = baseData.NoMR
data.PractitionerIHS = baseData.PractitionerIHS
data.StartedDate = baseData.StartedDate
// Gunakan placeholder '?' jika MySQL, atau '$1' jika PostgreSQL
err = mySQLDB.QueryRowContext(ctx, queryMySQL, data.NoRegister.String).Scan(&modality, &foto)
if err == nil {
data.Modality = modality
data.StartedDate = foto
// --- Langkah 2: Ambil Accession Number dan Modality dari `ris` (MySQL) ---
if baseData.NoRegister.Valid {
qbRis := query.NewSQLQueryBuilder(query.DBTypeMySQL).SetSecurityOptions(false, 0)
// PERBAIKAN: Secara eksplisit mengizinkan kolom yang akan digunakan dalam filter query RIS.
// Ini untuk mengatasi masalah validasi pada query builder yang menganggap kolom 'noregister' tidak valid.
qbRis.SetAllowedColumns([]string{
"noregister", "DATE(daftar)",
})
risFilters := []query.DynamicFilter{
query.CreateFilter("noregister", query.OpEqual, baseData.NoRegister.String),
}
// Filter berdasarkan tanggal order jika ada, untuk mempersempit pencarian
if baseData.TanggalOrder.Valid {
risFilters = append(risFilters, query.CreateFilter("DATE(daftar)", query.OpEqual, baseData.TanggalOrder.Time.Format("2006-01-02")))
}
qRis := query.DynamicQuery{
From: "periksa",
Fields: []query.SelectField{
{Expression: "nofoto", Alias: "accession_number"},
{Expression: "modality", Alias: "modality"},
},
Filters: []query.FilterGroup{
{Filters: risFilters},
},
Limit: 1,
}
// Mengeluarkan Log hasil SQL query yang di-generate
sqlStr, args, errBuild := qbRis.BuildQuery(qRis)
if errBuild == nil {
logger.Default().Debug("[IMAGING STUDY REPO] RIS Query Generated", logger.String("query", sqlStr), logger.Any("args", args))
} else {
logger.Default().Error("[IMAGING STUDY REPO] Failed to build RIS Query", logger.ErrorField(errBuild))
}
// PERBAIKAN: Jangan abaikan semua error. Log error jika bukan 'sql.ErrNoRows'.
// Ini penting untuk mendeteksi masalah koneksi atau query ke database RIS.
if errRis := qbRis.ExecuteQueryRow(ctx, risDB, qRis, &data); errRis != nil && errRis != sql.ErrNoRows {
logger.Default().Warn("[IMAGING STUDY REPO] Gagal mengambil data dari RIS, melanjutkan tanpa data RIS.",
logger.String("query", sqlStr), logger.Any("args", args), logger.ErrorField(errRis))
}
}
return &data, nil
}
// UpdateSyncStatus memperbarui status pengiriman di PostgreSQL
func (r *repository) UpdateSyncStatus(ctx context.Context, reqID string, status string, responsePayload string) error {
pgDB, err := r.dbManager.GetDB("satudata")
func (r *repository) SaveSatuSehatID(ctx context.Context, sourceID string, fhirID string) error {
// Perhatian: Karena ID dari data_service_requests adalah UUID,
// kita tidak bisa lagi menggunakan `id` (int64) dari tabel log untuk melakukan UPDATE.
// Fungsi ini perlu penyesuaian lebih lanjut jika Anda perlu menyimpan fhir_id kembali ke tabel `data_service_requests`.
// Untuk sementara, fungsi ini tidak akan melakukan apa-apa untuk menghindari error.
// return nil
// satudataDB, err := r.dbManager.GetSQLXDB("satudata")
// if err != nil {
// return err
// }
// qb := query.NewSQLQueryBuilder(query.DBTypePostgreSQL).SetSecurityOptions(false, 0)
// _, err = qb.ExecuteUpdate(ctx, satudataDB, "public.data_service_requests", query.UpdateData{Columns: []string{"fhir_id"}, Values: []interface{}{fhirID}}, []query.FilterGroup{query.CreateAndFilterGroup([]query.DynamicFilter{query.CreateEqualFilter("id", sourceID)})})
// return err
return nil
}
func (r *repository) SaveSyncLog(ctx context.Context, logData SyncLog) error {
satudataDB, err := r.dbManager.GetSQLXDB("satudata")
if err != nil {
return err
}
query := `
UPDATE public.data_service_requests
SET "Send_status" = $1, "Send_response" = $2, "Updated_at" = NOW()
WHERE id = $3
`
_, err = pgDB.ExecContext(ctx, query, status, responsePayload, reqID)
qb := query.NewSQLQueryBuilder(query.DBTypePostgreSQL).SetSecurityOptions(false, 0)
insertData := query.InsertData{
Columns: []string{"imaging_id", "fhir_id", "request_payload", "response_payload", "status", "error_message"},
Values: []interface{}{logData.ImagingStudyID, logData.SatuSehatID, logData.RequestPayload, logData.ResponsePayload, logData.Status, logData.ErrorMessage},
}
_, err = qb.ExecuteUpsert(ctx, satudataDB, "public.log_satusehat_imagingstudy", insertData, []string{"imaging_id"}, []string{"fhir_id", "request_payload", "response_payload", "status", "error_message"})
return err
}
internal/worker/satusehat/imagingstudy/service.go
+260 -40
View File
@@ -5,94 +5,314 @@ import (
"database/sql"
"encoding/json"
"fmt"
"log"
"net/http"
"os"
"strings"
"time"
"service/internal/infrastructure/database"
extapi "service/internal/worker/interface"
"service/pkg/logger"
)
const trackerFile = "internal/worker/satusehat/imagingstudy/last_imagingstudy_id.txt"
type Config struct {
DBManager database.Service
InternalBaseURL string
InternalToken string
OrganizationID string
}
// TokenManager defines the interface for a centralized token provider.
type TokenManager interface {
GetAccessToken() string
ForceRefreshAndGetToken(ctx context.Context) (string, error)
}
type WorkerService interface {
Run(ctx context.Context)
}
type worker struct {
cfg Config
repo Repository
apiClient extapi.Client
cfg Config
repo Repository
apiClient extapi.Client
tokenManager TokenManager
}
func NewWorker(cfg Config) WorkerService {
return &worker{cfg: cfg, repo: NewRepository(cfg.DBManager), apiClient: extapi.NewClient(15 * time.Second)}
func NewWorker(cfg Config, tokenManager TokenManager) WorkerService {
return &worker{
cfg: cfg,
repo: NewRepository(cfg.DBManager),
apiClient: extapi.NewClient(60 * time.Second),
tokenManager: tokenManager,
}
}
func (w *worker) Run(ctx context.Context) {
log.Println("[IMAGINGSTUDY WORKER] Memulai proses sinkronisasi...")
logger.Default().Info("[IMAGING STUDY WORKER] Memulai proses sinkronisasi API...")
lastCreatedAt := w.readLastCreatedAt()
logger.Default().Info("[IMAGING STUDY WORKER] Melanjutkan proses dari", logger.Time("last_created_at", lastCreatedAt))
for {
select {
case <-ctx.Done():
log.Println("[IMAGINGSTUDY WORKER] Proses dihentikan.")
logger.Default().Warn("[IMAGING STUDY WORKER] Proses dihentikan oleh sistem (Graceful Shutdown)")
return
default:
}
// 1. Ambil data dengan status belum terkirim (IS NULL atau FAILED)
isData, err := w.repo.GetNextPendingRequest(ctx)
// 1. Fetch Data
imagingData, err := w.repo.GetNextImagingStudy(ctx, lastCreatedAt)
if err != nil {
if err == sql.ErrNoRows {
// Jika tidak ada data antrean, istirahat lebih lama
time.Sleep(10 * time.Second)
time.Sleep(5 * time.Second) // Tunggu data baru
continue
}
log.Printf("[IMAGINGSTUDY WORKER] Error query db: %v", err)
logger.Default().Error("[IMAGING STUDY WORKER] Gagal query ke database", logger.ErrorField(err))
time.Sleep(5 * time.Second)
continue
}
// 2. Petakan ke payload SATUSEHAT via Internal API
internalPayload := MapToInternalAPI(isData, w.cfg.OrganizationID)
reqURL := fmt.Sprintf("%s/satusehat/imagingstudy", strings.TrimRight(w.cfg.InternalBaseURL, "/"))
// Fallback pencarian Encounter ID lewat API jika data di tabel kosong
if (!imagingData.EncounterIHS.Valid || imagingData.EncounterIHS.String == "") && imagingData.ServiceRequestID.Valid && imagingData.ServiceRequestID.String != "" {
logger.Default().Info("[IMAGING STUDY WORKER] Encounter ID kosong, mencoba fetch dari API ServiceRequest", logger.String("servicerequest_id", imagingData.ServiceRequestID.String))
// 3. Eksekusi pengiriman API
respBody, statusCode, err := w.apiClient.PostJSON(ctx, reqURL, w.cfg.InternalToken, internalPayload)
reqURL := fmt.Sprintf("%s/satusehat/servicerequest/%s", strings.TrimRight(w.cfg.InternalBaseURL, "/"), imagingData.ServiceRequestID.String)
srRespBody, srStatusCode, srErr := w.apiClient.GetJSON(ctx, reqURL, w.tokenManager.GetAccessToken())
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)
} 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)
if srStatusCode == http.StatusUnauthorized {
logger.Default().Warn("[IMAGING STUDY WORKER] 401 Unauthorized saat GET ServiceRequest, refresh token...")
if newToken, errRef := w.tokenManager.ForceRefreshAndGetToken(ctx); errRef == nil {
srRespBody, srStatusCode, srErr = w.apiClient.GetJSON(ctx, reqURL, newToken)
}
}
if srErr == nil && (srStatusCode == http.StatusOK || srStatusCode == http.StatusCreated) {
var parsedResp map[string]interface{}
if errParse := json.Unmarshal(srRespBody, &parsedResp); errParse == nil {
var encounterRef string
if data, ok := parsedResp["data"].(map[string]interface{}); ok {
if enc, ok := data["encounter"].(map[string]interface{}); ok {
encounterRef, _ = enc["reference"].(string)
}
} else if enc, ok := parsedResp["encounter"].(map[string]interface{}); ok {
encounterRef, _ = enc["reference"].(string)
}
if encounterRef != "" {
if parts := strings.Split(encounterRef, "/"); len(parts) > 1 {
imagingData.EncounterIHS.String = parts[len(parts)-1]
imagingData.EncounterIHS.Valid = true
logger.Default().Info("[IMAGING STUDY WORKER] Berhasil mendapatkan Encounter ID dari API", logger.String("encounter_id", imagingData.EncounterIHS.String))
}
}
}
}
}
// Buat response data berupa string JSON untuk log & penyimpanan error
logData := MapToSyncLog(isData.RequestID, fhirID, internalPayload, respBody, status, errMsg)
logBytes, _ := json.Marshal(logData)
// Gunakan SourceID (UUID) untuk logging dan tracking
logger.Default().Info("[IMAGING STUDY WORKER] 🔄 Memproses Dokumen",
logger.String("source_id", imagingData.SourceID),
logger.String("no_mr", imagingData.NoMR.String),
)
// 4. Update status ke tabel data_service_requests
errUpdate := w.repo.UpdateSyncStatus(ctx, isData.RequestID, status, string(logBytes))
if errUpdate != nil {
log.Printf("[IMAGINGSTUDY WORKER] Gagal update status di database untuk ID %s: %v", isData.RequestID, errUpdate)
} else {
log.Printf("[IMAGINGSTUDY WORKER] Sukses proses ID %s dengan status %s", isData.RequestID, status)
// Skip jika accession_number kosong / null
if !imagingData.AccessionNumber.Valid || imagingData.AccessionNumber.String == "" {
logger.Default().Warn("[IMAGING STUDY WORKER] ⏭️ Skip Dokumen: Accession Number kosong (Data RIS tidak ditemukan)",
logger.String("source_id", imagingData.SourceID),
logger.String("no_mr", imagingData.NoMR.String),
)
w.updateTracker(&lastCreatedAt, imagingData.CreatedAt)
continue
}
time.Sleep(500 * time.Millisecond)
// 2. Mapping Payload
internalPayload := MapToInternalAPI(imagingData, w.cfg.OrganizationID)
reqURL := fmt.Sprintf("%s/satusehat/imagingstudy", strings.TrimRight(w.cfg.InternalBaseURL, "/"))
logger.Default().Debug("[IMAGING STUDY WORKER] 📡 Mengirim Request ke API Part 3",
logger.String("url", reqURL),
logger.Any("payload", internalPayload),
)
// 3. Eksekusi Kirim HTTP via Client Helper Anda
fhirID, respBody, status, errMsg, shouldRetry := w.sendToInternalAPI(ctx, internalPayload, imagingData.SourceID)
/*
if err != nil {
// Kegagalan Jaringan/Timeout
status = "FAILED"
errMsg = err.Error()
logger.Default().Error("[IMAGING STUDY WORKER] ❌ Gagal Koneksi ke API Part 3",
logger.String("source_id", imagingData.SourceID),
logger.String("no_mr", imagingData.NoMR.String),
logger.Any("payload_sent", internalPayload),
logger.ErrorField(err),
)
shouldRetry = true // Wajib retry jika jaringan terputus/timeout
} else if statusCode != http.StatusOK && statusCode != http.StatusCreated {
// Kegagalan Response API
status = "FAILED"
errMsg = fmt.Sprintf("HTTP %d", statusCode)
logger.Default().Error("[IMAGING STUDY WORKER] ❌ Respons Error dari API Part 3",
logger.String("source_id", imagingData.SourceID),
logger.Int("status_code", statusCode),
logger.String("response", truncate(string(respBody), 200)),
)
// Retry jika Internal Server Error (5xx) atau Rate Limit (429)
if statusCode >= 500 || statusCode == http.StatusTooManyRequests {
shouldRetry = true
}
} else {
// Sukses
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
}
}
logger.Default().Info("[IMAGING STUDY WORKER] ✅ Sukses mengirim data Imaging Study",
logger.String("source_id", imagingData.SourceID),
logger.String("fhir_id", fhirID),
)
}*/
// 4. Catat Log ke Database
syncLog := MapToSyncLog(imagingData.SourceID, fhirID, internalPayload, respBody, status, errMsg)
if errLog := w.repo.SaveSyncLog(ctx, syncLog); errLog != nil {
logger.Default().Error("[IMAGING STUDY WORKER] Gagal menyimpan histori log ke DB", logger.ErrorField(errLog))
}
// 5. Update FHIR ID di tabel utama bila sukses
if status == "SUCCESS" && fhirID != "" {
if errUpdate := w.repo.SaveSatuSehatID(ctx, imagingData.SourceID, fhirID); errUpdate != nil {
logger.Default().Error("[IMAGING STUDY WORKER] Gagal update FHIR ID ke transaksi utama", logger.ErrorField(errUpdate))
}
}
// Update Tracker TXT agar tidak dikirim ulang terus-menerus KECUALI jika butuh retry
if !shouldRetry {
w.updateTracker(&lastCreatedAt, imagingData.CreatedAt)
} else {
// Memberitahu log bahwa data yang sama akan dicoba lagi pada loop berikutnya
logger.Default().Warn("[IMAGING STUDY WORKER] ⚠️ Retrying data", logger.String("source_id", imagingData.SourceID), logger.Time("created_at", imagingData.CreatedAt), logger.String("reason", errMsg))
}
// Jeda
if shouldRetry {
time.Sleep(5 * time.Second) // Tunggu sebentar jika server tujuan sedang sibuk
} else {
time.Sleep(2 * time.Second) // Jeda antar pengiriman diperpanjang untuk mencegah Rate Limit
}
}
}
// sendToInternalAPI sends the payload to the internal FHIR server, handles authentication and retries.
func (w *worker) sendToInternalAPI(ctx context.Context, payload map[string]interface{}, sourceID string) (fhirID string, respBody []byte, status string, errMsg string, shouldRetry bool) {
reqURL := fmt.Sprintf("%s/satusehat/imagingstudy", strings.TrimRight(w.cfg.InternalBaseURL, "/"))
// --- First Attempt ---
respBody, statusCode, err := w.apiClient.PostJSON(ctx, reqURL, w.tokenManager.GetAccessToken(), payload)
if err != nil {
// Network errors are always retried
logger.Default().Error("[IMAGING STUDY WORKER] ❌ Gagal Koneksi ke API Part 3",
logger.String("source_id", sourceID),
logger.Any("payload_sent", payload),
logger.ErrorField(err),
)
return "", nil, "FAILED", err.Error(), true
}
// --- Handle Unauthorized (401) ---
if statusCode == http.StatusUnauthorized {
logger.Default().Warn("[IMAGING STUDY WORKER] Menerima status 401 Unauthorized, mencoba refresh token...", logger.String("source_id", sourceID))
newToken, refreshErr := w.tokenManager.ForceRefreshAndGetToken(ctx)
if refreshErr != nil {
logger.Default().Error("[IMAGING STUDY WORKER] Gagal refresh token, akan mencoba lagi pada iterasi berikutnya.", logger.String("source_id", sourceID), logger.ErrorField(refreshErr))
return "", respBody, "FAILED", refreshErr.Error(), true
}
logger.Default().Info("[IMAGING STUDY WORKER] Token berhasil di-refresh, mencoba ulang request...", logger.String("source_id", sourceID))
// --- Second Attempt ---
respBody, statusCode, err = w.apiClient.PostJSON(ctx, reqURL, newToken, payload)
if err != nil {
logger.Default().Error("[IMAGING STUDY WORKER] ❌ Gagal Koneksi ke API Part 3 (setelah retry)",
logger.String("source_id", sourceID),
logger.ErrorField(err),
)
return "", nil, "FAILED", err.Error(), true
}
}
// --- Process Final Response ---
if statusCode != http.StatusOK && statusCode != http.StatusCreated {
errMsg = fmt.Sprintf("HTTP %d", statusCode)
logger.Default().Error("[IMAGING STUDY WORKER] ❌ Respons Error dari API Part 3",
logger.String("source_id", sourceID),
logger.Int("status_code", statusCode),
logger.String("response", truncate(string(respBody), 200)),
)
// Retry on server-side errors
if statusCode >= 500 || statusCode == http.StatusTooManyRequests {
return "", respBody, "FAILED", errMsg, true
}
// Do not retry on client-side errors (4xx)
return "", respBody, "FAILED", errMsg, false
}
// --- 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
}
}
logger.Default().Info("[IMAGING STUDY WORKER] ✅ Sukses mengirim data Imaging Study",
logger.String("source_id", sourceID),
logger.String("fhir_id", fhirID),
)
return fhirID, respBody, "SUCCESS", "", false
}
func (w *worker) updateTracker(lastCreatedAt *time.Time, currentCreatedAt time.Time) {
*lastCreatedAt = currentCreatedAt
os.WriteFile(trackerFile, []byte(currentCreatedAt.Format(time.RFC3339Nano)), 0644)
}
func (w *worker) readLastCreatedAt() time.Time {
data, err := os.ReadFile(trackerFile)
if err != nil {
// Jika file tidak ada, mulai dari awal (zero time)
return time.Time{}
}
timestampStr := strings.TrimSpace(string(data))
t, err := time.Parse(time.RFC3339Nano, timestampStr)
if err != nil {
logger.Default().Warn("[IMAGING STUDY WORKER] Gagal parsing timestamp dari tracker file, memulai dari awal.", logger.String("content", timestampStr), logger.ErrorField(err))
return time.Time{}
}
return t
}
func truncate(s string, max int) string {
if len(s) > max {
return s[:max] + "..."
}
return s
}
internal/worker/satusehat/medication/dto.go
+5 -12
View File
@@ -4,22 +4,15 @@ import (
"database/sql"
)
// MedicationDB merepresentasikan tabel master/transaksi obat (KFA)
// TODO: Sesuaikan field dan tag `db` dengan tabel SIMRS Anda
// MedicationDB merepresentasikan tabel transaksi obat dari t_permintaan_apotek_rajal
type MedicationDB struct {
IdxObat int64 `db:"idxobat"`
KodeKFA sql.NullString `db:"kode_kfa"`
NamaKFA sql.NullString `db:"nama_kfa"`
KodeBentuk sql.NullString `db:"kode_bentuk"`
NamaBentuk sql.NullString `db:"nama_bentuk"`
ManufacturerID sql.NullString `db:"manufacturer_id"`
NoBatch sql.NullString `db:"no_batch"`
TglKadaluarsa sql.NullTime `db:"tgl_kadaluarsa"`
IdxPesanObat int64 `db:"idxpesanobat"`
KfaCode sql.NullString `db:"kode_kfa"`
}
type MedicationSyncLog struct {
IdxObat int64 `db:"idxobat"`
MedicationID string `db:"medication_id"`
IdxPesanObat int64 `db:"idxpesanobat"`
SatuSehatID string `db:"satusehat_id"`
RequestPayload string `db:"request_payload"`
ResponsePayload string `db:"response_payload"`
Status string `db:"status"`
internal/worker/satusehat/medication/last_medication_id.txt
+1
View File
@@ -0,0 +1 @@
3384484
internal/worker/satusehat/medication/mapper.go
+27 -16
View File
@@ -2,31 +2,42 @@ package medication
import (
"encoding/json"
"time"
"strconv"
)
// TODO: Sesuaikan dengan payload internal API Medication
func MapToInternalAPI(dbData *MedicationDB, orgID string) map[string]interface{} {
payload := map[string]interface{}{
"status_code": "active",
"kfa_code": dbData.KodeKFA.String,
"kfa_display": dbData.NamaKFA.String,
"form_code": dbData.KodeBentuk.String,
"form_display": dbData.NamaBentuk.String,
"manufacturer_id": orgID, // Bisa disesuaikan menjadi dbData.ManufacturerID.String jika beda pabrik
"batch_number": dbData.NoBatch.String,
// MapToInternalAPI memetakan data DB ke payload API Internal Medication
func MapToInternalAPI(dbData *MedicationDB, orgID string, kfaDisplay, formCode, formDisplay string) map[string]interface{} {
payload := make(map[string]interface{})
payload["status_code"] = "active"
if dbData.IdxPesanObat != 0 {
payload["medication_id"] = strconv.FormatInt(dbData.IdxPesanObat, 10)
}
if dbData.TglKadaluarsa.Valid {
payload["expiration_date"] = dbData.TglKadaluarsa.Time.Format(time.RFC3339)
if dbData.KfaCode.Valid && dbData.KfaCode.String != "" {
payload["kfa_code"] = dbData.KfaCode.String
}
if orgID != "" {
payload["manufacturer_id"] = orgID
}
if kfaDisplay != "" {
payload["kfa_display"] = kfaDisplay
}
if formCode != "" {
payload["form_code"] = formCode
}
if formDisplay != "" {
payload["form_display"] = formDisplay
}
return payload
}
func MapToSyncLog(idx int64, fhirID string, reqPayload interface{}, respBody []byte, status string, errMsg string) MedicationSyncLog {
reqBytes, _ := json.MarshalIndent(reqPayload, "", " ")
reqBytes, _ := json.Marshal(reqPayload)
return MedicationSyncLog{
IdxObat: idx,
MedicationID: fhirID,
IdxPesanObat: idx,
SatuSehatID: fhirID,
RequestPayload: string(reqBytes),
ResponsePayload: string(respBody),
Status: status,
internal/worker/satusehat/medication/medication_mapping.txt
+121
View File
@@ -0,0 +1,121 @@
3384065|7d731949-ac7c-409c-b997-5ae48898087b
3384067|75922941-7c0d-40c9-b09e-07155d05f19e
3384068|3a1d2828-a67b-4b0b-8a7f-d501bff494eb
3384074|3269526f-6c8a-432e-a0c2-3cb0fb7d58f2
3384075|bf75589f-689a-4f6e-990d-b64bf7d19655
3384084|9b5e40a0-a908-4fe9-98fb-faeb23380f4d
3384088|6ab7a573-7327-4349-8e0a-e6adeece3e21
3384089|6ef8c7d0-b9b6-4473-b803-435e5d2da43b
3384090|740eeed2-7815-477d-8c1b-653d7b0a6dc8
3384093|e9916e04-ba83-44be-a59a-f9477cc9fba8
3384094|f8106ab5-048b-4b08-b647-b9263fcbeed1
3384095|4078f9b7-d895-4656-a215-30aeb291d0ec
3384096|2d9e19e5-1547-4e34-b05c-19bcea5dd099
3384097|a99b8793-1360-485c-8384-8c672f38ea14
3384099|e293a828-46a1-4829-b95e-b7d36bb17460
3384107|fb41a9e5-57ce-4259-ab22-c6d428d3cc5e
3384113|0640a712-16fc-4b68-832e-0b041c7fb0ee
3384138|67db8861-92dc-41b5-a2bc-14082cc04e76
3384146|581e3187-ac4a-413c-ae85-39560c300f4f
3384147|a3109162-df95-4581-9852-0cc0cda1372a
3384154|e46eda4e-ec0b-46f2-bc6b-3cca56e5c889
3384170|2634b8d9-7ab8-4556-87d9-118d82d281e7
3384175|3842777d-ceab-43ab-9a2e-08ff0397d7ca
3384177|18736b7f-fcb1-4e92-ae78-7073526ee605
3384180|b70c06f2-f06c-43f3-9ae9-ff1f5d7cec92
3384181|8c146256-a004-4d37-97ff-5d0efc07b28d
3384184|a0b4addb-cf57-4cf7-a8ce-7fa3c1733614
3384186|e5d36d3d-01f3-446b-b558-5c86712e1873
3384187|4a6549a1-e4c2-4165-8e5a-7a9cd068a522
3384190|49e2481c-d370-4367-b070-d2c652b1c325
3384191|7c1a7315-d8f4-4e55-8890-f0772e99115c
3384192|16ea6748-d1a2-4b46-acdf-3d0e84e4ab6f
3384193|ddf97d42-c953-45d9-b763-e7072665ac84
3384194|3da3d066-ca09-4fc6-8e83-5bde3c8a1d76
3384195|2bfef15f-2c1a-4c0a-a598-34f229139556
3384196|e9b94407-a524-4d0f-8efb-5e13080d4e2e
3384197|c6799b62-8777-488f-8931-dc1c820adebc
3384198|d1ab6618-d278-4c28-b9fd-129ad4345c07
3384200|a876cd27-65ca-4e78-82e3-7b47455119dc
3384208|f6b0c2b1-ac58-4639-83bd-633cc30ffa74
3384215|7d2ecbe4-5553-404b-a732-7d15b7cf9355
3384230|f8d15006-08aa-4fcd-9f25-8921cef87860
3384239|f8f506fb-0c8c-493e-8e2a-8401a064458f
3384253|993f2410-5a24-4bdc-aa77-0c740630d043
3384258|5153d58a-a487-4201-b198-b3db83d47568
3384260|28a28511-a1e1-4c75-aba1-97b67ab0bb39
3384261|f29fa145-5199-408b-9512-8184d5dbfa1e
3384263|2a621669-e2d0-444c-bd2c-4a77a65be14b
3384265|9d61147e-0edf-43ca-b93e-f9b6e546e173
3384277|84ce68b9-cbd9-4cdb-bc48-81c3068f2d9b
3384289|1d946f71-a050-444d-b97b-ffac87af3bec
3384291|629ae24f-f117-4adb-8122-5033739f9a77
3384298|e6b9a50c-1d23-4fdb-ada0-c79e605cc545
3384299|a14e9b01-5043-45a0-8d8a-bda67707d30c
3384300|add6ad7d-e32e-4840-a565-49c9bba605b9
3384304|50d251d8-7208-419d-a8bc-84ad979f17d8
3384305|e8001b64-b90f-41cc-95e2-5efda731c504
3384308|4d33f376-1437-4874-9573-924d1c02fefa
3384310|0a19db00-a042-4acb-8a8f-35485e1b2aad
3384311|38621f80-ae46-46ac-8f62-4bc7b66c2776
3384314|5d978686-2e70-4ed3-825a-e7ed18f400e8
3384315|7d5ef2c5-5079-43e5-b0ce-5284f51006fe
3384316|d789fe64-cf62-41ac-bbb6-227c23d90650
3384317|a1b6f81e-a194-4b23-bd84-9cf249d8a40d
3384321|f56e3cce-7bfc-479a-8bb0-c2a5e35ff956
3384325|919c9238-965c-40aa-81ae-7374d178475e
3384330|14b64e7b-fdd4-4e95-9152-54ba88fb1f02
3384331|a73228ae-6021-41ca-9edd-61fdd21647d4
3384333|f9d36faa-e97f-4cb9-a189-1c5c45598ea1
3384334|fa39f375-2a60-4782-bec4-1919c178795e
3384335|10691f24-78ea-45bf-9419-0300e2a24320
3384336|528649f6-0654-40f9-b310-806a35e0f3bf
3384340|e738cad7-8ff5-4d96-b01a-01117847bfd8
3384343|7b628d79-28fa-4ee1-ad1b-94760b6be2f0
3384349|2abd44ab-3465-4b77-aa8f-6bf0f3459fb4
3384354|a361bb0e-5764-4ad6-b0a1-6ea402a8170b
3384355|50c35eca-5cdb-4f26-97d8-0f1a16f812f4
3384358|f33107bf-e1d1-49f9-9d49-58e0ae4919af
3384361|413056a4-450e-4eb7-942e-752aeea608fd
3384363|a6dbe25e-0c99-41ab-85d4-468c10adff68
3384375|ec7b064e-5f3c-4d90-9776-0ec61d34833c
3384379|0495eb25-d3dd-4c31-b9bc-dbfa770a2152
3384380|b4abceee-6535-4839-b463-76df18f51a22
3384383|10989bab-0c22-4231-a676-8e41a0f60021
3384384|db35de2c-1777-4a30-b859-ffa88a01723a
3384387|7fc38f6a-6579-459b-bfa5-6eed9a24f274
3384388|b25684e7-e8d4-4470-b26a-ee1e54241af8
3384399|419a8949-3489-439c-ae2a-1f6ec6fa7d0f
3384400|ebb3d1e4-6d30-4822-908d-9e768c62ac80
3384405|80593d6a-749e-4602-b8a6-688d039e1f1f
3384408|abb1b18a-6915-4999-b016-68366d96be2d
3384416|e3921a30-dc8f-4085-a5be-ce24c077c768
3384417|7aeaec56-fd32-4d19-b561-5820c86eb803
3384419|93d437c5-8118-4e3b-ae68-5c289fb8c004
3384420|921bcc71-d632-4156-8a6f-7d79d887520a
3384422|056092a6-bec8-46b4-85db-2e5f9da4b628
3384423|d8026a01-5b2f-423c-9042-577cb3221d8c
3384425|01adcc5c-349c-4381-9125-06816fa19034
3384426|0eb873f5-054d-4861-9b57-ba4cd45c33a4
3384427|effbe803-2e15-4624-a423-595bdf670556
3384428|ac8a89d3-3c82-468a-8b99-5b4975d30b46
3384439|1a14cfa3-bfd6-49e3-9a87-3c496e07b18b
3384442|6dd478f9-9d3e-4478-af15-a9aed00cee8a
3384444|f75da4cd-1847-4f95-a1d2-e2b3a58f71e9
3384445|50800e76-b7b1-4d62-afe7-dada10212ace
3384454|2c0291d0-d609-419f-bb6a-1adafe71d685
3384456|88e3ea14-2027-482a-b928-8ecbb773bfff
3384459|41ab8bf3-bc2e-4d52-8933-b269669370fb
3384460|f82235cf-69ad-45a8-9318-5053cce14d00
3384461|632da84c-c45c-46b1-854c-4a34cd60ddd2
3384467|96beb57a-67b2-45a5-abc2-f5550edb0426
3384468|d43c2f05-33e9-442e-9b48-7d0f5f17454a
3384469|5487d5c6-4213-4674-90eb-d0410dfba7d2
3384470|50e2f103-d491-4644-8316-715711a3aa99
3384471|ca217f91-315a-454d-aa49-85454e30e66e
3384472|df86b609-92a2-4ecb-bb49-8780f5f3061c
3384473|4141ed35-9cb1-41c7-ace3-abd137c2fcd6
3384475|18ae8627-c025-4f4a-8522-2dcfa406e336
3384482|30a18fd7-4e57-41cf-ae5a-83bbda050b9d
3384483|0cfbf15e-0916-420c-8a14-db322a9f8355
3384484|1bf3dd22-9ba7-4786-b5f1-67d40f3b2fb2
internal/worker/satusehat/medication/repository.go
+28 -13
View File
@@ -8,7 +8,7 @@ import (
type Repository interface {
GetNextMedication(ctx context.Context, lastID int64) (*MedicationDB, error)
SaveSatuSehatID(ctx context.Context, idx int64, fhirID string) error
SaveSatuSehatID(ctx context.Context, idxPesanObat int64, fhirID string) error
SaveSyncLog(ctx context.Context, logData MedicationSyncLog) error
}
@@ -27,20 +27,29 @@ func (r *repository) GetNextMedication(ctx context.Context, lastID int64) (*Medi
return nil, err
}
qb := query.NewSQLQueryBuilder(query.DBTypePostgreSQL).SetSecurityOptions(false, 0)
qb := query.NewSQLQueryBuilder(query.DBTypePostgreSQL).
SetSecurityOptions(false, 0).
SetQueryLogging(false).
SetAllowedColumns([]string{"idxpesanobat", "kode_kfa", "tgl_pesan"})
// startDate, _ := time.Parse("2006-01-02", "2026-01-02")
// TODO: Sesuaikan nama tabel referensi/master obat Anda
qSimrs := query.DynamicQuery{
From: "public.m_obat",
From: "public.t_permintaan_apotek_rajal",
Fields: []query.SelectField{
{Expression: "idxpesanobat"},
{Expression: "kode_kfa"},
},
Filters: []query.FilterGroup{
{
Filters: []query.DynamicFilter{
query.CreateFilter("idxobat", query.OpGreaterThan, lastID),
query.CreateFilter("idxpesanobat", query.OpGreaterThanEqual, lastID+1),
// query.CreateFilter("tgl_pesan", query.OpGreaterThanEqual, startDate),
},
},
},
Sort: []query.SortField{
query.CreateAscSort("idxobat"),
query.CreateAscSort("idxpesanobat"),
},
Limit: 1,
}
@@ -52,13 +61,16 @@ func (r *repository) GetNextMedication(ctx context.Context, lastID int64) (*Medi
return &med, nil
}
func (r *repository) SaveSatuSehatID(ctx context.Context, idx int64, fhirID string) error {
func (r *repository) SaveSatuSehatID(ctx context.Context, idxPesanObat int64, fhirID string) error {
simrsDB, err := r.dbManager.GetSQLXDB("simrs")
if err != nil {
return err
}
qb := query.NewSQLQueryBuilder(query.DBTypePostgreSQL).SetSecurityOptions(false, 0)
_, err = qb.ExecuteUpdate(ctx, simrsDB, "public.m_obat", query.UpdateData{Columns: []string{"status_bridging"}, Values: []interface{}{fhirID}}, []query.FilterGroup{query.CreateAndFilterGroup([]query.DynamicFilter{query.CreateEqualFilter("idxobat", idx)})})
qb := query.NewSQLQueryBuilder(query.DBTypePostgreSQL).
SetSecurityOptions(false, 0).
SetQueryLogging(false).
SetAllowedColumns([]string{"id_satusehat", "idxpesanobat"})
_, err = qb.ExecuteUpdate(ctx, simrsDB, "public.t_permintaan_apotek_rajal", query.UpdateData{Columns: []string{"id_satusehat"}, Values: []interface{}{fhirID}}, []query.FilterGroup{query.CreateAndFilterGroup([]query.DynamicFilter{query.CreateEqualFilter("idxpesanobat", idxPesanObat)})})
return err
}
@@ -67,11 +79,14 @@ func (r *repository) SaveSyncLog(ctx context.Context, logData MedicationSyncLog)
if err != nil {
return err
}
qb := query.NewSQLQueryBuilder(query.DBTypePostgreSQL).SetSecurityOptions(false, 0)
qb := query.NewSQLQueryBuilder(query.DBTypePostgreSQL).
SetSecurityOptions(false, 0).
SetQueryLogging(false).
SetAllowedColumns([]string{"idxpesanobat", "satusehat_id", "request_payload", "response_payload", "status", "error_message"})
insertData := query.InsertData{
Columns: []string{"idxobat", "medication_id", "request_payload", "response_payload", "status", "error_message"},
Values: []interface{}{logData.IdxObat, logData.MedicationID, logData.RequestPayload, logData.ResponsePayload, logData.Status, logData.ErrorMessage},
Columns: []string{"idxpesanobat", "satusehat_id", "request_payload", "response_payload", "status", "error_message"},
Values: []interface{}{logData.IdxPesanObat, logData.SatuSehatID, logData.RequestPayload, logData.ResponsePayload, logData.Status, logData.ErrorMessage},
}
_, err = qb.ExecuteUpsert(ctx, simrsDB, "public.log_satusehat_medication", insertData, []string{"idxobat"}, []string{"medication_id", "request_payload", "response_payload", "status", "error_message"})
_, err = qb.ExecuteUpsert(ctx, simrsDB, "public.log_satusehat_medication", insertData, []string{"idxpesanobat"}, []string{"satusehat_id", "request_payload", "response_payload", "status", "error_message"})
return err
}
internal/worker/satusehat/medication/service.go
+263 -33
View File
@@ -2,92 +2,322 @@ package medication
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"log"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"time"
"service/internal/infrastructure/database"
extapi "service/internal/worker/interface"
"service/pkg/logger"
)
const trackerFile = "last_migrated_medication_id.txt"
const trackerFile = "internal/worker/satusehat/medication/last_medication_id.txt"
const mappingFile = "internal/worker/satusehat/medication/medication_mapping.txt"
type Config struct {
DBManager database.Service
InternalBaseURL string
InternalToken string
OrganizationID string
}
type TokenManager interface {
GetAccessToken() string
ForceRefreshAndGetToken(ctx context.Context) (string, error)
}
type WorkerService interface {
Run(ctx context.Context)
}
type worker struct {
cfg Config
repo Repository
apiClient extapi.Client
cfg Config
repo Repository
apiClient extapi.Client
tokenManager TokenManager
}
func NewWorker(cfg Config) WorkerService {
return &worker{cfg: cfg, repo: NewRepository(cfg.DBManager), apiClient: extapi.NewClient(15 * time.Second)}
func NewWorker(cfg Config, tokenManager TokenManager) WorkerService {
return &worker{
cfg: cfg,
repo: NewRepository(cfg.DBManager),
apiClient: extapi.NewClient(60 * time.Second),
tokenManager: tokenManager,
}
}
func (w *worker) Run(ctx context.Context) {
log.Println("[MEDICATION WORKER] Memulai proses sinkronisasi...")
logger.Default().Info("[MEDICATION WORKER] Memulai proses sinkronisasi API...")
w.ensureTrackerFileExists()
lastID := w.readLastID()
// if lastID == 0 {
// lastID = 3384064 // Dikurangi 1 agar ID 3384065 ikut terambil
// w.updateTracker(&lastID, lastID)
// }
logger.Default().Info("[MEDICATION WORKER] Melanjutkan proses dari", logger.Int64("last_id", lastID))
for {
select {
case <-ctx.Done():
log.Println("[MEDICATION WORKER] Proses dihentikan.")
logger.Default().Warn("[MEDICATION WORKER] Proses dihentikan oleh sistem (Graceful Shutdown)")
return
default:
}
med, err := w.repo.GetNextMedication(ctx, lastID)
if err != nil {
if err == sql.ErrNoRows {
time.Sleep(5 * time.Second) // Tunggu data baru
continue
}
logger.Default().Error("[MEDICATION WORKER] Gagal query ke database", logger.ErrorField(err))
time.Sleep(5 * time.Second)
continue
}
internalPayload := MapToInternalAPI(med, w.cfg.OrganizationID)
reqURL := fmt.Sprintf("%s/satusehat/medication", strings.TrimRight(w.cfg.InternalBaseURL, "/"))
respBody, statusCode, err := w.apiClient.PostJSON(ctx, reqURL, w.cfg.InternalToken, internalPayload)
logger.Default().Info("[MEDICATION WORKER] 🔄 Memproses Dokumen",
logger.Int64("idxpesanobat", med.IdxPesanObat),
logger.String("kode_kfa", med.KfaCode.String),
)
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)
} 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)
}
}
if !med.KfaCode.Valid || med.KfaCode.String == "" {
logger.Default().Warn("[MEDICATION WORKER] ⏭️ Skip Dokumen: KFA Code kosong",
logger.Int64("idxpesanobat", med.IdxPesanObat),
)
w.updateTracker(&lastID, med.IdxPesanObat)
continue
}
kfaDisplay, formCode, formDisplay := w.fetchKFADetails(ctx, med.KfaCode.String)
internalPayload := MapToInternalAPI(med, w.cfg.OrganizationID, kfaDisplay, formCode, formDisplay)
fhirID, respBody, status, errMsg, shouldRetry := w.sendToInternalAPI(ctx, internalPayload, med.IdxPesanObat)
// Menyimpan payload request dan response ke dalam log
logger.Default().Info("[MEDICATION WORKER] Detail Transaksi",
logger.Int64("idxpesanobat", med.IdxPesanObat),
logger.Any("request_payload", internalPayload),
logger.String("response_payload", string(respBody)),
logger.String("status", status),
)
syncLog := MapToSyncLog(med.IdxPesanObat, fhirID, internalPayload, respBody, status, errMsg)
if errLog := w.repo.SaveSyncLog(ctx, syncLog); errLog != nil {
logger.Default().Error("[MEDICATION WORKER] Gagal menyimpan histori log ke DB", logger.ErrorField(errLog))
}
_ = w.repo.SaveSyncLog(ctx, MapToSyncLog(med.IdxObat, fhirID, internalPayload, respBody, status, errMsg))
if status == "SUCCESS" && fhirID != "" {
_ = w.repo.SaveSatuSehatID(ctx, med.IdxObat, fhirID)
// if errUpdate := w.repo.SaveSatuSehatID(ctx, med.IdxPesanObat, fhirID); errUpdate != nil {
// logger.Default().Error("[MEDICATION WORKER] Gagal update FHIR ID ke transaksi utama", logger.ErrorField(errUpdate))
// }
w.saveMapping(med.IdxPesanObat, fhirID)
}
lastID = med.IdxObat
os.WriteFile(trackerFile, []byte(strconv.FormatInt(lastID, 10)), 0644)
time.Sleep(500 * time.Millisecond)
// SELALU update tracker agar lastID berjalan maju terus dan tidak mengulang data yang sama
w.updateTracker(&lastID, med.IdxPesanObat)
if shouldRetry {
logger.Default().Warn("[MEDICATION WORKER] ⚠️ Gagal memproses data (melewati dokumen)", logger.Int64("idxpesanobat", med.IdxPesanObat), logger.String("reason", errMsg))
time.Sleep(5 * time.Second)
} else {
time.Sleep(2 * time.Second)
}
}
}
func (w *worker) sendToInternalAPI(ctx context.Context, payload map[string]interface{}, idxPesanObat int64) (fhirID string, respBody []byte, status string, errMsg string, shouldRetry bool) {
reqURL := fmt.Sprintf("%s/satusehat/medication", strings.TrimRight(w.cfg.InternalBaseURL, "/"))
respBody, statusCode, err := w.apiClient.PostJSON(ctx, reqURL, w.tokenManager.GetAccessToken(), payload)
if err != nil {
logger.Default().Error("[MEDICATION WORKER] ❌ Gagal Koneksi ke API Part 3",
logger.Int64("idxpesanobat", idxPesanObat),
logger.Any("payload_sent", payload),
logger.ErrorField(err),
)
return "", nil, "FAILED", err.Error(), true
}
if statusCode == http.StatusUnauthorized {
logger.Default().Warn("[MEDICATION WORKER] Menerima status 401 Unauthorized, mencoba refresh token...", logger.Int64("idxpesanobat", idxPesanObat))
newToken, refreshErr := w.tokenManager.ForceRefreshAndGetToken(ctx)
if refreshErr != nil {
logger.Default().Error("[MEDICATION WORKER] Gagal refresh token, akan mencoba lagi pada iterasi berikutnya.", logger.Int64("idxpesanobat", idxPesanObat), logger.ErrorField(refreshErr))
return "", respBody, "FAILED", refreshErr.Error(), true
}
logger.Default().Info("[MEDICATION WORKER] Token berhasil di-refresh, mencoba ulang request...", logger.Int64("idxpesanobat", idxPesanObat))
respBody, statusCode, err = w.apiClient.PostJSON(ctx, reqURL, newToken, payload)
if err != nil {
logger.Default().Error("[MEDICATION WORKER] ❌ Gagal Koneksi ke API Part 3 (setelah retry)",
logger.Int64("idxpesanobat", idxPesanObat),
logger.ErrorField(err),
)
return "", nil, "FAILED", err.Error(), true
}
}
if statusCode != http.StatusOK && statusCode != http.StatusCreated {
errMsg = fmt.Sprintf("HTTP %d", statusCode)
logger.Default().Error("[MEDICATION WORKER] ❌ Respons Error dari API Part 3",
logger.Int64("idxpesanobat", idxPesanObat),
logger.Int("status_code", statusCode),
logger.String("response", truncate(string(respBody), 200)),
)
if statusCode >= 500 || statusCode == http.StatusTooManyRequests {
return "", respBody, "FAILED", errMsg, true
}
return "", respBody, "FAILED", errMsg, false
}
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
}
}
logger.Default().Info("[MEDICATION WORKER] ✅ Sukses mengirim data Medication",
logger.Int64("idxpesanobat", idxPesanObat),
logger.String("fhir_id", fhirID),
)
return fhirID, respBody, "SUCCESS", "", false
}
func (w *worker) fetchKFADetails(ctx context.Context, kfaCode string) (kfaDisplay, formCode, formDisplay string) {
reqURL := fmt.Sprintf("%s/satusehat/reference/kfa/products/%s", strings.TrimRight(w.cfg.InternalBaseURL, "/"), kfaCode)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
if err != nil {
logger.Default().Error("[MEDICATION WORKER] Gagal membuat request KFA", logger.ErrorField(err))
return "", "", ""
}
req.Header.Set("Authorization", "Bearer "+w.tokenManager.GetAccessToken())
req.Header.Set("Accept", "application/json")
client := &http.Client{Timeout: 10 * time.Second}
resp, err := client.Do(req)
if err != nil {
logger.Default().Error("[MEDICATION WORKER] Gagal request data ke Part 3 KFA", logger.ErrorField(err))
return "", "", ""
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
logger.Default().Warn("[MEDICATION WORKER] KFA tidak ditemukan / Error", logger.Int("status", resp.StatusCode), logger.String("kfa_code", kfaCode))
return "", "", ""
}
var result map[string]interface{}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
logger.Default().Error("[MEDICATION WORKER] Gagal decode response KFA", logger.ErrorField(err))
return "", "", ""
}
// Parsing respons bersarang dari KFA API (data -> result -> dosage_form)
if dataMap, ok := result["data"].(map[string]interface{}); ok {
if resultMap, ok := dataMap["result"].(map[string]interface{}); ok {
// Ekstrak nama obat
if val, ok := resultMap["name"].(string); ok {
kfaDisplay = val
}
// Ekstrak bentuk obat (dosage_form)
if dosageForm, ok := resultMap["dosage_form"].(map[string]interface{}); ok {
if val, ok := dosageForm["code"].(string); ok {
formCode = val
}
if val, ok := dosageForm["name"].(string); ok {
formDisplay = val
}
}
}
}
// Fallback jika format respons langsung flat
if kfaDisplay == "" {
if val, ok := result["name"].(string); ok {
kfaDisplay = val
}
}
if formCode == "" {
if val, ok := result["form_code"].(string); ok {
formCode = val
}
}
if formDisplay == "" {
if val, ok := result["form_display"].(string); ok {
formDisplay = val
}
}
return kfaDisplay, formCode, formDisplay
}
func (w *worker) updateTracker(lastID *int64, currentID int64) {
*lastID = currentID
os.MkdirAll("internal/worker/satusehat/medication", 0755)
os.WriteFile(trackerFile, []byte(strconv.FormatInt(currentID, 10)), 0644)
}
func (w *worker) readLastID() int64 {
data, _ := os.ReadFile(trackerFile)
id, _ := strconv.ParseInt(strings.TrimSpace(string(data)), 10, 64)
data, err := os.ReadFile(trackerFile)
if err != nil {
return 0
}
idStr := strings.TrimSpace(string(data))
id, err := strconv.ParseInt(idStr, 10, 64)
if err != nil {
logger.Default().Warn("[MEDICATION WORKER] Gagal parsing ID dari tracker file, memulai dari awal.", logger.String("content", idStr), logger.ErrorField(err))
return 0
}
return id
}
func (w *worker) ensureTrackerFileExists() {
if _, err := os.Stat(trackerFile); os.IsNotExist(err) {
dir := filepath.Dir(trackerFile)
if err := os.MkdirAll(dir, 0755); err != nil {
logger.Default().Error("[MEDICATION WORKER] Gagal membuat direktori tracker", logger.ErrorField(err))
}
if err := os.WriteFile(trackerFile, []byte("0"), 0644); err != nil {
logger.Default().Error("[MEDICATION WORKER] Gagal membuat file tracker default", logger.ErrorField(err))
}
}
}
func (w *worker) saveMapping(idxPesanObat int64, fhirID string) {
os.MkdirAll("internal/worker/satusehat/medication", 0755)
f, err := os.OpenFile(mappingFile, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
logger.Default().Error("[MEDICATION WORKER] Gagal membuka file mapping", logger.ErrorField(err))
return
}
defer f.Close()
line := fmt.Sprintf("%d|%s\n", idxPesanObat, fhirID)
if _, err := f.WriteString(line); err != nil {
logger.Default().Error("[MEDICATION WORKER] Gagal menulis ke file mapping", logger.ErrorField(err))
}
}
func truncate(s string, max int) string {
if len(s) > max {
return s[:max] + "..."
}
return s
}
internal/worker/satusehat/servicerequest/dto.go
+37
View File
@@ -2,6 +2,7 @@ package servicerequest
import (
"database/sql"
"time"
)
// ServiceRequestDB merepresentasikan tabel order lab/radiologi
@@ -22,3 +23,39 @@ type ServiceRequestSyncLog struct {
Status string `db:"status"`
ErrorMessage string `db:"error_message"`
}
type ServiceRequestRadDB struct {
ID string `db:"id"`
JsonData sql.NullString `db:"Json_data"`
PatientIHS sql.NullString `db:"Nomor_satusehat_pasien"`
PatientName sql.NullString `db:"Nama_lengkap"`
EncounterID sql.NullString `db:"IDXDAFTAR_satusehat"`
CreatedAt time.Time `db:"Created_at"`
RequesterID sql.NullString `db:"requester_id"`
PerformerID sql.NullString `db:"performer_id"`
}
type ServiceRequestJson struct {
Status string `json:"status"`
Intent string `json:"intent"`
Category []struct {
Coding []struct {
Code string `json:"code"`
Display string `json:"display"`
} `json:"coding"`
} `json:"category"`
Code struct {
Coding []struct {
Code string `json:"code"`
Display string `json:"display"`
} `json:"coding"`
} `json:"code"`
Requester struct {
Reference string `json:"reference"`
} `json:"requester"`
Performer []struct {
Reference string `json:"reference"`
Display string `json:"display"`
} `json:"performer"`
AuthoredOn string `json:"authoredOn"`
}
internal/worker/satusehat/servicerequest/last_servicerequest_rad_time.txt
+1
View File
@@ -0,0 +1 @@
2026-04-24T13:24:32Z
internal/worker/satusehat/servicerequest/mapper.go
+56
View File
@@ -3,6 +3,7 @@ package servicerequest
import (
"encoding/json"
"fmt"
"strings"
"time"
)
@@ -33,3 +34,58 @@ func MapToSyncLog(idx int64, fhirID string, reqPayload interface{}, respBody []b
ErrorMessage: errMsg,
}
}
func MapRadiologyToInternalAPI(dbData *ServiceRequestRadDB) map[string]interface{} {
var parsedData ServiceRequestJson
_ = json.Unmarshal([]byte(dbData.JsonData.String), &parsedData)
categoryCode := ""
categoryDisplay := ""
if len(parsedData.Category) > 0 && len(parsedData.Category[0].Coding) > 0 {
categoryCode = parsedData.Category[0].Coding[0].Code
categoryDisplay = parsedData.Category[0].Coding[0].Display
}
codeVal := ""
codeDisplay := ""
if len(parsedData.Code.Coding) > 0 {
codeVal = parsedData.Code.Coding[0].Code
codeDisplay = parsedData.Code.Coding[0].Display
}
requesterID := dbData.RequesterID.String
// Ambil dari DB terlebih dahulu, fallback ke JSON data jika kosong
performerID := dbData.PerformerID.String
if performerID == "" && len(parsedData.Performer) > 0 {
performerID = parsedData.Performer[0].Reference
if parts := strings.Split(performerID, "/"); len(parts) > 1 {
performerID = parts[len(parts)-1]
}
}
status := parsedData.Status
if status == "" {
status = "active" // Fallback jika kosong
}
intent := parsedData.Intent
if intent == "" {
intent = "order" // Fallback jika kosong
}
return map[string]interface{}{
"patient_id": dbData.PatientIHS.String,
"patient_name": dbData.PatientName.String,
"encounter_id": dbData.EncounterID.String,
"status": status,
"intent": intent,
"category_code": categoryCode,
"category_display": categoryDisplay,
"code": codeVal,
"display": codeDisplay,
"requester_id": requesterID,
"performer_id": performerID,
"authored_on": parsedData.AuthoredOn,
}
}
internal/worker/satusehat/servicerequest/repository.go
+85
View File
@@ -4,12 +4,16 @@ import (
"context"
"service/internal/infrastructure/database"
"service/pkg/utils/query"
"time"
)
type Repository interface {
GetNextServiceRequest(ctx context.Context, lastID int64) (*ServiceRequestDB, error)
SaveSatuSehatID(ctx context.Context, idx int64, fhirID string) error
SaveSyncLog(ctx context.Context, logData ServiceRequestSyncLog) error
GetNextRadiologyServiceRequest(ctx context.Context, lastCreatedAt time.Time) (*ServiceRequestRadDB, error)
UpdateRadiologyServiceRequestStatus(ctx context.Context, id string, fhirID string, encounterID string, status bool) error
}
type repository struct {
@@ -75,3 +79,84 @@ func (r *repository) SaveSyncLog(ctx context.Context, logData ServiceRequestSync
_, err = qb.ExecuteUpsert(ctx, simrsDB, "public.log_satusehat_servicerequest", insertData, []string{"idxorder"}, []string{"servicerequest_id", "request_payload", "response_payload", "status", "error_message"})
return err
}
func (r *repository) GetNextRadiologyServiceRequest(ctx context.Context, lastCreatedAt time.Time) (*ServiceRequestRadDB, error) {
var data ServiceRequestRadDB
satudataDB, err := r.dbManager.GetSQLXDB("satudata")
if err != nil {
return nil, err
}
qb := query.NewSQLQueryBuilder(query.DBTypePostgreSQL).SetSecurityOptions(false, 0)
qRadiology := query.DynamicQuery{
From: "public.data_service_requests",
Aliases: "dsr",
Fields: []query.SelectField{
{Expression: "dsr.id"},
{Expression: `dsr."Json_data"`, Alias: "Json_data"},
{Expression: `dkp."Nomor_satusehat_pasien"`, Alias: "Nomor_satusehat_pasien"},
{Expression: `dkp."Nama_lengkap"`, Alias: "Nama_lengkap"},
{Expression: `dkp."IDXDAFTAR_satusehat"`, Alias: "IDXDAFTAR_satusehat"},
{Expression: `dsr."Created_at"`, Alias: "Created_at"},
{Expression: `dp."Kode_satusehat"`, Alias: "requester_id"},
},
Joins: []query.Join{
{
Type: "INNER",
Table: "public.data_kunjungan_pasien",
Alias: "dkp",
OnConditions: query.CreateAndFilterGroup([]query.DynamicFilter{
query.CreateFilter(`dsr."No_register"`, query.OpEqual, `dkp."NOMR"`),
query.CreateFilter(`dsr."Tanggal_order"`, query.OpEqual, `dkp."Tanggal_masuk"`),
}),
},
{
Type: "LEFT",
Table: "public.data_pegawai",
Alias: "dp",
OnConditions: query.CreateAndFilterGroup([]query.DynamicFilter{
query.CreateFilter(`dkp."DPJP"`, query.OpEqual, `dp."id"`),
}),
},
},
Filters: []query.FilterGroup{
{Filters: []query.DynamicFilter{
query.CreateFilter(`dsr."Send_status"`, query.OpEqual, "false"),
query.CreateFilter(`dsr."Created_at"`, query.OpGreaterThan, lastCreatedAt),
query.CreateFilter(`dsr."Order_from"`, query.OpEqual, "radiologi"),
}},
},
Sort: []query.SortField{query.CreateAscSort(`dsr."Created_at"`)},
Limit: 1,
}
err = qb.ExecuteQueryRow(ctx, satudataDB, qRadiology, &data)
return &data, err
}
func (r *repository) UpdateRadiologyServiceRequestStatus(ctx context.Context, id string, fhirID string, encounterID string, status bool) error {
satudataDB, err := r.dbManager.GetSQLXDB("satudata")
if err != nil {
return err
}
qb := query.NewSQLQueryBuilder(query.DBTypePostgreSQL).SetSecurityOptions(false, 0)
// Secara eksplisit mengizinkan kolom yang memiliki case sensitive (huruf besar) dengan quotes
qb.SetAllowedColumns([]string{
`"Entity_Id"`, `"Send_status"`, "id", `"Encounter_Id"`,
})
statusStr := "false"
if status {
statusStr = "true"
}
updateData := query.UpdateData{
Columns: []string{`"Entity_Id"`, `"Send_status"`, `"Encounter_Id"`},
Values: []interface{}{fhirID, statusStr, encounterID},
}
filters := []query.FilterGroup{query.CreateAndFilterGroup([]query.DynamicFilter{query.CreateEqualFilter("id", id)})}
_, err = qb.ExecuteUpdate(ctx, satudataDB, "public.data_service_requests", updateData, filters)
return err
}
internal/worker/satusehat/servicerequest/service.go
+277 -31
View File
@@ -2,92 +2,338 @@ package servicerequest
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"log"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"time"
"service/internal/infrastructure/database"
extapi "service/internal/worker/interface"
"service/pkg/logger"
)
const trackerFile = "last_migrated_servicerequest_id.txt"
const trackerFile = "internal/worker/satusehat/servicerequest/last_servicerequest_id.txt"
const trackerFileRad = "internal/worker/satusehat/servicerequest/last_servicerequest_rad_time.txt"
type Config struct {
DBManager database.Service
InternalBaseURL string
InternalToken string
OrganizationID string
}
type TokenManager interface {
GetAccessToken() string
ForceRefreshAndGetToken(ctx context.Context) (string, error)
}
type WorkerService interface {
Run(ctx context.Context)
}
type worker struct {
cfg Config
repo Repository
apiClient extapi.Client
cfg Config
repo Repository
apiClient extapi.Client
tokenManager TokenManager
mode string
}
func NewWorker(cfg Config) WorkerService {
return &worker{cfg: cfg, repo: NewRepository(cfg.DBManager), apiClient: extapi.NewClient(15 * time.Second)}
func NewDefaultWorker(cfg Config, tokenManager TokenManager) WorkerService {
return &worker{cfg: cfg, repo: NewRepository(cfg.DBManager), apiClient: extapi.NewClient(15 * time.Second), tokenManager: tokenManager, mode: "default"}
}
func NewRadiologyWorker(cfg Config, tokenManager TokenManager) WorkerService {
return &worker{cfg: cfg, repo: NewRepository(cfg.DBManager), apiClient: extapi.NewClient(15 * time.Second), tokenManager: tokenManager, mode: "radiology"}
}
func (w *worker) Run(ctx context.Context) {
log.Println("[SERVICEREQUEST WORKER] Memulai proses sinkronisasi...")
if w.mode == "radiology" {
w.runRadiology(ctx)
} else {
w.runDefault(ctx)
}
}
func (w *worker) runDefault(ctx context.Context) {
logger.Default().Info("[SERVICEREQUEST WORKER] Memulai proses sinkronisasi...")
w.ensureTrackerFileExists()
lastID := w.readLastID()
logger.Default().Info("[SERVICEREQUEST WORKER] Melanjutkan proses dari", logger.Int64("last_id", lastID))
for {
select {
case <-ctx.Done():
log.Println("[SERVICEREQUEST WORKER] Proses dihentikan.")
logger.Default().Warn("[SERVICEREQUEST WORKER] Proses dihentikan oleh sistem (Graceful Shutdown)")
return
default:
}
sr, err := w.repo.GetNextServiceRequest(ctx, lastID)
if err != nil {
if err == sql.ErrNoRows {
time.Sleep(5 * time.Second) // Tunggu data baru
continue
}
logger.Default().Error("[SERVICEREQUEST WORKER] Gagal query ke database", logger.ErrorField(err))
time.Sleep(5 * time.Second)
continue
}
internalPayload := MapToInternalAPI(sr, w.cfg.OrganizationID)
reqURL := fmt.Sprintf("%s/satusehat/servicerequest", strings.TrimRight(w.cfg.InternalBaseURL, "/"))
respBody, statusCode, err := w.apiClient.PostJSON(ctx, reqURL, w.cfg.InternalToken, internalPayload)
sourceID := strconv.FormatInt(sr.IdxOrder, 10)
logger.Default().Info("[SERVICEREQUEST WORKER] 🔄 Memproses Dokumen",
logger.String("source_id", sourceID),
logger.String("no_mr", sr.NoMR.String),
)
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)
} 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)
}
internalPayload := MapToInternalAPI(sr, w.cfg.OrganizationID)
patientID, _ := internalPayload["patient_id"].(string)
encounterID, _ := internalPayload["encounter_id"].(string)
if patientID == "" || encounterID == "" || encounterID == "0" {
logger.Default().Warn("[SERVICEREQUEST WORKER] ⚠️ Data dilewati karena patient_id atau encounter_id kosong", logger.String("source_id", sourceID))
lastID = sr.IdxOrder
os.WriteFile(trackerFile, []byte(strconv.FormatInt(lastID, 10)), 0644)
continue
}
fhirID, respBody, status, errMsg, retryDelay := w.sendToInternalAPI(ctx, internalPayload, sourceID)
syncLog := MapToSyncLog(sr.IdxOrder, fhirID, internalPayload, respBody, status, errMsg)
if errLog := w.repo.SaveSyncLog(ctx, syncLog); errLog != nil {
logger.Default().Error("[SERVICEREQUEST WORKER] Gagal menyimpan histori log ke DB", logger.ErrorField(errLog))
}
if status == "SUCCESS" && fhirID != "" {
if errUpdate := w.repo.SaveSatuSehatID(ctx, sr.IdxOrder, fhirID); errUpdate != nil {
logger.Default().Error("[SERVICEREQUEST WORKER] Gagal update FHIR ID ke transaksi utama", logger.ErrorField(errUpdate))
}
}
_ = w.repo.SaveSyncLog(ctx, MapToSyncLog(sr.IdxOrder, fhirID, internalPayload, respBody, status, errMsg))
if status == "SUCCESS" && fhirID != "" {
_ = w.repo.SaveSatuSehatID(ctx, sr.IdxOrder, fhirID)
if retryDelay == 0 {
lastID = sr.IdxOrder
os.WriteFile(trackerFile, []byte(strconv.FormatInt(lastID, 10)), 0644)
} else {
logger.Default().Warn("[SERVICEREQUEST WORKER] ⚠️ Retrying data", logger.String("source_id", sourceID), logger.String("reason", errMsg), logger.Duration("delay", retryDelay))
}
lastID = sr.IdxOrder
os.WriteFile(trackerFile, []byte(strconv.FormatInt(lastID, 10)), 0644)
time.Sleep(500 * time.Millisecond)
if retryDelay > 0 {
time.Sleep(retryDelay)
} else {
time.Sleep(2 * time.Second) // Mencegah Rate Limit Exceeded
}
}
}
func (w *worker) runRadiology(ctx context.Context) {
logger.Default().Info("[SERVICEREQUEST-RAD WORKER] Memulai proses sinkronisasi radiologi...")
w.ensureTrackerFileRadExists()
lastTime := w.readLastTime()
// Custom data: Override membaca mulai tanggal 2025-12-18 (Format: YYYY-MM-DD)
customDate, _ := time.Parse("2006-01-02", "2026-24-04")
if lastTime.Before(customDate) {
lastTime = customDate
}
logger.Default().Info("[SERVICEREQUEST-RAD WORKER] Melanjutkan proses dari", logger.Time("last_time", lastTime))
for {
select {
case <-ctx.Done():
logger.Default().Warn("[SERVICEREQUEST-RAD WORKER] Proses dihentikan oleh sistem (Graceful Shutdown)")
return
default:
}
sr, err := w.repo.GetNextRadiologyServiceRequest(ctx, lastTime)
if err != nil {
if err == sql.ErrNoRows {
time.Sleep(5 * time.Second) // Tunggu data baru
continue
}
logger.Default().Error("[SERVICEREQUEST-RAD WORKER] Gagal query ke database", logger.ErrorField(err))
time.Sleep(5 * time.Second)
continue
}
logger.Default().Info("[SERVICEREQUEST-RAD WORKER] 🔄 Memproses Dokumen",
logger.String("source_id", sr.ID),
logger.String("patient_name", sr.PatientName.String),
)
internalPayload := MapRadiologyToInternalAPI(sr)
patientID, _ := internalPayload["patient_id"].(string)
encounterID, _ := internalPayload["encounter_id"].(string)
requesterID, _ := internalPayload["requester_id"].(string)
performerID, _ := internalPayload["performer_id"].(string)
if patientID == "" || encounterID == "" || encounterID == "0" || requesterID == "" || requesterID == "0" || performerID == "" || performerID == "0" {
logger.Default().Warn("[SERVICEREQUEST-RAD WORKER] ⚠️ Data dilewati karena patient_id atau encounter_id atau requester_id kosong", logger.String("source_id", sr.ID))
lastTime = sr.CreatedAt
os.WriteFile(trackerFileRad, []byte(lastTime.Format(time.RFC3339Nano)), 0644)
continue
}
fhirID, respBody, status, errMsg, retryDelay := w.sendToInternalAPI(ctx, internalPayload, sr.ID)
// Menyimpan payload request dan response ke dalam log
logger.Default().Info("[SERVICEREQUEST-RAD WORKER] Detail Transaksi",
logger.String("source_id", sr.ID),
logger.Any("request_payload", internalPayload),
logger.String("response_payload", string(respBody)),
logger.String("status", status),
)
if status == "SUCCESS" && fhirID != "" {
if errUpdate := w.repo.UpdateRadiologyServiceRequestStatus(ctx, sr.ID, fhirID, encounterID, true); errUpdate != nil {
logger.Default().Error("[SERVICEREQUEST-RAD WORKER] Gagal update status ke transaksi utama", logger.ErrorField(errUpdate))
}
}
if retryDelay == 0 {
lastTime = sr.CreatedAt
os.WriteFile(trackerFileRad, []byte(lastTime.Format(time.RFC3339Nano)), 0644)
} else {
logger.Default().Warn("[SERVICEREQUEST-RAD WORKER] ⚠️ Retrying data", logger.String("source_id", sr.ID), logger.String("reason", errMsg), logger.Duration("delay", retryDelay))
}
if retryDelay > 0 {
time.Sleep(retryDelay)
} else {
time.Sleep(2 * time.Second) // Mencegah Rate Limit Exceeded
}
}
}
// sendToInternalAPI mengirimkan payload ke internal FHIR server, menangani auth retry (401) dan response error.
func (w *worker) sendToInternalAPI(ctx context.Context, payload map[string]interface{}, sourceID string) (fhirID string, respBody []byte, status string, errMsg string, retryDelay time.Duration) {
reqURL := fmt.Sprintf("%s/satusehat/servicerequest", strings.TrimRight(w.cfg.InternalBaseURL, "/"))
workerName := "SERVICEREQUEST WORKER"
if w.mode == "radiology" {
workerName = "SERVICEREQUEST-RAD WORKER"
}
// --- Percobaan Pertama ---
respBody, statusCode, err := w.apiClient.PostJSON(ctx, reqURL, w.tokenManager.GetAccessToken(), payload)
if err != nil {
logger.Default().Error(fmt.Sprintf("[%s] ❌ Gagal Koneksi ke API Part 3", workerName),
logger.String("source_id", sourceID),
logger.Any("payload_sent", payload),
logger.ErrorField(err),
)
return "", nil, "FAILED", err.Error(), 5 * time.Second
}
// --- Penanganan Token Unauthorized (401) ---
if statusCode == http.StatusUnauthorized {
logger.Default().Warn(fmt.Sprintf("[%s] Menerima status 401 Unauthorized, mencoba refresh token...", workerName), logger.String("source_id", sourceID))
newToken, refreshErr := w.tokenManager.ForceRefreshAndGetToken(ctx)
if refreshErr != nil {
logger.Default().Error(fmt.Sprintf("[%s] Gagal refresh token, akan mencoba lagi pada iterasi berikutnya.", workerName), logger.String("source_id", sourceID), logger.ErrorField(refreshErr))
return "", respBody, "FAILED", refreshErr.Error(), 5 * time.Second
}
logger.Default().Info(fmt.Sprintf("[%s] Token berhasil di-refresh, mencoba ulang request...", workerName), logger.String("source_id", sourceID))
respBody, statusCode, err = w.apiClient.PostJSON(ctx, reqURL, newToken, payload)
if err != nil {
logger.Default().Error(fmt.Sprintf("[%s] ❌ Gagal Koneksi ke API Part 3 (setelah retry)", workerName),
logger.String("source_id", sourceID),
logger.ErrorField(err),
)
return "", nil, "FAILED", err.Error(), 5 * time.Second
}
}
// --- Evaluasi Final Respons ---
if statusCode != http.StatusOK && statusCode != http.StatusCreated {
errMsg = fmt.Sprintf("HTTP %d", statusCode)
logger.Default().Error(fmt.Sprintf("[%s] ❌ Respons Error dari API Part 3", workerName),
logger.String("source_id", sourceID),
logger.Int("status_code", statusCode),
logger.String("response", truncate(string(respBody), 200)),
)
// Lakukan Retry jika server Internal error (5xx) atau Rate Limit (429)
if statusCode >= 500 || statusCode == http.StatusTooManyRequests {
delay := 5 * time.Second
if statusCode == http.StatusTooManyRequests {
delay = 60 * time.Second // Default fallback jika parse gagal
var errResp struct {
RetryAfter int `json:"retry_after"`
}
if errParse := json.Unmarshal(respBody, &errResp); errParse == nil && errResp.RetryAfter > 0 {
delay = time.Duration(errResp.RetryAfter) * time.Second
}
}
return "", respBody, "FAILED", errMsg, delay
}
return "", respBody, "FAILED", errMsg, 0
}
// --- Sukses ---
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
}
}
logger.Default().Info(fmt.Sprintf("[%s] ✅ Sukses mengirim data ServiceRequest", workerName),
logger.String("source_id", sourceID),
logger.String("fhir_id", fhirID),
)
return fhirID, respBody, "SUCCESS", "", 0
}
func (w *worker) readLastID() int64 {
data, _ := os.ReadFile(trackerFile)
id, _ := strconv.ParseInt(strings.TrimSpace(string(data)), 10, 64)
return id
}
func (w *worker) ensureTrackerFileExists() {
if _, err := os.Stat(trackerFile); os.IsNotExist(err) {
dir := filepath.Dir(trackerFile)
if err := os.MkdirAll(dir, 0755); err != nil {
logger.Default().Error("[SERVICEREQUEST WORKER] Gagal membuat direktori tracker", logger.ErrorField(err))
}
if err := os.WriteFile(trackerFile, []byte("0"), 0644); err != nil {
logger.Default().Error("[SERVICEREQUEST WORKER] Gagal membuat file tracker default", logger.ErrorField(err))
}
}
}
func (w *worker) readLastTime() time.Time {
data, err := os.ReadFile(trackerFileRad)
if err != nil {
return time.Time{}
}
t, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(string(data)))
if err != nil {
return time.Time{}
}
return t
}
func (w *worker) ensureTrackerFileRadExists() {
if _, err := os.Stat(trackerFileRad); os.IsNotExist(err) {
dir := filepath.Dir(trackerFileRad)
_ = os.MkdirAll(dir, 0755)
_ = os.WriteFile(trackerFileRad, []byte(time.Time{}.Format(time.RFC3339Nano)), 0644)
}
}
func truncate(s string, max int) string {
if len(s) > max {
return s[:max] + "..."
}
return s
}
internal/worker/worker.go
+201 -200
View File
@@ -10,25 +10,7 @@ import (
"service/internal/infrastructure/config"
"service/internal/infrastructure/database"
"service/internal/interfaces/satusehat"
"service/internal/worker/satusehat/allergyintolerance"
"service/internal/worker/satusehat/careplan"
"service/internal/worker/satusehat/clinicalimpression"
"service/internal/worker/satusehat/composition"
"service/internal/worker/satusehat/condition"
"service/internal/worker/satusehat/diagnosticreport"
"service/internal/worker/satusehat/encounter"
"service/internal/worker/satusehat/episodeofcare"
"service/internal/worker/satusehat/imagingstudy"
"service/internal/worker/satusehat/immunization"
"service/internal/worker/satusehat/medication"
"service/internal/worker/satusehat/medicationdispense"
"service/internal/worker/satusehat/medicationrequest"
"service/internal/worker/satusehat/medicationstatement"
"service/internal/worker/satusehat/observation"
"service/internal/worker/satusehat/procedure"
"service/internal/worker/satusehat/questionnaireresponse"
"service/internal/worker/satusehat/servicerequest"
"service/internal/worker/satusehat/specimen"
"service/pkg/logger"
)
@@ -37,6 +19,11 @@ type Manager struct {
cfg *config.Config
db database.Service
satusehatClient satusehat.SatuSehatClient
// Auth state for all workers
accessToken string
refreshToken string
tokenMutex sync.RWMutex
}
// NewManager membuat instance Manager baru
@@ -52,20 +39,21 @@ func NewManager(cfg *config.Config, db database.Service, satusehatClient satuseh
func (m *Manager) Start(ctx context.Context) {
logger.Default().Info("Starting background workers...")
// Lakukan login awal untuk mendapatkan token bagi semua worker
if err := m.login(ctx); err != nil {
logger.Default().Fatal("Initial login for workers failed, stopping.", logger.ErrorField(err))
return
}
var wg sync.WaitGroup
// Tentukan Organization ID (Bisa dari Environment Variable atau Config Struct)
orgID := os.Getenv("SATUSEHAT_ORGANIZATION_ID")
orgID := os.Getenv("SATUSEHAT_ORG_ID")
if orgID == "" {
logger.Default().Warn("SATUSEHAT_ORGANIZATION_ID is not set in environment variables")
logger.Default().Warn("SATUSEHAT_ORG_ID is not set in environment variables")
}
internalPort := 8080
if m.cfg != nil && m.cfg.Server.REST.Port != 0 {
internalPort = m.cfg.Server.REST.Port
}
internalBaseURL := fmt.Sprintf("http://localhost:%d/api/v1", internalPort)
internalToken := os.Getenv("INTERNAL_API_TOKEN")
internalBaseURL := m.cfg.SatuSehat.InternalFHIRServerURL
// Jobs mendefinisikan urutan dan jeda waktu (delay) boot-up setiap worker.
// Konfigurasi waktu ini mencegah perebutan resource di awal (thundering herd)
@@ -76,195 +64,208 @@ func (m *Manager) Start(ctx context.Context) {
delay time.Duration
run func(context.Context)
}{
// {
// name: "Encounter",
// delay: 5 * time.Second, // Memberikan jeda 5 detik agar Encounter mendapat porsi jalan lebih dulu
// run: func(c context.Context) {
// encounter.NewWorker(encounter.Config{
// DBManager: m.db, InternalBaseURL: internalBaseURL, OrganizationID: orgID,
// }).Run(c)
// },
// },
// {
// name: "Condition",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // condition.NewWorker(...).Run(c)
// },
// },
{
name: "Medication (Master)",
delay: 0, // Master data berjalan seketika
run: func(c context.Context) {
medication.NewWorker(medication.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "Encounter",
delay: 5 * time.Second, // Memberikan jeda 5 detik agar Encounter mendapat porsi jalan lebih dulu
run: func(c context.Context) {
encounter.NewWorker(encounter.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "Condition",
delay: 2 * time.Second,
run: func(c context.Context) {
condition.NewWorker(condition.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "Observation",
delay: 2 * time.Second,
run: func(c context.Context) {
observation.NewWorker(observation.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "Procedure",
delay: 2 * time.Second,
run: func(c context.Context) {
procedure.NewWorker(procedure.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "AllergyIntolerance",
delay: 2 * time.Second,
run: func(c context.Context) {
allergyintolerance.NewWorker(allergyintolerance.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "ClinicalImpression",
delay: 2 * time.Second,
run: func(c context.Context) {
clinicalimpression.NewWorker(clinicalimpression.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "Immunization",
delay: 2 * time.Second,
run: func(c context.Context) {
immunization.NewWorker(immunization.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "ServiceRequest",
delay: 2 * time.Second,
run: func(c context.Context) {
servicerequest.NewWorker(servicerequest.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "Specimen",
delay: 2 * time.Second,
run: func(c context.Context) {
specimen.NewWorker(specimen.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "DiagnosticReport",
delay: 2 * time.Second,
run: func(c context.Context) {
diagnosticreport.NewWorker(diagnosticreport.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "ImagingStudy",
delay: 2 * time.Second,
run: func(c context.Context) {
imagingstudy.NewWorker(imagingstudy.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "Composition",
delay: 2 * time.Second,
run: func(c context.Context) {
composition.NewWorker(composition.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "CarePlan",
delay: 2 * time.Second,
run: func(c context.Context) {
careplan.NewWorker(careplan.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "EpisodeOfCare",
delay: 2 * time.Second,
run: func(c context.Context) {
episodeofcare.NewWorker(episodeofcare.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "MedicationRequest",
delay: 2 * time.Second,
run: func(c context.Context) {
medicationrequest.NewWorker(medicationrequest.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "MedicationDispense",
delay: 2 * time.Second,
run: func(c context.Context) {
medicationdispense.NewWorker(medicationdispense.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "MedicationStatement",
delay: 2 * time.Second,
run: func(c context.Context) {
medicationstatement.NewWorker(medicationstatement.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
},
},
{
name: "QuestionnaireResponse",
delay: 2 * time.Second,
run: func(c context.Context) {
questionnaireresponse.NewWorker(questionnaireresponse.Config{
DBManager: m.db, InternalBaseURL: internalBaseURL, InternalToken: internalToken, OrganizationID: orgID,
}).Run(c)
DBManager: m.db,
InternalBaseURL: internalBaseURL,
OrganizationID: orgID,
}, m).Run(c)
},
},
// {
// name: "Observation",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // observation.NewWorker(...).Run(c)
// },
// },
// {
// name: "Procedure",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // procedure.NewWorker(...).Run(c)
// },
// },
// {
// name: "AllergyIntolerance",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // allergyintolerance.NewWorker(...).Run(c)
// },
// },
// {
// name: "ClinicalImpression",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // clinicalimpression.NewWorker(...).Run(c)
// },
// },
// {
// name: "Immunization",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // immunization.NewWorker(...).Run(c)
// },
// },
// {
// name: "ServiceRequest (Default)",
// delay: 0,
// run: func(c context.Context) {
// servicerequest.NewDefaultWorker(servicerequest.Config{
// DBManager: m.db,
// InternalBaseURL: internalBaseURL,
// OrganizationID: orgID,
// }, m).Run(c)
// },
// },
// {
// name: "ServiceRequest (Radiology)",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// servicerequest.NewRadiologyWorker(servicerequest.Config{
// DBManager: m.db,
// InternalBaseURL: internalBaseURL,
// OrganizationID: orgID,
// }, m).Run(c)
// },
// },
// {
// name: "Specimen",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // specimen.NewWorker(...).Run(c)
// },
// },
// {
// name: "DiagnosticReport",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // diagnosticreport.NewWorker(...).Run(c)
// },
// },
// {
// name: "ImagingStudy",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // The Manager 'm' now acts as the TokenManager
// imagingstudy.NewWorker(imagingstudy.Config{
// DBManager: m.db, InternalBaseURL: internalBaseURL, OrganizationID: orgID,
// }, m).Run(c)
// },
// },
// {
// name: "Composition",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // composition.NewWorker(...).Run(c)
// },
// },
// {
// name: "CarePlan",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // careplan.NewWorker(...).Run(c)
// },
// },
// {
// name: "EpisodeOfCare",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // episodeofcare.NewWorker(...).Run(c)
// },
// },
// {
// name: "MedicationRequest",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // medicationrequest.NewWorker(...).Run(c)
// },
// },
// {
// name: "MedicationDispense",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // medicationdispense.NewWorker(...).Run(c)
// },
// },
// {
// name: "MedicationStatement",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // medicationstatement.NewWorker(...).Run(c)
// },
// },
// {
// name: "QuestionnaireResponse",
// delay: 2 * time.Second,
// run: func(c context.Context) {
// // TODO: Refactor this worker
// // questionnaireresponse.NewWorker(...).Run(c)
// },
// },
}
// Eksekusi semua background worker secara tertib
// Eksekusi semua background worker secara tertib dengan delay bertingkat
go func() {
for _, job := range jobs {
// Jeda startup, dapat di-interrupt oleh Context Cancel (Shutdown System)
if job.delay > 0 {
select {
case <-ctx.Done():
return
case <-time.After(job.delay):
}
baseDelay := 10 * time.Second // Jeda 10 detik antar tiap worker untuk cegah spike di awal
for i, job := range jobs {
// Hitung delay bertingkat: (index) * baseDelay
staggeredDelay := time.Duration(i) * baseDelay
// Jika job memiliki delay statis yang lebih besar, utamakan nilai tersebut
if job.delay > staggeredDelay {
staggeredDelay = job.delay
}
wg.Add(1)
j := job // copy pointer untuk goroutine loop
j := job
d := staggeredDelay
go func() {
defer wg.Done()
if d > 0 {
select {
case <-ctx.Done():
return
case <-time.After(d):
}
}
logger.Default().Info(fmt.Sprintf("[%s WORKER] Started", j.name))
j.run(ctx)
logger.Default().Info(fmt.Sprintf("[%s WORKER] Stopped", j.name))