class DataForgePipeline:
    """Flow: S3 raw files → Extract → Validate/Transform → Load (Redshift) → DLQ"""

    def run(self) -> RunStats:
        stats, started = RunStats(), time.monotonic()

        for idx, batch in enumerate(self.extractor.extract(batch_size=self.config.batch_size)):
            stats.batches_total  += 1
            stats.rows_extracted += len(batch)
            try:
                clean, rejected = self.transformer.transform(batch)  # validate + coerce
                stats.rows_rejected += len(rejected)

                if rejected:
                    self._send_to_dlq(rejected, reason="validation_failure")

                with_retry(                          # exponential backoff
                    lambda: self.loader.load(clean),
                    max_attempts=self.config.max_retries,
                )
                stats.rows_loaded += len(clean)

            except Exception as exc:
                stats.batches_failed += 1
                self._send_to_dlq(batch, reason=exc)   # whole batch → DLQ

        stats.duration_sec = time.monotonic() - started
        return stats   # rows_extracted, loaded, rejected, success_rate, duration