Drinkup

An Elixir library for consuming various AT Protocol sync services.

Drinkup provides a unified interface for connecting to various AT Protocol data streams, handling reconnection logic, sequence tracking, and event dispatch. Choose the sync service that fits your needs:

• Firehose - Raw event stream from the full AT Protocol network.
• Jetstream - Lightweight, cherry-picked event stream with filtering by record collections and DIDs.
• Tap - Managed backfill and indexing solution.

Installation

Add drinkup to your mix.exs:

def deps do
  [
    {:drinkup, "~> 0.2"}
  ]
end

Documentation can be found on HexDocs at https://hexdocs.pm/drinkup.

Quick Start

Firehose

defmodule MyApp.FirehoseConsumer do
  use Drinkup.Firehose

  @impl true
  def handle_event(%Drinkup.Firehose.Event.Commit{} = event) do
    IO.inspect(event, label: "Commit")
  end

  def handle_event(_), do: :noop
end

# In your supervision tree:
children = [MyApp.FirehoseConsumer]

For filtered commit events by collection:

defmodule MyApp.PostConsumer do
  use Drinkup.Firehose.RecordConsumer,
    collections: ["app.bsky.feed.post"]

  @impl true
  def handle_create(record) do
    IO.inspect(record, label: "New post")
  end
end

Jetstream

defmodule MyApp.JetstreamConsumer do
  use Drinkup.Jetstream,
    wanted_collections: ["app.bsky.feed.post"]

  @impl true
  def handle_event(%Drinkup.Jetstream.Event.Commit{} = event) do
    IO.inspect(event, label: "Commit")
  end

  def handle_event(_), do: :noop
end

# In your supervision tree:
children = [MyApp.JetstreamConsumer]

# Update filters dynamically:
Drinkup.Jetstream.update_options(MyApp.JetstreamConsumer, %{
  wanted_collections: ["app.bsky.graph.follow"]
})

Tap

defmodule MyApp.TapConsumer do
  use Drinkup.Tap,
    host: "http://localhost:2480"

  @impl true
  def handle_event(%Drinkup.Tap.Event.Record{} = event) do
    IO.inspect(event, label: "Record")
    :ok
  end

  def handle_event(_), do: :ok
end

# In your supervision tree:
children = [MyApp.TapConsumer]

# Track specific repos:
Drinkup.Tap.add_repos(MyTap.TapConsumer, ["did:plc:abc123"])

See the examples for some more complete samples.

TODO

• Validation for Firehose events (signatures, active account tracking) — see Firehose Validation Best Practices
• Pluggable cursor persistence (Mnesia, Postgres, etc.)
• Multi-node/distribution testing
• More comprehensive test coverage
• Additional documentation

Special thanks

The process structure used in Drinkup is heavily inspired by the work done on Nostrum, an incredible Elixir library for Discord.

License

This project is licensed under the MIT License