增加分区限量outbox回放工具

pkg/outboxreplay/replay.go

@@ -0,0 +1,88 @@
+// Package outboxreplay provides the process-level hard bound used by manual
+// owner-outbox repairs. Normal workers intentionally keep polling; an operator
+// replay must instead stop after an exact maximum even while the partition
+// remains non-empty.
+package outboxreplay
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"hyapp/pkg/appcode"
+)
+
+const ExecuteConfirmation = "REPLAY_EXACT_APP_OUTBOX_BATCH"
+
+// Scope is the immutable tenant and total-record bound for one process run.
+// Limit is not a worker batch size: Run enforces it across every claim/publish
+// iteration and returns once that many rows have been handled.
+type Scope struct {
+	AppCode string
+	Limit   int
+}
+
+// Result reports whether the selected App partition became empty before the
+// hard limit. Processed counts rows claimed by processOne, including a row that
+// was released as retryable after a publish failure.
+type Result struct {
+	Processed int
+	Drained   bool
+}
+
+// NewScope rejects an implicit tenant or an unbounded run before any database
+// or MQ client is opened. maxLimit is command-specific so a service may impose
+// a smaller operational envelope without weakening this shared contract.
+func NewScope(rawAppCode string, limit int, maxLimit int) (Scope, error) {
+	if strings.TrimSpace(rawAppCode) == "" {
+		// appcode.Normalize intentionally preserves a legacy empty=>lalu default
+		// for request compatibility. An operator command must reject omission
+		// before normalization or a missing flag would target a real tenant.
+		return Scope{}, errors.New("app-code is required")
+	}
+	appCode := appcode.Normalize(rawAppCode)
+	if maxLimit <= 0 {
+		return Scope{}, errors.New("max limit must be positive")
+	}
+	if limit <= 0 || limit > maxLimit {
+		return Scope{}, fmt.Errorf("limit must be between 1 and %d", maxLimit)
+	}
+	return Scope{AppCode: appCode, Limit: limit}, nil
+}
+
+// Run calls processOne at most Scope.Limit times. processOne must use the
+// owner's ordinary indexed claim path with an effective batch size of one;
+// accepting a larger count here would let an implementation silently strand
+// already-claimed tail rows after this process reaches its advertised limit.
+func Run(ctx context.Context, scope Scope, processOne func(context.Context) (int, error)) (Result, error) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if strings.TrimSpace(scope.AppCode) == "" || scope.Limit <= 0 {
+		return Result{}, errors.New("valid replay scope is required")
+	}
+	if processOne == nil {
+		return Result{}, errors.New("processOne is required")
+	}
+
+	result := Result{}
+	for result.Processed < scope.Limit {
+		if err := ctx.Err(); err != nil {
+			return result, err
+		}
+		processed, err := processOne(ctx)
+		if processed < 0 || processed > 1 {
+			return result, fmt.Errorf("single-record replay processed %d rows", processed)
+		}
+		result.Processed += processed
+		if err != nil {
+			return result, err
+		}
+		if processed == 0 {
+			result.Drained = true
+			return result, nil
+		}
+	}
+	return result, nil
+}

pkg/outboxreplay/replay_test.go

