diff --git a/API.md b/API.md index 1cc0556e..89115ff0 100644 --- a/API.md +++ b/API.md @@ -1952,3 +1952,21 @@ Content-Type: application/json - The `form` mode ensures compatibility with legacy or older webhook systems. - The `json` mode is recommended for modern integrations and easier backend parsing. - If you do not set the variable, the system will use `form` mode by default. + +### Streaming large media + +When `WEBHOOK_FORMAT=json`, large media attachments can be streamed directly +into the outgoing webhook request instead of being fully buffered in memory +first. This is opt-in via the `WEBHOOK_STREAM_MEDIA` environment variable, +independent of the format choice above: + +```bash +export WEBHOOK_STREAM_MEDIA=true # enable streaming; only takes effect when WEBHOOK_FORMAT=json +``` + +Left unset, it defaults to `false`, so upgrading does not silently change how +existing `WEBHOOK_FORMAT=json` deployments send webhooks — you must opt in +explicitly. It has no effect in `form` mode, since streaming is only +implemented for the JSON body format. The resulting webhook payload is +byte-for-byte identical either way — this setting only affects how much memory +sending it uses. diff --git a/README.md b/README.md index e797ceff..ce361d41 100644 --- a/README.md +++ b/README.md @@ -126,6 +126,7 @@ WUZAPI_GLOBAL_HMAC_KEY=your_global_hmac_key_here ``` TZ=America/New_York WEBHOOK_FORMAT=json +WEBHOOK_STREAM_MEDIA=true SESSION_DEVICE_NAME=WuzAPI WUZAPI_PORT=8080 WUZAPI_GLOBAL_WEBHOOK=https://your-global-webhook.url @@ -166,6 +167,7 @@ No database configuration needed - SQLite is used by default if no PostgreSQL se ``` TZ=America/New_York WEBHOOK_FORMAT=json # or "form" for the default +WEBHOOK_STREAM_MEDIA=true # opt-in, defaults to "false"; only takes effect when WEBHOOK_FORMAT=json SESSION_DEVICE_NAME=WuzAPI WUZAPI_PORT=8080 # Port for the WuzAPI server WUZAPI_GLOBAL_WEBHOOK= # Global webhook URL for all instances diff --git a/helpers.go b/helpers.go index ba12b4db..aca2f633 100644 --- a/helpers.go +++ b/helpers.go @@ -8,6 +8,7 @@ import ( "crypto/hmac" "crypto/rand" "crypto/sha256" + "encoding/base64" "encoding/binary" "encoding/hex" "encoding/json" @@ -25,7 +26,6 @@ import ( "runtime/debug" "strings" "sync" - "time" "github.com/go-resty/resty/v2" @@ -94,9 +94,9 @@ type WebhookErrorPayload struct { // ProxyConfig holds per-user proxy settings for WhatsApp and webhook delivery. type ProxyConfig struct { - Enabled bool `json:"enabled"` + Enabled bool `json:"enabled"` ProxyURL string `json:"proxyURL"` - WebhookUseProxy *bool `json:"webhookUseProxy,omitempty"` + WebhookUseProxy *bool `json:"webhookUseProxy,omitempty"` } func resolveWebhookUseProxy(perUser *bool) bool { @@ -310,6 +310,7 @@ func callHookWithHmac(myurl string, payload map[string]string, userID string, en var lastError error var body interface{} = payload + var lastJsonBody []byte // Starts the retry loop. for attempt := 0; attempt < maxRetries; attempt++ { @@ -330,30 +331,15 @@ func callHookWithHmac(myurl string, payload map[string]string, userID string, en var req *resty.Request var hmacSignature string - var marshalErr error format := os.Getenv("WEBHOOK_FORMAT") if format == "json" { - var jsonBody []byte - - if jsonStr, ok := payload["jsonData"]; ok { - var postmap map[string]interface{} - - if err := json.Unmarshal([]byte(jsonStr), &postmap); err == nil { - if instanceName, ok := payload["instanceName"]; ok { - postmap["instanceName"] = instanceName - } - postmap["userID"] = userID - body = postmap - } - } - - // Marshal body to JSON for HMAC signature - jsonBody, marshalErr = json.Marshal(body) - if marshalErr != nil { - log.Error().Err(marshalErr).Msg("Failed to marshal body for HMAC") - } + // payload["jsonData"] is already the final JSON (the caller + // embeds userID/instanceName before marshaling), so sign and + // send it as-is. + jsonBody := []byte(payload["jsonData"]) + lastJsonBody = jsonBody // Generate HMAC signature if key exists if len(encryptedHmacKey) > 0 && len(jsonBody) > 0 { @@ -364,7 +350,7 @@ func callHookWithHmac(myurl string, payload map[string]string, userID string, en } } - req = client.R().SetHeader("Content-Type", "application/json").SetBody(body) + req = client.R().SetHeader("Content-Type", "application/json").SetBody(jsonBody) } else { @@ -381,7 +367,6 @@ func callHookWithHmac(myurl string, payload map[string]string, userID string, en } } req = client.R().SetFormData(payload) - body = payload } if hmacSignature != "" { @@ -424,10 +409,161 @@ func callHookWithHmac(myurl string, payload map[string]string, userID string, en for k, v := range p { errorPayloadMap[k] = v } - } else if p, ok := body.(map[string]interface{}); ok { + } + + if len(lastJsonBody) > 0 { + var postmap map[string]interface{} + if err := json.Unmarshal(lastJsonBody, &postmap); err == nil { + errorPayloadMap = postmap + } + } + + errorPayload := WebhookErrorPayload{ + URL: myurl, + Payload: errorPayloadMap, + UserID: userID, + EncryptedHmacKey: hex.EncodeToString(encryptedHmacKey), + AttemptTime: time.Now(), + ErrorMessage: lastError.Error(), + } + + PublishDataErrorToQueue(errorPayload) + } +} + +// webhook body as a stream +func newStreamedWebhookBody(metaJSON []byte, mediaBytes []byte) io.Reader { + prefix := metaJSON[:len(metaJSON)-1] // drop trailing '}' + sep := []byte(`,"base64":"`) + if len(prefix) == 1 { // metaJSON was "{}" -- no fields to comma-separate from + sep = []byte(`"base64":"`) + } + + pr, pw := io.Pipe() + go func() { + _, werr := pw.Write(prefix) + if werr == nil { + _, werr = pw.Write(sep) + } + if werr == nil { + enc := base64.NewEncoder(base64.StdEncoding, pw) + if _, werr = enc.Write(mediaBytes); werr == nil { + werr = enc.Close() + } + } + if werr == nil { + _, werr = pw.Write([]byte(`"}`)) + } + pw.CloseWithError(werr) + }() + return pr +} + +// reports whether large media should be streamed directly +func shouldStreamMedia() bool { + if os.Getenv("WEBHOOK_FORMAT") != "json" { + return false + } + switch strings.ToLower(os.Getenv("WEBHOOK_STREAM_MEDIA")) { + case "true", "1", "yes": + return true + default: + return false + } +} + +// sends a webhook whose body embeds large media directly into the request body +func callHookStreamedWithHmac(myurl string, metaFields map[string]interface{}, mediaBytes []byte, userID string, encryptedHmacKey []byte) { + log.Info().Str("url", myurl).Str("userID", userID).Msg("Sending streamed POST to client with retry logic") + + client := clientManager.GetHTTPClient(userID) + if client == nil { + log.Warn().Str("url", myurl).Str("userID", userID).Msg("HTTP client is nil for user, skipping streamed webhook") + return + } + + metaJSON, err := json.Marshal(metaFields) + if err != nil { + log.Error().Err(err).Msg("Failed to marshal webhook metadata") + return + } + + // newBody returns a fresh streaming reader for the full JSON body + newBody := func() io.Reader { + return newStreamedWebhookBody(metaJSON, mediaBytes) + } + + maxRetries := 1 + if *webhookRetryEnabled { + maxRetries = *webhookRetryCount + } + + var lastError error + + for attempt := 0; attempt < maxRetries; attempt++ { + if attempt > 0 { + backoffFactor := 1 << uint(attempt-1) + delayDuration := time.Duration(*webhookRetryDelaySeconds) * time.Second * time.Duration(backoffFactor) + log.Warn(). + Int("attempt", attempt+1). + Str("url", myurl). + Dur("delay", delayDuration). + Msg("Retrying streamed webhook request with exponential backoff...") + time.Sleep(delayDuration) + } + + var hmacSignature string + if len(encryptedHmacKey) > 0 { + sig, herr := generateHmacSignatureFromReader(newBody(), encryptedHmacKey) + if herr != nil { + log.Error().Err(herr).Msg("Failed to generate HMAC signature for streamed webhook") + } else { + hmacSignature = sig + } + } + + req := client.R().SetHeader("Content-Type", "application/json").SetBody(newBody()) + if hmacSignature != "" { + req.SetHeader("x-hmac-signature", hmacSignature) + } + + resp, postErr := req.Post(myurl) + lastError = postErr + + if postErr != nil { + log.Error().Err(postErr).Int("attempt", attempt+1).Str("url", myurl).Msg("Streamed webhook failed due to network/IO error") + continue + } + + if resp.StatusCode() < 200 || resp.StatusCode() >= 300 { + lastError = fmt.Errorf("unexpected status code: %d. Body: %s", resp.StatusCode(), string(resp.Body())) + log.Error(). + Int("status", resp.StatusCode()). + Int("attempt", attempt+1). + Str("url", myurl). + Msg("Streamed webhook failed due to non-2xx status code") + + if !*webhookRetryEnabled { + break + } + continue + } + + log.Info().Int("status", resp.StatusCode()).Str("url", myurl).Msg("Streamed webhook call successful") + return + } - errorPayloadMap = p + if lastError != nil { + log.Error().Str("url", myurl).Msg("Streamed webhook permanently failed after all retries. Sending to error queue...") + + // Only the rare permanent-failure path pays for a full base64 + // string — needed so the error queue has a complete, replayable + // payload. + errorPayloadMap := make(map[string]interface{}, len(metaFields)+1) + for k, v := range metaFields { + errorPayloadMap[k] = v } + errorPayloadMap["base64"] = base64.StdEncoding.EncodeToString(mediaBytes) errorPayload := WebhookErrorPayload{ URL: myurl, @@ -623,6 +759,11 @@ func ProcessOutgoingMedia(userID string, contactJID string, messageID string, da // generateHmacSignature generates HMAC-SHA256 signature for webhook payload func generateHmacSignature(payload []byte, encryptedHmacKey []byte) (string, error) { + return generateHmacSignatureFromReader(bytes.NewReader(payload), encryptedHmacKey) +} + +// generateHmacSignature generates HMAC-SHA256 signature for streamed webhook payload +func generateHmacSignatureFromReader(r io.Reader, encryptedHmacKey []byte) (string, error) { if len(encryptedHmacKey) == 0 { return "", nil } @@ -635,7 +776,9 @@ func generateHmacSignature(payload []byte, encryptedHmacKey []byte) (string, err // Generate HMAC h := hmac.New(sha256.New, []byte(hmacKey)) - h.Write(payload) + if _, err := io.Copy(h, r); err != nil { + return "", fmt.Errorf("failed to hash streamed payload: %w", err) + } return hex.EncodeToString(h.Sum(nil)), nil } diff --git a/media.go b/media.go index 29d0a940..1796edea 100644 --- a/media.go +++ b/media.go @@ -2,7 +2,10 @@ package main import ( "context" + "encoding/base64" + "encoding/json" "mime" + "net/http" "os" "path/filepath" "time" @@ -24,6 +27,24 @@ type mediaS3Config struct { MediaDelivery string } +// lazyBase64 carries raw media bytes through postmap without eagerly +// encoding them +type lazyBase64 struct { + data []byte +} + +func (b lazyBase64) MarshalJSON() ([]byte, error) { + return json.Marshal(base64.StdEncoding.EncodeToString(b.data)) +} + +// resolves MIME Type based on WhatsApp declared preferably +func resolveMimeType(declaredMimeType string, data []byte) string { + if declaredMimeType != "" { + return declaredMimeType + } + return http.DetectContentType(data) +} + func (mycli *MyClient) processMedia( msg whatsmeow.DownloadableMessage, mimeType string, @@ -52,8 +73,11 @@ func (mycli *MyClient) processMedia( } ext := fallbackExt - if exts, _ := mime.ExtensionsByType(mimeType); len(exts) > 0 { - ext = exts[0] + //prefer fallbackExt when available + if len(ext) == 0 { + if exts, _ := mime.ExtensionsByType(mimeType); len(exts) > 0 { + ext = exts[0] + } } tmpPath := filepath.Join(tmpDir, messageID+ext) @@ -88,13 +112,8 @@ func (mycli *MyClient) processMedia( } if s3cfg.MediaDelivery == "base64" || s3cfg.MediaDelivery == "both" { - b64, mime_, err := fileToBase64(tmpPath) - if err != nil { - log.Error().Err(err).Msg("Failed to convert media to base64") - return - } - postmap["base64"] = b64 - postmap["mimeType"] = mime_ + postmap["base64"] = lazyBase64{data: data} + postmap["mimeType"] = resolveMimeType(mimeType, data) postmap["fileName"] = filepath.Base(tmpPath) } diff --git a/wmiau.go b/wmiau.go index a8fec072..ccd3a595 100644 --- a/wmiau.go +++ b/wmiau.go @@ -9,13 +9,13 @@ import ( "encoding/json" "errors" "fmt" - "net/http" "net/url" "os" "path/filepath" "runtime/debug" "strconv" "strings" + "sync" "time" "github.com/go-resty/resty/v2" @@ -32,7 +32,6 @@ import ( "go.mau.fi/whatsmeow/types/events" waLog "go.mau.fi/whatsmeow/util/log" "golang.org/x/net/proxy" - "sync" ) // db field declaration as *sqlx.DB @@ -284,16 +283,12 @@ func sendEventWithWebHook(mycli *MyClient, postmap map[string]interface{}, path return } - // Prepare webhook data - jsonData, err := json.Marshal(postmap) - if err != nil { - log.Error().Err(err).Msg("Failed to marshal postmap to JSON") - return - } - - // Get HMAC key for this user + // Embed userID/instanceName directly so downstream webhook senders + // don't need to unmarshal the JSON just to add two fields and re-marshal + postmap["instanceName"] = "" var encryptedHmacKey []byte if userinfo, found := userinfocache.Get(mycli.token); found { + postmap["instanceName"] = userinfo.(Values).Get("Name") encryptedB64 := userinfo.(Values).Get("HmacKeyEncrypted") if encryptedB64 != "" { var err error @@ -303,13 +298,43 @@ func sendEventWithWebHook(mycli *MyClient, postmap map[string]interface{}, path } } } + postmap["userID"] = mycli.userID + + lb, hasMedia := postmap["base64"].(lazyBase64) + streaming := hasMedia && webhookurl != "" && shouldStreamMedia() + + if streaming { + // Stream large media into the outgoing HTTP body instead of + // materializing the base64 string + full JSON document in memory + meta := make(map[string]interface{}, len(postmap)) + for k, v := range postmap { + if k != "base64" { + meta[k] = v + } + } + safeGo("callHookStreamedWithHmac", func() { + callHookStreamedWithHmac(webhookurl, meta, lb.data, mycli.userID, encryptedHmacKey) + }) + } - sendToUserWebHookWithHmac(webhookurl, path, jsonData, mycli.userID, mycli.token, encryptedHmacKey) + // global webhook / RabbitMQ still need a fully marshaled copy + needSideChannels := *globalWebhook != "" || rabbitEnabled - // Get global webhook if configured - safeGo("sendToGlobalWebHook", func() { sendToGlobalWebHook(jsonData, mycli.token, mycli.userID) }) + if !streaming || needSideChannels { + jsonData, err := json.Marshal(postmap) + if err != nil { + log.Error().Err(err).Msg("Failed to marshal postmap to JSON") + return + } + if !streaming { + sendToUserWebHookWithHmac(webhookurl, path, jsonData, mycli.userID, mycli.token, encryptedHmacKey) + } + if needSideChannels { + safeGo("sendToGlobalWebHook", func() { sendToGlobalWebHook(jsonData, mycli.token, mycli.userID) }) - safeGo("sendToGlobalRabbit", func() { sendToGlobalRabbit(jsonData, mycli.token, mycli.userID) }) + safeGo("sendToGlobalRabbit", func() { sendToGlobalRabbit(jsonData, mycli.token, mycli.userID) }) + } + } } func checkIfSubscribedToEvent(subscribedEvents []string, eventType string, userId string) bool { @@ -922,15 +947,6 @@ func (s *server) startClient(userID string, textjid string, token string, kill c deleteKillChannel(userID, kill) } -func fileToBase64(filepath string) (string, string, error) { - data, err := os.ReadFile(filepath) - if err != nil { - return "", "", err - } - mimeType := http.DetectContentType(data) - return base64.StdEncoding.EncodeToString(data), mimeType, nil -} - func (mycli *MyClient) myEventHandler(rawEvt interface{}) { txtid := mycli.userID postmap := make(map[string]interface{}) @@ -1176,24 +1192,24 @@ func (mycli *MyClient) myEventHandler(rawEvt interface{}) { } } } - - if encMessage := evt.Message.GetSecretEncryptedMessage(); encMessage != nil { - decrypted, derr := mycli.WAClient.DecryptSecretEncryptedMessage(context.Background(), evt) - if derr != nil { - log.Warn(). - Err(derr). - Str("messageID", evt.Info.ID). - Str("secretEncType", encMessage.GetSecretEncType().String()). - Msg("DecryptSecretEncryptedMessage failed") - } else if decrypted != nil { - log.Info(). - Str("messageID", evt.Info.ID). - Str("secretEncType", encMessage.GetSecretEncType().String()). - Msg("Decrypted secretEncryptedMessage; swapping evt.Message") - evt.Message = decrypted - } - } - + + if encMessage := evt.Message.GetSecretEncryptedMessage(); encMessage != nil { + decrypted, derr := mycli.WAClient.DecryptSecretEncryptedMessage(context.Background(), evt) + if derr != nil { + log.Warn(). + Err(derr). + Str("messageID", evt.Info.ID). + Str("secretEncType", encMessage.GetSecretEncType().String()). + Msg("DecryptSecretEncryptedMessage failed") + } else if decrypted != nil { + log.Info(). + Str("messageID", evt.Info.ID). + Str("secretEncType", encMessage.GetSecretEncType().String()). + Msg("Decrypted secretEncryptedMessage; swapping evt.Message") + evt.Message = decrypted + } + } + if !*skipMedia { isIncoming := !evt.Info.IsFromMe