Skip to content

Commit e51a958

Browse files
adityachoudhari26adityachoudhari26
committed
chore: job variables in memory
1 parent 8e9fd9a commit e51a958

File tree

3 files changed

+268
-2
lines changed

3 files changed

+268
-2
lines changed

‎apps/event-queue/src/repository/in-memory/job-variable.ts‎

Lines changed: 137 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,137 @@
1+
import type { Tx } from "@ctrlplane/db";
2+
3+
import { eq } from "@ctrlplane/db";
4+
import { db as dbClient } from "@ctrlplane/db/client";
5+
import * as schema from "@ctrlplane/db/schema";
6+
7+
import type { Repository } from "../repository";
8+
import { createSpanWrapper } from "../../traces.js";
9+
10+
type JobVariable = typeof schema.jobVariable.$inferSelect;
11+
12+
const getReleaseJobVariables = createSpanWrapper(
13+
"job-variable-getReleaseJobVariables",
14+
async (_span, workspaceId: string) =>
15+
dbClient
16+
.select()
17+
.from(schema.jobVariable)
18+
.innerJoin(schema.job, eq(schema.jobVariable.jobId, schema.job.id))
19+
.innerJoin(schema.releaseJob, eq(schema.releaseJob.jobId, schema.job.id))
20+
.innerJoin(
21+
schema.release,
22+
eq(schema.releaseJob.releaseId, schema.release.id),
23+
)
24+
.innerJoin(
25+
schema.versionRelease,
26+
eq(schema.release.versionReleaseId, schema.versionRelease.id),
27+
)
28+
.innerJoin(
29+
schema.releaseTarget,
30+
eq(schema.versionRelease.releaseTargetId, schema.releaseTarget.id),
31+
)
32+
.innerJoin(
33+
schema.resource,
34+
eq(schema.releaseTarget.resourceId, schema.resource.id),
35+
)
36+
.where(eq(schema.resource.workspaceId, workspaceId))
37+
.then((rows) => rows.map((row) => row.job_variable)),
38+
);
39+
40+
const getRunbookJobVariables = createSpanWrapper(
41+
"job-variable-getRunbookJobVariables",
42+
async (_span, workspaceId: string) => {
43+
return dbClient
44+
.select()
45+
.from(schema.jobVariable)
46+
.innerJoin(schema.job, eq(schema.jobVariable.jobId, schema.job.id))
47+
.innerJoin(
48+
schema.runbookJobTrigger,
49+
eq(schema.runbookJobTrigger.jobId, schema.job.id),
50+
)
51+
.innerJoin(
52+
schema.runbook,
53+
eq(schema.runbookJobTrigger.runbookId, schema.runbook.id),
54+
)
55+
.innerJoin(schema.system, eq(schema.runbook.systemId, schema.system.id))
56+
.where(eq(schema.system.workspaceId, workspaceId))
57+
.then((rows) => rows.map((row) => row.job_variable));
58+
},
59+
);
60+
61+
const getInitialEntities = createSpanWrapper(
62+
"job-variable-getInitialEntities",
63+
async (span, workspaceId: string) => {
64+
const [releaseJobVariables, runbookJobVariables] = await Promise.all([
65+
getReleaseJobVariables(workspaceId),
66+
getRunbookJobVariables(workspaceId),
67+
]);
68+
const initialEntities = [...releaseJobVariables, ...runbookJobVariables];
69+
span.setAttributes({ "job-variable.count": initialEntities.length });
70+
return initialEntities;
71+
},
72+
);
73+
74+
type InMemoryJobVariableRepositoryOptions = {
75+
initialEntities: JobVariable[];
76+
tx?: Tx;
77+
};
78+
79+
export class InMemoryJobVariableRepository implements Repository<JobVariable> {
80+
private entities: Map<string, JobVariable>;
81+
private db: Tx;
82+
83+
constructor(opts: InMemoryJobVariableRepositoryOptions) {
84+
this.entities = new Map();
85+
for (const entity of opts.initialEntities)
86+
this.entities.set(entity.id, entity);
87+
this.db = opts.tx ?? dbClient;
88+
}
89+
90+
static async create(workspaceId: string) {
91+
const initialEntities = await getInitialEntities(workspaceId);
92+
return new InMemoryJobVariableRepository({
93+
initialEntities,
94+
tx: dbClient,
95+
});
96+
}
97+
98+
get(id: string) {
99+
return this.entities.get(id) ?? null;
100+
}
101+
102+
getAll() {
103+
return Array.from(this.entities.values());
104+
}
105+
106+
async create(entity: JobVariable) {
107+
this.entities.set(entity.id, entity);
108+
await this.db
109+
.insert(schema.jobVariable)
110+
.values(entity)
111+
.onConflictDoNothing();
112+
return entity;
113+
}
114+
115+
async update(entity: JobVariable) {
116+
this.entities.set(entity.id, entity);
117+
await this.db
118+
.update(schema.jobVariable)
119+
.set(entity)
120+
.where(eq(schema.jobVariable.id, entity.id));
121+
return entity;
122+
}
123+
124+
async delete(id: string) {
125+
const entity = this.entities.get(id);
126+
if (entity == null) return null;
127+
this.entities.delete(id);
128+
await this.db
129+
.delete(schema.jobVariable)
130+
.where(eq(schema.jobVariable.id, id));
131+
return entity;
132+
}
133+
134+
exists(id: string) {
135+
return this.entities.has(id);
136+
}
137+
}

‎apps/event-queue/src/repository/in-memory/job.ts‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,129 @@
1+
import type { Tx } from "@ctrlplane/db";
2+
3+
import { eq } from "@ctrlplane/db";
4+
import { db as dbClient } from "@ctrlplane/db/client";
5+
import * as schema from "@ctrlplane/db/schema";
6+
7+
import type { Repository } from "../repository.js";
8+
import { createSpanWrapper } from "../../traces.js";
9+
10+
type InMemoryJobRepositoryOptions = {
11+
initialEntities: schema.Job[];
12+
tx?: Tx;
13+
};
14+
15+
const getReleaseJobs = createSpanWrapper(
16+
"job-getReleaseJobs",
17+
async (_span, workspaceId: string) => {
18+
return dbClient
19+
.select()
20+
.from(schema.job)
21+
.innerJoin(schema.releaseJob, eq(schema.releaseJob.jobId, schema.job.id))
22+
.innerJoin(
23+
schema.release,
24+
eq(schema.releaseJob.releaseId, schema.release.id),
25+
)
26+
.innerJoin(
27+
schema.versionRelease,
28+
eq(schema.release.versionReleaseId, schema.versionRelease.id),
29+
)
30+
.innerJoin(
31+
schema.releaseTarget,
32+
eq(schema.versionRelease.releaseTargetId, schema.releaseTarget.id),
33+
)
34+
.innerJoin(
35+
schema.resource,
36+
eq(schema.releaseTarget.resourceId, schema.resource.id),
37+
)
38+
.where(eq(schema.resource.workspaceId, workspaceId))
39+
.then((rows) => rows.map((row) => row.job));
40+
},
41+
);
42+
43+
const getRunbookJobs = createSpanWrapper(
44+
"job-getRunbookJobs",
45+
async (_span, workspaceId: string) => {
46+
return dbClient
47+
.select()
48+
.from(schema.job)
49+
.innerJoin(
50+
schema.runbookJobTrigger,
51+
eq(schema.runbookJobTrigger.jobId, schema.job.id),
52+
)
53+
.innerJoin(
54+
schema.runbook,
55+
eq(schema.runbookJobTrigger.runbookId, schema.runbook.id),
56+
)
57+
.innerJoin(schema.system, eq(schema.runbook.systemId, schema.system.id))
58+
.where(eq(schema.system.workspaceId, workspaceId))
59+
.then((rows) => rows.map((row) => row.job));
60+
},
61+
);
62+
63+
const getInitialEntities = createSpanWrapper(
64+
"job-getInitialEntities",
65+
async (span, workspaceId: string) => {
66+
const [releaseJobs, runbookJobs] = await Promise.all([
67+
getReleaseJobs(workspaceId),
68+
getRunbookJobs(workspaceId),
69+
]);
70+
const initialEntities = [...releaseJobs, ...runbookJobs];
71+
span.setAttributes({ "job.count": initialEntities.length });
72+
return initialEntities;
73+
},
74+
);
75+
76+
export class InMemoryJobRepository implements Repository<schema.Job> {
77+
private entities: Map<string, schema.Job>;
78+
private db: Tx;
79+
80+
constructor(opts: InMemoryJobRepositoryOptions) {
81+
this.entities = new Map();
82+
for (const entity of opts.initialEntities)
83+
this.entities.set(entity.id, entity);
84+
this.db = opts.tx ?? dbClient;
85+
}
86+
87+
static async create(workspaceId: string) {
88+
const initialEntities = await getInitialEntities(workspaceId);
89+
return new InMemoryJobRepository({
90+
initialEntities,
91+
tx: dbClient,
92+
});
93+
}
94+
95+
get(id: string) {
96+
return this.entities.get(id) ?? null;
97+
}
98+
99+
getAll() {
100+
return Array.from(this.entities.values());
101+
}
102+
103+
async create(entity: schema.Job) {
104+
this.entities.set(entity.id, entity);
105+
await this.db.insert(schema.job).values(entity).onConflictDoNothing();
106+
return entity;
107+
}
108+
109+
async update(entity: schema.Job) {
110+
this.entities.set(entity.id, entity);
111+
await this.db
112+
.update(schema.job)
113+
.set(entity)
114+
.where(eq(schema.job.id, entity.id));
115+
return entity;
116+
}
117+
118+
async delete(id: string) {
119+
const entity = this.entities.get(id);
120+
if (entity == null) return null;
121+
this.entities.delete(id);
122+
await this.db.delete(schema.job).where(eq(schema.job.id, id));
123+
return entity;
124+
}
125+
126+
exists(id: string) {
127+
return this.entities.has(id);
128+
}
129+
}

‎e2e/tests/api/releases/version-release.spec.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -153,6 +153,8 @@ test.describe("Version Release Creation", () => {
153153
);
154154
const resourceId = resourceResponse.data?.id ?? "";
155155

156+
await page.waitForTimeout(12_000);
157+
156158
const releaseTargetResponse = await api.GET(
157159
"/v1/resources/{resourceId}/release-targets",
158160
{
@@ -170,8 +172,6 @@ test.describe("Version Release Creation", () => {
170172
);
171173
expect(releaseTarget).toBeDefined();
172174

173-
await page.waitForTimeout(12_000);
174-
175175
const releaseResponse = await api.GET(
176176
"/v1/release-targets/{releaseTargetId}/releases",
177177
{

0 commit comments

Comments
 (0)