Import: Process Implementation

┌─────────────────────┐
│   ImportResource    │  REST API
│   (Controller)      │
└──────────┬──────────┘
           │
┌──────────▼──────────┐
│ ImportProcess       │  Orchestrator Service
│ Orchestrator        │
└──────────┬──────────┘
           │
┌──────────▼──────────┐
│  ImportController   │  Workflow Controller (like FormController)
│                     │
│  - processInstance  │
│  - stepHandlers     │
│  - next()           │
└──────────┬──────────┘
           │
    ┌──────┴──────┐
    │             │
┌───▼───┐   ┌─────▼─────┐   ┌────────────┐
│Column │   │   Cell    │   │    Row     │
│Mapping│   │  Mapping  │   │ Processing │
│Handler│   │  Handler  │   │  Handler   │
└───────┘   └───────────┘   └────────────┘

CREATE TABLE import_job (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,

    -- Import type discriminator
    import_type VARCHAR(20) NOT NULL,  -- EVENT_PARTICIPANT, MEMBERSHIP, RESULT

    -- Current workflow state
    current_step VARCHAR(20) NOT NULL,  -- UPLOADED, COLUMN_MAPPING, etc.

    -- File storage (FK to existing Attachment entity)
    attachment_id BIGINT NOT NULL,

    -- Organisation scope
    organisation_id BIGINT NOT NULL,

    -- Generic context reference (event, period, or race ID based on import_type)
    context_id BIGINT NOT NULL,

    -- Progress tracking
    total_rows INT,
    processed_rows INT DEFAULT 0,
    success_count INT DEFAULT 0,
    error_count INT DEFAULT 0,

    -- Timestamps
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP,
    completed_at TIMESTAMP,
    expire_at TIMESTAMP,

    -- Foreign keys
    CONSTRAINT fk_import_attachment FOREIGN KEY (attachment_id) REFERENCES attachment(id),
    CONSTRAINT fk_import_org FOREIGN KEY (organisation_id) REFERENCES organisation(id)
);

public interface ImportStepHandler<I, O> {

    /**
     * Execute this step's logic.
     * @param context Import context with job data and services
     * @param input Input data for this step (e.g., submitted mappings)
     * @return Result indicating success, needs input, or error
     */
    StepResult<O> execute(ImportContext context, I input);

    /**
     * Check if this step can proceed without user input.
     */
    boolean canProceed(ImportContext context);

    /**
     * Get pending items requiring user resolution.
     */
    List<?> getPendingItems(ImportContext context);
}

public class StepResult<T> {

    public enum Outcome {
        COMPLETED,    // Step finished successfully
        NEEDS_INPUT,  // Waiting for user to provide mappings
        ERROR         // Fatal error occurred
    }

    private Outcome outcome;
    private T output;
    private String message;
    private List<?> pendingItems;

    // Static factory methods
    public static <T> StepResult<T> completed(T output) { ... }
    public static <T> StepResult<T> needsInput(List<?> pending) { ... }
    public static <T> StepResult<T> error(String message) { ... }
}

public class ImportContext {

    private ImportJob job;
    private SpreadsheetReader reader;
    private Organisation organisation;

    // Cached reference data for FK resolution
    private List<EventCategory> eventCategories;
    private List<Country> countries;
    private List<CustomListValue> customLists;

    // Services (injected via orchestrator)
    private EventParticipantServiceEx eventParticipantService;
    private MembershipImportService membershipService;
    private RaceResultServiceEx resultService;

    // Utility methods
    public Long getContextId() {
        return job.getContextId();
    }

    public ImportType getImportType() {
        return job.getImportType();
    }
}

@Component
public class ColumnMappingHandler implements ImportStepHandler<Void, List<ImportColumnMapping>> {

    private final ImportColumnMappingRepository mappingRepository;

    @Override
    public StepResult<List<ImportColumnMapping>> execute(ImportContext context, Void input) {
        ImportJob job = context.getJob();
        SpreadsheetReader reader = context.getReader();

        // Get header row
        List<String> headers = reader.getHeaders();

        // Match each header against dictionary
        List<ImportColumnMapping> mappings = new ArrayList<>();
        for (int i = 0; i < headers.size(); i++) {
            ImportColumnMapping mapping = matchHeader(headers.get(i), i, job);
            mappings.add(mapping);
        }

        // Save mappings
        mappingRepository.saveAll(mappings);

        // Check if all required columns matched
        boolean allRequiredMapped = mappings.stream()
            .filter(ImportColumnMapping::isRequired)
            .allMatch(m -> m.getStatus() != MappingStatus.UNMATCHED);

        if (allRequiredMapped) {
            return StepResult.completed(mappings);
        } else {
            return StepResult.needsInput(getUnmatchedRequired(mappings));
        }
    }