@@ -0,0 +1,81 @@
+package outboxreplay
+
+import (
+	"context"
+	"errors"
+	"testing"
+)
+
+func TestNewScopeNormalizesAndBoundsTenantReplay(t *testing.T) {
+	scope, err := NewScope(" HUWAA ", 3, 10)
+	if err != nil {
+		t.Fatalf("NewScope failed: %v", err)
+	}
+	if scope.AppCode != "huwaa" || scope.Limit != 3 {
+		t.Fatalf("scope mismatch: %+v", scope)
+	}
+	for _, test := range []struct {
+		name  string
+		app   string
+		limit int
+		max   int
+	}{
+		{name: "missing app", limit: 1, max: 10},
+		{name: "zero limit", app: "huwaa", max: 10},
+		{name: "above maximum", app: "huwaa", limit: 11, max: 10},
+		{name: "invalid maximum", app: "huwaa", limit: 1},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			if _, err := NewScope(test.app, test.limit, test.max); err == nil {
+				t.Fatal("expected validation error")
+			}
+		})
+	}
+}
+
+func TestRunStopsAtProcessWideLimit(t *testing.T) {
+	calls := 0
+	result, err := Run(context.Background(), Scope{AppCode: "huwaa", Limit: 3}, func(context.Context) (int, error) {
+		calls++
+		return 1, nil
+	})
+	if err != nil {
+		t.Fatalf("Run failed: %v", err)
+	}
+	if calls != 3 || result.Processed != 3 || result.Drained {
+		t.Fatalf("hard-limit result mismatch: calls=%d result=%+v", calls, result)
+	}
+}
+
+func TestRunStopsWhenPartitionDrains(t *testing.T) {
+	calls := 0
+	result, err := Run(context.Background(), Scope{AppCode: "huwaa", Limit: 5}, func(context.Context) (int, error) {
+		calls++
+		if calls == 3 {
+			return 0, nil
+		}
+		return 1, nil
+	})
+	if err != nil {
+		t.Fatalf("Run failed: %v", err)
+	}
+	if calls != 3 || result.Processed != 2 || !result.Drained {
+		t.Fatalf("drained result mismatch: calls=%d result=%+v", calls, result)
+	}
+}
+
+func TestRunCountsClaimedFailureAndRejectsBatchClaims(t *testing.T) {
+	publishErr := errors.New("publish failed")
+	result, err := Run(context.Background(), Scope{AppCode: "huwaa", Limit: 5}, func(context.Context) (int, error) {
+		return 1, publishErr
+	})
+	if !errors.Is(err, publishErr) || result.Processed != 1 {
+		t.Fatalf("failed row result mismatch: result=%+v err=%v", result, err)
+	}
+
+	if _, err := Run(context.Background(), Scope{AppCode: "huwaa", Limit: 5}, func(context.Context) (int, error) {
+		return 2, nil
+	}); err == nil {
+		t.Fatal("expected multi-row claim rejection")
+	}
+}

services/game-service/Dockerfile

@@ -15,13 +15,14 @@ COPY . .
 RUN --mount=type=cache,id=hyapp-go-mod,target=/go/pkg/mod \
     --mount=type=cache,id=hyapp-go-build,target=/root/.cache/go-build \
     mkdir -p /out && \
-    GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/game-service/cmd/server ./cmd/grpc-health-probe
+    GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/game-service/cmd/server ./services/game-service/cmd/replay-outbox ./cmd/grpc-health-probe
 
 FROM alpine:3.20
 ENV TZ=UTC
 WORKDIR /app
 RUN apk add --no-cache ca-certificates && adduser -D -g '' appuser
 COPY --from=builder /out/server /app/server
+COPY --from=builder /out/replay-outbox /app/replay-outbox
 COPY --from=builder /out/grpc-health-probe /app/grpc-health-probe
 COPY services/game-service/configs/config.docker.yaml /app/config.yaml
 USER appuser

services/game-service/cmd/replay-outbox/main.go

@@ -0,0 +1,59 @@
+// Command replay-outbox publishes a hard-bounded game owner-outbox batch for
+// one App without starting game APIs or downstream gRPC connections.
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"hyapp/pkg/outboxreplay"
+	"hyapp/services/game-service/internal/app"
+	"hyapp/services/game-service/internal/config"
+)
+
+func main() {
+	configPath := flag.String("config", "services/game-service/configs/config.yaml", "path to game-service config")
+	appCode := flag.String("app-code", "", "exact App partition to replay")
+	limit := flag.Int("limit", 0, "process-wide maximum rows, 1..500")
+	execute := flag.Bool("execute", false, "publish the bounded batch; default is a no-write dry-run")
+	confirmation := flag.String("confirm", "", "required execution confirmation")
+	timeout := flag.Duration("timeout", 2*time.Minute, "total command timeout, at most 30m")
+	flag.Parse()
+
+	scope, err := outboxreplay.NewScope(*appCode, *limit, 500)
+	if err != nil {
+		fatal(err)
+	}
+	if *timeout <= 0 || *timeout > 30*time.Minute {
+		fatal(errors.New("timeout must be within (0,30m]"))
+	}
+	if *execute && strings.TrimSpace(*confirmation) != outboxreplay.ExecuteConfirmation {
+		fatal(fmt.Errorf("--execute requires --confirm=%s", outboxreplay.ExecuteConfirmation))
+	}
+	cfg, err := config.Load(strings.TrimSpace(*configPath))
+	if err != nil {
+		fatal(err)
+	}
+	if !*execute {
+		fmt.Printf("mode=dry-run app_code=%s limit=%d no_writes=true\n", scope.AppCode, scope.Limit)
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), *timeout)
+	defer cancel()
+	result, err := app.ReplayGameOutboxOnce(ctx, cfg, scope)
+	if err != nil {
+		fatal(fmt.Errorf("game replay stopped after %d rows: %w", result.Processed, err))
+	}
+	fmt.Printf("mode=execute app_code=%s limit=%d processed=%d drained=%t\n", scope.AppCode, scope.Limit, result.Processed, result.Drained)
+}
+
+func fatal(err error) {
+	fmt.Fprintln(os.Stderr, err)
+	os.Exit(1)
+}

