test: add CDN prefetch tests for 304 responses and snapshot handling

- Introduced `TestRefreshModuleIngest_CDN304UsesStoredSnapshot` to verify that the module ingest correctly uses stored snapshots when receiving a 304 Not Modified response.
- Added `TestCollectModulePrefixRows_CDN304RetriesWithoutETag` to ensure that the system retries without the If-None-Match header after a 304 response, and correctly collects prefixes on subsequent successful requests.
- Updated `cachedCDNPrefixRows` function to improve handling of cached prefixes during CDN source processing.
- Modified `RefreshModuleIngest` to utilize prior snapshots when collecting module prefix rows.

internal/pipeline/cdn_prefetch_test.go

@@ -97,3 +97,96 @@ func TestCollectModulePrefixRows_CDN304UsesSnapshot(t *testing.T) {
 		t.Fatalf("want cached prefix on 304, got %+v", collected)
 	}
 }
+
+func TestRefreshModuleIngest_CDN304UsesStoredSnapshot(t *testing.T) {
+	m := store.NewMemory()
+	m.SeedDemo()
+	tenant, _, _, _, _ := m.DemoIDs()
+
+	mod, err := m.CreateModule(tenant, &store.Module{
+		Type:    "CDN_CIDRS",
+		Name:    "cdn-304-ingest",
+		Enabled: true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusNotModified)
+	}))
+	defer srv.Close()
+
+	etag := "etag-stable"
+	src, err := m.CreateCDNSource(tenant, mod.ID, &store.CDNSource{
+		SourceKind: "txt",
+		URL:        srv.URL,
+		Etag:       etag,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	prior := []store.PrefixRow{{
+		Prefix: "203.0.113.0/24",
+		Source: cdnSourceKey(src.ID),
+	}}
+	if err := mergeCDNSourceIntoModuleSnapshot(m, tenant, mod, src.ID, prior); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := RefreshModuleIngest(context.Background(), m, srv.Client(), tenant, mod.ID); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestCollectModulePrefixRows_CDN304RetriesWithoutETag(t *testing.T) {
+	m := store.NewMemory()
+	m.SeedDemo()
+	tenant, _, _, _, _ := m.DemoIDs()
+
+	mod, err := m.CreateModule(tenant, &store.Module{
+		Type:    "CDN_CIDRS",
+		Name:    "cdn-304-retry",
+		Enabled: true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var calls int
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls++
+		if calls == 1 {
+			if got := strings.TrimSpace(r.Header.Get("If-None-Match")); got != "etag-stable" {
+				t.Fatalf("first request want If-None-Match etag-stable, got %q", got)
+			}
+			w.WriteHeader(http.StatusNotModified)
+			return
+		}
+		if got := strings.TrimSpace(r.Header.Get("If-None-Match")); got != "" {
+			t.Fatalf("retry must omit If-None-Match, got %q", got)
+		}
+		w.Header().Set("ETag", "etag-stable")
+		_, _ = w.Write([]byte("198.51.100.0/24\n"))
+	}))
+	defer srv.Close()
+
+	if _, err := m.CreateCDNSource(tenant, mod.ID, &store.CDNSource{
+		SourceKind: "txt",
+		URL:        srv.URL,
+		Etag:       "etag-stable",
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	collected, err := collectModulePrefixRows(context.Background(), m, srv.Client(), tenant, mod, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if calls != 2 {
+		t.Fatalf("want 2 HTTP calls (304 then 200), got %d", calls)
+	}
+	if len(collected) != 1 || collected[0].Prefix != "198.51.100.0/24" {
+		t.Fatalf("unexpected collected rows: %+v", collected)
+	}
+}

internal/pipeline/cdn_snapshot.go

@@ -15,6 +15,23 @@ func cdnSourceKey(sourceID string) string {
 	return "cdn:" + strings.TrimSpace(sourceID)
 }
 
+func cachedCDNPrefixRows(st store.Backend, tenantID, moduleID string, priorSnapshot []store.PrefixRow, sourceKey string) []store.PrefixRow {
+	if cached := prefixRowsForSource(priorSnapshot, sourceKey); len(cached) > 0 {
+		return cached
+	}
+	if st != nil {
+		if snap, ok, _ := st.GetModulePrefixSnapshot(tenantID, moduleID); ok && snap != nil {
+			if cached := prefixRowsForSource(snap.Prefixes, sourceKey); len(cached) > 0 {
+				return cached
+			}
+		}
+		if cached := latestCDNRowsBySource(st, tenantID)[sourceKey]; len(cached) > 0 {
+			return cached
+		}
+	}
+	return nil
+}
+
 func mergeSnapshotDropSource(rows []store.PrefixRow, sourceKey string) []store.PrefixRow {
 	if len(rows) == 0 {
 		return nil
@@ -88,14 +105,29 @@ func applyCDNSourceHTTPResult(ctx context.Context, st store.Backend, hc *http.Cl
 	if err != nil {
 		return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
 	}
-	defer resp.Body.Close()
 
 	if resp.StatusCode == http.StatusNotModified {
-		if cached := prefixRowsForSource(priorSnapshot, sourceKey); len(cached) > 0 {
+		if cached := cachedCDNPrefixRows(st, tenantID, moduleID, priorSnapshot, sourceKey); len(cached) > 0 {
+			_ = resp.Body.Close()
 			return cached, nil
 		}
-		return nil, fmt.Errorf("cdn url %s: 304 without cached prefixes", u)
+		// ETag is known but local snapshot is empty — force a full download.
+		_ = resp.Body.Close()
+		req2, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
+		if err != nil {
+			return nil, err
+		}
+		resp, err = hc.Do(req2)
+		if err != nil {
+			return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
+		}
+		if resp.StatusCode == http.StatusNotModified {
+			_ = resp.Body.Close()
+			return nil, fmt.Errorf("cdn url %s: 304 without cached prefixes", u)
+		}
 	}
+	defer resp.Body.Close()
+
 	if resp.StatusCode != http.StatusOK {
 		_, _ = io.Copy(io.Discard, resp.Body)
 		return nil, fmt.Errorf("cdn url %s: %s", u, resp.Status)

internal/pipeline/refresh.go

@@ -53,7 +53,11 @@ func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client,
 		return fmt.Errorf("module disabled")
 	}
 
-	rows, err := collectModulePrefixRows(ctx, st, hc, tenantID, mod, nil)
+	var prior []store.PrefixRow
+	if snap, ok, _ := st.GetModulePrefixSnapshot(tenantID, moduleID); ok && snap != nil {
+		prior = snap.Prefixes
+	}
+	rows, err := collectModulePrefixRows(ctx, st, hc, tenantID, mod, prior)
 	if err != nil {
 		return err
 	}