Genarrative/scripts/spacetime-editor-idempotency-smoke.mjs

#!/usr/bin/env node

import { spawn } from 'node:child_process';
import { once } from 'node:events';
import { chmod, mkdtemp, rm } from 'node:fs/promises';
import net from 'node:net';
import os from 'node:os';
import path from 'node:path';
import { setTimeout as delay } from 'node:timers/promises';
import { fileURLToPath } from 'node:url';

import { createSpacetimeWebIdentity } from './spacetime-migration-common.mjs';

const repoRoot = path.resolve(
  path.dirname(fileURLToPath(import.meta.url)),
  '..',
);
const database = 'editor-idempotency-smoke';
const expectedSpacetimeVersion = '2.8.3';
const expectedSpacetimeCommit = '8e410d2842147bd8e5a32a9589cc00c19f7478e2';
const commandTimeoutMs = 5 * 60 * 1000;

function assert(condition, message) {
  if (!condition) {
    throw new Error(message);
  }
}

function appendOutput(current, chunk) {
  const next = `${current}${chunk}`;
  return next.length <= 24_000 ? next : next.slice(-24_000);
}

export function sanitizeDiagnostic(value, sensitiveValues = []) {
  let sanitized = String(value);
  for (const sensitiveValue of sensitiveValues) {
    if (typeof sensitiveValue === 'string' && sensitiveValue.length > 0) {
      sanitized = sanitized.split(sensitiveValue).join('[redacted]');
    }
  }
  return sanitized;
}

export function runCommand(command, args, options = {}) {
  return new Promise((resolve, reject) => {
    const child = spawn(command, args, {
      cwd: options.cwd ?? repoRoot,
      env: options.env ?? process.env,
      shell: false,
      stdio: ['ignore', 'pipe', 'pipe'],
    });
    let stdout = '';
    let stderr = '';
    const timeout = setTimeout(() => {
      child.kill('SIGKILL');
      reject(
        new Error(
          sanitizeDiagnostic(
            `${command} timed out after ${options.timeoutMs ?? commandTimeoutMs}ms`,
            options.sensitiveValues,
          ),
        ),
      );
    }, options.timeoutMs ?? commandTimeoutMs);

    child.stdout.on('data', (chunk) => {
      stdout = appendOutput(stdout, chunk.toString());
    });
    child.stderr.on('data', (chunk) => {
      stderr = appendOutput(stderr, chunk.toString());
    });
    child.on('error', (error) => {
      clearTimeout(timeout);
      reject(error);
    });
    child.on('exit', (code, signal) => {
      clearTimeout(timeout);
      const diagnostic = sanitizeDiagnostic(
        `${stdout}\n${stderr}`.trim(),
        options.sensitiveValues,
      );
      if (signal) {
        reject(new Error(`${command} exited via ${signal}: ${diagnostic}`));
      } else if (code !== 0) {
        reject(new Error(`${command} exited with ${code}: ${diagnostic}`));
      } else {
        resolve({ stdout, stderr });
      }
    });
  });
}

export function parseProcedureOutput(stdout, procedureName = 'procedure') {
  let value;
  try {
    value = JSON.parse(String(stdout).trim());
  } catch (error) {
    throw new Error(
      `${procedureName} stdout is not JSON: ${error instanceof Error ? error.message : String(error)}`,
    );
  }
  assert(
    Array.isArray(value),
    `${procedureName} returned an invalid result shape.`,
  );
  if (value.length === 4) {
    const folder = decodeOption(value[1]);
    const library = decodeOption(value[2]);
    return {
      ok: value[0],
      record: folder ?? library,
      errorMessage: decodeOption(value[3]),
    };
  }
  assert(
    value.length === 3,
    `${procedureName} returned an invalid result shape.`,
  );
  return {
    ok: value[0],
    record: decodeOption(value[1]),
    errorMessage: decodeOption(value[2]),
  };
}

export function parseSqlRows(stdout, queryLabel = 'SQL') {
  let payload;
  try {
    payload = JSON.parse(String(stdout).trim());
  } catch (error) {
    throw new Error(
      `${queryLabel} stdout is not JSON: ${error instanceof Error ? error.message : String(error)}`,
    );
  }
  assert(
    Array.isArray(payload) &&
      payload.length === 1 &&
      Array.isArray(payload[0]?.rows),
    `${queryLabel} returned an invalid result shape.`,
  );
  return payload[0].rows;
}

function decodeOption(value) {
  if (Array.isArray(value) && value.length === 2 && value[0] === 0) {
    return value[1];
  }
  if (Array.isArray(value) && value.length === 2 && value[0] === 1) {
    return null;
  }
  if (value === null || value === undefined) {
    return null;
  }
  throw new Error(`Unexpected SATS option: ${JSON.stringify(value)}`);
}

async function reservePort() {
  const server = net.createServer();
  await new Promise((resolve, reject) => {
    server.once('error', reject);
    server.listen(0, '127.0.0.1', resolve);
  });
  const address = server.address();
  assert(
    address && typeof address === 'object',
    'Failed to reserve a local port.',
  );
  const port = address.port;
  await new Promise((resolve, reject) =>
    server.close((error) => (error ? reject(error) : resolve())),
  );
  return port;
}

function startStandalone(dataDir, port) {
  const child = spawn(
    'spacetime',
    [
      'start',
      '--in-memory',
      '--data-dir',
      dataDir,
      '--listen-addr',
      `127.0.0.1:${port}`,
      '--non-interactive',
    ],
    {
      cwd: repoRoot,
      env: process.env,
      shell: false,
      stdio: ['ignore', 'pipe', 'pipe'],
    },
  );
  let output = '';
  child.stdout.on('data', (chunk) => {
    output = appendOutput(output, chunk.toString());
  });
  child.stderr.on('data', (chunk) => {
    output = appendOutput(output, chunk.toString());
  });
  return { child, output: () => output };
}

async function waitForStandalone(serverUrl, processState) {
  const deadline = Date.now() + 30_000;
  while (Date.now() < deadline) {
    if (processState.child.exitCode !== null) {
      throw new Error(
        `SpacetimeDB exited during startup: ${processState.output().trim()}`,
      );
    }
    try {
      const response = await fetch(`${serverUrl}/v1/ping`);
      if (response.ok) {
        return;
      }
    } catch {
      // Startup is still in progress.
    }
    await delay(200);
  }
  throw new Error(
    `Timed out waiting for SpacetimeDB: ${processState.output().trim()}`,
  );
}

export async function stopStandalone(child) {
  if (child.exitCode !== null) {
    return;
  }
  if (process.platform === 'win32') {
    await stopWindowsProcessTree(child);
    return;
  }
  child.kill('SIGTERM');
  const exited = await Promise.race([
    once(child, 'exit').then(() => true),
    delay(5_000).then(() => false),
  ]);
  if (!exited && child.exitCode === null) {
    child.kill('SIGKILL');
    await once(child, 'exit');
  }
}

async function stopWindowsProcessTree(child) {
  if (typeof child.pid === 'number') {
    await runTaskKill(child.pid);
  }
  const exited = await Promise.race([
    once(child, 'exit').then(() => true),
    delay(5_000).then(() => false),
  ]);
  if (!exited && child.exitCode === null) {
    child.kill('SIGKILL');
    await once(child, 'exit');
  }
}

function runTaskKill(pid) {
  return new Promise((resolve, reject) => {
    const taskKill = spawn('taskkill', ['/PID', String(pid), '/T', '/F'], {
      stdio: 'ignore',
      shell: false,
    });
    taskKill.once('error', reject);
    taskKill.once('exit', (code, signal) => {
      if (code === 0 || code === 128 || code === 1) {
        resolve();
        return;
      }
      reject(new Error(`taskkill exited with ${signal ?? code}`));
    });
  });
}

function cliPrefix(configPath) {
  return ['--config-path', configPath];
}

async function callProcedure(configPath, serverUrl, procedureName, input) {
  const result = await runCommand('spacetime', [
    ...cliPrefix(configPath),
    'call',
    '--server',
    serverUrl,
    '--yes',
    '--no-config',
    database,
    procedureName,
    JSON.stringify(input),
  ]);
  return parseProcedureOutput(result.stdout, procedureName);
}

async function sqlRows(configPath, serverUrl, query, queryLabel) {
  const result = await runCommand('spacetime', [
    ...cliPrefix(configPath),
    'sql',
    '--server',
    serverUrl,
    '--yes',
    '--no-config',
    '--format',
    'json',
    database,
    query,
  ]);
  return parseSqlRows(result.stdout, queryLabel);
}

function projectInput(projectId, title, nowMicros = 1_000) {
  return {
    project_id: projectId,
    owner_user_id: 'receipt-owner',
    title,
    now_micros: nowMicros,
  };
}

function resourceInput(overrides = {}) {
  const none = [1, []];
  const some = (value) => [0, value];
  return {
    resource_id: 'resource-a',
    project_id: 'project-a',
    owner_user_id: 'receipt-owner',
    asset_object_id: none,
    image_src: '/generated/editor/resource-a.png',
    object_key: none,
    width: 32,
    height: 32,
    source_type: 'generated',
    prompt: some('receipt smoke'),
    actual_prompt: none,
    model: some('smoke'),
    provider: some('smoke'),
    task_id: some('receipt-smoke-task'),
    source_resource_id: none,
    asset_kind: some('icon'),
    generation_inputs_json: none,
    updated_at_micros: 1_000,
    image_sequence_frames_json: none,
    image_sequence_duration_ms: none,
    ...overrides,
  };
}

function folderInput(overrides = {}) {
  return {
    folder_id: 'folder-a',
    owner_user_id: 'receipt-owner',
    label: 'Folder A',
    sort_order: 100,
    now_micros: 1_000,
    ...overrides,
  };
}

function assertOk(result, label) {
  assert(result.ok === true, `${label} failed: ${result.errorMessage}`);
  assert(result.record !== null, `${label} did not return a record.`);
}

function assertBodyConflict(result, label) {
  assert(result.ok === false, `${label} unexpectedly succeeded.`);
  assert(
    result.errorMessage ===
      '幂等创建冲突：同一 Idempotency-Key 的创建正文不一致',
    `${label} returned an unexpected error: ${result.errorMessage}`,
  );
}

function assertDeletedConflict(result, recordLabel) {
  assert(
    result.ok === false,
    `${recordLabel} replay unexpectedly recreated the record.`,
  );
  assert(
    typeof result.errorMessage === 'string' &&
      result.errorMessage.includes('已删除'),
    `${recordLabel} replay returned an unexpected error: ${result.errorMessage}`,
  );
}

async function runSmoke(configPath, serverUrl) {
  const projectA = projectInput('project-a', 'First');
  assertOk(
    await callProcedure(
      configPath,
      serverUrl,
      'create_editor_project_and_return',
      projectA,
    ),
    'project first create',
  );
  assertOk(
    await callProcedure(
      configPath,
      serverUrl,
      'create_editor_project_and_return',
      { ...projectA, now_micros: 2_000 },
    ),
    'project exact replay',
  );
  assertBodyConflict(
    await callProcedure(
      configPath,
      serverUrl,
      'create_editor_project_and_return',
      { ...projectA, title: 'Drifted' },
    ),
    'project drift replay',
  );

  const resourceA = resourceInput();
  assertOk(
    await callProcedure(
      configPath,
      serverUrl,
      'create_editor_project_resource_and_return',
      resourceA,
    ),
    'resource first create',
  );
  assertOk(
    await callProcedure(
      configPath,
      serverUrl,
      'create_editor_project_resource_and_return',
      { ...resourceA, updated_at_micros: 2_000 },
    ),
    'resource exact replay',
  );
  assertBodyConflict(
    await callProcedure(
      configPath,
      serverUrl,
      'create_editor_project_resource_and_return',
      { ...resourceA, width: 64 },
    ),
    'resource drift replay',
  );

  const folderA = folderInput();
  assertOk(
    await callProcedure(
      configPath,
      serverUrl,
      'create_editor_asset_folder_and_return',
      folderA,
    ),
    'asset folder first create',
  );
  assertOk(
    await callProcedure(
      configPath,
      serverUrl,
      'create_editor_asset_folder_and_return',
      { ...folderA, now_micros: 2_000 },
    ),
    'asset folder exact replay',
  );
  assertBodyConflict(
    await callProcedure(
      configPath,
      serverUrl,
      'create_editor_asset_folder_and_return',
      { ...folderA, label: 'Drifted Folder' },
    ),
    'asset folder drift replay',
  );

  const sameRaceInput = projectInput('project-same-race', 'Same Race');
  const sameRace = await Promise.all([
    callProcedure(
      configPath,
      serverUrl,
      'create_editor_project_and_return',
      sameRaceInput,
    ),
    callProcedure(
      configPath,
      serverUrl,
      'create_editor_project_and_return',
      sameRaceInput,
    ),
  ]);
  sameRace.forEach((result, index) =>
    assertOk(result, `same-body race ${index + 1}`),
  );

  const conflictRace = await Promise.all([
    callProcedure(
      configPath,
      serverUrl,
      'create_editor_project_and_return',
      projectInput('project-conflict-race', 'Race A'),
    ),
    callProcedure(
      configPath,
      serverUrl,
      'create_editor_project_and_return',
      projectInput('project-conflict-race', 'Race B'),
    ),
  ]);
  assert(
    conflictRace.filter((result) => result.ok).length === 1,
    'Different-body race did not commit exactly one request.',
  );
  assert(
    conflictRace.filter((result) => !result.ok).length === 1,
    'Different-body race did not reject exactly one request.',
  );
  assertBodyConflict(
    conflictRace.find((result) => !result.ok),
    'different-body race loser',
  );

  const projectsBeforeDelete = await sqlRows(
    configPath,
    serverUrl,
    'SELECT project_id, owner_user_id, title FROM editor_project',
    'project rows before delete',
  );
  const resourcesBeforeDelete = await sqlRows(
    configPath,
    serverUrl,
    'SELECT resource_id, project_id, owner_user_id FROM editor_project_resource',
    'resource rows before delete',
  );
  const receiptsBeforeDelete = await sqlRows(
    configPath,
    serverUrl,
    'SELECT owner_user_id, namespace, request_record_id, result_record_id FROM editor_idempotent_create_receipt',
    'receipt rows before delete',
  );
  const foldersBeforeDelete = await sqlRows(
    configPath,
    serverUrl,
    'SELECT folder_id, owner_user_id, system_default FROM editor_asset_folder',
    'asset folder rows before delete',
  );
  assert(
    projectsBeforeDelete.length === 3,
    'Unexpected project count before deletion.',
  );
  assert(
    resourcesBeforeDelete.length === 1,
    'Unexpected resource count before deletion.',
  );
  assert(
    receiptsBeforeDelete.length === 5,
    'Unexpected receipt count before deletion.',
  );
  const customFoldersBeforeDelete = foldersBeforeDelete.filter(
    ([, , systemDefault]) => !systemDefault,
  );
  assert(
    customFoldersBeforeDelete.length === 1,
    'Unexpected custom folder count before deletion.',
  );
  for (const [projectId, ownerUserId] of projectsBeforeDelete) {
    assert(
      ownerUserId === 'receipt-owner',
      `Project ${projectId} has the wrong owner.`,
    );
    assert(
      receiptsBeforeDelete.some(
        ([owner, namespace, requestId, resultId]) =>
          owner === ownerUserId &&
          namespace === 'editor-project' &&
          requestId === projectId &&
          resultId === projectId,
      ),
      `Project ${projectId} is missing its durable receipt.`,
    );
  }
  for (const [resourceId, projectId, ownerUserId] of resourcesBeforeDelete) {
    assert(
      projectsBeforeDelete.some(([candidate]) => candidate === projectId),
      `Resource ${resourceId} is orphaned from project ${projectId}.`,
    );
    assert(
      receiptsBeforeDelete.some(
        ([owner, namespace, requestId, resultId]) =>
          owner === ownerUserId &&
          namespace === 'editor-project-resource' &&
          requestId === resourceId &&
          resultId === resourceId,
      ),
      `Resource ${resourceId} is missing its durable receipt.`,
    );
  }
  for (const [folderId, ownerUserId] of customFoldersBeforeDelete) {
    assert(
      receiptsBeforeDelete.some(
        ([owner, namespace, requestId, resultId]) =>
          owner === ownerUserId &&
          namespace === 'editor-asset-folder' &&
          requestId === folderId &&
          resultId === folderId,
      ),
      `Asset folder ${folderId} is missing its durable receipt.`,
    );
  }

  const deleteResult = await callProcedure(
    configPath,
    serverUrl,
    'delete_editor_project_and_return',
    { project_id: 'project-a', owner_user_id: 'receipt-owner' },
  );
  assertOk(deleteResult, 'project delete');
  assertDeletedConflict(
    await callProcedure(
      configPath,
      serverUrl,
      'create_editor_project_and_return',
      projectA,
    ),
    'project',
  );
  assertDeletedConflict(
    await callProcedure(
      configPath,
      serverUrl,
      'create_editor_project_resource_and_return',
      resourceA,
    ),
    'resource',
  );
  const deleteFolderResult = await callProcedure(
    configPath,
    serverUrl,
    'delete_editor_asset_folder_and_return',
    {
      folder_id: 'folder-a',
      owner_user_id: 'receipt-owner',
      updated_at_micros: 3_000,
    },
  );
  assertOk(deleteFolderResult, 'asset folder delete');
  assertDeletedConflict(
    await callProcedure(
      configPath,
      serverUrl,
      'create_editor_asset_folder_and_return',
      folderA,
    ),
    'asset folder',
  );

  const projectsAfterDelete = await sqlRows(
    configPath,
    serverUrl,
    'SELECT project_id FROM editor_project',
    'project rows after delete',
  );
  const resourcesAfterDelete = await sqlRows(
    configPath,
    serverUrl,
    'SELECT resource_id FROM editor_project_resource',
    'resource rows after delete',
  );
  const receiptsAfterDelete = await sqlRows(
    configPath,
    serverUrl,
    'SELECT owner_user_id, namespace, request_record_id, result_record_id FROM editor_idempotent_create_receipt',
    'receipt rows after delete',
  );
  const foldersAfterDelete = await sqlRows(
    configPath,
    serverUrl,
    'SELECT folder_id, system_default FROM editor_asset_folder',
    'asset folder rows after delete',
  );
  assert(
    projectsAfterDelete.length === 2 &&
      projectsAfterDelete.every(([projectId]) => projectId !== 'project-a'),
    'Deleted project was recreated or another project was lost.',
  );
  assert(
    resourcesAfterDelete.length === 0,
    'Project deletion left an orphan resource.',
  );
  assert(
    foldersAfterDelete.length === 1 && foldersAfterDelete[0][1] === true,
    'Asset folder deletion removed the default folder or left a custom-folder orphan.',
  );
  assert(
    receiptsAfterDelete.length === receiptsBeforeDelete.length,
    'Deletion removed or duplicated a durable receipt.',
  );
}

export async function main() {
  let tempDir;
  let standalone;
  let loginToken = '';
  try {
    const version = await runCommand('spacetime', ['--version'], {
      timeoutMs: 30_000,
    });
    assert(
      version.stdout.includes(
        `spacetimedb tool version ${expectedSpacetimeVersion}`,
      ),
      `Expected SpacetimeDB ${expectedSpacetimeVersion}, got: ${version.stdout.trim()}`,
    );
    assert(
      version.stdout.includes(`Commit: ${expectedSpacetimeCommit}`),
      `Expected SpacetimeDB commit ${expectedSpacetimeCommit}, got: ${version.stdout.trim()}`,
    );

    tempDir = await mkdtemp(
      path.join(os.tmpdir(), 'genarrative-editor-idempotency-smoke-'),
    );
    const port = await reservePort();
    const serverUrl = `http://127.0.0.1:${port}`;
    standalone = startStandalone(path.join(tempDir, 'data'), port);
    await waitForStandalone(serverUrl, standalone);

    const identity = await createSpacetimeWebIdentity({ database, serverUrl });
    loginToken = identity.token;
    const configPath = path.join(tempDir, 'cli.toml');
    await runCommand(
      'spacetime',
      [...cliPrefix(configPath), 'login', '--token', loginToken],
      { sensitiveValues: [loginToken] },
    );
    await chmod(configPath, 0o600);

    const wasmPath =
      process.env.GENARRATIVE_EDITOR_IDEMPOTENCY_SMOKE_WASM?.trim();
    const publishArgs = [
      ...cliPrefix(configPath),
      'publish',
      database,
      '--server',
      serverUrl,
      '--yes=all',
      '--no-config',
    ];
    if (wasmPath) {
      publishArgs.push('--bin-path', path.resolve(repoRoot, wasmPath));
    } else {
      publishArgs.push('--module-path', 'server-rs/crates/spacetime-module');
    }
    await runCommand('spacetime', publishArgs, {
      env: {
        ...process.env,
        CARGO_TARGET_DIR: path.join(tempDir, 'cargo-target'),
      },
      sensitiveValues: [loginToken],
    });

    await runSmoke(configPath, serverUrl);
    console.log(
      '[editor-idempotency-smoke] Passed exact replay, body conflicts, deletion fail-close, concurrent calls, and orphan checks.',
    );
  } catch (error) {
    const standaloneOutput = sanitizeDiagnostic(
      standalone?.output().trim() ?? '',
      [loginToken],
    );
    console.error(
      `[editor-idempotency-smoke] Failed: ${sanitizeDiagnostic(
        error instanceof Error ? error.message : String(error),
        [loginToken],
      )}`,
    );
    if (standaloneOutput) {
      console.error(standaloneOutput);
    }
    process.exitCode = 1;
  } finally {
    if (standalone) {
      await stopStandalone(standalone.child);
    }
    if (tempDir) {
      await rm(tempDir, { recursive: true, force: true });
    }
  }
}

if (
  process.argv[1] &&
  fileURLToPath(import.meta.url) === path.resolve(process.argv[1])
) {
  await main();
}