Previously, on Mythical Data Pulls

This is a continuation of a previous post, which you can find here. Some of the content or locations to find certain information related to building Farcaster apps using your own Hub data may be skipped over or implied. I'll try to re-introduce topics where appropriate.

Previously we fetched a user's profile information: username, display name, bio, profile pic, and URL, all things that could be considered essential displayable information for a user in a Farcaster application. A slightly more difficult metric to track that is just as expected to be seen on a user's profile is the people they follow and more importantly, the people that follow them. If you recall the Hub's perspective on data and the available RPC calls we have you might remember we have the ability to query against every link message created by a fid:

This works fine for getting the list of people that user follows but now we are faced with a new problem - we don't know who follows them. Instead, we could use the GetLinksByTarget RPC, although I am unsure on whether Hubs will store this data properly in the future as they are keyed based on inserting LinkAdd/LinkRemoves, which may not be used via the LinkCompactStateBody messages and lead to new hubs having incomplete data in the future(?). Either way, it probably pays to just iterate every fid to get all their data anyway, arriving at a reduced state of the network we care about, such as links, reactions, casts, etc.

Since we are already iterating through every known fid, we could just as easily add some other query information in there as we go. Picture the following flow: Get the total fid count N from our hub, then from fid i in 1 to N we would simply query that fid's follow list, add it to some database or however we want to store it, and while we're at it we could get the current user profile information (from before) alongside their follow list. After iterating through every fid we should have the current state of the network mapped to show everyone's follows and their profile information and give us insights like who is the most popular user or even build up some social score metrics by weighting the followers based on some other value. We could imagine a follow by a user who is also followed by a lot of other users is weighted higher than a follow by a fresh account for example.

It's probably worth exploring the actual messages and how Hubs communicate them between each other and what it means for a consumer to receive certain types of follows, as recently the link compact messages threw me for a loop and might not be intuitive to understand compared to the normal link add / remove message types:

MessageData is a protobuf 'message' or type which contains a MessageType, the fid sending it, the timestamp they sent it (in seconds from Farcaster epoch, which is different to unix epoch), and a body which can be one of a few different types. The body types can have their own that make sense, for example a ReactionBody might have the target cast or URL(?) the user is reacting to, a LinkBody has the target_fid that the link add/remove is relevant to.

  /**
 * A MessageData object contains properties common to all messages and wraps a body object which
 * contains properties specific to the MessageType.
 */
message MessageData {
  MessageType type = 1; // Type of message contained in the body
  uint64 fid = 2; // Farcaster ID of the user producing the message
  uint32 timestamp = 3; // Farcaster epoch timestamp in seconds
  FarcasterNetwork network = 4; // Farcaster network the message is intended for
  oneof body {
    UserDataBody user_data_body = 12;
    LinkBody link_body = 14;
    /* other message body types etc ... */
    // Compaction messages
    LinkCompactStateBody link_compact_state_body = 17;
  } // Properties specific to the MessageType
}

Keen-eyed readers will notice the LinkCompactStateBody type, which was recently added as part of the FIP-15 Link Defragmentation improvement discussion. Essentially this new message should be handled differently by both Hubs and applications consuming hub messages. The improvement took a little bit of re-reading to understand but the crux of the problem this improvement addresses is that users have only a limited number of messages that hubs can store and are pruned once the user's storage limits are exceeded for that message type. As a refresher, you can check out the Farcaster protocol architecture for how storage works and the limits for each type of data.

Presently, at 2500 Link messages including removes and adds, your historical message will start getting pruned from Hubs until your message count for Links is inside that limit. The Link CRDT resolves merging link messages in a specific way, such that older messages for the same link type and target fid for any given user will be removed and updated with the latest message resolved by later timestamp. This means that fid 1 storing a LinkAdd with a type of "follow" and a target_fid of 2 will be overwritten by a LinkRemove with type of "follow" and a target_fid of 2 if the timestamp of the Remove is later than the Add. The LinkCompactStateBody is a kind of "signalling" message in this way, as well as useful for storing a larger amount of "currently followed" fids (A LinkCompactStateBody can have as many as 10 * STORAGE_UNITS * LINKS_PER_STORAGE_UNIT fids specified as "currently following", meaning that a user with a single storage unit (the minimum requirement) would be able to express 10 * 1 * 2500 = 250,000 "follows" in a single follow message).

