91 lines
3.4 KiB
Go
91 lines
3.4 KiB
Go
package integration
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
activityv1 "hyapp.local/api/proto/activity/v1"
|
||
"hyapp/services/cron-service/internal/scheduler"
|
||
)
|
||
|
||
// ActivityCronClient delegates activity-owned background batches to activity-service.
|
||
type ActivityCronClient struct {
|
||
client activityv1.ActivityCronServiceClient
|
||
}
|
||
|
||
// NewActivityCronClient creates scheduler handlers backed by activity-service gRPC.
|
||
func NewActivityCronClient(client activityv1.ActivityCronServiceClient) *ActivityCronClient {
|
||
return &ActivityCronClient{client: client}
|
||
}
|
||
|
||
// ProcessMessageFanoutBatch handles message_fanout_jobs.
|
||
func (c *ActivityCronClient) ProcessMessageFanoutBatch(ctx context.Context, req scheduler.BatchRequest) (scheduler.BatchResult, error) {
|
||
resp, err := c.client.ProcessMessageFanoutBatch(ctx, activityCronBatchRequest(req))
|
||
if err != nil {
|
||
return scheduler.BatchResult{}, err
|
||
}
|
||
return activityCronBatchResult(resp), nil
|
||
}
|
||
|
||
// ProcessLevelRewardBatch handles growth_level_reward_jobs.
|
||
func (c *ActivityCronClient) ProcessLevelRewardBatch(ctx context.Context, req scheduler.BatchRequest) (scheduler.BatchResult, error) {
|
||
resp, err := c.client.ProcessLevelRewardBatch(ctx, activityCronBatchRequest(req))
|
||
if err != nil {
|
||
return scheduler.BatchResult{}, err
|
||
}
|
||
return activityCronBatchResult(resp), nil
|
||
}
|
||
|
||
// ProcessAchievementRewardBatch handles achievement_reward_jobs.
|
||
func (c *ActivityCronClient) ProcessAchievementRewardBatch(ctx context.Context, req scheduler.BatchRequest) (scheduler.BatchResult, error) {
|
||
resp, err := c.client.ProcessAchievementRewardBatch(ctx, activityCronBatchRequest(req))
|
||
if err != nil {
|
||
return scheduler.BatchResult{}, err
|
||
}
|
||
return activityCronBatchResult(resp), nil
|
||
}
|
||
|
||
// ProcessWeeklyStarSettlementBatch handles ended weekly-star cycle settlements.
|
||
func (c *ActivityCronClient) ProcessWeeklyStarSettlementBatch(ctx context.Context, req scheduler.BatchRequest) (scheduler.BatchResult, error) {
|
||
resp, err := c.client.ProcessWeeklyStarSettlementBatch(ctx, activityCronBatchRequest(req))
|
||
if err != nil {
|
||
return scheduler.BatchResult{}, err
|
||
}
|
||
return activityCronBatchResult(resp), nil
|
||
}
|
||
|
||
// ProcessRoomTurnoverRewardSettlementBatch handles ended room-turnover-reward cycle settlements.
|
||
func (c *ActivityCronClient) ProcessRoomTurnoverRewardSettlementBatch(ctx context.Context, req scheduler.BatchRequest) (scheduler.BatchResult, error) {
|
||
// cron-service 不直连 activity MySQL；它只把 run_id、worker_id 和 batch_size 传给活动服务，让 owner service 自己处理幂等和状态推进。
|
||
resp, err := c.client.ProcessRoomTurnoverRewardSettlementBatch(ctx, activityCronBatchRequest(req))
|
||
if err != nil {
|
||
return scheduler.BatchResult{}, err
|
||
}
|
||
return activityCronBatchResult(resp), nil
|
||
}
|
||
|
||
func activityCronBatchRequest(req scheduler.BatchRequest) *activityv1.CronBatchRequest {
|
||
return &activityv1.CronBatchRequest{
|
||
Meta: &activityv1.RequestMeta{
|
||
RequestId: req.RunID,
|
||
Caller: "cron-service",
|
||
SentAtMs: time.Now().UnixMilli(),
|
||
AppCode: req.AppCode,
|
||
},
|
||
RunId: req.RunID,
|
||
WorkerId: req.WorkerID,
|
||
BatchSize: int32(req.BatchSize),
|
||
LockTtlMs: req.LockTTL.Milliseconds(),
|
||
}
|
||
}
|
||
|
||
func activityCronBatchResult(resp *activityv1.CronBatchResponse) scheduler.BatchResult {
|
||
return scheduler.BatchResult{
|
||
ClaimedCount: int(resp.GetClaimedCount()),
|
||
ProcessedCount: int(resp.GetProcessedCount()),
|
||
SuccessCount: int(resp.GetSuccessCount()),
|
||
FailureCount: int(resp.GetFailureCount()),
|
||
HasMore: resp.GetHasMore(),
|
||
}
|
||
}
|