package kfa import ( "context" "encoding/json" "fmt" "io" "net/http" "os" "strconv" "strings" "time" "service/pkg/logger" ) type TokenManager interface { GetAccessToken() string ForceRefreshAndGetToken(ctx context.Context) (string, error) } type Config struct { InternalBaseURL string PageSize int } type Worker struct { cfg Config repo CommandRepository tokenManager TokenManager } func NewWorker(cfg Config, repo CommandRepository, tokenManager TokenManager) *Worker { if cfg.PageSize == 0 { cfg.PageSize = 100 } return &Worker{ cfg: cfg, repo: repo, tokenManager: tokenManager, } } func (w *Worker) readTracker() int { b, err := os.ReadFile("internal/master/kfa/kfa_tracker.txt") if err != nil { return 1 } p, err := strconv.Atoi(strings.TrimSpace(string(b))) if err != nil || p < 1 { return 1 } return p } func (w *Worker) writeTracker(page int) { os.WriteFile("internal/master/kfa/kfa_tracker.txt", []byte(strconv.Itoa(page)), 0644) } func (w *Worker) Run(ctx context.Context) { page := w.readTracker() for { select { case <-ctx.Done(): logger.Default().Info("KFA Master Puller stopped") return default: err := w.processPage(ctx, page) if err != nil { logger.Default().Error("KFA Master Puller error processing page", logger.ErrorField(err), logger.Int("page", page)) time.Sleep(30 * time.Second) continue } page++ w.writeTracker(page) time.Sleep(5 * time.Second) // Jedah halus antar halaman } } } func (w *Worker) processPage(ctx context.Context, page int) error { listURL := fmt.Sprintf("%s/satusehat/reference/kfa/products?page=%d&size=%d&product_type=farmasi", strings.TrimRight(w.cfg.InternalBaseURL, "/"), page, w.cfg.PageSize) req, err := http.NewRequestWithContext(ctx, http.MethodGet, listURL, nil) if err != nil { return err } req.Header.Set("Authorization", "Bearer "+w.tokenManager.GetAccessToken()) req.Header.Set("Accept", "application/json") client := &http.Client{Timeout: 30 * time.Second} resp, err := client.Do(req) if err != nil { return err } if resp.StatusCode == http.StatusUnauthorized { resp.Body.Close() logger.Default().Warn("KFA Master Puller list API 401, refreshing token...", logger.Int("page", page)) newToken, err := w.tokenManager.ForceRefreshAndGetToken(ctx) if err != nil { return err } req, _ = http.NewRequestWithContext(ctx, http.MethodGet, listURL, nil) req.Header.Set("Authorization", "Bearer "+newToken) req.Header.Set("Accept", "application/json") resp, err = client.Do(req) if err != nil { return err } } defer resp.Body.Close() if resp.StatusCode == http.StatusTooManyRequests { logger.Default().Warn("[KFA WORKER] ⏸️ Rate limit (429) pada List API, jeda 30s & ulangi halaman", logger.Int("page", page)) return fmt.Errorf("rate limited on list API") // Akan memicu waktu tidur 30 detik di method Run() dan mencoba halaman yang sama } if resp.StatusCode != http.StatusOK { return fmt.Errorf("list API returned status: %d", resp.StatusCode) } body, err := io.ReadAll(resp.Body) if err != nil { return err } var listRes KfaListResponse if err := json.Unmarshal(body, &listRes); err != nil { return err } if len(listRes.Data.Items.Data) == 0 { logger.Default().Info("No more KFA products found, resetting to page 1 and sleeping", logger.Int("page", page)) w.writeTracker(1) time.Sleep(6 * time.Hour) return fmt.Errorf("empty page, reset triggered") } logger.Default().Info("Pulling KFA items", logger.Int("page", page), logger.Int("items", len(listRes.Data.Items.Data))) for _, item := range listRes.Data.Items.Data { for { rateLimited, err := w.fetchAndSaveDetail(ctx, item.KfaCode) if rateLimited { logger.Default().Warn("[KFA WORKER] ⏸️ Rate limit (429) dari Satu Sehat - jeda 60 detik dan retry dokumen yang sama", logger.String("kfa_code", item.KfaCode)) time.Sleep(60 * time.Second) continue // Ulangi request untuk item ini (Tidak melewatkan ID) } if err != nil { logger.Default().Error("Failed to fetch/save KFA detail", logger.String("kfa_code", item.KfaCode), logger.ErrorField(err)) } break // Lanjut ke item/ID berikutnya bila sukses atau gagal karena alasan non-ratelimit } time.Sleep(500 * time.Millisecond) // Pencegahan Rate Limit Part 3 } return nil } func (w *Worker) fetchAndSaveDetail(ctx context.Context, kfaCode string) (bool, error) { detailURL := fmt.Sprintf("%s/satusehat/reference/kfa/products/%s", strings.TrimRight(w.cfg.InternalBaseURL, "/"), kfaCode) req, err := http.NewRequestWithContext(ctx, http.MethodGet, detailURL, nil) if err != nil { return false, err } req.Header.Set("Authorization", "Bearer "+w.tokenManager.GetAccessToken()) req.Header.Set("Accept", "application/json") client := &http.Client{Timeout: 30 * time.Second} resp, err := client.Do(req) if err != nil { return false, err } if resp.StatusCode == http.StatusUnauthorized { resp.Body.Close() newToken, err := w.tokenManager.ForceRefreshAndGetToken(ctx) if err != nil { return false, err } req, _ = http.NewRequestWithContext(ctx, http.MethodGet, detailURL, nil) req.Header.Set("Authorization", "Bearer "+newToken) req.Header.Set("Accept", "application/json") resp, err = client.Do(req) if err != nil { return false, err } } defer resp.Body.Close() if resp.StatusCode == http.StatusTooManyRequests { return true, fmt.Errorf("rate limited (429)") } if resp.StatusCode != http.StatusOK { return false, fmt.Errorf("detail API returned status: %d", resp.StatusCode) } body, err := io.ReadAll(resp.Body) if err != nil { return false, err } var detailRes KfaDetailResponse if err := json.Unmarshal(body, &detailRes); err != nil { return false, err } product, ingredients, packagings := MapKfaDetailToEntity(detailRes.Data.Result) return false, w.repo.UpsertProductBundle(ctx, product, ingredients, packagings) }