Compare commits
4 Commits
main
...
55b9ca427c
| Author | SHA1 | Date | |
|---|---|---|---|
| 55b9ca427c | |||
| 8bb6ce7dc9 | |||
| a0ec37e4ed | |||
| e863bc7f10 |
@@ -0,0 +1,97 @@
|
||||
package bootstrap
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"git.fossy.my.id/bagas/tunnel-please-controller/internal/config"
|
||||
"git.fossy.my.id/bagas/tunnel-please-controller/internal/db/sqlc/repository"
|
||||
"git.fossy.my.id/bagas/tunnel-please-controller/internal/server"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/lestrrat-go/httprc/v3"
|
||||
"github.com/lestrrat-go/jwx/v3/jwk"
|
||||
)
|
||||
|
||||
type bootstrap struct {
|
||||
config config.Config
|
||||
errChan chan error
|
||||
signalChan chan os.Signal
|
||||
}
|
||||
|
||||
type Bootstrap interface {
|
||||
Run() error
|
||||
}
|
||||
|
||||
func New(config config.Config) Bootstrap {
|
||||
errChan := make(chan error, 5)
|
||||
signalChan := make(chan os.Signal, 1)
|
||||
return &bootstrap{
|
||||
config: config,
|
||||
errChan: errChan,
|
||||
signalChan: signalChan,
|
||||
}
|
||||
}
|
||||
|
||||
func (b *bootstrap) Run() error {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
signal.Notify(b.signalChan, os.Interrupt, syscall.SIGTERM)
|
||||
|
||||
dbURL := b.config.DatabaseURL()
|
||||
repo, err := startDatabase(ctx, dbURL)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
client := httprc.NewClient()
|
||||
jwkCache, err := jwk.NewCache(ctx, client)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s := server.New(repo, jwkCache, b.config)
|
||||
|
||||
b.errChan = make(chan error, 2)
|
||||
|
||||
go func() {
|
||||
if err = s.StartAPI(ctx, b.config.ApiAddr()); err != nil && !errors.Is(err, context.Canceled) {
|
||||
b.errChan <- err
|
||||
}
|
||||
}()
|
||||
|
||||
go func() {
|
||||
if err = s.StartController(ctx, b.config.ControllerAddr()); err != nil && !errors.Is(err, context.Canceled) {
|
||||
b.errChan <- err
|
||||
}
|
||||
}()
|
||||
|
||||
select {
|
||||
case err = <-b.errChan:
|
||||
return fmt.Errorf("service error: %w", err)
|
||||
case sig := <-b.signalChan:
|
||||
log.Printf("Received signal %s, initiating graceful shutdown", sig)
|
||||
cancel()
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func startDatabase(ctx context.Context, dbURL string) (*repository.Queries, error) {
|
||||
connect, err := pgx.Connect(ctx, dbURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
defer func(connect *pgx.Conn, ctx context.Context) {
|
||||
err = connect.Close(ctx)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}(connect, ctx)
|
||||
|
||||
return repository.New(connect), nil
|
||||
}
|
||||
@@ -1,8 +1,81 @@
|
||||
package config
|
||||
|
||||
import "os"
|
||||
import (
|
||||
"crypto/rand"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"log"
|
||||
"os"
|
||||
|
||||
func Getenv(key, defaultValue string) string {
|
||||
"github.com/joho/godotenv"
|
||||
)
|
||||
|
||||
type config struct {
|
||||
databaseUrl string
|
||||
controllerAddr string
|
||||
apiAddr string
|
||||
authToken string
|
||||
jwksUrl string
|
||||
}
|
||||
|
||||
type Config interface {
|
||||
DatabaseURL() string
|
||||
ControllerAddr() string
|
||||
ApiAddr() string
|
||||
AuthToken() string
|
||||
JwksURL() string
|
||||
}
|
||||
|
||||
func (c *config) DatabaseURL() string { return c.databaseUrl }
|
||||
func (c *config) ControllerAddr() string { return c.controllerAddr }
|
||||
func (c *config) ApiAddr() string { return c.apiAddr }
|
||||
func (c *config) AuthToken() string { return c.authToken }
|
||||
func (c *config) JwksURL() string { return c.jwksUrl }
|
||||
|
||||
func MustLoad() (Config, error) {
|
||||
if err := loadEnvFile(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
cfg, err := parse()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
func loadEnvFile() error {
|
||||
if _, err := os.Stat(".env"); err == nil {
|
||||
return godotenv.Load(".env")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func parse() (*config, error) {
|
||||
databaseUrl := getenv("DATABASE_URL", "")
|
||||
if databaseUrl == "" {
|
||||
return nil, errors.New("DATABASE_URL environment variable not set")
|
||||
}
|
||||
controllerAddr := getenv("CONTROLLER_ADDR", ":8080")
|
||||
apiAddr := getenv("API_ADDR", ":8081")
|
||||
authToken := getenv("AUTH_TOKEN", "")
|
||||
if authToken == "" {
|
||||
authToken = generateAuthToken()
|
||||
log.Printf("No AUTH_TOKEN provided. Generated token: %s", authToken)
|
||||
}
|
||||
jwksUrl := getenv("JWKS_URL", "")
|
||||
|
||||
return &config{
|
||||
databaseUrl: databaseUrl,
|
||||
controllerAddr: controllerAddr,
|
||||
apiAddr: apiAddr,
|
||||
authToken: authToken,
|
||||
jwksUrl: jwksUrl,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func getenv(key, defaultValue string) string {
|
||||
val := os.Getenv(key)
|
||||
if val == "" {
|
||||
val = defaultValue
|
||||
@@ -10,3 +83,11 @@ func Getenv(key, defaultValue string) string {
|
||||
|
||||
return val
|
||||
}
|
||||
|
||||
func generateAuthToken() string {
|
||||
buf := make([]byte, 32)
|
||||
if _, err := rand.Read(buf); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return base64.StdEncoding.EncodeToString(buf)
|
||||
}
|
||||
|
||||
@@ -12,8 +12,8 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.fossy.my.id/bagas/tunnel-please-controller/db/sqlc/repository"
|
||||
"git.fossy.my.id/bagas/tunnel-please-controller/internal/config"
|
||||
"git.fossy.my.id/bagas/tunnel-please-controller/internal/db/sqlc/repository"
|
||||
proto "git.fossy.my.id/bagas/tunnel-please-grpc/gen"
|
||||
"github.com/lestrrat-go/jwx/v3/jwk"
|
||||
"github.com/lestrrat-go/jwx/v3/jwt"
|
||||
@@ -42,19 +42,19 @@ type Server struct {
|
||||
Database *repository.Queries
|
||||
Subscribers map[string]*Subscriber
|
||||
mu *sync.RWMutex
|
||||
authToken string
|
||||
jwkCache *jwk.Cache
|
||||
config config.Config
|
||||
proto.UnimplementedEventServiceServer
|
||||
proto.UnimplementedUserServiceServer
|
||||
}
|
||||
|
||||
func New(database *repository.Queries, authToken string, jwkCache *jwk.Cache) *Server {
|
||||
func New(database *repository.Queries, jwkCache *jwk.Cache, config config.Config) *Server {
|
||||
return &Server{
|
||||
Database: database,
|
||||
Subscribers: make(map[string]*Subscriber),
|
||||
mu: new(sync.RWMutex),
|
||||
authToken: authToken,
|
||||
jwkCache: jwkCache,
|
||||
config: config,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -94,7 +94,7 @@ func (s *Server) Subscribe(event grpc.BidiStreamingServer[proto.Node, proto.Even
|
||||
return status.Error(codes.InvalidArgument, "missing identity")
|
||||
}
|
||||
token := payload.AuthEvent.AuthToken
|
||||
if token != s.authToken {
|
||||
if token != s.config.AuthToken() {
|
||||
return status.Error(codes.Unauthenticated, "invalid auth token")
|
||||
}
|
||||
|
||||
@@ -332,7 +332,7 @@ func (s *Server) StartAPI(ctx context.Context, Addr string) error {
|
||||
WriteTimeout: 15 * time.Second,
|
||||
IdleTimeout: 60 * time.Second,
|
||||
}
|
||||
jwkURL := config.Getenv("JWKS_URL", "")
|
||||
jwkURL := s.config.JwksURL()
|
||||
if jwkURL != "" {
|
||||
registerCtx, cancel := context.WithTimeout(ctx, jwkRegisterTimeout)
|
||||
defer cancel()
|
||||
@@ -629,6 +629,7 @@ func (s *Server) StartAPI(ctx context.Context, Addr string) error {
|
||||
|
||||
errCh := make(chan error, 1)
|
||||
go func() {
|
||||
log.Printf("Listening api on %s", httpServer.Addr)
|
||||
errCh <- httpServer.ListenAndServe()
|
||||
}()
|
||||
|
||||
@@ -697,18 +698,19 @@ func (s *Server) StartController(ctx context.Context, Addr string) error {
|
||||
|
||||
serveErr := make(chan error, 1)
|
||||
go func() {
|
||||
log.Printf("Listening controller on %s", s.config.ControllerAddr())
|
||||
serveErr <- grpcServer.Serve(listener)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
grpcServer.GracefulStop()
|
||||
err := <-serveErr
|
||||
err = <-serveErr
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return ctx.Err()
|
||||
case err := <-serveErr:
|
||||
case err = <-serveErr:
|
||||
return err
|
||||
}
|
||||
}
|
||||
@@ -1,104 +1,25 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"git.fossy.my.id/bagas/tunnel-please-controller/db/sqlc/repository"
|
||||
"git.fossy.my.id/bagas/tunnel-please-controller/server"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/joho/godotenv"
|
||||
"github.com/lestrrat-go/httprc/v3"
|
||||
"github.com/lestrrat-go/jwx/v3/jwk"
|
||||
"git.fossy.my.id/bagas/tunnel-please-controller/internal/bootstrap"
|
||||
"git.fossy.my.id/bagas/tunnel-please-controller/internal/config"
|
||||
)
|
||||
|
||||
func main() {
|
||||
if _, err := os.Stat(".env"); err == nil {
|
||||
if err = godotenv.Load(".env"); err != nil {
|
||||
log.Printf("Warning: Failed to load .env file: %s", err)
|
||||
}
|
||||
}
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
|
||||
log.SetOutput(os.Stdout)
|
||||
log.SetFlags(log.LstdFlags | log.Lshortfile)
|
||||
|
||||
dbURL := os.Getenv("DATABASE_URL")
|
||||
if dbURL == "" {
|
||||
log.Fatal("DATABASE_URL is required")
|
||||
}
|
||||
|
||||
controllerAddr := getenv("CONTROLLER_ADDR", ":8080")
|
||||
apiAddr := getenv("API_ADDR", ":8081")
|
||||
authToken := getenv("AUTH_TOKEN", "")
|
||||
if authToken == "" {
|
||||
authToken = generateAuthToken()
|
||||
log.Printf("No AUTH_TOKEN provided. Generated token: %s", authToken)
|
||||
}
|
||||
|
||||
connect, err := pgx.Connect(ctx, dbURL)
|
||||
conf, err := config.MustLoad()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
log.Fatalf("Config load error: %v", err)
|
||||
}
|
||||
defer func(connect *pgx.Conn, ctx context.Context) {
|
||||
err = connect.Close(ctx)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}(connect, ctx)
|
||||
|
||||
repo := repository.New(connect)
|
||||
client := httprc.NewClient()
|
||||
jwkCache, err := jwk.NewCache(ctx, client)
|
||||
if err != nil {
|
||||
log.Printf("failed to initialize jwk cache : %s", err)
|
||||
}
|
||||
s := server.New(repo, authToken, jwkCache)
|
||||
boot := bootstrap.New(conf)
|
||||
|
||||
log.Printf("Listening controller on %s", controllerAddr)
|
||||
log.Printf("Listening api on %s", apiAddr)
|
||||
|
||||
errCh := make(chan error, 2)
|
||||
|
||||
go func() {
|
||||
if err = s.StartAPI(ctx, apiAddr); err != nil && !errors.Is(err, context.Canceled) {
|
||||
errCh <- err
|
||||
}
|
||||
}()
|
||||
|
||||
go func() {
|
||||
if err = s.StartController(ctx, controllerAddr); err != nil && !errors.Is(err, context.Canceled) {
|
||||
errCh <- err
|
||||
}
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
log.Printf("shutting down: %v", ctx.Err())
|
||||
case err = <-errCh:
|
||||
log.Fatalf("server error: %v", err)
|
||||
if err = boot.Run(); err != nil {
|
||||
log.Fatalf("Application error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func getenv(key, def string) string {
|
||||
if val := os.Getenv(key); val != "" {
|
||||
return val
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
func generateAuthToken() string {
|
||||
buf := make([]byte, 32)
|
||||
if _, err := rand.Read(buf); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return base64.StdEncoding.EncodeToString(buf)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user