source/teaql-ts/tests/sql-mask-lifecycle.test.ts

import { UserContext } from '../src/core/context';
import { SelectQuery } from '../src/core/ast';
import { PLAINTEXT_LOG_ENV, PLAINTEXT_LOG_ACK, projectSQLLog } from '../src/core/log-privacy';
import { AbstractSQLTeaQLClient, EntitySchema, SQLExecutionMetadata, SQLExecutionEvidenceStore,
  TextDiagnosticSQLLogSink, SqlQueryResult } from '../src/sql/core';
import { SQLiteDriver } from '../src/sql/sqlite';
import { GraphMutationSession, MutationIntent } from '../src/core/request-intent';

const schema: Record<string, EntitySchema> = { Customer: {
  table: 'customer_data', auditMaskFields: ['display_name'], columns: {
    id: { columnName: 'id', logicalType: 'integer', decode: 'string', logPolicy: 'plain' },
    version: { columnName: 'version', logicalType: 'integer', decode: 'number', logPolicy: 'plain' },
    displayName: { columnName: 'display_name', logicalType: 'text', decode: 'native', logPolicy: 'plain' },
    publicAddress: { columnName: 'public_address', logicalType: 'text', decode: 'native', logPolicy: 'plain' },
    passwordHash: { columnName: 'password_hash', logicalType: 'text', decode: 'native', logPolicy: 'plain' },
  },
} };

class Driver extends SQLiteDriver {
  fault?: Error;
  readbackFault?: Error;
  readbackRows?: number;
  failStreamAfter?: number;
  closedCursor = false;
  values: unknown[] = [];
  async query(sql: string, values: any[] = []): Promise<SqlQueryResult> {
    if (sql.startsWith('SELECT') && sql.includes('customer_data') && sql.includes('WHERE "id" = ?')) {
      if (this.readbackFault) throw this.readbackFault;
      if (this.readbackRows !== undefined) {
        const result = await super.query(sql, values);
        return { rowCount: this.readbackRows, rows: Array.from({ length: this.readbackRows }, () => result.rows[0]) };
      }
    }
    if (this.fault && sql.includes('customer_data')) {
      this.values = [...values];
      throw this.fault;
    }
    return super.query(sql, values);
  }
  async *stream(sql: string, values: any[] = []): AsyncIterable<any> {
    this.values = [...values];
    let seen = 0;
    try {
      for await (const row of super.stream(sql, values)) {
        if (this.failStreamAfter === seen++) throw this.fault;
        yield row;
      }
    } finally { this.closedCursor = true; }
  }
}
class Client extends AbstractSQLTeaQLClient {
  constructor(driver: Driver) { super(driver, schema); }
}
const previous = process.env[PLAINTEXT_LOG_ENV];
beforeEach(() => { delete process.env[PLAINTEXT_LOG_ENV]; });
afterEach(() => {
  if (previous === undefined) delete process.env[PLAINTEXT_LOG_ENV];
  else process.env[PLAINTEXT_LOG_ENV] = previous;
});

async function fixture() {
  const driver = new Driver(':memory:');
  const client = new Client(driver).setDiagnosticSQLLogSink(undefined);
  await new UserContext().insertResource('dataService', client).ensureSchema();
  for (let id = 1; id <= 3; id++) await client.executeMutation({
    entity: 'Customer', action: 'Create', id: String(id),
    payload: { displayName: 'Riverside', publicAddress: '1 Runtime Road', passwordHash: 'PASSWORD-CANARY' },
    comment: 'seed lifecycle fixture',
  });
  const logs: SQLExecutionMetadata[] = [], output: string[] = [];
  const evidence = new SQLExecutionEvidenceStore();
  client.setRuntimeTelemetrySink(evidence).setDiagnosticSQLLogSink({ write(entry) {
    logs.push(entry); new TextDiagnosticSQLLogSink(line => output.push(line)).write(entry);
  } });
  return { driver, client, logs, output, evidence };
}
function query() {
  return new SelectQuery('Customer').filter({ displayName: { $eq: 'Riverside' },
    publicAddress: { $eq: '1 Runtime Road' }, passwordHash: { $eq: 'PASSWORD-CANARY' } })
    .limit(10).comment('what: stream matching customers').purpose('why: verify lifecycle masking');
}
function verify(f: Awaited<ReturnType<typeof fixture>>, outcome: string, count?: number) {
  expect(f.logs).toHaveLength(1);
  const entry = f.logs[0];
  expect(entry.executionOutcome).toBe(outcome);
  expect(entry.resultCount).toBe(count);
  expect(entry.tracePath.length).toBeGreaterThan(0);
  const all = JSON.stringify([f.logs, f.output, f.evidence.snapshot()]);
  expect(all).not.toContain('Riverside');
  expect(all).not.toContain('PASSWORD-CANARY');
  expect(all).not.toContain('DRIVER-CANARY');
  expect(entry.debugSQL).toContain('Ri*****de');
  expect(entry.debugSQL).toContain('1 Runtime Road');
  expect(f.output.join('\n')).toContain(`outcome=${outcome}`);
}

