feat(message-bus): match arguments with matchers

packages/message-bus/src/Broker.ts

@@ -1,4 +1,3 @@
-import { InvokerRegistrationArgs } from '@plugola/invoke'
 import type MessageBus from './MessageBus.js'
 import SubscriptionDisposer from './SubscriptionDisposer.js'
 import type {
@@ -8,7 +7,10 @@ import type {
   UntilRtn,
 } from './types/events.js'
 import type { EventGeneratorArgs } from './types/generators.js'
-import type { InvokerInterceptorArgs } from './types/invokables.js'
+import type {
+  InvokerInterceptorArgs,
+  MatchableInvokerRegistrationArgs,
+} from './types/invokables.js'
 import {
   ErrorHandler,
   MessageBusContext,
@@ -183,7 +185,7 @@ export default class Broker<$ extends MessageBusContext = MessageBusContext> {
 
   register<InvokableName extends keyof $['invokables']>(
     invokableName: InvokableName,
-    ...args: InvokerRegistrationArgs<
+    ...args: MatchableInvokerRegistrationArgs<
       $['invokables'][InvokableName]['args'],
       $['invokables'][InvokableName]['return']
     >

packages/message-bus/src/MessageBus.ts

@@ -23,6 +23,7 @@ import {
   InvokerInterceptorArgs,
   InvokerInterceptors,
   Invokers,
+  MatchableInvokerRegistrationArgs,
 } from './types/invokables.js'
 import { Stringable, UnpackResolvableValue } from './types/util.js'
 import {
@@ -33,7 +34,8 @@ import {
 } from './types/MessageBus.js'
 import { anySignal, fromSignal } from './AbortController.js'
 import { InvokableNotRegisteredError } from './errors/InvokableNotRegisteredError.js'
-import { InvokerFn, InvokerRegistrationArgs } from '@plugola/invoke'
+import { InvokerFn } from '@plugola/invoke'
+import { match } from './matcher.js'
 import { StreamReader, StreamReaderArgs, Streams } from './types/streams.js'
 import { WritableReadablePair } from '@johngw/stream/transformers/WritableReadablePair'
 import { mergeUnderlyingSource } from '@johngw/stream'
@@ -341,7 +343,7 @@ export default class MessageBus<
   register<InvokableName extends keyof $['invokables']>(
     broker: Broker<$>,
     invokableName: InvokableName,
-    allArgs: InvokerRegistrationArgs<
+    allArgs: MatchableInvokerRegistrationArgs<
       $['invokables'][InvokableName]['args'],
       $['invokables'][InvokableName]['return']
     >,
@@ -595,7 +597,7 @@ export default class MessageBus<
     else if (args1.length > args2.length) return -1
 
     let i = 0
-    for (; i < args1.length; i++) if (args1[i] !== args2[i]) return -1
+    for (; i < args1.length; i++) if (!match(args1[i], args2[i])) return -1
     return i
   }
 

packages/message-bus/src/index.ts

@@ -2,6 +2,7 @@ export { AbortError } from '@johngw/async'
 export { default as Broker } from './Broker.js'
 export { CancelEvent } from './symbols.js'
 export { default as MessageBus } from './MessageBus.js'
+export * from './matcher.js'
 export * from './types/MessageBus.js'
 export * from './types/events.js'
 export * from './types/generators.js'

packages/message-bus/src/matcher.ts

@@ -0,0 +1,86 @@
+/**
+ * A {@link Matcher} stands in for a literal argument in a filter prefix. When
+ * one sits in a registered argument position, the bus runs its match method
+ * against the emitted value instead of comparing for strict equality.
+ *
+ * The contract is a single symbol-keyed method, so anything — a plain object, a
+ * class instance, a larger value that wants to double as a matcher — becomes a
+ * matcher simply by implementing it.
+ *
+ * `T` is contravariant (it only appears in the match parameter), so a narrow
+ * matcher such as `Matcher<{ bar: number }>` is accepted wherever a wider
+ * object that structurally includes it is expected.
+ */
+export const MatcherSymbol = Symbol.for('@plugola/message-bus matcher')
+
+export interface Matcher<in T = unknown> {
+  [MatcherSymbol](value: T): boolean
+}
+
+export function isMatcher(value: unknown): value is Matcher {
+  return (
+    typeof value === 'object' &&
+    value !== null &&
+    typeof (value as Partial<Matcher>)[MatcherSymbol] === 'function'
+  )
+}
+
+/**
+ * Compare a registered filter argument against an emitted value: run the
+ * predicate if it's a {@link Matcher}, otherwise compare for strict equality.
+ */
+export function match(matcher: unknown, value: unknown): boolean {
+  return isMatcher(matcher) ? matcher[MatcherSymbol](value) : matcher === value
+}
+
+/**
+ * Widen a filter-prefix tuple so a {@link Matcher} is accepted in place of any
+ * literal argument.
+ *
+ * @example
+ * type A = Matchable<[string, { bar: number }]>
+ * // [string | Matcher<string>, { bar: number } | Matcher<{ bar: number }>]
+ */
+export type Matchable<A extends unknown[]> = {
+  [K in keyof A]: A[K] | Matcher<A[K]>
+}
+
+/**
+ * Match an argument against any boolean test.
+ *
+ * @example
+ * broker.on('foo', predicate((n: number) => n > 2), () => {})
+ */
+export function predicate<T>(match: (value: T) => boolean): Matcher<T> {
+  return { [MatcherSymbol]: match }
+}
+
+/**
+ * Match an object argument that deeply contains the given subset. Nested
+ * objects are compared recursively as partials; every other value is compared
+ * with strict equality.
+ *
+ * @example
+ * broker.on('foo', objectWith({ bar: 2 }), () => {})
+ * // fires for emit('foo', { bar: 2, baz: 9 })
+ */
+export function objectWith<S extends object>(subset: S): Matcher<S> {
+  return { [MatcherSymbol]: (value) => objectContains(value, subset) }
+}
+
+function objectContains(value: unknown, subset: object): boolean {
+  if (typeof value !== 'object' || value === null) return false
+
+  for (const key of Reflect.ownKeys(subset)) {
+    const expected = (subset as Record<PropertyKey, unknown>)[key]
+    const actual = (value as Record<PropertyKey, unknown>)[key]
+
+    if (expected !== null && typeof expected === 'object') {
+      if (!objectContains(actual, expected)) return false
+    } else if (actual !== expected) {
+      return false
+    }
+  }
+
+  return true
+}

packages/message-bus/src/types/events.ts

@@ -1,4 +1,5 @@
 import type { L } from 'ts-toolbelt'
+import type { Matcher, Matchable } from '../matcher.js'
 import type Broker from '../Broker.js'
 import type { CancelEvent } from '../symbols.js'
 import type { AddAbortSignal, MessageBusContext } from './MessageBus.js'
@@ -36,7 +37,7 @@ type _SubscriberArgs<
     : _SubscriberArgs<
         L.Pop<A>,
         L.Prepend<B, L.Last<A>>,
-        L.Append<A, SubscriberFn<B>> | Acc
+        L.Append<Matchable<A>, SubscriberFn<B>> | Acc
       >
 
 /**
@@ -48,10 +49,10 @@ type _SubscriberArgs<
  * // | [string]
  * // | [string, number]
  */
-export type UntilArgs<A extends unknown[]> = _UntilArgs<A, A>
+export type UntilArgs<A extends unknown[]> = _UntilArgs<A, Matchable<A>>
 
 type _UntilArgs<A extends unknown[], Acc extends unknown[]> =
-  L.Length<A> extends 0 ? Acc : _UntilArgs<L.Pop<A>, L.Pop<A> | Acc>
+  L.Length<A> extends 0 ? Acc : _UntilArgs<L.Pop<A>, Matchable<L.Pop<A>> | Acc>
 
 /**
  * The return value of `.until` from given arguments and
@@ -70,9 +71,11 @@ type _UntilArgs<A extends unknown[], Acc extends unknown[]> =
 export type UntilRtn<T extends unknown[], Args extends unknown[]> =
   L.Length<Args> extends 0
     ? T
-    : L.Head<T> extends L.Head<Args>
+    : L.Head<Args> extends Matcher<any>
       ? UntilRtn<L.Tail<T>, L.Tail<Args>>
-      : never
+      : L.Head<T> extends L.Head<Args>
+        ? UntilRtn<L.Tail<T>, L.Tail<Args>>
+        : never
 
 export type Subscribers<$ extends MessageBusContext> = Partial<{
   [EventName in keyof $['events']]: Subscriber<$>[]
@@ -114,7 +117,7 @@ type _EventInterceptorArgs<
         L.Pop<A>,
         L.Prepend<B, L.Last<A>>,
         C,
-        Acc | [...A, EventInterceptorFn<B, C>]
+        Acc | [...Matchable<A>, EventInterceptorFn<B, C>]
       >
 
 export type EventInterceptors<$ extends MessageBusContext> = Partial<{

packages/message-bus/src/types/generators.ts

@@ -1,4 +1,5 @@
 import type { L } from 'ts-toolbelt'
+import type { Matchable } from '../matcher.js'
 import type Broker from '../Broker.js'
 import type { AddAbortSignal, MessageBusContext } from './MessageBus.js'
 
@@ -38,7 +39,7 @@ export type _EventGeneratorArgs<
         L.Pop<A>,
         R,
         L.Prepend<B, L.Last<A>>,
-        Acc | L.Append<A, EventGeneratorFn<B, R>>
+        Acc | L.Append<Matchable<A>, EventGeneratorFn<B, R>>
       >
 
 export type EventGenerators<$ extends MessageBusContext> = Partial<{

packages/message-bus/src/types/invokables.ts

@@ -1,6 +1,7 @@
-import type { InvokerFn } from '@plugola/invoke'
+import type { InvokerFn, InvokerRegistrationArgs } from '@plugola/invoke'
 import type Broker from '../Broker.js'
 import type { L } from 'ts-toolbelt'
+import type { Matchable } from '../matcher.js'
 import type { MessageBusContext } from './MessageBus.js'
 
 export {
@@ -10,6 +11,23 @@ export {
   InvokerRegistrationArgs,
 } from '@plugola/invoke'
 
+/**
+ * {@link InvokerRegistrationArgs} with matchers allowed in every filter-prefix
+ * position. Invoke owns the base type and stays matcher-agnostic; we widen each
+ * union member's prefix here, at the message-bus boundary.
+ */
+export type MatchableInvokerRegistrationArgs<
+  A extends unknown[],
+  Return,
+> = WrapPrefix<InvokerRegistrationArgs<A, Return>>
+
+type WrapPrefix<U extends unknown[]> = U extends [
+  ...infer Pre extends unknown[],
+  infer Fn,
+]
+  ? [...Matchable<Pre>, Fn]
+  : U
+
 export interface Invoker<
   $ extends MessageBusContext,
   Args extends unknown[],
@@ -50,7 +68,7 @@ export type _InvokerInterceptorArgs<
         L.Pop<A>,
         Return,
         B,
-        Acc | L.Append<A, InvokerInterceptorFn<B, Return>>
+        Acc | L.Append<Matchable<A>, InvokerInterceptorFn<B, Return>>
       >
 
 export type InvokerInterceptors<$ extends MessageBusContext> = Partial<{

packages/message-bus/src/types/streams.ts

@@ -1,4 +1,5 @@
 import { L } from 'ts-toolbelt'
+import type { Matchable } from '../matcher.js'
 import Broker from '../Broker.js'
 import { AddAbortSignal, MessageBusContext } from './MessageBus.js'
 import { UnderlyingDefaultSource } from 'node:stream/web'
@@ -55,5 +56,5 @@ type _StreamReaderArgs<
         L.Pop<A>,
         Item,
         L.Prepend<B, L.Last<A>>,
-        L.Append<A, ReaderFn<B, Item>> | Acc
+        L.Append<Matchable<A>, ReaderFn<B, Item>> | Acc
       >

packages/message-bus/test/MessageBus.test.ts

@@ -10,6 +10,7 @@ import { CreateMessageBusContext } from '../src/types/MessageBus.js'
 import { CreateInvokablesDict } from '../src/types/invokables.js'
 import { CreateEventGenerators } from '../src/types/generators.js'
 import { write } from '@johngw/stream'
+import { objectWith, predicate } from '../src/matcher.js'
 
 describe('events', () => {
   type Events = CreateEvents<{ foo: []; bar: [string]; mung: [string, number] }>
@@ -439,3 +440,106 @@ describe('streams', () => {
     expect(fn.mock.calls[1][0]).toBe('10')
   })
 })
+
+describe('matchers', () => {
+  type $ = CreateMessageBusContext<{
+    events: {
+      foo: [{ bar: number; baz?: string }]
+      keyed: [{ id: number }, string]
+      nested: [{ meta: { tag: string } }]
+      num: [number]
+    }
+    invokables: {
+      handle: { args: [{ id: number }]; return: string }
+    }
+    streams: { feed: { args: [{ kind: string }]; item: string } }
+    generators: {
+      gen: { args: [{ tag: string }]; yield: string }
+    }
+  }>
+
+  let messageBus: MessageBus<$>
+  let broker: Broker<$>
+
+  beforeEach(() => {
+    messageBus = new MessageBus()
+    broker = messageBus.broker('test')
+    messageBus.start()
+  })
+
+  test('objectWith fires only on matching objects', () => {
+    const fn = vi.fn()
+    broker.on('foo', objectWith({ bar: 2 }), fn)
+    broker.emit('foo', { bar: 1 })
+    broker.emit('foo', { bar: 2 })
+    broker.emit('foo', { bar: 2, baz: 'x' })
+    expect(fn).toHaveBeenCalledTimes(2)
+  })
+
+  test('the matched argument is consumed', () => {
+    const fn = vi.fn()
+    broker.on('keyed', objectWith({ id: 1 }), fn)
+    broker.emit('keyed', { id: 1 }, 'hello')
+    expect(fn).toHaveBeenCalledWith('hello', expect.any(AbortSignal))
+  })
+
+  test('objectWith matches nested objects deeply', () => {
+    const fn = vi.fn()
+    broker.on('nested', objectWith({ meta: { tag: 'a' } }), fn)
+    broker.emit('nested', { meta: { tag: 'b' } })
+    broker.emit('nested', { meta: { tag: 'a' } })
+    expect(fn).toHaveBeenCalledTimes(1)
+  })
+
+  test('predicate matches arbitrary values', () => {
+    const fn = vi.fn()
+    broker.on(
+      'num',
+      predicate((n: number) => n > 2),
+      fn,
+    )
+    broker.emit('num', 1)
+    broker.emit('num', 3)
+    expect(fn).toHaveBeenCalledTimes(1)
+    expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal))
+  })
+
+  test('matchers filter invoker registrations', async () => {
+    broker.register('handle', objectWith({ id: 1 }), () => 'one')
+    broker.register('handle', () => 'other')
+    expect(await broker.invoke('handle', { id: 1 })).toBe('one')
+    expect(await broker.invoke('handle', { id: 2 })).toBe('other')
+  })
+
+  test('matchers filter stream readers', async () => {
+    const fn = vi.fn()
+    broker.reader('feed', objectWith({ kind: 'a' }), () => ({
+      start(controller) {
+        controller.enqueue('matched')
+        controller.close()
+      },
+    }))
+    broker.reader('feed', objectWith({ kind: 'b' }), () => ({
+      start(controller) {
+        controller.enqueue('ERROR')
+        controller.close()
+      },
+    }))
+    await broker.stream('feed', { kind: 'a' }).pipeTo(write(fn))
+    expect(fn).toHaveBeenCalledTimes(1)
+    expect(fn.mock.calls[0][0]).toBe('matched')
+  })
+
+  test('matchers filter generators', async () => {
+    const results: string[] = []
+    broker.generator('gen', objectWith({ tag: 'a' }), async function* () {
+      yield 'matched'
+    })
+    broker.generator('gen', objectWith({ tag: 'b' }), async function* () {
+      yield 'ERROR'
+    })
+    for await (const result of broker.iterate('gen', { tag: 'a' }))
+      results.push(result)
+    expect(results).toEqual(['matched'])
+  })
+})