services/game-service/internal/app/outbox_replay.go

@@ -0,0 +1,63 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"hyapp/pkg/outboxpartition"
+	"hyapp/pkg/outboxreplay"
+	"hyapp/pkg/rocketmqx"
+	servicemq "hyapp/pkg/servicekit/mq"
+	"hyapp/services/game-service/internal/config"
+	mysqlstorage "hyapp/services/game-service/internal/storage/mysql"
+)
+
+// ReplayGameOutboxOnce runs the ordinary game fact publisher for one App and a
+// process-wide hard limit. It opens no gRPC listener and no downstream service
+// connection, so an empty partition is a read-only no-op apart from MQ producer
+// registration.
+func ReplayGameOutboxOnce(ctx context.Context, cfg config.Config, scope outboxreplay.Scope) (outboxreplay.Result, error) {
+	validatedScope, err := outboxreplay.NewScope(scope.AppCode, scope.Limit, 500)
+	if err != nil {
+		return outboxreplay.Result{}, err
+	}
+	if !cfg.RocketMQ.GameOutbox.Enabled {
+		return outboxreplay.Result{}, errors.New("game RocketMQ outbox is not enabled")
+	}
+	repository, err := mysqlstorage.Open(ctx, cfg.MySQLDSN)
+	if err != nil {
+		return outboxreplay.Result{}, err
+	}
+	defer func() { _ = repository.Close() }()
+	producer, err := rocketmqx.NewProducer(rocketMQProducerConfig(cfg.RocketMQ))
+	if err != nil {
+		return outboxreplay.Result{}, err
+	}
+	if err := servicemq.StartProducers([]*rocketmqx.Producer{producer}); err != nil {
+		return outboxreplay.Result{}, err
+	}
+	defer servicemq.ShutdownProducers([]*rocketmqx.Producer{producer})
+
+	replayConfig := cfg
+	replayConfig.OutboxWorker.BatchSize = 1
+	replayConfig.OutboxWorker.AppCodes = []string{validatedScope.AppCode}
+	replayCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	replayApp := &App{
+		repo:             repository,
+		outboxProducer:   producer,
+		outboxPartitions: outboxpartition.New(replayConfig.OutboxWorker.AppCodes, outboxpartition.DefaultDiscoveryRefreshInterval),
+		outboxCtx:        replayCtx,
+		outboxCancel:     cancel,
+		cfg:              replayConfig,
+	}
+	workerID := fmt.Sprintf("ops-game-outbox-%s-%d", validatedScope.AppCode, time.Now().UTC().UnixMilli())
+	return outboxreplay.Run(ctx, validatedScope, func(context.Context) (int, error) {
+		// processOutboxBatch derives the same worker prefix as the live service;
+		// attach the unique run ID to NodeID so lock ownership remains auditable.
+		replayApp.cfg.NodeID = workerID
+		return replayApp.processOutboxBatch()
+	})
+}

services/user-service/Dockerfile

@@ -19,7 +19,7 @@ COPY . .
 RUN --mount=type=cache,id=hyapp-go-mod,target=/go/pkg/mod \
     --mount=type=cache,id=hyapp-go-build,target=/root/.cache/go-build \
     mkdir -p /out && \
