feat: Init release upsert in workers

apps/event-worker/src/workers/new-deployment-version.ts

@@ -1,10 +1,12 @@
 import type { Tx } from "@ctrlplane/db";
+import type { ReleaseTargetIdentifier } from "@ctrlplane/rule-engine";
 import _ from "lodash";
 
 import { and, eq, isNull } from "@ctrlplane/db";
 import { db } from "@ctrlplane/db/client";
 import * as schema from "@ctrlplane/db/schema";
 import { Channel, createWorker } from "@ctrlplane/events";
+import { DatabaseReleaseRepository } from "@ctrlplane/rule-engine";
 
 const getDeploymentResources = async (
   tx: Tx,
@@ -56,21 +58,51 @@ export const newDeploymentVersionWorker = createWorker(
   async ({ data: version }) => {
     const deployment = await db.query.deployment.findFirst({
       where: eq(schema.deployment.id, version.deploymentId),
+      with: { system: true },
     });
 
     if (!deployment) throw new Error("Deployment not found");
 
     const resources = await getDeploymentResources(db, deployment);
 
-    const releaseTargets = resources.map((resource) => ({
+    const releaseTargetInserts = resources.map((resource) => ({
       resourceId: resource.id,
       environmentId: resource.environment.id,
       deploymentId: version.deploymentId,
     }));
 
-    await db
+    const releaseTargets = await db
       .insert(schema.releaseTarget)
-      .values(releaseTargets)
-      .onConflictDoNothing();
+      .values(releaseTargetInserts)
+      .onConflictDoNothing()
+      .returning();
+
+    const { system } = deployment;
+    const { workspaceId } = system;
+
+    const promises = releaseTargets.map(async (rt) => {
+      const rtWithWorkspaceId = { ...rt, workspaceId };
+      const repo = await DatabaseReleaseRepository.create(rtWithWorkspaceId);
+      const identifier: ReleaseTargetIdentifier = {
+        deploymentId: version.deploymentId,
+        environmentId: rt.environmentId,
+        resourceId: rt.resourceId,
+      };
+
+      const variables = await repo.getLatestVariables();
+      const { created, release } = await repo.upsert(
+        identifier,
+        version.id,
+        variables,
+      );
+      if (!created) return;
+
+      const newestRelease = await repo.getNewestRelease();
+      if (newestRelease?.id !== release.id) return;
+
+      await repo.setDesired(release.id);
+    });
+
+    await Promise.all(promises);
   },
 );

packages/rule-engine/src/releases/repositories/release-repository.ts

@@ -180,12 +180,10 @@ export class DatabaseReleaseRepository implements ReleaseRepository {
    * Sets the desired release for this release target
    * @param options - The release target identifier and desired release ID
    */
-  async setDesired(
-    options: ReleaseTargetIdentifier & { desiredReleaseId: string },
-  ) {
+  async setDesired(desiredReleaseId: string) {
     await this.db
       .update(schema.releaseTarget)
-      .set({ desiredReleaseId: options.desiredReleaseId })
+      .set({ desiredReleaseId })
       .where(eq(schema.releaseTarget.id, this.releaseTarget.id));
   }
 }

packages/rule-engine/src/releases/repositories/types.ts

@@ -26,7 +26,5 @@ export interface ReleaseRepository {
   /**
    * Set a specific release as the desired release
    */
-  setDesired(
-    options: ReleaseTargetIdentifier & { desiredReleaseId: string },
-  ): Promise<void>;
+  setDesired(desiredReleaseId: string): Promise<void>;
 }