11162546455

Committed 03 Oct 2024 01:05PM UTC coverage: 88.824% (-0.4%) from 89.213%

Build # 11162546455

Build Type

Pull #2797

github

Committed by

web-flow

Commit Message

Merge a2bc76b94 into 33bb091be

Pull Request Pull Request #2797: feat(queue): add new upsertJobScheduler method

Run Details

1267 of 1523 branches covered (83.19%)

Branch coverage included in aggregate %.

130 of 140 new or added lines in 8 files covered. (92.86%)

12 existing lines in 3 files now uncovered.

2365 of 2566 relevant lines covered (92.17%)

4192.49 hits per line

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

93.2

/src/classes/repeat.ts

import { parseExpression } from 'cron-parser';
import { createHash } from 'crypto';
import {
  RedisClient,
  RepeatBaseOptions,
  RepeatableJob,
  RepeatOptions,
} from '../interfaces';
import { JobsOptions, RepeatStrategy } from '../types';
import { Job } from './job';
import { QueueBase } from './queue-base';
import { RedisConnection } from './redis-connection';

export class Repeat extends QueueBase {
  private repeatStrategy: RepeatStrategy;
  private repeatKeyHashAlgorithm: string;

  constructor(
    name: string,
    opts: RepeatBaseOptions,
    Connection?: typeof RedisConnection,
  ) {
    super(name, opts, Connection);

    this.repeatStrategy =
      (opts.settings && opts.settings.repeatStrategy) || getNextMillis;

    this.repeatKeyHashAlgorithm =
      (opts.settings && opts.settings.repeatKeyHashAlgorithm) || 'md5';
  }

  async updateRepeatableJob<T = any, R = any, N extends string = string>(
    name: N,
    data: T,
    opts: JobsOptions,
    { override }: { override: boolean },
  ): Promise<Job<T, R, N> | undefined> {
    // Backwards compatibility for repeatable jobs for versions <= 3.0.0
    const repeatOpts: RepeatOptions & { cron?: string } = { ...opts.repeat };
    repeatOpts.pattern ??= repeatOpts.cron;
    delete repeatOpts.cron;

    // Check if we reached the limit of the repeatable job's iterations
    const iterationCount = repeatOpts.count ? repeatOpts.count + 1 : 1;
    if (
      typeof repeatOpts.limit !== 'undefined' &&
      iterationCount > repeatOpts.limit
    ) {
      return;
    }

    // Check if we reached the end date of the repeatable job
    let now = Date.now();
    const { endDate } = repeatOpts;
    if (!(typeof endDate === undefined) && now > new Date(endDate!).getTime()) {
      return;
    }

    const prevMillis = opts.prevMillis || 0;
    now = prevMillis < now ? now : prevMillis;

    const nextMillis = await this.repeatStrategy(now, repeatOpts, name);
    const { every, pattern } = repeatOpts;

    const hasImmediately = Boolean(
      (every || pattern) && repeatOpts.immediately,
    );
    const offset = hasImmediately && every ? now - nextMillis : undefined;
    if (nextMillis) {
      // We store the undecorated opts.jobId into the repeat options
      if (!prevMillis && opts.jobId) {
        repeatOpts.jobId = opts.jobId;
      }

      const legacyRepeatKey = getRepeatConcatOptions(name, repeatOpts);
      const newRepeatKey = opts.repeat.key ?? this.hash(legacyRepeatKey);

      let repeatJobKey;
      if (override) {
        repeatJobKey = await this.scripts.addRepeatableJob(
          newRepeatKey,
          nextMillis,
          {
            name,
            endDate: endDate ? new Date(endDate).getTime() : undefined,
            tz: repeatOpts.tz,
            pattern,
            every,
          },
          legacyRepeatKey,
        );
      } else {
        const client = await this.client;

        repeatJobKey = await this.scripts.updateRepeatableJobMillis(
          client,
          newRepeatKey,
          nextMillis,
          legacyRepeatKey,
        );
      }

      const { immediately, ...filteredRepeatOpts } = repeatOpts;

      return this.createNextJob<T, R, N>(
        name,
        nextMillis,
        repeatJobKey,
        { ...opts, repeat: { offset, ...filteredRepeatOpts } },
        data,
        iterationCount,
        hasImmediately,
      );
    }
  }

  private async createNextJob<T = any, R = any, N extends string = string>(
    name: N,
    nextMillis: number,
    repeatJobKey: string,
    opts: JobsOptions,
    data: T,
    currentCount: number,
    hasImmediately: boolean,
  ) {
    //
    // Generate unique job id for this iteration.
    //
    const jobId = this.getRepeatJobKey(name, nextMillis, repeatJobKey, data);

    const now = Date.now();
    const delay =
      nextMillis + (opts.repeat.offset ? opts.repeat.offset : 0) - now;

    const mergedOpts = {
      ...opts,
      jobId,
      delay: delay < 0 || hasImmediately ? 0 : delay,
      timestamp: now,
      prevMillis: nextMillis,
      repeatJobKey,
    };

    mergedOpts.repeat = { ...opts.repeat, count: currentCount };

    return this.Job.create<T, R, N>(this, name, data, mergedOpts);
  }

  // TODO: remove legacy code in next breaking change
  getRepeatJobKey<T = any, N extends string = string>(
    name: N,
    nextMillis: number,
    repeatJobKey: string,
    data: T,
  ) {
    if (repeatJobKey.split(':').length > 2) {
      return this.getRepeatJobId({
        name: name,
        nextMillis: nextMillis,
        namespace: this.hash(repeatJobKey),
        jobId: (data as any)?.id,
      });
    }

    return this.getRepeatDelayedJobId({
      customKey: repeatJobKey,
      nextMillis,
    });
  }

  async removeRepeatable(
    name: string,
    repeat: RepeatOptions,
    jobId?: string,
  ): Promise<number> {
    const repeatConcatOptions = getRepeatConcatOptions(name, {
      ...repeat,
      jobId,
    });
    const repeatJobKey = repeat.key ?? this.hash(repeatConcatOptions);
    const legacyRepeatJobId = this.getRepeatJobId({
      name,
      nextMillis: '',
      namespace: this.hash(repeatConcatOptions),
      jobId: jobId ?? repeat.jobId,
      key: repeat.key,
    });

    return this.scripts.removeRepeatable(
      legacyRepeatJobId,
      repeatConcatOptions,
      repeatJobKey,
    );
  }

  async removeRepeatableByKey(repeatJobKey: string): Promise<number> {
    const data = this.keyToData(repeatJobKey);

    const legacyRepeatJobId = this.getRepeatJobId({
      name: data.name,
      nextMillis: '',
      namespace: this.hash(repeatJobKey),
      jobId: data.id,
    });

    return this.scripts.removeRepeatable(legacyRepeatJobId, '', repeatJobKey);
  }

  private async getRepeatableData(
    client: RedisClient,
    key: string,
    next?: number,
  ): Promise<RepeatableJob> {
    const jobData = await client.hgetall(this.toKey('repeat:' + key));

    if (jobData) {
      return {
        key,
        name: jobData.name,
        endDate: parseInt(jobData.endDate) || null,
        tz: jobData.tz || null,
        pattern: jobData.pattern || null,
        every: jobData.every || null,
        next,
      };
    }

    return this.keyToData(key, next);
  }

  private keyToData(key: string, next?: number): RepeatableJob {
    const data = key.split(':');
    const pattern = data.slice(4).join(':') || null;

    return {
      key,
      name: data[0],
      id: data[1] || null,
      endDate: parseInt(data[2]) || null,
      tz: data[3] || null,
      pattern,
      next,
    };
  }

  async getRepeatableJobs(
    start = 0,
    end = -1,
    asc = false,
  ): Promise<RepeatableJob[]> {
    const client = await this.client;

    const key = this.keys.repeat;
    const result = asc
      ? await client.zrange(key, start, end, 'WITHSCORES')
      : await client.zrevrange(key, start, end, 'WITHSCORES');

    const jobs = [];
    for (let i = 0; i < result.length; i += 2) {
      jobs.push(
        this.getRepeatableData(client, result[i], parseInt(result[i + 1])),
      );
    }
    return Promise.all(jobs);
  }

  async getRepeatableCount(): Promise<number> {
    const client = await this.client;
    return client.zcard(this.toKey('repeat'));
  }

  private hash(str: string) {
    return createHash(this.repeatKeyHashAlgorithm).update(str).digest('hex');
  }

  private getRepeatDelayedJobId({
    nextMillis,
    customKey,
  }: {
    customKey: string;
    nextMillis: number | string;
  }) {
    return `repeat:${customKey}:${nextMillis}`;
  }

  private getRepeatJobId({
    name,
    nextMillis,
    namespace,
    jobId,
    key,
  }: {
    name?: string;
    nextMillis: number | string;
    namespace?: string;
    jobId?: string;
    key?: string;
  }) {
    const checksum = key ?? this.hash(`${name}${jobId || ''}${namespace}`);
    return `repeat:${checksum}:${nextMillis}`;
  }
}

function getRepeatConcatOptions(name: string, repeat: RepeatOptions) {
  const endDate = repeat.endDate ? new Date(repeat.endDate).getTime() : '';
  const tz = repeat.tz || '';
  const pattern = repeat.pattern;
  const suffix = (pattern ? pattern : String(repeat.every)) || '';
  const jobId = repeat.jobId ? repeat.jobId : '';

  return `${name}:${jobId}:${endDate}:${tz}:${suffix}`;
}

export const getNextMillis = (
  millis: number,
  opts: RepeatOptions,
): number | undefined => {
  const pattern = opts.pattern;
  if (pattern && opts.every) {
    throw new Error(
      'Both .pattern and .every options are defined for this repeatable job',
    );
  }

  if (opts.every) {
    return (
      Math.floor(millis / opts.every) * opts.every +
      (opts.immediately ? 0 : opts.every)
    );
  }

  const currentDate =
    opts.startDate && new Date(opts.startDate) > new Date(millis)
      ? new Date(opts.startDate)
      : new Date(millis);
  const interval = parseExpression(pattern, {
    ...opts,
    currentDate,
  });

  try {
    if (opts.immediately) {
      return new Date().getTime();
    } else {
      return interval.next().getTime();
    }
  } catch (e) {
    // Ignore error
  }
};

