84 lines
3.3 KiB
Go
84 lines
3.3 KiB
Go
// Package integration adapts owner-service gRPC APIs into scheduler handlers.
|
||
package integration
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
userv1 "hyapp.local/api/proto/user/v1"
|
||
"hyapp/services/cron-service/internal/scheduler"
|
||
)
|
||
|
||
// UserCronClient delegates user-owned background batches to user-service.
|
||
type UserCronClient struct {
|
||
client userv1.UserCronServiceClient
|
||
}
|
||
|
||
// NewUserCronClient creates scheduler handlers backed by user-service gRPC.
|
||
func NewUserCronClient(client userv1.UserCronServiceClient) *UserCronClient {
|
||
return &UserCronClient{client: client}
|
||
}
|
||
|
||
// ProcessLoginIPRiskBatch handles login_ip_risk_jobs.
|
||
func (c *UserCronClient) ProcessLoginIPRiskBatch(ctx context.Context, req scheduler.BatchRequest) (scheduler.BatchResult, error) {
|
||
resp, err := c.client.ProcessLoginIPRiskBatch(ctx, userCronRequest(req))
|
||
return userCronResult(resp), err
|
||
}
|
||
|
||
// ProcessRegionRebuildBatch handles user_region_rebuild_tasks.
|
||
func (c *UserCronClient) ProcessRegionRebuildBatch(ctx context.Context, req scheduler.BatchRequest) (scheduler.BatchResult, error) {
|
||
resp, err := c.client.ProcessRegionRebuildBatch(ctx, userCronRequest(req))
|
||
return userCronResult(resp), err
|
||
}
|
||
|
||
// CompensateMicOpenSessions handles abnormal user_mic_sessions closure.
|
||
func (c *UserCronClient) CompensateMicOpenSessions(ctx context.Context, req scheduler.BatchRequest) (scheduler.BatchResult, error) {
|
||
protoReq := userCronRequest(req)
|
||
resp, err := c.client.CompensateMicOpenSessions(ctx, protoReq)
|
||
return userCronResult(resp), err
|
||
}
|
||
|
||
// ExpireManagerUserBlocks handles timed manager-center bans after their unblock time.
|
||
func (c *UserCronClient) ExpireManagerUserBlocks(ctx context.Context, req scheduler.BatchRequest) (scheduler.BatchResult, error) {
|
||
resp, err := c.client.ExpireManagerUserBlocks(ctx, userCronRequest(req))
|
||
return userCronResult(resp), err
|
||
}
|
||
|
||
// RefreshCPIntimacyLeaderboard 只触发 user-service 重建 CP 亲密值 zset 读模型，cron 不读取业务表。
|
||
func (c *UserCronClient) RefreshCPIntimacyLeaderboard(ctx context.Context, req scheduler.BatchRequest) (scheduler.BatchResult, error) {
|
||
// userCronRequest 会带 run_id、worker_id、app_code 和 batch_size；user-service 用这些值做日志追踪、租户隔离和刷新上限。
|
||
resp, err := c.client.RefreshCPIntimacyLeaderboard(ctx, userCronRequest(req))
|
||
return userCronResult(resp), err
|
||
}
|
||
|
||
func userCronRequest(req scheduler.BatchRequest) *userv1.CronBatchRequest {
|
||
return &userv1.CronBatchRequest{
|
||
Meta: &userv1.RequestMeta{
|
||
RequestId: req.RunID,
|
||
Caller: "cron-service",
|
||
SentAtMs: time.Now().UnixMilli(),
|
||
AppCode: req.AppCode,
|
||
GatewayNodeId: req.WorkerID,
|
||
},
|
||
RunId: req.RunID,
|
||
WorkerId: req.WorkerID,
|
||
BatchSize: int32(req.BatchSize),
|
||
LockTtlMs: req.LockTTL.Milliseconds(),
|
||
PendingPublishMaxAgeMs: req.PendingPublishMaxAge.Milliseconds(),
|
||
PublishingSessionMaxAgeMs: req.PublishingSessionMaxAge.Milliseconds(),
|
||
}
|
||
}
|
||
|
||
func userCronResult(resp *userv1.CronBatchResponse) scheduler.BatchResult {
|
||
if resp == nil {
|
||
return scheduler.BatchResult{}
|
||
}
|
||
return scheduler.BatchResult{
|
||
ClaimedCount: int(resp.GetClaimedCount()),
|
||
ProcessedCount: int(resp.GetProcessedCount()),
|
||
SuccessCount: int(resp.GetSuccessCount()),
|
||
FailureCount: int(resp.GetFailureCount()),
|
||
HasMore: resp.GetHasMore(),
|
||
}
|
||
}
|