import type { UniDeskConfig } from "./config";
import { rootPath } from "./config";
import type { RenderedCliResult } from "./output";
import {
  capture,
  compactCapture,
  createYamlFieldReader,
  parseJsonOutput,
  readYamlRecord,
  sha256Hex,
  shQuote,
} from "./platform-infra-ops-library";

const configFile = rootPath("config", "platform-infra", "kafka.yaml");
const configLabel = "config/platform-infra/kafka.yaml";
const fieldManager = "unidesk-platform-infra-kafka";
const y = createYamlFieldReader(configLabel);

interface KafkaTarget {
  id: string;
  route: string;
  namespace: string;
  role: "active" | "standby";
  enabled: boolean;
  createNamespace: boolean;
  storageClassName: string;
}

interface KafkaTopicSpec {
  name: string;
  partitions: number;
  replicas: number;
  retentionMs: number;
  cleanupPolicy: "delete" | "compact";
  description: string;
}

interface KafkaClientSpec {
  id: string;
  namespace: string;
  serviceAccountName: string;
  kafkaUserName: string;
  secretName: string;
  produceTopics: string[];
  consumeTopics: string[];
  dlqTopics: string[];
}

interface PlatformKafkaConfig {
  version: number;
  kind: "platform-infra-kafka";
  metadata: {
    id: string;
    owner: string;
    spec: string;
    relatedIssues: number[];
  };
  defaults: {
    targetId: string;
    switch: {
      enabled: boolean;
      mode: string;
      appIntegrationEnabled: boolean;
      shadowProduceEnabled: boolean;
      shadowConsumeEnabled: boolean;
    };
  };
  operator: {
    implementation: "strimzi";
    version: string;
    manifestUrl: string;
    deploymentName: string;
    serviceAccountName: string;
    crds: string[];
  };
  targets: KafkaTarget[];
  cluster: {
    name: string;
    nodePoolName: string;
    kafkaVersion: string;
    metadataVersion: string;
    replicas: number;
    storage: {
      size: string;
      deleteClaim: boolean;
    };
    listeners: {
      plain: { enabled: boolean; port: number };
      tls: { enabled: boolean; port: number };
    };
    authorization: {
      enabled: boolean;
      mode: string;
    };
  };
  topics: KafkaTopicSpec[];
  clients: KafkaClientSpec[];
  validation: {
    timeoutSeconds: number;
    pollSeconds: number;
    smokeTopic: string;
  };
  management: {
    defaultTailLimit: number;
    maxTailLimit: number;
    defaultShadowTopic: string;
    shadowProducerId: string;
  };
}

interface CommonOptions {
  targetId: string | null;
  full: boolean;
  raw: boolean;
}

interface ApplyOptions extends CommonOptions {
  dryRun: boolean;
  confirm: boolean;
  wait: boolean;
}

interface KafkaInspectOptions extends CommonOptions {
  topic: string | null;
  group: string | null;
  limit: number | null;
}

interface KafkaProduceOptions extends CommonOptions {
  topic: string | null;
  key: string | null;
  source: string;
  eventType: string;
  payload: string | null;
}

export async function runPlatformInfraKafkaCommand(config: UniDeskConfig, args: string[]): Promise<Record<string, unknown> | RenderedCliResult> {
  const [action = "plan"] = args;
  if (action === "plan") {
    const options = parseCommonOptions(args.slice(1));
    const result = plan(options);
    return options.full || options.raw ? result : renderPlan(result);
  }
  if (action === "apply") return await apply(config, parseApplyOptions(args.slice(1)));
  if (action === "status") {
    const options = parseCommonOptions(args.slice(1));
    const result = await status(config, options);
    return options.full || options.raw ? result : renderStatus(result);
  }
  if (action === "validate") return await validate(config, parseCommonOptions(args.slice(1)));
  if (action === "topics") {
    const options = parseInspectOptions(args.slice(1));
    const result = await topics(config, options);
    return options.full || options.raw ? result : renderTopics(result);
  }
  if (action === "groups") {
    const options = parseInspectOptions(args.slice(1));
    const result = await groups(config, options);
    return options.full || options.raw ? result : renderGroups(result);
  }
  if (action === "offsets") {
    const options = parseInspectOptions(args.slice(1));
    const result = await offsets(config, options);
    return options.full || options.raw ? result : renderOffsets(result);
  }
  if (action === "tail") {
    const options = parseInspectOptions(args.slice(1));
    const result = await tail(config, options);
    return options.full || options.raw ? result : renderTail(result);
  }
  if (action === "produce") return await produce(config, await parseProduceOptions(args.slice(1)));
  return {
    ok: false,
    error: "unsupported-platform-infra-kafka-command",
    args,
    help: kafkaHelp(),
  };
}

export function kafkaHelp(): Record<string, unknown> {
  return {
    command: "platform-infra kafka plan|apply|status|validate|topics|groups|offsets|tail|produce",
    configTruth: configLabel,
    usage: [
      "bun scripts/cli.ts platform-infra kafka plan --target D518",
      "bun scripts/cli.ts platform-infra kafka plan --node D518",
      "bun scripts/cli.ts platform-infra kafka apply --target D518 --dry-run",
      "bun scripts/cli.ts platform-infra kafka apply --target D518 --confirm",
      "bun scripts/cli.ts platform-infra kafka status --target D518 [--full|--raw]",
      "bun scripts/cli.ts platform-infra kafka validate --target D518 [--full|--raw]",
      "bun scripts/cli.ts platform-infra kafka topics --node D518",
      "bun scripts/cli.ts platform-infra kafka groups --node D518",
      "bun scripts/cli.ts platform-infra kafka offsets --node D518 [--topic hwlab.agentrun.command.v1]",
      "bun scripts/cli.ts platform-infra kafka tail --node D518 --topic hwlab.agentrun.command.v1 --limit 5",
      "bun scripts/cli.ts platform-infra kafka produce --node D518 --topic hwlab.agentrun.command.v1 --key <trace-or-session>",
    ],
    boundary: "Kafka runtime belongs to platform-infra. P2 supports shadow produce/query only; Kafka consumer cutover stays disabled.",
  };
}

function readKafkaConfig(): PlatformKafkaConfig {
  const root = readYamlRecord<Record<string, unknown>>(configFile, "platform-infra-kafka");
  const version = y.integerField(root, "version", "");
  if (version !== 1) throw new Error(`${configLabel}.version must be 1`);
  const metadata = y.objectField(root, "metadata", "");
  const defaults = y.objectField(root, "defaults", "");
  const switchRecord = y.objectField(defaults, "switch", "defaults");
  const operator = y.objectField(root, "operator", "");
  const cluster = y.objectField(root, "cluster", "");
  const storage = y.objectField(cluster, "storage", "cluster");
  const listeners = y.objectField(cluster, "listeners", "cluster");
  const plain = y.objectField(listeners, "plain", "cluster.listeners");
  const tls = y.objectField(listeners, "tls", "cluster.listeners");
  const authorization = y.objectField(cluster, "authorization", "cluster");
  const validation = y.objectField(root, "validation", "");
  const parsed: PlatformKafkaConfig = {
    version,
    kind: "platform-infra-kafka",
    metadata: {
      id: y.stringField(metadata, "id", "metadata"),
      owner: y.stringField(metadata, "owner", "metadata"),
      spec: y.stringField(metadata, "spec", "metadata"),
      relatedIssues: y.numberArrayField(metadata, "relatedIssues", "metadata"),
    },
    defaults: {
      targetId: y.stringField(defaults, "targetId", "defaults"),
      switch: {
        enabled: y.booleanField(switchRecord, "enabled", "defaults.switch"),
        mode: y.stringField(switchRecord, "mode", "defaults.switch"),
        appIntegrationEnabled: y.booleanField(switchRecord, "appIntegrationEnabled", "defaults.switch"),
        shadowProduceEnabled: y.booleanField(switchRecord, "shadowProduceEnabled", "defaults.switch"),
        shadowConsumeEnabled: y.booleanField(switchRecord, "shadowConsumeEnabled", "defaults.switch"),
      },
    },
    operator: {
      implementation: y.enumField(operator, "implementation", "operator", ["strimzi"] as const),
      version: y.stringField(operator, "version", "operator"),
      manifestUrl: y.httpsUrlField(operator, "manifestUrl", "operator"),
      deploymentName: y.kubernetesNameField(operator, "deploymentName", "operator"),
      serviceAccountName: y.kubernetesNameField(operator, "serviceAccountName", "operator"),
      crds: y.stringArrayField(operator, "crds", "operator"),
    },
    targets: y.arrayOfRecords(root.targets, "targets").map(parseTarget),
    cluster: {
      name: y.kubernetesNameField(cluster, "name", "cluster"),
      nodePoolName: y.kubernetesNameField(cluster, "nodePoolName", "cluster"),
      kafkaVersion: y.stringField(cluster, "kafkaVersion", "cluster"),
      metadataVersion: y.stringField(cluster, "metadataVersion", "cluster"),
      replicas: y.integerField(cluster, "replicas", "cluster"),
      storage: {
        size: y.stringField(storage, "size", "cluster.storage"),
        deleteClaim: y.booleanField(storage, "deleteClaim", "cluster.storage"),
      },
      listeners: {
        plain: { enabled: y.booleanField(plain, "enabled", "cluster.listeners.plain"), port: y.portField(plain, "port", "cluster.listeners.plain") },
        tls: { enabled: y.booleanField(tls, "enabled", "cluster.listeners.tls"), port: y.portField(tls, "port", "cluster.listeners.tls") },
      },
      authorization: {
        enabled: y.booleanField(authorization, "enabled", "cluster.authorization"),
        mode: y.stringField(authorization, "mode", "cluster.authorization"),
      },
    },
    topics: y.arrayOfRecords(root.topics, "topics").map(parseTopic),
    clients: y.arrayOfRecords(root.clients, "clients").map(parseClient),
    validation: {
      timeoutSeconds: y.integerField(validation, "timeoutSeconds", "validation"),
      pollSeconds: y.integerField(validation, "pollSeconds", "validation"),
      smokeTopic: y.stringField(validation, "smokeTopic", "validation"),
    },
    management: parseManagement(root),
  };
  validateConfig(parsed);
  return parsed;
}

function parseTarget(record: Record<string, unknown>, index: number): KafkaTarget {
  const path = `targets[${index}]`;
  return {
    id: y.stringField(record, "id", path),
    route: y.stringField(record, "route", path),
    namespace: y.kubernetesNameField(record, "namespace", path),
    role: y.enumField(record, "role", path, ["active", "standby"] as const),
    enabled: y.booleanField(record, "enabled", path),
    createNamespace: y.booleanField(record, "createNamespace", path),
    storageClassName: y.stringField(record, "storageClassName", path),
  };
}

