makes PostActivity fully async

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")
+	go func() {
-	if len(encryptionKey) == 0 {
+		defer func() {
-		return fmt.Errorf("POCKETBASE_ENCRYPTION_KEY not set")
+			if r := recover(); r != nil {
-	}
+				app.Logger().Error(fmt.Sprintf("Recovered from panic in PostActivity: %v", r))
-	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
 			}
+		}()
 
-			if err := sem.Acquire(context.Background(), 1); err != nil {
+		encryptionKey := os.Getenv("POCKETBASE_ENCRYPTION_KEY")
-				app.Logger().Error(fmt.Sprintf("Semaphore acquire failed: %s", err))
+		if len(encryptionKey) == 0 {
-				return
+			app.Logger().Error("POCKETBASE_ENCRYPTION_KEY not set")
-			}
+			return
-			defer sem.Release(1)
+		}
+		origin := os.Getenv("ORIGIN")
+		if origin == "" {
+			app.Logger().Error("ORIGIN not set")
+			return
+		}
 
-			buf := bytes.NewBuffer(body)
+		algs := []httpsig.Algorithm{httpsig.RSA_SHA256}
-			req, err := http.NewRequest(http.MethodPost, inbox, buf)
+		postHeaders := []string{"(request-target)", "Date", "Digest", "Content-Type", "Host"}
-			if err != nil {
+		expiresIn := 60
-				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 {
+		body, err := jsonld.WithContext(
-				app.Logger().Error(fmt.Sprintf("Signing request failed: %s", err))
+			jsonld.IRI(pub.ActivityBaseURI),
-				return
+			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)
+		decryptedPrivateKey, err := security.Decrypt(actor.GetString("private_key"), encryptionKey)
-			if err != nil {
+		if err != nil {
-				app.Logger().Error(fmt.Sprintf("Error sending request to inbox %s: %s", inbox, err))
+			app.Logger().Error(fmt.Sprintf("Failed to decrypt key: %s", err))
-				return
+			return
-			}
+		}
-			defer resp.Body.Close()
+		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 {
+		client := &http.Client{}
-				body, _ := io.ReadAll(resp.Body)
+		sem := semaphore.NewWeighted(5)
-				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)
-			}
 
-		}(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
 }