Function activeReplyRequestIds

packages/infra/src/ClusterCosmos.ts:765–779

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

  })

const activeReplyRequestIds = <E>(
  docs: ReadonlyArray<MessageDoc>,
  queryReplies: (
    query: string,
    parameters: ReadonlyArray<CosmosParameter>
  ) => Effect.Effect<Array<ReplyDoc>, E>
) => {
  const requestIds = Array.from(new Set(docs.map((doc) => doc.requestId)))
  if (!Arr.isArrayNonEmpty(requestIds)) return Effect.succeed(new Set<string>())
  return queryReplies(
    "SELECT * FROM c WHERE c.type = 'reply' AND ARRAY_CONTAINS(@requestIds, c.requestId) AND (c.kind = 'WithExit' OR (c.kind = 'Chunk' AND c.acked = false))",
    [{ name: "@requestIds", value: requestIds }]
  )
    .pipe(Effect.map((replies) => new Set(replies.map((reply) => reply.requestId))))
}

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