pub(super) async fn create_active_app_cells(
        &self,
        conductor_handle: ConductorHandle,
    ) -> ConductorResult<Vec<Result<Vec<(Cell, InitialQueueTriggers)>, CreateAppError>>> {
        // Only create the active apps
        let active_apps = self.get_state().await?.active_apps;

        // Data required to create apps
        let root_env_dir = self.root_env_dir.clone();
        let keystore = self.keystore.clone();
        let task_manager = self
            .task_manager
            .as_ref()
            .expect("Task manager not initialized");

        // Closure for creating all cells in an app
        let tasks = active_apps.into_iter().map(
            move |(installed_app_id, app): (InstalledAppId, ActiveApp)| {
                // Clone data for async block
                let root_env_dir = std::path::PathBuf::from(root_env_dir.clone());
                let conductor_handle = conductor_handle.clone();
                let keystore = keystore.clone();

                // Task that creates the cells
                async move {
                    // Only create cells not already created
                    let cells_to_create = app
                        .all_cells()
                        .filter(|cell_id| !self.cells.contains_key(cell_id))
                        .map(|cell_id| {
                            (
                                cell_id,
                                root_env_dir.clone(),
                                keystore.clone(),
                                conductor_handle.clone(),
                            )
                        });

                    use aingle_p2p::actor::AIngleP2pRefToCell;

                    // Create each cell
                    let cells_tasks = cells_to_create.map(
                        |(cell_id, dir, keystore, conductor_handle)| async move {
                            let aingle_p2p_cell = self.aingle_p2p.to_cell(
                                cell_id.saf_hash().clone(),
                                cell_id.agent_pubkey().clone(),
                            );

                            let env = EnvWrite::open(
                                &dir,
                                DbKind::Cell(cell_id.clone()),
                                keystore.clone(),
                            )?;
                            let cache = self
                                .get_or_create_cache(cell_id.saf_hash())
                                .map_err(|e| CellError::FailedToCreateCache(e.into()))?;
                            Cell::create(
                                cell_id.clone(),
                                conductor_handle.clone(),
                                env,
                                cache,
                                aingle_p2p_cell,
                                task_manager.task_add_sender().clone(),
                                task_manager.task_stop_broadcaster().clone(),
                            )
                            .await
                        },
                    );

                    // Join all the cell create tasks for this app
                    // and separate any errors
                    let (success, errors): (Vec<_>, Vec<_>) =
                        futures::future::join_all(cells_tasks)
                            .await
                            .into_iter()
                            .partition(Result::is_ok);
                    // unwrap safe because of the partition
                    let success = success.into_iter().map(Result::unwrap);

                    // If there were errors, cleanup and return the errors
                    if !errors.is_empty() {
                        for cell in success {
                            // Error needs to capture which app failed
                            cell.0.destroy().await.map_err(|e| CreateAppError::Failed {
                                installed_app_id: installed_app_id.clone(),
                                errors: vec![e],
                            })?;
                        }
                        // match needed to avoid Debug requirement on unwrap_err
                        let errors = errors
                            .into_iter()
                            .map(|e| match e {
                                Err(e) => e,
                                Ok(_) => unreachable!("Safe because of the partition"),
                            })
                            .collect();
                        Err(CreateAppError::Failed {
                            installed_app_id,
                            errors,
                        })
                    } else {
                        // No errors so return the cells
                        Ok(success.collect())
                    }
                }
            },
        );

        // Join on all apps and return a list of
        // apps that had succelly created cells
        // and any apps that encounted errors
        Ok(futures::future::join_all(tasks).await)
    }
/// Create Cells for each CellId marked active in the ConductorState db
Create Cells for each CellId marked active in the ConductorState db
Create Cells for each CellId marked active in the ConductorState db
pub(super) async fn create_active_app_cells(
        &self,
        conductor_handle: ConductorHandle,
    ) -> ConductorResult<Vec<Result<Vec<(Cell, InitialQueueTriggers)>, CreateAppError>>> {
        // Only create the active apps
        let active_apps = self.get_state().await?.active_apps;
        // Data required to create apps
        let root_env_dir = self.root_env_dir.clone();
        let keystore = self.keystore.clone();
        let task_manager = self
            .task_manager
            .as_ref()
            .expect("Task manager not initialized");
        // Closure for creating all cells in an app
        let tasks = active_apps.into_iter().map(
            move |(installed_app_id, app): (InstalledAppId, ActiveApp)| {
                // Clone data for async block
                let root_env_dir = std::path::PathBuf::from(root_env_dir.clone());
                let conductor_handle = conductor_handle.clone();
                let keystore = keystore.clone();
                // Task that creates the cells
                async move {
                    // Only create cells not already created
                    let cells_to_create = app
                        .all_cells()
                        .filter(|cell_id| !self.cells.contains_key(cell_id))
                        .map(|cell_id| {
                            (
                                cell_id,
                                root_env_dir.clone(),
                                keystore.clone(),
                                conductor_handle.clone(),
                            )
                        });
                    use aingle_p2p::actor::AIngleP2pRefToCell;
                    // Create each cell
                    let cells_tasks = cells_to_create.map(
                        |(cell_id, dir, keystore, conductor_handle)| async move {
                            let aingle_p2p_cell = self.aingle_p2p.to_cell(
                                cell_id.saf_hash().clone(),
                                cell_id.agent_pubkey().clone(),
                            );
                            let env = EnvWrite::open(
                                &dir,
                                DbKind::Cell(cell_id.clone()),
                                keystore.clone(),
                            )?;
                            let cache = self
                                .get_or_create_cache(cell_id.saf_hash())
                                .map_err(|e| CellError::FailedToCreateCache(e.into()))?;
                            Cell::create(
                                cell_id.clone(),
                                conductor_handle.clone(),
                                env,
                                cache,
                                aingle_p2p_cell,
                                task_manager.task_add_sender().clone(),
                                task_manager.task_stop_broadcaster().clone(),
                            )
                            .await
                        },
                    );
                    // Join all the cell create tasks for this app
                    // and separate any errors
                    let (success, errors): (Vec<_>, Vec<_>) =
                        futures::future::join_all(cells_tasks)
                            .await
                            .into_iter()
                            .partition(Result::is_ok);
                    // unwrap safe because of the partition
                    let success = success.into_iter().map(Result::unwrap);
                    // If there were errors, cleanup and return the errors
                    if !errors.is_empty() {
                        for cell in success {
                            // Error needs to capture which app failed
                            cell.0.destroy().await.map_err(|e| CreateAppError::Failed {
                                installed_app_id: installed_app_id.clone(),
                                errors: vec![e],
                            })?;
                        }
                        // match needed to avoid Debug requirement on unwrap_err
                        let errors = errors
                            .into_iter()
                            .map(|e| match e {
                                Err(e) => e,
                                Ok(_) => unreachable!("Safe because of the partition"),
                            })
                            .collect();
                        Err(CreateAppError::Failed {
                            installed_app_id,
                            errors,
                        })
                    } else {
                        // No errors so return the cells
                        Ok(success.collect())
                    }
                }
            },
        );
        // Join on all apps and return a list of
        // apps that had succelly created cells
        // and any apps that encounted errors
        Ok(futures::future::join_all(tasks).await)
    }
