#!/usr/bin/env node
// 仅在临时 standalone 验证客户端埋点；显式传入带测试 bootstrap hash 的 WASM。
import assert from 'node:assert/strict';
import { spawn } from 'node:child_process';
import { randomUUID } from 'node:crypto';
import { once } from 'node:events';
import { access, chmod, mkdtemp, rm, writeFile } 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 root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');
const database = 'agc-analytics-smoke';
const testSecret = 'a'.repeat(64); // 公开测试值，绝不能用于正式部署。
const sensitive = [testSecret];
const redact = (value) =>
  sensitive.reduce(
    (text, secret) => text.replaceAll(secret, '[REDACTED]'),
    String(value),
  );

function command(args) {
  return new Promise((resolve, reject) => {
    const child = spawn('spacetime', args, {
      cwd: root,
      windowsHide: true,
      shell: false,
      stdio: ['ignore', 'pipe', 'pipe'],
    });
    let output = '';
    for (const stream of [child.stdout, child.stderr])
      stream.on('data', (chunk) => {
        output = (output + chunk).slice(-16000);
      });
    const timer = setTimeout(() => {
      child.kill();
      reject(new Error('SpacetimeDB command timed out'));
    }, 120000);
    child.once('error', (error) => {
      clearTimeout(timer);
      reject(error);
    });
    child.once('exit', (code) => {
      clearTimeout(timer);
      if (code === 0) resolve(output);
      else reject(new Error(redact(output)));
    });
  });
}

async function localPort() {
  const listener = net.createServer();
  listener.listen(0, '127.0.0.1');
  await once(listener, 'listening');
  const port = listener.address().port;
  await new Promise((resolve) => listener.close(resolve));
  return port;
}

async function call(url, token, name, input) {
  const response = await fetch(`${url}/v1/database/${database}/call/${name}`, {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${token}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify([input]),
    signal: AbortSignal.timeout(30000),
  });
  const text = await response.text();
  assert(response.ok, `${name}: HTTP ${response.status}: ${redact(text)}`);
  return JSON.parse(text);
}

function ok(result) {
  assert.equal(result[0], 0, `Expected Ok, got ${JSON.stringify(result)}`);
  return JSON.parse(result[1]);
}

const event = (user = 'smoke-user-a') => ({
  schema_version: 1,
  event_id: randomUUID(),
  event_name: 'editor_session_start',
  event_time: '2026-09-21T12:00:00.123Z',
  user_id: user,
  editor_session_id: randomUUID(),
  project_id: null,
  creative_task_id: null,
  agent_run_id: null,
  agent_turn_id: null,
  status: 'success',
  error_code: null,
  source: 'editor',
  client_version: 'smoke-1',
  properties: { entry_source: 'direct_launch', first_project_id: null },
});
const batch = (events) => ({
  schema_version: 1,
  batch_id: randomUUID(),
  destination_origin: 'https://analytics-smoke.invalid',
  user_id: events[0].user_id,
  events,
});

function pricingInput() {
  const tiered = (model, unit, keys) => ({
    model,
    unit,
    price: [1, []],
    prices: keys.map((key) => ({ key, price: 1 })),
  });
  return {
    admin_user_id: 'smoke-bootstrap',
    updated_at_micros: 1,
    bootstrap_secret: testSecret,
    models: [
      tiered('gemini-3.1-flash-image-preview', 'perGeneration', [
        '0.5K',
        '1K',
        '2K',
      ]),
      tiered('gpt-image-2', 'perGeneration', ['1K', '2K']),
      ...['audio1.0', 'eleven_text_to_sound_v2', 'chirp-v5'].map((model) => ({
        model,
        unit: 'perGeneration',
        price: [0, 1],
        prices: [],
      })),
      ...[
        'seedance2.0-fast',
        'seedance2.0',
        'kling3.0',
        'kling3.0-omni',
        'veo3.1',
        'veo3.1-fast',
      ].map((model) => tiered(model, 'perSecond', ['480p', '720p', '1080p'])),
    ],
  };
}

