diff --git a/.gitignore b/.gitignore index 32a3b6c1..179eee25 100644 --- a/.gitignore +++ b/.gitignore @@ -38,3 +38,6 @@ deploy-*.sh docs/technical-diaries plans release-manifest.json + +# Test credentials (never commit) +tests/**/creds.json diff --git a/docker-compose.otel.yml b/docker-compose.otel.yml index 20e41155..03c960ab 100644 --- a/docker-compose.otel.yml +++ b/docker-compose.otel.yml @@ -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 diff --git a/internal/bootstrap/seed_store.go b/internal/bootstrap/seed_store.go index e47f0546..39729992 100644 --- a/internal/bootstrap/seed_store.go +++ b/internal/bootstrap/seed_store.go @@ -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. diff --git a/internal/channels/zalo/personal/channel.go b/internal/channels/zalo/personal/channel.go index 5f0945bb..54e2fe31 100644 --- a/internal/channels/zalo/personal/channel.go +++ b/internal/channels/zalo/personal/channel.go @@ -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) { diff --git a/internal/channels/zalo/personal/protocol/listener.go b/internal/channels/zalo/personal/protocol/listener.go index f86d630c..79a1cebb 100644 --- a/internal/channels/zalo/personal/protocol/listener.go +++ b/internal/channels/zalo/personal/protocol/listener.go @@ -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() diff --git a/internal/channels/zalo/personal/protocol/listener_handlers.go b/internal/channels/zalo/personal/protocol/listener_handlers.go index 786f0028..d066779c 100644 --- a/internal/channels/zalo/personal/protocol/listener_handlers.go +++ b/internal/channels/zalo/personal/protocol/listener_handlers.go @@ -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 { diff --git a/internal/channels/zalo/personal/protocol/listener_handlers_test.go b/internal/channels/zalo/personal/protocol/listener_handlers_test.go new file mode 100644 index 00000000..396bb57d --- /dev/null +++ b/internal/channels/zalo/personal/protocol/listener_handlers_test.go @@ -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, ""}, + } + + 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") + } +} diff --git a/internal/channels/zalo/personal/protocol/send.go b/internal/channels/zalo/personal/protocol/send.go index 0a507f2f..8c133bb9 100644 --- a/internal/channels/zalo/personal/protocol/send.go +++ b/internal/channels/zalo/personal/protocol/send.go @@ -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":""} + 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) diff --git a/internal/channels/zalo/personal/protocol/send_test.go b/internal/channels/zalo/personal/protocol/send_test.go new file mode 100644 index 00000000..a7a870fa --- /dev/null +++ b/internal/channels/zalo/personal/protocol/send_test.go @@ -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") + } +} diff --git a/internal/http/tools_invoke.go b/internal/http/tools_invoke.go index 76a57d07..a9530612 100644 --- a/internal/http/tools_invoke.go +++ b/internal/http/tools_invoke.go @@ -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 { diff --git a/internal/tools/message.go b/internal/tools/message.go index 58e6648c..8b4dc722 100644 --- a/internal/tools/message.go +++ b/internal/tools/message.go @@ -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 +} diff --git a/tests/zalo_e2e/main.go b/tests/zalo_e2e/main.go new file mode 100644 index 00000000..263c78fb --- /dev/null +++ b/tests/zalo_e2e/main.go @@ -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= \ +// 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] + "..." +}