global-setup.mjs

import { GenericContainer, Network, Wait } from 'testcontainers';

import { KAFKA_CONNECT_IMAGE, OWL_SHOP_IMAGE, REDPANDA_IMAGE } from './test-images.mjs';
import { exec } from 'node:child_process';
import { existsSync, readFileSync, realpathSync, writeFileSync } from 'node:fs';
import { dirname, join, resolve } from 'node:path';
import { fileURLToPath } from 'node:url';
import { promisify } from 'node:util';

const execAsync = promisify(exec);
const __filename = fileURLToPath(import.meta.url);
const __dirname = dirname(__filename);

// Regex for extracting container ID from error messages
const CONTAINER_ID_REGEX = /container ([a-f0-9]+)/;

const getStateFile = (variantName) => resolve(__dirname, '..', `.testcontainers-state-${variantName}.json`);

/**
 * Load variant configuration from test-variant-{name}/config/variant.json
 * @param {string} variantName - The variant name (e.g., "console", "console-enterprise")
 * @returns {object} - The variant configuration including ports
 */
function loadVariantConfig(variantName) {
  const variantDir = resolve(__dirname, '..', `test-variant-${variantName}`);
  const configPath = join(variantDir, 'config', 'variant.json');

  if (!existsSync(configPath)) {
    throw new Error(`Variant config not found: ${configPath}`);
  }

  const config = JSON.parse(readFileSync(configPath, 'utf-8'));
  return config;
}

async function waitForPort(port, maxAttempts = 30, delayMs = 500) {
  for (let i = 0; i < maxAttempts; i++) {
    try {
      const { stdout } = await execAsync(
        `curl -4 -s -m 5 -o /dev/null -w "%{http_code}" http://localhost:${port}/ || echo "0"`
      );
      const statusCode = Number.parseInt(stdout.trim(), 10);
      if (statusCode > 0 && statusCode < 500) {
        return true;
      }
      if ((i + 1) % 5 === 0) {
        console.log(
          `  Still waiting for port ${port}... (attempt ${i + 1}/${maxAttempts}, last status: ${statusCode})`
        );
      }
    } catch (error) {
      // Port not ready yet
      if ((i + 1) % 5 === 0) {
        console.log(`  Still waiting for port ${port}... (attempt ${i + 1}/${maxAttempts}, error: ${error.message})`);
      }
    }
    await new Promise((resolve) => setTimeout(resolve, delayMs));
  }
  throw new Error(`Port ${port} failed to become available after ${maxAttempts} attempts`);
}
async function setupDockerNetwork(state) {
  console.log('Creating Docker network...');
  const network = await new Network().start();
  state.networkId = network.getId();
  state.network = network;
  console.log(`✓ Network created: ${state.networkId}`);
  return network;
}

async function startRedpandaContainer(network, state, ports) {
  console.log('Starting Redpanda container...');
  const redpanda = await new GenericContainer(REDPANDA_IMAGE)
    .withNetwork(network)
    .withNetworkAliases('redpanda')
    .withExposedPorts(
      { container: 19_092, host: ports.redpandaKafka },
      { container: 18_081, host: ports.redpandaSchemaRegistry },
      { container: 18_082, host: ports.redpandaPandaproxy },
      { container: 9644, host: ports.redpandaAdmin }
    )
    .withCommand([
      'redpanda',
      'start',
      '--smp',
      '1',
      '--mode dev-container',
      '--overprovisioned',
      '--kafka-addr',
      'internal://0.0.0.0:9092,external://0.0.0.0:19092',
      '--advertise-kafka-addr',
      'internal://redpanda:9092,external://localhost:19092',
      '--pandaproxy-addr',
      'internal://0.0.0.0:8082,external://0.0.0.0:18082',
      '--advertise-pandaproxy-addr',
      `internal://redpanda:8082,external://localhost:${ports.redpandaPandaproxy}`,
      '--schema-registry-addr',
      'internal://0.0.0.0:8081,external://0.0.0.0:18081',
      '--rpc-addr',
      'redpanda:33145',
      '--advertise-rpc-addr',
      'redpanda:33145',
    ])
    .withEnvironment({
      RP_BOOTSTRAP_USER: 'e2euser:very-secret',
    })
    .withBindMounts([
      {
        source: resolve(__dirname, 'conf/.bootstrap.yaml'),
        target: '/etc/redpanda/.bootstrap.yaml',
      },
    ])
    .withHealthCheck({
      test: ['CMD-SHELL', "rpk cluster health | grep -E 'Healthy:.+true' || exit 1"],
      interval: 10_000,
      timeout: 3000,
      retries: 5,
      startPeriod: 5000,
    })
    .withWaitStrategy(Wait.forHealthCheck())
    .withStartupTimeout(90_000)
    .start();

  state.redpandaId = redpanda.getId();
  state.redpandaContainer = redpanda;
  console.log(`✓ Redpanda container started: ${state.redpandaId}`);
}

async function verifyRedpandaServices(state, ports) {
  // Docker health check already verified rpk cluster health passed.
  // Just verify Schema Registry is responding (it can lag behind the broker).
  console.log('Checking if Schema Registry is ready...');
  for (let i = 0; i < 30; i++) {
    try {
      await execAsync(
        `docker exec ${state.redpandaId} curl -s -m 3 -o /dev/null -w "%{http_code}" http://localhost:18081/subjects`
      );
      break;
    } catch {
      if ((i + 1) % 5 === 0) {
        console.log(`  Still waiting for Schema Registry... (attempt ${i + 1}/30)`);
      }
      await new Promise((resolve) => setTimeout(resolve, 1000));
    }
  }
  console.log('✓ Schema Registry is ready');

  // Verify SASL authentication is working
  console.log('Verifying Redpanda SASL authentication...');
  try {
    await execAsync(
      `docker exec ${state.redpandaId} rpk cluster info -X brokers=redpanda:9092 -X user=e2euser -X pass=very-secret -X sasl.mechanism=SCRAM-SHA-256`
    );
    console.log('✓ SASL authentication verified');
  } catch (error) {
    console.log('⚠ SASL authentication check failed:', error.message);
  }
}

