feat: add tenant refresh functionality to HTTP API and job processing

Implemented a new endpoint for tenant refresh in the HTTP API, allowing for the refresh of modules associated with a tenant. Enhanced job processing to handle tenant refresh jobs, including logic for managing module IDs and job status updates. Updated the scheduler to enqueue tenant refresh jobs based on module due dates, improving the overall efficiency of module management. Additionally, introduced caching for ASN prefix data to optimize performance during refresh operations.

internal/httpapi/routes.go

@@ -18,6 +18,7 @@ import (
 	"evobgp/internal/bundle"
 	"evobgp/internal/jobs"
 	"evobgp/internal/observability"
+	"evobgp/internal/pipeline"
 	"evobgp/internal/store"
 )
 
@@ -52,6 +53,7 @@ func (s *Server) registerV1(m *http.ServeMux) {
 	m.HandleFunc("GET /peers", s.handleListPeers)
 	m.HandleFunc("GET /speakers", s.handleListSpeakers)
 	m.HandleFunc("POST /modules/{module_id}/refresh", s.handleModuleRefresh)
+	m.HandleFunc("POST /tenant/refresh", s.handleTenantRefresh)
 	m.HandleFunc("GET /revisions", s.handleListRevisions)
 	m.HandleFunc("GET /revisions/{revision_id}", s.handleGetRevision)
 	m.HandleFunc("GET /revisions/{revision_id}/preview", s.handleRevisionPreview)
@@ -462,6 +464,60 @@ func (s *Server) handleModuleRefresh(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusAccepted, map[string]any{"job_id": snap["job_id"], "status": snap["status"]})
 }
 
