From 7e1a1e76aefe9f2990b6a9356fece069d7143ef8 Mon Sep 17 00:00:00 2001 From: Lewis Date: Fri, 5 Jun 2026 09:31:00 +0300 Subject: [PATCH] appview/knotacl: knot acl client & cache Lewis: May this revision serve well! --- appview/knotacl/cache.go | 181 +++++++++++++++++++++++++++++++++++++ appview/knotacl/client.go | 134 +++++++++++++++++++++++++++ appview/xrpcclient/xrpc.go | 3 + 3 files changed, 318 insertions(+) create mode 100644 appview/knotacl/cache.go create mode 100644 appview/knotacl/client.go diff --git a/appview/knotacl/cache.go b/appview/knotacl/cache.go new file mode 100644 index 00000000..cb764dfd --- /dev/null +++ b/appview/knotacl/cache.go @@ -0,0 +1,181 @@ +package knotacl + +import ( + "context" + "maps" + "net/http" + "slices" + "sync" + "time" + + "golang.org/x/sync/singleflight" +) + +const ( + cacheTTL = 15 * time.Second + cacheMaxEntries = 4096 +) + +type lister interface { + GetKnotMembers(ctx context.Context, host string) ([]string, error) + GetRepoCollaborators(ctx context.Context, host, repoDid string) ([]string, error) +} + +type cacheEntry struct { + subjects []string + storedAt time.Time +} + +type cache struct { + inner lister + ttl time.Duration + now func() time.Time + + mu sync.Mutex + entries map[string]cacheEntry + group singleflight.Group +} + +func newCache(inner lister, ttl time.Duration, now func() time.Time) *cache { + if now == nil { + now = time.Now + } + return &cache{inner: inner, ttl: ttl, now: now, entries: map[string]cacheEntry{}} +} + +func (c *cache) GetKnotMembers(ctx context.Context, host string) ([]string, error) { + return c.fetch(ctx, memberCacheKey(host), func() ([]string, error) { + return c.inner.GetKnotMembers(ctx, host) + }) +} + +func (c *cache) GetRepoCollaborators(ctx context.Context, host, repoDid string) ([]string, error) { + return c.fetch(ctx, collabCacheKey(host, repoDid), func() ([]string, error) { + return c.inner.GetRepoCollaborators(ctx, host, repoDid) + }) +} + +func memberCacheKey(host string) string { return "m\x00" + host } + +func collabCacheKey(host, repoDid string) string { return "c\x00" + host + "\x00" + repoDid } + +func (c *cache) InvalidateMembers(host string) { + c.forget(memberCacheKey(host)) +} + +func (c *cache) InvalidateCollaborators(host, repoDid string) { + c.forget(collabCacheKey(host, repoDid)) +} + +func (c *cache) forget(key string) { + c.mu.Lock() + defer c.mu.Unlock() + delete(c.entries, key) +} + +func (c *cache) fetch(ctx context.Context, key string, load func() ([]string, error)) ([]string, error) { + if memo := memoFrom(ctx); memo != nil { + if v, ok := memo.get(key); ok { + return slices.Clone(v), nil + } + } + v, err := c.load(key, load) + if err != nil { + return nil, err + } + if memo := memoFrom(ctx); memo != nil { + memo.put(key, v) + } + return slices.Clone(v), nil +} + +func (c *cache) load(key string, load func() ([]string, error)) ([]string, error) { + if v, ok := c.lookup(key); ok { + return v, nil + } + v, err, _ := c.group.Do(key, func() (any, error) { + if v, ok := c.lookup(key); ok { + return v, nil + } + fresh, err := load() + if err != nil { + return nil, err + } + c.store(key, fresh) + return fresh, nil + }) + if err != nil { + return nil, err + } + return v.([]string), nil +} + +func (c *cache) lookup(key string) ([]string, bool) { + c.mu.Lock() + defer c.mu.Unlock() + e, ok := c.entries[key] + if !ok || c.now().Sub(e.storedAt) >= c.ttl { + return nil, false + } + return e.subjects, true +} + +func (c *cache) store(key string, subjects []string) { + c.mu.Lock() + defer c.mu.Unlock() + if _, exists := c.entries[key]; !exists && len(c.entries) >= cacheMaxEntries { + maps.DeleteFunc(c.entries, func(_ string, e cacheEntry) bool { + return c.now().Sub(e.storedAt) >= c.ttl + }) + c.evictOldestLocked() + } + c.entries[key] = cacheEntry{subjects: subjects, storedAt: c.now()} +} + +func (c *cache) evictOldestLocked() { + oldestKey := "" + var oldestAt time.Time + for k, e := range c.entries { + if oldestKey == "" || e.storedAt.Before(oldestAt) { + oldestKey, oldestAt = k, e.storedAt + } + } + if len(c.entries) >= cacheMaxEntries && oldestKey != "" { + delete(c.entries, oldestKey) + } +} + +type requestMemo struct { + mu sync.Mutex + entries map[string][]string +} + +type memoCtxKey struct{} + +func WithMemo(ctx context.Context) context.Context { + return context.WithValue(ctx, memoCtxKey{}, &requestMemo{entries: map[string][]string{}}) +} + +func memoFrom(ctx context.Context) *requestMemo { + memo, _ := ctx.Value(memoCtxKey{}).(*requestMemo) + return memo +} + +func (m *requestMemo) get(key string) ([]string, bool) { + m.mu.Lock() + defer m.mu.Unlock() + v, ok := m.entries[key] + return v, ok +} + +func (m *requestMemo) put(key string, subjects []string) { + m.mu.Lock() + defer m.mu.Unlock() + m.entries[key] = subjects +} + +func MemoMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + next.ServeHTTP(w, r.WithContext(WithMemo(r.Context()))) + }) +} diff --git a/appview/knotacl/client.go b/appview/knotacl/client.go new file mode 100644 index 00000000..fd63c2de --- /dev/null +++ b/appview/knotacl/client.go @@ -0,0 +1,134 @@ +package knotacl + +import ( + "context" + "fmt" + "log/slog" + "net/http" + "slices" + "time" + + indigoxrpc "github.com/bluesky-social/indigo/xrpc" + "tangled.org/core/api/tangled" +) + +const ( + listPageLimit = 1000 + maxListPages = 256 + requestTimeout = 5 * time.Second + listDrainBudget = 30 * time.Second +) + +type Client struct { + dev bool + http *http.Client + logger *slog.Logger +} + +func NewClient(dev bool, logger *slog.Logger) *Client { + return &Client{dev: dev, http: &http.Client{Timeout: requestTimeout}, logger: logger} +} + +func (c *Client) xrpcClient(host string) *indigoxrpc.Client { + scheme := "https" + if c.dev { + scheme = "http" + } + return &indigoxrpc.Client{ + Host: fmt.Sprintf("%s://%s", scheme, host), + Client: c.http, + } +} + +func (c *Client) GetKnotMembers(ctx context.Context, host string) ([]string, error) { + ctx, cancel := context.WithTimeout(ctx, listDrainBudget) + defer cancel() + + xc := c.xrpcClient(host) + subjects, truncated, err := drainList( + "", + make(map[string]struct{}), + func(cursor string) ([]*tangled.KnotListMembers_ListItem, *string, error) { + out, err := tangled.KnotListMembers(ctx, xc, cursor, listPageLimit, "", host) + if err != nil { + return nil, nil, err + } + return out.Items, out.Cursor, nil + }, + func(i *tangled.KnotListMembers_ListItem) string { return i.Subject }, + ) + if err != nil { + return nil, err + } + if truncated { + c.logger.Warn("knot member list truncated before draining all pages", "host", host, "limit", maxListPages) + } + return dedup(subjects), nil +} + +func (c *Client) GetRepoCollaborators(ctx context.Context, host, repoDid string) ([]string, error) { + ctx, cancel := context.WithTimeout(ctx, listDrainBudget) + defer cancel() + + xc := c.xrpcClient(host) + subjects, truncated, err := drainList( + "", + make(map[string]struct{}), + func(cursor string) ([]*tangled.RepoListCollaborators_ListItem, *string, error) { + out, err := tangled.RepoListCollaborators(ctx, xc, cursor, listPageLimit, "", repoDid) + if err != nil { + return nil, nil, err + } + return out.Items, out.Cursor, nil + }, + func(i *tangled.RepoListCollaborators_ListItem) string { return i.Subject }, + ) + if err != nil { + return nil, err + } + if truncated { + c.logger.Warn("repo collaborator list truncated before draining all pages", "host", host, "repoDid", repoDid, "limit", maxListPages) + } + return dedup(subjects), nil +} + +func drainList[T any]( + cursor string, + seen map[string]struct{}, + page func(cursor string) ([]*T, *string, error), + subject func(*T) string, +) (subjects []string, truncated bool, err error) { + if len(seen) >= maxListPages { + return nil, true, nil + } + if _, repeated := seen[cursor]; repeated { + return nil, true, nil + } + seen[cursor] = struct{}{} + items, next, err := page(cursor) + if err != nil { + return nil, false, err + } + subjects = mapSlice(items, subject) + if len(items) == 0 || next == nil || *next == "" { + return subjects, false, nil + } + rest, truncated, err := drainList(*next, seen, page, subject) + if err != nil { + return nil, false, err + } + return append(subjects, rest...), truncated, nil +} + +func dedup(subjects []string) []string { + slices.Sort(subjects) + return slices.Compact(subjects) +} + +func mapSlice[T, U any](items []T, f func(T) U) []U { + out := make([]U, len(items)) + for i, it := range items { + out[i] = f(it) + } + return out +} diff --git a/appview/xrpcclient/xrpc.go b/appview/xrpcclient/xrpc.go index 4e0ca538..3d492e07 100644 --- a/appview/xrpcclient/xrpc.go +++ b/appview/xrpcclient/xrpc.go @@ -10,6 +10,7 @@ import ( var ( ErrXrpcUnsupported = errors.New("xrpc not supported on this knot") ErrXrpcUnauthorized = errors.New("unauthorized xrpc request") + ErrXrpcForbidden = errors.New("forbidden xrpc request") ErrXrpcFailed = errors.New("xrpc request failed") ErrXrpcInvalid = errors.New("invalid xrpc request") ) @@ -30,6 +31,8 @@ func HandleXrpcErr(err error) error { return ErrXrpcUnsupported case http.StatusUnauthorized: return ErrXrpcUnauthorized + case http.StatusForbidden: + return ErrXrpcForbidden default: return ErrXrpcFailed } -- 2.51.2