[Call for proposal] Add support for Bluesky backfills + more Bluesky data types

[mark-torres10]

Issue Description

User request

We got a request from a user to get a large amount of data from the Bluesky network. Basically they want to be able to provide a list of a thousand users and get all of their interactions and activity on the app over the course of the last six months. This would naturally complement the currents. We work in the data platform and it is asked that it is repeated a variety of times across the lab. There is a demand for data that is not easily accessible via the Bluesky API, as you have seen before (see, for example, your past work on the Bluesky Jetstream).

Background

In the current data platform we specify a particular way to retrieve records from Bluesky. We use the direct Bluesky API. We then also get only posts currently. We take this and pass it into the rest of the pipeline. This is okay for now but we want to expand to more methods of ingestion from Bluesky as well as more data types.

https://github.com/bluesky-social/jetstream

There are three ways to retrieve data from Bluesky:

1. Via API
2. The firehose
3. The Jetstream backfill application

In past work, you have shown a way to backfill records from Bluesky. Now we need to be able to natively integrate that into the data platform layer. We also need to support more data types in the ingestion model. Right now we support just post records. The current application would not need to support passing these other data types of entities through the entire pipeline.

We basically only need to do ingestion for these for now as we don't have a use case that requires downstream filtering and enrichment and curation. However, we would want to build Glue tables on top of these raw data types, so we can query them.

There are two distinct but related workflows needed in this proposal:

1. Introduce the new backfill integration into the data pipeline. This will mean that Bluesky records will have two ways to interact with it: (1) via API, which is currently supported, (2) the backfill application.
2. Add support for more raw data types.

Task

Please create an RFC for how you'd suggest we implement this.

Feel free to split this into 2 RFCs if that makes sense, though since they're related (and part of the same ask from an end user), that's why they're mentioned together in this issue.

Beyond the template provided in the link above, please provide a list of ticket-scoped tasks that need to be done. It's OK to use Claude for this, but I'd like your commentary on what scope is appropriate and reasonable for each ticket (especially given your background knowledge of the repo). Now that the app is getting more mature, let's aim to have PRs be the size of a single feature or unit of work, perhaps max 500-600 lines (which includes testing + docs).

Requirements

Functional requirements

Jetstream

• Ingestion should support 2 ingestion methods for Bluesky, selectable via YAML.
• Backfill config accepts: list of actors (handles and/or DIDs), record types, time window. Possibly can include a specific cursor, though I'll leave that as an implementation choice.
• Handle -> DID resolution, and excluding users who don't have valid DIDs (e.g., their accounts are deleted/suspended).

New records

• V1 should support posts, likes, reposts, follows
• Each record should have a documented Pydantic schema.
• On ingestion, store provenance fields (uri, author_did, collection, operation (create/delete), created_at (when was the record created), sync_timestamp)
• Consider whether you want to use separate tables based on ingestion method or not. For example, the post record from the API will have fields not available when you sync posts from Jetstream or the firehose. Please evaluate these and provide a proposal as well as tradeoffs with the alternative. Please write this up as an ADR (like your previous ADR) for us to discuss more about, as this has implications for how we grow the data platform over time.

Orchestration

• Prefect DAG skips downstream stages for the backfill step. This can be done a variety of ways, e.g., adding ingestion_only: true in the config.

Deduplication

• We should have a way to track "we've recorded XYZ records for user A in timeframe B to C", and then on subsequent runs, we can skip subsets of work accordingly.

Non-functional requirements

Reliability

• Jobs should survive disconnect/restart without data loss.
• For transient API errors, we want to retry with exponential backoff.
• We want to store run metadata in DynamoDB.

Performance/scale

• We want to support >= 1,000 actors for a single job.
• We want to suppport a 180-day lookback.
• We want to include checkpointing, in order to avoid losing hours of work.

Deduplication

• Ideally syncs are idempotent: we don't write records twice if they've already been synced.

Cost + storage

• Use .parquet + S3 (to match the current data storage usage)

What counts as complete?

Task completion

• We can run backfill on some combination of (user(s), collection(s), time_chunk).
• A task is considered complete when (1) we've confirmed that our replay cursor has advanced through the entire code chunk, (2) rows are flushed to storage, (3) we store run metadata. We want to explicitly see that the replay cursor hits a timestamp that is beyond our requested range, and only if we see that should we conclude that the cursor has gone through the entire intended range.
• A task fails if the above aren't true for the given set of parameters, after X retries (e.g., 3 retries).
• We skip backfill if (1) after filtering for valid users, there's no more users to sync (e.g., accounts are deleted), or (2) for that combination of (user(s), collection(s), time_chunk), we already have all their data (e.g., we synced them previously, or some combination of our previous runs creates the requested run (e.g., (user1, posts, last two weeks) = (user1, posts, 1 week ago) + (user1, posts, 1 week ago, but we ran it 1 week ago, so it's the data from 2 weeks ago).