The only complications to understand for a developer is how a Hub should treat this message, and what to interpret receiving a message of this type as. Jumping back to the Rust context, you could essentially treat all link messages as a Vec<LinkAction> here, where a single LinkBody with MessageType == MessageType::MESSAGE_TYPE_LINK_ADD is a single-element collection with an add link type. A single LinkBody with a MessageType == MessageType::MESSAGE_TYPE_LINK_REMOVE could be thought of as a single-element collection with a remove link type. The LinkCompactStateBody then can be considered a 1..250,000 element collection with a LinkAdd type, as any fid in this list should be considered "currently followed" by the application. An important thing to note here is that the LinkCompactStateBody body should probably a MessageType == MessageType::MESSAGE_TYPE_LINK_COMPACT_STATE, so if you are filtering by MessageType in a subscription of merge messages, you might miss it expecting only link adds or removes.

Another thing to note is that the LinkCompactStateBody inherently cannot preserve the original follow times for each of the users in the list, so it could be assumed that the follow time is the compact message's timestamp if we haven't already merged in a LinkAdd with the same target fid and link type for that user. We might want to therefore try to merge all the links we can, and then request any user's LinkCompactStateBody, of which there should only be one message anyway, and only add in any missing LinkAdds, not overwrite any timestamped links we have already added of those types and targets. The LinkCompactStateBody also doesn't have a link type, so you might assume it is always going to be a "follow" type, even if more link types are added in the future.

Let's start by defining the overall interaction we want to be able to facilitate, a user might want a know their follow list and who follows them, let's say we keep this as a simple index that keeps updated from a subscription to a hub built with an initial index of every fid as well. The end user could be anything from a custom client, a terminal UI, a website dashboard, a frame or even just exported to run some analysis on for building a social graph or user power score.

To start off we are going to bring in the usual suspects from the previous exercise, this time also adding in some sort of data persistence library because we want to actually keep this follow list to query against in the future to save having to make RPC calls or re-indexing every time a user talks to it. Let's start by setting up a simple table to track the link state, here I'm going to use SQLite and Diesel in Rust to make it easier:

~/git/fid-playground$ echo DATABASE_URL=sqlite://data.db > .env
~/git/fid-playground$ diesel setup
~/git/fid-playground$ diesel migration generate create_links

Now we can fill in a super super super basic sql up and down to create the required table, we won't worry about other link types for now and we can filter on only "follow" links in our code

-- up.sql
CREATE TABLE IF NOT EXISTS links (
    fid INTEGER,
    target INTEGER,
    timestamp INTEGER,
    PRIMARY KEY (fid, target)
);

-- down.sql
DROP TABLE links;

Running diesel migration run should work without any issues and give us a brand new links tables to track the from<->to follow link relationship. I'm going to first define some Message to LinkAction processes to interpret the LinkBody messages with MESSAGE_TYPE_LINK_ADD and MESSAGE_TYPE_LINK_REMOVE message types, recall the protobuf structure from above if any of this seems like we're speeding through it (This implementation is pretty much all taken from fatline-rs). Recall that we want to create a Vec of LinkAction's through the stream of every user's links via the GetAllLinkMessagesByFid request which takes a FidRequest. This in turn will give us back a list of Messages, which should be all of the user's link adds and removes.

We'll start by mapping out our LinkAction, to represent the possible actions that someone can perform via a link add or remove message

#[derive(Debug, Serialize, Deserialize, Clone)]
pub enum LinkAction {
    AddFollow(LinkInfo),
    RemoveFollow(LinkInfo)
}

