init new policy engine

apps/event-worker/package.json

@@ -20,6 +20,7 @@
     "@ctrlplane/db": "workspace:*",
     "@ctrlplane/job-dispatch": "workspace:*",
     "@ctrlplane/logger": "workspace:*",
+    "@ctrlplane/rule-engine": "workspace:*",
     "@ctrlplane/validators": "workspace:*",
     "@google-cloud/compute": "^4.9.0",
     "@google-cloud/container": "^5.16.0",
@@ -36,6 +37,7 @@
     "js-yaml": "^4.1.0",
     "lodash": "catalog:",
     "ms": "^2.1.3",
+    "redis-semaphore": "^5.6.2",
     "semver": "catalog:",
     "ts-is-present": "^1.2.2",
     "uuid": "^10.0.0",

apps/event-worker/src/rule-engine-evaluation/index.ts

@@ -0,0 +1,150 @@
+import type {
+  DeploymentResourceContext,
+  Release,
+} from "@ctrlplane/rule-engine";
+import type { RuleEngineEvaluationEvent } from "@ctrlplane/validators/events";
+import { Worker } from "bullmq";
+import { Mutex } from "redis-semaphore";
+
+import { and, desc, eq, sql, takeFirstOrNull } from "@ctrlplane/db";
+import { db } from "@ctrlplane/db/client";
+import * as schema from "@ctrlplane/db/schema";
+import {
+  Releases,
+  RuleEngine,
+  VersionCooldownRule,
+} from "@ctrlplane/rule-engine";
+import { Channel } from "@ctrlplane/validators/events";
+
+import { redis } from "../redis.js";
+
+const createDeploymentResourceContext = ({
+  resourceId,
+  deploymentId,
+  environmentId,
+}: RuleEngineEvaluationEvent) => {
+  return db
+    .select({
+      desiredReleaseId: schema.resourceDesiredRelease.desiredReleaseId,
+      deployment: schema.deployment,
+      environment: schema.environment,
+      resource: schema.resource,
+    })
+    .from(schema.resourceDesiredRelease)
+    .innerJoin(
+      schema.deployment,
+      eq(schema.resourceDesiredRelease.deploymentId, schema.deployment.id),
+    )
+    .innerJoin(
+      schema.environment,
+      eq(schema.resourceDesiredRelease.environmentId, schema.environment.id),
+    )
+    .innerJoin(
+      schema.resource,
+      eq(schema.resourceDesiredRelease.resourceId, schema.resource.id),
+    )
+    .where(
+      and(
+        eq(schema.resourceDesiredRelease.resourceId, resourceId),
+        eq(schema.resourceDesiredRelease.environmentId, environmentId),
+        eq(schema.resourceDesiredRelease.deploymentId, deploymentId),
+      ),
+    )
+    .then(takeFirstOrNull);
+};
+
+const getReleaseCandidates = async (
+  ctx: DeploymentResourceContext,
+): Promise<Release[]> => {
+  return db
+    .select({
+      id: schema.release.id,
+      createdAt: schema.release.createdAt,
+      version: schema.deploymentVersion,
+      variables: sql<Record<string, unknown>>`COALESCE(jsonb_object_agg(
+              ${schema.releaseVariable.key},
+              ${schema.releaseVariable.value}
+            ) FILTER (WHERE ${schema.releaseVariable.key} IS NOT NULL), '{}'::jsonb)`.as(
+        "variables",
+      ),
+    })
+    .from(schema.release)
+    .where(
+      and(
+        eq(schema.release.id, ctx.resource.id),
+        eq(schema.release.environmentId, ctx.environment.id),
+        eq(schema.release.deploymentId, ctx.deployment.id),
+      ),
+    )
+    .innerJoin(
+      schema.deploymentVersion,
+      eq(schema.release.versionId, schema.deploymentVersion.id),
+    )
+    .leftJoin(
+      schema.releaseVariable,
+      and(
+        eq(schema.release.id, schema.releaseVariable.releaseId),
+        eq(schema.releaseVariable.sensitive, false),
+      ),
+    )
+    .groupBy(
+      schema.release.id,
+      schema.release.createdAt,
+      schema.deploymentVersion.id,
+    )
+    .orderBy(desc(schema.release.createdAt))
+    .limit(100)
+    .then((releases) =>
+      releases.map((r) => ({
+        ...r,
+        version: {
+          ...r.version,
+          metadata: {} as Record<string, string>,
+        },
+      })),
+    );
+};
+
+const versionCooldownRule = () =>
+  new VersionCooldownRule({
+    cooldownMinutes: 1440,
+    getLastSuccessfulDeploymentTime: () => new Date(),
+  });
+
+export const createRuleEngineEvaluationWorker = () =>
+  new Worker<RuleEngineEvaluationEvent>(
+    Channel.RuleEngineEvaluation,
+    async (job) => {
+      const { resourceId, deploymentId, environmentId } = job.data;
+
+      const key = `rule-engine-evaluation:${resourceId}-${deploymentId}-${environmentId}`;
+      const mutex = new Mutex(redis, key);
+
+      await mutex.acquire();
+      try {
+        const ctx = await createDeploymentResourceContext(job.data);
+        if (ctx == null)
+          throw new Error(
+            "Resource desired release not found. Could not build context.",
+          );
+
+        const allReleaseCandidates = await getReleaseCandidates(ctx);
+
+        const releases = Releases.from(allReleaseCandidates);
+        if (releases.isEmpty()) return;
+
+        const ruleEngine = new RuleEngine([versionCooldownRule()]);
+        const result = await ruleEngine.evaluate(releases, ctx);
+
+        console.log(result);
+      } finally {
+        await mutex.release();
+      }
+    },
+    {
+      connection: redis,
+      removeOnComplete: { age: 1 * 60 * 60, count: 5000 },
+      removeOnFail: { age: 12 * 60 * 60, count: 5000 },
+      concurrency: 10,
+    },
+  );

