Ux 1341 remove rest for topic listing

backend/pkg/api/connect/service/topic/v1/console.go

@@ -0,0 +1,103 @@
+// Copyright 2023 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package topic
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"sort"
+	"strings"
+
+	"connectrpc.com/connect"
+	"github.com/redpanda-data/common-go/api/pagination"
+
+	apierrors "github.com/redpanda-data/console/backend/pkg/api/connect/errors"
+	"github.com/redpanda-data/console/backend/pkg/console"
+	v1alpha1 "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/console/v1alpha1"
+	"github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/console/v1alpha1/consolev1alpha1connect"
+	v1 "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/dataplane/v1"
+)
+
+var _ consolev1alpha1connect.TopicServiceHandler = (*ConsoleService)(nil)
+
+// ConsoleService is the Console-facing implementation of the topic service.
+//
+// Unlike the other Console services it does not wrap the dataplane handler: the dataplane
+// ListTopics deliberately returns only cheap metadata, whereas the Console topic list also renders
+// cleanup policy and on-disk size. Those come from console.GetTopicsOverview, which fans out a
+// DescribeLogDirs request to every broker — a cost the public dataplane API should not pay.
+type ConsoleService struct {
+	logger     *slog.Logger
+	consoleSvc console.Servicer
+	mapper     consoleMapper
+	defaulter  defaulter
+}
+
+// NewConsoleService creates a new ConsoleService.
+func NewConsoleService(logger *slog.Logger, consoleSvc console.Servicer) *ConsoleService {
+	return &ConsoleService{
+		logger:     logger,
+		consoleSvc: consoleSvc,
+		mapper:     consoleMapper{},
+		defaulter:  defaulter{},
+	}
+}
+
+// ListTopics lists Kafka topics along with the metadata the Console topic list renders.
+func (s *ConsoleService) ListTopics(ctx context.Context, req *connect.Request[v1alpha1.ListTopicsRequest]) (*connect.Response[v1alpha1.ListTopicsResponse], error) {
+	dataplaneReq := req.Msg.GetRequest()
+	if dataplaneReq == nil {
+		dataplaneReq = &v1.ListTopicsRequest{}
+	}
+	s.defaulter.applyListTopicsRequest(dataplaneReq)
+
+	topicSummaries, err := s.consoleSvc.GetTopicsOverview(ctx)
+	if err != nil {
+		return nil, apierrors.NewConnectError(
+			connect.CodeInternal,
+			err,
+			apierrors.NewErrorInfo(v1.Reason_REASON_KAFKA_API_ERROR.String(), apierrors.KeyValsFromKafkaError(err)...),
+		)
+	}
+
+	if nameContains := dataplaneReq.GetFilter().GetNameContains(); nameContains != "" {
+		filteredTopics := make([]*console.TopicSummary, 0, len(topicSummaries))
+		for _, topic := range topicSummaries {
+			if strings.Contains(topic.TopicName, nameContains) {
+				filteredTopics = append(filteredTopics, topic)
+			}
+		}
+		topicSummaries = filteredTopics
+	}
+
+	topics := s.mapper.topicSummariesToProto(topicSummaries)
+
+	var nextPageToken string
+	if dataplaneReq.GetPageSize() > 0 {
+		sort.SliceStable(topics, func(i, j int) bool {
+			return topics[i].GetName() < topics[j].GetName()
+		})
+		page, token, err := pagination.SliceToPaginatedWithToken(topics, int(dataplaneReq.GetPageSize()), dataplaneReq.GetPageToken(), "name", func(x *v1alpha1.ListTopicsResponse_Topic) string {
+			return x.GetName()
+		})
+		if err != nil {
+			return nil, apierrors.NewConnectError(
+				connect.CodeInternal,
+				fmt.Errorf("failed to apply pagination: %w", err),
+				apierrors.NewErrorInfo(v1.Reason_REASON_CONSOLE_ERROR.String()),
+			)
+		}
+		topics = page
+		nextPageToken = token
+	}
+
+	return connect.NewResponse(&v1alpha1.ListTopicsResponse{Topics: topics, NextPageToken: nextPageToken}), nil
+}

backend/pkg/api/connect/service/topic/v1/console_mapper.go

