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);