setup infra for pusher service
This commit is contained in:
talksik
@@ -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"]
|
||||
+1
-1
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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",
|
||||
}
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
package pusher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"sync"
|
||||
|
||||
"nhooyr.io/websocket"
|
||||
)
|
||||
|
||||
const sendBufferSize = 256
|
||||
|
||||
// Conn wraps a WebSocket connection with identity and a send buffer.
|
||||
type Conn struct {
|
||||
id string
|
||||
humanID string
|
||||
ws *websocket.Conn
|
||||
send chan []byte
|
||||
once sync.Once // ensures close logic runs once
|
||||
}
|
||||
|
||||
func newConn(id, humanID string, ws *websocket.Conn) *Conn {
|
||||
return &Conn{
|
||||
id: id,
|
||||
humanID: humanID,
|
||||
ws: ws,
|
||||
send: make(chan []byte, sendBufferSize),
|
||||
}
|
||||
}
|
||||
|
||||
// ReadPump reads messages from the WebSocket and forwards them to the hub.
|
||||
// It blocks until the connection is closed or the context is cancelled.
|
||||
func (c *Conn) ReadPump(ctx context.Context, hub *Hub) {
|
||||
defer hub.disconnect(c)
|
||||
|
||||
for {
|
||||
_, data, err := c.ws.Read(ctx)
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
slog.Debug("websocket read error", "connId", c.id, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
var msg ClientMessage
|
||||
if err := json.Unmarshal(data, &msg); err != nil {
|
||||
c.sendError("invalid message format")
|
||||
continue
|
||||
}
|
||||
|
||||
switch msg.Type {
|
||||
case TypeSubscribe:
|
||||
if msg.Channel == "" {
|
||||
c.sendError("channel is required")
|
||||
continue
|
||||
}
|
||||
hub.subscribeCh <- &subscribeRequest{conn: c, channelID: msg.Channel}
|
||||
case TypeUnsubscribe:
|
||||
if msg.Channel == "" {
|
||||
c.sendError("channel is required")
|
||||
continue
|
||||
}
|
||||
hub.unsubscribeCh <- &unsubscribeRequest{conn: c, channelID: msg.Channel}
|
||||
case TypeMessage:
|
||||
if msg.Channel == "" {
|
||||
c.sendError("channel is required")
|
||||
continue
|
||||
}
|
||||
hub.broadcastCh <- &broadcastRequest{conn: c, channelID: msg.Channel, payload: msg.Payload}
|
||||
default:
|
||||
c.sendError("unknown message type: " + msg.Type)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WritePump drains the send buffer and writes messages to the WebSocket.
|
||||
func (c *Conn) WritePump(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case data, ok := <-c.send:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if err := c.ws.Write(ctx, websocket.MessageText, data); err != nil {
|
||||
slog.Debug("websocket write error", "connId", c.id, "error", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Send enqueues a ServerMessage to be written to the WebSocket.
|
||||
// If the send buffer is full, the connection is closed (slow client).
|
||||
func (c *Conn) Send(msg ServerMessage) {
|
||||
data, err := json.Marshal(msg)
|
||||
if err != nil {
|
||||
slog.Error("failed to marshal server message", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case c.send <- data:
|
||||
default:
|
||||
slog.Warn("slow client, closing connection", "connId", c.id, "humanId", c.humanID)
|
||||
c.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// Close closes the WebSocket connection and the send channel.
|
||||
func (c *Conn) Close() {
|
||||
c.once.Do(func() {
|
||||
c.ws.Close(websocket.StatusNormalClosure, "closing")
|
||||
close(c.send)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *Conn) sendError(msg string) {
|
||||
c.Send(ServerMessage{Type: TypeError, Message: msg})
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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"`
|
||||
}
|
||||
@@ -0,0 +1,104 @@
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: pusher
|
||||
spec:
|
||||
selector:
|
||||
matchLabels:
|
||||
app: pusher
|
||||
replicas: 1
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app: pusher
|
||||
spec:
|
||||
serviceAccountName: default-service-account
|
||||
nodeSelector:
|
||||
cloud.google.com/gke-spot: "true"
|
||||
terminationGracePeriodSeconds: 15
|
||||
containers:
|
||||
- name: pusher
|
||||
image: "pusher"
|
||||
ports:
|
||||
- containerPort: 8080
|
||||
- containerPort: 50051
|
||||
resources:
|
||||
requests:
|
||||
memory: "52Mi"
|
||||
cpu: 50m
|
||||
limits:
|
||||
memory: "52Mi"
|
||||
cpu: 50m
|
||||
env:
|
||||
- name: "PORT"
|
||||
value: "8080"
|
||||
- name: "GRPC_PORT"
|
||||
value: "50051"
|
||||
- name: "REDIS_HOST"
|
||||
value: "10.138.57.187"
|
||||
- name: "LLINK_POSTGRES_CONNECTION_URL"
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: shared-secrets
|
||||
key: LLINK_POSTGRES_CONNECTION_URL
|
||||
|
||||
---
|
||||
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
name: pusher
|
||||
spec:
|
||||
selector:
|
||||
app: pusher
|
||||
ports:
|
||||
- name: http
|
||||
port: 8080
|
||||
targetPort: 8080
|
||||
protocol: TCP
|
||||
- name: grpc
|
||||
port: 50051
|
||||
targetPort: 50051
|
||||
protocol: TCP
|
||||
|
||||
---
|
||||
|
||||
kind: HTTPRoute
|
||||
apiVersion: gateway.networking.k8s.io/v1beta1
|
||||
metadata:
|
||||
name: pusher
|
||||
spec:
|
||||
parentRefs:
|
||||
- kind: Gateway
|
||||
name: external-gateway
|
||||
hostnames:
|
||||
- pusher.dev.flowy.live
|
||||
rules:
|
||||
- backendRefs:
|
||||
- name: pusher
|
||||
port: 8080
|
||||
|
||||
---
|
||||
|
||||
apiVersion: networking.gke.io/v1
|
||||
kind: HealthCheckPolicy
|
||||
metadata:
|
||||
name: pusher-service-health-check
|
||||
spec:
|
||||
default:
|
||||
checkIntervalSec: 15
|
||||
timeoutSec: 15
|
||||
healthyThreshold: 1
|
||||
unhealthyThreshold: 2
|
||||
logConfig:
|
||||
enabled: true
|
||||
config:
|
||||
type: HTTP
|
||||
httpHealthCheck:
|
||||
portSpecification: USE_FIXED_PORT
|
||||
port: 8080
|
||||
requestPath: /health
|
||||
targetRef:
|
||||
group: ""
|
||||
kind: Service
|
||||
name: pusher
|
||||
@@ -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
|
||||
+1
-1
Submodule go/protocol updated: d323ba3d2b...9e3e1e0e6a
Executable
BIN
Binary file not shown.
@@ -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: {}
|
||||
|
||||
Reference in New Issue
Block a user