async def refresh_all_source_connections(
    *,
    readable_id: str,
    db: AsyncSession = Depends(deps.get_db),
    current_user: User = Depends(deps.get_user),
    background_tasks: BackgroundTasks,
) -> list[schemas.SourceConnectionJob]:
    """Start sync jobs for all source connections in the collection.

    Args:
        readable_id: The readable ID of the collection
        db: The database session
        current_user: The current user
        background_tasks: Background tasks for async operations

    Returns:
        A list of created sync jobs
    """
    # Check if collection exists
    collection = await crud.collection.get_by_readable_id(
        db, readable_id=readable_id, current_user=current_user
    )
    if collection is None:
        raise HTTPException(status_code=404, detail="Collection not found")

    # Convert to Pydantic model immediately
    collection_obj = schemas.Collection.model_validate(collection, from_attributes=True)

    # Get all source connections for this collection
    source_connections = await source_connection_service.get_source_connections_by_collection(
        db=db, collection=readable_id, current_user=current_user
    )

    if not source_connections:
        return []

    # Create a sync job for each source connection and run it in the background
    sync_jobs = []

    for sc in source_connections:
        # Create the sync job
        sync_job = await source_connection_service.run_source_connection(
            db=db, source_connection_id=sc.id, current_user=current_user
        )

        # Get necessary objects for running the sync
        sync = await crud.sync.get(
            db=db, id=sync_job.sync_id, current_user=current_user, with_connections=True
        )
        sync_dag = await sync_service.get_sync_dag(
            db=db, sync_id=sync_job.sync_id, current_user=current_user
        )

        # Get source connection with auth_fields for temporal processing
        source_connection = await source_connection_service.get_source_connection(
            db=db,
            source_connection_id=sc.id,
            show_auth_fields=True,  # Important: Need actual auth_fields for temporal
            current_user=current_user,
        )

        # Prepare objects for background task
        sync = schemas.Sync.model_validate(sync, from_attributes=True)
        sync_dag = schemas.SyncDag.model_validate(sync_dag, from_attributes=True)
        source_connection = schemas.SourceConnection.from_orm_with_collection_mapping(
            source_connection
        )

        # Add to jobs list
        sync_jobs.append(sync_job.to_source_connection_job(sc.id))

        # Start the sync job in the background or via Temporal
        if await temporal_service.is_temporal_enabled():
            # Use Temporal workflow
            await temporal_service.run_source_connection_workflow(
                sync=sync,
                sync_job=sync_job,
                sync_dag=sync_dag,
                collection=collection_obj,  # Use the already converted object
                source_connection=source_connection,
                user=current_user,
            )
        else:
            # Fall back to background tasks
            background_tasks.add_task(
                sync_service.run,
                sync,
                sync_job,
                sync_dag,
                collection_obj,  # Use the already converted object
                source_connection,
                current_user,
            )

    return sync_jobs
Start sync jobs for all source connections in the collection.

    Args:
        readable_id: The readable ID of the collection
        db: The database session
        current_user: The current user
        background_tasks: Background tasks for async operations

    Returns:
        A list of created sync jobs
    
backend/airweave/api/v1/endpoints/collections.py