+func (s *Server) handleTenantRefresh(w http.ResponseWriter, r *http.Request) {
+	a, ok := authFromContext(r.Context())
+	if !ok {
+		writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
+		return
+	}
+	if !s.requireAtLeast(w, a, "editor") {
+		return
+	}
+	var body struct {
+		ModuleIDs []string `json:"module_ids"`
+	}
+	if r.Body != nil && r.ContentLength != 0 {
+		if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&body); err != nil && err != io.EOF {
+			writeProblem(w, http.StatusBadRequest, "Bad Request", "invalid JSON body")
+			return
+		}
+	}
+	moduleIDs := body.ModuleIDs
+	if len(moduleIDs) == 0 {
+		now := time.Now().UTC()
+		for _, mod := range s.store.ListModules(a.TenantID) {
+			if pipeline.ModuleDueForScheduler(mod, now) {
+				moduleIDs = append(moduleIDs, mod.ID)
+			}
+		}
+	}
+	if len(moduleIDs) == 0 {
+		writeJSON(w, http.StatusNoContent, map[string]any{"message": "no modules due for refresh"})
+		return
+	}
+	idem := strings.TrimSpace(r.Header.Get("Idempotency-Key"))
+	var idemPtr *string
+	if idem != "" {
+		idemPtr = &idem
+	}
+	j, _, err := s.jobs.Enqueue(a.TenantID, jobs.KindTenantRefresh, idemPtr, nil, map[string]any{
+		"module_ids": moduleIDs,
+		"trigger":    "api",
+	})
+	if err != nil {
+		writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error())
+		return
+	}
+	w.Header().Set("Location", "/v1/jobs/"+j.ID)
+	snap := j.Snapshot()
+	writeJSON(w, http.StatusAccepted, map[string]any{
+		"job_id":      snap["job_id"],
+		"status":      snap["status"],
+		"module_ids":  moduleIDs,
+		"modules_due": len(moduleIDs),
+	})
+}
+
 func (s *Server) handleListRevisions(w http.ResponseWriter, r *http.Request) {
 	a, ok := authFromContext(r.Context())
 	if !ok {

internal/jobs/job.go

@@ -346,14 +346,21 @@ func (r *Registry) List(tenantID, statusFilter, kindFilter, cursor string, limit
 // CountOtherActiveModuleRefresh returns how many module_refresh jobs for the tenant are still
 // queued or running, excluding excludeJobID (the current job). Used to batch deploy_apply.
 func (r *Registry) CountOtherActiveModuleRefresh(tenantID, excludeJobID string) int {
+	return r.CountOtherActiveRefresh(tenantID, excludeJobID)
+}
+
+// CountOtherActiveRefresh counts queued/running module_refresh and tenant_refresh jobs for the tenant.
+func (r *Registry) CountOtherActiveRefresh(tenantID, excludeJobID string) int {
 	if r == nil {
 		return 0
 	}
-	// Two-phase: snapshot job pointers under r.mu, then read j.Status under Job.mu.
 	r.mu.RLock()
 	candidates := make([]*Job, 0, 8)
 	for _, j := range r.byID {
-		if j.TenantID != tenantID || j.Kind != KindModuleRefresh {
+		if j.TenantID != tenantID {
+			continue
+		}
+		if j.Kind != KindModuleRefresh && j.Kind != KindTenantRefresh {
 			continue
 		}
 		if j.ID == excludeJobID {

internal/jobs/worker.go

@@ -42,6 +42,7 @@ func mergeBirdPostApplyMeta(j *Job) {
 
 const (
 	KindModuleRefresh    = "module_refresh"
+	KindTenantRefresh    = "tenant_refresh"
 	KindPeerReconcile    = "peer_reconcile"
 	KindDeployApply      = "deploy_apply"
 	KindRevisionRollback = "revision_rollback"
@@ -106,6 +107,8 @@ func (w *Worker) Process(j *Job) {
 			return
 		}
 		w.finishModuleRefreshSuccess(j, mid)
+	case KindTenantRefresh:
+		w.runTenantRefresh(j)
 	case KindPeerReconcile:
 		w.runPeerReconcile(j)
 	case KindDeployApply:
@@ -221,6 +224,49 @@ func (w *Worker) tenantRefreshMu(tenantID string) *sync.Mutex {
 
 // finishModuleRefreshSuccess marks the refresh job and, for the last active refresh in tenant,
 // creates one aggregate revision and enqueues a single deploy_apply.
+func (w *Worker) runTenantRefresh(j *Job) {
+	moduleIDs := moduleIDsFromJobMeta(j.Meta)
+	if len(moduleIDs) == 0 {
+		j.Fail("missing module_ids in job meta")
+		return
+	}
+	if err := pipeline.RefreshTenantModules(context.Background(), w.Store, w.httpClient(), j.TenantID, moduleIDs); err != nil {
+		j.Fail(err.Error())
+		return
+	}
+	trigger, err := pipeline.PickTenantRefreshTriggerModule(w.Store, j.TenantID, moduleIDs)
+	if err != nil {
+		j.Fail(err.Error())
+		return
+	}
+	j.mergeMeta(map[string]any{"module_ids": moduleIDs, "modules_refreshed": len(moduleIDs)})
+	w.finishModuleRefreshSuccess(j, trigger)
+}
+
+func moduleIDsFromJobMeta(meta map[string]any) []string {
+	if meta == nil {
+		return nil
+	}
+	raw, ok := meta["module_ids"]
+	if !ok {
+		return nil
+	}
+	switch v := raw.(type) {
+	case []string:
+		return v
+	case []any:
+		var out []string
+		for _, x := range v {
+			if s, ok := x.(string); ok && strings.TrimSpace(s) != "" {
+				out = append(out, strings.TrimSpace(s))
+			}
+		}
+		return out
+	default:
+		return nil
+	}
+}
+
 func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) {
 	if w == nil || w.Store == nil {
 		j.Succeed()
@@ -231,7 +277,7 @@ func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) {
 	defer mu.Unlock()
 	deferDeploy := false
 	if w.Registry != nil {
-		deferDeploy = w.Registry.CountOtherActiveModuleRefresh(j.TenantID, j.ID) > 0
+		deferDeploy = w.Registry.CountOtherActiveRefresh(j.TenantID, j.ID) > 0
 	}
 	if deferDeploy {
 		j.mergeMeta(map[string]any{

internal/pipeline/asn_cache.go

@@ -0,0 +1,58 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/netip"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"evobgp/internal/asnresolve"
+	"evobgp/internal/store"
+)
+
+func asnCacheTTL() time.Duration {
+	sec := 1800
+	if s := strings.TrimSpace(os.Getenv("EVOBGP_ASN_CACHE_TTL_SEC")); s != "" {
+		if v, err := strconv.Atoi(s); err == nil && v > 0 {
+			sec = v
+		}
+	}
+	return time.Duration(sec) * time.Second
+}
+
+// resolveASNForEntry fetches prefixes and holder with shared TTL cache (asn_prefix_cache).
+func resolveASNForEntry(ctx context.Context, st store.Backend, hc *http.Client, asn int64) ([]netip.Prefix, string, error) {
+	ttl := asnCacheTTL()
+	if st != nil {
+		if ent, ok, err := st.GetASNPrefixCache(asn); err == nil && ok && ent != nil && time.Since(ent.FetchedAt) < ttl {
+			out := make([]netip.Prefix, 0, len(ent.Prefixes))
+			for _, p := range ent.Prefixes {
+				pfx, perr := netip.ParsePrefix(strings.TrimSpace(p))
+				if perr != nil {
+					continue
+				}
+				out = append(out, pfx.Masked())
+			}
+			return out, ent.Holder, nil
+		}
+	}
+	pfxs, err := asnresolve.AnnouncedPrefixes(ctx, hc, asn)
+	if err != nil {
+		return nil, "", err
+	}
+	holder, _ := asnresolve.ASHolderName(ctx, hc, asn)
+	if st != nil {
+		strs := make([]string, len(pfxs))
+		for i, p := range pfxs {
+			strs[i] = p.String()
+		}
+		if err := st.SetASNPrefixCache(asn, holder, strs); err != nil {
+			return nil, "", fmt.Errorf("asn cache AS%d: %w", asn, err)
+		}
+	}
+	return pfxs, holder, nil
+}

internal/pipeline/asn_cache_test.go

@@ -0,0 +1,64 @@
+package pipeline
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"evobgp/internal/store"
+)
+
+func TestResolveASNForEntry_UsesTTLCache(t *testing.T) {
+	m := store.NewMemory()
+	t.Setenv("EVOBGP_ASN_CACHE_TTL_SEC", "3600")
+
+	var calls atomic.Int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls.Add(1)
+		if strings.Contains(r.URL.Path, "/announced") {
+			_, _ = w.Write([]byte(`{"status":"ok","data":{"prefixes":[{"prefix":"203.0.113.0/24"}]}}`))
+			return
+		}
+		_, _ = w.Write([]byte(`{"status":"ok","data":{"holder":"Test AS"}}`))
+	}))
+	defer srv.Close()
+	t.Setenv("EVOBGP_RIPESTAT_ANNOUNCED_PREFIXES_URL", srv.URL+"/announced")
+	t.Setenv("EVOBGP_RIPESTAT_AS_OVERVIEW_URL", srv.URL+"/overview")
+
+	ctx := context.Background()
+	hc := srv.Client()
+	p1, h1, err := resolveASNForEntry(ctx, m, hc, 64512)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(p1) != 1 || h1 != "Test AS" {
+		t.Fatalf("unexpected first resolve: %+v holder=%q", p1, h1)
+	}
+	if calls.Load() < 2 {
+		t.Fatalf("expected ripestat calls on first resolve, got %d", calls.Load())
+	}
+	firstCalls := calls.Load()
+	_, _, err = resolveASNForEntry(ctx, m, hc, 64512)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if calls.Load() != firstCalls {
+		t.Fatalf("expected cache hit (no new HTTP), calls went from %d to %d", firstCalls, calls.Load())
+	}
+}
+
+func TestModuleDueForScheduler_BucketRollover(t *testing.T) {
+	mod := &store.Module{Enabled: true, Type: "CDN_CIDRS", RefreshIntervalSec: 300}
+	boundary := time.Unix(300, 0)
+	if !ModuleDueForScheduler(mod, boundary) {
+		t.Fatal("expected due when refresh bucket rolls")
+	}
+	mid := time.Unix(330, 0)
+	if ModuleDueForScheduler(mod, mid) {
+		t.Fatal("expected not due within same bucket")
+	}
+}

internal/pipeline/cdn_prefetch_test.go

@@ -0,0 +1,99 @@
+package pipeline
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"evobgp/internal/store"
+)
+
+func TestCollectModulePrefixRows_CDNSendsIfNoneMatch(t *testing.T) {
+	m := store.NewMemory()
+	m.SeedDemo()
+	tenant, _, _, _, _ := m.DemoIDs()
+
+	mod, err := m.CreateModule(tenant, &store.Module{
+		Type:    "CDN_CIDRS",
+		Name:    "cdn-conditional",
+		Enabled: true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var gotIfNoneMatch string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotIfNoneMatch = strings.TrimSpace(r.Header.Get("If-None-Match"))
+		w.Header().Set("ETag", "etag-new")
+		_, _ = w.Write([]byte("198.51.100.0/24\n"))
+	}))
+	defer srv.Close()
+
+	etag := "etag-old"
+	if _, err := m.CreateCDNSource(tenant, mod.ID, &store.CDNSource{
+		SourceKind: "txt",
+		URL:        srv.URL,
+		Etag:       etag,
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	collected, err := collectModulePrefixRows(context.Background(), m, srv.Client(), tenant, mod, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if gotIfNoneMatch != etag {
+		t.Fatalf("want If-None-Match %q, got %q", etag, gotIfNoneMatch)
+	}
+	if len(collected) != 1 || collected[0].Prefix != "198.51.100.0/24" {
+		t.Fatalf("unexpected collected rows: %+v", collected)
+	}
+}
+
+func TestCollectModulePrefixRows_CDN304UsesSnapshot(t *testing.T) {
+	m := store.NewMemory()
+	m.SeedDemo()
+	tenant, _, _, _, _ := m.DemoIDs()
+
+	mod, err := m.CreateModule(tenant, &store.Module{
+		Type:    "CDN_CIDRS",
+		Name:    "cdn-304",
+		Enabled: true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusNotModified)
+	}))
+	defer srv.Close()
+
+	etag := "etag-stable"
+	src, err := m.CreateCDNSource(tenant, mod.ID, &store.CDNSource{
+		SourceKind: "txt",
+		URL:        srv.URL,
+		Etag:       etag,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	prior := []store.PrefixRow{{
+		Prefix: "203.0.113.0/24",
+		Source: cdnSourceKey(src.ID),
+	}}
+	if err := mergeCDNSourceIntoModuleSnapshot(m, tenant, mod, src.ID, prior); err != nil {
+		t.Fatal(err)
+	}
+
+	collected, err := collectModulePrefixRows(context.Background(), m, srv.Client(), tenant, mod, prior)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(collected) != 1 || collected[0].Prefix != "203.0.113.0/24" {
+		t.Fatalf("want cached prefix on 304, got %+v", collected)
+	}
+}

internal/pipeline/cdn_snapshot.go

@@ -0,0 +1,126 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	"evobgp/internal/store"
+)
+
+func cdnSourceKey(sourceID string) string {
+	return "cdn:" + strings.TrimSpace(sourceID)
+}
+
+func mergeSnapshotDropSource(rows []store.PrefixRow, sourceKey string) []store.PrefixRow {
+	if len(rows) == 0 {
+		return nil
+	}
+	out := make([]store.PrefixRow, 0, len(rows))
+	for _, row := range rows {
+		if row.Source != sourceKey {
+			out = append(out, row)
+		}
+	}
+	return out
+}
+
+func cdnRowsFromParsed(mod *store.Module, src *store.CDNSource, pfxStrings []string) []store.PrefixRow {
+	var rows []store.PrefixRow
+	for _, p := range pfxStrings {
+		p = strings.TrimSpace(p)
+		if p == "" {
+			continue
+		}
+		comm := src.CommunityID
+		if comm == nil && mod != nil && mod.DefaultCommunityID != nil {
+			c := *mod.DefaultCommunityID
+			comm = &c
+		}
+		rows = append(rows, store.PrefixRow{Prefix: p, CommunityID: comm, Source: cdnSourceKey(src.ID)})
+	}
+	return rows
+}
+
+// mergeCDNSourceIntoModuleSnapshot replaces rows for one CDN source in the module snapshot.
+func mergeCDNSourceIntoModuleSnapshot(st store.Backend, tenantID string, mod *store.Module, sourceID string, newRows []store.PrefixRow) error {
+	if st == nil || mod == nil {
+		return nil
+	}
+	sourceKey := cdnSourceKey(sourceID)
+	var base []store.PrefixRow
+	if snap, ok, _ := st.GetModulePrefixSnapshot(tenantID, mod.ID); ok && snap != nil {
+		base = mergeSnapshotDropSource(snap.Prefixes, sourceKey)
+	}
+	merged := append(base, newRows...)
+	return persistModuleSnapshot(st, tenantID, mod, merged)
+}
+
+func parseCDNBody(body string, src *store.CDNSource) ([]string, error) {
+	pfxs, err := ExtractCIDRs(body, src.SourceKind, src.PrefixPath)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]string, 0, len(pfxs))
+	for _, pfx := range pfxs {
+		out = append(out, pfx.String())
+	}
+	return out, nil
+}
+
+func applyCDNSourceHTTPResult(ctx context.Context, st store.Backend, hc *http.Client, tenantID, moduleID string, mod *store.Module, src *store.CDNSource, priorSnapshot []store.PrefixRow, now time.Time) ([]store.PrefixRow, error) {
+	u := strings.TrimSpace(src.URL)
+	if u == "" {
+		return nil, nil
+	}
+	sourceKey := cdnSourceKey(src.ID)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
+	if err != nil {
+		return nil, err
+	}
+	if etag := strings.TrimSpace(src.Etag); etag != "" {
+		req.Header.Set("If-None-Match", etag)
+	}
+	resp, err := hc.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode == http.StatusNotModified {
+		if cached := prefixRowsForSource(priorSnapshot, sourceKey); len(cached) > 0 {
+			return cached, nil
+		}
+		return nil, fmt.Errorf("cdn url %s: 304 without cached prefixes", u)
+	}
+	if resp.StatusCode != http.StatusOK {
+		_, _ = io.Copy(io.Discard, resp.Body)
+		return nil, fmt.Errorf("cdn url %s: %s", u, resp.Status)
+	}
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
+	if err != nil {
+		return nil, err
+	}
+	prefixStrs, err := parseCDNBody(string(body), src)
+	if err != nil {
+		return nil, fmt.Errorf("cdn parse %s: %w", u, err)
+	}
+	etag := strings.TrimSpace(resp.Header.Get("ETag"))
+	patch := &store.CDNSourcePatch{}
+	if etag != "" && etag != strings.TrimSpace(src.Etag) {
+		e := etag
+		patch.Etag = &e
+	}
+	refreshedAt := now
+	patch.LastRefreshedAt = &refreshedAt
+	_, _ = st.UpdateCDNSource(tenantID, moduleID, src.ID, patch)
+
+	rows := cdnRowsFromParsed(mod, src, prefixStrs)
+	if err := mergeCDNSourceIntoModuleSnapshot(st, tenantID, mod, src.ID, rows); err != nil {
+		return nil, err
+	}
+	return rows, nil
+}

internal/pipeline/collect_parallel.go

@@ -3,14 +3,12 @@ package pipeline
 import (
 	"context"
 	"fmt"
-	"io"
 	"net/http"
 	"os"
 	"strings"
 	"sync"
 	"time"
 
-	"evobgp/internal/asnresolve"
 	"evobgp/internal/store"
 )
 
@@ -76,15 +74,11 @@ func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client,
 				c := *mod.DefaultCommunityID
 				comm = &c
 			}
-			pfxs, err := asnresolve.AnnouncedPrefixes(ctx, hc, entry.ASN)
+			pfxs, holder, err := resolveASNForEntry(ctx, st, hc, entry.ASN)
 			if err != nil {
 				results[idx] = entryResult{err: fmt.Errorf("resolve AS%d: %w", entry.ASN, err)}
 				return
 			}
-			holder := ""
-			if h, err := asnresolve.ASHolderName(ctx, hc, entry.ASN); err == nil {
-				holder = h
-			}
 			src := fmt.Sprintf("as:%d", entry.ASN)
 			var rows []store.PrefixRow
 			for _, pfx := range pfxs {
@@ -161,18 +155,11 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client
 					return
 				}
 			}
-			u := strings.TrimSpace(src.URL)
-			if u == "" {
-				return
-			}
-			rows, err := fetchAndParseCDNSource(ctx, hc, st, tenantID, moduleID, mod, src, now)
+			rows, err := applyCDNSourceHTTPResult(ctx, st, hc, tenantID, moduleID, mod, src, priorSnapshot, now)
 			if err != nil {
 				results[idx] = srcResult{err: err}
 				return
 			}
-			for i := range rows {
-				rows[i].Source = sourceKey
-			}
 			results[idx] = srcResult{rows: rows}
 		}(i, src)
 	}
@@ -188,51 +175,6 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client
 	return out, nil
 }
 
-func fetchAndParseCDNSource(ctx context.Context, hc *http.Client, st store.Backend, tenantID, moduleID string, mod *store.Module, src *store.CDNSource, now time.Time) ([]store.PrefixRow, error) {
-	u := strings.TrimSpace(src.URL)
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
-	if err != nil {
-		return nil, err
-	}
-	resp, err := hc.Do(req)
-	if err != nil {
-		return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
-	}
-	if resp.StatusCode != http.StatusOK {
-		_, _ = io.Copy(io.Discard, resp.Body)
-		_ = resp.Body.Close()
-		return nil, fmt.Errorf("cdn url %s: %s", u, resp.Status)
-	}
-	body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
-	_ = resp.Body.Close()
-	if err != nil {
-		return nil, err
-	}
-	etag := strings.TrimSpace(resp.Header.Get("ETag"))
-	patch := &store.CDNSourcePatch{}
-	if etag != "" && etag != strings.TrimSpace(src.Etag) {
-		e := etag
-		patch.Etag = &e
-	}
-	refreshedAt := now
-	patch.LastRefreshedAt = &refreshedAt
-	_, _ = st.UpdateCDNSource(tenantID, moduleID, src.ID, patch)
-	pfxs, err := ExtractCIDRs(string(body), src.SourceKind, src.PrefixPath)
-	if err != nil {
-		return nil, fmt.Errorf("cdn parse %s: %w", u, err)
-	}
-	var rows []store.PrefixRow
-	for _, pfx := range pfxs {
-		comm := src.CommunityID
-		if comm == nil && mod.DefaultCommunityID != nil {
-			c := *mod.DefaultCommunityID
-			comm = &c
-		}
-		rows = append(rows, store.PrefixRow{Prefix: pfx.String(), CommunityID: comm})
-	}
-	return rows, nil
-}
-
 func collectDomainPrefixRows(ctx context.Context, hc *http.Client, mod *store.Module, profile *store.DohProfile, entries []*store.DomainEntry) ([]store.PrefixRow, error) {
 	var validDom []*store.DomainEntry
 	for _, e := range entries {

internal/pipeline/materialize_regression_test.go

@@ -1,9 +1,6 @@
 package pipeline
 
 import (
-	"context"
-	"net/http"
-	"net/http/httptest"
 	"strings"
 	"testing"
 
@@ -50,45 +47,3 @@ func TestBuildPreviewFragments_SamePrefixDifferentCommunity(t *testing.T) {
 	}
 }
 
-func TestCollectModulePrefixRows_CDNRefreshForcesFullGet(t *testing.T) {
-	m := store.NewMemory()
-	m.SeedDemo()
-	tenant, _, _, _, _ := m.DemoIDs()
-
-	mod, err := m.CreateModule(tenant, &store.Module{
-		Type:    "CDN_CIDRS",
-		Name:    "cdn-refresh-force-full",
-		Enabled: true,
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	var gotIfNoneMatch string
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		gotIfNoneMatch = strings.TrimSpace(r.Header.Get("If-None-Match"))
-		w.Header().Set("ETag", "etag-new")
-		_, _ = w.Write([]byte("198.51.100.0/24\n"))
-	}))
-	defer srv.Close()
-
-	etag := "etag-old"
-	if _, err := m.CreateCDNSource(tenant, mod.ID, &store.CDNSource{
-		SourceKind: "txt",
-		URL:        srv.URL,
-		Etag:       etag,
-	}); err != nil {
-		t.Fatal(err)
-	}
-
-	collected, err := collectModulePrefixRows(context.Background(), m, srv.Client(), tenant, mod, nil)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if gotIfNoneMatch != "" {
-		t.Fatalf("refresh path must not send If-None-Match, got %q", gotIfNoneMatch)
-	}
-	if len(collected) != 1 || collected[0].Prefix != "198.51.100.0/24" {
-		t.Fatalf("unexpected collected rows: %+v", collected)
-	}
-}

internal/pipeline/prefetch.go

@@ -5,11 +5,12 @@ import (
 	"io"
 	"net/http"
 	"strings"
+	"time"
 
 	"evobgp/internal/store"
 )
 
-// PrefetchCDNSourceETags performs conditional GETs for CDN module sources and updates stored ETags when the origin responds 200.
+// 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 = http.DefaultClient
@@ -18,42 +19,70 @@ func PrefetchCDNSourceETags(ctx context.Context, st store.Backend, hc *http.Clie
 	if err != nil {
 		return err
 	}
+	now := time.Now().UTC()
 	for _, tid := range tenants {
 		for _, mod := range st.ListModules(tid) {
-			if !mod.Enabled || mod.Type != "CDN_CIDRS" {
+			if mod == nil || !mod.Enabled || mod.Type != "CDN_CIDRS" {
+				continue
+			}
+			omod, err := st.GetModule(tid, mod.ID)
+			if err != nil {
 				continue
 			}
 			sources, err := st.ListCDNSources(tid, mod.ID)
 			if err != nil {
 				continue
 			}
+			var prior []store.PrefixRow
+			if snap, ok, _ := st.GetModulePrefixSnapshot(tid, mod.ID); ok && snap != nil {
+				prior = snap.Prefixes
+			}
 			for _, src := range sources {
-				u := strings.TrimSpace(src.URL)
-				if u == "" {
+				if src == nil || strings.TrimSpace(src.URL) == "" {
 					continue
 				}
-				req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
+				req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimSpace(src.URL), nil)
 				if err != nil {
 					continue
 				}
-				if strings.TrimSpace(src.Etag) != "" {
-					req.Header.Set("If-None-Match", strings.TrimSpace(src.Etag))
+				if etag := strings.TrimSpace(src.Etag); etag != "" {
+					req.Header.Set("If-None-Match", etag)
 				}
 				resp, err := hc.Do(req)
 				if err != nil {
 					continue
 				}
-				_, _ = io.Copy(io.Discard, resp.Body)
-				_ = resp.Body.Close()
+				if resp.StatusCode == http.StatusNotModified {
+					_ = resp.Body.Close()
+					continue
+				}
 				if resp.StatusCode != http.StatusOK {
+					_, _ = io.Copy(io.Discard, resp.Body)
+					_ = resp.Body.Close()
 					continue
 				}
-				etag := strings.TrimSpace(resp.Header.Get("ETag"))
-				if etag == "" || etag == strings.TrimSpace(src.Etag) {
+				body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
+				_ = resp.Body.Close()
+				if err != nil {
 					continue
 				}
-				e := etag
-				_, _ = st.UpdateCDNSource(tid, mod.ID, src.ID, &store.CDNSourcePatch{Etag: &e})
+				prefixStrs, err := parseCDNBody(string(body), src)
+				if err != nil {
+					continue
+				}
+				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)
+				_ = prior // prior may be stale after merge; refresh for next source in loop
+				if snap, ok, _ := st.GetModulePrefixSnapshot(tid, mod.ID); ok && snap != nil {
+					prior = snap.Prefixes
+				}
 			}
 		}
 	}

internal/pipeline/scheduler_due.go

@@ -0,0 +1,24 @@
+package pipeline
+
+import (
+	"time"
+
+	"evobgp/internal/store"
+)
+
+// SchedulerTickSec matches the scheduler ticker interval (internal/scheduler).
+const SchedulerTickSec = 30
+
+// ModuleDueForScheduler reports whether a module's refresh interval bucket rolled since the last scheduler tick.
+func ModuleDueForScheduler(mod *store.Module, now time.Time) bool {
+	if mod == nil || !mod.Enabled || mod.Type == "IP_RANGES" || mod.RefreshIntervalSec <= 0 {
+		return false
+	}
+	win := int64(mod.RefreshIntervalSec)
+	if win < 60 {
+		win = 60
+	}
+	cur := now.Unix() / win
+	prev := (now.Unix() - SchedulerTickSec) / win
+	return cur != prev
+}

internal/pipeline/tenant_refresh.go

@@ -0,0 +1,80 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"strings"
+	"sync"
+
+	"evobgp/internal/store"
+)
+
+// RefreshTenantModules ingests all listed modules in parallel and updates per-module snapshots.
+func RefreshTenantModules(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, moduleIDs []string) error {
+	if hc == nil {
+		hc = http.DefaultClient
+	}
+	var ids []string
+	seen := make(map[string]struct{})
+	for _, id := range moduleIDs {
+		id = strings.TrimSpace(id)
+		if id == "" {
+			continue
+		}
+		if _, ok := seen[id]; ok {
+			continue
+		}
+		seen[id] = struct{}{}
+		ids = append(ids, id)
+	}
+	if len(ids) == 0 {
+		return nil
+	}
+	if len(ids) == 1 {
+		return RefreshModuleIngest(ctx, st, hc, tenantID, ids[0])
+	}
+
+	sem := make(chan struct{}, collectConcurrency())
+	errs := make([]error, len(ids))
+	var wg sync.WaitGroup
+	for i, mid := range ids {
+		wg.Add(1)
+		go func(idx int, moduleID string) {
+			defer wg.Done()
+			sem <- struct{}{}
+			defer func() { <-sem }()
+			errs[idx] = RefreshModuleIngest(ctx, st, hc, tenantID, moduleID)
+		}(i, mid)
+	}
+	wg.Wait()
+	for _, err := range errs {
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// PickTenantRefreshTriggerModule returns a module id for revision metadata (first enabled due module).
+func PickTenantRefreshTriggerModule(st store.Backend, tenantID string, moduleIDs []string) (string, error) {
+	for _, id := range moduleIDs {
+		id = strings.TrimSpace(id)
+		if id == "" {
+			continue
+		}
+		mod, err := st.GetModule(tenantID, id)
+		if err != nil {
+			return "", err
+		}
+		if mod.Enabled {
+			return mod.ID, nil
+		}
+	}
+	for _, mod := range st.ListModules(tenantID) {
+		if mod != nil && mod.Enabled {
+			return mod.ID, nil
+		}
+	}
+	return "", fmt.Errorf("no enabled module for tenant refresh")
+}

internal/repository/asn_cache.go

@@ -0,0 +1,55 @@
+package repository
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"evobgp/internal/store"
+
+	"github.com/jackc/pgx/v5"
+)
+
+func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, bool, error) {
+	ctx := context.Background()
+	var holder string
+	var fetchedAt time.Time
+	var raw []byte
+	err := p.pool.QueryRow(ctx, `
+		SELECT holder, fetched_at, prefixes_json FROM asn_prefix_cache WHERE asn = $1`, asn).
+		Scan(&holder, &fetchedAt, &raw)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, false, nil
+		}
+		return nil, false, err
+	}
+	var prefixes []string
+	if len(raw) > 0 {
+		_ = json.Unmarshal(raw, &prefixes)
+	}
+	return &store.ASNPrefixCacheEntry{
+		ASN:       asn,
+		Holder:    holder,
+		Prefixes:  prefixes,
+		FetchedAt: fetchedAt.UTC(),
+	}, true, nil
+}
+
+func (p *Postgres) SetASNPrefixCache(asn int64, holder string, prefixes []string) error {
+	raw, err := json.Marshal(prefixes)
+	if err != nil {
+		return err
+	}
+	ctx := context.Background()
+	_, err = p.pool.Exec(ctx, `
+		INSERT INTO asn_prefix_cache (asn, holder, prefixes_json, fetched_at)
+		VALUES ($1, $2, $3::jsonb, now())
+		ON CONFLICT (asn) DO UPDATE SET
+			holder = EXCLUDED.holder,
+			prefixes_json = EXCLUDED.prefixes_json,
+			fetched_at = EXCLUDED.fetched_at`,
+		asn, holder, string(raw))
+	return err
+}

internal/scheduler/run.go

@@ -1,8 +1,10 @@
-// Package scheduler drives module refresh intervals and enqueues module_refresh jobs on the shared Registry.
+// Package scheduler drives module refresh intervals and enqueues tenant_refresh jobs on the shared Registry.
 package scheduler
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
 	"log"
@@ -13,6 +15,7 @@ import (
 	"evobgp/internal/broker"
 	"evobgp/internal/config"
 	"evobgp/internal/jobs"
+	"evobgp/internal/pipeline"
 	"evobgp/internal/store"
 )
 
@@ -39,9 +42,9 @@ func Run(ctx context.Context, deps *Deps) {
 	t := time.NewTicker(30 * time.Second)
 	defer t.Stop()
 	if deps.Jobs != nil {
-		log.Printf("evobgp-scheduler: active (in-process enqueue module_refresh)")
+		log.Printf("evobgp-scheduler: active (in-process enqueue tenant_refresh)")
 	} else {
-		log.Printf("evobgp-scheduler: active (HTTP POST .../modules/{id}/refresh → %s)", strings.TrimSpace(deps.APIBase))
+		log.Printf("evobgp-scheduler: active (HTTP POST .../tenant/refresh → %s)", strings.TrimSpace(deps.APIBase))
 	}
 	for {
 		select {
@@ -60,51 +63,53 @@ func tick(ctx context.Context, deps *Deps) {
 		log.Printf("evobgp-scheduler: list tenants: %v", err)
 		return
 	}
+	now := time.Now().UTC()
+	tickBucket := now.Unix() / pipeline.SchedulerTickSec
 	for _, tid := range tenants {
+		var due []string
 		for _, mod := range deps.Store.ListModules(tid) {
-			if !mod.Enabled || mod.Type == "IP_RANGES" {
+			if pipeline.ModuleDueForScheduler(mod, now) {
+				due = append(due, mod.ID)
+			}
+		}
+		if len(due) == 0 {
+			continue
+		}
+		key := fmt.Sprintf("sched-tenant-%s-%d", tid, tickBucket)
+		if deps.Jobs != nil {
+			ids := append([]string(nil), due...)
+			_, created, err := deps.Jobs.Enqueue(tid, jobs.KindTenantRefresh, &key, nil, map[string]any{
+				"module_ids": ids,
+				"trigger":    "scheduler",
+			})
+			if err != nil {
+				log.Printf("evobgp-scheduler: enqueue tenant %s: %v", tid, err)
 				continue
 			}
-			interval := mod.RefreshIntervalSec
-			if interval <= 0 {
-				continue
-			}
-			win := interval
-			if win < 60 {
-				win = 60
-			}
-			bucket := time.Now().Unix() / int64(win)
-			key := fmt.Sprintf("sched-%s-%d", mod.ID, bucket)
-			if deps.Jobs != nil {
-				mid := mod.ID
-				_, created, err := deps.Jobs.Enqueue(tid, jobs.KindModuleRefresh, &key, &mid, map[string]any{
-					"module_id": mod.ID,
-					"trigger":   "scheduler",
-				})
-				if err != nil {
-					log.Printf("evobgp-scheduler: enqueue module %s: %v", mod.ID, err)
-					continue
-				}
-				if created {
-					log.Printf("evobgp-scheduler: queued refresh for module %s (%s)", mod.ID, mod.Type)
-				}
-				continue
-			}
-			if err := postModuleRefresh(ctx, deps, mod.ID, key); err != nil {
-				log.Printf("evobgp-scheduler: http refresh module %s: %v", mod.ID, err)
+			if created {
+				log.Printf("evobgp-scheduler: queued tenant refresh for %d module(s) in tenant %s", len(due), tid)
 			}
+			continue
+		}
+		if err := postTenantRefresh(ctx, deps, due, key); err != nil {
+			log.Printf("evobgp-scheduler: http tenant refresh %s: %v", tid, err)
 		}
 	}
 }
 
-func postModuleRefresh(ctx context.Context, deps *Deps, moduleID, idempotencyKey string) error {
+func postTenantRefresh(ctx context.Context, deps *Deps, moduleIDs []string, idempotencyKey string) error {
 	base := strings.TrimRight(strings.TrimSpace(deps.APIBase), "/")
-	u := base + "/v1/modules/" + moduleID + "/refresh"
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, nil)
+	u := base + "/v1/tenant/refresh"
+	body, err := json.Marshal(map[string]any{"module_ids": moduleIDs})
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewReader(body))
 	if err != nil {
 		return err
 	}
 	req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(deps.APIToken))
+	req.Header.Set("Content-Type", "application/json")
 	if idempotencyKey != "" {
 		req.Header.Set("Idempotency-Key", idempotencyKey)
 	}

internal/store/backend.go

@@ -84,6 +84,18 @@ type Backend interface {
 	GetModulePrefixSnapshot(tenantID, moduleID string) (*ModulePrefixSnapshot, bool, error)
 	SetModulePrefixSnapshot(tenantID, moduleID, inputHash string, prefixes []PrefixRow) error
 	DeleteModulePrefixSnapshot(tenantID, moduleID string) error
+
+	// ASNPrefixCache stores RIPEstat announced-prefixes per ASN (global TTL cache).
+	GetASNPrefixCache(asn int64) (*ASNPrefixCacheEntry, bool, error)
+	SetASNPrefixCache(asn int64, holder string, prefixes []string) error
+}
+
+// ASNPrefixCacheEntry is a cached RIPEstat response for one ASN.
+type ASNPrefixCacheEntry struct {
+	ASN       int64
+	Holder    string
+	Prefixes  []string
+	FetchedAt time.Time
 }
 
 // ModulePrefixSnapshot is the cached materialization for one module between refreshes.

internal/store/memory.go

@@ -41,6 +41,7 @@ type Memory struct {
 	settings    map[string]map[string]any // tenantID -> key -> JSON-compatible value
 	revPrefixes     map[string][]PrefixRow
 	moduleSnapshots map[string]*moduleSnapshotRec
+	asnPrefixCache  map[int64]*ASNPrefixCacheEntry
 
 	// DemoIDs valid after SeedDemo()
 	demoTenantID   string
@@ -128,6 +129,7 @@ func NewMemory() *Memory {
 		settings:          make(map[string]map[string]any),
 		revPrefixes:       make(map[string][]PrefixRow),
 		moduleSnapshots:   make(map[string]*moduleSnapshotRec),
+		asnPrefixCache:    make(map[int64]*ASNPrefixCacheEntry),
 	}
 }
 

internal/store/memory_asn_cache.go

@@ -0,0 +1,40 @@
+package store
+
+import (
+	"time"
+)
+
+func (m *Memory) GetASNPrefixCache(asn int64) (*ASNPrefixCacheEntry, bool, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	if m.asnPrefixCache == nil {
+		return nil, false, nil
+	}
+	e, ok := m.asnPrefixCache[asn]
+	if !ok || e == nil {
+		return nil, false, nil
+	}
+	pfx := append([]string(nil), e.Prefixes...)
+	return &ASNPrefixCacheEntry{
+		ASN:       asn,
+		Holder:    e.Holder,
+		Prefixes:  pfx,
+		FetchedAt: e.FetchedAt,
+	}, true, nil
+}
+
+func (m *Memory) SetASNPrefixCache(asn int64, holder string, prefixes []string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.asnPrefixCache == nil {
+		m.asnPrefixCache = make(map[int64]*ASNPrefixCacheEntry)
+	}
+	cp := append([]string(nil), prefixes...)
+	m.asnPrefixCache[asn] = &ASNPrefixCacheEntry{
+		ASN:       asn,
+		Holder:    holder,
+		Prefixes:  cp,
+		FetchedAt: time.Now().UTC(),
+	}
+	return nil
+}

migrations/postgres/000010_asn_prefix_cache.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS asn_prefix_cache;

migrations/postgres/000010_asn_prefix_cache.up.sql

@@ -0,0 +1,9 @@
+-- TTL cache for RIPEstat announced-prefixes per ASN (pipeline ingest).
+CREATE TABLE asn_prefix_cache (
+    asn           BIGINT PRIMARY KEY,
+    holder        TEXT NOT NULL DEFAULT '',
+    prefixes_json JSONB NOT NULL DEFAULT '[]',
+    fetched_at    TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX idx_asn_prefix_cache_fetched ON asn_prefix_cache (fetched_at);

migrations/sqlite/000010_asn_prefix_cache.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS asn_prefix_cache;

migrations/sqlite/000010_asn_prefix_cache.up.sql

@@ -0,0 +1,8 @@
+CREATE TABLE asn_prefix_cache (
+    asn           INTEGER PRIMARY KEY,
+    holder        TEXT NOT NULL DEFAULT '',
+    prefixes_json TEXT NOT NULL DEFAULT '[]',
+    fetched_at    TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+);
+
+CREATE INDEX idx_asn_prefix_cache_fetched ON asn_prefix_cache (fetched_at);