Documento de referencia del refactor del feature de importación de evaluaciones desde Cosmic Latte hacia ERAS. Cubre backend (ERAS-BE, ASP.NET Core 8 / EF Core 9 / PostgreSQL) y frontend (ERAS-FE, Angular 19). Pensado como guía para mantener y extender el feature.

El proceso de importación pasó de ser síncrono, bloqueante y frágil a una arquitectura asíncrona por estudiante con extracción y confirmación separadas:

El trabajo se hizo en fases incrementales (0→5). Cada fase es independientemente desplegable.

flowchart LR
    subgraph FE["ERAS-FE (Angular 19)"]
        EP["Evaluation Process List<br/>(trigger import)"]
        IV["Import Status View<br/>(unified, polling)"]
    end

    subgraph API["ERAS-BE · Eras.Api"]
        CC["CosmicLatteController<br/>extract / confirm / status / items / retry"]
    end

    subgraph APP["Eras.Application"]
        IJS["ImportJobService"]
        ORC["PollOrchestratorService<br/>+ importers (Phase 3)"]
        UOW["IUnitOfWork"]
    end

    subgraph INFRA["Eras.Infrastructure"]
        Q["ImportJobQueue<br/>Channel of int · singleton"]
        W["ImportQueueBackgroundService<br/>(BackgroundService)"]
        CL["CosmicLatteAPIService<br/>ExtractRespondentsAsync"]
        REPO["ImportJob / ImportJobItem<br/>repositories (set-based)"]
    end

    DB[("PostgreSQL")]
    CLAPI[["Cosmic Latte API"]]

    EP -->|"POST imports/extract"| CC
    IV -->|"poll GET imports/:id (+items)"| CC
    IV -->|"POST confirm / retry"| CC
    CC --> IJS
    IJS -->|"crea ImportJob + encola id"| Q
    IJS --> REPO
    W -->|"dequeue id"| Q
    W -->|"fase Extracting"| CL
    W -->|"fase Importing"| ORC
    CL -->|"detalle por estudiante"| CLAPI
    ORC --> UOW
    ORC --> REPO
    REPO --> DB
    UOW --> DB

Idea central: el ImportJob es una máquina de estados persistida; un único worker la avanza por fases consumiendo de una cola in-process. El frontend nunca espera; hace polling y muestra progreso.

stateDiagram-v2
    [*] --> Extracting: POST imports/extract
    Extracting --> Ready: extracción completa
    Extracting --> Failed: error de extracción
    Ready --> Importing: POST confirm (ids)
    Importing --> Completed: todos OK
    Importing --> PartiallyCompleted: algunos fallaron
    Importing --> Failed: setup falló
    PartiallyCompleted --> Importing: POST retry (ids)
    Failed --> Importing: POST retry (ids)
    Completed --> [*]

Estados por item (ImportJobItem.Status): Extracting → Extracted → (Skipped | Queued) → Running → Completed | Failed.

• Extracted: respondiente traído de Cosmic Latte, a la espera de confirmación.
• Skipped: no seleccionado al confirmar.
• Queued/Running/Completed/Failed: ciclo de la importación real (persistencia).

El estado agregado del job se recomputa por SQL (GROUP BY status) contando solo la fase de import (Pending = Queued+Running, Completed, Failed); Extracted/Skipped no cuentan.

sequenceDiagram
    autonumber
    participant U as Usuario
    participant FE as Import Status View
    participant API as CosmicLatteController
    participant SVC as ImportJobService
    participant Q as ImportJobQueue
    participant W as Worker
    participant CL as CosmicLatteAPIService
    participant ORC as PollOrchestrator
    participant DB as PostgreSQL

    U->>FE: Importar (modal: poll, fechas, config)
    FE->>API: POST imports/extract
    API->>SVC: StartExtractionAsync(...)
    SVC->>DB: ImportJob(status=Extracting)
    SVC->>Q: enqueue(jobId)
    API-->>FE: 202 (importJobId)
    FE->>FE: navega a import-status/:id

    W->>Q: dequeue(jobId)
    W->>CL: ExtractRespondentsAsync(...)
    CL->>CL: lista + detalle por estudiante (paralelo acotado)
    loop por respondiente
        CL-->>W: onExtracted(pollDto, alreadyImported)
        W->>DB: ImportJobItem(status=Extracted) + extractedCount++
    end
    W->>DB: job → Ready (totalCount = extraídos)

    loop polling cada 3s
        FE->>API: GET imports/:id (+items)
        API-->>FE: progreso (barra) / lista
    end

    U->>FE: selecciona y "Import selected"
    FE->>API: POST imports/:id/confirm (itemIds)
    API->>SVC: ConfirmImportAsync
    SVC->>DB: seleccionados=Queued, resto=Skipped, job=Importing
    SVC->>Q: enqueue(jobId)
    API-->>FE: 202

    W->>Q: dequeue(jobId)
    W->>ORC: SetupImportStructureAsync (1 transacción)
    loop por item Queued
        W->>ORC: ProcessStudentAsync(poll) (1 transacción/estudiante)
        ORC->>DB: student + poll instance + answers
        W->>DB: item → Completed | Failed
    end
    W->>DB: job → Completed | PartiallyCompleted

