merge part 5 draft and part 4 updates
This commit is contained in:
commit
36e6270d61
|
|
@ -39,9 +39,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "aho-corasick"
|
||||
version = "0.7.18"
|
||||
version = "0.7.19"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1e37cfd5e7657ada45f742d6e99ca5788580b5c529dc78faf11ece6dc702656f"
|
||||
checksum = "b4f55bd91a0978cbfd91c457a164bab8b4001c833b7f323132c0a4e1922dd44e"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
|
@ -101,9 +101,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "async-global-executor"
|
||||
version = "2.2.0"
|
||||
version = "2.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5262ed948da60dd8956c6c5aca4d4163593dddb7b32d73267c93dab7b2e98940"
|
||||
checksum = "0da5b41ee986eed3f524c380e6d64965aea573882a8907682ad100f7859305ca"
|
||||
dependencies = [
|
||||
"async-channel",
|
||||
"async-executor",
|
||||
|
|
@ -111,16 +111,16 @@ dependencies = [
|
|||
"async-lock",
|
||||
"blocking",
|
||||
"futures-lite",
|
||||
"num_cpus",
|
||||
"once_cell",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "async-io"
|
||||
version = "1.7.0"
|
||||
version = "1.9.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e5e18f61464ae81cde0a23e713ae8fd299580c54d697a35820cfd0625b8b0e07"
|
||||
checksum = "83e21f3a490c72b3b0cf44962180e60045de2925d8dff97918f7ee43c8f637c7"
|
||||
dependencies = [
|
||||
"autocfg",
|
||||
"concurrent-queue",
|
||||
"futures-lite",
|
||||
"libc",
|
||||
|
|
@ -145,11 +145,12 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "async-process"
|
||||
version = "1.4.0"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cf2c06e30a24e8c78a3987d07f0930edf76ef35e027e7bdb063fccafdad1f60c"
|
||||
checksum = "02111fd8655a613c25069ea89fc8d9bb89331fa77486eb3bc059ee757cfa481c"
|
||||
dependencies = [
|
||||
"async-io",
|
||||
"autocfg",
|
||||
"blocking",
|
||||
"cfg-if 1.0.0",
|
||||
"event-listener",
|
||||
|
|
@ -305,9 +306,9 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
|
|||
|
||||
[[package]]
|
||||
name = "block-buffer"
|
||||
version = "0.10.2"
|
||||
version = "0.10.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0bf7fe51849ea569fd452f37822f606a5cabb684dc918707a0193fd4664ff324"
|
||||
checksum = "69cce20737498f97b993470a6e536b8523f0af7892a4f928cceb1ac5e52ebe7e"
|
||||
dependencies = [
|
||||
"generic-array",
|
||||
]
|
||||
|
|
@ -337,9 +338,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "bumpalo"
|
||||
version = "3.10.0"
|
||||
version = "3.11.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "37ccbd214614c6783386c1af30caf03192f17891059cecc394b4fb119e363de3"
|
||||
checksum = "c1ad822118d20d2c234f427000d5acc36eabe1e29a348c89b63dd60b13f28e5d"
|
||||
|
||||
[[package]]
|
||||
name = "bytes"
|
||||
|
|
@ -455,9 +456,9 @@ checksum = "5827cebf4670468b8772dd191856768aedcb1b0278a04f989f7766351917b9dc"
|
|||
|
||||
[[package]]
|
||||
name = "cpufeatures"
|
||||
version = "0.2.2"
|
||||
version = "0.2.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "59a6001667ab124aebae2a495118e11d30984c3a653e99d86d58971708cf5e4b"
|
||||
checksum = "28d997bd5e24a5928dd43e46dc529867e207907fe0b239c3477d924f7f2ca320"
|
||||
dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
|
@ -574,9 +575,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "either"
|
||||
version = "1.7.0"
|
||||
version = "1.8.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3f107b87b6afc2a64fd13cac55fe06d6c8859f12d4b14cbcdd2c67d0976781be"
|
||||
checksum = "90e5c1c8368803113bf0c9584fc495a58b86dc8a29edbf8fe877d21d9507e797"
|
||||
|
||||
[[package]]
|
||||
name = "encoding_rs"
|
||||
|
|
@ -671,9 +672,9 @@ checksum = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7"
|
|||
|
||||
[[package]]
|
||||
name = "futures"
|
||||
version = "0.3.21"
|
||||
version = "0.3.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f73fe65f54d1e12b726f517d3e2135ca3125a437b6d998caf1962961f7172d9e"
|
||||
checksum = "7f21eda599937fba36daeb58a22e8f5cee2d14c4a17b5b7739c7c8e5e3b8230c"
|
||||
dependencies = [
|
||||
"futures-channel",
|
||||
"futures-core",
|
||||
|
|
@ -686,9 +687,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "futures-channel"
|
||||
version = "0.3.21"
|
||||
version = "0.3.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c3083ce4b914124575708913bca19bfe887522d6e2e6d0952943f5eac4a74010"
|
||||
checksum = "30bdd20c28fadd505d0fd6712cdfcb0d4b5648baf45faef7f852afb2399bb050"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
"futures-sink",
|
||||
|
|
@ -696,15 +697,15 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "futures-core"
|
||||
version = "0.3.21"
|
||||
version = "0.3.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0c09fd04b7e4073ac7156a9539b57a484a8ea920f79c7c675d05d289ab6110d3"
|
||||
checksum = "4e5aa3de05362c3fb88de6531e6296e85cde7739cccad4b9dfeeb7f6ebce56bf"
|
||||
|
||||
[[package]]
|
||||
name = "futures-executor"
|
||||
version = "0.3.21"
|
||||
version = "0.3.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9420b90cfa29e327d0429f19be13e7ddb68fa1cccb09d65e5706b8c7a749b8a6"
|
||||
checksum = "9ff63c23854bee61b6e9cd331d523909f238fc7636290b96826e9cfa5faa00ab"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
"futures-task",
|
||||
|
|
@ -713,9 +714,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "futures-io"
|
||||
version = "0.3.21"
|
||||
version = "0.3.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fc4045962a5a5e935ee2fdedaa4e08284547402885ab326734432bed5d12966b"
|
||||
checksum = "bbf4d2a7a308fd4578637c0b17c7e1c7ba127b8f6ba00b29f717e9655d85eb68"
|
||||
|
||||
[[package]]
|
||||
name = "futures-lite"
|
||||
|
|
@ -734,9 +735,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "futures-macro"
|
||||
version = "0.3.21"
|
||||
version = "0.3.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "33c1e13800337f4d4d7a316bf45a567dbcb6ffe087f16424852d97e97a91f512"
|
||||
checksum = "42cd15d1c7456c04dbdf7e88bcd69760d74f3a798d6444e16974b505b0e62f17"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
|
|
@ -745,21 +746,21 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "futures-sink"
|
||||
version = "0.3.21"
|
||||
version = "0.3.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "21163e139fa306126e6eedaf49ecdb4588f939600f0b1e770f4205ee4b7fa868"
|
||||
checksum = "21b20ba5a92e727ba30e72834706623d94ac93a725410b6a6b6fbc1b07f7ba56"
|
||||
|
||||
[[package]]
|
||||
name = "futures-task"
|
||||
version = "0.3.21"
|
||||
version = "0.3.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "57c66a976bf5909d801bbef33416c41372779507e7a6b3a5e25e4749c58f776a"
|
||||
checksum = "a6508c467c73851293f390476d4491cf4d227dbabcd4170f3bb6044959b294f1"
|
||||
|
||||
[[package]]
|
||||
name = "futures-util"
|
||||
version = "0.3.21"
|
||||
version = "0.3.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d8b7abd5d659d9b90c8cba917f6ec750a74e2dc23902ef9cd4cc8c8b22e6036a"
|
||||
checksum = "44fb6cb1be61cc1d2e43b262516aafcf63b241cffdb1d3fa115f91d9c7b09c90"
|
||||
dependencies = [
|
||||
"futures-channel",
|
||||
"futures-core",
|
||||
|
|
@ -908,9 +909,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "h2"
|
||||
version = "0.3.13"
|
||||
version = "0.3.14"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "37a82c6d637fc9515a4694bbf1cb2457b79d81ce52b3108bdeea58b07dd34a57"
|
||||
checksum = "5ca32592cf21ac7ccab1825cd87f6c9b3d9022c44d086172ed0966bec8af30be"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"fnv",
|
||||
|
|
@ -988,9 +989,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "httparse"
|
||||
version = "1.7.1"
|
||||
version = "1.8.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "496ce29bb5a52785b44e0f7ca2847ae0bb839c9bd28f69acac9b99d461c0c04c"
|
||||
checksum = "d897f394bad6a705d5f4104762e116a75639e470d80901eed05a860a95cb1904"
|
||||
|
||||
[[package]]
|
||||
name = "httpdate"
|
||||
|
|
@ -1209,9 +1210,9 @@ checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55"
|
|||
|
||||
[[package]]
|
||||
name = "libc"
|
||||
version = "0.2.129"
|
||||
version = "0.2.132"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "64de3cc433455c14174d42e554d4027ee631c4d046d43e3ecc6efc4636cdc7a7"
|
||||
checksum = "8371e4e5341c3a96db127eb2465ac681ced4c433e01dd0e938adbef26ba93ba5"
|
||||
|
||||
[[package]]
|
||||
name = "libsodium-sys"
|
||||
|
|
@ -1227,9 +1228,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "lock_api"
|
||||
version = "0.4.7"
|
||||
version = "0.4.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "327fa5b6a6940e4699ec49a9beae1ea4845c6bab9314e4f84ac68742139d8c53"
|
||||
checksum = "9f80bf5aacaf25cbfc8210d1cfb718f2bf3b11c4c54e5afe36c236853a8ec390"
|
||||
dependencies = [
|
||||
"autocfg",
|
||||
"scopeguard",
|
||||
|
|
@ -1632,10 +1633,11 @@ checksum = "1df8c4ec4b0627e53bdf214615ad287367e482558cf84b109250b37464dc03ae"
|
|||
|
||||
[[package]]
|
||||
name = "polling"
|
||||
version = "2.2.0"
|
||||
version = "2.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "685404d509889fade3e86fe3a5803bca2ec09b0c0778d5ada6ec8bf7a8de5259"
|
||||
checksum = "899b00b9c8ab553c743b3e11e87c5c7d423b2a2de229ba95b24a756344748011"
|
||||
dependencies = [
|
||||
"autocfg",
|
||||
"cfg-if 1.0.0",
|
||||
"libc",
|
||||
"log",
|
||||
|
|
@ -1926,18 +1928,18 @@ checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd"
|
|||
|
||||
[[package]]
|
||||
name = "serde"
|
||||
version = "1.0.143"
|
||||
version = "1.0.144"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "53e8e5d5b70924f74ff5c6d64d9a5acd91422117c60f48c4e07855238a254553"
|
||||
checksum = "0f747710de3dcd43b88c9168773254e809d8ddbdf9653b84e2554ab219f17860"
|
||||
dependencies = [
|
||||
"serde_derive",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_derive"
|
||||
version = "1.0.143"
|
||||
version = "1.0.144"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d3d8e8de557aee63c26b85b947f5e59b690d0454c753f3adeb5cd7835ab88391"
|
||||
checksum = "94ed3a816fb1d101812f83e789f888322c34e291f894f19590dc310963e87a00"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
|
|
@ -1946,9 +1948,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "serde_json"
|
||||
version = "1.0.83"
|
||||
version = "1.0.85"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "38dd04e3c8279e75b31ef29dbdceebfe5ad89f4d0937213c53f7d49d01b3d5a7"
|
||||
checksum = "e55a28e3aaef9d5ce0506d0a14dbba8054ddc7e499ef522dd8b26859ec9d4a44"
|
||||
dependencies = [
|
||||
"indexmap",
|
||||
"itoa",
|
||||
|
|
@ -1969,9 +1971,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "sha2"
|
||||
version = "0.10.2"
|
||||
version = "0.10.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "55deaec60f81eefe3cce0dc50bda92d6d8e88f2a27df7c5033b42afeb1ed2676"
|
||||
checksum = "cf9db03534dff993187064c4e0c05a5708d2a9728ace9a8959b77bedf415dac5"
|
||||
dependencies = [
|
||||
"cfg-if 1.0.0",
|
||||
"cpufeatures",
|
||||
|
|
@ -2038,9 +2040,9 @@ checksum = "2fd0db749597d91ff862fd1d55ea87f7855a744a8425a64695b6fca237d1dad1"
|
|||
|
||||
[[package]]
|
||||
name = "socket2"
|
||||
version = "0.4.4"
|
||||
version = "0.4.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "66d72b759436ae32898a2af0a14218dbf55efde3feeb170eb623637db85ee1e0"
|
||||
checksum = "02e2d2db9033d13a1567121ddd7a095ee144db4e1ca1b1bda3419bc0da294ebd"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"winapi 0.3.9",
|
||||
|
|
@ -2125,18 +2127,18 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "thiserror"
|
||||
version = "1.0.32"
|
||||
version = "1.0.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f5f6586b7f764adc0231f4c79be7b920e766bb2f3e51b3661cdb263828f19994"
|
||||
checksum = "8c1b05ca9d106ba7d2e31a9dab4a64e7be2cce415321966ea3132c49a656e252"
|
||||
dependencies = [
|
||||
"thiserror-impl",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "thiserror-impl"
|
||||
version = "1.0.32"
|
||||
version = "1.0.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "12bafc5b54507e0149cdf1b145a5d80ab80a90bcd9275df43d4fff68460f6c21"
|
||||
checksum = "e8f2591983642de85c921015f3f070c665a197ed69e417af436115e3a1407487"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
|
|
@ -2154,9 +2156,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "time"
|
||||
version = "0.3.13"
|
||||
version = "0.3.14"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "db76ff9fa4b1458b3c7f077f3ff9887394058460d21e634355b273aaf11eea45"
|
||||
checksum = "3c3f9a28b618c3a6b9251b6908e9c99e04b9e5c02e6581ccbb67d59c34ef7f9b"
|
||||
dependencies = [
|
||||
"itoa",
|
||||
"libc",
|
||||
|
|
@ -2172,9 +2174,9 @@ checksum = "42657b1a6f4d817cda8e7a0ace261fe0cc946cf3a80314390b22cc61ae080792"
|
|||
|
||||
[[package]]
|
||||
name = "tokio"
|
||||
version = "1.20.1"
|
||||
version = "1.21.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7a8325f63a7d4774dd041e363b2409ed1c5cbbd0f867795e661df066b2b0a581"
|
||||
checksum = "89797afd69d206ccd11fb0ea560a44bbb87731d020670e79416d442919257d42"
|
||||
dependencies = [
|
||||
"autocfg",
|
||||
"bytes",
|
||||
|
|
@ -2317,9 +2319,9 @@ checksum = "dcf81ac59edc17cc8697ff311e8f5ef2d99fcbd9817b34cec66f90b6c3dfd987"
|
|||
|
||||
[[package]]
|
||||
name = "ubyte"
|
||||
version = "0.10.2"
|
||||
version = "0.10.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a58e29f263341a29bb79e14ad7fda5f63b1c7e48929bad4c685d7876b1d04e94"
|
||||
checksum = "c81f0dae7d286ad0d9366d7679a77934cfc3cf3a8d67e82669794412b2368fe6"
|
||||
dependencies = [
|
||||
"serde",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -5,4 +5,5 @@ members = [
|
|||
"part_2_subscribe_form",
|
||||
#"part_3_database_follows",
|
||||
#"part_4_posts_streams"
|
||||
#"part_5_task_loop"
|
||||
]
|
||||
|
|
|
|||
|
|
@ -12,7 +12,11 @@ Author: [@glyph](https://mycelial.technology/)
|
|||
- [Part 2: Subscription Form and Key Validation](https://git.coopcloud.tech/glyph/lykin_tutorial/src/branch/main/part_2_subscribe_form)
|
||||
- Part 3: Database and Follows
|
||||
- Part 4: Posts and Message Streams
|
||||
- ...
|
||||
- Part 5: Task Loop
|
||||
- Part 6: Display Peers and Update UI
|
||||
- Part 7: Fetch Latest Posts
|
||||
- Part 8: Read, Unread and Delete Posts
|
||||
- Part 9: Extension Ideas and Conclusion
|
||||
|
||||
## Links
|
||||
|
||||
|
|
|
|||
|
|
@ -6,11 +6,15 @@ edition = "2021"
|
|||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||
|
||||
[dependencies]
|
||||
async-std = "1.10"
|
||||
bincode = "1.3"
|
||||
chrono = "0.4"
|
||||
futures = "0.3"
|
||||
golgi = { git = "https://git.coopcloud.tech/golgi-ssb/golgi.git" }
|
||||
log = "0.4"
|
||||
rocket = "0.5.0-rc.1"
|
||||
rocket_dyn_templates = { version = "0.1.0-rc.1", features = ["tera"] }
|
||||
serde = "1"
|
||||
serde_json = "1"
|
||||
sled = "0.34"
|
||||
xdg = "2.4.1"
|
||||
|
|
|
|||
|
|
@ -17,6 +17,15 @@ Here's what we'll tackle in this fourth part of the series:
|
|||
- Add a post to the database
|
||||
- Add a post batch to the database
|
||||
|
||||
### Libraries
|
||||
|
||||
The following libraries are introduced in this part:
|
||||
|
||||
- [async-std](https://crates.io/crates/async-std)
|
||||
- [chrono](https://crates.io/crates/chrono)
|
||||
- [futures](https://crates.io/crates/futures)
|
||||
- [serde_json](https://crates.io/crates/serde_json)
|
||||
|
||||
### Create a Post Data Structure
|
||||
|
||||
We'll begin by creating a `Post` struct to store data about each Scuttlebutt post we want to render in our application. The fields of our struct will diverge from the fields we expect in a Scuttlebutt post-type message. Open `src/db.rs` and add the following code (I've included code comments to further define each field):
|
||||
|
|
@ -130,6 +139,16 @@ Now that we have the ability to obtain all messages authored by a specific peer,
|
|||
`src/sbot.rs`
|
||||
|
||||
```rust
|
||||
use chrono::NaiveDateTime;
|
||||
use golgi::{
|
||||
api::history_stream::CreateHistoryStream},
|
||||
messages::{SsbMessageContentType, SsbMessageKVT},
|
||||
sbot::GolgiError,
|
||||
};
|
||||
use serde_json::value::Value;
|
||||
|
||||
use crate::db::Post;
|
||||
|
||||
// Filter a stream of messages and return a vector of root posts.
|
||||
pub async fn get_root_posts(
|
||||
history_stream: impl futures::Stream<Item = Result<SsbMessageKVT, GolgiError>>,
|
||||
|
|
@ -279,6 +298,8 @@ On most occasions we'll find ourselves in a situation where we wish to add multi
|
|||
`src/db.rs`
|
||||
|
||||
```rust
|
||||
use sled::Batch;
|
||||
|
||||
impl Database {
|
||||
// ...
|
||||
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ use std::path::Path;
|
|||
|
||||
use log::{debug, info};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sled::{Db, IVec, Result, Tree};
|
||||
use sled::{Batch, Db, IVec, Result, Tree};
|
||||
|
||||
/// Scuttlebutt peer data.
|
||||
#[derive(Debug, Deserialize, Serialize)]
|
||||
|
|
@ -31,6 +31,49 @@ impl Peer {
|
|||
}
|
||||
}
|
||||
|
||||
/// The text and metadata of a Scuttlebutt root post.
|
||||
#[derive(Debug, Deserialize, Serialize)]
|
||||
pub struct Post {
|
||||
/// The key of the post-type message, also known as a message reference.
|
||||
pub key: String,
|
||||
/// The text of the post (may be formatted as markdown).
|
||||
pub text: String,
|
||||
/// The date the post was published (e.g. 17 May 2021).
|
||||
pub date: String,
|
||||
/// The sequence number of the post-type message.
|
||||
pub sequence: u64,
|
||||
/// The read state of the post; true if read, false if unread.
|
||||
pub read: bool,
|
||||
/// The timestamp representing the date the post was published.
|
||||
pub timestamp: i64,
|
||||
/// The subject of the post, represented as the first 53 characters of
|
||||
/// the post text.
|
||||
pub subject: Option<String>,
|
||||
}
|
||||
|
||||
impl Post {
|
||||
// Create a new instance of the Post struct. A default value of `false` is
|
||||
// set for `read`.
|
||||
pub fn new(
|
||||
key: String,
|
||||
text: String,
|
||||
date: String,
|
||||
sequence: u64,
|
||||
timestamp: i64,
|
||||
subject: Option<String>,
|
||||
) -> Post {
|
||||
Post {
|
||||
key,
|
||||
text,
|
||||
date,
|
||||
sequence,
|
||||
timestamp,
|
||||
subject,
|
||||
read: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// An instance of the key-value database and relevant trees.
|
||||
#[allow(dead_code)]
|
||||
#[derive(Clone)]
|
||||
|
|
@ -40,6 +83,9 @@ pub struct Database {
|
|||
/// A database tree containing Peer struct instances for all the peers
|
||||
/// we are subscribed to.
|
||||
peer_tree: Tree,
|
||||
/// A database tree containing Post struct instances for all of the posts
|
||||
/// we have downloaded from the peer to whom we subscribe.
|
||||
pub post_tree: Tree,
|
||||
}
|
||||
|
||||
impl Database {
|
||||
|
|
@ -55,8 +101,16 @@ impl Database {
|
|||
let peer_tree = db
|
||||
.open_tree("peers")
|
||||
.expect("Failed to open 'peers' database tree");
|
||||
debug!("Opening 'posts' database tree");
|
||||
let post_tree = db
|
||||
.open_tree("posts")
|
||||
.expect("Failed to open 'posts' database tree");
|
||||
|
||||
Database { db, peer_tree }
|
||||
Database {
|
||||
db,
|
||||
peer_tree,
|
||||
post_tree,
|
||||
}
|
||||
}
|
||||
|
||||
/// Add a peer to the database by inserting the public key into the peer
|
||||
|
|
@ -78,4 +132,33 @@ impl Database {
|
|||
debug!("Removing peer {} from 'peers' database tree", &public_key);
|
||||
self.peer_tree.remove(&public_key).map(|_| ())
|
||||
}
|
||||
|
||||
/// Add a post to the database by inserting an instance of the Post struct
|
||||
/// into the post tree.
|
||||
pub fn add_post(&self, public_key: &str, post: Post) -> Result<Option<IVec>> {
|
||||
let post_key = format!("{}_{}", public_key, post.key);
|
||||
debug!("Serializing post data for {} to bincode", &post_key);
|
||||
let post_bytes = bincode::serialize(&post).unwrap();
|
||||
|
||||
debug!("Inserting post {} into 'posts' database tree", &post_key);
|
||||
self.post_tree.insert(post_key.as_bytes(), post_bytes)
|
||||
}
|
||||
|
||||
/// Add a batch of posts to the database by inserting a vector of instances
|
||||
/// of the Post struct into the post tree.
|
||||
pub fn add_post_batch(&self, public_key: &str, posts: Vec<Post>) -> Result<()> {
|
||||
let mut post_batch = Batch::default();
|
||||
|
||||
for post in posts {
|
||||
let post_key = format!("{}_{}", public_key, post.key);
|
||||
debug!("Serializing post data for {} to bincode", &post_key);
|
||||
let post_bytes = bincode::serialize(&post).unwrap();
|
||||
|
||||
debug!("Inserting post {} into 'posts' database tree", &post_key);
|
||||
post_batch.insert(post_key.as_bytes(), post_bytes)
|
||||
}
|
||||
|
||||
debug!("Applying batch insertion into 'posts' database tree");
|
||||
self.post_tree.apply_batch(post_batch)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,17 @@
|
|||
use std::env;
|
||||
|
||||
use golgi::{api::friends::RelationshipQuery, sbot::Keystore, Sbot};
|
||||
use async_std::stream::StreamExt;
|
||||
use chrono::NaiveDateTime;
|
||||
use golgi::{
|
||||
api::{friends::RelationshipQuery, history_stream::CreateHistoryStream},
|
||||
messages::{SsbMessageContentType, SsbMessageKVT},
|
||||
sbot::Keystore,
|
||||
GolgiError, Sbot,
|
||||
};
|
||||
use log::{info, warn};
|
||||
use serde_json::value::Value;
|
||||
|
||||
use crate::db::Post;
|
||||
|
||||
/// Initialise a connection to a Scuttlebutt server.
|
||||
pub async fn init_sbot() -> Result<Sbot, String> {
|
||||
|
|
@ -53,16 +63,6 @@ pub async fn unfollow_peer(public_key: &str) -> Result<String, String> {
|
|||
sbot.unfollow(public_key).await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// Return the name (self-identifier) for the peer associated with the given
|
||||
/// public key.
|
||||
///
|
||||
/// The public key of the peer will be returned if a name is not found.
|
||||
pub async fn get_name(public_key: &str) -> Result<String, String> {
|
||||
let mut sbot = init_sbot().await?;
|
||||
|
||||
sbot.get_name(public_key).await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// Check the follow status of a remote peer and follow them if not already
|
||||
/// following.
|
||||
pub async fn follow_if_not_following(remote_peer: &str) -> Result<(), String> {
|
||||
|
|
@ -128,3 +128,86 @@ pub async fn unfollow_if_following(remote_peer: &str) -> Result<(), String> {
|
|||
Err(err_msg)
|
||||
}
|
||||
}
|
||||
|
||||
/// Return a stream of messages authored by the given public key.
|
||||
///
|
||||
/// This returns all messages regardless of type.
|
||||
pub async fn get_message_stream(
|
||||
public_key: &str,
|
||||
sequence_number: u64,
|
||||
) -> impl futures::Stream<Item = Result<SsbMessageKVT, GolgiError>> {
|
||||
let mut sbot = init_sbot().await.unwrap();
|
||||
|
||||
let history_stream_args = CreateHistoryStream::new(public_key.to_string())
|
||||
.keys_values(true, true)
|
||||
.after_seq(sequence_number);
|
||||
|
||||
sbot.create_history_stream(history_stream_args)
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
/// Return the name (self-identifier) for the peer associated with the given
|
||||
/// public key.
|
||||
///
|
||||
/// The public key of the peer will be returned if a name is not found.
|
||||
pub async fn get_name(public_key: &str) -> Result<String, String> {
|
||||
let mut sbot = init_sbot().await?;
|
||||
|
||||
sbot.get_name(public_key).await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// Filter a stream of messages and return a vector of root posts.
|
||||
///
|
||||
/// Each returned vector element includes the key of the post, the content
|
||||
/// text, the date the post was published, the sequence number of the post
|
||||
/// and whether it is read or unread.
|
||||
pub async fn get_root_posts(
|
||||
history_stream: impl futures::Stream<Item = Result<SsbMessageKVT, GolgiError>>,
|
||||
) -> (u64, Vec<Post>) {
|
||||
let mut latest_sequence = 0;
|
||||
let mut posts = Vec::new();
|
||||
|
||||
futures::pin_mut!(history_stream);
|
||||
|
||||
while let Some(res) = history_stream.next().await {
|
||||
match res {
|
||||
Ok(msg) => {
|
||||
if msg.value.is_message_type(SsbMessageContentType::Post) {
|
||||
let content = msg.value.content.to_owned();
|
||||
if let Value::Object(content_map) = content {
|
||||
if !content_map.contains_key("root") {
|
||||
latest_sequence = msg.value.sequence;
|
||||
|
||||
let text = match content_map.get_key_value("text") {
|
||||
Some(value) => value.1.to_string(),
|
||||
None => String::from(""),
|
||||
};
|
||||
let timestamp = msg.value.timestamp.round() as i64 / 1000;
|
||||
let datetime = NaiveDateTime::from_timestamp(timestamp, 0);
|
||||
let date = datetime.format("%d %b %Y").to_string();
|
||||
let subject = text.get(0..52).map(|s| s.to_string());
|
||||
|
||||
let post = Post::new(
|
||||
msg.key.to_owned(),
|
||||
text,
|
||||
date,
|
||||
msg.value.sequence,
|
||||
timestamp,
|
||||
subject,
|
||||
);
|
||||
|
||||
posts.push(post)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
// Print the `GolgiError` of this element to `stderr`.
|
||||
warn!("err: {:?}", err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
(latest_sequence, posts)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,20 @@
|
|||
[package]
|
||||
name = "part_5_task_loop"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
|
||||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||
|
||||
[dependencies]
|
||||
async-std = "1.10"
|
||||
bincode = "1.3"
|
||||
chrono = "0.4"
|
||||
futures = "0.3"
|
||||
golgi = { git = "https://git.coopcloud.tech/golgi-ssb/golgi.git" }
|
||||
log = "0.4"
|
||||
rocket = "0.5.0-rc.1"
|
||||
rocket_dyn_templates = { version = "0.1.0-rc.1", features = ["tera"] }
|
||||
serde = "1"
|
||||
serde_json = "1"
|
||||
sled = "0.34"
|
||||
xdg = "2.4.1"
|
||||
|
|
@ -0,0 +1,329 @@
|
|||
# lykin tutorial
|
||||
|
||||
## Part 5: Task Loop and Post Fetching
|
||||
|
||||
### Introduction
|
||||
|
||||
In the last installment we added support to our key-value database for dealing with Scuttlebutt posts and wrote code to create and filter streams of Scuttlebutt messages. Since our peers may have authored tens of thousands of messages, it's useful to create a way of fetching and filtering message streams as a background process. Today we'll do just that; writing a task loop that can be invoked from our web application route handlers and used to execute potentially long-running processes.
|
||||
|
||||
### Outline
|
||||
|
||||
Here's what we'll tackle in this fifth part of the series:
|
||||
|
||||
- Create an asynchronous task loop
|
||||
- Create a message passing channel and spawn the task loop
|
||||
- Write sbot-related task functions
|
||||
- Pass database instance into task loop
|
||||
- Fetch root posts on subscription
|
||||
|
||||
### Create an Asynchronous Task Loop
|
||||
|
||||
Let's start by defining a task type that enumerates the various tasks we might want to carry out. We'll create a separate module for our task loop:
|
||||
|
||||
`src/task_loop.rs`
|
||||
|
||||
```rust
|
||||
pub enum Task {
|
||||
Cancel,
|
||||
FetchAllPosts(String),
|
||||
FetchLatestName(String),
|
||||
}
|
||||
```
|
||||
|
||||
The `Task` enum is simple enough: we can fetch all the posts by a given peer (the `String` value is the public key of the peer we're interested in), fetch the latest name assigned to a peer or cancel the task loop.
|
||||
|
||||
We're going to use a message passing approach in order to trigger tasks inside the loop. Let's write the basic loop code now, adding it below the `Task` we just defined, while also adding the necessary crate imports:
|
||||
|
||||
```rust
|
||||
use async_std::{channel::Receiver, task};
|
||||
use log::info;
|
||||
|
||||
// Spawn an asynchronous loop which receives tasks over an unbounded channel
|
||||
// and invokes task functions accordingly.
|
||||
pub async fn spawn(rx: Receiver<Task>) {
|
||||
task::spawn(async move {
|
||||
while let Ok(task) = rx.recv().await {
|
||||
match task {
|
||||
// Fetch all messages authored by the given peer, filter
|
||||
// the root posts and insert them into the posts tree of the
|
||||
// database.
|
||||
Task::FetchAllPosts(peer_id) => {
|
||||
info!("Fetching all posts for peer: {}", peer_id);
|
||||
}
|
||||
// Fetch the latest name for the given peer and update the
|
||||
// peer entry in the peers tree of the database.
|
||||
Task::FetchLatestName(peer_id) => {
|
||||
info!("Fetching latest name for peer: {}", peer_id);
|
||||
}
|
||||
// Break out of the task loop.
|
||||
Task::Cancel => {
|
||||
info!("Exiting task loop...");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
```
|
||||
|
||||
The loop spawning function is fairly simple: it takes the receiver half of a channel and expects messages of type `Task` to be delivered; it matches on the `Task` variant each time a message is received on the channel and acts accordingly. Writing an async loop like this means that we can call functions without blocking the execution of the rest of our program. This is a particularly useful in route handlers where we want to be able to trigger a task and then immediately respond to the request in order to keep the UI snappy and responsive.
|
||||
|
||||
### Create Message Passing Channel and Spawn the Task Loop
|
||||
|
||||
Let's return to the root of our application to create the message passing channel, spawn the task loop and add the channel transmitter to managed state:
|
||||
|
||||
`src/main.rs`
|
||||
|
||||
```rust
|
||||
mod task_loop;
|
||||
|
||||
use async_std::channel;
|
||||
use log::info;
|
||||
use rocket::fairing::AdHoc;
|
||||
|
||||
use crate::task_loop::Task;
|
||||
|
||||
#[launch] async fn rocket() -> _ {
|
||||
// ...
|
||||
|
||||
// Create the key-value database.
|
||||
// ...
|
||||
|
||||
// Create a message passing channel.
|
||||
let (tx, rx) = channel::unbounded();
|
||||
let tx_clone = tx.clone();
|
||||
|
||||
// Spawn the task loop, passing in the receiver half of the channel.
|
||||
info!("Spawning task loop");
|
||||
task_loop::spawn(rx).await;
|
||||
|
||||
rocket::build()
|
||||
.manage(db)
|
||||
// Add the transmitter half of the channel to the managed state
|
||||
// of the Rocket application.
|
||||
.manage(tx)
|
||||
// ...
|
||||
// Send a task loop cancellation message when the application
|
||||
// is shutting down.
|
||||
.attach(AdHoc::on_shutdown("cancel task loop", |_| {
|
||||
Box::pin(async move {
|
||||
tx_clone.send(Task::Cancel).await.unwrap();
|
||||
})
|
||||
}))
|
||||
}
|
||||
```
|
||||
|
||||
Reviewing the code above: first an unbounded, asynchronous channel is created and split into transmitting (`tx`) and receiving (`rx`) ends, after which the transmitting channel is cloned. The task loop is then spawned and takes with it the receiving end of the channel. As we did previously with the `db` instance, the transmitting half of the channel is added to the managed state of the Rocket application; this will allow us to transmit tasks to the task loop from our web route handlers. Finally, a shutdown handler is attached to the Rocket application in order to send a cancellation task to the task loop before the program ends. This ensures that the task loop closes cleanly.
|
||||
|
||||
### Write Sbot-Related Task Functions
|
||||
|
||||
Before we can write the sbot-related task functions, we first need to add a method to our database code to allow the retrieval of data for a specific peer. Since we serialized the peer data as bincode before inserting it into the database, we need to deserialize the value after fetching it.
|
||||
|
||||
`src/db.rs`
|
||||
|
||||
```rust
|
||||
impl Database {
|
||||
// pub fn add_peer(&self, peer: Peer) -> Result<Option<IVec>> {
|
||||
// ...
|
||||
// }
|
||||
|
||||
// Get a single peer from the peer tree, defined by the given public key.
|
||||
// The byte value for the matching entry, if found, is deserialized from
|
||||
// bincode into an instance of the Peer struct.
|
||||
pub fn get_peer(&self, public_key: &str) -> Result<Option<Peer>> {
|
||||
debug!(
|
||||
"Retrieving peer data for {} from 'peers' database tree",
|
||||
&public_key
|
||||
);
|
||||
let peer = self
|
||||
.peer_tree
|
||||
.get(public_key.as_bytes())
|
||||
.unwrap()
|
||||
.map(|peer| {
|
||||
debug!("Deserializing peer data for {} from bincode", &public_key);
|
||||
bincode::deserialize(&peer).unwrap()
|
||||
});
|
||||
|
||||
Ok(peer)
|
||||
}
|
||||
|
||||
// ...
|
||||
}
|
||||
```
|
||||
|
||||
Now it's time to write the functions that will be executed when the `FetchAllPosts` and `FetchLatestName` tasks are invoked. These functions will be responsible for retrieving data from the sbot and updating the database with the latest values. We can keep our task loop neat and readable by separating this logic into functions:
|
||||
|
||||
`src/task_loop.rs`
|
||||
|
||||
```rust
|
||||
use log::warn;
|
||||
|
||||
use crate::{Database, sbot};
|
||||
|
||||
// Retrieve a set of posts from the local sbot instance and add them to the
|
||||
// posts tree of the database.
|
||||
//
|
||||
// A stream of messages is first requested for the peer represented by the
|
||||
// given public key (ID), starting after the given sequence number. The root
|
||||
// posts are filtered from the set of messages and added to the database as a
|
||||
// batch. Finally, the value of the latest sequence for the peer is updated
|
||||
// and saved to the existing database entry.
|
||||
async fn fetch_posts_and_update_db(db: &Database, peer_id: String, after_sequence: u64) {
|
||||
let peer_msgs = sbot::get_message_stream(&peer_id, after_sequence).await;
|
||||
let (_latest_sequence, root_posts) = sbot::get_root_posts(peer_msgs).await;
|
||||
|
||||
match db.add_post_batch(&peer_id, root_posts) {
|
||||
Ok(_) => {
|
||||
info!(
|
||||
"Inserted batch of posts into database post tree for peer: {}",
|
||||
&peer_id
|
||||
)
|
||||
}
|
||||
Err(e) => warn!(
|
||||
"Failed to insert batch of posts into database post tree for peer: {}: {}",
|
||||
&peer_id, e
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
// Request the name of the peer represented by the given public key (ID)
|
||||
// and update the existing entry in the database.
|
||||
async fn fetch_name_and_update_db(db: &Database, peer_id: String) {
|
||||
match sbot::get_name(&peer_id).await {
|
||||
Ok(name) => {
|
||||
if let Ok(Some(peer)) = db.get_peer(&peer_id) {
|
||||
let updated_peer = peer.set_name(&name);
|
||||
match db.add_peer(updated_peer) {
|
||||
Ok(_) => info!("Updated name for peer: {}", &peer_id),
|
||||
Err(e) => {
|
||||
warn!("Failed to update name for peer: {}: {}", &peer_id, e)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => warn!("Failed to fetch name for {}: {}", &peer_id, e),
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
These function calls can now be added to our task matching code in the task loop. Note that we also need to add the database instance as a parameter in the function isgnature:
|
||||
|
||||
`src/task_loop.rs`
|
||||
|
||||
```rust
|
||||
pub async fn spawn(db: Database, rx: Receiver<Task>) {
|
||||
task::spawn(async move {
|
||||
while let Ok(task) = rx.recv().await {
|
||||
match task {
|
||||
Task::FetchAllPosts(peer_id) => {
|
||||
info!("Fetching all posts for peer: {}", peer_id);
|
||||
fetch_posts_and_update_db(&db, peer_id, 0).await;
|
||||
}
|
||||
Task::FetchLatestName(peer_id) => {
|
||||
info!("Fetching latest name for peer: {}", peer_id);
|
||||
fetch_name_and_update_db(&db, peer_id).await;
|
||||
}
|
||||
Task::Cancel => {
|
||||
info!("Exiting task loop...");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
```
|
||||
|
||||
### Pass Database Instance Into Task Loop
|
||||
|
||||
As it currently stands, our code will fail to compile because `task_loop::spawn()` expects a database instance which has not yet been provided. We need to revisit the code in the root of our application to clone the database and pass it into the task loop:
|
||||
|
||||
`src/main.rs`
|
||||
|
||||
```rust
|
||||
#[launch]
|
||||
async fn rocket() -> _ {
|
||||
// ...
|
||||
let db = Database::init(&db_path);
|
||||
// Clone the database instance.
|
||||
let db_clone = db.clone();
|
||||
|
||||
// Create a message passing channel.
|
||||
let (tx, rx) = channel::unbounded();
|
||||
let tx_clone = tx.clone();
|
||||
|
||||
// Spawn the task loop.
|
||||
info!("Spawning task loop");
|
||||
// Pass the clone database instance and the rx channel into the task loop.
|
||||
task_loop::spawn(db_clone, rx).await;
|
||||
|
||||
// ...
|
||||
}
|
||||
```
|
||||
|
||||
### Fetch Root Posts on Subscription
|
||||
|
||||
Great, the task loop is primed and ready for action. We are very close to being able to initiate tasks from the route handler(s) of our web application. Earlier in this installment of the tutorial we created a message passing channel in `src.main.rs` and added the transmission end of the channel to the managed state of our Rocket instance. We need to add the transmitter as a parameter of the `subscribe_form` function before we can invoke tasks:
|
||||
|
||||
`src/routes.rs`
|
||||
|
||||
```rust
|
||||
use async_std::channel::Sender;
|
||||
|
||||
use crate::task_loop::Task;
|
||||
|
||||
#[post("/subscribe", data = "<peer>")]
|
||||
pub async fn subscribe_form(
|
||||
db: &State<Database>,
|
||||
tx: &State<Sender<Task>>,
|
||||
peer: Form<PeerForm>,
|
||||
) -> Result<Redirect, Flash<Redirect>> {
|
||||
info!("Subscribing to peer {}", &peer.public_key);
|
||||
// ...
|
||||
}
|
||||
```
|
||||
|
||||
Now, when a subscription event occurs (ie. the subscribe form is submitted with a peer ID), we can trigger a task to fetch all the root posts for that peer and add them to the key-value database. Note that I've omitted most of the code we've already written from the sample below. The most important three lines are those beginning with `if let Err(e) = tx.send...`.
|
||||
|
||||
```rust
|
||||
#[post("/subscribe", data = "<peer>")]
|
||||
pub async fn subscribe_form(
|
||||
db: &State<Database>,
|
||||
tx: &State<Sender<Task>>,
|
||||
peer: Form<PeerForm>,
|
||||
) -> Result<Redirect, Flash<Redirect>> {
|
||||
// ... {
|
||||
match sbot::follow_if_not_following(&peer.public_key).await {
|
||||
Ok(_) => {
|
||||
if db.add_peer(peer_info).is_ok() {
|
||||
// ...
|
||||
|
||||
// Fetch all root posts authored by the peer we're subscribing
|
||||
// to. Posts will be added to the key-value database.
|
||||
if let Err(e) = tx.send(Task::FetchAllPosts(peer_id)).await {
|
||||
warn!("Task loop error: {}", e)
|
||||
}
|
||||
} else {
|
||||
// ...
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
// ...
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(Redirect::to(uri!(home)))
|
||||
}
|
||||
```
|
||||
|
||||
### Conclusion
|
||||
|
||||
In this installment we wrote an asynchronous task loop and `Task` type to be able to execute background processes in our application. We created task variants and functions for two primary operations: 1. fetching all the root posts for a peer and adding them to the key-value database, and 2. fetching the latest name assigned to a peer. We created a message passing channel, passed the receiving end to the task loop and the transmitting end to the managed state of our web application, and invoked the fetch-all task from our subscription route handler.
|
||||
|
||||
The `Task` type and loop we wrote today can be easily extended by adding more variants. It's a part of the code we will return to in a future installment.
|
||||
|
||||
In the next tutorial installment we'll focus on updating the web interface. We'll add more templates to create a modular layout, write some CSS and populate a list of peers from the data in our key-value store. Soon the application will begin to take shape!
|
||||
|
||||
## Funding
|
||||
|
||||
This work has been funded by a Scuttlebutt Community Grant.
|
||||
|
|
@ -0,0 +1,184 @@
|
|||
use std::path::Path;
|
||||
|
||||
use log::{debug, info};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sled::{Batch, Db, IVec, Result, Tree};
|
||||
|
||||
/// Scuttlebutt peer data.
|
||||
#[derive(Debug, Deserialize, Serialize)]
|
||||
pub struct Peer {
|
||||
pub public_key: String,
|
||||
pub name: String,
|
||||
}
|
||||
|
||||
impl Peer {
|
||||
/// Create a new instance of the Peer struct using the given public
|
||||
/// key. A default value is set for name.
|
||||
pub fn new(public_key: &str) -> Peer {
|
||||
Peer {
|
||||
public_key: public_key.to_string(),
|
||||
name: "".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Modify the name field of an instance of the Peer struct, leaving
|
||||
/// the other values unchanged.
|
||||
pub fn set_name(self, name: &str) -> Peer {
|
||||
Self {
|
||||
name: name.to_string(),
|
||||
..self
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The text and metadata of a Scuttlebutt root post.
|
||||
#[derive(Debug, Deserialize, Serialize)]
|
||||
pub struct Post {
|
||||
/// The key of the post-type message, also known as a message reference.
|
||||
pub key: String,
|
||||
/// The text of the post (may be formatted as markdown).
|
||||
pub text: String,
|
||||
/// The date the post was published (e.g. 17 May 2021).
|
||||
pub date: String,
|
||||
/// The sequence number of the post-type message.
|
||||
pub sequence: u64,
|
||||
/// The read state of the post; true if read, false if unread.
|
||||
pub read: bool,
|
||||
/// The timestamp representing the date the post was published.
|
||||
pub timestamp: i64,
|
||||
/// The subject of the post, represented as the first 53 characters of
|
||||
/// the post text.
|
||||
pub subject: Option<String>,
|
||||
}
|
||||
|
||||
impl Post {
|
||||
// Create a new instance of the Post struct. A default value of `false` is
|
||||
// set for `read`.
|
||||
pub fn new(
|
||||
key: String,
|
||||
text: String,
|
||||
date: String,
|
||||
sequence: u64,
|
||||
timestamp: i64,
|
||||
subject: Option<String>,
|
||||
) -> Post {
|
||||
Post {
|
||||
key,
|
||||
text,
|
||||
date,
|
||||
sequence,
|
||||
timestamp,
|
||||
subject,
|
||||
read: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// An instance of the key-value database and relevant trees.
|
||||
#[allow(dead_code)]
|
||||
#[derive(Clone)]
|
||||
pub struct Database {
|
||||
/// The sled database instance.
|
||||
db: Db,
|
||||
/// A database tree containing Peer struct instances for all the peers
|
||||
/// we are subscribed to.
|
||||
peer_tree: Tree,
|
||||
/// A database tree containing Post struct instances for all of the posts
|
||||
/// we have downloaded from the peer to whom we subscribe.
|
||||
pub post_tree: Tree,
|
||||
}
|
||||
|
||||
impl Database {
|
||||
/// Initialise the database by opening the database file, loading the
|
||||
/// peers tree and returning an instantiated Database struct.
|
||||
pub fn init(path: &Path) -> Self {
|
||||
// Open the database at the given path.
|
||||
// The database will be created if it does not yet exist.
|
||||
// This code will panic if an IO error is encountered.
|
||||
info!("Initialising sled database");
|
||||
let db = sled::open(path).expect("Failed to open database");
|
||||
debug!("Opening 'peers' database tree");
|
||||
let peer_tree = db
|
||||
.open_tree("peers")
|
||||
.expect("Failed to open 'peers' database tree");
|
||||
debug!("Opening 'posts' database tree");
|
||||
let post_tree = db
|
||||
.open_tree("posts")
|
||||
.expect("Failed to open 'posts' database tree");
|
||||
|
||||
Database {
|
||||
db,
|
||||
peer_tree,
|
||||
post_tree,
|
||||
}
|
||||
}
|
||||
|
||||
/// Add a peer to the database by inserting the public key into the peer
|
||||
/// tree.
|
||||
pub fn add_peer(&self, peer: Peer) -> Result<Option<IVec>> {
|
||||
debug!("Serializing peer data for {} to bincode", &peer.public_key);
|
||||
let peer_bytes = bincode::serialize(&peer).unwrap();
|
||||
|
||||
debug!(
|
||||
"Inserting peer {} into 'peers' database tree",
|
||||
&peer.public_key
|
||||
);
|
||||
self.peer_tree.insert(&peer.public_key, peer_bytes)
|
||||
}
|
||||
|
||||
/// Get a single peer from the peer tree, defined by the given public key.
|
||||
/// The byte value for the matching entry, if found, is deserialized from
|
||||
/// bincode into an instance of the Peer struct.
|
||||
pub fn get_peer(&self, public_key: &str) -> Result<Option<Peer>> {
|
||||
debug!(
|
||||
"Retrieving peer data for {} from 'peers' database tree",
|
||||
&public_key
|
||||
);
|
||||
let peer = self
|
||||
.peer_tree
|
||||
.get(public_key.as_bytes())
|
||||
.unwrap()
|
||||
.map(|peer| {
|
||||
debug!("Deserializing peer data for {} from bincode", &public_key);
|
||||
bincode::deserialize(&peer).unwrap()
|
||||
});
|
||||
|
||||
Ok(peer)
|
||||
}
|
||||
|
||||
/// Remove a peer from the database, as represented by the given public
|
||||
/// key.
|
||||
pub fn remove_peer(&self, public_key: &str) -> Result<()> {
|
||||
debug!("Removing peer {} from 'peers' database tree", &public_key);
|
||||
self.peer_tree.remove(&public_key).map(|_| ())
|
||||
}
|
||||
|
||||
/// Add a post to the database by inserting an instance of the Post struct
|
||||
/// into the post tree.
|
||||
pub fn add_post(&self, public_key: &str, post: Post) -> Result<Option<IVec>> {
|
||||
let post_key = format!("{}_{}", public_key, post.key);
|
||||
debug!("Serializing post data for {} to bincode", &post_key);
|
||||
let post_bytes = bincode::serialize(&post).unwrap();
|
||||
|
||||
debug!("Inserting post {} into 'posts' database tree", &post_key);
|
||||
self.post_tree.insert(post_key.as_bytes(), post_bytes)
|
||||
}
|
||||
|
||||
/// Add a batch of posts to the database by inserting a vector of instances
|
||||
/// of the Post struct into the post tree.
|
||||
pub fn add_post_batch(&self, public_key: &str, posts: Vec<Post>) -> Result<()> {
|
||||
let mut post_batch = Batch::default();
|
||||
|
||||
for post in posts {
|
||||
let post_key = format!("{}_{}", public_key, post.key);
|
||||
debug!("Serializing post data for {} to bincode", &post_key);
|
||||
let post_bytes = bincode::serialize(&post).unwrap();
|
||||
|
||||
debug!("Inserting post {} into 'posts' database tree", &post_key);
|
||||
post_batch.insert(post_key.as_bytes(), post_bytes)
|
||||
}
|
||||
|
||||
debug!("Applying batch insertion into 'posts' database tree");
|
||||
self.post_tree.apply_batch(post_batch)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,43 @@
|
|||
mod db;
|
||||
mod routes;
|
||||
mod sbot;
|
||||
mod task_loop;
|
||||
mod utils;
|
||||
|
||||
use async_std::channel;
|
||||
use log::info;
|
||||
use rocket::{fairing::AdHoc, launch, routes};
|
||||
use rocket_dyn_templates::Template;
|
||||
use xdg::BaseDirectories;
|
||||
|
||||
use crate::{db::Database, routes::*, task_loop::Task};
|
||||
|
||||
#[launch]
|
||||
async fn rocket() -> _ {
|
||||
// Create the key-value database.
|
||||
let xdg_dirs = BaseDirectories::with_prefix("lykin").unwrap();
|
||||
let db_path = xdg_dirs
|
||||
.place_config_file("database")
|
||||
.expect("cannot create database directory");
|
||||
let db = Database::init(&db_path);
|
||||
let db_clone = db.clone();
|
||||
|
||||
// Create a message passing channel.
|
||||
let (tx, rx) = channel::unbounded();
|
||||
let tx_clone = tx.clone();
|
||||
|
||||
// Spawn the task loop, passing in the receiver half of the channel.
|
||||
info!("Spawning task loop");
|
||||
task_loop::spawn(db_clone, rx).await;
|
||||
|
||||
rocket::build()
|
||||
.manage(db)
|
||||
.manage(tx)
|
||||
.attach(Template::fairing())
|
||||
.mount("/", routes![home, subscribe_form, unsubscribe_form])
|
||||
.attach(AdHoc::on_shutdown("cancel task loop", |_| {
|
||||
Box::pin(async move {
|
||||
tx_clone.send(Task::Cancel).await.unwrap();
|
||||
})
|
||||
}))
|
||||
}
|
||||
|
|
@ -0,0 +1,122 @@
|
|||
use async_std::channel::Sender;
|
||||
use log::{info, warn};
|
||||
use rocket::{
|
||||
form::Form,
|
||||
get, post,
|
||||
request::FlashMessage,
|
||||
response::{Flash, Redirect},
|
||||
uri, FromForm, State,
|
||||
};
|
||||
use rocket_dyn_templates::{context, Template};
|
||||
|
||||
use crate::{
|
||||
db::{Database, Peer},
|
||||
sbot,
|
||||
task_loop::Task,
|
||||
utils,
|
||||
};
|
||||
|
||||
#[derive(FromForm)]
|
||||
pub struct PeerForm {
|
||||
pub public_key: String,
|
||||
}
|
||||
|
||||
#[get("/")]
|
||||
pub async fn home(flash: Option<FlashMessage<'_>>) -> Template {
|
||||
let whoami = match sbot::whoami().await {
|
||||
Ok(id) => id,
|
||||
Err(e) => format!("Error making `whoami` RPC call: {}. Please ensure the local go-sbot is running and refresh.", e),
|
||||
};
|
||||
|
||||
Template::render("base", context! { whoami: whoami, flash: flash })
|
||||
}
|
||||
|
||||
#[post("/subscribe", data = "<peer>")]
|
||||
pub async fn subscribe_form(
|
||||
db: &State<Database>,
|
||||
tx: &State<Sender<Task>>,
|
||||
peer: Form<PeerForm>,
|
||||
) -> Result<Redirect, Flash<Redirect>> {
|
||||
if let Err(e) = utils::validate_public_key(&peer.public_key) {
|
||||
let validation_err_msg = format!("Public key {} is invalid: {}", &peer.public_key, e);
|
||||
warn!("{}", validation_err_msg);
|
||||
return Err(Flash::error(Redirect::to(uri!(home)), validation_err_msg));
|
||||
} else {
|
||||
info!("Public key {} is valid", &peer.public_key);
|
||||
// Retrieve the name of the peer to which we are subscribing.
|
||||
let peer_name = match sbot::get_name(&peer.public_key).await {
|
||||
Ok(name) => name,
|
||||
Err(e) => {
|
||||
warn!("Failed to fetch name for peer {}: {}", &peer.public_key, e);
|
||||
// Return an empty string if an error occurs.
|
||||
String::from("")
|
||||
}
|
||||
};
|
||||
let peer_info = Peer::new(&peer.public_key).set_name(&peer_name);
|
||||
|
||||
match sbot::follow_if_not_following(&peer.public_key).await {
|
||||
Ok(_) => {
|
||||
// Add the peer to the database.
|
||||
if db.add_peer(peer_info).is_ok() {
|
||||
info!("Added {} to 'peers' database tree", &peer.public_key);
|
||||
let peer_id = peer.public_key.to_string();
|
||||
|
||||
// Fetch all root posts authored by the peer we're subscribing
|
||||
// to. Posts will be added to the key-value database.
|
||||
if let Err(e) = tx.send(Task::FetchAllPosts(peer_id)).await {
|
||||
warn!("Task loop error: {}", e)
|
||||
}
|
||||
} else {
|
||||
let err_msg = format!(
|
||||
"Failed to add peer {} to 'peers' database tree",
|
||||
&peer.public_key
|
||||
);
|
||||
warn!("{}", err_msg);
|
||||
return Err(Flash::error(Redirect::to(uri!(home)), err_msg));
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("{}", e);
|
||||
return Err(Flash::error(Redirect::to(uri!(home)), e));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(Redirect::to(uri!(home)))
|
||||
}
|
||||
|
||||
#[post("/unsubscribe", data = "<peer>")]
|
||||
pub async fn unsubscribe_form(
|
||||
db: &State<Database>,
|
||||
peer: Form<PeerForm>,
|
||||
) -> Result<Redirect, Flash<Redirect>> {
|
||||
if let Err(e) = utils::validate_public_key(&peer.public_key) {
|
||||
let validation_err_msg = format!("Public key {} is invalid: {}", &peer.public_key, e);
|
||||
warn!("{}", validation_err_msg);
|
||||
return Err(Flash::error(Redirect::to(uri!(home)), validation_err_msg));
|
||||
} else {
|
||||
info!("Public key {} is valid", &peer.public_key);
|
||||
match sbot::unfollow_if_following(&peer.public_key).await {
|
||||
Ok(_) => {
|
||||
// Remove the peer from the database.
|
||||
if db.remove_peer(&peer.public_key).is_ok() {
|
||||
info!(
|
||||
"Removed peer {} from 'peers' database tree",
|
||||
&peer.public_key
|
||||
);
|
||||
} else {
|
||||
warn!(
|
||||
"Failed to remove peer {} from 'peers' database tree",
|
||||
&peer.public_key
|
||||
);
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("{}", e);
|
||||
return Err(Flash::error(Redirect::to(uri!(home)), e));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(Redirect::to(uri!(home)))
|
||||
}
|
||||
|
|
@ -0,0 +1,213 @@
|
|||
use std::env;
|
||||
|
||||
use async_std::stream::StreamExt;
|
||||
use chrono::NaiveDateTime;
|
||||
use golgi::{
|
||||
api::{friends::RelationshipQuery, history_stream::CreateHistoryStream},
|
||||
messages::{SsbMessageContentType, SsbMessageKVT},
|
||||
sbot::Keystore,
|
||||
GolgiError, Sbot,
|
||||
};
|
||||
use log::{info, warn};
|
||||
use serde_json::value::Value;
|
||||
|
||||
use crate::db::Post;
|
||||
|
||||
/// Initialise a connection to a Scuttlebutt server.
|
||||
pub async fn init_sbot() -> Result<Sbot, String> {
|
||||
let go_sbot_port = env::var("GO_SBOT_PORT").unwrap_or_else(|_| "8021".to_string());
|
||||
|
||||
let keystore = Keystore::GoSbot;
|
||||
let ip_port = Some(format!("127.0.0.1:{}", go_sbot_port));
|
||||
let net_id = None;
|
||||
|
||||
Sbot::init(keystore, ip_port, net_id)
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// Return the public key of the local sbot instance.
|
||||
pub async fn whoami() -> Result<String, String> {
|
||||
let mut sbot = init_sbot().await?;
|
||||
|
||||
sbot.whoami().await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// Check follow status.
|
||||
///
|
||||
/// Is peer A (`public_key_a`) following peer B (`public_key_b`)?
|
||||
pub async fn is_following(public_key_a: &str, public_key_b: &str) -> Result<String, String> {
|
||||
let mut sbot = init_sbot().await?;
|
||||
|
||||
let query = RelationshipQuery {
|
||||
source: public_key_a.to_string(),
|
||||
dest: public_key_b.to_string(),
|
||||
};
|
||||
|
||||
sbot.friends_is_following(query)
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// Follow a peer.
|
||||
pub async fn follow_peer(public_key: &str) -> Result<String, String> {
|
||||
let mut sbot = init_sbot().await?;
|
||||
|
||||
sbot.follow(public_key).await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// Unfollow a peer.
|
||||
pub async fn unfollow_peer(public_key: &str) -> Result<String, String> {
|
||||
let mut sbot = init_sbot().await?;
|
||||
|
||||
sbot.unfollow(public_key).await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// Return the name (self-identifier) for the peer associated with the given
|
||||
/// public key.
|
||||
///
|
||||
/// The public key of the peer will be returned if a name is not found.
|
||||
pub async fn get_name(public_key: &str) -> Result<String, String> {
|
||||
let mut sbot = init_sbot().await?;
|
||||
|
||||
sbot.get_name(public_key).await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// Check the follow status of a remote peer and follow them if not already
|
||||
/// following.
|
||||
pub async fn follow_if_not_following(remote_peer: &str) -> Result<(), String> {
|
||||
if let Ok(whoami) = whoami().await {
|
||||
match is_following(&whoami, remote_peer).await {
|
||||
Ok(status) if status.as_str() == "false" => match follow_peer(remote_peer).await {
|
||||
Ok(_) => {
|
||||
info!("Followed peer {}", &remote_peer);
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => {
|
||||
let err_msg = format!("Failed to follow peer {}: {}", &remote_peer, e);
|
||||
warn!("{}", err_msg);
|
||||
Err(err_msg)
|
||||
}
|
||||
},
|
||||
Ok(status) if status.as_str() == "true" => {
|
||||
info!(
|
||||
"Already following peer {}. No further action taken",
|
||||
&remote_peer
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
_ => Err(
|
||||
"Failed to determine follow status: received unrecognised response from local sbot"
|
||||
.to_string(),
|
||||
),
|
||||
}
|
||||
} else {
|
||||
let err_msg = String::from("Received an error during `whoami` RPC call. Please ensure the go-sbot is running and try again");
|
||||
warn!("{}", err_msg);
|
||||
Err(err_msg)
|
||||
}
|
||||
}
|
||||
|
||||
/// Check the follow status of a remote peer and unfollow them if already
|
||||
/// following.
|
||||
pub async fn unfollow_if_following(remote_peer: &str) -> Result<(), String> {
|
||||
if let Ok(whoami) = whoami().await {
|
||||
match is_following(&whoami, remote_peer).await {
|
||||
Ok(status) if status.as_str() == "true" => {
|
||||
info!("Unfollowing peer {}", &remote_peer);
|
||||
match unfollow_peer(remote_peer).await {
|
||||
Ok(_) => {
|
||||
info!("Unfollowed peer {}", &remote_peer);
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => {
|
||||
let err_msg = format!("Failed to unfollow peer {}: {}", &remote_peer, e);
|
||||
warn!("{}", err_msg);
|
||||
Err(err_msg)
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => Err(
|
||||
"Failed to determine follow status: received unrecognised response from local sbot"
|
||||
.to_string(),
|
||||
),
|
||||
}
|
||||
} else {
|
||||
let err_msg = String::from("Received an error during `whoami` RPC call. Please ensure the go-sbot is running and try again");
|
||||
warn!("{}", err_msg);
|
||||
Err(err_msg)
|
||||
}
|
||||
}
|
||||
|
||||
/// Return a stream of messages authored by the given public key.
|
||||
///
|
||||
/// This returns all messages regardless of type.
|
||||
pub async fn get_message_stream(
|
||||
public_key: &str,
|
||||
sequence_number: u64,
|
||||
) -> impl futures::Stream<Item = Result<SsbMessageKVT, GolgiError>> {
|
||||
let mut sbot = init_sbot().await.unwrap();
|
||||
|
||||
let history_stream_args = CreateHistoryStream::new(public_key.to_string())
|
||||
.keys_values(true, true)
|
||||
.after_seq(sequence_number);
|
||||
|
||||
sbot.create_history_stream(history_stream_args)
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
/// Filter a stream of messages and return a vector of root posts.
|
||||
///
|
||||
/// Each returned vector element includes the key of the post, the content
|
||||
/// text, the date the post was published, the sequence number of the post
|
||||
/// and whether it is read or unread.
|
||||
pub async fn get_root_posts(
|
||||
history_stream: impl futures::Stream<Item = Result<SsbMessageKVT, GolgiError>>,
|
||||
) -> (u64, Vec<Post>) {
|
||||
let mut latest_sequence = 0;
|
||||
let mut posts = Vec::new();
|
||||
|
||||
futures::pin_mut!(history_stream);
|
||||
|
||||
while let Some(res) = history_stream.next().await {
|
||||
match res {
|
||||
Ok(msg) => {
|
||||
if msg.value.is_message_type(SsbMessageContentType::Post) {
|
||||
let content = msg.value.content.to_owned();
|
||||
if let Value::Object(content_map) = content {
|
||||
if !content_map.contains_key("root") {
|
||||
latest_sequence = msg.value.sequence;
|
||||
|
||||
let text = match content_map.get_key_value("text") {
|
||||
Some(value) => value.1.to_string(),
|
||||
None => String::from(""),
|
||||
};
|
||||
let timestamp = msg.value.timestamp.round() as i64 / 1000;
|
||||
let datetime = NaiveDateTime::from_timestamp(timestamp, 0);
|
||||
let date = datetime.format("%d %b %Y").to_string();
|
||||
let subject = text.get(0..52).map(|s| s.to_string());
|
||||
|
||||
let post = Post::new(
|
||||
msg.key.to_owned(),
|
||||
text,
|
||||
date,
|
||||
msg.value.sequence,
|
||||
timestamp,
|
||||
subject,
|
||||
);
|
||||
|
||||
posts.push(post)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
// Print the `GolgiError` of this element to `stderr`.
|
||||
warn!("err: {:?}", err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
(latest_sequence, posts)
|
||||
}
|
||||
|
|
@ -0,0 +1,76 @@
|
|||
use async_std::{channel::Receiver, task};
|
||||
use log::{info, warn};
|
||||
|
||||
use crate::{sbot, Database};
|
||||
|
||||
async fn fetch_posts_and_update_db(db: &Database, peer_id: String, after_sequence: u64) {
|
||||
let peer_msgs = sbot::get_message_stream(&peer_id, after_sequence).await;
|
||||
let (_latest_sequence, root_posts) = sbot::get_root_posts(peer_msgs).await;
|
||||
|
||||
match db.add_post_batch(&peer_id, root_posts) {
|
||||
Ok(_) => {
|
||||
info!(
|
||||
"Inserted batch of posts into database post tree for peer: {}",
|
||||
&peer_id
|
||||
)
|
||||
}
|
||||
Err(e) => warn!(
|
||||
"Failed to insert batch of posts into database post tree for peer: {}: {}",
|
||||
&peer_id, e
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
/// Request the name of the peer represented by the given public key (ID)
|
||||
/// and update the existing entry in the database.
|
||||
async fn fetch_name_and_update_db(db: &Database, peer_id: String) {
|
||||
match sbot::get_name(&peer_id).await {
|
||||
Ok(name) => {
|
||||
if let Ok(Some(peer)) = db.get_peer(&peer_id) {
|
||||
let updated_peer = peer.set_name(&name);
|
||||
match db.add_peer(updated_peer) {
|
||||
Ok(_) => info!("Updated name for peer: {}", &peer_id),
|
||||
Err(e) => {
|
||||
warn!("Failed to update name for peer: {}: {}", &peer_id, e)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => warn!("Failed to fetch name for {}: {}", &peer_id, e),
|
||||
}
|
||||
}
|
||||
|
||||
pub enum Task {
|
||||
Cancel,
|
||||
FetchAllPosts(String),
|
||||
FetchLatestName(String),
|
||||
}
|
||||
|
||||
/// Spawn an asynchronous loop which receives tasks over an unbounded channel
|
||||
/// and invokes task functions accordingly.
|
||||
pub async fn spawn(db: Database, rx: Receiver<Task>) {
|
||||
task::spawn(async move {
|
||||
while let Ok(task) = rx.recv().await {
|
||||
match task {
|
||||
// Fetch all messages authored by the given peer, filter
|
||||
// the root posts and insert them into the posts tree of the
|
||||
// database.
|
||||
Task::FetchAllPosts(peer_id) => {
|
||||
info!("Fetching all posts for peer: {}", peer_id);
|
||||
fetch_posts_and_update_db(&db, peer_id, 0).await;
|
||||
}
|
||||
// Fetch the latest name for the given peer and update the
|
||||
// peer entry in the peers tree of the database.
|
||||
Task::FetchLatestName(peer_id) => {
|
||||
info!("Fetching latest name for peer: {}", peer_id);
|
||||
fetch_name_and_update_db(&db, peer_id).await;
|
||||
}
|
||||
// Break out of the task loop.
|
||||
Task::Cancel => {
|
||||
info!("Exiting task loop...");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
|
@ -0,0 +1,32 @@
|
|||
//! Public key validation.
|
||||
|
||||
/// Ensure that the given public key is a valid ed25519 key.
|
||||
///
|
||||
/// Return an error string if the key is invalid.
|
||||
pub fn validate_public_key(public_key: &str) -> Result<(), String> {
|
||||
// Ensure the ID starts with the correct sigil link.
|
||||
if !public_key.starts_with('@') {
|
||||
return Err("expected '@' sigil as first character".to_string());
|
||||
}
|
||||
|
||||
// Find the dot index denoting the start of the algorithm definition tag.
|
||||
let dot_index = match public_key.rfind('.') {
|
||||
Some(index) => index,
|
||||
None => return Err("no dot index was found".to_string()),
|
||||
};
|
||||
|
||||
// Check the hashing algorithm (must end with ".ed25519").
|
||||
if !&public_key.ends_with(".ed25519") {
|
||||
return Err("hashing algorithm must be ed25519".to_string());
|
||||
}
|
||||
|
||||
// Obtain the base64 portion (substring) of the public key.
|
||||
let base64_str = &public_key[1..dot_index];
|
||||
|
||||
// Ensure the length of the base64 encoded ed25519 public key is correct.
|
||||
if base64_str.len() != 44 {
|
||||
return Err("base64 data length is incorrect".to_string());
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -0,0 +1,21 @@
|
|||
<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<title>lykin</title>
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||
</head>
|
||||
<body>
|
||||
<h1><a href="/">lykin</a></h1>
|
||||
<p>{{ whoami }}</p>
|
||||
<form action="/subscribe" method="post">
|
||||
<label for="public_key">Public Key</label>
|
||||
<input type="text" id="public_key" name="public_key" maxlength=53>
|
||||
<input type="submit" value="Subscribe">
|
||||
<input type="submit" value="Unsubscribe" formaction="/unsubscribe">
|
||||
</form>
|
||||
{% if flash and flash.kind == "error" %}
|
||||
<p style="color: red;">[ {{ flash.message }} ]</p>
|
||||
{% endif %}
|
||||
</body>
|
||||
</html>
|
||||
Loading…
Reference in New Issue