Files
digital-psychology/apps/api/internal/repository/admin_repo.go
T
jackyu66gitandCursor b5a05941d9 feat(ECR-013A): Admin RBAC 实现并 Closed
角色权限、RequirePermission、/me permissions 与 migration 000015;
Reviewer Approve → Closed。Next:ECR-013B Contract Definition。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-07 17:19:45 +08:00

535 lines
16 KiB
Go

package repository
import (
"context"
"encoding/json"
"errors"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
// AdminRepo persists ops-admin accounts, sessions, and audit logs.
type AdminRepo struct {
Pool *pgxpool.Pool
}
// AdminAccount is an internal operator account.
type AdminAccount struct {
ID uuid.UUID
Username string
PasswordHash string
Status string
}
// CountAccounts returns non-deleted admin count.
func (r *AdminRepo) CountAccounts(ctx context.Context) (int, error) {
var n int
err := r.Pool.QueryRow(ctx, `
SELECT count(*) FROM admin_accounts WHERE deleted_at IS NULL`).Scan(&n)
return n, err
}
// CreateAccount inserts an admin account with seeded super_admin role.
func (r *AdminRepo) CreateAccount(ctx context.Context, username, hash string) (uuid.UUID, error) {
var id uuid.UUID
err := r.Pool.QueryRow(ctx, `
INSERT INTO admin_accounts(username, password_hash, role_id)
VALUES (
$1, $2,
(SELECT id FROM admin_roles WHERE name = 'super_admin' LIMIT 1)
) RETURNING id`, username, hash).Scan(&id)
return id, err
}
// FindByUsername loads an active admin by username.
func (r *AdminRepo) FindByUsername(ctx context.Context, username string) (*AdminAccount, error) {
var a AdminAccount
err := r.Pool.QueryRow(ctx, `
SELECT id, username, password_hash, status
FROM admin_accounts
WHERE username=$1 AND deleted_at IS NULL`, username,
).Scan(&a.ID, &a.Username, &a.PasswordHash, &a.Status)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, err
}
return &a, nil
}
// FindAccountByID loads admin by id.
func (r *AdminRepo) FindAccountByID(ctx context.Context, id uuid.UUID) (*AdminAccount, error) {
var a AdminAccount
err := r.Pool.QueryRow(ctx, `
SELECT id, username, password_hash, status
FROM admin_accounts
WHERE id=$1 AND deleted_at IS NULL`, id,
).Scan(&a.ID, &a.Username, &a.PasswordHash, &a.Status)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, err
}
return &a, nil
}
// CreateSession stores an opaque admin session token.
func (r *AdminRepo) CreateSession(ctx context.Context, adminID uuid.UUID, token string, expires time.Time) error {
_, err := r.Pool.Exec(ctx, `
INSERT INTO admin_sessions(admin_id, token, expires_at)
VALUES ($1,$2,$3)`, adminID, token, expires)
return err
}
// ResolveSession returns admin_id for a valid token.
func (r *AdminRepo) ResolveSession(ctx context.Context, token string) (uuid.UUID, error) {
var id uuid.UUID
err := r.Pool.QueryRow(ctx, `
SELECT s.admin_id FROM admin_sessions s
JOIN admin_accounts a ON a.id=s.admin_id AND a.deleted_at IS NULL AND a.status='active'
WHERE s.token=$1 AND s.expires_at > now()`, token).Scan(&id)
if errors.Is(err, pgx.ErrNoRows) {
return uuid.Nil, errors.New("invalid session")
}
return id, err
}
// DeleteSession removes a session by token.
func (r *AdminRepo) DeleteSession(ctx context.Context, token string) error {
_, err := r.Pool.Exec(ctx, `DELETE FROM admin_sessions WHERE token=$1`, token)
return err
}
// InsertAudit appends an immutable audit row.
func (r *AdminRepo) InsertAudit(ctx context.Context, adminID uuid.UUID, action, targetType, targetID string, meta json.RawMessage) error {
if meta == nil {
meta = json.RawMessage(`{}`)
}
_, err := r.Pool.Exec(ctx, `
INSERT INTO admin_audit_logs(admin_id, action, target_type, target_id, meta)
VALUES ($1,$2,$3,$4,$5)`, adminID, action, targetType, targetID, meta)
return err
}
// UserListItem is a compact user row for admin tables.
type UserListItem struct {
ID uuid.UUID `json:"id"`
Status string `json:"status"`
Phone *string `json:"phone,omitempty"`
Nickname *string `json:"nickname,omitempty"`
AskPaidQuotaLeft int `json:"ask_paid_quota_left"`
ProfileCount int `json:"profile_count"`
MembershipActive bool `json:"membership_active"`
MembershipPlan *string `json:"membership_plan,omitempty"`
CreatedAt time.Time `json:"created_at"`
}
// ListUsers returns users newest first; q matches id / phone / nickname.
func (r *AdminRepo) ListUsers(ctx context.Context, q string, limit, offset int) ([]UserListItem, error) {
if limit <= 0 || limit > 100 {
limit = 20
}
if offset < 0 {
offset = 0
}
rows, err := r.Pool.Query(ctx, `
SELECT u.id, u.status, u.phone, u.nickname, u.ask_paid_quota_left, u.created_at,
(SELECT count(*) FROM profiles p WHERE p.user_id=u.id AND p.deleted_at IS NULL) AS profile_count,
EXISTS(
SELECT 1 FROM memberships m
WHERE m.user_id=u.id AND m.deleted_at IS NULL AND m.status='active' AND m.expires_at > now()
) AS membership_active,
(
SELECT m.plan FROM memberships m
WHERE m.user_id=u.id AND m.deleted_at IS NULL
LIMIT 1
) AS membership_plan
FROM users u
WHERE u.deleted_at IS NULL
AND (
$1 = ''
OR u.id::text = $1
OR COALESCE(u.phone,'') ILIKE '%' || $1 || '%'
OR COALESCE(u.nickname,'') ILIKE '%' || $1 || '%'
)
ORDER BY u.created_at DESC
LIMIT $2 OFFSET $3`, q, limit, offset)
if err != nil {
return nil, err
}
defer rows.Close()
var out []UserListItem
for rows.Next() {
var u UserListItem
if err := rows.Scan(
&u.ID, &u.Status, &u.Phone, &u.Nickname, &u.AskPaidQuotaLeft, &u.CreatedAt,
&u.ProfileCount, &u.MembershipActive, &u.MembershipPlan,
); err != nil {
return nil, err
}
out = append(out, u)
}
return out, rows.Err()
}
// DashboardStats is ops overview counters.
type DashboardStats struct {
UsersTotal int `json:"users_total"`
MembershipActive int `json:"membership_active"`
OrdersToday int `json:"orders_today"`
PaidCentsToday int `json:"paid_cents_today"`
AskRepliesToday int `json:"ask_replies_today"`
ProfilesTotal int `json:"profiles_total"`
ReportsTotal int `json:"reports_total"`
Series []DashboardDay `json:"series"`
ReportsByType []ReportTypeCnt `json:"reports_by_type"`
}
// DashboardDay is one day of trend metrics.
type DashboardDay struct {
Day string `json:"day"`
NewUsers int `json:"new_users"`
Orders int `json:"orders"`
PaidCents int `json:"paid_cents"`
AskReplies int `json:"ask_replies"`
}
// ReportTypeCnt counts reports by type.
type ReportTypeCnt struct {
Type string `json:"type"`
Count int `json:"count"`
}
// GetDashboardStats aggregates key ops metrics.
func (r *AdminRepo) GetDashboardStats(ctx context.Context) (*DashboardStats, error) {
s := &DashboardStats{}
if err := r.Pool.QueryRow(ctx, `
SELECT count(*) FROM users WHERE deleted_at IS NULL`).Scan(&s.UsersTotal); err != nil {
return nil, err
}
if err := r.Pool.QueryRow(ctx, `
SELECT count(*) FROM memberships
WHERE deleted_at IS NULL AND status='active' AND expires_at > now()`).Scan(&s.MembershipActive); err != nil {
return nil, err
}
if err := r.Pool.QueryRow(ctx, `
SELECT count(*) FROM orders
WHERE deleted_at IS NULL AND created_at >= date_trunc('day', now())`).Scan(&s.OrdersToday); err != nil {
return nil, err
}
if err := r.Pool.QueryRow(ctx, `
SELECT coalesce(sum(amount_cents),0) FROM orders
WHERE deleted_at IS NULL AND status='paid' AND created_at >= date_trunc('day', now())`).Scan(&s.PaidCentsToday); err != nil {
return nil, err
}
if err := r.Pool.QueryRow(ctx, `
SELECT count(*) FROM ask_messages m
JOIN ask_threads t ON t.id=m.thread_id AND t.deleted_at IS NULL
WHERE m.deleted_at IS NULL AND m.role='assistant'
AND m.created_at >= date_trunc('day', now())`).Scan(&s.AskRepliesToday); err != nil {
return nil, err
}
if err := r.Pool.QueryRow(ctx, `
SELECT count(*) FROM profiles WHERE deleted_at IS NULL`).Scan(&s.ProfilesTotal); err != nil {
return nil, err
}
if err := r.Pool.QueryRow(ctx, `
SELECT count(*) FROM growth_reports WHERE deleted_at IS NULL`).Scan(&s.ReportsTotal); err != nil {
return nil, err
}
rows, err := r.Pool.Query(ctx, `
WITH days AS (
SELECT generate_series(
date_trunc('day', now()) - interval '6 day',
date_trunc('day', now()),
interval '1 day'
)::date AS d
)
SELECT to_char(d.d, 'YYYY-MM-DD') AS day,
(SELECT count(*) FROM users u
WHERE u.deleted_at IS NULL AND u.created_at::date = d.d) AS new_users,
(SELECT count(*) FROM orders o
WHERE o.deleted_at IS NULL AND o.created_at::date = d.d) AS orders,
(SELECT coalesce(sum(o.amount_cents),0) FROM orders o
WHERE o.deleted_at IS NULL AND o.status='paid' AND o.created_at::date = d.d) AS paid_cents,
(SELECT count(*) FROM ask_messages m
JOIN ask_threads t ON t.id=m.thread_id AND t.deleted_at IS NULL
WHERE m.deleted_at IS NULL AND m.role='assistant' AND m.created_at::date = d.d) AS ask_replies
FROM days d
ORDER BY d.d ASC`)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var p DashboardDay
if err := rows.Scan(&p.Day, &p.NewUsers, &p.Orders, &p.PaidCents, &p.AskReplies); err != nil {
return nil, err
}
s.Series = append(s.Series, p)
}
if err := rows.Err(); err != nil {
return nil, err
}
trows, err := r.Pool.Query(ctx, `
SELECT type, count(*) FROM growth_reports
WHERE deleted_at IS NULL
GROUP BY type
ORDER BY count(*) DESC`)
if err != nil {
return nil, err
}
defer trows.Close()
for trows.Next() {
var c ReportTypeCnt
if err := trows.Scan(&c.Type, &c.Count); err != nil {
return nil, err
}
s.ReportsByType = append(s.ReportsByType, c)
}
return s, trows.Err()
}
// UserExists reports whether user id is present.
func (r *AdminRepo) UserExists(ctx context.Context, id uuid.UUID) (bool, error) {
var n int
err := r.Pool.QueryRow(ctx, `
SELECT 1 FROM users WHERE id=$1 AND deleted_at IS NULL`, id).Scan(&n)
if errors.Is(err, pgx.ErrNoRows) {
return false, nil
}
return err == nil, err
}
// ProfileBrief for admin user detail.
type ProfileBrief struct {
ID uuid.UUID `json:"id"`
Relation string `json:"relation"`
DisplayName string `json:"display_name"`
BirthDate string `json:"birth_date,omitempty"`
}
// ListProfilesForUser returns profile briefs.
func (r *AdminRepo) ListProfilesForUser(ctx context.Context, userID uuid.UUID) ([]ProfileBrief, error) {
rows, err := r.Pool.Query(ctx, `
SELECT id, relation, display_name, to_char(birth_date, 'YYYY-MM-DD')
FROM profiles
WHERE user_id=$1 AND deleted_at IS NULL
ORDER BY created_at ASC`, userID)
if err != nil {
return nil, err
}
defer rows.Close()
var out []ProfileBrief
for rows.Next() {
var p ProfileBrief
if err := rows.Scan(&p.ID, &p.Relation, &p.DisplayName, &p.BirthDate); err != nil {
return nil, err
}
out = append(out, p)
}
return out, rows.Err()
}
// ReportBrief for admin user detail.
type ReportBrief struct {
ID uuid.UUID `json:"id"`
Type string `json:"type"`
CreatedAt time.Time `json:"created_at"`
}
// ListReportsForUser returns recent growth reports.
func (r *AdminRepo) ListReportsForUser(ctx context.Context, userID uuid.UUID, limit int) ([]ReportBrief, error) {
if limit <= 0 || limit > 50 {
limit = 20
}
rows, err := r.Pool.Query(ctx, `
SELECT id, type, created_at FROM growth_reports
WHERE user_id=$1 AND deleted_at IS NULL
ORDER BY created_at DESC
LIMIT $2`, userID, limit)
if err != nil {
return nil, err
}
defer rows.Close()
var out []ReportBrief
for rows.Next() {
var rep ReportBrief
if err := rows.Scan(&rep.ID, &rep.Type, &rep.CreatedAt); err != nil {
return nil, err
}
out = append(out, rep)
}
return out, rows.Err()
}
// GetUserAccount loads phone/nickname/paid ask quota for one user.
func (r *AdminRepo) GetUserAccount(ctx context.Context, userID uuid.UUID) (phone, nickname *string, paidLeft int, status string, createdAt time.Time, err error) {
err = r.Pool.QueryRow(ctx, `
SELECT phone, nickname, ask_paid_quota_left, status, created_at
FROM users WHERE id=$1 AND deleted_at IS NULL`, userID,
).Scan(&phone, &nickname, &paidLeft, &status, &createdAt)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil, 0, "", time.Time{}, err
}
return
}
// GrantAskQuotaWithAudit adds paid ask quota and writes audit.
func (r *AdminRepo) GrantAskQuotaWithAudit(ctx context.Context, adminID, userID uuid.UUID, delta int, meta json.RawMessage) (int, error) {
if delta <= 0 {
return 0, errors.New("delta must be positive")
}
tx, err := r.Pool.Begin(ctx)
if err != nil {
return 0, err
}
defer tx.Rollback(ctx)
var left int
err = tx.QueryRow(ctx, `
UPDATE users SET ask_paid_quota_left = ask_paid_quota_left + $2, updated_at=now()
WHERE id=$1 AND deleted_at IS NULL
RETURNING ask_paid_quota_left`, userID, delta,
).Scan(&left)
if err != nil {
return 0, err
}
if meta == nil {
meta = json.RawMessage(`{}`)
}
if _, err := tx.Exec(ctx, `
INSERT INTO admin_audit_logs(admin_id, action, target_type, target_id, meta)
VALUES ($1,'ask_quota.grant','user',$2,$3)`, adminID, userID.String(), meta); err != nil {
return 0, err
}
if err := tx.Commit(ctx); err != nil {
return 0, err
}
return left, nil
}
// OrderListItem for admin order tables.
type OrderListItem struct {
ID uuid.UUID `json:"id"`
UserID uuid.UUID `json:"user_id"`
Kind string `json:"kind"`
Plan *string `json:"plan,omitempty"`
AmountCents int `json:"amount_cents"`
Status string `json:"status"`
CreatedAt time.Time `json:"created_at"`
}
// ListOrders lists orders; optional user filter.
func (r *AdminRepo) ListOrders(ctx context.Context, userID *uuid.UUID, limit, offset int) ([]OrderListItem, error) {
if limit <= 0 || limit > 100 {
limit = 20
}
if offset < 0 {
offset = 0
}
rows, err := r.Pool.Query(ctx, `
SELECT id, user_id, kind, plan, amount_cents, status, created_at
FROM orders
WHERE deleted_at IS NULL
AND ($1::uuid IS NULL OR user_id = $1)
ORDER BY created_at DESC
LIMIT $2 OFFSET $3`, userID, limit, offset)
if err != nil {
return nil, err
}
defer rows.Close()
var out []OrderListItem
for rows.Next() {
var o OrderListItem
if err := rows.Scan(&o.ID, &o.UserID, &o.Kind, &o.Plan, &o.AmountCents, &o.Status, &o.CreatedAt); err != nil {
return nil, err
}
out = append(out, o)
}
return out, rows.Err()
}
// GrantMembershipWithAudit upserts membership and appends audit in one transaction.
func (r *AdminRepo) GrantMembershipWithAudit(
ctx context.Context,
adminID, userID uuid.UUID,
plan string,
days int,
meta json.RawMessage,
) error {
tx, err := r.Pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
if _, err := tx.Exec(ctx, `
INSERT INTO memberships(user_id, plan, status, expires_at, ask_quota_left)
VALUES ($1,$2,'active', now() + ($3 * interval '1 day'), 100)
ON CONFLICT (user_id) DO UPDATE SET
plan=EXCLUDED.plan, status='active',
expires_at=(CASE
WHEN memberships.expires_at IS NOT NULL AND memberships.expires_at > now()
THEN memberships.expires_at ELSE now()
END) + ($3 * interval '1 day'),
ask_quota_left=100, updated_at=now()`,
userID, plan, days); err != nil {
return err
}
if meta == nil {
meta = json.RawMessage(`{}`)
}
if _, err := tx.Exec(ctx, `
INSERT INTO admin_audit_logs(admin_id, action, target_type, target_id, meta)
VALUES ($1,'membership.grant','user',$2,$3)`, adminID, userID.String(), meta); err != nil {
return err
}
return tx.Commit(ctx)
}
// AuditListItem for admin audit table.
type AuditListItem struct {
ID uuid.UUID `json:"id"`
AdminID uuid.UUID `json:"admin_id"`
Action string `json:"action"`
TargetType string `json:"target_type"`
TargetID string `json:"target_id"`
Meta json.RawMessage `json:"meta"`
CreatedAt time.Time `json:"created_at"`
}
// ListAuditLogs returns newest audit rows.
func (r *AdminRepo) ListAuditLogs(ctx context.Context, limit, offset int) ([]AuditListItem, error) {
if limit <= 0 || limit > 100 {
limit = 20
}
if offset < 0 {
offset = 0
}
rows, err := r.Pool.Query(ctx, `
SELECT id, admin_id, action, target_type, target_id, meta, created_at
FROM admin_audit_logs
ORDER BY created_at DESC
LIMIT $1 OFFSET $2`, limit, offset)
if err != nil {
return nil, err
}
defer rows.Close()
var out []AuditListItem
for rows.Next() {
var a AuditListItem
if err := rows.Scan(&a.ID, &a.AdminID, &a.Action, &a.TargetType, &a.TargetID, &a.Meta, &a.CreatedAt); err != nil {
return nil, err
}
out = append(out, a)
}
return out, rows.Err()
}