hy/hyapp-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix databi statistics dimensions and backfill

Browse Source
This commit is contained in:
zhx 2026-06-08 12:01:55 +08:00
parent 1b6c77c6dc
commit b49a2109c7
25 changed files with 9164 additions and 5154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

914
api/proto/events/room/v1/events.pb.go
View File

File diff suppressed because it is too large Load Diff

4
api/proto/events/room/v1/events.proto
View File

@ -63,6 +63,8 @@ message RoomUserJoined {
// visible_region_id 是用户进房时所在房间的国家/区域桶，统计服务不能反查 room-service。
int64 visible_region_id = 3;
RoomEntryVehicleSnapshot entry_vehicle = 4;
int64 country_id = 5;
int64 region_id = 6;
}
// RoomUserLeft 表达用户离房成功。
@ -154,6 +156,8 @@ message RoomGiftSent {
string command_id = 8;
string pool_id = 9;
int64 coin_spent = 10;
int64 country_id = 11;
int64 region_id = 12;
}
// RoomHeatChanged 表达热度变化结果。

4502
api/proto/room/v1/room.pb.go
View File

File diff suppressed because it is too large Load Diff

4
api/proto/room/v1/room.proto
View File

@ -468,6 +468,8 @@ message JoinRoomRequest {
// password 只用于本次入房校验；room-service 不把明文写入 command log 或快照。
string password = 3;
RoomEntryVehicleSnapshot entry_vehicle = 4;
int64 actor_country_id = 5;
int64 actor_region_id = 6;
}
// JoinRoomResponse 返回加入后的房间快照。
@ -841,6 +843,8 @@ message SendGiftRequest {
int64 target_agency_owner_user_id = 10;
// sender_region_id 是 gateway 从 user-service 当前送礼用户资料注入的区域，用于 wallet 匹配礼物钻石比例。
int64 sender_region_id = 11;
// sender_country_id 是 gateway 从 user-service 当前送礼用户资料注入的真实国家，用于统计国家维度。
int64 sender_country_id = 12;
}
// SendGiftResponse 返回扣费成功并落到房间后的状态结果。

106
api/proto/room/v1/room_grpc.pb.go
View File

