feat: update collectModulePrefixRows to support prior snapshots and enhance CDN/domain prefix collection

Modified the collectModulePrefixRows function to accept an optional priorSnapshot parameter, allowing for more efficient data retrieval by skipping unnecessary CDN fetches. Refactored the logic for collecting prefix rows from AS, CDN, and domain sources to utilize dedicated functions, improving code organization and maintainability. Additionally, introduced caching for module prefix snapshots to optimize performance during refresh operations.

internal/pipeline/aggregate.go

@@ -0,0 +1,106 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+
+	"evobgp/internal/store"
+)
+
+func collectConcurrency() int {
+	n := 8
+	if s := strings.TrimSpace(os.Getenv("EVOBGP_COLLECT_CONCURRENCY")); s != "" {
+		if v, err := strconv.Atoi(s); err == nil && v > 0 {
+			n = v
+		}
+	}
+	if n > 32 {
+		n = 32
+	}
+	return n
+}
+
+// aggregateTenantPrefixRowsAll builds the union of materialized prefixes for all enabled modules.
+// Uses per-module snapshots when inputs are unchanged to avoid duplicate external fetches on render.
+func aggregateTenantPrefixRowsAll(ctx context.Context, st store.Backend, hc *http.Client, tenantID string) ([]store.PrefixRow, error) {
+	mods := st.ListModules(tenantID)
+	type modRef struct {
+		id string
+	}
+	var enabled []modRef
+	for _, m := range mods {
+		if m != nil && m.Enabled {
+			enabled = append(enabled, modRef{id: m.ID})
+		}
+	}
+	if len(enabled) == 0 {
+		return nil, nil
+	}
+	if len(enabled) == 1 {
+		omod, err := st.GetModule(tenantID, enabled[0].id)
+		if err != nil {
+			return nil, err
+		}
+		return rowsForModule(ctx, st, hc, tenantID, omod)
+	}
+
+	sem := make(chan struct{}, collectConcurrency())
+	results := make([][]store.PrefixRow, len(enabled))
+	errs := make([]error, len(enabled))
+	var wg sync.WaitGroup
+	for i, ref := range enabled {
+		wg.Add(1)
+		go func(idx int, moduleID string) {
+			defer wg.Done()
+			sem <- struct{}{}
+			defer func() { <-sem }()
+			omod, err := st.GetModule(tenantID, moduleID)
+			if err != nil {
+				errs[idx] = err
+				return
+			}
+			rows, err := rowsForModule(ctx, st, hc, tenantID, omod)
+			if err != nil {
+				errs[idx] = fmt.Errorf("module %s: %w", moduleID, err)
+				return
+			}
+			results[idx] = rows
+		}(i, ref.id)
+	}
+	wg.Wait()
+	for _, err := range errs {
+		if err != nil {
+			return nil, err
+		}
+	}
+	var out []store.PrefixRow
+	for _, part := range results {
+		out = append(out, part...)
+	}
+	return out, nil
+}
+
+func rowsForModule(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, mod *store.Module) ([]store.PrefixRow, error) {
+	if rows, ok, err := moduleRowsFromSnapshot(st, tenantID, mod); err != nil {
+		return nil, err
+	} else if ok {
+		return rows, nil
+	}
+	var prior []store.PrefixRow
+	if snap, ok, _ := st.GetModulePrefixSnapshot(tenantID, mod.ID); ok && snap != nil {
+		prior = snap.Prefixes
+	}
+	rows, err := collectModulePrefixRows(ctx, st, hc, tenantID, mod, prior)
+	if err != nil {
+		return nil, err
+	}
+	if err := persistModuleSnapshot(st, tenantID, mod, rows); err != nil {
+		return nil, err
+	}
+	return rows, nil
+}

internal/pipeline/collect_parallel.go

