comet.sh
Elixir ATProtocol ingestion and sync library.
docs: update README
ovyerus.com
+60
-58
README.md
+60
-58
README.md
···
1
1
# Drinkup
2
2
3
-
An Elixir library for listening to events from an ATProtocol relay
4
-
(firehose/`com.atproto.sync.subscribeRepos`). Eventually aiming to support any
5
-
ATProtocol subscription.
3
+
An Elixir library for consuming various AT Protocol sync services.
6
4
7
-
## TODO
5
+
Drinkup provides a unified interface for connecting to various AT Protocol data
6
+
streams, handling reconnection logic, sequence tracking, and event dispatch.
7
+
Choose the sync service that fits your needs:
8
8
9
-
- Support for different subscriptions other than
10
-
`com.atproto.sync.subscribeRepo'.
11
-
- Validation (signatures, making sure to only track handle active accounts,
12
-
etc.) (see
13
-
[Firehose Validation Best Practices](https://atproto.com/specs/sync#firehose-validation-best-practices))
14
-
- Look into backfilling? See if there's better ways to do it.
15
-
- Built-in solutions for tracking resumption? (probably a pluggable solution to
16
-
allow for different things like Mnesia, Postgres, etc.)
17
-
- Testing of multi-node/distribution.
18
-
- Tests
19
-
- Documentation
9
+
- **Firehose** - Raw event stream from the full AT Protocol network.
10
+
- **Jetstream** - Lightweight, cherry-picked event stream with filtering by
11
+
record collections and DIDs.
12
+
- **Tap** - Managed backfill and indexing solution.
20
13
21
14
## Installation
22
15
23
-
Add `drinkup` to your `mix.exs`.
16
+
Add `drinkup` to your `mix.exs`:
24
17
25
18
```elixir
26
19
def deps do
···
32
25
33
26
Documentation can be found on HexDocs at https://hexdocs.pm/drinkup.
34
27
35
-
## Example Usage
28
+
## Quick Start
36
29
37
-
First, create a module implementing the `Drinkup.Consumer` behaviour (only
38
-
requires a `handle_event/1` function):
30
+
### Firehose
39
31
40
32
```elixir
41
-
defmodule ExampleConsumer do
42
-
@behaviour Drinkup.Consumer
33
+
defmodule MyApp.FirehoseConsumer do
34
+
@behaviour Drinkup.Firehose.Consumer
43
35
44
-
def handle_event(%Drinkup.Event.Commit{} = event) do
45
-
IO.inspect(event, label: "Got commit event")
36
+
def handle_event(%Drinkup.Firehose.Event.Commit{} = event) do
37
+
IO.inspect(event, label: "Commit")
46
38
end
47
39
48
40
def handle_event(_), do: :noop
49
41
end
42
+
43
+
# In your supervision tree:
44
+
children = [{Drinkup.Firehose, %{consumer: MyApp.FirehoseConsumer}}]
50
45
```
51
46
52
-
Then add Drinkup and your consumer to your application's supervision tree:
47
+
### Jetstream
53
48
54
49
```elixir
55
-
defmodule MyApp.Application do
56
-
use Application
50
+
defmodule MyApp.JetstreamConsumer do
51
+
@behaviour Drinkup.Jetstream.Consumer
57
52
58
-
def start(_type, _args) do
59
-
children = [{Drinkup, %{consumer: ExampleConsumer}}]
60
-
Supervisor.start_link(children, strategy: :one_for_one)
53
+
def handle_event(%Drinkup.Jetstream.Event.Commit{} = event) do
54
+
IO.inspect(event, label: "Commit")
61
55
end
56
+
57
+
def handle_event(_), do: :noop
62
58
end
63
-
```
64
59
65
-
You should then be able to start your application and start seeing
66
-
`Got commit event: ...` in the terminal.
67
-
68
-
### Record Consumer
60
+
# In your supervision tree:
61
+
children = [
62
+
{Drinkup.Jetstream, %{
63
+
consumer: MyApp.JetstreamConsumer,
64
+
wanted_collections: ["app.bsky.feed.post"]
65
+
}}
66
+
]
67
+
```
69
68
70
-
One of the main reasons for listening to an ATProto relay is to synchronise a
71
-
database with records. As a result, Drinkup provides a light extension around a
72
-
basic consumer, the `RecordConsumer`, which only listens to commit events, and
73
-
transforms them into a slightly nicer structure to work around, calling your
74
-
`handle_create/1`, `handle_update/1`, and `handle_delete/1` functions for each
75
-
record it comes across. It also allows for filtering of specific types of
76
-
records either by full name or with a
77
-
[Regex](https://hexdocs.pm/elixir/1.18.4/Regex.html) match.
69
+
### Tap
78
70
79
71
```elixir
80
-
defmodule ExampleRecordConsumer do
81
-
# Will respond to any events either `app.bsky.feed.post` records, or anything under `app.bsky.graph`.
82
-
use Drinkup.RecordConsumer, collections: [~r/app\.bsky\.graph\..+/, "app.bsky.feed.post"]
83
-
alias Drinkup.RecordConsumer.Record
72
+
defmodule MyApp.TapConsumer do
73
+
@behaviour Drinkup.Tap.Consumer
84
74
85
-
def handle_create(%Record{type: "app.bsky.feed.post"} = record) do
86
-
IO.inspect(record, label: "Bluesky post created")
75
+
def handle_event(%Drinkup.Tap.Event.Record{} = event) do
76
+
IO.inspect(event, label: "Record")
87
77
end
88
78
89
-
def handle_create(%Record{type: "app.bsky.graph" <> _} = record) do
90
-
IO.inspect(record, label: "Bluesky graph updated")
91
-
end
79
+
def handle_event(_), do: :noop
80
+
end
92
81
93
-
def handle_update(record) do
94
-
# ...
95
-
end
82
+
# In your supervision tree:
83
+
children = [
84
+
{Drinkup.Tap, %{
85
+
consumer: MyApp.TapConsumer,
86
+
host: "http://localhost:2480"
87
+
}}
88
+
]
96
89
97
-
def handle_delete(record) do
98
-
# ...
99
-
end
100
-
end
90
+
# Track specific repos:
91
+
Drinkup.Tap.add_repos(Drinkup.Tap, ["did:plc:abc123"])
101
92
```
93
+
94
+
See [the examples](./examples) for some more complete samples.
95
+
96
+
## TODO
97
+
98
+
- Validation for Firehose events (signatures, active account tracking) — see
99
+
[Firehose Validation Best Practices](https://atproto.com/specs/sync#firehose-validation-best-practices)
100
+
- Pluggable cursor persistence (Mnesia, Postgres, etc.)
101
+
- Multi-node/distribution testing
102
+
- More comprehensive test coverage
103
+
- Additional documentation
102
104
103
105
## Special thanks
104
106