982330465

Build Type

github

Committed by Unknown Committer

Commit Message

Unknown Commit Message

Pull Request Pull Request #307: Cron timezone

Run Details

532 of 843 branches covered (63.11%)

Branch coverage included in aggregate %.

62 of 68 new or added lines in 7 files covered. (91.18%)

17 existing lines in 4 files now uncovered.

1014 of 1368 relevant lines covered (74.12%)

15.22 hits per line

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

84.03

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

import { EnqueueJob } from "./types/queues/POST/body";
import {
  encodeQueueDescriptor,
  decodeQueueDescriptor,
} from "../shared/queue-descriptor";
import * as uuid from "uuid";
import { cron, embedTimezone, parseTimezone } from "../../shared/repeat";
import Owl, { Closable, Job } from "@quirrel/owl";

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;
    cron_tz?: string;
  };
}

export class JobsRepo implements Closable {
  protected producer;

  constructor(protected readonly owl: Owl<"every" | "cron">) {
    this.producer = this.owl.createProducer();
  }

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

    let cron: { cron?: string; cronTimezone?: string } = {};
    if (job.schedule?.type === "cron") {
      const [cronExpression, cronTimezone] = parseTimezone(job.schedule.meta);
      cron = { cron: cronExpression, cronTimezone };
    }

    return {
      id: job.id,
      endpoint,
      body: job.payload,
      runAt: job.runAt.toISOString(),
      exclusive: job.exclusive,
      retry: job.retry,
      count: job.count,
      repeat: job.schedule
        ? {
            ...cron,
            count: job.count,
            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";

      if (repeat?.cronTimezone) {
        schedule_meta = embedTimezone(repeat.cron, repeat.cronTimezone);
      } else {
        schedule_meta = repeat.cron;
      }
    }

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

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

    return JobsRepo.toJobDTO(createdJob);
  }

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

quirrel-dev / quirrel / 982330465

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