fix cdn download logic
This commit is contained in:
Huaaudio
parent
607037472d
commit
d952e391ec
2 changed files with 228 additions and 40 deletions
|
|
@ -34,6 +34,8 @@ const (
|
|||
weixinMediaMaxBytes = 100 << 20
|
||||
weixinTypingKeepAlive = 5 * time.Second
|
||||
weixinUploadRetryMax = 3
|
||||
weixinDownloadRetryMax = 2
|
||||
weixinDownloadRetryDelay = 300 * time.Millisecond
|
||||
weixinVoiceTranscodeTimeout = 15 * time.Second
|
||||
)
|
||||
|
||||
|
|
@ -163,49 +165,99 @@ func buildCDNDownloadURL(base, encryptedQueryParam string) string {
|
|||
"/download?encrypted_query_param=" + url.QueryEscape(encryptedQueryParam)
|
||||
}
|
||||
|
||||
func shouldRetryCDNDownload(statusCode int) bool {
|
||||
// statusCode=0 represents transport/build errors from the HTTP client.
|
||||
return statusCode == 0 || statusCode >= 500 || statusCode == http.StatusTooManyRequests
|
||||
}
|
||||
|
||||
func buildCDNUploadURL(base, uploadParam, filekey string) string {
|
||||
return strings.TrimRight(base, "/") +
|
||||
"/upload?encrypted_query_param=" + url.QueryEscape(uploadParam) +
|
||||
"&filekey=" + url.QueryEscape(filekey)
|
||||
}
|
||||
|
||||
func uniqCDNURLs(urls []string) []string {
|
||||
seen := make(map[string]struct{}, len(urls))
|
||||
out := make([]string, 0, len(urls))
|
||||
for _, raw := range urls {
|
||||
u := strings.TrimSpace(raw)
|
||||
if u == "" {
|
||||
continue
|
||||
}
|
||||
if _, ok := seen[u]; ok {
|
||||
continue
|
||||
}
|
||||
seen[u] = struct{}{}
|
||||
out = append(out, u)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (c *WeixinChannel) downloadCDNBufferOnce(ctx context.Context, downloadURL string) ([]byte, int, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
resp, err := c.api.HttpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
|
||||
return nil, resp.StatusCode, fmt.Errorf("cdn download HTTP %d: %s", resp.StatusCode, string(body))
|
||||
}
|
||||
|
||||
data, err := io.ReadAll(io.LimitReader(resp.Body, weixinMediaMaxBytes+1))
|
||||
if err != nil {
|
||||
return nil, resp.StatusCode, err
|
||||
}
|
||||
if len(data) > weixinMediaMaxBytes {
|
||||
return nil, resp.StatusCode, fmt.Errorf("cdn media too large: %d bytes", len(data))
|
||||
}
|
||||
return data, resp.StatusCode, nil
|
||||
}
|
||||
|
||||
func (c *WeixinChannel) downloadCDNBuffer(
|
||||
ctx context.Context,
|
||||
encryptedQueryParam,
|
||||
fullURL string,
|
||||
) ([]byte, error) {
|
||||
downloadURL := strings.TrimSpace(fullURL)
|
||||
if downloadURL == "" {
|
||||
downloadURL = buildCDNDownloadURL(c.cdnBaseURL(), encryptedQueryParam)
|
||||
candidates := uniqCDNURLs([]string{
|
||||
strings.TrimSpace(fullURL),
|
||||
func() string {
|
||||
if strings.TrimSpace(encryptedQueryParam) == "" {
|
||||
return ""
|
||||
}
|
||||
req, err := http.NewRequestWithContext(
|
||||
ctx,
|
||||
http.MethodGet,
|
||||
downloadURL,
|
||||
nil,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp, err := c.api.HttpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
|
||||
return nil, fmt.Errorf("cdn download HTTP %d: %s", resp.StatusCode, string(body))
|
||||
return buildCDNDownloadURL(c.cdnBaseURL(), encryptedQueryParam)
|
||||
}(),
|
||||
})
|
||||
if len(candidates) == 0 {
|
||||
return nil, fmt.Errorf("missing CDN download URL")
|
||||
}
|
||||
|
||||
data, err := io.ReadAll(io.LimitReader(resp.Body, weixinMediaMaxBytes+1))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(data) > weixinMediaMaxBytes {
|
||||
return nil, fmt.Errorf("cdn media too large: %d bytes", len(data))
|
||||
}
|
||||
var lastErr error
|
||||
for _, downloadURL := range candidates {
|
||||
for attempt := 1; attempt <= weixinDownloadRetryMax; attempt++ {
|
||||
data, statusCode, err := c.downloadCDNBufferOnce(ctx, downloadURL)
|
||||
if err == nil {
|
||||
return data, nil
|
||||
}
|
||||
lastErr = fmt.Errorf("%w (attempt=%d url=%s)", err, attempt, downloadURL)
|
||||
if !shouldRetryCDNDownload(statusCode) {
|
||||
break
|
||||
}
|
||||
if attempt < weixinDownloadRetryMax {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case <-time.After(weixinDownloadRetryDelay):
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil, lastErr
|
||||
}
|
||||
|
||||
func (c *WeixinChannel) downloadAndDecryptCDNBuffer(
|
||||
|
|
@ -224,6 +276,33 @@ func (c *WeixinChannel) downloadAndDecryptCDNBuffer(
|
|||
return decryptAESECB(data, key)
|
||||
}
|
||||
|
||||
func (c *WeixinChannel) downloadImageBuffer(
|
||||
ctx context.Context,
|
||||
img *ImageItem,
|
||||
key []byte,
|
||||
) ([]byte, error) {
|
||||
if img == nil {
|
||||
return nil, fmt.Errorf("image item is nil")
|
||||
}
|
||||
if img.Media != nil {
|
||||
data, err := c.downloadAndDecryptCDNBuffer(ctx, img.Media.EncryptQueryParam, img.Media.FullURL, key)
|
||||
if err == nil {
|
||||
return data, nil
|
||||
}
|
||||
if img.ThumbMedia == nil {
|
||||
return nil, fmt.Errorf("image download failed: %w", err)
|
||||
}
|
||||
}
|
||||
if img.ThumbMedia != nil {
|
||||
data, err := c.downloadAndDecryptCDNBuffer(ctx, img.ThumbMedia.EncryptQueryParam, img.ThumbMedia.FullURL, key)
|
||||
if err == nil {
|
||||
return data, nil
|
||||
}
|
||||
return nil, fmt.Errorf("image download failed: %w", err)
|
||||
}
|
||||
return nil, fmt.Errorf("image media is nil")
|
||||
}
|
||||
|
||||
func detectMediaMetadata(data []byte, fallbackName, fallbackContentType string) (string, string) {
|
||||
contentType := strings.TrimSpace(fallbackContentType)
|
||||
ext := filepath.Ext(fallbackName)
|
||||
|
|
@ -446,21 +525,20 @@ func (c *WeixinChannel) downloadMediaFromItem(
|
|||
|
||||
switch item.Type {
|
||||
case MessageItemTypeImage:
|
||||
if item.ImageItem == nil {
|
||||
return "", fmt.Errorf("image media is nil")
|
||||
}
|
||||
key, ok, err := imageAESKey(item.ImageItem)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
data, err := c.downloadAndDecryptCDNBuffer(
|
||||
ctx,
|
||||
item.ImageItem.Media.EncryptQueryParam,
|
||||
item.ImageItem.Media.FullURL,
|
||||
func() []byte {
|
||||
decryptKey := func() []byte {
|
||||
if ok {
|
||||
return key
|
||||
}
|
||||
return nil
|
||||
}(),
|
||||
)
|
||||
}()
|
||||
data, err := c.downloadImageBuffer(ctx, item.ImageItem, decryptKey)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -81,6 +81,116 @@ func TestDownloadAndDecryptCDNBuffer(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestDownloadAndDecryptCDNBufferUsesFullURLWhenProvided(t *testing.T) {
|
||||
key := []byte("1234567890abcdef")
|
||||
plaintext := []byte("hello weixin")
|
||||
ciphertext, err := encryptAESECB(plaintext, key)
|
||||
if err != nil {
|
||||
t.Fatalf("encryptAESECB() error = %v", err)
|
||||
}
|
||||
|
||||
fullURLAttempts := 0
|
||||
ch := &WeixinChannel{
|
||||
api: &ApiClient{
|
||||
HttpClient: &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
|
||||
if r.URL.String() == "https://full.example.com/download" {
|
||||
fullURLAttempts++
|
||||
return &http.Response{
|
||||
StatusCode: http.StatusOK,
|
||||
Body: io.NopCloser(bytes.NewReader(ciphertext)),
|
||||
Header: make(http.Header),
|
||||
}, nil
|
||||
}
|
||||
t.Fatalf("unexpected fallback request: %s", r.URL.String())
|
||||
return nil, nil
|
||||
})},
|
||||
},
|
||||
config: config.WeixinConfig{
|
||||
CDNBaseURL: "https://cdn.example.com",
|
||||
},
|
||||
typingCache: make(map[string]typingTicketCacheEntry),
|
||||
}
|
||||
|
||||
got, err := ch.downloadAndDecryptCDNBuffer(context.Background(), "token", "https://full.example.com/download", key)
|
||||
if err != nil {
|
||||
t.Fatalf("downloadAndDecryptCDNBuffer() error = %v", err)
|
||||
}
|
||||
if !bytes.Equal(got, plaintext) {
|
||||
t.Fatalf("downloadAndDecryptCDNBuffer() = %q, want %q", got, plaintext)
|
||||
}
|
||||
if fullURLAttempts == 0 {
|
||||
t.Fatalf("fullURLAttempts = %d, want > 0", fullURLAttempts)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDownloadAndDecryptCDNBufferFallsBackToConstructedURLWhenFullURLFails(t *testing.T) {
|
||||
key := []byte("1234567890abcdef")
|
||||
plaintext := []byte("hello weixin")
|
||||
ciphertext, err := encryptAESECB(plaintext, key)
|
||||
if err != nil {
|
||||
t.Fatalf("encryptAESECB() error = %v", err)
|
||||
}
|
||||
|
||||
fullURLAttempts := 0
|
||||
constructedAttempts := 0
|
||||
ch := &WeixinChannel{
|
||||
api: &ApiClient{
|
||||
HttpClient: &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
|
||||
if r.URL.String() == "https://full.example.com/download?encrypted_query_param=token&taskid=123" {
|
||||
fullURLAttempts++
|
||||
return &http.Response{
|
||||
StatusCode: http.StatusInternalServerError,
|
||||
Body: io.NopCloser(bytes.NewReader(nil)),
|
||||
Header: make(http.Header),
|
||||
}, nil
|
||||
}
|
||||
if r.URL.String() != "https://cdn.example.com/download?encrypted_query_param=token" {
|
||||
t.Fatalf("unexpected fallback request: %s", r.URL.String())
|
||||
}
|
||||
constructedAttempts++
|
||||
return &http.Response{
|
||||
StatusCode: http.StatusOK,
|
||||
Body: io.NopCloser(bytes.NewReader(ciphertext)),
|
||||
Header: make(http.Header),
|
||||
}, nil
|
||||
})},
|
||||
},
|
||||
config: config.WeixinConfig{
|
||||
CDNBaseURL: "https://cdn.example.com",
|
||||
},
|
||||
typingCache: make(map[string]typingTicketCacheEntry),
|
||||
}
|
||||
|
||||
got, err := ch.downloadAndDecryptCDNBuffer(
|
||||
context.Background(),
|
||||
"token",
|
||||
"https://full.example.com/download?encrypted_query_param=token&taskid=123",
|
||||
key,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("downloadAndDecryptCDNBuffer() error = %v", err)
|
||||
}
|
||||
if !bytes.Equal(got, plaintext) {
|
||||
t.Fatalf("downloadAndDecryptCDNBuffer() = %q, want %q", got, plaintext)
|
||||
}
|
||||
if fullURLAttempts == 0 {
|
||||
t.Fatalf("fullURLAttempts = %d, want > 0", fullURLAttempts)
|
||||
}
|
||||
if constructedAttempts == 0 {
|
||||
t.Fatalf("constructedAttempts = %d, want > 0", constructedAttempts)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildCDNDownloadURLEscapesOpaqueToken(t *testing.T) {
|
||||
token := "MFcCAQAESzBJAgEAAgSieMV9AgM9CcwCBEoKPqICBGnHZB0EJDk4OWY5YWU0LTc4OGItNGQ5Ni1iMjZhLWU4YjhlMmEwOWVkZgIEIR0IAgIBAAQFAExUPQA%3D"
|
||||
|
||||
got := buildCDNDownloadURL("https://cdn.example.com", token)
|
||||
|
||||
if got != "https://cdn.example.com/download?encrypted_query_param=MFcCAQAESzBJAgEAAgSieMV9AgM9CcwCBEoKPqICBGnHZB0EJDk4OWY5YWU0LTc4OGItNGQ5Ni1iMjZhLWU4YjhlMmEwOWVkZgIEIR0IAgIBAAQFAExUPQA%253D" {
|
||||
t.Fatalf("buildCDNDownloadURL() = %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUploadBufferToCDN(t *testing.T) {
|
||||
key := []byte("1234567890abcdef")
|
||||
plaintext := []byte("upload me")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue