Implement UserManager

.factory/automation.yml

@@ -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_arg=--test-threads=1 --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_arg=--test-threads=1 --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_arg=--test-threads=1 --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(|result| (address, result))
+            })
+            .try_collect()?;
+
+        let errors: Vec<Error> =
+            server_connections.iter().map(|(_addr, 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().and_then(|s| Some(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)]

src/connection/network/proto/message.rs

@@ -24,7 +24,8 @@ use std::time::Duration;
 use itertools::Itertools;
 use typedb_protocol::{
     attribute, attribute_type, concept_manager, database, database_manager, entity_type, logic_manager, query_manager,
-    r#type, relation, relation_type, role_type, rule, server_manager, session, thing, thing_type, transaction,
+    r#type, relation, relation_type, role_type, rule, server_manager, session, thing, thing_type, transaction, user,
+    user_manager,
 };
 
 use super::{FromProto, IntoProto, TryFromProto, TryIntoProto};
@@ -42,6 +43,7 @@ use crate::{
     },
     error::{ConnectionError, InternalError},
     logic::{Explanation, Rule},
+    user::User,
 };
 
 impl TryIntoProto<server_manager::all::Req> for Request {
@@ -165,6 +167,71 @@ impl TryIntoProto<transaction::Client> for Request {
     }
 }
 
+impl TryIntoProto<user_manager::all::Req> for Request {
+    fn try_into_proto(self) -> Result<user_manager::all::Req> {
+        match self {
+            Self::UsersAll => Ok(user_manager::all::Req {}),
+            other => Err(InternalError::UnexpectedRequestType(format!("{other:?}")).into()),
+        }
+    }
+}
+
+impl TryIntoProto<user_manager::contains::Req> for Request {
+    fn try_into_proto(self) -> Result<user_manager::contains::Req> {
+        match self {
+            Self::UsersContain { username } => Ok(user_manager::contains::Req { username }),
+            other => Err(InternalError::UnexpectedRequestType(format!("{other:?}")).into()),
+        }
+    }
+}
+
+impl TryIntoProto<user_manager::create::Req> for Request {
+    fn try_into_proto(self) -> Result<user_manager::create::Req> {
+        match self {
+            Self::UsersCreate { username, password } => Ok(user_manager::create::Req { username, password }),
+            other => Err(InternalError::UnexpectedRequestType(format!("{other:?}")).into()),
+        }
+    }
+}
+
+impl TryIntoProto<user_manager::delete::Req> for Request {
+    fn try_into_proto(self) -> Result<user_manager::delete::Req> {
+        match self {
+            Self::UsersDelete { username } => Ok(user_manager::delete::Req { username }),
+            other => Err(InternalError::UnexpectedRequestType(format!("{other:?}")).into()),
+        }
+    }
+}
+
+impl TryIntoProto<user_manager::get::Req> for Request {
+    fn try_into_proto(self) -> Result<user_manager::get::Req> {
+        match self {
+            Self::UsersGet { username } => Ok(user_manager::get::Req { username }),
+            other => Err(InternalError::UnexpectedRequestType(format!("{other:?}")).into()),
+        }
+    }
+}
+
+impl TryIntoProto<user_manager::password_set::Req> for Request {
+    fn try_into_proto(self) -> Result<user_manager::password_set::Req> {
+        match self {
+            Self::UsersPasswordSet { username, password } => Ok(user_manager::password_set::Req { username, password }),
+            other => Err(InternalError::UnexpectedRequestType(format!("{other:?}")).into()),
+        }
+    }
+}
+
+impl TryIntoProto<user::password_update::Req> for Request {
+    fn try_into_proto(self) -> Result<user::password_update::Req> {
+        match self {
+            Self::UserPasswordUpdate { username, password_old, password_new } => {
+                Ok(user::password_update::Req { username, password_old, password_new })
+            }
+            other => Err(InternalError::UnexpectedRequestType(format!("{other:?}")).into()),
+        }
+    }
+}
+
 impl TryFromProto<server_manager::all::Res> for Response {
     fn try_from_proto(proto: server_manager::all::Res) -> Result<Self> {
         let servers = proto.servers.into_iter().map(|server| server.address.parse()).try_collect()?;
@@ -335,6 +402,48 @@ impl TryFromProto<transaction::ResPart> for TransactionResponse {
     }
 }
 
+impl FromProto<user_manager::all::Res> for Response {
+    fn from_proto(proto: user_manager::all::Res) -> Self {
+        Self::UsersAll { users: proto.users.into_iter().map(User::from_proto).collect() }
+    }
+}
+
+impl FromProto<user_manager::contains::Res> for Response {
+    fn from_proto(proto: user_manager::contains::Res) -> Self {
+        Self::UsersContain { contains: proto.contains }
+    }
+}
+
+impl FromProto<user_manager::create::Res> for Response {
+    fn from_proto(_proto: user_manager::create::Res) -> Self {
+        Self::UsersCreate
+    }
+}
+
+impl FromProto<user_manager::delete::Res> for Response {
+    fn from_proto(_proto: user_manager::delete::Res) -> Self {
+        Self::UsersDelete
+    }
+}
+
+impl FromProto<user_manager::get::Res> for Response {
+    fn from_proto(proto: user_manager::get::Res) -> Self {
+        Self::UsersGet { user: proto.user.map(User::from_proto) }
+    }
+}
+
+impl FromProto<user_manager::password_set::Res> for Response {
+    fn from_proto(_proto: user_manager::password_set::Res) -> Self {
+        Self::UsersPasswordSet
+    }
+}
+
+impl FromProto<user::password_update::Res> for Response {
+    fn from_proto(_proto: user::password_update::Res) -> Self {
+        Self::UserPasswordUpdate
+    }
+}
+
 impl IntoProto<query_manager::Req> for QueryRequest {
     fn into_proto(self) -> query_manager::Req {
         let (req, options) = match self {

src/connection/network/proto/mod.rs

@@ -24,6 +24,7 @@ mod concept;
 mod database;
 mod logic;
 mod message;
+mod user;
 
 use crate::Result;