1055780845

Build Type

github

Committed by Unknown Committer

Commit Message

Unknown Commit Message

Pull Request Pull Request #621: Quirrel ci diff preview depls

Run Details

576 of 849 branches covered (67.84%)

Branch coverage included in aggregate %.

48 of 64 new or added lines in 8 files covered. (75.0%)

5 existing lines in 2 files now uncovered.

1165 of 1454 relevant lines covered (80.12%)

17.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.82

/src/api/scheduler/jobs-repo.ts

import { EnqueueJob } from "./types/queues/POST/body";
import { QueuesUpdateCronBody } from "./types/queues/update-cron";
import {
  encodeQueueDescriptor,
  decodeQueueDescriptor,
} from "../shared/queue-descriptor";

import * as uuid from "uuid";
import { cron } from "../shared/owl";
import Owl, { Job, Closable } from "@quirrel/owl";
import { QueueRepo } from "./queue-repo";
import { Redis } from "ioredis";

interface PaginationOpts {
  cursor: number;
  count?: number;
}

interface JobDTO {
  id: string;
  endpoint: string;
  body: string;
  runAt: string;
  exclusive: boolean;
  retry: number[];
  count: number;
  repeat?: {
    every?: number;
    times?: number;
    count: number;
    cron?: string;
  };
}

export class JobsRepo implements Closable {
  protected producer;
  public readonly queueRepo: QueueRepo;

  constructor(protected readonly owl: Owl<"every" | "cron">, redis: Redis) {
    this.producer = this.owl.createProducer();
    this.queueRepo = new QueueRepo(redis, this);
  }

  private static toJobDTO(job: Job<"every" | "cron">): JobDTO {
    const { endpoint } = decodeQueueDescriptor(job.queue);

    return {
      id: job.id,
      endpoint,
      body: job.payload,
      runAt: job.runAt.toISOString(),
      exclusive: job.exclusive,
      retry: job.retry,
      count: job.count,
      repeat: job.schedule
        ? {
            count: job.count,
            cron: job.schedule?.type === "cron" ? job.schedule.meta : undefined,
            every:
              job.schedule?.type === "every" ? +job.schedule.meta : undefined,
            times: job.schedule?.times,
          }
        : undefined,
    };
  }

  public async close() {
    await this.producer.close();
  }

  public async find(
    byTokenId: string,
    endpoint: string,
    { count, cursor }: PaginationOpts
  ) {
    const { newCursor, jobs } = await this.producer.scanQueue(
      encodeQueueDescriptor(byTokenId, endpoint),
      cursor,
      count
    );

    return {
      cursor: newCursor,
      jobs: jobs.map(JobsRepo.toJobDTO),
    };
  }

  public async findAll({ count, cursor }: PaginationOpts) {
    const { newCursor, jobs } = await this.producer.scanQueuePattern(
      encodeQueueDescriptor("*", "*"),
      cursor,
      count
    );

    return {
      cursor: newCursor,
      jobs: jobs.map((job) => {
        const { tokenId } = decodeQueueDescriptor(job.queue);
        return {
          ...JobsRepo.toJobDTO(job),
          tokenId,
        };
      }),
    };
  }

  public async findByTokenId(
    byTokenId: string,
    { count, cursor }: PaginationOpts
  ) {
    const { newCursor, jobs } = await this.producer.scanQueuePattern(
      encodeQueueDescriptor(byTokenId, "*"),
      cursor,
      count
    );

    return {
      cursor: newCursor,
      jobs: jobs.map(JobsRepo.toJobDTO),
    };
  }

  public async findById(tokenId: string, endpoint: string, id: string) {
    const job = await this.producer.findById(
      encodeQueueDescriptor(tokenId, endpoint),
      id
    );
    return job ? JobsRepo.toJobDTO(job) : undefined;
  }

  public async invoke(tokenId: string, endpoint: string, id: string) {
    return await this.producer.invoke(
      encodeQueueDescriptor(tokenId, endpoint),
      id
    );
  }

  public async delete(tokenId: string, endpoint: string, id: string) {
    return await this.producer.delete(
      encodeQueueDescriptor(tokenId, endpoint),
      id
    );
  }

