diff --git a/api/proto/wallet/v1/wallet.pb.go b/api/proto/wallet/v1/wallet.pb.go index c712ebc6..9016f5fa 100644 --- a/api/proto/wallet/v1/wallet.pb.go +++ b/api/proto/wallet/v1/wallet.pb.go @@ -8991,6 +8991,9 @@ type VipRewardItem struct { Name string `protobuf:"bytes,4,opt,name=name,proto3" json:"name,omitempty"` Quantity int64 `protobuf:"varint,5,opt,name=quantity,proto3" json:"quantity,omitempty"` ExpiresAtMs int64 `protobuf:"varint,6,opt,name=expires_at_ms,json=expiresAtMs,proto3" json:"expires_at_ms,omitempty"` + AssetUrl string `protobuf:"bytes,7,opt,name=asset_url,json=assetUrl,proto3" json:"asset_url,omitempty"` + PreviewUrl string `protobuf:"bytes,8,opt,name=preview_url,json=previewUrl,proto3" json:"preview_url,omitempty"` + AnimationUrl string `protobuf:"bytes,9,opt,name=animation_url,json=animationUrl,proto3" json:"animation_url,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -9067,6 +9070,27 @@ func (x *VipRewardItem) GetExpiresAtMs() int64 { return 0 } +func (x *VipRewardItem) GetAssetUrl() string { + if x != nil { + return x.AssetUrl + } + return "" +} + +func (x *VipRewardItem) GetPreviewUrl() string { + if x != nil { + return x.PreviewUrl + } + return "" +} + +func (x *VipRewardItem) GetAnimationUrl() string { + if x != nil { + return x.AnimationUrl + } + return "" +} + type VipLevel struct { state protoimpl.MessageState `protogen:"open.v1"` Level int32 `protobuf:"varint,1,opt,name=level,proto3" json:"level,omitempty"` @@ -13451,7 +13475,7 @@ const file_proto_wallet_v1_wallet_proto_rawDesc = "" + "\n" + "withdrawal\x18\x01 \x01(\v2\".hyapp.wallet.v1.WithdrawalRequestR\n" + "withdrawal\x127\n" + - "\abalance\x18\x02 \x01(\v2\x1d.hyapp.wallet.v1.AssetBalanceR\abalance\"\xce\x01\n" + + "\abalance\x18\x02 \x01(\v2\x1d.hyapp.wallet.v1.AssetBalanceR\abalance\"\xb1\x02\n" + "\rVipRewardItem\x12\x1f\n" + "\vresource_id\x18\x01 \x01(\x03R\n" + "resourceId\x12#\n" + @@ -13459,7 +13483,11 @@ const file_proto_wallet_v1_wallet_proto_rawDesc = "" + "\rresource_type\x18\x03 \x01(\tR\fresourceType\x12\x12\n" + "\x04name\x18\x04 \x01(\tR\x04name\x12\x1a\n" + "\bquantity\x18\x05 \x01(\x03R\bquantity\x12\"\n" + - "\rexpires_at_ms\x18\x06 \x01(\x03R\vexpiresAtMs\"\xfc\x04\n" + + "\rexpires_at_ms\x18\x06 \x01(\x03R\vexpiresAtMs\x12\x1b\n" + + "\tasset_url\x18\a \x01(\tR\bassetUrl\x12\x1f\n" + + "\vpreview_url\x18\b \x01(\tR\n" + + "previewUrl\x12#\n" + + "\ranimation_url\x18\t \x01(\tR\fanimationUrl\"\xfc\x04\n" + "\bVipLevel\x12\x14\n" + "\x05level\x18\x01 \x01(\x05R\x05level\x12\x12\n" + "\x04name\x18\x02 \x01(\tR\x04name\x12\x16\n" + diff --git a/api/proto/wallet/v1/wallet.proto b/api/proto/wallet/v1/wallet.proto index a06d6ba5..a559048d 100644 --- a/api/proto/wallet/v1/wallet.proto +++ b/api/proto/wallet/v1/wallet.proto @@ -1007,6 +1007,9 @@ message VipRewardItem { string name = 4; int64 quantity = 5; int64 expires_at_ms = 6; + string asset_url = 7; + string preview_url = 8; + string animation_url = 9; } message VipLevel { diff --git a/run/im-c2c-broadcast-20260602-061847.jsonl b/run/im-c2c-broadcast-20260602-061847.jsonl new file mode 100644 index 00000000..32bcf75e --- /dev/null +++ b/run/im-c2c-broadcast-20260602-061847.jsonl @@ -0,0 +1,92 @@ +{"at_ms":1780381127156,"mode":"dry-run","to_account":"424242","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"12345678","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"880585934","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"880585935","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"881586014","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"881586015","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"882586332","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"882586333","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990000001","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990000002","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990686001","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990686002","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990686003","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990686004","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990686005","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990686006","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990686007","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990686008","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990686009","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990686010","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990687001","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990687002","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990687003","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990687004","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990687005","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990687006","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990687007","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990687008","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990687009","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990687010","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990688001","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990688002","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990688003","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990688004","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990688005","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990688006","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990688007","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990688008","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990688009","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990688010","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"900000971589","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"900000971590","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"900000971617","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"900000971618","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"900000971718","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"900000971719","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"900503651195","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"900503651196","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"900503651197","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"900503651198","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"910004935471","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"910004935472","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"910004935473","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"910004935474","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"910004936181","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"910004936182","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"910004936183","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"910004936184","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"930000000001","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"930000000002","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"975824912277","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"975824912278","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"975825864464","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"975825864465","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"975827047904","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"975827047905","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"975828064123","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"975828064124","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990778061358","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"990778062287","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"991205120003","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"1778070991803","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"1778070991804","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"1778071196761","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"1778071196762","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"7049835865153","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"7049835865154","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"7049835865155","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"7049835865156","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"7049835865157","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"7049835865158","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"7049835865159","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"7049835865160","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"7049835865161","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"7049835865162","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"318705991371722752","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"318707093894860800","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"318707097803952128","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"318707100643495936","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"318707103256547328","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"318707105987039232","status":"dry_run"} +{"at_ms":1780381127156,"mode":"dry-run","to_account":"320078559995498496","status":"dry_run"} diff --git a/scripts/ops/grant_all_users_coin.py b/scripts/ops/grant_all_users_coin.py new file mode 100755 index 00000000..3824a4e0 --- /dev/null +++ b/scripts/ops/grant_all_users_coin.py @@ -0,0 +1,343 @@ +#!/usr/bin/env python3 +"""Grant COIN to every current production user. + +This script intentionally creates a fresh command_id for every run. Running it +twice credits users twice. If a run fails halfway, re-running it credits the +already processed users again, matching the requested non-idempotent behavior. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import shlex +import subprocess +import time +import uuid +from pathlib import Path +from typing import Any + +import pymysql +import yaml + + +ASSET_COIN = "COIN" +BIZ_TYPE_MANUAL_CREDIT = "manual_credit" +OUTBOX_STATUS_PENDING = "pending" +TRANSACTION_STATUS_SUCCEEDED = "succeeded" +MAX_INT64 = 2**63 - 1 + + +def normalize_app_code(value: str) -> str: + value = (value or "").strip().lower() + return value or "lalu" + + +def stable_hash(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def transaction_id(app_code: str, command_id: str) -> str: + return "wtx_" + stable_hash(f"{normalize_app_code(app_code)}|{command_id}") + + +def event_id(tx_id: str, event_type: str, user_id: int, asset_type: str) -> str: + return "wev_" + stable_hash(f"{tx_id}|{event_type}|{user_id}|{asset_type}") + + +def admin_credit_request_hash( + app_code: str, + target_user_id: int, + asset_type: str, + amount: int, + operator_user_id: int, + reason: str, + evidence_ref: str, +) -> str: + return stable_hash( + f"admin_credit|{normalize_app_code(app_code)}|{target_user_id}|{asset_type}|{amount}|{operator_user_id}|{reason}|{evidence_ref}" + ) + + +def load_yaml(path: str) -> dict[str, Any]: + payload = yaml.safe_load(read_config_text(path)) or {} + if not isinstance(payload, dict): + raise RuntimeError(f"config must be a YAML object: {path}") + return payload + + +def read_config_text(location: str) -> str: + location = location.strip() + if looks_like_ssh_location(location): + host, remote_path = location.split(":", 1) + return subprocess.check_output( + ["ssh", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=5", host, "cat " + shlex.quote(remote_path)], + text=True, + ) + return Path(location).expanduser().read_text(encoding="utf-8") + + +def looks_like_ssh_location(location: str) -> bool: + if location.startswith("/") or location.startswith("./") or location.startswith("../") or "://" in location: + return False + head, sep, tail = location.partition(":") + return bool(sep and tail and ("@" in head or "." in head or head.replace("-", "").isalnum())) + + +def parse_go_mysql_dsn(dsn: str) -> dict[str, Any]: + # Supports the project form: USER:PASSWORD@tcp(HOST:PORT)/DATABASE?params. + prefix, rest = dsn.split("@tcp(", 1) + user, password = prefix.split(":", 1) if ":" in prefix else (prefix, "") + addr, rest = rest.split(")/", 1) + host, port = addr.rsplit(":", 1) if ":" in addr else (addr, "3306") + database = rest.split("?", 1)[0] + return { + "host": host, + "port": int(port), + "user": user, + "password": password, + "database": database, + "charset": "utf8mb4", + "autocommit": False, + "connect_timeout": 8, + "read_timeout": 60, + "write_timeout": 60, + } + + +def connect_from_config(path: str): + cfg = load_yaml(path) + dsn = str(cfg.get("mysql_dsn") or "").strip() + if not dsn: + raise RuntimeError(f"mysql_dsn is required in {path}") + return pymysql.connect(**parse_go_mysql_dsn(dsn)) + + +def current_ms() -> int: + return int(time.time() * 1000) + + +def fetch_users(conn, app_code: str, statuses: list[str]) -> list[int]: + where = ["app_code = %s"] + args: list[Any] = [app_code] + if statuses: + where.append("status IN (" + ",".join(["%s"] * len(statuses)) + ")") + args.extend(statuses) + with conn.cursor() as cursor: + cursor.execute(f"SELECT user_id FROM users WHERE {' AND '.join(where)} ORDER BY user_id ASC", args) + return [int(row[0]) for row in cursor.fetchall()] + + +def fetch_user_status_counts(conn, app_code: str) -> dict[str, int]: + with conn.cursor() as cursor: + cursor.execute("SELECT status, COUNT(*) FROM users WHERE app_code = %s GROUP BY status ORDER BY status", (app_code,)) + return {str(status): int(count) for status, count in cursor.fetchall()} + + +def grant_one(conn, *, app_code: str, user_id: int, amount: int, operator_user_id: int, reason: str, evidence_ref: str, run_id: str) -> tuple[str, int]: + command_id = f"bulk_coin_grant:{run_id}:{user_id}" + tx_id = transaction_id(app_code, command_id) + req_hash = admin_credit_request_hash(app_code, user_id, ASSET_COIN, amount, operator_user_id, reason, evidence_ref) + now_ms = current_ms() + metadata = { + "app_code": app_code, + "target_user_id": user_id, + "asset_type": ASSET_COIN, + "amount": amount, + "operator_user_id": operator_user_id, + "reason": reason, + "evidence_ref": evidence_ref, + } + + cursor = conn.cursor() + try: + cursor.execute("START TRANSACTION") + # Positive manual credit may create a missing COIN account; version then advances exactly like wallet-service. + cursor.execute( + """ + INSERT IGNORE INTO wallet_accounts ( + app_code, user_id, asset_type, available_amount, frozen_amount, version, created_at_ms, updated_at_ms + ) VALUES (%s, %s, %s, 0, 0, 1, %s, %s) + """, + (app_code, user_id, ASSET_COIN, now_ms, now_ms), + ) + cursor.execute( + """ + SELECT available_amount, frozen_amount, version + FROM wallet_accounts + WHERE app_code = %s AND user_id = %s AND asset_type = %s + FOR UPDATE + """, + (app_code, user_id, ASSET_COIN), + ) + row = cursor.fetchone() + if row is None: + raise RuntimeError(f"wallet account creation failed for user_id={user_id}") + available, frozen, version = int(row[0]), int(row[1]), int(row[2]) + available_after = available + amount + if available_after < 0 or available_after > MAX_INT64 - frozen: + raise RuntimeError(f"wallet balance overflow for user_id={user_id}") + + cursor.execute( + """ + INSERT INTO wallet_transactions ( + app_code, transaction_id, command_id, biz_type, status, request_hash, external_ref, + metadata_json, created_at_ms, updated_at_ms + ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + """, + ( + app_code, + tx_id, + command_id, + BIZ_TYPE_MANUAL_CREDIT, + TRANSACTION_STATUS_SUCCEEDED, + req_hash, + evidence_ref, + json.dumps(metadata, separators=(",", ":"), ensure_ascii=False), + now_ms, + now_ms, + ), + ) + cursor.execute( + """ + UPDATE wallet_accounts + SET available_amount = %s, frozen_amount = %s, version = version + 1, updated_at_ms = %s + WHERE app_code = %s AND user_id = %s AND asset_type = %s AND version = %s + """, + (available_after, frozen, now_ms, app_code, user_id, ASSET_COIN, version), + ) + if cursor.rowcount != 1: + raise RuntimeError(f"wallet account version conflict for user_id={user_id}") + cursor.execute( + """ + INSERT INTO wallet_entries ( + app_code, transaction_id, user_id, asset_type, available_delta, frozen_delta, + available_after, frozen_after, counterparty_user_id, room_id, created_at_ms + ) VALUES (%s, %s, %s, %s, %s, 0, %s, %s, 0, '', %s) + """, + (app_code, tx_id, user_id, ASSET_COIN, amount, available_after, frozen, now_ms), + ) + + event_payload = { + "transaction_id": tx_id, + "command_id": command_id, + "user_id": user_id, + "asset_type": ASSET_COIN, + "available_delta": amount, + "frozen_delta": 0, + "available_after": available_after, + "frozen_after": frozen, + "balance_version": version + 1, + "metadata": metadata, + "created_at_ms": now_ms, + } + cursor.execute( + """ + INSERT INTO wallet_outbox ( + app_code, event_id, event_type, transaction_id, command_id, user_id, asset_type, + available_delta, frozen_delta, payload, status, retry_count, created_at_ms, updated_at_ms + ) VALUES (%s, %s, 'WalletBalanceChanged', %s, %s, %s, %s, %s, 0, %s, %s, 0, %s, %s) + """, + ( + app_code, + event_id(tx_id, "WalletBalanceChanged", user_id, ASSET_COIN), + tx_id, + command_id, + user_id, + ASSET_COIN, + amount, + json.dumps(event_payload, separators=(",", ":"), ensure_ascii=False), + OUTBOX_STATUS_PENDING, + now_ms, + now_ms, + ), + ) + conn.commit() + return tx_id, available_after + except Exception: + conn.rollback() + raise + finally: + cursor.close() + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Non-idempotently grant COIN to every current user.") + parser.add_argument( + "--user-config", + default="services/user-service/configs/config.yaml", + help="YAML containing user-service mysql_dsn. Supports local path or user@host:/path/config.yaml.", + ) + parser.add_argument( + "--wallet-config", + default="services/wallet-service/configs/config.yaml", + help="YAML containing wallet-service mysql_dsn. Supports local path or user@host:/path/config.yaml.", + ) + parser.add_argument("--app-code", default="lalu", help="Tenant app_code to grant.") + parser.add_argument("--amount", type=int, default=1_000_000, help="COIN amount credited to each user per execution.") + parser.add_argument("--operator-user-id", type=int, default=0, help="Required with --yes; stored in manual credit metadata.") + parser.add_argument("--reason", default="bulk coin grant to all users", help="Manual credit reason stored with every transaction.") + parser.add_argument("--evidence-ref-prefix", default="ops-bulk-coin-grant", help="Evidence reference prefix; run_id and user_id are appended.") + parser.add_argument("--user-status", action="append", default=[], help="Optional repeatable users.status filter. Omit to include all users.") + parser.add_argument("--progress-every", type=int, default=100, help="Print progress every N credited users.") + parser.add_argument("--yes", action="store_true", help="Actually write wallet credits. Omit for dry-run.") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + app_code = normalize_app_code(args.app_code) + if args.amount <= 0 or args.amount > MAX_INT64: + raise RuntimeError("--amount must be a positive int64") + if args.yes and args.operator_user_id <= 0: + raise RuntimeError("--operator-user-id is required with --yes") + if args.yes and not args.reason.strip(): + raise RuntimeError("--reason is required with --yes") + + user_conn = connect_from_config(args.user_config) + wallet_conn = connect_from_config(args.wallet_config) + try: + status_counts = fetch_user_status_counts(user_conn, app_code) + users = fetch_users(user_conn, app_code, [status.strip() for status in args.user_status if status.strip()]) + summary = { + "mode": "execute" if args.yes else "dry_run", + "app_code": app_code, + "amount_per_user": args.amount, + "target_user_count": len(users), + "total_coin_delta": len(users) * args.amount, + "user_status_counts": status_counts, + "user_status_filter": args.user_status, + } + print(json.dumps(summary, ensure_ascii=False, indent=2)) + if not args.yes: + print("dry-run only; rerun with --yes --operator-user-id to credit users") + return 0 + + run_id = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) + "_" + uuid.uuid4().hex[:12] + credited = 0 + for user_id in users: + evidence_ref = f"{args.evidence_ref_prefix}:{run_id}:{user_id}" + grant_one( + wallet_conn, + app_code=app_code, + user_id=user_id, + amount=args.amount, + operator_user_id=args.operator_user_id, + reason=args.reason.strip(), + evidence_ref=evidence_ref, + run_id=run_id, + ) + credited += 1 + if args.progress_every > 0 and (credited % args.progress_every == 0 or credited == len(users)): + print(json.dumps({"run_id": run_id, "credited": credited, "total": len(users)}, ensure_ascii=False)) + print(json.dumps({"ok": True, "run_id": run_id, "credited": credited, "amount_per_user": args.amount}, ensure_ascii=False, indent=2)) + return 0 + finally: + user_conn.close() + wallet_conn.close() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/server/admin/internal/modules/upload/handler.go b/server/admin/internal/modules/upload/handler.go index c3f27e2c..14dd4c9d 100644 --- a/server/admin/internal/modules/upload/handler.go +++ b/server/admin/internal/modules/upload/handler.go @@ -21,6 +21,7 @@ import ( const ( uploadSniffBytes = 512 + maxBatchFiles = 10 ) var ( @@ -62,6 +63,10 @@ func (h *Handler) UploadImage(c *gin.Context) { h.uploadObject(c, imageUploadPolicy) } +func (h *Handler) UploadImagesBatch(c *gin.Context) { + h.uploadObjects(c, imageUploadPolicy) +} + func (h *Handler) uploadObject(c *gin.Context, policy uploadPolicy) { if h.uploader == nil { response.ServerError(c, "文件上传未配置") @@ -109,9 +114,72 @@ func (h *Handler) uploadObject(c *gin.Context, policy uploadPolicy) { ObjectKey: objectKey, ContentType: contentType, SizeBytes: header.Size, + Filename: header.Filename, }) } +func (h *Handler) uploadObjects(c *gin.Context, policy uploadPolicy) { + if h.uploader == nil { + response.ServerError(c, "文件上传未配置") + return + } + headers, ok := parseUploadFiles(c) + if !ok { + return + } + defer func() { + if c.Request.MultipartForm != nil { + _ = c.Request.MultipartForm.RemoveAll() + } + }() + + actor := shared.ActorFromContext(c) + results := make([]uploadResult, 0, len(headers)) + for _, header := range headers { + if header.Size <= 0 { + response.BadRequest(c, "文件大小不正确") + return + } + file, err := header.Open() + if err != nil { + response.BadRequest(c, "文件参数不正确") + return + } + head, contentType, detectedContentType, sniffOK := sniffUploadContent(c, file, header) + if !sniffOK { + _ = file.Close() + return + } + if policy.requireImage { + contentType = resolveImageUploadContentType(header.Filename, contentType, detectedContentType) + } + if !validateUploadContent(policy, header.Filename, contentType) { + _ = file.Close() + response.BadRequest(c, "文件类型不支持") + return + } + + objectKey := buildUploadObjectKey(h.objectPrefix, policy, actor.UserID, header.Filename, contentType, time.Now()) + objectURL, err := h.uploader.PutObject(c.Request.Context(), objectKey, io.MultiReader(bytes.NewReader(head), file), header.Size, contentType) + _ = file.Close() + if err != nil { + response.ServerError(c, "文件上传失败") + return + } + results = append(results, uploadResult{ + URL: objectURL, + ObjectKey: objectKey, + ContentType: contentType, + SizeBytes: header.Size, + Filename: header.Filename, + }) + } + + shared.OperationLogWithResourceID(c, h.audit, "batch-upload-file", "cos_objects", "batch", "success", + fmt.Sprintf("kind=%s count=%d", policy.kind, len(results))) + response.OK(c, results) +} + func parseUploadFile(c *gin.Context) (multipart.File, *multipart.FileHeader, bool) { if err := c.Request.ParseMultipartForm(1 << 20); err != nil { response.BadRequest(c, "文件参数不正确") @@ -127,6 +195,33 @@ func parseUploadFile(c *gin.Context) (multipart.File, *multipart.FileHeader, boo return nil, nil, false } +func parseUploadFiles(c *gin.Context) ([]*multipart.FileHeader, bool) { + if err := c.Request.ParseMultipartForm(1 << 20); err != nil { + response.BadRequest(c, "文件参数不正确") + return nil, false + } + if c.Request.MultipartForm == nil || len(c.Request.MultipartForm.File) == 0 { + response.BadRequest(c, "文件参数不正确") + return nil, false + } + + headers := make([]*multipart.FileHeader, 0, maxBatchFiles) + for _, field := range []string{"files", "file", "images", "image"} { + for _, header := range c.Request.MultipartForm.File[field] { + headers = append(headers, header) + } + } + if len(headers) == 0 { + response.BadRequest(c, "文件参数不正确") + return nil, false + } + if len(headers) > maxBatchFiles { + response.BadRequest(c, "一次最多上传 10 个素材") + return nil, false + } + return headers, true +} + func sniffUploadContent(c *gin.Context, file multipart.File, header *multipart.FileHeader) ([]byte, string, string, bool) { head := make([]byte, uploadSniffBytes) n, err := io.ReadFull(file, head) diff --git a/server/admin/internal/modules/upload/handler_test.go b/server/admin/internal/modules/upload/handler_test.go index 464ea163..d3c50364 100644 --- a/server/admin/internal/modules/upload/handler_test.go +++ b/server/admin/internal/modules/upload/handler_test.go @@ -20,6 +20,7 @@ import ( type fakeUploader struct { contentType string key string + keys []string sizeBytes int64 body []byte } @@ -30,6 +31,7 @@ func (u *fakeUploader) PutObject(_ context.Context, key string, reader io.Reader return "", err } u.key = key + u.keys = append(u.keys, key) u.body = body u.sizeBytes = sizeBytes u.contentType = contentType @@ -138,6 +140,53 @@ func TestUploadImageAllowsPayloadOverPreviousImageLimit(t *testing.T) { } } +func TestUploadImagesBatchUploadsSeriallyAndLimitsCount(t *testing.T) { + uploader := &fakeUploader{} + handler := New(uploader, "admin", fakeAudit{}) + router := ginRouter(handler) + body, contentType := multipartBatchUploadBody(t, []string{"one.png", "two.pag"}) + req := httptest.NewRequest(http.MethodPost, "/upload-images-batch", body) + req.Header.Set("Content-Type", contentType) + rec := httptest.NewRecorder() + + router.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } + if len(uploader.keys) != 2 { + t.Fatalf("expected 2 uploads, got %d", len(uploader.keys)) + } + var envelope struct { + Code int `json:"code"` + Data []struct { + Filename string `json:"filename"` + URL string `json:"url"` + } `json:"data"` + } + if err := json.NewDecoder(rec.Body).Decode(&envelope); err != nil { + t.Fatalf("decode response: %v", err) + } + if len(envelope.Data) != 2 || envelope.Data[0].Filename != "one.png" || envelope.Data[1].Filename != "two.pag" { + t.Fatalf("unexpected response order: %+v", envelope.Data) + } + + tooMany := make([]string, 11) + for index := range tooMany { + tooMany[index] = "asset.png" + } + body, contentType = multipartBatchUploadBody(t, tooMany) + req = httptest.NewRequest(http.MethodPost, "/upload-images-batch", body) + req.Header.Set("Content-Type", contentType) + rec = httptest.NewRecorder() + + router.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("too many status = %d body=%s", rec.Code, rec.Body.String()) + } +} + func ginRouter(handler *Handler) http.Handler { router := gin.New() router.Use(func(c *gin.Context) { @@ -146,6 +195,7 @@ func ginRouter(handler *Handler) http.Handler { c.Next() }) router.POST("/upload-image", handler.UploadImage) + router.POST("/upload-images-batch", handler.UploadImagesBatch) return router } @@ -168,3 +218,31 @@ func multipartUploadBody(t *testing.T, fieldName string, filename string, conten } return body, writer.FormDataContentType() } + +func multipartBatchUploadBody(t *testing.T, filenames []string) (*bytes.Buffer, string) { + t.Helper() + body := &bytes.Buffer{} + writer := multipart.NewWriter(body) + for _, filename := range filenames { + contentType := "image/png" + payload := append([]byte{0x89, 'P', 'N', 'G', '\r', '\n', 0x1a, '\n'}, bytes.Repeat([]byte{1}, 8)...) + if strings.HasSuffix(filename, ".pag") { + contentType = "application/octet-stream" + payload = []byte("PAG animated asset") + } + part, err := writer.CreatePart(map[string][]string{ + "Content-Disposition": {`form-data; name="files"; filename="` + filename + `"`}, + "Content-Type": {contentType}, + }) + if err != nil { + t.Fatalf("create multipart part: %v", err) + } + if _, err := part.Write(payload); err != nil { + t.Fatalf("write multipart part: %v", err) + } + } + if err := writer.Close(); err != nil { + t.Fatalf("close multipart writer: %v", err) + } + return body, writer.FormDataContentType() +} diff --git a/server/admin/internal/modules/upload/request.go b/server/admin/internal/modules/upload/request.go index a6c0e423..0436806a 100644 --- a/server/admin/internal/modules/upload/request.go +++ b/server/admin/internal/modules/upload/request.go @@ -11,4 +11,5 @@ type uploadResult struct { ObjectKey string `json:"object_key"` ContentType string `json:"content_type"` SizeBytes int64 `json:"size_bytes"` + Filename string `json:"filename,omitempty"` } diff --git a/server/admin/internal/modules/upload/routes.go b/server/admin/internal/modules/upload/routes.go index 4aa3ec3c..218aa5ea 100644 --- a/server/admin/internal/modules/upload/routes.go +++ b/server/admin/internal/modules/upload/routes.go @@ -13,4 +13,5 @@ func RegisterRoutes(protected *gin.RouterGroup, h *Handler) { protected.POST("/admin/files/upload", middleware.RequirePermission("upload:create"), h.UploadFile) protected.POST("/admin/files/image/upload", middleware.RequirePermission("upload:create"), h.UploadImage) + protected.POST("/admin/files/image/batch-upload", middleware.RequirePermission("upload:create"), h.UploadImagesBatch) } diff --git a/services/game-service/deploy/mysql/initdb/001_game_service.sql b/services/game-service/deploy/mysql/initdb/001_game_service.sql index 5368cad0..4bc41da9 100644 --- a/services/game-service/deploy/mysql/initdb/001_game_service.sql +++ b/services/game-service/deploy/mysql/initdb/001_game_service.sql @@ -197,24 +197,3 @@ CREATE TABLE IF NOT EXISTS game_outbox ( KEY idx_game_outbox_retention(app_code, status, updated_at_ms, event_id), KEY idx_game_outbox_order(app_code, order_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='游戏统计事件 outbox 表'; - -INSERT INTO game_platforms ( - app_code, platform_code, platform_name, status, api_base_url, sort_order, created_at_ms, updated_at_ms -) VALUES ( - 'lalu', 'demo', 'Demo Games', 'active', 'https://game.example.local/h5', 10, 0, 0 -) ON DUPLICATE KEY UPDATE platform_name = VALUES(platform_name); - -INSERT INTO game_catalog ( - app_code, game_id, platform_code, provider_game_id, game_name, category, icon_url, cover_url, - launch_mode, orientation, min_coin, status, sort_order, tags, created_at_ms, updated_at_ms -) VALUES ( - 'lalu', 'demo_rocket_001', 'demo', 'rocket_001', 'Rocket', 'casual', - 'https://cdn.example.com/game/rocket.png', 'https://cdn.example.com/game/rocket-cover.png', - 'h5_popup', 'portrait', 100, 'active', 10, JSON_ARRAY('demo','voice_room'), 0, 0 -) ON DUPLICATE KEY UPDATE game_name = VALUES(game_name); - -INSERT INTO game_display_rules ( - app_code, rule_id, game_id, scene, region_id, language, platform, visible, enabled, sort_order, created_at_ms, updated_at_ms -) VALUES ( - 'lalu', 'demo_rocket_voice_default', 'demo_rocket_001', 'voice_room', 0, '', '', 1, 1, 10, 0, 0 -) ON DUPLICATE KEY UPDATE visible = VALUES(visible), enabled = VALUES(enabled); diff --git a/services/game-service/internal/service/game/baishun_adapter.go b/services/game-service/internal/service/game/baishun_adapter.go index 5306f11c..d2ef9133 100644 --- a/services/game-service/internal/service/game/baishun_adapter.go +++ b/services/game-service/internal/service/game/baishun_adapter.go @@ -263,7 +263,7 @@ func (s *Service) handleBaishunGetSSToken(ctx context.Context, app string, confi } snapshot, err := s.yomiUserSnapshot(ctx, app, req, session.UserID) if err != nil { - return baishunAdapterError(req.GetMeta().GetRequestId(), baishunCodeFromError(err), baishunMessageFromError(err), true, "") + return baishunAdapterError(req.GetMeta().GetRequestId(), baishunUserSnapshotCodeFromError(err), baishunUserSnapshotMessageFromError(err), true, "") } response := map[string]any{ "code": baishunCodeOK, @@ -289,7 +289,7 @@ func (s *Service) handleBaishunUserInfo(ctx context.Context, app string, config } snapshot, err := s.yomiUserSnapshot(ctx, app, req, session.UserID) if err != nil { - return baishunAdapterError(req.GetMeta().GetRequestId(), baishunCodeFromError(err), baishunMessageFromError(err), true, "") + return baishunAdapterError(req.GetMeta().GetRequestId(), baishunUserSnapshotCodeFromError(err), baishunUserSnapshotMessageFromError(err), true, "") } raw, contentType := baishunJSON(req.GetMeta().GetRequestId(), baishunCodeOK, baishunDefaultSuccessMessage, baishunUserInfoData(session, config, snapshot)) return raw, contentType, strconv.Itoa(baishunCodeOK), true, "", nil @@ -466,3 +466,17 @@ func baishunMessageFromError(err error) string { return "api server error" } } + +func baishunUserSnapshotCodeFromError(err error) int { + if xerr.IsCode(err, xerr.NotFound) { + return baishunCodeInvalidData + } + return baishunCodeFromError(err) +} + +func baishunUserSnapshotMessageFromError(err error) string { + if xerr.IsCode(err, xerr.NotFound) { + return "token invalid" + } + return baishunMessageFromError(err) +} diff --git a/services/game-service/internal/service/game/service.go b/services/game-service/internal/service/game/service.go index 047e5304..cf981eb3 100644 --- a/services/game-service/internal/service/game/service.go +++ b/services/game-service/internal/service/game/service.go @@ -704,13 +704,14 @@ func buildLaunchURL(game gamedomain.LaunchableGame, session gamedomain.LaunchSes if strings.EqualFold(game.AdapterType, gamedomain.AdapterBaishunV1) { // 百顺前端规范以 NativeBridge.getConfig 为准；这里把同一份配置也拼到 URL，方便 Flutter/H5 壳层读取。 config := baishunConfigFromPlatform(game) - query.Set("bridge_keys", "appChannel,appId,code,gameMode,gsp,language,roomId,sceneMode") + query.Set("bridge_keys", "appChannel,appId,userId,code,gameMode,gsp,language,roomId,sceneMode,currencyIcon,isDataByUrl") query.Set("appChannel", config.AppChannel) if config.AppID > 0 { query.Set("appId", strconv.FormatInt(config.AppID, 10)) } query.Set("userId", externalUserID(session, config.UIDMode)) query.Set("code", token) + query.Set("isDataByUrl", "1") if session.RoomID != "" { query.Set("roomId", session.RoomID) } diff --git a/services/game-service/internal/service/game/service_test.go b/services/game-service/internal/service/game/service_test.go index 288a5303..43e92722 100644 --- a/services/game-service/internal/service/game/service_test.go +++ b/services/game-service/internal/service/game/service_test.go @@ -323,7 +323,8 @@ func TestLaunchGameBuildsBaishunURL(t *testing.T) { "gameMode": "3", "language": "2", "sceneMode": "1", - "bridge_keys": "appChannel,appId,code,gameMode,gsp,language,roomId,sceneMode", + "isDataByUrl": "1", + "bridge_keys": "appChannel,appId,userId,code,gameMode,gsp,language,roomId,sceneMode,currencyIcon,isDataByUrl", "currencyIcon": "https://cdn.example/coin.png", "gsp": "101", } @@ -1049,6 +1050,51 @@ func TestHandleBaishunAccessTokenIgnoresStaleLaunchSessionGameID(t *testing.T) { } } +func TestHandleBaishunGetSSTokenReturnsTokenInvalidWhenUserMissing(t *testing.T) { + secret := "baishun-test-app-key" + token := "app_access_token_baishun" + repo := &fakeRepository{ + platform: gamedomain.Platform{ + PlatformCode: "baishun", + AdapterType: gamedomain.AdapterBaishunV1, + CallbackSecretCiphertext: secret, + AdapterConfigJSON: `{"app_id":21397507,"app_channel":"mesh","uid_mode":"display_user_id"}`, + }, + session: gamedomain.LaunchSession{ + AppCode: "lalu", + SessionID: "game_sess_baishun_missing_user", + UserID: 42, + DisplayUserID: "420001", + RoomID: "room_1", + PlatformCode: "baishun", + GameID: "baishun_greedy", + ProviderGameID: "1006", + LaunchTokenHash: stableHash(token), + Status: gamedomain.SessionActive, + ExpiresAtMS: 1700086400000, + }, + } + svc := New(Config{}, repo, &fakeWallet{}, &fakeUser{err: xerr.New(xerr.NotFound, "user not found")}) + svc.now = func() time.Time { return time.UnixMilli(1700000000000) } + + raw, contentType, err := svc.HandleCallback(context.Background(), &gamev1.CallbackRequest{ + Meta: &gamev1.RequestMeta{RequestId: "req-baishun-missing-user", AppCode: "lalu"}, + PlatformCode: "baishun", + Operation: "get_sstoken", + RawBody: baishunTestBody(t, secret, `{"app_id":21397507,"user_id":"420001","code":"app_access_token_baishun"}`, 1700000000), + RemoteAddr: "3.1.174.194:443", + }) + if err != nil { + t.Fatalf("HandleCallback get_sstoken failed: %v", err) + } + if contentType != "application/json" || !strings.Contains(string(raw), `"code":1001`) || !strings.Contains(string(raw), `"message":"token invalid"`) { + t.Fatalf("baishun missing user response mismatch: %s", raw) + } + if strings.Contains(string(raw), `"code":1002`) || strings.Contains(string(raw), "read data failed") { + t.Fatalf("baishun missing user must not look like read data failure: %s", raw) + } +} + func TestHandleCallbackChangeCoinUsesWalletAndOrderIdempotency(t *testing.T) { repo := &fakeRepository{ launchable: gamedomain.LaunchableGame{CatalogItem: gamedomain.CatalogItem{GameID: "demo_rocket_001", ProviderGameID: "rocket_001"}}, @@ -1313,10 +1359,14 @@ func (f *fakeActivity) ConsumeLevelEvent(_ context.Context, req *activityv1.Cons type fakeUser struct { lastGet *userv1.GetUserRequest + err error } func (f *fakeUser) GetUser(_ context.Context, req *userv1.GetUserRequest) (*userv1.GetUserResponse, error) { f.lastGet = req + if f.err != nil { + return nil, f.err + } return &userv1.GetUserResponse{User: &userv1.User{ UserId: req.GetUserId(), DisplayUserId: "420001", diff --git a/services/game-service/internal/storage/mysql/repository.go b/services/game-service/internal/storage/mysql/repository.go index 05ccd9a3..a74d866d 100644 --- a/services/game-service/internal/storage/mysql/repository.go +++ b/services/game-service/internal/storage/mysql/repository.go @@ -285,7 +285,7 @@ func (r *Repository) Migrate(ctx context.Context) error { if err := r.ensureIndexDefinition(ctx, "game_level_event_outbox", "idx_game_level_event_retention", []string{"app_code", "status", "updated_at_ms", "event_id"}, "ALTER TABLE game_level_event_outbox ADD INDEX idx_game_level_event_retention(app_code, status, updated_at_ms, event_id)"); err != nil { return err } - return r.seedDefaults(ctx) + return nil } func (r *Repository) ensureColumn(ctx context.Context, tableName string, columnName string, columnDDL string) error { @@ -352,33 +352,6 @@ func equalStringSlices(left []string, right []string) bool { return true } -func (r *Repository) seedDefaults(ctx context.Context) error { - if _, err := r.db.ExecContext(ctx, - `INSERT INTO game_platforms (app_code, platform_code, platform_name, status, api_base_url, sort_order, created_at_ms, updated_at_ms) - VALUES ('lalu', 'demo', 'Demo Games', 'active', 'https://game.example.local/h5', 10, 0, 0) - ON DUPLICATE KEY UPDATE platform_name = VALUES(platform_name)`); err != nil { - return err - } - if _, err := r.db.ExecContext(ctx, - `INSERT INTO game_catalog ( - app_code, game_id, platform_code, provider_game_id, game_name, category, icon_url, cover_url, - launch_mode, orientation, min_coin, status, sort_order, tags, created_at_ms, updated_at_ms - ) VALUES ( - 'lalu', 'demo_rocket_001', 'demo', 'rocket_001', 'Rocket', 'casual', - 'https://cdn.example.com/game/rocket.png', 'https://cdn.example.com/game/rocket-cover.png', - 'h5_popup', 'portrait', 100, 'active', 10, JSON_ARRAY('demo','voice_room'), 0, 0 - ) ON DUPLICATE KEY UPDATE game_name = VALUES(game_name)`); err != nil { - return err - } - _, err := r.db.ExecContext(ctx, - `INSERT INTO game_display_rules ( - app_code, rule_id, game_id, scene, region_id, language, platform, visible, enabled, sort_order, created_at_ms, updated_at_ms - ) VALUES ( - 'lalu', 'demo_rocket_voice_default', 'demo_rocket_001', 'voice_room', 0, '', '', 1, 1, 10, 0, 0 - ) ON DUPLICATE KEY UPDATE visible = VALUES(visible), enabled = VALUES(enabled)`) - return err -} - func (r *Repository) ListGames(ctx context.Context, query gameservice.ListGamesQuery) ([]gamedomain.AppGame, error) { rows, err := r.db.QueryContext(ctx, `SELECT c.game_id, c.platform_code, c.game_name, c.icon_url, c.cover_url, c.category, diff --git a/services/gateway-service/configs/config.yaml b/services/gateway-service/configs/config.yaml index 8e7f2051..ba2490de 100644 --- a/services/gateway-service/configs/config.yaml +++ b/services/gateway-service/configs/config.yaml @@ -15,6 +15,8 @@ cors: - "http://127.0.0.1:3000" - "http://localhost:5173" - "http://127.0.0.1:5173" + - "http://localhost:30000" + - "http://127.0.0.1:30000" # 本地工具页会用 10300/8787/8788 承载静态 HTML；精确放行测试端口，避免用 "*" 破坏凭证请求边界。 - "http://localhost:10300" - "http://127.0.0.1:10300" diff --git a/services/gateway-service/internal/transport/http/appapi/handler.go b/services/gateway-service/internal/transport/http/appapi/handler.go index 0f6fec54..1f8e6796 100644 --- a/services/gateway-service/internal/transport/http/appapi/handler.go +++ b/services/gateway-service/internal/transport/http/appapi/handler.go @@ -72,6 +72,10 @@ func (h *Handler) UploadAvatar(writer http.ResponseWriter, request *http.Request h.uploadAvatar(writer, request) } +func (h *Handler) ProxyEffectMedia(writer http.ResponseWriter, request *http.Request) { + h.proxyEffectMedia(writer, request) +} + func (h *Handler) HandlePushToken(writer http.ResponseWriter, request *http.Request) { h.handlePushToken(writer, request) } diff --git a/services/gateway-service/internal/transport/http/appapi/media_proxy_handler.go b/services/gateway-service/internal/transport/http/appapi/media_proxy_handler.go new file mode 100644 index 00000000..d5237b45 --- /dev/null +++ b/services/gateway-service/internal/transport/http/appapi/media_proxy_handler.go @@ -0,0 +1,94 @@ +package appapi + +import ( + "io" + "net/http" + "net/url" + "path" + "strconv" + "strings" + "time" + + "hyapp/services/gateway-service/internal/transport/http/httpkit" +) + +const effectMediaProxyMaxBytes int64 = 30 << 20 + +var ( + effectMediaProxyHTTPClient = &http.Client{Timeout: 15 * time.Second} + effectMediaProxyHosts = map[string]struct{}{ + "media.haiyihy.com": {}, + } + effectMediaProxyExtensions = map[string]string{ + ".pag": "application/octet-stream", + ".svga": "application/octet-stream", + } +) + +// proxyEffectMedia streams trusted binary animation assets through gateway so +// browser-only SVGA/PAG runtimes can read them even when the media CDN has no CORS. +func (h *Handler) proxyEffectMedia(writer http.ResponseWriter, request *http.Request) { + rawURL := strings.TrimSpace(request.URL.Query().Get("url")) + sourceURL, contentType, ok := validateEffectMediaProxyURL(rawURL) + if !ok { + httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidArgument, "invalid argument") + return + } + + upstreamRequest, err := http.NewRequestWithContext(request.Context(), http.MethodGet, sourceURL.String(), nil) + if err != nil { + httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidArgument, "invalid argument") + return + } + upstreamRequest.Header.Set("Accept", "application/octet-stream,*/*") + upstreamRequest.Header.Set("User-Agent", "hyapp-gateway-effect-media-proxy/1.0") + + upstream, err := effectMediaProxyHTTPClient.Do(upstreamRequest) + if err != nil { + httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error") + return + } + defer upstream.Body.Close() + + if upstream.StatusCode < http.StatusOK || upstream.StatusCode >= http.StatusMultipleChoices { + httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error") + return + } + if upstream.ContentLength > effectMediaProxyMaxBytes { + httpkit.WriteError(writer, request, http.StatusRequestEntityTooLarge, httpkit.CodeInvalidArgument, "invalid argument") + return + } + if value := strings.TrimSpace(upstream.Header.Get("Content-Type")); value != "" { + contentType = value + } + + header := writer.Header() + header.Set("Content-Type", contentType) + header.Set("Cache-Control", "public, max-age=86400") + header.Set("X-Content-Type-Options", "nosniff") + if upstream.ContentLength >= 0 { + header.Set("Content-Length", strconv.FormatInt(upstream.ContentLength, 10)) + } + + writer.WriteHeader(http.StatusOK) + _, _ = io.Copy(writer, io.LimitReader(upstream.Body, effectMediaProxyMaxBytes+1)) +} + +func validateEffectMediaProxyURL(rawURL string) (*url.URL, string, bool) { + sourceURL, err := url.Parse(rawURL) + if err != nil || sourceURL == nil { + return nil, "", false + } + if sourceURL.Scheme != "https" { + return nil, "", false + } + host := strings.ToLower(sourceURL.Hostname()) + if _, ok := effectMediaProxyHosts[host]; !ok { + return nil, "", false + } + contentType, ok := effectMediaProxyExtensions[strings.ToLower(path.Ext(sourceURL.Path))] + if !ok { + return nil, "", false + } + return sourceURL, contentType, true +} diff --git a/services/gateway-service/internal/transport/http/appapi/media_proxy_handler_test.go b/services/gateway-service/internal/transport/http/appapi/media_proxy_handler_test.go new file mode 100644 index 00000000..04d51007 --- /dev/null +++ b/services/gateway-service/internal/transport/http/appapi/media_proxy_handler_test.go @@ -0,0 +1,46 @@ +package appapi + +import "testing" + +func TestValidateEffectMediaProxyURL(t *testing.T) { + tests := []struct { + name string + url string + ok bool + }{ + { + name: "svga media host", + url: "https://media.haiyihy.com/admin/files/1/20260602/file.svga", + ok: true, + }, + { + name: "pag media host", + url: "https://media.haiyihy.com/admin/files/1/20260602/file.pag?sign=1", + ok: true, + }, + { + name: "reject http", + url: "http://media.haiyihy.com/admin/files/1/file.svga", + ok: false, + }, + { + name: "reject untrusted host", + url: "https://example.com/admin/files/1/file.svga", + ok: false, + }, + { + name: "reject regular image", + url: "https://media.haiyihy.com/admin/files/1/file.png", + ok: false, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, _, ok := validateEffectMediaProxyURL(test.url) + if ok != test.ok { + t.Fatalf("ok mismatch: got=%t want=%t", ok, test.ok) + } + }) + } +} diff --git a/services/gateway-service/internal/transport/http/httproutes/router.go b/services/gateway-service/internal/transport/http/httproutes/router.go index 7ef19858..62629a84 100644 --- a/services/gateway-service/internal/transport/http/httproutes/router.go +++ b/services/gateway-service/internal/transport/http/httproutes/router.go @@ -57,6 +57,7 @@ type AppHandlers struct { HandleTencentRTCCallback http.HandlerFunc UploadFile http.HandlerFunc UploadAvatar http.HandlerFunc + ProxyEffectMedia http.HandlerFunc HandlePushToken http.HandlerFunc } @@ -276,6 +277,7 @@ func (r routes) registerAppRoutes() { r.public("/tencent-rtc/callback", "", h.HandleTencentRTCCallback) r.auth("/files/upload", "", h.UploadFile) r.auth("/files/avatar/upload", "", h.UploadAvatar) + r.public("/media/effects/proxy", http.MethodGet, h.ProxyEffectMedia) r.profile("/devices/push-token", "", h.HandlePushToken) } diff --git a/services/gateway-service/internal/transport/http/router.go b/services/gateway-service/internal/transport/http/router.go index 3bf162bf..dadea1e9 100644 --- a/services/gateway-service/internal/transport/http/router.go +++ b/services/gateway-service/internal/transport/http/router.go @@ -149,6 +149,7 @@ func (h *Handler) Routes(jwtVerifier *auth.Verifier) http.Handler { HandleTencentRTCCallback: callbackAPI.HandleTencentRTCCallback, UploadFile: appAPI.UploadFile, UploadAvatar: appAPI.UploadAvatar, + ProxyEffectMedia: appAPI.ProxyEffectMedia, HandlePushToken: appAPI.HandlePushToken, }, User: userHandlers, diff --git a/services/gateway-service/internal/transport/http/walletapi/app_wallet_handler.go b/services/gateway-service/internal/transport/http/walletapi/app_wallet_handler.go index a2667a13..e763ca0a 100644 --- a/services/gateway-service/internal/transport/http/walletapi/app_wallet_handler.go +++ b/services/gateway-service/internal/transport/http/walletapi/app_wallet_handler.go @@ -175,6 +175,10 @@ type vipRewardItemData struct { Name string `json:"name"` Quantity int64 `json:"quantity"` ExpiresAtMS int64 `json:"expires_at_ms"` + CoverURL string `json:"cover_url"` + AssetURL string `json:"asset_url"` + PreviewURL string `json:"preview_url"` + AnimationURL string `json:"animation_url"` } // getWalletOverview 返回钱包二级页摘要；我的页首屏使用更轻的 WalletValueSummary。 diff --git a/services/gateway-service/internal/transport/http/walletapi/vip_handler.go b/services/gateway-service/internal/transport/http/walletapi/vip_handler.go index f39d7023..86dd03e5 100644 --- a/services/gateway-service/internal/transport/http/walletapi/vip_handler.go +++ b/services/gateway-service/internal/transport/http/walletapi/vip_handler.go @@ -5,9 +5,10 @@ import ( "net/http" "strings" - walletv1 "hyapp.local/api/proto/wallet/v1" "hyapp/pkg/appcode" "hyapp/services/gateway-service/internal/auth" + + walletv1 "hyapp.local/api/proto/wallet/v1" ) // getMyVIP 返回当前用户 VIP 状态。 @@ -135,7 +136,21 @@ func vipRewardItemsFromProto(items []*walletv1.VipRewardItem) []vipRewardItemDat Name: item.GetName(), Quantity: item.GetQuantity(), ExpiresAtMS: item.GetExpiresAtMs(), + CoverURL: vipRewardCoverURL(item), + AssetURL: item.GetAssetUrl(), + PreviewURL: item.GetPreviewUrl(), + AnimationURL: item.GetAnimationUrl(), }) } return resp } + +func vipRewardCoverURL(item *walletv1.VipRewardItem) string { + if item == nil { + return "" + } + if strings.TrimSpace(item.GetPreviewUrl()) != "" { + return item.GetPreviewUrl() + } + return item.GetAssetUrl() +} diff --git a/services/wallet-service/internal/domain/ledger/ledger.go b/services/wallet-service/internal/domain/ledger/ledger.go index 0cf6307f..1207385c 100644 --- a/services/wallet-service/internal/domain/ledger/ledger.go +++ b/services/wallet-service/internal/domain/ledger/ledger.go @@ -504,6 +504,9 @@ type VipRewardItem struct { Name string Quantity int64 ExpiresAtMS int64 + AssetURL string + PreviewURL string + AnimationURL string } // VipLevel 是可购买 VIP 等级配置。 diff --git a/services/wallet-service/internal/service/wallet/service_test.go b/services/wallet-service/internal/service/wallet/service_test.go index 0b03cd92..e3b00312 100644 --- a/services/wallet-service/internal/service/wallet/service_test.go +++ b/services/wallet-service/internal/service/wallet/service_test.go @@ -2486,6 +2486,15 @@ func TestApplyWithdrawalFreezesUSDBalance(t *testing.T) { func TestPurchaseVipDebitsCoinAndGrantsReward(t *testing.T) { repository := mysqltest.NewRepository(t) repository.SetBalance(51001, 20000) + repository.SeedVIPRewardGroup(1, 71001, []mysqltest.VipRewardResourceSeed{ + { + ResourceID: 7100101, + ResourceCode: "vip1_avatar_frame", + ResourceType: resourcedomain.TypeAvatarFrame, + Name: "VIP1 Avatar Frame", + GrantStrategy: resourcedomain.GrantStrategyExtendExpiry, + }, + }) svc := walletservice.New(repository) receipt, err := svc.PurchaseVip(context.Background(), ledger.PurchaseVipCommand{ @@ -2554,6 +2563,80 @@ func TestPurchaseVipDebitsCoinAndGrantsReward(t *testing.T) { } } +func TestPurchaseVipRenewalAlignsRewardEntitlementExpiry(t *testing.T) { + repository := mysqltest.NewRepository(t) + repository.SetBalance(51004, 20000) + resources := []mysqltest.VipRewardResourceSeed{ + { + ResourceID: 7100401, + ResourceCode: "vip_renew_frame", + ResourceType: resourcedomain.TypeAvatarFrame, + Name: "VIP Renew Frame", + GrantStrategy: resourcedomain.GrantStrategyExtendExpiry, + }, + { + ResourceID: 7100402, + ResourceCode: "vip_renew_mic", + ResourceType: resourcedomain.TypeMicSeatAnimation, + Name: "VIP Renew Mic", + GrantStrategy: resourcedomain.GrantStrategyIncreaseQuantity, + }, + { + ResourceID: 7100403, + ResourceCode: "vip_renew_badge", + ResourceType: resourcedomain.TypeBadge, + Name: "VIP Renew Badge", + GrantStrategy: resourcedomain.GrantStrategySetActiveFlag, + }, + } + repository.SeedVIPRewardGroup(1, 71004, resources) + svc := walletservice.New(repository) + + first, err := svc.PurchaseVip(context.Background(), ledger.PurchaseVipCommand{ + CommandID: "cmd-vip-renew-expiry-first", + UserID: 51004, + Level: 1, + }) + if err != nil { + t.Fatalf("first VIP purchase failed: %v", err) + } + if len(first.RewardItems) != len(resources) { + t.Fatalf("first reward item count mismatch: %+v", first.RewardItems) + } + for _, resource := range resources { + expiresAt, _, _ := repository.ActiveResourceEntitlement(51004, resource.ResourceID) + if expiresAt != first.Vip.ExpiresAtMS { + t.Fatalf("first entitlement expiry mismatch for %d: got %d want %d", resource.ResourceID, expiresAt, first.Vip.ExpiresAtMS) + } + } + + renew, err := svc.PurchaseVip(context.Background(), ledger.PurchaseVipCommand{ + CommandID: "cmd-vip-renew-expiry-second", + UserID: 51004, + Level: 1, + }) + if err != nil { + t.Fatalf("VIP renewal failed: %v", err) + } + if renew.Vip.ExpiresAtMS != first.Vip.ExpiresAtMS+604800000 { + t.Fatalf("renewed VIP expiry mismatch: first=%d renewed=%d", first.Vip.ExpiresAtMS, renew.Vip.ExpiresAtMS) + } + if len(renew.RewardItems) != len(resources) { + t.Fatalf("renew reward item count mismatch: %+v", renew.RewardItems) + } + for _, item := range renew.RewardItems { + if item.ExpiresAtMS != renew.Vip.ExpiresAtMS { + t.Fatalf("renew reward response expiry mismatch for %d: got %d want %d", item.ResourceID, item.ExpiresAtMS, renew.Vip.ExpiresAtMS) + } + } + for _, resource := range resources { + expiresAt, _, _ := repository.ActiveResourceEntitlement(51004, resource.ResourceID) + if expiresAt != renew.Vip.ExpiresAtMS { + t.Fatalf("renew entitlement expiry mismatch for %d: got %d want %d", resource.ResourceID, expiresAt, renew.Vip.ExpiresAtMS) + } + } +} + // TestGrantVipUsesUnifiedActivationAndNoticeOutbox 验证后台和活动赠送 VIP 走统一激活入口并写 IM outbox。 func TestGrantVipUsesUnifiedActivationAndNoticeOutbox(t *testing.T) { repository := mysqltest.NewRepository(t) diff --git a/services/wallet-service/internal/storage/mysql/app_wallet_repository.go b/services/wallet-service/internal/storage/mysql/app_wallet_repository.go index cba8fdbd..0ac96574 100644 --- a/services/wallet-service/internal/storage/mysql/app_wallet_repository.go +++ b/services/wallet-service/internal/storage/mysql/app_wallet_repository.go @@ -1067,6 +1067,11 @@ func (r *Repository) applyVipRewardGroup(ctx context.Context, tx *sql.Tx, comman if err != nil { return nil, "", err } + if inserted.EntitlementID != "" { + if err := r.alignEntitlementExpiresAt(ctx, tx, inserted.EntitlementID, expiresAtMS, nowMs); err != nil { + return nil, "", err + } + } rewards = append(rewards, ledger.VipRewardItem{ ResourceID: item.Resource.ResourceID, ResourceCode: item.Resource.ResourceCode, @@ -1074,6 +1079,9 @@ func (r *Repository) applyVipRewardGroup(ctx context.Context, tx *sql.Tx, comman Name: item.Resource.Name, Quantity: inserted.Quantity, ExpiresAtMS: expiresAtMS, + AssetURL: item.Resource.AssetURL, + PreviewURL: item.Resource.PreviewURL, + AnimationURL: item.Resource.AnimationURL, }) } if err := r.insertWalletOutbox(ctx, tx, []walletOutboxEvent{ @@ -1104,6 +1112,9 @@ func (r *Repository) vipRewardItems(ctx context.Context, groupID int64, expiresA Name: item.Resource.Name, Quantity: item.Quantity, ExpiresAtMS: expiresAtMS, + AssetURL: item.Resource.AssetURL, + PreviewURL: item.Resource.PreviewURL, + AnimationURL: item.Resource.AnimationURL, }) } return rewards, nil @@ -1151,6 +1162,9 @@ func vipRewardItemsNoticePayload(items []ledger.VipRewardItem) []map[string]any "name": item.Name, "quantity": item.Quantity, "expires_at_ms": item.ExpiresAtMS, + "asset_url": item.AssetURL, + "preview_url": item.PreviewURL, + "animation_url": item.AnimationURL, }) } return payload @@ -1217,7 +1231,8 @@ func (r *Repository) rewardItemsByGrant(ctx context.Context, tx *sql.Tx, grantID return nil, nil } rows, err := tx.QueryContext(ctx, ` - SELECT i.resource_id, r.resource_code, r.resource_type, r.name, i.quantity + SELECT i.resource_id, r.resource_code, r.resource_type, r.name, i.quantity, + COALESCE(r.asset_url, ''), COALESCE(r.preview_url, ''), COALESCE(r.animation_url, '') FROM resource_grant_items i JOIN resources r ON r.app_code = i.app_code AND r.resource_id = i.resource_id WHERE i.app_code = ? AND i.grant_id = ? AND i.result_type = ? @@ -1231,7 +1246,16 @@ func (r *Repository) rewardItemsByGrant(ctx context.Context, tx *sql.Tx, grantID items := make([]ledger.VipRewardItem, 0) for rows.Next() { var item ledger.VipRewardItem - if err := rows.Scan(&item.ResourceID, &item.ResourceCode, &item.ResourceType, &item.Name, &item.Quantity); err != nil { + if err := rows.Scan( + &item.ResourceID, + &item.ResourceCode, + &item.ResourceType, + &item.Name, + &item.Quantity, + &item.AssetURL, + &item.PreviewURL, + &item.AnimationURL, + ); err != nil { return nil, err } item.ExpiresAtMS = expiresAtMS diff --git a/services/wallet-service/internal/storage/mysql/resource_repository.go b/services/wallet-service/internal/storage/mysql/resource_repository.go index 43f96ff0..66410d72 100644 --- a/services/wallet-service/internal/storage/mysql/resource_repository.go +++ b/services/wallet-service/internal/storage/mysql/resource_repository.go @@ -1135,18 +1135,18 @@ func (r *Repository) PurchaseResourceShopItem(ctx context.Context, command resou durationMS := int64(item.DurationDays) * 24 * int64(time.Hour/time.Millisecond) coinBalanceAfter := account.AvailableAmount - item.CoinPrice metadata := map[string]any{ - "app_code": command.AppCode, - "order_id": orderID, - "user_id": command.UserID, - "shop_item_id": item.ShopItemID, - "resource_id": item.ResourceID, - "resource_code": item.Resource.ResourceCode, - "resource_type": item.Resource.ResourceType, - "duration_days": item.DurationDays, - "duration_ms": durationMS, - "price_type": item.PriceType, - "coin_spent": item.CoinPrice, - "balance_after": coinBalanceAfter, + "app_code": command.AppCode, + "order_id": orderID, + "user_id": command.UserID, + "shop_item_id": item.ShopItemID, + "resource_id": item.ResourceID, + "resource_code": item.Resource.ResourceCode, + "resource_type": item.Resource.ResourceType, + "duration_days": item.DurationDays, + "duration_ms": durationMS, + "price_type": item.PriceType, + "coin_spent": item.CoinPrice, + "balance_after": coinBalanceAfter, "resource_grant_id": grantID, } if err := r.insertTransaction(ctx, tx, transactionID, command.CommandID, bizTypeResourceShopPurchase, requestHash, orderID, metadata, nowMs); err != nil { @@ -1563,11 +1563,25 @@ func (r *Repository) applyEntitlement(ctx context.Context, tx *sql.Tx, userID in return entitlementID, nil } +func (r *Repository) alignEntitlementExpiresAt(ctx context.Context, tx *sql.Tx, entitlementID string, expiresAtMS int64, nowMs int64) error { + if strings.TrimSpace(entitlementID) == "" { + return nil + } + _, err := tx.ExecContext(ctx, ` + UPDATE user_resource_entitlements + SET expires_at_ms = ?, updated_at_ms = ? + WHERE app_code = ? AND entitlement_id = ?`, + expiresAtMS, nowMs, appcode.FromContext(ctx), entitlementID, + ) + return err +} + func (r *Repository) queryActiveEntitlementForUpdate(ctx context.Context, tx *sql.Tx, userID int64, resourceID int64, nowMs int64) (resourcedomain.UserResourceEntitlement, bool, error) { row := tx.QueryRowContext(ctx, ` SELECT e.app_code, e.entitlement_id, e.user_id, e.resource_id, e.status, e.quantity, e.remaining_quantity, e.effective_at_ms, e.expires_at_ms, e.source_grant_id, e.created_at_ms, e.updated_at_ms, + FALSE AS equipped, `+resourceColumnsWithAlias("r")+` FROM user_resource_entitlements e JOIN resources r ON r.app_code = e.app_code AND r.resource_id = e.resource_id diff --git a/services/wallet-service/internal/testutil/mysqltest/mysqltest.go b/services/wallet-service/internal/testutil/mysqltest/mysqltest.go index 8f4cecb8..11dec55b 100644 --- a/services/wallet-service/internal/testutil/mysqltest/mysqltest.go +++ b/services/wallet-service/internal/testutil/mysqltest/mysqltest.go @@ -21,6 +21,15 @@ type Repository struct { schema *mysqlschema.Schema } +type VipRewardResourceSeed struct { + ResourceID int64 + ResourceCode string + ResourceType string + Name string + GrantStrategy string + SortOrder int +} + // CountRows returns a simple count from a known wallet table for persistence assertions. func (r *Repository) CountRows(table string, where string, args ...any) int { r.t.Helper() @@ -36,6 +45,22 @@ func (r *Repository) CountRows(table string, where string, args ...any) int { return count } +func (r *Repository) ActiveResourceEntitlement(userID int64, resourceID int64) (expiresAtMS int64, quantity int64, remainingQuantity int64) { + r.t.Helper() + + err := r.schema.DB.QueryRowContext(context.Background(), ` + SELECT expires_at_ms, quantity, remaining_quantity + FROM user_resource_entitlements + WHERE user_id = ? AND resource_id = ? AND status = 'active' + ORDER BY expires_at_ms DESC, created_at_ms DESC + LIMIT 1 + `, userID, resourceID).Scan(&expiresAtMS, &quantity, &remainingQuantity) + if err != nil { + r.t.Fatalf("query active resource entitlement failed: %v", err) + } + return expiresAtMS, quantity, remainingQuantity +} + // HostSalarySettlementIDs returns the persisted settlement IDs for one host/cycle in creation order. func (r *Repository) HostSalarySettlementIDs(userID int64, cycleKey string) []string { r.t.Helper() @@ -191,6 +216,74 @@ func (r *Repository) SetBalance(userID int64, balance int64) { r.SetAssetBalance(userID, "COIN", balance) } +func (r *Repository) SeedVIPRewardGroup(level int32, groupID int64, resources []VipRewardResourceSeed) { + r.t.Helper() + + nowMs := time.Now().UnixMilli() + _, err := r.schema.DB.ExecContext(context.Background(), ` + INSERT INTO resource_groups ( + group_id, group_code, name, status, description, sort_order, + created_by_user_id, updated_by_user_id, created_at_ms, updated_at_ms + ) VALUES (?, ?, ?, 'active', '', 0, 0, 0, ?, ?) + ON DUPLICATE KEY UPDATE + status = VALUES(status), + updated_at_ms = VALUES(updated_at_ms) + `, groupID, fmt.Sprintf("vip_test_group_%d", groupID), fmt.Sprintf("VIP test group %d", groupID), nowMs, nowMs) + if err != nil { + r.t.Fatalf("seed vip reward group failed: %v", err) + } + if _, err := r.schema.DB.ExecContext(context.Background(), `DELETE FROM resource_group_items WHERE group_id = ?`, groupID); err != nil { + r.t.Fatalf("clear vip reward group items failed: %v", err) + } + for index, resource := range resources { + sortOrder := resource.SortOrder + if sortOrder == 0 { + sortOrder = (index + 1) * 10 + } + name := resource.Name + if name == "" { + name = resource.ResourceCode + } + _, err := r.schema.DB.ExecContext(context.Background(), ` + INSERT INTO resources ( + resource_id, resource_code, resource_type, name, status, + grantable, manager_grant_enabled, grant_strategy, wallet_asset_type, + wallet_asset_amount, price_type, coin_price, gift_point_amount, + usage_scope_json, asset_url, preview_url, animation_url, + metadata_json, sort_order, created_by_user_id, updated_by_user_id, + created_at_ms, updated_at_ms + ) VALUES (?, ?, ?, ?, 'active', TRUE, TRUE, ?, '', 0, '', 0, 0, + JSON_ARRAY(), '', '', '', NULL, ?, 0, 0, ?, ?) + ON DUPLICATE KEY UPDATE + resource_type = VALUES(resource_type), + name = VALUES(name), + status = VALUES(status), + grant_strategy = VALUES(grant_strategy), + sort_order = VALUES(sort_order), + updated_at_ms = VALUES(updated_at_ms) + `, resource.ResourceID, resource.ResourceCode, resource.ResourceType, name, resource.GrantStrategy, sortOrder, nowMs, nowMs) + if err != nil { + r.t.Fatalf("seed vip reward resource failed: %v", err) + } + _, err = r.schema.DB.ExecContext(context.Background(), ` + INSERT INTO resource_group_items ( + group_id, item_type, resource_id, wallet_asset_type, wallet_asset_amount, + quantity, duration_ms, sort_order, created_at_ms, updated_at_ms + ) VALUES (?, 'resource', ?, '', 0, 1, 0, ?, ?, ?) + `, groupID, resource.ResourceID, sortOrder, nowMs, nowMs) + if err != nil { + r.t.Fatalf("seed vip reward group item failed: %v", err) + } + } + if _, err := r.schema.DB.ExecContext(context.Background(), ` + UPDATE vip_levels + SET reward_resource_group_id = ?, updated_at_ms = ? + WHERE level = ? + `, groupID, nowMs, level); err != nil { + r.t.Fatalf("update vip level reward group failed: %v", err) + } +} + // SetAssetBalance prepares a sender account for debit tests. func (r *Repository) SetAssetBalance(userID int64, assetType string, balance int64) { r.t.Helper() diff --git a/services/wallet-service/internal/transport/grpc/server.go b/services/wallet-service/internal/transport/grpc/server.go index 3c5dfc32..5c60befe 100644 --- a/services/wallet-service/internal/transport/grpc/server.go +++ b/services/wallet-service/internal/transport/grpc/server.go @@ -621,6 +621,9 @@ func vipRewardItemsToProto(items []ledger.VipRewardItem) []*walletv1.VipRewardIt Name: item.Name, Quantity: item.Quantity, ExpiresAtMs: item.ExpiresAtMS, + AssetUrl: item.AssetURL, + PreviewUrl: item.PreviewURL, + AnimationUrl: item.AnimationURL, }) } return resp diff --git a/tools/im-c2c-broadcast/main.go b/tools/im-c2c-broadcast/main.go new file mode 100644 index 00000000..1051e13c --- /dev/null +++ b/tools/im-c2c-broadcast/main.go @@ -0,0 +1,354 @@ +package main + +import ( + "bytes" + "context" + "crypto/rand" + "database/sql" + "encoding/binary" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "net/http" + "net/url" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + _ "github.com/go-sql-driver/mysql" + "hyapp/pkg/configx" + "hyapp/pkg/tencentim" +) + +const ( + sendC2CMsgCommand = "v4/openim/sendmsg" + confirmSendValue = "SEND_REAL_TENCENT_IM_C2C" +) + +type serviceConfig struct { + MySQLDSN string `yaml:"mysql_dsn"` + TencentIM tencentIMConfig `yaml:"tencent_im"` +} + +type tencentIMConfig struct { + Enabled bool `yaml:"enabled"` + SDKAppID int64 `yaml:"sdk_app_id"` + SecretKey string `yaml:"secret_key"` + AdminIdentifier string `yaml:"admin_identifier"` + Endpoint string `yaml:"endpoint"` +} + +type sendTextC2CRequest struct { + SyncOtherMachine int `json:"SyncOtherMachine"` + FromAccount string `json:"From_Account,omitempty"` + ToAccount string `json:"To_Account"` + MsgRandom uint32 `json:"MsgRandom"` + MsgBody []messageElement `json:"MsgBody"` +} + +type messageElement struct { + MsgType string `json:"MsgType"` + MsgContent textMsgContent `json:"MsgContent"` +} + +type textMsgContent struct { + Text string `json:"Text"` +} + +type restResponse struct { + ActionStatus string `json:"ActionStatus"` + ErrorInfo string `json:"ErrorInfo"` + ErrorCode int `json:"ErrorCode"` + MsgKey string `json:"MsgKey,omitempty"` +} + +type resultLine struct { + AtMS int64 `json:"at_ms"` + Mode string `json:"mode"` + ToAccount string `json:"to_account"` + Status string `json:"status"` + ErrorCode int `json:"error_code,omitempty"` + ErrorInfo string `json:"error_info,omitempty"` + MsgKey string `json:"msg_key,omitempty"` +} + +func main() { + if err := run(context.Background(), os.Args[1:]); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func run(ctx context.Context, args []string) error { + var userConfigPath string + var imConfigPath string + var fromAccount string + var appCode string + var message string + var outputPath string + var limit int + var sleepMS int + var timeoutSeconds int + var send bool + var confirmSend string + var includeSender bool + + flags := flag.NewFlagSet("im-c2c-broadcast", flag.ContinueOnError) + flags.StringVar(&userConfigPath, "user-config", "services/user-service/configs/config.yaml", "user-service YAML config used for mysql_dsn") + flags.StringVar(&imConfigPath, "im-config", "services/notice-service/configs/config.yaml", "service YAML config used for tencent_im") + flags.StringVar(&fromAccount, "from-account", "123456", "Tencent IM From_Account") + flags.StringVar(&appCode, "app-code", "lalu", "users.app_code filter") + flags.StringVar(&message, "message", "", "plain C2C text message body") + flags.StringVar(&outputPath, "output", "", "JSONL audit output path") + flags.IntVar(&limit, "limit", 0, "maximum active users to process; 0 means all") + flags.IntVar(&sleepMS, "sleep-ms", 25, "sleep between real sends") + flags.IntVar(&timeoutSeconds, "timeout-seconds", 10, "Tencent REST request timeout") + flags.BoolVar(&send, "send", false, "perform real Tencent IM sends; omitted means dry-run") + flags.StringVar(&confirmSend, "confirm-send", "", "must equal "+confirmSendValue+" when --send is set") + flags.BoolVar(&includeSender, "include-sender", false, "include from-account when it is also an active numeric user") + if err := flags.Parse(args); err != nil { + return err + } + + fromAccount = strings.TrimSpace(fromAccount) + appCode = strings.TrimSpace(appCode) + message = strings.TrimSpace(message) + if fromAccount == "" || appCode == "" || message == "" { + return errors.New("from-account, app-code and message are required") + } + if send && confirmSend != confirmSendValue { + return fmt.Errorf("--send requires --confirm-send=%s", confirmSendValue) + } + if sleepMS < 0 || timeoutSeconds <= 0 || limit < 0 { + return errors.New("sleep-ms, timeout-seconds and limit are invalid") + } + + userCfg, err := loadConfig(userConfigPath) + if err != nil { + return fmt.Errorf("load user config: %w", err) + } + imCfg, err := loadConfig(imConfigPath) + if err != nil { + return fmt.Errorf("load im config: %w", err) + } + if strings.TrimSpace(userCfg.MySQLDSN) == "" { + return errors.New("user config mysql_dsn is empty") + } + if err := validateTencentIMConfig(imCfg.TencentIM); err != nil { + return err + } + + db, err := sql.Open("mysql", userCfg.MySQLDSN) + if err != nil { + return err + } + defer db.Close() + if err := db.PingContext(ctx); err != nil { + return fmt.Errorf("ping mysql: %w", err) + } + + targets, totalActive, err := listActiveUserIDs(ctx, db, appCode, fromAccount, includeSender, limit) + if err != nil { + return err + } + mode := "dry-run" + if send { + mode = "send" + } + if outputPath == "" { + outputPath = filepath.Join("run", fmt.Sprintf("im-c2c-broadcast-%s.jsonl", time.Now().UTC().Format("20060102-150405"))) + } + if err := os.MkdirAll(filepath.Dir(outputPath), 0o755); err != nil { + return err + } + output, err := os.Create(outputPath) + if err != nil { + return err + } + defer output.Close() + + fmt.Printf("mode=%s app_code=%s sdk_app_id=%d endpoint=%s from_account=%s total_active=%d targets=%d output=%s\n", + mode, appCode, imCfg.TencentIM.SDKAppID, firstNonEmpty(imCfg.TencentIM.Endpoint, tencentim.DefaultEndpoint), fromAccount, totalActive, len(targets), outputPath) + if len(targets) > 0 { + fmt.Printf("first_target=%s last_target=%s\n", targets[0], targets[len(targets)-1]) + } + if !send { + for _, target := range targets { + if err := writeResult(output, resultLine{AtMS: time.Now().UTC().UnixMilli(), Mode: mode, ToAccount: target, Status: "dry_run"}); err != nil { + return err + } + } + fmt.Println("dry_run_complete=true") + return nil + } + + client := &http.Client{Timeout: time.Duration(timeoutSeconds) * time.Second} + for i, target := range targets { + response, err := sendTencentText(ctx, client, imCfg.TencentIM, fromAccount, target, message) + line := resultLine{AtMS: time.Now().UTC().UnixMilli(), Mode: mode, ToAccount: target, Status: "ok"} + if err != nil { + line.Status = "failed" + line.ErrorInfo = err.Error() + } else if response.ErrorCode != 0 || !strings.EqualFold(firstNonEmpty(response.ActionStatus, "OK"), "OK") { + line.Status = "failed" + line.ErrorCode = response.ErrorCode + line.ErrorInfo = response.ErrorInfo + } else { + line.MsgKey = response.MsgKey + } + if err := writeResult(output, line); err != nil { + return err + } + if line.Status != "ok" { + fmt.Printf("target=%s status=failed error_code=%d error=%s\n", target, line.ErrorCode, line.ErrorInfo) + } + if sleepMS > 0 && i+1 < len(targets) { + time.Sleep(time.Duration(sleepMS) * time.Millisecond) + } + } + + fmt.Println("send_complete=true") + return nil +} + +func loadConfig(path string) (serviceConfig, error) { + var cfg serviceConfig + err := configx.LoadYAML(strings.TrimSpace(path), &cfg) + return cfg, err +} + +func validateTencentIMConfig(cfg tencentIMConfig) error { + if cfg.SDKAppID <= 0 || strings.TrimSpace(cfg.SecretKey) == "" || strings.TrimSpace(cfg.AdminIdentifier) == "" { + return errors.New("tencent_im sdk_app_id, secret_key and admin_identifier are required") + } + return nil +} + +func listActiveUserIDs(ctx context.Context, db *sql.DB, appCode string, fromAccount string, includeSender bool, limit int) ([]string, int64, error) { + var totalActive int64 + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM users WHERE app_code = ? AND status = 'active'`, appCode).Scan(&totalActive); err != nil { + return nil, 0, err + } + + excludeUserID := int64(0) + if !includeSender { + if value, err := strconv.ParseInt(fromAccount, 10, 64); err == nil && value > 0 { + excludeUserID = value + } + } + query := `SELECT user_id FROM users WHERE app_code = ? AND status = 'active'` + params := []any{appCode} + if excludeUserID > 0 { + query += ` AND user_id <> ?` + params = append(params, excludeUserID) + } + query += ` ORDER BY user_id ASC` + if limit > 0 { + query += ` LIMIT ?` + params = append(params, limit) + } + rows, err := db.QueryContext(ctx, query, params...) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + targets := make([]string, 0) + for rows.Next() { + var userID int64 + if err := rows.Scan(&userID); err != nil { + return nil, 0, err + } + targets = append(targets, strconv.FormatInt(userID, 10)) + } + return targets, totalActive, rows.Err() +} + +func sendTencentText(ctx context.Context, client *http.Client, cfg tencentIMConfig, fromAccount string, toAccount string, text string) (restResponse, error) { + requestBody := sendTextC2CRequest{ + SyncOtherMachine: 2, + FromAccount: fromAccount, + ToAccount: toAccount, + MsgRandom: randomUint32(), + MsgBody: []messageElement{ + {MsgType: "TIMTextElem", MsgContent: textMsgContent{Text: text}}, + }, + } + body, err := json.Marshal(requestBody) + if err != nil { + return restResponse{}, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint(cfg, sendC2CMsgCommand), bytes.NewReader(body)) + if err != nil { + return restResponse{}, err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := client.Do(req) + if err != nil { + return restResponse{}, err + } + defer resp.Body.Close() + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return restResponse{}, err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return restResponse{}, fmt.Errorf("tencent im http status %d: %s", resp.StatusCode, string(respBody)) + } + var parsed restResponse + if err := json.Unmarshal(respBody, &parsed); err != nil { + return restResponse{}, err + } + return parsed, nil +} + +func endpoint(cfg tencentIMConfig, command string) string { + adminSig, _ := tencentim.GenerateUserSig(tencentim.UserSigConfig{ + SDKAppID: cfg.SDKAppID, + SecretKey: cfg.SecretKey, + TTL: 24 * time.Hour, + }, strings.TrimSpace(cfg.AdminIdentifier), time.Now().UTC()) + + values := url.Values{} + values.Set("sdkappid", strconv.FormatInt(cfg.SDKAppID, 10)) + values.Set("identifier", strings.TrimSpace(cfg.AdminIdentifier)) + values.Set("usersig", adminSig.UserSig) + values.Set("random", strconv.FormatUint(uint64(randomUint32()), 10)) + values.Set("contenttype", "json") + + return "https://" + strings.TrimSuffix(firstNonEmpty(cfg.Endpoint, tencentim.DefaultEndpoint), "/") + "/" + strings.TrimPrefix(command, "/") + "?" + values.Encode() +} + +func writeResult(output *os.File, line resultLine) error { + body, err := json.Marshal(line) + if err != nil { + return err + } + if _, err := output.Write(append(body, '\n')); err != nil { + return err + } + return nil +} + +func randomUint32() uint32 { + var payload [4]byte + if _, err := rand.Read(payload[:]); err != nil { + return uint32(time.Now().UnixNano()) + } + return binary.BigEndian.Uint32(payload[:]) +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + value = strings.TrimSpace(value) + if value != "" { + return value + } + } + return "" +}