diff --git a/consumer/src/db/record.rs b/consumer/src/db/record.rs index 2c11df3f..31bb5a06 100644 --- a/consumer/src/db/record.rs +++ b/consumer/src/db/record.rs @@ -343,7 +343,7 @@ pub async fn post_insert( let tags = merge_tags(tags, rec.tags); - tracing::info!( + tracing::debug!( "Inserting post URI: {} , repo: {} , cid: {}", at_uri, repo, diff --git a/consumer/src/indexer/mod.rs b/consumer/src/indexer/mod.rs index 8b6ca35a..64c04385 100644 --- a/consumer/src/indexer/mod.rs +++ b/consumer/src/indexer/mod.rs @@ -12,7 +12,6 @@ use crate::utils::at_uri_is_by; use deadpool_postgres::{Object, Pool, Transaction}; use did_resolver::Resolver; use parakeet_db::types::ActorStatus; -use serde_json::Value; use foldhash::quality::RandomState; use futures::StreamExt; @@ -1200,7 +1199,40 @@ fn decode_op(op: &CommitOp, blocks: &HashMap>) -> Option<(Cid, Reco return Some((cid, record)); } Err(err) => { - tracing::warn!("Failed to deserialize Jetstream record: {}", err); + // Enhanced error reporting with record type and structure details + if let Some(type_value) = record_value.get("$type") { + let type_str = type_value.as_str().unwrap_or("unknown"); + tracing::warn!("Failed to deserialize {} record: {}", type_str, err); + + // For actor profile records, provide detailed debugging + if type_str == "app.bsky.actor.profile" { + if let Ok(json_str) = serde_json::to_string(record_value) { + tracing::warn!( + "Failed profile record content (truncated): {:.200}...", + json_str + ); + + // Check for specific fields that might cause problems + if record_value.get("avatar").is_some() { + tracing::warn!( + "Profile has avatar field with structure: {:?}", + record_value.get("avatar") + ); + } + if record_value.get("banner").is_some() { + tracing::warn!( + "Profile has banner field with structure: {:?}", + record_value.get("banner") + ); + } + } + } + } else { + tracing::warn!( + "Failed to deserialize Jetstream record (no $type): {}", + err + ); + } return None; } } diff --git a/lexica/src/lib.rs b/lexica/src/lib.rs index 60b046ae..64f31066 100644 --- a/lexica/src/lib.rs +++ b/lexica/src/lib.rs @@ -1,12 +1,12 @@ use cid::Cid; use serde::{Deserialize, Serialize}; -pub use utils::LinkRef; +pub use utils::{LinkObject, LinkRef}; pub mod app_bsky; pub mod com_atproto; pub mod community_lexicon; -mod utils; +pub mod utils; #[derive(Clone, Debug, Serialize, Deserialize)] pub struct JsonBytes { @@ -38,7 +38,10 @@ impl StrongRef { pub struct Blob { pub mime_type: String, #[serde(rename = "ref")] - #[serde(serialize_with = "utils::cid_as_link")] + #[serde( + serialize_with = "utils::cid_as_link", + deserialize_with = "utils::cid_from_link_or_string" + )] pub cid: Cid, pub size: i32, } diff --git a/lexica/src/utils.rs b/lexica/src/utils.rs index 5d1e5fbd..1c205d61 100644 --- a/lexica/src/utils.rs +++ b/lexica/src/utils.rs @@ -1,5 +1,6 @@ use cid::Cid; use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde_json::Value; pub fn cid_from_string<'de, D: Deserializer<'de>>(deserializer: D) -> Result { let str = String::deserialize(deserializer)?; @@ -20,6 +21,12 @@ pub struct LinkRef { link: String, } +#[derive(Debug, Deserialize, Serialize)] +pub struct LinkObject { + #[serde(rename = "$link")] + pub link: String, +} + pub fn cid_as_link(inp: &Cid, serializer: S) -> Result where S: Serializer, @@ -29,3 +36,32 @@ where } .serialize(serializer) } + +pub fn cid_from_link_or_string<'de, D: Deserializer<'de>>( + deserializer: D, +) -> Result { + // First try to deserialize as a raw value to check what type it is + let value = Value::deserialize(deserializer)?; + + match value { + // If it's a string, parse directly + Value::String(s) => Cid::try_from(s).map_err(serde::de::Error::custom), + + // If it's an object, it might be a link object + Value::Object(map) => { + if let Some(link_value) = map.get("$link") { + if let Some(link_str) = link_value.as_str() { + return Cid::try_from(link_str).map_err(serde::de::Error::custom); + } + } + Err(serde::de::Error::custom( + "Expected string CID or object with $link field", + )) + } + + // For any other type, we can't handle it + _ => Err(serde::de::Error::custom( + "Expected string or object with $link field", + )), + } +}