1055705071

Build Type

github

Committed by Unknown Committer

Commit Message

Unknown Commit Message

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

Run Details

492 of 709 branches covered (69.39%)

Branch coverage included in aggregate %.

44 of 52 new or added lines in 6 files covered. (84.62%)

13 existing lines in 3 files now uncovered.

910 of 1084 relevant lines covered (83.95%)

14.73 hits per line

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

78.85

/src/client/index.ts

import { Job, JobDTO } from "./job";
import * as config from "./config";
import * as z from "zod";
import type { IsExact, AssertTrue } from "conditional-type-checks";
import Encryptor from "secure-e2ee";
import { symmetric, asymmetric } from "secure-webhooks";
import ms from "ms";
import fetch from "cross-fetch";
import type { IncomingHttpHeaders } from "http";
import pack from "../../package.json";
import * as EnhancedJSON from "./enhanced-json";
import { isValidRegex } from "../shared/is-valid-regex";

export { Job };

export interface JobMeta
  extends Pick<JobDTO, "id" | "count" | "exclusive" | "retry"> {
  /**
   * If this is a repeated job, the next repetition will be scheduled for this Date.
   */
  readonly nextRepetition?: Date;
}

export type QuirrelJobHandler<T> = (job: T, meta: JobMeta) => Promise<void>;
export type DefaultJobOptions = Pick<EnqueueJobOptions, "exclusive" | "retry">;

interface CreateQuirrelClientArgs<T> {
  route: string;
  handler: QuirrelJobHandler<T>;
  defaultJobOptions?: DefaultJobOptions;
  config?: {
    /**
     * Recommended way to set this: process.env.QUIRREL_BASE_URL
     */
    applicationBaseUrl?: string;

    /**
     * Overrides URL of the Quirrel Endpoint.
     * @default https://api.quirrel.dev or http://localhost:9181
     * Recommended way to set this: process.env.QUIRREL_API_URL
     */
    quirrelBaseUrl?: string;

    /**
     * Bearer Secret for authenticating with Quirrel.
     * Obtain on quirrel.dev or using the API of a self-hosted instance.
     * Recommended way to set this: process.env.QUIRREL_TOKEN
     */
    token?: string;

    /**
     * Secret used for end-to-end encryption.
     * Needs to be 32 characters long.
     * Recommended way to set this: process.env.QUIRREL_ENCRYPTION_SECRET
     */
    encryptionSecret?: string;

    /**
     * Public key used for verifying signatures.
     * Recommended way to set this: process.env.QUIRREL_SIGNATURE_PUBLIC_KEY
     */
    signaturePublicKey?: string;

    /**
     * Old Secrets that have been rotated out.
     * @see https://docs.quirrel.dev/docs/faq#my-encryption-secret-has-been-leaked-what-now
     * Recommended way to set this: process.env.QUIRREL_OLD_SECRETS
     */
    oldSecrets?: string[];
  };

  fetch?: typeof fetch;
  catchDecryptionErrors?: (error: Error) => void;
}

const vercelMs = z
  .string()
  .regex(
    /^(-?(?:\d+)?\.?\d+) *(milliseconds?|msecs?|ms|seconds?|secs?|s|minutes?|mins?|m|hours?|hrs?|h|days?|d|weeks?|w|years?|yrs?|y)?$/i,
    "Please provide a valid time span, according to https://github.com/vercel/ms"
  );

const timeDuration = (fieldName = "duration") =>
  z.union([
    z.number().min(1, { message: `${fieldName} must be positive` }),
    vercelMs,
  ]);

export const cron = z
  .string()
  .refine(
    isValidRegex,
    "Please provide a valid Cron expression. See https://github.com/harrisiirak/cron-parser for reference"
  );

const EnqueueJobOptionsSchema = z.object({
  id: z.string().optional(),
  exclusive: z.boolean().optional(),
  override: z.boolean().optional(),
  retry: z.array(timeDuration("retry")).min(1).max(10).optional(),
  delay: timeDuration("delay").optional(),
  runAt: z
    .date()
    .refine((date) => Date.now() <= +date, {
      message: "runAt must not be in the past",
    })
    .optional(),
  repeat: z
    .object({
      every: timeDuration("every").optional(),
      times: z.number().nonnegative().optional(),
      cron: cron.optional(),
    })
    .optional(),
});

