hashiverse_lib/client/timeline/
single_timeline.rs

1//! # A single stateful timeline cursor
2//!
3//! `SingleTimeline` walks one feed (a specific user's posts, a specific hashtag, the
4//! mentions of a given client, …) backwards through time, yielding pages of posts via
5//! `get_more_posts()`. It owns:
6//!
7//! - a [`crate::client::timeline::recursive_bucket_visitor::RecursiveBucketVisitor`]
8//!   that walks the hierarchical bucket ladder for the timeline's `(BucketType, base_id)`;
9//! - a `post_ids_already_seen` set that deduplicates across overlapping buckets and
10//!   across pages (important for Sequel / ReplyToPost timelines where posts can surface
11//!   from multiple places);
12//! - a [`crate::client::timeline::recent_posts_pen::RecentPostsPen`] reference so the
13//!   user's own very recent posts appear instantly instead of waiting for the next
14//!   DHT read.
15//!
16//! Timelines never reach an end — the walk just keeps stepping further back in time on
17//! demand (a hard-earned invariant: manually triggered "fetch more" must always produce
18//! *something*, even if it has to go a year further back to find it).
19
20use crate::client::post_bundle::post_bundle_manager::PostBundleManager;
21use crate::client::timeline::recent_posts_pen::RecentPostsPen;
22use crate::protocol::posting::encoded_post_bundle::EncodedPostBundleV1;
23use crate::client::timeline::recursive_bucket_visitor::{RecursiveBucketVisitor, RecursiveBucketVisitorCloseCallbackResult, RecursiveBucketVisitorOpenCallbackResult};
24use crate::tools::buckets::{BucketLocation, BucketType};
25use crate::tools::time::{DurationMillis, TimeMillis, MILLIS_IN_MONTH};
26use crate::tools::types::Id;
27use bytes::Bytes;
28use std::collections::HashSet;
29use std::sync::Arc;
30use tokio::sync::RwLock;
31
32/// A cursor that walks one feed backward through time, yielding encoded posts page by page.
33///
34/// Every feed in hashiverse — a user's profile, a hashtag, a mention stream, a reply thread
35/// — is represented on the wire as a sequence of time-bucketed post bundles keyed by a
36/// single [`Id`] (the "base id" of the feed) and a [`BucketType`]. `SingleTimeline` is the
37/// client-side state machine that reads such a feed: given a wall-clock `time_millis` and a
38/// desired page size, each call to `get_more_posts` walks to the next-older bucket via
39/// [`RecursiveBucketVisitor`], pulls the bundle through a [`PostBundleManager`], and emits
40/// the posts in reverse chronological order while deduplicating against `post_ids_already_seen`.
41///
42/// It also reads from the shared [`RecentPostsPen`] so posts the local user has just
43/// authored appear at the top of the feed before they have propagated through the network.
44/// For aggregated feeds (following multiple people, multiple hashtags) see `MultipleTimeline`,
45/// which composes several `SingleTimeline`s.
46pub struct SingleTimeline {
47    bucket_type: BucketType,
48    base_id: Id,
49    post_bundle_manager: Arc<dyn PostBundleManager>,
50    recent_posts_pen: Arc<RwLock<RecentPostsPen>>,
51    oldest_allowed_post_bundle_time_millis: TimeMillis,
52    oldest_processed_post_bundle_time_millis: TimeMillis,
53    post_ids_already_seen: HashSet<Id>,
54}
55
56impl SingleTimeline {
57    pub fn new(bucket_type: BucketType, location_id: &Id, post_bundle_manager: Arc<dyn PostBundleManager>, recent_posts_pen: Arc<RwLock<RecentPostsPen>>) -> Self {
58        Self {
59            bucket_type,
60            base_id: *location_id,
61            post_bundle_manager,
62            recent_posts_pen,
63            oldest_allowed_post_bundle_time_millis: TimeMillis::MAX,
64            oldest_processed_post_bundle_time_millis: TimeMillis::MAX,
65            post_ids_already_seen: HashSet::new(),
66        }
67    }
68
69    pub fn bucket_type(&self) -> BucketType {
70        self.bucket_type
71    }
72
73    pub fn base_id(&self) -> Id {
74        self.base_id
75    }
76
77    pub async fn get_more_posts(&mut self, time_millis: TimeMillis, max_posts: usize, bucket_durations: &[DurationMillis]) -> anyhow::Result<Vec<(BucketLocation, Bytes, bool)>> {
78        let mut encoded_posts = Vec::new();
79
80        if TimeMillis::MAX == self.oldest_allowed_post_bundle_time_millis {
81            self.oldest_allowed_post_bundle_time_millis = time_millis;
82        }
83        if TimeMillis::MAX == self.oldest_processed_post_bundle_time_millis {
84            self.oldest_processed_post_bundle_time_millis = time_millis;
85        }
86
87        let time_millis_max = time_millis;
88        let time_millis_min = self.oldest_allowed_post_bundle_time_millis - MILLIS_IN_MONTH;
89
90        RecursiveBucketVisitor::visit(
91            time_millis_max,
92            time_millis_min,
93            bucket_durations,
94            // on_bucket_open
95            &mut async |bucket_time_millis: TimeMillis, bucket_duration_millis: DurationMillis| {
96                // log::info!(
97                //     "open {}: bucket_time_millis: {}, bucket_duration_millis: {}, time_millis_min: {}",
98                //     self.base_id, bucket_time_millis, bucket_duration_millis, time_millis_min
99                // );
100
101                // Keep pushing back the oldest time we have queried - so that future calls are allowed to go back even further than that
102                self.oldest_allowed_post_bundle_time_millis = self.oldest_allowed_post_bundle_time_millis.min(bucket_time_millis);
103
104                // Get location's postbundle
105                let bucket_location = BucketLocation::new(self.bucket_type, self.base_id, bucket_duration_millis, bucket_time_millis)?;
106                let encoded_post_bundle = self.post_bundle_manager.get_post_bundle(&bucket_location, time_millis).await?;
107
108                // Do we keep going deeper?  Either way the visitor hands this same location and bundle to on_bucket_close, so it is fetched once.
109                let overflowed = encoded_post_bundle.header.overflowed;
110                let payload = (bucket_location, encoded_post_bundle);
111                match overflowed {
112                    true => Ok(RecursiveBucketVisitorOpenCallbackResult::ContinueWithChildren(payload)),
113                    false => Ok(RecursiveBucketVisitorOpenCallbackResult::ContinueWithoutChildren(payload)),
114                }
115            },
116            // on_bucket_close
117            &mut async |bucket_time_millis: TimeMillis, _bucket_duration_millis: DurationMillis, (bucket_location, encoded_post_bundle): (BucketLocation, EncodedPostBundleV1)| {
118                // log::info!("close {}: bucket_time_millis: {}, bucket_duration_millis: {}", self.base_id, bucket_time_millis, _bucket_duration_millis);
119
120                // Keep pushing back the oldest time we have processed
121                self.oldest_processed_post_bundle_time_millis = self.oldest_processed_post_bundle_time_millis.min(bucket_time_millis);
122
123                // Extract all the posts we have not seen before (some post bundles may add posts over time if they are not yet sealed)
124                let mut extraction_start_i = 0;
125                for i in 0..(encoded_post_bundle.header.num_posts as usize) {
126                    let extraction_end_i = extraction_start_i + encoded_post_bundle.header.encoded_post_lengths[i];
127                    let encoded_post_id = encoded_post_bundle.header.encoded_post_ids[i];
128                    if self.post_ids_already_seen.insert(encoded_post_id) {
129                        let encoded_post = encoded_post_bundle.encoded_posts_bytes.slice(extraction_start_i..extraction_end_i);
130                        let healed = encoded_post_bundle.header.encoded_post_healed.contains(&encoded_post_id);
131                        encoded_posts.push((bucket_location.clone(), encoded_post, healed));
132                    }
133                    extraction_start_i = extraction_end_i;
134                }
135
136                // Do we have enough posts this round?
137                match encoded_posts.len() < max_posts {
138                    true => Ok(RecursiveBucketVisitorCloseCallbackResult::Continue),
139                    false => Ok(RecursiveBucketVisitorCloseCallbackResult::Stop),
140                }
141            },
142        )
143        .await?;
144
145        // Consult the recent posts pen for any posts we submitted that haven't yet appeared from the network
146        let pen_posts = self.recent_posts_pen.write().await.get_matching_posts(self.bucket_type, &self.base_id, &self.post_ids_already_seen, time_millis);
147        for (bucket_location, encoded_post_bytes, post_id) in pen_posts {
148            if self.post_ids_already_seen.insert(post_id) {
149                encoded_posts.push((bucket_location, encoded_post_bytes, false));
150            }
151        }
152
153        Ok(encoded_posts)
154    }
155
156    pub fn post_count(&self) -> usize {
157        self.post_ids_already_seen.len()
158    }
159
160    pub fn oldest_allowed_post_bundle_time_millis(&self) -> TimeMillis {
161        self.oldest_allowed_post_bundle_time_millis
162    }
163
164    pub fn oldest_processed_post_bundle_time_millis(&self) -> TimeMillis {
165        self.oldest_processed_post_bundle_time_millis
166    }
167}
168
169#[cfg(test)]
170pub mod tests {
171    use crate::client::post_bundle::stub_post_bundle_manager::StubPostBundleManager;
172    use crate::client::timeline::recent_posts_pen::RecentPostsPen;
173    use crate::client::timeline::single_timeline::SingleTimeline;
174    use crate::tools::buckets::{BucketType, BUCKET_DURATIONS};
175    use crate::tools::time::{TimeMillis, MILLIS_IN_DAY, MILLIS_IN_MONTH, MILLIS_IN_WEEK};
176    use crate::tools::types::Id;
177    use log::info;
178    use std::collections::HashSet;
179    use std::sync::Arc;
180    use tokio::sync::RwLock;
181
182    fn empty_pen() -> Arc<RwLock<RecentPostsPen>> {
183        Arc::new(RwLock::new(RecentPostsPen::new()))
184    }
185
186    #[tokio::test]
187    async fn timeline_single_test() -> anyhow::Result<()> {
188        // configure_logging();
189
190        let id = Id::random();
191        let stub_post_bundle_manager = StubPostBundleManager::default();
192        {
193            stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_MONTH, "1D", 23)?;
194        }
195
196        let stub_post_bundle_manager = Arc::new(stub_post_bundle_manager);
197        let mut single_timeline = SingleTimeline::new(BucketType::User, &id, stub_post_bundle_manager, empty_pen());
198        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("1M")?, 5, &BUCKET_DURATIONS[1..]).await?;
199        assert_eq!(posts.len(), 23);
200        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("0")?);
201        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("0")?);
202
203        Ok(())
204    }
205
206    #[tokio::test]
207    async fn timeline_multiple_test() -> anyhow::Result<()> {
208        // configure_logging();
209
210        let id = Id::random();
211        let stub_post_bundle_manager = StubPostBundleManager::default();
212        {
213            stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_MONTH, "1D", 27)?;
214            stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_WEEK, "1W", 17)?;
215            stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_WEEK, "2W", 6)?;
216            stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_WEEK, "3W", 22)?;
217        }
218
219        let stub_post_bundle_manager = Arc::new(stub_post_bundle_manager);
220        let mut single_timeline = SingleTimeline::new(BucketType::User, &id, stub_post_bundle_manager, empty_pen());
221        info!("--- FETCH 1 -----------------------");
222        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("1M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
223        assert_eq!(posts.len(), 22);
224        assert_eq!(single_timeline.post_count(), 22);
225        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("0")?);
226        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("3W")?);
227        info!("--- FETCH 2 -----------------------");
228        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("1M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
229        assert_eq!(posts.len(), 23);
230        assert_eq!(single_timeline.post_count(), 22 + 23);
231        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("0")?);
232        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("1W")?);
233        info!("--- FETCH 3 -----------------------");
234        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("1M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
235        assert_eq!(posts.len(), 27);
236        assert_eq!(single_timeline.post_count(), 22 + 23 + 27);
237        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("0")?);
238        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("0")?);
239        info!("--- FETCH 4 -----------------------");
240        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("1M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
241        assert_eq!(posts.len(), 0);
242        assert_eq!(single_timeline.post_count(), 22 + 23 + 27);
243        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("-1M")?);
244        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("-1M")?);
245
246        Ok(())
247    }
248
249    #[tokio::test]
250    async fn timeline_multiple_months_test() -> anyhow::Result<()> {
251        // configure_logging();
252
253        let id = Id::random();
254        let stub_post_bundle_manager = StubPostBundleManager::default();
255        {
256            stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_MONTH, "1D", 23)?;
257            stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_WEEK, "2W", 5)?;
258            stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_WEEK, "3W", 16)?;
259            stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_MONTH, "1M1D", 29)?;
260            stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_WEEK, "1M2W", 21)?;
261            stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_DAY, "1M2W", 7)?;
262            stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_WEEK, "1M3W", 6)?;
263        }
264
265        let stub_post_bundle_manager = Arc::new(stub_post_bundle_manager);
266        let mut single_timeline = SingleTimeline::new(BucketType::User, &id, stub_post_bundle_manager, empty_pen());
267        info!("--- FETCH 1 -----------------------");
268        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("2M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
269        assert_eq!(posts.len(), 34);
270        assert_eq!(single_timeline.post_count(), 34);
271        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("1M")?);
272        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("1M2W")?);
273        info!("--- FETCH 2 -----------------------");
274        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("2M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
275        assert_eq!(posts.len(), 29);
276        assert_eq!(single_timeline.post_count(), 34 + 29);
277        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("1M")?);
278        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("1M")?);
279        info!("--- FETCH 3 -----------------------");
280        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("2M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
281        assert_eq!(posts.len(), 21);
282        assert_eq!(single_timeline.post_count(), 34 + 29 + 21);
283        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("0")?);
284        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("2W")?);
285        info!("--- FETCH 4 -----------------------");
286        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("2M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
287        assert_eq!(posts.len(), 23);
288        assert_eq!(single_timeline.post_count(), 34 + 29 + 21 + 23);
289        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("0")?);
290        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("0")?);
291        info!("--- FETCH 5 -----------------------");
292        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("2M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
293        assert_eq!(posts.len(), 0);
294        assert_eq!(single_timeline.post_count(), 34 + 29 + 21 + 23);
295        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("-1M")?);
296        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("-1M")?);
297        info!("--- FETCH 6 -----------------------");
298        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("2M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
299        assert_eq!(posts.len(), 0);
300        assert_eq!(single_timeline.post_count(), 34 + 29 + 21 + 23);
301        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("-2M")?);
302        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("-2M")?);
303
304        Ok(())
305    }
306
307    #[tokio::test]
308    async fn timeline_bundle_update_test() -> anyhow::Result<()> {
309        // configure_logging();
310
311        // Note that each time we do the request, the "distance back" it has searched should be increasing by 1 month because we never actually get 20 posts...
312
313        let id = Id::random();
314        let stub_post_bundle_manager = Arc::new(StubPostBundleManager::default());
315        let mut single_timeline = SingleTimeline::new(BucketType::User, &id, stub_post_bundle_manager.clone(), empty_pen());
316
317        info!("--- FETCH -----------------------");
318        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("1M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
319        assert_eq!(posts.len(), 0);
320        assert_eq!(single_timeline.post_count(), 0);
321        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("0")?);
322        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("0")?);
323
324        let mut post_bundle = stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_MONTH, "1D", 12)?;
325
326        info!("--- FETCH -----------------------");
327        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("1M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
328        assert_eq!(posts.len(), 12);
329        assert_eq!(single_timeline.post_count(), 12);
330        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("-1M")?);
331        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("-1M")?);
332
333        // Add a post to the old bundle - as if someone has posted since we last checked
334        post_bundle.header.num_posts += 1;
335        post_bundle.header.encoded_post_ids.push(Id::random());
336        post_bundle.header.encoded_post_lengths.push(0);
337        stub_post_bundle_manager.add_stub_post_bundle(&id, MILLIS_IN_MONTH, "1D", &post_bundle)?;
338
339        info!("--- FETCH -----------------------");
340        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("1M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
341        assert_eq!(posts.len(), 1);
342        assert_eq!(single_timeline.post_count(), 13);
343        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("-2M")?);
344        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("-2M")?);
345
346        info!("--- FETCH -----------------------");
347        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("1M")?, 20, &BUCKET_DURATIONS[1..4]).await?;
348        assert_eq!(posts.len(), 0);
349        assert_eq!(single_timeline.post_count(), 13);
350        assert_eq!(single_timeline.oldest_allowed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("-3M")?);
351        assert_eq!(single_timeline.oldest_processed_post_bundle_time_millis(), TimeMillis::from_epoch_offset_str("-3M")?);
352
353        Ok(())
354    }
355
356    #[tokio::test]
357    async fn timeline_fetches_each_bundle_once() -> anyhow::Result<()> {
358        // An overflowed month forces the walk down into its weeks, so buckets are opened and closed at several depths.
359        let id = Id::random();
360        let stub_post_bundle_manager = Arc::new(StubPostBundleManager::default());
361        stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_MONTH, "1D", 27)?;
362        stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_WEEK, "1W", 17)?;
363        stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_WEEK, "3W", 22)?;
364
365        let mut single_timeline = SingleTimeline::new(BucketType::User, &id, stub_post_bundle_manager.clone(), empty_pen());
366        single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("1M")?, 1000, &BUCKET_DURATIONS[1..4]).await?;
367
368        let requested_location_ids = stub_post_bundle_manager.requested_location_ids();
369        let distinct_requested_location_ids: HashSet<Id> = requested_location_ids.iter().copied().collect();
370        assert!(requested_location_ids.len() > 1, "the walk should have visited several buckets");
371        assert_eq!(requested_location_ids.len(), distinct_requested_location_ids.len(), "each bucket's bundle must be fetched exactly once");
372
373        Ok(())
374    }
375
376    #[tokio::test]
377    async fn timeline_healed_flag_is_per_post() -> anyhow::Result<()> {
378        // A bundle marks individual post_ids as healed via header.encoded_post_healed.
379        // get_more_posts must surface that as the third tuple element per post.
380        let id = Id::random();
381        let stub_post_bundle_manager = Arc::new(StubPostBundleManager::default());
382
383        let mut post_bundle = stub_post_bundle_manager.add_random_stub_post_bundle(&id, MILLIS_IN_MONTH, "1D", 5)?;
384        let healed_post_id = post_bundle.header.encoded_post_ids[2];
385        post_bundle.header.encoded_post_healed.insert(healed_post_id);
386        stub_post_bundle_manager.add_stub_post_bundle(&id, MILLIS_IN_MONTH, "1D", &post_bundle)?;
387
388        let mut single_timeline = SingleTimeline::new(BucketType::User, &id, stub_post_bundle_manager, empty_pen());
389        let posts = single_timeline.get_more_posts(TimeMillis::from_epoch_offset_str("1M")?, 20, &BUCKET_DURATIONS[1..]).await?;
390        assert_eq!(posts.len(), 5);
391
392        let healed_count = posts.iter().filter(|(_, _, healed)| *healed).count();
393        assert_eq!(healed_count, 1, "exactly one post in this bundle is marked healed");
394
395        // Pen-only posts (locally-submitted, no bundle yet) must be reported as not healed.
396        // The existing bundle test already covers the bundle case; here just spot-check the
397        // unhealed posts come through with `false`.
398        let unhealed_count = posts.iter().filter(|(_, _, healed)| !*healed).count();
399        assert_eq!(unhealed_count, 4);
400
401        Ok(())
402    }
403}
hashiverse_lib/client/timeline/single_timeline.rs

hashiverse_lib/client/timeline/
single_timeline.rs
