Bulk Operations

Relevant source files

The following files were used as context for generating this wiki page:

• backend/src/workflows/domain/bulk_enrollment_entities.py
• backend/src/workflows/application/use_cases/create_bulk_job.py
• backend/src/workflows/infrastructure/bulk_enrollment_models.py
• backend/src/workflows/presentation/bulk_enrollment_routes.py
• frontend/src/components/bulk/bulk-enrollment-modal.tsx
• DDD_IMPLEMENTATION_REPORT.md

Bulk Operations in the platform provide an asynchronous batch contact enrollment system designed to handle large-scale data processing within workflows. This system allows for the enrollment of up to 10,000 contacts per job through various selection methods, including manual selection, filter-based criteria, or CSV uploads. The architecture is built to ensure reliable execution through retry logic, real-time progress tracking, and detailed failure recording for debugging.

The primary focus of bulk operations is the Bulk Enrollment system, which automates the entry of multiple contacts into specific automation Workflows. This feature is essential for large marketing campaigns, database-wide updates, and initial onboarding processes.

Sources: DDD_IMPLEMENTATION_REPORT.md, backend/src/workflows/domain/bulk_enrollment_entities.py:1-20

Architecture and Domain Model

The system follows Domain-Driven Design (DDD) principles, organizing Bulk Operations into a four-layer architecture: Domain, Application, Infrastructure, and Presentation. The domain layer encapsulates the core business logic, defining the lifecycle of a bulk job and its constituent batches.

Core Domain Entities

The Bulk Enrollment system is governed by several key entities that manage the lifecycle and processing of batch jobs.

• BulkEnrollmentJob: The aggregate root representing a single bulk operation. It tracks the overall progress, status, and configuration of the enrollment task.
• BulkEnrollmentBatch: Individual processing units that divide a large job into smaller, manageable chunks for asynchronous execution.
• EnrollmentFailure: A record used for debugging, capturing specific reasons why a contact failed to enroll in a workflow.
• SelectionCriteria: A value object defining how contacts are selected (e.g., via specific IDs, filter sets, or CSV data).

Sources: backend/src/workflows/domain/bulk_enrollment_entities.py, DDD_IMPLEMENTATION_REPORT.md

Job Lifecycle and Statuses

A bulk job transitions through various states during its execution. These states are defined by the JobStatus and BatchStatus enums.

Status	Description
PENDING	Job is created and awaiting processing.
PROCESSING	Job is currently enrolling contacts in batches.
COMPLETED	All contacts have been successfully processed (or attempted).
FAILED	The job encountered a critical error preventing further processing.
CANCELLED	The operation was manually terminated by the user.

Sources: backend/src/workflows/domain/bulk_enrollment_entities.py, DDD_IMPLEMENTATION_REPORT.md

Data Flow and Execution

The execution flow begins in the Presentation layer via REST API endpoints, moves through the Application use cases to validate and create domain entities, and finally utilizes the Infrastructure layer for persistence and worker-based processing.

Bulk Enrollment Sequence

The following diagram illustrates the sequence of events from a user initiating a bulk enrollment to the system processing it asynchronously.

Database Schema

The infrastructure layer implements the domain entities using SQLAlchemy models, ensuring that bulk jobs are persistent and trackable.

Entity Relationship Diagram

The relationship between jobs, batches, and failures is managed through foreign key constraints and optimized with partial indexes for active jobs.

API Reference

The Presentation layer exposes several endpoints to manage the lifecycle of Bulk Operations. All endpoints are account-scoped to ensure multi-tenancy security.

Method	Endpoint	Description
POST	/api/v1/bulk-enrollment/workflows/{id}/jobs	Initiates a new bulk enrollment job for a specific workflow.
GET	/api/v1/bulk-enrollment/jobs	Lists all bulk jobs with filtering support (e.g., by status).
GET	/api/v1/bulk-enrollment/jobs/{id}/progress	Retrieves real-time progress percentages and counts.
POST	/api/v1/bulk-enrollment/jobs/{id}/cancel	Manually cancels an active job.
GET	/api/v1/bulk-enrollment/jobs/{id}/failures	Fetches detailed failure records for a specific job.

Sources: backend/src/workflows/presentation/bulk_enrollment_routes.py, DDD_IMPLEMENTATION_REPORT.md

Implementation Details (Python)

The backend implementation utilizes Pydantic DTOs for input validation and use cases for business logic orchestration.

# Example of the CreateBulkJobUseCase structure
class CreateBulkJobUseCase:
    async def execute(self, workflow_id: UUID, request: CreateBulkJobRequestDTO):
        # 1. Validate contact selection (limit 10,000)
        # 2. Check for active jobs on the same workflow
        # 3. Create BulkEnrollmentJob aggregate root
        # 4. Persist and dispatch to background worker
        pass

Sources: backend/src/workflows/application/use_cases/create_bulk_job.py, DDD_IMPLEMENTATION_REPORT.md

Conclusion

The Bulk Operations system is a robust, asynchronous framework designed to handle high-volume contact processing within the platform's automation engine. By leveraging a DDD-based architecture and background workers, it provides users with the ability to manage thousands of enrollments efficiently while maintaining a clear audit trail of successes and failures. The integration of real-time progress tracking and cancellation capabilities ensures that large-scale operations remain transparent and controllable.