async function startOwlShop(network, state) {
  console.log('Starting OwlShop container...');
  const owlshopConfigContent = `
shop:
  requestRate: 1
  interval: 0.1s
  topicReplicationFactor: 1
  topicPartitionCount: 1
kafka:
  brokers: "redpanda:9092"
  sasl:
    enabled: true
    mechanism: SCRAM-SHA-256
    username: e2euser
    password: very-secret
schemaRegistry:
  address: "http://redpanda:8081"
`;

  const owlshop = await new GenericContainer(OWL_SHOP_IMAGE)
    .withPlatform('linux/amd64')
    .withNetwork(network)
    .withNetworkAliases('owlshop')
    .withEnvironment({
      CONFIG_FILEPATH: '/tmp/config.yml',
      OWLSHOP_CONFIG_FILE: owlshopConfigContent,
    })
    .withEntrypoint(['/bin/sh'])
    .withCommand(['-c', 'echo "$OWLSHOP_CONFIG_FILE" > /tmp/config.yml && /app/owlshop'])
    .withStartupTimeout(120_000)
    .start();

  state.owlshopId = owlshop.getId();
  state.owlshopContainer = owlshop;
  console.log(`✓ OwlShop started: ${state.owlshopId}`);
}

async function createKafkaConnectTopics(state) {
  console.log('Pre-creating Kafka Connect internal topics...');

  // Kafka Connect internal storage topics must have cleanup.policy=compact
  const compactTopics = ['_internal_connectors_offsets', '_internal_connectors_configs', '_internal_connectors_status'];
  // Log topic uses default delete policy
  const deleteTopics = ['__redpanda.connectors_logs'];

  const saslFlags = '-X user=e2euser -X pass=very-secret -X sasl.mechanism=SCRAM-SHA-256';

  for (const topic of compactTopics) {
    try {
      await execAsync(
        `docker exec ${state.redpandaId} rpk topic create ${topic} --replicas 1 --partitions 1 --topic-config cleanup.policy=compact ${saslFlags}`
      );
      console.log(`  ✓ Created topic: ${topic} (compact)`);
    } catch (_error) {
      console.log(`  - Topic ${topic} already exists or creation skipped`);
    }
  }

  for (const topic of deleteTopics) {
    try {
      await execAsync(
        `docker exec ${state.redpandaId} rpk topic create ${topic} --replicas 1 --partitions 1 ${saslFlags}`
      );
      console.log(`  ✓ Created topic: ${topic}`);
    } catch (_error) {
      console.log(`  - Topic ${topic} already exists or creation skipped`);
    }
  }
}

async function startKafkaConnect(network, state, ports) {
  console.log('Starting Kafka Connect container...');

  // Write SASL password to a temp file so it can be mounted into the container.
  // The Redpanda Connectors image reads the password from a file at
  // /opt/kafka/connect-password/${CONNECT_SASL_PASSWORD_FILE} — passing SASL
  // settings via CONNECT_CONFIGURATION doesn't work because the config generator
  // appends security.protocol=PLAINTEXT after the user-provided config, overriding it.
  const passwordFile = resolve(__dirname, '.connect-sasl-password.tmp');
  writeFileSync(passwordFile, 'very-secret');

  const connectConfig = `
key.converter=org.apache.kafka.connect.converters.ByteArrayConverter
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
group.id=connectors-cluster
offset.storage.topic=_internal_connectors_offsets
config.storage.topic=_internal_connectors_configs
status.storage.topic=_internal_connectors_status
config.storage.replication.factor=1
offset.storage.replication.factor=1
status.storage.replication.factor=1
offset.flush.interval.ms=1000
producer.linger.ms=50
producer.batch.size=131072
topic.creation.enable=false
`;

  let connect;
  try {
    connect = await new GenericContainer(KAFKA_CONNECT_IMAGE)
      .withPlatform('linux/amd64')
      .withNetwork(network)
      .withNetworkAliases('connect')
      .withExposedPorts({ container: 8083, host: ports.kafkaConnect })
      .withBindMounts([
        {
          source: passwordFile,
          target: '/opt/kafka/connect-password/e2e-password',
          mode: 'ro',
        },
      ])
      .withEnvironment({
        CONNECT_CONFIGURATION: connectConfig,
        CONNECT_BOOTSTRAP_SERVERS: 'redpanda:9092',
        CONNECT_SASL_MECHANISM: 'SCRAM-SHA-256',
        CONNECT_SASL_USERNAME: 'e2euser',
        CONNECT_SASL_PASSWORD_FILE: 'e2e-password',
        CONNECT_GC_LOG_ENABLED: 'false',
        CONNECT_HEAP_OPTS: '-Xms512M -Xmx512M',
        CONNECT_LOG_LEVEL: 'info',
        CONNECT_TOPIC_LOG_ENABLED: 'true',
      })
      // Use log-based wait: avoids IPv4/IPv6 port-forwarding issues on macOS Docker Desktop.
      // Kafka Connect prints "REST resources initialized" when the HTTP API is ready.
      .withWaitStrategy(Wait.forLogMessage(/.*REST resources initialized.*/i))
      .withStartupTimeout(300_000)
      .start();

    state.connectId = connect.getId();
    state.connectContainer = connect;
    console.log(`✓ Kafka Connect container started: ${state.connectId}`);

    // Verify it's responding via docker exec (avoids macOS Docker Desktop port-forwarding latency)
    console.log('Verifying Kafka Connect API...');
    for (let i = 0; i < 30; i++) {
      try {
        await execAsync(
          `docker exec ${state.redpandaId} curl -s -m 5 -o /dev/null -w "%{http_code}" http://connect:8083/`
        );
        break;
      } catch {
        if ((i + 1) % 5 === 0) console.log(`  Still waiting for Kafka Connect API... (attempt ${i + 1}/30)`);
        await new Promise((resolve) => setTimeout(resolve, 2000));
      }
    }
    console.log('✓ Kafka Connect API ready (internal)');

    // Also wait for the host port to be forwarded (needed for tests that POST directly to the API)
    console.log(`Waiting for Kafka Connect host port ${ports.kafkaConnect}...`);
    await waitForPort(ports.kafkaConnect, 60, 2000);
    console.log(`✓ Kafka Connect host port ${ports.kafkaConnect} ready`);
  } catch (error) {
    console.log('⚠ Kafka Connect failed to start (connector tests may fail)');
    console.log(`  Error: ${error.message}`);

    // Try to get container ID and logs for debugging
    if (connect) {
      try {
        const containerId = connect.getId();
        state.connectId = containerId;
        state.connectContainer = connect;
        console.log(`\n  Container ID: ${containerId}`);
        console.log('  Last 50 lines of Kafka Connect logs:');
        const { stdout } = await execAsync(`docker logs --tail 50 ${containerId}`);
        console.log(stdout);
      } catch {
        console.log('  Could not retrieve container logs');
      }
    } else {
      console.log('  Container failed to start - no logs available');
    }
  } finally {
    // Clean up temp password file
    try {
      await execAsync(`rm -f "${passwordFile}"`);
    } catch {
      // ignore cleanup errors
    }
  }
}