Specific deliverables

• Works on a test case of "get all activity (posts/likes/follows/reposts) for a set of 100 Bluesky users for the past 2 weeks" (this as a .csv file the explicit deliverable that we'll want to return for the immediate user use case).

User journeys

Here are some user journeys that we'll want to keep in mind to fulfill for the workflow. We also can codify these as end-to-end tests of the pipeline itself to verify functionality.

Journey 1: Researcher requests an actor activity export

• Actor: Lab researcher
• Goal: 1,000 handles, last 180 days, posts + likes + reposts + follows

1. Researcher provides actors.csv (handles, optional notes).
2. Engineer creates ingestion YAML: ingestion_method: backfill, record_types, start_time / end_time, dataset_id.
3. Engineer runs pipeline.
4. Job runs for hours/days; researcher can see status if long-running (e.g., via [Call for proposal] Add an observability dashboard for ease of evaluation #113).
5. Sync is then done, and then the researcher is informed that it's done, and then they can directly query the app to get the data they need.

What success looks like: Researcher can answer “what did these users do on Bluesky in this window?” without manual Jetstream scripts.

On future use, if the researcher wants the same dataset, they should be able to just query the webapp and the data is already available.

Journey 2: Ops resumes an interrupted backfill

• Actor: Platform engineer
• Goal: Job died at actor 600/1000 after network drop; no duplicate work

1. Engineer inspects run.
2. Engineer re-runs with --run-dir (same pattern as API sync today).
3. Completed (actor, collection, chunk) tasks are skipped.
4. In-progress tasks restart from last persisted cursor_end (or chunk start if cursor not trusted).
5. Run eventually reaches run complete or partial with failed tasks listed.

What success looks like: No manual "figure out where we left off"; at worst, re-fetch one chunk with row-level dedupe.

Note: This ties into the borrowed discussion that we've been having about how to handle retries and retry semantics on failure. Feel free to include a discussion of that here if you think that's applicable and if you think that we should consider retries across the entire pipeline as a single sprint or a unit of work.

Journey 3: Existing keyword/API pipeline (unchanged)

• Actor: Research engineer
• Goal: Keyword search → preprocess → features → curate

1. Run sync_bluesky.py --config mirrorview.yaml (API, posts only).
2. Prefect orchestrator runs full DAG (no ingestion_only).
3. Output: curated Mirrorview CSV + existing Glue path.

What success looks like: Zero regression; backfill work does not change this path.

Journey 4: Ingestion-only backfill via Prefect

• Actor: Engineer scheduling lab jobs
• Goal: Backfill ingested and uploaded; no preprocess/features/curate

1. YAML includes ingestion_only: true (or equivalent).
2. Engineer runs orchestration DAG.
3. DAG runs ingestion only; downstream stages recorded as skipped in DynamoDB pipeline run.
4. Ingestion reaches run complete → S3 + Glue ready.

What success looks like: No spurious failures from preprocess expecting posts.csv shape on a likes-only job.

Journey 5: Engineer adds a new record type for the backfill to ingest

• Actor: Platform engineer
• Goal: Add app.bsky.graph.block without touching downstream stages

1. Add Pydantic model + raw schema doc.
2. Add collection to record_types enum / filename mapping.
3. Add Glue DDL script + register on upload.
4. No changes to preprocess / features / curate.

What success looks like: New table queryable; backfill task builder picks it up automatically.

Initial thoughts about implementation

Jetstream

User vs. keyword focus

The current Bluesky ingestion work focuses on keyword-style queries. We'll want to support queries for both keywords and users.

Subtasks for Jetstream

For introducing Jetstream, there's also 2 subtasks within it:

1. Get Jetstream to work at some nontrivial scale (building on your previous experimentation work). This might involve getting more API keys (so you can have more accounts accessing the Jetstream endpoint), having multiple requests in parallel, etc.
2. As you found in your previous work, it takes a while to run the backfill, but to get data further back in time, you need to run Jetstream for a while. Therefore, for nontrivial queries or requests, you'll want to be able to make sure that the backfill can run for long periods of time (and restart as needed). Additionally, you'll want to consider, for larger data collection jobs, things like network disconnects and reducing duplicated work. This should be built with the assumption that a 1-2 hour data collection job can (and likely will) be interrupted halfway through, and should be robust for that.

Testing

For load testing, your previous Locust experimentation is pretty helpful (also generally helpful for any load testing of the pipeline)