    @Override
    public boolean canProceed(ImportContext context) {
        // Check if mappings exist and all required are matched
        List<ImportColumnMapping> mappings = getMappings(context.getJob());
        return mappings.stream()
            .filter(ImportColumnMapping::isRequired)
            .allMatch(m -> m.getStatus() != MappingStatus.UNMATCHED);
    }

    @Override
    public List<?> getPendingItems(ImportContext context) {
        return getUnmatchedRequired(getMappings(context.getJob()));
    }
}

@Component
public class CellMappingHandler implements ImportStepHandler<Void, List<ImportCellMapping>> {

    @Override
    public StepResult<List<ImportCellMapping>> execute(ImportContext context, Void input) {
        ImportJob job = context.getJob();

        // Get FK columns that need resolution
        List<ImportColumnMapping> fkColumns = getForeignKeyColumns(job);

        // Scan spreadsheet for unique values in FK columns
        Map<String, Set<String>> uniqueValues = scanForUniqueValues(context, fkColumns);

        // Try to resolve each value
        List<ImportCellMapping> cellMappings = new ArrayList<>();
        for (var entry : uniqueValues.entrySet()) {
            String field = entry.getKey();
            for (String value : entry.getValue()) {
                ImportCellMapping mapping = resolveValue(context, field, value);
                cellMappings.add(mapping);
            }
        }

        // Save mappings
        cellMappingRepository.saveAll(cellMappings);

        // Check if all resolved
        boolean allResolved = cellMappings.stream()
            .allMatch(m -> m.getStatus() != MappingStatus.UNMATCHED);

        if (allResolved) {
            return StepResult.completed(cellMappings);
        } else {
            return StepResult.needsInput(getUnresolved(cellMappings));
        }
    }
}

@Component
public class RowProcessingHandler implements ImportStepHandler<Void, ImportJobDTO> {

    @Override
    @Async("importTaskExecutor")
    public StepResult<ImportJobDTO> execute(ImportContext context, Void input) {
        ImportJob job = context.getJob();
        SpreadsheetReader reader = context.getReader();

        // Get resolved mappings
        Map<Integer, String> columnMap = getColumnMap(job);
        Map<String, Map<String, Long>> cellMap = getCellMap(job);

        // Process each row
        int rowNum = 1;  // Skip header
        while (reader.hasNext()) {
            Row row = reader.next();

            try {
                ImportRowResult result = processRow(context, row, columnMap, cellMap);
                result.setRowNumber(rowNum);
                rowResultRepository.save(result);

                if (result.getOutcome() == ImportOutcome.CREATED ||
                    result.getOutcome() == ImportOutcome.UPDATED) {
                    job.incrementSuccess();
                } else {
                    job.incrementError();
                }
            } catch (UnmappedCellException e) {
                // New unmapped value discovered - need to pause
                savePendingCellMapping(job, e);
                job.setCurrentStep(ImportStep.CELL_MAPPING);
                jobRepository.save(job);
                return StepResult.needsInput(List.of(e.getMapping()));
            } catch (Exception e) {
                saveErrorResult(job, rowNum, e);
                job.incrementError();
            }

            job.incrementProcessed();
            if (rowNum % 100 == 0) {
                jobRepository.save(job);  // Periodic progress save
            }
            rowNum++;
        }

        return StepResult.completed(toDTO(job));
    }
}

@Configuration
@EnableAsync
public class ImportAsyncConfig {

    @Bean(name = "importTaskExecutor")
    public Executor importTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(25);
        executor.setThreadNamePrefix("import-");
        executor.setRejectedExecutionHandler(new CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

@Service
public class ImportJobCleanupService {

    @Scheduled(cron = "0 0 2 * * ?")  // Daily at 2 AM
    @Transactional
    public void cleanupExpiredImports() {
        Instant now = Instant.now();

        // Terminal states: 7 days
        deleteByStatusesAndDate(
            Set.of(COMPLETED, FAILED, CANCELLED),
            now.minus(7, ChronoUnit.DAYS)
        );

        // Abandoned: 90 days
        deleteByStatusesAndDate(
            Set.of(UPLOADED, COLUMN_MAPPING, CELL_MAPPING),
            now.minus(90, ChronoUnit.DAYS)
        );
    }
}