  public async enqueue(
    tokenId: string,
    endpoint: string,
    {
      body,
      runAt: runAtOption,
      id,
      delay,
      repeat,
      override,
      exclusive,
      retry,
    }: EnqueueJob
  ) {
    if (typeof id === "undefined") {
      id = uuid.v4();
    }

    let runAt: Date | undefined = undefined;

    if (runAtOption) {
      runAt = new Date(runAtOption);
    } else if (delay) {
      runAt = new Date(Date.now() + delay);
    }

    if (repeat?.cron) {
      runAt = cron(runAt ?? new Date(), repeat.cron);
    }

    if (typeof repeat?.times === "number" && repeat.times < 1) {
      throw new Error("repeat.times must be positive");
    }

    let schedule_type: "every" | "cron" | undefined = undefined;
    let schedule_meta: string | undefined = undefined;

    if (repeat?.cron) {
      schedule_type = "cron";
      schedule_meta = repeat.cron;
    }

    if (repeat?.every) {
      schedule_type = "every";
      schedule_meta = "" + repeat.every;
    }

    const createdJob = await this.producer.enqueue({
      queue: encodeQueueDescriptor(tokenId, endpoint),
      id,
      payload: body ?? "",
      runAt,
      exclusive,
      schedule: schedule_type
        ? {
            type: schedule_type,
            meta: schedule_meta!,
            times: repeat?.times,
          }
        : undefined,
      override,
      retry,
    });
    await this.queueRepo.add(endpoint, tokenId);

    return JobsRepo.toJobDTO(createdJob);
  }

  public async updateCron(
    tokenId: string,
    { baseUrl, crons, dryRun }: QueuesUpdateCronBody
  ) {
    const deleted: string[] = [];

    const queues = await this.queueRepo.get(tokenId);
    const queuesOnSameDeployment = queues.filter((q) => q.startsWith(baseUrl));

    if (!dryRun) {
      await Promise.all(
        crons.map(async ({ route, schedule }) => {
          await this.enqueue(tokenId, `${baseUrl}/${route}`, {
            id: "@cron",
            body: "null",
            override: true,
            repeat: { cron: schedule },
          });
        })
      );
    }

    const routesThatShouldPersist = crons.map((c) => c.route);
    await Promise.all(
      queuesOnSameDeployment.map(async (queue) => {
        const route = queue.slice(baseUrl.length + 1);
        const shouldPersist = routesThatShouldPersist.includes(route);
        if (shouldPersist) {
          return;
        }

        if (dryRun) {
          const exists = await this.findById(tokenId, queue, "@cron");
          if (exists) {
            deleted.push(queue);
          }
        } else {
          const result = await this.delete(tokenId, queue, "@cron");
          if (result === "deleted") {
            deleted.push(queue);
          }
        }
      })
    );

    return { deleted };
  }

  public onEvent(
    requesterTokenId: string,
    cb: (
      event: string,
      job: { endpoint: string; id: string; runAt?: string } | JobDTO
    ) => void
  ) {
    const activity = this.owl.createActivity(
      async (event) => {
        if (event.type === "scheduled") {
          cb(
            "scheduled",
            JobsRepo.toJobDTO(event.job as Job<"every" | "cron">)
          );
          return;
        }

        const { endpoint } = decodeQueueDescriptor(event.queue);

        switch (event.type) {
          case "acknowledged":
            cb("completed", { endpoint, id: event.id });
            break;
          case "requested":
            cb("started", { endpoint, id: event.id });
            break;
          case "invoked":
            cb("invoked", { endpoint, id: event.id });
            break;
          case "rescheduled":
            cb("rescheduled", {
              endpoint,
              id: event.id,
              runAt: event.runAt.toISOString(),
            });
            break;
          case "deleted":
            cb("deleted", { endpoint, id: event.id });
            break;
        }
      },
      {
        queue: encodeQueueDescriptor(requesterTokenId, "*"),
      }
    );

    return () => activity.close();
  }
}

