@ -352,6 +352,123 @@ async fn post_events_collect(data: Json<Vec<EventCollection>>, headers: Headers,
Ok ( ( ) )
}
pub async fn log_user_event ( event_type : i32 , user_id : & UserId , device_type : i32 , ip : & IpAddr , conn : & DbConn ) {
if ! CONFIG . org_events_enabled ( ) {
return ;
}
log_user_event_impl ( event_type , user_id , device_type , None , ip , conn ) . await ;
}
async fn log_user_event_impl (
event_type : i32 ,
user_id : & UserId ,
device_type : i32 ,
event_date : Option < NaiveDateTime > ,
ip : & IpAddr ,
conn : & DbConn ,
) {
let memberships = Membership ::find_confirmed_by_user ( user_id , conn ) . await ;
let mut events : Vec < Event > = Vec ::with_capacity ( memberships . len ( ) + 1 ) ; // We need an event per org and one without an org
// Upstream saves the event also without any org_id.
let mut event = Event ::new ( event_type , event_date ) ;
event . user_uuid = Some ( user_id . clone ( ) ) ;
event . act_user_uuid = Some ( user_id . clone ( ) ) ;
event . device_type = Some ( device_type ) ;
event . ip_address = Some ( ip . to_string ( ) ) ;
events . push ( event ) ;
// For each org a user is a member of store these events per org
for membership in memberships {
let mut event = Event ::new ( event_type , event_date ) ;
event . user_uuid = Some ( user_id . clone ( ) ) ;
event . org_uuid = Some ( membership . org_uuid ) ;
event . org_user_uuid = Some ( membership . uuid ) ;
event . act_user_uuid = Some ( user_id . clone ( ) ) ;
event . device_type = Some ( device_type ) ;
event . ip_address = Some ( ip . to_string ( ) ) ;
events . push ( event ) ;
}
Event ::save_user_event ( events , conn ) . await . unwrap_or ( ( ) ) ;
}
pub async fn log_event (
event_type : i32 ,
source_uuid : & str ,
org_id : & OrganizationId ,
act_user_id : & UserId ,
device_type : i32 ,
ip : & IpAddr ,
conn : & DbConn ,
) {
if ! CONFIG . org_events_enabled ( ) {
return ;
}
log_event_impl ( event_type , source_uuid , org_id , act_user_id , device_type , None , ip , conn ) . await ;
}
#[ expect(clippy::too_many_arguments) ]
async fn log_event_impl (
event_type : i32 ,
source_uuid : & str ,
org_id : & OrganizationId ,
act_user_id : & UserId ,
device_type : i32 ,
event_date : Option < NaiveDateTime > ,
ip : & IpAddr ,
conn : & DbConn ,
) {
// Create a new empty event
let mut event = Event ::new ( event_type , event_date ) ;
match event_type {
// 1000..=1099 Are user events, they need to be logged via log_user_event()
// Cipher Events
1100 . . = 1199 = > {
event . cipher_uuid = Some ( source_uuid . to_owned ( ) . into ( ) ) ;
}
// Collection Events
1300 . . = 1399 = > {
event . collection_uuid = Some ( source_uuid . to_owned ( ) . into ( ) ) ;
}
// Group Events
1400 . . = 1499 = > {
event . group_uuid = Some ( source_uuid . to_owned ( ) . into ( ) ) ;
}
// Org User Events
1500 . . = 1599 = > {
event . org_user_uuid = Some ( source_uuid . to_owned ( ) . into ( ) ) ;
}
// 1600..=1699 Are organizational events, and they do not need the source_uuid
// Policy Events
1700 . . = 1799 = > {
event . policy_uuid = Some ( source_uuid . to_owned ( ) . into ( ) ) ;
}
// Ignore others
_ = > { }
}
event . org_uuid = Some ( org_id . clone ( ) ) ;
event . act_user_uuid = Some ( act_user_id . clone ( ) ) ;
event . device_type = Some ( device_type ) ;
event . ip_address = Some ( ip . to_string ( ) ) ;
event . save ( conn ) . await . unwrap_or ( ( ) ) ;
}
pub async fn event_cleanup_job ( pool : DbPool ) {
debug ! ( "Start events cleanup job" ) ;
if CONFIG . events_days_retain ( ) . is_none ( ) {
debug ! ( "events_days_retain is not configured, abort" ) ;
return ;
}
if let Ok ( conn ) = pool . get ( ) . await {
Event ::clean_events ( & conn ) . await . ok ( ) ;
} else {
error ! ( "Failed to get DB connection while trying to cleanup the events table" ) ;
}
}
#[ cfg(test) ]
mod tests {
use super ::* ;
@ -390,10 +507,7 @@ mod tests {
cipher . organization_uuid = Some ( org_id . clone ( ) ) ;
let admin = membership ( MembershipType ::Admin , MembershipStatus ::Confirmed ) ;
assert_eq ! (
cipher_event_scope ( & cipher , & user_id , Some ( & admin ) ) ,
Some ( CipherEventScope ::Organization ( org_id . clone ( ) ) )
) ;
assert_eq ! ( cipher_event_scope ( & cipher , & user_id , Some ( & admin ) ) , Some ( CipherEventScope ::Organization ( org_id ) ) ) ;
let accepted_admin = membership ( MembershipType ::Admin , MembershipStatus ::Accepted ) ;
assert_eq ! ( cipher_event_scope ( & cipher , & user_id , Some ( & accepted_admin ) ) , None ) ;
@ -503,120 +617,3 @@ mod tests {
assert ! ( validate_client_event_batch_size ( MAX_CLIENT_EVENT_BATCH_SIZE + 1 ) . is_err ( ) ) ;
}
}
pub async fn log_user_event ( event_type : i32 , user_id : & UserId , device_type : i32 , ip : & IpAddr , conn : & DbConn ) {
if ! CONFIG . org_events_enabled ( ) {
return ;
}
log_user_event_impl ( event_type , user_id , device_type , None , ip , conn ) . await ;
}
async fn log_user_event_impl (
event_type : i32 ,
user_id : & UserId ,
device_type : i32 ,
event_date : Option < NaiveDateTime > ,
ip : & IpAddr ,
conn : & DbConn ,
) {
let memberships = Membership ::find_confirmed_by_user ( user_id , conn ) . await ;
let mut events : Vec < Event > = Vec ::with_capacity ( memberships . len ( ) + 1 ) ; // We need an event per org and one without an org
// Upstream saves the event also without any org_id.
let mut event = Event ::new ( event_type , event_date ) ;
event . user_uuid = Some ( user_id . clone ( ) ) ;
event . act_user_uuid = Some ( user_id . clone ( ) ) ;
event . device_type = Some ( device_type ) ;
event . ip_address = Some ( ip . to_string ( ) ) ;
events . push ( event ) ;
// For each org a user is a member of store these events per org
for membership in memberships {
let mut event = Event ::new ( event_type , event_date ) ;
event . user_uuid = Some ( user_id . clone ( ) ) ;
event . org_uuid = Some ( membership . org_uuid ) ;
event . org_user_uuid = Some ( membership . uuid ) ;
event . act_user_uuid = Some ( user_id . clone ( ) ) ;
event . device_type = Some ( device_type ) ;
event . ip_address = Some ( ip . to_string ( ) ) ;
events . push ( event ) ;
}
Event ::save_user_event ( events , conn ) . await . unwrap_or ( ( ) ) ;
}
pub async fn log_event (
event_type : i32 ,
source_uuid : & str ,
org_id : & OrganizationId ,
act_user_id : & UserId ,
device_type : i32 ,
ip : & IpAddr ,
conn : & DbConn ,
) {
if ! CONFIG . org_events_enabled ( ) {
return ;
}
log_event_impl ( event_type , source_uuid , org_id , act_user_id , device_type , None , ip , conn ) . await ;
}
#[ expect(clippy::too_many_arguments) ]
async fn log_event_impl (
event_type : i32 ,
source_uuid : & str ,
org_id : & OrganizationId ,
act_user_id : & UserId ,
device_type : i32 ,
event_date : Option < NaiveDateTime > ,
ip : & IpAddr ,
conn : & DbConn ,
) {
// Create a new empty event
let mut event = Event ::new ( event_type , event_date ) ;
match event_type {
// 1000..=1099 Are user events, they need to be logged via log_user_event()
// Cipher Events
1100 . . = 1199 = > {
event . cipher_uuid = Some ( source_uuid . to_owned ( ) . into ( ) ) ;
}
// Collection Events
1300 . . = 1399 = > {
event . collection_uuid = Some ( source_uuid . to_owned ( ) . into ( ) ) ;
}
// Group Events
1400 . . = 1499 = > {
event . group_uuid = Some ( source_uuid . to_owned ( ) . into ( ) ) ;
}
// Org User Events
1500 . . = 1599 = > {
event . org_user_uuid = Some ( source_uuid . to_owned ( ) . into ( ) ) ;
}
// 1600..=1699 Are organizational events, and they do not need the source_uuid
// Policy Events
1700 . . = 1799 = > {
event . policy_uuid = Some ( source_uuid . to_owned ( ) . into ( ) ) ;
}
// Ignore others
_ = > { }
}
event . org_uuid = Some ( org_id . clone ( ) ) ;
event . act_user_uuid = Some ( act_user_id . clone ( ) ) ;
event . device_type = Some ( device_type ) ;
event . ip_address = Some ( ip . to_string ( ) ) ;
event . save ( conn ) . await . unwrap_or ( ( ) ) ;
}
pub async fn event_cleanup_job ( pool : DbPool ) {
debug ! ( "Start events cleanup job" ) ;
if CONFIG . events_days_retain ( ) . is_none ( ) {
debug ! ( "events_days_retain is not configured, abort" ) ;
return ;
}
if let Ok ( conn ) = pool . get ( ) . await {
Event ::clean_events ( & conn ) . await . ok ( ) ;
} else {
error ! ( "Failed to get DB connection while trying to cleanup the events table" ) ;
}
}