Arjun Patel
0f73d685c2
Easier for testing, and omits need for full dep injection for pusher service, where we passed in nil for deps dangerously.
113 lines
3.1 KiB
Go
113 lines
3.1 KiB
Go
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"
|
|
)
|
|
|
|
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(db.RedisDBAuth)
|
|
|
|
// Redis for pusher state (presence hashes, pub/sub)
|
|
pusherRedis := internal.ConnectAndTestRedis(db.RedisDBPusher)
|
|
|
|
// Services
|
|
sessionReader := auth.NewSessionReader(authRedis) // pusher only validates sessions
|
|
networkReader := network.NewReader(db.Pool()) // pusher only checks membership
|
|
|
|
// 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)
|
|
// Context for graceful shutdown
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
|
|
authorizer := pusher.NewAuthorizer(networkReader)
|
|
hub := pusher.NewHub(bridge, authorizer)
|
|
bridge.SetHub(hub)
|
|
server := pusher.NewServer(ctx, hub, bridge, sessionReader)
|
|
|
|
// 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")
|
|
}
|