user_handler.go 31 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972
  1. package admin
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "log/slog"
  7. "math"
  8. "strconv"
  9. "strings"
  10. "time"
  11. "github.com/Wei-Shaw/sub2api/internal/handler/dto"
  12. "github.com/Wei-Shaw/sub2api/internal/handler/quotaview"
  13. "github.com/Wei-Shaw/sub2api/internal/pkg/response"
  14. "github.com/Wei-Shaw/sub2api/internal/server/middleware"
  15. "github.com/Wei-Shaw/sub2api/internal/service"
  16. "github.com/gin-gonic/gin"
  17. )
  18. // UserWithConcurrency wraps AdminUser with current concurrency info
  19. type UserWithConcurrency struct {
  20. dto.AdminUser
  21. CurrentConcurrency int `json:"current_concurrency"`
  22. }
  23. // UserHandler handles admin user management
  24. type UserHandler struct {
  25. adminService service.AdminService
  26. concurrencyService *service.ConcurrencyService
  27. userPlatformQuotaRepo service.UserPlatformQuotaRepository // T13 admin quota view
  28. billingCache service.BillingCache // T17/T18 缓存失效(PUT/POST 路径)
  29. totpService *service.TotpService // 角色提升为管理员的 step-up 门控
  30. userService *service.UserService
  31. settingService *service.SettingService // step-up 功能开关
  32. }
  33. // NewUserHandler creates a new admin user handler
  34. func NewUserHandler(
  35. adminService service.AdminService,
  36. concurrencyService *service.ConcurrencyService,
  37. userPlatformQuotaRepo service.UserPlatformQuotaRepository,
  38. billingCache service.BillingCache,
  39. totpService *service.TotpService,
  40. userService *service.UserService,
  41. settingService *service.SettingService,
  42. ) *UserHandler {
  43. return &UserHandler{
  44. adminService: adminService,
  45. concurrencyService: concurrencyService,
  46. userPlatformQuotaRepo: userPlatformQuotaRepo,
  47. billingCache: billingCache,
  48. totpService: totpService,
  49. userService: userService,
  50. settingService: settingService,
  51. }
  52. }
  53. // CreateUserRequest represents admin create user request
  54. type CreateUserRequest struct {
  55. Email string `json:"email" binding:"required,email"`
  56. Password string `json:"password" binding:"required,min=6"`
  57. Username string `json:"username"`
  58. Notes string `json:"notes"`
  59. Role string `json:"role" binding:"omitempty,oneof=super_admin admin user"`
  60. Balance *float64 `json:"balance"`
  61. Concurrency int `json:"concurrency"`
  62. RPMLimit int `json:"rpm_limit"`
  63. AllowedGroups []int64 `json:"allowed_groups"`
  64. RestrictPublicGroups bool `json:"restrict_public_groups"`
  65. }
  66. // UpdateUserRequest represents admin update user request
  67. // 使用指针类型来区分"未提供"和"设置为0"
  68. type UpdateUserRequest struct {
  69. Email string `json:"email" binding:"omitempty,email"`
  70. Password string `json:"password" binding:"omitempty,min=6"`
  71. Username *string `json:"username"`
  72. Notes *string `json:"notes"`
  73. Role string `json:"role" binding:"omitempty,oneof=super_admin admin user"`
  74. Balance *float64 `json:"balance"`
  75. Concurrency *int `json:"concurrency"`
  76. RPMLimit *int `json:"rpm_limit"`
  77. Status string `json:"status" binding:"omitempty,oneof=active disabled"`
  78. AllowedGroups *[]int64 `json:"allowed_groups"`
  79. RestrictPublicGroups *bool `json:"restrict_public_groups"`
  80. // GroupRates 用户专属分组倍率配置
  81. // map[groupID]*rate,nil 表示删除该分组的专属倍率
  82. GroupRates map[int64]*float64 `json:"group_rates"`
  83. }
  84. // UpdateBalanceRequest represents balance update request
  85. type UpdateBalanceRequest struct {
  86. Balance float64 `json:"balance" binding:"required,gt=0"`
  87. Operation string `json:"operation" binding:"required,oneof=set add subtract"`
  88. Notes string `json:"notes"`
  89. }
  90. type BindUserAuthIdentityRequest struct {
  91. ProviderType string `json:"provider_type"`
  92. ProviderKey string `json:"provider_key"`
  93. ProviderSubject string `json:"provider_subject"`
  94. Issuer *string `json:"issuer"`
  95. Metadata map[string]any `json:"metadata"`
  96. Channel *BindUserAuthIdentityChannelRequest `json:"channel"`
  97. }
  98. type BindUserAuthIdentityChannelRequest struct {
  99. Channel string `json:"channel"`
  100. ChannelAppID string `json:"channel_app_id"`
  101. ChannelSubject string `json:"channel_subject"`
  102. Metadata map[string]any `json:"metadata"`
  103. }
  104. // List handles listing all users with pagination
  105. // GET /api/v1/admin/users
  106. // Query params:
  107. // - status: filter by user status
  108. // - role: filter by user role
  109. // - search: search in email, username
  110. // - attr[{id}]: filter by custom attribute value, e.g. attr[1]=company
  111. // - group_name: fuzzy filter by allowed group name
  112. // - api_key_group_id: filter by the exact group bound to the user's API keys
  113. func (h *UserHandler) List(c *gin.Context) {
  114. page, pageSize := response.ParsePagination(c)
  115. search := c.Query("search")
  116. // 标准化和验证 search 参数
  117. search = strings.TrimSpace(search)
  118. if runes := []rune(search); len(runes) > 100 {
  119. search = string(runes[:100])
  120. }
  121. filters := service.UserListFilters{
  122. Status: c.Query("status"),
  123. Role: c.Query("role"),
  124. Search: search,
  125. GroupName: strings.TrimSpace(c.Query("group_name")),
  126. Attributes: parseAttributeFilters(c),
  127. }
  128. if raw := strings.TrimSpace(c.Query("api_key_group_id")); raw != "" {
  129. if id, parseErr := strconv.ParseInt(raw, 10, 64); parseErr == nil && id > 0 {
  130. filters.APIKeyGroupID = id
  131. }
  132. }
  133. sortBy := c.DefaultQuery("sort_by", "created_at")
  134. sortOrder := c.DefaultQuery("sort_order", "desc")
  135. if raw, ok := c.GetQuery("include_subscriptions"); ok {
  136. includeSubscriptions := parseBoolQueryWithDefault(raw, true)
  137. filters.IncludeSubscriptions = &includeSubscriptions
  138. }
  139. users, total, err := h.adminService.ListUsers(c.Request.Context(), page, pageSize, filters, sortBy, sortOrder)
  140. if err != nil {
  141. response.ErrorFrom(c, err)
  142. return
  143. }
  144. // Batch get current concurrency (nil map if unavailable)
  145. var loadInfo map[int64]*service.UserLoadInfo
  146. if len(users) > 0 && h.concurrencyService != nil {
  147. usersConcurrency := make([]service.UserWithConcurrency, len(users))
  148. for i := range users {
  149. usersConcurrency[i] = service.UserWithConcurrency{
  150. ID: users[i].ID,
  151. MaxConcurrency: users[i].Concurrency,
  152. }
  153. }
  154. loadInfo, _ = h.concurrencyService.GetUsersLoadBatch(c.Request.Context(), usersConcurrency)
  155. }
  156. // Build response with concurrency info
  157. out := make([]UserWithConcurrency, len(users))
  158. for i := range users {
  159. out[i] = UserWithConcurrency{
  160. AdminUser: *dto.UserFromServiceAdmin(&users[i]),
  161. }
  162. if info := loadInfo[users[i].ID]; info != nil {
  163. out[i].CurrentConcurrency = info.CurrentConcurrency
  164. }
  165. }
  166. response.Paginated(c, out, total, page, pageSize)
  167. }
  168. // parseAttributeFilters extracts attribute filters from query params
  169. // Format: attr[{attributeID}]=value, e.g. attr[1]=company&attr[2]=developer
  170. func parseAttributeFilters(c *gin.Context) map[int64]string {
  171. result := make(map[int64]string)
  172. // Get all query params and look for attr[*] pattern
  173. for key, values := range c.Request.URL.Query() {
  174. if len(values) == 0 || values[0] == "" {
  175. continue
  176. }
  177. // Check if key matches pattern attr[{id}]
  178. if len(key) > 5 && key[:5] == "attr[" && key[len(key)-1] == ']' {
  179. idStr := key[5 : len(key)-1]
  180. id, err := strconv.ParseInt(idStr, 10, 64)
  181. if err == nil && id > 0 {
  182. result[id] = values[0]
  183. }
  184. }
  185. }
  186. return result
  187. }
  188. // GetByID handles getting a user by ID
  189. // GET /api/v1/admin/users/:id
  190. func (h *UserHandler) GetByID(c *gin.Context) {
  191. userID, err := strconv.ParseInt(c.Param("id"), 10, 64)
  192. if err != nil {
  193. response.BadRequest(c, "Invalid user ID")
  194. return
  195. }
  196. var user *service.User
  197. if c.Query("include_deleted") == "true" {
  198. user, err = h.adminService.GetUserIncludeDeleted(c.Request.Context(), userID)
  199. } else {
  200. user, err = h.adminService.GetUser(c.Request.Context(), userID)
  201. }
  202. if err != nil {
  203. response.ErrorFrom(c, err)
  204. return
  205. }
  206. response.Success(c, dto.UserFromServiceAdmin(user))
  207. }
  208. // BindAuthIdentity manually binds a canonical auth identity to a user.
  209. // POST /api/v1/admin/users/:id/auth-identities
  210. func (h *UserHandler) BindAuthIdentity(c *gin.Context) {
  211. userID, err := strconv.ParseInt(c.Param("id"), 10, 64)
  212. if err != nil {
  213. response.BadRequest(c, "Invalid user ID")
  214. return
  215. }
  216. var req BindUserAuthIdentityRequest
  217. if err := c.ShouldBindJSON(&req); err != nil {
  218. response.BadRequest(c, "Invalid request: "+err.Error())
  219. return
  220. }
  221. input := service.AdminBindAuthIdentityInput{
  222. ProviderType: req.ProviderType,
  223. ProviderKey: req.ProviderKey,
  224. ProviderSubject: req.ProviderSubject,
  225. Issuer: req.Issuer,
  226. Metadata: req.Metadata,
  227. }
  228. if req.Channel != nil {
  229. input.Channel = &service.AdminBindAuthIdentityChannelInput{
  230. Channel: req.Channel.Channel,
  231. ChannelAppID: req.Channel.ChannelAppID,
  232. ChannelSubject: req.Channel.ChannelSubject,
  233. Metadata: req.Channel.Metadata,
  234. }
  235. }
  236. result, err := h.adminService.BindUserAuthIdentity(c.Request.Context(), userID, input)
  237. if err != nil {
  238. response.ErrorFrom(c, err)
  239. return
  240. }
  241. response.Success(c, result)
  242. }
  243. // Create handles creating a new user
  244. // POST /api/v1/admin/users
  245. func (h *UserHandler) Create(c *gin.Context) {
  246. var req CreateUserRequest
  247. if err := c.ShouldBindJSON(&req); err != nil {
  248. response.BadRequest(c, "Invalid request: "+err.Error())
  249. return
  250. }
  251. // 创建管理员账号属权限敏感操作:需最近完成 step-up 2FA 验证。
  252. if service.IsAdminRole(req.Role) {
  253. if !middleware.EnforceStepUp(c, h.totpService, h.userService, h.settingService) {
  254. return
  255. }
  256. }
  257. user, err := h.adminService.CreateUser(c.Request.Context(), &service.CreateUserInput{
  258. Email: req.Email,
  259. Password: req.Password,
  260. Username: req.Username,
  261. Notes: req.Notes,
  262. Role: req.Role,
  263. Balance: req.Balance,
  264. Concurrency: req.Concurrency,
  265. RPMLimit: req.RPMLimit,
  266. AllowedGroups: req.AllowedGroups,
  267. RestrictPublicGroups: req.RestrictPublicGroups,
  268. ActorAdminID: getAdminIDFromContext(c),
  269. })
  270. if err != nil {
  271. response.ErrorFrom(c, err)
  272. return
  273. }
  274. response.Success(c, dto.UserFromServiceAdmin(user))
  275. }
  276. // Update handles updating a user
  277. // PUT /api/v1/admin/users/:id
  278. func (h *UserHandler) Update(c *gin.Context) {
  279. userID, err := strconv.ParseInt(c.Param("id"), 10, 64)
  280. if err != nil {
  281. response.BadRequest(c, "Invalid user ID")
  282. return
  283. }
  284. var req UpdateUserRequest
  285. if err := c.ShouldBindJSON(&req); err != nil {
  286. response.BadRequest(c, "Invalid request: "+err.Error())
  287. return
  288. }
  289. // 防锁死保护:管理员不能把自己降级为普通用户(单管理员场景下会失去后台访问权)。
  290. // 与既有"不能禁用/删除 admin"保护一致。降级其他管理员仍然允许。
  291. if req.Role == service.RoleUser && userID == getAdminIDFromContext(c) {
  292. response.BadRequest(c, "cannot demote yourself from admin")
  293. return
  294. }
  295. // 把普通用户提升为管理员属权限敏感操作:需最近完成 step-up 2FA 验证。
  296. // 目标已是管理员时(前端编辑表单总是携带 role)不触发,避免日常编辑被打断。
  297. if service.IsAdminRole(req.Role) {
  298. target, err := h.adminService.GetUser(c.Request.Context(), userID)
  299. if err != nil {
  300. response.ErrorFrom(c, err)
  301. return
  302. }
  303. if target.Role != req.Role {
  304. if !middleware.EnforceStepUp(c, h.totpService, h.userService, h.settingService) {
  305. return
  306. }
  307. }
  308. }
  309. // 使用指针类型直接传递,nil 表示未提供该字段
  310. user, err := h.adminService.UpdateUser(c.Request.Context(), userID, &service.UpdateUserInput{
  311. Email: req.Email,
  312. Password: req.Password,
  313. Username: req.Username,
  314. Notes: req.Notes,
  315. Role: req.Role,
  316. Balance: req.Balance,
  317. Concurrency: req.Concurrency,
  318. RPMLimit: req.RPMLimit,
  319. Status: req.Status,
  320. AllowedGroups: req.AllowedGroups,
  321. RestrictPublicGroups: req.RestrictPublicGroups,
  322. GroupRates: req.GroupRates,
  323. ActorAdminID: getAdminIDFromContext(c),
  324. })
  325. if err != nil {
  326. response.ErrorFrom(c, err)
  327. return
  328. }
  329. response.Success(c, dto.UserFromServiceAdmin(user))
  330. }
  331. // Delete handles deleting a user
  332. // DELETE /api/v1/admin/users/:id
  333. func (h *UserHandler) Delete(c *gin.Context) {
  334. userID, err := strconv.ParseInt(c.Param("id"), 10, 64)
  335. if err != nil {
  336. response.BadRequest(c, "Invalid user ID")
  337. return
  338. }
  339. err = h.adminService.DeleteUser(c.Request.Context(), userID)
  340. if err != nil {
  341. response.ErrorFrom(c, err)
  342. return
  343. }
  344. response.Success(c, gin.H{"message": "User deleted successfully"})
  345. }
  346. // UpdateBalance handles updating user balance
  347. // POST /api/v1/admin/users/:id/balance
  348. func (h *UserHandler) UpdateBalance(c *gin.Context) {
  349. userID, err := strconv.ParseInt(c.Param("id"), 10, 64)
  350. if err != nil {
  351. response.BadRequest(c, "Invalid user ID")
  352. return
  353. }
  354. var req UpdateBalanceRequest
  355. if err := c.ShouldBindJSON(&req); err != nil {
  356. response.BadRequest(c, "Invalid request: "+err.Error())
  357. return
  358. }
  359. idempotencyPayload := struct {
  360. UserID int64 `json:"user_id"`
  361. Body UpdateBalanceRequest `json:"body"`
  362. }{
  363. UserID: userID,
  364. Body: req,
  365. }
  366. executeAdminIdempotentJSON(c, "admin.users.balance.update", idempotencyPayload, service.DefaultWriteIdempotencyTTL(), func(ctx context.Context) (any, error) {
  367. user, execErr := h.adminService.UpdateUserBalance(ctx, userID, req.Balance, req.Operation, req.Notes)
  368. if execErr != nil {
  369. return nil, execErr
  370. }
  371. return dto.UserFromServiceAdmin(user), nil
  372. })
  373. }
  374. // GetUserAPIKeys handles getting user's API keys
  375. // GET /api/v1/admin/users/:id/api-keys
  376. func (h *UserHandler) GetUserAPIKeys(c *gin.Context) {
  377. userID, err := strconv.ParseInt(c.Param("id"), 10, 64)
  378. if err != nil {
  379. response.BadRequest(c, "Invalid user ID")
  380. return
  381. }
  382. page, pageSize := response.ParsePagination(c)
  383. sortBy := c.DefaultQuery("sort_by", "created_at")
  384. sortOrder := c.DefaultQuery("sort_order", "desc")
  385. keys, total, err := h.adminService.GetUserAPIKeys(c.Request.Context(), userID, page, pageSize, sortBy, sortOrder)
  386. if err != nil {
  387. response.ErrorFrom(c, err)
  388. return
  389. }
  390. out := make([]dto.APIKey, 0, len(keys))
  391. for i := range keys {
  392. out = append(out, *dto.APIKeyFromService(&keys[i]))
  393. }
  394. response.Paginated(c, out, total, page, pageSize)
  395. }
  396. // GetUserUsage handles getting user's usage statistics
  397. // GET /api/v1/admin/users/:id/usage
  398. func (h *UserHandler) GetUserUsage(c *gin.Context) {
  399. userID, err := strconv.ParseInt(c.Param("id"), 10, 64)
  400. if err != nil {
  401. response.BadRequest(c, "Invalid user ID")
  402. return
  403. }
  404. period := c.DefaultQuery("period", "month")
  405. stats, err := h.adminService.GetUserUsageStats(c.Request.Context(), userID, period)
  406. if err != nil {
  407. response.ErrorFrom(c, err)
  408. return
  409. }
  410. response.Success(c, stats)
  411. }
  412. // GetBalanceHistory handles getting user's balance/concurrency change history
  413. // GET /api/v1/admin/users/:id/balance-history
  414. // Query params:
  415. // - type: filter by record type (balance, affiliate_balance, admin_balance, concurrency, admin_concurrency, subscription)
  416. func (h *UserHandler) GetBalanceHistory(c *gin.Context) {
  417. userID, err := strconv.ParseInt(c.Param("id"), 10, 64)
  418. if err != nil {
  419. response.BadRequest(c, "Invalid user ID")
  420. return
  421. }
  422. page, pageSize := response.ParsePagination(c)
  423. codeType := c.Query("type")
  424. codes, total, totalRecharged, err := h.adminService.GetUserBalanceHistory(c.Request.Context(), userID, page, pageSize, codeType)
  425. if err != nil {
  426. response.ErrorFrom(c, err)
  427. return
  428. }
  429. // Convert to admin DTO (includes notes field for admin visibility)
  430. out := make([]dto.AdminRedeemCode, 0, len(codes))
  431. for i := range codes {
  432. out = append(out, *dto.RedeemCodeFromServiceAdmin(&codes[i]))
  433. }
  434. // Custom response with total_recharged alongside pagination
  435. pages := int((total + int64(pageSize) - 1) / int64(pageSize))
  436. if pages < 1 {
  437. pages = 1
  438. }
  439. response.Success(c, gin.H{
  440. "items": out,
  441. "total": total,
  442. "page": page,
  443. "page_size": pageSize,
  444. "pages": pages,
  445. "total_recharged": totalRecharged,
  446. })
  447. }
  448. // ReplaceGroupRequest represents the request to replace a user's exclusive group
  449. type ReplaceGroupRequest struct {
  450. OldGroupID int64 `json:"old_group_id" binding:"required,gt=0"`
  451. NewGroupID int64 `json:"new_group_id" binding:"required,gt=0"`
  452. }
  453. // ReplaceGroup handles replacing a user's exclusive group
  454. // POST /api/v1/admin/users/:id/replace-group
  455. func (h *UserHandler) ReplaceGroup(c *gin.Context) {
  456. userID, err := strconv.ParseInt(c.Param("id"), 10, 64)
  457. if err != nil {
  458. response.BadRequest(c, "Invalid user ID")
  459. return
  460. }
  461. var req ReplaceGroupRequest
  462. if err := c.ShouldBindJSON(&req); err != nil {
  463. response.BadRequest(c, "Invalid request: "+err.Error())
  464. return
  465. }
  466. result, err := h.adminService.ReplaceUserGroup(c.Request.Context(), userID, req.OldGroupID, req.NewGroupID)
  467. if err != nil {
  468. response.ErrorFrom(c, err)
  469. return
  470. }
  471. response.Success(c, gin.H{
  472. "migrated_keys": result.MigratedKeys,
  473. })
  474. }
  475. // GetUserRPMStatus 返回指定用户当前分钟的 RPM 用量
  476. // GET /api/v1/admin/users/:id/rpm-status
  477. func (h *UserHandler) GetUserRPMStatus(c *gin.Context) {
  478. userID, err := strconv.ParseInt(c.Param("id"), 10, 64)
  479. if err != nil {
  480. response.BadRequest(c, "Invalid user ID")
  481. return
  482. }
  483. status, err := h.adminService.GetUserRPMStatus(c.Request.Context(), userID)
  484. if err != nil {
  485. response.ErrorFrom(c, err)
  486. return
  487. }
  488. response.Success(c, status)
  489. }
  490. // BatchUpdateConcurrency 批量修改用户并发数
  491. // POST /api/v1/admin/users/batch-concurrency
  492. type BatchUpdateConcurrencyRequest struct {
  493. UserIDs []int64 `json:"user_ids"`
  494. All bool `json:"all"`
  495. Concurrency int `json:"concurrency"`
  496. Mode string `json:"mode" binding:"required,oneof=set add"`
  497. }
  498. func (h *UserHandler) BatchUpdateConcurrency(c *gin.Context) {
  499. var req BatchUpdateConcurrencyRequest
  500. if err := c.ShouldBindJSON(&req); err != nil {
  501. response.BadRequest(c, "Invalid request: "+err.Error())
  502. return
  503. }
  504. if !req.All && len(req.UserIDs) == 0 {
  505. response.BadRequest(c, "user_ids is required unless all=true")
  506. return
  507. }
  508. if len(req.UserIDs) > 500 {
  509. response.BadRequest(c, "user_ids cannot exceed 500")
  510. return
  511. }
  512. var userIDs []int64
  513. if req.All {
  514. // Fetch all user IDs via pagination
  515. page := 1
  516. const pageSize = 500
  517. for {
  518. users, _, err := h.adminService.ListUsers(c.Request.Context(), page, pageSize, service.UserListFilters{}, "id", "asc")
  519. if err != nil {
  520. response.ErrorFrom(c, err)
  521. return
  522. }
  523. for _, u := range users {
  524. userIDs = append(userIDs, u.ID)
  525. }
  526. if len(users) < pageSize {
  527. break
  528. }
  529. page++
  530. }
  531. } else {
  532. userIDs = req.UserIDs
  533. }
  534. if len(userIDs) == 0 {
  535. response.Success(c, gin.H{"affected": 0})
  536. return
  537. }
  538. affected, err := h.adminService.BatchUpdateConcurrency(c.Request.Context(), userIDs, req.Concurrency, req.Mode)
  539. if err != nil {
  540. response.ErrorFrom(c, err)
  541. return
  542. }
  543. response.Success(c, gin.H{"affected": affected})
  544. }
  545. // BatchUpdateLimits overwrites concurrency and/or RPM limits for multiple users.
  546. // POST /api/v1/admin/users/batch-limits
  547. type BatchUpdateLimitsRequest struct {
  548. UserIDs []int64 `json:"user_ids"`
  549. All bool `json:"all"`
  550. Concurrency *int `json:"concurrency" binding:"omitempty,min=0"`
  551. RPMLimit *int `json:"rpm_limit" binding:"omitempty,min=0"`
  552. }
  553. func (h *UserHandler) BatchUpdateLimits(c *gin.Context) {
  554. var req BatchUpdateLimitsRequest
  555. if err := c.ShouldBindJSON(&req); err != nil {
  556. response.BadRequest(c, "Invalid request: "+err.Error())
  557. return
  558. }
  559. if req.Concurrency == nil && req.RPMLimit == nil {
  560. response.BadRequest(c, "at least one of concurrency or rpm_limit is required")
  561. return
  562. }
  563. if !req.All && len(req.UserIDs) == 0 {
  564. response.BadRequest(c, "user_ids is required unless all=true")
  565. return
  566. }
  567. if !req.All && len(req.UserIDs) > 500 {
  568. response.BadRequest(c, "user_ids cannot exceed 500")
  569. return
  570. }
  571. userIDs := req.UserIDs
  572. if req.All {
  573. userIDs = nil
  574. page := 1
  575. const pageSize = 500
  576. for {
  577. users, _, err := h.adminService.ListUsers(c.Request.Context(), page, pageSize, service.UserListFilters{}, "id", "asc")
  578. if err != nil {
  579. response.ErrorFrom(c, err)
  580. return
  581. }
  582. for _, user := range users {
  583. userIDs = append(userIDs, user.ID)
  584. }
  585. if len(users) < pageSize {
  586. break
  587. }
  588. page++
  589. }
  590. }
  591. if len(userIDs) == 0 {
  592. response.Success(c, gin.H{"affected": 0})
  593. return
  594. }
  595. affected, err := h.adminService.BatchUpdateLimits(
  596. c.Request.Context(),
  597. userIDs,
  598. req.Concurrency,
  599. req.RPMLimit,
  600. )
  601. if err != nil {
  602. response.ErrorFrom(c, err)
  603. return
  604. }
  605. response.Success(c, gin.H{"affected": affected})
  606. }
  607. // GetUserPlatformQuotas GET /admin/users/:id/platform-quotas
  608. // admin 视角:D14 lazy 归零 + 暴露 *_window_start 调试字段
  609. func (h *UserHandler) GetUserPlatformQuotas(c *gin.Context) {
  610. idStr := c.Param("id")
  611. userID, err := strconv.ParseInt(idStr, 10, 64)
  612. if err != nil {
  613. response.BadRequest(c, "invalid user id")
  614. return
  615. }
  616. if h.userPlatformQuotaRepo == nil {
  617. response.Success(c, map[string]any{"platform_quotas": []any{}})
  618. return
  619. }
  620. // 校验用户存在:与 PUT/POST 路径一致,不存在返回 404 而非空数组(避免 admin 界面误判用户存在)。
  621. if _, err := h.adminService.GetUser(c.Request.Context(), userID); err != nil {
  622. response.ErrorFrom(c, err)
  623. return
  624. }
  625. records, err := h.userPlatformQuotaRepo.ListByUser(c.Request.Context(), userID)
  626. if err != nil {
  627. response.ErrorFrom(c, err)
  628. return
  629. }
  630. now := time.Now().UTC()
  631. out := make([]map[string]any, 0, len(records))
  632. for _, r := range records {
  633. out = append(out, quotaview.LazyZeroQuotaForResponse(r, now, true)) // true = 暴露 window_start
  634. }
  635. response.Success(c, map[string]any{"platform_quotas": out})
  636. }
  637. // UpdateUserPlatformQuotasRequest is the body for PUT /admin/users/:id/platform-quotas.
  638. type UpdateUserPlatformQuotasRequest struct {
  639. Quotas []PlatformQuotaInput `json:"quotas" binding:"required"`
  640. }
  641. // PlatformQuotaInput 单平台限额输入;limit 字段为 nil 表示不限制。
  642. type PlatformQuotaInput struct {
  643. Platform string `json:"platform" binding:"required"`
  644. DailyLimitUSD *float64 `json:"daily_limit_usd"`
  645. WeeklyLimitUSD *float64 `json:"weekly_limit_usd"`
  646. MonthlyLimitUSD *float64 `json:"monthly_limit_usd"`
  647. }
  648. // platform 合法性由 service.IsAllowedQuotaPlatform / service.AllowedQuotaPlatforms 统一判断(单一源)。
  649. // UpdateUserPlatformQuotas PUT /admin/users/:id/platform-quotas
  650. // 全量替换该用户所有平台限额。
  651. func (h *UserHandler) UpdateUserPlatformQuotas(c *gin.Context) {
  652. if h.userPlatformQuotaRepo == nil {
  653. response.Error(c, 503, "platform quota service not available")
  654. return
  655. }
  656. userID, err := strconv.ParseInt(c.Param("id"), 10, 64)
  657. if err != nil {
  658. response.BadRequest(c, "Invalid user ID")
  659. return
  660. }
  661. var req UpdateUserPlatformQuotasRequest
  662. if err := c.ShouldBindJSON(&req); err != nil {
  663. response.BadRequest(c, "Invalid request: "+err.Error())
  664. return
  665. }
  666. if len(req.Quotas) > len(service.AllowedQuotaPlatforms) {
  667. response.BadRequest(c, fmt.Sprintf("quotas length must be <= %d", len(service.AllowedQuotaPlatforms)))
  668. return
  669. }
  670. seen := make(map[string]struct{}, len(req.Quotas))
  671. for _, q := range req.Quotas {
  672. if !service.IsAllowedQuotaPlatform(q.Platform) {
  673. response.BadRequest(c, "invalid platform: "+q.Platform)
  674. return
  675. }
  676. if _, dup := seen[q.Platform]; dup {
  677. response.BadRequest(c, "duplicate platform: "+q.Platform)
  678. return
  679. }
  680. seen[q.Platform] = struct{}{}
  681. // daily_limit_usd / weekly_limit_usd / monthly_limit_usd 的语义:
  682. // nil / not set → 无限额(完全放行)
  683. // 0 → 完全禁用(任何请求都会被拒绝,因为 usage >= 0 恒成立)
  684. // > 0 → USD 限额上限
  685. // 拦截 NaN / ±Inf:客户端可发送超大数(如 1e308 × 2)使 JSON 反序列化得到 +Inf,
  686. // 进入 DB 后 cache check 中 usage >= limit 永不成立,limit 等同失效。
  687. for _, f := range []struct {
  688. name string
  689. val *float64
  690. }{
  691. {"daily_limit_usd", q.DailyLimitUSD},
  692. {"weekly_limit_usd", q.WeeklyLimitUSD},
  693. {"monthly_limit_usd", q.MonthlyLimitUSD},
  694. } {
  695. if f.val == nil {
  696. continue
  697. }
  698. v := *f.val
  699. if v < 0 {
  700. response.BadRequest(c, f.name+" must be >= 0")
  701. return
  702. }
  703. if math.IsNaN(v) || math.IsInf(v, 0) {
  704. response.BadRequest(c, f.name+" must be a finite number")
  705. return
  706. }
  707. }
  708. }
  709. records := make([]service.UserPlatformQuotaRecord, 0, len(req.Quotas))
  710. for _, q := range req.Quotas {
  711. records = append(records, service.UserPlatformQuotaRecord{
  712. UserID: userID,
  713. Platform: q.Platform,
  714. DailyLimitUSD: q.DailyLimitUSD,
  715. WeeklyLimitUSD: q.WeeklyLimitUSD,
  716. MonthlyLimitUSD: q.MonthlyLimitUSD,
  717. })
  718. }
  719. ctx := c.Request.Context()
  720. // 校验用户是否存在,避免 FK 违反导致 500;用户不存在时返回 404。
  721. if _, err := h.adminService.GetUser(ctx, userID); err != nil {
  722. response.ErrorFrom(c, err)
  723. return
  724. }
  725. // 在 UpsertForUser 之前抓取 before snapshot 用于审计 before/after 对比。
  726. // ListByUser 失败不阻断主操作(best-effort),仅记录降级 warn。
  727. beforeRecords, beforeErr := h.userPlatformQuotaRepo.ListByUser(ctx, userID)
  728. if beforeErr != nil {
  729. slog.Warn("quota audit before snapshot failed", "user_id", userID, "err", beforeErr)
  730. }
  731. if err := h.userPlatformQuotaRepo.UpsertForUser(ctx, userID, records); err != nil {
  732. response.ErrorFrom(c, err)
  733. return
  734. }
  735. beforeByPlatform := make(map[string]service.UserPlatformQuotaRecord, len(beforeRecords))
  736. for _, r := range beforeRecords {
  737. beforeByPlatform[r.Platform] = r
  738. }
  739. afterPlatforms := make(map[string]struct{}, len(records))
  740. for _, r := range records {
  741. afterPlatforms[r.Platform] = struct{}{}
  742. }
  743. changes := make([]map[string]any, 0, len(records))
  744. for _, r := range records {
  745. entry := map[string]any{
  746. "platform": r.Platform,
  747. "daily_limit_usd": r.DailyLimitUSD,
  748. "weekly_limit_usd": r.WeeklyLimitUSD,
  749. "monthly_limit_usd": r.MonthlyLimitUSD,
  750. }
  751. if prev, ok := beforeByPlatform[r.Platform]; ok {
  752. entry["before_daily_limit_usd"] = prev.DailyLimitUSD
  753. entry["before_weekly_limit_usd"] = prev.WeeklyLimitUSD
  754. entry["before_monthly_limit_usd"] = prev.MonthlyLimitUSD
  755. }
  756. changes = append(changes, entry)
  757. }
  758. // 补 removed 条目:before 存在但 after 缺失 = 该平台被软删除。
  759. // 缺少这条记录,审计消费方无法察觉"管理员把某平台从配额列表移除"的操作(合规盲区)。
  760. for _, prev := range beforeRecords {
  761. if _, kept := afterPlatforms[prev.Platform]; kept {
  762. continue
  763. }
  764. changes = append(changes, map[string]any{
  765. "platform": prev.Platform,
  766. "removed": true,
  767. "before_daily_limit_usd": prev.DailyLimitUSD,
  768. "before_weekly_limit_usd": prev.WeeklyLimitUSD,
  769. "before_monthly_limit_usd": prev.MonthlyLimitUSD,
  770. })
  771. }
  772. // before_snapshot_available 让审计消费方能识别 changes 中是否带 before_* 字段;
  773. // false 时所有 entry 都会缺失 before_*_limit_usd,仅有 after 视图。
  774. slog.Info("admin.quota_updated",
  775. "actor_admin_id", getAdminIDFromContext(c),
  776. "target_user_id", userID,
  777. "platform_count", len(records),
  778. "before_snapshot_available", beforeErr == nil,
  779. "changes", changes)
  780. // 失效 cache:对全部允许的 platform 统一 invalidate。
  781. // Trade-off:精确失效(仅 req 涉及平台 + 被软删平台)需 upsert 前额外 ListByUser,
  782. // 增加一次 DB 查询和逻辑复杂度。由于 AllowedQuotaPlatforms 数量很少,
  783. // 全量 invalidate 的额外开销可接受,且能可靠覆盖软删除场景。
  784. if h.billingCache != nil {
  785. for _, p := range service.AllowedQuotaPlatforms {
  786. if err := h.billingCache.DeleteUserPlatformQuotaCache(ctx, userID, p); err != nil {
  787. slog.Error("ALERT: quota cache invalidation failed after UpsertForUser; limit 生效可能延迟至 sentinel TTL(最长 1h),需人工确认或重试失效", "user_id", userID, "platform", p, "err", err)
  788. }
  789. }
  790. }
  791. // 返回最新状态
  792. now := time.Now().UTC()
  793. records2, err := h.userPlatformQuotaRepo.ListByUser(ctx, userID)
  794. if err != nil {
  795. response.ErrorFrom(c, err)
  796. return
  797. }
  798. out := make([]map[string]any, 0, len(records2))
  799. for i := range records2 {
  800. out = append(out, quotaview.LazyZeroQuotaForResponse(records2[i], now, true))
  801. }
  802. response.Success(c, map[string]any{"platform_quotas": out})
  803. }
  804. // ResetUserPlatformQuotaWindowRequest is the body for POST /admin/users/:id/platform-quotas/reset.
  805. type ResetUserPlatformQuotaWindowRequest struct {
  806. Platform string `json:"platform" binding:"required"`
  807. Window string `json:"window" binding:"required"`
  808. }
  809. var allowedWindowsForQuotaReset = map[string]struct{}{
  810. "daily": {},
  811. "weekly": {},
  812. "monthly": {},
  813. }
  814. // ResetUserPlatformQuotaWindow POST /admin/users/:id/platform-quotas/reset
  815. // 立即归零指定 (platform, window) 的用量并更新 window_start。
  816. func (h *UserHandler) ResetUserPlatformQuotaWindow(c *gin.Context) {
  817. if h.userPlatformQuotaRepo == nil {
  818. response.Error(c, 503, "platform quota service not available")
  819. return
  820. }
  821. userID, err := strconv.ParseInt(c.Param("id"), 10, 64)
  822. if err != nil {
  823. response.BadRequest(c, "Invalid user ID")
  824. return
  825. }
  826. var req ResetUserPlatformQuotaWindowRequest
  827. if err := c.ShouldBindJSON(&req); err != nil {
  828. response.BadRequest(c, "Invalid request: "+err.Error())
  829. return
  830. }
  831. if !service.IsAllowedQuotaPlatform(req.Platform) {
  832. response.BadRequest(c, "invalid platform: "+req.Platform)
  833. return
  834. }
  835. if _, ok := allowedWindowsForQuotaReset[req.Window]; !ok {
  836. response.BadRequest(c, "invalid window: "+req.Window)
  837. return
  838. }
  839. ctx := c.Request.Context()
  840. // 校验用户是否存在,避免对不存在的用户执行操作返回误导性的 500。
  841. if _, err := h.adminService.GetUser(ctx, userID); err != nil {
  842. response.ErrorFrom(c, err)
  843. return
  844. }
  845. now := time.Now().UTC()
  846. if err := h.userPlatformQuotaRepo.ResetExpiredWindow(ctx, userID, req.Platform, req.Window, now); err != nil {
  847. if errors.Is(err, service.ErrUserPlatformQuotaNotFound) {
  848. response.NotFound(c, "user platform quota not found")
  849. return
  850. }
  851. response.ErrorFrom(c, err)
  852. return
  853. }
  854. slog.Info("admin.quota_window_reset",
  855. "actor_admin_id", getAdminIDFromContext(c),
  856. "target_user_id", userID,
  857. "platform", req.Platform,
  858. "window", req.Window)
  859. if h.billingCache != nil {
  860. if err := h.billingCache.DeleteUserPlatformQuotaCache(ctx, userID, req.Platform); err != nil {
  861. slog.Error("ALERT: quota cache invalidation failed after ResetExpiredWindow; 窗口重置可能延迟至 sentinel TTL(最长 1h)", "user_id", userID, "platform", req.Platform, "err", err)
  862. }
  863. }
  864. records, err := h.userPlatformQuotaRepo.ListByUser(ctx, userID)
  865. if err != nil {
  866. response.ErrorFrom(c, err)
  867. return
  868. }
  869. out := make([]map[string]any, 0, len(records))
  870. for i := range records {
  871. out = append(out, quotaview.LazyZeroQuotaForResponse(records[i], now, true))
  872. }
  873. response.Success(c, map[string]any{"platform_quotas": out})
  874. }