@@ -0,0 +1,302 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"evobgp/internal/asnresolve"
+	"evobgp/internal/store"
+)
+
+func prefixRowsForSource(rows []store.PrefixRow, sourceKey string) []store.PrefixRow {
+	if len(rows) == 0 {
+		return nil
+	}
+	var out []store.PrefixRow
+	for _, row := range rows {
+		if row.Source == sourceKey {
+			out = append(out, row)
+		}
+	}
+	return out
+}
+
+func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, mod *store.Module, list []*store.ASEntry) ([]store.PrefixRow, error) {
+	moduleID := mod.ID
+	legacy := strings.TrimSpace(os.Getenv("EVOBGP_ASN_RESOLVE")) == "0"
+	if legacy {
+		var rows []store.PrefixRow
+		for _, e := range list {
+			if !store.ValidASN(e.ASN) {
+				continue
+			}
+			comm := e.CommunityID
+			if comm == nil && mod.DefaultCommunityID != nil {
+				c := *mod.DefaultCommunityID
+				comm = &c
+			}
+			rows = append(rows, store.PrefixRow{Prefix: MaterializedASPrefixKey(e.ASN), CommunityID: comm, Source: "as_entry"})
+		}
+		return rows, nil
+	}
+
+	type entryResult struct {
+		rows   []store.PrefixRow
+		metaID string
+		asn    int64
+		holder string
+		count  int64
+		err    error
+	}
+
+	var valid []*store.ASEntry
+	for _, e := range list {
+		if e != nil && store.ValidASN(e.ASN) {
+			valid = append(valid, e)
+		}
+	}
+	sem := make(chan struct{}, collectConcurrency())
+	results := make([]entryResult, len(valid))
+	var wg sync.WaitGroup
+	for i, e := range valid {
+		wg.Add(1)
+		go func(idx int, entry *store.ASEntry) {
+			defer wg.Done()
+			sem <- struct{}{}
+			defer func() { <-sem }()
+
+			comm := entry.CommunityID
+			if comm == nil && mod.DefaultCommunityID != nil {
+				c := *mod.DefaultCommunityID
+				comm = &c
+			}
+			pfxs, err := asnresolve.AnnouncedPrefixes(ctx, 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 {
+				rows = append(rows, store.PrefixRow{Prefix: pfx.String(), CommunityID: comm, Source: src})
+			}
+			results[idx] = entryResult{
+				rows:   rows,
+				metaID: entry.ID,
+				asn:    entry.ASN,
+				holder: holder,
+				count:  int64(len(pfxs)),
+			}
+		}(i, e)
+	}
+	wg.Wait()
+
+	seenPfx := make(map[string]struct{})
+	var out []store.PrefixRow
+	now := time.Now().UTC()
+	for _, r := range results {
+		if r.err != nil {
+			return nil, r.err
+		}
+		if r.metaID != "" {
+			if err := st.UpdateASEntryResolveMeta(tenantID, moduleID, r.metaID, r.holder, r.count, now); err != nil {
+				return nil, fmt.Errorf("as entry meta AS%d: %w", r.asn, err)
+			}
+		}
+		for _, row := range r.rows {
+			k := row.Prefix
+			if _, ok := seenPfx[k]; ok {
+				continue
+			}
+			seenPfx[k] = struct{}{}
+			out = append(out, row)
+		}
+	}
+	return out, nil
+}
+
+func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, mod *store.Module, sources []*store.CDNSource, priorSnapshot []store.PrefixRow) ([]store.PrefixRow, error) {
+	moduleID := mod.ID
+	now := time.Now().UTC()
+
+	var valid []*store.CDNSource
+	for _, s := range sources {
+		if s != nil {
+			valid = append(valid, s)
+		}
+	}
+	type srcResult struct {
+		rows []store.PrefixRow
+		err  error
+	}
+	results := make([]srcResult, len(valid))
+	sem := make(chan struct{}, collectConcurrency())
+	var wg sync.WaitGroup
+
+	for i, src := range valid {
+		wg.Add(1)
+		go func(idx int, src *store.CDNSource) {
+			defer wg.Done()
+			sem <- struct{}{}
+			defer func() { <-sem }()
+
+			sourceKey := "cdn:" + src.ID
+			if shouldSkipCDNSourceFetch(src, now) {
+				if cached := prefixRowsForSource(priorSnapshot, sourceKey); len(cached) > 0 {
+					results[idx] = srcResult{rows: cached}
+					return
+				}
+				if cached := latestCDNRowsBySource(st, tenantID)[sourceKey]; len(cached) > 0 {
+					results[idx] = srcResult{rows: cached}
+					return
+				}
+			}
+			u := strings.TrimSpace(src.URL)
+			if u == "" {
+				return
+			}
+			rows, err := fetchAndParseCDNSource(ctx, hc, st, tenantID, moduleID, mod, src, 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)
+	}
+	wg.Wait()
+
+	var out []store.PrefixRow
+	for _, r := range results {
+		if r.err != nil {
+			return nil, r.err
+		}
+		out = append(out, r.rows...)
+	}
+	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 {
+		if e != nil {
+			validDom = append(validDom, e)
+		}
+	}
+	type domResult struct {
+		rows []store.PrefixRow
+		err  error
+	}
+	results := make([]domResult, len(validDom))
+	sem := make(chan struct{}, collectConcurrency())
+	var wg sync.WaitGroup
+
+	for i, e := range validDom {
+		wg.Add(1)
+		go func(idx int, entry *store.DomainEntry) {
+			defer wg.Done()
+			sem <- struct{}{}
+			defer func() { <-sem }()
+
+			comm := entry.CommunityID
+			if comm == nil && mod.DefaultCommunityID != nil {
+				c := *mod.DefaultCommunityID
+				comm = &c
+			}
+			addrs, err := resolveDomainIPs(ctx, hc, profile, entry.FQDN)
+			if err != nil {
+				results[idx] = domResult{err: fmt.Errorf("resolve domain %q: %w", entry.FQDN, err)}
+				return
+			}
+			src := "domain:" + strings.TrimSpace(entry.FQDN)
+			var rows []store.PrefixRow
+			for _, ip := range addrs {
+				cidr := ipToHostPrefix(ip)
+				if cidr == "" {
+					continue
+				}
+				rows = append(rows, store.PrefixRow{
+					Prefix:      cidr,
+					CommunityID: comm,
+					Source:      src,
+				})
+			}
+			results[idx] = domResult{rows: rows}
+		}(i, e)
+	}
+	wg.Wait()
+
+	seen := make(map[string]struct{})
+	var out []store.PrefixRow
+	for _, r := range results {
+		if r.err != nil {
+			return nil, r.err
+		}
+		for _, row := range r.rows {
+			key := row.Prefix + "|" + row.Source
+			if _, ok := seen[key]; ok {
+				continue
+			}
+			seen[key] = struct{}{}
+			out = append(out, row)
+		}
+	}
+	return out, nil
+}

internal/pipeline/materialize_regression_test.go

@@ -81,7 +81,7 @@ func TestCollectModulePrefixRows_CDNRefreshForcesFullGet(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	collected, err := collectModulePrefixRows(context.Background(), m, srv.Client(), tenant, mod)
+	collected, err := collectModulePrefixRows(context.Background(), m, srv.Client(), tenant, mod, nil)
 	if err != nil {
 		t.Fatal(err)
 	}

internal/pipeline/module_hash.go

@@ -0,0 +1,123 @@
+package pipeline
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"sort"
+	"strings"
+
+	"evobgp/internal/store"
+)
+
+// moduleIngestInputHash fingerprints module config and child entries so snapshots invalidate on CRUD.
+func moduleIngestInputHash(st store.Backend, tenantID string, mod *store.Module) (string, error) {
+	if st == nil || mod == nil {
+		return "", fmt.Errorf("module hash: missing store or module")
+	}
+	h := sha256.New()
+	_, _ = fmt.Fprintf(h, "type=%s\n", strings.TrimSpace(mod.Type))
+	_, _ = fmt.Fprintf(h, "enabled=%t\n", mod.Enabled)
+	if mod.DefaultCommunityID != nil {
+		_, _ = fmt.Fprintf(h, "default_community=%s\n", strings.TrimSpace(*mod.DefaultCommunityID))
+	}
+	if mod.DohProfileID != nil {
+		_, _ = fmt.Fprintf(h, "doh_profile=%s\n", strings.TrimSpace(*mod.DohProfileID))
+		if pid := strings.TrimSpace(*mod.DohProfileID); pid != "" {
+			if prof, err := st.GetDohProfile(tenantID, pid); err == nil && prof != nil {
+				_, _ = fmt.Fprintf(h, "doh_url=%s\n", strings.TrimSpace(prof.URL))
+				if prof.TimeoutMs != nil {
+					_, _ = fmt.Fprintf(h, "doh_timeout=%d\n", *prof.TimeoutMs)
+				}
+			}
+		}
+	}
+
+	switch mod.Type {
+	case "IP_RANGES":
+		list, err := st.ListIPRangeEntries(tenantID, mod.ID)
+		if err != nil {
+			return "", err
+		}
+		sort.Slice(list, func(i, j int) bool { return list[i].Prefix < list[j].Prefix })
+		for _, e := range list {
+			comm := ""
+			if e.CommunityID != nil {
+				comm = *e.CommunityID
+			}
+			_, _ = fmt.Fprintf(h, "ip=%s|c=%s\n", e.Prefix, comm)
+		}
+	case "AS_PREFIXES":
+		list, err := st.ListASEntries(tenantID, mod.ID)
+		if err != nil {
+			return "", err
+		}
+		sort.Slice(list, func(i, j int) bool { return list[i].ASN < list[j].ASN })
+		for _, e := range list {
+			comm := ""
+			if e.CommunityID != nil {
+				comm = *e.CommunityID
+			}
+			_, _ = fmt.Fprintf(h, "as=%d|c=%s\n", e.ASN, comm)
+		}
+	case "CDN_CIDRS":
+		list, err := st.ListCDNSources(tenantID, mod.ID)
+		if err != nil {
+			return "", err
+		}
+		sort.Slice(list, func(i, j int) bool { return list[i].ID < list[j].ID })
+		for _, s := range list {
+			comm := ""
+			if s.CommunityID != nil {
+				comm = *s.CommunityID
+			}
+			interval := 0
+			if s.RefreshIntervalSec != nil {
+				interval = *s.RefreshIntervalSec
+			}
+			_, _ = fmt.Fprintf(h, "cdn=%s|url=%s|kind=%s|path=%s|c=%s|etag=%s|interval=%d\n",
+				s.ID, strings.TrimSpace(s.URL), s.SourceKind, strings.TrimSpace(s.PrefixPath), comm,
+				strings.TrimSpace(s.Etag), interval)
+		}
+	case "DOMAINS":
+		list, err := st.ListDomainEntries(tenantID, mod.ID)
+		if err != nil {
+			return "", err
+		}
+		sort.Slice(list, func(i, j int) bool { return list[i].FQDN < list[j].FQDN })
+		for _, e := range list {
+			comm := ""
+			if e.CommunityID != nil {
+				comm = *e.CommunityID
+			}
+			_, _ = fmt.Fprintf(h, "dom=%s|c=%s\n", strings.TrimSpace(e.FQDN), comm)
+		}
+	default:
+		_, _ = fmt.Fprintf(h, "unknown_type=%s\n", mod.Type)
+	}
+	return hex.EncodeToString(h.Sum(nil)), nil
+}
+
+func persistModuleSnapshot(st store.Backend, tenantID string, mod *store.Module, rows []store.PrefixRow) error {
+	if st == nil || mod == nil {
+		return nil
+	}
+	hash, err := moduleIngestInputHash(st, tenantID, mod)
+	if err != nil {
+		return err
+	}
+	return st.SetModulePrefixSnapshot(tenantID, mod.ID, hash, rows)
+}
+
+func moduleRowsFromSnapshot(st store.Backend, tenantID string, mod *store.Module) ([]store.PrefixRow, bool, error) {
+	hash, err := moduleIngestInputHash(st, tenantID, mod)
+	if err != nil {
+		return nil, false, err
+	}
+	snap, ok, err := st.GetModulePrefixSnapshot(tenantID, mod.ID)
+	if err != nil || !ok || snap == nil || snap.InputHash != hash {
+		return nil, false, err
+	}
+	cp := append([]store.PrefixRow(nil), snap.Prefixes...)
+	return cp, true, nil
+}

internal/pipeline/refresh.go

@@ -12,13 +12,11 @@ import (
 	"net/http"
 	"net/netip"
 	"net/url"
-	"os"
 	"sort"
 	"strconv"
 	"strings"
 	"time"
 
-	"evobgp/internal/asnresolve"
 	"evobgp/internal/birdfmt"
 	"evobgp/internal/store"
 
@@ -55,10 +53,13 @@ func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client,
 		return fmt.Errorf("module disabled")
 	}
 
-	_, err = collectModulePrefixRows(ctx, st, hc, tenantID, mod)
+	rows, err := collectModulePrefixRows(ctx, st, hc, tenantID, mod, nil)
 	if err != nil {
 		return err
 	}
+	if err := persistModuleSnapshot(st, tenantID, mod, rows); err != nil {
+		return err
+	}
 	refreshedAt := time.Now().UTC()
 	_, _ = st.UpdateModule(tenantID, moduleID, &store.ModulePatch{LastRefreshedAt: &refreshedAt})
 	return nil
@@ -129,7 +130,8 @@ func RefreshModule(ctx context.Context, st store.Backend, hc *http.Client, tenan
 }
 
 // collectModulePrefixRows returns materialized prefix rows for a single module (source of truth from store / ASN resolve / CDN fetch).
-func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, mod *store.Module) ([]store.PrefixRow, error) {
+// priorSnapshot is the last stored module snapshot (used to skip CDN fetches when refresh interval has not elapsed).
+func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, mod *store.Module, priorSnapshot []store.PrefixRow) ([]store.PrefixRow, error) {
 	moduleID := mod.ID
 	switch mod.Type {
 	case "IP_RANGES":
@@ -153,110 +155,13 @@ func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Cli
 			return nil, err
 		}
 		sort.Slice(list, func(i, j int) bool { return list[i].ASN < list[j].ASN })
-		legacy := strings.TrimSpace(os.Getenv("EVOBGP_ASN_RESOLVE")) == "0"
+		return collectASPrefixRows(ctx, st, hc, tenantID, mod, list)
-		seenPfx := make(map[string]struct{})
-		var rows []store.PrefixRow
-		for i, e := range list {
-			if !store.ValidASN(e.ASN) {
-				continue
-			}
-			comm := e.CommunityID
-			if comm == nil && mod.DefaultCommunityID != nil {
-				c := *mod.DefaultCommunityID
-				comm = &c
-			}
-			if legacy {
-				rows = append(rows, store.PrefixRow{Prefix: MaterializedASPrefixKey(e.ASN), CommunityID: comm, Source: "as_entry"})
-				continue
-			}
-			if i > 0 {
-				asnresolve.PolitePause()
-			}
-			pfxs, err := asnresolve.AnnouncedPrefixes(ctx, hc, e.ASN)
-			if err != nil {
-				return nil, fmt.Errorf("resolve AS%d: %w", e.ASN, err)
-			}
-			holder := ""
-			asnresolve.PolitePause()
-			if h, err := asnresolve.ASHolderName(ctx, hc, e.ASN); err == nil {
-				holder = h
-			}
-			now := time.Now().UTC()
-			if err := st.UpdateASEntryResolveMeta(tenantID, moduleID, e.ID, holder, int64(len(pfxs)), now); err != nil {
-				return nil, fmt.Errorf("as entry meta AS%d: %w", e.ASN, err)
-			}
-			src := fmt.Sprintf("as:%d", e.ASN)
-			for _, pfx := range pfxs {
-				k := pfx.String()
-				if _, ok := seenPfx[k]; ok {
-					continue
-				}
-				seenPfx[k] = struct{}{}
-				rows = append(rows, store.PrefixRow{Prefix: k, CommunityID: comm, Source: src})
-			}
-		}
-		return rows, nil
 	case "CDN_CIDRS":
 		sources, err := st.ListCDNSources(tenantID, moduleID)
 		if err != nil {
 			return nil, err
 		}
-		var rows []store.PrefixRow
+		return collectCDNPrefixRows(ctx, st, hc, tenantID, mod, sources, priorSnapshot)
-		latestCDNRows := latestCDNRowsBySource(st, tenantID)
-		for _, src := range sources {
-			sourceKey := "cdn:" + src.ID
-			now := time.Now().UTC()
-			if shouldSkipCDNSourceFetch(src, now) {
-				if cached := latestCDNRows[sourceKey]; len(cached) > 0 {
-					rows = append(rows, cached...)
-					continue
-				}
-			}
-			u := strings.TrimSpace(src.URL)
-			if u == "" {
-				continue
-			}
-			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)
-			}
-			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, Source: sourceKey})
-			}
-		}
-		return rows, nil
 	case "DOMAINS":
 		entries, err := st.ListDomainEntries(tenantID, moduleID)
 		if err != nil {
@@ -269,40 +174,7 @@ func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Cli
 				return nil, fmt.Errorf("get doh profile: %w", err)
 			}
 		}