it.each(['complete', 'cancel', 'consumer-error', 'provider-error'])(
  'logs masked SQL once when the stream ends: %s', async mode => {
    const f = await fixture();
    const fault = new Error('Riverside DRIVER-CANARY PASSWORD-CANARY');
    if (mode === 'provider-error') { f.driver.fault = fault; f.driver.failStreamAfter = 1; }
    try {
      const consume = async () => {
        for await (const rows of f.client.executeForStream(query(), 1)) {
          expect(rows[0].displayName).toBe('Riverside');
          if (mode === 'cancel') break;
          if (mode === 'consumer-error') throw fault;
        }
      };
      if (mode.endsWith('error')) await expect(consume()).rejects.toBe(fault);
      else await consume();
      verify(f, mode === 'complete' ? 'success' : mode === 'provider-error' ? 'failure' : 'cancelled', mode === 'complete' ? 3 : 1);
      expect(f.driver.closedCursor).toBe(true);
      expect(f.driver.values).toContain('Riverside');
      expect(f.driver.values).toContain('PASSWORD-CANARY');
    } finally { await f.client.close(); }
  });

it('records an empty stream and honors the query logging switch', async () => {
  const f = await fixture();
  try {
    const empty = query().filter({ id: { $eq: '999' } });
    for await (const rows of f.client.executeForStream(empty, 2)) expect(rows).toHaveLength(0);
    expect(f.logs).toHaveLength(1);
    expect(f.logs[0].resultCount).toBe(0);
    f.logs.length = 0; f.evidence.enableAll(); f.client.setQueryLoggingEnabled(false);
    for await (const _rows of f.client.executeForStream(query(), 2)) { /* consume */ }
    expect(f.logs).toHaveLength(0);
    expect(f.evidence.snapshot()).toHaveLength(1);
    expect(f.evidence.snapshot()[0].resultCount).toBe(3);
  } finally { await f.client.close(); }
});

it('records failed list SQL without logging or changing the original driver error', async () => {
  const f = await fixture();
  const fault = new Error('Riverside DRIVER-CANARY PASSWORD-CANARY');
  f.driver.fault = fault;
  try {
    await expect(f.client.executeQuery(query())).rejects.toBe(fault);
    verify(f, 'failure');
    expect(f.logs[0].affectedRows).toBeUndefined();
  } finally { await f.client.close(); }
});

it.each(['Create', 'Update', 'Delete'])('records failed %s with audit reason, not exception payload', async action => {
  const f = await fixture();
  const fault = new Error('Riverside DRIVER-CANARY PASSWORD-CANARY');
  f.driver.fault = fault;
  try {
    const targetID = action === 'Create' ? '4' : '1';
    await expect(f.client.executeMutation({ entity: 'Customer', action, id: targetID,
      version: 1, payload: { displayName: 'Riverside', publicAddress: '1 Runtime Road', passwordHash: 'PASSWORD-CANARY' },
      comment: `verify failed mutation ${targetID}` })).rejects.toBe(fault);
    expect(f.logs).toHaveLength(1);
    expect(f.logs[0].executionOutcome).toBe('failure');
    expect(f.logs[0].auditReason).toBe('verify failed mutation [REDACTED]');
    expect(f.logs[0].affectedRows).toBeUndefined();
    const all = JSON.stringify([f.logs, f.output, f.evidence.snapshot()]);
    expect(all).not.toMatch(/Riverside|PASSWORD-CANARY|DRIVER-CANARY/);
    expect(f.logs[0].debugSQL).not.toContain('?');
  } finally { await f.client.close(); }
});

it('does not redact a structural row count when the target ID is short', async () => {
  const f = await fixture();
  try {
    await f.client.executeMutation({entity:'Customer',action:'Update',id:'1',version:1,
      payload:{publicAddress:'Changed Road'},comment:'update target 1'});
    expect(f.logs.map(entry => entry.operation)).toEqual(['update', 'select']);
    expect(f.logs[0].auditReason).toBe('update target [REDACTED]');
    expect(f.logs[0].resultSummary).toBe('1 rows affected');
    expect(f.logs[0].parameters).toContain('1');
  } finally { await f.client.close(); }
});

