Compare commits
3 commits
7b1022658c
...
8d580c7fc7
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8d580c7fc7 | ||
|
|
351e6c2265 | ||
|
|
1c7c32e76e |
6 changed files with 938 additions and 4 deletions
1
Cargo.lock
generated
1
Cargo.lock
generated
|
|
@ -2724,6 +2724,7 @@ dependencies = [
|
||||||
"nigig-uikit",
|
"nigig-uikit",
|
||||||
"proptest",
|
"proptest",
|
||||||
"robius-contacts",
|
"robius-contacts",
|
||||||
|
"robius-file-picker",
|
||||||
"robius-sms",
|
"robius-sms",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -10,6 +10,9 @@ nigig-uikit = { path = "../../nigig-uikit" }
|
||||||
chrono = { version = "0.4", features = ["serde"] }
|
chrono = { version = "0.4", features = ["serde"] }
|
||||||
robius-sms = { version = "0.1.0", path = "../../robius-sms" }
|
robius-sms = { version = "0.1.0", path = "../../robius-sms" }
|
||||||
robius-contacts = { version = "0.1.0", path = "../../robius-contacts" }
|
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]
|
[dev-dependencies]
|
||||||
proptest = "1"
|
proptest = "1"
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,7 @@
|
||||||
pub mod sms_bulk_page;
|
pub mod sms_bulk_page;
|
||||||
pub mod directory_scraper;
|
pub mod directory_scraper;
|
||||||
pub mod companies_list;
|
pub mod companies_list;
|
||||||
|
pub mod recipient_csv;
|
||||||
|
|
||||||
use makepad_widgets::ScriptVm;
|
use makepad_widgets::ScriptVm;
|
||||||
|
|
||||||
|
|
|
||||||
460
crates/apps/nigig-sms/src/sms_frame/bulk/recipient_csv.rs
Normal file
460
crates/apps/nigig-sms/src/sms_frame/bulk/recipient_csv.rs
Normal file
|
|
@ -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<CsvRecipient>,
|
||||||
|
/// `(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<String> {
|
||||||
|
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 `+<country><number>` 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<String> {
|
||||||
|
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<String>]) -> (usize, Option<usize>, 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<String> = 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<CsvImport, String> {
|
||||||
|
// 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<Vec<String>> = 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<String> = 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}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -15,6 +15,7 @@ use super::companies_list::{
|
||||||
use super::directory_scraper::{
|
use super::directory_scraper::{
|
||||||
get_extraction_state, run_extraction, DirectoryScraperAction, ExtractionState,
|
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::lock_ext::LockRecover;
|
||||||
use crate::chats::sms_frame::sms_utils::now_ms;
|
use crate::chats::sms_frame::sms_utils::now_ms;
|
||||||
|
|
||||||
|
|
@ -31,6 +32,20 @@ struct BulkSendProgress {
|
||||||
/// D5: guards against two overlapping bulk sends.
|
/// D5: guards against two overlapping bulk sends.
|
||||||
static BULK_SEND_IN_FLIGHT: AtomicBool = AtomicBool::new(false);
|
static BULK_SEND_IN_FLIGHT: AtomicBool = AtomicBool::new(false);
|
||||||
static BULK_SEND_PROGRESS: Mutex<Option<BulkSendProgress>> = Mutex::new(None);
|
static BULK_SEND_PROGRESS: Mutex<Option<BulkSendProgress>> = 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<Option<Result<CsvImport, String>>> = Mutex::new(None);
|
||||||
|
|
||||||
static FILTER_TEXT: Mutex<String> = Mutex::new(String::new());
|
static FILTER_TEXT: Mutex<String> = Mutex::new(String::new());
|
||||||
static SHOW_FAVORITES: Mutex<bool> = Mutex::new(false);
|
static SHOW_FAVORITES: Mutex<bool> = 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 } }
|
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_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 } }
|
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 } }
|
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 } }
|
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_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_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 } }
|
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.
|
// D5: the bulk-send worker publishes progress and signals the UI.
|
||||||
if matches!(event, Event::Signal) {
|
if matches!(event, Event::Signal) {
|
||||||
self.drain_bulk_send_progress(cx);
|
self.drain_bulk_send_progress(cx);
|
||||||
|
self.drain_csv_import(cx);
|
||||||
}
|
}
|
||||||
|
|
||||||
let actions = cx.capture_actions(|cx| self.view.handle_event(cx, event, scope));
|
let actions = cx.capture_actions(|cx| self.view.handle_event(cx, event, scope));
|
||||||
|
|
@ -355,8 +399,28 @@ impl Widget for SmsBulkPage {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if self.view.button(cx, ids!(bulk_send_btn)).clicked(&actions) {
|
if self.view.button(cx, ids!(bulk_send_btn)).clicked(&actions) {
|
||||||
|
// 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);
|
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.
|
// F10 / A5: show what this body will actually COST, live.
|
||||||
//
|
//
|
||||||
// The old compose page rendered a hardcoded "0 / 160 characters"
|
// The old compose page rendered a hardcoded "0 / 160 characters"
|
||||||
|
|
@ -670,22 +734,38 @@ impl SmsBulkPage {
|
||||||
.as_ref()
|
.as_ref()
|
||||||
.is_some_and(|(b, n)| b == trimmed_body && *n == total);
|
.is_some_and(|(b, n)| b == trimmed_body && *n == total);
|
||||||
|
|
||||||
|
let pacing = self.current_pacing(cx);
|
||||||
|
|
||||||
if !confirmation_matches {
|
if !confirmation_matches {
|
||||||
self.pending_bulk_send = Some((trimmed_body.to_string(), total));
|
self.pending_bulk_send = Some((trimmed_body.to_string(), total));
|
||||||
let encoding = match seg.encoding {
|
let encoding = match seg.encoding {
|
||||||
robius_sms::SmsEncoding::Gsm7 => "GSM-7",
|
robius_sms::SmsEncoding::Gsm7 => "GSM-7",
|
||||||
robius_sms::SmsEncoding::Ucs2 => "Unicode",
|
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(
|
self.view.label(cx, ids!(bulk_status)).set_text(
|
||||||
cx,
|
cx,
|
||||||
&format!(
|
&format!(
|
||||||
"Send {} SMS ({} {} segment{} each, {} total)? \
|
"Send {} SMS ({} {} segment{} each, {} total)?{} \
|
||||||
This cannot be undone. Tap Send again to confirm.",
|
This cannot be undone. Tap Send again to confirm.",
|
||||||
total,
|
total,
|
||||||
seg.segments,
|
seg.segments,
|
||||||
encoding,
|
encoding,
|
||||||
if seg.segments == 1 { "" } else { "s" },
|
if seg.segments == 1 { "" } else { "s" },
|
||||||
total_segments,
|
total_segments,
|
||||||
|
timing,
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
self.view.redraw(cx);
|
self.view.redraw(cx);
|
||||||
|
|
@ -733,6 +813,7 @@ impl SmsBulkPage {
|
||||||
}
|
}
|
||||||
|
|
||||||
let body_owned = trimmed_body.to_string();
|
let body_owned = trimmed_body.to_string();
|
||||||
|
BULK_SEND_CANCEL.store(false, Ordering::Relaxed);
|
||||||
self.view
|
self.view
|
||||||
.label(cx, ids!(bulk_status))
|
.label(cx, ids!(bulk_status))
|
||||||
.set_text(cx, &format!("Sending 0/{total}…"));
|
.set_text(cx, &format!("Sending 0/{total}…"));
|
||||||
|
|
@ -747,11 +828,51 @@ impl SmsBulkPage {
|
||||||
let mut failed = 0usize;
|
let mut failed = 0usize;
|
||||||
let mut throttled = 0usize;
|
let mut throttled = 0usize;
|
||||||
|
|
||||||
for recipient in &recipients {
|
for (i, recipient) in recipients.iter().enumerate() {
|
||||||
if limiter.allow_at(now_ms()).is_err() {
|
// 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;
|
throttled = recipients.len() - sent - failed;
|
||||||
break;
|
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) {
|
match send_single(recipient, &body_owned) {
|
||||||
Ok(()) => sent += 1,
|
Ok(()) => sent += 1,
|
||||||
Err(_) => failed += 1,
|
Err(_) => failed += 1,
|
||||||
|
|
@ -775,10 +896,146 @@ impl SmsBulkPage {
|
||||||
done: true,
|
done: true,
|
||||||
});
|
});
|
||||||
BULK_SEND_IN_FLIGHT.store(false, Ordering::Relaxed);
|
BULK_SEND_IN_FLIGHT.store(false, Ordering::Relaxed);
|
||||||
|
BULK_SEND_CANCEL.store(false, Ordering::Relaxed);
|
||||||
SignalToUI::set_ui_signal();
|
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::<i64>() {
|
||||||
|
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<String> =
|
||||||
|
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<String> = 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.
|
/// Apply bulk-send progress published by the D5 worker, if any.
|
||||||
fn drain_bulk_send_progress(&mut self, cx: &mut Cx) {
|
fn drain_bulk_send_progress(&mut self, cx: &mut Cx) {
|
||||||
let Some(p) = BULK_SEND_PROGRESS.lock_recover().take() else {
|
let Some(p) = BULK_SEND_PROGRESS.lock_recover().take() else {
|
||||||
|
|
|
||||||
|
|
@ -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 {
|
impl BulkSendRequest {
|
||||||
/// Validate a bulk request.
|
/// Validate a bulk request.
|
||||||
///
|
///
|
||||||
|
|
@ -864,6 +973,109 @@ mod tests {
|
||||||
assert_eq!(rl.in_window(31), 30);
|
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]
|
#[test]
|
||||||
fn a_bulk_blast_of_200_is_stopped_well_short() {
|
fn a_bulk_blast_of_200_is_stopped_well_short() {
|
||||||
// The scenario the bulk UI makes one tap away.
|
// The scenario the bulk UI makes one tap away.
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue