From 3e35850bdfc79e19cc5f772b3d45e1332898750d Mon Sep 17 00:00:00 2001 From: talksik Date: Thu, 9 Apr 2026 11:12:35 -0700 Subject: [PATCH] setup infra for pusher service --- go/Dockerfile.pusherservice | 22 ++ go/Makefile | 2 +- go/README.md | 1 + go/cmd/pusherservice/main.go | 117 +++++++ go/genproto/llink/pusher/pusher.pb.go | 251 ++++++++++++++ go/genproto/llink/pusher/pusher_grpc.pb.go | 121 +++++++ go/go.mod | 1 + go/go.sum | 2 + go/internal/pusher/authorizer.go | 66 ++++ go/internal/pusher/channel.go | 59 ++++ go/internal/pusher/conn.go | 123 +++++++ go/internal/pusher/hub.go | 237 +++++++++++++ go/internal/pusher/redis_bridge.go | 372 +++++++++++++++++++++ go/internal/pusher/server.go | 90 +++++ go/internal/pusher/types.go | 36 ++ go/k8s/dev/pusher.yaml | 104 ++++++ go/k8s/prod/pusher.yaml | 101 ++++++ go/protocol | 2 +- go/pusherservice | Bin 0 -> 30076466 bytes go/skaffold.yaml | 35 ++ 20 files changed, 1740 insertions(+), 2 deletions(-) create mode 100644 go/Dockerfile.pusherservice create mode 100644 go/cmd/pusherservice/main.go create mode 100644 go/genproto/llink/pusher/pusher.pb.go create mode 100644 go/genproto/llink/pusher/pusher_grpc.pb.go create mode 100644 go/internal/pusher/authorizer.go create mode 100644 go/internal/pusher/channel.go create mode 100644 go/internal/pusher/conn.go create mode 100644 go/internal/pusher/hub.go create mode 100644 go/internal/pusher/redis_bridge.go create mode 100644 go/internal/pusher/server.go create mode 100644 go/internal/pusher/types.go create mode 100644 go/k8s/dev/pusher.yaml create mode 100644 go/k8s/prod/pusher.yaml create mode 100755 go/pusherservice diff --git a/go/Dockerfile.pusherservice b/go/Dockerfile.pusherservice new file mode 100644 index 0000000..fb808d7 --- /dev/null +++ b/go/Dockerfile.pusherservice @@ -0,0 +1,22 @@ +# golang two stage build +FROM golang:1.25 AS first-stage + +WORKDIR /app + +COPY go.mod go.sum ./ +RUN go mod download && go mod verify + +COPY . . + +WORKDIR /app/cmd/pusherservice +RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o main +RUN ls + +FROM alpine:latest AS second-stage +# for health check +RUN apk --no-cache add curl + +WORKDIR /app +COPY --from=first-stage /app/cmd/pusherservice . +RUN echo "copied over binary to production stage" +CMD ["./main"] diff --git a/go/Makefile b/go/Makefile index 868512b..d9ab0d9 100644 --- a/go/Makefile +++ b/go/Makefile @@ -46,7 +46,7 @@ migrate-prod: SKAFFOLD_DEFAULT_REPO=$(PROD_REPO) skaffold run -p migrations --kube-context prod --tail # ---- Deploy ---- -# Use MODULE=orion or MODULE=worker to deploy a single service, e.g.: +# Use MODULE=orion or MODULE=worker or MODULE=pusher to deploy a single service, e.g.: # make deploy-dev MODULE=orion .PHONY: deploy-dev diff --git a/go/README.md b/go/README.md index d0b86cf..fdd6085 100644 --- a/go/README.md +++ b/go/README.md @@ -12,6 +12,7 @@ make deploy-prod # Deploy a single service make deploy-dev MODULE=orion make deploy-dev MODULE=worker +make deploy-dev MODULE=pusher ``` ## Migrations diff --git a/go/cmd/pusherservice/main.go b/go/cmd/pusherservice/main.go new file mode 100644 index 0000000..816bb9d --- /dev/null +++ b/go/cmd/pusherservice/main.go @@ -0,0 +1,117 @@ +package main + +import ( + "context" + "fmt" + "log/slog" + "net" + "net/http" + "os" + "os/signal" + "syscall" + + "github.com/flowy-live/llink/internal" + "github.com/flowy-live/llink/internal/auth" + "github.com/flowy-live/llink/internal/db" + "github.com/flowy-live/llink/internal/network" + "github.com/flowy-live/llink/internal/pusher" + "github.com/flowy-live/llink/internal/utils" + + pbpusher "github.com/flowy-live/llink/genproto/llink/pusher" + "google.golang.org/grpc" +) + +const ( + redisDBAuth = 4 // shared with orion for session validation + redisDBPusher = 5 // dedicated to pusher state (presence, pub/sub) +) + +func main() { + port := utils.MustGetEnv("PORT") + grpcPort := utils.MustGetEnv("GRPC_PORT") + + // Initialize database (for network membership checks) + db.Init() + defer db.Cleanup() + + // Redis for auth session validation (same DB as orion) + authRedis := internal.ConnectAndTestRedis(redisDBAuth) + + // Redis for pusher state (presence hashes, pub/sub) + pusherRedis := internal.ConnectAndTestRedis(redisDBPusher) + + // Services + authSvc := auth.NewAuthService(authRedis, nil) // nil aeroSvc — pusher only calls GetSession + networkSvc := network.NewService(db.Pool()) + + // Pod identity (use hostname in k8s, which is the pod name) + podID, err := os.Hostname() + if err != nil { + podID = fmt.Sprintf("pod-%d", os.Getpid()) + } + + // Pusher core + bridge := pusher.NewRedisBridge(pusherRedis, podID) + authorizer := pusher.NewAuthorizer(networkSvc) + hub := pusher.NewHub(bridge, authorizer) + bridge.SetHub(hub) + server := pusher.NewServer(hub, bridge, authSvc) + + // Context for graceful shutdown + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Start hub event loop + go hub.Run(ctx) + + // Start Redis Pub/Sub listener + go bridge.Listen(ctx) + + // Start pod heartbeat + stale pod cleanup + go bridge.Heartbeat(ctx) + + // --- gRPC server (internal, for presence queries) --- + grpcListener, err := net.Listen("tcp", fmt.Sprintf("0.0.0.0:%s", grpcPort)) + if err != nil { + slog.Error("failed to listen for gRPC", "port", grpcPort, "error", err) + os.Exit(1) + } + grpcServer := grpc.NewServer() + pbpusher.RegisterPusherServiceServer(grpcServer, server) + go func() { + slog.Info("gRPC server listening", "port", grpcPort) + if err := grpcServer.Serve(grpcListener); err != nil { + slog.Error("gRPC server failed", "error", err) + } + }() + + // --- HTTP server (WebSocket + health) --- + mux := http.NewServeMux() + mux.HandleFunc("GET /health", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc("GET /ws", server.HandleWebSocket) + + httpAddr := fmt.Sprintf("0.0.0.0:%s", port) + httpServer := &http.Server{Addr: httpAddr, Handler: mux} + + go func() { + slog.Info("HTTP server listening", "addr", httpAddr) + if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + slog.Error("HTTP server failed", "error", err) + os.Exit(1) + } + }() + + // --- Graceful shutdown --- + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) + <-sigCh + + slog.Info("shutting down...") + cancel() // stops hub, bridge listener, heartbeat + + grpcServer.GracefulStop() + httpServer.Shutdown(context.Background()) + slog.Info("shutdown complete") +} diff --git a/go/genproto/llink/pusher/pusher.pb.go b/go/genproto/llink/pusher/pusher.pb.go new file mode 100644 index 0000000..beb85d5 --- /dev/null +++ b/go/genproto/llink/pusher/pusher.pb.go @@ -0,0 +1,251 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.35.1 +// protoc v5.28.3 +// source: llink/pusher/pusher.proto + +package pbpusher + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" +) + +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) +) + +type BulkGetPresenceRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + ChannelIds []string `protobuf:"bytes,1,rep,name=channel_ids,json=channelIds,proto3" json:"channel_ids,omitempty"` +} + +func (x *BulkGetPresenceRequest) Reset() { + *x = BulkGetPresenceRequest{} + mi := &file_llink_pusher_pusher_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *BulkGetPresenceRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*BulkGetPresenceRequest) ProtoMessage() {} + +func (x *BulkGetPresenceRequest) ProtoReflect() protoreflect.Message { + mi := &file_llink_pusher_pusher_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 BulkGetPresenceRequest.ProtoReflect.Descriptor instead. +func (*BulkGetPresenceRequest) Descriptor() ([]byte, []int) { + return file_llink_pusher_pusher_proto_rawDescGZIP(), []int{0} +} + +func (x *BulkGetPresenceRequest) GetChannelIds() []string { + if x != nil { + return x.ChannelIds + } + return nil +} + +type BulkGetPresenceResponse struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Presences map[string]*ChannelPresence `protobuf:"bytes,1,rep,name=presences,proto3" json:"presences,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` +} + +func (x *BulkGetPresenceResponse) Reset() { + *x = BulkGetPresenceResponse{} + mi := &file_llink_pusher_pusher_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *BulkGetPresenceResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*BulkGetPresenceResponse) ProtoMessage() {} + +func (x *BulkGetPresenceResponse) ProtoReflect() protoreflect.Message { + mi := &file_llink_pusher_pusher_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 BulkGetPresenceResponse.ProtoReflect.Descriptor instead. +func (*BulkGetPresenceResponse) Descriptor() ([]byte, []int) { + return file_llink_pusher_pusher_proto_rawDescGZIP(), []int{1} +} + +func (x *BulkGetPresenceResponse) GetPresences() map[string]*ChannelPresence { + if x != nil { + return x.Presences + } + return nil +} + +type ChannelPresence struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + HumanIds []string `protobuf:"bytes,1,rep,name=human_ids,json=humanIds,proto3" json:"human_ids,omitempty"` +} + +func (x *ChannelPresence) Reset() { + *x = ChannelPresence{} + mi := &file_llink_pusher_pusher_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ChannelPresence) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ChannelPresence) ProtoMessage() {} + +func (x *ChannelPresence) ProtoReflect() protoreflect.Message { + mi := &file_llink_pusher_pusher_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 ChannelPresence.ProtoReflect.Descriptor instead. +func (*ChannelPresence) Descriptor() ([]byte, []int) { + return file_llink_pusher_pusher_proto_rawDescGZIP(), []int{2} +} + +func (x *ChannelPresence) GetHumanIds() []string { + if x != nil { + return x.HumanIds + } + return nil +} + +var File_llink_pusher_pusher_proto protoreflect.FileDescriptor + +var file_llink_pusher_pusher_proto_rawDesc = []byte{ + 0x0a, 0x19, 0x6c, 0x6c, 0x69, 0x6e, 0x6b, 0x2f, 0x70, 0x75, 0x73, 0x68, 0x65, 0x72, 0x2f, 0x70, + 0x75, 0x73, 0x68, 0x65, 0x72, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0c, 0x6c, 0x6c, 0x69, + 0x6e, 0x6b, 0x2e, 0x70, 0x75, 0x73, 0x68, 0x65, 0x72, 0x22, 0x39, 0x0a, 0x16, 0x42, 0x75, 0x6c, + 0x6b, 0x47, 0x65, 0x74, 0x50, 0x72, 0x65, 0x73, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x65, 0x71, 0x75, + 0x65, 0x73, 0x74, 0x12, 0x1f, 0x0a, 0x0b, 0x63, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x5f, 0x69, + 0x64, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0a, 0x63, 0x68, 0x61, 0x6e, 0x6e, 0x65, + 0x6c, 0x49, 0x64, 0x73, 0x22, 0xca, 0x01, 0x0a, 0x17, 0x42, 0x75, 0x6c, 0x6b, 0x47, 0x65, 0x74, + 0x50, 0x72, 0x65, 0x73, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, + 0x12, 0x52, 0x0a, 0x09, 0x70, 0x72, 0x65, 0x73, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x18, 0x01, 0x20, + 0x03, 0x28, 0x0b, 0x32, 0x34, 0x2e, 0x6c, 0x6c, 0x69, 0x6e, 0x6b, 0x2e, 0x70, 0x75, 0x73, 0x68, + 0x65, 0x72, 0x2e, 0x42, 0x75, 0x6c, 0x6b, 0x47, 0x65, 0x74, 0x50, 0x72, 0x65, 0x73, 0x65, 0x6e, + 0x63, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x2e, 0x50, 0x72, 0x65, 0x73, 0x65, + 0x6e, 0x63, 0x65, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x09, 0x70, 0x72, 0x65, 0x73, 0x65, + 0x6e, 0x63, 0x65, 0x73, 0x1a, 0x5b, 0x0a, 0x0e, 0x50, 0x72, 0x65, 0x73, 0x65, 0x6e, 0x63, 0x65, + 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x33, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, + 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1d, 0x2e, 0x6c, 0x6c, 0x69, 0x6e, 0x6b, 0x2e, + 0x70, 0x75, 0x73, 0x68, 0x65, 0x72, 0x2e, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x50, 0x72, + 0x65, 0x73, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, + 0x01, 0x22, 0x2e, 0x0a, 0x0f, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x50, 0x72, 0x65, 0x73, + 0x65, 0x6e, 0x63, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x68, 0x75, 0x6d, 0x61, 0x6e, 0x5f, 0x69, 0x64, + 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x08, 0x68, 0x75, 0x6d, 0x61, 0x6e, 0x49, 0x64, + 0x73, 0x32, 0x6f, 0x0a, 0x0d, 0x50, 0x75, 0x73, 0x68, 0x65, 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, + 0x63, 0x65, 0x12, 0x5e, 0x0a, 0x0f, 0x42, 0x75, 0x6c, 0x6b, 0x47, 0x65, 0x74, 0x50, 0x72, 0x65, + 0x73, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x24, 0x2e, 0x6c, 0x6c, 0x69, 0x6e, 0x6b, 0x2e, 0x70, 0x75, + 0x73, 0x68, 0x65, 0x72, 0x2e, 0x42, 0x75, 0x6c, 0x6b, 0x47, 0x65, 0x74, 0x50, 0x72, 0x65, 0x73, + 0x65, 0x6e, 0x63, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x25, 0x2e, 0x6c, 0x6c, + 0x69, 0x6e, 0x6b, 0x2e, 0x70, 0x75, 0x73, 0x68, 0x65, 0x72, 0x2e, 0x42, 0x75, 0x6c, 0x6b, 0x47, + 0x65, 0x74, 0x50, 0x72, 0x65, 0x73, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, + 0x73, 0x65, 0x42, 0x3c, 0x5a, 0x3a, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, + 0x2f, 0x66, 0x6c, 0x6f, 0x77, 0x79, 0x2d, 0x6c, 0x69, 0x76, 0x65, 0x2f, 0x6c, 0x6c, 0x69, 0x6e, + 0x6b, 0x2f, 0x67, 0x65, 0x6e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x6c, 0x6c, 0x69, 0x6e, 0x6b, + 0x2f, 0x70, 0x75, 0x73, 0x68, 0x65, 0x72, 0x3b, 0x70, 0x62, 0x70, 0x75, 0x73, 0x68, 0x65, 0x72, + 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, +} + +var ( + file_llink_pusher_pusher_proto_rawDescOnce sync.Once + file_llink_pusher_pusher_proto_rawDescData = file_llink_pusher_pusher_proto_rawDesc +) + +func file_llink_pusher_pusher_proto_rawDescGZIP() []byte { + file_llink_pusher_pusher_proto_rawDescOnce.Do(func() { + file_llink_pusher_pusher_proto_rawDescData = protoimpl.X.CompressGZIP(file_llink_pusher_pusher_proto_rawDescData) + }) + return file_llink_pusher_pusher_proto_rawDescData +} + +var file_llink_pusher_pusher_proto_msgTypes = make([]protoimpl.MessageInfo, 4) +var file_llink_pusher_pusher_proto_goTypes = []any{ + (*BulkGetPresenceRequest)(nil), // 0: llink.pusher.BulkGetPresenceRequest + (*BulkGetPresenceResponse)(nil), // 1: llink.pusher.BulkGetPresenceResponse + (*ChannelPresence)(nil), // 2: llink.pusher.ChannelPresence + nil, // 3: llink.pusher.BulkGetPresenceResponse.PresencesEntry +} +var file_llink_pusher_pusher_proto_depIdxs = []int32{ + 3, // 0: llink.pusher.BulkGetPresenceResponse.presences:type_name -> llink.pusher.BulkGetPresenceResponse.PresencesEntry + 2, // 1: llink.pusher.BulkGetPresenceResponse.PresencesEntry.value:type_name -> llink.pusher.ChannelPresence + 0, // 2: llink.pusher.PusherService.BulkGetPresence:input_type -> llink.pusher.BulkGetPresenceRequest + 1, // 3: llink.pusher.PusherService.BulkGetPresence:output_type -> llink.pusher.BulkGetPresenceResponse + 3, // [3:4] is the sub-list for method output_type + 2, // [2:3] is the sub-list for method input_type + 2, // [2:2] is the sub-list for extension type_name + 2, // [2:2] is the sub-list for extension extendee + 0, // [0:2] is the sub-list for field type_name +} + +func init() { file_llink_pusher_pusher_proto_init() } +func file_llink_pusher_pusher_proto_init() { + if File_llink_pusher_pusher_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: file_llink_pusher_pusher_proto_rawDesc, + NumEnums: 0, + NumMessages: 4, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_llink_pusher_pusher_proto_goTypes, + DependencyIndexes: file_llink_pusher_pusher_proto_depIdxs, + MessageInfos: file_llink_pusher_pusher_proto_msgTypes, + }.Build() + File_llink_pusher_pusher_proto = out.File + file_llink_pusher_pusher_proto_rawDesc = nil + file_llink_pusher_pusher_proto_goTypes = nil + file_llink_pusher_pusher_proto_depIdxs = nil +} diff --git a/go/genproto/llink/pusher/pusher_grpc.pb.go b/go/genproto/llink/pusher/pusher_grpc.pb.go new file mode 100644 index 0000000..26f6600 --- /dev/null +++ b/go/genproto/llink/pusher/pusher_grpc.pb.go @@ -0,0 +1,121 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.5.1 +// - protoc v5.28.3 +// source: llink/pusher/pusher.proto + +package pbpusher + +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 ( + PusherService_BulkGetPresence_FullMethodName = "/llink.pusher.PusherService/BulkGetPresence" +) + +// PusherServiceClient is the client API for PusherService 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. +type PusherServiceClient interface { + BulkGetPresence(ctx context.Context, in *BulkGetPresenceRequest, opts ...grpc.CallOption) (*BulkGetPresenceResponse, error) +} + +type pusherServiceClient struct { + cc grpc.ClientConnInterface +} + +func NewPusherServiceClient(cc grpc.ClientConnInterface) PusherServiceClient { + return &pusherServiceClient{cc} +} + +func (c *pusherServiceClient) BulkGetPresence(ctx context.Context, in *BulkGetPresenceRequest, opts ...grpc.CallOption) (*BulkGetPresenceResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(BulkGetPresenceResponse) + err := c.cc.Invoke(ctx, PusherService_BulkGetPresence_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +// PusherServiceServer is the server API for PusherService service. +// All implementations must embed UnimplementedPusherServiceServer +// for forward compatibility. +type PusherServiceServer interface { + BulkGetPresence(context.Context, *BulkGetPresenceRequest) (*BulkGetPresenceResponse, error) + mustEmbedUnimplementedPusherServiceServer() +} + +// UnimplementedPusherServiceServer 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 UnimplementedPusherServiceServer struct{} + +func (UnimplementedPusherServiceServer) BulkGetPresence(context.Context, *BulkGetPresenceRequest) (*BulkGetPresenceResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method BulkGetPresence not implemented") +} +func (UnimplementedPusherServiceServer) mustEmbedUnimplementedPusherServiceServer() {} +func (UnimplementedPusherServiceServer) testEmbeddedByValue() {} + +// UnsafePusherServiceServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to PusherServiceServer will +// result in compilation errors. +type UnsafePusherServiceServer interface { + mustEmbedUnimplementedPusherServiceServer() +} + +func RegisterPusherServiceServer(s grpc.ServiceRegistrar, srv PusherServiceServer) { + // If the following call pancis, it indicates UnimplementedPusherServiceServer 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(&PusherService_ServiceDesc, srv) +} + +func _PusherService_BulkGetPresence_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(BulkGetPresenceRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(PusherServiceServer).BulkGetPresence(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: PusherService_BulkGetPresence_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(PusherServiceServer).BulkGetPresence(ctx, req.(*BulkGetPresenceRequest)) + } + return interceptor(ctx, in, info, handler) +} + +// PusherService_ServiceDesc is the grpc.ServiceDesc for PusherService service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var PusherService_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "llink.pusher.PusherService", + HandlerType: (*PusherServiceServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "BulkGetPresence", + Handler: _PusherService_BulkGetPresence_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "llink/pusher/pusher.proto", +} diff --git a/go/go.mod b/go/go.mod index 89fd4a3..2ff2e45 100644 --- a/go/go.mod +++ b/go/go.mod @@ -176,4 +176,5 @@ require ( google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect k8s.io/klog/v2 v2.110.1 // indirect + nhooyr.io/websocket v1.8.17 // indirect ) diff --git a/go/go.sum b/go/go.sum index 000cd72..c849499 100644 --- a/go/go.sum +++ b/go/go.sum @@ -493,3 +493,5 @@ gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q= gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA= k8s.io/klog/v2 v2.110.1 h1:U/Af64HJf7FcwMcXyKm2RPM22WZzyR7OSpYj5tg3cL0= k8s.io/klog/v2 v2.110.1/go.mod h1:YGtd1984u+GgbuZ7e08/yBuAfKLSO0+uR1Fhi6ExXjo= +nhooyr.io/websocket v1.8.17 h1:KEVeLJkUywCKVsnLIDlD/5gtayKp8VoCkksHCGGfT9Y= +nhooyr.io/websocket v1.8.17/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c= diff --git a/go/internal/pusher/authorizer.go b/go/internal/pusher/authorizer.go new file mode 100644 index 0000000..b0c4c3d --- /dev/null +++ b/go/internal/pusher/authorizer.go @@ -0,0 +1,66 @@ +package pusher + +import ( + "context" + "errors" + "strings" + + "github.com/flowy-live/llink/internal/network" +) + +var ErrUnauthorized = errors.New("unauthorized") + +// Authorizer validates whether a user can access a given channel. +type Authorizer struct { + networkSvc network.Service +} + +// NewAuthorizer creates a new channel authorizer. +func NewAuthorizer(networkSvc network.Service) *Authorizer { + return &Authorizer{networkSvc: networkSvc} +} + +// Authorize checks if the given humanID is allowed to subscribe to the channel. +// Channel formats: +// - network:{networkId} +// - stream:{networkId}:{streamId} +func (a *Authorizer) Authorize(ctx context.Context, channelID, humanID string) error { + parts := strings.SplitN(channelID, ":", 2) + if len(parts) < 2 { + return ErrUnauthorized + } + + channelType := parts[0] + rest := parts[1] + + switch channelType { + case "network": + return a.authorizeNetwork(ctx, rest, humanID) + case "stream": + return a.authorizeStream(ctx, rest, humanID) + default: + return ErrUnauthorized + } +} + +func (a *Authorizer) authorizeNetwork(ctx context.Context, networkID, humanID string) error { + isMember, err := a.networkSvc.IsMember(ctx, networkID, humanID) + if err != nil { + return err + } + if !isMember { + return ErrUnauthorized + } + return nil +} + +// authorizeStream expects rest to be "{networkId}:{streamId}". +// We only check network membership — stream visibility is handled by network access. +func (a *Authorizer) authorizeStream(ctx context.Context, rest, humanID string) error { + parts := strings.SplitN(rest, ":", 2) + if len(parts) < 2 { + return ErrUnauthorized + } + networkID := parts[0] + return a.authorizeNetwork(ctx, networkID, humanID) +} diff --git a/go/internal/pusher/channel.go b/go/internal/pusher/channel.go new file mode 100644 index 0000000..e1025c1 --- /dev/null +++ b/go/internal/pusher/channel.go @@ -0,0 +1,59 @@ +package pusher + +// Channel tracks the local connections subscribed to a channel on this pod. +// All methods are only called from the Hub goroutine — no locks needed. +type Channel struct { + id string + members map[*Conn]string // conn → humanID +} + +func newChannel(id string) *Channel { + return &Channel{ + id: id, + members: make(map[*Conn]string), + } +} + +func (ch *Channel) addMember(conn *Conn, humanID string) { + ch.members[conn] = humanID +} + +func (ch *Channel) removeMember(conn *Conn) { + delete(ch.members, conn) +} + +func (ch *Channel) isEmpty() bool { + return len(ch.members) == 0 +} + +// localHumanIDs returns the deduplicated set of humanIDs connected on this pod. +func (ch *Channel) localHumanIDs() []string { + seen := make(map[string]bool, len(ch.members)) + ids := make([]string, 0, len(ch.members)) + for _, hid := range ch.members { + if !seen[hid] { + seen[hid] = true + ids = append(ids, hid) + } + } + return ids +} + +// hasHumanID returns true if the given humanID has at least one local connection. +func (ch *Channel) hasHumanID(humanID string) bool { + for _, hid := range ch.members { + if hid == humanID { + return true + } + } + return false +} + +// broadcast sends a message to all local connections except the excluded one. +func (ch *Channel) broadcast(msg ServerMessage, exclude *Conn) { + for conn := range ch.members { + if conn != exclude { + conn.Send(msg) + } + } +} diff --git a/go/internal/pusher/conn.go b/go/internal/pusher/conn.go new file mode 100644 index 0000000..bcaa3f9 --- /dev/null +++ b/go/internal/pusher/conn.go @@ -0,0 +1,123 @@ +package pusher + +import ( + "context" + "encoding/json" + "log/slog" + "sync" + + "nhooyr.io/websocket" +) + +const sendBufferSize = 256 + +// Conn wraps a WebSocket connection with identity and a send buffer. +type Conn struct { + id string + humanID string + ws *websocket.Conn + send chan []byte + once sync.Once // ensures close logic runs once +} + +func newConn(id, humanID string, ws *websocket.Conn) *Conn { + return &Conn{ + id: id, + humanID: humanID, + ws: ws, + send: make(chan []byte, sendBufferSize), + } +} + +// ReadPump reads messages from the WebSocket and forwards them to the hub. +// It blocks until the connection is closed or the context is cancelled. +func (c *Conn) ReadPump(ctx context.Context, hub *Hub) { + defer hub.disconnect(c) + + for { + _, data, err := c.ws.Read(ctx) + if err != nil { + if ctx.Err() != nil { + return + } + slog.Debug("websocket read error", "connId", c.id, "error", err) + return + } + + var msg ClientMessage + if err := json.Unmarshal(data, &msg); err != nil { + c.sendError("invalid message format") + continue + } + + switch msg.Type { + case TypeSubscribe: + if msg.Channel == "" { + c.sendError("channel is required") + continue + } + hub.subscribeCh <- &subscribeRequest{conn: c, channelID: msg.Channel} + case TypeUnsubscribe: + if msg.Channel == "" { + c.sendError("channel is required") + continue + } + hub.unsubscribeCh <- &unsubscribeRequest{conn: c, channelID: msg.Channel} + case TypeMessage: + if msg.Channel == "" { + c.sendError("channel is required") + continue + } + hub.broadcastCh <- &broadcastRequest{conn: c, channelID: msg.Channel, payload: msg.Payload} + default: + c.sendError("unknown message type: " + msg.Type) + } + } +} + +// WritePump drains the send buffer and writes messages to the WebSocket. +func (c *Conn) WritePump(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case data, ok := <-c.send: + if !ok { + return + } + if err := c.ws.Write(ctx, websocket.MessageText, data); err != nil { + slog.Debug("websocket write error", "connId", c.id, "error", err) + return + } + } + } +} + +// Send enqueues a ServerMessage to be written to the WebSocket. +// If the send buffer is full, the connection is closed (slow client). +func (c *Conn) Send(msg ServerMessage) { + data, err := json.Marshal(msg) + if err != nil { + slog.Error("failed to marshal server message", "error", err) + return + } + + select { + case c.send <- data: + default: + slog.Warn("slow client, closing connection", "connId", c.id, "humanId", c.humanID) + c.Close() + } +} + +// Close closes the WebSocket connection and the send channel. +func (c *Conn) Close() { + c.once.Do(func() { + c.ws.Close(websocket.StatusNormalClosure, "closing") + close(c.send) + }) +} + +func (c *Conn) sendError(msg string) { + c.Send(ServerMessage{Type: TypeError, Message: msg}) +} diff --git a/go/internal/pusher/hub.go b/go/internal/pusher/hub.go new file mode 100644 index 0000000..689d455 --- /dev/null +++ b/go/internal/pusher/hub.go @@ -0,0 +1,237 @@ +package pusher + +import ( + "context" + "encoding/json" + "log/slog" +) + +type subscribeRequest struct { + conn *Conn + channelID string +} + +type unsubscribeRequest struct { + conn *Conn + channelID string +} + +type broadcastRequest struct { + conn *Conn + channelID string + payload json.RawMessage +} + +type remoteEvent struct { + channelID string + event redisEvent +} + +// Hub manages all local WebSocket connections and channels on this pod. +// All state mutations happen in a single goroutine via Go channels — no locks. +type Hub struct { + channels map[string]*Channel + connChannels map[*Conn]map[string]bool // reverse index: conn → set of channel IDs + + bridge *RedisBridge + authorizer *Authorizer + + subscribeCh chan *subscribeRequest + unsubscribeCh chan *unsubscribeRequest + broadcastCh chan *broadcastRequest + disconnectCh chan *Conn + remoteEventCh chan *remoteEvent +} + +// NewHub creates a new Hub. +func NewHub(bridge *RedisBridge, authorizer *Authorizer) *Hub { + return &Hub{ + channels: make(map[string]*Channel), + connChannels: make(map[*Conn]map[string]bool), + bridge: bridge, + authorizer: authorizer, + subscribeCh: make(chan *subscribeRequest, 256), + unsubscribeCh: make(chan *unsubscribeRequest, 256), + broadcastCh: make(chan *broadcastRequest, 256), + disconnectCh: make(chan *Conn, 256), + remoteEventCh: make(chan *remoteEvent, 256), + } +} + +// Run starts the hub event loop. Blocks until the context is cancelled. +func (h *Hub) Run(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + + case req := <-h.subscribeCh: + h.handleSubscribe(ctx, req) + + case req := <-h.unsubscribeCh: + h.handleUnsubscribe(ctx, req) + + case req := <-h.broadcastCh: + h.handleBroadcast(ctx, req) + + case conn := <-h.disconnectCh: + h.handleDisconnect(ctx, conn) + + case evt := <-h.remoteEventCh: + h.handleRemoteEvent(evt) + } + } +} + +func (h *Hub) handleSubscribe(ctx context.Context, req *subscribeRequest) { + // Authorize channel access + if err := h.authorizer.Authorize(ctx, req.channelID, req.conn.humanID); err != nil { + req.conn.Send(ServerMessage{ + Type: TypeError, + Channel: req.channelID, + Message: "unauthorized", + }) + return + } + + // Get or create local channel + ch, ok := h.channels[req.channelID] + if !ok { + ch = newChannel(req.channelID) + h.channels[req.channelID] = ch + } + + // Add to local channel + ch.addMember(req.conn, req.conn.humanID) + + // Track in reverse index + if h.connChannels[req.conn] == nil { + h.connChannels[req.conn] = make(map[string]bool) + } + h.connChannels[req.conn][req.channelID] = true + + // Register in Redis and get global presence + presence, err := h.bridge.Subscribe(ctx, req.channelID, req.conn.id, req.conn.humanID) + if err != nil { + slog.Error("redis subscribe failed", "channelId", req.channelID, "error", err) + // Still send local presence as fallback + presence = ch.localHumanIDs() + } + + // Send subscribed ack with presence snapshot + req.conn.Send(ServerMessage{ + Type: TypeSubscribed, + Channel: req.channelID, + Presence: presence, + }) +} + +func (h *Hub) handleUnsubscribe(ctx context.Context, req *unsubscribeRequest) { + ch, ok := h.channels[req.channelID] + if !ok { + return + } + + ch.removeMember(req.conn) + + // Remove from reverse index + if chans, ok := h.connChannels[req.conn]; ok { + delete(chans, req.channelID) + } + + // Update Redis + if err := h.bridge.Unsubscribe(ctx, req.channelID, req.conn.id, req.conn.humanID); err != nil { + slog.Error("redis unsubscribe failed", "channelId", req.channelID, "error", err) + } + + // Clean up empty local channel + if ch.isEmpty() { + delete(h.channels, req.channelID) + } +} + +func (h *Hub) handleBroadcast(ctx context.Context, req *broadcastRequest) { + ch, ok := h.channels[req.channelID] + if !ok { + return + } + + // Check that the sender is actually in the channel + if _, isMember := ch.members[req.conn]; !isMember { + req.conn.sendError("not subscribed to channel: " + req.channelID) + return + } + + // Deliver to local connections (except sender) + ch.broadcast(ServerMessage{ + Type: TypeMessage, + Channel: req.channelID, + HumanID: req.conn.humanID, + Payload: req.payload, + }, req.conn) + + // Publish to Redis for other pods + h.bridge.Broadcast(ctx, req.channelID, req.conn.humanID, req.payload) +} + +func (h *Hub) handleDisconnect(ctx context.Context, conn *Conn) { + chans, ok := h.connChannels[conn] + if !ok { + return + } + + for channelID := range chans { + ch, ok := h.channels[channelID] + if !ok { + continue + } + + ch.removeMember(conn) + + if err := h.bridge.Unsubscribe(ctx, channelID, conn.id, conn.humanID); err != nil { + slog.Error("redis unsubscribe on disconnect failed", "channelId", channelID, "error", err) + } + + if ch.isEmpty() { + delete(h.channels, channelID) + } + } + + delete(h.connChannels, conn) +} + +func (h *Hub) handleRemoteEvent(evt *remoteEvent) { + ch, ok := h.channels[evt.channelID] + if !ok { + return // no local connections care about this channel + } + + switch evt.event.Type { + case TypeJoin: + ch.broadcast(ServerMessage{ + Type: TypeJoin, + Channel: evt.channelID, + HumanID: evt.event.HumanID, + }, nil) + + case TypeLeave: + ch.broadcast(ServerMessage{ + Type: TypeLeave, + Channel: evt.channelID, + HumanID: evt.event.HumanID, + }, nil) + + case TypeMessage: + ch.broadcast(ServerMessage{ + Type: TypeMessage, + Channel: evt.channelID, + HumanID: evt.event.HumanID, + Payload: evt.event.Payload, + }, nil) + } +} + +// disconnect sends a connection to the disconnect channel. +func (h *Hub) disconnect(conn *Conn) { + h.disconnectCh <- conn +} diff --git a/go/internal/pusher/redis_bridge.go b/go/internal/pusher/redis_bridge.go new file mode 100644 index 0000000..908829b --- /dev/null +++ b/go/internal/pusher/redis_bridge.go @@ -0,0 +1,372 @@ +package pusher + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "strings" + "time" + + "github.com/redis/go-redis/v9" +) + +const ( + podHeartbeatInterval = 30 * time.Second + podHeartbeatTTL = 60 * time.Second + cleanupInterval = 60 * time.Second + + // Redis key prefixes + channelConnsPrefix = "pusher:ch:" + channelConnsSuffix = ":conns" + podKeyPrefix = "pusher:pod:" + pubsubPrefix = "pusher:events:" +) + +// redisEvent is published/received via Redis Pub/Sub for cross-pod communication. +type redisEvent struct { + Type string `json:"type"` // "join", "leave", "message" + HumanID string `json:"humanId,omitempty"` // who triggered the event + PodID string `json:"podId,omitempty"` // originating pod + Payload json.RawMessage `json:"payload,omitempty"` // for message events +} + +// RedisBridge handles cross-pod coordination via Redis Pub/Sub and presence tracking. +type RedisBridge struct { + client *redis.Client + podID string + hub *Hub // set after hub is created +} + +// NewRedisBridge creates a new Redis bridge for cross-pod coordination. +func NewRedisBridge(client *redis.Client, podID string) *RedisBridge { + return &RedisBridge{ + client: client, + podID: podID, + } +} + +// SetHub sets the hub reference. Called during initialization. +func (rb *RedisBridge) SetHub(hub *Hub) { + rb.hub = hub +} + +// --- Presence management (called by hub goroutine) --- + +// Subscribe adds a connection to a channel in Redis. +// Returns the current presence set for the channel. +func (rb *RedisBridge) Subscribe(ctx context.Context, channelID, connID, humanID string) ([]string, error) { + key := channelConnsKey(channelID) + field := rb.connField(connID) + + // Check if humanID was already present before adding + existingMembers, err := rb.client.HVals(ctx, key).Result() + if err != nil && err != redis.Nil { + return nil, fmt.Errorf("failed to get channel members: %w", err) + } + + wasPresent := containsString(existingMembers, humanID) + + // Add this connection + if err := rb.client.HSet(ctx, key, field, humanID).Err(); err != nil { + return nil, fmt.Errorf("failed to add connection to channel: %w", err) + } + + // Publish join event if this is a new humanID in the channel + if !wasPresent { + rb.publishEvent(ctx, channelID, redisEvent{ + Type: TypeJoin, + HumanID: humanID, + PodID: rb.podID, + }) + } + + // Return deduplicated presence set + allMembers, err := rb.client.HVals(ctx, key).Result() + if err != nil { + return nil, fmt.Errorf("failed to get channel members: %w", err) + } + return deduplicateStrings(allMembers), nil +} + +// Unsubscribe removes a connection from a channel in Redis. +func (rb *RedisBridge) Unsubscribe(ctx context.Context, channelID, connID, humanID string) error { + key := channelConnsKey(channelID) + field := rb.connField(connID) + + if err := rb.client.HDel(ctx, key, field).Err(); err != nil { + return fmt.Errorf("failed to remove connection from channel: %w", err) + } + + // Check if this humanID is still present via other connections + remainingMembers, err := rb.client.HVals(ctx, key).Result() + if err != nil && err != redis.Nil { + return fmt.Errorf("failed to get remaining members: %w", err) + } + + if !containsString(remainingMembers, humanID) { + rb.publishEvent(ctx, channelID, redisEvent{ + Type: TypeLeave, + HumanID: humanID, + PodID: rb.podID, + }) + } + + // Clean up empty channel hash + if len(remainingMembers) == 0 { + rb.client.Del(ctx, key) + } + + return nil +} + +// Broadcast publishes a message event to all pods. +func (rb *RedisBridge) Broadcast(ctx context.Context, channelID, humanID string, payload json.RawMessage) { + rb.publishEvent(ctx, channelID, redisEvent{ + Type: TypeMessage, + HumanID: humanID, + PodID: rb.podID, + Payload: payload, + }) +} + +// GetPresence returns the deduplicated humanIDs for the given channels. +func (rb *RedisBridge) GetPresence(ctx context.Context, channelIDs []string) (map[string][]string, error) { + result := make(map[string][]string, len(channelIDs)) + for _, chID := range channelIDs { + members, err := rb.client.HVals(ctx, channelConnsKey(chID)).Result() + if err != nil && err != redis.Nil { + return nil, fmt.Errorf("failed to get presence for %s: %w", chID, err) + } + result[chID] = deduplicateStrings(members) + } + return result, nil +} + +// --- Pub/Sub listener (runs in its own goroutine) --- + +// Listen subscribes to Redis Pub/Sub and forwards events to the local hub. +// Blocks until the context is cancelled. +func (rb *RedisBridge) Listen(ctx context.Context) { + pubsub := rb.client.PSubscribe(ctx, pubsubPrefix+"*") + defer pubsub.Close() + + ch := pubsub.Channel() + for { + select { + case <-ctx.Done(): + return + case msg, ok := <-ch: + if !ok { + return + } + rb.handlePubSubMessage(msg) + } + } +} + +func (rb *RedisBridge) handlePubSubMessage(msg *redis.Message) { + // Extract channel ID from topic: "pusher:events:{channelID}" + channelID := strings.TrimPrefix(msg.Channel, pubsubPrefix) + if channelID == "" { + return + } + + var event redisEvent + if err := json.Unmarshal([]byte(msg.Payload), &event); err != nil { + slog.Error("failed to parse pub/sub event", "error", err) + return + } + + // Skip events originating from this pod — the local hub already handled them + if event.PodID == rb.podID { + return + } + + if rb.hub == nil { + return + } + + // Forward to local hub for delivery to local WebSocket connections + rb.hub.remoteEventCh <- &remoteEvent{ + channelID: channelID, + event: event, + } +} + +// --- Heartbeat + cleanup (runs in its own goroutine) --- + +// Heartbeat maintains this pod's liveness key and cleans up stale pods. +func (rb *RedisBridge) Heartbeat(ctx context.Context) { + podKey := podKeyPrefix + rb.podID + + // Initial heartbeat + rb.client.Set(ctx, podKey, "alive", podHeartbeatTTL) + + heartbeatTicker := time.NewTicker(podHeartbeatInterval) + cleanupTicker := time.NewTicker(cleanupInterval) + defer heartbeatTicker.Stop() + defer cleanupTicker.Stop() + + for { + select { + case <-ctx.Done(): + // On shutdown, remove our pod key and clean up our connections + rb.client.Del(context.Background(), podKey) + rb.cleanupPod(context.Background(), rb.podID) + return + case <-heartbeatTicker.C: + rb.client.Set(ctx, podKey, "alive", podHeartbeatTTL) + case <-cleanupTicker.C: + rb.cleanupStalePods(ctx) + } + } +} + +func (rb *RedisBridge) cleanupStalePods(ctx context.Context) { + // Scan all channel conn hashes for pod IDs, then check if those pods are still alive + var cursor uint64 + knownPods := make(map[string]bool) + alivePods := make(map[string]bool) + + for { + keys, nextCursor, err := rb.client.Scan(ctx, cursor, channelConnsPrefix+"*"+channelConnsSuffix, 100).Result() + if err != nil { + slog.Error("failed to scan channel keys", "error", err) + return + } + + for _, key := range keys { + fields, err := rb.client.HKeys(ctx, key).Result() + if err != nil { + continue + } + for _, field := range fields { + podID := extractPodID(field) + if podID != "" { + knownPods[podID] = true + } + } + } + + cursor = nextCursor + if cursor == 0 { + break + } + } + + // Check which pods are still alive + for podID := range knownPods { + exists, err := rb.client.Exists(ctx, podKeyPrefix+podID).Result() + if err != nil { + continue + } + if exists > 0 { + alivePods[podID] = true + } + } + + // Clean up dead pods + for podID := range knownPods { + if !alivePods[podID] { + slog.Info("cleaning up stale pod", "podId", podID) + rb.cleanupPod(ctx, podID) + } + } +} + +func (rb *RedisBridge) cleanupPod(ctx context.Context, podID string) { + var cursor uint64 + for { + keys, nextCursor, err := rb.client.Scan(ctx, cursor, channelConnsPrefix+"*"+channelConnsSuffix, 100).Result() + if err != nil { + return + } + + for _, key := range keys { + fields, err := rb.client.HGetAll(ctx, key).Result() + if err != nil { + continue + } + + channelID := extractChannelID(key) + for field, humanID := range fields { + if extractPodID(field) == podID { + rb.client.HDel(ctx, key, field) + // Check if this humanID is now gone from the channel + remaining, _ := rb.client.HVals(ctx, key).Result() + if !containsString(remaining, humanID) { + rb.publishEvent(ctx, channelID, redisEvent{ + Type: TypeLeave, + HumanID: humanID, + PodID: rb.podID, + }) + } + } + } + } + + cursor = nextCursor + if cursor == 0 { + break + } + } +} + +// --- Helpers --- + +func (rb *RedisBridge) connField(connID string) string { + return rb.podID + ":" + connID +} + +func (rb *RedisBridge) publishEvent(ctx context.Context, channelID string, event redisEvent) { + data, err := json.Marshal(event) + if err != nil { + slog.Error("failed to marshal event", "error", err) + return + } + if err := rb.client.Publish(ctx, pubsubPrefix+channelID, data).Err(); err != nil { + slog.Error("failed to publish event", "channelId", channelID, "error", err) + } +} + +func channelConnsKey(channelID string) string { + return channelConnsPrefix + channelID + channelConnsSuffix +} + +func extractChannelID(redisKey string) string { + // "pusher:ch:{channelID}:conns" → channelID + s := strings.TrimPrefix(redisKey, channelConnsPrefix) + s = strings.TrimSuffix(s, channelConnsSuffix) + return s +} + +func extractPodID(field string) string { + // "{podID}:{connID}" → podID + parts := strings.SplitN(field, ":", 2) + if len(parts) == 2 { + return parts[0] + } + return "" +} + +func containsString(slice []string, s string) bool { + for _, v := range slice { + if v == s { + return true + } + } + return false +} + +func deduplicateStrings(slice []string) []string { + seen := make(map[string]bool, len(slice)) + result := make([]string, 0, len(slice)) + for _, s := range slice { + if !seen[s] { + seen[s] = true + result = append(result, s) + } + } + return result +} diff --git a/go/internal/pusher/server.go b/go/internal/pusher/server.go new file mode 100644 index 0000000..80ae174 --- /dev/null +++ b/go/internal/pusher/server.go @@ -0,0 +1,90 @@ +package pusher + +import ( + "context" + "log/slog" + "net/http" + + "github.com/flowy-live/llink/genproto/llink/pusher" + "github.com/flowy-live/llink/internal/auth" + "github.com/google/uuid" + "nhooyr.io/websocket" +) + +// Server handles WebSocket upgrades and gRPC presence queries. +type Server struct { + pbpusher.UnimplementedPusherServiceServer + + hub *Hub + bridge *RedisBridge + authSvc auth.AuthService +} + +// NewServer creates a new pusher server. +func NewServer(hub *Hub, bridge *RedisBridge, authSvc auth.AuthService) *Server { + return &Server{ + hub: hub, + bridge: bridge, + authSvc: authSvc, + } +} + +// HandleWebSocket handles the WebSocket upgrade and connection lifecycle. +func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) { + // Authenticate via query param (WebSocket upgrade can't use custom headers) + token := r.URL.Query().Get("token") + if token == "" { + http.Error(w, "token required", http.StatusUnauthorized) + return + } + + session, err := s.authSvc.GetSession(r.Context(), token) + if err != nil { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + + // Accept WebSocket upgrade + ws, err := websocket.Accept(w, r, &websocket.AcceptOptions{ + // Allow all origins for now — CORS is handled at the gateway level + InsecureSkipVerify: true, + }) + if err != nil { + slog.Error("websocket accept failed", "error", err) + return + } + + connID := uuid.New().String() + conn := newConn(connID, session.HumanId, ws) + + slog.Info("websocket connected", "connId", connID, "humanId", session.HumanId) + + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + + // WritePump runs in a separate goroutine + go conn.WritePump(ctx) + + // ReadPump blocks until the connection closes + conn.ReadPump(ctx, s.hub) + + slog.Info("websocket disconnected", "connId", connID, "humanId", session.HumanId) +} + +// BulkGetPresence implements the gRPC PusherService. +func (s *Server) BulkGetPresence(ctx context.Context, req *pbpusher.BulkGetPresenceRequest) (*pbpusher.BulkGetPresenceResponse, error) { + presence, err := s.bridge.GetPresence(ctx, req.ChannelIds) + if err != nil { + return nil, err + } + + resp := &pbpusher.BulkGetPresenceResponse{ + Presences: make(map[string]*pbpusher.ChannelPresence, len(presence)), + } + for chID, humanIDs := range presence { + resp.Presences[chID] = &pbpusher.ChannelPresence{ + HumanIds: humanIDs, + } + } + return resp, nil +} diff --git a/go/internal/pusher/types.go b/go/internal/pusher/types.go new file mode 100644 index 0000000..a518046 --- /dev/null +++ b/go/internal/pusher/types.go @@ -0,0 +1,36 @@ +package pusher + +import "encoding/json" + +// Client → Server message types +const ( + TypeSubscribe = "subscribe" + TypeUnsubscribe = "unsubscribe" + TypeMessage = "message" +) + +// Server → Client message types +const ( + TypeSubscribed = "subscribed" + TypeJoin = "join" + TypeLeave = "leave" + // TypeMessage is reused for server → client messages + TypeError = "error" +) + +// ClientMessage is a message sent from a WebSocket client to the server. +type ClientMessage struct { + Type string `json:"type"` + Channel string `json:"channel,omitempty"` + Payload json.RawMessage `json:"payload,omitempty"` +} + +// ServerMessage is a message sent from the server to a WebSocket client. +type ServerMessage struct { + Type string `json:"type"` + Channel string `json:"channel,omitempty"` + HumanID string `json:"humanId,omitempty"` + Presence []string `json:"presence,omitempty"` + Payload json.RawMessage `json:"payload,omitempty"` + Message string `json:"message,omitempty"` +} diff --git a/go/k8s/dev/pusher.yaml b/go/k8s/dev/pusher.yaml new file mode 100644 index 0000000..a28bbe2 --- /dev/null +++ b/go/k8s/dev/pusher.yaml @@ -0,0 +1,104 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: pusher +spec: + selector: + matchLabels: + app: pusher + replicas: 1 + template: + metadata: + labels: + app: pusher + spec: + serviceAccountName: default-service-account + nodeSelector: + cloud.google.com/gke-spot: "true" + terminationGracePeriodSeconds: 15 + containers: + - name: pusher + image: "pusher" + ports: + - containerPort: 8080 + - containerPort: 50051 + resources: + requests: + memory: "52Mi" + cpu: 50m + limits: + memory: "52Mi" + cpu: 50m + env: + - name: "PORT" + value: "8080" + - name: "GRPC_PORT" + value: "50051" + - name: "REDIS_HOST" + value: "10.138.57.187" + - name: "LLINK_POSTGRES_CONNECTION_URL" + valueFrom: + secretKeyRef: + name: shared-secrets + key: LLINK_POSTGRES_CONNECTION_URL + +--- + +apiVersion: v1 +kind: Service +metadata: + name: pusher +spec: + selector: + app: pusher + ports: + - name: http + port: 8080 + targetPort: 8080 + protocol: TCP + - name: grpc + port: 50051 + targetPort: 50051 + protocol: TCP + +--- + +kind: HTTPRoute +apiVersion: gateway.networking.k8s.io/v1beta1 +metadata: + name: pusher +spec: + parentRefs: + - kind: Gateway + name: external-gateway + hostnames: + - pusher.dev.flowy.live + rules: + - backendRefs: + - name: pusher + port: 8080 + +--- + +apiVersion: networking.gke.io/v1 +kind: HealthCheckPolicy +metadata: + name: pusher-service-health-check +spec: + default: + checkIntervalSec: 15 + timeoutSec: 15 + healthyThreshold: 1 + unhealthyThreshold: 2 + logConfig: + enabled: true + config: + type: HTTP + httpHealthCheck: + portSpecification: USE_FIXED_PORT + port: 8080 + requestPath: /health + targetRef: + group: "" + kind: Service + name: pusher diff --git a/go/k8s/prod/pusher.yaml b/go/k8s/prod/pusher.yaml new file mode 100644 index 0000000..2eed66b --- /dev/null +++ b/go/k8s/prod/pusher.yaml @@ -0,0 +1,101 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: pusher +spec: + selector: + matchLabels: + app: pusher + replicas: 1 + template: + metadata: + labels: + app: pusher + spec: + serviceAccountName: default-service-account + containers: + - name: pusher + image: "pusher" + ports: + - containerPort: 8080 + - containerPort: 50051 + resources: + requests: + memory: "52Mi" + cpu: 50m + limits: + memory: "52Mi" + cpu: 50m + env: + - name: "PORT" + value: "8080" + - name: "GRPC_PORT" + value: "50051" + - name: "REDIS_HOST" + value: "10.204.2.99" + - name: "LLINK_POSTGRES_CONNECTION_URL" + valueFrom: + secretKeyRef: + name: shared-secrets + key: LLINK_POSTGRES_CONNECTION_URL + +--- + +apiVersion: v1 +kind: Service +metadata: + name: pusher +spec: + selector: + app: pusher + ports: + - name: http + port: 8080 + targetPort: 8080 + protocol: TCP + - name: grpc + port: 50051 + targetPort: 50051 + protocol: TCP + +--- + +kind: HTTPRoute +apiVersion: gateway.networking.k8s.io/v1beta1 +metadata: + name: pusher +spec: + parentRefs: + - kind: Gateway + name: external-gateway + hostnames: + - pusher.flowy.live + rules: + - backendRefs: + - name: pusher + port: 8080 + +--- + +apiVersion: networking.gke.io/v1 +kind: HealthCheckPolicy +metadata: + name: pusher-service-health-check +spec: + default: + checkIntervalSec: 15 + timeoutSec: 15 + healthyThreshold: 1 + unhealthyThreshold: 2 + logConfig: + enabled: true + config: + type: HTTP + httpHealthCheck: + portSpecification: USE_FIXED_PORT + port: 8080 + requestPath: /health + targetRef: + group: "" + kind: Service + name: pusher diff --git a/go/protocol b/go/protocol index d323ba3..9e3e1e0 160000 --- a/go/protocol +++ b/go/protocol @@ -1 +1 @@ -Subproject commit d323ba3d2b80577048bdfa06a247f12b6b3848c2 +Subproject commit 9e3e1e0e6a0cc942be32d6f7e86c49fec596ff87 diff --git a/go/pusherservice b/go/pusherservice new file mode 100755 index 0000000000000000000000000000000000000000..ecd242133188bc7502dd6d9c030f5ef6d7db472f GIT binary patch literal 30076466 zcmeFa37nPHegA*%G8>EmBdH{yv$DBiq5>1s&Ygu-g|w1tYunjjMiktVKwKDPSY)s| zN@4?TSrlZ(+8Tn5qzfPl2uaFfmYTH8J_Dl(Sli4MbpG$pd7kIqXD$P3V%lH(&%9pG z-1|JwS-$7HpYQpebN}Vtx86@TCSdpra($BPCxeU$n@qkL^J%VWTqPx6ESNg^k0<}Z zEIsZ2XDj!Abk?2PnMA>olG&4Qo!$SntsmLY|Dk)I!qX(Kko&fzb93` z?^OSu1(jv@mASWPk9UmI-XHyRm{Wv%e|ug%?^Adi3t>u1iVJ2JloZXHd9#DP$Oa47 zt#7S#igK?%?_T)QCAdJr0B{Z5-WOlVr8IxEcRsjEO6D%`k>M=(z8HSh!neUCobJV6 z-k;#=U;n;EcU9d{Gk=~_;;iweI_*Wi?6$!D`QfX%udMvs?L{7R+S@T@J6qKG=eF0J>OSMXpM3d>>Z@&UPR(8S zI?cI!5--#|&4F*B`$+fa8)r$$`~`R2cTY^j9K6NbE1mANS9X&$hcUW8e|xc}Z283Y zn!j3`4{UGFJ!N;&VB&bc z|D|bmyy>$YY<>Uh#U7+ylL_!S!<}1MUUF~6JvE8#{cGb*wmlD9|9X7-2ejwnFD@7T z=uyz*wD+UOF1PR$ocNIK-Lb$bj70b@{`?{f-z9ft+V(E~+Lu2ZeD{>ipIdeBS@^rQ z>6^B_!>>E|yY|G_&)c5(ZW>of$&J%z-Y~g%x?{yKs198}chz_cm)W(>w&UIF#^3Pg zxE~&ShUa2*VIyft&v$*_<52t9Wb^!LeWl&%5hy5OF9_l_@Ga8K2g1z(cGo#h34?xo^?HJ$P?vQ-344G={dily~p_<>%jRzxDKYzxey7 zzkLbM>2LAtzfxgVC3o@<0pE_pS^DGHPth2 z{?eCfu9-07nyQI^G=AQkTgtDVUtaXJnyRX=&Y$py*WO#RD0@(5Uku5Q4~|4ng#w53 zgF}19U0`OvKV(?%drLEV-^($Fo9CM7pS)+*)J`^sOy=bc8%^zzg?(ntS9$i%+(>j? zur}{M+f2hcRQY?q_`SyWD$L>6-wLd;bwe3V*p^|<7n^7cWil6*_BLN+qSL6mldJin zNc8#l&6-Mmzc>;#VbhQiF{kR^@0*>^v%Oq@%Jl`V4U^3aW@xyfWpHHMt@M8j&zk9f zmaJJ)A(7~#^k1y+^v|TJL2c#(!#rTP zl;@kd)c*)Bi-T(&983g;2|ROP2%iPRljp<12op`A?nA0SA`%VLR*t?W;Nbl;X3a95 zkK}r9X-V(9omY2q8 zS(0f^bqCDsmjh;dQP>+>bD|c^dO3V`iK5K)usfq0?6=(9{xoVB3N>>FZ|m zxG@tOc9)uyS&?xKmmUnP$;b@-uz7`PYz41^=|aJ@BEo$n`tL4W9y~P(@TlMHGmD?h zey6a$=uFX*i&M=hV4GcZ^1&xe`or~ECUh_>(EFXBiR86~BHPL`cdsolfhK&drRZ8C6rFbSTwS{-9#a4HXwf z8cR&Lrw5v@H|EqM;I%XyY4pPy9{Ir0((20I@}a@{f^a12f7gucw1iC4sU8=fIR+YB z+VoA~GbfVg$EWBoKJ>;J_>)Lp=FrF!vHLGZ^74&uJoLr*PsPSh`Fu0tp5}{Pe0cXR ze!THM67J8>A==FeWln4j_I@WcG!kux2iiiRgYb2a_`JjfdWytPCU9_+2_77EX4$G* zlT_bAxmNhP|9Hjc!KTBEx7e8NUE2d|KJj8;O$VR-_`Wu?q4YqU$2`8j+2u3w{p&tn z6Wv3pkwt>-8Dp+<@S0)!7kY;&(^gp|9To*zxV3L*X~*Gf%ClYZ~TDwUp)Igf#I0!30^ZhlGl&M zxAwmmypo-z&@cqwRKPbC&~UL!!!l@?WrjCgzp%p6Fa!-hi5(4@Km+r&VIgS5TzOas z8p^(hpdrf>_IaxNNi5mE58L;qfjRuJHhA%AY)>Wp-4O~mF-^a&oo9N^>~B~VvQUA1 zI=Va@lM&HKwp=o+ae4Y+9Q@|j@l`u;Y?l zS9UBe`IksvD_<_Z#Q4W6%%=xmqP+`(b3y_Wf@!r!gpbz24{i&UIU? zXoB_m=(0_sDDvbVi^d^6_Q;92E|Px!tj2-;_3LLtAAWuaR7MtMPKxVi58mlX{qY&3 zoZu*#WbRo441!HCUL=|_7J^z6Lhe}F7)zLbMAt&vE<~Rdej#!+WJ2{jP0;G&XVAfQ z*!#Nlq*a9`>0kwA@_DE6l@2(*VAJn2zUEMp)y)qjjP<{oov|KD)VCJ4$PN;Ghbc1(*#bvdiDCXPlLB--p`*H zp}hFt!9_u@V11&;^z;4ZyZq6dz#kqw$E`mRhi3zJUt@K2O-7S+nB?2vM;86&0Antn zf>Y%ypj{C@$TDEvoVjQ1Vqkq5Sj!nt9k7;!%_-@lKq`29AaHjJevZbhe&YJb8AF*j z2A7t<>(cV5Zwwy%HI^UG4L?y=3%~mLFk^W7JY&d7Wo~BqB2-@pUE1&s9_`q)zbD>O%ZqcM@5AXHZi4ki;HDMaw1$G#4|QhaSpMHsO)PH{7x`6qldFDvIN$HvK88dq^ zpO+0cv-@sN>g_w4w6O2n;oiP$k`}6;PV7b(;}vg@YhJYN#4Uk_=`~{}by;~zZ5n%K z>XR#xgTArmRH(0S_TNxW-&=>7*<%_1amtUSyhD>vQ@7jJJlcA5e4wGeX6&TYB-79q z8q{<Z%#GQT{oMPOcBu! z>DFiX;^DxWw#5Z`mFB^19nTi!?L>Di9lCUF%T#9`ay%VB@HP|02g%D}UW~8q=wfs^ zcDcTzC=%TQeiobLp7Ph1t!)DbiJr4fz1EC)IEP5u}-F1zL zc1&cxehYX9_Lfi@cuH;RJRVrnGBFZuzQ*FENBy?l!n}&UI^Y|%-34!Z<1E5WE{dDG` zv+VcOrh-82ksE#Ar`zu-_W2aw_agf}qbX|&zU)h%haZDY?V+!1TQ=D|zWM6Q8ah)V z+sZ?=N4``OSTppW88_o{_-0~5_ujx7>K)3O_PY(OgCE)^U6)u_c3}#20||BimAY1* z6%%5;}}#vZqoUR=k) zYqTq{=B2{^I%tF7#D91L*e(o4qVGIy8rBCd?Ad*~Z}vMmreRpnM1Pu~pA>I8&iu-+ zzXkuvfO!!)5&Y2zx>dOG+lg;_eS}?(IDNOg7w`K`U*Cp54}Syoe|7Nu z{V($MFM0`vQXic(AFw!Z;n?Vds$EI*!Ac2|C+^rJl?1N4}t$(;9qmZm*4Aq%IVkPfzK!Oo9*^1{QBu0 zAQtM;{Vm>k{e8IcIzFaPZq|GC?&`IZN> zE1%yK%;+%5zn?c-e7fOhKHmH&{P|DS?8on)d-EW#ykf@wXa4$LEK2q^K#V|f6UC%jkWJkyZlYL}eNR7%n-t z?e0=+EpgRBulj7KyPojg&m4YJTvhW~zh2t?Zd@<<=SKnH?-y_O;>o`FvW=AzCmIMp z!k>*td+~WMw(8LhIu1k^k3UquU;Nvv5C2)ZW`1ye3m^Rr1fS@i@1uVpA)cNX4~+2x zvCDzl6TZ$9&waeG_S3O=ZorKN2HaR+z>Nh4+*n}1jRoqtVuAV;PwTz>lE5w}7U<1= ztsEu9D`W8fhU8WF+xhPQF8O(`!;zoee;ZiS&9##&QGP!E%GvUB<^Ln{^P853`+fBM zjmv`>@qbPpPDlTD$wODWBM*Pro30@|8Vj%R|ji{}0)d^Vo;) z_~`kYl!wAHbF#2@0c}bKP(f=YJ{eR=~W3qmu@{?_9cZ9e~OlaP2;(9^y`*OH+U!Z&Pn{v2V z8=touJHz`OSzP40%=2;_{dq%PJZ_wg-(M2Tk0`vl_Q<12UhZ#Eukfl|ORZ5%bouNQ zF|ll7`ynp+@3HU@56a=v+PsBlXkQ%ppSNCcs)!sx^2^gkX^rUO*ch8x`(@pgJOyXnbN5hVb1;+} z@;uLjruK#WMLDH&=`UCz2m zl8Nk?lm6#Nw)|l8+dED2yd`J)W@{}6o)|;F`-!iYu%^0aBNoBi;W9E-Er>Asp<{S;@#qH_i%qVZMCuf-9A1Nt%o;T$IuV!_BZgHb`F`JUDS{j zu01kW?XaHRGA0tOOlYT+`x0L}i)cr1-@hlLp)@7ZD1I@uhIZg7);Zq-NAPna`CfU& zeJK;jsf{L8vc4KL6PA-hvL$FHwn5ugXxpN3lV>o7_Ko0qJ=~DRI&hKT0qz#yj_{p# zkDE20ZfFVC9{B?I*HgZk_3xI;xqyehej8Um)H~ z4#gKKuiO&x)PXnQ#p<$2JKu5MRq)R4dmy)ht~p0EOk-^~wP|VCtjP?RiOo5z*}M4ffJU9b+qE;2*A2}&eR!;( zeU()oL3prYn5l0k$G)=E;Su3lW%RxkzN_MWC+%#hWG&lm|03G^;Ys?PLcYZdgBn_f zEZ%nNZSdss$Itoh=Z_QIAE&%mpZt=G|EL*%lMubHQT&`V3XQ!4n4H(5BZ#IzlWM`wx3SUxAJxJxzA7@j;p_p z{m8N7hu@E{cfLk0$NNL#`nh>F{9``dFlm-q^PjZwI(f5evajj47bZ%(+6%L8tvOj1 znzdtN&|FI%STx8QLTD^&^^@S$X_091t&wO-btIa)FcMuz4p7!$)0lHnpmEfQVB?sa zP^0*w>~v8yxRQE*f$pYE_7KyUnG$H+xYj(LWv*-pq?)Mx9_IUS_j@Ye$GYFs`98`0 zK9uj%-0#EqeyjU^INz(??<4uX(EXmt_m%GVv3&mv6ICvTg)0<}dVE`&Vj7A2yEehG zS+ZC4;PY?A>-@>yo!aBrgXy0TPCpg12iK!(?tjkNzhJ-rG_|2>M5OWbH9__?n8uEE zX^qW%Zy_&3HlgO918Yi6?*y%TN-v-f?r!_>`nCT6PAW34%o}0b`=s3uar!8^M6?m= zsy|Y3#Fcr#_4xhwtF1_*{r+sgfpMbZL{s7aO3GQZhN7C)m3Wm^7FNK zvD@ByUweC;_D-)i76%<8LXFLQm+n)07rX7rCU|&kcJWv`=E}Ux%-SQFnH$~~UhfZG z*-#oXI~?72gVlYa@h742Bhc9MXI``G)p2=YA4FsBut?)h^k5k}B`Gx$9nEzNeb$0p zlQ)|;nu}944s14&Ba;JWg7o-kuFWRdJ|lN`g6Att37IuohZD>WEd70?pa0O2;d8D> zlhfvxXZajzuW%@OoLu;(L1uPnSg<~qs|tT=IhTbMo+Vd3y4;1mO!vUO+z0!s7nlhZ zz+K0+)PxRHFrFNq$zKT%3)Sav`P(cY2W=Vc&2ih@J(zpio8xP9)L=6~?WtXIr)`@{ z2hk?EW?q}ifN_os=SK9Ezs)S&C$zaN_3Sp=QfQN$Mz2jjoB`~Szs)7&q9wN3k#u&O z75|wsaU8e})kUJYTMkfoW(vW!==4E8-n;#@1;yo-3sX4PTghDdlO|VDBD3h8>=gN zYq`Eo+2%7T6I+p+W^9^#3(p4qp3A?pUA-yYsXxhJ>6kn7WK&A_W#lE!w>A`B|HDA( ztw$MGUP(ddhl%giPrqkpj;-|H2hObTcdz-I`l=`?$=d_Yu74o%Wbn-Cm5W1xd2RTj z%E=vZro`TtX})u2#L9JjeJ|>rWYyoc;P=`qr#<}73BlTym$WtTV|$t`(NIcr`$ zfBA=1-n5Xi7N}2h>*%}s{D*{k>dXF}Ikw+xOZSYg-@VtqH@?&8L&mqC@%8w|XYbwd z?f;15>vhMspYiSg5aa9hjc>m@zWwJJUuwem`rUiu6TXGN)1j=U;Ly{Pgul?4lGJh& zSOxx4)xO&Cm$&q}y2W1>+{DWYpYodmXO5-%_eA>Pao(?w!<4{pCgDDj{}S&#_&j=j zT>SWPxA&{^w@aUwT>7kZ>0|l(9Q5&*6@6l5MIR4e9{qnSKmGUP!hikrLE?R)zDT_H z@ayr<$D}_F`S|C<=nq@o)sG)lfAkxF;`ls&&WTUO<3Sqtf4;!{#_mn;AOCsmsN;Jn zKIP))(8q8OZC$dgD36_dV6$lcVSX4*)M!H)vv20FNyab#1G1C&f?Pi-ixQH z9q;dh-Zv9h!M+|Uos|?--zCQML{hbnBhWMFg(Yh{Q_S`QZ$9t@j`?=Z?y3)G?%p~% zclTE9A1J^V$q%IV6yTR-m_a>R#2faeN1m9%eu2ICum`cn;krogWGL1MlN_ z-}qkN_8e%`0S-5(2WDoOw1W{oyGD+=D)o}FS6N%Q;n!2g=vh9_6`2N z_MVBCN+RYIbC20Y?O%JMf_&j*6RI~?9J?`>Yvt34_EdVBUoHigYA-x&_tr7+Wfyakx#W>@%HM%; ze5wucWg&c79yZ$_c6n+PJhhp!N$hc$;_~Ducv8=*$ZO`zsA0m_!!BP@euv&IBd2}} z@mP5HCCa~~XC36ZYcI0OYtL9XH?nXPm-qx)&)($n7Hx>P?0y&zcg!Cz^UJ))PeGSw zat-@|jH%~b(FX02t_H6m#xNQ9GiYlo_)Q)b>DAm2-hpS$u%$^uCxk+&6Tpqi?U-_P z?@ zS6d6cwv0(V=-{%}p8qf}7x{qqwn78xpt%uqcuMu$UOPVcCN}kJ=#_NpuXpRGQ-8f%KmDY>JN46PYkgpljkO-|$zP&A*@ztU5dRxhTYDra zxbo>Fhj;C}Rz&h+eUBQc8ihkPa$t7~SxqlMUhEAe!b z8HpbFHv8;|{Z>{-c2t_7#Az~`#M5Qmmzm-AUOuhkFHGsRe1c46AXDY=SlPn-sS=g0=>ayMd#M(`5gNky|Z_o^zo}lzv-VGn0{LGLO;G}{)~8bZB|3w zbhF0pVFk~B^^f2a++>Eej)T5~R*y*kv9F}^iMk`X)3?1{m;2h=#ZR=qUH8DAb;P9( zE)5seXQUSO>_^vCoGCn7d1gpM-I*Cj(V;JD4NrWbeGpo!tAp=4LSNhV4E3zdgg4il zz^P~0UmGtUv3opKZ$sEj>|h_?<_BI`_bN7G+3R09+6le)jg3URA4poa`GI}wmLV_N zgSpeB_T(TxZSYPz^jRuir@x$9_T#0nZ%#9-m~7AQ`#q`dRl)VesP`XDc;g&Hu1({ zm1n&nGu1?M)3v7LosHv-*V!xhbIQrSTAp+F%65%DFelgo&2pOSE2WqH($en@XTb8)N0{kZ1?v*1E0i zGyWIrYTyr>mQ5T;@l(OLm-8e_!bv^l#Ae65`M^<)znbx{H0Ih=>`w|~8J%MqldH${ zPNj`L%5Flx^((hwe|%RC%5Kez+W- zYDM>yfcr*d#l!s)`Yr>$ZraRboum|ems@%!5!n}K5|bB5cowNq$gi%G6ep+9d--S78Z+r4m3 z?|%BvX6*j~{OV(}?IY=+*M6ZLr_(;?ZPk~#u$0?!!GjMcz)=7kik}LOr{Qhkqr`=y zlD3LXxIW2+V-7Uv1P<}(7U0N%PX&i#6N-Cf6C~#?==48;Ms^JkT+XuJ*~iNF8wcRS z|M-_UAC|g&DBI=y^*b$LYP^wOhLxWbb#l>6XN? zyEG786W3Ned-Dw+PImfm;^AZZ0Qd;uE1d%$k9+vI<2Q_tD?S81@QL9)XMMQe_$^+} zjg#N^;pCrPoOt*chc58zL**h!?&T}Bs2FIb)e29 zugXB~>vjd6kpJDm+5%@19M-dEcqjcK-BAp`wX-&njej9Mxg47(-)`@jqNANd4E{#) zI{8AX=Ur1xQvFzbv}yQk&*E=%p?kCOUH%w;D})c7^|q9z2z(eh!@352Rj!%b*oiE> zGAz7K{nWW_E~l*u+A6}|+roNCGxFLJN`}WG+Xj;HIDfgaFMr0hE${yzF4KQE0DsA@ z`Q=UPIBnpRiT;{4c)SAtPe6xPE^!3%|>#N(>m-L56zZ&cRj)>W_X6#lA z^lQZr_T;*m^94+AX2VAEM!x$U;76yoK$q5u1Iu+YbeG;QLN8RXMp(4cBl?XXc=5l!LR3c9R0;I-4uvgY8Md298u_ z31x_19a%!(4t^&0I(is=o6x`XjUSest}gQI&}4K`Qtqp3r!a@X6oV zkz&>a-cOm}jTv2do%DepZ+8eb^qtBSxUdx@!shdP@AGIkBoh5PwDahd_9^klr{nZu zp3;J^+jfI1bI{L|xen+cx!Z@_tu!eIk^_>vG5zFD>w}%Z#rYam?p`q|t87kC5%|Bs zk+*iqALEhybprniq7n{SFH~O&yv4s*y(Y|%yi^$Kl0ZN51%c6saF1sYUyd7pKSJ9kt<{Q4(tUtUE@le=c(M z4{&?>XA3aQ1%|1?;S)N5p&b}BZ@eY}25Tc-7`*RA;ADwlAg?A;oz>e(-^vG*P0mU$ zv}accH?7EGwlR4v`T9gh;_q-DZAQ=e@wq{HQENsuY)q~_@}0kjr(?F%!MER!-t&Fl zuW-%=QNEbtw>bY}{A}|n;_-LQi!(DFdbOZm?_rGcJ8vTnE__#j@1^*h1<2wJ8W%n% zu|P}TmOX)O0bq9E?H{|};KAa;;q~80KEy!tf5%QfK<DBRmrquei?mtXg