-    GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/user-service/cmd/server ./cmd/grpc-health-probe
+    GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/user-service/cmd/server ./services/user-service/cmd/replay-outbox ./cmd/grpc-health-probe
 
 FROM alpine:3.20
 
@@ -30,6 +30,7 @@ WORKDIR /app
 RUN apk add --no-cache ca-certificates && adduser -D -g '' appuser
 
 COPY --from=builder /out/server /app/server
+COPY --from=builder /out/replay-outbox /app/replay-outbox
 COPY --from=builder /out/grpc-health-probe /app/grpc-health-probe
 COPY services/user-service/configs/config.docker.yaml /app/config.yaml
 

services/user-service/cmd/replay-outbox/main.go

@@ -0,0 +1,59 @@
+// Command replay-outbox publishes a hard-bounded user owner-outbox batch for
+// one App without starting the user-service API or unrelated MQ consumers.
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"hyapp/pkg/outboxreplay"
+	"hyapp/services/user-service/internal/app"
+	"hyapp/services/user-service/internal/config"
+)
+
+func main() {
+	configPath := flag.String("config", "services/user-service/configs/config.yaml", "path to user-service config")
+	appCode := flag.String("app-code", "", "exact App partition to replay")
+	limit := flag.Int("limit", 0, "process-wide maximum rows, 1..500")
+	execute := flag.Bool("execute", false, "publish the bounded batch; default is a no-write dry-run")
+	confirmation := flag.String("confirm", "", "required execution confirmation")
+	timeout := flag.Duration("timeout", 2*time.Minute, "total command timeout, at most 30m")
+	flag.Parse()
+
+	scope, err := outboxreplay.NewScope(*appCode, *limit, 500)
+	if err != nil {
+		fatal(err)
+	}
+	if *timeout <= 0 || *timeout > 30*time.Minute {
+		fatal(errors.New("timeout must be within (0,30m]"))
+	}
+	if *execute && strings.TrimSpace(*confirmation) != outboxreplay.ExecuteConfirmation {
+		fatal(fmt.Errorf("--execute requires --confirm=%s", outboxreplay.ExecuteConfirmation))
+	}
+	cfg, err := config.Load(strings.TrimSpace(*configPath))
+	if err != nil {
+		fatal(err)
+	}
+	if !*execute {
+		fmt.Printf("mode=dry-run app_code=%s limit=%d no_writes=true\n", scope.AppCode, scope.Limit)
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), *timeout)
+	defer cancel()
+	result, err := app.ReplayUserOutboxOnce(ctx, cfg, scope)
+	if err != nil {
+		fatal(fmt.Errorf("user replay stopped after %d rows: %w", result.Processed, err))
+	}
+	fmt.Printf("mode=execute app_code=%s limit=%d processed=%d drained=%t\n", scope.AppCode, scope.Limit, result.Processed, result.Drained)
+}
+
+func fatal(err error) {
+	fmt.Fprintln(os.Stderr, err)
+	os.Exit(1)
+}

services/user-service/internal/app/outbox_replay.go

