Merge pull request #13 from led0nk/abstraction

Abstraction & renaming pkgs & chan-opt

ark-overseer.spec

@@ -27,9 +27,9 @@ ark-overseer is a steam observation tool to track players
 %goprep
 %autosetup 
 
-#%generate_buildrequires
+#%%generate_buildrequires
 
-#%go_generate_buildrequires
+#%%go_generate_buildrequires
 
 %build
 go build -v -buildmode pie -mod vendor -o %{gobuilddir}/bin/%{name} cmd/api/main.go

cmd/api/main.go

@@ -3,123 +3,169 @@ package main
 import (
 	"context"
 	"flag"
+	"fmt"
 	"log/slog"
 	"os"
 	"os/signal"
+	"path/filepath"
 	"sync"
 	"syscall"
 
-	"github.com/led0nk/ark-overseer/internal"
-	blist "github.com/led0nk/ark-overseer/internal/blacklist"
-	"github.com/led0nk/ark-overseer/internal/jsondb"
-	"github.com/led0nk/ark-overseer/internal/notifier"
-	v1 "github.com/led0nk/ark-overseer/internal/server"
+	"github.com/led0nk/ark-overseer/internal/blacklist"
+	"github.com/led0nk/ark-overseer/internal/interfaces"
+	"github.com/led0nk/ark-overseer/internal/observer"
+	"github.com/led0nk/ark-overseer/internal/server"
 	"github.com/led0nk/ark-overseer/internal/services"
-	"github.com/led0nk/ark-overseer/observer"
+	"github.com/led0nk/ark-overseer/internal/storage"
+	"github.com/led0nk/ark-overseer/internal/storagewrapper"
 	"github.com/led0nk/ark-overseer/pkg/config"
 	"github.com/led0nk/ark-overseer/pkg/events"
 )
 
 func main() {
 
 	var (
-		addr   = flag.String("addr", "localhost:8080", "server port")
-		db     = flag.String("db", "testdata", "path to the database")
-		blpath = flag.String("blacklist", "testdata", "path to the blacklist")
-		//grpcaddr    = flag.String("grpcaddr", "", "grpc address, e.g. localhost:4317")
+		addr        = flag.String("addr", "localhost:8080", "server port")
+		dbpath      = flag.String("db", "testdata", "path to the database")
+		blpath      = flag.String("blacklist", "testdata", "path to the blacklist")
 		domain      = flag.String("domain", "127.0.0.1", "given domain for cookies/mail")
 		logLevelStr = flag.String("loglevel", "INFO", "define the level for logs")
 		configPath  = flag.String("config", "config", "path to config-file")
-		sStore      internal.ServerStore
-		obs         internal.Observer
-		blacklist   internal.Blacklist
-		cfg         config.Configuration
 		logLevel    slog.Level
 		shutdownWg  sync.WaitGroup
 		initWg      sync.WaitGroup
+		listenerWg  sync.WaitGroup
 	)
 	flag.Parse()
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	err := logLevel.UnmarshalText([]byte(*logLevelStr))
-	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: logLevel}))
+	logger, err := setupLogger(logLevelStr, logLevel)
 	if err != nil {
-		logger.ErrorContext(ctx, "error parsing loglevel", "loglevel", *logLevelStr, "error", err)
+		logger.ErrorContext(ctx, "failed to setup logger", "error", err)
 		os.Exit(1)
 	}
-	slog.SetDefault(logger)
 
 	logger.Info("server address", "addr", *addr)
 
-	sStore, err = jsondb.NewServerStorage(ctx, *db+"/cluster.json")
+	eventManager := events.NewEventManager()
+	serviceManager := services.NewServiceManager(eventManager, &initWg)
+
+	database, blackList, obs, cfg, err := initServices(ctx, dbpath, blpath, configPath, eventManager)
 	if err != nil {
-		logger.ErrorContext(ctx, "failed to create new cluster", "error", err)
+		logger.ErrorContext(ctx, "failed to initialize services", "error", err)
 		os.Exit(1)
 	}
 
-	em := events.NewEventManager()
-	sm := services.NewServiceManager(em, &initWg)
+	listenerWg.Add(2)
+	startEventListeners(ctx, eventManager, &listenerWg, &shutdownWg, serviceManager, obs)
+	listenerWg.Wait()
 
