refactor: phase 2 structural alignment — reports, importer, nodecli, pagination

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

internal/httpapi/list_params.go

@@ -0,0 +1,24 @@
+package httpapi
+
+import (
+	"net/http"
+	"strconv"
+)
+
+func parseListLimit(r *http.Request) int {
+	limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
+	if limit <= 0 {
+		return 50
+	}
+	if limit > 500 {
+		return 500
+	}
+	return limit
+}
+
+func strPtrOrNull(s string) any {
+	if s == "" {
+		return nil
+	}
+	return s
+}

internal/httpapi/routes.go

@@ -19,6 +19,7 @@ import (
 	"evobgp/internal/jobs"
 	"evobgp/internal/observability"
 	"evobgp/internal/pipeline"
+	"evobgp/internal/reports"
 	"evobgp/internal/store"
 )
 
@@ -189,19 +190,23 @@ func (s *Server) handleListModules(w http.ResponseWriter, r *http.Request) {
 		enabledFilter = &v
 	}
 
-	mods := s.store.ListModules(a.TenantID)
-	items := make([]map[string]any, 0, len(mods))
-	for _, mod := range mods {
+	filtered := make([]*store.Module, 0)
+	for _, mod := range s.store.ListModules(a.TenantID) {
 		if typeFilter != "" && mod.Type != typeFilter {
 			continue
 		}
 		if enabledFilter != nil && mod.Enabled != *enabledFilter {
 			continue
 		}
+		filtered = append(filtered, mod)
+	}
+	page, next, more := store.PaginateOffset(filtered, r.URL.Query().Get("cursor"), parseListLimit(r))
+	items := make([]map[string]any, 0, len(page))
+	for _, mod := range page {
 		items = append(items, moduleJSON(mod))
 	}
 	writeJSON(w, http.StatusOK, map[string]any{
-		"items": items, "next_cursor": nil, "has_more": false,
+		"items": items, "next_cursor": strPtrOrNull(next), "has_more": more,
 	})
 }
 
@@ -214,91 +219,17 @@ func (s *Server) handleRouterListsCatalog(w http.ResponseWriter, r *http.Request
 	if !s.requireAtLeast(w, a, "viewer") {
 		return
 	}
-
-	mods := s.store.ListModules(a.TenantID)
-	moduleItems := make([]map[string]any, 0, len(mods))
-	domains := make([]map[string]any, 0)
-	asns := make([]map[string]any, 0)
-	ipRanges := make([]map[string]any, 0)
-
-	for _, mod := range mods {
-		switch mod.Type {
-		case "DOMAINS", "AS_PREFIXES", "IP_RANGES":
-			moduleItems = append(moduleItems, moduleJSON(mod))
-		default:
-			continue
-		}
-
-		switch mod.Type {
-		case "DOMAINS":
-			list, err := s.store.ListDomainEntries(a.TenantID, mod.ID)
-			if err != nil {
-				writeStoreErr(w, err)
-				return
-			}
-			for _, x := range list {
-				domains = append(domains, map[string]any{
-					"module_id": mod.ID,
-					"entry":     domainEntryJSON(x),
-				})
-			}
-		case "AS_PREFIXES":
-			list, err := s.store.ListASEntries(a.TenantID, mod.ID)
-			if err != nil {
-				writeStoreErr(w, err)
-				return
-			}
-			for _, x := range list {
-				asns = append(asns, map[string]any{
-					"module_id": mod.ID,
-					"entry":     asEntryJSON(x),
-				})
-			}
-		case "IP_RANGES":
-			list, err := s.store.ListIPRangeEntries(a.TenantID, mod.ID)
-			if err != nil {
-				writeStoreErr(w, err)
-				return
-			}
-			for _, x := range list {
-				ipRanges = append(ipRanges, map[string]any{
-					"module_id": mod.ID,
-					"entry":     ipRangeJSON(x),
-				})
-			}
-		}
-	}
-
-	comms, err := s.store.ListCommunities(a.TenantID)
+	cat, err := reports.BuildRouterListsCatalog(s.store, a.TenantID)
 	if err != nil {
 		writeStoreErr(w, err)
 		return
 	}
-	communityItems := make([]map[string]any, 0, len(comms))
-	for _, c := range comms {
-		communityItems = append(communityItems, map[string]any{
-			"id":        c.ID,
-			"community": c.Community,
-			"title":     c.Title,
-		})
-	}
-
 	writeJSON(w, http.StatusOK, map[string]any{
-		"modules": map[string]any{
-			"items": moduleItems,
-		},
-		"domains": map[string]any{
-			"items": domains,
-		},
-		"asns": map[string]any{
-			"items": asns,
-		},
-		"ip_ranges": map[string]any{
-			"items": ipRanges,
-		},
-		"communities": map[string]any{
-			"items": communityItems,
-		},
+		"modules":     map[string]any{"items": cat.Modules},
+		"domains":     map[string]any{"items": cat.Domains},
+		"asns":        map[string]any{"items": cat.ASNs},
+		"ip_ranges":   map[string]any{"items": cat.IPRanges},
+		"communities": map[string]any{"items": cat.Communities},
 	})
 }
 