-		var rows []store.PrefixRow
+		return collectDomainPrefixRows(ctx, hc, mod, profile, entries)
-		seen := make(map[string]struct{})
-		for _, e := range entries {
-			if e == nil {
-				continue
-			}
-			comm := e.CommunityID
-			if comm == nil && mod.DefaultCommunityID != nil {
-				c := *mod.DefaultCommunityID
-				comm = &c
-			}
-			addrs, err := resolveDomainIPs(ctx, hc, profile, e.FQDN)
-			if err != nil {
-				return nil, fmt.Errorf("resolve domain %q: %w", e.FQDN, err)
-			}
-			src := "domain:" + strings.TrimSpace(e.FQDN)
-			for _, ip := range addrs {
-				cidr := ipToHostPrefix(ip)
-				if cidr == "" {
-					continue
-				}
-				key := cidr + "|" + src
-				if _, ok := seen[key]; ok {
-					continue
-				}
-				seen[key] = struct{}{}
-				rows = append(rows, store.PrefixRow{
-					Prefix:      cidr,
-					CommunityID: comm,
-					Source:      src,
-				})
-			}
-		}
-		return rows, nil
 	default:
 		return nil, fmt.Errorf("unknown module type %q", mod.Type)
 	}
@@ -548,28 +420,6 @@ func ipToHostPrefix(ip netip.Addr) string {
 	return netip.PrefixFrom(ip, bits).Masked().String()
 }
 
