Merge pull request 'feat: add devtodev event report' (#2) from feat/devtodev-report into test

Reviewed-on: #2

.gitignore

@@ -1,3 +1,5 @@
 # 忽略日志目录和文件
 log/
 *.log.*
+.idea/
+.DS_Store

api/v1/sqs/aws-sqs.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"runtime/debug"
 	"strings"
 	"sync"
 	"syscall"
@@ -85,6 +86,16 @@ func ProcessSqsMessage() {
 				wg.Add(1)
 				go func(m *sqs.Message) {
 					defer func() {
+						if r := recover(); r != nil {
+							msgID := ""
+							if m != nil && m.MessageId != nil {
+								msgID = *m.MessageId
+							}
+							global.GVA_LOG.Error("SQS worker panic recovered",
+								zap.String("msgId", msgID),
+								zap.Any("panic", r),
+								zap.ByteString("stack", debug.Stack()))
+						}
 						<-semaphore // 释放 semaphore
 						wg.Done()
 					}()
@@ -278,6 +289,8 @@ func processMessage(svc *sqs.SQS, queueURL string, msg *sqs.Message) {
 			if err != nil {
 				processErr = err
 				global.GVA_LOG.Error("Save SqsUserBehaviorLog error", zap.Error(err))
+			} else {
+				account.ReportUserBehaviorEvent(context.Background(), sqsMessage.Action, req)
 			}
 		}
 	} else if sqsMessage.Action == awssqs.SqsActionWalletBalanceChange {
@@ -301,6 +314,8 @@ func processMessage(svc *sqs.SQS, queueURL string, msg *sqs.Message) {
 			if err != nil {
 				processErr = err
 				global.GVA_LOG.Error("Save SqsWalletBalanceChangeLog error", zap.Error(err))
+			} else {
+				account.ReportWalletBalanceChangeEvent(context.Background(), req)
 			}
 		}
 	} else {

config-dev.yaml

@@ -176,3 +176,16 @@ aws:
   aws-sqs-access-key: AKIAUWKJ5EVVM2APLKGR
   aws-sqs-secret-key: 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
   sqs-region: ""
+
+devtodev:
+  app-id: "2d6fd1f8-a02e-0143-9497-9db2c6abac47_WW-qfj"
+  default-currency: "USD"
+  platform: "web"
+  app-version: ""
+  bundle: ""
+  current-balance-min-interval-seconds: 86400
+  deposit-source-types: []
+  withdraw-source-types: []
+  currency-accrual-source-types: []
+  virtual-currency-spent-source-types: []
+  virtual-currency-payment-source-types: []

config.yaml

@@ -176,3 +176,16 @@ aws:
   aws-sqs-access-key: AKIAUWKJ5EVVM2APLKGR
   aws-sqs-secret-key: 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
   sqs-region: ""
+
+devtodev:
+  app-id: "2d6fd1f8-a02e-0143-9497-9db2c6abac47_WW-qfj"
+  default-currency: "USD"
+  platform: "web"
+  app-version: ""
+  bundle: ""
+  current-balance-min-interval-seconds: 86400
+  deposit-source-types: []
+  withdraw-source-types: []
+  currency-accrual-source-types: []
+  virtual-currency-spent-source-types: []
+  virtual-currency-payment-source-types: []

config/config.go

@@ -40,4 +40,7 @@ type Server struct {
 
 	// AwsConfig 配置
 	AWS AwsConfig `mapstructure:"aws" json:"aws" yaml:"aws"`
+
+	// devtodev 配置
+	DevToDev DevToDev `mapstructure:"devtodev" json:"devtodev" yaml:"devtodev"`
 }

config/devtodev.go

@@ -0,0 +1,15 @@
+package config
+
+type DevToDev struct {
+	AppID                             string  `mapstructure:"app-id" json:"app-id" yaml:"app-id"`
+	DefaultCurrency                   string  `mapstructure:"default-currency" json:"default-currency" yaml:"default-currency"`
+	Platform                          string  `mapstructure:"platform" json:"platform" yaml:"platform"`
+	AppVersion                        string  `mapstructure:"app-version" json:"app-version" yaml:"app-version"`
+	Bundle                            string  `mapstructure:"bundle" json:"bundle" yaml:"bundle"`
+	CurrentBalanceMinIntervalSeconds  int64   `mapstructure:"current-balance-min-interval-seconds" json:"current-balance-min-interval-seconds" yaml:"current-balance-min-interval-seconds"`
+	DepositSourceTypes                []int32 `mapstructure:"deposit-source-types" json:"deposit-source-types" yaml:"deposit-source-types"`
+	WithdrawSourceTypes               []int32 `mapstructure:"withdraw-source-types" json:"withdraw-source-types" yaml:"withdraw-source-types"`
+	CurrencyAccrualSourceTypes        []int32 `mapstructure:"currency-accrual-source-types" json:"currency-accrual-source-types" yaml:"currency-accrual-source-types"`
+	VirtualCurrencySpentSourceTypes   []int32 `mapstructure:"virtual-currency-spent-source-types" json:"virtual-currency-spent-source-types" yaml:"virtual-currency-spent-source-types"`
+	VirtualCurrencyPaymentSourceTypes []int32 `mapstructure:"virtual-currency-payment-source-types" json:"virtual-currency-payment-source-types" yaml:"virtual-currency-payment-source-types"`
+}

go.mod

@@ -5,6 +5,7 @@ go 1.24.0
 toolchain go1.24.2
 
 require (
+	devtodev-sdk v0.0.0
 	github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible
 	github.com/aws/aws-sdk-go v1.55.6
 	github.com/casbin/casbin/v2 v2.103.0
@@ -56,6 +57,8 @@ require (
 	gorm.io/gorm v1.25.12
 )
 
+replace devtodev-sdk => ./pkg/devtodev
+
 require (
 	filippo.io/edwards25519 v1.1.0 // indirect
 	github.com/BurntSushi/toml v1.4.0 // indirect
@@ -114,7 +117,7 @@ require (
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
-	github.com/klauspost/compress v1.18.0 // indirect
+	github.com/klauspost/compress v1.18.4 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.9 // indirect
 	github.com/klauspost/pgzip v1.2.6 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect

go.sum

@@ -290,6 +290,8 @@ github.com/klauspost/compress v1.4.1/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
+github.com/klauspost/compress v1.18.4 h1:RPhnKRAQ4Fh8zU2FY/6ZFDwTVTxgJ/EMydqSTzE9a2c=
+github.com/klauspost/compress v1.18.4/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4=
 github.com/klauspost/cpuid v1.2.0/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
 github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.2.9 h1:66ze0taIn2H33fBvCkXuv9BmCwDfafmiIVpKV9kKGuY=

model/awssqs/sqs.go

@@ -2,6 +2,7 @@ package awssqs
 
 import (
 	"encoding/json"
+	"strings"
 
 	"github.com/shopspring/decimal"
 )
@@ -150,6 +151,43 @@ type SqsActionWalletBalanceChangeContent struct {
 	RecordNo      string `json:"recordNo"`      // 账变流水号
 }
 
+func (c SqsActionUserBehaviorContent) GetDeviceID() string {
+	if c.Header == "" {
+		return ""
+	}
+	var header map[string]interface{}
+	if err := json.Unmarshal([]byte(c.Header), &header); err != nil {
+		return ""
+	}
+	for _, key := range []string{"X-Devid"} {
+		if value := findStringValue(header, key); value != "" {
+			return value
+		}
+	}
+	return ""
+}
+
+func (c SqsActionWalletBalanceChangeContent) GetDeviceID() string {
+	return ""
+}
+
+func findStringValue(values map[string]interface{}, key string) string {
+	if value, ok := values[key]; ok {
+		if text, ok := value.(string); ok && strings.TrimSpace(text) != "" {
+			return strings.TrimSpace(text)
+		}
+	}
+	for currentKey, value := range values {
+		if !strings.EqualFold(currentKey, key) {
+			continue
+		}
+		if text, ok := value.(string); ok && strings.TrimSpace(text) != "" {
+			return strings.TrimSpace(text)
+		}
+	}
+	return ""
+}
+
 func (m SqsMessage) Json() (string, error) {
 	marshal, err := json.Marshal(m)
 	if err != nil {

pkg/devtodev/README.md

@@ -0,0 +1,140 @@
+# devtodev SDK (Go)
+
+Minimal Go SDK for devtodev Data API 2.0 event reporting.
+
+## Install
+
+Use as a local module or set your module path in `go.mod`.
+
+## Usage
+
+```go
+package main
+
+import (
+    "context"
+    "time"
+
+    devtodev "devtodev-sdk"
+)
+
+func main() {
+    client := devtodev.NewClient("YOUR_APP_ID")
+
+    payload := devtodev.Payload{
+        Reports: []devtodev.Report{
+            {
+                DeviceID: "device-123",
+                Packages: []devtodev.DevtodevPackage{
+                    *devtodev.NewReporter(nil, "device-123").NewPackage().
+                        WithLanguage("en").
+                        WithCountry("US").
+                        Append(devtodev.RawEvent{
+                            "code":      "ce",
+                            "timestamp": time.Now().UnixMilli(),
+                            "level":     5,
+                            "name":      "custom_event",
+                            "parameters": map[string]interface{}{
+                                "level": 5,
+                            },
+                        }),
+                },
+            },
+        },
+    }
+
+    if err := client.Send(context.Background(), payload); err != nil {
+        panic(err)
+    }
+}
+```
+
+## Event Builder
+
+This SDK uses a `Reporter -> Package -> Event` builder flow. Build one or more events, append them into a package, then report once.
+
+```go
+client := devtodev.NewClient("YOUR_APP_ID")
+
+reporter := devtodev.NewReporter(client, "device-123")
+pkg := reporter.NewPackage().
+    WithPlatform("web").
+    WithLanguage("en").
+    WithCountry("US").
+    WithIP("127.0.0.1").
+    WithAppVersion("1.0.0")
+
+pkg.Append(devtodev.NewDeviceInfoEvent(time.Now().UnixMilli(), map[string]interface{}{
+    "platform": "ios",
+    "device":   "iPhone14,3",
+}))
+
+pkg.Append(devtodev.NewCustomEvent(time.Now().UnixMilli(), 1, "custom_event", map[string]interface{}{
+    "score": 123,
+}, nil))
+
+_, _ = pkg.Report(context.Background())
+```
+
+## 中文事件文档
+
+**Service Events**
+- Device Info（`di`）：设备信息上报。必填字段：`code=di`、`timestamp`。新用户必须先发送该事件，否则不会注册用户。
+- Session Start（`ss`）：会话开始。必填字段：`code=ss`、`timestamp`、`level`。
+- User Engagement（`ue`）：用户活跃/心跳，记录活跃时长。必填字段：`code=ue`、`timestamp`、`level`、`length`（秒）。
+- Setting User Tracking Status (GDPR)（`ts`）：用户追踪授权状态。必填字段：`code=ts`、`timestamp`、`trackingAllowed`（bool）。
+- Alive（`al`）：应用存活/心跳。必填字段：`code=al`、`timestamp`。
+
+**User properties**
+- People（`pl`）：用户属性更新。必填字段：`code=pl`、`timestamp`、`level`、`parameters`（用户属性键值对）。
+
+**Basic Events**
+- Custom Event（`ce`）：自定义事件。必填字段：`code=ce`、`timestamp`、`level`、`name`；`parameters` 可选。
+- Real Payment（`rp`）：真实支付（IAP）。必填字段：`code=rp`、`timestamp`、`level`、`productId`、`orderId`、`price`、`currencyCode`。
+- Onboarding / Tutorial（`tr`）：新手引导。必填字段：`code=tr`、`timestamp`、`level`、`step`。
+- Virtual Currency Payment（`vp`）：虚拟货币消费。必填字段：`code=vp`、`timestamp`、`level`、`purchaseAmount`、`purchasePrice`、`purchaseType`、`purchaseId`。
+- Currency Accrual（`ca`）：虚拟货币获得。必填字段：`code=ca`、`timestamp`、`level`，且 `bought` 与 `earned` 至少提供一个。
+- Current Balance（`cb`）：当前余额。必填字段：`code=cb`、`timestamp`、`level`、`balance`。
+- Level Up（`lu`）：升级。必填字段：`code=lu`、`timestamp`、`level`；`balance`/`spent`/`earned`/`bought` 可选。
+- Progression Event（`pe`）：进度事件。必填字段：`code=pe`、`timestamp`、`level`、`name`、`parameters`。`parameters` 内必填：`success`、`duration`。
+
+**Secondary Events**
+- Referral（`rf`）：邀请推荐/安装来源。必填字段：`code=rf`、`timestamp`。
+- Ad Impression（`adrv`）：广告展示。必填字段：`code=adrv`、`timestamp`、`ad_network`、`revenue`。
+- Social Connect（`sc`）：社交绑定。必填字段：`code=sc`、`timestamp`、`level`、`socialNetwork`。
+- Social Post（`sp`）：社交分享。必填字段：`code=sp`、`timestamp`、`level`、`socialNetwork`、`postReason`。
+
+**事件相关性说明**
+- Session Start（`ss`）需要搭配 User Engagement（`ue`）才能完整统计会话时长。
+- Device Info（`di`）必须作为新用户的首个事件上报；建议每次会话开始时也上报一次。
+- Alive（`al`）在用户超过 5 分钟没有任何事件时，需要上报以保持“在线”状态展示。
+- Setting User Tracking Status（`ts`）当用户拒绝追踪时必须上报；`trackingAllowed=false` 会触发删除该用户数据并禁止继续收集，若之后改为 `true` 则视为新用户。
+- Session Start（`ss`）/ User Engagement（`ue`）/ Alive（`al`）可通过 `sessionId` 关联为同一会话。
+- Currency Accrual（`ca`）不建议按“每笔交易”上报，应按 5-10 分钟聚合上报；当用户等级变化时应中断并上报一次聚合结果。
+- Current Balance（`cb`）不应对同一用户一天上报超过一次。
+- Referral（`rf`）每个用户仅需上报一次；若已接入 AppsFlyer 等广告追踪或 devtodev 自定义回调，可不再上报。
+- Progression Event（`pe`）的 `parameters.source` 用于串联上一个关卡/区域，便于形成进度链路。
+
+## Notes
+
+- Default endpoint: `https://api.devtodev.com/v2/analytics/report?appId=YOUR_APP_ID`
+- Payload size limit is 2 MB uncompressed.
+- Timestamps are in milliseconds since epoch.
+- For new users, send the Device Info event (`code: "di"`) first so the user is registered.
+- SDK validates required fields (`reports`, `deviceId`, `packages`, `events`, `code`, `timestamp`) before sending.
+- Content-Type values follow the API: `application/json`, `application/zstd`, `application/gzip`.
+- Use `SendWithResponse` if you need response details (`status`, `headers`, `body`, parsed JSON).
+- Retries happen on common transient HTTP statuses and network errors.
+
+## Customize
+
+```go
+client := devtodev.NewClient("YOUR_APP_ID")
+client.Endpoint = "https://api.devtodev.com/v2/analytics/report"
+client.Retry.MaxAttempts = 5
+client.Retry.BaseDelay = 300 * time.Millisecond
+client.Retry.MaxDelay = 3 * time.Second
+
+// Compression options
+client.Compression = devtodev.CompressionGzip // or CompressionZstd / CompressionNone
+```

pkg/devtodev/cmd/send/main.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"time"
+
+	devtodev "devtodev-sdk"
+)
+
+func main() {
+	rand.Seed(time.Now().UnixNano())
+
+	appID := "2d942fe9-c1d3-081c-bdb3-ec6080273fe6"
+	client := devtodev.NewClient(appID)
+	client.Compression = devtodev.CompressionNone
+
+	ctx := context.Background()
+	userCount := rand.Intn(2) + 2 // 2-3 users
+
+	for u := 0; u < userCount; u++ {
+		deviceID := fmt.Sprintf("device-%d", rand.Int63())
+		base := time.Now().Add(time.Duration(-rand.Intn(120)) * time.Second)
+		level := rand.Intn(20) + 1
+
+		reporter := devtodev.NewReporter(client, deviceID)
+		pkg := reporter.NewPackage().
+			WithLanguage("en").
+			WithCountry("US")
+
+		deviceInfoEvent := devtodev.NewDeviceInfoEvent(base.UnixMilli(), map[string]interface{}{
+			"platform": "ios",
+			"device":   "iPhone14,3",
+		})
+		pkg.Append(deviceInfoEvent)
+
+		sessionStartEvent := devtodev.NewSessionStartEvent(base.Add(2*time.Second).UnixMilli(), level, nil)
+		pkg.Append(sessionStartEvent)
+
+		// 2-4 user engagement heartbeats
+		ueCount := rand.Intn(3) + 2
+		for i := 0; i < ueCount; i++ {
+			ueEvent := devtodev.NewUserEngagementEvent(base.Add(time.Duration(10*(i+1))*time.Second).UnixMilli(), level, 10, nil)
+			pkg.Append(ueEvent)
+		}
+
+		// 1-3 real payments, each price 5-10
+		payments := rand.Intn(3) + 1
+		for i := 0; i < payments; i++ {
+			price := float64(rand.Intn(6) + 5)
+			paymentEvent := devtodev.NewRealPaymentEvent(
+				base.Add(time.Duration(60+(i*10))*time.Second).UnixMilli(),
+				level,
+				fmt.Sprintf("com.demo.product.%d", rand.Intn(5)+1),
+				fmt.Sprintf("order-%d-%d", time.Now().UnixNano(), i),
+				price,
+				"USD",
+				nil,
+			)
+			pkg.Append(paymentEvent)
+		}
+
+		if _, err := pkg.Report(ctx); err != nil {
+			panic(err)
+		}
+	}
+
+	fmt.Printf("sent events for %d users\n", userCount)
+}

pkg/devtodev/devtodev.go

@@ -0,0 +1,338 @@
+package devtodev
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"math"
+	"math/rand"
+	"net"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/klauspost/compress/zstd"
+)
+
+const (
+	defaultEndpoint  = "https://api.devtodev.com/v2/analytics/report"
+	defaultUserAgent = "devtodev-go-sdk/0.1"
+	maxPayloadBytes  = 2 * 1024 * 1024
+)
+
+// Compression controls request body encoding.
+type Compression string
+
+const (
+	CompressionNone Compression = "none"
+	CompressionGzip Compression = "gzip"
+	CompressionZstd Compression = "zstd"
+)
+
+// Client sends events to devtodev Data API 2.0.
+type Client struct {
+	AppID       string
+	Endpoint    string
+	HTTP        *http.Client
+	Retry       RetryConfig
+	UserAgent   string
+	Compression Compression
+}
+
+// Response captures HTTP response details for a send.
+type Response struct {
+	StatusCode int
+	Headers    http.Header
+	Body       []byte
+	JSON       map[string]interface{}
+}
+
+// APIError provides details for non-2xx responses.
+type APIError struct {
+	StatusCode int
+	Message    string
+	Body       []byte
+}
+
+func (e *APIError) Error() string {
+	if e.Message != "" {
+		return fmt.Sprintf("request failed with status %d: %s", e.StatusCode, e.Message)
+	}
+	return fmt.Sprintf("request failed with status %d", e.StatusCode)
+}
+
+// RetryConfig configures retry behavior for transient failures.
+type RetryConfig struct {
+	MaxAttempts   int
+	BaseDelay     time.Duration
+	MaxDelay      time.Duration
+	Jitter        bool
+	RetryStatuses map[int]struct{}
+}
+
+// NewClient creates a client with defaults.
+func NewClient(appID string) *Client {
+	return &Client{
+		AppID:    appID,
+		Endpoint: defaultEndpoint,
+		HTTP:     &http.Client{Timeout: 10 * time.Second},
+		Retry: RetryConfig{
+			MaxAttempts:   3,
+			BaseDelay:     500 * time.Millisecond,
+			MaxDelay:      5 * time.Second,
+			Jitter:        true,
+			RetryStatuses: map[int]struct{}{408: {}, 425: {}, 429: {}, 500: {}, 502: {}, 503: {}, 504: {}},
+		},
+		UserAgent:   defaultUserAgent,
+		Compression: CompressionNone,
+	}
+}
+
+// Send sends a payload to devtodev.
+func (c *Client) Send(ctx context.Context, payload Payload) error {
+	_, err := c.SendWithResponse(ctx, payload)
+	return err
+}
+
+// SendWithResponse sends a payload and returns the HTTP response details.
+func (c *Client) SendWithResponse(ctx context.Context, payload Payload) (*Response, error) {
+	if c == nil {
+		return nil, errors.New("client is nil")
+	}
+	if c.AppID == "" {
+		return nil, errors.New("AppID is required")
+	}
+	if err := ValidatePayload(payload); err != nil {
+		return nil, err
+	}
+	endpoint := c.Endpoint
+	if endpoint == "" {
+		endpoint = defaultEndpoint
+	}
+
+	body, err := json.Marshal(payload)
+	if err != nil {
+		return nil, fmt.Errorf("marshal payload: %w", err)
+	}
+	if len(body) > maxPayloadBytes {
+		return nil, fmt.Errorf("payload too large: %d bytes exceeds %d bytes", len(body), maxPayloadBytes)
+	}
+
+	contentType := "application/json"
+	payloadBytes := body
+	switch c.Compression {
+	case "", CompressionNone:
+		// no-op
+	case CompressionGzip:
+		contentType = "application/gzip"
+		payloadBytes, err = gzipCompress(body)
+		if err != nil {
+			return nil, fmt.Errorf("gzip compress: %w", err)
+		}
+	case CompressionZstd:
+		contentType = "application/zstd"
+		payloadBytes, err = zstdCompress(body)
+		if err != nil {
+			return nil, fmt.Errorf("zstd compress: %w", err)
+		}
+	default:
+		return nil, fmt.Errorf("unsupported compression: %s", c.Compression)
+	}
+
+	reqURL, err := withAppID(endpoint, c.AppID)
+	if err != nil {
+		return nil, fmt.Errorf("build url: %w", err)
+	}
+
+	httpClient := c.HTTP
+	if httpClient == nil {
+		httpClient = &http.Client{Timeout: 10 * time.Second}
+	}
+
+	attempts := max(1, c.Retry.MaxAttempts)
+	for i := 1; i <= attempts; i++ {
+		req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL, bytes.NewReader(payloadBytes))
+		if err != nil {
+			return nil, fmt.Errorf("new request: %w", err)
+		}
+		req.Header.Set("Content-Type", contentType)
+		if c.UserAgent != "" {
+			req.Header.Set("User-Agent", c.UserAgent)
+		}
+
+		resp, err := httpClient.Do(req)
+		var parsed *Response
+		if err == nil && resp != nil {
+			parsed, err = parseResponse(resp)
+			if err != nil {
+				return nil, fmt.Errorf("read response: %w", err)
+			}
+		}
+
+		if err == nil && resp != nil && resp.StatusCode >= 200 && resp.StatusCode < 300 {
+			return parsed, nil
+		}
+
+		retry := shouldRetry(err, resp, c.Retry.RetryStatuses)
+		if !retry || i == attempts {
+			if err != nil {
+				return parsed, fmt.Errorf("request failed: %w", err)
+			}
+			return parsed, apiError(parsed)
+		}
+
+		wait := backoff(c.Retry, i)
+		if wait > 0 {
+			timer := time.NewTimer(wait)
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return nil, ctx.Err()
+			case <-timer.C:
+			}
+		}
+	}
+
+	return nil, errors.New("unexpected retry loop exit")
+}
+
+func withAppID(endpoint, appID string) (string, error) {
+	u, err := url.Parse(endpoint)
+	if err != nil {
+		return "", err
+	}
+	q := u.Query()
+	if q.Get("appId") == "" {
+		q.Set("appId", appID)
+	}
+	u.RawQuery = q.Encode()
+	return u.String(), nil
+}
+
+func shouldRetry(err error, resp *http.Response, retryStatuses map[int]struct{}) bool {
+	if err != nil {
+		var netErr net.Error
+		if errors.As(err, &netErr) {
+			return true
+		}
+		if strings.Contains(err.Error(), "timeout") {
+			return true
+		}
+		return true
+	}
+	if resp == nil {
+		return true
+	}
+	if _, ok := retryStatuses[resp.StatusCode]; ok {
+		return true
+	}
+	if resp.StatusCode >= 500 {
+		return true
+	}
+	return false
+}
+
+func backoff(cfg RetryConfig, attempt int) time.Duration {
+	if cfg.BaseDelay <= 0 {
+		return 0
+	}
+	exp := math.Pow(2, float64(attempt-1))
+	delay := time.Duration(float64(cfg.BaseDelay) * exp)
+	if cfg.MaxDelay > 0 && delay > cfg.MaxDelay {
+		delay = cfg.MaxDelay
+	}
+	if cfg.Jitter {
+		maxJitter := int64(delay / 2)
+		if maxJitter > 0 {
+			delay = delay/2 + time.Duration(rand.Int63n(maxJitter))
+		}
+	}
+	return delay
+}
+
+func max(a, b int) int {
+	if a > b {
+		return a
+	}
+	return b
+}
+
+func gzipCompress(data []byte) ([]byte, error) {
+	var buf bytes.Buffer
+	zw := gzip.NewWriter(&buf)
+	if _, err := zw.Write(data); err != nil {
+		zw.Close()
+		return nil, err
+	}
+	if err := zw.Close(); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+func zstdCompress(data []byte) ([]byte, error) {
+	var buf bytes.Buffer
+	encoder, err := zstd.NewWriter(&buf)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := encoder.Write(data); err != nil {
+		encoder.Close()
+		return nil, err
+	}
+	if err := encoder.Close(); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+func parseResponse(resp *http.Response) (*Response, error) {
+	if resp == nil {
+		return nil, nil
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	out := &Response{
+		StatusCode: resp.StatusCode,
+		Headers:    resp.Header.Clone(),
+		Body:       body,
+	}
+	if isJSONResponse(resp.Header.Get("Content-Type")) && len(body) > 0 {
+		var m map[string]interface{}
+		if err := json.Unmarshal(body, &m); err == nil {
+			out.JSON = m
+		}
+	}
+	return out, nil
+}
+
+func apiError(resp *Response) error {
+	if resp == nil {
+		return &APIError{StatusCode: 0, Message: "no response"}
+	}
+	msg := ""
+	if resp.JSON != nil {
+		if v, ok := resp.JSON["message"].(string); ok {
+			msg = v
+		} else if v, ok := resp.JSON["error"].(string); ok {
+			msg = v
+		}
+	}
+	return &APIError{
+		StatusCode: resp.StatusCode,
+		Message:    msg,
+		Body:       resp.Body,
+	}
+}
+
+func isJSONResponse(contentType string) bool {
+	return strings.Contains(contentType, "application/json") || strings.Contains(contentType, "+json")
+}

pkg/devtodev/devtodev_test.go

@@ -0,0 +1,249 @@
+package devtodev
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/klauspost/compress/zstd"
+)
+
+func TestSend_NoCompression(t *testing.T) {
+	payload := samplePayload()
+	expected, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatalf("marshal payload: %v", err)
+	}
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost {
+			t.Fatalf("method: %s", r.Method)
+		}
+		if ct := r.Header.Get("Content-Type"); ct != "application/json" {
+			t.Fatalf("content-type: %s", ct)
+		}
+		if r.URL.Query().Get("appId") != "app-123" {
+			t.Fatalf("appId: %s", r.URL.Query().Get("appId"))
+		}
+		got, err := io.ReadAll(r.Body)
+		if err != nil {
+			t.Fatalf("read body: %v", err)
+		}
+		if !bytes.Equal(got, expected) {
+			t.Fatalf("body mismatch")
+		}
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	client := NewClient("app-123")
+	client.Endpoint = server.URL
+
+	if err := client.Send(context.Background(), payload); err != nil {
+		t.Fatalf("send: %v", err)
+	}
+}
+
+func TestSend_GzipCompression(t *testing.T) {
+	payload := samplePayload()
+	expected, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatalf("marshal payload: %v", err)
+	}
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if ct := r.Header.Get("Content-Type"); ct != "application/gzip" {
+			t.Fatalf("content-type: %s", ct)
+		}
+		zr, err := gzip.NewReader(r.Body)
+		if err != nil {
+			t.Fatalf("gzip reader: %v", err)
+		}
+		got, err := io.ReadAll(zr)
+		if err != nil {
+			t.Fatalf("read body: %v", err)
+		}
+		zr.Close()
+		if !bytes.Equal(got, expected) {
+			t.Fatalf("body mismatch")
+		}
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	client := NewClient("app-123")
+	client.Endpoint = server.URL
+	client.Compression = CompressionGzip
+
+	if err := client.Send(context.Background(), payload); err != nil {
+		t.Fatalf("send: %v", err)
+	}
+}
+
+func TestSend_ZstdCompression(t *testing.T) {
+	payload := samplePayload()
+	expected, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatalf("marshal payload: %v", err)
+	}
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if ct := r.Header.Get("Content-Type"); ct != "application/zstd" {
+			t.Fatalf("content-type: %s", ct)
+		}
+		decoder, err := zstd.NewReader(r.Body)
+		if err != nil {
+			t.Fatalf("zstd reader: %v", err)
+		}
+		got, err := io.ReadAll(decoder)
+		if err != nil {
+			t.Fatalf("read body: %v", err)
+		}
+		decoder.Close()
+		if !bytes.Equal(got, expected) {
+			t.Fatalf("body mismatch")
+		}
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	client := NewClient("app-123")
+	client.Endpoint = server.URL
+	client.Compression = CompressionZstd
+
+	if err := client.Send(context.Background(), payload); err != nil {
+		t.Fatalf("send: %v", err)
+	}
+}
+
+func TestValidatePayload(t *testing.T) {
+	cases := []struct {
+		name    string
+		payload Payload
+		ok      bool
+	}{
+		{
+			name:    "empty",
+			payload: Payload{},
+			ok:      false,
+		},
+		{
+			name: "missing device id",
+			payload: Payload{
+				Reports: []Report{{}},
+			},
+			ok: false,
+		},
+		{
+			name: "missing packages",
+			payload: Payload{
+				Reports: []Report{{DeviceID: "d1"}},
+			},
+			ok: false,
+		},
+		{
+			name: "missing events",
+			payload: Payload{
+				Reports: []Report{{DeviceID: "d1", Packages: []DevtodevPackage{{}}}},
+			},
+			ok: false,
+		},
+		{
+			name: "missing code",
+			payload: Payload{
+				Reports: []Report{{DeviceID: "d1", Packages: []DevtodevPackage{{events: []Event{RawEvent{"timestamp": int64(1)}}}}}},
+			},
+			ok: false,
+		},
+		{
+			name: "missing timestamp",
+			payload: Payload{
+				Reports: []Report{{DeviceID: "d1", Packages: []DevtodevPackage{{events: []Event{RawEvent{"code": "ce"}}}}}},
+			},
+			ok: false,
+		},
+		{
+			name: "ok",
+			payload: Payload{
+				Reports: []Report{{DeviceID: "d1", Packages: []DevtodevPackage{{events: []Event{RawEvent{"code": "ce", "timestamp": int64(1)}}}}}},
+			},
+			ok: true,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := ValidatePayload(tc.payload)
+			if tc.ok && err != nil {
+				t.Fatalf("expected ok, got %v", err)
+			}
+			if !tc.ok && err == nil {
+				t.Fatalf("expected error")
+			}
+		})
+	}
+}
+
+func TestSendWithResponse_ParsesJSONError(t *testing.T) {
+	payload := samplePayload()
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusBadRequest)
+		_, _ = w.Write([]byte(`{"message":"bad request"}`))
+	}))
+	defer server.Close()
+
+	client := NewClient("app-123")
+	client.Endpoint = server.URL
+	client.Retry.MaxAttempts = 1
+
+	resp, err := client.SendWithResponse(context.Background(), payload)
+	if err == nil {
+		t.Fatalf("expected error")
+	}
+	if resp == nil || resp.StatusCode != http.StatusBadRequest {
+		t.Fatalf("unexpected response: %+v", resp)
+	}
+	var apiErr *APIError
+	if !errors.As(err, &apiErr) {
+		t.Fatalf("expected APIError, got %T", err)
+	}
+	if apiErr.Message != "bad request" {
+		t.Fatalf("unexpected message: %q", apiErr.Message)
+	}
+}
+
+func samplePayload() Payload {
+	return Payload{
+		Reports: []Report{
+			{
+				DeviceID: "device-123",
+				Packages: []DevtodevPackage{
+					{
+						language: "en",
+						country:  "US",
+						events: []Event{
+							RawEvent{
+								"code":      "ce",
+								"timestamp": time.Now().UnixMilli(),
+								"level":     5,
+								"name":      "custom_event",
+								"parameters": map[string]interface{}{
+									"level": 5,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}

pkg/devtodev/event/common.go

@@ -0,0 +1,20 @@
+package event
+
+import "fmt"
+
+func addFields(event map[string]interface{}, fields map[string]interface{}, protected ...string) error {
+	if len(fields) == 0 {
+		return nil
+	}
+	block := map[string]struct{}{}
+	for _, key := range protected {
+		block[key] = struct{}{}
+	}
+	for k, v := range fields {
+		if _, ok := block[k]; ok {
+			return fmt.Errorf("fields must not override %s", k)
+		}
+		event[k] = v
+	}
+	return nil
+}

pkg/devtodev/event/game.go

@@ -0,0 +1,91 @@
+package event
+
+import "fmt"
+
+func Onboarding(timestamp int64, level int, step int, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":      "tr",
+		"timestamp": timestamp,
+		"level":     level,
+		"step":      step,
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level", "step"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func CurrencyAccrual(timestamp int64, level int, bought, earned map[string]map[string]float64, fields map[string]interface{}) (map[string]interface{}, error) {
+	if len(bought) == 0 && len(earned) == 0 {
+		return nil, fmt.Errorf("either bought or earned must be provided")
+	}
+	event := map[string]interface{}{
+		"code":      "ca",
+		"timestamp": timestamp,
+		"level":     level,
+	}
+	if len(bought) > 0 {
+		event["bought"] = bought
+	}
+	if len(earned) > 0 {
+		event["earned"] = earned
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level", "bought", "earned"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func CurrentBalance(timestamp int64, level int, balance map[string]float64, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":      "cb",
+		"timestamp": timestamp,
+		"level":     level,
+		"balance":   balance,
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level", "balance"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func LevelUp(timestamp int64, level int, balance, spent, earned, bought map[string]float64, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":      "lu",
+		"timestamp": timestamp,
+		"level":     level,
+	}
+	if len(balance) > 0 {
+		event["balance"] = balance
+	}
+	if len(spent) > 0 {
+		event["spent"] = spent
+	}
+	if len(earned) > 0 {
+		event["earned"] = earned
+	}
+	if len(bought) > 0 {
+		event["bought"] = bought
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level", "balance", "spent", "earned", "bought"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func ProgressionEvent(timestamp int64, level int, name string, parameters map[string]interface{}, fields map[string]interface{}) (map[string]interface{}, error) {
+	if len(parameters) == 0 {
+		return nil, fmt.Errorf("parameters is required")
+	}
+	event := map[string]interface{}{
+		"code":       "pe",
+		"timestamp":  timestamp,
+		"level":      level,
+		"name":       name,
+		"parameters": parameters,
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level", "name", "parameters"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}

pkg/devtodev/event/payment.go

@@ -0,0 +1,46 @@
+package event
+
+func RealPayment(timestamp int64, level int, productID, orderID string, price float64, currencyCode string, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":         "rp",
+		"timestamp":    timestamp,
+		"level":        level,
+		"productId":    productID,
+		"orderId":      orderID,
+		"price":        price,
+		"currencyCode": currencyCode,
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level", "productId", "orderId", "price", "currencyCode"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func VirtualCurrencyPayment(timestamp int64, level int, purchaseAmount int, purchasePrice map[string]float64, purchaseType, purchaseID string, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":           "vp",
+		"timestamp":      timestamp,
+		"level":          level,
+		"purchaseAmount": purchaseAmount,
+		"purchasePrice":  purchasePrice,
+		"purchaseType":   purchaseType,
+		"purchaseId":     purchaseID,
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level", "purchaseAmount", "purchasePrice", "purchaseType", "purchaseId"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func AdImpression(timestamp int64, adNetwork string, revenue float64, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":       "adrv",
+		"timestamp":  timestamp,
+		"ad_network": adNetwork,
+		"revenue":    revenue,
+	}
+	if err := addFields(event, fields, "code", "timestamp", "ad_network", "revenue"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}

pkg/devtodev/event/session.go

@@ -0,0 +1,60 @@
+package event
+
+func DeviceInfo(timestamp int64, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":      "di",
+		"timestamp": timestamp,
+	}
+	if err := addFields(event, fields, "code", "timestamp"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func SessionStart(timestamp int64, level int, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":      "ss",
+		"timestamp": timestamp,
+		"level":     level,
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func UserEngagement(timestamp int64, level int, length int, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":      "ue",
+		"timestamp": timestamp,
+		"level":     level,
+		"length":    length,
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level", "length"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func TrackingStatus(timestamp int64, trackingAllowed bool, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":            "ts",
+		"timestamp":       timestamp,
+		"trackingAllowed": trackingAllowed,
+	}
+	if err := addFields(event, fields, "code", "timestamp", "trackingAllowed"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func Alive(timestamp int64, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":      "al",
+		"timestamp": timestamp,
+	}
+	if err := addFields(event, fields, "code", "timestamp"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}

pkg/devtodev/event/user.go

@@ -0,0 +1,73 @@
+package event
+
+import "fmt"
+
+func People(timestamp int64, level int, properties map[string]interface{}, fields map[string]interface{}) (map[string]interface{}, error) {
+	if len(properties) == 0 {
+		return nil, fmt.Errorf("parameters is required")
+	}
+	event := map[string]interface{}{
+		"code":       "pl",
+		"timestamp":  timestamp,
+		"level":      level,
+		"parameters": properties,
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level", "parameters"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func CustomEvent(timestamp int64, level int, name string, parameters map[string]interface{}, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":      "ce",
+		"timestamp": timestamp,
+		"level":     level,
+		"name":      name,
+	}
+	if parameters != nil {
+		event["parameters"] = parameters
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level", "name", "parameters"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func Referral(timestamp int64, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":      "rf",
+		"timestamp": timestamp,
+	}
+	if err := addFields(event, fields, "code", "timestamp"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func SocialConnect(timestamp int64, level int, socialNetwork string, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":          "sc",
+		"timestamp":     timestamp,
+		"level":         level,
+		"socialNetwork": socialNetwork,
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level", "socialNetwork"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func SocialPost(timestamp int64, level int, socialNetwork, postReason string, fields map[string]interface{}) (map[string]interface{}, error) {
+	event := map[string]interface{}{
+		"code":          "sp",
+		"timestamp":     timestamp,
+		"level":         level,
+		"socialNetwork": socialNetwork,
+		"postReason":    postReason,
+	}
+	if err := addFields(event, fields, "code", "timestamp", "level", "socialNetwork", "postReason"); err != nil {
+		return nil, err
+	}
+	return event, nil
+}

pkg/devtodev/events.go

@@ -0,0 +1,163 @@
+package devtodev
+
+import (
+	"context"
+	"fmt"
+
+	"devtodev-sdk/event"
+)
+
+// Reporter wraps Client with report-level identity fields.
+type Reporter struct {
+	Client           *Client
+	DeviceID         string
+	UserID           string
+	PreviousDeviceID string
+	PreviousUserID   string
+	DevtodevID       string
+}
+
+// NewReporter creates a Reporter with report-level fields.
+func NewReporter(client *Client, deviceID string) *Reporter {
+	return &Reporter{
+		Client:   client,
+		DeviceID: deviceID,
+	}
+}
+
+// NewPackage creates a package bound to the reporter.
+func (r *Reporter) NewPackage() *DevtodevPackage {
+	return &DevtodevPackage{
+		reporter: r,
+		events:   make([]Event, 0),
+	}
+}
+
+// Report sends one request with one report and one or more packages.
+func (r *Reporter) Report(ctx context.Context, packages ...*DevtodevPackage) (*Response, error) {
+	if r == nil {
+		return nil, fmt.Errorf("reporter is nil")
+	}
+	if r.Client == nil {
+		return nil, fmt.Errorf("client is nil")
+	}
+	if r.DeviceID == "" {
+		return nil, fmt.Errorf("deviceId is required")
+	}
+	if len(packages) == 0 {
+		return nil, fmt.Errorf("packages is required")
+	}
+
+	rawPackages := make([]DevtodevPackage, 0, len(packages))
+	for _, pkg := range packages {
+		if pkg == nil {
+			return nil, fmt.Errorf("package is nil")
+		}
+		pkg.reporter = r
+		rawPackages = append(rawPackages, *pkg)
+	}
+
+	payload := Payload{
+		Reports: []Report{
+			{
+				DeviceID:         r.DeviceID,
+				UserID:           r.UserID,
+				PreviousDeviceID: r.PreviousDeviceID,
+				PreviousUserID:   r.PreviousUserID,
+				DevtodevID:       r.DevtodevID,
+				Packages:         rawPackages,
+			},
+		},
+	}
+	return r.Client.SendWithResponse(ctx, payload)
+}
+
+// Report sends the package using the reporter bound by NewPackage.
+func (p *DevtodevPackage) Report(ctx context.Context) (*Response, error) {
+	if p == nil {
+		return nil, fmt.Errorf("package is nil")
+	}
+	if p.reporter == nil {
+		return nil, fmt.Errorf("package reporter is nil")
+	}
+	return p.reporter.Report(ctx, p)
+}
+
+func wrapEvent(raw map[string]interface{}, err error) Event {
+	if err != nil {
+		return builtEvent{err: err}
+	}
+	return builtEvent{payload: raw}
+}
+
+func NewDeviceInfoEvent(timestamp int64, fields map[string]interface{}) Event {
+	return wrapEvent(event.DeviceInfo(timestamp, fields))
+}
+
+func NewSessionStartEvent(timestamp int64, level int, fields map[string]interface{}) Event {
+	return wrapEvent(event.SessionStart(timestamp, level, fields))
+}
+
+func NewUserEngagementEvent(timestamp int64, level int, length int, fields map[string]interface{}) Event {
+	return wrapEvent(event.UserEngagement(timestamp, level, length, fields))
+}
+
+func NewTrackingStatusEvent(timestamp int64, trackingAllowed bool, fields map[string]interface{}) Event {
+	return wrapEvent(event.TrackingStatus(timestamp, trackingAllowed, fields))
+}
+
+func NewAliveEvent(timestamp int64, fields map[string]interface{}) Event {
+	return wrapEvent(event.Alive(timestamp, fields))
+}
+
+func NewPeopleEvent(timestamp int64, level int, properties map[string]interface{}, fields map[string]interface{}) Event {
+	return wrapEvent(event.People(timestamp, level, properties, fields))
+}
+
+func NewCustomEvent(timestamp int64, level int, name string, parameters map[string]interface{}, fields map[string]interface{}) Event {
+	return wrapEvent(event.CustomEvent(timestamp, level, name, parameters, fields))
+}
+
+func NewRealPaymentEvent(timestamp int64, level int, productID, orderID string, price float64, currencyCode string, fields map[string]interface{}) Event {
+	return wrapEvent(event.RealPayment(timestamp, level, productID, orderID, price, currencyCode, fields))
+}
+
+func NewOnboardingEvent(timestamp int64, level int, step int, fields map[string]interface{}) Event {
+	return wrapEvent(event.Onboarding(timestamp, level, step, fields))
+}
+
+func NewVirtualCurrencyPaymentEvent(timestamp int64, level int, purchaseAmount int, purchasePrice map[string]float64, purchaseType, purchaseID string, fields map[string]interface{}) Event {
+	return wrapEvent(event.VirtualCurrencyPayment(timestamp, level, purchaseAmount, purchasePrice, purchaseType, purchaseID, fields))
+}
+
+func NewCurrencyAccrualEvent(timestamp int64, level int, bought, earned map[string]map[string]float64, fields map[string]interface{}) Event {
+	return wrapEvent(event.CurrencyAccrual(timestamp, level, bought, earned, fields))
+}
+
+func NewCurrentBalanceEvent(timestamp int64, level int, balance map[string]float64, fields map[string]interface{}) Event {
+	return wrapEvent(event.CurrentBalance(timestamp, level, balance, fields))
+}
+
+func NewLevelUpEvent(timestamp int64, level int, balance, spent, earned, bought map[string]float64, fields map[string]interface{}) Event {
+	return wrapEvent(event.LevelUp(timestamp, level, balance, spent, earned, bought, fields))
+}
+
+func NewProgressionEvent(timestamp int64, level int, name string, parameters map[string]interface{}, fields map[string]interface{}) Event {
+	return wrapEvent(event.ProgressionEvent(timestamp, level, name, parameters, fields))
+}
+
+func NewReferralEvent(timestamp int64, fields map[string]interface{}) Event {
+	return wrapEvent(event.Referral(timestamp, fields))
+}
+
+func NewAdImpressionEvent(timestamp int64, adNetwork string, revenue float64, fields map[string]interface{}) Event {
+	return wrapEvent(event.AdImpression(timestamp, adNetwork, revenue, fields))
+}
+
+func NewSocialConnectEvent(timestamp int64, level int, socialNetwork string, fields map[string]interface{}) Event {
+	return wrapEvent(event.SocialConnect(timestamp, level, socialNetwork, fields))
+}
+
+func NewSocialPostEvent(timestamp int64, level int, socialNetwork, postReason string, fields map[string]interface{}) Event {
+	return wrapEvent(event.SocialPost(timestamp, level, socialNetwork, postReason, fields))
+}

pkg/devtodev/go.mod

@@ -0,0 +1,5 @@
+module devtodev-sdk
+
+go 1.23
+
+require github.com/klauspost/compress v1.18.4

pkg/devtodev/go.sum

@@ -0,0 +1,4 @@
+github.com/klauspost/compress v1.18.1 h1:bcSGx7UbpBqMChDtsF28Lw6v/G94LPrrbMbdC3JH2co=
+github.com/klauspost/compress v1.18.1/go.mod h1:ZQFFVG+MdnR0P+l6wpXgIL4NTtwiKIdBnrBd8Nrxr+0=
+github.com/klauspost/compress v1.18.4 h1:RPhnKRAQ4Fh8zU2FY/6ZFDwTVTxgJ/EMydqSTzE9a2c=
+github.com/klauspost/compress v1.18.4/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4=

pkg/devtodev/note.md

@@ -0,0 +1,35 @@
+1.登录，2.注册，3.token刷新，4.充值，5.进入游戏，6.下注，7.提现，8.客服反馈
+
+
+1. 
+device_info 设备信息
+2.
+Virtual Currency Payment # 表示用户用虚拟币购买某个“具体商品”
+虚拟币 → 商品
+
+3.
+Current Balance # 实时资产余额
+
+4.
+session_start   ✅  会话开启,登陆
+5.
+register 注册事件 (自定义事件)
+6.
+token刷新 事件(自定义事件)
+7.
+Real Payment  # 真实支付，收到支付回调时调用,充值
+8.
+enter game 进入游戏事件(自定义事件)
+9.
+currency_accrual 游戏收益，等等其他收益,下注
+典型场景：
+游戏胜利获得筹码
+活动奖励发币
+任务奖励
+充值后发放筹
+系统补偿
+
+9.
+withdraw 提现事件(自定义事件)
+10.
+custom support 客服反馈(自定义事件)

pkg/devtodev/report_event.md

@@ -0,0 +1,16 @@
+## 事件埋点列表
+
+| 事件名称 | 上报场景 | 事件类型 |
+|---|---|---|
+| device_info | 首次安装 / 版本升级 / 设备环境变化时上报设备信息 | 预设事件 |
+| session_start | 用户产生一次会话（进入平台 / 打开App） | 预设事件 |
+| real_currency_payment | 用户真实支付成功（充值成功回调时） | 预设事件 |
+| virtual_currency_payment | 用户用虚拟币购买具体商品（VIP、礼包、道具等） | 预设事件 |
+| currency_accrual | 用户虚拟币增加（赢币、奖励、充值发筹码、系统补偿等） | 预设事件 |
+| currency_spent | 用户虚拟币消费（下注） | 预设事件 |
+| current_balance | 余额变动后 / 登录时 / 每日汇总时上报当前资产余额 | 预设事件 |
+| register | 用户完成注册时 | 自定义事件 |
+| token_refresh | 用户 token 刷新时 | 自定义事件 |
+| enter_game | 用户进入某个具体游戏时 | 自定义事件 |
+| withdraw | 用户提现成功时 | 自定义事件 |
+| custom_support | 用户提交客服反馈时 | 自定义事件 |

pkg/devtodev/types.go

@@ -0,0 +1,174 @@
+package devtodev
+
+import "encoding/json"
+
+// Payload is the top-level request body for Data API 2.0.
+type Payload struct {
+	Reports []Report `json:"reports"`
+}
+
+// Report groups events for a device (or a player).
+type Report struct {
+	DeviceID         string            `json:"deviceId"`
+	UserID           string            `json:"userId,omitempty"`
+	PreviousDeviceID string            `json:"previousDeviceId,omitempty"`
+	PreviousUserID   string            `json:"previousUserId,omitempty"`
+	DevtodevID       string            `json:"devtodevId,omitempty"`
+	Packages         []DevtodevPackage `json:"packages"`
+}
+
+// DevtodevPackage groups events under package-level metadata.
+type DevtodevPackage struct {
+	platform           string
+	language           string
+	country            string
+	ip                 string
+	appVersion         string
+	appBuildVersion    string
+	sdkVersion         string
+	bundle             string
+	engine             string
+	installationSource string
+	events             []Event
+
+	reporter *Reporter
+}
+
+// Event is the common event interface used by package builders.
+type Event interface {
+	Payload() map[string]interface{}
+	Err() error
+}
+
+// RawEvent is a generic event implementation backed by a plain map.
+type RawEvent map[string]interface{}
+
+func (e RawEvent) Payload() map[string]interface{} {
+	return map[string]interface{}(e)
+}
+
+func (e RawEvent) Err() error {
+	return nil
+}
+
+type builtEvent struct {
+	payload map[string]interface{}
+	err     error
+}
+
+func (e builtEvent) Payload() map[string]interface{} {
+	return e.payload
+}
+
+func (e builtEvent) Err() error {
+	return e.err
+}
+
+func (p *DevtodevPackage) WithPlatform(platform string) *DevtodevPackage {
+	p.platform = platform
+	return p
+}
+
+func (p *DevtodevPackage) WithLanguage(language string) *DevtodevPackage {
+	p.language = language
+	return p
+}
+
+func (p *DevtodevPackage) WithCountry(country string) *DevtodevPackage {
+	p.country = country
+	return p
+}
+
+func (p *DevtodevPackage) WithIP(ip string) *DevtodevPackage {
+	p.ip = ip
+	return p
+}
+
+func (p *DevtodevPackage) WithAppVersion(appVersion string) *DevtodevPackage {
+	p.appVersion = appVersion
+	return p
+}
+
+func (p *DevtodevPackage) WithAppBuildVersion(appBuildVersion string) *DevtodevPackage {
+	p.appBuildVersion = appBuildVersion
+	return p
+}
+
+func (p *DevtodevPackage) WithSDKVersion(sdkVersion string) *DevtodevPackage {
+	p.sdkVersion = sdkVersion
+	return p
+}
+
+func (p *DevtodevPackage) WithBundle(bundle string) *DevtodevPackage {
+	p.bundle = bundle
+	return p
+}
+
+func (p *DevtodevPackage) WithEngine(engine string) *DevtodevPackage {
+	p.engine = engine
+	return p
+}
+
+func (p *DevtodevPackage) WithInstallationSource(installationSource string) *DevtodevPackage {
+	p.installationSource = installationSource
+	return p
+}
+
+func (p *DevtodevPackage) Append(events ...Event) *DevtodevPackage {
+	if p == nil {
+		return nil
+	}
+	for _, evt := range events {
+		if evt == nil {
+			continue
+		}
+		p.events = append(p.events, evt)
+	}
+	return p
+}
+
+func (p *DevtodevPackage) Events() []Event {
+	if p == nil {
+		return nil
+	}
+	return p.events
+}
+
+func (p DevtodevPackage) MarshalJSON() ([]byte, error) {
+	type packageJSON struct {
+		Platform           string                   `json:"platform,omitempty"`
+		Language           string                   `json:"language,omitempty"`
+		Country            string                   `json:"country,omitempty"`
+		IP                 string                   `json:"ip,omitempty"`
+		AppVersion         string                   `json:"appVersion,omitempty"`
+		AppBuildVersion    string                   `json:"appBuildVersion,omitempty"`
+		SDKVersion         string                   `json:"sdkVersion,omitempty"`
+		Bundle             string                   `json:"bundle,omitempty"`
+		Engine             string                   `json:"engine,omitempty"`
+		InstallationSource string                   `json:"installationSource,omitempty"`
+		Events             []map[string]interface{} `json:"events"`
+	}
+
+	events := make([]map[string]interface{}, 0, len(p.events))
+	for _, evt := range p.events {
+		if evt == nil {
+			events = append(events, nil)
+			continue
+		}
+		events = append(events, evt.Payload())
+	}
+
+	return json.Marshal(packageJSON{
+		Platform:           p.platform,
+		Language:           p.language,
+		Country:            p.country,
+		IP:                 p.ip,
+		AppVersion:         p.appVersion,
+		AppBuildVersion:    p.appBuildVersion,
+		SDKVersion:         p.sdkVersion,
+		Bundle:             p.bundle,
+		Engine:             p.engine,
+		InstallationSource: p.installationSource,
+		Events:             events,
+	})
+}

pkg/devtodev/validation.go

@@ -0,0 +1,61 @@
+package devtodev
+
+import "fmt"
+
+// ValidatePayload checks for required fields and common schema errors.
+func ValidatePayload(payload Payload) error {
+	if len(payload.Reports) == 0 {
+		return fmt.Errorf("reports is required")
+	}
+	for ri, report := range payload.Reports {
+		if report.DeviceID == "" {
+			return fmt.Errorf("reports[%d].deviceId is required", ri)
+		}
+		if len(report.Packages) == 0 {
+			return fmt.Errorf("reports[%d].packages is required", ri)
+		}
+		for pi, pkg := range report.Packages {
+			events := pkg.Events()
+			if len(events) == 0 {
+				return fmt.Errorf("reports[%d].packages[%d].events is required", ri, pi)
+			}
+			for ei, event := range events {
+				if event == nil {
+					return fmt.Errorf("reports[%d].packages[%d].events[%d] is required", ri, pi, ei)
+				}
+				if err := event.Err(); err != nil {
+					return fmt.Errorf("reports[%d].packages[%d].events[%d] is invalid: %w", ri, pi, ei, err)
+				}
+				payload := event.Payload()
+				code, ok := payload["code"]
+				if !ok {
+					return fmt.Errorf("reports[%d].packages[%d].events[%d].code is required", ri, pi, ei)
+				}
+				if codeStr, ok := code.(string); !ok || codeStr == "" {
+					return fmt.Errorf("reports[%d].packages[%d].events[%d].code must be a non-empty string", ri, pi, ei)
+				}
+				ts, ok := payload["timestamp"]
+				if !ok {
+					return fmt.Errorf("reports[%d].packages[%d].events[%d].timestamp is required", ri, pi, ei)
+				}
+				if !isNumber(ts) {
+					return fmt.Errorf("reports[%d].packages[%d].events[%d].timestamp must be a number", ri, pi, ei)
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func isNumber(v interface{}) bool {
+	switch v.(type) {
+	case int, int8, int16, int32, int64:
+		return true
+	case uint, uint8, uint16, uint32, uint64:
+		return true
+	case float32, float64:
+		return true
+	default:
+		return false
+	}
+}

service/account/devtodev.go

@@ -0,0 +1,353 @@
+package account
+
+import (
+	"bygdata/global"
+	"bygdata/model/awssqs"
+	"context"
+	"encoding/json"
+	"fmt"
+	"runtime/debug"
+	"sync"
+	"time"
+
+	devtodev "devtodev-sdk"
+
+	"github.com/shopspring/decimal"
+	"go.uber.org/zap"
+)
+
+var (
+	devToDevOnce          sync.Once
+	devToDevClient        *devtodev.Client
+	devToDevSourceTypeSet map[int32]string
+	currentBalanceLastAt  sync.Map
+)
+
+const (
+	devToDevEventRegister              = "register"
+	devToDevEventWithdraw              = "withdraw"
+	devToDevWalletEventDeposit         = "deposit"
+	devToDevWalletEventWithdraw        = "withdraw"
+	devToDevWalletEventAccrual         = "currency_accrual"
+	devToDevWalletEventSpent           = "currency_spent"
+	devToDevWalletEventVirtualPayment  = "virtual_currency_payment"
+	devToDevDefaultCurrency            = "USD"
+	devToDevDefaultPlatform            = "web"
+	devToDevDefaultBalanceMinIntervalS = int64(86400)
+)
+
+func ReportUserBehaviorEvent(ctx context.Context, action awssqs.SqsAction, req awssqs.SqsActionUserBehaviorContent) {
+	defer recoverDevToDevPanic("user_behavior", req.Userno)
+
+	reporter := newDevToDevReporter(req.Userno, req.GetDeviceID())
+	if reporter == nil {
+		logUserBehaviorReporterSkip(req)
+		return
+	}
+	now := time.Now().UnixMilli()
+	pkg := baseDevtodevPackage(reporter, req.Ip)
+	switch action {
+	case awssqs.SqsActionUserBehaviorRegister:
+		deviceInfoEvent := devtodev.NewDeviceInfoEvent(now, buildDeviceInfoFields(req.Header))
+		pkg.Append(deviceInfoEvent)
+		parameters := map[string]interface{}{}
+		if req.Phone != "" {
+			parameters["phone"] = req.Phone
+		}
+		registerEvent := devtodev.NewCustomEvent(now, 1, devToDevEventRegister, parameters, buildCommonFields(action, "", ""))
+		pkg.Append(registerEvent)
+		if _, err := pkg.Report(ctx); err != nil {
+			logDevToDevError("register_bundle", req.Userno, err)
+		}
+	case awssqs.SqsActionUserBehaviorLogin:
+		sessionStartEvent := devtodev.NewSessionStartEvent(now, 1, buildCommonFields(action, "", ""))
+		pkg.Append(sessionStartEvent)
+		if _, err := pkg.Report(ctx); err != nil {
+			logDevToDevError("login_bundle", req.Userno, err)
+		}
+	}
+}
+
+func ReportWalletBalanceChangeEvent(ctx context.Context, req awssqs.SqsActionWalletBalanceChangeContent) {
+	defer recoverDevToDevPanic("wallet_balance_change", req.Userno)
+
+	reporter := newDevToDevReporter(req.Userno, req.GetDeviceID())
+	if reporter == nil {
+		logDevToDevSkip("wallet_balance_change", req.Userno, "reporter unavailable: missing app-id/userno/deviceId")
+		return
+	}
+
+	now := time.Now().UnixMilli()
+	amount, err := decimal.NewFromString(req.Amount)
+	if err != nil {
+		logDevToDevError("parse_amount", req.Userno, err)
+		return
+	}
+	pkg := baseDevtodevPackage(reporter, "")
+
+	if shouldReportCurrentBalance(req.Userno) {
+		if balance, ok := parseDecimalFloat(req.AfterBalance); ok {
+			balanceEvent := devtodev.NewCurrentBalanceEvent(now, 1, map[string]float64{defaultCurrency(): balance}, buildCommonFields(awssqs.SqsActionWalletBalanceChange, req.SourceId, req.RecordNo))
+			pkg.Append(balanceEvent)
+		} else {
+			logDevToDevSkip("current_balance", req.Userno, fmt.Sprintf("invalid afterBalance: %q", req.AfterBalance))
+		}
+	}
+
+	matchedEvent := false
+	switch walletSourceTypeKind(req.SourceType) {
+	case devToDevWalletEventDeposit:
+		matchedEvent = true
+		price, ok := decimalAbsFloat(amount)
+		if !ok {
+			logDevToDevSkip("real_currency_payment", req.Userno, fmt.Sprintf("invalid amount: %q", req.Amount))
+			return
+		}
+		paymentEvent := devtodev.NewRealPaymentEvent(now, 1, resolveProductID(req), req.SourceId, price, defaultCurrency(), buildCommonFields(awssqs.SqsActionWalletBalanceChange, req.SourceId, req.RecordNo))
+		pkg.Append(paymentEvent)
+	case devToDevWalletEventWithdraw:
+		matchedEvent = true
+		value, ok := decimalAbsFloat(amount)
+		if !ok {
+			logDevToDevSkip("withdraw", req.Userno, fmt.Sprintf("invalid amount: %q", req.Amount))
+			return
+		}
+		withdrawEvent := devtodev.NewCustomEvent(now, 1, devToDevEventWithdraw, map[string]interface{}{
+			"amount":   value,
+			"sourceId": req.SourceId,
+			"recordNo": req.RecordNo,
+		}, buildCommonFields(awssqs.SqsActionWalletBalanceChange, req.SourceId, req.RecordNo))
+		pkg.Append(withdrawEvent)
+	case devToDevWalletEventAccrual:
+		matchedEvent = true
+		value, ok := decimalAbsFloat(amount)
+		if !ok {
+			logDevToDevSkip("currency_accrual", req.Userno, fmt.Sprintf("invalid amount: %q", req.Amount))
+			return
+		}
+		accrualEvent := devtodev.NewCurrencyAccrualEvent(now, 1, nil, map[string]map[string]float64{
+			"default": {defaultCurrency(): value},
+		}, buildCommonFields(awssqs.SqsActionWalletBalanceChange, req.SourceId, req.RecordNo))
+		pkg.Append(accrualEvent)
+	case devToDevWalletEventVirtualPayment:
+		matchedEvent = true
+		value, ok := decimalAbsFloat(amount)
+		if !ok {
+			logDevToDevSkip("virtual_currency_payment", req.Userno, fmt.Sprintf("invalid amount: %q", req.Amount))
+			return
+		}
+		virtualPaymentEvent := devtodev.NewVirtualCurrencyPaymentEvent(now, 1, 1, map[string]float64{defaultCurrency(): value}, "wallet_source_type", fmt.Sprintf("%d", req.SourceType), buildCommonFields(awssqs.SqsActionWalletBalanceChange, req.SourceId, req.RecordNo))
+		pkg.Append(virtualPaymentEvent)
+	case devToDevWalletEventSpent:
+		matchedEvent = true
+		value, ok := decimalAbsFloat(amount)
+		if !ok {
+			logDevToDevSkip("currency_spent", req.Userno, fmt.Sprintf("invalid amount: %q", req.Amount))
+			return
+		}
+		spentEvent := devtodev.NewCustomEvent(now, 1, devToDevWalletEventSpent, map[string]interface{}{
+			"amount":     value,
+			"sourceId":   req.SourceId,
+			"recordNo":   req.RecordNo,
+			"gameId":     req.GameId,
+			"sourceType": req.SourceType,
+		}, buildCommonFields(awssqs.SqsActionWalletBalanceChange, req.SourceId, req.RecordNo))
+		pkg.Append(spentEvent)
+	}
+
+	if !matchedEvent {
+		logDevToDevSkip("wallet_balance_change", req.Userno, fmt.Sprintf("no sourceType mapping for sourceType=%d", req.SourceType))
+	}
+
+	if len(pkg.Events()) == 0 {
+		logDevToDevSkip("wallet_balance_change", req.Userno, "no valid events to report")
+		return
+	}
+	if _, err := pkg.Report(ctx); err != nil {
+		logDevToDevError("wallet_bundle", req.Userno, err)
+	}
+}
+
+func newDevToDevReporter(userno, deviceID string) *devtodev.Reporter {
+	client := getDevToDevClient()
+	if client == nil || userno == "" || deviceID == "" {
+		return nil
+	}
+	reporter := devtodev.NewReporter(client, deviceID)
+	reporter.UserID = userno
+	return reporter
+}
+
+func baseDevtodevPackage(reporter *devtodev.Reporter, ip string) *devtodev.DevtodevPackage {
+	return reporter.NewPackage().
+		WithPlatform(defaultPlatform()).
+		WithIP(ip).
+		WithAppVersion(global.GVA_CONFIG.DevToDev.AppVersion).
+		WithBundle(global.GVA_CONFIG.DevToDev.Bundle)
+}
+
+func getDevToDevClient() *devtodev.Client {
+	devToDevOnce.Do(func() {
+		cfg := global.GVA_CONFIG.DevToDev
+		if cfg.AppID == "" {
+			return
+		}
+		client := devtodev.NewClient(cfg.AppID)
+		devToDevClient = client
+		devToDevSourceTypeSet = make(map[int32]string)
+		registerSourceTypes(devToDevWalletEventDeposit, cfg.DepositSourceTypes)
+		registerSourceTypes(devToDevWalletEventWithdraw, cfg.WithdrawSourceTypes)
+		registerSourceTypes(devToDevWalletEventAccrual, cfg.CurrencyAccrualSourceTypes)
+		registerSourceTypes(devToDevWalletEventSpent, cfg.VirtualCurrencySpentSourceTypes)
+		registerSourceTypes(devToDevWalletEventVirtualPayment, cfg.VirtualCurrencyPaymentSourceTypes)
+	})
+	return devToDevClient
+}
+
+func registerSourceTypes(kind string, values []int32) {
+	for _, value := range values {
+		devToDevSourceTypeSet[value] = kind
+	}
+}
+
+func walletSourceTypeKind(sourceType int32) string {
+	if len(devToDevSourceTypeSet) == 0 {
+		return ""
+	}
+	return devToDevSourceTypeSet[sourceType]
+}
+
+func shouldReportCurrentBalance(userno string) bool {
+	interval := global.GVA_CONFIG.DevToDev.CurrentBalanceMinIntervalSeconds
+	if interval <= 0 {
+		interval = devToDevDefaultBalanceMinIntervalS
+	}
+	now := time.Now().Unix()
+	if value, ok := currentBalanceLastAt.Load(userno); ok {
+		lastAt, ok := value.(int64)
+		if ok && now-lastAt < interval {
+			return false
+		}
+	}
+	currentBalanceLastAt.Store(userno, now)
+	return true
+}
+
+func buildDeviceInfoFields(header string) map[string]interface{} {
+	if header == "" {
+		return nil
+	}
+	var values map[string]interface{}
+	if err := json.Unmarshal([]byte(header), &values); err == nil && len(values) > 0 {
+		return map[string]interface{}{"header": values}
+	}
+	return map[string]interface{}{"header_raw": header}
+}
+
+func buildCommonFields(action awssqs.SqsAction, sourceID, recordNo string) map[string]interface{} {
+	fields := map[string]interface{}{
+		"sqsAction": action.GetName(),
+	}
+	if sourceID != "" {
+		fields["sourceId"] = sourceID
+	}
+	if recordNo != "" {
+		fields["recordNo"] = recordNo
+	}
+	return fields
+}
+
+func resolveProductID(req awssqs.SqsActionWalletBalanceChangeContent) string {
+	if req.GameId != "" {
+		return req.GameId
+	}
+	if req.SourceId != "" {
+		return req.SourceId
+	}
+	return fmt.Sprintf("wallet_source_type_%d", req.SourceType)
+}
+
+func defaultCurrency() string {
+	if global.GVA_CONFIG.DevToDev.DefaultCurrency != "" {
+		return global.GVA_CONFIG.DevToDev.DefaultCurrency
+	}
+	return devToDevDefaultCurrency
+}
+
+func defaultPlatform() string {
+	if global.GVA_CONFIG.DevToDev.Platform != "" {
+		return global.GVA_CONFIG.DevToDev.Platform
+	}
+	return devToDevDefaultPlatform
+}
+
+func parseDecimalFloat(value string) (float64, bool) {
+	parsed, err := decimal.NewFromString(value)
+	if err != nil {
+		return 0, false
+	}
+	floatValue, _ := parsed.Float64()
+	return floatValue, true
+}
+
+func decimalAbsFloat(value decimal.Decimal) (float64, bool) {
+	floatValue, _ := value.Abs().Float64()
+	return floatValue, true
+}
+
+func logDevToDevError(eventName, userno string, err error) {
+	global.GVA_LOG.Warn("devtodev report failed", zap.String("event", eventName), zap.String("userno", userno), zap.Error(err))
+}
+
+func logDevToDevSkip(scene, userno, reason string) {
+	global.GVA_LOG.Warn("devtodev report skipped", zap.String("scene", scene), zap.String("userno", userno), zap.String("reason", reason))
+}
+
+func logUserBehaviorReporterSkip(req awssqs.SqsActionUserBehaviorContent) {
+	switch {
+	case global.GVA_CONFIG.DevToDev.AppID == "":
+		logDevToDevSkip("user_behavior", req.Userno, "reporter unavailable: missing app-id")
+	case req.Userno == "":
+		logDevToDevSkip("user_behavior", req.Userno, "reporter unavailable: missing userno")
+	case req.Header == "":
+		global.GVA_LOG.Warn(
+			"devtodev report skipped",
+			zap.String("scene", "user_behavior"),
+			zap.String("userno", req.Userno),
+			zap.String("reason", "reporter unavailable: missing deviceId, header is empty"),
+			zap.Strings("deviceIdCandidateKeys", deviceIDCandidateKeys()),
+		)
+	default:
+		global.GVA_LOG.Warn(
+			"devtodev report skipped",
+			zap.String("scene", "user_behavior"),
+			zap.String("userno", req.Userno),
+			zap.String("reason", "reporter unavailable: missing deviceId in header"),
+			zap.Strings("deviceIdCandidateKeys", deviceIDCandidateKeys()),
+			zap.String("headerPreview", truncateForLog(req.Header, 512)),
+		)
+	}
+}
+
+func deviceIDCandidateKeys() []string {
+	return []string{"deviceId", "deviceID", "device_id", "device-id", "x-device-id", "X-Device-Id", "X-DEVICE-ID"}
+}
+
+func truncateForLog(value string, max int) string {
+	if max <= 0 || len(value) <= max {
+		return value
+	}
+	return value[:max] + "...(truncated)"
+}
+
+func recoverDevToDevPanic(scene, userno string) {
+	if r := recover(); r != nil {
+		global.GVA_LOG.Error(
+			"devtodev report panic recovered",
+			zap.String("scene", scene),
+			zap.String("userno", userno),
+			zap.Any("panic", r),
+			zap.ByteString("stack", debug.Stack()),
+		)
+	}
+}

work_flow/map_event.md

@@ -0,0 +1,31 @@
+# 200-205 到 devtodev 映射表
+
+基于当前仓库信息整理，依据主要来自 `model/awssqs/sqs.go` 和 `pkg/devtodev/report_event.md`。
+
+## 总表
+
+| SQS Action | 含义 | 推荐映射到 devtodev | 类型 | 是否满足需求 | 备注 |
+|---|---|---|---|---|---|
+| `200` `SqsActionUserBehaviorRegister` | 注册 | `register` | 自定义事件 | `是，部分满足` | 同时建议补发 `device_info`，因为新用户首事件应先有设备信息 |
+| `201` `SqsActionUserBehaviorLogin` | 登录 | `session_start` | 预设事件 | `是，部分满足` | 只能覆盖“会话开始”；若要统计会话时长，还要配合 `user_engagement` |
+| `202` `SqsActionUserBehaviorEditPassword` | 修改登录密码 | `不映射` | 无 | `否` | `report_event.md` 里没有这个需求，除非额外埋成自定义事件如 `edit_password` |
+| `203` `SqsActionUserBehaviorEditPayPassword` | 修改支付密码 | `不映射` | 无 | `否` | 当前需求里没有 |
+| `204` `SqsActionUserBehaviorUpdateWallet` | 更新钱包地址/绑定银行卡 | `不映射` | 无 | `否` | 当前需求里没有；如果要分析绑卡，可单独加自定义事件 |
+| `205` `SqsActionWalletBalanceChange` | 余额变更 | `按来源拆分` | 混合 | `部分满足` | 不能直接统一映射成一个 devtodev 事件，必须按 `sourceType` 或业务场景拆 |
+
+## `205` 拆分建议
+
+| `205` 子场景 | 推荐映射到 devtodev | 类型 | 需要字段 |
+|---|---|---|---|
+| 充值成功，法币/真实支付入账 | `real_currency_payment` | 预设事件 | `orderId`、`price`、`currencyCode`、`productId` |
+| 提现成功 | `withdraw` | 自定义事件 | 建议参数：`amount`、`sourceId`、`recordNo` |
+| 游戏赢币、活动奖励、补偿、赠送 | `currency_accrual` | 预设事件 | 需要明确币种和金额 |
+| 下注、扣减虚拟币 | `currency_spent` | 预设事件 | 需要明确币种和金额 |
+| 每次账变后同步当前余额 | `current_balance` | 预设事件 | 需要当前余额；但文档建议不要对同一用户一天上报超过一次 |
+| 用虚拟币购买 VIP/礼包/道具 | `virtual_currency_payment` | 预设事件 | `purchaseAmount`、`purchasePrice`、`purchaseType`、`purchaseId` |
+
+## 结论
+
+- `200-205` 里能直接对上需求的主要是 `200`、`201`，以及需要拆分后的 `205`
+- `token_refresh`、`enter_game`、`custom_support` 不在 `200-205` 范围内，需要从其他业务入口补埋点
+- 当前代码里这组事件在 SQS 消费后只是落库日志，还没有真正接到 devtodev 上报