Implement UserManager

.factory/automation.yml

@@ -58,7 +58,7 @@ build:
         bazel run @vaticle_dependencies//tool/bazelinstall:remote_cache_setup.sh
         bazel run @vaticle_dependencies//distribution/artifact:create-netrc
         tools/start-core-server.sh
-        bazel test //tests --test_arg=-- --test_arg=integration::queries::core --test_arg=--test-threads=1 --test_output=streamed && export TEST_SUCCESS=0 || export TEST_SUCCESS=1
+        bazel test //tests --test_arg=-- --test_arg=integration::queries::core --test_output=streamed && export TEST_SUCCESS=0 || export TEST_SUCCESS=1
         tools/stop-core-server.sh
         exit $TEST_SUCCESS
     test-integration-cluster:
@@ -72,7 +72,7 @@ build:
         bazel run @vaticle_dependencies//tool/bazelinstall:remote_cache_setup.sh
         bazel run @vaticle_dependencies//distribution/artifact:create-netrc
         source tools/start-cluster-servers.sh # use source to receive export vars
-        bazel test //tests --test_env=ROOT_CA=$ROOT_CA --test_arg=-- --test_arg=integration::queries::cluster --test_arg=--test-threads=1 --test_output=streamed && export TEST_SUCCESS=0 || export TEST_SUCCESS=1
+        bazel test //tests --test_env=ROOT_CA=$ROOT_CA --test_arg=-- --test_arg=integration::queries::cluster --test_output=streamed && export TEST_SUCCESS=0 || export TEST_SUCCESS=1
         tools/stop-cluster-servers.sh
         exit $TEST_SUCCESS
     test-integration-runtimes:
@@ -85,7 +85,7 @@ build:
         bazel run @vaticle_dependencies//tool/bazelinstall:remote_cache_setup.sh
         bazel run @vaticle_dependencies//distribution/artifact:create-netrc
         source tools/start-cluster-servers.sh # use source to receive export vars
-        bazel test //tests --test_env=ROOT_CA=$ROOT_CA --test_arg=-- --test_arg=integration::runtimes --test_arg=--test-threads=1 --test_output=streamed && export TEST_SUCCESS=0 || export TEST_SUCCESS=1
+        bazel test //tests --test_env=ROOT_CA=$ROOT_CA --test_arg=-- --test_arg=integration::runtimes --test_output=streamed && export TEST_SUCCESS=0 || export TEST_SUCCESS=1
         tools/stop-cluster-servers.sh
         exit $TEST_SUCCESS
     test-behaviour-concept:
@@ -97,9 +97,9 @@ build:
         export ARTIFACT_PASSWORD=$REPO_VATICLE_PASSWORD
         bazel run @vaticle_dependencies//tool/bazelinstall:remote_cache_setup.sh
         bazel run @vaticle_dependencies//distribution/artifact:create-netrc
-        tools/start-core-server.sh
-        bazel test //tests --test_arg=-- --test_arg=behaviour::concept --test_arg=--test-threads=1 --test_output=streamed && export TEST_SUCCESS=0 || export TEST_SUCCESS=1
-        tools/stop-core-server.sh
+        source tools/start-cluster-servers.sh # use source to receive export vars
+        bazel test //tests --test_env=ROOT_CA=$ROOT_CA --test_arg=-- --test_arg=behaviour::concept --test_output=streamed && export TEST_SUCCESS=0 || export TEST_SUCCESS=1
+        tools/stop-cluster-servers.sh
         exit $TEST_SUCCESS
     test-behaviour-connection:
       image: vaticle-ubuntu-22.04
@@ -110,9 +110,23 @@ build:
         export ARTIFACT_PASSWORD=$REPO_VATICLE_PASSWORD
         bazel run @vaticle_dependencies//tool/bazelinstall:remote_cache_setup.sh
         bazel run @vaticle_dependencies//distribution/artifact:create-netrc