export async function buildBackendImage(isEnterprise) {
  console.log(`Building backend Docker image ${isEnterprise ? '(Enterprise)' : '(OSS)'}...`);

  let backendDir;
  if (isEnterprise) {
    // Check for ENTERPRISE_BACKEND_DIR environment variable
    if (process.env.ENTERPRISE_BACKEND_DIR) {
      backendDir = resolve(process.env.ENTERPRISE_BACKEND_DIR);
      console.log(`Using ENTERPRISE_BACKEND_DIR from environment: ${backendDir}`);
    } else {
      // Default to relative path
      backendDir = resolve(__dirname, '../../../../console-enterprise/backend');
    }
  } else {
    backendDir = resolve(__dirname, '../../../backend');
  }

  // Resolve symlinks to real path (needed for Docker build context)
  if (existsSync(backendDir)) {
    backendDir = realpathSync(backendDir);
  }

  const imageTag = isEnterprise ? 'console-backend:e2e-test-enterprise' : 'console-backend:e2e-test';

  console.log(`Building from: ${backendDir}`);

  let embedDir = null;

  try {
    // Copy frontend assets before build (required for both OSS and Enterprise)
    // The pkg/embed/frontend/ directory has .gitignore with *, so assets don't exist in CI
    if (isEnterprise && !existsSync(backendDir)) {
      throw new Error(
        `Enterprise backend directory not found: ${backendDir}\nEnterprise E2E tests require console-enterprise repo to be checked out alongside console repo.`
      );
    }

    const frontendBuildDir = resolve(__dirname, '../../build');

    // Check if frontend build exists
    if (!existsSync(frontendBuildDir)) {
      throw new Error(
        `Frontend build directory not found: ${frontendBuildDir}\nRun "bun run build" before running E2E tests.`
      );
    }

    embedDir = join(backendDir, 'pkg/embed/frontend');

    console.log(`Copying frontend assets to ${isEnterprise ? 'enterprise' : 'OSS'} backend...`);
    console.log(`  From: ${frontendBuildDir}`);
    console.log(`  To: ${embedDir}`);

    // Copy all files from build/ to pkg/embed/frontend/
    await execAsync(`cp -r "${frontendBuildDir}"/* "${embedDir}"/`);
    console.log('✓ Frontend assets copied');

    // Build Docker image using testcontainers
    // Docker doesn't allow Dockerfiles to reference files outside build context,
    // so we temporarily copy the Dockerfile into the build context
    const dockerfilePath = resolve(__dirname, 'Dockerfile.backend');
    const tempDockerfile = join(backendDir, '.dockerfile.e2e.tmp');

    // For enterprise workspace builds (go.work exists), the workspace references
    // modules outside the backend/ directory (e.g., ../console-oss/backend).
    // We copy those into the build context and rewrite go.work to use local paths.
    const isWorkspaceBuild = isEnterprise && existsSync(join(backendDir, 'go.work'));
    const workspaceDir = join(backendDir, '.e2e-workspace');

    if (isWorkspaceBuild) {
      console.log('Workspace build detected (go.work found)');
      const goWorkContent = readFileSync(join(backendDir, 'go.work'), 'utf-8');

      // Parse workspace module paths (skip "." which is the backend itself)
      const useRegex = /^\s+(\S+)\s*$/gm;
      const rewrittenPaths = [];
      let match;
      while ((match = useRegex.exec(goWorkContent)) !== null) {
        const modulePath = match[1];
        if (modulePath === '.' || modulePath === 'use' || modulePath === '(' || modulePath === ')') continue;

        // Resolve the actual path relative to backendDir
        const absModulePath = resolve(backendDir, modulePath);
        if (!existsSync(absModulePath)) {
          console.warn(`  Workspace module not found: ${absModulePath}, skipping`);
          continue;
        }

        // Create a sanitized directory name
        const localName = modulePath.replace(/[./]/g, '-').replace(/^-+|-+$/g, '');
        const destPath = join(workspaceDir, localName);

        console.log(`  Copying workspace module: ${modulePath} -> .e2e-workspace/${localName}`);
        await execAsync(`mkdir -p "${destPath}" && cp -r "${absModulePath}"/* "${destPath}"/`);
        rewrittenPaths.push({ original: modulePath, local: `.e2e-workspace/${localName}` });
      }

      // Rewrite go.work to use local paths
      if (rewrittenPaths.length > 0) {
        let newGoWork = goWorkContent;
        for (const { original, local } of rewrittenPaths) {
          newGoWork = newGoWork.replace(original, local);
        }
        writeFileSync(join(backendDir, 'go.work'), newGoWork);
        console.log('  ✓ Rewrote go.work with local paths');
      }
    }

    console.log('Building Docker image with BuildKit...');
    await execAsync(`cp "${dockerfilePath}" "${tempDockerfile}"`);

    try {
      // Use docker buildx with BuildKit cache mounts for Go module and build caches.
      // This is significantly faster than testcontainers build on repeat runs.
      await execAsync(`DOCKER_BUILDKIT=1 docker build -f .dockerfile.e2e.tmp -t ${imageTag} .`, {
        cwd: backendDir,
        maxBuffer: 50 * 1024 * 1024,
      });
      console.log('✓ Backend image built');
    } finally {
      // Clean up temporary Dockerfile and workspace directory
      await execAsync(`rm -f "${tempDockerfile}"`).catch(() => {});
      if (isWorkspaceBuild) {
        await execAsync(`rm -rf "${workspaceDir}"`).catch(() => {});
      }
    }

    return imageTag;
  } finally {
    // Cleanup: remove copied frontend assets (for both OSS and Enterprise)
    if (embedDir) {
      console.log('Cleaning up copied frontend assets...');
      // Keep .gitignore, remove everything else
      await execAsync(`find "${embedDir}" -mindepth 1 ! -name '.gitignore' -delete`).catch(() => {});
      console.log('✓ Cleanup complete');
    }
  }
}

