lit/commands/
batch.rs

1use crate::core::find_repo_root;
2use crate::response::{BatchOperationResult, BatchResponse};
3use serde::Deserialize;
4use std::io::{self, BufRead};
5
6/// Maximum size of a single JSONL request line (1 MiB). A larger line is
7/// rejected without being fully buffered, so a malformed or hostile stream
8/// (e.g. a very long line with no newline) cannot exhaust memory.
9const MAX_LINE_BYTES: usize = 1_048_576;
10
11/// Read a single newline-terminated record from `reader`, capping buffering at
12/// `max_bytes`. Returns `Ok(None)` at end of input, otherwise the line bytes
13/// (without the trailing newline) and a flag indicating the line exceeded
14/// `max_bytes` and was truncated (the remainder is drained from the stream).
15fn read_capped_line<R: BufRead>(
16    reader: &mut R,
17    max_bytes: usize,
18) -> io::Result<Option<(Vec<u8>, bool)>> {
19    let mut buf = Vec::new();
20    let mut oversized = false;
21    let mut saw_any = false;
22    loop {
23        let available = match reader.fill_buf() {
24            Ok(b) => b,
25            Err(ref e) if e.kind() == io::ErrorKind::Interrupted => continue,
26            Err(e) => return Err(e),
27        };
28        if available.is_empty() {
29            if !saw_any {
30                return Ok(None);
31            }
32            return Ok(Some((buf, oversized)));
33        }
34        saw_any = true;
35        let newline = available.iter().position(|&b| b == b'\n');
36        let chunk_len = newline.map_or(available.len(), |idx| idx);
37        let room = max_bytes.saturating_sub(buf.len());
38        let take = room.min(chunk_len);
39        buf.extend_from_slice(&available[..take]);
40        if chunk_len > room {
41            oversized = true;
42        }
43        match newline {
44            Some(idx) => {
45                reader.consume(idx + 1);
46                return Ok(Some((buf, oversized)));
47            }
48            None => {
49                let consumed = available.len();
50                reader.consume(consumed);
51            }
52        }
53    }
54}
55
56/// A single operation in a batch JSONL stream
57#[derive(Debug, Deserialize)]
58struct BatchOperation {
59    command: String,
60    #[serde(default)]
61    args: serde_json::Value,
62}
63
64pub fn execute(atomic: bool, dry_run: bool) -> Result<BatchResponse, crate::errors::LitError> {
65    let _repo_root = find_repo_root()?;
66
67    let stdin = io::stdin();
68    let mut reader = stdin.lock();
69    let mut operations: Vec<BatchOperation> = Vec::new();
70    while let Some((raw, oversized)) = read_capped_line(&mut reader, MAX_LINE_BYTES)
71        .map_err(|e| crate::errors::LitError::io(e.to_string()))?
72    {
73        // Skip oversized lines: the reader never buffered more than
74        // MAX_LINE_BYTES, so an unbounded line cannot exhaust memory.
75        if oversized {
76            continue;
77        }
78        let line = String::from_utf8_lossy(&raw);
79        let trimmed = line.trim();
80        if trimmed.is_empty() {
81            continue;
82        }
83        if let Ok(op) = serde_json::from_str(trimmed) {
84            operations.push(op);
85        }
86    }
87
88    if operations.is_empty() {
89        return Err("No operations provided on stdin (expected JSONL)".into());
90    }
91
92    let total = operations.len();
93    let mut results = Vec::with_capacity(total);
94    let mut succeeded = 0usize;
95    let mut failed = 0usize;
96
97    for (i, op) in operations.iter().enumerate() {
98        if dry_run {
99            results.push(BatchOperationResult {
100                index: i,
101                command: op.command.clone(),
102                status: "ok".to_string(),
103                result: Some(serde_json::json!({"dry_run": true})),
104                error: None,
105            });
106            succeeded += 1;
107            continue;
108        }
109
110        match execute_single_operation(op) {
111            Ok(value) => {
112                results.push(BatchOperationResult {
113                    index: i,
114                    command: op.command.clone(),
115                    status: "ok".to_string(),
116                    result: Some(value),
117                    error: None,
118                });
119                succeeded += 1;
120            }
121            Err(e) => {
122                results.push(BatchOperationResult {
123                    index: i,
124                    command: op.command.clone(),
125                    status: "error".to_string(),
126                    result: None,
127                    error: Some(e.internal_message().to_string()),
128                });
129                failed += 1;
130
131                if atomic {
132                    // In atomic mode, stop on first failure
133                    // Mark remaining as skipped
134                    for (j, op) in operations.iter().enumerate().skip(i + 1) {
135                        results.push(BatchOperationResult {
136                            index: j,
137                            command: op.command.clone(),
138                            status: "skipped".to_string(),
139                            result: None,
140                            error: Some("Skipped due to atomic rollback".to_string()),
141                        });
142                    }
143                    break;
144                }
145            }
146        }
147    }
148
149    Ok(BatchResponse {
150        total,
151        succeeded,
152        failed,
153        atomic,
154        dry_run,
155        results,
156    })
157}
158
159fn execute_single_operation(
160    op: &BatchOperation,
161) -> Result<serde_json::Value, crate::errors::LitError> {
162    match op.command.as_str() {
163        "add" => {
164            let files: Vec<String> = op
165                .args
166                .get("files")
167                .and_then(|v| serde_json::from_value(v.clone()).ok())
168                .unwrap_or_default();
169            if files.is_empty() {
170                return Err("'files' argument required for add".into());
171            }
172            let resp = crate::commands::add::execute(files)?;
173            serde_json::to_value(&resp).map_err(|e| e.to_string().into())
174        }
175        "commit" => {
176            let message = op
177                .args
178                .get("message")
179                .and_then(|v| v.as_str())
180                .ok_or("'message' argument required for commit")?
181                .to_string();
182            let author = op
183                .args
184                .get("author")
185                .and_then(|v| v.as_str())
186                .map(|s| s.to_string());
187            let resp = crate::commands::commit::execute(message, author)?;
188            serde_json::to_value(&resp).map_err(|e| e.to_string().into())
189        }
190        "status" => {
191            let resp = crate::commands::status::execute()?;
192            serde_json::to_value(&resp).map_err(|e| e.to_string().into())
193        }
194        "branch" => {
195            let name = op
196                .args
197                .get("name")
198                .and_then(|v| v.as_str())
199                .map(|s| s.to_string());
200            let delete = op
201                .args
202                .get("delete")
203                .and_then(|v| v.as_bool())
204                .unwrap_or(false);
205            let all = op
206                .args
207                .get("all")
208                .and_then(|v| v.as_bool())
209                .unwrap_or(false);
210            let resp = crate::commands::branch::execute(name, delete, all)?;
211            serde_json::to_value(&resp).map_err(|e| e.to_string().into())
212        }
213        "checkout" => {
214            let target = op
215                .args
216                .get("target")
217                .and_then(|v| v.as_str())
218                .ok_or("'target' argument required for checkout")?
219                .to_string();
220            let b = op
221                .args
222                .get("create")
223                .and_then(|v| v.as_bool())
224                .unwrap_or(false);
225            let resp = crate::commands::checkout::execute(target, b)?;
226            serde_json::to_value(&resp).map_err(|e| e.to_string().into())
227        }
228        "log" => {
229            let count = op.args.get("count").and_then(|v| v.as_u64()).unwrap_or(10) as usize;
230            let oneline = op
231                .args
232                .get("oneline")
233                .and_then(|v| v.as_bool())
234                .unwrap_or(false);
235            let resp = crate::commands::log::execute(count, oneline)?;
236            serde_json::to_value(&resp).map_err(|e| e.to_string().into())
237        }
238        "snapshot" => {
239            let message = op
240                .args
241                .get("message")
242                .and_then(|v| v.as_str())
243                .ok_or("'message' argument required for snapshot")?
244                .to_string();
245            let author = op
246                .args
247                .get("author")
248                .and_then(|v| v.as_str())
249                .map(|s| s.to_string());
250            let metadata = op.args.get("metadata").cloned();
251            let resp = crate::commands::snapshot::execute(message, author, metadata)?;
252            serde_json::to_value(&resp).map_err(|e| e.to_string().into())
253        }
254        _ => Err(format!("Unknown batch command: '{}'", op.command).into()),
255    }
256}
lit/commands/batch.rs

lit/commands/
batch.rs
