diff --git a/api/proto/user/v1/user.pb.go b/api/proto/user/v1/user.pb.go index 7ce4bd73..33ab235a 100644 --- a/api/proto/user/v1/user.pb.go +++ b/api/proto/user/v1/user.pb.go @@ -15143,7 +15143,7 @@ const file_proto_user_v1_user_proto_rawDesc = "" + "\x17ListCPFormationGiftFeed\x12-.hyapp.user.v1.ListCPFormationGiftFeedRequest\x1a..hyapp.user.v1.ListCPFormationGiftFeedResponse2\x88\x02\n" + "\x15UserCPInternalService\x12u\n" + "\x16ConsumeRoomGiftCPEvent\x12,.hyapp.user.v1.ConsumeRoomGiftCPEventRequest\x1a-.hyapp.user.v1.ConsumeRoomGiftCPEventResponse\x12x\n" + - "\x17ListCPWeeklyRankEntries\x12-.hyapp.user.v1.ListCPWeeklyRankEntriesRequest\x1a..hyapp.user.v1.ListCPWeeklyRankEntriesResponse2\xab\x05\n" + + "\x17ListCPWeeklyRankEntries\x12-.hyapp.user.v1.ListCPWeeklyRankEntriesRequest\x1a..hyapp.user.v1.ListCPWeeklyRankEntriesResponse2\x93\x06\n" + "\x0fUserCronService\x12\\\n" + "\x17ProcessLoginIPRiskBatch\x12\x1f.hyapp.user.v1.CronBatchRequest\x1a .hyapp.user.v1.CronBatchResponse\x12^\n" + "\x19ProcessRegionRebuildBatch\x12\x1f.hyapp.user.v1.CronBatchRequest\x1a .hyapp.user.v1.CronBatchResponse\x12^\n" + @@ -15151,7 +15151,8 @@ const file_proto_user_v1_user_proto_rawDesc = "" + "\x1aCompensateRoomOpenSessions\x12\x1f.hyapp.user.v1.CronBatchRequest\x1a .hyapp.user.v1.CronBatchResponse\x12\\\n" + "\x17ExpireManagerUserBlocks\x12\x1f.hyapp.user.v1.CronBatchRequest\x1a .hyapp.user.v1.CronBatchResponse\x12X\n" + "\x13ExpireAdminUserBans\x12\x1f.hyapp.user.v1.CronBatchRequest\x1a .hyapp.user.v1.CronBatchResponse\x12a\n" + - "\x1cRefreshCPIntimacyLeaderboard\x12\x1f.hyapp.user.v1.CronBatchRequest\x1a .hyapp.user.v1.CronBatchResponse2\xd1\x01\n" + + "\x1cRefreshCPIntimacyLeaderboard\x12\x1f.hyapp.user.v1.CronBatchRequest\x1a .hyapp.user.v1.CronBatchResponse\x12f\n" + + "!ProcessPrivateMessageHistoryBatch\x12\x1f.hyapp.user.v1.CronBatchRequest\x1a .hyapp.user.v1.CronBatchResponse2\xd1\x01\n" + "\x11UserDeviceService\x12Z\n" + "\rBindPushToken\x12#.hyapp.user.v1.BindPushTokenRequest\x1a$.hyapp.user.v1.BindPushTokenResponse\x12`\n" + "\x0fDeletePushToken\x12%.hyapp.user.v1.DeletePushTokenRequest\x1a&.hyapp.user.v1.DeletePushTokenResponse2\xb4\x01\n" + @@ -15634,120 +15635,122 @@ var file_proto_user_v1_user_proto_depIdxs = []int32{ 14, // 232: hyapp.user.v1.UserCronService.ExpireManagerUserBlocks:input_type -> hyapp.user.v1.CronBatchRequest 14, // 233: hyapp.user.v1.UserCronService.ExpireAdminUserBans:input_type -> hyapp.user.v1.CronBatchRequest 14, // 234: hyapp.user.v1.UserCronService.RefreshCPIntimacyLeaderboard:input_type -> hyapp.user.v1.CronBatchRequest - 130, // 235: hyapp.user.v1.UserDeviceService.BindPushToken:input_type -> hyapp.user.v1.BindPushTokenRequest - 132, // 236: hyapp.user.v1.UserDeviceService.DeletePushToken:input_type -> hyapp.user.v1.DeletePushTokenRequest - 3, // 237: hyapp.user.v1.AppRegistryService.ResolveApp:input_type -> hyapp.user.v1.ResolveAppRequest - 186, // 238: hyapp.user.v1.AppRegistryService.ListApps:input_type -> hyapp.user.v1.ListAppsRequest - 136, // 239: hyapp.user.v1.CountryAdminService.ListCountries:input_type -> hyapp.user.v1.ListCountriesRequest - 138, // 240: hyapp.user.v1.CountryAdminService.UpdateCountry:input_type -> hyapp.user.v1.UpdateCountryRequest - 140, // 241: hyapp.user.v1.CountryQueryService.ListRegistrationCountries:input_type -> hyapp.user.v1.ListRegistrationCountriesRequest - 143, // 242: hyapp.user.v1.CountryQueryService.ListLoginRiskBlockedCountries:input_type -> hyapp.user.v1.ListLoginRiskBlockedCountriesRequest - 145, // 243: hyapp.user.v1.RegionAdminService.ListRegions:input_type -> hyapp.user.v1.ListRegionsRequest - 147, // 244: hyapp.user.v1.RegionAdminService.GetRegion:input_type -> hyapp.user.v1.GetRegionRequest - 148, // 245: hyapp.user.v1.RegionAdminService.UpdateRegion:input_type -> hyapp.user.v1.UpdateRegionRequest - 149, // 246: hyapp.user.v1.RegionAdminService.ReplaceRegionCountries:input_type -> hyapp.user.v1.ReplaceRegionCountriesRequest - 152, // 247: hyapp.user.v1.UserIdentityService.GetUserIdentity:input_type -> hyapp.user.v1.GetUserIdentityRequest - 154, // 248: hyapp.user.v1.UserIdentityService.ResolveDisplayUserID:input_type -> hyapp.user.v1.ResolveDisplayUserIDRequest - 188, // 249: hyapp.user.v1.UserIdentityService.ResolveAdminUserIdentifier:input_type -> hyapp.user.v1.ResolveAdminUserIdentifierRequest - 156, // 250: hyapp.user.v1.UserIdentityService.ChangeDisplayUserID:input_type -> hyapp.user.v1.ChangeDisplayUserIDRequest - 158, // 251: hyapp.user.v1.UserIdentityService.ApplyPrettyDisplayUserID:input_type -> hyapp.user.v1.ApplyPrettyDisplayUserIDRequest - 165, // 252: hyapp.user.v1.UserIdentityService.ListAvailablePrettyDisplayIDs:input_type -> hyapp.user.v1.ListAvailablePrettyDisplayIDsRequest - 167, // 253: hyapp.user.v1.UserIdentityService.ApplyPrettyDisplayIDFromPool:input_type -> hyapp.user.v1.ApplyPrettyDisplayIDFromPoolRequest - 160, // 254: hyapp.user.v1.UserIdentityService.ExpirePrettyDisplayUserID:input_type -> hyapp.user.v1.ExpirePrettyDisplayUserIDRequest - 169, // 255: hyapp.user.v1.UserPrettyDisplayIDAdminService.ListPrettyDisplayIDPools:input_type -> hyapp.user.v1.ListPrettyDisplayIDPoolsRequest - 171, // 256: hyapp.user.v1.UserPrettyDisplayIDAdminService.CreatePrettyDisplayIDPool:input_type -> hyapp.user.v1.CreatePrettyDisplayIDPoolRequest - 172, // 257: hyapp.user.v1.UserPrettyDisplayIDAdminService.UpdatePrettyDisplayIDPool:input_type -> hyapp.user.v1.UpdatePrettyDisplayIDPoolRequest - 174, // 258: hyapp.user.v1.UserPrettyDisplayIDAdminService.GeneratePrettyDisplayIDs:input_type -> hyapp.user.v1.GeneratePrettyDisplayIDsRequest - 176, // 259: hyapp.user.v1.UserPrettyDisplayIDAdminService.ListPrettyDisplayIDs:input_type -> hyapp.user.v1.ListPrettyDisplayIDsRequest - 178, // 260: hyapp.user.v1.UserPrettyDisplayIDAdminService.SetPrettyDisplayIDStatus:input_type -> hyapp.user.v1.SetPrettyDisplayIDStatusRequest - 179, // 261: hyapp.user.v1.UserPrettyDisplayIDAdminService.RecyclePrettyDisplayID:input_type -> hyapp.user.v1.RecyclePrettyDisplayIDRequest - 181, // 262: hyapp.user.v1.UserPrettyDisplayIDAdminService.AdminGrantPrettyDisplayID:input_type -> hyapp.user.v1.AdminGrantPrettyDisplayIDRequest - 18, // 263: hyapp.user.v1.UserService.GetUser:output_type -> hyapp.user.v1.GetUserResponse - 10, // 264: hyapp.user.v1.UserService.GetInviteAttribution:output_type -> hyapp.user.v1.GetInviteAttributionResponse - 21, // 265: hyapp.user.v1.UserService.BusinessUserLookup:output_type -> hyapp.user.v1.BusinessUserLookupResponse - 24, // 266: hyapp.user.v1.UserService.GetMyProfileStats:output_type -> hyapp.user.v1.GetMyProfileStatsResponse - 82, // 267: hyapp.user.v1.UserService.BatchGetUsers:output_type -> hyapp.user.v1.BatchGetUsersResponse - 86, // 268: hyapp.user.v1.UserService.BatchGetUserAdminProfiles:output_type -> hyapp.user.v1.BatchGetUserAdminProfilesResponse - 88, // 269: hyapp.user.v1.UserService.AdminIssueUserAccessToken:output_type -> hyapp.user.v1.AdminIssueUserAccessTokenResponse - 91, // 270: hyapp.user.v1.UserService.BatchGetRoomBasicUsers:output_type -> hyapp.user.v1.BatchGetRoomBasicUsersResponse - 93, // 271: hyapp.user.v1.UserService.ListUserIDs:output_type -> hyapp.user.v1.ListUserIDsResponse - 13, // 272: hyapp.user.v1.UserService.GetUserMicLifetimeStats:output_type -> hyapp.user.v1.GetUserMicLifetimeStatsResponse - 98, // 273: hyapp.user.v1.UserService.AuthorizeUserAvatarUpload:output_type -> hyapp.user.v1.AuthorizeUserAvatarUploadResponse - 100, // 274: hyapp.user.v1.UserService.CompleteUserAvatarUpload:output_type -> hyapp.user.v1.CompleteUserAvatarUploadResponse - 95, // 275: hyapp.user.v1.UserService.UpdateUserProfile:output_type -> hyapp.user.v1.UpdateUserProfileResponse - 102, // 276: hyapp.user.v1.UserService.UpdateUserProfileBackground:output_type -> hyapp.user.v1.UpdateUserProfileBackgroundResponse - 104, // 277: hyapp.user.v1.UserService.UpdateUserContactInfo:output_type -> hyapp.user.v1.UpdateUserContactInfoResponse - 106, // 278: hyapp.user.v1.UserService.UpdateUserWithdrawAddress:output_type -> hyapp.user.v1.UpdateUserWithdrawAddressResponse - 109, // 279: hyapp.user.v1.UserService.ChangeUserCountry:output_type -> hyapp.user.v1.ChangeUserCountryResponse - 109, // 280: hyapp.user.v1.UserService.AdminChangeUserCountry:output_type -> hyapp.user.v1.ChangeUserCountryResponse - 111, // 281: hyapp.user.v1.UserService.SetUserStatus:output_type -> hyapp.user.v1.SetUserStatusResponse - 114, // 282: hyapp.user.v1.UserService.AdminBanUser:output_type -> hyapp.user.v1.AdminBanUserResponse - 116, // 283: hyapp.user.v1.UserService.AdminUnbanUser:output_type -> hyapp.user.v1.AdminUnbanUserResponse - 119, // 284: hyapp.user.v1.UserService.CreateManagerUserBlock:output_type -> hyapp.user.v1.CreateManagerUserBlockResponse - 121, // 285: hyapp.user.v1.UserService.ListManagerUserBlocks:output_type -> hyapp.user.v1.ListManagerUserBlocksResponse - 123, // 286: hyapp.user.v1.UserService.UnblockManagerUser:output_type -> hyapp.user.v1.UnblockManagerUserResponse - 125, // 287: hyapp.user.v1.UserService.CompleteOnboarding:output_type -> hyapp.user.v1.CompleteOnboardingResponse - 127, // 288: hyapp.user.v1.UserService.SearchInviteReferrer:output_type -> hyapp.user.v1.SearchInviteReferrerResponse - 129, // 289: hyapp.user.v1.UserService.BindInviteReferrer:output_type -> hyapp.user.v1.BindInviteReferrerResponse - 26, // 290: hyapp.user.v1.UserSocialService.RecordProfileVisit:output_type -> hyapp.user.v1.RecordProfileVisitResponse - 29, // 291: hyapp.user.v1.UserSocialService.ListProfileVisitors:output_type -> hyapp.user.v1.ListProfileVisitorsResponse - 31, // 292: hyapp.user.v1.UserSocialService.FollowUser:output_type -> hyapp.user.v1.FollowUserResponse - 33, // 293: hyapp.user.v1.UserSocialService.UnfollowUser:output_type -> hyapp.user.v1.UnfollowUserResponse - 36, // 294: hyapp.user.v1.UserSocialService.ListFollowing:output_type -> hyapp.user.v1.ListFollowingResponse - 38, // 295: hyapp.user.v1.UserSocialService.ApplyFriend:output_type -> hyapp.user.v1.ApplyFriendResponse - 40, // 296: hyapp.user.v1.UserSocialService.AcceptFriendApplication:output_type -> hyapp.user.v1.AcceptFriendApplicationResponse - 42, // 297: hyapp.user.v1.UserSocialService.DeleteFriend:output_type -> hyapp.user.v1.DeleteFriendResponse - 45, // 298: hyapp.user.v1.UserSocialService.ListFriends:output_type -> hyapp.user.v1.ListFriendsResponse - 48, // 299: hyapp.user.v1.UserSocialService.ListFriendApplications:output_type -> hyapp.user.v1.ListFriendApplicationsResponse - 80, // 300: hyapp.user.v1.UserSocialService.SubmitReport:output_type -> hyapp.user.v1.SubmitReportResponse - 60, // 301: hyapp.user.v1.UserCPService.ListCPApplications:output_type -> hyapp.user.v1.ListCPApplicationsResponse - 62, // 302: hyapp.user.v1.UserCPService.AcceptCPApplication:output_type -> hyapp.user.v1.AcceptCPApplicationResponse - 64, // 303: hyapp.user.v1.UserCPService.RejectCPApplication:output_type -> hyapp.user.v1.RejectCPApplicationResponse - 66, // 304: hyapp.user.v1.UserCPService.ListCPRelationships:output_type -> hyapp.user.v1.ListCPRelationshipsResponse - 68, // 305: hyapp.user.v1.UserCPService.ListCPIntimacyLeaderboard:output_type -> hyapp.user.v1.ListCPIntimacyLeaderboardResponse - 70, // 306: hyapp.user.v1.UserCPService.PrepareBreakCPRelationship:output_type -> hyapp.user.v1.PrepareBreakCPRelationshipResponse - 72, // 307: hyapp.user.v1.UserCPService.ConfirmBreakCPRelationship:output_type -> hyapp.user.v1.ConfirmBreakCPRelationshipResponse - 74, // 308: hyapp.user.v1.UserCPService.CancelBreakCPRelationship:output_type -> hyapp.user.v1.CancelBreakCPRelationshipResponse - 185, // 309: hyapp.user.v1.UserCPService.ListCPFormationGiftFeed:output_type -> hyapp.user.v1.ListCPFormationGiftFeedResponse - 77, // 310: hyapp.user.v1.UserCPInternalService.ConsumeRoomGiftCPEvent:output_type -> hyapp.user.v1.ConsumeRoomGiftCPEventResponse - 58, // 311: hyapp.user.v1.UserCPInternalService.ListCPWeeklyRankEntries:output_type -> hyapp.user.v1.ListCPWeeklyRankEntriesResponse - 15, // 312: hyapp.user.v1.UserCronService.ProcessLoginIPRiskBatch:output_type -> hyapp.user.v1.CronBatchResponse - 15, // 313: hyapp.user.v1.UserCronService.ProcessRegionRebuildBatch:output_type -> hyapp.user.v1.CronBatchResponse - 15, // 314: hyapp.user.v1.UserCronService.CompensateMicOpenSessions:output_type -> hyapp.user.v1.CronBatchResponse - 15, // 315: hyapp.user.v1.UserCronService.CompensateRoomOpenSessions:output_type -> hyapp.user.v1.CronBatchResponse - 15, // 316: hyapp.user.v1.UserCronService.ExpireManagerUserBlocks:output_type -> hyapp.user.v1.CronBatchResponse - 15, // 317: hyapp.user.v1.UserCronService.ExpireAdminUserBans:output_type -> hyapp.user.v1.CronBatchResponse - 15, // 318: hyapp.user.v1.UserCronService.RefreshCPIntimacyLeaderboard:output_type -> hyapp.user.v1.CronBatchResponse - 131, // 319: hyapp.user.v1.UserDeviceService.BindPushToken:output_type -> hyapp.user.v1.BindPushTokenResponse - 133, // 320: hyapp.user.v1.UserDeviceService.DeletePushToken:output_type -> hyapp.user.v1.DeletePushTokenResponse - 4, // 321: hyapp.user.v1.AppRegistryService.ResolveApp:output_type -> hyapp.user.v1.ResolveAppResponse - 187, // 322: hyapp.user.v1.AppRegistryService.ListApps:output_type -> hyapp.user.v1.ListAppsResponse - 137, // 323: hyapp.user.v1.CountryAdminService.ListCountries:output_type -> hyapp.user.v1.ListCountriesResponse - 139, // 324: hyapp.user.v1.CountryAdminService.UpdateCountry:output_type -> hyapp.user.v1.CountryResponse - 141, // 325: hyapp.user.v1.CountryQueryService.ListRegistrationCountries:output_type -> hyapp.user.v1.ListRegistrationCountriesResponse - 144, // 326: hyapp.user.v1.CountryQueryService.ListLoginRiskBlockedCountries:output_type -> hyapp.user.v1.ListLoginRiskBlockedCountriesResponse - 146, // 327: hyapp.user.v1.RegionAdminService.ListRegions:output_type -> hyapp.user.v1.ListRegionsResponse - 150, // 328: hyapp.user.v1.RegionAdminService.GetRegion:output_type -> hyapp.user.v1.RegionResponse - 150, // 329: hyapp.user.v1.RegionAdminService.UpdateRegion:output_type -> hyapp.user.v1.RegionResponse - 150, // 330: hyapp.user.v1.RegionAdminService.ReplaceRegionCountries:output_type -> hyapp.user.v1.RegionResponse - 153, // 331: hyapp.user.v1.UserIdentityService.GetUserIdentity:output_type -> hyapp.user.v1.GetUserIdentityResponse - 155, // 332: hyapp.user.v1.UserIdentityService.ResolveDisplayUserID:output_type -> hyapp.user.v1.ResolveDisplayUserIDResponse - 189, // 333: hyapp.user.v1.UserIdentityService.ResolveAdminUserIdentifier:output_type -> hyapp.user.v1.ResolveAdminUserIdentifierResponse - 157, // 334: hyapp.user.v1.UserIdentityService.ChangeDisplayUserID:output_type -> hyapp.user.v1.ChangeDisplayUserIDResponse - 159, // 335: hyapp.user.v1.UserIdentityService.ApplyPrettyDisplayUserID:output_type -> hyapp.user.v1.ApplyPrettyDisplayUserIDResponse - 166, // 336: hyapp.user.v1.UserIdentityService.ListAvailablePrettyDisplayIDs:output_type -> hyapp.user.v1.ListAvailablePrettyDisplayIDsResponse - 168, // 337: hyapp.user.v1.UserIdentityService.ApplyPrettyDisplayIDFromPool:output_type -> hyapp.user.v1.ApplyPrettyDisplayIDFromPoolResponse - 161, // 338: hyapp.user.v1.UserIdentityService.ExpirePrettyDisplayUserID:output_type -> hyapp.user.v1.ExpirePrettyDisplayUserIDResponse - 170, // 339: hyapp.user.v1.UserPrettyDisplayIDAdminService.ListPrettyDisplayIDPools:output_type -> hyapp.user.v1.ListPrettyDisplayIDPoolsResponse - 173, // 340: hyapp.user.v1.UserPrettyDisplayIDAdminService.CreatePrettyDisplayIDPool:output_type -> hyapp.user.v1.PrettyDisplayIDPoolResponse - 173, // 341: hyapp.user.v1.UserPrettyDisplayIDAdminService.UpdatePrettyDisplayIDPool:output_type -> hyapp.user.v1.PrettyDisplayIDPoolResponse - 175, // 342: hyapp.user.v1.UserPrettyDisplayIDAdminService.GeneratePrettyDisplayIDs:output_type -> hyapp.user.v1.GeneratePrettyDisplayIDsResponse - 177, // 343: hyapp.user.v1.UserPrettyDisplayIDAdminService.ListPrettyDisplayIDs:output_type -> hyapp.user.v1.ListPrettyDisplayIDsResponse - 180, // 344: hyapp.user.v1.UserPrettyDisplayIDAdminService.SetPrettyDisplayIDStatus:output_type -> hyapp.user.v1.PrettyDisplayIDResponse - 180, // 345: hyapp.user.v1.UserPrettyDisplayIDAdminService.RecyclePrettyDisplayID:output_type -> hyapp.user.v1.PrettyDisplayIDResponse - 182, // 346: hyapp.user.v1.UserPrettyDisplayIDAdminService.AdminGrantPrettyDisplayID:output_type -> hyapp.user.v1.AdminGrantPrettyDisplayIDResponse - 263, // [263:347] is the sub-list for method output_type - 179, // [179:263] is the sub-list for method input_type + 14, // 235: hyapp.user.v1.UserCronService.ProcessPrivateMessageHistoryBatch:input_type -> hyapp.user.v1.CronBatchRequest + 130, // 236: hyapp.user.v1.UserDeviceService.BindPushToken:input_type -> hyapp.user.v1.BindPushTokenRequest + 132, // 237: hyapp.user.v1.UserDeviceService.DeletePushToken:input_type -> hyapp.user.v1.DeletePushTokenRequest + 3, // 238: hyapp.user.v1.AppRegistryService.ResolveApp:input_type -> hyapp.user.v1.ResolveAppRequest + 186, // 239: hyapp.user.v1.AppRegistryService.ListApps:input_type -> hyapp.user.v1.ListAppsRequest + 136, // 240: hyapp.user.v1.CountryAdminService.ListCountries:input_type -> hyapp.user.v1.ListCountriesRequest + 138, // 241: hyapp.user.v1.CountryAdminService.UpdateCountry:input_type -> hyapp.user.v1.UpdateCountryRequest + 140, // 242: hyapp.user.v1.CountryQueryService.ListRegistrationCountries:input_type -> hyapp.user.v1.ListRegistrationCountriesRequest + 143, // 243: hyapp.user.v1.CountryQueryService.ListLoginRiskBlockedCountries:input_type -> hyapp.user.v1.ListLoginRiskBlockedCountriesRequest + 145, // 244: hyapp.user.v1.RegionAdminService.ListRegions:input_type -> hyapp.user.v1.ListRegionsRequest + 147, // 245: hyapp.user.v1.RegionAdminService.GetRegion:input_type -> hyapp.user.v1.GetRegionRequest + 148, // 246: hyapp.user.v1.RegionAdminService.UpdateRegion:input_type -> hyapp.user.v1.UpdateRegionRequest + 149, // 247: hyapp.user.v1.RegionAdminService.ReplaceRegionCountries:input_type -> hyapp.user.v1.ReplaceRegionCountriesRequest + 152, // 248: hyapp.user.v1.UserIdentityService.GetUserIdentity:input_type -> hyapp.user.v1.GetUserIdentityRequest + 154, // 249: hyapp.user.v1.UserIdentityService.ResolveDisplayUserID:input_type -> hyapp.user.v1.ResolveDisplayUserIDRequest + 188, // 250: hyapp.user.v1.UserIdentityService.ResolveAdminUserIdentifier:input_type -> hyapp.user.v1.ResolveAdminUserIdentifierRequest + 156, // 251: hyapp.user.v1.UserIdentityService.ChangeDisplayUserID:input_type -> hyapp.user.v1.ChangeDisplayUserIDRequest + 158, // 252: hyapp.user.v1.UserIdentityService.ApplyPrettyDisplayUserID:input_type -> hyapp.user.v1.ApplyPrettyDisplayUserIDRequest + 165, // 253: hyapp.user.v1.UserIdentityService.ListAvailablePrettyDisplayIDs:input_type -> hyapp.user.v1.ListAvailablePrettyDisplayIDsRequest + 167, // 254: hyapp.user.v1.UserIdentityService.ApplyPrettyDisplayIDFromPool:input_type -> hyapp.user.v1.ApplyPrettyDisplayIDFromPoolRequest + 160, // 255: hyapp.user.v1.UserIdentityService.ExpirePrettyDisplayUserID:input_type -> hyapp.user.v1.ExpirePrettyDisplayUserIDRequest + 169, // 256: hyapp.user.v1.UserPrettyDisplayIDAdminService.ListPrettyDisplayIDPools:input_type -> hyapp.user.v1.ListPrettyDisplayIDPoolsRequest + 171, // 257: hyapp.user.v1.UserPrettyDisplayIDAdminService.CreatePrettyDisplayIDPool:input_type -> hyapp.user.v1.CreatePrettyDisplayIDPoolRequest + 172, // 258: hyapp.user.v1.UserPrettyDisplayIDAdminService.UpdatePrettyDisplayIDPool:input_type -> hyapp.user.v1.UpdatePrettyDisplayIDPoolRequest + 174, // 259: hyapp.user.v1.UserPrettyDisplayIDAdminService.GeneratePrettyDisplayIDs:input_type -> hyapp.user.v1.GeneratePrettyDisplayIDsRequest + 176, // 260: hyapp.user.v1.UserPrettyDisplayIDAdminService.ListPrettyDisplayIDs:input_type -> hyapp.user.v1.ListPrettyDisplayIDsRequest + 178, // 261: hyapp.user.v1.UserPrettyDisplayIDAdminService.SetPrettyDisplayIDStatus:input_type -> hyapp.user.v1.SetPrettyDisplayIDStatusRequest + 179, // 262: hyapp.user.v1.UserPrettyDisplayIDAdminService.RecyclePrettyDisplayID:input_type -> hyapp.user.v1.RecyclePrettyDisplayIDRequest + 181, // 263: hyapp.user.v1.UserPrettyDisplayIDAdminService.AdminGrantPrettyDisplayID:input_type -> hyapp.user.v1.AdminGrantPrettyDisplayIDRequest + 18, // 264: hyapp.user.v1.UserService.GetUser:output_type -> hyapp.user.v1.GetUserResponse + 10, // 265: hyapp.user.v1.UserService.GetInviteAttribution:output_type -> hyapp.user.v1.GetInviteAttributionResponse + 21, // 266: hyapp.user.v1.UserService.BusinessUserLookup:output_type -> hyapp.user.v1.BusinessUserLookupResponse + 24, // 267: hyapp.user.v1.UserService.GetMyProfileStats:output_type -> hyapp.user.v1.GetMyProfileStatsResponse + 82, // 268: hyapp.user.v1.UserService.BatchGetUsers:output_type -> hyapp.user.v1.BatchGetUsersResponse + 86, // 269: hyapp.user.v1.UserService.BatchGetUserAdminProfiles:output_type -> hyapp.user.v1.BatchGetUserAdminProfilesResponse + 88, // 270: hyapp.user.v1.UserService.AdminIssueUserAccessToken:output_type -> hyapp.user.v1.AdminIssueUserAccessTokenResponse + 91, // 271: hyapp.user.v1.UserService.BatchGetRoomBasicUsers:output_type -> hyapp.user.v1.BatchGetRoomBasicUsersResponse + 93, // 272: hyapp.user.v1.UserService.ListUserIDs:output_type -> hyapp.user.v1.ListUserIDsResponse + 13, // 273: hyapp.user.v1.UserService.GetUserMicLifetimeStats:output_type -> hyapp.user.v1.GetUserMicLifetimeStatsResponse + 98, // 274: hyapp.user.v1.UserService.AuthorizeUserAvatarUpload:output_type -> hyapp.user.v1.AuthorizeUserAvatarUploadResponse + 100, // 275: hyapp.user.v1.UserService.CompleteUserAvatarUpload:output_type -> hyapp.user.v1.CompleteUserAvatarUploadResponse + 95, // 276: hyapp.user.v1.UserService.UpdateUserProfile:output_type -> hyapp.user.v1.UpdateUserProfileResponse + 102, // 277: hyapp.user.v1.UserService.UpdateUserProfileBackground:output_type -> hyapp.user.v1.UpdateUserProfileBackgroundResponse + 104, // 278: hyapp.user.v1.UserService.UpdateUserContactInfo:output_type -> hyapp.user.v1.UpdateUserContactInfoResponse + 106, // 279: hyapp.user.v1.UserService.UpdateUserWithdrawAddress:output_type -> hyapp.user.v1.UpdateUserWithdrawAddressResponse + 109, // 280: hyapp.user.v1.UserService.ChangeUserCountry:output_type -> hyapp.user.v1.ChangeUserCountryResponse + 109, // 281: hyapp.user.v1.UserService.AdminChangeUserCountry:output_type -> hyapp.user.v1.ChangeUserCountryResponse + 111, // 282: hyapp.user.v1.UserService.SetUserStatus:output_type -> hyapp.user.v1.SetUserStatusResponse + 114, // 283: hyapp.user.v1.UserService.AdminBanUser:output_type -> hyapp.user.v1.AdminBanUserResponse + 116, // 284: hyapp.user.v1.UserService.AdminUnbanUser:output_type -> hyapp.user.v1.AdminUnbanUserResponse + 119, // 285: hyapp.user.v1.UserService.CreateManagerUserBlock:output_type -> hyapp.user.v1.CreateManagerUserBlockResponse + 121, // 286: hyapp.user.v1.UserService.ListManagerUserBlocks:output_type -> hyapp.user.v1.ListManagerUserBlocksResponse + 123, // 287: hyapp.user.v1.UserService.UnblockManagerUser:output_type -> hyapp.user.v1.UnblockManagerUserResponse + 125, // 288: hyapp.user.v1.UserService.CompleteOnboarding:output_type -> hyapp.user.v1.CompleteOnboardingResponse + 127, // 289: hyapp.user.v1.UserService.SearchInviteReferrer:output_type -> hyapp.user.v1.SearchInviteReferrerResponse + 129, // 290: hyapp.user.v1.UserService.BindInviteReferrer:output_type -> hyapp.user.v1.BindInviteReferrerResponse + 26, // 291: hyapp.user.v1.UserSocialService.RecordProfileVisit:output_type -> hyapp.user.v1.RecordProfileVisitResponse + 29, // 292: hyapp.user.v1.UserSocialService.ListProfileVisitors:output_type -> hyapp.user.v1.ListProfileVisitorsResponse + 31, // 293: hyapp.user.v1.UserSocialService.FollowUser:output_type -> hyapp.user.v1.FollowUserResponse + 33, // 294: hyapp.user.v1.UserSocialService.UnfollowUser:output_type -> hyapp.user.v1.UnfollowUserResponse + 36, // 295: hyapp.user.v1.UserSocialService.ListFollowing:output_type -> hyapp.user.v1.ListFollowingResponse + 38, // 296: hyapp.user.v1.UserSocialService.ApplyFriend:output_type -> hyapp.user.v1.ApplyFriendResponse + 40, // 297: hyapp.user.v1.UserSocialService.AcceptFriendApplication:output_type -> hyapp.user.v1.AcceptFriendApplicationResponse + 42, // 298: hyapp.user.v1.UserSocialService.DeleteFriend:output_type -> hyapp.user.v1.DeleteFriendResponse + 45, // 299: hyapp.user.v1.UserSocialService.ListFriends:output_type -> hyapp.user.v1.ListFriendsResponse + 48, // 300: hyapp.user.v1.UserSocialService.ListFriendApplications:output_type -> hyapp.user.v1.ListFriendApplicationsResponse + 80, // 301: hyapp.user.v1.UserSocialService.SubmitReport:output_type -> hyapp.user.v1.SubmitReportResponse + 60, // 302: hyapp.user.v1.UserCPService.ListCPApplications:output_type -> hyapp.user.v1.ListCPApplicationsResponse + 62, // 303: hyapp.user.v1.UserCPService.AcceptCPApplication:output_type -> hyapp.user.v1.AcceptCPApplicationResponse + 64, // 304: hyapp.user.v1.UserCPService.RejectCPApplication:output_type -> hyapp.user.v1.RejectCPApplicationResponse + 66, // 305: hyapp.user.v1.UserCPService.ListCPRelationships:output_type -> hyapp.user.v1.ListCPRelationshipsResponse + 68, // 306: hyapp.user.v1.UserCPService.ListCPIntimacyLeaderboard:output_type -> hyapp.user.v1.ListCPIntimacyLeaderboardResponse + 70, // 307: hyapp.user.v1.UserCPService.PrepareBreakCPRelationship:output_type -> hyapp.user.v1.PrepareBreakCPRelationshipResponse + 72, // 308: hyapp.user.v1.UserCPService.ConfirmBreakCPRelationship:output_type -> hyapp.user.v1.ConfirmBreakCPRelationshipResponse + 74, // 309: hyapp.user.v1.UserCPService.CancelBreakCPRelationship:output_type -> hyapp.user.v1.CancelBreakCPRelationshipResponse + 185, // 310: hyapp.user.v1.UserCPService.ListCPFormationGiftFeed:output_type -> hyapp.user.v1.ListCPFormationGiftFeedResponse + 77, // 311: hyapp.user.v1.UserCPInternalService.ConsumeRoomGiftCPEvent:output_type -> hyapp.user.v1.ConsumeRoomGiftCPEventResponse + 58, // 312: hyapp.user.v1.UserCPInternalService.ListCPWeeklyRankEntries:output_type -> hyapp.user.v1.ListCPWeeklyRankEntriesResponse + 15, // 313: hyapp.user.v1.UserCronService.ProcessLoginIPRiskBatch:output_type -> hyapp.user.v1.CronBatchResponse + 15, // 314: hyapp.user.v1.UserCronService.ProcessRegionRebuildBatch:output_type -> hyapp.user.v1.CronBatchResponse + 15, // 315: hyapp.user.v1.UserCronService.CompensateMicOpenSessions:output_type -> hyapp.user.v1.CronBatchResponse + 15, // 316: hyapp.user.v1.UserCronService.CompensateRoomOpenSessions:output_type -> hyapp.user.v1.CronBatchResponse + 15, // 317: hyapp.user.v1.UserCronService.ExpireManagerUserBlocks:output_type -> hyapp.user.v1.CronBatchResponse + 15, // 318: hyapp.user.v1.UserCronService.ExpireAdminUserBans:output_type -> hyapp.user.v1.CronBatchResponse + 15, // 319: hyapp.user.v1.UserCronService.RefreshCPIntimacyLeaderboard:output_type -> hyapp.user.v1.CronBatchResponse + 15, // 320: hyapp.user.v1.UserCronService.ProcessPrivateMessageHistoryBatch:output_type -> hyapp.user.v1.CronBatchResponse + 131, // 321: hyapp.user.v1.UserDeviceService.BindPushToken:output_type -> hyapp.user.v1.BindPushTokenResponse + 133, // 322: hyapp.user.v1.UserDeviceService.DeletePushToken:output_type -> hyapp.user.v1.DeletePushTokenResponse + 4, // 323: hyapp.user.v1.AppRegistryService.ResolveApp:output_type -> hyapp.user.v1.ResolveAppResponse + 187, // 324: hyapp.user.v1.AppRegistryService.ListApps:output_type -> hyapp.user.v1.ListAppsResponse + 137, // 325: hyapp.user.v1.CountryAdminService.ListCountries:output_type -> hyapp.user.v1.ListCountriesResponse + 139, // 326: hyapp.user.v1.CountryAdminService.UpdateCountry:output_type -> hyapp.user.v1.CountryResponse + 141, // 327: hyapp.user.v1.CountryQueryService.ListRegistrationCountries:output_type -> hyapp.user.v1.ListRegistrationCountriesResponse + 144, // 328: hyapp.user.v1.CountryQueryService.ListLoginRiskBlockedCountries:output_type -> hyapp.user.v1.ListLoginRiskBlockedCountriesResponse + 146, // 329: hyapp.user.v1.RegionAdminService.ListRegions:output_type -> hyapp.user.v1.ListRegionsResponse + 150, // 330: hyapp.user.v1.RegionAdminService.GetRegion:output_type -> hyapp.user.v1.RegionResponse + 150, // 331: hyapp.user.v1.RegionAdminService.UpdateRegion:output_type -> hyapp.user.v1.RegionResponse + 150, // 332: hyapp.user.v1.RegionAdminService.ReplaceRegionCountries:output_type -> hyapp.user.v1.RegionResponse + 153, // 333: hyapp.user.v1.UserIdentityService.GetUserIdentity:output_type -> hyapp.user.v1.GetUserIdentityResponse + 155, // 334: hyapp.user.v1.UserIdentityService.ResolveDisplayUserID:output_type -> hyapp.user.v1.ResolveDisplayUserIDResponse + 189, // 335: hyapp.user.v1.UserIdentityService.ResolveAdminUserIdentifier:output_type -> hyapp.user.v1.ResolveAdminUserIdentifierResponse + 157, // 336: hyapp.user.v1.UserIdentityService.ChangeDisplayUserID:output_type -> hyapp.user.v1.ChangeDisplayUserIDResponse + 159, // 337: hyapp.user.v1.UserIdentityService.ApplyPrettyDisplayUserID:output_type -> hyapp.user.v1.ApplyPrettyDisplayUserIDResponse + 166, // 338: hyapp.user.v1.UserIdentityService.ListAvailablePrettyDisplayIDs:output_type -> hyapp.user.v1.ListAvailablePrettyDisplayIDsResponse + 168, // 339: hyapp.user.v1.UserIdentityService.ApplyPrettyDisplayIDFromPool:output_type -> hyapp.user.v1.ApplyPrettyDisplayIDFromPoolResponse + 161, // 340: hyapp.user.v1.UserIdentityService.ExpirePrettyDisplayUserID:output_type -> hyapp.user.v1.ExpirePrettyDisplayUserIDResponse + 170, // 341: hyapp.user.v1.UserPrettyDisplayIDAdminService.ListPrettyDisplayIDPools:output_type -> hyapp.user.v1.ListPrettyDisplayIDPoolsResponse + 173, // 342: hyapp.user.v1.UserPrettyDisplayIDAdminService.CreatePrettyDisplayIDPool:output_type -> hyapp.user.v1.PrettyDisplayIDPoolResponse + 173, // 343: hyapp.user.v1.UserPrettyDisplayIDAdminService.UpdatePrettyDisplayIDPool:output_type -> hyapp.user.v1.PrettyDisplayIDPoolResponse + 175, // 344: hyapp.user.v1.UserPrettyDisplayIDAdminService.GeneratePrettyDisplayIDs:output_type -> hyapp.user.v1.GeneratePrettyDisplayIDsResponse + 177, // 345: hyapp.user.v1.UserPrettyDisplayIDAdminService.ListPrettyDisplayIDs:output_type -> hyapp.user.v1.ListPrettyDisplayIDsResponse + 180, // 346: hyapp.user.v1.UserPrettyDisplayIDAdminService.SetPrettyDisplayIDStatus:output_type -> hyapp.user.v1.PrettyDisplayIDResponse + 180, // 347: hyapp.user.v1.UserPrettyDisplayIDAdminService.RecyclePrettyDisplayID:output_type -> hyapp.user.v1.PrettyDisplayIDResponse + 182, // 348: hyapp.user.v1.UserPrettyDisplayIDAdminService.AdminGrantPrettyDisplayID:output_type -> hyapp.user.v1.AdminGrantPrettyDisplayIDResponse + 264, // [264:349] is the sub-list for method output_type + 179, // [179:264] is the sub-list for method input_type 179, // [179:179] is the sub-list for extension type_name 179, // [179:179] is the sub-list for extension extendee 0, // [0:179] is the sub-list for field type_name diff --git a/api/proto/user/v1/user.proto b/api/proto/user/v1/user.proto index ae234dee..fcfe4d53 100644 --- a/api/proto/user/v1/user.proto +++ b/api/proto/user/v1/user.proto @@ -1613,6 +1613,7 @@ service UserCronService { rpc ExpireManagerUserBlocks(CronBatchRequest) returns (CronBatchResponse); rpc ExpireAdminUserBans(CronBatchRequest) returns (CronBatchResponse); rpc RefreshCPIntimacyLeaderboard(CronBatchRequest) returns (CronBatchResponse); + rpc ProcessPrivateMessageHistoryBatch(CronBatchRequest) returns (CronBatchResponse); } // UserDeviceService 承载 App 设备推送 token 的绑定和失效。 diff --git a/api/proto/user/v1/user_grpc.pb.go b/api/proto/user/v1/user_grpc.pb.go index ae2f3e0d..6e686605 100644 --- a/api/proto/user/v1/user_grpc.pb.go +++ b/api/proto/user/v1/user_grpc.pb.go @@ -2153,13 +2153,14 @@ var UserCPInternalService_ServiceDesc = grpc.ServiceDesc{ } const ( - UserCronService_ProcessLoginIPRiskBatch_FullMethodName = "/hyapp.user.v1.UserCronService/ProcessLoginIPRiskBatch" - UserCronService_ProcessRegionRebuildBatch_FullMethodName = "/hyapp.user.v1.UserCronService/ProcessRegionRebuildBatch" - UserCronService_CompensateMicOpenSessions_FullMethodName = "/hyapp.user.v1.UserCronService/CompensateMicOpenSessions" - UserCronService_CompensateRoomOpenSessions_FullMethodName = "/hyapp.user.v1.UserCronService/CompensateRoomOpenSessions" - UserCronService_ExpireManagerUserBlocks_FullMethodName = "/hyapp.user.v1.UserCronService/ExpireManagerUserBlocks" - UserCronService_ExpireAdminUserBans_FullMethodName = "/hyapp.user.v1.UserCronService/ExpireAdminUserBans" - UserCronService_RefreshCPIntimacyLeaderboard_FullMethodName = "/hyapp.user.v1.UserCronService/RefreshCPIntimacyLeaderboard" + UserCronService_ProcessLoginIPRiskBatch_FullMethodName = "/hyapp.user.v1.UserCronService/ProcessLoginIPRiskBatch" + UserCronService_ProcessRegionRebuildBatch_FullMethodName = "/hyapp.user.v1.UserCronService/ProcessRegionRebuildBatch" + UserCronService_CompensateMicOpenSessions_FullMethodName = "/hyapp.user.v1.UserCronService/CompensateMicOpenSessions" + UserCronService_CompensateRoomOpenSessions_FullMethodName = "/hyapp.user.v1.UserCronService/CompensateRoomOpenSessions" + UserCronService_ExpireManagerUserBlocks_FullMethodName = "/hyapp.user.v1.UserCronService/ExpireManagerUserBlocks" + UserCronService_ExpireAdminUserBans_FullMethodName = "/hyapp.user.v1.UserCronService/ExpireAdminUserBans" + UserCronService_RefreshCPIntimacyLeaderboard_FullMethodName = "/hyapp.user.v1.UserCronService/RefreshCPIntimacyLeaderboard" + UserCronService_ProcessPrivateMessageHistoryBatch_FullMethodName = "/hyapp.user.v1.UserCronService/ProcessPrivateMessageHistoryBatch" ) // UserCronServiceClient is the client API for UserCronService service. @@ -2175,6 +2176,7 @@ type UserCronServiceClient interface { ExpireManagerUserBlocks(ctx context.Context, in *CronBatchRequest, opts ...grpc.CallOption) (*CronBatchResponse, error) ExpireAdminUserBans(ctx context.Context, in *CronBatchRequest, opts ...grpc.CallOption) (*CronBatchResponse, error) RefreshCPIntimacyLeaderboard(ctx context.Context, in *CronBatchRequest, opts ...grpc.CallOption) (*CronBatchResponse, error) + ProcessPrivateMessageHistoryBatch(ctx context.Context, in *CronBatchRequest, opts ...grpc.CallOption) (*CronBatchResponse, error) } type userCronServiceClient struct { @@ -2255,6 +2257,16 @@ func (c *userCronServiceClient) RefreshCPIntimacyLeaderboard(ctx context.Context return out, nil } +func (c *userCronServiceClient) ProcessPrivateMessageHistoryBatch(ctx context.Context, in *CronBatchRequest, opts ...grpc.CallOption) (*CronBatchResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(CronBatchResponse) + err := c.cc.Invoke(ctx, UserCronService_ProcessPrivateMessageHistoryBatch_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // UserCronServiceServer is the server API for UserCronService service. // All implementations must embed UnimplementedUserCronServiceServer // for forward compatibility. @@ -2268,6 +2280,7 @@ type UserCronServiceServer interface { ExpireManagerUserBlocks(context.Context, *CronBatchRequest) (*CronBatchResponse, error) ExpireAdminUserBans(context.Context, *CronBatchRequest) (*CronBatchResponse, error) RefreshCPIntimacyLeaderboard(context.Context, *CronBatchRequest) (*CronBatchResponse, error) + ProcessPrivateMessageHistoryBatch(context.Context, *CronBatchRequest) (*CronBatchResponse, error) mustEmbedUnimplementedUserCronServiceServer() } @@ -2299,6 +2312,9 @@ func (UnimplementedUserCronServiceServer) ExpireAdminUserBans(context.Context, * func (UnimplementedUserCronServiceServer) RefreshCPIntimacyLeaderboard(context.Context, *CronBatchRequest) (*CronBatchResponse, error) { return nil, status.Error(codes.Unimplemented, "method RefreshCPIntimacyLeaderboard not implemented") } +func (UnimplementedUserCronServiceServer) ProcessPrivateMessageHistoryBatch(context.Context, *CronBatchRequest) (*CronBatchResponse, error) { + return nil, status.Error(codes.Unimplemented, "method ProcessPrivateMessageHistoryBatch not implemented") +} func (UnimplementedUserCronServiceServer) mustEmbedUnimplementedUserCronServiceServer() {} func (UnimplementedUserCronServiceServer) testEmbeddedByValue() {} @@ -2446,6 +2462,24 @@ func _UserCronService_RefreshCPIntimacyLeaderboard_Handler(srv interface{}, ctx return interceptor(ctx, in, info, handler) } +func _UserCronService_ProcessPrivateMessageHistoryBatch_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(CronBatchRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(UserCronServiceServer).ProcessPrivateMessageHistoryBatch(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: UserCronService_ProcessPrivateMessageHistoryBatch_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(UserCronServiceServer).ProcessPrivateMessageHistoryBatch(ctx, req.(*CronBatchRequest)) + } + return interceptor(ctx, in, info, handler) +} + // UserCronService_ServiceDesc is the grpc.ServiceDesc for UserCronService service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -2481,6 +2515,10 @@ var UserCronService_ServiceDesc = grpc.ServiceDesc{ MethodName: "RefreshCPIntimacyLeaderboard", Handler: _UserCronService_RefreshCPIntimacyLeaderboard_Handler, }, + { + MethodName: "ProcessPrivateMessageHistoryBatch", + Handler: _UserCronService_ProcessPrivateMessageHistoryBatch_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "proto/user/v1/user.proto", diff --git a/pkg/tencentim/history.go b/pkg/tencentim/history.go new file mode 100644 index 00000000..7724c850 --- /dev/null +++ b/pkg/tencentim/history.go @@ -0,0 +1,176 @@ +package tencentim + +import ( + "compress/gzip" + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/url" + "strings" + "time" +) + +const ( + // historyFileCommand 获取一个自然小时的 C2C 消息归档下载地址；它是离线统计接口， + // 不进入客户端发消息链路，也不会把业务服务延迟传播到腾讯 IM。 + historyFileCommand = "v4/open_msg_svc/get_history" + // 小时归档通常是 MB 级文件，不能复用实时 REST 调用的 5 秒传输预算； + // cron RPC 仍有 2 分钟总 deadline，这里只放宽单文件下载而不影响发消息接口。 + historyDownloadTimeout = 60 * time.Second +) + +var tencentHistoryLocation = time.FixedZone("UTC+8", 8*60*60) + +// C2CHistoryFile 是腾讯 IM 返回的短期下载凭据。URL 含临时授权信息，只能在内存中使用， +// 调用方不得把它写入日志、数据库或业务错误。 +type C2CHistoryFile struct { + URL string `json:"URL"` + FileSize int64 `json:"FileSize"` + GzipSize int64 `json:"GzipSize"` +} + +// C2CHistoryMessage 只解析私聊人数统计所需字段。MsgBody 等正文由流式 decoder 跳过， +// 避免消息内容进入 user-service 内存对象、日志或统计库。 +type C2CHistoryMessage struct { + FromAccount string `json:"From_Account"` + ToAccount string `json:"To_Account"` + Timestamp int64 `json:"MsgTimestamp"` + Sequence uint64 `json:"MsgSeq"` + Random uint64 `json:"MsgRandom"` +} + +type historyFilesRequest struct { + ChatType string `json:"ChatType"` + MsgTime string `json:"MsgTime"` +} + +type historyFilesResponse struct { + restResponse + Files []C2CHistoryFile `json:"File"` +} + +// ListC2CHistoryFiles 获取指定 UTC 小时对应的 C2C 归档。 +// 腾讯接口的 MsgTime 使用 UTC+8 小时字符串，而消息本身仍携带 Unix 时间戳；在 REST +// 边界集中转换可防止 cron 把 UTC 小时错取成前一天的数据。 +func (c *RESTClient) ListC2CHistoryFiles(ctx context.Context, hourStartUTC time.Time) ([]C2CHistoryFile, error) { + hourStartUTC = hourStartUTC.UTC().Truncate(time.Hour) + request := historyFilesRequest{ + ChatType: "C2C", + MsgTime: hourStartUTC.In(tencentHistoryLocation).Format("2006010215"), + } + var response historyFilesResponse + if err := c.post(ctx, historyFileCommand, request, &response); err != nil { + return nil, err + } + if err := response.err(); err != nil { + return nil, err + } + + files := make([]C2CHistoryFile, 0, len(response.Files)) + for _, file := range response.Files { + file.URL = strings.TrimSpace(file.URL) + if file.URL == "" { + continue + } + files = append(files, file) + } + return files, nil +} + +// StreamC2CHistory 下载并逐条解码一个 gzip 归档。decoder 只保留当前消息的统计字段， +// 即使单小时归档很大也不会把整个 MsgList 或消息正文一次性加载进内存。 +func (c *RESTClient) StreamC2CHistory(ctx context.Context, file C2CHistoryFile, consume func(C2CHistoryMessage) error) (int, error) { + if c == nil || c.httpClient == nil { + return 0, fmt.Errorf("tencent im rest client is not configured") + } + if consume == nil { + return 0, fmt.Errorf("tencent im history consumer is required") + } + downloadURL, err := url.Parse(strings.TrimSpace(file.URL)) + if err != nil || downloadURL.Scheme != "https" || downloadURL.Host == "" { + return 0, fmt.Errorf("tencent im history download url is invalid") + } + request, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL.String(), nil) + if err != nil { + return 0, fmt.Errorf("build tencent im history request: %w", err) + } + downloadClient := *c.httpClient + if downloadClient.Timeout <= 0 || downloadClient.Timeout < historyDownloadTimeout { + downloadClient.Timeout = historyDownloadTimeout + } + response, err := downloadClient.Do(request) + if err != nil { + var transportErr *url.Error + if errors.As(err, &transportErr) { + // url.Error 会把带临时下载签名的完整 URL 拼进错误；只保留底层网络原因。 + return 0, fmt.Errorf("tencent im history download failed: %w", transportErr.Err) + } + return 0, fmt.Errorf("tencent im history download failed: %w", err) + } + defer response.Body.Close() + if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { + return 0, fmt.Errorf("tencent im history download status %d", response.StatusCode) + } + + archive, err := gzip.NewReader(response.Body) + if err != nil { + return 0, fmt.Errorf("open tencent im history gzip: %w", err) + } + defer archive.Close() + + decoder := json.NewDecoder(archive) + first, err := decoder.Token() + if err != nil { + return 0, fmt.Errorf("decode tencent im history document: %w", err) + } + if delimiter, ok := first.(json.Delim); !ok || delimiter != '{' { + return 0, fmt.Errorf("tencent im history document is not an object") + } + + processed := 0 + for decoder.More() { + keyToken, err := decoder.Token() + if err != nil { + return processed, fmt.Errorf("decode tencent im history field: %w", err) + } + key, ok := keyToken.(string) + if !ok { + return processed, fmt.Errorf("tencent im history field name is invalid") + } + if key != "MsgList" { + // 元数据字段与统计无关；RawMessage 只短暂承载单个标量/小对象，不读取消息正文。 + var ignored json.RawMessage + if err := decoder.Decode(&ignored); err != nil { + return processed, fmt.Errorf("skip tencent im history metadata: %w", err) + } + continue + } + + listStart, err := decoder.Token() + if err != nil { + return processed, fmt.Errorf("decode tencent im history list: %w", err) + } + if delimiter, ok := listStart.(json.Delim); !ok || delimiter != '[' { + return processed, fmt.Errorf("tencent im history MsgList is not an array") + } + for decoder.More() { + var message C2CHistoryMessage + if err := decoder.Decode(&message); err != nil { + return processed, fmt.Errorf("decode tencent im history message: %w", err) + } + if err := consume(message); err != nil { + return processed, err + } + processed++ + } + if _, err := decoder.Token(); err != nil { + return processed, fmt.Errorf("close tencent im history list: %w", err) + } + } + if _, err := decoder.Token(); err != nil { + return processed, fmt.Errorf("close tencent im history document: %w", err) + } + return processed, nil +} diff --git a/services/cron-service/configs/config.docker.yaml b/services/cron-service/configs/config.docker.yaml index 3ea0efa9..549a83c8 100644 --- a/services/cron-service/configs/config.docker.yaml +++ b/services/cron-service/configs/config.docker.yaml @@ -81,6 +81,12 @@ tasks: timeout: "20s" lock_ttl: "5m" batch_size: 10000 + private_message_history: + enabled: true + interval: "1m" + timeout: "2m" + lock_ttl: "3m" + batch_size: 500 game_level_event_relay: enabled: true interval: "5s" diff --git a/services/cron-service/configs/config.tencent.example.yaml b/services/cron-service/configs/config.tencent.example.yaml index 806dbeac..74c5a9f0 100644 --- a/services/cron-service/configs/config.tencent.example.yaml +++ b/services/cron-service/configs/config.tencent.example.yaml @@ -81,6 +81,12 @@ tasks: timeout: "20s" lock_ttl: "5m" batch_size: 10000 + private_message_history: + enabled: true + interval: "1m" + timeout: "2m" + lock_ttl: "3m" + batch_size: 500 game_level_event_relay: enabled: true interval: "5s" diff --git a/services/cron-service/configs/config.yaml b/services/cron-service/configs/config.yaml index 2a6bb3b9..a221c726 100644 --- a/services/cron-service/configs/config.yaml +++ b/services/cron-service/configs/config.yaml @@ -81,6 +81,12 @@ tasks: timeout: "20s" lock_ttl: "5m" batch_size: 10000 + private_message_history: + enabled: true + interval: "1m" + timeout: "2m" + lock_ttl: "3m" + batch_size: 500 game_level_event_relay: enabled: true interval: "5s" diff --git a/services/cron-service/internal/app/app.go b/services/cron-service/internal/app/app.go index e4980126..b59bfd74 100644 --- a/services/cron-service/internal/app/app.go +++ b/services/cron-service/internal/app/app.go @@ -123,6 +123,7 @@ func New(cfg config.Config) (*App, error) { "admin_user_ban_expiry": userCron.ExpireAdminUserBans, // CP 榜单任务只占 cron 租约并调用 user-service；真实 MySQL 读取、wallet 头像框补齐和 Redis 替换都在 user-service 内完成。 "cp_intimacy_leaderboard": userCron.RefreshCPIntimacyLeaderboard, + "private_message_history": userCron.ProcessPrivateMessageHistoryBatch, "message_fanout": activityCron.ProcessMessageFanoutBatch, "growth_level_reward": activityCron.ProcessLevelRewardBatch, "temporary_growth_level": activityCron.ProcessTemporaryLevelBatch, diff --git a/services/cron-service/internal/config/config.go b/services/cron-service/internal/config/config.go index 198ccae3..9b435d2e 100644 --- a/services/cron-service/internal/config/config.go +++ b/services/cron-service/internal/config/config.go @@ -214,6 +214,15 @@ func defaultTasks() map[string]TaskConfig { LockTTL: "5m", BatchSize: 10000, }, + "private_message_history": { + // 腾讯消息文件在小时结束后异步生成；每分钟最多认领一个小时并在 user-service + // 内按 500 条写入，既能回补最近七天，又不会形成下载或数据库突发。 + Enabled: true, + Interval: "1m", + Timeout: "2m", + LockTTL: "3m", + BatchSize: 500, + }, "game_level_event_relay": { Enabled: true, Interval: "5s", diff --git a/services/cron-service/internal/integration/user.go b/services/cron-service/internal/integration/user.go index 120a06ba..616a8219 100644 --- a/services/cron-service/internal/integration/user.go +++ b/services/cron-service/internal/integration/user.go @@ -63,6 +63,13 @@ func (c *UserCronClient) RefreshCPIntimacyLeaderboard(ctx context.Context, req s return userCronResult(resp), err } +// ProcessPrivateMessageHistoryBatch 只触发 user-service owner 拉取离线小时归档； +// cron 不接触腾讯凭证、下载地址、消息内容或 user 数据库。 +func (c *UserCronClient) ProcessPrivateMessageHistoryBatch(ctx context.Context, req scheduler.BatchRequest) (scheduler.BatchResult, error) { + resp, err := c.client.ProcessPrivateMessageHistoryBatch(ctx, userCronRequest(req)) + return userCronResult(resp), err +} + func userCronRequest(req scheduler.BatchRequest) *userv1.CronBatchRequest { return &userv1.CronBatchRequest{ Meta: &userv1.RequestMeta{ diff --git a/services/user-service/deploy/mysql/initdb/001_user_service.sql b/services/user-service/deploy/mysql/initdb/001_user_service.sql index 8826d41e..cea4d564 100644 --- a/services/user-service/deploy/mysql/initdb/001_user_service.sql +++ b/services/user-service/deploy/mysql/initdb/001_user_service.sql @@ -211,6 +211,23 @@ CREATE TABLE IF NOT EXISTS user_private_message_events ( KEY idx_private_message_target_time (app_code, target_user_id, occurred_at_ms, sender_user_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='用户私信发送事实，仅保存统计字段'; +CREATE TABLE IF NOT EXISTS user_private_message_history_sync ( + sdk_app_id BIGINT NOT NULL COMMENT '腾讯 IM SDKAppID', + hour_start_ms BIGINT NOT NULL COMMENT '待同步 UTC 小时起点 epoch ms', + status VARCHAR(16) NOT NULL DEFAULT 'pending' COMMENT 'pending/processing/retry/completed', + attempts INT NOT NULL DEFAULT 0 COMMENT '外部归档拉取尝试次数', + next_retry_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT '失败后最早重试时间', + locked_by VARCHAR(96) NOT NULL DEFAULT '' COMMENT '当前 cron run ID', + locked_until_ms BIGINT NOT NULL DEFAULT 0 COMMENT '处理租约截止时间', + source_message_count INT NOT NULL DEFAULT 0 COMMENT '归档原始消息数', + imported_event_count INT NOT NULL DEFAULT 0 COMMENT '新增 Host 私信事实数', + last_error VARCHAR(512) NOT NULL DEFAULT '' COMMENT '最近一次脱敏错误', + created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', + updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', + PRIMARY KEY (sdk_app_id, hour_start_ms), + KEY idx_private_message_history_sync_ready (sdk_app_id, status, next_retry_at_ms, hour_start_ms, locked_until_ms) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='腾讯 IM 小时归档离线同步状态'; + CREATE TABLE IF NOT EXISTS user_friend_applications ( app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码，用于多租户隔离', requester_user_id BIGINT NOT NULL COMMENT '申请人用户 ID', diff --git a/services/user-service/deploy/mysql/migrations/023_private_message_history_sync.sql b/services/user-service/deploy/mysql/migrations/023_private_message_history_sync.sql new file mode 100644 index 00000000..4ad1d144 --- /dev/null +++ b/services/user-service/deploy/mysql/migrations/023_private_message_history_sync.sql @@ -0,0 +1,20 @@ +USE hyapp_user; + +-- 归档同步表每个 SDKAppID 每小时仅一行，cron 只在该小表上 claim； +-- 私信事实仍通过 user_id 主键校验租户后写入既有索引表，不扫描用户或消息全表。 +CREATE TABLE IF NOT EXISTS user_private_message_history_sync ( + sdk_app_id BIGINT NOT NULL COMMENT '腾讯 IM SDKAppID', + hour_start_ms BIGINT NOT NULL COMMENT '待同步 UTC 小时起点 epoch ms', + status VARCHAR(16) NOT NULL DEFAULT 'pending' COMMENT 'pending/processing/retry/completed', + attempts INT NOT NULL DEFAULT 0 COMMENT '外部归档拉取尝试次数', + next_retry_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT '失败后最早重试时间', + locked_by VARCHAR(96) NOT NULL DEFAULT '' COMMENT '当前 cron run ID', + locked_until_ms BIGINT NOT NULL DEFAULT 0 COMMENT '处理租约截止时间', + source_message_count INT NOT NULL DEFAULT 0 COMMENT '归档原始消息数', + imported_event_count INT NOT NULL DEFAULT 0 COMMENT '新增 Host 私信事实数', + last_error VARCHAR(512) NOT NULL DEFAULT '' COMMENT '最近一次脱敏错误', + created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', + updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', + PRIMARY KEY (sdk_app_id, hour_start_ms), + KEY idx_private_message_history_sync_ready (sdk_app_id, status, next_retry_at_ms, hour_start_ms, locked_until_ms) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='腾讯 IM 小时归档离线同步状态'; diff --git a/services/user-service/internal/app/app.go b/services/user-service/internal/app/app.go index 86355b14..4c9bd1be 100644 --- a/services/user-service/internal/app/app.go +++ b/services/user-service/internal/app/app.go @@ -192,6 +192,7 @@ func New(cfg config.Config) (*App, error) { } var imLoginKicker userservice.IMLoginKicker var imAccountImporter authservice.IMAccountImporter + var imHistoryClient hostservice.PrivateMessageHistoryClient if cfg.TencentIM.Enabled { imClient, err := tencentim.NewRESTClient(cfg.TencentIM.RESTConfig()) if err != nil { @@ -207,6 +208,7 @@ func New(cfg config.Config) (*App, error) { } imLoginKicker = imClient imAccountImporter = tencentIMAccountImporter{client: imClient} + imHistoryClient = imClient } var roomConn *grpc.ClientConn var roomEvictor userservice.RoomEvictor @@ -323,6 +325,8 @@ func New(cfg config.Config) (*App, error) { ) hostSvc := hostservice.New(hostRepo, hostservice.WithIDGenerator(idgen.NewInt64Generator(cfg.IDGenerator.NodeID)), + // 私聊人数来自腾讯每小时离线归档；该依赖只暴露给 cron RPC，不进入 IM 发消息链路。 + hostservice.WithPrivateMessageHistorySource(cfg.TencentIM.SDKAppID, imHistoryClient), ) // user service 负责用户主状态和 display_user_id/靓号用例，不进入房间高频流程。 userSvc := userservice.New(userRepo, diff --git a/services/user-service/internal/service/host/private_message_history.go b/services/user-service/internal/service/host/private_message_history.go new file mode 100644 index 00000000..b5622263 --- /dev/null +++ b/services/user-service/internal/service/host/private_message_history.go @@ -0,0 +1,233 @@ +package host + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "hyapp/pkg/tencentim" + "hyapp/pkg/xerr" + hostdomain "hyapp/services/user-service/internal/domain/host" +) + +const ( + privateMessageHistoryHours = 7 * 24 + privateMessageHistoryReadyDelay = 2 * time.Hour + privateMessageHistoryRetryDelay = 10 * time.Minute + privateMessageHistoryImportMax = 500 + privateMessageHistoryEmptyMaxWait = 4 * time.Hour +) + +// PrivateMessageHistoryClient 是腾讯 IM 离线归档边界；它只在 cron 批处理中调用， +// 不参与客户端发送、接收或 UserSig 签发链路。 +type PrivateMessageHistoryClient interface { + ListC2CHistoryFiles(ctx context.Context, hourStartUTC time.Time) ([]tencentim.C2CHistoryFile, error) + StreamC2CHistory(ctx context.Context, file tencentim.C2CHistoryFile, consume func(tencentim.C2CHistoryMessage) error) (int, error) +} + +// PrivateMessageHistoryRepository 保存小时同步租约和最小私聊统计事实。 +type PrivateMessageHistoryRepository interface { + PreparePrivateMessageHistoryHours(ctx context.Context, sdkAppID int64, fromHourMS int64, throughHourMS int64, nowMS int64) error + ClaimPrivateMessageHistoryHour(ctx context.Context, sdkAppID int64, runID string, fromHourMS int64, throughHourMS int64, nowMS int64, lockTTL time.Duration) (hourStartMS int64, attempts int, claimed bool, err error) + ImportPrivateMessageEvents(ctx context.Context, events []hostdomain.PrivateMessageEvent) (int, error) + CompletePrivateMessageHistoryHour(ctx context.Context, sdkAppID int64, hourStartMS int64, runID string, sourceMessages int, importedEvents int, nowMS int64) error + RetryPrivateMessageHistoryHour(ctx context.Context, sdkAppID int64, hourStartMS int64, runID string, retryAtMS int64, errorMessage string, nowMS int64) error +} + +// PrivateMessageHistoryBatchResult 汇总一次小时归档处理，不把下载地址或消息正文带到 cron 日志。 +type PrivateMessageHistoryBatchResult struct { + Claimed bool + SourceCount int + ImportedCount int + RetryPending bool +} + +// WithPrivateMessageHistorySource 挂载腾讯归档和 owner repository。SDKAppID 是共享归档主键， +// 即使 cron 按多个 app_code 并发触发，同一小时也只会被认领一次。 +func WithPrivateMessageHistorySource(sdkAppID int64, client PrivateMessageHistoryClient) Option { + return func(s *Service) { + if sdkAppID <= 0 || client == nil { + return + } + repository, ok := s.repository.(PrivateMessageHistoryRepository) + if !ok { + return + } + s.privateMessageHistorySDKAppID = sdkAppID + s.privateMessageHistoryClient = client + s.privateMessageHistoryRepository = repository + } +} + +// ProcessPrivateMessageHistoryBatch 处理最近七天中一个已完成的 UTC 小时。 +// get_history 文件由腾讯异步生成，因此统计最多延迟约两小时，但不会给 IM 发消息增加等待。 +func (s *Service) ProcessPrivateMessageHistoryBatch(ctx context.Context, runID string, lockTTL time.Duration, importBatchSize int) (PrivateMessageHistoryBatchResult, error) { + if s == nil || s.privateMessageHistoryClient == nil || s.privateMessageHistoryRepository == nil || s.privateMessageHistorySDKAppID <= 0 { + return PrivateMessageHistoryBatchResult{}, xerr.New(xerr.Unavailable, "private message history sync is not configured") + } + runID = strings.TrimSpace(runID) + if runID == "" { + return PrivateMessageHistoryBatchResult{}, xerr.New(xerr.InvalidArgument, "run_id is required") + } + if importBatchSize <= 0 || importBatchSize > privateMessageHistoryImportMax { + importBatchSize = privateMessageHistoryImportMax + } + if lockTTL <= 0 { + lockTTL = 5 * time.Minute + } + + now := s.now().UTC() + throughHour := now.Truncate(time.Hour).Add(-privateMessageHistoryReadyDelay) + fromHour := throughHour.Add(-time.Duration(privateMessageHistoryHours-1) * time.Hour) + if err := s.privateMessageHistoryRepository.PreparePrivateMessageHistoryHours( + ctx, + s.privateMessageHistorySDKAppID, + fromHour.UnixMilli(), + throughHour.UnixMilli(), + now.UnixMilli(), + ); err != nil { + return PrivateMessageHistoryBatchResult{}, err + } + + hourStartMS, attempts, claimed, err := s.privateMessageHistoryRepository.ClaimPrivateMessageHistoryHour( + ctx, + s.privateMessageHistorySDKAppID, + runID, + fromHour.UnixMilli(), + throughHour.UnixMilli(), + now.UnixMilli(), + lockTTL, + ) + if err != nil || !claimed { + return PrivateMessageHistoryBatchResult{}, err + } + result := PrivateMessageHistoryBatchResult{Claimed: true} + hourStart := time.UnixMilli(hourStartMS).UTC() + + files, err := s.privateMessageHistoryClient.ListC2CHistoryFiles(ctx, hourStart) + if err != nil { + if tencentim.IsRESTErrorCode(err, 1004) { + // 1004 同时表示“无消息”和“文件尚未生成”。最近小时先重试；超过生成窗口后 + // 视为空小时完成，避免一个永久空小时阻塞七天回补。 + if attempts < 3 && now.Sub(hourStart) < privateMessageHistoryEmptyMaxWait { + if retryErr := s.markPrivateMessageHistoryRetry(ctx, hourStartMS, runID, err, now); retryErr != nil { + return result, retryErr + } + result.RetryPending = true + return result, nil + } + if completeErr := s.privateMessageHistoryRepository.CompletePrivateMessageHistoryHour( + ctx, s.privateMessageHistorySDKAppID, hourStartMS, runID, 0, 0, now.UnixMilli(), + ); completeErr != nil { + return result, completeErr + } + return result, nil + } + if retryErr := s.markPrivateMessageHistoryRetry(ctx, hourStartMS, runID, err, now); retryErr != nil { + return result, retryErr + } + return result, err + } + + pending := make([]hostdomain.PrivateMessageEvent, 0, importBatchSize) + flush := func() error { + if len(pending) == 0 { + return nil + } + imported, importErr := s.privateMessageHistoryRepository.ImportPrivateMessageEvents(ctx, pending) + if importErr != nil { + return importErr + } + result.ImportedCount += imported + pending = pending[:0] + return nil + } + consume := func(message tencentim.C2CHistoryMessage) error { + event, ok := privateMessageEventFromHistory(message) + if !ok { + return nil + } + pending = append(pending, event) + if len(pending) >= importBatchSize { + return flush() + } + return nil + } + + for _, file := range files { + processed, streamErr := s.privateMessageHistoryClient.StreamC2CHistory(ctx, file, consume) + result.SourceCount += processed + if streamErr != nil { + if retryErr := s.markPrivateMessageHistoryRetry(ctx, hourStartMS, runID, streamErr, now); retryErr != nil { + return result, retryErr + } + return result, streamErr + } + } + if err := flush(); err != nil { + if retryErr := s.markPrivateMessageHistoryRetry(ctx, hourStartMS, runID, err, now); retryErr != nil { + return result, retryErr + } + return result, err + } + if err := s.privateMessageHistoryRepository.CompletePrivateMessageHistoryHour( + ctx, + s.privateMessageHistorySDKAppID, + hourStartMS, + runID, + result.SourceCount, + result.ImportedCount, + s.now().UTC().UnixMilli(), + ); err != nil { + return result, err + } + return result, nil +} + +func (s *Service) markPrivateMessageHistoryRetry(ctx context.Context, hourStartMS int64, runID string, cause error, now time.Time) error { + retryErr := s.privateMessageHistoryRepository.RetryPrivateMessageHistoryHour( + ctx, + s.privateMessageHistorySDKAppID, + hourStartMS, + runID, + now.Add(privateMessageHistoryRetryDelay).UnixMilli(), + cause.Error(), + now.UnixMilli(), + ) + if retryErr == nil { + return nil + } + return errors.Join(cause, retryErr) +} + +func privateMessageEventFromHistory(message tencentim.C2CHistoryMessage) (hostdomain.PrivateMessageEvent, bool) { + senderText := strings.TrimSpace(message.FromAccount) + targetText := strings.TrimSpace(message.ToAccount) + senderUserID, senderErr := strconv.ParseInt(senderText, 10, 64) + targetUserID, targetErr := strconv.ParseInt(targetText, 10, 64) + if senderErr != nil || targetErr != nil || senderUserID <= 0 || targetUserID <= 0 || message.Timestamp <= 0 { + // 共享 SDKAppID 可能包含管理员/第三方非数字 identifier；只有项目内部不可变 user_id + // 才能安全映射 users 主键，其余消息不属于 Host Center 统计事实。 + return hostdomain.PrivateMessageEvent{}, false + } + identity := fmt.Sprintf( + "%s\x00%s\x00%d\x00%d\x00%d", + senderText, + targetText, + message.Timestamp, + message.Sequence, + message.Random, + ) + hash := sha256.Sum256([]byte(identity)) + return hostdomain.PrivateMessageEvent{ + EventID: "im-history:" + hex.EncodeToString(hash[:]), + SenderUserID: senderUserID, + TargetUserID: targetUserID, + OccurredAtMS: message.Timestamp * int64(time.Second/time.Millisecond), + }, true +} diff --git a/services/user-service/internal/service/host/service.go b/services/user-service/internal/service/host/service.go index dea22f8a..96dedc88 100644 --- a/services/user-service/internal/service/host/service.go +++ b/services/user-service/internal/service/host/service.go @@ -70,8 +70,12 @@ type IDGenerator interface { type Service struct { repository Repository roleScopePolicyRepository RoleScopePolicyRepository - idGenerator IDGenerator - now func() time.Time + // privateMessageHistory* 只用于 cron 离线同步，不进入任何 Host Center 查询或 IM 实时链路。 + privateMessageHistoryRepository PrivateMessageHistoryRepository + privateMessageHistoryClient PrivateMessageHistoryClient + privateMessageHistorySDKAppID int64 + idGenerator IDGenerator + now func() time.Time } // Option 调整 Host 业务层的运行时依赖，测试可以注入固定时钟和确定性发号器。 diff --git a/services/user-service/internal/storage/mysql/host/private_message_history.go b/services/user-service/internal/storage/mysql/host/private_message_history.go new file mode 100644 index 00000000..c7bebe3a --- /dev/null +++ b/services/user-service/internal/storage/mysql/host/private_message_history.go @@ -0,0 +1,210 @@ +package host + +import ( + "context" + "database/sql" + "fmt" + "strings" + "time" + + "hyapp/pkg/xerr" + hostdomain "hyapp/services/user-service/internal/domain/host" +) + +const privateMessageHistoryMaxHours = 7 * 24 + +// PreparePrivateMessageHistoryHours 只向小时状态小表补齐最近七天的主键，不更新已完成行。 +// 多个 app_code 的 cron 并发调用会由 INSERT IGNORE 和共享 SDKAppID 主键自然收敛。 +func (r *Repository) PreparePrivateMessageHistoryHours(ctx context.Context, sdkAppID int64, fromHourMS int64, throughHourMS int64, nowMS int64) error { + if r == nil || r.db == nil { + return xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + if sdkAppID <= 0 || fromHourMS <= 0 || throughHourMS < fromHourMS { + return xerr.New(xerr.InvalidArgument, "private message history range is invalid") + } + const hourMS = int64(time.Hour / time.Millisecond) + hourCount := int((throughHourMS-fromHourMS)/hourMS) + 1 + if hourCount <= 0 || hourCount > privateMessageHistoryMaxHours { + return xerr.New(xerr.InvalidArgument, "private message history range exceeds seven days") + } + + values := make([]string, 0, hourCount) + args := make([]any, 0, hourCount*4) + for hourStartMS := fromHourMS; hourStartMS <= throughHourMS; hourStartMS += hourMS { + values = append(values, "(?, ?, 'pending', ?, ?)") + args = append(args, sdkAppID, hourStartMS, nowMS, nowMS) + } + _, err := r.db.ExecContext(ctx, ` + INSERT IGNORE INTO user_private_message_history_sync ( + sdk_app_id, hour_start_ms, status, created_at_ms, updated_at_ms + ) VALUES `+strings.Join(values, ","), args...) + return err +} + +// ClaimPrivateMessageHistoryHour 原子认领最新的可处理小时，让当天 Host Center 数据优先恢复。 +// 状态表最多保留最近七天新增的小时行，ORDER BY/LIMIT 不会触碰消息事实大表。 +func (r *Repository) ClaimPrivateMessageHistoryHour(ctx context.Context, sdkAppID int64, runID string, fromHourMS int64, throughHourMS int64, nowMS int64, lockTTL time.Duration) (int64, int, bool, error) { + if r == nil || r.db == nil { + return 0, 0, false, xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + runID = strings.TrimSpace(runID) + if sdkAppID <= 0 || runID == "" || fromHourMS <= 0 || throughHourMS < fromHourMS { + return 0, 0, false, xerr.New(xerr.InvalidArgument, "private message history claim is invalid") + } + if lockTTL <= 0 { + lockTTL = 5 * time.Minute + } + lockedUntilMS := nowMS + lockTTL.Milliseconds() + result, err := r.db.ExecContext(ctx, ` + UPDATE user_private_message_history_sync + SET status = 'processing', + attempts = attempts + 1, + locked_by = ?, + locked_until_ms = ?, + updated_at_ms = ? + WHERE sdk_app_id = ? + AND hour_start_ms >= ? AND hour_start_ms <= ? + AND ( + (status IN ('pending', 'retry') AND next_retry_at_ms <= ?) + OR (status = 'processing' AND locked_until_ms <= ?) + ) + ORDER BY hour_start_ms DESC + LIMIT 1`, + runID, lockedUntilMS, nowMS, + sdkAppID, fromHourMS, throughHourMS, nowMS, nowMS, + ) + if err != nil { + return 0, 0, false, err + } + affected, err := result.RowsAffected() + if err != nil || affected == 0 { + return 0, 0, false, err + } + + var hourStartMS int64 + var attempts int + err = r.db.QueryRowContext(ctx, ` + SELECT hour_start_ms, attempts + FROM user_private_message_history_sync + WHERE sdk_app_id = ? AND status = 'processing' AND locked_by = ? AND locked_until_ms = ? + LIMIT 1`, + sdkAppID, runID, lockedUntilMS, + ).Scan(&hourStartMS, &attempts) + if err != nil { + if err == sql.ErrNoRows { + return 0, 0, false, xerr.New(xerr.Conflict, "private message history claim was lost") + } + return 0, 0, false, err + } + return hourStartMS, attempts, true, nil +} + +// ImportPrivateMessageEvents 批量把腾讯归档转换为最小统计事实。users 主键分别验证收发双方， +// app_code 取接收者主数据且要求双方同租户；host_profiles 联接避免保存与 Host Center 无关的普通私聊。 +func (r *Repository) ImportPrivateMessageEvents(ctx context.Context, events []hostdomain.PrivateMessageEvent) (int, error) { + if r == nil || r.db == nil { + return 0, xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + if len(events) == 0 { + return 0, nil + } + if len(events) > 500 { + return 0, xerr.New(xerr.InvalidArgument, "private message history import batch exceeds 500") + } + + createdAtMS := time.Now().UTC().UnixMilli() + selects := make([]string, 0, len(events)) + args := make([]any, 0, len(events)*5) + for index, event := range events { + prefix := "SELECT" + if index > 0 { + prefix = "UNION ALL SELECT" + } + selects = append(selects, prefix+" ? AS event_id, ? AS sender_user_id, ? AS target_user_id, ? AS occurred_at_ms, ? AS created_at_ms") + args = append(args, event.EventID, event.SenderUserID, event.TargetUserID, event.OccurredAtMS, createdAtMS) + } + query := ` + INSERT IGNORE INTO user_private_message_events ( + app_code, event_id, sender_user_id, target_user_id, occurred_at_ms, created_at_ms + ) + SELECT target.app_code, + imported.event_id, + sender.user_id, + target.user_id, + imported.occurred_at_ms, + imported.created_at_ms + FROM (` + strings.Join(selects, "\n") + `) imported + INNER JOIN users target + ON target.user_id = imported.target_user_id + INNER JOIN users sender + ON sender.user_id = imported.sender_user_id AND sender.app_code = target.app_code + INNER JOIN host_profiles host + ON host.app_code = target.app_code AND host.user_id = target.user_id` + + result, err := r.db.ExecContext(ctx, query, args...) + if err != nil { + return 0, fmt.Errorf("import private message history events: %w", err) + } + affected, err := result.RowsAffected() + return int(affected), err +} + +// CompletePrivateMessageHistoryHour 仅允许当前 run 完成自己的租约，防止超时旧 worker +// 覆盖新 worker 的重试结果。 +func (r *Repository) CompletePrivateMessageHistoryHour(ctx context.Context, sdkAppID int64, hourStartMS int64, runID string, sourceMessages int, importedEvents int, nowMS int64) error { + result, err := r.db.ExecContext(ctx, ` + UPDATE user_private_message_history_sync + SET status = 'completed', + source_message_count = ?, + imported_event_count = ?, + next_retry_at_ms = 0, + locked_by = '', + locked_until_ms = 0, + last_error = '', + updated_at_ms = ? + WHERE sdk_app_id = ? AND hour_start_ms = ? AND status = 'processing' AND locked_by = ?`, + sourceMessages, importedEvents, nowMS, sdkAppID, hourStartMS, strings.TrimSpace(runID), + ) + if err != nil { + return err + } + affected, err := result.RowsAffected() + if err != nil { + return err + } + if affected == 0 { + return xerr.New(xerr.Conflict, "private message history completion lease was lost") + } + return nil +} + +// RetryPrivateMessageHistoryHour 记录脱敏错误并延后重试；下载 URL 和 UserSig 已在 REST +// 边界被剥离，last_error 仍限制长度，避免外部响应放大状态表。 +func (r *Repository) RetryPrivateMessageHistoryHour(ctx context.Context, sdkAppID int64, hourStartMS int64, runID string, retryAtMS int64, errorMessage string, nowMS int64) error { + errorMessage = strings.TrimSpace(errorMessage) + if len(errorMessage) > 512 { + errorMessage = errorMessage[:512] + } + result, err := r.db.ExecContext(ctx, ` + UPDATE user_private_message_history_sync + SET status = 'retry', + next_retry_at_ms = ?, + locked_by = '', + locked_until_ms = 0, + last_error = ?, + updated_at_ms = ? + WHERE sdk_app_id = ? AND hour_start_ms = ? AND status = 'processing' AND locked_by = ?`, + retryAtMS, errorMessage, nowMS, sdkAppID, hourStartMS, strings.TrimSpace(runID), + ) + if err != nil { + return err + } + affected, err := result.RowsAffected() + if err != nil { + return err + } + if affected == 0 { + return xerr.New(xerr.Conflict, "private message history retry lease was lost") + } + return nil +} diff --git a/services/user-service/internal/transport/grpc/cron.go b/services/user-service/internal/transport/grpc/cron.go index c6d8a721..41379335 100644 --- a/services/user-service/internal/transport/grpc/cron.go +++ b/services/user-service/internal/transport/grpc/cron.go @@ -162,6 +162,40 @@ func (s *Server) RefreshCPIntimacyLeaderboard(ctx context.Context, req *userv1.C }, nil } +// ProcessPrivateMessageHistoryBatch 从腾讯 IM 小时归档恢复 Host 私聊发送人数事实。 +// 任务只在 cron 后台运行；即使腾讯归档延迟或下载失败，也不会影响客户端私聊发送。 +func (s *Server) ProcessPrivateMessageHistoryBatch(ctx context.Context, req *userv1.CronBatchRequest) (*userv1.CronBatchResponse, error) { + ctx = contextWithApp(ctx, req.GetMeta()) + if s.hostSvc == nil { + return nil, xerr.ToGRPCError(xerr.New(xerr.Unavailable, "host service is not configured")) + } + result, err := s.hostSvc.ProcessPrivateMessageHistoryBatch( + ctx, + req.GetRunId(), + durationFromMillis(req.GetLockTtlMs()), + int(req.GetBatchSize()), + ) + if err != nil { + return nil, xerr.ToGRPCError(err) + } + claimed := int32(0) + if result.Claimed { + claimed = 1 + } + failureCount := int32(0) + if result.RetryPending { + failureCount = 1 + } + return &userv1.CronBatchResponse{ + ClaimedCount: claimed, + ProcessedCount: int32(result.SourceCount), + SuccessCount: int32(result.ImportedCount), + FailureCount: failureCount, + // 小时归档有意按 cron interval 节流；回补不通过 has_more 形成无间隔下载突发。 + HasMore: false, + }, nil +} + func durationFromMillis(value int64) time.Duration { if value <= 0 { return 0