feat(api): 开发auth、login等接口，规范格式化

2026-09-03 13:11:06 +08:00
parent 195b6d25e9
commit 037928c778
82 changed files with 9199 additions and 5236 deletions
@@ -0,0 +1,340 @@
+import { Inject, Injectable, UnauthorizedException } from '@nestjs/common';
+import type {
+    AuthSessionResponse,
+    AuthUser,
+    LoginRequest,
+    RegisterRequest,
+    Workspace,
+} from '@atlas/contracts';
+import type { Database, DatabaseTransaction } from '@atlas/database';
+import { ApiError } from '../common/api-error.js';
+import { APP_CONFIG, type AppConfig } from '../config/app-config.module.js';
+import { DATABASE } from '../database/database.module.js';
+import { AuthRateLimiter } from '../security/auth-rate-limiter.service.js';
+import {
+    createOpaqueToken,
+    hashPassword,
+    signAccessToken,
+    tokenHash,
+    verifyAccessToken,
+    verifyPassword,
+} from './auth.crypto.js';
+
+type UserRow = {
+    id: string;
+    email: string;
+    display_name: string | null;
+    email_verified_at: Date | null;
+    created_at: Date;
+    status: 'active' | 'disabled' | 'pending_review';
+    password_hash: string;
+};
+type WorkspaceRow = { id: string; name: string; role: Workspace['role'] };
+
+@Injectable()
+export class AuthService {
+    private readonly accessTtlSeconds = 900;
+    private readonly refreshTtlDays = 30;
+
+    public constructor(
+        @Inject(DATABASE) private readonly database: Database,
+        @Inject(APP_CONFIG) private readonly config: AppConfig,
+        private readonly rateLimiter: AuthRateLimiter,
+    ) {}
+
+    public async register(input: RegisterRequest, metadata: RequestMetadata) {
+        if (input.password !== input.password_confirmation)
+            throw validationError(
+                'password_confirmation',
+                'PASSWORD_CONFIRMATION_MISMATCH',
+                '两次输入的密码不一致',
+            );
+        if (input.password.length < this.config.AUTH_PASSWORD_MIN_LENGTH)
+            throw validationError(
+                'password',
+                'PASSWORD_TOO_SHORT',
+                `密码至少需要 ${this.config.AUTH_PASSWORD_MIN_LENGTH} 个字符`,
+            );
+        const email = input.email.trim().toLowerCase();
+        // Apply the shared limiter before spending CPU on password hashing or querying account existence.
+        await this.rateLimiter.consume('register', { ip: metadata.ip, email });
+        const passwordHash = await hashPassword(input.password);
+        try {
+            return await this.database.transaction(async (transaction) => {
+                // A user must never exist without its initial tenant and owner membership.
+                const user = await insertUser(
+                    transaction,
+                    email,
+                    passwordHash,
+                    input.display_name?.trim() || null,
+                );
+                const workspace = await createPersonalWorkspace(
+                    transaction,
+                    user.id,
+                    user.display_name ?? email,
+                );
+                return this.createSessionResponse(
+                    transaction,
+                    user,
+                    workspace,
+                    metadata,
+                );
+            });
+        } catch (error: unknown) {
+            if (isUniqueViolation(error))
+                throw new ApiError(
+                    409,
+                    'EMAIL_ALREADY_REGISTERED',
+                    '该邮箱已注册',
+                    [
+                        {
+                            field: 'email',
+                            code: 'EMAIL_ALREADY_REGISTERED',
+                            message: '该邮箱已注册',
+                        },
+                    ],
+                );
+            throw error;
+        }
+    }
+
+    public async login(input: LoginRequest, metadata: RequestMetadata) {
+        const email = input.email.trim().toLowerCase();
+        // Use the same normalized value for the limiter and citext account lookup.
+        await this.rateLimiter.consume('login', { ip: metadata.ip, email });
+        const result = await this.database.query<UserRow>(
+            'SELECT id, email, display_name, email_verified_at, created_at, status, password_hash FROM atlas.users WHERE email = $1',
+            [email],
+        );
+        const user = result.rows[0];
+        if (
+            !user ||
+            !(await verifyPassword(input.password, user.password_hash))
+        )
+            throw new ApiError(401, 'INVALID_CREDENTIALS', '邮箱或密码错误');
+        if (user.status === 'disabled')
+            throw new ApiError(403, 'ACCOUNT_DISABLED', '账户已停用');
+        const workspace = await this.findWorkspace(user.id);
+        return this.database.transaction(async (transaction) => {
+            await transaction.query(
+                'UPDATE atlas.users SET last_login_at = now() WHERE id = $1',
+                [user.id],
+            );
+            return this.createSessionResponse(
+                transaction,
+                user,
+                workspace,
+                metadata,
+            );
+        });
+    }
+
+    public async refresh(
+        refreshToken: string | undefined,
+        metadata: RequestMetadata,
+    ) {
+        // Refresh has no trusted email until its opaque token is validated, so it is IP-scoped only.
+        await this.rateLimiter.consume('refresh', { ip: metadata.ip });
+        if (!refreshToken)
+            throw new ApiError(401, 'SESSION_EXPIRED', '登录会话已过期');
+        return this.database.transaction(async (transaction) => {
+            const sessions = await transaction.query<
+                UserRow & { session_id: string }
+            >(
+                `SELECT u.id, u.email, u.display_name, u.email_verified_at, u.created_at, u.status, u.password_hash, s.id AS session_id
+        FROM atlas.user_sessions s JOIN atlas.users u ON u.id = s.user_id
+        WHERE s.refresh_token_hash = $1 AND s.revoked_at IS NULL AND s.expires_at > now() FOR UPDATE`,
+                [tokenHash(refreshToken)],
+            );
+            const user = sessions.rows[0];
+            if (!user || user.status !== 'active')
+                throw new ApiError(401, 'SESSION_EXPIRED', '登录会话已过期');
+            // Row locking makes a refresh token single-use even when two browser requests arrive together.
+            await transaction.query(
+                'UPDATE atlas.user_sessions SET revoked_at = now() WHERE id = $1',
+                [user.session_id],
+            );
+            const workspace = await findWorkspaceInTransaction(
+                transaction,
+                user.id,
+            );
+            return this.createSessionResponse(
+                transaction,
+                user,
+                workspace,
+                metadata,
+            );
+        });
+    }
+
+    public async logout(sessionId: string) {
+        await this.database.query(
+            'UPDATE atlas.user_sessions SET revoked_at = now() WHERE id = $1 AND revoked_at IS NULL',
+            [sessionId],
+        );
+    }
+
+    public async authenticate(accessToken: string | undefined) {
+        const token = accessToken?.replace(/^Bearer\s+/i, '');
+        const claims = token
+            ? verifyAccessToken(token, this.config.AUTH_JWT_SECRET)
+            : null;
+        if (!claims) throw new UnauthorizedException();
+        const active = await this.database.query(
+            'SELECT 1 FROM atlas.user_sessions WHERE id = $1 AND user_id = $2 AND revoked_at IS NULL AND expires_at > now()',
+            [claims.sid, claims.sub],
+        );
+        if (!active.rowCount) throw new UnauthorizedException();
+        return { userId: claims.sub, sessionId: claims.sid };
+    }
+
+    public async me(userId: string) {
+        const workspace = await this.findWorkspace(userId);
+        const users = await this.database.query<UserRow>(
+            'SELECT id, email, display_name, email_verified_at, created_at, status, password_hash FROM atlas.users WHERE id = $1',
+            [userId],
+        );
+        const user = users.rows[0];
+        if (!user) throw new UnauthorizedException();
+        const platforms = await this.database.workspaceTransaction(
+            workspace.id,
+            async (transaction) => {
+                const rows = await transaction.query<{
+                    id: string;
+                    code: string;
+                    name: string;
+                }>(
+                    "SELECT id, code, name FROM atlas.platforms WHERE status = 'active' ORDER BY CASE WHEN code = 'OZON' THEN 0 ELSE 1 END, name",
+                );
+                return rows.rows;
+            },
+        );
+        return {
+            user: publicUser(user),
+            workspace,
+            platforms,
+            suggested_context: {
+                platform_id: platforms[0]?.id ?? null,
+                default_store: null,
+            },
+        };
+    }
+
+    private async findWorkspace(userId: string) {
+        const result = await this.database.query<WorkspaceRow>(
+            `SELECT w.id, w.name, m.role FROM atlas.workspace_memberships m JOIN atlas.workspaces w ON w.id = m.workspace_id WHERE m.user_id = $1 AND w.status = 'active' ORDER BY m.joined_at LIMIT 1`,
+            [userId],
+        );
+        return requireWorkspace(result.rows[0]);
+    }
+    private async createSessionResponse(
+        transaction: DatabaseTransaction,
+        user: UserRow,
+        workspace: WorkspaceRow,
+        metadata: RequestMetadata,
+    ) {
+        const refreshToken = createOpaqueToken();
+        const session = await transaction.query<{ id: string }>(
+            `INSERT INTO atlas.user_sessions (user_id, refresh_token_hash, expires_at, ip_hash, user_agent)
+      VALUES ($1, $2, now() + ($3 || ' days')::interval, $4, $5) RETURNING id`,
+            [
+                user.id,
+                tokenHash(refreshToken),
+                this.refreshTtlDays,
+                metadata.ipHash,
+                metadata.userAgent,
+            ],
+        );
+        const now = Math.floor(Date.now() / 1000);
+        const accessToken = signAccessToken(
+            {
+                sub: user.id,
+                sid: session.rows[0].id,
+                iat: now,
+                exp: now + this.accessTtlSeconds,
+            },
+            this.config.AUTH_JWT_SECRET,
+        );
+        const payload: AuthSessionResponse = {
+            access_token: accessToken,
+            access_token_expires_in: this.accessTtlSeconds,
+            user: publicUser(user),
+            workspace,
+        };
+        return { payload, refreshToken };
+    }
+}
+
+export type RequestMetadata = {
+    ip: string | null;
+    ipHash: string | null;
+    userAgent: string | null;
+};
+function validationError(field: string, code: string, message: string) {
+    return new ApiError(422, 'VALIDATION_FAILED', '请求参数不符合要求', [
+        { field, code, message },
+    ]);
+}
+function isUniqueViolation(error: unknown): error is { code: string } {
+    return (
+        typeof error === 'object' &&
+        error !== null &&
+        'code' in error &&
+        (error as { code: string }).code === '23505'
+    );
+}
+function publicUser(user: UserRow): AuthUser {
+    return {
+        id: user.id,
+        email: user.email,
+        display_name: user.display_name,
+        email_verified_at: user.email_verified_at?.toISOString() ?? null,
+        created_at: user.created_at.toISOString(),
+    };
+}
+async function insertUser(
+    transaction: DatabaseTransaction,
+    email: string,
+    passwordHash: string,
+    displayName: string | null,
+) {
+    const result = await transaction.query<UserRow>(
+        'INSERT INTO atlas.users (email, password_hash, display_name) VALUES ($1, $2, $3) RETURNING id, email, display_name, email_verified_at, created_at, status, password_hash',
+        [email, passwordHash, displayName],
+    );
+    return result.rows[0];
+}
+async function createPersonalWorkspace(
+    transaction: DatabaseTransaction,
+    userId: string,
+    label: string,
+) {
+    const workspace = await transaction.query<{ id: string; name: string }>(
+        'INSERT INTO atlas.workspaces (name, slug) VALUES ($1, $2) RETURNING id, name',
+        [`${label} 的工作空间`, `personal-${userId}`],
+    );
+    await transaction.query(
+        "INSERT INTO atlas.workspace_memberships (workspace_id, user_id, role) VALUES ($1, $2, 'owner')",
+        [workspace.rows[0].id, userId],
+    );
+    return { ...workspace.rows[0], role: 'owner' as const };
+}
+async function findWorkspaceInTransaction(
+    transaction: DatabaseTransaction,
+    userId: string,
+): Promise<WorkspaceRow> {
+    const result = await transaction.query<WorkspaceRow>(
+        `SELECT w.id, w.name, m.role FROM atlas.workspace_memberships m JOIN atlas.workspaces w ON w.id = m.workspace_id WHERE m.user_id = $1 AND w.status = 'active' ORDER BY m.joined_at LIMIT 1`,
+        [userId],
+    );
+    return requireWorkspace(result.rows[0]);
+}
+function requireWorkspace(workspace: WorkspaceRow | undefined): WorkspaceRow {
+    if (!workspace)
+        throw new ApiError(
+            403,
+            'WORKSPACE_MEMBERSHIP_REQUIRED',
+            '当前账户没有可用工作空间',
+        );
+    return workspace;
+}