+5 -1

.github/workflows/checks.yml

reviewed

··· 23 23 runs-on: ubuntu-24.04 24 24 steps: 25 25 - uses: actions/checkout@v4 26 26 + - name: get nightly toolchain for jetstream fmt 27 27 + run: rustup toolchain install nightly --allow-downgrade -c rustfmt 26 28 - name: fmt 27 27 - run: cargo fmt --all -- --check 29 29 + run: cargo fmt --package links --package constellation -- --check 30 30 + - name: fmt jetstream (nightly) 31 31 + run: cargo +nightly fmt --package jetstream -- --check 28 32 - name: clippy 29 33 run: cargo clippy --all-targets --all-features -- -D warnings

+1

.gitignore

reviewed

··· 1 1 /target 2 2 + local/

+556 -1

Cargo.lock

reviewed

··· 162 162 ] 163 163 164 164 [[package]] 165 165 + name = "async-trait" 166 166 + version = "0.1.87" 167 167 + source = "registry+https://github.com/rust-lang/crates.io-index" 168 168 + checksum = "d556ec1359574147ec0c4fc5eb525f3f23263a592b1a9c07e0a75b427de55c97" 169 169 + dependencies = [ 170 170 + "proc-macro2", 171 171 + "quote", 172 172 + "syn", 173 173 + ] 174 174 + 175 175 + [[package]] 165 176 name = "atomic-waker" 166 177 version = "1.1.2" 167 178 source = "registry+https://github.com/rust-lang/crates.io-index" 168 179 checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" 169 180 170 181 [[package]] 182 182 + name = "atrium-api" 183 183 + version = "0.24.10" 184 184 + source = "registry+https://github.com/rust-lang/crates.io-index" 185 185 + checksum = "9c5d74937642f6b21814e82d80f54d55ebd985b681bffbe27c8a76e726c3c4db" 186 186 + dependencies = [ 187 187 + "atrium-xrpc", 188 188 + "chrono", 189 189 + "http", 190 190 + "ipld-core", 191 191 + "langtag", 192 192 + "regex", 193 193 + "serde", 194 194 + "serde_bytes", 195 195 + "serde_json", 196 196 + "thiserror 1.0.69", 197 197 + "trait-variant", 198 198 + ] 199 199 + 200 200 + [[package]] 201 201 + name = "atrium-xrpc" 202 202 + version = "0.12.1" 203 203 + source = "registry+https://github.com/rust-lang/crates.io-index" 204 204 + checksum = "6b4956d94147cfbb669c68f654eb4fd6a1d00648c810cec79d04ec5425b8f378" 205 205 + dependencies = [ 206 206 + "http", 207 207 + "serde", 208 208 + "serde_html_form", 209 209 + "serde_json", 210 210 + "thiserror 1.0.69", 211 211 + "trait-variant", 212 212 + ] 213 213 + 214 214 + [[package]] 171 215 name = "autocfg" 172 216 version = "1.4.0" 173 217 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 278 322 "rustc-demangle", 279 323 "windows-targets", 280 324 ] 325 325 + 326 326 + [[package]] 327 327 + name = "base-x" 328 328 + version = "0.2.11" 329 329 + source = "registry+https://github.com/rust-lang/crates.io-index" 330 330 + checksum = "4cbbc9d0964165b47557570cce6c952866c2678457aca742aafc9fb771d30270" 281 331 282 332 [[package]] 283 333 name = "base64" ··· 437 487 dependencies = [ 438 488 "android-tzdata", 439 489 "iana-time-zone", 490 490 + "js-sys", 440 491 "num-traits", 441 492 "serde", 493 493 + "wasm-bindgen", 442 494 "windows-targets", 443 495 ] 444 496 445 497 [[package]] 498 498 + name = "cid" 499 499 + version = "0.11.1" 500 500 + source = "registry+https://github.com/rust-lang/crates.io-index" 501 501 + checksum = "3147d8272e8fa0ccd29ce51194dd98f79ddfb8191ba9e3409884e751798acf3a" 502 502 + dependencies = [ 503 503 + "core2", 504 504 + "multibase", 505 505 + "multihash", 506 506 + "serde", 507 507 + "serde_bytes", 508 508 + "unsigned-varint", 509 509 + ] 510 510 + 511 511 + [[package]] 446 512 name = "clang-sys" 447 513 version = "1.8.1" 448 514 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 541 607 "tokio", 542 608 "tokio-util", 543 609 "tower-http", 544 544 - "tungstenite", 610 610 + "tungstenite 0.26.1", 545 611 "zstd", 546 612 ] 547 613 ··· 562 628 checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" 563 629 564 630 [[package]] 631 631 + name = "core2" 632 632 + version = "0.4.0" 633 633 + source = "registry+https://github.com/rust-lang/crates.io-index" 634 634 + checksum = "b49ba7ef1ad6107f8824dbe97de947cbaac53c44e7f9756a1fba0d37c1eec505" 635 635 + dependencies = [ 636 636 + "memchr", 637 637 + ] 638 638 + 639 639 + [[package]] 565 640 name = "cpufeatures" 566 641 version = "0.2.17" 567 642 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 647 722 checksum = "0e60eed09d8c01d3cee5b7d30acb059b76614c918fa0f992e0dd6eeb10daad6f" 648 723 649 724 [[package]] 725 725 + name = "data-encoding-macro" 726 726 + version = "0.1.16" 727 727 + source = "registry+https://github.com/rust-lang/crates.io-index" 728 728 + checksum = "5b16d9d0d88a5273d830dac8b78ceb217ffc9b1d5404e5597a3542515329405b" 729 729 + dependencies = [ 730 730 + "data-encoding", 731 731 + "data-encoding-macro-internal", 732 732 + ] 733 733 + 734 734 + [[package]] 735 735 + name = "data-encoding-macro-internal" 736 736 + version = "0.1.14" 737 737 + source = "registry+https://github.com/rust-lang/crates.io-index" 738 738 + checksum = "1145d32e826a7748b69ee8fc62d3e6355ff7f1051df53141e7048162fc90481b" 739 739 + dependencies = [ 740 740 + "data-encoding", 741 741 + "syn", 742 742 + ] 743 743 + 744 744 + [[package]] 650 745 name = "deranged" 651 746 version = "0.3.11" 652 747 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 667 762 ] 668 763 669 764 [[package]] 765 765 + name = "displaydoc" 766 766 + version = "0.2.5" 767 767 + source = "registry+https://github.com/rust-lang/crates.io-index" 768 768 + checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0" 769 769 + dependencies = [ 770 770 + "proc-macro2", 771 771 + "quote", 772 772 + "syn", 773 773 + ] 774 774 + 775 775 + [[package]] 670 776 name = "either" 671 777 version = "1.13.0" 672 778 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 710 816 source = "registry+https://github.com/rust-lang/crates.io-index" 711 817 checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" 712 818 dependencies = [ 819 819 + "futures-core", 820 820 + "futures-sink", 821 821 + "nanorand", 713 822 "spin", 714 823 ] 715 824 ··· 775 884 checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" 776 885 777 886 [[package]] 887 887 + name = "futures-macro" 888 888 + version = "0.3.31" 889 889 + source = "registry+https://github.com/rust-lang/crates.io-index" 890 890 + checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" 891 891 + dependencies = [ 892 892 + "proc-macro2", 893 893 + "quote", 894 894 + "syn", 895 895 + ] 896 896 + 897 897 + [[package]] 778 898 name = "futures-sink" 779 899 version = "0.3.31" 780 900 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 793 913 checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" 794 914 dependencies = [ 795 915 "futures-core", 916 916 + "futures-macro", 917 917 + "futures-sink", 796 918 "futures-task", 797 919 "pin-project-lite", 798 920 "pin-utils", 921 921 + "slab", 799 922 ] 800 923 801 924 [[package]] ··· 815 938 checksum = "c4567c8db10ae91089c99af84c68c38da3ec2f087c3f82960bcdbf3656b6f4d7" 816 939 dependencies = [ 817 940 "cfg-if", 941 941 + "js-sys", 818 942 "libc", 819 943 "wasi 0.11.0+wasi-snapshot-preview1", 944 944 + "wasm-bindgen", 820 945 ] 821 946 822 947 [[package]] ··· 1043 1168 ] 1044 1169 1045 1170 [[package]] 1171 1171 + name = "icu_collections" 1172 1172 + version = "1.5.0" 1173 1173 + source = "registry+https://github.com/rust-lang/crates.io-index" 1174 1174 + checksum = "db2fa452206ebee18c4b5c2274dbf1de17008e874b4dc4f0aea9d01ca79e4526" 1175 1175 + dependencies = [ 1176 1176 + "displaydoc", 1177 1177 + "yoke", 1178 1178 + "zerofrom", 1179 1179 + "zerovec", 1180 1180 + ] 1181 1181 + 1182 1182 + [[package]] 1183 1183 + name = "icu_locid" 1184 1184 + version = "1.5.0" 1185 1185 + source = "registry+https://github.com/rust-lang/crates.io-index" 1186 1186 + checksum = "13acbb8371917fc971be86fc8057c41a64b521c184808a698c02acc242dbf637" 1187 1187 + dependencies = [ 1188 1188 + "displaydoc", 1189 1189 + "litemap", 1190 1190 + "tinystr", 1191 1191 + "writeable", 1192 1192 + "zerovec", 1193 1193 + ] 1194 1194 + 1195 1195 + [[package]] 1196 1196 + name = "icu_locid_transform" 1197 1197 + version = "1.5.0" 1198 1198 + source = "registry+https://github.com/rust-lang/crates.io-index" 1199 1199 + checksum = "01d11ac35de8e40fdeda00d9e1e9d92525f3f9d887cdd7aa81d727596788b54e" 1200 1200 + dependencies = [ 1201 1201 + "displaydoc", 1202 1202 + "icu_locid", 1203 1203 + "icu_locid_transform_data", 1204 1204 + "icu_provider", 1205 1205 + "tinystr", 1206 1206 + "zerovec", 1207 1207 + ] 1208 1208 + 1209 1209 + [[package]] 1210 1210 + name = "icu_locid_transform_data" 1211 1211 + version = "1.5.0" 1212 1212 + source = "registry+https://github.com/rust-lang/crates.io-index" 1213 1213 + checksum = "fdc8ff3388f852bede6b579ad4e978ab004f139284d7b28715f773507b946f6e" 1214 1214 + 1215 1215 + [[package]] 1216 1216 + name = "icu_normalizer" 1217 1217 + version = "1.5.0" 1218 1218 + source = "registry+https://github.com/rust-lang/crates.io-index" 1219 1219 + checksum = "19ce3e0da2ec68599d193c93d088142efd7f9c5d6fc9b803774855747dc6a84f" 1220 1220 + dependencies = [ 1221 1221 + "displaydoc", 1222 1222 + "icu_collections", 1223 1223 + "icu_normalizer_data", 1224 1224 + "icu_properties", 1225 1225 + "icu_provider", 1226 1226 + "smallvec", 1227 1227 + "utf16_iter", 1228 1228 + "utf8_iter", 1229 1229 + "write16", 1230 1230 + "zerovec", 1231 1231 + ] 1232 1232 + 1233 1233 + [[package]] 1234 1234 + name = "icu_normalizer_data" 1235 1235 + version = "1.5.0" 1236 1236 + source = "registry+https://github.com/rust-lang/crates.io-index" 1237 1237 + checksum = "f8cafbf7aa791e9b22bec55a167906f9e1215fd475cd22adfcf660e03e989516" 1238 1238 + 1239 1239 + [[package]] 1240 1240 + name = "icu_properties" 1241 1241 + version = "1.5.1" 1242 1242 + source = "registry+https://github.com/rust-lang/crates.io-index" 1243 1243 + checksum = "93d6020766cfc6302c15dbbc9c8778c37e62c14427cb7f6e601d849e092aeef5" 1244 1244 + dependencies = [ 1245 1245 + "displaydoc", 1246 1246 + "icu_collections", 1247 1247 + "icu_locid_transform", 1248 1248 + "icu_properties_data", 1249 1249 + "icu_provider", 1250 1250 + "tinystr", 1251 1251 + "zerovec", 1252 1252 + ] 1253 1253 + 1254 1254 + [[package]] 1255 1255 + name = "icu_properties_data" 1256 1256 + version = "1.5.0" 1257 1257 + source = "registry+https://github.com/rust-lang/crates.io-index" 1258 1258 + checksum = "67a8effbc3dd3e4ba1afa8ad918d5684b8868b3b26500753effea8d2eed19569" 1259 1259 + 1260 1260 + [[package]] 1261 1261 + name = "icu_provider" 1262 1262 + version = "1.5.0" 1263 1263 + source = "registry+https://github.com/rust-lang/crates.io-index" 1264 1264 + checksum = "6ed421c8a8ef78d3e2dbc98a973be2f3770cb42b606e3ab18d6237c4dfde68d9" 1265 1265 + dependencies = [ 1266 1266 + "displaydoc", 1267 1267 + "icu_locid", 1268 1268 + "icu_provider_macros", 1269 1269 + "stable_deref_trait", 1270 1270 + "tinystr", 1271 1271 + "writeable", 1272 1272 + "yoke", 1273 1273 + "zerofrom", 1274 1274 + "zerovec", 1275 1275 + ] 1276 1276 + 1277 1277 + [[package]] 1278 1278 + name = "icu_provider_macros" 1279 1279 + version = "1.5.0" 1280 1280 + source = "registry+https://github.com/rust-lang/crates.io-index" 1281 1281 + checksum = "1ec89e9337638ecdc08744df490b221a7399bf8d164eb52a665454e60e075ad6" 1282 1282 + dependencies = [ 1283 1283 + "proc-macro2", 1284 1284 + "quote", 1285 1285 + "syn", 1286 1286 + ] 1287 1287 + 1288 1288 + [[package]] 1046 1289 name = "ident_case" 1047 1290 version = "1.0.1" 1048 1291 source = "registry+https://github.com/rust-lang/crates.io-index" 1049 1292 checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" 1050 1293 1051 1294 [[package]] 1295 1295 + name = "idna" 1296 1296 + version = "1.0.3" 1297 1297 + source = "registry+https://github.com/rust-lang/crates.io-index" 1298 1298 + checksum = "686f825264d630750a544639377bae737628043f20d38bbc029e8f29ea968a7e" 1299 1299 + dependencies = [ 1300 1300 + "idna_adapter", 1301 1301 + "smallvec", 1302 1302 + "utf8_iter", 1303 1303 + ] 1304 1304 + 1305 1305 + [[package]] 1306 1306 + name = "idna_adapter" 1307 1307 + version = "1.2.0" 1308 1308 + source = "registry+https://github.com/rust-lang/crates.io-index" 1309 1309 + checksum = "daca1df1c957320b2cf139ac61e7bd64fed304c5040df000a745aa1de3b4ef71" 1310 1310 + dependencies = [ 1311 1311 + "icu_normalizer", 1312 1312 + "icu_properties", 1313 1313 + ] 1314 1314 + 1315 1315 + [[package]] 1052 1316 name = "indexmap" 1053 1317 version = "1.9.3" 1054 1318 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1068 1332 "equivalent", 1069 1333 "hashbrown 0.15.2", 1070 1334 "serde", 1335 1335 + ] 1336 1336 + 1337 1337 + [[package]] 1338 1338 + name = "ipld-core" 1339 1339 + version = "0.4.2" 1340 1340 + source = "registry+https://github.com/rust-lang/crates.io-index" 1341 1341 + checksum = "104718b1cc124d92a6d01ca9c9258a7df311405debb3408c445a36452f9bf8db" 1342 1342 + dependencies = [ 1343 1343 + "cid", 1344 1344 + "serde", 1345 1345 + "serde_bytes", 1071 1346 ] 1072 1347 1073 1348 [[package]] ··· 1107 1382 checksum = "d75a2a4b1b190afb6f5425f10f6a8f959d2ea0b9c2b1d79553551850539e4674" 1108 1383 1109 1384 [[package]] 1385 1385 + name = "jetstream" 1386 1386 + version = "0.1.1" 1387 1387 + dependencies = [ 1388 1388 + "anyhow", 1389 1389 + "async-trait", 1390 1390 + "atrium-api", 1391 1391 + "chrono", 1392 1392 + "clap", 1393 1393 + "flume", 1394 1394 + "futures-util", 1395 1395 + "log", 1396 1396 + "serde", 1397 1397 + "serde_json", 1398 1398 + "thiserror 2.0.11", 1399 1399 + "tokio", 1400 1400 + "tokio-tungstenite", 1401 1401 + "tokio-util", 1402 1402 + "url", 1403 1403 + "zstd", 1404 1404 + ] 1405 1405 + 1406 1406 + [[package]] 1110 1407 name = "jobserver" 1111 1408 version = "0.1.32" 1112 1409 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1123 1420 dependencies = [ 1124 1421 "once_cell", 1125 1422 "wasm-bindgen", 1423 1423 + ] 1424 1424 + 1425 1425 + [[package]] 1426 1426 + name = "langtag" 1427 1427 + version = "0.3.4" 1428 1428 + source = "registry+https://github.com/rust-lang/crates.io-index" 1429 1429 + checksum = "ed60c85f254d6ae8450cec15eedd921efbc4d1bdf6fcf6202b9a58b403f6f805" 1430 1430 + dependencies = [ 1431 1431 + "serde", 1126 1432 ] 1127 1433 1128 1434 [[package]] ··· 1214 1520 checksum = "d26c52dbd32dccf2d10cac7725f8eae5296885fb5703b261f7d0a0739ec807ab" 1215 1521 1216 1522 [[package]] 1523 1523 + name = "litemap" 1524 1524 + version = "0.7.5" 1525 1525 + source = "registry+https://github.com/rust-lang/crates.io-index" 1526 1526 + checksum = "23fb14cb19457329c82206317a5663005a4d404783dc74f4252769b0d5f42856" 1527 1527 + 1528 1528 + [[package]] 1217 1529 name = "lock_api" 1218 1530 version = "0.4.12" 1219 1531 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1368 1680 "libc", 1369 1681 "wasi 0.11.0+wasi-snapshot-preview1", 1370 1682 "windows-sys 0.52.0", 1683 1683 + ] 1684 1684 + 1685 1685 + [[package]] 1686 1686 + name = "multibase" 1687 1687 + version = "0.9.1" 1688 1688 + source = "registry+https://github.com/rust-lang/crates.io-index" 1689 1689 + checksum = "9b3539ec3c1f04ac9748a260728e855f261b4977f5c3406612c884564f329404" 1690 1690 + dependencies = [ 1691 1691 + "base-x", 1692 1692 + "data-encoding", 1693 1693 + "data-encoding-macro", 1694 1694 + ] 1695 1695 + 1696 1696 + [[package]] 1697 1697 + name = "multihash" 1698 1698 + version = "0.19.3" 1699 1699 + source = "registry+https://github.com/rust-lang/crates.io-index" 1700 1700 + checksum = "6b430e7953c29dd6a09afc29ff0bb69c6e306329ee6794700aee27b76a1aea8d" 1701 1701 + dependencies = [ 1702 1702 + "core2", 1703 1703 + "serde", 1704 1704 + "unsigned-varint", 1705 1705 + ] 1706 1706 + 1707 1707 + [[package]] 1708 1708 + name = "nanorand" 1709 1709 + version = "0.7.0" 1710 1710 + source = "registry+https://github.com/rust-lang/crates.io-index" 1711 1711 + checksum = "6a51313c5820b0b02bd422f4b44776fbf47961755c74ce64afc73bfad10226c3" 1712 1712 + dependencies = [ 1713 1713 + "getrandom 0.2.15", 1371 1714 ] 1372 1715 1373 1716 [[package]] ··· 1837 2180 ] 1838 2181 1839 2182 [[package]] 2183 2183 + name = "serde_bytes" 2184 2184 + version = "0.11.16" 2185 2185 + source = "registry+https://github.com/rust-lang/crates.io-index" 2186 2186 + checksum = "364fec0df39c49a083c9a8a18a23a6bcfd9af130fe9fe321d18520a0d113e09e" 2187 2187 + dependencies = [ 2188 2188 + "serde", 2189 2189 + ] 2190 2190 + 2191 2191 + [[package]] 1840 2192 name = "serde_derive" 1841 2193 version = "1.0.217" 1842 2194 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1845 2197 "proc-macro2", 1846 2198 "quote", 1847 2199 "syn", 2200 2200 + ] 2201 2201 + 2202 2202 + [[package]] 2203 2203 + name = "serde_html_form" 2204 2204 + version = "0.2.7" 2205 2205 + source = "registry+https://github.com/rust-lang/crates.io-index" 2206 2206 + checksum = "9d2de91cf02bbc07cde38891769ccd5d4f073d22a40683aa4bc7a95781aaa2c4" 2207 2207 + dependencies = [ 2208 2208 + "form_urlencoded", 2209 2209 + "indexmap 2.7.1", 2210 2210 + "itoa", 2211 2211 + "ryu", 2212 2212 + "serde", 1848 2213 ] 1849 2214 1850 2215 [[package]] ··· 1978 2343 ] 1979 2344 1980 2345 [[package]] 2346 2346 + name = "stable_deref_trait" 2347 2347 + version = "1.2.0" 2348 2348 + source = "registry+https://github.com/rust-lang/crates.io-index" 2349 2349 + checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3" 2350 2350 + 2351 2351 + [[package]] 1981 2352 name = "strsim" 1982 2353 version = "0.11.1" 1983 2354 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1999 2370 version = "1.0.2" 2000 2371 source = "registry+https://github.com/rust-lang/crates.io-index" 2001 2372 checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" 2373 2373 + 2374 2374 + [[package]] 2375 2375 + name = "synstructure" 2376 2376 + version = "0.13.1" 2377 2377 + source = "registry+https://github.com/rust-lang/crates.io-index" 2378 2378 + checksum = "c8af7666ab7b6390ab78131fb5b0fce11d6b7a6951602017c35fa82800708971" 2379 2379 + dependencies = [ 2380 2380 + "proc-macro2", 2381 2381 + "quote", 2382 2382 + "syn", 2383 2383 + ] 2002 2384 2003 2385 [[package]] 2004 2386 name = "tempfile" ··· 2092 2474 checksum = "9ab95735ea2c8fd51154d01e39cf13912a78071c2d89abc49a7ef102a7dd725a" 2093 2475 2094 2476 [[package]] 2477 2477 + name = "tinystr" 2478 2478 + version = "0.7.6" 2479 2479 + source = "registry+https://github.com/rust-lang/crates.io-index" 2480 2480 + checksum = "9117f5d4db391c1cf6927e7bea3db74b9a1c1add8f7eda9ffd5364f40f57b82f" 2481 2481 + dependencies = [ 2482 2482 + "displaydoc", 2483 2483 + "zerovec", 2484 2484 + ] 2485 2485 + 2486 2486 + [[package]] 2095 2487 name = "tokio" 2096 2488 version = "1.43.0" 2097 2489 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 2118 2510 "proc-macro2", 2119 2511 "quote", 2120 2512 "syn", 2513 2513 + ] 2514 2514 + 2515 2515 + [[package]] 2516 2516 + name = "tokio-native-tls" 2517 2517 + version = "0.3.1" 2518 2518 + source = "registry+https://github.com/rust-lang/crates.io-index" 2519 2519 + checksum = "bbae76ab933c85776efabc971569dd6119c580d8f5d448769dec1764bf796ef2" 2520 2520 + dependencies = [ 2521 2521 + "native-tls", 2522 2522 + "tokio", 2523 2523 + ] 2524 2524 + 2525 2525 + [[package]] 2526 2526 + name = "tokio-tungstenite" 2527 2527 + version = "0.24.0" 2528 2528 + source = "registry+https://github.com/rust-lang/crates.io-index" 2529 2529 + checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9" 2530 2530 + dependencies = [ 2531 2531 + "futures-util", 2532 2532 + "log", 2533 2533 + "native-tls", 2534 2534 + "tokio", 2535 2535 + "tokio-native-tls", 2536 2536 + "tungstenite 0.24.0", 2121 2537 ] 2122 2538 2123 2539 [[package]] ··· 2196 2612 ] 2197 2613 2198 2614 [[package]] 2615 2615 + name = "trait-variant" 2616 2616 + version = "0.1.2" 2617 2617 + source = "registry+https://github.com/rust-lang/crates.io-index" 2618 2618 + checksum = "70977707304198400eb4835a78f6a9f928bf41bba420deb8fdb175cd965d77a7" 2619 2619 + dependencies = [ 2620 2620 + "proc-macro2", 2621 2621 + "quote", 2622 2622 + "syn", 2623 2623 + ] 2624 2624 + 2625 2625 + [[package]] 2199 2626 name = "try-lock" 2200 2627 version = "0.2.5" 2201 2628 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 2203 2630 2204 2631 [[package]] 2205 2632 name = "tungstenite" 2633 2633 + version = "0.24.0" 2634 2634 + source = "registry+https://github.com/rust-lang/crates.io-index" 2635 2635 + checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a" 2636 2636 + dependencies = [ 2637 2637 + "byteorder", 2638 2638 + "bytes", 2639 2639 + "data-encoding", 2640 2640 + "http", 2641 2641 + "httparse", 2642 2642 + "log", 2643 2643 + "native-tls", 2644 2644 + "rand", 2645 2645 + "sha1", 2646 2646 + "thiserror 1.0.69", 2647 2647 + "url", 2648 2648 + "utf-8", 2649 2649 + ] 2650 2650 + 2651 2651 + [[package]] 2652 2652 + name = "tungstenite" 2206 2653 version = "0.26.1" 2207 2654 source = "registry+https://github.com/rust-lang/crates.io-index" 2208 2655 checksum = "413083a99c579593656008130e29255e54dcaae495be556cc26888f211648c24" ··· 2239 2686 checksum = "a210d160f08b701c8721ba1c726c11662f877ea6b7094007e1ca9a1041945034" 2240 2687 2241 2688 [[package]] 2689 2689 + name = "unsigned-varint" 2690 2690 + version = "0.8.0" 2691 2691 + source = "registry+https://github.com/rust-lang/crates.io-index" 2692 2692 + checksum = "eb066959b24b5196ae73cb057f45598450d2c5f71460e98c49b738086eff9c06" 2693 2693 + 2694 2694 + [[package]] 2695 2695 + name = "url" 2696 2696 + version = "2.5.4" 2697 2697 + source = "registry+https://github.com/rust-lang/crates.io-index" 2698 2698 + checksum = "32f8b686cadd1473f4bd0117a5d28d36b1ade384ea9b5069a1c40aefed7fda60" 2699 2699 + dependencies = [ 2700 2700 + "form_urlencoded", 2701 2701 + "idna", 2702 2702 + "percent-encoding", 2703 2703 + ] 2704 2704 + 2705 2705 + [[package]] 2242 2706 name = "utf-8" 2243 2707 version = "0.7.6" 2244 2708 source = "registry+https://github.com/rust-lang/crates.io-index" 2245 2709 checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" 2710 2710 + 2711 2711 + [[package]] 2712 2712 + name = "utf16_iter" 2713 2713 + version = "1.0.5" 2714 2714 + source = "registry+https://github.com/rust-lang/crates.io-index" 2715 2715 + checksum = "c8232dd3cdaed5356e0f716d285e4b40b932ac434100fe9b7e0e8e935b9e6246" 2716 2716 + 2717 2717 + [[package]] 2718 2718 + name = "utf8_iter" 2719 2719 + version = "1.0.4" 2720 2720 + source = "registry+https://github.com/rust-lang/crates.io-index" 2721 2721 + checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" 2246 2722 2247 2723 [[package]] 2248 2724 name = "utf8parse" ··· 2541 3017 ] 2542 3018 2543 3019 [[package]] 3020 3020 + name = "write16" 3021 3021 + version = "1.0.0" 3022 3022 + source = "registry+https://github.com/rust-lang/crates.io-index" 3023 3023 + checksum = "d1890f4022759daae28ed4fe62859b1236caebfc61ede2f63ed4e695f3f6d936" 3024 3024 + 3025 3025 + [[package]] 3026 3026 + name = "writeable" 3027 3027 + version = "0.5.5" 3028 3028 + source = "registry+https://github.com/rust-lang/crates.io-index" 3029 3029 + checksum = "1e9df38ee2d2c3c5948ea468a8406ff0db0b29ae1ffde1bcf20ef305bcc95c51" 3030 3030 + 3031 3031 + [[package]] 3032 3032 + name = "yoke" 3033 3033 + version = "0.7.5" 3034 3034 + source = "registry+https://github.com/rust-lang/crates.io-index" 3035 3035 + checksum = "120e6aef9aa629e3d4f52dc8cc43a015c7724194c97dfaf45180d2daf2b77f40" 3036 3036 + dependencies = [ 3037 3037 + "serde", 3038 3038 + "stable_deref_trait", 3039 3039 + "yoke-derive", 3040 3040 + "zerofrom", 3041 3041 + ] 3042 3042 + 3043 3043 + [[package]] 3044 3044 + name = "yoke-derive" 3045 3045 + version = "0.7.5" 3046 3046 + source = "registry+https://github.com/rust-lang/crates.io-index" 3047 3047 + checksum = "2380878cad4ac9aac1e2435f3eb4020e8374b5f13c296cb75b4620ff8e229154" 3048 3048 + dependencies = [ 3049 3049 + "proc-macro2", 3050 3050 + "quote", 3051 3051 + "syn", 3052 3052 + "synstructure", 3053 3053 + ] 3054 3054 + 3055 3055 + [[package]] 2544 3056 name = "zerocopy" 2545 3057 version = "0.7.35" 2546 3058 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 2555 3067 version = "0.7.35" 2556 3068 source = "registry+https://github.com/rust-lang/crates.io-index" 2557 3069 checksum = "fa4f8080344d4671fb4e831a13ad1e68092748387dfc4f55e356242fae12ce3e" 3070 3070 + dependencies = [ 3071 3071 + "proc-macro2", 3072 3072 + "quote", 3073 3073 + "syn", 3074 3074 + ] 3075 3075 + 3076 3076 + [[package]] 3077 3077 + name = "zerofrom" 3078 3078 + version = "0.1.6" 3079 3079 + source = "registry+https://github.com/rust-lang/crates.io-index" 3080 3080 + checksum = "50cc42e0333e05660c3587f3bf9d0478688e15d870fab3346451ce7f8c9fbea5" 3081 3081 + dependencies = [ 3082 3082 + "zerofrom-derive", 3083 3083 + ] 3084 3084 + 3085 3085 + [[package]] 3086 3086 + name = "zerofrom-derive" 3087 3087 + version = "0.1.6" 3088 3088 + source = "registry+https://github.com/rust-lang/crates.io-index" 3089 3089 + checksum = "d71e5d6e06ab090c67b5e44993ec16b72dcbaabc526db883a360057678b48502" 3090 3090 + dependencies = [ 3091 3091 + "proc-macro2", 3092 3092 + "quote", 3093 3093 + "syn", 3094 3094 + "synstructure", 3095 3095 + ] 3096 3096 + 3097 3097 + [[package]] 3098 3098 + name = "zerovec" 3099 3099 + version = "0.10.4" 3100 3100 + source = "registry+https://github.com/rust-lang/crates.io-index" 3101 3101 + checksum = "aa2b893d79df23bfb12d5461018d408ea19dfafe76c2c7ef6d4eba614f8ff079" 3102 3102 + dependencies = [ 3103 3103 + "yoke", 3104 3104 + "zerofrom", 3105 3105 + "zerovec-derive", 3106 3106 + ] 3107 3107 + 3108 3108 + [[package]] 3109 3109 + name = "zerovec-derive" 3110 3110 + version = "0.10.3" 3111 3111 + source = "registry+https://github.com/rust-lang/crates.io-index" 3112 3112 + checksum = "6eafa6dfb17584ea3e2bd6e76e0cc15ad7af12b09abdd1ca55961bed9b1063c6" 2558 3113 dependencies = [ 2559 3114 "proc-macro2", 2560 3115 "quote",