erDiagram
    import_jobs ||--o{ import_job_items : "1:N"
    import_jobs {
        int Id PK
        int evaluation_id
        string status "varchar(20)"
        int total_count
        int processed_count
        int extracted_count
        int retry_count
        string error_message
        string evaluation_set_name
        int configuration_id
        string start_date
        string end_date
        jsonb polls_payload "legacy/empty en flujo extract"
        timestamptz created_at_utc
        timestamptz updated_at_utc
    }
    import_job_items {
        int Id PK
        int import_job_id FK
        string student_email
        string student_name
        string cohort
        string status "varchar(20)"
        int retry_count
        bool is_already_imported
        string error_message
        jsonb poll_payload "PollDTO del estudiante"
        timestamptz created_at_utc
        timestamptz updated_at_utc
    }

• Índices: ix_import_jobs_evaluation_id, ix_import_job_items_import_job_id_status.
• poll_instances.answers_hash (+ índice ix_poll_instances_student_id_answers_hash): detección de duplicados por hash persistido e indexado (antes se cargaba todo y se hasheaba en memoria).
• Migraciones escritas a mano (AddAnswersHashToPollInstance, AddImportJobs, AddImportJobItems, AddExtractionFields) para no arrastrar drift preexistente de poll_instances (un FK/índice de EvaluationId que el modelo configura pero que ninguna migración previa capturó). Se aplican solas al arrancar vía Database.Migrate() en Program.cs.

• Una transacción por estudiante (ProcessStudentAsync envuelto en IUnitOfWork.ExecuteInTransactionAsync): un fallo aísla a ese estudiante; los demás se confirman → habilita el reintento selectivo.
• El reintento no depende de estado en memoria: ProcessStudentAsync reconstruye la estructura del poll desde la BD (IVariableRepository.GetAllByPollUuidAsync), por eso un retry en otra request funciona.
• Updates set-based (ExecuteUpdateAsync) en estado de job/item: evitan el conflicto "entidad ya trackeada" que aparece al hacer GetByIdAsync + UpdateAsync repetidos en un mismo scope.
• Worker resiliente: cada job se procesa en try/catch; una excepción nunca detiene el host (BackgroundService por defecto haría StopHost).
• Concurrencia de extracción: HTTP en paralelo acotado (SemaphoreSlim(8)), pero la persistencia del callback es serializada (SemaphoreSlim(1)) porque el DbContext (scoped) no es thread-safe.

flowchart TD
    A["Evaluation Process (modal)"] -->|"startExtraction"| B["POST imports/extract → 202"]
    B --> C["navega import-status/:id"]
    C --> D{"polling status"}
    D -->|"Extracting"| E["barra indeterminada + contador"]
    D -->|"Ready"| F["selección de Extracted + Import selected"]
    F -->|"confirmImport (itemIds)"| G["POST confirm → 202"]
    G --> D
    D -->|"Importing"| H["barra determinada processed/total"]
    D -->|"Completed / Partial / Failed"| I["resultado + Retry selected"]
    I -->|"retryImportItems"| G

• Polling con timer(0,3000)+switchMap+takeWhile: solo activo en fases Extracting/Importing; pausa en Ready (espera confirmación) y en terminales.
• No bloquea la UI: la ruta es navegable; el job persiste y se puede volver.
• Selección: en Ready se seleccionan Extracted (pre-deselecciona ya-importados e inválidos); en fase import, multi-select de Failed para reintentar.

• Nuevo estado de item/job: agregar al enum ImportJobStatus (la columna es varchar(20)), mapear en ImportStatusBadgeComponent (el Record<ImportJobStatus,...> debe ser exhaustivo) y en el union del FE.
• Otro paso en la importación de un estudiante: extender PollOrchestratorService.ProcessStudentAsync (sigue dentro de la transacción por estudiante); reusar los importers.
• Cambiar el paralelismo de extracción: ajustar SemaphoreSlim en ExtractRespondentsAsync (httpGate); mantener persistGate(1) para no usar el DbContext concurrentemente.
• Persistencia/idempotencia entre reinicios: los payloads viven en ImportJobItem.poll_payload; el worker reprocesa por estado, así que un job puede re-encolarse sin perder datos.
• Migraciones: escribirlas a mano (ver §5) para no empaquetar el drift de poll_instances; actualizar AppDbContextModelSnapshot.cs en consecuencia.

• Backend unit: worker en fase Extracting (crea items, extractedCount sube, → Ready); confirm marca Queued/Skipped; per-student con un fallo → PartiallyCompleted; retry vuelve a Completed.
• Frontend (Jasmine/Karma): servicio (URLs), vista unificada (barra por fase, selección por estado, polling se detiene en Ready/terminal), badge.
• E2E local (Docker): ver docker-compose.local.yml (stack para Docker nativo en WSL). Flujo validado: extracción incremental 0→N → Ready, confirmar selección → Importing → Completed, retry de fallidos. Notas de despliegue local: proxy nginx /api (same-origin, evita CORS), backend en host networking (issuer de Keycloak coherente), client_max_body_size elevado.

• Endpoints/método legacy (POST polls/{id}, GetAllPollsPreview, CosmicLatteAPIService.SavePreviewPolls, importAnswerBySurvey en el FE) quedaron sin uso; candidatos a eliminación.
• Drift de poll_instances (FK/índice de EvaluationId no migrado): abordar en una migración dedicada.
• Reintentos automáticos con back-off ante fallos transitorios (hoy el retry es manual).
• Tests automatizados de las Fases 4–5 (se validaron E2E manualmente).