Something went wrong. Try again.
🔥 🔥 🔥 A Free & Self-hostable Airtable Alternative nocodb.com
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513export interface RateLimitOptions { enabled: boolean maxRequestsPerWindow: number windowSizeMs: number}
export interface QueueOptions { maxConcurrent: number rateLimit?: Partial<RateLimitOptions> autoStart?: boolean priorityLevels?: number retryOptions?: RetryOptions timeout?: number}export interface RetryOptions { maxRetries: number retryDelay: number | ((attempt: number) => number) retryCondition?: (error: any) => boolean}
export interface QueueTask<T> { id: string task: () => Promise<T> priority: number addedAt: number attempts: number timeout?: number resolve: (value: T | PromiseLike<T>) => void reject: (reason?: any) => void abortController?: AbortController}
export enum QueueEvents { TASK_ADDED = 'taskAdded', TASK_STARTED = 'taskStarted', TASK_COMPLETED = 'taskCompleted', TASK_FAILED = 'taskFailed', TASK_RETRIED = 'taskRetried', QUEUE_IDLE = 'queueIdle', QUEUE_PAUSED = 'queuePaused', QUEUE_RESUMED = 'queueResumed',}
export type QueueEventCallback = (data: any) => void
export class Queue { private queue: QueueTask<any>[] = [] private runningTasks: Map<string, QueueTask<any>> = new Map() private runningCount = 0 private requestTimestamps: number[] = [] private rateLimitOptions: RateLimitOptions private taskCounter = 0 private paused = false private eventListeners: Map<QueueEvents, QueueEventCallback[]> = new Map() private options: QueueOptions
constructor(options: Partial<QueueOptions> = {}) { this.options = { maxConcurrent: 4, autoStart: true, priorityLevels: 1, ...options, }
this.rateLimitOptions = { enabled: false, maxRequestsPerWindow: 10, windowSizeMs: 1000, ...(this.options.rateLimit || {}), } }
/** * Add a task to the queue * @param task The task function to execute * @param options Optional configuration for this specific task * @returns Promise that resolves with the task result */ async add<T>( task: () => Promise<T>, options: { priority?: number id?: string timeout?: number signal?: AbortSignal } = {}, ): Promise<T> { return new Promise<T>((resolve, reject) => { const id = options.id || `task-${++this.taskCounter}` const priority = options.priority !== undefined ? Math.min(Math.max(0, options.priority), (this.options.priorityLevels || 1) - 1) : 0
const abortController = new AbortController()
if (options.signal) { if (options.signal.aborted) { reject(new Error('Task was cancelled before being added to queue')) return } options.signal.addEventListener('abort', () => abortController.abort()) }
const queueTask: QueueTask<T> = { id, task, priority, addedAt: Date.now(), attempts: 0, timeout: options.timeout, resolve, reject, abortController, }
// Add task to queue with priority sorting this.queue.push(queueTask) this.sortQueue()
this.emit(QueueEvents.TASK_ADDED, { id, priority })
// If autoStart is enabled and we're not paused, try to run tasks if (this.options.autoStart && !this.paused) { this.tryRunNext() } }) }
/** * Add multiple tasks to the queue at once * @param tasks Array of tasks to add * @param options * @returns Promise that resolves when all tasks are completed */ async addBatch<T>( tasks: Array<() => Promise<T>>, options: { priority?: number concurrency?: number } = {}, ): Promise<T[]> { const promises = tasks.map((task) => this.add(task, { priority: options.priority, }), )
return Promise.all(promises) }
/** * Execute a task from the queue * @param queueTask The task to execute */ private async executeTask<T>(queueTask: QueueTask<T>): Promise<void> { const { id, task, resolve, reject } = queueTask queueTask.attempts++
this.runningTasks.set(id, queueTask) this.runningCount++ this.emit(QueueEvents.TASK_STARTED, { id, attempts: queueTask.attempts })
let shouldCleanup = true
try { // Check if task was cancelled before execution if (queueTask.abortController?.signal.aborted) { throw new Error('Task was cancelled before execution') }
// Check if we need to wait for rate limit await this.waitForRateLimit()
// Check again after waiting (task might have been cancelled) if (queueTask.abortController?.signal.aborted) { throw new Error('Task was cancelled during rate limit wait') }
// Record this request timestamp for rate limiting if (this.rateLimitOptions.enabled) { this.requestTimestamps.push(Date.now()) }
// Execute the task with optional timeout and cancellation let result: T if (this.options.timeout || queueTask.timeout) { const timeoutMs = queueTask.timeout || this.options.timeout result = await this.withTimeoutAndCancellation(task(), timeoutMs!, queueTask.abortController?.signal) } else { result = await this.withCancellation(task(), queueTask.abortController?.signal) }
// Task completed successfully resolve(result) this.emit(QueueEvents.TASK_COMPLETED, { id }) } catch (error) { // Check if this was a cancellation if (queueTask.abortController?.signal.aborted || (error as Error).message.includes('cancelled')) { reject(new Error(`Task ${id} was cancelled`)) this.emit(QueueEvents.TASK_FAILED, { id, error: 'Task cancelled', attempts: queueTask.attempts }) } else { // Check if we should retry the task if (this.shouldRetry(queueTask, error)) { this.handleRetry(queueTask, error) shouldCleanup = false // Don't cleanup here, handleRetry will manage it return } else { // No more retries, reject the promise reject(error) this.emit(QueueEvents.TASK_FAILED, { id, error, attempts: queueTask.attempts }) } } } finally { // Only cleanup if we're not retrying if (shouldCleanup) { this.runningTasks.delete(id) this.runningCount--
// If the queue is empty and no tasks are running, emit idle event if (this.queue.length === 0 && this.runningCount === 0) { this.emit(QueueEvents.QUEUE_IDLE, {}) }
// Try to run next task if not paused if (!this.paused) { this.tryRunNext() } } } }
/** * Wrap a promise with cancellation support */ private async withCancellation<T>(promise: Promise<T>, signal?: AbortSignal): Promise<T> { if (!signal) return promise
return Promise.race([ promise, new Promise<T>((_resolve, reject) => { if (signal.aborted) { reject(new Error('Task was cancelled')) return } signal.addEventListener('abort', () => reject(new Error('Task was cancelled'))) }), ]) }
/** * Wrap a promise with timeout and cancellation */ private async withTimeoutAndCancellation<T>(promise: Promise<T>, timeoutMs: number, signal?: AbortSignal): Promise<T> { const promises: Promise<T>[] = [promise]
// Add timeout promise promises.push( new Promise<T>((_resolve, reject) => setTimeout(() => reject(new Error(`Task timed out after ${timeoutMs}ms`)), timeoutMs)), )
// Add cancellation promise if signal provided if (signal) { promises.push( new Promise<T>((_resolve, reject) => { if (signal.aborted) { reject(new Error('Task was cancelled')) return } signal.addEventListener('abort', () => reject(new Error('Task was cancelled'))) }), ) }
return Promise.race(promises) }
/** * Determine if a task should be retried */ private shouldRetry(queueTask: QueueTask<any>, error: any): boolean { const retryOptions = this.options.retryOptions if (!retryOptions) return false
const underMaxRetries = queueTask.attempts <= retryOptions.maxRetries
// If there's a retry condition function, use it if (retryOptions.retryCondition) { return underMaxRetries && retryOptions.retryCondition(error) }
return underMaxRetries }
/** * Handle retrying a failed task */ private handleRetry(queueTask: QueueTask<any>, error: any) { const retryOptions = this.options.retryOptions!
// Remove from running tasks since we're going to retry this.runningTasks.delete(queueTask.id) this.runningCount--
// Create new abort controller for retry queueTask.abortController = new AbortController()
// Calculate delay based on retry count let delay: number if (typeof retryOptions.retryDelay === 'function') { delay = retryOptions.retryDelay(queueTask.attempts) } else { delay = retryOptions.retryDelay }
this.emit(QueueEvents.TASK_RETRIED, { id: queueTask.id, attempts: queueTask.attempts, delay, error, })
// Add task back to queue after delay setTimeout(() => { this.queue.push(queueTask) this.sortQueue() this.tryRunNext() }, delay) }
/** * Wait for rate limit if necessary */ private async waitForRateLimit(): Promise<void> { // Skip rate limiting if disabled if (!this.rateLimitOptions.enabled) { return }
// Clean up old timestamps outside the window const now = Date.now() const windowStart = now - this.rateLimitOptions.windowSizeMs this.requestTimestamps = this.requestTimestamps.filter((timestamp) => timestamp > windowStart)
// If we're at the rate limit, wait until we can proceed if (this.requestTimestamps.length >= this.rateLimitOptions.maxRequestsPerWindow) { const oldestTimestamp = this.requestTimestamps[0]
// Check if oldestTimestamp is defined if (oldestTimestamp !== undefined) { const timeToWait = oldestTimestamp + this.rateLimitOptions.windowSizeMs - now
if (timeToWait > 0) { await new Promise((resolve) => setTimeout(resolve, timeToWait)) // Recursively check again after waiting return this.waitForRateLimit() } } } }
/** * Sort the queue by priority (higher values first) */ private sortQueue(): void { this.queue.sort((a, b) => { // First sort by priority (descending) if (b.priority !== a.priority) { return b.priority - a.priority } // Then by age (older tasks first) return a.addedAt - b.addedAt }) }
/** * Try to run the next task from the queue */ private tryRunNext(): void { // Schedule this to run on the next tick to avoid stack overflow // when many tasks are added at once setTimeout(() => this.runNext(), 0) }
/** * Run the next task from the queue if possible */ private runNext(): void { if (this.paused || this.queue.length === 0) return if (this.runningCount >= this.options.maxConcurrent) return
const nextTask = this.queue.shift() if (nextTask) { this.executeTask(nextTask).catch(() => { // Errors are already handled in executeTask }) } }
/** * Pause the queue - stops processing new tasks but allows running tasks to complete */ public pause(): void { if (this.paused) return this.paused = true this.emit(QueueEvents.QUEUE_PAUSED, {}) }
/** * Resume processing tasks from the queue */ public resume(): void { if (!this.paused) return this.paused = false this.emit(QueueEvents.QUEUE_RESUMED, {})
// Try to run tasks if any are queued for (let i = 0; i < this.options.maxConcurrent; i++) { this.tryRunNext() } }
/** * Clear all pending tasks from the queue * @returns The number of tasks cleared */ public clear(): number { const count = this.queue.length
// Cancel all pending tasks for (const task of this.queue) { task.abortController?.abort() task.reject(new Error('Task was cancelled via clear()')) } this.queue = [] return count }
/** * Remove a specific task from the queue by ID * This will cancel running tasks and remove pending tasks * @param id The ID of the task to remove * @returns True if the task was found and removed/cancelled */ public remove(id: string): boolean { let found = false
// Check if task is currently running const runningTask = this.runningTasks.get(id) if (runningTask) { // Cancel the running task runningTask.abortController?.abort() found = true this.emit(QueueEvents.TASK_FAILED, { id, error: 'Task cancelled via remove()', attempts: runningTask.attempts }) }
this.queue = this.queue.filter((task) => { if (task.id === id) { // Cancel pending task task.abortController?.abort() task.reject(new Error(`Task ${id} was cancelled`)) found = true return false } return true })
return found }
/** * Register an event listener * @param event The event to listen for * @param callback Function to call when event occurs */ public on(event: QueueEvents, callback: QueueEventCallback): void { if (!this.eventListeners.has(event)) { this.eventListeners.set(event, []) } this.eventListeners.get(event)!.push(callback) }
/** * Remove an event listener * @param event The event to remove listener from * @param callback The callback to remove */ public off(event: QueueEvents, callback: QueueEventCallback): void { if (!this.eventListeners.has(event)) return
const listeners = this.eventListeners.get(event)! const index = listeners.indexOf(callback) if (index !== -1) { listeners.splice(index, 1) } }
/** * Emit an event * @param event The event to emit * @param data Data to pass to callbacks */ private emit(event: QueueEvents, data: any): void { if (!this.eventListeners.has(event)) return
for (const callback of this.eventListeners.get(event)!) { try { callback(data) } catch (error) { console.error(`Error in ${event} event handler:`, error) } } }}