chore: replace onceSatisfied with p-event (#522)

Lets us remove a bunch of code!

package.json

@@ -89,6 +89,7 @@
     "drizzle-kit": "^0.19.12",
     "eslint": "^8.57.0",
     "husky": "^8.0.0",
+    "iterpal": "^0.3.0",
     "light-my-request": "^5.10.0",
     "lint-staged": "^14.0.1",
     "mapeo-offline-map": "^2.0.0",
@@ -139,6 +140,7 @@
     "mime": "^4.0.1",
     "multi-core-indexer": "^1.0.0-alpha.9",
     "p-defer": "^4.0.0",
+    "p-event": "^6.0.1",
     "p-timeout": "^6.1.2",
     "patch-package": "^8.0.0",
     "protobufjs": "^7.2.3",

src/invite-api.js

@@ -1,7 +1,8 @@
 // @ts-check
 import { TypedEmitter } from 'tiny-typed-emitter'
+import { pEvent } from 'p-event'
 import { InviteResponse_Decision } from './generated/rpc.js'
-import { assert, keyToId, onceSatisfied, noop } from './utils.js'
+import { assert, keyToId, noop } from './utils.js'
 import HashMap from './lib/hashmap.js'
 import timingSafeEqual from './lib/timing-safe-equal.js'
 
@@ -251,19 +252,18 @@ export class InviteApi extends TypedEmitter {
     const projectDetailsAbortController = new AbortController()
 
     const projectDetailsPromise =
-      /** @type {typeof onceSatisfied<TypedEmitter<import('./local-peers.js').LocalPeersEvents>, 'got-project-details'>} */ (
-        onceSatisfied
-      )(
-        this.rpc,
-        'got-project-details',
-        (projectDetailsPeerId, details) =>
+      /** @type {typeof pEvent<'got-project-details', [string, ProjectJoinDetails]>} */ (
+        pEvent
+      )(this.rpc, 'got-project-details', {
+        multiArgs: true,
+        filter: ([projectDetailsPeerId, details]) =>
           // This peer ID check is probably superfluous because the invite ID
           // should be unguessable, but might be useful if someone forwards an
           // invite message (or if there's an unforeseen bug).
           timingSafeEqual(projectDetailsPeerId, peerId) &&
           timingSafeEqual(inviteId, details.inviteId),
-        { signal: projectDetailsAbortController.signal }
-      )
+        signal: projectDetailsAbortController.signal,
+      })
         .then((args) => args?.[1])
         .catch(noop)
 

src/member-api.js

@@ -1,11 +1,11 @@
 import * as crypto from 'node:crypto'
 import { TypedEmitter } from 'tiny-typed-emitter'
+import { pEvent } from 'p-event'
 import { InviteResponse_Decision } from './generated/rpc.js'
 import {
   assert,
   noop,
   ExhaustivenessError,
-  onceSatisfied,
   projectKeyToId,
   projectKeyToPublicId,
 } from './utils.js'
@@ -180,16 +180,15 @@ export class MemberApi extends TypedEmitter {
     const timeoutId = setTimeout(() => abortController.abort(), timeout)
 
     const responsePromise =
-      /** @type {typeof onceSatisfied<TypedEmitter<import('./local-peers.js').LocalPeersEvents>, 'invite-response'>} */ (
-        onceSatisfied
-      )(
-        this.#rpc,
-        'invite-response',
-        (peerId, inviteResponse) =>
+      /** @type {typeof pEvent<'invite-response', [string, InviteResponse]>} */ (
+        pEvent
+      )(this.#rpc, 'invite-response', {
+        multiArgs: true,
+        filter: ([peerId, inviteResponse]) =>
           timingSafeEqual(peerId, deviceId) &&
           timingSafeEqual(invite.inviteId, inviteResponse.inviteId),
-        { signal: abortController.signal }
-      ).then((args) => args?.[1])
+        signal: abortController.signal,
+      }).then((args) => args?.[1])
 
     responsePromise.catch(noop)
 

src/utils.js

@@ -110,111 +110,6 @@ export function setHas(set) {
   return (value) => set.has(/** @type {*} */ (value))
 }
 
-/**
- * @internal
- * @template {import('tiny-typed-emitter').ListenerSignature<L>} L
- * @typedef {import('tiny-typed-emitter').TypedEmitter<L>} TypedEmitter
- */
-
-/**
- * @internal
- * @template {TypedEmitter<any>} T
- * @typedef {import('./utils_types.d.ts').TypedEvents<T>} TypedEvents
- */
-
-/**
- * @internal
- * @template {TypedEmitter<any>} T
- * @typedef {import('./utils_types.d.ts').TypedEventsFor<T>} TypedEventsFor
- */
-
-/**
- * @internal
- * @template {TypedEmitter<any>} Emitter
- * @template {TypedEventsFor<Emitter>} Event
- * @typedef {import('./utils_types.d.ts').TypedEventArgs<Emitter, Event>} TypedEventArgs
- */
-
-/**
- * Like `once`, but only resolves after the event's arguments satisfy `check`.
- * Useful when you want to listen for a particular instance of an event.
- *
- * Due to an unfortunate TypeScript quirk, you need to manually specify the
- * generic types to use this function with `TypedEmitter` subclasses. For
- * example:
- *
- * ```
- * const res = await onceSatisfied<TypedEmitter<MyEvents>, 'my-event'>(
- *   myEmitterSubclass,
- *   'my-event'
- * )
- * ```
- *
- * It would be great to remove this verbosity, but it is the only way to achieve
- * type safety. Luckily, you'll get a type error if you do it incorrectly.
- *
- * @template {TypedEmitter<any>} Emitter
- * @template {TypedEventsFor<Emitter>} Event
- * @param {Emitter} emitter A `tiny-typed-emitter` event emitter.
- * @param {Event} eventName The event to listen to.
- * @param {(...args: TypedEventArgs<Emitter, Event>) => boolean} check
- *   Called with the event's arguments. If this function returns `false`, the
- *   event will be ignored. If this function returns `true`, the promise will
- *   resolve with these arguments and the event listener will be cleaned up.
- * @param {object} [options]
- * @param {AbortSignal} [options.signal] An `AbortSignal`. If this signal is
- *   aborted, the promise will resolve with `null`.
- * @returns {Promise<TypedEventArgs<Emitter, Event>>} Resolves with the
- *   event arguments that satisfy `check`, or `null` if the aborted.
- */
-export const onceSatisfied = (emitter, eventName, check, { signal } = {}) =>
-  new Promise((res, rej) => {
-    /** @type {() => void} */
-    let cleanup
-    if (signal) {
-      if (signal.aborted) {
-        rej(signal.reason)
-        return
-      }
-
-      const onAbort = () => {
-        cleanup()
-        rej(signal.reason)
-      }
-      signal.addEventListener('abort', onAbort, { once: true })
-
-      cleanup = () => {
-        emitter.off(eventName, onEvent)
-        signal.removeEventListener('abort', onAbort)
-      }
-    } else {
-      cleanup = () => {
-        emitter.off(eventName, onEvent)
-      }
-    }
-
-    /** @param {TypedEventArgs<Emitter, Event>} args */
-    const onEvent = (...args) => {
-      /** @type {unknown} */
-      let err
-
-      try {
-        if (!check(...args)) return
-      } catch (e) {
-        err = e
-      }
-
-      cleanup()
-      if (err) {
-        rej(err)
-      } else {
-        res(args)
-      }
-    }
-
-    emitter.on(eventName, onEvent)
-  })
-
 /**
  * When reading from SQLite, any optional properties are set to `null`. This
  * converts `null` back to `undefined` to match the input types (e.g. the types

tests/helpers/events.js

@@ -1,56 +1,22 @@
 // @ts-check
-import { assert, onceSatisfied } from '../../src/utils.js'
-
-/**
- * @internal
- * @template {import('tiny-typed-emitter').ListenerSignature<L>} L
- * @typedef {import('tiny-typed-emitter').TypedEmitter<L>} TypedEmitter
- */
-
-/**
- * @internal
- * @template {TypedEmitter<any>} T
- * @typedef {import('../../src/utils_types.d.ts').TypedEvents<T>} TypedEvents
- */
-
-/**
- * @internal
- * @template {TypedEmitter<any>} T
- * @typedef {import('../../src/utils_types.d.ts').TypedEventsFor<T>} TypedEventsFor
- */
-
-/**
- * @internal
- * @template {TypedEmitter<any>} Emitter
- * @template {TypedEventsFor<Emitter>} Event
- * @typedef {import('../../src/utils_types.d.ts').TypedEventArgs<Emitter, Event>} TypedEventArgs
- */
+import { assert } from '../../src/utils.js'
+import { pEventIterator } from 'p-event'
+import { arrayFrom } from 'iterpal'
 
 /**
  * Like `once`, but listens to events up to a certain number of times.
  *
- * @template {TypedEmitter<any>} Emitter
- * @template {TypedEventsFor<Emitter>} Event
- * @param {Emitter} emitter
- * @param {Event} eventName
+ * @param {import('node:events').EventEmitter} emitter
+ * @param {string | symbol} eventName
  * @param {number} count
- * @returns {Promise<TypedEventArgs<Emitter, Event>[]>}
+ * @returns {Promise<unknown[]>}
  */
-export async function onTimes(emitter, eventName, count) {
+export function onTimes(emitter, eventName, count) {
   assert(
     Number.isSafeInteger(count) && count >= 0,
     'onTimes called with an invalid count'
   )
 
-  /** @type {TypedEventArgs<Emitter, Event>[]} */
-  const result = []
-
-  if (count === 0) return result
-
-  await onceSatisfied(emitter, eventName, (...args) => {
-    result.push(args)
-    return result.length === count
-  })
-
-  return result
+  const events = pEventIterator(emitter, eventName, { limit: count })
+  return arrayFrom(events)
 }

tests/invite-api.js

@@ -101,11 +101,7 @@ test('invite-received event has expected payload', async (t) => {
     },
   ]
   const receivedInvitesArgs = await invitesReceivedPromise
-  t.alike(
-    receivedInvitesArgs,
-    expectedInvites.map((i) => [i]),
-    'received expected invites'
-  )
+  t.alike(receivedInvitesArgs, expectedInvites, 'received expected invites')
   t.alike(inviteApi.getPending(), expectedInvites)
 })
 
@@ -492,7 +488,7 @@ test('Receiving invite for project that peer already belongs to', async (t) => {
       'got expected responses'
     )
 
-    const removedInvites = (await invitesRemovedPromise).map((args) => args[0])
+    const removedInvites = await invitesRemovedPromise
     const allButLastRemoved = removedInvites.slice(0, -1)
     const lastRemoved = removedInvites[removedInvites.length - 1]
     t.alike(