type EnqueueJobOptionsSchema = z.TypeOf<typeof EnqueueJobOptionsSchema>;

type EnqueueJobOptionssSchemaMatchesDocs = AssertTrue<
  IsExact<EnqueueJobOptions, EnqueueJobOptionsSchema>
>;

/**
 * @deprecated renamed to EnqueueJobOptions
 */
export type EnqueueJobOpts = EnqueueJobOptions;

export interface EnqueueJobOptions {
  /**
   * Can be used to make a job easier to manage.
   * If there's already a job with the same ID, this job will be trashed.
   * @tutorial https://demo.quirrel.dev/managed
   */
  id?: string;

  /**
   * If set to `true`,
   * no other job (on the same queue)
   * will be executed at the same time.
   */
  exclusive?: boolean;

  /**
   * If a job fails, retry it at along the specified intervals.
   * @example ["5min", "1h", "1d"] // retries it after 5 minutes, 1:05 hours, and 1 day 1:05 hours
   */
  retry?: (number | string)[];

  /**
   * Determines what to do when a job
   * with the same ID already exists.
   * false: do nothing (default)
   * true: replace the job
   */
  override?: boolean;

  /**
   * Will delay the job's execution by the specified amount of milliseconds.
   * Supports human-readable notation as of @see https://github.com/vercel/ms.
   * If used together with `repeat`, this will delay the first job to be executed.
   */
  delay?: number | string;

  /**
   * Schedules the job for execution at the specified timestamp.
   */
  runAt?: Date;

  repeat?: {
    /**
     * Will make the job repeat every X milliseconds.
     * Supports human-readable notation as of @see https://github.com/vercel/ms.
     * If `delay` isn't set, the first repetition will be executed immediately.
     */
    every?: number | string;

    /**
     * Can be used in conjunction with @field every and @field cron
     * to limit the number of executions.
     */
    times?: number;

    /**
     * Schedules the job according to the Cron expression.
     * @see https://github.com/harrisiirak/cron-parser for supported syntax
     * If `delay` isn't set, the first repetition will be executed immediately.
     */
    cron?: string;
  };
}

function parseDuration(value: number | string | undefined): number | undefined {
  if (!value) {
    return undefined;
  }

  if (typeof value === "string") {
    return ms(value);
  }

  return value;
}

function runAtToDelay(value: Date) {
  return +value - Date.now();
}

function getEncryptor(
  encryptionSecret: string | undefined,
  oldSecrets: string[] = []
) {
  if (!encryptionSecret) {
    return undefined;
  }

  return new Encryptor(encryptionSecret, [encryptionSecret, ...oldSecrets]);
}

function getAuthHeaders(
  token: string | undefined
): { Authorization: string } | {} {
  if (!token) {
    return {};
  }

  return { Authorization: `Bearer ${token}` };
}

export class QuirrelClient<T> {
  private handler;
  private route;
  private defaultJobOptions;
  private encryptor;
  private defaultHeaders: Record<string, string>;
  private quirrelBaseUrl;
  private baseUrl;
  private token;
  private fetch;
  private catchDecryptionErrors;
  private signaturePublicKey;

  constructor(args: CreateQuirrelClientArgs<T>) {
    this.handler = args.handler;
    this.defaultJobOptions = args.defaultJobOptions;

    const token = args.config?.token ?? config.getQuirrelToken();
    this.defaultHeaders = {
      ...getAuthHeaders(token),
      "X-QuirrelClient-Version": pack.version,
    };

    const quirrelBaseUrl =
      args.config?.quirrelBaseUrl ?? config.getQuirrelBaseUrl();
    const applicationBaseUrl = config.withoutTrailingSlash(
      config.prefixWithProtocol(
        args.config?.applicationBaseUrl ?? config.getApplicationBaseUrl()!
      )
    );
    this.quirrelBaseUrl = quirrelBaseUrl;
    this.baseUrl =
      quirrelBaseUrl +
      "/queues/" +
      encodeURIComponent(applicationBaseUrl + "/" + args.route);
    this.token = args.config?.token ?? config.getQuirrelToken();
    this.route = args.route;
    this.encryptor = getEncryptor(
      args.config?.encryptionSecret ?? config.getEncryptionSecret(),
      args.config?.oldSecrets ?? config.getOldEncryptionSecrets() ?? undefined
    );
    this.catchDecryptionErrors = args.catchDecryptionErrors;
    this.fetch = args.fetch ?? fetch;
    this.signaturePublicKey =
      args.config?.signaturePublicKey ?? config.getSignaturePublicKey();
  }

