* Cursors on terminal `completedAt` so in-flight runs (mutable `status`, * `error`, `completedAt`) are not exported until they reach a terminal state.
(input: SourcePageInput)
| 18 | * `error`, `completedAt`) are not exported until they reach a terminal state. |
| 19 | */ |
| 20 | async function* pages(input: SourcePageInput): AsyncIterable<CopilotRunRow[]> { |
| 21 | const workspaceIds = await getOrganizationWorkspaceIds(input.organizationId) |
| 22 | if (workspaceIds.length === 0) return |
| 23 | |
| 24 | let cursor = decodeTimeCursor(input.cursor) |
| 25 | while (!input.signal.aborted) { |
| 26 | const cursorClause = timeCursorPredicate(copilotRuns.completedAt, copilotRuns.id, cursor) |
| 27 | |
| 28 | const rows = await dbReplica |
| 29 | .select() |
| 30 | .from(copilotRuns) |
| 31 | .where( |
| 32 | and( |
| 33 | inArray(copilotRuns.workspaceId, workspaceIds), |
| 34 | isNotNull(copilotRuns.completedAt), |
| 35 | timeCursorStabilityBound(copilotRuns.completedAt), |
| 36 | cursorClause |
| 37 | ) |
| 38 | ) |
| 39 | .orderBy(...timeCursorOrderBy(copilotRuns.completedAt, copilotRuns.id)) |
| 40 | .limit(input.chunkSize) |
| 41 | |
| 42 | if (rows.length === 0) return |
| 43 | yield rows |
| 44 | const last = rows[rows.length - 1] |
| 45 | cursor = { ts: last.completedAt!.toISOString(), id: last.id } |
| 46 | if (rows.length < input.chunkSize) return |
| 47 | } |
| 48 | } |
| 49 | |
| 50 | export const copilotRunsSource: DrainSource<CopilotRunRow> = { |
| 51 | type: 'copilot_runs', |
nothing calls this directly
no test coverage detected