From ee364c8b6dcff5f565a1d025230baea9a9cee55b Mon Sep 17 00:00:00 2001 From: Denozordec Date: Tue, 19 May 2026 10:22:33 +0700 Subject: [PATCH] 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 | 106 ++++++ internal/pipeline/collect_parallel.go | 302 ++++++++++++++++++ .../pipeline/materialize_regression_test.go | 2 +- internal/pipeline/module_hash.go | 123 +++++++ internal/pipeline/refresh.go | 168 +--------- internal/repository/module_snapshot.go | 70 ++++ internal/repository/postgres.go | 29 +- internal/store/backend.go | 12 + internal/store/memory.go | 4 +- internal/store/memory_crud.go | 3 + internal/store/memory_snapshot.go | 62 ++++ .../000009_module_prefix_snapshot.down.sql | 1 + .../000009_module_prefix_snapshot.up.sql | 11 + .../000009_module_prefix_snapshot.down.sql | 1 + .../000009_module_prefix_snapshot.up.sql | 10 + 15 files changed, 730 insertions(+), 174 deletions(-) create mode 100644 internal/pipeline/aggregate.go create mode 100644 internal/pipeline/collect_parallel.go create mode 100644 internal/pipeline/module_hash.go create mode 100644 internal/repository/module_snapshot.go create mode 100644 internal/store/memory_snapshot.go create mode 100644 migrations/postgres/000009_module_prefix_snapshot.down.sql create mode 100644 migrations/postgres/000009_module_prefix_snapshot.up.sql create mode 100644 migrations/sqlite/000009_module_prefix_snapshot.down.sql create mode 100644 migrations/sqlite/000009_module_prefix_snapshot.up.sql diff --git a/internal/pipeline/aggregate.go b/internal/pipeline/aggregate.go new file mode 100644 index 0000000..60d85fd --- /dev/null +++ b/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 +} diff --git a/internal/pipeline/collect_parallel.go b/internal/pipeline/collect_parallel.go new file mode 100644 index 0000000..e74d69c --- /dev/null +++ b/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 +} diff --git a/internal/pipeline/materialize_regression_test.go b/internal/pipeline/materialize_regression_test.go index cfffc68..a59eea8 100644 --- a/internal/pipeline/materialize_regression_test.go +++ b/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) } diff --git a/internal/pipeline/module_hash.go b/internal/pipeline/module_hash.go new file mode 100644 index 0000000..2e3e633 --- /dev/null +++ b/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 +} diff --git a/internal/pipeline/refresh.go b/internal/pipeline/refresh.go index bc4b14a..645d7af 100644 --- a/internal/pipeline/refresh.go +++ b/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" - 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 + return collectASPrefixRows(ctx, st, hc, tenantID, mod, list) case "CDN_CIDRS": sources, err := st.ListCDNSources(tenantID, moduleID) if err != nil { return nil, err } - var rows []store.PrefixRow - 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 + return collectCDNPrefixRows(ctx, st, hc, tenantID, mod, sources, priorSnapshot) 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 - 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 + return collectDomainPrefixRows(ctx, hc, mod, profile, entries) 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 diff --git a/internal/repository/module_snapshot.go b/internal/repository/module_snapshot.go new file mode 100644 index 0000000..ffe8aa1 --- /dev/null +++ b/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 +} diff --git a/internal/repository/postgres.go b/internal/repository/postgres.go index 0a13594..919581f 100644 --- a/internal/repository/postgres.go +++ b/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 { - var comm any - if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" { - comm = strings.TrimSpace(*pr.CommunityID) - } - src := pr.Source - if strings.TrimSpace(src) == "" { - src = "render" - } - _, err = tx.Exec(ctx, ` - INSERT INTO revision_materialized_prefix (revision_id, prefix, community_id, source) - VALUES ($1::uuid, $2, $3::uuid, $4)`, - strings.TrimSpace(revisionID), strings.TrimSpace(pr.Prefix), comm, src) + if len(prefixes) > 0 { + _, err = tx.CopyFrom(ctx, + pgx.Identifier{"revision_materialized_prefix"}, + []string{"revision_id", "prefix", "community_id", "source"}, + pgx.CopyFromSlice(len(prefixes), func(i int) ([]any, error) { + pr := prefixes[i] + var comm any + if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" { + comm = strings.TrimSpace(*pr.CommunityID) + } + src := pr.Source + if strings.TrimSpace(src) == "" { + src = "render" + } + return []any{strings.TrimSpace(revisionID), strings.TrimSpace(pr.Prefix), comm, src}, nil + })) if err != nil { return err } diff --git a/internal/store/backend.go b/internal/store/backend.go index 6a56b7e..b9342d2 100644 --- a/internal/store/backend.go +++ b/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. diff --git a/internal/store/memory.go b/internal/store/memory.go index c30b162..447dcbc 100644 --- a/internal/store/memory.go +++ b/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), } } diff --git a/internal/store/memory_crud.go b/internal/store/memory_crud.go index f12f6c1..bc8ceee 100644 --- a/internal/store/memory_crud.go +++ b/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 diff --git a/internal/store/memory_snapshot.go b/internal/store/memory_snapshot.go new file mode 100644 index 0000000..27c44d5 --- /dev/null +++ b/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) +} diff --git a/migrations/postgres/000009_module_prefix_snapshot.down.sql b/migrations/postgres/000009_module_prefix_snapshot.down.sql new file mode 100644 index 0000000..e3f66bf --- /dev/null +++ b/migrations/postgres/000009_module_prefix_snapshot.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS module_prefix_snapshot; diff --git a/migrations/postgres/000009_module_prefix_snapshot.up.sql b/migrations/postgres/000009_module_prefix_snapshot.up.sql new file mode 100644 index 0000000..d25f400 --- /dev/null +++ b/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); diff --git a/migrations/sqlite/000009_module_prefix_snapshot.down.sql b/migrations/sqlite/000009_module_prefix_snapshot.down.sql new file mode 100644 index 0000000..e3f66bf --- /dev/null +++ b/migrations/sqlite/000009_module_prefix_snapshot.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS module_prefix_snapshot; diff --git a/migrations/sqlite/000009_module_prefix_snapshot.up.sql b/migrations/sqlite/000009_module_prefix_snapshot.up.sql new file mode 100644 index 0000000..407df1a --- /dev/null +++ b/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);