From d6e08e6d80c02cb86edc0f74cf3640d4cc5b9234 Mon Sep 17 00:00:00 2001 From: Christian Beutel <> Date: Thu, 10 Jul 2025 14:46:12 +0200 Subject: [PATCH] makes PostActivity fully async --- db/federation/activity.go | 171 ++++++++++++++++++++------------------ 1 file changed, 90 insertions(+), 81 deletions(-) diff --git a/db/federation/activity.go b/db/federation/activity.go index 9cbd29b6..8874bcd4 100644 --- a/db/federation/activity.go +++ b/db/federation/activity.go @@ -27,95 +27,104 @@ import ( ) func PostActivity(app core.App, actor *core.Record, activity *pub.Activity, recipients []string) error { - encryptionKey := os.Getenv("POCKETBASE_ENCRYPTION_KEY") - if len(encryptionKey) == 0 { - return fmt.Errorf("POCKETBASE_ENCRYPTION_KEY not set") - } - origin := os.Getenv("ORIGIN") - if origin == "" { - return fmt.Errorf("ORIGIN not set") - } - - algs := []httpsig.Algorithm{httpsig.RSA_SHA256} - postHeaders := []string{"(request-target)", "Date", "Digest", "Content-Type", "Host"} - expiresIn := 60 - - body, err := jsonld.WithContext( - jsonld.IRI(pub.ActivityBaseURI), - jsonld.IRI(pub.SecurityContextURI), - ).Marshal(activity) - if err != nil { - return err - } - - decryptedPrivateKey, err := security.Decrypt(actor.GetString("private_key"), encryptionKey) - if err != nil { - return err - } - privateKey, err := x509.ParsePKCS1PrivateKey(decryptedPrivateKey) - if err != nil { - return err - } - pubID := actor.GetString("iri") + "#main-key" - - client := &http.Client{} - var wg sync.WaitGroup - - sem := semaphore.NewWeighted(5) // Limit to 5 concurrent sends - - slices.Sort(recipients) - uniqueRecipients := slices.Compact(recipients) - - for _, v := range uniqueRecipients { - - wg.Add(1) - go func(inbox string) { - defer wg.Done() - - signer, _, err := httpsig.NewSigner(algs, httpsig.DigestSha256, postHeaders, httpsig.Signature, int64(expiresIn)) - if err != nil { - return + go func() { + defer func() { + if r := recover(); r != nil { + app.Logger().Error(fmt.Sprintf("Recovered from panic in PostActivity: %v", r)) } + }() - if err := sem.Acquire(context.Background(), 1); err != nil { - app.Logger().Error(fmt.Sprintf("Semaphore acquire failed: %s", err)) - return - } - defer sem.Release(1) + encryptionKey := os.Getenv("POCKETBASE_ENCRYPTION_KEY") + if len(encryptionKey) == 0 { + app.Logger().Error("POCKETBASE_ENCRYPTION_KEY not set") + return + } + origin := os.Getenv("ORIGIN") + if origin == "" { + app.Logger().Error("ORIGIN not set") + return + } - buf := bytes.NewBuffer(body) - req, err := http.NewRequest(http.MethodPost, inbox, buf) - if err != nil { - app.Logger().Error(fmt.Sprintf("Request creation failed: %s", err)) - return - } - req.Header.Add("Content-Type", "application/activity+json") - req.Header.Add("Date", strings.ReplaceAll(time.Now().UTC().Format(time.RFC1123), "UTC", "GMT")) - req.Header.Add("Host", req.Host) + algs := []httpsig.Algorithm{httpsig.RSA_SHA256} + postHeaders := []string{"(request-target)", "Date", "Digest", "Content-Type", "Host"} + expiresIn := 60 - if err := signer.SignRequest(privateKey, pubID, req, body); err != nil { - app.Logger().Error(fmt.Sprintf("Signing request failed: %s", err)) - return - } + body, err := jsonld.WithContext( + jsonld.IRI(pub.ActivityBaseURI), + jsonld.IRI(pub.SecurityContextURI), + ).Marshal(activity) + if err != nil { + app.Logger().Error(fmt.Sprintf("Failed to marshal activity: %s", err)) + return + } - resp, err := client.Do(req) - if err != nil { - app.Logger().Error(fmt.Sprintf("Error sending request to inbox %s: %s", inbox, err)) - return - } - defer resp.Body.Close() + decryptedPrivateKey, err := security.Decrypt(actor.GetString("private_key"), encryptionKey) + if err != nil { + app.Logger().Error(fmt.Sprintf("Failed to decrypt key: %s", err)) + return + } + privateKey, err := x509.ParsePKCS1PrivateKey(decryptedPrivateKey) + if err != nil { + app.Logger().Error(fmt.Sprintf("Failed to parse private key: %s", err)) + return + } + pubID := actor.GetString("iri") + "#main-key" - if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted { - body, _ := io.ReadAll(resp.Body) - app.Logger().Error(fmt.Sprintf("Inbox %s responded with %d: %s", inbox, resp.StatusCode, body)) - } else { - app.Logger().Info(fmt.Sprintf("Sent %s to %s", activity.Type, inbox), "activity", activity) - } + client := &http.Client{} + sem := semaphore.NewWeighted(5) - }(v) - } + slices.Sort(recipients) + uniqueRecipients := slices.Compact(recipients) - wg.Wait() + var wg sync.WaitGroup + for _, v := range uniqueRecipients { + wg.Add(1) + go func(inbox string) { + defer wg.Done() + + signer, _, err := httpsig.NewSigner(algs, httpsig.DigestSha256, postHeaders, httpsig.Signature, int64(expiresIn)) + if err != nil { + app.Logger().Error(fmt.Sprintf("Signer creation failed: %s", err)) + return + } + + if err := sem.Acquire(context.Background(), 1); err != nil { + app.Logger().Error(fmt.Sprintf("Semaphore acquire failed: %s", err)) + return + } + defer sem.Release(1) + + req, err := http.NewRequest(http.MethodPost, inbox, bytes.NewBuffer(body)) + if err != nil { + app.Logger().Error(fmt.Sprintf("Request creation failed: %s", err)) + return + } + req.Header.Add("Content-Type", "application/activity+json") + req.Header.Add("Date", strings.ReplaceAll(time.Now().UTC().Format(time.RFC1123), "UTC", "GMT")) + req.Header.Add("Host", req.Host) + + if err := signer.SignRequest(privateKey, pubID, req, body); err != nil { + app.Logger().Error(fmt.Sprintf("Signing request failed: %s", err)) + return + } + + resp, err := client.Do(req) + if err != nil { + app.Logger().Error(fmt.Sprintf("Error sending to inbox %s: %s", inbox, err)) + return + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted { + respBody, _ := io.ReadAll(resp.Body) + app.Logger().Error(fmt.Sprintf("Inbox %s responded with %d: %s", inbox, resp.StatusCode, respBody)) + } else { + app.Logger().Info(fmt.Sprintf("Sent %s to %s", activity.Type, inbox), "activity", activity) + } + }(v) + } + wg.Wait() + }() return nil }