perf: optimize data path, indexes and frontend virtualization

- Убран latestCDNRowsBySource; expanded BIRD preview генерируется on-demand
- Batch AS meta updates; миграция perf-индексов 000012
- VirtualPrefixList для preview префиксов; метрики pipeline refresh
- PolitePause для RIPEstat после cache miss

Co-authored-by: Cursor <[email protected]>

internal/pipeline/asn_cache.go

@@ -44,6 +44,7 @@ func resolveASNForEntry(ctx context.Context, st store.Backend, hc *http.Client,
 	if err != nil {
 		return nil, "", err
 	}
+	asnresolve.PolitePause()
 	holder, _ := asnresolve.ASHolderName(ctx, hc, asn)
 	if st != nil {
 		strs := make([]string, len(pfxs))

internal/pipeline/cdn_snapshot.go

@@ -25,9 +25,6 @@ func cachedCDNPrefixRows(st store.Backend, tenantID, moduleID string, priorSnaps
 				return cached
 			}
 		}
-		if cached := latestCDNRowsBySource(st, tenantID)[sourceKey]; len(cached) > 0 {
-			return cached
-		}
 	}
 	return nil
 }

internal/pipeline/collect_parallel.go

@@ -97,15 +97,18 @@ func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client,
 
 	seenPfx := make(map[string]struct{})
 	var out []store.PrefixRow
+	var metaUpdates []store.ASEntryResolveMetaUpdate
 	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)
-			}
+			metaUpdates = append(metaUpdates, store.ASEntryResolveMetaUpdate{
+				EntryID:     r.metaID,
+				ASNName:     r.holder,
+				PrefixCount: r.count,
+			})
 		}
 		for _, row := range r.rows {
 			k := row.Prefix
@@ -116,6 +119,11 @@ func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client,
 			out = append(out, row)
 		}
 	}
+	if len(metaUpdates) > 0 {
+		if err := st.UpdateASEntryResolveMetaBatch(tenantID, moduleID, metaUpdates, now); err != nil {
+			return nil, fmt.Errorf("as entry meta batch: %w", err)
+		}
+	}
 	return out, nil
 }
 
@@ -150,9 +158,11 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client
 					results[idx] = srcResult{rows: cached}
 					return
 				}
-				if cached := latestCDNRowsBySource(st, tenantID)[sourceKey]; len(cached) > 0 {
-					results[idx] = srcResult{rows: cached}
-					return
+				if snap, ok, _ := st.GetModulePrefixSnapshot(tenantID, moduleID); ok && snap != nil {
+					if cached := prefixRowsForSource(snap.Prefixes, sourceKey); len(cached) > 0 {
+						results[idx] = srcResult{rows: cached}
+						return
+					}
 				}
 			}
 			rows, err := fetchCDNSourceRows(ctx, st, hc, tenantID, moduleID, mod, src, priorSnapshot, now)

internal/pipeline/refresh.go

@@ -36,6 +36,11 @@ const (
 	revisionDefaultTTL  = 30 * 24 * time.Hour
 )
 
