Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | 11x 34x 34x 34x 13x 1x 12x 5x 8x 4x 4x 6x 6x 3x 2x 2x 1x 1x 10x 5x 4x 4x 4x 4x 4x 1x 1x 1x 1x 1x 1x 5x 5x 5x 5x 5x 2x 2x 2x 1x 5x 5x 5x | import pino from "pino"
import DynamicConfig, { VConfig } from "../../modules/config/dynamic"
import { getNotifier } from "../../modules/notifiers"
import { redis } from "../../redis"
import { renderTemplate, TemplateContextMap, TemplateName } from "../../utils/renderTemplate"
import { getQueue } from "../queue"
const logger = pino()
export abstract class BaseJob {
abstract readonly id: string
readonly retryable: boolean = true
readonly startDelay: number = 0
readonly retryDelay: number = 60
readonly uniqueNotificationKey?: string
readonly cronPattern?: string
readonly cronConfigKey?: VConfig
getStartDelay(asap: boolean = false): number {
if (asap) {
return 2000 // 2 seconds
}
return this.startDelay * 1000
}
getRetryDelay(retryCount: number): number {
return retryCount * this.retryDelay * 1000
}
async resolveCronPattern(): Promise<string | undefined> {
if (this.cronConfigKey) {
return (await DynamicConfig.get(this.cronConfigKey)) ?? undefined
}
return this.cronPattern
}
async init(): Promise<void> {
const pattern = await this.resolveCronPattern()
if (pattern) {
await this.scheduleCronJob(pattern)
}
}
async rescheduleCronJob(): Promise<void> {
const pattern = await this.resolveCronPattern()
if (pattern) {
await this.scheduleCronJob(pattern)
} else {
await this.removeCronJob()
}
}
private get repeatJobId(): string {
return `${this.id}-repeat`
}
// The scheduler key BullMQ registers the repeatable job under.
private get schedulerId(): string {
return `${this.repeatJobId}-repeat`
}
private async scheduleCronJob(pattern: string): Promise<void> {
const queue = await getQueue()
const id = this.repeatJobId
logger.info("Setting up scheduler for %s with cron '%s'", id, pattern)
try {
await queue.upsertJobScheduler(this.schedulerId, { pattern }, { name: id, data: { job: id } })
} catch (err) {
logger.error({ err }, "Failed to set up scheduler for job %s", id)
}
}
private async removeCronJob(): Promise<void> {
const queue = await getQueue()
const id = this.repeatJobId
logger.info("Removing scheduler for %s", id)
try {
await queue.removeJobScheduler(this.schedulerId)
} catch (err) {
logger.error({ err }, "Failed to remove scheduler for job %s", id)
}
}
async sendUniqueNotification<T extends TemplateName>(template: T, data: TemplateContextMap[T]): Promise<void> {
Iif (!this.uniqueNotificationKey) {
throw new Error("uniqueNotificationKey is not set for this job")
}
const notifier = await getNotifier()
Iif (!notifier) {
logger.warn("No notifier configured, skipping notification for job %s", this.id)
return
}
const previousNotificationId = await redis.get(this.uniqueNotificationKey)
if (previousNotificationId) {
logger.info("Deleting previous notification with ID %s", previousNotificationId)
try {
await notifier.deleteMessage(previousNotificationId)
} catch (err) {
logger.error({ err }, "Failed to delete previous notification with ID %s", previousNotificationId)
}
}
const { title, content } = await renderTemplate(template, data)
const notificationId = await notifier.sendMessage(title, content)
await redis.set(this.uniqueNotificationKey, notificationId)
}
}
export abstract class SimpleJob extends BaseJob {
abstract run(): Promise<void>
}
export abstract class TransactionJob extends BaseJob {
abstract run(transactionId: string): Promise<void>
}
export abstract class BudgetJob extends BaseJob {
abstract run(budgetId: string): Promise<void>
}
export abstract class EndpointJob extends BaseJob {
abstract run(transactionId: string, data: unknown): Promise<void>
}
|