// biome-ignore lint/complexity/noExcessiveCognitiveComplexity: (21) nested test environment setup with multiple configuration checks
async function startBackendServer(network, isEnterprise, imageTag, state, variantName, configFile, ports) {
  console.log('Starting backend server container...');
  console.log(`Image tag: ${imageTag}`);
  console.log(`Enterprise mode: ${isEnterprise}`);

  const backendConfigPath = resolve(__dirname, '..', `test-variant-${variantName}`, 'config', configFile);

  console.log(`Backend config path: ${backendConfigPath}`);

  const bindMounts = [
    {
      source: backendConfigPath,
      target: '/etc/console/config.yaml',
      mode: 'ro',
    },
  ];

  // Mount license file for enterprise mode
  if (isEnterprise) {
    let licensePath;
    const fs = await import('node:fs');

    // Check if license is provided as GitHub secret (environment variable)
    if (process.env.ENTERPRISE_LICENSE_CONTENT) {
      console.log('Using ENTERPRISE_LICENSE_CONTENT from environment variable (GitHub secret)');
      // Write the license content to a temporary file
      const { mkdtempSync, writeFileSync } = await import('node:fs');
      const { tmpdir } = await import('node:os');
      const tempDir = mkdtempSync(`${tmpdir()}/redpanda-license-`);
      licensePath = `${tempDir}/redpanda.license`;
      writeFileSync(licensePath, process.env.ENTERPRISE_LICENSE_CONTENT);
      console.log(`✓ License written to temporary file: ${licensePath}`);
    } else {
      // Default to relative path based on backend directory
      const backendDir = process.env.ENTERPRISE_BACKEND_DIR
        ? resolve(process.env.ENTERPRISE_BACKEND_DIR)
        : resolve(__dirname, '../../../../console-enterprise/backend');

      const defaultLicensePath = resolve(backendDir, '../frontend/tests/config/redpanda.license');
      licensePath = process.env.REDPANDA_LICENSE_PATH || defaultLicensePath;
      console.log(`Enterprise license path: ${licensePath}`);

      if (!fs.existsSync(licensePath)) {
        throw new Error(`License file not found at: ${licensePath}`);
      }
      console.log('✓ License file found');
    }

    bindMounts.push({
      source: licensePath,
      target: '/etc/console/redpanda.license',
      mode: 'ro',
    });
  }

  console.log('Creating container with bind mounts:');
  bindMounts.forEach((mount, i) => {
    console.log(`  [${i}] ${mount.source} -> ${mount.target}`);
  });

  let backend;
  let containerId;
  try {
    console.log('Starting container...');
    console.log('Configuration summary:');
    console.log(`  - Network: ${network.getId ? network.getId() : 'unknown'}`);
    console.log('  - Alias: console-backend');
    console.log(`  - Port: ${ports.backend}:3000`);
    console.log('  - Command: --config.filepath=/etc/console/config.yaml');

    // Create container without wait strategy first to get the ID immediately
    const container = new GenericContainer(imageTag)
      .withNetwork(network)
      .withNetworkAliases('console-backend')
      .withNetworkMode(network.getName())
      .withExposedPorts({ container: 3000, host: ports.backend })
      .withBindMounts(bindMounts)
      .withCommand(['--config.filepath=/etc/console/config.yaml']);

    console.log('Calling container.start()...');

    try {
      backend = await container.start();
      containerId = backend.getId();
      state.backendId = containerId;
      state.backendContainer = backend;
      console.log(`✓ Container.start() returned with ID: ${containerId}`);
    } catch (startError) {
      console.error('Error during container.start():', startError.message);

      // Extract container ID from error message if available
      const containerIdMatch = startError.message.match(CONTAINER_ID_REGEX);
      const failedContainerId = containerIdMatch ? containerIdMatch[1] : null;

      if (failedContainerId) {
        console.log(`Container ID from error: ${failedContainerId}`);
        containerId = failedContainerId;
        state.backendId = failedContainerId;

        try {
          // Get logs from this container
          console.log('Fetching logs from failed container...');
          const { stdout: logs } = await execAsync(`docker logs ${failedContainerId} 2>&1`);
          console.log('=== CONTAINER LOGS START ===');
          console.log(logs || '(no logs)');
          console.log('=== CONTAINER LOGS END ===');

          // Get container state
          console.log('Fetching container state...');
          const { stdout: stateJson } = await execAsync(
            `docker inspect ${failedContainerId} --format='{{json .State}}'`
          );
          console.log('Container state:');
          const state = JSON.parse(stateJson);
          console.log(JSON.stringify(state, null, 2));

          // Get container config to see the actual command and mounts
          console.log('Fetching container config...');
          const { stdout: configJson } = await execAsync(
            `docker inspect ${failedContainerId} --format='{{json .Config}}'`
          );
          const config = JSON.parse(configJson);
          console.log('Container command:', config.Cmd);
          console.log('Container entrypoint:', config.Entrypoint);

          // Get mount info
          console.log('Fetching mount info...');
          const { stdout: mountsJson } = await execAsync(
            `docker inspect ${failedContainerId} --format='{{json .Mounts}}'`
          );
          const mounts = JSON.parse(mountsJson);
          console.log('Container mounts:');
          console.log(JSON.stringify(mounts, null, 2));
        } catch (inspectError) {
          console.error('Failed to inspect container:', inspectError.message);
        }
      } else {
        console.log('Could not extract container ID from error message');
        console.log('Full error:', JSON.stringify(startError, null, 2));
      }

      throw startError;
    }

    console.log(`Container created with ID: ${containerId}`);

    // Check if container is still running
    const { stdout: inspectOutput } = await execAsync(`docker inspect ${containerId} --format='{{.State.Status}}'`);
    const containerStatus = inspectOutput.trim();
    console.log(`Container status: ${containerStatus}`);

    if (containerStatus !== 'running') {
      console.error(`Container is not running (status: ${containerStatus})`);
      console.log('Fetching container logs...');
      const { stdout: logs } = await execAsync(`docker logs ${containerId} 2>&1`);
      console.log('Container logs:');
      console.log(logs);

      // Get exit code
      const { stdout: exitCode } = await execAsync(`docker inspect ${containerId} --format='{{.State.ExitCode}}'`);
      console.log(`Container exit code: ${exitCode.trim()}`);

      throw new Error(`Container stopped immediately with status: ${containerStatus}`);
    }

    console.log(`✓ Backend container started and running: ${containerId}`);

    // Get initial logs to see startup
    console.log('Fetching initial container logs...');
    const { stdout: logs } = await execAsync(`docker logs ${containerId} 2>&1 | tail -30`);
    if (logs) {
      console.log('Container logs:');
      console.log(logs);
    }

    // Wait for backend via docker exec (avoids macOS Docker Desktop port-forwarding latency)
    console.log('Waiting for backend to be ready (internal port 3000)...');
    for (let i = 0; i < 90; i++) {
      try {
        await execAsync(
          `docker exec ${state.redpandaId} curl -s -m 5 -o /dev/null -w "%{http_code}" http://console-backend:3000/`
        );
        break;
      } catch {
        if ((i + 1) % 10 === 0) console.log(`  Still waiting for backend... (attempt ${i + 1}/90)`);
        await new Promise((resolve) => setTimeout(resolve, 2000));
      }
    }
    console.log('✓ Backend ready (internal)');

    // Also wait for host port to be forwarded (browser navigates to localhost:port)
    console.log(`Waiting for backend host port ${ports.backend}...`);
    await waitForPort(ports.backend, 90, 2000);
    console.log(`✓ Backend host port ${ports.backend} ready`);
  } catch (error) {
    console.error('Failed to start backend container:', error.message);

    // Try to get container logs if container was created but failed
    if (containerId) {
      try {
        console.log('Attempting to fetch logs from failed container...');
        const { stdout: logs } = await execAsync(`docker logs ${containerId} 2>&1`);
        console.log('Full container logs:');
        console.log(logs);

        // Get container inspect info
        const { stdout: inspect } = await execAsync(`docker inspect ${containerId}`);
        console.log('Container inspect (state):');
        const inspectJson = JSON.parse(inspect);
        console.log(JSON.stringify(inspectJson[0].State, null, 2));
      } catch (logError) {
        console.error('Could not fetch logs:', logError.message);
      }
    }

    throw error;
  }
}

