hailin
/
rwadurian
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(mpc-system): replace K8s party discovery with Message Router-based discovery 

- Add GetRegisteredParties gRPC method to Message Router for party discovery
- Create MessageRouterPartyDiscovery adapter in Session Coordinator
- Remove K8s dependency from Session Coordinator (works in any environment)
- Add party registration to server-party-api on startup
- Fix docker-compose.yml: add MESSAGE_ROUTER_ADDR to session-coordinator

This change implements a fully decentralized party discovery mechanism:
- Parties register themselves to Message Router on startup
- Session Coordinator queries Message Router for available parties
- Works in Docker Compose, K8s, or any deployment environment

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
hailin 2025-12-05 16:37:16 -08:00
parent d7f181f2ec
commit c52b6aa980
15 changed files with 534 additions and 3023 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
backend/.claude/settings.local.json
View File

@ -6,7 +6,17 @@
"Bash(go list:*)",
"Bash(go get:*)",
"Bash(ls:*)",
"Bash(find:*)"
"Bash(find:*)",
"Bash(certutil:*)",
"Bash(git checkout:*)",
"Bash(wsl.exe:*)",
"Bash(protoc:*)",
"Bash(powershell -Command \"Select-String -Path ''api\\proto\\message_router_grpc.pb.go'' -Pattern ''GetRegisteredParties'' | Select-Object -First 3\")",
"Bash(copy /Y \"api\\proto\\message_router.pb.go\" \"api\\grpc\\router\\v1\"\" && copy /Y \"apiprotomessage_router_grpc.pb.go\" \"apigrpcrouterv1\"\")",
"Bash(powershell -Command:*)",
"Bash(go build:*)",
"Bash(git add:*)",
"Bash(git commit:*)"
],
"deny": [],
"ask": []

252
backend/mpc-system/api/grpc/router/v1/message_router.pb.go
View File

