127 lines
4.0 KiB
Go
127 lines
4.0 KiB
Go
package walletnotice
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"hyapp/pkg/tencentim"
|
|
)
|
|
|
|
func TestProcessWalletBalanceNoticesPublishesPrivateIM(t *testing.T) {
|
|
repo := &fakeRepository{
|
|
events: []WalletBalanceEvent{{
|
|
AppCode: "lalu",
|
|
EventID: "evt-1",
|
|
EventType: "WalletBalanceChanged",
|
|
TransactionID: "tx-1",
|
|
CommandID: "cmd-1",
|
|
UserID: 123,
|
|
AssetType: "COIN",
|
|
AvailableDelta: -10,
|
|
FrozenDelta: 0,
|
|
PayloadJSON: `{"available_after":90,"frozen_after":0,"balance_version":7}`,
|
|
CreatedAtMS: 1710000000000,
|
|
}},
|
|
}
|
|
publisher := &fakePublisher{}
|
|
service := New(Config{NodeID: "node-a"}, repo, publisher)
|
|
|
|
processed, err := service.ProcessWalletBalanceNotices(context.Background(), WalletNoticeWorkerOptions{MaxRetryCount: 3})
|
|
if err != nil {
|
|
t.Fatalf("ProcessWalletBalanceNotices failed: %v", err)
|
|
}
|
|
if processed != 1 || len(repo.delivered) != 1 || len(publisher.messages) != 1 {
|
|
t.Fatalf("unexpected process result: processed=%d delivered=%d messages=%d", processed, len(repo.delivered), len(publisher.messages))
|
|
}
|
|
message := publisher.messages[0]
|
|
if message.ToAccount != "123" || message.EventID != "evt-1" || message.Ext != "wallet_notice" {
|
|
t.Fatalf("unexpected c2c message: %+v", message)
|
|
}
|
|
var payload map[string]any
|
|
if err := json.Unmarshal(message.PayloadJSON, &payload); err != nil {
|
|
t.Fatalf("decode payload: %v", err)
|
|
}
|
|
if payload["event_id"] != "evt-1" || payload["asset_type"] != "COIN" || payload["balance_version"].(float64) != 7 {
|
|
t.Fatalf("unexpected wallet payload: %+v", payload)
|
|
}
|
|
}
|
|
|
|
func TestProcessWalletBalanceNoticesMarksRetryable(t *testing.T) {
|
|
repo := &fakeRepository{
|
|
events: []WalletBalanceEvent{{
|
|
AppCode: "lalu",
|
|
EventID: "evt-2",
|
|
EventType: "WalletBalanceChanged",
|
|
UserID: 123,
|
|
AssetType: "COIN",
|
|
PayloadJSON: `{"available_after":90,"balance_version":8}`,
|
|
RetryCount: 0,
|
|
}},
|
|
}
|
|
service := New(Config{NodeID: "node-a"}, repo, &fakePublisher{err: errors.New("im timeout")})
|
|
|
|
processed, err := service.ProcessWalletBalanceNotices(context.Background(), WalletNoticeWorkerOptions{
|
|
MaxRetryCount: 3,
|
|
InitialBackoff: time.Second,
|
|
MaxBackoff: time.Minute,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("ProcessWalletBalanceNotices failed: %v", err)
|
|
}
|
|
if processed != 1 || len(repo.retryable) != 1 {
|
|
t.Fatalf("unexpected retry result: processed=%d retryable=%d", processed, len(repo.retryable))
|
|
}
|
|
if repo.retryable[0].retryCount != 1 || repo.retryable[0].nextRetryAtMS <= 0 {
|
|
t.Fatalf("unexpected retry marker: %+v", repo.retryable[0])
|
|
}
|
|
}
|
|
|
|
type fakeRepository struct {
|
|
events []WalletBalanceEvent
|
|
delivered []WalletBalanceEvent
|
|
retryable []retryMarker
|
|
failed []WalletBalanceEvent
|
|
}
|
|
|
|
type retryMarker struct {
|
|
event WalletBalanceEvent
|
|
retryCount int
|
|
nextRetryAtMS int64
|
|
lastErr string
|
|
}
|
|
|
|
func (r *fakeRepository) ClaimWalletBalanceEvents(context.Context, string, int, time.Duration) ([]WalletBalanceEvent, error) {
|
|
return r.events, nil
|
|
}
|
|
|
|
func (r *fakeRepository) MarkWalletBalanceDelivered(_ context.Context, event WalletBalanceEvent, _ json.RawMessage, _ int64) error {
|
|
r.delivered = append(r.delivered, event)
|
|
return nil
|
|
}
|
|
|
|
func (r *fakeRepository) MarkWalletBalanceRetryable(_ context.Context, event WalletBalanceEvent, retryCount int, nextRetryAtMS int64, lastErr string, _ int64) error {
|
|
r.retryable = append(r.retryable, retryMarker{event: event, retryCount: retryCount, nextRetryAtMS: nextRetryAtMS, lastErr: lastErr})
|
|
return nil
|
|
}
|
|
|
|
func (r *fakeRepository) MarkWalletBalanceFailed(_ context.Context, event WalletBalanceEvent, _ int, _ string, _ int64) error {
|
|
r.failed = append(r.failed, event)
|
|
return nil
|
|
}
|
|
|
|
type fakePublisher struct {
|
|
messages []tencentim.CustomUserMessage
|
|
err error
|
|
}
|
|
|
|
func (p *fakePublisher) PublishUserCustomMessage(_ context.Context, message tencentim.CustomUserMessage) error {
|
|
if p.err != nil {
|
|
return p.err
|
|
}
|
|
p.messages = append(p.messages, message)
|
|
return nil
|
|
}
|