panda/qipai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(M09-A): 收口保洁任务事务与并发

Browse Source
This commit is contained in:
Codex
2026-08-10 14:48:04 +08:00
parent b35af9037f
commit 9b2f055ecd
2 changed files with 572 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cleaning/cleaning-task-repository.ts
+158 -112
View File
@@ -72,6 +72,7 @@ interface CountRow extends RowDataPacket { total: number }
interface StatusCountRow extends RowDataPacket { status: CleaningTaskStatus; total: number }
interface AmountRow extends RowDataPacket { amount: number | null }
interface CurrentStatusRow extends RowDataPacket { status: CleaningTaskStatus; cleanerUserId: string | null }
interface ReclaimRow extends CurrentStatusRow { id: string }
interface FinishedOrderRow extends RowDataPacket {
id: string;
orderNo: string;
@@ -232,16 +233,28 @@ export class CleaningTaskRepository {
async claim(input: CleaningActor & { taskId: string }) {
this.assertCleaner(input.access, 'write');
await this.assertStoreVisible(input, input.taskId);
const [result] = await this.pool.execute<ResultSetHeader>(
`UPDATE qipai_cleaning_tasks
SET status = 'CLAIMED', cleaner_user_id = ?, claimed_at = UTC_TIMESTAMP(3)
WHERE tenant_id = ? AND id = ? AND status = 'WAITING' AND cleaner_user_id IS NULL
AND deleted_at IS NULL`,
[input.userId, input.tenantId, input.taskId]
);
if (result.affectedRows !== 1) throw new CleaningTaskError('CLEANING_TASK_NOT_CLAIMABLE');
await this.ensureLeadMember(this.pool, input.tenantId, input.taskId);
await this.recordEvent(input, input.taskId, 'WAITING', 'CLAIMED', 'CLAIM', '');
await this.transaction(async (connection) => {
const [rows] = await connection.execute<CurrentStatusRow[]>(
`SELECT status, cleaner_user_id AS cleanerUserId
FROM qipai_cleaning_tasks
WHERE tenant_id = ? AND id = ? AND deleted_at IS NULL
FOR UPDATE`,
[input.tenantId, input.taskId]
);
if (rows[0]?.status !== 'WAITING' || rows[0].cleanerUserId) {
throw new CleaningTaskError('CLEANING_TASK_NOT_CLAIMABLE');
}
await connection.execute(
`UPDATE qipai_cleaning_tasks
SET status = 'CLAIMED', cleaner_user_id = ?, claimed_at = UTC_TIMESTAMP(3)
WHERE tenant_id = ? AND id = ?`,
[input.userId, input.tenantId, input.taskId]
);
await this.ensureLeadMember(connection, input.tenantId, input.taskId);
await this.recordEventWithConnection(
connection, input, input.taskId, 'WAITING', 'CLAIMED', 'CLAIM', ''
);
});
return this.getMineTask(input, input.taskId);
}
@@ -259,38 +272,58 @@ export class CleaningTaskRepository {
async assign(input: CleaningActor & { taskId: string; cleanerUserId: string; note?: string }) {
this.assertCleaner(input.access, 'write');
await this.assertStoreVisible(input, input.taskId);
await this.assertCleanerUserForTask(input.tenantId, input.taskId, input.cleanerUserId);
const [result] = await this.pool.execute<ResultSetHeader>(
`UPDATE qipai_cleaning_tasks
SET status = 'CLAIMED', cleaner_user_id = ?, claimed_at = COALESCE(claimed_at, UTC_TIMESTAMP(3)),
reject_reason = ''
WHERE tenant_id = ? AND id = ? AND status IN ('WAITING', 'CLAIMED', 'REJECTED')
AND deleted_at IS NULL`,
[input.cleanerUserId, input.tenantId, input.taskId]
);
if (result.affectedRows !== 1) throw new CleaningTaskError('CLEANING_TASK_ASSIGN_CONFLICT');
await this.ensureLeadMember(this.pool, input.tenantId, input.taskId);
await this.recordEvent(input, input.taskId, 'WAITING', 'CLAIMED', 'ASSIGN', input.note ?? '');
await this.transaction(async (connection) => {
const [rows] = await connection.execute<CurrentStatusRow[]>(
`SELECT status, cleaner_user_id AS cleanerUserId
FROM qipai_cleaning_tasks
WHERE tenant_id = ? AND id = ? AND deleted_at IS NULL
FOR UPDATE`,
[input.tenantId, input.taskId]
);
const current = rows[0];
if (!current || !['WAITING', 'CLAIMED', 'REJECTED'].includes(current.status)) {
throw new CleaningTaskError('CLEANING_TASK_ASSIGN_CONFLICT');
}
await this.assertCleanerUserForTask(
input.tenantId, input.taskId, input.cleanerUserId, connection
);
await connection.execute(
`UPDATE qipai_cleaning_tasks
SET status = 'CLAIMED', cleaner_user_id = ?, claimed_at = UTC_TIMESTAMP(3),
started_at = NULL, submitted_at = NULL, completed_at = NULL, settled_at = NULL,
cancelled_at = NULL, photo_urls_json = JSON_ARRAY(), reject_reason = ''
WHERE tenant_id = ? AND id = ?`,
[input.cleanerUserId, input.tenantId, input.taskId]
);
await this.ensureLeadMember(connection, input.tenantId, input.taskId);
await this.recordEventWithConnection(
connection, input, input.taskId, current.status, 'CLAIMED', 'ASSIGN', input.note ?? ''
);
});
return this.getTask(input, input.taskId);
}
async complete(input: CleaningActor & { taskId: string; note?: string }) {
const task = await this.moveManaged(input, 'SUBMITTED', 'COMPLETED', 'COMPLETE', 'completed_at', input.note ?? '');
await this.ensureLeadMember(this.pool, input.tenantId, input.taskId);
return task;
return this.moveManaged(
input, 'SUBMITTED', 'COMPLETED', 'COMPLETE', 'completed_at', input.note ?? ''
);
}
async reject(input: CleaningActor & { taskId: string; reason: string }) {
this.assertCleaner(input.access, 'write');
await this.assertStoreVisible(input, input.taskId);
const [result] = await this.pool.execute<ResultSetHeader>(
`UPDATE qipai_cleaning_tasks
SET status = 'REJECTED', reject_reason = ?
WHERE tenant_id = ? AND id = ? AND status = 'SUBMITTED' AND deleted_at IS NULL`,
[input.reason.slice(0, 512), input.tenantId, input.taskId]
);
if (result.affectedRows !== 1) throw new CleaningTaskError('CLEANING_TASK_STATUS_CONFLICT');
await this.recordEvent(input, input.taskId, 'SUBMITTED', 'REJECTED', 'REJECT', input.reason);
await this.transaction(async (connection) => {
const [result] = await connection.execute<ResultSetHeader>(
`UPDATE qipai_cleaning_tasks
SET status = 'REJECTED', reject_reason = ?
WHERE tenant_id = ? AND id = ? AND status = 'SUBMITTED' AND deleted_at IS NULL`,
[input.reason.slice(0, 512), input.tenantId, input.taskId]
);
if (result.affectedRows !== 1) throw new CleaningTaskError('CLEANING_TASK_STATUS_CONFLICT');
await this.recordEventWithConnection(
connection, input, input.taskId, 'SUBMITTED', 'REJECTED', 'REJECT', input.reason
);
});
return this.getTask(input, input.taskId);
}
@@ -753,54 +786,69 @@ export class CleaningTaskRepository {
async reclaimTimeouts(input: CleaningActor & { olderThanMinutes: number; limit: number }) {
this.assertCleaner(input.access, 'write');
const [rows] = await this.pool.execute<RowDataPacket[]>(
`SELECT id FROM qipai_cleaning_tasks t
WHERE t.tenant_id = ? AND t.status IN ('CLAIMED', 'STARTED')
AND t.updated_at < TIMESTAMPADD(MINUTE, -?, UTC_TIMESTAMP(3))
AND t.deleted_at IS NULL AND ${storeScopeSql(input.access, 't.store_id')}
ORDER BY t.updated_at ASC, t.id ASC
LIMIT ?`,
[input.tenantId, input.olderThanMinutes, input.limit]
);
const ids = rows.map((row) => String(row.id));
if (ids.length === 0) return { reclaimed: 0, taskIds: [] as string[] };
await this.pool.execute(
`UPDATE qipai_cleaning_tasks
SET status = 'WAITING', cleaner_user_id = NULL, claimed_at = NULL,
started_at = NULL, reject_reason = ''
WHERE tenant_id = ? AND id IN (${ids.map(() => '?').join(',')})`,
[input.tenantId, ...ids]
);
for (const taskId of ids) {
await this.recordEvent(input, taskId, 'CLAIMED', 'WAITING', 'TIMEOUT_RECLAIM', '');
}
return { reclaimed: ids.length, taskIds: ids };
const safeLimit = Math.max(1, Math.min(100, Math.trunc(input.limit)));
return this.transaction(async (connection) => {
const [rows] = await connection.execute<ReclaimRow[]>(
`SELECT t.id, t.status, t.cleaner_user_id AS cleanerUserId
FROM qipai_cleaning_tasks t
WHERE t.tenant_id = ? AND t.status IN ('CLAIMED', 'STARTED')
AND t.updated_at < TIMESTAMPADD(MINUTE, -?, UTC_TIMESTAMP(3))
AND t.deleted_at IS NULL AND ${storeScopeSql(input.access, 't.store_id')}
ORDER BY t.updated_at ASC, t.id ASC
LIMIT ${safeLimit} FOR UPDATE SKIP LOCKED`,
[input.tenantId, input.olderThanMinutes]
);
for (const row of rows) {
const taskId = String(row.id);
await connection.execute(
`UPDATE qipai_cleaning_tasks
SET status = 'WAITING', cleaner_user_id = NULL, claimed_at = NULL,
started_at = NULL, submitted_at = NULL, completed_at = NULL,
settled_at = NULL, cancelled_at = NULL, photo_urls_json = JSON_ARRAY(),
reject_reason = ''
WHERE tenant_id = ? AND id = ? AND status = ?`,
[input.tenantId, taskId, row.status]
);
await connection.execute(
`UPDATE qipai_cleaning_task_members
SET removed_at = UTC_TIMESTAMP(3), reward_cents = 0, updated_at = UTC_TIMESTAMP(3)
WHERE tenant_id = ? AND task_id = ? AND removed_at IS NULL AND settled_at IS NULL`,
[input.tenantId, taskId]
);
await this.recordEventWithConnection(
connection, input, taskId, row.status, 'WAITING', 'TIMEOUT_RECLAIM', ''
);
}
return { reclaimed: rows.length, taskIds: rows.map((row) => String(row.id)) };
});
}
async submit(input: CleaningActor & { taskId: string; photoUrls: string[]; note?: string }) {
this.assertCleaner(input.access, 'write');
if (input.photoUrls.length === 0) throw new CleaningTaskError('CLEANING_PHOTO_REQUIRED');
const [rows] = await this.pool.execute<CurrentStatusRow[]>(
`SELECT status, cleaner_user_id AS cleanerUserId
FROM qipai_cleaning_tasks
WHERE tenant_id = ? AND id = ? AND cleaner_user_id = ? AND deleted_at IS NULL`,
[input.tenantId, input.taskId, input.userId]
);
const current = rows[0];
if (!current || !['STARTED', 'REJECTED'].includes(current.status)) {
throw new CleaningTaskError('CLEANING_TASK_STATUS_CONFLICT');
}
const [result] = await this.pool.execute<ResultSetHeader>(
`UPDATE qipai_cleaning_tasks
SET status = 'SUBMITTED', submitted_at = UTC_TIMESTAMP(3),
photo_urls_json = ?, reject_reason = ''
WHERE tenant_id = ? AND id = ? AND cleaner_user_id = ? AND status = ?
AND deleted_at IS NULL`,
[JSON.stringify(input.photoUrls.slice(0, 9)), input.tenantId, input.taskId,
input.userId, current.status]
);
if (result.affectedRows !== 1) throw new CleaningTaskError('CLEANING_TASK_STATUS_CONFLICT');
await this.recordEvent(input, input.taskId, current.status, 'SUBMITTED', 'SUBMIT', input.note ?? '');
await this.transaction(async (connection) => {
const [rows] = await connection.execute<CurrentStatusRow[]>(
`SELECT status, cleaner_user_id AS cleanerUserId
FROM qipai_cleaning_tasks
WHERE tenant_id = ? AND id = ? AND cleaner_user_id = ? AND deleted_at IS NULL
FOR UPDATE`,
[input.tenantId, input.taskId, input.userId]
);
const current = rows[0];
if (!current || !['STARTED', 'REJECTED'].includes(current.status)) {
throw new CleaningTaskError('CLEANING_TASK_STATUS_CONFLICT');
}
await connection.execute(
`UPDATE qipai_cleaning_tasks
SET status = 'SUBMITTED', submitted_at = UTC_TIMESTAMP(3),
photo_urls_json = ?, reject_reason = ''
WHERE tenant_id = ? AND id = ?`,
[JSON.stringify(input.photoUrls.slice(0, 9)), input.tenantId, input.taskId]
);
await this.recordEventWithConnection(
connection, input, input.taskId, current.status, 'SUBMITTED', 'SUBMIT', input.note ?? ''
);
});
return this.getMineTask(input, input.taskId);
}
@@ -1104,14 +1152,19 @@ export class CleaningTaskRepository {
) {
this.assertCleaner(input.access, 'write');
await this.assertStoreVisible(input, input.taskId);
const [result] = await this.pool.execute<ResultSetHeader>(
`UPDATE qipai_cleaning_tasks
SET status = ?, ${timestampColumn} = UTC_TIMESTAMP(3)
WHERE tenant_id = ? AND id = ? AND status = ? AND deleted_at IS NULL`,
[to, input.tenantId, input.taskId, from]
);
if (result.affectedRows !== 1) throw new CleaningTaskError('CLEANING_TASK_STATUS_CONFLICT');
await this.recordEvent(input, input.taskId, from, to, action, note);
await this.transaction(async (connection) => {
const [result] = await connection.execute<ResultSetHeader>(
`UPDATE qipai_cleaning_tasks
SET status = ?, ${timestampColumn} = UTC_TIMESTAMP(3)
WHERE tenant_id = ? AND id = ? AND status = ? AND deleted_at IS NULL`,
[to, input.tenantId, input.taskId, from]
);
if (result.affectedRows !== 1) throw new CleaningTaskError('CLEANING_TASK_STATUS_CONFLICT');
await this.ensureLeadMember(connection, input.tenantId, input.taskId);
await this.recordEventWithConnection(
connection, input, input.taskId, from, to, action, note
);
});
return this.getTask(input, input.taskId);
}
@@ -1136,15 +1189,19 @@ export class CleaningTaskRepository {
extraParams.push(extra.reject_reason);
}
const setExtra = extraAssignments.length > 0 ? `, ${extraAssignments.join(', ')}` : '';
const [result] = await this.pool.execute<ResultSetHeader>(
`UPDATE qipai_cleaning_tasks
SET status = ?, ${timestampColumn} = UTC_TIMESTAMP(3)${setExtra}
WHERE tenant_id = ? AND id = ? AND cleaner_user_id = ? AND status = ?
AND deleted_at IS NULL`,
[to, ...extraParams, input.tenantId, input.taskId, input.userId, from]
);
if (result.affectedRows !== 1) throw new CleaningTaskError('CLEANING_TASK_STATUS_CONFLICT');
await this.recordEvent(input, input.taskId, from, to, action, note);
await this.transaction(async (connection) => {
const [result] = await connection.execute<ResultSetHeader>(
`UPDATE qipai_cleaning_tasks
SET status = ?, ${timestampColumn} = UTC_TIMESTAMP(3)${setExtra}
WHERE tenant_id = ? AND id = ? AND cleaner_user_id = ? AND status = ?
AND deleted_at IS NULL`,
[to, ...extraParams, input.tenantId, input.taskId, input.userId, from]
);
if (result.affectedRows !== 1) throw new CleaningTaskError('CLEANING_TASK_STATUS_CONFLICT');
await this.recordEventWithConnection(
connection, input, input.taskId, from, to, action, note
);
});
return this.getMineTask(input, input.taskId);
}
@@ -1236,8 +1293,13 @@ export class CleaningTaskRepository {
}
}
private async assertCleanerUserForTask(tenantId: string, taskId: string, cleanerUserId: string) {
const [rows] = await this.pool.execute<RowDataPacket[]>(
private async assertCleanerUserForTask(
tenantId: string,
taskId: string,
cleanerUserId: string,
connection: Pick<MySqlPool, 'execute'> | PoolConnection = this.pool
) {
const [rows] = await connection.execute<RowDataPacket[]>(
`SELECT 1
FROM qipai_cleaning_tasks t
INNER JOIN qipai_users u ON u.tenant_id = t.tenant_id AND u.id = ?
@@ -1254,22 +1316,6 @@ export class CleaningTaskRepository {
if (!rows[0]) throw new CleaningTaskError('CLEANING_ASSIGNEE_INVALID');
}
private async recordEvent(
input: CleaningActor,
taskId: string,
from: CleaningTaskStatus,
to: CleaningTaskStatus,
action: string,
note: string
) {
await this.pool.execute(
`INSERT IGNORE INTO qipai_cleaning_task_events
(tenant_id, task_id, from_status, to_status, action, actor_id, trace_id, note, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, JSON_OBJECT())`,
[input.tenantId, taskId, from, to, action, input.userId, input.traceId, note.slice(0, 512)]
);
}
private async recordEventWithConnection(
connection: PoolConnection,
input: CleaningActor,
@@ -1280,7 +1326,7 @@ export class CleaningTaskRepository {
note: string
) {
await connection.execute(
`INSERT IGNORE INTO qipai_cleaning_task_events
`INSERT INTO qipai_cleaning_task_events
(tenant_id, task_id, from_status, to_status, action, actor_id, trace_id, note, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, JSON_OBJECT())`,
[input.tenantId, taskId, from, to, action, input.userId,