diff --git a/Cargo.lock b/Cargo.lock index 21d0bba..bf277ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14,18 +14,18 @@ dependencies = [ [[package]] name = "aho-corasick" -version = "1.1.5" +version = "1.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba" +checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" dependencies = [ "memchr", ] [[package]] name = "android_system_properties" -version = "0.1.6" +version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ae221649c9976a6f6c56ae1facf410f3ddb33cc661c4b7b61020a912d4237fbc" +checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311" dependencies = [ "libc", ] @@ -109,9 +109,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" [[package]] name = "aws-lc-rs" -version = "1.18.0" +version = "1.17.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ce2b2dcc879c3bae0d371e77c99f2238400ef24ec001394befa67b6e543add9e" +checksum = "00bdb5da18dac48ca2cc7cd4a98e533e8635a58e2361d13a1a4ee3888e0d72f1" dependencies = [ "aws-lc-sys", "untrusted 0.7.1", @@ -120,9 +120,9 @@ dependencies = [ [[package]] name = "aws-lc-sys" -version = "0.44.0" +version = "0.43.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f09fae7be8bb3174e05c6afdb34199e6dc0c7c04ba9fa237b1967adfbde27483" +checksum = "43103168cc76fe62678a375e722fc9cb3a0146159ac5828bc4f0dfd755c2224c" dependencies = [ "cc", "cmake", @@ -143,12 +143,6 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" -[[package]] -name = "base64" -version = "0.23.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" - [[package]] name = "base64ct" version = "1.8.3" @@ -217,9 +211,9 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "cc" -version = "1.4.1" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9066c49992464636f92905fa096ec58baaa4d57ec19a5c096c68d3e25ef3d136" +checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9" dependencies = [ "find-msvc-tools", "jobserver", @@ -325,6 +319,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "const-oid" +version = "0.9.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" + [[package]] name = "const-oid" version = "0.10.2" @@ -411,23 +411,8 @@ dependencies = [ "cfg-if", "cpufeatures 0.2.17", "curve25519-dalek-derive", - "fiat-crypto 0.2.9", - "rustc_version", - "subtle", - "zeroize", -] - -[[package]] -name = "curve25519-dalek" -version = "5.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b5eed333089e2e1c1ac8c6c0398e5e2497b4c9926ca6d0365ed1e099afa5bc23" -dependencies = [ - "cfg-if", - "cpufeatures 0.3.0", - "curve25519-dalek-derive", - "digest 0.11.3", - "fiat-crypto 0.3.0", + "digest 0.10.7", + "fiat-crypto", "rustc_version", "subtle", "zeroize", @@ -460,9 +445,20 @@ dependencies = [ [[package]] name = "data-encoding" -version = "2.11.1" +version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" +checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" + +[[package]] +name = "der" +version = "0.7.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" +dependencies = [ + "const-oid 0.9.6", + "pem-rfc7468", + "zeroize", +] [[package]] name = "der" @@ -470,8 +466,7 @@ version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a69dedd701da44b0536442edf09c81a64b0ab97a7a4a5e3d1971f00027cbc63d" dependencies = [ - "const-oid", - "pem-rfc7468", + "const-oid 0.10.2", "zeroize", ] @@ -528,7 +523,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" dependencies = [ "block-buffer 0.12.1", - "const-oid", + "const-oid 0.10.2", "crypto-common 0.2.2", "ctutils", ] @@ -568,25 +563,24 @@ checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" [[package]] name = "ed25519" -version = "3.0.0" +version = "2.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "29fcf32e6c73d1079f83ab4d782de2d81620346a5f38c6237a86a22f8368980a" +checksum = "115531babc129696a58c64a4fef0a8bf9e9698629fb97e9e40767d235cfbcd53" dependencies = [ - "pkcs8", - "signature 3.0.0", + "pkcs8 0.10.2", + "signature 2.2.0", ] [[package]] name = "ed25519-dalek" -version = "3.0.0" +version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ebaa1a2bf1290ab3bfe5a7b771d050ebffab2711c19a81691c683a5144a25de" +checksum = "70e796c081cee67dc755e1a36a0a172b897fab85fc3f6bc48307991f64e4eca9" dependencies = [ - "curve25519-dalek 5.0.0", + "curve25519-dalek", "ed25519", "serde", - "sha2 0.11.0", - "signature 3.0.0", + "sha2 0.10.9", "subtle", "zeroize", ] @@ -622,11 +616,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", -<<<<<<< HEAD - "windows-sys 0.61.2", -======= - "windows-sys 0.52.0", ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 + "windows-sys 0.59.0", ] [[package]] @@ -653,17 +643,11 @@ version = "0.2.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" -[[package]] -name = "fiat-crypto" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "64cd1e32ddd350061ae6edb1b082d7c54915b5c672c389143b9a63403a109f24" - [[package]] name = "find-msvc-tools" -version = "0.1.10" +version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "26b73573e6edcd2af0cdf47bd6cb58f0b3839491263c314eaad1ccf24430e1de" +checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" [[package]] name = "fixedbitset" @@ -969,9 +953,9 @@ checksum = "1a9fcbcc408c5526c3ab80d534e5c86e7967c1fb7aa0a8c76abd1edc27deb877" [[package]] name = "http" -version = "1.5.0" +version = "1.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425" dependencies = [ "bytes", "itoa", @@ -1014,9 +998,9 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hybrid-array" -version = "0.4.14" +version = "0.4.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b" +checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c" dependencies = [ "ctutils", "typenum", @@ -1230,9 +1214,9 @@ dependencies = [ [[package]] name = "ipnet" -version = "2.12.1" +version = "2.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a756c3fac73139e83f14c2d742155dd2b78d3ee56597b419a0579b7bdd6dd78" +checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" [[package]] name = "itertools" @@ -1379,9 +1363,9 @@ dependencies = [ [[package]] name = "keccak" -version = "0.2.1" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ffd9697dc4a9a62e2da93389f34400b77a28f0287711263cabb203b3ccb9c0e4" +checksum = "9e24a010dd405bd7ed803e5253182815b41bf2e6a80cc3bfc066658e03a198aa" dependencies = [ "cfg-if", "cpufeatures 0.3.0", @@ -1468,15 +1452,9 @@ dependencies = [ [[package]] name = "livekit-protocol" -<<<<<<< HEAD -version = "0.7.10" +version = "0.7.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4d26880e94e2f9bab298445e7d86a3794453d211a12ddbd051bd9991a343f9ff" -======= -version = "0.7.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "526f22bddf409e5f15449d55cf341647d7c16a1f43df9db9b05be106e4208e1c" ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 +checksum = "735a237deeff60124edd7831fcabce32b4b9f667d5c4c5688c9d18a2808422c4" dependencies = [ "pbjson", "pbjson-types", @@ -1553,12 +1531,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "add6b9d92e496f16f4526d68ff29da1483aba4b119baeab8bed3b9e3544a6f3d" dependencies = [ - "const-oid", + "const-oid 0.10.2", "crypto-common 0.2.2", "ctutils", "hybrid-array", "module-lattice", - "pkcs8", + "pkcs8 0.11.0", "shake", "signature 3.0.0", ] @@ -1603,7 +1581,7 @@ dependencies = [ [[package]] name = "mtp" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" dependencies = [ "mtp-client", "mtp-codec", @@ -1619,7 +1597,7 @@ dependencies = [ [[package]] name = "mtp-client" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" dependencies = [ "mtp-codec", "mtp-common", @@ -1632,9 +1610,9 @@ dependencies = [ [[package]] name = "mtp-codec" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" dependencies = [ - "base64 0.23.1", + "base64 0.22.1", "byteorder", "mtp-common", "mtp-crypto", @@ -1645,7 +1623,7 @@ dependencies = [ [[package]] name = "mtp-common" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" dependencies = [ "quinn", "rustls", @@ -1656,9 +1634,9 @@ dependencies = [ [[package]] name = "mtp-crypto" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" dependencies = [ - "base64 0.23.1", + "base64 0.22.1", "chacha20poly1305", "ed25519-dalek", "getrandom 0.4.3", @@ -1670,7 +1648,7 @@ dependencies = [ "rustls", "serde", "sha2 0.11.0", - "thiserror 2.0.19", + "thiserror 1.0.69", "tokio", "zeroize", ] @@ -1678,24 +1656,24 @@ dependencies = [ [[package]] name = "mtp-files" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" dependencies = [ "mtp-crypto", "rand 0.10.2", - "thiserror 2.0.19", + "thiserror 1.0.69", "zeroize", ] [[package]] name = "mtp-host" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" dependencies = [ "mtp-codec", "mtp-common", "mtp-crypto", "mtp-transport", - "rand 0.10.2", + "rand 0.8.7", "tokio", "tracing", "wtransport", @@ -1704,7 +1682,7 @@ dependencies = [ [[package]] name = "mtp-transport" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" dependencies = [ "async-trait", "mtp-codec", @@ -1722,7 +1700,7 @@ dependencies = [ [[package]] name = "mtp-type-map" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" dependencies = [ "serde", "serde_yaml", @@ -1731,7 +1709,7 @@ dependencies = [ [[package]] name = "mtp-webserver" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" dependencies = [ "async-trait", "bytes", @@ -2124,9 +2102,9 @@ dependencies = [ [[package]] name = "pem-rfc7468" -version = "1.0.0" +version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a6305423e0e7738146434843d1694d621cce767262b2a86910beab705e4493d9" +checksum = "88b39c9bfcfc231068454382784bb460aae594343fb030d46e9f50a645418412" dependencies = [ "base64ct", ] @@ -2153,14 +2131,24 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pkcs8" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7" +dependencies = [ + "der 0.7.10", + "spki 0.7.3", +] + [[package]] name = "pkcs8" version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "451913da69c775a56034ea8d9003d27ee8948e12443eae7c038ba100a4f21cb7" dependencies = [ - "der", - "spki", + "der 0.8.1", + "spki 0.8.0", ] [[package]] @@ -2339,11 +2327,7 @@ dependencies = [ "once_cell", "socket2", "tracing", -<<<<<<< HEAD - "windows-sys 0.61.2", -======= - "windows-sys 0.52.0", ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 + "windows-sys 0.59.0", ] [[package]] @@ -2490,9 +2474,9 @@ dependencies = [ [[package]] name = "regex-automata" -version = "0.4.18" +version = "0.4.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" +checksum = "8fcfdb36bda0c880c5931cdc7a2bcdc8ba4556847b9d912bca70bc94708711ad" dependencies = [ "aho-corasick", "memchr", @@ -2597,18 +2581,14 @@ dependencies = [ "errno", "libc", "linux-raw-sys", -<<<<<<< HEAD - "windows-sys 0.61.2", -======= - "windows-sys 0.52.0", ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 + "windows-sys 0.59.0", ] [[package]] name = "rustls" -version = "0.23.43" +version = "0.23.42" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138" dependencies = [ "aws-lc-rs", "log", @@ -2660,11 +2640,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", -<<<<<<< HEAD - "windows-sys 0.61.2", -======= - "windows-sys 0.52.0", ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 + "windows-sys 0.59.0", ] [[package]] @@ -2868,7 +2844,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09057cb2149ad4cbd2da1e26b351f9a4c354219421229c69c3063e6f61947c4a" dependencies = [ "digest 0.11.3", - "keccak 0.2.1", + "keccak 0.2.0", "sponge-cursor", ] @@ -2951,6 +2927,16 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "spki" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d" +dependencies = [ + "base64ct", + "der 0.7.10", +] + [[package]] name = "spki" version = "0.8.0" @@ -2958,7 +2944,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d9efca8738c78ee9484207732f728b1ef517bbb1833d6fc0879ca898a522f6f" dependencies = [ "base64ct", - "der", + "der 0.8.1", ] [[package]] @@ -3046,14 +3032,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix", -<<<<<<< HEAD - "windows-sys 0.61.2", -======= - "windows-sys 0.52.0", ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 + "windows-sys 0.59.0", ] [[package]] @@ -3098,9 +3080,9 @@ dependencies = [ [[package]] name = "time" -version = "0.3.55" +version = "0.3.54" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cdb87b95ec50ddfa440816d227a17b2ccbdda963a316a727fda0fc4334f7d134" +checksum = "3e1d5e639ff6bab73cb6885cc7e7b1de96c3f32c68ec55f3952614bec1092244" dependencies = [ "deranged", "num-conv", @@ -3170,13 +3152,13 @@ dependencies = [ [[package]] name = "tokio-macros" -version = "2.7.2" +version = "2.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +checksum = "6328af13490e73a9b4694030fafd93f8c8c6a9dede33e821c3fc63eddf8042ba" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 2.0.119", ] [[package]] @@ -3614,11 +3596,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ -<<<<<<< HEAD - "windows-sys 0.61.2", -======= - "windows-sys 0.52.0", ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 + "windows-sys 0.59.0", ] [[package]] @@ -3873,7 +3851,7 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c7e468321c81fb07fa7f4c636c3972b9100f0346e5b6a9f2bd0603a52f7ed277" dependencies = [ - "curve25519-dalek 4.1.3", + "curve25519-dalek", "rand_core 0.6.4", "serde", "zeroize", @@ -3933,18 +3911,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.56" +version = "0.8.55" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb" +checksum = "b5a105cd7b140f6eeec8acff2ea38135d3cab283ada58540f629fe51e46696eb" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.56" +version = "0.8.55" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1" +checksum = "0fe976fb70c78cd64cccfe3a6fc142244e8a77b70959b30faf9d0ac37ee228eb" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index a9aa0d8..b0fb5f7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,7 +30,7 @@ dotenv = "0.15.0" strum = "0.28.0" strum_macros = "0.28.0" livekit-api = { version = "0.6.0", features = ["rustls-tls-native-roots"] } -livekit-protocol = "=0.7.10" +livekit-protocol = "0.7.10" thiserror = "2.0.19" trust-dns-resolver = "0.23.2" serde_json = "1.0.151" diff --git a/mtp-type-maps b/mtp-type-maps index 486541b..f753cfd 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit 486541b9483356ff49ff3ec7016f87d3ecbeaa0e +Subproject commit f753cfd406fc0050a8ccd76a90897205efceacd8 diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs index 2ec1b78..a333260 100644 --- a/src/anonymous_clients/anonymous_client_connection.rs +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -8,6 +8,7 @@ use uuid::Uuid; use crate::anonymous_clients::anonymous_manager::{self, generate_username}; use crate::app_state::AppState; use crate::calls::call_group::call_invite_secret_from_cv; +use crate::data::user::UserStatus; use crate::rho::connection::{GeneralConnection, MtpReceiver, MtpSender}; use crate::util::data_type_id; use crate::util::logger::PrintType; @@ -221,13 +222,9 @@ impl AnonymousClientConnection { .await; } - // Presence is account-scoped and anonymous sessions have no - // persisted account preference to change. - if cv.is_type(CommunicationType::SetUserState) - || cv.is_type(CommunicationType::ClientChanged) - { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoUserId) - .await; + // Handle client status changes + if cv.is_type(CommunicationType::ClientChanged) { + self.handle_client_changed(cv).await; return; } @@ -302,6 +299,7 @@ impl AnonymousClientConnection { if cv.is_type(CommunicationType::GetUserData) || cv.is_type(CommunicationType::GetIotaData) + || cv.is_type(CommunicationType::DeleteUser) { self.handle_omega_forward(cv).await; return; @@ -323,6 +321,22 @@ impl AnonymousClientConnection { }); } + /// Handle client status change + async fn handle_client_changed(self: Arc, cv: CommunicationValue) { + if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) { + let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online); + self.state + .omega + .client_changed( + self.user_id as i64, + self.user_id as i64, + self.user_id as i64, + user_status, + ) + .await; + } + } + /// Handle call invite async fn handle_call_invite(self: Arc, cv: CommunicationValue) { let receiver_id: i64 = cv.get_data(DataType::ReceiverId).as_number().unwrap_or(0) as i64; diff --git a/src/config.rs b/src/config.rs index 484da67..7c10773 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,12 +1,10 @@ -use std::{env, time::Duration}; +use std::env; use thiserror::Error; const DEFAULT_RHO_PORT: u16 = 443; const DEFAULT_OMEGA_HOST: &str = "tensamin.net"; const DEFAULT_OMEGA_PORT: u16 = 9187; -const DEFAULT_OMEGA_SYNC_TIMEOUT_SECONDS: u64 = 20; -const DEFAULT_OMEGA_SYNC_RETRIES: u32 = 3; #[derive(Clone, Debug, Eq, PartialEq)] pub struct LiveKitConfig { @@ -21,11 +19,6 @@ pub struct Config { pub omega_host: String, pub omega_port: u16, pub omikron_id: u64, - /// Maximum duration of one route or subscription synchronization request. - pub omega_sync_timeout: Duration, - /// Number of synchronization requests before the transport is closed and - /// the normal reconnect loop starts. - pub omega_sync_retries: u32, pub livekit: Option, } @@ -45,15 +38,6 @@ impl Config { let rho_port = parse_or_default("RHO_PORT", DEFAULT_RHO_PORT)?; let omega_port = parse_or_default("OMEGA_PORT", DEFAULT_OMEGA_PORT)?; let omikron_id = parse_or_default("ID", 0_u64)?; - // Each synchronization request is bounded by this timeout. After - // omega_sync_retries attempts, Omikron closes the authenticated - // transport so the normal reconnect loop can establish a clean state. - let omega_sync_timeout = Duration::from_secs(parse_or_default( - "OMEGA_SYNC_TIMEOUT_SECONDS", - DEFAULT_OMEGA_SYNC_TIMEOUT_SECONDS, - )?); - let omega_sync_retries = - parse_or_default("OMEGA_SYNC_RETRIES", DEFAULT_OMEGA_SYNC_RETRIES)?.max(1); let omega_host = env::var("OMEGA_HOST") .unwrap_or_else(|_| DEFAULT_OMEGA_HOST.to_string()) .trim() @@ -71,8 +55,6 @@ impl Config { omega_host, omega_port, omikron_id, - omega_sync_timeout, - omega_sync_retries, livekit: livekit_from_environment()?, }) } diff --git a/src/data/user.rs b/src/data/user.rs index 1eda0ba..8587ffe 100644 --- a/src/data/user.rs +++ b/src/data/user.rs @@ -1,4 +1,7 @@ -#[derive(Debug, Clone, PartialEq, Eq)] +use strum::IntoEnumIterator; +use strum_macros::EnumIter; + +#[derive(Debug, Clone, PartialEq, EnumIter, Eq)] #[allow(unused, non_camel_case_types)] pub enum UserStatus { user_offline, @@ -17,40 +20,12 @@ impl UserStatus { pub fn to_string(&self) -> String { format!("{:?}", self) } - pub fn from_client_preference(s: &str) -> Option { - match s { - "user_online" => Some(Self::user_online), - "user_idle" => Some(Self::user_idle), - "user_dnd" => Some(Self::user_dnd), - "user_wc" => Some(Self::user_wc), - "user_invisible" => Some(Self::user_invisible), - _ => None, + pub fn from_str(s: &str) -> Option { + for sel in UserStatus::iter() { + if &sel.to_string() == s { + return Some(sel); + } } - } - - pub fn public_value(&self) -> Self { - match self { - Self::user_invisible => Self::user_offline, - value => value.clone(), - } - } - - pub fn from_str(s: &str) -> Option { - Self::from_client_preference(s) - } -} - -#[cfg(test)] -mod tests { - use super::UserStatus; - - #[test] - fn network_parser_rejects_derived_states() { - assert_eq!(UserStatus::from_str("user_offline"), None); - assert_eq!(UserStatus::from_str("iota_offline"), None); - assert_eq!( - UserStatus::from_str("user_online"), - Some(UserStatus::user_online) - ); + None } } diff --git a/src/omega/capabilities.rs b/src/omega/capabilities.rs deleted file mode 100644 index b658068..0000000 --- a/src/omega/capabilities.rs +++ /dev/null @@ -1,145 +0,0 @@ -use std::collections::BTreeSet; - -const OMIKRON_PREFIX: &str = "omikron;caps="; -const OMEGA_PREFIX: &str = "omega;caps="; -const SET_USER_STATE: &str = "set_user_state_v1"; -const STATE_SUBSCRIBE: &str = "state_subscribe_v1"; -const SESSION_SNAPSHOT: &str = "session_snapshot_v1"; -const CLIENT_STATE_PUSH: &str = "client_state_push_v1"; - -#[derive(Clone, Debug, Default, PartialEq, Eq)] -pub struct PeerCapabilities { - pub set_user_state_v1: bool, - pub state_subscribe_v1: bool, - pub session_snapshot_v1: bool, - pub client_state_push_v1: bool, -} - -impl PeerCapabilities { - pub fn current() -> Self { - Self { - set_user_state_v1: true, - state_subscribe_v1: true, - session_snapshot_v1: true, - client_state_push_v1: true, - } - } - - pub fn identification_description(&self) -> String { - let mut names = Vec::new(); - if self.set_user_state_v1 { - names.push(SET_USER_STATE); - } - if self.state_subscribe_v1 { - names.push(STATE_SUBSCRIBE); - } - if self.session_snapshot_v1 { - names.push(SESSION_SNAPSHOT); - } - if self.client_state_push_v1 { - names.push(CLIENT_STATE_PUSH); - } - format!("{OMIKRON_PREFIX}{}", names.join(",")) - } - - pub fn from_identification_description(description: Option<&str>) -> Result { - parse_capabilities(description, OMIKRON_PREFIX) - } - - pub fn from_omega_identification_description(description: Option<&str>) -> Result { - // A missing Omega descriptor selects the documented legacy wire - // contracts: tuple route snapshots, GetStates-only subscriptions, - // and ClientChanged pushes. - parse_capabilities(description, OMEGA_PREFIX) - } -} - -fn parse_capabilities(description: Option<&str>, prefix: &str) -> Result { - let Some(description) = description else { - return Ok(PeerCapabilities::default()); - }; - if description == "omega" { - return Ok(PeerCapabilities::default()); - } - let Some(capabilities) = description.strip_prefix(prefix) else { - return Err(()); - }; - let mut seen = BTreeSet::new(); - for capability in capabilities.split(',') { - if capability.is_empty() || !seen.insert(capability) { - return Err(()); - } - } - if seen.iter().any(|capability| { - !matches!( - *capability, - SET_USER_STATE | STATE_SUBSCRIBE | SESSION_SNAPSHOT | CLIENT_STATE_PUSH - ) - }) { - return Err(()); - } - Ok(PeerCapabilities { - set_user_state_v1: seen.contains(SET_USER_STATE), - state_subscribe_v1: seen.contains(STATE_SUBSCRIBE), - session_snapshot_v1: seen.contains(SESSION_SNAPSHOT), - client_state_push_v1: seen.contains(CLIENT_STATE_PUSH), - }) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn capability_description_round_trips() { - let capabilities = PeerCapabilities::current(); - assert_eq!( - PeerCapabilities::from_identification_description(Some( - &capabilities.identification_description() - )), - Ok(capabilities) - ); - } - - #[test] - fn malformed_capabilities_are_rejected() { - assert!( - PeerCapabilities::from_identification_description(Some("omikron;caps=unknown_v9")) - .is_err() - ); - assert!( - PeerCapabilities::from_identification_description(Some( - "omikron;caps=state_subscribe_v1,state_subscribe_v1" - )) - .is_err() - ); - assert!(PeerCapabilities::from_identification_description(Some("omikron;caps=")).is_err()); - } - - #[test] - fn missing_capabilities_mean_legacy_peer() { - assert_eq!( - PeerCapabilities::from_identification_description(None), - Ok(PeerCapabilities::default()) - ); - assert_eq!( - PeerCapabilities::from_omega_identification_description(Some("omega")), - Ok(PeerCapabilities::default()) - ); - } - - #[test] - fn omega_capabilities_use_the_omega_namespace() { - assert!( - PeerCapabilities::from_omega_identification_description(Some( - "omega;caps=set_user_state_v1" - )) - .unwrap() - .set_user_state_v1 - ); - assert!( - PeerCapabilities::from_identification_description(Some("omega;caps=set_user_state_v1")) - .is_err() - ); - } -} diff --git a/src/omega/mod.rs b/src/omega/mod.rs index 4108cc6..cbe77d9 100644 --- a/src/omega/mod.rs +++ b/src/omega/mod.rs @@ -1,2 +1 @@ -pub mod capabilities; pub mod omega_connection; diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index 97f025b..78a2d38 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -1,9 +1,6 @@ -use super::capabilities::PeerCapabilities; use crate::{ - config::Config, - log_cv_in, log_cv_out, log_err, log_in, log_out, - rho::rho_manager::RhoManager, - util::{data_type_id, logger::PrintType}, + config::Config, data::user::UserStatus, log_cv_in, log_cv_out, log_err, log_in, log_out, + rho::rho_manager::RhoManager, util::logger::PrintType, }; use dashmap::DashMap; use mtp::client::{Client, MTPConnection, Sender}; @@ -16,7 +13,7 @@ use mtp::{ use mtp_transport::ConnectionHandle; use std::{sync::Arc, time::Duration}; use tokio::{ - sync::{Mutex, RwLock, mpsc, oneshot, watch}, + sync::{Mutex, RwLock, mpsc, watch}, task::JoinHandle, time::{Instant, sleep}, }; @@ -26,7 +23,6 @@ use uuid::Uuid; const RECONNECT_DELAY: Duration = Duration::from_secs(5); const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(300); const CONNECTION_TIMEOUT: Duration = Duration::from_secs(10); -const CAPABILITY_NEGOTIATION_TIMEOUT: Duration = Duration::from_secs(1); const PING_INTERVAL: Duration = Duration::from_secs(5); const TASK_CLEANUP_INTERVAL: Duration = Duration::from_secs(60); const TASK_MAX_AGE: Duration = Duration::from_secs(60); @@ -34,37 +30,6 @@ const MAX_CONCURRENT_REQUESTS: usize = 128; const CIRCUIT_BREAKER_FAILURE_THRESHOLD: u32 = 3; const CIRCUIT_BREAKER_COOLDOWN: Duration = Duration::from_secs(30); -fn client_changed_target(value: &CommunicationValue) -> Option<(i64, i64)> { - if !value.is_type(CommunicationType::ClientChanged) { - return None; - } - let receiver = i64::try_from(value.get_receiver()) - .ok() - .filter(|id| *id > 0)?; - let session_id = value - .get_data(DataType::SessionId) - .as_signed_number() - .and_then(|id| i64::try_from(id).ok()) - .filter(|id| *id > 0)?; - value - .get_data(DataType::UserId) - .as_signed_number() - .and_then(|id| i64::try_from(id).ok()) - .filter(|id| *id > 0)?; - value.get_data(DataType::UserState).as_str()?; - Some((receiver, session_id)) -} - -fn parse_omega_capability_response(value: &CommunicationValue) -> Result { - if !value.is_type(CommunicationType::IdentificationResponse) { - return Err("not an identification response".to_string()); - } - PeerCapabilities::from_omega_identification_description( - value.get_data(DataType::Description).as_str(), - ) - .map_err(|_| "Omega advertised an invalid capability set".to_string()) -} - // ============================================================================ // Waiting Task System // ============================================================================ @@ -120,33 +85,24 @@ pub fn start_task_cleanup_loop(omega: Arc) { pub enum ConnectionState { Disconnected, Connecting, - Identifying, - SynchronizingRoutes, - SynchronizingSubscriptions, - Ready, + Connected { identified: bool }, } #[allow(unused_variables)] impl ConnectionState { pub fn is_connected(&self) -> bool { - !matches!( - self, - ConnectionState::Disconnected | ConnectionState::Connecting - ) + match self { + ConnectionState::Connected { identified } => true, + _ => false, + } } #[allow(dead_code)] pub fn is_identified(&self) -> bool { - matches!( - self, - ConnectionState::SynchronizingRoutes - | ConnectionState::SynchronizingSubscriptions - | ConnectionState::Ready - ) - } - - pub fn is_ready(&self) -> bool { - matches!(self, ConnectionState::Ready) + match self { + ConnectionState::Connected { identified: true } => true, + _ => false, + } } } @@ -161,8 +117,6 @@ pub struct OmegaConnection { host: String, port: u16, omikron_id: u64, - sync_timeout: Duration, - sync_retries: u32, pub connection_id: Uuid, shutdown_tx: Arc>>>, // Track if we should reconnect on close @@ -172,14 +126,6 @@ pub struct OmegaConnection { keyring: mtp::crypto::Keyring, rho: Arc, waiting_tasks: DashMap, - session_presence: DashMap<(i64, i64), SessionPresenceState>, - peer_capabilities: Arc>, -} - -#[derive(Clone, Debug, Default, PartialEq, Eq)] -pub struct SessionPresenceState { - pub subscribed_user_ids: Vec, - pub pending: bool, } impl OmegaConnection { @@ -197,8 +143,6 @@ impl OmegaConnection { host: config.omega_host.clone(), port: config.omega_port, omikron_id: config.omikron_id, - sync_timeout: config.omega_sync_timeout, - sync_retries: config.omega_sync_retries, connection_id: Uuid::new_v4(), shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))), reconnect_on_close: Arc::new(RwLock::new(true)), @@ -210,8 +154,6 @@ impl OmegaConnection { keyring, rho, waiting_tasks: DashMap::new(), - session_presence: DashMap::new(), - peer_capabilities: Arc::new(RwLock::new(PeerCapabilities::default())), } } @@ -325,7 +267,6 @@ impl OmegaConnection { let client_config = ClientConfig::new(format!("https://{}:{}", self.host, self.port)) .with_client_id(self.omikron_id) - .with_description(PeerCapabilities::current().identification_description()) .with_policy( Policy::default() .with_send_mode(SendMode::SingleStreamPerMessage) @@ -361,43 +302,18 @@ impl OmegaConnection { // Store sender let sender_arc = Arc::new(connection.sender.clone()); *self.sender.write().await = Some(sender_arc.clone()); - // A new authenticated transport gets a fresh request circuit. A - // previous transport can have failed while synchronizing and must not - // prevent this connection from making its initial requests. - self.circuit_breaker.lock().await.record_success(); - *self.state.write().await = ConnectionState::Identifying; + *self.state.write().await = ConnectionState::Connected { identified: false }; // Get handle for close monitoring let sender_handle = sender_arc.handle().clone(); let connection = Arc::new(connection); - let (capability_tx, capability_rx) = oneshot::channel(); // Start read loop let read_self = self.clone(); let read_handle = tokio::spawn(async move { - read_self - .read_loop(connection, sender_handle, Some(capability_tx)) - .await; + read_self.read_loop(connection, sender_handle).await; }); - let negotiated_capabilities = - match tokio::time::timeout(CAPABILITY_NEGOTIATION_TIMEOUT, capability_rx).await { - Ok(Ok(Ok(capabilities))) => capabilities, - Ok(Ok(Err(error))) => { - self.fail_synchronization("capability negotiation", error) - .await; - let _ = read_handle.await; - return Err("Omega capability negotiation failed".to_string()); - } - Ok(Err(_)) | Err(_) => { - // An Omega from before capability negotiation sends no second - // identification response. Fall back to its documented - // legacy contracts rather than guessing modern support. - PeerCapabilities::default() - } - }; - *self.peer_capabilities.write().await = negotiated_capabilities; - // Tell omega our current state now that we're actually connected - // doing this after teardown (as before) sent into a sender that had // already been cleared, silently dropping the sync every time. @@ -438,10 +354,8 @@ impl OmegaConnection { } } - async fn sync_client_iota_status(&self) { - *self.state.write().await = ConnectionState::SynchronizingRoutes; + async fn sync_client_iota_status(self: Arc) { let mut connected_iota_ids: Vec = Vec::new(); - let mut connected_sessions: Vec = Vec::new(); let mut connected_user_ids: Vec = Vec::new(); let iota_ids = self.rho.iota_ids().await; @@ -453,120 +367,21 @@ impl OmegaConnection { for rho in rho_connections { for client_conn in rho.get_client_connections().await { - let type_map = mtp::codec::TypeMap::latest(); - let user_id = DataValue::SignedNumber(client_conn.get_user_id().await.into()); - let session_id = DataValue::SignedNumber((client_conn.session_id as i128).into()); - let iota_id = DataValue::SignedNumber((rho.get_iota_id().await as i128).into()); - let peer_capabilities = self.peer_capabilities.read().await.clone(); - if peer_capabilities.session_snapshot_v1 { - connected_sessions.push(DataValue::Container(vec![ - (data_type_id(DataType::UserId, &type_map), user_id), - (data_type_id(DataType::SessionId, &type_map), session_id), - (data_type_id(DataType::IotaId, &type_map), iota_id), - ])); - } else { - connected_user_ids.push(user_id); - } + connected_user_ids.push(DataValue::SignedNumber( + client_conn.get_user_id().await.into(), + )); } } - let peer_capabilities = self.peer_capabilities.read().await.clone(); - let sync_msg = if peer_capabilities.session_snapshot_v1 { - CommunicationValue::new(CommunicationType::SyncClientIotaStatus) - .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids)) - .add_typed_default(DataType::UserStates, DataValue::Array(connected_sessions)) - .add_typed_default( - DataType::RhoConnections, - DataValue::SignedNumber(self.rho.connection_count().await as i128), - ) - } else { - // Legacy Omega has no request/response contract for this message. - // Its exact snapshot is IotaIds + UserIds + RhoConnections. - CommunicationValue::new(CommunicationType::SyncClientIotaStatus) - .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids)) - .add_typed_default(DataType::UserIds, DataValue::Array(connected_user_ids)) - .add_typed_default( - DataType::RhoConnections, - DataValue::SignedNumber(self.rho.connection_count().await as i128), - ) - }; + let sync_msg = CommunicationValue::new(CommunicationType::SyncClientIotaStatus) + .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids)) + .add_typed_default(DataType::UserIds, DataValue::Array(connected_user_ids)) + .add_typed_default( + DataType::RhoConnections, + DataValue::SignedNumber(self.rho.connection_count().await as i128), + ); - if !peer_capabilities.session_snapshot_v1 { - if let Err(error) = self.send_message_result(&sync_msg).await { - self.fail_synchronization("legacy route synchronization", error) - .await; - return; - } - *self.state.write().await = ConnectionState::SynchronizingSubscriptions; - if let Err(error) = self.restore_presence_subscriptions().await { - self.fail_synchronization("subscription restoration", error) - .await; - return; - } - *self.state.write().await = ConnectionState::Ready; - return; - } - - let mut response = Err("route synchronization did not start".to_string()); - for attempt in 0..self.sync_retries { - response = self - .await_response(&sync_msg, Some(self.sync_timeout)) - .await; - match &response { - Ok(value) if value.is_type(CommunicationType::Success) => break, - Ok(value) if value.is_type(CommunicationType::ErrorInvalidData) => break, - _ if attempt + 1 < self.sync_retries => { - sleep(Duration::from_millis(250 * 2u64.pow(attempt))).await - } - _ => break, - } - } - match response { - Ok(response) if response.is_type(CommunicationType::Success) => { - *self.state.write().await = ConnectionState::SynchronizingSubscriptions; - if let Err(error) = self.restore_presence_subscriptions().await { - log_err!( - 0, - PrintType::Omega, - "Presence subscription restoration failed: {}", - error - ); - self.fail_synchronization("subscription restoration", error) - .await; - return; - } - *self.state.write().await = ConnectionState::Ready; - } - Ok(response) => { - let error = format!( - "Omega rejected presence snapshot with {}", - response.get_type() - ); - self.fail_synchronization("route synchronization", error) - .await; - return; - } - Err(error) => { - self.fail_synchronization("route synchronization", error) - .await; - return; - } - } - } - - async fn fail_synchronization(&self, phase: &str, error: String) { - log_err!( - 0, - PrintType::Omega, - "Omega {} failed after {} attempts: {}; closing transport for reconnect", - phase, - self.sync_retries, - error - ); - *self.state.write().await = ConnectionState::Disconnected; - if let Some(sender) = self.sender.read().await.as_ref() { - sender.close().await; - } + self.send_message(&sync_msg).await; } // ------------------------------------------------------------------------- @@ -577,7 +392,6 @@ impl OmegaConnection { self: Arc, connection: Arc, sender_handle: Arc, - mut capability_tx: Option>>, ) { // Monitor both receiver and sender handle for close let mut close_rx = sender_handle.subscribe_close(); @@ -589,14 +403,6 @@ impl OmegaConnection { Ok(cv) => { log_cv_in!(PrintType::Omega, &cv); - if cv.is_type(CommunicationType::IdentificationResponse) { - if let Some(tx) = capability_tx.take() { - let result = parse_omega_capability_response(&cv); - let _ = tx.send(result); - } - continue; - } - let msg_id = cv.get_id(); if let Some((_, task)) = self.waiting_tasks.remove(&msg_id) { if (task.task)(self.clone(), cv.clone()) { @@ -624,36 +430,6 @@ impl OmegaConnection { } } } - if cv.is_type(CommunicationType::EraseHostedUserData) { - let Some(iota_id) = cv - .get_data(DataType::IotaId) - .as_number() - .and_then(|id| i64::try_from(id).ok()) - else { - log_err!(0, PrintType::Omega, "Discarded hosted-data erasure without IotaId"); - continue; - }; - if let Some(rho) = self.rho.get_by_iota(iota_id).await { - rho.get_iota_connection().send_message(&cv).await; - } else { - log_err!(iota_id, PrintType::Omega, "Hosted-data erasure is pending because the Iota is offline"); - } - continue; - } - if cv.is_type(CommunicationType::ClientChanged) { - let Some((receiver, session_id)) = client_changed_target(&cv) - else { - continue; - }; - if let Some(client) = self - .rho - .get_client_connection(receiver, session_id) - .await - { - client.send_message(&cv).await; - } - continue; - } } Err(e) => { log_err!(0, PrintType::Omega, "Receive error: {}", e); @@ -712,14 +488,6 @@ impl OmegaConnection { let _ = self.try_send_message(cv).await; } - pub async fn send_message_result(&self, cv: &CommunicationValue) -> Result<(), String> { - self.try_send_message(cv).await - } - - pub async fn supports_set_user_state(&self) -> bool { - self.peer_capabilities.read().await.set_user_state_v1 - } - pub async fn await_connection(&self, timeout_duration: Option) -> Result<(), String> { if self.state.read().await.is_connected() { return Ok(()); @@ -839,371 +607,72 @@ impl OmegaConnection { self.state.read().await.is_identified() } - pub async fn is_ready(&self) -> bool { - self.state.read().await.is_ready() - } - #[allow(dead_code)] - pub async fn close_iota(&self, iota_id: i64) -> Result<(), String> { + pub async fn close_iota(&self, iota_id: i64) { let cv = CommunicationValue::new(CommunicationType::IotaDisconnected) .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())); - let request_id = cv.get_id(); - let result = self.lifecycle_request(&cv).await; - if let Err(error) = &result { - self.log_lifecycle_failure("IotaDisconnected", iota_id, None, request_id, error); - self.sync_client_iota_status().await; - } - result + self.send_message(&cv).await; } - pub async fn client_disconnected(&self, user_id: i64, session_id: i64) -> Result<(), String> { - self.session_presence.remove(&(user_id, session_id)); + pub async fn client_changed( + &self, + _iota_id: i64, + user_id: i64, + session_id: i64, + state: UserStatus, + ) { + let cv = CommunicationValue::new(CommunicationType::ClientChanged) + .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(session_id.into()), + ) + .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())); + self.send_message(&cv).await; + } + + pub async fn client_disconnected(&self, user_id: i64, session_id: i64) { let cv = CommunicationValue::new(CommunicationType::UserDisconnected) .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) .add_typed_default( DataType::SessionId, DataValue::SignedNumber(session_id.into()), ); - let request_id = cv.get_id(); - let result = self.lifecycle_request(&cv).await; - if let Err(error) = &result { - self.log_lifecycle_failure( - "UserDisconnected", - user_id, - Some(session_id), - request_id, - error, - ); - self.sync_client_iota_status().await; - } - result + self.send_message(&cv).await; } - pub async fn user_connected( - &self, - user_id: i64, - session_id: i64, - iota_id: i64, - ) -> Result<(), String> { - let request = CommunicationValue::new(CommunicationType::UserConnected) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) - .add_typed_default( - DataType::SessionId, - DataValue::SignedNumber(session_id.into()), - ) - .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())) - .add_typed_default( - DataType::UserState, - DataValue::Str("user_online".to_string()), - ) - .add_typed_default( - DataType::UpdatedAt, - DataValue::SignedNumber( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i128, + pub async fn user_states(&self, user_id: i64, user_ids: Vec) { + let user_ids = user_ids + .iter() + .map(|v| DataValue::SignedNumber((*v).into())) + .collect(); + + let cv = CommunicationValue::new(CommunicationType::GetStates) + .add_typed_default(DataType::UserIds, DataValue::Array(user_ids)); + let msg_id = cv.get_id(); + + let rho_manager = self.rho.clone(); + self.waiting_tasks.insert( + msg_id, + WaitingTask { + task: Box::new( + move |_: Arc, response: CommunicationValue| { + let rho_manager = rho_manager.clone(); + tokio::spawn(async move { + let rho = rho_manager.get_for_user(user_id).await; + if let Some(rho) = rho { + for client in rho.get_client_connections_for_user(user_id).await { + client.send_message(&response).await; + } + } + }); + true + }, ), - ); - let request_id = request.get_id(); - let result = self.lifecycle_request(&request).await; - if let Err(error) = &result { - self.log_lifecycle_failure( - "UserConnected", - user_id, - Some(session_id), - request_id, - error, - ); - self.sync_client_iota_status().await; - } - result - } - - pub async fn iota_connected(&self, iota_id: i64) -> Result<(), String> { - let request = CommunicationValue::new(CommunicationType::IotaConnected) - .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())); - let request_id = request.get_id(); - let result = self.lifecycle_request(&request).await; - if let Err(error) = &result { - self.log_lifecycle_failure("IotaConnected", iota_id, None, request_id, error); - self.sync_client_iota_status().await; - } - result - } - - pub async fn reconcile_routes(&self) { - self.sync_client_iota_status().await; - } - - async fn lifecycle_request(&self, request: &CommunicationValue) -> Result<(), String> { - if !self.is_ready().await { - return Err("Omega is not synchronized".to_string()); - } - if !self.peer_capabilities.read().await.session_snapshot_v1 { - // Legacy Omega handles lifecycle messages as notifications and - // never returns Success. A transport-level send result is the - // only acknowledgement available under that contract. - return self.send_message_result(request).await; - } - let response = self - .await_response(request, Some(Duration::from_secs(20))) - .await?; - if response.is_type(CommunicationType::Success) { - Ok(()) - } else { - Err(format!("Omega rejected {}", response.get_type())) - } - } - - fn log_lifecycle_failure( - &self, - operation: &str, - entity_id: i64, - session_id: Option, - request_id: u32, - error: &str, - ) { - log_err!( - 0, - PrintType::Omega, - "Lifecycle operation={} entity_id={} session_id={:?} request_id={} failed: {}", - operation, - entity_id, - session_id, - request_id, - error - ); - } - - pub async fn replace_state_subscription( - &self, - user_id: i64, - session_id: i64, - mut user_ids: Vec, - ) -> Result<(), String> { - if user_id <= 0 || session_id <= 0 { - return Err("invalid user or session id".to_string()); - } - - user_ids.sort_unstable(); - user_ids.dedup(); - - let key = (user_id, session_id); - // Store the latest authoritative snapshot before checking readiness. - // A client can synchronize while the transport is reconnecting. - self.session_presence.insert( - key, - SessionPresenceState { - subscribed_user_ids: user_ids.clone(), - pending: true, + inserted_at: Instant::now(), }, ); - if !self.is_ready().await { - return Err("Omega is not synchronized".to_string()); - } - - self.deliver_state_subscription(key, user_ids).await - } - - async fn deliver_state_subscription( - &self, - key: (i64, i64), - user_ids: Vec, - ) -> Result<(), String> { - let (user_id, session_id) = key; - - if !self.peer_capabilities.read().await.state_subscribe_v1 { - let delivered_ids = user_ids.clone(); - let result = self - .request_user_states(user_id, session_id, user_ids) - .await; - if result.is_ok() { - self.mark_presence_delivered(key, &delivered_ids).await; - } - return result; - } - - let subscription = CommunicationValue::new(CommunicationType::StateSubscribe) - .with_sender(user_id as u64) - .add_typed_default( - DataType::SessionId, - DataValue::SignedNumber(session_id.into()), - ) - .add_typed_default( - DataType::UserIds, - DataValue::Array( - user_ids - .iter() - .map(|user_id| DataValue::SignedNumber((*user_id).into())) - .collect(), - ), - ); - let response = self - .await_response(&subscription, Some(Duration::from_secs(20))) - .await?; - if !response.is_type(CommunicationType::Success) { - return Err(format!( - "StateSubscribe rejected with {}", - response.get_type() - )); - } - - let delivered_ids = user_ids.clone(); - self.request_user_states(user_id, session_id, user_ids) - .await?; - self.mark_presence_delivered(key, &delivered_ids).await; - Ok(()) - } - - async fn restore_presence_subscriptions(&self) -> Result<(), String> { - let mut active = Vec::new(); - for rho in self.rho.connections().await { - active.extend(rho.get_client_connections().await); - } - let active_keys: std::collections::HashSet<_> = active - .iter() - .map(|client| (client.user_id as i64, client.session_id as i64)) - .collect(); - - // Do not replay subscriptions for sessions that disappeared while the - // transport was down. - self.session_presence - .retain(|key, _| active_keys.contains(key)); - - for key in active_keys { - let Some(snapshot) = self.session_presence.get(&key).map(|entry| entry.clone()) else { - continue; - }; - if !snapshot.pending { - continue; - } - self.deliver_state_subscription(key, snapshot.subscribed_user_ids) - .await?; - } - Ok(()) - } - - async fn mark_presence_delivered(&self, key: (i64, i64), delivered_ids: &[i64]) { - if let Some(mut entry) = self.session_presence.get_mut(&key) - && entry.subscribed_user_ids == delivered_ids - { - entry.pending = false; - } - } - - async fn request_user_states( - &self, - user_id: i64, - session_id: i64, - user_ids: Vec, - ) -> Result<(), String> { - let request = CommunicationValue::new(CommunicationType::GetStates) - .add_typed_default( - DataType::SessionId, - DataValue::SignedNumber(session_id.into()), - ) - .add_typed_default( - DataType::UserIds, - DataValue::Array( - user_ids - .iter() - .map(|user_id| DataValue::SignedNumber((*user_id).into())) - .collect(), - ), - ); - let response = self - .await_response(&request, Some(Duration::from_secs(20))) - .await?; - let legacy_peer = !self.peer_capabilities.read().await.client_state_push_v1; - let response = - if legacy_peer && response.get_data(DataType::SessionId).as_number().is_none() { - response.add_typed_default( - DataType::SessionId, - DataValue::SignedNumber(session_id.into()), - ) - } else { - response - }; - if !legacy_peer - && response.get_data(DataType::SessionId).as_number() != Some(session_id as i128) - { - return Err("GetStates response had the wrong session".to_string()); - } - - if let Some(client) = self.rho.get_client_connection(user_id, session_id).await { - client.send_message(&response).await; - } - Ok(()) - } -} - -#[cfg(test)] -mod tests { - use super::{PeerCapabilities, client_changed_target, parse_omega_capability_response}; - use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; - - fn notification() -> CommunicationValue { - CommunicationValue::new(CommunicationType::ClientChanged) - .with_receiver(7) - .add_typed_default(DataType::SessionId, DataValue::SignedNumber(11)) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(20)) - .add_typed_default(DataType::UserState, DataValue::Str("user_online".into())) - } - - #[test] - fn client_changed_target_requires_all_routing_fields() { - assert_eq!(client_changed_target(¬ification()), Some((7, 11))); - - let mut missing_session = notification(); - missing_session.remove_data(DataType::SessionId); - assert_eq!(client_changed_target(&missing_session), None); - - let mut missing_state = notification(); - missing_state.remove_data(DataType::UserState); - assert_eq!(client_changed_target(&missing_state), None); - } - - #[test] - fn client_changed_target_rejects_non_positive_ids() { - let invalid = CommunicationValue::new(CommunicationType::ClientChanged) - .with_receiver(0) - .add_typed_default(DataType::SessionId, DataValue::SignedNumber(11)) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(20)) - .add_typed_default(DataType::UserState, DataValue::Str("user_online".into())); - assert_eq!(client_changed_target(&invalid), None); - } - - #[test] - fn capability_response_negotiates_new_omega() { - let response = CommunicationValue::new(CommunicationType::IdentificationResponse) - .add_typed_default( - DataType::Description, - DataValue::Str( - "omega;caps=set_user_state_v1,state_subscribe_v1,session_snapshot_v1,client_state_push_v1" - .into(), - ), - ); - let capabilities = parse_omega_capability_response(&response).unwrap(); - assert!(capabilities.state_subscribe_v1); - assert!(capabilities.client_state_push_v1); - } - - #[test] - fn missing_capability_response_is_legacy() { - let response = CommunicationValue::new(CommunicationType::IdentificationResponse); - let capabilities = parse_omega_capability_response(&response).unwrap(); - assert_eq!(capabilities, PeerCapabilities::default()); - } - - #[test] - fn unsupported_capability_response_is_rejected() { - let response = CommunicationValue::new(CommunicationType::IdentificationResponse) - .add_typed_default( - DataType::Description, - DataValue::Str("omega;caps=unsupported_v9".into()), - ); - assert!(parse_omega_capability_response(&response).is_err()); + self.send_message(&cv).await; } } diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 0a6f30f..64cb052 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -24,6 +24,7 @@ pub struct ClientConnection { pub receiver: Arc, pub_key: Arc>>>, pub rho_connection: Arc>>>, + pub interested_users: Arc>>, is_open: Arc>, message_slots: Arc, } @@ -34,6 +35,7 @@ impl ClientConnection { state: general.state.clone(), pub_key: Arc::new(RwLock::new(None)), rho_connection: general.rho_connection.clone(), + interested_users: Arc::new(RwLock::new(Vec::new())), is_open: Arc::new(RwLock::new(true)), sender: general.sender.clone(), receiver: general.receiver.clone(), @@ -77,12 +79,6 @@ impl ClientConnection { *self.rho_connection.write().await = Some(rho_connection); } - /// A membership snapshot can remove an account from an Iota while keeping - /// its authenticated Omega connection alive. - pub async fn clear_rho_connection(&self) { - *self.rho_connection.write().await = None; - } - /// Send a CommunicationValue to the client pub async fn send_message(self: Arc, cv: &CommunicationValue) { if !*self.is_open.read().await { @@ -108,26 +104,9 @@ impl ClientConnection { let mut cv = cv; - if cv.is_type(CommunicationType::SetUserState) { - self.handle_set_user_state(cv).await; - return; - } - - // Compatibility for clients predating SetUserState. The target - // user fields, if present, are deliberately ignored: an - // authenticated connection may only change its own state. - if cv.is_type(CommunicationType::ClientChanged) - && cv.get_data_opt(DataType::UserState).is_some() - { - self.handle_set_user_state( - CommunicationValue::new(CommunicationType::SetUserState) - .with_id(cv.get_id()) - .add_typed_default( - DataType::UserState, - cv.get_data(DataType::UserState).clone(), - ), - ) - .await; + // Handle client status changes + if cv.is_type(CommunicationType::ClientChanged) { + self.handle_client_changed(cv).await; return; } @@ -212,46 +191,6 @@ impl ClientConnection { || cv.is_type(CommunicationType::GetIotaData) || cv.is_type(CommunicationType::DeleteUser) { - if cv.is_type(CommunicationType::ChangeUserData) - && cv.get_data_opt(DataType::OnlineStatus).is_some() - { - let mut profile_request = cv.clone(); - let preference = profile_request.remove_data(DataType::OnlineStatus); - let state_request = CommunicationValue::new(CommunicationType::SetUserState) - .with_id(cv.get_id()) - .add_typed_default(DataType::UserState, preference.unwrap()); - let state_response = match self.request_set_user_state(state_request).await { - Ok(response) => response, - Err(error_type) => { - self.send_error_response(cv.get_id(), error_type).await; - return; - } - }; - if !state_response.is_type(CommunicationType::Success) { - self.send_message(&state_response).await; - return; - } - - if profile_request.data_len() == 0 { - self.send_message(&state_response).await; - return; - } - - match self - .await_omega_response(profile_request.with_sender(self.user_id)) - .await - { - Ok(response) if response.is_type(CommunicationType::Success) => { - self.send_message(&response).await; - } - Ok(response) => self.send_message(&response).await, - Err(_) => { - self.send_error_response(cv.get_id(), CommunicationType::ErrorInternal) - .await; - } - } - return; - } let sender = self.get_user_id().await; self.handle_omega_forward(cv.with_sender(sender as u64)) .await; @@ -351,91 +290,34 @@ impl ClientConnection { }); } async fn handle_omega_forward(self: Arc, cv: CommunicationValue) { - let request_id = cv.get_id(); - match self.await_omega_response(cv).await { - Ok(response_cv) => self.send_message(&response_cv).await, - Err(_) => { - self.send_error_response(request_id, CommunicationType::ErrorInternal) - .await; - } - } - } - - async fn await_omega_response(&self, cv: CommunicationValue) -> Result { - self.state + let client_for_closure = self.clone(); + let response_cv = self + .state .omega .clone() .await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20))) - .await - .map_err(|_| ()) + .await; + if let Ok(response_cv) = response_cv { + client_for_closure.send_message(&response_cv).await; + } } - async fn request_set_user_state( - &self, - cv: CommunicationValue, - ) -> Result { - if !self.state.omega.is_ready().await { - return Err(CommunicationType::ErrorInternal); - } - let Some(state) = cv - .get_data(DataType::UserState) - .as_str() - .and_then(UserStatus::from_client_preference) - else { - return Err(CommunicationType::ErrorInvalidData); - }; - let Some(rho) = self.get_rho_connection().await else { - return Err(CommunicationType::ErrorNoIota); - }; - let request = if self.state.omega.supports_set_user_state().await { - CommunicationValue::new(CommunicationType::SetUserState) - .with_id(cv.get_id()) - .with_sender(self.user_id) - .add_typed_default( - DataType::UserId, - DataValue::SignedNumber(self.user_id as i128), - ) - .add_typed_default( - DataType::SessionId, - DataValue::SignedNumber(self.session_id as i128), - ) - .add_typed_default( - DataType::IotaId, - DataValue::SignedNumber(rho.get_iota_id().await as i128), - ) - .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())) - } else { - // Legacy Omega accepts the original ClientChanged request. It is - // intentionally limited to the authenticated user's ID and a - // writable preference, with no derived connectivity state. - CommunicationValue::new(CommunicationType::ClientChanged) - .with_id(cv.get_id()) - .with_sender(self.user_id) - .add_typed_default( - DataType::UserId, - DataValue::SignedNumber(self.user_id as i128), - ) - .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())) - }; - if !self.state.omega.supports_set_user_state().await { - self.state - .omega - .send_message_result(&request) - .await - .map_err(|_| CommunicationType::ErrorInternal)?; - return Ok(CommunicationValue::new(CommunicationType::Success) - .with_id(cv.get_id()) - .add_typed_default(DataType::UserState, DataValue::Str(state.to_string()))); - } - self.await_omega_response(request) - .await - .map_err(|_| CommunicationType::ErrorInternal) - } - - async fn handle_set_user_state(self: Arc, cv: CommunicationValue) { - match self.request_set_user_state(cv.clone()).await { - Ok(response) => self.send_message(&response).await, - Err(error_type) => self.send_error_response(cv.get_id(), error_type).await, + /// Handle client status change + async fn handle_client_changed(self: Arc, cv: CommunicationValue) { + let user_id = self.get_user_id().await; + if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) { + let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online); + if let Some(rho_conn) = self.get_rho_connection().await { + self.state + .omega + .client_changed( + rho_conn.get_iota_id().await as i64, + user_id as i64, + self.session_id as i64, + user_status, + ) + .await; + } } } @@ -948,6 +830,35 @@ impl ClientConnection { let _ = self.sender.close(); } + /// Set interested users list + pub async fn set_interested_users(self: Arc, interested_ids: Vec) { + let mut interested_guard = self.interested_users.write().await; + *interested_guard = interested_ids; + } + #[allow(dead_code)] + pub async fn get_interested_users(self: Arc) -> Vec { + let interested_guard = self.interested_users.read().await; + interested_guard.clone() + } + + /// Check if interested in a user and send notification + #[allow(dead_code)] + pub async fn are_you_interested(self: Arc, user_id: i64, user_status: &str) { + let interested_guard = self.clone().get_interested_users().await; + if interested_guard.contains(&user_id) { + let status = if user_status == "user_invisible" { + "user_offline" + } else { + user_status + }; + let notification = CommunicationValue::new(CommunicationType::ClientChanged) + .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) + .add_typed_default(DataType::UserState, DataValue::Str(status.to_string())); + + self.send_message(¬ification).await; + } + } + /// Handle connection close pub async fn handle_close(self: Arc) { let user_id = self.get_user_id().await; @@ -969,6 +880,7 @@ impl Clone for ClientConnection { client_version: self.client_version.clone(), pub_key: Arc::clone(&self.pub_key), rho_connection: Arc::clone(&self.rho_connection), + interested_users: Arc::clone(&self.interested_users), is_open: Arc::clone(&self.is_open), message_slots: Arc::clone(&self.message_slots), } diff --git a/src/rho/connection.rs b/src/rho/connection.rs index cede813..137929b 100755 --- a/src/rho/connection.rs +++ b/src/rho/connection.rs @@ -136,23 +136,10 @@ impl GeneralConnection { if let Some(rho_conn) = rho { rho_conn.bind_user_id(user_id).await; rho_conn.add_client_connection(client.clone()).await; - match self - .notify_user_connected(user_id, rho_conn.get_iota_id().await as i64) - .await - { - Ok(()) => { - self.send_initial_client_state_request(&rho_conn, user_id) - .await; - } - Err(error) => { - log_err!( - user_id, - PrintType::Client, - "UserConnected acknowledgement failed: {}", - error - ); - } - } + self.notify_user_connected(user_id, rho_conn.get_iota_id().await as i64) + .await; + self.send_initial_client_state_request(&rho_conn, user_id) + .await; } else { log_err!( user_id, @@ -168,13 +155,29 @@ impl GeneralConnection { true } - async fn notify_user_connected(&self, user_id: i64, iota_id: i64) -> Result<(), String> { + async fn notify_user_connected(&self, user_id: i64, iota_id: i64) { let session_id = *self.session_id.read().await as i64; - self.state - .omega - .clone() - .user_connected(user_id, session_id, iota_id) - .await + let notify = CommunicationValue::new(CommunicationType::UserConnected) + .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(session_id.into()), + ) + .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())) + .add_typed_default( + DataType::UserState, + DataValue::Str("user_online".to_string()), + ) + .add_typed_default( + DataType::UpdatedAt, + DataValue::SignedNumber( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i128, + ), + ); + self.state.omega.clone().send_message(¬ify).await; } async fn find_user_rho(&self, user_id: i64) -> Option> { @@ -224,6 +227,9 @@ impl GeneralConnection { async fn migrate_iota(self: &Arc) { let id = self.id; log_in!(id as i64, PrintType::Iota, "Iota {} connected", id); + let notify = CommunicationValue::new(CommunicationType::IotaConnected) + .add_typed_default(DataType::IotaId, DataValue::SignedNumber(id.into())); + self.state.omega.clone().send_message(¬ify).await; let iota = IotaConnection::from_general(self.clone(), id).await; let previous = self.state.rho.get_by_iota(id as i64).await; @@ -240,14 +246,6 @@ impl GeneralConnection { } } self.state.rho.add(rho).await; - if let Err(error) = self.state.omega.clone().iota_connected(id as i64).await { - log_err!( - id as i64, - PrintType::Iota, - "IotaConnected acknowledgement failed: {}", - error - ); - } self.load_iota_users(id).await; iota.start(); } diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 747a394..7f7086c 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -23,47 +23,6 @@ use tokio::sync::mpsc; use super::rho_connection::RhoConnection; -fn contact_snapshot(value: &CommunicationValue) -> Option<(i64, i64, Vec)> { - if !value.is_type(CommunicationType::ClientStateSync) - && !value.is_type(CommunicationType::AddConversation) - { - return None; - } - - let user_id = i64::try_from(value.get_receiver()) - .ok() - .filter(|id| *id > 0)?; - let session_id = value - .get_data(DataType::SessionId) - .as_signed_number() - .and_then(|id| i64::try_from(id).ok()) - .filter(|id| *id > 0)?; - let DataValue::Array(values) = value.get_data(DataType::UserIds) else { - return None; - }; - - let user_id_type = data_type_id(DataType::UserId, &TypeMap::latest()); - let mut contact_ids = Vec::with_capacity(values.len()); - for value in values { - let contact_id = match value { - DataValue::SignedNumber(id) => i64::try_from(*id).ok(), - DataValue::Container(entries) => entries.iter().find_map(|(key, value)| { - (*key == user_id_type) - .then(|| value.as_signed_number()) - .flatten() - .and_then(|id| i64::try_from(id).ok()) - }), - _ => None, - } - .filter(|id| *id > 0)?; - contact_ids.push(contact_id); - } - contact_ids.sort_unstable(); - contact_ids.dedup(); - - Some((user_id, session_id, contact_ids)) -} - #[allow(dead_code)] pub struct IotaConnection { pub state: Arc, @@ -213,29 +172,6 @@ impl IotaConnection { } } - async fn replace_state_subscription( - &self, - user_id: i64, - session_id: i64, - contact_ids: Vec, - ) -> Result<(), String> { - let Some(rho_connection) = self.get_rho_connection().await else { - return Err("Rho connection is unavailable".to_string()); - }; - if rho_connection - .get_client_connection(user_id, session_id) - .await - .is_none() - { - return Err("target client session is no longer connected".to_string()); - } - - self.state - .omega - .replace_state_subscription(user_id, session_id, contact_ids) - .await - } - /// Send a CommunicationValue to the Iota pub async fn send_message(&self, cv: &CommunicationValue) { log_cv_out!(PrintType::Iota, cv); @@ -270,34 +206,6 @@ impl IotaConnection { cv }; - // Iota's UserIds field is an authoritative replacement snapshot for - // the requesting client session, including contact mutations. - if let Some((user_id, session_id, contact_ids)) = contact_snapshot(&cv) { - if let Err(error) = self - .replace_state_subscription(user_id, session_id, contact_ids) - .await - { - log_err!( - self.iota_id as i64, - PrintType::Omega, - "Failed to replace presence subscription for user {} session {}: {}", - user_id, - session_id, - error - ); - } - } - - if cv.is_type(CommunicationType::StateSubscribe) { - self.send_error_response( - cv.get_id(), - CommunicationType::ErrorInvalidData, - Some("StateSubscribe must come from an authoritative contact snapshot"), - ) - .await; - return; - } - // Handle GET_CHATS if cv.is_type(CommunicationType::GetChats) { self.handle_get_chats(cv).await; @@ -442,12 +350,6 @@ impl IotaConnection { || cv.is_type(CommunicationType::GetUserData) || cv.is_type(CommunicationType::GetIotaData) || cv.is_type(CommunicationType::DeleteIota) - || cv.is_type(CommunicationType::AttachUserBegin) - || cv.is_type(CommunicationType::AttachUserComplete) - || cv.is_type(CommunicationType::ReleaseUserFromIota) - || cv.is_type(CommunicationType::DeleteUserCredentialBegin) - || cv.is_type(CommunicationType::DeleteUserCredentialComplete) - || cv.is_type(CommunicationType::EraseHostedUserDataAck) { let sender = self.get_iota_id().await; @@ -615,20 +517,7 @@ impl IotaConnection { } let mut interested_ids: Vec = Vec::new(); - let Some(session_id) = cv - .get_data(DataType::SessionId) - .as_signed_number() - .and_then(|id| i64::try_from(id).ok()) - .filter(|id| *id > 0) - else { - self.forward_to_client( - CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) - .with_receiver(user_id), - ) - .await; - return; - }; + let session_id = cv.get_data(DataType::SessionId).as_signed_number(); let tm = TypeMap::latest(); // Presence interest is the complete contact set, independent of @@ -769,18 +658,20 @@ impl IotaConnection { // ============================ // Notify Omega // ============================ - if let Err(error) = self - .replace_state_subscription(user_id as i64, session_id, interested_ids.clone()) - .await - { - log_err!( - self.iota_id as i64, - PrintType::Omega, - "Failed to replace presence subscription for user {} session {}: {}", - user_id, - session_id, - error - ); + self.state + .omega + .user_states(user_id as i64, interested_ids.clone()) + .await; + + // ============================ + // Notify Rho + // ============================ + if let Some(rho_conn) = self.get_rho_connection().await { + if let Some(session_id) = session_id.and_then(|id| i64::try_from(id).ok()) { + rho_conn + .set_interested(user_id as i64, session_id, interested_ids) + .await; + } } // ============================ @@ -888,48 +779,3 @@ impl std::fmt::Debug for IotaConnection { .finish() } } - -#[cfg(test)] -mod tests { - use super::contact_snapshot; - use crate::util::data_type_id; - use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue, TypeMap}; - - fn snapshot(user_ids: DataValue) -> CommunicationValue { - CommunicationValue::new(CommunicationType::ClientStateSync) - .with_receiver(7) - .add_typed_default(DataType::SessionId, DataValue::SignedNumber(11)) - .add_typed_default(DataType::UserIds, user_ids) - } - - #[test] - fn contact_snapshot_deduplicates_flat_and_typed_ids() { - let user_id_type = data_type_id(DataType::UserId, &TypeMap::latest()); - let parsed = contact_snapshot(&snapshot(DataValue::Array(vec![ - DataValue::SignedNumber(20), - DataValue::Container(vec![(user_id_type, DataValue::SignedNumber(21))]), - DataValue::SignedNumber(20), - ]))) - .unwrap(); - - assert_eq!(parsed, (7, 11, vec![20, 21])); - } - - #[test] - fn contact_snapshot_accepts_empty_contact_sets() { - assert_eq!( - contact_snapshot(&snapshot(DataValue::Array(Vec::new()))), - Some((7, 11, Vec::new())) - ); - } - - #[test] - fn contact_snapshot_rejects_malformed_contact_sets() { - assert_eq!( - contact_snapshot(&snapshot(DataValue::Array(vec![DataValue::Str( - "bad".into(), - )]))), - None - ); - } -} diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 9944572..46f0330 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -1,6 +1,6 @@ use super::{client_connection::ClientConnection, iota_connection::IotaConnection}; -use crate::{log_err, rho::app_connection::AppConnection}; +use crate::{data::user::UserStatus, rho::app_connection::AppConnection}; use dashmap::DashMap; use mtp::codec::{CommunicationValue, DataType}; use std::sync::Arc; @@ -86,14 +86,6 @@ impl RhoConnection { .collect() } - pub async fn detach_user_clients(&self, user_id: i64) { - let clients = self.get_client_connections_for_user(user_id).await; - for client in clients { - client.clear_rho_connection().await; - } - self.client_connections.retain(|(id, _), _| *id != user_id as u64); - } - #[allow(unused)] pub async fn get_app_connections( &self, @@ -186,21 +178,11 @@ impl RhoConnection { return; } self.client_connections.remove(&key); - if let Err(error) = self - .iota_connection + self.iota_connection .state .omega .client_disconnected(target_user_id, target_session_id) - .await - { - log_err!( - target_user_id, - crate::util::logger::PrintType::Client, - "UserDisconnected acknowledgement failed for session {}: {}", - target_session_id, - error - ); - } + .await; let remaining_for_user = self .client_connections .iter() @@ -212,6 +194,16 @@ impl RhoConnection { .rho .remove_user_binding(target_user_id, self.get_iota_id().await as i64) .await; + self.iota_connection + .state + .omega + .client_changed( + self.get_iota_id().await as i64, + target_user_id, + target_session_id, + UserStatus::user_offline, + ) + .await; } } @@ -236,14 +228,7 @@ impl RhoConnection { } // Notify OmegaConnection - if let Err(error) = self.iota_connection.state.omega.close_iota(iota_id).await { - log_err!( - iota_id, - crate::util::logger::PrintType::Iota, - "IotaDisconnected acknowledgement failed: {}", - error - ); - } + self.iota_connection.state.omega.close_iota(iota_id).await; } /// Send message from Iota to specific client @@ -270,6 +255,25 @@ impl RhoConnection { self.iota_connection.send_message(&cv).await; } + /// Set interested users for a specific client + pub async fn set_interested(&self, user_id: i64, session_id: i64, interested_ids: Vec) { + if let Some(connection) = self.get_client_connection(user_id, session_id).await { + connection.set_interested_users(interested_ids).await; + } + } + + /// Check if clients are interested in a user + #[allow(dead_code)] + pub async fn are_they_interested(&self, user_id: i64, user_status: &str) { + let connections = self.get_client_connections().await; + for connection in connections.iter() { + connection + .clone() + .are_you_interested(user_id, user_status) + .await; + } + } + /// Check if this RhoConnection contains a specific user ID #[allow(dead_code)] pub async fn contains_user(&self, user_id: &i64) -> bool { diff --git a/src/rho/rho_manager.rs b/src/rho/rho_manager.rs index 5cb7ce3..0345ae8 100644 --- a/src/rho/rho_manager.rs +++ b/src/rho/rho_manager.rs @@ -66,11 +66,7 @@ impl RhoManager { ) .await; - let replacement = user_ids.iter().copied().collect::>(); for user_id in previous_users { - if !replacement.contains(&user_id) { - rho.detach_user_clients(user_id).await; - } if self .users .get(&user_id) @@ -139,17 +135,4 @@ impl RhoManager { .map(|entry| entry.value().clone()) .collect() } - - pub async fn get_client_connection( - &self, - user_id: i64, - session_id: i64, - ) -> Option> { - for rho in self.connections().await { - if let Some(client) = rho.get_client_connection(user_id, session_id).await { - return Some(client); - } - } - None - } }