-        tools/start-core-server.sh
-        bazel test //tests --test_arg=-- --test_arg=behaviour::connection --test_arg=--test-threads=1 --test_output=streamed && export TEST_SUCCESS=0 || export TEST_SUCCESS=1
-        tools/stop-core-server.sh
+        source tools/start-cluster-servers.sh # use source to receive export vars
+        bazel test //tests --test_env=ROOT_CA=$ROOT_CA --test_arg=-- --test_arg=behaviour::connection --test_output=streamed && export TEST_SUCCESS=0 || export TEST_SUCCESS=1
+        tools/stop-cluster-servers.sh
+        exit $TEST_SUCCESS
+    test-behaviour-typeql:
+      machine: 8-core-32-gb
+      image: vaticle-ubuntu-22.04
+      dependencies:
+        - build
+      command: |
+        export ARTIFACT_USERNAME=$REPO_VATICLE_USERNAME
+        export ARTIFACT_PASSWORD=$REPO_VATICLE_PASSWORD
+        bazel run @vaticle_dependencies//tool/bazelinstall:remote_cache_setup.sh
+        bazel run @vaticle_dependencies//distribution/artifact:create-netrc
+        source tools/start-cluster-servers.sh # use source to receive export vars
+        bazel test //tests --test_env=ROOT_CA=$ROOT_CA --test_arg=-- --test_arg=behaviour::typeql --test_output=streamed && export TEST_SUCCESS=0 || export TEST_SUCCESS=1
+        tools/stop-cluster-servers.sh
         exit $TEST_SUCCESS
     deploy-crate-snapshot:
       filter:
@@ -124,7 +138,9 @@ build:
         - test-integration-core
         - test-integration-cluster
         - test-integration-runtimes
+        - test-behaviour-concept
         - test-behaviour-connection
+        - test-behaviour-typeql
       command: |
         export DEPLOY_CRATE_TOKEN=$REPO_VATICLE_CRATES_TOKEN
         bazel run @vaticle_dependencies//tool/bazelinstall:remote_cache_setup.sh

src/connection/connection.rs

@@ -47,21 +47,23 @@ use crate::{
     },
     connection::message::{Request, Response, TransactionRequest},
     error::InternalError,
