diff --git a/Cargo.lock b/Cargo.lock index 9373125..5c0ef70 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2724,6 +2724,7 @@ dependencies = [ "nigig-uikit", "proptest", "robius-contacts", + "robius-file-picker", "robius-sms", ] diff --git a/crates/apps/nigig-sms/Cargo.toml b/crates/apps/nigig-sms/Cargo.toml index 0a1e945..a472c7a 100644 --- a/crates/apps/nigig-sms/Cargo.toml +++ b/crates/apps/nigig-sms/Cargo.toml @@ -10,6 +10,9 @@ nigig-uikit = { path = "../../nigig-uikit" } chrono = { version = "0.4", features = ["serde"] } robius-sms = { version = "0.1.0", path = "../../robius-sms" } robius-contacts = { version = "0.1.0", path = "../../robius-contacts" } +# Same pin as nigig-build and nigig-pay-ui, which already use it to +# pick a CSV. Bulk SMS needs the identical capability. +robius-file-picker = { git = "https://github.com/project-robius/robius", rev = "b766e62b0600f5d2ee21cc6995648346fc277bd8" } [dev-dependencies] proptest = "1" diff --git a/crates/apps/nigig-sms/src/sms_frame/bulk/mod.rs b/crates/apps/nigig-sms/src/sms_frame/bulk/mod.rs index 5f0021c..d148e89 100644 --- a/crates/apps/nigig-sms/src/sms_frame/bulk/mod.rs +++ b/crates/apps/nigig-sms/src/sms_frame/bulk/mod.rs @@ -1,6 +1,7 @@ pub mod sms_bulk_page; pub mod directory_scraper; pub mod companies_list; +pub mod recipient_csv; use makepad_widgets::ScriptVm; diff --git a/crates/apps/nigig-sms/src/sms_frame/bulk/recipient_csv.rs b/crates/apps/nigig-sms/src/sms_frame/bulk/recipient_csv.rs new file mode 100644 index 0000000..148b318 --- /dev/null +++ b/crates/apps/nigig-sms/src/sms_frame/bulk/recipient_csv.rs @@ -0,0 +1,460 @@ +// Parse an arbitrary user-supplied CSV into a recipient list. +// +// This is deliberately NOT `import_business_listings_from_csv_path`. +// That importer exists for one specific artefact: an 11-column export +// (`category,company_name,address,phones,emails,industry,source_url, +// page_number,website,is_favorite,notes`) with a hardcoded filename, +// discovered by probing /sdcard/Download and three dev paths. It fails +// the whole file on the first malformed row, and its own error message +// tells the user to run `adb push`. None of that is reachable for +// someone who just wants to text a list of numbers they exported from +// a spreadsheet. +// +// What people actually have is a phone column, maybe a name column, +// maybe a header, and inevitably some junk rows. So: +// +// * find the phone column by NAME if there is a header, otherwise by +// content -- the first column where most values look like numbers; +// * skip bad rows and report how many, rather than rejecting the file; +// * de-duplicate, because sending the same person four copies of a +// marketing message is worse than not sending at all; +// * keep every row's original line number so the UI can point at what +// it skipped. +// +// Kept free of Makepad and of any file I/O so it is testable on the +// host: the SMS crates' CI runs on Linux, where the whole Android +// backend is a stub, and an untested parser is exactly how bug A3 +// (byte-offset slicing) shipped. + +/// One accepted row. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CsvRecipient { + /// Normalised E.164-ish number, ready to hand to the SMS provider. + pub phone: String, + /// Display name if the file had one, else empty. + pub name: String, +} + +/// Outcome of parsing a file: what we can send to, and what we ignored. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct CsvImport { + pub recipients: Vec, + /// `(line_number, reason)` for each row we could not use. + pub skipped: Vec<(usize, String)>, + /// Rows dropped because the number already appeared earlier. + pub duplicates: usize, +} + +impl CsvImport { + pub fn is_empty(&self) -> bool { + self.recipients.is_empty() + } + + /// One-line summary for the status label. + pub fn summary(&self) -> String { + let mut s = format!( + "{} recipient{}", + self.recipients.len(), + if self.recipients.len() == 1 { "" } else { "s" } + ); + if self.duplicates > 0 { + s.push_str(&format!(", {} duplicate(s) removed", self.duplicates)); + } + if !self.skipped.is_empty() { + s.push_str(&format!(", {} row(s) skipped", self.skipped.len())); + } + s + } +} + +/// Header names we accept for the phone column, lowercased. +const PHONE_HEADERS: &[&str] = &[ + "phone", "phones", "number", "msisdn", "mobile", "cell", "telephone", + "tel", "contact", "phone_number", "phonenumber", "mobile_number", +]; + +/// Header names we accept for the display-name column, lowercased. +const NAME_HEADERS: &[&str] = &[ + "name", "company_name", "company", "customer", "contact_name", + "full_name", "fullname", "business", "client", +]; + +/// Split one CSV line, honouring double-quoted fields. +/// +/// Not a general CSV implementation -- no embedded newlines -- but it +/// does handle `"Acme, Inc.",+254...`, which a naive `split(',')` gets +/// wrong and which is extremely common in spreadsheet exports. +fn split_row(line: &str) -> Vec { + let mut out = Vec::new(); + let mut cur = String::new(); + let mut in_quotes = false; + let mut chars = line.chars().peekable(); + + while let Some(c) = chars.next() { + match c { + '"' if in_quotes && chars.peek() == Some(&'"') => { + // Escaped quote inside a quoted field. + cur.push('"'); + chars.next(); + } + '"' => in_quotes = !in_quotes, + ',' | ';' | '\t' if !in_quotes => { + out.push(cur.trim().to_string()); + cur = String::new(); + } + _ => cur.push(c), + } + } + out.push(cur.trim().to_string()); + out +} + +/// Normalise a phone number, or return None if it cannot be one. +/// +/// Kenyan-friendly (this app's directory is Nairobi businesses) but not +/// Kenya-only: anything already in `+` form is kept. +/// +/// 0712345678 -> +254712345678 +/// 254712345678 -> +254712345678 +/// +254 712 345678 -> +254712345678 +/// 712345678 -> +254712345678 +/// +/// Returns None for anything too short, too long, or non-numeric, so a +/// stray "N/A" or an email address in the phone column is skipped +/// rather than dispatched to the radio. +pub fn normalise_phone(raw: &str) -> Option { + let trimmed = raw.trim(); + if trimmed.is_empty() { + return None; + } + + let had_plus = trimmed.starts_with('+'); + // Strip formatting: spaces, dashes, parens, dots. + let digits: String = trimmed.chars().filter(|c| c.is_ascii_digit()).collect(); + if digits.is_empty() { + return None; + } + + // A number with letters mixed in is not a number -- reject rather + // than silently salvaging the digits out of "call 0712345678 ext 4". + let stripped: String = trimmed + .chars() + .filter(|c| !c.is_whitespace() && !"+-().".contains(*c)) + .collect(); + if !stripped.chars().all(|c| c.is_ascii_digit()) { + return None; + } + + let normalised = if had_plus { + // Already international; trust it. + format!("+{digits}") + } else if let Some(rest) = digits.strip_prefix("254") { + format!("+254{rest}") + } else if let Some(rest) = digits.strip_prefix('0') { + format!("+254{rest}") + } else if digits.len() == 9 { + // Bare subscriber number, e.g. 712345678. + format!("+254{digits}") + } else { + // Unknown shape with no country code. Keep the digits but do + // not invent a country -- an 11-digit US number must not become + // a Kenyan one. + format!("+{digits}") + }; + + // Sanity bounds. E.164 allows at most 15 digits; anything under 7 + // is not dialable. + let n = normalised.chars().filter(|c| c.is_ascii_digit()).count(); + if !(7..=15).contains(&n) { + return None; + } + Some(normalised) +} + +/// Does this cell look like a phone number? +fn looks_like_phone(cell: &str) -> bool { + normalise_phone(cell).is_some() +} + +/// Decide which column holds phones and which holds names. +/// +/// Returns `(phone_idx, name_idx, header_consumed)`. +fn detect_columns(rows: &[Vec]) -> (usize, Option, bool) { + let Some(first) = rows.first() else { + return (0, None, false); + }; + + // A header row is one where the phone column is named and the cell + // is NOT itself a phone number. "phone" is a header; "+254..." is + // data even if a column is called that. + let lowered: Vec = first.iter().map(|c| c.to_lowercase()).collect(); + let header_phone = lowered + .iter() + .position(|c| PHONE_HEADERS.contains(&c.as_str())); + + if let Some(pi) = header_phone { + if !looks_like_phone(&first[pi]) { + let ni = lowered + .iter() + .position(|c| NAME_HEADERS.contains(&c.as_str())); + return (pi, ni, true); + } + } + + // No usable header. Pick the column with the most phone-looking + // values across the sample. + let width = rows.iter().map(|r| r.len()).max().unwrap_or(1); + // `take(50)` rather than `rows[..sample]`: the SMS crates gate + // against byte-offset indexing because `s[..n]` on a &str panics on + // a non-boundary (bug A3). These are Vec slices and would be safe, + // but the gate matches on shape, and an iterator is no worse here. + const SAMPLE: usize = 50; + let mut best = (0usize, 0usize); // (column, hits) + for col in 0..width { + let hits = rows + .iter() + .take(SAMPLE) + .filter(|r| r.get(col).is_some_and(|c| looks_like_phone(c))) + .count(); + if hits > best.1 { + best = (col, hits); + } + } + let phone_idx = best.0; + + // Name column: the first column that is not the phone column and + // holds mostly non-numeric text. + let name_idx = (0..width).find(|&col| { + col != phone_idx + && rows.iter().take(SAMPLE).any(|r| { + r.get(col) + .is_some_and(|c| !c.is_empty() && !looks_like_phone(c)) + }) + }); + + (phone_idx, name_idx, false) +} + +/// Parse CSV text into a recipient list. +/// +/// Never returns Err for a merely messy file -- bad rows land in +/// `skipped`. Err is reserved for "there is nothing usable here at +/// all", which is the only case where the user must pick a different +/// file. +pub fn parse_recipients(input: &str) -> Result { + // Strip a UTF-8 BOM; Excel writes one and it corrupts the first + // header cell, which would break column detection. + let input = input.strip_prefix('\u{feff}').unwrap_or(input); + + let raw_lines: Vec<(usize, &str)> = input + .lines() + .enumerate() + .map(|(i, l)| (i + 1, l.trim_end_matches('\r'))) + .filter(|(_, l)| !l.trim().is_empty()) + .collect(); + + if raw_lines.is_empty() { + return Err("The file is empty.".into()); + } + + let rows: Vec> = raw_lines.iter().map(|(_, l)| split_row(l)).collect(); + let (phone_idx, name_idx, header) = detect_columns(&rows); + + let mut out = CsvImport::default(); + let mut seen: std::collections::HashSet = std::collections::HashSet::new(); + + for (idx, (lineno, _)) in raw_lines.iter().enumerate() { + if header && idx == 0 { + continue; + } + let row = &rows[idx]; + let Some(cell) = row.get(phone_idx) else { + out.skipped.push((*lineno, "no phone column".into())); + continue; + }; + let Some(phone) = normalise_phone(cell) else { + let shown = if cell.is_empty() { "(blank)" } else { cell.as_str() }; + out.skipped + .push((*lineno, format!("not a phone number: {shown}"))); + continue; + }; + if !seen.insert(phone.clone()) { + out.duplicates += 1; + continue; + } + let name = name_idx + .and_then(|i| row.get(i)) + .map(|s| s.to_string()) + .unwrap_or_default(); + out.recipients.push(CsvRecipient { phone, name }); + } + + if out.recipients.is_empty() { + return Err(format!( + "No phone numbers found. Checked {} row(s); the file needs a \ + column of numbers such as 0712345678 or +254712345678.", + raw_lines.len() + )); + } + Ok(out) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_a_plain_two_column_file_with_header() { + let csv = "name,phone\nAlice,0712345678\nBob,+254798765432\n"; + let out = parse_recipients(csv).unwrap(); + assert_eq!(out.recipients.len(), 2); + assert_eq!(out.recipients[0].name, "Alice"); + assert_eq!(out.recipients[0].phone, "+254712345678"); + assert_eq!(out.recipients[1].phone, "+254798765432"); + assert!(out.skipped.is_empty()); + } + + #[test] + fn parses_a_bare_list_with_no_header_at_all() { + let csv = "0712345678\n0798765432\n"; + let out = parse_recipients(csv).unwrap(); + assert_eq!(out.recipients.len(), 2); + assert_eq!(out.recipients[0].phone, "+254712345678"); + } + + /// A header cell named "phone" must not be sent a message. + #[test] + fn the_header_row_is_never_a_recipient() { + let csv = "name,phone\nAlice,0712345678\n"; + let out = parse_recipients(csv).unwrap(); + assert_eq!(out.recipients.len(), 1); + assert!(!out.recipients.iter().any(|r| r.name == "name")); + } + + /// The phone column is not always second, and not always named. + #[test] + fn finds_the_phone_column_by_content_when_unnamed() { + let csv = "Acme Ltd,Nairobi,0712345678,acme@example.com\n\ + Beta Co,Mombasa,0798765432,beta@example.com\n"; + let out = parse_recipients(csv).unwrap(); + assert_eq!(out.recipients.len(), 2); + assert_eq!(out.recipients[0].phone, "+254712345678"); + } + + /// An email in the row must not be mistaken for a number. + #[test] + fn an_email_column_is_not_treated_as_phones() { + let csv = "phone,email\n0712345678,a@b.com\n"; + let out = parse_recipients(csv).unwrap(); + assert_eq!(out.recipients.len(), 1); + assert_eq!(out.recipients[0].phone, "+254712345678"); + } + + #[test] + fn quoted_fields_containing_commas_do_not_shift_columns() { + let csv = "name,phone\n\"Acme, Inc.\",0712345678\n"; + let out = parse_recipients(csv).unwrap(); + assert_eq!(out.recipients.len(), 1); + assert_eq!(out.recipients[0].name, "Acme, Inc."); + assert_eq!(out.recipients[0].phone, "+254712345678"); + } + + #[test] + fn semicolon_and_tab_separated_files_also_work() { + let out = parse_recipients("name;phone\nAlice;0712345678\n").unwrap(); + assert_eq!(out.recipients[0].phone, "+254712345678"); + let out = parse_recipients("name\tphone\nBob\t0798765432\n").unwrap(); + assert_eq!(out.recipients[0].phone, "+254798765432"); + } + + /// Excel writes a BOM. It must not corrupt header detection. + #[test] + fn a_utf8_bom_does_not_break_the_header() { + let csv = "\u{feff}name,phone\nAlice,0712345678\n"; + let out = parse_recipients(csv).unwrap(); + assert_eq!(out.recipients.len(), 1); + assert_eq!(out.recipients[0].name, "Alice"); + } + + /// One bad row must not reject the whole file -- the directory + /// importer's behaviour, and the reason it is unusable here. + #[test] + fn bad_rows_are_skipped_and_reported_not_fatal() { + let csv = "name,phone\nAlice,0712345678\nBroken,N/A\nBob,0798765432\n"; + let out = parse_recipients(csv).unwrap(); + assert_eq!(out.recipients.len(), 2); + assert_eq!(out.skipped.len(), 1); + assert_eq!(out.skipped[0].0, 3, "should name the offending line"); + } + + /// Sending the same person the same marketing text twice is a bug + /// that costs money, so dedupe is not cosmetic. + #[test] + fn duplicate_numbers_are_collapsed_even_when_written_differently() { + let csv = "phone\n0712345678\n+254712345678\n254712345678\n"; + let out = parse_recipients(csv).unwrap(); + assert_eq!(out.recipients.len(), 1); + assert_eq!(out.duplicates, 2); + } + + #[test] + fn an_empty_file_is_an_error_not_an_empty_batch() { + assert!(parse_recipients("").is_err()); + assert!(parse_recipients("\n\n \n").is_err()); + } + + #[test] + fn a_file_with_no_numbers_at_all_is_an_error() { + let err = parse_recipients("name,email\nAlice,a@b.com\n").unwrap_err(); + assert!(err.contains("No phone numbers"), "got: {err}"); + } + + #[test] + fn normalises_the_common_kenyan_forms_identically() { + for raw in ["0712345678", "254712345678", "+254712345678", "712345678"] { + assert_eq!( + normalise_phone(raw).as_deref(), + Some("+254712345678"), + "failed for {raw}" + ); + } + } + + #[test] + fn strips_spaces_dashes_and_parens() { + assert_eq!( + normalise_phone("+254 (712) 345-678").as_deref(), + Some("+254712345678") + ); + } + + /// An international number must keep its own country code. + #[test] + fn a_non_kenyan_international_number_is_left_alone() { + assert_eq!(normalise_phone("+14155552671").as_deref(), Some("+14155552671")); + assert_eq!(normalise_phone("+442071838750").as_deref(), Some("+442071838750")); + } + + #[test] + fn rejects_things_that_are_not_numbers() { + for raw in ["", " ", "N/A", "none", "a@b.com", "call me", "12", "abc123"] { + assert!(normalise_phone(raw).is_none(), "should reject {raw:?}"); + } + } + + /// 16+ digits is not dialable; do not hand it to the radio. + #[test] + fn rejects_absurdly_long_numbers() { + assert!(normalise_phone("+1234567890123456789").is_none()); + } + + #[test] + fn summary_mentions_duplicates_and_skips() { + let csv = "phone\n0712345678\n0712345678\nbroken\n"; + let out = parse_recipients(csv).unwrap(); + let s = out.summary(); + assert!(s.contains("1 recipient"), "got: {s}"); + assert!(s.contains("duplicate"), "got: {s}"); + assert!(s.contains("skipped"), "got: {s}"); + } +} diff --git a/crates/apps/nigig-sms/src/sms_frame/bulk/sms_bulk_page.rs b/crates/apps/nigig-sms/src/sms_frame/bulk/sms_bulk_page.rs index ac11860..582b197 100644 --- a/crates/apps/nigig-sms/src/sms_frame/bulk/sms_bulk_page.rs +++ b/crates/apps/nigig-sms/src/sms_frame/bulk/sms_bulk_page.rs @@ -15,6 +15,7 @@ use super::companies_list::{ use super::directory_scraper::{ get_extraction_state, run_extraction, DirectoryScraperAction, ExtractionState, }; +use super::recipient_csv::{parse_recipients, CsvImport}; use crate::chats::sms_frame::lock_ext::LockRecover; use crate::chats::sms_frame::sms_utils::now_ms; @@ -31,6 +32,20 @@ struct BulkSendProgress { /// D5: guards against two overlapping bulk sends. static BULK_SEND_IN_FLIGHT: AtomicBool = AtomicBool::new(false); static BULK_SEND_PROGRESS: Mutex> = Mutex::new(None); +/// Set by the UI to stop a paced batch early. +/// +/// Pacing turns a bulk send from a few seconds into potentially hours, +/// so "wait for it to finish" stops being an acceptable answer: the user +/// needs a way out that does not involve killing the app. The worker +/// checks this while sleeping, not just between sends. +static BULK_SEND_CANCEL: AtomicBool = AtomicBool::new(false); + +/// Recipients parsed from a picked CSV, handed back to the UI thread. +/// +/// The file-picker callback runs off the UI thread, so it cannot touch +/// `Cx`. Same shape as the D1/D5 worker handoff: park the result, raise +/// the signal, drain it in handle_event. +static PENDING_CSV_IMPORT: Mutex>> = Mutex::new(None); static FILTER_TEXT: Mutex = Mutex::new(String::new()); static SHOW_FAVORITES: Mutex = Mutex::new(false); @@ -228,8 +243,36 @@ company_card := mod.widgets.SmsBulkCompanyCard {} Label { text: "Recipients" draw_text +: { color: (LABEL), text_style.font_size: 11.0 } } bulk_recipients := TextInput { width: Fill, height: 100, empty_text: "+254712345678\n+254798765432", draw_bg.color: (INPUT_BG), draw_text +: { color: (TEXT), text_style.font_size: 13.0 } } bulk_recipient_count := Label { width: Fill, text: "0 recipients" draw_text +: { color: (LABEL), text_style.font_size: 10.0 } } + + // Import recipients from any CSV on the device. + csv_import_row := View { + width: Fill, height: Fit + flow: Right + spacing: 8 + csv_import_btn := Button { + width: Fill, height: 34 + text: "📂 Import recipients from CSV" + draw_bg +: { color: (ACCENT), border_radius: 8.0 } + draw_text +: { color: #x000000, text_style.font_size: 12.0 } + } + } + csv_import_status := Label { width: Fill, text: "" draw_text +: { color: (LABEL), text_style.font_size: 10.0 } } + Label { text: "Message" draw_text +: { color: (LABEL), text_style.font_size: 11.0 } } bulk_body := TextInput { width: Fill, height: 100, empty_text: "Type your message here...", draw_bg.color: (INPUT_BG), draw_text +: { color: (TEXT), text_style.font_size: 14.0 } } + + // Gap between messages. Defaults to the carrier-safe 60s + // rather than 0, so a long list completes instead of + // stopping at the throttle. + Label { text: "Seconds between messages" draw_text +: { color: (LABEL), text_style.font_size: 11.0 } } + bulk_delay_row := View { + width: Fill, height: Fit + flow: Right + spacing: 8 + bulk_delay := TextInput { width: 90, height: 34, text: "60", empty_text: "60", draw_bg.color: (INPUT_BG), draw_text +: { color: (TEXT), text_style.font_size: 13.0 } } + bulk_delay_hint := Label { width: Fill, text: "" draw_text +: { color: (LABEL), text_style.font_size: 10.0 } } + } + bulk_send_btn := Button { width: Fill, height: 48, text: "Send Bulk SMS", draw_bg +: { color: (SEND_BG), border_radius: 14.0 }, draw_text +: { color: #xFFFFFF, text_style.font_size: 15.0 } } bulk_body_cost := Label { width: Fill, text: "" draw_text +: { color: (LABEL), text_style.font_size: 10.0 } } bulk_status := Label { width: Fill, text: "" draw_text +: { color: (LABEL), text_style.font_size: 11.0 } } @@ -312,6 +355,7 @@ impl Widget for SmsBulkPage { // D5: the bulk-send worker publishes progress and signals the UI. if matches!(event, Event::Signal) { self.drain_bulk_send_progress(cx); + self.drain_csv_import(cx); } let actions = cx.capture_actions(|cx| self.view.handle_event(cx, event, scope)); @@ -355,7 +399,27 @@ impl Widget for SmsBulkPage { } } if self.view.button(cx, ids!(bulk_send_btn)).clicked(&actions) { - self.handle_send_bulk(cx); + // While a paced batch is running the same button cancels it: + // a 200-message run at 60s apart is over three hours, and + // force-quitting the app is not an acceptable stop button. + if BULK_SEND_IN_FLIGHT.load(Ordering::Relaxed) { + BULK_SEND_CANCEL.store(true, Ordering::Relaxed); + self.view + .label(cx, ids!(bulk_status)) + .set_text(cx, "Stopping after the current message…"); + self.view.redraw(cx); + } else { + self.handle_send_bulk(cx); + } + } + + if self.view.button(cx, ids!(csv_import_btn)).clicked(&actions) { + self.handle_pick_csv(cx); + } + + // Re-estimate the batch duration as the delay is edited. + if self.view.text_input(cx, ids!(bulk_delay)).changed(&actions).is_some() { + self.refresh_delay_hint(cx); } // F10 / A5: show what this body will actually COST, live. // @@ -670,22 +734,38 @@ impl SmsBulkPage { .as_ref() .is_some_and(|(b, n)| b == trimmed_body && *n == total); + let pacing = self.current_pacing(cx); + if !confirmation_matches { self.pending_bulk_send = Some((trimmed_body.to_string(), total)); let encoding = match seg.encoding { robius_sms::SmsEncoding::Gsm7 => "GSM-7", robius_sms::SmsEncoding::Ucs2 => "Unicode", }; + // Quote the wall-clock duration too. At the default 60s gap + // a 200-recipient batch runs for over three hours, and the + // user must know that before the first tap, not discover it + // afterwards. + let timing = if pacing.is_immediate() { + String::new() + } else { + format!( + " Sending {}s apart, so this takes {}.", + pacing.delay_ms() / 1000, + pacing.describe_duration(total) + ) + }; self.view.label(cx, ids!(bulk_status)).set_text( cx, &format!( - "Send {} SMS ({} {} segment{} each, {} total)? \ + "Send {} SMS ({} {} segment{} each, {} total)?{} \ This cannot be undone. Tap Send again to confirm.", total, seg.segments, encoding, if seg.segments == 1 { "" } else { "s" }, total_segments, + timing, ), ); self.view.redraw(cx); @@ -733,6 +813,7 @@ impl SmsBulkPage { } let body_owned = trimmed_body.to_string(); + BULK_SEND_CANCEL.store(false, Ordering::Relaxed); self.view .label(cx, ids!(bulk_status)) .set_text(cx, &format!("Sending 0/{total}…")); @@ -747,11 +828,51 @@ impl SmsBulkPage { let mut failed = 0usize; let mut throttled = 0usize; - for recipient in &recipients { - if limiter.allow_at(now_ms()).is_err() { + for (i, recipient) in recipients.iter().enumerate() { + // Wait BEFORE every send except the first. Doing it here + // rather than at the end of the body means a cancel + // between messages takes effect without burning the gap, + // and the arithmetic matches SendPacing::total_duration_ms + // (n messages, n-1 gaps). + if i > 0 && !pacing.is_immediate() { + let step = std::time::Duration::from_millis(200); + let mut waited = 0i64; + while waited < pacing.delay_ms() { + if BULK_SEND_CANCEL.load(Ordering::Relaxed) { + break; + } + thread::sleep(step); + waited += step.as_millis() as i64; + } + } + + if BULK_SEND_CANCEL.load(Ordering::Relaxed) { throttled = recipients.len() - sent - failed; break; } + + // The limiter is still the backstop. With a sane gap it + // never fires; if the user forced delay=0 on a list + // longer than the cap, wait it out rather than + // abandoning the batch the way this loop used to. + loop { + match limiter.allow_at(now_ms()) { + Ok(()) => break, + Err(wait_ms) => { + if BULK_SEND_CANCEL.load(Ordering::Relaxed) { + break; + } + // Cap each nap so cancel stays responsive. + let nap = wait_ms.clamp(0, 1000) as u64; + thread::sleep(std::time::Duration::from_millis(nap.max(50))); + } + } + } + if BULK_SEND_CANCEL.load(Ordering::Relaxed) { + throttled = recipients.len() - sent - failed; + break; + } + match send_single(recipient, &body_owned) { Ok(()) => sent += 1, Err(_) => failed += 1, @@ -775,10 +896,146 @@ impl SmsBulkPage { done: true, }); BULK_SEND_IN_FLIGHT.store(false, Ordering::Relaxed); + BULK_SEND_CANCEL.store(false, Ordering::Relaxed); SignalToUI::set_ui_signal(); }); } + /// Current pacing from the delay field. + /// + /// Empty or unparseable falls back to the carrier-safe default + /// rather than to zero: a typo must not turn a paced batch into a + /// burst that trips the throttle at message 30. + fn current_pacing(&mut self, cx: &mut Cx) -> robius_sms::SendPacing { + let raw = self.view.text_input(cx, ids!(bulk_delay)).text(); + let t = raw.trim(); + if t.is_empty() { + return robius_sms::SendPacing::default(); + } + match t.parse::() { + Ok(secs) => robius_sms::SendPacing::from_seconds(secs), + Err(_) => robius_sms::SendPacing::default(), + } + } + + /// Show what the chosen delay means for this recipient count. + fn refresh_delay_hint(&mut self, cx: &mut Cx) { + let n = self + .view + .text_input(cx, ids!(bulk_recipients)) + .text() + .lines() + .filter(|l| !l.trim().is_empty()) + .count(); + let pacing = self.current_pacing(cx); + let hint = if n == 0 { + String::new() + } else if pacing.is_immediate() { + format!( + "No delay — sends as fast as allowed, and stops at the \ + carrier limit of {} per {} min.", + robius_sms::SendRateLimiter::DEFAULT_CAPACITY, + robius_sms::SendRateLimiter::DEFAULT_WINDOW_MS / 60_000, + ) + } else { + format!( + "{} message(s), {}s apart — {}.", + n, + pacing.delay_ms() / 1000, + pacing.describe_duration(n) + ) + }; + self.view.label(cx, ids!(bulk_delay_hint)).set_text(cx, &hint); + self.view.redraw(cx); + } + + /// Open the system file picker and parse whatever CSV comes back. + /// + /// Deliberately not `directory_scraper::find_csv`, which only looks + /// for one hardcoded filename in /sdcard/Download and tells the user + /// to run `adb push` when it is missing. This accepts any file the + /// user can reach through the platform picker, including Drive and + /// other storage providers, which is what "upload a CSV" means on a + /// phone. + fn handle_pick_csv(&mut self, cx: &mut Cx) { + self.view + .label(cx, ids!(csv_import_status)) + .set_text(cx, "Opening file picker…"); + self.view.redraw(cx); + + use robius_file_picker::FileDialog; + let _ = FileDialog::new() + .add_filter("CSV", &["csv", "txt"]) + .set_title("Pick a CSV of recipients") + .pick_file(|result| { + // Runs off the UI thread: park the outcome and signal. + let parked = match result { + Ok(Some(picked)) => match picked.into_local_file() { + Ok(local) => match std::fs::read_to_string(local.path()) { + Ok(text) => Some(parse_recipients(&text)), + Err(e) => Some(Err(format!("Could not read the file: {e}"))), + }, + Err(e) => Some(Err(format!("Could not open the file: {e}"))), + }, + // User cancelled: leave the previous list alone. + Ok(None) => None, + Err(e) => Some(Err(format!("File picker failed: {e}"))), + }; + if let Some(outcome) = parked { + *PENDING_CSV_IMPORT.lock_recover() = Some(outcome); + SignalToUI::set_ui_signal(); + } + }); + } + + /// Apply a CSV import parked by the picker callback, if any. + fn drain_csv_import(&mut self, cx: &mut Cx) { + let Some(outcome) = PENDING_CSV_IMPORT.lock_recover().take() else { + return; + }; + match outcome { + Err(e) => { + self.view.label(cx, ids!(csv_import_status)).set_text(cx, &e); + } + Ok(import) => { + let numbers: Vec = + import.recipients.iter().map(|r| r.phone.clone()).collect(); + self.view + .text_input(cx, ids!(bulk_recipients)) + .set_text(cx, &numbers.join("\n")); + self.view + .label(cx, ids!(bulk_recipient_count)) + .set_text(cx, &format!("{} recipients", numbers.len())); + + // Name the first few skipped lines. "3 rows skipped" with + // no detail is not actionable when the file has 900 rows. + let mut status = format!("Imported {}", import.summary()); + if !import.skipped.is_empty() { + let detail: Vec = import + .skipped + .iter() + .take(3) + .map(|(line, why)| format!("line {line}: {why}")) + .collect(); + status.push_str(&format!(" — {}", detail.join("; "))); + if import.skipped.len() > 3 { + status.push_str(&format!(" (+{} more)", import.skipped.len() - 3)); + } + } + self.view + .label(cx, ids!(csv_import_status)) + .set_text(cx, &status); + + // A freshly imported list invalidates any armed + // confirmation: the count it was armed for is stale. + self.pending_bulk_send = None; + self.set_sub_tab(cx, BulkSubTab::Compose); + self.refresh_delay_hint(cx); + } + } + self.view.redraw(cx); + } + /// Apply bulk-send progress published by the D5 worker, if any. fn drain_bulk_send_progress(&mut self, cx: &mut Cx) { let Some(p) = BULK_SEND_PROGRESS.lock_recover().take() else { diff --git a/crates/robius-sms/src/lib.rs b/crates/robius-sms/src/lib.rs index 47776b0..44cb419 100644 --- a/crates/robius-sms/src/lib.rs +++ b/crates/robius-sms/src/lib.rs @@ -335,6 +335,115 @@ impl Default for SendRateLimiter { } } +/// How long to wait between two sends in a batch, and how long a whole +/// batch will therefore take. +/// +/// `SendRateLimiter` above answers "may I send now?" and, when the +/// answer is no, the bulk worker used to `break` -- abandoning the rest +/// of the list and telling the user to come back later. That is correct +/// for protecting the carrier ceiling but useless as a way to actually +/// deliver 200 messages: the user has to babysit the app and re-run it +/// seven times. +/// +/// Pacing is the other half. Instead of firing as fast as the loop +/// allows and then stopping dead at the cap, put a fixed gap between +/// sends so the batch stays under the ceiling by construction and runs +/// to completion. +/// +/// Pure arithmetic, no sleeping, so the schedule can be asserted in a +/// unit test on any host. +#[derive(Copy, Clone, Debug, Eq, PartialEq)] +pub struct SendPacing { + /// Gap between consecutive sends. Zero means "as fast as possible", + /// which is still bounded by `SendRateLimiter`. + delay_ms: i64, +} + +impl SendPacing { + /// Longest gap we let a user choose: 10 minutes. Past this a batch + /// of any size takes days, and the app is not a scheduler. + pub const MAX_DELAY_MS: i64 = 10 * 60 * 1000; + + /// Gap that keeps a batch just inside the default carrier ceiling + /// (30 messages / 30 minutes), i.e. one per minute. + pub const RECOMMENDED_DELAY_MS: i64 = + SendRateLimiter::DEFAULT_WINDOW_MS / SendRateLimiter::DEFAULT_CAPACITY as i64; + + /// Clamps rather than rejecting: this is driven by a text field, and + /// refusing to send because someone typed 9999 would be worse than + /// quietly using the maximum. + pub fn from_millis(delay_ms: i64) -> Self { + Self { + delay_ms: delay_ms.clamp(0, Self::MAX_DELAY_MS), + } + } + + pub fn from_seconds(delay_s: i64) -> Self { + Self::from_millis(delay_s.saturating_mul(1000)) + } + + /// A gap that spreads `count` messages evenly across the rate + /// limiter's window, so the batch never trips the cap. + /// + /// Returns `RECOMMENDED_DELAY_MS` when the batch already fits. + pub fn to_stay_under(capacity: u32, window_ms: i64, count: usize) -> Self { + if capacity == 0 || count == 0 { + return Self::from_millis(0); + } + if count <= capacity as usize { + return Self::from_millis(0); + } + // Need `count` sends; only `capacity` fit per window. + Self::from_millis(window_ms / capacity as i64) + } + + pub fn delay_ms(self) -> i64 { + self.delay_ms + } + + pub fn is_immediate(self) -> bool { + self.delay_ms == 0 + } + + /// Wall-clock duration of a batch of `count` messages. + /// + /// There are `count - 1` gaps, not `count`: nothing is waited before + /// the first message or after the last. Off-by-one here would show + /// the user a materially wrong estimate on small batches. + pub fn total_duration_ms(self, count: usize) -> i64 { + let gaps = count.saturating_sub(1) as i64; + self.delay_ms.saturating_mul(gaps) + } + + /// Human estimate for the confirmation prompt, e.g. "about 3m 20s". + pub fn describe_duration(self, count: usize) -> String { + let ms = self.total_duration_ms(count); + if ms <= 0 { + return "a few seconds".into(); + } + let total_s = ms / 1000; + let h = total_s / 3600; + let m = (total_s % 3600) / 60; + let s = total_s % 60; + if h > 0 { + format!("about {h}h {m}m") + } else if m > 0 { + format!("about {m}m {s}s") + } else { + format!("about {s}s") + } + } +} + +impl Default for SendPacing { + /// Default to the carrier-safe gap rather than to zero. A user who + /// never touches the field gets a batch that completes instead of + /// one that stops a third of the way through. + fn default() -> Self { + Self::from_millis(Self::RECOMMENDED_DELAY_MS) + } +} + impl BulkSendRequest { /// Validate a bulk request. /// @@ -864,6 +973,109 @@ mod tests { assert_eq!(rl.in_window(31), 30); } + // ---- SendPacing ----------------------------------------------------- + + #[test] + fn pacing_defaults_to_the_carrier_safe_gap_not_to_zero() { + // A user who never touches the delay field must still get a + // batch that completes rather than one that stops at 30. + let p = SendPacing::default(); + assert_eq!(p.delay_ms(), SendPacing::RECOMMENDED_DELAY_MS); + assert!(!p.is_immediate()); + } + + #[test] + fn the_recommended_gap_is_one_per_minute() { + // 30 minutes / 30 messages. + assert_eq!(SendPacing::RECOMMENDED_DELAY_MS, 60_000); + } + + /// The point of pacing: a batch bigger than the cap must still + /// finish. Walk the limiter with the recommended gap and check + /// nothing is refused. + #[test] + fn the_recommended_gap_keeps_a_long_batch_under_the_limiter() { + let p = SendPacing::default(); + let mut rl = SendRateLimiter::default(); + let mut now = 0i64; + for i in 0..200 { + assert!( + rl.allow_at(now).is_ok(), + "send {i} refused at t={now}ms; pacing failed to protect the cap" + ); + now += p.delay_ms(); + } + } + + /// And the negative: with no gap, the same batch dies at 30. This is + /// the behaviour that made the old bulk worker abandon the list. + #[test] + fn without_pacing_the_same_batch_is_refused_at_the_cap() { + let mut rl = SendRateLimiter::default(); + let allowed = (0..200).filter(|i| rl.allow_at(*i).is_ok()).count(); + assert_eq!(allowed, SendRateLimiter::DEFAULT_CAPACITY as usize); + } + + #[test] + fn pacing_clamps_instead_of_rejecting_out_of_range_input() { + assert_eq!(SendPacing::from_millis(-5).delay_ms(), 0); + assert_eq!( + SendPacing::from_millis(i64::MAX).delay_ms(), + SendPacing::MAX_DELAY_MS + ); + assert_eq!( + SendPacing::from_seconds(i64::MAX).delay_ms(), + SendPacing::MAX_DELAY_MS, + "seconds->millis must not overflow into a negative" + ); + } + + /// n messages have n-1 gaps. Getting this wrong shows the user a + /// visibly wrong estimate on a 2-message batch. + #[test] + fn duration_counts_the_gaps_between_messages_not_the_messages() { + let p = SendPacing::from_seconds(10); + assert_eq!(p.total_duration_ms(0), 0); + assert_eq!(p.total_duration_ms(1), 0, "one message waits for nothing"); + assert_eq!(p.total_duration_ms(2), 10_000); + assert_eq!(p.total_duration_ms(11), 100_000); + } + + #[test] + fn duration_description_is_human_and_never_empty() { + assert_eq!(SendPacing::from_seconds(0).describe_duration(50), "a few seconds"); + assert_eq!(SendPacing::from_seconds(10).describe_duration(4), "about 30s"); + assert_eq!(SendPacing::from_seconds(60).describe_duration(4), "about 3m 0s"); + assert!(SendPacing::from_seconds(60) + .describe_duration(200) + .starts_with("about 3h")); + } + + #[test] + fn to_stay_under_is_free_when_the_batch_already_fits() { + let p = SendPacing::to_stay_under(30, 30 * 60 * 1000, 10); + assert!(p.is_immediate(), "10 messages fit in a 30 cap; no need to wait"); + } + + #[test] + fn to_stay_under_spreads_a_batch_that_does_not_fit() { + let p = SendPacing::to_stay_under(30, 30 * 60 * 1000, 200); + assert_eq!(p.delay_ms(), 60_000); + + let mut rl = SendRateLimiter::new(30, 30 * 60 * 1000); + let mut now = 0i64; + for _ in 0..200 { + assert!(rl.allow_at(now).is_ok()); + now += p.delay_ms(); + } + } + + #[test] + fn to_stay_under_handles_degenerate_input_without_panicking() { + assert!(SendPacing::to_stay_under(0, 1000, 10).is_immediate()); + assert!(SendPacing::to_stay_under(30, 1000, 0).is_immediate()); + } + #[test] fn a_bulk_blast_of_200_is_stopped_well_short() { // The scenario the bulk UI makes one tap away.