🎨 ♻️ Restructure code around the actor model

.dockerignore

@@ -1,8 +1,8 @@
 /bin/
 /data/
 /internal/app/gen_version.go
-/internal/ffi/filterdsl/rust-crate/target/
-/internal/ffi/vrl/rust-crate/target/
+/internal/utils/ffi/filterdsl/rust-crate/target/
+/internal/utils/ffi/vrl/rust-crate/target/
 /web/public/
 /web/app/node_modules/
 /web/app/dist/

.github/dependabot.yml

@@ -26,15 +26,15 @@ updates:
     labels:
       - "deps: go"
   - package-ecosystem: "cargo"
-    directory: "/internal/ffi/vrl/rust-crate"
+    directory: "/internal/utils/ffi/vrl/rust-crate"
     schedule:
       interval: "weekly"
     commit-message:
       prefix: ":arrow_up: "
     labels:
       - "deps: rust"
   - package-ecosystem: "cargo"
-    directory: "/internal/ffi/filterdsl/rust-crate"
+    directory: "/internal/utils/ffi/filterdsl/rust-crate"
     schedule:
       interval: "weekly"
     commit-message:

api/change_password.go

@@ -8,7 +8,9 @@ import (
 	"github.com/swaggest/usecase"
 	"github.com/swaggest/usecase/status"
 
-	"link-society.com/flowg/internal/data/auth"
+	apiUtils "link-society.com/flowg/internal/utils/api"
+
+	"link-society.com/flowg/internal/storage/auth"
 )
 
 type ChangePasswordRequest struct {
@@ -20,25 +22,23 @@ type ChangePasswordResponse struct {
 	Success bool `json:"success"`
 }
 
-func ChangePasswordUsecase(authDb *auth.Database) usecase.Interactor {
-	userSys := auth.NewUserSystem(authDb)
-
+func ChangePasswordUsecase(authStorage *auth.Storage) usecase.Interactor {
 	u := usecase.NewInteractor(
 		func(
 			ctx context.Context,
 			req ChangePasswordRequest,
 			resp *ChangePasswordResponse,
 		) error {
-			user := auth.GetContextUser(ctx)
+			user := apiUtils.GetContextUser(ctx)
 
-			switch valid, err := userSys.VerifyUserPassword(user.Name, req.OldPassword); {
+			switch valid, err := authStorage.VerifyUserPassword(ctx, user.Name, req.OldPassword); {
 			case err != nil:
 				slog.ErrorContext(
 					ctx,
 					"Failed to verify user password",
-					"channel", "api",
-					"username", user.Name,
-					"error", err.Error(),
+					slog.String("channel", "api"),
+					slog.String("username", user.Name),
+					slog.String("error", err.Error()),
 				)
 
 				resp.Success = false
@@ -49,13 +49,13 @@ func ChangePasswordUsecase(authDb *auth.Database) usecase.Interactor {
 				return status.Wrap(errors.New("invalid credentials"), status.PermissionDenied)
 			}
 
-			if err := userSys.SaveUser(*user, req.NewPassword); err != nil {
+			if err := authStorage.SaveUser(ctx, *user, req.NewPassword); err != nil {
 				slog.ErrorContext(
 					ctx,
 					"Failed to update user password",
-					"channel", "api",
-					"username", user.Name,
-					"error", err.Error(),
+					slog.String("channel", "api"),
+					slog.String("username", user.Name),
+					slog.String("error", err.Error()),
 				)
 
 				resp.Success = false

api/configure_stream.go

@@ -7,41 +7,42 @@ import (
 	"github.com/swaggest/usecase"
 	"github.com/swaggest/usecase/status"
 
-	"link-society.com/flowg/internal/data/auth"
-	"link-society.com/flowg/internal/data/logstorage"
+	apiUtils "link-society.com/flowg/internal/utils/api"
+
+	"link-society.com/flowg/internal/models"
+	"link-society.com/flowg/internal/storage/auth"
+	"link-society.com/flowg/internal/storage/log"
 )
 
 type ConfigureStreamRequest struct {
-	Stream string                  `path:"stream" minLength:"1"`
-	Config logstorage.StreamConfig `json:"config"`
+	Stream string              `path:"stream" minLength:"1"`
+	Config models.StreamConfig `json:"config"`
 }
 
 type ConfigureStreamResponse struct {
 	Success bool `json:"success"`
 }
 
 func ConfigureStreamUsecase(
-	authDb *auth.Database,
-	logDb *logstorage.Storage,
+	authStorage *auth.Storage,
+	logStorage *log.Storage,
 ) usecase.Interactor {
-	metaSys := logstorage.NewMetaSystem(logDb)
-
 	u := usecase.NewInteractor(
-		auth.RequireScopeApiDecorator(
-			authDb,
-			auth.SCOPE_WRITE_STREAMS,
+		apiUtils.RequireScopeApiDecorator(
+			authStorage,
+			models.SCOPE_WRITE_STREAMS,
 			func(
 				ctx context.Context,
 				req ConfigureStreamRequest,
 				resp *ConfigureStreamResponse,
 			) error {
-				err := metaSys.ConfigureStream(req.Stream, req.Config)
+				err := logStorage.ConfigureStream(ctx, req.Stream, req.Config)
 				if err != nil {
 					slog.ErrorContext(
 						ctx,
 						"Failed to configure stream",
-						"stream", req.Stream,
-						"error", err.Error(),
+						slog.String("stream", req.Stream),
+						slog.String("error", err.Error()),
 					)
 
 					return status.Wrap(err, status.Internal)

api/create_token.go

@@ -7,7 +7,9 @@ import (
 	"github.com/swaggest/usecase"
 	"github.com/swaggest/usecase/status"
 
-	"link-society.com/flowg/internal/data/auth"
+	apiUtils "link-society.com/flowg/internal/utils/api"
+
+	"link-society.com/flowg/internal/storage/auth"
 )
 
 type CreateTokenRequest struct{}
@@ -18,25 +20,23 @@ type CreateTokenResponse struct {
 	TokenUUID string `json:"token_uuid"`
 }
 
-func CreateTokenUsecase(authDb *auth.Database) usecase.Interactor {
-	tokenSys := auth.NewTokenSystem(authDb)
-
+func CreateTokenUsecase(authStorage *auth.Storage) usecase.Interactor {
 	u := usecase.NewInteractor(
 		func(
 			ctx context.Context,
 			req CreateTokenRequest,
 			resp *CreateTokenResponse,
 		) error {
-			user := auth.GetContextUser(ctx)
+			user := apiUtils.GetContextUser(ctx)
 
-			token, tokenUuid, err := tokenSys.CreateToken(user.Name)
+			token, tokenUuid, err := authStorage.CreateToken(ctx, user.Name)
 			if err != nil {
 				slog.ErrorContext(
 					ctx,
 					"Failed to create token",
-					"channel", "api",
-					"user", user.Name,
-					"error", err.Error(),
+					slog.String("channel", "api"),
+					slog.String("user", user.Name),
+					slog.String("error", err.Error()),
 				)
 
 				resp.Success = false

api/delete_alert.go

@@ -7,8 +7,11 @@ import (
 	"github.com/swaggest/usecase"
 	"github.com/swaggest/usecase/status"
 
-	"link-society.com/flowg/internal/data/auth"
-	"link-society.com/flowg/internal/data/config"
+	apiUtils "link-society.com/flowg/internal/utils/api"
+
+	"link-society.com/flowg/internal/models"
+	"link-society.com/flowg/internal/storage/auth"
+	"link-society.com/flowg/internal/storage/config"
 )
 
 type DeleteAlertRequest struct {
@@ -20,28 +23,26 @@ type DeleteAlertResponse struct {
 }
 
 func DeleteAlertUsecase(
-	authDb *auth.Database,
+	authStorage *auth.Storage,
 	configStorage *config.Storage,
 ) usecase.Interactor {
-	alertSys := config.NewAlertSystem(configStorage)
-
 	u := usecase.NewInteractor(
-		auth.RequireScopeApiDecorator(
-			authDb,
-			auth.SCOPE_WRITE_ALERTS,
+		apiUtils.RequireScopeApiDecorator(
+			authStorage,
+			models.SCOPE_WRITE_ALERTS,
 			func(
 				ctx context.Context,
 				req DeleteAlertRequest,
 				resp *DeleteAlertResponse,
 			) error {
-				err := alertSys.Delete(req.Alert)
+				err := configStorage.DeleteAlert(ctx, req.Alert)
 				if err != nil {
 					slog.ErrorContext(
 						ctx,
 						"Failed to delete alert",
-						"channel", "api",
-						"alert", req.Alert,
-						"error", err.Error(),
+						slog.String("channel", "api"),
+						slog.String("alert", req.Alert),
+						slog.String("error", err.Error()),
 					)
 
 					resp.Success = false

api/delete_pipeline.go

@@ -7,8 +7,11 @@ import (
 	"github.com/swaggest/usecase"
 	"github.com/swaggest/usecase/status"
 
-	"link-society.com/flowg/internal/data/auth"
-	"link-society.com/flowg/internal/data/config"
+	apiUtils "link-society.com/flowg/internal/utils/api"
+
+	"link-society.com/flowg/internal/models"
+	"link-society.com/flowg/internal/storage/auth"
+	"link-society.com/flowg/internal/storage/config"
 )
 
 type DeletePipelineRequest struct {
@@ -20,28 +23,26 @@ type DeletePipelineResponse struct {
 }
 
 func DeletePipelineUsecase(
-	authDb *auth.Database,
+	authStorage *auth.Storage,
 	configStorage *config.Storage,
 ) usecase.Interactor {
-	pipelineSys := config.NewPipelineSystem(configStorage)
-
 	u := usecase.NewInteractor(
-		auth.RequireScopeApiDecorator(
-			authDb,
-			auth.SCOPE_WRITE_PIPELINES,
+		apiUtils.RequireScopeApiDecorator(
+			authStorage,
+			models.SCOPE_WRITE_PIPELINES,
 			func(
 				ctx context.Context,
 				req DeletePipelineRequest,
 				resp *DeletePipelineResponse,
 			) error {
-				err := pipelineSys.Delete(req.Pipeline)
+				err := configStorage.DeletePipeline(ctx, req.Pipeline)
 				if err != nil {
 					slog.ErrorContext(
 						ctx,
 						"Failed to delete pipeline",
-						"channel", "api",
-						"pipeline", req.Pipeline,
-						"error", err.Error(),
+						slog.String("channel", "api"),
+						slog.String("pipeline", req.Pipeline),
+						slog.String("error", err.Error()),
 					)
 
 					resp.Success = false

api/delete_role.go

@@ -7,7 +7,10 @@ import (
 	"github.com/swaggest/usecase"
 	"github.com/swaggest/usecase/status"
 
-	"link-society.com/flowg/internal/data/auth"
+	apiUtils "link-society.com/flowg/internal/utils/api"
+
+	"link-society.com/flowg/internal/models"
+	"link-society.com/flowg/internal/storage/auth"
 )
 
 type DeleteRoleRequest struct {
@@ -18,26 +21,24 @@ type DeleteRoleResponse struct {
 	Success bool `json:"success"`
 }
 
-func DeleteRoleUsecase(authDb *auth.Database) usecase.Interactor {
-	roleSys := auth.NewRoleSystem(authDb)
-
+func DeleteRoleUsecase(authStorage *auth.Storage) usecase.Interactor {
 	u := usecase.NewInteractor(
-		auth.RequireScopeApiDecorator(
-			authDb,
-			auth.SCOPE_WRITE_ACLS,
+		apiUtils.RequireScopeApiDecorator(
+			authStorage,
+			models.SCOPE_WRITE_ACLS,
 			func(
 				ctx context.Context,
 				req DeleteRoleRequest,
 				resp *DeleteRoleResponse,
 			) error {
-				err := roleSys.DeleteRole(req.Role)
+				err := authStorage.DeleteRole(ctx, req.Role)
 				if err != nil {
 					slog.ErrorContext(
 						ctx,
 						"Failed to delete role",
-						"channel", "api",
-						"role", req.Role,
-						"error", err.Error(),
+						slog.String("channel", "api"),
+						slog.String("role", req.Role),
+						slog.String("error", err.Error()),
 					)
 
 					resp.Success = false

api/delete_token.go

@@ -7,7 +7,9 @@ import (
 	"github.com/swaggest/usecase"
 	"github.com/swaggest/usecase/status"
 
-	"link-society.com/flowg/internal/data/auth"
+	apiUtils "link-society.com/flowg/internal/utils/api"
+
+	"link-society.com/flowg/internal/storage/auth"
 )
 
 type DeleteTokenRequest struct {
@@ -18,26 +20,24 @@ type DeleteTokenResponse struct {
 	Success bool `json:"success"`
 }
 
-func DeleteTokenUsecase(authDb *auth.Database) usecase.Interactor {
-	tokenSys := auth.NewTokenSystem(authDb)
-
+func DeleteTokenUsecase(authStorage *auth.Storage) usecase.Interactor {
 	u := usecase.NewInteractor(
 		func(
 			ctx context.Context,
 			req DeleteTokenRequest,
 			resp *DeleteTokenResponse,
 		) error {
-			user := auth.GetContextUser(ctx)
+			user := apiUtils.GetContextUser(ctx)
 
-			err := tokenSys.DeleteToken(user.Name, req.TokenUUID)
+			err := authStorage.DeleteToken(ctx, user.Name, req.TokenUUID)
 			if err != nil {
 				slog.ErrorContext(
 					ctx,
 					"Failed to delete token",
-					"channel", "api",
-					"user", user.Name,
-					"token-uuid", req.TokenUUID,
-					"error", err.Error(),
+					slog.String("channel", "api"),
+					slog.String("user", user.Name),
+					slog.String("token-uuid", req.TokenUUID),
+					slog.String("error", err.Error()),
 				)
 
 				resp.Success = false