@@ -0,0 +1,60 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"hyapp/pkg/outboxpartition"
+	"hyapp/pkg/outboxreplay"
+	"hyapp/pkg/rocketmqx"
+	serviceapp "hyapp/pkg/servicekit/app"
+	servicemq "hyapp/pkg/servicekit/mq"
+	"hyapp/services/user-service/internal/config"
+	mysqlstorage "hyapp/services/user-service/internal/storage/mysql"
+)
+
+// ReplayUserOutboxOnce processes a process-wide bounded App batch without
+// starting user-service listeners or its unrelated consumers. Claiming one row
+// per iteration keeps the existing 30-second row lease attached only to the
+// message currently being published.
+func ReplayUserOutboxOnce(ctx context.Context, cfg config.Config, scope outboxreplay.Scope) (outboxreplay.Result, error) {
+	validatedScope, err := outboxreplay.NewScope(scope.AppCode, scope.Limit, 500)
+	if err != nil {
+		return outboxreplay.Result{}, err
+	}
+	if !cfg.RocketMQ.UserOutbox.Enabled {
+		return outboxreplay.Result{}, errors.New("user RocketMQ outbox is not enabled")
+	}
+	repository, err := mysqlstorage.Open(ctx, cfg.MySQLDSN)
+	if err != nil {
+		return outboxreplay.Result{}, err
+	}
+	defer func() { _ = repository.Close() }()
+	producer, err := rocketmqx.NewProducer(userOutboxProducerConfig(cfg.RocketMQ))
+	if err != nil {
+		return outboxreplay.Result{}, err
+	}
+	if err := servicemq.StartProducers([]*rocketmqx.Producer{producer}); err != nil {
+		return outboxreplay.Result{}, err
+	}
+	defer servicemq.ShutdownProducers([]*rocketmqx.Producer{producer})
+
+	replayConfig := cfg
+	replayConfig.OutboxWorker.BatchSize = 1
+	replayConfig.OutboxWorker.AppCodes = []string{validatedScope.AppCode}
+	workers := serviceapp.NewBackground(ctx)
+	defer workers.StopAndWait()
+	replayApp := &App{
+		mysqlRepo:            repository,
+		userOutboxProducer:   producer,
+		userOutboxPartitions: outboxpartition.New(replayConfig.OutboxWorker.AppCodes, outboxpartition.DefaultDiscoveryRefreshInterval),
+		cfg:                  replayConfig,
+		workers:              workers,
+	}
+	workerID := fmt.Sprintf("ops-user-outbox-%s-%d", validatedScope.AppCode, time.Now().UTC().UnixMilli())
+	return outboxreplay.Run(ctx, validatedScope, func(context.Context) (int, error) {
+		return replayApp.processUserOutboxBatch(workerID)
+	})
+}

services/wallet-service/Dockerfile

@@ -19,7 +19,7 @@ COPY . .
 RUN --mount=type=cache,id=hyapp-go-mod,target=/go/pkg/mod \
     --mount=type=cache,id=hyapp-go-build,target=/root/.cache/go-build \
     mkdir -p /out && \
-    GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/wallet-service/cmd/server ./cmd/grpc-health-probe
+    GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/wallet-service/cmd/server ./services/wallet-service/cmd/replay-outbox ./cmd/grpc-health-probe
 
 FROM alpine:3.20
 
@@ -30,6 +30,7 @@ WORKDIR /app
 RUN apk add --no-cache ca-certificates && adduser -D -g '' appuser
 
 COPY --from=builder /out/server /app/server
+COPY --from=builder /out/replay-outbox /app/replay-outbox
 COPY --from=builder /out/grpc-health-probe /app/grpc-health-probe
 COPY services/wallet-service/configs/config.docker.yaml /app/config.yaml
 

services/wallet-service/cmd/replay-outbox/main.go

