From 9b2f055ecdf635497a6d86f998c65f3e8c4ed7b0 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 10 Aug 2026 14:48:04 +0800 Subject: [PATCH] =?UTF-8?q?fix(M09-A):=20=E6=94=B6=E5=8F=A3=E4=BF=9D?= =?UTF-8?q?=E6=B4=81=E4=BB=BB=E5=8A=A1=E4=BA=8B=E5=8A=A1=E4=B8=8E=E5=B9=B6?= =?UTF-8?q?=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/cleaning/cleaning-task-repository.ts | 270 +++++++----- .../tests/mysql-migration-roundtrip.test.mjs | 414 ++++++++++++++++++ 2 files changed, 572 insertions(+), 112 deletions(-) diff --git a/backend/src/cleaning/cleaning-task-repository.ts b/backend/src/cleaning/cleaning-task-repository.ts index fed21e2..876c10a 100644 --- a/backend/src/cleaning/cleaning-task-repository.ts +++ b/backend/src/cleaning/cleaning-task-repository.ts @@ -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( - `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( + `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( - `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( + `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( - `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( + `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( - `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( + `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( - `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( - `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( + `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( - `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( + `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( - `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( + `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( + private async assertCleanerUserForTask( + tenantId: string, + taskId: string, + cleanerUserId: string, + connection: Pick | PoolConnection = this.pool + ) { + const [rows] = await connection.execute( `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, diff --git a/backend/tests/mysql-migration-roundtrip.test.mjs b/backend/tests/mysql-migration-roundtrip.test.mjs index da9e7cd..b1aed28 100644 --- a/backend/tests/mysql-migration-roundtrip.test.mjs +++ b/backend/tests/mysql-migration-roundtrip.test.mjs @@ -45,6 +45,9 @@ import { BusinessStatisticsRepository } from '../dist/operations/business-statis import { SystemOperationsRepository } from '../dist/operations/system-operations-repository.js'; import { AdminAuthRepository, AdminAuthError, hashToken } from '../dist/auth/admin-auth-repository.js'; import { hashPassword } from '../dist/auth/password.js'; +import { + CleaningTaskError, CleaningTaskRepository +} from '../dist/cleaning/cleaning-task-repository.js'; import { executeMigrationPlan, loadMigrationPlan, @@ -2129,6 +2132,408 @@ async function assertIotMessages(pool, context) { assert.deepEqual(linkRows, [{ subId: 'SUB-AUTO-001', subtype: '14', model: '701C' }]); } +async function assertCleaningTaskTransactions(pool, context) { + const [scopeRows] = await pool.query( + `SELECT s.id AS storeId, r.id AS roomId + FROM qipai_stores s + INNER JOIN qipai_rooms r ON r.tenant_id = s.tenant_id AND r.store_id = s.id + WHERE s.tenant_id = ? AND s.name = 'M03A Store' + ORDER BY r.id LIMIT 1`, + [context.tenantId] + ); + const [managerRows] = await pool.query( + `SELECT u.id + FROM qipai_users u + INNER JOIN qipai_user_roles ur ON ur.tenant_id = u.tenant_id AND ur.user_id = u.id + INNER JOIN qipai_roles r ON r.tenant_id = ur.tenant_id AND r.id = ur.role_id + WHERE u.tenant_id = ? AND r.code = 'TENANT_ADMIN' + ORDER BY u.id LIMIT 1`, + [context.tenantId] + ); + assert.ok(scopeRows[0], 'M09-A cleaning test requires the M03-A store and room.'); + assert.ok(managerRows[0], 'M09-A cleaning test requires a tenant administrator.'); + const storeId = String(scopeRows[0].storeId); + const roomId = String(scopeRows[0].roomId); + const managerId = String(managerRows[0].id); + const managerAccess = await new RbacRepository(pool).getAccessProfile(context.tenantId, managerId); + assert.ok(managerAccess.capabilities.includes('tenant.manage')); + + const createCleaner = async (nickname, phone) => { + const [result] = await pool.query( + `INSERT INTO qipai_users (tenant_id, user_type, nickname, phone) + VALUES (?, 'STAFF', ?, ?)`, + [context.tenantId, nickname, phone] + ); + const cleanerId = String(result.insertId); + await pool.query( + `INSERT INTO qipai_user_roles (tenant_id, user_id, role_id) + SELECT ?, ?, id FROM qipai_roles + WHERE tenant_id = ? AND code = 'CLEANER'`, + [context.tenantId, cleanerId, context.tenantId] + ); + await pool.query( + `INSERT INTO qipai_user_store_scopes (tenant_id, user_id, store_id, scope_type) + VALUES (?, ?, ?, 'STAFF')`, + [context.tenantId, cleanerId, storeId] + ); + return cleanerId; + }; + const firstCleanerId = await createCleaner('M09A Cleaner One', '13900000901'); + const secondCleanerId = await createCleaner('M09A Cleaner Two', '13900000902'); + const cleanerAccess = { + roles: ['CLEANER'], + capabilities: ['cleaning.task.read', 'cleaning.task.write'], + storeIds: [storeId] + }; + const cleanerActor = (userId, traceId) => ({ + tenantId: context.tenantId, + userId, + access: cleanerAccess, + traceId + }); + const managerActor = (traceId) => ({ + tenantId: context.tenantId, + userId: managerId, + access: managerAccess, + traceId + }); + const repository = new CleaningTaskRepository(pool); + let taskSequence = 0; + const insertTask = async ({ + status = 'WAITING', cleanerUserId = null, rewardCents = 100, + photoUrls = [], rejectReason = '', claimedAt = null, startedAt = null, + submittedAt = null, completedAt = null, settledAt = null, cancelledAt = null, + updatedAt = new Date() + } = {}) => { + taskSequence += 1; + const [result] = await pool.query( + `INSERT INTO qipai_cleaning_tasks + (tenant_id, store_id, room_id, task_no, status, cleaner_user_id, + priority, reward_cents, requirement, photo_urls_json, reject_reason, + claimed_at, started_at, submitted_at, completed_at, settled_at, cancelled_at, + updated_at) + VALUES (?, ?, ?, ?, ?, ?, 5, ?, 'M09-A transaction test', ?, ?, + ?, ?, ?, ?, ?, ?, ?)`, + [ + context.tenantId, storeId, roomId, `M09A-${Date.now()}-${taskSequence}`, + status, cleanerUserId, rewardCents, JSON.stringify(photoUrls), rejectReason, + claimedAt, startedAt, submittedAt, completedAt, settledAt, cancelledAt, updatedAt + ] + ); + return String(result.insertId); + }; + const insertMember = async ({ + taskId, userId, memberRole = 'LEAD', rewardCents = 0, + removedAt = null, settledAt = null + }) => { + await pool.query( + `INSERT INTO qipai_cleaning_task_members + (tenant_id, task_id, user_id, member_role, reward_cents, joined_at, removed_at, settled_at) + VALUES (?, ?, ?, ?, ?, UTC_TIMESTAMP(3), ?, ?)`, + [context.tenantId, taskId, userId, memberRole, rewardCents, removedAt, settledAt] + ); + }; + + const rollbackTaskId = await insertTask(); + const rollbackTrace = `m09a-event-failure-${rollbackTaskId}`; + await pool.query( + `INSERT INTO qipai_cleaning_task_events + (tenant_id, task_id, from_status, to_status, action, actor_id, trace_id, note, metadata) + VALUES (?, ?, 'WAITING', 'WAITING', 'TEST_BLOCKER', ?, ?, '', JSON_OBJECT())`, + [context.tenantId, rollbackTaskId, firstCleanerId, rollbackTrace] + ); + await assert.rejects( + () => repository.claim({ + ...cleanerActor(firstCleanerId, 'm09a-event-failure'), taskId: rollbackTaskId + }), + (error) => error?.code === 'ER_DUP_ENTRY' + ); + const [rollbackRows] = await pool.query( + `SELECT t.status, t.cleaner_user_id AS cleanerUserId, + (SELECT COUNT(*) FROM qipai_cleaning_task_members m + WHERE m.tenant_id = t.tenant_id AND m.task_id = t.id) AS memberCount + FROM qipai_cleaning_tasks t WHERE t.tenant_id = ? AND t.id = ?`, + [context.tenantId, rollbackTaskId] + ); + assert.equal(rollbackRows[0].status, 'WAITING'); + assert.equal(rollbackRows[0].cleanerUserId, null); + assert.equal(Number(rollbackRows[0].memberCount), 0); + + const concurrentTaskId = await insertTask(); + const claimResults = await Promise.allSettled([ + repository.claim({ + ...cleanerActor(firstCleanerId, 'm09a-concurrent-claim-one'), taskId: concurrentTaskId + }), + repository.claim({ + ...cleanerActor(secondCleanerId, 'm09a-concurrent-claim-two'), taskId: concurrentTaskId + }) + ]); + assert.equal(claimResults.filter((result) => result.status === 'fulfilled').length, 1); + const rejectedClaim = claimResults.find((result) => result.status === 'rejected'); + assert.ok(rejectedClaim?.reason instanceof CleaningTaskError); + assert.equal(rejectedClaim.reason.code, 'CLEANING_TASK_NOT_CLAIMABLE'); + const [claimedRows] = await pool.query( + `SELECT t.status, t.cleaner_user_id AS cleanerUserId, + (SELECT COUNT(*) FROM qipai_cleaning_task_events e + WHERE e.tenant_id = t.tenant_id AND e.task_id = t.id AND e.action = 'CLAIM') AS eventCount, + (SELECT COUNT(*) FROM qipai_cleaning_task_members m + WHERE m.tenant_id = t.tenant_id AND m.task_id = t.id + AND m.member_role = 'LEAD' AND m.removed_at IS NULL) AS leadCount + FROM qipai_cleaning_tasks t WHERE t.tenant_id = ? AND t.id = ?`, + [context.tenantId, concurrentTaskId] + ); + assert.equal(claimedRows[0].status, 'CLAIMED'); + assert.ok([firstCleanerId, secondCleanerId].includes(String(claimedRows[0].cleanerUserId))); + assert.equal(Number(claimedRows[0].eventCount), 1); + assert.equal(Number(claimedRows[0].leadCount), 1); + + const lifecycleTaskId = await insertTask(); + await repository.assign({ + ...managerActor('m09a-assign-waiting'), taskId: lifecycleTaskId, + cleanerUserId: firstCleanerId, note: 'WAITING assignment' + }); + await repository.assign({ + ...managerActor('m09a-assign-claimed'), taskId: lifecycleTaskId, + cleanerUserId: secondCleanerId, note: 'CLAIMED reassignment' + }); + const expiredAt = new Date(Date.now() - 3_600_000); + await pool.query( + `UPDATE qipai_cleaning_tasks + SET status = 'REJECTED', photo_urls_json = JSON_ARRAY('https://old.example/photo.jpg'), + reject_reason = 'old rejection', started_at = ?, submitted_at = ?, completed_at = ?, + settled_at = ?, cancelled_at = ? + WHERE tenant_id = ? AND id = ?`, + [expiredAt, expiredAt, expiredAt, expiredAt, expiredAt, context.tenantId, lifecycleTaskId] + ); + await repository.assign({ + ...managerActor('m09a-assign-rejected'), taskId: lifecycleTaskId, + cleanerUserId: firstCleanerId, note: 'REJECTED reassignment' + }); + const [resetRows] = await pool.query( + `SELECT status, cleaner_user_id AS cleanerUserId, + claimed_at IS NOT NULL AS claimedSet, + started_at IS NULL AS startedCleared, + submitted_at IS NULL AS submittedCleared, + completed_at IS NULL AS completedCleared, + settled_at IS NULL AS settledCleared, + cancelled_at IS NULL AS cancelledCleared, + JSON_LENGTH(photo_urls_json) AS photoCount, reject_reason AS rejectReason + FROM qipai_cleaning_tasks WHERE tenant_id = ? AND id = ?`, + [context.tenantId, lifecycleTaskId] + ); + assert.deepEqual({ + status: resetRows[0].status, + cleanerUserId: String(resetRows[0].cleanerUserId), + claimedSet: Number(resetRows[0].claimedSet), + startedCleared: Number(resetRows[0].startedCleared), + submittedCleared: Number(resetRows[0].submittedCleared), + completedCleared: Number(resetRows[0].completedCleared), + settledCleared: Number(resetRows[0].settledCleared), + cancelledCleared: Number(resetRows[0].cancelledCleared), + photoCount: Number(resetRows[0].photoCount), + rejectReason: resetRows[0].rejectReason + }, { + status: 'CLAIMED', cleanerUserId: firstCleanerId, claimedSet: 1, + startedCleared: 1, submittedCleared: 1, completedCleared: 1, + settledCleared: 1, cancelledCleared: 1, photoCount: 0, rejectReason: '' + }); + const [reassignedMemberRows] = await pool.query( + `SELECT user_id AS userId, member_role AS memberRole, reward_cents AS rewardCents, + removed_at IS NOT NULL AS removed + FROM qipai_cleaning_task_members + WHERE tenant_id = ? AND task_id = ? ORDER BY user_id`, + [context.tenantId, lifecycleTaskId] + ); + assert.deepEqual(reassignedMemberRows.map((row) => ({ + userId: String(row.userId), memberRole: row.memberRole, + rewardCents: Number(row.rewardCents), removed: Number(row.removed) + })), [ + { userId: firstCleanerId, memberRole: 'LEAD', rewardCents: 100, removed: 0 }, + { userId: secondCleanerId, memberRole: 'LEAD', rewardCents: 0, removed: 1 } + ].sort((left, right) => Number(left.userId) - Number(right.userId))); + + await repository.start({ + ...cleanerActor(firstCleanerId, 'm09a-start'), taskId: lifecycleTaskId + }); + await repository.submit({ + ...cleanerActor(firstCleanerId, 'm09a-submit-first'), taskId: lifecycleTaskId, + photoUrls: ['https://api.txyundm.cn/uploads/m09a-first.webp'], note: 'first submit' + }); + await repository.reject({ + ...managerActor('m09a-reject'), taskId: lifecycleTaskId, reason: 'needs rework' + }); + const [rejectedRows] = await pool.query( + `SELECT status, reject_reason AS rejectReason + FROM qipai_cleaning_tasks WHERE tenant_id = ? AND id = ?`, + [context.tenantId, lifecycleTaskId] + ); + assert.deepEqual(rejectedRows, [{ status: 'REJECTED', rejectReason: 'needs rework' }]); + await repository.rework({ + ...cleanerActor(firstCleanerId, 'm09a-rework'), taskId: lifecycleTaskId + }); + await repository.submit({ + ...cleanerActor(firstCleanerId, 'm09a-submit-second'), taskId: lifecycleTaskId, + photoUrls: ['https://api.txyundm.cn/uploads/m09a-second.webp'], note: 'second submit' + }); + const completeInput = { + ...managerActor('m09a-complete'), taskId: lifecycleTaskId, note: 'accepted' + }; + await repository.complete(completeInput); + await assert.rejects( + () => repository.complete(completeInput), + (error) => error instanceof CleaningTaskError + && error.code === 'CLEANING_TASK_STATUS_CONFLICT' + ); + const [lifecycleRows] = await pool.query( + `SELECT status, completed_at IS NOT NULL AS completedSet + FROM qipai_cleaning_tasks WHERE tenant_id = ? AND id = ?`, + [context.tenantId, lifecycleTaskId] + ); + assert.deepEqual({ + status: lifecycleRows[0].status, + completedSet: Number(lifecycleRows[0].completedSet) + }, { status: 'COMPLETED', completedSet: 1 }); + const [lifecycleEventRows] = await pool.query( + `SELECT action FROM qipai_cleaning_task_events + WHERE tenant_id = ? AND task_id = ? ORDER BY id`, + [context.tenantId, lifecycleTaskId] + ); + assert.deepEqual(lifecycleEventRows.map((row) => row.action), [ + 'ASSIGN', 'ASSIGN', 'ASSIGN', 'START', 'SUBMIT', 'REJECT', 'REWORK', 'SUBMIT', 'COMPLETE' + ]); + + const staleBase = Date.now() - 24 * 3_600_000; + const lockedTaskId = await insertTask({ + status: 'CLAIMED', cleanerUserId: firstCleanerId, + claimedAt: new Date(staleBase), updatedAt: new Date(staleBase) + }); + const reclaimTaskId = await insertTask({ + status: 'STARTED', cleanerUserId: secondCleanerId, + claimedAt: new Date(staleBase + 3_600_000), startedAt: new Date(staleBase + 3_600_000), + photoUrls: ['https://old.example/reclaim.jpg'], rejectReason: 'stale', + updatedAt: new Date(staleBase + 3_600_000) + }); + const outsideBatchTaskId = await insertTask({ + status: 'CLAIMED', cleanerUserId: firstCleanerId, + claimedAt: new Date(staleBase + 7_200_000), updatedAt: new Date(staleBase + 7_200_000) + }); + const settledTaskId = await insertTask({ + status: 'SETTLED', cleanerUserId: secondCleanerId, + claimedAt: new Date(staleBase - 3_600_000), completedAt: new Date(staleBase - 3_600_000), + settledAt: new Date(staleBase - 3_600_000), updatedAt: new Date(staleBase - 3_600_000) + }); + await insertMember({ taskId: lockedTaskId, userId: firstCleanerId, rewardCents: 100 }); + await insertMember({ taskId: reclaimTaskId, userId: secondCleanerId, rewardCents: 70 }); + await insertMember({ + taskId: reclaimTaskId, userId: firstCleanerId, memberRole: 'ASSIST', rewardCents: 30 + }); + await insertMember({ taskId: outsideBatchTaskId, userId: firstCleanerId, rewardCents: 100 }); + await insertMember({ + taskId: settledTaskId, userId: secondCleanerId, rewardCents: 100, + settledAt: new Date(staleBase) + }); + + const lockConnection = await pool.getConnection(); + let reclaimed; + try { + await lockConnection.beginTransaction(); + await lockConnection.query( + `SELECT id FROM qipai_cleaning_tasks + WHERE tenant_id = ? AND id = ? FOR UPDATE`, + [context.tenantId, lockedTaskId] + ); + reclaimed = await repository.reclaimTimeouts({ + ...managerActor('m09a-reclaim'), olderThanMinutes: 30, limit: 1 + }); + } finally { + await lockConnection.rollback(); + lockConnection.release(); + } + assert.deepEqual(reclaimed, { reclaimed: 1, taskIds: [reclaimTaskId] }); + const [reclaimRows] = await pool.query( + `SELECT id, status, cleaner_user_id AS cleanerUserId, + JSON_LENGTH(photo_urls_json) AS photoCount, reject_reason AS rejectReason + FROM qipai_cleaning_tasks + WHERE tenant_id = ? AND id IN (?, ?, ?, ?) ORDER BY id`, + [context.tenantId, lockedTaskId, reclaimTaskId, outsideBatchTaskId, settledTaskId] + ); + const reclaimById = new Map(reclaimRows.map((row) => [String(row.id), row])); + assert.equal(reclaimById.get(lockedTaskId).status, 'CLAIMED'); + assert.equal(reclaimById.get(outsideBatchTaskId).status, 'CLAIMED'); + assert.equal(reclaimById.get(settledTaskId).status, 'SETTLED'); + assert.equal(reclaimById.get(reclaimTaskId).status, 'WAITING'); + assert.equal(reclaimById.get(reclaimTaskId).cleanerUserId, null); + assert.equal(Number(reclaimById.get(reclaimTaskId).photoCount), 0); + assert.equal(reclaimById.get(reclaimTaskId).rejectReason, ''); + const [reclaimedMembers] = await pool.query( + `SELECT reward_cents AS rewardCents, removed_at IS NOT NULL AS removed + FROM qipai_cleaning_task_members + WHERE tenant_id = ? AND task_id = ? ORDER BY user_id`, + [context.tenantId, reclaimTaskId] + ); + assert.deepEqual(reclaimedMembers.map((row) => ({ + rewardCents: Number(row.rewardCents), removed: Number(row.removed) + })), [ + { rewardCents: 0, removed: 1 }, { rewardCents: 0, removed: 1 } + ]); + const [settledMemberRows] = await pool.query( + `SELECT reward_cents AS rewardCents, removed_at AS removedAt, + settled_at IS NOT NULL AS settled + FROM qipai_cleaning_task_members + WHERE tenant_id = ? AND task_id = ?`, + [context.tenantId, settledTaskId] + ); + assert.equal(Number(settledMemberRows[0].rewardCents), 100); + assert.equal(settledMemberRows[0].removedAt, null); + assert.equal(Number(settledMemberRows[0].settled), 1); + + const [orderResult] = await pool.query( + `INSERT INTO qipai_orders + (tenant_id, store_id, room_id, order_no, status, start_at, end_at, + total_amount_cents, paid_amount_cents) + VALUES (?, ?, ?, ?, 'FINISHED', ?, ?, 1000, 1000)`, + [ + context.tenantId, storeId, roomId, `M09A-FINISHED-${Date.now()}`, + new Date(Date.now() - 7_200_000), new Date(Date.now() - 3_600_000) + ] + ); + const finishedOrderId = String(orderResult.insertId); + const orderConnection = await pool.getConnection(); + try { + await orderConnection.beginTransaction(); + const creationInput = { + tenantId: context.tenantId, + orderId: finishedOrderId, + actorId: managerId, + traceId: 'm09a-finished-order' + }; + await repository.createForFinishedOrder(orderConnection, creationInput); + await repository.createForFinishedOrder(orderConnection, creationInput); + await orderConnection.commit(); + } catch (error) { + await orderConnection.rollback(); + throw error; + } finally { + orderConnection.release(); + } + const [generatedRows] = await pool.query( + `SELECT t.status, + (SELECT COUNT(*) FROM qipai_cleaning_task_events e + WHERE e.tenant_id = t.tenant_id AND e.task_id = t.id + AND e.action = 'AUTO_CREATE') AS eventCount + FROM qipai_cleaning_tasks t + WHERE t.tenant_id = ? AND t.order_id = ?`, + [context.tenantId, finishedOrderId] + ); + assert.equal(generatedRows.length, 1); + assert.equal(generatedRows[0].status, 'WAITING'); + assert.equal(Number(generatedRows[0].eventCount), 1); + + console.log( + 'PASS: M09-A cleaning claims, lifecycle events, rollback, reassignment, timeout reclaim and order generation are transactional.' + ); +} + const config = loadConfig(); assert.equal(config.mysql.passwordConfigured, true, 'Live migration test requires a temporary password.'); assert.match( @@ -2197,6 +2602,7 @@ try { await assertProfitSharingDomain(pool, loginContext); await assertDeviceTopology(pool, loginContext); await assertIotMessages(pool, loginContext); + await assertCleaningTaskTransactions(pool, loginContext); await assertLegacyCompatibility(pool); console.log('PASS: first up, verify, tenant isolation and revocable auth checks completed.'); @@ -2351,6 +2757,14 @@ try { 'door record credential hashing and masking', 'low-battery alert upsert', 'AddDevice ACK creates 701C parent-child topology' + , + 'concurrent cleaning claim has one winner', + 'cleaning state and event transaction rollback', + 'WAITING CLAIMED and REJECTED reassignment cleanup', + 'idempotent cleaning completion trace', + 'SKIP LOCKED cleaning timeout reclaim batch', + 'settled cleaning member preservation', + 'finished order creates one cleaning task' ] }, null, 2)); } finally {