fatline_rs/

lib.rs

pub use ed25519::{SigningKey, VerifyingKey};
use ed25519_dalek as ed25519;
pub use prost::Message as MessageTrait;
use tonic::transport::Channel;

#[cfg(feature = "subscription")]
pub mod sub;

pub mod action;
mod error;

pub mod cast_builder;
#[cfg(feature = "service_types")]
pub mod hub;
#[cfg(feature = "service_types")]
pub mod posts;
pub mod profile_builder;
#[cfg(feature = "service_types")]
pub mod stream;
pub mod to_messages;
#[cfg(feature = "service_types")]
pub mod users;

// Simplified hub service type
pub type HubService = proto::hub_service_client::HubServiceClient<Channel>;

// Re-export ed25519 constants so they're available for services
pub const SIGNATURE_LENGTH: usize = ed25519_dalek::SIGNATURE_LENGTH;
pub const PUBLIC_KEY_LENGTH: usize = ed25519_dalek::PUBLIC_KEY_LENGTH;

// 160-bit blake3 hash length used by Farcaster rpc requests
pub const HASH_LENGTH: usize = 20;
// Farcaster epoch in seconds, used to calculate timestamps in rpc
pub const FARCASTER_EPOCH: u64 = 1609459200;

pub mod proto {
    use tonic::{IntoRequest, Request};

    tonic::include_proto!("_"); // farcaster protos don't use package

    impl IntoRequest<SubmitBulkMessagesRequest> for Vec<Message> {
        fn into_request(self) -> Request<SubmitBulkMessagesRequest> {
            Request::new(SubmitBulkMessagesRequest { messages: self })
        }
    }
}

pub mod utils {
    use std::time::{Duration, SystemTime, SystemTimeError, UNIX_EPOCH};

    use crate::proto::{Message as FMessage, VerificationAddAddressBody};
    use ed25519_dalek::{Signature, Signer, VerifyingKey, PUBLIC_KEY_LENGTH, SIGNATURE_LENGTH};
    use eyre::Result;
    use prost::Message;
    use rand::rngs::OsRng;
    use tonic::Response;

    use super::ed25519::SigningKey;
    use crate::action::*;
    use crate::posts::{Cast, CastId, Embed, Mention, Parent};
    use crate::proto::cast_add_body::Parent as PParent;
    use crate::proto::embed::Embed as PEmbed;
    use crate::proto::link_body::Target::TargetFid;
    use crate::proto::message_data::Body;
    use crate::proto::reaction_body::Target::TargetCastId;
    use crate::proto::{LinkBody, MessageData, MessageType, ReactionBody, UserDataBody};
    use crate::{FARCASTER_EPOCH, HASH_LENGTH};

    pub fn generate_signing_key() -> SigningKey {
        SigningKey::generate(&mut OsRng)
    }

    pub fn optional_get_user_data(response: Response<FMessage>) -> Option<UserDataBody> {
        response
            .into_inner()
            .data
            // map the inner data
            .and_then(|data| data.body)
            // map the inner UserDataBody
            .and_then(|body| match body {
                Body::UserDataBody(body) => Some(body),
                _ => None,
            })
    }

    pub fn optional_get_user_data_value(user_data_body: UserDataBody) -> Option<String> {
        Some(user_data_body.value)
    }

    pub fn optional_get_reaction_message(response: Response<FMessage>) -> Option<ReactionBody> {
        response
            .into_inner()
            .data
            .and_then(|data| data.body)
            .and_then(|body| match body {
                Body::ReactionBody(body) => Some(body),
                _ => None,
            })
    }

    pub fn optional_get_follow_message(response: Response<FMessage>) -> Option<LinkBody> {
        response
            .into_inner()
            .data
            .and_then(|data| data.body)
            .and_then(|body| match body {
                Body::LinkBody(body) => Some(body),
                _ => None,
            })
    }

    pub fn truncated_hash(bytes: &[u8]) -> [u8; HASH_LENGTH] {
        let hash = blake3::hash(&bytes);
        let mut truncated: [u8; HASH_LENGTH] = [0u8; HASH_LENGTH];
        truncated.copy_from_slice(&hash.as_bytes()[0..HASH_LENGTH]);
        truncated
    }