@@ -0,0 +1,79 @@
+// Command replay-outbox publishes one explicitly bounded wallet owner-outbox
+// batch. It is intentionally separate from the long-running server worker:
+// batch-size there controls one poll, while --limit here caps the whole process.
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"hyapp/pkg/outboxreplay"
+	"hyapp/services/wallet-service/internal/app"
+	"hyapp/services/wallet-service/internal/config"
+)
+
+type repeatedFlag []string
+
+func (values *repeatedFlag) String() string { return strings.Join(*values, ",") }
+
+func (values *repeatedFlag) Set(value string) error {
+	value = strings.TrimSpace(value)
+	if value == "" {
+		return errors.New("event-type cannot be empty")
+	}
+	*values = append(*values, value)
+	return nil
+}
+
+func main() {
+	configPath := flag.String("config", "services/wallet-service/configs/config.yaml", "path to wallet-service config")
+	appCode := flag.String("app-code", "", "exact App partition to replay")
+	limit := flag.Int("limit", 0, "process-wide maximum rows, 1..500")
+	execute := flag.Bool("execute", false, "publish the bounded batch; default is a no-write dry-run")
+	confirmation := flag.String("confirm", "", "required execution confirmation")
+	timeout := flag.Duration("timeout", 2*time.Minute, "total command timeout, at most 30m")
+	var eventTypes repeatedFlag
+	flag.Var(&eventTypes, "event-type", "exact wallet event type to include; repeat for multiple types")
+	flag.Parse()
+
+	scope, err := outboxreplay.NewScope(*appCode, *limit, 500)
+	if err != nil {
+		fatal(err)
+	}
+	if *timeout <= 0 || *timeout > 30*time.Minute {
+		fatal(errors.New("timeout must be within (0,30m]"))
+	}
+	if *execute && strings.TrimSpace(*confirmation) != outboxreplay.ExecuteConfirmation {
+		fatal(fmt.Errorf("--execute requires --confirm=%s", outboxreplay.ExecuteConfirmation))
+	}
+	cfg, err := config.Load(strings.TrimSpace(*configPath))
+	if err != nil {
+		fatal(err)
+	}
+	normalizedTypes, err := app.ValidateWalletReplayEventTypes(cfg, eventTypes)
+	if err != nil {
+		fatal(err)
+	}
+	if !*execute {
+		fmt.Printf("mode=dry-run app_code=%s limit=%d event_types=%s no_writes=true\n", scope.AppCode, scope.Limit, strings.Join(normalizedTypes, ","))
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), *timeout)
+	defer cancel()
+	result, err := app.ReplayWalletOutboxOnce(ctx, cfg, scope, normalizedTypes)
+	if err != nil {
+		fatal(fmt.Errorf("wallet replay stopped after %d rows: %w", result.Processed, err))
+	}
+	fmt.Printf("mode=execute app_code=%s limit=%d processed=%d drained=%t event_types=%s\n", scope.AppCode, scope.Limit, result.Processed, result.Drained, strings.Join(normalizedTypes, ","))
+}
+
+func fatal(err error) {
+	fmt.Fprintln(os.Stderr, err)
+	os.Exit(1)
+}

services/wallet-service/cmd/replay-outbox/main_test.go

@@ -0,0 +1,19 @@
+package main
+
+import "testing"
+
+func TestRepeatedEventTypeFlagAcceptsMultipleAndRejectsEmpty(t *testing.T) {
+	var values repeatedFlag
+	if err := values.Set(" GiftConfigChanged "); err != nil {
+		t.Fatalf("first event type failed: %v", err)
+	}
+	if err := values.Set("ResourceChanged"); err != nil {
+		t.Fatalf("second event type failed: %v", err)
+	}
+	if len(values) != 2 || values[0] != "GiftConfigChanged" || values[1] != "ResourceChanged" {
+		t.Fatalf("repeated event types mismatch: %v", values)
+	}
+	if err := values.Set("   "); err == nil {
+		t.Fatal("expected empty event type rejection")
+	}
+}

services/wallet-service/internal/app/outbox_replay.go

