From cce783019f285205d749c00bad88d1506fe5ad8b Mon Sep 17 00:00:00 2001 From: Albert Wu Date: Mon, 13 Jul 2026 15:05:49 -0700 Subject: [PATCH] feat(gateway): add queue request listing --- api/submitqueue/gateway/proto/gateway.proto | 25 ++ api/submitqueue/gateway/protopb/gateway.pb.go | 233 +++++++++++++++--- .../gateway/protopb/gateway.pb.yarpc.go | 158 ++++++++---- .../gateway/protopb/gateway_grpc.pb.go | 40 +++ service/submitqueue/gateway/server/main.go | 12 + .../gateway/server/mapper/BUILD.bazel | 2 + .../submitqueue/gateway/server/mapper/list.go | 56 +++++ .../gateway/server/mapper/list_test.go | 60 +++++ submitqueue/entity/BUILD.bazel | 1 + submitqueue/entity/list.go | 37 +++ submitqueue/gateway/controller/BUILD.bazel | 2 + submitqueue/gateway/controller/list.go | 174 +++++++++++++ submitqueue/gateway/controller/list_test.go | 137 ++++++++++ .../submitqueue/gateway/suite_test.go | 46 +++- 14 files changed, 891 insertions(+), 92 deletions(-) create mode 100644 service/submitqueue/gateway/server/mapper/list.go create mode 100644 service/submitqueue/gateway/server/mapper/list_test.go create mode 100644 submitqueue/entity/list.go create mode 100644 submitqueue/gateway/controller/list.go create mode 100644 submitqueue/gateway/controller/list_test.go diff --git a/api/submitqueue/gateway/proto/gateway.proto b/api/submitqueue/gateway/proto/gateway.proto index 45e6ad60..40793059 100644 --- a/api/submitqueue/gateway/proto/gateway.proto +++ b/api/submitqueue/gateway/proto/gateway.proto @@ -124,6 +124,28 @@ message GetRequestSummaryByChangeURIResponse { repeated RequestSummary requests = 1; } +// ListRequest defines one bounded queue receipt-history query. +message ListRequest { + // Exact queue to query. + string queue = 1; + // Inclusive lower receipt-time bound in Unix milliseconds. + int64 received_at_or_after_ms = 2; + // Exclusive upper receipt-time bound in Unix milliseconds. + int64 received_before_ms = 3; + // Maximum results to return. Zero selects the server default; values above the server maximum are invalid. + int32 page_size = 4; + // Opaque continuation token from a previous response. Empty selects the first page. + string page_token = 5; +} + +// ListResponse defines one page of queue receipt history. +message ListResponse { + // Requests ordered by receipt time descending, then sqid descending. + repeated RequestSummary requests = 1; + // Opaque continuation token. Empty when this is the last page. + string next_page_token = 2; +} + // *************** // Error messages, returned as `google.rpc.Status` messages. // *************** @@ -184,4 +206,7 @@ service SubmitQueueGateway { // GetRequestSummaryByChangeURI returns current materialized statuses for an exact pinned change URI. rpc GetRequestSummaryByChangeURI(GetRequestSummaryByChangeURIRequest) returns (GetRequestSummaryByChangeURIResponse) {} + // List returns requests received for one queue during a bounded receipt-time range. + rpc List(ListRequest) returns (ListResponse) {} + } diff --git a/api/submitqueue/gateway/protopb/gateway.pb.go b/api/submitqueue/gateway/protopb/gateway.pb.go index 4ba88ee7..603afd66 100644 --- a/api/submitqueue/gateway/protopb/gateway.pb.go +++ b/api/submitqueue/gateway/protopb/gateway.pb.go @@ -651,6 +651,143 @@ func (x *GetRequestSummaryByChangeURIResponse) GetRequests() []*RequestSummary { return nil } +// ListRequest defines one bounded queue receipt-history query. +type ListRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Exact queue to query. + Queue string `protobuf:"bytes,1,opt,name=queue,proto3" json:"queue,omitempty"` + // Inclusive lower receipt-time bound in Unix milliseconds. + ReceivedAtOrAfterMs int64 `protobuf:"varint,2,opt,name=received_at_or_after_ms,json=receivedAtOrAfterMs,proto3" json:"received_at_or_after_ms,omitempty"` + // Exclusive upper receipt-time bound in Unix milliseconds. + ReceivedBeforeMs int64 `protobuf:"varint,3,opt,name=received_before_ms,json=receivedBeforeMs,proto3" json:"received_before_ms,omitempty"` + // Maximum results to return. Zero selects the server default; values above the server maximum are invalid. + PageSize int32 `protobuf:"varint,4,opt,name=page_size,json=pageSize,proto3" json:"page_size,omitempty"` + // Opaque continuation token from a previous response. Empty selects the first page. + PageToken string `protobuf:"bytes,5,opt,name=page_token,json=pageToken,proto3" json:"page_token,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ListRequest) Reset() { + *x = ListRequest{} + mi := &file_gateway_proto_msgTypes[11] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ListRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ListRequest) ProtoMessage() {} + +func (x *ListRequest) ProtoReflect() protoreflect.Message { + mi := &file_gateway_proto_msgTypes[11] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ListRequest.ProtoReflect.Descriptor instead. +func (*ListRequest) Descriptor() ([]byte, []int) { + return file_gateway_proto_rawDescGZIP(), []int{11} +} + +func (x *ListRequest) GetQueue() string { + if x != nil { + return x.Queue + } + return "" +} + +func (x *ListRequest) GetReceivedAtOrAfterMs() int64 { + if x != nil { + return x.ReceivedAtOrAfterMs + } + return 0 +} + +func (x *ListRequest) GetReceivedBeforeMs() int64 { + if x != nil { + return x.ReceivedBeforeMs + } + return 0 +} + +func (x *ListRequest) GetPageSize() int32 { + if x != nil { + return x.PageSize + } + return 0 +} + +func (x *ListRequest) GetPageToken() string { + if x != nil { + return x.PageToken + } + return "" +} + +// ListResponse defines one page of queue receipt history. +type ListResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Requests ordered by receipt time descending, then sqid descending. + Requests []*RequestSummary `protobuf:"bytes,1,rep,name=requests,proto3" json:"requests,omitempty"` + // Opaque continuation token. Empty when this is the last page. + NextPageToken string `protobuf:"bytes,2,opt,name=next_page_token,json=nextPageToken,proto3" json:"next_page_token,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ListResponse) Reset() { + *x = ListResponse{} + mi := &file_gateway_proto_msgTypes[12] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ListResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ListResponse) ProtoMessage() {} + +func (x *ListResponse) ProtoReflect() protoreflect.Message { + mi := &file_gateway_proto_msgTypes[12] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ListResponse.ProtoReflect.Descriptor instead. +func (*ListResponse) Descriptor() ([]byte, []int) { + return file_gateway_proto_rawDescGZIP(), []int{12} +} + +func (x *ListResponse) GetRequests() []*RequestSummary { + if x != nil { + return x.Requests + } + return nil +} + +func (x *ListResponse) GetNextPageToken() string { + if x != nil { + return x.NextPageToken + } + return "" +} + // Generic error with metadata. Each custom error type should extend this message. type Error struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -662,7 +799,7 @@ type Error struct { func (x *Error) Reset() { *x = Error{} - mi := &file_gateway_proto_msgTypes[11] + mi := &file_gateway_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -674,7 +811,7 @@ func (x *Error) String() string { func (*Error) ProtoMessage() {} func (x *Error) ProtoReflect() protoreflect.Message { - mi := &file_gateway_proto_msgTypes[11] + mi := &file_gateway_proto_msgTypes[13] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -687,7 +824,7 @@ func (x *Error) ProtoReflect() protoreflect.Message { // Deprecated: Use Error.ProtoReflect.Descriptor instead. func (*Error) Descriptor() ([]byte, []int) { - return file_gateway_proto_rawDescGZIP(), []int{11} + return file_gateway_proto_rawDescGZIP(), []int{13} } func (x *Error) GetMessage() string { @@ -710,7 +847,7 @@ type UnrecognizedQueueError struct { func (x *UnrecognizedQueueError) Reset() { *x = UnrecognizedQueueError{} - mi := &file_gateway_proto_msgTypes[12] + mi := &file_gateway_proto_msgTypes[14] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -722,7 +859,7 @@ func (x *UnrecognizedQueueError) String() string { func (*UnrecognizedQueueError) ProtoMessage() {} func (x *UnrecognizedQueueError) ProtoReflect() protoreflect.Message { - mi := &file_gateway_proto_msgTypes[12] + mi := &file_gateway_proto_msgTypes[14] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -735,7 +872,7 @@ func (x *UnrecognizedQueueError) ProtoReflect() protoreflect.Message { // Deprecated: Use UnrecognizedQueueError.ProtoReflect.Descriptor instead. func (*UnrecognizedQueueError) Descriptor() ([]byte, []int) { - return file_gateway_proto_rawDescGZIP(), []int{12} + return file_gateway_proto_rawDescGZIP(), []int{14} } func (x *UnrecognizedQueueError) GetError() *Error { @@ -767,7 +904,7 @@ type RequestNotFoundError struct { func (x *RequestNotFoundError) Reset() { *x = RequestNotFoundError{} - mi := &file_gateway_proto_msgTypes[13] + mi := &file_gateway_proto_msgTypes[15] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -779,7 +916,7 @@ func (x *RequestNotFoundError) String() string { func (*RequestNotFoundError) ProtoMessage() {} func (x *RequestNotFoundError) ProtoReflect() protoreflect.Message { - mi := &file_gateway_proto_msgTypes[13] + mi := &file_gateway_proto_msgTypes[15] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -792,7 +929,7 @@ func (x *RequestNotFoundError) ProtoReflect() protoreflect.Message { // Deprecated: Use RequestNotFoundError.ProtoReflect.Descriptor instead. func (*RequestNotFoundError) Descriptor() ([]byte, []int) { - return file_gateway_proto_rawDescGZIP(), []int{13} + return file_gateway_proto_rawDescGZIP(), []int{15} } func (x *RequestNotFoundError) GetError() *Error { @@ -859,7 +996,17 @@ const file_gateway_proto_rawDesc = "" + "\n" + "change_uri\x18\x01 \x01(\tR\tchangeUri\"l\n" + "$GetRequestSummaryByChangeURIResponse\x12D\n" + - "\brequests\x18\x01 \x03(\v2(.uber.submitqueue.gateway.RequestSummaryR\brequests\"!\n" + + "\brequests\x18\x01 \x03(\v2(.uber.submitqueue.gateway.RequestSummaryR\brequests\"\xc3\x01\n" + + "\vListRequest\x12\x14\n" + + "\x05queue\x18\x01 \x01(\tR\x05queue\x124\n" + + "\x17received_at_or_after_ms\x18\x02 \x01(\x03R\x13receivedAtOrAfterMs\x12,\n" + + "\x12received_before_ms\x18\x03 \x01(\x03R\x10receivedBeforeMs\x12\x1b\n" + + "\tpage_size\x18\x04 \x01(\x05R\bpageSize\x12\x1d\n" + + "\n" + + "page_token\x18\x05 \x01(\tR\tpageToken\"|\n" + + "\fListResponse\x12D\n" + + "\brequests\x18\x01 \x03(\v2(.uber.submitqueue.gateway.RequestSummaryR\brequests\x12&\n" + + "\x0fnext_page_token\x18\x02 \x01(\tR\rnextPageToken\"!\n" + "\x05Error\x12\x18\n" + "\amessage\x18\x01 \x01(\tR\amessage\"e\n" + "\x16UnrecognizedQueueError\x125\n" + @@ -869,13 +1016,14 @@ const file_gateway_proto_rawDesc = "" + "\x05error\x18\x01 \x01(\v2\x1f.uber.submitqueue.gateway.ErrorR\x05error\x12\x12\n" + "\x04sqid\x18\x02 \x01(\tR\x04sqid\x12\x1d\n" + "\n" + - "change_uri\x18\x03 \x01(\tR\tchangeUri2\xd4\x04\n" + + "change_uri\x18\x03 \x01(\tR\tchangeUri2\xad\x05\n" + "\x12SubmitQueueGateway\x12W\n" + "\x04Ping\x12%.uber.submitqueue.gateway.PingRequest\x1a&.uber.submitqueue.gateway.PingResponse\"\x00\x12W\n" + "\x04Land\x12%.uber.submitqueue.gateway.LandRequest\x1a&.uber.submitqueue.gateway.LandResponse\"\x00\x12]\n" + "\x06Cancel\x12'.uber.submitqueue.gateway.CancelRequest\x1a(.uber.submitqueue.gateway.CancelResponse\"\x00\x12\x8a\x01\n" + "\x15GetRequestSummaryByID\x126.uber.submitqueue.gateway.GetRequestSummaryByIDRequest\x1a7.uber.submitqueue.gateway.GetRequestSummaryByIDResponse\"\x00\x12\x9f\x01\n" + - "\x1cGetRequestSummaryByChangeURI\x12=.uber.submitqueue.gateway.GetRequestSummaryByChangeURIRequest\x1a>.uber.submitqueue.gateway.GetRequestSummaryByChangeURIResponse\"\x00Bk\n" + + "\x1cGetRequestSummaryByChangeURI\x12=.uber.submitqueue.gateway.GetRequestSummaryByChangeURIRequest\x1a>.uber.submitqueue.gateway.GetRequestSummaryByChangeURIResponse\"\x00\x12W\n" + + "\x04List\x12%.uber.submitqueue.gateway.ListRequest\x1a&.uber.submitqueue.gateway.ListResponse\"\x00Bk\n" + "\x1ccom.uber.submitqueue.gatewayB\fGatewayProtoP\x01Z;github.com/uber/submitqueue/api/submitqueue/gateway/protopbb\x06proto3" var ( @@ -890,7 +1038,7 @@ func file_gateway_proto_rawDescGZIP() []byte { return file_gateway_proto_rawDescData } -var file_gateway_proto_msgTypes = make([]protoimpl.MessageInfo, 15) +var file_gateway_proto_msgTypes = make([]protoimpl.MessageInfo, 17) var file_gateway_proto_goTypes = []any{ (*PingRequest)(nil), // 0: uber.submitqueue.gateway.PingRequest (*PingResponse)(nil), // 1: uber.submitqueue.gateway.PingResponse @@ -903,36 +1051,41 @@ var file_gateway_proto_goTypes = []any{ (*GetRequestSummaryByIDResponse)(nil), // 8: uber.submitqueue.gateway.GetRequestSummaryByIDResponse (*GetRequestSummaryByChangeURIRequest)(nil), // 9: uber.submitqueue.gateway.GetRequestSummaryByChangeURIRequest (*GetRequestSummaryByChangeURIResponse)(nil), // 10: uber.submitqueue.gateway.GetRequestSummaryByChangeURIResponse - (*Error)(nil), // 11: uber.submitqueue.gateway.Error - (*UnrecognizedQueueError)(nil), // 12: uber.submitqueue.gateway.UnrecognizedQueueError - (*RequestNotFoundError)(nil), // 13: uber.submitqueue.gateway.RequestNotFoundError - nil, // 14: uber.submitqueue.gateway.RequestSummary.MetadataEntry - (*protopb.Change)(nil), // 15: uber.base.change.Change - (protopb1.Strategy)(0), // 16: uber.base.mergestrategy.Strategy + (*ListRequest)(nil), // 11: uber.submitqueue.gateway.ListRequest + (*ListResponse)(nil), // 12: uber.submitqueue.gateway.ListResponse + (*Error)(nil), // 13: uber.submitqueue.gateway.Error + (*UnrecognizedQueueError)(nil), // 14: uber.submitqueue.gateway.UnrecognizedQueueError + (*RequestNotFoundError)(nil), // 15: uber.submitqueue.gateway.RequestNotFoundError + nil, // 16: uber.submitqueue.gateway.RequestSummary.MetadataEntry + (*protopb.Change)(nil), // 17: uber.base.change.Change + (protopb1.Strategy)(0), // 18: uber.base.mergestrategy.Strategy } var file_gateway_proto_depIdxs = []int32{ - 15, // 0: uber.submitqueue.gateway.LandRequest.change:type_name -> uber.base.change.Change - 16, // 1: uber.submitqueue.gateway.LandRequest.strategy:type_name -> uber.base.mergestrategy.Strategy - 14, // 2: uber.submitqueue.gateway.RequestSummary.metadata:type_name -> uber.submitqueue.gateway.RequestSummary.MetadataEntry + 17, // 0: uber.submitqueue.gateway.LandRequest.change:type_name -> uber.base.change.Change + 18, // 1: uber.submitqueue.gateway.LandRequest.strategy:type_name -> uber.base.mergestrategy.Strategy + 16, // 2: uber.submitqueue.gateway.RequestSummary.metadata:type_name -> uber.submitqueue.gateway.RequestSummary.MetadataEntry 6, // 3: uber.submitqueue.gateway.GetRequestSummaryByIDResponse.request:type_name -> uber.submitqueue.gateway.RequestSummary 6, // 4: uber.submitqueue.gateway.GetRequestSummaryByChangeURIResponse.requests:type_name -> uber.submitqueue.gateway.RequestSummary - 11, // 5: uber.submitqueue.gateway.UnrecognizedQueueError.error:type_name -> uber.submitqueue.gateway.Error - 11, // 6: uber.submitqueue.gateway.RequestNotFoundError.error:type_name -> uber.submitqueue.gateway.Error - 0, // 7: uber.submitqueue.gateway.SubmitQueueGateway.Ping:input_type -> uber.submitqueue.gateway.PingRequest - 2, // 8: uber.submitqueue.gateway.SubmitQueueGateway.Land:input_type -> uber.submitqueue.gateway.LandRequest - 4, // 9: uber.submitqueue.gateway.SubmitQueueGateway.Cancel:input_type -> uber.submitqueue.gateway.CancelRequest - 7, // 10: uber.submitqueue.gateway.SubmitQueueGateway.GetRequestSummaryByID:input_type -> uber.submitqueue.gateway.GetRequestSummaryByIDRequest - 9, // 11: uber.submitqueue.gateway.SubmitQueueGateway.GetRequestSummaryByChangeURI:input_type -> uber.submitqueue.gateway.GetRequestSummaryByChangeURIRequest - 1, // 12: uber.submitqueue.gateway.SubmitQueueGateway.Ping:output_type -> uber.submitqueue.gateway.PingResponse - 3, // 13: uber.submitqueue.gateway.SubmitQueueGateway.Land:output_type -> uber.submitqueue.gateway.LandResponse - 5, // 14: uber.submitqueue.gateway.SubmitQueueGateway.Cancel:output_type -> uber.submitqueue.gateway.CancelResponse - 8, // 15: uber.submitqueue.gateway.SubmitQueueGateway.GetRequestSummaryByID:output_type -> uber.submitqueue.gateway.GetRequestSummaryByIDResponse - 10, // 16: uber.submitqueue.gateway.SubmitQueueGateway.GetRequestSummaryByChangeURI:output_type -> uber.submitqueue.gateway.GetRequestSummaryByChangeURIResponse - 12, // [12:17] is the sub-list for method output_type - 7, // [7:12] is the sub-list for method input_type - 7, // [7:7] is the sub-list for extension type_name - 7, // [7:7] is the sub-list for extension extendee - 0, // [0:7] is the sub-list for field type_name + 6, // 5: uber.submitqueue.gateway.ListResponse.requests:type_name -> uber.submitqueue.gateway.RequestSummary + 13, // 6: uber.submitqueue.gateway.UnrecognizedQueueError.error:type_name -> uber.submitqueue.gateway.Error + 13, // 7: uber.submitqueue.gateway.RequestNotFoundError.error:type_name -> uber.submitqueue.gateway.Error + 0, // 8: uber.submitqueue.gateway.SubmitQueueGateway.Ping:input_type -> uber.submitqueue.gateway.PingRequest + 2, // 9: uber.submitqueue.gateway.SubmitQueueGateway.Land:input_type -> uber.submitqueue.gateway.LandRequest + 4, // 10: uber.submitqueue.gateway.SubmitQueueGateway.Cancel:input_type -> uber.submitqueue.gateway.CancelRequest + 7, // 11: uber.submitqueue.gateway.SubmitQueueGateway.GetRequestSummaryByID:input_type -> uber.submitqueue.gateway.GetRequestSummaryByIDRequest + 9, // 12: uber.submitqueue.gateway.SubmitQueueGateway.GetRequestSummaryByChangeURI:input_type -> uber.submitqueue.gateway.GetRequestSummaryByChangeURIRequest + 11, // 13: uber.submitqueue.gateway.SubmitQueueGateway.List:input_type -> uber.submitqueue.gateway.ListRequest + 1, // 14: uber.submitqueue.gateway.SubmitQueueGateway.Ping:output_type -> uber.submitqueue.gateway.PingResponse + 3, // 15: uber.submitqueue.gateway.SubmitQueueGateway.Land:output_type -> uber.submitqueue.gateway.LandResponse + 5, // 16: uber.submitqueue.gateway.SubmitQueueGateway.Cancel:output_type -> uber.submitqueue.gateway.CancelResponse + 8, // 17: uber.submitqueue.gateway.SubmitQueueGateway.GetRequestSummaryByID:output_type -> uber.submitqueue.gateway.GetRequestSummaryByIDResponse + 10, // 18: uber.submitqueue.gateway.SubmitQueueGateway.GetRequestSummaryByChangeURI:output_type -> uber.submitqueue.gateway.GetRequestSummaryByChangeURIResponse + 12, // 19: uber.submitqueue.gateway.SubmitQueueGateway.List:output_type -> uber.submitqueue.gateway.ListResponse + 14, // [14:20] is the sub-list for method output_type + 8, // [8:14] is the sub-list for method input_type + 8, // [8:8] is the sub-list for extension type_name + 8, // [8:8] is the sub-list for extension extendee + 0, // [0:8] is the sub-list for field type_name } func init() { file_gateway_proto_init() } @@ -946,7 +1099,7 @@ func file_gateway_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_gateway_proto_rawDesc), len(file_gateway_proto_rawDesc)), NumEnums: 0, - NumMessages: 15, + NumMessages: 17, NumExtensions: 0, NumServices: 1, }, diff --git a/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go b/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go index 0eafe53b..0b808dbe 100644 --- a/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go +++ b/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go @@ -26,6 +26,7 @@ type SubmitQueueGatewayYARPCClient interface { Cancel(context.Context, *CancelRequest, ...yarpc.CallOption) (*CancelResponse, error) GetRequestSummaryByID(context.Context, *GetRequestSummaryByIDRequest, ...yarpc.CallOption) (*GetRequestSummaryByIDResponse, error) GetRequestSummaryByChangeURI(context.Context, *GetRequestSummaryByChangeURIRequest, ...yarpc.CallOption) (*GetRequestSummaryByChangeURIResponse, error) + List(context.Context, *ListRequest, ...yarpc.CallOption) (*ListResponse, error) } func newSubmitQueueGatewayYARPCClient(clientConfig transport.ClientConfig, anyResolver v2.AnyResolver, options ...v2.ClientOption) SubmitQueueGatewayYARPCClient { @@ -51,6 +52,7 @@ type SubmitQueueGatewayYARPCServer interface { Cancel(context.Context, *CancelRequest) (*CancelResponse, error) GetRequestSummaryByID(context.Context, *GetRequestSummaryByIDRequest) (*GetRequestSummaryByIDResponse, error) GetRequestSummaryByChangeURI(context.Context, *GetRequestSummaryByChangeURIRequest) (*GetRequestSummaryByChangeURIResponse, error) + List(context.Context, *ListRequest) (*ListResponse, error) } type buildSubmitQueueGatewayYARPCProceduresParams struct { @@ -114,6 +116,16 @@ func buildSubmitQueueGatewayYARPCProcedures(params buildSubmitQueueGatewayYARPCP }, ), }, + { + MethodName: "List", + Handler: v2.NewUnaryHandler( + v2.UnaryHandlerParams{ + Handle: handler.List, + NewRequest: newSubmitQueueGatewayServiceListYARPCRequest, + AnyResolver: params.AnyResolver, + }, + ), + }, }, OnewayHandlerParams: []v2.BuildProceduresOnewayHandlerParams{}, StreamHandlerParams: []v2.BuildProceduresStreamHandlerParams{}, @@ -286,6 +298,18 @@ func (c *_SubmitQueueGatewayYARPCCaller) GetRequestSummaryByChangeURI(ctx contex return response, err } +func (c *_SubmitQueueGatewayYARPCCaller) List(ctx context.Context, request *ListRequest, options ...yarpc.CallOption) (*ListResponse, error) { + responseMessage, err := c.streamClient.Call(ctx, "List", request, newSubmitQueueGatewayServiceListYARPCResponse, options...) + if responseMessage == nil { + return nil, err + } + response, ok := responseMessage.(*ListResponse) + if !ok { + return nil, v2.CastError(emptySubmitQueueGatewayServiceListYARPCResponse, responseMessage) + } + return response, err +} + type _SubmitQueueGatewayYARPCHandler struct { server SubmitQueueGatewayYARPCServer } @@ -370,6 +394,22 @@ func (h *_SubmitQueueGatewayYARPCHandler) GetRequestSummaryByChangeURI(ctx conte return response, err } +func (h *_SubmitQueueGatewayYARPCHandler) List(ctx context.Context, requestMessage proto.Message) (proto.Message, error) { + var request *ListRequest + var ok bool + if requestMessage != nil { + request, ok = requestMessage.(*ListRequest) + if !ok { + return nil, v2.CastError(emptySubmitQueueGatewayServiceListYARPCRequest, requestMessage) + } + } + response, err := h.server.List(ctx, request) + if response == nil { + return nil, err + } + return response, err +} + func newSubmitQueueGatewayServicePingYARPCRequest() proto.Message { return &PingRequest{} } @@ -410,6 +450,14 @@ func newSubmitQueueGatewayServiceGetRequestSummaryByChangeURIYARPCResponse() pro return &GetRequestSummaryByChangeURIResponse{} } +func newSubmitQueueGatewayServiceListYARPCRequest() proto.Message { + return &ListRequest{} +} + +func newSubmitQueueGatewayServiceListYARPCResponse() proto.Message { + return &ListResponse{} +} + var ( emptySubmitQueueGatewayServicePingYARPCRequest = &PingRequest{} emptySubmitQueueGatewayServicePingYARPCResponse = &PingResponse{} @@ -421,61 +469,71 @@ var ( emptySubmitQueueGatewayServiceGetRequestSummaryByIDYARPCResponse = &GetRequestSummaryByIDResponse{} emptySubmitQueueGatewayServiceGetRequestSummaryByChangeURIYARPCRequest = &GetRequestSummaryByChangeURIRequest{} emptySubmitQueueGatewayServiceGetRequestSummaryByChangeURIYARPCResponse = &GetRequestSummaryByChangeURIResponse{} + emptySubmitQueueGatewayServiceListYARPCRequest = &ListRequest{} + emptySubmitQueueGatewayServiceListYARPCResponse = &ListResponse{} ) var yarpcFileDescriptorClosuref1a937782ebbded5 = [][]byte{ // gateway.proto []byte{ - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xa4, 0x56, 0x5f, 0x4f, 0xfb, 0x36, - 0x14, 0x25, 0x4d, 0x5b, 0xda, 0xdb, 0x52, 0x21, 0x8b, 0xa1, 0x28, 0x02, 0x51, 0x3c, 0x36, 0xfa, - 0x94, 0xa2, 0x4e, 0xfb, 0xa3, 0x21, 0x26, 0xad, 0xc0, 0x10, 0xd2, 0x40, 0x2c, 0x08, 0x4d, 0x9a, - 0x34, 0x55, 0x6e, 0x6a, 0x85, 0x88, 0x26, 0x29, 0xb6, 0xc3, 0xd4, 0x3d, 0xed, 0x65, 0x2f, 0x7b, - 0xde, 0xfb, 0x3e, 0xcf, 0xde, 0xf7, 0x81, 0xa6, 0xd8, 0x4e, 0x48, 0x51, 0x53, 0xe0, 0xf7, 0x7b, - 0xaa, 0xaf, 0x7d, 0xcf, 0xb9, 0xc7, 0xc7, 0xf6, 0x6d, 0x60, 0xc3, 0x27, 0x82, 0xfe, 0x46, 0xe6, - 0xce, 0x8c, 0xc5, 0x22, 0x46, 0x56, 0x32, 0xa6, 0xcc, 0xe1, 0xc9, 0x38, 0x0c, 0xc4, 0x63, 0x42, - 0x13, 0xea, 0xe8, 0x75, 0x1b, 0x93, 0x59, 0xd0, 0x1f, 0x13, 0x4e, 0xfb, 0xde, 0x3d, 0x89, 0x7c, - 0xda, 0x97, 0x00, 0x1d, 0x28, 0xb4, 0x7d, 0x94, 0xe7, 0x84, 0x94, 0xf9, 0x94, 0x0b, 0x46, 0x04, - 0xf5, 0xe7, 0x3a, 0x75, 0x61, 0x4e, 0x21, 0xf0, 0x21, 0xb4, 0x6e, 0x82, 0xc8, 0x77, 0xe9, 0x63, - 0x42, 0xb9, 0x40, 0x16, 0xac, 0x87, 0x94, 0x73, 0xe2, 0x53, 0xcb, 0xe8, 0x1a, 0xbd, 0xa6, 0x9b, - 0x85, 0xf8, 0x4f, 0x03, 0xda, 0x2a, 0x93, 0xcf, 0xe2, 0x88, 0xd3, 0xf2, 0x54, 0xb4, 0x0f, 0x6d, - 0x4e, 0xd9, 0x53, 0xe0, 0xd1, 0x51, 0x44, 0x42, 0x6a, 0x55, 0xe4, 0x72, 0x4b, 0xcf, 0x5d, 0x93, - 0x90, 0xa2, 0x1d, 0x68, 0x8a, 0x20, 0xa4, 0x5c, 0x90, 0x70, 0x66, 0x99, 0x5d, 0xa3, 0x67, 0xba, - 0xcf, 0x13, 0xc8, 0x86, 0xc6, 0x7d, 0xcc, 0x85, 0x04, 0x57, 0x25, 0x38, 0x8f, 0xf1, 0xdf, 0x06, - 0xb4, 0x7e, 0x24, 0xd1, 0x24, 0x53, 0xbc, 0x05, 0x35, 0xe9, 0x93, 0x16, 0xa1, 0x02, 0x74, 0x04, - 0x75, 0x65, 0x8c, 0x2c, 0xde, 0x1a, 0x58, 0x8e, 0xf4, 0x35, 0xb5, 0xc6, 0xd1, 0x8e, 0x9d, 0xca, - 0x1f, 0x57, 0xe7, 0xa1, 0x13, 0x68, 0x64, 0xd6, 0xc8, 0x9a, 0x9d, 0xc1, 0x7e, 0x01, 0xb3, 0x68, - 0xdd, 0xad, 0x1e, 0xb8, 0x39, 0x04, 0x63, 0x68, 0x2b, 0x55, 0xda, 0x1d, 0x04, 0x55, 0xfe, 0x18, - 0x4c, 0xb4, 0x2a, 0x39, 0xc6, 0xc7, 0xb0, 0x71, 0x4a, 0x22, 0x8f, 0x4e, 0x33, 0xed, 0x4b, 0x92, - 0xd0, 0x36, 0xd4, 0x19, 0x25, 0x3c, 0x8e, 0xb4, 0x6d, 0x3a, 0xc2, 0x9b, 0xd0, 0xc9, 0xc0, 0xaa, - 0x04, 0xfe, 0xb7, 0x02, 0x1d, 0xcd, 0x74, 0x9b, 0x84, 0x21, 0x61, 0xf3, 0xa5, 0x84, 0xb9, 0x41, - 0x95, 0xa2, 0x41, 0x7b, 0xd0, 0x52, 0x1b, 0x1f, 0x25, 0x2c, 0xe0, 0x96, 0xd9, 0x35, 0x7b, 0x4d, - 0x17, 0xd4, 0xd4, 0x1d, 0x0b, 0x38, 0x3a, 0x80, 0x0e, 0xa3, 0x1e, 0x0d, 0x9e, 0xe8, 0x64, 0x44, - 0xc4, 0x28, 0xe4, 0xd2, 0x15, 0xd3, 0x6d, 0x67, 0xb3, 0xdf, 0x8b, 0x2b, 0x9e, 0xaa, 0xe5, 0x82, - 0x88, 0x84, 0x5b, 0x35, 0xa5, 0x56, 0x45, 0x68, 0x17, 0x60, 0x4a, 0xb8, 0x18, 0x51, 0xc6, 0x62, - 0x66, 0xd5, 0xe5, 0x5a, 0x33, 0x9d, 0x39, 0x4f, 0x27, 0x90, 0x0b, 0x8d, 0x90, 0x0a, 0x32, 0x21, - 0x82, 0x58, 0xeb, 0x5d, 0xb3, 0xd7, 0x1a, 0x7c, 0xe5, 0x94, 0x5d, 0x7c, 0x67, 0x71, 0x8f, 0xce, - 0x95, 0x06, 0x9e, 0x47, 0x82, 0xcd, 0xdd, 0x9c, 0xc7, 0x3e, 0x86, 0x8d, 0x85, 0x25, 0xb4, 0x09, - 0xe6, 0x03, 0x9d, 0x6b, 0x2f, 0xd2, 0x61, 0x6a, 0xc5, 0x13, 0x99, 0x3e, 0x5b, 0x21, 0x83, 0x6f, - 0x2b, 0xdf, 0x18, 0x78, 0x00, 0x3b, 0x17, 0x54, 0x2c, 0x56, 0x1a, 0xce, 0x2f, 0xcf, 0x56, 0x9c, - 0x14, 0xf6, 0x60, 0xb7, 0x04, 0xa3, 0xef, 0xc0, 0x10, 0xd6, 0x99, 0x5a, 0x95, 0xb8, 0xd6, 0xa0, - 0xf7, 0xd6, 0x4d, 0xba, 0x19, 0x10, 0x9f, 0xc1, 0xa7, 0x4b, 0x8a, 0xa8, 0xbb, 0x7b, 0xe7, 0x5e, - 0x66, 0xfa, 0x76, 0x01, 0x9e, 0x8f, 0x53, 0xab, 0x6c, 0xe6, 0xa7, 0x89, 0xa7, 0x70, 0xb0, 0x9a, - 0x45, 0x2b, 0x3e, 0x83, 0x86, 0x2e, 0xcc, 0x2d, 0x43, 0x9e, 0xcb, 0xdb, 0x25, 0xe7, 0x48, 0xbc, - 0x0f, 0x35, 0x75, 0xcc, 0xe5, 0xdd, 0x84, 0xc2, 0xf6, 0x5d, 0xc4, 0xa8, 0x17, 0xfb, 0x51, 0xf0, - 0x3b, 0x9d, 0xfc, 0x94, 0x12, 0x2b, 0xcc, 0x97, 0x50, 0x53, 0x97, 0x46, 0x59, 0xb6, 0x57, 0x5e, - 0x5f, 0xe6, 0xbb, 0x2a, 0x7b, 0xf9, 0x2d, 0xc7, 0x7f, 0x18, 0xb0, 0xa5, 0x65, 0x5e, 0xc7, 0xe2, - 0x87, 0x38, 0x89, 0x26, 0x1f, 0x55, 0x25, 0xbb, 0x06, 0x95, 0xc2, 0xfb, 0x5a, 0xb4, 0xde, 0x7c, - 0x61, 0xfd, 0xe0, 0xbf, 0x2a, 0xa0, 0x5b, 0xc9, 0x2b, 0x37, 0x79, 0xa1, 0x68, 0xd1, 0xcf, 0x50, - 0x4d, 0xbb, 0x29, 0xfa, 0xac, 0xbc, 0x72, 0xa1, 0x2f, 0xdb, 0x9f, 0xbf, 0x96, 0xa6, 0x7b, 0xc2, - 0x5a, 0x4a, 0x9c, 0x36, 0xa2, 0x55, 0xc4, 0x85, 0xf6, 0xb9, 0x8a, 0xb8, 0xd8, 0xcf, 0xf0, 0x1a, - 0xfa, 0x15, 0xea, 0xaa, 0x01, 0xa1, 0xc3, 0x72, 0xcc, 0x42, 0x7f, 0xb3, 0x7b, 0xaf, 0x27, 0xe6, - 0xf4, 0x7f, 0x19, 0xf0, 0xc9, 0xd2, 0xe7, 0x84, 0x56, 0xb4, 0x86, 0x55, 0x6f, 0xd6, 0xfe, 0xfa, - 0xdd, 0xb8, 0x5c, 0xcc, 0x3f, 0xc6, 0xd2, 0x7e, 0x90, 0x3f, 0x18, 0x74, 0xf2, 0x2e, 0xee, 0x97, - 0xcf, 0xd5, 0xfe, 0xee, 0x43, 0xe1, 0x99, 0xc2, 0xe1, 0x03, 0xec, 0x78, 0x71, 0x58, 0x4a, 0x33, - 0x6c, 0xeb, 0x8b, 0x76, 0x93, 0xfe, 0xcb, 0xdf, 0x18, 0xbf, 0x1c, 0xfb, 0x81, 0xb8, 0x4f, 0xc6, - 0x8e, 0x17, 0x87, 0xfd, 0x14, 0xd4, 0x2f, 0x80, 0xfa, 0xe9, 0x57, 0x43, 0x31, 0xd6, 0x24, 0xea, - 0xb3, 0x61, 0x36, 0x1e, 0xd7, 0xe5, 0xe0, 0x8b, 0xff, 0x03, 0x00, 0x00, 0xff, 0xff, 0x40, 0x17, - 0x90, 0xe2, 0xac, 0x08, 0x00, 0x00, + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x56, 0x5f, 0x6f, 0x1b, 0x45, + 0x10, 0xcf, 0xd9, 0xb1, 0x63, 0x8f, 0x9d, 0x10, 0x2d, 0xa5, 0x9c, 0x4c, 0xa2, 0x26, 0x4b, 0x69, + 0xfd, 0x80, 0xce, 0x95, 0xf9, 0x2b, 0xa2, 0x22, 0xd5, 0x4d, 0xa9, 0x2a, 0x91, 0x12, 0x2e, 0x44, + 0x48, 0x48, 0xc8, 0x5a, 0xdb, 0xd3, 0xcb, 0x29, 0xb9, 0x3b, 0x67, 0x77, 0x2f, 0xe0, 0x88, 0x07, + 0x5e, 0x78, 0xe1, 0x99, 0x77, 0x3e, 0x01, 0x9f, 0x82, 0x27, 0xbe, 0x15, 0xda, 0x3f, 0x77, 0x3e, + 0x57, 0xf6, 0xb5, 0x81, 0x3e, 0xf9, 0x76, 0x66, 0x7e, 0x33, 0xb3, 0xbf, 0x9d, 0xfd, 0xad, 0x61, + 0x33, 0x60, 0x12, 0x7f, 0x62, 0x33, 0x6f, 0xca, 0x13, 0x99, 0x10, 0x37, 0x1d, 0x21, 0xf7, 0x44, + 0x3a, 0x8a, 0x42, 0x79, 0x99, 0x62, 0x8a, 0x9e, 0xf5, 0x77, 0x28, 0x9b, 0x86, 0xbd, 0x11, 0x13, + 0xd8, 0x1b, 0x9f, 0xb1, 0x38, 0xc0, 0x9e, 0x06, 0xd8, 0x85, 0x41, 0x77, 0x1e, 0xe4, 0x31, 0x11, + 0xf2, 0x00, 0x85, 0xe4, 0x4c, 0x62, 0x30, 0xb3, 0xa1, 0x0b, 0x36, 0x83, 0xa0, 0xf7, 0xa1, 0x75, + 0x1c, 0xc6, 0x81, 0x8f, 0x97, 0x29, 0x0a, 0x49, 0x5c, 0xd8, 0x88, 0x50, 0x08, 0x16, 0xa0, 0xeb, + 0xec, 0x39, 0xdd, 0xa6, 0x9f, 0x2d, 0xe9, 0x6f, 0x0e, 0xb4, 0x4d, 0xa4, 0x98, 0x26, 0xb1, 0xc0, + 0xd5, 0xa1, 0x64, 0x1f, 0xda, 0x02, 0xf9, 0x55, 0x38, 0xc6, 0x61, 0xcc, 0x22, 0x74, 0x2b, 0xda, + 0xdd, 0xb2, 0xb6, 0xe7, 0x2c, 0x42, 0xb2, 0x03, 0x4d, 0x19, 0x46, 0x28, 0x24, 0x8b, 0xa6, 0x6e, + 0x75, 0xcf, 0xe9, 0x56, 0xfd, 0xb9, 0x81, 0x74, 0xa0, 0x71, 0x96, 0x08, 0xa9, 0xc1, 0xeb, 0x1a, + 0x9c, 0xaf, 0xe9, 0x1f, 0x0e, 0xb4, 0xbe, 0x66, 0xf1, 0x24, 0xeb, 0xf8, 0x16, 0xd4, 0x34, 0x4f, + 0xb6, 0x09, 0xb3, 0x20, 0x0f, 0xa0, 0x6e, 0x88, 0xd1, 0xc5, 0x5b, 0x7d, 0xd7, 0xd3, 0xbc, 0x2a, + 0x6a, 0x3c, 0xcb, 0xd8, 0x63, 0xfd, 0xe3, 0xdb, 0x38, 0xf2, 0x10, 0x1a, 0x19, 0x35, 0xba, 0xe6, + 0x56, 0x7f, 0xbf, 0x80, 0x59, 0xa4, 0xee, 0xc4, 0x7e, 0xf8, 0x39, 0x84, 0x52, 0x68, 0x9b, 0xae, + 0x2c, 0x3b, 0x04, 0xd6, 0xc5, 0x65, 0x38, 0xb1, 0x5d, 0xe9, 0x6f, 0x7a, 0x00, 0x9b, 0x8f, 0x59, + 0x3c, 0xc6, 0x8b, 0xac, 0xf7, 0x25, 0x41, 0xe4, 0x36, 0xd4, 0x39, 0x32, 0x91, 0xc4, 0x96, 0x36, + 0xbb, 0xa2, 0xdb, 0xb0, 0x95, 0x81, 0x4d, 0x09, 0xfa, 0x4f, 0x05, 0xb6, 0x6c, 0xa6, 0x93, 0x34, + 0x8a, 0x18, 0x9f, 0x2d, 0x4d, 0x98, 0x13, 0x54, 0x29, 0x12, 0x74, 0x07, 0x5a, 0x66, 0xe3, 0xc3, + 0x94, 0x87, 0xc2, 0xad, 0xee, 0x55, 0xbb, 0x4d, 0x1f, 0x8c, 0xe9, 0x94, 0x87, 0x82, 0xdc, 0x85, + 0x2d, 0x8e, 0x63, 0x0c, 0xaf, 0x70, 0x32, 0x64, 0x72, 0x18, 0x09, 0xcd, 0x4a, 0xd5, 0x6f, 0x67, + 0xd6, 0x47, 0xf2, 0x48, 0xa8, 0x6e, 0x85, 0x64, 0x32, 0x15, 0x6e, 0xcd, 0x74, 0x6b, 0x56, 0x64, + 0x17, 0xe0, 0x82, 0x09, 0x39, 0x44, 0xce, 0x13, 0xee, 0xd6, 0xb5, 0xaf, 0xa9, 0x2c, 0x4f, 0x94, + 0x81, 0xf8, 0xd0, 0x88, 0x50, 0xb2, 0x09, 0x93, 0xcc, 0xdd, 0xd8, 0xab, 0x76, 0x5b, 0xfd, 0x4f, + 0xbd, 0x55, 0x83, 0xef, 0x2d, 0xee, 0xd1, 0x3b, 0xb2, 0xc0, 0x27, 0xb1, 0xe4, 0x33, 0x3f, 0xcf, + 0xd3, 0x39, 0x80, 0xcd, 0x05, 0x17, 0xd9, 0x86, 0xea, 0x39, 0xce, 0x2c, 0x17, 0xea, 0x53, 0x51, + 0x71, 0xc5, 0x2e, 0xe6, 0x54, 0xe8, 0xc5, 0x17, 0x95, 0xcf, 0x1d, 0xda, 0x87, 0x9d, 0xa7, 0x28, + 0x17, 0x2b, 0x0d, 0x66, 0xcf, 0x0e, 0x4b, 0x4e, 0x8a, 0x8e, 0x61, 0x77, 0x05, 0xc6, 0xce, 0xc0, + 0x00, 0x36, 0xb8, 0xf1, 0x6a, 0x5c, 0xab, 0xdf, 0x7d, 0xdd, 0x4d, 0xfa, 0x19, 0x90, 0x1e, 0xc2, + 0xfb, 0x4b, 0x8a, 0x98, 0xd9, 0x3d, 0xf5, 0x9f, 0x65, 0xfd, 0xed, 0x02, 0xcc, 0x8f, 0xd3, 0x76, + 0xd9, 0xcc, 0x4f, 0x93, 0x5e, 0xc0, 0xdd, 0xf2, 0x2c, 0xb6, 0xe3, 0x43, 0x68, 0xd8, 0xc2, 0xc2, + 0x75, 0xf4, 0xb9, 0xbc, 0x7e, 0xcb, 0x39, 0x92, 0xfe, 0xad, 0xae, 0x68, 0x28, 0x64, 0xf9, 0x15, + 0xfd, 0x18, 0xde, 0x2d, 0x0e, 0x58, 0xc2, 0x87, 0xec, 0x85, 0x44, 0xae, 0x26, 0xad, 0xa2, 0x27, + 0xed, 0xed, 0xf9, 0xa4, 0x7d, 0xc3, 0x1f, 0x29, 0xdf, 0x91, 0x20, 0x1f, 0x02, 0xc9, 0x51, 0x23, + 0x7c, 0x91, 0x70, 0x54, 0x00, 0xa3, 0x20, 0xdb, 0x99, 0x67, 0xa0, 0x1d, 0x47, 0x82, 0xbc, 0x07, + 0xcd, 0x29, 0x0b, 0x70, 0x28, 0xc2, 0x6b, 0xa3, 0x24, 0x35, 0xbf, 0xa1, 0x0c, 0x27, 0xe1, 0x35, + 0x2a, 0xce, 0xb4, 0x53, 0x26, 0xe7, 0x18, 0xdb, 0xf9, 0xd5, 0xe1, 0xdf, 0x29, 0x03, 0xfd, 0x05, + 0xda, 0x66, 0x13, 0x6f, 0x92, 0x1b, 0x72, 0x0f, 0xde, 0x8a, 0xf1, 0x67, 0x39, 0x2c, 0x54, 0x36, + 0xc3, 0xb8, 0xa9, 0xcc, 0xc7, 0x79, 0xf5, 0x7d, 0xa8, 0x99, 0xab, 0xb2, 0x5a, 0x91, 0x11, 0x6e, + 0x9f, 0xc6, 0x1c, 0xc7, 0x49, 0x10, 0x87, 0xd7, 0x38, 0xf9, 0x56, 0x35, 0x60, 0x30, 0x9f, 0x40, + 0xcd, 0x5c, 0x3c, 0x33, 0x76, 0x77, 0x56, 0xf7, 0xa9, 0xe3, 0x7d, 0x13, 0xbd, 0x5c, 0x29, 0xe8, + 0xaf, 0x0e, 0xdc, 0xb2, 0xdb, 0x79, 0x9e, 0xc8, 0xaf, 0x92, 0x34, 0x9e, 0xfc, 0xaf, 0x2a, 0xd9, + 0x55, 0xaa, 0x14, 0x34, 0x6a, 0x71, 0x7c, 0xab, 0x2f, 0x8d, 0x6f, 0xff, 0xaf, 0x1a, 0x90, 0x13, + 0x9d, 0x57, 0x6f, 0xf2, 0xa9, 0x49, 0x4b, 0xbe, 0x87, 0x75, 0xf5, 0x22, 0x91, 0x0f, 0x56, 0x57, + 0x2e, 0xbc, 0x6d, 0x9d, 0x7b, 0xaf, 0x0a, 0xb3, 0xba, 0xba, 0xa6, 0x12, 0x2b, 0x31, 0x2f, 0x4b, + 0x5c, 0x78, 0x82, 0xca, 0x12, 0x17, 0xdf, 0x04, 0xba, 0x46, 0x7e, 0x84, 0xba, 0x11, 0x71, 0x72, + 0x7f, 0x35, 0x66, 0xe1, 0x8d, 0xe8, 0x74, 0x5f, 0x1d, 0x98, 0xa7, 0xff, 0xdd, 0x81, 0x77, 0x96, + 0x4a, 0x12, 0x29, 0x91, 0xd7, 0x32, 0xdd, 0xeb, 0x7c, 0x76, 0x63, 0x5c, 0xde, 0xcc, 0x9f, 0xce, + 0x52, 0x4d, 0xcd, 0x45, 0x87, 0x3c, 0xbc, 0x51, 0xee, 0x97, 0x25, 0xaf, 0xf3, 0xe5, 0x7f, 0x85, + 0x2f, 0x1c, 0x73, 0x28, 0x64, 0xe9, 0x31, 0xcf, 0x65, 0xac, 0xf4, 0x98, 0x0b, 0x42, 0x41, 0xd7, + 0x06, 0xe7, 0xb0, 0x33, 0x4e, 0xa2, 0x95, 0xe1, 0x83, 0xb6, 0x9d, 0xe0, 0x63, 0xf5, 0x17, 0xec, + 0xd8, 0xf9, 0xe1, 0x20, 0x08, 0xe5, 0x59, 0x3a, 0xf2, 0xc6, 0x49, 0xd4, 0x53, 0xa0, 0x5e, 0x01, + 0xd4, 0x53, 0x7f, 0xe9, 0x8a, 0x6b, 0x9b, 0xc4, 0xfc, 0xa7, 0x9b, 0x8e, 0x46, 0x75, 0xfd, 0xf1, + 0xd1, 0xbf, 0x01, 0x00, 0x00, 0xff, 0xff, 0x93, 0xae, 0xe0, 0xd4, 0x49, 0x0a, 0x00, 0x00, }, // api/base/change/proto/change.proto []byte{ diff --git a/api/submitqueue/gateway/protopb/gateway_grpc.pb.go b/api/submitqueue/gateway/protopb/gateway_grpc.pb.go index 8cd12a7f..86c8f6ec 100644 --- a/api/submitqueue/gateway/protopb/gateway_grpc.pb.go +++ b/api/submitqueue/gateway/protopb/gateway_grpc.pb.go @@ -39,6 +39,7 @@ const ( SubmitQueueGateway_Cancel_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/Cancel" SubmitQueueGateway_GetRequestSummaryByID_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/GetRequestSummaryByID" SubmitQueueGateway_GetRequestSummaryByChangeURI_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/GetRequestSummaryByChangeURI" + SubmitQueueGateway_List_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/List" ) // SubmitQueueGatewayClient is the client API for SubmitQueueGateway service. @@ -68,6 +69,8 @@ type SubmitQueueGatewayClient interface { GetRequestSummaryByID(ctx context.Context, in *GetRequestSummaryByIDRequest, opts ...grpc.CallOption) (*GetRequestSummaryByIDResponse, error) // GetRequestSummaryByChangeURI returns current materialized statuses for an exact pinned change URI. GetRequestSummaryByChangeURI(ctx context.Context, in *GetRequestSummaryByChangeURIRequest, opts ...grpc.CallOption) (*GetRequestSummaryByChangeURIResponse, error) + // List returns requests received for one queue during a bounded receipt-time range. + List(ctx context.Context, in *ListRequest, opts ...grpc.CallOption) (*ListResponse, error) } type submitQueueGatewayClient struct { @@ -128,6 +131,16 @@ func (c *submitQueueGatewayClient) GetRequestSummaryByChangeURI(ctx context.Cont return out, nil } +func (c *submitQueueGatewayClient) List(ctx context.Context, in *ListRequest, opts ...grpc.CallOption) (*ListResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(ListResponse) + err := c.cc.Invoke(ctx, SubmitQueueGateway_List_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // SubmitQueueGatewayServer is the server API for SubmitQueueGateway service. // All implementations must embed UnimplementedSubmitQueueGatewayServer // for forward compatibility. @@ -155,6 +168,8 @@ type SubmitQueueGatewayServer interface { GetRequestSummaryByID(context.Context, *GetRequestSummaryByIDRequest) (*GetRequestSummaryByIDResponse, error) // GetRequestSummaryByChangeURI returns current materialized statuses for an exact pinned change URI. GetRequestSummaryByChangeURI(context.Context, *GetRequestSummaryByChangeURIRequest) (*GetRequestSummaryByChangeURIResponse, error) + // List returns requests received for one queue during a bounded receipt-time range. + List(context.Context, *ListRequest) (*ListResponse, error) mustEmbedUnimplementedSubmitQueueGatewayServer() } @@ -180,6 +195,9 @@ func (UnimplementedSubmitQueueGatewayServer) GetRequestSummaryByID(context.Conte func (UnimplementedSubmitQueueGatewayServer) GetRequestSummaryByChangeURI(context.Context, *GetRequestSummaryByChangeURIRequest) (*GetRequestSummaryByChangeURIResponse, error) { return nil, status.Errorf(codes.Unimplemented, "method GetRequestSummaryByChangeURI not implemented") } +func (UnimplementedSubmitQueueGatewayServer) List(context.Context, *ListRequest) (*ListResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method List not implemented") +} func (UnimplementedSubmitQueueGatewayServer) mustEmbedUnimplementedSubmitQueueGatewayServer() {} func (UnimplementedSubmitQueueGatewayServer) testEmbeddedByValue() {} @@ -291,6 +309,24 @@ func _SubmitQueueGateway_GetRequestSummaryByChangeURI_Handler(srv interface{}, c return interceptor(ctx, in, info, handler) } +func _SubmitQueueGateway_List_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(ListRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(SubmitQueueGatewayServer).List(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: SubmitQueueGateway_List_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(SubmitQueueGatewayServer).List(ctx, req.(*ListRequest)) + } + return interceptor(ctx, in, info, handler) +} + // SubmitQueueGateway_ServiceDesc is the grpc.ServiceDesc for SubmitQueueGateway service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -318,6 +354,10 @@ var SubmitQueueGateway_ServiceDesc = grpc.ServiceDesc{ MethodName: "GetRequestSummaryByChangeURI", Handler: _SubmitQueueGateway_GetRequestSummaryByChangeURI_Handler, }, + { + MethodName: "List", + Handler: _SubmitQueueGateway_List_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "gateway.proto", diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 45f9c638..1bfeb252 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -56,6 +56,7 @@ type GatewayServer struct { landController *controller.LandController cancelController *controller.CancelController requestSummaryController *controller.RequestSummaryController + listController *controller.ListController } // Ping delegates to the controller @@ -104,6 +105,15 @@ func (s *GatewayServer) GetRequestSummaryByChangeURI(ctx context.Context, req *p return &pb.GetRequestSummaryByChangeURIResponse{Requests: mapper.RequestSummariesToProto(summaries)}, nil } +// List maps the wire request to an entity, delegates to the controller, and maps the result back to the wire response. +func (s *GatewayServer) List(ctx context.Context, req *pb.ListRequest) (*pb.ListResponse, error) { + result, err := s.listController.List(ctx, mapper.ProtoToListRequest(req)) + if err != nil { + return nil, err + } + return mapper.ListResultToProto(result), nil +} + func gatewayStatusError(err error) error { switch { case controller.IsRequestNotFound(err): @@ -299,11 +309,13 @@ func run() error { store.GetRequestSummaryStore(), store.GetRequestURIStore(), ) + listController := controller.NewListController(logger.Sugar(), scope, store.GetRequestQueueSummaryStore(), queueConfigs) gatewayServer := &GatewayServer{ pingController: pingController, landController: landController, cancelController: cancelController, requestSummaryController: requestSummaryController, + listController: listController, } pb.RegisterSubmitQueueGatewayServer(grpcServer, gatewayServer) diff --git a/service/submitqueue/gateway/server/mapper/BUILD.bazel b/service/submitqueue/gateway/server/mapper/BUILD.bazel index 572a438a..3ef3401b 100644 --- a/service/submitqueue/gateway/server/mapper/BUILD.bazel +++ b/service/submitqueue/gateway/server/mapper/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "cancel.go", "land.go", + "list.go", "request_summary.go", ], importpath = "github.com/uber/submitqueue/service/submitqueue/gateway/server/mapper", @@ -23,6 +24,7 @@ go_test( srcs = [ "cancel_test.go", "land_test.go", + "list_test.go", "request_summary_test.go", ], embed = [":go_default_library"], diff --git a/service/submitqueue/gateway/server/mapper/list.go b/service/submitqueue/gateway/server/mapper/list.go new file mode 100644 index 00000000..aca2613a --- /dev/null +++ b/service/submitqueue/gateway/server/mapper/list.go @@ -0,0 +1,56 @@ +// Copyright (c) 2025 Uber Technologies, 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 mapper + +import ( + pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/submitqueue/entity" +) + +// ProtoToListRequest maps the wire request to the entity request the controller operates on. +func ProtoToListRequest(req *pb.ListRequest) entity.ListRequest { + return entity.ListRequest{ + Queue: req.GetQueue(), + ReceivedAtOrAfterMs: req.GetReceivedAtOrAfterMs(), + ReceivedBeforeMs: req.GetReceivedBeforeMs(), + PageSize: req.GetPageSize(), + PageToken: req.GetPageToken(), + } +} + +// ListResultToProto maps the domain result to the wire response. +func ListResultToProto(result entity.ListResult) *pb.ListResponse { + requests := make([]*pb.RequestSummary, 0, len(result.Requests)) + for _, summary := range result.Requests { + requests = append(requests, RequestQueueSummaryToProto(summary)) + } + return &pb.ListResponse{ + Requests: requests, + NextPageToken: result.NextPageToken, + } +} + +// RequestQueueSummaryToProto maps a queue projection to the wire request summary. +func RequestQueueSummaryToProto(summary entity.RequestQueueSummary) *pb.RequestSummary { + return &pb.RequestSummary{ + Sqid: summary.RequestID, + Queue: summary.Queue, + ChangeUris: append([]string{}, summary.ChangeURIs...), + ReceivedAtMs: summary.ReceivedAtMs, + Status: string(summary.Status), + LastError: summary.LastError, + Metadata: cloneStringMap(summary.Metadata), + } +} diff --git a/service/submitqueue/gateway/server/mapper/list_test.go b/service/submitqueue/gateway/server/mapper/list_test.go new file mode 100644 index 00000000..0c9e6261 --- /dev/null +++ b/service/submitqueue/gateway/server/mapper/list_test.go @@ -0,0 +1,60 @@ +// Copyright (c) 2025 Uber Technologies, 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 mapper + +import ( + "testing" + + "github.com/stretchr/testify/assert" + pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/submitqueue/entity" +) + +func TestProtoToListRequest(t *testing.T) { + req := &pb.ListRequest{ + Queue: "q", + ReceivedAtOrAfterMs: 100, + ReceivedBeforeMs: 200, + PageSize: 25, + PageToken: "token", + } + + assert.Equal(t, entity.ListRequest{ + Queue: "q", + ReceivedAtOrAfterMs: 100, + ReceivedBeforeMs: 200, + PageSize: 25, + PageToken: "token", + }, ProtoToListRequest(req)) +} + +func TestListResultToProto(t *testing.T) { + result := entity.ListResult{ + Requests: []entity.RequestQueueSummary{{ + RequestID: "q/1", + Queue: "q", + ChangeURIs: []string{"github://uber/repo/pull/1/abc"}, + ReceivedAtMs: 100, + Status: entity.RequestStatusAccepted, + Metadata: map[string]string{}, + }}, + NextPageToken: "next", + } + + response := ListResultToProto(result) + + assert.Equal(t, "q/1", response.Requests[0].Sqid) + assert.Equal(t, "next", response.NextPageToken) +} diff --git a/submitqueue/entity/BUILD.bazel b/submitqueue/entity/BUILD.bazel index 030209da..6f6a9190 100644 --- a/submitqueue/entity/BUILD.bazel +++ b/submitqueue/entity/BUILD.bazel @@ -12,6 +12,7 @@ go_library( "change_record.go", "conflict.go", "land.go", + "list.go", "merge_result.go", "push_result.go", "queue.go", diff --git a/submitqueue/entity/list.go b/submitqueue/entity/list.go new file mode 100644 index 00000000..ffef78a7 --- /dev/null +++ b/submitqueue/entity/list.go @@ -0,0 +1,37 @@ +// Copyright (c) 2025 Uber Technologies, 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 entity + +// ListRequest defines one bounded queue receipt-history query. +type ListRequest struct { + // Queue is the exact queue to query. + Queue string + // ReceivedAtOrAfterMs is the inclusive lower receipt-time bound in Unix milliseconds. + ReceivedAtOrAfterMs int64 + // ReceivedBeforeMs is the exclusive upper receipt-time bound in Unix milliseconds. + ReceivedBeforeMs int64 + // PageSize is the maximum number of results to return. Zero selects the server default. + PageSize int32 + // PageToken is an opaque continuation token from a previous result. + PageToken string +} + +// ListResult contains one page of queue receipt history. +type ListResult struct { + // Requests are ordered by receipt time descending, then request ID descending. + Requests []RequestQueueSummary + // NextPageToken is an opaque continuation token. Empty means this is the last page. + NextPageToken string +} diff --git a/submitqueue/gateway/controller/BUILD.bazel b/submitqueue/gateway/controller/BUILD.bazel index c71a3d4f..c116f0ce 100644 --- a/submitqueue/gateway/controller/BUILD.bazel +++ b/submitqueue/gateway/controller/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "cancel.go", "land.go", + "list.go", "ping.go", "read_errors.go", "request_summary.go", @@ -33,6 +34,7 @@ go_test( srcs = [ "cancel_test.go", "land_test.go", + "list_test.go", "ping_test.go", "request_summary_test.go", "storage_fixture_test.go", diff --git a/submitqueue/gateway/controller/list.go b/submitqueue/gateway/controller/list.go new file mode 100644 index 00000000..c629777e --- /dev/null +++ b/submitqueue/gateway/controller/list.go @@ -0,0 +1,174 @@ +// Copyright (c) 2025 Uber Technologies, 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 controller + +import ( + "context" + "encoding/base64" + "errors" + "fmt" + "net/url" + "strconv" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/queueconfig" + "github.com/uber/submitqueue/submitqueue/extension/storage" + "go.uber.org/zap" +) + +const ( + defaultListPageSize = 50 + maxListPageSize = 200 +) + +type listPageToken struct { + Queue string + ReceivedAtOrAfterMs int64 + ReceivedBeforeMs int64 + LastReceivedAtMs int64 + LastRequestID string +} + +// ListController handles bounded queue receipt-history queries. +type ListController struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + requestQueueSummaryStore storage.RequestQueueSummaryStore + queueConfigs queueconfig.Store +} + +// NewListController creates a gateway list controller. +func NewListController(logger *zap.SugaredLogger, scope tally.Scope, requestQueueSummaryStore storage.RequestQueueSummaryStore, queueConfigs queueconfig.Store) *ListController { + return &ListController{ + logger: logger, + metricsScope: scope.SubScope("list_controller"), + requestQueueSummaryStore: requestQueueSummaryStore, + queueConfigs: queueConfigs, + } +} + +// List returns one page of requests received for a queue in the supplied half-open time range. +func (c *ListController) List(ctx context.Context, req entity.ListRequest) (result entity.ListResult, retErr error) { + op := metrics.Begin(c.metricsScope, "list") + defer func() { op.Complete(retErr) }() + + if err := validateStoredIdentifier("queue", req.Queue); err != nil { + return entity.ListResult{}, fmt.Errorf("ListController invalid queue: %w", err) + } + if _, err := c.queueConfigs.Get(ctx, req.Queue); err != nil { + if errors.Is(err, queueconfig.ErrNotFound) { + return entity.ListResult{}, errs.NewUserError(&UnrecognizedQueueError{Queue: req.Queue}) + } + return entity.ListResult{}, fmt.Errorf("ListController failed to look up queue %q: %w", req.Queue, err) + } + if req.ReceivedAtOrAfterMs >= req.ReceivedBeforeMs { + return entity.ListResult{}, fmt.Errorf("ListController requires received_at_or_after_ms < received_before_ms: %w", ErrInvalidRequest) + } + pageSize := int(req.PageSize) + if pageSize == 0 { + pageSize = defaultListPageSize + } + if pageSize < 0 || pageSize > maxListPageSize { + return entity.ListResult{}, fmt.Errorf("ListController page_size must be between 0 and %d: %w", maxListPageSize, ErrInvalidRequest) + } + + query := storage.RequestQueueSummaryQuery{ + Queue: req.Queue, + ReceivedAtOrAfterMs: req.ReceivedAtOrAfterMs, + ReceivedBeforeMs: req.ReceivedBeforeMs, + Limit: pageSize + 1, + } + if req.PageToken != "" { + token, err := decodeListPageToken(req.PageToken) + if err != nil { + return entity.ListResult{}, fmt.Errorf("ListController invalid page token: %w", ErrInvalidRequest) + } + if token.Queue != req.Queue || token.ReceivedAtOrAfterMs != req.ReceivedAtOrAfterMs || token.ReceivedBeforeMs != req.ReceivedBeforeMs { + return entity.ListResult{}, fmt.Errorf("ListController page token does not match query: %w", ErrInvalidRequest) + } + query.HasCursor = true + query.Cursor = storage.RequestQueueSummaryCursor{ReceivedAtMs: token.LastReceivedAtMs, RequestID: token.LastRequestID} + } + + summaries, err := c.requestQueueSummaryStore.List(ctx, query) + if err != nil { + return entity.ListResult{}, fmt.Errorf("ListController failed to list queue=%s: %w", req.Queue, err) + } + + visible := summaries + result = entity.ListResult{Requests: make([]entity.RequestQueueSummary, 0, min(len(visible), pageSize))} + if len(visible) > pageSize { + visible = visible[:pageSize] + last := visible[len(visible)-1] + result.NextPageToken = encodeListPageToken(listPageToken{ + Queue: req.Queue, + ReceivedAtOrAfterMs: req.ReceivedAtOrAfterMs, + ReceivedBeforeMs: req.ReceivedBeforeMs, + LastReceivedAtMs: last.ReceivedAtMs, + LastRequestID: last.RequestID, + }) + } + result.Requests = append(result.Requests, visible...) + c.logger.Debugw("queue requests listed", "queue", req.Queue, "request_count", len(result.Requests), "has_next_page", result.NextPageToken != "") + return result, nil +} + +func encodeListPageToken(token listPageToken) string { + values := url.Values{ + "queue": {token.Queue}, + "received_at_or_after_ms": {strconv.FormatInt(token.ReceivedAtOrAfterMs, 10)}, + "received_before_ms": {strconv.FormatInt(token.ReceivedBeforeMs, 10)}, + "last_received_at_ms": {strconv.FormatInt(token.LastReceivedAtMs, 10)}, + "last_request_id": {token.LastRequestID}, + } + return base64.RawURLEncoding.EncodeToString([]byte(values.Encode())) +} + +func decodeListPageToken(encoded string) (listPageToken, error) { + data, err := base64.RawURLEncoding.DecodeString(encoded) + if err != nil { + return listPageToken{}, err + } + values, err := url.ParseQuery(string(data)) + if err != nil { + return listPageToken{}, err + } + receivedAtOrAfterMs, err := strconv.ParseInt(values.Get("received_at_or_after_ms"), 10, 64) + if err != nil { + return listPageToken{}, err + } + receivedBeforeMs, err := strconv.ParseInt(values.Get("received_before_ms"), 10, 64) + if err != nil { + return listPageToken{}, err + } + lastReceivedAtMs, err := strconv.ParseInt(values.Get("last_received_at_ms"), 10, 64) + if err != nil { + return listPageToken{}, err + } + token := listPageToken{ + Queue: values.Get("queue"), + ReceivedAtOrAfterMs: receivedAtOrAfterMs, + ReceivedBeforeMs: receivedBeforeMs, + LastReceivedAtMs: lastReceivedAtMs, + LastRequestID: values.Get("last_request_id"), + } + if token.Queue == "" || token.LastRequestID == "" || token.ReceivedAtOrAfterMs >= token.ReceivedBeforeMs { + return listPageToken{}, fmt.Errorf("invalid token fields") + } + return token, nil +} diff --git a/submitqueue/gateway/controller/list_test.go b/submitqueue/gateway/controller/list_test.go new file mode 100644 index 00000000..3ec52e52 --- /dev/null +++ b/submitqueue/gateway/controller/list_test.go @@ -0,0 +1,137 @@ +// Copyright (c) 2025 Uber Technologies, 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 controller + +import ( + "context" + "encoding/base64" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/queueconfig" + qcmock "github.com/uber/submitqueue/submitqueue/extension/queueconfig/mock" + "github.com/uber/submitqueue/submitqueue/extension/storage" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +func TestList_ReturnsPageAndCursor(t *testing.T) { + ctrl := gomock.NewController(t) + store := storagemock.NewMockRequestQueueSummaryStore(ctrl) + store.EXPECT().List(gomock.Any(), storage.RequestQueueSummaryQuery{ + Queue: "q", ReceivedAtOrAfterMs: 100, ReceivedBeforeMs: 200, Limit: 3, + }).Return([]entity.RequestQueueSummary{ + {RequestID: "q/3", Queue: "q", ChangeURIs: []string{}, ReceivedAtMs: 190, Status: entity.RequestStatusAccepted, Metadata: map[string]string{}}, + {RequestID: "q/2", Queue: "q", ChangeURIs: []string{}, ReceivedAtMs: 180, Status: entity.RequestStatusLanded, Metadata: map[string]string{}}, + {RequestID: "q/1", Queue: "q", ChangeURIs: []string{}, ReceivedAtMs: 170, Status: entity.RequestStatusError, Metadata: map[string]string{}}, + }, nil) + controller := newConfiguredListController(ctrl, store) + + result, err := controller.List(context.Background(), entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 100, ReceivedBeforeMs: 200, PageSize: 2}) + + require.NoError(t, err) + require.Len(t, result.Requests, 2) + assert.Equal(t, []string{"q/3", "q/2"}, []string{result.Requests[0].RequestID, result.Requests[1].RequestID}) + assert.Equal(t, int64(190), result.Requests[0].ReceivedAtMs) + require.NotEmpty(t, result.NextPageToken) + token, err := decodeListPageToken(result.NextPageToken) + require.NoError(t, err) + assert.Equal(t, int64(180), token.LastReceivedAtMs) + assert.Equal(t, "q/2", token.LastRequestID) +} + +func TestList_UsesCursor(t *testing.T) { + ctrl := gomock.NewController(t) + store := storagemock.NewMockRequestQueueSummaryStore(ctrl) + token := encodeListPageToken(listPageToken{Queue: "q", ReceivedAtOrAfterMs: 100, ReceivedBeforeMs: 200, LastReceivedAtMs: 180, LastRequestID: "q/2"}) + store.EXPECT().List(gomock.Any(), storage.RequestQueueSummaryQuery{ + Queue: "q", ReceivedAtOrAfterMs: 100, ReceivedBeforeMs: 200, Limit: 51, + HasCursor: true, Cursor: storage.RequestQueueSummaryCursor{ReceivedAtMs: 180, RequestID: "q/2"}, + }).Return([]entity.RequestQueueSummary{}, nil) + controller := newConfiguredListController(ctrl, store) + + result, err := controller.List(context.Background(), entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 100, ReceivedBeforeMs: 200, PageToken: token}) + + require.NoError(t, err) + assert.Empty(t, result.Requests) + assert.Empty(t, result.NextPageToken) +} + +func TestList_Errors(t *testing.T) { + validToken := encodeListPageToken(listPageToken{Queue: "other", ReceivedAtOrAfterMs: 100, ReceivedBeforeMs: 200, LastReceivedAtMs: 150, LastRequestID: "other/1"}) + invalidFieldsToken := base64.RawURLEncoding.EncodeToString([]byte("queue=q&received_at_or_after_ms=100&received_before_ms=200&last_received_at_ms=150")) + invalidNumberToken := base64.RawURLEncoding.EncodeToString([]byte("queue=q&received_at_or_after_ms=x&received_before_ms=200&last_received_at_ms=150&last_request_id=q%2F1")) + backendErr := fmt.Errorf("store down") + tests := []struct { + name string + request entity.ListRequest + setup func(*storagemock.MockRequestQueueSummaryStore) + wantInvalid bool + wantUnknown bool + }{ + {name: "empty queue", request: entity.ListRequest{ReceivedAtOrAfterMs: 1, ReceivedBeforeMs: 2}, wantInvalid: true}, + {name: "unknown queue", request: entity.ListRequest{Queue: "missing", ReceivedAtOrAfterMs: 1, ReceivedBeforeMs: 2}, wantUnknown: true}, + {name: "invalid range", request: entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 2, ReceivedBeforeMs: 2}, wantInvalid: true}, + {name: "negative page size", request: entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 1, ReceivedBeforeMs: 2, PageSize: -1}, wantInvalid: true}, + {name: "page size above maximum", request: entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 1, ReceivedBeforeMs: 2, PageSize: 201}, wantInvalid: true}, + {name: "malformed token", request: entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 1, ReceivedBeforeMs: 2, PageToken: "%%%"}, wantInvalid: true}, + {name: "invalid token number", request: entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 100, ReceivedBeforeMs: 200, PageToken: invalidNumberToken}, wantInvalid: true}, + {name: "invalid token fields", request: entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 100, ReceivedBeforeMs: 200, PageToken: invalidFieldsToken}, wantInvalid: true}, + {name: "token query mismatch", request: entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 100, ReceivedBeforeMs: 200, PageToken: validToken}, wantInvalid: true}, + { + name: "store failure", + request: entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 1, ReceivedBeforeMs: 2}, + setup: func(store *storagemock.MockRequestQueueSummaryStore) { + store.EXPECT().List(gomock.Any(), storage.RequestQueueSummaryQuery{Queue: "q", ReceivedAtOrAfterMs: 1, ReceivedBeforeMs: 2, Limit: 51}).Return(nil, backendErr) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + store := storagemock.NewMockRequestQueueSummaryStore(ctrl) + queueConfigs := qcmock.NewMockStore(ctrl) + if tt.request.Queue != "" { + if tt.wantUnknown { + queueConfigs.EXPECT().Get(gomock.Any(), tt.request.Queue).Return(entity.QueueConfig{}, queueconfig.ErrNotFound) + } else { + queueConfigs.EXPECT().Get(gomock.Any(), tt.request.Queue).Return(entity.QueueConfig{}, nil) + } + } + if tt.setup != nil { + tt.setup(store) + } + controller := NewListController(zap.NewNop().Sugar(), tally.NoopScope, store, queueConfigs) + _, err := controller.List(context.Background(), tt.request) + require.Error(t, err) + if tt.wantInvalid { + assert.True(t, IsInvalidRequest(err)) + } + assert.Equal(t, tt.wantUnknown, IsUnrecognizedQueue(err)) + }) + } +} + +func newConfiguredListController(ctrl *gomock.Controller, store storage.RequestQueueSummaryStore) *ListController { + queueConfigs := qcmock.NewMockStore(ctrl) + queueConfigs.EXPECT().Get(gomock.Any(), "q").Return(entity.QueueConfig{}, nil) + return NewListController(zap.NewNop().Sugar(), tally.NoopScope, store, queueConfigs) +} diff --git a/test/integration/submitqueue/gateway/suite_test.go b/test/integration/submitqueue/gateway/suite_test.go index cbf087b5..d52fea65 100644 --- a/test/integration/submitqueue/gateway/suite_test.go +++ b/test/integration/submitqueue/gateway/suite_test.go @@ -170,14 +170,56 @@ func (s *GatewayIntegrationSuite) TestLandAPI() { s.log.Logf("Land API test passed: request stored and message published") } -// TestRequestSummaryAPIErrorCodes verifies request-summary controller errors reach stable gRPC codes. -func (s *GatewayIntegrationSuite) TestRequestSummaryAPIErrorCodes() { +// TestListAPI verifies the queue projection is exposed in deterministic receipt order. +func (s *GatewayIntegrationSuite) TestListAPI() { + t := s.T() + store, err := mysqlstorage.NewStorage(s.db, tally.NoopScope) + require.NoError(t, err) + materializer := corerequest.NewMaterializer(store) + for _, summary := range []entity.RequestSummary{ + {RequestID: "test-queue/list-1", Queue: "test-queue", ChangeURIs: []string{"uri/1"}, ReceivedAtMs: 100, Status: entity.RequestStatusAccepted, StatusTimestampMs: 100, Version: 1, Metadata: map[string]string{}}, + {RequestID: "test-queue/list-2", Queue: "test-queue", ChangeURIs: []string{"uri/2"}, ReceivedAtMs: 200, Status: entity.RequestStatusLanded, StatusTimestampMs: 200, Version: 1, Metadata: map[string]string{}}, + } { + publicStatus := summary.Status + summary.Status = entity.RequestStatusAccepting + require.NoError(t, store.GetRequestSummaryStore().Create(s.ctx, summary)) + require.NoError(t, materializer.PersistLog(s.ctx, entity.RequestLog{ + RequestID: summary.RequestID, + TimestampMs: summary.StatusTimestampMs, + Status: publicStatus, + Metadata: map[string]string{}, + })) + } + + resp, err := s.client.List(s.ctx, &pb.ListRequest{Queue: "test-queue", ReceivedAtOrAfterMs: 50, ReceivedBeforeMs: 250, PageSize: 1}) + require.NoError(t, err) + require.Len(t, resp.Requests, 1) + assert.Equal(t, "test-queue/list-2", resp.Requests[0].Sqid) + require.NotEmpty(t, resp.NextPageToken) + + resp, err = s.client.List(s.ctx, &pb.ListRequest{Queue: "test-queue", ReceivedAtOrAfterMs: 50, ReceivedBeforeMs: 250, PageSize: 1, PageToken: resp.NextPageToken}) + require.NoError(t, err) + require.Len(t, resp.Requests, 1) + assert.Equal(t, "test-queue/list-1", resp.Requests[0].Sqid) + assert.Empty(t, resp.NextPageToken) +} + +// TestReadAPIErrorCodes verifies controller error classes reach stable gRPC codes. +func (s *GatewayIntegrationSuite) TestReadAPIErrorCodes() { t := s.T() _, err := s.client.GetRequestSummaryByID(s.ctx, &pb.GetRequestSummaryByIDRequest{Sqid: "missing/1"}) require.Error(t, err) assert.Equal(t, codes.NotFound, status.Code(err)) + _, err = s.client.List(s.ctx, &pb.ListRequest{ + Queue: "missing-queue", + ReceivedAtOrAfterMs: 1, + ReceivedBeforeMs: 2, + }) + require.Error(t, err) + assert.Equal(t, codes.InvalidArgument, status.Code(err)) + store, err := mysqlstorage.NewStorage(s.db, tally.NoopScope) require.NoError(t, err) const overflowChangeURI = "uri/read-api-overflow"