Function pages

apps/sim/lib/data-drains/sources/audit-logs.ts:23–52

* Drains audit events scoped to the organization: rows from any of the org's * workspaces, plus org-level rows (`workspace_id IS NULL`) where * `metadata->>'organizationId'` matches. Audit-log writers consistently set * `metadata.organizationId` for org-scoped actions even though the table has *

(input: SourcePageInput)

 * no dedicated FK column.
 */
async function* pages(input: SourcePageInput): AsyncIterable<AuditLogRow[]> {
  const workspaceIds = await getOrganizationWorkspaceIds(input.organizationId)

  const orgScopedClause = and(
    isNull(auditLog.workspaceId),
    sql`${auditLog.metadata}->>'organizationId' = ${input.organizationId}`
  )
  const scopeClause =
    workspaceIds.length === 0
      ? orgScopedClause
      : or(inArray(auditLog.workspaceId, workspaceIds), orgScopedClause)

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

    const rows = await dbReplica
      .select()
      .from(auditLog)
      .where(and(scopeClause, timeCursorStabilityBound(auditLog.createdAt), cursorClause))
      .orderBy(...timeCursorOrderBy(auditLog.createdAt, auditLog.id))
      .limit(input.chunkSize)

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

export const auditLogsSource: DrainSource<AuditLogRow> = {
  type: 'audit_logs',