kmsvc-cli/internal/cli/queue.go

package cli

import (
	"encoding/json"
	"fmt"
	"io"
	"strconv"
	"strings"
	"text/tabwriter"

	"github.com/spf13/cobra"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
)

// queueSummary is the subset of a Queue CRD's spec/status this CLI surfaces.
// Read directly off the unstructured object rather than a generated
// clientset, since kmsvc-cli otherwise has no dependency on the
// kafaka-management-service module's API types.
type queueSummary struct {
	Name        string `json:"name"`
	Namespace   string `json:"namespace"`
	Phase       string `json:"phase"`
	FIFO        bool   `json:"fifoQueue"`
	ShardCount  int    `json:"shardCount"`
	MaxReceives int64  `json:"maxReceiveCount"`
}

func newQueueCmd(flags *globalFlags) *cobra.Command {
	cmd := &cobra.Command{
		Use:   "queue",
		Short: "List and describe Queue CRDs (the queue-operator's source of truth)",
	}
	cmd.AddCommand(
		newQueueListCmd(flags),
		newQueueDescribeCmd(flags),
	)
	return cmd
}

// queueGVK is the apiVersion/kind pair for the Queue CRD, matching queueGVR
// (kmsvc.io/v1, plural "queues") by standard k8s singular-Kind convention.
const (
	queueAPIVersion = "kmsvc.io/v1"
	queueKind       = "Queue"
)

func newQueueCreateCmd(flags *globalFlags) *cobra.Command {
	var namespace string
	var setFields []string

	cmd := &cobra.Command{
		Use:   "create-queue [name]",
		Short: "Create a Queue CRD (operator defaults apply unless overridden with --set)",
		Args:  cobra.ExactArgs(1),
		RunE: func(cmd *cobra.Command, args []string) error {
			cl, err := newDynamicClient()
			if err != nil {
				return err
			}

			spec := map[string]any{}
			for _, kv := range setFields {
				if err := applySetField(spec, kv); err != nil {
					return err
				}
			}

			obj := &unstructured.Unstructured{Object: map[string]any{
				"apiVersion": queueAPIVersion,
				"kind":       queueKind,
				"metadata": map[string]any{
					"name":      args[0],
					"namespace": namespace,
				},
				"spec": spec,
			}}

			created, err := cl.Resource(queueGVR).Namespace(namespace).Create(cmd.Context(), obj, metav1.CreateOptions{})
			if err != nil {
				return fmt.Errorf("create queue %s: %w", args[0], err)
			}
			fmt.Fprintf(cmd.OutOrStdout(), "queue/%s created\n", created.GetName())
			return nil
		},
	}
	cmd.Flags().StringVarP(&namespace, "namespace", "n", "sqs", "namespace to create the Queue CRD in")
	cmd.Flags().StringArrayVar(&setFields, "set", nil, "override a spec field, key=value (e.g. --set fifoQueue=true), repeatable")
	return cmd
}

func newQueueDeleteCmd(flags *globalFlags) *cobra.Command {
	var namespace string

	cmd := &cobra.Command{
		Use:   "delete-queue [name]",
		Short: "Delete a Queue CRD",
		Args:  cobra.ExactArgs(1),
		RunE: func(cmd *cobra.Command, args []string) error {
			cl, err := newDynamicClient()
			if err != nil {
				return err
			}
			if err := cl.Resource(queueGVR).Namespace(namespace).Delete(cmd.Context(), args[0], metav1.DeleteOptions{}); err != nil {
				return fmt.Errorf("delete queue %s: %w", args[0], err)
			}
			fmt.Fprintf(cmd.OutOrStdout(), "queue/%s deleted\n", args[0])
			return nil
		},
	}
	cmd.Flags().StringVarP(&namespace, "namespace", "n", "sqs", "namespace the Queue CRD lives in")
	return cmd
}