-	notify := notifier.NewNotifier(sStore, em)
-	sStore = notify
+	initWg.Add(2)
+	go func(config.Configuration) {
+		defer initWg.Done()
+		eventManager.Publish(events.EventMessage{Type: "init.services", Payload: cfg})
+	}(cfg)
+	initWg.Wait()
+
+	initWg.Add(1)
+	go func() {
+		defer initWg.Done()
+		eventManager.Publish(events.EventMessage{Type: "init"})
+	}()
 
-	cfg, err = config.NewConfiguration(*configPath+"/config.yaml", em)
+	srv := server.NewServer(*addr, *domain, database, blackList, cfg)
+	startHTTPServer(ctx, srv, &shutdownWg)
+
+	handleShutdown(ctx, cancel, &initWg, &shutdownWg, database)
+}
+
+func initServices(
+	ctx context.Context,
+	dbpath *string,
+	blpath *string,
+	configPath *string,
+	eventManager *events.EventManager,
+) (
+	interfaces.Database,
+	interfaces.Blacklist,
+	observer.Overseer,
+	config.Configuration,
+	error) {
+	var (
+		database  interfaces.Database
+		blackList interfaces.Blacklist
+		obs       observer.Overseer
+		cfg       config.Configuration
+	)
+
+	database, err := storage.NewServerStorage(ctx, filepath.Join(*dbpath, "cluster.json"))
 	if err != nil {
-		logger.Error("failed to create new config", "error", err)
+		return nil, nil, nil, nil, fmt.Errorf("failed to create new server storage: %w", err)
 	}
 
-	blacklist, err = blist.NewBlacklist(*blpath + "/blacklist.json")
+	storageWrapper := storagewrapper.NewStorageWrapper(database, eventManager)
+	database = storageWrapper
+
+	blackList, err = blacklist.NewBlacklist(filepath.Join(*blpath, "blacklist.json"))
 	if err != nil {
-		logger.ErrorContext(ctx, "failed to create blacklist", "error", err)
-		os.Exit(1)
+		return nil, nil, nil, nil, fmt.Errorf("failed to create blacklist: %w", err)
 	}
 
-	obs, err = observer.NewObserver(ctx, sStore, blacklist, em)
+	obs, err = observer.NewObserver(ctx, database, blackList, eventManager)
 	if err != nil {
-		logger.ErrorContext(ctx, "failed to create endpoint storage", "error", err)
-		os.Exit(1)
+		return nil, nil, nil, nil, fmt.Errorf("failed to create observer: %w", err)
+	}
+
+	cfg, err = config.NewConfiguration(filepath.Join(*configPath, "config.yaml"), eventManager)
+	if err != nil {
+		return nil, nil, nil, nil, fmt.Errorf("failed to create config: %w", err)
 	}
 
+	return database, blackList, obs, cfg, nil
+}
+
+func startHTTPServer(
+	ctx context.Context,
+	server *server.Server,
+	shutdownWg *sync.WaitGroup,
+) {
 	shutdownWg.Add(1)
 	go func() {
 		defer shutdownWg.Done()
-		em.StartListening(ctx, sm, "serviceManager")
-	}()
-
-	//NOTE: Wait group for initialization, 2 because the first 1 is the publish for init.services and the 2nd is the handled event
-	initWg.Add(2)
-	go func() {
-		defer initWg.Done()
-		em.Publish(events.EventMessage{Type: "init.services", Payload: cfg})
+		server.ServeHTTP(ctx)
 	}()
+}
 
+func startEventListeners(
+	ctx context.Context,
+	em *events.EventManager,
+	listenerWg, shutdownWg *sync.WaitGroup,
+	sm *services.ServiceManager,
+	obs observer.Overseer,
+) {
 	shutdownWg.Add(1)
-	go func(cfg config.Configuration) {
-		defer shutdownWg.Done()
-		em.StartListening(ctx, obs, "observer")
-	}(cfg)
-
-	initWg.Wait()
-	initWg.Add(1)
 	go func() {
-		defer initWg.Done()
-		em.Publish(events.EventMessage{Type: "init"})
+		defer shutdownWg.Done()
+		em.StartListening(ctx, sm, "serviceManager", func() { listenerWg.Done() })
+		fmt.Println("after listening")
 	}()
 
-	server := v1.NewServer(*addr, *domain, logger, sStore, blacklist, cfg)
-
 	shutdownWg.Add(1)
 	go func() {
 		defer shutdownWg.Done()
-		err := server.ServeHTTP(ctx)
-		if err != nil {
-			logger.ErrorContext(ctx, "failed to shutdown http server", "error", err)
-			return
-		}
+		em.StartListening(ctx, obs, "observer", func() { listenerWg.Done() })
 	}()
+}
 
+func handleShutdown(
+	ctx context.Context,
+	cancel context.CancelFunc,
+	initWg, shutdownWg *sync.WaitGroup,
+	database interfaces.Database,
+) {
+	logger := slog.Default()
 	sigCh := make(chan os.Signal, 1)
 	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
 
@@ -134,10 +180,22 @@ func main() {
 	shutdownWg.Add(1)
 
 	logger.InfoContext(ctx, "finally saving server storage", "info", "shutdown")
-	err = sStore.Save()
+	err := database.Save()
 	if err != nil {
 		logger.ErrorContext(ctx, "failed to save server storage", "error", err)
+		return
 	}
 
 	logger.InfoContext(ctx, "application stopped gracefully", "info", "shutdown")
 }
+
+func setupLogger(logLevelStr *string, logLevel slog.Level) (*slog.Logger, error) {
+	err := logLevel.UnmarshalText([]byte(*logLevelStr))
+	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: logLevel}))
+	if err != nil {
+		return nil, fmt.Errorf("error parsing logLevel: %w", err)
+	}
+	slog.SetDefault(logger)
+
+	return logger, nil
+}

go.mod

@@ -7,17 +7,20 @@ require (
 	github.com/a-h/templ v0.2.680
 	github.com/bwmarrin/discordgo v0.28.1
 	github.com/google/uuid v1.6.0
-	github.com/joho/godotenv v1.5.1
 	github.com/samber/slog-http v1.3.1
+	github.com/stretchr/testify v1.8.4
 	gopkg.in/yaml.v2 v2.4.0
 )
 
 require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/gorilla/websocket v1.5.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	go.opentelemetry.io/otel v1.19.0 // indirect
 	go.opentelemetry.io/otel/trace v1.19.0 // indirect
 	golang.org/x/crypto v0.22.0 // indirect
 	golang.org/x/net v0.24.0 // indirect
 	golang.org/x/sys v0.19.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -14,8 +14,6 @@ github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
 github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY=
-github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
-github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

internal/blacklist/blacklist.go

@@ -1,4 +1,4 @@
-package blist
+package blacklist
 
 import (
 	"context"
@@ -22,13 +22,13 @@ func NewBlacklist(filename string) (*Blacklist, error) {
 		filename:  filename,
 		blacklist: make(map[uuid.UUID]*model.BlacklistPlayers),
 	}
-	if err := blacklist.readJSON(); err != nil {
+	if err := blacklist.load(); err != nil {
 		return nil, err
 	}
 	return blacklist, nil
 }
 
-func (b *Blacklist) writeJSON() error {
+func (b *Blacklist) save() error {
 	as_json, err := json.MarshalIndent(b.blacklist, "", "\t")
 	if err != nil {
 		return err
@@ -41,13 +41,13 @@ func (b *Blacklist) writeJSON() error {
 	return nil
 }
 
-func (b *Blacklist) readJSON() error {
+func (b *Blacklist) load() error {
 	if _, err := os.Stat(b.filename); os.IsNotExist(err) {
 		err = os.MkdirAll(filepath.Dir(b.filename), 0644)
 		if err != nil {
 			return err
 		}
-		err = b.writeJSON()
+		err = b.save()
 		if err != nil {
 			return err
 		}
@@ -68,7 +68,7 @@ func (b *Blacklist) Create(ctx context.Context, player *model.BlacklistPlayers)
 	}
 
 	b.blacklist[player.ID] = player
-	if err := b.writeJSON(); err != nil {
+	if err := b.save(); err != nil {
 		return nil, err
 	}
 	return player, nil
@@ -79,7 +79,7 @@ func (b *Blacklist) Delete(ctx context.Context, id uuid.UUID) error {
 	defer b.mu.Unlock()
 
 	delete(b.blacklist, id)
-	if err := b.writeJSON(); err != nil {
+	if err := b.save(); err != nil {
 		return err
 	}
 	return nil