Next, we'll create two more functions, one to map the message type and link's body information into the previous enum, and one to handle the unwrapping of the protobuf Message itself,

fn map_link_action(message_type: &MessageType, target: u64, timestamp: u32) -> Option<LinkAction> {
    let info = LinkInfo {
        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
    }
}

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

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

Here we simply add in a way to get a Vec of LinkAction's from any message, whether it's a LinkCompactStateBody or a LinkBody, we just use the MessageType to figure out whether it will be an add or remove, treating the MessagaeType::LinkCompactState as a list of adds as well. We use an optional in case the data body isn't what we expect, as we are dealing with the proto Message, which could contain any of the valid body types as its inner body and lets us extend the types at a future date if other link type messages are added.

fatline-rs gives us a way to receive all of these messages as a Stream, which is useful for just iterating or collecting them all from some intermediate steps, here we are going to use that property to iterate over all the link add and remove actions to run a database query for all of them, either creating a row or deleting a row in the table we defined earlier. The primary key being composed of both the source and target fids should let us have only one entry per person per target, which is fine for our needs. We need to define the model class for generating some diesel database types that allow insertion and query that match the previously defined SQLite table:

// model.rs
use diesel::prelude::*;

#[derive(Queryable,Selectable,Insertable,Debug, Clone)]
#[diesel(table_name = crate::schema::links)]
#[diesel(check_for_backend(diesel::sqlite::Sqlite))]
pub(crate) struct LinkEntry {
    pub fid: i32,
    pub target: i32,
    pub timestamp: Option<i32>
}

Now we can wire it all up with a very simple main.rs file to consume the hub messages transformed into LinkActions

use std::pin::pin;

use diesel::{BoolExpressionMethods, Connection, ExpressionMethods, QueryDsl, RunQueryDsl, SqliteConnection};
use diesel::associations::HasTable;
use dotenvy_macro::dotenv;
use fatline_rs::hub::HubInfoService;
use fatline_rs::HubService;
use fatline_rs::stream::{LinkAction, StreamService};
use tokio_stream::StreamExt;

use crate::model::LinkEntry;
use crate::schema::links::dsl::{fid as link_fid, links, target as link_target};

mod schema;
mod model;

const DB_URL: &'static str = dotenv!("DATABASE_URL");
const HUB_URL: &'static str = dotenv!("HUB_URL");

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut connection = SqliteConnection::establish(DB_URL)?;
    let mut client = HubService::connect(HUB_URL).await?;

    // for iterating
    let total_fid_count = client.get_current_fid_count().await?;

    for fid in 1..=total_fid_count {
        let page_size = None;
        let reverse = false;
        let fid_links = client.get_all_link_messages(fid, reverse, page_size);
        let mut fid_links = pin!(fid_links);

        // our actual stream iteration happens here
        while let Some(link_action) = fid_links.next().await {
            match link_action {
                LinkAction::AddFollow(info) => {
                    // when it's an add then insert values
                    let model = LinkEntry {
                        fid: fid as i32,
                        target: info.target_fid as i32,
                        timestamp: Some(info.timestamp as i32),
                    };
                    // Database insertion
                    diesel::insert_into(links::table())
                        .values(model)
                        .on_conflict_do_nothing() // my hub has some weird duplicate adds with differing timestamps, maybe a hub sync issue?
                        .execute(&mut connection)?;
                },
                LinkAction::RemoveFollow(info) => {
                    // when it's a remove then remove values (if any exist)
                    let source = fid as i32;
                    let target = info.target_fid as i32;
                    // Database deletion
                    diesel::delete(links.filter(link_target.eq(target).and(link_fid.eq(source))))
                        .execute(&mut connection)?;
                }
            }
        }
        println!("Mapped fid {fid}");
    }
    // we now have every fid link mapped
    Ok(())
}

