Build a Restart-Safe Email Scheduler

Build a durable email scheduler with rate limits, recovery, and Slack alerts.

Introduction

30 Second Summary

A scheduled email can become a broken promise when the app closes before delivery time. A busy hour can also send too much mail before anyone notices.

In this project, you will build a restart-safe email scheduling backend in TypeScript with Slack alerts for breached sender limits. Your backend will keep delayed work durable through restarts.

What You'll Build

You will watch a delayed email reach its Ethereal Email preview exactly once after restarting the scheduler.

By the end of this project, you'll have:

A frontend-ready REST API that returns the same stored email when a client repeats a scheduling request with one idempotency key.
Durable delivery that preserves delayed jobs through API or worker restarts. Every successful record includes an Ethereal Email preview link.
Operational visibility through Bull Board for live queue inspection. Distributed rate limiting keeps parallel workers inside the same hourly quota. Slack posts one alert when that quota is reached.
Secret Mission: Run a five-email burst that forces deferrals. Capture one Slack alert. Restart the processes while a future job waits. Prove the remaining email reaches sent once.

Are there any prerequisites?

This project requires a Linux machine plus access to a Slack workspace where you can install an app. The guide covers every required tool setup.

Before We Start

Before any hands-on work begins, this checkpoint locks in what you are building. The backend gives tenants durable delayed email delivery that survives process interruptions without duplicate sends.

Start the Persistent Development Stack

A scheduler loses its promises when email records or queue work disappear with the application process. Durable storage gives every scheduled email somewhere to survive an interruption.

You will prepare Node.js, Docker, PostgreSQL, and Redis on Linux. You will also create the TypeScript workspace that the API uses in the next step.

In this step, get ready to:

Verify the Linux development tools.
Create the TypeScript workspace with its local configuration.
Start PostgreSQL and Redis with persistent Docker volumes.

Verify the Linux toolchain

The project pins Node.js to v24.20.0 so every dependency runs against the same runtime. npm arrives with Node.js.

Check Node.js, npm, Git, and Docker Compose by running:

node --version
npm --version
git --version
docker compose version

What do these checks prove?

The first command confirms that the Node.js runtime is available.
The second command confirms that npm can install the project dependencies.
The third command confirms that Git is available for source control.
The final command confirms that Docker Compose can manage the local services.

✔️ I see v24.20.0 and all tools

Your toolchain is ready. Node.js reports v24.20.0. The remaining commands each report a version.

Open your installed code editor before continuing.

ⓧ I see an older Node.js version

The installed Node.js release does not match the project runtime. Upgrade it before installing the dependencies.

Open the official Node.js v24.20.0 release page.
Download the binary that matches your Linux architecture.
Follow the installation method shown for your Linux environment.
Open a new terminal after the installation finishes.
Run the four tool checks again.
Open your installed code editor after every command reports a version.

ⓧ One or more commands are missing

A missing command identifies the tool that still needs installation. Use each vendor's Linux guide so your package source matches your distribution.

Install Node.js v24.20.0 from the official Node.js download page if the Node.js or npm check failed.
Install Git from the official Git Linux installation page if the Git check failed.
Install Docker Engine from the official Docker Engine guide if the Docker check failed.
Include the Docker Compose plugin during the Docker Engine installation.
Install a code editor that can create TypeScript files.
Open a new terminal after the installations finish.
Run the four tool checks again.

Still missing a tool?

Close the terminal after an installation completes. A new terminal reloads the executable paths supplied by the installer.

Use the installation page for your exact Linux distribution if an apt command is unavailable.

Help me troubleshoot my Linux Node.js, npm, Git, or Docker setup.

Create and configure the TypeScript workspace

The workspace keeps dependency versions, compiler rules, runtime settings, and documentation together. The folder name is reachinbox-email-scheduler.

Move to your Desktop by running this command:

cd ~/Desktop

What does this command do?

This makes your Desktop the starting location. You will know exactly where to find the project folder.

Create the project folder with its source folder by running:

mkdir reachinbox-email-scheduler
cd reachinbox-email-scheduler
mkdir src

What does this setup create?

The first directory holds the complete project. The nested src directory holds the TypeScript source files.

Confirm the source folder exists by running:

ls

What should I see?

You should see src in the output. This confirms that your terminal is inside reachinbox-email-scheduler.

Open the reachinbox-email-scheduler folder in your installed editor.
Create package.json inside reachinbox-email-scheduler.
Paste this package definition into package.json:

{
  "name": "reachinbox-email-scheduler",
  "version": "1.0.0",
  "private": true,
  "type": "module",
  "scripts": {
    "build": "tsc -p tsconfig.json",
    "dev:api": "tsx watch src/server.ts",
    "dev:worker": "tsx watch src/worker.ts",
    "start:api": "node dist/server.js",
    "start:worker": "node dist/worker.js"
  },
  "dependencies": {
    "@bull-board/api": "9.8.0",
    "@bull-board/express": "9.8.0",
    "bullmq": "6.3.4",
    "dotenv": "17.4.2",
    "express": "5.2.1",
    "ioredis": "6.0.0",
    "nodemailer": "10.0.1",
    "pg": "8.23.0"
  },
  "devDependencies": {
    "@types/express": "5.0.6",
    "@types/node": "26.5.0",
    "@types/pg": "8.23.1",
    "tsx": "4.23.13",
    "typescript": "7.0.2"
  }
}

What does this file control?

The scripts define development processes for the API and worker.
The dependencies provide HTTP routing, delayed jobs, database access, Redis access, test email delivery, queue inspection, and environment loading.
The development dependencies provide TypeScript compilation, file watching, and type declarations.

Save package.json.
Install the pinned dependencies by running:

npm install

What does npm install create?

npm reads the exact dependency versions from package.json. It downloads them into node_modules and records the resolved dependency tree.

Good progress. Your project now has the libraries needed for the API, worker, database, queue, email transport, and Slack integration.

Did the dependency installation fail?

Confirm that the terminal is inside reachinbox-email-scheduler. Confirm that package.json contains valid JSON.

A network or registry failure can interrupt the download. Retry after your connection is stable.

Help me troubleshoot npm install for this project.

Create tsconfig.json inside reachinbox-email-scheduler.
Paste these compiler settings into tsconfig.json:

{
  "compilerOptions": {
    "target": "ES2023",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "rootDir": "src",
    "outDir": "dist",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "noUncheckedIndexedAccess": true
  },
  "include": ["src/**/*.ts"]
}

Why use these compiler settings?

NodeNext keeps TypeScript's module behavior aligned with the project's ESM package type. Strict checks catch unsafe values before the server runs.

Source files live in src. Compiled JavaScript goes into dist.

Save tsconfig.json.
Create .gitignore inside reachinbox-email-scheduler.
Paste these ignored paths into .gitignore:

node_modules
dist
.env

Why ignore these paths?

Dependencies and compiled output can be recreated from the project files. The local .env file can contain credentials that must stay outside source control.

Save .gitignore.
Confirm the current files by running:

ls -a

What should I see now?

You should see package.json, tsconfig.json, .gitignore, node_modules, and src.

Environment variables keep machine-specific settings outside the TypeScript source. The example file documents every required key without containing live credentials.

Create .env.example inside reachinbox-email-scheduler.
Paste this configuration template into .env.example:

PORT=3000
DATABASE_URL=postgres://scheduler:scheduler@localhost:5432/scheduler
REDIS_URL=redis://localhost:6379
FRONTEND_ORIGIN=http://localhost:5173
WORKER_CONCURRENCY=5
MIN_SEND_DELAY_MS=2000
MAX_EMAILS_PER_HOUR_PER_SENDER=200
SEND_LEASE_MS=60000
ADMIN_KEY=replace-with-a-long-random-value
ETHEREAL_USER=
ETHEREAL_PASS=
SLACK_CLIENT_ID=
SLACK_CLIENT_SECRET=
SLACK_REDIRECT_URI=https://your-quick-tunnel.trycloudflare.com/api/slack/callback

What does this configuration cover?

The file defines local service addresses, worker controls, limiter values, and the protected queue-board key. It also reserves empty fields for Ethereal and Slack credentials used later.

The values use local ports for development. The PostgreSQL and Redis services remain reachable only through your machine.

Save .env.example.

Your local .env stays on this machine. The .gitignore entry prevents it from being committed.

Copy the template into the local environment file by running:

cp .env.example .env

Why keep two environment files?

The example file documents the expected keys for every developer. The local copy holds the values used by your machine.

Create config.ts inside the src folder.
Paste this configuration loader into src/config.ts:

import "dotenv/config";

function integer(name: string, fallback: number): number {
  const value = Number(process.env[name] ?? fallback);
  if (!Number.isInteger(value) || value <= 0) {
    throw new Error(`${name} must be a positive integer`);
  }
  return value;
}

export const config = {
  port: integer("PORT", 3000),
  databaseUrl:
    process.env.DATABASE_URL ??
    "postgres://scheduler:scheduler@localhost:5432/scheduler",
  redisUrl: process.env.REDIS_URL ?? "redis://localhost:6379",
  frontendOrigin: process.env.FRONTEND_ORIGIN ?? "http://localhost:5173",
  workerConcurrency: integer("WORKER_CONCURRENCY", 5),
  minSendDelayMs: integer("MIN_SEND_DELAY_MS", 2000),
  maxEmailsPerHourPerSender: integer(
    "MAX_EMAILS_PER_HOUR_PER_SENDER",
    200,
  ),
  sendLeaseMs: integer("SEND_LEASE_MS", 60000),
  adminKey: process.env.ADMIN_KEY ?? "replace-with-a-long-random-value",
  etherealUser: process.env.ETHEREAL_USER,
  etherealPass: process.env.ETHEREAL_PASS,
  slackClientId: process.env.SLACK_CLIENT_ID,
  slackClientSecret: process.env.SLACK_CLIENT_SECRET,
  slackRedirectUri: process.env.SLACK_REDIRECT_URI,
};

What does this code do?

The integer() helper rejects missing or invalid positive-number settings.
The exported config object gives the API and worker one typed source for runtime settings.
The optional credential fields remain empty until their integrations are configured.

Save src/config.ts.
Compile the current TypeScript source by running:

npm run build

What does the build confirm?

npm runs the build script from package.json. A successful exit confirms that TypeScript accepts src/config.ts under the project compiler rules.

Seeing a TypeScript error?

Check that src/config.ts matches the reference exactly. Confirm that tsconfig.json is saved in reachinbox-email-scheduler.

Help me fix the TypeScript build error in src/config.ts.

Start the persistent services

PostgreSQL stores business records. Redis stores BullMQ queue state plus rate-limit state.

Named Docker volumes keep that data when the containers stop. Redis also uses append-only file persistence to record write operations.

Create compose.yaml inside reachinbox-email-scheduler.
Paste this service definition into compose.yaml:

services:
  postgres:
    image: postgres:18.6
    environment:
      POSTGRES_USER: scheduler
      POSTGRES_PASSWORD: scheduler
      POSTGRES_DB: scheduler
    ports:
      - "127.0.0.1:5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:8.10.1
    command:
      - redis-server
      - --appendonly
      - "yes"
      - --appendfsync
      - everysec
      - --maxmemory-policy
      - noeviction
    ports:
      - "127.0.0.1:6379:6379"
    volumes:
      - redis_data:/data

volumes:
  postgres_data:
  redis_data:

How does persistence work?

The postgres_data volume stores PostgreSQL data. The redis_data volume stores Redis data.

Redis uses --appendonly with everysec persistence. The noeviction policy prevents queue keys from being discarded under memory pressure.

Save compose.yaml.
Start PostgreSQL and Redis in the background by running:

docker compose up -d

What does this command start?

Docker Compose creates the PostgreSQL and Redis containers from compose.yaml. Detached mode returns control of the terminal while both services continue running.

The first start may take a few minutes while Docker downloads the PostgreSQL and Redis images. A quiet download does not mean the command is stuck.

Could not start the services?

Confirm that Docker Engine is running. A permissions error means your Linux account cannot currently access the Docker daemon.

A port conflict means another process is already using PostgreSQL port 5432 or Redis port 6379.

Help me troubleshoot docker compose up for PostgreSQL and Redis.

Create README.md inside reachinbox-email-scheduler.
Copy the exact README.md contents from the full-code check below.
Save README.md.

Before you check, which two service names do you expect Docker Compose to report as running?

Check the service state and pinned Node.js version by running:

docker compose ps
node --version

What should I see?

You should see postgres and redis listed as running. The final line should report v24.20.0.

This proves that the application runtime is pinned correctly. It also proves that both persistent services are ready.

That is the foundation in place. Your database and queue storage now keep running outside the future API process.

✔️ Awesome, I've got everything!

Great. Double check that every project file is saved. Keep PostgreSQL and Redis running for the next step.

ⓧ I'd like to double check the full code

Compare each file with these exact versions. Use the filename shown above each reference.

{
  "name": "reachinbox-email-scheduler",
  "version": "1.0.0",
  "private": true,
  "type": "module",
  "scripts": {
    "build": "tsc -p tsconfig.json",
    "dev:api": "tsx watch src/server.ts",
    "dev:worker": "tsx watch src/worker.ts",
    "start:api": "node dist/server.js",
    "start:worker": "node dist/worker.js"
  },
  "dependencies": {
    "@bull-board/api": "9.8.0",
    "@bull-board/express": "9.8.0",
    "bullmq": "6.3.4",
    "dotenv": "17.4.2",
    "express": "5.2.1",
    "ioredis": "6.0.0",
    "nodemailer": "10.0.1",
    "pg": "8.23.0"
  },
  "devDependencies": {
    "@types/express": "5.0.6",
    "@types/node": "26.5.0",
    "@types/pg": "8.23.1",
    "tsx": "4.23.13",
    "typescript": "7.0.2"
  }
}

Package reference

This is the complete package definition for the workspace.

{
  "compilerOptions": {
    "target": "ES2023",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "rootDir": "src",
    "outDir": "dist",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "noUncheckedIndexedAccess": true
  },
  "include": ["src/**/*.ts"]
}

Compiler reference

This is the complete TypeScript compiler configuration.

node_modules
dist
.env

Ignored-file reference

This is the complete list of generated or local paths excluded from Git.

PORT=3000
DATABASE_URL=postgres://scheduler:scheduler@localhost:5432/scheduler
REDIS_URL=redis://localhost:6379
FRONTEND_ORIGIN=http://localhost:5173
WORKER_CONCURRENCY=5
MIN_SEND_DELAY_MS=2000
MAX_EMAILS_PER_HOUR_PER_SENDER=200
SEND_LEASE_MS=60000
ADMIN_KEY=replace-with-a-long-random-value
ETHEREAL_USER=
ETHEREAL_PASS=
SLACK_CLIENT_ID=
SLACK_CLIENT_SECRET=
SLACK_REDIRECT_URI=https://your-quick-tunnel.trycloudflare.com/api/slack/callback

Environment reference

This is the complete environment template. Your local .env is a private copy of this file.

services:
  postgres:
    image: postgres:18.6
    environment:
      POSTGRES_USER: scheduler
      POSTGRES_PASSWORD: scheduler
      POSTGRES_DB: scheduler
    ports:
      - "127.0.0.1:5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:8.10.1
    command:
      - redis-server
      - --appendonly
      - "yes"
      - --appendfsync
      - everysec
      - --maxmemory-policy
      - noeviction
    ports:
      - "127.0.0.1:6379:6379"
    volumes:
      - redis_data:/data

volumes:
  postgres_data:
  redis_data:

Service reference

This is the complete Docker Compose definition for PostgreSQL and Redis.

import "dotenv/config";

