merge main

backend/cmd/consumer.go

@@ -14,7 +14,7 @@ import (
 	"github.com/coze-dev/coze-loop/backend/pkg/conf"
 )
 
-func mustInitConsumerWorkers(
+func MustInitConsumerWorkers(
 	cfactory conf.IConfigLoaderFactory,
 	experimentApplication exptapp.IExperimentApplication,
 	datasetApplication dataapp.IJobRunMsgHandler,
@@ -45,4 +45,4 @@ func mustInitConsumerWorkers(
 	res = append(res, workers...)
 
 	return res
-}
+}

backend/cmd/main.go

@@ -61,7 +61,8 @@ func main() {
 		panic(err)
 	}
 
-	if err := registry.NewConsumerRegistry(c.mqFactory).Register(mustInitConsumerWorkers(c.cfgFactory, handler, handler, handler)).StartAll(ctx); err != nil {
+	consumerRegistry := registry.NewConsumerRegistry(c.mqFactory).Register(MustInitConsumerWorkers(c.cfgFactory, handler, handler, handler))
+	if err := consumerRegistry.StartAll(ctx); err != nil {
 		panic(err)
 	}
 

backend/conf/infrastructure.yaml

@@ -0,0 +1,34 @@
+# Infrastructure configuration for coze-loop backend
+# This is a minimal configuration for testing
+
+# Database configuration - use SQLite for testing
+sqlite:
+  path: "/tmp/coze_loop.db"
+
+# Redis configuration - disabled for testing
+redis:
+  enabled: false
+
+# ClickHouse configuration - disabled for testing
+clickhouse:
+  enabled: false
+
+# Message Queue configuration - disabled for testing
+mq:
+  enabled: false
+
+# File server configuration
+fileserver:
+  type: "local"
+  local:
+    root: "/tmp/coze-loop-files"
+
+# Logging configuration
+log:
+  level: "info"
+  format: "json"
+
+# Server configuration
+server:
+  port: 8080
+  host: "0.0.0.0"

backend/go.mod

@@ -1,8 +1,8 @@
 module github.com/coze-dev/coze-loop/backend
 
-go 1.24.0
+go 1.24.1
 
-toolchain go1.24.4
+toolchain go1.24.6
 
 replace github.com/apache/thrift => github.com/apache/thrift v0.13.0
 
@@ -33,7 +33,7 @@ require (
 	github.com/cloudwego/eino-ext/components/model/qwen v0.0.0-20250520101807-b2008771903a
 	github.com/cloudwego/eino-ext/libs/acl/openai v0.0.0-20250519084852-38fafa73d9ea
 	github.com/cloudwego/gopkg v0.1.4
-	github.com/cloudwego/hertz v0.9.7
+	github.com/cloudwego/hertz v0.10.1
 	github.com/cloudwego/kitex v0.13.1
 	github.com/coocood/freecache v1.2.4
 	github.com/coreos/go-semver v0.3.0
@@ -101,6 +101,8 @@ require (
 
 require github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.1 // indirect
 
+require github.com/brianvoe/gofakeit/v6 v6.28.0
+
 require (
 	cloud.google.com/go v0.116.0 // indirect
 	cloud.google.com/go/ai v0.8.0 // indirect

backend/go.sum

@@ -152,6 +152,8 @@ github.com/bitly/go-simplejson v0.5.0/go.mod h1:cXHtHw4XUPsvGaxgjIAn8PhEWG9NfngE
 github.com/bluele/gcache v0.0.2 h1:WcbfdXICg7G/DGBh1PFfcirkWOQV+v077yF1pSy3DGw=
 github.com/bluele/gcache v0.0.2/go.mod h1:m15KV+ECjptwSPxKhOhQoAFQVtUFjTVkc3H8o0t/fp0=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
+github.com/brianvoe/gofakeit/v6 v6.28.0 h1:Xib46XXuQfmlLS2EXRuJpqcw8St6qSZz75OUo0tgAW4=
+github.com/brianvoe/gofakeit/v6 v6.28.0/go.mod h1:Xj58BMSnFqcn/fAQeSK+/PLtC5kSb7FJIq4JyGa8vEs=
 github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
 github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
@@ -227,8 +229,8 @@ github.com/cloudwego/frugal v0.2.5 h1:zRICkWpBCQ6TY4QmRf+uINzcHbv7ogCHOM8h7ltPRz
 github.com/cloudwego/frugal v0.2.5/go.mod h1:nC1U47gswLRiaxv6dybrhZvsDGCfQP9RGiiWC73CnoI=
 github.com/cloudwego/gopkg v0.1.4 h1:EoQiCG4sTonTPHxOGE0VlQs+sQR+Hsi2uN0qqwu8O50=
 github.com/cloudwego/gopkg v0.1.4/go.mod h1:FQuXsRWRsSqJLsMVd5SYzp8/Z1y5gXKnVvRrWUOsCMI=
-github.com/cloudwego/hertz v0.9.7 h1:tAVaiO+vTf+ZkQhvNhKbDJ0hmC4oJ7bzwDi1KhvhHy4=
-github.com/cloudwego/hertz v0.9.7/go.mod h1:t6d7NcoQxPmETvzPMMIVPHMn5C5QzpqIiFsaavoLJYQ=
+github.com/cloudwego/hertz v0.10.1 h1:gTM2JIGO7vmRoaDz71GctyoUE19pXGuznFX55HjGs1g=
+github.com/cloudwego/hertz v0.10.1/go.mod h1:0sofikwk5YcHCerClgCzcaoamY61JiRwR5G0mAUo+Y0=
 github.com/cloudwego/iasm v0.2.0 h1:1KNIy1I1H9hNNFEEH3DVnI4UujN+1zjpuk6gwHLTssg=
 github.com/cloudwego/iasm v0.2.0/go.mod h1:8rXZaNYT2n95jn+zTI1sDr+IgcD2GVs0nlbbQPiEFhY=
 github.com/cloudwego/kitex v0.13.1 h1:oPJS/hy9gvo0rlfQmJAKJj8F4PMLG74IYzpaPlCRgg8=

backend/infra/http/client.go

@@ -0,0 +1,108 @@
+// Copyright (c) 2025 coze-dev Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package http
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+)
+
+// HTTPClient HTTP客户端实现
+type HTTPClient struct {
+	client *http.Client
+}
+
+// NewHTTPClient 创建HTTP客户端实例
+func NewHTTPClient() IClient {
+	return &HTTPClient{
+		client: &http.Client{
+			Timeout: 30 * time.Second,
+		},
+	}
+}
+
+// DoHTTPRequest 执行HTTP请求
+func (c *HTTPClient) DoHTTPRequest(ctx context.Context, requestParam *RequestParam) error {
+	if requestParam == nil {
+		return fmt.Errorf("request param is nil")
+	}
+
+	// 设置超时
+	if requestParam.Timeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, requestParam.Timeout)
+		defer cancel()
+	}
+
+	// 序列化请求体
+	var body io.Reader
+	if requestParam.Body != nil {
+		// 如果Body已经是io.Reader类型，直接使用
+		if reader, ok := requestParam.Body.(io.Reader); ok {
+			body = reader
+		} else {
+			// 否则进行JSON序列化
+			bodyBytes, err := json.Marshal(requestParam.Body)
+			if err != nil {
+				return fmt.Errorf("failed to marshal request body: %w", err)
+			}
+			body = bytes.NewReader(bodyBytes)
+		}
+	}
+
+	// 创建HTTP请求
+	req, err := http.NewRequestWithContext(ctx, requestParam.Method, requestParam.RequestURI, body)
+	if err != nil {
+		return fmt.Errorf("failed to create request: %w", err)
+	}
+
+	// 设置请求头
+	if requestParam.Header != nil {
+		for key, value := range requestParam.Header {
+			req.Header.Set(key, value)
+		}
+	}
+
+	// 如果没有设置Content-Type且有body，设置为application/json
+	if requestParam.Body != nil && req.Header.Get("Content-Type") == "" {
+		// 如果Body是io.Reader类型，默认设置为text/plain
+		if _, ok := requestParam.Body.(io.Reader); ok {
+			req.Header.Set("Content-Type", "text/plain")
+		} else {
+			req.Header.Set("Content-Type", "application/json")
+		}
+	}
+
+	// 发送请求
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("failed to send request: %w", err)
+	}
+	defer resp.Body.Close()
+
+	// 检查HTTP状态码
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		bodyBytes, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("HTTP request failed with status %d: %s", resp.StatusCode, string(bodyBytes))
+	}
+
+	// 解析响应体
+	if requestParam.Response != nil {
+		bodyBytes, err := io.ReadAll(resp.Body)
+		if err != nil {
+			return fmt.Errorf("failed to read response body: %w", err)
+		}
+
+		if err := json.Unmarshal(bodyBytes, requestParam.Response); err != nil {
+			return fmt.Errorf("failed to unmarshal response body: %w", err)
+		}
+	}
+
+	return nil
+}