[backmeupplz]

The following feed only merges followed users' own casts, so recasts never show up. Clients that want a retweet-style timeline have no way to build one: they'd have to call reactionsByFid once per followed account.

This adds with_recasts=true to /v2/farcaster/feed/following and to /v2/farcaster/feed?feed_type=following. Without the flag, the feed is unchanged.

With the flag:

• Each followed account's recasts are fetched with get_reactions_by_fid (type Recast, newest first, same page size). This happens in the same per-account task as get_casts_by_fid and goes into the same timestamp heap.
• A recast sits at the time of the recast. The timestamp cursor works the same way for both kinds of entry.
• A cast shows once per page, as its newest entry. If a cast was recast by two people you follow, or posted by one and recast by another, you get one item.
• Recast casts are fetched (get_cast) only for the items on the page. A cast that has since been deleted is skipped.
• The item carries the recaster as a hydrated user in a new optional recaster field on Cast. The field is left out of the JSON when it's empty, so other endpoints don't change.

Cost: with the flag, a page makes twice as many hub reads (one more per followed account), plus one get_cast per recast shown.

Tests: test_recasts_merge_newest_first_once covers the entry mapping (cast adds and recasts in, likes out) and dedupe and ordering in take_newest. cargo test --lib api:: passes (300 tests). I haven't run it on a synced node.

Separate issue I noticed while reading this code, not changed here: max_following_fetch (1000) takes the first 1000 followed accounts in key order, which means the 1000 lowest FIDs. Someone following 17k accounts never sees casts from the other 16k.

[martinvr]

Low compatibility risk because it's opt-in, but potentially significant performance impact for busy nodes. Should we setup performance benchmarks?

[backmeupplz]

Ran benchmarks. Setup: the real hub service on RocksDB (the server_tests harness with unlimited store limits), a viewer following N accounts, each with casts and recasts of other followed accounts' casts interleaved in time, page size 25, page 1 plus page 2 via the cursor, 60 page loads per row, release build on an M1 Max.

followed accounts × messages each	main (2e8733d)	this PR, flag off	this PR, with_recasts=true
200 × 50	p50 1.76–1.94 ms	p50 1.87–1.96 ms	p50 3.27–3.38 ms (12 of 25 items are recasts)
1,000 × 100	p50 10.0–10.9 ms	p50 9.7–9.8 ms	p50 18.2 ms, p90 19–20 ms (12 of 25)

(ranges are two separate runs)

• Without the flag the cost is the same as main, within run-to-run noise, so nodes and clients that don't ask for recasts see no change.
• With the flag a page costs about 1.85× as much. That matches the extra work: one get_reactions_by_fid per followed account (capped by max_following_fetch, 1,000) next to the existing get_casts_by_fid, plus one get_cast per recast shown (up to 25). It scales linearly with the number of followed accounts, like the feed already does.

If that's still too much for busy nodes, two cheap options: fetch recasts with a smaller page size than casts (most accounts recast less than they cast), or put the flag behind a node config switch so operators can turn it off. Happy to add either.