pub ( super ) async fn create_active_app_cells ( & self , conductor_handle : ConductorHandle , ) -> ConductorResult < Vec < Result < Vec < ( Cell , InitialQueueTriggers ) > , CreateAppError > > > { // Only create the active apps let active_apps = self . get_state ( ) . await ? . active_apps ; // Data required to create apps let root_env_dir = self . root_env_dir . clone ( ) ; let keystore = self . keystore . clone ( ) ; let task_manager = self . task_manager . as_ref ( ) . expect ( " " ) ; // Closure for creating all cells in an app let tasks = active_apps . into_iter ( ) . map ( move | ( installed_app_id , app ) : ( InstalledAppId , ActiveApp ) | { // Clone data for async block let root_env_dir = std :: path :: PathBuf :: from ( root_env_dir . clone ( ) ) ; let conductor_handle = conductor_handle . clone ( ) ; let keystore = keystore . clone ( ) ; // Task that creates the cells async move { // Only create cells not already created let cells_to_create = app . all_cells ( ) . filter ( | cell_id | ! self . cells . contains_key ( cell_id ) ) . map ( | cell_id | { ( cell_id , root_env_dir . clone ( ) , keystore . clone ( ) , conductor_handle . clone ( ) , ) } ) ; use aingle_p2p :: actor :: AIngleP2pRefToCell ; // Create each cell let cells_tasks = cells_to_create . map ( | ( cell_id , dir , keystore , conductor_handle ) | async move { let aingle_p2p_cell = self . aingle_p2p . to_cell ( cell_id . saf_hash ( ) . clone ( ) , cell_id . agent_pubkey ( ) . clone ( ) , ) ; let env = EnvWrite :: open ( & dir , DbKind :: Cell ( cell_id . clone ( ) ) , keystore . clone ( ) , ) ? ; let cache = self . get_or_create_cache ( cell_id . saf_hash ( ) ) . map_err ( | e | CellError :: FailedToCreateCache ( e . into ( ) ) ) ? ; Cell :: create ( cell_id . clone ( ) , conductor_handle . clone ( ) , env , cache , aingle_p2p_cell , task_manager . task_add_sender ( ) . clone ( ) , task_manager . task_stop_broadcaster ( ) . clone ( ) , ) . await } , ) ; // Join all the cell create tasks for this app // and separate any errors let ( success , errors ) : ( Vec < _ > , Vec < _ > ) = futures :: future :: join_all ( cells_tasks ) . await . into_iter ( ) . partition ( Result :: is_ok ) ; // unwrap safe because of the partition let success = success . into_iter ( ) . map ( Result :: unwrap ) ; // If there were errors, cleanup and return the errors if ! errors . is_empty ( ) { for cell in success { // Error needs to capture which app failed cell . 0 . destroy ( ) . await . map_err ( | e | CreateAppError :: Failed { installed_app_id : installed_app_id . clone ( ) , errors : vec ! [ e ] , } ) ? ; } // match needed to avoid Debug requirement on unwrap_err let errors = errors . into_iter ( ) . map ( | e | match e { Err ( e ) => e , Ok ( _ ) => unreachable ! ( " " ) , } ) . collect ( ) ; Err ( CreateAppError :: Failed { installed_app_id , errors , } ) } else { // No errors so return the cells Ok ( success . collect ( ) ) } } } , ) ; // Join on all apps and return a list of // apps that had succelly created cells // and any apps that encounted errors Ok ( futures :: future :: join_all ( tasks ) . await ) }
Create Cells for each CellId marked active in the ConductorState db
Create Cells for each CellId marked active in the ConductorState db
&self
conductor_handle
ConductorHandle
&self
conductor_handle
ConductorHandle