npm package: @fapoli/pg-scheduler
Description: PostgreSQL-backed distributed task scheduler for Node.js
Latest version: 1.0.3

# @fapoli/pg-scheduler

[![npm](https://img.shields.io/npm/v/@fapoli/pg-scheduler)](https://www.npmjs.com/package/@fapoli/pg-scheduler)
[![GitHub repo](https://img.shields.io/badge/github-fapoli%2Flospec--cli-blue?logo=github)](https://github.com/fapoli/lospec-cli)
[![license](https://img.shields.io/badge/license-MIT-blue)](./LICENSE)

A PostgreSQL-backed distributed task scheduler for Node.js. Inspired by [db-scheduler](https://github.com/kagkarlsson/db-scheduler).

## Features

- Distributed-safe via `FOR UPDATE SKIP LOCKED`
- One-time and recurring tasks
- Heartbeat-based dead execution recovery
- Configurable failure handlers with exponential backoff, fixed delay, or max retries
- Graceful shutdown
- No external dependencies beyond `pg`

## Installation

```bash
npm install @fapoli/pg-scheduler
```

`pg` is a peer dependency — install it if you haven't already:

```bash
npm install pg
```

## Database setup

Run this migration against your PostgreSQL database:

```sql
CREATE TABLE IF NOT EXISTS scheduled_tasks (
  task_name            TEXT        NOT NULL,
  task_instance        TEXT        NOT NULL,
  task_data            JSONB,
  execution_time       TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  picked               BOOLEAN     NOT NULL DEFAULT FALSE,
  picked_by            TEXT,
  last_success         TIMESTAMPTZ,
  last_failure         TIMESTAMPTZ,
  consecutive_failures INT         NOT NULL DEFAULT 0,
  last_heartbeat       TIMESTAMPTZ,
  version              BIGINT      NOT NULL DEFAULT 1,
  priority             INT,
  PRIMARY KEY (task_name, task_instance)
);

CREATE INDEX IF NOT EXISTS idx_scheduled_tasks_picker
  ON scheduled_tasks (priority DESC NULLS LAST, execution_time ASC)
  WHERE picked = FALSE;
```

## Usage

### Register tasks and start the worker

```ts
import { Pool } from 'pg';
import { oneTime, recurring, scheduleTask, startWorker } from '@fapoli/pg-scheduler';

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

// Register a one-time task
oneTime({
  name: 'send-welcome-email',
  run: async (data) => {
    await sendEmail(data.userId);
  },
});

// Register a recurring task
recurring({
  name: 'cleanup-expired-sessions',
  intervalMs: 60 * 60 * 1000, // every hour
  run: async () => {
    await deleteExpiredSessions();
  },
});

// Schedule initial executions (ON CONFLICT DO NOTHING — safe to call on every boot)
await scheduleTask({ pool, taskName: 'send-welcome-email', taskInstance: 'user-123', taskData: { userId: 'user-123' } });
await scheduleTask({ pool, taskName: 'cleanup-expired-sessions', taskInstance: 'default', taskData: null });

// Start the worker
const worker = startWorker({ pool });

// Graceful shutdown
process.on('SIGTERM', async () => {
  await worker.shutdown();
  process.exit(0);
});
```

### One-time tasks

One-time tasks are deleted after successful execution.

```ts
oneTime({
  name: 'my-task',
  run: async (data) => { /* ... */ },
});
```

### Recurring tasks

Recurring tasks reschedule themselves after each successful execution based on `intervalMs`.

```ts
recurring({
  name: 'my-recurring-task',
  intervalMs: 5 * 60 * 1000, // every 5 minutes
  run: async (data) => { /* ... */ },
});
```

### Scheduling tasks

```ts
// Schedule immediately (or supply an executionTime)
await scheduleTask({
  pool,
  taskName: 'my-task',
  taskInstance: 'unique-instance-id',
  taskData: { foo: 'bar' }, // pass null if no data
  executionTime: new Date(Date.now() + 5000), // optional, defaults to now
  tableName: 'scheduled_tasks',              // optional, defaults to 'scheduled_tasks'
});

// Reschedule an existing task (throws if not found or currently running)
await rescheduleTask({
  pool,
  taskName: 'my-task',
  taskInstance: 'unique-instance-id',
  taskData: { foo: 'baz' },
  executionTime: new Date(Date.now() + 10000),
  tableName: 'scheduled_tasks',              // optional, defaults to 'scheduled_tasks'
});
```

### Failure handlers

```ts
import { exponentialBackoff, fixedDelay, maxRetries } from '@fapoli/pg-scheduler';

recurring({
  name: 'my-task',
  intervalMs: 60000,
  run: async () => { /* ... */ },
  failureHandler: exponentialBackoff(1000, 5), // 1s initial delay, max 5 retries
});

// Other built-ins
fixedDelay(5000, 3)   // retry every 5s, max 3 times
maxRetries(3)          // retry immediately, max 3 times
```

### Worker options

```ts
startWorker({
  pool,
  tableName: 'scheduled_tasks',     // default
  workerId: 'my-worker-1',          // default: hostname-pid
  pollingIntervalMs: 10000,         // default: 10s
  batchSize: 10,                    // default: 10
  heartbeatTimeoutMs: 300000,       // default: 5 minutes
  heartbeatIntervalMs: 100000,      // default: heartbeatTimeoutMs / 3
  logger: console,                  // default: console — inject your own
});
```

### Custom logger

```ts
startWorker({
  pool,
  logger: {
    info: (msg) => myLogger.info(msg),
    warn: (msg) => myLogger.warn(msg),
    error: (msg, err) => myLogger.error(msg, err),
  },
});
```

## License

MIT