feat(fxcore): Added core tasks system (ankorstore#326)

Author: ekkinox

fxcore/README.md

@@ -134,6 +134,9 @@ modules:
         liveness:            
           expose: true                 # to expose health check liveness route, disabled by default
           path: /livez                 # health check liveness route path (default /livez)
+      tasks:
+        expose: true                   # to expose tasks route, disabled by default
+        path: /tasks/:name             # tasks route path (default /tasks/:name)
       debug:
         config:
           expose: true                 # to expose debug config route

fxcore/go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/ankorstore/yokai/healthcheck v1.1.0
 	github.com/ankorstore/yokai/httpserver v1.6.0
 	github.com/ankorstore/yokai/log v1.2.0
-	github.com/ankorstore/yokai/trace v1.3.0
+	github.com/ankorstore/yokai/trace v1.4.0
 	github.com/arl/statsviz v0.6.0
 	github.com/labstack/echo/v4 v4.13.3
 	github.com/labstack/gommon v0.4.2

fxcore/go.sum

@@ -22,6 +22,8 @@ github.com/ankorstore/yokai/log v1.2.0 h1:jiuDiC0dtqIGIOsFQslUHYoFJ1qjI+rOMa6dI1
 github.com/ankorstore/yokai/log v1.2.0/go.mod h1:MVvUcms1AYGo0BT6l88B9KJdvtK6/qGKdgyKVXfbmyc=
 github.com/ankorstore/yokai/trace v1.3.0 h1:0ji32oymIcxTmH5h6GRWLo5ypwBbWrZkXRf9rWF9070=
 github.com/ankorstore/yokai/trace v1.3.0/go.mod h1:m7EL2MRBilgCtrly5gA4F0jkGSXR2EbG6LsotbTJ4nA=
+github.com/ankorstore/yokai/trace v1.4.0 h1:AdEQs/4TEuqOJ9p/EfsQmrtmkSG3pcmE7r/l+FQFxY8=
+github.com/ankorstore/yokai/trace v1.4.0/go.mod h1:m7EL2MRBilgCtrly5gA4F0jkGSXR2EbG6LsotbTJ4nA=
 github.com/arl/statsviz v0.6.0 h1:jbW1QJkEYQkufd//4NDYRSNBpwJNrdzPahF7ZmoGdyE=
 github.com/arl/statsviz v0.6.0/go.mod h1:0toboo+YGSUXDaS4g1D5TVS4dXs7S7YYT5J/qnW2h8s=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=

fxcore/info.go

@@ -1,6 +1,9 @@
 package fxcore
 
 import (
+	"fmt"
+	"sort"
+
 	"github.com/ankorstore/yokai/config"
 	"github.com/ankorstore/yokai/log"
 	"github.com/ankorstore/yokai/trace"
@@ -129,3 +132,57 @@ func (i *FxCoreModuleInfo) Data() map[string]interface{} {
 		"extra": i.ExtraInfos,
 	}
 }
+
+// FxModuleInfoRegistry is the registry collecting info about registered modules.
+type FxModuleInfoRegistry struct {
+	infos map[string]FxModuleInfo
+}
+
+// FxModuleInfoRegistryParam allows injection of the required dependencies in [NewFxModuleInfoRegistry].
+type FxModuleInfoRegistryParam struct {
+	fx.In
+	Infos []any `group:"core-module-infos"`
+}
+
+// NewFxModuleInfoRegistry returns a new [FxModuleInfoRegistry].
+func NewFxModuleInfoRegistry(p FxModuleInfoRegistryParam) *FxModuleInfoRegistry {
+	infos := make(map[string]FxModuleInfo)
+
+	for _, info := range p.Infos {
+		if castInfo, ok := info.(FxModuleInfo); ok {
+			infos[castInfo.Name()] = castInfo
+		}
+	}
+
+	return &FxModuleInfoRegistry{
+		infos: infos,
+	}
+}
+
+func (r *FxModuleInfoRegistry) Names() []string {
+	names := make([]string, len(r.infos))
+
+	i := 0
+	for name := range r.infos {
+		names[i] = name
+		i++
+	}
+
+	sort.Strings(names)
+
+	return names
+}
+
+// All returns a map of all registered [FxModuleInfo].
+func (r *FxModuleInfoRegistry) All() map[string]FxModuleInfo {
+	return r.infos
+}
+
+// Find returns a [FxModuleInfo] by name.
+func (r *FxModuleInfoRegistry) Find(name string) (FxModuleInfo, error) {
+	if info, ok := r.infos[name]; ok {
+		return info, nil
+	}
+
+	return nil, fmt.Errorf("fx module info with name %s was not found", name)
+}

fxcore/info_test.go

@@ -8,7 +8,17 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestNewFxCoreModuleInfo(t *testing.T) {
+type testModuleInfo struct{}
+
+func (i *testModuleInfo) Name() string {
+	return "test"
+}
+
+func (i *testModuleInfo) Data() map[string]interface{} {
+	return map[string]interface{}{}
+}
+
+func TestFxCoreModuleInfo(t *testing.T) {
 	t.Setenv("APP_ENV", "test")
 
 	cfg, err := config.NewDefaultConfigFactory().Create(
@@ -54,3 +64,70 @@ func TestNewFxCoreModuleInfo(t *testing.T) {
 		info.Data(),
 	)
 }
+
+func TestFxModuleInfoRegistry(t *testing.T) {
+	t.Parallel()
+
+	createRegistry := func(tb testing.TB) *fxcore.FxModuleInfoRegistry {
+		tb.Helper()
+
+		cfg, err := config.NewDefaultConfigFactory().Create(
+			config.WithFilePaths("./testdata/config"),
+		)
+		assert.NoError(tb, err)
+
+		return fxcore.NewFxModuleInfoRegistry(fxcore.FxModuleInfoRegistryParam{
+			Infos: []interface{}{
+				&testModuleInfo{},
+				fxcore.NewFxCoreModuleInfo(fxcore.FxCoreModuleInfoParam{
+					Config:     cfg,
+					ExtraInfos: []fxcore.FxExtraInfo{},
+				}),
+				"invalid",
+			},
+		})
+	}
+
+	t.Run("test type", func(t *testing.T) {
+		t.Parallel()
+
+		registry := createRegistry(t)
+
+		assert.IsType(t, &fxcore.FxModuleInfoRegistry{}, registry)
+	})
+
+	t.Run("test all", func(t *testing.T) {
+		t.Parallel()
+
+		registry := createRegistry(t)
+
+		assert.Len(t, registry.All(), 2)
+	})
+
+	t.Run("test names", func(t *testing.T) {
+		t.Parallel()
+
+		registry := createRegistry(t)
+
+		assert.Equal(t, []string{fxcore.ModuleName, "test"}, registry.Names())
+	})
+
+	t.Run("test find", func(t *testing.T) {
+		t.Parallel()
+
+		registry := createRegistry(t)
+
+		testInfo, err := registry.Find("test")
+		assert.NoError(t, err)
+		assert.Equal(t, "test", testInfo.Name())
+
+		coreInfo, err := registry.Find(fxcore.ModuleName)
+		assert.NoError(t, err)
+		assert.Equal(t, fxcore.ModuleName, coreInfo.Name())
+
+		invalidInfo, err := registry.Find("invalid")
+		assert.Error(t, err)
+		assert.Equal(t, "fx module info with name invalid was not found", err.Error())
+		assert.Nil(t, invalidInfo)
+	})
+}

fxcore/module.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"embed"
 	"fmt"
+	"io"
 	"net/http"
 	"strconv"
 
@@ -37,6 +38,7 @@ const (
 	DefaultHealthCheckStartupPath   = "/healthz"
 	DefaultHealthCheckLivenessPath  = "/livez"
 	DefaultHealthCheckReadinessPath = "/readyz"
+	DefaultTasksPath                = "/tasks"
 	DefaultDebugConfigPath          = "/debug/config"
 	DefaultDebugPProfPath           = "/debug/pprof"
 	DefaultDebugBuildPath           = "/debug/build"
@@ -63,6 +65,7 @@ var FxCoreModule = fx.Module(
 	fxhealthcheck.FxHealthcheckModule,
 	fx.Provide(
 		NewFxModuleInfoRegistry,
+		NewTaskRegistry,
 		NewFxCore,
 		fx.Annotate(
 			NewFxCoreModuleInfo,
@@ -92,7 +95,8 @@ type FxCoreParam struct {
 	Checker         *healthcheck.Checker
 	Config          *config.Config
 	Logger          *log.Logger
-	Registry        *FxModuleInfoRegistry
+	InfoRegistry    *FxModuleInfoRegistry
+	TaskRegistry    *TaskRegistry
 	MetricsRegistry *prometheus.Registry
 }
 
@@ -232,7 +236,7 @@ func withHandlers(coreServer *echo.Echo, p FxCoreParam) (*echo.Echo, error) {
 	dashboardEnabled := p.Config.GetBool("modules.core.server.dashboard.enabled")
 
 	// dashboard overview
-	overviewInfo, err := p.Registry.Find(ModuleName)
+	overviewInfo, err := p.InfoRegistry.Find(ModuleName)
 	if err != nil {
 		return nil, err
 	}
@@ -248,6 +252,7 @@ func withHandlers(coreServer *echo.Echo, p FxCoreParam) (*echo.Echo, error) {
 	overviewTraceProcessorExpose := p.Config.GetBool("modules.core.server.dashboard.overview.trace_processor")
 
 	// template expositions
+	tasksExpose := p.Config.GetBool("modules.core.server.tasks.expose")
 	metricsExpose := p.Config.GetBool("modules.core.server.metrics.expose")
 	startupExpose := p.Config.GetBool("modules.core.server.healthcheck.startup.expose")
 	livenessExpose := p.Config.GetBool("modules.core.server.healthcheck.liveness.expose")
@@ -260,6 +265,7 @@ func withHandlers(coreServer *echo.Echo, p FxCoreParam) (*echo.Echo, error) {
 	modulesExpose := p.Config.GetBool("modules.core.server.debug.modules.expose")
 
 	// template paths
+	tasksPath := p.Config.GetString("modules.core.server.tasks.path")
 	metricsPath := p.Config.GetString("modules.core.server.metrics.path")
 	startupPath := p.Config.GetString("modules.core.server.healthcheck.startup.path")
 	livenessPath := p.Config.GetString("modules.core.server.healthcheck.liveness.path")
@@ -271,6 +277,48 @@ func withHandlers(coreServer *echo.Echo, p FxCoreParam) (*echo.Echo, error) {
 	buildPath := p.Config.GetString("modules.core.server.debug.build.path")
 	modulesPath := p.Config.GetString("modules.core.server.debug.modules.path")
 
+	// tasks
+	if tasksExpose {
+		if tasksPath == "" {
+			tasksPath = DefaultTasksPath
+		}
+
+		coreServer.POST(fmt.Sprintf("%s/:name", tasksPath), func(c echo.Context) error {
+			ctx := c.Request().Context()
+
+			logger := log.CtxLogger(ctx)
+
+			name := c.Param("name")
+
+			input, err := io.ReadAll(c.Request().Body)
+			if err != nil {
+				logger.Error().Err(err).Str("task", name).Msg("request body read error")
+
+				return echo.NewHTTPError(http.StatusInternalServerError, fmt.Sprintf("cannot read request body: %v", err.Error()))
+			}
+
+			err = c.Request().Body.Close()
+			if err != nil {
+				logger.Error().Err(err).Str("task", name).Msg("request body close error")
+
+				return echo.NewHTTPError(http.StatusInternalServerError, fmt.Sprintf("cannot close request body: %v", err.Error()))
+			}
+
+			res := p.TaskRegistry.Run(ctx, name, input)
+			if !res.Success {
+				logger.Error().Err(err).Str("task", name).Msg("task execution error")
+
+				return c.JSON(http.StatusInternalServerError, res)
+			}
+
+			logger.Info().Str("task", name).Msg("task execution success")
+
+			return c.JSON(http.StatusOK, res)
+		})
+
+		coreServer.Logger.Debug("registered tasks handler")
+	}
+
 	// metrics
 	if metricsExpose {
 		if metricsPath == "" {
@@ -393,14 +441,14 @@ func withHandlers(coreServer *echo.Echo, p FxCoreParam) (*echo.Echo, error) {
 		coreServer.Logger.Debug("registered debug build handler")
 	}
 
-	// debug modules
+	// modules
 	if modulesExpose || appDebug {
 		if modulesPath == "" {
 			modulesPath = DefaultDebugModulesPath
 		}
 
 		coreServer.GET(fmt.Sprintf("%s/:name", modulesPath), func(c echo.Context) error {
-			info, err := p.Registry.Find(c.Param("name"))
+			info, err := p.InfoRegistry.Find(c.Param("name"))
 			if err != nil {
 				return echo.NewHTTPError(http.StatusNotFound, err.Error())
 			}
@@ -466,6 +514,9 @@ func withHandlers(coreServer *echo.Echo, p FxCoreParam) (*echo.Echo, error) {
 				"overviewLogOutputExpose":      overviewLogOutputExpose,
 				"overviewTraceSamplerExpose":   overviewTraceSamplerExpose,
 				"overviewTraceProcessorExpose": overviewTraceProcessorExpose,
+				"tasksExpose":                  tasksExpose,
+				"tasksPath":                    tasksPath,
+				"tasksNames":                   p.TaskRegistry.Names(),
 				"metricsExpose":                metricsExpose,
 				"metricsPath":                  metricsPath,
 				"startupExpose":                startupExpose,
@@ -486,7 +537,7 @@ func withHandlers(coreServer *echo.Echo, p FxCoreParam) (*echo.Echo, error) {
 				"buildPath":                    buildPath,
 				"modulesExpose":                modulesExpose || appDebug,
 				"modulesPath":                  modulesPath,
-				"modulesNames":                 p.Registry.Names(),
+				"modulesNames":                 p.InfoRegistry.Names(),
 				"theme":                        theme,
 			})
 		})