it.each(['Create', 'Update', 'Delete'])(
  'scrubs the target id from successful %s audit reason but keeps the typed id', async action => {
    const f = await fixture();
    try {
      if (action !== 'Create') {
        await f.client.executeMutation({
          entity: 'Customer', action: 'Create', id: '1001',
          payload: { displayName: 'Seed', publicAddress: 'Seed Road' },
          comment: 'seed audit target',
        });
      }
      const mutation = {
        entity: 'Customer', action, id: '1001', version: 1,
        payload: action === 'Delete' ? {} : { displayName: 'Changed' },
        comment: `${action.toLowerCase()} target 1001`,
      };
      await f.client.executeMutation(mutation);
      const auditTrace = f.client.auditTrace;
      const audit = auditTrace[auditTrace.length - 1];
      expect(audit?.id).toBe('1001');
      expect(audit?.reason).toBe(`${action.toLowerCase()} target [REDACTED]`);
      const sqlEntry = f.logs[f.logs.length - 1];
      expect(sqlEntry.auditReason).toBe(`${action.toLowerCase()} target [REDACTED]`);
      expect(JSON.stringify(sqlEntry.tracePath)).not.toContain('1001');
      expect(sqlEntry).not.toHaveProperty('targetID');
      expect(sqlEntry.parameters).toContain('1001');
      expect(sqlEntry.mutationLineage?.[0].entityId).toBe('1001');
      expect(sqlEntry.mutationLineage?.[0].detail).toBe(`${action.toLowerCase()} target [REDACTED]`);
      expect(Object.isFrozen(sqlEntry.mutationLineage?.[0])).toBe(true);
      expect(mutation.id).toBe('1001');
      expect(mutation.comment).toContain('1001');
    } finally { await f.client.close(); }
  });

it('does not report prefetched but undelivered rows on a stream failure', async () => {
  const f = await fixture();
  const fault = new Error('DRIVER-CANARY');
  f.driver.fault = fault; f.driver.failStreamAfter = 1;
  try {
    const consume = async () => { for await (const _rows of f.client.executeForStream(query(), 3)) { /* consume */ } };
    await expect(consume()).rejects.toBe(fault);
    verify(f, 'failure', 0);
    expect(f.driver.closedCursor).toBe(true);
  } finally { await f.client.close(); }
});

it('keeps credentials private and labels each debug stream/failure record', async () => {
  const f = await fixture();
  process.env[PLAINTEXT_LOG_ENV] = PLAINTEXT_LOG_ACK;
  try {
    for await (const _rows of f.client.executeForStream(query(), 1)) break;
    f.driver.fault = new Error('DRIVER-CANARY');
    await expect(f.client.executeQuery(query())).rejects.toBe(f.driver.fault);
    expect(f.logs).toHaveLength(2);
    for (const entry of f.logs) {
      expect(entry.debugSQL).toContain('Riverside');
      expect(entry.debugSQL).toContain('EXPLICIT OPT-IN');
      expect(entry.debugSQL).toContain('NOT REPLAYABLE');
      expect(JSON.stringify(entry)).not.toMatch(/PASSWORD-CANARY|DRIVER-CANARY/);
    }
  } finally { await f.client.close(); }
});

it('does not replace a driver error with a diagnostic-sink error', async () => {
  const f = await fixture();
  const fault = new Error('DRIVER-CANARY');
  f.driver.fault = fault;
  f.client.setDiagnosticSQLLogSink({ write() { throw new Error('SINK-FAILURE'); } });
  try {
    await expect(f.client.executeQuery(query())).rejects.toBe(fault);
    f.driver.failStreamAfter = 0;
    const consume = async () => { for await (const _rows of f.client.executeForStream(query(), 1)) { /* consume */ } };
    await expect(consume()).rejects.toBe(fault);
    expect(f.driver.closedCursor).toBe(true);
  } finally { await f.client.close(); }
});

