diff --git a/FOLLOWUPS.md b/FOLLOWUPS.md index 2eaacfe..e7f5215 100644 --- a/FOLLOWUPS.md +++ b/FOLLOWUPS.md @@ -133,3 +133,11 @@ task documents and git history rather than this list. - The e2e suite has not yet completed on GitHub Actions. It cold-builds Lemmy from source and may exceed practical runner time/disk limits. Prefer a prebuilt pinned Lemmy debug image in GHCR, or persist a buildx GHA cache. +- The declarative follow list (FOLLOW_LIST_PATH) has no e2e coverage: an + end-to-end convergence test would mount a follow-list YAML into the + tidepool compose service, restart it, and poll GET /admin/communities + until the listed community turns `accepted`, then remove the entry and + assert an Undo lands at Lemmy. The reconciler is unit-tested against the + ingest harness (real Postgres + fake Lemmy inbox) and was smoke-tested + live (startup fail-fast, sweep, POST /admin/communities/reconcile); + compose plumbing for the file mount is the missing piece. diff --git a/README.md b/README.md index 111c21a..f5f928a 100644 --- a/README.md +++ b/README.md @@ -264,6 +264,8 @@ production**: | `INBOX_TOMBSTONE_CONFIRMS_PER_MINUTE` / `INBOX_TOMBSTONE_CONFIRM_BURST` | `6` / `10` | dedicated per-IP cap on the tombstoned-self-delete confirmation branch (an unauthenticated POST that costs an outbound fetch + durable writes); over-limit deliveries defer (503) so legitimate deletions redeliver | | `SYNC_RATE_PER_SECOND` / `SYNC_RATE_BURST` | `25` / `200` | per-client-IP token bucket over the public `com.atproto.sync.*` surface (429; `_health` exempt) | | `SYNC_MAX_SUBSCRIBERS` | `100` | concurrent `subscribeRepos` connection cap | +| `FOLLOW_LIST_PATH` | *(optional)* | declarative follow list (see below); unset = the `/admin` API is the only subscription control | +| `FOLLOW_LIST_INTERVAL` | `15m` | follow-list reconciler sweep cadence | ## Subscribing to communities (admin API) @@ -298,6 +300,35 @@ curl localhost:8091/admin/metrics \ # where validation failures log-and-write, means investigate) ``` +### Declarative follow list (`FOLLOW_LIST_PATH`) + +Instead of driving subscriptions one curl at a time, point +`FOLLOW_LIST_PATH` at a repo-committed YAML naming every community the +bridge should follow (see [communities.yaml](communities.yaml)): + +```yaml +communities: + - "!comicstrips@lemmy.world" # entries MUST be quoted — bare ! is a YAML tag + - "https://lemmy.ml/c/linux" # AP group URLs work too +``` + +A reconciler converges the subscription table to the file on startup and +every `FOLLOW_LIST_INTERVAL` (`POST /admin/communities/reconcile` forces a +pass and reports what changed). **The file is authoritative**: entries +missing from the table are subscribed; subscriptions missing from the file +are unfollowed (`Undo{Follow}`, records kept — content just stops flowing) +— including manual `POST /admin/communities` additions. Community consent +(`#nobridge`) still overrides the file. + +Git history is the moderation audit log: additions and removals arrive as +reviewed PRs (requests via GitHub issue), each entry carrying its rationale +in a comment. Guard rails, so a bad deploy can't mass-unfollow: a missing +or malformed file **fails startup**; a file that breaks after startup skips +sweeps (never "unfollow everything"); an entry that stops matching is +diffed against the table offline, so a resolver or remote-instance outage +can't make a desired community look removed. Only an explicit +`communities: []` unfollows all. + The bridge's AP face lives next to the inbox: the service actor document at `/actor`, an **instance actor at the origin apex** (`GET /`, type `Application` — Lemmy resolves every peer's "Site" actor there and delivers diff --git a/cmd/tidepool/main.go b/cmd/tidepool/main.go index 55d676e..a090fae 100644 --- a/cmd/tidepool/main.go +++ b/cmd/tidepool/main.go @@ -397,6 +397,29 @@ func run(logger *slog.Logger) error { } admin.Routes(router) + // Declarative follow list (FOLLOW_LIST_PATH): converge subscriptions to + // the repo-committed YAML on startup and every FOLLOW_LIST_INTERVAL. A + // missing or malformed file fails startup (fail fast on a bad deploy, + // like the migration gate); the reconciler goroutine itself only ever + // logs — a file that breaks AFTER startup skips sweeps rather than + // unfollowing anything. + if cfg.FollowListPath != "" { + if _, err := ingest.ParseFollowList(cfg.FollowListPath); err != nil { + return err + } + reconciler, err := ingest.NewFollowReconciler(ingest.FollowReconcilerOptions{ + Admin: admin, + Path: cfg.FollowListPath, + Interval: cfg.FollowListInterval, + Logger: logger, + }) + if err != nil { + return err + } + admin.SetFollowReconciler(reconciler) + go reconciler.Run(ctx) + } + // The vote-aggregate XRPC (the AppView's side-channel read). votesXRPC, err := votes.NewXRPC(votes.XRPCOptions{DB: database, Logger: logger}) if err != nil { diff --git a/communities.yaml b/communities.yaml new file mode 100644 index 0000000..72c1197 --- /dev/null +++ b/communities.yaml @@ -0,0 +1,28 @@ +# Declarative follow list: every Lemmy community this Tidepool instance +# bridges. Deploys point FOLLOW_LIST_PATH at this file; the reconciler +# converges subscriptions to it on startup and every FOLLOW_LIST_INTERVAL +# (default 15m), and POST /admin/communities/reconcile forces a pass. +# +# THIS FILE IS AUTHORITATIVE while FOLLOW_LIST_PATH is set: +# - adding an entry subscribes it (profile bridged, Follow sent) +# - removing an entry unfollows it (Undo{Follow}; already-bridged records +# are kept — content just stops flowing) +# - manual POST /admin/communities subscriptions NOT listed here are +# unfollowed at the next sweep +# Community-side consent always wins: a community advertising #nobridge is +# refused no matter what this file says. +# +# Inclusion criteria (evaluate requests against these, not vibes): +# - active: posts within the last 30 days +# - established moderation (named mod team, posted rules) +# - not NSFW-primary +# Request additions or removals via GitHub issue; each entry should carry a +# comment linking its rationale. +# +# Entry format: "!name@host" or the community's AP URL ("https://host/c/name"). +# NOTE: entries MUST be quoted — a bare !name@host is a YAML tag and will +# fail to parse. +communities: + - "!comicstrips@lemmy.world" # seed set, 2026-07: small, active, cross-checks lemmy.world + - "!selfhosted@lemmy.world" # seed set, 2026-07 + - "!linux@lemmy.ml" # seed set, 2026-07: second instance, exercises cross-instance federation diff --git a/go.mod b/go.mod index 6898229..5984223 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module tidepool go 1.25.7 require ( + github.com/JohannesKaufmann/html-to-markdown/v2 v2.5.2 github.com/bluesky-social/indigo v0.0.0-20260202181658-ea3d39eec464 github.com/go-chi/chi/v5 v5.3.1 github.com/gorilla/websocket v1.5.3 @@ -13,14 +14,15 @@ require ( github.com/lib/pq v1.12.3 github.com/multiformats/go-multihash v0.2.3 github.com/pressly/goose/v3 v3.27.2 + github.com/rivo/uniseg v0.4.7 github.com/stretchr/testify v1.11.1 golang.org/x/sync v0.21.0 golang.org/x/time v0.15.0 + gopkg.in/yaml.v3 v3.0.1 ) require ( github.com/JohannesKaufmann/dom v0.3.1 // indirect - github.com/JohannesKaufmann/html-to-markdown/v2 v2.5.2 // indirect github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect @@ -71,7 +73,6 @@ require ( github.com/prometheus/client_model v0.5.0 // indirect github.com/prometheus/common v0.45.0 // indirect github.com/prometheus/procfs v0.20.1 // indirect - github.com/rivo/uniseg v0.4.7 // indirect github.com/sethvargo/go-retry v0.3.0 // indirect github.com/spaolacci/murmur3 v1.1.0 // indirect github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e // indirect @@ -90,7 +91,6 @@ require ( golang.org/x/sys v0.45.0 // indirect golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect google.golang.org/protobuf v1.36.11 // indirect - gopkg.in/yaml.v3 v3.0.1 // indirect gorm.io/gorm v1.25.9 // indirect lukechampine.com/blake3 v1.2.1 // indirect ) diff --git a/go.sum b/go.sum index c51b1c9..2643f93 100644 --- a/go.sum +++ b/go.sum @@ -51,8 +51,6 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8= github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= -github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY= -github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ= @@ -232,6 +230,10 @@ github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFR github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/sebdah/goldie/v2 v2.8.0 h1:dZb9wR8q5++oplmEiJT+U/5KyotVD+HNGCAc5gNr8rc= +github.com/sebdah/goldie/v2 v2.8.0/go.mod h1:oZ9fp0+se1eapSRjfYbsV/0Hqhbuu3bJVvKI/NNtssI= +github.com/sergi/go-diff v1.4.0 h1:n/SP9D5ad1fORl+llWyN+D6qoUETXNZARKjyY2/KVCw= +github.com/sergi/go-diff v1.4.0/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4= github.com/sethvargo/go-retry v0.3.0 h1:EEt31A35QhrcRZtrYFDTBg91cqZVnFL2navjDrah2SE= github.com/sethvargo/go-retry v0.3.0/go.mod h1:mNX17F0C/HguQMyMyJxcnU471gOZGxCLyYaFyAZraas= github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= @@ -260,6 +262,8 @@ github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e/go.mod h1 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= +github.com/yuin/goldmark v1.8.2 h1:kEGpgqJXdgbkhcOgBxkC0X0PmoPG1ZyoZ117rDVp4zE= +github.com/yuin/goldmark v1.8.2/go.mod h1:ip/1k0VRfGynBgxOz0yCqHrbZXhcjxyuS66Brc7iBKg= gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b h1:CzigHMRySiX3drau9C6Q5CAbNIApmLdat5jPMqChvDA= gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b/go.mod h1:/y/V339mxv2sZmYYR64O07VuCpdNZqCTwO8ZcouTMI8= gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02 h1:qwDnMxjkyLmAFgcfgTnfJrmYKWhHnci3GjDqcZp1M3Q= diff --git a/internal/config/config.go b/internal/config/config.go index 00dc120..b1ba1e0 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -175,6 +175,24 @@ type Config struct { // (STATS_REFRESH_BATCH, default 200, must be positive). Commits are // globally serialized, so the batch keeps one sweep from flooding the lock. StatsRefreshBatch int + // FollowListPath optionally points at the declarative follow list — a + // repo-committed YAML file naming every community the bridge should + // follow (FOLLOW_LIST_PATH). When set, a reconciler converges the + // communities table to the file on startup and every + // FollowListInterval: entries missing from the table are subscribed, + // subscriptions missing from the file are unfollowed (records are + // kept; content just stops flowing). Empty = feature off, the /admin + // API is the only subscription control (the BRIDGE_SERVICE_DID + // pattern). NOTE: when active, the file is authoritative — manual + // /admin subscriptions not in the file are unfollowed at the next + // sweep. + FollowListPath string + // FollowListInterval is the reconciler's sweep cadence + // (FOLLOW_LIST_INTERVAL, a Go duration, default 15m, must be + // positive). Periodic re-sweeps self-heal subscribes that failed + // because the remote was down at startup; pending→accepted retries are + // the follow retrier's job, not the reconciler's. + FollowListInterval time.Duration } // Load reads configuration from the environment. logger must not be nil; @@ -415,6 +433,16 @@ func Load(logger *slog.Logger) (*Config, error) { return nil, err } + // Declarative follow list: optional in every environment (unset = the + // /admin API is the only subscription control). The path's existence and + // contents are validated at startup by the caller, not here — config + // only carries the knob. + cfg.FollowListPath = os.Getenv("FOLLOW_LIST_PATH") + cfg.FollowListInterval, err = durationVar(logger, "FOLLOW_LIST_INTERVAL", 15*time.Minute) + if err != nil { + return nil, err + } + defaultUserAgent := fmt.Sprintf("tidepool/0.1 (+https://%s)", cfg.BridgeHostname) cfg.UserAgent = os.Getenv("USER_AGENT") if cfg.UserAgent == "" { diff --git a/internal/ingest/follow.go b/internal/ingest/follow.go index 2f2fd3e..4bca716 100644 --- a/internal/ingest/follow.go +++ b/internal/ingest/follow.go @@ -5,6 +5,7 @@ import ( "crypto/rand" "encoding/hex" "encoding/json" + stderrors "errors" "expvar" "fmt" "io" @@ -47,10 +48,11 @@ type AdminOptions struct { // Admin is the operator API driving the community subscription lifecycle: // -// POST /admin/communities {"community":"!tech@lemmy.world"} -// DELETE /admin/communities {"community":"!tech@lemmy.world"} +// POST /admin/communities {"community":"!tech@lemmy.world"} +// DELETE /admin/communities {"community":"!tech@lemmy.world"} // GET /admin/communities -// POST /admin/communities/backfill {"community":"!tech@lemmy.world"} +// POST /admin/communities/backfill {"community":"!tech@lemmy.world"} +// POST /admin/communities/reconcile (follow list configured only) // // All endpoints require "Authorization: Bearer $ADMIN_TOKEN". type Admin struct { @@ -61,8 +63,17 @@ type Admin struct { service *ap.ServiceActor backfill Backfiller logger *slog.Logger + // reconciler serves POST /admin/communities/reconcile; nil (the + // endpoint answers 501) unless a follow list is configured. Set once + // during startup via SetFollowReconciler, before the server listens. + reconciler *FollowReconciler } +// SetFollowReconciler wires the optional follow-list reconciler in after +// construction (the reconciler itself needs the Admin's subscribe cores, so +// it is necessarily built second). +func (a *Admin) SetFollowReconciler(r *FollowReconciler) { a.reconciler = r } + // NewAdmin validates options and builds the Admin API. func NewAdmin(opts AdminOptions) (*Admin, error) { if opts.Token == "" { @@ -108,6 +119,7 @@ func (a *Admin) Routes(r chi.Router) { r.Delete("/communities", a.handleUnsubscribe) r.Get("/communities", a.handleList) r.Post("/communities/backfill", a.handleBackfill) + r.Post("/communities/reconcile", a.handleReconcile) r.Method(http.MethodGet, "/metrics", http.HandlerFunc(scopedMetrics)) }) } @@ -158,23 +170,69 @@ func communityJSON(c *store.Community) communityResponse { return resp } +// adminError carries the admin API's HTTP mapping for a failed +// subscribe/unsubscribe step, so the HTTP handlers and the follow-list +// reconciler can share one transport-agnostic core. The core logs each +// failure at the site that understands it; handlers only translate. +type adminError struct { + status int // HTTP status the admin API reports + public string // operator-facing message (response body) + err error // underlying cause; nil for pure policy refusals +} + +func (e *adminError) Error() string { + if e.err != nil { + return e.public + ": " + e.err.Error() + } + return e.public +} + +func (e *adminError) Unwrap() error { return e.err } + +// writeAdminError translates a subscribe/unsubscribe core failure into the +// HTTP response. Non-adminError errors cannot happen today (the cores wrap +// everything), but map to 500 rather than panicking on a future oversight. +func writeAdminError(w http.ResponseWriter, err error) { + var ae *adminError + if stderrors.As(err, &ae) { + http.Error(w, ae.public, ae.status) + return + } + http.Error(w, "internal error", http.StatusInternalServerError) +} + // handleSubscribe resolves, bridges, and follows a community: // WebFinger → fetch Group → materialize community profile → signed Follow // from the service actor → follow_state pending (Accept arrives via the // inbox and flips it to accepted, which triggers backfill). func (a *Admin) handleSubscribe(w http.ResponseWriter, r *http.Request) { - ctx := r.Context() var req communityRequest if err := decodeJSONBody(r, &req); err != nil || strings.TrimSpace(req.Community) == "" { http.Error(w, `body must be {"community":"!name@instance"}`, http.StatusBadRequest) return } - groupIRI, err := a.resolveCommunity(ctx, req.Community) + community, err := a.subscribe(r.Context(), req.Community) if err != nil { - a.writeResolveError(w, req.Community, err) + writeAdminError(w, err) return } + if community.FollowState == store.FollowStateAccepted { + // Already subscribed; idempotent success. + writeJSON(w, http.StatusOK, communityJSON(community)) + return + } + writeJSON(w, http.StatusAccepted, communityJSON(community)) +} + +// subscribe is the transport-agnostic core of handleSubscribe, shared with +// the follow-list reconciler. On success the returned community is either +// already accepted (idempotent no-op) or freshly pending. +func (a *Admin) subscribe(ctx context.Context, ref string) (*store.Community, error) { + groupIRI, err := a.resolveCommunity(ctx, ref) + if err != nil { + return nil, a.resolveError(ref, err) + } // Bridge the community first: DID minted, community.profile committed — // content referencing it can land the moment announces start. @@ -182,52 +240,50 @@ func (a *Admin) handleSubscribe(w http.ResponseWriter, r *http.Request) { if err != nil { if materialize.IsSkip(err) { a.logger.Warn("subscribe refused", "community", groupIRI, "reason", err.Error()) - http.Error(w, "community cannot be bridged: "+err.Error(), http.StatusUnprocessableEntity) - return + return nil, &adminError{status: http.StatusUnprocessableEntity, + public: "community cannot be bridged: " + err.Error(), err: err} } a.logger.Error("subscribe: materialize community", "community", groupIRI, "error", err) - http.Error(w, "failed to bridge community", http.StatusBadGateway) - return + return nil, &adminError{status: http.StatusBadGateway, + public: "failed to bridge community", err: err} } if community.FollowState == store.FollowStateAccepted { - // Already subscribed; idempotent success. - writeJSON(w, http.StatusOK, communityJSON(community)) - return + return community, nil } group, err := a.client.FetchActor(ctx, groupIRI) if err != nil { a.logger.Error("subscribe: fetch group", "community", groupIRI, "error", err) - http.Error(w, "failed to fetch community actor", http.StatusBadGateway) - return + return nil, &adminError{status: http.StatusBadGateway, + public: "failed to fetch community actor", err: err} } inbox := group.SharedInboxOrInbox() if inbox == "" { - http.Error(w, "community actor advertises no inbox", http.StatusUnprocessableEntity) - return + return nil, &adminError{status: http.StatusUnprocessableEntity, + public: "community actor advertises no inbox"} } follow, err := a.buildFollow(groupIRI) if err != nil { a.logger.Error("subscribe: build follow", "community", groupIRI, "error", err) - http.Error(w, "failed to build Follow", http.StatusInternalServerError) - return + return nil, &adminError{status: http.StatusInternalServerError, + public: "failed to build Follow", err: err} } if err := a.client.SendActivity(ctx, inbox, follow); err != nil { a.logger.Error("subscribe: deliver follow", "community", groupIRI, "error", err) - http.Error(w, "failed to deliver Follow", http.StatusBadGateway) - return + return nil, &adminError{status: http.StatusBadGateway, + public: "failed to deliver Follow", err: err} } if err := a.communities.SetFollowState(ctx, groupIRI, store.FollowStatePending); err != nil { a.logger.Error("subscribe: record pending follow", "community", groupIRI, "error", err) - http.Error(w, "failed to record follow state", http.StatusInternalServerError) - return + return nil, &adminError{status: http.StatusInternalServerError, + public: "failed to record follow state", err: err} } a.logger.Info("follow sent; awaiting accept", "community", groupIRI, "did", community.DID) community.FollowState = store.FollowStatePending - writeJSON(w, http.StatusAccepted, communityJSON(community)) + return community, nil } // handleUnsubscribe sends Undo{Follow} and clears the follow state. The @@ -246,15 +302,29 @@ func (a *Admin) handleUnsubscribe(w http.ResponseWriter, r *http.Request) { a.writeResolveError(w, req.Community, err) return } + community, err := a.unsubscribeByGroupIRI(ctx, groupIRI) + if err != nil { + writeAdminError(w, err) + return + } + writeJSON(w, http.StatusOK, communityJSON(community)) +} + +// unsubscribeByGroupIRI is the transport-agnostic core of handleUnsubscribe, +// shared with the follow-list reconciler (which diffs by canonical group IRI +// and so never needs the resolve step). Undo{Follow} delivery is best-effort; +// clearing local follow state is the authoritative effect. Materialized +// records are never deleted — content just stops flowing. +func (a *Admin) unsubscribeByGroupIRI(ctx context.Context, groupIRI string) (*store.Community, error) { community, err := a.communities.GetByAPGroupID(ctx, groupIRI) if errors.IsNotFound(err) { - http.Error(w, "community is not bridged", http.StatusNotFound) - return + return nil, &adminError{status: http.StatusNotFound, + public: "community is not bridged", err: err} } if err != nil { a.logger.Error("unsubscribe: look up community", "community", groupIRI, "error", err) - http.Error(w, "internal error", http.StatusInternalServerError) - return + return nil, &adminError{status: http.StatusInternalServerError, + public: "internal error", err: err} } // Best-effort remote notification. @@ -279,12 +349,12 @@ func (a *Admin) handleUnsubscribe(w http.ResponseWriter, r *http.Request) { if err := a.communities.SetFollowState(ctx, groupIRI, store.FollowStateNone); err != nil { a.logger.Error("unsubscribe: clear follow state", "community", groupIRI, "error", err) - http.Error(w, "failed to clear follow state", http.StatusInternalServerError) - return + return nil, &adminError{status: http.StatusInternalServerError, + public: "failed to clear follow state", err: err} } a.logger.Info("community unfollowed", "community", groupIRI) community.FollowState = store.FollowStateNone - writeJSON(w, http.StatusOK, communityJSON(community)) + return community, nil } // handleList reports every community in accepted or pending state. @@ -336,6 +406,26 @@ func (a *Admin) handleBackfill(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusAccepted, communityJSON(community)) } +// handleReconcile runs one synchronous follow-list sweep on demand, so an +// operator can converge right after editing the file instead of waiting for +// the next tick. 501 when no follow list is configured (the handleBackfill +// nil-dependency pattern). +func (a *Admin) handleReconcile(w http.ResponseWriter, r *http.Request) { + if a.reconciler == nil { + http.Error(w, "follow list reconciliation is not configured", http.StatusNotImplemented) + return + } + result, err := a.reconciler.Sweep(r.Context()) + if err != nil { + // Parse/list failures abort the pass with no state changes; the + // admin API is operator-facing, so the real reason (file path, + // entry number) goes straight back. + http.Error(w, "reconcile failed: "+err.Error(), http.StatusInternalServerError) + return + } + writeJSON(w, http.StatusOK, result) +} + // resolveCommunity turns the request's community reference into the Group's // AP id: URLs pass through, handles go through WebFinger. func (a *Admin) resolveCommunity(ctx context.Context, ref string) (string, error) { @@ -346,18 +436,26 @@ func (a *Admin) resolveCommunity(ctx context.Context, ref string) (string, error return a.client.ResolveHandle(ctx, ref) } -func (a *Admin) writeResolveError(w http.ResponseWriter, ref string, err error) { +// resolveError maps a resolveCommunity failure onto the admin API's HTTP +// vocabulary (shared by the HTTP handlers and the reconciler-driven +// subscribe core). +func (a *Admin) resolveError(ref string, err error) *adminError { switch { case errors.IsValidation(err): - http.Error(w, err.Error(), http.StatusBadRequest) + return &adminError{status: http.StatusBadRequest, public: err.Error(), err: err} case errors.IsNotFound(err): - http.Error(w, "community not found: "+ref, http.StatusNotFound) + return &adminError{status: http.StatusNotFound, public: "community not found: " + ref, err: err} default: a.logger.Error("resolve community", "community", ref, "error", err) - http.Error(w, "failed to resolve community", http.StatusBadGateway) + return &adminError{status: http.StatusBadGateway, public: "failed to resolve community", err: err} } } +func (a *Admin) writeResolveError(w http.ResponseWriter, ref string, err error) { + e := a.resolveError(ref, err) + http.Error(w, e.public, e.status) +} + // buildFollow constructs the signed Follow activity (delivery signing // happens in the client; this is the payload Lemmy validates). func (a *Admin) buildFollow(groupIRI string) (*ap.Object, error) { diff --git a/internal/ingest/reconcile.go b/internal/ingest/reconcile.go new file mode 100644 index 0000000..036393d --- /dev/null +++ b/internal/ingest/reconcile.go @@ -0,0 +1,313 @@ +package ingest + +import ( + "bytes" + "context" + stderrors "errors" + "fmt" + "io" + "log/slog" + "net/url" + "os" + "strings" + "sync" + "time" + + "gopkg.in/yaml.v3" + + "tidepool/internal/errors" + "tidepool/internal/store" +) + +// defaultFollowListInterval is the reconciler's sweep cadence when the +// options leave it zero (config's FOLLOW_LIST_INTERVAL default mirrors it). +const defaultFollowListInterval = 15 * time.Minute + +// followListDoc is the FOLLOW_LIST_PATH YAML document: +// +// communities: +// - "!technology@lemmy.world" # entries MUST be quoted (bare ! is a YAML tag) +// - "https://lemmy.ml/c/linux" +// +// Communities is a pointer so a present-but-null key (`communities:`) and a +// missing key (`{}`) are distinguishable from an explicit empty sequence +// (`communities: []`): only the last may unfollow everything. +type followListDoc struct { + Communities *[]string `yaml:"communities"` +} + +// ParseFollowList reads and validates the declarative follow list: every +// entry must be a "!name@host" handle or an http(s) Group URL, deduplicated +// case-insensitively (first spelling wins). Validation is strict because a +// typo'd entry would otherwise mint a garbage did:plc via EnsureCommunity — +// PLC registrations are forever. Unfollowing everything requires an +// explicit `communities: []`: an empty file, a missing communities key, or +// a present-but-null one (`communities:` with nothing under it) are all +// errors, so a truncated or accidentally emptied file cannot read as +// "unfollow all". +func ParseFollowList(path string) ([]string, error) { + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("follow list %s: %w", path, err) + } + dec := yaml.NewDecoder(bytes.NewReader(raw)) + dec.KnownFields(true) // catch "communites:" and friends + var doc followListDoc + if err := dec.Decode(&doc); err != nil { + if stderrors.Is(err, io.EOF) { + return nil, fmt.Errorf("follow list %s: file is empty; write \"communities: []\" to explicitly unfollow everything", path) + } + return nil, fmt.Errorf("follow list %s: %w", path, err) + } + if doc.Communities == nil { + return nil, fmt.Errorf("follow list %s: communities key is missing or null; write \"communities: []\" to explicitly unfollow everything", path) + } + + seen := make(map[string]struct{}, len(*doc.Communities)) + entries := make([]string, 0, len(*doc.Communities)) + for i, raw := range *doc.Communities { + entry, err := validateFollowEntry(raw) + if err != nil { + return nil, fmt.Errorf("follow list %s: entry %d: %w", path, i+1, err) + } + key := strings.ToLower(entry) + if _, dup := seen[key]; dup { + continue + } + seen[key] = struct{}{} + entries = append(entries, entry) + } + return entries, nil +} + +// validateFollowEntry normalizes one follow-list entry and rejects anything +// that is neither a "!name@host" handle nor an http(s) URL. +func validateFollowEntry(raw string) (string, error) { + entry := strings.TrimSpace(raw) + if entry == "" { + return "", fmt.Errorf("entry must not be empty") + } + if strings.ContainsAny(entry, " \t") { + return "", fmt.Errorf("entry %q must not contain whitespace", entry) + } + if strings.HasPrefix(entry, "https://") || strings.HasPrefix(entry, "http://") { + u, err := url.Parse(entry) + if err != nil || u.Host == "" { + return "", fmt.Errorf("entry %q is not a valid group URL", entry) + } + return entry, nil + } + if !strings.HasPrefix(entry, "!") { + return "", fmt.Errorf("entry %q must be \"!name@host\" or an http(s) group URL (handles need the leading '!', and YAML requires quoting it)", entry) + } + if name, host, ok := splitHandle(entry); !ok || name == "" || host == "" { + return "", fmt.Errorf("entry %q must be \"!name@host\" with exactly one '@'", entry) + } + return entry, nil +} + +// splitHandle splits "!name@host" into (name, host). +func splitHandle(entry string) (name, host string, ok bool) { + rest := strings.TrimPrefix(entry, "!") + name, host, ok = strings.Cut(rest, "@") + if strings.Contains(host, "@") { + return "", "", false + } + return name, host, ok +} + +// entryMatchesCommunity reports whether a follow-list entry names an +// already-tracked community WITHOUT any network resolution: URL entries +// compare against the canonical AP group id, handle entries against the +// stored (preferred_username, instance) pair. Keeping this offline is the +// reconciler's mass-unfollow fail-safe — a resolver or remote-instance +// outage can never make a desired community look "removed". +func entryMatchesCommunity(entry string, c *store.Community) bool { + if strings.HasPrefix(entry, "https://") || strings.HasPrefix(entry, "http://") { + return strings.EqualFold(entry, c.APGroupID) + } + name, host, ok := splitHandle(entry) + if !ok { + return false + } + return strings.EqualFold(name, c.PreferredUsername) && strings.EqualFold(host, c.Instance) +} + +// FollowReconcilerOptions configures NewFollowReconciler. Admin and Path are +// required; a zero Interval takes defaultFollowListInterval. +type FollowReconcilerOptions struct { + // Admin supplies the subscribe/unsubscribe cores (and their deps) the + // sweeps drive. + Admin *Admin + // Path is the follow-list YAML (config.FollowListPath). + Path string + // Interval is the sweep cadence (config.FollowListInterval). + Interval time.Duration + Logger *slog.Logger +} + +// FollowReconciler converges the communities table to the declarative +// follow list: file entries missing from the table are subscribed, tracked +// subscriptions missing from the file are unfollowed. The file is +// authoritative — manual /admin subscriptions not in it are unfollowed at +// the next sweep. Materialized records are never deleted. +type FollowReconciler struct { + admin *Admin + path string + interval time.Duration + logger *slog.Logger + // mu serializes sweeps: the ticker and POST /admin/communities/reconcile + // could otherwise both snapshot a community as absent and subscribe it + // twice — duplicate Follows are merely noisy, but the racing + // EnsureCommunity calls can mint twice, and the materializer documents + // that the losing mint leaves an orphaned permanent DID. + mu sync.Mutex +} + +// NewFollowReconciler validates options and builds the reconciler. +func NewFollowReconciler(opts FollowReconcilerOptions) (*FollowReconciler, error) { + if opts.Admin == nil { + return nil, errors.NewValidationError("admin", "must not be nil") + } + if opts.Path == "" { + return nil, errors.NewValidationError("path", "must not be empty") + } + logger := opts.Logger + if logger == nil { + logger = slog.Default() + } + interval := opts.Interval + if interval <= 0 { + interval = defaultFollowListInterval + } + return &FollowReconciler{ + admin: opts.Admin, + path: opts.Path, + interval: interval, + logger: logger, + }, nil +} + +// SweepResult reports what one reconcile pass changed. Subscribed and +// Failed carry follow-list entries; Unsubscribed carries canonical AP group +// ids (the file no longer names those communities, so their entries are +// gone by definition). +type SweepResult struct { + Subscribed []string `json:"subscribed"` + Unsubscribed []string `json:"unsubscribed"` + Failed []string `json:"failed"` +} + +// Run converges once immediately, then on every interval tick until ctx is +// cancelled. Sweep failures are logged, never fatal — the periodic re-sweep +// is the retry. +func (r *FollowReconciler) Run(ctx context.Context) { + r.logger.Info("follow reconciler started", "path", r.path, "interval", r.interval) + if _, err := r.Sweep(ctx); err != nil && ctx.Err() == nil { + r.logger.Error("follow reconciler: startup sweep failed", "error", err) + } + ticker := time.NewTicker(r.interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if _, err := r.Sweep(ctx); err != nil && ctx.Err() == nil { + r.logger.Error("follow reconciler: sweep failed", "error", err) + } + } + } +} + +// Sweep runs one reconcile pass. Exported so tests and the admin trigger +// (POST /admin/communities/reconcile) can drive one synchronously, like +// FollowRetrier.Sweep. A parse or list failure aborts the pass with NO +// state changes (a broken file must never read as "unfollow everything"); +// per-community subscribe/unsubscribe failures are recorded in the result +// and the pass keeps converging the rest. Sweeps are serialized (see mu): a +// concurrent caller blocks, then runs against the converged state (a no-op +// when nothing changed in between). +func (r *FollowReconciler) Sweep(ctx context.Context) (SweepResult, error) { + r.mu.Lock() + defer r.mu.Unlock() + + result := SweepResult{Subscribed: []string{}, Unsubscribed: []string{}, Failed: []string{}} + + entries, err := ParseFollowList(r.path) + if err != nil { + return result, err + } + + // Union of accepted+pending — everything the bridge currently follows + // (the same union the admin list endpoint reports). + var current []*store.Community + for _, state := range []store.FollowState{store.FollowStateAccepted, store.FollowStatePending} { + list, err := r.admin.communities.ListByFollowState(ctx, state) + if err != nil { + return result, fmt.Errorf("follow reconciler: list %s communities: %w", state, err) + } + current = append(current, list...) + } + + if len(entries) == 0 && len(current) > 0 { + r.logger.Warn("follow list is empty: unfollowing every current subscription", + "path", r.path, "current", len(current)) + } + + // Offline diff (see entryMatchesCommunity): which tracked communities + // the file still names, and which entries are not tracked yet. + matched := make([]bool, len(entries)) + var extras []*store.Community + for _, c := range current { + found := false + for i, entry := range entries { + if entryMatchesCommunity(entry, c) { + matched[i] = true + found = true + break + } + } + if !found { + extras = append(extras, c) + } + } + + for _, c := range extras { + if ctx.Err() != nil { + return result, ctx.Err() + } + if _, err := r.admin.unsubscribeByGroupIRI(ctx, c.APGroupID); err != nil { + // unsubscribeByGroupIRI already logged the details. + result.Failed = append(result.Failed, c.APGroupID) + continue + } + result.Unsubscribed = append(result.Unsubscribed, c.APGroupID) + } + + for i, entry := range entries { + if matched[i] { + continue + } + if ctx.Err() != nil { + return result, ctx.Err() + } + if _, err := r.admin.subscribe(ctx, entry); err != nil { + // subscribe already logged the details (including consent + // refusals, which stay refused until the community drops its + // opt-out marker — the sweep keeps going either way). + result.Failed = append(result.Failed, entry) + continue + } + result.Subscribed = append(result.Subscribed, entry) + } + + if len(result.Subscribed) > 0 || len(result.Unsubscribed) > 0 || len(result.Failed) > 0 { + r.logger.Info("follow reconciler: sweep complete", + "desired", len(entries), + "subscribed", len(result.Subscribed), + "unsubscribed", len(result.Unsubscribed), + "failed", len(result.Failed)) + } + return result, nil +} diff --git a/internal/ingest/reconcile_test.go b/internal/ingest/reconcile_test.go new file mode 100644 index 0000000..6178ac5 --- /dev/null +++ b/internal/ingest/reconcile_test.go @@ -0,0 +1,344 @@ +package ingest + +import ( + "context" + "encoding/json" + "net/http" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" + "tidepool/internal/errors" + "tidepool/internal/store" +) + +const gamingGroupID = "https://lemmy.world/c/gaming" + +// writeFollowList writes a follow-list YAML into a temp dir and returns its +// path. +func writeFollowList(t *testing.T, content string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "communities.yaml") + require.NoError(t, os.WriteFile(path, []byte(content), 0o644)) + return path +} + +func (h *harness) newReconciler(path string) *FollowReconciler { + h.t.Helper() + rec, err := NewFollowReconciler(FollowReconcilerOptions{Admin: h.admin, Path: path}) + require.NoError(h.t, err) + return rec +} + +// gamingGroup serves a second lemmy.world community (the technology fixture +// re-badged), reachable by URL so tests with two communities don't collide +// on the single webfinger fixture path. +func (h *harness) gamingGroup(extra map[string]any) *remoteActor { + h.t.Helper() + doc := loadFixture(h.t, "group_lemmy_world.json") + doc["id"] = gamingGroupID + doc["preferredUsername"] = "gaming" + doc["name"] = "Gaming" + for k, v := range extra { + doc[k] = v + } + return h.newRemoteActor(gamingGroupID, doc) +} + +// serveTechnologyWebfinger registers the webfinger fixture that resolves +// !technology@lemmy.world. +func (h *harness) serveTechnologyWebfinger() { + h.t.Helper() + webfinger, err := os.ReadFile(filepath.Join("..", "ap", "testdata", "webfinger_group.json")) + require.NoError(h.t, err) + h.serveJSON("/.well-known/webfinger", webfinger) +} + +// TestReconcileConverges: an empty table converges to the file (one handle +// entry, one URL entry — both spellings subscribe), and a second sweep over +// the converged state is a complete no-op that never touches the network. +func TestReconcileConverges(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + h.technologyGroup() + h.serveTechnologyWebfinger() + h.gamingGroup(nil) + + rec := h.newReconciler(writeFollowList(t, ` +communities: + - "!technology@lemmy.world" + - "https://lemmy.world/c/gaming" +`)) + + result, err := rec.Sweep(ctx) + require.NoError(t, err) + assert.Equal(t, []string{"!technology@lemmy.world", "https://lemmy.world/c/gaming"}, result.Subscribed) + assert.Empty(t, result.Unsubscribed) + assert.Empty(t, result.Failed) + + h.mu.Lock() + follows := len(h.inboxLog) + h.mu.Unlock() + assert.Equal(t, 2, follows, "one Follow per subscribed community") + for _, id := range []string{groupID, gamingGroupID} { + community, err := h.communities.GetByAPGroupID(ctx, id) + require.NoError(t, err) + assert.Equal(t, store.FollowStatePending, community.FollowState, id) + } + + // The second sweep matches both rows offline: no writes, no deliveries, + // and — the mass-unfollow fail-safe — no handle re-resolution. + webfingerHits := h.hitCount("/.well-known/webfinger") + result, err = rec.Sweep(ctx) + require.NoError(t, err) + assert.Empty(t, result.Subscribed) + assert.Empty(t, result.Unsubscribed) + assert.Empty(t, result.Failed) + h.mu.Lock() + assert.Equal(t, follows, len(h.inboxLog), "a converged sweep must deliver nothing") + h.mu.Unlock() + assert.Equal(t, webfingerHits, h.hitCount("/.well-known/webfinger"), + "matched entries must never re-resolve (offline diff)") +} + +// TestReconcileUnsubscribesExtra: a community the file no longer names is +// unfollowed (Undo delivered, state none) while its row, DID, and records +// survive; a new entry in the same sweep is subscribed. +func TestReconcileUnsubscribesExtra(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + h.subscribeTechnology() // accepted, via the admin API + h.gamingGroup(nil) + + before, err := h.communities.GetByAPGroupID(ctx, groupID) + require.NoError(t, err) + require.NotEmpty(t, before.DID) + + rec := h.newReconciler(writeFollowList(t, ` +communities: + - "https://lemmy.world/c/gaming" +`)) + deliveriesBefore := len(h.inboxLog) + + result, err := rec.Sweep(ctx) + require.NoError(t, err) + assert.Equal(t, []string{groupID}, result.Unsubscribed) + assert.Equal(t, []string{gamingGroupID}, result.Subscribed) + assert.Empty(t, result.Failed) + + // Unsubscribes go out before subscribes: Undo{Follow(technology)}, then + // Follow(gaming). + h.mu.Lock() + require.Equal(t, deliveriesBefore+2, len(h.inboxLog)) + undoRaw := h.inboxLog[len(h.inboxLog)-2] + followRaw := h.inboxLog[len(h.inboxLog)-1] + h.mu.Unlock() + undo, err := ap.ParseObject(undoRaw) + require.NoError(t, err) + assert.Equal(t, ap.TypeUndo, undo.Type) + require.NotNil(t, undo.Object) + assert.Equal(t, groupID, undo.Object.Object.ID) + follow, err := ap.ParseObject(followRaw) + require.NoError(t, err) + assert.Equal(t, ap.TypeFollow, follow.Type) + assert.Equal(t, gamingGroupID, follow.Object.ID) + + after, err := h.communities.GetByAPGroupID(ctx, groupID) + require.NoError(t, err) + assert.Equal(t, store.FollowStateNone, after.FollowState) + assert.Equal(t, before.DID, after.DID, "unfollowing must not touch the community's DID") +} + +// TestReconcileEmptyListUnfollowsAll: an explicit `communities: []` is +// honored — every subscription is unfollowed. +func TestReconcileEmptyListUnfollowsAll(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + h.subscribeTechnology() + + rec := h.newReconciler(writeFollowList(t, "communities: []\n")) + result, err := rec.Sweep(ctx) + require.NoError(t, err) + assert.Equal(t, []string{groupID}, result.Unsubscribed) + + community, err := h.communities.GetByAPGroupID(ctx, groupID) + require.NoError(t, err) + assert.Equal(t, store.FollowStateNone, community.FollowState) +} + +// TestReconcileBrokenFileKeepsState: a file that breaks after startup makes +// the sweep fail with NO state changes — a parse error must never read as +// "unfollow everything". +func TestReconcileBrokenFileKeepsState(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + h.subscribeTechnology() + + path := writeFollowList(t, ` +communities: + - "!technology@lemmy.world" +`) + rec := h.newReconciler(path) + + deliveriesBefore := len(h.inboxLog) + for _, broken := range []string{ + "communities:\n - \"technology@lemmy.world\"\n", // missing the leading '!' + "communites:\n - \"!technology@lemmy.world\"\n", // typo'd key (KnownFields) + "", // truncated to empty ≠ empty list + "{}\n", // no communities key ≠ empty list + "communities:\n", // present-but-null key ≠ empty list + } { + require.NoError(t, os.WriteFile(path, []byte(broken), 0o644)) + _, err := rec.Sweep(ctx) + require.Error(t, err) + } + + h.mu.Lock() + assert.Equal(t, deliveriesBefore, len(h.inboxLog), "failed sweeps must deliver nothing") + h.mu.Unlock() + community, err := h.communities.GetByAPGroupID(ctx, groupID) + require.NoError(t, err) + assert.Equal(t, store.FollowStateAccepted, community.FollowState) +} + +// TestReconcileConcurrentSweeps: overlapping sweeps (ticker vs the admin +// trigger) are serialized — the loser runs against the converged state, so +// a community absent in both snapshots is subscribed exactly once (a +// duplicate EnsureCommunity race could orphan a permanent DID). +func TestReconcileConcurrentSweeps(t *testing.T) { + h := newHarness(t) + h.technologyGroup() + h.serveTechnologyWebfinger() + + rec := h.newReconciler(writeFollowList(t, ` +communities: + - "!technology@lemmy.world" +`)) + + const sweeps = 4 + results := make([]SweepResult, sweeps) + var wg sync.WaitGroup + for i := range sweeps { + wg.Add(1) + go func() { + defer wg.Done() + result, err := rec.Sweep(context.Background()) + require.NoError(t, err) + results[i] = result + }() + } + wg.Wait() + + subscribed := 0 + for _, result := range results { + subscribed += len(result.Subscribed) + } + assert.Equal(t, 1, subscribed, "exactly one sweep must win the subscribe") + h.mu.Lock() + assert.Equal(t, 1, len(h.inboxLog), "concurrent sweeps must deliver exactly one Follow") + h.mu.Unlock() +} + +// TestReconcileConsentRefusal: a community advertising #nobridge stays +// refused no matter what the file says; the sweep records the failure and +// keeps going. +func TestReconcileConsentRefusal(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + h.gamingGroup(map[string]any{"summary": "