packages/db/src/schema/index.ts

@@ -22,6 +22,8 @@ export * from "./job-agent.js";
 export * from "./event.js";
 export * from "./release-channel.js";
 export * from "./release-job-trigger.js";
+export * from "./release.js";
+export * from "./rule.js";
 
 // relations
 export * from "./environment-relations.js";

packages/db/src/schema/release.ts

@@ -0,0 +1,63 @@
+import {
+  boolean,
+  json,
+  pgTable,
+  text,
+  timestamp,
+  uniqueIndex,
+  uuid,
+} from "drizzle-orm/pg-core";
+
+import { deploymentVersion } from "./deployment-version.js";
+import { deployment } from "./deployment.js";
+import { environment } from "./environment.js";
+import { job } from "./job.js";
+import { resource } from "./resource.js";
+
+export const release = pgTable("release", {
+  id: uuid("id").primaryKey().defaultRandom(),
+
+  versionId: uuid("version_id")
+    .notNull()
+    .references(() => deploymentVersion.id, { onDelete: "cascade" }),
+  resourceId: uuid("resource_id")
+    .notNull()
+    .references(() => resource.id, { onDelete: "cascade" }),
+  deploymentId: uuid("deployment_id")
+    .notNull()
+    .references(() => deployment.id, { onDelete: "cascade" }),
+  environmentId: uuid("environment_id")
+    .references(() => environment.id, { onDelete: "cascade" })
+    .notNull(),
+
+  createdAt: timestamp("created_at", { withTimezone: true })
+    .notNull()
+    .defaultNow(),
+});
+
+export const releaseVariable = pgTable(
+  "release_variable",
+  {
+    id: uuid("id").primaryKey().defaultRandom(),
+    releaseId: uuid("release_id")
+      .notNull()
+      .references(() => release.id, { onDelete: "cascade" }),
+    key: text("key").notNull(),
+    value: json("value").notNull(),
+    sensitive: boolean("sensitive").notNull().default(false),
+  },
+  (t) => ({ uniq: uniqueIndex().on(t.releaseId, t.key) }),
+);
+
+export const releaseJob = pgTable("release_job", {
+  id: uuid("id").primaryKey().defaultRandom(),
+  releaseId: uuid("release_id")
+    .notNull()
+    .references(() => release.id, { onDelete: "cascade" }),
+  jobId: uuid("job_id")
+    .notNull()
+    .references(() => job.id, { onDelete: "cascade" }),
+  createdAt: timestamp("created_at", { withTimezone: true })
+    .notNull()
+    .defaultNow(),
+});

packages/db/src/schema/resource.ts

@@ -50,8 +50,11 @@ import {
 } from "@ctrlplane/validators/resources";
 
 import type { Tx } from "../common.js";
+import { deployment } from "./deployment.js";
+import { environment } from "./environment.js";
 import { job } from "./job.js";
 import { releaseJobTrigger } from "./release-job-trigger.js";
+import { release } from "./release.js";
 import { resourceProvider } from "./resource-provider.js";
 import { workspace } from "./workspace.js";
 
@@ -405,3 +408,27 @@ export const createResourceVariable = createInsertSchema(resourceVariable, {
 
 export const updateResourceVariable = createResourceVariable.partial();
 export type ResourceVariable = InferSelectModel<typeof resourceVariable>;
+
+export const resourceDesiredRelease = pgTable(
+  "resource_desired_release",
+  {
+    id: uuid("id").primaryKey().defaultRandom(),
+
+    resourceId: uuid("resource_id")
+      .references(() => resource.id, { onDelete: "cascade" })
+      .notNull(),
+    environmentId: uuid("environment_id")
+      .references(() => environment.id, { onDelete: "cascade" })
+      .notNull(),
+    deploymentId: uuid("deployment_id")
+      .references(() => deployment.id, { onDelete: "cascade" })
+      .notNull(),
+
+    desiredReleaseId: uuid("desired_release_id")
+      .references(() => release.id, { onDelete: "set null" })
+      .default(sql`NULL`),
+  },
+  (t) => ({
+    uniq: uniqueIndex().on(t.resourceId, t.environmentId, t.deploymentId),
+  }),
+);