function integer(name: string, fallback: number): number {
  const value = Number(process.env[name] ?? fallback);
  if (!Number.isInteger(value) || value <= 0) {
    throw new Error(`${name} must be a positive integer`);
  }
  return value;
}

export const config = {
  port: integer("PORT", 3000),
  databaseUrl:
    process.env.DATABASE_URL ??
    "postgres://scheduler:scheduler@localhost:5432/scheduler",
  redisUrl: process.env.REDIS_URL ?? "redis://localhost:6379",
  frontendOrigin: process.env.FRONTEND_ORIGIN ?? "http://localhost:5173",
  workerConcurrency: integer("WORKER_CONCURRENCY", 5),
  minSendDelayMs: integer("MIN_SEND_DELAY_MS", 2000),
  maxEmailsPerHourPerSender: integer(
    "MAX_EMAILS_PER_HOUR_PER_SENDER",
    200,
  ),
  sendLeaseMs: integer("SEND_LEASE_MS", 60000),
  adminKey: process.env.ADMIN_KEY ?? "replace-with-a-long-random-value",
  etherealUser: process.env.ETHEREAL_USER,
  etherealPass: process.env.ETHEREAL_PASS,
  slackClientId: process.env.SLACK_CLIENT_ID,
  slackClientSecret: process.env.SLACK_CLIENT_SECRET,
  slackRedirectUri: process.env.SLACK_REDIRECT_URI,
};

Runtime configuration reference

This is the complete typed runtime configuration loader.

# ReachInbox Email Scheduler Backend

This backend-first milestone implements REST scheduling, PostgreSQL persistence, BullMQ delayed jobs, Redis AOF persistence, Ethereal SMTP delivery, configurable concurrency, distributed per-sender hourly limits, minimum sender spacing, Slack OAuth alerts, restart reconciliation, and Bull Board.

## Run

1. Copy `.env.example` to `.env` and replace the secret placeholders.
2. Run `npm install`.
3. Run `docker compose up -d`.
4. In one terminal run `npm run dev:api`.
5. In another terminal run `npm run dev:worker`.
6. Open `http://localhost:3000/admin/queues` and sign in as `admin` with `ADMIN_KEY`.

## Slack

1. Create a Slack app with the bot scope `chat:write`.
2. Run `cloudflared tunnel --url http://localhost:3000`.
3. Set `SLACK_REDIRECT_URI` to the generated HTTPS URL plus `/api/slack/callback` and add the same URL under Slack OAuth Redirect URLs.
4. Restart the API and open `/api/slack/connect` with the tenant header supplied by your frontend or API client.
5. Invite the app to the destination channel and send its channel ID to `PUT /api/slack/channel`.
6. Use `DELETE /api/slack/disconnect` to revoke the token and delete the installation.

## API

- `POST /api/senders`
- `POST /api/emails` with `Idempotency-Key`
- `GET /api/emails?status=sent`
- `GET /api/slack/connect`
- `GET /api/slack/callback`
- `GET /api/slack/status`
- `PUT /api/slack/channel`
- `DELETE /api/slack/disconnect`
- `GET /api/health`
- `GET /admin/queues`

Use `X-Tenant-Id: demo` during the assignment demo. Every API response is JSON except the Slack callback success page and Bull Board.

## Scheduling and restart behavior

PostgreSQL is the business source of truth. Redis stores BullMQ job state. The API calculates the delay from `scheduledAt`, uses `email-<email-id>` as the BullMQ job ID, and retains finalized jobs. At API startup, non-final database rows are re-added. Existing BullMQ IDs are ignored, so reconciliation does not create a second queue job.

Redis runs with AOF enabled and named Docker storage. PostgreSQL also uses a named volume. `docker compose down` keeps both volumes. `docker compose down -v` deletes both.

## Concurrency and rate limiting

`WORKER_CONCURRENCY` controls parallel BullMQ processing. A short Redis Lua script atomically reserves a sender's hourly counter and spacing key. The configured values are `MAX_EMAILS_PER_HOUR_PER_SENDER` and `MIN_SEND_DELAY_MS`. A denied job is moved back to BullMQ's delayed set with `moveToDelayed` and `DelayedError`, so it is not dropped or permanently failed.

The hourly limiter uses aligned fixed UTC hour windows. A Redis notification key ensures only one Slack alert is attempted per sender and hour. If Slack is not connected or has no channel configured, email jobs continue without crashing.

## Idempotency and SMTP trade-off

The scheduling API has a PostgreSQL unique constraint on `(tenant_id, idempotency_key)` and BullMQ uses a stable custom job ID. This prevents duplicate API retries from creating multiple records or queue jobs.

SMTP does not expose a provider idempotency key. A process can fail after the SMTP server accepts a message but before PostgreSQL records `sent`. Strict exactly-once delivery is therefore impossible with plain SMTP. This implementation chooses the no-automatic-duplicate side of that trade-off: stale or failed `sending` work becomes `uncertain` and requires manual review instead of being resent automatically.

## Deliberate scope boundary

This milestone does not include Elasticsearch, login, or the custom React or Next.js dashboard. Those belong in the next project because they require independent security, indexing consistency, and UI work. The REST endpoints and CORS contract are ready for that frontend.

## Cleanup

- Pause while retaining data: `docker compose down`
- Delete local data: `docker compose down -v`
- Revoke Slack first: call `DELETE /api/slack/disconnect`

Documentation reference

This is the complete project documentation for local operation, Slack setup, API behavior, recovery, rate limiting, SMTP safety, scope, and cleanup.

Confirm the root project files exist by running:

ls -a

What should the project contain?

You should see package.json, tsconfig.json, .gitignore, .env.example, .env, compose.yaml, README.md, and src.

Your persistent development stack is ready. Next up, you will turn it into a frontend-ready scheduling API backed by PostgreSQL.

Create the Scheduling API

Your persistent PostgreSQL and Redis services are running. The scheduler now needs a request path that stores email work outside the Node.js process.

This step builds that path with an Express API. You will make scheduling visible before discovering why a stored email still needs a delivery worker.

In this step, get ready to:

Create the PostgreSQL schema for tenants, senders, emails, and Slack installations.
Expose tenant-aware sender creation and email-listing endpoints with CORS headers.
Schedule email records idempotently and inspect their stored status.

Build the durable database schema

The API needs a durable source of truth for every sender and scheduled email. A connection pool lets requests share PostgreSQL connections while the schema keeps tenant data separated.

Use your editor's file explorer to create src/db.ts inside the src folder.
Add the connection pool and the first two tables by pasting this code:

import { Pool } from "pg";
import { config } from "./config.js";

export const pool = new Pool({
  connectionString: config.databaseUrl,
  max: 10,
});

export async function initializeDatabase(): Promise<void> {
  await pool.query(`
    CREATE TABLE IF NOT EXISTS tenants (
      id TEXT PRIMARY KEY,
      name TEXT NOT NULL
    );

    CREATE TABLE IF NOT EXISTS senders (
      id TEXT PRIMARY KEY,
      tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
      name TEXT NOT NULL,
      email TEXT NOT NULL,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      UNIQUE (tenant_id, email)
    );
  `);
}

What does this code do?

The Pool uses the existing DATABASE_URL value to connect to PostgreSQL.
The tenants table gives each customer's data a stable owner.
The senders table stores the identities that can schedule email.
The unique tenant and email pair prevents one tenant from registering the same sender address twice.

Insert the email table inside the SQL template before its closing backtick:

    CREATE TABLE IF NOT EXISTS emails (
      id TEXT PRIMARY KEY,
      tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
      sender_id TEXT NOT NULL REFERENCES senders(id),
      idempotency_key TEXT NOT NULL,
      to_email TEXT NOT NULL,
      subject TEXT NOT NULL,
      text_body TEXT NOT NULL,
      scheduled_at TIMESTAMPTZ NOT NULL,
      status TEXT NOT NULL DEFAULT 'scheduled'
        CHECK (status IN (
          'scheduled', 'queued', 'processing', 'sending',
          'rate_limited', 'sent', 'failed', 'uncertain'
        )),
      processing_started_at TIMESTAMPTZ,
      rate_limited_until TIMESTAMPTZ,
      attempts INTEGER NOT NULL DEFAULT 0,
      provider_message_id TEXT,
      preview_url TEXT,
      sent_at TIMESTAMPTZ,
      last_error TEXT,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      UNIQUE (tenant_id, idempotency_key)
    );

Why store delivery state?

Each email row records its business status independently from the future queue job. The fields also preserve attempts, delivery evidence, rate-limit timing, and ambiguous outcomes.

The unique tenant and idempotency key pair turns a repeated scheduling request into the same logical email. That constraint is the database foundation for duplicate protection.

Save src/db.ts.
Check the TypeScript structure by running this command:

npm run build

What does this check prove?

The build script asks TypeScript to compile the project using tsconfig.json. A successful exit confirms that the pool and schema function are valid TypeScript.

Seeing a build error?

Check that src/db.ts imports config from ./config.js. NodeNext imports use the emitted JavaScript extension.

Check that the SQL remains between one opening backtick and one closing backtick.

Help me fix the TypeScript error in src/db.ts.

Insert the remaining schema objects before the SQL template's closing backtick:

    CREATE INDEX IF NOT EXISTS emails_tenant_status_idx
      ON emails (tenant_id, status, scheduled_at);

    CREATE TABLE IF NOT EXISTS slack_installations (
      tenant_id TEXT PRIMARY KEY REFERENCES tenants(id) ON DELETE CASCADE,
      team_id TEXT NOT NULL,
      team_name TEXT NOT NULL,
      access_token TEXT NOT NULL,
      bot_user_id TEXT,
      channel_id TEXT,
      connected_at TIMESTAMPTZ NOT NULL DEFAULT now()
    );

    CREATE TABLE IF NOT EXISTS slack_oauth_states (
      state TEXT PRIMARY KEY,
      tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
      expires_at TIMESTAMPTZ NOT NULL
    );

    INSERT INTO tenants (id, name)
    VALUES ('demo', 'Demo Tenant')
    ON CONFLICT (id) DO NOTHING;

What completes the schema?

The emails_tenant_status_idx index supports tenant-scoped status queries ordered by schedule time.
The Slack tables reserve durable storage for a tenant installation and short-lived OAuth state.
The final insert creates the demo tenant once.
The conflict clause makes repeated API starts safe.

Save src/db.ts.
Confirm the completed schema compiles by running this command:

npm run build

What should you see?

The command should finish without a TypeScript error. This confirms that the completed database module can be imported by the API.

Is the schema failing to compile?

Check the comma after each column except the final column in a table definition. Also confirm that every opening parenthesis has a matching closing parenthesis.

Help me compare my database schema with the required structure.

✔️ Awesome, I've got everything!

Your database module compiles. Keep src/db.ts saved before you build the API.

ⓧ I'd like to double check the full code

Compare your src/db.ts file with this complete version.

import { Pool } from "pg";
import { config } from "./config.js";

export const pool = new Pool({
  connectionString: config.databaseUrl,
  max: 10,
});

export async function initializeDatabase(): Promise<void> {
  await pool.query(`
    CREATE TABLE IF NOT EXISTS tenants (
      id TEXT PRIMARY KEY,
      name TEXT NOT NULL
    );

    CREATE TABLE IF NOT EXISTS senders (
      id TEXT PRIMARY KEY,
      tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
      name TEXT NOT NULL,
      email TEXT NOT NULL,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      UNIQUE (tenant_id, email)
    );

    CREATE TABLE IF NOT EXISTS emails (
      id TEXT PRIMARY KEY,
      tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
      sender_id TEXT NOT NULL REFERENCES senders(id),
      idempotency_key TEXT NOT NULL,
      to_email TEXT NOT NULL,
      subject TEXT NOT NULL,
      text_body TEXT NOT NULL,
      scheduled_at TIMESTAMPTZ NOT NULL,
      status TEXT NOT NULL DEFAULT 'scheduled'
        CHECK (status IN (
          'scheduled', 'queued', 'processing', 'sending',
          'rate_limited', 'sent', 'failed', 'uncertain'
        )),
      processing_started_at TIMESTAMPTZ,
      rate_limited_until TIMESTAMPTZ,
      attempts INTEGER NOT NULL DEFAULT 0,
      provider_message_id TEXT,
      preview_url TEXT,
      sent_at TIMESTAMPTZ,
      last_error TEXT,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      UNIQUE (tenant_id, idempotency_key)
    );

    CREATE INDEX IF NOT EXISTS emails_tenant_status_idx
      ON emails (tenant_id, status, scheduled_at);

    CREATE TABLE IF NOT EXISTS slack_installations (
      tenant_id TEXT PRIMARY KEY REFERENCES tenants(id) ON DELETE CASCADE,
      team_id TEXT NOT NULL,
      team_name TEXT NOT NULL,
      access_token TEXT NOT NULL,
      bot_user_id TEXT,
      channel_id TEXT,
      connected_at TIMESTAMPTZ NOT NULL DEFAULT now()
    );

    CREATE TABLE IF NOT EXISTS slack_oauth_states (
      state TEXT PRIMARY KEY,
      tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
      expires_at TIMESTAMPTZ NOT NULL
    );

    INSERT INTO tenants (id, name)
    VALUES ('demo', 'Demo Tenant')
    ON CONFLICT (id) DO NOTHING;
  `);
}

How to compare this file

Check the table names, column names, status values, and closing delimiters. Your saved file should match this reference exactly.

Create the tenant-aware Express service

The service initializes the schema before accepting requests. Its CORS headers also prepare the API for the later frontend without adding that frontend to this project.

Use your editor's file explorer to create src/server.ts inside the src folder.
Add the imports and CORS middleware by pasting this code:

import { randomUUID } from "node:crypto";
import express, {
  type NextFunction,
  type Request,
  type Response,
} from "express";
import { config } from "./config.js";
import { initializeDatabase, pool } from "./db.js";

await initializeDatabase();

const app = express();
app.use(express.json({ limit: "100kb" }));
app.use((req, res, next) => {
  res.setHeader("Access-Control-Allow-Origin", config.frontendOrigin);
  res.setHeader("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS");
  res.setHeader(
    "Access-Control-Allow-Headers",
    "Content-Type,Idempotency-Key,X-Tenant-Id,Authorization",
  );
  if (req.method === "OPTIONS") {
    res.sendStatus(204);
    return;
  }
  next();
});

What does this foundation do?

The service waits for initializeDatabase() before it creates the HTTP listener.
The JSON limit bounds the size of incoming request bodies.
The CORS middleware allows the configured frontend origin to use the API methods and headers.
Preflight requests receive a 204 response.

Add the tenant helpers, health route, error middleware, and listener below the CORS middleware:

function tenantId(req: Request): string {
  const value = req.header("X-Tenant-Id") ?? "demo";
  if (!/^[a-zA-Z0-9_-]{1,64}$/.test(value)) {
    throw new Error("Invalid X-Tenant-Id header");
  }
  return value;
}

function requiredString(value: unknown, field: string): string {
  if (typeof value !== "string" || value.trim() === "") {
    throw new Error(`${field} is required`);
  }
  return value.trim();
}

app.get("/api/health", async (_req, res) => {
  await pool.query("SELECT 1");
  res.json({ ok: true });
});

app.use(
  (error: unknown, _req: Request, res: Response, _next: NextFunction) => {
    const message = error instanceof Error ? error.message : "Unknown error";
    res.status(400).json({ error: message });
  },
);

const server = app.listen(config.port, () => {
  console.log(`API listening on http://localhost:${config.port}`);
});

How does request handling work?

The tenantId() helper defaults requests to the demo tenant.
The header pattern blocks unexpected tenant identifier characters.
The requiredString() helper rejects missing or blank text fields.
The error middleware converts thrown errors into JSON responses.

Save src/server.ts.
Start the API from the reachinbox-email-scheduler terminal by running this command:

npm run dev:api

What should you see?

The development process initializes the PostgreSQL schema before listening on port 3000. You should see the API listening address in the terminal.

That is the first live slice of your scheduler. The API can now prove that its database connection works.

API not starting?

Confirm the PostgreSQL container is still running. Also check that DATABASE_URL in .env points to the local scheduler database.

If port 3000 is already occupied, stop the older local process before retrying.

Help me diagnose why the Express API cannot start.

Open a second terminal in the existing reachinbox-email-scheduler folder.
Check the database-backed health route by running this command:

curl http://localhost:3000/api/health

What does the response prove?

You should receive a JSON response with ok set to true. The response proves that Express can query PostgreSQL.

Health request failing?

Keep the first terminal running npm run dev:api while you send the request from the second terminal.

If the response mentions a database connection, confirm the postgres container is running.

Help me fix the API health request.

Insert the sender creation route after the health route and before the error middleware:

app.post("/api/senders", async (req, res) => {
  const tenant = tenantId(req);
  const id = requiredString(req.body.id, "id");
  const name = requiredString(req.body.name, "name");
  const email = requiredString(req.body.email, "email");
  if (!email.includes("@")) throw new Error("email must contain @");

  const result = await pool.query(
    `INSERT INTO senders (id, tenant_id, name, email)
     VALUES ($1, $2, $3, $4)
     ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, email = EXCLUDED.email
     RETURNING *`,
    [id, tenant, name, email],
  );
  res.status(201).json(result.rows[0]);
});

How is the sender stored?

The route validates each required field before using a parameterized query. The tenant comes from X-Tenant-Id or defaults to demo.

A repeated sender ID updates the name and email. The API returns the persisted row with a 201 status.

Save src/server.ts.
Create the demo sender from the second terminal by running this request:

curl -X POST http://localhost:3000/api/senders \
  -H "Content-Type: application/json" \
  -d '{"id":"sender-demo","name":"Demo Sender","email":"sender@example.com"}'

What should the API return?

You should receive the stored sender as JSON. Its tenant_id value should be demo because the request did not supply a tenant header.

Sender request rejected?

Check that the request body uses double quotes around every JSON key and value. Confirm that the email value contains an @ character.

Help me fix the sender creation request.

Insert the email-listing route below the sender route and before the error middleware:

app.get("/api/emails", async (req, res) => {
  const tenant = tenantId(req);
  const status = typeof req.query.status === "string" ? req.query.status : null;
  const values: unknown[] = [tenant];
  let sql = `SELECT * FROM emails WHERE tenant_id = $1`;
  if (status) {
    sql += ` AND status = $2`;
    values.push(status);
  }
  sql += ` ORDER BY scheduled_at DESC LIMIT 200`;
  const result = await pool.query(sql, values);
  res.json(result.rows);
});

How does filtering work?

Every listing stays within the resolved tenant. Supplying a status query value adds a parameterized filter.

The response includes up to 200 records with the latest scheduled time first.

Save src/server.ts.

The watch process reloads the service after the save. Your sender persists because PostgreSQL holds it outside the Node.js process.

API stopped after the edit?

Read the first terminal for the exact TypeScript line number. Check the route's braces and the backticks around each SQL fragment.

Help me fix the GET /api/emails route.

Add idempotent email scheduling

Clients can retry a request when a network response is lost. An idempotency key gives every logical scheduling request a stable identity inside its tenant.

Insert the first half of the scheduling route below the sender route and before the listing route:

app.post("/api/emails", async (req, res) => {
  const tenant = tenantId(req);
  const idempotencyKey = req.header("Idempotency-Key");
  if (!idempotencyKey) throw new Error("Idempotency-Key header is required");

  const senderId = requiredString(req.body.senderId, "senderId");
  const toEmail = requiredString(req.body.toEmail, "toEmail");
  const subject = requiredString(req.body.subject, "subject");
  const textBody = requiredString(req.body.textBody, "textBody");
  const scheduledAt = new Date(requiredString(req.body.scheduledAt, "scheduledAt"));
  if (Number.isNaN(scheduledAt.getTime())) {
    throw new Error("scheduledAt must be an ISO date-time");
  }

  const sender = await pool.query(
    `SELECT id FROM senders WHERE id = $1 AND tenant_id = $2`,
    [senderId, tenant],
  );
  if (sender.rowCount !== 1) {
    res.status(404).json({ error: "Sender not found" });
    return;
  }

What does the validation protect?

The route requires an idempotency key plus the sender, recipient, subject, body, and schedule time. It rejects an invalid date before touching the email table.

The sender lookup also verifies tenant ownership. A tenant cannot schedule mail through another tenant's sender.

Complete the scheduling route by adding this code directly below the sender check:

  const id = randomUUID();
  const result = await pool.query<{
    id: string;
    scheduled_at: Date;
    status: string;
  }>(
    `INSERT INTO emails
       (id, tenant_id, sender_id, idempotency_key,
        to_email, subject, text_body, scheduled_at)
     VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
     ON CONFLICT (tenant_id, idempotency_key)
     DO UPDATE SET idempotency_key = EXCLUDED.idempotency_key
     RETURNING id, scheduled_at, status`,
    [
      id,
      tenant,
      senderId,
      idempotencyKey,
      toEmail,
      subject,
      textBody,
      scheduledAt,
    ],
  );
  const email = result.rows[0]!;

  const saved = await pool.query(`SELECT * FROM emails WHERE id = $1`, [email.id]);
  res.status(202).json(saved.rows[0]);
});

How does idempotency work?

A new request generates a UUID and inserts one email row. PostgreSQL resolves a repeated tenant and idempotency key through ON CONFLICT.

The conflict update keeps the existing row in place. The final query returns that persisted record to the client.

Save src/server.ts.
Schedule a future email from the second terminal by running this request:

curl -X POST http://localhost:3000/api/emails \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: scheduling-demo-1" \
  -d '{"senderId":"sender-demo","toEmail":"recipient@example.com","subject":"Restart-safe demo","textBody":"This email is stored durably.","scheduledAt":"2026-12-01T10:00:00.000Z"}'

What should the scheduling response contain?

You should receive a JSON email record with a generated ID. Its status should be scheduled.

The 202 response confirms that the API accepted durable work. Delivery is a separate responsibility.

Record the returned email ID here: your email ID.
Repeat the same scheduling request with the same idempotency key:

curl -X POST http://localhost:3000/api/emails \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: scheduling-demo-1" \
  -d '{"senderId":"sender-demo","toEmail":"recipient@example.com","subject":"Restart-safe demo","textBody":"This email is stored durably.","scheduledAt":"2026-12-01T10:00:00.000Z"}'

What should stay the same?

The second response should return the same ID as your email ID. PostgreSQL reused the existing record instead of creating a duplicate.

Received a different email ID?

Confirm both requests use exactly scheduling-demo-1 in the Idempotency-Key header. Also confirm both requests resolve to the default demo tenant.

Help me diagnose duplicate scheduling records.

Before you check the stored record, do you think a PostgreSQL row can send an email by itself?

List scheduled emails by running this command:

curl "http://localhost:3000/api/emails?status=scheduled"

What did the database-only API reveal?

You should see the email inside a JSON array with the same ID. Its status remains scheduled.

This is the intended shortfall. The request is durable, but no independent process is watching for its due time or sending it.

Scheduled list is empty?

Check that the scheduling response used the scheduled status. Confirm that the listing URL includes ?status=scheduled.

If the sender request failed earlier, recreate sender-demo before scheduling the email.

Help me find why GET /api/emails returns no scheduled records.

✔️ Awesome, I've got everything!

Your API creates senders, stores idempotent email requests, and lists their status. Keep the API process running for the next step.

ⓧ I'd like to double check the full code

Compare your src/server.ts file with this complete version for the current step.

import { randomUUID } from "node:crypto";
import express, {
  type NextFunction,
  type Request,
  type Response,
} from "express";
import { config } from "./config.js";
import { initializeDatabase, pool } from "./db.js";

await initializeDatabase();

const app = express();
app.use(express.json({ limit: "100kb" }));
app.use((req, res, next) => {
  res.setHeader("Access-Control-Allow-Origin", config.frontendOrigin);
  res.setHeader("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS");
  res.setHeader(
    "Access-Control-Allow-Headers",
    "Content-Type,Idempotency-Key,X-Tenant-Id,Authorization",
  );
  if (req.method === "OPTIONS") {
    res.sendStatus(204);
    return;
  }
  next();
});

function tenantId(req: Request): string {
  const value = req.header("X-Tenant-Id") ?? "demo";
  if (!/^[a-zA-Z0-9_-]{1,64}$/.test(value)) {
    throw new Error("Invalid X-Tenant-Id header");
  }
  return value;
}

function requiredString(value: unknown, field: string): string {
  if (typeof value !== "string" || value.trim() === "") {
    throw new Error(`${field} is required`);
  }
  return value.trim();
}

app.get("/api/health", async (_req, res) => {
  await pool.query("SELECT 1");
  res.json({ ok: true });
});

app.post("/api/senders", async (req, res) => {
  const tenant = tenantId(req);
  const id = requiredString(req.body.id, "id");
  const name = requiredString(req.body.name, "name");
  const email = requiredString(req.body.email, "email");
  if (!email.includes("@")) throw new Error("email must contain @");

  const result = await pool.query(
    `INSERT INTO senders (id, tenant_id, name, email)
     VALUES ($1, $2, $3, $4)
     ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, email = EXCLUDED.email
     RETURNING *`,
    [id, tenant, name, email],
  );
  res.status(201).json(result.rows[0]);
});

app.post("/api/emails", async (req, res) => {
  const tenant = tenantId(req);
  const idempotencyKey = req.header("Idempotency-Key");
  if (!idempotencyKey) throw new Error("Idempotency-Key header is required");

  const senderId = requiredString(req.body.senderId, "senderId");
  const toEmail = requiredString(req.body.toEmail, "toEmail");
  const subject = requiredString(req.body.subject, "subject");
  const textBody = requiredString(req.body.textBody, "textBody");
  const scheduledAt = new Date(requiredString(req.body.scheduledAt, "scheduledAt"));
  if (Number.isNaN(scheduledAt.getTime())) {
    throw new Error("scheduledAt must be an ISO date-time");
  }

  const sender = await pool.query(
    `SELECT id FROM senders WHERE id = $1 AND tenant_id = $2`,
    [senderId, tenant],
  );
  if (sender.rowCount !== 1) {
    res.status(404).json({ error: "Sender not found" });
    return;
  }

  const id = randomUUID();
  const result = await pool.query<{
    id: string;
    scheduled_at: Date;
    status: string;
  }>(
    `INSERT INTO emails
       (id, tenant_id, sender_id, idempotency_key,
        to_email, subject, text_body, scheduled_at)
     VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
     ON CONFLICT (tenant_id, idempotency_key)
     DO UPDATE SET idempotency_key = EXCLUDED.idempotency_key
     RETURNING id, scheduled_at, status`,
    [
      id,
      tenant,
      senderId,
      idempotencyKey,
      toEmail,
      subject,
      textBody,
      scheduledAt,
    ],
  );
  const email = result.rows[0]!;

  const saved = await pool.query(`SELECT * FROM emails WHERE id = $1`, [email.id]);
  res.status(202).json(saved.rows[0]);
});

app.get("/api/emails", async (req, res) => {
  const tenant = tenantId(req);
  const status = typeof req.query.status === "string" ? req.query.status : null;
  const values: unknown[] = [tenant];
  let sql = `SELECT * FROM emails WHERE tenant_id = $1`;
  if (status) {
    sql += ` AND status = $2`;
    values.push(status);
  }
  sql += ` ORDER BY scheduled_at DESC LIMIT 200`;
  const result = await pool.query(sql, values);
  res.json(result.rows);
});

app.use(
  (error: unknown, _req: Request, res: Response, _next: NextFunction) => {
    const message = error instanceof Error ? error.message : "Unknown error";
    res.status(400).json({ error: message });
  },
);

const server = app.listen(config.port, () => {
  console.log(`API listening on http://localhost:${config.port}`);
});

How to compare this file

Check the import paths, middleware order, route order, SQL parameters, and closing braces. Your saved file should match this reference exactly.

Your frontend-ready API now preserves senders and scheduled emails across process restarts. Next, you will add the queue and worker that turn those durable records into Ethereal deliveries.

Queue and Deliver Emails

Your PostgreSQL record proves that the API can remember a scheduled email. The record stays scheduled because no delivery process is watching for its due time.

This step connects the API to a delayed BullMQ queue backed by Redis. A separate worker sends due messages through Ethereal Email without delivering anything to a real inbox.

In this step, get ready to:

Add scheduled emails to a persistent delayed queue.
Deliver due emails through a reusable Ethereal transport.
Inspect delayed and completed jobs through a protected queue dashboard.

Prove the database-only limitation

Before you check the current record, do you expect its status to change without a worker?

Return to the browser tab for http://localhost:3000/api/emails?status=scheduled.
Refresh the page after the scheduled time has passed.

You will still see the email with a scheduled status. This shortfall is intentional because a database record cannot execute delivery work.

Why use a separate queue?

The API accepts requests while BullMQ tracks when each delivery becomes eligible. This separation lets the API respond without staying attached to a delayed task.

Redis stores the queue state outside the Node.js process. Delayed jobs can remain available while the API or worker restarts.

Add the persistent email queue

The queue producer calculates the delay from scheduledAt. Its stable job ID ties one BullMQ job to one PostgreSQL email record.

Create src/queue.ts in the existing src folder.
Add the Redis connection and queue producer with this first chunk:

import { Queue } from "bullmq";
import { Redis } from "ioredis";
import { config } from "./config.js";
import { pool } from "./db.js";

export const producerRedis = new Redis(config.redisUrl, {
  maxRetriesPerRequest: 1,
});

export const emailQueue = new Queue("email-send", {
  connection: producerRedis,
});

export async function enqueueEmail(
  emailId: string,
  scheduledAt: Date,
): Promise<void> {
  const delay = Math.max(0, scheduledAt.getTime() - Date.now());
  await emailQueue.add(
    "send-email",
    { emailId },
    {
      delay,
      jobId: `email-${emailId}`,
    },
  );
}

What does this queue code do?

The producer connection gives BullMQ access to the persistent Redis service.
The email-send queue holds every delayed delivery job.
The calculated delay keeps future jobs dormant until their scheduled time.
The stable email-<email-id> job ID prevents a retained email job from being added twice.

Add startup reconciliation below enqueueEmail() with this code:

export async function recoverNonFinalEmails(): Promise<number> {
  const result = await pool.query<{
    id: string;
    scheduled_at: Date;
  }>(`
    SELECT id, scheduled_at
    FROM emails
    WHERE status IN (
      'scheduled', 'queued', 'processing', 'sending', 'rate_limited'
    )
    ORDER BY scheduled_at ASC
  `);

  for (const email of result.rows) {
    await enqueueEmail(email.id, new Date(email.scheduled_at));
    await pool.query(
      `UPDATE emails
       SET status = CASE
         WHEN status = 'scheduled' THEN 'queued'
         ELSE status
       END,
       updated_at = now()
       WHERE id = $1`,
      [email.id],
    );
  }

  return result.rowCount ?? 0;
}

How does reconciliation help?

The API reads every non-final email when it starts. Each record is offered to the queue using the same stable job ID.

A retained BullMQ job keeps its existing identity. A missing job is restored from PostgreSQL.

Save src/queue.ts.
Confirm that src/queue.ts appears beside the existing TypeScript files in the editor sidebar.

Seeing queue import errors?

Confirm that the import paths end in .js because this project uses NodeNext module resolution.

