tensamin/omega
Watch
6
0
Fork
You've already forked omega
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

[Fix] Stability

Browse source
This commit is contained in:
Alex 2026-07-27 20:36:23 +02:00
commit 8082050170
Signed by: alex
SSH key fingerprint: SHA256:D1+Ub8o0v4K5y1JNivW8IxEOelqLSvPmUzBbDIoZkRQ
20 changed files with 563 additions and 464 deletions
Download patch file Download diff file Expand all files Collapse all files

59
src/transport/handlers/presence.rs
View file

@ -21,7 +21,18 @@ pub async fn user_connected(
.and_then(UserStatus::from_str)
.unwrap_or(UserStatus::user_online);
if let Ok(user_id) = i64::try_from(user_id) {
user_online_tracker::track_user_status(user_id, status, omikron_id);
if let Some(session_id) = value
.get_data(DataType::SessionId)
.as_number()
.and_then(|id| i64::try_from(id).ok())
.filter(|id| *id > 0)
{
user_online_tracker::track_user_session_status(
user_id, session_id, status, omikron_id,
);
} else {
user_online_tracker::track_user_status(user_id, status, omikron_id);
}
}
}
Ok(())
@ -34,11 +45,55 @@ pub async fn user_disconnected(
) -> OmikronResult<()> {
log_in!(crate::util::logger::PrintType::Omega, "User disconnected");
if let Some(user_id) = value.get_data(DataType::UserId).as_number() {
user_online_tracker::untrack_user_status(user_id as i64, omikron_id);
if let Some(session_id) = value
.get_data(DataType::SessionId)
.as_number()
.and_then(|id| i64::try_from(id).ok())
.filter(|id| *id > 0)
{
user_online_tracker::untrack_user_session_status(
user_id as i64,
session_id,
omikron_id,
);
} else {
user_online_tracker::untrack_user_status(user_id as i64, omikron_id);
}
}
Ok(())
}
pub async fn client_changed(
_: Arc<OmikronConnection>,
value: CommunicationValue,
_: i64,
) -> OmikronResult<()> {
let Some(user_id) = value
.get_data(DataType::UserId)
.as_number()
.and_then(|id| i64::try_from(id).ok())
else {
return Ok(());
};
let Some(status) = value
.get_data(DataType::UserState)
.as_str()
.and_then(UserStatus::from_str)
else {
return Ok(());
};
// Connectivity is derived from routes. Clients may choose only public
// presence preferences, never server/offline states.
if matches!(
status,
UserStatus::user_offline | UserStatus::iota_offline | UserStatus::iota_online
) {
return Ok(());
}
user_online_tracker::update_user_session_status(user_id, status);
Ok(())
}
pub async fn iota_connected(
connection: Arc<OmikronConnection>,
value: CommunicationValue,

119
src/transport/handlers/register.rs
View file

@ -13,13 +13,25 @@ pub async fn get_register(
connection: Arc<OmikronConnection>,
value: CommunicationValue,
) -> OmikronResult<()> {
let register_id = user_repo::get_register_id().await?;
let iota_id = value
.get_data(DataType::IotaId)
.as_number()
.and_then(|id| i64::try_from(id).ok())
.filter(|id| user_repo::valid_protocol_id(*id));
let Some(iota_id) = iota_id else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidData)
.await;
};
let (register_id, registration_token) =
user_repo::allocate_registration(IotaId::from(iota_id), value.get_id()).await?;
let response = CommunicationValue::new(CommunicationType::GetRegister)
.with_id(value.get_id())
.add_typed_default(
DataType::UserId,
DataValue::SignedNumber(register_id.0.into()),
);
)
.add_typed_default(DataType::RegisterId, DataValue::Str(registration_token));
connection.send(&response).await
}
@ -27,10 +39,6 @@ pub async fn complete_iota(
connection: Arc<OmikronConnection>,
value: CommunicationValue,
) -> OmikronResult<()> {
let iota_id = value
.get_data(DataType::IotaId)
.as_number()
.map(|id| id as i64);
let public_key = value
.get_data(DataType::PublicKey)
.as_str()
@ -40,57 +48,25 @@ pub async fn complete_iota(
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidData)
.await;
};
match iota_id {
Some(iota_id) => {
match iota_repo::register_complete_iota(IotaId::from(iota_id), public_key).await {
Ok(()) => {
connection
.send(
&CommunicationValue::new(CommunicationType::Success)
.with_id(value.get_id()),
)
.await
}
Err(error) => {
connection
.send(
&CommunicationValue::new(CommunicationType::ErrorInternal)
.with_id(value.get_id())
.add_typed_default(
DataType::ErrorType,
DataValue::Str(error.to_string()),
),
)
.await
}
}
match iota_repo::create_new_iota(public_key).await {
Ok(id) => {
connection
.send(
&CommunicationValue::new(CommunicationType::CompleteRegisterIota)
.with_id(value.get_id())
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(id.0.into())),
)
.await
}
Err(error) => {
connection
.send(
&CommunicationValue::new(CommunicationType::ErrorInternal)
.with_id(value.get_id())
.add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string())),
)
.await
}
None => match iota_repo::create_new_iota(public_key).await {
Ok(id) => {
connection
.send(
&CommunicationValue::new(CommunicationType::CompleteRegisterIota)
.with_id(value.get_id())
.add_typed_default(
DataType::IotaId,
DataValue::SignedNumber(id.0.into()),
),
)
.await
}
Err(error) => {
connection
.send(
&CommunicationValue::new(CommunicationType::ErrorInternal)
.with_id(value.get_id())
.add_typed_default(
DataType::ErrorType,
DataValue::Str(error.to_string()),
),
)
.await
}
},
}
}
@ -101,7 +77,8 @@ pub async fn complete_user(
let user_id = value
.get_data(DataType::UserId)
.as_number()
.map(|id| id as i64);
.and_then(|id| i64::try_from(id).ok())
.filter(|id| user_repo::valid_protocol_id(*id));
let username = value
.get_data(DataType::Username)
.as_str()
@ -114,22 +91,44 @@ pub async fn complete_user(
.get_data(DataType::ResetToken)
.as_str()
.map(str::to_owned);
let Some((user_id, username, public_key, reset_token)) = user_id
let registration_token = value
.get_data(DataType::RegisterId)
.as_str()
.filter(|token| uuid::Uuid::parse_str(token).is_ok())
.map(str::to_owned);
let Some((user_id, username, public_key, reset_token, registration_token)) = user_id
.zip(username)
.zip(public_key)
.zip(reset_token)
.map(|(((id, name), key), token)| (id, name, key, token))
.zip(registration_token)
.map(|((((id, name), key), token), registration_token)| {
(id, name, key, token, registration_token)
})
else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidData)
.await;
};
// Omikron supplies the authenticated Iota ID in the payload. The lease
// check below binds completion to that Iota rather than trusting sender.
let iota_id = value
.get_data(DataType::IotaId)
.as_number()
.and_then(|id| i64::try_from(id).ok())
.filter(|id| user_repo::valid_protocol_id(*id));
let Some(iota_id) = iota_id else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidData)
.await;
};
match user_repo::register_complete_user(
UserId::from(user_id),
username,
public_key,
IotaId::from(value.get_sender() as i64),
IotaId::from(iota_id),
reset_token,
registration_token,
)
.await
{