use std::collections::{BTreeMap, HashSet};
use std::path::{Path, PathBuf};
#[cfg(any(feature = "auth-import", test))]
use cratefield_core::HttpClient;
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[cfg(any(feature = "auth-import", test))]
use std::sync::Arc;
pub const IMPORT_PATH: &str = "/v1/auth-core/admin/users/import";
pub const DEFAULT_BATCH_SIZE: usize = 500;
pub const MAX_BATCH_SIZE: usize = 1000;
const STATUSES: [&str; 5] = ["created", "unchanged", "merged", "conflict", "invalid"];
pub struct ImportOptions {
pub file: PathBuf,
pub target: String,
pub admin_token: String,
pub apply: bool,
pub merge_by_email: bool,
pub report: Option<PathBuf>,
pub batch_size: usize,
}
impl ImportOptions {
#[must_use]
pub fn dry_run(&self) -> bool {
!self.apply
}
#[must_use]
pub fn report_path(&self) -> PathBuf {
if let Some(path) = &self.report {
return path.clone();
}
let mut name = self.file.clone().into_os_string();
name.push(".report.jsonl");
PathBuf::from(name)
}
}
#[derive(Debug, Serialize, Deserialize)]
pub(crate) struct ResultRow {
pub(crate) external_provider: String,
pub(crate) external_id: String,
pub(crate) status: String,
pub(crate) sub: Option<String>,
pub(crate) reason: Option<String>,
}
#[cfg(any(feature = "auth-import", test))]
#[derive(Deserialize)]
struct ImportResponse {
results: Vec<ResultRow>,
}
#[cfg(any(feature = "auth-import", test))]
#[derive(Serialize)]
struct ImportRequest<'a> {
dry_run: bool,
merge_by_email: bool,
users: &'a [Value],
}
#[derive(Debug)]
struct Outcome {
dry_run: bool,
results: Vec<ResultRow>,
}
pub fn run(options: &ImportOptions) -> Result<(), String> {
let users = prepare(options)?;
let outcome = import_dispatch(options, &users)?;
finish(options, &outcome)
}
fn prepare(options: &ImportOptions) -> Result<Vec<Value>, String> {
if options.batch_size == 0 || options.batch_size > MAX_BATCH_SIZE {
return Err(format!(
"--batch-size must be between 1 and {MAX_BATCH_SIZE}, got {}",
options.batch_size
));
}
load_users(&options.file)
}
fn load_users(path: &Path) -> Result<Vec<Value>, String> {
let text = std::fs::read_to_string(path)
.map_err(|err| format!("cannot read {}: {err}", path.display()))?;
let mut users = Vec::new();
let mut seen: HashSet<(String, String)> = HashSet::new();
for (index, raw) in text.lines().enumerate() {
let line = index + 1;
if raw.trim().is_empty() {
continue;
}
let value: Value =
serde_json::from_str(raw).map_err(|err| format!("line {line}: not JSON: {err}"))?;
let Some(object) = value.as_object() else {
return Err(format!("line {line}: not a JSON object"));
};
for field in ["external_provider", "external_id", "email"] {
match object.get(field) {
Some(Value::String(text)) if !text.is_empty() => {}
Some(Value::String(_)) => {
return Err(format!("line {line}: `{field}` must not be empty"));
}
Some(_) => return Err(format!("line {line}: `{field}` must be a string")),
None => return Err(format!("line {line}: missing required field `{field}`")),
}
}
match object.get("email_verified") {
Some(Value::Bool(_)) => {}
Some(_) => return Err(format!("line {line}: `email_verified` must be a boolean")),
None => {
return Err(format!(
"line {line}: missing required field `email_verified`"
));
}
}
let provider = object["external_provider"].as_str().unwrap_or_default();
let id = object["external_id"].as_str().unwrap_or_default();
if !seen.insert((provider.to_owned(), id.to_owned())) {
return Err(format!(
"line {line}: duplicate (external_provider, external_id) — ({provider:?}, {id:?}) \
appears twice"
));
}
users.push(value);
}
if users.is_empty() {
return Err(format!("{}: no users to import", path.display()));
}
Ok(users)
}
#[cfg(any(feature = "auth-import", test))]
pub(crate) fn import_url(target: &str) -> String {
format!("{}{IMPORT_PATH}", target.trim_end_matches('/'))
}
#[cfg(any(feature = "auth-import", test))]
async fn import_all(
http: &Arc<dyn HttpClient>,
options: &ImportOptions,
users: &[Value],
) -> Result<Outcome, String> {
let url = import_url(&options.target);
let dry_run = options.dry_run();
let mut results = Vec::with_capacity(users.len());
for batch in users.chunks(options.batch_size) {
results.extend(
send_batch(
http,
&url,
&options.admin_token,
options.merge_by_email,
dry_run,
batch,
)
.await?,
);
}
Ok(Outcome { dry_run, results })
}
#[cfg(any(feature = "auth-import", test))]
pub(crate) async fn send_batch(
http: &Arc<dyn HttpClient>,
url: &str,
admin_token: &str,
merge_by_email: bool,
dry_run: bool,
users: &[Value],
) -> Result<Vec<ResultRow>, String> {
let uri = url
.parse::<http::Uri>()
.map_err(|_| format!("`{url}` is not a valid URL"))?;
let body = serde_json::to_vec(&ImportRequest {
dry_run,
merge_by_email,
users,
})
.map_err(|err| format!("cannot encode the import request: {err}"))?;
let authorization =
http::HeaderValue::from_str(&format!("Bearer {admin_token}")).map_err(|_| {
"the environment variable `--admin-token-env` named does not hold a value usable as \
an `Authorization` header"
.to_owned()
})?;
let request = http::Request::builder()
.method(http::Method::POST)
.uri(uri)
.header(http::header::AUTHORIZATION, authorization)
.header(http::header::CONTENT_TYPE, "application/json")
.body(bytes::Bytes::from(body))
.map_err(|err| format!("cannot build the import request: {err}"))?;
let response = http
.send(request)
.await
.map_err(|err| format!("the import request to {url} failed: {err}"))?;
let status = response.status();
if !status.is_success() {
return Err(format!(
"the import request to {url} failed: HTTP {status}{}",
problem_detail(response.body())
));
}
let parsed: ImportResponse = serde_json::from_slice(response.body())
.map_err(|err| format!("the import response was not the expected JSON: {err}"))?;
if parsed.results.len() != users.len() {
return Err(format!(
"the server returned {} results for {} users",
parsed.results.len(),
users.len()
));
}
Ok(parsed.results)
}
#[cfg(any(feature = "auth-import", test))]
fn problem_detail(body: &[u8]) -> String {
let Ok(value) = serde_json::from_slice::<Value>(body) else {
return String::new();
};
value
.get("detail")
.and_then(Value::as_str)
.or_else(|| value.get("title").and_then(Value::as_str))
.map(|detail| format!(" — {detail}"))
.unwrap_or_default()
}
fn finish(options: &ImportOptions, outcome: &Outcome) -> Result<(), String> {
let path = options.report_path();
write_report(&path, &outcome.results)?;
print_summary(outcome, &path);
Ok(())
}
fn write_report(path: &Path, results: &[ResultRow]) -> Result<(), String> {
let mut out = String::new();
for row in results {
let line = serde_json::to_string(row)
.map_err(|err| format!("cannot encode a report row: {err}"))?;
out.push_str(&line);
out.push('\n');
}
std::fs::write(path, out).map_err(|err| format!("cannot write {}: {err}", path.display()))
}
fn print_summary(outcome: &Outcome, path: &Path) {
if outcome.dry_run {
println!(
"fz auth import: dry run — the server validated {} user(s) and wrote nothing; pass \
--apply to import",
outcome.results.len()
);
} else {
println!("fz auth import: imported {} user(s)", outcome.results.len());
}
let mut counts: BTreeMap<&str, usize> = BTreeMap::new();
for row in &outcome.results {
*counts.entry(row.status.as_str()).or_default() += 1;
}
for status in STATUSES {
println!(" {status}: {}", counts.get(status).copied().unwrap_or(0));
}
println!("report: {}", path.display());
}
#[cfg(feature = "auth-import")]
fn import_dispatch(options: &ImportOptions, users: &[Value]) -> Result<Outcome, String> {
let (http, _clock) = native_stack();
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|err| format!("cannot start the async runtime this import needs: {err}"))?;
runtime.block_on(import_all(&http, options, users))
}
#[cfg(feature = "auth-import")]
pub(crate) fn native_stack() -> (Arc<dyn HttpClient>, Arc<dyn cratefield_core::Clock>) {
let clock: Arc<dyn cratefield_core::Clock> = Arc::new(cratefield_runtime_native::TokioClock);
let http: Arc<dyn HttpClient> = Arc::new(cratefield_core::BoundedHttpClient::new(
Arc::new(cratefield_runtime_native::ReqwestClient::new()),
Arc::clone(&clock),
));
(http, clock)
}
#[cfg(not(feature = "auth-import"))]
fn import_dispatch(_options: &ImportOptions, _users: &[Value]) -> Result<Outcome, String> {
Err(
"this `fz` was built without the `auth-import` feature, so it has no HTTP client and \
cannot import. Install one that has it — `cargo install cratefield-cli --features \
auth-import` — and run that binary (it needs no compiled-in harness); do not add the \
feature to the venture's own `cratefield-cli` dependency, which would pull the native \
runtime into its wasm build."
.to_owned(),
)
}
#[cfg(test)]
mod tests {
use super::*;
use cratefield_core::HttpError;
use cratefield_testing::{FakeHttpClient, TempDir};
use http::Response;
fn json_response(body: &str) -> Result<Response<bytes::Bytes>, HttpError> {
Response::builder()
.status(200)
.body(bytes::Bytes::from(body.to_owned()))
.map_err(|err| HttpError::Transport(err.to_string()))
}
fn result_body(pairs: &[(&str, &str)]) -> String {
let results: Vec<Value> = pairs
.iter()
.map(|(provider, id)| {
serde_json::json!({
"external_provider": provider,
"external_id": id,
"status": "created",
"sub": format!("sub-{id}"),
"reason": Value::Null,
})
})
.collect();
serde_json::json!({ "dry_run": true, "results": results }).to_string()
}
fn options(dir: &TempDir, rows: &[(&str, &str)], batch_size: usize) -> ImportOptions {
let file = dir.join("users.jsonl");
let mut text = String::new();
for (provider, id) in rows {
use std::fmt::Write as _;
writeln!(
text,
"{{\"external_provider\":{provider:?},\"external_id\":{id:?},\
\"email\":\"u{id}@example.com\",\"email_verified\":true}}"
)
.expect("writes to a String");
}
std::fs::write(&file, text).expect("writes the fixture");
ImportOptions {
file,
target: "https://venture.example/".to_owned(),
admin_token: "test-admin-token-0123456789abcdef".to_owned(),
apply: false,
merge_by_email: false,
report: None,
batch_size,
}
}
#[cfg(test)]
fn import_with(http: &Arc<dyn HttpClient>, options: &ImportOptions) -> Result<Outcome, String> {
let users = prepare(options)?;
pollster::block_on(import_all(http, options, &users))
}
#[test]
fn batching_splits_into_the_requested_sizes() {
let dir = TempDir::new("fz-auth-import-batch");
let options = options(
&dir,
&[("p", "1"), ("p", "2"), ("p", "3"), ("p", "4"), ("p", "5")],
2,
);
let fake = FakeHttpClient::scripted(vec![
json_response(&result_body(&[("p", "1"), ("p", "2")])),
json_response(&result_body(&[("p", "3"), ("p", "4")])),
json_response(&result_body(&[("p", "5")])),
]);
let http: Arc<dyn HttpClient> = Arc::new(fake.clone());
let outcome = import_with(&http, &options).expect("imports");
assert_eq!(outcome.results.len(), 5);
let captured = fake.captured();
assert_eq!(captured.len(), 3, "5 users at 2 per batch is 3 requests");
for (method, uri, _) in &captured {
assert_eq!(method, "POST");
assert_eq!(
uri,
"https://venture.example/v1/auth-core/admin/users/import"
);
}
let sizes: Vec<usize> = captured
.iter()
.map(|(_, _, body)| {
serde_json::from_str::<Value>(body).expect("request is JSON")["users"]
.as_array()
.expect("users array")
.len()
})
.collect();
assert_eq!(sizes, vec![2, 2, 1]);
}
#[test]
fn the_report_never_carries_an_email_or_a_password_hash() {
let dir = TempDir::new("fz-auth-import-report-privacy");
let email = "alice@example.com";
let hash = "argon2id$v=19$m=19456,t=2,p=1$c2FsdA$c2FsdA";
let file = dir.join("users.jsonl");
std::fs::write(
&file,
format!(
"{{\"external_provider\":\"legacy\",\"external_id\":\"42\",\"email\":\"{email}\",\
\"email_verified\":true,\"password_hash\":\"{hash}\"}}\n"
),
)
.expect("writes the fixture");
let options = ImportOptions {
file,
target: "https://venture.example".to_owned(),
admin_token: "test-admin-token-0123456789abcdef".to_owned(),
apply: false,
merge_by_email: false,
report: None,
batch_size: DEFAULT_BATCH_SIZE,
};
let fake = FakeHttpClient::scripted(vec![json_response(&result_body(&[("legacy", "42")]))]);
let http: Arc<dyn HttpClient> = Arc::new(fake.clone());
let outcome = import_with(&http, &options).expect("imports");
let report = options.report_path();
write_report(&report, &outcome.results).expect("writes the report");
let bytes = std::fs::read_to_string(&report).expect("reads the report");
assert!(
!bytes.contains(email),
"the report carries no email: {bytes}"
);
assert!(!bytes.contains(hash), "the report carries no hash: {bytes}");
let (_, _, request_body) = fake.captured().into_iter().next().expect("one request");
assert!(
request_body.contains(email),
"the request carries the email"
);
assert!(request_body.contains(hash), "the request carries the hash");
}
#[test]
fn a_missing_field_aborts_locally_before_any_request() {
let dir = TempDir::new("fz-auth-import-missing-field");
let file = dir.join("users.jsonl");
std::fs::write(
&file,
"{\"external_provider\":\"p\",\"external_id\":\"1\",\"email_verified\":true}\n",
)
.expect("writes the fixture");
let options = ImportOptions {
file,
target: "https://venture.example".to_owned(),
admin_token: "test-admin-token-0123456789abcdef".to_owned(),
apply: false,
merge_by_email: false,
report: None,
batch_size: DEFAULT_BATCH_SIZE,
};
let fake = FakeHttpClient::ok_json("{}");
let http: Arc<dyn HttpClient> = Arc::new(fake.clone());
let error = import_with(&http, &options).expect_err("email is required");
assert!(error.contains("line 1"), "{error}");
assert!(error.contains("email"), "{error}");
assert!(fake.captured().is_empty(), "nothing was sent");
}
#[test]
fn a_duplicate_id_aborts_locally_before_any_request() {
let dir = TempDir::new("fz-auth-import-duplicate");
let options = options(&dir, &[("p", "7"), ("p", "7")], DEFAULT_BATCH_SIZE);
let fake = FakeHttpClient::ok_json("{}");
let http: Arc<dyn HttpClient> = Arc::new(fake.clone());
let error = import_with(&http, &options).expect_err("the id is duplicated");
assert!(error.contains("line 2"), "{error}");
assert!(error.contains("duplicate"), "{error}");
assert!(fake.captured().is_empty(), "nothing was sent");
}
#[test]
fn a_line_that_is_not_an_object_is_a_local_error_with_its_line_number() {
let dir = TempDir::new("fz-auth-import-not-object");
let file = dir.join("users.jsonl");
std::fs::write(&file, "\n[1, 2, 3]\n").expect("writes the fixture");
let error = load_users(&file).expect_err("an array is not a user");
assert!(error.contains("line 2"), "{error}");
assert!(error.contains("JSON object"), "{error}");
}
#[test]
fn dry_run_is_true_by_default_and_false_with_apply() {
let dir = TempDir::new("fz-auth-import-dry-run");
let dry = options(&dir, &[("p", "1")], DEFAULT_BATCH_SIZE);
let fake = FakeHttpClient::scripted(vec![json_response(&result_body(&[("p", "1")]))]);
let http: Arc<dyn HttpClient> = Arc::new(fake.clone());
assert!(dry.dry_run());
import_with(&http, &dry).expect("dry-runs");
let body = fake.captured().into_iter().next().expect("one request").2;
assert!(body.contains("\"dry_run\":true"), "{body}");
let mut apply = options(&dir, &[("p", "1")], DEFAULT_BATCH_SIZE);
apply.apply = true;
apply.merge_by_email = true;
let fake = FakeHttpClient::scripted(vec![json_response(&result_body(&[("p", "1")]))]);
let http: Arc<dyn HttpClient> = Arc::new(fake.clone());
assert!(!apply.dry_run());
import_with(&http, &apply).expect("applies");
let body = fake.captured().into_iter().next().expect("one request").2;
assert!(body.contains("\"dry_run\":false"), "{body}");
assert!(body.contains("\"merge_by_email\":true"), "{body}");
}
#[test]
fn an_out_of_range_batch_size_is_a_local_error() {
let dir = TempDir::new("fz-auth-import-batch-range");
for batch_size in [0, MAX_BATCH_SIZE + 1] {
let options = options(&dir, &[("p", "1")], batch_size);
let error = prepare(&options).expect_err("out of range");
assert!(error.contains("--batch-size"), "{error}");
}
}
#[test]
fn a_non_2xx_batch_fails_the_run_with_the_problem_detail() {
let dir = TempDir::new("fz-auth-import-non-2xx");
let options = options(&dir, &[("p", "1")], DEFAULT_BATCH_SIZE);
let fake = FakeHttpClient::scripted(vec![
Response::builder()
.status(400)
.body(bytes::Bytes::from(
r#"{"type":"about:blank","title":"Bad Request","detail":"too many users"}"#,
))
.map_err(|err| HttpError::Transport(err.to_string())),
]);
let http: Arc<dyn HttpClient> = Arc::new(fake.clone());
let error = import_with(&http, &options).expect_err("a 400 fails the run");
assert!(error.contains("400"), "{error}");
assert!(error.contains("too many users"), "{error}");
}
#[test]
#[cfg(not(feature = "auth-import"))]
fn the_refusal_points_at_an_installed_binary() {
let dir = TempDir::new("fz-auth-import-refusal");
let options = options(&dir, &[("p", "1")], DEFAULT_BATCH_SIZE);
let error = run(&options).expect_err("no client, no import");
assert!(
error.contains("cargo install cratefield-cli --features auth-import"),
"{error}"
);
assert!(!error.contains("cratefield-cli = {"), "{error}");
}
}