feat(zalo): add file and image upload/send support (#62)

## Summary
- Implement UploadImage, SendImage, UploadFile, SendFile in Zalo Personal protocol
- Handle WebSocket control events (cmd=601) for async file upload callbacks
- Add media attachment routing in channel.Send() — auto-detects image vs file
- Add MEDIA: prefix support in message tool for agent file attachments
- Add group_id metadata routing for correct Zalo group API selection

## Review fixes (dc475bc)
- Fix CI: %s for json.Number (was %d)
- Fix path traversal: parseMediaPath restricted to os.TempDir()
- Add 25MB file size limit before upload reads
- Drain stale uploadCallbacks on Listener.reset()
- Split send.go (688 LOC) into send.go, send_image.go, send_file.go, send_helpers.go
- Add 12 unit tests for parseMediaPath
This commit is contained in:
Luan Vu authored and GitHub committed 2026-03-06 23:20:50 +07:00
1 parent d35046a224
commit 95885da14e
12 files changed
+1033 -2

No files matched your search

+3
View File
@@ -38,3 +38,6 @@ deploy-*.sh
docs/technical-diaries
plans
release-manifest.json
# Test credentials (never commit)
tests/**/creds.json
+1 -1
View File
@@ -7,7 +7,7 @@
services:
jaeger:
image: jaegertracing/all-in-one:1.68
image: jaegertracing/all-in-one:1.68.0
ports:
- "16686:16686" # Jaeger UI
- "4317:4317" # OTLP gRPC
+1
View File
@@ -95,6 +95,7 @@ var userSeedFilesOpen = []string{
// and don't need a bootstrap onboarding ritual. They just need to learn the user's profile.
var userSeedFilesPredefined = []string{
UserFile,
BootstrapFile,
}
// SeedUserFiles seeds embedded templates into user_context_files for a new user.
+44 -1
View File
@@ -159,7 +159,50 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error {
}
}
return c.sendChunkedText(ctx, sess, msg.ChatID, threadType, msg.Content)
// Send media attachments.
for _, media := range msg.Media {
if protocol.IsImageFile(media.URL) {
if err := c.sendImage(ctx, sess, msg.ChatID, threadType, media.URL, media.Caption); err != nil {
slog.Warn("zalo_personal: failed to send image", "path", media.URL, "error", err)
}
} else {
if err := c.sendFile(ctx, sess, msg.ChatID, threadType, media.URL); err != nil {
slog.Warn("zalo_personal: failed to send file", "path", media.URL, "error", err)
}
}
}
// Send text content (if any remains after media).
if msg.Content != "" {
return c.sendChunkedText(ctx, sess, msg.ChatID, threadType, msg.Content)
}
return nil
}
// sendImage uploads and sends an image file to a Zalo thread.
func (c *Channel) sendImage(ctx context.Context, sess *protocol.Session, chatID string, threadType protocol.ThreadType, filePath, caption string) error {
upload, err := protocol.UploadImage(ctx, sess, chatID, threadType, filePath)
if err != nil {
return fmt.Errorf("upload: %w", err)
}
_, err = protocol.SendImage(ctx, sess, chatID, threadType, upload, caption)
return err
}
// sendFile uploads and sends a file to a Zalo thread.
func (c *Channel) sendFile(ctx context.Context, sess *protocol.Session, chatID string, threadType protocol.ThreadType, filePath string) error {
ln := c.getListener()
if ln == nil {
return fmt.Errorf("listener not available for file upload")
}
upload, err := protocol.UploadFile(ctx, sess, ln, chatID, threadType, filePath)
if err != nil {
return fmt.Errorf("upload: %w", err)
}
_, err = protocol.SendFile(ctx, sess, chatID, threadType, upload)
return err
}
func (c *Channel) listenLoop(ctx context.Context) {
@@ -36,6 +36,8 @@ type Listener struct {
retryStates map[string]*retryState
uploadCallbacks sync.Map // fileID (string) → chan string (fileURL)
messageCh chan Message
disconnectedCh chan CloseInfo
closedCh chan CloseInfo
@@ -46,6 +48,19 @@ type Listener struct {
wg sync.WaitGroup
}
// RegisterUploadCallback registers a callback channel for a file upload completion.
// Returns a channel that will receive the fileURL when the upload is done.
func (ln *Listener) RegisterUploadCallback(fileID string) <-chan string {
ch := make(chan string, 1)
ln.uploadCallbacks.Store(fileID, ch)
return ch
}
// CancelUploadCallback removes a pending upload callback.
func (ln *Listener) CancelUploadCallback(fileID string) {
ln.uploadCallbacks.Delete(fileID)
}
// CloseInfo carries the WebSocket close code and reason.
type CloseInfo struct {
Code int
@@ -230,6 +245,8 @@ func (ln *Listener) handleFrame(ctx context.Context, data []byte) {
ln.handleUserMessages(ctx, envelope.Data, envelope.Encrypt)
case "1_521_0":
ln.handleGroupMessages(ctx, envelope.Data, envelope.Encrypt)
case "1_601_0":
ln.handleControlEvents(ctx, envelope.Data, envelope.Encrypt)
case "1_3000_0":
slog.Warn("zalo_personal: duplicate connection detected, closing")
ln.mu.RLock()
@@ -12,6 +12,7 @@ import (
"log/slog"
"net/url"
"slices"
"strconv"
"time"
"unicode/utf8"
)
@@ -86,6 +87,62 @@ func (ln *Listener) handleGroupMessages(ctx context.Context, data string, encTyp
}
}
// --- Control events (cmd=601) ---
func (ln *Listener) handleControlEvents(ctx context.Context, data string, encType uint) {
ln.mu.RLock()
ck := ln.cipherKey
ln.mu.RUnlock()
payload, err := ln.decryptEventData(data, encType, ck)
if err != nil {
emit(ctx, ln.errorCh, fmt.Errorf("zalo_personal: decrypt control event: %w", err))
return
}
var envelope struct {
Data struct {
Controls []struct {
Content struct {
ActType string `json:"act_type"`
FileID any `json:"fileId"` // can be string or number
Data struct {
URL string `json:"url"`
} `json:"data"`
} `json:"content"`
} `json:"controls"`
} `json:"data"`
}
if err := json.Unmarshal(payload, &envelope); err != nil {
emit(ctx, ln.errorCh, fmt.Errorf("zalo_personal: parse control event: %w", err))
return
}
for _, ctrl := range envelope.Data.Controls {
if ctrl.Content.ActType != "file_done" {
continue
}
if ctrl.Content.FileID == nil {
continue
}
// FileID can be string or number (float64 from JSON).
// fmt.Sprint on large float64 produces scientific notation (3.15e+11),
// but callbacks are registered with the plain decimal string.
fileID := anyToDecimalString(ctrl.Content.FileID)
fileURL := ctrl.Content.Data.URL
slog.Debug("zalo_personal file upload completed", "file_id", fileID, "url_len", len(fileURL))
if val, ok := ln.uploadCallbacks.LoadAndDelete(fileID); ok {
ch := val.(chan string)
select {
case ch <- fileURL:
default:
}
}
}
}
// --- Decryption ---
func (ln *Listener) decryptEventData(data string, encType uint, cipherKey string) ([]byte, error) {
@@ -309,6 +366,22 @@ func (ln *Listener) emitClosed(ci CloseInfo) {
}
}
// anyToDecimalString converts a value (string, float64, json.Number, etc.) to a
// plain decimal string. Needed because JSON numbers unmarshalled into `any`
// become float64, and fmt.Sprint produces scientific notation for large values.
func anyToDecimalString(v any) string {
switch val := v.(type) {
case string:
return val
case float64:
return strconv.FormatFloat(val, 'f', -1, 64)
case json.Number:
return val.String()
default:
return fmt.Sprint(v)
}
}
// --- Helpers ---
func buildWSURL(sess *Session, base string) string {
@@ -0,0 +1,50 @@
package protocol
import (
"encoding/json"
"testing"
)
func TestAnyToDecimalString(t *testing.T) {
tests := []struct {
name string
input any
want string
}{
{"string", "315077459047", "315077459047"},
{"float64 large", float64(315077459047), "315077459047"},
{"float64 small", float64(42), "42"},
{"float64 zero", float64(0), "0"},
{"json.Number", json.Number("315077459047"), "315077459047"},
{"int", 42, "42"},
{"nil", nil, "<nil>"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := anyToDecimalString(tt.input)
if got != tt.want {
t.Errorf("anyToDecimalString(%v) = %q, want %q", tt.input, got, tt.want)
}
})
}
}
// TestAnyToDecimalString_JSONUnmarshal verifies the fix for the actual bug:
// JSON number unmarshalled into `any` becomes float64, which fmt.Sprint
// renders in scientific notation for large values.
func TestAnyToDecimalString_JSONUnmarshal(t *testing.T) {
// Simulate what happens when Zalo sends fileId as a number in JSON
raw := `{"fileId": 315077459047}`
var parsed struct {
FileID any `json:"fileId"`
}
if err := json.Unmarshal([]byte(raw), &parsed); err != nil {
t.Fatal(err)
}
got := anyToDecimalString(parsed.FileID)
if got != "315077459047" {
t.Errorf("JSON number → anyToDecimalString = %q, want %q", got, "315077459047")
}
}
@@ -1,11 +1,24 @@
package protocol
import (
"bytes"
"context"
"crypto/md5"
"crypto/rand"
"encoding/base64"
"encoding/hex"
"encoding/json"
"fmt"
"image"
_ "image/jpeg"
_ "image/png"
"io"
"mime/multipart"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"time"
)
@@ -168,6 +181,477 @@ func getServiceURL(sess *Session, service string) string {
return urls[0]
}
// FlexBool handles JSON fields that may be bool (true/false) or number (0/1).
type FlexBool bool
func (b *FlexBool) UnmarshalJSON(data []byte) error {
switch string(data) {
case "true", "1":
*b = true
default:
*b = false
}
return nil
}
// ImageUploadResult holds the response from uploading an image to Zalo's file service.
type ImageUploadResult struct {
NormalURL string `json:"normalUrl"`
HDUrl string `json:"hdUrl"`
ThumbURL string `json:"thumbUrl"`
PhotoID json.Number `json:"photoId"` // Zalo may return string or number
ClientFileID json.Number `json:"clientFileId"` // Zalo may return string or number
ChunkID int `json:"chunkId"`
Finished FlexBool `json:"finished"` // Zalo returns bool or int depending on endpoint
// Set by caller (not from API response).
Width int `json:"-"`
Height int `json:"-"`
TotalSize int `json:"-"`
}
// UploadImage uploads an image file to Zalo's file service.
// Returns upload metadata needed for SendImage.
func UploadImage(ctx context.Context, sess *Session, threadID string, threadType ThreadType, filePath string) (*ImageUploadResult, error) {
fileURL := getServiceURL(sess, "file")
if fileURL == "" {
return nil, fmt.Errorf("zalo_personal: no file service URL")
}
data, err := os.ReadFile(filePath)
if err != nil {
return nil, fmt.Errorf("zalo_personal: read image: %w", err)
}
fileName := filepath.Base(filePath)
totalSize := len(data)
width, height := imageDimensions(data)
uploadParams := map[string]any{
"totalChunk": 1,
"fileName": fileName,
"clientId": time.Now().UnixMilli(),
"totalSize": totalSize,
"imei": sess.IMEI,
"isE2EE": 0,
"jxl": 0,
"chunkId": 1,
}
if threadType == ThreadTypeGroup {
uploadParams["grid"] = threadID
} else {
uploadParams["toid"] = threadID
}
encParams, err := encryptPayload(sess, uploadParams)
if err != nil {
return nil, fmt.Errorf("zalo_personal: encrypt upload params: %w", err)
}
// Build URL: file_service/api/{message|group}/photo_original/upload
pathPrefix := "/api/message/"
typeParam := "2"
if threadType == ThreadTypeGroup {
pathPrefix = "/api/group/"
typeParam = "11"
}
uploadURL := makeURL(sess, fileURL+pathPrefix+"photo_original/upload", map[string]any{
"type": typeParam,
"params": encParams,
}, true)
// Build multipart body
body, contentType, err := buildMultipartBody("chunkContent", fileName, data)
if err != nil {
return nil, fmt.Errorf("zalo_personal: build multipart: %w", err)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, uploadURL, body)
if err != nil {
return nil, err
}
setDefaultHeaders(req, sess)
req.Header.Set("Content-Type", contentType)
resp, err := sess.Client.Do(req)
if err != nil {
return nil, fmt.Errorf("zalo_personal: upload image: %w", err)
}
defer resp.Body.Close()
// Response: {"error_code":0, "data":"<encrypted>"}
var envelope Response[*string]
if err := readJSON(resp, &envelope); err != nil {
return nil, fmt.Errorf("zalo_personal: parse upload response: %w", err)
}
if envelope.ErrorCode != 0 {
return nil, fmt.Errorf("zalo_personal: upload error code %d", envelope.ErrorCode)
}
if envelope.Data == nil {
return nil, fmt.Errorf("zalo_personal: empty upload response")
}
plain, err := decryptDataField(sess, *envelope.Data)
if err != nil {
return nil, fmt.Errorf("zalo_personal: decrypt upload response: %w", err)
}
var result ImageUploadResult
if err := json.Unmarshal(plain, &result); err != nil {
return nil, fmt.Errorf("zalo_personal: parse upload result: %w", err)
}
result.Width = width
result.Height = height
result.TotalSize = totalSize
return &result, nil
}
// SendImage sends a previously uploaded image as a message.
func SendImage(ctx context.Context, sess *Session, threadID string, threadType ThreadType, upload *ImageUploadResult, caption string) (string, error) {
fileURL := getServiceURL(sess, "file")
if fileURL == "" {
return "", fmt.Errorf("zalo_personal: no file service URL")
}
params := map[string]any{
"photoId": upload.PhotoID,
"clientId": strconv.FormatInt(time.Now().UnixMilli(), 10),
"desc": caption,
"width": upload.Width,
"height": upload.Height,
"rawUrl": upload.NormalURL,
"hdUrl": upload.HDUrl,
"thumbUrl": upload.ThumbURL,
"hdSize": strconv.Itoa(upload.TotalSize),
"zsource": -1,
"ttl": 0,
"jcp": `{"convertible":"jxl"}`,
}
if threadType == ThreadTypeGroup {
params["grid"] = threadID
params["oriUrl"] = upload.NormalURL
} else {
params["toid"] = threadID
params["normalUrl"] = upload.NormalURL
}
encData, err := encryptPayload(sess, params)
if err != nil {
return "", fmt.Errorf("zalo_personal: encrypt image send params: %w", err)
}
pathPrefix := "/api/message/"
if threadType == ThreadTypeGroup {
pathPrefix = "/api/group/"
}
sendURL := makeURL(sess, fileURL+pathPrefix+"photo_original/send", map[string]any{"nretry": 0}, true)
form := buildFormBody(map[string]string{"params": encData})
req, err := http.NewRequestWithContext(ctx, http.MethodPost, sendURL, form)
if err != nil {
return "", err
}
setDefaultHeaders(req, sess)
resp, err := sess.Client.Do(req)
if err != nil {
return "", fmt.Errorf("zalo_personal: send image: %w", err)
}
defer resp.Body.Close()
var envelope Response[*string]
if err := readJSON(resp, &envelope); err != nil {
return "", fmt.Errorf("zalo_personal: parse image send response: %w", err)
}
if envelope.ErrorCode != 0 {
return "", fmt.Errorf("zalo_personal: image send error code %d", envelope.ErrorCode)
}
if envelope.Data == nil {
return "", nil
}
plain, err := decryptDataField(sess, *envelope.Data)
if err != nil {
return "", fmt.Errorf("zalo_personal: decrypt image send response: %w", err)
}
var result struct {
MsgID json.Number `json:"msgId"`
}
if err := json.Unmarshal(plain, &result); err != nil {
return "", fmt.Errorf("zalo_personal: parse image send result: %w", err)
}
return result.MsgID.String(), nil
}
// buildMultipartBody creates a multipart/form-data body with a single file field.
func buildMultipartBody(fieldName, fileName string, data []byte) (io.Reader, string, error) {
var buf bytes.Buffer
w := multipart.NewWriter(&buf)
boundary := "----GoClaw" + randomBoundary()
if err := w.SetBoundary(boundary); err != nil {
return nil, "", err
}
part, err := w.CreateFormFile(fieldName, fileName)
if err != nil {
return nil, "", err
}
if _, err := part.Write(data); err != nil {
return nil, "", err
}
if err := w.Close(); err != nil {
return nil, "", err
}
return &buf, w.FormDataContentType(), nil
}
func randomBoundary() string {
b := make([]byte, 8)
_, _ = rand.Read(b)
return hex.EncodeToString(b)
}
// imageDimensions extracts width and height from an image (PNG, JPEG, etc.).
// Returns (0, 0) if the format is unrecognized.
func imageDimensions(data []byte) (int, int) {
cfg, _, err := image.DecodeConfig(bytes.NewReader(data))
if err != nil {
return 0, 0
}
return cfg.Width, cfg.Height
}
// IsImageFile returns true if the file extension is a supported image type.
func IsImageFile(filePath string) bool {
ext := strings.ToLower(filepath.Ext(filePath))
switch ext {
case ".jpg", ".jpeg", ".png", ".webp":
return true
}
return false
}
// FileUploadResult holds the response from uploading a file to Zalo's file service.
type FileUploadResult struct {
FileID string `json:"fileId"`
FileURL string `json:"fileUrl"` // populated from WS callback
ClientFileID json.Number `json:"clientFileId"` // Zalo may return string or number
ChunkID int `json:"chunkId"`
Finished int `json:"finished"`
// Set by caller.
TotalSize int `json:"-"`
FileName string `json:"-"`
Checksum string `json:"-"` // MD5 hex
}
// UploadFile uploads a non-image file to Zalo's file service.
// The upload response only contains fileId; the fileUrl arrives via WebSocket callback.
// The caller must provide the Listener so we can register a callback for the fileUrl.
func UploadFile(ctx context.Context, sess *Session, ln *Listener, threadID string, threadType ThreadType, filePath string) (*FileUploadResult, error) {
fileURL := getServiceURL(sess, "file")
if fileURL == "" {
return nil, fmt.Errorf("zalo_personal: no file service URL")
}
data, err := os.ReadFile(filePath)
if err != nil {
return nil, fmt.Errorf("zalo_personal: read file: %w", err)
}
fileName := filepath.Base(filePath)
totalSize := len(data)
clientID := time.Now().UnixMilli()
uploadParams := map[string]any{
"totalChunk": 1,
"fileName": fileName,
"clientId": clientID,
"totalSize": totalSize,
"imei": sess.IMEI,
"isE2EE": 0,
"jxl": 0,
"chunkId": 1,
}
if threadType == ThreadTypeGroup {
uploadParams["grid"] = threadID
} else {
uploadParams["toid"] = threadID
}
encParams, err := encryptPayload(sess, uploadParams)
if err != nil {
return nil, fmt.Errorf("zalo_personal: encrypt file upload params: %w", err)
}
pathPrefix := "/api/message/"
typeParam := "2"
if threadType == ThreadTypeGroup {
pathPrefix = "/api/group/"
typeParam = "11"
}
uploadURL := makeURL(sess, fileURL+pathPrefix+"asyncfile/upload", map[string]any{
"type": typeParam,
"params": encParams,
}, true)
body, contentType, err := buildMultipartBody("chunkContent", fileName, data)
if err != nil {
return nil, fmt.Errorf("zalo_personal: build multipart: %w", err)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, uploadURL, body)
if err != nil {
return nil, err
}
setDefaultHeaders(req, sess)
req.Header.Set("Content-Type", contentType)
resp, err := sess.Client.Do(req)
if err != nil {
return nil, fmt.Errorf("zalo_personal: upload file: %w", err)
}
defer resp.Body.Close()
var envelope Response[*string]
if err := readJSON(resp, &envelope); err != nil {
return nil, fmt.Errorf("zalo_personal: parse file upload response: %w", err)
}
if envelope.ErrorCode != 0 {
return nil, fmt.Errorf("zalo_personal: file upload error code %d", envelope.ErrorCode)
}
if envelope.Data == nil {
return nil, fmt.Errorf("zalo_personal: empty file upload response")
}
plain, err := decryptDataField(sess, *envelope.Data)
if err != nil {
return nil, fmt.Errorf("zalo_personal: decrypt file upload response: %w", err)
}
var result FileUploadResult
if err := json.Unmarshal(plain, &result); err != nil {
return nil, fmt.Errorf("zalo_personal: parse file upload result: %w", err)
}
result.TotalSize = totalSize
result.FileName = fileName
// Compute MD5 checksum
h := md5Hash(data)
result.Checksum = h
// Wait for fileUrl from WebSocket callback
if ln != nil && result.FileID != "" && result.FileID != "-1" {
urlCh := ln.RegisterUploadCallback(result.FileID)
select {
case fileURL := <-urlCh:
result.FileURL = fileURL
case <-time.After(30 * time.Second):
ln.CancelUploadCallback(result.FileID)
return nil, fmt.Errorf("zalo_personal: timeout waiting for file upload callback (fileId=%s)", result.FileID)
case <-ctx.Done():
ln.CancelUploadCallback(result.FileID)
return nil, ctx.Err()
}
}
return &result, nil
}
// SendFile sends a previously uploaded file as a message.
func SendFile(ctx context.Context, sess *Session, threadID string, threadType ThreadType, upload *FileUploadResult) (string, error) {
fileURL := getServiceURL(sess, "file")
if fileURL == "" {
return "", fmt.Errorf("zalo_personal: no file service URL")
}
ext := strings.TrimPrefix(filepath.Ext(upload.FileName), ".")
params := map[string]any{
"fileId": upload.FileID,
"checksum": upload.Checksum,
"checksumSha": "",
"extention": ext, // Zalo typo: "extention" not "extension"
"totalSize": upload.TotalSize,
"fileName": upload.FileName,
"clientId": upload.ClientFileID.String(),
"fType": 1,
"fileCount": 0,
"fdata": "{}",
"fileUrl": upload.FileURL,
"zsource": -1,
"ttl": 0,
}
if threadType == ThreadTypeGroup {
params["grid"] = threadID
} else {
params["toid"] = threadID
}
encData, err := encryptPayload(sess, params)
if err != nil {
return "", fmt.Errorf("zalo_personal: encrypt file send params: %w", err)
}
pathPrefix := "/api/message/"
if threadType == ThreadTypeGroup {
pathPrefix = "/api/group/"
}
sendURL := makeURL(sess, fileURL+pathPrefix+"asyncfile/msg", map[string]any{"nretry": 0}, true)
form := buildFormBody(map[string]string{"params": encData})
req, err := http.NewRequestWithContext(ctx, http.MethodPost, sendURL, form)
if err != nil {
return "", err
}
setDefaultHeaders(req, sess)
resp, err := sess.Client.Do(req)
if err != nil {
return "", fmt.Errorf("zalo_personal: send file: %w", err)
}
defer resp.Body.Close()
var respEnvelope Response[*string]
if err := readJSON(resp, &respEnvelope); err != nil {
return "", fmt.Errorf("zalo_personal: parse file send response: %w", err)
}
if respEnvelope.ErrorCode != 0 {
return "", fmt.Errorf("zalo_personal: file send error code %d", respEnvelope.ErrorCode)
}
if respEnvelope.Data == nil {
return "", nil
}
plain, err := decryptDataField(sess, *respEnvelope.Data)
if err != nil {
return "", fmt.Errorf("zalo_personal: decrypt file send response: %w", err)
}
var result struct {
MsgID json.Number `json:"msgId"`
}
if err := json.Unmarshal(plain, &result); err != nil {
return "", fmt.Errorf("zalo_personal: parse file send result: %w", err)
}
return result.MsgID.String(), nil
}
// md5Hash returns the MD5 hex digest. Required by Zalo's file upload API checksum field.
func md5Hash(data []byte) string {
h := md5.Sum(data)
return hex.EncodeToString(h[:])
}
// encryptPayload encrypts a JSON payload with the session's secret key via AES-CBC.
func encryptPayload(sess *Session, payload map[string]any) (string, error) {
blob, err := json.Marshal(payload)
@@ -0,0 +1,135 @@
package protocol
import (
"bytes"
"image"
"image/color"
"image/jpeg"
"image/png"
"io"
"testing"
)
func TestImageDimensions(t *testing.T) {
tests := []struct {
name string
data []byte
wantW int
wantH int
}{
{
name: "valid PNG 200x100",
data: makePNG(200, 100),
wantW: 200,
wantH: 100,
},
{
name: "valid PNG 1x1",
data: makePNG(1, 1),
wantW: 1,
wantH: 1,
},
{
name: "valid JPEG 50x30",
data: makeJPEG(50, 30),
wantW: 50,
wantH: 30,
},
{
name: "not an image",
data: []byte("hello world"),
wantW: 0,
wantH: 0,
},
{
name: "empty",
data: nil,
wantW: 0,
wantH: 0,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
w, h := imageDimensions(tt.data)
if w != tt.wantW || h != tt.wantH {
t.Errorf("imageDimensions() = (%d, %d), want (%d, %d)", w, h, tt.wantW, tt.wantH)
}
})
}
}
func makePNG(width, height int) []byte {
img := image.NewRGBA(image.Rect(0, 0, width, height))
img.Set(0, 0, color.RGBA{R: 255})
var buf bytes.Buffer
_ = png.Encode(&buf, img)
return buf.Bytes()
}
func makeJPEG(width, height int) []byte {
img := image.NewRGBA(image.Rect(0, 0, width, height))
img.Set(0, 0, color.RGBA{R: 255})
var buf bytes.Buffer
_ = jpeg.Encode(&buf, img, nil)
return buf.Bytes()
}
func TestIsImageFile(t *testing.T) {
tests := []struct {
path string
want bool
}{
{"photo.jpg", true},
{"photo.JPG", true},
{"photo.jpeg", true},
{"photo.png", true},
{"photo.webp", true},
{"photo.PNG", true},
{"file.md", false},
{"file.pdf", false},
{"file.gif", false},
{"file.mp4", false},
{"file.txt", false},
{"noext", false},
{"", false},
}
for _, tt := range tests {
t.Run(tt.path, func(t *testing.T) {
if got := IsImageFile(tt.path); got != tt.want {
t.Errorf("IsImageFile(%q) = %v, want %v", tt.path, got, tt.want)
}
})
}
}
func TestBuildMultipartBody(t *testing.T) {
data := []byte("hello world")
reader, contentType, err := buildMultipartBody("chunkContent", "test.png", data)
if err != nil {
t.Fatalf("buildMultipartBody() error: %v", err)
}
if contentType == "" {
t.Fatal("content type is empty")
}
if !bytes.Contains([]byte(contentType), []byte("multipart/form-data")) {
t.Errorf("content type %q does not contain multipart/form-data", contentType)
}
body, err := io.ReadAll(reader)
if err != nil {
t.Fatalf("read body error: %v", err)
}
if !bytes.Contains(body, data) {
t.Error("body does not contain file data")
}
if !bytes.Contains(body, []byte("chunkContent")) {
t.Error("body does not contain field name")
}
if !bytes.Contains(body, []byte("test.png")) {
t.Error("body does not contain filename")
}
}
+14
View File
@@ -33,6 +33,9 @@ type toolsInvokeRequest struct {
SessionKey string `json:"sessionKey,omitempty"`
AgentID string `json:"agentId,omitempty"`
DryRun bool `json:"dryRun,omitempty"`
Channel string `json:"channel,omitempty"` // tool context: channel name
ChatID string `json:"chatId,omitempty"` // tool context: chat ID
PeerKind string `json:"peerKind,omitempty"` // tool context: "direct" or "group"
}
func (h *ToolsInvokeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -97,6 +100,17 @@ func (h *ToolsInvokeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
}
// Inject tool context keys (channel, chatID, peerKind) for message routing.
if req.Channel != "" {
ctx = tools.WithToolChannel(ctx, req.Channel)
}
if req.ChatID != "" {
ctx = tools.WithToolChatID(ctx, req.ChatID)
}
if req.PeerKind != "" {
ctx = tools.WithToolPeerKind(ctx, req.PeerKind)
}
// Execute the tool
args := req.Args
if args == nil {
+53
View File
@@ -2,7 +2,11 @@ package tools
import (
"context"
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
"github.com/nextlevelbuilder/goclaw/internal/bus"
)
@@ -76,6 +80,11 @@ func (t *MessageTool) Execute(ctx context.Context, args map[string]interface{})
return ErrorResult("target chat ID is required (no current chat in context)")
}
// Handle MEDIA: prefix — send file as attachment instead of text.
if filePath, ok := parseMediaPath(message); ok {
return t.sendMedia(ctx, channel, target, filePath)
}
// Prefer direct channel sender (channels.Manager.SendToChannel)
if t.sender != nil {
if err := t.sender(ctx, channel, target, message); err != nil {
@@ -96,3 +105,47 @@ func (t *MessageTool) Execute(ctx context.Context, args map[string]interface{})
return ErrorResult("no channel sender or message bus available")
}
// sendMedia sends a file as a media attachment via the outbound message bus.
func (t *MessageTool) sendMedia(ctx context.Context, channel, target, filePath string) *Result {
if _, err := os.Stat(filePath); err != nil {
return ErrorResult(fmt.Sprintf("file not found: %s", filePath))
}
if t.msgBus == nil {
return ErrorResult("media sending requires message bus")
}
// Build metadata for group routing (Zalo needs group_id to choose group API).
var meta map[string]string
peerKind := ToolPeerKindFromCtx(ctx)
if peerKind == "group" {
meta = map[string]string{"group_id": target}
}
t.msgBus.PublishOutbound(bus.OutboundMessage{
Channel: channel,
ChatID: target,
Media: []bus.MediaAttachment{{URL: filePath}},
Metadata: meta,
})
out, _ := json.Marshal(map[string]string{
"status": "sent",
"channel": channel,
"target": target,
"media": filepath.Base(filePath),
})
return SilentResult(string(out))
}
// parseMediaPath extracts a file path from a "MEDIA:/path/to/file" string.
func parseMediaPath(s string) (string, bool) {
s = strings.TrimSpace(s)
if !strings.HasPrefix(s, "MEDIA:") {
return "", false
}
path := filepath.Clean(strings.TrimSpace(s[len("MEDIA:"):]))
if path == "" || path == "." {
return "", false
}
return path, true
}
+158
View File
@@ -0,0 +1,158 @@
// Manual E2E test for Zalo Personal file & image sending via Go protocol package.
// Requires real Zalo credentials — not run by CI.
//
// Usage:
//
// docker run --rm -v $(pwd):/src -w /src \
// -e ZALO_CREDS_FILE=/src/tests/zalo_e2e/creds.json \
// -e ZALO_GROUP_ID=<your-group-id> \
// golang:1.25-alpine go run ./tests/zalo_e2e/
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"os"
"time"
"github.com/nextlevelbuilder/goclaw/internal/channels/zalo/personal/protocol"
)
func main() {
credsFile := os.Getenv("ZALO_CREDS_FILE")
if credsFile == "" {
log.Fatal("ZALO_CREDS_FILE env var is required")
}
groupID := os.Getenv("ZALO_GROUP_ID")
if groupID == "" {
log.Fatal("ZALO_GROUP_ID env var is required")
}
log.Println("=== Zalo E2E Test ===")
log.Println("Creds:", credsFile)
log.Println("Group:", groupID)
// Load credentials
raw, err := os.ReadFile(credsFile)
if err != nil {
log.Fatalf("read creds: %v", err)
}
var creds protocol.Credentials
if err := json.Unmarshal(raw, &creds); err != nil {
log.Fatalf("parse creds: %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
// Login
log.Println("\n--- Step 1: Login ---")
sess := protocol.NewSession()
if err := protocol.LoginWithCredentials(ctx, sess, creds); err != nil {
log.Fatalf("login: %v", err)
}
log.Println("UID:", sess.UID)
// Start listener (needed for file upload WS callback)
log.Println("\n--- Step 2: Start Listener ---")
ln, err := protocol.NewListener(sess)
if err != nil {
log.Fatalf("new listener: %v", err)
}
if err := ln.Start(ctx); err != nil {
log.Fatalf("start listener: %v", err)
}
defer ln.Stop()
// Wait for cipher key
log.Println("Waiting for cipher key (max 15s)...")
time.Sleep(3 * time.Second)
log.Println("Listener ready")
// Test 1: Send text message
log.Println("\n--- Test 1: Send Text Message ---")
msgID, err := protocol.SendMessage(ctx, sess, groupID, protocol.ThreadTypeGroup,
fmt.Sprintf("GoClaw Go E2E test - %s", time.Now().Format(time.RFC3339)))
if err != nil {
log.Fatalf("send text: %v", err)
}
log.Println("OK, msgId:", msgID)
// Test 2: Upload & send image
log.Println("\n--- Test 2: Upload & Send Image ---")
imgPath := createTestImage()
defer os.Remove(imgPath)
upload, err := protocol.UploadImage(ctx, sess, groupID, protocol.ThreadTypeGroup, imgPath)
if err != nil {
log.Fatalf("upload image: %v", err)
}
log.Printf("Upload OK: photoId=%d normalUrl=%s", upload.PhotoID, truncate(upload.NormalURL, 60))
imgMsgID, err := protocol.SendImage(ctx, sess, groupID, protocol.ThreadTypeGroup, upload, "Go E2E test image")
if err != nil {
log.Fatalf("send image: %v", err)
}
log.Println("OK, msgId:", imgMsgID)
// Test 3: Upload & send file (requires WS callback)
log.Println("\n--- Test 3: Upload & Send File ---")
filePath := createTestFile()
defer os.Remove(filePath)
fileUpload, err := protocol.UploadFile(ctx, sess, ln, groupID, protocol.ThreadTypeGroup, filePath)
if err != nil {
log.Fatalf("upload file: %v", err)
}
log.Printf("Upload OK: fileId=%s fileUrl=%s", fileUpload.FileID, truncate(fileUpload.FileURL, 60))
fileMsgID, err := protocol.SendFile(ctx, sess, groupID, protocol.ThreadTypeGroup, fileUpload)
if err != nil {
log.Fatalf("send file: %v", err)
}
log.Println("OK, msgId:", fileMsgID)
log.Println("\n=== ALL TESTS PASSED ===")
}
func createTestImage() string {
// Minimal valid PNG: 1x1 red pixel
png := []byte{
0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, // PNG signature
0x00, 0x00, 0x00, 0x0d, 0x49, 0x48, 0x44, 0x52, // IHDR chunk
0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x01, // 1x1
0x08, 0x02, 0x00, 0x00, 0x00, 0x90, 0x77, 0x53, 0xde, // 8-bit RGB
0x00, 0x00, 0x00, 0x0c, 0x49, 0x44, 0x41, 0x54, // IDAT chunk
0x08, 0xd7, 0x63, 0xf8, 0xcf, 0xc0, 0x00, 0x00,
0x00, 0x02, 0x00, 0x01, 0xe2, 0x21, 0xbc, 0x33,
0x00, 0x00, 0x00, 0x00, 0x49, 0x45, 0x4e, 0x44, // IEND chunk
0xae, 0x42, 0x60, 0x82,
}
f, err := os.CreateTemp("", "goclaw_e2e_*.png")
if err != nil {
log.Fatal(err)
}
f.Write(png)
f.Close()
return f.Name()
}
func createTestFile() string {
content := fmt.Sprintf("GoClaw E2E Test File\nTimestamp: %s\nThis file tests the file upload pipeline.\n", time.Now().Format(time.RFC3339))
f, err := os.CreateTemp("", "goclaw_e2e_*.txt")
if err != nil {
log.Fatal(err)
}
f.WriteString(content)
f.Close()
return f.Name()
}
func truncate(s string, n int) string {
if len(s) <= n {
return s
}
return s[:n] + "..."
}