package pipeline import ( "context" "net/http" "strings" "sync" "time" "evobgp/internal/httpclient" "evobgp/internal/store" ) type prefetchTask struct { tenantID string mod *store.Module src *store.CDNSource } // PrefetchCDNSourceETags performs conditional GETs for CDN sources; on 200 parses CIDRs into module_prefix_snapshot. func PrefetchCDNSourceETags(ctx context.Context, st store.Backend, hc *http.Client) error { if hc == nil { hc = httpclient.New(httpclient.DefaultTimeout) } if ctx == nil { ctx = context.Background() } tenants, err := st.ListTenantIDs() if err != nil { return err } var tasks []prefetchTask now := time.Now().UTC() for _, tid := range tenants { for _, mod := range st.ListModules(tid) { if mod == nil || !mod.Enabled || mod.Type != "CDN_CIDRS" { continue } sources, err := st.ListCDNSources(tid, mod.ID) if err != nil { continue } for _, src := range sources { if src == nil || strings.TrimSpace(src.URL) == "" { continue } // Respect per-source refresh intervals: conditional GET only for due sources. // The ETag probe still lets 304s skip body downloads for the rest. if shouldSkipCDNSourceFetch(src, now) { continue } tasks = append(tasks, prefetchTask{tenantID: tid, mod: mod, src: src}) } } } if len(tasks) == 0 { return nil } sem := make(chan struct{}, collectConcurrency()) var wg sync.WaitGroup for _, task := range tasks { wg.Add(1) go func(t prefetchTask) { defer wg.Done() sem <- struct{}{} defer func() { <-sem }() prefetchOneCDNSource(ctx, st, hc, t) }(task) } wg.Wait() return nil } func prefetchOneCDNSource(ctx context.Context, st store.Backend, hc *http.Client, t prefetchTask) { now := time.Now().UTC() tid, mod, src := t.tenantID, t.mod, t.src omod, err := st.GetModule(tid, mod.ID) if err != nil || omod == nil { return } rows, err := fetchCDNSourceRows(ctx, st, hc, tid, mod.ID, omod, src, nil, now) if err != nil { return } _ = mergeCDNSourceIntoModuleSnapshot(st, tid, omod, src.ID, rows) }