1	import { EnqueueJob } from "./types/queues/POST/body";
2	import { QueuesUpdateCronBody } from "./types/queues/update-cron";
3	import {	17✔
4	encodeQueueDescriptor,
5	decodeQueueDescriptor,
6	} from "../shared/queue-descriptor";
7
8	import * as uuid from "uuid";	17✔
9	import { cron } from "../shared/owl";	17✔
10	import Owl, { Job, Closable } from "@quirrel/owl";
11	import { QueueRepo } from "./queue-repo";	17✔
12	import { Redis } from "ioredis";
13
14	interface PaginationOpts {
15	cursor: number;
16	count?: number;
17	}
18
19	interface JobDTO {
20	id: string;
21	endpoint: string;
22	body: string;
23	runAt: string;
24	exclusive: boolean;
25	retry: number[];
26	count: number;
27	repeat?: {
28	every?: number;
29	times?: number;
30	count: number;
31	cron?: string;
32	};
33	}
34
35	export class JobsRepo implements Closable {	17✔
36	protected producer;
37	public readonly queueRepo: QueueRepo;
38
39	constructor(protected readonly owl: Owl<"every" \| "cron">, redis: Redis) {	62✔
40	this.producer = this.owl.createProducer();	62✔
41	this.queueRepo = new QueueRepo(redis, this);	62✔
42	}
43
44	private static toJobDTO(job: Job<"every" \| "cron">): JobDTO {
45	const { endpoint } = decodeQueueDescriptor(job.queue);	120✔
46
47	return {	120✔
48	id: job.id,
49	endpoint,
50	body: job.payload,
51	runAt: job.runAt.toISOString(),
52	exclusive: job.exclusive,
53	retry: job.retry,
54	count: job.count,
55	repeat: job.schedule
56	? {	120✔
57	count: job.count,
58	cron: job.schedule?.type === "cron" ? job.schedule.meta : undefined,	92!
59	every:
60	job.schedule?.type === "every" ? +job.schedule.meta : undefined,	92!
61	times: job.schedule?.times,	69!
62	}
63	: undefined,
64	};
65	}
66
67	public async close() {
68	await this.producer.close();	42✔
69	}
70
71	public async find(
72	byTokenId: string,
73	endpoint: string,
74	{ count, cursor }: PaginationOpts
75	) {
76	const { newCursor, jobs } = await this.producer.scanQueue(	4✔
77	encodeQueueDescriptor(byTokenId, endpoint),
78	cursor,
79	count
80	);
81
82	return {	4✔
83	cursor: newCursor,
84	jobs: jobs.map(JobsRepo.toJobDTO),
85	};
86	}
87
88	public async findAll({ count, cursor }: PaginationOpts) {
89	const { newCursor, jobs } = await this.producer.scanQueuePattern(	6✔
90	encodeQueueDescriptor("", ""),
91	cursor,
92	count
93	);
94
95	return {	6✔
96	cursor: newCursor,
97	jobs: jobs.map((job) => {
98	const { tokenId } = decodeQueueDescriptor(job.queue);	12✔
99	return {	12✔
100	...JobsRepo.toJobDTO(job),
101	tokenId,
102	};
103	}),
104	};
105	}
106
107	public async findByTokenId(
108	byTokenId: string,
109	{ count, cursor }: PaginationOpts
110	) {
111	const { newCursor, jobs } = await this.producer.scanQueuePattern(	×
112	encodeQueueDescriptor(byTokenId, "*"),
113	cursor,
114	count
115	);
116
117	return {	×
118	cursor: newCursor,
119	jobs: jobs.map(JobsRepo.toJobDTO),
120	};
121	}
122
123	public async findById(tokenId: string, endpoint: string, id: string) {
124	const job = await this.producer.findById(	12✔
125	encodeQueueDescriptor(tokenId, endpoint),
126	id
127	);
128	return job ? JobsRepo.toJobDTO(job) : undefined;	12✔
129	}
130
131	public async invoke(tokenId: string, endpoint: string, id: string) {
132	return await this.producer.invoke(	×
133	encodeQueueDescriptor(tokenId, endpoint),
134	id
135	);
136	}
137
138	public async delete(tokenId: string, endpoint: string, id: string) {
139	return await this.producer.delete(	13✔
140	encodeQueueDescriptor(tokenId, endpoint),
141	id
142	);
143	}
144
145	public async enqueue(
146	tokenId: string,
147	endpoint: string,
148	{
149	body,
150	runAt: runAtOption,
151	id,
152	delay,
153	repeat,
154	override,
155	exclusive,
156	retry,
157	}: EnqueueJob
158	) {
159	if (typeof id === "undefined") {	90✔
160	id = uuid.v4();	64✔
161	}
162
163	let runAt: Date \| undefined = undefined;	90✔
164
165	if (runAtOption) {	90✔
166	runAt = new Date(runAtOption);	32✔
167	} else if (delay) {	58✔
168	runAt = new Date(Date.now() + delay);	17✔
169	}
170
171	if (repeat?.cron) {	90✔
172	runAt = cron(runAt ?? new Date(), repeat.cron);	4!
173	}
174
175	if (typeof repeat?.times === "number" && repeat.times < 1) {	90!
176	throw new Error("repeat.times must be positive");	×
177	}
178
179	let schedule_type: "every" \| "cron" \| undefined = undefined;	90✔
180	let schedule_meta: string \| undefined = undefined;	90✔
181
182	if (repeat?.cron) {	90✔
183	schedule_type = "cron";	4✔
184	schedule_meta = repeat.cron;	4✔
185	}
186
187	if (repeat?.every) {	90✔
188	schedule_type = "every";	18✔
189	schedule_meta = "" + repeat.every;	18✔
190	}
191
192	const createdJob = await this.producer.enqueue({	90✔
193	queue: encodeQueueDescriptor(tokenId, endpoint),
194	id,
195	payload: body ?? "",	270!
196	runAt,
197	exclusive,
198	schedule: schedule_type
199	? {	90✔
200	type: schedule_type,
201	meta: schedule_meta!,
202	times: repeat?.times,	66!
203	}
204	: undefined,
205	override,
206	retry,
207	});
208	await this.queueRepo.add(endpoint, tokenId);	90✔
209
210	return JobsRepo.toJobDTO(createdJob);	90✔
211	}
212
213	public async updateCron(
214	tokenId: string,
215	{ baseUrl, crons, dryRun }: QueuesUpdateCronBody
216	) {
217	const deleted: string[] = [];	2✔
218
219	const queues = await this.queueRepo.get(tokenId);	2✔
220	const queuesOnSameDeployment = queues.filter((q) => q.startsWith(baseUrl));	2✔
221
222	if (!dryRun) {	2!
223	await Promise.all(	2✔
224	crons.map(async ({ route, schedule }) => {
225	await this.enqueue(tokenId, `${baseUrl}/${route}`, {	1✔
226	id: "@cron",
227	body: "null",
228	override: true,
229	repeat: { cron: schedule },
230	});
231	})
232	);
233	}
234
235	const routesThatShouldPersist = crons.map((c) => c.route);	2✔
236	await Promise.all(	2✔
237	queuesOnSameDeployment.map(async (queue) => {
238	const route = queue.slice(baseUrl.length + 1);	1✔
239	const shouldPersist = routesThatShouldPersist.includes(route);	1✔
240	if (shouldPersist) {	1!
NEW 241	return;	×
242	}
243
244	if (dryRun) {	1!
NEW 245	const exists = await this.findById(tokenId, queue, "@cron");	×
NEW 246	if (exists) {	×
NEW 247	deleted.push(queue);	×
248	}
249	} else {
250	const result = await this.delete(tokenId, queue, "@cron");	1✔
251	if (result === "deleted") {	1!
252	deleted.push(queue);	1✔
253	}
254	}
255	})
256	);
257
258	return { deleted };	2✔
259	}
260
261	public onEvent(
262	requesterTokenId: string,
263	cb: (
264	event: string,
265	job: { endpoint: string; id: string; runAt?: string } \| JobDTO
266	) => void
267	) {
268	const activity = this.owl.createActivity(	2✔
269	async (event) => {
270	if (event.type === "scheduled") {	12✔
271	cb(	4✔
272	"scheduled",
273	JobsRepo.toJobDTO(event.job as Job<"every" \| "cron">)
274	);
275	return;	4✔
276	}
277
278	const { endpoint } = decodeQueueDescriptor(event.queue);	8✔
279
280	switch (event.type) {	8✔
281	case "acknowledged":	8!
282	cb("completed", { endpoint, id: event.id });	4✔
283	break;	4✔
284	case "requested":
285	cb("started", { endpoint, id: event.id });	4✔
286	break;	4✔
287	case "invoked":
288	cb("invoked", { endpoint, id: event.id });	×
289	break;	×
290	case "rescheduled":
291	cb("rescheduled", {	×
292	endpoint,
293	id: event.id,
294	runAt: event.runAt.toISOString(),
295	});
296	break;	×
297	case "deleted":
298	cb("deleted", { endpoint, id: event.id });	×
299	break;	×
300	}
301	},
302	{
303	queue: encodeQueueDescriptor(requesterTokenId, "*"),
304	}
305	);
306
307	return () => activity.close();	2✔
308	}
309	}

quirrel-dev / quirrel / 1055780845

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous