feat(ops): add tenant admin user management module
Adds /admin-users CRUD on the ops backend, covering listing, plan/role/KOL toggles, subscription expiry, status flips and password resets, with a configurable default plan code. The ops console exposes a new top-level "用户管理" view and reorganises the sidebar so 操作员管理 and 审计日志 move under a "系统设置" group; AccountsView is renamed accordingly. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,833 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
tenantrepo "github.com/geo-platform/tenant-api/internal/tenant/repository"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrAdminUserNotFound = errors.New("admin user not found")
|
||||
ErrDefaultPlanNotFound = errors.New("default plan not found")
|
||||
ErrActivePlanNotFound = errors.New("active plan subscription not found")
|
||||
ErrPrimaryWorkspaceEmpty = errors.New("primary workspace not created")
|
||||
)
|
||||
|
||||
type AdminUserRepository struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewAdminUserRepository(pool *pgxpool.Pool) *AdminUserRepository {
|
||||
return &AdminUserRepository{pool: pool}
|
||||
}
|
||||
|
||||
type AdminUserRecord struct {
|
||||
ID int64
|
||||
Email *string
|
||||
Phone string
|
||||
Name *string
|
||||
Status string
|
||||
TenantID *int64
|
||||
TenantName *string
|
||||
TenantStatus *string
|
||||
TenantRole *string
|
||||
PrimaryWorkspaceID *int64
|
||||
PlanCode *string
|
||||
PlanName *string
|
||||
SubscriptionStatus *string
|
||||
SubscriptionEndAt *time.Time
|
||||
KolProfileID *int64
|
||||
KolDisplayName *string
|
||||
KolStatus *string
|
||||
KolMarketEnabled *bool
|
||||
CreatedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
type PlanRecord struct {
|
||||
ID int64
|
||||
Code string
|
||||
Name string
|
||||
Status string
|
||||
DurationSecs int64
|
||||
ArticleLimit int
|
||||
AIPointsMonthly int
|
||||
BrandLimit int
|
||||
}
|
||||
|
||||
type AdminUserListFilter struct {
|
||||
Keyword string
|
||||
Status string
|
||||
Limit int
|
||||
Offset int
|
||||
}
|
||||
|
||||
type CreateAdminUserInput struct {
|
||||
Email *string
|
||||
Phone string
|
||||
PasswordHash string
|
||||
Name *string
|
||||
PlanCode string
|
||||
TenantRole string
|
||||
WorkspaceRole string
|
||||
}
|
||||
|
||||
type UpdateAdminUserProfileInput struct {
|
||||
Email *string
|
||||
Phone string
|
||||
Name *string
|
||||
}
|
||||
|
||||
type ResetPlanUsageResult struct {
|
||||
TenantID int64
|
||||
SubscriptionID int64
|
||||
ArticleReservations int64
|
||||
ArticleAmount int64
|
||||
AIPointReservations int64
|
||||
AIPointsAmount int64
|
||||
AIPointUsageLogs int64
|
||||
ArticleWindowStart time.Time
|
||||
ArticleWindowEnd time.Time
|
||||
AIPointsWindowStart time.Time
|
||||
AIPointsWindowEnd time.Time
|
||||
}
|
||||
|
||||
const adminUserColumns = `
|
||||
u.id,
|
||||
u.email,
|
||||
u.phone,
|
||||
u.name,
|
||||
u.status,
|
||||
tm.tenant_id,
|
||||
t.name AS tenant_name,
|
||||
t.status AS tenant_status,
|
||||
tm.tenant_role,
|
||||
wm.workspace_id AS primary_workspace_id,
|
||||
sub.plan_code,
|
||||
sub.plan_name,
|
||||
sub.subscription_status,
|
||||
sub.subscription_end_at,
|
||||
kp.id AS kol_profile_id,
|
||||
kp.display_name AS kol_display_name,
|
||||
kp.status AS kol_status,
|
||||
kp.market_enabled AS kol_market_enabled,
|
||||
u.created_at,
|
||||
u.updated_at`
|
||||
|
||||
const adminUserFromClause = `
|
||||
FROM users u
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT tm.tenant_id, tm.tenant_role
|
||||
FROM tenant_memberships tm
|
||||
WHERE tm.user_id = u.id
|
||||
AND tm.deleted_at IS NULL
|
||||
ORDER BY tm.created_at ASC, tm.id ASC
|
||||
LIMIT 1
|
||||
) tm ON TRUE
|
||||
LEFT JOIN tenants t
|
||||
ON t.id = tm.tenant_id
|
||||
AND t.deleted_at IS NULL
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT wm.workspace_id
|
||||
FROM workspace_memberships wm
|
||||
WHERE wm.user_id = u.id
|
||||
AND wm.tenant_id = tm.tenant_id
|
||||
AND wm.is_primary = TRUE
|
||||
ORDER BY wm.created_at ASC, wm.id ASC
|
||||
LIMIT 1
|
||||
) wm ON TRUE
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT
|
||||
p.plan_code,
|
||||
p.name AS plan_name,
|
||||
s.status AS subscription_status,
|
||||
s.end_at AS subscription_end_at
|
||||
FROM tenant_plan_subscriptions s
|
||||
JOIN plans p
|
||||
ON p.id = s.plan_id
|
||||
AND p.deleted_at IS NULL
|
||||
WHERE s.tenant_id = tm.tenant_id
|
||||
AND s.deleted_at IS NULL
|
||||
ORDER BY
|
||||
CASE
|
||||
WHEN s.status = 'active'
|
||||
AND p.status = 'active'
|
||||
AND s.start_at <= NOW()
|
||||
AND s.end_at > NOW()
|
||||
THEN 0
|
||||
ELSE 1
|
||||
END,
|
||||
s.start_at DESC,
|
||||
s.id DESC
|
||||
LIMIT 1
|
||||
) sub ON TRUE
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT kp.id, kp.display_name, kp.status, kp.market_enabled
|
||||
FROM kol_profiles kp
|
||||
WHERE kp.user_id = u.id
|
||||
AND kp.deleted_at IS NULL
|
||||
ORDER BY kp.created_at ASC, kp.id ASC
|
||||
LIMIT 1
|
||||
) kp ON TRUE`
|
||||
|
||||
func scanAdminUser(row pgx.Row) (*AdminUserRecord, error) {
|
||||
var u AdminUserRecord
|
||||
err := row.Scan(
|
||||
&u.ID,
|
||||
&u.Email,
|
||||
&u.Phone,
|
||||
&u.Name,
|
||||
&u.Status,
|
||||
&u.TenantID,
|
||||
&u.TenantName,
|
||||
&u.TenantStatus,
|
||||
&u.TenantRole,
|
||||
&u.PrimaryWorkspaceID,
|
||||
&u.PlanCode,
|
||||
&u.PlanName,
|
||||
&u.SubscriptionStatus,
|
||||
&u.SubscriptionEndAt,
|
||||
&u.KolProfileID,
|
||||
&u.KolDisplayName,
|
||||
&u.KolStatus,
|
||||
&u.KolMarketEnabled,
|
||||
&u.CreatedAt,
|
||||
&u.UpdatedAt,
|
||||
)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrAdminUserNotFound
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return &u, nil
|
||||
}
|
||||
|
||||
func (r *AdminUserRepository) GetByID(ctx context.Context, id int64) (*AdminUserRecord, error) {
|
||||
row := r.pool.QueryRow(ctx, `
|
||||
SELECT `+adminUserColumns+`
|
||||
`+adminUserFromClause+`
|
||||
WHERE u.id = $1
|
||||
AND u.deleted_at IS NULL`, id)
|
||||
return scanAdminUser(row)
|
||||
}
|
||||
|
||||
func (r *AdminUserRepository) List(ctx context.Context, f AdminUserListFilter) ([]AdminUserRecord, int64, error) {
|
||||
args := []any{}
|
||||
where := "u.deleted_at IS NULL"
|
||||
|
||||
if f.Keyword != "" {
|
||||
args = append(args, "%"+f.Keyword+"%")
|
||||
where += fmt.Sprintf(" AND (u.name ILIKE $%d OR u.email ILIKE $%d OR u.phone ILIKE $%d OR t.name ILIKE $%d)",
|
||||
len(args), len(args), len(args), len(args))
|
||||
}
|
||||
if f.Status != "" {
|
||||
args = append(args, f.Status)
|
||||
where += fmt.Sprintf(" AND u.status = $%d", len(args))
|
||||
}
|
||||
|
||||
var total int64
|
||||
if err := r.pool.QueryRow(ctx, `
|
||||
SELECT COUNT(*)
|
||||
`+adminUserFromClause+`
|
||||
WHERE `+where, args...).Scan(&total); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
limit := f.Limit
|
||||
if limit <= 0 || limit > 200 {
|
||||
limit = 20
|
||||
}
|
||||
offset := f.Offset
|
||||
if offset < 0 {
|
||||
offset = 0
|
||||
}
|
||||
args = append(args, limit, offset)
|
||||
limitArg := len(args) - 1
|
||||
offsetArg := len(args)
|
||||
|
||||
rows, err := r.pool.Query(ctx, fmt.Sprintf(`
|
||||
SELECT %s
|
||||
%s
|
||||
WHERE %s
|
||||
ORDER BY u.created_at DESC, u.id DESC
|
||||
LIMIT $%d OFFSET $%d`, adminUserColumns, adminUserFromClause, where, limitArg, offsetArg), args...)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := make([]AdminUserRecord, 0, limit)
|
||||
for rows.Next() {
|
||||
record, err := scanAdminUser(rows)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
out = append(out, *record)
|
||||
}
|
||||
return out, total, rows.Err()
|
||||
}
|
||||
|
||||
func (r *AdminUserRepository) ListActivePlans(ctx context.Context) ([]PlanRecord, error) {
|
||||
rows, err := r.pool.Query(ctx, `
|
||||
SELECT
|
||||
id,
|
||||
plan_code,
|
||||
name,
|
||||
status,
|
||||
COALESCE(NULLIF(quota_policy_json->>'subscription_duration_seconds', '')::BIGINT, 259200),
|
||||
COALESCE(NULLIF(quota_policy_json->>'article_generation', '')::INT, 0),
|
||||
COALESCE(NULLIF(quota_policy_json->>'ai_points_monthly', '')::INT, 0),
|
||||
COALESCE(NULLIF(quota_policy_json->>'brand_limit', '')::INT, 0)
|
||||
FROM plans
|
||||
WHERE status = 'active'
|
||||
AND deleted_at IS NULL
|
||||
ORDER BY id ASC`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := []PlanRecord{}
|
||||
for rows.Next() {
|
||||
var p PlanRecord
|
||||
if err := rows.Scan(
|
||||
&p.ID,
|
||||
&p.Code,
|
||||
&p.Name,
|
||||
&p.Status,
|
||||
&p.DurationSecs,
|
||||
&p.ArticleLimit,
|
||||
&p.AIPointsMonthly,
|
||||
&p.BrandLimit,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, p)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (r *AdminUserRepository) CreatePersonalTenantUser(ctx context.Context, in CreateAdminUserInput) (*AdminUserRecord, error) {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
_ = tx.Rollback(ctx)
|
||||
}()
|
||||
|
||||
var userID int64
|
||||
if err := tx.QueryRow(ctx, `
|
||||
INSERT INTO users (email, phone, password_hash, name, status)
|
||||
VALUES ($1, $2, $3, $4, 'active')
|
||||
RETURNING id`, in.Email, in.Phone, in.PasswordHash, in.Name).Scan(&userID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var tenantID int64
|
||||
tenantName := fmt.Sprintf("user-%d", userID)
|
||||
if err := tx.QueryRow(ctx, `
|
||||
INSERT INTO tenants (name, status)
|
||||
VALUES ($1, 'active')
|
||||
RETURNING id`, tenantName).Scan(&tenantID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO tenant_memberships (tenant_id, user_id, tenant_role)
|
||||
VALUES ($1, $2, $3)`, tenantID, userID, in.TenantRole); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var workspaceID int64
|
||||
if err := tx.QueryRow(ctx, `
|
||||
INSERT INTO workspaces (tenant_id, name, slug, is_default)
|
||||
VALUES ($1, 'Default', 'default', TRUE)
|
||||
RETURNING id`, tenantID).Scan(&workspaceID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if workspaceID == 0 {
|
||||
return nil, ErrPrimaryWorkspaceEmpty
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO workspace_memberships (workspace_id, user_id, tenant_id, role, is_primary)
|
||||
VALUES ($1, $2, $3, $4, TRUE)`, workspaceID, userID, tenantID, in.WorkspaceRole); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var (
|
||||
planID int64
|
||||
durationSeconds int64
|
||||
)
|
||||
if err := tx.QueryRow(ctx, `
|
||||
SELECT id,
|
||||
COALESCE(NULLIF(quota_policy_json->>'subscription_duration_seconds', '')::BIGINT, 259200)
|
||||
FROM plans
|
||||
WHERE plan_code = $1
|
||||
AND status = 'active'
|
||||
AND deleted_at IS NULL
|
||||
LIMIT 1`, in.PlanCode).Scan(&planID, &durationSeconds); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrDefaultPlanNotFound
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if durationSeconds <= 0 {
|
||||
durationSeconds = 259200
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO tenant_plan_subscriptions (tenant_id, plan_id, start_at, end_at, status)
|
||||
VALUES ($1, $2, NOW(), NOW() + ($3::BIGINT * INTERVAL '1 second'), 'active')`,
|
||||
tenantID, planID, durationSeconds); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return r.GetByID(ctx, userID)
|
||||
}
|
||||
|
||||
func (r *AdminUserRepository) ChangePlan(ctx context.Context, userID int64, planCode string) (*AdminUserRecord, error) {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
_ = tx.Rollback(ctx)
|
||||
}()
|
||||
|
||||
var tenantID int64
|
||||
if err := tx.QueryRow(ctx, `
|
||||
SELECT tm.tenant_id
|
||||
FROM tenant_memberships tm
|
||||
WHERE tm.user_id = $1
|
||||
AND tm.deleted_at IS NULL
|
||||
ORDER BY tm.created_at ASC, tm.id ASC
|
||||
LIMIT 1`, userID).Scan(&tenantID); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrAdminUserNotFound
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var (
|
||||
planID int64
|
||||
durationSeconds int64
|
||||
)
|
||||
if err := tx.QueryRow(ctx, `
|
||||
SELECT id,
|
||||
COALESCE(NULLIF(quota_policy_json->>'subscription_duration_seconds', '')::BIGINT, 259200)
|
||||
FROM plans
|
||||
WHERE plan_code = $1
|
||||
AND status = 'active'
|
||||
AND deleted_at IS NULL
|
||||
LIMIT 1`, planCode).Scan(&planID, &durationSeconds); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrDefaultPlanNotFound
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if durationSeconds <= 0 {
|
||||
durationSeconds = 259200
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE tenant_plan_subscriptions
|
||||
SET status = 'replaced',
|
||||
updated_at = NOW()
|
||||
WHERE tenant_id = $1
|
||||
AND status = 'active'
|
||||
AND deleted_at IS NULL`, tenantID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO tenant_plan_subscriptions (tenant_id, plan_id, start_at, end_at, status)
|
||||
VALUES ($1, $2, NOW(), NOW() + ($3::BIGINT * INTERVAL '1 second'), 'active')`,
|
||||
tenantID, planID, durationSeconds); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return r.GetByID(ctx, userID)
|
||||
}
|
||||
|
||||
func (r *AdminUserRepository) UpdateSubscriptionEndAt(ctx context.Context, userID int64, endAt time.Time) (*AdminUserRecord, error) {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
_ = tx.Rollback(ctx)
|
||||
}()
|
||||
|
||||
var tenantID int64
|
||||
if err := tx.QueryRow(ctx, `
|
||||
SELECT tm.tenant_id
|
||||
FROM tenant_memberships tm
|
||||
WHERE tm.user_id = $1
|
||||
AND tm.deleted_at IS NULL
|
||||
ORDER BY tm.created_at ASC, tm.id ASC
|
||||
LIMIT 1`, userID).Scan(&tenantID); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrAdminUserNotFound
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
tag, err := tx.Exec(ctx, `
|
||||
UPDATE tenant_plan_subscriptions
|
||||
SET end_at = $1,
|
||||
updated_at = NOW()
|
||||
WHERE tenant_id = $2
|
||||
AND status = 'active'
|
||||
AND deleted_at IS NULL`, endAt.UTC(), tenantID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return nil, ErrActivePlanNotFound
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return r.GetByID(ctx, userID)
|
||||
}
|
||||
|
||||
func (r *AdminUserRepository) ResetCurrentPlanUsage(ctx context.Context, userID int64) (*AdminUserRecord, ResetPlanUsageResult, error) {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, ResetPlanUsageResult{}, err
|
||||
}
|
||||
defer func() {
|
||||
_ = tx.Rollback(ctx)
|
||||
}()
|
||||
|
||||
var tenantID int64
|
||||
if err := tx.QueryRow(ctx, `
|
||||
SELECT tm.tenant_id
|
||||
FROM tenant_memberships tm
|
||||
WHERE tm.user_id = $1
|
||||
AND tm.deleted_at IS NULL
|
||||
ORDER BY tm.created_at ASC, tm.id ASC
|
||||
LIMIT 1`, userID).Scan(&tenantID); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ResetPlanUsageResult{}, ErrAdminUserNotFound
|
||||
}
|
||||
return nil, ResetPlanUsageResult{}, err
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(20260428, hashint8($1)::int)`, tenantID); err != nil {
|
||||
return nil, ResetPlanUsageResult{}, err
|
||||
}
|
||||
|
||||
now := time.Now().UTC()
|
||||
access, err := tenantrepo.NewTenantPlanRepository(tx).GetTenantPlanAccess(ctx, tenantID, now)
|
||||
if err != nil {
|
||||
return nil, ResetPlanUsageResult{}, err
|
||||
}
|
||||
if access == nil || access.SubscriptionStatus != "active" {
|
||||
return nil, ResetPlanUsageResult{}, ErrActivePlanNotFound
|
||||
}
|
||||
|
||||
referenceTime := now
|
||||
if !access.EndAt.IsZero() && !access.EndAt.After(now) && access.EndAt.After(access.StartAt) {
|
||||
referenceTime = access.EndAt.Add(-time.Nanosecond)
|
||||
}
|
||||
if referenceTime.Before(access.StartAt) {
|
||||
referenceTime = access.StartAt
|
||||
}
|
||||
|
||||
articleStart, articleEnd, _ := access.ArticleQuotaWindow(referenceTime)
|
||||
aiStart, aiEnd, _ := access.AIPointsWindow(referenceTime)
|
||||
|
||||
result := ResetPlanUsageResult{
|
||||
TenantID: tenantID,
|
||||
SubscriptionID: access.SubscriptionID,
|
||||
ArticleWindowStart: articleStart,
|
||||
ArticleWindowEnd: articleEnd,
|
||||
AIPointsWindowStart: aiStart,
|
||||
AIPointsWindowEnd: aiEnd,
|
||||
}
|
||||
|
||||
if articleEnd.After(articleStart) {
|
||||
count, amount, err := resetQuotaReservations(ctx, tx, tenantID, "article_generation", articleStart, articleEnd)
|
||||
if err != nil {
|
||||
return nil, ResetPlanUsageResult{}, err
|
||||
}
|
||||
result.ArticleReservations = count
|
||||
result.ArticleAmount = amount
|
||||
}
|
||||
|
||||
if aiEnd.After(aiStart) {
|
||||
count, amount, logCount, err := resetAIPointReservations(ctx, tx, tenantID, aiStart, aiEnd)
|
||||
if err != nil {
|
||||
return nil, ResetPlanUsageResult{}, err
|
||||
}
|
||||
result.AIPointReservations = count
|
||||
result.AIPointsAmount = amount
|
||||
result.AIPointUsageLogs = logCount
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, ResetPlanUsageResult{}, err
|
||||
}
|
||||
|
||||
user, err := r.GetByID(ctx, userID)
|
||||
if err != nil {
|
||||
return nil, ResetPlanUsageResult{}, err
|
||||
}
|
||||
return user, result, nil
|
||||
}
|
||||
|
||||
func resetQuotaReservations(ctx context.Context, tx pgx.Tx, tenantID int64, quotaType string, startAt, endAt time.Time) (int64, int64, error) {
|
||||
row := tx.QueryRow(ctx, `
|
||||
WITH reset AS (
|
||||
UPDATE quota_reservations
|
||||
SET status = 'refunded',
|
||||
refunded_amount = reserved_amount,
|
||||
updated_at = NOW()
|
||||
WHERE tenant_id = $1
|
||||
AND quota_type = $2
|
||||
AND status IN ('pending', 'confirmed')
|
||||
AND created_at >= $3
|
||||
AND created_at < $4
|
||||
RETURNING reserved_amount
|
||||
)
|
||||
SELECT COUNT(*)::BIGINT,
|
||||
COALESCE(SUM(reserved_amount), 0)::BIGINT
|
||||
FROM reset`, tenantID, quotaType, startAt.UTC(), endAt.UTC())
|
||||
|
||||
var count int64
|
||||
var amount int64
|
||||
if err := row.Scan(&count, &amount); err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
return count, amount, nil
|
||||
}
|
||||
|
||||
func resetAIPointReservations(ctx context.Context, tx pgx.Tx, tenantID int64, startAt, endAt time.Time) (int64, int64, int64, error) {
|
||||
row := tx.QueryRow(ctx, `
|
||||
WITH reset AS (
|
||||
UPDATE quota_reservations
|
||||
SET status = 'refunded',
|
||||
refunded_amount = reserved_amount,
|
||||
updated_at = NOW()
|
||||
WHERE tenant_id = $1
|
||||
AND quota_type = 'ai_points'
|
||||
AND status IN ('pending', 'confirmed')
|
||||
AND created_at >= $2
|
||||
AND created_at < $3
|
||||
RETURNING id, reserved_amount
|
||||
),
|
||||
logs AS (
|
||||
UPDATE ai_point_usage_logs l
|
||||
SET status = 'refunded',
|
||||
error_message = 'ops reset current plan usage',
|
||||
completed_at = COALESCE(l.completed_at, NOW()),
|
||||
updated_at = NOW()
|
||||
FROM reset r
|
||||
WHERE l.tenant_id = $1
|
||||
AND l.quota_reservation_id = r.id
|
||||
AND l.status IN ('pending', 'completed')
|
||||
RETURNING l.id
|
||||
)
|
||||
SELECT (SELECT COUNT(*) FROM reset)::BIGINT,
|
||||
COALESCE((SELECT SUM(reserved_amount) FROM reset), 0)::BIGINT,
|
||||
(SELECT COUNT(*) FROM logs)::BIGINT`, tenantID, startAt.UTC(), endAt.UTC())
|
||||
|
||||
var reservationCount int64
|
||||
var amount int64
|
||||
var logCount int64
|
||||
if err := row.Scan(&reservationCount, &amount, &logCount); err != nil {
|
||||
return 0, 0, 0, err
|
||||
}
|
||||
return reservationCount, amount, logCount, nil
|
||||
}
|
||||
|
||||
func (r *AdminUserRepository) ChangeRole(ctx context.Context, userID int64, tenantRole, workspaceRole string) (*AdminUserRecord, error) {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
_ = tx.Rollback(ctx)
|
||||
}()
|
||||
|
||||
var tenantID int64
|
||||
if err := tx.QueryRow(ctx, `
|
||||
SELECT tm.tenant_id
|
||||
FROM tenant_memberships tm
|
||||
WHERE tm.user_id = $1
|
||||
AND tm.deleted_at IS NULL
|
||||
ORDER BY tm.created_at ASC, tm.id ASC
|
||||
LIMIT 1`, userID).Scan(&tenantID); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrAdminUserNotFound
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
tag, err := tx.Exec(ctx, `
|
||||
UPDATE tenant_memberships
|
||||
SET tenant_role = $1,
|
||||
updated_at = NOW()
|
||||
WHERE tenant_id = $2
|
||||
AND user_id = $3
|
||||
AND deleted_at IS NULL`, tenantRole, tenantID, userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return nil, ErrAdminUserNotFound
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE workspace_memberships
|
||||
SET role = $1
|
||||
WHERE tenant_id = $2
|
||||
AND user_id = $3`, workspaceRole, tenantID, userID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return r.GetByID(ctx, userID)
|
||||
}
|
||||
|
||||
func (r *AdminUserRepository) SetKOL(ctx context.Context, userID int64, enabled bool, displayName string, marketEnabled bool) (*AdminUserRecord, error) {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
_ = tx.Rollback(ctx)
|
||||
}()
|
||||
|
||||
var (
|
||||
tenantID int64
|
||||
userName *string
|
||||
phone string
|
||||
)
|
||||
if err := tx.QueryRow(ctx, `
|
||||
SELECT tm.tenant_id, u.name, u.phone
|
||||
FROM users u
|
||||
JOIN tenant_memberships tm
|
||||
ON tm.user_id = u.id
|
||||
AND tm.deleted_at IS NULL
|
||||
WHERE u.id = $1
|
||||
AND u.deleted_at IS NULL
|
||||
ORDER BY tm.created_at ASC, tm.id ASC
|
||||
LIMIT 1`, userID).Scan(&tenantID, &userName, &phone); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrAdminUserNotFound
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if displayName == "" {
|
||||
if userName != nil && *userName != "" {
|
||||
displayName = *userName
|
||||
} else {
|
||||
displayName = phone
|
||||
}
|
||||
}
|
||||
|
||||
if enabled {
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO kol_profiles (tenant_id, user_id, display_name, market_enabled, status)
|
||||
VALUES ($1, $2, $3, $4, 'active')
|
||||
ON CONFLICT (user_id) WHERE deleted_at IS NULL
|
||||
DO UPDATE SET
|
||||
display_name = EXCLUDED.display_name,
|
||||
market_enabled = EXCLUDED.market_enabled,
|
||||
status = 'active',
|
||||
updated_at = NOW()`, tenantID, userID, displayName, marketEnabled); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
} else {
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE kol_profiles
|
||||
SET status = 'suspended',
|
||||
market_enabled = FALSE,
|
||||
updated_at = NOW()
|
||||
WHERE user_id = $1
|
||||
AND deleted_at IS NULL`, userID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return r.GetByID(ctx, userID)
|
||||
}
|
||||
|
||||
func (r *AdminUserRepository) UpdateProfile(ctx context.Context, id int64, in UpdateAdminUserProfileInput) error {
|
||||
cmd, err := r.pool.Exec(ctx, `
|
||||
UPDATE users
|
||||
SET email = $1,
|
||||
phone = $2,
|
||||
name = $3,
|
||||
updated_at = NOW()
|
||||
WHERE id = $4
|
||||
AND deleted_at IS NULL`, in.Email, in.Phone, in.Name, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if cmd.RowsAffected() == 0 {
|
||||
return ErrAdminUserNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *AdminUserRepository) UpdateStatus(ctx context.Context, id int64, status string) error {
|
||||
cmd, err := r.pool.Exec(ctx, `
|
||||
UPDATE users
|
||||
SET status = $1,
|
||||
updated_at = NOW()
|
||||
WHERE id = $2
|
||||
AND deleted_at IS NULL`, status, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if cmd.RowsAffected() == 0 {
|
||||
return ErrAdminUserNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *AdminUserRepository) UpdatePassword(ctx context.Context, id int64, passwordHash string) error {
|
||||
cmd, err := r.pool.Exec(ctx, `
|
||||
UPDATE users
|
||||
SET password_hash = $1,
|
||||
updated_at = NOW()
|
||||
WHERE id = $2
|
||||
AND deleted_at IS NULL`, passwordHash, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if cmd.RowsAffected() == 0 {
|
||||
return ErrAdminUserNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user