diff --git a/internal/httpapi/routes.go b/internal/httpapi/routes.go index ecdb8a2..6e94479 100644 --- a/internal/httpapi/routes.go +++ b/internal/httpapi/routes.go @@ -579,7 +579,7 @@ func (s *Server) handleGetRevision(w http.ResponseWriter, r *http.Request) { if !s.requireAtLeast(w, a, "viewer") { return } - rev, err := s.store.GetRevision(a.TenantID, r.PathValue("revision_id")) + rev, err := s.store.GetRevisionSummary(a.TenantID, r.PathValue("revision_id")) if err != nil { writeProblem(w, http.StatusNotFound, "Not Found", "revision not found") return diff --git a/internal/ingest/run.go b/internal/ingest/run.go index 7c918fa..58a1579 100644 --- a/internal/ingest/run.go +++ b/internal/ingest/run.go @@ -3,11 +3,11 @@ package ingest import ( "context" "log" - "net/http" "time" "evobgp/internal/broker" "evobgp/internal/config" + "evobgp/internal/httpclient" "evobgp/internal/pipeline" "evobgp/internal/store" ) @@ -24,7 +24,7 @@ func Run(ctx context.Context, deps *Deps) { if deps == nil || deps.Store == nil { log.Fatalf("evobgp-ingest: missing store (pass ingest.Deps from BootstrapWorkers or evobgp-all)") } - hc := &http.Client{Timeout: 45 * time.Second} + hc := httpclient.New(httpclient.DefaultTimeout) t := time.NewTicker(60 * time.Second) defer t.Stop() log.Printf("evobgp-ingest: active (CDN conditional GET / ETag prefetch)") @@ -34,7 +34,10 @@ func Run(ctx context.Context, deps *Deps) { log.Printf("evobgp-ingest: stopped") return case <-t.C: - if err := pipeline.PrefetchCDNSourceETags(context.Background(), deps.Store, hc); err != nil { + prefetchCtx, cancel := context.WithTimeout(ctx, 50*time.Second) + err := pipeline.PrefetchCDNSourceETags(prefetchCtx, deps.Store, hc) + cancel() + if err != nil { log.Printf("evobgp-ingest: prefetch: %v", err) } } diff --git a/internal/pipeline/prefetch.go b/internal/pipeline/prefetch.go index 025069d..8563713 100644 --- a/internal/pipeline/prefetch.go +++ b/internal/pipeline/prefetch.go @@ -5,87 +5,116 @@ import ( "io" "net/http" "strings" + "sync" "time" "evobgp/internal/httpclient" "evobgp/internal/store" ) +type prefetchTask struct { + tenantID string + mod *store.Module + src *store.CDNSource +} + // PrefetchCDNSourceETags performs conditional GETs for CDN sources; on 200 parses CIDRs into module_prefix_snapshot. func PrefetchCDNSourceETags(ctx context.Context, st store.Backend, hc *http.Client) error { if hc == nil { hc = httpclient.New(httpclient.DefaultTimeout) } + if ctx == nil { + ctx = context.Background() + } tenants, err := st.ListTenantIDs() if err != nil { return err } - now := time.Now().UTC() + var tasks []prefetchTask for _, tid := range tenants { for _, mod := range st.ListModules(tid) { if mod == nil || !mod.Enabled || mod.Type != "CDN_CIDRS" { continue } - omod, err := st.GetModule(tid, mod.ID) - if err != nil { - continue - } sources, err := st.ListCDNSources(tid, mod.ID) if err != nil { continue } - var prior []store.PrefixRow - if snap, ok, _ := st.GetModulePrefixSnapshot(tid, mod.ID); ok && snap != nil { - prior = snap.Prefixes - } for _, src := range sources { - if src == nil || strings.TrimSpace(src.URL) == "" { - continue - } - req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimSpace(src.URL), nil) - if err != nil { - continue - } - if etag := strings.TrimSpace(src.Etag); etag != "" { - req.Header.Set("If-None-Match", etag) - } - resp, err := hc.Do(req) - if err != nil { - continue - } - if resp.StatusCode == http.StatusNotModified { - _ = resp.Body.Close() - continue - } - if resp.StatusCode != http.StatusOK { - _, _ = io.Copy(io.Discard, resp.Body) - _ = resp.Body.Close() - continue - } - body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) - _ = resp.Body.Close() - if err != nil { - continue - } - prefixStrs, err := parseCDNBody(string(body), src) - if err != nil { - continue - } - newEtag := strings.TrimSpace(resp.Header.Get("ETag")) - patch := &store.CDNSourcePatch{LastRefreshedAt: &now} - if newEtag != "" && newEtag != strings.TrimSpace(src.Etag) { - e := newEtag - patch.Etag = &e - } - _, _ = st.UpdateCDNSource(tid, mod.ID, src.ID, patch) - rows := cdnRowsFromParsed(omod, src, prefixStrs) - _ = mergeCDNSourceIntoModuleSnapshot(st, tid, omod, src.ID, rows) - _ = prior // prior may be stale after merge; refresh for next source in loop - if snap, ok, _ := st.GetModulePrefixSnapshot(tid, mod.ID); ok && snap != nil { - prior = snap.Prefixes + if src != nil && strings.TrimSpace(src.URL) != "" { + tasks = append(tasks, prefetchTask{tenantID: tid, mod: mod, src: src}) } } } } + if len(tasks) == 0 { + return nil + } + sem := make(chan struct{}, collectConcurrency()) + var wg sync.WaitGroup + for _, task := range tasks { + wg.Add(1) + go func(t prefetchTask) { + defer wg.Done() + sem <- struct{}{} + defer func() { <-sem }() + prefetchOneCDNSource(ctx, st, hc, t) + }(task) + } + wg.Wait() return nil } + +func prefetchOneCDNSource(ctx context.Context, st store.Backend, hc *http.Client, t prefetchTask) { + now := time.Now().UTC() + tid, mod, src := t.tenantID, t.mod, t.src + u := strings.TrimSpace(src.URL) + if _, err := ValidateCDNURL(u); err != nil { + return + } + if err := ResolveCDNURLHost(ctx, u); err != nil { + return + } + omod, err := st.GetModule(tid, mod.ID) + if err != nil { + return + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return + } + if etag := strings.TrimSpace(src.Etag); etag != "" { + req.Header.Set("If-None-Match", etag) + } + resp, err := upstreamHTTPDo(ctx, hc, req) + if err != nil { + return + } + if resp.StatusCode == http.StatusNotModified { + _ = resp.Body.Close() + return + } + if resp.StatusCode != http.StatusOK { + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + return + } + body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) + _ = resp.Body.Close() + if err != nil { + return + } + prefixStrs, err := parseCDNBody(string(body), src) + if err != nil { + return + } + newEtag := strings.TrimSpace(resp.Header.Get("ETag")) + patch := &store.CDNSourcePatch{LastRefreshedAt: &now} + if newEtag != "" && newEtag != strings.TrimSpace(src.Etag) { + e := newEtag + patch.Etag = &e + } + _, _ = st.UpdateCDNSource(tid, mod.ID, src.ID, patch) + rows := cdnRowsFromParsed(omod, src, prefixStrs) + _ = mergeCDNSourceIntoModuleSnapshot(st, tid, omod, src.ID, rows) +} diff --git a/internal/repository/ctx.go b/internal/repository/ctx.go new file mode 100644 index 0000000..4797dcb --- /dev/null +++ b/internal/repository/ctx.go @@ -0,0 +1,19 @@ +package repository + +import ( + "context" + "time" +) + +const defaultRepoTimeout = 60 * time.Second + +// boundedRepoCtx returns a context with default repository I/O timeout. +func boundedRepoCtx(parent context.Context) (context.Context, context.CancelFunc) { + if parent == nil { + parent = context.Background() + } + if _, ok := parent.Deadline(); ok { + return parent, func() {} + } + return context.WithTimeout(parent, defaultRepoTimeout) +} diff --git a/internal/repository/postgres.go b/internal/repository/postgres.go index 965c435..e0fcf46 100644 --- a/internal/repository/postgres.go +++ b/internal/repository/postgres.go @@ -659,7 +659,8 @@ func (p *Postgres) DeleteSpeaker(tenantID, id string) error { } func (p *Postgres) GetRevision(tenantID, revisionID string) (*store.Revision, error) { - ctx := context.Background() + ctx, cancel := boundedRepoCtx(context.Background()) + defer cancel() var r store.Revision var mod *string var parent *string @@ -691,6 +692,33 @@ func (p *Postgres) GetRevision(tenantID, revisionID string) (*store.Revision, er return &r, nil } +func (p *Postgres) GetRevisionSummary(tenantID, revisionID string) (*store.Revision, error) { + ctx, cancel := boundedRepoCtx(context.Background()) + defer cancel() + var r store.Revision + var mod *string + var parent *string + var prefixCount int + err := p.pool.QueryRow(ctx, ` + SELECT id::text, tenant_id::text, module_id::text, content_hash, parent_revision_id::text, + COALESCE((meta_json->>'materialized_prefix_count')::int, 0), created_at + FROM config_revision WHERE id=$1 AND tenant_id=$2`, revisionID, tenantID).Scan( + &r.ID, &r.TenantID, &mod, &r.ContentHash, &parent, &prefixCount, &r.CreatedAt) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, store.ErrNotFound + } + return nil, err + } + if mod != nil { + r.ModuleID = *mod + } + r.ParentRevisionID = strOrNil(parent) + r.MaterializedPrefixCount = prefixCount + r.PreviewFragments = map[string]string{} + return &r, nil +} + func (p *Postgres) ListRevisions(tenantID, moduleID string, cursor string, limit int) ([]*store.Revision, string, bool) { if limit <= 0 { limit = 50 diff --git a/internal/store/backend.go b/internal/store/backend.go index d157e5c..35b9a01 100644 --- a/internal/store/backend.go +++ b/internal/store/backend.go @@ -76,6 +76,8 @@ type Backend interface { DeleteSpeaker(tenantID, id string) error GetRevision(tenantID, revisionID string) (*Revision, error) + // GetRevisionSummary returns revision metadata without preview_fragments payloads. + GetRevisionSummary(tenantID, revisionID string) (*Revision, error) ListRevisions(tenantID, moduleID string, cursor string, limit int) (items []*Revision, nextCursor string, hasMore bool) ListRevisionPrefixes(tenantID, revisionID string, cursor string, limit int) (prefixes []PrefixRow, next string, more bool) CreateRollbackRevision(tenantID, sourceRevisionID string) (newID string, err error) diff --git a/internal/store/memory.go b/internal/store/memory.go index 7c16d53..8b31954 100644 --- a/internal/store/memory.go +++ b/internal/store/memory.go @@ -424,14 +424,19 @@ func (m *Memory) GetModule(tenantID, moduleID string) (*Module, error) { func (m *Memory) GetRevision(tenantID, revisionID string) (*Revision, error) { m.mu.RLock() defer m.mu.RUnlock() - rev, ok := m.revisions[revisionID] - if !ok { - return nil, ErrNotFound + return m.getRevisionLocked(tenantID, revisionID) +} + +func (m *Memory) GetRevisionSummary(tenantID, revisionID string) (*Revision, error) { + m.mu.RLock() + defer m.mu.RUnlock() + rev, err := m.getRevisionLocked(tenantID, revisionID) + if err != nil { + return nil, err } - if rev.TenantID != tenantID { - return nil, ErrTenantScope - } - return rev, nil + cp := *rev + cp.PreviewFragments = nil + return &cp, nil } func (m *Memory) GetSpeaker(tenantID, speakerID string) (*Speaker, error) {