From 5249c9eaab98ac44ed3e466a02178d36b5dd7c25 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sun, 22 Feb 2026 12:06:02 -0600 Subject: [PATCH] add relay-compare tool --- cmd/relay-compare/main.go | 292 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 292 insertions(+) create mode 100644 cmd/relay-compare/main.go diff --git a/cmd/relay-compare/main.go b/cmd/relay-compare/main.go new file mode 100644 index 0000000..ede8ac3 --- /dev/null +++ b/cmd/relay-compare/main.go @@ -0,0 +1,292 @@ +// relay-compare compares ATProto relays by querying listReposByCollection +// for all io.atcr.* record types and showing what's missing from each relay. +// +// Usage: +// +// go run ./cmd/relay-compare https://relay1.us-east.bsky.network https://relay1.us-west.bsky.network +package main + +import ( + "context" + "flag" + "fmt" + "net/url" + "os" + "sort" + "strings" + "sync" + "time" + + "atcr.io/pkg/atproto" +) + +// ANSI color codes (disabled via --no-color or NO_COLOR env) +var ( + cRed = "\033[31m" + cGreen = "\033[32m" + cYellow = "\033[33m" + cCyan = "\033[36m" + cBold = "\033[1m" + cDim = "\033[2m" + cReset = "\033[0m" +) + +func disableColors() { + cRed, cGreen, cYellow, cCyan, cBold, cDim, cReset = "", "", "", "", "", "", "" +} + +// All io.atcr.* collections to compare +var allCollections = []string{ + atproto.ManifestCollection, // io.atcr.manifest + atproto.TagCollection, // io.atcr.tag + atproto.SailorProfileCollection, // io.atcr.sailor.profile + atproto.StarCollection, // io.atcr.sailor.star + atproto.SailorWebhookCollection, // io.atcr.sailor.webhook + atproto.RepoPageCollection, // io.atcr.repo.page + atproto.CaptainCollection, // io.atcr.hold.captain + atproto.CrewCollection, // io.atcr.hold.crew + atproto.LayerCollection, // io.atcr.hold.layer + atproto.StatsCollection, // io.atcr.hold.stats + atproto.ScanCollection, // io.atcr.hold.scan + atproto.WebhookCollection, // io.atcr.hold.webhook +} + +type summaryRow struct { + collection string + counts []int + status string // "sync", "diff", "error" + diffCount int +} + +func main() { + noColor := flag.Bool("no-color", false, "disable colored output") + collection := flag.String("collection", "", "compare only this collection") + timeout := flag.Duration("timeout", 2*time.Minute, "timeout for all relay queries") + flag.Usage = func() { + fmt.Fprintf(os.Stderr, "Compare ATProto relays by querying listReposByCollection for io.atcr.* records.\n\n") + fmt.Fprintf(os.Stderr, "Usage:\n relay-compare [flags] [relay-url...]\n\n") + fmt.Fprintf(os.Stderr, "Example:\n") + fmt.Fprintf(os.Stderr, " go run ./cmd/relay-compare https://relay1.us-east.bsky.network https://relay1.us-west.bsky.network\n\n") + fmt.Fprintf(os.Stderr, "Flags:\n") + flag.PrintDefaults() + } + flag.Parse() + + if *noColor || os.Getenv("NO_COLOR") != "" { + disableColors() + } + + relays := flag.Args() + if len(relays) < 2 { + flag.Usage() + os.Exit(1) + } + + for i, r := range relays { + relays[i] = strings.TrimRight(r, "/") + } + + cols := allCollections + if *collection != "" { + cols = []string{*collection} + } + + ctx, cancel := context.WithTimeout(context.Background(), *timeout) + defer cancel() + + // Short display names for each relay + names := make([]string, len(relays)) + maxNameLen := 0 + for i, r := range relays { + names[i] = shortName(r) + if len(names[i]) > maxNameLen { + maxNameLen = len(names[i]) + } + } + + fmt.Printf("%sFetching %d collections from %d relays...%s\n", cDim, len(cols), len(relays), cReset) + + // Fetch all data in parallel: every (collection, relay) pair concurrently + type key struct{ col, relay string } + type fetchResult struct { + dids map[string]struct{} + err error + } + allResults := make(map[key]fetchResult) + var mu sync.Mutex + var wg sync.WaitGroup + + for _, col := range cols { + for _, relay := range relays { + wg.Add(1) + go func(col, relay string) { + defer wg.Done() + dids, err := fetchAllDIDs(ctx, relay, col) + mu.Lock() + allResults[key{col, relay}] = fetchResult{dids, err} + mu.Unlock() + }(col, relay) + } + } + wg.Wait() + + // Display per-collection diffs and collect summary + var summary []summaryRow + totalMissing := 0 + + for _, col := range cols { + fmt.Printf("\n%s%s━━━ %s ━━━%s\n", cBold, cCyan, col, cReset) + + row := summaryRow{collection: col, counts: make([]int, len(relays))} + hasError := false + + // Show counts per relay + for ri, relay := range relays { + r := allResults[key{col, relay}] + if r.err != nil { + hasError = true + fmt.Printf(" %-*s %s%serror%s: %v\n", maxNameLen, names[ri], cBold, cRed, cReset, r.err) + } else { + row.counts[ri] = len(r.dids) + fmt.Printf(" %-*s %s%d%s DIDs\n", maxNameLen, names[ri], cBold, len(r.dids), cReset) + } + } + + if hasError { + row.status = "error" + summary = append(summary, row) + continue + } + + // Build union of all DIDs across relays + union := make(map[string]struct{}) + for _, relay := range relays { + for did := range allResults[key{col, relay}].dids { + union[did] = struct{}{} + } + } + + // For each relay, show what it's missing + inSync := true + for ri, relay := range relays { + var missing []string + for did := range union { + if _, ok := allResults[key{col, relay}].dids[did]; !ok { + missing = append(missing, did) + } + } + if len(missing) == 0 { + continue + } + + inSync = false + totalMissing += len(missing) + row.diffCount += len(missing) + sort.Strings(missing) + + fmt.Printf("\n %sMissing from %s (%d):%s\n", cRed, names[ri], len(missing), cReset) + for _, did := range missing { + fmt.Printf(" %s- %s%s\n", cRed, did, cReset) + } + } + + if inSync { + fmt.Printf(" %s✓ in sync%s\n", cGreen, cReset) + row.status = "sync" + } else { + row.status = "diff" + } + summary = append(summary, row) + } + + // Summary table + printSummary(summary, names, maxNameLen, totalMissing) +} + +func printSummary(rows []summaryRow, names []string, maxNameLen, totalMissing int) { + fmt.Printf("\n%s%s━━━ Summary ━━━%s\n\n", cBold, cCyan, cReset) + + colW := 28 + relayW := maxNameLen + 2 + if relayW < 8 { + relayW = 8 + } + + // Header + fmt.Printf(" %-*s", colW, "Collection") + for _, name := range names { + fmt.Printf(" %*s", relayW, name) + } + fmt.Printf(" Status\n") + + // Separator + fmt.Printf(" %s", strings.Repeat("─", colW)) + for range names { + fmt.Printf(" %s", strings.Repeat("─", relayW)) + } + fmt.Printf(" %s\n", strings.Repeat("─", 14)) + + // Data rows + for _, row := range rows { + fmt.Printf(" %-*s", colW, row.collection) + for _, c := range row.counts { + switch row.status { + case "error": + fmt.Printf(" %*s", relayW, fmt.Sprintf("%s—%s", cDim, cReset)) + default: + fmt.Printf(" %*d", relayW, c) + } + } + switch row.status { + case "sync": + fmt.Printf(" %s✓ in sync%s", cGreen, cReset) + case "diff": + fmt.Printf(" %s≠ %d missing%s", cYellow, row.diffCount, cReset) + case "error": + fmt.Printf(" %s✗ error%s", cRed, cReset) + } + fmt.Println() + } + + // Footer + fmt.Println() + if totalMissing > 0 { + fmt.Printf("%s%d total missing DID-collection pairs across relays%s\n", cYellow, totalMissing, cReset) + } else { + fmt.Printf("%s✓ All relays fully in sync%s\n", cGreen, cReset) + } +} + +// fetchAllDIDs paginates through listReposByCollection to collect all DIDs. +func fetchAllDIDs(ctx context.Context, relay, collection string) (map[string]struct{}, error) { + client := atproto.NewClient(relay, "", "") + dids := make(map[string]struct{}) + var cursor string + + for { + result, err := client.ListReposByCollection(ctx, collection, 1000, cursor) + if err != nil { + return dids, err + } + + for _, repo := range result.Repos { + dids[repo.DID] = struct{}{} + } + + if result.Cursor == "" { + break + } + cursor = result.Cursor + } + + return dids, nil +} + +// shortName extracts the hostname from a relay URL for display. +func shortName(relayURL string) string { + u, err := url.Parse(relayURL) + if err != nil { + return relayURL + } + return u.Hostname() +}