chore: version jobs go endpoint (#689)

Author: adityachoudhari26

apps/workspace-engine/oapi/openapi.json

@@ -1183,10 +1183,6 @@
                                           },
                                           "resourceId": {
                                              "type": "string"
-                                          },
-                                          "rolloutTime": {
-                                             "format": "date-time",
-                                             "type": "string"
                                           }
                                        },
                                        "type": "object"

apps/workspace-engine/oapi/spec/paths/deployment-version.jsonnet

@@ -57,8 +57,6 @@ local openapi = import '../lib/openapi.libsonnet';
                         },
                       },
                     },
-                    // Optional rollout time
-                    rolloutTime: { type: 'string', format: 'date-time' },
                   },
                 },
               },

apps/workspace-engine/pkg/db/changeset.go

@@ -5,12 +5,13 @@ import (
 	"fmt"
 	"workspace-engine/pkg/changeset"
 	"workspace-engine/pkg/oapi"
+	"workspace-engine/pkg/workspace/store"
 
 	"github.com/jackc/pgx/v5"
 	"go.opentelemetry.io/otel/attribute"
 )
 
-func FlushChangeset(ctx context.Context, cs *changeset.ChangeSet[any], workspaceID string) error {
+func FlushChangeset(ctx context.Context, cs *changeset.ChangeSet[any], workspaceID string, store *store.Store) error {
 	ctx, span := tracer.Start(ctx, "DBFlushChangeset")
 	defer span.End()
 
@@ -37,7 +38,7 @@ func FlushChangeset(ctx context.Context, cs *changeset.ChangeSet[any], workspace
 	defer func() { _ = tx.Rollback(ctx) }()
 
 	for _, change := range cs.Changes {
-		if err := applyChange(ctx, tx, change, workspaceID); err != nil {
+		if err := applyChange(ctx, tx, change, workspaceID, store); err != nil {
 			return err
 		}
 	}
@@ -51,7 +52,7 @@ func FlushChangeset(ctx context.Context, cs *changeset.ChangeSet[any], workspace
 	return nil
 }
 
-func applyChange(ctx context.Context, conn pgx.Tx, change changeset.Change[any], workspaceID string) error {
+func applyChange(ctx context.Context, conn pgx.Tx, change changeset.Change[any], workspaceID string, store *store.Store) error {
 	if e, ok := change.Entity.(*oapi.Resource); ok && e != nil {
 		if change.Type == changeset.ChangeTypeDelete {
 			return deleteResource(ctx, e.Id, conn)
@@ -133,7 +134,7 @@ func applyChange(ctx context.Context, conn pgx.Tx, change changeset.Change[any],
 		if change.Type == changeset.ChangeTypeDelete {
 			return deleteJob(ctx, e.Id, conn)
 		}
-		return writeJob(ctx, e, conn)
+		return writeJob(ctx, e, store, conn)
 	}
 
 	if e, ok := change.Entity.(*oapi.ReleaseTarget); ok && e != nil {
@@ -148,16 +149,18 @@ func applyChange(ctx context.Context, conn pgx.Tx, change changeset.Change[any],
 
 type DbChangesetConsumer struct {
 	workspaceID string
+	store       *store.Store
 }
 
 var _ changeset.ChangesetConsumer[any] = (*DbChangesetConsumer)(nil)
 
-func NewChangesetConsumer(workspaceID string) *DbChangesetConsumer {
+func NewChangesetConsumer(workspaceID string, store *store.Store) *DbChangesetConsumer {
 	return &DbChangesetConsumer{
 		workspaceID: workspaceID,
+		store:       store,
 	}
 }
 
 func (c *DbChangesetConsumer) FlushChangeset(ctx context.Context, changeset *changeset.ChangeSet[any]) error {
-	return FlushChangeset(ctx, changeset, c.workspaceID)
+	return FlushChangeset(ctx, changeset, c.workspaceID, c.store)
 }

apps/workspace-engine/pkg/db/jobs.go

@@ -2,8 +2,10 @@ package db
 
 import (
 	"context"
+	"fmt"
 	"time"
 	"workspace-engine/pkg/oapi"
+	"workspace-engine/pkg/workspace/store"
 
 	"github.com/jackc/pgx/v5"
 )
@@ -129,6 +131,20 @@ const JOB_UPSERT_QUERY = `
 		updated_at = EXCLUDED.updated_at
 `
 
+const RELEASE_JOB_CHECK_QUERY = `
+	SELECT EXISTS(SELECT 1 FROM release_job WHERE release_id = $1 AND job_id = $2)
+`
+
+const RELEASE_JOB_INSERT_QUERY = `
+	INSERT INTO release_job (release_id, job_id)
+	VALUES ($1, $2)
+`
+
+func writeReleaseJob(ctx context.Context, releaseId string, jobId string, tx pgx.Tx) error {
+	_, err := tx.Exec(ctx, RELEASE_JOB_INSERT_QUERY, releaseId, jobId)
+	return err
+}
+
 func convertOapiJobStatusToStr(status oapi.JobStatus) string {
 	switch status {
 	case oapi.Pending:
@@ -156,7 +172,11 @@ func convertOapiJobStatusToStr(status oapi.JobStatus) string {
 	}
 }
 
-func writeJob(ctx context.Context, job *oapi.Job, tx pgx.Tx) error {
+func writeJob(ctx context.Context, job *oapi.Job, store *store.Store, tx pgx.Tx) error {
+	release, ok := store.Releases.Get(job.ReleaseId)
+	if !ok {
+		return fmt.Errorf("release not found for job %s", job.Id)
+	}
 	statusStr := convertOapiJobStatusToStr(job.Status)
 	_, err := tx.Exec(
 		ctx,
@@ -170,7 +190,16 @@ func writeJob(ctx context.Context, job *oapi.Job, tx pgx.Tx) error {
 		job.StartedAt,
 		job.CompletedAt,
 		job.UpdatedAt)
-	return err
+	if err != nil {
+		return err
+	}
+
+	if job.ReleaseId != "" {
+		if err := writeReleaseJob(ctx, release.UUID().String(), job.Id, tx); err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 const DELETE_JOB_QUERY = `