async function verify(url, serviceToken, outsiderToken) {
  const upload = async (payload) =>
    call(
      url,
      serviceToken,
      'upload_agc_analytics_batch',
      JSON.stringify(payload),
    );
  const list = async (query = {}) =>
    ok(
      await call(
        url,
        serviceToken,
        'list_agc_tracking_events',
        JSON.stringify(query),
      ),
    );
  const first = batch([event(), event(), event()]);
  assert.deepEqual(ok(await upload(first)), {
    acknowledged_batch_ids: [first.batch_id],
    event_count: 3,
  });
  const initial = await list();
  assert.equal(initial.entries.length, 3);
  ok(await upload(first));
  assert.deepEqual(
    await list(),
    initial,
    'Replay must preserve rows, first batch ID and receipt time',
  );
  const reordered = structuredClone(first);
  reordered.events[0].properties = {
    first_project_id: null,
    entry_source: 'direct_launch',
  };
  ok(await upload(reordered));
  assert.deepEqual(
    await list(),
    initial,
    'JSON property order must not cause conflict',
  );

  const rollbackCandidate = event();
  const conflict = batch([
    rollbackCandidate,
    { ...first.events[0], client_version: 'changed' },
  ]);
  const rejected = await upload(conflict);
  assert.equal(rejected[0], 1);
  assert.equal(rejected[1], 'agc_event_conflict');
  assert.deepEqual(
    await list(),
    initial,
    'Earlier insert in failed batch must roll back',
  );
  const userB = batch([event('smoke-user-b')]);
  ok(await upload(userB));
  assert.equal(
    (await list({ userId: 'smoke-user-b' })).entries[0].eventId,
    userB.events[0].event_id,
  );

  const page1 = await list({ limit: 2 });
  assert(page1.nextCursor);
  const late = batch([event()]);
  ok(await upload(late));
  const seen = [...page1.entries];
  let cursor = page1.nextCursor;
  while (cursor) {
    const page = await list({ limit: 2, cursor });
    seen.push(...page.entries);
    cursor = page.nextCursor;
  }
  assert.equal(seen.length, 4);
  assert.equal(new Set(seen.map((row) => row.eventId)).size, 4);
  assert(
    !seen.some((row) => row.eventId === late.events[0].event_id),
    'Cursor snapshot must exclude later insert',
  );
  for (let index = 1; index < seen.length; index++) {
    const previous = seen[index - 1];
    const current = seen[index];
    assert(
      previous.receivedAt > current.receivedAt ||
        (previous.receivedAt === current.receivedAt &&
          previous.eventId > current.eventId),
      'Stable receipt/event ID descending order',
    );
  }
  const refreshed = await list();
  assert.equal(refreshed.entries.length, 5);
  assert.equal(refreshed.entries[0].eventId, late.events[0].event_id);
  assert.equal(refreshed.entries[0].projectId, null);
  assert.equal(refreshed.entries[0].eventTime, '2026-09-21T12:00:00.123Z');
  for (const [name, input] of [
    ['upload_agc_analytics_batch', first],
    ['list_agc_tracking_events', {}],
  ]) {
    const unauthorized = await call(
      url,
      outsiderToken,
      name,
      JSON.stringify(input),
    );
    assert.equal(unauthorized[0], 1, 'Nonservice identity must be rejected');
    assert.match(unauthorized[1], /无权/);
  }
  console.log(
    '[agc-analytics-smoke] PASS: first/replay, JSON order, atomic rollback, user filter, stable snapshot pagination, refresh, null/time, service authorization. 5 rows persisted.',
  );
}

