Metadata-Version: 2.5
Name: bluesky-zerobus
Version: 0.1.0
Summary: Bluesky Jetstream WebSocket producer for Databricks Zerobus Ingest
Project-URL: Homepage, https://github.com/jesinity/bluesky-zerobus
Project-URL: Repository, https://github.com/jesinity/bluesky-zerobus
Project-URL: Issues, https://github.com/jesinity/bluesky-zerobus/issues
Author: jesinity
License-Expression: Apache-2.0
License-File: LICENSE
Requires-Python: >=3.11
Requires-Dist: databricks-zerobus-ingest-sdk<2,>=1.9
Requires-Dist: dynaconf<4,>=3.3.5
Requires-Dist: jetstream-lakehouse<0.2,>=0.1
Requires-Dist: kafka-python<3,>=2.2
Requires-Dist: pydantic<3,>=2.11
Requires-Dist: requests<3,>=2.32
Description-Content-Type: text/markdown

# bluesky-zerobus

`bluesky-zerobus` reads Bluesky Jetstream's **v2 WebSocket** with
[`jetstream-lakehouse`](https://pypi.org/project/jetstream-lakehouse/) and sends
normalized events to a Unity Catalog Delta table. The sink uses either the official
[Databricks Zerobus Python SDK](https://github.com/databricks/zerobus-sdk/tree/main/python)
over gRPC or Zerobus' Kafka-compatible producer API. The connector follows live
data and resumes from the last locally saved Jetstream sequence after a restart.
Archived backfill can be added later.

## Prepare Databricks

Zerobus Ingest must be available in your workspace region. Create a table with
the same column names and types as the mapper below; change the fully qualified
table name as needed:

```sql
CREATE TABLE main.bronze.bluesky_events (
  source_id STRING NOT NULL,
  seq BIGINT NOT NULL,
  event_time STRING,
  witnessed_at STRING,
  did STRING,
  kind STRING,
  operation STRING,
  collection STRING,
  rkey STRING,
  cid STRING,
  rev STRING,
  is_resync BOOLEAN,
  record STRING,
  event_payload STRING
) USING DELTA;
```

`record` and `event_payload` contain JSON text. Identity, account, sync, create,
update, and delete events all land as separate rows. Build a downstream view or
table to fold events into current record state if needed.

Create a Databricks service principal and grant `USE CATALOG`, `USE SCHEMA`,
`SELECT`, and `MODIFY` on the destination. Its application ID and secret belong
in `.secrets.toml`. Follow the [Databricks setup guide](https://docs.databricks.com/aws/en/ingestion/zerobus-ingest)
for the endpoint and grants.

## Configure and run

Install dependencies with [uv](https://docs.astral.sh/uv/), then copy the example
settings into the ignored `settings.local.toml`. Set your workspace, Zerobus
endpoint, table, source ID, and batch limits there. On Azure, the endpoint is
`https://<workspace-id>.zerobus.<region>.azuredatabricks.net`; on AWS, it is
`https://<workspace-id>.zerobus.<region>.cloud.databricks.com`. Get a retained
start cursor from the same WebSocket subscription, and set
`jetstream.start_cursor` to the printed number before the first run. Cursor
discovery needs no Databricks credentials. Copy the credential template into a
local, ignored file:

```sh
uv sync --locked
cp settings.toml settings.local.toml
uv run bluesky-zerobus --config settings.local.toml --discover-cursor
# Set jetstream.start_cursor in settings.local.toml to the printed number.
cp credentials.example.toml .secrets.toml
chmod 600 .secrets.toml
uv run bluesky-zerobus --config settings.local.toml --secrets .secrets.toml
```

Edit `.secrets.toml` before running. Dynaconf reads the two explicit TOML files.
The connector does not load application settings or credentials from environment
variables. You can pass other file paths with `--config` and `--secrets`.

Choose `[sink].type = "grpc"` (the default) or `"kafka"` in the settings TOML.
Both sinks use the same `[zerobus]` endpoint, workspace URL, destination table,
and service principal credentials. For a side-by-side comparison, use separate
destination tables, pipeline IDs, and checkpoint files so each sink processes
its own source window. The Kafka-compatible API is Beta and may need to be
enabled in the workspace Previews page.

A new checkpoint requires an explicit `start_cursor`, so a failure before
the first acknowledged batch cannot silently restart from a later live tip.
The discovery command subscribes at the current tip, waits for one matching
event, and prints its sequence minus one. A saved checkpoint takes precedence
on subsequent runs. If the WebSocket can no longer serve a saved cursor, the
connector fails rather than skip data. Choose a recovery or archive backfill
procedure before changing that checkpoint.

The SQLite checkpoint is stored at `pipeline.checkpoint_path`, relative to the
settings file unless absolute. Put it on persistent storage when deploying. A
process lock prevents simultaneous consumers using the same checkpoint. The
checkpoint is tied to `pipeline.id` and a fingerprint of source, filters, source
ID, and destination table; changing those settings requires a new pipeline ID.
The Databricks credentials are excluded from the fingerprint. `source_id`
identifies a Jetstream serving instance for downstream deduplication.

Stop with SIGINT or SIGTERM. The reader stops, pending rows are flushed, the
confirmed cursor is saved, and the Zerobus stream closes.

Set `[logging].level` in the settings TOML to `DEBUG`, `INFO`, `WARNING`,
`ERROR`, or `CRITICAL` (`INFO` by default). This controls only the connector's
loggers. Logs go to stderr and report startup, cursor origin, confirmed batches,
shutdown, and failure types. Batch confirmation is logged only after the
Zerobus acknowledgment and SQLite checkpoint commit. The connector's own log
messages omit event payloads and credentials. When running as a Databricks
Python task, these messages appear in the task output;
Zerobus service activity is available separately in its system tables.

## Delivery contract

The source runs in a separate thread with a bounded queue. The WebSocket frame
limit matches the configured batch byte limit, and the default queue holds at
most 32 events. The producer groups
rows by count, UTF-8 byte size, or time. The SDK's `ingest_records_offset()`
queues a batch; `flush()` waits until Zerobus confirms it is durable. Only then
does the connector save the last event sequence in SQLite. The default 4 MiB
batch ceiling stays below Zerobus' 10 MB JSON message limit; the ceiling is an
estimate of row JSON bytes and leaves room for SDK framing. A record larger than
the configured ceiling stops the connector without advancing its cursor.

This provides **at-least-once** delivery. A crash between Zerobus' confirmation
and SQLite's checkpoint commit can repeat a row. Use `(source_id, seq)` for
deduplication downstream. A durability acknowledgment precedes query visibility
in the Delta table. See [Databricks' acknowledgment guide](https://docs.databricks.com/aws/en/ingestion/zerobus-message-blocking).

`Sink.write_batch(rows)` defines the destination contract: it returns only after
all rows are durable, or raises. `ZerobusGrpcSink` uses the official SDK's JSON
stream and recovery. `ZerobusKafkaSink` reuses a long-lived Kafka producer,
obtains a fresh table-scoped OAuth token on reconnect, and sends UTF-8 JSON values
to a topic named after the full table name. It uses `SASL_SSL`, `OAUTHBEARER`,
`acks=all`, and no compression. It checks every produce future before the source
checkpoint can advance. A failed batch may have written some rows; replay can
therefore duplicate them. Kafka's consumer, admin, and transaction APIs are not
available on this endpoint. See the [Azure Databricks Kafka-compatible API guide](https://learn.microsoft.com/en-us/azure/databricks/ingestion/zerobus-kafka).

## Development

```sh
uv sync --locked
uv run ruff check .
uv run pytest
```

Unit tests use a fake WebSocket iterator and sink. A live end-to-end run requires
your Jetstream endpoint and a Databricks workspace and is not part of the default
test suite.

## License

Apache License 2.0. See [LICENSE](LICENSE).