function parseTopic(record: Record<string, unknown>, index: number): KafkaTopicSpec {
  const path = `topics[${index}]`;
  return {
    name: topicNameField(record, "name", path),
    partitions: positiveInteger(record, "partitions", path),
    replicas: positiveInteger(record, "replicas", path),
    retentionMs: positiveInteger(record, "retentionMs", path),
    cleanupPolicy: y.enumField(record, "cleanupPolicy", path, ["delete", "compact"] as const),
    description: y.stringField(record, "description", path),
  };
}

function parseClient(record: Record<string, unknown>, index: number): KafkaClientSpec {
  const path = `clients[${index}]`;
  return {
    id: y.stringField(record, "id", path),
    namespace: y.kubernetesNameField(record, "namespace", path),
    serviceAccountName: y.kubernetesNameField(record, "serviceAccountName", path),
    kafkaUserName: y.kubernetesNameField(record, "kafkaUserName", path),
    secretName: y.kubernetesNameField(record, "secretName", path),
    produceTopics: topicArray(record, "produceTopics", path),
    consumeTopics: topicArray(record, "consumeTopics", path),
    dlqTopics: topicArray(record, "dlqTopics", path),
  };
}

function parseManagement(root: Record<string, unknown>): PlatformKafkaConfig["management"] {
  const management = y.objectField(root, "management", "");
  return {
    defaultTailLimit: positiveInteger(management, "defaultTailLimit", "management"),
    maxTailLimit: positiveInteger(management, "maxTailLimit", "management"),
    defaultShadowTopic: topicNameField(management, "defaultShadowTopic", "management"),
    shadowProducerId: y.stringField(management, "shadowProducerId", "management"),
  };
}

function validateConfig(kafka: PlatformKafkaConfig): void {
  resolveTarget(kafka, kafka.defaults.targetId);
  if (!/^https:\/\/github\.com\/strimzi\/strimzi-kafka-operator\/releases\/download\//u.test(kafka.operator.manifestUrl)) {
    throw new Error(`${configLabel}.operator.manifestUrl must be an official Strimzi GitHub release URL`);
  }
  if (kafka.cluster.replicas !== 1) throw new Error(`${configLabel}.cluster.replicas must be 1 for the D518 POC`);
  if (!/^[0-9]+Gi$/u.test(kafka.cluster.storage.size)) throw new Error(`${configLabel}.cluster.storage.size must be a Gi quantity`);
  if (!kafka.topics.some((topic) => topic.name === kafka.validation.smokeTopic)) throw new Error(`${configLabel}.validation.smokeTopic must reference one of topics[].name`);
  if (!kafka.topics.some((topic) => topic.name === kafka.management.defaultShadowTopic)) throw new Error(`${configLabel}.management.defaultShadowTopic must reference one of topics[].name`);
  if (kafka.management.defaultTailLimit > kafka.management.maxTailLimit) throw new Error(`${configLabel}.management.defaultTailLimit must be <= management.maxTailLimit`);
  if (kafka.defaults.switch.shadowConsumeEnabled) throw new Error(`${configLabel}.defaults.switch.shadowConsumeEnabled must stay false for shadow-produce-only stage`);
  const topicNames = new Set<string>();
  for (const topic of kafka.topics) {
    if (topicNames.has(topic.name)) throw new Error(`${configLabel}.topics contains duplicate topic ${topic.name}`);
    topicNames.add(topic.name);
  }
  const clientIds = new Set<string>();
  for (const client of kafka.clients) {
    if (clientIds.has(client.id)) throw new Error(`${configLabel}.clients contains duplicate id ${client.id}`);
    clientIds.add(client.id);
    for (const topic of [...client.produceTopics, ...client.consumeTopics, ...client.dlqTopics]) {
      if (!topicNames.has(topic)) throw new Error(`${configLabel}.clients.${client.id} references unknown topic ${topic}`);
    }
  }
  if (kafka.validation.timeoutSeconds < 5 || kafka.validation.timeoutSeconds > 55) throw new Error(`${configLabel}.validation.timeoutSeconds must be between 5 and 55`);
  if (kafka.validation.pollSeconds < 1 || kafka.validation.pollSeconds > kafka.validation.timeoutSeconds) throw new Error(`${configLabel}.validation.pollSeconds must be between 1 and timeoutSeconds`);
}

function resolveTarget(kafka: PlatformKafkaConfig, targetId: string | null): KafkaTarget {
  const resolved = targetId ?? kafka.defaults.targetId;
  const target = kafka.targets.find((item) => item.id.toLowerCase() === resolved.toLowerCase());
  if (target === undefined) throw new Error(`unknown kafka target ${resolved}; known targets: ${kafka.targets.map((item) => item.id).join(", ")}`);
  if (!target.enabled) throw new Error(`kafka target ${target.id} is disabled in ${configLabel}`);
  return target;
}

function plan(options: CommonOptions): Record<string, unknown> {
  const kafka = readKafkaConfig();
  const target = resolveTarget(kafka, options.targetId);
  const manifest = renderKafkaManifest(kafka, target);
  const policy = policyChecks(kafka, target, manifest);
  return {
    ok: policy.every((check) => check.ok),
    action: "platform-infra-kafka-plan",
    mutation: false,
    config: configSummary(kafka, target),
    renderPlan: {
      target: targetSummary(target),
      objects: manifestObjectSummary(manifest),
      operator: operatorSummary(kafka, target),
      cluster: clusterSummary(kafka, target),
      topics: kafka.topics.map((topic) => topicSummary(topic)),
      clients: kafka.clients.map((client) => clientSummary(client)),
    },
    policy,
    next: {
      dryRun: `bun scripts/cli.ts platform-infra kafka apply --target ${target.id} --dry-run`,
      apply: `bun scripts/cli.ts platform-infra kafka apply --target ${target.id} --confirm`,
      status: `bun scripts/cli.ts platform-infra kafka status --target ${target.id}`,
      validate: `bun scripts/cli.ts platform-infra kafka validate --target ${target.id}`,
    },
  };
}

async function apply(config: UniDeskConfig, options: ApplyOptions): Promise<Record<string, unknown>> {
  const kafka = readKafkaConfig();
  const target = resolveTarget(kafka, options.targetId);
  const manifest = renderKafkaManifest(kafka, target);
  const policy = policyChecks(kafka, target, manifest);
  if (!policy.every((check) => check.ok)) return { ok: false, action: "platform-infra-kafka-apply", mode: "policy-blocked", mutation: false, policy };
  const operator = await fetchOperatorManifest(kafka, target);
  if (operator.ok !== true) {
    return {
      ok: false,
      action: "platform-infra-kafka-apply",
      mode: "operator-manifest-fetch-failed",
      mutation: false,
      target: targetSummary(target),
      config: compactConfigSummary(kafka, target),
      policy,
      operator,
    };
  }
  const result = await capture(config, target.route, ["sh"], applyScript(kafka, target, manifest, operator, options));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed?.ok === true,
    action: "platform-infra-kafka-apply",
    mode: options.dryRun ? "dry-run" : "confirmed",
    mutation: !options.dryRun,
    target: targetSummary(target),
    config: compactConfigSummary(kafka, target),
    policy,
    remote: parsed ?? compactCapture(result, { full: true }),
    next: {
      status: `bun scripts/cli.ts platform-infra kafka status --target ${target.id}`,
      validate: `bun scripts/cli.ts platform-infra kafka validate --target ${target.id}`,
    },
  };
}

async function status(config: UniDeskConfig, options: CommonOptions): Promise<Record<string, unknown>> {
  const kafka = readKafkaConfig();
  const target = resolveTarget(kafka, options.targetId);
  const result = await capture(config, target.route, ["sh"], statusScript(kafka, target, options.full));
  const parsed = parseJsonOutput(result.stdout);
  const summary = parsed === null ? null : statusSummary(parsed);
  return {
    ok: result.exitCode === 0 && summary?.ready === true,
    action: "platform-infra-kafka-status",
    mutation: false,
    target: targetSummary(target),
    config: configSummary(kafka, target),
    summary,
    remote: options.raw ? parsed : options.full ? parsed : summary ?? compactCapture(result, { full: true }),
    next: {
      plan: `bun scripts/cli.ts platform-infra kafka plan --target ${target.id}`,
      apply: `bun scripts/cli.ts platform-infra kafka apply --target ${target.id} --confirm`,
      validate: `bun scripts/cli.ts platform-infra kafka validate --target ${target.id}`,
    },
  };
}

async function validate(config: UniDeskConfig, options: CommonOptions): Promise<Record<string, unknown>> {
  const kafka = readKafkaConfig();
  const target = resolveTarget(kafka, options.targetId);
  const result = await capture(config, target.route, ["sh"], validateScript(kafka, target, options.full));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed?.ok === true,
    action: "platform-infra-kafka-validate",
    mutation: true,
    target: targetSummary(target),
    config: compactConfigSummary(kafka, target),
    validation: parsed ?? null,
    remote: options.raw && parsed !== null ? parsed : compactCapture(result, { full: options.full || result.exitCode !== 0 }),
  };
}

async function topics(config: UniDeskConfig, options: KafkaInspectOptions): Promise<Record<string, unknown>> {
  const kafka = readKafkaConfig();
  const target = resolveTarget(kafka, options.targetId);
  const result = await capture(config, target.route, ["sh"], topicsScript(kafka, target));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed?.ok === true,
    action: "platform-infra-kafka-topics",
    mutation: false,
    target: targetSummary(target),
    config: compactConfigSummary(kafka, target),
    query: parsed ?? null,
    remote: options.raw && parsed !== null ? parsed : compactCapture(result, { full: options.full || result.exitCode !== 0 }),
  };
}

async function groups(config: UniDeskConfig, options: KafkaInspectOptions): Promise<Record<string, unknown>> {
  const kafka = readKafkaConfig();
  const target = resolveTarget(kafka, options.targetId);
  const result = await capture(config, target.route, ["sh"], groupsScript(kafka, target));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed?.ok === true,
    action: "platform-infra-kafka-groups",
    mutation: false,
    target: targetSummary(target),
    config: compactConfigSummary(kafka, target),
    query: parsed ?? null,
    remote: options.raw && parsed !== null ? parsed : compactCapture(result, { full: options.full || result.exitCode !== 0 }),
  };
}

