diff --git a/backend/go.mod b/backend/go.mod
index 83a9bd528..b0edebe30 100644
--- a/backend/go.mod
+++ b/backend/go.mod
@@ -3,10 +3,12 @@ module github.com/redpanda-data/console/backend
 go 1.20
 
 require (
+	buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.31.0-20230721003620-2341cbb21958.1
 	connectrpc.com/connect v1.11.0
 	connectrpc.com/grpcreflect v1.2.0
 	github.com/basgys/goxml2json v1.1.0
 	github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869
+	github.com/bufbuild/protovalidate-go v0.2.1
 	github.com/cloudhut/common v0.10.0
 	github.com/cloudhut/connect-client v0.0.0-20230417124247-963e5bcdfee7
 	github.com/docker/docker v24.0.4+incompatible
@@ -55,6 +57,7 @@ require (
 	github.com/Microsoft/go-winio v0.6.1 // indirect
 	github.com/ProtonMail/go-crypto v0.0.0-20230717121422-5aa5874ade95 // indirect
 	github.com/acomagu/bufpipe v1.0.4 // indirect
+	github.com/antlr/antlr4/runtime/Go/antlr/v4 v4.0.0-20230512164433-5d1fd1a340c9 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bitly/go-simplejson v0.5.0 // indirect
 	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
@@ -74,6 +77,7 @@ require (
 	github.com/goccy/go-json v0.10.2 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
+	github.com/google/cel-go v0.17.1 // indirect
 	github.com/google/pprof v0.0.0-20230705174524-200ffdc848b8 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
@@ -125,6 +129,7 @@ require (
 	github.com/spf13/afero v1.9.5 // indirect
 	github.com/spf13/cobra v1.7.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/stoewer/go-strcase v1.3.0 // indirect
 	github.com/twmb/tlscfg v1.2.1 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	github.com/xanzy/ssh-agent v0.3.3 // indirect
@@ -136,6 +141,7 @@ require (
 	golang.org/x/term v0.10.0 // indirect
 	golang.org/x/tools v0.11.0 // indirect
 	google.golang.org/genproto v0.0.0-20230717213848-3f92550aa753 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20230706204954-ccb25ca9f130 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230706204954-ccb25ca9f130 // indirect
 	google.golang.org/grpc v1.56.2 // indirect
 	gopkg.in/warnings.v0 v0.1.2 // indirect
diff --git a/backend/go.sum b/backend/go.sum
index a6242b068..696b9424c 100644
--- a/backend/go.sum
+++ b/backend/go.sum
@@ -1,3 +1,5 @@
+buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.31.0-20230721003620-2341cbb21958.1 h1:mnhf3O5uBs95ngTaQbGZfAnoZC0lM6yWkpdgjtqPbNE=
+buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.31.0-20230721003620-2341cbb21958.1/go.mod h1:xafc+XIsTxTy76GJQ1TKgvJWsSugFBqMaN27WhUblew=
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.38.0/go.mod h1:990N+gfupTy94rShfmMCWGDn0LpTmnzTp2qbd1dvSRU=
@@ -60,6 +62,8 @@ github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRF
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
 github.com/anmitsu/go-shlex v0.0.0-20200514113438-38f4b401e2be h1:9AeTilPcZAjCFIImctFaOjnTIavg87rW78vTPkQqLI8=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
+github.com/antlr/antlr4/runtime/Go/antlr/v4 v4.0.0-20230512164433-5d1fd1a340c9 h1:goHVqTbFX3AIo0tzGr14pgfAW2ZfPChKO21Z9MGf/gk=
+github.com/antlr/antlr4/runtime/Go/antlr/v4 v4.0.0-20230512164433-5d1fd1a340c9/go.mod h1:pSwJ0fSY5KhvocuWSx4fz3BA8OrA1bQn+K1Eli3BRwM=
 github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
 github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
@@ -87,6 +91,8 @@ github.com/bitly/go-simplejson v0.5.0 h1:6IH+V8/tVMab511d5bn4M7EwGXZf9Hj6i2xSwkN
 github.com/bitly/go-simplejson v0.5.0/go.mod h1:cXHtHw4XUPsvGaxgjIAn8PhEWG9NfngEKAMDJEczWVA=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
+github.com/bufbuild/protovalidate-go v0.2.1 h1:pJr07sYhliyfj/STAM7hU4J3FKpVeLVKvOBmOTN8j+s=
+github.com/bufbuild/protovalidate-go v0.2.1/go.mod h1:e7XXDtlxj5vlEyAgsrxpzayp4cEMKCSSb8ZCkin+MVA=
 github.com/bwesterb/go-ristretto v1.2.3/go.mod h1:fUIoIZaG73pV5biE2Blr2xEzDoMj7NFEuV9ekS419A0=
 github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
 github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
@@ -154,6 +160,7 @@ github.com/envoyproxy/go-control-plane v0.9.7/go.mod h1:cwu0lG7PUMfa9snN8LXBig5y
 github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
+github.com/envoyproxy/protoc-gen-validate v1.0.2 h1:QkIBuU5k+x7/QXPvPPnWXWlCdaBFApVqftFV6k087DA=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
 github.com/fatih/structs v1.1.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga6PJ7M=
@@ -231,6 +238,8 @@ github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiu
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
+github.com/google/cel-go v0.17.1 h1:s2151PDGy/eqpCI80/8dl4VL3xTkqI/YubXLXCFw0mw=
+github.com/google/cel-go v0.17.1/go.mod h1:HXZKzB0LXqer5lHHgfWAnlYwJaQBDKMjxjulNQzhwhY=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
@@ -537,6 +546,8 @@ github.com/spf13/cobra v1.7.0 h1:hyqWnYt1ZQShIddO5kBpj3vu05/++x6tJ6dg8EC572I=
 github.com/spf13/cobra v1.7.0/go.mod h1:uLxZILRyS/50WlhOIKD7W6V5bgeIt+4sICxh6uRMrb0=
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/stoewer/go-strcase v1.3.0 h1:g0eASXYtp+yvN9fK8sH94oCIk0fau9uV1/ZdJ0AVEzs=
+github.com/stoewer/go-strcase v1.3.0/go.mod h1:fAH5hQ5pehh+j3nZfvwdk2RgEgQjAoM8wodgtPmh1xo=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
@@ -967,6 +978,8 @@ google.golang.org/genproto v0.0.0-20210226172003-ab064af71705/go.mod h1:FWY/as6D
 google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c/go.mod h1:UODoCrxHCcBojKKwX1terBiRUaqAsFqJiF615XL43r0=
 google.golang.org/genproto v0.0.0-20230717213848-3f92550aa753 h1:+VoAg+OKmWaommL56xmZSE2sUK8A7m6SUO7X89F2tbw=
 google.golang.org/genproto v0.0.0-20230717213848-3f92550aa753/go.mod h1:iqkVr8IRpZ53gx1dEnWlCUIEwDWqWARWrbzpasaTNYM=
+google.golang.org/genproto/googleapis/api v0.0.0-20230706204954-ccb25ca9f130 h1:XVeBY8d/FaK4848myy41HBqnDwvxeV3zMZhwN1TvAMU=
+google.golang.org/genproto/googleapis/api v0.0.0-20230706204954-ccb25ca9f130/go.mod h1:mPBs5jNgx2GuQGvFwUvVKqtn6HsUw9nP64BedgvqEsQ=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20230706204954-ccb25ca9f130 h1:2FZP5XuJY9zQyGM5N0rtovnoXjiMUEIUMvw0m9wlpLc=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20230706204954-ccb25ca9f130/go.mod h1:8mL13HKkDa+IuJ8yruA3ci0q+0vsUz4m//+ottjwS5o=
 google.golang.org/grpc v1.14.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw=
diff --git a/backend/pkg/api/api.go b/backend/pkg/api/api.go
index 397c66820..97cacbd1d 100644
--- a/backend/pkg/api/api.go
+++ b/backend/pkg/api/api.go
@@ -14,15 +14,20 @@
 package api
 
 import (
+	"context"
+	"errors"
 	"io/fs"
 	"math"
 
+	connect_go "connectrpc.com/connect"
 	"connectrpc.com/grpcreflect"
+	"github.com/bufbuild/protovalidate-go"
 	"github.com/cloudhut/common/logging"
 	"github.com/cloudhut/common/rest"
 	"go.uber.org/zap"
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/h2c"
+	"google.golang.org/protobuf/reflect/protoreflect"
 
 	"github.com/redpanda-data/console/backend/pkg/config"
 	"github.com/redpanda-data/console/backend/pkg/connect"
@@ -124,8 +129,21 @@ func (api *API) Start() {
 
 	mux := api.routes()
 
+	v, err := protovalidate.New()
+	if err != nil {
+		api.Logger.Fatal("failed to create proto validator", zap.Error(err))
+	}
+
+	interceptors := []connect_go.Interceptor{}
+
+	// we want the actual request validation after all authorization and permission checks
+	interceptors = append(interceptors, NewRequestValidationInterceptor(api.Logger, v))
+
 	// Connect service(s)
-	mux.Mount(consolev1alphaconnect.NewConsoleServiceHandler(api))
+	mux.Mount(consolev1alphaconnect.NewConsoleServiceHandler(
+		api,
+		connect_go.WithInterceptors(interceptors...),
+	))
 
 	// Connect reflection
 	reflector := grpcreflect.NewStaticReflector(consolev1alphaconnect.ConsoleServiceName)
@@ -146,3 +164,26 @@ func (api *API) Start() {
 		api.Logger.Fatal("REST Server returned an error", zap.Error(err))
 	}
 }
+
+func NewRequestValidationInterceptor(logger *zap.Logger, validator *protovalidate.Validator) connect_go.UnaryInterceptorFunc {
+	interceptor := func(next connect_go.UnaryFunc) connect_go.UnaryFunc {
+		return connect_go.UnaryFunc(func(
+			ctx context.Context,
+			req connect_go.AnyRequest,
+		) (connect_go.AnyResponse, error) {
+			msg, ok := req.Any().(protoreflect.ProtoMessage)
+			if !ok {
+				return nil, connect_go.NewError(connect_go.CodeInvalidArgument, errors.New("request is not a protocol buffer message"))
+			}
+
+			err := validator.Validate(msg)
+			if err != nil {
+				return nil, err
+			}
+
+			return next(ctx, req)
+		})
+	}
+
+	return connect_go.UnaryInterceptorFunc(interceptor)
+}
diff --git a/backend/pkg/api/handle_topic_messages.go b/backend/pkg/api/handle_topic_messages.go
index 4ac970251..1d6fb15c7 100644
--- a/backend/pkg/api/handle_topic_messages.go
+++ b/backend/pkg/api/handle_topic_messages.go
@@ -200,7 +200,69 @@ func (api *API) handleGetMessages() http.HandlerFunc {
 }
 
 // ListMessages consumes a Kafka topic and streams the Kafka records back.
-func (api *API) ListMessages(context.Context, *connect.Request[v1alpha.ListMessagesRequest], *connect.ServerStream[v1alpha.ListMessagesResponse]) error {
-	api.Logger.Info("ListMessages handler")
-	return fmt.Errorf("not implemented yet")
+func (api *API) ListMessages(ctx context.Context, req *connect.Request[v1alpha.ListMessagesRequest], stream *connect.ServerStream[v1alpha.ListMessagesResponse]) error {
+	logger := api.Logger
+
+	logger.Info("ListMessages handler")
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	lmq := ListMessagesRequest{
+		TopicName:             req.Msg.GetTopic(),
+		StartOffset:           req.Msg.GetStartOffset(),
+		StartTimestamp:        req.Msg.GetStartTimestamp(),
+		PartitionID:           req.Msg.GetPartitionId(),
+		MaxResults:            int(req.Msg.GetMaxResults()),
+		FilterInterpreterCode: req.Msg.GetFilterInterpreterCode(),
+		Enterprise:            req.Msg.GetEnterprise(),
+	}
+
+	// Check if logged in user is allowed to list messages for the given request
+	canViewMessages, restErr := api.Hooks.Authorization.CanViewTopicMessages(ctx, &lmq)
+	if restErr != nil || !canViewMessages {
+		return connect.NewError(connect.CodePermissionDenied, restErr.Err)
+	}
+
+	if len(lmq.FilterInterpreterCode) > 0 {
+		canUseMessageSearchFilters, restErr := api.Hooks.Authorization.CanUseMessageSearchFilters(ctx, &lmq)
+		if restErr != nil || !canUseMessageSearchFilters {
+			return connect.NewError(connect.CodePermissionDenied, restErr.Err)
+		}
+	}
+
+	interpreterCode, _ := lmq.DecodeInterpreterCode() // Error has been checked in validation function
+
+	// Request messages from kafka and return them once we got all the messages or the context is done
+	listReq := console.ListMessageRequest{
+		TopicName:             lmq.TopicName,
+		PartitionID:           lmq.PartitionID,
+		StartOffset:           lmq.StartOffset,
+		StartTimestamp:        lmq.StartTimestamp,
+		MessageCount:          lmq.MaxResults,
+		FilterInterpreterCode: interpreterCode,
+	}
+	// api.Hooks.Authorization.PrintListMessagesAuditLog(r, &listReq)
+
+	// Use 30min duration if we want to search a whole topic or forward messages as they arrive
+	duration := 45 * time.Second
+	if listReq.FilterInterpreterCode != "" || listReq.StartOffset == console.StartOffsetNewest {
+		duration = 30 * time.Minute
+	}
+
+	childCtx, cancel := context.WithTimeout(ctx, duration)
+	defer cancel()
+
+	progress := &streamProgressReporter{
+		ctx:              childCtx,
+		logger:           api.Logger,
+		request:          &listReq,
+		stream:           stream,
+		statsMutex:       &sync.RWMutex{},
+		messagesConsumed: 0,
+		bytesConsumed:    0,
+	}
+	progress.Start()
+
+	return api.ConsoleSvc.ListMessages(childCtx, listReq, progress)
 }
diff --git a/backend/pkg/api/stream_progress_reporter.go b/backend/pkg/api/stream_progress_reporter.go
new file mode 100644
index 000000000..a6a7c1ca3
--- /dev/null
+++ b/backend/pkg/api/stream_progress_reporter.go
@@ -0,0 +1,90 @@
+// Copyright 2022 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file https://github.com/redpanda-data/redpanda/blob/dev/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 api
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"connectrpc.com/connect"
+	"go.uber.org/zap"
+
+	"github.com/redpanda-data/console/backend/pkg/console"
+	"github.com/redpanda-data/console/backend/pkg/kafka"
+	v1alpha "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/console/v1alpha"
+)
+
+// streamProgressReporter is in charge of sending status updates and messages regularly to the frontend.
+type streamProgressReporter struct {
+	ctx     context.Context
+	logger  *zap.Logger
+	request *console.ListMessageRequest
+	stream  *connect.ServerStream[v1alpha.ListMessagesResponse]
+
+	statsMutex       *sync.RWMutex
+	messagesConsumed int64
+	bytesConsumed    int64
+}
+
+func (p *streamProgressReporter) Start() {
+	// If search is disabled do not report progress regularly as each consumed message will be sent through the socket
+	// anyways
+	if p.request.FilterInterpreterCode == "" {
+		return
+	}
+
+	// Report the current progress every second to the user. If there's a search request which has to browse a whole
+	// topic it may take some time until there are messages. This go routine is in charge of keeping the user up to
+	// date about the progress Kowl made streaming the topic
+	go func() {
+		for {
+			select {
+			case <-p.ctx.Done():
+				return
+			default:
+				p.reportProgress()
+			}
+			time.Sleep(1 * time.Second)
+		}
+	}()
+}
+
+func (p *streamProgressReporter) reportProgress() {
+}
+
+func (p *streamProgressReporter) OnPhase(name string) {
+}
+
+func (p *streamProgressReporter) OnMessageConsumed(size int64) {
+}
+
+func (p *streamProgressReporter) OnMessage(message *kafka.TopicMessage) {
+	p.stream.Send(&v1alpha.ListMessagesResponse{
+		Value: &v1alpha.KafkaRecordPayload{
+			OriginalPayload:     message.Value.Payload.Payload,
+			PayloadSize:         int32(message.Value.Size),
+			DeserializedPayload: message.Value.Payload.Payload,
+			IsPayloadTooLarge:   false, // TODO check for size
+		},
+		Key: &v1alpha.KafkaRecordPayload{
+			OriginalPayload:     message.Key.Payload.Payload,
+			PayloadSize:         int32(message.Key.Size),
+			DeserializedPayload: message.Key.Payload.Payload,
+			IsPayloadTooLarge:   false, // TODO check for size
+		},
+	})
+}
+
+func (p *streamProgressReporter) OnComplete(elapsedMs int64, isCancelled bool) {
+}
+
+func (p *streamProgressReporter) OnError(message string) {
+}
diff --git a/backend/buf.gen.yaml b/backend/pkg/kafka/testdata/proto/buf.gen.yaml
similarity index 100%
rename from backend/buf.gen.yaml
rename to backend/pkg/kafka/testdata/proto/buf.gen.yaml
diff --git a/backend/buf.yaml b/backend/pkg/kafka/testdata/proto/buf.yaml
similarity index 100%
rename from backend/buf.yaml
rename to backend/pkg/kafka/testdata/proto/buf.yaml
diff --git a/backend/pkg/protogen/redpanda/api/console/v1alpha/consolev1alphaconnect/list_messages.connect.go b/backend/pkg/protogen/redpanda/api/console/v1alpha/consolev1alphaconnect/list_messages.connect.go
index 5afb57ed4..a80da5103 100644
--- a/backend/pkg/protogen/redpanda/api/console/v1alpha/consolev1alphaconnect/list_messages.connect.go
+++ b/backend/pkg/protogen/redpanda/api/console/v1alpha/consolev1alphaconnect/list_messages.connect.go
@@ -5,12 +5,14 @@
 package consolev1alphaconnect
 
 import (
-	connect "connectrpc.com/connect"
 	context "context"
 	errors "errors"
-	v1alpha "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/console/v1alpha"
 	http "net/http"
 	strings "strings"
+
+	connect "connectrpc.com/connect"
+
+	v1alpha "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/console/v1alpha"
 )
 
 // This is a compile-time assertion to ensure that this generated file and the connect package are
@@ -40,6 +42,7 @@ const (
 
 // ConsoleServiceClient is a client for the redpanda.api.console.v1alpha.ConsoleService service.
 type ConsoleServiceClient interface {
+	// ListMessages lists the messages according to the requested query.
 	ListMessages(context.Context, *connect.Request[v1alpha.ListMessagesRequest]) (*connect.ServerStreamForClient[v1alpha.ListMessagesResponse], error)
 }
 
@@ -74,6 +77,7 @@ func (c *consoleServiceClient) ListMessages(ctx context.Context, req *connect.Re
 // ConsoleServiceHandler is an implementation of the redpanda.api.console.v1alpha.ConsoleService
 // service.
 type ConsoleServiceHandler interface {
+	// ListMessages lists the messages according to the requested query.
 	ListMessages(context.Context, *connect.Request[v1alpha.ListMessagesRequest], *connect.ServerStream[v1alpha.ListMessagesResponse]) error
 }
 
diff --git a/backend/pkg/protogen/redpanda/api/console/v1alpha/list_messages.pb.go b/backend/pkg/protogen/redpanda/api/console/v1alpha/list_messages.pb.go
index 58eab5048..ffb25ab1d 100644
--- a/backend/pkg/protogen/redpanda/api/console/v1alpha/list_messages.pb.go
+++ b/backend/pkg/protogen/redpanda/api/console/v1alpha/list_messages.pb.go
@@ -7,10 +7,12 @@
 package consolev1alpha
 
 import (
-	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
-	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
 	reflect "reflect"
 	sync "sync"
+
+	_ "buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go/buf/validate"
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
 )
 
 const (
@@ -20,12 +22,19 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
+// ListMessagesRequest is the request for ListMessages call.
 type ListMessagesRequest struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	Topic string `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"`
+	Topic                 string `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"`                                                                // Topic name.
+	StartOffset           int64  `protobuf:"zigzag64,2,opt,name=start_offset,json=startOffset,proto3" json:"start_offset,omitempty"`                              // Start offset. -1 for recent (newest - results), -2 for oldest offset, -3 for newest, -4 for timestamp.
+	StartTimestamp        int64  `protobuf:"varint,3,opt,name=start_timestamp,json=startTimestamp,proto3" json:"start_timestamp,omitempty"`                       // Start offset by unix timestamp in ms (only considered if start offset is set to -4).
+	PartitionId           int32  `protobuf:"varint,4,opt,name=partition_id,json=partitionId,proto3" json:"partition_id,omitempty"`                                // -1 for all partition ids
+	MaxResults            int32  `protobuf:"varint,5,opt,name=max_results,json=maxResults,proto3" json:"max_results,omitempty"`                                   // Maximum number of results
+	FilterInterpreterCode string `protobuf:"bytes,6,opt,name=filter_interpreter_code,json=filterInterpreterCode,proto3" json:"filter_interpreter_code,omitempty"` // Base64 encoded code
+	Enterprise            []byte `protobuf:"bytes,7,opt,name=enterprise,proto3" json:"enterprise,omitempty"`                                                      // Enterprise may only be set in the Enterprise mode. The JSON deserialization is deferred.
 }
 
 func (x *ListMessagesRequest) Reset() {
@@ -67,6 +76,49 @@ func (x *ListMessagesRequest) GetTopic() string {
 	return ""
 }
 
+func (x *ListMessagesRequest) GetStartOffset() int64 {
+	if x != nil {
+		return x.StartOffset
+	}
+	return 0
+}
+
+func (x *ListMessagesRequest) GetStartTimestamp() int64 {
+	if x != nil {
+		return x.StartTimestamp
+	}
+	return 0
+}
+
+func (x *ListMessagesRequest) GetPartitionId() int32 {
+	if x != nil {
+		return x.PartitionId
+	}
+	return 0
+}
+
+func (x *ListMessagesRequest) GetMaxResults() int32 {
+	if x != nil {
+		return x.MaxResults
+	}
+	return 0
+}
+
+func (x *ListMessagesRequest) GetFilterInterpreterCode() string {
+	if x != nil {
+		return x.FilterInterpreterCode
+	}
+	return ""
+}
+
+func (x *ListMessagesRequest) GetEnterprise() []byte {
+	if x != nil {
+		return x.Enterprise
+	}
+	return nil
+}
+
+// ListMessagesResponse is the response for ListMessages call.
 type ListMessagesResponse struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -122,6 +174,7 @@ func (x *ListMessagesResponse) GetValue() *KafkaRecordPayload {
 	return nil
 }
 
+// KafkaRecordPayload is record payload representation.
 type KafkaRecordPayload struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -201,60 +254,79 @@ var file_redpanda_api_console_v1alpha_list_messages_proto_rawDesc = []byte{
 	0x69, 0x73, 0x74, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x70, 0x72, 0x6f,
 	0x74, 0x6f, 0x12, 0x1c, 0x72, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x2e, 0x61, 0x70, 0x69,
 	0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61,
-	0x22, 0x2b, 0x0a, 0x13, 0x4c, 0x69, 0x73, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73,
-	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x22, 0xa2, 0x01,
-	0x0a, 0x14, 0x4c, 0x69, 0x73, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x52, 0x65,
-	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x42, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20,
-	0x01, 0x28, 0x0b, 0x32, 0x30, 0x2e, 0x72, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x2e, 0x61,
-	0x70, 0x69, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70,
-	0x68, 0x61, 0x2e, 0x4b, 0x61, 0x66, 0x6b, 0x61, 0x52, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x50, 0x61,
-	0x79, 0x6c, 0x6f, 0x61, 0x64, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x46, 0x0a, 0x05, 0x76, 0x61,
-	0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x30, 0x2e, 0x72, 0x65, 0x64, 0x70,
-	0x61, 0x6e, 0x64, 0x61, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65,
-	0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x2e, 0x4b, 0x61, 0x66, 0x6b, 0x61, 0x52, 0x65,
-	0x63, 0x6f, 0x72, 0x64, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x52, 0x05, 0x76, 0x61, 0x6c,
-	0x75, 0x65, 0x22, 0xc6, 0x01, 0x0a, 0x12, 0x4b, 0x61, 0x66, 0x6b, 0x61, 0x52, 0x65, 0x63, 0x6f,
-	0x72, 0x64, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x12, 0x29, 0x0a, 0x10, 0x6f, 0x72, 0x69,
-	0x67, 0x69, 0x6e, 0x61, 0x6c, 0x5f, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18, 0x01, 0x20,
-	0x01, 0x28, 0x0c, 0x52, 0x0f, 0x6f, 0x72, 0x69, 0x67, 0x69, 0x6e, 0x61, 0x6c, 0x50, 0x61, 0x79,
-	0x6c, 0x6f, 0x61, 0x64, 0x12, 0x21, 0x0a, 0x0c, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x5f,
-	0x73, 0x69, 0x7a, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0b, 0x70, 0x61, 0x79, 0x6c,
-	0x6f, 0x61, 0x64, 0x53, 0x69, 0x7a, 0x65, 0x12, 0x31, 0x0a, 0x14, 0x64, 0x65, 0x73, 0x65, 0x72,
-	0x69, 0x61, 0x6c, 0x69, 0x7a, 0x65, 0x64, 0x5f, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18,
-	0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x13, 0x64, 0x65, 0x73, 0x65, 0x72, 0x69, 0x61, 0x6c, 0x69,
-	0x7a, 0x65, 0x64, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x12, 0x2f, 0x0a, 0x14, 0x69, 0x73,
-	0x5f, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x5f, 0x74, 0x6f, 0x6f, 0x5f, 0x6c, 0x61, 0x72,
-	0x67, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x08, 0x52, 0x11, 0x69, 0x73, 0x50, 0x61, 0x79, 0x6c,
-	0x6f, 0x61, 0x64, 0x54, 0x6f, 0x6f, 0x4c, 0x61, 0x72, 0x67, 0x65, 0x32, 0x8b, 0x01, 0x0a, 0x0e,
-	0x43, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x79,
-	0x0a, 0x0c, 0x4c, 0x69, 0x73, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x12, 0x31,
-	0x2e, 0x72, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x63, 0x6f,
-	0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x2e, 0x4c, 0x69,
-	0x73, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
-	0x74, 0x1a, 0x32, 0x2e, 0x72, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x2e, 0x61, 0x70, 0x69,
+	0x1a, 0x1b, 0x62, 0x75, 0x66, 0x2f, 0x76, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x2f, 0x76,
+	0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xb5, 0x02,
+	0x0a, 0x13, 0x4c, 0x69, 0x73, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x52, 0x65,
+	0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x20, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x09, 0x42, 0x0a, 0xba, 0x48, 0x07, 0x72, 0x05, 0x10, 0x01, 0x18, 0x80, 0x01,
+	0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x2e, 0x0a, 0x0c, 0x73, 0x74, 0x61, 0x72, 0x74,
+	0x5f, 0x6f, 0x66, 0x66, 0x73, 0x65, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x12, 0x42, 0x0b, 0xba,
+	0x48, 0x08, 0x3a, 0x06, 0x32, 0x04, 0x01, 0x03, 0x05, 0x07, 0x52, 0x0b, 0x73, 0x74, 0x61, 0x72,
+	0x74, 0x4f, 0x66, 0x66, 0x73, 0x65, 0x74, 0x12, 0x27, 0x0a, 0x0f, 0x73, 0x74, 0x61, 0x72, 0x74,
+	0x5f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x03,
+	0x52, 0x0e, 0x73, 0x74, 0x61, 0x72, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70,
+	0x12, 0x2a, 0x0a, 0x0c, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x64,
+	0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x42, 0x07, 0xba, 0x48, 0x04, 0x3a, 0x02, 0x28, 0x01, 0x52,
+	0x0b, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x1f, 0x0a, 0x0b,
+	0x6d, 0x61, 0x78, 0x5f, 0x72, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73, 0x18, 0x05, 0x20, 0x01, 0x28,
+	0x05, 0x52, 0x0a, 0x6d, 0x61, 0x78, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73, 0x12, 0x36, 0x0a,
+	0x17, 0x66, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x5f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x70, 0x72, 0x65,
+	0x74, 0x65, 0x72, 0x5f, 0x63, 0x6f, 0x64, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x15,
+	0x66, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x49, 0x6e, 0x74, 0x65, 0x72, 0x70, 0x72, 0x65, 0x74, 0x65,
+	0x72, 0x43, 0x6f, 0x64, 0x65, 0x12, 0x1e, 0x0a, 0x0a, 0x65, 0x6e, 0x74, 0x65, 0x72, 0x70, 0x72,
+	0x69, 0x73, 0x65, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0a, 0x65, 0x6e, 0x74, 0x65, 0x72,
+	0x70, 0x72, 0x69, 0x73, 0x65, 0x22, 0xa2, 0x01, 0x0a, 0x14, 0x4c, 0x69, 0x73, 0x74, 0x4d, 0x65,
+	0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x42,
+	0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x30, 0x2e, 0x72, 0x65,
+	0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x6f,
+	0x6c, 0x65, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x2e, 0x4b, 0x61, 0x66, 0x6b, 0x61,
+	0x52, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x52, 0x03, 0x6b,
+	0x65, 0x79, 0x12, 0x46, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28,
+	0x0b, 0x32, 0x30, 0x2e, 0x72, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x2e, 0x61, 0x70, 0x69,
 	0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61,
-	0x2e, 0x4c, 0x69, 0x73, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x52, 0x65, 0x73,
-	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x30, 0x01, 0x42, 0xab, 0x02, 0x0a, 0x20, 0x63, 0x6f,
-	0x6d, 0x2e, 0x72, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x63,
-	0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x42, 0x11,
-	0x4c, 0x69, 0x73, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x50, 0x72, 0x6f, 0x74,
-	0x6f, 0x50, 0x01, 0x5a, 0x61, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f,
-	0x72, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x2d, 0x64, 0x61, 0x74, 0x61, 0x2f, 0x63, 0x6f,
-	0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x2f, 0x62, 0x61, 0x63, 0x6b, 0x65, 0x6e, 0x64, 0x2f, 0x70, 0x6b,
-	0x67, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x67, 0x65, 0x6e, 0x2f, 0x72, 0x65, 0x64, 0x70, 0x61,
-	0x6e, 0x64, 0x61, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x63, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x2f,
-	0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x3b, 0x63, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x76,
-	0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0xa2, 0x02, 0x03, 0x52, 0x41, 0x43, 0xaa, 0x02, 0x1c, 0x52,
-	0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x2e, 0x41, 0x70, 0x69, 0x2e, 0x43, 0x6f, 0x6e, 0x73,
-	0x6f, 0x6c, 0x65, 0x2e, 0x56, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0xca, 0x02, 0x1c, 0x52, 0x65,
-	0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x5c, 0x41, 0x70, 0x69, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x6f,
-	0x6c, 0x65, 0x5c, 0x56, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0xe2, 0x02, 0x28, 0x52, 0x65, 0x64,
-	0x70, 0x61, 0x6e, 0x64, 0x61, 0x5c, 0x41, 0x70, 0x69, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x6f, 0x6c,
-	0x65, 0x5c, 0x56, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74,
-	0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x1f, 0x52, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61,
-	0x3a, 0x3a, 0x41, 0x70, 0x69, 0x3a, 0x3a, 0x43, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x3a, 0x3a,
-	0x56, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x2e, 0x4b, 0x61, 0x66, 0x6b, 0x61, 0x52, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x50, 0x61, 0x79, 0x6c,
+	0x6f, 0x61, 0x64, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, 0xc6, 0x01, 0x0a, 0x12, 0x4b,
+	0x61, 0x66, 0x6b, 0x61, 0x52, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61,
+	0x64, 0x12, 0x29, 0x0a, 0x10, 0x6f, 0x72, 0x69, 0x67, 0x69, 0x6e, 0x61, 0x6c, 0x5f, 0x70, 0x61,
+	0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0f, 0x6f, 0x72, 0x69,
+	0x67, 0x69, 0x6e, 0x61, 0x6c, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x12, 0x21, 0x0a, 0x0c,
+	0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x5f, 0x73, 0x69, 0x7a, 0x65, 0x18, 0x02, 0x20, 0x01,
+	0x28, 0x05, 0x52, 0x0b, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x53, 0x69, 0x7a, 0x65, 0x12,
+	0x31, 0x0a, 0x14, 0x64, 0x65, 0x73, 0x65, 0x72, 0x69, 0x61, 0x6c, 0x69, 0x7a, 0x65, 0x64, 0x5f,
+	0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x13, 0x64,
+	0x65, 0x73, 0x65, 0x72, 0x69, 0x61, 0x6c, 0x69, 0x7a, 0x65, 0x64, 0x50, 0x61, 0x79, 0x6c, 0x6f,
+	0x61, 0x64, 0x12, 0x2f, 0x0a, 0x14, 0x69, 0x73, 0x5f, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64,
+	0x5f, 0x74, 0x6f, 0x6f, 0x5f, 0x6c, 0x61, 0x72, 0x67, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x08,
+	0x52, 0x11, 0x69, 0x73, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x54, 0x6f, 0x6f, 0x4c, 0x61,
+	0x72, 0x67, 0x65, 0x32, 0x8b, 0x01, 0x0a, 0x0e, 0x43, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x53,
+	0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x79, 0x0a, 0x0c, 0x4c, 0x69, 0x73, 0x74, 0x4d, 0x65,
+	0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x12, 0x31, 0x2e, 0x72, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64,
+	0x61, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x2e, 0x76, 0x31,
+	0x61, 0x6c, 0x70, 0x68, 0x61, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67,
+	0x65, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x32, 0x2e, 0x72, 0x65, 0x64, 0x70,
+	0x61, 0x6e, 0x64, 0x61, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65,
+	0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x4d, 0x65, 0x73,
+	0x73, 0x61, 0x67, 0x65, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x30,
+	0x01, 0x42, 0xab, 0x02, 0x0a, 0x20, 0x63, 0x6f, 0x6d, 0x2e, 0x72, 0x65, 0x64, 0x70, 0x61, 0x6e,
+	0x64, 0x61, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x2e, 0x76,
+	0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x42, 0x11, 0x4c, 0x69, 0x73, 0x74, 0x4d, 0x65, 0x73, 0x73,
+	0x61, 0x67, 0x65, 0x73, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x61, 0x67, 0x69, 0x74,
+	0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x72, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61,
+	0x2d, 0x64, 0x61, 0x74, 0x61, 0x2f, 0x63, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x2f, 0x62, 0x61,
+	0x63, 0x6b, 0x65, 0x6e, 0x64, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x67,
+	0x65, 0x6e, 0x2f, 0x72, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x2f, 0x61, 0x70, 0x69, 0x2f,
+	0x63, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x2f, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x3b,
+	0x63, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0xa2, 0x02,
+	0x03, 0x52, 0x41, 0x43, 0xaa, 0x02, 0x1c, 0x52, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x2e,
+	0x41, 0x70, 0x69, 0x2e, 0x43, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x2e, 0x56, 0x31, 0x61, 0x6c,
+	0x70, 0x68, 0x61, 0xca, 0x02, 0x1c, 0x52, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x5c, 0x41,
+	0x70, 0x69, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x5c, 0x56, 0x31, 0x61, 0x6c, 0x70,
+	0x68, 0x61, 0xe2, 0x02, 0x28, 0x52, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x5c, 0x41, 0x70,
+	0x69, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x5c, 0x56, 0x31, 0x61, 0x6c, 0x70, 0x68,
+	0x61, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x1f,
+	0x52, 0x65, 0x64, 0x70, 0x61, 0x6e, 0x64, 0x61, 0x3a, 0x3a, 0x41, 0x70, 0x69, 0x3a, 0x3a, 0x43,
+	0x6f, 0x6e, 0x73, 0x6f, 0x6c, 0x65, 0x3a, 0x3a, 0x56, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x62,
+	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
diff --git a/frontend/src/protogen/redpanda/api/console/v1alpha/list_messages_connect.ts b/frontend/src/protogen/redpanda/api/console/v1alpha/list_messages_connect.ts
index 562385684..df45f5b2c 100644
--- a/frontend/src/protogen/redpanda/api/console/v1alpha/list_messages_connect.ts
+++ b/frontend/src/protogen/redpanda/api/console/v1alpha/list_messages_connect.ts
@@ -7,12 +7,16 @@ import { ListMessagesRequest, ListMessagesResponse } from "./list_messages_pb";
 import { MethodKind } from "@bufbuild/protobuf";
 
 /**
+ * ConsoleService represents the Console API service.
+ *
  * @generated from service redpanda.api.console.v1alpha.ConsoleService
  */
 export const ConsoleService = {
   typeName: "redpanda.api.console.v1alpha.ConsoleService",
   methods: {
     /**
+     * ListMessages lists the messages according to the requested query.
+     *
      * @generated from rpc redpanda.api.console.v1alpha.ConsoleService.ListMessages
      */
     listMessages: {
diff --git a/frontend/src/protogen/redpanda/api/console/v1alpha/list_messages_pb.ts b/frontend/src/protogen/redpanda/api/console/v1alpha/list_messages_pb.ts
index bca0980d3..d1a2671ed 100644
--- a/frontend/src/protogen/redpanda/api/console/v1alpha/list_messages_pb.ts
+++ b/frontend/src/protogen/redpanda/api/console/v1alpha/list_messages_pb.ts
@@ -4,17 +4,63 @@
 // @ts-nocheck
 
 import type { BinaryReadOptions, FieldList, JsonReadOptions, JsonValue, PartialMessage, PlainMessage } from "@bufbuild/protobuf";
-import { Message, proto3 } from "@bufbuild/protobuf";
+import { Message, proto3, protoInt64 } from "@bufbuild/protobuf";
 
 /**
+ * ListMessagesRequest is the request for ListMessages call.
+ *
  * @generated from message redpanda.api.console.v1alpha.ListMessagesRequest
  */
 export class ListMessagesRequest extends Message<ListMessagesRequest> {
   /**
+   * Topic name.
+   *
    * @generated from field: string topic = 1;
    */
   topic = "";
 
+  /**
+   * Start offset. -1 for recent (newest - results), -2 for oldest offset, -3 for newest, -4 for timestamp.
+   *
+   * @generated from field: sint64 start_offset = 2;
+   */
+  startOffset = protoInt64.zero;
+
+  /**
+   * Start offset by unix timestamp in ms (only considered if start offset is set to -4).
+   *
+   * @generated from field: int64 start_timestamp = 3;
+   */
+  startTimestamp = protoInt64.zero;
+
+  /**
+   * -1 for all partition ids
+   *
+   * @generated from field: int32 partition_id = 4;
+   */
+  partitionId = 0;
+
+  /**
+   * Maximum number of results
+   *
+   * @generated from field: int32 max_results = 5;
+   */
+  maxResults = 0;
+
+  /**
+   * Base64 encoded code
+   *
+   * @generated from field: string filter_interpreter_code = 6;
+   */
+  filterInterpreterCode = "";
+
+  /**
+   * Enterprise may only be set in the Enterprise mode. The JSON deserialization is deferred.
+   *
+   * @generated from field: bytes enterprise = 7;
+   */
+  enterprise = new Uint8Array(0);
+
   constructor(data?: PartialMessage<ListMessagesRequest>) {
     super();
     proto3.util.initPartial(data, this);
@@ -24,6 +70,12 @@ export class ListMessagesRequest extends Message<ListMessagesRequest> {
   static readonly typeName = "redpanda.api.console.v1alpha.ListMessagesRequest";
   static readonly fields: FieldList = proto3.util.newFieldList(() => [
     { no: 1, name: "topic", kind: "scalar", T: 9 /* ScalarType.STRING */ },
+    { no: 2, name: "start_offset", kind: "scalar", T: 18 /* ScalarType.SINT64 */ },
+    { no: 3, name: "start_timestamp", kind: "scalar", T: 3 /* ScalarType.INT64 */ },
+    { no: 4, name: "partition_id", kind: "scalar", T: 5 /* ScalarType.INT32 */ },
+    { no: 5, name: "max_results", kind: "scalar", T: 5 /* ScalarType.INT32 */ },
+    { no: 6, name: "filter_interpreter_code", kind: "scalar", T: 9 /* ScalarType.STRING */ },
+    { no: 7, name: "enterprise", kind: "scalar", T: 12 /* ScalarType.BYTES */ },
   ]);
 
   static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): ListMessagesRequest {
@@ -44,6 +96,8 @@ export class ListMessagesRequest extends Message<ListMessagesRequest> {
 }
 
 /**
+ * ListMessagesResponse is the response for ListMessages call.
+ *
  * @generated from message redpanda.api.console.v1alpha.ListMessagesResponse
  */
 export class ListMessagesResponse extends Message<ListMessagesResponse> {
@@ -87,6 +141,8 @@ export class ListMessagesResponse extends Message<ListMessagesResponse> {
 }
 
 /**
+ * KafkaRecordPayload is record payload representation.
+ *
  * @generated from message redpanda.api.console.v1alpha.KafkaRecordPayload
  */
 export class KafkaRecordPayload extends Message<KafkaRecordPayload> {
diff --git a/proto/buf.lock b/proto/buf.lock
new file mode 100644
index 000000000..10446a630
--- /dev/null
+++ b/proto/buf.lock
@@ -0,0 +1,13 @@
+# Generated by buf. DO NOT EDIT.
+version: v1
+deps:
+  - remote: buf.build
+    owner: bufbuild
+    repository: protovalidate
+    commit: 1c33ebd9ecfa4653b8c1e662f84d1acf
+    digest: shake256:4e3d93b2859a799ed4c08f530c62da1647cb85adacb3ff085dff6959a8d0a73989f3b8527145ee8b0783cd4e5264eaf0c0a76c206404f9520fce608b525559cb
+  - remote: buf.build
+    owner: googleapis
+    repository: googleapis
+    commit: 711e289f6a384c4caeebaff7c6931ade
+    digest: shake256:e08fb55dad7469f69df00304eed31427d2d1576e9aab31e6bf86642688e04caaf0372f15fe6974cf79432779a635b3ea401ca69c943976dc42749524e4c25d94
diff --git a/proto/redpanda/api/console/v1alpha/list_messages.proto b/proto/redpanda/api/console/v1alpha/list_messages.proto
index fb86ed38a..ff47732bb 100644
--- a/proto/redpanda/api/console/v1alpha/list_messages.proto
+++ b/proto/redpanda/api/console/v1alpha/list_messages.proto
@@ -2,15 +2,33 @@ syntax = "proto3";
 
 package redpanda.api.console.v1alpha;
 
+import "buf/validate/validate.proto";
+
+// ListMessagesRequest is the request for ListMessages call.
 message ListMessagesRequest {
-  string topic = 1;
+  string topic = 1 [
+    (buf.validate.field).string.min_len = 1,
+    (buf.validate.field).string.max_len = 128
+  ]; // Topic name.
+
+  sint64 start_offset = 2 [
+    (buf.validate.field).sint32 = { in: [-1, -2, -3, -4] }
+  ]; // Start offset. -1 for recent (newest - results), -2 for oldest offset, -3 for newest, -4 for timestamp.
+
+  int64 start_timestamp = 3; // Start offset by unix timestamp in ms (only considered if start offset is set to -4).
+  int32 partition_id = 4 [(buf.validate.field).sint32 = { gte: -1 }]; // -1 for all partition ids
+  int32 max_results = 5; // Maximum number of results
+  string filter_interpreter_code = 6; // Base64 encoded code
+  bytes enterprise = 7; // Enterprise may only be set in the Enterprise mode. The JSON deserialization is deferred.
 }
 
+// ListMessagesResponse is the response for ListMessages call.
 message ListMessagesResponse {
   KafkaRecordPayload key = 1;
   KafkaRecordPayload value = 2;
 }
 
+// KafkaRecordPayload is record payload representation.
 message KafkaRecordPayload {
   bytes original_payload = 1;
   int32 payload_size = 2;
@@ -18,6 +36,8 @@ message KafkaRecordPayload {
   bool is_payload_too_large = 4;
 }
 
+// ConsoleService represents the Console API service.
 service ConsoleService {
+  // ListMessages lists the messages according to the requested query.
   rpc ListMessages(ListMessagesRequest) returns (stream ListMessagesResponse) {}
 }
diff --git a/taskfiles/backend.yaml b/taskfiles/backend.yaml
index fcb5fa137..bec44b444 100644
--- a/taskfiles/backend.yaml
+++ b/taskfiles/backend.yaml
@@ -121,10 +121,10 @@ tasks:
         {{ .BUILD_ROOT }}/bin/buf format -w pkg/kafka/testdata/proto
       - |
         export PATH=$PATH:{{ .BUILD_ROOT }}/bin/go
-        {{ .BUILD_ROOT }}/bin/buf lint --config buf.yaml pkg/kafka/testdata/proto
+        {{ .BUILD_ROOT }}/bin/buf lint --config pkg/kafka/testdata/proto/buf.yaml pkg/kafka/testdata/proto
       - |
         export PATH=$PATH:{{ .BUILD_ROOT }}/bin/go
-        {{ .BUILD_ROOT }}/bin/buf generate pkg/kafka/testdata/proto
+        {{ .BUILD_ROOT }}/bin/buf generate --template pkg/kafka/testdata/proto/buf.gen.yaml pkg/kafka/testdata/proto
       - task: 'fmt'
 
   test-unit:
diff --git a/taskfiles/proto.yaml b/taskfiles/proto.yaml
index d409ce52f..eabfc5ca4 100644
--- a/taskfiles/proto.yaml
+++ b/taskfiles/proto.yaml
@@ -30,7 +30,7 @@ tasks:
       # Delete previously generated files
       - rm -rf {{.BACKEND_ROOT}}/internal/protogen
       - rm -rf {{.FRONTEND_ROOT}}/src/protogen
-      - PATH={{.BUILD_ROOT}}/bin:$PATH buf generate --include-imports
+      - PATH={{.BUILD_ROOT}}/bin:$PATH buf generate
       - if [[ $CI == "true" ]]; then git diff --exit-code; fi
 
   install-buf: