Basic server app with WS connection.

server/internal/application/logger.go

@@ -3,12 +3,37 @@ package application
 import (
 	"log/slog"
 	"os"
+	"strings"
+)
+
+const (
+	defaultLogLevel slog.Level = slog.LevelInfo
+
+	logLevelEnvVar = "BUNKERD_LOG_LEVEL"
 )
 
 func (a *Application) initializeLogger() {
+	logLevel := defaultLogLevel
+
+	logLevelAsString, found := os.LookupEnv(logLevelEnvVar)
+	if found {
+		switch strings.ToLower(logLevelAsString) {
+		case "debug":
+			logLevel = slog.LevelDebug
+		case "info":
+			logLevel = slog.LevelInfo
+		case "warn":
+			logLevel = slog.LevelWarn
+		case "error":
+			logLevel = slog.LevelError
+		}
+	}
+
+	slog.Warn("Setting log level.", "level", logLevel.String())
+
 	a.baseLogger = slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
 		AddSource: true,
-		Level:     slog.LevelDebug,
+		Level:     logLevel,
 	}))
 
 	a.appLogger = a.baseLogger.With("module", "application")

server/internal/services/core/database.go

@@ -0,0 +1,44 @@
+package core
+
+import (
+	"context"
+	"errors"
+	"io/fs"
+
+	"github.com/jmoiron/sqlx"
+)
+
+// ServiceNameDatabase is a name for database service.
+const ServiceNameDatabase = "core/database"
+
+var (
+	// ErrDatabase indicates that error appeared somewhere in database service.
+	ErrDatabase = errors.New("database service")
+	// ErrDatabaseIsInvalid indicates that database service implementation is invalid.
+	ErrDatabaseIsInvalid = errors.New("database service implementation is invalid")
+)
+
+// Database is an interface for database service.
+type Database interface {
+	// Exec is a proxy for ExecContext from sqlx.
+	Exec(ctx context.Context, query string, params ...interface{}) error
+	// Get is a proxy for GetContext from sqlx.
+	Get(ctx context.Context, target interface{}, query string, params ...interface{}) error
+	// NamedExec is a proxy for NamedExecContext from sqlx.
+	NamedExec(ctx context.Context, query string, param interface{}) error
+	// RegisterMigrations registers migrations for applying from other services. Migrations should reside
+	// in "migrations" directory in passed filesystem.
+	RegisterMigrations(moduleName string, fs fs.FS) error
+	// Select is a proxy for SelectContext from sqlx.
+	Select(ctx context.Context, target interface{}, query string, params ...interface{}) error
+	// Transaction is a wrapper for transactions processing which wraps sqlx's transactions.
+	Transaction(ctx context.Context) (DatabaseTransaction, error)
+}
+
+// DatabaseTransaction is an interface for database transactions controllers implementations.
+type DatabaseTransaction interface {
+	Apply(steps ...TransactionFunc) error
+}
+
+// TransactionFunc is a function that is used in transactions to mangle with data.
+type TransactionFunc func(*sqlx.Tx) error

server/internal/services/core/database/connection.go

@@ -0,0 +1,83 @@
+package database
+
+import (
+	"errors"
+	"fmt"
+	"net/url"
+	"os"
+	"strconv"
+	"strings"
+
+	"github.com/jmoiron/sqlx"
+	// postgresql driver.
+	_ "github.com/jackc/pgx/v5/stdlib"
+)
+
+const (
+	databaseDSNEnvVar            = "BUNKERD_DATABASE_DSN"
+	databaseMaxIdleConnsEnvVar   = "BUNKERD_DATABASE_MAX_IDLE_CONNS"
+	databaseMaxOpenedConnsEnvVar = "BUNKERD_DATABASE_MAX_OPENED_CONNS"
+)
+
+var (
+	errDSNInvalid            = errors.New("BUNKERD_DATABASE_DSN environment variable is empty or invalid")
+	errNoMaxIdleConns        = errors.New("no BUNKERD_DATABASE_MAX_IDLE_CONNS defined")
+	errNoMaxOpenedConns      = errors.New("no BUNKERD_DATABASE_MAX_OPENED_CONNS defined")
+	errPostgresOnlySupported = errors.New("only PostgreSQL database is currently supported")
+)
+
+func (d *database) initializeConnection() error {
+	// Getting database DSN from environment as well as other required settings.
+	dsn, found := os.LookupEnv(databaseDSNEnvVar)
+	if !found {
+		return fmt.Errorf("initialize connection: getting database DSN: %w", errDSNInvalid)
+	}
+
+	maxOpenedConnsRaw, found := os.LookupEnv(databaseMaxOpenedConnsEnvVar)
+	if !found {
+		return fmt.Errorf("initialize connection: getting maximum number of opened conections: %w", errNoMaxOpenedConns)
+	}
+
+	maxOpenedConns, err := strconv.ParseInt(maxOpenedConnsRaw, 10, 64)
+	if err != nil {
+		return fmt.Errorf("initialize connection: parsing maximum number of opened conections: %w", err)
+	}
+
+	maxIdleConnsRaw, found := os.LookupEnv(databaseMaxIdleConnsEnvVar)
+	if !found {
+		return fmt.Errorf("initialize connection: getting maximum number of idle conections: %w", errNoMaxIdleConns)
+	}
+
+	maxIdleConns, err := strconv.ParseInt(maxIdleConnsRaw, 10, 64)
+	if err != nil {
+		return fmt.Errorf("initialize connection: parsing maximum number of opened conections: %w", err)
+	}
+
+	// While database/sql (and sqlx) supports all possible DSN formats, we will force user to use DSN in form
+	// "proto://user:passowrd@host:port/dbname" as it is easier to parse.
+	if _, err := url.Parse(dsn); err != nil {
+		return fmt.Errorf("initialize connection: validate DSN: %w", err)
+	}
+
+	// Currently we're support only postgresql, but this may change in future.
+	if !strings.HasPrefix(dsn, "postgres://") {
+		return fmt.Errorf("initialize connection: validate DSN: %w", errPostgresOnlySupported)
+	}
+
+	proto := strings.Split(dsn, ":")[0]
+	if proto == "postgres" {
+		proto = "pgx"
+	}
+
+	db, err := sqlx.Open(proto, dsn)
+	if err != nil {
+		return fmt.Errorf("initialize connection: open database: %w", err)
+	}
+
+	d.db = db
+
+	d.db.SetMaxOpenConns(int(maxOpenedConns))
+	d.db.SetMaxIdleConns(int(maxIdleConns))
+
+	return nil
+}

server/internal/services/core/database/database.go

@@ -0,0 +1,73 @@
+package database
+
+import (
+	"fmt"
+	"io/fs"
+	"log/slog"
+
+	"bunker/server/internal/application"
+	"bunker/server/internal/services/core"
+
+	"github.com/jmoiron/sqlx"
+)
+
+var _ = core.Database(&database{})
+
+type database struct {
+	app        *application.Application
+	db         *sqlx.DB
+	logger     *slog.Logger
+	migrations map[string]fs.FS
+	version    int64
+}
+
+// Initialize initializes service.
+func Initialize(app *application.Application) error {
+	db := &database{
+		app: app,
+	}
+
+	if err := app.RegisterService(db); err != nil {
+		return fmt.Errorf("%w: %w", core.ErrDatabase, err)
+	}
+
+	return nil
+}
+
+func (d *database) Configure() error {
+	if err := d.initializeConnection(); err != nil {
+		return fmt.Errorf("configure: %w", err)
+	}
+
+	return nil
+}
+
+func (d *database) ConnectDependencies() error {
+	return nil
+}
+
+func (d *database) Initialize() error {
+	d.logger = d.app.NewLogger("service", core.ServiceNameDatabase)
+
+	d.logger.Info("Initializing...")
+
+	d.migrations = make(map[string]fs.FS, 0)
+
+	return nil
+}
+
+func (d *database) Name() string {
+	return core.ServiceNameDatabase
+}
+
+func (d *database) LaunchStartupTasks() error {
+	if err := d.applyMigrations(); err != nil {
+		return fmt.Errorf("launch startup tasks: %w", err)
+	}
+
+	return nil
+}
+
+func (d *database) Shutdown() error {
+	return nil
+}

server/internal/services/core/database/migrations.go

@@ -0,0 +1,82 @@
+package database
+
+import (
+	"errors"
+	"fmt"
+	"io/fs"
+	"log/slog"
+	"sort"
+	"strings"
+
+	"bunker/commons"
+	"bunker/server/internal/services/core"
+
+	"github.com/pressly/goose/v3"
+)
+
+var errMigrationsAlreadyRegistered = errors.New("migrations already registered")
+
+func (d *database) applyMigrations() error {
+	d.logger.Info("Migrating database...")
+
+	modules := make([]string, 0)
+
+	for module := range d.migrations {
+		modules = append(modules, module)
+	}
+
+	sort.Strings(modules)
+
+	_ = goose.SetDialect(string(goose.DialectPostgres))
+
+	gooseLogger := commons.NewGooseLogger(d.logger)
+	goose.SetLogger(gooseLogger)
+
+	for _, module := range modules {
+		d.logger.Info("Migrating database for module...", "module", module)
+
+		goose.SetBaseFS(d.migrations[module])
+		goose.SetTableName(strings.ReplaceAll(module, "/", "_") + "_migrations")
+
+		if err := goose.Up(d.db.DB, "migrations"); err != nil {
+			return fmt.Errorf("%w: applying migrations for module '%s': %w", core.ErrDatabase, module, err)
+		}
+
+		moduleDBVersion, err := goose.GetDBVersion(d.db.DB)
+		if err != nil {
+			return fmt.Errorf("%w: get database version for module '%s': %w", core.ErrDatabase, module, err)
+		}
+
+		d.version += moduleDBVersion
+
+		d.logger.Info(
+			"Database for module migrated to latest version",
+			"module", module,
+			"module_db_version", moduleDBVersion,
+			"db_version", d.version,
+		)
+	}
+
+	d.logger.Info("Database migrated.", "version", d.version)
+
+	return nil
+}
+
+func (d *database) RegisterMigrations(moduleName string, fSys fs.FS) error {
+	slog.Debug("Registering migrations for service.", "service", moduleName)
+
+	if _, found := d.migrations[moduleName]; found {
+		return fmt.Errorf(
+			"%w: RegisterMigrations: module '%s': %w",
+			core.ErrDatabase,
+			moduleName,
+			errMigrationsAlreadyRegistered,
+		)
+	}
+
+	d.migrations[moduleName] = fSys
+
+	slog.Debug("Migrations for service successfully registered.", "service", moduleName)
+
+	return nil
+}

server/internal/services/core/database/queries.go

@@ -0,0 +1,69 @@
+package database
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"bunker/server/internal/services/core"
+)
+
+// Exec is a proxy for ExecContext from sqlx.
+func (d *database) Exec(ctx context.Context, query string, params ...interface{}) error {
+	if strings.Contains(query, "?") {
+		query = d.db.Rebind(query)
+	}
+
+	d.logger.Debug("Executing query.", "query", query, "params", fmt.Sprintf("%+v", params))
+
+	if _, err := d.db.ExecContext(ctx, query, params...); err != nil {
+		return fmt.Errorf("%w: failed to Exec(): %w", core.ErrDatabase, err)
+	}
+
+	return nil
+}
+
+// Get is a proxy for GetContext from sqlx.
+func (d *database) Get(ctx context.Context, target interface{}, query string, params ...interface{}) error {
+	if strings.Contains(query, "?") {
+		query = d.db.Rebind(query)
+	}
+
+	d.logger.Debug("Getting single data from database with query.", "query", query, "params", fmt.Sprintf("%+v", params))
+
+	if err := d.db.GetContext(ctx, target, query, params...); err != nil {
+		return fmt.Errorf("%w: failed to Get(): %w", core.ErrDatabase, err)
+	}
+
+	return nil
+}
+
+// NamedExec is a proxy for NamedExecContext from sqlx.
+func (d *database) NamedExec(ctx context.Context, query string, param interface{}) error {
+	if strings.Contains(query, "?") {
+		query = d.db.Rebind(query)
+	}
+
+	d.logger.Debug("Executing named query.", "query", query, "params", fmt.Sprintf("%+v", param))
+
+	if _, err := d.db.NamedExecContext(ctx, query, param); err != nil {
+		return fmt.Errorf("%w: failed to NamedExec(): %w", core.ErrDatabase, err)
+	}
+
+	return nil
+}
+
+// Select is a proxy for SelectContext from sqlx.
+func (d *database) Select(ctx context.Context, target interface{}, query string, params ...interface{}) error {
+	if strings.Contains(query, "?") {
+		query = d.db.Rebind(query)
+	}
+
+	d.logger.Debug("Selecting from database with query.", "query", query, "params", fmt.Sprintf("%+v", params))
+
+	if err := d.db.SelectContext(ctx, target, query, params...); err != nil {
+		return fmt.Errorf("%w: failed to Select(): %w", core.ErrDatabase, err)
+	}
+
+	return nil
+}

server/internal/services/core/database/transaction.go

@@ -0,0 +1,81 @@
+package database
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	"bunker/server/internal/services/core"
+
+	"github.com/jmoiron/sqlx"
+)
+
+type transaction struct {
+	transaction *sqlx.Tx
+	logger      *slog.Logger
+}
+
+func (d *database) Transaction(ctx context.Context) (core.DatabaseTransaction, error) {
+	txn, err := d.db.BeginTxx(ctx, nil)
+	if err != nil {
+		return nil, fmt.Errorf("%w: starting transaction: %w", core.ErrDatabase, err)
+	}
+
+	txHandler := &transaction{
+		transaction: txn,
+		logger:      d.logger.With("module", "transactioner"),
+	}
+
+	return txHandler, nil
+}
+
+func (t *transaction) Apply(steps ...core.TransactionFunc) error {
+	for stepNumber, stepFunc := range steps {
+		if err := stepFunc(t.transaction); err != nil {
+			t.logger.Error(
+				"Error occurred.",
+				"step", stepNumber,
+				"error", err.Error(),
+				"module", "core/database",
+				"subsystem", "transaction",
+			)
+
+			if rollbackErr := t.transaction.Rollback(); rollbackErr != nil {
+				t.logger.Error(
+					"Transaction rollback failed.",
+					"error", err.Error(),
+					"module", "core/database",
+					"subsystem", "transaction",
+				)
+
+				return fmt.Errorf("%w: transaction rollback: %w", core.ErrDatabase, rollbackErr)
+			}
+
+			return err
+		}
+	}
+
+	if err := t.transaction.Commit(); err != nil {
+		t.logger.Error(
+			"Transaction commit failed.",
+			"error", err.Error(),
+			"module", "core/database",
+			"subsystem", "transaction",
+		)
+
+		if rollbackErr := t.transaction.Rollback(); rollbackErr != nil {
+			t.logger.Error(
+				"Transaction rollback failed.",
+				"error", err.Error(),
+				"module", "core/database",
+				"subsystem", "transaction",
+			)
+
+			return fmt.Errorf("%w: transaction rollback: %w", core.ErrDatabase, rollbackErr)
+		}
+
+		return fmt.Errorf("%w: transaction commit: %w", core.ErrDatabase, err)
+	}
+
+	return nil
+}

server/internal/services/core/httpserver.go

@@ -0,0 +1,14 @@
+package core
+
+import (
+	"errors"
+)
+
+// ServiceNameHTTPServer is a name for HTTP server service.
+const ServiceNameHTTPServer = "core/http_server"
+
+// ErrHTTPServerIsInvalid appears when HTTP server service implementation is invalid.
+var ErrHTTPServerIsInvalid = errors.New("HTTP server service implementation is invalid")
+
+// HTTPServer is an interface for HTTP server service.
+type HTTPServer interface{}

server/internal/services/core/httpserver/httpserver.go

@@ -0,0 +1,91 @@
+package httpserver
+
+import (
+	"errors"
+	"fmt"
+	"log/slog"
+	"net/http"
+
+	"bunker/server/internal/application"
+	"bunker/server/internal/services/core"
+)
+
+var (
+	_ = core.HTTPServer(&httpServer{})
+
+	errHTTPServer = errors.New("HTTP server core service")
+)
+
+type httpServer struct {
+	app     *application.Application
+	logger  *slog.Logger
+	db      core.Database
+	httpSrv *http.Server
+}
+
+// Initialize initializes service.
+func Initialize(app *application.Application) error {
+	httpSrv := &httpServer{
+		app: app,
+	}
+
+	if err := app.RegisterService(httpSrv); err != nil {
+		return fmt.Errorf("%w: %w", errHTTPServer, err)
+	}
+
+	return nil
+}
+
+func (h *httpServer) Configure() error {
+	h.logger.Debug("Configuring service...")
+
+	if err := h.configureHTTPServer(); err != nil {
+		return fmt.Errorf("configure: %w", err)
+	}
+
+	return nil
+}
+
+func (h *httpServer) ConnectDependencies() error {
+	databaseRaw := h.app.Service(core.ServiceNameDatabase)
+	if databaseRaw == nil {
+		return fmt.Errorf("connect dependencies: get database service: %w", application.ErrServiceNotFound)
+	}
+
+	database, valid := databaseRaw.(core.Database)
+	if !valid {
+		return fmt.Errorf("connect dependencies: type assert database service: %w", core.ErrDatabaseIsInvalid)
+	}
+
+	h.db = database
+
+	return nil
+}
+
+func (h *httpServer) Initialize() error {
+	h.logger = h.app.NewLogger("service", core.ServiceNameHTTPServer)
+
+	h.logger.Info("Initializing...")
+
+	return nil
+}
+
+func (h *httpServer) Name() string {
+	return core.ServiceNameHTTPServer
+}
+
+func (h *httpServer) LaunchStartupTasks() error {
+	h.logger.Debug("Launching startup tasks...")
+
+	go h.startHTTPServer()
+
+	return nil
+}
+
+func (h *httpServer) Shutdown() error {
+	if err := h.stopHTTPServer(); err != nil {
+		return fmt.Errorf("%w: Shutdown: %w", errHTTPServer, err)
+	}
+
+	return nil
+}

server/internal/services/core/httpserver/server.go

@@ -0,0 +1,85 @@
+package httpserver
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"time"
+
+	"github.com/coder/websocket"
+)
+
+const httpServerAddrEnvVar = "BUNKERD_HTTP_ADDRESS"
+
+var (
+	errHTTPServerAddrInvalid = errors.New("BUNKERD_HTTP_ADDRESS environment variable contains invalid address to " +
+		"listen, should be 'host:port'")
+	errHTTPServerAddrNotFound = errors.New("BUNKERD_HTTP_ADDRESS environment variable empty")
+)
+
+func (h *httpServer) configureHTTPServer() error {
+	httpSrvAddr, found := os.LookupEnv(httpServerAddrEnvVar)
+	if !found {
+		return fmt.Errorf("configure HTTP server: get address from environment variable: %w", errHTTPServerAddrNotFound)
+	}
+
+	host, port, err := net.SplitHostPort(httpSrvAddr)
+	if err != nil {
+		return fmt.Errorf("configure HTTP server: validate HTTP server address: %w", err)
+	}
+
+	if httpSrvAddr != host+":"+port {
+		return fmt.Errorf("configure HTTP server: validate HTTP server address: %w", errHTTPServerAddrInvalid)
+	}
+
+	mux := new(http.ServeMux)
+	mux.HandleFunc("GET /api/v1/socket", h.handleWebsocketRequest)
+
+	h.httpSrv = &http.Server{
+		Addr:              httpSrvAddr,
+		Handler:           mux,
+		ReadHeaderTimeout: time.Second * 3,
+	}
+
+	return nil
+}
+
+func (h *httpServer) handleWebsocketRequest(w http.ResponseWriter, r *http.Request) {
+	wsConn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
+		OnPingReceived: func(_ context.Context, _ []byte) bool {
+			return true
+		},
+	})
+	if err != nil {
+		h.logger.Error("Failed to accept WS connection!", "error", err.Error())
+
+		return
+	}
+
+	defer func() {
+		if err := wsConn.CloseNow(); err != nil {
+			h.logger.Warn("Failed to close WS connection in defer!", "error", err.Error())
+		}
+	}()
+}
+
+func (h *httpServer) startHTTPServer() {
+	h.logger.Info("Starting listening for HTTP requests.", "address", h.httpSrv.Addr)
+
+	if err := h.httpSrv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		h.logger.Warn("Error when listening to ", "error", err.Error())
+	}
+}
+
+func (h *httpServer) stopHTTPServer() error {
+	h.logger.Info("Stopping HTTP server...")
+
+	if err := h.httpSrv.Shutdown(h.app.ContextWithTimeout(time.Second * 3)); err != nil {
+		return fmt.Errorf("stopping HTTP server: %w", err)
+	}
+
+	return nil
+}

server/internal/services/core/options.go

@@ -0,0 +1,14 @@
+package core
+
+import (
+	"errors"
+)
+
+// ServiceNameOptions is a name for options service which controls options storage.
+const ServiceNameOptions = "core/options"
+
+// ErrOptionsIsInvalid appears when options service implementation is invalid.
+var ErrOptionsIsInvalid = errors.New("options service implementation is invalid")
+
+// Options is an interface for options service.
+type Options interface{}

server/internal/services/core/options/migrations.go

@@ -0,0 +1,17 @@
+package options
+
+import (
+	"embed"
+	"fmt"
+)
+
+//go:embed migrations
+var migrations embed.FS
+
+func (o *options) registerMigrations() error {
+	if err := o.db.RegisterMigrations("core/options", migrations); err != nil {
+		return fmt.Errorf("register migrations: %w", err)
+	}
+
+	return nil
+}

server/internal/services/core/options/migrations/0001_initial.sql

@@ -0,0 +1,12 @@
+-- +goose Up
+CREATE TABLE IF NOT EXISTS options (
+    id UUID NOT NULL PRIMARY KEY,
+    user_id UUID NOT NULL,
+    key VARCHAR(1024) NOT NULL,
+    value VARCHAR(8192)
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS options_user_id_key_idx ON options(user_id, key);
+
+-- +goose Down
+DROP TABLE IF EXISTS options;

server/internal/services/core/options/options.go

@@ -0,0 +1,79 @@
+package options
+
+import (
+	"errors"
+	"fmt"
+	"log/slog"
+
+	"bunker/server/internal/application"
+	"bunker/server/internal/services/core"
+)
+
+var (
+	_ = core.Options(&options{})
+
+	errOptions = errors.New("options core service")
+)
+
+type options struct {
+	app    *application.Application
+	logger *slog.Logger
+	db     core.Database
+}
+
+// Initialize initializes service.
+func Initialize(app *application.Application) error {
+	opts := &options{
+		app: app,
+	}
+
+	if err := app.RegisterService(opts); err != nil {
+		return fmt.Errorf("%w: %w", errOptions, err)
+	}
+
+	return nil
+}
+
+func (o *options) Configure() error {
+	if err := o.registerMigrations(); err != nil {
+		return fmt.Errorf("configure: %w", err)
+	}
+
+	return nil
+}
+
+func (o *options) ConnectDependencies() error {
+	databaseRaw := o.app.Service(core.ServiceNameDatabase)
+	if databaseRaw == nil {
+		return fmt.Errorf("connect dependencies: get database service: %w", application.ErrServiceNotFound)
+	}
+
+	database, valid := databaseRaw.(core.Database)
+	if !valid {
+		return fmt.Errorf("connect dependencies: type assert database service: %w", core.ErrDatabaseIsInvalid)
+	}
+
+	o.db = database
+
+	return nil
+}
+
+func (o *options) Initialize() error {
+	o.logger = o.app.NewLogger("service", core.ServiceNameOptions)
+
+	o.logger.Info("Initializing...")
+
+	return nil
+}
+
+func (o *options) Name() string {
+	return core.ServiceNameOptions
+}
+
+func (o *options) LaunchStartupTasks() error {
+	return nil
+}
+
+func (o *options) Shutdown() error {
+	return nil
+}