    pub fn message_hash(message_data: &MessageData) -> [u8; HASH_LENGTH] {
        let hash = blake3::hash(&message_data.encode_to_vec());
        let mut truncated: [u8; HASH_LENGTH] = [0u8; HASH_LENGTH];
        truncated.copy_from_slice(&hash.as_bytes()[0..HASH_LENGTH]);
        truncated
    }

    pub fn sign_hash(signing_key: &SigningKey, hash: &[u8; HASH_LENGTH]) -> [u8; SIGNATURE_LENGTH] {
        signing_key.sign(hash).to_bytes()
    }

    pub fn now() -> u32 {
        (SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs()
            - FARCASTER_EPOCH) as u32
    }

    pub fn fc_timestamp_to_unix(fc_epoch_time_seconds: u32) -> Result<u32, SystemTimeError> {
        Ok((UNIX_EPOCH
            + Duration::from_secs(FARCASTER_EPOCH)
            + Duration::from_secs(fc_epoch_time_seconds as u64))
        .duration_since(UNIX_EPOCH)?
        .as_secs() as u32)
    }

    pub fn validate_signed_by(
        msg: &[u8; HASH_LENGTH],
        sig: &[u8; SIGNATURE_LENGTH],
        pub_key: &[u8; PUBLIC_KEY_LENGTH],
    ) -> Result<bool> {
        let signature = Signature::from_slice(sig)?;

        let verifying_key = VerifyingKey::from_bytes(pub_key)?;

        let verify_result = verifying_key.verify_strict(msg, &signature);
        if verify_result.is_ok() {
            Ok(true)
        } else {
            Ok(false)
        }
    }

    pub struct NonExistentTypeError;

    impl TryInto<MessageType> for String {
        type Error = NonExistentTypeError;

        fn try_into(self) -> std::result::Result<MessageType, Self::Error> {
            let link_add_str = MessageType::LinkAdd.as_str_name().to_string();
            let link_rem_str = MessageType::LinkRemove.as_str_name().to_string();
            if self == link_add_str {
                Ok(MessageType::LinkAdd)
            } else if self == link_rem_str {
                Ok(MessageType::LinkRemove)
            } else {
                Err(NonExistentTypeError)
            }
        }
    }

    pub fn cast_action_from_message(message: crate::proto::Message) -> Option<CastAction> {
        let data = message.data?;
        let author = data.fid;
        let hash = message.hash.clone();

        match data.body? {
            Body::CastAddBody(add) => {
                // transform optional and vecs to fatline types
                let embeds = add
                    .embeds
                    .iter()
                    .cloned()
                    .filter_map(|e| {
                        e.embed.map(|e| match e {
                            PEmbed::Url(url) => Embed::Url(url),
                            PEmbed::CastId(cast) => Embed::CastId(CastId {
                                fid: cast.fid,
                                hash: cast.hash,
                            }),
                        })
                    })
                    .collect::<Vec<Embed>>();

                let parent = add.parent.map(|p| match p {
                    PParent::ParentUrl(url) => Parent::Url(url),
                    PParent::ParentCastId(cast) => Parent::CastId(CastId {
                        fid: cast.fid,
                        hash: cast.hash,
                    }),
                });

                let mentions = add
                    .mentions
                    .iter()
                    .copied()
                    .enumerate()
                    .filter_map(|(pos, fid)| {
                        let start = add.mentions_positions.get(pos)?.to_owned();
                        Some(Mention {
                            fid,
                            start_pos: start,
                        })
                    })
                    .collect();

                // copy text out
                let text = add.text;
                Some(CastAction::CastAdd(Cast {
                    cast_id: CastId { fid: author, hash },
                    text,
                    mentions,
                    embeds,
                    parent,
                    cast_time: data.timestamp,
                }))
            }
            Body::CastRemoveBody(remove) => Some(CastAction::CastRemove(CastId {
                fid: author,
                hash: remove.target_hash,
            })),
            _ => None,
        }
    }

