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 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, err := resetAIPointReservations(ctx, tx, tenantID, aiStart, aiEnd) if err != nil { return nil, ResetPlanUsageResult{}, err } result.AIPointReservations = count result.AIPointsAmount = amount } 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 = 'reset', 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, error) { row := tx.QueryRow(ctx, ` WITH reset AS ( UPDATE quota_reservations SET status = 'reset', 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 reserved_amount ) SELECT (SELECT COUNT(*) FROM reset)::BIGINT, COALESCE((SELECT SUM(reserved_amount) FROM reset), 0)::BIGINT`, tenantID, startAt.UTC(), endAt.UTC()) var reservationCount int64 var amount int64 if err := row.Scan(&reservationCount, &amount); err != nil { return 0, 0, err } return reservationCount, amount, 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 }