async function main() {
  const wasm = process.env.GENARRATIVE_AGC_ANALYTICS_SMOKE_WASM;
  assert(
    wasm,
    'Set GENARRATIVE_AGC_ANALYTICS_SMOKE_WASM to an isolated test WASM compiled with SHA256(a repeated 64 times) bootstrap hash.',
  );
  await access(wasm);
  const version = await command(['--version']);
  assert(
    version.includes('version 2.8.3') &&
      version.includes('8e410d2842147bd8e5a32a9589cc00c19f7478e2'),
  );
  const temp = await mkdtemp(
    path.join(os.tmpdir(), 'genarrative-agc-analytics-smoke-'),
  );
  let standalone;
  try {
    const port = await localPort();
    const url = `http://127.0.0.1:${port}`;
    standalone = spawn(
      'spacetime',
      [
        'start',
        '--data-dir',
        path.join(temp, 'data'),
        '--listen-addr',
        `127.0.0.1:${port}`,
        '--non-interactive',
      ],
      { windowsHide: true, shell: false, stdio: ['ignore', 'pipe', 'pipe'] },
    );
    let logs = '';
    for (const stream of [standalone.stdout, standalone.stderr])
      stream.on('data', (chunk) => {
        logs = (logs + chunk).slice(-8000);
      });
    const deadline = Date.now() + 30000;
    for (;;) {
      assert(standalone.exitCode === null, redact(logs));
      try {
        if (
          (await fetch(`${url}/v1/ping`, { signal: AbortSignal.timeout(1000) }))
            .ok
        )
          break;
      } catch {
        /* 等待启动 */
      }
      assert(
        Date.now() < deadline,
        `Standalone startup timeout: ${redact(logs)}`,
      );
      await delay(200);
    }
    const owner = await createSpacetimeWebIdentity({
      database,
      serverUrl: url,
    });
    const service = await createSpacetimeWebIdentity({
      database,
      serverUrl: url,
    });
    sensitive.push(owner.token, service.token);
    const config = path.join(temp, 'cli.toml');
    await command(['--config-path', config, 'login', '--token', owner.token]);
    await chmod(config, 0o600);
    await command([
      '--config-path',
      config,
      'publish',
      database,
      '--server',
      url,
      '--yes=all',
      '--no-config',
      '--bin-path',
      path.resolve(wasm),
    ]);
    const bootstrap = await call(
      url,
      service.token,
      'initialize_editor_generation_pricing_config_if_missing_and_return',
      pricingInput(),
    );
    assert.equal(
      bootstrap[0],
      true,
      `Pricing service bootstrap failed: ${redact(JSON.stringify(bootstrap))}`,
    );
    await verify(url, service.token, owner.token);
    if (process.env.AGC_ANALYTICS_SMOKE_KEEP === '1') {
      const stopFile = path.join(temp, 'stop');
      const contextPath = path.join(temp, 'connection.json');
      await writeFile(
        contextPath,
        JSON.stringify({
          serverUrl: url,
          database,
          token: service.token,
          operatorToken: owner.token,
          configPath: config,
          stopFile,
        }),
        { mode: 0o600 },
      );
      console.log(
        `[agc-analytics-smoke] Integration connection file: ${contextPath}`,
      );
      const stopDeadline = Date.now() + 45 * 60 * 1000;
      while (Date.now() < stopDeadline) {
        try {
          await access(stopFile);
          break;
        } catch {
          /* 联调结束后由调用方创建 stop 文件 */
        }
        await delay(1000);
      }
    }
  } finally {
    if (standalone && standalone.exitCode === null) {
      if (process.platform === 'win32') {
        const killer = spawn(
          'taskkill',
          ['/PID', String(standalone.pid), '/T', '/F'],
          { windowsHide: true, stdio: 'ignore' },
        );
        await once(killer, 'exit');
      } else {
        standalone.kill('SIGTERM');
      }
      if (standalone.exitCode === null) await once(standalone, 'exit');
    }
    // 仅删除本脚本 mkdtemp 创建的系统临时子目录。
    assert(
      path.dirname(path.resolve(temp)) === path.resolve(os.tmpdir()) &&
        path.basename(temp).startsWith('genarrative-agc-analytics-smoke-'),
    );
    await rm(temp, {
      recursive: true,
      force: true,
      maxRetries: 5,
      retryDelay: 200,
    });
  }
}

main().catch((error) => {
  console.error(redact(error.stack ?? error));
  process.exitCode = 1;
});