s3/weed/notification/webhook/http.go

package webhook

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"sync"
	"time"

	"github.com/seaweedfs/seaweedfs/weed/glog"
	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
)

const (
	maxWebhookRetryDepth = 2
)

type httpClient struct {
	endpoint   string
	token      string
	timeout    time.Duration
	client     *http.Client // Reused HTTP client with redirect prevention
	endpointMu sync.RWMutex
	finalURL   string // Cached final URL after following redirects
}

func newHTTPClient(cfg *config) (*httpClient, error) {
	return &httpClient{
		endpoint: cfg.endpoint,
		token:    cfg.authBearerToken,
		timeout:  time.Duration(cfg.timeoutSeconds) * time.Second,
		client: &http.Client{
			CheckRedirect: func(req *http.Request, via []*http.Request) error {
				return http.ErrUseLastResponse
			},
		},
	}, nil
}

func (h *httpClient) sendMessage(message *webhookMessage) error {
	return h.sendMessageWithRetry(message, 0)
}

func (h *httpClient) sendMessageWithRetry(message *webhookMessage, depth int) error {
	// Prevent infinite recursion
	if depth > maxWebhookRetryDepth {
		return fmt.Errorf("webhook max retry depth exceeded")
	}

	// Serialize the protobuf message to JSON for HTTP payload
	notificationData, err := json.Marshal(message.Notification)
	if err != nil {
		return fmt.Errorf("failed to marshal notification: %w", err)
	}

	payload := map[string]interface{}{
		"key":        message.Key,
		"event_type": message.EventType,
		"message":    json.RawMessage(notificationData),
	}

	jsonData, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("failed to marshal message: %w", err)
	}

	// Use cached final URL if available, otherwise use original endpoint
	h.endpointMu.RLock()
	targetURL := h.finalURL
	usingCachedURL := targetURL != ""
	if targetURL == "" {
		targetURL = h.endpoint
	}
	h.endpointMu.RUnlock()

	req, err := http.NewRequest(http.MethodPost, targetURL, bytes.NewBuffer(jsonData))
	if err != nil {
		return fmt.Errorf("failed to create request: %w", err)
	}

	req.Header.Set("Content-Type", "application/json")
	if h.token != "" {
		req.Header.Set("Authorization", "Bearer "+h.token)
	}

	// Apply timeout via context (not on client) to avoid redundancy
	if h.timeout > 0 {
		ctx, cancel := context.WithTimeout(context.Background(), h.timeout)
		defer cancel()
		req = req.WithContext(ctx)
	}

	resp, err := h.client.Do(req)
	if err != nil {
		if drainErr := drainResponse(resp); drainErr != nil {
			glog.Errorf("failed to drain response: %v", drainErr)
		}

		// If using cached URL and request failed, clear cache and retry with original endpoint
		if usingCachedURL && depth == 0 {
			glog.V(1).Infof("Webhook request to cached URL %s failed, clearing cache and retrying with original endpoint", targetURL)
			h.setFinalURL("")
			return h.sendMessageWithRetry(message, depth+1)
		}

		return fmt.Errorf("failed to send request: %w", err)
	}

	// Handle redirects by caching the final destination and recreating POST request
	if resp.StatusCode >= 300 && resp.StatusCode < 400 {
		// Drain and close response body to enable connection reuse
		util_http.CloseResponse(resp)

		location := resp.Header.Get("Location")
		if location == "" {
			return fmt.Errorf("webhook returned redirect status %d without Location header", resp.StatusCode)
		}

		// Resolve relative URLs against the request URL
		reqURL := req.URL
		finalURL, err := reqURL.Parse(location)
		if err != nil {
			return fmt.Errorf("failed to parse redirect location: %w", err)
		}

		// Update finalURL to follow the redirect for this attempt
		finalURLStr := finalURL.String()
		h.setFinalURL(finalURLStr)

		if depth == 0 {
			glog.V(1).Infof("Webhook endpoint redirected from %s to %s, caching final destination", targetURL, finalURLStr)
		} else {
			glog.V(1).Infof("Webhook endpoint redirected from %s to %s (following redirect on retry)", targetURL, finalURLStr)
		}

		// Recreate the POST request to the final destination (increment depth to prevent infinite loops)
		return h.sendMessageWithRetry(message, depth+1)
	}

	// If using cached URL and got an error response, clear cache and retry with original endpoint
	if resp.StatusCode >= 400 && usingCachedURL && depth == 0 {
		// Drain and close response body to enable connection reuse
		util_http.CloseResponse(resp)

		glog.V(1).Infof("Webhook request to cached URL %s returned error %d, clearing cache and retrying with original endpoint", targetURL, resp.StatusCode)
		h.setFinalURL("")
		return h.sendMessageWithRetry(message, depth+1)
	}

	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		// Drain and close response body before returning error to enable connection reuse
		util_http.CloseResponse(resp)
		return fmt.Errorf("webhook returned status code: %d", resp.StatusCode)
	}

	// Drain and close response body on success to enable connection reuse
	util_http.CloseResponse(resp)

	return nil
}

func (h *httpClient) setFinalURL(url string) {
	h.endpointMu.Lock()
	h.finalURL = url
	h.endpointMu.Unlock()
}

func drainResponse(resp *http.Response) error {
	if resp == nil || resp.Body == nil {
		return nil
	}

	_, err := io.ReadAll(resp.Body)

	return errors.Join(
		err,
		resp.Body.Close(),
	)
}