Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

add very basic web ui #25

Merged
merged 3 commits into from
Mar 12, 2024
Merged

add very basic web ui #25

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
15fbfe5
add very basic web ui
pavadeli Mar 12, 2024
86d4271
feat: Add listen method to FirestoreProject
pavadeli Mar 12, 2024
ee7da95
refactor: Remove unused Debug import.
pavadeli Mar 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
457 changes: 447 additions & 10 deletions Cargo.lock
View file
Open in desktop

Large diffs are not rendered by default.

23 changes: 10 additions & 13 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[package]
name = "firestore-emulator"
version = "0.1.0"
version = { workspace = true }
edition = "2021"

[features]
Expand All @@ -19,15 +19,21 @@ tracing = [
[workspace]
members = ["crates/*"]

[workspace.package]
version = "0.1.0"

[workspace.dependencies]
async-trait = "0.1.77"
axum = "0.6.20"
firestore-database = { path = "crates/firestore-database" }
futures = "0.3.30"
googleapis = { path = "crates/googleapis" }
hyper = "0.14"
itertools = "0.12.1"
pin-project = "1.1.5"
prost = "0.12"
string_cache = "0.8.7"
serde = "1.0.197"
serde_with = "3.6.1"
thiserror = "1.0.57"
time = "0.3.34"
tokio = "1.36.0"
Expand All @@ -42,25 +48,16 @@ clap = { version = "4.5.1", features = ["derive", "env"] }
color-eyre = "0.6.2"
console = { version = "0.15.8", optional = true }
console-subscriber = { version = "0.2.0", optional = true }
firestore-database = { path = "crates/firestore-database" }
futures = { workspace = true }
googleapis = { workspace = true }
itertools = { workspace = true }
emulator-grpc = { path = "crates/emulator-grpc" }
emulator-ui = { path = "crates/emulator-ui" }
hybrid-axum-tonic = { path = "crates/hybrid-axum-tonic" }
string_cache = { workspace = true }
tikv-jemallocator = "0.5.4"
time = { workspace = true, optional = true }
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
tokio-stream = { workspace = true }
tonic = { workspace = true, features = ["gzip"] }
tracing = { workspace = true }
tracing-subscriber = { version = "0.3.18", optional = true }

[profile.release]
codegen-units = 1
lto = "fat"
panic = "abort"

[lints.clippy]
# Until this is resolved: https://github.com/rust-lang/rust-clippy/issues/12281
blocks_in_conditions = "allow"
20 changes: 20 additions & 0 deletions crates/emulator-grpc/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
[package]
name = "emulator-grpc"
edition = "2021"
version = { workspace = true }

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
firestore-database = { workspace = true }
futures = { workspace = true }
googleapis = { workspace = true }
itertools = { workspace = true }
tokio = { workspace = true }
tokio-stream = { workspace = true }
tonic = { workspace = true, features = ["gzip"] }
tracing = { workspace = true }

[lints.clippy]
# Until this is resolved: https://github.com/rust-lang/rust-clippy/issues/12281
blocks_in_conditions = "allow"
101 changes: 49 additions & 52 deletions src/emulator.rs → crates/emulator-grpc/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,59 +1,49 @@
use std::{collections::HashMap, ops::Deref, sync::Arc};
use std::sync::Arc;

use firestore_database::{
event::DatabaseEvent,
get_doc_name_from_write,
reference::{DocumentRef, Ref, RootRef},
utils::RwLockHashMapExt,
Database, ReadConsistency,
reference::{DocumentRef, Ref},
FirestoreDatabase, FirestoreProject, ReadConsistency,
};
use futures::{future::try_join_all, stream::BoxStream, StreamExt};
use futures::{future::try_join_all, stream::BoxStream, TryStreamExt};
use googleapis::google::{
firestore::v1::{
firestore_server::FirestoreServer,
structured_query::{CollectionSelector, FieldReference},
transaction_options::ReadWrite,
*,
},
protobuf::{Empty, Timestamp},
};
use itertools::Itertools;
use string_cache::DefaultAtom;
use tokio::sync::{mpsc, RwLock};
use tokio_stream::wrappers::ReceiverStream;
use tonic::{async_trait, Code, Request, Response, Result, Status};
use tokio::sync::mpsc;
use tokio_stream::{wrappers::ReceiverStream, StreamExt};
use tonic::{async_trait, codec::CompressionEncoding, Code, Request, Response, Result, Status};
use tracing::{info, info_span, instrument, Instrument};

use crate::{unimplemented, unimplemented_bool, unimplemented_collection, unimplemented_option};
#[macro_use]
mod utils;

pub struct FirestoreEmulator {
databases: RwLock<HashMap<DefaultAtom, Arc<Database>>>,
const MAX_MESSAGE_SIZE: usize = 50 * 1024 * 1024;

pub fn service() -> FirestoreServer<FirestoreEmulator> {
FirestoreServer::new(FirestoreEmulator::default())
.accept_compressed(CompressionEncoding::Gzip)
.send_compressed(CompressionEncoding::Gzip)
.max_decoding_message_size(MAX_MESSAGE_SIZE)
}

impl std::fmt::Debug for FirestoreEmulator {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FirestoreEmulator").finish_non_exhaustive()
}
pub struct FirestoreEmulator {
project: &'static FirestoreProject,
}

impl FirestoreEmulator {
pub fn new() -> Self {
impl Default for FirestoreEmulator {
fn default() -> Self {
Self {
databases: Default::default(),
project: FirestoreProject::get(),
}
}

// pub async fn clear(&mut self) {
// self.databases.write().await.clear();
// }

pub async fn database(&self, name: &RootRef) -> Arc<Database> {
Arc::clone(
self.databases
.get_or_insert(&name.database_id, || Database::new(name.clone()))
.await
.deref(),
)
}
}

#[async_trait]
Expand All @@ -74,6 +64,7 @@ impl firestore_server::Firestore for FirestoreEmulator {
let name: DocumentRef = name.parse()?;

let doc = self
.project
.database(&name.collection_ref.root_ref)
.await
.get_doc(&name, &consistency_selector.try_into()?)
Expand Down Expand Up @@ -105,7 +96,7 @@ impl firestore_server::Firestore for FirestoreEmulator {
} = request.into_inner();
unimplemented_option!(mask);

let database = self.database(&database.parse()?).await;
let database = self.project.database(&database.parse()?).await;
let documents: Vec<_> = documents
.into_iter()
.map(|name| name.parse::<DocumentRef>())
Expand Down Expand Up @@ -139,7 +130,7 @@ impl firestore_server::Firestore for FirestoreEmulator {
read_time: Some(Timestamp::now()),
transaction: new_transaction.take().unwrap_or_default(),
}),
Err(err) => Err(err),
Err(err) => Err(Status::from(err)),
};
if tx.send(msg).await.is_err() {
break;
Expand All @@ -164,10 +155,10 @@ impl firestore_server::Firestore for FirestoreEmulator {
transaction,
} = request.into_inner();

let database = self.database(&database.parse()?).await;
let database = self.project.database(&database.parse()?).await;

let (commit_time, write_results) = if transaction.is_empty() {
perform_writes(database.as_ref(), writes).await?
perform_writes(&database, writes).await?
} else {
let txn_id = transaction.try_into()?;
info!(?txn_id);
Expand Down Expand Up @@ -215,7 +206,7 @@ impl firestore_server::Firestore for FirestoreEmulator {
}

let parent: Ref = parent.parse()?;
let database = self.database(parent.root()).await;
let database = self.project.database(parent.root()).await;

let documents = database
.run_query(
Expand Down Expand Up @@ -275,7 +266,7 @@ impl firestore_server::Firestore for FirestoreEmulator {
) -> Result<Response<BeginTransactionResponse>> {
let BeginTransactionRequest { database, options } = request.into_inner();

let database = self.database(&database.parse()?).await;
let database = self.project.database(&database.parse()?).await;

use transaction_options::Mode;

Expand Down Expand Up @@ -308,7 +299,7 @@ impl firestore_server::Firestore for FirestoreEmulator {
database,
transaction,
} = request.into_inner();
let database = self.database(&database.parse()?).await;
let database = self.project.database(&database.parse()?).await;
database.rollback(&transaction.try_into()?).await?;
Ok(Response::new(Empty {}))
}
Expand Down Expand Up @@ -338,6 +329,7 @@ impl firestore_server::Firestore for FirestoreEmulator {
let parent: Ref = parent.parse()?;

let docs = self
.project
.database(parent.root())
.await
.run_query(parent, query, consistency_selector.try_into()?)
Expand All @@ -354,7 +346,7 @@ impl firestore_server::Firestore for FirestoreEmulator {
})
});

Ok(Response::new(stream.boxed()))
Ok(Response::new(Box::pin(stream)))
}

/// Server streaming response type for the RunAggregationQuery method.
Expand Down Expand Up @@ -406,7 +398,7 @@ impl firestore_server::Firestore for FirestoreEmulator {
}

/// Server streaming response type for the Listen method.
type ListenStream = ReceiverStream<Result<ListenResponse>>;
type ListenStream = BoxStream<'static, Result<ListenResponse>>;

/// Listens to changes. This method is only available via gRPC or WebChannel
/// (not REST).
Expand All @@ -415,15 +407,17 @@ impl firestore_server::Firestore for FirestoreEmulator {
&self,
request: Request<tonic::Streaming<ListenRequest>>,
) -> Result<Response<Self::ListenStream>> {
// TODO: refactor to be able to use database properly
Ok(Response::new(
self.database(&"projects/whatever/databases/(default)".parse()?)
.await
.listen(request.into_inner()),
))
let stream = self
.project
.listen(request.into_inner().filter_map(Result::ok))
.map_err(Into::into);
Ok(Response::new(Box::pin(stream)))
}

/// Lists all the collection IDs underneath a document.
///
/// TODO: Check that this indeed needs to be a list of deeply nested collection IDs or only the
/// direct children of the given document.
#[instrument(skip_all, err)]
async fn list_collection_ids(
&self,
Expand All @@ -437,9 +431,10 @@ impl firestore_server::Firestore for FirestoreEmulator {
} = request.into_inner();
let parent: DocumentRef = parent.parse()?;
let collection_ids = self
.project
.database(&parent.collection_ref.root_ref)
.await
.get_collection_ids(&parent)
.get_collection_ids_deep(&Ref::Document(parent))
.await?
.into_iter()
.map(|name| name.to_string())
Expand Down Expand Up @@ -478,7 +473,7 @@ impl firestore_server::Firestore for FirestoreEmulator {

let time: Timestamp = Timestamp::now();

let database = self.database(&database.parse()?).await;
let database = self.project.database(&database.parse()?).await;

let (status, write_results, updates): (Vec<_>, Vec<_>, Vec<_>) =
try_join_all(writes.into_iter().map(|write| async {
Expand All @@ -492,8 +487,8 @@ impl firestore_server::Firestore for FirestoreEmulator {
Ok((wr, update)) => (Default::default(), wr, Some((name.clone(), update))),
Err(err) => (
rpc::Status {
code: err.code() as _,
message: err.message().to_string(),
code: err.grpc_code() as _,
message: err.to_string(),
details: vec![],
},
Default::default(),
Expand All @@ -506,6 +501,7 @@ impl firestore_server::Firestore for FirestoreEmulator {
.multiunzip();

database.send_event(DatabaseEvent {
database: Arc::downgrade(&database),
update_time: time.clone(),
updates: updates.into_iter().flatten().collect(),
});
Expand All @@ -518,7 +514,7 @@ impl firestore_server::Firestore for FirestoreEmulator {
}

async fn perform_writes(
database: &Database,
database: &Arc<FirestoreDatabase>,
writes: Vec<Write>,
) -> Result<(Timestamp, Vec<WriteResult>)> {
let time: Timestamp = Timestamp::now();
Expand All @@ -534,6 +530,7 @@ async fn perform_writes(
.into_iter()
.unzip();
database.send_event(DatabaseEvent {
database: Arc::downgrade(database),
update_time: time.clone(),
updates: updates.into_iter().map(|u| (u.name().clone(), u)).collect(),
});
Expand Down
10 changes: 3 additions & 7 deletions src/utils.rs → crates/emulator-grpc/src/utils.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
#[macro_export]
macro_rules! unimplemented {
($name:expr) => {{
use tonic::Status;
Expand All @@ -8,29 +7,26 @@ macro_rules! unimplemented {
}};
}

#[macro_export]
macro_rules! unimplemented_option {
($val:expr) => {
if $val.is_some() {
$crate::unimplemented!(stringify!($val))
unimplemented!(stringify!($val))
}
};
}

#[macro_export]
macro_rules! unimplemented_collection {
($val:expr) => {
if !$val.is_empty() {
$crate::unimplemented!(stringify!($val))
unimplemented!(stringify!($val))
}
};
}

#[macro_export]
macro_rules! unimplemented_bool {
($val:expr) => {
if $val {
$crate::unimplemented!(stringify!($val))
unimplemented!(stringify!($val))
}
};
}
12 changes: 12 additions & 0 deletions crates/emulator-ui/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
[package]
name = "emulator-ui"
version = { workspace = true }
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
axum = { workspace = true, features = ["macros"] }
firestore-database = { workspace = true }
serde_json = "1.0.114"
tower-http = { version = "0.4.4", features = ["full"] }
Loading
Loading