Running this application should now ingest all the link information and print every fid that is mapped after consuming all the current link events, minus the compact state list events, but for now it should get us started. The output looks something like this:

... etc
Mapped fid 14
Mapped fid 15
Mapped fid 16
Mapped fid 17
Mapped fid 18
Mapped fid 19
Mapped fid 20
... etc

And output a table that looks something like this:

You should be able to query against this and serve this data in... several days time... or less, depending on your latency to the hub and transfer speed I'm guessing, as well as any SQL tuning we do to optimise the writes. Another optimisation could be starting the initial sync in a breadth-first approach, you might have a specific fid in mind, and queue up that initial fid, followed by all the fids they follow, and their fids they follow etc, instead of brute forcing 1 to 700,000+. At the end of the day these are just some trade-offs you might want to make in your own application to get to a faster result quicker. You might even fall back to using the get links by target RPC for the specific fids you want, and branch out that way to at least build two-way links from an initial fid.

As of writing this I don't have a checkpointed data source for people to analyze, but in theory it would be pretty easy to spin one up if you left the application running, although you might want to throw in some extra data into that export reactions, profile info and casts before actually wanting to use it for anything useful.

Thanks for getting this far and let me know if you have any other useful data questions you want to explore for a future post, or how you use the link data from the Farcaster network in your own applications!

Some Background

As a developer particularly focused in building the future of France, the idea of an open, permission-less and alternative social network protocol to the Big Ones seems appealing. Having played around with previous iterations of social media data endpoints back when you were allowed to do so without paying, even the most restricted API limits were enough to generate some ideas and validate them. That's why the idea of being able to run your own hardware and contribute to the sufficient decentralisation of the network appealed greatly to me and I wanted to play around with and understand how it worked from the ground up.

What kind of design and architecture choices were taken? What are the primitives for communication, message storage, etc? What are the balances and trade-offs or potential hazards to look out for when writing applications that consume or broadcast this data? Are there any improvements you can suggest to the protocol overall, or implementation-specific to make your life a little easier? The documentation on the official Farcaster websites for the protocol as well as the message types for the Hubble APIs get us enough information to start building and are a pretty easy way to get across the basics to understand this information on your own as well:
https://docs.farcaster.xyz/learn/architecture/overview
https://docs.farcaster.xyz/reference/hubble/datatypes/messages

A good way to conceptualise all of this, much like any other distributed and shared data source, is as an event stream that every node resolves to get to an overall view of the state. Although in the Hub's case it is out of scope to assume you could query any "reduced" view of this state for any given user or post, instead relying on that stream's state reduction being done in your app/indexer-specific process; either a service you can pay for, an open source implementation, or writing your own application-specific implementations using libraries or clients to query against your own hub (my personal favourite, and fitting with the open and decentralised theme of the protocol). You can visualise this separation conceptually for a single user's "Profile Update" messages like this:

This diagram shows the separation between what you could consider the data streams of the network, real-time, historical for every user and cast on the network. Every like and recast, every follow and unfollow, every cast and the data associated with them. Today I want to go through the process of creating a simple script that could give you the "up to date" view of a specific user's profile information as that's probably one of the earliest and easiest tasks one might want to do with their own Farcaster application.