  async makeRequest(uri: string, init?: RequestInit) {
    return await this.fetch(this.quirrelBaseUrl + uri, {
      credentials: "omit",
      ...init,
      headers: {
        ...this.defaultHeaders,
        ...init?.headers,
      },
    });
  }

  private async payloadAndOptionsToBody(
    payload: T,
    options: EnqueueJobOptionsSchema
  ) {
    if (typeof payload === "undefined") {
      throw new Error("Passing `undefined` as Payload is not allowed");
    }

    if (options.repeat && options.retry?.length) {
      throw new Error("retry and repeat cannot be used together");
    }

    options = EnqueueJobOptionsSchema.parse(options);

    let delay = parseDuration(options.delay);

    if ("runAt" in options && options.runAt) {
      delay = runAtToDelay(options.runAt);
    }

    if (options.repeat) {
      options.repeat.every = parseDuration(options.repeat?.every);
    }

    let stringifiedBody = EnhancedJSON.stringify(payload);

    if (this.encryptor) {
      stringifiedBody = await this.encryptor.encrypt(stringifiedBody);
    }

    return {
      ...this.defaultJobOptions,
      body: stringifiedBody,
      delay,
      id: options.id,
      repeat: options.repeat,
      retry: options.retry?.map(parseDuration),
      override: options.override,
    };
  }

