cratefield_import_supabase/
lib.rs

1//! `cratefield-import-supabase`: step one of moving a Supabase project onto
2//! the harness (ADR 0026, issue #658). [`inspect`] connects read-only to the
3//! project's Postgres — plus, optionally, the Supabase Management API for
4//! what Postgres cannot see — and returns a [`Report`]: every schema object,
5//! RLS policy, auth provider, storage bucket, Edge Function and Realtime
6//! publication, each classified as automatic, needs work or a blocker, with
7//! a size and transfer-time estimate.
8//!
9//! It never writes. The later steps (auth users, schema and data, storage,
10//! cutover) read the report's JSON; `docs/import/supabase-report.md`
11//! documents it.
12//!
13//! Step two (issue #659) is [`read_users`]: the same read-only snapshot,
14//! but rows — the auth users as the admin import contract, plus the map
15//! [`record_mapping`] writes so the data step can rewrite their references.
16//!
17//! ```no_run
18//! # async fn demo() -> Result<(), cratefield_import_supabase::InspectError> {
19//! use cratefield_import_supabase::{InspectOptions, Secret, inspect};
20//!
21//! let options = InspectOptions::new("abcdefghijklmnopqrst", Secret::new(std::env::var("SUPABASE_DB_URL").unwrap()));
22//! let report = inspect(&options).await?;
23//! println!("{}", report.to_markdown());
24//! # Ok(())
25//! # }
26//! ```
27
28#![forbid(unsafe_code)]
29
30// Native only: the sqlx driver does not compile to wasm, and an importer is
31// an operator's command, never something a Worker serves.
32// `no_wasm_package_can_reach_a_native_only_crate` in
33// `crates/cli-acceptance/tests/native_only_off_wasm.rs` finds this crate by
34// the guard below.
35#[cfg(any(target_arch = "wasm32", target_arch = "wasm64"))]
36compile_error!(
37    "cratefield-import-supabase is native-only: it reads Postgres through sqlx, which does not \
38     compile to wasm (ADR 0026). Run it from the fz CLI, never from a Worker."
39);
40
41mod classify;
42mod collect;
43pub mod dispositions;
44pub mod extensions;
45pub mod management;
46mod markdown;
47pub mod plan;
48pub mod policy;
49pub mod report;
50mod secret;
51mod session;
52mod storage_policy;
53mod users;
54mod visibility;
55
56use std::sync::Arc;
57
58use cratefield_core::Classifier;
59
60pub use dispositions::{
61    DispositionsError, DispositionsFile, apply as apply_dispositions, skeleton,
62};
63pub use management::{AuthConfig, DEFAULT_API_BASE, ManagementApi, ManagementError};
64pub use plan::{
65    PLAN_VERSION, Plan, PlanBlocker, PlanDrift, TargetFacts, build_plan, check_drift,
66    inspection_hash, read_target,
67};
68pub use report::*;
69pub use secret::{Secret, SourceName, source_name};
70pub use session::{InspectError, ReadOnlySession};
71pub use users::{
72    DEFAULT_BATCH_SIZE, EXTERNAL_PROVIDER, ImportIdentity, ImportPlan, ImportRecord, MappedUser,
73    MappingCounts, MappingError, SkipReason, SkippedUser, UnmappedProvider, UserMetadata,
74    UsersOptions, connect_mapping, read_users, record_mapping,
75};
76
77/// The throughput the transfer estimate assumes when none is given, in
78/// megabits per second.
79pub const DEFAULT_TRANSFER_MBPS: u32 = 100;
80
81/// What [`inspect`] needs.
82pub struct InspectOptions {
83    /// The Supabase project ref.
84    pub project_ref: String,
85    /// The database URL, ideally for the read-only role.
86    pub db_url: Secret,
87    /// The Management API, when a token was given.
88    pub management: Option<ManagementApi>,
89    /// The classifier asked about policies no rule placed. `None` leaves
90    /// them `needs_review`.
91    pub classifier: Option<Arc<dyn Classifier>>,
92    /// The confidence a classifier answer needs (per adapter; see
93    /// [`policy`]).
94    pub classify_threshold: f32,
95    /// The transfer estimate's throughput.
96    pub transfer_mbps: u32,
97}
98
99impl InspectOptions {
100    /// Options with no Management API, no classifier, the default
101    /// threshold and throughput.
102    #[must_use]
103    pub fn new(project_ref: impl Into<String>, db_url: Secret) -> Self {
104        Self {
105            project_ref: project_ref.into(),
106            db_url,
107            management: None,
108            classifier: None,
109            classify_threshold: policy::DEFAULT_THRESHOLD,
110            transfer_mbps: DEFAULT_TRANSFER_MBPS,
111        }
112    }
113}
114
115/// A project ref is lower-case letters and digits: it goes into a URL path.
116///
117/// # Errors
118///
119/// The refusal, when it is not one.
120pub fn validate_project_ref(project_ref: &str) -> Result<(), String> {
121    let ok = !project_ref.is_empty()
122        && project_ref.len() <= 64
123        && project_ref
124            .chars()
125            .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit());
126    if ok {
127        Ok(())
128    } else {
129        Err(format!(
130            "{project_ref:?} is not a Supabase project ref (lower-case letters and digits, as in \
131             the project URL https://<ref>.supabase.co)"
132        ))
133    }
134}
135
136/// Inspects the project. Read-only throughout; see [`ReadOnlySession`].
137///
138/// # Errors
139///
140/// [`InspectError`] when the database cannot be reached or read, or the
141/// Management API refuses the token. A blocker is never an error: it is in
142/// the report.
143#[allow(clippy::too_many_lines)] // the report, assembled in field order
144pub async fn inspect(options: &InspectOptions) -> Result<Report, InspectError> {
145    validate_project_ref(&options.project_ref).map_err(InspectError::InvalidInput)?;
146    let source = source_name(&options.db_url).map_err(InspectError::InvalidInput)?;
147    if !(0.0..=1.0).contains(&options.classify_threshold) {
148        return Err(InspectError::InvalidInput(
149            "the classifier threshold must be between 0 and 1".to_owned(),
150        ));
151    }
152
153    let mut session = ReadOnlySession::open(&options.db_url).await?;
154    let catalog = collect::read(&mut session).await?;
155    let no_transaction_id_assigned = session.finish().await?;
156    let mut warnings = catalog.warnings.clone();
157
158    // The Management API, for what Postgres cannot see.
159    let mut auth = catalog.auth.clone();
160    let mut edge = EdgeFunctions {
161        status: SourceStatus::NotInspected,
162        functions: Vec::new(),
163    };
164    let mut management_status = SourceStatus::NotInspected;
165    if let Some(api) = &options.management {
166        management_status = SourceStatus::Inspected;
167        match api.edge_functions(&options.project_ref).await {
168            Ok(functions) => {
169                edge = EdgeFunctions {
170                    status: SourceStatus::Inspected,
171                    functions,
172                };
173            }
174            Err(ManagementError::Unauthorized(status)) => {
175                return Err(InspectError::ManagementUnauthorized(
176                    ManagementError::Unauthorized(status).to_string(),
177                ));
178            }
179            Err(error) => {
180                management_status = SourceStatus::Failed;
181                edge.status = SourceStatus::Failed;
182                warnings.push(format!("Edge Functions not listed: {error}"));
183            }
184        }
185        match api.auth_config(&options.project_ref).await {
186            Ok(config) => {
187                auth.enabled_providers = Some(config.providers);
188                auth.enabled_mfa = Some(config.mfa);
189            }
190            Err(ManagementError::Unauthorized(status)) => {
191                return Err(InspectError::ManagementUnauthorized(
192                    ManagementError::Unauthorized(status).to_string(),
193                ));
194            }
195            Err(error) => {
196                management_status = SourceStatus::Failed;
197                warnings.push(format!("auth configuration not read: {error}"));
198            }
199        }
200    }
201
202    // Policies: rules, then the classifier for the rest.
203    let mut classifier = options.classifier.as_deref();
204    let mut classifier_status = if classifier.is_some() {
205        SourceStatus::Inspected
206    } else {
207        SourceStatus::NotInspected
208    };
209    let mut policies = Vec::with_capacity(catalog.policies.len());
210    let mut managed_policies: Vec<ManagedPolicy> = Vec::new();
211    let mut storage_policies: Vec<collect::RawPolicy> = Vec::new();
212    for raw in &catalog.policies {
213        // Policies on Supabase-managed schemas are the platform's, not the
214        // venture's: listed for audit, never classified, never a finding.
215        // The two storage tables are handled by the storage phase instead.
216        if collect::MANAGED_SCHEMAS.contains(&raw.schema.as_str()) {
217            if storage_policy::is_bucket_table(&raw.schema, &raw.table) {
218                storage_policies.push(raw.clone());
219            } else {
220                managed_policies.push(ManagedPolicy {
221                    schema: raw.schema.clone(),
222                    table: raw.table.clone(),
223                    name: raw.name.clone(),
224                    command: raw.command.clone(),
225                    roles: raw.roles.clone(),
226                    using: raw.using.clone(),
227                    with_check: raw.with_check.clone(),
228                });
229            }
230            continue;
231        }
232        let table = format!("{}.{}", raw.schema, raw.table);
233        let input = policy::PolicyInput {
234            table: &table,
235            columns: &raw.columns,
236            command: &raw.command,
237            roles: &raw.roles,
238            permissive: raw.permissive,
239            using: raw.using.as_deref(),
240            with_check: raw.with_check.as_deref(),
241        };
242        let placement = match policy::place(&input, classifier, options.classify_threshold).await {
243            Ok(placement) => placement,
244            Err(error) => {
245                // One failure ends the classifier's part: the rest of the
246                // policies are placed by rule alone.
247                warnings.push(format!(
248                    "the policy classifier failed, so unplaced policies stay needs_review: \
249                     {error}"
250                ));
251                classifier = None;
252                classifier_status = SourceStatus::Failed;
253                policy::place(&input, None, options.classify_threshold)
254                    .await
255                    .unwrap_or_else(|_| unreachable!("rules alone never fail"))
256            }
257        };
258        let test_stub = policy::test_stub(&input, &raw.name, &placement.suggested_equivalent);
259        policies.push(Policy {
260            schema: raw.schema.clone(),
261            table: raw.table.clone(),
262            name: raw.name.clone(),
263            command: raw.command.clone(),
264            permissive: raw.permissive,
265            roles: raw.roles.clone(),
266            using: raw.using.clone(),
267            with_check: raw.with_check.clone(),
268            pattern: placement.pattern,
269            confidence: placement.confidence,
270            source: placement.source,
271            classifier_label: placement.classifier_label,
272            suggested_equivalent: placement.suggested_equivalent,
273            test_stub,
274            disposition: Disposition::Undecided,
275        });
276    }
277
278    // Storage policies on `storage.objects`/`storage.buckets` go with the
279    // bucket they name; ones on any other `storage` table stayed managed.
280    let mut storage = catalog.storage.clone();
281    storage_policy::attach(&mut storage, &storage_policies);
282
283    let role_can_write = catalog.role_can_write;
284    if catalog.role_is_superuser {
285        warnings.push(format!(
286            "inspected as the superuser `{}`: the session was read-only, but use the read-only \
287             role from docs/import/supabase.md for the real run",
288            catalog.role
289        ));
290    } else if role_can_write {
291        warnings.push(format!(
292            "the role `{}` holds write privileges: the session was read-only, but a role that \
293             cannot write is the second guard (docs/import/supabase.md)",
294            catalog.role
295        ));
296    }
297
298    let findings = classify::findings(&catalog, &policies, &storage, &auth, &edge);
299    let count = |class: Classification| {
300        findings
301            .iter()
302            .filter(|finding| finding.classification == class)
303            .count()
304    };
305    let data_bytes: u64 = catalog.tables.iter().map(|table| table.data_bytes).sum();
306    // `None` when the storage section is not visible: a sum over hidden
307    // buckets would be a wrong fact, not a small one.
308    let storage_objects: Option<u64> = storage
309        .buckets
310        .as_ref()
311        .and_then(|buckets| buckets.iter().map(|bucket| bucket.objects).sum());
312    let storage_bytes: Option<u64> = storage
313        .buckets
314        .as_ref()
315        .and_then(|buckets| buckets.iter().map(|bucket| bucket.bytes).sum());
316    let mbps = u64::from(options.transfer_mbps.max(1));
317    // The estimate is data-only when storage is not visible: adding a zero
318    // would silently understate it, so the Markdown says so instead (the
319    // JSON keeps its shape).
320    let bits = (data_bytes + storage_bytes.unwrap_or(0)).saturating_mul(8);
321    let summary = Summary {
322        automatic: count(Classification::Automatic),
323        needs_work: count(Classification::NeedsWork),
324        blockers: count(Classification::Blocker),
325        // Both counts are set by `dispositions::apply` below.
326        decided: 0,
327        undecided: 0,
328        ready: count(Classification::Blocker) == 0,
329        tables: catalog.tables.len(),
330        estimated_rows: catalog
331            .tables
332            .iter()
333            .filter_map(|table| table.estimated_rows)
334            .sum(),
335        data_bytes,
336        index_bytes: catalog.tables.iter().map(|table| table.index_bytes).sum(),
337        storage_objects,
338        storage_bytes,
339        transfer_assumed_mbps: options.transfer_mbps.max(1),
340        estimated_transfer_seconds: bits.div_ceil(mbps * 1_000_000),
341    };
342
343    let schemas = catalog
344        .schemas
345        .iter()
346        .map(|(name, tables)| {
347            let managed = collect::MANAGED_SCHEMAS.contains(&name.as_str());
348            Schema {
349                name: name.clone(),
350                kind: if managed {
351                    SchemaKind::SupabaseManaged
352                } else {
353                    SchemaKind::User
354                },
355                target: (!managed).then(|| classify::target_schema(name)),
356                tables: *tables,
357            }
358        })
359        .collect();
360
361    warnings.sort();
362    warnings.dedup();
363    let mut report = Report {
364        report_version: REPORT_VERSION,
365        tool: Tool {
366            name: env!("CARGO_PKG_NAME").to_owned(),
367            version: env!("CARGO_PKG_VERSION").to_owned(),
368        },
369        project: Project {
370            project_ref: options.project_ref.clone(),
371            host: source.host,
372            port: source.port,
373            database: source.database,
374            server_version: catalog.server_version.clone(),
375        },
376        read_only: ReadOnlyEvidence {
377            transaction_read_only: catalog.transaction_read_only,
378            session_read_only: catalog.session_read_only,
379            no_transaction_id_assigned,
380            role: catalog.role.clone(),
381            role_is_superuser: catalog.role_is_superuser,
382            role_can_write,
383        },
384        coverage: Coverage {
385            database: SourceStatus::Inspected,
386            management_api: management_status,
387            policy_classifier: classifier_status,
388            sections: catalog.visibility.sections.clone(),
389        },
390        summary,
391        // Set just below, with an empty file: without a dispositions file
392        // every needs-work and blocker item is undecided.
393        dispositions: DispositionsReport::default(),
394        schemas,
395        tables: catalog.tables.clone(),
396        views: catalog.views.clone(),
397        sequences: catalog.sequences.clone(),
398        enums: catalog.enums.clone(),
399        extensions: catalog
400            .extensions
401            .iter()
402            .map(|(name, version, schema)| Extension {
403                support: extensions::support(name).0,
404                name: name.clone(),
405                version: version.clone(),
406                schema: schema.clone(),
407            })
408            .collect(),
409        functions: catalog.functions.clone(),
410        triggers: catalog.triggers.clone(),
411        policies,
412        managed_policies,
413        api_role_grants: catalog.grants.clone(),
414        auth,
415        storage,
416        edge_functions: edge,
417        realtime: Realtime {
418            publications: catalog.publications.clone(),
419        },
420        cron_jobs: catalog.cron_jobs.clone(),
421        findings,
422        warnings,
423    };
424    dispositions::apply(&mut report, &DispositionsFile::default());
425    Ok(report)
426}
cratefield_import_supabase/lib.rs

cratefield_import_supabase/
lib.rs
