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"]

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

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

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")
+}

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
+}

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",
+}

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
 )

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=

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)
+}

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)
+		}
+	}
+}

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})
+}

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
+}

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
+}

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
+}

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"`
+}

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

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

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: {}

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 <LoginPage />;
   }
 
-  return <AuthenticatedApp />;
+  return (
+    <PusherProvider>
+      <AuthenticatedApp />
+    </PusherProvider>
+  );
 };
 
 function AutoplayNavigationListener() {

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 (
+    <div
+      className="z-100 absolute left-2 top-1/2 z-20 flex -translate-y-1/2 flex-col gap-1.5 animate-in fade-in duration-200"
+      style={{ writingMode: "vertical-rl" }}
+    >
+      {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 (
+          <div
+            key={u.humanId}
+            className="flex rotate-180 items-center gap-1.5 rounded-full bg-white/10 px-2 py-1 backdrop-blur-sm"
+          >
+            <span className="flex gap-0.5">
+              <span className="size-1 rounded-full bg-white/70 animate-bounce [animation-delay:0ms]" />
+              <span className="size-1 rounded-full bg-white/70 animate-bounce [animation-delay:150ms]" />
+              <span className="size-1 rounded-full bg-white/70 animate-bounce [animation-delay:300ms]" />
+            </span>
+            <span className="whitespace-nowrap text-[10px] text-white/50">
+              {name} {modeLabel}
+            </span>
+          </div>
+        );
+      })}
+    </div>
+  );
+}

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) {

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<number, HumanPresence[]>;
+  /** Set of humanIds currently online in the stream channel. */
+  onlineHumanIds?: Set<string>;
   /** 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 (
           <div key={i} className="flex flex-1 flex-col items-stretch">
             {showAvatars && presence && presence.length > 0 && (
-              <SegmentPresenceAvatars presence={presence} />
+              <SegmentPresenceAvatars presence={presence} onlineHumanIds={onlineHumanIds} />
             )}
             {showTracks && (
               <button
@@ -74,8 +77,10 @@ export function PlaybackPageIndicator({
 
 function SegmentPresenceAvatars({
   presence,
+  onlineHumanIds,
 }: {
   presence: HumanPresence[];
+  onlineHumanIds?: Set<string>;
 }) {
   const visible = presence.slice(0, MAX_VISIBLE_AVATARS);
   const overflow = presence.length - MAX_VISIBLE_AVATARS;
@@ -89,6 +94,9 @@ function SegmentPresenceAvatars({
               <AvatarFallback>
                 {human.emailPrefix.slice(0, 2).toUpperCase()}
               </AvatarFallback>
+              {onlineHumanIds?.has(human.humanId) && (
+                <AvatarBadge className="bg-green-500" />
+              )}
             </Avatar>
           </TooltipTrigger>
           <TooltipContent side="top" className="text-xs">

js/src/features/particles/stream-presence-context.tsx

@@ -0,0 +1,229 @@
+import {
+  createContext,
+  useCallback,
+  useContext,
+  useEffect,
+  useMemo,
+  useRef,
+  useState,
+  type ReactNode,
+} from "react";
+import { useChannel } from "@/hooks/use-channel";
+import { useAuthStore } from "@/stores/auth-store";
+import type { ChannelMessage } from "@/lib/pusher-client";
+
+// ---------------------------------------------------------------------------
+// Types
+// ---------------------------------------------------------------------------
+
+export type ComposingMode = "recording" | "typing" | "screen";
+
+export interface ComposingUser {
+  humanId: string;
+  mode: ComposingMode;
+  lastSeen: number;
+}
+
+interface StreamPresenceContextValue {
+  onlineHumanIds: Set<string>;
+  composingUsers: ComposingUser[];
+  startComposing: (mode: ComposingMode) => void;
+  stopComposing: () => void;
+}
+
+// ---------------------------------------------------------------------------
+// Constants
+// ---------------------------------------------------------------------------
+
+const COMPOSING_TIMEOUT_MS = 10_000;
+const COMPOSING_HEARTBEAT_MS = 5_000;
+const COMPOSING_CLEANUP_INTERVAL_MS = 2_000;
+
+// ---------------------------------------------------------------------------
+// Context
+// ---------------------------------------------------------------------------
+
+const StreamPresenceContext = createContext<StreamPresenceContextValue | null>(
+  null,
+);
+
+// ---------------------------------------------------------------------------
+// Provider
+// ---------------------------------------------------------------------------
+
+interface StreamPresenceProviderProps {
+  networkId: string;
+  streamId: string;
+  children: ReactNode;
+}
+
+export function StreamPresenceProvider({
+  networkId,
+  streamId,
+  children,
+}: StreamPresenceProviderProps) {
+  const channelId = `stream:${networkId}:${streamId}`;
+  const { presence, messages, sendMessage } = useChannel(channelId);
+  const currentUserId = useAuthStore((s) => s.user?.id);
+
+  // --- Online presence ---
+  const onlineHumanIds = useMemo(() => new Set(presence), [presence]);
+
+  // --- Composing state ---
+  const [composingUsers, setComposingUsers] = useState<ComposingUser[]>([]);
+  const composingMapRef = useRef(new Map<string, ComposingUser>());
+  const processedCountRef = useRef(0);
+
+  // Process new messages incrementally
+  useEffect(() => {
+    if (messages.length <= processedCountRef.current) return;
+
+    const newMessages = messages.slice(processedCountRef.current);
+    processedCountRef.current = messages.length;
+
+    let changed = false;
+    const map = composingMapRef.current;
+
+    for (const msg of newMessages) {
+      const payload = msg.payload as
+        | { type: string; mode?: string }
+        | undefined;
+      if (!payload?.type) continue;
+
+      // Skip own events
+      if (msg.humanId === currentUserId) continue;
+
+      if (payload.type === "composing_start" && payload.mode) {
+        map.set(msg.humanId, {
+          humanId: msg.humanId,
+          mode: payload.mode as ComposingMode,
+          lastSeen: Date.now(),
+        });
+        changed = true;
+      } else if (payload.type === "composing_stop") {
+        if (map.delete(msg.humanId)) changed = true;
+      }
+    }
+
+    if (changed) {
+      setComposingUsers(Array.from(map.values()));
+    }
+  }, [messages, currentUserId]);
+
+  // Also clear composing when a user leaves the channel
+  useEffect(() => {
+    const map = composingMapRef.current;
+    const onlineSet = new Set(presence);
+    let changed = false;
+
+    for (const humanId of map.keys()) {
+      if (!onlineSet.has(humanId)) {
+        map.delete(humanId);
+        changed = true;
+      }
+    }
+
+    if (changed) {
+      setComposingUsers(Array.from(map.values()));
+    }
+  }, [presence]);
+
+  // Cleanup stale composing entries
+  useEffect(() => {
+    const interval = setInterval(() => {
+      const map = composingMapRef.current;
+      const now = Date.now();
+      let changed = false;
+
+      for (const [humanId, entry] of map) {
+        if (now - entry.lastSeen > COMPOSING_TIMEOUT_MS) {
+          map.delete(humanId);
+          changed = true;
+        }
+      }
+
+      if (changed) {
+        setComposingUsers(Array.from(map.values()));
+      }
+    }, COMPOSING_CLEANUP_INTERVAL_MS);
+
+    return () => clearInterval(interval);
+  }, []);
+
+  // --- Composing broadcast ---
+  const heartbeatRef = useRef<ReturnType<typeof setInterval>>(undefined);
+
+  const startComposing = useCallback(
+    (mode: ComposingMode) => {
+      // Send immediately
+      sendMessage({ type: "composing_start", mode });
+
+      // Clear any existing heartbeat
+      clearInterval(heartbeatRef.current);
+
+      // Start heartbeat
+      heartbeatRef.current = setInterval(() => {
+        sendMessage({ type: "composing_start", mode });
+      }, COMPOSING_HEARTBEAT_MS);
+    },
+    [sendMessage],
+  );
+
+  const stopComposing = useCallback(() => {
+    clearInterval(heartbeatRef.current);
+    heartbeatRef.current = undefined;
+    sendMessage({ type: "composing_stop" });
+  }, [sendMessage]);
+
+  // Cleanup heartbeat on unmount
+  useEffect(() => {
+    return () => {
+      clearInterval(heartbeatRef.current);
+    };
+  }, []);
+
+  const value = useMemo<StreamPresenceContextValue>(
+    () => ({
+      onlineHumanIds,
+      composingUsers,
+      startComposing,
+      stopComposing,
+    }),
+    [onlineHumanIds, composingUsers, startComposing, stopComposing],
+  );
+
+  return (
+    <StreamPresenceContext.Provider value={value}>
+      {children}
+    </StreamPresenceContext.Provider>
+  );
+}
+
+// ---------------------------------------------------------------------------
+// Hooks
+// ---------------------------------------------------------------------------
+
+function useStreamPresenceContext() {
+  const ctx = useContext(StreamPresenceContext);
+  if (!ctx) {
+    throw new Error(
+      "useStreamPresence must be used within a StreamPresenceProvider",
+    );
+  }
+  return ctx;
+}
+
+export function useStreamPresence() {
+  const { onlineHumanIds } = useStreamPresenceContext();
+  return { onlineHumanIds };
+}
+
+export function useStreamComposing() {
+  const { composingUsers } = useStreamPresenceContext();
+  return { composingUsers };
+}
+
+export function useStreamComposingBroadcast() {
+  const { startComposing, stopComposing } = useStreamPresenceContext();
+  return { startComposing, stopComposing };
+}

js/src/features/particles/stream-view.tsx

@@ -4,12 +4,12 @@ import { useAuthStore } from "@/stores/auth-store";
 import { apiClient } from "@/api/client";
 import { type Particle, REACTION_EMOJIS } from "@/api/types";
 import { parseParticlePath, particlePath, toFirestoreDocPath, type ParticlePath } from "@/lib/particle-path";
-import { ComposeOverlay } from "@/features/compose/compose-overlay";
+import { ComposeOverlay, type ComposeStep } from "@/features/compose/compose-overlay";
 import { PlaybackPageIndicator } from "@/features/particles/playback-page-indicator";
 import { MediaParticleView, type MediaParticleHandle } from "@/features/particles/media-particle-view";
 import { TextParticleView } from "@/features/particles/text-particle-view";
 import { FallbackParticleView } from "@/features/particles/fallback-particle-view";
-import { Avatar, AvatarFallback, AvatarGroup } from "@/components/ui/avatar";
+import { Avatar, AvatarBadge, AvatarFallback, AvatarGroup } from "@/components/ui/avatar";
 import { VideoAudioToggle } from "@/components/video-audio-toggle";
 import { useMediaSettingsStore } from "@/stores/media-settings-store";
 import { KeybindingsOverlay, type KeybindingGroup } from "@/components/keybindings-overlay";
@@ -31,6 +31,8 @@ import { RelativeTimestamp } from "@/components/relative-timestamp";
 import { useStreamPlayback } from "@/hooks/use-stream-playback";
 import { usePrefetchAdjacentMedia } from "@/hooks/use-prefetch-adjacent-media";
 import { usePresencePositions } from "@/hooks/use-presence-positions";
+import { StreamPresenceProvider, useStreamPresence, useStreamComposing, useStreamComposingBroadcast, type ComposingMode } from "@/features/particles/stream-presence-context";
+import { ComposingIndicator } from "@/components/composing-indicator";
 import { cn, getInitials } from "@/lib/utils";
 import { useMount } from "react-use";
 
@@ -153,6 +155,16 @@ interface StreamViewProps {
 
 export function StreamView({ path, streamParticle }: StreamViewProps) {
   const { networkId } = parseParticlePath(path);
+
+  return (
+    <StreamPresenceProvider networkId={networkId} streamId={streamParticle.id}>
+      <StreamViewInner path={path} streamParticle={streamParticle} />
+    </StreamPresenceProvider>
+  );
+}
+
+function StreamViewInner({ path, streamParticle }: StreamViewProps) {
+  const { networkId } = parseParticlePath(path);
   const navigate = useNavigate();
 
   useMount(() => {
@@ -185,6 +197,11 @@ export function StreamView({ path, streamParticle }: StreamViewProps) {
     authedUser?.id,
   );
 
+  // --- Stream presence (realtime via pusher) ---
+  const { onlineHumanIds } = useStreamPresence();
+  const { composingUsers } = useStreamComposing();
+  const { startComposing, stopComposing } = useStreamComposingBroadcast();
+
   const mediaRef = useRef<MediaParticleHandle>(null);
 
   const handleToggleReaction = useCallback((emoji: string) => {
@@ -201,10 +218,30 @@ export function StreamView({ path, streamParticle }: StreamViewProps) {
   }, [authedUser, currentParticle]);
 
   const [composeActive, setComposeActive] = useState(false);
+  const [composeStep, setComposeStep] = useState<ComposeStep>("idle");
   const [progress, setProgress] = useState(0);
   const [fastPlayback, setFastPlayback] = useState(false);
   const [showKeybindings, setShowKeybindings] = useState(false);
 
+  // Broadcast composing state to other viewers
+  useEffect(() => {
+    const stepToMode: Record<string, ComposingMode | null> = {
+      idle: null,
+      submitting: null,
+      recording: "recording",
+      typing: "typing",
+      reviewing: "typing",
+      configuring: "typing",
+      picking: "screen",
+    };
+    const mode = stepToMode[composeStep] ?? null;
+    if (mode) {
+      startComposing(mode);
+    } else {
+      stopComposing();
+    }
+  }, [composeStep, startComposing, stopComposing]);
+
   // Show/hide chrome on mouse activity (YouTube-style)
   const [showControls, setShowControls] = useState(true);
   const idleTimerRef = useRef<ReturnType<typeof setTimeout>>(undefined);
@@ -436,10 +473,14 @@ export function StreamView({ path, streamParticle }: StreamViewProps) {
         </div>
       )}
 
+      {/* Composing indicator — left edge, always visible */}
+      <ComposingIndicator users={composingUsers} networkHumans={network?.humans} />
+
       <ComposeOverlay
         networkId={networkId}
         targetPath={path}
         onActiveChange={setComposeActive}
