@racunis/postgresql

@racunis/postgresql is a TypeScript-based, robust, and flexible job queueing system designed for PostgreSQL databases. It facilitates easy management of background tasks in Node.js applications using PostgreSQL. The system abstracts the complexities of database operations and job processing, offering a straightforward and powerful API for job queueing and processing.

Features

• PostgreSQL-specific implementations of core abstractions.
• Type definitions for jobs and events.
• Easy to extend and integrate into existing applications.
• Configurable options for queues and workers.
• Support for handling multiple job types within a worker.
• Event handling for job states and errors.

To install @racunis/postgresql, use one of the following commands:

# npm
npm install @racunis/postgresql

# yarn
yarn add @racunis/postgresql

# pnpm
pnpm add @racunis/postgresql

# bun
bun add @racunis/postgresql
Copy

Usage

Example

Here is a complete example of how to use @racunis/postgresql to manage a job queue and worker with PostgreSQL.

import { PostgreSqlQueue, PostgreSqlWorker } from '@racunis/postgresql'

interface JobPayload {
  task: string
}

const main = async () => {
  // Create the queue
  const queue = await PostgreSqlQueue.create<JobPayload>('Queue', {
    connectionString: 'postgresql://user:password@localhost:port/db',
  })

  // Add a job to the queue
  await queue.add({ task: 'processData' }, 10)

  // Create a worker to process jobs from the queue
  const worker = PostgreSqlWorker.create(queue, (job) => {
    console.log('Processing job:', job.payload.task)
  })
}

try {
  await main()
}
catch (error) {
  console.error(error)
}
Copy

Handling Multiple Job Types

To handle different job types within a worker, you can use a switch statement to process each job type accordingly.

import { PostgreSqlQueue, PostgreSqlWorker } from '@racunis/postgresql'
import type { Job } from '@racunis/core'

// Define job payload interfaces for different job types
export interface EmailProcessingJobPayload {
  type: 'EmailProcessing'
  details: {
    recipient: string
    content: string
  }
}

export interface DataMigrationJobPayload {
  type: 'DataMigration'
  details: {
    source: string
    destination: string
  }
}

export interface ReportGenerationJobPayload {
  type: 'ReportGeneration'
  details: {
    reportId: string
    requestedBy: string
  }
}

// Union type for job payloads
export type JobPayload = EmailProcessingJobPayload | DataMigrationJobPayload | ReportGenerationJobPayload

const main = async () => {
  // Create the queue
  const queue = await PostgreSqlQueue.create<JobPayload>('Queue', {
    connectionString: 'postgresql://user:password@localhost:port/db',
  })

  // Add jobs to the queue
  await queue.add({ type: 'EmailProcessing', details: { recipient: 'user@example.com', content: 'Hello, World!' } }, 5)
  await queue.add({ type: 'DataMigration', details: { source: 'System A', destination: 'System B' } }, 3)
  await queue.add({ type: 'ReportGeneration', details: { reportId: 'R123', requestedBy: 'admin' } }, 4)

  // Create a worker to process jobs from the queue
  const worker = PostgreSqlWorker.create(queue, (job) => {
    const { type, details } = job.payload
    switch (type) {
      case 'EmailProcessing':
        console.log(`Processing Email Job: Recipient - ${details.recipient}, Content - ${details.content}`)
        break
      case 'DataMigration':
        console.log(`Processing Data Migration Job: Source - ${details.source}, Destination - ${details.destination}`)
        break
      case 'ReportGeneration':
        console.log(`Processing Report Generation Job: Report ID - ${details.reportId}, Requested By - ${details.requestedBy}`)
        break
    }
  })
}

try {
  await main()
}
catch (error) {
  console.error(error)
}
Copy

Queue and Worker Options

@racunis/postgresql provides configurable options for queues and workers to customize their behavior.

Queue Options

• autostart: Automatically start processing jobs when the queue is created. Default is true.

  const queue = await PostgreSqlQueue.create<JobPayload>('Queue', {
    connectionString: 'postgresql://user:password@localhost:port/db',
  }, {
    autostart: false, // Do not start automatically
  })
  Copy

Worker Options

• autostart: Automatically start the worker when it is created. Default is true.

• processingInterval: Interval in milliseconds between job processing attempts. Default is 0.

• waitingInterval: Interval in milliseconds for checking new jobs when the worker is waiting. Default is 1000.

• maxRetries: Maximum number of retries for processing a job. Default is 3.

• retryInterval: Interval in milliseconds between retry attempts. Default is 500.

  const worker = PostgreSqlWorker.create<JobPayload>(queue, (job) => {
    console.log('Processing job:', job.payload.task)
  }, {
    autostart: false, // Do not start automatically
    processingInterval: 1000, // Process jobs every 1 second
    waitingInterval: 2000, // Check for new jobs every 2 seconds
    maxRetries: 5, // Retry failed jobs up to 5 times
    retryInterval: 1000, // Wait 1 second between retries
  })
  Copy

These options allow you to tailor the behavior of queues and workers to fit your specific requirements.

Queue and Worker Events

@racunis/postgresql supports various events for job queues and workers to handle job states and errors effectively. Below is an overview of the events and how you can use them.

Queue Events

These events are triggered by the queue during different stages of job processing.

• activated: Emitted when a job is activated.
• completed: Emitted when a job is completed.
• failed: Emitted when a job fails.
• error: Emitted when an error occurs.

