fix(sip): implicitly paginate sip trunk queries to skirt twirp limits

Author: rektdeckard

cmd/lk/proto.go

@@ -26,6 +26,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"github.com/livekit/livekit-cli/v2/pkg/util"
+	"github.com/livekit/protocol/livekit"
 )
 
 const flagRequest = "request"
@@ -44,6 +45,16 @@ type protoTypeValidator[T any] interface {
 	Validate() error
 }
 
+type paginatedType[T any] interface {
+	protoType[T]
+	GetPage() *livekit.Pagination
+}
+
+type paginatedResponseType[T any, D any] interface {
+	protoType[T]
+	GetItems() []*D
+}
+
 func ReadRequest[T any, P protoType[T]](cmd *cli.Command) (*T, error) {
 	return ReadRequestFileOrLiteral[T, P](cmd.String(flagRequest))
 }
@@ -99,6 +110,35 @@ func RequestDesc[T any, _ protoType[T]]() string {
 	return typ + " as JSON file"
 }
 
+// Repeatedly calls a paginated list function until all items are retrieved,
+// requiring the caller to update the request's pagination parameters as needed.
+func ExhaustivePaginatedList[
+	ReqT any, Req paginatedType[ReqT],
+	ResT any, ResD any, Res paginatedResponseType[ResT, ResD],
+](
+	ctx context.Context,
+	req Req,
+	list func(context.Context, Req) (Res, error),
+	iter func(items []*ResD),
+	page *livekit.Pagination,
+) error {
+	for {
+		res, err := list(ctx, req)
+		if err != nil {
+			return err
+		}
+		resultItems := res.GetItems()
+		if len(resultItems) > 0 {
+			iter(resultItems)
+		}
+		// List exhausted
+		if len(resultItems) < int(page.Limit) {
+			break
+		}
+	}
+	return nil
+}
+
 func createAndPrintFile[T any, P protoTypeValidator[T], R any](
 	ctx context.Context,
 	cmd *cli.Command, file string,
@@ -206,7 +246,7 @@ func listAndPrint[
 ](
 	ctx context.Context,
 	cmd *cli.Command,
-	getList func(ctx context.Context, req Req) (Resp, error), req Req,
+	getList func(context.Context, Req) (Resp, error), req Req,
 	header []string, tableRow func(item *T) []string,
 ) error {
 	res, err := getList(ctx, req)

cmd/lk/sip.go

@@ -790,9 +790,33 @@ func listSipTrunk(ctx context.Context, cmd *cli.Command) error {
 func listSipInboundTrunk(ctx context.Context, cmd *cli.Command) error {
 	cli, err := createSIPClient(ctx, cmd)
 	if err != nil {
-		return err
+		return fmt.Errorf("could not create SIP client: %w", err)
+	}
+
+	// NOTE: twirp has a maximum payload size of 4MB, which some customer data may exceed.
+	// We implement pagination here behind the scenes to split requests into manageable chunks
+	// unlikely to exceed the limit. This should be used on all listing commands that may
+	// return a large number of items.
+	page := &livekit.Pagination{Limit: 500}
+	req := &livekit.ListSIPInboundTrunkRequest{Page: page}
+	list := func(ctx context.Context, req *livekit.ListSIPInboundTrunkRequest) (*livekit.ListSIPInboundTrunkResponse, error) {
+		res := &livekit.ListSIPInboundTrunkResponse{}
+		if err := ExhaustivePaginatedList(
+			ctx,
+			req,
+			cli.ListSIPInboundTrunk,
+			func(items []*livekit.SIPInboundTrunkInfo) {
+				res.Items = append(res.Items, items...)
+				page.AfterId = items[len(items)-1].SipTrunkId
+			},
+			page,
+		); err != nil {
+			return nil, fmt.Errorf("could not list SIP inbound trunks: %w", err)
+		}
+		return res, nil
 	}
-	return listAndPrint(ctx, cmd, cli.ListSIPInboundTrunk, &livekit.ListSIPInboundTrunkRequest{}, []string{
+
+	return listAndPrint(ctx, cmd, list, req, []string{
 		"SipTrunkID", "Name", "Numbers",
 		"AllowedAddresses", "AllowedNumbers",
 		"Authentication",
@@ -816,7 +840,31 @@ func listSipOutboundTrunk(ctx context.Context, cmd *cli.Command) error {
 	if err != nil {
 		return err
 	}
-	return listAndPrint(ctx, cmd, cli.ListSIPOutboundTrunk, &livekit.ListSIPOutboundTrunkRequest{}, []string{
+
+	// NOTE: twirp has a maximum payload size of 4MB, which some customer data may exceed.
+	// We implement pagination here behind the scenes to split requests into manageable chunks
+	// unlikely to exceed the limit. This should be used on all listing commands that may
+	// return a large number of items.
+	page := &livekit.Pagination{Limit: 500}
+	req := &livekit.ListSIPOutboundTrunkRequest{Page: page}
+	list := func(ctx context.Context, req *livekit.ListSIPOutboundTrunkRequest) (*livekit.ListSIPOutboundTrunkResponse, error) {
+		res := &livekit.ListSIPOutboundTrunkResponse{}
+		if err := ExhaustivePaginatedList(
+			ctx,
+			req,
+			cli.ListSIPOutboundTrunk,
+			func(items []*livekit.SIPOutboundTrunkInfo) {
+				res.Items = append(res.Items, items...)
+				page.AfterId = items[len(items)-1].SipTrunkId
+			},
+			page,
+		); err != nil {
+			return nil, fmt.Errorf("could not list SIP outbound trunks: %w", err)
+		}
+		return res, nil
+	}
+
+	return listAndPrint(ctx, cmd, list, req, []string{
 		"SipTrunkID", "Name",
 		"Address", "Transport",
 		"Numbers",

pkg/util/pagination.go

@@ -0,0 +1,56 @@
+// Copyright 2023-2024 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package util
+
+import (
+	// "github.com/livekit/protocol/livekit"
+	"google.golang.org/protobuf/proto"
+)
+
+type ProtoType[T any] interface {
+	*T
+	proto.Message
+}
+
+type Res[T any] interface {
+	proto.Message
+	GetItems() []*T
+}
+
+// func FetchExhaustiveWithPagination[ReqT any, Req ProtoType[ReqT]](fetch func(pagination *livekit.Pagination) error, initialPage *livekit.Pagination) error {
+// 	page := initialPage
+//
+// 	for {
+// 		if err := fetch(page); err != nil {
+// 			return err
+// 		}
+// 		if
+// 	}
+//
+// 	var response = livekit.ListSIPInboundTrunkResponse{}
+// 	for {
+// 		subResponse, err := s.sipClient.ListSIPInboundTrunk(ctx, in)
+// 		if err != nil {
+// 			return nil, err
+// 		}
+// 		response.Items = append(response.Items, subResponse.Items...)
+// 		if len(subResponse.Items) < int(in.Page.Limit) {
+// 			break
+// 		}
+// 		in.Page.AfterId = subResponse.Items[len(subResponse.Items)-1].SipTrunkId
+// 	}
+// 	return &response, nil
+//
+// }