+        onStepChange={setComposeStep}
         disabled={streamParticle.status === "closed"}
       />
 
@@ -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<number, import("@/hooks/use-presence-positions").HumanPresence[]>;
+  onlineHumanIds: Set<string>;
   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 (
-    <span className="flex
-        items-center gap-1.5">
+    <span className="flex items-center gap-1.5">
       <Avatar size="sm">
         <AvatarFallback>
           {initials}
         </AvatarFallback>
+        {isOnline && <AvatarBadge className="bg-green-500" />}
       </Avatar>
       {prefix} - <RelativeTimestamp date={particle.created_at} />
     </span>

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<string[]>([]);
+  const [messages, setMessages] = useState<ChannelMessage[]>([]);
+
+  // 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 };
+}

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<ChannelEventType, Set<ChannelEventCallback>>
+  >();
+
+  // Active subscriptions for re-subscribe on reconnect
+  private activeSubscriptions = new Set<string>();
+
+  // Reconnection state
+  private reconnectDelay = INITIAL_RECONNECT_DELAY;
+  private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+  private shouldReconnect = false;
+
+  // Keep-alive ping
+  private pingTimer: ReturnType<typeof setInterval> | 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);
+    }
+  }
+}

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<PusherClient | null>(null);
+const PusherStateContext = createContext<ConnectionState>("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<PusherClient | null>(null);
+  const [connectionState, setConnectionState] =
+    useState<ConnectionState>("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 (
+    <PusherContext.Provider value={clientRef.current}>
+      <PusherStateContext.Provider value={connectionState}>
+        {children}
+      </PusherStateContext.Provider>
+    </PusherContext.Provider>
+  );
+}
+
+/**
+ * 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);
+}