chore: add discriminated union management

Makes it easier to persist and retrieve dynamic types such as deployment Source data and Job payloads.

This is still a WIP, and the front has not been updated to reflect the structure change yet.

cmd/configuration.go

@@ -25,7 +25,7 @@ var (
 )
 
 const (
-	databaseFilename              = "seelf.db?_foreign_keys=yes"
+	databaseFilename              = "seelf.db?_foreign_keys=yes&_txlock=immediate"
 	defaultConfigFilename         = "conf.yml"
 	defaultPort                   = 8080
 	defaultHost                   = ""

cmd/serve/deployments.go

@@ -19,7 +19,7 @@ type queueDeploymentBody struct {
 	Environment string                   `json:"environment" form:"environment"`
 	Raw         monad.Maybe[string]      `json:"raw"`
 	Archive     *multipart.FileHeader    `form:"archive"`
-	Git         monad.Maybe[git.Payload] `json:"git"`
+	Git         monad.Maybe[git.Request] `json:"git"`
 }
 
 func (s *server) queueDeploymentHandler() gin.HandlerFunc {

cmd/serve/server.go

@@ -209,7 +209,7 @@ func (s *server) configureServices() error {
 	)
 
 	s.docker = docker.New(s.options, s.logger)
-	handler := jobs.NewFacade(s.logger,
+	handler := jobs.NewFacade(
 		deploy.New(s.logger, deplcmd.Deploy(deploymentsStore, deploymentsStore, sourceFacade, s.docker)),
 		cleanup.New(s.logger, deplcmd.CleanupApp(deploymentsStore, appsStore, appsStore, s.docker)),
 	)
@@ -221,7 +221,7 @@ func (s *server) configureServices() error {
 		func(ctx context.Context, tags []string) error {
 			return processNextJob(ctx, workercmd.ProcessNextCommand{Names: tags})
 		},
-		async.Group(s.options.RunnersDeploymentCount(), deploy.JobName, cleanup.JobName),
+		async.Group(s.options.RunnersDeploymentCount(), deploy.Data{}.Discriminator(), cleanup.Data("").Discriminator()),
 	)
 	s.usersReader = usersStore
 
@@ -247,7 +247,7 @@ func (s *server) configureServices() error {
 	s.failRunningDeployments = deplcmd.FailRunningDeployments(deploymentsStore, deploymentsStore)
 
 	s.failRunningJobs = workercmd.FailRunningJobs(jobsStore, jobsStore)
-	s.queueJob = workercmd.Queue(jobsStore)
+	s.queueJob = workercmd.Queue(jobsStore, handler)
 
 	return nil
 }
@@ -361,9 +361,13 @@ func (s *server) configureRouter() {
 func (s *server) domainEventHandler(ctx context.Context, e event.Event) error {
 	switch evt := e.(type) {
 	case domain.DeploymentCreated:
-		s.queueJob(ctx, deploy.Queue(evt))
+		s.queueJob(ctx, workercmd.QueueCommand{
+			Payload: deploy.Request(evt),
+		})
 	case domain.AppCleanupRequested:
-		s.queueJob(ctx, cleanup.Queue(evt.ID))
+		s.queueJob(ctx, workercmd.QueueCommand{
+			Payload: cleanup.Request(evt),
+		})
 	}
 
 	return nil

internal/deployment/app/command/cleanup_app_test.go

@@ -49,7 +49,7 @@ func Test_CleanupApp(t *testing.T) {
 
 	t.Run("should fail if there are still pending or running deployments", func(t *testing.T) {
 		app := domain.NewApp("my-app", "uid")
-		depl, _ := app.NewDeployment(1, domain.NewMeta("some", "data"), domain.Production, options{}, "uid")
+		depl, _ := app.NewDeployment(1, meta{}, domain.Production, options{}, "uid")
 		app.RequestCleanup("uid")
 
 		uc := cleanup(initialData{

internal/deployment/app/command/deploy_test.go

@@ -107,8 +107,6 @@ func Test_Deploy(t *testing.T) {
 	})
 }
 
-const kind domain.Kind = "dummy"
-
 type dummySource struct {
 	err error
 }
@@ -117,8 +115,8 @@ func source(failedWithErr error) domain.Source {
 	return &dummySource{failedWithErr}
 }
 
-func (*dummySource) Prepare(domain.App, any) (domain.Meta, error) {
-	return domain.NewMeta(kind, ""), nil
+func (*dummySource) Prepare(domain.App, any) (domain.SourceData, error) {
+	return meta{}, nil
 }
 
 func (t *dummySource) Fetch(context.Context, domain.Deployment) error {
@@ -140,3 +138,8 @@ func (b *dummyBackend) Run(context.Context, domain.Deployment) (domain.Services,
 func (b *dummyBackend) Cleanup(context.Context, domain.App) error {
 	return nil
 }
+
+type meta struct{}
+
+func (meta) Discriminator() string { return "test" }
+func (m meta) NeedVCS() bool       { return false }

internal/deployment/app/command/fail_running_deployments_test.go

@@ -26,12 +26,12 @@ func Test_FailRunningDeployments(t *testing.T) {
 	t.Run("should reset running deployments", func(t *testing.T) {
 		errReset := errors.New("server_reset")
 
-		started, _ := app.NewDeployment(2, domain.Meta{}, domain.Production, opts, "some-uid")
+		started, _ := app.NewDeployment(2, meta{}, domain.Production, opts, "some-uid")
 		err := started.HasStarted()
 
 		testutil.IsNil(t, err)
 
-		succeeded, _ := app.NewDeployment(1, domain.Meta{}, domain.Production, opts, "some-uid")
+		succeeded, _ := app.NewDeployment(1, meta{}, domain.Production, opts, "some-uid")
 		succeeded.HasStarted()
 		err = succeeded.HasEnded(domain.Services{}, nil)
 

internal/deployment/app/command/promote_test.go

@@ -43,7 +43,7 @@ func Test_Promote(t *testing.T) {
 	})
 
 	t.Run("should correctly creates a new deployment based on the provided one", func(t *testing.T) {
-		dpl, _ := app.NewDeployment(1, domain.NewMeta("some", "data"), domain.Staging, opts, "some-uid")
+		dpl, _ := app.NewDeployment(1, meta{}, domain.Staging, opts, "some-uid")
 		uc := promote(dpl)
 
 		number, err := uc(ctx, command.PromoteCommand{

internal/deployment/app/command/redeploy_test.go

@@ -43,7 +43,7 @@ func Test_Redeploy(t *testing.T) {
 	})
 
 	t.Run("should correctly creates a new deployment based on the provided one", func(t *testing.T) {
-		dpl, _ := app.NewDeployment(1, domain.NewMeta("some", "data"), domain.Production, opts, "some-uid")
+		dpl, _ := app.NewDeployment(1, meta{}, domain.Production, opts, "some-uid")
 		uc := redeploy(dpl)
 
 		number, err := uc(ctx, command.RedeployCommand{

internal/deployment/app/query/gateway.go

@@ -9,6 +9,8 @@ import (
 	"github.com/YuukanOO/seelf/pkg/storage"
 )
 
+var SourceDataTypes = storage.NewDiscriminatedMapper[SourceData]()
+
 type (
 	// Access to the underlying storage adapter for read use cases
 	Gateway interface {
@@ -43,17 +45,19 @@ type (
 		AppID            string    `json:"app_id"`
 		DeploymentNumber int       `json:"deployment_number"`
 		Environment      string    `json:"environment"`
-		Meta             Meta      `json:"meta"`
+		Source           Source    `json:"source"`
 		State            State     `json:"state"`
 		RequestedAt      time.Time `json:"requested_at"`
 		RequestedBy      User      `json:"requested_by"`
 	}
 
-	Meta struct {
-		Kind string              `json:"kind"`
-		Data monad.Maybe[string] `json:"data"` // Contain source data only when the information is not sensitive
+	Source struct {
+		Discriminator string     `json:"discriminator"`
+		Data          SourceData `json:"data"`
 	}
 
+	SourceData storage.Discriminated
+
 	VCSConfig struct {
 		Url   string                          `json:"url"`
 		Token monad.Maybe[query.SecretString] `json:"token"`

internal/deployment/domain/deployment.go

@@ -42,10 +42,10 @@ type (
 		event.Emitter
 
 		id        DeploymentID
-		path      string
+		path      string // Relative path where deployment data will be stored
 		config    Config
 		state     State
-		source    Meta
+		source    SourceData
 		requested shared.Action[domain.UserID]
 	}
 
@@ -69,7 +69,7 @@ type (
 		Path      string
 		Config    Config
 		State     State
-		Source    Meta
+		Source    SourceData
 		Requested shared.Action[domain.UserID]
 	}
 
@@ -83,7 +83,7 @@ type (
 // entity to make sure a new deployment can be created for an app.
 func (a App) NewDeployment(
 	deployNumber DeploymentNumber,
-	meta Meta,
+	meta SourceData,
 	env Environment,
 	tmpl DeploymentDirTemplate,
 	requestedBy domain.UserID,
@@ -92,7 +92,7 @@ func (a App) NewDeployment(
 		return d, ErrAppCleanupRequested
 	}
 
-	if meta.kind.IsVCS() && !a.vcs.HasValue() {
+	if meta.NeedVCS() && !a.vcs.HasValue() {
 		return d, ErrVCSNotConfigured
 	}
 
@@ -148,8 +148,10 @@ func (a App) Promote(
 
 func DeploymentFrom(scanner storage.Scanner) (d Deployment, err error) {
 	var (
-		requestedAt time.Time
-		requestedBy domain.UserID
+		requestedAt             time.Time
+		requestedBy             domain.UserID
+		sourceMetaDiscriminator string
+		sourceMetaData          string
 	)
 
 	err = scanner.Scan(
@@ -165,20 +167,25 @@ func DeploymentFrom(scanner storage.Scanner) (d Deployment, err error) {
 		&d.state.services,
 		&d.state.startedAt,
 		&d.state.finishedAt,
-		&d.source.kind,
-		&d.source.data,
+		&sourceMetaDiscriminator,
+		&sourceMetaData,
 		&requestedAt,
 		&requestedBy,
 	)
 
+	if err != nil {
+		return d, err
+	}
+
+	d.source, err = SourceDataTypes.From(sourceMetaDiscriminator, sourceMetaData)
 	d.requested = shared.ActionFrom(requestedBy, requestedAt)
 
 	return d, err
 }
 
-func (d Deployment) ID() DeploymentID { return d.id }
-func (d Deployment) Config() Config   { return d.config }
-func (d Deployment) Source() Meta     { return d.source }
+func (d Deployment) ID() DeploymentID   { return d.id }
+func (d Deployment) Config() Config     { return d.config }
+func (d Deployment) Source() SourceData { return d.source }
 
 // Retrieve the deployment path relative to the given directories.
 func (d Deployment) Path(relativeTo ...string) string {