it('keeps successful SQL and graph mutation independent of broken diagnostic sinks', async () => {
  const f = await fixture();
  let evidenceCalls = 0;
  let diagnosticCalls = 0;
  f.client.setRuntimeTelemetrySink({ record(entry) {
    evidenceCalls++;
    expect(JSON.stringify(entry)).not.toContain('PASSWORD-CANARY');
    throw new Error('EVIDENCE-SINK-FAILURE');
  } }).setDiagnosticSQLLogSink({ write(entry) {
    diagnosticCalls++;
    expect(JSON.stringify(entry)).not.toContain('PASSWORD-CANARY');
    throw new Error('DIAGNOSTIC-SINK-FAILURE');
  } });
  try {
    await expect(f.client.executeMutation({
      entity: 'Customer', action: 'Create', id: '4',
      payload: { displayName: 'Riverside', publicAddress: '1 Runtime Road',
        passwordHash: 'PASSWORD-CANARY' },
      comment: 'what: create customer despite logging failure',
    })).resolves.toMatchObject({ success: true, id: '4' });
    const rows = await f.client.executeQuery(query());
    expect(rows).toHaveLength(4);
    const driverFault = new Error('DRIVER-CANARY');
    f.driver.fault = driverFault;
    await expect(f.client.executeQuery(query())).rejects.toBe(driverFault);
    expect(evidenceCalls).toBeGreaterThan(0);
    expect(diagnosticCalls).toBeGreaterThan(0);
  } finally { await f.client.close(); }
});

it.each(['Create', 'Update', 'Delete'].flatMap(action => ['error','empty','multiple'].map(mode => ({action,mode}))))(
  'retains independent write/readback facts for $action/$mode', async ({action,mode}) => {
    const f = await fixture();
    const fault = new Error('DRIVER-CANARY Riverside PASSWORD-CANARY');
    if (mode === 'error') f.driver.readbackFault = fault;
    else f.driver.readbackRows = mode === 'empty' ? 0 : 2;
    try {
      const mutation = { entity:'Customer', action, id:action === 'Create' ? '4':'1', version:1,
        payload:{displayName:'Riverside',passwordHash:'PASSWORD-CANARY'},
        comment: action === 'Delete' ? 'what: delete customer' : 'what: write Riverside PASSWORD-CANARY' };
      const pending = f.client.executeMutation(mutation);
      if (mode === 'error') await expect(pending).rejects.toBe(fault);
      else await expect(pending).rejects.toThrow('could not be read back');
      expect(f.logs).toHaveLength(2);
      expect(f.logs[0].executionOutcome).toBe('success');
      expect(f.logs[0].affectedRows).toBe(1);
      expect(f.logs[1].operation).toBe('select');
      expect(f.logs[1].executionOutcome).toBe(mode === 'error' ? 'failure':'success');
      expect(f.logs[1].resultCount).toBe(mode === 'error' ? undefined : mode === 'empty' ? 0:2);
      expect(f.logs[1].affectedRows).toBeUndefined();
      expect(f.logs[1].auditReason).toContain('what:');
      expect(f.logs[1].debugSQL).toContain('SELECT');
      expect(f.logs[1].debugSQL).not.toContain('?');
      expect(JSON.stringify([f.logs,f.output,f.evidence.snapshot()])).not.toMatch(/Riverside|PASSWORD-CANARY|DRIVER-CANARY/);
      f.driver.readbackFault = undefined; f.driver.readbackRows = undefined;
      const rows = await f.driver.query('SELECT * FROM customer_data');
      expect(rows.rowCount).toBe(3);
      expect(rows.rows[0].version).toBe(1);
      expect(rows.rows[0].display_name).toBe('Riverside');
    } finally { await f.client.close(); }
  });

it('preserves original error when the readback diagnostic sink fails', async () => {
  const f = await fixture();
  const fault = new Error('READBACK-FAILURE');
  f.driver.readbackFault = fault;
  let observed = 0;
  f.client.setDiagnosticSQLLogSink({write(entry) { observed++; if(entry.operation === 'select') throw new Error('SINK-FAILURE'); }});
  try {
    await expect(f.client.executeMutation({entity:'Customer',action:'Update',id:'1',version:1,payload:{displayName:'Riverside'},comment:'verify failed readback'})).rejects.toBe(fault);
    expect(observed).toBe(2);
    expect((await f.driver.query('SELECT version FROM customer_data')).rows[0].version).toBe(1);
  } finally { await f.client.close(); }
});

