From ff85f21f5d4d363c1d3d7f89535efb44310fe33d Mon Sep 17 00:00:00 2001 From: BergaBruh <49639421+BergaBruh@users.noreply.github.com> Date: Sat, 9 May 2026 18:21:43 +0500 Subject: [PATCH] proto: add RecordSpeedTest RPC + request/response messages --- gen/stats/v1/stats.pb.go | 191 +++++++++++++++++++++++++++++++--- gen/stats/v1/stats_grpc.pb.go | 46 +++++++- proto/stats.proto | 25 +++++ 3 files changed, 243 insertions(+), 19 deletions(-) diff --git a/gen/stats/v1/stats.pb.go b/gen/stats/v1/stats.pb.go index 0d326cf..68144cb 100644 --- a/gen/stats/v1/stats.pb.go +++ b/gen/stats/v1/stats.pb.go @@ -450,6 +450,150 @@ func (x *Frame) GetPadding() []byte { return nil } +type RecordSpeedTestRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Unix epoch seconds. Required. + Timestamp int64 `protobuf:"varint,1,opt,name=timestamp,proto3" json:"timestamp,omitempty"` + // Node identifier: "fl" or "pl1". Required. + Node string `protobuf:"bytes,2,opt,name=node,proto3" json:"node,omitempty"` + // ISO-2 country code resolved from client IP at the edge node. "XX" if unknown. + Country string `protobuf:"bytes,3,opt,name=country,proto3" json:"country,omitempty"` + // Mbps. >= 0. + DlMbps float64 `protobuf:"fixed64,4,opt,name=dl_mbps,json=dlMbps,proto3" json:"dl_mbps,omitempty"` + // Mbps. >= 0. + UlMbps float64 `protobuf:"fixed64,5,opt,name=ul_mbps,json=ulMbps,proto3" json:"ul_mbps,omitempty"` + // Milliseconds. >= 0. + PingMs float64 `protobuf:"fixed64,6,opt,name=ping_ms,json=pingMs,proto3" json:"ping_ms,omitempty"` + // Milliseconds. >= 0. + JitterMs float64 `protobuf:"fixed64,7,opt,name=jitter_ms,json=jitterMs,proto3" json:"jitter_ms,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *RecordSpeedTestRequest) Reset() { + *x = RecordSpeedTestRequest{} + mi := &file_stats_proto_msgTypes[6] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *RecordSpeedTestRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RecordSpeedTestRequest) ProtoMessage() {} + +func (x *RecordSpeedTestRequest) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[6] + 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 RecordSpeedTestRequest.ProtoReflect.Descriptor instead. +func (*RecordSpeedTestRequest) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{6} +} + +func (x *RecordSpeedTestRequest) GetTimestamp() int64 { + if x != nil { + return x.Timestamp + } + return 0 +} + +func (x *RecordSpeedTestRequest) GetNode() string { + if x != nil { + return x.Node + } + return "" +} + +func (x *RecordSpeedTestRequest) GetCountry() string { + if x != nil { + return x.Country + } + return "" +} + +func (x *RecordSpeedTestRequest) GetDlMbps() float64 { + if x != nil { + return x.DlMbps + } + return 0 +} + +func (x *RecordSpeedTestRequest) GetUlMbps() float64 { + if x != nil { + return x.UlMbps + } + return 0 +} + +func (x *RecordSpeedTestRequest) GetPingMs() float64 { + if x != nil { + return x.PingMs + } + return 0 +} + +func (x *RecordSpeedTestRequest) GetJitterMs() float64 { + if x != nil { + return x.JitterMs + } + return 0 +} + +type RecordSpeedTestResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Server-assigned monotonic ID for the inserted row. + Id int64 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *RecordSpeedTestResponse) Reset() { + *x = RecordSpeedTestResponse{} + mi := &file_stats_proto_msgTypes[7] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *RecordSpeedTestResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RecordSpeedTestResponse) ProtoMessage() {} + +func (x *RecordSpeedTestResponse) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[7] + 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 RecordSpeedTestResponse.ProtoReflect.Descriptor instead. +func (*RecordSpeedTestResponse) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{7} +} + +func (x *RecordSpeedTestResponse) GetId() int64 { + if x != nil { + return x.Id + } + return 0 +} + var File_stats_proto protoreflect.FileDescriptor const file_stats_proto_rawDesc = "" + @@ -486,11 +630,22 @@ const file_stats_proto_rawDesc = "" + "\aUNKNOWN\x10\x00\x12\t\n" + "\x05HELLO\x10\x01\x12\b\n" + "\x04DATA\x10\x02\x12\t\n" + - "\x05CLOSE\x10\x032\xa7\x01\n" + + "\x05CLOSE\x10\x03\"\xcc\x01\n" + + "\x16RecordSpeedTestRequest\x12\x1c\n" + + "\ttimestamp\x18\x01 \x01(\x03R\ttimestamp\x12\x12\n" + + "\x04node\x18\x02 \x01(\tR\x04node\x12\x18\n" + + "\acountry\x18\x03 \x01(\tR\acountry\x12\x17\n" + + "\adl_mbps\x18\x04 \x01(\x01R\x06dlMbps\x12\x17\n" + + "\aul_mbps\x18\x05 \x01(\x01R\x06ulMbps\x12\x17\n" + + "\aping_ms\x18\x06 \x01(\x01R\x06pingMs\x12\x1b\n" + + "\tjitter_ms\x18\a \x01(\x01R\bjitterMs\")\n" + + "\x17RecordSpeedTestResponse\x12\x0e\n" + + "\x02id\x18\x01 \x01(\x03R\x02id2\xff\x01\n" + "\vSpeedStatus\x125\n" + "\rRecentResults\x12\x10.stats.v1.Filter\x1a\x10.stats.v1.Sample0\x01\x121\n" + "\tAggregate\x12\x0f.stats.v1.Range\x1a\x13.stats.v1.Histogram\x12.\n" + - "\x06Tunnel\x12\x0f.stats.v1.Frame\x1a\x0f.stats.v1.Frame(\x010\x01B9Z7github.com/bergabruh/stats-gateway/gen/stats/v1;statsv1b\x06proto3" + "\x06Tunnel\x12\x0f.stats.v1.Frame\x1a\x0f.stats.v1.Frame(\x010\x01\x12V\n" + + "\x0fRecordSpeedTest\x12 .stats.v1.RecordSpeedTestRequest\x1a!.stats.v1.RecordSpeedTestResponseB9Z7github.com/bergabruh/stats-gateway/gen/stats/v1;statsv1b\x06proto3" var ( file_stats_proto_rawDescOnce sync.Once @@ -505,15 +660,17 @@ func file_stats_proto_rawDescGZIP() []byte { } var file_stats_proto_enumTypes = make([]protoimpl.EnumInfo, 1) -var file_stats_proto_msgTypes = make([]protoimpl.MessageInfo, 6) +var file_stats_proto_msgTypes = make([]protoimpl.MessageInfo, 8) var file_stats_proto_goTypes = []any{ - (Frame_Kind)(0), // 0: stats.v1.Frame.Kind - (*Filter)(nil), // 1: stats.v1.Filter - (*Sample)(nil), // 2: stats.v1.Sample - (*Range)(nil), // 3: stats.v1.Range - (*Histogram)(nil), // 4: stats.v1.Histogram - (*Bucket)(nil), // 5: stats.v1.Bucket - (*Frame)(nil), // 6: stats.v1.Frame + (Frame_Kind)(0), // 0: stats.v1.Frame.Kind + (*Filter)(nil), // 1: stats.v1.Filter + (*Sample)(nil), // 2: stats.v1.Sample + (*Range)(nil), // 3: stats.v1.Range + (*Histogram)(nil), // 4: stats.v1.Histogram + (*Bucket)(nil), // 5: stats.v1.Bucket + (*Frame)(nil), // 6: stats.v1.Frame + (*RecordSpeedTestRequest)(nil), // 7: stats.v1.RecordSpeedTestRequest + (*RecordSpeedTestResponse)(nil), // 8: stats.v1.RecordSpeedTestResponse } var file_stats_proto_depIdxs = []int32{ 5, // 0: stats.v1.Histogram.buckets:type_name -> stats.v1.Bucket @@ -521,11 +678,13 @@ var file_stats_proto_depIdxs = []int32{ 1, // 2: stats.v1.SpeedStatus.RecentResults:input_type -> stats.v1.Filter 3, // 3: stats.v1.SpeedStatus.Aggregate:input_type -> stats.v1.Range 6, // 4: stats.v1.SpeedStatus.Tunnel:input_type -> stats.v1.Frame - 2, // 5: stats.v1.SpeedStatus.RecentResults:output_type -> stats.v1.Sample - 4, // 6: stats.v1.SpeedStatus.Aggregate:output_type -> stats.v1.Histogram - 6, // 7: stats.v1.SpeedStatus.Tunnel:output_type -> stats.v1.Frame - 5, // [5:8] is the sub-list for method output_type - 2, // [2:5] is the sub-list for method input_type + 7, // 5: stats.v1.SpeedStatus.RecordSpeedTest:input_type -> stats.v1.RecordSpeedTestRequest + 2, // 6: stats.v1.SpeedStatus.RecentResults:output_type -> stats.v1.Sample + 4, // 7: stats.v1.SpeedStatus.Aggregate:output_type -> stats.v1.Histogram + 6, // 8: stats.v1.SpeedStatus.Tunnel:output_type -> stats.v1.Frame + 8, // 9: stats.v1.SpeedStatus.RecordSpeedTest:output_type -> stats.v1.RecordSpeedTestResponse + 6, // [6:10] is the sub-list for method output_type + 2, // [2:6] is the sub-list for method input_type 2, // [2:2] is the sub-list for extension type_name 2, // [2:2] is the sub-list for extension extendee 0, // [0:2] is the sub-list for field type_name @@ -542,7 +701,7 @@ func file_stats_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_stats_proto_rawDesc), len(file_stats_proto_rawDesc)), NumEnums: 1, - NumMessages: 6, + NumMessages: 8, NumExtensions: 0, NumServices: 1, }, diff --git a/gen/stats/v1/stats_grpc.pb.go b/gen/stats/v1/stats_grpc.pb.go index ffb3507..b4d85c3 100644 --- a/gen/stats/v1/stats_grpc.pb.go +++ b/gen/stats/v1/stats_grpc.pb.go @@ -19,9 +19,10 @@ import ( const _ = grpc.SupportPackageIsVersion9 const ( - SpeedStatus_RecentResults_FullMethodName = "/stats.v1.SpeedStatus/RecentResults" - SpeedStatus_Aggregate_FullMethodName = "/stats.v1.SpeedStatus/Aggregate" - SpeedStatus_Tunnel_FullMethodName = "/stats.v1.SpeedStatus/Tunnel" + SpeedStatus_RecentResults_FullMethodName = "/stats.v1.SpeedStatus/RecentResults" + SpeedStatus_Aggregate_FullMethodName = "/stats.v1.SpeedStatus/Aggregate" + SpeedStatus_Tunnel_FullMethodName = "/stats.v1.SpeedStatus/Tunnel" + SpeedStatus_RecordSpeedTest_FullMethodName = "/stats.v1.SpeedStatus/RecordSpeedTest" ) // SpeedStatusClient is the client API for SpeedStatus service. @@ -36,6 +37,8 @@ type SpeedStatusClient interface { // Auth: Bearer JWT (HS256) в gRPC metadata "authorization: Bearer ". // Server validates: signature, exp, scope=="tunnel", optional revocation list. Tunnel(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[Frame, Frame], error) + // RecordSpeedTest ingests a single speedtest sample from an edge node. + RecordSpeedTest(ctx context.Context, in *RecordSpeedTestRequest, opts ...grpc.CallOption) (*RecordSpeedTestResponse, error) } type speedStatusClient struct { @@ -88,6 +91,16 @@ func (c *speedStatusClient) Tunnel(ctx context.Context, opts ...grpc.CallOption) // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type SpeedStatus_TunnelClient = grpc.BidiStreamingClient[Frame, Frame] +func (c *speedStatusClient) RecordSpeedTest(ctx context.Context, in *RecordSpeedTestRequest, opts ...grpc.CallOption) (*RecordSpeedTestResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(RecordSpeedTestResponse) + err := c.cc.Invoke(ctx, SpeedStatus_RecordSpeedTest_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // SpeedStatusServer is the server API for SpeedStatus service. // All implementations must embed UnimplementedSpeedStatusServer // for forward compatibility. @@ -100,6 +113,8 @@ type SpeedStatusServer interface { // Auth: Bearer JWT (HS256) в gRPC metadata "authorization: Bearer ". // Server validates: signature, exp, scope=="tunnel", optional revocation list. Tunnel(grpc.BidiStreamingServer[Frame, Frame]) error + // RecordSpeedTest ingests a single speedtest sample from an edge node. + RecordSpeedTest(context.Context, *RecordSpeedTestRequest) (*RecordSpeedTestResponse, error) mustEmbedUnimplementedSpeedStatusServer() } @@ -119,6 +134,9 @@ func (UnimplementedSpeedStatusServer) Aggregate(context.Context, *Range) (*Histo func (UnimplementedSpeedStatusServer) Tunnel(grpc.BidiStreamingServer[Frame, Frame]) error { return status.Error(codes.Unimplemented, "method Tunnel not implemented") } +func (UnimplementedSpeedStatusServer) RecordSpeedTest(context.Context, *RecordSpeedTestRequest) (*RecordSpeedTestResponse, error) { + return nil, status.Error(codes.Unimplemented, "method RecordSpeedTest not implemented") +} func (UnimplementedSpeedStatusServer) mustEmbedUnimplementedSpeedStatusServer() {} func (UnimplementedSpeedStatusServer) testEmbeddedByValue() {} @@ -176,6 +194,24 @@ func _SpeedStatus_Tunnel_Handler(srv interface{}, stream grpc.ServerStream) erro // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type SpeedStatus_TunnelServer = grpc.BidiStreamingServer[Frame, Frame] +func _SpeedStatus_RecordSpeedTest_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(RecordSpeedTestRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(SpeedStatusServer).RecordSpeedTest(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: SpeedStatus_RecordSpeedTest_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(SpeedStatusServer).RecordSpeedTest(ctx, req.(*RecordSpeedTestRequest)) + } + return interceptor(ctx, in, info, handler) +} + // SpeedStatus_ServiceDesc is the grpc.ServiceDesc for SpeedStatus service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -187,6 +223,10 @@ var SpeedStatus_ServiceDesc = grpc.ServiceDesc{ MethodName: "Aggregate", Handler: _SpeedStatus_Aggregate_Handler, }, + { + MethodName: "RecordSpeedTest", + Handler: _SpeedStatus_RecordSpeedTest_Handler, + }, }, Streams: []grpc.StreamDesc{ { diff --git a/proto/stats.proto b/proto/stats.proto index f34738d..c1475f8 100644 --- a/proto/stats.proto +++ b/proto/stats.proto @@ -14,6 +14,9 @@ service SpeedStatus { // Auth: Bearer JWT (HS256) в gRPC metadata "authorization: Bearer ". // Server validates: signature, exp, scope=="tunnel", optional revocation list. rpc Tunnel(stream Frame) returns (stream Frame); + + // RecordSpeedTest ingests a single speedtest sample from an edge node. + rpc RecordSpeedTest(RecordSpeedTestRequest) returns (RecordSpeedTestResponse); } message Filter { @@ -60,3 +63,25 @@ message Frame { bytes payload = 3; // raw TCP bytes bytes padding = 4; // random padding для frame-length obfuscation } + +message RecordSpeedTestRequest { + // Unix epoch seconds. Required. + int64 timestamp = 1; + // Node identifier: "fl" or "pl1". Required. + string node = 2; + // ISO-2 country code resolved from client IP at the edge node. "XX" if unknown. + string country = 3; + // Mbps. >= 0. + double dl_mbps = 4; + // Mbps. >= 0. + double ul_mbps = 5; + // Milliseconds. >= 0. + double ping_ms = 6; + // Milliseconds. >= 0. + double jitter_ms = 7; +} + +message RecordSpeedTestResponse { + // Server-assigned monotonic ID for the inserted row. + int64 id = 1; +}