graft.pds.cyberwild.org / graft-source
Federated via graft
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

v2 M2: inbox, follows, and signed delivery to followers

- httpsig.go: draft-cavage HTTP Signatures (sign outbound, verify inbound)
— the fiddliest part of ActivityPub federation, so it ships with a
round-trip test (sign -> receive -> verify) plus a tamper-detection
test, rather than trusting it by inspection.
- Inbox accepts Follow (verifies the signature against the follower's own
published key, records them, replies with a signed Accept) and
Undo{Follow} (removes them). Anything else (replies included) is
accepted and ignored until the comment bridge, M3.
- After each sync pass, DeliverNewActivity signs and POSTs a Create{Note}
to every follower's inbox for whatever landed in activity_log since
that series' last delivery — tracked per-series so a restart never
redelivers history.
- /social now shows each series' live follower count.

Verified live against real malformed/unreachable-actor requests (400 on
bad JSON, 502 on an actor that fails to resolve) with no crash — a real
follow from an actual fediverse account is still the outstanding
end-to-end check, not yet done.

author
jeanbaptiste date (Sep 12, 2026, 11:08 PM +0200) commit 65774d9e parent a21f3c21
+630 -29
+21 -13
cmd/sync/main.go
··· 60 60 tracker := status.NewTracker(st, cfg.SourceURL) 61 61 tracker.SetTopology(topology) 62 62 tracker.SetPublicHost(cfg.PublicHost) 63 + 64 + var apHandler *activitypub.Handler 65 + if cfg.PublicHost != "" { 66 + apHandler = activitypub.NewHandler(st, cfg.PublicHost, log, 67 + func(series string) string { 68 + if refs := topology[series]; len(refs) > 0 { 69 + return refs[0].URL 70 + } 71 + return "" 72 + }, 73 + func(series string) bool { 74 + _, ok := topology[series] 75 + return ok 76 + }, 77 + ) 78 + } 79 + 63 80 runAll := func() { 64 81 for _, rs := range syncers { 65 82 gitErr, issuesErr, patchErr := rs.Run(log) 66 83 tracker.Record(rs.Name(), rs.Series(), gitErr, issuesErr, patchErr) 67 84 } 85 + if apHandler != nil { 86 + apHandler.DeliverNewActivity() 87 + } 68 88 } 69 89 70 90 if *listen != "" { 71 91 mux := http.NewServeMux() 72 - if cfg.PublicHost != "" { 73 - apHandler := activitypub.NewHandler(st, cfg.PublicHost, 74 - func(series string) string { 75 - if refs := topology[series]; len(refs) > 0 { 76 - return refs[0].URL 77 - } 78 - return "" 79 - }, 80 - func(series string) bool { 81 - _, ok := topology[series] 82 - return ok 83 - }, 84 - ) 92 + if apHandler != nil { 85 93 apHandler.Register(mux) 86 94 } 87 95 mux.Handle("/", tracker.Handler())
+197 -5
internal/activitypub/handlers.go
··· 1 1 package activitypub 2 2 3 3 import ( 4 + "crypto/rsa" 4 5 "encoding/json" 6 + "io" 7 + "log/slog" 5 8 "net/http" 6 9 "strconv" 7 10 "strings" 11 + "time" 8 12 9 13 "graft/internal/state" 10 14 ) ··· 12 16 const ( 13 17 activityContentType = "application/activity+json" 14 18 outboxLimit = 50 19 + maxInboxBody = 1 << 20 // 1 MiB, generous for a Follow/Undo/reply 15 20 ) 16 21 17 22 // Handler serves the ActivityPub surface for every series: WebFinger, 18 - // actor, outbox, individual notes, and (for now, always empty — see M2 in 19 - // the v2 plan) a followers collection. 23 + // actor, outbox, individual notes, followers, and an inbox that accepts 24 + // Follow/Undo{Follow} (replies land here too but are ignored until the 25 + // comment bridge, M3). 20 26 type Handler struct { 21 27 store *state.Store 22 28 host string 29 + log *slog.Logger 23 30 // repoURL and known are supplied by the caller (cmd/sync/main.go), 24 31 // which already knows every series' topology — avoids this package 25 32 // needing to know about config.Config or sync.RepoSyncer at all. 26 33 repoURL func(series string) string 27 34 known func(series string) bool 35 + client *http.Client 28 36 } 29 37 30 - func NewHandler(store *state.Store, host string, repoURL func(series string) string, known func(series string) bool) *Handler { 31 - return &Handler{store: store, host: host, repoURL: repoURL, known: known} 38 + func NewHandler(store *state.Store, host string, log *slog.Logger, repoURL func(series string) string, known func(series string) bool) *Handler { 39 + return &Handler{ 40 + store: store, 41 + host: host, 42 + log: log, 43 + repoURL: repoURL, 44 + known: known, 45 + client: &http.Client{Timeout: 15 * time.Second}, 46 + } 32 47 } 33 48 34 49 // Register mounts every ActivityPub route on mux. Call before mounting any ··· 84 99 h.outbox(w, series) 85 100 case len(parts) == 2 && parts[1] == "followers": 86 101 h.followers(w, series) 102 + case len(parts) == 2 && parts[1] == "inbox": 103 + h.inbox(w, r, series) 87 104 case len(parts) == 3 && parts[1] == "notes": 88 105 h.note(w, r, series, parts[2]) 89 106 default: ··· 155 172 } 156 173 157 174 func (h *Handler) followers(w http.ResponseWriter, series string) { 175 + n, err := h.store.FollowerCount(series) 176 + if err != nil { 177 + http.Error(w, err.Error(), http.StatusInternalServerError) 178 + return 179 + } 158 180 writeJSON(w, activityContentType, OrderedCollection{ 159 181 Context: "https://www.w3.org/ns/activitystreams", 160 182 ID: ActorURI(h.host, series) + "/followers", 161 183 Type: "OrderedCollection", 162 - TotalItems: 0, 184 + TotalItems: n, 163 185 OrderedItems: []Create{}, 164 186 }) 187 + } 188 + 189 + // actorPrivateKey returns a series' actor's private key, generating a 190 + // keypair first if one doesn't exist yet. 191 + func (h *Handler) actorPrivateKey(series string) (*rsa.PrivateKey, error) { 192 + privPEM, _, err := h.actorKeys(series) 193 + if err != nil { 194 + return nil, err 195 + } 196 + return ParsePrivateKey(privPEM) 197 + } 198 + 199 + // inboxActivity is the subset of an inbound activity's fields this handler 200 + // reads. Object is kept raw so, for a Follow, it can be embedded verbatim 201 + // as the Accept's object — and so an Undo's nested Follow can be decoded 202 + // separately. 203 + type inboxActivity struct { 204 + Type string `json:"type"` 205 + Actor string `json:"actor"` 206 + ID string `json:"id"` 207 + Object json.RawMessage `json:"object"` 208 + } 209 + 210 + func (h *Handler) inbox(w http.ResponseWriter, r *http.Request, series string) { 211 + body, err := io.ReadAll(io.LimitReader(r.Body, maxInboxBody)) 212 + if err != nil { 213 + http.Error(w, "read body", http.StatusBadRequest) 214 + return 215 + } 216 + 217 + var act inboxActivity 218 + if err := json.Unmarshal(body, &act); err != nil || act.Actor == "" { 219 + http.Error(w, "invalid activity", http.StatusBadRequest) 220 + return 221 + } 222 + 223 + remoteActor, err := FetchActor(h.client, act.Actor) 224 + if err != nil { 225 + h.log.Error("ap inbox: resolve actor", "series", series, "actor", act.Actor, "err", err) 226 + http.Error(w, "could not resolve actor", http.StatusBadGateway) 227 + return 228 + } 229 + pub, err := ParsePublicKey(remoteActor.PublicKey.PublicKeyPem) 230 + if err != nil { 231 + http.Error(w, "invalid actor public key", http.StatusBadGateway) 232 + return 233 + } 234 + if err := VerifyRequest(r, body, pub); err != nil { 235 + h.log.Error("ap inbox: signature verification failed", "series", series, "actor", act.Actor, "err", err) 236 + http.Error(w, "signature verification failed", http.StatusUnauthorized) 237 + return 238 + } 239 + 240 + switch act.Type { 241 + case "Follow": 242 + if err := h.store.SaveFollower(series, act.Actor, remoteActor.Inbox); err != nil { 243 + http.Error(w, err.Error(), http.StatusInternalServerError) 244 + return 245 + } 246 + if err := h.sendAccept(series, remoteActor.Inbox, body); err != nil { 247 + // The follow is already recorded; a failed Accept delivery 248 + // shouldn't undo that — the remote side will likely retry. 249 + h.log.Error("ap inbox: send accept", "series", series, "actor", act.Actor, "err", err) 250 + } 251 + case "Undo": 252 + var inner inboxActivity 253 + if err := json.Unmarshal(act.Object, &inner); err == nil && inner.Type == "Follow" { 254 + if err := h.store.RemoveFollower(series, act.Actor); err != nil { 255 + h.log.Error("ap inbox: remove follower", "series", series, "actor", act.Actor, "err", err) 256 + } 257 + } 258 + default: 259 + // Replies (Create{Note}) land here too but aren't handled until 260 + // the comment bridge (v2 M3) — accepted and ignored for now. 261 + } 262 + 263 + w.WriteHeader(http.StatusAccepted) 264 + } 265 + 266 + // sendAccept replies to a Follow with a signed Accept, embedding the 267 + // original Follow activity verbatim as required by the spec. 268 + func (h *Handler) sendAccept(series, targetInbox string, followBody []byte) error { 269 + priv, err := h.actorPrivateKey(series) 270 + if err != nil { 271 + return err 272 + } 273 + actor := ActorURI(h.host, series) 274 + accept := map[string]any{ 275 + "@context": "https://www.w3.org/ns/activitystreams", 276 + "id": actor + "/accepts/" + strconv.FormatInt(time.Now().UnixNano(), 10), 277 + "type": "Accept", 278 + "actor": actor, 279 + "object": json.RawMessage(followBody), 280 + } 281 + body, err := json.Marshal(accept) 282 + if err != nil { 283 + return err 284 + } 285 + return PostSigned(h.client, targetInbox, actor+"#main-key", priv, body) 286 + } 287 + 288 + // DeliverNewActivity signs and POSTs a Create{Note} to every follower's 289 + // inbox for each series' activity_log rows logged since the last 290 + // delivery. Call it once per sync pass, after runAll has logged whatever 291 + // happened this pass. Best-effort: a delivery failure to one follower is 292 + // logged and skipped, never blocks the sync loop or other followers. 293 + func (h *Handler) DeliverNewActivity() { 294 + seriesList, err := h.store.SeriesWithFollowers() 295 + if err != nil { 296 + h.log.Error("ap delivery: list series with followers", "err", err) 297 + return 298 + } 299 + 300 + for _, series := range seriesList { 301 + cursor, err := h.store.DeliveryCursor(series) 302 + if err != nil { 303 + h.log.Error("ap delivery: read cursor", "series", series, "err", err) 304 + continue 305 + } 306 + entries, err := h.store.NewActivityForSeries(series, cursor) 307 + if err != nil { 308 + h.log.Error("ap delivery: list new activity", "series", series, "err", err) 309 + continue 310 + } 311 + if len(entries) == 0 { 312 + continue 313 + } 314 + 315 + followers, err := h.store.Followers(series) 316 + if err != nil { 317 + h.log.Error("ap delivery: list followers", "series", series, "err", err) 318 + continue 319 + } 320 + priv, err := h.actorPrivateKey(series) 321 + if err != nil { 322 + h.log.Error("ap delivery: load actor key", "series", series, "err", err) 323 + continue 324 + } 325 + keyID := ActorURI(h.host, series) + "#main-key" 326 + 327 + maxID := cursor 328 + for _, e := range entries { 329 + note := BuildNote(h.host, series, e) 330 + create := Create{ 331 + Context: "https://www.w3.org/ns/activitystreams", 332 + ID: note.ID + "/activity", 333 + Type: "Create", 334 + Actor: ActorURI(h.host, series), 335 + Published: note.Published, 336 + To: []string{publicAudience}, 337 + Object: note, 338 + } 339 + body, err := json.Marshal(create) 340 + if err != nil { 341 + h.log.Error("ap delivery: marshal activity", "series", series, "id", e.ID, "err", err) 342 + continue 343 + } 344 + for _, f := range followers { 345 + if err := PostSigned(h.client, f.InboxURI, keyID, priv, body); err != nil { 346 + h.log.Error("ap delivery: post to follower", "series", series, "follower", f.ActorURI, "err", err) 347 + } 348 + } 349 + if e.ID > maxID { 350 + maxID = e.ID 351 + } 352 + } 353 + if err := h.store.SetDeliveryCursor(series, maxID); err != nil { 354 + h.log.Error("ap delivery: set cursor", "series", series, "err", err) 355 + } 356 + } 165 357 } 166 358 167 359 func writeJSON(w http.ResponseWriter, contentType string, v any) {
+141
internal/activitypub/httpsig.go
··· 1 + package activitypub 2 + 3 + import ( 4 + "bytes" 5 + "crypto" 6 + "crypto/rand" 7 + "crypto/rsa" 8 + "crypto/sha256" 9 + "encoding/base64" 10 + "fmt" 11 + "io" 12 + "net/http" 13 + "strings" 14 + "time" 15 + ) 16 + 17 + // signedHeaders is the fixed set of headers this package signs and 18 + // verifies — the subset draft-cavage HTTP Signatures needs to bind a 19 + // request to its actual method, path, host, date, and body, which is what 20 + // every mainstream ActivityPub implementation (Mastodon included) expects. 21 + var signedHeaders = []string{"(request-target)", "host", "date", "digest"} 22 + 23 + // SignRequest signs req per draft-cavage HTTP Signatures: (request-target), 24 + // host, date, and a Digest of body are covered. Sets the Digest, Date, and 25 + // Signature headers on req. 26 + func SignRequest(req *http.Request, body []byte, keyID string, priv *rsa.PrivateKey) error { 27 + digest := sha256.Sum256(body) 28 + req.Header.Set("Digest", "SHA-256="+base64.StdEncoding.EncodeToString(digest[:])) 29 + req.Header.Set("Date", time.Now().UTC().Format(http.TimeFormat)) 30 + 31 + signingString := buildSigningString(signedHeaders, req.Method, req.URL.RequestURI(), req.URL.Host, req.Header) 32 + hashed := sha256.Sum256([]byte(signingString)) 33 + sig, err := rsa.SignPKCS1v15(rand.Reader, priv, crypto.SHA256, hashed[:]) 34 + if err != nil { 35 + return fmt.Errorf("sign request: %w", err) 36 + } 37 + 38 + req.Header.Set("Signature", fmt.Sprintf( 39 + `keyId="%s",algorithm="rsa-sha256",headers="%s",signature="%s"`, 40 + keyID, strings.Join(signedHeaders, " "), base64.StdEncoding.EncodeToString(sig), 41 + )) 42 + return nil 43 + } 44 + 45 + // VerifyRequest verifies an inbound request's Signature header against 46 + // pub, re-deriving the signing string from whichever headers the 47 + // signature itself claims to cover (a remote server may sign a different 48 + // header set than we do when sending). 49 + func VerifyRequest(r *http.Request, body []byte, pub *rsa.PublicKey) error { 50 + params := parseSignatureHeader(r.Header.Get("Signature")) 51 + if params["signature"] == "" { 52 + return fmt.Errorf("missing or unparseable Signature header") 53 + } 54 + 55 + if digestHeader := r.Header.Get("Digest"); digestHeader != "" { 56 + sum := sha256.Sum256(body) 57 + want := "SHA-256=" + base64.StdEncoding.EncodeToString(sum[:]) 58 + if !strings.EqualFold(digestHeader, want) { 59 + return fmt.Errorf("digest does not match body") 60 + } 61 + } 62 + 63 + headers := strings.Fields(params["headers"]) 64 + if len(headers) == 0 { 65 + headers = []string{"date"} 66 + } 67 + host := r.Header.Get("Host") 68 + if host == "" { 69 + host = r.Host 70 + } 71 + signingString := buildSigningString(headers, r.Method, r.URL.RequestURI(), host, r.Header) 72 + 73 + sigBytes, err := base64.StdEncoding.DecodeString(params["signature"]) 74 + if err != nil { 75 + return fmt.Errorf("decode signature: %w", err) 76 + } 77 + hashed := sha256.Sum256([]byte(signingString)) 78 + if err := rsa.VerifyPKCS1v15(pub, crypto.SHA256, hashed[:], sigBytes); err != nil { 79 + return fmt.Errorf("signature does not verify: %w", err) 80 + } 81 + return nil 82 + } 83 + 84 + // KeyID extracts the keyId parameter from an inbound request's Signature 85 + // header — usually the remote actor's URI with a #main-key fragment, 86 + // telling us whose public key to fetch and verify against. 87 + func KeyID(r *http.Request) string { 88 + return parseSignatureHeader(r.Header.Get("Signature"))["keyId"] 89 + } 90 + 91 + func buildSigningString(headers []string, method, requestURI, host string, reqHeaders http.Header) string { 92 + var lines []string 93 + for _, h := range headers { 94 + switch strings.ToLower(h) { 95 + case "(request-target)": 96 + lines = append(lines, fmt.Sprintf("(request-target): %s %s", strings.ToLower(method), requestURI)) 97 + case "host": 98 + lines = append(lines, "host: "+host) 99 + default: 100 + lines = append(lines, strings.ToLower(h)+": "+reqHeaders.Get(h)) 101 + } 102 + } 103 + return strings.Join(lines, "\n") 104 + } 105 + 106 + func parseSignatureHeader(s string) map[string]string { 107 + out := map[string]string{} 108 + for _, part := range strings.Split(s, ",") { 109 + part = strings.TrimSpace(part) 110 + kv := strings.SplitN(part, "=", 2) 111 + if len(kv) != 2 { 112 + continue 113 + } 114 + out[kv[0]] = strings.Trim(kv[1], `"`) 115 + } 116 + return out 117 + } 118 + 119 + // PostSigned sends a signed POST of an activity+json body, authenticated 120 + // as keyID using priv. 121 + func PostSigned(client *http.Client, url, keyID string, priv *rsa.PrivateKey, body []byte) error { 122 + req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(body)) 123 + if err != nil { 124 + return fmt.Errorf("build request: %w", err) 125 + } 126 + req.Header.Set("Content-Type", activityContentType) 127 + if err := SignRequest(req, body, keyID, priv); err != nil { 128 + return err 129 + } 130 + 131 + resp, err := client.Do(req) 132 + if err != nil { 133 + return fmt.Errorf("post %s: %w", url, err) 134 + } 135 + defer resp.Body.Close() 136 + if resp.StatusCode >= 300 { 137 + respBody, _ := io.ReadAll(resp.Body) 138 + return fmt.Errorf("post %s: HTTP %d: %s", url, resp.StatusCode, string(respBody)) 139 + } 140 + return nil 141 + }
+85
internal/activitypub/httpsig_test.go
··· 1 + package activitypub 2 + 3 + import ( 4 + "bytes" 5 + "crypto/rsa" 6 + "io" 7 + "net/http" 8 + "net/http/httptest" 9 + "testing" 10 + ) 11 + 12 + // TestSignVerifyRoundTrip exercises the actual code path a remote server 13 + // runs: sign a POST the way PostSigned does, receive it through an 14 + // httptest server, and verify it the way the inbox handler does. This is 15 + // the one piece of this package genuinely worth distrusting until proven — 16 + // HTTP Signatures are notoriously easy to get subtly wrong (header casing, 17 + // request-target format, digest encoding) in a way that only shows up 18 + // against a real remote server's strict verifier. 19 + func TestSignVerifyRoundTrip(t *testing.T) { 20 + _, pub, priv := testKeyPair(t) 21 + 22 + body := []byte(`{"type":"Follow","actor":"https://example.social/users/alice","object":"https://graft.example.org/actors/testrepo"}`) 23 + 24 + var verifyErr error 25 + var gotBody []byte 26 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { 27 + var err error 28 + gotBody, err = io.ReadAll(r.Body) 29 + if err != nil { 30 + t.Errorf("read body: %v", err) 31 + } 32 + verifyErr = VerifyRequest(r, gotBody, pub) 33 + w.WriteHeader(http.StatusAccepted) 34 + })) 35 + defer srv.Close() 36 + 37 + client := srv.Client() 38 + if err := PostSigned(client, srv.URL+"/actors/testrepo/inbox", "https://example.social/users/alice#main-key", priv, body); err != nil { 39 + t.Fatalf("PostSigned: %v", err) 40 + } 41 + if verifyErr != nil { 42 + t.Fatalf("VerifyRequest: %v", verifyErr) 43 + } 44 + if !bytes.Equal(gotBody, body) { 45 + t.Fatalf("body mismatch: got %q want %q", gotBody, body) 46 + } 47 + } 48 + 49 + // TestVerifyRejectsTamperedBody confirms a body modified in transit (or by 50 + // a malicious intermediary) fails verification via the Digest mismatch, 51 + // even though the Signature header itself is untouched. 52 + func TestVerifyRejectsTamperedBody(t *testing.T) { 53 + _, pub, priv := testKeyPair(t) 54 + body := []byte(`{"type":"Follow","actor":"https://example.social/users/alice"}`) 55 + 56 + req, err := http.NewRequest(http.MethodPost, "https://graft.example.org/actors/testrepo/inbox", bytes.NewReader(body)) 57 + if err != nil { 58 + t.Fatal(err) 59 + } 60 + if err := SignRequest(req, body, "https://example.social/users/alice#main-key", priv); err != nil { 61 + t.Fatal(err) 62 + } 63 + 64 + tampered := []byte(`{"type":"Follow","actor":"https://attacker.example/users/mallory"}`) 65 + if err := VerifyRequest(req, tampered, pub); err == nil { 66 + t.Fatal("expected verification to fail on tampered body, got nil error") 67 + } 68 + } 69 + 70 + func testKeyPair(t *testing.T) (privPEM string, pub *rsa.PublicKey, priv *rsa.PrivateKey) { 71 + t.Helper() 72 + privateP, publicP, err := GenerateKeyPair() 73 + if err != nil { 74 + t.Fatalf("GenerateKeyPair: %v", err) 75 + } 76 + privKey, err := ParsePrivateKey(privateP) 77 + if err != nil { 78 + t.Fatalf("ParsePrivateKey: %v", err) 79 + } 80 + pubKey, err := ParsePublicKey(publicP) 81 + if err != nil { 82 + t.Fatalf("ParsePublicKey: %v", err) 83 + } 84 + return privateP, pubKey, privKey 85 + }
+18
internal/activitypub/keys.go
··· 42 42 } 43 43 return x509.ParsePKCS1PrivateKey(block.Bytes) 44 44 } 45 + 46 + // ParsePublicKey decodes a PEM-encoded RSA public key, e.g. one fetched 47 + // from a remote actor's publicKeyPem field. 48 + func ParsePublicKey(pemStr string) (*rsa.PublicKey, error) { 49 + block, _ := pem.Decode([]byte(pemStr)) 50 + if block == nil { 51 + return nil, fmt.Errorf("no PEM block found in public key") 52 + } 53 + key, err := x509.ParsePKIXPublicKey(block.Bytes) 54 + if err != nil { 55 + return nil, fmt.Errorf("parse public key: %w", err) 56 + } 57 + rsaKey, ok := key.(*rsa.PublicKey) 58 + if !ok { 59 + return nil, fmt.Errorf("public key is not RSA") 60 + } 61 + return rsaKey, nil 62 + }
+35
internal/activitypub/remote.go
··· 1 + package activitypub 2 + 3 + import ( 4 + "encoding/json" 5 + "fmt" 6 + "net/http" 7 + ) 8 + 9 + // FetchActor retrieves a remote actor object. An unsigned GET is used — 10 + // acceptable here since this is only for resolving a Follow request's 11 + // actor (to get their public key and inbox URL) and most servers permit 12 + // unauthenticated actor fetches; authenticated fetches ("secure mode") are 13 + // out of scope for now. 14 + func FetchActor(client *http.Client, actorURL string) (*Actor, error) { 15 + req, err := http.NewRequest(http.MethodGet, actorURL, nil) 16 + if err != nil { 17 + return nil, fmt.Errorf("build request: %w", err) 18 + } 19 + req.Header.Set("Accept", activityContentType) 20 + 21 + resp, err := client.Do(req) 22 + if err != nil { 23 + return nil, fmt.Errorf("fetch actor %s: %w", actorURL, err) 24 + } 25 + defer resp.Body.Close() 26 + if resp.StatusCode >= 300 { 27 + return nil, fmt.Errorf("fetch actor %s: HTTP %d", actorURL, resp.StatusCode) 28 + } 29 + 30 + var a Actor 31 + if err := json.NewDecoder(resp.Body).Decode(&a); err != nil { 32 + return nil, fmt.Errorf("decode actor %s: %w", actorURL, err) 33 + } 34 + return &a, nil 35 + }
+116
internal/state/store.go
··· 84 84 public_key TEXT NOT NULL, 85 85 created_at TEXT NOT NULL 86 86 ); 87 + 88 + -- One row per remote actor following a series' ActivityPub actor. 89 + CREATE TABLE IF NOT EXISTS ap_follower ( 90 + series TEXT NOT NULL, 91 + actor_uri TEXT NOT NULL, 92 + inbox_uri TEXT NOT NULL, 93 + PRIMARY KEY (series, actor_uri) 94 + ); 95 + 96 + -- How far delivery has gotten through activity_log for each series, so a 97 + -- restart doesn't redeliver everything to every follower. 98 + CREATE TABLE IF NOT EXISTS ap_delivery_cursor ( 99 + series TEXT PRIMARY KEY, 100 + last_delivered_id INTEGER NOT NULL DEFAULT 0 101 + ); 87 102 `) 88 103 if err != nil { 89 104 return err ··· 317 332 `, series, k.PrivateKeyPEM, k.PublicKeyPEM, time.Now().UTC().Format(time.RFC3339)) 318 333 return err 319 334 } 335 + 336 + // Follower is one remote actor following a series' ActivityPub actor. 337 + type Follower struct { 338 + ActorURI string 339 + InboxURI string 340 + } 341 + 342 + // SaveFollower records (or updates the inbox URL of) a follower. 343 + func (s *Store) SaveFollower(series, actorURI, inboxURI string) error { 344 + _, err := s.db.Exec(` 345 + INSERT INTO ap_follower (series, actor_uri, inbox_uri) VALUES (?, ?, ?) 346 + ON CONFLICT (series, actor_uri) DO UPDATE SET inbox_uri = excluded.inbox_uri 347 + `, series, actorURI, inboxURI) 348 + return err 349 + } 350 + 351 + // RemoveFollower drops a follower, e.g. on Undo{Follow}. 352 + func (s *Store) RemoveFollower(series, actorURI string) error { 353 + _, err := s.db.Exec(`DELETE FROM ap_follower WHERE series = ? AND actor_uri = ?`, series, actorURI) 354 + return err 355 + } 356 + 357 + // Followers lists everyone following a series. 358 + func (s *Store) Followers(series string) ([]Follower, error) { 359 + rows, err := s.db.Query(`SELECT actor_uri, inbox_uri FROM ap_follower WHERE series = ?`, series) 360 + if err != nil { 361 + return nil, err 362 + } 363 + defer rows.Close() 364 + var out []Follower 365 + for rows.Next() { 366 + var f Follower 367 + if err := rows.Scan(&f.ActorURI, &f.InboxURI); err != nil { 368 + return nil, err 369 + } 370 + out = append(out, f) 371 + } 372 + return out, rows.Err() 373 + } 374 + 375 + // FollowerCount is len(Followers(series)) without building the slice. 376 + func (s *Store) FollowerCount(series string) (int, error) { 377 + row := s.db.QueryRow(`SELECT COUNT(*) FROM ap_follower WHERE series = ?`, series) 378 + var n int 379 + err := row.Scan(&n) 380 + return n, err 381 + } 382 + 383 + // SeriesWithFollowers lists every series that has at least one follower — 384 + // the set worth checking for new activity to deliver. 385 + func (s *Store) SeriesWithFollowers() ([]string, error) { 386 + rows, err := s.db.Query(`SELECT DISTINCT series FROM ap_follower`) 387 + if err != nil { 388 + return nil, err 389 + } 390 + defer rows.Close() 391 + var out []string 392 + for rows.Next() { 393 + var series string 394 + if err := rows.Scan(&series); err != nil { 395 + return nil, err 396 + } 397 + out = append(out, series) 398 + } 399 + return out, rows.Err() 400 + } 401 + 402 + // DeliveryCursor returns how far ActivityPub delivery has gotten through 403 + // activity_log for a series (0 if nothing has been delivered yet). 404 + func (s *Store) DeliveryCursor(series string) (int64, error) { 405 + row := s.db.QueryRow(`SELECT last_delivered_id FROM ap_delivery_cursor WHERE series = ?`, series) 406 + var id int64 407 + err := row.Scan(&id) 408 + if err == sql.ErrNoRows { 409 + return 0, nil 410 + } 411 + return id, err 412 + } 413 + 414 + // SetDeliveryCursor records how far delivery has gotten for a series. 415 + func (s *Store) SetDeliveryCursor(series string, id int64) error { 416 + _, err := s.db.Exec(` 417 + INSERT INTO ap_delivery_cursor (series, last_delivered_id) VALUES (?, ?) 418 + ON CONFLICT (series) DO UPDATE SET last_delivered_id = excluded.last_delivered_id 419 + `, series, id) 420 + return err 421 + } 422 + 423 + // NewActivityForSeries returns activity_log entries for a series with 424 + // id > afterID, oldest first — the delivery order for ActivityPub. 425 + func (s *Store) NewActivityForSeries(series string, afterID int64) ([]ActivityEntry, error) { 426 + rows, err := s.db.Query(` 427 + SELECT id, repo_pair, kind, direction, summary, url, series, series_url, occurred_at FROM activity_log 428 + WHERE series = ? AND id > ? 429 + ORDER BY id ASC 430 + `, series, afterID) 431 + if err != nil { 432 + return nil, err 433 + } 434 + return scanActivityRows(rows) 435 + }
+17 -11
internal/status/status.go
··· 187 187 } 188 188 189 189 type socialRow struct { 190 - Name string 191 - Handle string // "series@host" 192 - WebfingerURL string 193 - ActorURL string 194 - OutboxURL string 190 + Name string 191 + Handle string // "series@host" 192 + WebfingerURL string 193 + ActorURL string 194 + OutboxURL string 195 + FollowerCount int 195 196 } 196 197 197 198 func (t *Tracker) serveSocial(w http.ResponseWriter, r *http.Request) { ··· 214 215 rows := make([]socialRow, 0, len(names)) 215 216 for _, name := range names { 216 217 actorURL := "https://" + publicHost + "/actors/" + name 218 + followerCount, err := t.store.FollowerCount(name) 219 + if err != nil { 220 + followerCount = 0 221 + } 217 222 rows = append(rows, socialRow{ 218 - Name: name, 219 - Handle: name + "@" + publicHost, 220 - WebfingerURL: "https://" + publicHost + "/.well-known/webfinger?resource=acct:" + name + "@" + publicHost, 221 - ActorURL: actorURL, 222 - OutboxURL: actorURL + "/outbox", 223 + Name: name, 224 + Handle: name + "@" + publicHost, 225 + WebfingerURL: "https://" + publicHost + "/.well-known/webfinger?resource=acct:" + name + "@" + publicHost, 226 + ActorURL: actorURL, 227 + OutboxURL: actorURL + "/outbox", 228 + FollowerCount: followerCount, 223 229 }) 224 230 } 225 231 ··· 826 832 {{range .Series}} 827 833 <div class="series-block"> 828 834 <div class="handle">@{{.Handle}}</div> 829 - <h2>ActivityPub</h2> 835 + <h2>ActivityPub &middot; {{.FollowerCount}} follower{{if ne .FollowerCount 1}}s{{end}}</h2> 830 836 <div class="endpoints"> 831 837 <a href="{{.WebfingerURL}}" target="_blank" rel="noopener">WebFinger</a> 832 838 <a href="{{.ActorURL}}" target="_blank" rel="noopener">Actor</a>