diff --git a/.cargo/config.toml b/.cargo/config.toml new file mode 100644 index 0000000..46adaad --- /dev/null +++ b/.cargo/config.toml @@ -0,0 +1,2 @@ +[env] +MTP_TYPE_MAPS = { value = "mtp-type-maps/type-maps.yaml", relative = true } diff --git a/apps/tauri/.cargo/config.toml b/apps/tauri/.cargo/config.toml deleted file mode 100644 index 161678f..0000000 --- a/apps/tauri/.cargo/config.toml +++ /dev/null @@ -1,2 +0,0 @@ -[env] -MTP_TYPE_MAPS = { value = "../../mtp-type-maps/type-maps.yaml", relative = true } diff --git a/apps/tauri/src-tauri/Cargo.lock b/apps/tauri/src-tauri/Cargo.lock index d404d61..822aa43 100644 --- a/apps/tauri/src-tauri/Cargo.lock +++ b/apps/tauri/src-tauri/Cargo.lock @@ -2745,7 +2745,7 @@ dependencies = [ [[package]] name = "mtp" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git?rev=a5c8d4f0c898c78351e9d54124886c86e789a22a#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git?rev=f6a8b464e0daae6c7753208f3d8601d21cd47ab2#f6a8b464e0daae6c7753208f3d8601d21cd47ab2" dependencies = [ "mtp-client", "mtp-codec", @@ -2760,7 +2760,7 @@ dependencies = [ [[package]] name = "mtp-client" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git?rev=a5c8d4f0c898c78351e9d54124886c86e789a22a#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git?rev=f6a8b464e0daae6c7753208f3d8601d21cd47ab2#f6a8b464e0daae6c7753208f3d8601d21cd47ab2" dependencies = [ "mtp-codec", "mtp-common", @@ -2773,7 +2773,7 @@ dependencies = [ [[package]] name = "mtp-codec" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git?rev=a5c8d4f0c898c78351e9d54124886c86e789a22a#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git?rev=f6a8b464e0daae6c7753208f3d8601d21cd47ab2#f6a8b464e0daae6c7753208f3d8601d21cd47ab2" dependencies = [ "base64 0.23.1", "byteorder", @@ -2787,10 +2787,9 @@ dependencies = [ [[package]] name = "mtp-common" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git?rev=a5c8d4f0c898c78351e9d54124886c86e789a22a#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git?rev=f6a8b464e0daae6c7753208f3d8601d21cd47ab2#f6a8b464e0daae6c7753208f3d8601d21cd47ab2" dependencies = [ "quinn", - "rustls", "thiserror 2.0.19", "wtransport", ] @@ -2798,7 +2797,7 @@ dependencies = [ [[package]] name = "mtp-crypto" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git?rev=a5c8d4f0c898c78351e9d54124886c86e789a22a#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git?rev=f6a8b464e0daae6c7753208f3d8601d21cd47ab2#f6a8b464e0daae6c7753208f3d8601d21cd47ab2" dependencies = [ "base64 0.22.1", "chacha20poly1305", @@ -2820,7 +2819,7 @@ dependencies = [ [[package]] name = "mtp-host" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git?rev=a5c8d4f0c898c78351e9d54124886c86e789a22a#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git?rev=f6a8b464e0daae6c7753208f3d8601d21cd47ab2#f6a8b464e0daae6c7753208f3d8601d21cd47ab2" dependencies = [ "mtp-codec", "mtp-common", @@ -2836,7 +2835,7 @@ dependencies = [ [[package]] name = "mtp-transport" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git?rev=a5c8d4f0c898c78351e9d54124886c86e789a22a#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git?rev=f6a8b464e0daae6c7753208f3d8601d21cd47ab2#f6a8b464e0daae6c7753208f3d8601d21cd47ab2" dependencies = [ "async-trait", "mtp-codec", @@ -2856,7 +2855,7 @@ dependencies = [ [[package]] name = "mtp-type-map" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git?rev=a5c8d4f0c898c78351e9d54124886c86e789a22a#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git?rev=f6a8b464e0daae6c7753208f3d8601d21cd47ab2#f6a8b464e0daae6c7753208f3d8601d21cd47ab2" dependencies = [ "serde", "serde_yaml", @@ -2865,7 +2864,7 @@ dependencies = [ [[package]] name = "mtp-webserver" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git?rev=a5c8d4f0c898c78351e9d54124886c86e789a22a#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git?rev=f6a8b464e0daae6c7753208f3d8601d21cd47ab2#f6a8b464e0daae6c7753208f3d8601d21cd47ab2" dependencies = [ "async-trait", "bytes", diff --git a/apps/tauri/src-tauri/Cargo.toml b/apps/tauri/src-tauri/Cargo.toml index 2a68e95..fd27d12 100644 --- a/apps/tauri/src-tauri/Cargo.toml +++ b/apps/tauri/src-tauri/Cargo.toml @@ -24,7 +24,7 @@ serde_json = "1" base64 = "0.22" reqwest = { version = "0.13", default-features = false, features = ["json", "rustls"] } tokio = { version = "1", features = ["rt-multi-thread", "sync", "time"] } -mtp = { git = "https://git.methanium.net/Methanium/mtp.git", rev = "a5c8d4f0c898c78351e9d54124886c86e789a22a", features = ["client", "crypto"] } +mtp = { git = "https://git.methanium.net/Methanium/mtp.git", rev = "f6a8b464e0daae6c7753208f3d8601d21cd47ab2", features = ["client", "crypto"] } webpki-root-certs = "1" tauri-plugin-deep-link = "2" tauri-plugin-notification = "2" diff --git a/apps/tauri/src-tauri/src/mtp_backend.rs b/apps/tauri/src-tauri/src/mtp_backend.rs index 14365b8..fa28597 100644 --- a/apps/tauri/src-tauri/src/mtp_backend.rs +++ b/apps/tauri/src-tauri/src/mtp_backend.rs @@ -341,7 +341,7 @@ async fn supervise(generation: u64) { let notification_config = config.clone(); let notification_worker = tokio::spawn(async move { while let Some(frame) = notification_rx.recv().await { - if !manager().is_current(generation) { + if !manager.is_current(generation) { break; } if let Err(error) = notify_message( @@ -497,7 +497,7 @@ fn prepare_initial_state_ack( .ok_or("ClientStateSync payload is not an object")?; let session_id = required_integer(data, "SessionId")?; let version = required_integer(data, "VersionNumber")?; - let ack = CommunicationValue::new(CommunicationType::ClientStateAck) + let ack = CommunicationValue::new(communication_type("ClientStateAck")?) .with_id(request_ids.next()?) .add_typed_default(DataType::SessionId, number_to_data(session_id)) .add_typed_default(DataType::VersionNumber, number_to_data(version)); @@ -645,7 +645,7 @@ async fn await_initial_state( return Err("No Iota is currently connected".into()); } - if frame.is_type(CommunicationType::ClientStateSync) { + if frame.get_type_name() == Some("ClientStateSync") { return Ok((frame, buffered)); } @@ -740,7 +740,7 @@ async fn notify_message( let message = frame .get_data(DataType::Message) .ok_or("MessageLive omitted Message")?; - let content = container_value(message, DataType::AppContent) + let content = container_value_by_name(message, "Content") .and_then(DataValue::as_str) .ok_or("MessageLive omitted Content")?; let keyring_bytes = decode_browser_base64(&config.keyring)?; @@ -888,42 +888,24 @@ fn container_value(value: &DataValue, field: DataType) -> Option<&DataValue> { value.get_field(id) } -fn wire_field_name(name: &str) -> &str { - match name { - "Content" => "AppContent", - "CreatedAt" => "AppCreatedAt", - "MessageId" => "AppMessageId", - _ => name, - } +fn container_value_by_name<'a>(value: &'a DataValue, field: &str) -> Option<&'a DataValue> { + container_value(value, DataType::from_name(field)?) } -fn application_field_name(name: &str) -> &str { - match name { - "AppContent" => "Content", - "AppCreatedAt" => "CreatedAt", - "AppMessageId" => "MessageId", - _ => name, - } +fn communication_type(name: &str) -> Result { + CommunicationType::from_name(name).ok_or_else(|| format!("unknown communication type: {name}")) } fn json_to_frame(type_name: &str, data: Value, id: u32) -> Result { - let comm_type = CommunicationType::from_name(type_name) - .ok_or_else(|| format!("unknown communication type: {type_name}"))?; + let comm_type = communication_type(type_name)?; let mut frame = CommunicationValue::new(comm_type).with_id(id); let Value::Object(fields) = data else { return Err("MTP request data must be an object".into()); }; - let mut translated_fields = std::collections::HashSet::::with_capacity(fields.len()); for (name, value) in fields { - let wire_name = wire_field_name(&name).to_owned(); - if !translated_fields.insert(wire_name.clone()) { - return Err(format!( - "duplicate MTP field after translation: {wire_name}" - )); - } let data_type = - DataType::from_name(&wire_name).ok_or_else(|| format!("unknown data type: {name}"))?; - frame = frame.add_typed_default(data_type, json_to_data(&wire_name, value)?); + DataType::from_name(&name).ok_or_else(|| format!("unknown data type: {name}"))?; + frame = frame.add_typed_default(data_type, json_to_data(&name, value)?); } Ok(frame) } @@ -961,21 +943,13 @@ fn json_to_data(field: &str, value: Value) -> Result { ), Value::Object(fields) => { let mut entries = Vec::with_capacity(fields.len()); - let mut translated_fields = - std::collections::HashSet::::with_capacity(fields.len()); for (name, value) in fields { - let wire_name = wire_field_name(&name).to_owned(); - if !translated_fields.insert(wire_name.clone()) { - return Err(format!( - "duplicate MTP field after translation: {wire_name}" - )); - } - let data_type = DataType::from_name(&wire_name) + let data_type = DataType::from_name(&name) .ok_or_else(|| format!("unknown nested data type: {name}"))?; let id = data_type .try_to_id(&TypeMap::latest()) .ok_or_else(|| format!("unmapped data type: {name}"))?; - entries.push((id, json_to_data(&wire_name, value)?)); + entries.push((id, json_to_data(&name, value)?)); } DataValue::Container(entries) } @@ -1020,13 +994,7 @@ fn frame_data_to_json(frame: &CommunicationValue) -> Result { let name = map .data_type_name(id.0) .ok_or_else(|| format!("unknown data type id: {}", id.0))?; - let application_name = application_field_name(name); - if result.contains_key(application_name) { - return Err(format!( - "duplicate MTP field after translation: {application_name}" - )); - } - result.insert(application_name.to_owned(), data_to_json(value, &map)?); + result.insert(name.to_owned(), data_to_json(value, &map)?); } Ok(Value::Object(result)) } @@ -1057,13 +1025,7 @@ fn data_to_json(value: &DataValue, map: &TypeMap) -> Result { let name = map .data_type_name(id.0) .ok_or_else(|| format!("unknown nested data type id: {}", id.0))?; - let application_name = application_field_name(name); - if object.contains_key(application_name) { - return Err(format!( - "duplicate MTP field after translation: {application_name}" - )); - } - object.insert(application_name.to_owned(), data_to_json(value, map)?); + object.insert(name.to_owned(), data_to_json(value, map)?); } Value::Object(object) } @@ -1084,7 +1046,7 @@ mod tests { use serde_json::json; use super::{ - container_value, decode_browser_base64, decode_sdk_bytes, frame_to_json, + container_value_by_name, decode_browser_base64, decode_sdk_bytes, frame_to_json, jittered_retry_delay, json_to_frame, prepare_initial_state_ack, RequestIdAllocator, }; @@ -1117,7 +1079,7 @@ mod tests { } #[test] - fn json_content_uses_app_content_wire_type() { + fn json_content_uses_content_wire_type() { let frame = json_to_frame( "MessageEdit", json!({ @@ -1131,14 +1093,14 @@ mod tests { assert_eq!( frame - .get_data(DataType::AppContent) + .get_data(DataType::Content) .and_then(DataValue::as_str), Some("ciphertext") ); } #[test] - fn nested_json_content_uses_app_content_wire_type() { + fn nested_json_content_uses_content_wire_type() { let frame = json_to_frame( "MessageEdit", json!({ @@ -1150,35 +1112,24 @@ mod tests { let message = frame.get_data(DataType::Message).unwrap(); assert_eq!( - container_value(message, DataType::AppContent).and_then(DataValue::as_str), + container_value_by_name(message, "Content").and_then(DataValue::as_str), Some("ciphertext") ); } #[test] - fn app_content_is_exposed_as_content_to_frontend() { + fn content_is_exposed_to_frontend() { let frame = CommunicationValue::new(CommunicationType::MessageEditLive) .with_id(1) - .add_typed_default(DataType::AppContent, DataValue::Str("ciphertext".into())); + .add_typed_default(DataType::Content, DataValue::Str("ciphertext".into())); let json = frame_to_json(&frame).unwrap(); assert_eq!(json["data"]["Content"], "ciphertext"); - assert!(json["data"].get("AppContent").is_none()); - } - - #[test] - fn translated_field_collisions_are_rejected() { - assert!(json_to_frame( - "MessageEdit", - json!({ "Content": "a", "AppContent": "b" }), - 1, - ) - .is_err()); } fn valid_initial_state() -> CommunicationValue { - CommunicationValue::new(CommunicationType::ClientStateSync) + CommunicationValue::new(communication_type("ClientStateSync").unwrap()) .add_typed_default(DataType::SessionId, DataValue::UnsignedNumber(1)) .add_typed_default(DataType::VersionNumber, DataValue::UnsignedNumber(0)) .add_typed_default(DataType::CacheSchemaVersion, DataValue::UnsignedNumber(0)) @@ -1195,7 +1146,7 @@ mod tests { let (state, ack) = prepare_initial_state_ack(&valid_initial_state(), &ids).unwrap(); assert_eq!(state["SyncMode"], "full"); - assert!(ack.is_type(CommunicationType::ClientStateAck)); + assert_eq!(ack.get_type_name(), Some("ClientStateAck")); assert_eq!(ack.id(), Some(1)); } @@ -1216,7 +1167,7 @@ mod tests { #[test] fn malformed_nested_initial_state_does_not_prepare_ack() { let ids = RequestIdAllocator::new(); - let malformed = CommunicationValue::new(CommunicationType::ClientStateSync) + let malformed = CommunicationValue::new(communication_type("ClientStateSync").unwrap()) .add_typed_default(DataType::SessionId, DataValue::UnsignedNumber(1)) .add_typed_default(DataType::VersionNumber, DataValue::UnsignedNumber(0)) .add_typed_default(DataType::CacheSchemaVersion, DataValue::UnsignedNumber(0)) diff --git a/packages/cache/src/sync.tsx b/packages/cache/src/sync.tsx index 37076c1..8928e23 100644 --- a/packages/cache/src/sync.tsx +++ b/packages/cache/src/sync.tsx @@ -21,8 +21,7 @@ export function removeMissingContactSnapshots( } export default function CacheSync() { - const { addInterceptor, contextReady, freshContacts, subscribePush } = - useMTP(); + const { addInterceptor, contextReady, freshContacts, subscribe } = useMTP(); const { load } = useStorage(); const [accountId, setAccountId] = useState(0); const queueRef = useRef(Promise.resolve()); @@ -321,10 +320,19 @@ export default function CacheSync() { useEffect(() => { if (!accountId || !contextReady) return; - return subscribePush((message) => { + const handleMessage = (message: ProtocolMessage) => { void enqueue(() => synchronizePush(message)); - }); - }, [accountId, contextReady, enqueue, subscribePush, synchronizePush]); + }; + const unsubscribers = [ + subscribe("GetStates", handleMessage), + subscribe("MessageLive", handleMessage), + subscribe("MessageEditLive", handleMessage), + subscribe("MessageDeleteLive", handleMessage), + subscribe("MessageState", handleMessage), + subscribe("MessageReactionLive", handleMessage), + ]; + return () => unsubscribers.forEach((unsubscribe) => unsubscribe()); + }, [accountId, contextReady, enqueue, subscribe, synchronizePush]); return null; } diff --git a/packages/call/src/store.tsx b/packages/call/src/store.tsx index 6f7f524..bbf3f4e 100644 --- a/packages/call/src/store.tsx +++ b/packages/call/src/store.tsx @@ -1223,7 +1223,7 @@ export const useCall = create<{ export function useInitializeCall() { const navigate = useNavigate(); const location = useLocation(); - const { send, subscribePush } = useMTP(); + const { send, subscribe } = useMTP(); const { load } = useStorage(); const { insertCall } = useSession(); const { get } = useUser(); @@ -1313,14 +1313,9 @@ export function useInitializeCall() { // listen to call invites useEffect(() => { - return subscribePush(async (message) => { - if (message.type !== "CallInvite") return; - - const { CallId, CallSecret, SenderId } = message.data as { - CallId: string; - CallSecret: ProtocolCallSecret; - SenderId: number; - }; + return subscribe("CallInvite", async ({ data }) => { + const { CallId, CallSecret, SenderId } = data; + if (!CallId || !CallSecret || !SenderId) return; if (SenderId === Number(await load("user_id"))) { return; @@ -1337,7 +1332,7 @@ export function useInitializeCall() { SenderId, ); }); - }, [load, subscribePush, showCallingScreen]); + }, [load, subscribe, showCallingScreen]); // get callId from url useEffect(() => { diff --git a/packages/chat/src/context.tsx b/packages/chat/src/context.tsx index b4f1f5b..7c88df1 100644 --- a/packages/chat/src/context.tsx +++ b/packages/chat/src/context.tsx @@ -226,7 +226,7 @@ export async function fetchReplyMessage({ export default function Provider({ children }: { children: ReactNode }) { const { load } = useStorage(); - const { send, subscribePush } = useMTP(); + const { send, subscribe } = useMTP(); const { get: getUser } = useUser(); const { moveUserIdToTop } = useSession(); @@ -912,124 +912,47 @@ export default function Provider({ children }: { children: ReactNode }) { // Get live updates for message states useEffect(() => { - return subscribePush((message) => { - if (message.type === "MessageEditLive") { - if (!currentChatSecret) return; - - const rawData = message.data as { - ChatPartnerId: unknown; - SendTime: unknown; - Content: string; - }; - - const chatPartnerId = Number(rawData.ChatPartnerId); - const sendTime = Number(rawData.SendTime); - - if (!Number.isFinite(chatPartnerId) || !Number.isFinite(sendTime)) { - log( - 3, - "chat", - "yellow", - "Cancel message edit update due to invalid data", - ); - return; - } - - if (chatPartnerId !== userIdValue) { - log( - 3, - "chat", - "yellow", - "Cancel message edit update due to user ID mismatch", - { - expected: userIdValue, - received: chatPartnerId, - }, - ); - return; - } - - void decryptChatText(currentChatSecret, rawData.Content) - .then((content) => { - editMessage(sendTime, { Content: content, Edited: true }); - }) - .catch((err) => { - log(1, "chat", "red", "Failed to decrypt message edit", err, { - SendTime: sendTime, - }); - }); - return; - } - - if (message.type === "MessageReactionLive") { - const rawData = message.data as { - ChatPartnerId: unknown; - SendTime: unknown; - Reaction: string; - SenderId: unknown; - Accepted: boolean; - }; - const chatPartnerId = Number(rawData.ChatPartnerId); - const sendTime = Number(rawData.SendTime); - const senderId = Number(rawData.SenderId); - - if ( - chatPartnerId !== userIdValue || - !Number.isFinite(sendTime) || - !Number.isFinite(senderId) - ) { - return; - } - - applyLiveReaction( - sendTime, - rawData.Reaction, - senderId, - rawData.Accepted, - ); - return; - } - - if (message.type === "MessageDeleteLive") { - const data = message.data as { - ChatPartnerId: number; - SendTime: number; - }; - - if (data.ChatPartnerId !== userIdValue) return; - - removeMessage(data.SendTime); - return; - } - - if (message.type !== "MessageState") return; - - const rawData = message.data as { - ChatPartnerId: unknown; - SendTime: unknown; - MessageState: RawMessage["MessageState"]; - }; - - const nextState = { - ChatPartnerId: Number(rawData.ChatPartnerId), - SendTime: Number(rawData.SendTime), - MessageState: rawData.MessageState, - }; - - if ( - !Number.isFinite(nextState.ChatPartnerId) || - !Number.isFinite(nextState.SendTime) - ) { + const unsubscribeEdit = subscribe("MessageEditLive", ({ data }) => { + if (!currentChatSecret) return; + if (data.ChatPartnerId !== userIdValue) { log( 3, "chat", "yellow", - "Cancel message state update due to invalid data", + "Cancel message edit update due to user ID mismatch", + { + expected: userIdValue, + received: data.ChatPartnerId, + }, ); return; } - if (nextState.ChatPartnerId !== userIdValue) { + void decryptChatText(currentChatSecret, data.Content) + .then((content) => { + editMessage(data.SendTime, { Content: content, Edited: true }); + }) + .catch((err) => { + log(1, "chat", "red", "Failed to decrypt message edit", err, { + SendTime: data.SendTime, + }); + }); + }); + const unsubscribeReaction = subscribe("MessageReactionLive", ({ data }) => { + if (data.ChatPartnerId !== userIdValue) return; + applyLiveReaction( + data.SendTime, + data.Reaction, + data.SenderId, + data.Accepted, + ); + }); + const unsubscribeDelete = subscribe("MessageDeleteLive", ({ data }) => { + if (data.ChatPartnerId !== userIdValue) return; + removeMessage(data.SendTime); + }); + const unsubscribeState = subscribe("MessageState", ({ data }) => { + if (data.ChatPartnerId !== userIdValue) { log( 3, "chat", @@ -1037,22 +960,27 @@ export default function Provider({ children }: { children: ReactNode }) { "Cancel message state update due to user ID mismatch", { expected: userIdValue, - received: nextState.ChatPartnerId, + received: data.ChatPartnerId, }, ); return; } - - editMessage(nextState.SendTime, { - MessageState: nextState.MessageState, + editMessage(data.SendTime, { + MessageState: data.MessageState, }); }); + return () => { + unsubscribeEdit(); + unsubscribeReaction(); + unsubscribeDelete(); + unsubscribeState(); + }; }, [ currentChatSecret, applyLiveReaction, editMessage, removeMessage, - subscribePush, + subscribe, userIdValue, ]); diff --git a/packages/markdown/src/markdown.tsx b/packages/markdown/src/markdown.tsx index cd2a210..f8684c4 100644 --- a/packages/markdown/src/markdown.tsx +++ b/packages/markdown/src/markdown.tsx @@ -159,7 +159,7 @@ function CopyableCode({ * @param input Parameter input. * @returns InlineNode[]. */ -export function parseInlineNodes(input: string): InlineNode[] { +function parseInlineNodes(input: string): InlineNode[] { const nodes: InlineNode[] = []; let cursor = 0; @@ -205,7 +205,7 @@ export function parseInlineNodes(input: string): InlineNode[] { return nodes; } -export function parseEmojiText(input: string): InlineNode[] { +function parseEmojiText(input: string): InlineNode[] { const nodes: InlineNode[] = []; let cursor = 0; diff --git a/packages/mtp/package.json b/packages/mtp/package.json index e036029..44f5750 100644 --- a/packages/mtp/package.json +++ b/packages/mtp/package.json @@ -9,19 +9,16 @@ "scripts": { "format": "pnpm exec prettier --write .", "lint": "eslint src --ext .ts,.tsx", - "test": "vitest run", + "test": "vitest run --passWithNoTests", "build": "pnpm run test && tsc -p tsconfig.json --noEmit" }, "dependencies": { "@methanium/ui": "*", "@tauri-apps/api": "^2.11.1", - "@tensamin/crypto": "workspace:*", "@tensamin/shared": "workspace:*", "@tensamin/storage": "workspace:*", "mtp": "*", - "react": "^19.2.8", - "react-dom": "^19.2.8", - "zod": "^4.4.3" + "react": "^19.2.8" }, "devDependencies": { "eslint": "^10.8.0" diff --git a/packages/mtp/src/browser.tsx b/packages/mtp/src/browser.tsx new file mode 100644 index 0000000..755a021 --- /dev/null +++ b/packages/mtp/src/browser.tsx @@ -0,0 +1,529 @@ +import { type ReactNode, useEffect, useMemo, useRef, useState } from "react"; +import { toast as sonnerToast } from "@methanium/ui"; +import { base64ToBytes, ConnectionState, MTPClient } from "mtp"; +import createAsyncQueue from "@tensamin/shared/asyncQueue"; +import { + mtp as mtpSchemas, + type Calls, + type Communities, + type Contacts, +} from "@tensamin/shared/data"; +import { log } from "@tensamin/shared/log"; +import { useStorage } from "@tensamin/storage/context"; + +import { + type BoundSendFn, + MTPContext, + type MTPContextType, + type ProtocolMessage, + removeMissingContacts, + useMessageHandlers, +} from "./mtpContext"; +import { + DISCOVERY_TIMEOUT, + INITIAL_SYNC_TIMEOUT, + RECONNECT_JITTER, + RECONNECT_LONG_INTERVAL, + RECONNECT_RESET, + RECONNECT_TRIES, + RETRY_INTERVAL, + STATE_ACK_TIMEOUT, +} from "./values"; + +type BrowserMtpClient = Awaited>; + +function createBrowserClient( + options: Omit[0], "schemas">, +) { + return MTPClient.create({ + ...options, + schemas: mtpSchemas, + throwProtocolErrors: true, + onValidationError: (error) => { + log(1, "mtp", "red", "Failed to validate push message", error); + }, + }); +} + +function abortError(signal: AbortSignal): Error { + return signal.reason instanceof Error + ? signal.reason + : new Error("Initial state synchronization was cancelled"); +} + +function withDeadline( + promise: Promise, + timeoutMs: number, + timeoutMessage: string, + signal: AbortSignal, +): Promise { + return new Promise((resolve, reject) => { + if (signal.aborted) { + reject(abortError(signal)); + return; + } + let settled = false; + const finish = (complete: () => void) => { + if (settled) return; + settled = true; + clearTimeout(timeout); + signal.removeEventListener("abort", onAbort); + complete(); + }; + const timeout = setTimeout( + () => finish(() => reject(new Error(timeoutMessage))), + timeoutMs, + ); + const onAbort = () => finish(() => reject(abortError(signal))); + signal.addEventListener("abort", onAbort, { once: true }); + promise.then( + (value) => finish(() => resolve(value)), + (error: unknown) => finish(() => reject(error)), + ); + }); +} + +async function completeInitialSynchronization( + client: BrowserMtpClient, + subscribe: MTPContextType["subscribe"], + signal: AbortSignal, + syncTimeoutMs = INITIAL_SYNC_TIMEOUT, + ackTimeoutMs = STATE_ACK_TIMEOUT, +): Promise> { + const stateSync = new Promise>( + (resolve, reject) => { + let unsubscribeStateSync = () => {}; + let unsubscribeNoIota = () => {}; + const cleanup = () => { + clearTimeout(timeout); + unsubscribeStateSync(); + unsubscribeNoIota(); + signal.removeEventListener("abort", onAbort); + }; + const onAbort = () => { + cleanup(); + reject(abortError(signal)); + }; + const timeout = setTimeout(() => { + cleanup(); + reject(new Error("Initial state synchronization timed out")); + }, syncTimeoutMs); + signal.addEventListener("abort", onAbort, { once: true }); + if (signal.aborted) { + onAbort(); + return; + } + unsubscribeStateSync = subscribe("ClientStateSync", (message) => { + cleanup(); + resolve(message); + }); + unsubscribeNoIota = subscribe("ErrorNoIota", () => { + cleanup(); + reject(new Error("No Iota is currently connected")); + }); + }, + ); + const [, state] = await Promise.all([ + withDeadline( + client.auth(), + syncTimeoutMs, + "MTP authentication timed out", + signal, + ), + stateSync, + ]); + await withDeadline( + client.request("ClientStateAck", { + SessionId: state.data.SessionId, + VersionNumber: state.data.VersionNumber, + }), + ackTimeoutMs, + "State acknowledgement timed out", + signal, + ); + if (signal.aborted) throw abortError(signal); + return state; +} + +function protocolErrorDetails(error: unknown) { + if (typeof error !== "object" || error === null || !("type" in error)) { + return null; + } + const protocolError = error as { + id?: unknown; + type?: unknown; + frame?: unknown; + }; + return { + id: protocolError.id, + type: protocolError.type, + frame: protocolError.frame, + }; +} + +export function BrowserProvider(props: { + children: ReactNode; + blockConnection?: boolean; +}) { + const { load } = useStorage(); + const [readyState, setReadyState] = useState( + ConnectionState.Disconnected, + ); + const [identified, setIdentified] = useState(false); + const [identifying, setIdentifying] = useState(false); + const [freshCommunities, setFreshCommunities] = useState([]); + const [freshContacts, setFreshContacts] = useState([]); + const [freshCalls, setFreshCalls] = useState([]); + const clientRef = useRef(null); + const { addInterceptor, attachSubscriptions, interceptorsRef, subscribe } = + useMessageHandlers(); + const connected = readyState === ConnectionState.Connected; + + const [mtpUrl, setMtpUrl] = useState(null); + useEffect(() => { + load("omega_url").then(setMtpUrl); + }, [load]); + + const send: BoundSendFn = useMemo( + () => async (type, data, options) => { + const client = clientRef.current; + if (!client) throw new Error("mtp is not connected"); + const response = await client.request(type, data, options); + if (response.type === "GetStates") { + setFreshContacts((contacts) => + removeMissingContacts( + contacts, + response as ProtocolMessage<"GetStates">, + ), + ); + } + return response; + }, + [], + ); + + const resolveConnectionRef = useRef(() => {}); + useEffect(() => { + if (!mtpUrl) return; + let attempts = 0; + let reconnectTimer: ReturnType | null = null; + let reconnectResetTimer: ReturnType | null = null; + let reconnectScheduled = false; + let disposed = false; + let connectionGeneration = 0; + let cleanupConnection = () => {}; + + const clearReconnectTimer = () => { + if (!reconnectTimer) return; + clearTimeout(reconnectTimer); + reconnectTimer = null; + reconnectScheduled = false; + }; + const clearReconnectResetTimer = () => { + if (!reconnectResetTimer) return; + clearTimeout(reconnectResetTimer); + reconnectResetTimer = null; + }; + const scheduleReconnect = (error: unknown) => { + if (disposed || reconnectScheduled) return; + attempts += 1; + const shortRetry = attempts <= RECONNECT_TRIES; + if (!shortRetry) { + log(0, "mtp", "red", "Reconnection attempts exhausted", error); + sonnerToast.error("Connection failed", { + id: "mtp-connection-toast", + description: + error instanceof Error + ? `${error.message.split(":")[0]}. Retrying in the background.` + : "Connection lost. Retrying in the background.", + icon: null, + duration: Infinity, + closeButton: true, + promise: null, + } as unknown as Parameters[1]); + } else { + sonnerToast.loading( + `Reconnecting to server... (attempt ${attempts} of ${RECONNECT_TRIES})`, + { id: "mtp-connection-toast" }, + ); + } + const baseDelay = shortRetry ? RETRY_INTERVAL : RECONNECT_LONG_INTERVAL; + const jitter = 1 + (Math.random() * 2 - 1) * RECONNECT_JITTER; + reconnectScheduled = true; + reconnectTimer = setTimeout( + () => { + reconnectScheduled = false; + reconnectTimer = null; + void connect(); + }, + Math.round(baseDelay * jitter), + ); + }; + + async function connect() { + if (disposed || props.blockConnection) return; + const generation = ++connectionGeneration; + let client: BrowserMtpClient | null = null; + let failed = false; + let connectionReady = false; + let detachSubscriptions = () => {}; + let unsubscribeNoIota = () => {}; + const attemptAbort = new AbortController(); + const cleanup = () => { + attemptAbort.abort( + new Error("Initial state synchronization was cancelled"), + ); + unsubscribeNoIota(); + detachSubscriptions(); + client?.disconnect(); + if (clientRef.current === client) clientRef.current = null; + clearReconnectResetTimer(); + if (generation === connectionGeneration) { + setReadyState(ConnectionState.Disconnected); + setIdentified(false); + setIdentifying(false); + } + }; + cleanupConnection = cleanup; + try { + setIdentified(false); + setIdentifying(false); + const [userId, keyring] = await Promise.all([ + load("user_id"), + load("mtp_keyring"), + ]); + if (!userId || !keyring) throw new Error("Missing login credentials"); + const forcedOmikronUrl = await load("forced_omikron_url"); + const forcedOmikronPublicKey = await load("forced_omikron_public_key"); + let url = null; + let omikronPublicKey = null; + if (forcedOmikronUrl && forcedOmikronPublicKey) { + url = forcedOmikronUrl; + omikronPublicKey = forcedOmikronPublicKey; + } else { + log(2, "mtp", "purple", "Fetching Omikron data."); + const data = await fetch(`${mtpUrl}api/get/omikron/${userId}`, { + signal: AbortSignal.any([ + attemptAbort.signal, + AbortSignal.timeout(DISCOVERY_TIMEOUT), + ]), + }); + if (data.status === 404) { + throw new Error("No Omikron assignment is currently available"); + } + if (!data.ok) + throw new Error(`Omikron discovery failed: HTTP ${data.status}`); + const omikronData = (await data.json()) as { + ip_address: string; + port: number; + public_key: string; + }; + if ( + !omikronData.ip_address || + !omikronData.port || + !omikronData.public_key + ) { + throw new Error("Invalid Omikron data"); + } + url = `https://${omikronData.ip_address}:${omikronData.port}`; + omikronPublicKey = omikronData.public_key; + } + if (!url || !omikronPublicKey) { + throw new Error("Missing Omikron URL or Public Key"); + } + log(2, "mtp", "green", "Connecting to: " + url); + client = await createBrowserClient({ + url, + credentials: { clientId: userId, keyring: base64ToBytes(keyring) }, + hostPublicKey: { value: omikronPublicKey, encoding: "base64" }, + descriptor: "client", + pings: true, + logger: (event) => { + if (event.type === "state") { + if (generation !== connectionGeneration) return; + const state = client?.state ?? ConnectionState.Disconnected; + setReadyState(state); + if ( + state === ConnectionState.Disconnected && + clientRef.current === client && + !failed + ) { + failed = true; + const error = new Error("MTP connection lost"); + attemptAbort.abort(error); + if (connectionReady) { + cleanup(); + scheduleReconnect(error); + } + } + } + if (event.type !== "Pong" && event.type !== "Ping") { + log( + 2, + "mtp", + event.type === "state" + ? "purple" + : event.direction === "recv" + ? "cyan" + : event.direction === "send" + ? "gray" + : "blue", + event.type === "state" + ? event.data + : event.direction === "recv" + ? "< " + event.type + : event.direction === "send" + ? "> " + event.type + : event.type, + event, + ); + } + }, + }); + if (disposed || generation !== connectionGeneration) { + client.disconnect(); + return; + } + const activeClient = client; + clientRef.current = activeClient; + detachSubscriptions = attachSubscriptions(activeClient); + unsubscribeNoIota = subscribe("ErrorNoIota", () => { + if (clientRef.current !== activeClient || failed) return; + failed = true; + const error = new Error("No Iota is currently connected"); + attemptAbort.abort(error); + cleanup(); + scheduleReconnect(error); + }); + setReadyState(activeClient.state); + setIdentifying(true); + const finalResponse = await completeInitialSynchronization( + activeClient, + subscribe, + attemptAbort.signal, + ); + if (disposed || clientRef.current !== activeClient) return; + setFreshContacts(finalResponse.data.Contacts); + setFreshCommunities(finalResponse.data.Communities); + setFreshCalls(finalResponse.data.Calls); + connectionReady = true; + setIdentifying(false); + setIdentified(true); + clearReconnectTimer(); + clearReconnectResetTimer(); + reconnectResetTimer = setTimeout(() => { + attempts = 0; + reconnectResetTimer = null; + }, RECONNECT_RESET * 1_000); + resolveConnectionRef.current?.(); + } catch (connectError) { + if (disposed || generation !== connectionGeneration) { + client?.disconnect(); + return; + } + failed = true; + cleanup(); + const message = + connectError instanceof Error + ? connectError.message + : String(connectError ?? "Unknown error"); + log( + 0, + "mtp", + "red", + `Connection/authentication attempt failed: ${message}`, + protocolErrorDetails(connectError) ?? connectError, + ); + scheduleReconnect(connectError); + } + } + + void connect(); + return () => { + disposed = true; + clearReconnectTimer(); + clearReconnectResetTimer(); + cleanupConnection(); + setReadyState(ConnectionState.Disconnected); + setIdentified(false); + setIdentifying(false); + sonnerToast.dismiss("mtp-connection-toast"); + }; + }, [attachSubscriptions, load, mtpUrl, props.blockConnection, subscribe]); + + useEffect(() => { + return subscribe("ErrorNoIota", () => { + setIdentified(false); + setIdentifying(false); + sonnerToast.error("We couldn't reach your Iota", { + description: + "Check your network connection and try restarting your Iota", + icon: null, + duration: Infinity, + closeButton: true, + }); + resolveConnectionRef.current?.(); + }); + }, [subscribe]); + + useEffect( + () => + subscribe("GetStates", (message) => { + setFreshContacts((contacts) => + removeMissingContacts(contacts, message), + ); + }), + [subscribe], + ); + + const loadingDescription = useMemo(() => { + if (!mtpUrl) return "Loading connection details"; + if (readyState === ConnectionState.Connecting || !connected) { + return "Establishing transport channel"; + } + if (identifying || !identified) return "Waiting for authenticated session"; + return "Loading..."; + }, [connected, identified, identifying, readyState, mtpUrl]); + const contextReady = connected && identified && mtpUrl !== null; + const mtpRef = useMemo(() => createAsyncQueue<{ send: typeof send }>(), []); + useEffect(() => { + if (connected && identified && mtpUrl) { + mtpRef.set({ send }); + } + }, [connected, identified, mtpUrl, send, mtpRef]); + + const sendQueued: BoundSendFn = useMemo( + () => async (type, data, options) => { + const mtp = await mtpRef.get(); + const response = await mtp.send(type, data, options); + for (const interceptor of interceptorsRef.current) { + void Promise.resolve(interceptor({ type, data, response })).catch( + (error) => { + log(1, "mtp", "yellow", "MTP interceptor failed", error, { type }); + }, + ); + } + return response; + }, + [interceptorsRef, mtpRef], + ); + + return ( + + {props.children} + + ); +} diff --git a/packages/mtp/src/context.test.tsx b/packages/mtp/src/context.test.tsx deleted file mode 100644 index b15043a..0000000 --- a/packages/mtp/src/context.test.tsx +++ /dev/null @@ -1,220 +0,0 @@ -import { afterEach, describe, expect, it, vi } from "vitest"; - -import { RequestIdAllocator } from "./requestIds"; - -vi.mock("@methanium/ui", () => ({ - toast: { - dismiss: vi.fn(), - error: vi.fn(), - loading: vi.fn(), - }, -})); - -vi.mock("@tensamin/shared/log", () => ({ log: vi.fn() })); -vi.mock("@tensamin/storage/context", () => ({ - useStorage: () => ({ load: vi.fn() }), -})); - -const { completeInitialSynchronization, isPushType, validateResponse } = - await import("./context"); - -const validState = { - SessionId: 7, - VersionNumber: 2, - CacheSchemaVersion: 0, - SyncMode: "full", - Contacts: [], - Communities: [], - Calls: [], - Messages: [], - DeletedMessageIds: [], - DeletedContactIds: [], -}; - -function mockInitialSyncClient( - state: { type: string; data: unknown } = { - type: "ClientStateSync", - data: validState, - }, - acknowledgement: unknown = { type: "ClientStateAck", data: {} }, -) { - const handlers = new Map void>(); - const request = vi.fn().mockResolvedValue(acknowledgement); - const client = { - auth: vi.fn(async () => { - handlers.get(state.type)?.(state as never); - }), - subscribe: vi.fn((type: string, handler: (message: never) => void) => { - handlers.set(type, handler); - return () => handlers.delete(type); - }), - request, - disconnect: vi.fn(), - } as unknown as Parameters[0]; - return { client, handlers, request }; -} - -afterEach(() => { - vi.useRealTimers(); -}); - -describe("MTP protocol dispatch", () => { - it("preserves protocol errors for the request layer", () => { - const error = validateResponse("GetStates", { - id: 12, - type: "ErrorInternal", - data: { ErrorType: "temporary" }, - }); - expect(error.type).toBe("ErrorInternal"); - expect(error.id).toBe(12); - }); - - it("recognizes initial and live presence pushes", () => { - expect(isPushType("GetStates")).toBe(true); - expect(isPushType("ClientChanged")).toBe(true); - expect(isPushType("UnknownMessage")).toBe(false); - }); -}); - -describe("browser initial synchronization", () => { - it("validates state, sends a nonzero acknowledgement, then resolves", async () => { - const { client, request } = mockInitialSyncClient(); - - const state = await completeInitialSynchronization( - client, - new RequestIdAllocator(), - new AbortController().signal, - ); - - expect(state.data).toEqual(validState); - expect(request).toHaveBeenCalledWith( - "ClientStateAck", - { SessionId: 7, VersionNumber: 2 }, - { id: 1 }, - ); - }); - - it("does not acknowledge malformed state", async () => { - const { client, request } = mockInitialSyncClient({ - type: "ClientStateSync", - data: { ...validState, SyncMode: "invalid" }, - }); - - await expect( - completeInitialSynchronization( - client, - new RequestIdAllocator(), - new AbortController().signal, - ), - ).rejects.toThrow("Response validation failed"); - expect(request).not.toHaveBeenCalled(); - }); - - it("rejects ErrorNoIota during initial synchronization", async () => { - const { client, request } = mockInitialSyncClient({ - type: "ErrorNoIota", - data: {}, - }); - - await expect( - completeInitialSynchronization( - client, - new RequestIdAllocator(), - new AbortController().signal, - ), - ).rejects.toThrow("No Iota is currently connected"); - expect(request).not.toHaveBeenCalled(); - }); - - it("rejects an acknowledgement protocol error", async () => { - const { client } = mockInitialSyncClient(undefined, { - type: "ErrorInvalidData", - data: {}, - }); - - await expect( - completeInitialSynchronization( - client, - new RequestIdAllocator(), - new AbortController().signal, - ), - ).rejects.toThrow("State acknowledgement failed: ErrorInvalidData"); - }); - - it("times out a missing acknowledgement", async () => { - vi.useFakeTimers(); - const { client } = mockInitialSyncClient(); - vi.mocked(client.request).mockReturnValue(new Promise(() => {})); - const result = completeInitialSynchronization( - client, - new RequestIdAllocator(), - new AbortController().signal, - 100, - 10, - ); - const assertion = expect(result).rejects.toThrow( - "State acknowledgement timed out", - ); - - await vi.advanceTimersByTimeAsync(10); - await assertion; - }); - - it("stops immediately when the connection attempt is cancelled", async () => { - const { client } = mockInitialSyncClient(); - vi.mocked(client.auth).mockImplementation(() => new Promise(() => {})); - const controller = new AbortController(); - const result = completeInitialSynchronization( - client, - new RequestIdAllocator(), - controller.signal, - ); - - controller.abort(new Error("MTP connection lost")); - await expect(result).rejects.toThrow("MTP connection lost"); - }); - - it("times out authentication while waiting for initial state", async () => { - vi.useFakeTimers(); - const { client, handlers } = mockInitialSyncClient(); - vi.mocked(client.auth).mockImplementation(() => { - handlers.get("ClientStateSync")?.({ - type: "ClientStateSync", - data: validState, - } as never); - return new Promise(() => {}); - }); - const result = completeInitialSynchronization( - client, - new RequestIdAllocator(), - new AbortController().signal, - 100, - 10, - ); - const assertion = expect(result).rejects.toThrow( - "MTP authentication timed out", - ); - - await vi.advanceTimersByTimeAsync(100); - await assertion; - }); - - it("uses a fresh request ID namespace for each connection", async () => { - const first = mockInitialSyncClient(); - const second = mockInitialSyncClient(); - - await completeInitialSynchronization( - first.client, - new RequestIdAllocator(), - new AbortController().signal, - ); - await completeInitialSynchronization( - second.client, - new RequestIdAllocator(), - new AbortController().signal, - ); - - expect(first.request.mock.calls[0]?.[2]).toEqual({ id: 1 }); - expect(second.request.mock.calls[0]?.[2]).toEqual({ id: 1 }); - }); -}); diff --git a/packages/mtp/src/context.tsx b/packages/mtp/src/context.tsx index 53c1862..c07579f 100644 --- a/packages/mtp/src/context.tsx +++ b/packages/mtp/src/context.tsx @@ -1,1110 +1,41 @@ -import { - createContext, - type ReactNode, - useCallback, - useContext, - useEffect, - useMemo, - useRef, - useState, -} from "react"; -import { invoke, isTauri } from "@tauri-apps/api/core"; -import { listen, type UnlistenFn } from "@tauri-apps/api/event"; +import { type ReactNode, useContext, useEffect, useState } from "react"; +import { isTauri } from "@tauri-apps/api/core"; import { MTPClient } from "mtp"; -import { type z } from "zod"; -import { ConnectionState } from "mtp"; -import createAsyncQueue from "@tensamin/shared/asyncQueue"; -import { toast as sonnerToast } from "@methanium/ui"; -import { - type Calls, - type Communities, - type Contacts, - mtp as schemas, - type MTP as Schemas, -} from "@tensamin/shared/data"; -import { log } from "@tensamin/shared/log"; -import { ProtocolError } from "@tensamin/shared/errors"; -import { useStorage } from "@tensamin/storage/context"; - -import { fromWireMessage, toWireData } from "./protocolFields"; -import { RequestIdAllocator } from "./requestIds"; -import { - DISCOVERY_TIMEOUT, - INITIAL_SYNC_TIMEOUT, - RECONNECT_JITTER, - RECONNECT_LONG_INTERVAL, - RECONNECT_RESET, - RECONNECT_TRIES, - RETRY_INTERVAL, - STATE_ACK_TIMEOUT, -} from "./values"; - -type BrowserMtpClient = Awaited>; - -function base64ToUint8Array(b64: string) { - const bin = atob(b64); - const out = new Uint8Array(bin.length); - - for (let i = 0; i < bin.length; i++) { - out[i] = bin.charCodeAt(i); - } - - return out; -} - -export type ProtocolMessage< - T extends keyof Schemas & string = keyof Schemas & string, -> = { - id?: number; - type: T | string; - data: z.infer; -}; - -export type BoundSendFn = ( - type: T, - data?: z.infer, -) => Promise>; - -export type PushHandler = (message: ProtocolMessage) => void | Promise; - -const PUSH_TYPES = [ - "MessageLive", - "MessageEditLive", - "MessageReactionLive", - "MessageDeleteLive", - "MessageState", - "CallInvite", - "GetStates", - "ClientChanged", - "ErrorNoIota", -] as const; - -export function isPushType(type: string): boolean { - return (PUSH_TYPES as readonly string[]).includes(type); -} - -function normalizeMtpMessage(message: T): T { - return fromWireMessage(message); -} - -async function requestWithId( - client: BrowserMtpClient, - ids: RequestIdAllocator, - type: string, - data: Record, -) { - let id: number; - try { - id = ids.allocate(); - } catch (error) { - client.disconnect(); - throw error; - } - - return client.request(type, toWireData(data) as Record, { - id, - }); -} - -function removeMissingContacts( - contacts: Contacts, - message: ProtocolMessage, -): Contacts { - if (message.type !== "GetStates") return contacts; - const data = message.data as { MissingUserIds?: unknown }; - if (!Array.isArray(data.MissingUserIds)) return contacts; - const missing = new Set( - data.MissingUserIds.filter( - (userId): userId is number => typeof userId === "number", - ), - ); - return contacts.filter((contact) => !missing.has(contact.UserId)); -} - -export type MTPExchange = { - type: keyof Schemas & string; - data: unknown; - response: ProtocolMessage; -}; - -export type MTPInterceptor = (exchange: MTPExchange) => void | Promise; - -type ContextType = { - send: BoundSendFn; - subscribe: ( - type: T, - handler: (message: ProtocolMessage) => void, - ) => () => void; - subscribePush: (handler: PushHandler) => () => void; - addInterceptor: (interceptor: MTPInterceptor) => () => void; - readyState: number; - identified: boolean; - freshContacts: Contacts; - freshCommunities: Communities; - freshCalls: Calls; - contextReady: boolean; - loadingDescription: string; -}; - -const MTPContext = createContext(undefined); - -function getProtocolErrorDetails(error: unknown) { - if (typeof error !== "object" || error === null || !("type" in error)) { - return null; - } - - const protocolError = error as { - id?: unknown; - type?: unknown; - data?: unknown; - }; - return { - id: protocolError.id, - type: protocolError.type, - data: protocolError.data, - }; -} - -// Zod schema validation -export function validateResponse( - type: T, - message: { id?: number; type: string; data: unknown }, -): ProtocolMessage { - if (message.type.startsWith("Error")) { - return message as ProtocolMessage; - } - - const schema = - schemas[message.type as keyof Schemas & string]?.response ?? - schemas[type]?.response; - if (!schema) { - return message as ProtocolMessage; - } - - const parsed = schema.safeParse(message.data); - if (!parsed.success) { - throw new Error( - `Response validation failed for ${type}: ${parsed.error.message}`, - ); - } - - return { - id: message.id, - type: message.type, - data: parsed.data, - } as ProtocolMessage; -} - -function abortError(signal: AbortSignal): Error { - return signal.reason instanceof Error - ? signal.reason - : new Error("Initial state synchronization was cancelled"); -} - -function withDeadline( - promise: Promise, - timeoutMs: number, - timeoutMessage: string, - signal: AbortSignal, -): Promise { - return new Promise((resolve, reject) => { - if (signal.aborted) { - reject(abortError(signal)); - return; - } - let settled = false; - const finish = (complete: () => void) => { - if (settled) return; - settled = true; - clearTimeout(timeout); - signal.removeEventListener("abort", onAbort); - complete(); - }; - const timeout = setTimeout( - () => finish(() => reject(new Error(timeoutMessage))), - timeoutMs, - ); - const onAbort = () => finish(() => reject(abortError(signal))); - signal.addEventListener("abort", onAbort, { once: true }); - promise.then( - (value) => finish(() => resolve(value)), - (error: unknown) => finish(() => reject(error)), - ); - }); -} - -export async function completeInitialSynchronization( - client: BrowserMtpClient, - ids: RequestIdAllocator, - signal: AbortSignal, - syncTimeoutMs = INITIAL_SYNC_TIMEOUT, - ackTimeoutMs = STATE_ACK_TIMEOUT, -): Promise> { - const stateSync = new Promise>( - (resolve, reject) => { - let unsubscribeStateSync = () => {}; - let unsubscribeNoIota = () => {}; - const cleanup = () => { - clearTimeout(timeout); - unsubscribeStateSync(); - unsubscribeNoIota(); - signal.removeEventListener("abort", onAbort); - }; - const onAbort = () => { - cleanup(); - reject(abortError(signal)); - }; - const timeout = setTimeout(() => { - cleanup(); - reject(new Error("Initial state synchronization timed out")); - }, syncTimeoutMs); - signal.addEventListener("abort", onAbort, { once: true }); - if (signal.aborted) { - onAbort(); - return; - } - unsubscribeStateSync = client.subscribe("ClientStateSync", (message) => { - cleanup(); - try { - resolve( - validateResponse("ClientStateSync", normalizeMtpMessage(message)), - ); - } catch (error) { - reject(error); - } - }); - unsubscribeNoIota = client.subscribe("ErrorNoIota", () => { - cleanup(); - reject(new Error("No Iota is currently connected")); - }); - }, - ); - const [, state] = await Promise.all([ - withDeadline( - client.auth(), - syncTimeoutMs, - "MTP authentication timed out", - signal, - ), - stateSync, - ]); - if (state.type.startsWith("Error")) { - throw new Error(`State synchronization failed: ${state.type}`); - } - const acknowledgement = normalizeMtpMessage( - await withDeadline( - requestWithId(client, ids, "ClientStateAck", { - SessionId: state.data.SessionId, - VersionNumber: state.data.VersionNumber, - }), - ackTimeoutMs, - "State acknowledgement timed out", - signal, - ), - ); - if (acknowledgement.type.startsWith("Error")) { - throw new Error(`State acknowledgement failed: ${acknowledgement.type}`); - } - if (signal.aborted) throw abortError(signal); - return state; -} - -function useMessageHandlers() { - const interceptorsRef = useRef(new Set()); - const pushHandlersRef = useRef(new Set()); - const lastInitialStateRef = useRef(null); - const subscribePush = useCallback((handler: PushHandler) => { - pushHandlersRef.current.add(handler); - const initialState = lastInitialStateRef.current; - if (initialState?.type === "GetStates") { - void Promise.resolve(handler(initialState)).catch(() => undefined); - } - return () => pushHandlersRef.current.delete(handler); - }, []); - const addInterceptor = useCallback((interceptor: MTPInterceptor) => { - interceptorsRef.current.add(interceptor); - return () => interceptorsRef.current.delete(interceptor); - }, []); - return { - addInterceptor, - interceptorsRef, - lastInitialStateRef, - pushHandlersRef, - subscribePush, - }; -} - -function BrowserProvider(props: { - children: ReactNode; - blockConnection?: boolean; -}) { - const { load } = useStorage(); - - const [readyState, setReadyState] = useState( - ConnectionState.Disconnected, - ); - const [identified, setIdentified] = useState(false); - const [identifying, setIdentifying] = useState(false); - - const [freshCommunities, setFreshCommunities] = useState([]); - const [freshContacts, setFreshContacts] = useState([]); - const [freshCalls, setFreshCalls] = useState([]); - - const clientRef = useRef> | null>( - null, - ); - const requestIdsRef = useRef(null); - const { - addInterceptor, - interceptorsRef, - lastInitialStateRef, - pushHandlersRef, - subscribePush, - } = useMessageHandlers(); - - const connected = readyState === ConnectionState.Connected; - - // MTP url - const [mtpUrl, setMtpUrl] = useState(null); - useEffect(() => { - load("omega_url").then(setMtpUrl); - }, [load]); - - // Validation override functions - const send: BoundSendFn = useMemo( - () => async (type, data) => { - const client = clientRef.current; - const ids = requestIdsRef.current; - - if (!client) { - throw new Error("mtp is not connected"); - } - if (!ids) { - throw new Error("MTP request allocator is unavailable"); - } - - const rawMessage = await requestWithId( - client, - ids, - type, - (data ?? {}) as Record, - ); - const response = validateResponse(type, normalizeMtpMessage(rawMessage)); - setFreshContacts((contacts) => removeMissingContacts(contacts, response)); - if (response.type.startsWith("Error")) { - const errorData = response.data as Record; - throw new ProtocolError({ - type: response.type, - requestId: response.id, - errorType: - typeof errorData.ErrorType === "string" - ? errorData.ErrorType - : undefined, - }); - } - return response; - }, - [], - ); - - const subscribe = useCallback((type, handler) => { - const client = clientRef.current; - if (!client) { - return () => {}; - } - - return client.subscribe(type, (message) => { - handler(validateResponse(type, normalizeMtpMessage(message))); - }); - }, []); - - // Reconnect stuff - const resolveConnectionRef = useRef(() => {}); - useEffect(() => { - if (!mtpUrl) return; - - let attempts = 0; - let reconnectTimer: ReturnType | null = null; - let reconnectResetTimer: ReturnType | null = null; - let reconnectScheduled = false; - let disposed = false; - let connectionGeneration = 0; - - const clearReconnectTimer = () => { - if (!reconnectTimer) return; - clearTimeout(reconnectTimer); - reconnectTimer = null; - reconnectScheduled = false; - }; - - const clearReconnectResetTimer = () => { - if (!reconnectResetTimer) return; - clearTimeout(reconnectResetTimer); - reconnectResetTimer = null; - }; - - const scheduleReconnect = (error: unknown) => { - if (disposed || reconnectScheduled) return; - attempts += 1; - const shortRetry = attempts <= RECONNECT_TRIES; - if (!shortRetry) { - log(0, "mtp", "red", "Reconnection attempts exhausted", error); - sonnerToast.error("Connection failed", { - id: "mtp-connection-toast", - description: - error instanceof Error - ? `${error.message.split(":")[0]}. Retrying in the background.` - : "Connection lost. Retrying in the background.", - icon: null, - duration: Infinity, - closeButton: true, - promise: null, - } as unknown as Parameters[1]); - } else { - sonnerToast.loading( - `Reconnecting to server... (attempt ${attempts} of ${RECONNECT_TRIES})`, - { id: "mtp-connection-toast" }, - ); - } - - const baseDelay = shortRetry ? RETRY_INTERVAL : RECONNECT_LONG_INTERVAL; - const jitter = 1 + (Math.random() * 2 - 1) * RECONNECT_JITTER; - reconnectScheduled = true; - reconnectTimer = setTimeout( - () => { - reconnectScheduled = false; - reconnectTimer = null; - void connect(); - }, - Math.round(baseDelay * jitter), - ); - }; - - async function connect() { - if (disposed || props.blockConnection) return; - - const generation = ++connectionGeneration; - let client: Awaited> | null = null; - let failed = false; - let connectionReady = false; - const attemptAbort = new AbortController(); - const cleanup = () => { - attemptAbort.abort( - new Error("Initial state synchronization was cancelled"), - ); - client?.disconnect(); - if (clientRef.current === client) { - clientRef.current = null; - requestIdsRef.current = null; - } - clearReconnectResetTimer(); - if (generation === connectionGeneration) { - setReadyState(ConnectionState.Disconnected); - setIdentified(false); - setIdentifying(false); - } - }; - try { - setIdentified(false); - setIdentifying(false); - - const [userId, keyring] = await Promise.all([ - load("user_id"), - load("mtp_keyring"), - ]); - if (!userId || !keyring) { - throw new Error("Missing login credentials"); - } - const forcedOmikronUrl = await load("forced_omikron_url"); - const forcedOmikronPublicKey = await load("forced_omikron_public_key"); - - let url = null; - let omikronPublicKey = null; - if (forcedOmikronUrl && forcedOmikronPublicKey) { - url = forcedOmikronUrl; - omikronPublicKey = forcedOmikronPublicKey; - } else { - log(2, "mtp", "purple", "Fetching Omikron data."); - const data = await fetch(`${mtpUrl}api/get/omikron/${userId}`, { - signal: AbortSignal.any([ - attemptAbort.signal, - AbortSignal.timeout(DISCOVERY_TIMEOUT), - ]), - }); - - if (data.status === 404) { - throw new Error("No Omikron assignment is currently available"); - } - if (!data.ok) { - throw new Error(`Omikron discovery failed: HTTP ${data.status}`); - } - const omikronData = (await data.json()) as { - id: number; - ip_address: string; - port: number; - public_key: string; - status: string; - }; - - if ( - !omikronData.ip_address || - !omikronData.port || - !omikronData.public_key - ) - throw new Error("Invalid Omikron data"); - - url = `https://${omikronData.ip_address}:${omikronData.port}`; - omikronPublicKey = omikronData.public_key; - } - //codec.decode(new Uint8Array(await res.arrayBuffer())), - - if (!url || !omikronPublicKey) - throw new Error("Missing Omikron URL or Public Key"); - - log(2, "mtp", "green", "Connecting to: " + url); - - client = await MTPClient.create({ - url, - credentials: { - clientId: userId, - keyring: base64ToUint8Array(keyring), - }, - hostPublicKey: { - value: omikronPublicKey, - encoding: "base64", - }, - descriptor: "client", - pings: true, - logger: (event) => { - if (event.type === "state") { - if (generation !== connectionGeneration) return; - const state = client?.state ?? ConnectionState.Disconnected; - setReadyState(state); - if ( - state === ConnectionState.Disconnected && - clientRef.current === client && - !failed - ) { - failed = true; - const error = new Error("MTP connection lost"); - attemptAbort.abort(error); - if (connectionReady) { - cleanup(); - scheduleReconnect(error); - } - } - } - - if (event.type !== "Pong" && event.type !== "Ping") { - log( - 2, - "mtp", - event.type === "state" - ? "purple" - : event.direction === "recv" - ? "cyan" - : event.direction === "send" - ? "gray" - : "blue", - event.type === "state" - ? event.data - : event.direction === "recv" - ? "< " + event.type - : event.direction === "send" - ? "> " + event.type - : event.type, - event, - ); - } - }, - }); - - if (disposed || generation !== connectionGeneration) { - client.disconnect(); - return; - } - const activeClient = client; - - clientRef.current = activeClient; - requestIdsRef.current = new RequestIdAllocator(); - for (const type of PUSH_TYPES) { - activeClient.subscribe(type, (message) => { - let validated: ProtocolMessage; - try { - validated = validateResponse(type, normalizeMtpMessage(message)); - } catch (error) { - log(1, "mtp", "red", "Failed to validate push message", error, { - type, - data: message.data, - }); - return; - } - - setFreshContacts((contacts) => - removeMissingContacts(contacts, validated), - ); - for (const handler of [...pushHandlersRef.current]) { - void Promise.resolve() - .then(() => handler(validated)) - .catch((error) => { - log(1, "mtp", "red", "Push handler failed", error, { type }); - }); - } - if (validated.type === "GetStates") { - lastInitialStateRef.current = validated; - } - if ( - validated.type === "ErrorNoIota" && - clientRef.current === activeClient && - !failed - ) { - failed = true; - const error = new Error("No Iota is currently connected"); - attemptAbort.abort(error); - cleanup(); - scheduleReconnect(error); - } - }); - } - setReadyState(activeClient.state); - setIdentifying(true); - - const ids = requestIdsRef.current; - if (!ids) throw new Error("MTP request allocator is unavailable"); - const finalResponse = await completeInitialSynchronization( - activeClient, - ids, - attemptAbort.signal, - ); - - if (disposed || clientRef.current !== activeClient) return; - - setFreshContacts(finalResponse.data.Contacts); - setFreshCommunities(finalResponse.data.Communities); - setFreshCalls(finalResponse.data.Calls); - connectionReady = true; - setIdentifying(false); - setIdentified(true); - - clearReconnectTimer(); - clearReconnectResetTimer(); - reconnectResetTimer = setTimeout(() => { - attempts = 0; - reconnectResetTimer = null; - }, RECONNECT_RESET * 1_000); - resolveConnectionRef.current?.(); - } catch (connectError) { - if (disposed || generation !== connectionGeneration) { - client?.disconnect(); - return; - } - failed = true; - cleanup(); - const connectErrorMessage = - connectError instanceof Error - ? connectError.message - : String(connectError ?? "Unknown error"); - log( - 0, - "mtp", - "red", - `Connection/authentication attempt failed: ${connectErrorMessage}`, - getProtocolErrorDetails(connectError) ?? connectError, - ); - - scheduleReconnect(connectError); - } - } - - void connect(); - - return () => { - disposed = true; - clearReconnectTimer(); - clearReconnectResetTimer(); - - clientRef.current?.disconnect(); - clientRef.current = null; - requestIdsRef.current = null; - setReadyState(ConnectionState.Disconnected); - setIdentified(false); - setIdentifying(false); - sonnerToast.dismiss("mtp-connection-toast"); - }; - }, [ - lastInitialStateRef, - mtpUrl, - props.blockConnection, - load, - pushHandlersRef, - ]); - - // No Iota check - useEffect(() => { - if (!connected) return; - - return subscribe("ErrorNoIota", () => { - setIdentified(false); - setIdentifying(false); - sonnerToast.error("We couldn't reach your Iota", { - description: - "Check your network connection and try restarting your Iota", - icon: null, - duration: Infinity, - closeButton: true, - }); - resolveConnectionRef.current?.(); - }); - }, [connected, subscribe]); - - // Async queue - const loadingDescription = useMemo(() => { - if (!mtpUrl) return "Loading connection details"; - if (readyState === ConnectionState.Connecting || !connected) { - return "Establishing transport channel"; - } - if (identifying || !identified) return "Waiting for authenticated session"; - return "Loading..."; - }, [connected, identified, identifying, readyState, mtpUrl]); - const contextReady = connected && identified && mtpUrl !== null; - const mtpRef = useMemo( - () => - createAsyncQueue<{ - send: typeof send; - subscribe: typeof subscribe; - subscribePush: typeof subscribePush; - }>(), - [], - ); - useEffect(() => { - if (connected && identified && mtpUrl) { - mtpRef.set({ - send, - subscribe, - subscribePush, - }); - } - }, [connected, identified, mtpUrl, send, subscribe, subscribePush, mtpRef]); - - const sendQueued: BoundSendFn = useMemo( - () => async (type, data) => { - const mtp = await mtpRef.get(); - const response = await mtp.send(type, data); - for (const interceptor of interceptorsRef.current) { - void Promise.resolve( - interceptor({ type, data, response: response as ProtocolMessage }), - ).catch((error) => { - log(1, "mtp", "yellow", "MTP interceptor failed", error, { type }); - }); - } - return response; - }, - [interceptorsRef, mtpRef], - ); - - return ( - - {props.children} - - ); -} - -type NativeSnapshot = { - generation: number; - readyState: number; - identified: boolean; - state?: unknown; - error?: string; -}; - -function TauriProvider(props: { - children: ReactNode; - blockConnection?: boolean; -}) { - const [snapshot, setSnapshot] = useState({ - generation: 0, - readyState: ConnectionState.Disconnected, - identified: false, - }); - const [freshContacts, setFreshContacts] = useState([]); - const [freshCommunities, setFreshCommunities] = useState([]); - const [freshCalls, setFreshCalls] = useState([]); - const generationRef = useRef(0); - const { - addInterceptor, - interceptorsRef, - lastInitialStateRef, - pushHandlersRef, - subscribePush, - } = useMessageHandlers(); - const subscriptionsRef = useRef( - new Map void>>(), - ); - - const applySnapshot = useCallback((next: NativeSnapshot) => { - if (next.generation < generationRef.current) return; - generationRef.current = next.generation; - if (next.error) { - log(0, "android", "orange", "MTP connection failed", next.error); - } - if (!next.identified) { - setSnapshot(next); - return; - } - if (next.state === undefined) { - setSnapshot({ - ...next, - identified: false, - error: "Native MTP connection omitted initial state", - }); - return; - } - const parsed = schemas.ClientStateSync.response.safeParse(next.state); - if (!parsed.success) { - log(0, "mtp", "red", "Invalid native MTP state", parsed.error); - setSnapshot({ - ...next, - identified: false, - error: "Invalid ClientStateSync payload", - }); - return; - } - setFreshContacts(parsed.data.Contacts); - setFreshCommunities(parsed.data.Communities); - setFreshCalls(parsed.data.Calls); - setSnapshot(next); - }, []); - - const dispatchMessage = useCallback( - (raw: unknown) => { - if (!raw || typeof raw !== "object" || !("type" in raw)) return; - const message = raw as { id?: number; type: string; data: unknown }; - let validated: ProtocolMessage; - try { - validated = validateResponse( - message.type as keyof Schemas & string, - normalizeMtpMessage(message), - ); - } catch (error) { - log(1, "mtp", "red", "Failed to validate native MTP message", error); - return; - } - for (const handler of subscriptionsRef.current.get(validated.type) ?? - []) { - handler(validated); - } - if (!isPushType(validated.type)) return; - setFreshContacts((contacts) => - removeMissingContacts(contacts, validated), - ); - for (const handler of [...pushHandlersRef.current]) { - void Promise.resolve(handler(validated)).catch((error) => { - log(1, "mtp", "red", "Native MTP push handler failed", error, { - type: validated.type, - }); - }); - } - if (validated.type === "GetStates") { - lastInitialStateRef.current = validated; - } - }, - [lastInitialStateRef, pushHandlersRef], - ); - - useEffect(() => { - if (props.blockConnection) return; - let disposed = false; - let unlisten: UnlistenFn | undefined; - void (async () => { - try { - const nextUnlisten = await listen< - | { kind: "state"; snapshot: NativeSnapshot } - | { kind: "message"; generation: number; message: unknown } - | { - kind: "log"; - level: number; - message: string; - details?: unknown; - } - >("mtp://event", ({ payload }) => { - if (disposed) return; - if (payload.kind === "state") { - applySnapshot(payload.snapshot); - return; - } - if (payload.kind === "message") { - if (payload.generation === generationRef.current) { - dispatchMessage(payload.message); - } - return; - } - log( - payload.level, - "android", - "orange", - payload.message, - payload.details, - ); - }); - if (disposed) nextUnlisten(); - else unlisten = nextUnlisten; - } catch (error) { - log(0, "mtp", "red", "Failed to subscribe to native MTP events", error); - } - - try { - const current = await invoke("mtp_status"); - if (!disposed) applySnapshot(current); - } catch (error) { - log(0, "mtp", "red", "Failed to load native MTP status", error); - } - })(); - return () => { - disposed = true; - unlisten?.(); - }; - }, [applySnapshot, dispatchMessage, props.blockConnection]); - - useEffect(() => { - if (props.blockConnection) return; - const updateVisibility = () => { - void invoke("mtp_set_ui_visible", { - visible: document.visibilityState === "visible" && document.hasFocus(), - }); - }; - updateVisibility(); - document.addEventListener("visibilitychange", updateVisibility); - window.addEventListener("focus", updateVisibility); - window.addEventListener("blur", updateVisibility); - return () => { - document.removeEventListener("visibilitychange", updateVisibility); - window.removeEventListener("focus", updateVisibility); - window.removeEventListener("blur", updateVisibility); - void invoke("mtp_set_ui_visible", { visible: false }); - }; - }, [props.blockConnection]); - - const send = useCallback( - async (type, data) => { - const response = await invoke("mtp_request", { - typeName: type, - data: data ?? {}, - }); - const validated = validateResponse(type, normalizeMtpMessage(response)); - setFreshContacts((contacts) => - removeMissingContacts(contacts, validated), - ); - if (validated.type.startsWith("Error")) { - const errorData = validated.data as Record; - throw new ProtocolError({ - type: validated.type, - requestId: validated.id, - errorType: - typeof errorData.ErrorType === "string" - ? errorData.ErrorType - : undefined, - }); - } - for (const interceptor of interceptorsRef.current) { - void Promise.resolve( - interceptor({ type, data, response: validated as ProtocolMessage }), - ).catch((error) => { - log(1, "mtp", "yellow", "MTP interceptor failed", error, { type }); - }); - } - return validated; - }, - [interceptorsRef], - ); - - const subscribe = useCallback((type, handler) => { - const handlers = - subscriptionsRef.current.get(type) ?? - new Set<(message: ProtocolMessage) => void>(); - handlers.add(handler as (message: ProtocolMessage) => void); - subscriptionsRef.current.set(type, handlers); - return () => { - handlers.delete(handler as (message: ProtocolMessage) => void); - if (handlers.size === 0) subscriptionsRef.current.delete(type); - }; - }, []); - const connected = snapshot.readyState === ConnectionState.Connected; - const contextReady = connected && snapshot.identified; - - return ( - - {props.children} - - ); -} +import { BrowserProvider } from "./browser"; +import { MTPContext, type MTPContextType } from "./mtpContext"; +import { TauriProvider } from "./tauri"; export function Provider(props: { children: ReactNode; blockConnection?: boolean; +}) { + if (isTauri()) return ; + return ; +} + +function BrowserWasmProvider(props: { + children: ReactNode; + blockConnection?: boolean; }) { const [wasmReady, setWasmReady] = useState(false); const [wasmError, setWasmError] = useState(); - useEffect(() => { let active = true; void MTPClient.init().then( - () => { - if (active) setWasmReady(true); - }, - (error: unknown) => { - if (active) setWasmError(() => error); - }, + () => active && setWasmReady(true), + (error: unknown) => active && setWasmError(() => error), ); return () => { active = false; }; }, []); - if (wasmError) throw wasmError; - if (!wasmReady) return null; - - return isTauri() ? ( - - ) : ( - - ); + return wasmReady ? : null; } -export function useMTP(): ContextType { +export function useMTP(): MTPContextType { const context = useContext(MTPContext); - if (!context) { - throw new Error("useMTP must be used within an MTPProvider"); - } + if (!context) throw new Error("useMTP must be used within an MTPProvider"); return context; } diff --git a/packages/mtp/src/index.ts b/packages/mtp/src/index.ts index 19f5acd..a4f05c8 100644 --- a/packages/mtp/src/index.ts +++ b/packages/mtp/src/index.ts @@ -1,9 +1,7 @@ export { Provider, useMTP } from "./context"; -export { RequestIdAllocator } from "./requestIds"; export type { BoundSendFn, MTPExchange, MTPInterceptor, - PushHandler, ProtocolMessage, -} from "./context"; +} from "./mtpContext"; diff --git a/packages/mtp/src/mtpContext.tsx b/packages/mtp/src/mtpContext.tsx new file mode 100644 index 0000000..1f1654d --- /dev/null +++ b/packages/mtp/src/mtpContext.tsx @@ -0,0 +1,159 @@ +import { createContext, useCallback, useRef } from "react"; +import type { + MTPRequestFunction, + MTPResponseFrame, + MTPSubscriptionFunction, +} from "mtp"; +import { + mtp as mtpSchemas, + type Calls, + type Communities, + type Contacts, +} from "@tensamin/shared/data"; +import { log } from "@tensamin/shared/log"; + +export type ProtocolMessage< + Type extends keyof typeof mtpSchemas & string = keyof typeof mtpSchemas & + string, +> = MTPResponseFrame; + +export type BoundSendFn = MTPRequestFunction; + +export type MTPExchange = { + type: keyof typeof mtpSchemas & string; + data: unknown; + response: ProtocolMessage; +}; + +export type MTPInterceptor = (exchange: MTPExchange) => void | Promise; + +export type MTPContextType = { + send: BoundSendFn; + subscribe: MTPSubscriptionFunction; + addInterceptor: (interceptor: MTPInterceptor) => () => void; + readyState: number; + identified: boolean; + freshContacts: Contacts; + freshCommunities: Communities; + freshCalls: Calls; + contextReady: boolean; + loadingDescription: string; +}; + +export const MTPContext = createContext(undefined); + +export function removeMissingContacts( + contacts: Contacts, + message: ProtocolMessage<"GetStates">, +): Contacts { + const missing = new Set(message.data.MissingUserIds ?? []); + return contacts.filter((contact) => !missing.has(contact.UserId)); +} + +export function useMessageHandlers() { + const interceptorsRef = useRef(new Set()); + const subscriptionHandlersRef = useRef( + new Map void | Promise>>(), + ); + const transportRef = useRef<{ + subscribe: MTPSubscriptionFunction; + } | null>(null); + const transportGenerationRef = useRef(0); + const transportUnsubscribersRef = useRef(new Map void>()); + const lastInitialStateRef = useRef | null>(null); + + const attachType = useCallback( + (type: Type) => { + const transport = transportRef.current; + if (!transport || transportUnsubscribersRef.current.has(type)) return; + const generation = transportGenerationRef.current; + const unsubscribe = transport.subscribe(type, (message) => { + if ( + transportRef.current !== transport || + transportGenerationRef.current !== generation + ) + return; + if (type === "GetStates") { + lastInitialStateRef.current = message as ProtocolMessage<"GetStates">; + } + for (const handler of [ + ...(subscriptionHandlersRef.current.get(type) ?? []), + ]) { + void Promise.resolve(handler(message as ProtocolMessage)).catch( + (error) => { + log(1, "mtp", "red", "Subscription handler failed", error, { + type, + }); + }, + ); + } + }); + transportUnsubscribersRef.current.set(type, unsubscribe); + }, + [], + ); + + const attachSubscriptions = useCallback( + (transport: { subscribe: MTPSubscriptionFunction }) => { + for (const unsubscribe of transportUnsubscribersRef.current.values()) { + unsubscribe(); + } + transportUnsubscribersRef.current.clear(); + transportRef.current = transport; + const generation = ++transportGenerationRef.current; + for (const type of subscriptionHandlersRef.current.keys()) { + attachType(type as keyof typeof mtpSchemas & string); + } + return () => { + if ( + transportRef.current !== transport || + transportGenerationRef.current !== generation + ) + return; + transportRef.current = null; + transportGenerationRef.current += 1; + for (const unsubscribe of transportUnsubscribersRef.current.values()) { + unsubscribe(); + } + transportUnsubscribersRef.current.clear(); + }; + }, + [attachType], + ); + + const subscribe = useCallback>( + (type, handler) => { + const handlers = subscriptionHandlersRef.current.get(type) ?? new Set(); + const untypedHandler = handler as ( + message: ProtocolMessage, + ) => void | Promise; + handlers.add(untypedHandler); + subscriptionHandlersRef.current.set(type, handlers); + attachType(type); + const initialState = lastInitialStateRef.current; + if (type === "GetStates" && initialState) { + void Promise.resolve(untypedHandler(initialState)).catch( + () => undefined, + ); + } + return () => { + handlers.delete(untypedHandler); + if (handlers.size !== 0) return; + subscriptionHandlersRef.current.delete(type); + transportUnsubscribersRef.current.get(type)?.(); + transportUnsubscribersRef.current.delete(type); + }; + }, + [attachType], + ); + const addInterceptor = useCallback((interceptor: MTPInterceptor) => { + interceptorsRef.current.add(interceptor); + return () => interceptorsRef.current.delete(interceptor); + }, []); + return { + addInterceptor, + attachSubscriptions, + interceptorsRef, + subscribe, + }; +} diff --git a/packages/mtp/src/protocolFields.test.ts b/packages/mtp/src/protocolFields.test.ts deleted file mode 100644 index ef0c18f..0000000 --- a/packages/mtp/src/protocolFields.test.ts +++ /dev/null @@ -1,44 +0,0 @@ -import { describe, expect, it } from "vitest"; - -import { fromWireData, toWireData } from "./protocolFields"; - -describe("MTP protocol field translation", () => { - it("maps application message fields to MTP wire fields", () => { - expect( - toWireData({ - Messages: [{ Content: "abc", MessageId: 4 }], - }), - ).toEqual({ - Messages: [{ AppContent: "abc", AppMessageId: 4 }], - }); - }); - - it("maps MTP wire fields back to application fields", () => { - expect( - fromWireData({ - AppCreatedAt: 123, - Message: { AppContent: "abc", AppMessageId: 4 }, - }), - ).toEqual({ - CreatedAt: 123, - Message: { Content: "abc", MessageId: 4 }, - }); - }); - - it("preserves byte arrays and unrelated fields", () => { - const bytes = new Uint8Array([1, 2, 3]); - const translated = toWireData({ Payload: bytes, Other: "value" }) as { - Payload: Uint8Array; - Other: string; - }; - - expect(translated.Payload).toBe(bytes); - expect(translated.Other).toBe("value"); - }); - - it("rejects field mapping collisions", () => { - expect(() => toWireData({ Content: "a", AppContent: "b" })).toThrow( - "MTP field translation collision", - ); - }); -}); diff --git a/packages/mtp/src/protocolFields.ts b/packages/mtp/src/protocolFields.ts deleted file mode 100644 index 3ec3d5d..0000000 --- a/packages/mtp/src/protocolFields.ts +++ /dev/null @@ -1,58 +0,0 @@ -const APPLICATION_TO_WIRE_FIELDS = { - Content: "AppContent", - CreatedAt: "AppCreatedAt", - MessageId: "AppMessageId", -} as const; - -const WIRE_TO_APPLICATION_FIELDS = { - AppContent: "Content", - AppCreatedAt: "CreatedAt", - AppMessageId: "MessageId", -} as const; - -function mapProtocolFields( - value: unknown, - fieldMap: Readonly>, -): unknown { - if ( - value === null || - typeof value !== "object" || - value instanceof Uint8Array - ) { - return value; - } - - if (Array.isArray(value)) { - return value.map((item) => mapProtocolFields(item, fieldMap)); - } - - const source = value as Record; - const target: Record = {}; - - for (const [key, child] of Object.entries(source)) { - const mappedKey = fieldMap[key] ?? key; - - if (mappedKey in target) { - throw new Error(`MTP field translation collision for ${mappedKey}`); - } - - target[mappedKey] = mapProtocolFields(child, fieldMap); - } - - return target; -} - -export function toWireData(value: unknown): unknown { - return mapProtocolFields(value, APPLICATION_TO_WIRE_FIELDS); -} - -export function fromWireData(value: unknown): unknown { - return mapProtocolFields(value, WIRE_TO_APPLICATION_FIELDS); -} - -export function fromWireMessage(message: T): T { - return { - ...message, - data: fromWireData(message.data), - }; -} diff --git a/packages/mtp/src/requestIds.test.ts b/packages/mtp/src/requestIds.test.ts deleted file mode 100644 index 82fd508..0000000 --- a/packages/mtp/src/requestIds.test.ts +++ /dev/null @@ -1,25 +0,0 @@ -import { describe, expect, it } from "vitest"; - -import { RequestIdAllocator } from "./requestIds"; - -describe("MTP request ID allocation", () => { - it("allocates nonzero request IDs monotonically", () => { - const ids = new RequestIdAllocator(); - - expect(ids.allocate()).toBe(1); - expect(ids.allocate()).toBe(2); - expect(new RequestIdAllocator().allocate()).toBe(1); - }); - - it("does not wrap exhausted request IDs", () => { - const ids = new RequestIdAllocator(0x1_0000_0000); - - expect(() => ids.allocate()).toThrow("MTP request ID space exhausted"); - }); - - it("rejects invalid allocator states", () => { - expect(() => new RequestIdAllocator(0)).toThrow( - "invalid MTP request ID allocator state", - ); - }); -}); diff --git a/packages/mtp/src/requestIds.ts b/packages/mtp/src/requestIds.ts deleted file mode 100644 index 5581e8a..0000000 --- a/packages/mtp/src/requestIds.ts +++ /dev/null @@ -1,24 +0,0 @@ -const MAX_MTP_REQUEST_ID = 0xffff_ffff; - -export class RequestIdAllocator { - #next: number; - - constructor(next = 1) { - if ( - !Number.isSafeInteger(next) || - next <= 0 || - next > MAX_MTP_REQUEST_ID + 1 - ) { - throw new RangeError("invalid MTP request ID allocator state"); - } - this.#next = next; - } - - allocate(): number { - if (this.#next > MAX_MTP_REQUEST_ID) { - throw new Error("MTP request ID space exhausted for this connection"); - } - - return this.#next++; - } -} diff --git a/packages/mtp/src/tauri.tsx b/packages/mtp/src/tauri.tsx new file mode 100644 index 0000000..92ac209 --- /dev/null +++ b/packages/mtp/src/tauri.tsx @@ -0,0 +1,258 @@ +import { + type ReactNode, + useCallback, + useEffect, + useRef, + useState, +} from "react"; +import { invoke } from "@tauri-apps/api/core"; +import { listen, type UnlistenFn } from "@tauri-apps/api/event"; +import { + ConnectionState, + MTPProxyConnection, + type MTPFrame, + type MTPProxyAdapter, +} from "mtp"; +import { + mtp as mtpSchemas, + type Calls, + type Communities, + type Contacts, +} from "@tensamin/shared/data"; +import { log } from "@tensamin/shared/log"; + +import { + type BoundSendFn, + MTPContext, + type ProtocolMessage, + removeMissingContacts, + useMessageHandlers, +} from "./mtpContext"; + +type NativeSnapshot = { + generation: number; + readyState: number; + identified: boolean; + state?: unknown; + error?: string; +}; + +function createTauriAdapter() { + const subscriptions = new Map void>>(); + const adapter: MTPProxyAdapter = { + request: (type, data) => + invoke("mtp_request", { typeName: type, data }), + subscribe(type, handler) { + const handlers = subscriptions.get(type) ?? new Set(); + handlers.add(handler); + subscriptions.set(type, handlers); + return () => { + handlers.delete(handler); + if (handlers.size === 0) subscriptions.delete(type); + }; + }, + }; + return { + adapter, + dispatch(message: MTPFrame) { + for (const handler of subscriptions.get(message.type) ?? []) + handler(message); + }, + }; +} + +export function TauriProvider(props: { + children: ReactNode; + blockConnection?: boolean; +}) { + const [snapshot, setSnapshot] = useState({ + generation: 0, + readyState: ConnectionState.Disconnected, + identified: false, + }); + const [freshContacts, setFreshContacts] = useState([]); + const [freshCommunities, setFreshCommunities] = useState([]); + const [freshCalls, setFreshCalls] = useState([]); + const generationRef = useRef(0); + const { addInterceptor, attachSubscriptions, interceptorsRef, subscribe } = + useMessageHandlers(); + const [{ bridge, connection }] = useState(() => { + const bridge = createTauriAdapter(); + return { + bridge, + connection: new MTPProxyConnection(bridge.adapter, { + schemas: mtpSchemas, + throwProtocolErrors: true, + onValidationError: (error) => { + log(1, "mtp", "red", "Failed to validate native MTP message", error); + }, + }), + }; + }); + + const applySnapshot = useCallback(async (next: NativeSnapshot) => { + if (next.generation < generationRef.current) return; + generationRef.current = next.generation; + if (next.error) + log(0, "android", "orange", "MTP connection failed", next.error); + if (!next.identified) { + setSnapshot(next); + return; + } + if (next.state === undefined) { + setSnapshot({ + ...next, + identified: false, + error: "Native MTP connection omitted initial state", + }); + return; + } + try { + const state = await mtpSchemas.ClientStateSync.response.parseAsync( + next.state, + ); + setFreshContacts(state.Contacts); + setFreshCommunities(state.Communities); + setFreshCalls(state.Calls); + setSnapshot(next); + } catch (error) { + log(0, "mtp", "red", "Invalid native MTP state", error); + setSnapshot({ + ...next, + identified: false, + error: "Invalid ClientStateSync payload", + }); + } + }, []); + + const dispatchMessage = useCallback( + (message: MTPFrame) => { + bridge.dispatch(message); + }, + [bridge], + ); + + useEffect(() => { + return attachSubscriptions(connection); + }, [attachSubscriptions, connection]); + + useEffect( + () => + subscribe("GetStates", (message) => { + setFreshContacts((contacts) => + removeMissingContacts(contacts, message), + ); + }), + [subscribe], + ); + + useEffect(() => { + if (props.blockConnection) return; + let disposed = false; + let unlisten: UnlistenFn | undefined; + void (async () => { + try { + const nextUnlisten = await listen< + | { kind: "state"; snapshot: NativeSnapshot } + | { kind: "message"; generation: number; message: MTPFrame } + | { kind: "log"; level: number; message: string; details?: unknown } + >("mtp://event", ({ payload }) => { + if (disposed) return; + if (payload.kind === "state") { + void applySnapshot(payload.snapshot); + } else if (payload.kind === "message") { + if (payload.generation === generationRef.current) { + dispatchMessage(payload.message); + } + } else { + log( + payload.level, + "android", + "orange", + payload.message, + payload.details, + ); + } + }); + if (disposed) nextUnlisten(); + else unlisten = nextUnlisten; + } catch (error) { + log(0, "mtp", "red", "Failed to subscribe to native MTP events", error); + } + try { + const current = await invoke("mtp_status"); + if (!disposed) await applySnapshot(current); + } catch (error) { + log(0, "mtp", "red", "Failed to load native MTP status", error); + } + })(); + return () => { + disposed = true; + unlisten?.(); + }; + }, [applySnapshot, dispatchMessage, props.blockConnection]); + + useEffect(() => { + if (props.blockConnection) return; + const updateVisibility = () => { + void invoke("mtp_set_ui_visible", { + visible: document.visibilityState === "visible" && document.hasFocus(), + }); + }; + updateVisibility(); + document.addEventListener("visibilitychange", updateVisibility); + window.addEventListener("focus", updateVisibility); + window.addEventListener("blur", updateVisibility); + return () => { + document.removeEventListener("visibilitychange", updateVisibility); + window.removeEventListener("focus", updateVisibility); + window.removeEventListener("blur", updateVisibility); + void invoke("mtp_set_ui_visible", { visible: false }); + }; + }, [props.blockConnection]); + + const send = useCallback( + async (type, data, options) => { + const response = await connection.request(type, data, options); + if (response.type === "GetStates") { + setFreshContacts((contacts) => + removeMissingContacts( + contacts, + response as ProtocolMessage<"GetStates">, + ), + ); + } + for (const interceptor of interceptorsRef.current) { + void Promise.resolve(interceptor({ type, data, response })).catch( + (error) => { + log(1, "mtp", "yellow", "MTP interceptor failed", error, { type }); + }, + ); + } + return response; + }, + [connection, interceptorsRef], + ); + const connected = snapshot.readyState === ConnectionState.Connected; + + return ( + + {props.children} + + ); +} diff --git a/packages/notifications/src/context.tsx b/packages/notifications/src/context.tsx index 7a8b32a..e0da277 100644 --- a/packages/notifications/src/context.tsx +++ b/packages/notifications/src/context.tsx @@ -16,7 +16,6 @@ import { useLocation, useNavigate } from "@tanstack/react-router"; import { decryptChatText } from "@tensamin/crypto/chatSecret"; import { log } from "@tensamin/shared/log"; import { playSound } from "@tensamin/shared/sounds"; -import { type RawMessage } from "@tensamin/chat/values"; export const context = createContext(undefined); @@ -30,7 +29,7 @@ async function requestNotificationPermission() { } export default function Provider(props: { children: React.ReactNode }) { - const { subscribePush, send } = useMTP(); + const { subscribe, send } = useMTP(); const { load } = useStorage(); const { get } = useUser(); const { addLiveMessage, chatSecret, getChatSecret, userId } = useChat(); @@ -39,148 +38,140 @@ export default function Provider(props: { children: React.ReactNode }) { const location = useLocation(); useEffect(() => { - return subscribePush(async (message) => { - if (message.type === "MessageLive") { - const data = message.data as { - Message?: RawMessage; - SenderId?: number; - }; + return subscribe("MessageLive", async ({ data }) => { + if (!data.SenderId) return; - if (!data.SenderId) return; + const isCurrentChat = + location.pathname === "/chat" && userId === data.SenderId; + const appFocused = + document.hasFocus() && document.visibilityState === "visible"; + const shouldAlert = !isCurrentChat || !appFocused; - const isCurrentChat = - location.pathname === "/chat" && userId === data.SenderId; - const appFocused = - document.hasFocus() && document.visibilityState === "visible"; - const shouldAlert = !isCurrentChat || !appFocused; + const messageSecret = + isCurrentChat && chatSecret + ? chatSecret + : await getChatSecret(data.SenderId); - const messageSecret = - isCurrentChat && chatSecret - ? chatSecret - : await getChatSecret(data.SenderId); + if (!data.Message || !messageSecret) return; + if (shouldAlert) playSound("message"); - if (!data.Message || !messageSecret) return; - if (shouldAlert) playSound("message"); + void decryptChatText(messageSecret, data.Message.Content) + .catch((err) => { + log(1, "chat", "red", "Failed to decrypt live message", err, { + SendTime: data.Message?.SendTime, + }); + return null; + }) + .then(async (content) => { + if (!data.Message || !content || !data.SenderId) return; - void decryptChatText(messageSecret, data.Message.Content) - .catch((err) => { - log(1, "chat", "red", "Failed to decrypt live message", err, { - SendTime: data.Message?.SendTime, + if (isCurrentChat) { + addLiveMessage({ + ...data.Message, + Content: content ?? "Failed to decrypt message", + decryptionFailed: content === null, }); - return null; - }) - .then(async (content) => { - if (!data.Message || !content || !data.SenderId) return; + } - if (isCurrentChat) { - addLiveMessage({ - ...data.Message, - Content: content ?? "Failed to decrypt message", - decryptionFailed: content === null, + if (!shouldAlert) return; + + if (!isCurrentChat) { + // todo: add notification symbol to conversation cards (incl. message start) + moveUserIdToTop(data.SenderId); + + if (await load("settings.receive_confirmations")) { + void send("MessageState", { + MessageState: "received", }); } + } - if (!shouldAlert) return; + const user = await get(data.SenderId, [ + "UserId", + "Display", + "Avatar", + ]); - if (!isCurrentChat) { - // todo: add notification symbol to conversation cards (incl. message start) - moveUserIdToTop(data.SenderId); + if (isTauri()) { + if (!appFocused) return; + const permissionGranted = + (await isTauriNotificationPermissionGranted()) || + (await requestTauriNotificationPermission()) === "granted"; - if (await load("settings.receive_confirmations")) { - void send("MessageState", { - MessageState: "received", - }); - } - } - - const user = await get(data.SenderId, [ - "UserId", - "Display", - "Avatar", - ]); - - if (isTauri()) { - if (!appFocused) return; - const permissionGranted = - (await isTauriNotificationPermissionGranted()) || - (await requestTauriNotificationPermission()) === "granted"; - - if (permissionGranted) { - let handledNatively = false; - try { - handledNatively = await invoke( - "mtp_post_message_notification", - { - senderId: user.UserId, - sender: user.Display, - body: content, - avatar: user.Avatar, - }, - ); - } catch (error) { - log( - 1, - "notifications", - "red", - "Failed to create native message notification", - error, - ); - } - - if (!handledNatively) { - sendTauriNotification({ title: user.Display, body: content }); - } - } - } else { - const hasPermissions = await requestNotificationPermission(); - - if (hasPermissions) { - const options: NotificationOptions = { - body: content, - icon: user.Avatar || "/icons/icon-192.png", - badge: "/icons/notification-badge.png", - tag: `message-${user.UserId}`, - silent: true, - }; - if ("serviceWorker" in navigator) { - const registration = - await navigator.serviceWorker.getRegistration(); - if (registration) { - await registration.showNotification(user.Display, { - ...options, - data: { url: `/chat?id=${user.UserId}` }, - }); - return; - } - } - const notification = new Notification(user.Display, options); - notification.onclick = () => { - window.focus(); - navigate({ - to: `/chat?id=${user.UserId}`, - }); - notification.close(); - }; - } else { - sonnerToast(user.Display, { - classNames: { - content: "pl-4", + if (permissionGranted) { + let handledNatively = false; + try { + handledNatively = await invoke( + "mtp_post_message_notification", + { + senderId: user.UserId, + sender: user.Display, + body: content, + avatar: user.Avatar, }, - description: content, - icon: ( - - - - {user.Display.slice(0, 2).toUpperCase()} - - - ), - }); + ); + } catch (error) { + log( + 1, + "notifications", + "red", + "Failed to create native message notification", + error, + ); + } + + if (!handledNatively) { + sendTauriNotification({ title: user.Display, body: content }); } } - }); - return; - } + } else { + const hasPermissions = await requestNotificationPermission(); + + if (hasPermissions) { + const options: NotificationOptions = { + body: content, + icon: user.Avatar || "/icons/icon-192.png", + badge: "/icons/notification-badge.png", + tag: `message-${user.UserId}`, + silent: true, + }; + if ("serviceWorker" in navigator) { + const registration = + await navigator.serviceWorker.getRegistration(); + if (registration) { + await registration.showNotification(user.Display, { + ...options, + data: { url: `/chat?id=${user.UserId}` }, + }); + return; + } + } + const notification = new Notification(user.Display, options); + notification.onclick = () => { + window.focus(); + navigate({ + to: `/chat?id=${user.UserId}`, + }); + notification.close(); + }; + } else { + sonnerToast(user.Display, { + classNames: { + content: "pl-4", + }, + description: content, + icon: ( + + + + {user.Display.slice(0, 2).toUpperCase()} + + + ), + }); + } + } + }); }); }, [ addLiveMessage, @@ -191,7 +182,7 @@ export default function Provider(props: { children: React.ReactNode }) { navigate, send, moveUserIdToTop, - subscribePush, + subscribe, getChatSecret, userId, ]); diff --git a/packages/user/src/context.tsx b/packages/user/src/context.tsx index 423661c..8c41aa4 100644 --- a/packages/user/src/context.tsx +++ b/packages/user/src/context.tsx @@ -9,13 +9,12 @@ import { useState, useSyncExternalStore, } from "react"; -import { useMTP, type ProtocolMessage } from "@tensamin/mtp"; +import { useMTP } from "@tensamin/mtp"; import { clientUserStateSchema, mtp as schemas, publicUserStateSchema, - userStateEntrySchema, } from "@tensamin/shared/data"; import type z from "zod"; import { createCache } from "@tensamin/cache"; @@ -89,7 +88,7 @@ export default function UserProvider(props: { children: ReactNode }) { ); const revisionsRef = useRef(new Map>()); - const { send, subscribePush } = useMTP(); + const { send, subscribe: subscribeMTP } = useMTP(); const { load } = useStorage(); const { contacts } = useSession(); const [accountId, setAccountId] = useState(null); @@ -174,39 +173,6 @@ export default function UserProvider(props: { children: ReactNode }) { [publishUser], ); - const handleStatePush = useCallback( - async (message: ProtocolMessage) => { - if (!accountId) return; - const data = message.data as Record; - if (message.type === "GetStates") { - if (Array.isArray(data.MissingUserIds)) { - for (const userId of data.MissingUserIds) { - if (typeof userId === "number") removePresence(userId); - } - } - if (!Array.isArray(data.UserStates)) return; - for (const entry of data.UserStates) { - const parsed = userStateEntrySchema.safeParse(entry); - if (!parsed.success) continue; - if (parsed.data.UserId === accountId) continue; - initialStatesRef.current.set( - parsed.data.UserId, - parsed.data.UserState, - ); - if (applyUserState(parsed.data.UserId, parsed.data.UserState)) { - initialStatesRef.current.delete(parsed.data.UserId); - } - } - return; - } - if (message.type !== "ClientChanged") return; - const parsed = schemas.ClientChanged.response.safeParse(data); - if (!parsed.success) return; - applyUserState(parsed.data.UserId, parsed.data.UserState, true); - }, - [accountId, applyUserState, removePresence], - ); - useEffect(() => { void load("user_id").then((accountId) => { accountIdRef.current = accountId; @@ -223,8 +189,24 @@ export default function UserProvider(props: { children: ReactNode }) { useEffect(() => { if (!accountId) return; - return subscribePush(handleStatePush); - }, [accountId, handleStatePush, subscribePush]); + const unsubscribeStates = subscribeMTP("GetStates", ({ data }) => { + for (const userId of data.MissingUserIds ?? []) removePresence(userId); + for (const entry of data.UserStates) { + if (!entry || entry.UserId === accountId) continue; + initialStatesRef.current.set(entry.UserId, entry.UserState); + if (applyUserState(entry.UserId, entry.UserState)) { + initialStatesRef.current.delete(entry.UserId); + } + } + }); + const unsubscribeChanged = subscribeMTP("ClientChanged", ({ data }) => { + applyUserState(data.UserId, data.UserState, true); + }); + return () => { + unsubscribeStates(); + unsubscribeChanged(); + }; + }, [accountId, applyUserState, removePresence, subscribeMTP]); const loadUser = useCallback( async (userId: number): Promise => { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 107149b..83d88ff 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -6,7 +6,7 @@ settings: overrides: '@methanium/ui': https://git.methanium.net/methanium/ui/releases/download/0.0.29/methanium-ui.tgz - mtp: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz + mtp: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz importers: @@ -16,8 +16,8 @@ importers: specifier: https://git.methanium.net/methanium/ui/releases/download/0.0.29/methanium-ui.tgz version: https://git.methanium.net/methanium/ui/releases/download/0.0.29/methanium-ui.tgz(@types/react-dom@19.2.4(@types/react@19.2.18))(@types/react@19.2.18)(emojibase@17.0.0)(react-dom@19.2.8(react@19.2.8))(react-is@19.2.8)(react@19.2.8)(redux@5.0.1)(supports-color@7.2.0)(typescript@6.0.3) mtp: - specifier: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz - version: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz + specifier: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz + version: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz sonner: specifier: ^2.0.8 version: 2.0.8(@types/react@19.2.18)(react-dom@19.2.8(react@19.2.8))(react@19.2.8) @@ -98,8 +98,8 @@ importers: specifier: workspace:* version: link:../../packages/storage mtp: - specifier: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz - version: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz + specifier: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz + version: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz react: specifier: ^19.2.8 version: 19.2.8 @@ -290,8 +290,8 @@ importers: specifier: ^17.9.0 version: 17.9.0 mtp: - specifier: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz - version: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz + specifier: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz + version: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz typescript: specifier: ~6.0.3 version: 6.0.3 @@ -356,8 +356,8 @@ importers: specifier: ^1.29.0 version: 1.30.0(react@19.2.8) mtp: - specifier: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz - version: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz + specifier: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz + version: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz react: specifier: ^19.2.8 version: 19.2.8 @@ -431,8 +431,8 @@ importers: packages/crypto: dependencies: mtp: - specifier: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz - version: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz + specifier: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz + version: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz react: specifier: ^19.2.8 version: 19.2.8 @@ -506,9 +506,6 @@ importers: '@tauri-apps/api': specifier: ^2.11.1 version: 2.11.1 - '@tensamin/crypto': - specifier: workspace:* - version: link:../crypto '@tensamin/shared': specifier: workspace:* version: link:../shared @@ -516,17 +513,11 @@ importers: specifier: workspace:* version: link:../storage mtp: - specifier: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz - version: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz + specifier: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz + version: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz react: specifier: ^19.2.8 version: 19.2.8 - react-dom: - specifier: ^19.2.8 - version: 19.2.8(react@19.2.8) - zod: - specifier: ^4.4.3 - version: 4.4.3 devDependencies: eslint: specifier: ^10.8.0 @@ -5085,8 +5076,8 @@ packages: ms@2.1.3: resolution: {integrity: sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==} - mtp@https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz: - resolution: {integrity: sha512-rI+xskgAp93o9EdBO/b7HnA0PZTQQPTYuF4Snon/hTvayN+x9O40oad4n/ZUF0ahS6grw00MPKPe89Mfe3BHnw==, tarball: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz} + mtp@https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz: + resolution: {integrity: sha512-MzqeWSaS2lVoiK0coNfY8EPgGNk7QYFS3eRqVYeYBCT0NNm2lQ7T8lLjBuIuMZ6Zh2KAVWhwI2xQIkMdZ7QThA==, tarball: https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz} version: 0.3.0 nanoid@3.3.18: @@ -11337,7 +11328,7 @@ snapshots: ms@2.1.3: {} - mtp@https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz: + mtp@https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz: dependencies: yaml: 2.9.0 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index f713a2d..6ec3531 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -7,4 +7,4 @@ allowBuilds: esbuild: true overrides: "@methanium/ui": "https://git.methanium.net/methanium/ui/releases/download/0.0.29/methanium-ui.tgz" - mtp: "https://git.methanium.net/methanium/mtp/releases/download/0.3.0-b331b9f6a3/mtp-0.3.0.tgz" + mtp: "https://git.methanium.net/methanium/mtp/releases/download/0.3.0-dev-c7c7afe/mtp-0.3.0.tgz"