excel唯一逻辑修复

control-plane/test/passenger-roster-target.test.ts

@@ -0,0 +1,156 @@
+import assert from 'node:assert/strict';
+import test, { type TestContext } from 'node:test';
+import { loadConfig } from '../src/config.js';
+import { sha256Bytes, sha256Text } from '../src/crypto.js';
+import { closePool, getPool } from '../src/db.js';
+import { TaskError, TaskService, type TaskContext } from '../src/task-service.js';
+
+const context: TaskContext = {
+  organizationId: 'org-a', userId: 'user-a', role: 'user',
+  source: 'agentbus', requestId: 'fixture', channelId: 'channel-a'
+};
+const selection = { conversationId: 'conversation-a', channelId: 'channel-a' };
+type Selection = Parameters<TaskService['attachPassengerRosterAttachment']>[2];
+
+function task(id: string) {
+  return {
+    id, task_id: `TASK-${id}`, business_route_id: 'passenger_list_import_shared_child',
+    status: 'awaiting_attachment', channel_id: 'channel-a', conversation_id: 'conversation-a',
+    organization_id: 'org-a', assigned_user_id: 'user-a'
+  };
+}
+function result(rows: Array<Record<string, unknown>> = []) {
+  return { rowCount: rows.length, rows };
+}
+function setup(t: TestContext) {
+  const config = loadConfig({ NODE_ENV: 'test', FIELD_ENCRYPTION_KEY: Buffer.alloc(32, 29).toString('base64'),
+    DATABASE_URL: 'postgresql://fixture:fixture@127.0.0.1:1/fixture' });
+  const pool = getPool(config);
+  const service = new TaskService(config);
+  t.after(async () => { t.mock.restoreAll(); await closePool(); });
+  const target = (input: Selection = selection, actor = context) =>
+    (service as unknown as {
+      passengerRosterAttachmentTarget(c: TaskContext, s: Selection): Promise<{ taskId: string }>;
+    }).passengerRosterAttachmentTarget(actor, input);
+  return { pool, service, target };
+}
+
+test('automatic roster target query uses newest creation time and preserves all scope filters', async t => {
+  const { pool, target } = setup(t);
+  t.mock.method(pool, 'query', async (sql: string, params: unknown[]) => {
+    assert.match(sql, /t\.organization_id = \$1/);
+    assert.match(sql, /t\.status = 'awaiting_attachment'/);
+    assert.match(sql, /t\.business_route_id IN \('passenger_list_import_independent', 'passenger_list_import_shared_child'\)/);
+    assert.match(sql, /t\.archived_at IS NULL/);
+    assert.match(sql, /s\.conversation_id = \$2/);
+    assert.match(sql, /\(\$3::uuid IS NULL OR t\.channel_id = \$3::uuid\)/);
+    assert.match(sql, /\(\$4::boolean = false OR t\.assigned_user_id = \$5\)/);
+    assert.match(sql, /ORDER BY t\.created_at DESC, t\.id DESC\s+LIMIT 1\s*$/);
+    assert.doesNotMatch(sql, /updated_at/);
+    assert.deepEqual(params, ['org-a', 'conversation-a', 'channel-a', true, 'user-a']);
+    return result([task('newest')]);
+  });
+  for (const role of ['user', 'team_lead'] as const) {
+    assert.equal((await target(selection, { ...context, role })).taskId, 'TASK-newest');
+  }
+});
+
+test('explicit target still selects an older task and validates channel and conversation', async t => {
+  const { pool, target } = setup(t);
+  t.mock.method(pool, 'query', async (sql: string, params: unknown[]) => {
+    assert.match(sql, /t\.organization_id = \$1 AND t\.task_id = \$2/);
+    assert.match(sql, /t\.archived_at IS NULL/);
+    assert.match(sql, /t\.assigned_user_id = \$4/);
+    assert.doesNotMatch(sql, /ORDER BY|LIMIT/);
+    assert.deepEqual(params, ['org-a', 'TASK-older', true, 'user-a']);
+    return result([task('older')]);
+  });
+  const explicit = { ...selection, taskId: 'TASK-older' };
+  assert.equal((await target(explicit)).taskId, 'TASK-older');
+  await assert.rejects(target({ ...explicit, channelId: 'channel-b' }), { code: 'channel_mismatch' });
+  await assert.rejects(target({ ...explicit, conversationId: 'conversation-b' }), { code: 'conversation_mismatch' });
+});
+
+test('missing conversation or eligible target never falls back to an unrelated task', async t => {
+  const { pool, target } = setup(t);
+  let queries = 0;
+  t.mock.method(pool, 'query', async () => { queries++; return result(); });
+  await assert.rejects(target({ channelId: 'channel-a' }), { code: 'task_selection_required' });
+  assert.equal(queries, 0);
+  await assert.rejects(target(), { code: 'awaiting_attachment_task_not_found' });
+  await assert.rejects(target({ ...selection, taskId: 'TASK-missing' }), { code: 'task_not_found' });
+});
+
+test('attachment transaction writes only its selected task and keeps race and replay protections', async t => {
+  const { pool, service } = setup(t);
+  const attachment = {
+    fileName: 'synthetic.xlsx', contentType: 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
+    content: Buffer.from('synthetic-normalization-fixture'), source: 'agentbus' as const
+  };
+  const requestHash = sha256Text(`passenger-workbook\0${sha256Bytes(attachment.content)}`);
+  let chosen = task('newest');
+  let locked = { ...chosen };
+  let previousKey: Record<string, unknown> | undefined;
+  let attachedTo: unknown[] = [];
+  let updatedTasks: unknown[] = [];
+  let statements: string[] = [];
+  let returnedTaskId = '';
+  let targetReads = 0;
+  t.mock.method(pool, 'query', async (sql: string, params: unknown[]) => {
+    if (sql.includes('FROM tasks t')) { targetReads++; return result([{ ...chosen }]); }
+    assert.match(sql, /FROM task_input_attachments/);
+    assert.equal(params[0], chosen.id);
+    return result();
+  });
+  const client = { release() {}, async query(sql: string, params: unknown[] = []) {
+    statements.push(sql.trim());
+    if (sql.includes('SELECT t.*, s.conversation_id')) {
+      assert.match(sql, /FOR UPDATE OF t, s/);
+      assert.deepEqual(params, ['org-a', chosen.id, true, 'user-a']);
+      return result([{ ...locked }]);
+    }
+    if (sql.includes('FROM idempotency_keys i')) return result(previousKey ? [previousKey] : []);
+    if (sql.includes('FROM task_input_attachments')) return result();
+    if (sql.includes('INSERT INTO task_input_attachments')) { attachedTo.push(params[1]); return result(); }
+    if (/UPDATE tasks\s/.test(sql)) {
+      updatedTasks.push(params[0]); locked.status = 'parse_queued'; return result([{ ...locked }]);
+    }
+    if (/^(BEGIN|COMMIT|ROLLBACK)$/.test(sql) || sql.includes('pg_advisory_xact_lock')
+      || sql.includes('INSERT INTO idempotency_keys')) return result();
+    assert.fail(`Unexpected query: ${sql}`);
+  } };
+  t.mock.method(pool, 'connect', async () => client);
+  // Keep normalization and authorization outside this association regression;
+  // execute the real attachment transaction, state checks and idempotency logic.
+  Object.assign(service, {
+    requireBusinessAuthorization: async () => {}, assertBusinessAuthorizationInTransaction: async () => {},
+    normalizePassengerRosterAttachment: async () => ({ canonicalTsv: 'synthetic', rowCount: 1, sourceFormat: 'xlsx' }),
+    emitEvent: async (_client: unknown, _row: unknown, event: unknown) => event,
+    audit: async () => {}, notify: () => {}, recordAgentBusAcceptedDelivery: async () => {},
+    getTask: async (_org: unknown, id: string) => { returnedTaskId = id; return { ...locked }; }
+  });
+  const first = await service.attachPassengerRosterAttachment(context, attachment, selection);
+  assert.equal(first.input_attachment?.status, 'normalized');
+  assert.equal(returnedTaskId, 'TASK-newest');
+  assert.deepEqual(attachedTo, ['newest']);
+  assert.deepEqual(updatedTasks, ['newest']);
+  assert.ok(statements.includes('COMMIT'));
+
+  for (const scenario of ['state_changed', 'replay_after_newest_completed'] as const) {
+    chosen = task(scenario === 'state_changed' ? 'newest' : 'older');
+    locked = { ...chosen, status: scenario === 'state_changed' ? 'parse_queued' : 'awaiting_attachment' };
+    previousKey = scenario === 'replay_after_newest_completed'
+      ? { id: 'newest', task_id: 'TASK-newest', request_hash: requestHash } : undefined;
+    attachedTo = []; updatedTasks = []; statements = []; targetReads = 0;
+    await assert.rejects(service.attachPassengerRosterAttachment(context, attachment,
+      { ...selection, idempotencyKey: 'same-agentbus-message' }), (error: unknown) => {
+      assert.ok(error instanceof TaskError);
+      assert.equal(error.code, scenario === 'state_changed' ? 'invalid_transition' : 'idempotency_conflict');
+      return true;
+    });
+    assert.equal(targetReads, 1, 'never reselect an older task after a transaction failure');
+    assert.deepEqual(attachedTo, [], scenario);
+    assert.deepEqual(updatedTasks, [], scenario);
+    assert.ok(statements.includes('ROLLBACK'), scenario);
+  }
+});