-// aggregateTenantPrefixRowsAll builds the union of materialized prefixes for all enabled modules
-// using current source data from store/external resolvers.
-func aggregateTenantPrefixRowsAll(ctx context.Context, st store.Backend, hc *http.Client, tenantID string) ([]store.PrefixRow, error) {
-	mods := st.ListModules(tenantID)
-	var out []store.PrefixRow
-	for _, m := range mods {
-		if m == nil || !m.Enabled {
-			continue
-		}
-		omod, err := st.GetModule(tenantID, m.ID)
-		if err != nil {
-			return nil, err
-		}
-		rows, err := collectModulePrefixRows(ctx, st, hc, tenantID, omod)
-		if err != nil {
-			return nil, fmt.Errorf("module %s: %w", m.ID, err)
-		}
-		out = append(out, rows...)
-	}
-	return out, nil
-}
-
 type prefixGroupKey struct {
 	community string
 	source    string

internal/repository/module_snapshot.go

@@ -0,0 +1,70 @@
+package repository
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"strings"
+	"time"
+
+	"evobgp/internal/store"
+
+	"github.com/jackc/pgx/v5"
+)
+
+func (p *Postgres) GetModulePrefixSnapshot(tenantID, moduleID string) (*store.ModulePrefixSnapshot, bool, error) {
+	ctx := context.Background()
+	var inputHash string
+	var collectedAt time.Time
+	var raw []byte
+	err := p.pool.QueryRow(ctx, `
+		SELECT input_hash, collected_at, prefixes_json
+		FROM module_prefix_snapshot
+		WHERE tenant_id = $1 AND module_id = $2`,
+		tenantID, moduleID).Scan(&inputHash, &collectedAt, &raw)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, false, nil
+		}
+		return nil, false, err
+	}
+	var prefixes []store.PrefixRow
+	if len(raw) > 0 {
+		if err := json.Unmarshal(raw, &prefixes); err != nil {
+			return nil, false, err
+		}
+	}
+	return &store.ModulePrefixSnapshot{
+		InputHash:   inputHash,
+		CollectedAt: collectedAt.UTC(),
+		Prefixes:    prefixes,
+	}, true, nil
+}
+
+func (p *Postgres) SetModulePrefixSnapshot(tenantID, moduleID, inputHash string, prefixes []store.PrefixRow) error {
+	if strings.TrimSpace(tenantID) == "" || strings.TrimSpace(moduleID) == "" || strings.TrimSpace(inputHash) == "" {
+		return store.ErrInvalidInput
+	}
+	raw, err := json.Marshal(prefixes)
+	if err != nil {
+		return err
+	}
+	ctx := context.Background()
+	_, err = p.pool.Exec(ctx, `
+		INSERT INTO module_prefix_snapshot (tenant_id, module_id, input_hash, collected_at, prefixes_json)
+		VALUES ($1::uuid, $2::uuid, $3, now(), $4::jsonb)
+		ON CONFLICT (tenant_id, module_id) DO UPDATE SET
+			input_hash = EXCLUDED.input_hash,
+			collected_at = EXCLUDED.collected_at,
+			prefixes_json = EXCLUDED.prefixes_json`,
+		tenantID, moduleID, inputHash, string(raw))
+	return err
+}
+
+func (p *Postgres) DeleteModulePrefixSnapshot(tenantID, moduleID string) error {
+	ctx := context.Background()
+	_, err := p.pool.Exec(ctx, `
+		DELETE FROM module_prefix_snapshot WHERE tenant_id = $1 AND module_id = $2`,
+		tenantID, moduleID)
+	return err
+}

