|
|
@ -1,6 +1,9 @@
|
|
|
|
use crate::{utils, Database, PduEvent};
|
|
|
|
use crate::{utils, Database, PduEvent};
|
|
|
|
use log::debug;
|
|
|
|
use log::debug;
|
|
|
|
use ruma_events::{room::message::MessageEvent, EventType};
|
|
|
|
use ruma_events::{
|
|
|
|
|
|
|
|
room::message::{MessageEvent, MessageEventContent},
|
|
|
|
|
|
|
|
EventType,
|
|
|
|
|
|
|
|
};
|
|
|
|
use ruma_federation_api::RoomV3Pdu;
|
|
|
|
use ruma_federation_api::RoomV3Pdu;
|
|
|
|
use ruma_identifiers::{EventId, RoomId, UserId};
|
|
|
|
use ruma_identifiers::{EventId, RoomId, UserId};
|
|
|
|
use std::{
|
|
|
|
use std::{
|
|
|
@ -122,8 +125,7 @@ impl Data {
|
|
|
|
})
|
|
|
|
})
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
// TODO: Make sure this isn't called twice in parallel
|
|
|
|
pub fn pdu_leaves_get(&self, room_id: &RoomId) -> Vec<EventId> {
|
|
|
|
pub fn pdu_leaves_replace(&self, room_id: &RoomId, event_id: &EventId) -> Vec<EventId> {
|
|
|
|
|
|
|
|
let event_ids = self
|
|
|
|
let event_ids = self
|
|
|
|
.db
|
|
|
|
.db
|
|
|
|
.roomid_pduleaves
|
|
|
|
.roomid_pduleaves
|
|
|
@ -135,6 +137,10 @@ impl Data {
|
|
|
|
})
|
|
|
|
})
|
|
|
|
.collect();
|
|
|
|
.collect();
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
event_ids
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
pub fn pdu_leaves_replace(&self, room_id: &RoomId, event_id: &EventId) {
|
|
|
|
self.db
|
|
|
|
self.db
|
|
|
|
.roomid_pduleaves
|
|
|
|
.roomid_pduleaves
|
|
|
|
.clear(room_id.to_string().as_bytes());
|
|
|
|
.clear(room_id.to_string().as_bytes());
|
|
|
@ -143,15 +149,20 @@ impl Data {
|
|
|
|
&room_id.to_string().as_bytes(),
|
|
|
|
&room_id.to_string().as_bytes(),
|
|
|
|
(*event_id.to_string()).into(),
|
|
|
|
(*event_id.to_string()).into(),
|
|
|
|
);
|
|
|
|
);
|
|
|
|
|
|
|
|
|
|
|
|
event_ids
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
/// Add a persisted data unit from this homeserver
|
|
|
|
/// Add a persisted data unit from this homeserver
|
|
|
|
pub fn pdu_append_message(&self, event_id: &EventId, room_id: &RoomId, event: MessageEvent) {
|
|
|
|
pub fn pdu_append(
|
|
|
|
|
|
|
|
&self,
|
|
|
|
|
|
|
|
room_id: RoomId,
|
|
|
|
|
|
|
|
sender: UserId,
|
|
|
|
|
|
|
|
event_type: EventType,
|
|
|
|
|
|
|
|
content: MessageEventContent,
|
|
|
|
|
|
|
|
) -> EventId {
|
|
|
|
// prev_events are the leaves of the current graph. This method removes all leaves from the
|
|
|
|
// prev_events are the leaves of the current graph. This method removes all leaves from the
|
|
|
|
// room and replaces them with our event
|
|
|
|
// room and replaces them with our event
|
|
|
|
let prev_events = self.pdu_leaves_replace(room_id, event_id);
|
|
|
|
// TODO: Make sure this isn't called twice in parallel
|
|
|
|
|
|
|
|
let prev_events = self.pdu_leaves_get(&room_id);
|
|
|
|
|
|
|
|
|
|
|
|
// Our depth is the maximum depth of prev_events + 1
|
|
|
|
// Our depth is the maximum depth of prev_events + 1
|
|
|
|
let depth = prev_events
|
|
|
|
let depth = prev_events
|
|
|
@ -166,26 +177,36 @@ impl Data {
|
|
|
|
.unwrap_or(0_u64)
|
|
|
|
.unwrap_or(0_u64)
|
|
|
|
+ 1;
|
|
|
|
+ 1;
|
|
|
|
|
|
|
|
|
|
|
|
let pdu = PduEvent {
|
|
|
|
let mut pdu = PduEvent {
|
|
|
|
event_id: event_id.clone(),
|
|
|
|
event_id: EventId::try_from("$thiswillbefilledinlater").unwrap(),
|
|
|
|
room_id: room_id.clone(),
|
|
|
|
room_id: room_id.clone(),
|
|
|
|
sender: event.sender,
|
|
|
|
sender: sender.clone(),
|
|
|
|
origin: self.hostname.clone(),
|
|
|
|
origin: self.hostname.clone(),
|
|
|
|
origin_server_ts: event.origin_server_ts,
|
|
|
|
origin_server_ts: utils::millis_since_unix_epoch(),
|
|
|
|
kind: EventType::RoomMessage,
|
|
|
|
kind: event_type,
|
|
|
|
content: serde_json::to_value(event.content).unwrap(),
|
|
|
|
content: serde_json::to_value(content).expect("message content is valid json"),
|
|
|
|
state_key: None,
|
|
|
|
state_key: None,
|
|
|
|
prev_events,
|
|
|
|
prev_events,
|
|
|
|
depth: depth.try_into().unwrap(),
|
|
|
|
depth: depth.try_into().unwrap(),
|
|
|
|
auth_events: Vec::new(),
|
|
|
|
auth_events: Vec::new(),
|
|
|
|
redacts: None,
|
|
|
|
redacts: None,
|
|
|
|
unsigned: Default::default(),
|
|
|
|
unsigned: Default::default(), // TODO
|
|
|
|
hashes: ruma_federation_api::EventHash {
|
|
|
|
hashes: ruma_federation_api::EventHash {
|
|
|
|
sha256: "aaa".to_owned(),
|
|
|
|
sha256: "aaa".to_owned(),
|
|
|
|
},
|
|
|
|
},
|
|
|
|
signatures: HashMap::new(),
|
|
|
|
signatures: HashMap::new(),
|
|
|
|
};
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Generate event id
|
|
|
|
|
|
|
|
pdu.event_id = EventId::try_from(&*format!(
|
|
|
|
|
|
|
|
"${}",
|
|
|
|
|
|
|
|
ruma_signatures::reference_hash(&serde_json::to_value(&pdu).unwrap())
|
|
|
|
|
|
|
|
.expect("ruma can calculate reference hashes")
|
|
|
|
|
|
|
|
))
|
|
|
|
|
|
|
|
.expect("ruma's reference hashes are correct");
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
self.pdu_leaves_replace(&room_id, &pdu.event_id);
|
|
|
|
|
|
|
|
|
|
|
|
// The new value will need a new index. We store the last used index in 'n' + id
|
|
|
|
// The new value will need a new index. We store the last used index in 'n' + id
|
|
|
|
let mut count_key: Vec<u8> = vec![b'n'];
|
|
|
|
let mut count_key: Vec<u8> = vec![b'n'];
|
|
|
|
count_key.extend_from_slice(&room_id.to_string().as_bytes());
|
|
|
|
count_key.extend_from_slice(&room_id.to_string().as_bytes());
|
|
|
@ -213,8 +234,10 @@ impl Data {
|
|
|
|
|
|
|
|
|
|
|
|
self.db
|
|
|
|
self.db
|
|
|
|
.eventid_pduid
|
|
|
|
.eventid_pduid
|
|
|
|
.insert(event_id.to_string(), pdu_id.clone())
|
|
|
|
.insert(pdu.event_id.to_string(), pdu_id.clone())
|
|
|
|
.unwrap();
|
|
|
|
.unwrap();
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
pdu.event_id
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
/// Returns a vector of all PDUs.
|
|
|
|
/// Returns a vector of all PDUs.
|
|
|
|