async function offsets(config: UniDeskConfig, options: KafkaInspectOptions): Promise<Record<string, unknown>> {
  const kafka = readKafkaConfig();
  const target = resolveTarget(kafka, options.targetId);
  const topic = resolveTopic(kafka, options.topic, null);
  const result = await capture(config, target.route, ["sh"], offsetsScript(kafka, target, topic, options.group));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed?.ok === true,
    action: "platform-infra-kafka-offsets",
    mutation: false,
    target: targetSummary(target),
    config: compactConfigSummary(kafka, target),
    query: parsed ?? null,
    remote: options.raw && parsed !== null ? parsed : compactCapture(result, { full: options.full || result.exitCode !== 0 }),
  };
}

async function tail(config: UniDeskConfig, options: KafkaInspectOptions): Promise<Record<string, unknown>> {
  const kafka = readKafkaConfig();
  const target = resolveTarget(kafka, options.targetId);
  const topic = resolveTopic(kafka, options.topic, kafka.management.defaultShadowTopic);
  const limit = boundedLimit(kafka, options.limit);
  const result = await capture(config, target.route, ["sh"], tailScript(kafka, target, topic, limit));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed?.ok === true,
    action: "platform-infra-kafka-tail",
    mutation: false,
    target: targetSummary(target),
    config: compactConfigSummary(kafka, target),
    query: parsed ?? null,
    remote: options.raw && parsed !== null ? parsed : compactCapture(result, { full: options.full || result.exitCode !== 0 }),
  };
}

async function produce(config: UniDeskConfig, options: KafkaProduceOptions): Promise<Record<string, unknown>> {
  const kafka = readKafkaConfig();
  const target = resolveTarget(kafka, options.targetId);
  const topic = resolveTopic(kafka, options.topic, kafka.management.defaultShadowTopic);
  const payload = buildShadowPayload(kafka, target, options, topic);
  const result = await capture(config, target.route, ["sh"], produceScript(kafka, target, topic, options.key, payload));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed?.ok === true,
    action: "platform-infra-kafka-produce",
    mutation: true,
    mode: "shadow-produce-only",
    target: targetSummary(target),
    config: compactConfigSummary(kafka, target),
    production: parsed ?? null,
    remote: options.raw ? parsed : compactCapture(result, { full: options.full || result.exitCode !== 0 }),
  };
}

function renderKafkaManifest(kafka: PlatformKafkaConfig, target: KafkaTarget): string {
  const listeners = [
    kafka.cluster.listeners.plain.enabled
      ? `      - name: plain
        port: ${kafka.cluster.listeners.plain.port}
        type: internal
        tls: false`
      : "",
    kafka.cluster.listeners.tls.enabled
      ? `      - name: tls
        port: ${kafka.cluster.listeners.tls.port}
        type: internal
        tls: true`
      : "",
  ].filter(Boolean).join("\n");
  const topicDocs = kafka.topics.map((topic) => kafkaTopicManifest(kafka, target, topic)).join("---\n");
  const userDocs = kafka.clients.map((client) => kafkaUserManifest(kafka, target, client)).join("---\n");
  return `apiVersion: v1
kind: Namespace
metadata:
  name: ${target.namespace}
  labels:
    app.kubernetes.io/name: platform-infra
    app.kubernetes.io/managed-by: unidesk
    unidesk.ai/runtime-node: ${target.id}
---
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: allow-all
  namespace: ${target.namespace}
  labels:
    app.kubernetes.io/name: platform-infra
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
spec:
  podSelector: {}
  policyTypes:
    - Ingress
    - Egress
  ingress:
    - {}
  egress:
    - {}
---
apiVersion: kafka.strimzi.io/v1
kind: KafkaNodePool
metadata:
  name: ${kafka.cluster.nodePoolName}
  namespace: ${target.namespace}
  labels:
    strimzi.io/cluster: ${kafka.cluster.name}
    app.kubernetes.io/name: ${kafka.cluster.name}
    app.kubernetes.io/component: kafka-node-pool
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
spec:
  replicas: ${kafka.cluster.replicas}
  roles:
    - controller
    - broker
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: ${kafka.cluster.storage.size}
        class: ${target.storageClassName}
        deleteClaim: ${kafka.cluster.storage.deleteClaim}
        kraftMetadata: shared
---
apiVersion: kafka.strimzi.io/v1
kind: Kafka
metadata:
  name: ${kafka.cluster.name}
  namespace: ${target.namespace}
  labels:
    app.kubernetes.io/name: ${kafka.cluster.name}
    app.kubernetes.io/component: kafka
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
    unidesk.ai/runtime-node: ${target.id}
  annotations:
    unidesk.ai/spec: ${kafka.metadata.spec}
    unidesk.ai/app-integration-enabled: "${kafka.defaults.switch.appIntegrationEnabled}"
spec:
  kafka:
    version: ${kafka.cluster.kafkaVersion}
    metadataVersion: ${kafka.cluster.metadataVersion}
    listeners:
${listeners}
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      min.insync.replicas: 1
      auto.create.topics.enable: false
  entityOperator:
    topicOperator: {}
    userOperator: {}
---
${topicDocs}---\n${userDocs}`;
}

function kafkaTopicManifest(kafka: PlatformKafkaConfig, target: KafkaTarget, topic: KafkaTopicSpec): string {
  return `apiVersion: kafka.strimzi.io/v1
kind: KafkaTopic
metadata:
  name: ${topicResourceName(topic.name)}
  namespace: ${target.namespace}
  labels:
    strimzi.io/cluster: ${kafka.cluster.name}
    app.kubernetes.io/name: ${topicResourceName(topic.name)}
    app.kubernetes.io/component: kafka-topic
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
  annotations:
    unidesk.ai/topic-name: ${topic.name}
    unidesk.ai/description: ${quoteAnnotation(topic.description)}
spec:
  topicName: ${topic.name}
  partitions: ${topic.partitions}
  replicas: ${topic.replicas}
  config:
    retention.ms: ${topic.retentionMs}
    cleanup.policy: ${topic.cleanupPolicy}
`;
}

function kafkaUserManifest(kafka: PlatformKafkaConfig, target: KafkaTarget, client: KafkaClientSpec): string {
  const authorization = kafka.cluster.authorization.enabled
    ? `  authorization:
    type: simple
    acls:
${clientAcls(client)}`
    : "";
  return `apiVersion: kafka.strimzi.io/v1
kind: KafkaUser
metadata:
  name: ${client.kafkaUserName}
  namespace: ${target.namespace}
  labels:
    strimzi.io/cluster: ${kafka.cluster.name}
    app.kubernetes.io/name: ${client.kafkaUserName}
    app.kubernetes.io/component: kafka-client
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
  annotations:
    unidesk.ai/client-id: ${client.id}
    unidesk.ai/client-namespace: ${client.namespace}
    unidesk.ai/client-service-account: ${client.serviceAccountName}
    unidesk.ai/secret-target: ${client.namespace}/${client.secretName}
spec:
  authentication:
    type: tls
${authorization}`;
}

function clientAcls(client: KafkaClientSpec): string {
  const lines: string[] = [];
  for (const topic of client.produceTopics) {
    lines.push(`      - resource:
          type: topic
          name: ${topic}
          patternType: literal
        operations:
          - Write
          - Describe`);
  }
  for (const topic of client.consumeTopics) {
    lines.push(`      - resource:
          type: topic
          name: ${topic}
          patternType: literal
        operations:
          - Read
          - Describe`);
  }
  for (const topic of client.dlqTopics) {
    lines.push(`      - resource:
          type: topic
          name: ${topic}
          patternType: literal
        operations:
          - Write
          - Describe`);
  }
  lines.push(`      - resource:
          type: group
          name: ${client.id}
          patternType: prefix
        operations:
          - Read`);
  return lines.join("\n");
}

async function fetchOperatorManifest(kafka: PlatformKafkaConfig, target: KafkaTarget): Promise<Record<string, unknown> & { ok: boolean; manifest?: string }> {
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), 30_000);
  try {
    const response = await fetch(kafka.operator.manifestUrl, { signal: controller.signal });
    if (!response.ok) {
      return {
        ok: false,
        url: kafka.operator.manifestUrl,
        status: response.status,
        statusText: response.statusText,
      };
    }
    const source = await response.text();
    const manifest = source.replaceAll("namespace: myproject", `namespace: ${target.namespace}`);
    if (!manifest.includes("kind: CustomResourceDefinition") || !manifest.includes(`namespace: ${target.namespace}`)) {
      return {
        ok: false,
        url: kafka.operator.manifestUrl,
        error: "unexpected-strimzi-operator-manifest-shape",
        bytes: Buffer.byteLength(source, "utf8"),
      };
    }
    return {
      ok: true,
      url: kafka.operator.manifestUrl,
      bytes: Buffer.byteLength(source, "utf8"),
      sha256: sha256Hex(source),
      namespaceRewrite: target.namespace,
      manifest,
    };
  } catch (error) {
    return {
      ok: false,
      url: kafka.operator.manifestUrl,
      error: error instanceof Error ? error.message : String(error),
    };
  } finally {
    clearTimeout(timer);
  }
}