// applySetField parses a "key=value" pair and writes it into spec, coercing
// value to bool/int64 when it parses as one, else leaving it as a string —
// mirrors helm --set's pragmatic type inference since the CRD schema isn't
// known to this CLI.
func applySetField(spec map[string]any, kv string) error {
	key, value, ok := strings.Cut(kv, "=")
	if !ok {
		return fmt.Errorf("--set %q: expected key=value", kv)
	}
	if b, err := strconv.ParseBool(value); err == nil {
		spec[key] = b
		return nil
	}
	if i, err := strconv.ParseInt(value, 10, 64); err == nil {
		spec[key] = i
		return nil
	}
	spec[key] = value
	return nil
}

func newQueueListCmd(flags *globalFlags) *cobra.Command {
	var namespace string

	cmd := &cobra.Command{
		Use:   "list",
		Short: "List Queue CRDs in a namespace",
		RunE: func(cmd *cobra.Command, args []string) error {
			cl, err := newDynamicClient()
			if err != nil {
				return err
			}
			list, err := cl.Resource(queueGVR).Namespace(namespace).List(cmd.Context(), metav1.ListOptions{})
			if err != nil {
				return fmt.Errorf("list queues: %w", err)
			}

			summaries := make([]queueSummary, 0, len(list.Items))
			for _, item := range list.Items {
				summaries = append(summaries, summarizeQueue(item.Object))
			}
			return renderQueues(cmd.OutOrStdout(), flags.output, summaries)
		},
	}
	cmd.Flags().StringVarP(&namespace, "namespace", "n", "sqs", "namespace the Queue CRDs live in")
	return cmd
}

func newQueueDescribeCmd(flags *globalFlags) *cobra.Command {
	var namespace string

	cmd := &cobra.Command{
		Use:   "describe [name]",
		Short: "Show full status (shards, phase) for one Queue CRD",
		Args:  cobra.ExactArgs(1),
		RunE: func(cmd *cobra.Command, args []string) error {
			cl, err := newDynamicClient()
			if err != nil {
				return err
			}
			obj, err := cl.Resource(queueGVR).Namespace(namespace).Get(cmd.Context(), args[0], metav1.GetOptions{})
			if err != nil {
				return fmt.Errorf("get queue %s: %w", args[0], err)
			}

			data, err := json.MarshalIndent(obj.Object, "", "  ")
			if err != nil {
				return err
			}
			fmt.Fprintln(cmd.OutOrStdout(), string(data))
			return nil
		},
	}
	cmd.Flags().StringVarP(&namespace, "namespace", "n", "sqs", "namespace the Queue CRD lives in")
	return cmd
}

func summarizeQueue(obj map[string]any) queueSummary {
	name, _, _ := unstructured.NestedString(obj, "metadata", "name")
	namespace, _, _ := unstructured.NestedString(obj, "metadata", "namespace")
	phase, _, _ := unstructured.NestedString(obj, "status", "phase")
	fifo, _, _ := unstructured.NestedBool(obj, "spec", "fifoQueue")
	maxReceives, _, _ := unstructured.NestedInt64(obj, "spec", "maxReceiveCount")

	shardCount := 0
	if shards, ok, _ := unstructured.NestedSlice(obj, "status", "shards"); ok {
		shardCount = len(shards)
	}

	return queueSummary{
		Name:        name,
		Namespace:   namespace,
		Phase:       phase,
		FIFO:        fifo,
		ShardCount:  shardCount,
		MaxReceives: maxReceives,
	}
}

func renderQueues(w io.Writer, format string, queues []queueSummary) error {
	if format == "json" {
		return json.NewEncoder(w).Encode(queues)
	}

	tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0)
	fmt.Fprintln(tw, "NAME\tNAMESPACE\tPHASE\tFIFO\tSHARDS\tMAX_RECEIVES")
	for _, q := range queues {
		fmt.Fprintf(tw, "%s\t%s\t%s\t%t\t%d\t%d\n", q.Name, q.Namespace, q.Phase, q.FIFO, q.ShardCount, q.MaxReceives)
	}
	return tw.Flush()
}