#nobridge

"}) + + rec := h.newReconciler(writeFollowList(t, ` +communities: + - "https://lemmy.world/c/gaming" +`)) + result, err := rec.Sweep(ctx) + require.NoError(t, err, "a consent refusal is a per-entry failure, not a sweep failure") + assert.Equal(t, []string{gamingGroupID}, result.Failed) + assert.Empty(t, result.Subscribed) + + _, err = h.communities.GetByAPGroupID(ctx, gamingGroupID) + assert.True(t, errors.IsNotFound(err), "a refused community must not be bridged") +} + +// TestAdminReconcileEndpoint: 501 without a configured follow list, a +// synchronous sweep with one, 500 (and no state changes) on a broken file. +func TestAdminReconcileEndpoint(t *testing.T) { + h := newHarness(t) + h.technologyGroup() + h.serveTechnologyWebfinger() + + rec := h.adminRequest(http.MethodPost, "/admin/communities/reconcile", nil) + assert.Equal(t, http.StatusNotImplemented, rec.Code) + + path := writeFollowList(t, ` +communities: + - "!technology@lemmy.world" +`) + h.admin.SetFollowReconciler(h.newReconciler(path)) + + rec = h.adminRequest(http.MethodPost, "/admin/communities/reconcile", nil) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + var result SweepResult + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &result)) + assert.Equal(t, []string{"!technology@lemmy.world"}, result.Subscribed) + + require.NoError(t, os.WriteFile(path, []byte("not: [valid"), 0o644)) + rec = h.adminRequest(http.MethodPost, "/admin/communities/reconcile", nil) + assert.Equal(t, http.StatusInternalServerError, rec.Code) +} + +// TestParseFollowList covers entry validation, dedupe, and the guard rails +// around empty/malformed files. +func TestParseFollowList(t *testing.T) { + parse := func(t *testing.T, content string) ([]string, error) { + t.Helper() + return ParseFollowList(writeFollowList(t, content)) + } + + t.Run("valid entries pass, case-insensitive dupes collapse", func(t *testing.T) { + entries, err := parse(t, ` +communities: + - "!technology@lemmy.world" + - "https://lemmy.ml/c/linux" + - "!Technology@Lemmy.World" +`) + require.NoError(t, err) + assert.Equal(t, []string{"!technology@lemmy.world", "https://lemmy.ml/c/linux"}, entries) + }) + + t.Run("explicit empty list is valid", func(t *testing.T) { + entries, err := parse(t, "communities: []\n") + require.NoError(t, err) + assert.Empty(t, entries) + }) + + t.Run("missing file", func(t *testing.T) { + _, err := ParseFollowList(filepath.Join(t.TempDir(), "absent.yaml")) + require.Error(t, err) + }) + + for name, content := range map[string]string{ + "empty file is not an empty list": "", + "missing communities key": "{}\n", + "null communities key": "communities:\n", + "explicitly null communities key": "communities: null\n", + "handle without leading bang": "communities:\n - \"technology@lemmy.world\"\n", + "handle with two ats": "communities:\n - \"!a@b@c\"\n", + "handle without host": "communities:\n - \"!technology@\"\n", + "URL without host": "communities:\n - \"https://\"\n", + "entry with whitespace": "communities:\n - \"!tech @lemmy.world\"\n", + "blank entry": "communities:\n - \"\"\n", + "unknown key": "communites:\n - \"!technology@lemmy.world\"\n", + "malformed yaml": "communities: [unclosed\n", + } { + t.Run(name, func(t *testing.T) { + _, err := parse(t, content) + require.Error(t, err) + }) + } +}