@ -1,6 +1,6 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.6.2
// - protoc-gen-go-grpc v1.5.1
// - protoc v7.35.0
// source: proto/room/v1/room.proto
@ -470,100 +470,100 @@ type RoomCommandServiceServer interface {
type UnimplementedRoomCommandServiceServer struct{}
func (UnimplementedRoomCommandServiceServer) CreateRoom(context.Context, *CreateRoomRequest) (*CreateRoomResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CreateRoom not implemented")
return nil, status.Errorf(codes.Unimplemented, "method CreateRoom not implemented")
}
func (UnimplementedRoomCommandServiceServer) UpdateRoomProfile(context.Context, *UpdateRoomProfileRequest) (*UpdateRoomProfileResponse, error) {
return nil, status.Error(codes.Unimplemented, "method UpdateRoomProfile not implemented")
return nil, status.Errorf(codes.Unimplemented, "method UpdateRoomProfile not implemented")
}
func (UnimplementedRoomCommandServiceServer) SaveRoomBackground(context.Context, *SaveRoomBackgroundRequest) (*SaveRoomBackgroundResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SaveRoomBackground not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SaveRoomBackground not implemented")
}
func (UnimplementedRoomCommandServiceServer) SetRoomBackground(context.Context, *SetRoomBackgroundRequest) (*SetRoomBackgroundResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SetRoomBackground not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SetRoomBackground not implemented")
}
func (UnimplementedRoomCommandServiceServer) JoinRoom(context.Context, *JoinRoomRequest) (*JoinRoomResponse, error) {
return nil, status.Error(codes.Unimplemented, "method JoinRoom not implemented")
return nil, status.Errorf(codes.Unimplemented, "method JoinRoom not implemented")
}
func (UnimplementedRoomCommandServiceServer) RoomHeartbeat(context.Context, *RoomHeartbeatRequest) (*RoomHeartbeatResponse, error) {
return nil, status.Error(codes.Unimplemented, "method RoomHeartbeat not implemented")
return nil, status.Errorf(codes.Unimplemented, "method RoomHeartbeat not implemented")
}
func (UnimplementedRoomCommandServiceServer) LeaveRoom(context.Context, *LeaveRoomRequest) (*LeaveRoomResponse, error) {
return nil, status.Error(codes.Unimplemented, "method LeaveRoom not implemented")
return nil, status.Errorf(codes.Unimplemented, "method LeaveRoom not implemented")
}
func (UnimplementedRoomCommandServiceServer) CloseRoom(context.Context, *CloseRoomRequest) (*CloseRoomResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CloseRoom not implemented")
return nil, status.Errorf(codes.Unimplemented, "method CloseRoom not implemented")
}
func (UnimplementedRoomCommandServiceServer) AdminUpdateRoom(context.Context, *AdminUpdateRoomRequest) (*AdminUpdateRoomResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminUpdateRoom not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminUpdateRoom not implemented")
}
func (UnimplementedRoomCommandServiceServer) AdminDeleteRoom(context.Context, *AdminDeleteRoomRequest) (*AdminDeleteRoomResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminDeleteRoom not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminDeleteRoom not implemented")
}
func (UnimplementedRoomCommandServiceServer) AdminUpdateRoomRocketConfig(context.Context, *AdminUpdateRoomRocketConfigRequest) (*AdminUpdateRoomRocketConfigResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminUpdateRoomRocketConfig not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminUpdateRoomRocketConfig not implemented")
}
func (UnimplementedRoomCommandServiceServer) AdminUpdateRoomSeatConfig(context.Context, *AdminUpdateRoomSeatConfigRequest) (*AdminUpdateRoomSeatConfigResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminUpdateRoomSeatConfig not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminUpdateRoomSeatConfig not implemented")
}
func (UnimplementedRoomCommandServiceServer) AdminCreateRoomPin(context.Context, *AdminCreateRoomPinRequest) (*AdminCreateRoomPinResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminCreateRoomPin not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminCreateRoomPin not implemented")
}
func (UnimplementedRoomCommandServiceServer) AdminCancelRoomPin(context.Context, *AdminCancelRoomPinRequest) (*AdminCancelRoomPinResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminCancelRoomPin not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminCancelRoomPin not implemented")
}
func (UnimplementedRoomCommandServiceServer) MicUp(context.Context, *MicUpRequest) (*MicUpResponse, error) {
return nil, status.Error(codes.Unimplemented, "method MicUp not implemented")
return nil, status.Errorf(codes.Unimplemented, "method MicUp not implemented")
}
func (UnimplementedRoomCommandServiceServer) MicDown(context.Context, *MicDownRequest) (*MicDownResponse, error) {
return nil, status.Error(codes.Unimplemented, "method MicDown not implemented")
return nil, status.Errorf(codes.Unimplemented, "method MicDown not implemented")
}
func (UnimplementedRoomCommandServiceServer) ChangeMicSeat(context.Context, *ChangeMicSeatRequest) (*ChangeMicSeatResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ChangeMicSeat not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ChangeMicSeat not implemented")
}
func (UnimplementedRoomCommandServiceServer) ConfirmMicPublishing(context.Context, *ConfirmMicPublishingRequest) (*ConfirmMicPublishingResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ConfirmMicPublishing not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ConfirmMicPublishing not implemented")
}
func (UnimplementedRoomCommandServiceServer) MicHeartbeat(context.Context, *MicHeartbeatRequest) (*MicHeartbeatResponse, error) {
return nil, status.Error(codes.Unimplemented, "method MicHeartbeat not implemented")
return nil, status.Errorf(codes.Unimplemented, "method MicHeartbeat not implemented")
}
func (UnimplementedRoomCommandServiceServer) SetMicMute(context.Context, *SetMicMuteRequest) (*SetMicMuteResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SetMicMute not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SetMicMute not implemented")
}
func (UnimplementedRoomCommandServiceServer) ApplyRTCEvent(context.Context, *ApplyRTCEventRequest) (*ApplyRTCEventResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ApplyRTCEvent not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ApplyRTCEvent not implemented")
}
func (UnimplementedRoomCommandServiceServer) SetMicSeatLock(context.Context, *SetMicSeatLockRequest) (*SetMicSeatLockResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SetMicSeatLock not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SetMicSeatLock not implemented")
}
func (UnimplementedRoomCommandServiceServer) SetChatEnabled(context.Context, *SetChatEnabledRequest) (*SetChatEnabledResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SetChatEnabled not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SetChatEnabled not implemented")
}
func (UnimplementedRoomCommandServiceServer) SetRoomPassword(context.Context, *SetRoomPasswordRequest) (*SetRoomPasswordResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SetRoomPassword not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SetRoomPassword not implemented")
}
func (UnimplementedRoomCommandServiceServer) SetRoomAdmin(context.Context, *SetRoomAdminRequest) (*SetRoomAdminResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SetRoomAdmin not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SetRoomAdmin not implemented")
}
func (UnimplementedRoomCommandServiceServer) MuteUser(context.Context, *MuteUserRequest) (*MuteUserResponse, error) {
return nil, status.Error(codes.Unimplemented, "method MuteUser not implemented")
return nil, status.Errorf(codes.Unimplemented, "method MuteUser not implemented")
}
func (UnimplementedRoomCommandServiceServer) KickUser(context.Context, *KickUserRequest) (*KickUserResponse, error) {
return nil, status.Error(codes.Unimplemented, "method KickUser not implemented")
return nil, status.Errorf(codes.Unimplemented, "method KickUser not implemented")
}
func (UnimplementedRoomCommandServiceServer) UnbanUser(context.Context, *UnbanUserRequest) (*UnbanUserResponse, error) {
return nil, status.Error(codes.Unimplemented, "method UnbanUser not implemented")
return nil, status.Errorf(codes.Unimplemented, "method UnbanUser not implemented")
}
func (UnimplementedRoomCommandServiceServer) SystemEvictUser(context.Context, *SystemEvictUserRequest) (*SystemEvictUserResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SystemEvictUser not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SystemEvictUser not implemented")
}
func (UnimplementedRoomCommandServiceServer) SendGift(context.Context, *SendGiftRequest) (*SendGiftResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SendGift not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SendGift not implemented")
}
func (UnimplementedRoomCommandServiceServer) FollowRoom(context.Context, *FollowRoomRequest) (*FollowRoomResponse, error) {
return nil, status.Error(codes.Unimplemented, "method FollowRoom not implemented")
return nil, status.Errorf(codes.Unimplemented, "method FollowRoom not implemented")
}
func (UnimplementedRoomCommandServiceServer) UnfollowRoom(context.Context, *UnfollowRoomRequest) (*UnfollowRoomResponse, error) {
return nil, status.Error(codes.Unimplemented, "method UnfollowRoom not implemented")
return nil, status.Errorf(codes.Unimplemented, "method UnfollowRoom not implemented")
}
func (UnimplementedRoomCommandServiceServer) mustEmbedUnimplementedRoomCommandServiceServer() {}
func (UnimplementedRoomCommandServiceServer) testEmbeddedByValue() {}
@ -576,7 +576,7 @@ type UnsafeRoomCommandServiceServer interface {
}
func RegisterRoomCommandServiceServer(s grpc.ServiceRegistrar, srv RoomCommandServiceServer) {
// If the following call panics, it indicates UnimplementedRoomCommandServiceServer was
// If the following call pancis, it indicates UnimplementedRoomCommandServiceServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
@ -1364,10 +1364,10 @@ type RoomGuardServiceServer interface {
type UnimplementedRoomGuardServiceServer struct{}
func (UnimplementedRoomGuardServiceServer) CheckSpeakPermission(context.Context, *CheckSpeakPermissionRequest) (*CheckSpeakPermissionResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CheckSpeakPermission not implemented")
return nil, status.Errorf(codes.Unimplemented, "method CheckSpeakPermission not implemented")
}
func (UnimplementedRoomGuardServiceServer) VerifyRoomPresence(context.Context, *VerifyRoomPresenceRequest) (*VerifyRoomPresenceResponse, error) {
return nil, status.Error(codes.Unimplemented, "method VerifyRoomPresence not implemented")
return nil, status.Errorf(codes.Unimplemented, "method VerifyRoomPresence not implemented")
}
func (UnimplementedRoomGuardServiceServer) mustEmbedUnimplementedRoomGuardServiceServer() {}
func (UnimplementedRoomGuardServiceServer) testEmbeddedByValue() {}
@ -1380,7 +1380,7 @@ type UnsafeRoomGuardServiceServer interface {
}
func RegisterRoomGuardServiceServer(s grpc.ServiceRegistrar, srv RoomGuardServiceServer) {
// If the following call panics, it indicates UnimplementedRoomGuardServiceServer was
// If the following call pancis, it indicates UnimplementedRoomGuardServiceServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
@ -1677,49 +1677,49 @@ type RoomQueryServiceServer interface {
type UnimplementedRoomQueryServiceServer struct{}
func (UnimplementedRoomQueryServiceServer) AdminListRooms(context.Context, *AdminListRoomsRequest) (*AdminListRoomsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminListRooms not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminListRooms not implemented")
}
func (UnimplementedRoomQueryServiceServer) AdminGetRoom(context.Context, *AdminGetRoomRequest) (*AdminGetRoomResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminGetRoom not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminGetRoom not implemented")
}
func (UnimplementedRoomQueryServiceServer) AdminGetRoomRocketConfig(context.Context, *AdminGetRoomRocketConfigRequest) (*AdminGetRoomRocketConfigResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminGetRoomRocketConfig not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminGetRoomRocketConfig not implemented")
}
func (UnimplementedRoomQueryServiceServer) AdminGetRoomSeatConfig(context.Context, *AdminGetRoomSeatConfigRequest) (*AdminGetRoomSeatConfigResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminGetRoomSeatConfig not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminGetRoomSeatConfig not implemented")
}
func (UnimplementedRoomQueryServiceServer) AdminListRoomPins(context.Context, *AdminListRoomPinsRequest) (*AdminListRoomPinsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminListRoomPins not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminListRoomPins not implemented")
}
func (UnimplementedRoomQueryServiceServer) ListRooms(context.Context, *ListRoomsRequest) (*ListRoomsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListRooms not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListRooms not implemented")
}
func (UnimplementedRoomQueryServiceServer) ListRoomFeeds(context.Context, *ListRoomFeedsRequest) (*ListRoomsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListRoomFeeds not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListRoomFeeds not implemented")
}
func (UnimplementedRoomQueryServiceServer) ListRoomGiftLeaderboard(context.Context, *ListRoomGiftLeaderboardRequest) (*ListRoomGiftLeaderboardResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListRoomGiftLeaderboard not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListRoomGiftLeaderboard not implemented")
}
func (UnimplementedRoomQueryServiceServer) GetMyRoom(context.Context, *GetMyRoomRequest) (*GetMyRoomResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetMyRoom not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetMyRoom not implemented")
}
func (UnimplementedRoomQueryServiceServer) GetCurrentRoom(context.Context, *GetCurrentRoomRequest) (*GetCurrentRoomResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetCurrentRoom not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetCurrentRoom not implemented")
}
func (UnimplementedRoomQueryServiceServer) GetRoomSnapshot(context.Context, *GetRoomSnapshotRequest) (*GetRoomSnapshotResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetRoomSnapshot not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetRoomSnapshot not implemented")
}
func (UnimplementedRoomQueryServiceServer) ListRoomBackgrounds(context.Context, *ListRoomBackgroundsRequest) (*ListRoomBackgroundsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListRoomBackgrounds not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListRoomBackgrounds not implemented")
}
func (UnimplementedRoomQueryServiceServer) GetRoomRocket(context.Context, *GetRoomRocketRequest) (*GetRoomRocketResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetRoomRocket not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetRoomRocket not implemented")
}
func (UnimplementedRoomQueryServiceServer) ListRoomOnlineUsers(context.Context, *ListRoomOnlineUsersRequest) (*ListRoomOnlineUsersResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListRoomOnlineUsers not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListRoomOnlineUsers not implemented")
}
func (UnimplementedRoomQueryServiceServer) ListRoomBannedUsers(context.Context, *ListRoomBannedUsersRequest) (*ListRoomBannedUsersResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListRoomBannedUsers not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListRoomBannedUsers not implemented")
}
func (UnimplementedRoomQueryServiceServer) mustEmbedUnimplementedRoomQueryServiceServer() {}
func (UnimplementedRoomQueryServiceServer) testEmbeddedByValue() {}
@ -1732,7 +1732,7 @@ type UnsafeRoomQueryServiceServer interface {
}
func RegisterRoomQueryServiceServer(s grpc.ServiceRegistrar, srv RoomQueryServiceServer) {
// If the following call panics, it indicates UnimplementedRoomQueryServiceServer was
// If the following call pancis, it indicates UnimplementedRoomQueryServiceServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.

7487
api/proto/wallet/v1/wallet.pb.go
View File

File diff suppressed because it is too large Load Diff

5
api/proto/wallet/v1/wallet.proto
View File

@ -196,6 +196,8 @@ message AdminCreditCoinSellerStockRequest {
int64 operator_user_id = 9;
string reason = 10;
string app_code = 11;
int64 seller_country_id = 12;
int64 seller_region_id = 13;
}
message AdminCreditCoinSellerStockResponse {
@ -218,9 +220,10 @@ message TransferCoinFromSellerRequest {
int64 amount = 4;
string reason = 5;
string app_code = 6;
// seller_region_id 和 target_region_id 由 gateway 从 user-service 查询，不接受客户端提交。
// seller_region_id、target_region_id 和 target_country_id 由 gateway 从 user-service 查询，不接受客户端提交。
int64 seller_region_id = 7;
int64 target_region_id = 8;
int64 target_country_id = 9;
}
// TransferCoinFromSellerResponse 同时返回金币到账结果和本次转账对应的充值金额口径。

148
api/proto/wallet/v1/wallet_grpc.pb.go
View File

@ -1,6 +1,6 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.6.2
// - protoc-gen-go-grpc v1.5.1
// - protoc v7.35.0
// source: proto/wallet/v1/wallet.proto
@ -93,13 +93,13 @@ type WalletCronServiceServer interface {
type UnimplementedWalletCronServiceServer struct{}
func (UnimplementedWalletCronServiceServer) ProcessHostSalaryDailySettlementBatch(context.Context, *CronBatchRequest) (*CronBatchResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ProcessHostSalaryDailySettlementBatch not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ProcessHostSalaryDailySettlementBatch not implemented")
}
func (UnimplementedWalletCronServiceServer) ProcessHostSalaryHalfMonthSettlementBatch(context.Context, *CronBatchRequest) (*CronBatchResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ProcessHostSalaryHalfMonthSettlementBatch not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ProcessHostSalaryHalfMonthSettlementBatch not implemented")
}
func (UnimplementedWalletCronServiceServer) ProcessHostSalaryMonthEndBatch(context.Context, *CronBatchRequest) (*CronBatchResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ProcessHostSalaryMonthEndBatch not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ProcessHostSalaryMonthEndBatch not implemented")
}
func (UnimplementedWalletCronServiceServer) mustEmbedUnimplementedWalletCronServiceServer() {}
func (UnimplementedWalletCronServiceServer) testEmbeddedByValue() {}
@ -112,7 +112,7 @@ type UnsafeWalletCronServiceServer interface {
}
func RegisterWalletCronServiceServer(s grpc.ServiceRegistrar, srv WalletCronServiceServer) {
// If the following call panics, it indicates UnimplementedWalletCronServiceServer was
// If the following call pancis, it indicates UnimplementedWalletCronServiceServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
@ -1120,208 +1120,208 @@ type WalletServiceServer interface {
type UnimplementedWalletServiceServer struct{}
func (UnimplementedWalletServiceServer) DebitGift(context.Context, *DebitGiftRequest) (*DebitGiftResponse, error) {
return nil, status.Error(codes.Unimplemented, "method DebitGift not implemented")
return nil, status.Errorf(codes.Unimplemented, "method DebitGift not implemented")
}
func (UnimplementedWalletServiceServer) BatchDebitGift(context.Context, *BatchDebitGiftRequest) (*BatchDebitGiftResponse, error) {
return nil, status.Error(codes.Unimplemented, "method BatchDebitGift not implemented")
return nil, status.Errorf(codes.Unimplemented, "method BatchDebitGift not implemented")
}
func (UnimplementedWalletServiceServer) GetBalances(context.Context, *GetBalancesRequest) (*GetBalancesResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetBalances not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetBalances not implemented")
}
func (UnimplementedWalletServiceServer) GetActiveHostSalaryPolicy(context.Context, *GetActiveHostSalaryPolicyRequest) (*GetActiveHostSalaryPolicyResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetActiveHostSalaryPolicy not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetActiveHostSalaryPolicy not implemented")
}
func (UnimplementedWalletServiceServer) GetHostSalaryProgress(context.Context, *GetHostSalaryProgressRequest) (*GetHostSalaryProgressResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetHostSalaryProgress not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetHostSalaryProgress not implemented")
}
func (UnimplementedWalletServiceServer) AdminCreditAsset(context.Context, *AdminCreditAssetRequest) (*AdminCreditAssetResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminCreditAsset not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminCreditAsset not implemented")
}
func (UnimplementedWalletServiceServer) AdminCreditCoinSellerStock(context.Context, *AdminCreditCoinSellerStockRequest) (*AdminCreditCoinSellerStockResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AdminCreditCoinSellerStock not implemented")
return nil, status.Errorf(codes.Unimplemented, "method AdminCreditCoinSellerStock not implemented")
}
func (UnimplementedWalletServiceServer) TransferCoinFromSeller(context.Context, *TransferCoinFromSellerRequest) (*TransferCoinFromSellerResponse, error) {
return nil, status.Error(codes.Unimplemented, "method TransferCoinFromSeller not implemented")
return nil, status.Errorf(codes.Unimplemented, "method TransferCoinFromSeller not implemented")
}
func (UnimplementedWalletServiceServer) ListCoinSellerSalaryExchangeRateTiers(context.Context, *ListCoinSellerSalaryExchangeRateTiersRequest) (*ListCoinSellerSalaryExchangeRateTiersResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListCoinSellerSalaryExchangeRateTiers not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListCoinSellerSalaryExchangeRateTiers not implemented")
}
func (UnimplementedWalletServiceServer) ExchangeSalaryToCoin(context.Context, *ExchangeSalaryToCoinRequest) (*ExchangeSalaryToCoinResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ExchangeSalaryToCoin not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ExchangeSalaryToCoin not implemented")
}
func (UnimplementedWalletServiceServer) TransferSalaryToCoinSeller(context.Context, *TransferSalaryToCoinSellerRequest) (*TransferSalaryToCoinSellerResponse, error) {
return nil, status.Error(codes.Unimplemented, "method TransferSalaryToCoinSeller not implemented")
return nil, status.Errorf(codes.Unimplemented, "method TransferSalaryToCoinSeller not implemented")
}
func (UnimplementedWalletServiceServer) ListResources(context.Context, *ListResourcesRequest) (*ListResourcesResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListResources not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListResources not implemented")
}
func (UnimplementedWalletServiceServer) GetResource(context.Context, *GetResourceRequest) (*GetResourceResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetResource not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetResource not implemented")
}
func (UnimplementedWalletServiceServer) CreateResource(context.Context, *CreateResourceRequest) (*ResourceResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CreateResource not implemented")
return nil, status.Errorf(codes.Unimplemented, "method CreateResource not implemented")
}
func (UnimplementedWalletServiceServer) UpdateResource(context.Context, *UpdateResourceRequest) (*ResourceResponse, error) {
return nil, status.Error(codes.Unimplemented, "method UpdateResource not implemented")
return nil, status.Errorf(codes.Unimplemented, "method UpdateResource not implemented")
}
func (UnimplementedWalletServiceServer) SetResourceStatus(context.Context, *SetResourceStatusRequest) (*ResourceResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SetResourceStatus not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SetResourceStatus not implemented")
}
func (UnimplementedWalletServiceServer) ListResourceGroups(context.Context, *ListResourceGroupsRequest) (*ListResourceGroupsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListResourceGroups not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListResourceGroups not implemented")
}
func (UnimplementedWalletServiceServer) GetResourceGroup(context.Context, *GetResourceGroupRequest) (*GetResourceGroupResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetResourceGroup not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetResourceGroup not implemented")
}
func (UnimplementedWalletServiceServer) CreateResourceGroup(context.Context, *CreateResourceGroupRequest) (*ResourceGroupResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CreateResourceGroup not implemented")
return nil, status.Errorf(codes.Unimplemented, "method CreateResourceGroup not implemented")
}
func (UnimplementedWalletServiceServer) UpdateResourceGroup(context.Context, *UpdateResourceGroupRequest) (*ResourceGroupResponse, error) {
return nil, status.Error(codes.Unimplemented, "method UpdateResourceGroup not implemented")
return nil, status.Errorf(codes.Unimplemented, "method UpdateResourceGroup not implemented")
}
func (UnimplementedWalletServiceServer) SetResourceGroupStatus(context.Context, *SetResourceGroupStatusRequest) (*ResourceGroupResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SetResourceGroupStatus not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SetResourceGroupStatus not implemented")
}
func (UnimplementedWalletServiceServer) ListGiftConfigs(context.Context, *ListGiftConfigsRequest) (*ListGiftConfigsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListGiftConfigs not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListGiftConfigs not implemented")
}
func (UnimplementedWalletServiceServer) ListGiftTypeConfigs(context.Context, *ListGiftTypeConfigsRequest) (*ListGiftTypeConfigsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListGiftTypeConfigs not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListGiftTypeConfigs not implemented")
}
func (UnimplementedWalletServiceServer) CreateGiftConfig(context.Context, *CreateGiftConfigRequest) (*GiftConfigResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CreateGiftConfig not implemented")
return nil, status.Errorf(codes.Unimplemented, "method CreateGiftConfig not implemented")
}
func (UnimplementedWalletServiceServer) UpdateGiftConfig(context.Context, *UpdateGiftConfigRequest) (*GiftConfigResponse, error) {
return nil, status.Error(codes.Unimplemented, "method UpdateGiftConfig not implemented")
return nil, status.Errorf(codes.Unimplemented, "method UpdateGiftConfig not implemented")
}
func (UnimplementedWalletServiceServer) SetGiftConfigStatus(context.Context, *SetGiftConfigStatusRequest) (*GiftConfigResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SetGiftConfigStatus not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SetGiftConfigStatus not implemented")
}
func (UnimplementedWalletServiceServer) UpsertGiftTypeConfig(context.Context, *UpsertGiftTypeConfigRequest) (*GiftTypeConfigResponse, error) {
return nil, status.Error(codes.Unimplemented, "method UpsertGiftTypeConfig not implemented")
return nil, status.Errorf(codes.Unimplemented, "method UpsertGiftTypeConfig not implemented")
}
func (UnimplementedWalletServiceServer) GrantResource(context.Context, *GrantResourceRequest) (*ResourceGrantResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GrantResource not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GrantResource not implemented")
}
func (UnimplementedWalletServiceServer) GrantResourceGroup(context.Context, *GrantResourceGroupRequest) (*ResourceGrantResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GrantResourceGroup not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GrantResourceGroup not implemented")
}
func (UnimplementedWalletServiceServer) ListUserResources(context.Context, *ListUserResourcesRequest) (*ListUserResourcesResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListUserResources not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListUserResources not implemented")
}
func (UnimplementedWalletServiceServer) EquipUserResource(context.Context, *EquipUserResourceRequest) (*EquipUserResourceResponse, error) {
return nil, status.Error(codes.Unimplemented, "method EquipUserResource not implemented")
return nil, status.Errorf(codes.Unimplemented, "method EquipUserResource not implemented")
}
func (UnimplementedWalletServiceServer) UnequipUserResource(context.Context, *UnequipUserResourceRequest) (*UnequipUserResourceResponse, error) {
return nil, status.Error(codes.Unimplemented, "method UnequipUserResource not implemented")
return nil, status.Errorf(codes.Unimplemented, "method UnequipUserResource not implemented")
}
func (UnimplementedWalletServiceServer) BatchGetUserEquippedResources(context.Context, *BatchGetUserEquippedResourcesRequest) (*BatchGetUserEquippedResourcesResponse, error) {
return nil, status.Error(codes.Unimplemented, "method BatchGetUserEquippedResources not implemented")
return nil, status.Errorf(codes.Unimplemented, "method BatchGetUserEquippedResources not implemented")
}
func (UnimplementedWalletServiceServer) ListResourceGrants(context.Context, *ListResourceGrantsRequest) (*ListResourceGrantsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListResourceGrants not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListResourceGrants not implemented")
}
func (UnimplementedWalletServiceServer) ListResourceShopItems(context.Context, *ListResourceShopItemsRequest) (*ListResourceShopItemsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListResourceShopItems not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListResourceShopItems not implemented")
}
func (UnimplementedWalletServiceServer) UpsertResourceShopItems(context.Context, *UpsertResourceShopItemsRequest) (*UpsertResourceShopItemsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method UpsertResourceShopItems not implemented")
return nil, status.Errorf(codes.Unimplemented, "method UpsertResourceShopItems not implemented")
}
func (UnimplementedWalletServiceServer) SetResourceShopItemStatus(context.Context, *SetResourceShopItemStatusRequest) (*ResourceShopItemResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SetResourceShopItemStatus not implemented")
return nil, status.Errorf(codes.Unimplemented, "method SetResourceShopItemStatus not implemented")
}
func (UnimplementedWalletServiceServer) PurchaseResourceShopItem(context.Context, *PurchaseResourceShopItemRequest) (*PurchaseResourceShopItemResponse, error) {
return nil, status.Error(codes.Unimplemented, "method PurchaseResourceShopItem not implemented")
return nil, status.Errorf(codes.Unimplemented, "method PurchaseResourceShopItem not implemented")
}
func (UnimplementedWalletServiceServer) ListRechargeBills(context.Context, *ListRechargeBillsRequest) (*ListRechargeBillsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListRechargeBills not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListRechargeBills not implemented")
}
func (UnimplementedWalletServiceServer) GetWalletOverview(context.Context, *GetWalletOverviewRequest) (*GetWalletOverviewResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetWalletOverview not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetWalletOverview not implemented")
}
func (UnimplementedWalletServiceServer) GetWalletValueSummary(context.Context, *GetWalletValueSummaryRequest) (*GetWalletValueSummaryResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetWalletValueSummary not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetWalletValueSummary not implemented")
}
func (UnimplementedWalletServiceServer) GetUserGiftWall(context.Context, *GetUserGiftWallRequest) (*GetUserGiftWallResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetUserGiftWall not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetUserGiftWall not implemented")
}
func (UnimplementedWalletServiceServer) ListRechargeProducts(context.Context, *ListRechargeProductsRequest) (*ListRechargeProductsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListRechargeProducts not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListRechargeProducts not implemented")
}
func (UnimplementedWalletServiceServer) ConfirmGooglePayment(context.Context, *ConfirmGooglePaymentRequest) (*ConfirmGooglePaymentResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ConfirmGooglePayment not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ConfirmGooglePayment not implemented")
}
func (UnimplementedWalletServiceServer) ListAdminRechargeProducts(context.Context, *ListAdminRechargeProductsRequest) (*ListAdminRechargeProductsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListAdminRechargeProducts not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListAdminRechargeProducts not implemented")
}
func (UnimplementedWalletServiceServer) CreateRechargeProduct(context.Context, *CreateRechargeProductRequest) (*RechargeProductResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CreateRechargeProduct not implemented")
return nil, status.Errorf(codes.Unimplemented, "method CreateRechargeProduct not implemented")
}
func (UnimplementedWalletServiceServer) UpdateRechargeProduct(context.Context, *UpdateRechargeProductRequest) (*RechargeProductResponse, error) {
return nil, status.Error(codes.Unimplemented, "method UpdateRechargeProduct not implemented")
return nil, status.Errorf(codes.Unimplemented, "method UpdateRechargeProduct not implemented")
}
func (UnimplementedWalletServiceServer) DeleteRechargeProduct(context.Context, *DeleteRechargeProductRequest) (*DeleteRechargeProductResponse, error) {
return nil, status.Error(codes.Unimplemented, "method DeleteRechargeProduct not implemented")
return nil, status.Errorf(codes.Unimplemented, "method DeleteRechargeProduct not implemented")
}
func (UnimplementedWalletServiceServer) GetDiamondExchangeConfig(context.Context, *GetDiamondExchangeConfigRequest) (*GetDiamondExchangeConfigResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetDiamondExchangeConfig not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetDiamondExchangeConfig not implemented")
}
func (UnimplementedWalletServiceServer) ListWalletTransactions(context.Context, *ListWalletTransactionsRequest) (*ListWalletTransactionsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListWalletTransactions not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListWalletTransactions not implemented")
}
func (UnimplementedWalletServiceServer) ListVipPackages(context.Context, *ListVipPackagesRequest) (*ListVipPackagesResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListVipPackages not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListVipPackages not implemented")
}
func (UnimplementedWalletServiceServer) GetMyVip(context.Context, *GetMyVipRequest) (*GetMyVipResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetMyVip not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetMyVip not implemented")
}
func (UnimplementedWalletServiceServer) PurchaseVip(context.Context, *PurchaseVipRequest) (*PurchaseVipResponse, error) {
return nil, status.Error(codes.Unimplemented, "method PurchaseVip not implemented")
return nil, status.Errorf(codes.Unimplemented, "method PurchaseVip not implemented")
}
func (UnimplementedWalletServiceServer) GrantVip(context.Context, *GrantVipRequest) (*GrantVipResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GrantVip not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GrantVip not implemented")
}
func (UnimplementedWalletServiceServer) ListAdminVipLevels(context.Context, *ListAdminVipLevelsRequest) (*ListAdminVipLevelsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListAdminVipLevels not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListAdminVipLevels not implemented")
}
func (UnimplementedWalletServiceServer) UpdateAdminVipLevels(context.Context, *UpdateAdminVipLevelsRequest) (*UpdateAdminVipLevelsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method UpdateAdminVipLevels not implemented")
return nil, status.Errorf(codes.Unimplemented, "method UpdateAdminVipLevels not implemented")
}
func (UnimplementedWalletServiceServer) CreditTaskReward(context.Context, *CreditTaskRewardRequest) (*CreditTaskRewardResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CreditTaskReward not implemented")
return nil, status.Errorf(codes.Unimplemented, "method CreditTaskReward not implemented")
}
func (UnimplementedWalletServiceServer) CreditLuckyGiftReward(context.Context, *CreditLuckyGiftRewardRequest) (*CreditLuckyGiftRewardResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CreditLuckyGiftReward not implemented")
return nil, status.Errorf(codes.Unimplemented, "method CreditLuckyGiftReward not implemented")
}
func (UnimplementedWalletServiceServer) CreditRoomTurnoverReward(context.Context, *CreditRoomTurnoverRewardRequest) (*CreditRoomTurnoverRewardResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CreditRoomTurnoverReward not implemented")
return nil, status.Errorf(codes.Unimplemented, "method CreditRoomTurnoverReward not implemented")
}
func (UnimplementedWalletServiceServer) ApplyGameCoinChange(context.Context, *ApplyGameCoinChangeRequest) (*ApplyGameCoinChangeResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ApplyGameCoinChange not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ApplyGameCoinChange not implemented")
}
func (UnimplementedWalletServiceServer) GetRedPacketConfig(context.Context, *GetRedPacketConfigRequest) (*GetRedPacketConfigResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetRedPacketConfig not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetRedPacketConfig not implemented")
}
func (UnimplementedWalletServiceServer) UpdateRedPacketConfig(context.Context, *UpdateRedPacketConfigRequest) (*UpdateRedPacketConfigResponse, error) {
return nil, status.Error(codes.Unimplemented, "method UpdateRedPacketConfig not implemented")
return nil, status.Errorf(codes.Unimplemented, "method UpdateRedPacketConfig not implemented")
}
func (UnimplementedWalletServiceServer) CreateRedPacket(context.Context, *CreateRedPacketRequest) (*CreateRedPacketResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CreateRedPacket not implemented")
return nil, status.Errorf(codes.Unimplemented, "method CreateRedPacket not implemented")
}
func (UnimplementedWalletServiceServer) ClaimRedPacket(context.Context, *ClaimRedPacketRequest) (*ClaimRedPacketResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ClaimRedPacket not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ClaimRedPacket not implemented")
}
func (UnimplementedWalletServiceServer) ListRedPackets(context.Context, *ListRedPacketsRequest) (*ListRedPacketsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListRedPackets not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ListRedPackets not implemented")
}
func (UnimplementedWalletServiceServer) GetRedPacket(context.Context, *GetRedPacketRequest) (*GetRedPacketResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetRedPacket not implemented")
return nil, status.Errorf(codes.Unimplemented, "method GetRedPacket not implemented")
}
func (UnimplementedWalletServiceServer) ExpireRedPackets(context.Context, *ExpireRedPacketsRequest) (*ExpireRedPacketsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ExpireRedPackets not implemented")
return nil, status.Errorf(codes.Unimplemented, "method ExpireRedPackets not implemented")
}
func (UnimplementedWalletServiceServer) RetryRedPacketRefund(context.Context, *RetryRedPacketRefundRequest) (*RetryRedPacketRefundResponse, error) {
return nil, status.Error(codes.Unimplemented, "method RetryRedPacketRefund not implemented")
return nil, status.Errorf(codes.Unimplemented, "method RetryRedPacketRefund not implemented")
}
func (UnimplementedWalletServiceServer) mustEmbedUnimplementedWalletServiceServer() {}
func (UnimplementedWalletServiceServer) testEmbeddedByValue() {}
@ -1334,7 +1334,7 @@ type UnsafeWalletServiceServer interface {
}
func RegisterWalletServiceServer(s grpc.ServiceRegistrar, srv WalletServiceServer) {
// If the following call panics, it indicates UnimplementedWalletServiceServer was
// If the following call pancis, it indicates UnimplementedWalletServiceServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.

13
server/admin/internal/modules/hostorg/service.go
View File

@ -303,10 +303,23 @@ func (s *Service) CreditCoinSellerStock(ctx context.Context, actorID int64, sell
if profile == nil || profile.Status != coinSellerStatusActive || profile.MerchantAssetType != coinSellerMerchantAssetType {
return nil, fmt.Errorf("target user is not an active coin seller")
}
seller, err := s.userClient.GetUser(ctx, userclient.GetUserRequest{
RequestID: requestID,
Caller: "hyapp-admin-server",
UserID: sellerUserID,
})
if err != nil {
return nil, err
}
if seller == nil || seller.CountryID <= 0 || seller.RegionID <= 0 {
return nil, fmt.Errorf("coin seller country and region are required")
}
return s.walletClient.AdminCreditCoinSellerStock(ctx, &walletv1.AdminCreditCoinSellerStockRequest{
CommandId: commandID,
SellerUserId: sellerUserID,
SellerCountryId: seller.CountryID,
SellerRegionId: seller.RegionID,
StockType: stockType,
CoinAmount: req.CoinAmount,
PaidCurrencyCode: paidCurrencyCode,

30
services/gateway-service/internal/transport/http/roomapi/room_handler.go
View File

@ -836,11 +836,18 @@ func (h *Handler) joinRoom(writer http.ResponseWriter, request *http.Request) {
if !httpkit.Decode(writer, request, &body) {
return
}
actorCountryID, actorRegionID, err := h.resolveRoomActorCountryRegion(request)
if err != nil {
httpkit.WriteRPCError(writer, request, err)
return
}
resp, err := h.roomClient.JoinRoom(request.Context(), &roomv1.JoinRoomRequest{
Meta: httpkit.RoomMeta(request, body.RoomID, body.CommandID),
Role: body.Role,
Password: body.Password,
Meta: httpkit.RoomMeta(request, body.RoomID, body.CommandID),
Role: body.Role,
Password: body.Password,
ActorCountryId: actorCountryID,
ActorRegionId: actorRegionID,
// 座驾快照在进房命令提交前生成；room-service 只保存快照，不回查 wallet。
EntryVehicle: h.joinRoomEntryVehicleSnapshot(request),
})
@ -1307,7 +1314,7 @@ func (h *Handler) sendGift(writer http.ResponseWriter, request *http.Request) {
httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidArgument, "invalid argument")
return
}
senderRegionID, err := h.resolveGiftSenderRegionID(request)
senderCountryID, senderRegionID, err := h.resolveRoomActorCountryRegion(request)
if err != nil {
httpkit.WriteRPCError(writer, request, err)
return
@ -1326,6 +1333,7 @@ func (h *Handler) sendGift(writer http.ResponseWriter, request *http.Request) {
GiftId: body.GiftID,
GiftCount: body.GiftCount,
PoolId: body.PoolID,
SenderCountryId: senderCountryID,
SenderRegionId: senderRegionID,
TargetIsHost: targetIsHost,
TargetHostRegionId: targetHostRegionID,
@ -1334,21 +1342,25 @@ func (h *Handler) sendGift(writer http.ResponseWriter, request *http.Request) {
httpkit.Write(writer, request, resp, err)
}
func (h *Handler) resolveGiftSenderRegionID(request *http.Request) (int64, error) {
func (h *Handler) resolveRoomActorCountryRegion(request *http.Request) (int64, int64, error) {
if h.userProfileClient == nil {
return 0, xerr.New(xerr.Unavailable, "user service is not configured")
return 0, 0, xerr.New(xerr.Unavailable, "user service is not configured")
}
resp, err := h.userProfileClient.GetUser(request.Context(), &userv1.GetUserRequest{
Meta: httpkit.UserMeta(request, ""),
UserId: auth.UserIDFromContext(request.Context()),
})
if err != nil {
return 0, err
return 0, 0, err
}
if resp.GetUser() == nil {
return 0, xerr.New(xerr.NotFound, "sender not found")
return 0, 0, xerr.New(xerr.NotFound, "user not found")
}
return resp.GetUser().GetRegionId(), nil
if resp.GetUser().GetCountryId() <= 0 || resp.GetUser().GetRegionId() <= 0 {
return 0, 0, xerr.New(xerr.InvalidArgument, "user country and region are required")
}
// room-service 事件必须携带用户真实国家和区域；房间 visible_region_id 只表达房间列表归属。
return resp.GetUser().GetCountryId(), resp.GetUser().GetRegionId(), nil
}
func (h *Handler) resolveGiftTargetHostScope(request *http.Request, targetUserID int64) (bool, int64, int64, error) {

47
services/gateway-service/internal/transport/http/walletapi/wallet_handler.go
View File

@ -133,20 +133,21 @@ func (h *Handler) transferCoinFromSeller(writer http.ResponseWriter, request *ht
if !ok {
return
}
sellerRegionID, targetRegionID, ok := h.coinSellerTransferRegions(writer, request, sellerUserID, targetUserID)
sellerRegionID, targetRegionID, targetCountryID, ok := h.coinSellerTransferRegions(writer, request, sellerUserID, targetUserID)
if !ok {
return
}
resp, err := h.walletClient.TransferCoinFromSeller(request.Context(), &walletv1.TransferCoinFromSellerRequest{
CommandId: commandID,
SellerUserId: sellerUserID,
TargetUserId: targetUserID,
SellerRegionId: sellerRegionID,
TargetRegionId: targetRegionID,
Amount: body.Amount,
Reason: reason,
AppCode: appcode.FromContext(request.Context()),
CommandId: commandID,
SellerUserId: sellerUserID,
TargetUserId: targetUserID,
SellerRegionId: sellerRegionID,
TargetRegionId: targetRegionID,
TargetCountryId: targetCountryID,
Amount: body.Amount,
Reason: reason,
AppCode: appcode.FromContext(request.Context()),
})
if err != nil {
httpkit.WriteRPCError(writer, request, err)
@ -184,22 +185,24 @@ func (h *Handler) resolveCoinSellerTransferTargetUserID(writer http.ResponseWrit
return targetUserID, true
}
func (h *Handler) coinSellerTransferRegions(writer http.ResponseWriter, request *http.Request, sellerUserID int64, targetUserID int64) (int64, int64, bool) {
func (h *Handler) coinSellerTransferRegions(writer http.ResponseWriter, request *http.Request, sellerUserID int64, targetUserID int64) (int64, int64, int64, bool) {
sellerResp, err := h.userProfileClient.GetUser(request.Context(), &userv1.GetUserRequest{
Meta: httpkit.UserMeta(request, ""),
UserId: sellerUserID,
})
if err != nil {
httpkit.WriteRPCError(writer, request, err)
return 0, 0, false
return 0, 0, 0, false
}
sellerRegionID := sellerResp.GetUser().GetRegionId()
seller := sellerResp.GetUser()
sellerCountryID := seller.GetCountryId()
sellerRegionID := seller.GetRegionId()
if sellerUserID == targetUserID {
if sellerRegionID <= 0 {
if sellerCountryID <= 0 || sellerRegionID <= 0 {
httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidArgument, "invalid argument")
return 0, 0, false
return 0, 0, 0, false
}
return sellerRegionID, sellerRegionID, true
return sellerRegionID, sellerRegionID, sellerCountryID, true
}
targetResp, err := h.userProfileClient.GetUser(request.Context(), &userv1.GetUserRequest{
Meta: httpkit.UserMeta(request, ""),
@ -207,19 +210,21 @@ func (h *Handler) coinSellerTransferRegions(writer http.ResponseWriter, request
})
if err != nil {
httpkit.WriteRPCError(writer, request, err)
return 0, 0, false
return 0, 0, 0, false
}
targetRegionID := targetResp.GetUser().GetRegionId()
if sellerRegionID <= 0 || targetRegionID <= 0 {
target := targetResp.GetUser()
targetCountryID := target.GetCountryId()
targetRegionID := target.GetRegionId()
if sellerRegionID <= 0 || targetCountryID <= 0 || targetRegionID <= 0 {
httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidArgument, "invalid argument")
return 0, 0, false
return 0, 0, 0, false
}
if sellerRegionID != targetRegionID {
// 充值政策当前按单一区域生效；跨区转账必须先设计明确的结算区域，不能静默套用任意一方政策。
httpkit.WriteError(writer, request, http.StatusConflict, string(xerr.Conflict), "conflict")
return 0, 0, false
return 0, 0, 0, false
}
return sellerRegionID, targetRegionID, true
return sellerRegionID, targetRegionID, targetCountryID, true
}
func walletAssetTypes(request *http.Request) []string {

6
services/room-service/internal/room/command/command.go
View File

@ -121,6 +121,10 @@ type JoinRoom struct {
Base
// Role 是用户在房间内的轻量角色，默认 audience。
Role string `json:"role"`
// ActorCountryID 是 gateway 从 user-service 注入的真实国家，用于统计进房活跃国家维度。
ActorCountryID int64 `json:"actor_country_id,omitempty"`
// ActorRegionID 是 gateway 从 user-service 注入的真实区域；房间 visible_region_id 不能代替用户国家。
ActorRegionID int64 `json:"actor_region_id,omitempty"`
// EntryVehicle 是 gateway 在进房时解析出的当前有效佩戴座驾快照，用于进房 IM。
EntryVehicle EntryVehicleSnapshot `json:"entry_vehicle,omitempty"`
}
@ -396,6 +400,8 @@ type SendGift struct {
GiftCount int32 `json:"gift_count"`
// SenderRegionID 是送礼用户所属区域，由 gateway 从 user-service 注入。
SenderRegionID int64 `json:"sender_region_id,omitempty"`
// SenderCountryID 是送礼用户所属国家，由 gateway 从 user-service 注入，统计服务不能用房间区域反推。
SenderCountryID int64 `json:"sender_country_id,omitempty"`
// TargetIsHost 是 gateway 从 user-service active host profile 注入的工资入账开关。
TargetIsHost bool `json:"target_is_host,omitempty"`
// TargetHostRegionID 是工资政策匹配区域；room-service 不自行推导主播身份或区域。

19
services/room-service/internal/room/service/gift.go
View File

@ -26,14 +26,15 @@ func (s *Service) SendGift(ctx context.Context, req *roomv1.SendGiftRequest) (*r
ctx = contextFromMeta(ctx, req.GetMeta())
// SendGift 的账务不在 room-service 内结算，但房间表现、热度和榜单由 Room Cell 同步更新。
cmd := command.SendGift{
Base: baseFromMeta(req.GetMeta()),
TargetType: normalizeGiftTargetType(req.GetTargetType()),
TargetUserIDs: normalizeGiftTargetUserIDs(req.GetTargetUserId(), req.GetTargetUserIds()),
GiftID: req.GetGiftId(),
PoolID: strings.TrimSpace(req.GetPoolId()),
GiftCount: req.GetGiftCount(),
SenderRegionID: req.GetSenderRegionId(),
TargetIsHost: req.GetTargetIsHost(),
Base: baseFromMeta(req.GetMeta()),
TargetType: normalizeGiftTargetType(req.GetTargetType()),
TargetUserIDs: normalizeGiftTargetUserIDs(req.GetTargetUserId(), req.GetTargetUserIds()),
GiftID: req.GetGiftId(),
PoolID: strings.TrimSpace(req.GetPoolId()),
GiftCount: req.GetGiftCount(),
SenderCountryID: req.GetSenderCountryId(),
SenderRegionID: req.GetSenderRegionId(),
TargetIsHost: req.GetTargetIsHost(),
// 工资区域由 gateway 从 user-service host profile 注入；房间 visible_region_id 只用于房间/礼物可见性。
TargetHostRegionID: req.GetTargetHostRegionId(),
TargetAgencyOwnerUserID: req.GetTargetAgencyOwnerUserId(),
@ -175,6 +176,8 @@ func (s *Service) SendGift(ctx context.Context, req *roomv1.SendGiftRequest) (*r
CoinSpent: targetBilling.Billing.GetCoinSpent(),
BillingReceiptId: targetBilling.Billing.GetBillingReceiptId(),
VisibleRegionId: roomMeta.VisibleRegionID,
CountryId: cmd.SenderCountryID,
RegionId: firstNonZeroInt64(cmd.SenderRegionID, roomMeta.VisibleRegionID),
CommandId: targetBilling.CommandID,
})
if err != nil {

9
services/room-service/internal/room/service/helpers.go
View File

@ -98,6 +98,15 @@ func runtimeRoomRef(key string) loadedRoomRef {
return loadedRoomRef{AppCode: "", RoomID: key}
}
func firstNonZeroInt64(values ...int64) int64 {
for _, value := range values {
if value != 0 {
return value
}
}
return 0
}
func commandResult(applied bool, roomVersion int64, now time.Time) *roomv1.CommandResult {
// 所有命令响应统一带 applied、room_version 和服务端时间，便于客户端处理幂等结果。
return &roomv1.CommandResult{

8
services/room-service/internal/room/service/presence.go
View File

@ -25,8 +25,10 @@ func (s *Service) JoinRoom(ctx context.Context, req *roomv1.JoinRoomRequest) (*r
password := strings.TrimSpace(req.GetPassword())
// JoinRoom 只维护 room-service presence，不创建任何本地长连接订阅。
cmd := command.JoinRoom{
Base: baseFromMeta(req.GetMeta()),
Role: req.GetRole(),
Base: baseFromMeta(req.GetMeta()),
Role: req.GetRole(),
ActorCountryID: req.GetActorCountryId(),
ActorRegionID: req.GetActorRegionId(),
// entry_vehicle 来自 gateway 入房瞬间的 wallet 快照，写入 command log 后可稳定重放。
EntryVehicle: entryVehicleSnapshotFromProto(req.GetEntryVehicle()),
}
@ -78,6 +80,8 @@ func (s *Service) JoinRoom(ctx context.Context, req *roomv1.JoinRoomRequest) (*r
UserId: cmd.ActorUserID(),
Role: cmd.Role,
VisibleRegionId: current.VisibleRegionID,
CountryId: cmd.ActorCountryID,
RegionId: firstNonZeroInt64(cmd.ActorRegionID, current.VisibleRegionID),
// outbox 事件必须带座驾快照；否则异步补偿投递 IM 时会丢失入场座驾。
EntryVehicle: eventEntryVehicleFromCommand(cmd.EntryVehicle),
})

647
services/statistics-service/cmd/backfill-last7/main.go Normal file
View File

@ -0,0 +1,647 @@
package main
import (
"context"
"database/sql"
"flag"
"fmt"
"log"
"os"
"sort"
"strings"
"time"
_ "github.com/go-sql-driver/mysql"
"hyapp/pkg/appcode"
)
type countryRegion struct {
countryID int64
regionID int64
}
type appDayKey struct {
day string
countryID int64
regionID int64
}
type appDayAgg struct {
newUsers int64
activeUsers int64
paidUsers int64
rechargeUSDMinor int64
newUserRechargeUSDMinor int64
coinSellerRechargeUSDMinor int64
googleRechargeUSDMinor int64
coinSellerTransferCoin int64
}
type userRegistrationRow struct {
userID int64
registeredDay string
dim countryRegion
createdAtMS int64
}
type payerRow struct {
day string
userID int64
dim countryRegion
firstMS int64
}
type activeRow struct {
day string
userID int64
dim countryRegion
firstMS int64
}
type rebuildState struct {
appDays map[appDayKey]*appDayAgg
registrations []userRegistrationRow
registeredByUserDay map[string]struct{}
payersByKey map[string]payerRow
activeByKey map[string]activeRow
userIDs map[int64]struct{}
userDims map[int64]countryRegion
}
func main() {
ctx := context.Background()
var (
appFlag = flag.String("app", envDefault("APP_CODE", "lalu"), "app_code to backfill")
tzFlag = flag.String("timezone", envDefault("BACKFILL_TIMEZONE", "UTC"), "day boundary timezone, e.g. UTC or Asia/Shanghai")
startFlag = flag.String("start-day", os.Getenv("START_DAY"), "inclusive yyyy-mm-dd in timezone; defaults to last 7 days")
endFlag = flag.String("end-day", os.Getenv("END_DAY"), "inclusive yyyy-mm-dd in timezone; defaults to today")
statsDSNFlag = flag.String("statistics-dsn", firstNonEmpty(os.Getenv("STATISTICS_MYSQL_DSN"), os.Getenv("MYSQL_DSN")), "statistics mysql dsn")
userDSNFlag = flag.String("user-dsn", os.Getenv("USER_MYSQL_DSN"), "user-service mysql dsn")
walletDSNFlag = flag.String("wallet-dsn", os.Getenv("WALLET_MYSQL_DSN"), "wallet-service mysql dsn")
applyFlag = flag.Bool("apply", envBool("APPLY"), "apply replacement; default dry-run")
)
flag.Parse()
if strings.TrimSpace(*statsDSNFlag) == "" || strings.TrimSpace(*userDSNFlag) == "" || strings.TrimSpace(*walletDSNFlag) == "" {
log.Fatal("statistics-dsn, user-dsn and wallet-dsn are required")
}
loc, err := time.LoadLocation(strings.TrimSpace(*tzFlag))
if err != nil {
log.Fatalf("load timezone: %v", err)
}
startDay, endDay, windowStartMS, windowEndMS, err := resolveWindow(*startFlag, *endFlag, loc)
if err != nil {
log.Fatal(err)
}
app := appcode.Normalize(*appFlag)
statsDB := mustOpen(ctx, *statsDSNFlag)
defer statsDB.Close()
userDB := mustOpen(ctx, *userDSNFlag)
defer userDB.Close()
walletDB := mustOpen(ctx, *walletDSNFlag)
defer walletDB.Close()
state := &rebuildState{
appDays: map[appDayKey]*appDayAgg{},
registeredByUserDay: map[string]struct{}{},
payersByKey: map[string]payerRow{},
activeByKey: map[string]activeRow{},
userIDs: map[int64]struct{}{},
userDims: map[int64]countryRegion{},
}
// 第一阶段只收集窗口内会参与重算的 user_id，后续统一批量加载用户国家和区域，避免按充值/活跃记录逐条访问 user-service 数据库。
if err := collectRegistrationUsers(ctx, userDB, state, app, windowStartMS, windowEndMS); err != nil {
log.Fatalf("collect registrations: %v", err)
}
if err := collectActiveUserIDs(ctx, statsDB, state, app, startDay, endDay); err != nil {
log.Fatalf("collect active ids: %v", err)
}
if err := collectRechargeUserIDs(ctx, walletDB, state, app, windowStartMS, windowEndMS); err != nil {
log.Fatalf("collect recharge ids: %v", err)
}
if err := loadUserDims(ctx, userDB, state, app); err != nil {
log.Fatalf("load user dims: %v", err)
}
if err := rebuildRegistrations(ctx, userDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
log.Fatalf("rebuild registrations: %v", err)
}
if err := rebuildActive(ctx, statsDB, state, app, startDay, endDay); err != nil {
log.Fatalf("rebuild active: %v", err)
}
if err := rebuildWalletRecharge(ctx, walletDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
log.Fatalf("rebuild wallet recharge: %v", err)
}
if err := rebuildCoinSellerStock(ctx, walletDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
log.Fatalf("rebuild coin seller stock: %v", err)
}
if err := stageAndMaybeApply(ctx, statsDB, state, app, startDay, endDay, *applyFlag); err != nil {
log.Fatalf("stage/apply: %v", err)
}
mode := "dry-run"
if *applyFlag {
mode = "applied"
}
fmt.Printf("%s app=%s timezone=%s window=%s..%s app_day_rows=%d registrations=%d active=%d payers=%d\n",
mode, app, loc.String(), startDay, endDay, len(state.appDays), len(state.registrations), len(state.activeByKey), len(state.payersByKey))
}
func envDefault(key string, fallback string) string {
if value := strings.TrimSpace(os.Getenv(key)); value != "" {
return value
}
return fallback
}
func envBool(key string) bool {
switch strings.ToLower(strings.TrimSpace(os.Getenv(key))) {
case "1", "true", "yes", "y":
return true
default:
return false
}
}
func firstNonEmpty(values ...string) string {
for _, value := range values {
if strings.TrimSpace(value) != "" {
return strings.TrimSpace(value)
}
}
return ""
}
func mustOpen(ctx context.Context, dsn string) *sql.DB {
db, err := sql.Open("mysql", strings.TrimSpace(dsn))
if err != nil {
log.Fatal(err)
}
if err := db.PingContext(ctx); err != nil {
log.Fatal(err)
}
return db
}
func resolveWindow(rawStart, rawEnd string, loc *time.Location) (string, string, int64, int64, error) {
now := time.Now().In(loc)
endDayTime := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, loc)
startDayTime := endDayTime.AddDate(0, 0, -6)
var err error
if strings.TrimSpace(rawEnd) != "" {
endDayTime, err = parseDay(rawEnd, loc)
if err != nil {
return "", "", 0, 0, err
}
}
if strings.TrimSpace(rawStart) != "" {
startDayTime, err = parseDay(rawStart, loc)
if err != nil {
return "", "", 0, 0, err
}
}
if startDayTime.After(endDayTime) {
return "", "", 0, 0, fmt.Errorf("start-day must not be after end-day")
}
windowEnd := endDayTime.AddDate(0, 0, 1)
return formatDay(startDayTime), formatDay(endDayTime), startDayTime.UTC().UnixMilli(), windowEnd.UTC().UnixMilli(), nil
}
func parseDay(value string, loc *time.Location) (time.Time, error) {
parsed, err := time.ParseInLocation("2006-01-02", strings.TrimSpace(value), loc)
if err != nil {
return time.Time{}, err
}
return time.Date(parsed.Year(), parsed.Month(), parsed.Day(), 0, 0, 0, 0, loc), nil
}
func dayFromMS(ms int64, loc *time.Location) string {
return formatDay(time.UnixMilli(ms).In(loc))
}
func formatDay(value time.Time) string {
return value.Format("2006-01-02")
}
func collectRegistrationUsers(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64) error {
rows, err := db.QueryContext(ctx, `
SELECT u.user_id
FROM users u
WHERE u.app_code = ? AND u.created_at_ms >= ? AND u.created_at_ms < ?`,
app, startMS, endMS)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var userID int64
if err := rows.Scan(&userID); err != nil {
return err
}
state.userIDs[userID] = struct{}{}
}
return rows.Err()
}
func collectActiveUserIDs(ctx context.Context, db *sql.DB, state *rebuildState, app, startDay, endDay string) error {
rows, err := db.QueryContext(ctx, `
SELECT DISTINCT user_id
FROM stat_user_day_activity
WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
app, startDay, endDay)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var userID int64
if err := rows.Scan(&userID); err != nil {
return err
}
state.userIDs[userID] = struct{}{}
}
return rows.Err()
}
func collectRechargeUserIDs(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64) error {
for _, query := range []string{
`SELECT DISTINCT user_id FROM wallet_recharge_records WHERE app_code = ? AND created_at_ms >= ? AND created_at_ms < ?`,
`SELECT DISTINCT seller_user_id FROM coin_seller_stock_records WHERE app_code = ? AND counts_as_seller_recharge = 1 AND created_at_ms >= ? AND created_at_ms < ?`,
} {
rows, err := db.QueryContext(ctx, query, app, startMS, endMS)
if err != nil {
return err
}
for rows.Next() {
var userID int64
if err := rows.Scan(&userID); err != nil {
rows.Close()
return err
}
state.userIDs[userID] = struct{}{}
}
if err := rows.Close(); err != nil {
return err
}
}
return nil
}
func loadUserDims(ctx context.Context, db *sql.DB, state *rebuildState, app string) error {
userIDs := make([]int64, 0, len(state.userIDs))
for userID := range state.userIDs {
userIDs = append(userIDs, userID)
}
sort.Slice(userIDs, func(i, j int) bool { return userIDs[i] < userIDs[j] })
for start := 0; start < len(userIDs); start += 500 {
end := start + 500
if end > len(userIDs) {
end = len(userIDs)
}
placeholders := strings.TrimRight(strings.Repeat("?,", end-start), ",")
args := []any{app}
for _, userID := range userIDs[start:end] {
args = append(args, userID)
}
rows, err := db.QueryContext(ctx, `
SELECT u.user_id, COALESCE(c.country_id, 0), COALESCE(u.region_id, 0)
FROM users u
LEFT JOIN countries c ON c.app_code = u.app_code AND c.country_code = u.country
WHERE u.app_code = ? AND u.user_id IN (`+placeholders+`)`, args...)
if err != nil {
return err
}
for rows.Next() {
var userID int64
var dim countryRegion
if err := rows.Scan(&userID, &dim.countryID, &dim.regionID); err != nil {
rows.Close()
return err
}
state.userDims[userID] = dim
}
if err := rows.Close(); err != nil {
return err
}
}
return nil
}
func rebuildRegistrations(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64, loc *time.Location) error {
// 注册国家以 users.country 映射出的真实 country_id 为准；region_id 只作为独立维度写入，不能再兜底成国家。
rows, err := db.QueryContext(ctx, `
SELECT u.user_id, COALESCE(c.country_id, 0), COALESCE(u.region_id, 0), u.created_at_ms
FROM users u
LEFT JOIN countries c ON c.app_code = u.app_code AND c.country_code = u.country
WHERE u.app_code = ? AND u.created_at_ms >= ? AND u.created_at_ms < ?`,
app, startMS, endMS)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var row userRegistrationRow
if err := rows.Scan(&row.userID, &row.dim.countryID, &row.dim.regionID, &row.createdAtMS); err != nil {
return err
}
row.registeredDay = dayFromMS(row.createdAtMS, loc)
state.registrations = append(state.registrations, row)
state.registeredByUserDay[userDayKey(row.registeredDay, row.userID)] = struct{}{}
agg := state.ensureAgg(row.registeredDay, row.dim)
agg.newUsers++
}
return rows.Err()
}
func rebuildActive(ctx context.Context, db *sql.DB, state *rebuildState, app, startDay, endDay string) error {
// 活跃的原始房间事件已经被实时消费进 stat_user_day_activity；离线回填只重刷该表的国家/区域快照，避免在大窗口内扫描房间流水。
rows, err := db.QueryContext(ctx, `
SELECT stat_day, user_id, MIN(first_active_at_ms)
FROM stat_user_day_activity
WHERE app_code = ? AND stat_day BETWEEN ? AND ?
GROUP BY stat_day, user_id`,
app, startDay, endDay)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var day string
var userID, firstMS int64
if err := rows.Scan(&day, &userID, &firstMS); err != nil {
return err
}
dim := state.userDims[userID]
key := fmt.Sprintf("%s:%d", day, userID)
if _, exists := state.activeByKey[key]; exists {
continue
}
state.activeByKey[key] = activeRow{day: day, userID: userID, dim: dim, firstMS: firstMS}
state.ensureAgg(day, dim).activeUsers++
}
return rows.Err()
}
func rebuildWalletRecharge(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64, loc *time.Location) error {
// wallet_recharge_records 是到账事实表；Google 也会写入该表，payment_orders 只是 provider 审计，不能在回填里再扫一次导致双算。
// 币商向用户转账只代表金币充值用户，不产生用户 USDT 充值；Google Play 则按记录里的 usd_minor_amount 计入用户充值。
rows, err := db.QueryContext(ctx, `
SELECT rr.user_id, rr.recharge_sequence, rr.coin_amount, rr.usd_minor_amount, rr.created_at_ms, wt.biz_type
FROM wallet_recharge_records rr
JOIN wallet_transactions wt ON wt.app_code = rr.app_code AND wt.transaction_id = rr.transaction_id
WHERE rr.app_code = ? AND rr.created_at_ms >= ? AND rr.created_at_ms < ?`,
app, startMS, endMS)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var userID, sequence, coinAmount, usdMinor, createdAtMS int64
var bizType string
if err := rows.Scan(&userID, &sequence, &coinAmount, &usdMinor, &createdAtMS, &bizType); err != nil {
return err
}
day := dayFromMS(createdAtMS, loc)
dim := state.userDims[userID]
state.addPayer(day, userID, dim, createdAtMS)
agg := state.ensureAgg(day, dim)
switch strings.ToLower(strings.TrimSpace(bizType)) {
case "coin_seller_transfer":
agg.coinSellerTransferCoin += coinAmount
case "google_play_recharge":
agg.rechargeUSDMinor += usdMinor
agg.googleRechargeUSDMinor += usdMinor
default:
if usdMinor > 0 {
agg.rechargeUSDMinor += usdMinor
}
}
if sequence == 1 && usdMinor > 0 {
agg.newUserRechargeUSDMinor += usdMinor
}
}
return rows.Err()
}
func rebuildCoinSellerStock(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64, loc *time.Location) error {
// 币商进货是平台收到的 USDT 充值，按 seller 的国家/区域进入总充值，但不写入用户金币转账金额。
rows, err := db.QueryContext(ctx, `
SELECT seller_user_id, paid_amount_micro, created_at_ms
FROM coin_seller_stock_records
WHERE app_code = ? AND counts_as_seller_recharge = 1 AND paid_amount_micro > 0 AND created_at_ms >= ? AND created_at_ms < ?`,
app, startMS, endMS)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var sellerUserID, amountMicro, createdAtMS int64
if err := rows.Scan(&sellerUserID, &amountMicro, &createdAtMS); err != nil {
return err
}
usdMinor := amountMicro / 10_000
day := dayFromMS(createdAtMS, loc)
dim := state.userDims[sellerUserID]
agg := state.ensureAgg(day, dim)
agg.rechargeUSDMinor += usdMinor
agg.coinSellerRechargeUSDMinor += usdMinor
}
return rows.Err()
}
func (s *rebuildState) ensureAgg(day string, dim countryRegion) *appDayAgg {
key := appDayKey{day: day, countryID: dim.countryID, regionID: dim.regionID}
if s.appDays[key] == nil {
s.appDays[key] = &appDayAgg{}
}
return s.appDays[key]
}
func (s *rebuildState) addPayer(day string, userID int64, dim countryRegion, firstMS int64) {
key := fmt.Sprintf("%s:%d", day, userID)
if existing, exists := s.payersByKey[key]; exists && existing.firstMS <= firstMS {
return
}
if _, exists := s.payersByKey[key]; !exists {
s.ensureAgg(day, dim).paidUsers++
}
s.payersByKey[key] = payerRow{day: day, userID: userID, dim: dim, firstMS: firstMS}
}
func (s *rebuildState) userRegisteredOn(userID int64, day string) bool {
_, ok := s.registeredByUserDay[userDayKey(day, userID)]
return ok
}
func userDayKey(day string, userID int64) string {
return fmt.Sprintf("%s:%d", day, userID)
}
func stageAndMaybeApply(ctx context.Context, db *sql.DB, state *rebuildState, app, startDay, endDay string, apply bool) error {
conn, err := db.Conn(ctx)
if err != nil {
return err
}
defer conn.Close()
tx, err := conn.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
if err := createTempTables(ctx, tx); err != nil {
return err
}
if err := insertStageRows(ctx, tx, state, app); err != nil {
return err
}
if err := validateStage(ctx, tx); err != nil {
return err
}
if !apply {
return tx.Rollback()
}
nowMS := time.Now().UTC().UnixMilli()
// 本 job 只覆盖能从当前源表确定的近 7 天列；礼物、游戏、MifaPay 等不在本次源表范围内的指标保留原值，避免修国家维度时误删其他业务数据。
if _, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET new_users = 0, active_users = 0, paid_users = 0,
recharge_usd_minor = 0, new_user_recharge_usd_minor = 0,
coin_seller_recharge_usd_minor = 0,
google_recharge_usd_minor = 0, coin_seller_transfer_coin = 0,
updated_at_ms = ?
WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
nowMS, app, startDay, endDay); err != nil {
return err
}
for _, statement := range []string{
`DELETE FROM stat_user_registration WHERE app_code = ? AND registered_day BETWEEN ? AND ?`,
`DELETE FROM stat_user_day_activity WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
`DELETE FROM stat_recharge_day_payers WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
} {
if _, err := tx.ExecContext(ctx, statement, app, startDay, endDay); err != nil {
return err
}
}
if _, err := tx.ExecContext(ctx, `
INSERT INTO stat_app_day_country (
app_code, stat_day, country_id, region_id, new_users, active_users, paid_users,
recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
google_recharge_usd_minor, coin_seller_transfer_coin, updated_at_ms
)
SELECT app_code, stat_day, country_id, region_id, new_users, active_users, paid_users,
recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
google_recharge_usd_minor, coin_seller_transfer_coin, ?
FROM tmp_stat_backfill_app_day
ON DUPLICATE KEY UPDATE
new_users = VALUES(new_users), active_users = VALUES(active_users), paid_users = VALUES(paid_users),
recharge_usd_minor = VALUES(recharge_usd_minor),
new_user_recharge_usd_minor = VALUES(new_user_recharge_usd_minor),
coin_seller_recharge_usd_minor = VALUES(coin_seller_recharge_usd_minor),
google_recharge_usd_minor = VALUES(google_recharge_usd_minor),
coin_seller_transfer_coin = VALUES(coin_seller_transfer_coin),
updated_at_ms = VALUES(updated_at_ms)`, nowMS); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `
INSERT INTO stat_user_registration (app_code, user_id, registered_day, country_id, region_id, registered_at_ms)
SELECT app_code, user_id, registered_day, country_id, region_id, registered_at_ms FROM tmp_stat_backfill_registration`); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `
INSERT INTO stat_user_day_activity (app_code, stat_day, country_id, region_id, user_id, first_active_at_ms)
SELECT app_code, stat_day, country_id, region_id, user_id, first_active_at_ms FROM tmp_stat_backfill_active`); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `
INSERT INTO stat_recharge_day_payers (app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms)
SELECT app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms FROM tmp_stat_backfill_payers`); err != nil {
return err
}
return tx.Commit()
}
func createTempTables(ctx context.Context, tx *sql.Tx) error {
statements := []string{
`CREATE TEMPORARY TABLE tmp_stat_backfill_app_day (
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL, region_id BIGINT NOT NULL,
new_users BIGINT NOT NULL, active_users BIGINT NOT NULL, paid_users BIGINT NOT NULL,
recharge_usd_minor BIGINT NOT NULL, new_user_recharge_usd_minor BIGINT NOT NULL,
coin_seller_recharge_usd_minor BIGINT NOT NULL, google_recharge_usd_minor BIGINT NOT NULL,
coin_seller_transfer_coin BIGINT NOT NULL,
PRIMARY KEY(app_code, stat_day, country_id, region_id)
)`,
`CREATE TEMPORARY TABLE tmp_stat_backfill_registration (
app_code VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, registered_day DATE NOT NULL,
country_id BIGINT NOT NULL, region_id BIGINT NOT NULL, registered_at_ms BIGINT NOT NULL,
PRIMARY KEY(app_code, user_id)
)`,
`CREATE TEMPORARY TABLE tmp_stat_backfill_active (
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL,
region_id BIGINT NOT NULL, user_id BIGINT NOT NULL, first_active_at_ms BIGINT NOT NULL,
PRIMARY KEY(app_code, stat_day, user_id)
)`,
`CREATE TEMPORARY TABLE tmp_stat_backfill_payers (
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL,
region_id BIGINT NOT NULL, user_id BIGINT NOT NULL, first_paid_at_ms BIGINT NOT NULL,
PRIMARY KEY(app_code, stat_day, user_id)
)`,
}
for _, statement := range statements {
if _, err := tx.ExecContext(ctx, statement); err != nil {
return err
}
}
return nil
}
func insertStageRows(ctx context.Context, tx *sql.Tx, state *rebuildState, app string) error {
for key, agg := range state.appDays {
if _, err := tx.ExecContext(ctx, `
INSERT INTO tmp_stat_backfill_app_day (
app_code, stat_day, country_id, region_id, new_users, active_users, paid_users,
recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
google_recharge_usd_minor, coin_seller_transfer_coin
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
app, key.day, key.countryID, key.regionID, agg.newUsers, agg.activeUsers, agg.paidUsers,
agg.rechargeUSDMinor, agg.newUserRechargeUSDMinor, agg.coinSellerRechargeUSDMinor,
agg.googleRechargeUSDMinor, agg.coinSellerTransferCoin); err != nil {
return err
}
}
for _, row := range state.registrations {
if _, err := tx.ExecContext(ctx, `
INSERT INTO tmp_stat_backfill_registration (app_code, user_id, registered_day, country_id, region_id, registered_at_ms)
VALUES (?, ?, ?, ?, ?, ?)`, app, row.userID, row.registeredDay, row.dim.countryID, row.dim.regionID, row.createdAtMS); err != nil {
return err
}
}
for _, row := range state.activeByKey {
if _, err := tx.ExecContext(ctx, `
INSERT INTO tmp_stat_backfill_active (app_code, stat_day, country_id, region_id, user_id, first_active_at_ms)
VALUES (?, ?, ?, ?, ?, ?)`, app, row.day, row.dim.countryID, row.dim.regionID, row.userID, row.firstMS); err != nil {
return err
}
}
for _, row := range state.payersByKey {
if _, err := tx.ExecContext(ctx, `
INSERT INTO tmp_stat_backfill_payers (app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms)
VALUES (?, ?, ?, ?, ?, ?)`, app, row.day, row.dim.countryID, row.dim.regionID, row.userID, row.firstMS); err != nil {
return err
}
}
return nil
}
func validateStage(ctx context.Context, tx *sql.Tx) error {
var negativeCount int64
if err := tx.QueryRowContext(ctx, `
SELECT COUNT(*)
FROM tmp_stat_backfill_app_day
WHERE new_users < 0 OR active_users < 0 OR paid_users < 0 OR recharge_usd_minor < 0 OR coin_seller_transfer_coin < 0`).Scan(&negativeCount); err != nil {
return err
}
if negativeCount > 0 {
return fmt.Errorf("stage contains negative counters")
}
return nil
}

6
services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql
View File

@ -40,7 +40,7 @@ CREATE TABLE IF NOT EXISTS stat_app_day_country (
game_payout BIGINT NOT NULL DEFAULT 0 COMMENT '游戏返奖',
game_refund BIGINT NOT NULL DEFAULT 0 COMMENT '游戏退款',
updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms',
PRIMARY KEY (app_code, stat_day, country_id),
PRIMARY KEY (app_code, stat_day, country_id, region_id),
KEY idx_stat_app_day_region (app_code, stat_day, region_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='App 国家日统计聚合表';
@ -101,7 +101,7 @@ CREATE TABLE IF NOT EXISTS stat_lucky_gift_pool_day_country (
turnover_coin BIGINT NOT NULL DEFAULT 0 COMMENT '奖池幸运礼物流水金币',
payout_coin BIGINT NOT NULL DEFAULT 0 COMMENT '奖池幸运礼物返奖金币',
updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms',
PRIMARY KEY (app_code, stat_day, country_id, pool_id),
PRIMARY KEY (app_code, stat_day, country_id, region_id, pool_id),
KEY idx_stat_lucky_pool_overview (app_code, stat_day, pool_id),
KEY idx_stat_lucky_pool_region (app_code, stat_day, region_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物奖池国家日统计聚合表';
@ -118,7 +118,7 @@ CREATE TABLE IF NOT EXISTS stat_game_day_country (
refund_coin BIGINT NOT NULL DEFAULT 0,
player_count BIGINT NOT NULL DEFAULT 0,
updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id),
PRIMARY KEY (app_code, stat_day, country_id, region_id, platform_code, game_id),
KEY idx_stat_game_day_region (app_code, stat_day, region_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='游戏国家日统计聚合表';

8
services/statistics-service/internal/app/app.go
View File

@ -365,8 +365,8 @@ func roomGiftEvent(body []byte) (mysqlstorage.RoomGiftEvent, bool, error) {
AppCode: envelope.GetAppCode(),
EventID: envelope.GetEventId(),
SenderUserID: gift.GetSenderUserId(),
CountryID: 0,
VisibleRegionID: gift.GetVisibleRegionId(),
CountryID: gift.GetCountryId(),
VisibleRegionID: firstNonZeroInt64(gift.GetRegionId(), gift.GetVisibleRegionId()),
GiftValue: gift.GetGiftValue(),
CoinSpent: roomGiftCoinSpent(&gift),
PoolID: gift.GetPoolId(),
@ -402,8 +402,8 @@ func roomJoinActiveEvent(body []byte) (mysqlstorage.UserActiveEvent, bool, error
EventID: envelope.GetEventId(),
EventType: envelope.GetEventType(),
UserID: joined.GetUserId(),
CountryID: 0,
RegionID: joined.GetVisibleRegionId(),
CountryID: joined.GetCountryId(),
RegionID: firstNonZeroInt64(joined.GetRegionId(), joined.GetVisibleRegionId()),
OccurredAtMS: envelope.GetOccurredAtMs(),
}, true, nil
}

34
services/statistics-service/internal/storage/mysql/query.go
View File

@ -241,9 +241,8 @@ func (r *Repository) queryAppDayOverviewTotals(ctx context.Context, app, startDa
args = append(args, countryID)
}
if regionID > 0 {
// 旧统计行曾把 region_id 写入 country_id；兼容分支只服务历史读数，新事件写入后会走显式 region_id。
filter += " AND (region_id = ? OR (region_id = 0 AND country_id = ?))"
args = append(args, regionID, regionID)
filter += " AND region_id = ?"
args = append(args, regionID)
}
row := r.db.QueryRowContext(ctx, `
SELECT COALESCE(SUM(new_users),0), COALESCE(SUM(active_users),0), COALESCE(SUM(paid_users),0),
@ -272,8 +271,8 @@ func (r *Repository) queryNewPaidUsers(ctx context.Context, app, startDay, endDa
}
if regionID > 0 {
// p 表和 r 表都已经是统计小表，这里只按充值发生地过滤，保证新增付费用户口径和付费用户卡片一致。
filter += " AND (p.region_id = ? OR (p.region_id = 0 AND p.country_id = ?))"
args = append(args, regionID, regionID)
filter += " AND p.region_id = ?"
args = append(args, regionID)
}
row := r.db.QueryRowContext(ctx, `
SELECT COUNT(DISTINCT p.user_id)
@ -296,9 +295,8 @@ func (r *Repository) queryDailySeries(ctx context.Context, app, startDay, endDay
args = append(args, countryID)
}
if regionID > 0 {
// 和总览一致兼容历史 region_id 写到 country_id 的聚合行；新数据优先读显式 region_id。
filter += " AND (region_id = ? OR (region_id = 0 AND country_id = ?))"
args = append(args, regionID, regionID)
filter += " AND region_id = ?"
args = append(args, regionID)
}
rows, err := r.db.QueryContext(ctx, `
SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'),
@ -355,8 +353,8 @@ func (r *Repository) queryDailyNewPaidUsers(ctx context.Context, app, startDay,
args = append(args, countryID)
}
if regionID > 0 {
filter += " AND (p.region_id = ? OR (p.region_id = 0 AND p.country_id = ?))"
args = append(args, regionID, regionID)
filter += " AND p.region_id = ?"
args = append(args, regionID)
}
rows, err := r.db.QueryContext(ctx, `
SELECT DATE_FORMAT(p.stat_day, '%Y-%m-%d'), COUNT(DISTINCT p.user_id)
@ -442,8 +440,8 @@ func (r *Repository) queryRetention(ctx context.Context, app, startDay, endDay s
args = append(args, countryID)
}
if regionID > 0 {
filter += " AND (r.region_id = ? OR (r.region_id = 0 AND r.country_id = ?))"
args = append(args, regionID, regionID)
filter += " AND r.region_id = ?"
args = append(args, regionID)
}
row := r.db.QueryRowContext(ctx, `
SELECT COUNT(*),
@ -470,8 +468,8 @@ func (r *Repository) queryGameRanking(ctx context.Context, app, startDay, endDay
args = append(args, countryID)
}
if regionID > 0 {
filter += " AND (region_id = ? OR (region_id = 0 AND country_id = ?))"
args = append(args, regionID, regionID)
filter += " AND region_id = ?"
args = append(args, regionID)
}
rows, err := r.db.QueryContext(ctx, `
SELECT platform_code, game_id, COALESCE(SUM(turnover_coin),0), COALESCE(SUM(payout_coin),0),
@ -502,8 +500,8 @@ func (r *Repository) queryCountryBreakdown(ctx context.Context, app, startDay, e
args := []any{app, startDay, endDay}
filter := ""
if regionID > 0 {
filter = " AND (region_id = ? OR (region_id = 0 AND country_id = ?))"
args = append(args, regionID, regionID)
filter = " AND region_id = ?"
args = append(args, regionID)
}
rows, err := r.db.QueryContext(ctx, `
SELECT country_id, COALESCE(MAX(region_id),0),
@ -561,8 +559,8 @@ func (r *Repository) queryLuckyGiftPools(ctx context.Context, app, startDay, end
args = append(args, countryID)
}
if regionID > 0 {
filter += " AND (region_id = ? OR (region_id = 0 AND country_id = ?))"
args = append(args, regionID, regionID)
filter += " AND region_id = ?"
args = append(args, regionID)
}
rows, err := r.db.QueryContext(ctx, `
SELECT pool_id, COALESCE(SUM(turnover_coin),0), COALESCE(SUM(payout_coin),0)

100
services/statistics-service/internal/storage/mysql/repository.go
View File

@ -113,7 +113,7 @@ func (r *Repository) Migrate(ctx context.Context) error {
lucky_gift_turnover BIGINT NOT NULL DEFAULT 0, lucky_gift_payout BIGINT NOT NULL DEFAULT 0, lucky_gift_payers BIGINT NOT NULL DEFAULT 0,
game_turnover BIGINT NOT NULL DEFAULT 0, game_players BIGINT NOT NULL DEFAULT 0,
game_payout BIGINT NOT NULL DEFAULT 0, game_refund BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_day, country_id),
PRIMARY KEY (app_code, stat_day, country_id, region_id),
KEY idx_stat_app_day_region (app_code, stat_day, region_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS stat_user_day_activity (
@ -150,7 +150,7 @@ func (r *Repository) Migrate(ctx context.Context) error {
region_id BIGINT NOT NULL DEFAULT 0,
pool_id VARCHAR(96) NOT NULL, turnover_coin BIGINT NOT NULL DEFAULT 0, payout_coin BIGINT NOT NULL DEFAULT 0,
updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_day, country_id, pool_id),
PRIMARY KEY (app_code, stat_day, country_id, region_id, pool_id),
KEY idx_stat_lucky_pool_overview (app_code, stat_day, pool_id),
KEY idx_stat_lucky_pool_region (app_code, stat_day, region_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
@ -160,7 +160,7 @@ func (r *Repository) Migrate(ctx context.Context) error {
platform_code VARCHAR(64) NOT NULL DEFAULT '', game_id VARCHAR(96) NOT NULL,
turnover_coin BIGINT NOT NULL DEFAULT 0, payout_coin BIGINT NOT NULL DEFAULT 0, refund_coin BIGINT NOT NULL DEFAULT 0,
player_count BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id),
PRIMARY KEY (app_code, stat_day, country_id, region_id, platform_code, game_id),
KEY idx_stat_game_day_region (app_code, stat_day, region_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS stat_game_day_players (
@ -205,9 +205,60 @@ func (r *Repository) Migrate(ctx context.Context) error {
return err
}
}
for table, columns := range map[string][]string{
"stat_app_day_country": {"app_code", "stat_day", "country_id", "region_id"},
"stat_lucky_gift_pool_day_country": {"app_code", "stat_day", "country_id", "region_id", "pool_id"},
"stat_game_day_country": {"app_code", "stat_day", "country_id", "region_id", "platform_code", "game_id"},
} {
if err := r.ensurePrimaryKeyColumns(ctx, table, columns); err != nil {
return err
}
}
return nil
}
func (r *Repository) ensurePrimaryKeyColumns(ctx context.Context, table string, expected []string) error {
rows, err := r.db.QueryContext(ctx, `
SELECT COLUMN_NAME
FROM information_schema.KEY_COLUMN_USAGE
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND CONSTRAINT_NAME = 'PRIMARY'
ORDER BY ORDINAL_POSITION`, table)
if err != nil {
return err
}
defer rows.Close()
actual := make([]string, 0, len(expected))
for rows.Next() {
var column string
if err := rows.Scan(&column); err != nil {
return err
}
actual = append(actual, column)
}
if err := rows.Err(); err != nil {
return err
}
if sameStringSlice(actual, expected) {
return nil
}
// 统计聚合表必须同时以国家和区域做主键；否则区域筛选和国家筛选会写入同一行，导致阿富汗这类历史错位国家继续污染榜单。
_, err = r.db.ExecContext(ctx, "ALTER TABLE "+table+" DROP PRIMARY KEY, ADD PRIMARY KEY ("+strings.Join(expected, ", ")+")")
return err
}
func sameStringSlice(left, right []string) bool {
if len(left) != len(right) {
return false
}
for index := range left {
if left[index] != right[index] {
return false
}
}
return true
}
type UserRegisteredEvent struct {
AppCode string
EventID string
@ -305,8 +356,8 @@ func (r *Repository) ConsumeUserRegistered(ctx context.Context, event UserRegist
_, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET new_users = new_users + 1, updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ?
`, nowMS, appcode.Normalize(event.AppCode), day, countryID)
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
return err
})
}
@ -350,8 +401,8 @@ func (r *Repository) ConsumeRecharge(ctx context.Context, event RechargeEvent) e
coin_seller_transfer_coin = coin_seller_transfer_coin + ?,
paid_users = paid_users + ?,
updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ?
`, event.USDMinor, newUserUSD, coinSellerUSD, mifapayUSD, googleUSD, coinSellerTransferCoin, paidUsers, nowMS, appcode.Normalize(event.AppCode), day, countryID)
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, event.USDMinor, newUserUSD, coinSellerUSD, mifapayUSD, googleUSD, coinSellerTransferCoin, paidUsers, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
return err
})
}
@ -369,8 +420,8 @@ func (r *Repository) ConsumeCoinSellerStock(ctx context.Context, event CoinSelle
SET recharge_usd_minor = recharge_usd_minor + ?,
coin_seller_recharge_usd_minor = coin_seller_recharge_usd_minor + ?,
updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ?
`, event.USDMinor, event.USDMinor, nowMS, appcode.Normalize(event.AppCode), day, countryID)
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, event.USDMinor, event.USDMinor, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
return err
})
}
@ -401,8 +452,8 @@ func (r *Repository) ConsumeRoomGift(ctx context.Context, event RoomGiftEvent) e
if _, err := tx.ExecContext(ctx, `
UPDATE stat_lucky_gift_pool_day_country
SET turnover_coin = turnover_coin + ?, updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND pool_id = ?
`, luckyTurnover, nowMS, appcode.Normalize(event.AppCode), day, countryID, poolID); err != nil {
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ? AND pool_id = ?
`, luckyTurnover, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID, poolID); err != nil {
return err
}
}
@ -410,8 +461,8 @@ func (r *Repository) ConsumeRoomGift(ctx context.Context, event RoomGiftEvent) e
UPDATE stat_app_day_country
SET gift_coin_spent = gift_coin_spent + ?, lucky_gift_turnover = lucky_gift_turnover + ?,
lucky_gift_payers = lucky_gift_payers + ?, updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ?
`, event.CoinSpent, luckyTurnover, luckyPayers, nowMS, appcode.Normalize(event.AppCode), day, countryID)
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, event.CoinSpent, luckyTurnover, luckyPayers, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
return err
})
}
@ -432,16 +483,16 @@ func (r *Repository) ConsumeLuckyGiftReward(ctx context.Context, event LuckyGift
if _, err := tx.ExecContext(ctx, `
UPDATE stat_lucky_gift_pool_day_country
SET payout_coin = payout_coin + ?, updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND pool_id = ?
`, event.Amount, nowMS, appcode.Normalize(event.AppCode), day, countryID, poolID); err != nil {
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ? AND pool_id = ?
`, event.Amount, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID, poolID); err != nil {
return err
}
}
_, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET lucky_gift_payout = lucky_gift_payout + ?, updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ?
`, event.Amount, nowMS, appcode.Normalize(event.AppCode), day, countryID)
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, event.Amount, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
return err
})
}
@ -490,16 +541,16 @@ func (r *Repository) ConsumeGameOrder(ctx context.Context, event GameOrderEvent)
UPDATE stat_game_day_country
SET turnover_coin = turnover_coin + ?, payout_coin = payout_coin + ?, refund_coin = refund_coin + ?,
player_count = player_count + ?, updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND platform_code = ? AND game_id = ?
`, turnover, payout, refund, playerDelta, nowMS, appcode.Normalize(event.AppCode), day, countryID, event.PlatformCode, event.GameID); err != nil {
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ? AND platform_code = ? AND game_id = ?
`, turnover, payout, refund, playerDelta, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID, event.PlatformCode, event.GameID); err != nil {
return err
}
_, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET game_turnover = game_turnover + ?, game_payout = game_payout + ?, game_refund = game_refund + ?,
game_players = game_players + ?, updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ?
`, turnover, payout, refund, playerDelta, nowMS, appcode.Normalize(event.AppCode), day, countryID)
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, turnover, payout, refund, playerDelta, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
return err
})
}
@ -596,8 +647,8 @@ func applyActive(ctx context.Context, tx *sql.Tx, app string, day string, countr
_, err = tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET active_users = active_users + 1, updated_at_ms = ?
WHERE app_code = ? AND stat_day = ? AND country_id = ?
`, nowMS, appcode.Normalize(app), day, countryID)
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, nowMS, appcode.Normalize(app), day, countryID, regionID)
return err
}
@ -606,7 +657,6 @@ func ensureAppDay(ctx context.Context, tx *sql.Tx, app string, day string, count
INSERT INTO stat_app_day_country (app_code, stat_day, country_id, region_id, updated_at_ms)
VALUES (?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
region_id = IF(region_id = 0 AND VALUES(region_id) > 0, VALUES(region_id), region_id),
updated_at_ms = VALUES(updated_at_ms)
`, appcode.Normalize(app), day, countryID, regionID, nowMS)
return err
@ -617,7 +667,6 @@ func ensureGameDay(ctx context.Context, tx *sql.Tx, app string, day string, coun
INSERT INTO stat_game_day_country (app_code, stat_day, country_id, region_id, platform_code, game_id, updated_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
region_id = IF(region_id = 0 AND VALUES(region_id) > 0, VALUES(region_id), region_id),
updated_at_ms = VALUES(updated_at_ms)
`, appcode.Normalize(app), day, countryID, regionID, platformCode, gameID, nowMS)
return err
@ -628,7 +677,6 @@ func ensureLuckyGiftPoolDay(ctx context.Context, tx *sql.Tx, app string, day str
INSERT INTO stat_lucky_gift_pool_day_country (app_code, stat_day, country_id, region_id, pool_id, updated_at_ms)
VALUES (?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
region_id = IF(region_id = 0 AND VALUES(region_id) > 0, VALUES(region_id), region_id),
updated_at_ms = VALUES(updated_at_ms)
`, appcode.Normalize(app), day, countryID, regionID, strings.TrimSpace(poolID), nowMS)
return err

21
services/wallet-service/internal/domain/ledger/ledger.go
View File

@ -243,14 +243,15 @@ type HostSalaryProgress struct {
// CoinSellerTransferCommand 是币商给玩家转普通金币的账务命令。
type CoinSellerTransferCommand struct {
AppCode string
CommandID string
SellerUserID int64
TargetUserID int64
SellerRegionID int64
TargetRegionID int64
Amount int64
Reason string
AppCode string
CommandID string
SellerUserID int64
TargetUserID int64
TargetCountryID int64
SellerRegionID int64
TargetRegionID int64
Amount int64
Reason string
}
// CoinSellerTransferReceipt 是币商转账完成后的稳定回执。
@ -335,6 +336,8 @@ type CoinSellerStockCreditCommand struct {
AppCode string
CommandID string
SellerUserID int64
SellerCountryID int64
SellerRegionID int64
StockType string
CoinAmount int64
PaidCurrencyCode string
@ -349,6 +352,8 @@ type CoinSellerStockCreditCommand struct {
type CoinSellerStockCreditReceipt struct {
TransactionID string
SellerUserID int64
SellerCountryID int64
SellerRegionID int64
StockType string
CoinAmount int64
PaidCurrencyCode string

6
services/wallet-service/internal/service/wallet/service.go
View File

@ -315,6 +315,9 @@ func (s *Service) AdminCreditCoinSellerStock(ctx context.Context, command ledger
if command.CommandID == "" || command.SellerUserID <= 0 || command.OperatorUserID <= 0 {
return ledger.CoinSellerStockCreditReceipt{}, xerr.New(xerr.InvalidArgument, "coin seller stock command is incomplete")
}
if command.SellerCountryID <= 0 || command.SellerRegionID <= 0 {
return ledger.CoinSellerStockCreditReceipt{}, xerr.New(xerr.InvalidArgument, "seller country and region are required")
}
if len(command.CommandID) > 128 {
return ledger.CoinSellerStockCreditReceipt{}, xerr.New(xerr.InvalidArgument, "command_id is too long")
}
@ -364,6 +367,9 @@ func (s *Service) TransferCoinFromSeller(ctx context.Context, command ledger.Coi
if command.SellerRegionID <= 0 || command.TargetRegionID <= 0 {
return ledger.CoinSellerTransferReceipt{}, xerr.New(xerr.InvalidArgument, "seller and target region are required")
}
if command.TargetCountryID <= 0 {
return ledger.CoinSellerTransferReceipt{}, xerr.New(xerr.InvalidArgument, "target country is required")
}
if command.SellerRegionID != command.TargetRegionID {
return ledger.CoinSellerTransferReceipt{}, xerr.New(xerr.Conflict, "seller and target region must match")
}

120
services/wallet-service/internal/service/wallet/service_test.go
View File

@ -1448,14 +1448,15 @@ func TestWalletOutboxClaimUsesRealMySQLStatusFlow(t *testing.T) {
ctx := appcode.WithContext(context.Background(), "lalu")
receipt, err := svc.TransferCoinFromSeller(ctx, ledger.CoinSellerTransferCommand{
AppCode: "lalu",
CommandID: "cmd-real-outbox-transfer",
SellerUserID: 30001,
TargetUserID: 30002,
SellerRegionID: 86,
TargetRegionID: 86,
Amount: 200,
Reason: "real outbox claim test",
AppCode: "lalu",
CommandID: "cmd-real-outbox-transfer",
SellerUserID: 30001,
TargetUserID: 30002,
TargetCountryID: 63,
SellerRegionID: 86,
TargetRegionID: 86,
Amount: 200,
Reason: "real outbox claim test",
})
if err != nil {
t.Fatalf("TransferCoinFromSeller failed: %v", err)
@ -1593,13 +1594,14 @@ func TestTransferCoinFromSellerMovesDedicatedAssetToPlayerCoin(t *testing.T) {
}
command := ledger.CoinSellerTransferCommand{
CommandID: "cmd-seller-transfer",
SellerUserID: 30001,
TargetUserID: 30002,
SellerRegionID: 1001,
TargetRegionID: 1001,
Amount: 80000,
Reason: "player recharge",
CommandID: "cmd-seller-transfer",
SellerUserID: 30001,
TargetUserID: 30002,
TargetCountryID: 63,
SellerRegionID: 1001,
TargetRegionID: 1001,
Amount: 80000,
Reason: "player recharge",
}
first, err := svc.TransferCoinFromSeller(context.Background(), command)
if err != nil {
@ -1639,6 +1641,9 @@ func TestTransferCoinFromSellerMovesDedicatedAssetToPlayerCoin(t *testing.T) {
if got := repository.CountRows("wallet_outbox", "transaction_id = ?", first.TransactionID); got != 4 {
t.Fatalf("seller transfer should write two balance events, one transfer event and one recharge event, got %d", got)
}
if got := repository.CountRows("wallet_outbox", "transaction_id = ? AND event_type = ? AND JSON_EXTRACT(payload_json, '$.target_country_id') = 63 AND JSON_EXTRACT(payload_json, '$.country_id') = 63 AND JSON_EXTRACT(payload_json, '$.target_region_id') = 1001", first.TransactionID, "WalletRechargeRecorded"); got != 1 {
t.Fatalf("seller transfer recharge event must carry target country/region for statistics, got %d", got)
}
}
// TestTransferCoinFromSellerAllowsSelfTransfer 验证币商可以把自己的专用库存转入自己的普通金币账户。
@ -1659,13 +1664,14 @@ func TestTransferCoinFromSellerAllowsSelfTransfer(t *testing.T) {
}
command := ledger.CoinSellerTransferCommand{
CommandID: "cmd-seller-self-transfer",
SellerUserID: 30003,
TargetUserID: 30003,
SellerRegionID: 1001,
TargetRegionID: 1001,
Amount: 30000,
Reason: "self recharge",
CommandID: "cmd-seller-self-transfer",
SellerUserID: 30003,
TargetUserID: 30003,
TargetCountryID: 63,
SellerRegionID: 1001,
TargetRegionID: 1001,
Amount: 30000,
Reason: "self recharge",
}
first, err := svc.TransferCoinFromSeller(context.Background(), command)
if err != nil {
@ -1718,13 +1724,14 @@ func TestTransferCoinFromSellerRecordsCoinsWithoutUSDConversion(t *testing.T) {
}
receipt, err := svc.TransferCoinFromSeller(context.Background(), ledger.CoinSellerTransferCommand{
CommandID: "cmd-seller-small-transfer",
SellerUserID: 30004,
TargetUserID: 30004,
SellerRegionID: 1001,
TargetRegionID: 1001,
Amount: 1,
Reason: "self recharge",
CommandID: "cmd-seller-small-transfer",
SellerUserID: 30004,
TargetUserID: 30004,
TargetCountryID: 63,
SellerRegionID: 1001,
TargetRegionID: 1001,
Amount: 1,
Reason: "self recharge",
})
if err != nil {
t.Fatalf("TransferCoinFromSeller small amount failed: %v", err)
@ -1750,13 +1757,14 @@ func TestTransferCoinFromSellerInsufficientBalance(t *testing.T) {
svc := walletservice.New(repository)
_, err := svc.TransferCoinFromSeller(context.Background(), ledger.CoinSellerTransferCommand{
CommandID: "cmd-seller-low",
SellerUserID: 31001,
TargetUserID: 31002,
SellerRegionID: 1001,
TargetRegionID: 1001,
Amount: 1,
Reason: "player recharge",
CommandID: "cmd-seller-low",
SellerUserID: 31001,
TargetUserID: 31002,
TargetCountryID: 63,
SellerRegionID: 1001,
TargetRegionID: 1001,
Amount: 1,
Reason: "player recharge",
})
if !xerr.IsCode(err, xerr.InsufficientBalance) {
t.Fatalf("expected INSUFFICIENT_BALANCE, got %v", err)
@ -1784,13 +1792,14 @@ func TestTransferCoinFromSellerDoesNotRequireRechargePolicy(t *testing.T) {
}
receipt, err := svc.TransferCoinFromSeller(context.Background(), ledger.CoinSellerTransferCommand{
CommandID: "cmd-seller-no-policy",
SellerUserID: 32001,
TargetUserID: 32002,
SellerRegionID: 2001,
TargetRegionID: 2001,
Amount: 80000,
Reason: "player recharge",
CommandID: "cmd-seller-no-policy",
SellerUserID: 32001,
TargetUserID: 32002,
TargetCountryID: 63,
SellerRegionID: 2001,
TargetRegionID: 2001,
Amount: 80000,
Reason: "player recharge",
})
if err != nil {
t.Fatalf("TransferCoinFromSeller without recharge policy failed: %v", err)
@ -1945,6 +1954,8 @@ func TestAdminCreditCoinSellerStockPurchaseCreditsDedicatedBalance(t *testing.T)
command := ledger.CoinSellerStockCreditCommand{
CommandID: "cmd-stock-usdt",
SellerUserID: 33001,
SellerCountryID: 63,
SellerRegionID: 7,
StockType: ledger.StockTypeUSDTPurchase,
CoinAmount: 8000000,
PaidAmountMicro: 100000000,
@ -1982,6 +1993,9 @@ func TestAdminCreditCoinSellerStockPurchaseCreditsDedicatedBalance(t *testing.T)
if got := repository.CountRows("wallet_outbox", "transaction_id = ?", first.TransactionID); got != 2 {
t.Fatalf("stock purchase should write balance and stock outbox events, got %d", got)
}
if got := repository.CountRows("wallet_outbox", "transaction_id = ? AND event_type = ? AND JSON_EXTRACT(payload_json, '$.seller_country_id') = 63 AND JSON_EXTRACT(payload_json, '$.seller_region_id') = 7 AND JSON_EXTRACT(payload_json, '$.country_id') = 63 AND JSON_EXTRACT(payload_json, '$.region_id') = 7", first.TransactionID, "WalletCoinSellerStockPurchased"); got != 1 {
t.Fatalf("stock outbox must carry seller country/region for statistics, got %d", got)
}
}
// TestAdminCreditCoinSellerStockCompensationRejectsPaidAmount 锁定补偿不能携带充值金额的边界。
@ -1991,6 +2005,8 @@ func TestAdminCreditCoinSellerStockCompensationRejectsPaidAmount(t *testing.T) {
_, err := svc.AdminCreditCoinSellerStock(context.Background(), ledger.CoinSellerStockCreditCommand{
CommandID: "cmd-stock-comp-invalid",
SellerUserID: 34001,
SellerCountryID: 63,
SellerRegionID: 7,
StockType: ledger.StockTypeCoinCompensation,
CoinAmount: 50000,
PaidCurrencyCode: ledger.PaidCurrencyUSDT,
@ -2010,11 +2026,13 @@ func TestAdminCreditCoinSellerStockCompensationCreditsWithoutRechargeAmount(t *t
repository := mysqltest.NewRepository(t)
svc := walletservice.New(repository)
receipt, err := svc.AdminCreditCoinSellerStock(context.Background(), ledger.CoinSellerStockCreditCommand{
CommandID: "cmd-stock-comp",
SellerUserID: 35001,
StockType: ledger.StockTypeCoinCompensation,
CoinAmount: 50000,
OperatorUserID: 90001,
CommandID: "cmd-stock-comp",
SellerUserID: 35001,
SellerCountryID: 63,
SellerRegionID: 7,
StockType: ledger.StockTypeCoinCompensation,
CoinAmount: 50000,
OperatorUserID: 90001,
})
if err != nil {
t.Fatalf("AdminCreditCoinSellerStock compensation failed: %v", err)
@ -2034,6 +2052,8 @@ func TestAdminCreditCoinSellerStockRejectsDuplicatePaymentRef(t *testing.T) {
command := ledger.CoinSellerStockCreditCommand{
CommandID: "cmd-stock-ref-1",
SellerUserID: 36001,
SellerCountryID: 63,
SellerRegionID: 7,
StockType: ledger.StockTypeUSDTPurchase,
CoinAmount: 100,
PaidAmountMicro: 1000000,
@ -2047,6 +2067,8 @@ func TestAdminCreditCoinSellerStockRejectsDuplicatePaymentRef(t *testing.T) {
}
command.CommandID = "cmd-stock-ref-2"
command.SellerUserID = 36002
command.SellerCountryID = 971
command.SellerRegionID = 8
command.EvidenceRef = "oss://receipt-dup-2"
_, err := svc.AdminCreditCoinSellerStock(context.Background(), command)
if !xerr.IsCode(err, xerr.CoinSellerPaymentRefDuplicated) {
@ -2064,6 +2086,8 @@ func TestAdminCreditCoinSellerStockRejectsIdempotencyConflict(t *testing.T) {
command := ledger.CoinSellerStockCreditCommand{
CommandID: "cmd-stock-conflict",
SellerUserID: 37001,
SellerCountryID: 63,
SellerRegionID: 7,
StockType: ledger.StockTypeUSDTPurchase,
CoinAmount: 100,
PaidAmountMicro: 1000000,

55
services/wallet-service/internal/storage/mysql/repository.go
View File

@ -1142,6 +1142,8 @@ func (r *Repository) AdminCreditCoinSellerStock(ctx context.Context, command led
metadata := coinSellerStockMetadata{
AppCode: command.AppCode,
SellerUserID: command.SellerUserID,
SellerCountryID: command.SellerCountryID,
SellerRegionID: command.SellerRegionID,
StockType: command.StockType,
CoinAmount: command.CoinAmount,
PaidCurrencyCode: command.PaidCurrencyCode,
@ -1241,6 +1243,7 @@ func (r *Repository) TransferCoinFromSeller(ctx context.Context, command ledger.
AppCode: command.AppCode,
SellerUserID: command.SellerUserID,
TargetUserID: command.TargetUserID,
TargetCountryID: command.TargetCountryID,
SellerRegionID: command.SellerRegionID,
TargetRegionID: command.TargetRegionID,
Amount: command.Amount,
@ -2676,6 +2679,7 @@ type coinSellerTransferMetadata struct {
AppCode string `json:"app_code"`
SellerUserID int64 `json:"seller_user_id"`
TargetUserID int64 `json:"target_user_id"`
TargetCountryID int64 `json:"target_country_id"`
SellerRegionID int64 `json:"seller_region_id"`
TargetRegionID int64 `json:"target_region_id"`
Amount int64 `json:"amount"`
@ -2728,6 +2732,8 @@ type salaryTransferToCoinSellerMetadata struct {
type coinSellerStockMetadata struct {
AppCode string `json:"app_code"`
SellerUserID int64 `json:"seller_user_id"`
SellerCountryID int64 `json:"seller_country_id"`
SellerRegionID int64 `json:"seller_region_id"`
StockType string `json:"stock_type"`
CoinAmount int64 `json:"coin_amount"`
PaidCurrencyCode string `json:"paid_currency_code"`
@ -2889,13 +2895,14 @@ func receiptFromGameMetadata(transactionID string, metadata gameCoinMetadata, id
func (r *Repository) receiptForCoinSellerStockTransaction(ctx context.Context, tx *sql.Tx, transactionID string) (ledger.CoinSellerStockCreditReceipt, error) {
row := tx.QueryRowContext(ctx,
`SELECT seller_user_id, stock_type, coin_amount, paid_currency_code, paid_amount_micro,
counts_as_seller_recharge, balance_after, created_at_ms
counts_as_seller_recharge, balance_after, metadata_json, created_at_ms
FROM coin_seller_stock_records
WHERE app_code = ? AND transaction_id = ?`,
appcode.FromContext(ctx),
transactionID,
)
var receipt ledger.CoinSellerStockCreditReceipt
var metadataJSON string
if err := row.Scan(
&receipt.SellerUserID,
&receipt.StockType,
@ -2904,10 +2911,18 @@ func (r *Repository) receiptForCoinSellerStockTransaction(ctx context.Context, t
&receipt.PaidAmountMicro,
&receipt.CountsAsSellerRecharge,
&receipt.BalanceAfter,
&metadataJSON,
&receipt.CreatedAtMS,
); err != nil {
return ledger.CoinSellerStockCreditReceipt{}, err
}
if metadataJSON != "" {
var metadata coinSellerStockMetadata
if err := json.Unmarshal([]byte(metadataJSON), &metadata); err == nil {
receipt.SellerCountryID = metadata.SellerCountryID
receipt.SellerRegionID = metadata.SellerRegionID
}
}
receipt.TransactionID = transactionID
return receipt, nil
}
@ -2916,6 +2931,8 @@ func receiptFromCoinSellerStockMetadata(transactionID string, metadata coinSelle
return ledger.CoinSellerStockCreditReceipt{
TransactionID: transactionID,
SellerUserID: metadata.SellerUserID,
SellerCountryID: metadata.SellerCountryID,
SellerRegionID: metadata.SellerRegionID,
StockType: metadata.StockType,
CoinAmount: metadata.CoinAmount,
PaidCurrencyCode: metadata.PaidCurrencyCode,
@ -3070,6 +3087,34 @@ func coinSellerTransferredEvent(transactionID string, commandID string, sellerUs
}
func rechargeRecordedEvent(transactionID string, commandID string, targetUserID int64, availableDelta int64, payload any, nowMs int64) walletOutboxEvent {
if metadata, ok := payload.(coinSellerTransferMetadata); ok {
// WalletRechargeRecorded 是统计服务的充值入口；币商转账没有 USDT 金额，但必须带目标用户真实国家/区域，避免统计继续把 region 当 country。
payload = map[string]any{
"app_code": metadata.AppCode,
"seller_user_id": metadata.SellerUserID,
"target_user_id": metadata.TargetUserID,
"target_country_id": metadata.TargetCountryID,
"country_id": metadata.TargetCountryID,
"seller_region_id": metadata.SellerRegionID,
"target_region_id": metadata.TargetRegionID,
"region_id": metadata.TargetRegionID,
"amount": metadata.Amount,
"reason": metadata.Reason,
"seller_asset_type": metadata.SellerAssetType,
"target_asset_type": metadata.TargetAssetType,
"seller_balance_after": metadata.SellerBalanceAfter,
"target_balance_after": metadata.TargetBalanceAfter,
"recharge_sequence": metadata.RechargeSequence,
"recharge_usd_minor": metadata.RechargeUSDMinor,
"recharge_currency_code": metadata.RechargeCurrencyCode,
"recharge_policy_id": metadata.RechargePolicyID,
"recharge_policy_version": metadata.RechargePolicyVersion,
"recharge_policy_coin_amount": metadata.RechargePolicyCoinAmount,
"recharge_policy_usd_minor_amount": metadata.RechargePolicyUSDMinorAmount,
"recharge_type": "coin_seller_transfer",
"created_at_ms": nowMs,
}
}
return walletOutboxEvent{
EventID: eventID(transactionID, "WalletRechargeRecorded", targetUserID, ledger.AssetCoin),
EventType: "WalletRechargeRecorded",
@ -3184,6 +3229,10 @@ func coinSellerStockCreditedEvent(transactionID string, commandID string, metada
"transaction_id": transactionID,
"command_id": commandID,
"seller_user_id": metadata.SellerUserID,
"seller_country_id": metadata.SellerCountryID,
"seller_region_id": metadata.SellerRegionID,
"country_id": metadata.SellerCountryID,
"region_id": metadata.SellerRegionID,
"stock_type": metadata.StockType,
"coin_amount": metadata.CoinAmount,
"paid_currency_code": metadata.PaidCurrencyCode,
@ -3451,9 +3500,11 @@ func gameBizTypeAndDelta(opType string, coinAmount int64) (string, int64, error)
}
func coinSellerStockRequestHash(command ledger.CoinSellerStockCreditCommand) string {
return stableHash(fmt.Sprintf("coin_seller_stock|%s|%d|%s|%d|%s|%d|%s|%s|%d|%s",
return stableHash(fmt.Sprintf("coin_seller_stock|%s|%d|%d|%d|%s|%d|%s|%d|%s|%s|%d|%s",
appcode.Normalize(command.AppCode),
command.SellerUserID,
command.SellerCountryID,
command.SellerRegionID,
command.StockType,
command.CoinAmount,
command.PaidCurrencyCode,

19
services/wallet-service/internal/transport/grpc/server.go
View File

@ -842,6 +842,8 @@ func (s *Server) AdminCreditCoinSellerStock(ctx context.Context, req *walletv1.A
AppCode: req.GetAppCode(),
CommandID: req.GetCommandId(),
SellerUserID: req.GetSellerUserId(),
SellerCountryID: req.GetSellerCountryId(),
SellerRegionID: req.GetSellerRegionId(),
StockType: req.GetStockType(),
CoinAmount: req.GetCoinAmount(),
PaidCurrencyCode: req.GetPaidCurrencyCode(),
@ -872,14 +874,15 @@ func (s *Server) AdminCreditCoinSellerStock(ctx context.Context, req *walletv1.A
func (s *Server) TransferCoinFromSeller(ctx context.Context, req *walletv1.TransferCoinFromSellerRequest) (*walletv1.TransferCoinFromSellerResponse, error) {
ctx = appcode.WithContext(ctx, req.GetAppCode())
receipt, err := s.svc.TransferCoinFromSeller(ctx, ledger.CoinSellerTransferCommand{
AppCode: req.GetAppCode(),
CommandID: req.GetCommandId(),
SellerUserID: req.GetSellerUserId(),
TargetUserID: req.GetTargetUserId(),
SellerRegionID: req.GetSellerRegionId(),
TargetRegionID: req.GetTargetRegionId(),
Amount: req.GetAmount(),
Reason: req.GetReason(),
AppCode: req.GetAppCode(),
CommandID: req.GetCommandId(),
SellerUserID: req.GetSellerUserId(),
TargetUserID: req.GetTargetUserId(),
TargetCountryID: req.GetTargetCountryId(),
SellerRegionID: req.GetSellerRegionId(),
TargetRegionID: req.GetTargetRegionId(),
Amount: req.GetAmount(),
Reason: req.GetReason(),
})
if err != nil {
return nil, xerr.ToGRPCError(err)