refactor: build records upfront with fail-fast, use iterators

- Move events instead of borrowing
- Build PutRecordsRequestEntry upfront with fail-fast on errors
- Use flat_map for cleaner stream/chunk handling

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: psteinroe

src/sink/kinesis.rs

@@ -171,62 +171,67 @@ impl Sink for KinesisSink {
             return Ok(());
         }
 
-        // Group events by stream name for batch sending.
-        let mut events_by_stream: HashMap<String, Vec<&TriggeredEvent>> = HashMap::new();
-
-        for event in &events {
-            let stream_name = self.resolve_stream_name(event).ok_or_else(|| {
-                etl::etl_error!(
-                    etl::error::ErrorKind::ConfigError,
-                    "No stream name configured",
-                    "Stream name must be provided in sink config or event metadata"
-                )
-            })?;
-
-            events_by_stream
-                .entry(stream_name.to_string())
-                .or_default()
-                .push(event);
-        }
-
-        // Prepare all batch requests.
-        let mut batch_futures = Vec::new();
-
-        for (stream_name, stream_events) in events_by_stream {
-            // Process in chunks of KINESIS_MAX_BATCH_SIZE (Kinesis limit is 500 per batch).
-            for chunk in stream_events.chunks(KINESIS_MAX_BATCH_SIZE) {
-                let mut records = Vec::with_capacity(chunk.len());
-                let chunk_len = chunk.len();
-
-                for event in chunk {
-                    let data = serde_json::to_vec(&event.payload).map_err(|e| {
+        // Build records upfront (fail fast on config/serialization errors).
+        let records_with_stream: Vec<_> = events
+            .into_iter()
+            .map(|event| {
+                let stream_name = self.resolve_stream_name(&event).ok_or_else(|| {
+                    etl::etl_error!(
+                        etl::error::ErrorKind::ConfigError,
+                        "No stream name configured",
+                        "Stream name must be provided in sink config or event metadata"
+                    )
+                })?;
+
+                let data = serde_json::to_vec(&event.payload).map_err(|e| {
+                    etl::etl_error!(
+                        etl::error::ErrorKind::InvalidData,
+                        "Failed to serialize payload to JSON",
+                        e.to_string()
+                    )
+                })?;
+
+                let record = PutRecordsRequestEntry::builder()
+                    .partition_key(event.id.id)
+                    .data(aws_sdk_kinesis::primitives::Blob::new(data))
+                    .build()
+                    .map_err(|e| {
                         etl::etl_error!(
                             etl::error::ErrorKind::InvalidData,
-                            "Failed to serialize payload to JSON",
+                            "Failed to build record entry",
                             e.to_string()
                         )
                     })?;
 
-                    // Use event ID as partition key.
-                    let record = PutRecordsRequestEntry::builder()
-                        .partition_key(&event.id.id)
-                        .data(aws_sdk_kinesis::primitives::Blob::new(data))
-                        .build()
-                        .map_err(|e| {
-                            etl::etl_error!(
-                                etl::error::ErrorKind::InvalidData,
-                                "Failed to build record entry",
-                                e.to_string()
-                            )
-                        })?;
+                Ok((stream_name.to_string(), record))
+            })
+            .collect::<EtlResult<Vec<_>>>()?;
 
-                    records.push(record);
-                }
+        // Group records by stream name.
+        let mut records_by_stream: HashMap<String, Vec<PutRecordsRequestEntry>> = HashMap::new();
+        for (stream_name, record) in records_with_stream {
+            records_by_stream
+                .entry(stream_name)
+                .or_default()
+                .push(record);
+        }
 
-                // Create future for this batch.
-                let client = self.client.clone();
-                let stream_name = stream_name.clone();
-                batch_futures.push(async move {
+        // Create batch futures for all streams and chunks.
+        let batch_futures: Vec<_> = records_by_stream
+            .into_iter()
+            .flat_map(|(stream_name, records)| {
+                records
+                    .into_iter()
+                    .collect::<Vec<_>>()
+                    .chunks(KINESIS_MAX_BATCH_SIZE)
+                    .map(|chunk| chunk.to_vec())
+                    .map(|chunk| (stream_name.clone(), chunk))
+                    .collect::<Vec<_>>()
+            })
+            .map(|(stream_name, records)| {
+                let client = &self.client;
+                let chunk_len = records.len();
+                async move {
                     let result = client
                         .put_records()
                         .stream_name(&stream_name)
@@ -244,7 +249,6 @@ impl Sink for KinesisSink {
                     // Check for partial failures.
                     let failed_count = result.failed_record_count.unwrap_or(0);
                     if failed_count > 0 {
-                        // Find first failed record for error details.
                         let first_error = result
                             .records
                             .iter()
@@ -269,9 +273,9 @@ impl Sink for KinesisSink {
                     }
 
                     Ok(())
-                });
-            }
-        }
+                }
+            })
+            .collect();
 
         // Send all batches concurrently.
         try_join_all(batch_futures).await?;