internal/repository/postgres.go

@@ -929,19 +929,22 @@ func (p *Postgres) CreateRenderRevision(revisionID, tenantID, moduleID string, p
 	if err != nil {
 		return err
 	}
-	for _, pr := range prefixes {
+	if len(prefixes) > 0 {
-		var comm any
+		_, err = tx.CopyFrom(ctx,
-		if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" {
+			pgx.Identifier{"revision_materialized_prefix"},
-			comm = strings.TrimSpace(*pr.CommunityID)
+			[]string{"revision_id", "prefix", "community_id", "source"},
-		}
+			pgx.CopyFromSlice(len(prefixes), func(i int) ([]any, error) {
-		src := pr.Source
+				pr := prefixes[i]
-		if strings.TrimSpace(src) == "" {
+				var comm any
-			src = "render"
+				if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" {
-		}
+					comm = strings.TrimSpace(*pr.CommunityID)
-		_, err = tx.Exec(ctx, `
+				}
-			INSERT INTO revision_materialized_prefix (revision_id, prefix, community_id, source)
+				src := pr.Source
-			VALUES ($1::uuid, $2, $3::uuid, $4)`,
+				if strings.TrimSpace(src) == "" {
-			strings.TrimSpace(revisionID), strings.TrimSpace(pr.Prefix), comm, src)
+					src = "render"
+				}
+				return []any{strings.TrimSpace(revisionID), strings.TrimSpace(pr.Prefix), comm, src}, nil
+			}))
 		if err != nil {
 			return err
 		}

internal/store/backend.go

@@ -79,6 +79,18 @@ type Backend interface {
 
 	ListGlobalSettings(tenantID string) (map[string]any, error)
 	PatchGlobalSettings(tenantID string, patch map[string]any) error
+
+	// Module prefix snapshots cache last successful collect per module (pipeline ingest/render).
+	GetModulePrefixSnapshot(tenantID, moduleID string) (*ModulePrefixSnapshot, bool, error)
+	SetModulePrefixSnapshot(tenantID, moduleID, inputHash string, prefixes []PrefixRow) error
+	DeleteModulePrefixSnapshot(tenantID, moduleID string) error
+}
+
+// ModulePrefixSnapshot is the cached materialization for one module between refreshes.
+type ModulePrefixSnapshot struct {
+	InputHash   string
+	CollectedAt time.Time
+	Prefixes    []PrefixRow
 }
 
 // ModulePatch is a partial update for module.

internal/store/memory.go

@@ -39,7 +39,8 @@ type Memory struct {
 	domainEnt   map[string]*DomainEntry
 	ipRanges    map[string]*IPRangeEntry
 	settings    map[string]map[string]any // tenantID -> key -> JSON-compatible value
-	revPrefixes map[string][]PrefixRow
+	revPrefixes     map[string][]PrefixRow
+	moduleSnapshots map[string]*moduleSnapshotRec
 
 	// DemoIDs valid after SeedDemo()
 	demoTenantID   string
@@ -126,6 +127,7 @@ func NewMemory() *Memory {
 		ipRanges:          make(map[string]*IPRangeEntry),
 		settings:          make(map[string]map[string]any),
 		revPrefixes:       make(map[string][]PrefixRow),
+		moduleSnapshots:   make(map[string]*moduleSnapshotRec),
 	}
 }
 

internal/store/memory_crud.go

@@ -89,6 +89,9 @@ func (m *Memory) UpdateModule(tenantID, moduleID string, patch *ModulePatch) (*M
 func (m *Memory) SoftDeleteModule(tenantID, moduleID string) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
+	if m.moduleSnapshots != nil {
+		delete(m.moduleSnapshots, moduleSnapshotKey(tenantID, moduleID))
+	}
 	mod, ok := m.modules[moduleID]
 	if !ok || mod.TenantID != tenantID {
 		return ErrNotFound

internal/store/memory_snapshot.go

@@ -0,0 +1,62 @@
+package store
+
+import (
+	"strings"
+	"time"
+)
+
+func (m *Memory) GetModulePrefixSnapshot(tenantID, moduleID string) (*ModulePrefixSnapshot, bool, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	key := moduleSnapshotKey(tenantID, moduleID)
+	snap, ok := m.moduleSnapshots[key]
+	if !ok || snap == nil {
+		return nil, false, nil
+	}
+	cp := make([]PrefixRow, len(snap.Prefixes))
+	copy(cp, snap.Prefixes)
+	return &ModulePrefixSnapshot{
+		InputHash:   snap.InputHash,
+		CollectedAt: snap.CollectedAt,
+		Prefixes:    cp,
+	}, true, nil
+}
+
+func (m *Memory) SetModulePrefixSnapshot(tenantID, moduleID, inputHash string, prefixes []PrefixRow) error {
+	if strings.TrimSpace(tenantID) == "" || strings.TrimSpace(moduleID) == "" || strings.TrimSpace(inputHash) == "" {
+		return ErrInvalidInput
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.moduleSnapshots == nil {
+		m.moduleSnapshots = make(map[string]*moduleSnapshotRec)
+	}
+	cp := make([]PrefixRow, len(prefixes))
+	copy(cp, prefixes)
+	key := moduleSnapshotKey(tenantID, moduleID)
+	m.moduleSnapshots[key] = &moduleSnapshotRec{
+		InputHash:   inputHash,
+		CollectedAt: time.Now().UTC(),
+		Prefixes:    cp,
+	}
+	return nil
+}
+
+func (m *Memory) DeleteModulePrefixSnapshot(tenantID, moduleID string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.moduleSnapshots != nil {
+		delete(m.moduleSnapshots, moduleSnapshotKey(tenantID, moduleID))
+	}
+	return nil
+}
+
+type moduleSnapshotRec struct {
+	InputHash   string
+	CollectedAt time.Time
+	Prefixes    []PrefixRow
+}
+
+func moduleSnapshotKey(tenantID, moduleID string) string {
+	return strings.TrimSpace(tenantID) + "\x00" + strings.TrimSpace(moduleID)
+}

migrations/postgres/000009_module_prefix_snapshot.down.sql

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

migrations/postgres/000009_module_prefix_snapshot.up.sql

@@ -0,0 +1,11 @@
+-- Per-module materialized prefix cache to avoid re-fetching external sources on every tenant render.
+CREATE TABLE module_prefix_snapshot (
+    tenant_id      UUID NOT NULL REFERENCES tenant (id) ON DELETE CASCADE,
+    module_id      UUID NOT NULL REFERENCES module (id) ON DELETE CASCADE,
+    input_hash     TEXT NOT NULL,
+    collected_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+    prefixes_json  JSONB NOT NULL DEFAULT '[]',
+    PRIMARY KEY (tenant_id, module_id)
+);
+
+CREATE INDEX idx_module_prefix_snapshot_collected ON module_prefix_snapshot (collected_at);

migrations/sqlite/000009_module_prefix_snapshot.down.sql

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

migrations/sqlite/000009_module_prefix_snapshot.up.sql

@@ -0,0 +1,10 @@
+CREATE TABLE module_prefix_snapshot (
+    tenant_id      TEXT NOT NULL REFERENCES tenant (id) ON DELETE CASCADE,
+    module_id      TEXT NOT NULL REFERENCES module (id) ON DELETE CASCADE,
+    input_hash     TEXT NOT NULL,
+    collected_at   TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    prefixes_json  TEXT NOT NULL DEFAULT '[]',
+    PRIMARY KEY (tenant_id, module_id)
+);
+
+CREATE INDEX idx_module_prefix_snapshot_collected ON module_prefix_snapshot (collected_at);