Function collectUnprocessedById

packages/infra/src/ClusterCosmos.ts:743–763

(
  docs: ReadonlyArray<MessageDoc>,
  queryReplies: (
    query: string,
    parameters: ReadonlyArray<CosmosParameter>
  ) => Effect.Effect<Array<ReplyDoc>, E>
)

  })

const collectUnprocessedById = <E>(
  docs: ReadonlyArray<MessageDoc>,
  queryReplies: (
    query: string,
    parameters: ReadonlyArray<CosmosParameter>
  ) => Effect.Effect<Array<ReplyDoc>, E>
) =>
  Effect.gen(function*() {
    const messages: Array<{
      readonly envelope: Envelope.Encoded
      readonly lastSentReply: Option.Option<Reply.Encoded>
    }> = []
    const activeRequestIds = yield* activeReplyRequestIds(docs, queryReplies)
    const lastReplies = yield* lastRepliesById(docs, activeRequestIds, queryReplies)
    for (const doc of docs) {
      if (activeRequestIds.has(doc.requestId)) continue
      const sentReply = Option.fromNullishOr(doc.lastReplyId === null ? undefined : lastReplies.get(doc.lastReplyId))
      messages.push(envelopeFromDoc(doc, sentReply))
    }
    return messages
  })

const activeReplyRequestIds = <E>(
  docs: ReadonlyArray<MessageDoc>,