user_subscription_repo.go 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733
  1. package repository
  2. import (
  3. "context"
  4. "time"
  5. dbent "github.com/Wei-Shaw/sub2api/ent"
  6. "github.com/Wei-Shaw/sub2api/ent/group"
  7. "github.com/Wei-Shaw/sub2api/ent/predicate"
  8. "github.com/Wei-Shaw/sub2api/ent/schema/mixins"
  9. "github.com/Wei-Shaw/sub2api/ent/user"
  10. "github.com/Wei-Shaw/sub2api/ent/usersubscription"
  11. "github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
  12. "github.com/Wei-Shaw/sub2api/internal/service"
  13. )
  14. type userSubscriptionRepository struct {
  15. client *dbent.Client
  16. }
  17. func NewUserSubscriptionRepository(client *dbent.Client) service.UserSubscriptionRepository {
  18. return &userSubscriptionRepository{client: client}
  19. }
  20. func (r *userSubscriptionRepository) Create(ctx context.Context, sub *service.UserSubscription) error {
  21. if sub == nil {
  22. return service.ErrSubscriptionNilInput
  23. }
  24. if needsManagedResourceGuard(ctx) {
  25. return withManagedResourceWrite(ctx, r.client, "", 0, sub.UserID, func(next context.Context) error { return r.Create(next, sub) })
  26. }
  27. client := clientFromContext(ctx, r.client)
  28. builder := client.UserSubscription.Create().
  29. SetUserID(sub.UserID).
  30. SetGroupID(sub.GroupID).
  31. SetExpiresAt(sub.ExpiresAt).
  32. SetNillableDailyWindowStart(sub.DailyWindowStart).
  33. SetNillableWeeklyWindowStart(sub.WeeklyWindowStart).
  34. SetNillableMonthlyWindowStart(sub.MonthlyWindowStart).
  35. SetDailyUsageUsd(sub.DailyUsageUSD).
  36. SetWeeklyUsageUsd(sub.WeeklyUsageUSD).
  37. SetMonthlyUsageUsd(sub.MonthlyUsageUSD).
  38. SetNillableAssignedBy(sub.AssignedBy)
  39. if sub.StartsAt.IsZero() {
  40. builder.SetStartsAt(time.Now())
  41. } else {
  42. builder.SetStartsAt(sub.StartsAt)
  43. }
  44. if sub.Status != "" {
  45. builder.SetStatus(sub.Status)
  46. }
  47. if !sub.AssignedAt.IsZero() {
  48. builder.SetAssignedAt(sub.AssignedAt)
  49. }
  50. // Keep compatibility with historical behavior: always store notes as a string value.
  51. builder.SetNotes(sub.Notes)
  52. created, err := builder.Save(ctx)
  53. if err == nil {
  54. applyUserSubscriptionEntityToService(sub, created)
  55. }
  56. return translatePersistenceError(err, nil, service.ErrSubscriptionAlreadyExists)
  57. }
  58. func (r *userSubscriptionRepository) GetByID(ctx context.Context, id int64) (*service.UserSubscription, error) {
  59. client := clientFromContext(ctx, r.client)
  60. m, err := client.UserSubscription.Query().
  61. Where(usersubscription.IDEQ(id)).
  62. WithUser().
  63. WithGroup().
  64. WithAssignedByUser().
  65. Only(ctx)
  66. if err != nil {
  67. return nil, translatePersistenceError(err, service.ErrSubscriptionNotFound, nil)
  68. }
  69. return userSubscriptionEntityToService(m), nil
  70. }
  71. func (r *userSubscriptionRepository) GetByIDForUpdate(ctx context.Context, id int64) (*service.UserSubscription, error) {
  72. if needsManagedResourceGuard(ctx) {
  73. var result *service.UserSubscription
  74. err := withManagedResourceWrite(ctx, r.client, "user_subscriptions", id, 0, func(next context.Context) error {
  75. var err error
  76. result, err = r.GetByIDForUpdate(next, id)
  77. return err
  78. })
  79. return result, err
  80. }
  81. client := clientFromContext(ctx, r.client)
  82. m, err := client.UserSubscription.Query().
  83. Where(usersubscription.IDEQ(id)).
  84. ForUpdate().
  85. Only(ctx)
  86. if err != nil {
  87. return nil, translatePersistenceError(err, service.ErrSubscriptionNotFound, nil)
  88. }
  89. return userSubscriptionEntityToService(m), nil
  90. }
  91. func (r *userSubscriptionRepository) GetByIDIncludeDeleted(ctx context.Context, id int64) (*service.UserSubscription, error) {
  92. client := clientFromContext(ctx, r.client)
  93. queryCtx := mixins.SkipSoftDelete(ctx)
  94. m, err := client.UserSubscription.Query().
  95. Where(usersubscription.IDEQ(id)).
  96. WithUser().
  97. WithGroup().
  98. WithAssignedByUser().
  99. Only(queryCtx)
  100. if err != nil {
  101. return nil, translatePersistenceError(err, service.ErrSubscriptionNotFound, nil)
  102. }
  103. return userSubscriptionEntityToServicePreserveStatus(m), nil
  104. }
  105. func (r *userSubscriptionRepository) GetByUserIDAndGroupID(ctx context.Context, userID, groupID int64) (*service.UserSubscription, error) {
  106. client := clientFromContext(ctx, r.client)
  107. m, err := client.UserSubscription.Query().
  108. Where(usersubscription.UserIDEQ(userID), usersubscription.GroupIDEQ(groupID)).
  109. WithGroup().
  110. Only(ctx)
  111. if err != nil {
  112. return nil, translatePersistenceError(err, service.ErrSubscriptionNotFound, nil)
  113. }
  114. return userSubscriptionEntityToService(m), nil
  115. }
  116. func (r *userSubscriptionRepository) GetActiveByUserIDAndGroupID(ctx context.Context, userID, groupID int64) (*service.UserSubscription, error) {
  117. client := clientFromContext(ctx, r.client)
  118. m, err := client.UserSubscription.Query().
  119. Where(
  120. usersubscription.UserIDEQ(userID),
  121. usersubscription.GroupIDEQ(groupID),
  122. usersubscription.StatusEQ(service.SubscriptionStatusActive),
  123. usersubscription.ExpiresAtGT(time.Now()),
  124. ).
  125. WithGroup().
  126. Only(ctx)
  127. if err != nil {
  128. return nil, translatePersistenceError(err, service.ErrSubscriptionNotFound, nil)
  129. }
  130. return userSubscriptionEntityToService(m), nil
  131. }
  132. func (r *userSubscriptionRepository) Update(ctx context.Context, sub *service.UserSubscription) error {
  133. if sub == nil {
  134. return service.ErrSubscriptionNilInput
  135. }
  136. if needsManagedResourceGuard(ctx) {
  137. return withManagedResourceWrite(ctx, r.client, "user_subscriptions", sub.ID, sub.UserID, func(next context.Context) error { return r.Update(next, sub) })
  138. }
  139. client := clientFromContext(ctx, r.client)
  140. builder := client.UserSubscription.UpdateOneID(sub.ID).
  141. SetUserID(sub.UserID).
  142. SetGroupID(sub.GroupID).
  143. SetStartsAt(sub.StartsAt).
  144. SetExpiresAt(sub.ExpiresAt).
  145. SetStatus(sub.Status).
  146. SetNillableDailyWindowStart(sub.DailyWindowStart).
  147. SetNillableWeeklyWindowStart(sub.WeeklyWindowStart).
  148. SetNillableMonthlyWindowStart(sub.MonthlyWindowStart).
  149. SetDailyUsageUsd(sub.DailyUsageUSD).
  150. SetWeeklyUsageUsd(sub.WeeklyUsageUSD).
  151. SetMonthlyUsageUsd(sub.MonthlyUsageUSD).
  152. SetNillableAssignedBy(sub.AssignedBy).
  153. SetAssignedAt(sub.AssignedAt).
  154. SetNotes(sub.Notes)
  155. updated, err := builder.Save(ctx)
  156. if err == nil {
  157. applyUserSubscriptionEntityToService(sub, updated)
  158. return nil
  159. }
  160. return translatePersistenceError(err, service.ErrSubscriptionNotFound, service.ErrSubscriptionAlreadyExists)
  161. }
  162. func (r *userSubscriptionRepository) Delete(ctx context.Context, id int64) error {
  163. if needsManagedResourceGuard(ctx) {
  164. return withManagedResourceWrite(ctx, r.client, "user_subscriptions", id, 0, func(next context.Context) error { return r.Delete(next, id) })
  165. }
  166. // Match GORM semantics: deleting a missing row is not an error.
  167. client := clientFromContext(ctx, r.client)
  168. _, err := client.UserSubscription.Delete().Where(usersubscription.IDEQ(id)).Exec(ctx)
  169. return err
  170. }
  171. func (r *userSubscriptionRepository) Restore(ctx context.Context, subscriptionID int64, restoredStatus string) (*service.UserSubscription, error) {
  172. if needsManagedResourceGuard(ctx) {
  173. var result *service.UserSubscription
  174. err := withManagedResourceWrite(ctx, r.client, "user_subscriptions", subscriptionID, 0, func(next context.Context) error {
  175. var err error
  176. result, err = r.Restore(next, subscriptionID, restoredStatus)
  177. return err
  178. })
  179. return result, err
  180. }
  181. client := clientFromContext(ctx, r.client)
  182. queryCtx := mixins.SkipSoftDelete(ctx)
  183. _, err := client.UserSubscription.UpdateOneID(subscriptionID).
  184. SetStatus(restoredStatus).
  185. ClearDeletedAt().
  186. SetUpdatedAt(time.Now()).
  187. Save(queryCtx)
  188. if err != nil {
  189. return nil, translatePersistenceError(err, service.ErrSubscriptionNotFound, service.ErrSubscriptionRestoreConflict)
  190. }
  191. return r.GetByID(ctx, subscriptionID)
  192. }
  193. func (r *userSubscriptionRepository) ListByUserID(ctx context.Context, userID int64) ([]service.UserSubscription, error) {
  194. client := clientFromContext(ctx, r.client)
  195. subs, err := client.UserSubscription.Query().
  196. Where(usersubscription.UserIDEQ(userID)).
  197. WithGroup().
  198. Order(dbent.Desc(usersubscription.FieldCreatedAt)).
  199. All(ctx)
  200. if err != nil {
  201. return nil, err
  202. }
  203. return userSubscriptionEntitiesToService(subs), nil
  204. }
  205. func (r *userSubscriptionRepository) ListActiveByUserID(ctx context.Context, userID int64) ([]service.UserSubscription, error) {
  206. client := clientFromContext(ctx, r.client)
  207. subs, err := client.UserSubscription.Query().
  208. Where(
  209. usersubscription.UserIDEQ(userID),
  210. usersubscription.StatusEQ(service.SubscriptionStatusActive),
  211. usersubscription.ExpiresAtGT(time.Now()),
  212. ).
  213. WithGroup().
  214. Order(dbent.Desc(usersubscription.FieldCreatedAt)).
  215. All(ctx)
  216. if err != nil {
  217. return nil, err
  218. }
  219. return userSubscriptionEntitiesToService(subs), nil
  220. }
  221. func (r *userSubscriptionRepository) ListByGroupID(ctx context.Context, groupID int64, params pagination.PaginationParams) ([]service.UserSubscription, *pagination.PaginationResult, error) {
  222. client := clientFromContext(ctx, r.client)
  223. q := client.UserSubscription.Query().Where(usersubscription.GroupIDEQ(groupID))
  224. total, err := q.Clone().Count(ctx)
  225. if err != nil {
  226. return nil, nil, err
  227. }
  228. subs, err := q.
  229. WithUser().
  230. WithGroup().
  231. Order(dbent.Desc(usersubscription.FieldCreatedAt)).
  232. Offset(params.Offset()).
  233. Limit(params.Limit()).
  234. All(ctx)
  235. if err != nil {
  236. return nil, nil, err
  237. }
  238. return userSubscriptionEntitiesToService(subs), paginationResultFromTotal(int64(total), params), nil
  239. }
  240. func (r *userSubscriptionRepository) List(ctx context.Context, params pagination.PaginationParams, userID, groupID *int64, status, platform, sortBy, sortOrder string) ([]service.UserSubscription, *pagination.PaginationResult, error) {
  241. client := clientFromContext(ctx, r.client)
  242. q := client.UserSubscription.Query()
  243. includeSoftDeleted := status == "" || status == service.SubscriptionStatusRevoked
  244. if userID != nil {
  245. q = q.Where(usersubscription.UserIDEQ(*userID))
  246. }
  247. if groupID != nil {
  248. q = q.Where(usersubscription.GroupIDEQ(*groupID))
  249. }
  250. if platform != "" {
  251. groupPredicates := []predicate.Group{group.PlatformEQ(platform)}
  252. if includeSoftDeleted {
  253. groupPredicates = append(groupPredicates, group.DeletedAtIsNil())
  254. }
  255. q = q.Where(usersubscription.HasGroupWith(groupPredicates...))
  256. }
  257. // Status filtering with real-time expiration check
  258. now := time.Now()
  259. switch status {
  260. case service.SubscriptionStatusActive:
  261. // Active: status is active AND not yet expired
  262. q = q.Where(
  263. usersubscription.StatusEQ(service.SubscriptionStatusActive),
  264. usersubscription.ExpiresAtGT(now),
  265. )
  266. case service.SubscriptionStatusExpired:
  267. // Expired: status is expired OR (status is active but already expired)
  268. q = q.Where(
  269. usersubscription.Or(
  270. usersubscription.StatusEQ(service.SubscriptionStatusExpired),
  271. usersubscription.And(
  272. usersubscription.StatusEQ(service.SubscriptionStatusActive),
  273. usersubscription.ExpiresAtLTE(now),
  274. ),
  275. ),
  276. )
  277. case service.SubscriptionStatusRevoked:
  278. // Revoked is a DTO/API display state backed by user_subscriptions.deleted_at.
  279. q = q.Where(usersubscription.DeletedAtNotNil())
  280. case "":
  281. // No filter. Use SkipSoftDelete below so admin "all status" includes revoked history.
  282. default:
  283. // Other persisted status.
  284. q = q.Where(usersubscription.StatusEQ(status))
  285. }
  286. queryCtx := ctx
  287. if includeSoftDeleted {
  288. queryCtx = mixins.SkipSoftDelete(ctx)
  289. }
  290. total, err := q.Clone().Count(queryCtx)
  291. if err != nil {
  292. return nil, nil, err
  293. }
  294. if !includeSoftDeleted {
  295. q = q.WithUser().WithGroup().WithAssignedByUser()
  296. }
  297. // Determine sort field
  298. var field string
  299. switch sortBy {
  300. case "expires_at":
  301. field = usersubscription.FieldExpiresAt
  302. case "status":
  303. field = usersubscription.FieldStatus
  304. default:
  305. field = usersubscription.FieldCreatedAt
  306. }
  307. // Determine sort order (default: desc)
  308. if sortOrder == "asc" && sortBy != "" {
  309. q = q.Order(dbent.Asc(field))
  310. } else {
  311. q = q.Order(dbent.Desc(field))
  312. }
  313. subs, err := q.
  314. Offset(params.Offset()).
  315. Limit(params.Limit()).
  316. All(queryCtx)
  317. if err != nil {
  318. return nil, nil, err
  319. }
  320. result := userSubscriptionEntitiesToService(subs)
  321. if includeSoftDeleted {
  322. if err := r.attachUserSubscriptionRelations(ctx, result); err != nil {
  323. return nil, nil, err
  324. }
  325. }
  326. return result, paginationResultFromTotal(int64(total), params), nil
  327. }
  328. func (r *userSubscriptionRepository) ExistsByUserIDAndGroupID(ctx context.Context, userID, groupID int64) (bool, error) {
  329. client := clientFromContext(ctx, r.client)
  330. return client.UserSubscription.Query().
  331. Where(usersubscription.UserIDEQ(userID), usersubscription.GroupIDEQ(groupID)).
  332. Exist(ctx)
  333. }
  334. func (r *userSubscriptionRepository) ExistsActiveByUserIDAndGroupID(ctx context.Context, userID, groupID int64) (bool, error) {
  335. return r.ExistsByUserIDAndGroupID(ctx, userID, groupID)
  336. }
  337. func (r *userSubscriptionRepository) ExtendExpiry(ctx context.Context, subscriptionID int64, newExpiresAt time.Time) error {
  338. if needsManagedResourceGuard(ctx) {
  339. return withManagedResourceWrite(ctx, r.client, "user_subscriptions", subscriptionID, 0, func(next context.Context) error { return r.ExtendExpiry(next, subscriptionID, newExpiresAt) })
  340. }
  341. client := clientFromContext(ctx, r.client)
  342. _, err := client.UserSubscription.UpdateOneID(subscriptionID).
  343. SetExpiresAt(newExpiresAt).
  344. Save(ctx)
  345. return translatePersistenceError(err, service.ErrSubscriptionNotFound, nil)
  346. }
  347. func (r *userSubscriptionRepository) UpdateStatus(ctx context.Context, subscriptionID int64, status string) error {
  348. if needsManagedResourceGuard(ctx) {
  349. return withManagedResourceWrite(ctx, r.client, "user_subscriptions", subscriptionID, 0, func(next context.Context) error { return r.UpdateStatus(next, subscriptionID, status) })
  350. }
  351. client := clientFromContext(ctx, r.client)
  352. _, err := client.UserSubscription.UpdateOneID(subscriptionID).
  353. SetStatus(status).
  354. Save(ctx)
  355. return translatePersistenceError(err, service.ErrSubscriptionNotFound, nil)
  356. }
  357. func (r *userSubscriptionRepository) UpdateNotes(ctx context.Context, subscriptionID int64, notes string) error {
  358. if needsManagedResourceGuard(ctx) {
  359. return withManagedResourceWrite(ctx, r.client, "user_subscriptions", subscriptionID, 0, func(next context.Context) error { return r.UpdateNotes(next, subscriptionID, notes) })
  360. }
  361. client := clientFromContext(ctx, r.client)
  362. _, err := client.UserSubscription.UpdateOneID(subscriptionID).
  363. SetNotes(notes).
  364. Save(ctx)
  365. return translatePersistenceError(err, service.ErrSubscriptionNotFound, nil)
  366. }
  367. func (r *userSubscriptionRepository) ActivateWindows(ctx context.Context, id int64, dailyStart, periodicStart time.Time) error {
  368. client := clientFromContext(ctx, r.client)
  369. n, err := client.UserSubscription.Update().
  370. Where(
  371. usersubscription.IDEQ(id),
  372. usersubscription.DailyWindowStartIsNil(),
  373. usersubscription.WeeklyWindowStartIsNil(),
  374. usersubscription.MonthlyWindowStartIsNil(),
  375. ).
  376. SetDailyWindowStart(dailyStart).
  377. SetWeeklyWindowStart(periodicStart).
  378. SetMonthlyWindowStart(periodicStart).
  379. Save(ctx)
  380. return r.translateConditionalWindowReset(ctx, client, id, n, err)
  381. }
  382. func (r *userSubscriptionRepository) ResetUsageWindows(ctx context.Context, id int64, resetDaily, resetWeekly, resetMonthly bool, dailyStart, periodicStart time.Time) error {
  383. if needsManagedResourceGuard(ctx) {
  384. return withManagedResourceWrite(ctx, r.client, "user_subscriptions", id, 0, func(next context.Context) error {
  385. return r.ResetUsageWindows(next, id, resetDaily, resetWeekly, resetMonthly, dailyStart, periodicStart)
  386. })
  387. }
  388. client := clientFromContext(ctx, r.client)
  389. update := client.UserSubscription.UpdateOneID(id)
  390. if resetDaily {
  391. update.SetDailyUsageUsd(0).SetDailyWindowStart(dailyStart)
  392. }
  393. if resetWeekly {
  394. update.SetWeeklyUsageUsd(0).SetWeeklyWindowStart(periodicStart)
  395. }
  396. if resetMonthly {
  397. update.SetMonthlyUsageUsd(0).SetMonthlyWindowStart(periodicStart)
  398. }
  399. _, err := update.Save(ctx)
  400. return translatePersistenceError(err, service.ErrSubscriptionNotFound, nil)
  401. }
  402. func (r *userSubscriptionRepository) ResetDailyUsage(ctx context.Context, id int64, expectedWindowStart *time.Time, newWindowStart time.Time) error {
  403. client := clientFromContext(ctx, r.client)
  404. query := client.UserSubscription.Update().Where(usersubscription.IDEQ(id))
  405. if expectedWindowStart == nil {
  406. query = query.Where(usersubscription.DailyWindowStartIsNil())
  407. } else {
  408. query = query.Where(usersubscription.DailyWindowStartEQ(*expectedWindowStart))
  409. }
  410. n, err := query.
  411. SetDailyUsageUsd(0).
  412. SetDailyWindowStart(newWindowStart).
  413. Save(ctx)
  414. return r.translateConditionalWindowReset(ctx, client, id, n, err)
  415. }
  416. func (r *userSubscriptionRepository) ResetWeeklyUsage(ctx context.Context, id int64, expectedWindowStart *time.Time, newWindowStart time.Time) error {
  417. client := clientFromContext(ctx, r.client)
  418. query := client.UserSubscription.Update().Where(usersubscription.IDEQ(id))
  419. if expectedWindowStart == nil {
  420. query = query.Where(usersubscription.WeeklyWindowStartIsNil())
  421. } else {
  422. query = query.Where(usersubscription.WeeklyWindowStartEQ(*expectedWindowStart))
  423. }
  424. n, err := query.
  425. SetWeeklyUsageUsd(0).
  426. SetWeeklyWindowStart(newWindowStart).
  427. Save(ctx)
  428. return r.translateConditionalWindowReset(ctx, client, id, n, err)
  429. }
  430. func (r *userSubscriptionRepository) ResetMonthlyUsage(ctx context.Context, id int64, expectedWindowStart *time.Time, newWindowStart time.Time) error {
  431. client := clientFromContext(ctx, r.client)
  432. query := client.UserSubscription.Update().Where(usersubscription.IDEQ(id))
  433. if expectedWindowStart == nil {
  434. query = query.Where(usersubscription.MonthlyWindowStartIsNil())
  435. } else {
  436. query = query.Where(usersubscription.MonthlyWindowStartEQ(*expectedWindowStart))
  437. }
  438. n, err := query.
  439. SetMonthlyUsageUsd(0).
  440. SetMonthlyWindowStart(newWindowStart).
  441. Save(ctx)
  442. return r.translateConditionalWindowReset(ctx, client, id, n, err)
  443. }
  444. func (r *userSubscriptionRepository) translateConditionalWindowReset(ctx context.Context, client *dbent.Client, id int64, affected int, err error) error {
  445. if err != nil {
  446. return translatePersistenceError(err, service.ErrSubscriptionNotFound, nil)
  447. }
  448. if affected > 0 {
  449. return nil
  450. }
  451. // A stale reset is an expected no-op: another request already advanced the
  452. // window. Preserve not-found semantics for callers that target a missing row.
  453. exists, err := client.UserSubscription.Query().Where(usersubscription.IDEQ(id)).Exist(ctx)
  454. if err != nil {
  455. return translatePersistenceError(err, service.ErrSubscriptionNotFound, nil)
  456. }
  457. if !exists {
  458. return service.ErrSubscriptionNotFound
  459. }
  460. return nil
  461. }
  462. // IncrementUsage 原子性地累加订阅用量。
  463. // 限额检查已在请求前由 BillingCacheService.CheckBillingEligibility 完成,
  464. // 此处仅负责记录实际消费,确保消费数据的完整性。
  465. func (r *userSubscriptionRepository) IncrementUsage(ctx context.Context, id int64, costUSD float64) error {
  466. const updateSQL = `
  467. UPDATE user_subscriptions us
  468. SET
  469. daily_usage_usd = us.daily_usage_usd + $1,
  470. weekly_usage_usd = us.weekly_usage_usd + $1,
  471. monthly_usage_usd = us.monthly_usage_usd + $1,
  472. updated_at = NOW()
  473. FROM groups g
  474. WHERE us.id = $2
  475. AND us.deleted_at IS NULL
  476. AND us.group_id = g.id
  477. AND g.deleted_at IS NULL
  478. `
  479. client := clientFromContext(ctx, r.client)
  480. result, err := client.ExecContext(ctx, updateSQL, costUSD, id)
  481. if err != nil {
  482. return err
  483. }
  484. affected, err := result.RowsAffected()
  485. if err != nil {
  486. return err
  487. }
  488. if affected > 0 {
  489. return nil
  490. }
  491. // affected == 0:订阅不存在或已删除
  492. return service.ErrSubscriptionNotFound
  493. }
  494. func (r *userSubscriptionRepository) BatchUpdateExpiredStatus(ctx context.Context) (int64, error) {
  495. client := clientFromContext(ctx, r.client)
  496. n, err := client.UserSubscription.Update().
  497. Where(
  498. usersubscription.StatusEQ(service.SubscriptionStatusActive),
  499. usersubscription.ExpiresAtLTE(time.Now()),
  500. ).
  501. SetStatus(service.SubscriptionStatusExpired).
  502. Save(ctx)
  503. return int64(n), err
  504. }
  505. // Extra repository helpers (currently used only by integration tests).
  506. func (r *userSubscriptionRepository) ListExpired(ctx context.Context) ([]service.UserSubscription, error) {
  507. client := clientFromContext(ctx, r.client)
  508. subs, err := client.UserSubscription.Query().
  509. Where(
  510. usersubscription.StatusEQ(service.SubscriptionStatusActive),
  511. usersubscription.ExpiresAtLTE(time.Now()),
  512. ).
  513. All(ctx)
  514. if err != nil {
  515. return nil, err
  516. }
  517. return userSubscriptionEntitiesToService(subs), nil
  518. }
  519. func (r *userSubscriptionRepository) CountByGroupID(ctx context.Context, groupID int64) (int64, error) {
  520. client := clientFromContext(ctx, r.client)
  521. count, err := client.UserSubscription.Query().Where(usersubscription.GroupIDEQ(groupID)).Count(ctx)
  522. return int64(count), err
  523. }
  524. func (r *userSubscriptionRepository) CountActiveByGroupID(ctx context.Context, groupID int64) (int64, error) {
  525. client := clientFromContext(ctx, r.client)
  526. count, err := client.UserSubscription.Query().
  527. Where(
  528. usersubscription.GroupIDEQ(groupID),
  529. usersubscription.StatusEQ(service.SubscriptionStatusActive),
  530. usersubscription.ExpiresAtGT(time.Now()),
  531. ).
  532. Count(ctx)
  533. return int64(count), err
  534. }
  535. func (r *userSubscriptionRepository) DeleteByGroupID(ctx context.Context, groupID int64) (int64, error) {
  536. client := clientFromContext(ctx, r.client)
  537. n, err := client.UserSubscription.Delete().Where(usersubscription.GroupIDEQ(groupID)).Exec(ctx)
  538. return int64(n), err
  539. }
  540. func (r *userSubscriptionRepository) attachUserSubscriptionRelations(ctx context.Context, subs []service.UserSubscription) error {
  541. if len(subs) == 0 {
  542. return nil
  543. }
  544. userIDs := make([]int64, 0, len(subs))
  545. groupIDs := make([]int64, 0, len(subs))
  546. assignedByIDs := make([]int64, 0, len(subs))
  547. for i := range subs {
  548. userIDs = append(userIDs, subs[i].UserID)
  549. groupIDs = append(groupIDs, subs[i].GroupID)
  550. if subs[i].AssignedBy != nil {
  551. assignedByIDs = append(assignedByIDs, *subs[i].AssignedBy)
  552. }
  553. }
  554. client := clientFromContext(ctx, r.client)
  555. users, err := client.User.Query().Where(user.IDIn(uniqueInt64s(userIDs)...)).All(ctx)
  556. if err != nil {
  557. return err
  558. }
  559. userByID := make(map[int64]*service.User, len(users))
  560. for _, u := range users {
  561. userByID[u.ID] = userEntityToService(u)
  562. }
  563. groups, err := client.Group.Query().Where(group.IDIn(uniqueInt64s(groupIDs)...)).All(ctx)
  564. if err != nil {
  565. return err
  566. }
  567. groupByID := make(map[int64]*service.Group, len(groups))
  568. for _, g := range groups {
  569. groupByID[g.ID] = groupEntityToService(g)
  570. }
  571. assignedByID := map[int64]*service.User{}
  572. if len(assignedByIDs) > 0 {
  573. assignedUsers, err := client.User.Query().Where(user.IDIn(uniqueInt64s(assignedByIDs)...)).All(ctx)
  574. if err != nil {
  575. return err
  576. }
  577. assignedByID = make(map[int64]*service.User, len(assignedUsers))
  578. for _, u := range assignedUsers {
  579. assignedByID[u.ID] = userEntityToService(u)
  580. }
  581. }
  582. for i := range subs {
  583. subs[i].User = userByID[subs[i].UserID]
  584. subs[i].Group = groupByID[subs[i].GroupID]
  585. if subs[i].AssignedBy != nil {
  586. subs[i].AssignedByUser = assignedByID[*subs[i].AssignedBy]
  587. }
  588. }
  589. return nil
  590. }
  591. func uniqueInt64s(values []int64) []int64 {
  592. seen := make(map[int64]struct{}, len(values))
  593. out := make([]int64, 0, len(values))
  594. for _, v := range values {
  595. if _, ok := seen[v]; ok {
  596. continue
  597. }
  598. seen[v] = struct{}{}
  599. out = append(out, v)
  600. }
  601. return out
  602. }
  603. func userSubscriptionEntityToService(m *dbent.UserSubscription) *service.UserSubscription {
  604. return userSubscriptionEntityToServiceWithStatusMapping(m, true)
  605. }
  606. func userSubscriptionEntityToServicePreserveStatus(m *dbent.UserSubscription) *service.UserSubscription {
  607. return userSubscriptionEntityToServiceWithStatusMapping(m, false)
  608. }
  609. func userSubscriptionEntityToServiceWithStatusMapping(m *dbent.UserSubscription, mapDeletedToRevoked bool) *service.UserSubscription {
  610. if m == nil {
  611. return nil
  612. }
  613. status := m.Status
  614. if mapDeletedToRevoked && m.DeletedAt != nil {
  615. status = service.SubscriptionStatusRevoked
  616. }
  617. out := &service.UserSubscription{
  618. ID: m.ID,
  619. UserID: m.UserID,
  620. GroupID: m.GroupID,
  621. StartsAt: m.StartsAt,
  622. ExpiresAt: m.ExpiresAt,
  623. Status: status,
  624. DailyWindowStart: m.DailyWindowStart,
  625. WeeklyWindowStart: m.WeeklyWindowStart,
  626. MonthlyWindowStart: m.MonthlyWindowStart,
  627. DailyUsageUSD: m.DailyUsageUsd,
  628. WeeklyUsageUSD: m.WeeklyUsageUsd,
  629. MonthlyUsageUSD: m.MonthlyUsageUsd,
  630. AssignedBy: m.AssignedBy,
  631. AssignedAt: m.AssignedAt,
  632. Notes: derefString(m.Notes),
  633. CreatedAt: m.CreatedAt,
  634. UpdatedAt: m.UpdatedAt,
  635. DeletedAt: m.DeletedAt,
  636. }
  637. if m.Edges.User != nil {
  638. out.User = userEntityToService(m.Edges.User)
  639. }
  640. if m.Edges.Group != nil {
  641. out.Group = groupEntityToService(m.Edges.Group)
  642. }
  643. if m.Edges.AssignedByUser != nil {
  644. out.AssignedByUser = userEntityToService(m.Edges.AssignedByUser)
  645. }
  646. return out
  647. }
  648. func userSubscriptionEntitiesToService(models []*dbent.UserSubscription) []service.UserSubscription {
  649. out := make([]service.UserSubscription, 0, len(models))
  650. for i := range models {
  651. if s := userSubscriptionEntityToService(models[i]); s != nil {
  652. out = append(out, *s)
  653. }
  654. }
  655. return out
  656. }
  657. func applyUserSubscriptionEntityToService(dst *service.UserSubscription, src *dbent.UserSubscription) {
  658. if dst == nil || src == nil {
  659. return
  660. }
  661. dst.ID = src.ID
  662. dst.CreatedAt = src.CreatedAt
  663. dst.UpdatedAt = src.UpdatedAt
  664. }