async function startDestinationRedpandaContainer(network, state, ports) {
  console.log('Starting destination Redpanda container for shadowlink...');
  const destRedpanda = await new GenericContainer(REDPANDA_IMAGE)
    .withNetwork(network)
    .withNetworkAliases('dest-cluster')
    .withExposedPorts(
      { container: 19_093, host: ports.destRedpandaKafka },
      { container: 18_091, host: ports.destRedpandaSchemaRegistry },
      { container: 18_092, host: ports.destRedpandaPandaproxy },
      { container: 9644, host: ports.destRedpandaAdmin }
    )
    .withCommand([
      'redpanda',
      'start',
      '--smp',
      '1',
      '--mode dev-container',
      '--overprovisioned',
      '--kafka-addr',
      'internal://0.0.0.0:9092,external://0.0.0.0:19093',
      '--advertise-kafka-addr',
      'internal://dest-cluster:9092,external://localhost:19093',
      '--pandaproxy-addr',
      'internal://0.0.0.0:8082,external://0.0.0.0:18092',
      '--advertise-pandaproxy-addr',
      `internal://dest-cluster:8082,external://localhost:${ports.destRedpandaPandaproxy}`,
      '--schema-registry-addr',
      'internal://0.0.0.0:8081,external://0.0.0.0:18091',
      '--rpc-addr',
      'dest-cluster:33145',
      '--advertise-rpc-addr',
      'dest-cluster:33145',
    ])
    .withEnvironment({
      RP_BOOTSTRAP_USER: 'e2euser:very-secret',
    })
    .withBindMounts([
      {
        source: resolve(__dirname, 'conf/.bootstrap-dest.yaml'),
        target: '/etc/redpanda/.bootstrap.yaml',
      },
    ])
    .withHealthCheck({
      test: ['CMD-SHELL', "rpk cluster health | grep -E 'Healthy:.+true' || exit 1"],
      interval: 15_000,
      timeout: 3000,
      retries: 5,
      startPeriod: 5000,
    })
    .withWaitStrategy(Wait.forHealthCheck())
    .withStartupTimeout(120_000)
    .start();

  state.destRedpandaId = destRedpanda.getId();
  state.destRedpandaContainer = destRedpanda;
  console.log(`✓ Destination Redpanda container started: ${state.destRedpandaId}`);

  // Debug: Check port mappings
  try {
    const { stdout: portOutput } = await execAsync(`docker port ${state.destRedpandaId}`);
    console.log('Destination container port mappings:');
    console.log(portOutput);
  } catch (e) {
    console.log('Could not get port mappings:', e.message);
  }
}

