floci-ui/packages/frontend/src/components/MskPanel.tsx

import {useEffect, useState} from "react";
import {Activity, Loader2, MessageSquare, Network, Plus, RefreshCw, Send, Server, Trash2} from "lucide-react";
import {useMutation, useQuery, useQueryClient} from "@tanstack/react-query";
import {createMskCluster, createMskTopic, deleteMskCluster, deleteMskTopic, getMskConsumerGroupLag, listMskClusters, listMskTopics, peekMskMessages, produceMskMessage} from "@/api/mskClient";
import {listEc2Subnets} from "@/api/aws/ec2.api";

export function MskPanel({runtimeReachable}: {runtimeReachable: boolean}) {
  const queryClient = useQueryClient();
  const [clusterArn, setClusterArn] = useState("");
  const [deleteClusterConfirmation, setDeleteClusterConfirmation] = useState(false);
  const [clusterName, setClusterName] = useState("");
  const [brokers, setBrokers] = useState("1");
  const [version, setVersion] = useState("3.7.x");
  const [subnets, setSubnets] = useState<string[]>([]);
  const [topic, setTopic] = useState("");
  const [partitions, setPartitions] = useState("1");
  const [retentionMs, setRetentionMs] = useState("");
  const [messageTopic, setMessageTopic] = useState("");
  const [messageKey, setMessageKey] = useState("");
  const [messageValue, setMessageValue] = useState("");
  const [groupId, setGroupId] = useState("");
  const [messages, setMessages] = useState<Awaited<ReturnType<typeof peekMskMessages>>>([]);
  const [lag, setLag] = useState<Awaited<ReturnType<typeof getMskConsumerGroupLag>>>([]);
  const clustersQuery = useQuery({queryKey: ["msk-clusters"], queryFn: ({signal}) => listMskClusters(signal), enabled: runtimeReachable});
  const subnetsQuery = useQuery({queryKey: ["ec2", "subnets"], queryFn: ({signal}) => listEc2Subnets(undefined, signal), enabled: runtimeReachable});
  const topicsQuery = useQuery({queryKey: ["msk-topics", clusterArn], queryFn: ({signal}) => listMskTopics(clusterArn, signal), enabled: runtimeReachable && Boolean(clusterArn)});

  useEffect(() => { const firstArn = clustersQuery.data?.[0]?.ClusterArn; if (!clusterArn && firstArn) setClusterArn(firstArn); }, [clusterArn, clustersQuery.data]);
  useEffect(() => { setMessageTopic(""); setMessages([]); setLag([]); }, [clusterArn]);

  const createCluster = useMutation({mutationFn: () => createMskCluster({clusterName: clusterName.trim(), kafkaVersion: version.trim(), numberOfBrokerNodes: Number(brokers), brokerNodeGroupInfo: {clientSubnets: subnets}}), onSuccess: () => { setClusterName(""); void queryClient.invalidateQueries({queryKey: ["msk-clusters"]}); }});
  const deleteCluster = useMutation({mutationFn: () => deleteMskCluster(clusterArn), onSuccess: () => { setClusterArn(""); setDeleteClusterConfirmation(false); void queryClient.invalidateQueries({queryKey: ["msk-clusters"]}); }});
  const createTopic = useMutation({mutationFn: () => createMskTopic(clusterArn, {topic: topic.trim(), numPartitions: Number(partitions), ...(retentionMs.trim() ? {configEntries: [{name: "retention.ms", value: retentionMs.trim()}]} : {})}), onSuccess: () => { setTopic(""); setRetentionMs(""); void queryClient.invalidateQueries({queryKey: ["msk-topics", clusterArn]}); }});
  const deleteTopic = useMutation({mutationFn: (topicName: string) => deleteMskTopic(clusterArn, topicName), onSuccess: () => { setMessages([]); void queryClient.invalidateQueries({queryKey: ["msk-topics", clusterArn]}); }});
  const produce = useMutation({mutationFn: () => produceMskMessage(clusterArn, messageTopic, {value: messageValue, ...(messageKey.trim() ? {key: messageKey} : {})}), onSuccess: () => { setMessageKey(""); setMessageValue(""); }});
  const peek = useMutation({mutationFn: () => peekMskMessages(clusterArn, messageTopic, 50), onSuccess: setMessages});
  const loadLag = useMutation({mutationFn: () => getMskConsumerGroupLag(clusterArn, groupId.trim(), messageTopic || undefined), onSuccess: setLag});
  const error = createCluster.error ?? deleteCluster.error ?? createTopic.error ?? deleteTopic.error ?? produce.error ?? peek.error ?? loadLag.error;

  return <div className="sns-topic-workspace msk-workspace">
    <section className="table-panel msk-service-section">
      <header className="msk-section-header">
        <div className="msk-section-heading"><span><Server size={17}/></span><div><p className="eyebrow">Amazon MSK</p><h3>Kafka clusters</h3><p className="muted compact-text">Provision broker nodes in existing EC2 subnets for local Kafka workloads.</p></div></div>
        <button className="button" type="button" disabled={!runtimeReachable || clustersQuery.isFetching} onClick={() => clustersQuery.refetch()}><RefreshCw size={13}/>{clustersQuery.isFetching ? "Loading" : "Refresh"}</button>
      </header>
      <form className="msk-form" onSubmit={(event) => { event.preventDefault(); if (clusterName.trim() && Number(brokers) > 0 && version.trim() && subnets.length > 0) createCluster.mutate(); }}>
        <label className="dynamic-field"><span>Cluster name <em className="field-required">*</em></span><input className="input" value={clusterName} onChange={(event) => setClusterName(event.target.value)} placeholder="events-cluster" required /></label>
        <label className="dynamic-field"><span>Broker nodes <em className="field-required">*</em></span><input className="input" type="number" min="1" value={brokers} onChange={(event) => setBrokers(event.target.value)} required /></label>
        <label className="dynamic-field"><span>Kafka version <em className="field-required">*</em></span><input className="input" value={version} onChange={(event) => setVersion(event.target.value)} required /></label>
        <label className="dynamic-field"><span>Client subnets <em className="field-required">*</em></span><select className="input subnet-multiselect" multiple size={3} value={subnets} onChange={(event) => setSubnets(Array.from(event.currentTarget.selectedOptions, (option) => option.value))} required>{subnetsQuery.data?.map((subnet) => <option key={subnet.subnetId} value={subnet.subnetId}>{subnet.subnetId} ({subnet.availabilityZone})</option>)}</select><small>Select one or more existing EC2 subnets.</small></label>
        <div className="msk-form-actions"><span>Broker access is open: no IAM, SASL, or TLS authentication is configured.</span><button className="button primary" type="submit" disabled={!runtimeReachable || subnets.length === 0 || createCluster.isPending}>{createCluster.isPending ? <Loader2 size={13} className="spin"/> : <Plus size={13}/>}{createCluster.isPending ? "Creating" : "Create cluster"}</button></div>
      </form>
      <div className="msk-active-resource">
        <label className="dynamic-field"><span>Active cluster</span><select className="input" value={clusterArn} onChange={(event) => { setClusterArn(event.target.value); setDeleteClusterConfirmation(false); }} disabled={!runtimeReachable}><option value="">Select a cluster</option>{clustersQuery.data?.map((cluster) => <option key={cluster.ClusterArn} value={cluster.ClusterArn}>{cluster.ClusterName ?? cluster.ClusterArn} {cluster.State ? `(${cluster.State})` : ""}</option>)}</select></label>
        {clusterArn && (deleteClusterConfirmation ? <div className="apigateway-route-confirm"><span>Delete this cluster and all topics?</span><button className="button danger compact" type="button" disabled={deleteCluster.isPending} onClick={() => deleteCluster.mutate()}>{deleteCluster.isPending ? "Deleting" : "Delete cluster"}</button><button className="button compact" type="button" disabled={deleteCluster.isPending} onClick={() => setDeleteClusterConfirmation(false)}>Cancel</button></div> : <button className="button danger compact" type="button" disabled={!runtimeReachable || deleteCluster.isPending} onClick={() => setDeleteClusterConfirmation(true)}><Trash2 size={13}/>Delete cluster</button>)}
      </div>
    </section>

    <section className="table-panel msk-service-section">
      <header className="msk-section-header"><div className="msk-section-heading"><span><Network size={17}/></span><div><p className="eyebrow">Topics</p><h3>Kafka Admin API</h3><p className="muted compact-text">Define partitions and optional retention for the active cluster.</p></div></div></header>
      <form className="msk-form" onSubmit={(event) => { event.preventDefault(); if (clusterArn && topic.trim() && Number(partitions) > 0) createTopic.mutate(); }}>
        <label className="dynamic-field"><span>Topic name <em className="field-required">*</em></span><input className="input" aria-label="Topic" value={topic} onChange={(event) => setTopic(event.target.value)} placeholder="orders.created" required /></label>
        <label className="dynamic-field"><span>Partitions <em className="field-required">*</em></span><input className="input" type="number" min="1" value={partitions} onChange={(event) => setPartitions(event.target.value)} required /></label>
        <label className="dynamic-field"><span>Retention in milliseconds</span><input className="input" type="number" min="0" value={retentionMs} onChange={(event) => setRetentionMs(event.target.value)} placeholder="Optional"/><small>Configuration is accepted, but enforcement unverified.</small></label>
        <div className="msk-form-actions"><button className="button primary" type="submit" disabled={!runtimeReachable || !clusterArn || createTopic.isPending}><Plus size={13}/>{createTopic.isPending ? "Creating" : "Create topic"}</button></div>
      </form>
      <div className="msk-resource-list">{topicsQuery.isLoading && <p className="muted compact-text">Loading topics...</p>}{topicsQuery.data?.length === 0 && <p className="muted compact-text">No topics found for this cluster.</p>}{topicsQuery.data?.map((item) => <div className="apigateway-route-item" key={item.name}><div className="apigateway-route-select"><code>{item.name}</code><small>{item.partitionCount} partition{item.partitionCount === 1 ? "" : "s"}</small></div><button className="icon-btn danger" type="button" aria-label={`Delete topic ${item.name}`} disabled={deleteTopic.isPending} onClick={() => deleteTopic.mutate(item.name)}><Trash2 size={14}/></button></div>)}</div>
    </section>

    <section className="table-panel msk-service-section">
      <header className="msk-section-header"><div className="msk-section-heading"><span><MessageSquare size={17}/></span><div><p className="eyebrow">Messages</p><h3>Produce and inspect</h3><p className="muted compact-text">Publish records or peek from the beginning with a non-committing consumer.</p></div></div></header>
      <form className="msk-form" onSubmit={(event) => { event.preventDefault(); if (clusterArn && messageTopic && messageValue) produce.mutate(); }}>
        <label className="dynamic-field"><span>Topic <em className="field-required">*</em></span><select className="input" aria-label="Topic" value={messageTopic} onChange={(event) => setMessageTopic(event.target.value)} required><option value="">Select a topic</option>{topicsQuery.data?.map((item) => <option key={item.name} value={item.name}>{item.name}</option>)}</select></label>
        <label className="dynamic-field"><span>Message key</span><input className="input" value={messageKey} onChange={(event) => setMessageKey(event.target.value)} placeholder="Optional partition key"/></label>
        <label className="dynamic-field dynamic-field--span"><span>Message value <em className="field-required">*</em></span><textarea className="textarea code-textarea small" aria-label="Value" value={messageValue} onChange={(event) => setMessageValue(event.target.value)} placeholder="Enter the record payload" required /></label>
        <div className="msk-form-actions"><button className="button" type="button" disabled={!runtimeReachable || !clusterArn || !messageTopic || peek.isPending} onClick={() => peek.mutate()}>{peek.isPending ? "Peeking" : "Peek messages"}</button><button className="button primary" type="submit" disabled={!runtimeReachable || !clusterArn || !messageTopic || !messageValue || produce.isPending}>{produce.isPending ? <Loader2 size={13} className="spin"/> : <Send size={13}/>}{produce.isPending ? "Producing" : "Produce message"}</button></div>
      </form>
      <div className="msk-resource-list">{messages.map((message) => <article className="apigateway-route-item" key={`${message.partition}:${message.offset}`}><div className="apigateway-route-select"><pre>{message.value ?? ""}</pre><small>Partition {message.partition} · Offset {message.offset} · Timestamp {formatTimestamp(message.timestamp)}</small>{message.key && <small>Key: <code>{message.key}</code></small>}</div></article>)}{messages.length === 0 && <p className="muted compact-text">Select a topic and peek to inspect up to 50 messages.</p>}</div>
    </section>

    <section className="table-panel msk-service-section">
      <header className="msk-section-header"><div className="msk-section-heading"><span><Activity size={17}/></span><div><p className="eyebrow">Consumer groups</p><h3>Lag inspection</h3><p className="muted compact-text">Compare committed offsets with partition end offsets.</p></div></div></header>
      <form className="msk-form msk-lag-form" onSubmit={(event) => { event.preventDefault(); if (clusterArn && groupId.trim()) loadLag.mutate(); }}><label className="dynamic-field"><span>Consumer group ID <em className="field-required">*</em></span><input className="input" aria-label="Consumer group ID" value={groupId} onChange={(event) => setGroupId(event.target.value)} placeholder="checkout-workers" required /></label><div className="msk-form-actions"><button className="button primary" type="submit" disabled={!runtimeReachable || !clusterArn || !groupId.trim() || loadLag.isPending}>{loadLag.isPending ? "Loading" : "View lag"}</button></div></form>
      <div className="msk-resource-list">{lag.map((topicLag) => <div className="apigateway-route-item" key={topicLag.topic}><div className="apigateway-route-select"><code>{topicLag.topic}</code><small>Total lag: {topicLag.totalLag}</small>{topicLag.partitions.map((partition) => <small key={partition.partition}>Partition {partition.partition}: {partition.currentOffset} / {partition.endOffset}, lag {partition.lag}</small>)}</div></div>)}</div>
      {error instanceof Error && <p className="error-text compact-text msk-error">{error.message}</p>}
    </section>
  </div>;
}

function formatTimestamp(timestamp: string): string { const date = new Date(Number(timestamp)); return Number.isNaN(date.getTime()) ? timestamp : date.toLocaleString(); }