feat: add prefix method to publisher, rate limiter, and lock adapters

apps/content/docs/helpers/lock.mdx

@@ -183,6 +183,9 @@ const locker = new RedisLocker(client, {
    */
   retryInterval: 100,
 })
+
+// Creates a locker with an extra prefix that shares this client, e.g. per tenant
+const tenantLocker = locker.prefix('tenant:123:')
 ```
 
 ### Upstash
@@ -225,6 +228,9 @@ const locker = new UpstashLocker(redis, {
    */
   retryInterval: 100,
 })
+
+// Creates a locker with an extra prefix that shares this client, e.g. per tenant
+const tenantLocker = locker.prefix('tenant:123:')
 ```
 
 ### Bun Redis
@@ -265,6 +271,9 @@ const locker = new BunRedisLocker(redis, {
    */
   retryInterval: 100,
 })
+
+// Creates a locker with an extra prefix that shares this client, e.g. per tenant
+const tenantLocker = locker.prefix('tenant:123:')
 ```
 
 ### Cloudflare
@@ -316,6 +325,9 @@ export default {
        */
       getStubByName: (namespace, key) => namespace.getByName(key),
     })
+
+    // Creates a locker with an extra prefix that shares this namespace, e.g. per tenant
+    const tenantLocker = locker.prefix('tenant:123:')
   },
 }
 ```

apps/content/docs/helpers/publisher.mdx

@@ -150,6 +150,9 @@ const publisher = new RedisPublisher<Events>(client, {
     ttl: 300_000
   }
 })
+
+// Creates a publisher with an extra prefix that shares this client, e.g. per tenant
+const tenantPublisher = publisher.prefix('tenant:123:')
 ```
 
 ```ts upstash
@@ -196,6 +199,9 @@ const publisher = new UpstashPublisher<Events>(redis, {
     ttl: 300_000
   }
 })
+
+// Creates a publisher with an extra prefix that shares this client, e.g. per tenant
+const tenantPublisher = publisher.prefix('tenant:123:')
 ```
 
 ```ts bun
@@ -253,6 +259,9 @@ const publisher = new BunRedisPublisher<Events>(redis, {
     ttl: 300_000
   }
 })
+
+// Creates a publisher with an extra prefix that shares this client, e.g. per tenant
+const tenantPublisher = publisher.prefix('tenant:123:')
 ```
 
 ```ts cloudflare
@@ -331,6 +340,9 @@ export default {
        */
       getStubByName: (namespace, event) => namespace.getByName(event)
     })
+
+    // Creates a publisher with an extra prefix that shares this namespace, e.g. per tenant
+    const tenantPublisher = publisher.prefix('tenant:123:')
   },
 }
 ```

apps/content/docs/helpers/ratelimit.mdx

@@ -216,6 +216,9 @@ const limiter = new RedisRateLimiter(client, {
    */
   timeout: 0,
 })
+
+// Creates a limiter with an extra prefix that shares this client, e.g. per tenant
+const tenantLimiter = limiter.prefix('tenant:123:')
 ```
 
 ### Upstash
@@ -254,6 +257,9 @@ const limiter = new UpstashRateLimiter(redis, {
    */
   timeout: 0,
 })
+
+// Creates a limiter with an extra prefix that shares this client, e.g. per tenant
+const tenantLimiter = limiter.prefix('tenant:123:')
 ```
 
 ### Bun Redis
@@ -290,6 +296,9 @@ const limiter = new BunRedisRateLimiter(redis, {
    */
   timeout: 0,
 })
+
+// Creates a limiter with an extra prefix that shares this client, e.g. per tenant
+const tenantLimiter = limiter.prefix('tenant:123:')
 ```
 
 ### Cloudflare
@@ -309,6 +318,9 @@ export default {
        */
       prefix: ''
     })