async function verifyDestinationRedpandaServices(state, ports) {
  console.log('Waiting for destination Redpanda services...');

  // Debug: Check if container is still running
  try {
    const { stdout: status } = await execAsync(`docker inspect ${state.destRedpandaId} --format='{{.State.Status}}'`);
    console.log(`Destination container status: ${status.trim()}`);
    if (status.trim() !== 'running') {
      const { stdout: logs } = await execAsync(`docker logs ${state.destRedpandaId} 2>&1 | tail -30`);
      console.log('Destination container logs:');
      console.log(logs);
    }
  } catch (e) {
    console.log('Could not check container status:', e.message);
  }

  console.log(`Checking destination Admin API (port ${ports.destRedpandaAdmin})...`);
  await waitForPort(ports.destRedpandaAdmin, 60, 2000);
  console.log('✓ Destination Admin API ready');

  console.log(`Checking destination Schema Registry (port ${ports.destRedpandaSchemaRegistry})...`);
  await waitForPort(ports.destRedpandaSchemaRegistry, 60, 2000);
  console.log('✓ Destination Schema Registry ready');
}

async function startBackendServerWithConfig(
  network,
  isEnterprise,
  imageTag,
  state,
  configPath,
  externalPort,
  networkAlias,
  extraHosts = []
) {
  console.log(`Starting backend server container on port ${externalPort} with alias ${networkAlias}...`);

  const bindMounts = [
    {
      source: configPath,
      target: '/etc/console/config.yaml',
      mode: 'ro',
    },
  ];

  if (isEnterprise) {
    let licensePath;
    const fs = await import('node:fs');

    if (process.env.ENTERPRISE_LICENSE_CONTENT) {
      const { mkdtempSync, writeFileSync } = await import('node:fs');
      const { tmpdir } = await import('node:os');
      const tempDir = mkdtempSync(`${tmpdir()}/redpanda-license-`);
      licensePath = `${tempDir}/redpanda.license`;
      writeFileSync(licensePath, process.env.ENTERPRISE_LICENSE_CONTENT);
    } else {
      const defaultLicensePath = resolve(
        __dirname,
        '../../../../console-enterprise/frontend/tests/config/redpanda.license'
      );
      licensePath = process.env.REDPANDA_LICENSE_PATH || defaultLicensePath;

      if (!fs.existsSync(licensePath)) {
        throw new Error(`License file not found at: ${licensePath}`);
      }
    }

    bindMounts.push({
      source: licensePath,
      target: '/etc/console/redpanda.license',
      mode: 'ro',
    });
  }

  let containerId;
  try {
    let container = new GenericContainer(imageTag)
      .withNetwork(network)
      .withNetworkAliases(networkAlias)
      .withExposedPorts({ container: 3000, host: externalPort })
      .withBindMounts(bindMounts)
      .withCommand(['--config.filepath=/etc/console/config.yaml']);

    if (extraHosts.length > 0) {
      container = container.withExtraHosts(extraHosts);
    }

    const backend = await container.start();

    containerId = backend.getId();
    state.backendId = containerId;
    state.backendContainer = backend;

    await new Promise((resolve) => setTimeout(resolve, 2000));

    // Check if container is still running
    const { stdout: status } = await execAsync(`docker inspect ${containerId} --format='{{.State.Status}}'`);
    if (status.trim() !== 'running') {
      const { stdout: logs } = await execAsync(`docker logs ${containerId} 2>&1`);
      const { stdout: exitCode } = await execAsync(`docker inspect ${containerId} --format='{{.State.ExitCode}}'`);
      console.error(`Container ${containerId} stopped (exit ${exitCode.trim()}):`);
      console.error(logs);
      throw new Error(`Container stopped immediately with status: ${status.trim()}`);
    }

    await waitForPort(externalPort, 60, 1000);
    console.log(`✓ Backend ready at http://localhost:${externalPort}`);
  } catch (error) {
    console.error(`Failed to start backend on port ${externalPort}:`, error.message);

    if (containerId) {
      try {
        const { stdout: logs } = await execAsync(`docker logs ${containerId} 2>&1`);
        const { stdout: inspect } = await execAsync(`docker inspect ${containerId}`);
        const inspectJson = JSON.parse(inspect);
        console.error('Container state:', JSON.stringify(inspectJson[0].State, null, 2));
        console.error('Container logs:', logs);
      } catch (logError) {
        console.error('Could not fetch container diagnostics:', logError.message);
      }
    }

    throw error;
  }
}

/**
 * Start the Console backend as a host process (not in Docker).
 * Used for OIDC tests where the backend must reach both localhost services
 * (Zitadel) and port-mapped Docker services (Redpanda).
 */