+    user::User,
     Credential, Options,
 };
 
 #[derive(Clone)]
 pub struct Connection {
     server_connections: HashMap<Address, ServerConnection>,
     background_runtime: Arc<BackgroundRuntime>,
+    username: Option<String>,
 }
 
 impl Connection {
     pub fn new_plaintext(address: impl AsRef<str>) -> Result<Self> {
         let address: Address = address.as_ref().parse()?;
         let background_runtime = Arc::new(BackgroundRuntime::new()?);
         let server_connection = ServerConnection::new_plaintext(background_runtime.clone(), address.clone())?;
-        Ok(Self { server_connections: [(address, server_connection)].into(), background_runtime })
+        Ok(Self { server_connections: [(address, server_connection)].into(), background_runtime, username: None })
     }
 
     pub fn new_encrypted<T: AsRef<str> + Sync>(init_addresses: &[T], credential: Credential) -> Result<Self> {
@@ -70,14 +72,23 @@ impl Connection {
         let init_addresses = init_addresses.iter().map(|addr| addr.as_ref().parse()).try_collect()?;
         let addresses = Self::fetch_current_addresses(background_runtime.clone(), init_addresses, credential.clone())?;
 
-        let mut server_connections = HashMap::with_capacity(addresses.len());
-        for address in addresses {
-            let server_connection =
-                ServerConnection::new_encrypted(background_runtime.clone(), address.clone(), credential.clone())?;
-            server_connections.insert(address, server_connection);
+        let server_connections: HashMap<Address, ServerConnection> = addresses
+            .into_iter()
+            .map(|address| {
+                ServerConnection::new_encrypted(background_runtime.clone(), address.clone(), credential.clone())
+                    .map(|server_connection| (address, server_connection))
+            })
+            .try_collect()?;
+
+        let errors: Vec<Error> =
+            server_connections.values().map(|conn| conn.validate()).filter_map(Result::err).collect();
+        if errors.len() == server_connections.len() {
+            Err(ConnectionError::ClusterAllNodesFailed(
+                errors.into_iter().map(|err| err.to_string()).collect::<Vec<_>>().join("\n"),
+            ))?
+        } else {
+            Ok(Self { server_connections, background_runtime, username: Some(credential.username().to_string()) })
         }
-
-        Ok(Self { server_connections, background_runtime })
     }
 
     fn fetch_current_addresses(
@@ -124,6 +135,10 @@ impl Connection {
         self.server_connections.values()
     }
 
+    pub(crate) fn username(&self) -> Option<&str> {
+        self.username.as_ref().map(|s| s.as_ref())
+    }
+
     pub(crate) fn unable_to_connect_error(&self) -> Error {
         Error::Connection(ConnectionError::ClusterUnableToConnect(
             self.addresses().map(Address::to_string).collect::<Vec<_>>().join(","),
@@ -194,6 +209,13 @@ impl ServerConnection {
         }
     }
 
+    pub(crate) fn validate(&self) -> Result {
+        match self.request_blocking(Request::DatabasesAll)? {
+            Response::DatabasesAll { databases: _ } => Ok(()),
+            _other => Err(ConnectionError::UnableToConnect().into()),
+        }
+    }
+
     pub(crate) async fn database_exists(&self, database_name: String) -> Result<bool> {
         match self.request_async(Request::DatabasesContains { database_name }).await? {
             Response::DatabasesContains { contains } => Ok(contains),
@@ -303,6 +325,60 @@ impl ServerConnection {
             other => Err(InternalError::UnexpectedResponseType(format!("{other:?}")).into()),
         }
     }
+
+    pub(crate) async fn all_users(&self) -> Result<Vec<User>> {
+        match self.request_async(Request::UsersAll).await? {
+            Response::UsersAll { users } => Ok(users),
+            other => Err(InternalError::UnexpectedResponseType(format!("{other:?}")).into()),
+        }
+    }
+
+    pub(crate) async fn contains_user(&self, username: String) -> Result<bool> {
+        match self.request_async(Request::UsersContain { username }).await? {
+            Response::UsersContain { contains } => Ok(contains),
+            other => Err(InternalError::UnexpectedResponseType(format!("{other:?}")).into()),
+        }
+    }
+
+    pub(crate) async fn create_user(&self, username: String, password: String) -> Result {
+        match self.request_async(Request::UsersCreate { username, password }).await? {
+            Response::UsersCreate => Ok(()),
+            other => Err(InternalError::UnexpectedResponseType(format!("{other:?}")).into()),
+        }
+    }
+
+    pub(crate) async fn delete_user(&self, username: String) -> Result {
+        match self.request_async(Request::UsersDelete { username }).await? {
+            Response::UsersDelete => Ok(()),
+            other => Err(InternalError::UnexpectedResponseType(format!("{other:?}")).into()),
+        }
+    }
+
+    pub(crate) async fn get_user(&self, username: String) -> Result<Option<User>> {
+        match self.request_async(Request::UsersGet { username }).await? {
+            Response::UsersGet { user } => Ok(user),
+            other => Err(InternalError::UnexpectedResponseType(format!("{other:?}")).into()),
+        }
+    }
+
+    pub(crate) async fn set_user_password(&self, username: String, password: String) -> Result {
+        match self.request_async(Request::UsersPasswordSet { username, password }).await? {
+            Response::UsersPasswordSet => Ok(()),
+            other => Err(InternalError::UnexpectedResponseType(format!("{other:?}")).into()),
+        }
+    }
+
+    pub(crate) async fn update_user_password(
+        &self,
+        username: String,
+        password_old: String,
+        password_new: String,
+    ) -> Result {
+        match self.request_async(Request::UserPasswordUpdate { username, password_old, password_new }).await? {
+            Response::UserPasswordUpdate => Ok(()),
+            other => Err(InternalError::UnexpectedResponseType(format!("{other:?}")).into()),
+        }
+    }
 }
 
 impl fmt::Debug for ServerConnection {

src/connection/message.rs

@@ -34,6 +34,7 @@ use crate::{
         Thing, ThingType, Transitivity, Value, ValueType,
     },
     logic::{Explanation, Rule},
+    user::User,
     Options, SessionType, TransactionType,
 };
 
@@ -56,6 +57,15 @@ pub(super) enum Request {
     SessionPulse { session_id: SessionID },
 
     Transaction(TransactionRequest),
+
+    UsersAll,
+    UsersContain { username: String },
+    UsersCreate { username: String, password: String },
+    UsersDelete { username: String },
+    UsersGet { username: String },
+    UsersPasswordSet { username: String, password: String },
+
+    UserPasswordUpdate { username: String, password_old: String, password_new: String },
 }
 
 #[derive(Debug)]
@@ -97,6 +107,21 @@ pub(super) enum Response {
         request_sink: UnboundedSender<transaction::Client>,
         response_source: Streaming<transaction::Server>,
     },
+
+    UsersAll {
+        users: Vec<User>,
+    },
+    UsersContain {
+        contains: bool,
+    },
+    UsersCreate,
+    UsersDelete,
+    UsersGet {
+        user: Option<User>,
+    },
+    UsersPasswordSet,
+
+    UserPasswordUpdate,
 }
 
 #[derive(Debug)]