hyapp-server/tools/migrate-room-visible-region/main.go

package main

import (
	"context"
	"crypto/sha1"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"strings"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	roomv1 "hyapp.local/api/proto/room/v1"
	"hyapp/pkg/appcode"
)

type roomCandidate struct {
	RoomID          string `json:"room_id"`
	RoomShortID     string `json:"room_short_id"`
	OwnerUserID     int64  `json:"owner_user_id"`
	VisibleRegionID int64  `json:"visible_region_id"`
	OwnerRegionID   int64  `json:"owner_region_id"`
}

func main() {
	var roomDSN string
	var userDSN string
	var roomGRPCAddr string
	var app string
	var shortID string
	var limit int
	var apply bool
	var adminID uint64
	flag.StringVar(&roomDSN, "room_mysql_dsn", "", "room-service MySQL DSN")
	flag.StringVar(&userDSN, "user_mysql_dsn", "", "user-service MySQL DSN")
	flag.StringVar(&roomGRPCAddr, "room_grpc_addr", "", "room-service gRPC address; required with -apply")
	flag.StringVar(&app, "app_code", appcode.Default, "app_code scope")
	flag.StringVar(&shortID, "room_short_id", "", "optional room_short_id filter")
	flag.IntVar(&limit, "limit", 500, "maximum rooms to scan")
	flag.BoolVar(&apply, "apply", false, "call room-service AdminUpdateRoom; omitted means dry-run")
	flag.Uint64Var(&adminID, "admin_id", 0, "admin id written to AdminUpdateRoom")
	flag.Parse()

	if strings.TrimSpace(roomDSN) == "" || strings.TrimSpace(userDSN) == "" {
		log.Fatal("room_mysql_dsn and user_mysql_dsn are required")
	}
	if apply && strings.TrimSpace(roomGRPCAddr) == "" {
		log.Fatal("room_grpc_addr is required when -apply is set")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()

	roomDB, err := sql.Open("mysql", roomDSN)
	if err != nil {
		log.Fatalf("open room mysql failed: %v", err)
	}
	defer roomDB.Close()
	userDB, err := sql.Open("mysql", userDSN)
	if err != nil {
		log.Fatalf("open user mysql failed: %v", err)
	}
	defer userDB.Close()

	mismatches, err := findMismatches(ctx, roomDB, userDB, appcode.Normalize(app), shortID, limit)
	if err != nil {
		log.Fatalf("find mismatches failed: %v", err)
	}
	encoder := json.NewEncoder(log.Writer())
	for _, item := range mismatches {
		if err := encoder.Encode(item); err != nil {
			log.Fatalf("write dry-run row failed: %v", err)
		}
	}
	log.Printf("mismatch_count=%d apply=%t", len(mismatches), apply)
	if !apply || len(mismatches) == 0 {
		return
	}

	conn, err := grpc.DialContext(ctx, roomGRPCAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatalf("dial room-service failed: %v", err)
	}
	defer conn.Close()
	client := roomv1.NewRoomCommandServiceClient(conn)
	for _, item := range mismatches {
		// 迁移必须走 Room Cell 命令链路，让 command log、snapshot、rooms 和 room_list_entries 在同一事务内收敛。
		if _, err := client.AdminUpdateRoom(ctx, &roomv1.AdminUpdateRoomRequest{
			Meta: &roomv1.RequestMeta{
				AppCode:   appcode.Normalize(app),
				RequestId: "room-visible-region-migration",
				CommandId: migrationCommandID(item),
				RoomId:    item.RoomID,
				SentAtMs:  time.Now().UTC().UnixMilli(),
			},
			VisibleRegionId: &item.OwnerRegionID,
			AdminId:         adminID,
			AdminName:       "room-visible-region-migration",
		}); err != nil {
			log.Fatalf("migrate room_id=%s short_id=%s failed: %v", item.RoomID, item.RoomShortID, err)
		}
	}
	log.Printf("migrated_count=%d", len(mismatches))
}

func findMismatches(ctx context.Context, roomDB *sql.DB, userDB *sql.DB, app string, shortID string, limit int) ([]roomCandidate, error) {
	if limit <= 0 {
		limit = 500
	}
	where := "WHERE app_code = ? AND status <> 'deleted'"
	args := []any{app}
	if strings.TrimSpace(shortID) != "" {
		where += " AND room_short_id = ?"
		args = append(args, strings.TrimSpace(shortID))
	}
	args = append(args, limit)
	rows, err := roomDB.QueryContext(ctx, `
		SELECT room_id, room_short_id, owner_user_id, COALESCE(visible_region_id, 0)
		FROM rooms
		`+where+`
		ORDER BY room_id ASC
		LIMIT ?
	`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	mismatches := make([]roomCandidate, 0)
	for rows.Next() {
		var item roomCandidate
		if err := rows.Scan(&item.RoomID, &item.RoomShortID, &item.OwnerUserID, &item.VisibleRegionID); err != nil {
			return nil, err
		}
		var ownerRegionID int64
		err := userDB.QueryRowContext(ctx, `
			SELECT COALESCE(region_id, 0)
			FROM users
			WHERE app_code = ? AND user_id = ? AND COALESCE(region_id, 0) > 0
		`, app, item.OwnerUserID).Scan(&ownerRegionID)
		if err == sql.ErrNoRows {
			continue
		}
		if err != nil {
			return nil, err
		}
		if ownerRegionID == item.VisibleRegionID {
			continue
		}
		item.OwnerRegionID = ownerRegionID
		mismatches = append(mismatches, item)
	}
	return mismatches, rows.Err()
}

func migrationCommandID(item roomCandidate) string {
	sum := sha1.Sum([]byte(fmt.Sprintf("%s:%d:%d", item.RoomID, item.VisibleRegionID, item.OwnerRegionID)))
	return "room-region-migrate:" + hex.EncodeToString(sum[:])
}