it.each(['debug','disabled','success'])('handles readback mode %s without changing business values', async mode => {
  const f = await fixture();
  if (mode === 'debug') process.env[PLAINTEXT_LOG_ENV] = PLAINTEXT_LOG_ACK;
  if (mode === 'disabled') f.client.setQueryLoggingEnabled(false).setMutationLoggingEnabled(false);
  if (mode !== 'success') f.driver.readbackFault = new Error('READBACK-FAILURE');
  try {
    const pending = f.client.executeMutation({entity:'Customer',action:'Update',id:'1',version:1,
      payload:{displayName:'Riverside',passwordHash:'PASSWORD-CANARY'},comment:'what: update Riverside PASSWORD-CANARY'});
    if (mode === 'success') {
      const result = await pending;
      expect(result.persistedRecord?.displayName).toBe('Riverside');
      expect(result.persistedRecord?.passwordHash).toBe('PASSWORD-CANARY');
      expect(f.logs.map(entry => entry.operation)).toEqual(['update', 'select']);
    } else await expect(pending).rejects.toBe(f.driver.readbackFault);
    if (mode === 'disabled') {
      expect(f.logs).toHaveLength(0);
      expect(f.evidence.snapshot().map(entry => entry.executionOutcome)).toEqual(['success', 'failure']);
    }
    if (mode === 'debug') {
      expect(f.logs).toHaveLength(2);
      for (const entry of f.logs) {
        expect(entry.auditReason).toContain('Riverside');
        expect(entry.debugSQL).toContain('EXPLICIT OPT-IN');
        expect(JSON.stringify(entry)).not.toContain('PASSWORD-CANARY');
      }
      delete process.env[PLAINTEXT_LOG_ENV];
      for (const entry of f.logs) {
        const safe = projectSQLLog(entry);
        expect(safe.auditReason).toContain('what: update');
        expect(JSON.stringify(safe)).not.toMatch(/Riverside|PASSWORD-CANARY/);
        // Copying a record loses the private safe alternative. Unknown inherited
        // intent must then fail closed rather than survive in plaintext.
        expect(JSON.stringify(projectSQLLog({...entry}))).not.toMatch(/Riverside|PASSWORD-CANARY/);
      }
    }
  } finally { await f.client.close(); }
});

it.each(['write','readback'])('retains partial graph SQL facts and rolls back after %s failure', async phase => {
  const f = await fixture();
  const fault = new Error('GRAPH-FAILURE');
  const create = (id:string, graph?: GraphMutationSession) => {
    const mutation = {entity:'Customer',action:'Create',id,
      payload:{displayName:'Riverside',passwordHash:'PASSWORD-CANARY'},comment:'what: insert Riverside PASSWORD-CANARY'};
    return f.client.executeMutation(graph ? graph.request(mutation) : mutation);
  };
  try {
    await expect(f.client.executeGraphSave(new MutationIntent('what: insert Riverside PASSWORD-CANARY'), async graph => {
      await create('4', graph);
      if(phase === 'write') f.driver.fault = fault;
      else f.driver.readbackFault = fault;
      await create('5', graph);
      await create('6', graph);
    })).rejects.toBe(fault);
    expect(f.logs.map(entry => entry.executionOutcome)).toEqual(phase === 'write'
      ? ['success','success','failure'] : ['success','success','success','failure']);
    expect(f.logs[1].operation).toBe('select');
    expect(f.logs[1].resultCount).toBe(1);
    expect(JSON.stringify([f.logs,f.output,f.evidence.snapshot()])).not.toMatch(/Riverside|PASSWORD-CANARY|GRAPH-FAILURE/);
    f.driver.fault = undefined; f.driver.readbackFault = undefined;
    expect((await f.driver.query('SELECT * FROM customer_data')).rowCount).toBe(3);
    f.logs.length = 0;
    await create('7');
    expect(f.logs.map(entry => entry.operation)).toEqual(['insert', 'select']);
    expect((await f.driver.query('SELECT * FROM customer_data')).rowCount).toBe(4);
  } finally { await f.client.close(); }
});

it.each(['AbortError','TimeoutError'])('records readback cancellation reported by the driver: %s', async name => {
  const f = await fixture();
  const fault = new Error('CANCEL-CANARY'); fault.name = name;
  f.driver.readbackFault = fault;
  try {
    await expect(f.client.executeMutation({entity:'Customer',action:'Update',id:'1',version:1,
      payload:{displayName:'Riverside'},comment:'what: update Riverside'})).rejects.toBe(fault);
    expect(f.logs.map(entry=>entry.executionOutcome)).toEqual(['success','cancelled']);
    expect(f.logs[1].resultCount).toBeUndefined();
    expect(JSON.stringify(f.logs)).not.toMatch(/Riverside|CANCEL-CANARY/);
    expect((await f.driver.query('SELECT version FROM customer_data')).rows[0].version).toBe(1);
  } finally { await f.client.close(); }
});