@@ -0,0 +1,168 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"hyapp/pkg/appcode"
+	"hyapp/pkg/outboxreplay"
+	"hyapp/pkg/rocketmqx"
+	servicemq "hyapp/pkg/servicekit/mq"
+	"hyapp/pkg/walletmq"
+	"hyapp/services/wallet-service/internal/config"
+	mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql"
+)
+
+// ReplayWalletOutboxOnce publishes at most scope.Limit claimable facts from
+// exactly one App partition, then closes both MySQL and MQ. Each iteration uses
+// the production claim/publish/mark path with batch size one, so a process
+// timeout cannot leave a claimed tail waiting for lease expiry.
+func ReplayWalletOutboxOnce(ctx context.Context, cfg config.Config, scope outboxreplay.Scope, eventTypes []string) (outboxreplay.Result, error) {
+	validatedScope, err := outboxreplay.NewScope(scope.AppCode, scope.Limit, 500)
+	if err != nil {
+		return outboxreplay.Result{}, err
+	}
+	includeEventTypes, err := ValidateWalletReplayEventTypes(cfg, eventTypes)
+	if err != nil {
+		return outboxreplay.Result{}, err
+	}
+	if !cfg.RocketMQ.WalletOutbox.Enabled {
+		return outboxreplay.Result{}, errors.New("wallet RocketMQ outbox is not enabled")
+	}
+
+	repository, err := mysqlstorage.Open(ctx, cfg.MySQLDSN)
+	if err != nil {
+		return outboxreplay.Result{}, err
+	}
+	defer func() { _ = repository.Close() }()
+	producer, err := rocketmqx.NewProducer(rocketMQProducerConfig(cfg.RocketMQ, cfg.RocketMQ.WalletOutbox.ProducerGroup))
+	if err != nil {
+		return outboxreplay.Result{}, err
+	}
+	if err := servicemq.StartProducers([]*rocketmqx.Producer{producer}); err != nil {
+		return outboxreplay.Result{}, err
+	}
+	defer servicemq.ShutdownProducers([]*rocketmqx.Producer{producer})
+
+	workerOptions := cfg.OutboxWorker
+	workerOptions.BatchSize = 1
+	workerOptions.Concurrency = 1
+	replayApp := &App{
+		mysqlRepo:          repository,
+		walletEventTagMode: cfg.RocketMQ.WalletEventTagMode,
+		walletOutboxTopic:  cfg.RocketMQ.WalletOutbox.Topic,
+		outboxWorkerCfg:    workerOptions,
+	}
+	workerID := fmt.Sprintf("ops-wallet-outbox-%s-%d", validatedScope.AppCode, time.Now().UTC().UnixMilli())
+	return outboxreplay.Run(ctx, validatedScope, func(runCtx context.Context) (int, error) {
+		return replayApp.replayOneWalletOutboxRecord(runCtx, workerID, workerOptions, producer, cfg.RocketMQ.WalletOutbox.Topic, validatedScope.AppCode, includeEventTypes)
+	})
+}
+
+func (a *App) replayOneWalletOutboxRecord(ctx context.Context, workerID string, options config.OutboxWorkerConfig, producer *rocketmqx.Producer, topic string, targetAppCode string, includeEventTypes []string) (int, error) {
+	if options.PublishTimeout <= 0 {
+		options.PublishTimeout = 3 * time.Second
+	}
+	claimCtx := appcode.WithContext(ctx, targetAppCode)
+	records, err := a.mysqlRepo.ClaimPendingWalletOutboxFiltered(
+		claimCtx,
+		workerID,
+		1,
+		time.Now().UTC().Add(walletOutboxSingleRecordLease(options.PublishTimeout)).UnixMilli(),
+		mysqlstorage.WalletOutboxClaimFilter{IncludeEventTypes: includeEventTypes},
+	)
+	if err != nil || len(records) == 0 {
+		return 0, err
+	}
+	record := records[0]
+	if record.RetryCount >= options.MaxRetryCount {
+		markCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
+		markErr := a.mysqlRepo.MarkWalletOutboxDead(markCtx, record.EventID, deadWalletOutboxReason(record))
+		cancel()
+		if markErr != nil {
+			return 1, markErr
+		}
+		return 1, fmt.Errorf("wallet outbox event %s reached max retry count and was marked dead", record.EventID)
+	}
+
+	publishCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
+	publishErr := a.publishWalletOutboxRecord(publishCtx, producer, topic, record)
+	cancel()
+	markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
+	defer markCancel()
+	if publishErr != nil {
+		nextRetryAtMS := time.Now().UTC().Add(walletOutboxBackoff(record.RetryCount+1, options)).UnixMilli()
+		if markErr := a.mysqlRepo.MarkWalletOutboxRetryable(markCtx, record.EventID, publishErr.Error(), nextRetryAtMS); markErr != nil {
+			return 1, errors.Join(publishErr, markErr)
+		}
+		// The row was safely released for a later retry, but this explicit run must
+		// fail instead of reporting a merely claimed row as a delivered replay.
+		return 1, publishErr
+	}
+	if err := a.mysqlRepo.MarkWalletOutboxDelivered(markCtx, record.EventID); err != nil {
+		return 1, err
+	}
+	return 1, nil
+}
+
+// NormalizeWalletReplayEventTypes validates the exact MQ tags before an
+// operator dry-run is promoted to an executing run.
+func NormalizeWalletReplayEventTypes(values []string) ([]string, error) {
+	if len(values) > 32 {
+		return nil, errors.New("wallet event-type count cannot exceed 32")
+	}
+	normalized := make([]string, 0, len(values))
+	seen := make(map[string]struct{}, len(values))
+	for _, value := range values {
+		eventType, err := walletmq.EventTypeTag(strings.TrimSpace(value))
+		if err != nil {
+			return nil, err
+		}
+		if _, exists := seen[eventType]; exists {
+			continue
+		}
+		seen[eventType] = struct{}{}
+		normalized = append(normalized, eventType)
+	}
+	return normalized, nil
+}
+
+// ValidateWalletReplayEventTypes applies the lane boundary in both dry-run and
+// execute modes. The normal-topic repair must reject realtime event types even
+// while that worker or topic is temporarily disabled; pausing a lane does not
+// transfer ownership of its durable rows to another topic.
+func ValidateWalletReplayEventTypes(cfg config.Config, values []string) ([]string, error) {
+	normalized, err := NormalizeWalletReplayEventTypes(values)
+	if err != nil {
+		return nil, err
+	}
+	if len(normalized) == 0 {
+		// Wallet fanout has both high-volume balance notices and low-volume config
+		// facts. Requiring an explicit set prevents a small operational command
+		// from accidentally mixing stages merely because their timestamps overlap.
+		return nil, errors.New("at least one wallet event-type is required")
+	}
+	if err := rejectRealtimeReplayEventTypes(normalized, cfg.RealtimeOutboxWorker.EventTypes); err != nil {
+		return nil, err
+	}
+	return normalized, nil
+}
+
+func rejectRealtimeReplayEventTypes(included []string, realtime []string) error {
+	realtimeSet := make(map[string]struct{}, len(realtime))
+	for _, eventType := range realtime {
+		eventType = strings.TrimSpace(eventType)
+		if eventType != "" {
+			realtimeSet[eventType] = struct{}{}
+		}
+	}
+	for _, eventType := range included {
+		if _, exists := realtimeSet[eventType]; exists {
+			return fmt.Errorf("wallet event-type %s belongs to the realtime outbox lane", eventType)
+		}
+	}
+	return nil
+}

