feat: add devtodev event report #2
@ -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 {
|
||||
|
||||
@ -176,3 +176,16 @@ aws:
|
||||
aws-sqs-access-key: AKIAUWKJ5EVVM2APLKGR
|
||||
aws-sqs-secret-key: JYJRe2S1vpQvbrzy8gVp5OABXoJVZXePnwvCbhKe
|
||||
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: []
|
||||
|
||||
13
config.yaml
13
config.yaml
@ -176,3 +176,16 @@ aws:
|
||||
aws-sqs-access-key: AKIAUWKJ5EVVM2APLKGR
|
||||
aws-sqs-secret-key: JYJRe2S1vpQvbrzy8gVp5OABXoJVZXePnwvCbhKe
|
||||
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: []
|
||||
|
||||
@ -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"`
|
||||
}
|
||||
|
||||
15
config/devtodev.go
Normal file
15
config/devtodev.go
Normal file
@ -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"`
|
||||
}
|
||||
5
go.mod
5
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
|
||||
|
||||
2
go.sum
2
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=
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -25,12 +25,11 @@ func main() {
|
||||
Reports: []devtodev.Report{
|
||||
{
|
||||
DeviceID: "device-123",
|
||||
Packages: []devtodev.Package{
|
||||
{
|
||||
Language: "en",
|
||||
Country: "US",
|
||||
Events: []devtodev.Event{
|
||||
{
|
||||
Packages: []devtodev.DevtodevPackage{
|
||||
*devtodev.NewReporter(nil, "device-123").NewPackage().
|
||||
WithLanguage("en").
|
||||
WithCountry("US").
|
||||
Append(devtodev.RawEvent{
|
||||
"code": "ce",
|
||||
"timestamp": time.Now().UnixMilli(),
|
||||
"level": 5,
|
||||
@ -38,9 +37,7 @@ func main() {
|
||||
"parameters": map[string]interface{}{
|
||||
"level": 5,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}),
|
||||
},
|
||||
},
|
||||
},
|
||||
@ -52,29 +49,31 @@ func main() {
|
||||
}
|
||||
```
|
||||
|
||||
## Event Helpers
|
||||
## Event Builder
|
||||
|
||||
This SDK includes a `Reporter` wrapper. Each event has its own function and sends a single report.
|
||||
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")
|
||||
reporter.Package = devtodev.Package{
|
||||
Language: "en",
|
||||
Country: "US",
|
||||
}
|
||||
pkg := reporter.NewPackage().
|
||||
WithPlatform("web").
|
||||
WithLanguage("en").
|
||||
WithCountry("US").
|
||||
WithIP("127.0.0.1").
|
||||
WithAppVersion("1.0.0")
|
||||
|
||||
// Device Info (di)
|
||||
_, _ = reporter.DeviceInfo(context.Background(), time.Now().UnixMilli(), map[string]interface{}{
|
||||
pkg.Append(devtodev.NewDeviceInfoEvent(time.Now().UnixMilli(), map[string]interface{}{
|
||||
"platform": "ios",
|
||||
"device": "iPhone14,3",
|
||||
})
|
||||
}))
|
||||
|
||||
// Custom Event (ce)
|
||||
_, _ = reporter.CustomEvent(context.Background(), time.Now().UnixMilli(), 1, "custom_event", map[string]interface{}{
|
||||
pkg.Append(devtodev.NewCustomEvent(time.Now().UnixMilli(), 1, "custom_event", map[string]interface{}{
|
||||
"score": 123,
|
||||
}, nil)
|
||||
}, nil))
|
||||
|
||||
_, _ = pkg.Report(context.Background())
|
||||
```
|
||||
|
||||
## 中文事件文档
|
||||
|
||||
@ -25,36 +25,31 @@ func main() {
|
||||
level := rand.Intn(20) + 1
|
||||
|
||||
reporter := devtodev.NewReporter(client, deviceID)
|
||||
reporter.Package = devtodev.Package{
|
||||
Language: "en",
|
||||
Country: "US",
|
||||
}
|
||||
pkg := reporter.NewPackage().
|
||||
WithLanguage("en").
|
||||
WithCountry("US")
|
||||
|
||||
if _, err := reporter.DeviceInfo(ctx, base.UnixMilli(), map[string]interface{}{
|
||||
deviceInfoEvent := devtodev.NewDeviceInfoEvent(base.UnixMilli(), map[string]interface{}{
|
||||
"platform": "ios",
|
||||
"device": "iPhone14,3",
|
||||
}); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
})
|
||||
pkg.Append(deviceInfoEvent)
|
||||
|
||||
if _, err := reporter.SessionStart(ctx, base.Add(2*time.Second).UnixMilli(), level, nil); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
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++ {
|
||||
if _, err := reporter.UserEngagement(ctx, base.Add(time.Duration(10*(i+1))*time.Second).UnixMilli(), level, 10, nil); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
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)
|
||||
if _, err := reporter.RealPayment(
|
||||
ctx,
|
||||
paymentEvent := devtodev.NewRealPaymentEvent(
|
||||
base.Add(time.Duration(60+(i*10))*time.Second).UnixMilli(),
|
||||
level,
|
||||
fmt.Sprintf("com.demo.product.%d", rand.Intn(5)+1),
|
||||
@ -62,9 +57,12 @@ func main() {
|
||||
price,
|
||||
"USD",
|
||||
nil,
|
||||
); err != nil {
|
||||
panic(err)
|
||||
)
|
||||
pkg.Append(paymentEvent)
|
||||
}
|
||||
|
||||
if _, err := pkg.Report(ctx); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -151,28 +151,28 @@ func TestValidatePayload(t *testing.T) {
|
||||
{
|
||||
name: "missing events",
|
||||
payload: Payload{
|
||||
Reports: []Report{{DeviceID: "d1", Packages: []Package{{}}}},
|
||||
Reports: []Report{{DeviceID: "d1", Packages: []DevtodevPackage{{}}}},
|
||||
},
|
||||
ok: false,
|
||||
},
|
||||
{
|
||||
name: "missing code",
|
||||
payload: Payload{
|
||||
Reports: []Report{{DeviceID: "d1", Packages: []Package{{Events: []Event{{"timestamp": int64(1)}}}}}},
|
||||
Reports: []Report{{DeviceID: "d1", Packages: []DevtodevPackage{{events: []Event{RawEvent{"timestamp": int64(1)}}}}}},
|
||||
},
|
||||
ok: false,
|
||||
},
|
||||
{
|
||||
name: "missing timestamp",
|
||||
payload: Payload{
|
||||
Reports: []Report{{DeviceID: "d1", Packages: []Package{{Events: []Event{{"code": "ce"}}}}}},
|
||||
Reports: []Report{{DeviceID: "d1", Packages: []DevtodevPackage{{events: []Event{RawEvent{"code": "ce"}}}}}},
|
||||
},
|
||||
ok: false,
|
||||
},
|
||||
{
|
||||
name: "ok",
|
||||
payload: Payload{
|
||||
Reports: []Report{{DeviceID: "d1", Packages: []Package{{Events: []Event{{"code": "ce", "timestamp": int64(1)}}}}}},
|
||||
Reports: []Report{{DeviceID: "d1", Packages: []DevtodevPackage{{events: []Event{RawEvent{"code": "ce", "timestamp": int64(1)}}}}}},
|
||||
},
|
||||
ok: true,
|
||||
},
|
||||
@ -226,12 +226,12 @@ func samplePayload() Payload {
|
||||
Reports: []Report{
|
||||
{
|
||||
DeviceID: "device-123",
|
||||
Packages: []Package{
|
||||
{
|
||||
Language: "en",
|
||||
Country: "US",
|
||||
Events: []Event{
|
||||
Packages: []DevtodevPackage{
|
||||
{
|
||||
language: "en",
|
||||
country: "US",
|
||||
events: []Event{
|
||||
RawEvent{
|
||||
"code": "ce",
|
||||
"timestamp": time.Now().UnixMilli(),
|
||||
"level": 5,
|
||||
|
||||
@ -7,8 +7,7 @@ import (
|
||||
"devtodev-sdk/event"
|
||||
)
|
||||
|
||||
// Reporter wraps Client with shared report/package fields and per-event helpers.
|
||||
// Each event helper sends exactly one report with one package.
|
||||
// Reporter wraps Client with report-level identity fields.
|
||||
type Reporter struct {
|
||||
Client *Client
|
||||
DeviceID string
|
||||
@ -16,10 +15,9 @@ type Reporter struct {
|
||||
PreviousDeviceID string
|
||||
PreviousUserID string
|
||||
DevtodevID string
|
||||
Package Package
|
||||
}
|
||||
|
||||
// NewReporter creates a Reporter with a client and device ID.
|
||||
// NewReporter creates a Reporter with report-level fields.
|
||||
func NewReporter(client *Client, deviceID string) *Reporter {
|
||||
return &Reporter{
|
||||
Client: client,
|
||||
@ -27,8 +25,16 @@ func NewReporter(client *Client, deviceID string) *Reporter {
|
||||
}
|
||||
}
|
||||
|
||||
// Report sends a single event using the current reporter context.
|
||||
func (r *Reporter) Report(ctx context.Context, event Event) (*Response, error) {
|
||||
// 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")
|
||||
}
|
||||
@ -38,8 +44,19 @@ func (r *Reporter) Report(ctx context.Context, event Event) (*Response, error) {
|
||||
if r.DeviceID == "" {
|
||||
return nil, fmt.Errorf("deviceId is required")
|
||||
}
|
||||
pkg := r.Package
|
||||
pkg.Events = []Event{event}
|
||||
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{
|
||||
{
|
||||
@ -48,171 +65,99 @@ func (r *Reporter) Report(ctx context.Context, event Event) (*Response, error) {
|
||||
PreviousDeviceID: r.PreviousDeviceID,
|
||||
PreviousUserID: r.PreviousUserID,
|
||||
DevtodevID: r.DevtodevID,
|
||||
Packages: []Package{pkg},
|
||||
Packages: rawPackages,
|
||||
},
|
||||
},
|
||||
}
|
||||
return r.Client.SendWithResponse(ctx, payload)
|
||||
}
|
||||
|
||||
// DeviceInfo (code "di").
|
||||
func (r *Reporter) DeviceInfo(ctx context.Context, timestamp int64, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.DeviceInfo(timestamp, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
// 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")
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
if p.reporter == nil {
|
||||
return nil, fmt.Errorf("package reporter is nil")
|
||||
}
|
||||
return p.reporter.Report(ctx, p)
|
||||
}
|
||||
|
||||
// SessionStart (code "ss").
|
||||
func (r *Reporter) SessionStart(ctx context.Context, timestamp int64, level int, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.SessionStart(timestamp, level, fields)
|
||||
func wrapEvent(raw map[string]interface{}, err error) Event {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return builtEvent{err: err}
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
return builtEvent{payload: raw}
|
||||
}
|
||||
|
||||
// UserEngagement (code "ue").
|
||||
func (r *Reporter) UserEngagement(ctx context.Context, timestamp int64, level int, length int, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.UserEngagement(timestamp, level, length, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
func NewDeviceInfoEvent(timestamp int64, fields map[string]interface{}) Event {
|
||||
return wrapEvent(event.DeviceInfo(timestamp, fields))
|
||||
}
|
||||
|
||||
// TrackingStatus (GDPR) (code "ts").
|
||||
func (r *Reporter) TrackingStatus(ctx context.Context, timestamp int64, trackingAllowed bool, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.TrackingStatus(timestamp, trackingAllowed, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
func NewSessionStartEvent(timestamp int64, level int, fields map[string]interface{}) Event {
|
||||
return wrapEvent(event.SessionStart(timestamp, level, fields))
|
||||
}
|
||||
|
||||
// Alive (code "al").
|
||||
func (r *Reporter) Alive(ctx context.Context, timestamp int64, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.Alive(timestamp, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
func NewUserEngagementEvent(timestamp int64, level int, length int, fields map[string]interface{}) Event {
|
||||
return wrapEvent(event.UserEngagement(timestamp, level, length, fields))
|
||||
}
|
||||
|
||||
// People (user properties) (code "pl").
|
||||
func (r *Reporter) People(ctx context.Context, timestamp int64, level int, properties map[string]interface{}, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.People(timestamp, level, properties, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
func NewTrackingStatusEvent(timestamp int64, trackingAllowed bool, fields map[string]interface{}) Event {
|
||||
return wrapEvent(event.TrackingStatus(timestamp, trackingAllowed, fields))
|
||||
}
|
||||
|
||||
// CustomEvent (code "ce").
|
||||
func (r *Reporter) CustomEvent(ctx context.Context, timestamp int64, level int, name string, parameters map[string]interface{}, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.CustomEvent(timestamp, level, name, parameters, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
func NewAliveEvent(timestamp int64, fields map[string]interface{}) Event {
|
||||
return wrapEvent(event.Alive(timestamp, fields))
|
||||
}
|
||||
|
||||
// RealPayment (code "rp").
|
||||
func (r *Reporter) RealPayment(ctx context.Context, timestamp int64, level int, productID, orderID string, price float64, currencyCode string, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.RealPayment(timestamp, level, productID, orderID, price, currencyCode, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
func NewPeopleEvent(timestamp int64, level int, properties map[string]interface{}, fields map[string]interface{}) Event {
|
||||
return wrapEvent(event.People(timestamp, level, properties, fields))
|
||||
}
|
||||
|
||||
// Onboarding (tutorial) (code "tr").
|
||||
func (r *Reporter) Onboarding(ctx context.Context, timestamp int64, level int, step int, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.Onboarding(timestamp, level, step, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
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))
|
||||
}
|
||||
|
||||
// VirtualCurrencyPayment (code "vp").
|
||||
func (r *Reporter) VirtualCurrencyPayment(ctx context.Context, timestamp int64, level int, purchaseAmount int, purchasePrice map[string]float64, purchaseType, purchaseID string, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.VirtualCurrencyPayment(timestamp, level, purchaseAmount, purchasePrice, purchaseType, purchaseID, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
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))
|
||||
}
|
||||
|
||||
// CurrencyAccrual (code "ca"). At least one of bought or earned must be provided.
|
||||
func (r *Reporter) CurrencyAccrual(ctx context.Context, timestamp int64, level int, bought, earned map[string]map[string]float64, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.CurrencyAccrual(timestamp, level, bought, earned, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
func NewOnboardingEvent(timestamp int64, level int, step int, fields map[string]interface{}) Event {
|
||||
return wrapEvent(event.Onboarding(timestamp, level, step, fields))
|
||||
}
|
||||
|
||||
// CurrentBalance (code "cb").
|
||||
func (r *Reporter) CurrentBalance(ctx context.Context, timestamp int64, level int, balance map[string]float64, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.CurrentBalance(timestamp, level, balance, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
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))
|
||||
}
|
||||
|
||||
// LevelUp (code "lu").
|
||||
func (r *Reporter) LevelUp(ctx context.Context, timestamp int64, level int, balance, spent, earned, bought map[string]float64, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.LevelUp(timestamp, level, balance, spent, earned, bought, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
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))
|
||||
}
|
||||
|
||||
// ProgressionEvent (code "pe"). "parameters" is required and should include success and duration.
|
||||
func (r *Reporter) ProgressionEvent(ctx context.Context, timestamp int64, level int, name string, parameters map[string]interface{}, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.ProgressionEvent(timestamp, level, name, parameters, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
func NewCurrentBalanceEvent(timestamp int64, level int, balance map[string]float64, fields map[string]interface{}) Event {
|
||||
return wrapEvent(event.CurrentBalance(timestamp, level, balance, fields))
|
||||
}
|
||||
|
||||
// Referral (code "rf").
|
||||
func (r *Reporter) Referral(ctx context.Context, timestamp int64, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.Referral(timestamp, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
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))
|
||||
}
|
||||
|
||||
// AdImpression (code "adrv").
|
||||
func (r *Reporter) AdImpression(ctx context.Context, timestamp int64, adNetwork string, revenue float64, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.AdImpression(timestamp, adNetwork, revenue, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
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))
|
||||
}
|
||||
|
||||
// SocialConnect (code "sc").
|
||||
func (r *Reporter) SocialConnect(ctx context.Context, timestamp int64, level int, socialNetwork string, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.SocialConnect(timestamp, level, socialNetwork, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
func NewReferralEvent(timestamp int64, fields map[string]interface{}) Event {
|
||||
return wrapEvent(event.Referral(timestamp, fields))
|
||||
}
|
||||
|
||||
// SocialPost (code "sp").
|
||||
func (r *Reporter) SocialPost(ctx context.Context, timestamp int64, level int, socialNetwork, postReason string, fields map[string]interface{}) (*Response, error) {
|
||||
e, err := event.SocialPost(timestamp, level, socialNetwork, postReason, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Report(ctx, Event(e))
|
||||
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))
|
||||
}
|
||||
|
||||
@ -1,5 +1,7 @@
|
||||
package devtodev
|
||||
|
||||
import "encoding/json"
|
||||
|
||||
// Payload is the top-level request body for Data API 2.0.
|
||||
type Payload struct {
|
||||
Reports []Report `json:"reports"`
|
||||
@ -12,25 +14,161 @@ type Report struct {
|
||||
PreviousDeviceID string `json:"previousDeviceId,omitempty"`
|
||||
PreviousUserID string `json:"previousUserId,omitempty"`
|
||||
DevtodevID string `json:"devtodevId,omitempty"`
|
||||
Packages []Package `json:"packages"`
|
||||
Packages []DevtodevPackage `json:"packages"`
|
||||
}
|
||||
|
||||
// Package groups events under a locale and metadata.
|
||||
type Package struct {
|
||||
// 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"`
|
||||
Platform string `json:"platform,omitempty"`
|
||||
IP string `json:"ip,omitempty"`
|
||||
AppVersion string `json:"appVersion,omitempty"`
|
||||
AppBuildVersion string `json:"appBuildVersion,omitempty"`
|
||||
SDKVersion string `json:"sdkVersion,omitempty"`
|
||||
SDKCodeVersion int `json:"sdkCodeVersion,omitempty"`
|
||||
Bundle string `json:"bundle,omitempty"`
|
||||
InstallationSource string `json:"installationSource,omitempty"`
|
||||
Engine string `json:"engine,omitempty"`
|
||||
Events []Event `json:"events"`
|
||||
}
|
||||
InstallationSource string `json:"installationSource,omitempty"`
|
||||
Events []map[string]interface{} `json:"events"`
|
||||
}
|
||||
|
||||
// Event is a raw devtodev event object.
|
||||
// Use the Data API 2.0 event schema when building this map.
|
||||
type Event map[string]interface{}
|
||||
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,
|
||||
})
|
||||
}
|
||||
|
||||
@ -15,18 +15,26 @@ func ValidatePayload(payload Payload) error {
|
||||
return fmt.Errorf("reports[%d].packages is required", ri)
|
||||
}
|
||||
for pi, pkg := range report.Packages {
|
||||
if len(pkg.Events) == 0 {
|
||||
events := pkg.Events()
|
||||
if len(events) == 0 {
|
||||
return fmt.Errorf("reports[%d].packages[%d].events is required", ri, pi)
|
||||
}
|
||||
for ei, event := range pkg.Events {
|
||||
code, ok := event["code"]
|
||||
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 := event["timestamp"]
|
||||
ts, ok := payload["timestamp"]
|
||||
if !ok {
|
||||
return fmt.Errorf("reports[%d].packages[%d].events[%d].timestamp is required", ri, pi, ei)
|
||||
}
|
||||
|
||||
353
service/account/devtodev.go
Normal file
353
service/account/devtodev.go
Normal file
@ -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()),
|
||||
)
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user