hashiverse_lib/client/meta_post/
meta_post_manager.rs

1//! # Meta-post lifecycle orchestration
2//!
3//! The glue between [`crate::client::meta_post::meta_post`] (data model) and the rest of
4//! the client. `MetaPostManager`:
5//!
6//! - loads the current profile from `BUCKET_CONFIG` on startup,
7//! - merges incoming meta-post updates (from other devices or from the network) via the
8//!   per-field `VersionedField` CRDT rule,
9//! - writes updates back to storage, and
10//! - re-publishes the account's meta-post to the network if the current month's User
11//!   bucket no longer contains a fresh enough copy — so following clients and other
12//!   devices can discover current settings without every profile edit triggering an
13//!   immediate post.
14
15use std::collections::HashMap;
16use std::sync::Arc;
17use log::{info, warn};
18use crate::client::client_storage::client_storage::{self as cs, ClientStorage, BUCKET_META_POST_PUBLIC, BUCKET_CONFIG, BUCKET_CONFIG_KEY_META_POST_V1_PUBLIC, BUCKET_CONFIG_KEY_META_POST_V1_PRIVATE};
19use crate::client::key_locker::key_locker::KeyLocker;
20use crate::client::meta_post::meta_post::{MetaPost, MetaPostPrivateV1, MetaPostPublicV1, MetaPostV1, VersionedField, merge_public, merge_private};
21use crate::client::meta_post::meta_post_crypto;
22use crate::client::post_bundle::post_bundle_manager::PostBundleManager;
23use crate::protocol::posting::encoded_post::EncodedPostV1;
24use crate::tools::buckets::{bucket_durations_for_type, generate_bucket_location, BucketType};
25use crate::tools::client_id::ClientId;
26use crate::tools::json;
27use crate::tools::runtime_services::RuntimeServices;
28use crate::tools::time::TimeMillis;
29use crate::tools::types::{Id, Salt};
30
31/// Owns all of a user's config state — profile bio, follows, content thresholds, skip-warnings,
32/// etc. — backed by a `MetaPostV1` envelope that can be published to the network.
33///
34/// ## Storage
35///
36/// Public and private config are stored as two separate keys in `BUCKET_CONFIG`:
37/// - `{client_id}.meta_post_v1_public`  → `MetaPostPublicV1`
38/// - `{client_id}.meta_post_v1_private` → `MetaPostPrivateV1`
39///
40/// `MetaPostV1` is used strictly as the network envelope — it is assembled from the two
41/// stored sections at publish time and disassembled on receive. Because every field carries
42/// its own `TimeMillis` timestamp, concurrent edits from multiple devices merge cleanly
43/// via last-writer-wins.
44///
45/// ## Publishing policy
46///
47/// Local config changes (follows, thresholds, skip_warnings, bio) are written to
48/// `BUCKET_CONFIG` immediately but are **not** published to the network on every change —
49/// that would spam the network whenever a user toggles follows or adjusts sliders. A
50/// `MetaPostV1` is published to the network in exactly two situations:
51///
52/// 1. **Explicit publish** — the user saves their bio in the UI, which calls
53///    `build_meta_post_json()` followed by `submit_post()` on the client. This bundles the
54///    latest config (including any pending follow/threshold changes) into the post.
55/// 2. **Auto-publish on startup** — `should_auto_publish()` checks whether the current
56///    month's User bucket already contains a `MetaPostV1` from this client. If not (and the
57///    bucket isn't full), the client publishes one so other users and the user's own
58///    devices can always find an up-to-date config within a reasonable time window.
59pub struct MetaPostManager {
60    runtime_services: Arc<RuntimeServices>,
61    client_storage: Arc<dyn ClientStorage>,
62    key_locker: Arc<dyn KeyLocker>,
63    client_id: ClientId,
64}
65
66impl MetaPostManager {
67    pub fn new(
68        runtime_services: Arc<RuntimeServices>,
69        client_storage: Arc<dyn ClientStorage>,
70        key_locker: Arc<dyn KeyLocker>,
71        client_id: ClientId,
72    ) -> Self {
73        Self { runtime_services, client_storage, key_locker, client_id }
74    }
75
76    fn now(&self) -> TimeMillis {
77        self.runtime_services.time_provider.current_time_millis()
78    }
79
80    fn public_config_key(&self) -> String {
81        cs::config_key_for_user(self.client_id.id, BUCKET_CONFIG_KEY_META_POST_V1_PUBLIC)
82    }
83
84    fn private_config_key(&self) -> String {
85        cs::config_key_for_user(self.client_id.id, BUCKET_CONFIG_KEY_META_POST_V1_PRIVATE)
86    }
87
88    // ------------------------------------------------------------------
89    // Config read/write
90    // ------------------------------------------------------------------
91
92    pub async fn get_public(&self) -> anyhow::Result<MetaPostPublicV1> {
93        let public = cs::get_struct::<MetaPostPublicV1>(self.client_storage.as_ref(), BUCKET_CONFIG, &self.public_config_key(), TimeMillis::zero()).await?;
94        Ok(public.unwrap_or_else(MetaPostPublicV1::empty))
95    }
96
97    async fn put_public(&self, public: &MetaPostPublicV1) -> anyhow::Result<()> {
98        cs::put_struct(self.client_storage.as_ref(), BUCKET_CONFIG, &self.public_config_key(), public, TimeMillis::zero()).await
99    }
100
101    pub async fn get_private(&self) -> anyhow::Result<MetaPostPrivateV1> {
102        let private = cs::get_struct::<MetaPostPrivateV1>(self.client_storage.as_ref(), BUCKET_CONFIG, &self.private_config_key(), TimeMillis::zero()).await?;
103        Ok(private.unwrap_or_else(MetaPostPrivateV1::empty))
104    }
105
106    async fn put_private(&self, private: &MetaPostPrivateV1) -> anyhow::Result<()> {
107        cs::put_struct(self.client_storage.as_ref(), BUCKET_CONFIG, &self.private_config_key(), private, TimeMillis::zero()).await
108    }
109
110    // ------------------------------------------------------------------
111    // Bio (public section)
112    // ------------------------------------------------------------------
113
114    /// Update the public bio fields in the local config.
115    /// Does NOT automatically publish to the network.
116    pub async fn set_bio(&self, nickname: String, status: String, selfie: String, avatar: String) -> anyhow::Result<()> {
117        info!("set_bio: nickname={}, status={}, selfie={}, avatar={}", nickname, status, selfie, avatar);
118
119        let time_millis = self.now();
120        let local_public = self.get_public().await?;
121        let incoming_public = MetaPostPublicV1::from_bio(time_millis, nickname, status, selfie, avatar);
122        let merged_public = merge_public(&local_public, &incoming_public);
123
124        // Store in our client_storage::BUCKET_CONFIG
125        self.put_public(&merged_public).await?;
126
127        // Store in our client_storage::BUCKET_META_POST_PUBLIC as if we already received this from the network so local get_meta_post_public() works immediately
128        cs::put_struct(self.client_storage.as_ref(), BUCKET_META_POST_PUBLIC, &self.client_id.id.to_hex_str(), &merged_public, time_millis).await?;
129
130        Ok(())
131    }
132
133    pub async fn get_meta_post_public(&self, id: Id) -> anyhow::Result<Option<MetaPostPublicV1>> {
134        let meta_post_public = cs::get_struct::<MetaPostPublicV1>(self.client_storage.as_ref(), BUCKET_META_POST_PUBLIC, &id.to_hex_str(), self.now()).await?;
135        Ok(meta_post_public)
136    }
137
138    pub async fn get_all_meta_post_publics(&self) -> anyhow::Result<Vec<(String, MetaPostPublicV1)>> {
139        let mut meta_post_publics = Vec::new();
140        let keys = self.client_storage.keys(BUCKET_META_POST_PUBLIC).await?;
141        for key in keys {
142            let result: anyhow::Result<()> = try {
143                if let Some(meta_post_public) = cs::get_struct::<MetaPostPublicV1>(self.client_storage.as_ref(), BUCKET_META_POST_PUBLIC, &key, TimeMillis::zero()).await? {
144                    meta_post_publics.push((key.clone(), meta_post_public));
145                }
146            };
147            if let Err(e) = result {
148                warn!("Failed to decode MetaPostPublicV1 with key {}: {}", key, e);
149            }
150        }
151        Ok(meta_post_publics)
152    }
153
154    // ------------------------------------------------------------------
155    // Publishing
156    // ------------------------------------------------------------------
157
158    /// Build the MetaPostV1 JSON string ready for `submit_post()`.
159    /// Also updates the local BUCKET_META_POST_PUBLIC cache.
160    pub async fn build_meta_post_json(&self) -> anyhow::Result<String> {
161        info!("build_meta_post_json");
162
163        let public = self.get_public().await?;
164        let private = self.get_private().await?;
165        let salt = Salt::random();
166        let private_encrypted = meta_post_crypto::encrypt_private_section(self.key_locker.as_ref(), &salt, &private).await?;
167
168        let meta_post = MetaPostV1::new(
169            self.client_id.id_hex(),
170            salt,
171            public.clone(),
172            private_encrypted,
173        );
174
175        // Update local meta post public cache
176        cs::put_struct(self.client_storage.as_ref(), BUCKET_META_POST_PUBLIC, &self.client_id.id.to_hex_str(), &public, self.now()).await?;
177
178        json::struct_to_string(&meta_post)
179    }
180
181    /// Check whether the current month's User bucket already contains a
182    /// MetaPostV1 from this client.  Returns `true` if we should publish.
183    pub async fn should_auto_publish(&self, post_bundle_manager: &dyn PostBundleManager) -> anyhow::Result<bool> {
184        let timestamp = self.now();
185
186        let bucket_durations = bucket_durations_for_type(BucketType::User);
187        let bucket_duration = bucket_durations.first().ok_or_else(|| anyhow::anyhow!("No bucket durations for User type"))?;
188        let bucket_location = generate_bucket_location(BucketType::User, self.client_id.id, *bucket_duration, timestamp)?;
189
190        // Settled, not provisional: a provisional bundle may lack our existing MetaPost, and we would publish a duplicate
191        let post_bundle = post_bundle_manager.get_post_bundle_settled(&bucket_location, timestamp).await?;
192
193        if post_bundle.header.overflowed || post_bundle.header.sealed {
194            info!("Current bucket is full/sealed, skipping auto-publish of MetaPostV1");
195            return Ok(false);
196        }
197
198        let mut offset = 0;
199        for i in 0..(post_bundle.header.num_posts as usize) {
200            let len = post_bundle.header.encoded_post_lengths[i];
201            let post_bytes = post_bundle.encoded_posts_bytes.slice(offset..offset + len);
202            offset += len;
203
204            let decode_result = EncodedPostV1::decode_from_bytes(post_bytes, &bucket_location.base_id, true, true);
205            if let Ok(encoded_post) = decode_result {
206                if let Ok(post_client_id) = encoded_post.header.client_id() {
207                    if post_client_id.id == self.client_id.id {
208                        if let Ok(MetaPost::MetaPostV1(_)) = MetaPost::try_parse_meta_post(&encoded_post.post) {
209                            info!("MetaPostV1 already exists in current bucket, skipping auto-publish");
210                            return Ok(false);
211                        }
212                    }
213                }
214            }
215        }
216
217        info!("No MetaPostV1 found in most-recent, most-granular bucket, should auto-publish");
218        Ok(true)
219    }
220
221    // ------------------------------------------------------------------
222    // Receiving
223    // ------------------------------------------------------------------
224
225    /// Process an incoming MetaPostV1 from the network.
226    /// Caches the public section for any user; decrypts and merges the private section
227    /// if the post is from our own client_id.
228    pub async fn process_incoming_meta_post(&self, meta_post_v1: &MetaPostV1, post_client_id: &ClientId) -> anyhow::Result<()> {
229        let id = post_client_id.id.to_hex_str();
230        let now = self.now();
231
232        // Always cache the public section for any user
233        let incoming_timestamp = meta_post_v1.public.max_timestamp();
234        let existing_timestamp = cs::get_struct::<MetaPostPublicV1>(self.client_storage.as_ref(), BUCKET_META_POST_PUBLIC, &id, now).await?.map(|b| b.max_timestamp());
235        if existing_timestamp.is_none_or(|t| incoming_timestamp > t) {
236            cs::put_struct(self.client_storage.as_ref(), BUCKET_META_POST_PUBLIC, &id, &meta_post_v1.public, now).await?;
237        }
238
239        // If this is our own post, decrypt and merge both sections
240        if post_client_id.id == self.client_id.id {
241            // Merge public
242            let local_public = self.get_public().await?;
243            let merged_public = merge_public(&local_public, &meta_post_v1.public);
244            self.put_public(&merged_public).await?;
245
246            // Merge private
247            match meta_post_crypto::decrypt_private_section(self.key_locker.as_ref(), &meta_post_v1.encryption_salt, &meta_post_v1.private_encrypted).await {
248                Ok(incoming_private) => {
249                    let local_private = self.get_private().await?;
250                    let merged_private = merge_private(&local_private, &incoming_private, now);
251                    self.put_private(&merged_private).await?;
252                }
253                Err(e) => {
254                    warn!("Failed to decrypt own MetaPostV1 private section: {}", e);
255                }
256            }
257        }
258
259        Ok(())
260    }
261
262    // ------------------------------------------------------------------
263    // Followed users
264    // ------------------------------------------------------------------
265
266    pub async fn get_followed_client_ids(&self) -> anyhow::Result<Vec<Id>> {
267        let private = self.get_private().await?;
268        let client_ids: Vec<Id> = private.followed_client_ids.iter()
269            .filter(|(_key, field)| field.value == Some(true))
270            .map(|(key, _field)| Id::from_hex_str(key))
271            .collect::<anyhow::Result<Vec<_>>>()?;
272        Ok(client_ids)
273    }
274
275    pub async fn set_followed_client_ids(&self, client_ids: Vec<Id>) -> anyhow::Result<()> {
276        let timestamp = self.now();
277        let mut private = self.get_private().await?;
278
279        for field in private.followed_client_ids.values_mut() {
280            if field.value == Some(true) {
281                *field = VersionedField::tombstone(timestamp);
282            }
283        }
284        for client_id in &client_ids {
285            private.followed_client_ids.insert(client_id.to_hex_str(), VersionedField::new(true, timestamp));
286        }
287
288        self.put_private(&private).await
289    }
290
291    pub async fn set_followed_client_id(&self, client_id: Id, is_followed: bool) -> anyhow::Result<()> {
292        let timestamp = self.now();
293        let mut private = self.get_private().await?;
294
295        let field = if is_followed {
296            VersionedField::new(true, timestamp)
297        } else {
298            VersionedField::tombstone(timestamp)
299        };
300        private.followed_client_ids.insert(client_id.to_hex_str(), field);
301
302        self.put_private(&private).await
303    }
304
305    // ------------------------------------------------------------------
306    // Followed hashtags
307    // ------------------------------------------------------------------
308
309    pub async fn get_followed_hashtags(&self) -> anyhow::Result<Vec<String>> {
310        let private = self.get_private().await?;
311        let hashtags: Vec<String> = private.followed_hashtags.iter()
312            .filter(|(_key, field)| field.value == Some(true))
313            .map(|(key, _field)| key.clone())
314            .collect();
315        Ok(hashtags)
316    }
317
318    pub async fn set_followed_hashtags(&self, hashtags: Vec<String>) -> anyhow::Result<()> {
319        let timestamp = self.now();
320        let mut private = self.get_private().await?;
321
322        for field in private.followed_hashtags.values_mut() {
323            if field.value == Some(true) {
324                *field = VersionedField::tombstone(timestamp);
325            }
326        }
327        for hashtag in &hashtags {
328            private.followed_hashtags.insert(hashtag.clone(), VersionedField::new(true, timestamp));
329        }
330
331        self.put_private(&private).await
332    }
333
334    pub async fn set_followed_hashtag(&self, hashtag: String, is_followed: bool) -> anyhow::Result<()> {
335        let timestamp = self.now();
336        let mut private = self.get_private().await?;
337
338        let field = if is_followed {
339            VersionedField::new(true, timestamp)
340        } else {
341            VersionedField::tombstone(timestamp)
342        };
343        private.followed_hashtags.insert(hashtag, field);
344
345        self.put_private(&private).await
346    }
347
348    // ------------------------------------------------------------------
349    // Content thresholds
350    // ------------------------------------------------------------------
351
352    pub async fn get_content_thresholds(&self) -> anyhow::Result<HashMap<u8, u32>> {
353        let private = self.get_private().await?;
354        let thresholds: HashMap<u8, u32> = private.content_thresholds.iter()
355            .filter_map(|(key, field)| {
356                let threshold = field.value?;
357                Some((*key, threshold))
358            })
359            .collect();
360        Ok(thresholds)
361    }
362
363    pub async fn set_content_threshold(&self, feedback_type: u8, threshold: u32) -> anyhow::Result<()> {
364        let timestamp = self.now();
365        let mut private = self.get_private().await?;
366        private.content_thresholds.insert(feedback_type, VersionedField::new(threshold, timestamp));
367        self.put_private(&private).await
368    }
369
370    pub async fn set_content_thresholds(&self, thresholds: HashMap<u8, u32>) -> anyhow::Result<()> {
371        let timestamp = self.now();
372        let mut private = self.get_private().await?;
373        for (feedback_type, threshold) in thresholds {
374            private.content_thresholds.insert(feedback_type, VersionedField::new(threshold, timestamp));
375        }
376        self.put_private(&private).await
377    }
378
379    // ------------------------------------------------------------------
380    // Skip warnings for followed
381    // ------------------------------------------------------------------
382
383    pub async fn get_skip_warnings_for_followed(&self) -> anyhow::Result<bool> {
384        let private = self.get_private().await?;
385        Ok(private.skip_warnings_for_followed.value.unwrap_or(false))
386    }
387
388    pub async fn set_skip_warnings_for_followed(&self, value: bool) -> anyhow::Result<()> {
389        let timestamp = self.now();
390        let mut private = self.get_private().await?;
391        private.skip_warnings_for_followed = VersionedField::new(value, timestamp);
392        self.put_private(&private).await
393    }
394}
hashiverse_lib/client/meta_post/meta_post_manager.rs

hashiverse_lib/client/meta_post/
meta_post_manager.rs