import { PostgreSqlQueue } from '@racunis/postgresql'
import type { Job } from '@racunis/core'

interface JobPayload {
  type: string
  details: any
}

const main = async () => {
  // Create the queue
  const queue = await PostgreSqlQueue.create<JobPayload>('Queue', {
    connectionString: 'postgresql://user:password@localhost:port/db',
  })

  // Event listener for when a job is activated
  queue.on('activated', ({ job }) => {
    console.log(`Job activated: ${job.id}`)
  })

  // Event listener for when a job is completed
  queue.on('completed', ({ job }) => {
    console.log(`Job completed: ${job.id}`)
  })

  // Event listener for when a job fails
  queue.on('failed', ({ job, error }) => {
    console.log(`Job failed: ${job.id}, Error: ${error.message}`)
  })

  // Event listener for queue errors
  queue.on('error', ({ error }) => {
    console.error(`Queue error: ${error.message}`)
  })

  // Add jobs to the queue
  await queue.add({ type: 'EmailProcessing', details: { recipient: 'user@example.com', content: 'Hello, World!' } }, 5)
}

try {
  await main()
}
catch (error) {
  console.error(error)
}
Copy

Worker Events

These events are triggered by the worker while processing jobs.

• waiting: Emitted when the worker is waiting for a job.
• activated: Emitted when a job is activated.
• completed: Emitted when a job is completed.
• failed: Emitted when a job fails.

import { PostgreSqlQueue, PostgreSqlWorker } from '@racunis/postgresql'
import type { Job } from '@racunis/core'

interface JobPayload {
  type: string
  details: any
}

const main = async () => {
  // Create the queue
  const queue = await PostgreSqlQueue.create<JobPayload>('Queue', {
    connectionString: 'postgresql://user:password@localhost:port/db',
  })

  // Create a worker to process jobs from the queue
  const worker = PostgreSqlWorker.create(queue, (job) => {
    const { type, details } = job.payload
    switch (type) {
      case 'EmailProcessing':
        console.log(`Processing Email Job: Recipient - ${details.recipient}, Content - ${details.content}`)
        break
      case 'DataMigration':
        console.log(`Processing Data Migration Job: Source - ${details.source}, Destination - ${details.destination}`)
        break
      case 'ReportGeneration':
        console.log(`Processing Report Generation Job: Report ID - ${details.reportId}, Requested By - ${details.requestedBy}`)
        break
    }
  }, { autostart: false })

  // Event listener for when the worker is waiting for a job
  worker.on('waiting', () => {
    console.log('Worker is waiting for a job...')
  })

  // Event listener for when a job is activated
  worker.on('activated', ({ job }) => {
    console.log(`Job activated: ${job.id}`)
  })

  // Event listener for when a job is completed
  worker.on('completed', ({ job }) => {
    console.log(`Job completed: ${job.id}`)
  })

  // Event listener for when a job fails
  worker.on('failed', ({ job, error }) => {
    console.log(`Job failed: ${job.id}, Error: ${error.message}`)
  })

  // Start the worker
  worker.start()
}

try {
  await main()
}
catch (error) {
  console.error(error)
}
Copy

Queue and Worker Control

@racunis/postgresql provides mechanisms to control the execution of queues and workers, as well as methods to manage the state of jobs in the queue.

Getting Job Counts

You can get the count of jobs in various states using the getJobCounts method. This is useful for monitoring and managing the state of your job queue.

const jobCounts = await queue.getJobCounts('waiting', 'active', 'completed', 'failed')
console.log(`Waiting: ${jobCounts.waiting}`)
console.log(`Active: ${jobCounts.active}`)
console.log(`Completed: ${jobCounts.completed}`)
console.log(`Failed: ${jobCounts.failed}`)
Copy

This will give you an object with the counts of jobs in the specified states, helping you keep track of the queue's status.

Starting and Stopping Queues and Workers

You can start and stop job processing in queues and workers using the start and stop methods.

• Queue Start/Stop: Use start() to begin processing jobs in the queue and stop() to halt job processing.

  await queue.start() // Start processing jobs
  await queue.stop()  // Stop processing jobs
  Copy

• Worker Start/Stop: Use start() to begin job processing by the worker and stop() to halt job processing.

  worker.start() // Start the worker
  await worker.stop() // Stop the worker
  Copy

If you attempt to start a worker on a stopped queue, the worker will not start. Conversely, if you restart a queue that has a previously stopped worker, the worker will automatically start processing jobs again.

Draining and Emptying the Queue

@racunis/postgresql provides methods to manage the jobs in the queue:

• drain(): Deletes all waiting jobs in the queue.

  await queue.drain() // Clear all waiting jobs
  Copy

• empty(): Deletes all jobs in all states (waiting, active, completed, and failed).

  await queue.empty() // Clear all jobs in the queue
  Copy

These methods allow you to maintain control over the job queue, ensuring you can clear jobs as needed.

Closing the Queue

The close method allows you to close the queue and release all associated resources. This is particularly useful for clean-up operations and ensuring that all connections are properly terminated.

await queue.close() // Close the queue and release resources
Copy

This method stops the queue from processing jobs, closes all associated workers, and releases any resources used by the queue, such as database connections.

Contributing

I appreciate your interest in contributing! Please see the CONTRIBUTING.md file for guidelines on how to contribute.

License

@racunis/core is licensed under the MIT License. See the LICENSE file for more details.