fix: New targets trigger release

apps/event-worker/package.json

@@ -13,6 +13,7 @@
   },
   "dependencies": {
     "@ctrlplane/db": "workspace:*",
+    "@ctrlplane/job-dispatch": "workspace:*",
     "@ctrlplane/logger": "workspace:*",
     "@ctrlplane/validators": "workspace:*",
     "@google-cloud/container": "^5.16.0",

apps/event-worker/src/target-scan/upsert.ts

@@ -3,6 +3,14 @@ import type { Target } from "@ctrlplane/db/schema";
 
 import { buildConflictUpdateColumns, sql } from "@ctrlplane/db";
 import { target } from "@ctrlplane/db/schema";
+import {
+  cancelOldJobConfigsOnJobDispatch,
+  createJobConfigs,
+  createJobExecutionApprovals,
+  dispatchJobConfigs,
+  isPassingAllPolicies,
+  isPassingReleaseSequencingCancelPolicy,
+} from "@ctrlplane/job-dispatch";
 
 export type UpsertTarget = Pick<
   Target,
@@ -25,4 +33,18 @@ export const upsertTargets = (db: Tx, providerId: string, ts: UpsertTarget[]) =>
       setWhere: sql`target.provider_id = ${providerId}`,
       set: buildConflictUpdateColumns(target, ["labels"]),
     })
-    .returning();
+    .returning()
+    .then((targets) =>
+      createJobConfigs(db, "new_target")
+        .targets(targets.map((t) => t.id))
+        .filterAsync(isPassingReleaseSequencingCancelPolicy)
+        .then(createJobExecutionApprovals)
+        .insert()
+        .then((jobConfigs) =>
+          dispatchJobConfigs(db)
+            .jobConfigs(jobConfigs)
+            .filter(isPassingAllPolicies)
+            .then(cancelOldJobConfigsOnJobDispatch)
+            .dispatch(),
+        ),
+    );

apps/webservice/src/app/[workspaceSlug]/settings/(settings)/workspace/integrations/(integration)/github/SelectPreconnectedOrgDialogContent.tsx

@@ -39,24 +39,34 @@ export const SelectPreconnectedOrgDialogContent: React.FC<
   const router = useRouter();
 
   const githubOrgCreate = api.github.organizations.create.useMutation();
+  const jobAgentCreate = api.job.agent.create.useMutation();
 
   const handleSave = () => {
     if (value == null) return;
     const org = githubOrgs.find((o) => o.login === value);
     if (org == null) return;
 
-    githubOrgCreate
-      .mutateAsync({
+    Promise.all([
+      githubOrgCreate.mutateAsync({
         installationId: org.installationId,
         workspaceId,
         organizationName: org.login,
         addedByUserId: githubUser.userId,
         avatarUrl: org.avatar_url,
-      })
-      .then(() => {
-        onSave();
-        router.refresh();
-      });
+      }),
+      jobAgentCreate.mutateAsync({
+        workspaceId,
+        type: "github-app",
+        name: org.login,
+        config: {
+          installationId: org.installationId,
+          owner: org.login,
+        },
+      }),
+    ]).then(() => {
+      onSave();
+      router.refresh();
+    });
   };
 
   return (

apps/webservice/src/app/[workspaceSlug]/systems/[systemSlug]/deployments/[deploymentSlug]/configure/job-agent/page.tsx

@@ -57,7 +57,6 @@ export default function ConfigureJobAgentPage({
       jobAgentId: "",
       jobAgentConfig: {},
     },
-    disabled: deployment.isLoading || jobAgents.isLoading,
   });
 
   const { jobAgentId } = form.watch();

apps/webservice/src/app/api/v1/[workspace]/target-provider/[providerId]/set/route.ts

@@ -11,6 +11,14 @@ import {
   targetProvider,
   workspace,
 } from "@ctrlplane/db/schema";
+import {
+  cancelOldJobConfigsOnJobDispatch,
+  createJobConfigs,
+  createJobExecutionApprovals,
+  dispatchJobConfigs,
+  isPassingAllPolicies,
+  isPassingReleaseSequencingCancelPolicy,
+} from "@ctrlplane/job-dispatch";
 import { Permission } from "@ctrlplane/validators/auth";
 
 import { getUser } from "~/app/api/v1/auth";
@@ -56,7 +64,22 @@ export const PATCH = async (
       target: [target.name, target.providerId],
       set: buildConflictUpdateColumns(target, ["labels"]),
     })
-    .returning();
+    .returning()
+    .then((targets) =>
+      createJobConfigs(db, "new_target")
+        .causedById(user.id)
+        .targets(targets.map((t) => t.id))
+        .filterAsync(isPassingReleaseSequencingCancelPolicy)
+        .then(createJobExecutionApprovals)
+        .insert()
+        .then((jobConfigs) =>
+          dispatchJobConfigs(db)
+            .jobConfigs(jobConfigs)
+            .filter(isPassingAllPolicies)
+            .then(cancelOldJobConfigsOnJobDispatch)
+            .dispatch(),
+        ),
+    );
 
   return NextResponse.json({ targets: results });
 };

