Function collectUnprocessed

packages/infra/src/ClusterCosmos.ts:716–741

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

})

const collectUnprocessed = <E>(
  docs: ReadonlyArray<MessageDoc>,
  now: number,
  claimMessageRead: (doc: MessageDoc, now: number) => Effect.Effect<boolean, E>,
  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))
      const claimed = yield* claimMessageRead(doc, now)
      if (claimed) {
        messages.push(envelopeFromDoc({ ...doc, lastRead: now }, sentReply))
      }
    }
    return messages
  })

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