async function startBackendProcess(state, configPath, ports) {
  console.log('Starting backend as host process...');

  // Rewrite the config to use localhost ports instead of Docker hostnames
  const fs = await import('node:fs');
  let config = fs.readFileSync(configPath, 'utf-8');
  config = config.replace('redpanda:9092', `localhost:${ports.redpandaKafka}`);
  config = config.replace('http://redpanda:9644', `http://localhost:${ports.redpandaAdmin}`);
  config = config.replace('http://redpanda:8081', `http://localhost:${ports.redpandaSchemaRegistry}`);
  config = config.replace('listenPort: 3000', `listenPort: ${ports.backend}`);

  // Resolve the license file path to the host filesystem
  const defaultLicensePath = resolve(
    __dirname,
    '../../../../console-enterprise/frontend/tests/config/redpanda.license'
  );
  const hostLicensePath = process.env.REDPANDA_LICENSE_PATH || defaultLicensePath;
  config = config.replace(/licenseFilepath:.*/, `licenseFilepath: ${hostLicensePath}`);

  fs.writeFileSync(configPath, config);

  // Find the enterprise backend binary or build it
  const backendDir = process.env.ENTERPRISE_BACKEND_DIR
    ? resolve(process.env.ENTERPRISE_BACKEND_DIR)
    : resolve(__dirname, '../../../../console-enterprise/backend');

  const cmdDir = join(backendDir, 'cmd');

  // Copy frontend assets into the embed directory for the Go binary
  const frontendBuildDir = resolve(__dirname, '../../build');
  const embedDir = join(backendDir, 'pkg', 'embed', 'frontend');
  if (fs.existsSync(frontendBuildDir)) {
    console.log('  Copying frontend assets for host binary...');
    await execAsync(`cp -r "${frontendBuildDir}"/* "${embedDir}"/`);
  }

  // Build the binary
  console.log(`  Building backend from ${cmdDir}...`);
  await execAsync('go build -o /tmp/console-enterprise-e2e .', { cwd: cmdDir, maxBuffer: 50 * 1024 * 1024 });
  console.log('  ✓ Backend binary built');

  // Clean up the copied frontend assets
  await execAsync(`find "${embedDir}" -mindepth 1 ! -name '.gitignore' -delete`).catch(() => {});

  // If a license is available, inject it as the REDPANDA_LICENSE env var
  // so the backend can use it without a filepath.
  let licenseEnv = {};
  if (process.env.ENTERPRISE_LICENSE_CONTENT) {
    licenseEnv.REDPANDA_LICENSE = process.env.ENTERPRISE_LICENSE_CONTENT;
  } else {
    const defaultLicensePath = resolve(
      __dirname,
      '../../../../console-enterprise/frontend/tests/config/redpanda.license'
    );
    const licensePath = process.env.REDPANDA_LICENSE_PATH || defaultLicensePath;
    if (fs.existsSync(licensePath)) {
      licenseEnv.REDPANDA_LICENSE = fs.readFileSync(licensePath, 'utf-8').trim();
    }
  }

  // Start the backend process
  const { spawn } = await import('node:child_process');
  const args = [`--config.filepath=${configPath}`];

  const proc = spawn('/tmp/console-enterprise-e2e', args, {
    stdio: ['ignore', 'pipe', 'pipe'],
    detached: true,
    env: { ...process.env, ...licenseEnv },
  });

  state.backendProcess = proc;
  state.backendPid = proc.pid;

  // Log output for debugging
  proc.stdout.on('data', (data) => {
    const line = data.toString().trim();
    if (line) console.log(`[backend] ${line}`);
  });
  proc.stderr.on('data', (data) => {
    const line = data.toString().trim();
    if (line) console.error(`[backend] ${line}`);
  });

  proc.on('exit', (code) => {
    if (code !== null && code !== 0) {
      console.error(`Backend process exited with code ${code}`);
    }
  });

  // Wait for backend to be ready
  console.log(`  Waiting for backend on port ${ports.backend}...`);
  await waitForPort(ports.backend, 60, 1000);
  console.log(`  ✓ Backend ready at http://localhost:${ports.backend}`);
}

async function cleanupOnFailure(state) {
  if (state.backendProcess) {
    console.log('Stopping backend process...');
    try { state.backendProcess.kill('SIGTERM'); } catch { /* ignore */ }
  }

  if (state.sourceBackendContainer) {
    console.log('Stopping source backend container using testcontainers API...');
    await state.sourceBackendContainer.stop().catch((error) => {
      console.log(`Failed to stop source backend container: ${error.message}`);
    });
  }

  if (state.backendContainer) {
    console.log('Stopping backend container using testcontainers API...');
    await state.backendContainer.stop().catch((error) => {
      console.log(`Failed to stop backend container: ${error.message}`);
    });
  }

  if (state.connectContainer) {
    console.log('Stopping Kafka Connect container using testcontainers API...');
    await state.connectContainer.stop().catch((error) => {
      console.log(`Failed to stop Connect container: ${error.message}`);
    });
  }
  if (state.owlshopContainer) {
    console.log('Stopping OwlShop container using testcontainers API...');
    await state.owlshopContainer.stop().catch((error) => {
      console.log(`Failed to stop OwlShop container: ${error.message}`);
    });
  }
  for (const [key, label] of [
    ['zitadelProxyContainer', 'Zitadel proxy'],
    ['zitadelLoginContainer', 'Zitadel Login'],
    ['zitadelContainer', 'Zitadel API'],
    ['zitadelDbContainer', 'Zitadel DB'],
  ]) {
    if (state[key]) {
      console.log(`Stopping ${label} container using testcontainers API...`);
      await state[key].stop().catch((error) => {
        console.log(`Failed to stop ${label} container: ${error.message}`);
      });
    }
  }
  if (state.destRedpandaContainer) {
    console.log('Stopping destination Redpanda container using testcontainers API...');
    await state.destRedpandaContainer.stop().catch((error) => {
      console.log(`Failed to stop destination Redpanda container: ${error.message}`);
    });
  }
  if (state.redpandaContainer) {
    console.log('Stopping Redpanda container using testcontainers API...');
    await state.redpandaContainer.stop().catch((error) => {
      console.log(`Failed to stop Redpanda container: ${error.message}`);
    });
  }
  if (state.network) {
    console.log('Stopping network using testcontainers API...');
    await state.network.stop().catch((error) => {
      console.log(`Failed to stop network: ${error.message}`);
    });
  }
}