  /**
   * Enqueue a job to the specified endpoint.
   * @param options job options
   */
  async enqueue(payload: T, options: EnqueueJobOptions = {}): Promise<Job<T>> {
    const body = await this.payloadAndOptionsToBody(payload, options);

    const res = await this.fetch(this.baseUrl, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        ...this.defaultHeaders,
      },
      credentials: "omit",
      body: JSON.stringify(body),
    });

    if (res.status === 201) {
      return await this.toJob(await res.json());
    }

    throw new Error(`Unexpected response: ${await res.text()}`);
  }

  /**
   * Enqueue multiple jobs
   */
  async enqueueMany(
    jobs: { payload: T; options?: EnqueueJobOptions }[]
  ): Promise<Job<T>[]> {
    const body = await Promise.all(
      jobs.map(({ payload, options = {} }) =>
        this.payloadAndOptionsToBody(payload, options)
      )
    );

    const res = await this.fetch(this.baseUrl + "/batch", {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        ...this.defaultHeaders,
      },
      credentials: "omit",
      body: JSON.stringify(body),
    });

    if (res.status === 201) {
      const response = (await res.json()) as any[];
      return await Promise.all(response.map((job) => this.toJob(job)));
    }

    throw new Error(`Unexpected response: ${await res.text()}`);
  }

  private async decryptAndDecodeBody(body: string): Promise<T> {
    if (this.encryptor) {
      if (this.catchDecryptionErrors) {
        try {
          body = await this.encryptor.decrypt(body);
        } catch (error) {
          this.catchDecryptionErrors(error);
          return body as any;
        }
      } else {
        body = await this.encryptor.decrypt(body);
      }
    }

    return EnhancedJSON.parse(body);
  }

  private async toJob(dto: JobDTO): Promise<Job<T>> {
    return {
      ...dto,
      body: await this.decryptAndDecodeBody(dto.body),
      runAt: new Date(dto.runAt),
      delete: () => this.delete(dto.id),
      invoke: () => this.invoke(dto.id),
    };
  }

  /**
   * Iterate through scheduled jobs.
   * @example
   * for await (const jobs of queue.get()) {
   *   // do smth
   * }
   */
  async *get(): AsyncGenerator<Job<T>[]> {
    let cursor: number | null = 0;

    while (cursor !== null) {
      const res = await this.fetch(this.baseUrl + "?cursor=" + cursor, {
        headers: this.defaultHeaders,
      });

      const json = await res.json();

      const { cursor: newCursor, jobs } = json as {
        cursor: number | null;
        jobs: JobDTO[];
      };

      cursor = newCursor;

      yield await Promise.all(jobs.map((dto) => this.toJob(dto)));
    }
  }

  /**
   * Get a specific job.
   * @returns null if no job was found.
   */
  async getById(id: string): Promise<Job<T> | null> {
    const res = await this.fetch(this.baseUrl + "/" + id, {
      headers: this.defaultHeaders,
    });

    if (res.status === 404) {
      return null;
    }

    if (res.status === 200) {
      return await this.toJob(await res.json());
    }

    throw new Error("Unexpected response: " + (await res.text()));
  }

  /**
   * Schedule a job for immediate execution.
   * @returns false if job could not be found.
   */
  async invoke(id: string): Promise<boolean> {
    const res = await this.fetch(this.baseUrl + "/" + id, {
      method: "POST",
      headers: this.defaultHeaders,
    });

    if (res.status === 404) {
      return false;
    }

    if (res.status === 204) {
      return true;
    }

    throw new Error("Unexpected response: " + (await res.text()));
  }

  /**
   * Delete a job, preventing it from executing.
   * @returns false if job could not be found.
   */
  async delete(id: string): Promise<boolean> {
    const res = await this.fetch(this.baseUrl + "/" + id, {
      method: "DELETE",
      headers: this.defaultHeaders,
    });

    if (res.status === 404) {
      return false;
    }

    if (res.status === 204) {
      return true;
    }

    throw new Error("Unexpected response: " + (await res.text()));
  }

  private isValidSignature(body: string, signature: string): boolean {
    if (this.signaturePublicKey) {
      return asymmetric.verify(body, this.signaturePublicKey, signature);
    } else {
      return symmetric.verify(body, this.token!, signature);
    }
  }

  async respondTo(
    body: string,
    headers: IncomingHttpHeaders
  ): Promise<{
    status: number;
    headers: Record<string, string>;
    body: string;
  }> {
    if (process.env.NODE_ENV === "production") {
      const signature = headers["x-quirrel-signature"];
      if (typeof signature !== "string") {
        return {
          status: 401,
          headers: {},
          body: "Signature missing",
        };
      }

      if (!this.isValidSignature(body, signature)) {
        return {
          status: 401,
          headers: {},
          body: "Signature invalid",
        };
      }
    }

    const payload = await this.decryptAndDecodeBody(body);
    const { id, count, retry, nextRepetition, exclusive } = JSON.parse(
      (headers["x-quirrel-meta"] as string) ?? "{}"
    );

    console.log(`Received job to ${this.route}: `, payload);

    try {
      await this.handler(payload, {
        id,
        count,
        retry,
        nextRepetition,
        exclusive,
      });

      return {
        status: 200,
        headers: {},
        body: "OK",
      };
    } catch (error) {
      console.error(error);
      return {
        status: 500,
        headers: {},
        body: String(error),
      };
    }
  }
}

