hashiverse_lib/client/
hashiverse_client.rs

1//! # Top-level client API
2//!
3//! The one thing every external consumer of `hashiverse-lib` ends up holding: a
4//! [`HashiverseClient`] instance. The server binary's cache role, the WASM browser client,
5//! the Python wrapper, and the integration-test harness all build on top of this struct.
6//!
7//! `HashiverseClient` wires together the rest of the `client` module — peer tracking,
8//! post/feedback bundle management, timelines, the meta-post system, the key locker — and
9//! exposes the small set of verbs a social-network client actually performs:
10//!
11//! - submit posts, submit feedback, fetch URL previews, fetch trending hashtags;
12//! - walk per-user / per-hashtag / per-mention timelines;
13//! - manage the logged-in account (via [`crate::client::key_locker`]);
14//! - read and update the profile ([`crate::client::meta_post`]).
15//!
16//! Every dependency that varies by environment (clock, network, PoW engine, storage, key
17//! locker) is injected through [`crate::tools::runtime_services::RuntimeServices`] and
18//! pluggable traits so the same struct works unchanged on native, in the browser (WASM),
19//! and in deterministic in-memory integration tests.
20
21use crate::client::args::Args;
22use crate::client::client_storage::client_storage::{ClientStorage, BUCKETS, BUCKET_TRIMS};
23use crate::client::key_locker::key_locker::KeyLocker;
24use crate::client::meta_post::meta_post::MetaPost;
25use crate::client::meta_post::meta_post_manager::MetaPostManager;
26use crate::client::peer_tracker::peer_tracker::PeerTracker;
27use crate::client::post_bundle::live_post_bundle_manager::LivePostBundleManager;
28use crate::client::post_bundle::post_bundle_manager::PostBundleManager;
29use crate::client::post_bundle::posting;
30use crate::client::timeline::recent_posts_pen::RecentPostsPen;
31use crate::client::timeline::single_timeline::SingleTimeline;
32use crate::protocol::posting::encoded_post::EncodedPostV1;
33use crate::protocol::posting::encoded_post_feedback::EncodedPostFeedbackV1;
34use crate::tools::buckets::{bucket_durations_for_type, generate_bucket_location, BucketLocation, BucketType};
35use crate::tools::client_id::ClientId;
36use crate::tools::runtime_services::RuntimeServices;
37use crate::tools::types::Id;
38use bytes::Bytes;
39use log::{error, info, trace, warn};
40use scraper::{Html, Selector};
41use std::sync::Arc;
42use tokio::sync::{RwLock, RwLockWriteGuard};
43use crate::anyhow_assert_eq;
44use crate::client::post_bundle::live_post_bundle_feedback_manager::LivePostBundleFeedbackManager;
45use crate::client::post_bundle::post_bundle_feedback_manager::PostBundleFeedbackManager;
46use crate::client::timeline::multiple_timeline::MultipleTimeline;
47use crate::protocol::payload::payload::{FetchUrlPreviewResponseV1, FetchUrlPreviewV1, PayloadRequestKind, PayloadResponseKind, SubmitPostCommitTokenV1, TrendingHashtagsFetchResponseV1, TrendingHashtagsFetchV1};
48use crate::protocol::peer::Peer;
49use crate::protocol::rpc;
50use crate::tools::config::CLIENT_FEEDBACK_POW_NUMERAIRE;
51use crate::tools::plain_text_post::convert_text_to_hashiverse_html;
52use crate::tools::tools;
53use crate::tools::time::TimeMillis;
54
55/// The top-level client API for participating in a hashiverse network.
56///
57/// `HashiverseClient` is what every external consumer — the server binary's cache role, the
58/// WASM browser client, the Python wrapper, integration tests — builds on top of. It
59/// orchestrates all the subsystems the client side needs:
60///
61/// - a [`PeerTracker`] for Kademlia-style peer discovery and gossip,
62/// - a [`LivePostBundleManager`] / [`LivePostBundleFeedbackManager`] pair for fetching post
63///   bundles and their feedback from the network (with on-disk caching),
64/// - a [`SingleTimeline`] / [`MultipleTimeline`] pair for recursive bucket traversal when
65///   reading a feed,
66/// - a [`MetaPostManager`] for the user's profile / follow list / settings,
67/// - a [`RecentPostsPen`] so freshly-authored posts appear in the user's own timelines
68///   immediately,
69/// - a [`ClientId`] / [`KeyLocker`] pair for the signed identity used on every outbound
70///   action.
71///
72/// All external dependencies — clock, network, PoW — are carried through a single
73/// [`RuntimeServices`], which is what makes the same code run inside the in-memory integration
74/// test harness, on a native server, and in a browser Web Worker without touching any of the
75/// high-level logic above.
76///
77/// Construct one via [`HashiverseClient::new`] and keep a single instance per logged-in
78/// account; the struct is cheap to share (`Arc<Self>`) and its internal state uses `RwLock`s
79/// for concurrent access.
80pub struct HashiverseClient {
81    runtime_services: Arc<RuntimeServices>,
82    client_storage: Arc<dyn ClientStorage>,
83    key_locker: Arc<dyn KeyLocker>,
84    post_bundle_manager: Arc<LivePostBundleManager>,
85    post_bundle_feedback_manager: Arc<LivePostBundleFeedbackManager>,
86    meta_post_manager: MetaPostManager,
87    client_id: ClientId,
88    peer_tracker: Arc<RwLock<PeerTracker>>,
89    recent_posts_pen: Arc<RwLock<RecentPostsPen>>,
90    single_timeline: Arc<RwLock<Option<SingleTimeline>>>,
91    multiple_timeline: Arc<RwLock<Option<MultipleTimeline>>>,
92}
93
94impl HashiverseClient {
95    pub async fn new(runtime_services: Arc<RuntimeServices>, client_storage: Arc<dyn ClientStorage>, key_locker: Arc<dyn KeyLocker>, _args: Args) -> anyhow::Result<Self> {
96        let client_id = key_locker.client_id().clone();
97        info!("client_id={}", client_id);
98
99        let peer_tracker = Arc::new(RwLock::new(PeerTracker::new(runtime_services.clone(), client_storage.clone()).await?));
100        let post_bundle_manager = Arc::new(LivePostBundleManager::new(runtime_services.clone(), client_id.id, client_storage.clone(), peer_tracker.clone()));
101        let post_bundle_feedback_manager = Arc::new(LivePostBundleFeedbackManager::new(runtime_services.clone(), client_id.id, client_storage.clone(), peer_tracker.clone()));
102
103        // Trim the various buckets
104        anyhow_assert_eq!(BUCKETS.len(), BUCKET_TRIMS.len(), "Mismatch in length between BUCKETS and BUCKET_TRIMS");
105        info!("Trimming buckets: {} buckets, {} trims", BUCKETS.len(), BUCKET_TRIMS.len());
106        for i in 0..BUCKETS.len() {
107            let trim = BUCKET_TRIMS[i];
108            if trim > 0 {
109                client_storage.trim(BUCKETS[i], trim).await?;
110            }
111        }
112
113        let meta_post_manager = MetaPostManager::new(runtime_services.clone(), client_storage.clone(), key_locker.clone(), client_id.clone());
114
115        Ok(Self {
116            runtime_services,
117            client_storage,
118            key_locker,
119            post_bundle_manager,
120            post_bundle_feedback_manager,
121            meta_post_manager,
122            client_id,
123            peer_tracker,
124            recent_posts_pen: Arc::new(RwLock::new(RecentPostsPen::new())),
125            single_timeline: Arc::new(RwLock::new(None)),
126            multiple_timeline: Arc::new(RwLock::new(None)),
127        })
128    }
129
130    pub fn client_id(&self) -> &ClientId {
131        &self.client_id
132    }
133
134    pub async fn client_storage_reset(&self) -> anyhow::Result<()> {
135        self.client_storage.reset().await
136    }
137
138    pub async fn submit_post(&self, post: &str) -> Result<(Vec<SubmitPostCommitTokenV1>, (EncodedPostV1, Bytes)), anyhow::Error> {
139        trace!("submitting post: {}", post);
140
141        if post.is_empty() {
142            anyhow::bail!("Post cannot be empty");
143        }
144
145        let timestamp = self.runtime_services.time_provider.current_time_millis();
146
147        struct LinkedBaseIdDetail {
148            linked_base_id: Id,
149            bucket_type: BucketType,
150            referenced_post_header_bytes: Option<Bytes>,
151        }
152
153        // Parse the post for #hashtags, @mentions, replies, sequels, etc.
154        let mut linked_base_id_details: Vec<LinkedBaseIdDetail> = vec![];
155        let mut referenced_hashtags: Vec<String> = vec![];
156
157        // The original post itself is always present
158        linked_base_id_details.push(LinkedBaseIdDetail { linked_base_id: self.client_id.id, bucket_type: BucketType::User, referenced_post_header_bytes: None });
159
160        {
161            let html = Html::parse_fragment(post);
162            {
163                // Returns true if the element is nested inside a <reply> or <repost> — meaning it
164                // belongs to a quoted post and should not be indexed under the new post's buckets.
165                let is_quoted = |element: scraper::ElementRef| -> bool {
166                    let mut node = element.parent();
167                    while let Some(n) = node {
168                        if let Some(el) = scraper::ElementRef::wrap(n) {
169                            if matches!(el.value().name(), "reply" | "repost" | "sequel") { return true; }
170                        }
171                        node = n.parent();
172                    }
173                    false
174                };
175
176                let selector_hashtag = Selector::parse("hashtag").map_err(|e| anyhow::anyhow!("Failed to parse hashtag selector: {}", e))?;
177                for element in html.select(&selector_hashtag) {
178                    if is_quoted(element) { continue; }
179                    if let Some(hashtag) = element.attr("hashtag") {
180                        trace!("hashtag={:?}", hashtag);
181                        referenced_hashtags.push(hashtag.to_string());
182                        linked_base_id_details.push(LinkedBaseIdDetail { linked_base_id: Id::from_hashtag_str(hashtag)?, bucket_type: BucketType::Hashtag, referenced_post_header_bytes: None });
183                    } else {
184                        warn!("hashtag attribute not found in element {:?}", element);
185                    }
186                }
187
188                let selector_mention = Selector::parse("mention").map_err(|e| anyhow::anyhow!("Failed to parse mention selector: {}", e))?;
189                for element in html.select(&selector_mention) {
190                    if is_quoted(element) { continue; }
191                    if let Some(client_id_str) = element.attr("client_id") {
192                        match Id::from_hex_str(client_id_str) {
193                            Ok(client_id) => {
194                                trace!("mention_id={:?}", client_id);
195                                linked_base_id_details.push(LinkedBaseIdDetail { linked_base_id: client_id, bucket_type: BucketType::Mention, referenced_post_header_bytes: None });
196                            }
197                            Err(e) => warn!("mention_id corrupted in element {:?}:, {}", element, e),
198                        }
199                    } else {
200                        warn!("mention attribute not found in element {:?}", element);
201                    }
202                }
203
204                let selector_reply = Selector::parse("reply").map_err(|e| anyhow::anyhow!("Failed to parse reply selector: {}", e))?;
205                for element in html.select(&selector_reply) {
206                    if is_quoted(element) { continue; }
207                    if let Some(post_id_str) = element.attr("post_id") {
208                        match Id::from_hex_str(post_id_str) {
209                            Ok(post_id) => {
210                                trace!("reply post_id={:?}", post_id);
211                                let referenced_post_header_bytes = element.attr("post_header_hex")
212                                    .and_then(|hex_str| hex::decode(hex_str).ok())
213                                    .map(Bytes::from);
214                                linked_base_id_details.push(LinkedBaseIdDetail { linked_base_id: post_id, bucket_type: BucketType::ReplyToPost, referenced_post_header_bytes });
215                            }
216                            Err(e) => warn!("reply post_id corrupted in element {:?}: {}", element, e),
217                        }
218                    } else {
219                        warn!("post_id attribute not found in reply element {:?}", element);
220                    }
221                }
222
223                let selector_sequel = Selector::parse("sequel").map_err(|e| anyhow::anyhow!("Failed to parse sequel selector: {}", e))?;
224                for element in html.select(&selector_sequel) {
225                    if is_quoted(element) { continue; }
226                    if let Some(post_id_str) = element.attr("post_id") {
227                        match Id::from_hex_str(post_id_str) {
228                            Ok(post_id) => {
229                                trace!("sequel post_id={:?}", post_id);
230                                let referenced_post_header_bytes = element.attr("post_header_hex")
231                                    .and_then(|hex_str| hex::decode(hex_str).ok())
232                                    .map(Bytes::from);
233                                linked_base_id_details.push(LinkedBaseIdDetail { linked_base_id: post_id, bucket_type: BucketType::Sequel, referenced_post_header_bytes });
234                            }
235                            Err(e) => warn!("sequel post_id corrupted in element {:?}: {}", element, e),
236                        }
237                    } else {
238                        warn!("post_id attribute not found in sequel element {:?}", element);
239                    }
240                }
241            }
242        }
243
244        // Encode the post
245        let linked_base_ids: Vec<Id> = linked_base_id_details.iter().map(|d| d.linked_base_id).collect();
246        let mut encoded_post = EncodedPostV1::new(&self.client_id, timestamp, linked_base_ids, post);
247        let encoded_post_bytes = encoded_post.encode_to_bytes_direct(&self.key_locker).await?;
248
249        // The commit tokens for the User's post
250        let mut post_commit_tokens = Vec::new();
251
252        // Start the post machine
253        for linked_base_id_detail in &linked_base_id_details {
254            trace!("Posting to bucket type: {:?}, linked_base_id: {}", linked_base_id_detail.bucket_type, linked_base_id_detail.linked_base_id);
255            let try_result = try {
256                // Where are we going to post it?  Start at the widest bucket and work our way narrower till we succeed
257                for &bucket_duration in bucket_durations_for_type(linked_base_id_detail.bucket_type) {
258                    let bucket_location = generate_bucket_location(linked_base_id_detail.bucket_type, linked_base_id_detail.linked_base_id, bucket_duration, timestamp)?;
259                    info!("checking posting availability of {:?}", bucket_location);
260
261                    let post_bundle = self.post_bundle_manager.get_post_bundle(&bucket_location, timestamp).await?;
262                    if !post_bundle.header.overflowed && !post_bundle.header.sealed {
263                        info!("Posting to {:?}", bucket_location);
264                        let result = posting::post_to_location(&self.runtime_services, &self.client_id.id, &self.peer_tracker, &bucket_location, &encoded_post, &encoded_post_bytes, linked_base_id_detail.referenced_post_header_bytes.as_deref(), &referenced_hashtags).await;
265                        match result {
266                            Ok(mut result) => {
267                                post_commit_tokens.append(&mut result);
268                                break;
269                            }
270                            Err(e) => {
271                                warn!("Failed to post to {:?}: {}", bucket_location, e);
272                                continue;
273                            }
274                        }
275                    }
276
277                    else {
278                        trace!("no availability: overflowed={} sealed={}", post_bundle.header.overflowed, post_bundle.header.sealed);
279                    }
280                }
281            };
282
283            if let Err(e) = try_result {
284                warn!("Failed to post to any bucket location for linked_base_id {:?}: {}", linked_base_id_detail.linked_base_id, e);
285            }
286
287            // Check that we managed at least one commit by the end of the User bucket (healing should fix the rest).  We can happily continue with a failed hashtag, mention, etc.
288            if linked_base_id_detail.bucket_type == BucketType::User && post_commit_tokens.is_empty() {
289                anyhow::bail!("Failed to post to any User buckets, so bailing,");
290            }
291        }
292
293        let encoded_post_bytes_raw = Bytes::copy_from_slice(encoded_post_bytes.bytes());
294
295        // Record in the recent posts pen so these posts appear immediately in timeline fetches
296        {
297            let bucket_locations_and_post_ids: Vec<_> = post_commit_tokens.iter()
298                .map(|token| (token.bucket_location.clone(), token.post_id))
299                .collect();
300            self.recent_posts_pen.write().await.add_all(&bucket_locations_and_post_ids, encoded_post_bytes_raw.clone(), timestamp);
301        }
302
303        Ok((post_commit_tokens, (encoded_post, encoded_post_bytes_raw)))
304    }
305
306    // ------------------------------------------------------------------
307    // MetaPostManager delegation
308    // ------------------------------------------------------------------
309
310    pub fn meta_post_manager(&self) -> &MetaPostManager {
311        &self.meta_post_manager
312    }
313
314    pub async fn submit_meta_post(&self) -> anyhow::Result<()> {
315        let post_json = self.meta_post_manager.build_meta_post_json().await?;
316        self.submit_post(&post_json).await?;
317        Ok(())
318    }
319
320    pub async fn ensure_meta_post_in_current_bucket(&self) -> anyhow::Result<()> {
321        if self.meta_post_manager.should_auto_publish(self.post_bundle_manager.as_ref()).await? {
322            self.submit_meta_post().await?;
323        }
324        Ok(())
325    }
326
327    pub async fn submit_feedback(&self, bucket_location: BucketLocation, post_id: Id, feedback_type: u8) -> anyhow::Result<()> {
328        info!("submit_feedback: bucket_location={}, post_signature={}, feedback_type={}", bucket_location, post_id, feedback_type);
329
330        // Generate our PoW
331        let (salt, pow, _hash) = EncodedPostFeedbackV1::pow_generate(&post_id, feedback_type, self.runtime_services.pow_generator.as_ref()).await?;
332
333        // Is it better than the best we have seen so far?
334        let post_bundle_location_id = bucket_location.location_id;
335        let post_bundle_feedback = self.post_bundle_feedback_manager.get_post_bundle_feedback(bucket_location.clone(), self.runtime_services.time_provider.current_time_millis()).await?;
336        let pow_best_so_far = post_bundle_feedback.get_post_pow_for_feedback_type(&post_id, feedback_type);
337        if pow <= pow_best_so_far {
338            trace!("skipping feedback submission: pow_best_so_far: {}, pow: {}", pow_best_so_far, pow);
339            return Ok(());
340        }
341
342        // Submit our good work to the servers
343        let encoded_post_feedback = EncodedPostFeedbackV1::new(post_id, feedback_type, salt, pow);
344        let result = posting::post_feedback_to_location(
345            &self.runtime_services, &self.client_id.id, &self.peer_tracker,
346            &bucket_location, &encoded_post_feedback,
347        ).await;
348        if let Err(e) = result {
349            warn!("Failed to feedback to {:?}: {}", post_bundle_location_id, e);
350        }
351
352        Ok(())
353    }
354
355    pub async fn get_post(&self, bucket_location: BucketLocation, post_id: &Id) -> anyhow::Result<(BucketLocation, EncodedPostV1, Bytes)>
356    {
357        let post_bundle = self.post_bundle_manager.get_post_bundle(&bucket_location, self.runtime_services.time_provider.current_time_millis()).await?;
358
359        let mut offset = 0;
360        for i in 0..(post_bundle.header.num_posts as usize) {
361            let len = post_bundle.header.encoded_post_lengths[i];
362            if post_bundle.header.encoded_post_ids[i] == *post_id {
363                let post_bytes = post_bundle.encoded_posts_bytes.slice(offset..offset + len);
364                let encoded_post = EncodedPostV1::decode_from_bytes(post_bytes.clone(), &bucket_location.base_id, true, true)?;
365                return Ok((bucket_location, encoded_post, post_bytes));
366            }
367            offset += len;
368        }
369
370        anyhow::bail!("Post {} not found in bundle {}", post_id, bucket_location.location_id)
371    }
372
373    pub async fn get_post_feedbacks(&self, bucket_location: BucketLocation, post_id: Id) -> anyhow::Result<[u64; 256]>
374    {
375        let mut post_feedbacks = [0u64; 256];
376
377        let post_bundle_feedback = self.post_bundle_feedback_manager.get_post_bundle_feedback(bucket_location, self.runtime_services.time_provider.current_time_millis()).await?;
378        let post_pows = post_bundle_feedback.get_post_pows(&post_id);
379        for (i, &pow) in post_pows.iter().enumerate() {
380            if pow.0 > 0 {
381                let statistical_attempts = 1u64.checked_shl(pow.0 as u32).unwrap_or(u64::MAX);
382                post_feedbacks[i] = statistical_attempts / CLIENT_FEEDBACK_POW_NUMERAIRE as u64;
383            }
384            // pow=0 is the sentinel for "no feedback submitted" — leave as 0
385        }
386
387        Ok(post_feedbacks)
388    }
389
390
391    async fn post_process_timeline_posts(&self, encoded_posts_bytes: Vec<(BucketLocation, Bytes)>) -> anyhow::Result<Vec<(BucketLocation, EncodedPostV1, Bytes)>> {
392        let mut encoded_posts = Vec::new();
393        for (bucket_location, encoded_post_bytes) in encoded_posts_bytes {
394            let result = try {
395                let encoded_post = EncodedPostV1::decode_from_bytes(encoded_post_bytes.clone(), &bucket_location.base_id, true, true)?;
396                let meta_post = MetaPost::try_parse_meta_post(&encoded_post.post)?;
397                match meta_post {
398                    MetaPost::None => encoded_posts.push((bucket_location, encoded_post, encoded_post_bytes)),
399                    MetaPost::MetaPostV1(meta_post_v1) => {
400                        let post_client_id = encoded_post.header.client_id()?;
401                        self.meta_post_manager.process_incoming_meta_post(&meta_post_v1, &post_client_id).await?;
402                    }
403                }
404            };
405
406            if let Err(e) = result {
407                warn!("Failed to decode post: {}", e);
408            }
409        }
410
411        Ok(encoded_posts)
412    }
413
414    pub async fn single_timeline_reset(&self) -> anyhow::Result<()> {
415        info!("Resetting single timeline");
416        let mut single_timeline = self.single_timeline.write().await;
417        *single_timeline = None;
418        Ok(())
419    }
420
421    pub async fn single_timeline_lock(&self, bucket_type: BucketType, base_id: &Id) -> anyhow::Result<RwLockWriteGuard<'_, Option<SingleTimeline>>> {
422        let mut single_timeline = self.single_timeline.write().await;
423
424        // If our base_id has changed, blow it away
425        if let Some(single_timeline_instance) = single_timeline.as_ref() {
426            if single_timeline_instance.bucket_type() != bucket_type || single_timeline_instance.base_id() != *base_id {
427                *single_timeline = None;
428            }
429        }
430
431        // Create the single_timeline if necessary
432        if single_timeline.is_none() {
433            trace!("Starting a new SingleTimeline for bucket_type={} base_id={}", bucket_type, base_id);
434            *single_timeline = Some(SingleTimeline::new(bucket_type, base_id, self.post_bundle_manager.clone(), self.recent_posts_pen.clone()));
435        }
436
437        Ok(single_timeline)
438    }
439
440    pub async fn single_timeline_get_more(&self, bucket_type: BucketType, base_id: &Id) -> anyhow::Result<(Vec<(BucketLocation, EncodedPostV1, Bytes)>, TimeMillis)> {
441        trace!("Getting more posts for {}", base_id);
442
443        let mut single_timeline = self.single_timeline_lock(bucket_type, base_id).await?;
444        let single_timeline = single_timeline.as_mut().expect("we have ensured that our SingleTimeline exists");
445        let encoded_posts_bytes = single_timeline.get_more_posts(self.runtime_services.time_provider.current_time_millis(), 20, bucket_durations_for_type(bucket_type)).await?;
446        let oldest_processed_time_millis = single_timeline.oldest_processed_post_bundle_time_millis();
447        let posts = self.post_process_timeline_posts(encoded_posts_bytes).await?;
448        Ok((posts, oldest_processed_time_millis))
449    }
450
451    pub async fn multiple_timeline_reset(&self) -> anyhow::Result<()> {
452        info!("Resetting multiple timeline");
453        let mut multiple_timeline = self.multiple_timeline.write().await;
454        *multiple_timeline = None;
455        Ok(())
456    }
457
458    pub async fn multiple_timeline_lock(&self, bucket_type: BucketType, base_ids: &Vec<Id>) -> anyhow::Result<RwLockWriteGuard<'_, Option<MultipleTimeline>>> {
459        let mut multiple_timeline = self.multiple_timeline.write().await;
460
461        // If our base_ids have changed, blow it away
462        if let Some(multiple_timeline_instance) = multiple_timeline.as_ref() {
463            if multiple_timeline_instance.bucket_type() != bucket_type || multiple_timeline_instance.base_ids() != base_ids {
464                *multiple_timeline = None;
465            }
466        }
467
468        // Create the multiple_timeline if necessary
469        if multiple_timeline.is_none() {
470            trace!("Starting a new MultipleTimeline for base_ids.len()={}", base_ids.len());
471            *multiple_timeline = Some(MultipleTimeline::new(bucket_type, base_ids.clone(), self.post_bundle_manager.clone(), self.recent_posts_pen.clone()));
472        }
473
474        Ok(multiple_timeline)
475    }
476
477    pub async fn multiple_timeline_get_more(&self, bucket_type: BucketType, base_ids: &Vec<Id>) -> anyhow::Result<(Vec<(BucketLocation, EncodedPostV1, Bytes)>, TimeMillis)> {
478        trace!("Getting more posts for base_ids.len()={}", base_ids.len());
479
480        let mut multiple_timeline = self.multiple_timeline_lock(bucket_type, base_ids).await?;
481        let multiple_timeline = multiple_timeline.as_mut().expect("we have ensured that our MultipleTimeline exists");
482        let encoded_posts_bytes = multiple_timeline.get_more_posts(self.runtime_services.time_provider.current_time_millis(), 60, 5, bucket_durations_for_type(bucket_type)).await?;
483        let oldest_processed_time_millis = multiple_timeline.oldest_processed_post_bundle_time_millis();
484        let posts = self.post_process_timeline_posts(encoded_posts_bytes).await?;
485        Ok((posts, oldest_processed_time_millis))
486    }
487
488    async fn get_random_peer(&self) -> anyhow::Result<Peer> {
489        {
490            let peer_tracker = self.peer_tracker.read().await;
491            if !peer_tracker.peers().is_empty() {
492                return Ok(tools::random_element(peer_tracker.peers()).clone());
493            }
494        }
495
496        // No peers — try bootstrapping
497        {
498            let mut peer_tracker = self.peer_tracker.write().await;
499            peer_tracker.bootstrap().await?;
500            anyhow::ensure!(!peer_tracker.peers().is_empty(), "Still no known peers available after bootstrap");
501            Ok(tools::random_element(peer_tracker.peers()).clone())
502        }
503    }
504
505    pub async fn fetch_url_preview(&self, url: &str) -> anyhow::Result<FetchUrlPreviewResponseV1> {
506        let peer = self.get_random_peer().await?;
507
508        let payload = FetchUrlPreviewV1::new_to_bytes(url)?;
509        let sponsor_id = self.client_id.id;
510
511        let response = rpc::rpc::rpc_server_known_with_requisite_pow(
512            &self.runtime_services,
513            &sponsor_id,
514            &peer,
515            PayloadRequestKind::FetchUrlPreviewV1,
516            payload,
517            crate::tools::config::POW_MINIMUM_PER_URL_FETCH,
518        ).await?;
519
520        anyhow::ensure!(response.response_request_kind == PayloadResponseKind::FetchUrlPreviewResponseV1, "unexpected response kind: {}", response.response_request_kind);
521        FetchUrlPreviewResponseV1::from_bytes(&response.bytes)
522    }
523
524    pub async fn fetch_trending_hashtags(&self, limit: u16) -> anyhow::Result<TrendingHashtagsFetchResponseV1> {
525        let peer = self.get_random_peer().await?;
526
527        let payload = TrendingHashtagsFetchV1::new_to_bytes(limit)?;
528        let sponsor_id = self.client_id.id;
529
530        let response = rpc::rpc::rpc_server_known(
531            &self.runtime_services,
532            &sponsor_id,
533            &peer,
534            PayloadRequestKind::TrendingHashtagsFetchV1,
535            payload,
536        ).await?;
537
538        anyhow::ensure!(response.response_request_kind == PayloadResponseKind::TrendingHashtagsFetchResponseV1, "unexpected response kind: {}", response.response_request_kind);
539        TrendingHashtagsFetchResponseV1::from_bytes(&response.bytes)
540    }
541
542    pub async fn dispatch_command(&self, command: &String) -> Result<(), anyhow::Error> {
543        let mut command_parts: Vec<String> = command.splitn(2, " ").map(|s| s.to_string()).collect();
544        if command_parts.is_empty() {
545            anyhow::bail!("Command cannot be empty")
546        }
547
548        command_parts[0] = command_parts[0].to_uppercase();
549
550        match command_parts[0].as_str() {
551            // ID
552            "I" => {
553                info!("I am hashiverse_client {}", self.client_id);
554            }
555
556            // POST
557            "P" => {
558                if command_parts.len() < 2 {
559                    anyhow::bail!("Post message cannot be empty")
560                }
561                let post_html = convert_text_to_hashiverse_html(&command_parts[1]);
562                let payload_response = self.submit_post(&post_html).await;
563                match payload_response {
564                    Ok(_) => {
565                        info!("post succeeded");
566                    }
567                    Err(e) => {
568                        error!("post error: {}", e);
569                    }
570                }
571            }
572
573            // MY POSTS
574            "M" => {
575                let encoded_posts = self.single_timeline_get_more(BucketType::User, &self.client_id.id).await;
576                match encoded_posts {
577                    Ok((encoded_posts, _oldest_processed_time_millis)) => {
578                        info!("received {} more posts", encoded_posts.len());
579                        for (bucket_location_id, encoded_post, _raw_bytes) in encoded_posts {
580                            info!("post: {} {} {}", bucket_location_id, encoded_post.header.time_millis, encoded_post.post);
581                        }
582                    }
583                    Err(e) => {
584                        error!("post error: {}", e);
585                    }
586                }
587            }
588
589            _ => {
590                warn!("unknown command: {}", command);
591            }
592        }
593
594        Ok(())
595    }
596}
hashiverse_lib/client/hashiverse_client.rs

hashiverse_lib/client/
hashiverse_client.rs