export default async function globalSetup(config = {}) {
  // Get variant configuration from metadata
  const variantName = config?.metadata?.variantName ?? 'console';
  const configFile = config?.metadata?.configFile ?? 'console.config.yaml';
  const isEnterprise = config?.metadata?.isEnterprise ?? false;
  const needsShadowlink = config?.metadata?.needsShadowlink ?? false;
  const needsConnect = config?.metadata?.needsConnect ?? false;
  const needsZitadel = config?.metadata?.needsZitadel ?? false;

  // Load ports from variant's config/variant.json
  const variantConfig = loadVariantConfig(variantName);
  const ports = variantConfig.ports;

  console.log('\n\n========================================');
  console.log(`🚀 GLOBAL SETUP: ${variantName}${needsShadowlink ? ' + SHADOWLINK' : ''}${needsZitadel ? ' + ZITADEL' : ''}`);
  console.log('========================================\n');
  console.log('DEBUG - Config metadata:', {
    variantName,
    configFile,
    isEnterprise,
    needsShadowlink,
    needsZitadel,
    ports,
  });
  console.log('Starting testcontainers environment...');

  const state = {
    networkId: '',
    redpandaId: '',
    owlshopId: '',
    connectId: '',
    backendId: '',
    destRedpandaId: '',
    sourceBackendId: '',
    isEnterprise,
    needsShadowlink,
    needsZitadel,
  };

  try {
    // Use pre-built image tag if available (set by run-all-variants.mjs), otherwise build
    const prebuiltTag = isEnterprise
      ? process.env.E2E_PREBUILT_IMAGE_TAG_ENTERPRISE
      : process.env.E2E_PREBUILT_IMAGE_TAG;
    const imageTag = prebuiltTag || (await buildBackendImage(isEnterprise));

    // Setup Docker infrastructure
    const network = await setupDockerNetwork(state);

    // --- Group 1: Start clusters in parallel ---
    const clusterPromises = [
      startRedpandaContainer(network, state, ports).then(() => verifyRedpandaServices(state, ports)),
    ];
    if (isEnterprise && needsShadowlink) {
      clusterPromises.push(
        startDestinationRedpandaContainer(network, state, ports).then(() =>
          verifyDestinationRedpandaServices(state, ports)
        )
      );
    }
    await Promise.all(clusterPromises);

    console.log('');
    console.log('=== Docker Environment Ready ===');
    console.log(`  - Source Redpanda broker (external): localhost:${ports.redpandaKafka}`);
    console.log('  - Source Redpanda broker (internal): redpanda:9092');
    console.log(`  - Schema Registry: http://localhost:${ports.redpandaSchemaRegistry}`);
    console.log(`  - Admin API: http://localhost:${ports.redpandaAdmin}`);
    if (needsShadowlink) {
      console.log(`  - Destination Redpanda broker (external): localhost:${ports.destRedpandaKafka}`);
      console.log('  - Destination Redpanda broker (internal): dest-cluster:9092');
      console.log(`  - Destination Schema Registry: http://localhost:${ports.destRedpandaSchemaRegistry}`);
      console.log(`  - Destination Admin API: http://localhost:${ports.destRedpandaAdmin}`);
    }
    console.log(`  - Kafka Connect: http://localhost:${ports.kafkaConnect}`);
    console.log('================================\n');

    // --- Zitadel OIDC provider (must start before backend so config can be rewritten) ---
    let zitadelConfig = null;
    let effectiveConfigFile = configFile;
    if (needsZitadel) {
      const { startZitadel, rewriteConsoleConfig } = await import('./zitadel-setup.mjs');
      zitadelConfig = await startZitadel(network, state, ports);

      // Rewrite the Console config template with real Zitadel values
      const originalConfigPath = resolve(__dirname, '..', `test-variant-${variantName}`, 'config', configFile);
      const rewrittenConfigPath = rewriteConsoleConfig(originalConfigPath, zitadelConfig);
      // Store rewritten path for the backend to use
      state.zitadelConfig = zitadelConfig;
      state.rewrittenConfigPath = rewrittenConfigPath;
      console.log(`  ✓ Console config rewritten: ${rewrittenConfigPath}`);
    }

    // --- Group 2: Start services in parallel (all depend on Redpanda being ready) ---
    const servicePromises = [
      startOwlShop(network, state),
      createKafkaConnectTopics(state).then(() => (needsConnect ? startKafkaConnect(network, state, ports) : null)),
    ];

    // For variants that need Kafka Connect, the backend must start AFTER Connect is ready
    // (the backend proxies connector APIs and tests immediately hit those endpoints).
    // For other variants, start the backend in parallel with services.
    if (!needsConnect) {
      if (needsShadowlink) {
        const sourceBackendConfigPath = resolve(__dirname, '..', `test-variant-${variantName}`, 'config', configFile);
        const destBackendConfigPath = resolve(__dirname, 'console.dest.config.yaml');
        servicePromises.push(
          startBackendServerWithConfig(
            network,
            isEnterprise,
            imageTag,
            state,
            sourceBackendConfigPath,
            ports.backend,
            'console-backend'
          ),
          startBackendServerWithConfig(
            network,
            isEnterprise,
            imageTag,
            state,
            destBackendConfigPath,
            ports.backendDest,
            'console-backend-dest'
          )
        );
      } else if (needsZitadel && state.rewrittenConfigPath) {
        // For OIDC tests, the Console backend must reach both Zitadel (on localhost)
        // and Redpanda (on Docker network). Run the backend as a host process
        // with the config rewritten to use localhost-accessible ports for all services.
        servicePromises.push(
          startBackendProcess(state, state.rewrittenConfigPath, ports)
        );
      } else {
        servicePromises.push(
          startBackendServer(network, isEnterprise, imageTag, state, variantName, configFile, ports)
        );
      }
    }

    await Promise.all(servicePromises);

    // Start backend after Kafka Connect for connect variants
    if (needsConnect) {
      await startBackendServer(network, isEnterprise, imageTag, state, variantName, configFile, ports);
    }

    console.log('Backend is ready');

    // Give services extra time to stabilize in CI (especially shadowlink replication)
    if (isEnterprise && needsShadowlink && process.env.CI) {
      console.log('CI detected: Giving services 3 seconds to stabilize...');
      await new Promise((resolve) => setTimeout(resolve, 3000));
      console.log('✓ Stabilization period complete');
    }

    writeFileSync(getStateFile(variantName), JSON.stringify(state, null, 2));

    console.log('\n✅ All services ready! Starting tests...\n');
  } catch (error) {
    console.error('Failed to start environment:', error);
    await cleanupOnFailure(state);
    throw error;
  }
}