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

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);
  }
});