+1

Cargo.toml

reviewed

··· 3 3 members = [ 4 4 "links", 5 5 "constellation", 6 6 + "jetstream", 6 7 ]

+2 -1

Makefile

reviewed

··· 5 5 cargo test 6 6 7 7 fmt: 8 8 - cargo fmt --all 8 8 + cargo fmt --package links --package constellation 9 9 + cargo +nightly fmt --package jetstream 9 10 10 11 clippy: 11 12 cargo clippy --all-targets --all-features -- -D warnings

+8

jetstream/.gitignore

reviewed

··· 1 1 + target/ 2 2 + pkg/ 3 3 + **/*.rs.bk 4 4 + dist/ 5 5 + traces/ 6 6 + *.DS_Store 7 7 + .cargo/ 8 8 + .env

+35

jetstream/Cargo.toml

reviewed

··· 1 1 + [package] 2 2 + authors = ["videah <videah@selfish.systems>", "phil <phil@bad-example.com>"] 3 3 + name = "jetstream" 4 4 + version = "0.1.1" 5 5 + edition = "2021" 6 6 + license = "MIT" 7 7 + description = "Library for easily interacting with and consuming the Bluesky Jetstream service." 8 8 + repository = "https://github.com/at-microcosm/links" 9 9 + readme = "README.md" 10 10 + 11 11 + [dependencies] 12 12 + async-trait = "0.1.83" 13 13 + atrium-api = { version = "0.24.7", default-features = false, features = [ 14 14 + "namespace-appbsky", 15 15 + ] } 16 16 + tokio = { version = "1.41.1", features = ["full", "sync", "time"] } 17 17 + tokio-tungstenite = { version = "0.24.0", features = [ 18 18 + "connect", 19 19 + "native-tls", 20 20 + "url", 21 21 + ] } 22 22 + futures-util = "0.3.31" 23 23 + url = "2.5.4" 24 24 + serde = { version = "1.0.215", features = ["derive"] } 25 25 + serde_json = "1.0.132" 26 26 + chrono = "0.4.38" 27 27 + zstd = "0.13.2" 28 28 + thiserror = "2.0.3" 29 29 + flume = "0.11.1" 30 30 + log = "0.4.22" 31 31 + tokio-util = "0.7.13" 32 32 + 33 33 + [dev-dependencies] 34 34 + anyhow = "1.0.93" 35 35 + clap = { version = "4.5.20", features = ["derive"] }

+21

jetstream/LICENSE

reviewed

··· 1 1 + MIT License 2 2 + 3 3 + Copyright (c) 2024 videah 4 4 + 5 5 + Permission is hereby granted, free of charge, to any person obtaining a copy 6 6 + of this software and associated documentation files (the "Software"), to deal 7 7 + in the Software without restriction, including without limitation the rights 8 8 + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 9 9 + copies of the Software, and to permit persons to whom the Software is 10 10 + furnished to do so, subject to the following conditions: 11 11 + 12 12 + The above copyright notice and this permission notice shall be included in all 13 13 + copies or substantial portions of the Software. 14 14 + 15 15 + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 16 16 + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 17 17 + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 18 18 + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 19 19 + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 20 20 + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE 21 21 + SOFTWARE.

+69

jetstream/README.md

reviewed

··· 1 1 + # jetstream-oxide 2 2 + 3 3 + [](https://crates.io/crates/jetstream-oxide) 4 4 + [](https://docs.rs/jetstream-oxide/latest/jetstream_oxide) 5 5 + 6 6 + A typed Rust library for easily interacting with and consuming the 7 7 + Bluesky [Jetstream](https://github.com/bluesky-social/jetstream) 8 8 + service. 9 9 + 10 10 + ```rust 11 11 + let config = JetstreamConfig { 12 12 + endpoint: DefaultJetstreamEndpoints::USEastOne.into(), 13 13 + compression: JetstreamCompression::Zstd, 14 14 + ..Default::default() 15 15 + }; 16 16 + 17 17 + let jetstream = JetstreamConnector::new(config).unwrap(); 18 18 + let receiver = jetstream.connect().await?; 19 19 + 20 20 + while let Ok(event) = receiver.recv_async().await { 21 21 + if let Commit(commit) = event { 22 22 + match commit { 23 23 + CommitEvent::Create { info, commit } => { 24 24 + println!("Received create event: {:#?}", info); 25 25 + } 26 26 + CommitEvent::Update { info, commit } => { 27 27 + println!("Received update event: {:#?}", info); 28 28 + } 29 29 + CommitEvent::Delete { info, commit } => { 30 30 + println!("Received delete event: {:#?}", info); 31 31 + } 32 32 + } 33 33 + } 34 34 + } 35 35 + ``` 36 36 + 37 37 + ## Example 38 38 + 39 39 + A small example CLI utility to show how to use this crate can be found in the `examples` directory. To run it, use the 40 40 + following command: 41 41 + 42 42 + ```sh 43 43 + cargo run --example basic -- --nsid "app.bsky.feed.post" 44 44 + ``` 45 45 + 46 46 + This will display a real-time feed of every single post that is being made or deleted in the entire Bluesky network, 47 47 + right in your terminal! 48 48 + 49 49 + You can filter it down to just specific accounts like this: 50 50 + 51 51 + ```sh 52 52 + cargo run --example basic -- \ 53 53 + --nsid "app.bsky.feed.post" \ 54 54 + --did "did:plc:inze6wrmsm7pjl7yta3oig77" 55 55 + ``` 56 56 + 57 57 + This listens for posts that *I personally make*. You can substitute your own DID and make a few test posts yourself if 58 58 + you'd 59 59 + like of course! 60 60 + 61 61 + 62 62 + ## Running `rustfmt` (requires nightly) 63 63 + 64 64 + ```bash 65 65 + # get nightly set up 66 66 + rustup toolchain install nightly --allow-downgrade -c rustfmt 67 67 + # run the nightly version of fmt 68 68 + cargo +nightly fmt 69 69 + ```

+53

jetstream/examples/arbitrary_record.rs

reviewed

··· 1 1 + //! An example of how to listen for create/delete events on a specific DID and potentialy unknown 2 2 + //! NSID 3 3 + 4 4 + use atrium_api::types::string; 5 5 + use clap::Parser; 6 6 + use jetstream::{ 7 7 + events::{ 8 8 + commit::CommitEvent, 9 9 + JetstreamEvent::Commit, 10 10 + }, 11 11 + DefaultJetstreamEndpoints, 12 12 + JetstreamCompression, 13 13 + JetstreamConfig, 14 14 + JetstreamConnector, 15 15 + }; 16 16 + 17 17 + #[derive(Parser, Debug)] 18 18 + #[command(version, about, long_about = None)] 19 19 + struct Args { 20 20 + /// The DIDs to listen for events on, if not provided we will listen for all DIDs. 21 21 + #[arg(short, long)] 22 22 + did: Option<Vec<string::Did>>, 23 23 + /// The NSID for the collection to listen for (e.g. `blue.flashes.feed.post`). 24 24 + #[arg(short, long)] 25 25 + nsid: string::Nsid, 26 26 + } 27 27 + 28 28 + #[tokio::main] 29 29 + async fn main() -> anyhow::Result<()> { 30 30 + let args = Args::parse(); 31 31 + 32 32 + let dids = args.did.unwrap_or_default(); 33 33 + let config: JetstreamConfig<serde_json::Value> = JetstreamConfig { 34 34 + endpoint: DefaultJetstreamEndpoints::USEastOne.into(), 35 35 + wanted_collections: vec![args.nsid.clone()], 36 36 + wanted_dids: dids.clone(), 37 37 + compression: JetstreamCompression::Zstd, 38 38 + ..Default::default() 39 39 + }; 40 40 + 41 41 + let jetstream: JetstreamConnector<serde_json::Value> = JetstreamConnector::new(config)?; 42 42 + let receiver = jetstream.connect().await?; 43 43 + 44 44 + println!("Listening for '{}' events on DIDs: {:?}", &*args.nsid, dids); 45 45 + 46 46 + while let Ok(event) = receiver.recv_async().await { 47 47 + if let Commit(CommitEvent::Create { commit, .. }) = event { 48 48 + println!("got record: {:?}", commit.record); 49 49 + } 50 50 + } 51 51 + 52 52 + Ok(()) 53 53 + }

+68

jetstream/examples/basic.rs

reviewed

··· 1 1 + //! A very basic example of how to listen for create/delete events on a specific DID and NSID. 2 2 + 3 3 + use atrium_api::{ 4 4 + record::KnownRecord::AppBskyFeedPost, 5 5 + types::string, 6 6 + }; 7 7 + use clap::Parser; 8 8 + use jetstream::{ 9 9 + events::{ 10 10 + commit::CommitEvent, 11 11 + JetstreamEvent::Commit, 12 12 + }, 13 13 + DefaultJetstreamEndpoints, 14 14 + JetstreamCompression, 15 15 + JetstreamConfig, 16 16 + JetstreamConnector, 17 17 + }; 18 18 + 19 19 + #[derive(Parser, Debug)] 20 20 + #[command(version, about, long_about = None)] 21 21 + struct Args { 22 22 + /// The DIDs to listen for events on, if not provided we will listen for all DIDs. 23 23 + #[arg(short, long)] 24 24 + did: Option<Vec<string::Did>>, 25 25 + /// The NSID for the collection to listen for (e.g. `app.bsky.feed.post`). 26 26 + #[arg(short, long)] 27 27 + nsid: string::Nsid, 28 28 + } 29 29 + 30 30 + #[tokio::main] 31 31 + async fn main() -> anyhow::Result<()> { 32 32 + let args = Args::parse(); 33 33 + 34 34 + let dids = args.did.unwrap_or_default(); 35 35 + let config = JetstreamConfig { 36 36 + endpoint: DefaultJetstreamEndpoints::USEastOne.into(), 37 37 + wanted_collections: vec![args.nsid.clone()], 38 38 + wanted_dids: dids.clone(), 39 39 + compression: JetstreamCompression::Zstd, 40 40 + ..Default::default() 41 41 + }; 42 42 + 43 43 + let jetstream = JetstreamConnector::new(config)?; 44 44 + let receiver = jetstream.connect().await?; 45 45 + 46 46 + println!("Listening for '{}' events on DIDs: {:?}", &*args.nsid, dids); 47 47 + 48 48 + while let Ok(event) = receiver.recv_async().await { 49 49 + if let Commit(commit) = event { 50 50 + match commit { 51 51 + CommitEvent::Create { info: _, commit } => { 52 52 + if let AppBskyFeedPost(record) = commit.record { 53 53 + println!( 54 54 + "New post created! ({})\n\n'{}'", 55 55 + commit.info.rkey, record.text 56 56 + ); 57 57 + } 58 58 + } 59 59 + CommitEvent::Delete { info: _, commit } => { 60 60 + println!("A post has been deleted. ({})", commit.rkey); 61 61 + } 62 62 + _ => {} 63 63 + } 64 64 + } 65 65 + } 66 66 + 67 67 + Ok(()) 68 68 + }

+6

jetstream/rustfmt.toml

reviewed

··· 1 1 + group_imports = "StdExternalCrate" 2 2 + comment_width = 100 3 3 + format_code_in_doc_comments = true 4 4 + imports_granularity = "Crate" 5 5 + imports_layout = "Vertical" 6 6 + wrap_comments = true

+43

jetstream/src/error.rs

reviewed

··· 1 1 + //! Various error types. 2 2 + use std::io; 3 3 + 4 4 + use thiserror::Error; 5 5 + 6 6 + /// Possible errors that can occur when a [JetstreamConfig](crate::JetstreamConfig) that is passed 7 7 + /// to a [JetstreamConnector](crate::JetstreamConnector) is invalid. 8 8 + #[derive(Error, Debug)] 9 9 + pub enum ConfigValidationError { 10 10 + #[error("too many wanted collections: {0} > 100")] 11 11 + TooManyWantedCollections(usize), 12 12 + #[error("too many wanted DIDs: {0} > 10,000")] 13 13 + TooManyDids(usize), 14 14 + } 15 15 + 16 16 + /// Possible errors that can occur in the process of connecting to a Jetstream instance over 17 17 + /// WebSockets. 18 18 + /// 19 19 + /// See [JetstreamConnector::connect](crate::JetstreamConnector::connect). 20 20 + #[derive(Error, Debug)] 21 21 + pub enum ConnectionError { 22 22 + #[error("invalid endpoint: {0}")] 23 23 + InvalidEndpoint(#[from] url::ParseError), 24 24 + #[error("failed to connect to Jetstream instance: {0}")] 25 25 + WebSocketFailure(#[from] tokio_tungstenite::tungstenite::Error), 26 26 + #[error("the Jetstream config is invalid (this really should not happen here): {0}")] 27 27 + InvalidConfig(#[from] ConfigValidationError), 28 28 + } 29 29 + 30 30 + /// Possible errors that can occur when receiving events from a Jetstream instance over WebSockets. 31 31 + /// 32 32 + /// See [websocket_task](crate::websocket_task). 33 33 + #[derive(Error, Debug)] 34 34 + pub enum JetstreamEventError { 35 35 + #[error("received websocket message that could not be deserialized as JSON: {0}")] 36 36 + ReceivedMalformedJSON(#[from] serde_json::Error), 37 37 + #[error("failed to load built-in zstd dictionary for decoding: {0}")] 38 38 + CompressionDictionaryError(io::Error), 39 39 + #[error("failed to decode zstd-compressed message: {0}")] 40 40 + CompressionDecoderError(io::Error), 41 41 + #[error("all receivers were dropped but the websocket connection failed to close cleanly")] 42 42 + WebSocketCloseFailure, 43 43 + }

+40

jetstream/src/events/account.rs

reviewed

··· 1 1 + use chrono::Utc; 2 2 + use serde::Deserialize; 3 3 + 4 4 + use crate::{ 5 5 + events::EventInfo, 6 6 + exports, 7 7 + }; 8 8 + 9 9 + /// An event representing a change to an account. 10 10 + #[derive(Deserialize, Debug)] 11 11 + pub struct AccountEvent { 12 12 + /// Basic metadata included with every event. 13 13 + #[serde(flatten)] 14 14 + pub info: EventInfo, 15 15 + /// Account specific data bundled with this event. 16 16 + pub account: AccountData, 17 17 + } 18 18 + 19 19 + /// Account specific data bundled with an account event. 20 20 + #[derive(Deserialize, Debug)] 21 21 + pub struct AccountData { 22 22 + /// Whether the account is currently active. 23 23 + pub active: bool, 24 24 + /// The DID of the account. 25 25 + pub did: exports::Did, 26 26 + pub seq: u64, 27 27 + pub time: chrono::DateTime<Utc>, 28 28 + /// If `active` is `false` this will be present to explain why the account is inactive. 29 29 + pub status: Option<AccountStatus>, 30 30 + } 31 31 + 32 32 + /// The possible reasons an account might be listed as inactive. 33 33 + #[derive(Deserialize, Debug)] 34 34 + #[serde(rename_all = "lowercase")] 35 35 + pub enum AccountStatus { 36 36 + Deactivated, 37 37 + Deleted, 38 38 + Suspended, 39 39 + TakenDown, 40 40 + }

+60

jetstream/src/events/commit.rs

reviewed

··· 1 1 + use serde::Deserialize; 2 2 + 3 3 + use crate::{ 4 4 + events::EventInfo, 5 5 + exports, 6 6 + }; 7 7 + 8 8 + /// An event representing a repo commit, which can be a `create`, `update`, or `delete` operation. 9 9 + #[derive(Deserialize, Debug)] 10 10 + #[serde(untagged, rename_all = "snake_case")] 11 11 + pub enum CommitEvent<R> { 12 12 + Create { 13 13 + #[serde(flatten)] 14 14 + info: EventInfo, 15 15 + commit: CommitData<R>, 16 16 + }, 17 17 + Update { 18 18 + #[serde(flatten)] 19 19 + info: EventInfo, 20 20 + commit: CommitData<R>, 21 21 + }, 22 22 + Delete { 23 23 + #[serde(flatten)] 24 24 + info: EventInfo, 25 25 + commit: CommitInfo, 26 26 + }, 27 27 + } 28 28 + 29 29 + /// The type of commit operation that was performed. 30 30 + #[derive(Deserialize, Debug)] 31 31 + #[serde(rename_all = "snake_case")] 32 32 + pub enum CommitType { 33 33 + Create, 34 34 + Update, 35 35 + Delete, 36 36 + } 37 37 + 38 38 + /// Basic commit specific info bundled with every event, also the only data included with a `delete` 39 39 + /// operation. 40 40 + #[derive(Deserialize, Debug)] 41 41 + pub struct CommitInfo { 42 42 + /// The type of commit operation that was performed. 43 43 + pub operation: CommitType, 44 44 + pub rev: String, 45 45 + pub rkey: String, 46 46 + /// The NSID of the record type that this commit is associated with. 47 47 + pub collection: exports::Nsid, 48 48 + } 49 49 + 50 50 + /// Detailed data bundled with a commit event. This data is only included when the event is 51 51 + /// `create` or `update`. 52 52 + #[derive(Deserialize, Debug)] 53 53 + pub struct CommitData<R> { 54 54 + #[serde(flatten)] 55 55 + pub info: CommitInfo, 56 56 + /// The CID of the record that was operated on. 57 57 + pub cid: exports::Cid, 58 58 + /// The record that was operated on. 59 59 + pub record: R, 60 60 + }

+28

jetstream/src/events/identity.rs

reviewed

··· 1 1 + use chrono::Utc; 2 2 + use serde::Deserialize; 3 3 + 4 4 + use crate::{ 5 5 + events::EventInfo, 6 6 + exports, 7 7 + }; 8 8 + 9 9 + /// An event representing a change to an identity. 10 10 + #[derive(Deserialize, Debug)] 11 11 + pub struct IdentityEvent { 12 12 + /// Basic metadata included with every event. 13 13 + #[serde(flatten)] 14 14 + pub info: EventInfo, 15 15 + /// Identity specific data bundled with this event. 16 16 + pub identity: IdentityData, 17 17 + } 18 18 + 19 19 + /// Identity specific data bundled with an identity event. 20 20 + #[derive(Deserialize, Debug)] 21 21 + pub struct IdentityData { 22 22 + /// The DID of the identity. 23 23 + pub did: exports::Did, 24 24 + /// The handle associated with the identity. 25 25 + pub handle: Option<exports::Handle>, 26 26 + pub seq: u64, 27 27 + pub time: chrono::DateTime<Utc>, 28 28 + }

+31

jetstream/src/events/mod.rs

reviewed

··· 1 1 + pub mod account; 2 2 + pub mod commit; 3 3 + pub mod identity; 4 4 + 5 5 + use serde::Deserialize; 6 6 + 7 7 + use crate::exports; 8 8 + 9 9 + /// Basic data that is included with every event. 10 10 + #[derive(Deserialize, Debug)] 11 11 + pub struct EventInfo { 12 12 + pub did: exports::Did, 13 13 + pub time_us: u64, 14 14 + pub kind: EventKind, 15 15 + } 16 16 + 17 17 + #[derive(Deserialize, Debug)] 18 18 + #[serde(untagged)] 19 19 + pub enum JetstreamEvent<R> { 20 20 + Commit(commit::CommitEvent<R>), 21 21 + Identity(identity::IdentityEvent), 22 22 + Account(account::AccountEvent), 23 23 + } 24 24 + 25 25 + #[derive(Deserialize, Debug)] 26 26 + #[serde(rename_all = "snake_case")] 27 27 + pub enum EventKind { 28 28 + Commit, 29 29 + Identity, 30 30 + Account, 31 31 + }

+8

jetstream/src/exports.rs

reviewed

··· 1 1 + //! Useful exports for third-party crates used by this project. 2 2 + 3 3 + pub use atrium_api::types::string::{ 4 4 + Cid, 5 5 + Did, 6 6 + Handle, 7 7 + Nsid, 8 8 + };

+415

jetstream/src/lib.rs

reviewed

··· 1 1 + pub mod error; 2 2 + pub mod events; 3 3 + pub mod exports; 4 4 + 5 5 + use std::{ 6 6 + io::{ 7 7 + Cursor, 8 8 + Read, 9 9 + }, 10 10 + marker::PhantomData, 11 11 + sync::Arc, 12 12 + time::{ 13 13 + Duration, 14 14 + Instant, 15 15 + }, 16 16 + }; 17 17 + 18 18 + use atrium_api::record::KnownRecord; 19 19 + use chrono::Utc; 20 20 + use futures_util::{ 21 21 + stream::StreamExt, 22 22 + SinkExt, 23 23 + }; 24 24 + use serde::de::DeserializeOwned; 25 25 + use tokio::{ 26 26 + net::TcpStream, 27 27 + sync::Mutex, 28 28 + }; 29 29 + use tokio_tungstenite::{ 30 30 + connect_async, 31 31 + tungstenite::Message, 32 32 + MaybeTlsStream, 33 33 + WebSocketStream, 34 34 + }; 35 35 + use tokio_util::sync::CancellationToken; 36 36 + use url::Url; 37 37 + use zstd::dict::DecoderDictionary; 38 38 + 39 39 + use crate::{ 40 40 + error::{ 41 41 + ConfigValidationError, 42 42 + ConnectionError, 43 43 + JetstreamEventError, 44 44 + }, 45 45 + events::JetstreamEvent, 46 46 + }; 47 47 + 48 48 + /// The Jetstream endpoints officially provided by Bluesky themselves. 49 49 + /// 50 50 + /// There are no guarantees that these endpoints will always be available, but you are free 51 51 + /// to run your own Jetstream instance in any case. 52 52 + pub enum DefaultJetstreamEndpoints { 53 53 + /// `jetstream1.us-east.bsky.network` 54 54 + USEastOne, 55 55 + /// `jetstream2.us-east.bsky.network` 56 56 + USEastTwo, 57 57 + /// `jetstream1.us-west.bsky.network` 58 58 + USWestOne, 59 59 + /// `jetstream2.us-west.bsky.network` 60 60 + USWestTwo, 61 61 + } 62 62 + 63 63 + impl From<DefaultJetstreamEndpoints> for String { 64 64 + fn from(endpoint: DefaultJetstreamEndpoints) -> Self { 65 65 + match endpoint { 66 66 + DefaultJetstreamEndpoints::USEastOne => { 67 67 + "wss://jetstream1.us-east.bsky.network/subscribe".to_owned() 68 68 + } 69 69 + DefaultJetstreamEndpoints::USEastTwo => { 70 70 + "wss://jetstream2.us-east.bsky.network/subscribe".to_owned() 71 71 + } 72 72 + DefaultJetstreamEndpoints::USWestOne => { 73 73 + "wss://jetstream1.us-west.bsky.network/subscribe".to_owned() 74 74 + } 75 75 + DefaultJetstreamEndpoints::USWestTwo => { 76 76 + "wss://jetstream2.us-west.bsky.network/subscribe".to_owned() 77 77 + } 78 78 + } 79 79 + } 80 80 + } 81 81 + 82 82 + /// The maximum number of wanted collections that can be requested on a single Jetstream connection. 83 83 + const MAX_WANTED_COLLECTIONS: usize = 100; 84 84 + /// The maximum number of wanted DIDs that can be requested on a single Jetstream connection. 85 85 + const MAX_WANTED_DIDS: usize = 10_000; 86 86 + 87 87 + /// The custom `zstd` dictionary used for decoding compressed Jetstream messages. 88 88 + /// 89 89 + /// Sourced from the [official Bluesky Jetstream repo.](https://github.com/bluesky-social/jetstream/tree/main/pkg/models) 90 90 + const JETSTREAM_ZSTD_DICTIONARY: &[u8] = include_bytes!("../zstd/dictionary"); 91 91 + 92 92 + /// A receiver channel for consuming Jetstream events. 93 93 + pub type JetstreamReceiver<R> = flume::Receiver<JetstreamEvent<R>>; 94 94 + 95 95 + /// An internal sender channel for sending Jetstream events to [JetstreamReceiver]'s. 96 96 + type JetstreamSender<R> = flume::Sender<JetstreamEvent<R>>; 97 97 + 98 98 + /// A wrapper connector type for working with a WebSocket connection to a Jetstream instance to 99 99 + /// receive and consume events. See [JetstreamConnector::connect] for more info. 100 100 + pub struct JetstreamConnector<R: DeserializeOwned> { 101 101 + /// The configuration for the Jetstream connection. 102 102 + config: JetstreamConfig<R>, 103 103 + } 104 104 + 105 105 + pub enum JetstreamCompression { 106 106 + /// No compression, just raw plaintext JSON. 107 107 + None, 108 108 + /// Use the `zstd` compression algorithm, which can result in a ~56% smaller messages on 109 109 + /// average. See [here](https://github.com/bluesky-social/jetstream?tab=readme-ov-file#compression) for more info. 110 110 + Zstd, 111 111 + } 112 112 + 113 113 + impl From<JetstreamCompression> for bool { 114 114 + fn from(compression: JetstreamCompression) -> Self { 115 115 + match compression { 116 116 + JetstreamCompression::None => false, 117 117 + JetstreamCompression::Zstd => true, 118 118 + } 119 119 + } 120 120 + } 121 121 + 122 122 + pub struct JetstreamConfig<R: DeserializeOwned = KnownRecord> { 123 123 + /// A Jetstream endpoint to connect to with a WebSocket Scheme i.e. 124 124 + /// `wss://jetstream1.us-east.bsky.network/subscribe`. 125 125 + pub endpoint: String, 126 126 + /// A list of collection [NSIDs](https://atproto.com/specs/nsid) to filter events for. 127 127 + /// 128 128 + /// An empty list will receive events for *all* collections. 129 129 + /// 130 130 + /// Regardless of desired collections, all subscribers receive 131 131 + /// [AccountEvent](events::account::AccountEvent) and 132 132 + /// [IdentityEvent](events::identity::Identity) events. 133 133 + pub wanted_collections: Vec<exports::Nsid>, 134 134 + /// A list of repo [DIDs](https://atproto.com/specs/did) to filter events for. 135 135 + /// 136 136 + /// An empty list will receive events for *all* repos, which is a lot of events! 137 137 + pub wanted_dids: Vec<exports::Did>, 138 138 + /// The compression algorithm to request and use for the WebSocket connection (if any). 139 139 + pub compression: JetstreamCompression, 140 140 + /// An optional timestamp to begin playback from. 141 141 + /// 142 142 + /// An absent cursor or a cursor from the future will result in live-tail operation. 143 143 + /// 144 144 + /// When reconnecting, use the time_us from your most recently processed event and maybe 145 145 + /// provide a negative buffer (i.e. subtract a few seconds) to ensure gapless playback. 146 146 + pub cursor: Option<chrono::DateTime<Utc>>, 147 147 + /// Marker for record deserializable type. 148 148 + /// 149 149 + /// See examples/arbitrary_record.rs for an example using serde_json::Value 150 150 + /// 151 151 + /// You can omit this if you construct `JetstreamConfig { a: b, ..Default::default() }. 152 152 + /// If you have to specify it, use `std::marker::PhantomData` with no type parameters. 153 153 + pub record_type: PhantomData<R>, 154 154 + } 155 155 + 156 156 + impl<R: DeserializeOwned> Default for JetstreamConfig<R> { 157 157 + fn default() -> Self { 158 158 + JetstreamConfig { 159 159 + endpoint: DefaultJetstreamEndpoints::USEastOne.into(), 160 160 + wanted_collections: Vec::new(), 161 161 + wanted_dids: Vec::new(), 162 162 + compression: JetstreamCompression::None, 163 163 + cursor: None, 164 164 + record_type: PhantomData, 165 165 + } 166 166 + } 167 167 + } 168 168 + 169 169 + impl<R: DeserializeOwned> JetstreamConfig<R> { 170 170 + /// Constructs a new endpoint URL with the given [JetstreamConfig] applied. 171 171 + pub fn construct_endpoint(&self, endpoint: &str) -> Result<Url, url::ParseError> { 172 172 + let did_search_query = self 173 173 + .wanted_dids 174 174 + .iter() 175 175 + .map(|s| ("wantedDids", s.to_string())); 176 176 + 177 177 + let collection_search_query = self 178 178 + .wanted_collections 179 179 + .iter() 180 180 + .map(|s| ("wantedCollections", s.to_string())); 181 181 + 182 182 + let compression = ( 183 183 + "compress", 184 184 + match self.compression { 185 185 + JetstreamCompression::None => "false".to_owned(), 186 186 + JetstreamCompression::Zstd => "true".to_owned(), 187 187 + }, 188 188 + ); 189 189 + 190 190 + let cursor = self 191 191 + .cursor 192 192 + .map(|c| ("cursor", c.timestamp_micros().to_string())); 193 193 + 194 194 + let params = did_search_query 195 195 + .chain(collection_search_query) 196 196 + .chain(std::iter::once(compression)) 197 197 + .chain(cursor) 198 198 + .collect::<Vec<(&str, String)>>(); 199 199 + 200 200 + Url::parse_with_params(endpoint, params) 201 201 + } 202 202 + 203 203 + /// Validates the configuration to make sure it is within the limits of the Jetstream API. 204 204 + /// 205 205 + /// # Constants 206 206 + /// The following constants are used to validate the configuration and should only be changed 207 207 + /// if the Jetstream API has itself changed. 208 208 + /// - [MAX_WANTED_COLLECTIONS] 209 209 + /// - [MAX_WANTED_DIDS] 210 210 + pub fn validate(&self) -> Result<(), ConfigValidationError> { 211 211 + let collections = self.wanted_collections.len(); 212 212 + let dids = self.wanted_dids.len(); 213 213 + 214 214 + if collections > MAX_WANTED_COLLECTIONS { 215 215 + return Err(ConfigValidationError::TooManyWantedCollections(collections)); 216 216 + } 217 217 + 218 218 + if dids > MAX_WANTED_DIDS { 219 219 + return Err(ConfigValidationError::TooManyDids(dids)); 220 220 + } 221 221 + 222 222 + Ok(()) 223 223 + } 224 224 + } 225 225 + 226 226 + impl<R: DeserializeOwned + Send + 'static> JetstreamConnector<R> { 227 227 + /// Create a Jetstream connector with a valid [JetstreamConfig]. 228 228 + /// 229 229 + /// After creation, you can call [connect] to connect to the provided Jetstream instance. 230 230 + pub fn new(config: JetstreamConfig<R>) -> Result<Self, ConfigValidationError> { 231 231 + // We validate the configuration here so any issues are caught early. 232 232 + config.validate()?; 233 233 + Ok(JetstreamConnector { config }) 234 234 + } 235 235 + 236 236 + /// Connects to a Jetstream instance as defined in the [JetstreamConfig]. 237 237 + /// 238 238 + /// A [JetstreamReceiver] is returned which can be used to respond to events. When all instances 239 239 + /// of this receiver are dropped, the connection and task are automatically closed. 240 240 + pub async fn connect(&self) -> Result<JetstreamReceiver<R>, ConnectionError> { 241 241 + // We validate the config again for good measure. Probably not necessary but it can't hurt. 242 242 + self.config 243 243 + .validate() 244 244 + .map_err(ConnectionError::InvalidConfig)?; 245 245 + 246 246 + // TODO: Run some benchmarks and look into using a bounded channel instead. 247 247 + let (send_channel, receive_channel) = flume::unbounded(); 248 248 + 249 249 + let configured_endpoint = self 250 250 + .config 251 251 + .construct_endpoint(&self.config.endpoint) 252 252 + .map_err(ConnectionError::InvalidEndpoint)?; 253 253 + 254 254 + tokio::task::spawn(async move { 255 255 + let max_retries = 30; 256 256 + let base_delay_ms = 1_000; // 1 second 257 257 + let max_delay_ms = 30_000; // 30 seconds 258 258 + let success_threshold_s = 15; // 15 seconds, retry count is reset if we were connected at least this long 259 259 + 260 260 + let mut retry_attempt = 0; 261 261 + loop { 262 262 + let dict = DecoderDictionary::copy(JETSTREAM_ZSTD_DICTIONARY); 263 263 + 264 264 + retry_attempt += 1; 265 265 + if let Ok((ws_stream, _)) = connect_async(&configured_endpoint).await { 266 266 + let t_connected = Instant::now(); 267 267 + if let Err(e) = websocket_task(dict, ws_stream, send_channel.clone()).await { 268 268 + log::error!("Jetstream closed after encountering error: {e:?}"); 269 269 + } else { 270 270 + log::error!("Jetstream connection closed cleanly"); 271 271 + } 272 272 + if t_connected.elapsed() > Duration::from_secs(success_threshold_s) { 273 273 + retry_attempt = 0; 274 274 + continue; 275 275 + } 276 276 + } 277 277 + 278 278 + if retry_attempt >= max_retries { 279 279 + break; 280 280 + } 281 281 + 282 282 + // Exponential backoff 283 283 + let delay_ms = base_delay_ms * (2_u64.pow(retry_attempt)); 284 284 + 285 285 + log::error!("Connection failed, retrying in {delay_ms}ms..."); 286 286 + tokio::time::sleep(Duration::from_millis(delay_ms.min(max_delay_ms))).await; 287 287 + log::info!("Attempting to reconnect...") 288 288 + } 289 289 + log::error!("Connection retries exhausted. Jetstream is disconnected."); 290 290 + }); 291 291 + 292 292 + Ok(receive_channel) 293 293 + } 294 294 + } 295 295 + 296 296 + /// The main task that handles the WebSocket connection and sends [JetstreamEvent]'s to any 297 297 + /// receivers that are listening for them. 298 298 + async fn websocket_task<R: DeserializeOwned>( 299 299 + dictionary: DecoderDictionary<'_>, 300 300 + ws: WebSocketStream<MaybeTlsStream<TcpStream>>, 301 301 + send_channel: JetstreamSender<R>, 302 302 + ) -> Result<(), JetstreamEventError> { 303 303 + // TODO: Use the write half to allow the user to change configuration settings on the fly. 304 304 + let (socket_write, mut socket_read) = ws.split(); 305 305 + let shared_socket_write = Arc::new(Mutex::new(socket_write)); 306 306 + 307 307 + let ping_cancellation_token = CancellationToken::new(); 308 308 + let mut ping_interval = tokio::time::interval(Duration::from_secs(30)); 309 309 + let ping_cancelled = ping_cancellation_token.clone(); 310 310 + let ping_shared_socket_write = shared_socket_write.clone(); 311 311 + tokio::spawn(async move { 312 312 + loop { 313 313 + ping_interval.tick().await; 314 314 + let false = ping_cancelled.is_cancelled() else { 315 315 + break; 316 316 + }; 317 317 + log::trace!("Sending ping"); 318 318 + match ping_shared_socket_write 319 319 + .lock() 320 320 + .await 321 321 + .send(Message::Ping("ping".as_bytes().to_vec())) 322 322 + .await 323 323 + { 324 324 + Ok(_) => (), 325 325 + Err(error) => { 326 326 + log::error!("Ping failed: {error}"); 327 327 + break; 328 328 + } 329 329 + } 330 330 + } 331 331 + }); 332 332 + 333 333 + let mut closing_connection = false; 334 334 + loop { 335 335 + match socket_read.next().await { 336 336 + Some(Ok(message)) => { 337 337 + match message { 338 338 + Message::Text(json) => { 339 339 + let event = serde_json::from_str(&json) 340 340 + .map_err(JetstreamEventError::ReceivedMalformedJSON)?; 341 341 + 342 342 + if send_channel.send(event).is_err() { 343 343 + // We can assume that all receivers have been dropped, so we can close 344 344 + // the connection and exit the task. 345 345 + log::info!( 346 346 + "All receivers for the Jetstream connection have been dropped, closing connection." 347 347 + ); 348 348 + closing_connection = true; 349 349 + } 350 350 + } 351 351 + Message::Binary(zstd_json) => { 352 352 + let mut cursor = Cursor::new(zstd_json); 353 353 + let mut decoder = zstd::stream::Decoder::with_prepared_dictionary( 354 354 + &mut cursor, 355 355 + &dictionary, 356 356 + ) 357 357 + .map_err(JetstreamEventError::CompressionDictionaryError)?; 358 358 + 359 359 + let mut json = String::new(); 360 360 + decoder 361 361 + .read_to_string(&mut json) 362 362 + .map_err(JetstreamEventError::CompressionDecoderError)?; 363 363 + 364 364 + let event = serde_json::from_str(&json) 365 365 + .map_err(JetstreamEventError::ReceivedMalformedJSON)?; 366 366 + 367 367 + if send_channel.send(event).is_err() { 368 368 + // We can assume that all receivers have been dropped, so we can close 369 369 + // the connection and exit the task. 370 370 + log::info!( 371 371 + "All receivers for the Jetstream connection have been dropped, closing connection..." 372 372 + ); 373 373 + closing_connection = true; 374 374 + } 375 375 + } 376 376 + Message::Ping(vec) => { 377 377 + log::trace!("Ping recieved, responding"); 378 378 + _ = shared_socket_write 379 379 + .lock() 380 380 + .await 381 381 + .send(Message::Pong(vec)) 382 382 + .await; 383 383 + } 384 384 + Message::Close(close_frame) => { 385 385 + if let Some(close_frame) = close_frame { 386 386 + let reason = close_frame.reason; 387 387 + let code = close_frame.code; 388 388 + log::trace!("Connection closed. Reason: {reason}, Code: {code}"); 389 389 + } 390 390 + } 391 391 + Message::Pong(pong) => { 392 392 + let pong_payload = 393 393 + String::from_utf8(pong).unwrap_or("Invalid payload".to_string()); 394 394 + log::trace!("Pong recieved. Payload: {pong_payload}"); 395 395 + } 396 396 + Message::Frame(_) => (), 397 397 + } 398 398 + } 399 399 + Some(Err(error)) => { 400 400 + log::error!("Web socket error: {error}"); 401 401 + ping_cancellation_token.cancel(); 402 402 + closing_connection = true; 403 403 + } 404 404 + None => { 405 405 + log::error!("No web socket result"); 406 406 + ping_cancellation_token.cancel(); 407 407 + closing_connection = true; 408 408 + } 409 409 + } 410 410 + if closing_connection { 411 411 + _ = shared_socket_write.lock().await.close().await; 412 412 + return Ok(()); 413 413 + } 414 414 + } 415 415 + }

jetstream/zstd/dictionary

reviewed

This is a binary file and will not be displayed.