effect-mq with Upstash Redis - Upstash Documentation

You can use effect-mq with Upstash Redis. effect-mq is a background job library for Effect: you define a job once with a schema-typed payload, enqueue it from anywhere, and run it in a worker. Its Redis store (effect-mq/redis) keeps every operation atomic with a single Lua script, so you can use Upstash Redis as its storage.

Install

npm install effect-mq effect@rc @effect/platform-node@rc redis

effect-mq targets Effect v4, which is currently published under the rc tag; @effect/platform-node@latest still targets Effect v3, so install it with the rc tag as well. The Redis store does not bundle a client; on Node.js it uses node-redis through @effect/platform-node. On Bun, install @effect/platform-bun@rc instead and use BunRedis (no extra client needed).

Usage

First, get your connection string from the TCP tab of the Connect section on your database page in the Upstash Console:

Then create the Redis store layer with that URL. It must start with rediss:// since Upstash requires TLS; node-redis picks up TLS from the scheme automatically, so no extra options are needed:

import { RedisJobStore } from "effect-mq/redis"
import { NodeRedis } from "@effect/platform-node"
import { Layer } from "effect"

const StoreLive = RedisJobStore.layer({
  prefix: "myapp-jobs", // key namespace (default "effect-mq")
  historyTtl: "7 days"  // optional retention ceiling for finished jobs
}).pipe(
  Layer.provide(
    NodeRedis.layer({
      url: "rediss://default:UPSTASH_REDIS_PASSWORD@UPSTASH_REDIS_ENDPOINT:6379"
    })
  )
)

Define a job, enqueue it, and run a worker:

import { Job, Worker } from "effect-mq"
import { Effect, Layer, Schema } from "effect"

// Define the job once; producers and workers share this definition
class SendEmail extends Job.make("SendEmail", {
  payload: { to: Schema.String, subject: Schema.String },
  success: Schema.String,
  queue: "email",
  defaults: {
    attempts: 3,
    backoff: { type: "exponential", delay: "1 second" }
  }
}) {}

// Enqueue from anywhere that has the store in context
const producer = Effect.gen(function* () {
  const jobId = yield* SendEmail.enqueue({ to: "ada@example.com", subject: "hi" })

  // ...or enqueue and await the typed result
  const messageId = yield* SendEmail.execute(
    { to: "grace@example.com", subject: "now" },
    { delay: "5 seconds", priority: 2 }
  )
})

// Run a worker
const RunnerLive = SendEmail.toLayer(
  (payload) => Effect.succeed(`sent to ${payload.to}`),
  { concurrency: 5 }
).pipe(
  Layer.provideMerge(Worker.layer()),
  Layer.provideMerge(StoreLive)
)

Provide RunnerLive to your app and the worker claims, runs, retries, and records jobs. The producer only needs StoreLive, so your API server can enqueue without depending on any handler code.

Connecting over HTTP

The setup above talks to Upstash over a TCP connection, which suits a long-running worker process. You may prefer to connect over HTTP with @upstash/redis instead when:

You enqueue jobs from serverless or edge functions. Vercel Functions, Cloudflare Workers, Lambda, and similar runtimes either have no TCP sockets or would open a fresh TLS connection on every cold start. @upstash/redis uses fetch, so it works anywhere and has no connection to manage.
You run many short-lived instances. Each TCP client holds a connection open (two for the worker, one of them for pub/sub), which adds up across many serverless invocations or autoscaled replicas. HTTP requests are stateless, so there is no connection count to watch.

effect-mq does not need a separate storage driver for this. Its Redis store talks to Effect’s client-agnostic Redis service, which only requires a send function for raw commands and a subscribe function for pub/sub wake-ups. @upstash/redis provides both (exec and subscribe), so the same store, Lua scripts, and key layout work unchanged. Install the client:

npm install @upstash/redis

Add the following layer, which provides the Redis service over the Upstash REST API:

upstash-redis.ts

import { Redis as Upstash, type RedisConfigNodejs } from "@upstash/redis"
import { Redis } from "effect/unstable/persistence"
import { Deferred, Effect, Layer } from "effect"

const make = (options: Omit<RedisConfigNodejs, "automaticDeserialization">) =>
  Effect.gen(function* () {
    // The store reads back the JSON strings it wrote, so replies must not be auto-parsed
    const client = new Upstash({ ...options, automaticDeserialization: false })
    const fail = (cause: unknown) => new Redis.RedisError({ cause })

    return yield* Redis.make({
      send: <A = unknown>(command: string, ...args: ReadonlyArray<string>) =>
        Effect.tryPromise({ try: () => client.exec<A>([command, ...args]), catch: fail }),

      subscribe: (channel, onMessage) =>
        Effect.gen(function* () {
          const terminal = yield* Deferred.make<void, Redis.RedisError>()
          yield* Effect.acquireRelease(
            Effect.try({
              try: () => {
                const subscriber = client.subscribe<string>(channel)
                subscriber.on("message", (event) =>
                  onMessage({ channel: event.channel, message: String(event.message) })
                )
                subscriber.on("error", (error) => Effect.runSync(Deferred.fail(terminal, fail(error))))
                return subscriber
              },
              catch: fail
            }),
            (subscriber) => Effect.promise(() => subscriber.unsubscribe().catch(() => {}))
          )
          return Deferred.await(terminal)
        })
    })
  })

export const UpstashRedis = {
  /** Provides Effect's `Redis` service over the Upstash REST API. */
  layer: (options: Omit<RedisConfigNodejs, "automaticDeserialization">) =>
    Layer.effect(Redis.Redis, make(options))
}

Then provide it to the store in place of NodeRedis, using the REST URL and token from the REST tab of the Connect section on your database page:

import { RedisJobStore } from "effect-mq/redis"
import { Layer } from "effect"
import { UpstashRedis } from "./upstash-redis"

const StoreLive = RedisJobStore.layer({ prefix: "myapp-jobs" }).pipe(
  Layer.provide(
    UpstashRedis.layer({
      url: "UPSTASH_REDIS_REST_URL",
      token: "UPSTASH_REDIS_REST_TOKEN"
    })
  )
)

Everything else on this page stays the same: jobs, enqueue, execute, and workers all work over HTTP, including pub/sub wake-ups so idle workers pick up new jobs immediately instead of waiting for pollInterval. Because both clients run the same Lua scripts against the same keys, you can mix them under one prefix. A common split is to enqueue over HTTP from serverless functions and run the worker over TCP on a long-running process.

Billing Optimization

effect-mq workers poll Redis regularly (pollInterval), heartbeat locks on active jobs, and sweep for stalled jobs and history, even when there is no queue activity. This can incur extra costs because Upstash charges per request on the Pay-As-You-Go plan. With the introduction of our Fixed plans, we recommend switching to a Fixed plan to avoid increased command count and high costs in effect-mq use cases.

​Install

​Usage

​Connecting over HTTP

​Billing Optimization

Install

Usage

Connecting over HTTP

Billing Optimization