refactor: move events by value instead of borrowing

Use into_iter() and drain() to consume events by value rather than
borrowing. Events are now moved into the grouped structure and
drained into batch chunks, freeing memory as they're processed.

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

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

Author: psteinroe

src/sink/sqs.rs

@@ -172,10 +172,10 @@ impl Sink for SqsSink {
         }
 
         // Group events by queue URL for batch sending.
-        let mut events_by_queue: HashMap<String, Vec<(usize, &TriggeredEvent)>> = HashMap::new();
+        let mut events_by_queue: HashMap<String, Vec<(usize, TriggeredEvent)>> = HashMap::new();
 
-        for (idx, event) in events.iter().enumerate() {
-            let queue_url = self.resolve_queue_url(event).ok_or_else(|| {
+        for (idx, event) in events.into_iter().enumerate() {
+            let queue_url = self.resolve_queue_url(&event).ok_or_else(|| {
                 etl::etl_error!(
                     etl::error::ErrorKind::ConfigError,
                     "No queue URL configured",
@@ -192,9 +192,12 @@ impl Sink for SqsSink {
         // Prepare all batch requests.
         let mut batch_futures = Vec::new();
 
-        for (queue_url, queue_events) in events_by_queue {
+        for (queue_url, mut queue_events) in events_by_queue {
             // Process in chunks of SQS_MAX_BATCH_SIZE (SQS limit is 10 per batch).
-            for chunk in queue_events.chunks(SQS_MAX_BATCH_SIZE) {
+            while !queue_events.is_empty() {
+                let chunk_size = queue_events.len().min(SQS_MAX_BATCH_SIZE);
+                let chunk: Vec<_> = queue_events.drain(..chunk_size).collect();
+
                 let mut entries = Vec::with_capacity(chunk.len());
 
                 for (idx, event) in chunk {