chore: remove scheduled_jobs.resource_id

Use SQLite JSON functions to extract needed informations.

cmd/serve/front/src/lib/resources/jobs.ts

@@ -4,7 +4,6 @@ import type { Paginated } from '$lib/pagination';
 
 export type Job = {
 	id: string;
-	resource_id: string;
 	group: string;
 	message_name: string;
 	message_data: string;

cmd/serve/front/src/routes/(main)/jobs/+page.svelte

@@ -48,7 +48,6 @@
 		{:else if value === 'resource'}
 			<!-- @ts-ignore -->
 			<div>{translateMessageName(item.message_name)}</div>
-			<div class="meta">{item.resource_id}</div>
 		{/if}
 	</svelte:fragment>
 

internal/deployment/app/cleanup_app/cleanup_app.go

@@ -25,9 +25,8 @@ type Command struct {
 	To          time.Time `json:"to"`
 }
 
-func (Command) Name_() string        { return "deployment.command.cleanup_app" }
-func (c Command) ResourceID() string { return c.AppID }
-func (c Command) Group() string      { return bus.Group(c.AppID, c.Environment, c.TargetID) }
+func (Command) Name_() string   { return "deployment.command.cleanup_app" }
+func (c Command) Group() string { return bus.Group(c.AppID, c.Environment, c.TargetID) }
 
 func Handler(
 	reader domain.TargetsReader,

internal/deployment/app/cleanup_app/cleanup_app_test.go

@@ -108,7 +108,7 @@ func Test_CleanupApp(t *testing.T) {
 		var provider mockProvider
 		user := authfixture.User()
 		target := fixture.Target(fixture.WithTargetCreatedBy(user.ID()))
-		target.Configured(target.CurrentVersion(), nil, nil)
+		assert.Nil(t, target.Configured(target.CurrentVersion(), nil, nil))
 		assert.Nil(t, target.RequestDelete(false, "uid"))
 
 		handler, ctx := arrange(t, &provider,
@@ -146,7 +146,7 @@ func Test_CleanupApp(t *testing.T) {
 		var provider mockProvider
 		user := authfixture.User()
 		target := fixture.Target(fixture.WithTargetCreatedBy(user.ID()))
-		target.Configured(target.CurrentVersion(), nil, errors.New("configuration failed"))
+		assert.Nil(t, target.Configured(target.CurrentVersion(), nil, errors.New("configuration failed")))
 		app := fixture.App(fixture.WithAppCreatedBy(user.ID()),
 			fixture.WithEnvironmentConfig(
 				domain.NewEnvironmentConfig(target.ID()),
@@ -181,7 +181,7 @@ func Test_CleanupApp(t *testing.T) {
 		var provider mockProvider
 		user := authfixture.User()
 		target := fixture.Target(fixture.WithTargetCreatedBy(user.ID()))
-		target.Configured(target.CurrentVersion(), nil, nil)
+		assert.Nil(t, target.Configured(target.CurrentVersion(), nil, nil))
 		app := fixture.App(fixture.WithAppCreatedBy(user.ID()),
 			fixture.WithEnvironmentConfig(
 				domain.NewEnvironmentConfig(target.ID()),

internal/deployment/app/cleanup_app/on_app_cleanup_requested.go

@@ -13,22 +13,21 @@ func OnAppCleanupRequestedHandler(scheduler bus.Scheduler) bus.SignalHandler[dom
 	return func(ctx context.Context, evt domain.AppCleanupRequested) error {
 		now := time.Now().UTC()
 
-		if err := scheduler.Queue(ctx, Command{
-			AppID:       string(evt.ID),
-			Environment: string(domain.Production),
-			TargetID:    string(evt.ProductionConfig.Target()),
-			From:        evt.ProductionConfig.Version(),
-			To:          now,
-		}); err != nil {
-			return err
-		}
-
-		return scheduler.Queue(ctx, Command{
-			AppID:       string(evt.ID),
-			Environment: string(domain.Staging),
-			TargetID:    string(evt.StagingConfig.Target()),
-			From:        evt.StagingConfig.Version(),
-			To:          now,
-		})
+		return scheduler.Queue(ctx,
+			Command{
+				AppID:       string(evt.ID),
+				Environment: string(domain.Production),
+				TargetID:    string(evt.ProductionConfig.Target()),
+				From:        evt.ProductionConfig.Version(),
+				To:          now,
+			},
+			Command{
+				AppID:       string(evt.ID),
+				Environment: string(domain.Staging),
+				TargetID:    string(evt.StagingConfig.Target()),
+				From:        evt.StagingConfig.Version(),
+				To:          now,
+			},
+		)
 	}
 }

internal/deployment/app/cleanup_target/cleanup_target.go

@@ -14,12 +14,11 @@ import (
 type Command struct {
 	bus.AsyncCommand
 
-	ID string `json:"id"`
+	ID string `json:"target_id"`
 }
 
-func (Command) Name_() string        { return "deployment.command.cleanup_target" }
-func (c Command) ResourceID() string { return c.ID }
-func (c Command) Group() string      { return c.ID }
+func (Command) Name_() string   { return "deployment.command.cleanup_target" }
+func (c Command) Group() string { return c.ID }
 
 func Handler(
 	reader domain.TargetsReader,

internal/deployment/app/cleanup_target/on_job_dismissed.go

@@ -15,11 +15,13 @@ func OnJobDismissedHandler(
 	writer domain.TargetsWriter,
 ) bus.SignalHandler[embedded.JobDismissed] {
 	return func(ctx context.Context, evt embedded.JobDismissed) error {
-		if _, isCleanupJob := evt.Command.(Command); !isCleanupJob {
+		cmd, isCleanupJob := evt.Command.(Command)
+
+		if !isCleanupJob {
 			return nil
 		}
 
-		target, err := reader.GetByID(ctx, domain.TargetID(evt.Command.ResourceID()))
+		target, err := reader.GetByID(ctx, domain.TargetID(cmd.ID))
 
 		if err != nil {
 			// Target deleted, no need to go further

internal/deployment/app/configure_target/configure_target.go

@@ -14,13 +14,12 @@ import (
 type Command struct {
 	bus.AsyncCommand
 
-	ID      string    `json:"id"`
+	ID      string    `json:"target_id"`
 	Version time.Time `json:"version"`
 }
 
-func (Command) Name_() string        { return "deployment.command.configure_target" }
-func (c Command) ResourceID() string { return c.ID }
-func (c Command) Group() string      { return c.ID }
+func (Command) Name_() string   { return "deployment.command.configure_target" }
+func (c Command) Group() string { return c.ID }
 
 func Handler(
 	reader domain.TargetsReader,

internal/deployment/app/deploy/deploy.go

@@ -3,7 +3,6 @@ package deploy
 import (
 	"context"
 	"errors"
-	"strconv"
 
 	"github.com/YuukanOO/seelf/internal/deployment/domain"
 	"github.com/YuukanOO/seelf/pkg/apperr"
@@ -20,9 +19,8 @@ type Command struct {
 	TargetID         string `json:"target_id"`
 }
 
-func (Command) Name_() string        { return "deployment.command.deploy" }
-func (c Command) ResourceID() string { return c.AppID + "-" + strconv.Itoa(c.DeploymentNumber) }
-func (c Command) Group() string      { return bus.Group(c.AppID, c.Environment, c.TargetID) }
+func (Command) Name_() string   { return "deployment.command.deploy" }
+func (c Command) Group() string { return bus.Group(c.AppID, c.Environment, c.TargetID) }
 
 // Handle the deployment process.
 // If an unexpected error occurs during this process, it uses the bus.PreserveOrder function

internal/deployment/infra/sqlite/migrations/1728345059_add_app_cleanup_needed_on.up.sql

@@ -12,26 +12,48 @@ ALTER TABLE apps ADD version DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP;
 ALTER TABLE registries ADD version DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP;
 ALTER TABLE targets ADD version DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP;
 
-ALTER TABLE apps ADD history TEXT NOT NULL DEFAULT '{}';
+ALTER TABLE apps ADD history JSON NOT NULL DEFAULT '{}';
 UPDATE apps SET history = '{"production": ["' || production_target || '"], "staging": ["'|| staging_target ||'"]}';
 
 -- Since the group has changed for configure and cleanup, just update those jobs.
 UPDATE scheduled_jobs
 SET
     [group] = resource_id
-WHERE message_name IN ('deployment.command.configure_target', 'deployment.command.cleanup_target');
+    ,message_data = '{"target_id": "' || resource_id || '"}'
+WHERE message_name = 'deployment.command.cleanup_target';
+
+UPDATE scheduled_jobs
+SET
+    [group] = resource_id
+    ,message_data = '{"target_id": "' || resource_id || '", "version": "' || (message_data ->> '$.version') || '"}'
+WHERE message_name = 'deployment.command.configure_target';
 
 -- Since those messages no longer exists.
 DELETE FROM scheduled_jobs WHERE message_name IN ('deployment.command.delete_target', 'deployment.command.delete_app');
 
+DROP TRIGGER IF EXISTS on_deployment_failed_cleanup_jobs;
+
+-- When a deployment is marked as failed, remove all pending jobs for that resource.
+-- This is to avoid running jobs that are no longer needed.
+CREATE TRIGGER IF NOT EXISTS on_deployment_failed_cleanup_jobs AFTER UPDATE ON deployments
+    WHEN OLD.state_status != NEW.state_status AND NEW.state_status = 2 -- Only when the deployment goes to the failed state
+BEGIN
+    DELETE FROM scheduled_jobs
+    WHERE
+        message_name = 'deployment.command.deploy'
+        AND message_data ->> '$.app_id' = NEW.app_id
+        AND message_data ->> '$.deployment_number' = NEW.deployment_number
+        AND retrieved = 0;
+END;
+
 -- When a target is configured, no need to keep old configure jobs since they are outdated.
 CREATE TRIGGER IF NOT EXISTS on_target_configure_remove_outdated_jobs
 BEFORE INSERT ON scheduled_jobs
 WHEN NEW.message_name = 'deployment.command.configure_target'
 BEGIN
   DELETE FROM scheduled_jobs
   WHERE
-      resource_id = NEW.resource_id
-      AND message_name = 'deployment.command.configure_target'
+      message_name = 'deployment.command.configure_target'
+      AND message_data ->> '$.target_id' = NEW.message_data ->> '$.target_id'
       AND retrieved = 0;
-END
+END;

pkg/bus/embedded/get_jobs/get_jobs.go

@@ -11,7 +11,6 @@ import (
 type (
 	Job struct {
 		ID          string              `json:"id"`
-		ResourceID  string              `json:"resource_id"`
 		Group       string              `json:"group"`
 		MessageName string              `json:"message_name"`
 		MessageData string              `json:"message_data"`

pkg/bus/embedded/runner_test.go

@@ -126,17 +126,15 @@ type returnCommand struct {
 	err    error
 }
 
-func (r returnCommand) Name_() string      { return "returnCommand" }
-func (r returnCommand) ResourceID() string { return "" }
-func (r returnCommand) Group() string      { return "" }
+func (r returnCommand) Name_() string { return "returnCommand" }
+func (r returnCommand) Group() string { return "" }
 
 type unhandledCommand struct {
 	bus.AsyncCommand
 }
 
-func (u unhandledCommand) Name_() string      { return "unhandledCommand" }
-func (u unhandledCommand) ResourceID() string { return "" }
-func (u unhandledCommand) Group() string      { return "" }
+func (u unhandledCommand) Name_() string { return "unhandledCommand" }
+func (u unhandledCommand) Group() string { return "" }
 
 var (
 	_ embedded.JobsStore = (*adapter)(nil)
@@ -145,12 +143,12 @@ var (
 
 type job struct {
 	id  string
-	msg bus.AsyncRequest
+	cmd bus.AsyncRequest
 	err error
 }
 
 func (j *job) ID() string                { return j.id }
-func (j *job) Command() bus.AsyncRequest { return j.msg }
+func (j *job) Command() bus.AsyncRequest { return j.cmd }
 
 type adapter struct {
 	wg      sync.WaitGroup
@@ -160,9 +158,13 @@ type adapter struct {
 	done    []*job
 }
 
-func (a *adapter) Queue(ctx context.Context, msg bus.AsyncRequest) error {
+func (a *adapter) Queue(ctx context.Context, requests ...bus.AsyncRequest) error {
 	a.wg.Add(1)
-	a.jobs = append(a.jobs, &job{id: id.New[string](), msg: msg})
+
+	for _, req := range requests {
+		a.jobs = append(a.jobs, &job{id: id.New[string](), cmd: req})
+	}
+
 	return nil
 }
 

pkg/bus/message.go

@@ -65,8 +65,7 @@ type (
 	AsyncRequest interface {
 		MutateRequest
 		TypedRequest[AsyncResult]
-		ResourceID() string // ID of the main resource processed by the request
-		Group() string      // Work group for this request, at most one job per group is processed at any given time
+		Group() string // Work group for this request, at most one job per group is processed at any given time
 	}
 
 	// Request to query the system.

pkg/bus/message_test.go

@@ -181,9 +181,8 @@ type asyncCommand struct {
 	SomeValue int
 }
 
-func (asyncCommand) Name_() string      { return "AsyncCommand" }
-func (asyncCommand) ResourceID() string { return "" }
-func (asyncCommand) Group() string      { return "" }
+func (asyncCommand) Name_() string { return "AsyncCommand" }
+func (asyncCommand) Group() string { return "" }
 
 type getQuery struct {
 	bus.Query[int]

pkg/bus/scheduler.go

@@ -5,5 +5,5 @@ import "context"
 // Enable scheduled dispatching of a message.
 type Scheduler interface {
 	// Queue a request to be dispatched asynchronously at a later time.
-	Queue(context.Context, AsyncRequest) error
+	Queue(context.Context, ...AsyncRequest) error
 }

pkg/bus/sqlite/jobs.go

@@ -58,7 +58,6 @@ func (s *JobsStore) GetAllJobs(ctx context.Context, query get_jobs.Query) (stora
 	return builder.
 		Select[get_jobs.Job](`
 			id
-			,resource_id
 			,[group]
 			,message_name
 			,message_data
@@ -124,27 +123,33 @@ func (s *JobsStore) DismissJob(ctx context.Context, cmd dismiss_job.Command) (bu
 
 func (s *JobsStore) Queue(
 	ctx context.Context,
-	msg bus.AsyncRequest,
+	requests ...bus.AsyncRequest,
 ) error {
 	now := time.Now().UTC()
-	msgData, err := storage.ValueJSON(msg)
 
-	if err != nil {
-		return err
+	for _, req := range requests {
+		cmdData, err := storage.ValueJSON(req)
+
+		if err != nil {
+			return err
+		}
+
+		if err = builder.
+			Insert("scheduled_jobs", builder.Values{
+				"id":           id.New[string](),
+				"[group]":      req.Group(),
+				"message_name": req.Name_(),
+				"message_data": cmdData,
+				"queued_at":    now,
+				"not_before":   now,
+				"retrieved":    false,
+			}).
+			Exec(s.db, ctx); err != nil {
+			return err
+		}
 	}
 
-	return builder.
-		Insert("scheduled_jobs", builder.Values{
-			"id":           id.New[string](),
-			"resource_id":  msg.ResourceID(),
-			"[group]":      msg.Group(),
-			"message_name": msg.Name_(),
-			"message_data": msgData,
-			"queued_at":    now,
-			"not_before":   now,
-			"retrieved":    false,
-		}).
-		Exec(s.db, ctx)
+	return nil
 }
 
 func (s *JobsStore) GetNextPendingJobs(ctx context.Context) ([]embedded.Job, error) {
@@ -226,7 +231,6 @@ func jobMapper(scanner storage.Scanner) (embedded.Job, error) {
 func jobQueryMapper(scanner storage.Scanner) (j get_jobs.Job, err error) {
 	err = scanner.Scan(
 		&j.ID,
-		&j.ResourceID,
 		&j.Group,
 		&j.MessageName,
 		&j.MessageData,