+
+    // Creates a limiter with an extra prefix that shares this binding, e.g. per tenant
+    const tenantLimiter = limiter.prefix('tenant:123:')
   }
 }
 ```

packages/bun/src/redis-lock.test.ts

@@ -95,6 +95,16 @@ describe.skipIf(!REDIS_URL)('bun redis locker integration', async () => {
     await holder.release()
   }, { timeout: 20_000 })
 
+  it('shares locks with lockers created by prefix', async () => {
+    const { locker } = createTestingLocker()
+    const holder = await hold(locker, 'tenant:key')
+
+    await expect(locker.prefix('tenant:').lock('key', () => 'never', { timeout: 0 })).rejects.toBeInstanceOf(LockTimeoutError)
+    await expect(locker.prefix('other:').lock('key', () => 'ok', { timeout: 0 })).resolves.toBe('ok')
+
+    await holder.release()
+  })
+
   it('releases the lock when the callback throws', async () => {
     const { locker } = createTestingLocker()
 

packages/bun/src/redis-lock.ts

@@ -11,11 +11,21 @@ import { BaseRedisLocker } from '@orpc/experimental-lock/base-redis'
 export class experimental_BunRedisLocker extends BaseRedisLocker {
   constructor(
     private readonly redis: RedisClient,
-    options: BaseRedisLockerOptions,
+    private readonly options: BaseRedisLockerOptions,
   ) {
     super(options)
   }
 
+  /**
+   * Creates a locker with `prefix` appended to this locker's prefix, sharing its Redis client.
+   *
+   * @see {@link https://orpc.dev/docs/helpers/lock#adapters | Lock Helpers - Adapters}
+   */
+  prefix(prefix: string): experimental_BunRedisLocker {
+    // eslint-disable-next-line new-cap
+    return new experimental_BunRedisLocker(this.redis, { ...this.options, prefix: `${this.keyPrefix}${prefix}` })
+  }
+
   protected async acquire(key: string, token: string, ttl: number): Promise<boolean> {
     const result = await this.redis.send('SET', [key, token, 'NX', 'PX', String(ttl)])
 

packages/bun/src/redis-publisher.test.ts

@@ -81,6 +81,35 @@ describe.skipIf(!REDIS_URL)('bun redis publisher', () => {
     await unsubscribeResumeAttempt()
   }, { timeout: 20_000 })
 
+  it('shares the subscriber with publishers created by prefix', async () => {
+    const publisher = createTestingPublisher()
+    const tenant = publisher.prefix('tenant:')
+    const event = crypto.randomUUID()
+    const listener = vi.fn()
+    const onError = vi.fn()
+
+    const unsubscribe = await tenant.subscribe(event, listener, { onError })
+
+    await publisher.publish(`tenant:${event}`, { order: 1 })
+    await publisher.prefix('other:').publish(event, { order: 2 })
+    await tenant.publish(event, { order: 3 })
+
+    await waitFor(() => {
+      expect(listener).toHaveBeenCalledTimes(2)
+    })
+    expect(listener).toHaveBeenNthCalledWith(1, { order: 1 })
+    expect(listener).toHaveBeenNthCalledWith(2, { order: 3 })
+
+    const subscriber: RedisClient = await (publisher as any).getSubscriber()
+    expect(await (tenant as any).getSubscriber()).toBe(subscriber)
+
+    const error = new Error('boom')
+    subscriber.onclose!.call(subscriber, error)
+    expect(onError).toHaveBeenCalledWith(error)
+
+    await unsubscribe()
+  }, { timeout: 20_000 })
+
   it('resumes missed events in order and preserves event metadata while rewriting ids', async () => {
     const publisher = createTestingPublisher({
       resume: { enabled: true, ttl: 10_000 },

packages/bun/src/redis-publisher.ts

@@ -26,23 +26,38 @@ export interface BunRedisPublisherOptions extends BaseRedisPublisherOptions {
  * @see {@link https://orpc.dev/docs/helpers/publisher#adapters | Publisher Helpers - Adapters}
  */
 export class BunRedisPublisher<T extends Record<string, object>> extends BaseRedisPublisher<T> {
-  private readonly getSubscriber: () => Promise<RedisClient>
+  private getSubscriber: () => Promise<RedisClient>
 
-  private readonly onErrors: Array<(error: ThrowableError) => void> = []
+  private onErrors: Array<(error: ThrowableError) => void> = []
 
   constructor(
     private readonly redis: RedisClient,
-    { subscriber, ...options }: BunRedisPublisherOptions = {},
+    private readonly options: BunRedisPublisherOptions = {},
   ) {
     super(options)
 
     this.getSubscriber = once(async () => {
-      const client = await (subscriber ?? redis.duplicate())
+      const client = await (options.subscriber ?? redis.duplicate())
       this.watchSubscriber(client)
       return client
     })
   }
 
+  /**
+   * Creates a publisher with `prefix` appended to this publisher's prefix, sharing its Redis client and subscriber.
+   *
+   * @see {@link https://orpc.dev/docs/helpers/publisher#adapters | Publisher Helpers - Adapters}
+   */
+  prefix(prefix: string): BunRedisPublisher<T> {
+    const publisher = new BunRedisPublisher<T>(this.redis, { ...this.options, prefix: `${this.eventPrefix}${prefix}` })
+
+    // Share the lazily created subscriber, which reports its errors to `onErrors` of this publisher.
+    publisher.getSubscriber = this.getSubscriber
+    publisher.onErrors = this.onErrors
+
+    return publisher
+  }
+
   protected async publishMessage(channel: string, message: string): Promise<void> {
     await this.redis.publish(channel, message)
   }

packages/bun/src/redis-ratelimit.test.ts

@@ -64,6 +64,14 @@ describe.skipIf(!REDIS_URL)('bun redis rate limiter integration', async () => {
     })
   }, { timeout: 20_000 })
 
+  it('shares counters with limiters created by prefix', async () => {
+    const limiter = await createTestingRateLimiter({ maxRequests: 1, window: 10_000 })
+
+    await expect(limiter.limit('tenant:key')).resolves.toMatchObject({ success: true, remaining: 0 })
+    await expect(limiter.prefix('tenant:').limit('key')).resolves.toMatchObject({ success: false, remaining: 0 })
+    await expect(limiter.prefix('other:').limit('key')).resolves.toMatchObject({ success: true, remaining: 0 })
+  }, { timeout: 20_000 })
+
   it('returns a stable reset timestamp within the same window', async () => {
     const window = 2000
     const limiter = await createTestingRateLimiter({

packages/bun/src/redis-ratelimit.ts

@@ -13,11 +13,20 @@ export interface BunRedisRateLimiterOptions extends BaseRedisRateLimiterOptions
 export class BunRedisRateLimiter extends BaseRedisRateLimiter {
   constructor(
     private readonly redis: RedisClient,
-    options: BunRedisRateLimiterOptions,
+    private readonly options: BunRedisRateLimiterOptions,
   ) {
     super(options)
   }
 
+  /**
+   * Creates a limiter with `prefix` appended to this limiter's prefix, sharing its Redis client.
+   *
+   * @see {@link https://orpc.dev/docs/helpers/ratelimit#adapters | Rate Limit Helpers - Adapters}
+   */
+  prefix(prefix: string): BunRedisRateLimiter {
+    return new BunRedisRateLimiter(this.redis, { ...this.options, prefix: `${this.keyPrefix}${prefix}` })
+  }
+
   protected async evalScript(script: string, keys: string[], args: string[]): Promise<unknown> {
     return await this.redis.send('EVAL', [script, String(keys.length), ...keys, ...args])
   }

packages/cloudflare/src/lock.test.ts

@@ -93,6 +93,16 @@ describe('durableLocker', () => {
     await expect(other.lock('key', () => 'ok')).resolves.toBe('ok')
   })
 
+  it('shares locks with lockers created by prefix', async () => {
+    const { locker } = createTestingLocker()
+    const holder = await hold(locker, 'tenant:key')
+
+    await expect(locker.prefix('tenant:').lock('key', () => 'never', { timeout: 0 })).rejects.toBeInstanceOf(LockTimeoutError)
+    await expect(locker.prefix('other:').lock('key', () => 'ok', { timeout: 0 })).resolves.toBe('ok')
+
+    await holder.release()
+  })
+
   it('tracks locks independently per key', async () => {
     const { locker } = createTestingLocker()
     const holder = await hold(locker, 'alice')

packages/cloudflare/src/lock.ts

@@ -42,25 +42,35 @@ export interface experimental_DurableLockerOptions {
  * @see {@link https://orpc.dev/docs/helpers/lock#adapters | Lock Helpers - Adapters}
  */
 export class experimental_DurableLocker implements Locker {
-  private readonly prefix: string
+  private readonly keyPrefix: string
   private readonly ttl: number
   private readonly timeout: number
   private readonly getStubByName: Exclude<experimental_DurableLockerOptions['getStubByName'], undefined>
 
   constructor(
     private readonly namespace: DurableObjectNamespace<any>,
-    options: experimental_DurableLockerOptions,
+    private readonly options: experimental_DurableLockerOptions,
   ) {
-    this.prefix = options.prefix ?? ''
+    this.keyPrefix = options.prefix ?? ''
     this.ttl = options.ttl
     this.timeout = options.timeout ?? 10_000
     this.getStubByName = options.getStubByName ?? ((namespace, key) => namespace.getByName(key))
   }
 
+  /**
+   * Creates a locker with `prefix` appended to this locker's prefix, sharing its namespace.
+   *
+   * @see {@link https://orpc.dev/docs/helpers/lock#adapters | Lock Helpers - Adapters}
+   */
+  prefix(prefix: string): experimental_DurableLocker {
+    // eslint-disable-next-line new-cap
+    return new experimental_DurableLocker(this.namespace, { ...this.options, prefix: `${this.keyPrefix}${prefix}` })
+  }
+
   async lock<T>(key: string, fn: (options: LockCallbackOptions) => Promisable<T>, options: LockOptions = {}): Promise<T> {
     throwIfAborted(options.signal)
 
-    const response = await this.getStubByName(this.namespace, `${this.prefix}${key}`).fetch('http://localhost/acquire', {
+    const response = await this.getStubByName(this.namespace, `${this.keyPrefix}${key}`).fetch('http://localhost/acquire', {
       headers: { upgrade: 'websocket' },
     })
 

packages/cloudflare/src/publisher.test.ts

@@ -264,6 +264,24 @@ describe('durable publisher', () => {
     await unsubscribe()
   })
 
+  it('delivers events to publishers created by prefix', async () => {
+    const { publisher } = createTestingPublisher(env.PUBLISHER_DON)
+    const listener = vi.fn()
+
+    const unsubscribe = await publisher.prefix('tenant:').subscribe('message', listener)
+
+    await publisher.publish('tenant:message', { text: 'original' })
+    await publisher.prefix('tenant:').publish('message', { text: 'prefixed' })
+
+    await vi.waitFor(() => {
+      expect(listener).toHaveBeenCalledTimes(2)
+    })
+    expect(listener).toHaveBeenNthCalledWith(1, { text: 'original' })
+    expect(listener).toHaveBeenNthCalledWith(2, { text: 'prefixed' })
+
+    await unsubscribe()
+  })
+
   it('throws when publish fails', async () => {
     const stub = {
       fetch: vi.fn(async () => new Response(null, {