fix(saga): Update sql task with saga IDs (#4062)

Tasks are created prior to the atomic operation processor operation.  In order to bridge atomic operations and sagas, we store the saga IDs when added to the SqlTask during the bridge process.

clouddriver-core/src/main/groovy/com/netflix/spinnaker/clouddriver/data/task/SagaId.java

@@ -15,11 +15,19 @@
  */
 package com.netflix.spinnaker.clouddriver.data.task;
 
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonPOJOBuilder;
 import javax.annotation.Nonnull;
+import lombok.Builder;
 import lombok.Value;
 
 @Value
+@Builder(builderClassName = "SagaIdBuilder")
+@JsonDeserialize(builder = SagaId.SagaIdBuilder.class)
 public class SagaId {
   @Nonnull String name;
   @Nonnull String id;
+
+  @JsonPOJOBuilder(withPrefix = "")
+  public static class SagaIdBuilder {}
 }

clouddriver-core/src/main/groovy/com/netflix/spinnaker/clouddriver/orchestration/sagas/SagaAtomicOperationBridge.java

@@ -44,7 +44,7 @@ public <T> T apply(@Nonnull ApplyCommandWrapper applyCommand) {
     final String sagaName = applyCommand.sagaName;
     final String sagaId = Optional.ofNullable(task.getRequestId()).orElse(task.getId());
 
-    task.addSagaId(new SagaId(sagaName, sagaId));
+    task.addSagaId(SagaId.builder().id(sagaId).name(sagaName).build());
 
     applyCommand.sagaFlow.injectFirst(SnapshotAtomicOperationInput.class);
 

clouddriver-sql/src/main/kotlin/com/netflix/spinnaker/clouddriver/sql/SqlTask.kt

@@ -101,6 +101,7 @@ class SqlTask(
   override fun addSagaId(sagaId: SagaId) {
     this.dirty.set(true)
     sagaIds.add(sagaId)
+    repository.updateSagaIds(this)
   }
 
   override fun getSagaIds(): MutableList<SagaId> {

clouddriver-sql/src/main/kotlin/com/netflix/spinnaker/clouddriver/sql/SqlTaskRepository.kt

@@ -84,6 +84,17 @@ class SqlTaskRepository(
     return task
   }
 
+  fun updateSagaIds(task: Task) {
+    return withPool(POOL_NAME) {
+      jooq.transactional(sqlRetryProperties.transactions) { ctx ->
+        ctx.update(tasksTable)
+          .set(field("saga_ids"), mapper.writeValueAsString(task.sagaIds))
+          .where(field("id").eq(task.id))
+          .execute()
+      }
+    }
+  }
+
   override fun get(id: String): Task? {
     return retrieveInternal(id)
   }