Skip to main content

bnto_spreadsheet/
merge.rs

1// Merge CSV Node — combine multiple CSV files into one.
2//
3// Batch processor: receives all files at once, merges rows into a single
4// output CSV. Supports header reconciliation (first-file vs union) and
5// optional deduplication of rows across files.
6
7use bnto_core::context::ProcessContext;
8use bnto_core::errors::BntoError;
9use bnto_core::processor::{
10    BatchInput, FileData, NodeInput, NodeOutput, NodeProcessor, OutputFile,
11};
12use bnto_core::progress::ProgressReporter;
13
14/// The spreadsheet-merge node processor. Stateless — config comes from params.
15pub struct MergeSpreadsheets;
16
17impl MergeSpreadsheets {
18    pub fn new() -> Self {
19        Self
20    }
21}
22
23impl Default for MergeSpreadsheets {
24    fn default() -> Self {
25        Self::new()
26    }
27}
28
29// --- NodeProcessor Implementation ---
30
31impl NodeProcessor for MergeSpreadsheets {
32    fn name(&self) -> &str {
33        "spreadsheet-merge"
34    }
35
36    fn metadata(&self) -> bnto_core::NodeMetadata {
37        use bnto_core::metadata::*;
38        NodeMetadata {
39            node_type: "spreadsheet-merge".to_string(),
40            name: "Merge CSV".to_string(),
41            description:
42                "Combine multiple CSV files into one, with header reconciliation and deduplication."
43                    .to_string(),
44            category: NodeCategory::Spreadsheet,
45            accepts: vec!["text/csv".to_string()],
46            platforms: vec!["browser".to_string()],
47            parameters: build_merge_parameters(),
48            input_cardinality: InputCardinality::Batch,
49            requires: vec![],
50        }
51    }
52
53    /// Single-file process — not the primary path for a batch processor, but
54    /// required by the trait. Returns the file unchanged.
55    fn process(
56        &self,
57        input: NodeInput,
58        _progress: &ProgressReporter,
59        _ctx: &dyn ProcessContext,
60    ) -> Result<NodeOutput, BntoError> {
61        Ok(NodeOutput {
62            files: vec![OutputFile {
63                data: input.data,
64                filename: input.filename,
65                mime_type: input.mime_type.unwrap_or_else(|| "text/csv".to_string()),
66                metadata: serde_json::Map::new(),
67            }],
68            metadata: serde_json::Map::new(),
69        })
70    }
71
72    /// Merge all input CSV files into a single output.
73    fn process_batch(
74        &self,
75        input: BatchInput,
76        progress: &ProgressReporter,
77        _ctx: &dyn ProcessContext,
78    ) -> Result<NodeOutput, BntoError> {
79        if input.files.is_empty() {
80            return Err(BntoError::InvalidInput(
81                "No files to merge — need at least one CSV file".to_string(),
82            ));
83        }
84
85        let config = MergeConfig::from_params(&input.params);
86
87        progress.report(0, "Parsing CSV files...");
88        let parsed = parse_all_files(&input)?;
89
90        progress.report(30, "Reconciling headers...");
91        let merged_headers = reconcile_headers(&parsed, &config);
92
93        progress.report(50, "Merging rows...");
94        let mut all_rows = collect_all_rows(&parsed, &merged_headers)?;
95        let total_input_rows = all_rows.len();
96
97        progress.report(70, "Deduplicating...");
98        let duplicates_removed = deduplicate_rows(&mut all_rows, config.deduplicate);
99
100        progress.report(90, "Writing merged CSV...");
101        let output_bytes = write_csv_output(&merged_headers, &all_rows)?;
102
103        let metadata = build_merge_metadata(
104            input.files.len(),
105            total_input_rows,
106            all_rows.len(),
107            duplicates_removed,
108        );
109
110        progress.report(100, "Done!");
111        Ok(NodeOutput {
112            files: vec![OutputFile {
113                data: FileData::Bytes(output_bytes),
114                filename: "merged.csv".to_string(),
115                mime_type: "text/csv".to_string(),
116                metadata: serde_json::Map::new(),
117            }],
118            metadata,
119        })
120    }
121}
122
123// --- Configuration ---
124
125struct MergeConfig {
126    header_handling: HeaderHandling,
127    deduplicate: bool,
128}
129
130enum HeaderHandling {
131    /// Use headers from the first file. Other files' rows are mapped by position.
132    FirstFile,
133    /// Union of all headers across files. Missing columns get empty values.
134    Union,
135}
136
137impl MergeConfig {
138    fn from_params(params: &serde_json::Map<String, serde_json::Value>) -> Self {
139        let header_handling = match params.get("headerHandling").and_then(|v| v.as_str()) {
140            Some("union") => HeaderHandling::Union,
141            _ => HeaderHandling::FirstFile,
142        };
143
144        let deduplicate = params
145            .get("deduplicate")
146            .and_then(|v| v.as_bool())
147            .unwrap_or(false);
148
149        Self {
150            header_handling,
151            deduplicate,
152        }
153    }
154}
155
156// --- Metadata Parameter Definitions ---
157
158fn build_merge_parameters() -> Vec<bnto_core::metadata::ParameterDef> {
159    use bnto_core::metadata::*;
160    vec![
161        ParameterDef {
162            name: "headerHandling".to_string(),
163            label: "Header Handling".to_string(),
164            description: "How to reconcile headers across files".to_string(),
165            param_type: ParameterType::Enum {
166                options: vec![
167                    OptionEntry {
168                        value: "first-file".to_string(),
169                        label: "First File".to_string(),
170                    },
171                    OptionEntry {
172                        value: "union".to_string(),
173                        label: "Union".to_string(),
174                    },
175                ],
176            },
177            default: Some(serde_json::json!("first-file")),
178            ..Default::default()
179        },
180        ParameterDef {
181            name: "deduplicate".to_string(),
182            label: "Remove Duplicates".to_string(),
183            description: "Remove duplicate rows across all files".to_string(),
184            param_type: ParameterType::Boolean,
185            default: Some(serde_json::json!(false)),
186            ..Default::default()
187        },
188    ]
189}
190
191// --- Parsed File ---
192
193struct ParsedCsv {
194    headers: Vec<String>,
195    rows: Vec<Vec<String>>,
196}
197
198/// Parse all batch files into header+rows structures.
199fn parse_all_files(input: &BatchInput) -> Result<Vec<ParsedCsv>, BntoError> {
200    let mut parsed = Vec::with_capacity(input.files.len());
201
202    for file in &input.files {
203        let csv_text = std::str::from_utf8(&file.data).map_err(|e| {
204            BntoError::InvalidInput(format!("File '{}' is not valid UTF-8: {e}", file.filename))
205        })?;
206
207        if csv_text.trim().is_empty() {
208            return Err(BntoError::InvalidInput(format!(
209                "File '{}' is empty",
210                file.filename
211            )));
212        }
213
214        let mut reader = csv::ReaderBuilder::new()
215            .has_headers(true)
216            .flexible(true)
217            .from_reader(csv_text.as_bytes());
218
219        let headers: Vec<String> = reader
220            .headers()
221            .map_err(|e| {
222                BntoError::ProcessingFailed(format!(
223                    "Failed to read headers from '{}': {e}",
224                    file.filename
225                ))
226            })?
227            .iter()
228            .map(|h| h.trim().to_string())
229            .collect();
230
231        let mut rows = Vec::new();
232        for result in reader.records() {
233            let record = match result {
234                Ok(rec) => rec,
235                Err(_) => continue,
236            };
237            let row: Vec<String> = record.iter().map(|c| c.trim().to_string()).collect();
238            rows.push(row);
239        }
240
241        parsed.push(ParsedCsv { headers, rows });
242    }
243
244    Ok(parsed)
245}
246
247// --- Header Reconciliation ---
248
249/// Determine the merged header set based on the config strategy.
250fn reconcile_headers(parsed: &[ParsedCsv], config: &MergeConfig) -> Vec<String> {
251    match config.header_handling {
252        HeaderHandling::FirstFile => parsed
253            .first()
254            .map(|p| p.headers.clone())
255            .unwrap_or_default(),
256        HeaderHandling::Union => {
257            let mut seen = std::collections::HashSet::new();
258            let mut headers = Vec::new();
259            for file in parsed {
260                for header in &file.headers {
261                    if seen.insert(header.clone()) {
262                        headers.push(header.clone());
263                    }
264                }
265            }
266            headers
267        }
268    }
269}
270
271// --- Row Collection ---
272
273/// Map each file's rows to the merged header order.
274///
275/// For union mode: maps by header name (missing columns → empty).
276/// For first-file mode: if headers match by name, map by name. Otherwise
277/// fall back to positional mapping (column 0 → column 0, etc.).
278fn collect_all_rows(
279    parsed: &[ParsedCsv],
280    merged_headers: &[String],
281) -> Result<Vec<Vec<String>>, BntoError> {
282    let mut all_rows = Vec::new();
283
284    for file in parsed {
285        // Check if this file's headers overlap with merged headers by name.
286        let has_name_match = file.headers.iter().any(|h| merged_headers.contains(h));
287
288        for row in &file.rows {
289            let mapped: Vec<String> = if has_name_match {
290                // Map by header name
291                let col_map: Vec<Option<usize>> = merged_headers
292                    .iter()
293                    .map(|h| file.headers.iter().position(|fh| fh == h))
294                    .collect();
295                col_map
296                    .iter()
297                    .map(|idx| idx.and_then(|i| row.get(i)).cloned().unwrap_or_default())
298                    .collect()
299            } else {
300                // No header overlap — map by position, pad/truncate to match
301                let mut positional: Vec<String> = merged_headers
302                    .iter()
303                    .enumerate()
304                    .map(|(i, _)| row.get(i).cloned().unwrap_or_default())
305                    .collect();
306                positional.truncate(merged_headers.len());
307                positional
308            };
309            all_rows.push(mapped);
310        }
311    }
312
313    Ok(all_rows)
314}
315
316// --- Deduplication ---
317
318/// Remove duplicate rows in-place. Returns count removed.
319fn deduplicate_rows(rows: &mut Vec<Vec<String>>, enabled: bool) -> usize {
320    if !enabled {
321        return 0;
322    }
323
324    let mut seen = std::collections::HashSet::new();
325    let before = rows.len();
326    rows.retain(|row| seen.insert(row.join("\0")));
327    before - rows.len()
328}
329
330// --- CSV Output ---
331
332fn write_csv_output(headers: &[String], rows: &[Vec<String>]) -> Result<Vec<u8>, BntoError> {
333    let mut writer = csv::WriterBuilder::new().from_writer(Vec::new());
334
335    writer
336        .write_record(headers)
337        .map_err(|e| BntoError::ProcessingFailed(format!("Failed to write CSV headers: {e}")))?;
338
339    for row in rows {
340        writer
341            .write_record(row)
342            .map_err(|e| BntoError::ProcessingFailed(format!("Failed to write CSV row: {e}")))?;
343    }
344
345    writer
346        .into_inner()
347        .map_err(|e| BntoError::ProcessingFailed(format!("Failed to finalize CSV output: {e}")))
348}
349
350// --- Result Metadata ---
351
352fn build_merge_metadata(
353    files_merged: usize,
354    total_input_rows: usize,
355    output_rows: usize,
356    duplicates_removed: usize,
357) -> serde_json::Map<String, serde_json::Value> {
358    let mut metadata = serde_json::Map::new();
359    metadata.insert(
360        "filesMerged".to_string(),
361        serde_json::Value::Number(files_merged.into()),
362    );
363    metadata.insert(
364        "totalInputRows".to_string(),
365        serde_json::Value::Number(total_input_rows.into()),
366    );
367    metadata.insert(
368        "outputRows".to_string(),
369        serde_json::Value::Number(output_rows.into()),
370    );
371    metadata.insert(
372        "duplicatesRemoved".to_string(),
373        serde_json::Value::Number(duplicates_removed.into()),
374    );
375    metadata
376}
377
378// =============================================================================
379// Tests
380// =============================================================================
381
382#[cfg(test)]
383mod tests {
384    use super::*;
385    use bnto_core::NoopContext;
386    use bnto_core::processor::BatchFile;
387
388    // --- Test Helpers ---
389
390    fn make_batch_input(files: Vec<(&str, &str)>) -> BatchInput {
391        BatchInput {
392            files: files
393                .into_iter()
394                .map(|(name, content)| BatchFile {
395                    data: content.as_bytes().to_vec(),
396                    filename: name.to_string(),
397                    mime_type: Some("text/csv".to_string()),
398                })
399                .collect(),
400            params: serde_json::Map::new(),
401        }
402    }
403
404    fn make_batch_with_params(
405        files: Vec<(&str, &str)>,
406        params: serde_json::Map<String, serde_json::Value>,
407    ) -> BatchInput {
408        BatchInput {
409            files: files
410                .into_iter()
411                .map(|(name, content)| BatchFile {
412                    data: content.as_bytes().to_vec(),
413                    filename: name.to_string(),
414                    mime_type: Some("text/csv".to_string()),
415                })
416                .collect(),
417            params,
418        }
419    }
420
421    fn output_csv_text(output: &NodeOutput) -> String {
422        let bytes = output.files[0]
423            .data
424            .clone()
425            .into_bytes()
426            .expect("Should read bytes");
427        String::from_utf8(bytes).expect("Output should be valid UTF-8")
428    }
429
430    fn count_data_rows(csv_text: &str) -> usize {
431        let mut reader = csv::ReaderBuilder::new()
432            .has_headers(true)
433            .flexible(true)
434            .from_reader(csv_text.as_bytes());
435        reader.records().count()
436    }
437
438    // --- Trait Tests ---
439
440    #[test]
441    fn test_name_returns_merge_csv() {
442        let processor = MergeSpreadsheets::new();
443        assert_eq!(processor.name(), "spreadsheet-merge");
444    }
445
446    #[test]
447    fn test_metadata_has_batch_cardinality() {
448        let processor = MergeSpreadsheets::new();
449        let meta = processor.metadata();
450        assert_eq!(
451            meta.input_cardinality,
452            bnto_core::metadata::InputCardinality::Batch
453        );
454        assert_eq!(meta.node_type, "spreadsheet-merge");
455    }
456
457    #[test]
458    fn test_metadata_has_two_parameters() {
459        let processor = MergeSpreadsheets::new();
460        let meta = processor.metadata();
461        assert_eq!(meta.parameters.len(), 2);
462        let names: Vec<&str> = meta.parameters.iter().map(|p| p.name.as_str()).collect();
463        assert!(names.contains(&"headerHandling"));
464        assert!(names.contains(&"deduplicate"));
465    }
466
467    #[test]
468    fn test_default_creates_same_as_new() {
469        #[allow(clippy::default_constructed_unit_structs)]
470        let _processor = MergeSpreadsheets::default();
471    }
472
473    #[test]
474    fn test_validate_passes_with_no_params() {
475        let processor = MergeSpreadsheets::new();
476        let errors = processor.validate(&serde_json::Map::new());
477        assert!(errors.is_empty());
478    }
479
480    // --- Basic Merge Tests ---
481
482    #[test]
483    fn test_merge_two_files_same_headers() {
484        let processor = MergeSpreadsheets::new();
485        let progress = ProgressReporter::new_noop();
486        let input = make_batch_input(vec![
487            ("a.csv", "name,age\nAlice,30\nBob,25\n"),
488            ("b.csv", "name,age\nCharlie,35\nDiana,28\n"),
489        ]);
490
491        let output = processor
492            .process_batch(input, &progress, &NoopContext)
493            .unwrap();
494        let text = output_csv_text(&output);
495        let rows = count_data_rows(&text);
496
497        assert_eq!(rows, 4, "Should merge all 4 rows");
498        assert!(text.contains("Alice"));
499        assert!(text.contains("Bob"));
500        assert!(text.contains("Charlie"));
501        assert!(text.contains("Diana"));
502        assert_eq!(output.files[0].filename, "merged.csv");
503        assert_eq!(output.files[0].mime_type, "text/csv");
504    }
505
506    #[test]
507    fn test_merge_single_file() {
508        let processor = MergeSpreadsheets::new();
509        let progress = ProgressReporter::new_noop();
510        let input = make_batch_input(vec![("a.csv", "name,age\nAlice,30\n")]);
511
512        let output = processor
513            .process_batch(input, &progress, &NoopContext)
514            .unwrap();
515        let text = output_csv_text(&output);
516        let rows = count_data_rows(&text);
517
518        assert_eq!(rows, 1);
519        assert!(text.contains("Alice"));
520    }
521
522    #[test]
523    fn test_merge_three_files() {
524        let processor = MergeSpreadsheets::new();
525        let progress = ProgressReporter::new_noop();
526        let input = make_batch_input(vec![
527            ("a.csv", "id\n1\n2\n"),
528            ("b.csv", "id\n3\n4\n"),
529            ("c.csv", "id\n5\n"),
530        ]);
531
532        let output = processor
533            .process_batch(input, &progress, &NoopContext)
534            .unwrap();
535        let text = output_csv_text(&output);
536        let rows = count_data_rows(&text);
537
538        assert_eq!(rows, 5, "Should merge rows from all 3 files");
539    }
540
541    // --- Empty Input ---
542
543    #[test]
544    fn test_merge_empty_batch_returns_error() {
545        let processor = MergeSpreadsheets::new();
546        let progress = ProgressReporter::new_noop();
547        let input = BatchInput {
548            files: vec![],
549            params: serde_json::Map::new(),
550        };
551
552        let result = processor.process_batch(input, &progress, &NoopContext);
553        assert!(result.is_err());
554    }
555
556    #[test]
557    fn test_merge_empty_file_returns_error() {
558        let processor = MergeSpreadsheets::new();
559        let progress = ProgressReporter::new_noop();
560        let input = make_batch_input(vec![("a.csv", "")]);
561
562        let result = processor.process_batch(input, &progress, &NoopContext);
563        assert!(result.is_err());
564    }
565
566    // --- Header Handling: first-file (default) ---
567
568    #[test]
569    fn test_first_file_headers_columns_matched_by_position() {
570        let processor = MergeSpreadsheets::new();
571        let progress = ProgressReporter::new_noop();
572        // Second file has different header names — with first-file mode,
573        // rows map by column position, headers come from first file.
574        let input = make_batch_input(vec![
575            ("a.csv", "name,age\nAlice,30\n"),
576            ("b.csv", "nombre,edad\nCarlos,40\n"),
577        ]);
578
579        let output = processor
580            .process_batch(input, &progress, &NoopContext)
581            .unwrap();
582        let text = output_csv_text(&output);
583
584        // Headers should be from first file
585        assert!(text.starts_with("name,age"), "Headers from first file");
586        assert_eq!(count_data_rows(&text), 2);
587        assert!(text.contains("Carlos"));
588    }
589
590    // --- Header Handling: union ---
591
592    #[test]
593    fn test_union_headers_includes_all_columns() {
594        let processor = MergeSpreadsheets::new();
595        let progress = ProgressReporter::new_noop();
596        let mut params = serde_json::Map::new();
597        params.insert("headerHandling".to_string(), serde_json::json!("union"));
598        let input = make_batch_with_params(
599            vec![
600                ("a.csv", "name,age\nAlice,30\n"),
601                ("b.csv", "name,city\nBob,NYC\n"),
602            ],
603            params,
604        );
605
606        let output = processor
607            .process_batch(input, &progress, &NoopContext)
608            .unwrap();
609        let text = output_csv_text(&output);
610
611        // Union headers: name, age, city
612        let first_line = text.lines().next().unwrap();
613        assert!(first_line.contains("name"));
614        assert!(first_line.contains("age"));
615        assert!(first_line.contains("city"));
616
617        assert_eq!(count_data_rows(&text), 2);
618        // Alice should have empty city, Bob should have empty age
619        assert!(text.contains("Alice"));
620        assert!(text.contains("Bob"));
621    }
622
623    // --- Deduplication ---
624
625    #[test]
626    fn test_deduplicate_removes_cross_file_duplicates() {
627        let processor = MergeSpreadsheets::new();
628        let progress = ProgressReporter::new_noop();
629        let mut params = serde_json::Map::new();
630        params.insert("deduplicate".to_string(), serde_json::json!(true));
631        let input = make_batch_with_params(
632            vec![
633                ("a.csv", "name,age\nAlice,30\nBob,25\n"),
634                ("b.csv", "name,age\nAlice,30\nCharlie,35\n"),
635            ],
636            params,
637        );
638
639        let output = processor
640            .process_batch(input, &progress, &NoopContext)
641            .unwrap();
642        let text = output_csv_text(&output);
643        let rows = count_data_rows(&text);
644
645        assert_eq!(rows, 3, "Should have 3 unique rows (Alice, Bob, Charlie)");
646        let dupes = output
647            .metadata
648            .get("duplicatesRemoved")
649            .unwrap()
650            .as_u64()
651            .unwrap();
652        assert_eq!(dupes, 1);
653    }
654
655    #[test]
656    fn test_deduplicate_false_keeps_duplicates() {
657        let processor = MergeSpreadsheets::new();
658        let progress = ProgressReporter::new_noop();
659        let input = make_batch_input(vec![("a.csv", "name\nAlice\n"), ("b.csv", "name\nAlice\n")]);
660
661        let output = processor
662            .process_batch(input, &progress, &NoopContext)
663            .unwrap();
664        let text = output_csv_text(&output);
665        let rows = count_data_rows(&text);
666
667        assert_eq!(rows, 2, "Dedup off by default — keep both Alices");
668    }
669
670    // --- Parameterized: different params produce different output ---
671
672    #[test]
673    fn test_different_header_handling_produces_different_output() {
674        let processor = MergeSpreadsheets::new();
675        let progress = ProgressReporter::new_noop();
676
677        let files = vec![
678            ("a.csv", "name,age\nAlice,30\n"),
679            ("b.csv", "name,city\nBob,NYC\n"),
680        ];
681
682        // First-file mode
683        let input1 = make_batch_input(files.clone());
684        let out1 = processor
685            .process_batch(input1, &progress, &NoopContext)
686            .unwrap();
687
688        // Union mode
689        let mut params = serde_json::Map::new();
690        params.insert("headerHandling".to_string(), serde_json::json!("union"));
691        let input2 = make_batch_with_params(files, params);
692        let out2 = processor
693            .process_batch(input2, &progress, &NoopContext)
694            .unwrap();
695
696        // Outputs should differ — union has more columns
697        let bytes1 = out1.files[0].data.clone().into_bytes().unwrap();
698        let bytes2 = out2.files[0].data.clone().into_bytes().unwrap();
699        assert_ne!(bytes1, bytes2);
700    }
701
702    #[test]
703    fn test_different_deduplicate_produces_different_output() {
704        let processor = MergeSpreadsheets::new();
705        let progress = ProgressReporter::new_noop();
706
707        let files = vec![("a.csv", "name\nAlice\n"), ("b.csv", "name\nAlice\n")];
708
709        // Without dedup
710        let input1 = make_batch_input(files.clone());
711        let out1 = processor
712            .process_batch(input1, &progress, &NoopContext)
713            .unwrap();
714
715        // With dedup
716        let mut params = serde_json::Map::new();
717        params.insert("deduplicate".to_string(), serde_json::json!(true));
718        let input2 = make_batch_with_params(files, params);
719        let out2 = processor
720            .process_batch(input2, &progress, &NoopContext)
721            .unwrap();
722
723        assert_ne!(
724            count_data_rows(&output_csv_text(&out1)),
725            count_data_rows(&output_csv_text(&out2)),
726        );
727    }
728
729    // --- Metadata ---
730
731    #[test]
732    fn test_output_metadata_fields() {
733        let processor = MergeSpreadsheets::new();
734        let progress = ProgressReporter::new_noop();
735        let input = make_batch_input(vec![
736            ("a.csv", "name\nAlice\nBob\n"),
737            ("b.csv", "name\nCharlie\n"),
738        ]);
739
740        let output = processor
741            .process_batch(input, &progress, &NoopContext)
742            .unwrap();
743
744        assert_eq!(
745            output
746                .metadata
747                .get("filesMerged")
748                .unwrap()
749                .as_u64()
750                .unwrap(),
751            2
752        );
753        assert_eq!(
754            output
755                .metadata
756                .get("totalInputRows")
757                .unwrap()
758                .as_u64()
759                .unwrap(),
760            3
761        );
762        assert_eq!(
763            output.metadata.get("outputRows").unwrap().as_u64().unwrap(),
764            3
765        );
766        assert_eq!(
767            output
768                .metadata
769                .get("duplicatesRemoved")
770                .unwrap()
771                .as_u64()
772                .unwrap(),
773            0
774        );
775    }
776
777    // --- Error Handling ---
778
779    #[test]
780    fn test_non_utf8_file_returns_error() {
781        let processor = MergeSpreadsheets::new();
782        let progress = ProgressReporter::new_noop();
783        let input = BatchInput {
784            files: vec![BatchFile {
785                data: vec![0xFF, 0xFE, 0x00],
786                filename: "bad.csv".to_string(),
787                mime_type: Some("text/csv".to_string()),
788            }],
789            params: serde_json::Map::new(),
790        };
791
792        let result = processor.process_batch(input, &progress, &NoopContext);
793        assert!(result.is_err());
794        if let Err(e) = result {
795            assert!(e.to_string().contains("UTF-8"));
796        }
797    }
798
799    // --- Edge Cases ---
800
801    #[test]
802    fn test_merge_files_with_varying_column_counts() {
803        let processor = MergeSpreadsheets::new();
804        let progress = ProgressReporter::new_noop();
805        // First file has 2 cols, second has 3 — first-file mode uses 2 cols
806        let input = make_batch_input(vec![
807            ("a.csv", "name,age\nAlice,30\n"),
808            ("b.csv", "name,age,city\nBob,25,NYC\n"),
809        ]);
810
811        let output = processor
812            .process_batch(input, &progress, &NoopContext)
813            .unwrap();
814        let text = output_csv_text(&output);
815        let rows = count_data_rows(&text);
816
817        assert_eq!(rows, 2);
818        // With first-file headers (name, age), Bob's city gets dropped
819        let header_line = text.lines().next().unwrap();
820        assert_eq!(header_line, "name,age");
821    }
822
823    #[test]
824    fn test_merge_preserves_row_order() {
825        let processor = MergeSpreadsheets::new();
826        let progress = ProgressReporter::new_noop();
827        let input = make_batch_input(vec![("a.csv", "id\n1\n2\n"), ("b.csv", "id\n3\n4\n")]);
828
829        let output = processor
830            .process_batch(input, &progress, &NoopContext)
831            .unwrap();
832        let text = output_csv_text(&output);
833
834        let mut reader = csv::ReaderBuilder::new()
835            .has_headers(true)
836            .from_reader(text.as_bytes());
837        let ids: Vec<String> = reader
838            .records()
839            .filter_map(|r| r.ok())
840            .map(|r| r.get(0).unwrap_or("").to_string())
841            .collect();
842
843        assert_eq!(ids, vec!["1", "2", "3", "4"]);
844    }
845
846    #[test]
847    fn test_merge_with_test_fixtures() {
848        let processor = MergeSpreadsheets::new();
849        let progress = ProgressReporter::new_noop();
850        let csv1 = include_bytes!("../../../../test-fixtures/csv/simple.csv");
851        let csv2 = include_bytes!("../../../../test-fixtures/csv/simple.csv");
852
853        let input = BatchInput {
854            files: vec![
855                BatchFile {
856                    data: csv1.to_vec(),
857                    filename: "simple1.csv".to_string(),
858                    mime_type: Some("text/csv".to_string()),
859                },
860                BatchFile {
861                    data: csv2.to_vec(),
862                    filename: "simple2.csv".to_string(),
863                    mime_type: Some("text/csv".to_string()),
864                },
865            ],
866            params: serde_json::Map::new(),
867        };
868
869        let output = processor
870            .process_batch(input, &progress, &NoopContext)
871            .unwrap();
872        let text = output_csv_text(&output);
873        let rows = count_data_rows(&text);
874
875        // simple.csv has 5 data rows, merging two copies = 10 rows
876        assert_eq!(
877            rows, 10,
878            "Should have 10 rows from two copies of simple.csv"
879        );
880    }
881
882    #[test]
883    fn test_merge_with_dedup_and_fixtures() {
884        let processor = MergeSpreadsheets::new();
885        let progress = ProgressReporter::new_noop();
886        let csv_data = include_bytes!("../../../../test-fixtures/csv/simple.csv");
887
888        let mut params = serde_json::Map::new();
889        params.insert("deduplicate".to_string(), serde_json::json!(true));
890
891        let input = BatchInput {
892            files: vec![
893                BatchFile {
894                    data: csv_data.to_vec(),
895                    filename: "simple1.csv".to_string(),
896                    mime_type: Some("text/csv".to_string()),
897                },
898                BatchFile {
899                    data: csv_data.to_vec(),
900                    filename: "simple2.csv".to_string(),
901                    mime_type: Some("text/csv".to_string()),
902                },
903            ],
904            params,
905        };
906
907        let output = processor
908            .process_batch(input, &progress, &NoopContext)
909            .unwrap();
910        let text = output_csv_text(&output);
911        let rows = count_data_rows(&text);
912
913        // Deduplicated: should have 5 unique rows (all duplicates removed)
914        assert_eq!(rows, 5, "Should have 5 unique rows after dedup");
915        let dupes = output
916            .metadata
917            .get("duplicatesRemoved")
918            .unwrap()
919            .as_u64()
920            .unwrap();
921        assert_eq!(dupes, 5, "Should remove 5 duplicates");
922    }
923}