package main import ( "context" "fmt" "log/slog" "net/http" "os" "cloud.google.com/go/storage" pbaero "github.com/flowy-live/llink/genproto/aero" "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/depot" "github.com/flowy-live/llink/internal/handler" "github.com/flowy-live/llink/internal/human" "github.com/flowy-live/llink/internal/middleware" "github.com/flowy-live/llink/internal/network" "github.com/flowy-live/llink/internal/particle" "github.com/flowy-live/llink/internal/utils" "github.com/redis/go-redis/v9" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" ) const ( // FIX: Use separate redis instance. We start with higher number because use this same instance in helios. REDIS_DATABASE_FOR_AUTH int = 4 ) func redisForAuth() *redis.Client { return internal.ConnectAndTestRedis(REDIS_DATABASE_FOR_AUTH) } func main() { port := utils.MustGetEnv("PORT") gcsBucket := utils.MustGetEnv("GCS_BUCKET") // Initialize database db.Init() defer db.Cleanup() // Initialize Redis for auth redisClient := redisForAuth() // Initialize GCS client ctx := context.Background() storageClient, err := storage.NewClient(ctx) if err != nil { slog.Error("failed to create GCS client", "error", err) os.Exit(1) } defer storageClient.Close() aeroAddr := utils.MustGetEnv("AERO_ADDR") if aeroAddr == "" { slog.Error("must provide AERO_ADDR") os.Exit(1) } aeroServer, err := grpc.NewClient(aeroAddr, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { slog.Error("connection to aero server invalid", "error", err) os.Exit(1) } defer aeroServer.Close() aeroSvc := pbaero.NewPrimaryClient(aeroServer) // Initialize services authSvc := auth.NewAuthService(redisClient, aeroSvc) humanSvc := human.NewService(db.Pool()) networkSvc := network.NewService(db.Pool()) particleSvc := particle.NewService(db.Pool(), networkSvc) depotSvc := depot.NewService(db.Pool(), storageClient, depot.Config{ GoogleServiceAccountEmail: utils.MustGetEnv("GOOGLE_SERVICE_ACCOUNT_EMAIL"), BucketName: gcsBucket, }) // Initialize handler h := handler.NewHandler(authSvc, humanSvc, networkSvc, particleSvc, depotSvc) // Helper to wrap handlers with auth middleware withAuth := func(hf http.HandlerFunc) http.Handler { return middleware.Auth(authSvc)(http.HandlerFunc(hf)) } mux := http.NewServeMux() // ========================================================================== // Public routes (no auth) // ========================================================================== mux.HandleFunc("GET /", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) }) mux.HandleFunc("GET /health", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) }) mux.HandleFunc("POST /auth/request-code", h.RequestSignInCode) mux.HandleFunc("POST /auth/sign-in", h.SignIn) // ========================================================================== // Protected routes (auth required) // ========================================================================== // Auth mux.Handle("POST /auth/sign-out", withAuth(h.SignOut)) mux.Handle("GET /auth/me", withAuth(h.GetCurrentHuman)) // Bootstrap startup data mux.Handle("GET /startup", withAuth(h.StartupData)) // Networks mux.Handle("POST /networks", withAuth(h.CreateNetwork)) mux.Handle("GET /networks", withAuth(h.ListNetworks)) mux.Handle("GET /networks/{id}", withAuth(h.GetNetwork)) mux.Handle("POST /networks/{id}/members", withAuth(h.AddMembersToNetwork)) // TODO: what about members who are part of streams visibility within this network? mux.Handle("DELETE /networks/{id}/members/{email}", withAuth(h.RemoveMemberFromNetwork)) mux.Handle("PUT /networks/{id}/capacity", withAuth(h.SetOpenStreamCapacity)) // Streams mux.Handle("POST /networks/{network_id}/streams", withAuth(h.CreateStream)) mux.Handle("GET /streams/{id}", withAuth(h.GetStream)) mux.Handle("PATCH /streams/{id}", withAuth(h.UpdateStream)) mux.Handle("POST /streams/{id}/particles", withAuth(h.CreateStreamParticle)) mux.Handle("POST /streams/{id}/open", withAuth(h.OpenStream)) mux.Handle("POST /streams/{id}/close", withAuth(h.CloseStream)) mux.Handle("POST /streams/{id}/members", withAuth(h.AddMembers)) mux.Handle("DELETE /streams/{id}/members", withAuth(h.RemoveMembers)) // Particles mux.Handle("GET /networks/{network_id}/particles", withAuth(h.ListParticles)) mux.Handle("GET /particles/{id}", withAuth(h.GetParticle)) mux.Handle("PATCH /particles/{id}", withAuth(h.UpdateParticle)) mux.Handle("DELETE /particles/{id}", withAuth(h.DeleteParticle)) mux.Handle("POST /particles/{id}/seen", withAuth(h.MarkSeen)) mux.Handle("POST /particles/{id}/ack", withAuth(h.AckParticle)) mux.Handle("GET /particles/{id}/download", withAuth(h.DownloadParticle)) mux.Handle("POST /particles/seen", withAuth(h.MarkSeenBatch)) // Depot mux.Handle("POST /depot/upload", withAuth(h.PrepareUpload)) mux.Handle("POST /depot/objects/{id}/confirm", withAuth(h.ConfirmUpload)) addr := fmt.Sprintf("0.0.0.0:%s", port) slog.Info("running server", "addr", addr) if err := http.ListenAndServe(addr, mux); err != nil { slog.Error("server failed", "error", err) os.Exit(1) } }