R524809
134 lines
5.4 KiB
TypeScript
134 lines
5.4 KiB
TypeScript
import { createHmac } from 'node:crypto';
|
|
import { Inject, Injectable } from '@nestjs/common';
|
|
import type { Database, DatabaseTransaction } from '@atlas/database';
|
|
import { APP_CONFIG, type AppConfig } from '../config/app-config.module.js';
|
|
import { DATABASE } from '../database/database.module.js';
|
|
import { ApiError } from './api-error.js';
|
|
|
|
type IdempotentResponse = { status: number; body: unknown };
|
|
type IdempotencyInput = {
|
|
workspaceId: string;
|
|
userId: string;
|
|
method: 'POST' | 'PUT' | 'PATCH' | 'DELETE';
|
|
path: string;
|
|
key: string;
|
|
request: unknown;
|
|
};
|
|
type IdempotencyRow = {
|
|
request_fingerprint: string;
|
|
status: 'in_progress' | 'completed';
|
|
response_status: number | null;
|
|
response_body: unknown;
|
|
};
|
|
|
|
@Injectable()
|
|
export class IdempotencyService {
|
|
public constructor(
|
|
@Inject(DATABASE) private readonly database: Database,
|
|
@Inject(APP_CONFIG) private readonly config: AppConfig,
|
|
) {}
|
|
|
|
/**
|
|
* 幂等记录与业务写入共用同一个工作空间事务。只有两者都提交，重试才会得到可靠的历史响应。
|
|
*/
|
|
public async run(
|
|
input: IdempotencyInput,
|
|
work: (transaction: DatabaseTransaction) => Promise<IdempotentResponse>,
|
|
): Promise<IdempotentResponse> {
|
|
const fingerprint = createHmac(
|
|
'sha256',
|
|
this.config.IDEMPOTENCY_KEY_SECRET,
|
|
)
|
|
.update(canonicalize(input.request))
|
|
.digest('base64url');
|
|
|
|
return this.database.workspaceTransaction(
|
|
input.workspaceId,
|
|
async (transaction) => {
|
|
const inserted = await transaction.query<IdempotencyRow>(
|
|
`INSERT INTO atlas.idempotency_requests
|
|
(workspace_id, user_id, http_method, request_path, idempotency_key, request_fingerprint)
|
|
VALUES ($1, $2, $3, $4, $5, $6)
|
|
ON CONFLICT (workspace_id, user_id, http_method, request_path, idempotency_key) DO NOTHING
|
|
RETURNING request_fingerprint, status, response_status, response_body`,
|
|
[
|
|
input.workspaceId,
|
|
input.userId,
|
|
input.method,
|
|
input.path,
|
|
input.key,
|
|
fingerprint,
|
|
],
|
|
);
|
|
if (!inserted.rows[0]) {
|
|
// 冲突插入会等待首个事务结束；随后行锁确保只能读取完整的持久化结果。
|
|
const existing = await transaction.query<IdempotencyRow>(
|
|
`SELECT request_fingerprint, status, response_status, response_body
|
|
FROM atlas.idempotency_requests
|
|
WHERE workspace_id = $1 AND user_id = $2 AND http_method = $3
|
|
AND request_path = $4 AND idempotency_key = $5
|
|
FOR UPDATE`,
|
|
[
|
|
input.workspaceId,
|
|
input.userId,
|
|
input.method,
|
|
input.path,
|
|
input.key,
|
|
],
|
|
);
|
|
const row = existing.rows[0];
|
|
if (!row || row.request_fingerprint !== fingerprint) {
|
|
throw new ApiError(
|
|
409,
|
|
'IDEMPOTENCY_KEY_REUSED',
|
|
'幂等键已用于不同请求',
|
|
);
|
|
}
|
|
if (row.status !== 'completed' || !row.response_status) {
|
|
throw new ApiError(
|
|
409,
|
|
'IDEMPOTENCY_REQUEST_IN_PROGRESS',
|
|
'相同请求仍在处理中',
|
|
);
|
|
}
|
|
return {
|
|
status: row.response_status,
|
|
body: row.response_body,
|
|
};
|
|
}
|
|
|
|
const response = await work(transaction);
|
|
await transaction.query(
|
|
`UPDATE atlas.idempotency_requests
|
|
SET status = 'completed', response_status = $1, response_body = $2,
|
|
completed_at = now()
|
|
WHERE workspace_id = $3 AND user_id = $4 AND http_method = $5
|
|
AND request_path = $6 AND idempotency_key = $7`,
|
|
[
|
|
response.status,
|
|
response.body,
|
|
input.workspaceId,
|
|
input.userId,
|
|
input.method,
|
|
input.path,
|
|
input.key,
|
|
],
|
|
);
|
|
return response;
|
|
},
|
|
);
|
|
}
|
|
}
|
|
|
|
/** 规范化排序后再签名，使 JSON 字段顺序不会改变同一请求的身份。 */
|
|
function canonicalize(value: unknown): string {
|
|
if (value === null || typeof value !== 'object')
|
|
return JSON.stringify(value);
|
|
if (Array.isArray(value)) return `[${value.map(canonicalize).join(',')}]`;
|
|
const record = value as Record<string, unknown>;
|
|
return `{${Object.keys(record)
|
|
.sort()
|
|
.map((key) => `${JSON.stringify(key)}:${canonicalize(record[key])}`)
|
|
.join(',')}}`;
|
|
}
|