function applyScript(kafka: PlatformKafkaConfig, target: KafkaTarget, manifest: string, operator: Record<string, unknown> & { manifest?: string }, options: ApplyOptions): string {
  const dryRun = options.dryRun;
  const mode = dryRun ? "dry-run" : "confirmed";
  const dryServer = dryRun ? "--dry-run=server " : "";
  const stepTimeoutSeconds = Math.max(5, Math.min(kafka.validation.timeoutSeconds, 20));
  const checkTimeoutSeconds = Math.max(5, Math.min(kafka.validation.pollSeconds, 10));
  const operatorManifest = operator.manifest ?? "";
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
operator="$tmp/strimzi-operator.yaml"
manifest="$tmp/kafka.k8s.yaml"
cat >"$operator" <<'UNIDESK_STRIMZI_OPERATOR_YAML'
${operatorManifest}
UNIDESK_STRIMZI_OPERATOR_YAML
cat >"$manifest" <<'UNIDESK_KAFKA_RUNTIME_YAML'
${manifest}
UNIDESK_KAFKA_RUNTIME_YAML

kubectl create namespace ${target.namespace} --dry-run=client -o yaml >"$tmp/ns.yaml" 2>"$tmp/ns-create.err"
ns_create_rc=$?
if [ "$ns_create_rc" -eq 0 ]; then
  timeout ${stepTimeoutSeconds} kubectl apply --server-side ${dryServer}--force-conflicts --field-manager=${fieldManager} -f "$tmp/ns.yaml" >"$tmp/ns.out" 2>"$tmp/ns.err"
  ns_rc=$?
else
  ns_rc=1
  : >"$tmp/ns.out"
  cp "$tmp/ns-create.err" "$tmp/ns.err"
fi

if [ "$ns_rc" -eq 0 ]; then
  timeout ${stepTimeoutSeconds} kubectl -n ${target.namespace} apply --server-side ${dryServer}--force-conflicts --validate=false --field-manager=${fieldManager} -f "$operator" >"$tmp/operator.out" 2>"$tmp/operator.err"
  operator_rc=$?
else
  operator_rc=1
  printf '%s\\n' 'namespace apply failed; operator skipped' >"$tmp/operator.err"
  : >"$tmp/operator.out"
fi

if [ "$operator_rc" -eq 0 ]; then
  timeout ${checkTimeoutSeconds} kubectl get ${kafka.operator.crds.map((name) => `crd/${name}`).join(" ")} >"$tmp/crd-wait.out" 2>"$tmp/crd-wait.err"
  crd_ready_rc=$?
else
  crd_ready_rc=1
  printf '%s\\n' 'operator apply failed; CRD check skipped' >"$tmp/crd-wait.err"
  : >"$tmp/crd-wait.out"
fi

if [ "$operator_rc" -eq 0 ] && [ "$crd_ready_rc" -eq 0 ]; then
  timeout ${stepTimeoutSeconds} kubectl -n ${target.namespace} apply --server-side ${dryServer}--force-conflicts --validate=false --field-manager=${fieldManager} -f "$manifest" >"$tmp/kafka.out" 2>"$tmp/kafka.err"
  kafka_rc=$?
elif [ "$operator_rc" -eq 0 ] && [ "${dryRun ? "1" : "0"}" = "1" ]; then
  kafka_rc=0
  printf '%s\\n' 'dry-run: Kafka CR server validation skipped because Strimzi CRDs are not installed yet; operator dry-run succeeded.' >"$tmp/kafka.out"
  cp "$tmp/crd-wait.err" "$tmp/kafka.err"
else
  kafka_rc=1
  printf '%s\\n' 'Kafka CR apply skipped because Strimzi CRDs are not ready.' >"$tmp/kafka.err"
  : >"$tmp/kafka.out"
fi
python3 - "$tmp" "$ns_rc" "$operator_rc" "$crd_ready_rc" "$kafka_rc" <<'PY'
import json, os, sys
tmp = sys.argv[1]
ns_rc, operator_rc, crd_ready_rc, kafka_rc = [int(value) for value in sys.argv[2:6]]
def text(name, limit=1800):
    try:
        return open(os.path.join(tmp, name), encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
kafka_skipped_for_missing_crds = operator_rc == 0 and crd_ready_rc != 0 and kafka_rc == 0
payload = {
    "ok": ${dryRun ? "ns_rc == 0 and operator_rc == 0 and kafka_rc == 0" : "ns_rc == 0 and operator_rc == 0 and crd_ready_rc == 0 and kafka_rc == 0"},
    "target": "${target.id}",
    "route": "${target.route}",
    "namespace": "${target.namespace}",
    "mode": "${mode}",
    "mutation": ${dryRun ? "False" : "True"},
    "operator": {
        "implementation": "${kafka.operator.implementation}",
        "version": "${kafka.operator.version}",
        "manifestUrl": "${kafka.operator.manifestUrl}",
        "manifestSha256": "${stringValue(operator.sha256)}",
        "manifestBytes": ${Number(operator.bytes ?? 0)},
        "namespaceRewrite": "${target.namespace}",
    },
    "cluster": {
        "name": "${kafka.cluster.name}",
        "bootstrap": "${bootstrapService(kafka, target)}",
        "storage": "${kafka.cluster.storage.size}",
        "replicas": ${kafka.cluster.replicas},
    },
    "steps": {
        "namespace": {"exitCode": ns_rc, "stdoutTail": text("ns.out"), "stderrTail": text("ns.err")},
        "operatorApply": {"exitCode": operator_rc, "stdoutTail": text("operator.out"), "stderrTail": text("operator.err")},
        "crdPresent": {"exitCode": crd_ready_rc, "stdoutTail": text("crd-wait.out"), "stderrTail": text("crd-wait.err")},
        "kafkaApply": {"exitCode": kafka_rc, "skippedForMissingCrds": kafka_skipped_for_missing_crds, "stdoutTail": text("kafka.out"), "stderrTail": text("kafka.err")},
    },
    "valuesPrinted": False,
    "next": {
        "status": "bun scripts/cli.ts platform-infra kafka status --target ${target.id}",
        "validate": "bun scripts/cli.ts platform-infra kafka validate --target ${target.id}"
    }
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}

function statusScript(kafka: PlatformKafkaConfig, target: KafkaTarget, full: boolean): string {
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
capture_json() {
  name="$1"
  shift
  "$@" -o json >"$tmp/$name.json" 2>"$tmp/$name.err"
  rc=$?
  printf '%s' "$rc" >"$tmp/$name.rc"
}
capture_json ns kubectl get namespace ${target.namespace}
capture_json crds kubectl get ${kafka.operator.crds.map((name) => `crd/${name}`).join(" ")}
capture_json operator kubectl -n ${target.namespace} get deployment ${kafka.operator.deploymentName}
capture_json kafka kubectl -n ${target.namespace} get kafka.kafka.strimzi.io ${kafka.cluster.name}
capture_json nodepool kubectl -n ${target.namespace} get kafkanodepool.kafka.strimzi.io ${kafka.cluster.nodePoolName}
capture_json topics kubectl -n ${target.namespace} get kafkatopic.kafka.strimzi.io -l strimzi.io/cluster=${kafka.cluster.name}
capture_json users kubectl -n ${target.namespace} get kafkauser.kafka.strimzi.io -l strimzi.io/cluster=${kafka.cluster.name}
capture_json pods kubectl -n ${target.namespace} get pods -l strimzi.io/cluster=${kafka.cluster.name}
capture_json services kubectl -n ${target.namespace} get service -l strimzi.io/cluster=${kafka.cluster.name}
capture_json secrets kubectl -n ${target.namespace} get secret ${kafka.clients.map((client) => client.secretName).join(" ")}
capture_json events kubectl -n ${target.namespace} get events --sort-by=.lastTimestamp
python3 - "$tmp" <<'PY'
import json, os, sys
tmp = sys.argv[1]
expected_topics = ${JSON.stringify(kafka.topics.map((topic) => topic.name))}
expected_clients = ${JSON.stringify(kafka.clients.map((client) => client.kafkaUserName))}
def rc(name):
    try:
        return int(open(os.path.join(tmp, f"{name}.rc"), encoding="utf-8").read() or "1")
    except FileNotFoundError:
        return 1
def load(name):
    try:
        return json.load(open(os.path.join(tmp, f"{name}.json"), encoding="utf-8"))
    except Exception:
        return None
def items(name):
    data = load(name)
    if isinstance(data, dict) and isinstance(data.get("items"), list):
        return data["items"]
    if isinstance(data, dict) and data.get("kind") != "Status":
        return [data]
    return []
def meta_name(item):
    return ((item or {}).get("metadata") or {}).get("name")
def conditions(item):
    return (((item or {}).get("status") or {}).get("conditions") or [])
def condition_ready(item):
    return any(c.get("type") == "Ready" and c.get("status") == "True" for c in conditions(item))
def deployment_ready(item):
    spec = (item or {}).get("spec") or {}
    status = (item or {}).get("status") or {}
    desired = spec.get("replicas", 1)
    available = status.get("availableReplicas", 0)
    return {"name": meta_name(item), "desired": desired, "available": available, "ready": available >= desired}
def pod_summary(item):
    status = (item or {}).get("status") or {}
    return {"name": meta_name(item), "phase": status.get("phase"), "ready": any(c.get("type") == "Ready" and c.get("status") == "True" for c in status.get("conditions", []))}
def service_summary(item):
    spec = (item or {}).get("spec") or {}
    return {"name": meta_name(item), "type": spec.get("type"), "clusterIP": spec.get("clusterIP"), "ports": [{"name": p.get("name"), "port": p.get("port")} for p in spec.get("ports", [])]}
def topic_summary(item):
    spec = (item or {}).get("spec") or {}
    return {"resource": meta_name(item), "topicName": spec.get("topicName"), "ready": condition_ready(item)}
def user_summary(item):
    status = (item or {}).get("status") or {}
    return {"name": meta_name(item), "ready": condition_ready(item), "secret": status.get("secret")}
operator = deployment_ready(load("operator") or {})
kafka_obj = load("kafka") or {}
topics = [topic_summary(item) for item in items("topics")]
users = [user_summary(item) for item in items("users")]
for name in expected_topics:
    if not any(item["topicName"] == name for item in topics):
        topics.append({"resource": "missing", "topicName": name, "ready": False})
for name in expected_clients:
    if not any(item["name"] == name for item in users):
        users.append({"name": name, "ready": False, "secret": None})
topic_ready = all(any(item["topicName"] == name and item["ready"] for item in topics) for name in expected_topics)
user_ready = all(any(item["name"] == name and item["ready"] for item in users) for name in expected_clients)
secrets = items("secrets")
payload = {
    "ready": rc("ns") == 0 and rc("crds") == 0 and operator["ready"] and condition_ready(kafka_obj) and topic_ready and user_ready,
    "target": "${target.id}",
    "route": "${target.route}",
    "namespace": "${target.namespace}",
    "operator": operator,
    "cluster": {
        "name": "${kafka.cluster.name}",
        "ready": condition_ready(kafka_obj),
        "conditions": conditions(kafka_obj),
        "bootstrap": "${bootstrapService(kafka, target)}",
        "plain": "${kafka.cluster.name}-kafka-bootstrap.${target.namespace}.svc.cluster.local:${kafka.cluster.listeners.plain.port}",
        "tls": "${kafka.cluster.name}-kafka-bootstrap.${target.namespace}.svc.cluster.local:${kafka.cluster.listeners.tls.port}",
    },
    "crdsReady": rc("crds") == 0,
    "nodePoolReady": condition_ready(load("nodepool") or {}),
    "topicReady": topic_ready,
    "userReady": user_ready,
    "topics": topics,
    "users": users,
    "pods": [pod_summary(item) for item in items("pods")],
    "services": [service_summary(item) for item in items("services")],
    "clientSecrets": [{"name": meta_name(item), "keys": sorted(((item or {}).get("data") or {}).keys()), "valuesPrinted": False} for item in secrets],
    "valuesPrinted": False,
}
if ${full ? "False" : "True"}:
    payload["cluster"]["conditions"] = [{"type": c.get("type"), "status": c.get("status"), "reason": c.get("reason")} for c in payload["cluster"]["conditions"]]
print(json.dumps(payload, ensure_ascii=False, indent=2))
sys.exit(0 if payload["ready"] else 1)
PY
`;
}

function validateScript(kafka: PlatformKafkaConfig, target: KafkaTarget, full: boolean): string {
  const topic = kafka.validation.smokeTopic;
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
pod="$(kubectl -n ${target.namespace} get pod -l strimzi.io/cluster=${kafka.cluster.name} -o jsonpath='{.items[0].metadata.name}' 2>"$tmp/pod.err" || true)"
payload="unidesk-kafka-smoke-${target.id}-$(date +%s)-$$"
printf '%s' "$payload" >"$tmp/payload.txt"
if [ -n "$pod" ]; then
  printf '%s\\n' "$payload" | kubectl -n ${target.namespace} exec -i "$pod" -- bin/kafka-console-producer.sh --bootstrap-server ${kafka.cluster.name}-kafka-bootstrap:${kafka.cluster.listeners.plain.port} --topic ${topic} --command-property acks=all >"$tmp/produce.out" 2>"$tmp/produce.err"
  produce_rc=$?
  if [ "$produce_rc" -eq 0 ]; then
    timeout ${kafka.validation.timeoutSeconds} kubectl -n ${target.namespace} exec "$pod" -- bin/kafka-console-consumer.sh --bootstrap-server ${kafka.cluster.name}-kafka-bootstrap:${kafka.cluster.listeners.plain.port} --topic ${topic} --from-beginning --timeout-ms ${kafka.validation.timeoutSeconds * 1000} >"$tmp/consume.raw" 2>"$tmp/consume.err" || true
    grep -F "$payload" "$tmp/consume.raw" >"$tmp/consume.out" 2>>"$tmp/consume.err"
    consume_rc=$?
  else
    consume_rc=1
    printf '%s\\n' 'producer failed; consumer skipped' >"$tmp/consume.err"
  fi
else
  produce_rc=1
  consume_rc=1
  printf '%s\\n' 'kafka pod not found' >"$tmp/produce.err"
  printf '%s\\n' 'kafka pod not found' >"$tmp/consume.err"
fi
python3 - "$tmp" "$pod" "$payload" "$produce_rc" "$consume_rc" <<'PY'
import hashlib, json, os, sys
tmp, pod, payload = sys.argv[1], sys.argv[2], sys.argv[3]
produce_rc, consume_rc = int(sys.argv[4]), int(sys.argv[5])
def text(name, limit=1200):
    try:
        return open(os.path.join(tmp, name), encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
payload_obj = {
    "ok": produce_rc == 0 and consume_rc == 0,
    "target": "${target.id}",
    "namespace": "${target.namespace}",
    "cluster": "${kafka.cluster.name}",
    "topic": "${topic}",
    "pod": pod or None,
    "message": {
        "sha256": hashlib.sha256(payload.encode()).hexdigest(),
        "matched": consume_rc == 0,
        "valuesPrinted": False
    },
    "steps": {
        "produce": {"exitCode": produce_rc, "stdoutTail": text("produce.out"), "stderrTail": text("produce.err")},
        "consume": {
            "exitCode": consume_rc,
            "stdoutTail": "payload matched; stdout redacted" if consume_rc == 0 else "payload not found; stdout redacted",
            "stderrTail": text("consume.err"),
        },
    },
    "valuesPrinted": False,
}
print(json.dumps(payload_obj, ensure_ascii=False, indent=2))
sys.exit(0 if payload_obj["ok"] else 1)
PY
`;
}

function topicsScript(kafka: PlatformKafkaConfig, target: KafkaTarget): string {
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
pod="$(kubectl -n ${target.namespace} get pod -l strimzi.io/cluster=${kafka.cluster.name},strimzi.io/kind=Kafka -o jsonpath='{.items[0].metadata.name}' 2>"$tmp/pod.err" || true)"
if [ -n "$pod" ]; then
  kubectl -n ${target.namespace} exec "$pod" -- bin/kafka-topics.sh --bootstrap-server ${brokerBootstrap(kafka)} --describe >"$tmp/topics.out" 2>"$tmp/topics.err"
  topics_rc=$?
else
  topics_rc=1
  printf '%s\\n' 'kafka broker pod not found' >"$tmp/topics.err"
  : >"$tmp/topics.out"
fi
python3 - "$tmp" "$pod" "$topics_rc" <<'PY'
import json, os, re, sys
tmp, pod, rc_text = sys.argv[1], sys.argv[2], sys.argv[3]
configured = ${JSON.stringify(kafka.topics.map((topic) => topic.name))}
def text(name, limit=1600):
    try:
        return open(os.path.join(tmp, name), encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
topics = {}
for raw in text("topics.out", 200000).splitlines():
    line = raw.strip()
    if not line.startswith("Topic:"):
        continue
    parts = [part.strip() for part in line.split("\\t") if part.strip()]
    data = {}
    for part in parts:
        if ":" in part:
            key, value = part.split(":", 1)
            data[key.strip()] = value.strip()
    name = data.get("Topic")
    if not name or name == "__consumer_offsets":
        continue
    topic = topics.setdefault(name, {"name": name, "partitions": [], "configured": name in configured})
    if "PartitionCount" in data:
        topic["partitionCount"] = int(data.get("PartitionCount") or 0)
        topic["replicationFactor"] = int(data.get("ReplicationFactor") or 0)
        topic["configs"] = data.get("Configs") or ""
    if "Partition" in data:
        topic["partitions"].append({
            "partition": int(data.get("Partition") or 0),
            "leader": data.get("Leader"),
            "replicas": [item for item in re.split(r",\\s*", data.get("Replicas") or "") if item],
            "isr": [item for item in re.split(r",\\s*", data.get("Isr") or "") if item],
        })
for name in configured:
    topics.setdefault(name, {"name": name, "partitions": [], "configured": True, "missing": True})
payload = {
    "ok": int(rc_text) == 0,
    "target": "${target.id}",
    "namespace": "${target.namespace}",
    "cluster": "${kafka.cluster.name}",
    "pod": pod or None,
    "topics": sorted(topics.values(), key=lambda item: item["name"]),
    "valuesPrinted": False,
    "stderrTail": text("topics.err"),
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}

function groupsScript(kafka: PlatformKafkaConfig, target: KafkaTarget): string {
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
pod="$(kubectl -n ${target.namespace} get pod -l strimzi.io/cluster=${kafka.cluster.name},strimzi.io/kind=Kafka -o jsonpath='{.items[0].metadata.name}' 2>"$tmp/pod.err" || true)"
if [ -n "$pod" ]; then
  kubectl -n ${target.namespace} exec "$pod" -- bin/kafka-consumer-groups.sh --bootstrap-server ${brokerBootstrap(kafka)} --list >"$tmp/groups.out" 2>"$tmp/groups.err"
  groups_rc=$?
else
  groups_rc=1
  printf '%s\\n' 'kafka broker pod not found' >"$tmp/groups.err"
  : >"$tmp/groups.out"
fi
python3 - "$tmp" "$pod" "$groups_rc" <<'PY'
import json, os, sys
tmp, pod, rc_text = sys.argv[1], sys.argv[2], sys.argv[3]
def text(name, limit=1600):
    try:
        return open(os.path.join(tmp, name), encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
groups = [line.strip() for line in text("groups.out", 200000).splitlines() if line.strip()]
payload = {
    "ok": int(rc_text) == 0,
    "target": "${target.id}",
    "namespace": "${target.namespace}",
    "cluster": "${kafka.cluster.name}",
    "pod": pod or None,
    "groups": sorted(groups),
    "valuesPrinted": False,
    "stderrTail": text("groups.err"),
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}

function offsetsScript(kafka: PlatformKafkaConfig, target: KafkaTarget, topic: string | null, group: string | null): string {
  const topics = topic === null ? kafka.topics.map((item) => item.name) : [topic];
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
pod="$(kubectl -n ${target.namespace} get pod -l strimzi.io/cluster=${kafka.cluster.name},strimzi.io/kind=Kafka -o jsonpath='{.items[0].metadata.name}' 2>"$tmp/pod.err" || true)"
topics='${JSON.stringify(topics)}'
if [ -n "$pod" ]; then
  python3 - "$topics" <<'PY' >"$tmp/topic-list"
import json, sys
for item in json.loads(sys.argv[1]):
    print(item)
PY
  offsets_rc=0
  : >"$tmp/offsets.out"
  : >"$tmp/offsets.err"
  while IFS= read -r topic_name; do
    kubectl -n ${target.namespace} exec "$pod" -- bin/kafka-get-offsets.sh --bootstrap-server ${brokerBootstrap(kafka)} --topic "$topic_name" >>"$tmp/offsets.out" 2>>"$tmp/offsets.err" || offsets_rc=$?
  done <"$tmp/topic-list"
  if [ ${group === null ? "0" : "1"} -eq 1 ]; then
    kubectl -n ${target.namespace} exec "$pod" -- bin/kafka-consumer-groups.sh --bootstrap-server ${brokerBootstrap(kafka)} --describe --group ${shQuote(group ?? "")} >"$tmp/group.out" 2>"$tmp/group.err" || group_rc=$?
    group_rc=\${group_rc:-0}
  else
    group_rc=0
    : >"$tmp/group.out"
    : >"$tmp/group.err"
  fi
else
  offsets_rc=1
  group_rc=1
  printf '%s\\n' 'kafka broker pod not found' >"$tmp/offsets.err"
  : >"$tmp/offsets.out" >"$tmp/group.out" >"$tmp/group.err"
fi
python3 - "$tmp" "$pod" "$offsets_rc" "$group_rc" <<'PY'
import json, os, sys
tmp, pod, offsets_rc, group_rc = sys.argv[1], sys.argv[2], int(sys.argv[3]), int(sys.argv[4])
def text(name, limit=2000):
    try:
        return open(os.path.join(tmp, name), encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
offsets = []
for line in text("offsets.out", 200000).splitlines():
    parts = line.strip().split(":")
    if len(parts) != 3:
        continue
    offsets.append({"topic": parts[0], "partition": int(parts[1]), "endOffset": int(parts[2])})
group_rows = []
for raw in text("group.out", 200000).splitlines():
    line = raw.strip()
    if not line or line.startswith("GROUP") or line.startswith("Consumer group"):
        continue
    cols = line.split()
    if len(cols) >= 6 and cols[1] != "-":
        try:
            group_rows.append({"group": cols[0], "topic": cols[1], "partition": int(cols[2]), "currentOffset": int(cols[3]), "logEndOffset": int(cols[4]), "lag": int(cols[5])})
        except ValueError:
            pass
payload = {
    "ok": offsets_rc == 0 and group_rc == 0,
    "target": "${target.id}",
    "namespace": "${target.namespace}",
    "cluster": "${kafka.cluster.name}",
    "pod": pod or None,
    "topicFilter": ${pythonJsonExpr(topic)},
    "groupFilter": ${pythonJsonExpr(group)},
    "offsets": sorted(offsets, key=lambda item: (item["topic"], item["partition"])),
    "groupOffsets": group_rows,
    "valuesPrinted": False,
    "stderrTail": (text("offsets.err") + text("group.err"))[-2000:],
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}

function tailScript(kafka: PlatformKafkaConfig, target: KafkaTarget, topic: string, limit: number): string {
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
pod="$(kubectl -n ${target.namespace} get pod -l strimzi.io/cluster=${kafka.cluster.name},strimzi.io/kind=Kafka -o jsonpath='{.items[0].metadata.name}' 2>"$tmp/pod.err" || true)"
if [ -n "$pod" ]; then
  kubectl -n ${target.namespace} exec "$pod" -- bin/kafka-get-offsets.sh --bootstrap-server ${brokerBootstrap(kafka)} --topic ${shQuote(topic)} >"$tmp/offsets.out" 2>"$tmp/offsets.err"
  offsets_rc=$?
  tail_rc=0
  : >"$tmp/messages.raw"
  while IFS=: read -r topic_name partition end_offset; do
    case "$partition" in ''|*[!0-9]*) continue ;; esac
    case "$end_offset" in ''|*[!0-9]*) continue ;; esac
    start=$(( end_offset > ${limit} ? end_offset - ${limit} : 0 ))
    count=$(( end_offset - start ))
    if [ "$count" -gt 0 ]; then
      timeout ${Math.min(kafka.validation.timeoutSeconds, 20)} kubectl -n ${target.namespace} exec "$pod" -- bin/kafka-console-consumer.sh --bootstrap-server ${brokerBootstrap(kafka)} --topic ${shQuote(topic)} --partition "$partition" --offset "$start" --max-messages "$count" >>"$tmp/messages.raw" 2>>"$tmp/messages.err" || true
    fi
  done <"$tmp/offsets.out"
else
  offsets_rc=1
  tail_rc=1
  printf '%s\\n' 'kafka broker pod not found' >"$tmp/offsets.err"
  : >"$tmp/offsets.out" >"$tmp/messages.raw" >"$tmp/messages.err"
fi
python3 - "$tmp" "$pod" "$offsets_rc" "$tail_rc" <<'PY'
import hashlib, json, os, sys
tmp, pod, offsets_rc, tail_rc = sys.argv[1], sys.argv[2], int(sys.argv[3]), int(sys.argv[4])
def text(name, limit=2000):
    try:
        return open(os.path.join(tmp, name), encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
offsets = []
for line in text("offsets.out", 200000).splitlines():
    parts = line.strip().split(":")
    if len(parts) == 3:
        offsets.append({"topic": parts[0], "partition": int(parts[1]), "endOffset": int(parts[2])})
messages = []
for index, line in enumerate(text("messages.raw", 200000).splitlines()):
    encoded = line.encode()
    messages.append({"index": index, "sha256": hashlib.sha256(encoded).hexdigest(), "bytes": len(encoded), "valuePrinted": False})
payload = {
    "ok": offsets_rc == 0 and tail_rc == 0,
    "target": "${target.id}",
    "namespace": "${target.namespace}",
    "cluster": "${kafka.cluster.name}",
    "pod": pod or None,
    "topic": "${topic}",
    "limit": ${limit},
    "offsets": offsets,
    "messages": messages[-${limit}:],
    "valuesPrinted": False,
    "stderrTail": (text("offsets.err") + text("messages.err"))[-2000:],
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}

function produceScript(kafka: PlatformKafkaConfig, target: KafkaTarget, topic: string, key: string | null, payload: string): string {
  const keyValue = key ?? `${kafka.management.shadowProducerId}-${Date.now()}`;
  const separator = "|";
  if (keyValue.includes(separator) || keyValue.includes("\n")) throw new Error("kafka produce --key must not contain | or newline");
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
pod="$(kubectl -n ${target.namespace} get pod -l strimzi.io/cluster=${kafka.cluster.name},strimzi.io/kind=Kafka -o jsonpath='{.items[0].metadata.name}' 2>"$tmp/pod.err" || true)"
cat >"$tmp/payload.json" <<'UNIDESK_KAFKA_SHADOW_PAYLOAD'
${payload}
UNIDESK_KAFKA_SHADOW_PAYLOAD
printf '%s' ${shQuote(keyValue)} >"$tmp/key.txt"
if [ -n "$pod" ]; then
  printf '%s%s%s\\n' "$(cat "$tmp/key.txt")" ${shQuote(separator)} "$(cat "$tmp/payload.json")" | kubectl -n ${target.namespace} exec -i "$pod" -- bin/kafka-console-producer.sh --bootstrap-server ${brokerBootstrap(kafka)} --topic ${shQuote(topic)} --reader-property parse.key=true --reader-property key.separator=${shQuote(separator)} --command-property acks=all >"$tmp/produce.out" 2>"$tmp/produce.err"
  produce_rc=$?
else
  produce_rc=1
  printf '%s\\n' 'kafka broker pod not found' >"$tmp/produce.err"
  : >"$tmp/produce.out"
fi
python3 - "$tmp" "$pod" "$produce_rc" <<'PY'
import hashlib, json, os, sys
tmp, pod, produce_rc = sys.argv[1], sys.argv[2], int(sys.argv[3])
def text(name, limit=1200):
    try:
        return open(os.path.join(tmp, name), encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
payload = open(os.path.join(tmp, "payload.json"), encoding="utf-8").read()
key = open(os.path.join(tmp, "key.txt"), encoding="utf-8").read()
payload_obj = {
    "ok": produce_rc == 0,
    "target": "${target.id}",
    "namespace": "${target.namespace}",
    "cluster": "${kafka.cluster.name}",
    "topic": "${topic}",
    "pod": pod or None,
    "keySha256": hashlib.sha256(key.encode()).hexdigest(),
    "message": {
        "sha256": hashlib.sha256(payload.encode()).hexdigest(),
        "bytes": len(payload.encode()),
        "valuesPrinted": False,
    },
    "producer": "${kafka.management.shadowProducerId}",
    "mode": "shadow-produce-only",
    "consume": {"enabled": False, "performed": False},
    "steps": {"produce": {"exitCode": produce_rc, "stdoutTail": text("produce.out"), "stderrTail": text("produce.err")}},
    "valuesPrinted": False,
}
print(json.dumps(payload_obj, ensure_ascii=False, indent=2))
sys.exit(0 if payload_obj["ok"] else 1)
PY
`;
}

function configSummary(kafka: PlatformKafkaConfig, target: KafkaTarget): Record<string, unknown> {
  return {
    configPath: configLabel,
    spec: kafka.metadata.spec,
    target: targetSummary(target),
    switch: kafka.defaults.switch,
    operator: operatorSummary(kafka, target),
    cluster: clusterSummary(kafka, target),
    topics: kafka.topics.map((topic) => topicSummary(topic)),
    clients: kafka.clients.map((client) => clientSummary(client)),
    management: kafka.management,
  };
}

function compactConfigSummary(kafka: PlatformKafkaConfig, target: KafkaTarget): Record<string, unknown> {
  return {
    configPath: configLabel,
    spec: kafka.metadata.spec,
    target: targetSummary(target),
    switch: kafka.defaults.switch,
    operator: {
      implementation: kafka.operator.implementation,
      version: kafka.operator.version,
      deploymentName: kafka.operator.deploymentName,
    },
    cluster: {
      name: kafka.cluster.name,
      bootstrap: bootstrapService(kafka, target),
      replicas: kafka.cluster.replicas,
      storageSize: kafka.cluster.storage.size,
      authorizationEnabled: kafka.cluster.authorization.enabled,
    },
    topicCount: kafka.topics.length,
    clientCount: kafka.clients.length,
    management: {
      defaultShadowTopic: kafka.management.defaultShadowTopic,
      defaultTailLimit: kafka.management.defaultTailLimit,
      maxTailLimit: kafka.management.maxTailLimit,
      shadowProducerId: kafka.management.shadowProducerId,
    },
    valuesPrinted: false,
  };
}

function targetSummary(target: KafkaTarget): Record<string, unknown> {
  return {
    id: target.id,
    route: target.route,
    namespace: target.namespace,
    role: target.role,
    createNamespace: target.createNamespace,
    storageClassName: target.storageClassName,
  };
}

function operatorSummary(kafka: PlatformKafkaConfig, target: KafkaTarget): Record<string, unknown> {
  return {
    implementation: kafka.operator.implementation,
    version: kafka.operator.version,
    manifestUrl: kafka.operator.manifestUrl,
    deploymentName: kafka.operator.deploymentName,
    namespace: target.namespace,
    crds: kafka.operator.crds,
  };
}

function clusterSummary(kafka: PlatformKafkaConfig, target: KafkaTarget): Record<string, unknown> {
  return {
    name: kafka.cluster.name,
    nodePoolName: kafka.cluster.nodePoolName,
    kafkaVersion: kafka.cluster.kafkaVersion,
    metadataVersion: kafka.cluster.metadataVersion,
    replicas: kafka.cluster.replicas,
    storage: { ...kafka.cluster.storage, storageClassName: target.storageClassName },
    bootstrap: bootstrapService(kafka, target),
    authorization: kafka.cluster.authorization,
  };
}

function topicSummary(topic: KafkaTopicSpec): Record<string, unknown> {
  return {
    name: topic.name,
    resourceName: topicResourceName(topic.name),
    partitions: topic.partitions,
    replicas: topic.replicas,
    retentionMs: topic.retentionMs,
    cleanupPolicy: topic.cleanupPolicy,
  };
}

function clientSummary(client: KafkaClientSpec): Record<string, unknown> {
  return {
    id: client.id,
    namespace: client.namespace,
    serviceAccountName: client.serviceAccountName,
    kafkaUserName: client.kafkaUserName,
    secretName: client.secretName,
    produceTopics: client.produceTopics,
    consumeTopics: client.consumeTopics,
    dlqTopics: client.dlqTopics,
    valuesPrinted: false,
  };
}

function policyChecks(kafka: PlatformKafkaConfig, target: KafkaTarget, manifest: string): Array<Record<string, unknown>> {
  return [
    { name: "yaml-source-of-truth", ok: true, detail: "Kafka target, namespace, Strimzi version, topics, clients, storage and switch are read from config/platform-infra/kafka.yaml." },
    { name: "namespace-is-platform-infra", ok: target.namespace === "platform-infra", detail: "Kafka runtime belongs to platform-infra, not an HWLAB or AgentRun lane namespace." },
    { name: "no-public-exposure", ok: !/^\s*type:\s*(NodePort|LoadBalancer)\s*$/mu.test(manifest) && !/^\s*kind:\s*Ingress\s*$/mu.test(manifest), detail: "Kafka POC is ClusterIP/internal only." },
    { name: "single-broker-poc", ok: kafka.cluster.replicas === 1, detail: "D518 POC uses one KRaft broker/controller; production HA remains out of scope." },
    { name: "allow-all-network-policy", ok: manifest.includes("kind: NetworkPolicy") && manifest.includes("name: allow-all") && manifest.includes(`namespace: ${target.namespace}`), detail: `NetworkPolicy/allow-all is rendered in ${target.namespace}.` },
    { name: "shadow-produce-enabled", ok: kafka.defaults.switch.shadowProduceEnabled === true, detail: "Apps may publish shadow events to Kafka." },
    { name: "app-integration-shadow-only", ok: kafka.defaults.switch.appIntegrationEnabled === true && kafka.defaults.switch.mode === "shadow-produce-only", detail: "App integration is limited to shadow production in this stage." },
    { name: "shadow-consume-disabled", ok: kafka.defaults.switch.shadowConsumeEnabled === false, detail: "Kafka consumer cutover remains disabled." },
  ];
}

function statusSummary(payload: Record<string, unknown>): Record<string, unknown> {
  return {
    ready: payload.ready === true,
    target: payload.target,
    route: payload.route,
    namespace: payload.namespace,
    operator: payload.operator,
    cluster: payload.cluster,
    crdsReady: payload.crdsReady === true,
    nodePoolReady: payload.nodePoolReady === true,
    topics: Array.isArray(payload.topics) ? payload.topics : [],
    users: Array.isArray(payload.users) ? payload.users : [],
    pods: Array.isArray(payload.pods) ? payload.pods : [],
    services: Array.isArray(payload.services) ? payload.services : [],
    clientSecrets: Array.isArray(payload.clientSecrets) ? payload.clientSecrets : [],
    valuesPrinted: false,
  };
}

function manifestObjectSummary(yaml: string): Array<Record<string, unknown>> {
  const objects: Array<Record<string, unknown>> = [];
  for (const doc of yaml.split(/^---$/mu)) {
    const kind = doc.match(/^\s*kind:\s*([A-Za-z0-9._-]+)\s*$/mu)?.[1];
    const name = doc.match(/^\s*name:\s*([A-Za-z0-9._-]+)\s*$/mu)?.[1];
    const namespace = doc.match(/^\s*namespace:\s*([A-Za-z0-9._-]+)\s*$/mu)?.[1] ?? null;
    if (kind !== undefined && name !== undefined) objects.push({ kind, name, namespace });
  }
  return objects;
}

function renderPlan(result: Record<string, unknown>): RenderedCliResult {
  const config = record(result.config);
  const target = record(config.target);
  const cluster = record(config.cluster);
  const operator = record(config.operator);
  const switchRecord = record(config.switch);
  const policy = arrayRecords(result.policy);
  const failed = policy.filter((item) => item.ok === false);
  const next = record(result.next);
  return rendered(result, "platform-infra kafka plan", [
    "PLATFORM-INFRA KAFKA PLAN",
    ...table(["FIELD", "VALUE", "DETAIL", "VALUE"], [
      ["TARGET", stringValue(target.id), "route", stringValue(target.route)],
      ["NAMESPACE", stringValue(target.namespace), "role", stringValue(target.role)],
      ["OPERATOR", `strimzi ${stringValue(operator.version)}`, "manifest", stringValue(operator.manifestUrl)],
      ["CLUSTER", stringValue(cluster.name), "bootstrap", stringValue(cluster.bootstrap)],
      ["STORAGE", `${stringValue(record(cluster.storage).size)} ${stringValue(record(cluster.storage).storageClassName)}`, "replicas", stringValue(cluster.replicas)],
      ["TOPICS", String(arrayRecords(config.topics).length), "clients", String(arrayRecords(config.clients).length)],
      ["SWITCH", stringValue(switchRecord.mode), "appIntegration", boolText(switchRecord.appIntegrationEnabled)],
      ["POLICY", failed.length === 0 ? "ok" : `failed=${failed.length}`, "valuesPrinted", "false"],
    ]),
    "",
    "NEXT",
    `  dry-run: ${stringValue(next.dryRun)}`,
    `  apply: ${stringValue(next.apply)}`,
    `  status: ${stringValue(next.status)}`,
    `  validate: ${stringValue(next.validate)}`,
    "",
    "Boundary: Kafka runtime is platform-infra only; P2 allows shadow produce/query and keeps consumer cutover disabled.",
    "Disclosure: Secret values are not printed; client entries show only object/key metadata.",
  ]);
}

function renderStatus(result: Record<string, unknown>): RenderedCliResult {
  const summary = record(result.summary);
  const cluster = record(summary.cluster);
  const operator = record(summary.operator);
  const topics = arrayRecords(summary.topics).map((topic) => [stringValue(topic.topicName), boolText(topic.ready), stringValue(topic.resource)]);
  const users = arrayRecords(summary.users).map((user) => [stringValue(user.name), boolText(user.ready), stringValue(user.secret)]);
  const pods = arrayRecords(summary.pods).map((pod) => [stringValue(pod.name), stringValue(pod.phase), boolText(pod.ready)]);
  return rendered(result, "platform-infra kafka status", [
    "PLATFORM-INFRA KAFKA STATUS",
    ...table(["TARGET", "ROUTE", "NAMESPACE", "READY"], [[stringValue(summary.target), stringValue(summary.route), stringValue(summary.namespace), boolText(summary.ready)]]),
    "",
    "CONTROL",
    ...table(["CHECK", "VALUE", "DETAIL"], [
      ["crds", boolText(summary.crdsReady), "Strimzi Kafka CRDs"],
      ["operator", boolText(operator.ready), `${stringValue(operator.name)} ${stringValue(operator.available)}/${stringValue(operator.desired)}`],
      ["cluster", boolText(record(cluster).ready), stringValue(record(cluster).bootstrap)],
      ["nodePool", boolText(summary.nodePoolReady), "dual-role broker/controller"],
    ]),
    "",
    "TOPICS",
    ...(topics.length === 0 ? ["-"] : table(["TOPIC", "READY", "RESOURCE"], topics)),
    "",
    "CLIENTS",
    ...(users.length === 0 ? ["-"] : table(["USER", "READY", "SECRET"], users)),
    "",
    "PODS",
    ...(pods.length === 0 ? ["-"] : table(["POD", "PHASE", "READY"], pods)),
    ...remoteErrorLines(result),
    "",
    `NEXT  bun scripts/cli.ts platform-infra kafka validate --target ${stringValue(summary.target)}`,
  ]);
}

function renderTopics(result: Record<string, unknown>): RenderedCliResult {
  const query = record(result.query);
  const topics = arrayRecords(query.topics).map((topic) => [
    stringValue(topic.name),
    boolText(topic.configured),
    stringValue(topic.partitionCount, "0"),
    stringValue(topic.replicationFactor, "0"),
    stringValue(topic.configs),
  ]);
  return rendered(result, "platform-infra kafka topics", [
    "PLATFORM-INFRA KAFKA TOPICS",
    ...table(["TARGET", "NAMESPACE", "CLUSTER", "POD"], [[stringValue(query.target), stringValue(query.namespace), stringValue(query.cluster), stringValue(query.pod)]]),
    "",
    ...(topics.length === 0 ? ["TOPICS  -"] : table(["TOPIC", "YAML", "PARTITIONS", "RF", "CONFIGS"], topics)),
    ...remoteErrorLines(result),
    "",
    "Disclosure: message values and Secret values are not printed.",
  ]);
}

function renderGroups(result: Record<string, unknown>): RenderedCliResult {
  const query = record(result.query);
  const groups = arrayRecords(query.groups).length > 0 ? arrayRecords(query.groups).map((item) => [stringValue(item.name)]) : (Array.isArray(query.groups) ? query.groups.map((item) => [String(item)]) : []);
  return rendered(result, "platform-infra kafka groups", [
    "PLATFORM-INFRA KAFKA GROUPS",
    ...table(["TARGET", "NAMESPACE", "CLUSTER", "POD"], [[stringValue(query.target), stringValue(query.namespace), stringValue(query.cluster), stringValue(query.pod)]]),
    "",
    ...(groups.length === 0 ? ["GROUPS  -"] : table(["GROUP"], groups)),
    ...remoteErrorLines(result),
  ]);
}

function renderOffsets(result: Record<string, unknown>): RenderedCliResult {
  const query = record(result.query);
  const offsets = arrayRecords(query.offsets).map((item) => [stringValue(item.topic), stringValue(item.partition), stringValue(item.endOffset)]);
  const groupOffsets = arrayRecords(query.groupOffsets).map((item) => [stringValue(item.group), stringValue(item.topic), stringValue(item.partition), stringValue(item.currentOffset), stringValue(item.logEndOffset), stringValue(item.lag)]);
  return rendered(result, "platform-infra kafka offsets", [
    "PLATFORM-INFRA KAFKA OFFSETS",
    ...table(["TARGET", "NAMESPACE", "CLUSTER", "POD"], [[stringValue(query.target), stringValue(query.namespace), stringValue(query.cluster), stringValue(query.pod)]]),
    "",
    "END OFFSETS",
    ...(offsets.length === 0 ? ["-"] : table(["TOPIC", "PARTITION", "END"], offsets)),
    "",
    "GROUP OFFSETS",
    ...(groupOffsets.length === 0 ? ["-"] : table(["GROUP", "TOPIC", "PARTITION", "CURRENT", "END", "LAG"], groupOffsets)),
    ...remoteErrorLines(result),
  ]);
}

function renderTail(result: Record<string, unknown>): RenderedCliResult {
  const query = record(result.query);
  const messages = arrayRecords(query.messages).map((item) => [stringValue(item.index), stringValue(item.sha256), stringValue(item.bytes), boolText(item.valuePrinted)]);
  const offsets = arrayRecords(query.offsets).map((item) => [stringValue(item.partition), stringValue(item.endOffset)]);
  return rendered(result, "platform-infra kafka tail", [
    "PLATFORM-INFRA KAFKA TAIL",
    ...table(["TARGET", "TOPIC", "LIMIT", "POD"], [[stringValue(query.target), stringValue(query.topic), stringValue(query.limit), stringValue(query.pod)]]),
    "",
    "OFFSETS",
    ...(offsets.length === 0 ? ["-"] : table(["PARTITION", "END"], offsets)),
    "",
    "MESSAGES",
    ...(messages.length === 0 ? ["-"] : table(["INDEX", "SHA256", "BYTES", "VALUE_PRINTED"], messages)),
    ...remoteErrorLines(result),
    "",
    "Disclosure: values are hashed only; use application-side replay tools for payload inspection.",
  ]);
}

function remoteErrorLines(result: Record<string, unknown>): string[] {
  if (result.ok !== false) return [];
  const remote = record(result.remote);
  const exitCode = stringValue(remote.exitCode);
  const stderr = stringValue(remote.stderrTail).replace(/\s+/gu, " ").trim().slice(0, 220);
  const stdout = stringValue(remote.stdoutTail).replace(/\s+/gu, " ").trim().slice(0, 220);
  const detail = stderr !== "-" ? stderr : stdout;
  return ["", "ERROR", ...table(["EXIT", "DETAIL"], [[exitCode, detail]])];
}

function rendered(result: Record<string, unknown>, command: string, lines: string[]): RenderedCliResult {
  return { ok: result.ok !== false, command, renderedText: lines.join("\n"), contentType: "text/plain" };
}

function parseApplyOptions(args: string[]): ApplyOptions {
  const commonArgs: string[] = [];
  let confirm = false;
  let dryRun = false;
  let wait = false;
  for (let index = 0; index < args.length; index += 1) {
    const arg = args[index];
    if (arg === "--confirm") confirm = true;
    else if (arg === "--dry-run") dryRun = true;
    else if (arg === "--wait") wait = true;
    else {
      commonArgs.push(arg);
      if (arg === "--target" || arg === "--node") {
        commonArgs.push(args[index + 1] ?? "");
        index += 1;
      }
    }
  }
  if (confirm && dryRun) throw new Error("kafka apply accepts only one of --confirm or --dry-run");
  return { ...parseCommonOptions(commonArgs), confirm, dryRun: dryRun || !confirm, wait };
}

function parseCommonOptions(args: string[]): CommonOptions {
  let targetId: string | null = null;
  let full = false;
  let raw = false;
  for (let index = 0; index < args.length; index += 1) {
    const arg = args[index];
    if (arg === "--target" || arg === "--node") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
      if (!/^[A-Za-z0-9._-]+$/u.test(value)) throw new Error(`${arg} must be a simple target id`);
      targetId = value;
      index += 1;
    } else if (arg === "--full") {
      full = true;
    } else if (arg === "--raw") {
      raw = true;
      full = true;
    } else {
      throw new Error(`unsupported kafka option: ${arg}`);
    }
  }
  return { targetId, full, raw };
}

function parseInspectOptions(args: string[]): KafkaInspectOptions {
  const commonArgs: string[] = [];
  let topic: string | null = null;
  let group: string | null = null;
  let limit: number | null = null;
  for (let index = 0; index < args.length; index += 1) {
    const arg = args[index];
    if (arg === "--topic") {
      topic = parseTopicOption(args[index + 1], arg);
      index += 1;
    } else if (arg === "--group") {
      group = parseSimpleOption(args[index + 1], arg);
      index += 1;
    } else if (arg === "--limit") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--limit requires a value");
      limit = Number(value);
      if (!Number.isInteger(limit) || limit < 1) throw new Error("--limit must be a positive integer");
      index += 1;
    } else {
      commonArgs.push(arg);
      if (arg === "--target" || arg === "--node") {
        commonArgs.push(args[index + 1] ?? "");
        index += 1;
      }
    }
  }
  return { ...parseCommonOptions(commonArgs), topic, group, limit };
}

async function parseProduceOptions(args: string[]): Promise<KafkaProduceOptions> {
  const commonArgs: string[] = [];
  let topic: string | null = null;
  let key: string | null = null;
  let source = "manual-cli";
  let eventType = "platform-infra.kafka.shadow-produce.v1";
  let payload: string | null = null;
  for (let index = 0; index < args.length; index += 1) {
    const arg = args[index];
    if (arg === "--topic") {
      topic = parseTopicOption(args[index + 1], arg);
      index += 1;
    } else if (arg === "--key") {
      key = parseSimpleOption(args[index + 1], arg);
      index += 1;
    } else if (arg === "--source") {
      source = parseSimpleOption(args[index + 1], arg);
      index += 1;
    } else if (arg === "--event-type") {
      eventType = parseTopicOption(args[index + 1], arg);
      index += 1;
    } else if (arg === "--payload-stdin") {
      payload = await Bun.stdin.text();
    } else {
      commonArgs.push(arg);
      if (arg === "--target" || arg === "--node") {
        commonArgs.push(args[index + 1] ?? "");
        index += 1;
      }
    }
  }
  return { ...parseCommonOptions(commonArgs), topic, key, source, eventType, payload };
}

function bootstrapService(kafka: PlatformKafkaConfig, target: KafkaTarget): string {
  return `${kafka.cluster.name}-kafka-bootstrap.${target.namespace}.svc.cluster.local:${kafka.cluster.listeners.plain.port}`;
}

function brokerBootstrap(kafka: PlatformKafkaConfig): string {
  return `${kafka.cluster.name}-kafka-bootstrap:${kafka.cluster.listeners.plain.port}`;
}

function resolveTopic(kafka: PlatformKafkaConfig, topic: string | null, fallback: string | null): string | null {
  const resolved = topic ?? fallback;
  if (resolved === null) return null;
  if (!kafka.topics.some((item) => item.name === resolved)) throw new Error(`unknown kafka topic ${resolved}; known topics: ${kafka.topics.map((item) => item.name).join(", ")}`);
  return resolved;
}

function boundedLimit(kafka: PlatformKafkaConfig, limit: number | null): number {
  const value = limit ?? kafka.management.defaultTailLimit;
  if (!Number.isInteger(value) || value < 1) throw new Error("limit must be a positive integer");
  if (value > kafka.management.maxTailLimit) throw new Error(`limit ${value} exceeds ${configLabel}.management.maxTailLimit=${kafka.management.maxTailLimit}`);
  return value;
}

function buildShadowPayload(kafka: PlatformKafkaConfig, target: KafkaTarget, options: KafkaProduceOptions, topic: string): string {
  let userPayload: unknown = null;
  if (options.payload !== null) {
    const trimmed = options.payload.trim();
    if (trimmed.length > 0) {
      try {
        userPayload = JSON.parse(trimmed) as unknown;
      } catch {
        userPayload = { textSha256: sha256Hex(trimmed), valuesPrinted: false };
      }
    }
  }
  return JSON.stringify({
    eventId: `evt_${target.id}_${Date.now()}_${Math.random().toString(16).slice(2)}`,
    schemaVersion: 1,
    eventType: options.eventType,
    producer: kafka.management.shadowProducerId,
    source: options.source,
    node: target.id,
    namespace: target.namespace,
    topic,
    mode: "shadow-produce-only",
    consumeEnabled: false,
    appIntegrationEnabled: kafka.defaults.switch.appIntegrationEnabled,
    shadowProduceEnabled: kafka.defaults.switch.shadowProduceEnabled,
    shadowConsumeEnabled: kafka.defaults.switch.shadowConsumeEnabled,
    partitionKey: options.key ?? null,
    occurredAt: new Date().toISOString(),
    payload: userPayload ?? {
      kind: "manual-shadow-produce",
      spec: kafka.metadata.spec,
      valuesPrinted: false,
    },
    redaction: { valuesPrinted: false },
  });
}

function parseTopicOption(value: string | undefined, optionName: string): string {
  if (value === undefined || value.startsWith("--")) throw new Error(`${optionName} requires a value`);
  if (!/^[A-Za-z0-9._-]+$/u.test(value) || value.includes("..")) throw new Error(`${optionName} has an unsupported Kafka topic/event name`);
  return value;
}

function parseSimpleOption(value: string | undefined, optionName: string): string {
  if (value === undefined || value.startsWith("--")) throw new Error(`${optionName} requires a value`);
  if (!/^[A-Za-z0-9._:/=-]+$/u.test(value)) throw new Error(`${optionName} must be a simple token`);
  return value;
}

function pythonJsonExpr(value: unknown): string {
  return `json.loads(${JSON.stringify(JSON.stringify(value))})`;
}

function topicNameField(record: Record<string, unknown>, key: string, path: string): string {
  const value = y.stringField(record, key, path);
  if (!/^[A-Za-z0-9._-]+$/u.test(value) || value.includes("..")) throw new Error(`${configLabel}.${path}.${key} has an unsupported Kafka topic name`);
  return value;
}

function topicArray(record: Record<string, unknown>, key: string, path: string): string[] {
  return y.stringArrayField(record, key, path).map((value) => {
    if (!/^[A-Za-z0-9._-]+$/u.test(value) || value.includes("..")) throw new Error(`${configLabel}.${path}.${key} contains unsupported topic name ${value}`);
    return value;
  });
}

function topicResourceName(topic: string): string {
  return topic.toLowerCase().replace(/[^a-z0-9-]+/gu, "-").replace(/^-+|-+$/gu, "").slice(0, 253);
}

function positiveInteger(record: Record<string, unknown>, key: string, path: string): number {
  const value = y.integerField(record, key, path);
  if (value < 1) throw new Error(`${configLabel}.${path}.${key} must be >= 1`);
  return value;
}

function quoteAnnotation(value: string): string {
  return JSON.stringify(value);
}

function record(value: unknown): Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record<string, unknown> : {};
}

function arrayRecords(value: unknown): Record<string, unknown>[] {
  return Array.isArray(value) ? value.map(record) : [];
}

function stringValue(value: unknown, fallback = "-"): string {
  if (value === undefined || value === null || value === "") return fallback;
  return String(value);
}

function boolText(value: unknown): string {
  return value === true ? "yes" : value === false ? "no" : "-";
}

function table(headers: string[], rows: string[][]): string[] {
  const widths = headers.map((header, index) => Math.max(header.length, ...rows.map((row) => row[index]?.length ?? 0)));
  const renderRow = (row: string[]) => row.map((cell, index) => cell.padEnd(widths[index] ?? cell.length)).join("  ").trimEnd();
  return [renderRow(headers), ...rows.map(renderRow)];
}