The choices for implementation matter and are at the end of the day it is always better to work with languages and tooling you are more familiar with. Most of the Farcaster work I've done outside of frames or browser-facing applications are all in Rust, just because I like it and there wasn't a huge amount of existing tooling in it for building Farcaster applications (good learning opportunity compared to using existing libraries I wouldn't understand at the fundamental level). I started writing a library to make a lot of these hub to app-specific translations easier or help import the historical data and I'll continue to build it in a semi-leftcurve way because I'm not S-tier Rust or library developer, but I'm Doing My Part™ and it's been fun which is the most important thing I think. :)

The Hub endpoints we'll be communicating with use gRPC and protobufs, the first step of creating a library or application to help script out communications with the Hub is to compile the protobufs from the official Farcaster hub monorepo, I started by just cloning it as a submodule and creating a build step to compile and make available the compiled RPC services and types. You can usually get a gRPC library for this in your language of choice, I used tonic for the Rust code. It might be mentioned that you would need a Hub running and up to date with all the OP and mainnet data as well as the Farcaster network messages you care about querying to be synced with it, you can use a hosted hub from a provider or run your own on your own machine. The current DB requirement is probably 100-200+GB of storage.

The data we'll be querying comes from the Hub's UserData endpoints in the RPC service (protos), which we could crudely construct a couple of different ways depending on what specific information the application wanted. If we didn't want historical data we could just use the rpc GetUserData(UserDataRequest) returns (Message); function which returns a single Message for each type (to get the latest value for each field, although we have to make a different request for each type we cared about) or iterate through all of the messages using rpc GetAllUserDataMessagesByFid(FidRequest) returns (MessagesResponse); or the paged version rpc GetUserDataByFid(FidRequest) returns (MessagesResponse); which will return all the currently stored user data messages for that fid (minus any that have been pruned from storage limits or deletions I guess)

Writing this in a Rust script, you can start by adding my fatline-rs library, or just pull in the protobufs and add the tonic dependency and build.rs stuff yourself (bear in mind some code may reference utility functions or extra library code I've written for fatline-rs)

For adding the library to your existing Cargo.toml:

# Cargo.toml file
# other dependencies you might want and Cargo metadata etc ...
[depdencies]
eyre = "0.6.12"
tokio = { version = "1", features = ["full"] } # probably useful as the tonic responses will be async

[dependencies.fatline-rs]
git = "https://github.com/0x330a-public/fatline-rs.git"
rev = "c155d9f862c56e94ecf508d1185a114e1c5bc1a4"

In the src/main.rs we're just going to add a constant for our Hub's publicly accessible IP/URL, you could just as easily add this as a dotenv or something using dotenvy or equivalent library

const HUB_URL: &'static str = "http://somethingsomething:2283";

We're going to use the application to query and print out the current state of a specific user by their fid, we'll use the Farcaster account for this example (fid #1). The information we should expect to see should match the Warpcast display for Farcaster's user profile page and looks (presently) something like this:

We can see the user's username @farcaster, the display name Farcaster, the profile picture (Farcaster logo), and the bio A sufficiently decentralized social network. farcaster.xyz. To pull all this data we need to construct a client using our Hub URL, surrounded by some other async Rust program boilerplate and imports:

use eyre::Result;
use fatline_rs::HubService;

const HUB_URL: &'static str = "http://somethingsomething:2283";

#[tokio::main]
async fn main() -> Result<()> {

	// HubService here is a simplified, re-exposed type from fatline-rs, HubServiceClient<Channel> is the original type
	let mut service = HubService::connect(HUB_URL).await?;

	// return Result::Ok at the end of main
	Ok(())
}

Running this won't give us any useful information, except I guess that the execution didn't result in an error when connecting to our Hub. We can expand the code to fetch each specific piece of information for the Farcaster account (fid #1):

const FC_FID: u64 = 1;

In fatline-rs there is a function to simplify getting a User's current profile information, there's a custom Profile type I use that enables serialization/deserialization from an API as well. The implementation just calls the client's get_user_data endpoint for each type, with a shorthand function to build the specific FidRequest, as well as some utility functions to get the response out of the proto RPC response.

The basic gist is that we want to query every UserDataType for any given fid as individual requests, which should give us the latest message for every type of user data. Once we get back an optional body, we check that the body is a UserDataBody which indicates it's a message containing an update to this user's user-data aka pfp/username/displayname etc. After this check we just have to get the UserDataBody field and type to see which field was updated and what the new field value is. We could stream through or subscribe to all realtime messages and filter on this user's updates to set or update user fields in a DB for indexing purposes so we can always return the current state in some API or use it for whatever purpose we want to.

we can recreate the logic for getting the profile like so:

use eyre::Result;
use fatline_rs::proto::{UserDataType, UserDataRequest, Message, MessageData};
use fatline_rs::proto::message_data::Body;
use fatline_rs::HubService;
use tonic::{Response, Status};

/// The combined user's profile, holding values from all user update types
#[derive(Debug)]
pub struct Profile {
    pub fid: u64,
    pub username: Option<String>,
    pub display_name: Option<String>,
    pub profile_picture: Option<String>,
    pub bio: Option<String>,
    pub url: Option<String>
}

// shorthand so we can call this for each type
fn get_user_data_request(fid: u64, data_type: UserDataType) -> UserDataRequest {
	// the RPC method is expecting this as the "request"
    UserDataRequest {
        fid,
        user_data_type: data_type as i32
    }
}

// Actual function to get the profile
async fn get_user_profile(client: &mut HubService, fid: u64) -> Result<Profile> {
	// lets start by getting the username as an example:
	let username_request: Response<Message> = client.get_user_data(
		get_user_data_request(fid, UserDataType::Username)
	).await?;

	// Now we have a response, we can get the inner message and extract the message, data and body,
	// and finally see the client published "profile update" message body
	let message: Message = username_request.into_inner();
	let body: Option<Body> = message.data.and_then(| data: MessageData | data.body);
	// The UserDataBody contains the type (UserDataType) as well as the value (String)
	// We *should* expect the type to match the type we requested, in this case the username
	let username_data_body: Option<UserDataBody> =  match body {
		Some(Body::UserDataBody(body)) => Some(body),
		_ => None
	}; 
	// ignore the type since we requested it explicitly, map the body and pull the value out
	let username_value: Option<String> = username_data_body.map(|body| body.value);
	// technically usernames can be optional, so this is all we need for the profile for now
	let profile = Profile {
		fid,
		username: username_value,
		display_name: None,
		profile_picture: None,
		bio: None,
		url: None
	};
	// return our poorly populated profile
	Ok(profile)
}

At this point we would just populate all the other remaining fields, maybe write some useful helpers to extract other username data, shortcut all of the body/optional/message etc types and fill out the rest of the get_user_profile function here.

Returning to our main method, we would add the call for this new profile function and display the data however we want, like logging it to the terminal output:

use eyre::Result;
use fatline_rs::HubService;

const HUB_URL: &'static str = "http://somethingsomething:2283";
const FC_FID: u64 = 1;

#[tokio::main]
async fn main() -> Result<()> {

	// HubService here is a simplified, re-exposed type from fatline-rs, HubServiceClient<Channel> is the original type
	let mut service = HubService::connect(HUB_URL).await?;
	
	// call the new profile function from wherever we implemented it, assuming here it's in the main.rs
	let profile: Profile = get_user_profile(client, FC_FID).await?;
	// as username is optional, get a default if the user doesn't have one set
    let username = match profile.username {
        Some(value) => value,
        _ => "actually nothing".to_string()
    };

    println!("FC profile's username is: {}", username);

	// return Result::Ok at the end of main
	Ok(())
}

Running this should give us an output something like this:

FC profile's username is: farcaster

I'll leave implementing the helpers and other user profile fields as an exercise for the reader, alternatively read through the library implementation, mine isn't perfect or the ideal implementation but seems to work for me!

Reading data from your own Hub is easy and fun and can help to diversify client applications, reduce dependence on hosted or paid services which also helps decentralize the overall Farcaster network and keep your personal costs down. If you have the hardware to run one it provides very low latency access to the current entire state of the network. You can think of this similar to having your own Eth node running to query against vs relying on something like Infura. I'll try to write more content like this alongside writing a library and alternate client as a learning tool and to help diversify the open source Farcaster community. Any tips or feedback is welcome and appreciated.

Thanks for reading until the end! Hit me up on Farcaster @harris- for what kind of content you want to see next, and I would appreciate any stars or follows and requests for new features on my fatline client and libraries