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..c28ab33 --- /dev/null +++ b/go/internal/pusher/conn.go @@ -0,0 +1,132 @@ +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 + } + + // Respond to keep-alive pings + if string(data) == "ping" { + if err := c.ws.Write(ctx, websocket.MessageText, []byte("pong")); err != nil { + slog.Debug("websocket pong write error", "connId", c.id, "error", err) + return + } + continue + } + + 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..3053761 --- /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: 2 + 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 0000000..ecd2421 Binary files /dev/null and b/go/pusherservice differ diff --git a/go/skaffold.yaml b/go/skaffold.yaml index 0b8d805..49dd7c0 100644 --- a/go/skaffold.yaml +++ b/go/skaffold.yaml @@ -90,3 +90,38 @@ profiles: - k8s/prod/particleprocessorworker.yaml deploy: kubectl: {} +--- +apiVersion: skaffold/v4beta11 +kind: Config +metadata: + name: pusher +build: + local: {} + tagPolicy: + gitCommit: + variant: AbbrevCommitSha +profiles: + - name: dev + build: + artifacts: + - image: pusher + context: . + docker: + dockerfile: Dockerfile.pusherservice + manifests: + rawYaml: + - k8s/dev/pusher.yaml + deploy: + kubectl: {} + - name: prod + build: + artifacts: + - image: pusher + context: . + docker: + dockerfile: Dockerfile.pusherservice + manifests: + rawYaml: + - k8s/prod/pusher.yaml + deploy: + kubectl: {} diff --git a/js/src/App.tsx b/js/src/App.tsx index ffbfe62..514f73b 100644 --- a/js/src/App.tsx +++ b/js/src/App.tsx @@ -14,6 +14,7 @@ import ParticleViewResolver from "@/features/particles/particle-view-resolver"; import Layout from "@/features/layout"; import NetworkSettingsPage from "@/features/network-settings"; import { Toaster } from "@/components/ui/sonner"; +import { PusherProvider } from "@/lib/pusher-provider"; const queryClient = new QueryClient(); @@ -37,7 +38,11 @@ const App = () => { return ; } - return ; + return ( + + + + ); }; function AutoplayNavigationListener() { diff --git a/js/src/components/composing-indicator.tsx b/js/src/components/composing-indicator.tsx new file mode 100644 index 0000000..8c22b51 --- /dev/null +++ b/js/src/components/composing-indicator.tsx @@ -0,0 +1,47 @@ +import type { Human } from "@/api/types"; +import type { ComposingUser } from "@/features/particles/stream-presence-context"; + +interface ComposingIndicatorProps { + users: ComposingUser[]; + networkHumans?: Human[]; +} + +/** + * Composing indicators pinned to the left edge, text running bottom-to-top + * via writing-mode so it hugs the edge without transform math issues. + */ +export function ComposingIndicator({ + users, + networkHumans, +}: ComposingIndicatorProps) { + if (users.length === 0) return null; + + return ( +
+ {users.map((u) => { + const human = networkHumans?.find((h) => h.id === u.humanId); + const name = human?.email_prefix ?? u.humanId; + const modeLabel = u.mode === "typing" ? "typing" : "recording"; + + return ( +
+ + + + + + + {name} {modeLabel} + +
+ ); + })} +
+ ); +} diff --git a/js/src/features/compose/compose-overlay.tsx b/js/src/features/compose/compose-overlay.tsx index f2cb065..018affc 100644 --- a/js/src/features/compose/compose-overlay.tsx +++ b/js/src/features/compose/compose-overlay.tsx @@ -17,7 +17,7 @@ import { createImageThumbnail } from "@/lib/image-thumbnail"; import { MAX_ATTACHMENT_SIZE_BYTES, MAX_ATTACHMENTS } from "@/lib/constants"; import type { PendingAttachment } from "@/features/compose/attachment-strip"; -type ComposeStep = "idle" | "picking" | "recording" | "reviewing" | "typing" | "configuring" | "submitting"; +export type ComposeStep = "idle" | "picking" | "recording" | "reviewing" | "typing" | "configuring" | "submitting"; type RecordingSource = "media" | "screen"; @@ -26,6 +26,7 @@ interface ComposeOverlayProps { // Optional target path for reply mode. If not provided, compose creates a new stream. targetPath?: ParticlePath; onActiveChange?: (active: boolean) => void; + onStepChange?: (step: ComposeStep) => void; onParticleCreated?: (particleId: string) => void; /** When true, composing is blocked (e.g. stream is closed). */ disabled?: boolean; @@ -42,6 +43,7 @@ export function ComposeOverlay({ networkId, targetPath, onActiveChange, + onStepChange, onParticleCreated, disabled, }: ComposeOverlayProps) { @@ -77,7 +79,8 @@ export function ComposeOverlay({ // Notify parent when active state changes useEffect(() => { onActiveChange?.(step !== "idle"); - }, [step, onActiveChange]); + onStepChange?.(step); + }, [step, onActiveChange, onStepChange]); const revokeAttachmentThumbnails = useCallback((items: PendingAttachment[]) => { for (const a of items) { diff --git a/js/src/features/particles/playback-page-indicator.tsx b/js/src/features/particles/playback-page-indicator.tsx index 10cfa27..d7d1bee 100644 --- a/js/src/features/particles/playback-page-indicator.tsx +++ b/js/src/features/particles/playback-page-indicator.tsx @@ -1,4 +1,4 @@ -import { Avatar, AvatarFallback } from "@/components/ui/avatar"; +import { Avatar, AvatarBadge, AvatarFallback } from "@/components/ui/avatar"; import { Tooltip, TooltipContent, @@ -14,6 +14,8 @@ interface PlaybackPageIndicatorProps { progress: number; onGoTo: (index: number) => void; presenceBySegment?: Map; + /** Set of humanIds currently online in the stream channel. */ + onlineHumanIds?: Set; /** Render only avatars or only tracks. Omit to render both. */ layer?: "avatars" | "tracks"; } @@ -24,6 +26,7 @@ export function PlaybackPageIndicator({ progress, onGoTo, presenceBySegment, + onlineHumanIds, layer, }: PlaybackPageIndicatorProps) { if (total === 0) return null; @@ -38,7 +41,7 @@ export function PlaybackPageIndicator({ return (
{showAvatars && presence && presence.length > 0 && ( - + )} {showTracks && (
)} + {/* Composing indicator — left edge, always visible */} + + @@ -454,6 +495,7 @@ export function StreamView({ path, streamParticle }: StreamViewProps) { progress={progress} onGoTo={goTo} presenceBySegment={presenceBySegment} + onlineHumanIds={onlineHumanIds} exitRemainingMs={exitRemainingMs} onOpenKeybindings={() => setShowKeybindings(true)} /> @@ -475,6 +517,7 @@ function BottomBar({ progress, onGoTo, presenceBySegment, + onlineHumanIds, exitRemainingMs, onOpenKeybindings, }: { @@ -484,6 +527,7 @@ function BottomBar({ progress: number; onGoTo: (index: number) => void; presenceBySegment: Map; + onlineHumanIds: Set; exitRemainingMs: number | null; onOpenKeybindings: () => void; }) { @@ -499,6 +543,7 @@ function BottomBar({ progress={progress} onGoTo={onGoTo} presenceBySegment={presenceBySegment} + onlineHumanIds={onlineHumanIds} layer="avatars" /> {/* Blurred background container — tracks + controls */} @@ -711,17 +756,19 @@ function NetworkBreadcrumbContent({ networkId }: { networkId: string }) { function ParticleBreadcrumbContent({ particle, networkId }: { particle: Particle; networkId: string }) { const network = useNetwork(networkId); + const { onlineHumanIds } = useStreamPresence(); const creator = network?.humans?.find((h) => h.id === particle.created_by_human_id); const prefix = creator?.email_prefix ?? particle.created_by_human_id; const initials = prefix.slice(0, 2).toUpperCase(); + const isOnline = particle.created_by_human_id ? onlineHumanIds.has(particle.created_by_human_id) : false; return ( - + {initials} + {isOnline && } {prefix} - diff --git a/js/src/hooks/use-channel.ts b/js/src/hooks/use-channel.ts new file mode 100644 index 0000000..0cc1fd7 --- /dev/null +++ b/js/src/hooks/use-channel.ts @@ -0,0 +1,91 @@ +import { useEffect, useState, useCallback, useRef } from "react"; +import { usePusherClient } from "@/lib/pusher-provider"; +import type { ChannelMessage } from "@/lib/pusher-client"; + +interface UseChannelResult { + /** Current set of humanIds present in the channel */ + presence: string[]; + /** Messages received on this channel (since the hook mounted) */ + messages: ChannelMessage[]; + /** Send a message to the channel */ + sendMessage: (payload: unknown) => void; +} + +/** + * Subscribe to a pusher channel. Manages presence tracking and message delivery. + * Subscribes on mount, unsubscribes on unmount. + * + * @param channelId - The channel to subscribe to, or null to skip. + */ +export function useChannel(channelId: string | null): UseChannelResult { + const client = usePusherClient(); + const [presence, setPresence] = useState([]); + const [messages, setMessages] = useState([]); + + // Keep a ref to avoid re-subscribing when sendMessage changes + const clientRef = useRef(client); + const channelRef = useRef(channelId); + clientRef.current = client; + channelRef.current = channelId; + + useEffect(() => { + if (!client || !channelId) { + setPresence([]); + setMessages([]); + return; + } + + client.subscribe(channelId); + + const onSubscribed = (msg: { presence?: string[] }) => { + setPresence(msg.presence ?? []); + }; + + const onJoin = (msg: { humanId?: string }) => { + if (msg.humanId) { + setPresence((prev) => + prev.includes(msg.humanId!) ? prev : [...prev, msg.humanId!], + ); + } + }; + + const onLeave = (msg: { humanId?: string }) => { + if (msg.humanId) { + setPresence((prev) => prev.filter((id) => id !== msg.humanId)); + } + }; + + const onMessage = (msg: { humanId?: string; payload?: unknown }) => { + if (msg.humanId) { + setMessages((prev) => [ + ...prev, + { humanId: msg.humanId!, payload: msg.payload }, + ]); + } + }; + + client.on(channelId, "subscribed", onSubscribed); + client.on(channelId, "join", onJoin); + client.on(channelId, "leave", onLeave); + client.on(channelId, "message", onMessage); + + return () => { + client.off(channelId, "subscribed", onSubscribed); + client.off(channelId, "join", onJoin); + client.off(channelId, "leave", onLeave); + client.off(channelId, "message", onMessage); + client.unsubscribe(channelId); + }; + }, [client, channelId]); + + const sendMessage = useCallback( + (payload: unknown) => { + if (clientRef.current && channelRef.current) { + clientRef.current.sendMessage(channelRef.current, payload); + } + }, + [], + ); + + return { presence, messages, sendMessage }; +} diff --git a/js/src/lib/pusher-client.ts b/js/src/lib/pusher-client.ts new file mode 100644 index 0000000..1ec4e7b --- /dev/null +++ b/js/src/lib/pusher-client.ts @@ -0,0 +1,285 @@ +/** + * PusherClient manages a WebSocket connection to the pusher service. + * Handles authentication, reconnection with exponential backoff, + * channel subscriptions, and event dispatching. + */ + +export type ConnectionState = + | "disconnected" + | "connecting" + | "connected" + | "reconnecting"; + +export interface ChannelMessage { + humanId: string; + payload: unknown; +} + +// Server → Client message shape +interface ServerMessage { + type: "subscribed" | "join" | "leave" | "message" | "error"; + channel?: string; + humanId?: string; + presence?: string[]; + payload?: unknown; + message?: string; +} + +type ChannelEventType = "subscribed" | "join" | "leave" | "message"; +type ChannelEventCallback = (msg: ServerMessage) => void; + +interface PusherClientConfig { + url: string; + getToken: () => string | null; +} + +const INITIAL_RECONNECT_DELAY = 1000; +const MAX_RECONNECT_DELAY = 30000; +const PING_INTERVAL = 20000; // 20s — keeps alive through GKE gateway timeout + +export class PusherClient { + private config: PusherClientConfig; + private ws: WebSocket | null = null; + private state: ConnectionState = "disconnected"; + private stateListeners = new Set<(state: ConnectionState) => void>(); + + // Channel event listeners: channelId → eventType → callbacks + private listeners = new Map< + string, + Map> + >(); + + // Active subscriptions for re-subscribe on reconnect + private activeSubscriptions = new Set(); + + // Reconnection state + private reconnectDelay = INITIAL_RECONNECT_DELAY; + private reconnectTimer: ReturnType | null = null; + private shouldReconnect = false; + + // Keep-alive ping + private pingTimer: ReturnType | null = null; + + constructor(config: PusherClientConfig) { + this.config = config; + } + + get connectionState(): ConnectionState { + return this.state; + } + + connect(): void { + if (this.ws) return; + + const token = this.config.getToken(); + if (!token) { + console.warn("[pusher] no token available, cannot connect"); + return; + } + + this.shouldReconnect = true; + this.setState( + this.state === "reconnecting" ? "reconnecting" : "connecting", + ); + + const url = `${this.config.url}?token=${encodeURIComponent(token)}`; + this.ws = new WebSocket(url); + + this.ws.onopen = () => { + this.setState("connected"); + this.reconnectDelay = INITIAL_RECONNECT_DELAY; + this.startPing(); + this.resubscribeAll(); + }; + + this.ws.onclose = () => { + this.cleanup(); + if (this.shouldReconnect) { + this.scheduleReconnect(); + } + }; + + this.ws.onerror = (event) => { + console.warn("[pusher] websocket error", event); + // onclose will fire after onerror, so reconnection is handled there + }; + + this.ws.onmessage = (event) => { + this.handleMessage(event.data as string); + }; + } + + disconnect(): void { + this.shouldReconnect = false; + this.clearReconnectTimer(); + this.cleanup(); + this.activeSubscriptions.clear(); + this.setState("disconnected"); + } + + subscribe(channelId: string): void { + this.activeSubscriptions.add(channelId); + this.send({ type: "subscribe", channel: channelId }); + } + + unsubscribe(channelId: string): void { + this.activeSubscriptions.delete(channelId); + this.send({ type: "unsubscribe", channel: channelId }); + } + + sendMessage(channelId: string, payload: unknown): void { + this.send({ type: "message", channel: channelId, payload }); + } + + on( + channelId: string, + event: ChannelEventType, + callback: ChannelEventCallback, + ): void { + if (!this.listeners.has(channelId)) { + this.listeners.set(channelId, new Map()); + } + const channelListeners = this.listeners.get(channelId)!; + if (!channelListeners.has(event)) { + channelListeners.set(event, new Set()); + } + channelListeners.get(event)!.add(callback); + } + + off( + channelId: string, + event: ChannelEventType, + callback: ChannelEventCallback, + ): void { + const channelListeners = this.listeners.get(channelId); + if (!channelListeners) return; + const eventListeners = channelListeners.get(event); + if (!eventListeners) return; + eventListeners.delete(callback); + + // Cleanup empty maps + if (eventListeners.size === 0) channelListeners.delete(event); + if (channelListeners.size === 0) this.listeners.delete(channelId); + } + + onStateChange(callback: (state: ConnectionState) => void): () => void { + this.stateListeners.add(callback); + return () => this.stateListeners.delete(callback); + } + + // --- Private --- + + private send(msg: { type: string; channel?: string; payload?: unknown }): void { + if (this.ws?.readyState === WebSocket.OPEN) { + this.ws.send(JSON.stringify(msg)); + } + } + + private handleMessage(data: string): void { + // Ignore keep-alive pong responses + if (data === "pong") return; + + let msg: ServerMessage; + try { + msg = JSON.parse(data); + } catch { + console.warn("[pusher] failed to parse message", data); + return; + } + + if (msg.type === "error") { + console.warn("[pusher] server error:", msg.message); + return; + } + + if (!msg.channel) return; + + const channelListeners = this.listeners.get(msg.channel); + if (!channelListeners) return; + + const eventListeners = channelListeners.get(msg.type as ChannelEventType); + if (!eventListeners) return; + + for (const cb of eventListeners) { + try { + cb(msg); + } catch (err) { + console.error("[pusher] listener error", err); + } + } + } + + private resubscribeAll(): void { + for (const channelId of this.activeSubscriptions) { + this.send({ type: "subscribe", channel: channelId }); + } + } + + private scheduleReconnect(): void { + this.setState("reconnecting"); + + // Exponential backoff with jitter + const jitter = Math.random() * 0.5 + 0.75; // 0.75 - 1.25x + const delay = Math.min( + this.reconnectDelay * jitter, + MAX_RECONNECT_DELAY, + ); + + this.reconnectTimer = setTimeout(() => { + this.reconnectDelay = Math.min( + this.reconnectDelay * 2, + MAX_RECONNECT_DELAY, + ); + this.connect(); + }, delay); + } + + private cleanup(): void { + this.stopPing(); + if (this.ws) { + this.ws.onopen = null; + this.ws.onclose = null; + this.ws.onerror = null; + this.ws.onmessage = null; + if ( + this.ws.readyState === WebSocket.OPEN || + this.ws.readyState === WebSocket.CONNECTING + ) { + this.ws.close(); + } + this.ws = null; + } + } + + private clearReconnectTimer(): void { + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + } + + private startPing(): void { + this.stopPing(); + this.pingTimer = setInterval(() => { + // Send an empty message as a keep-alive + if (this.ws?.readyState === WebSocket.OPEN) { + this.ws.send("ping"); + } + }, PING_INTERVAL); + } + + private stopPing(): void { + if (this.pingTimer) { + clearInterval(this.pingTimer); + this.pingTimer = null; + } + } + + private setState(state: ConnectionState): void { + if (this.state === state) return; + this.state = state; + for (const cb of this.stateListeners) { + cb(state); + } + } +} diff --git a/js/src/lib/pusher-provider.tsx b/js/src/lib/pusher-provider.tsx new file mode 100644 index 0000000..8174d36 --- /dev/null +++ b/js/src/lib/pusher-provider.tsx @@ -0,0 +1,76 @@ +import { + createContext, + useContext, + useEffect, + useRef, + useState, + type ReactNode, +} from "react"; +import { PusherClient, type ConnectionState } from "./pusher-client"; +import { useSessionStore } from "@/stores/session-store"; + +const PusherContext = createContext(null); +const PusherStateContext = createContext("disconnected"); + +// TODO: make this configurable per environment +const PUSHER_URL = "wss://pusher.dev.flowy.live/ws"; + +export function PusherProvider({ children }: { children: ReactNode }) { + const token = useSessionStore((s) => s.token); + const clientRef = useRef(null); + const [connectionState, setConnectionState] = + useState("disconnected"); + + useEffect(() => { + if (!token) { + // Disconnect if token is cleared (logout) + if (clientRef.current) { + clientRef.current.disconnect(); + clientRef.current = null; + setConnectionState("disconnected"); + } + return; + } + + const client = new PusherClient({ + url: PUSHER_URL, + getToken: () => useSessionStore.getState().token, + }); + + clientRef.current = client; + + const unsubscribeState = client.onStateChange((state) => { + setConnectionState(state); + }); + + client.connect(); + + return () => { + unsubscribeState(); + client.disconnect(); + clientRef.current = null; + }; + }, [token]); + + return ( + + + {children} + + + ); +} + +/** + * Returns the PusherClient instance, or null if not connected. + */ +export function usePusherClient(): PusherClient | null { + return useContext(PusherContext); +} + +/** + * Returns the current WebSocket connection state. + */ +export function usePusherConnectionState(): ConnectionState { + return useContext(PusherStateContext); +}