use std::collections::HashSet;use std::ops::Index;use std::sync::Arc; use anyhow::anyhow;use chrono::{DateTime, Duration, Utc};use cloud_object_client::MockObjectClient;use cloud_objects::cloud_object::ServerPermissions;use firebase::FirebaseError;use itertools::Itertools;use warpui::r#async::Timer;use warpui::{App, Entity, ModelHandle, SingletonEntity}; use super::QueueDependency;use crate::ai::facts::{AIFact, AIMemory, CloudAIFactModel};use crate::cloud_object::model::actions::{    ObjectAction, ObjectActionHistory, ObjectActionSubtype, ObjectActionType,};use crate::cloud_object::model::generic_string_model::GenericStringObjectId;use crate::cloud_object::{    CloudModelType, CloudObjectEventEntrypoint, CreateCloudObjectResult, CreatedCloudObject,    GenericStringObjectFormat, JsonObjectType, ObjectIdType, ObjectType, Owner, Revision,    RevisionAndLastEditor, ServerCreationInfo, UpdateCloudObjectResult,};use crate::drive::CloudObjectTypeAndId;use crate::notebooks::{CloudNotebookModel, NotebookId};use crate::server::cloud_objects::update_manager::InitiatedBy;use crate::server::ids::{ClientId, HashableId, ServerId, ServerIdAndType, SyncId};use crate::server::server_api::ServerApiProvider;use crate::server::server_api::auth::UserAuthenticationError;use crate::server::sync_queue::{    CreationFailureReason, GenericStringObjectToCreate, QueueItemId, SerializedModel,    SyncQueueEvent,};use crate::system::SystemStats;use crate::workflows::CloudWorkflowModel;use crate::workflows::workflow::{Argument, ArgumentType, Workflow};use crate::{NetworkStatus, QueueItem, SyncQueue}; #[derive(Default)]struct Events(Vec<SyncQueueEvent>); impl Entity for Events {    type Event = ();} impl Index<usize> for Events {    type Output = SyncQueueEvent;     fn index(&self, index: usize) -> &Self::Output {        &self.0[index]    }} fn initialize_app(app: &mut App) {    app.add_singleton_model(|_| ServerApiProvider::new_for_test());    app.add_singleton_model(|_| NetworkStatus::new());    app.add_singleton_model(|_| SystemStats::new());} fn queue_item_dependencies<const N: usize>(    queue_item_ids: [QueueItemId; N],) -> HashSet<QueueDependency> {    queue_item_ids        .into_iter()        .map(QueueDependency::QueueItem)        .collect()} fn bulk_create_dependency(queue_item_id: QueueItemId, client_id: ClientId) -> QueueDependency {    QueueDependency::BulkCreateGenericStringObject {        queue_item_id,        client_id,    }} fn create_sync_queue(    app: &mut App,    queue_items: Vec<QueueItem>,    cloud_objects_client_mock: MockObjectClient,    immediately_start_dequeueing: bool,) -> ModelHandle<SyncQueue> {    let sync_queue = app.add_singleton_model(|ctx| {        SyncQueue::new(queue_items, Arc::new(cloud_objects_client_mock), ctx)    });     if immediately_start_dequeueing {        SyncQueue::handle(app).update(app, |sync_queue, ctx| {            sync_queue.start_dequeueing(ctx);        });    }     sync_queue} #[test]fn test_create_notebook() {    App::test((), |mut app| async move {        let notebook_id = ClientId::default();        let notebook_server_id = ServerId::from(1);        let notebook_ts = DateTime::<Utc>::default();        let notebook = CloudNotebookModel {            title: "Shared Notebook".to_string(),            data: "Hello".to_string(),            ai_document_id: None,            conversation_id: None,        };         let mut cloud_objects_client_mock = MockObjectClient::new();        cloud_objects_client_mock            .expect_create_notebook()            .times(1)            .return_once(move |_| {                Ok(CreateCloudObjectResult::Success {                    created_cloud_object: CreatedCloudObject {                        client_id: notebook_id,                        revision_and_editor: RevisionAndLastEditor {                            revision: notebook_ts.into(),                            last_editor_uid: None,                        },                        metadata_ts: notebook_ts.into(),                        server_id_and_type: ServerIdAndType {                            id: notebook_server_id,                            id_type: ObjectIdType::Notebook,                        },                        creator_uid: None,                        permissions: ServerPermissions::mock_personal(),                    },                })            });         initialize_app(&mut app);        let sync_queue = create_sync_queue(&mut app, vec![], cloud_objects_client_mock, true);         let sync_queue_events = app.add_model(|_ctx| Events::default());         sync_queue_events.update(&mut app, |_, ctx| {            ctx.subscribe_to_model(&sync_queue, |me, _, event, _ctx| me.0.push(event.clone()))        });         // Enqueue a single item and wait for the response to complete.        sync_queue            .update(&mut app, |sync_queue, ctx| {                sync_queue.enqueue(                    QueueItem::CreateObject {                        object_type: ObjectType::Notebook,                        owner: Owner::mock_current_user(),                        id: notebook_id,                        title: None,                        serialized_model: Some(Arc::new(notebook.serialized())),                        initial_folder_id: None,                        entrypoint: Default::default(),                        initiated_by: InitiatedBy::User,                    },                    ctx,                );                 ctx.await_spawned_future(sync_queue.spawned_futures[0])            })            .await;         sync_queue_events.update(&mut app, |sync_queue_events, _ctx| {            assert_eq!(sync_queue_events.0.len(), 1);            assert_eq!(                sync_queue_events[0],                SyncQueueEvent::ObjectCreationSuccessful {                    server_creation_info: ServerCreationInfo {                        server_id_and_type: ServerIdAndType {                            id: notebook_server_id,                            id_type: ObjectIdType::Notebook                        },                        creator_uid: None,                        permissions: ServerPermissions::mock_personal(),                    },                    client_id: notebook_id,                    revision_and_editor: RevisionAndLastEditor {                        revision: notebook_ts.into(),                        last_editor_uid: None                    },                    metadata_ts: notebook_ts.into(),                    initiated_by: InitiatedBy::User                }            );        })    });} #[test]fn test_generic_string_object_unique_key_failure() {    App::test((), |mut app| async move {        let owner = Owner::mock_current_user();        let gso_id = ClientId::default();        let gso_json = "{\"storage_key\":\"somepref\",\"value\":true,\"platform\":\"Global\"}";        let workflow_id = ClientId::default();        let workflow_server_id = ServerId::from(1);        let workflow_ts = DateTime::<Utc>::default();        let workflow_data = Workflow::new("my workflow", "echo hi");         // This pattern is used in a couple places to control the order of async operations.        // Basically, we wait to receive a message on a channel before certain mocks can continue,        // so that we have time to assert intermediate states, at which point we fire a message.        let (tx, rx) = std::sync::mpsc::channel();         initialize_app(&mut app);         let mut cloud_objects_client_mock = MockObjectClient::new();        cloud_objects_client_mock            .expect_create_generic_string_object()            .times(1)            .return_once(move |_, _, _| {                Ok(CreateCloudObjectResult::GenericStringObjectUniqueKeyConflict)            });        cloud_objects_client_mock            .expect_create_workflow()            .times(1)            .return_once(move |_| {                rx.recv().unwrap();                Ok(CreateCloudObjectResult::Success {                    created_cloud_object: CreatedCloudObject {                        client_id: workflow_id,                        revision_and_editor: RevisionAndLastEditor {                            revision: workflow_ts.into(),                            last_editor_uid: None,                        },                        metadata_ts: workflow_ts.into(),                        server_id_and_type: ServerIdAndType {                            id: workflow_server_id,                            id_type: ObjectIdType::Workflow,                        },                        creator_uid: None,                        permissions: ServerPermissions::mock_personal(),                    },                })            });         let sync_queue = create_sync_queue(            &mut app,            vec![                QueueItem::CreateObject {                    object_type: ObjectType::GenericStringObject(GenericStringObjectFormat::Json(                        JsonObjectType::Preference,                    )),                    owner,                    id: gso_id,                    title: None,                    serialized_model: Some(Arc::new(gso_json.to_string().into())),                    initial_folder_id: None,                    entrypoint: CloudObjectEventEntrypoint::Unknown,                    initiated_by: InitiatedBy::User,                },                QueueItem::CreateWorkflow {                    object_type: ObjectType::Workflow,                    owner,                    id: workflow_id,                    model: Arc::new(CloudWorkflowModel {                        data: workflow_data,                    }),                    initial_folder_id: None,                    entrypoint: CloudObjectEventEntrypoint::Unknown,                    initiated_by: InitiatedBy::User,                },            ],            cloud_objects_client_mock,            false,        );         let sync_queue_events = app.add_model(|_ctx| Events::default());        sync_queue_events.update(&mut app, |_, ctx| {            ctx.subscribe_to_model(&sync_queue, |me, _, event, _ctx| me.0.push(event.clone()))        });         // Await the first future.        sync_queue            .update(&mut app, |queue, ctx| {                queue.start_dequeueing(ctx);                ctx.await_spawned_future(queue.spawned_futures[0])            })            .await;         app.read(|ctx| {            // The GSO failure should not stop the queue.            assert!(sync_queue.as_ref(ctx).is_dequeueing());             let events = sync_queue_events.as_ref(ctx);            assert_eq!(events.0.len(), 1);            assert_eq!(                &events.0,                &[SyncQueueEvent::ObjectCreationFailure {                    reason: CreationFailureReason::UniqueKeyConflict {                        id: gso_id.to_hash(),                        initiated_by: InitiatedBy::User                    },                },]            );        });         // Allow the workflow to be created.        tx.send(()).unwrap();        sync_queue            .update(&mut app, |queue, ctx| {                assert!(queue.is_dequeueing());                ctx.await_spawned_future(queue.spawned_futures[1])            })            .await;         app.read(|ctx| {            // The queue should still be dequeueing.            assert!(sync_queue.as_ref(ctx).is_dequeueing());             let events = sync_queue_events.as_ref(ctx);            assert_eq!(events.0.len(), 2);            assert_eq!(                &events.0,                &[                    SyncQueueEvent::ObjectCreationFailure {                        reason: CreationFailureReason::UniqueKeyConflict {                            id: gso_id.to_hash(),                            initiated_by: InitiatedBy::User                        },                    },                    SyncQueueEvent::ObjectCreationSuccessful {                        server_creation_info: ServerCreationInfo {                            server_id_and_type: ServerIdAndType {                                id: workflow_server_id,                                id_type: ObjectIdType::Workflow                            },                            creator_uid: None,                            permissions: ServerPermissions::mock_personal(),                        },                        client_id: workflow_id,                        revision_and_editor: RevisionAndLastEditor {                            revision: workflow_ts.into(),                            last_editor_uid: None                        },                        metadata_ts: workflow_ts.into(),                        initiated_by: InitiatedBy::User                    }                ]            );        });    });} #[test]fn test_dequeue_after_transient_failure() {    App::test((), |mut app| async move {        let owner = Owner::mock_current_user();        let success_notebook_id = ClientId::default();        let success_notebook_server_id = ServerId::from(1);        let success_notebook_ts = DateTime::<Utc>::default();        let success_notebook = CloudNotebookModel {            title: "Successful Notebook".to_string(),            data: "Hello :)".to_string(),            ai_document_id: None,            conversation_id: None,        };        let failure_notebook_id = ClientId::default();        let failure_notebook = CloudNotebookModel {            title: "Failed Notebook".to_string(),            data: "Hello :(".to_string(),            ai_document_id: None,            conversation_id: None,        };         let (tx, rx) = std::sync::mpsc::channel();         let failure_attempts = 1 + super::DEFAULT_RETRY_OPTION.remaining_retries();        let mut cloud_objects_client_mock = MockObjectClient::new();        cloud_objects_client_mock            .expect_create_notebook()            .times(failure_attempts)            .returning(move |_| Err(anyhow!("Transient network error!!!!")));        cloud_objects_client_mock            .expect_create_notebook()            .times(1)            .return_once(move |_| {                rx.recv().unwrap();                Ok(CreateCloudObjectResult::Success {                    created_cloud_object: CreatedCloudObject {                        client_id: success_notebook_id,                        revision_and_editor: RevisionAndLastEditor {                            revision: success_notebook_ts.into(),                            last_editor_uid: None,                        },                        metadata_ts: success_notebook_ts.into(),                        server_id_and_type: ServerIdAndType {                            id: success_notebook_server_id,                            id_type: ObjectIdType::Notebook,                        },                        creator_uid: None,                        permissions: ServerPermissions::mock_personal(),                    },                })            });         initialize_app(&mut app);         let sync_queue = create_sync_queue(            &mut app,            vec![                QueueItem::CreateObject {                    object_type: ObjectType::Notebook,                    owner,                    id: failure_notebook_id,                    title: None,                    serialized_model: Some(Arc::new(failure_notebook.serialized())),                    initial_folder_id: None,                    entrypoint: CloudObjectEventEntrypoint::Unknown,                    initiated_by: InitiatedBy::User,                },                QueueItem::CreateObject {                    object_type: ObjectType::Notebook,                    owner,                    id: success_notebook_id,                    title: None,                    serialized_model: Some(Arc::new(success_notebook.serialized())),                    initial_folder_id: None,                    entrypoint: CloudObjectEventEntrypoint::Unknown,                    initiated_by: InitiatedBy::User,                },            ],            cloud_objects_client_mock,            false,        );         let sync_queue_events = app.add_model(|_ctx| Events::default());        sync_queue_events.update(&mut app, |_, ctx| {            ctx.subscribe_to_model(&sync_queue, |me, _, event, _ctx| me.0.push(event.clone()))        });         sync_queue            .update(&mut app, |queue, ctx| {                queue.start_dequeueing(ctx);                ctx.await_spawned_future(queue.spawned_futures[0])            })            .await;         // Wait for the first notebook's creation to fail.        // The failures are retried, but their futures are spawned on background threads,        // so we can't access them. Instead, we wait for a SyncQueue event to appear in the model.        let mut timeout = Timer::after(std::time::Duration::from_secs(20));        let mut has_event = false;        while !has_event {            if futures::poll!(&mut timeout).is_ready() {                panic!("Timed out waiting for failure");            }             Timer::after(std::time::Duration::from_millis(500)).await;            sync_queue_events.read(&app, |events, _ctx| {                has_event = !events.0.is_empty();            });        }         sync_queue_events.read(&app, |events, _| {            assert_eq!(                &events.0[0],                &SyncQueueEvent::ObjectCreationFailure {                    reason: CreationFailureReason::Other {                        id: failure_notebook_id.to_hash(),                        initiated_by: InitiatedBy::User                    },                }            );        });         // Allow the second notebook to be created.        tx.send(()).unwrap();        sync_queue            .update(&mut app, |queue, ctx| {                assert!(queue.is_dequeueing());                ctx.await_spawned_future(queue.spawned_futures[1])            })            .await;         sync_queue_events.read(&app, |events, _| {            assert_eq!(events.0.len(), 2);            assert_eq!(                &events.0,                &[                    SyncQueueEvent::ObjectCreationFailure {                        reason: CreationFailureReason::Other {                            id: failure_notebook_id.to_hash(),                            initiated_by: InitiatedBy::User                        },                    },                    SyncQueueEvent::ObjectCreationSuccessful {                        server_creation_info: ServerCreationInfo {                            server_id_and_type: ServerIdAndType {                                id: success_notebook_server_id,                                id_type: ObjectIdType::Notebook                            },                            creator_uid: None,                            permissions: ServerPermissions::mock_personal(),                        },                        client_id: success_notebook_id,                        revision_and_editor: RevisionAndLastEditor {                            revision: success_notebook_ts.into(),                            last_editor_uid: None                        },                        metadata_ts: success_notebook_ts.into(),

Lines 1–500 of 1844. Load more