Creating a Stream for Database Entity Updates

type objectForPrimaryKey = <A>(entityType: string, id: BSON.ObjectId) => Effect.Effect<R, E, A>
type wrapCallbackInStream = <A>(identifier: string, fn: (cb: (val: T) => void) => () => void) => Stream.Stream<never, never, A>

  // userActivityStream: (id: BSON.ObjectId) => Stream.Stream<R, E, UserActivity>
  const userActivityStream = (id: BSON.ObjectId) => pipe(
    Stream.fromEffect(dbWrapper.objectForPrimaryKey<UserActivityRO>('UserActivity', id)),
    Stream.flatMap(act =>
      wrapCallbackInStream<UserActivity>(
        `UserActivity/${id}`,
        cb => {
          act.addListener(cb)
          return () => act.removeListener(cb)
        }
      )
    )
  )