Process initial interests by registering them with the interest service
(
extra_interests: &[String],
interest_svc: &Arc<InterestService>,
network: &ceramic_core::Network,
node_id: &NodeId,
)
| 10 | |
| 11 | /// Process initial interests by registering them with the interest service |
| 12 | pub async fn process_extra_interests( |
| 13 | extra_interests: &[String], |
| 14 | interest_svc: &Arc<InterestService>, |
| 15 | network: &ceramic_core::Network, |
| 16 | node_id: &NodeId, |
| 17 | ) -> Result<()> { |
| 18 | if extra_interests.is_empty() { |
| 19 | return Ok(()); |
| 20 | } |
| 21 | |
| 22 | info!("Processing {} extra interests", extra_interests.len()); |
| 23 | |
| 24 | for stream_id_str in extra_interests { |
| 25 | let stream_id_str = stream_id_str.trim(); |
| 26 | if stream_id_str.is_empty() { |
| 27 | continue; |
| 28 | } |
| 29 | |
| 30 | // Validate that the model stream ID is parseable |
| 31 | let _stream_id = StreamId::from_str(stream_id_str) |
| 32 | .map_err(|e| anyhow!("Invalid model ID '{}': {}", stream_id_str, e))?; |
| 33 | |
| 34 | // Create an interest for the "model" separator key covering the full range for this stream |
| 35 | // This follows the same pattern as the API endpoint /ceramic/interests/model/{stream_id} |
| 36 | let stream_id_bytes = multibase::decode(stream_id_str) |
| 37 | .map_err(|e| anyhow!("Failed to decode stream ID '{}': {}", stream_id_str, e))? |
| 38 | .1; |
| 39 | let start = EventId::builder() |
| 40 | .with_network(network) |
| 41 | .with_sep("model", &stream_id_bytes) |
| 42 | .with_min_controller() |
| 43 | .with_min_init() |
| 44 | .with_min_event() |
| 45 | .build_fencepost(); |
| 46 | let stop = EventId::builder() |
| 47 | .with_network(network) |
| 48 | .with_sep("model", &stream_id_bytes) |
| 49 | .with_max_controller() |
| 50 | .with_max_init() |
| 51 | .with_max_event() |
| 52 | .build_fencepost(); |
| 53 | |
| 54 | let interest = Interest::builder() |
| 55 | .with_sep_key("model") |
| 56 | .with_peer_id(&node_id.peer_id()) |
| 57 | .with_range((start.as_slice(), stop.as_slice())) |
| 58 | .with_not_after(0) |
| 59 | .build(); |
| 60 | |
| 61 | match interest_svc.insert(interest).await { |
| 62 | Ok(was_inserted) => { |
| 63 | if was_inserted { |
| 64 | info!( |
| 65 | "Successfully registered extra interest for model: {}", |
| 66 | stream_id_str |
| 67 | ); |
| 68 | } else { |
| 69 | debug!( |