1	import { parseExpression } from 'cron-parser';	1✔
2	import { createHash } from 'crypto';	1✔
3	import {
4	RedisClient,
5	RepeatBaseOptions,
6	RepeatableJob,
7	RepeatOptions,
8	} from '../interfaces';
9	import { JobsOptions, RepeatStrategy } from '../types';
10	import { Job } from './job';
11	import { QueueBase } from './queue-base';	1✔
12	import { RedisConnection } from './redis-connection';
13
14	export class Repeat extends QueueBase {	1✔
15	private repeatStrategy: RepeatStrategy;
16	private repeatKeyHashAlgorithm: string;
17
18	constructor(
19	name: string,
20	opts: RepeatBaseOptions,
21	Connection?: typeof RedisConnection,
22	) {
23	super(name, opts, Connection);	130✔
24
25	this.repeatStrategy =	130✔
26	(opts.settings && opts.settings.repeatStrategy) \|\| getNextMillis;	261✔
27
28	this.repeatKeyHashAlgorithm =	130✔
29	(opts.settings && opts.settings.repeatKeyHashAlgorithm) \|\| 'md5';	262✔
30	}
31
32	async updateRepeatableJob<T = any, R = any, N extends string = string>(
33	name: N,
34	data: T,
35	opts: JobsOptions,
36	{ override }: { override: boolean },
37	): Promise<Job<T, R, N> \| undefined> {
38	// Backwards compatibility for repeatable jobs for versions <= 3.0.0
39	const repeatOpts: RepeatOptions & { cron?: string } = { ...opts.repeat };	54✔
40	repeatOpts.pattern ??= repeatOpts.cron;	54✔
41	delete repeatOpts.cron;	54✔
42
43	// Check if we reached the limit of the repeatable job's iterations
44	const iterationCount = repeatOpts.count ? repeatOpts.count + 1 : 1;	54!
45	if (	54!
46	typeof repeatOpts.limit !== 'undefined' &&	55✔
47	iterationCount > repeatOpts.limit
48	) {
UNCOV 49	return;	×
50	}
51
52	// Check if we reached the end date of the repeatable job
53	let now = Date.now();	54✔
54	const { endDate } = repeatOpts;	54✔
55	if (!(typeof endDate === undefined) && now > new Date(endDate!).getTime()) {	54!
UNCOV 56	return;	×
57	}
58
59	const prevMillis = opts.prevMillis \|\| 0;	54✔
60	now = prevMillis < now ? now : prevMillis;	54✔
61
62	const nextMillis = await this.repeatStrategy(now, repeatOpts, name);	54✔
63	const { every, pattern } = repeatOpts;	53✔
64
65	const hasImmediately = Boolean(	53✔
66	(every \|\| pattern) && repeatOpts.immediately,	143✔
67	);
68	const offset = hasImmediately && every ? now - nextMillis : undefined;	53✔
69	if (nextMillis) {	53!
70	// We store the undecorated opts.jobId into the repeat options
71	if (!prevMillis && opts.jobId) {	53✔
72	repeatOpts.jobId = opts.jobId;	2✔
73	}
74
75	const legacyRepeatKey = getRepeatConcatOptions(name, repeatOpts);	53✔
76	const newRepeatKey = opts.repeat.key ?? this.hash(legacyRepeatKey);	53✔
77
78	let repeatJobKey;
79	if (override) {	53!
80	repeatJobKey = await this.scripts.addRepeatableJob(	53✔
81	newRepeatKey,
82	nextMillis,
83	{
84	name,
85	endDate: endDate ? new Date(endDate).getTime() : undefined,	53✔
86	tz: repeatOpts.tz,
87	pattern,
88	every,
89	},
90	legacyRepeatKey,
91	);
92	} else {
UNCOV 93	const client = await this.client;	×
94
UNCOV 95	repeatJobKey = await this.scripts.updateRepeatableJobMillis(	×
96	client,
97	newRepeatKey,
98	nextMillis,
99	legacyRepeatKey,
100	);
101	}
102
103	const { immediately, ...filteredRepeatOpts } = repeatOpts;	53✔
104
105	return this.createNextJob<T, R, N>(	53✔
106	name,
107	nextMillis,
108	repeatJobKey,
109	{ ...opts, repeat: { offset, ...filteredRepeatOpts } },
110	data,
111	iterationCount,
112	hasImmediately,
113	);
114	}
115	}
116
117	private async createNextJob<T = any, R = any, N extends string = string>(
118	name: N,
119	nextMillis: number,
120	repeatJobKey: string,
121	opts: JobsOptions,
122	data: T,
123	currentCount: number,
124	hasImmediately: boolean,
125	) {
126	//
127	// Generate unique job id for this iteration.
128	//
129	const jobId = this.getRepeatJobKey(name, nextMillis, repeatJobKey, data);	53✔
130
131	const now = Date.now();	53✔
132	const delay =
133	nextMillis + (opts.repeat.offset ? opts.repeat.offset : 0) - now;	53!
134
135	const mergedOpts = {	53✔
136	...opts,
137	jobId,
138	delay: delay < 0 \|\| hasImmediately ? 0 : delay,	159✔
139	timestamp: now,
140	prevMillis: nextMillis,
141	repeatJobKey,
142	};
143
144	mergedOpts.repeat = { ...opts.repeat, count: currentCount };	53✔
145
146	return this.Job.create<T, R, N>(this, name, data, mergedOpts);	53✔
147	}
148
149	// TODO: remove legacy code in next breaking change
150	getRepeatJobKey<T = any, N extends string = string>(
151	name: N,
152	nextMillis: number,
153	repeatJobKey: string,
154	data: T,
155	) {
156	if (repeatJobKey.split(':').length > 2) {	53✔
157	return this.getRepeatJobId({	1✔
158	name: name,
159	nextMillis: nextMillis,
160	namespace: this.hash(repeatJobKey),
161	jobId: (data as any)?.id,	3!
162	});
163	}
164
165	return this.getRepeatDelayedJobId({	52✔
166	customKey: repeatJobKey,
167	nextMillis,
168	});
169	}
170
171	async removeRepeatable(
172	name: string,
173	repeat: RepeatOptions,
174	jobId?: string,
175	): Promise<number> {
176	const repeatConcatOptions = getRepeatConcatOptions(name, {	9✔
177	...repeat,
178	jobId,
179	});
180	const repeatJobKey = repeat.key ?? this.hash(repeatConcatOptions);	9✔
181	const legacyRepeatJobId = this.getRepeatJobId({	9✔
182	name,
183	nextMillis: '',
184	namespace: this.hash(repeatConcatOptions),
185	jobId: jobId ?? repeat.jobId,	27✔
186	key: repeat.key,
187	});
188
189	return this.scripts.removeRepeatable(	9✔
190	legacyRepeatJobId,
191	repeatConcatOptions,
192	repeatJobKey,
193	);
194	}
195
196	async removeRepeatableByKey(repeatJobKey: string): Promise<number> {
197	const data = this.keyToData(repeatJobKey);	5✔
198
199	const legacyRepeatJobId = this.getRepeatJobId({	5✔
200	name: data.name,
201	nextMillis: '',
202	namespace: this.hash(repeatJobKey),
203	jobId: data.id,
204	});
205
206	return this.scripts.removeRepeatable(legacyRepeatJobId, '', repeatJobKey);	5✔
207	}
208
209	private async getRepeatableData(
210	client: RedisClient,
211	key: string,
212	next?: number,
213	): Promise<RepeatableJob> {
214	const jobData = await client.hgetall(this.toKey('repeat:' + key));	27✔
215
216	if (jobData) {	27!
217	return {	27✔
218	key,
219	name: jobData.name,
220	endDate: parseInt(jobData.endDate) \|\| null,	50✔
221	tz: jobData.tz \|\| null,	48✔
222	pattern: jobData.pattern \|\| null,	38✔
223	every: jobData.every \|\| null,	46✔
224	next,
225	};
226	}
227
228	return this.keyToData(key, next);	×
229	}
230
231	private keyToData(key: string, next?: number): RepeatableJob {
232	const data = key.split(':');	5✔
233	const pattern = data.slice(4).join(':') \|\| null;	5✔
234
235	return {	5✔
236	key,
237	name: data[0],
238	id: data[1] \|\| null,	10✔
239	endDate: parseInt(data[2]) \|\| null,	10✔
240	tz: data[3] \|\| null,	10✔
241	pattern,
242	next,
243	};
244	}
245
246	async getRepeatableJobs(
247	start = 0,	18✔
248	end = -1,	18✔
249	asc = false,	18✔
250	): Promise<RepeatableJob[]> {
251	const client = await this.client;	24✔
252
253	const key = this.keys.repeat;	24✔
254	const result = asc	24✔
255	? await client.zrange(key, start, end, 'WITHSCORES')
256	: await client.zrevrange(key, start, end, 'WITHSCORES');
257
258	const jobs = [];	24✔
259	for (let i = 0; i < result.length; i += 2) {	24✔
260	jobs.push(	27✔
261	this.getRepeatableData(client, result[i], parseInt(result[i + 1])),
262	);
263	}
264	return Promise.all(jobs);	24✔
265	}
266
267	async getRepeatableCount(): Promise<number> {
268	const client = await this.client;	2✔
269	return client.zcard(this.toKey('repeat'));	2✔
270	}
271
272	private hash(str: string) {
273	return createHash(this.repeatKeyHashAlgorithm).update(str).digest('hex');	82✔
274	}
275
276	private getRepeatDelayedJobId({
277	nextMillis,
278	customKey,
279	}: {
280	customKey: string;
281	nextMillis: number \| string;
282	}) {
283	return `repeat:${customKey}:${nextMillis}`;	52✔
284	}
285
286	private getRepeatJobId({
287	name,
288	nextMillis,
289	namespace,
290	jobId,
291	key,
292	}: {
293	name?: string;
294	nextMillis: number \| string;
295	namespace?: string;
296	jobId?: string;
297	key?: string;
298	}) {
299	const checksum = key ?? this.hash(`${name}${jobId \|\| ''}${namespace}`);	15✔
300	return `repeat:${checksum}:${nextMillis}`;	15✔
301	}
302	}
303
304	function getRepeatConcatOptions(name: string, repeat: RepeatOptions) {
305	const endDate = repeat.endDate ? new Date(repeat.endDate).getTime() : '';	62✔
306	const tz = repeat.tz \|\| '';	62✔
307	const pattern = repeat.pattern;	62✔
308	const suffix = (pattern ? pattern : String(repeat.every)) \|\| '';	62!
309	const jobId = repeat.jobId ? repeat.jobId : '';	62✔
310
311	return `${name}:${jobId}:${endDate}:${tz}:${suffix}`;	62✔
312	}
313
314	export const getNextMillis = (	1✔
315	millis: number,
316	opts: RepeatOptions,
317	): number \| undefined => {
318	const pattern = opts.pattern;	64✔
319	if (pattern && opts.every) {	64✔
320	throw new Error(	1✔
321	'Both .pattern and .every options are defined for this repeatable job',
322	);
323	}
324
325	if (opts.every) {	63✔
326	return (	16✔
327	Math.floor(millis / opts.every) * opts.every +
328	(opts.immediately ? 0 : opts.every)	16✔
329	);
330	}
331
332	const currentDate =
333	opts.startDate && new Date(opts.startDate) > new Date(millis)	47✔
334	? new Date(opts.startDate)
335	: new Date(millis);
336	const interval = parseExpression(pattern, {	47✔
337	...opts,
338	currentDate,
339	});
340
341	try {	47✔
342	if (opts.immediately) {	47✔
343	return new Date().getTime();	2✔
344	} else {
345	return interval.next().getTime();	45✔
346	}
347	} catch (e) {
348	// Ignore error
349	}
350	};

taskforcesh / bullmq / 11162546455

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