Check that your earlier dependency installation completed successfully.

Ask for help with the exact compiler output if the imports remain unresolved. Help me diagnose the BullMQ or ioredis import error in src/queue.ts.

✔️ Awesome, I've got everything!

Great. Your queue producer now calculates delays and restores non-final database records.

ⓧ I'd like to double check the full code

Compare your complete src/queue.ts file with this reference.

import { Queue } from "bullmq";
import { Redis } from "ioredis";
import { config } from "./config.js";
import { pool } from "./db.js";

export const producerRedis = new Redis(config.redisUrl, {
  maxRetriesPerRequest: 1,
});

export const emailQueue = new Queue("email-send", {
  connection: producerRedis,
});

export async function enqueueEmail(
  emailId: string,
  scheduledAt: Date,
): Promise<void> {
  const delay = Math.max(0, scheduledAt.getTime() - Date.now());
  await emailQueue.add(
    "send-email",
    { emailId },
    {
      delay,
      jobId: `email-${emailId}`,
    },
  );
}

export async function recoverNonFinalEmails(): Promise<number> {
  const result = await pool.query<{
    id: string;
    scheduled_at: Date;
  }>(`
    SELECT id, scheduled_at
    FROM emails
    WHERE status IN (
      'scheduled', 'queued', 'processing', 'sending', 'rate_limited'
    )
    ORDER BY scheduled_at ASC
  `);

  for (const email of result.rows) {
    await enqueueEmail(email.id, new Date(email.scheduled_at));
    await pool.query(
      `UPDATE emails
       SET status = CASE
         WHEN status = 'scheduled' THEN 'queued'
         ELSE status
       END,
       updated_at = now()
       WHERE id = $1`,
      [email.id],
    );
  }

  return result.rowCount ?? 0;
}

Connect the API to the queue dashboard

The API now needs to enqueue each saved email and expose the queue through Bull Board. HTTP Basic authentication keeps the dashboard behind the configured admin credential.

Update src/server.ts so it imports the queue and Bull Board adapters.
Call recoverNonFinalEmails() immediately after database initialization.
Update POST /api/emails so it enqueues the saved email and changes a scheduled record to queued.
Mount the protected dashboard at /admin/queues.
Add queue cleanup to the existing shutdown flow.
Compare your updated file with the complete reference below.

✔️ Awesome, I've got everything!

Save src/server.ts once your queue imports, scheduling route, dashboard, and shutdown flow match the target.

ⓧ I'd like to double check the full code

At this point, your complete src/server.ts file should look like this.

import { randomUUID } from "node:crypto";
import { createBullBoard } from "@bull-board/api";
import { BullMQAdapter } from "@bull-board/api/bullMQAdapter";
import { ExpressAdapter } from "@bull-board/express";
import express, {
  type NextFunction,
  type Request,
  type Response,
} from "express";
import { config } from "./config.js";
import { initializeDatabase, pool } from "./db.js";
import {
  emailQueue,
  enqueueEmail,
  producerRedis,
  recoverNonFinalEmails,
} from "./queue.js";

await initializeDatabase();
const recovered = await recoverNonFinalEmails();

const app = express();
app.use(express.json({ limit: "100kb" }));
app.use((req, res, next) => {
  res.setHeader("Access-Control-Allow-Origin", config.frontendOrigin);
  res.setHeader("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS");
  res.setHeader(
    "Access-Control-Allow-Headers",
    "Content-Type,Idempotency-Key,X-Tenant-Id,Authorization",
  );
  if (req.method === "OPTIONS") {
    res.sendStatus(204);
    return;
  }
  next();
});

function tenantId(req: Request): string {
  const value = req.header("X-Tenant-Id") ?? "demo";
  if (!/^[a-zA-Z0-9_-]{1,64}$/.test(value)) {
    throw new Error("Invalid X-Tenant-Id header");
  }
  return value;
}

function requiredString(value: unknown, field: string): string {
  if (typeof value !== "string" || value.trim() === "") {
    throw new Error(`${field} is required`);
  }
  return value.trim();
}

app.get("/api/health", async (_req, res) => {
  await pool.query("SELECT 1");
  await producerRedis.ping();
  res.json({ ok: true, recovered });
});

app.post("/api/senders", async (req, res) => {
  const tenant = tenantId(req);
  const id = requiredString(req.body.id, "id");
  const name = requiredString(req.body.name, "name");
  const email = requiredString(req.body.email, "email");
  if (!email.includes("@")) throw new Error("email must contain @");

  const result = await pool.query(
    `INSERT INTO senders (id, tenant_id, name, email)
     VALUES ($1, $2, $3, $4)
     ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, email = EXCLUDED.email
     RETURNING *`,
    [id, tenant, name, email],
  );
  res.status(201).json(result.rows[0]);
});

app.post("/api/emails", async (req, res) => {
  const tenant = tenantId(req);
  const idempotencyKey = req.header("Idempotency-Key");
  if (!idempotencyKey) throw new Error("Idempotency-Key header is required");

  const senderId = requiredString(req.body.senderId, "senderId");
  const toEmail = requiredString(req.body.toEmail, "toEmail");
  const subject = requiredString(req.body.subject, "subject");
  const textBody = requiredString(req.body.textBody, "textBody");
  const scheduledAt = new Date(requiredString(req.body.scheduledAt, "scheduledAt"));
  if (Number.isNaN(scheduledAt.getTime())) {
    throw new Error("scheduledAt must be an ISO date-time");
  }

  const sender = await pool.query(
    `SELECT id FROM senders WHERE id = $1 AND tenant_id = $2`,
    [senderId, tenant],
  );
  if (sender.rowCount !== 1) {
    res.status(404).json({ error: "Sender not found" });
    return;
  }

  const id = randomUUID();
  const result = await pool.query<{
    id: string;
    scheduled_at: Date;
    status: string;
  }>(
    `INSERT INTO emails
       (id, tenant_id, sender_id, idempotency_key,
        to_email, subject, text_body, scheduled_at)
     VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
     ON CONFLICT (tenant_id, idempotency_key)
     DO UPDATE SET idempotency_key = EXCLUDED.idempotency_key
     RETURNING id, scheduled_at, status`,
    [
      id,
      tenant,
      senderId,
      idempotencyKey,
      toEmail,
      subject,
      textBody,
      scheduledAt,
    ],
  );
  const email = result.rows[0]!;

  try {
    await enqueueEmail(email.id, new Date(email.scheduled_at));
    await pool.query(
      `UPDATE emails
       SET status = CASE WHEN status = 'scheduled' THEN 'queued' ELSE status END,
           updated_at = now()
       WHERE id = $1`,
      [email.id],
    );
  } catch (error) {
    res.status(503).json({
      error: "Email was saved but Redis enqueue failed; startup recovery will retry",
      emailId: email.id,
    });
    return;
  }

  const saved = await pool.query(`SELECT * FROM emails WHERE id = $1`, [email.id]);
  res.status(202).json(saved.rows[0]);
});

app.get("/api/emails", async (req, res) => {
  const tenant = tenantId(req);
  const status = typeof req.query.status === "string" ? req.query.status : null;
  const values: unknown[] = [tenant];
  let sql = `SELECT * FROM emails WHERE tenant_id = $1`;
  if (status) {
    sql += ` AND status = $2`;
    values.push(status);
  }
  sql += ` ORDER BY scheduled_at DESC LIMIT 200`;
  const result = await pool.query(sql, values);
  res.json(result.rows);
});

const boardAdapter = new ExpressAdapter();
boardAdapter.setBasePath("/admin/queues");
createBullBoard({
  queues: [new BullMQAdapter(emailQueue)],
  serverAdapter: boardAdapter,
});

function protectBoard(req: Request, res: Response, next: NextFunction): void {
  const expected = `Basic ${Buffer.from(`admin:${config.adminKey}`).toString("base64")}`;
  if (req.header("Authorization") !== expected) {
    res.setHeader("WWW-Authenticate", 'Basic realm="queues"');
    res.status(401).send("Authentication required");
    return;
  }
  next();
}

app.use("/admin/queues", protectBoard, boardAdapter.getRouter());

app.use(
  (error: unknown, _req: Request, res: Response, _next: NextFunction) => {
    const message = error instanceof Error ? error.message : "Unknown error";
    res.status(400).json({ error: message });
  },
);

const server = app.listen(config.port, () => {
  console.log(`API listening on http://localhost:${config.port}`);
  console.log(`Recovered ${recovered} non-final email records`);
});

async function shutdown(): Promise<void> {
  server.close();
  await emailQueue.close();
  await producerRedis.quit();
  await pool.end();
}

process.on("SIGINT", () => void shutdown().then(() => process.exit(0)));
process.on("SIGTERM", () => void shutdown().then(() => process.exit(0)));

What changed in the API?

Startup reconciliation restores non-final records to BullMQ.
The scheduling route returns queued after the job reaches Redis.
The health route checks both PostgreSQL and Redis.
The protected dashboard exposes delayed and completed queue state.

Save src/server.ts.
Watch the existing API terminal for its automatic restart.

Before you check the API, do you expect startup reconciliation to find the scheduled record from the earlier test?

Refresh http://localhost:3000/api/health in your browser.

You will see ok set to true plus a numeric recovered value. The earlier non-final record is now eligible for queue processing.

Repeat the email scheduling request from the previous step with a new idempotency key.
Set scheduledAt to an ISO date-time about two minutes in the future.
Inspect the JSON response for the new email record.

You will see the new record with a queued status. The database record now has a matching delayed job.

API not restarting cleanly?

Confirm that the PostgreSQL and Redis services from the earlier step are still running.

Compare the queue import names with src/queue.ts if the API reports a missing export.

Use the first stack trace line to locate a mismatched brace or import. Help me debug the API restart after adding BullMQ and Bull Board.

Create the Ethereal mailer and worker

A Nodemailer transporter owns the SMTP connection details. Reusing one transporter prevents every job from creating a separate test account.

The worker claims one database record before sending it. It stores the provider result only after Ethereal accepts the message.

Create src/mailer.ts in the existing src folder.
Add the reusable Ethereal transport with this first chunk:

import nodemailer, { type Transporter } from "nodemailer";
import { config } from "./config.js";

let transporterPromise: Promise<Transporter> | undefined;

async function createEtherealTransport(): Promise<Transporter> {
  if (config.etherealUser && config.etherealPass) {
    return nodemailer.createTransport({
      service: "Ethereal",
      auth: {
        user: config.etherealUser,
        pass: config.etherealPass,
      },
    });
  }

  const account = await nodemailer.createTestAccount();
  console.log(`Generated Ethereal account: ${account.user}`);
  console.log(`Ethereal inbox: ${account.web}`);

  return nodemailer.createTransport({
    host: account.smtp.host,
    port: account.smtp.port,
    secure: account.smtp.secure,
    auth: {
      user: account.user,
      pass: account.pass,
    },
  });
}

How is the test transport selected?

Configured Ethereal credentials create a transport for the existing test account. Empty credentials make Nodemailer generate a test account when the first job needs it.

The promise is cached so later jobs reuse the same transporter.

Add getTransporter() and sendEmail() below createEtherealTransport() with this code:

function getTransporter(): Promise<Transporter> {
  transporterPromise ??= createEtherealTransport();
  return transporterPromise;
}

export async function sendEmail(input: {
  senderName: string;
  senderEmail: string;
  toEmail: string;
  subject: string;
  textBody: string;
}): Promise<{ messageId: string; previewUrl: string | null }> {
  const transporter = await getTransporter();
  const info = await transporter.sendMail({
    from: `"${input.senderName}" <${input.senderEmail}>`,
    to: input.toEmail,
    subject: input.subject,
    text: input.textBody,
  });

  const preview = nodemailer.getTestMessageUrl(info);
  return {
    messageId: info.messageId,
    previewUrl: typeof preview === "string" ? preview : null,
  };
}

What does the mailer return?

The transport sends the sender details and message body to Ethereal. Ethereal captures the message without delivering it to the real recipient address.

The returned message ID records the provider result. The preview URL gives you browser-visible proof of the captured email.

Save src/mailer.ts.
Confirm that src/mailer.ts appears in the editor sidebar.

Mailer types not resolving?

Check that the import uses the default nodemailer export plus the Transporter type from the same package.

Confirm that your dependency installation includes the version from the existing package manifest.

Share the compiler message without sharing any credentials. Help me fix the Nodemailer TypeScript error in src/mailer.ts.

✔️ Awesome, I've got everything!

Your mailer can now reuse a configured or generated Ethereal transport.

ⓧ I'd like to double check the full code

Compare your complete src/mailer.ts file with this reference.

import nodemailer, { type Transporter } from "nodemailer";
import { config } from "./config.js";

let transporterPromise: Promise<Transporter> | undefined;

async function createEtherealTransport(): Promise<Transporter> {
  if (config.etherealUser && config.etherealPass) {
    return nodemailer.createTransport({
      service: "Ethereal",
      auth: {
        user: config.etherealUser,
        pass: config.etherealPass,
      },
    });
  }

  const account = await nodemailer.createTestAccount();
  console.log(`Generated Ethereal account: ${account.user}`);
  console.log(`Ethereal inbox: ${account.web}`);

  return nodemailer.createTransport({
    host: account.smtp.host,
    port: account.smtp.port,
    secure: account.smtp.secure,
    auth: {
      user: account.user,
      pass: account.pass,
    },
  });
}

function getTransporter(): Promise<Transporter> {
  transporterPromise ??= createEtherealTransport();
  return transporterPromise;
}

export async function sendEmail(input: {
  senderName: string;
  senderEmail: string;
  toEmail: string;
  subject: string;
  textBody: string;
}): Promise<{ messageId: string; previewUrl: string | null }> {
  const transporter = await getTransporter();
  const info = await transporter.sendMail({
    from: `"${input.senderName}" <${input.senderEmail}>`,
    to: input.toEmail,
    subject: input.subject,
    text: input.textBody,
  });

  const preview = nodemailer.getTestMessageUrl(info);
  return {
    messageId: info.messageId,
    previewUrl: typeof preview === "string" ? preview : null,
  };
}

Create src/worker.ts in the existing src folder.
Build the worker so it locks email rows before delivery.
Use WORKER_CONCURRENCY to control parallel processing.
Store successful delivery details in PostgreSQL.
Preserve ambiguous delivery outcomes as uncertain.
Compare your worker with the complete reference below.

✔️ Awesome, I've got everything!

Save src/worker.ts once the claim, delivery, event handling, and shutdown sections match the target.

ⓧ I'd like to double check the full code

At this point, your complete src/worker.ts file should look like this.

import { DelayedError, Worker } from "bullmq";
import { Redis } from "ioredis";
import { config } from "./config.js";
import { initializeDatabase, pool } from "./db.js";
import { sendEmail } from "./mailer.js";

type EmailRow = {
  id: string;
  tenant_id: string;
  sender_id: string;
  sender_name: string;
  sender_email: string;
  to_email: string;
  subject: string;
  text_body: string;
  status: string;
  processing_started_at: Date | null;
};

type Claim =
  | { kind: "work"; email: EmailRow }
  | { kind: "delay"; waitMs: number }
  | { kind: "skip" };