Benchmark code (local only, not in this PR: it needs make_server to use unlimited store limits and keep its engines' temp dirs)

    /// Following feed latency with and without `with_recasts`. Not run by
    /// default: `FEED_BENCH=1 cargo test --release --lib feed_bench -- --ignored --nocapture`.
    #[tokio::test(flavor = "multi_thread")]
    #[ignore]
    async fn feed_bench() {
        use crate::api::config::{FeatureConfig, FeedConfig};
        use crate::api::events::IndexEvent;
        use crate::api::indexer::Indexer;
        use crate::api::social_graph::SocialGraphIndexer;
        use crate::api::FeedService;
        use crate::proto::{reaction_body, LinkBody, MessageData, MessageType, ReactionType};
        let followed: u64 = std::env::var("FOLLOWED")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(200);
        let per_user: u32 = std::env::var("PER_USER")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(50);
        let (_stores, _senders, [mut engine1, mut engine2], _be, service, _sd, _bd) =
            make_server(None).await;
        let now = crate::core::util::get_farcaster_time().unwrap() as u32;
        let viewer = 1_000_001u64;
        let graph_dir = tempfile::tempdir().unwrap();
        let db = Arc::new(RocksDB::new(graph_dir.path().to_str().unwrap()));
        db.open().unwrap();
        let graph = Arc::new(SocialGraphIndexer::new(
            FeatureConfig {
                enabled: true,
                ..Default::default()
            },
            db,
        ));
        let started = Instant::now();
        // casts first, then recasts of other followed users' real casts, interleaved in time
        let mut cast_ids: Vec<Vec<proto::CastId>> = Vec::new();
        for i in 0..followed {
            let fid = 1_000 + i;
            let engine = if fid % 2 == 0 {
                &mut engine2
            } else {
                &mut engine1
            };
            register_user(
                fid,
                test_helper::default_signer(),
                test_helper::default_custody_address(),
                engine,
            )
            .await;
            let casts: Vec<_> = (0..per_user / 2)
                .map(|j| {
                    let ts = now - 60 - (2 * j * followed as u32 + i as u32) * 7;
                    messages_factory::casts::create_cast_add(
                        fid,
                        &format!("cast {fid} {j}"),
                        Some(ts),
                        None,
                    )
                })
                .collect();
            cast_ids.push(
                casts
                    .iter()
                    .map(|m| proto::CastId {
                        fid,
                        hash: m.hash.clone(),
                    })
                    .collect(),
            );
            test_helper::commit_messages(engine, casts).await;
            let follow = proto::Message {
                data: Some(MessageData {
                    fid: viewer,
                    r#type: MessageType::LinkAdd as i32,
                    body: Some(proto::message_data::Body::LinkBody(LinkBody {
                        r#type: "follow".to_string(),
                        target: Some(proto::link_body::Target::TargetFid(fid)),
                        ..Default::default()
                    })),
                    ..Default::default()
                }),
                ..Default::default()
            };
            graph
                .process_event(&IndexEvent::message(follow, 0, 1))
                .await
                .unwrap();
        }
        for i in 0..followed {
            let fid = 1_000 + i;
            let engine = if fid % 2 == 0 {
                &mut engine2
            } else {
                &mut engine1
            };
            let others = &cast_ids[((i + 1) % followed) as usize];
            let recasts: Vec<_> = (0..per_user / 2)
                .map(|j| {
                    let ts = now - 60 - (2 * j * followed as u32 + i as u32) * 7 - 3;
                    messages_factory::reactions::create_reaction_add(
                        fid,
                        ReactionType::Recast,
                        reaction_body::Target::TargetCastId(
                            others[(j as usize + 1) % others.len()].clone(),
                        ),
                        Some(ts),
                        None,
                    )
                })
                .collect();
            test_helper::commit_messages(engine, recasts).await;
        }
        println!("setup: {followed} followed accounts x {per_user} messages (half casts, half recasts) in {:?}", started.elapsed());
        let feeds = FeedService::new(
            FeedConfig {
                enabled: true,
                max_following_fetch: 1000,
                ..Default::default()
            },
            Some(graph),
            None,
            Arc::new(service),
        );
        for with_recasts in [false, true, false, true] {
            let mut times = Vec::new();
            let mut items = 0;
            let mut recast_items = 0;
            for _ in 0..30 {
                let t = Instant::now();
                let page = feeds
                    .get_following_feed(viewer, None, 25, with_recasts)
                    .await
                    .unwrap();
                times.push(t.elapsed());
                items = page.items.len();
                recast_items = page.items.iter().filter(|i| i.recaster.is_some()).count();
                let cursor = page.next_cursor.clone();
                let t = Instant::now();
                feeds
                    .get_following_feed(viewer, cursor.as_deref(), 25, with_recasts)
                    .await
                    .unwrap();
                times.push(t.elapsed());
            }
            times.sort();
            println!(
                "with_recasts={with_recasts}: {} page loads, p50 {:?}, p90 {:?}, max {:?}, {items} items on page 1 ({recast_items} via recasts)",
                times.len(), times[times.len() / 2], times[times.len() * 9 / 10], times[times.len() - 1]
            );
        }
    }

Run: FOLLOWED=1000 PER_USER=100 FEED_BENCH=1 cargo test --release --lib feed_bench -- --ignored --nocapture