+// AuxBirdFullExpandedKey returns the preview map key for the expanded BIRD config (generated on demand).
+func AuxBirdFullExpandedKey() string {
+	return auxBirdFullExpanded
+}
+
 // MaterializedASPrefixKey returns the revision snapshot key for an AS-only entry (not a CIDR).
 func MaterializedASPrefixKey(asn int64) string {
 	return fmt.Sprintf("as:%d", asn)
@@ -47,10 +52,14 @@ func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client,
 	if hc == nil {
 		hc = http.DefaultClient
 	}
+	start := time.Now()
 	mod, err := st.GetModule(tenantID, moduleID)
 	if err != nil {
 		return err
 	}
+	defer func() {
+		observability.RecordPipelineRefresh(mod.Type, time.Since(start))
+	}()
 	if !mod.Enabled {
 		return fmt.Errorf("pipeline: module disabled")
 	}
@@ -197,33 +206,6 @@ func shouldSkipCDNSourceFetch(src *store.CDNSource, now time.Time) bool {
 	return now.UTC().Before(nextRefreshAt)
 }
 
-func latestCDNRowsBySource(st store.Backend, tenantID string) map[string][]store.PrefixRow {
-	out := make(map[string][]store.PrefixRow)
-	if st == nil {
-		return out
-	}
-	revs, _, _ := st.ListRevisions(tenantID, "", "", 1)
-	if len(revs) == 0 || strings.TrimSpace(revs[0].ID) == "" {
-		return out
-	}
-	revID := strings.TrimSpace(revs[0].ID)
-	cursor := ""
-	for {
-		page, next, more := st.ListRevisionPrefixes(tenantID, revID, cursor, 2000)
-		for _, row := range page {
-			if !strings.HasPrefix(strings.TrimSpace(row.Source), "cdn:") {
-				continue
-			}
-			out[row.Source] = append(out[row.Source], row)
-		}
-		if !more || strings.TrimSpace(next) == "" {
-			break
-		}
-		cursor = next
-	}
-	return out
-}
-
 type dohJSONAnswer struct {
 	Type int    `json:"type"`
 	Data string `json:"data"`
@@ -836,10 +818,18 @@ func buildPreviewFragments(st store.Backend, tenantID, moduleID, revisionID stri
 		px6:         birdfmt.JoinFragments(birdfmt.ManagedBanner(revisionID), staticV6),
 		pPeers:      peersBody,
 	}
-	out[auxBirdFullExpanded] = buildExpandedBirdText(main, out)
 	return out, nil
 }
 
+// BuildExpandedBirdPreview concatenates bird.conf and deployable includes for UI preview (not persisted in revision).
+func BuildExpandedBirdPreview(frags map[string]string) string {
+	if frags == nil {
+		return ""
+	}
+	main := frags["bird.conf"]
+	return buildExpandedBirdText(main, frags)
+}
+
 func renderStaticProtocolsByCommunity(groups []staticCommunityRoutes) (string, string) {
 	var b4 strings.Builder
 	var b6 strings.Builder

internal/repository/postgres_entities.go

@@ -325,6 +325,38 @@ func (p *Postgres) UpdateASEntryResolveMeta(tenantID, moduleID, entryID string,
 	return nil
 }
 
+func (p *Postgres) UpdateASEntryResolveMetaBatch(tenantID, moduleID string, updates []store.ASEntryResolveMetaUpdate, resolvedAt time.Time) error {
+	if len(updates) == 0 {
+		return nil
+	}
+	if _, err := p.GetModule(tenantID, moduleID); err != nil {
+		return err
+	}
+	ctx := context.Background()
+	batch := &pgx.Batch{}
+	for _, u := range updates {
+		var nameArg any
+		sn := strings.TrimSpace(u.ASNName)
+		if sn == "" {
+			nameArg = nil
+		} else {
+			nameArg = sn
+		}
+		batch.Queue(`
+			UPDATE module_as_entry SET asn_name=$3, prefix_count=$4, asn_resolved_at=$5, updated_at=now()
+			WHERE id=$1 AND module_id=$2`,
+			u.EntryID, moduleID, nameArg, u.PrefixCount, resolvedAt.UTC())
+	}
+	br := p.pool.SendBatch(ctx, batch)
+	defer func() { _ = br.Close() }()
+	for range updates {
+		if _, err := br.Exec(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (p *Postgres) DeleteASEntry(tenantID, moduleID, entryID string) error {
 	if _, err := p.GetModule(tenantID, moduleID); err != nil {
 		return err

internal/store/as_meta_batch.go

@@ -0,0 +1,8 @@
+package store
+
+// ASEntryResolveMetaUpdate is one row for batch AS resolve metadata writes.
+type ASEntryResolveMetaUpdate struct {
+	EntryID     string
+	ASNName     string
+	PrefixCount int64
+}

internal/store/backend.go

@@ -18,6 +18,8 @@ type Backend interface {
 
 	// ListModules returns all modules for a tenant (control plane may paginate in httpapi).
 	ListModules(tenantID string) []*Module
+	// ListModulesPage returns one page of modules (limit capped by caller).
+	ListModulesPage(tenantID, cursor string, limit int) ([]*Module, string, bool)
 	GetModule(tenantID, moduleID string) (*Module, error)
 	CreateModule(tenantID string, in *Module) (*Module, error)
 	UpdateModule(tenantID, moduleID string, patch *ModulePatch) (*Module, error)
@@ -34,6 +36,7 @@ type Backend interface {
 	DeleteASEntry(tenantID, moduleID, entryID string) error
 	// UpdateASEntryResolveMeta записывает имя AS, число объявленных префиксов и время успешного резолва (pipeline).
 	UpdateASEntryResolveMeta(tenantID, moduleID, entryID string, asnName string, prefixCount int64, resolvedAt time.Time) error
+	UpdateASEntryResolveMetaBatch(tenantID, moduleID string, updates []ASEntryResolveMetaUpdate, resolvedAt time.Time) error
 
 	ListDomainEntries(tenantID, moduleID string) ([]*DomainEntry, error)
 	CreateDomainEntry(tenantID, moduleID string, in *DomainEntry) (*DomainEntry, error)

internal/store/memory_crud.go

@@ -301,6 +301,15 @@ func (m *Memory) UpdateASEntryResolveMeta(tenantID, moduleID, entryID string, as
 	return nil
 }
 
+func (m *Memory) UpdateASEntryResolveMetaBatch(tenantID, moduleID string, updates []ASEntryResolveMetaUpdate, resolvedAt time.Time) error {
+	for _, u := range updates {
+		if err := m.UpdateASEntryResolveMeta(tenantID, moduleID, u.EntryID, u.ASNName, u.PrefixCount, resolvedAt); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (m *Memory) DeleteASEntry(tenantID, moduleID, entryID string) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()