diff --git a/cmd/pad/main.go b/cmd/pad/main.go index e2dc0cd9..92c1f39d 100644 --- a/cmd/pad/main.go +++ b/cmd/pad/main.go @@ -7,7 +7,7 @@ import ( "fmt" "io" "io/fs" - "log" + "log/slog" "net/http" "net/url" "os" @@ -30,7 +30,9 @@ import ( "regexp" "github.com/xarmian/pad/internal/email" + "github.com/redis/go-redis/v9" "github.com/xarmian/pad/internal/events" + "github.com/xarmian/pad/internal/logging" "github.com/xarmian/pad/internal/models" "github.com/xarmian/pad/internal/server" "github.com/xarmian/pad/internal/store" @@ -162,6 +164,17 @@ func serveCmd() *cobra.Command { RunE: func(cmd *cobra.Command, args []string) error { cfg := getConfig() + // Initialize structured logging + logLevel := os.Getenv("PAD_LOG_LEVEL") + if logLevel == "" { + logLevel = "info" + } + logFormat := os.Getenv("PAD_LOG_FORMAT") + if logFormat == "" { + logFormat = "text" + } + logging.Setup(logLevel, logFormat) + if cmd.Flags().Changed("host") { cfg.Host = host } @@ -169,23 +182,40 @@ func serveCmd() *cobra.Command { cfg.Port = port } - s, err := store.New(cfg.DBPath) - if err != nil { - return fmt.Errorf("open database: %w", err) + // Open database (SQLite default, PostgreSQL via PAD_DB_DRIVER) + var s *store.Store + var err error + dbDriver := os.Getenv("PAD_DB_DRIVER") + if dbDriver == "postgres" { + pgURL := os.Getenv("PAD_DATABASE_URL") + if pgURL == "" { + return fmt.Errorf("PAD_DATABASE_URL is required when PAD_DB_DRIVER=postgres") + } + s, err = store.NewPostgres(pgURL) + if err != nil { + return fmt.Errorf("open postgres: %w", err) + } + slog.Info("Database using PostgreSQL") + } else { + s, err = store.New(cfg.DBPath) + if err != nil { + return fmt.Errorf("open database: %w", err) + } + slog.Info("Database using SQLite", "path", cfg.DBPath) } defer s.Close() // Auto-upgrade: ensure all default collections exist in every workspace. // This is safe because SeedDefaultCollections skips collections that already exist. if workspaces, err := s.ListWorkspaces(); err == nil { - log.Printf("Auto-upgrade: checking %d workspace(s) for missing default collections", len(workspaces)) + slog.Info("auto-upgrade: checking workspaces for missing default collections", "count", len(workspaces)) for _, ws := range workspaces { if err := s.SeedDefaultCollections(ws.ID); err != nil { - log.Printf("Warning: failed to seed defaults for workspace %s: %v", ws.Slug, err) + slog.Warn("failed to seed defaults for workspace", "workspace", ws.Slug, "error", err) } } } else { - log.Printf("Warning: failed to list workspaces for auto-upgrade: %v", err) + slog.Warn("failed to list workspaces for auto-upgrade", "error", err) } srv := server.New(s) @@ -195,7 +225,23 @@ func serveCmd() *cobra.Command { srv.SetSecureCookies(cfg.SecureCookies) // Attach event bus for real-time SSE - srv.SetEventBus(events.New()) + var eventBus events.EventBus + if redisURL := os.Getenv("PAD_REDIS_URL"); redisURL != "" { + opts, err := redis.ParseURL(redisURL) + if err != nil { + return fmt.Errorf("invalid PAD_REDIS_URL: %w", err) + } + rc := redis.NewClient(opts) + if err := rc.Ping(context.Background()).Err(); err != nil { + return fmt.Errorf("redis connection failed: %w", err) + } + eventBus = events.NewRedisBus(rc) + slog.Info("Event bus using Redis pub/sub", "addr", opts.Addr, "db", opts.DB) + } else { + eventBus = events.New() + slog.Info("Event bus using in-memory (single instance)") + } + srv.SetEventBus(eventBus) // Attach webhook dispatcher for outgoing notifications srv.SetWebhookDispatcher(webhooks.NewDispatcher(s)) @@ -211,7 +257,7 @@ func serveCmd() *cobra.Command { fromName = "Pad" } srv.SetEmailSender(email.NewSender(cfg.MailerooAPIKey, fromAddr, fromName, cfg.BaseURL())) - log.Println("Email sending enabled via Maileroo (env)") + slog.Info("Email sending enabled via Maileroo (env)") } // Platform settings can override or provide email config srv.InitEmailFromSettings() @@ -221,11 +267,47 @@ func serveCmd() *cobra.Command { if err == nil { if entries, err := fs.ReadDir(webFS, "."); err == nil && len(entries) > 0 { srv.SetWebUI(webFS) - log.Println("Serving embedded web UI") + slog.Info("Serving embedded web UI") } } - return srv.ListenAndServe(cfg.Addr()) + // Graceful shutdown: listen for SIGINT/SIGTERM + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + // Start server in a goroutine + errCh := make(chan error, 1) + go func() { + errCh <- srv.ListenAndServe(cfg.Addr()) + }() + + // Wait for signal or server error + select { + case err := <-errCh: + // Server failed to start or crashed + return err + case <-ctx.Done(): + // Received shutdown signal + slog.Info("Shutting down server (30s grace period)...") + stop() // Reset signal handling so a second signal force-kills + + // Close event bus first — this terminates SSE handler + // goroutines so http.Server.Shutdown won't block on them. + if eventBus != nil { + eventBus.Close() + slog.Info("Event bus closed") + } + + shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + if err := srv.Shutdown(shutdownCtx); err != nil { + slog.Error("HTTP server shutdown error", "error", err) + } + + slog.Info("Server stopped") + return nil + } }, } diff --git a/go.mod b/go.mod index d6694577..8adf3cf0 100644 --- a/go.mod +++ b/go.mod @@ -3,26 +3,39 @@ module github.com/xarmian/pad go 1.25.0 require ( - github.com/BurntSushi/toml v1.6.0 // indirect + github.com/BurntSushi/toml v1.6.0 + github.com/fatih/color v1.19.0 + github.com/go-chi/chi/v5 v5.2.5 + github.com/go-chi/cors v1.2.2 + github.com/google/uuid v1.6.0 + github.com/jackc/pgx/v5 v5.9.1 + github.com/redis/go-redis/v9 v9.18.0 + github.com/sergi/go-diff v1.4.0 + github.com/spf13/cobra v1.10.2 + golang.org/x/crypto v0.49.0 + golang.org/x/term v0.41.0 + golang.org/x/time v0.15.0 + modernc.org/sqlite v1.47.0 +) + +require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/dustin/go-humanize v1.0.1 // indirect - github.com/fatih/color v1.19.0 // indirect - github.com/go-chi/chi/v5 v5.2.5 // indirect - github.com/go-chi/cors v1.2.2 // indirect - github.com/google/uuid v1.6.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/mattn/go-colorable v0.1.14 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect - github.com/sergi/go-diff v1.4.0 // indirect - github.com/spf13/cobra v1.10.2 // indirect github.com/spf13/pflag v1.0.9 // indirect - golang.org/x/crypto v0.49.0 // indirect + go.uber.org/atomic v1.11.0 // indirect + golang.org/x/sync v0.20.0 // indirect golang.org/x/sys v0.42.0 // indirect - golang.org/x/term v0.41.0 // indirect - golang.org/x/time v0.15.0 // indirect + golang.org/x/text v0.35.0 // indirect modernc.org/libc v1.70.0 // indirect modernc.org/mathutil v1.7.1 // indirect modernc.org/memory v1.11.0 // indirect - modernc.org/sqlite v1.47.0 // indirect ) diff --git a/go.sum b/go.sum index 6afcc6d5..e1387190 100644 --- a/go.sum +++ b/go.sum @@ -1,8 +1,17 @@ github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk= github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/fatih/color v1.19.0 h1:Zp3PiM21/9Ld6FzSKyL5c/BULoe/ONr9KlbYVOfG8+w= @@ -11,10 +20,24 @@ github.com/go-chi/chi/v5 v5.2.5 h1:Eg4myHZBjyvJmAFjFvWgrqDTXFyOzjj7YIm3L3mu6Ug= github.com/go-chi/chi/v5 v5.2.5/go.mod h1:X7Gx4mteadT3eDOMTsXzmI4/rwUpOwBHLpAfupzFJP0= github.com/go-chi/cors v1.2.2 h1:Jmey33TE+b+rB7fT8MUy1u0I4L+NARQlK6LhzKPSyQE= github.com/go-chi/cors v1.2.2/go.mod h1:sSbTewc+6wYHBBCW7ytsFSn836hqM7JxpglAy2Vzc58= +github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs= +github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.9.1 h1:uwrxJXBnx76nyISkhr33kQLlUqjv7et7b9FjCen/tdc= +github.com/jackc/pgx/v5 v5.9.1/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4= +github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= @@ -24,7 +47,10 @@ github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWE github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.18.0 h1:pMkxYPkEbMPwRdenAzUNyFNrDgHx9U+DrBabWNfSRQs= +github.com/redis/go-redis/v9 v9.18.0/go.mod h1:k3ufPphLU5YXwNTUcCRXGxUoF1fqxnhFQmscfkCoDA0= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= @@ -35,26 +61,65 @@ github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiT github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY= github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0= +github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4= golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA= +golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8= +golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.41.0 h1:QCgPso/Q3RTJx2Th4bDLqML4W6iJiaXFq2/ftQF13YU= golang.org/x/term v0.41.0/go.mod h1:3pfBgksrReYfZ5lvYM0kSO0LIkAl4Yl2bXOkKP7Ec2A= +golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8= +golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k= +golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +modernc.org/cc/v4 v4.27.1 h1:9W30zRlYrefrDV2JE2O8VDtJ1yPGownxciz5rrbQZis= +modernc.org/cc/v4 v4.27.1/go.mod h1:uVtb5OGqUKpoLWhqwNQo/8LwvoiEBLvZXIQ/SmO6mL0= +modernc.org/ccgo/v4 v4.32.0 h1:hjG66bI/kqIPX1b2yT6fr/jt+QedtP2fqojG2VrFuVw= +modernc.org/ccgo/v4 v4.32.0/go.mod h1:6F08EBCx5uQc38kMGl+0Nm0oWczoo1c7cgpzEry7Uc0= +modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM= +modernc.org/fileutil v1.4.0/go.mod h1:EqdKFDxiByqxLk8ozOxObDSfcVOv/54xDs/DUHdvCUU= +modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI= +modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito= +modernc.org/gc/v3 v3.1.2 h1:ZtDCnhonXSZexk/AYsegNRV1lJGgaNZJuKjJSWKyEqo= +modernc.org/gc/v3 v3.1.2/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY= +modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks= +modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI= modernc.org/libc v1.70.0 h1:U58NawXqXbgpZ/dcdS9kMshu08aiA6b7gusEusqzNkw= modernc.org/libc v1.70.0/go.mod h1:OVmxFGP1CI/Z4L3E0Q3Mf1PDE0BucwMkcXjjLntvHJo= modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI= modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/opt v0.1.4 h1:2kNGMRiUjrp4LcaPuLY2PzUfqM/w9N23quVwhKt5Qm8= +modernc.org/opt v0.1.4/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns= +modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w= +modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE= modernc.org/sqlite v1.47.0 h1:R1XyaNpoW4Et9yly+I2EeX7pBza/w+pmYee/0HJDyKk= modernc.org/sqlite v1.47.0/go.mod h1:hWjRO6Tj/5Ik8ieqxQybiEOUXy0NJFNp2tpvVpKlvig= +modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0= +modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A= +modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= +modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= diff --git a/internal/events/bus.go b/internal/events/bus.go index 8660b5c3..8aab1b87 100644 --- a/internal/events/bus.go +++ b/internal/events/bus.go @@ -1,7 +1,7 @@ package events import ( - "log" + "log/slog" "sync" "time" ) @@ -47,29 +47,49 @@ type Event struct { Timestamp int64 `json:"timestamp"` } +// EventBus is the interface for pub/sub event distribution. +// Implementations include MemoryBus (in-process) and RedisBus (cross-instance). +type EventBus interface { + // Subscribe registers a new subscriber for the given workspace. + // Returns a buffered channel that will receive events for that workspace. + Subscribe(workspaceID string) chan Event + + // Unsubscribe removes a subscriber and closes its channel. + Unsubscribe(ch chan Event) + + // Publish sends an event to all subscribers for the event's workspace. + Publish(event Event) + + // Close shuts down the event bus and cleans up resources. + Close() + + // SubscriberCount returns the number of active local subscribers. + SubscriberCount() int +} + // subscriber wraps a channel with its workspace filter. type subscriber struct { ch chan Event workspaceID string } -// Bus is an in-process pub/sub event bus that fans out events -// to all subscribers for a given workspace. -type Bus struct { +// MemoryBus is an in-process pub/sub event bus that fans out events +// to all subscribers for a given workspace. Suitable for single-instance deployments. +type MemoryBus struct { mu sync.RWMutex subscribers map[chan Event]*subscriber } -// New creates a new EventBus. -func New() *Bus { - return &Bus{ +// New creates a new in-memory EventBus. +func New() *MemoryBus { + return &MemoryBus{ subscribers: make(map[chan Event]*subscriber), } } // Subscribe registers a new subscriber for the given workspace. // Returns a buffered channel that will receive events for that workspace. -func (b *Bus) Subscribe(workspaceID string) chan Event { +func (b *MemoryBus) Subscribe(workspaceID string) chan Event { b.mu.Lock() defer b.mu.Unlock() @@ -82,7 +102,7 @@ func (b *Bus) Subscribe(workspaceID string) chan Event { } // Unsubscribe removes a subscriber and closes its channel. -func (b *Bus) Unsubscribe(ch chan Event) { +func (b *MemoryBus) Unsubscribe(ch chan Event) { b.mu.Lock() defer b.mu.Unlock() @@ -95,7 +115,7 @@ func (b *Bus) Unsubscribe(ch chan Event) { // Publish sends an event to all subscribers for the event's workspace. // Non-blocking: if a subscriber's channel is full, the event is dropped // and a warning is logged. -func (b *Bus) Publish(event Event) { +func (b *MemoryBus) Publish(event Event) { if event.Timestamp == 0 { event.Timestamp = time.Now().UnixMilli() } @@ -110,13 +130,25 @@ func (b *Bus) Publish(event Event) { select { case sub.ch <- event: default: - log.Printf("events: dropping event %s for slow subscriber (workspace=%s)", event.Type, event.WorkspaceID) + slog.Warn("dropping event for slow subscriber", "type", event.Type, "workspace", event.WorkspaceID) } } } +// Close shuts down the event bus by closing all subscriber channels. +// SSE handler goroutines will see the channel close and exit cleanly. +func (b *MemoryBus) Close() { + b.mu.Lock() + defer b.mu.Unlock() + + for ch := range b.subscribers { + delete(b.subscribers, ch) + close(ch) + } +} + // SubscriberCount returns the number of active subscribers (for testing/debugging). -func (b *Bus) SubscriberCount() int { +func (b *MemoryBus) SubscriberCount() int { b.mu.RLock() defer b.mu.RUnlock() return len(b.subscribers) diff --git a/internal/events/redis_bus.go b/internal/events/redis_bus.go new file mode 100644 index 00000000..552d4677 --- /dev/null +++ b/internal/events/redis_bus.go @@ -0,0 +1,209 @@ +package events + +import ( + "context" + "encoding/json" + "log/slog" + "sync" + "time" + + "github.com/redis/go-redis/v9" +) + +const ( + // redisChannelPrefix is prepended to workspace IDs for Redis pub/sub channels. + redisChannelPrefix = "pad:events:" + + // reconnectDelay is how long to wait before retrying a failed Redis subscription. + reconnectDelay = 2 * time.Second +) + +// RedisBus distributes events across multiple Pad instances via Redis pub/sub. +// Each instance subscribes to Redis channels for its locally-connected SSE clients, +// and publishes events to Redis so all instances see them. +type RedisBus struct { + client *redis.Client + + mu sync.RWMutex + subscribers map[chan Event]*subscriber + + // Track which workspace channels we're subscribed to in Redis, + // so we subscribe/unsubscribe as local SSE clients come and go. + wsCounts map[string]int // workspace → local subscriber count + wsSubs map[string]*redisSub // workspace → active Redis subscription + + ctx context.Context + cancel context.CancelFunc +} + +// redisSub tracks an active Redis subscription for a workspace. +type redisSub struct { + pubsub *redis.PubSub + cancel context.CancelFunc +} + +// NewRedisBus creates a new Redis-backed EventBus. +// The provided redis.Client should already be configured and connected. +func NewRedisBus(client *redis.Client) *RedisBus { + ctx, cancel := context.WithCancel(context.Background()) + return &RedisBus{ + client: client, + subscribers: make(map[chan Event]*subscriber), + wsCounts: make(map[string]int), + wsSubs: make(map[string]*redisSub), + ctx: ctx, + cancel: cancel, + } +} + +// Subscribe registers a local subscriber for the given workspace. +// Starts a Redis subscription for the workspace if this is the first local subscriber. +func (b *RedisBus) Subscribe(workspaceID string) chan Event { + b.mu.Lock() + defer b.mu.Unlock() + + ch := make(chan Event, 64) + b.subscribers[ch] = &subscriber{ + ch: ch, + workspaceID: workspaceID, + } + + b.wsCounts[workspaceID]++ + if b.wsCounts[workspaceID] == 1 { + // First local subscriber for this workspace — subscribe to Redis channel + b.startRedisSubscription(workspaceID) + } + + return ch +} + +// Unsubscribe removes a local subscriber and closes its channel. +// Cancels the Redis subscription if this was the last local subscriber for the workspace. +func (b *RedisBus) Unsubscribe(ch chan Event) { + b.mu.Lock() + defer b.mu.Unlock() + + sub, ok := b.subscribers[ch] + if !ok { + return + } + + delete(b.subscribers, ch) + close(ch) + + wsID := sub.workspaceID + b.wsCounts[wsID]-- + if b.wsCounts[wsID] <= 0 { + delete(b.wsCounts, wsID) + b.stopRedisSubscription(wsID) + } +} + +// Publish sends an event to Redis, which distributes it to all instances. +func (b *RedisBus) Publish(event Event) { + if event.Timestamp == 0 { + event.Timestamp = time.Now().UnixMilli() + } + + data, err := json.Marshal(event) + if err != nil { + slog.Error("failed to marshal event for Redis", "error", err) + return + } + + channel := redisChannelPrefix + event.WorkspaceID + if err := b.client.Publish(b.ctx, channel, data).Err(); err != nil { + slog.Error("failed to publish event to Redis", "channel", channel, "error", err) + } +} + +// Close shuts down all Redis subscriptions and closes local subscriber channels. +func (b *RedisBus) Close() { + b.cancel() // signal all subscription goroutines to stop + + b.mu.Lock() + defer b.mu.Unlock() + + for wsID, sub := range b.wsSubs { + sub.cancel() + sub.pubsub.Close() + delete(b.wsSubs, wsID) + } + + for ch := range b.subscribers { + delete(b.subscribers, ch) + close(ch) + } +} + +// SubscriberCount returns the number of active local subscribers. +func (b *RedisBus) SubscriberCount() int { + b.mu.RLock() + defer b.mu.RUnlock() + return len(b.subscribers) +} + +// startRedisSubscription begins listening on a Redis channel for a workspace. +// Must be called with b.mu held. +func (b *RedisBus) startRedisSubscription(workspaceID string) { + channel := redisChannelPrefix + workspaceID + pubsub := b.client.Subscribe(b.ctx, channel) + + subCtx, subCancel := context.WithCancel(b.ctx) + b.wsSubs[workspaceID] = &redisSub{ + pubsub: pubsub, + cancel: subCancel, + } + + go b.receiveMessages(subCtx, pubsub, workspaceID) +} + +// stopRedisSubscription cancels and cleans up the Redis subscription for a workspace. +// Must be called with b.mu held. +func (b *RedisBus) stopRedisSubscription(workspaceID string) { + sub, ok := b.wsSubs[workspaceID] + if !ok { + return + } + sub.cancel() + sub.pubsub.Close() + delete(b.wsSubs, workspaceID) +} + +// receiveMessages reads from a Redis pub/sub channel and fans out to local subscribers. +func (b *RedisBus) receiveMessages(ctx context.Context, pubsub *redis.PubSub, workspaceID string) { + ch := pubsub.Channel() + for { + select { + case <-ctx.Done(): + return + case msg, ok := <-ch: + if !ok { + return + } + var event Event + if err := json.Unmarshal([]byte(msg.Payload), &event); err != nil { + slog.Error("failed to unmarshal Redis event", "channel", msg.Channel, "error", err) + continue + } + b.fanOutLocally(event) + } + } +} + +// fanOutLocally distributes an event to all local subscribers for the event's workspace. +func (b *RedisBus) fanOutLocally(event Event) { + b.mu.RLock() + defer b.mu.RUnlock() + + for _, sub := range b.subscribers { + if sub.workspaceID != event.WorkspaceID { + continue + } + select { + case sub.ch <- event: + default: + slog.Warn("dropping event for slow subscriber", "type", event.Type, "workspace", event.WorkspaceID) + } + } +} diff --git a/internal/logging/logging.go b/internal/logging/logging.go new file mode 100644 index 00000000..50cf6882 --- /dev/null +++ b/internal/logging/logging.go @@ -0,0 +1,59 @@ +// Package logging provides structured logging using log/slog. +// +// Usage: +// +// logging.Setup("info", "json") // call once at startup +// slog.Info("something happened", "key", value) +// +// All application code should use the slog package directly after Setup has +// been called — it configures the default slog logger. +package logging + +import ( + "io" + "log/slog" + "os" + "strings" +) + +// Setup configures the default slog logger. +// +// - level: "debug", "info", "warn", "error" (default "info") +// - format: "json" or "text" (default "text") +// +// After calling Setup, use slog.Info / slog.Error / etc. everywhere. +func Setup(level, format string) { + SetupWriter(os.Stderr, level, format) +} + +// SetupWriter is like Setup but writes to w instead of stderr (useful for tests). +func SetupWriter(w io.Writer, level, format string) { + lvl := parseLevel(level) + + opts := &slog.HandlerOptions{ + Level: lvl, + } + + var handler slog.Handler + switch strings.ToLower(format) { + case "json": + handler = slog.NewJSONHandler(w, opts) + default: + handler = slog.NewTextHandler(w, opts) + } + + slog.SetDefault(slog.New(handler)) +} + +func parseLevel(s string) slog.Level { + switch strings.ToLower(s) { + case "debug": + return slog.LevelDebug + case "warn", "warning": + return slog.LevelWarn + case "error": + return slog.LevelError + default: + return slog.LevelInfo + } +} diff --git a/internal/server/handlers_auth.go b/internal/server/handlers_auth.go index ba920b82..f7ea280d 100644 --- a/internal/server/handlers_auth.go +++ b/internal/server/handlers_auth.go @@ -2,7 +2,7 @@ package server import ( "context" - "log" + "log/slog" "net" "net/http" "regexp" @@ -537,7 +537,7 @@ func (s *Server) handleForgotPassword(w http.ResponseWriter, r *http.Request) { // Generate reset token token, err := s.store.CreatePasswordReset(user.ID) if err != nil { - log.Printf("Failed to create password reset: %v", err) + slog.Error("failed to create password reset", "error", err) writeJSON(w, http.StatusOK, okResponse) return } @@ -547,11 +547,11 @@ func (s *Server) handleForgotPassword(w http.ResponseWriter, r *http.Request) { resetURL := s.baseURL + "/reset-password/" + token go func() { if err := s.email.SendPasswordReset(context.Background(), user.Email, user.Name, resetURL); err != nil { - log.Printf("Failed to send password reset email: %v", err) + slog.Error("failed to send password reset email", "error", err) } }() } else { - log.Printf("Password reset token generated (email not configured). Use pad auth reset-password to manage.") + slog.Info("password reset token generated (email not configured)") } writeJSON(w, http.StatusOK, okResponse) @@ -598,7 +598,7 @@ func (s *Server) handleResetPassword(w http.ResponseWriter, r *http.Request) { // Invalidate all existing sessions (force logout everywhere) if err := s.store.DeleteUserSessions(user.ID); err != nil { - log.Printf("Failed to invalidate sessions after password reset: %v", err) + slog.Error("failed to invalidate sessions after password reset", "error", err) } // Create a fresh session so the user is logged in diff --git a/internal/server/handlers_events.go b/internal/server/handlers_events.go index 4dbcad21..3a1152e2 100644 --- a/internal/server/handlers_events.go +++ b/internal/server/handlers_events.go @@ -3,7 +3,7 @@ package server import ( "encoding/json" "fmt" - "log" + "log/slog" "net/http" "time" ) @@ -89,7 +89,7 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { func writeSSEEvent(w http.ResponseWriter, eventType string, data interface{}) { jsonData, err := json.Marshal(data) if err != nil { - log.Printf("events: error marshaling SSE event: %v", err) + slog.Error("failed to marshal SSE event", "error", err) return } fmt.Fprintf(w, "event: %s\ndata: %s\n\n", eventType, jsonData) diff --git a/internal/server/handlers_items.go b/internal/server/handlers_items.go index 6253644b..1e5b29f1 100644 --- a/internal/server/handlers_items.go +++ b/internal/server/handlers_items.go @@ -4,7 +4,7 @@ import ( "database/sql" "encoding/json" "fmt" - "log" + "log/slog" "net/http" "strconv" "strings" @@ -369,7 +369,7 @@ func (s *Server) handleUpdateItem(w http.ResponseWriter, r *http.Request) { commentInput.Source = source comment, cerr := s.store.CreateComment(workspaceID, updated.ID, commentInput) if cerr != nil { - log.Printf("WARNING: failed to create comment on item update %s: %v", updated.ID, cerr) + slog.Warn("failed to create comment on item update", "item_id", updated.ID, "error", cerr) } if cerr == nil && comment != nil { s.publishCommentEvent(events.CommentCreated, workspaceID, updated.ID, comment.ID, updated.Title, updated.CollectionSlug, actor, source) diff --git a/internal/server/handlers_members.go b/internal/server/handlers_members.go index 0fb6e2e8..9b5209c0 100644 --- a/internal/server/handlers_members.go +++ b/internal/server/handlers_members.go @@ -2,7 +2,7 @@ package server import ( "context" - "log" + "log/slog" "net/http" "github.com/go-chi/chi/v5" @@ -149,7 +149,7 @@ func (s *Server) handleInviteMember(w http.ResponseWriter, r *http.Request) { wsName = ws.Name } if err := s.email.SendInvitation(context.Background(), inv.Email, inviterName, wsName, joinURL); err != nil { - log.Printf("Failed to send invitation email: %v", err) + slog.Error("failed to send invitation email", "error", err) } }() } diff --git a/internal/server/handlers_workspaces.go b/internal/server/handlers_workspaces.go index 92b6624a..d495dce2 100644 --- a/internal/server/handlers_workspaces.go +++ b/internal/server/handlers_workspaces.go @@ -72,6 +72,26 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, resp) } +// handleHealthLive is a lightweight liveness probe — always returns 200 if the +// process is running. Kubernetes uses this to decide whether to restart the pod. +func (s *Server) handleHealthLive(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) +} + +// handleHealthReady is a readiness probe — returns 200 only when the service +// can accept traffic (DB connection healthy). Kubernetes uses this to decide +// whether to route traffic to the pod. +func (s *Server) handleHealthReady(w http.ResponseWriter, r *http.Request) { + if err := s.store.Ping(); err != nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{ + "status": "not ready", + "error": "database unavailable", + }) + return + } + writeJSON(w, http.StatusOK, map[string]string{"status": "ready"}) +} + func (s *Server) handleListTemplates(w http.ResponseWriter, r *http.Request) { type templateInfo struct { Name string `json:"name"` diff --git a/internal/server/middleware_auth.go b/internal/server/middleware_auth.go index fcd7dc23..75209d9c 100644 --- a/internal/server/middleware_auth.go +++ b/internal/server/middleware_auth.go @@ -130,7 +130,7 @@ func (s *Server) RequireAuth(next http.Handler) http.Handler { path := r.URL.Path // Auth endpoints are always exempt - if strings.HasPrefix(path, "/api/v1/auth/") || path == "/api/v1/health" { + if strings.HasPrefix(path, "/api/v1/auth/") || path == "/api/v1/health" || strings.HasPrefix(path, "/api/v1/health/") { next.ServeHTTP(w, r) return } diff --git a/internal/server/middleware_logging.go b/internal/server/middleware_logging.go new file mode 100644 index 00000000..efafbff5 --- /dev/null +++ b/internal/server/middleware_logging.go @@ -0,0 +1,48 @@ +package server + +import ( + "log/slog" + "net/http" + "time" + + chimiddleware "github.com/go-chi/chi/v5/middleware" +) + +// StructuredLogger is a chi-compatible request logger that writes structured +// log entries via slog. It replaces chi's default Logger middleware. +func StructuredLogger(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + start := time.Now() + ww := chimiddleware.NewWrapResponseWriter(w, r.ProtoMajor) + + next.ServeHTTP(ww, r) + + duration := time.Since(start) + status := ww.Status() + + level := slog.LevelInfo + if status >= 500 { + level = slog.LevelError + } else if status >= 400 { + level = slog.LevelWarn + } + + attrs := []slog.Attr{ + slog.String("method", r.Method), + slog.String("path", r.URL.Path), + slog.Int("status", status), + slog.Duration("duration", duration), + slog.Int("bytes", ww.BytesWritten()), + } + + if reqID := chimiddleware.GetReqID(r.Context()); reqID != "" { + attrs = append(attrs, slog.String("request_id", reqID)) + } + + if r.URL.RawQuery != "" { + attrs = append(attrs, slog.String("query", r.URL.RawQuery)) + } + + slog.LogAttrs(r.Context(), level, "http request", attrs...) + }) +} diff --git a/internal/server/middleware_security.go b/internal/server/middleware_security.go index 941e6c55..1734ce94 100644 --- a/internal/server/middleware_security.go +++ b/internal/server/middleware_security.go @@ -24,9 +24,12 @@ func SecurityHeaders(next http.Handler) http.Handler { // Restrict browser features the app doesn't need h.Set("Permissions-Policy", "camera=(), microphone=(), geolocation=()") - // CSP: allow self-sourced scripts/styles, plus inline styles for Svelte + // CSP: allow self-sourced content, inline styles for Svelte component scoping, + // and inline scripts for SvelteKit's module bootstrap/hydration. + // Without 'unsafe-inline' on script-src, SvelteKit's generated inline