feat: enhance CDN source management with last refreshed timestamp

Added functionality to track the last refreshed timestamp for CDN sources. Updated the database schema and relevant methods to include the last refreshed timestamp during creation and updates. Implemented logic to skip fetching CDN sources based on their refresh interval, improving efficiency in the module prefix collection process. Enhanced the data retrieval methods to support the new timestamp field, ensuring accurate state management for CDN sources.

internal/pipeline/refresh.go

@@ -200,7 +200,16 @@ func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Cli
 			return nil, err
 		}
 		var rows []store.PrefixRow
+		latestCDNRows := latestCDNRowsBySource(st, tenantID)
 		for _, src := range sources {
+			sourceKey := "cdn:" + src.ID
+			now := time.Now().UTC()
+			if shouldSkipCDNSourceFetch(src, now) {
+				if cached := latestCDNRows[sourceKey]; len(cached) > 0 {
+					rows = append(rows, cached...)
+					continue
+				}
+			}
 			u := strings.TrimSpace(src.URL)
 			if u == "" {
 				continue
@@ -224,10 +233,14 @@ func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Cli
 				return nil, err
 			}
 			etag := strings.TrimSpace(resp.Header.Get("ETag"))
+			patch := &store.CDNSourcePatch{}
 			if etag != "" && etag != strings.TrimSpace(src.Etag) {
 				e := etag
-				_, _ = st.UpdateCDNSource(tenantID, moduleID, src.ID, &store.CDNSourcePatch{Etag: &e})
+				patch.Etag = &e
 			}
+			refreshedAt := now
+			patch.LastRefreshedAt = &refreshedAt
+			_, _ = st.UpdateCDNSource(tenantID, moduleID, src.ID, patch)
 			pfxs, err := ExtractCIDRs(string(body), src.SourceKind, src.PrefixPath)
 			if err != nil {
 				return nil, fmt.Errorf("cdn parse %s: %w", u, err)
@@ -238,7 +251,7 @@ func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Cli
 					c := *mod.DefaultCommunityID
 					comm = &c
 				}
-				rows = append(rows, store.PrefixRow{Prefix: pfx.String(), CommunityID: comm, Source: "cdn:" + src.ID})
+				rows = append(rows, store.PrefixRow{Prefix: pfx.String(), CommunityID: comm, Source: sourceKey})
 			}
 		}
 		return rows, nil
@@ -293,6 +306,41 @@ func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Cli
 	}
 }
 
+func shouldSkipCDNSourceFetch(src *store.CDNSource, now time.Time) bool {
+	if src == nil || src.RefreshIntervalSec == nil || *src.RefreshIntervalSec <= 0 || src.LastRefreshedAt == nil {
+		return false
+	}
+	nextRefreshAt := src.LastRefreshedAt.UTC().Add(time.Duration(*src.RefreshIntervalSec) * time.Second)
+	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"`