Data Pipeline Engineering

┌────────────┐    ┌────────────┐    ┌────────────┐    ┌────────────┐
│  Extract   │───▶│ Transform  │───▶│   Load     │───▶│  Monitor   │
│            │    │            │    │            │    │            │
│ API Fetch  │    │ Validate   │    │ JPA Upsert │    │ Metrics    │
│ XML/JSON   │    │ Map Fields │    │ Batch Save │    │ Alerts     │
│ Pagination │    │ Enrich     │    │ Index      │    │ Dashboards │
└────────────┘    └────────────┘    └────────────┘    └────────────┘

@Configuration
public class RiksdagPipelineConfig {

    @Bean
    public IntegrationFlow riksdagImportFlow() {
        return IntegrationFlow
            .from(pollingSource(), e -> e.poller(
                Pollers.cron("0 0 2 * * *")     // Daily at 2 AM
                    .maxMessagesPerPoll(1)
                    .errorHandler(pipelineErrorHandler())))
            .channel("riksdagRawChannel")
            .transform(xmlToJsonTransformer())
            .split(personListSplitter())
            .channel(c -> c.executor(taskExecutor()))
            .filter(dataQualityFilter())
            .transform(entityMapper())
            .aggregate(batchAggregator())
            .handle(jpaOutboundAdapter())
            .get();
    }

    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(1000);
        executor.setThreadNamePrefix("riksdag-pipeline-");
        return executor;
    }
}

@Bean
public IntegrationFlow errorFlow() {
    return IntegrationFlow
        .from("errorChannel")
        .handle(message -> {
            MessagingException exception = (MessagingException) message.getPayload();
            LOG.error("Pipeline error: {}", exception.getMessage(), exception);
            metricsService.incrementErrorCount("riksdag-pipeline");

            // Route to dead letter queue for manual review
            Message<?> failedMessage = exception.getFailedMessage();
            deadLetterRepository.save(new DeadLetterEntry(
                failedMessage.getPayload().toString(),
                exception.getMessage(),
                Instant.now()
            ));
        })
        .get();
}

@Configuration
public class VoteImportBatchConfig {

    @Bean
    public Job voteImportJob(JobRepository jobRepository, Step importStep) {
        return new JobBuilder("voteImportJob", jobRepository)
            .incrementer(new RunIdIncrementer())
            .start(importStep)
            .build();
    }

    @Bean
    public Step importStep(JobRepository jobRepository,
                           PlatformTransactionManager txManager) {
        return new StepBuilder("importVotes", jobRepository)
            .<RiksdagVote, VoteData>chunk(100, txManager)
            .reader(voteReader())
            .processor(voteProcessor())
            .writer(voteWriter())
            .faultTolerant()
            .retryLimit(3)
            .retry(TransientDataAccessException.class)
            .skipLimit(10)
            .skip(DataValidationException.class)
            .listener(stepListener())
            .build();
    }
}

@Component
public class RiksdagEntityMapper {

    public PersonData mapPerson(RiksdagPerson source) {
        PersonData target = new PersonData();
        target.setId(source.getIntressentId());
        target.setFirstName(sanitize(source.getFornamn()));
        target.setLastName(sanitize(source.getEfternamn()));
        target.setParty(normalizeParty(source.getParti()));
        target.setBornYear(parseYear(source.getFoddAr()));
        target.setGender(normalizeGender(source.getKon()));
        target.setStatus(source.getStatus());
        target.setImportTimestamp(Instant.now());
        return target;
    }

    private String sanitize(String input) {
        if (input == null) return null;
        String trimmed = input.trim().replaceAll("[\\p{Cntrl}]", "");  // Remove control characters
        return trimmed.substring(0, Math.min(trimmed.length(), 255));
    }

    private String normalizeParty(String party) {
        return Optional.ofNullable(party)
            .map(String::trim)
            .map(String::toUpperCase)
            .orElse("-");
    }
}

@Component
public class PipelineScheduler {

    @Scheduled(cron = "0 0 2 * * *")   // Daily full refresh
    public void dailyFullImport() {
        LOG.info("Starting daily full import");
        importService.importAll();
    }

    @Scheduled(cron = "0 */15 8-18 * * MON-FRI")  // Every 15 min during sessions
    public void incrementalVoteImport() {
        if (riksdagSessionActive()) {
            LOG.info("Starting incremental vote import");
            importService.importRecentVotes();
        }
    }
}

@Component
public class PipelineMetrics {

    private final MeterRegistry meterRegistry;

    public void recordImport(String pipeline, int recordCount, long durationMs) {
        meterRegistry.counter("pipeline.records.imported",
            "pipeline", pipeline).increment(recordCount);
        meterRegistry.timer("pipeline.duration",
            "pipeline", pipeline).record(durationMs, TimeUnit.MILLISECONDS);
    }

    public void recordError(String pipeline, String errorType) {
        meterRegistry.counter("pipeline.errors",
            "pipeline", pipeline,
            "error_type", errorType).increment();
    }
}