packages/api/src/latest-release-subquery.ts

@@ -0,0 +1,19 @@
+import type { Tx } from "@ctrlplane/db";
+
+import { sql } from "@ctrlplane/db";
+import { release } from "@ctrlplane/db/schema";
+
+export const latestReleaseSubQuery = (db: Tx) =>
+  db
+    .select({
+      id: release.id,
+      deploymentId: release.deploymentId,
+      version: release.version,
+      createdAt: release.createdAt,
+
+      rank: sql<number>`ROW_NUMBER() OVER (PARTITION BY deployment_id ORDER BY created_at DESC)`.as(
+        "rank",
+      ),
+    })
+    .from(release)
+    .as("release");

packages/api/src/router/deployment.ts

@@ -1,4 +1,3 @@
-import type { Tx } from "@ctrlplane/db";
 import { z } from "zod";
 
 import {
@@ -32,24 +31,10 @@ import {
 } from "@ctrlplane/job-dispatch";
 import { Permission } from "@ctrlplane/validators/auth";
 
+import { latestReleaseSubQuery } from "../latest-release-subquery";
 import { createTRPCRouter, protectedProcedure } from "../trpc";
 import { deploymentVariableRouter } from "./deployment-variable";
 
-const latestReleaseSubQuery = (db: Tx) =>
-  db
-    .select({
-      id: release.id,
-      deploymentId: release.deploymentId,
-      version: release.version,
-      createdAt: release.createdAt,
-
-      rank: sql<number>`ROW_NUMBER() OVER (PARTITION BY deployment_id ORDER BY created_at DESC)`.as(
-        "rank",
-      ),
-    })
-    .from(release)
-    .as("release");
-
 export const deploymentRouter = createTRPCRouter({
   variable: deploymentVariableRouter,
   distrubtionById: protectedProcedure

packages/api/src/router/environment.ts

@@ -40,6 +40,7 @@ import {
 } from "@ctrlplane/job-dispatch";
 import { Permission } from "@ctrlplane/validators/auth";
 
+import { latestReleaseSubQuery } from "../latest-release-subquery";
 import { createTRPCRouter, protectedProcedure } from "../trpc";
 
 const policyRouter = createTRPCRouter({
@@ -470,7 +471,7 @@ export const environmentRouter = createTRPCRouter({
         .causedById(ctx.session.user.id)
         .environments([env.id])
         .releases([rel.release.id])
-        .filter(isPassingReleaseSequencingCancelPolicy)
+        .filterAsync(isPassingReleaseSequencingCancelPolicy)
         .then(createJobExecutionApprovals)
         .insert();
 
@@ -563,14 +564,70 @@ export const environmentRouter = createTRPCRouter({
           .on({ type: "environment", id: input.id }),
     })
     .input(z.object({ id: z.string().uuid(), data: updateEnvironment }))
-    .mutation(({ ctx, input }) =>
-      ctx.db
+    .mutation(async ({ ctx, input }) => {
+      const latestRelease = latestReleaseSubQuery(ctx.db);
+      /*
+       * When updating the labels of an environment and retriggering releases
+       * we should exclude targets that were already part of the environment (would lead to uninteded duplicate job configs).
+       * We need to check if there is a job config matching the release and target
+       * in this environment that is not associated with a runbook.
+       */
+      const existingJobConfigs = await ctx.db
+        .select()
+        .from(deployment)
+        .innerJoin(
+          latestRelease,
+          and(
+            eq(latestRelease.deploymentId, deployment.id),
+            eq(latestRelease.rank, 1),
+          ),
+        )
+        .innerJoin(system, eq(system.id, deployment.systemId))
+        .innerJoin(environment, eq(environment.systemId, system.id))
+        .innerJoin(target, eq(target.workspaceId, system.workspaceId))
+        .innerJoin(
+          jobConfig,
+          and(
+            eq(jobConfig.environmentId, environment.id),
+            eq(jobConfig.releaseId, latestRelease.id),
+            eq(jobConfig.targetId, target.id),
+            isNull(jobConfig.runbookId),
+          ),
+        )
+        .where(eq(environment.id, input.id));
+
+      return ctx.db
         .update(environment)
         .set(input.data)
         .where(eq(environment.id, input.id))
         .returning()
-        .then(takeFirst),
-    ),
+        .then(takeFirst)
+        .then((env) =>
+          createJobConfigs(ctx.db, "new_target")
+            .environments([env.id])
+            .filter((jobConfigsInserts) =>
+              jobConfigsInserts.filter(
+                (jobConfigInsert) =>
+                  !existingJobConfigs.some(
+                    (existingJobConfig) =>
+                      existingJobConfig.release.id ===
+                        jobConfigInsert.releaseId &&
+                      existingJobConfig.target.id === jobConfigInsert.targetId,
+                  ),
+              ),
+            )
+            .filterAsync(isPassingReleaseSequencingCancelPolicy)
+            .then(createJobExecutionApprovals)
+            .insert()
+            .then((jobConfigs) =>
+              dispatchJobConfigs(ctx.db)
+                .jobConfigs(jobConfigs)
+                .filter(isPassingAllPolicies)
+                .then(cancelOldJobConfigsOnJobDispatch)
+                .dispatch(),
+            ),
+        );
+    }),
 
   delete: protectedProcedure
     .meta({

packages/api/src/router/release.ts

@@ -115,7 +115,8 @@ export const releaseRouter = createTRPCRouter({
           .causedById(ctx.session.user.id)
           .environments([input.environmentId])
           .releases([input.releaseId])
-          .filter(isPassingReleaseSequencingCancelPolicy)
+          .filterAsync(isPassingReleaseSequencingCancelPolicy)
+          .then(createJobExecutionApprovals)
           .insert();
 
         await dispatchJobConfigs(ctx.db)
@@ -154,9 +155,9 @@ export const releaseRouter = createTRPCRouter({
 
         const jobConfigs = await createJobConfigs(db, "new_release")
           .causedById(ctx.session.user.id)
-          .filter(isPassingEnvironmentPolicy)
+          .filterAsync(isPassingEnvironmentPolicy)
           .releases([rel.id])
-          .filter(isPassingReleaseSequencingCancelPolicy)
+          .filterAsync(isPassingReleaseSequencingCancelPolicy)
           .then(createJobExecutionApprovals)
           .insert();
 

packages/api/src/router/target.ts

@@ -21,6 +21,14 @@ import {
   updateTarget,
   workspace,
 } from "@ctrlplane/db/schema";
+import {
+  cancelOldJobConfigsOnJobDispatch,
+  createJobConfigs,
+  createJobExecutionApprovals,
+  dispatchJobConfigs,
+  isPassingAllPolicies,
+  isPassingReleaseSequencingCancelPolicy,
+} from "@ctrlplane/job-dispatch";
 import { Permission } from "@ctrlplane/validators/auth";
 
 import { createTRPCRouter, protectedProcedure } from "../trpc";
@@ -142,7 +150,26 @@ export const targetRouter = createTRPCRouter({
     })
     .input(createTarget)
     .mutation(({ ctx, input }) =>
-      ctx.db.insert(target).values(input).returning().then(takeFirst),
+      ctx.db
+        .insert(target)
+        .values(input)
+        .returning()
+        .then(takeFirst)
+        .then((target) =>
+          createJobConfigs(ctx.db, "new_target")
+            .causedById(ctx.session.user.id)
+            .targets([target.id])
+            .filterAsync(isPassingReleaseSequencingCancelPolicy)
+            .then(createJobExecutionApprovals)
+            .insert()
+            .then((jobConfigs) =>
+              dispatchJobConfigs(ctx.db)
+                .jobConfigs(jobConfigs)
+                .filter(isPassingAllPolicies)
+                .then(cancelOldJobConfigsOnJobDispatch)
+                .dispatch(),
+            ),
+        ),
     ),
 
   update: protectedProcedure

packages/db/drizzle.config.ts

@@ -12,4 +12,5 @@ export default defineConfig({
   verbose: true,
   dbCredentials: { url: nonPoolingUrl },
   tablesFilter: ["t3turbo_*"],
+  out: "drizzle",
 });

packages/job-dispatch/src/job-config.ts

@@ -15,7 +15,9 @@ import {
   target,
 } from "@ctrlplane/db/schema";
 
-type FilterFunc = (
+type FilterFunc = (insertJobConfigs: JobConfigInsert[]) => JobConfigInsert[];
+
+type AsyncFilterFunc = (
   tx: Tx,
   insertJobConfigs: JobConfigInsert[],
 ) => Promise<JobConfigInsert[]>;
@@ -33,6 +35,7 @@ class JobConfigBuilder {
   private releaseIds?: string[];
 
   private _filters: FilterFunc[] = [];
+  private _asyncFilters: AsyncFilterFunc[] = [];
   private _then: ThenFunc[] = [];
 
   constructor(
@@ -50,6 +53,11 @@ class JobConfigBuilder {
     return this;
   }
 
+  filterAsync(fn: AsyncFilterFunc) {
+    this._asyncFilters.push(fn);
+    return this;
+  }
+
   then(fn: ThenFunc) {
     this._then.push(fn);
     return this;
@@ -134,7 +142,8 @@ class JobConfigBuilder {
       releaseId: v.release.id,
     }));
 
-    for (const func of this._filters) wt = await func(this.tx, wt);
+    for (const func of this._filters) wt = func(wt);
+    for (const func of this._asyncFilters) wt = await func(this.tx, wt);
 
     if (wt.length === 0) return [];
 

packages/job-dispatch/src/policy-create.ts

@@ -15,6 +15,7 @@ export const createJobExecutionApprovals = async (
   db: Tx,
   jobConfigs: JobConfig[],
 ) => {
+  if (jobConfigs.length === 0) return;
   const policiesToCheck = await db
     .selectDistinctOn([release.id, environmentPolicy.id])
     .from(jobConfig)