refactor: update worker processes in EvoBGP to accept dependencies for shared store and job registry. Enhance scheduler, ingest, render, and deploy components to utilize a unified context and improve logging for drift detection. Update architecture documentation to reflect changes in process interactions and worker functionalities.

internal/pipeline/refresh.go

@@ -0,0 +1,214 @@
+package pipeline
+
+import (
+	"context"
+	"crypto/sha256"
+	"fmt"
+	"io"
+	"net/http"
+	"net/netip"
+	"sort"
+	"strings"
+
+	"evobgp/internal/birdfmt"
+	"evobgp/internal/store"
+
+	"github.com/google/uuid"
+)
+
+// RefreshModule runs ingest (where applicable) and creates a new rendered revision for the module.
+func RefreshModule(ctx context.Context, st store.Backend, hc *http.Client, tenantID, moduleID string) (revisionID string, err error) {
+	if hc == nil {
+		hc = http.DefaultClient
+	}
+	mod, err := st.GetModule(tenantID, moduleID)
+	if err != nil {
+		return "", err
+	}
+	if !mod.Enabled {
+		return "", fmt.Errorf("module disabled")
+	}
+
+	var rows []store.PrefixRow
+	switch mod.Type {
+	case "IP_RANGES":
+		list, err := st.ListIPRangeEntries(tenantID, moduleID)
+		if err != nil {
+			return "", err
+		}
+		for _, e := range list {
+			comm := e.CommunityID
+			if comm == nil && mod.DefaultCommunityID != nil {
+				c := *mod.DefaultCommunityID
+				comm = &c
+			}
+			rows = append(rows, store.PrefixRow{Prefix: e.Prefix, CommunityID: comm, Source: "ip_range"})
+		}
+	case "AS_PREFIXES":
+		list, err := st.ListASEntries(tenantID, moduleID)
+		if err != nil {
+			return "", err
+		}
+		for _, e := range list {
+			if e.Prefix == nil || strings.TrimSpace(*e.Prefix) == "" {
+				continue
+			}
+			comm := e.CommunityID
+			if comm == nil && mod.DefaultCommunityID != nil {
+				c := *mod.DefaultCommunityID
+				comm = &c
+			}
+			rows = append(rows, store.PrefixRow{Prefix: strings.TrimSpace(*e.Prefix), CommunityID: comm, Source: "as_entry"})
+		}
+	case "CDN_CIDRS":
+		sources, err := st.ListCDNSources(tenantID, moduleID)
+		if err != nil {
+			return "", err
+		}
+		for _, src := range sources {
+			u := strings.TrimSpace(src.URL)
+			if u == "" {
+				continue
+			}
+			req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
+			if err != nil {
+				return "", err
+			}
+			if strings.TrimSpace(src.Etag) != "" {
+				req.Header.Set("If-None-Match", strings.TrimSpace(src.Etag))
+			}
+			resp, err := hc.Do(req)
+			if err != nil {
+				return "", fmt.Errorf("cdn fetch %s: %w", u, err)
+			}
+			if resp.StatusCode == http.StatusNotModified {
+				_ = resp.Body.Close()
+				continue
+			}
+			if resp.StatusCode != http.StatusOK {
+				_, _ = io.Copy(io.Discard, resp.Body)
+				_ = resp.Body.Close()
+				return "", 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 "", err
+			}
+			etag := strings.TrimSpace(resp.Header.Get("ETag"))
+			if etag != "" && etag != strings.TrimSpace(src.Etag) {
+				e := etag
+				_, _ = st.UpdateCDNSource(tenantID, moduleID, src.ID, &store.CDNSourcePatch{Etag: &e})
+			}
+			for _, pfx := range ParseCIDRLines(string(body)) {
+				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: "cdn:" + src.ID})
+			}
+		}
+	case "DOMAINS":
+		if _, err := st.ListDomainEntries(tenantID, moduleID); err != nil {
+			return "", err
+		}
+		// DNS/DoH resolution not wired yet; emit empty prefix set (valid revision).
+	default:
+		return "", fmt.Errorf("unknown module type %q", mod.Type)
+	}
+
+	revisionID = uuid.NewString()
+	parent := parentRevision(st, tenantID, moduleID)
+	hash := hashMaterialization(moduleID, rows)
+	preview, err := buildPreviewFragments(revisionID, rows)
+	if err != nil {
+		return "", err
+	}
+	if err := st.CreateRenderRevision(revisionID, tenantID, moduleID, parent, hash, preview, rows); err != nil {
+		return "", err
+	}
+	return revisionID, nil
+}
+
+func parentRevision(st store.Backend, tenantID, moduleID string) *string {
+	items, _, _ := st.ListRevisions(tenantID, moduleID, "", 1)
+	if len(items) == 0 {
+		return nil
+	}
+	id := items[0].ID
+	return &id
+}
+
+func hashMaterialization(moduleID string, rows []store.PrefixRow) string {
+	type line struct{ p, c, s string }
+	var lines []line
+	for _, r := range rows {
+		c := ""
+		if r.CommunityID != nil {
+			c = *r.CommunityID
+		}
+		lines = append(lines, line{r.Prefix, c, r.Source})
+	}
+	sort.Slice(lines, func(i, j int) bool {
+		if lines[i].p != lines[j].p {
+			return lines[i].p < lines[j].p
+		}
+		if lines[i].c != lines[j].c {
+			return lines[i].c < lines[j].c
+		}
+		return lines[i].s < lines[j].s
+	})
+	h := sha256.New()
+	h.Write([]byte(moduleID))
+	h.Write([]byte{0})
+	for _, l := range lines {
+		h.Write([]byte(l.p))
+		h.Write([]byte{1})
+		h.Write([]byte(l.c))
+		h.Write([]byte{1})
+		h.Write([]byte(l.s))
+		h.Write([]byte{0})
+	}
+	return fmt.Sprintf("sha256:%x", h.Sum(nil))
+}
+
+func buildPreviewFragments(revisionID string, rows []store.PrefixRow) (map[string]string, error) {
+	var v4, v6 []netip.Prefix
+	for _, pr := range rows {
+		pfx, err := netip.ParsePrefix(strings.TrimSpace(pr.Prefix))
+		if err != nil {
+			continue
+		}
+		if pfx.Addr().Is4() {
+			v4 = append(v4, pfx.Masked())
+		} else {
+			v6 = append(v6, pfx.Masked())
+		}
+	}
+	f4, err := birdfmt.RenderExportFilterIPv4("evobgp_export_v4", v4)
+	if err != nil {
+		return nil, err
+	}
+	f6, err := birdfmt.RenderExportFilterIPv6("evobgp_export_v6", v6)
+	if err != nil {
+		return nil, err
+	}
+	birdD := birdfmt.JoinFragments(birdfmt.ManagedBanner(revisionID), f4, f6)
+	main := `# EvoBGP generated (pipeline refresh)
+router id 192.0.2.1;
+include "bird.d/evobgp_generated.conf";
+
+protocol device {
+}
+
+protocol direct {
+    ipv4;
+    ipv6;
+}
+`
+	return map[string]string{
+		"bird.conf":                    main,
+		"bird.d/evobgp_generated.conf": birdD,
+	}, nil
+}