    pub fn link_from_message(message: FMessage) -> Option<Vec<LinkAction>> {
        let data = message.data?;
        let data_type = data.r#type();

        match data.body? {
            Body::LinkBody(link) => {
                if link.r#type != "follow" {
                    return None;
                }
                let TargetFid(target) = link.target?;
                let mapped = map_link_action(&data_type, data.fid, target, data.timestamp)?;
                Some(vec![mapped])
            }
            Body::LinkCompactStateBody(compaction) => {
                let mapped = compaction
                    .target_fids
                    .iter()
                    .copied()
                    .filter_map(|target| {
                        map_link_action(&data_type, data.fid, target, data.timestamp)
                    })
                    .collect::<Vec<_>>();
                if mapped.is_empty() {
                    None
                } else {
                    Some(mapped)
                }
            }
            _ => None,
        }
    }

    pub fn verification_from_message(message: FMessage) -> Option<VerificationAddAddressBody> {
        let data = message.data?;
        match data.body? {
            Body::VerificationAddAddressBody(verification) => Some(verification),
            _ => None,
        }
    }

    pub fn reaction_from_message(message: FMessage) -> Option<ReactionAction> {
        let data = message.data?;
        let data_type = data.r#type();

        match data.body? {
            Body::ReactionBody(reaction) => {
                let reaction_info = match reaction.target? {
                    TargetCastId(cast_id) => Some(ReactionInfo {
                        reference: CastId {
                            hash: cast_id.hash,
                            fid: cast_id.fid,
                        },
                        timestamp: data.timestamp,
                        reaction_type: reaction.r#type,
                    }),
                    _ => None,
                }?;
                let is_add = match data_type {
                    MessageType::ReactionAdd => true,
                    MessageType::ReactionRemove => false,
                    _ => return None,
                };
                if is_add {
                    Some(ReactionAction::Add(reaction_info))
                } else {
                    Some(ReactionAction::Remove(reaction_info))
                }
            }
            _ => None,
        }
    }

    fn map_link_action(
        message_type: &MessageType,
        source: u64,
        target: u64,
        timestamp: u32,
    ) -> Option<LinkAction> {
        let info = LinkInfo {
            source_fid: source,
            target_fid: target,
            timestamp,
        };
        match message_type {
            MessageType::LinkAdd => Some(LinkAction::AddFollow(info)),
            MessageType::LinkRemove => Some(LinkAction::RemoveFollow(info)),
            MessageType::LinkCompactState => Some(LinkAction::AddFollow(info)),
            _ => None,
        }
    }

    #[derive(Debug, Eq, PartialEq, Clone)]
    pub enum FieldUpdate<T> {
        Add(T),
        Remove,
    }
}

#[cfg(test)]
mod tests {
    use crate::hub::HubInfoService;
    use crate::proto::{
        FidRequest, MessageData, Protocol,
    };
    use crate::utils::verification_from_message;
    use crate::HubService;
    use prost::Message;
    use tonic::transport::Uri;

    #[tokio::test]
    async fn test_ok() -> Result<(), Box<dyn std::error::Error>> {
        let data = MessageData {
            r#type: 0,
            fid: 0,
            timestamp: 0,
            network: 0,
            body: None,
        };

        data.encode_to_vec();

        Ok(())
    }

    #[tokio::test]
    async fn connect_hub() -> eyre::Result<()> {
        let hub_url = dotenvy::var("HUB_URL")?;
        let uri = hub_url.parse::<Uri>()?;
        let mut hub = HubService::connect(uri).await?;
        let total_count = hub.get_current_fid_count().await?;
        assert!(total_count > 0);
        Ok(())
    }

    #[tokio::test]
    async fn number_solana_accts() -> eyre::Result<()> {
        let hub_url = dotenvy::var("HUB_URL")?;
        let uri = hub_url.parse::<Uri>()?;
        let mut hub = HubService::connect(uri).await?;
        let total_count = hub.get_current_fid_count().await?;

        let mut num_sol = 0;
        let mut num_users = 0;

        for fid in 1..=total_count {
            let req = FidRequest {
                fid,
                page_size: Some(1000),
                page_token: None,
                reverse: None,
            };

            let result = hub.get_verifications_by_fid(req).await?.into_inner();

            let mut has_sol = false;

            result
                .messages
                .iter()
                .cloned()
                .filter_map(|m| verification_from_message(m))
                .for_each(|v| {
                    if v.protocol() == Protocol::Solana {
                        num_sol += 1;
                        has_sol = true;
                    }
                });
            if has_sol {
                num_users += 1;
            }

            if fid % 20_000 == 0 {
                println!("visited {fid}, num_sol: {num_sol}, num_users: {num_users}");
            }
        }

        println!("num sol: {num_sol}, num_users: {num_users}");

        Ok(())
    }
}