lenso_organization_postgres_plugin/
lib.rs

1//! PostgreSQL-backed Organization behavior for Lenso vNext.
2
3mod operator;
4mod schema;
5
6use std::{cell::RefCell, fmt, fmt::Write as _, rc::Rc, time::Duration};
7
8use lenso_capability_organization_admin::{
9    CreateOrganizationError, CreateOrganizationRequest, CreateOrganizationResponse,
10    OrganizationAdmin, OrganizationAdminEndpoint, OrganizationAdminProvider,
11};
12use lenso_capability_organization_membership::{
13    CheckMembershipError, CheckMembershipRequest, CheckMembershipResponse, OrganizationMembership,
14    OrganizationMembershipEndpoint, OrganizationMembershipProvider,
15};
16use lenso_capability_secrets::{ResolveRequest, SecretsClient, SecretsInvocationError};
17use lenso_kernel::{
18    DeactivateContext, InvocationContext, NativeRequestEndpoint, NativeRequestFuture, PluginFuture,
19    PluginLifecycle, PrepareContext, RuntimeFailure,
20};
21use lenso_native_adapter::{NativePluginFactory, NativePluginFactoryContext, NativePluginInstance};
22use lenso_postgres_kit::OwnedPostgres;
23use serde::{Deserialize, Serialize};
24use sqlx::Row;
25use thiserror::Error;
26use zeroize::Zeroizing;
27
28use crate::schema::schema_plan;
29
30pub use operator::{OrganizationOperator, OrganizationOperatorError};
31
32pub const PACKAGE_ID: &str = "lenso.organization.postgres";
33pub const PACKAGE_VERSION: &str = env!("CARGO_PKG_VERSION");
34const DEPENDENCY_TIMEOUT: Duration = Duration::from_secs(10);
35#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
36#[serde(deny_unknown_fields)]
37pub struct OrganizationConfig {
38    schema: String,
39    database_url_secret: String,
40    #[serde(default)]
41    admin_callers: Vec<String>,
42}
43
44impl OrganizationConfig {
45    pub fn new(
46        schema: impl Into<String>,
47        database_url_secret: impl Into<String>,
48        admin_callers: Vec<String>,
49    ) -> Result<Self, OrganizationConfigError> {
50        let value = Self {
51            schema: schema.into(),
52            database_url_secret: database_url_secret.into(),
53            admin_callers,
54        };
55        value.validate()?;
56        Ok(value)
57    }
58
59    fn validate(&self) -> Result<(), OrganizationConfigError> {
60        schema_plan(self.schema.clone()).map_err(|_| OrganizationConfigError::InvalidSchema)?;
61        if !valid_secret_reference(&self.database_url_secret) {
62            return Err(OrganizationConfigError::InvalidSecretReference);
63        }
64        if self.admin_callers.is_empty()
65            || self
66                .admin_callers
67                .iter()
68                .any(|value| !valid_name(value, 256))
69        {
70            return Err(OrganizationConfigError::InvalidAdminCallers);
71        }
72        Ok(())
73    }
74}
75
76#[derive(Clone, Debug, Error, Eq, PartialEq)]
77pub enum OrganizationConfigError {
78    #[error("invalid owned PostgreSQL schema")]
79    InvalidSchema,
80    #[error("invalid database URL secret reference")]
81    InvalidSecretReference,
82    #[error("at least one valid Organization Admin caller is required")]
83    InvalidAdminCallers,
84}
85
86#[derive(Clone, Copy, Debug, Default)]
87pub struct OrganizationFactory;
88
89impl NativePluginFactory for OrganizationFactory {
90    fn package_id(&self) -> &'static str {
91        PACKAGE_ID
92    }
93
94    fn package_version(&self) -> &'static str {
95        PACKAGE_VERSION
96    }
97
98    fn instantiate(
99        &self,
100        context: NativePluginFactoryContext<'_>,
101    ) -> Result<NativePluginInstance, RuntimeFailure> {
102        if context.entrypoint() != "default" {
103            return Err(RuntimeFailure::InvalidResolvedPlan {
104                detail: format!(
105                    "unsupported Organization entrypoint `{}`",
106                    context.entrypoint()
107                ),
108            });
109        }
110        let config: OrganizationConfig =
111            serde_json::from_str(context.configuration()).map_err(|error| {
112                RuntimeFailure::InvalidResolvedPlan {
113                    detail: format!("Organization configuration is invalid: {error}"),
114                }
115            })?;
116        config
117            .validate()
118            .map_err(|error| RuntimeFailure::InvalidResolvedPlan {
119                detail: error.to_string(),
120            })?;
121
122        let state = Rc::new(RefCell::new(None));
123        let provider = OrganizationProvider {
124            state: state.clone(),
125            admin_callers: config.admin_callers.clone(),
126        };
127        let endpoints: Vec<Rc<dyn NativeRequestEndpoint>> = vec![
128            Rc::new(OrganizationAdminEndpoint::new(provider.clone())),
129            Rc::new(OrganizationMembershipEndpoint::new(provider)),
130        ];
131        Ok(NativePluginInstance::with_lifecycle(
132            endpoints,
133            OrganizationLifecycle { config, state },
134        ))
135    }
136}
137
138#[derive(Clone)]
139struct PreparedOrganization {
140    postgres: OwnedPostgres,
141}
142
143impl fmt::Debug for PreparedOrganization {
144    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
145        formatter
146            .debug_struct("PreparedOrganization")
147            .field("schema", &self.postgres.schema())
148            .finish()
149    }
150}
151
152#[derive(Clone)]
153struct OrganizationProvider {
154    state: Rc<RefCell<Option<PreparedOrganization>>>,
155    admin_callers: Vec<String>,
156}
157
158impl fmt::Debug for OrganizationProvider {
159    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
160        formatter
161            .debug_struct("OrganizationProvider")
162            .field("prepared", &self.state.borrow().is_some())
163            .field("admin_caller_count", &self.admin_callers.len())
164            .finish()
165    }
166}
167
168impl OrganizationProvider {
169    fn prepared(&self) -> Result<PreparedOrganization, RuntimeFailure> {
170        self.state
171            .borrow()
172            .clone()
173            .ok_or(RuntimeFailure::PluginFailure {
174                detail: "Organization Plugin is not prepared".to_owned(),
175            })
176    }
177
178    fn authorized_admin_caller<'a>(&self, context: &'a InvocationContext) -> Option<&'a str> {
179        context
180            .caller_instance()
181            .filter(|caller| self.admin_callers.iter().any(|allowed| allowed == *caller))
182    }
183}
184
185impl OrganizationAdminProvider for OrganizationProvider {
186    fn create_organization(
187        &self,
188        context: InvocationContext,
189        request: CreateOrganizationRequest,
190    ) -> NativeRequestFuture<OrganizationAdmin> {
191        let caller_instance = self.authorized_admin_caller(&context).map(str::to_owned);
192        let prepared = self.prepared();
193        Box::pin(async move {
194            let Some(caller_instance) = caller_instance else {
195                return Ok(Err(CreateOrganizationError::Forbidden));
196            };
197            let name = request.name.trim().to_owned();
198            if !valid_name(&request.idempotency_key, 256)
199                || !valid_organization_name(&request.name)
200                || !valid_slug(&request.slug)
201                || !valid_name(&request.owner_subject, 256)
202            {
203                return Ok(Err(CreateOrganizationError::InvalidOrganization));
204            }
205            let prepared = prepared?;
206            create_organization_in_postgres(prepared, caller_instance, request, name).await
207        })
208    }
209}
210
211async fn create_organization_in_postgres(
212    prepared: PreparedOrganization,
213    caller_instance: String,
214    request: CreateOrganizationRequest,
215    name: String,
216) -> Result<Result<CreateOrganizationResponse, CreateOrganizationError>, RuntimeFailure> {
217    let organization_id = random_id("org_").map_err(runtime)?;
218    let owner_membership_id = random_id("member_").map_err(runtime)?;
219    let mut transaction = prepared.postgres.pool().begin().await.map_err(|source| {
220        runtime(OrganizationError::Database {
221            operation: "begin organization creation",
222            source,
223        })
224    })?;
225    if !reserve_creation(
226        &mut transaction,
227        &caller_instance,
228        &request,
229        &name,
230        &organization_id,
231        &owner_membership_id,
232    )
233    .await?
234    {
235        let response = match read_creation_replay(
236            &mut transaction,
237            &caller_instance,
238            &request,
239            &name,
240        )
241        .await?
242        {
243            Ok(response) => response,
244            Err(error) => return Ok(Err(error)),
245        };
246        transaction.commit().await.map_err(|source| {
247            runtime(OrganizationError::Database {
248                operation: "commit organization creation replay",
249                source,
250            })
251        })?;
252        return Ok(Ok(response));
253    }
254    if let Err(error) = insert_organization_and_owner(
255        &mut transaction,
256        &request,
257        &name,
258        &organization_id,
259        &owner_membership_id,
260    )
261    .await?
262    {
263        return Ok(Err(error));
264    }
265    transaction.commit().await.map_err(|source| {
266        runtime(OrganizationError::Database {
267            operation: "commit organization creation",
268            source,
269        })
270    })?;
271    Ok(Ok(CreateOrganizationResponse {
272        created: true,
273        organization_id,
274        owner_membership_id,
275    }))
276}
277
278async fn reserve_creation(
279    transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
280    caller_instance: &str,
281    request: &CreateOrganizationRequest,
282    name: &str,
283    organization_id: &str,
284    owner_membership_id: &str,
285) -> Result<bool, RuntimeFailure> {
286    sqlx::query(
287        "INSERT INTO organization_creation_requests (caller_instance,idempotency_key,name,slug,owner_subject,organization_id,owner_membership_id) VALUES ($1,$2,$3,$4,$5,$6,$7) ON CONFLICT (caller_instance,idempotency_key) DO NOTHING",
288    )
289    .bind(caller_instance)
290    .bind(&request.idempotency_key)
291    .bind(name)
292    .bind(&request.slug)
293    .bind(&request.owner_subject)
294    .bind(organization_id)
295    .bind(owner_membership_id)
296    .execute(&mut **transaction)
297    .await
298    .map(|result| result.rows_affected() == 1)
299    .map_err(|source| {
300        runtime(OrganizationError::Database {
301            operation: "reserve organization creation",
302            source,
303        })
304    })
305}
306
307async fn read_creation_replay(
308    transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
309    caller_instance: &str,
310    request: &CreateOrganizationRequest,
311    name: &str,
312) -> Result<Result<CreateOrganizationResponse, CreateOrganizationError>, RuntimeFailure> {
313    let (stored_name, stored_slug, stored_owner, organization_id, owner_membership_id): (
314        String,
315        String,
316        String,
317        String,
318        String,
319    ) = sqlx::query_as(
320        "SELECT name,slug,owner_subject,organization_id,owner_membership_id FROM organization_creation_requests WHERE caller_instance=$1 AND idempotency_key=$2",
321    )
322    .bind(caller_instance)
323    .bind(&request.idempotency_key)
324    .fetch_one(&mut **transaction)
325    .await
326    .map_err(|source| {
327        runtime(OrganizationError::Database {
328            operation: "read organization creation replay",
329            source,
330        })
331    })?;
332    if stored_name != name || stored_slug != request.slug || stored_owner != request.owner_subject {
333        return Ok(Err(CreateOrganizationError::IdempotencyConflict));
334    }
335    Ok(Ok(CreateOrganizationResponse {
336        created: false,
337        organization_id,
338        owner_membership_id,
339    }))
340}
341
342async fn insert_organization_and_owner(
343    transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
344    request: &CreateOrganizationRequest,
345    name: &str,
346    organization_id: &str,
347    owner_membership_id: &str,
348) -> Result<Result<(), CreateOrganizationError>, RuntimeFailure> {
349    let inserted =
350        sqlx::query("INSERT INTO organizations (organization_id,name,slug) VALUES ($1,$2,$3)")
351            .bind(organization_id)
352            .bind(name)
353            .bind(&request.slug)
354            .execute(&mut **transaction)
355            .await;
356    if let Err(error) = inserted {
357        if error
358            .as_database_error()
359            .and_then(|database| database.constraint())
360            == Some("organizations_active_slug_key")
361        {
362            return Ok(Err(CreateOrganizationError::SlugConflict));
363        }
364        return Err(runtime(OrganizationError::Database {
365            operation: "insert organization",
366            source: error,
367        }));
368    }
369    sqlx::query("INSERT INTO organization_memberships (membership_id,organization_id,subject,is_owner) VALUES ($1,$2,$3,true)")
370        .bind(owner_membership_id)
371        .bind(organization_id)
372        .bind(&request.owner_subject)
373        .execute(&mut **transaction)
374        .await
375        .map_err(|source| runtime(OrganizationError::Database { operation: "insert owner membership", source }))?;
376    Ok(Ok(()))
377}
378
379impl OrganizationMembershipProvider for OrganizationProvider {
380    fn check_membership(
381        &self,
382        _context: InvocationContext,
383        request: CheckMembershipRequest,
384    ) -> NativeRequestFuture<OrganizationMembership> {
385        let prepared = self.prepared();
386        Box::pin(async move {
387            if !valid_name(&request.organization_id, 256) || !valid_name(&request.subject, 256) {
388                return Ok(Err(CheckMembershipError::InvalidRequest));
389            }
390            let prepared = prepared?;
391            let row = sqlx::query(
392                "SELECT EXISTS(SELECT 1 FROM organizations WHERE organization_id=$1 AND archived_at IS NULL) AS organization_exists, COALESCE((SELECT removed_at IS NULL FROM organization_memberships WHERE organization_id=$1 AND subject=$2 ORDER BY created_at DESC LIMIT 1),false) AS active, COALESCE((SELECT is_owner AND removed_at IS NULL FROM organization_memberships WHERE organization_id=$1 AND subject=$2 ORDER BY created_at DESC LIMIT 1),false) AS owner",
393            )
394            .bind(&request.organization_id)
395            .bind(&request.subject)
396            .fetch_one(prepared.postgres.pool())
397            .await
398            .map_err(|source| runtime(OrganizationError::Database { operation: "check organization membership", source }))?;
399            let organization_exists: bool =
400                row.try_get("organization_exists").map_err(|source| {
401                    runtime(OrganizationError::Database {
402                        operation: "decode organization existence",
403                        source,
404                    })
405                })?;
406            if !organization_exists {
407                return Ok(Err(CheckMembershipError::OrganizationNotFound));
408            }
409            let active = row.try_get("active").map_err(|source| {
410                runtime(OrganizationError::Database {
411                    operation: "decode organization membership",
412                    source,
413                })
414            })?;
415            let owner = row.try_get("owner").map_err(|source| {
416                runtime(OrganizationError::Database {
417                    operation: "decode organization ownership",
418                    source,
419                })
420            })?;
421            Ok(Ok(CheckMembershipResponse { active, owner }))
422        })
423    }
424}
425
426#[derive(Debug)]
427struct OrganizationLifecycle {
428    config: OrganizationConfig,
429    state: Rc<RefCell<Option<PreparedOrganization>>>,
430}
431
432impl PluginLifecycle for OrganizationLifecycle {
433    fn prepare(&self, context: PrepareContext) -> PluginFuture {
434        let config = self.config.clone();
435        let state = self.state.clone();
436        let dependencies = context.dependencies().clone();
437        let cancellation = context.cancellation();
438        Box::pin(async move {
439            let secrets = SecretsClient::from_dependencies(&dependencies)?;
440            let invocation =
441                dependencies.invocation_context_after(DEPENDENCY_TIMEOUT, cancellation)?;
442            let database_url = secrets
443                .resolve_with_context(
444                    invocation,
445                    ResolveRequest {
446                        reference: config.database_url_secret.clone(),
447                    },
448                )
449                .await
450                .map_err(|error| match error {
451                    SecretsInvocationError::Domain(_) => RuntimeFailure::PluginFailure {
452                        detail: format!(
453                            "database URL secret `{}` was rejected",
454                            config.database_url_secret
455                        ),
456                    },
457                    SecretsInvocationError::Runtime(error) => error,
458                })?;
459            let database_url = Zeroizing::new(database_url.value);
460            let postgres = OwnedPostgres::prepare(
461                &database_url,
462                schema_plan(config.schema).map_err(|error| {
463                    RuntimeFailure::InvalidResolvedPlan {
464                        detail: error.to_string(),
465                    }
466                })?,
467            )
468            .await
469            .map_err(|error| RuntimeFailure::PluginFailure {
470                detail: error.to_string(),
471            })?;
472            state.replace(Some(PreparedOrganization { postgres }));
473            Ok(())
474        })
475    }
476
477    fn deactivate(&self, _context: DeactivateContext) -> PluginFuture {
478        let prepared = self.state.borrow_mut().take();
479        Box::pin(async move {
480            if let Some(prepared) = prepared {
481                prepared.postgres.pool().close().await;
482            }
483            Ok(())
484        })
485    }
486}
487
488#[derive(Debug, Error)]
489enum OrganizationError {
490    #[error("PostgreSQL operation `{operation}` failed")]
491    Database {
492        operation: &'static str,
493        #[source]
494        source: sqlx::Error,
495    },
496    #[error("random source unavailable")]
497    Random,
498}
499
500fn runtime(error: impl fmt::Display) -> RuntimeFailure {
501    RuntimeFailure::PluginFailure {
502        detail: error.to_string(),
503    }
504}
505
506fn random_id(prefix: &str) -> Result<String, OrganizationError> {
507    let mut bytes = [0_u8; 18];
508    getrandom::fill(&mut bytes).map_err(|_| OrganizationError::Random)?;
509    let mut id = String::with_capacity(prefix.len() + bytes.len() * 2);
510    id.push_str(prefix);
511    for byte in bytes {
512        write!(&mut id, "{byte:02x}").expect("writing to String cannot fail");
513    }
514    Ok(id)
515}
516
517fn valid_organization_name(value: &str) -> bool {
518    let value = value.trim();
519    !value.is_empty() && value.len() <= 200 && !value.chars().any(char::is_control)
520}
521
522fn valid_slug(value: &str) -> bool {
523    !value.is_empty()
524        && value.len() <= 100
525        && value
526            .bytes()
527            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-')
528        && !value.starts_with('-')
529        && !value.ends_with('-')
530}
531
532fn valid_name(value: &str, max: usize) -> bool {
533    !value.is_empty()
534        && value.len() <= max
535        && value
536            .bytes()
537            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b':'))
538}
539
540fn valid_secret_reference(reference: &str) -> bool {
541    !reference.is_empty()
542        && reference.len() <= 256
543        && !reference.starts_with('/')
544        && !reference.ends_with('/')
545        && !reference.contains("//")
546        && reference
547            .split('/')
548            .all(|segment| segment != "." && segment != "..")
549        && reference
550            .bytes()
551            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b'/'))
552}
553
554#[cfg(test)]
555mod tests {
556    use super::*;
557    use lenso_kernel::CancellationToken;
558    use lenso_postgres_kit::{Migration, SchemaOperator, SchemaPlan};
559    use sqlx::{AssertSqlSafe, Executor};
560
561    const LEGACY_MIGRATIONS: &[Migration] = &[Migration::new(
562        1,
563        "create-organizations",
564        include_str!("../migrations/001_create_organizations.sql"),
565    )];
566
567    fn legacy_schema_plan(schema: impl Into<std::sync::Arc<str>>) -> SchemaPlan {
568        SchemaPlan::new(schema, LEGACY_MIGRATIONS).unwrap()
569    }
570
571    #[test]
572    fn configuration_rejects_ambient_admin_authority() {
573        let error = OrganizationConfig::new("organization", "organization/database", Vec::new())
574            .unwrap_err();
575        assert_eq!(error, OrganizationConfigError::InvalidAdminCallers);
576    }
577
578    #[test]
579    fn slugs_are_stable_and_narrow() {
580        assert!(valid_slug("acme-platform"));
581        assert!(!valid_slug("Acme Platform"));
582        assert!(!valid_slug("-acme"));
583    }
584
585    #[test]
586    fn generated_ids_do_not_repeat() {
587        assert_ne!(random_id("org_").unwrap(), random_id("org_").unwrap());
588    }
589
590    #[tokio::test]
591    async fn forbidden_admin_is_a_domain_error_before_storage_access() {
592        let provider = OrganizationProvider {
593            state: Rc::new(RefCell::new(None)),
594            admin_callers: vec!["business-admin".to_owned()],
595        };
596        let context = InvocationContext::new(1, None, CancellationToken::new())
597            .with_caller_instance("untrusted");
598        let result = provider
599            .create_organization(
600                context,
601                CreateOrganizationRequest {
602                    idempotency_key: "forbidden-create".to_owned(),
603                    name: "Acme".to_owned(),
604                    owner_subject: "usr_owner".to_owned(),
605                    slug: "acme".to_owned(),
606                },
607            )
608            .await
609            .unwrap();
610        assert_eq!(result, Err(CreateOrganizationError::Forbidden));
611    }
612
613    #[tokio::test]
614    async fn invalid_idempotency_key_is_rejected_before_storage_access() {
615        let provider = OrganizationProvider {
616            state: Rc::new(RefCell::new(None)),
617            admin_callers: vec!["business-admin".to_owned()],
618        };
619        let context = InvocationContext::new(1, None, CancellationToken::new())
620            .with_caller_instance("business-admin");
621        let result = provider
622            .create_organization(
623                context,
624                CreateOrganizationRequest {
625                    idempotency_key: String::new(),
626                    name: "Acme".to_owned(),
627                    owner_subject: "usr_owner".to_owned(),
628                    slug: "acme".to_owned(),
629                },
630            )
631            .await
632            .unwrap();
633        assert_eq!(result, Err(CreateOrganizationError::InvalidOrganization));
634    }
635
636    #[tokio::test]
637    async fn unprepared_membership_reports_runtime_failure() {
638        let provider = OrganizationProvider {
639            state: Rc::new(RefCell::new(None)),
640            admin_callers: vec!["business-admin".to_owned()],
641        };
642        let result = provider
643            .check_membership(
644                InvocationContext::new(1, None, CancellationToken::new()),
645                CheckMembershipRequest {
646                    organization_id: "org_missing".to_owned(),
647                    subject: "usr_owner".to_owned(),
648                },
649            )
650            .await;
651        assert!(matches!(result, Err(RuntimeFailure::PluginFailure { .. })));
652    }
653
654    #[tokio::test]
655    #[ignore = "requires LENSO_POSTGRES_TEST_URL"]
656    #[allow(
657        clippy::too_many_lines,
658        reason = "the acceptance scenario keeps concurrent creation and every replay boundary together"
659    )]
660    async fn concurrent_create_is_caller_scoped_idempotent_and_preserves_ownership() {
661        let database_url =
662            std::env::var("LENSO_POSTGRES_TEST_URL").expect("LENSO_POSTGRES_TEST_URL is required");
663        let schema = random_id("organization_test_").unwrap();
664        OrganizationOperator::setup(&database_url, &schema)
665            .await
666            .unwrap();
667        let postgres = OwnedPostgres::prepare(&database_url, schema_plan(schema.clone()).unwrap())
668            .await
669            .unwrap();
670        let provider = OrganizationProvider {
671            state: Rc::new(RefCell::new(Some(PreparedOrganization { postgres }))),
672            admin_callers: vec!["business-admin".to_owned(), "second-admin".to_owned()],
673        };
674        let admin_context = InvocationContext::new(1, None, CancellationToken::new())
675            .with_caller_instance("business-admin");
676        let create_request = CreateOrganizationRequest {
677            idempotency_key: "create-acme".to_owned(),
678            name: "Acme".to_owned(),
679            owner_subject: "usr_owner".to_owned(),
680            slug: "acme".to_owned(),
681        };
682        let first_creation =
683            provider.create_organization(admin_context.clone(), create_request.clone());
684        let concurrent_replay = provider.create_organization(
685            InvocationContext::new(2, None, CancellationToken::new())
686                .with_caller_instance("business-admin"),
687            create_request.clone(),
688        );
689        let (first_creation, concurrent_replay) = tokio::join!(first_creation, concurrent_replay);
690        let first_creation = first_creation.unwrap().unwrap();
691        let concurrent_replay = concurrent_replay.unwrap().unwrap();
692        assert_ne!(first_creation.created, concurrent_replay.created);
693        assert_eq!(
694            first_creation.organization_id,
695            concurrent_replay.organization_id
696        );
697        assert_eq!(
698            first_creation.owner_membership_id,
699            concurrent_replay.owner_membership_id
700        );
701        let created = if first_creation.created {
702            first_creation
703        } else {
704            concurrent_replay
705        };
706        let membership = provider
707            .check_membership(
708                InvocationContext::new(3, None, CancellationToken::new()),
709                CheckMembershipRequest {
710                    organization_id: created.organization_id.clone(),
711                    subject: "usr_owner".to_owned(),
712                },
713            )
714            .await
715            .unwrap()
716            .unwrap();
717        assert!(membership.active);
718        assert!(membership.owner);
719
720        let replay = provider
721            .create_organization(admin_context.clone(), create_request.clone())
722            .await
723            .unwrap()
724            .unwrap();
725        assert!(!replay.created);
726        assert_eq!(replay.organization_id, created.organization_id);
727        assert_eq!(replay.owner_membership_id, created.owner_membership_id);
728
729        let conflict = provider
730            .create_organization(
731                admin_context.clone(),
732                CreateOrganizationRequest {
733                    owner_subject: "usr_other".to_owned(),
734                    ..create_request
735                },
736            )
737            .await
738            .unwrap();
739        assert_eq!(conflict, Err(CreateOrganizationError::IdempotencyConflict));
740
741        let second_caller = provider
742            .create_organization(
743                InvocationContext::new(4, None, CancellationToken::new())
744                    .with_caller_instance("second-admin"),
745                CreateOrganizationRequest {
746                    idempotency_key: "create-acme".to_owned(),
747                    name: "Second".to_owned(),
748                    owner_subject: "usr_second".to_owned(),
749                    slug: "second".to_owned(),
750                },
751            )
752            .await
753            .unwrap()
754            .unwrap();
755        assert!(second_caller.created);
756        assert_ne!(second_caller.organization_id, created.organization_id);
757
758        let duplicate = provider
759            .create_organization(
760                admin_context,
761                CreateOrganizationRequest {
762                    idempotency_key: "create-another-acme".to_owned(),
763                    name: "Another Acme".to_owned(),
764                    owner_subject: "usr_other".to_owned(),
765                    slug: "acme".to_owned(),
766                },
767            )
768            .await
769            .unwrap();
770        assert_eq!(duplicate, Err(CreateOrganizationError::SlugConflict));
771
772        let cleanup_pool = sqlx::PgPool::connect(&database_url).await.unwrap();
773        cleanup_pool
774            .execute(AssertSqlSafe(format!("DROP SCHEMA \"{schema}\" CASCADE")))
775            .await
776            .unwrap();
777        cleanup_pool.close().await;
778    }
779
780    #[tokio::test]
781    #[ignore = "requires LENSO_POSTGRES_TEST_URL"]
782    async fn upgrade_projects_legacy_owner_and_rejects_an_ownerless_active_organization() {
783        let database_url =
784            std::env::var("LENSO_POSTGRES_TEST_URL").expect("LENSO_POSTGRES_TEST_URL is required");
785        let schema = random_id("organization_upgrade_test_").unwrap();
786        SchemaOperator::connect(&database_url, legacy_schema_plan(schema.clone()))
787            .await
788            .unwrap()
789            .setup()
790            .await
791            .unwrap();
792        let legacy = OwnedPostgres::prepare(&database_url, legacy_schema_plan(schema.clone()))
793            .await
794            .unwrap();
795
796        sqlx::query("INSERT INTO organizations (organization_id,name,slug) VALUES ('org_good','Good','good'),('org_bad','Bad','bad')")
797            .execute(legacy.pool())
798            .await
799            .unwrap();
800        sqlx::query("INSERT INTO organization_roles (role_id,organization_id,name,permissions,system_key) VALUES ('role_good','org_good','Owner',ARRAY['organization.read'],'owner'),('role_bad','org_bad','Member',ARRAY['organization.read'],NULL)")
801            .execute(legacy.pool())
802            .await
803            .unwrap();
804        sqlx::query("INSERT INTO organization_memberships (membership_id,organization_id,subject,role_id) VALUES ('membership_good','org_good','usr_good','role_good'),('membership_bad','org_bad','usr_bad','role_bad')")
805            .execute(legacy.pool())
806            .await
807            .unwrap();
808
809        assert!(
810            OrganizationOperator::upgrade(&database_url, &schema)
811                .await
812                .is_err()
813        );
814        let legacy_roles_remain: bool =
815            sqlx::query_scalar("SELECT to_regclass('organization_roles') IS NOT NULL")
816                .fetch_one(legacy.pool())
817                .await
818                .unwrap();
819        assert!(legacy_roles_remain);
820
821        sqlx::query("UPDATE organization_roles SET system_key='owner' WHERE role_id='role_bad'")
822            .execute(legacy.pool())
823            .await
824            .unwrap();
825        legacy.pool().close().await;
826        OrganizationOperator::upgrade(&database_url, &schema)
827            .await
828            .unwrap();
829        let upgraded = OwnedPostgres::prepare(&database_url, schema_plan(schema.clone()).unwrap())
830            .await
831            .unwrap();
832        let owner_count: i64 = sqlx::query_scalar(
833            "SELECT count(*) FROM organization_memberships WHERE removed_at IS NULL AND is_owner",
834        )
835        .fetch_one(upgraded.pool())
836        .await
837        .unwrap();
838        assert_eq!(owner_count, 2);
839        let legacy_roles_remain: bool =
840            sqlx::query_scalar("SELECT to_regclass('organization_roles') IS NOT NULL")
841                .fetch_one(upgraded.pool())
842                .await
843                .unwrap();
844        assert!(!legacy_roles_remain);
845        let creation_receipts_exist: bool =
846            sqlx::query_scalar("SELECT to_regclass('organization_creation_requests') IS NOT NULL")
847                .fetch_one(upgraded.pool())
848                .await
849                .unwrap();
850        assert!(creation_receipts_exist);
851
852        upgraded.pool().close().await;
853        let cleanup_pool = sqlx::PgPool::connect(&database_url).await.unwrap();
854        cleanup_pool
855            .execute(AssertSqlSafe(format!("DROP SCHEMA \"{schema}\" CASCADE")))
856            .await
857            .unwrap();
858        cleanup_pool.close().await;
859    }
860}
lenso_organization_postgres_plugin/lib.rs

lenso_organization_postgres_plugin/
lib.rs
