Monorepo for Tangled
tangled.org
1package knotacl
2
3import (
4 "context"
5 "fmt"
6 "log/slog"
7 "net/http"
8 "slices"
9 "time"
10
11 indigoxrpc "github.com/bluesky-social/indigo/xrpc"
12 "tangled.org/core/api/tangled"
13)
14
15const (
16 listPageLimit = 1000
17 maxListPages = 256
18 requestTimeout = 5 * time.Second
19 listDrainBudget = 30 * time.Second
20)
21
22type Client struct {
23 dev bool
24 http *http.Client
25 logger *slog.Logger
26}
27
28func NewClient(dev bool, logger *slog.Logger) *Client {
29 return &Client{dev: dev, http: &http.Client{Timeout: requestTimeout}, logger: logger}
30}
31
32func (c *Client) xrpcClient(host string) *indigoxrpc.Client {
33 scheme := "https"
34 if c.dev {
35 scheme = "http"
36 }
37 return &indigoxrpc.Client{
38 Host: fmt.Sprintf("%s://%s", scheme, host),
39 Client: c.http,
40 }
41}
42
43func (c *Client) GetKnotMembers(ctx context.Context, host string) ([]string, error) {
44 ctx, cancel := context.WithTimeout(ctx, listDrainBudget)
45 defer cancel()
46
47 xc := c.xrpcClient(host)
48 subjects, truncated, err := drainList(
49 "",
50 make(map[string]struct{}),
51 func(cursor string) ([]*tangled.KnotListMembers_ListItem, *string, error) {
52 out, err := tangled.KnotListMembers(ctx, xc, cursor, listPageLimit, "", host)
53 if err != nil {
54 return nil, nil, err
55 }
56 return out.Items, out.Cursor, nil
57 },
58 func(i *tangled.KnotListMembers_ListItem) string { return i.Subject },
59 )
60 if err != nil {
61 return nil, err
62 }
63 if truncated {
64 c.logger.Warn("knot member list truncated before draining all pages", "host", host, "limit", maxListPages)
65 }
66 return dedup(subjects), nil
67}
68
69func (c *Client) GetRepoCollaborators(ctx context.Context, host, repoDid string) ([]string, error) {
70 ctx, cancel := context.WithTimeout(ctx, listDrainBudget)
71 defer cancel()
72
73 xc := c.xrpcClient(host)
74 subjects, truncated, err := drainList(
75 "",
76 make(map[string]struct{}),
77 func(cursor string) ([]*tangled.RepoListCollaborators_ListItem, *string, error) {
78 out, err := tangled.RepoListCollaborators(ctx, xc, cursor, listPageLimit, "", repoDid)
79 if err != nil {
80 return nil, nil, err
81 }
82 return out.Items, out.Cursor, nil
83 },
84 func(i *tangled.RepoListCollaborators_ListItem) string { return i.Subject },
85 )
86 if err != nil {
87 return nil, err
88 }
89 if truncated {
90 c.logger.Warn("repo collaborator list truncated before draining all pages", "host", host, "repoDid", repoDid, "limit", maxListPages)
91 }
92 return dedup(subjects), nil
93}
94
95func drainList[T any](
96 cursor string,
97 seen map[string]struct{},
98 page func(cursor string) ([]*T, *string, error),
99 subject func(*T) string,
100) (subjects []string, truncated bool, err error) {
101 if len(seen) >= maxListPages {
102 return nil, true, nil
103 }
104 if _, repeated := seen[cursor]; repeated {
105 return nil, true, nil
106 }
107 seen[cursor] = struct{}{}
108 items, next, err := page(cursor)
109 if err != nil {
110 return nil, false, err
111 }
112 subjects = mapSlice(items, subject)
113 if len(items) == 0 || next == nil || *next == "" {
114 return subjects, false, nil
115 }
116 rest, truncated, err := drainList(*next, seen, page, subject)
117 if err != nil {
118 return nil, false, err
119 }
120 return append(subjects, rest...), truncated, nil
121}
122
123func dedup(subjects []string) []string {
124 slices.Sort(subjects)
125 return slices.Compact(subjects)
126}
127
128func mapSlice[T, U any](items []T, f func(T) U) []U {
129 out := make([]U, len(items))
130 for i, it := range items {
131 out[i] = f(it)
132 }
133 return out
134}