services/wallet-service/internal/app/outbox_replay_test.go

@@ -0,0 +1,43 @@
+package app
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"hyapp/pkg/outboxreplay"
+	"hyapp/services/wallet-service/internal/config"
+)
+
+func TestNormalizeWalletReplayEventTypes(t *testing.T) {
+	eventTypes, err := NormalizeWalletReplayEventTypes([]string{" GiftConfigChanged ", "WalletBalanceChanged", "GiftConfigChanged"})
+	if err != nil {
+		t.Fatalf("NormalizeWalletReplayEventTypes failed: %v", err)
+	}
+	if len(eventTypes) != 2 || eventTypes[0] != "GiftConfigChanged" || eventTypes[1] != "WalletBalanceChanged" {
+		t.Fatalf("normalized event types mismatch: %v", eventTypes)
+	}
+	if _, err := NormalizeWalletReplayEventTypes([]string{"bad || tag"}); err == nil {
+		t.Fatal("expected invalid RocketMQ tag rejection")
+	}
+}
+
+func TestRejectRealtimeReplayEventTypes(t *testing.T) {
+	if err := rejectRealtimeReplayEventTypes([]string{"GiftConfigChanged"}, []string{"WalletRedPacketCreated"}); err != nil {
+		t.Fatalf("normal event rejected: %v", err)
+	}
+	if err := rejectRealtimeReplayEventTypes([]string{"WalletRedPacketCreated"}, []string{"WalletRedPacketCreated"}); err == nil {
+		t.Fatal("expected realtime lane rejection")
+	}
+}
+
+func TestWalletReplayRejectsRealtimeTypeWhenLaneIsDisabled(t *testing.T) {
+	cfg := config.Default()
+	cfg.RealtimeOutboxWorker.Enabled = false
+	cfg.RocketMQ.RealtimeOutbox.Enabled = false
+	cfg.RealtimeOutboxWorker.EventTypes = []string{"WalletRedPacketCreated"}
+	_, err := ReplayWalletOutboxOnce(context.Background(), cfg, outboxreplay.Scope{AppCode: "huwaa", Limit: 1}, []string{"WalletRedPacketCreated"})
+	if err == nil || !strings.Contains(err.Error(), "realtime outbox lane") {
+		t.Fatalf("disabled realtime lane must retain event ownership, err=%v", err)
+	}
+}