1	import { Job, JobDTO } from "./job";
2	import * as config from "./config";	12✔
3	import * as z from "zod";	12✔
4	import type { IsExact, AssertTrue } from "conditional-type-checks";
5	import Encryptor from "secure-e2ee";	12✔
6	import { symmetric, asymmetric } from "secure-webhooks";	12✔
7	import ms from "ms";	12✔
8	import fetch from "cross-fetch";	12✔
9	import type { IncomingHttpHeaders } from "http";
10	import pack from "../../package.json";	12✔
11	import * as EnhancedJSON from "./enhanced-json";	12✔
12	import { isValidRegex } from "../shared/is-valid-regex";	12✔
13
14	export { Job };
15
16	export interface JobMeta
17	extends Pick<JobDTO, "id" \| "count" \| "exclusive" \| "retry"> {
18	/**
19	* If this is a repeated job, the next repetition will be scheduled for this Date.
20	*/
21	readonly nextRepetition?: Date;
22	}
23
24	export type QuirrelJobHandler<T> = (job: T, meta: JobMeta) => Promise<void>;
25	export type DefaultJobOptions = Pick<EnqueueJobOptions, "exclusive" \| "retry">;
26
27	interface CreateQuirrelClientArgs<T> {
28	route: string;
29	handler: QuirrelJobHandler<T>;
30	defaultJobOptions?: DefaultJobOptions;
31	config?: {
32	/**
33	* Recommended way to set this: process.env.QUIRREL_BASE_URL
34	*/
35	applicationBaseUrl?: string;
36
37	/**
38	* Overrides URL of the Quirrel Endpoint.
39	* @default https://api.quirrel.dev or http://localhost:9181
40	* Recommended way to set this: process.env.QUIRREL_API_URL
41	*/
42	quirrelBaseUrl?: string;
43
44	/**
45	* Bearer Secret for authenticating with Quirrel.
46	* Obtain on quirrel.dev or using the API of a self-hosted instance.
47	* Recommended way to set this: process.env.QUIRREL_TOKEN
48	*/
49	token?: string;
50
51	/**
52	* Secret used for end-to-end encryption.
53	* Needs to be 32 characters long.
54	* Recommended way to set this: process.env.QUIRREL_ENCRYPTION_SECRET
55	*/
56	encryptionSecret?: string;
57
58	/**
59	* Public key used for verifying signatures.
60	* Recommended way to set this: process.env.QUIRREL_SIGNATURE_PUBLIC_KEY
61	*/
62	signaturePublicKey?: string;
63
64	/**
65	* Old Secrets that have been rotated out.
66	* @see https://docs.quirrel.dev/docs/faq#my-encryption-secret-has-been-leaked-what-now
67	* Recommended way to set this: process.env.QUIRREL_OLD_SECRETS
68	*/
69	oldSecrets?: string[];
70	};
71
72	fetch?: typeof fetch;
73	catchDecryptionErrors?: (error: Error) => void;
74	}
75
76	const vercelMs = z	12✔
77	.string()
78	.regex(
79	/^(-?(?:\d+)?\.?\d+) *(milliseconds?\|msecs?\|ms\|seconds?\|secs?\|s\|minutes?\|mins?\|m\|hours?\|hrs?\|h\|days?\|d\|weeks?\|w\|years?\|yrs?\|y)?$/i,
80	"Please provide a valid time span, according to https://github.com/vercel/ms"
81	);
82
83	const timeDuration = (fieldName = "duration") =>	12!
84	z.union([	36✔
85	z.number().min(1, { message: `${fieldName} must be positive` }),
86	vercelMs,
87	]);
88
89	export const cron = z	12✔
90	.string()
91	.refine(
92	isValidRegex,
93	"Please provide a valid Cron expression. See https://github.com/harrisiirak/cron-parser for reference"
94	);
95
96	const EnqueueJobOptionsSchema = z.object({	12✔
97	id: z.string().optional(),
98	exclusive: z.boolean().optional(),
99	override: z.boolean().optional(),
100	retry: z.array(timeDuration("retry")).min(1).max(10).optional(),
101	delay: timeDuration("delay").optional(),
102	runAt: z
103	.date()
104	.refine((date) => Date.now() <= +date, {	1✔
105	message: "runAt must not be in the past",
106	})
107	.optional(),
108	repeat: z
109	.object({
110	every: timeDuration("every").optional(),
111	times: z.number().nonnegative().optional(),
112	cron: cron.optional(),
113	})
114	.optional(),
115	});
116
117	type EnqueueJobOptionsSchema = z.TypeOf<typeof EnqueueJobOptionsSchema>;
118
119	type EnqueueJobOptionssSchemaMatchesDocs = AssertTrue<
120	IsExact<EnqueueJobOptions, EnqueueJobOptionsSchema>
121	>;
122
123	/**
124	* @deprecated renamed to EnqueueJobOptions
125	*/
126	export type EnqueueJobOpts = EnqueueJobOptions;
127
128	export interface EnqueueJobOptions {
129	/**
130	* Can be used to make a job easier to manage.
131	* If there's already a job with the same ID, this job will be trashed.
132	* @tutorial https://demo.quirrel.dev/managed
133	*/
134	id?: string;
135
136	/**
137	* If set to `true`,
138	* no other job (on the same queue)
139	* will be executed at the same time.
140	*/
141	exclusive?: boolean;
142
143	/**
144	* If a job fails, retry it at along the specified intervals.
145	* @example ["5min", "1h", "1d"] // retries it after 5 minutes, 1:05 hours, and 1 day 1:05 hours
146	*/
147	retry?: (number \| string)[];
148
149	/**
150	* Determines what to do when a job
151	* with the same ID already exists.
152	* false: do nothing (default)
153	* true: replace the job
154	*/
155	override?: boolean;
156
157	/**
158	* Will delay the job's execution by the specified amount of milliseconds.
159	* Supports human-readable notation as of @see https://github.com/vercel/ms.
160	* If used together with `repeat`, this will delay the first job to be executed.
161	*/
162	delay?: number \| string;
163
164	/**
165	* Schedules the job for execution at the specified timestamp.
166	*/
167	runAt?: Date;
168
169	repeat?: {
170	/**
171	* Will make the job repeat every X milliseconds.
172	* Supports human-readable notation as of @see https://github.com/vercel/ms.
173	* If `delay` isn't set, the first repetition will be executed immediately.
174	*/
175	every?: number \| string;
176
177	/**
178	* Can be used in conjunction with @field every and @field cron
179	* to limit the number of executions.
180	*/
181	times?: number;
182
183	/**
184	* Schedules the job according to the Cron expression.
185	* @see https://github.com/harrisiirak/cron-parser for supported syntax
186	* If `delay` isn't set, the first repetition will be executed immediately.
187	*/
188	cron?: string;
189	};
190	}
191
192	function parseDuration(value: number \| string \| undefined): number \| undefined {
193	if (!value) {	17✔
194	return undefined;	9✔
195	}
196
197	if (typeof value === "string") {	8!
198	return ms(value);	8✔
199	}
200
201	return value;	×
202	}
203
204	function runAtToDelay(value: Date) {
205	return +value - Date.now();	×
206	}
207
208	function getEncryptor(
209	encryptionSecret: string \| undefined,
210	oldSecrets: string[] = []	15✔
211	) {
212	if (!encryptionSecret) {	15✔
213	return undefined;	4✔
214	}
215
216	return new Encryptor(encryptionSecret, [encryptionSecret, ...oldSecrets]);	11✔
217	}
218
219	function getAuthHeaders(
220	token: string \| undefined
221	): { Authorization: string } \| {} {
222	if (!token) {	15!
223	return {};	15✔
224	}
225
226	return { Authorization: `Bearer ${token}` };	×
227	}
228
229	export class QuirrelClient<T> {	12✔
230	private handler;
231	private route;
232	private defaultJobOptions;
233	private encryptor;
234	private defaultHeaders: Record<string, string>;
235	private quirrelBaseUrl;
236	private baseUrl;
237	private token;
238	private fetch;
239	private catchDecryptionErrors;
240	private signaturePublicKey;
241
242	constructor(args: CreateQuirrelClientArgs<T>) {
243	this.handler = args.handler;	15✔
244	this.defaultJobOptions = args.defaultJobOptions;	15✔
245
246	const token = args.config?.token ?? config.getQuirrelToken();	15!
247	this.defaultHeaders = {	15✔
248	...getAuthHeaders(token),
249	"X-QuirrelClient-Version": pack.version,
250	};
251
252	const quirrelBaseUrl =
253	args.config?.quirrelBaseUrl ?? config.getQuirrelBaseUrl();	15✔
254	const applicationBaseUrl = config.withoutTrailingSlash(	15✔
255	config.prefixWithProtocol(
256	args.config?.applicationBaseUrl ?? config.getApplicationBaseUrl()!	90✔
257	)
258	);
259	this.quirrelBaseUrl = quirrelBaseUrl;	15✔
260	this.baseUrl =	15✔
261	quirrelBaseUrl +
262	"/queues/" +
263	encodeURIComponent(applicationBaseUrl + "/" + args.route);
264	this.token = args.config?.token ?? config.getQuirrelToken();	15!
265	this.route = args.route;	15✔
266	this.encryptor = getEncryptor(	15✔
267	args.config?.encryptionSecret ?? config.getEncryptionSecret(),	90✔
268	args.config?.oldSecrets ?? config.getOldEncryptionSecrets() ?? undefined	120!
269	);
270	this.catchDecryptionErrors = args.catchDecryptionErrors;	15✔
271	this.fetch = args.fetch ?? fetch;	15!
272	this.signaturePublicKey =	15✔
273	args.config?.signaturePublicKey ?? config.getSignaturePublicKey();	90✔
274	}
275
276	async makeRequest(uri: string, init?: RequestInit) {
277	return await this.fetch(this.quirrelBaseUrl + uri, {	2✔
278	credentials: "omit",
279	...init,
280	headers: {
281	...this.defaultHeaders,
282	...init?.headers,	6!
283	},
284	});
285	}
286
287	private async payloadAndOptionsToBody(
288	payload: T,
289	options: EnqueueJobOptionsSchema
290	) {
291	if (typeof payload === "undefined") {	18!
292	throw new Error("Passing `undefined` as Payload is not allowed");	×
293	}
294
295	if (options.repeat && options.retry?.length) {	18✔
296	throw new Error("retry and repeat cannot be used together");	1✔
297	}
298
299	options = EnqueueJobOptionsSchema.parse(options);	17✔
300
301	let delay = parseDuration(options.delay);	13✔
302
303	if ("runAt" in options && options.runAt) {	13!
304	delay = runAtToDelay(options.runAt);	×
305	}
306
307	if (options.repeat) {	13✔
308	options.repeat.every = parseDuration(options.repeat?.every);	1!
309	}
310
311	let stringifiedBody = EnhancedJSON.stringify(payload);	13✔
312
313	if (this.encryptor) {	13✔
314	stringifiedBody = await this.encryptor.encrypt(stringifiedBody);	12✔
315	}
316
317	return {	13✔
318	...this.defaultJobOptions,
319	body: stringifiedBody,
320	delay,
321	id: options.id,
322	repeat: options.repeat,
323	retry: options.retry?.map(parseDuration),	39✔
324	override: options.override,
325	};
326	}
327
328	/**
329	* Enqueue a job to the specified endpoint.
330	* @param options job options
331	*/
332	async enqueue(payload: T, options: EnqueueJobOptions = {}): Promise<Job<T>> {	6✔
333	const body = await this.payloadAndOptionsToBody(payload, options);	15✔
334
335	const res = await this.fetch(this.baseUrl, {	10✔
336	method: "POST",
337	headers: {
338	"Content-Type": "application/json",
339	...this.defaultHeaders,
340	},
341	credentials: "omit",
342	body: JSON.stringify(body),
343	});
344
345	if (res.status === 201) {	8!
346	return await this.toJob(await res.json());	8✔
347	}
348
349	throw new Error(`Unexpected response: ${await res.text()}`);	×
350	}
351
352	/**
353	* Enqueue multiple jobs
354	*/
355	async enqueueMany(
356	jobs: { payload: T; options?: EnqueueJobOptions }[]
357	): Promise<Job<T>[]> {
358	const body = await Promise.all(	1✔
359	jobs.map(({ payload, options = {} }) =>	×
360	this.payloadAndOptionsToBody(payload, options)	3✔
361	)
362	);
363
364	const res = await this.fetch(this.baseUrl + "/batch", {	1✔
365	method: "POST",
366	headers: {
367	"Content-Type": "application/json",
368	...this.defaultHeaders,
369	},
370	credentials: "omit",
371	body: JSON.stringify(body),
372	});
373
374	if (res.status === 201) {	1!
375	const response = (await res.json()) as any[];	1✔
376	return await Promise.all(response.map((job) => this.toJob(job)));	3✔
377	}
378
379	throw new Error(`Unexpected response: ${await res.text()}`);	×
380	}
381
382	private async decryptAndDecodeBody(body: string): Promise<T> {
383	if (this.encryptor) {	22✔
384	if (this.catchDecryptionErrors) {	20✔
385	try {	1✔
386	body = await this.encryptor.decrypt(body);	1✔
387	} catch (error) {
388	this.catchDecryptionErrors(error);	1✔
389	return body as any;	1✔
390	}
391	} else {
392	body = await this.encryptor.decrypt(body);	19✔
393	}
394	}
395
396	return EnhancedJSON.parse(body);	21✔
397	}
398
399	private async toJob(dto: JobDTO): Promise<Job<T>> {
400	return {	18✔
401	...dto,
402	body: await this.decryptAndDecodeBody(dto.body),
403	runAt: new Date(dto.runAt),
404	delete: () => this.delete(dto.id),	×
405	invoke: () => this.invoke(dto.id),	×
406	};
407	}
408
409	/**
410	* Iterate through scheduled jobs.
411	* @example
412	* for await (const jobs of queue.get()) {
413	* // do smth
414	* }
415	*/
416	async *get(): AsyncGenerator<Job<T>[]> {
417	let cursor: number \| null = 0;	1✔
418
419	while (cursor !== null) {	1✔
420	const res = await this.fetch(this.baseUrl + "?cursor=" + cursor, {	1✔
421	headers: this.defaultHeaders,
422	});
423
424	const json = await res.json();	1✔
425
426	const { cursor: newCursor, jobs } = json as {	1✔
427	cursor: number \| null;
428	jobs: JobDTO[];
429	};
430
431	cursor = newCursor;	1✔
432
433	yield await Promise.all(jobs.map((dto) => this.toJob(dto)));	3✔
434	}
435	}
436
437	/**
438	* Get a specific job.
439	* @returns null if no job was found.
440	*/
441	async getById(id: string): Promise<Job<T> \| null> {
442	const res = await this.fetch(this.baseUrl + "/" + id, {	8✔
443	headers: this.defaultHeaders,
444	});
445
446	if (res.status === 404) {	8✔
447	return null;	4✔
448	}
449
450	if (res.status === 200) {	4!
451	return await this.toJob(await res.json());	4✔
452	}
453
454	throw new Error("Unexpected response: " + (await res.text()));	×
455	}
456
457	/**
458	* Schedule a job for immediate execution.
459	* @returns false if job could not be found.
460	*/
461	async invoke(id: string): Promise<boolean> {
462	const res = await this.fetch(this.baseUrl + "/" + id, {	×
463	method: "POST",
464	headers: this.defaultHeaders,
465	});
466
467	if (res.status === 404) {	×
468	return false;	×
469	}
470
471	if (res.status === 204) {	×
472	return true;	×
473	}
474
475	throw new Error("Unexpected response: " + (await res.text()));	×
476	}
477
478	/**
479	* Delete a job, preventing it from executing.
480	* @returns false if job could not be found.
481	*/
482	async delete(id: string): Promise<boolean> {
UNCOV 483	const res = await this.fetch(this.baseUrl + "/" + id, {	×
484	method: "DELETE",
485	headers: this.defaultHeaders,
486	});
487
UNCOV 488	if (res.status === 404) {	×
489	return false;	×
490	}
491
UNCOV 492	if (res.status === 204) {	×
UNCOV 493	return true;	×
494	}
495
496	throw new Error("Unexpected response: " + (await res.text()));	×
497	}
498
499	private isValidSignature(body: string, signature: string): boolean {
500	if (this.signaturePublicKey) {	2!
501	return asymmetric.verify(body, this.signaturePublicKey, signature);	2✔
502	} else {
503	return symmetric.verify(body, this.token!, signature);	×
504	}
505	}
506
507	async respondTo(
508	body: string,
509	headers: IncomingHttpHeaders
510	): Promise<{
511	status: number;
512	headers: Record<string, string>;
513	body: string;
514	}> {
515	if (process.env.NODE_ENV === "production") {	4✔
516	const signature = headers["x-quirrel-signature"];	2✔
517	if (typeof signature !== "string") {	2!
518	return {	×
519	status: 401,
520	headers: {},
521	body: "Signature missing",
522	};
523	}
524
525	if (!this.isValidSignature(body, signature)) {	2✔
526	return {	1✔
527	status: 401,
528	headers: {},
529	body: "Signature invalid",
530	};
531	}
532	}
533
534	const payload = await this.decryptAndDecodeBody(body);	3✔
535	const { id, count, retry, nextRepetition, exclusive } = JSON.parse(	3✔
536	(headers["x-quirrel-meta"] as string) ?? "{}"	9!
537	);
538
539	console.log(`Received job to ${this.route}: `, payload);	3✔
540
541	try {	3✔
542	await this.handler(payload, {	3✔
543	id,
544	count,
545	retry,
546	nextRepetition,
547	exclusive,
548	});
549
550	return {	3✔
551	status: 200,
552	headers: {},
553	body: "OK",
554	};
555	} catch (error) {
556	console.error(error);	×
557	return {	×
558	status: 500,
559	headers: {},
560	body: String(error),
561	};
562	}
563	}
564	}

quirrel-dev / quirrel / 1055705071

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