async function claimEmail(emailId: string): Promise<Claim> {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const result = await client.query<EmailRow>(
      `SELECT e.*, s.name AS sender_name, s.email AS sender_email
       FROM emails e
       JOIN senders s ON s.id = e.sender_id
       WHERE e.id = $1
       FOR UPDATE OF e`,
      [emailId],
    );
    const email = result.rows[0];
    if (!email || ["sent", "failed", "uncertain"].includes(email.status)) {
      await client.query("COMMIT");
      return { kind: "skip" };
    }

    if (["processing", "sending"].includes(email.status)) {
      const started = email.processing_started_at?.getTime() ?? 0;
      const age = Date.now() - started;
      if (age < config.sendLeaseMs) {
        await client.query("COMMIT");
        return { kind: "delay", waitMs: config.sendLeaseMs - age };
      }
      if (email.status === "sending") {
        await client.query(
          `UPDATE emails
           SET status = 'uncertain',
               last_error = 'SMTP outcome unknown after worker interruption',
               updated_at = now()
           WHERE id = $1`,
          [emailId],
        );
        await client.query("COMMIT");
        return { kind: "skip" };
      }
    }

    const claimed = await client.query<EmailRow>(
      `UPDATE emails
       SET status = 'processing',
           processing_started_at = now(),
           attempts = attempts + 1,
           updated_at = now()
       WHERE id = $1
       RETURNING *,
         $2::text AS sender_name,
         $3::text AS sender_email`,
      [emailId, email.sender_name, email.sender_email],
    );
    await client.query("COMMIT");
    return { kind: "work", email: claimed.rows[0]! };
  } catch (error) {
    await client.query("ROLLBACK");
    throw error;
  } finally {
    client.release();
  }
}

await initializeDatabase();
const workerRedis = new Redis(config.redisUrl, {
  maxRetriesPerRequest: null,
});

const worker = new Worker(
  "email-send",
  async (job, token) => {
    const emailId = String(job.data.emailId);
    const claim = await claimEmail(emailId);

    if (claim.kind === "skip") return { skipped: true };
    if (claim.kind === "delay") {
      await job.moveToDelayed(Date.now() + claim.waitMs, token);
      throw new DelayedError();
    }

    const email = claim.email;
    await pool.query(
      `UPDATE emails SET status = 'sending', updated_at = now() WHERE id = $1`,
      [email.id],
    );

    try {
      const sent = await sendEmail({
        senderName: email.sender_name,
        senderEmail: email.sender_email,
        toEmail: email.to_email,
        subject: email.subject,
        textBody: email.text_body,
      });
      await pool.query(
        `UPDATE emails
         SET status = 'sent', provider_message_id = $2, preview_url = $3,
             sent_at = now(), last_error = NULL, updated_at = now()
         WHERE id = $1`,
        [email.id, sent.messageId, sent.previewUrl],
      );
      return sent;
    } catch (error) {
      const message = error instanceof Error ? error.message : String(error);
      await pool.query(
        `UPDATE emails
         SET status = 'uncertain', last_error = $2, updated_at = now()
         WHERE id = $1`,
        [email.id, message],
      );
      return { uncertain: true, reason: message };
    }
  },
  {
    connection: workerRedis,
    concurrency: config.workerConcurrency,
    maxStartedAttempts: 1000,
  },
);

worker.on("completed", (job) => {
  console.log(`Completed job ${job.id}`);
});
worker.on("failed", (job, error) => {
  console.error(`Failed job ${job?.id ?? "unknown"}`, error);
});
worker.on("error", (error) => {
  console.error("Worker error", error);
});

async function shutdown(): Promise<void> {
  await worker.close();
  await pool.end();
  await workerRedis.quit();
}

process.on("SIGINT", () => void shutdown().then(() => process.exit(0)));
process.on("SIGTERM", () => void shutdown().then(() => process.exit(0)));

console.log(
  `Email worker started with concurrency ${config.workerConcurrency}`,
);

How does the worker protect delivery state?

The PostgreSQL transaction locks the email row before changing its status to processing. A second worker cannot claim the same row at that moment.

A successful send stores provider_message_id, preview_url, and sent_at before the job completes.

A delivery exception becomes uncertain. This state prevents an automatic resend when the SMTP outcome cannot be proven.

Save src/worker.ts.
Return to your terminal panel from the earlier steps.
Create a second terminal session for the independent worker.
Start the worker in the second terminal by running:

npm run dev:worker

You will see the worker report its configured concurrency. The queued email remains delayed until its scheduled time.

Worker not starting?

Confirm that the API still owns the first terminal while the worker runs in a separate terminal session.

Check the Redis service if the worker repeatedly reports connection failures.

Use the first worker stack trace to identify the file that needs attention. Help me debug npm run dev:worker using this terminal output.

Watch a delayed email become sent

Bull Board shows the queue lifecycle while PostgreSQL shows the business lifecycle. The Ethereal preview connects those two views to a captured message you can open.

Update the ## Run section in README.md with these runtime steps:

4. In one terminal run `npm run dev:api`.
5. In another terminal run `npm run dev:worker`.
6. Open `http://localhost:3000/admin/queues` and sign in as `admin` with `ADMIN_KEY`.

Why document separate processes?

The API and worker have separate responsibilities. Running them in separate terminals makes it clear which process accepts requests and which process performs delivery.

The dashboard instruction records how to inspect the queue without exposing the admin credential in the README.

Save README.md.
Confirm that the ## Run section now names both development processes and the protected queue route.

Open http://localhost:3000/admin/queues in your browser.
Enter admin as the username.
Enter the value of ADMIN_KEY from your local .env file as the password.
Select the email-send queue.

You will see the scheduled job in the delayed state while its due time is still in the future.

Before the due time arrives, do you expect the job to remain delayed or complete immediately?

Wait until the scheduled time passes.
Refresh the email-send queue in Bull Board.

You will see the job move to the completed state. The worker terminal also reports the completed stable job ID.

Open http://localhost:3000/api/emails?status=sent in your browser.
Find the email you scheduled in the returned JSON.
Open its preview_url value in a new browser tab.

You will see the captured subject and body in Ethereal. The API record now includes sent, a provider message ID, a preview URL, and a send time.

That is the full delivery loop working: the API schedules the job, BullMQ releases it when due, the worker sends it, and PostgreSQL stores the result.

Job delayed after its due time?

Check that the worker terminal still shows an active process. A stopped worker leaves due jobs waiting in Redis.

Confirm that the API response used a valid future ISO date-time and that your machine clock is correct.

Share the worker output without sharing ADMIN_KEY or Ethereal credentials. Help me diagnose why my BullMQ email job did not complete.

Strong work. Your scheduler now turns persisted requests into delayed test deliveries with queue visibility and stored preview links. Next, you will make that flow safer across restarts and parallel workers.

Protect Recovery and Parallel Delivery

Your BullMQ worker now turns delayed jobs into test emails. However, a crash between a PostgreSQL write and a Redis enqueue can leave a stored email without a runnable job.

This step reconciles that split state after a restart. It also protects parallel delivery when an SMTP outcome becomes ambiguous.

In this step, get ready to:

Reconcile non-final email records into BullMQ during API startup.
Protect worker claims with row locking and processing leases.
Prove restart recovery and idempotent scheduling.

Reconcile non-final emails

PostgreSQL holds the scheduler's business records. Startup reconciliation reads every non-final record from that durable state and restores its stable queue job.

In src/queue.ts, find recoverNonFinalEmails().
Replace the existing recovery helper with this implementation:

export async function recoverNonFinalEmails(): Promise<number> {
  const result = await pool.query<{
    id: string;
    scheduled_at: Date;
  }>(`
    SELECT id, scheduled_at
    FROM emails
    WHERE status IN (
      'scheduled', 'queued', 'processing', 'sending', 'rate_limited'
    )
    ORDER BY scheduled_at ASC
  `);

  for (const email of result.rows) {
    await enqueueEmail(email.id, new Date(email.scheduled_at));
    await pool.query(
      `UPDATE emails
       SET status = CASE
         WHEN status = 'scheduled' THEN 'queued'
         ELSE status
       END,
       updated_at = now()
       WHERE id = $1`,
      [email.id],
    );
  }

  return result.rowCount ?? 0;
}

What does this recovery do?

The query selects records that still need delivery work.
Each record returns to enqueueEmail() with its original schedule.
The stable email-<email-id> job ID prevents a second retained job from being added.
A newly recovered scheduled record changes to queued.

Save src/queue.ts.
Confirm src/server.ts still calls recoverNonFinalEmails() before creating the Express app.
Return to the API terminal from earlier.
Stop the API with Ctrl+C.
Start the API with reconciliation enabled by running:

npm run dev:api

What does this command prove?

The API startup path initializes the database before running reconciliation. The recovered count records how many non-final rows were considered.

You should see the API listening message. You should also see a recovered count without a startup failure.

API failing during recovery?

Confirm the PostgreSQL and Redis containers from earlier are still running.

Check that every status in the recovery query matches the email status constraint in src/db.ts.

Use the startup output to help me diagnose why recoverNonFinalEmails() prevents my API from starting.

✔️ Awesome, I've got everything!

Your recovery helper now restores non-final records with stable BullMQ job IDs.

ⓧ I'd like to double check the full code

Compare your complete src/queue.ts file with this reference.

import { Queue } from "bullmq";
import { Redis } from "ioredis";
import { config } from "./config.js";
import { pool } from "./db.js";

export const producerRedis = new Redis(config.redisUrl, {
  maxRetriesPerRequest: 1,
});

export const emailQueue = new Queue("email-send", {
  connection: producerRedis,
});

export async function enqueueEmail(
  emailId: string,
  scheduledAt: Date,
): Promise<void> {
  const delay = Math.max(0, scheduledAt.getTime() - Date.now());
  await emailQueue.add(
    "send-email",
    { emailId },
    {
      delay,
      jobId: `email-${emailId}`,
    },
  );
}

export async function recoverNonFinalEmails(): Promise<number> {
  const result = await pool.query<{
    id: string;
    scheduled_at: Date;
  }>(`
    SELECT id, scheduled_at
    FROM emails
    WHERE status IN (
      'scheduled', 'queued', 'processing', 'sending', 'rate_limited'
    )
    ORDER BY scheduled_at ASC
  `);

  for (const email of result.rows) {
    await enqueueEmail(email.id, new Date(email.scheduled_at));
    await pool.query(
      `UPDATE emails
       SET status = CASE
         WHEN status = 'scheduled' THEN 'queued'
         ELSE status
       END,
       updated_at = now()
       WHERE id = $1`,
      [email.id],
    );
  }

  return result.rowCount ?? 0;
}

The recovery contract also belongs in the project documentation. It explains why both storage systems have distinct responsibilities.

In README.md, find the scheduling and restart section.
Replace that section with this documentation:

## Scheduling and restart behavior

PostgreSQL is the business source of truth. Redis stores BullMQ job state. The API calculates the delay from `scheduledAt`, uses `email-<email-id>` as the BullMQ job ID, and retains finalized jobs. At API startup, non-final database rows are re-added. Existing BullMQ IDs are ignored, so reconciliation does not create a second queue job.

Redis runs with AOF enabled and named Docker storage. PostgreSQL also uses a named volume. `docker compose down` keeps both volumes. `docker compose down -v` deletes both.

Why document both stores?

PostgreSQL remains the durable business source of truth. Redis holds the queue state needed to execute that work.

The recovery loop reconnects those responsibilities after an interrupted enqueue.

Save README.md.
Preview README.md and confirm the scheduling section describes stable job IDs plus startup reconciliation.

Guard final and leased records

Multiple workers can reach the same database record close together. The existing SELECT FOR UPDATE row lock serializes each claim inside a short transaction.

A processing lease handles interrupted work. A stale sending record receives the conservative uncertain state because the SMTP server may already have accepted it.

In src/worker.ts, find const email = result.rows[0]; inside claimEmail().
Replace the status checks below that line with this lease guard:

    if (!email || ["sent", "failed", "uncertain"].includes(email.status)) {
      await client.query("COMMIT");
      return { kind: "skip" };
    }

    if (["processing", "sending"].includes(email.status)) {
      const started = email.processing_started_at?.getTime() ?? 0;
      const age = Date.now() - started;
      if (age < config.sendLeaseMs) {
        await client.query("COMMIT");
        return { kind: "delay", waitMs: config.sendLeaseMs - age };
      }
      if (email.status === "sending") {
        await client.query(
          `UPDATE emails
           SET status = 'uncertain',
               last_error = 'SMTP outcome unknown after worker interruption',
               updated_at = now()
           WHERE id = $1`,
          [emailId],
        );
        await client.query("COMMIT");
        return { kind: "skip" };
      }
    }

How do the lease checks protect delivery?

Final sent, failed, and uncertain records leave the worker without another delivery attempt.
Recent processing or sending work moves back to delayed processing until its lease expires.
Expired processing work can be claimed again because no SMTP attempt had begun.
Expired sending work becomes uncertain because another send could create a duplicate email.

Save src/worker.ts.
Return to the worker terminal from earlier.
Stop the worker with Ctrl+C.
Restart the worker with the lease guard by running:

npm run dev:worker

What does this restart check?

The worker loads the updated claim path before accepting queue work. A successful startup confirms the lease code compiles.

You should see the worker startup message with its configured concurrency.

Worker failing after the lease edit?

Check that the lease guard remains inside the try block in claimEmail().

Confirm each early return commits the transaction before returning.

Share the compiler output to help me fix the lease checks inside claimEmail().

A successful claim must record when work started. It also increments the attempt count for later inspection.

In claimEmail(), find the database update that changes an email to processing.
Replace that claim update with this version:

    const claimed = await client.query<EmailRow>(
      `UPDATE emails
       SET status = 'processing',
           processing_started_at = now(),
           attempts = attempts + 1,
           updated_at = now()
       WHERE id = $1
       RETURNING *,
         $2::text AS sender_name,
         $3::text AS sender_email`,
      [emailId, email.sender_name, email.sender_email],
    );
    await client.query("COMMIT");
    return { kind: "work", email: claimed.rows[0]! };

What does the claim record?

The processing status shows that a worker owns the current attempt.
The processing_started_at timestamp anchors the lease calculation.
The attempts counter records how many times the record entered active processing.
The transaction commits before the worker starts the delivery path.

Save src/worker.ts.
Stop the worker with Ctrl+C.
Start the completed worker by running:

npm run dev:worker

What does the completed claim path do?

The worker locks one record before deciding whether to skip, delay, quarantine, or claim it. This keeps parallel workers from claiming the same email at once.

You should see the worker startup message again. The process should remain ready for jobs.

Worker exits during startup?

Confirm the claim update remains before the surrounding catch and finally blocks.

Check that claimed.rows[0]! is returned only after the transaction commits.

Use the terminal output to help me repair the claim transaction in src/worker.ts.

The conservative SMTP decision needs to be visible to anyone maintaining this scheduler. Documenting the trade-off prevents a future retry change from reintroducing duplicate delivery.

In README.md, locate the idempotency and SMTP section.
Replace that section with this explanation:

## Idempotency and SMTP trade-off

The scheduling API has a PostgreSQL unique constraint on `(tenant_id, idempotency_key)` and BullMQ uses a stable custom job ID. This prevents duplicate API retries from creating multiple records or queue jobs.

SMTP does not expose a provider idempotency key. A process can fail after the SMTP server accepts a message but before PostgreSQL records `sent`. Strict exactly-once delivery is therefore impossible with plain SMTP. This implementation chooses the no-automatic-duplicate side of that trade-off: stale or failed `sending` work becomes `uncertain` and requires manual review instead of being resent automatically.

Why choose uncertain?

The sending state means the worker crossed the boundary into an SMTP attempt. A crash after acceptance can hide a successful delivery from PostgreSQL.

The uncertain state stops automatic resend and leaves the outcome for manual review.

Save README.md.
Preview README.md and confirm it explains idempotency plus ambiguous SMTP outcomes.

✔️ Awesome, I've got everything!

Your worker now uses transactional claims, processing leases, and conservative delivery outcomes.

ⓧ I'd like to double check the full code

Compare your complete src/worker.ts file with this reference.

