108 lines
3.4 KiB
Go
108 lines
3.4 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"flag"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"os/signal"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
|
|
"hyapp/pkg/appcode"
|
|
"hyapp/services/wallet-service/internal/config"
|
|
walletservice "hyapp/services/wallet-service/internal/service/wallet"
|
|
mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql"
|
|
)
|
|
|
|
// main 只做一次性礼物墙回填：复用 wallet-service 的投影状态机，不手写聚合 SQL。
|
|
func main() {
|
|
configPath := flag.String("config", "services/wallet-service/configs/config.yaml", "wallet-service config path")
|
|
appCode := flag.String("app-code", appcode.Default, "app code to backfill")
|
|
workerID := flag.String("worker-id", "gift-wall-backfill", "projection worker id")
|
|
batchSize := flag.Int("batch-size", 50, "events claimed per batch")
|
|
maxBatches := flag.Int("max-batches", 0, "maximum batches to process; 0 means until drained")
|
|
progressEvery := flag.Int("progress-every", 10, "print progress every N non-empty batches")
|
|
lockTTL := flag.Duration("lock-ttl", 30*time.Second, "projection lock ttl")
|
|
sleep := flag.Duration("sleep", 200*time.Millisecond, "pause between non-empty batches")
|
|
dryRun := flag.Bool("dry-run", false, "load config and ping mysql without processing")
|
|
flag.Parse()
|
|
|
|
normalizedAppCode := appcode.Normalize(strings.TrimSpace(*appCode))
|
|
if normalizedAppCode == "" {
|
|
log.Fatal("app-code is required")
|
|
}
|
|
normalizedWorkerID := strings.TrimSpace(*workerID)
|
|
if normalizedWorkerID == "" {
|
|
log.Fatal("worker-id is required")
|
|
}
|
|
|
|
cfg, err := config.Load(*configPath)
|
|
if err != nil {
|
|
log.Fatalf("load config: %v", err)
|
|
}
|
|
|
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
|
defer stop()
|
|
ctx = appcode.WithContext(ctx, normalizedAppCode)
|
|
|
|
repository, err := mysqlstorage.Open(ctx, cfg.MySQLDSN)
|
|
if err != nil {
|
|
log.Fatalf("open mysql: %v", err)
|
|
}
|
|
defer func() { _ = repository.Close() }()
|
|
|
|
if *dryRun {
|
|
if err := repository.Ping(ctx); err != nil {
|
|
log.Fatalf("ping mysql: %v", err)
|
|
}
|
|
fmt.Printf("dry_run_ok app_code=%s batch_size=%d worker_id=%s\n", normalizedAppCode, *batchSize, normalizedWorkerID)
|
|
return
|
|
}
|
|
if *progressEvery <= 0 {
|
|
*progressEvery = 1
|
|
}
|
|
|
|
svc := walletservice.New(repository)
|
|
total := 0
|
|
emptyBatches := 0
|
|
started := time.Now()
|
|
for batchNo := 1; ; batchNo++ {
|
|
if *maxBatches > 0 && batchNo > *maxBatches {
|
|
break
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
fmt.Fprintf(os.Stderr, "stopped app_code=%s total=%d batches=%d duration=%s\n", normalizedAppCode, total, batchNo-1, time.Since(started).Round(time.Millisecond))
|
|
return
|
|
default:
|
|
}
|
|
|
|
// ProjectPendingGiftWallEvents 会认领缺失或失败的事件，并在同一事务里写 user_gift_wall 与 done 标记。
|
|
processed, err := svc.ProjectPendingGiftWallEvents(ctx, normalizedWorkerID, *batchSize, *lockTTL)
|
|
if err != nil {
|
|
log.Fatalf("project batch %d: %v", batchNo, err)
|
|
}
|
|
total += processed
|
|
if processed == 0 || batchNo == 1 || batchNo%*progressEvery == 0 {
|
|
fmt.Printf("batch=%d processed=%d total=%d elapsed=%s\n", batchNo, processed, total, time.Since(started).Round(time.Millisecond))
|
|
}
|
|
|
|
if processed == 0 {
|
|
emptyBatches++
|
|
if emptyBatches >= 2 {
|
|
break
|
|
}
|
|
time.Sleep(*sleep)
|
|
continue
|
|
}
|
|
emptyBatches = 0
|
|
if *sleep > 0 {
|
|
time.Sleep(*sleep)
|
|
}
|
|
}
|
|
fmt.Printf("done app_code=%s total=%d duration=%s\n", normalizedAppCode, total, time.Since(started).Round(time.Millisecond))
|
|
}
|