Function pages

apps/sim/lib/data-drains/sources/job-logs.ts:20–48

* Cursors on terminal `endedAt` so in-flight rows (mutable `status`, `endedAt`, * `totalDurationMs`, `executionData`) are not exported until finalized.

(input: SourcePageInput)

 * `totalDurationMs`, `executionData`) are not exported until finalized.
 */
async function* pages(input: SourcePageInput): AsyncIterable<JobLogRow[]> {
  const workspaceIds = await getOrganizationWorkspaceIds(input.organizationId)
  if (workspaceIds.length === 0) return

  let cursor = decodeTimeCursor(input.cursor)
  while (!input.signal.aborted) {
    const cursorClause = timeCursorPredicate(jobExecutionLogs.endedAt, jobExecutionLogs.id, cursor)

    const rows = await dbReplica
      .select()
      .from(jobExecutionLogs)
      .where(
        and(
          inArray(jobExecutionLogs.workspaceId, workspaceIds),
          isNotNull(jobExecutionLogs.endedAt),
          timeCursorStabilityBound(jobExecutionLogs.endedAt),
          cursorClause
        )
      )
      .orderBy(...timeCursorOrderBy(jobExecutionLogs.endedAt, jobExecutionLogs.id))
      .limit(input.chunkSize)

    if (rows.length === 0) return
    yield rows
    const last = rows[rows.length - 1]
    cursor = { ts: last.endedAt!.toISOString(), id: last.id }
    if (rows.length < input.chunkSize) return
  }
}

export const jobLogsSource: DrainSource<JobLogRow> = {
  type: 'job_logs',