Function pages

apps/sim/lib/data-drains/sources/copilot-chats.ts:51–102

* Cursor is `createdAt` (immutable) but rows themselves are mutable — * `messages`, `title`, `lastSeenAt`, etc. are updated in-place over the chat's * lifetime. This means a chat exported once will not be re-exported when its * messages change. Consumers who need the latest state should periodica

(input: SourcePageInput)

 * design and `data-drains` is not a CDC pipeline.
 */
async function* pages(input: SourcePageInput): AsyncIterable<CopilotChatRow[]> {
  const workspaceIds = await getOrganizationWorkspaceIds(input.organizationId)
  if (workspaceIds.length === 0) return

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

    const metaRows = await dbReplica
      .select(chatColumns)
      .from(copilotChats)
      .where(
        and(
          inArray(copilotChats.workspaceId, workspaceIds),
          timeCursorStabilityBound(copilotChats.createdAt),
          cursorClause
        )
      )
      .orderBy(...timeCursorOrderBy(copilotChats.createdAt, copilotChats.id))
      .limit(input.chunkSize)

    if (metaRows.length === 0) return

    const chatIds = metaRows.map((r) => r.id)
    const messageRows = await dbReplica
      .select({ chatId: copilotMessages.chatId, content: copilotMessages.content })
      .from(copilotMessages)
      .where(and(inArray(copilotMessages.chatId, chatIds), isNull(copilotMessages.deletedAt)))
      .orderBy(
        asc(copilotMessages.chatId),
        sql`${copilotMessages.seq} asc nulls last`,
        asc(copilotMessages.createdAt),
        asc(copilotMessages.id)
      )
    const messagesByChat = new Map<string, unknown[]>()
    for (const m of messageRows) {
      const existing = messagesByChat.get(m.chatId)
      if (existing) existing.push(m.content)
      else messagesByChat.set(m.chatId, [m.content])
    }

    const rows: CopilotChatRow[] = metaRows.map((r) => ({
      ...r,
      messages: messagesByChat.get(r.id) ?? [],
    }))

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

export const copilotChatsSource: DrainSource<CopilotChatRow> = {
  type: 'copilot_chats',