EvoBGP/internal/pipeline/prefetch.go

package pipeline

import (
	"context"
	"io"
	"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
	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) != "" {
					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
	u := strings.TrimSpace(src.URL)
	if _, err := ValidateCDNURL(u); err != nil {
		return
	}
	if err := ResolveCDNURLHost(ctx, u); err != nil {
		return
	}
	omod, err := st.GetModule(tid, mod.ID)
	if err != nil {
		return
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
	if err != nil {
		return
	}
	if etag := strings.TrimSpace(src.Etag); etag != "" {
		req.Header.Set("If-None-Match", etag)
	}
	resp, err := upstreamHTTPDo(ctx, hc, req)
	if err != nil {
		return
	}
	if resp.StatusCode == http.StatusNotModified {
		_ = resp.Body.Close()
		return
	}
	if resp.StatusCode != http.StatusOK {
		_, _ = io.Copy(io.Discard, resp.Body)
		_ = resp.Body.Close()
		return
	}
	body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
	_ = resp.Body.Close()
	if err != nil {
		return
	}
	prefixStrs, err := parseCDNBody(string(body), src)
	if err != nil {
		return
	}
	newEtag := strings.TrimSpace(resp.Header.Get("ETag"))
	patch := &store.CDNSourcePatch{LastRefreshedAt: &now}
	if newEtag != "" && newEtag != strings.TrimSpace(src.Etag) {
		e := newEtag
		patch.Etag = &e
	}
	_, _ = st.UpdateCDNSource(tid, mod.ID, src.ID, patch)
	rows := cdnRowsFromParsed(omod, src, prefixStrs)
	_ = mergeCDNSourceIntoModuleSnapshot(st, tid, omod, src.ID, rows)
}