@ -809,6 +809,189 @@ func (x *PublishSessionEventResponse) GetSubscriberCount() int32 {
return 0
}
// GetRegisteredPartiesRequest requests registered parties list
type GetRegisteredPartiesRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
RoleFilter string `protobuf:"bytes,1,opt,name=role_filter,json=roleFilter,proto3" json:"role_filter,omitempty"` // Optional: filter by role (persistent, delegate, temporary)
OnlyOnline bool `protobuf:"varint,2,opt,name=only_online,json=onlyOnline,proto3" json:"only_online,omitempty"` // Optional: only return online parties
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *GetRegisteredPartiesRequest) Reset() {
*x = GetRegisteredPartiesRequest{}
mi := &file_api_proto_message_router_proto_msgTypes[12]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *GetRegisteredPartiesRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*GetRegisteredPartiesRequest) ProtoMessage() {}
func (x *GetRegisteredPartiesRequest) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_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 GetRegisteredPartiesRequest.ProtoReflect.Descriptor instead.
func (*GetRegisteredPartiesRequest) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{12}
}
func (x *GetRegisteredPartiesRequest) GetRoleFilter() string {
if x != nil {
return x.RoleFilter
}
return ""
}
func (x *GetRegisteredPartiesRequest) GetOnlyOnline() bool {
if x != nil {
return x.OnlyOnline
}
return false
}
// RegisteredParty represents a registered party
type RegisteredParty struct {
state protoimpl.MessageState `protogen:"open.v1"`
PartyId string `protobuf:"bytes,1,opt,name=party_id,json=partyId,proto3" json:"party_id,omitempty"` // Unique party identifier
Role string `protobuf:"bytes,2,opt,name=role,proto3" json:"role,omitempty"` // persistent, delegate, or temporary
Online bool `protobuf:"varint,3,opt,name=online,proto3" json:"online,omitempty"` // Whether party is currently connected
RegisteredAt int64 `protobuf:"varint,4,opt,name=registered_at,json=registeredAt,proto3" json:"registered_at,omitempty"` // Unix timestamp milliseconds
LastSeenAt int64 `protobuf:"varint,5,opt,name=last_seen_at,json=lastSeenAt,proto3" json:"last_seen_at,omitempty"` // Unix timestamp milliseconds
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *RegisteredParty) Reset() {
*x = RegisteredParty{}
mi := &file_api_proto_message_router_proto_msgTypes[13]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *RegisteredParty) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*RegisteredParty) ProtoMessage() {}
func (x *RegisteredParty) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_proto_msgTypes[13]
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 RegisteredParty.ProtoReflect.Descriptor instead.
func (*RegisteredParty) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{13}
}
func (x *RegisteredParty) GetPartyId() string {
if x != nil {
return x.PartyId
}
return ""
}
func (x *RegisteredParty) GetRole() string {
if x != nil {
return x.Role
}
return ""
}
func (x *RegisteredParty) GetOnline() bool {
if x != nil {
return x.Online
}
return false
}
func (x *RegisteredParty) GetRegisteredAt() int64 {
if x != nil {
return x.RegisteredAt
}
return 0
}
func (x *RegisteredParty) GetLastSeenAt() int64 {
if x != nil {
return x.LastSeenAt
}
return 0
}
// GetRegisteredPartiesResponse returns registered parties
type GetRegisteredPartiesResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Parties []*RegisteredParty `protobuf:"bytes,1,rep,name=parties,proto3" json:"parties,omitempty"`
TotalCount int32 `protobuf:"varint,2,opt,name=total_count,json=totalCount,proto3" json:"total_count,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *GetRegisteredPartiesResponse) Reset() {
*x = GetRegisteredPartiesResponse{}
mi := &file_api_proto_message_router_proto_msgTypes[14]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *GetRegisteredPartiesResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*GetRegisteredPartiesResponse) ProtoMessage() {}
func (x *GetRegisteredPartiesResponse) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_proto_msgTypes[14]
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 GetRegisteredPartiesResponse.ProtoReflect.Descriptor instead.
func (*GetRegisteredPartiesResponse) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{14}
}
func (x *GetRegisteredPartiesResponse) GetParties() []*RegisteredParty {
if x != nil {
return x.Parties
}
return nil
}
func (x *GetRegisteredPartiesResponse) GetTotalCount() int32 {
if x != nil {
return x.TotalCount
}
return 0
}
var File_api_proto_message_router_proto protoreflect.FileDescriptor
const file_api_proto_message_router_proto_rawDesc = "" +
@ -892,14 +1075,31 @@ const file_api_proto_message_router_proto_rawDesc = "" +
"\x05event\x18\x01 \x01(\v2\x1b.mpc.router.v1.SessionEventR\x05event\"b\n" +
"\x1bPublishSessionEventResponse\x12\x18\n" +
"\asuccess\x18\x01 \x01(\bR\asuccess\x12)\n" +
"\x10subscriber_count\x18\x02 \x01(\x05R\x0fsubscriberCount2\xdf\x04\n" +
"\x10subscriber_count\x18\x02 \x01(\x05R\x0fsubscriberCount\"_\n" +
"\x1bGetRegisteredPartiesRequest\x12\x1f\n" +
"\vrole_filter\x18\x01 \x01(\tR\n" +
"roleFilter\x12\x1f\n" +
"\vonly_online\x18\x02 \x01(\bR\n" +
"onlyOnline\"\x9f\x01\n" +
"\x0fRegisteredParty\x12\x19\n" +
"\bparty_id\x18\x01 \x01(\tR\apartyId\x12\x12\n" +
"\x04role\x18\x02 \x01(\tR\x04role\x12\x16\n" +
"\x06online\x18\x03 \x01(\bR\x06online\x12#\n" +
"\rregistered_at\x18\x04 \x01(\x03R\fregisteredAt\x12 \n" +
"\flast_seen_at\x18\x05 \x01(\x03R\n" +
"lastSeenAt\"y\n" +
"\x1cGetRegisteredPartiesResponse\x128\n" +
"\aparties\x18\x01 \x03(\v2\x1e.mpc.router.v1.RegisteredPartyR\aparties\x12\x1f\n" +
"\vtotal_count\x18\x02 \x01(\x05R\n" +
"totalCount2\xd0\x05\n" +
"\rMessageRouter\x12W\n" +
"\fRouteMessage\x12\".mpc.router.v1.RouteMessageRequest\x1a#.mpc.router.v1.RouteMessageResponse\x12Y\n" +
"\x11SubscribeMessages\x12'.mpc.router.v1.SubscribeMessagesRequest\x1a\x19.mpc.router.v1.MPCMessage0\x01\x12i\n" +
"\x12GetPendingMessages\x12(.mpc.router.v1.GetPendingMessagesRequest\x1a).mpc.router.v1.GetPendingMessagesResponse\x12Z\n" +
"\rRegisterParty\x12#.mpc.router.v1.RegisterPartyRequest\x1a$.mpc.router.v1.RegisterPartyResponse\x12e\n" +
"\x16SubscribeSessionEvents\x12,.mpc.router.v1.SubscribeSessionEventsRequest\x1a\x1b.mpc.router.v1.SessionEvent0\x01\x12l\n" +
"\x13PublishSessionEvent\x12).mpc.router.v1.PublishSessionEventRequest\x1a*.mpc.router.v1.PublishSessionEventResponseB;Z9github.com/rwadurian/mpc-system/api/grpc/router/v1;routerb\x06proto3"
"\x13PublishSessionEvent\x12).mpc.router.v1.PublishSessionEventRequest\x1a*.mpc.router.v1.PublishSessionEventResponse\x12o\n" +
"\x14GetRegisteredParties\x12*.mpc.router.v1.GetRegisteredPartiesRequest\x1a+.mpc.router.v1.GetRegisteredPartiesResponseB;Z9github.com/rwadurian/mpc-system/api/grpc/router/v1;routerb\x06proto3"
var (
file_api_proto_message_router_proto_rawDescOnce sync.Once
@ -913,7 +1113,7 @@ func file_api_proto_message_router_proto_rawDescGZIP() []byte {
return file_api_proto_message_router_proto_rawDescData
}
var file_api_proto_message_router_proto_msgTypes = make([]protoimpl.MessageInfo, 13)
var file_api_proto_message_router_proto_msgTypes = make([]protoimpl.MessageInfo, 16)
var file_api_proto_message_router_proto_goTypes = []any{
(*RouteMessageRequest)(nil), // 0: mpc.router.v1.RouteMessageRequest
(*RouteMessageResponse)(nil), // 1: mpc.router.v1.RouteMessageResponse
@ -927,29 +1127,35 @@ var file_api_proto_message_router_proto_goTypes = []any{
(*SessionEvent)(nil), // 9: mpc.router.v1.SessionEvent
(*PublishSessionEventRequest)(nil), // 10: mpc.router.v1.PublishSessionEventRequest
(*PublishSessionEventResponse)(nil), // 11: mpc.router.v1.PublishSessionEventResponse
nil, // 12: mpc.router.v1.SessionEvent.JoinTokensEntry
(*GetRegisteredPartiesRequest)(nil), // 12: mpc.router.v1.GetRegisteredPartiesRequest
(*RegisteredParty)(nil), // 13: mpc.router.v1.RegisteredParty
(*GetRegisteredPartiesResponse)(nil), // 14: mpc.router.v1.GetRegisteredPartiesResponse
nil, // 15: mpc.router.v1.SessionEvent.JoinTokensEntry
}
var file_api_proto_message_router_proto_depIdxs = []int32{
3, // 0: mpc.router.v1.GetPendingMessagesResponse.messages:type_name -> mpc.router.v1.MPCMessage
12, // 1: mpc.router.v1.SessionEvent.join_tokens:type_name -> mpc.router.v1.SessionEvent.JoinTokensEntry
15, // 1: mpc.router.v1.SessionEvent.join_tokens:type_name -> mpc.router.v1.SessionEvent.JoinTokensEntry
9, // 2: mpc.router.v1.PublishSessionEventRequest.event:type_name -> mpc.router.v1.SessionEvent
0, // 3: mpc.router.v1.MessageRouter.RouteMessage:input_type -> mpc.router.v1.RouteMessageRequest
2, // 4: mpc.router.v1.MessageRouter.SubscribeMessages:input_type -> mpc.router.v1.SubscribeMessagesRequest
4, // 5: mpc.router.v1.MessageRouter.GetPendingMessages:input_type -> mpc.router.v1.GetPendingMessagesRequest
6, // 6: mpc.router.v1.MessageRouter.RegisterParty:input_type -> mpc.router.v1.RegisterPartyRequest
8, // 7: mpc.router.v1.MessageRouter.SubscribeSessionEvents:input_type -> mpc.router.v1.SubscribeSessionEventsRequest
10, // 8: mpc.router.v1.MessageRouter.PublishSessionEvent:input_type -> mpc.router.v1.PublishSessionEventRequest
1, // 9: mpc.router.v1.MessageRouter.RouteMessage:output_type -> mpc.router.v1.RouteMessageResponse
3, // 10: mpc.router.v1.MessageRouter.SubscribeMessages:output_type -> mpc.router.v1.MPCMessage
5, // 11: mpc.router.v1.MessageRouter.GetPendingMessages:output_type -> mpc.router.v1.GetPendingMessagesResponse
7, // 12: mpc.router.v1.MessageRouter.RegisterParty:output_type -> mpc.router.v1.RegisterPartyResponse
9, // 13: mpc.router.v1.MessageRouter.SubscribeSessionEvents:output_type -> mpc.router.v1.SessionEvent
11, // 14: mpc.router.v1.MessageRouter.PublishSessionEvent:output_type -> mpc.router.v1.PublishSessionEventResponse
9, // [9:15] is the sub-list for method output_type
3, // [3:9] is the sub-list for method input_type
3, // [3:3] is the sub-list for extension type_name
3, // [3:3] is the sub-list for extension extendee
0, // [0:3] is the sub-list for field type_name
13, // 3: mpc.router.v1.GetRegisteredPartiesResponse.parties:type_name -> mpc.router.v1.RegisteredParty
0, // 4: mpc.router.v1.MessageRouter.RouteMessage:input_type -> mpc.router.v1.RouteMessageRequest
2, // 5: mpc.router.v1.MessageRouter.SubscribeMessages:input_type -> mpc.router.v1.SubscribeMessagesRequest
4, // 6: mpc.router.v1.MessageRouter.GetPendingMessages:input_type -> mpc.router.v1.GetPendingMessagesRequest
6, // 7: mpc.router.v1.MessageRouter.RegisterParty:input_type -> mpc.router.v1.RegisterPartyRequest
8, // 8: mpc.router.v1.MessageRouter.SubscribeSessionEvents:input_type -> mpc.router.v1.SubscribeSessionEventsRequest
10, // 9: mpc.router.v1.MessageRouter.PublishSessionEvent:input_type -> mpc.router.v1.PublishSessionEventRequest
12, // 10: mpc.router.v1.MessageRouter.GetRegisteredParties:input_type -> mpc.router.v1.GetRegisteredPartiesRequest
1, // 11: mpc.router.v1.MessageRouter.RouteMessage:output_type -> mpc.router.v1.RouteMessageResponse
3, // 12: mpc.router.v1.MessageRouter.SubscribeMessages:output_type -> mpc.router.v1.MPCMessage
5, // 13: mpc.router.v1.MessageRouter.GetPendingMessages:output_type -> mpc.router.v1.GetPendingMessagesResponse
7, // 14: mpc.router.v1.MessageRouter.RegisterParty:output_type -> mpc.router.v1.RegisterPartyResponse
9, // 15: mpc.router.v1.MessageRouter.SubscribeSessionEvents:output_type -> mpc.router.v1.SessionEvent
11, // 16: mpc.router.v1.MessageRouter.PublishSessionEvent:output_type -> mpc.router.v1.PublishSessionEventResponse
14, // 17: mpc.router.v1.MessageRouter.GetRegisteredParties:output_type -> mpc.router.v1.GetRegisteredPartiesResponse
11, // [11:18] is the sub-list for method output_type
4, // [4:11] is the sub-list for method input_type
4, // [4:4] is the sub-list for extension type_name
4, // [4:4] is the sub-list for extension extendee
0, // [0:4] is the sub-list for field type_name
}
func init() { file_api_proto_message_router_proto_init() }
@ -963,7 +1169,7 @@ func file_api_proto_message_router_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_api_proto_message_router_proto_rawDesc), len(file_api_proto_message_router_proto_rawDesc)),
NumEnums: 0,
NumMessages: 13,
NumMessages: 16,
NumExtensions: 0,
NumServices: 1,
},

40
backend/mpc-system/api/grpc/router/v1/message_router_grpc.pb.go
View File

@ -25,6 +25,7 @@ const (
MessageRouter_RegisterParty_FullMethodName = "/mpc.router.v1.MessageRouter/RegisterParty"
MessageRouter_SubscribeSessionEvents_FullMethodName = "/mpc.router.v1.MessageRouter/SubscribeSessionEvents"
MessageRouter_PublishSessionEvent_FullMethodName = "/mpc.router.v1.MessageRouter/PublishSessionEvent"
MessageRouter_GetRegisteredParties_FullMethodName = "/mpc.router.v1.MessageRouter/GetRegisteredParties"
)
// MessageRouterClient is the client API for MessageRouter service.
@ -45,6 +46,8 @@ type MessageRouterClient interface {
SubscribeSessionEvents(ctx context.Context, in *SubscribeSessionEventsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[SessionEvent], error)
// PublishSessionEvent publishes a session event (called by Session Coordinator)
PublishSessionEvent(ctx context.Context, in *PublishSessionEventRequest, opts ...grpc.CallOption) (*PublishSessionEventResponse, error)
// GetRegisteredParties returns all registered parties (for Session Coordinator party discovery)
GetRegisteredParties(ctx context.Context, in *GetRegisteredPartiesRequest, opts ...grpc.CallOption) (*GetRegisteredPartiesResponse, error)
}
type messageRouterClient struct {
@ -133,6 +136,16 @@ func (c *messageRouterClient) PublishSessionEvent(ctx context.Context, in *Publi
return out, nil
}
func (c *messageRouterClient) GetRegisteredParties(ctx context.Context, in *GetRegisteredPartiesRequest, opts ...grpc.CallOption) (*GetRegisteredPartiesResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(GetRegisteredPartiesResponse)
err := c.cc.Invoke(ctx, MessageRouter_GetRegisteredParties_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
// MessageRouterServer is the server API for MessageRouter service.
// All implementations must embed UnimplementedMessageRouterServer
// for forward compatibility.
@ -151,6 +164,8 @@ type MessageRouterServer interface {
SubscribeSessionEvents(*SubscribeSessionEventsRequest, grpc.ServerStreamingServer[SessionEvent]) error
// PublishSessionEvent publishes a session event (called by Session Coordinator)
PublishSessionEvent(context.Context, *PublishSessionEventRequest) (*PublishSessionEventResponse, error)
// GetRegisteredParties returns all registered parties (for Session Coordinator party discovery)
GetRegisteredParties(context.Context, *GetRegisteredPartiesRequest) (*GetRegisteredPartiesResponse, error)
mustEmbedUnimplementedMessageRouterServer()
}
@ -179,6 +194,9 @@ func (UnimplementedMessageRouterServer) SubscribeSessionEvents(*SubscribeSession
func (UnimplementedMessageRouterServer) PublishSessionEvent(context.Context, *PublishSessionEventRequest) (*PublishSessionEventResponse, error) {
return nil, status.Error(codes.Unimplemented, "method PublishSessionEvent not implemented")
}
func (UnimplementedMessageRouterServer) GetRegisteredParties(context.Context, *GetRegisteredPartiesRequest) (*GetRegisteredPartiesResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetRegisteredParties not implemented")
}
func (UnimplementedMessageRouterServer) mustEmbedUnimplementedMessageRouterServer() {}
func (UnimplementedMessageRouterServer) testEmbeddedByValue() {}
@ -294,6 +312,24 @@ func _MessageRouter_PublishSessionEvent_Handler(srv interface{}, ctx context.Con
return interceptor(ctx, in, info, handler)
}
func _MessageRouter_GetRegisteredParties_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(GetRegisteredPartiesRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(MessageRouterServer).GetRegisteredParties(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: MessageRouter_GetRegisteredParties_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(MessageRouterServer).GetRegisteredParties(ctx, req.(*GetRegisteredPartiesRequest))
}
return interceptor(ctx, in, info, handler)
}
// MessageRouter_ServiceDesc is the grpc.ServiceDesc for MessageRouter service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
@ -317,6 +353,10 @@ var MessageRouter_ServiceDesc = grpc.ServiceDesc{
MethodName: "PublishSessionEvent",
Handler: _MessageRouter_PublishSessionEvent_Handler,
},
{
MethodName: "GetRegisteredParties",
Handler: _MessageRouter_GetRegisteredParties_Handler,
},
},
Streams: []grpc.StreamDesc{
{

977
backend/mpc-system/api/proto/message_router.pb.go
View File

@ -1,977 +0,0 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.10
// protoc v6.33.1
// source: api/proto/message_router.proto
package router
import (
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
unsafe "unsafe"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
// RouteMessageRequest routes an MPC message
type RouteMessageRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
SessionId string `protobuf:"bytes,1,opt,name=session_id,json=sessionId,proto3" json:"session_id,omitempty"`
FromParty string `protobuf:"bytes,2,opt,name=from_party,json=fromParty,proto3" json:"from_party,omitempty"`
ToParties []string `protobuf:"bytes,3,rep,name=to_parties,json=toParties,proto3" json:"to_parties,omitempty"` // Empty for broadcast
RoundNumber int32 `protobuf:"varint,4,opt,name=round_number,json=roundNumber,proto3" json:"round_number,omitempty"`
MessageType string `protobuf:"bytes,5,opt,name=message_type,json=messageType,proto3" json:"message_type,omitempty"`
Payload []byte `protobuf:"bytes,6,opt,name=payload,proto3" json:"payload,omitempty"` // Encrypted MPC message
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *RouteMessageRequest) Reset() {
*x = RouteMessageRequest{}
mi := &file_api_proto_message_router_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *RouteMessageRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*RouteMessageRequest) ProtoMessage() {}
func (x *RouteMessageRequest) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_proto_msgTypes[0]
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 RouteMessageRequest.ProtoReflect.Descriptor instead.
func (*RouteMessageRequest) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{0}
}
func (x *RouteMessageRequest) GetSessionId() string {
if x != nil {
return x.SessionId
}
return ""
}
func (x *RouteMessageRequest) GetFromParty() string {
if x != nil {
return x.FromParty
}
return ""
}
func (x *RouteMessageRequest) GetToParties() []string {
if x != nil {
return x.ToParties
}
return nil
}
func (x *RouteMessageRequest) GetRoundNumber() int32 {
if x != nil {
return x.RoundNumber
}
return 0
}
func (x *RouteMessageRequest) GetMessageType() string {
if x != nil {
return x.MessageType
}
return ""
}
func (x *RouteMessageRequest) GetPayload() []byte {
if x != nil {
return x.Payload
}
return nil
}
// RouteMessageResponse confirms message routing
type RouteMessageResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Success bool `protobuf:"varint,1,opt,name=success,proto3" json:"success,omitempty"`
MessageId string `protobuf:"bytes,2,opt,name=message_id,json=messageId,proto3" json:"message_id,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *RouteMessageResponse) Reset() {
*x = RouteMessageResponse{}
mi := &file_api_proto_message_router_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *RouteMessageResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*RouteMessageResponse) ProtoMessage() {}
func (x *RouteMessageResponse) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_proto_msgTypes[1]
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 RouteMessageResponse.ProtoReflect.Descriptor instead.
func (*RouteMessageResponse) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{1}
}
func (x *RouteMessageResponse) GetSuccess() bool {
if x != nil {
return x.Success
}
return false
}
func (x *RouteMessageResponse) GetMessageId() string {
if x != nil {
return x.MessageId
}
return ""
}
// SubscribeMessagesRequest subscribes to messages for a party
type SubscribeMessagesRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
SessionId string `protobuf:"bytes,1,opt,name=session_id,json=sessionId,proto3" json:"session_id,omitempty"`
PartyId string `protobuf:"bytes,2,opt,name=party_id,json=partyId,proto3" json:"party_id,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *SubscribeMessagesRequest) Reset() {
*x = SubscribeMessagesRequest{}
mi := &file_api_proto_message_router_proto_msgTypes[2]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *SubscribeMessagesRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*SubscribeMessagesRequest) ProtoMessage() {}
func (x *SubscribeMessagesRequest) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_proto_msgTypes[2]
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 SubscribeMessagesRequest.ProtoReflect.Descriptor instead.
func (*SubscribeMessagesRequest) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{2}
}
func (x *SubscribeMessagesRequest) GetSessionId() string {
if x != nil {
return x.SessionId
}
return ""
}
func (x *SubscribeMessagesRequest) GetPartyId() string {
if x != nil {
return x.PartyId
}
return ""
}
// MPCMessage represents an MPC protocol message
type MPCMessage struct {
state protoimpl.MessageState `protogen:"open.v1"`
MessageId string `protobuf:"bytes,1,opt,name=message_id,json=messageId,proto3" json:"message_id,omitempty"`
SessionId string `protobuf:"bytes,2,opt,name=session_id,json=sessionId,proto3" json:"session_id,omitempty"`
FromParty string `protobuf:"bytes,3,opt,name=from_party,json=fromParty,proto3" json:"from_party,omitempty"`
IsBroadcast bool `protobuf:"varint,4,opt,name=is_broadcast,json=isBroadcast,proto3" json:"is_broadcast,omitempty"`
RoundNumber int32 `protobuf:"varint,5,opt,name=round_number,json=roundNumber,proto3" json:"round_number,omitempty"`
MessageType string `protobuf:"bytes,6,opt,name=message_type,json=messageType,proto3" json:"message_type,omitempty"`
Payload []byte `protobuf:"bytes,7,opt,name=payload,proto3" json:"payload,omitempty"`
CreatedAt int64 `protobuf:"varint,8,opt,name=created_at,json=createdAt,proto3" json:"created_at,omitempty"` // Unix timestamp milliseconds
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *MPCMessage) Reset() {
*x = MPCMessage{}
mi := &file_api_proto_message_router_proto_msgTypes[3]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *MPCMessage) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*MPCMessage) ProtoMessage() {}
func (x *MPCMessage) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_proto_msgTypes[3]
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 MPCMessage.ProtoReflect.Descriptor instead.
func (*MPCMessage) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{3}
}
func (x *MPCMessage) GetMessageId() string {
if x != nil {
return x.MessageId
}
return ""
}
func (x *MPCMessage) GetSessionId() string {
if x != nil {
return x.SessionId
}
return ""
}
func (x *MPCMessage) GetFromParty() string {
if x != nil {
return x.FromParty
}
return ""
}
func (x *MPCMessage) GetIsBroadcast() bool {
if x != nil {
return x.IsBroadcast
}
return false
}
func (x *MPCMessage) GetRoundNumber() int32 {
if x != nil {
return x.RoundNumber
}
return 0
}
func (x *MPCMessage) GetMessageType() string {
if x != nil {
return x.MessageType
}
return ""
}
func (x *MPCMessage) GetPayload() []byte {
if x != nil {
return x.Payload
}
return nil
}
func (x *MPCMessage) GetCreatedAt() int64 {
if x != nil {
return x.CreatedAt
}
return 0
}
// GetPendingMessagesRequest retrieves pending messages
type GetPendingMessagesRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
SessionId string `protobuf:"bytes,1,opt,name=session_id,json=sessionId,proto3" json:"session_id,omitempty"`
PartyId string `protobuf:"bytes,2,opt,name=party_id,json=partyId,proto3" json:"party_id,omitempty"`
AfterTimestamp int64 `protobuf:"varint,3,opt,name=after_timestamp,json=afterTimestamp,proto3" json:"after_timestamp,omitempty"` // Get messages after this timestamp
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *GetPendingMessagesRequest) Reset() {
*x = GetPendingMessagesRequest{}
mi := &file_api_proto_message_router_proto_msgTypes[4]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *GetPendingMessagesRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*GetPendingMessagesRequest) ProtoMessage() {}
func (x *GetPendingMessagesRequest) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_proto_msgTypes[4]
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 GetPendingMessagesRequest.ProtoReflect.Descriptor instead.
func (*GetPendingMessagesRequest) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{4}
}
func (x *GetPendingMessagesRequest) GetSessionId() string {
if x != nil {
return x.SessionId
}
return ""
}
func (x *GetPendingMessagesRequest) GetPartyId() string {
if x != nil {
return x.PartyId
}
return ""
}
func (x *GetPendingMessagesRequest) GetAfterTimestamp() int64 {
if x != nil {
return x.AfterTimestamp
}
return 0
}
// GetPendingMessagesResponse contains pending messages
type GetPendingMessagesResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Messages []*MPCMessage `protobuf:"bytes,1,rep,name=messages,proto3" json:"messages,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *GetPendingMessagesResponse) Reset() {
*x = GetPendingMessagesResponse{}
mi := &file_api_proto_message_router_proto_msgTypes[5]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *GetPendingMessagesResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*GetPendingMessagesResponse) ProtoMessage() {}
func (x *GetPendingMessagesResponse) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_proto_msgTypes[5]
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 GetPendingMessagesResponse.ProtoReflect.Descriptor instead.
func (*GetPendingMessagesResponse) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{5}
}
func (x *GetPendingMessagesResponse) GetMessages() []*MPCMessage {
if x != nil {
return x.Messages
}
return nil
}
// RegisterPartyRequest registers a party with the router
type RegisterPartyRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
PartyId string `protobuf:"bytes,1,opt,name=party_id,json=partyId,proto3" json:"party_id,omitempty"` // Unique party identifier
PartyRole string `protobuf:"bytes,2,opt,name=party_role,json=partyRole,proto3" json:"party_role,omitempty"` // persistent, delegate, or temporary
Version string `protobuf:"bytes,3,opt,name=version,proto3" json:"version,omitempty"` // Party software version
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *RegisterPartyRequest) Reset() {
*x = RegisterPartyRequest{}
mi := &file_api_proto_message_router_proto_msgTypes[6]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *RegisterPartyRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*RegisterPartyRequest) ProtoMessage() {}
func (x *RegisterPartyRequest) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_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 RegisterPartyRequest.ProtoReflect.Descriptor instead.
func (*RegisterPartyRequest) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{6}
}
func (x *RegisterPartyRequest) GetPartyId() string {
if x != nil {
return x.PartyId
}
return ""
}
func (x *RegisterPartyRequest) GetPartyRole() string {
if x != nil {
return x.PartyRole
}
return ""
}
func (x *RegisterPartyRequest) GetVersion() string {
if x != nil {
return x.Version
}
return ""
}
// RegisterPartyResponse confirms party registration
type RegisterPartyResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Success bool `protobuf:"varint,1,opt,name=success,proto3" json:"success,omitempty"`
Message string `protobuf:"bytes,2,opt,name=message,proto3" json:"message,omitempty"`
RegisteredAt int64 `protobuf:"varint,3,opt,name=registered_at,json=registeredAt,proto3" json:"registered_at,omitempty"` // Unix timestamp milliseconds
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *RegisterPartyResponse) Reset() {
*x = RegisterPartyResponse{}
mi := &file_api_proto_message_router_proto_msgTypes[7]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *RegisterPartyResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*RegisterPartyResponse) ProtoMessage() {}
func (x *RegisterPartyResponse) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_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 RegisterPartyResponse.ProtoReflect.Descriptor instead.
func (*RegisterPartyResponse) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{7}
}
func (x *RegisterPartyResponse) GetSuccess() bool {
if x != nil {
return x.Success
}
return false
}
func (x *RegisterPartyResponse) GetMessage() string {
if x != nil {
return x.Message
}
return ""
}
func (x *RegisterPartyResponse) GetRegisteredAt() int64 {
if x != nil {
return x.RegisteredAt
}
return 0
}
// SubscribeSessionEventsRequest subscribes to session events
type SubscribeSessionEventsRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
PartyId string `protobuf:"bytes,1,opt,name=party_id,json=partyId,proto3" json:"party_id,omitempty"` // Party ID subscribing to events
EventTypes []string `protobuf:"bytes,2,rep,name=event_types,json=eventTypes,proto3" json:"event_types,omitempty"` // Event types to subscribe (empty = all)
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *SubscribeSessionEventsRequest) Reset() {
*x = SubscribeSessionEventsRequest{}
mi := &file_api_proto_message_router_proto_msgTypes[8]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *SubscribeSessionEventsRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*SubscribeSessionEventsRequest) ProtoMessage() {}
func (x *SubscribeSessionEventsRequest) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_proto_msgTypes[8]
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 SubscribeSessionEventsRequest.ProtoReflect.Descriptor instead.
func (*SubscribeSessionEventsRequest) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{8}
}
func (x *SubscribeSessionEventsRequest) GetPartyId() string {
if x != nil {
return x.PartyId
}
return ""
}
func (x *SubscribeSessionEventsRequest) GetEventTypes() []string {
if x != nil {
return x.EventTypes
}
return nil
}
// SessionEvent represents a session lifecycle event
type SessionEvent struct {
state protoimpl.MessageState `protogen:"open.v1"`
EventId string `protobuf:"bytes,1,opt,name=event_id,json=eventId,proto3" json:"event_id,omitempty"`
EventType string `protobuf:"bytes,2,opt,name=event_type,json=eventType,proto3" json:"event_type,omitempty"` // session_created, session_started, etc.
SessionId string `protobuf:"bytes,3,opt,name=session_id,json=sessionId,proto3" json:"session_id,omitempty"`
ThresholdN int32 `protobuf:"varint,4,opt,name=threshold_n,json=thresholdN,proto3" json:"threshold_n,omitempty"`
ThresholdT int32 `protobuf:"varint,5,opt,name=threshold_t,json=thresholdT,proto3" json:"threshold_t,omitempty"`
SelectedParties []string `protobuf:"bytes,6,rep,name=selected_parties,json=selectedParties,proto3" json:"selected_parties,omitempty"` // PartyIDs selected for this session
JoinTokens map[string]string `protobuf:"bytes,7,rep,name=join_tokens,json=joinTokens,proto3" json:"join_tokens,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` // PartyID -> JoinToken mapping
MessageHash []byte `protobuf:"bytes,8,opt,name=message_hash,json=messageHash,proto3" json:"message_hash,omitempty"` // For sign sessions
CreatedAt int64 `protobuf:"varint,9,opt,name=created_at,json=createdAt,proto3" json:"created_at,omitempty"` // Unix timestamp milliseconds
ExpiresAt int64 `protobuf:"varint,10,opt,name=expires_at,json=expiresAt,proto3" json:"expires_at,omitempty"` // Unix timestamp milliseconds
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *SessionEvent) Reset() {
*x = SessionEvent{}
mi := &file_api_proto_message_router_proto_msgTypes[9]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *SessionEvent) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*SessionEvent) ProtoMessage() {}
func (x *SessionEvent) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_proto_msgTypes[9]
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 SessionEvent.ProtoReflect.Descriptor instead.
func (*SessionEvent) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{9}
}
func (x *SessionEvent) GetEventId() string {
if x != nil {
return x.EventId
}
return ""
}
func (x *SessionEvent) GetEventType() string {
if x != nil {
return x.EventType
}
return ""
}
func (x *SessionEvent) GetSessionId() string {
if x != nil {
return x.SessionId
}
return ""
}
func (x *SessionEvent) GetThresholdN() int32 {
if x != nil {
return x.ThresholdN
}
return 0
}
func (x *SessionEvent) GetThresholdT() int32 {
if x != nil {
return x.ThresholdT
}
return 0
}
func (x *SessionEvent) GetSelectedParties() []string {
if x != nil {
return x.SelectedParties
}
return nil
}
func (x *SessionEvent) GetJoinTokens() map[string]string {
if x != nil {
return x.JoinTokens
}
return nil
}
func (x *SessionEvent) GetMessageHash() []byte {
if x != nil {
return x.MessageHash
}
return nil
}
func (x *SessionEvent) GetCreatedAt() int64 {
if x != nil {
return x.CreatedAt
}
return 0
}
func (x *SessionEvent) GetExpiresAt() int64 {
if x != nil {
return x.ExpiresAt
}
return 0
}
// PublishSessionEventRequest publishes a session event
type PublishSessionEventRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
Event *SessionEvent `protobuf:"bytes,1,opt,name=event,proto3" json:"event,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *PublishSessionEventRequest) Reset() {
*x = PublishSessionEventRequest{}
mi := &file_api_proto_message_router_proto_msgTypes[10]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *PublishSessionEventRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*PublishSessionEventRequest) ProtoMessage() {}
func (x *PublishSessionEventRequest) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_proto_msgTypes[10]
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 PublishSessionEventRequest.ProtoReflect.Descriptor instead.
func (*PublishSessionEventRequest) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{10}
}
func (x *PublishSessionEventRequest) GetEvent() *SessionEvent {
if x != nil {
return x.Event
}
return nil
}
// PublishSessionEventResponse confirms event publication
type PublishSessionEventResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Success bool `protobuf:"varint,1,opt,name=success,proto3" json:"success,omitempty"`
SubscriberCount int32 `protobuf:"varint,2,opt,name=subscriber_count,json=subscriberCount,proto3" json:"subscriber_count,omitempty"` // Number of parties that received the event
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *PublishSessionEventResponse) Reset() {
*x = PublishSessionEventResponse{}
mi := &file_api_proto_message_router_proto_msgTypes[11]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *PublishSessionEventResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*PublishSessionEventResponse) ProtoMessage() {}
func (x *PublishSessionEventResponse) ProtoReflect() protoreflect.Message {
mi := &file_api_proto_message_router_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 PublishSessionEventResponse.ProtoReflect.Descriptor instead.
func (*PublishSessionEventResponse) Descriptor() ([]byte, []int) {
return file_api_proto_message_router_proto_rawDescGZIP(), []int{11}
}
func (x *PublishSessionEventResponse) GetSuccess() bool {
if x != nil {
return x.Success
}
return false
}
func (x *PublishSessionEventResponse) GetSubscriberCount() int32 {
if x != nil {
return x.SubscriberCount
}
return 0
}
var File_api_proto_message_router_proto protoreflect.FileDescriptor
const file_api_proto_message_router_proto_rawDesc = "" +
"\n" +
"\x1eapi/proto/message_router.proto\x12\rmpc.router.v1\"\xd2\x01\n" +
"\x13RouteMessageRequest\x12\x1d\n" +
"\n" +
"session_id\x18\x01 \x01(\tR\tsessionId\x12\x1d\n" +
"\n" +
"from_party\x18\x02 \x01(\tR\tfromParty\x12\x1d\n" +
"\n" +
"to_parties\x18\x03 \x03(\tR\ttoParties\x12!\n" +
"\fround_number\x18\x04 \x01(\x05R\vroundNumber\x12!\n" +
"\fmessage_type\x18\x05 \x01(\tR\vmessageType\x12\x18\n" +
"\apayload\x18\x06 \x01(\fR\apayload\"O\n" +
"\x14RouteMessageResponse\x12\x18\n" +
"\asuccess\x18\x01 \x01(\bR\asuccess\x12\x1d\n" +
"\n" +
"message_id\x18\x02 \x01(\tR\tmessageId\"T\n" +
"\x18SubscribeMessagesRequest\x12\x1d\n" +
"\n" +
"session_id\x18\x01 \x01(\tR\tsessionId\x12\x19\n" +
"\bparty_id\x18\x02 \x01(\tR\apartyId\"\x8b\x02\n" +
"\n" +
"MPCMessage\x12\x1d\n" +
"\n" +
"message_id\x18\x01 \x01(\tR\tmessageId\x12\x1d\n" +
"\n" +
"session_id\x18\x02 \x01(\tR\tsessionId\x12\x1d\n" +
"\n" +
"from_party\x18\x03 \x01(\tR\tfromParty\x12!\n" +
"\fis_broadcast\x18\x04 \x01(\bR\visBroadcast\x12!\n" +
"\fround_number\x18\x05 \x01(\x05R\vroundNumber\x12!\n" +
"\fmessage_type\x18\x06 \x01(\tR\vmessageType\x12\x18\n" +
"\apayload\x18\a \x01(\fR\apayload\x12\x1d\n" +
"\n" +
"created_at\x18\b \x01(\x03R\tcreatedAt\"~\n" +
"\x19GetPendingMessagesRequest\x12\x1d\n" +
"\n" +
"session_id\x18\x01 \x01(\tR\tsessionId\x12\x19\n" +
"\bparty_id\x18\x02 \x01(\tR\apartyId\x12'\n" +
"\x0fafter_timestamp\x18\x03 \x01(\x03R\x0eafterTimestamp\"S\n" +
"\x1aGetPendingMessagesResponse\x125\n" +
"\bmessages\x18\x01 \x03(\v2\x19.mpc.router.v1.MPCMessageR\bmessages\"j\n" +
"\x14RegisterPartyRequest\x12\x19\n" +
"\bparty_id\x18\x01 \x01(\tR\apartyId\x12\x1d\n" +
"\n" +
"party_role\x18\x02 \x01(\tR\tpartyRole\x12\x18\n" +
"\aversion\x18\x03 \x01(\tR\aversion\"p\n" +
"\x15RegisterPartyResponse\x12\x18\n" +
"\asuccess\x18\x01 \x01(\bR\asuccess\x12\x18\n" +
"\amessage\x18\x02 \x01(\tR\amessage\x12#\n" +
"\rregistered_at\x18\x03 \x01(\x03R\fregisteredAt\"[\n" +
"\x1dSubscribeSessionEventsRequest\x12\x19\n" +
"\bparty_id\x18\x01 \x01(\tR\apartyId\x12\x1f\n" +
"\vevent_types\x18\x02 \x03(\tR\n" +
"eventTypes\"\xc2\x03\n" +
"\fSessionEvent\x12\x19\n" +
"\bevent_id\x18\x01 \x01(\tR\aeventId\x12\x1d\n" +
"\n" +
"event_type\x18\x02 \x01(\tR\teventType\x12\x1d\n" +
"\n" +
"session_id\x18\x03 \x01(\tR\tsessionId\x12\x1f\n" +
"\vthreshold_n\x18\x04 \x01(\x05R\n" +
"thresholdN\x12\x1f\n" +
"\vthreshold_t\x18\x05 \x01(\x05R\n" +
"thresholdT\x12)\n" +
"\x10selected_parties\x18\x06 \x03(\tR\x0fselectedParties\x12L\n" +
"\vjoin_tokens\x18\a \x03(\v2+.mpc.router.v1.SessionEvent.JoinTokensEntryR\n" +
"joinTokens\x12!\n" +
"\fmessage_hash\x18\b \x01(\fR\vmessageHash\x12\x1d\n" +
"\n" +
"created_at\x18\t \x01(\x03R\tcreatedAt\x12\x1d\n" +
"\n" +
"expires_at\x18\n" +
" \x01(\x03R\texpiresAt\x1a=\n" +
"\x0fJoinTokensEntry\x12\x10\n" +
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"O\n" +
"\x1aPublishSessionEventRequest\x121\n" +
"\x05event\x18\x01 \x01(\v2\x1b.mpc.router.v1.SessionEventR\x05event\"b\n" +
"\x1bPublishSessionEventResponse\x12\x18\n" +
"\asuccess\x18\x01 \x01(\bR\asuccess\x12)\n" +
"\x10subscriber_count\x18\x02 \x01(\x05R\x0fsubscriberCount2\xdf\x04\n" +
"\rMessageRouter\x12W\n" +
"\fRouteMessage\x12\".mpc.router.v1.RouteMessageRequest\x1a#.mpc.router.v1.RouteMessageResponse\x12Y\n" +
"\x11SubscribeMessages\x12'.mpc.router.v1.SubscribeMessagesRequest\x1a\x19.mpc.router.v1.MPCMessage0\x01\x12i\n" +
"\x12GetPendingMessages\x12(.mpc.router.v1.GetPendingMessagesRequest\x1a).mpc.router.v1.GetPendingMessagesResponse\x12Z\n" +
"\rRegisterParty\x12#.mpc.router.v1.RegisterPartyRequest\x1a$.mpc.router.v1.RegisterPartyResponse\x12e\n" +
"\x16SubscribeSessionEvents\x12,.mpc.router.v1.SubscribeSessionEventsRequest\x1a\x1b.mpc.router.v1.SessionEvent0\x01\x12l\n" +
"\x13PublishSessionEvent\x12).mpc.router.v1.PublishSessionEventRequest\x1a*.mpc.router.v1.PublishSessionEventResponseB;Z9github.com/rwadurian/mpc-system/api/grpc/router/v1;routerb\x06proto3"
var (
file_api_proto_message_router_proto_rawDescOnce sync.Once
file_api_proto_message_router_proto_rawDescData []byte
)
func file_api_proto_message_router_proto_rawDescGZIP() []byte {
file_api_proto_message_router_proto_rawDescOnce.Do(func() {
file_api_proto_message_router_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_api_proto_message_router_proto_rawDesc), len(file_api_proto_message_router_proto_rawDesc)))
})
return file_api_proto_message_router_proto_rawDescData
}
var file_api_proto_message_router_proto_msgTypes = make([]protoimpl.MessageInfo, 13)
var file_api_proto_message_router_proto_goTypes = []any{
(*RouteMessageRequest)(nil), // 0: mpc.router.v1.RouteMessageRequest
(*RouteMessageResponse)(nil), // 1: mpc.router.v1.RouteMessageResponse
(*SubscribeMessagesRequest)(nil), // 2: mpc.router.v1.SubscribeMessagesRequest
(*MPCMessage)(nil), // 3: mpc.router.v1.MPCMessage
(*GetPendingMessagesRequest)(nil), // 4: mpc.router.v1.GetPendingMessagesRequest
(*GetPendingMessagesResponse)(nil), // 5: mpc.router.v1.GetPendingMessagesResponse
(*RegisterPartyRequest)(nil), // 6: mpc.router.v1.RegisterPartyRequest
(*RegisterPartyResponse)(nil), // 7: mpc.router.v1.RegisterPartyResponse
(*SubscribeSessionEventsRequest)(nil), // 8: mpc.router.v1.SubscribeSessionEventsRequest
(*SessionEvent)(nil), // 9: mpc.router.v1.SessionEvent
(*PublishSessionEventRequest)(nil), // 10: mpc.router.v1.PublishSessionEventRequest
(*PublishSessionEventResponse)(nil), // 11: mpc.router.v1.PublishSessionEventResponse
nil, // 12: mpc.router.v1.SessionEvent.JoinTokensEntry
}
var file_api_proto_message_router_proto_depIdxs = []int32{
3, // 0: mpc.router.v1.GetPendingMessagesResponse.messages:type_name -> mpc.router.v1.MPCMessage
12, // 1: mpc.router.v1.SessionEvent.join_tokens:type_name -> mpc.router.v1.SessionEvent.JoinTokensEntry
9, // 2: mpc.router.v1.PublishSessionEventRequest.event:type_name -> mpc.router.v1.SessionEvent
0, // 3: mpc.router.v1.MessageRouter.RouteMessage:input_type -> mpc.router.v1.RouteMessageRequest
2, // 4: mpc.router.v1.MessageRouter.SubscribeMessages:input_type -> mpc.router.v1.SubscribeMessagesRequest
4, // 5: mpc.router.v1.MessageRouter.GetPendingMessages:input_type -> mpc.router.v1.GetPendingMessagesRequest
6, // 6: mpc.router.v1.MessageRouter.RegisterParty:input_type -> mpc.router.v1.RegisterPartyRequest
8, // 7: mpc.router.v1.MessageRouter.SubscribeSessionEvents:input_type -> mpc.router.v1.SubscribeSessionEventsRequest
10, // 8: mpc.router.v1.MessageRouter.PublishSessionEvent:input_type -> mpc.router.v1.PublishSessionEventRequest
1, // 9: mpc.router.v1.MessageRouter.RouteMessage:output_type -> mpc.router.v1.RouteMessageResponse
3, // 10: mpc.router.v1.MessageRouter.SubscribeMessages:output_type -> mpc.router.v1.MPCMessage
5, // 11: mpc.router.v1.MessageRouter.GetPendingMessages:output_type -> mpc.router.v1.GetPendingMessagesResponse
7, // 12: mpc.router.v1.MessageRouter.RegisterParty:output_type -> mpc.router.v1.RegisterPartyResponse
9, // 13: mpc.router.v1.MessageRouter.SubscribeSessionEvents:output_type -> mpc.router.v1.SessionEvent
11, // 14: mpc.router.v1.MessageRouter.PublishSessionEvent:output_type -> mpc.router.v1.PublishSessionEventResponse
9, // [9:15] is the sub-list for method output_type
3, // [3:9] is the sub-list for method input_type
3, // [3:3] is the sub-list for extension type_name
3, // [3:3] is the sub-list for extension extendee
0, // [0:3] is the sub-list for field type_name
}
func init() { file_api_proto_message_router_proto_init() }
func file_api_proto_message_router_proto_init() {
if File_api_proto_message_router_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_api_proto_message_router_proto_rawDesc), len(file_api_proto_message_router_proto_rawDesc)),
NumEnums: 0,
NumMessages: 13,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_api_proto_message_router_proto_goTypes,
DependencyIndexes: file_api_proto_message_router_proto_depIdxs,
MessageInfos: file_api_proto_message_router_proto_msgTypes,
}.Build()
File_api_proto_message_router_proto = out.File
file_api_proto_message_router_proto_goTypes = nil
file_api_proto_message_router_proto_depIdxs = nil
}

24
backend/mpc-system/api/proto/message_router.proto
View File

@ -23,6 +23,9 @@ service MessageRouter {
// PublishSessionEvent publishes a session event (called by Session Coordinator)
rpc PublishSessionEvent(PublishSessionEventRequest) returns (PublishSessionEventResponse);
// GetRegisteredParties returns all registered parties (for Session Coordinator party discovery)
rpc GetRegisteredParties(GetRegisteredPartiesRequest) returns (GetRegisteredPartiesResponse);
}
// RouteMessageRequest routes an MPC message
@ -115,3 +118,24 @@ message PublishSessionEventResponse {
bool success = 1;
int32 subscriber_count = 2; // Number of parties that received the event
}
// GetRegisteredPartiesRequest requests registered parties list
message GetRegisteredPartiesRequest {
string role_filter = 1; // Optional: filter by role (persistent, delegate, temporary)
bool only_online = 2; // Optional: only return online parties
}
// RegisteredParty represents a registered party
message RegisteredParty {
string party_id = 1; // Unique party identifier
string role = 2; // persistent, delegate, or temporary
bool online = 3; // Whether party is currently connected
int64 registered_at = 4; // Unix timestamp milliseconds
int64 last_seen_at = 5; // Unix timestamp milliseconds
}
// GetRegisteredPartiesResponse returns registered parties
message GetRegisteredPartiesResponse {
repeated RegisteredParty parties = 1;
int32 total_count = 2;
}

334
backend/mpc-system/api/proto/message_router_grpc.pb.go
View File

@ -1,334 +0,0 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.6.0
// - protoc v6.33.1
// source: api/proto/message_router.proto
package router
import (
context "context"
grpc "google.golang.org/grpc"
codes "google.golang.org/grpc/codes"
status "google.golang.org/grpc/status"
)
// This is a compile-time assertion to ensure that this generated file
// is compatible with the grpc package it is being compiled against.
// Requires gRPC-Go v1.64.0 or later.
const _ = grpc.SupportPackageIsVersion9
const (
MessageRouter_RouteMessage_FullMethodName = "/mpc.router.v1.MessageRouter/RouteMessage"
MessageRouter_SubscribeMessages_FullMethodName = "/mpc.router.v1.MessageRouter/SubscribeMessages"
MessageRouter_GetPendingMessages_FullMethodName = "/mpc.router.v1.MessageRouter/GetPendingMessages"
MessageRouter_RegisterParty_FullMethodName = "/mpc.router.v1.MessageRouter/RegisterParty"
MessageRouter_SubscribeSessionEvents_FullMethodName = "/mpc.router.v1.MessageRouter/SubscribeSessionEvents"
MessageRouter_PublishSessionEvent_FullMethodName = "/mpc.router.v1.MessageRouter/PublishSessionEvent"
)
// MessageRouterClient is the client API for MessageRouter service.
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
//
// MessageRouter service handles MPC message routing
type MessageRouterClient interface {
// RouteMessage routes a message from one party to others
RouteMessage(ctx context.Context, in *RouteMessageRequest, opts ...grpc.CallOption) (*RouteMessageResponse, error)
// SubscribeMessages subscribes to messages for a party (streaming)
SubscribeMessages(ctx context.Context, in *SubscribeMessagesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[MPCMessage], error)
// GetPendingMessages retrieves pending messages (polling alternative)
GetPendingMessages(ctx context.Context, in *GetPendingMessagesRequest, opts ...grpc.CallOption) (*GetPendingMessagesResponse, error)
// RegisterParty registers a party with the message router (party actively connects)
RegisterParty(ctx context.Context, in *RegisterPartyRequest, opts ...grpc.CallOption) (*RegisterPartyResponse, error)
// SubscribeSessionEvents subscribes to session lifecycle events (session start, etc.)
SubscribeSessionEvents(ctx context.Context, in *SubscribeSessionEventsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[SessionEvent], error)
// PublishSessionEvent publishes a session event (called by Session Coordinator)
PublishSessionEvent(ctx context.Context, in *PublishSessionEventRequest, opts ...grpc.CallOption) (*PublishSessionEventResponse, error)
}
type messageRouterClient struct {
cc grpc.ClientConnInterface
}
func NewMessageRouterClient(cc grpc.ClientConnInterface) MessageRouterClient {
return &messageRouterClient{cc}
}
func (c *messageRouterClient) RouteMessage(ctx context.Context, in *RouteMessageRequest, opts ...grpc.CallOption) (*RouteMessageResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(RouteMessageResponse)
err := c.cc.Invoke(ctx, MessageRouter_RouteMessage_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *messageRouterClient) SubscribeMessages(ctx context.Context, in *SubscribeMessagesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[MPCMessage], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
stream, err := c.cc.NewStream(ctx, &MessageRouter_ServiceDesc.Streams[0], MessageRouter_SubscribeMessages_FullMethodName, cOpts...)
if err != nil {
return nil, err
}
x := &grpc.GenericClientStream[SubscribeMessagesRequest, MPCMessage]{ClientStream: stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
return x, nil
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type MessageRouter_SubscribeMessagesClient = grpc.ServerStreamingClient[MPCMessage]
func (c *messageRouterClient) GetPendingMessages(ctx context.Context, in *GetPendingMessagesRequest, opts ...grpc.CallOption) (*GetPendingMessagesResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(GetPendingMessagesResponse)
err := c.cc.Invoke(ctx, MessageRouter_GetPendingMessages_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *messageRouterClient) RegisterParty(ctx context.Context, in *RegisterPartyRequest, opts ...grpc.CallOption) (*RegisterPartyResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(RegisterPartyResponse)
err := c.cc.Invoke(ctx, MessageRouter_RegisterParty_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *messageRouterClient) SubscribeSessionEvents(ctx context.Context, in *SubscribeSessionEventsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[SessionEvent], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
stream, err := c.cc.NewStream(ctx, &MessageRouter_ServiceDesc.Streams[1], MessageRouter_SubscribeSessionEvents_FullMethodName, cOpts...)
if err != nil {
return nil, err
}
x := &grpc.GenericClientStream[SubscribeSessionEventsRequest, SessionEvent]{ClientStream: stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
return x, nil
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type MessageRouter_SubscribeSessionEventsClient = grpc.ServerStreamingClient[SessionEvent]
func (c *messageRouterClient) PublishSessionEvent(ctx context.Context, in *PublishSessionEventRequest, opts ...grpc.CallOption) (*PublishSessionEventResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(PublishSessionEventResponse)
err := c.cc.Invoke(ctx, MessageRouter_PublishSessionEvent_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
// MessageRouterServer is the server API for MessageRouter service.
// All implementations must embed UnimplementedMessageRouterServer
// for forward compatibility.
//
// MessageRouter service handles MPC message routing
type MessageRouterServer interface {
// RouteMessage routes a message from one party to others
RouteMessage(context.Context, *RouteMessageRequest) (*RouteMessageResponse, error)
// SubscribeMessages subscribes to messages for a party (streaming)
SubscribeMessages(*SubscribeMessagesRequest, grpc.ServerStreamingServer[MPCMessage]) error
// GetPendingMessages retrieves pending messages (polling alternative)
GetPendingMessages(context.Context, *GetPendingMessagesRequest) (*GetPendingMessagesResponse, error)
// RegisterParty registers a party with the message router (party actively connects)
RegisterParty(context.Context, *RegisterPartyRequest) (*RegisterPartyResponse, error)
// SubscribeSessionEvents subscribes to session lifecycle events (session start, etc.)
SubscribeSessionEvents(*SubscribeSessionEventsRequest, grpc.ServerStreamingServer[SessionEvent]) error
// PublishSessionEvent publishes a session event (called by Session Coordinator)
PublishSessionEvent(context.Context, *PublishSessionEventRequest) (*PublishSessionEventResponse, error)
mustEmbedUnimplementedMessageRouterServer()
}
// UnimplementedMessageRouterServer must be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedMessageRouterServer struct{}
func (UnimplementedMessageRouterServer) RouteMessage(context.Context, *RouteMessageRequest) (*RouteMessageResponse, error) {
return nil, status.Error(codes.Unimplemented, "method RouteMessage not implemented")
}
func (UnimplementedMessageRouterServer) SubscribeMessages(*SubscribeMessagesRequest, grpc.ServerStreamingServer[MPCMessage]) error {
return status.Error(codes.Unimplemented, "method SubscribeMessages not implemented")
}
func (UnimplementedMessageRouterServer) GetPendingMessages(context.Context, *GetPendingMessagesRequest) (*GetPendingMessagesResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetPendingMessages not implemented")
}
func (UnimplementedMessageRouterServer) RegisterParty(context.Context, *RegisterPartyRequest) (*RegisterPartyResponse, error) {
return nil, status.Error(codes.Unimplemented, "method RegisterParty not implemented")
}
func (UnimplementedMessageRouterServer) SubscribeSessionEvents(*SubscribeSessionEventsRequest, grpc.ServerStreamingServer[SessionEvent]) error {
return status.Error(codes.Unimplemented, "method SubscribeSessionEvents not implemented")
}
func (UnimplementedMessageRouterServer) PublishSessionEvent(context.Context, *PublishSessionEventRequest) (*PublishSessionEventResponse, error) {
return nil, status.Error(codes.Unimplemented, "method PublishSessionEvent not implemented")
}
func (UnimplementedMessageRouterServer) mustEmbedUnimplementedMessageRouterServer() {}
func (UnimplementedMessageRouterServer) testEmbeddedByValue() {}
// UnsafeMessageRouterServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to MessageRouterServer will
// result in compilation errors.
type UnsafeMessageRouterServer interface {
mustEmbedUnimplementedMessageRouterServer()
}
func RegisterMessageRouterServer(s grpc.ServiceRegistrar, srv MessageRouterServer) {
// If the following call panics, it indicates UnimplementedMessageRouterServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&MessageRouter_ServiceDesc, srv)
}
func _MessageRouter_RouteMessage_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(RouteMessageRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(MessageRouterServer).RouteMessage(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: MessageRouter_RouteMessage_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(MessageRouterServer).RouteMessage(ctx, req.(*RouteMessageRequest))
}
return interceptor(ctx, in, info, handler)
}
func _MessageRouter_SubscribeMessages_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(SubscribeMessagesRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(MessageRouterServer).SubscribeMessages(m, &grpc.GenericServerStream[SubscribeMessagesRequest, MPCMessage]{ServerStream: stream})
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type MessageRouter_SubscribeMessagesServer = grpc.ServerStreamingServer[MPCMessage]
func _MessageRouter_GetPendingMessages_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(GetPendingMessagesRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(MessageRouterServer).GetPendingMessages(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: MessageRouter_GetPendingMessages_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(MessageRouterServer).GetPendingMessages(ctx, req.(*GetPendingMessagesRequest))
}
return interceptor(ctx, in, info, handler)
}
func _MessageRouter_RegisterParty_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(RegisterPartyRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(MessageRouterServer).RegisterParty(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: MessageRouter_RegisterParty_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(MessageRouterServer).RegisterParty(ctx, req.(*RegisterPartyRequest))
}
return interceptor(ctx, in, info, handler)
}
func _MessageRouter_SubscribeSessionEvents_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(SubscribeSessionEventsRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(MessageRouterServer).SubscribeSessionEvents(m, &grpc.GenericServerStream[SubscribeSessionEventsRequest, SessionEvent]{ServerStream: stream})
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type MessageRouter_SubscribeSessionEventsServer = grpc.ServerStreamingServer[SessionEvent]
func _MessageRouter_PublishSessionEvent_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(PublishSessionEventRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(MessageRouterServer).PublishSessionEvent(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: MessageRouter_PublishSessionEvent_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(MessageRouterServer).PublishSessionEvent(ctx, req.(*PublishSessionEventRequest))
}
return interceptor(ctx, in, info, handler)
}
// MessageRouter_ServiceDesc is the grpc.ServiceDesc for MessageRouter service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
var MessageRouter_ServiceDesc = grpc.ServiceDesc{
ServiceName: "mpc.router.v1.MessageRouter",
HandlerType: (*MessageRouterServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "RouteMessage",
Handler: _MessageRouter_RouteMessage_Handler,
},
{
MethodName: "GetPendingMessages",
Handler: _MessageRouter_GetPendingMessages_Handler,
},
{
MethodName: "RegisterParty",
Handler: _MessageRouter_RegisterParty_Handler,
},
{
MethodName: "PublishSessionEvent",
Handler: _MessageRouter_PublishSessionEvent_Handler,
},
},
Streams: []grpc.StreamDesc{
{
StreamName: "SubscribeMessages",
Handler: _MessageRouter_SubscribeMessages_Handler,
ServerStreams: true,
},
{
StreamName: "SubscribeSessionEvents",
Handler: _MessageRouter_SubscribeSessionEvents_Handler,
ServerStreams: true,
},
},
Metadata: "api/proto/message_router.proto",
}

1321
backend/mpc-system/api/proto/session_coordinator.pb.go
View File

File diff suppressed because it is too large Load Diff

355
backend/mpc-system/api/proto/session_coordinator_grpc.pb.go
View File

@ -1,355 +0,0 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.6.0
// - protoc v6.33.1
// source: api/proto/session_coordinator.proto
package coordinator
import (
context "context"
grpc "google.golang.org/grpc"
codes "google.golang.org/grpc/codes"
status "google.golang.org/grpc/status"
)
// This is a compile-time assertion to ensure that this generated file
// is compatible with the grpc package it is being compiled against.
// Requires gRPC-Go v1.64.0 or later.
const _ = grpc.SupportPackageIsVersion9
const (
SessionCoordinator_CreateSession_FullMethodName = "/mpc.coordinator.v1.SessionCoordinator/CreateSession"
SessionCoordinator_JoinSession_FullMethodName = "/mpc.coordinator.v1.SessionCoordinator/JoinSession"
SessionCoordinator_GetSessionStatus_FullMethodName = "/mpc.coordinator.v1.SessionCoordinator/GetSessionStatus"
SessionCoordinator_MarkPartyReady_FullMethodName = "/mpc.coordinator.v1.SessionCoordinator/MarkPartyReady"
SessionCoordinator_StartSession_FullMethodName = "/mpc.coordinator.v1.SessionCoordinator/StartSession"
SessionCoordinator_ReportCompletion_FullMethodName = "/mpc.coordinator.v1.SessionCoordinator/ReportCompletion"
SessionCoordinator_CloseSession_FullMethodName = "/mpc.coordinator.v1.SessionCoordinator/CloseSession"
)
// SessionCoordinatorClient is the client API for SessionCoordinator service.
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
//
// SessionCoordinator service manages MPC sessions
type SessionCoordinatorClient interface {
// Session management
CreateSession(ctx context.Context, in *CreateSessionRequest, opts ...grpc.CallOption) (*CreateSessionResponse, error)
JoinSession(ctx context.Context, in *JoinSessionRequest, opts ...grpc.CallOption) (*JoinSessionResponse, error)
GetSessionStatus(ctx context.Context, in *GetSessionStatusRequest, opts ...grpc.CallOption) (*GetSessionStatusResponse, error)
MarkPartyReady(ctx context.Context, in *MarkPartyReadyRequest, opts ...grpc.CallOption) (*MarkPartyReadyResponse, error)
StartSession(ctx context.Context, in *StartSessionRequest, opts ...grpc.CallOption) (*StartSessionResponse, error)
ReportCompletion(ctx context.Context, in *ReportCompletionRequest, opts ...grpc.CallOption) (*ReportCompletionResponse, error)
CloseSession(ctx context.Context, in *CloseSessionRequest, opts ...grpc.CallOption) (*CloseSessionResponse, error)
}
type sessionCoordinatorClient struct {
cc grpc.ClientConnInterface
}
func NewSessionCoordinatorClient(cc grpc.ClientConnInterface) SessionCoordinatorClient {
return &sessionCoordinatorClient{cc}
}
func (c *sessionCoordinatorClient) CreateSession(ctx context.Context, in *CreateSessionRequest, opts ...grpc.CallOption) (*CreateSessionResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(CreateSessionResponse)
err := c.cc.Invoke(ctx, SessionCoordinator_CreateSession_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *sessionCoordinatorClient) JoinSession(ctx context.Context, in *JoinSessionRequest, opts ...grpc.CallOption) (*JoinSessionResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(JoinSessionResponse)
err := c.cc.Invoke(ctx, SessionCoordinator_JoinSession_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *sessionCoordinatorClient) GetSessionStatus(ctx context.Context, in *GetSessionStatusRequest, opts ...grpc.CallOption) (*GetSessionStatusResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(GetSessionStatusResponse)
err := c.cc.Invoke(ctx, SessionCoordinator_GetSessionStatus_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *sessionCoordinatorClient) MarkPartyReady(ctx context.Context, in *MarkPartyReadyRequest, opts ...grpc.CallOption) (*MarkPartyReadyResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(MarkPartyReadyResponse)
err := c.cc.Invoke(ctx, SessionCoordinator_MarkPartyReady_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *sessionCoordinatorClient) StartSession(ctx context.Context, in *StartSessionRequest, opts ...grpc.CallOption) (*StartSessionResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(StartSessionResponse)
err := c.cc.Invoke(ctx, SessionCoordinator_StartSession_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *sessionCoordinatorClient) ReportCompletion(ctx context.Context, in *ReportCompletionRequest, opts ...grpc.CallOption) (*ReportCompletionResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(ReportCompletionResponse)
err := c.cc.Invoke(ctx, SessionCoordinator_ReportCompletion_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *sessionCoordinatorClient) CloseSession(ctx context.Context, in *CloseSessionRequest, opts ...grpc.CallOption) (*CloseSessionResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(CloseSessionResponse)
err := c.cc.Invoke(ctx, SessionCoordinator_CloseSession_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
// SessionCoordinatorServer is the server API for SessionCoordinator service.
// All implementations must embed UnimplementedSessionCoordinatorServer
// for forward compatibility.
//
// SessionCoordinator service manages MPC sessions
type SessionCoordinatorServer interface {
// Session management
CreateSession(context.Context, *CreateSessionRequest) (*CreateSessionResponse, error)
JoinSession(context.Context, *JoinSessionRequest) (*JoinSessionResponse, error)
GetSessionStatus(context.Context, *GetSessionStatusRequest) (*GetSessionStatusResponse, error)
MarkPartyReady(context.Context, *MarkPartyReadyRequest) (*MarkPartyReadyResponse, error)
StartSession(context.Context, *StartSessionRequest) (*StartSessionResponse, error)
ReportCompletion(context.Context, *ReportCompletionRequest) (*ReportCompletionResponse, error)
CloseSession(context.Context, *CloseSessionRequest) (*CloseSessionResponse, error)
mustEmbedUnimplementedSessionCoordinatorServer()
}
// UnimplementedSessionCoordinatorServer must be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedSessionCoordinatorServer struct{}
func (UnimplementedSessionCoordinatorServer) CreateSession(context.Context, *CreateSessionRequest) (*CreateSessionResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CreateSession not implemented")
}
func (UnimplementedSessionCoordinatorServer) JoinSession(context.Context, *JoinSessionRequest) (*JoinSessionResponse, error) {
return nil, status.Error(codes.Unimplemented, "method JoinSession not implemented")
}
func (UnimplementedSessionCoordinatorServer) GetSessionStatus(context.Context, *GetSessionStatusRequest) (*GetSessionStatusResponse, error) {
return nil, status.Error(codes.Unimplemented, "method GetSessionStatus not implemented")
}
func (UnimplementedSessionCoordinatorServer) MarkPartyReady(context.Context, *MarkPartyReadyRequest) (*MarkPartyReadyResponse, error) {
return nil, status.Error(codes.Unimplemented, "method MarkPartyReady not implemented")
}
func (UnimplementedSessionCoordinatorServer) StartSession(context.Context, *StartSessionRequest) (*StartSessionResponse, error) {
return nil, status.Error(codes.Unimplemented, "method StartSession not implemented")
}
func (UnimplementedSessionCoordinatorServer) ReportCompletion(context.Context, *ReportCompletionRequest) (*ReportCompletionResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ReportCompletion not implemented")
}
func (UnimplementedSessionCoordinatorServer) CloseSession(context.Context, *CloseSessionRequest) (*CloseSessionResponse, error) {
return nil, status.Error(codes.Unimplemented, "method CloseSession not implemented")
}
func (UnimplementedSessionCoordinatorServer) mustEmbedUnimplementedSessionCoordinatorServer() {}
func (UnimplementedSessionCoordinatorServer) testEmbeddedByValue() {}
// UnsafeSessionCoordinatorServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to SessionCoordinatorServer will
// result in compilation errors.
type UnsafeSessionCoordinatorServer interface {
mustEmbedUnimplementedSessionCoordinatorServer()
}
func RegisterSessionCoordinatorServer(s grpc.ServiceRegistrar, srv SessionCoordinatorServer) {
// If the following call panics, it indicates UnimplementedSessionCoordinatorServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&SessionCoordinator_ServiceDesc, srv)
}
func _SessionCoordinator_CreateSession_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(CreateSessionRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(SessionCoordinatorServer).CreateSession(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: SessionCoordinator_CreateSession_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(SessionCoordinatorServer).CreateSession(ctx, req.(*CreateSessionRequest))
}
return interceptor(ctx, in, info, handler)
}
func _SessionCoordinator_JoinSession_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(JoinSessionRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(SessionCoordinatorServer).JoinSession(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: SessionCoordinator_JoinSession_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(SessionCoordinatorServer).JoinSession(ctx, req.(*JoinSessionRequest))
}
return interceptor(ctx, in, info, handler)
}
func _SessionCoordinator_GetSessionStatus_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(GetSessionStatusRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(SessionCoordinatorServer).GetSessionStatus(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: SessionCoordinator_GetSessionStatus_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(SessionCoordinatorServer).GetSessionStatus(ctx, req.(*GetSessionStatusRequest))
}
return interceptor(ctx, in, info, handler)
}
func _SessionCoordinator_MarkPartyReady_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(MarkPartyReadyRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(SessionCoordinatorServer).MarkPartyReady(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: SessionCoordinator_MarkPartyReady_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(SessionCoordinatorServer).MarkPartyReady(ctx, req.(*MarkPartyReadyRequest))
}
return interceptor(ctx, in, info, handler)
}
func _SessionCoordinator_StartSession_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(StartSessionRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(SessionCoordinatorServer).StartSession(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: SessionCoordinator_StartSession_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(SessionCoordinatorServer).StartSession(ctx, req.(*StartSessionRequest))
}
return interceptor(ctx, in, info, handler)
}
func _SessionCoordinator_ReportCompletion_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ReportCompletionRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(SessionCoordinatorServer).ReportCompletion(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: SessionCoordinator_ReportCompletion_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(SessionCoordinatorServer).ReportCompletion(ctx, req.(*ReportCompletionRequest))
}
return interceptor(ctx, in, info, handler)
}
func _SessionCoordinator_CloseSession_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(CloseSessionRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(SessionCoordinatorServer).CloseSession(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: SessionCoordinator_CloseSession_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(SessionCoordinatorServer).CloseSession(ctx, req.(*CloseSessionRequest))
}
return interceptor(ctx, in, info, handler)
}
// SessionCoordinator_ServiceDesc is the grpc.ServiceDesc for SessionCoordinator service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
var SessionCoordinator_ServiceDesc = grpc.ServiceDesc{
ServiceName: "mpc.coordinator.v1.SessionCoordinator",
HandlerType: (*SessionCoordinatorServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "CreateSession",
Handler: _SessionCoordinator_CreateSession_Handler,
},
{
MethodName: "JoinSession",
Handler: _SessionCoordinator_JoinSession_Handler,
},
{
MethodName: "GetSessionStatus",
Handler: _SessionCoordinator_GetSessionStatus_Handler,
},
{
MethodName: "MarkPartyReady",
Handler: _SessionCoordinator_MarkPartyReady_Handler,
},
{
MethodName: "StartSession",
Handler: _SessionCoordinator_StartSession_Handler,
},
{
MethodName: "ReportCompletion",
Handler: _SessionCoordinator_ReportCompletion_Handler,
},
{
MethodName: "CloseSession",
Handler: _SessionCoordinator_CloseSession_Handler,
},
},
Streams: []grpc.StreamDesc{},
Metadata: "api/proto/session_coordinator.proto",
}

3
backend/mpc-system/docker-compose.yml
View File

@ -113,6 +113,7 @@ services:
MPC_RABBITMQ_PASSWORD: ${RABBITMQ_PASSWORD:?RABBITMQ_PASSWORD must be set}
MPC_JWT_SECRET_KEY: ${JWT_SECRET_KEY}
MPC_JWT_ISSUER: mpc-system
MESSAGE_ROUTER_ADDR: message-router:50051
depends_on:
postgres:
condition: service_healthy
@ -120,6 +121,8 @@ services:
condition: service_healthy
rabbitmq:
condition: service_healthy
message-router:
condition: service_healthy
healthcheck:
test: ["CMD", "curl", "-sf", "http://localhost:8080/health"]
interval: 30s

39
backend/mpc-system/services/message-router/adapters/input/grpc/message_grpc_handler.go
View File

@ -259,6 +259,45 @@ func (s *MessageRouterServer) PublishSessionEvent(
}, nil
}
// GetRegisteredParties returns all registered parties
// This is called by Session Coordinator for party discovery
func (s *MessageRouterServer) GetRegisteredParties(
ctx context.Context,
req *pb.GetRegisteredPartiesRequest,
) (*pb.GetRegisteredPartiesResponse, error) {
var parties []*domain.RegisteredParty
// Filter by role if specified
if req.RoleFilter != "" {
parties = s.partyRegistry.GetByRole(req.RoleFilter)
} else {
parties = s.partyRegistry.GetAll()
}
// Convert to protobuf format
protoParties := make([]*pb.RegisteredParty, 0, len(parties))
for _, party := range parties {
// For now, consider all registered parties as online
// TODO: Track actual online status via heartbeats
protoParties = append(protoParties, &pb.RegisteredParty{
PartyId: party.PartyID,
Role: party.Role,
Online: true, // Assume online if registered
RegisteredAt: party.RegisteredAt.UnixMilli(),
LastSeenAt: party.LastSeen.UnixMilli(),
})
}
logger.Debug("GetRegisteredParties called",
zap.String("role_filter", req.RoleFilter),
zap.Int("party_count", len(protoParties)))
return &pb.GetRegisteredPartiesResponse{
Parties: protoParties,
TotalCount: int32(len(protoParties)),
}, nil
}
func sendMessage(stream pb.MessageRouter_SubscribeMessagesServer, msg *entities.MessageDTO) error {
protoMsg := &pb.MPCMessage{
MessageId: msg.ID,

14
backend/mpc-system/services/message-router/domain/party_registry.go
View File

@ -66,6 +66,20 @@ func (r *PartyRegistry) GetAll() []*RegisteredParty {
return parties
}
// GetByRole returns registered parties filtered by role
func (r *PartyRegistry) GetByRole(role string) []*RegisteredParty {
r.mu.RLock()
defer r.mu.RUnlock()
parties := make([]*RegisteredParty, 0)
for _, party := range r.parties {
if party.Role == role {
parties = append(parties, party)
}
}
return parties
}
// UpdateLastSeen updates the last seen timestamp
func (r *PartyRegistry) UpdateLastSeen(partyID string) {
r.mu.Lock()

20
backend/mpc-system/services/server-party-api/cmd/server/main.go
View File

@ -96,6 +96,26 @@ func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Get party ID from environment (or generate one)
partyID := os.Getenv("PARTY_ID")
if partyID == "" {
partyID = "server-party-api"
}
// Register this party as a delegate party with Message Router
// This allows Session Coordinator to discover this party for session creation
logger.Info("Registering party with Message Router",
zap.String("party_id", partyID),
zap.String("role", "delegate"))
if err := messageRouter.RegisterParty(ctx, partyID, "delegate", "1.0.0"); err != nil {
logger.Fatal("Failed to register party", zap.Error(err))
}
logger.Info("Party registered successfully",
zap.String("party_id", partyID),
zap.String("role", "delegate"))
// Start HTTP server
errChan := make(chan error, 1)
go func() {

25
backend/mpc-system/services/session-coordinator/adapters/output/grpc/message_router_client.go
View File

@ -77,6 +77,31 @@ func (c *MessageRouterClient) PublishSessionEvent(
return nil
}
// GetRegisteredParties retrieves registered parties from Message Router
func (c *MessageRouterClient) GetRegisteredParties(
ctx context.Context,
roleFilter string,
) ([]*router.RegisteredParty, error) {
req := &router.GetRegisteredPartiesRequest{
RoleFilter: roleFilter,
OnlyOnline: true,
}
resp, err := c.client.GetRegisteredParties(ctx, req)
if err != nil {
logger.Error("Failed to get registered parties",
zap.Error(err),
zap.String("role_filter", roleFilter))
return nil, err
}
logger.Debug("Retrieved registered parties from Message Router",
zap.String("role_filter", roleFilter),
zap.Int32("count", resp.TotalCount))
return resp.Parties, nil
}
// PublishSessionCreated publishes a session_created event
func (c *MessageRouterClient) PublishSessionCreated(
ctx context.Context,

19
backend/mpc-system/services/session-coordinator/cmd/server/main.go
View File

@ -31,7 +31,7 @@ import (
redisadapter "github.com/rwadurian/mpc-system/services/session-coordinator/adapters/output/redis"
"github.com/rwadurian/mpc-system/services/session-coordinator/application/use_cases"
"github.com/rwadurian/mpc-system/services/session-coordinator/domain/repositories"
"github.com/rwadurian/mpc-system/services/session-coordinator/infrastructure/k8s"
"github.com/rwadurian/mpc-system/services/session-coordinator/infrastructure/discovery"
"go.uber.org/zap"
)
@ -98,17 +98,7 @@ func main() {
cfg.JWT.RefreshExpiry,
)
// Initialize K8s party discovery (optional - will fallback gracefully if not in K8s)
partyPool, err := k8s.NewPartyDiscovery(logger.Log)
if err != nil {
logger.Warn("K8s party discovery not available, will use dynamic join mode",
zap.Error(err))
partyPool = nil // Set to nil so CreateSessionUseCase can handle gracefully
} else {
logger.Info("K8s party discovery initialized successfully")
}
// Initialize Message Router gRPC client for event publishing
// Initialize Message Router gRPC client for event publishing and party discovery
routerAddr := os.Getenv("MESSAGE_ROUTER_ADDR")
if routerAddr == "" {
routerAddr = "localhost:9092" // Default for local development
@ -119,6 +109,11 @@ func main() {
}
defer messageRouterClient.Close()
// Initialize party discovery using Message Router (decentralized approach)
// This replaces K8s-based discovery and works in any environment
partyPool := discovery.NewMessageRouterPartyDiscovery(messageRouterClient, logger.Log)
logger.Info("Party discovery initialized using Message Router")
// Initialize use cases
createSessionUC := use_cases.NewCreateSessionUseCase(sessionRepo, jwtService, eventPublisher, partyPool, messageRouterClient)
joinSessionUC := use_cases.NewJoinSessionUseCase(sessionRepo, jwtService, eventPublisher)

122
backend/mpc-system/services/session-coordinator/infrastructure/discovery/message_router_discovery.go Normal file
View File

@ -0,0 +1,122 @@
package discovery
import (
"context"
"fmt"
"time"
grpcclient "github.com/rwadurian/mpc-system/services/session-coordinator/adapters/output/grpc"
"github.com/rwadurian/mpc-system/services/session-coordinator/application/ports/output"
"go.uber.org/zap"
)
// MessageRouterPartyDiscovery implements PartyPoolPort using Message Router
// This replaces K8s-based discovery with a decentralized approach
type MessageRouterPartyDiscovery struct {
client *grpcclient.MessageRouterClient
logger *zap.Logger
}
// NewMessageRouterPartyDiscovery creates a new Message Router-based party discovery
func NewMessageRouterPartyDiscovery(client *grpcclient.MessageRouterClient, logger *zap.Logger) *MessageRouterPartyDiscovery {
return &MessageRouterPartyDiscovery{
client: client,
logger: logger,
}
}
// GetAvailableParties returns all available party endpoints
// Implements output.PartyPoolPort interface
func (d *MessageRouterPartyDiscovery) GetAvailableParties() []output.PartyEndpoint {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
parties, err := d.client.GetRegisteredParties(ctx, "")
if err != nil {
d.logger.Error("Failed to get available parties from Message Router", zap.Error(err))
return []output.PartyEndpoint{}
}
endpoints := make([]output.PartyEndpoint, 0, len(parties))
for _, party := range parties {
endpoints = append(endpoints, output.PartyEndpoint{
PartyID: party.PartyId,
Ready: party.Online,
Role: output.PartyRole(party.Role),
})
}
return endpoints
}
// GetAvailablePartiesByRole returns available parties filtered by role
// Implements output.PartyPoolPort interface
func (d *MessageRouterPartyDiscovery) GetAvailablePartiesByRole(role output.PartyRole) []output.PartyEndpoint {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
parties, err := d.client.GetRegisteredParties(ctx, string(role))
if err != nil {
d.logger.Error("Failed to get parties by role from Message Router",
zap.String("role", string(role)),
zap.Error(err))
return []output.PartyEndpoint{}
}
endpoints := make([]output.PartyEndpoint, 0, len(parties))
for _, party := range parties {
endpoints = append(endpoints, output.PartyEndpoint{
PartyID: party.PartyId,
Ready: party.Online,
Role: output.PartyRole(party.Role),
})
}
return endpoints
}
// SelectParties randomly selects n parties from the available pool
// Implements output.PartyPoolPort interface
func (d *MessageRouterPartyDiscovery) SelectParties(n int) ([]output.PartyEndpoint, error) {
available := d.GetAvailableParties()
if len(available) < n {
return nil, fmt.Errorf("insufficient parties: need %d, have %d", n, len(available))
}
// For now, return first n parties
// TODO: Implement random selection or load balancing strategy
selected := make([]output.PartyEndpoint, n)
copy(selected, available[:n])
return selected, nil
}
// SelectPartiesWithFilter selects parties based on filter criteria
// Implements output.PartyPoolPort interface
func (d *MessageRouterPartyDiscovery) SelectPartiesWithFilter(filter output.PartySelectionFilter) ([]output.PartyEndpoint, error) {
var available []output.PartyEndpoint
// If role is specified, filter by role; otherwise get all available parties
if filter.Role != "" {
available = d.GetAvailablePartiesByRole(filter.Role)
} else {
available = d.GetAvailableParties()
}
if len(available) < filter.Count {
return nil, fmt.Errorf("insufficient parties with role %s: need %d, have %d", filter.Role, filter.Count, len(available))
}
// For now, return first n parties
// TODO: Implement random selection or load balancing strategy
selected := make([]output.PartyEndpoint, filter.Count)
copy(selected, available[:filter.Count])
d.logger.Info("Selected parties from Message Router",
zap.String("role", string(filter.Role)),
zap.Int("requested", filter.Count),
zap.Int("selected", len(selected)))
return selected, nil
}