@@ -0,0 +1,54 @@
+// Copyright 2023 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package topic
+
+import (
+	"github.com/redpanda-data/console/backend/pkg/console"
+	v1alpha1 "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/console/v1alpha1"
+)
+
+// consoleMapper maps console topic summaries to the Console v1alpha1 proto types.
+type consoleMapper struct{}
+
+func (k *consoleMapper) topicSummariesToProto(summaries []*console.TopicSummary) []*v1alpha1.ListTopicsResponse_Topic {
+	topics := make([]*v1alpha1.ListTopicsResponse_Topic, len(summaries))
+	for i, summary := range summaries {
+		topics[i] = k.topicSummaryToProto(summary)
+	}
+
+	return topics
+}
+
+func (k *consoleMapper) topicSummaryToProto(summary *console.TopicSummary) *v1alpha1.ListTopicsResponse_Topic {
+	return &v1alpha1.ListTopicsResponse_Topic{
+		Name:              summary.TopicName,
+		Internal:          summary.IsInternal,
+		PartitionCount:    int32(summary.PartitionCount),
+		ReplicationFactor: int32(summary.ReplicationFactor),
+		CleanupPolicy:     summary.CleanupPolicy,
+		LogDirSummary:     k.topicLogDirSummaryToProto(summary.LogDirSummary),
+	}
+}
+
+func (*consoleMapper) topicLogDirSummaryToProto(summary console.TopicLogDirSummary) *v1alpha1.ListTopicsResponse_LogDirSummary {
+	replicaErrors := make([]*v1alpha1.ListTopicsResponse_ReplicaError, len(summary.ReplicaErrors))
+	for i, replicaError := range summary.ReplicaErrors {
+		replicaErrors[i] = &v1alpha1.ListTopicsResponse_ReplicaError{
+			BrokerId: replicaError.BrokerID,
+			Error:    replicaError.Error,
+		}
+	}
+
+	return &v1alpha1.ListTopicsResponse_LogDirSummary{
+		TotalSizeBytes: summary.TotalSizeBytes,
+		ReplicaErrors:  replicaErrors,
+		Hint:           summary.Hint,
+	}
+}

backend/pkg/api/connect/service/topic/v1/console_mapper_test.go

@@ -0,0 +1,101 @@
+// Copyright 2024 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package topic
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/redpanda-data/console/backend/pkg/console"
+	v1alpha1 "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/console/v1alpha1"
+)
+
+// TestConsoleTopicSummariesToProto guards the Console ListTopics response shape: the cleanup policy
+// and the aggregated log dir summary (size, hint, per-broker replica errors) must survive the
+// mapping from console.TopicSummary. A regression here surfaces as empty cleanup-policy / size
+// columns in the UI. These fields live on the Console API rather than the dataplane one, which
+// deliberately exposes only cheap metadata.
+func TestConsoleTopicSummariesToProto(t *testing.T) {
+	testCases := []struct {
+		name       string
+		input      []*console.TopicSummary
+		validateFn func(t *testing.T, result []*v1alpha1.ListTopicsResponse_Topic)
+	}{
+		{
+			name: "maps cleanup policy and log dir summary",
+			input: []*console.TopicSummary{
+				{
+					TopicName:         "orders",
+					IsInternal:        false,
+					PartitionCount:    3,
+					ReplicationFactor: 2,
+					CleanupPolicy:     "compact",
+					LogDirSummary: console.TopicLogDirSummary{
+						TotalSizeBytes: 2048,
+						Hint:           "partial result",
+						ReplicaErrors: []console.TopicLogDirSummaryReplicaError{
+							{BrokerID: 1, Error: "broker down"},
+						},
+					},
+				},
+			},
+			validateFn: func(t *testing.T, result []*v1alpha1.ListTopicsResponse_Topic) {
+				require.Len(t, result, 1)
+				topic := result[0]
+				assert.Equal(t, "orders", topic.GetName())
+				assert.False(t, topic.GetInternal())
+				assert.Equal(t, int32(3), topic.GetPartitionCount())
+				assert.Equal(t, int32(2), topic.GetReplicationFactor())
+				assert.Equal(t, "compact", topic.GetCleanupPolicy())
+
+				summary := topic.GetLogDirSummary()
+				require.NotNil(t, summary)
+				assert.Equal(t, int64(2048), summary.GetTotalSizeBytes())
+				assert.Equal(t, "partial result", summary.GetHint())
+				require.Len(t, summary.GetReplicaErrors(), 1)
+				assert.Equal(t, int32(1), summary.GetReplicaErrors()[0].GetBrokerId())
+				assert.Equal(t, "broker down", summary.GetReplicaErrors()[0].GetError())
+			},
+		},
+		{
+			name: "preserves order and the N/A cleanup policy fallback",
+			input: []*console.TopicSummary{
+				{TopicName: "a", CleanupPolicy: "delete", LogDirSummary: console.TopicLogDirSummary{TotalSizeBytes: 1}},
+				{TopicName: "b", CleanupPolicy: "N/A", LogDirSummary: console.TopicLogDirSummary{TotalSizeBytes: 2}},
+			},
+			validateFn: func(t *testing.T, result []*v1alpha1.ListTopicsResponse_Topic) {
+				require.Len(t, result, 2)
+				assert.Equal(t, "a", result[0].GetName())
+				assert.Equal(t, "delete", result[0].GetCleanupPolicy())
+				assert.Equal(t, "b", result[1].GetName())
+				assert.Equal(t, "N/A", result[1].GetCleanupPolicy())
+				assert.Empty(t, result[1].GetLogDirSummary().GetReplicaErrors())
+			},
+		},
+		{
+			name:  "empty input yields empty slice",
+			input: []*console.TopicSummary{},
+			validateFn: func(t *testing.T, result []*v1alpha1.ListTopicsResponse_Topic) {
+				assert.Empty(t, result)
+			},
+		},
+	}
+
+	consoleMapper := consoleMapper{}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			out := consoleMapper.topicSummariesToProto(tc.input)
+			tc.validateFn(t, out)
+		})
+	}
+}

