cratefield_cli/
import_users.rs

1//! `fz import supabase users` (issue #659, step 2 of ADR 0026): moves a
2//! Supabase project's `auth.users` into a running venture's `auth-core`
3//! over the same admin import `fz auth import` uses (its `send_batch`, so
4//! there is one HTTP path), then records the source-to-account mapping the
5//! data step rewrites references through. `cratefield-import-supabase`'s
6//! `read_users` is the engine; the CLI flags are documented in `lib.rs`.
7//!
8//! Without `--apply` nothing is written: the server validates every user
9//! and reports a verdict, and no mapping row is recorded. `--apply` writes
10//! both and needs `--map-db-url-env`. Progress and the summary go to
11//! stderr; the JSON report (stdout, or `--report`) holds the counts, the
12//! skipped and unmapped ids and the server's verdict — never an email, a
13//! hash, a token or a credentialed URL.
14//!
15//! Exit code: non-zero only when the run cannot finish — a refused input, a
16//! failed read, batch or mapping write, or drift in the mapping table. A
17//! run that completes reporting `conflict` or `invalid` exits zero, like
18//! `fz auth import`: those are the server having answered, and the counts
19//! print either way.
20
21use std::path::PathBuf;
22
23/// Everything `fz import supabase users` was given. The field docs are the
24/// clap help, in `lib.rs`.
25pub struct UsersArgs {
26    pub db_url: Option<String>,
27    pub target: String,
28    pub admin_token_env: String,
29    /// Required with `--apply`.
30    pub map_db_url_env: Option<String>,
31    pub oidc_providers: Vec<String>,
32    pub batch_size: usize,
33    pub merge_by_email: bool,
34    /// `--apply`: write. Absent, the run is a dry run.
35    pub apply: bool,
36    /// Where the JSON report goes. Defaults to stdout.
37    pub report: Option<PathBuf>,
38}
39
40#[cfg(feature = "import-supabase")]
41mod imp {
42    use std::collections::{BTreeMap, BTreeSet};
43
44    use serde::Serialize;
45    use serde_json::Value;
46
47    use crate::EnvVars;
48    use crate::auth_import::{self, ResultRow};
49    use crate::import::source_db_url;
50    use cratefield_core::Config as _;
51    use cratefield_import_supabase as engine;
52
53    use super::UsersArgs;
54
55    /// The statuses the summary prints, in order.
56    const STATUSES: [&str; 5] = ["created", "unchanged", "merged", "conflict", "invalid"];
57
58    /// The statuses whose row names the account the import created for the
59    /// user, so a mapping row is recorded for them.
60    const MAPPED_STATUSES: [&str; 3] = ["created", "unchanged", "merged"];
61
62    /// The JSON report. `skipped`, `unmapped`, `mapping` and `results` are
63    /// the engine's and the server's own shapes, so they are not restated;
64    /// none of them names an email or a hash.
65    #[derive(Debug, Serialize)]
66    struct Report {
67        dry_run: bool,
68        counts: BTreeMap<String, usize>,
69        mapping: engine::MappingCounts,
70        skipped: Vec<engine::SkippedUser>,
71        unmapped: Vec<engine::UnmappedProvider>,
72        results: Vec<ResultRow>,
73    }
74
75    /// `fz import supabase users`: read, send, map, report.
76    ///
77    /// # Errors
78    ///
79    /// A refused input (a URL, a token, an out-of-range `--batch-size`, a
80    /// missing variable), any read or batch that fails, a mapping write
81    /// that drifts, or a report that cannot be written.
82    pub fn run(args: &UsersArgs) -> Result<(), String> {
83        if args.batch_size == 0 || args.batch_size > auth_import::MAX_BATCH_SIZE {
84            return Err(format!(
85                "--batch-size must be between 1 and {}, got {}",
86                auth_import::MAX_BATCH_SIZE,
87                args.batch_size
88            ));
89        }
90        let (db_url, note) = source_db_url(args.db_url.as_deref(), &EnvVars)?;
91        if let Some(note) = note {
92            eprintln!("{note}");
93        }
94        let admin_token = EnvVars.get(&args.admin_token_env).ok_or_else(|| {
95            format!(
96                "the environment variable `{}` is not set or is empty (it must hold the \
97                 auth-core admin token — the value the venture has as `ADMIN_TOKEN`)",
98                args.admin_token_env
99            )
100        })?;
101        let map_url = if args.apply {
102            let name = args.map_db_url_env.as_deref().ok_or_else(|| {
103                "--apply needs --map-db-url-env naming the environment variable that holds the \
104                 *target* database URL (the harness Postgres, where `import_supabase_users` \
105                 lives)"
106                    .to_owned()
107            })?;
108            let url = EnvVars.get(name).ok_or_else(|| {
109                format!(
110                    "the environment variable `{name}` is not set or is empty (it must hold the \
111                     *target* database URL — the harness Postgres, where `import_supabase_users` \
112                     lives)"
113                )
114            })?;
115            Some(engine::Secret::new(url))
116        } else {
117            None
118        };
119
120        let options = engine::UsersOptions {
121            db_url: engine::Secret::new(db_url.expose()),
122            batch_size: args.batch_size,
123            oidc_providers: args.oidc_providers.iter().cloned().collect::<BTreeSet<_>>(),
124        };
125        let runtime = tokio::runtime::Builder::new_current_thread()
126            .enable_all()
127            .build()
128            .map_err(|err| format!("cannot start the async runtime this import needs: {err}"))?;
129        runtime.block_on(transfer(args, &options, &admin_token, map_url.as_ref()))
130    }
131
132    /// Reads the plan, sends every batch, records the mapping and writes
133    /// the report.
134    async fn transfer(
135        args: &UsersArgs,
136        options: &engine::UsersOptions,
137        admin_token: &str,
138        map_url: Option<&engine::Secret>,
139    ) -> Result<(), String> {
140        let (http, _clock) = auth_import::native_stack();
141        let plan = engine::read_users(options)
142            .await
143            .map_err(|error| error.to_string())?;
144        // Each record rides with the metadata read alongside it, both in id
145        // order, so a result row pairs with its user by position: the server
146        // answers one verdict per user in request order.
147        let users: Vec<(&engine::ImportRecord, &engine::UserMetadata)> = plan.users().collect();
148        let url = auth_import::import_url(&args.target);
149        let dry_run = !args.apply;
150        let total = users.len();
151        let mut results: Vec<ResultRow> = Vec::with_capacity(total);
152        let mut mapped: Vec<engine::MappedUser> = Vec::new();
153        for chunk in users.chunks(args.batch_size) {
154            let body: Vec<Value> = chunk
155                .iter()
156                .map(|(record, _)| serde_json::to_value(record))
157                .collect::<Result<_, _>>()
158                .map_err(|error| format!("cannot encode a user record: {error}"))?;
159            let rows = auth_import::send_batch(
160                &http,
161                &url,
162                admin_token,
163                args.merge_by_email,
164                dry_run,
165                &body,
166            )
167            .await?;
168            for (row, (_, metadata)) in rows.iter().zip(chunk) {
169                if MAPPED_STATUSES.contains(&row.status.as_str())
170                    && let Some(sub) = row.sub.as_deref()
171                {
172                    mapped.push(metadata.mapping(sub));
173                }
174            }
175            results.extend(rows);
176            print_progress(results.len(), total, &results);
177        }
178
179        let mapping = match map_url {
180            Some(url) if !mapped.is_empty() => {
181                let mut conn = engine::connect_mapping(url)
182                    .await
183                    .map_err(|error| error.to_string())?;
184                engine::record_mapping(&mut conn, url, &mapped)
185                    .await
186                    .map_err(|error| error.to_string())?
187            }
188            _ => engine::MappingCounts::default(),
189        };
190
191        let report = Report {
192            dry_run,
193            counts: counts_by_status(&results),
194            mapping,
195            skipped: plan.skipped,
196            unmapped: plan.unmapped,
197            results,
198        };
199        let json = serde_json::to_string_pretty(&report)
200            .map_err(|error| format!("cannot encode the report: {error}"))?;
201        match &args.report {
202            Some(path) => std::fs::write(path, json)
203                .map_err(|error| format!("cannot write {}: {error}", path.display()))?,
204            None => println!("{json}"),
205        }
206        print_summary(&report, total, args.apply);
207        Ok(())
208    }
209
210    /// How many results carry each status. Every status is present, zero
211    /// included, so a status that never appeared is visibly a zero.
212    fn counts_by_status(results: &[ResultRow]) -> BTreeMap<String, usize> {
213        let mut counts: BTreeMap<String, usize> =
214            STATUSES.iter().map(|s| ((*s).to_owned(), 0)).collect();
215        for row in results {
216            *counts.entry(row.status.clone()).or_default() += 1;
217        }
218        counts
219    }
220
221    fn count(counts: &BTreeMap<String, usize>, status: &str) -> usize {
222        counts.get(status).copied().unwrap_or(0)
223    }
224
225    /// One line per batch, on stderr: `users: 500/1234 sent (created 480,
226    /// unchanged 20, merged 0, conflict 0, invalid 0)`. Counts are
227    /// cumulative, so the last line is the run.
228    fn print_progress(sent: usize, total: usize, results: &[ResultRow]) {
229        let counts = counts_by_status(results);
230        eprintln!(
231            "users: {sent}/{total} sent (created {}, unchanged {}, merged {}, conflict {}, invalid \
232             {})",
233            count(&counts, "created"),
234            count(&counts, "unchanged"),
235            count(&counts, "merged"),
236            count(&counts, "conflict"),
237            count(&counts, "invalid"),
238        );
239    }
240
241    /// The human summary, on stderr (stdout is the JSON report).
242    fn print_summary(report: &Report, total: usize, apply: bool) {
243        if report.dry_run {
244            eprintln!(
245                "fz import supabase users: dry run — the server validated {total} user(s) and \
246                 wrote nothing; pass --apply to import"
247            );
248        } else {
249            eprintln!("fz import supabase users: imported {total} user(s)");
250        }
251        for status in STATUSES {
252            eprintln!("  {status}: {}", count(&report.counts, status));
253        }
254        if apply {
255            eprintln!(
256                "  mapping inserted: {}, unchanged: {}",
257                report.mapping.inserted, report.mapping.unchanged
258            );
259        }
260        if !report.skipped.is_empty() {
261            let mut reasons: BTreeMap<&str, usize> = BTreeMap::new();
262            for user in &report.skipped {
263                *reasons.entry(user.reason.as_str()).or_default() += 1;
264            }
265            let reasons: Vec<String> = reasons.iter().map(|(r, n)| format!("{r} {n}")).collect();
266            eprintln!(
267                "  skipped: {} ({})",
268                report.skipped.len(),
269                reasons.join(", ")
270            );
271        }
272        if !report.unmapped.is_empty() {
273            let providers: Vec<String> = report
274                .unmapped
275                .iter()
276                .map(|p| format!("{} ({})", p.provider, p.supabase_ids.len()))
277                .collect();
278            eprintln!(
279                "  unmapped: {} — these users can sign in with a magic link",
280                providers.join(", ")
281            );
282        }
283    }
284
285    #[cfg(test)]
286    mod tests {
287        use super::*;
288
289        #[test]
290        fn the_report_carries_counts_skips_and_unmapped_but_no_email_or_hash() {
291            let email = "ada@example.test";
292            let hash = "$2b$12$abcdefghijklmnopqrstuuOJ4f0Zf1rSYx2Zl8pQ1fSl8sQf2aZ1u";
293            let plan = engine::ImportPlan {
294                records: vec![engine::ImportRecord {
295                    external_provider: engine::EXTERNAL_PROVIDER,
296                    external_id: "00000000-0000-4000-8000-000000000001".to_owned(),
297                    email: email.to_owned(),
298                    email_verified: true,
299                    password_hash: Some(engine::Secret::new(hash)),
300                    created_at: None,
301                    identities: vec![engine::ImportIdentity {
302                        provider: "google".to_owned(),
303                        subject: "sub-1".to_owned(),
304                    }],
305                }],
306                metadata: Vec::new(),
307                skipped: vec![engine::SkippedUser {
308                    supabase_id: "00000000-0000-4000-8000-000000000002".to_owned(),
309                    reason: engine::SkipReason::NoEmail,
310                }],
311                unmapped: vec![engine::UnmappedProvider {
312                    provider: "github".to_owned(),
313                    supabase_ids: vec!["00000000-0000-4000-8000-000000000003".to_owned()],
314                }],
315            };
316            let results = vec![ResultRow {
317                external_provider: "supabase".to_owned(),
318                external_id: "00000000-0000-4000-8000-000000000001".to_owned(),
319                status: "created".to_owned(),
320                sub: Some("sub-1".to_owned()),
321                reason: None,
322            }];
323            let report = Report {
324                dry_run: false,
325                counts: counts_by_status(&results),
326                mapping: engine::MappingCounts::default(),
327                skipped: plan.skipped,
328                unmapped: plan.unmapped,
329                results,
330            };
331            let json = serde_json::to_string(&report).expect("serializes");
332
333            assert!(!json.contains(email), "the report carries no email: {json}");
334            assert!(!json.contains(hash), "the report carries no hash: {json}");
335            assert!(!json.contains("password_hash"), "{json}");
336            // The skipped id and its reason, and the unmapped provider, are
337            // exactly what the operator acts on.
338            assert!(json.contains("no-email"), "{json}");
339            assert!(json.contains("github"), "{json}");
340            assert_eq!(count(&report.counts, "created"), 1);
341            assert_eq!(count(&report.counts, "conflict"), 0);
342        }
343    }
344}
345
346#[cfg(feature = "import-supabase")]
347pub use imp::run;
348
349/// The refusal an `fz` without the feature prints: one line naming the
350/// install, mirroring `import::inspect`'s.
351///
352/// # Errors
353///
354/// Always: this build has no `import-supabase` feature, so it cannot reach
355/// Postgres or the network.
356#[cfg(not(feature = "import-supabase"))]
357pub fn run(_args: &UsersArgs) -> Result<(), String> {
358    Err(
359        "this `fz` was built without the `import-supabase` feature, so it cannot move auth \
360         users. Install one that has it — `cargo install cratefield-cli --features \
361         import-supabase` — and run that binary (it needs no compiled-in harness)."
362            .to_owned(),
363    )
364}
cratefield_cli/import_users.rs

cratefield_cli/
import_users.rs