import { DelayedError, Worker } from "bullmq";
import { Redis } from "ioredis";
import { config } from "./config.js";
import { initializeDatabase, pool } from "./db.js";
import { sendEmail } from "./mailer.js";

type EmailRow = {
  id: string;
  tenant_id: string;
  sender_id: string;
  sender_name: string;
  sender_email: string;
  to_email: string;
  subject: string;
  text_body: string;
  status: string;
  processing_started_at: Date | null;
};

type Claim =
  | { kind: "work"; email: EmailRow }
  | { kind: "delay"; waitMs: number }
  | { kind: "skip" };

async function claimEmail(emailId: string): Promise<Claim> {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const result = await client.query<EmailRow>(
      `SELECT e.*, s.name AS sender_name, s.email AS sender_email
       FROM emails e
       JOIN senders s ON s.id = e.sender_id
       WHERE e.id = $1
       FOR UPDATE OF e`,
      [emailId],
    );
    const email = result.rows[0];
    if (!email || ["sent", "failed", "uncertain"].includes(email.status)) {
      await client.query("COMMIT");
      return { kind: "skip" };
    }

    if (["processing", "sending"].includes(email.status)) {
      const started = email.processing_started_at?.getTime() ?? 0;
      const age = Date.now() - started;
      if (age < config.sendLeaseMs) {
        await client.query("COMMIT");
        return { kind: "delay", waitMs: config.sendLeaseMs - age };
      }
      if (email.status === "sending") {
        await client.query(
          `UPDATE emails
           SET status = 'uncertain',
               last_error = 'SMTP outcome unknown after worker interruption',
               updated_at = now()
           WHERE id = $1`,
          [emailId],
        );
        await client.query("COMMIT");
        return { kind: "skip" };
      }
    }

    const claimed = await client.query<EmailRow>(
      `UPDATE emails
       SET status = 'processing',
           processing_started_at = now(),
           attempts = attempts + 1,
           updated_at = now()
       WHERE id = $1
       RETURNING *,
         $2::text AS sender_name,
         $3::text AS sender_email`,
      [emailId, email.sender_name, email.sender_email],
    );
    await client.query("COMMIT");
    return { kind: "work", email: claimed.rows[0]! };
  } catch (error) {
    await client.query("ROLLBACK");
    throw error;
  } finally {
    client.release();
  }
}

await initializeDatabase();
const workerRedis = new Redis(config.redisUrl, {
  maxRetriesPerRequest: null,
});

const worker = new Worker(
  "email-send",
  async (job, token) => {
    const emailId = String(job.data.emailId);
    const claim = await claimEmail(emailId);

    if (claim.kind === "skip") return { skipped: true };
    if (claim.kind === "delay") {
      await job.moveToDelayed(Date.now() + claim.waitMs, token);
      throw new DelayedError();
    }

    const email = claim.email;
    await pool.query(
      `UPDATE emails SET status = 'sending', updated_at = now() WHERE id = $1`,
      [email.id],
    );

    try {
      const sent = await sendEmail({
        senderName: email.sender_name,
        senderEmail: email.sender_email,
        toEmail: email.to_email,
        subject: email.subject,
        textBody: email.text_body,
      });
      await pool.query(
        `UPDATE emails
         SET status = 'sent', provider_message_id = $2, preview_url = $3,
             sent_at = now(), last_error = NULL, updated_at = now()
         WHERE id = $1`,
        [email.id, sent.messageId, sent.previewUrl],
      );
      return sent;
    } catch (error) {
      const message = error instanceof Error ? error.message : String(error);
      await pool.query(
        `UPDATE emails
         SET status = 'uncertain', last_error = $2, updated_at = now()
         WHERE id = $1`,
        [email.id, message],
      );
      return { uncertain: true, reason: message };
    }
  },
  {
    connection: workerRedis,
    concurrency: config.workerConcurrency,
    maxStartedAttempts: 1000,
  },
);

worker.on("completed", (job) => {
  console.log(`Completed job ${job.id}`);
});
worker.on("failed", (job, error) => {
  console.error(`Failed job ${job?.id ?? "unknown"}`, error);
});
worker.on("error", (error) => {
  console.error("Worker error", error);
});

async function shutdown(): Promise<void> {
  await worker.close();
  await pool.end();
  await workerRedis.quit();
}

process.on("SIGINT", () => void shutdown().then(() => process.exit(0)));
process.on("SIGTERM", () => void shutdown().then(() => process.exit(0)));

console.log(
  `Email worker started with concurrency ${config.workerConcurrency}`,
);

Prove restart recovery and idempotency

A future email gives you time to stop both Node processes while its durable database record and delayed queue job remain. Restarting before the due time tests recovery without entering the ambiguous SMTP window.

Return to the saved POST /api/emails request from earlier.
Set the Idempotency-Key header to recovery-drill-1.
Set scheduledAt to an ISO date-time at least five minutes in the future.
Send the scheduling request.
Record the returned email ID here: your recovery email ID.

What does this drill isolate?

The email remains non-final while both application processes stop. PostgreSQL plus the named Redis volume preserve the state needed for the restart.

Return to the API terminal.
Stop the API with Ctrl+C.
Return to the worker terminal.
Stop the worker with Ctrl+C.

Before you restart, how many retained jobs do you expect Bull Board to show for this email?

Restart the API in its terminal by running:

npm run dev:api

What happens during API startup?

The API reconciles every non-final row before it starts listening. BullMQ ignores the duplicate stable job ID while the retained job exists.

Restart the worker in its terminal by running:

npm run dev:worker

What happens during worker startup?

The worker reconnects to the persistent queue and waits for the retained delayed job. Its database claim protects the record when the due time arrives.

Wait until the scheduled time passes.
Return to /admin/queues in the browser from earlier.
Refresh Bull Board.
Open the completed email- job for your recovery email ID.
Return to your API client.
Send GET /api/emails?status=sent.

You should see one completed BullMQ job for the recorded ID. The matching API record should appear once with a preview URL and the sent status.

Before you repeat the request, do you expect the scheduler to create a new email ID?

Return to the original POST /api/emails request.
Send the request again with the same Idempotency-Key header and unchanged body.
Compare the returned ID with your recovery email ID.
Refresh Bull Board.

You should receive the same persisted email ID. Bull Board should still contain one retained job with that stable ID.

Seeing a second email or a missing job?

Confirm both scheduling requests used the same tenant plus the exact same Idempotency-Key value.

Check that enqueueEmail() still builds its job ID from the persisted email ID.

Share the two responses to help me diagnose why my recovery or idempotency check created the wrong result.

That is the recovery proof complete. Your scheduler survives application restarts without duplicating retained work.

Next, you will enforce shared sender limits and deliver operator alerts through Slack.

Enforce Limits and Connect Slack

Your scheduler now recovers jobs after a restart. The next risk comes from parallel workers racing through one sender's quota.

An atomic Redis reservation gives every worker the same decision. A tenant-scoped Slack connection then tells operators when the hourly limit is reached.

In this step, get ready to:

Enforce an hourly sender quota with minimum send spacing.
Connect tenant demo to Slack through OAuth 2.0.
Trigger one rate-limit alert with a three-email burst.

Build the distributed sender limiter

An in-memory counter belongs to one worker process. An atomic Lua script keeps the quota decision inside Redis so every worker shares the same count.

Why use a Redis Lua reservation?

The script checks the hourly counter before reserving the spacing key. Redis evaluates the complete decision atomically.

A notification key records the first hourly rejection. That key limits Slack alert attempts to one per sender-hour.

Create src/rate-limit.ts inside the existing project folder.
Replace src/worker.ts with the limiter-aware worker in the full-code tab below.

✔️ Awesome, I've got everything!

Confirm that src/rate-limit.ts exists. Confirm that src/worker.ts reserves a sender slot before delivery.

ⓧ I'd like to double check the full code

Use this final src/rate-limit.ts file for the atomic Redis reservation.

import { Redis } from "ioredis";
import { config } from "./config.js";

const redis = new Redis(config.redisUrl, {
  maxRetriesPerRequest: null,
});

const reserveScript = `
local current = tonumber(redis.call('GET', KEYS[1]) or '0')
local maximum = tonumber(ARGV[1])
local window_ms = tonumber(ARGV[2])
local spacing_ms = tonumber(ARGV[3])

if current >= maximum then
  local notify = redis.call('SET', KEYS[3], '1', 'NX', 'PX', window_ms)
  return {0, window_ms, 1, notify and 1 or 0, current}
end

local spacing = redis.call('SET', KEYS[2], '1', 'NX', 'PX', spacing_ms)
if not spacing then
  local wait_ms = redis.call('PTTL', KEYS[2])
  return {0, wait_ms, 0, 0, current}
end

current = redis.call('INCR', KEYS[1])
if current == 1 then
  redis.call('PEXPIRE', KEYS[1], window_ms)
end

return {1, 0, 0, 0, current}
`;

export type Reservation = {
  allowed: boolean;
  waitMs: number;
  hourlyLimitHit: boolean;
  shouldNotify: boolean;
  count: number;
};

export async function reserveSend(senderId: string): Promise<Reservation> {
  const now = Date.now();
  const hourStart = Math.floor(now / 3_600_000) * 3_600_000;
  const windowMs = hourStart + 3_600_000 - now;
  const keys = [
    `email-rate-hour-${senderId}-${hourStart}`,
    `email-spacing-${senderId}`,
    `email-rate-notified-${senderId}-${hourStart}`,
  ];

  const raw = (await redis.eval(
    reserveScript,
    3,
    ...keys,
    config.maxEmailsPerHourPerSender,
    windowMs,
    config.minSendDelayMs,
  )) as Array<number | string>;

  return {
    allowed: Number(raw[0]) === 1,
    waitMs: Math.max(250, Number(raw[1])),
    hourlyLimitHit: Number(raw[2]) === 1,
    shouldNotify: Number(raw[3]) === 1,
    count: Number(raw[4]),
  };
}

export async function closeRateLimiter(): Promise<void> {
  await redis.quit();
}

What does this limiter do?

The hourly key stores the sender's count for the aligned UTC hour.
The spacing key blocks another reservation until MIN_SEND_DELAY_MS expires.
The notification key allows one alert attempt during that sender-hour.
The returned wait time tells BullMQ when to retry a denied job.

Use this final src/worker.ts file to apply the reservation before SMTP delivery.

import { DelayedError, Worker } from "bullmq";
import { Redis } from "ioredis";
import { config } from "./config.js";
import { initializeDatabase, pool } from "./db.js";
import { sendEmail } from "./mailer.js";
import { closeRateLimiter, reserveSend } from "./rate-limit.js";
import { sendRateLimitAlert } from "./slack.js";

type EmailRow = {
  id: string;
  tenant_id: string;
  sender_id: string;
  sender_name: string;
  sender_email: string;
  to_email: string;
  subject: string;
  text_body: string;
  status: string;
  processing_started_at: Date | null;
};

type Claim =
  | { kind: "work"; email: EmailRow }
  | { kind: "delay"; waitMs: number }
  | { kind: "skip" };

async function claimEmail(emailId: string): Promise<Claim> {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const result = await client.query<EmailRow>(
      `SELECT e.*, s.name AS sender_name, s.email AS sender_email
       FROM emails e
       JOIN senders s ON s.id = e.sender_id
       WHERE e.id = $1
       FOR UPDATE OF e`,
      [emailId],
    );
    const email = result.rows[0];
    if (!email || ["sent", "failed", "uncertain"].includes(email.status)) {
      await client.query("COMMIT");
      return { kind: "skip" };
    }

    if (["processing", "sending"].includes(email.status)) {
      const started = email.processing_started_at?.getTime() ?? 0;
      const age = Date.now() - started;
      if (age < config.sendLeaseMs) {
        await client.query("COMMIT");
        return { kind: "delay", waitMs: config.sendLeaseMs - age };
      }
      if (email.status === "sending") {
        await client.query(
          `UPDATE emails
           SET status = 'uncertain',
               last_error = 'SMTP outcome unknown after worker interruption',
               updated_at = now()
           WHERE id = $1`,
          [emailId],
        );
        await client.query("COMMIT");
        return { kind: "skip" };
      }
    }

    const claimed = await client.query<EmailRow>(
      `UPDATE emails
       SET status = 'processing',
           processing_started_at = now(),
           attempts = attempts + 1,
           updated_at = now()
       WHERE id = $1
       RETURNING *,
         $2::text AS sender_name,
         $3::text AS sender_email`,
      [emailId, email.sender_name, email.sender_email],
    );
    await client.query("COMMIT");
    return { kind: "work", email: claimed.rows[0]! };
  } catch (error) {
    await client.query("ROLLBACK");
    throw error;
  } finally {
    client.release();
  }
}

await initializeDatabase();
const workerRedis = new Redis(config.redisUrl, {
  maxRetriesPerRequest: null,
});

const worker = new Worker(
  "email-send",
  async (job, token) => {
    const emailId = String(job.data.emailId);
    const claim = await claimEmail(emailId);

    if (claim.kind === "skip") return { skipped: true };
    if (claim.kind === "delay") {
      await job.moveToDelayed(Date.now() + claim.waitMs, token);
      throw new DelayedError();
    }

    const email = claim.email;
    const reservation = await reserveSend(email.sender_id);
    if (!reservation.allowed) {
      const retryAt = new Date(Date.now() + reservation.waitMs);
      await pool.query(
        `UPDATE emails
         SET status = 'rate_limited',
             rate_limited_until = $2,
             processing_started_at = NULL,
             updated_at = now()
         WHERE id = $1`,
        [email.id, retryAt],
      );

      if (reservation.hourlyLimitHit && reservation.shouldNotify) {
        try {
          await sendRateLimitAlert({
            tenantId: email.tenant_id,
            senderEmail: email.sender_email,
            maximum: config.maxEmailsPerHourPerSender,
          });
        } catch (error) {
          console.error("Slack alert failed without failing the email job", error);
        }
      }

      await job.moveToDelayed(retryAt.getTime(), token);
      throw new DelayedError();
    }

    await pool.query(
      `UPDATE emails SET status = 'sending', updated_at = now() WHERE id = $1`,
      [email.id],
    );

    try {
      const sent = await sendEmail({
        senderName: email.sender_name,
        senderEmail: email.sender_email,
        toEmail: email.to_email,
        subject: email.subject,
        textBody: email.text_body,
      });
      await pool.query(
        `UPDATE emails
         SET status = 'sent', provider_message_id = $2, preview_url = $3,
             sent_at = now(), last_error = NULL, updated_at = now()
         WHERE id = $1`,
        [email.id, sent.messageId, sent.previewUrl],
      );
      return sent;
    } catch (error) {
      const message = error instanceof Error ? error.message : String(error);
      await pool.query(
        `UPDATE emails
         SET status = 'uncertain', last_error = $2, updated_at = now()
         WHERE id = $1`,
        [email.id, message],
      );
      return { uncertain: true, reason: message };
    }
  },
  {
    connection: workerRedis,
    concurrency: config.workerConcurrency,
    maxStartedAttempts: 1000,
  },
);

worker.on("completed", (job) => {
  console.log(`Completed job ${job.id}`);
});
worker.on("failed", (job, error) => {
  console.error(`Failed job ${job?.id ?? "unknown"}`, error);
});
worker.on("error", (error) => {
  console.error("Worker error", error);
});

async function shutdown(): Promise<void> {
  await worker.close();
  await closeRateLimiter();
  await pool.end();
  await workerRedis.quit();
}

process.on("SIGINT", () => void shutdown().then(() => process.exit(0)));
process.on("SIGTERM", () => void shutdown().then(() => process.exit(0)));

console.log(
  `Email worker started with concurrency ${config.workerConcurrency}`,
);

How does the worker defer jobs?

The worker reserves quota after claiming the PostgreSQL record.
A denied reservation stores rate_limited with a retry timestamp.
BullMQ moves the active job back to its delayed set.
A Slack failure is logged without failing the email job.

Save src/rate-limit.ts.
Save src/worker.ts.
Compile the limiter changes by running:

npm run build

What does this check prove?

The TypeScript build checks the new imports. It also checks the reservation result used by the worker.

The build should complete without TypeScript errors. Your worker can now coordinate sender limits through Redis.

Seeing a TypeScript error?

Check that both new imports in src/worker.ts include their .js suffixes.

Confirm that src/rate-limit.ts exports reserveSend and closeRateLimiter.

Ask for help with the exact compiler output: Help me debug my rate limiter build error.

Add tenant-scoped Slack OAuth

OAuth 2.0 lets tenant demo approve a Slack app without placing a bot token in the browser. The callback validates a short-lived state before storing the installation.

Update src/config.ts with the Slack environment settings.
Update src/db.ts with tenant-scoped installation tables.

✔️ Awesome, I've got everything!

Confirm that src/config.ts reads the Slack settings. Confirm that src/db.ts creates both Slack tables.

ⓧ I'd like to double check the full code

Your final src/config.ts should match this file.

import "dotenv/config";

function integer(name: string, fallback: number): number {
  const value = Number(process.env[name] ?? fallback);
  if (!Number.isInteger(value) || value <= 0) {
    throw new Error(`${name} must be a positive integer`);
  }
  return value;
}

export const config = {
  port: integer("PORT", 3000),
  databaseUrl:
    process.env.DATABASE_URL ??
    "postgres://scheduler:scheduler@localhost:5432/scheduler",
  redisUrl: process.env.REDIS_URL ?? "redis://localhost:6379",
  frontendOrigin: process.env.FRONTEND_ORIGIN ?? "http://localhost:5173",
  workerConcurrency: integer("WORKER_CONCURRENCY", 5),
  minSendDelayMs: integer("MIN_SEND_DELAY_MS", 2000),
  maxEmailsPerHourPerSender: integer(
    "MAX_EMAILS_PER_HOUR_PER_SENDER",
    200,
  ),
  sendLeaseMs: integer("SEND_LEASE_MS", 60000),
  adminKey: process.env.ADMIN_KEY ?? "replace-with-a-long-random-value",
  etherealUser: process.env.ETHEREAL_USER,
  etherealPass: process.env.ETHEREAL_PASS,
  slackClientId: process.env.SLACK_CLIENT_ID,
  slackClientSecret: process.env.SLACK_CLIENT_SECRET,
  slackRedirectUri: process.env.SLACK_REDIRECT_URI,
};

What does this configuration add?

The Slack client credentials stay in the local environment. The redirect URI remains configurable because each Quick Tunnel receives a temporary hostname.

Your final src/db.ts should match this file.

import { Pool } from "pg";
import { config } from "./config.js";

export const pool = new Pool({
  connectionString: config.databaseUrl,
  max: 10,
});

export async function initializeDatabase(): Promise<void> {
  await pool.query(`
    CREATE TABLE IF NOT EXISTS tenants (
      id TEXT PRIMARY KEY,
      name TEXT NOT NULL
    );

    CREATE TABLE IF NOT EXISTS senders (
      id TEXT PRIMARY KEY,
      tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
      name TEXT NOT NULL,
      email TEXT NOT NULL,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      UNIQUE (tenant_id, email)
    );

    CREATE TABLE IF NOT EXISTS emails (
      id TEXT PRIMARY KEY,
      tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
      sender_id TEXT NOT NULL REFERENCES senders(id),
      idempotency_key TEXT NOT NULL,
      to_email TEXT NOT NULL,
      subject TEXT NOT NULL,
      text_body TEXT NOT NULL,
      scheduled_at TIMESTAMPTZ NOT NULL,
      status TEXT NOT NULL DEFAULT 'scheduled'
        CHECK (status IN (
          'scheduled', 'queued', 'processing', 'sending',
          'rate_limited', 'sent', 'failed', 'uncertain'
        )),
      processing_started_at TIMESTAMPTZ,
      rate_limited_until TIMESTAMPTZ,
      attempts INTEGER NOT NULL DEFAULT 0,
      provider_message_id TEXT,
      preview_url TEXT,
      sent_at TIMESTAMPTZ,
      last_error TEXT,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      UNIQUE (tenant_id, idempotency_key)
    );

    CREATE INDEX IF NOT EXISTS emails_tenant_status_idx
      ON emails (tenant_id, status, scheduled_at);

    CREATE TABLE IF NOT EXISTS slack_installations (
      tenant_id TEXT PRIMARY KEY REFERENCES tenants(id) ON DELETE CASCADE,
      team_id TEXT NOT NULL,
      team_name TEXT NOT NULL,
      access_token TEXT NOT NULL,
      bot_user_id TEXT,
      channel_id TEXT,
      connected_at TIMESTAMPTZ NOT NULL DEFAULT now()
    );

    CREATE TABLE IF NOT EXISTS slack_oauth_states (
      state TEXT PRIMARY KEY,
      tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
      expires_at TIMESTAMPTZ NOT NULL
    );

    INSERT INTO tenants (id, name)
    VALUES ('demo', 'Demo Tenant')
    ON CONFLICT (id) DO NOTHING;
  `);
}

Why use two Slack tables?

The installation table stores one connected workspace per tenant. The state table holds temporary OAuth checks that expire after ten minutes.

Deleting one installation leaves sender records and email processing untouched.

Save src/config.ts.
Save src/db.ts.
Compile the database changes by running:

npm run build

What does this build check?

The build confirms that the expanded configuration remains type-safe. The SQL runs when the API starts later in this step.

Create src/slack.ts inside the existing src folder.
Replace src/server.ts with the final Slack-enabled server in the full-code tab below.

✔️ Awesome, I've got everything!

Confirm that src/slack.ts exports the OAuth helpers. Confirm that src/server.ts exposes all five Slack endpoints.

ⓧ I'd like to double check the full code

Use this final src/slack.ts file for tenant-scoped OAuth.

import { randomUUID } from "node:crypto";
import { config } from "./config.js";
import { pool } from "./db.js";

function requireSlackConfig(): {
  clientId: string;
  clientSecret: string;
  redirectUri: string;
} {
  if (
    !config.slackClientId ||
    !config.slackClientSecret ||
    !config.slackRedirectUri
  ) {
    throw new Error("Slack OAuth environment variables are not configured");
  }
  return {
    clientId: config.slackClientId,
    clientSecret: config.slackClientSecret,
    redirectUri: config.slackRedirectUri,
  };
}

export async function createSlackAuthorizeUrl(
  tenantId: string,
): Promise<string> {
  const slack = requireSlackConfig();
  const state = randomUUID();

  await pool.query(
    `INSERT INTO slack_oauth_states (state, tenant_id, expires_at)
     VALUES ($1, $2, now() + interval '10 minutes')`,
    [state, tenantId],
  );

  const url = new URL("https://slack.com/oauth/v2/authorize");
  url.searchParams.set("client_id", slack.clientId);
  url.searchParams.set("scope", "chat:write");
  url.searchParams.set("redirect_uri", slack.redirectUri);
  url.searchParams.set("state", state);
  return url.toString();
}

type SlackAccessResponse = {
  ok: boolean;
  error?: string;
  access_token?: string;
  bot_user_id?: string;
  team?: { id: string; name: string };
};

export async function completeSlackOAuth(
  code: string,
  state: string,
): Promise<{ tenantId: string; teamName: string }> {
  const slack = requireSlackConfig();
  const stateResult = await pool.query<{ tenant_id: string }>(
    `DELETE FROM slack_oauth_states
     WHERE state = $1 AND expires_at > now()
     RETURNING tenant_id`,
    [state],
  );
  const tenantId = stateResult.rows[0]?.tenant_id;
  if (!tenantId) {
    throw new Error("Slack OAuth state is invalid or expired");
  }

  const body = new URLSearchParams({
    client_id: slack.clientId,
    client_secret: slack.clientSecret,
    code,
    redirect_uri: slack.redirectUri,
  });
  const response = await fetch("https://slack.com/api/oauth.v2.access", {
    method: "POST",
    headers: { "Content-Type": "application/x-www-form-urlencoded" },
    body,
  });
  const data = (await response.json()) as SlackAccessResponse;
  if (!data.ok || !data.access_token || !data.team) {
    throw new Error(data.error ?? "Slack OAuth token exchange failed");
  }

  await pool.query(
    `INSERT INTO slack_installations
       (tenant_id, team_id, team_name, access_token, bot_user_id)
     VALUES ($1, $2, $3, $4, $5)
     ON CONFLICT (tenant_id) DO UPDATE SET
       team_id = EXCLUDED.team_id,
       team_name = EXCLUDED.team_name,
       access_token = EXCLUDED.access_token,
       bot_user_id = EXCLUDED.bot_user_id,
       connected_at = now()`,
    [tenantId, data.team.id, data.team.name, data.access_token, data.bot_user_id],
  );

  return { tenantId, teamName: data.team.name };
}

export async function setSlackChannel(
  tenantId: string,
  channelId: string,
): Promise<void> {
  const result = await pool.query(
    `UPDATE slack_installations SET channel_id = $2 WHERE tenant_id = $1`,
    [tenantId, channelId],
  );
  if (result.rowCount !== 1) {
    throw new Error("Connect Slack before selecting a channel");
  }
}

export async function getSlackStatus(tenantId: string): Promise<object> {
  const result = await pool.query<{
    team_id: string;
    team_name: string;
    channel_id: string | null;
    connected_at: Date;
  }>(
    `SELECT team_id, team_name, channel_id, connected_at
     FROM slack_installations WHERE tenant_id = $1`,
    [tenantId],
  );
  return { connected: result.rowCount === 1, installation: result.rows[0] ?? null };
}

export async function disconnectSlack(tenantId: string): Promise<void> {
  const result = await pool.query<{ access_token: string }>(
    `SELECT access_token FROM slack_installations WHERE tenant_id = $1`,
    [tenantId],
  );
  const token = result.rows[0]?.access_token;
  if (!token) return;

  const response = await fetch("https://slack.com/api/auth.revoke", {
    headers: { Authorization: `Bearer ${token}` },
  });
  const data = (await response.json()) as { ok: boolean; error?: string };
  if (!data.ok) throw new Error(data.error ?? "Slack token revocation failed");

  await pool.query(
    `DELETE FROM slack_installations WHERE tenant_id = $1`,
    [tenantId],
  );
}

export async function sendRateLimitAlert(input: {
  tenantId: string;
  senderEmail: string;
  maximum: number;
}): Promise<boolean> {
  const result = await pool.query<{
    access_token: string;
    channel_id: string | null;
  }>(
    `SELECT access_token, channel_id
     FROM slack_installations WHERE tenant_id = $1`,
    [input.tenantId],
  );
  const installation = result.rows[0];
  if (!installation?.channel_id) return false;

  const response = await fetch("https://slack.com/api/chat.postMessage", {
    method: "POST",
    headers: {
      Authorization: `Bearer ${installation.access_token}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      channel: installation.channel_id,
      text: `Email rate limit reached for ${input.senderEmail}: ${input.maximum} emails per hour. Remaining jobs were delayed.`,
    }),
  });
  const data = (await response.json()) as { ok: boolean; error?: string };
  if (!data.ok) throw new Error(data.error ?? "Slack notification failed");
  return true;
}

How is the Slack connection scoped?

Each authorization state belongs to one tenant.
A valid callback deletes the temporary state before saving the installation.
The installation stores the workspace details with its selected channel.
Disconnect revokes the token before deleting the installation row.

Use this final src/server.ts file to expose the Slack routes.

import { randomUUID } from "node:crypto";
import { createBullBoard } from "@bull-board/api";
import { BullMQAdapter } from "@bull-board/api/bullMQAdapter";
import { ExpressAdapter } from "@bull-board/express";
import express, {
  type NextFunction,
  type Request,
  type Response,
} from "express";
import { config } from "./config.js";
import { initializeDatabase, pool } from "./db.js";
import {
  emailQueue,
  enqueueEmail,
  producerRedis,
  recoverNonFinalEmails,
} from "./queue.js";
import {
  completeSlackOAuth,
  createSlackAuthorizeUrl,
  disconnectSlack,
  getSlackStatus,
  setSlackChannel,
} from "./slack.js";

await initializeDatabase();
const recovered = await recoverNonFinalEmails();

const app = express();
app.use(express.json({ limit: "100kb" }));
app.use((req, res, next) => {
  res.setHeader("Access-Control-Allow-Origin", config.frontendOrigin);
  res.setHeader("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS");
  res.setHeader(
    "Access-Control-Allow-Headers",
    "Content-Type,Idempotency-Key,X-Tenant-Id,Authorization",
  );
  if (req.method === "OPTIONS") {
    res.sendStatus(204);
    return;
  }
  next();
});

function tenantId(req: Request): string {
  const value = req.header("X-Tenant-Id") ?? "demo";
  if (!/^[a-zA-Z0-9_-]{1,64}$/.test(value)) {
    throw new Error("Invalid X-Tenant-Id header");
  }
  return value;
}

function requiredString(value: unknown, field: string): string {
  if (typeof value !== "string" || value.trim() === "") {
    throw new Error(`${field} is required`);
  }
  return value.trim();
}

app.get("/api/health", async (_req, res) => {
  await pool.query("SELECT 1");
  await producerRedis.ping();
  res.json({ ok: true, recovered });
});

app.post("/api/senders", async (req, res) => {
  const tenant = tenantId(req);
  const id = requiredString(req.body.id, "id");
  const name = requiredString(req.body.name, "name");
  const email = requiredString(req.body.email, "email");
  if (!email.includes("@")) throw new Error("email must contain @");

  const result = await pool.query(
    `INSERT INTO senders (id, tenant_id, name, email)
     VALUES ($1, $2, $3, $4)
     ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, email = EXCLUDED.email
     RETURNING *`,
    [id, tenant, name, email],
  );
  res.status(201).json(result.rows[0]);
});

app.post("/api/emails", async (req, res) => {
  const tenant = tenantId(req);
  const idempotencyKey = req.header("Idempotency-Key");
  if (!idempotencyKey) throw new Error("Idempotency-Key header is required");

  const senderId = requiredString(req.body.senderId, "senderId");
  const toEmail = requiredString(req.body.toEmail, "toEmail");
  const subject = requiredString(req.body.subject, "subject");
  const textBody = requiredString(req.body.textBody, "textBody");
  const scheduledAt = new Date(requiredString(req.body.scheduledAt, "scheduledAt"));
  if (Number.isNaN(scheduledAt.getTime())) {
    throw new Error("scheduledAt must be an ISO date-time");
  }

  const sender = await pool.query(
    `SELECT id FROM senders WHERE id = $1 AND tenant_id = $2`,
    [senderId, tenant],
  );
  if (sender.rowCount !== 1) {
    res.status(404).json({ error: "Sender not found" });
    return;
  }

  const id = randomUUID();
  const result = await pool.query<{
    id: string;
    scheduled_at: Date;
    status: string;
  }>(
    `INSERT INTO emails
       (id, tenant_id, sender_id, idempotency_key,
        to_email, subject, text_body, scheduled_at)
     VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
     ON CONFLICT (tenant_id, idempotency_key)
     DO UPDATE SET idempotency_key = EXCLUDED.idempotency_key
     RETURNING id, scheduled_at, status`,
    [
      id,
      tenant,
      senderId,
      idempotencyKey,
      toEmail,
      subject,
      textBody,
      scheduledAt,
    ],
  );
  const email = result.rows[0]!;

  try {
    await enqueueEmail(email.id, new Date(email.scheduled_at));
    await pool.query(
      `UPDATE emails
       SET status = CASE WHEN status = 'scheduled' THEN 'queued' ELSE status END,
           updated_at = now()
       WHERE id = $1`,
      [email.id],
    );
  } catch (error) {
    res.status(503).json({
      error: "Email was saved but Redis enqueue failed; startup recovery will retry",
      emailId: email.id,
    });
    return;
  }

  const saved = await pool.query(`SELECT * FROM emails WHERE id = $1`, [email.id]);
  res.status(202).json(saved.rows[0]);
});

app.get("/api/emails", async (req, res) => {
  const tenant = tenantId(req);
  const status = typeof req.query.status === "string" ? req.query.status : null;
  const values: unknown[] = [tenant];
  let sql = `SELECT * FROM emails WHERE tenant_id = $1`;
  if (status) {
    sql += ` AND status = $2`;
    values.push(status);
  }
  sql += ` ORDER BY scheduled_at DESC LIMIT 200`;
  const result = await pool.query(sql, values);
  res.json(result.rows);
});

app.get("/api/slack/connect", async (req, res) => {
  const url = await createSlackAuthorizeUrl(tenantId(req));
  res.redirect(url);
});

app.get("/api/slack/callback", async (req, res) => {
  const code = requiredString(req.query.code, "code");
  const state = requiredString(req.query.state, "state");
  const installation = await completeSlackOAuth(code, state);
  res.type("html").send(
    `<h1>Slack connected</h1><p>${installation.teamName} is connected for tenant ${installation.tenantId}.</p>`,
  );
});

app.get("/api/slack/status", async (req, res) => {
  res.json(await getSlackStatus(tenantId(req)));
});

app.put("/api/slack/channel", async (req, res) => {
  const channelId = requiredString(req.body.channelId, "channelId");
  await setSlackChannel(tenantId(req), channelId);
  res.json({ ok: true, channelId });
});

app.delete("/api/slack/disconnect", async (req, res) => {
  await disconnectSlack(tenantId(req));
  res.status(204).end();
});

const boardAdapter = new ExpressAdapter();
boardAdapter.setBasePath("/admin/queues");
createBullBoard({
  queues: [new BullMQAdapter(emailQueue)],
  serverAdapter: boardAdapter,
});

function protectBoard(req: Request, res: Response, next: NextFunction): void {
  const expected = `Basic ${Buffer.from(`admin:${config.adminKey}`).toString("base64")}`;
  if (req.header("Authorization") !== expected) {
    res.setHeader("WWW-Authenticate", 'Basic realm="queues"');
    res.status(401).send("Authentication required");
    return;
  }
  next();
}

app.use("/admin/queues", protectBoard, boardAdapter.getRouter());

app.use(
  (error: unknown, _req: Request, res: Response, _next: NextFunction) => {
    const message = error instanceof Error ? error.message : "Unknown error";
    res.status(400).json({ error: message });
  },
);

const server = app.listen(config.port, () => {
  console.log(`API listening on http://localhost:${config.port}`);
  console.log(`Recovered ${recovered} non-final email records`);
});

async function shutdown(): Promise<void> {
  server.close();
  await emailQueue.close();
  await producerRedis.quit();
  await pool.end();
}

process.on("SIGINT", () => void shutdown().then(() => process.exit(0)));
process.on("SIGTERM", () => void shutdown().then(() => process.exit(0)));

What do the Slack routes provide?

The connect route starts authorization for the current tenant.
The callback route validates state before saving the installation.
The status route returns the connected workspace without exposing its token.
The channel route stores the destination for rate-limit alerts.
The disconnect route revokes the token without stopping email processing.

Save src/slack.ts.
Save src/server.ts.
Compile the OAuth changes by running:

npm run build

What does this build prove?

The build checks the Slack response types. It also confirms that the server imports each OAuth helper correctly.

The build should finish without errors. Your API now has the complete Slack connection contract.

Slack code not compiling?

Confirm that src/slack.ts exports every function imported by src/server.ts.

Check the callback route for the exact code and state query names.

Share the compiler output for focused help: Help me debug my Slack OAuth build error.

Connect Slack and trigger the alert

Slack requires an HTTPS OAuth callback. A Cloudflare Quick Tunnel supplies a temporary public address for the local API.

Quick Tunnels are intended for testing. Keep the tunnel process running while you complete the OAuth flow.

Check whether cloudflared is available by running:

cloudflared --version

What does this command check?

The command checks whether the Cloudflare Tunnel client is available in your terminal. Choose the tab that matches the result.

✔️ I see a version number

Cloudflared is ready. Continue to the tunnel command below.

ⓧ Command not found on Debian or Ubuntu

Cloudflare provides an APT repository for Debian-based Linux systems.

Add the Cloudflare package signing key by running:

sudo mkdir -p --mode=0755 /usr/share/keyrings
curl -fsSL https://pkg.cloudflare.com/cloudflare-main.gpg | sudo tee /usr/share/keyrings/cloudflare-main.gpg >/dev/null

What does this setup do?

These commands create the keyring directory. They then save Cloudflare's package signing key there.

Add the Cloudflare APT repository by running:

echo "deb [signed-by=/usr/share/keyrings/cloudflare-main.gpg] https://pkg.cloudflare.com/cloudflared any main" | sudo tee /etc/apt/sources.list.d/cloudflared.list

What does this repository provide?

The repository gives APT an official source for the Cloudflare Tunnel package.

Install Cloudflared by running:

sudo apt-get update && sudo apt-get install cloudflared

What happens during installation?

APT refreshes its package index before installing Cloudflared. Accept the package prompt if your system asks for confirmation.

ⓧ Command not found on another Linux distribution

Cloudflare provides Linux binaries plus package-specific installation options.

Follow the official Cloudflared downloads guide for your Linux distribution.
Return here after cloudflared --version prints a version number.

Start the Quick Tunnel in a separate terminal by running:

cloudflared tunnel --url http://localhost:3000

What does the tunnel do?

Cloudflared prints a random public trycloudflare.com URL. Requests to that address are forwarded to your API on port 3000.

The command keeps control of this terminal. Leave it running for the rest of the step.

Record the complete callback URL here: https://your-quick-tunnel.trycloudflare.com/api/slack/callback.

The next actions handle credentials. Keep the client secret inside .env so it stays outside version control.

Open the Slack app dashboard in your browser.
Create a Blank app for your development workspace.
Enter ReachInbox Scheduler as the app name.
Open OAuth & Permissions in the app sidebar.
Add chat:write under Bot Token Scopes.
Add https://your-quick-tunnel.trycloudflare.com/api/slack/callback under Redirect URLs.
Copy the app's client ID from its credentials area.
Paste the client ID as the value of SLACK_CLIENT_ID in .env.
Copy the app's client secret from its credentials area.
Paste the client secret as the value of SLACK_CLIENT_SECRET in .env.
Set SLACK_REDIRECT_URI in .env to https://your-quick-tunnel.trycloudflare.com/api/slack/callback.
Set MAX_EMAILS_PER_HOUR_PER_SENDER to 2 in .env.
Save .env.

Keep the callback identical

The Slack redirect URL must match SLACK_REDIRECT_URI exactly. A missing callback path prevents the authorization code from reaching your API.

Return to the API terminal from earlier.
Stop the running API with Ctrl+C.
Restart the API with the Slack configuration by running:

npm run dev:api

What should the API print?

The API should print its local address. It should also report how many non-final email records were recovered.

Leave this terminal running so Slack can reach the callback through the tunnel.

Return to the worker terminal from earlier.
Stop the running worker with Ctrl+C.
Restart the limiter-aware worker by running:

npm run dev:worker

What should the worker print?

The worker should report its configured concurrency. Leave this process running while you trigger the burst.

Open http://localhost:3000/api/slack/connect in your browser.
Select your development workspace on Slack's authorization screen.
Approve the requested permission with Allow.

You should see a Slack connected page naming the workspace. That confirms the state check passed before the installation was stored.

Invite the Slack app to the channel that should receive alerts.
Copy that channel's ID from its channel details.
Record the channel ID here: your Slack channel ID.
Store the channel for tenant demo by running:

curl -X PUT http://localhost:3000/api/slack/channel \
  -H "Content-Type: application/json" \
  -H "X-Tenant-Id: demo" \
  -d '{"channelId":"[[SLACK_CHANNEL_ID="your Slack channel ID"]]"}'

What does this request change?

The request stores the destination channel on tenant demo's Slack installation. The response should contain ok set to true.

Channel setup failing?

Confirm that OAuth completed before you send the channel request.

Check that the app is a member of the destination channel.

Get help with the response body: Help me fix my Slack channel configuration.

Enter the sender ID you created earlier here: sender ID from Step 2.
Enter the current ISO date-time here: current ISO date-time.

Before you run the burst, do you expect all three jobs to reach SMTP immediately?

Schedule three immediate emails with distinct idempotency keys by running:

curl -X POST http://localhost:3000/api/emails \
  -H "Content-Type: application/json" -H "X-Tenant-Id: demo" -H "Idempotency-Key: limit-test-1" \
  -d '{"senderId":"[[SENDER_ID="sender ID from Step 2"]]","toEmail":"test@example.com","subject":"Limit test 1","textBody":"First immediate email","scheduledAt":"[[SCHEDULED_AT="current ISO date-time"]]"}'
curl -X POST http://localhost:3000/api/emails \
  -H "Content-Type: application/json" -H "X-Tenant-Id: demo" -H "Idempotency-Key: limit-test-2" \
  -d '{"senderId":"[[SENDER_ID="sender ID from Step 2"]]","toEmail":"test@example.com","subject":"Limit test 2","textBody":"Second immediate email","scheduledAt":"[[SCHEDULED_AT="current ISO date-time"]]"}'
curl -X POST http://localhost:3000/api/emails \
  -H "Content-Type: application/json" -H "X-Tenant-Id: demo" -H "Idempotency-Key: limit-test-3" \
  -d '{"senderId":"[[SENDER_ID="sender ID from Step 2"]]","toEmail":"test@example.com","subject":"Limit test 3","textBody":"Third immediate email","scheduledAt":"[[SCHEDULED_AT="current ISO date-time"]]"}'

What does this burst test?

Each request creates one email with a distinct idempotency key. All three jobs compete for the same sender's hourly quota.

Two jobs can reserve hourly slots. The spacing key keeps their sends at least MIN_SEND_DELAY_MS apart.

Wait five seconds for the worker to process the burst.
List the rate-limited email records by running:

curl "http://localhost:3000/api/emails?status=rate_limited" -H "X-Tenant-Id: demo"

What should you see?

The response should contain the third email with status rate_limited. Its rate_limited_until value points to the next retry time.

Your Slack channel should contain one alert for this sender-hour. The alert confirms that the shared quota reached 2.

No rate-limited record or Slack alert?

Confirm that MAX_EMAILS_PER_HOUR_PER_SENDER is set to 2 in .env.

Confirm that the API and worker were restarted after the environment change.

Check the worker terminal for a Slack alert failure without sharing the stored token: Help me debug my missing rate-limit alert.

You have closed the parallel-worker gap. The scheduler now defers excess jobs safely while operators receive one actionable Slack alert per sender-hour.

Secret mission

Run the Burst and Recovery Drill

Stress the scheduler with five immediate emails from one sender. Watch excess jobs defer after the hourly quota is reached. Restart the API and worker while a future job is waiting to prove that delivery resumes exactly once.

Clean Up Your Resources

This demo path has no ongoing cost for local development or testing. Decide whether to keep the processes active, pause them while retaining data, or delete the resources entirely.

Resources you used:

The postgres service running as a Docker container for PostgreSQL.
The redis service running as a container for Redis.
The postgres_data named Docker volume.
The redis_data named Docker volume.
The tenant-scoped Slack OAuth installation for tenant demo.
The temporary Cloudflare Quick Tunnel process.
The local project folder identified by its package.json file.

Keep everything running

No action needed. Choose this if you want to keep the complete demo ready for another walkthrough.

Leave the API process running.
Leave the worker process running.
Leave the Quick Tunnel process running.
Keep the postgres service running.
Keep the redis service running.
Keep the Slack installation connected.

Keep the Tunnel for Testing

Quick Tunnels are intended for testing. They do not have a production SLA.

Pause - I'll come back to this later

Shut down the running processes to free local resources. Your PostgreSQL data and Redis queue state remain in the named volumes.

Press Ctrl+C in the API terminal.
Press Ctrl+C in the worker terminal.
Press Ctrl+C in the Quick Tunnel terminal.
Return to the terminal whose current directory contains compose.yaml.
Pause the Docker services by running this command:

docker compose down

What Does This Command Do?

This stops and removes the postgres container.

It also removes the redis container. The named volumes remain available for the next start.

Wait for Docker Compose to return control to the terminal.

You should see Docker Compose report that the containers were removed. Your scheduler is safely paused with its persistent data retained.

Docker Services Still Running?

Check that the terminal is inside the folder containing compose.yaml.

If Docker reports a permissions problem, use the same Linux shell configuration that worked when you started the stack.

Help me pause this Docker Compose stack without deleting its named volumes.

Delete - I don't want to use this again

Remove the project resources and start fresh. This cleanup permanently deletes the PostgreSQL and Redis volumes, but your demo recording already preserves the evidence you built.

Send a DELETE request to /api/slack/disconnect with the X-Tenant-Id: demo header from the API client you used earlier.

You should receive an empty 204 response. That response confirms the Slack disconnect flow finished.

Press Ctrl+C in the API terminal.
Press Ctrl+C in the worker terminal.
Press Ctrl+C in the Quick Tunnel terminal.
Return to the terminal whose current directory contains compose.yaml.
Delete the Docker services and named volumes by running this command:

docker compose down -v

What Does This Command Delete?

The -v flag removes the named volumes with the containers. This permanently deletes the PostgreSQL records and Redis AOF data.

Wait for Docker Compose to return control to the terminal.

You should see Docker Compose report volume removal. The persistent scheduler data is now deleted.

Volumes Not Removed?

Check that you included -v in the command. Running the command without that flag preserves the named volumes.

Help me confirm that this Docker Compose project removed its named volumes.

The final local resource is the folder identified by package.json. The next command records that folder before removing it from its parent directory.

Remove the local project folder by running these commands from inside it:

project_dir="$(pwd)"
cd ..
rm -rf "$project_dir"

What Do These Commands Do?

The project_dir variable stores the current folder path. The next command moves to its parent folder.

The final command permanently removes the stored project folder path. This includes the source files and local dependency folders inside it.

Use your Linux file manager to confirm that the folder containing package.json is gone.

Your local containers, persistent data, tunnel process, Slack connection, and project files are now cleaned up.

Project Folder Still Present?

Check that you ran the commands from the folder containing this project's package.json file.

Help me safely remove the local scheduler project folder from Linux.

Nice Work!

You pulled it off! Your restart-safe email scheduler now handles delayed delivery under real operational pressure.

You've learned how to:

Build a frontend-ready REST API with TypeScript plus Express. Create senders through stable endpoints. Schedule emails with idempotency keys. List each delivery status.
Deliver delayed jobs through BullMQ with Ethereal preview links. Reconcile PostgreSQL records with Redis queue state after a restart. Quarantine ambiguous SMTP outcomes as uncertain for manual review.
Protect parallel delivery with per-sender quotas. Enforce minimum send spacing across workers with an atomic Redis limiter. Inspect live queue state in Bull Board. Send one Slack alert for each sender-hour that reaches its quota.
Complete the Secret Mission burst and recovery drill. Prove that a five-email burst defers excess jobs after two sends. Show one Slack alert for the sender-hour. Confirm that a future delayed job reaches sent once after both Node processes restart.

Ready to quiz yourself?