backend/pkg/api/routes.go

@@ -124,6 +124,7 @@ func (api *API) setupConnectWithGRPCGateway(r chi.Router) {
 	transformSvcV1 := transformsvcv1.NewService(api.Cfg, loggerpkg.Named(api.Logger, "transform_service"), v, api.RedpandaClientProvider)
 	kafkaConnectSvcV1 := apikafkaconnectsvcv1.NewService(api.Cfg, loggerpkg.Named(api.Logger, "kafka_connect_service"), api.ConnectSvc)
 	consoleTransformSvcV1 := &transformsvcv1.ConsoleService{Impl: transformSvcV1}
+	consoleTopicSvcV1 := topicsvcv1.NewConsoleService(loggerpkg.Named(api.Logger, "topic_service"), api.ConsoleSvc)
 	monitoringSvcV1 := monitoringsvcv1.NewService(api.Cfg, loggerpkg.Named(api.Logger, "monitoring_service"), api.RedpandaClientProvider)
 
 	// v1alpha2
@@ -170,6 +171,7 @@ func (api *API) setupConnectWithGRPCGateway(r chi.Router) {
 			consolev1alpha1connect.SecurityServiceName:       consolev1alpha1connect.UnimplementedSecurityServiceHandler{},
 			consolev1alpha1connect.LicenseServiceName:        licenseSvc,
 			consolev1alpha1connect.TransformServiceName:      consoleTransformSvcV1,
+			consolev1alpha1connect.TopicServiceName:          consoleTopicSvcV1,
 			consolev1alpha1connect.AuthenticationServiceName: &AuthenticationDefaultHandler{},
 			consolev1alpha1connect.ClusterStatusServiceName:  clusterStatusSvc,
 			consolev1alpha1connect.SecretServiceName:         consolev1alpha1connect.UnimplementedSecretServiceHandler{},
@@ -247,6 +249,9 @@ func (api *API) setupConnectWithGRPCGateway(r chi.Router) {
 	consoleTransformSvcPath, consoleTransformSvcHandler := consolev1alpha1connect.NewTransformServiceHandler(
 		hookOutput.Services[consolev1alpha1connect.TransformServiceName].(consolev1alpha1connect.TransformServiceHandler),
 		connect.WithInterceptors(append(hookOutput.Interceptors, sunsetInterceptor)...))
+	consoleTopicSvcPath, consoleTopicSvcHandler := consolev1alpha1connect.NewTopicServiceHandler(
+		hookOutput.Services[consolev1alpha1connect.TopicServiceName].(consolev1alpha1connect.TopicServiceHandler),
+		connect.WithInterceptors(append(hookOutput.Interceptors, sunsetInterceptor)...))
 	licenseSvcPath, licenseSvcHandler := consolev1alpha1connect.NewLicenseServiceHandler(hookOutput.Services[consolev1alpha1connect.LicenseServiceName].(consolev1alpha1connect.LicenseServiceHandler),
 		connect.WithInterceptors(append(hookOutput.Interceptors, sunsetInterceptor)...))
 	authenticationSvcPath, authenticationSvcHandler := consolev1alpha1connect.NewAuthenticationServiceHandler(hookOutput.Services[consolev1alpha1connect.AuthenticationServiceName].(consolev1alpha1connect.AuthenticationServiceHandler),
@@ -355,6 +360,11 @@ func (api *API) setupConnectWithGRPCGateway(r chi.Router) {
 			MountPath:   consoleTransformSvcPath,
 			Handler:     consoleTransformSvcHandler,
 		},
+		{
+			ServiceName: consolev1alpha1connect.TopicServiceName,
+			MountPath:   consoleTopicSvcPath,
+			Handler:     consoleTopicSvcHandler,
+		},
 		{
 			ServiceName: consolev1alpha1connect.LicenseServiceName,
 			MountPath:   licenseSvcPath,