@@ -332,10 +263,11 @@ func (s *Server) handleListPeers(w http.ResponseWriter, r *http.Request) {
 	if !s.requireAtLeast(w, a, "viewer") {
 		return
 	}
-	peers := s.store.ListPeers(a.TenantID)
+	allPeers := s.store.ListPeers(a.TenantID)
+	page, next, more := store.PaginateOffset(allPeers, r.URL.Query().Get("cursor"), parseListLimit(r))
 	liveStates := s.liveBGPProtocolStates(r.Context())
-	items := make([]map[string]any, 0, len(peers))
-	for _, p := range peers {
+	items := make([]map[string]any, 0, len(page))
+	for _, p := range page {
 		row := peerJSON(p)
 		if st, ok := liveStates[peerProtocolNameForID(p.ID)]; ok && strings.TrimSpace(st) != "" {
 			row["session_state"] = strings.TrimSpace(st)
@@ -343,7 +275,7 @@ func (s *Server) handleListPeers(w http.ResponseWriter, r *http.Request) {
 		items = append(items, row)
 	}
 	writeJSON(w, http.StatusOK, map[string]any{
-		"items": items, "next_cursor": nil, "has_more": false,
+		"items": items, "next_cursor": strPtrOrNull(next), "has_more": more,
 	})
 }
 
@@ -570,13 +502,6 @@ func revisionJSON(rev *store.Revision) map[string]any {
 	return m
 }
 
-func strPtrOrNull(s string) any {
-	if s == "" {
-		return nil
-	}
-	return s
-}
-
 // enqueueModuleRefreshIfEnabled queues module_refresh when the module exists and is enabled (best-effort, no HTTP error).
 func (s *Server) enqueueModuleRefreshIfEnabled(tenantID, moduleID, trigger string) {
 	if s.jobs == nil {
@@ -672,225 +597,15 @@ func (s *Server) handleRevisionDiagnosticLog(w http.ResponseWriter, r *http.Requ
 		writeProblem(w, http.StatusNotFound, "Not Found", "revision not found")
 		return
 	}
-
-	moduleType := ""
-	moduleName := ""
-	if strings.TrimSpace(rev.ModuleID) != "" {
-		if mod, modErr := s.store.GetModule(a.TenantID, rev.ModuleID); modErr == nil && mod != nil {
-			moduleType = strings.TrimSpace(mod.Type)
-			moduleName = strings.TrimSpace(mod.Name)
-		}
+	logOut, err := reports.BuildRevisionDiagnosticLog(s.store, a.TenantID, rev)
+	if err != nil {
+		writeInternalError(w, "revision diagnostic log", err)
+		return
 	}
-
-	communityLabels := map[string]string{}
-	if communities, listErr := s.store.ListCommunities(a.TenantID); listErr == nil {
-		for _, c := range communities {
-			if c == nil {
-				continue
-			}
-			label := strings.TrimSpace(c.Title)
-			if label == "" {
-				label = strings.TrimSpace(c.Community)
-			}
-			if label == "" {
-				label = c.ID
-			}
-			communityLabels[c.ID] = label
-		}
-	}
-
-	type rawRow struct {
-		Prefix         string
-		Source         string
-		Kind           string
-		SourceName     string
-		SourceDetail   string
-		CommunityID    string
-		CommunityLabel string
-	}
-	type summary struct {
-		Kind           string
-		Source         string
-		SourceDetail   string
-		CommunityID    string
-		CommunityLabel string
-		Count          int
-		Sample         []string
-	}
-
-	cdnSourceURLByID := map[string]string{}
-	if moduleType == "CDN_CIDRS" && strings.TrimSpace(rev.ModuleID) != "" {
-		if sources, srcErr := s.store.ListCDNSources(a.TenantID, rev.ModuleID); srcErr == nil {
-			for _, src := range sources {
-				if src == nil {
-					continue
-				}
-				id := strings.TrimSpace(src.ID)
-				url := strings.TrimSpace(src.URL)
-				if id != "" && url != "" {
-					cdnSourceURLByID[id] = url
-				}
-			}
-		}
-	}
-
-	rows := make([]rawRow, 0, rev.MaterializedPrefixCount)
-	byGroup := map[string]*summary{}
-	cursor := ""
-	for {
-		page, next, more := s.store.ListRevisionPrefixes(a.TenantID, revID, cursor, 2000)
-		for _, p := range page {
-			kind, sourceName := classifyRevisionSource(p.Source)
-			sourceDetail := sourceName
-			if kind == "cdn" {
-				if url, ok := cdnSourceURLByID[sourceName]; ok && strings.TrimSpace(url) != "" {
-					sourceDetail = url
-				}
-			}
-			communityID := "none"
-			if p.CommunityID != nil && strings.TrimSpace(*p.CommunityID) != "" {
-				communityID = strings.TrimSpace(*p.CommunityID)
-			}
-			communityLabel := "без community"
-			if communityID != "none" {
-				if lbl, ok := communityLabels[communityID]; ok && strings.TrimSpace(lbl) != "" {
-					communityLabel = lbl
-				} else {
-					communityLabel = communityID
-				}
-			}
-			rows = append(rows, rawRow{
-				Prefix:         p.Prefix,
-				Source:         p.Source,
-				Kind:           kind,
-				SourceName:     sourceName,
-				SourceDetail:   sourceDetail,
-				CommunityID:    communityID,
-				CommunityLabel: communityLabel,
-			})
-			groupKey := kind + "|" + sourceName + "|" + communityID
-			g, ok := byGroup[groupKey]
-			if !ok {
-				g = &summary{
-					Kind:           kind,
-					Source:         sourceName,
-					SourceDetail:   sourceDetail,
-					CommunityID:    communityID,
-					CommunityLabel: communityLabel,
-					Sample:         make([]string, 0, 5),
-				}
-				byGroup[groupKey] = g
-			}
-			g.Count++
-			if len(g.Sample) < 5 {
-				g.Sample = append(g.Sample, p.Prefix)
-			}
-		}
-		if !more || strings.TrimSpace(next) == "" {
-			break
-		}
-		cursor = next
-	}
-
-	summaryKeys := make([]string, 0, len(byGroup))
-	for k := range byGroup {
-		summaryKeys = append(summaryKeys, k)
-	}
-	sort.Strings(summaryKeys)
-
-	sort.Slice(rows, func(i, j int) bool {
-		if rows[i].Kind != rows[j].Kind {
-			return rows[i].Kind < rows[j].Kind
-		}
-		if rows[i].SourceName != rows[j].SourceName {
-			return rows[i].SourceName < rows[j].SourceName
-		}
-		if rows[i].CommunityID != rows[j].CommunityID {
-			return rows[i].CommunityID < rows[j].CommunityID
-		}
-		return rows[i].Prefix < rows[j].Prefix
-	})
-
-	var b strings.Builder
-	b.WriteString("EvoBGP revision diagnostic log\n")
-	b.WriteString("generated_at=" + time.Now().UTC().Format(time.RFC3339Nano) + "\n")
-	b.WriteString("tenant_id=" + a.TenantID + "\n")
-	b.WriteString("revision_id=" + rev.ID + "\n")
-	b.WriteString("revision_created_at=" + rev.CreatedAt.UTC().Format(time.RFC3339Nano) + "\n")
-	b.WriteString("content_hash=" + rev.ContentHash + "\n")
-	b.WriteString("materialized_prefix_count=" + strconv.Itoa(rev.MaterializedPrefixCount) + "\n")
-	if rev.ModuleID != "" {
-		b.WriteString("module_id=" + rev.ModuleID + "\n")
-	} else {
-		b.WriteString("module_id=\n")
-	}
-	b.WriteString("module_type=" + moduleType + "\n")
-	b.WriteString("module_name=" + moduleName + "\n")
-	b.WriteString("fetched_rows=" + strconv.Itoa(len(rows)) + "\n\n")
-
-	b.WriteString("## Aggregation by source and community\n")
-	for _, k := range summaryKeys {
-		g := byGroup[k]
-		b.WriteString("- kind=" + g.Kind +
-			" source=" + g.Source +
-			" source_detail=" + g.SourceDetail +
-			" community_id=" + g.CommunityID +
-			" community_label=" + g.CommunityLabel +
-			" count=" + strconv.Itoa(g.Count))
-		if len(g.Sample) > 0 {
-			b.WriteString(" sample=" + strings.Join(g.Sample, ","))
-		}
-		b.WriteByte('\n')
-	}
-
-	b.WriteString("\n## Raw rows\n")
-	b.WriteString("prefix\tkind\tsource\tsource_detail\tcommunity_id\tcommunity_label\n")
-	for _, row := range rows {
-		b.WriteString(row.Prefix)
-		b.WriteByte('\t')
-		b.WriteString(row.Kind)
-		b.WriteByte('\t')
-		b.WriteString(row.Source)
-		b.WriteByte('\t')
-		b.WriteString(strings.ReplaceAll(row.SourceDetail, "\t", " "))
-		b.WriteByte('\t')
-		b.WriteString(row.CommunityID)
-		b.WriteByte('\t')
-		b.WriteString(strings.ReplaceAll(row.CommunityLabel, "\t", " "))
-		b.WriteByte('\n')
-	}
-
-	filename := "revision-" + shortRevisionID(rev.ID) + "-diagnostic.log"
 	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
-	w.Header().Set("Content-Disposition", `attachment; filename="`+filename+`"`)
+	w.Header().Set("Content-Disposition", `attachment; filename="`+logOut.Filename+`"`)
 	w.WriteHeader(http.StatusOK)
-	_, _ = w.Write([]byte(b.String()))
-}
-
-func classifyRevisionSource(src string) (kind, sourceName string) {
-	switch {
-	case strings.HasPrefix(src, "as:"):
-		return "asn", strings.TrimPrefix(src, "as:")
-	case strings.HasPrefix(src, "domain:"):
-		return "domain", strings.TrimPrefix(src, "domain:")
-	case strings.HasPrefix(src, "cdn:"):
-		return "cdn", strings.TrimPrefix(src, "cdn:")
-	case src == "ip_range":
-		return "ip_range", "manual_ranges"
-	default:
-		if strings.TrimSpace(src) == "" {
-			return "unknown", "unknown"
-		}
-		return "source", strings.TrimSpace(src)
-	}
-}
-
-func shortRevisionID(id string) string {
-	s := strings.TrimSpace(id)
-	if len(s) <= 8 {
-		return s
-	}
-	return s[:8]
+	_, _ = w.Write(logOut.Body)
 }
 
 func (s *Server) handleRevisionDiff(w http.ResponseWriter, r *http.Request) {

internal/httpapi/routes_crud.go

@@ -3,6 +3,7 @@ package httpapi
 import (
 	"encoding/csv"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -10,6 +11,7 @@ import (
 	"strings"
 	"time"
 
+	"evobgp/internal/importer"
 	"evobgp/internal/pipeline"
 	"evobgp/internal/store"
 )
@@ -689,163 +691,41 @@ func (s *Server) handleImportModuleEntriesCSV(w http.ResponseWriter, r *http.Req
 		return
 	}
 	moduleID := r.PathValue("module_id")
-	mod, err := s.store.GetModule(a.TenantID, moduleID)
+	res, err := importer.ImportModuleEntriesCSV(s.store, a.TenantID, moduleID, r.Body)
 	if err != nil {
+		if errors.Is(err, store.ErrInvalidInput) {
+			writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "csv is empty")
+			return
+		}
+		if strings.Contains(err.Error(), "importer: invalid csv") {
+			writeProblem(w, http.StatusBadRequest, "Bad Request", "invalid csv")
+			return
+		}
+		if strings.Contains(err.Error(), "importer: line") {
+			detail := strings.TrimPrefix(err.Error(), "importer: ")
+			writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", detail)
+			return
+		}
+		if strings.Contains(err.Error(), "importer: csv import/export") {
+			writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "csv import/export is supported only for AS_PREFIXES, DOMAINS, IP_RANGES")
+			return
+		}
 		writeStoreErr(w, err)
 		return
 	}
-
-	cr := csv.NewReader(io.LimitReader(r.Body, 8<<20))
-	cr.TrimLeadingSpace = true
-	cr.FieldsPerRecord = -1
-	rows, err := cr.ReadAll()
-	if err != nil {
-		writeProblem(w, http.StatusBadRequest, "Bad Request", "invalid csv")
-		return
-	}
-	if len(rows) == 0 {
-		writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "csv is empty")
-		return
-	}
-
-	communities, err := s.store.ListCommunities(a.TenantID)
-	if err != nil {
-		writeStoreErr(w, err)
-		return
-	}
-	communityIDByID := make(map[string]string, len(communities))
-	communityIDByValue := make(map[string]string, len(communities))
-	for _, c := range communities {
-		communityIDByID[c.ID] = c.ID
-		communityIDByValue[strings.TrimSpace(c.Community)] = c.ID
-	}
-
-	resolveCommunity := func(raw string, required bool) (*string, error) {
-		v := strings.TrimSpace(raw)
-		if v == "" {
-			if required {
-				return nil, fmt.Errorf("community is required")
-			}
-			return nil, nil
-		}
-		if id, ok := communityIDByID[v]; ok {
-			return &id, nil
-		}
-		if id, ok := communityIDByValue[v]; ok {
-			return &id, nil
-		}
-		return nil, fmt.Errorf("unknown community %q", v)
-	}
-
-	start := 0
-	if len(rows[0]) >= 2 {
-		key := strings.ToLower(strings.TrimSpace(rows[0][0]))
-		switch key {
-		case "asn", "domain", "iprange":
-			start = 1
-		}
-	}
-
-	imported := 0
-	switch mod.Type {
-	case "AS_PREFIXES":
-		for i := start; i < len(rows); i++ {
-			rec := rows[i]
-			if len(rec) == 0 || (strings.TrimSpace(rec[0]) == "" && (len(rec) < 2 || strings.TrimSpace(rec[1]) == "")) {
-				continue
-			}
-			if len(rec) < 2 {
-				writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", fmt.Sprintf("line %d: expected 2 columns", i+1))
-				return
-			}
-			asn, err := strconv.ParseInt(strings.TrimSpace(rec[0]), 10, 64)
-			if err != nil || asn <= 0 {
-				writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", fmt.Sprintf("line %d: invalid asn", i+1))
-				return
-			}
-			cid, err := resolveCommunity(rec[1], false)
-			if err != nil {
-				writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", fmt.Sprintf("line %d: %v", i+1, err))
-				return
-			}
-			_, err = s.store.CreateASEntry(a.TenantID, moduleID, &store.ASEntry{ASN: asn, CommunityID: cid})
-			if err != nil {
-				writeStoreErr(w, err)
-				return
-			}
-			imported++
-		}
-		if imported > 0 {
+	if res.Imported > 0 {
+		switch res.ModuleType {
+		case "AS_PREFIXES":
 			s.enqueueModuleRefreshIfEnabled(a.TenantID, moduleID, "as_entry_import_csv")
-		}
-	case "DOMAINS":
-		for i := start; i < len(rows); i++ {
-			rec := rows[i]
-			if len(rec) == 0 || (strings.TrimSpace(rec[0]) == "" && (len(rec) < 2 || strings.TrimSpace(rec[1]) == "")) {
-				continue
-			}
-			if len(rec) < 2 {
-				writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", fmt.Sprintf("line %d: expected 2 columns", i+1))
-				return
-			}
-			fqdn := strings.TrimSpace(rec[0])
-			if fqdn == "" {
-				writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", fmt.Sprintf("line %d: domain is required", i+1))
-				return
-			}
-			cid, err := resolveCommunity(rec[1], false)
-			if err != nil {
-				writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", fmt.Sprintf("line %d: %v", i+1, err))
-				return
-			}
-			_, err = s.store.CreateDomainEntry(a.TenantID, moduleID, &store.DomainEntry{FQDN: fqdn, CommunityID: cid})
-			if err != nil {
-				writeStoreErr(w, err)
-				return
-			}
-			imported++
-		}
-		if imported > 0 {
+		case "DOMAINS":
 			s.enqueueModuleRefreshIfEnabled(a.TenantID, moduleID, "domain_entry_import_csv")
-		}
-	case "IP_RANGES":
-		for i := start; i < len(rows); i++ {
-			rec := rows[i]
-			if len(rec) == 0 || (strings.TrimSpace(rec[0]) == "" && (len(rec) < 2 || strings.TrimSpace(rec[1]) == "")) {
-				continue
-			}
-			if len(rec) < 2 {
-				writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", fmt.Sprintf("line %d: expected 2 columns", i+1))
-				return
-			}
-			prefix := strings.TrimSpace(rec[0])
-			if prefix == "" {
-				writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", fmt.Sprintf("line %d: ipRange is required", i+1))
-				return
-			}
-			cid, err := resolveCommunity(rec[1], true)
-			if err != nil {
-				writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", fmt.Sprintf("line %d: %v", i+1, err))
-				return
-			}
-			_, err = s.store.CreateIPRangeEntry(a.TenantID, moduleID, &store.IPRangeEntry{Prefix: prefix, CommunityID: cid})
-			if err != nil {
-				writeStoreErr(w, err)
-				return
-			}
-			imported++
-		}
-		if imported > 0 {
+		case "IP_RANGES":
 			s.enqueueModuleRefreshIfEnabled(a.TenantID, moduleID, "ip_range_import_csv")
 		}
-	default:
-		writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "csv import/export is supported only for AS_PREFIXES, DOMAINS, IP_RANGES")
-		return
 	}
-
 	writeJSON(w, http.StatusOK, map[string]any{
-		"imported":    imported,
-		"module_type": mod.Type,
+		"imported":    res.Imported,
+		"module_type": res.ModuleType,
 	})
 }