comet.sh
Elixir ATProtocol ingestion and sync library.
refactor: integrate consumers and sockets directly
ovyerus.com
+28
-8
AGENTS.md
+28
-8
AGENTS.md
···
24
24
25
25
## Project Structure
26
26
27
-
- **Namespace**: All firehose functionality under `Drinkup.Firehose.*`
28
-
- `Drinkup.Firehose` - Main supervisor
29
-
- `Drinkup.Firehose.Consumer` - Behaviour for handling all events
30
-
- `Drinkup.Firehose.RecordConsumer` - Macro for handling commit record events with filtering
31
-
- `Drinkup.Firehose.Event` - Event types (`Commit`, `Sync`, `Identity`, `Account`, `Info`)
32
-
- `Drinkup.Firehose.Socket` - `:gen_statem` WebSocket connection manager
33
-
- **Consumer Pattern**: Implement `@behaviour Drinkup.Firehose.Consumer` with `handle_event/1`
34
-
- **RecordConsumer Pattern**: `use Drinkup.Firehose.RecordConsumer, collections: [~r/app\.bsky\.graph\..+/, "app.bsky.feed.post"]` with `handle_create/1`, `handle_update/1`, `handle_delete/1` overrides
27
+
Each namespace (`Drinkup.Firehose`, `Drinkup.Jetstream`, `Drinkup.Tap`) follows a common architecture:
28
+
29
+
- **Core Modules**:
30
+
- `Consumer` - Behaviour/macro for handling events; `use Namespace` with `handle_event/1` implementation
31
+
- `Event` - Typed event structs specific to the protocol
32
+
- `Socket` - `:gen_statem` WebSocket connection manager
33
+
- `Options` (or top-level utility module) - Configuration and runtime utilities
34
+
35
+
- **Consumer Pattern**: `use Namespace, opts...` with `handle_event/1` callback; consumer module becomes a supervisor
36
+
37
+
### Namespace-Specific Details
38
+
39
+
- **Firehose** (`Drinkup.Firehose.*`): Full AT Protocol firehose
40
+
- Events: `Commit`, `Sync`, `Identity`, `Account`, `Info`
41
+
- Additional: `RecordConsumer` macro for filtered commit records with `handle_create/1`, `handle_update/1`, `handle_delete/1` callbacks
42
+
- Pattern: `use Drinkup.Firehose.RecordConsumer, collections: [~r/app\.bsky\.graph\..+/, "app.bsky.feed.post"]`
43
+
44
+
- **Jetstream** (`Drinkup.Jetstream.*`): Simplified JSON event stream
45
+
- Events: `Commit`, `Identity`, `Account`
46
+
- Config: `:wanted_collections`, `:wanted_dids`, `:compress` (zstd)
47
+
- Utility: `Drinkup.Jetstream.update_options/2` for dynamic filtering
48
+
- Semantics: Fire-and-forget (no acks)
49
+
50
+
- **Tap** (`Drinkup.Tap.*`): HTTP API + WebSocket indexer/backfill service
51
+
- Events: `Record`, `Identity`
52
+
- Config: `:host`, `:admin_password`, `:disable_acks`
53
+
- Utility: `Drinkup.Tap` HTTP API functions (`add_repos/2`, `remove_repos/2`, `get_repo_info/2`)
54
+
- Semantics: Ack/nack - return `:ok`/`{:ok, _}`/`nil` to ack, `{:error, _}` to nack (Tap retries)
35
55
36
56
## Important Notes
37
57
+2
CHANGELOG.md
+2
CHANGELOG.md
+33
-18
README.md
+33
-18
README.md
···
31
31
32
32
```elixir
33
33
defmodule MyApp.FirehoseConsumer do
34
-
@behaviour Drinkup.Firehose.Consumer
34
+
use Drinkup.Firehose
35
35
36
+
@impl true
36
37
def handle_event(%Drinkup.Firehose.Event.Commit{} = event) do
37
38
IO.inspect(event, label: "Commit")
38
39
end
···
41
42
end
42
43
43
44
# In your supervision tree:
44
-
children = [{Drinkup.Firehose, %{consumer: MyApp.FirehoseConsumer}}]
45
+
children = [MyApp.FirehoseConsumer]
46
+
```
47
+
48
+
For filtered commit events by collection:
49
+
50
+
```elixir
51
+
defmodule MyApp.PostConsumer do
52
+
use Drinkup.Firehose.RecordConsumer,
53
+
collections: ["app.bsky.feed.post"]
54
+
55
+
@impl true
56
+
def handle_create(record) do
57
+
IO.inspect(record, label: "New post")
58
+
end
59
+
end
45
60
```
46
61
47
62
### Jetstream
48
63
49
64
```elixir
50
65
defmodule MyApp.JetstreamConsumer do
51
-
@behaviour Drinkup.Jetstream.Consumer
66
+
use Drinkup.Jetstream,
67
+
wanted_collections: ["app.bsky.feed.post"]
52
68
69
+
@impl true
53
70
def handle_event(%Drinkup.Jetstream.Event.Commit{} = event) do
54
71
IO.inspect(event, label: "Commit")
55
72
end
···
58
75
end
59
76
60
77
# In your supervision tree:
61
-
children = [
62
-
{Drinkup.Jetstream, %{
63
-
consumer: MyApp.JetstreamConsumer,
64
-
wanted_collections: ["app.bsky.feed.post"]
65
-
}}
66
-
]
78
+
children = [MyApp.JetstreamConsumer]
79
+
80
+
# Update filters dynamically:
81
+
Drinkup.Jetstream.update_options(MyApp.JetstreamConsumer, %{
82
+
wanted_collections: ["app.bsky.graph.follow"]
83
+
})
67
84
```
68
85
69
86
### Tap
70
87
71
88
```elixir
72
89
defmodule MyApp.TapConsumer do
73
-
@behaviour Drinkup.Tap.Consumer
90
+
use Drinkup.Tap,
91
+
host: "http://localhost:2480"
74
92
93
+
@impl true
75
94
def handle_event(%Drinkup.Tap.Event.Record{} = event) do
76
95
IO.inspect(event, label: "Record")
96
+
:ok
77
97
end
78
98
79
-
def handle_event(_), do: :noop
99
+
def handle_event(_), do: :ok
80
100
end
81
101
82
102
# In your supervision tree:
83
-
children = [
84
-
{Drinkup.Tap, %{
85
-
consumer: MyApp.TapConsumer,
86
-
host: "http://localhost:2480"
87
-
}}
88
-
]
103
+
children = [MyApp.TapConsumer]
89
104
90
105
# Track specific repos:
91
-
Drinkup.Tap.add_repos(Drinkup.Tap, ["did:plc:abc123"])
106
+
Drinkup.Tap.add_repos(MyTap.TapConsumer, ["did:plc:abc123"])
92
107
```
93
108
94
109
See [the examples](./examples) for some more complete samples.
+3
-4
examples/firehose/basic_consumer.ex
+3
-4
examples/firehose/basic_consumer.ex
···
1
1
defmodule BasicConsumer do
2
-
@behaviour Drinkup.Firehose.Consumer
2
+
use Drinkup.Firehose
3
3
4
+
@impl true
4
5
def handle_event(%Drinkup.Firehose.Event.Commit{} = event) do
5
6
IO.inspect(event, label: "Got commit event")
6
7
end
···
17
18
18
19
@impl true
19
20
def init(_) do
20
-
children = [
21
-
{Drinkup.Firehose, %{consumer: BasicConsumer}}
22
-
]
21
+
children = [BasicConsumer]
23
22
24
23
Supervisor.init(children, strategy: :one_for_one)
25
24
end
+4
-3
examples/firehose/multiple_consumers.ex
+4
-3
examples/firehose/multiple_consumers.ex
···
7
7
end
8
8
9
9
defmodule IdentityConsumer do
10
-
@behaviour Drinkup.Firehose.Consumer
10
+
use Drinkup.Firehose, name: :identities
11
11
12
+
@impl true
12
13
def handle_event(%Drinkup.Firehose.Event.Identity{} = event) do
13
14
IO.inspect(event, label: "identity event")
14
15
end
···
26
27
@impl true
27
28
def init(_) do
28
29
children = [
29
-
{Drinkup.Firehose, %{consumer: PostDeleteConsumer}},
30
-
{Drinkup.Firehose, %{consumer: IdentityConsumer, name: :identities}}
30
+
PostDeleteConsumer,
31
+
IdentityConsumer
31
32
]
32
33
33
34
Supervisor.init(children, strategy: :one_for_one)
+4
-3
examples/firehose/record_consumer.ex
+4
-3
examples/firehose/record_consumer.ex
···
2
2
use Drinkup.Firehose.RecordConsumer,
3
3
collections: [~r/app\.bsky\.graph\..+/, "app.bsky.feed.post"]
4
4
5
+
@impl true
5
6
def handle_create(record) do
6
7
IO.inspect(record, label: "create")
7
8
end
8
9
10
+
@impl true
9
11
def handle_update(record) do
10
12
IO.inspect(record, label: "update")
11
13
end
12
14
15
+
@impl true
13
16
def handle_delete(record) do
14
17
IO.inspect(record, label: "delete")
15
18
end
···
24
27
25
28
@impl true
26
29
def init(_) do
27
-
children = [
28
-
{Drinkup.Firehose, %{consumer: ExampleRecordConsumer}}
29
-
]
30
+
children = [ExampleRecordConsumer]
30
31
31
32
Supervisor.init(children, strategy: :one_for_one)
32
33
end
+13
-12
examples/jetstream/jetstream_consumer.ex
+13
-12
examples/jetstream/jetstream_consumer.ex
···
8
8
- Account events (status changes)
9
9
"""
10
10
11
-
@behaviour Drinkup.Jetstream.Consumer
11
+
use Drinkup.Jetstream,
12
+
name: MyJetstream,
13
+
wanted_collections: ["app.bsky.feed.post", "app.bsky.feed.like"]
12
14
15
+
@impl true
13
16
def handle_event(%Drinkup.Jetstream.Event.Commit{operation: :create} = event) do
14
17
IO.inspect(event, label: "New record created")
15
18
:ok
···
72
75
73
76
@impl true
74
77
def init(_) do
75
-
children = [
76
-
# Connect to public Jetstream instance and filter for posts and likes
77
-
{Drinkup.Jetstream,
78
-
%{
79
-
consumer: JetstreamConsumer,
80
-
name: MyJetstream,
81
-
wanted_collections: ["app.bsky.feed.post", "app.bsky.feed.like"]
82
-
}}
83
-
]
78
+
children = [JetstreamConsumer]
84
79
85
80
Supervisor.init(children, strategy: :one_for_one)
86
81
end
···
88
83
89
84
# Example: Filter for all graph operations (follows, blocks, etc.)
90
85
defmodule GraphEventsConsumer do
91
-
@behaviour Drinkup.Jetstream.Consumer
86
+
use Drinkup.Jetstream,
87
+
name: :graph_events,
88
+
wanted_collections: ["app.bsky.graph.*"]
92
89
90
+
@impl true
93
91
def handle_event(%Drinkup.Jetstream.Event.Commit{collection: "app.bsky.graph." <> _} = event) do
94
92
IO.puts("Graph event: #{event.collection} - #{event.operation}")
95
93
:ok
···
100
98
101
99
# Example: Filter for specific DIDs
102
100
defmodule SpecificDIDConsumer do
103
-
@behaviour Drinkup.Jetstream.Consumer
101
+
use Drinkup.Jetstream,
102
+
name: :specific_dids,
103
+
wanted_dids: ["did:plc:abc123", "did:plc:def456"]
104
104
105
105
@watched_dids [
106
106
"did:plc:abc123",
107
107
"did:plc:def456"
108
108
]
109
109
110
+
@impl true
110
111
def handle_event(%Drinkup.Jetstream.Event.Commit{did: did} = event)
111
112
when did in @watched_dids do
112
113
IO.puts("Activity from watched DID: #{did}")
+8
-10
examples/tap/tap_consumer.ex
+8
-10
examples/tap/tap_consumer.ex
···
1
1
defmodule TapConsumer do
2
-
@behaviour Drinkup.Tap.Consumer
2
+
use Drinkup.Tap,
3
+
name: MyTap,
4
+
host: "http://localhost:2480"
3
5
6
+
@impl true
4
7
def handle_event(%Drinkup.Tap.Event.Record{} = record) do
5
8
IO.inspect(record, label: "Tap record event")
9
+
:ok
6
10
end
7
11
8
12
def handle_event(%Drinkup.Tap.Event.Identity{} = identity) do
9
13
IO.inspect(identity, label: "Tap identity event")
14
+
:ok
10
15
end
11
16
end
12
17
13
-
defmodule TapExampleSupervisor do
18
+
defmodule ExampleTapConsumer do
14
19
use Supervisor
15
20
16
21
def start_link(arg \\ []) do
···
19
24
20
25
@impl true
21
26
def init(_) do
22
-
children = [
23
-
{Drinkup.Tap,
24
-
%{
25
-
consumer: TapConsumer,
26
-
name: MyTap,
27
-
host: "http://localhost:2480"
28
-
}}
29
-
]
27
+
children = [TapConsumer]
30
28
31
29
Supervisor.init(children, strategy: :one_for_one)
32
30
end
+133
-25
lib/firehose.ex
+133
-25
lib/firehose.ex
···
1
1
defmodule Drinkup.Firehose do
2
-
use Supervisor
3
-
alias Drinkup.Firehose.Options
2
+
@moduledoc """
3
+
Module for handling events from the AT Protocol [firehose](https://docs.bsky.app/docs/advanced-guides/firehose).
4
+
5
+
Due to the nature of the firehose, this will result in a lot of incoming
6
+
traffic as it receives every repo and identity event within the network. If
7
+
you're concerened about bandwidth constaints or just don't need a
8
+
whole-network sync, you may be better off using `Drinkup.Jetstream` or
9
+
`Drinkup.Tap`.
10
+
11
+
## Usage
12
+
13
+
defmodule MyFirehoseConsumer do
14
+
use Drinkup.Firehose,
15
+
name: :my_firehose,
16
+
host: "https://bsky.network",
17
+
cursor: nil
18
+
19
+
@impl true
20
+
def handle_event(%Drinkup.Firehose.Event.Commit{} = event) do
21
+
IO.inspect(event, label: "Commit")
22
+
:ok
23
+
end
24
+
25
+
def handle_event(_event), do: :ok
26
+
end
27
+
28
+
# In your application supervision tree:
29
+
children = [MyFirehoseConsumer]
30
+
31
+
Exceptions raised by `handle_event/1` will be logged instead of killing and
32
+
restarting the socket process.
33
+
34
+
## Options
35
+
36
+
- `:name` - Unique name for this Firehose instance (default: the module name)
37
+
- `:host` - Firehose relay URL (default: `"https://bsky.network"`)
38
+
- `:cursor` - Optional sequence number to resume streaming from
39
+
40
+
## Runtime Configuration
41
+
42
+
You can override options at runtime by providing them to `child_spec/1`:
4
43
5
-
@dialyzer nowarn_function: {:init, 1}
6
-
@impl true
7
-
def init({%Options{name: name} = drinkup_options, supervisor_options}) do
8
-
children = [
9
-
{Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, Tasks}}}},
10
-
{Drinkup.Firehose.Socket, drinkup_options}
11
-
]
44
+
children = [
45
+
{MyFirehoseConsumer, name: :runtime_name, cursor: 12345}
46
+
]
12
47
13
-
Supervisor.start_link(
14
-
children,
15
-
supervisor_options ++ [name: {:via, Registry, {Drinkup.Registry, {name, Supervisor}}}]
16
-
)
17
-
end
48
+
## Event Types
49
+
50
+
`handle_event/1` will receive the following event structs:
18
51
19
-
@spec child_spec(Options.options()) :: Supervisor.child_spec()
20
-
def child_spec(%{} = options), do: child_spec({options, [strategy: :one_for_one]})
52
+
- `Drinkup.Firehose.Event.Commit` - Repository commits
53
+
- `Drinkup.Firehose.Event.Sync` - Sync events
54
+
- `Drinkup.Firehose.Event.Identity` - Identity updates
55
+
- `Drinkup.Firehose.Event.Account` - Account status changes
56
+
- `Drinkup.Firehose.Event.Info` - Info messages
57
+
"""
21
58
22
-
@spec child_spec({Options.options(), Keyword.t()}) :: Supervisor.child_spec()
23
-
def child_spec({drinkup_options, supervisor_options}) do
24
-
%{
25
-
id: Map.get(drinkup_options, :name, __MODULE__),
26
-
start: {__MODULE__, :init, [{Options.from(drinkup_options), supervisor_options}]},
27
-
type: :supervisor,
28
-
restart: :permanent,
29
-
shutdown: 500
30
-
}
59
+
defmacro __using__(opts) do
60
+
quote location: :keep, bind_quoted: [opts: opts] do
61
+
use Supervisor
62
+
@behaviour Drinkup.Firehose.Consumer
63
+
64
+
alias Drinkup.Firehose.Options
65
+
66
+
# Store compile-time options as module attributes
67
+
@name Keyword.get(opts, :name)
68
+
@host Keyword.get(opts, :host, "https://bsky.network")
69
+
@cursor Keyword.get(opts, :cursor)
70
+
71
+
@doc """
72
+
Starts the Firehose consumer supervisor.
73
+
74
+
Accepts optional runtime configuration that overrides compile-time options.
75
+
"""
76
+
def start_link(runtime_opts \\ []) do
77
+
# Merge compile-time and runtime options
78
+
opts = build_options(runtime_opts)
79
+
Supervisor.start_link(__MODULE__, opts, name: via_tuple(opts.name))
80
+
end
81
+
82
+
@impl true
83
+
def init(%Options{name: name} = options) do
84
+
children = [
85
+
{Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, Tasks}}}},
86
+
{Drinkup.Firehose.Socket, options}
87
+
]
88
+
89
+
Supervisor.init(children, strategy: :one_for_one)
90
+
end
91
+
92
+
@doc """
93
+
Returns a child spec for adding this consumer to a supervision tree.
94
+
95
+
Runtime options override compile-time options.
96
+
"""
97
+
def child_spec(runtime_opts) when is_list(runtime_opts) do
98
+
opts = build_options(runtime_opts)
99
+
100
+
%{
101
+
id: opts.name,
102
+
start: {__MODULE__, :start_link, [runtime_opts]},
103
+
type: :supervisor,
104
+
restart: :permanent,
105
+
shutdown: 500
106
+
}
107
+
end
108
+
109
+
def child_spec(_opts) do
110
+
raise ArgumentError, "child_spec expects a keyword list of options"
111
+
end
112
+
113
+
defoverridable child_spec: 1
114
+
115
+
# Build Options struct from compile-time and runtime options
116
+
defp build_options(runtime_opts) do
117
+
# Compile-time defaults
118
+
compile_opts = [
119
+
name: @name || __MODULE__,
120
+
host: @host,
121
+
cursor: @cursor
122
+
]
123
+
124
+
# Merge with runtime opts (runtime takes precedence)
125
+
merged =
126
+
compile_opts
127
+
|> Keyword.merge(runtime_opts)
128
+
|> Enum.reject(fn {_k, v} -> is_nil(v) end)
129
+
|> Map.new()
130
+
|> Map.put(:consumer, __MODULE__)
131
+
132
+
Options.from(merged)
133
+
end
134
+
135
+
defp via_tuple(name) do
136
+
{:via, Registry, {Drinkup.Registry, {name, Supervisor}}}
137
+
end
138
+
end
31
139
end
32
140
end
+4
-4
lib/firehose/consumer.ex
+4
-4
lib/firehose/consumer.ex
···
1
1
defmodule Drinkup.Firehose.Consumer do
2
2
@moduledoc """
3
-
An unopinionated consumer of the Firehose. Will receive all events, not just commits.
4
-
"""
3
+
Behaviour for handling Firehose events.
5
4
6
-
alias Drinkup.Firehose.Event
5
+
Implemented by `Drinkup.Firehose`, you'll likely want to be using that instead.
6
+
"""
7
7
8
-
@callback handle_event(Event.t()) :: any()
8
+
@callback handle_event(Drinkup.Firehose.Event.t()) :: any()
9
9
end
+48
-3
lib/firehose/record_consumer.ex
+48
-3
lib/firehose/record_consumer.ex
···
1
1
defmodule Drinkup.Firehose.RecordConsumer do
2
2
@moduledoc """
3
-
An opinionated consumer of the Firehose that eats consumers
3
+
Opinionated consumer of the Firehose focused on record operations.
4
+
5
+
This is an abstraction over the core `Drinkup.Firehose` implementation
6
+
designed for easily handling `commit` events, with the ability to filter by
7
+
collection. It's similiar to `Drinkup.Jetstream`, but using the Firehose
8
+
directly (and currently more naive).
9
+
10
+
## Example
11
+
12
+
defmodule MyRecordConsumer do
13
+
use Drinkup.Firehose.RecordConsumer,
14
+
collections: ["app.bsky.feed.post", ~r/app\\.bsky\\.graph\\..+/],
15
+
name: :my_records,
16
+
host: "https://bsky.network"
17
+
18
+
@impl true
19
+
def handle_create(record) do
20
+
IO.inspect(record, label: "New record")
21
+
end
22
+
23
+
@impl true
24
+
def handle_delete(record) do
25
+
IO.inspect(record, label: "Deleted record")
26
+
end
27
+
end
28
+
29
+
# In your application supervision tree:
30
+
children = [MyRecordConsumer]
31
+
32
+
## Options
33
+
34
+
All options from `Drinkup.Firehose` are supported, plus:
35
+
36
+
- `:collections` - List of collection NSIDs (strings or regexes) to filter. If
37
+
empty or not provided, all collections are processed.
38
+
39
+
## Callbacks
40
+
41
+
Implement these callbacks to handle different record actions:
42
+
43
+
- `handle_create/1` - Called when a record is created
44
+
- `handle_update/1` - Called when a record is updated
45
+
- `handle_delete/1` - Called when a record is deleted
46
+
47
+
All callbacks receive a `Drinkup.Firehose.RecordConsumer.Record` struct.
4
48
"""
5
49
6
50
@callback handle_create(any()) :: any()
···
8
52
@callback handle_delete(any()) :: any()
9
53
10
54
defmacro __using__(opts) do
11
-
{collections, _opts} = Keyword.pop(opts, :collections, [])
55
+
{collections, firehose_opts} = Keyword.pop(opts, :collections, [])
12
56
13
57
quote location: :keep do
14
-
@behaviour Drinkup.Firehose.Consumer
58
+
use Drinkup.Firehose, unquote(firehose_opts)
15
59
@behaviour Drinkup.Firehose.RecordConsumer
16
60
61
+
@impl true
17
62
def handle_event(%Drinkup.Firehose.Event.Commit{} = event) do
18
63
event.ops
19
64
|> Enum.filter(fn %{path: path} ->
+121
-55
lib/jetstream.ex
+121
-55
lib/jetstream.ex
···
1
1
defmodule Drinkup.Jetstream do
2
2
@moduledoc """
3
-
Supervisor for Jetstream event stream connections.
3
+
Module for handling events from an AT Protocol
4
+
[Jetstream](https://github.com/bluesky-social/jetstream) instance.
4
5
5
-
Jetstream is a simplified JSON event stream that converts the CBOR-encoded
6
-
ATProto Firehose into lightweight, friendly JSON events. It provides zstd
7
-
compression and filtering capabilities for collections and DIDs.
6
+
Jetstream is an abstraction over the raw AT Protocol firehose that converts
7
+
the CBOR-encoded events into easier to handle JSON objects, and also provides
8
+
the ability to filter the events received by repository DID or collection
9
+
NSID. This is useful when you know specifically which repos or collections you
10
+
want events from, and thus reduces the amount of bandwidth consumed vs
11
+
consuming the raw firehose directly.
12
+
13
+
If you need a solution for easy backfilling from repositories and not just a
14
+
firehose translation layer, check out `Drinkup.Tap`.
8
15
9
16
## Usage
10
17
11
-
Add Jetstream to your supervision tree:
18
+
defmodule MyJetstreamConsumer do
19
+
use Drinkup.Jetstream,
20
+
name: :my_jetstream,
21
+
wanted_collections: ["app.bsky.feed.post"]
12
22
13
-
children = [
14
-
{Drinkup.Jetstream, %{
15
-
consumer: MyJetstreamConsumer,
16
-
name: MyJetstream,
17
-
wanted_collections: ["app.bsky.feed.post", "app.bsky.feed.like"]
18
-
}}
19
-
]
23
+
@impl true
24
+
def handle_event(event) do
25
+
IO.inspect(event)
26
+
end
27
+
end
28
+
29
+
# In your application supervision tree:
30
+
children = [MyJetstreamConsumer]
20
31
21
32
## Configuration
22
33
23
-
See `Drinkup.Jetstream.Options` for all available configuration options.
34
+
See `Drinkup.Jetstream.Consumer` for all available configuration options.
24
35
25
36
## Dynamic Filter Updates
26
37
27
38
You can update filters after the connection is established:
28
39
29
-
Drinkup.Jetstream.update_options(MyJetstream, %{
40
+
Drinkup.Jetstream.update_options(:my_jetstream, %{
30
41
wanted_collections: ["app.bsky.graph.follow"],
31
42
wanted_dids: ["did:plc:abc123"]
32
43
})
···
36
47
By default Drinkup connects to `jetstream2.us-east.bsky.network`.
37
48
38
49
Bluesky operates a few different Jetstream instances:
39
-
- `jetstream1.us-east.bsky.network`
40
-
- `jetstream2.us-east.bsky.network`
41
-
- `jetstream1.us-west.bsky.network`
42
-
- `jetstream2.us-west.bsky.network`
50
+
- `wss://jetstream1.us-east.bsky.network`
51
+
- `wss://jetstream2.us-east.bsky.network`
52
+
- `wss://jetstream1.us-west.bsky.network`
53
+
- `wss://jetstream2.us-west.bsky.network`
43
54
44
-
There also some third-party instances not run by Bluesky PBC:
45
-
- `jetstream.fire.hose.cam`
46
-
- `jetstream2.fr.hose.cam`
47
-
- `jetstream1.us-east.fire.hose.cam`
55
+
There also some third-party instances not run by Bluesky PBC, including but not limited to:
56
+
- `wss://jetstream.fire.hose.cam`
57
+
- `wss://jetstream2.fr.hose.cam`
58
+
- `wss://jetstream1.us-east.fire.hose.cam`
59
+
60
+
https://firehose.stream/ also hosts several instances around the world.
48
61
"""
49
62
50
-
use Supervisor
51
63
require Logger
52
-
alias Drinkup.Jetstream.Options
53
64
54
-
@dialyzer nowarn_function: {:init, 1}
65
+
defmacro __using__(opts) do
66
+
quote location: :keep, bind_quoted: [opts: opts] do
67
+
use Supervisor
68
+
@behaviour Drinkup.Jetstream.Consumer
55
69
56
-
@impl true
57
-
def init({%Options{name: name} = drinkup_options, supervisor_options}) do
58
-
children = [
59
-
{Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, JetstreamTasks}}}},
60
-
{Drinkup.Jetstream.Socket, drinkup_options}
61
-
]
70
+
alias Drinkup.Jetstream.Options
62
71
63
-
Supervisor.start_link(
64
-
children,
65
-
supervisor_options ++
66
-
[name: {:via, Registry, {Drinkup.Registry, {name, JetstreamSupervisor}}}]
67
-
)
68
-
end
72
+
# Store compile-time options as module attributes
73
+
@name Keyword.get(opts, :name)
74
+
@host Keyword.get(opts, :host, "wss://jetstream2.us-east.bsky.network")
75
+
@wanted_collections Keyword.get(opts, :wanted_collections, [])
76
+
@wanted_dids Keyword.get(opts, :wanted_dids, [])
77
+
@cursor Keyword.get(opts, :cursor)
78
+
@require_hello Keyword.get(opts, :require_hello, false)
79
+
@max_message_size_bytes Keyword.get(opts, :max_message_size_bytes)
69
80
70
-
@spec child_spec(Options.options()) :: Supervisor.child_spec()
71
-
def child_spec(%{} = options), do: child_spec({options, [strategy: :one_for_one]})
81
+
@doc """
82
+
Starts the Jetstream consumer supervisor.
72
83
73
-
@spec child_spec({Options.options(), Keyword.t()}) :: Supervisor.child_spec()
74
-
def child_spec({drinkup_options, supervisor_options}) do
75
-
%{
76
-
id: Map.get(drinkup_options, :name, __MODULE__),
77
-
start: {__MODULE__, :init, [{Options.from(drinkup_options), supervisor_options}]},
78
-
type: :supervisor,
79
-
restart: :permanent,
80
-
shutdown: 500
81
-
}
84
+
Accepts optional runtime configuration that overrides compile-time options.
85
+
"""
86
+
def start_link(runtime_opts \\ []) do
87
+
opts = build_options(runtime_opts)
88
+
Supervisor.start_link(__MODULE__, opts, name: via_tuple(opts.name))
89
+
end
90
+
91
+
@impl true
92
+
def init(%Options{name: name} = options) do
93
+
children = [
94
+
{Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, JetstreamTasks}}}},
95
+
{Drinkup.Jetstream.Socket, options}
96
+
]
97
+
98
+
Supervisor.init(children, strategy: :one_for_one)
99
+
end
100
+
101
+
@doc """
102
+
Returns a child spec for adding this consumer to a supervision tree.
103
+
104
+
Runtime options override compile-time options.
105
+
"""
106
+
def child_spec(runtime_opts) when is_list(runtime_opts) do
107
+
opts = build_options(runtime_opts)
108
+
109
+
%{
110
+
id: opts.name,
111
+
start: {__MODULE__, :start_link, [runtime_opts]},
112
+
type: :supervisor,
113
+
restart: :permanent,
114
+
shutdown: 500
115
+
}
116
+
end
117
+
118
+
def child_spec(_opts) do
119
+
raise ArgumentError, "child_spec expects a keyword list of options"
120
+
end
121
+
122
+
defoverridable child_spec: 1
123
+
124
+
# Build Options struct from compile-time and runtime options
125
+
defp build_options(runtime_opts) do
126
+
compile_opts = [
127
+
name: @name || __MODULE__,
128
+
host: @host,
129
+
wanted_collections: @wanted_collections,
130
+
wanted_dids: @wanted_dids,
131
+
cursor: @cursor,
132
+
require_hello: @require_hello,
133
+
max_message_size_bytes: @max_message_size_bytes
134
+
]
135
+
136
+
merged =
137
+
compile_opts
138
+
|> Keyword.merge(runtime_opts)
139
+
|> Enum.reject(fn {_k, v} -> is_nil(v) end)
140
+
|> Map.new()
141
+
|> Map.put(:consumer, __MODULE__)
142
+
143
+
Options.from(merged)
144
+
end
145
+
146
+
defp via_tuple(name) do
147
+
{:via, Registry, {Drinkup.Registry, {name, JetstreamSupervisor}}}
148
+
end
149
+
end
82
150
end
83
151
84
152
# Options Update API
···
107
175
108
176
## Parameters
109
177
110
-
- `name` - The name of the Jetstream instance (default: `Drinkup.Jetstream`)
178
+
- `name` - The name of the Jetstream consumer (the `:name` option passed to `use Drinkup.Jetstream`)
111
179
- `opts` - Map with optional fields:
112
180
- `:wanted_collections` - List of collection NSIDs or prefixes (max 100)
113
181
- `:wanted_dids` - List of DIDs to filter (max 10,000)
···
116
184
## Examples
117
185
118
186
# Filter to only posts
119
-
Drinkup.Jetstream.update_options(MyJetstream, %{
187
+
Drinkup.Jetstream.update_options(:my_jetstream, %{
120
188
wanted_collections: ["app.bsky.feed.post"]
121
189
})
122
190
123
191
# Filter to specific DIDs
124
-
Drinkup.Jetstream.update_options(MyJetstream, %{
192
+
Drinkup.Jetstream.update_options(:my_jetstream, %{
125
193
wanted_dids: ["did:plc:abc123", "did:plc:def456"]
126
194
})
127
195
128
196
# Disable all filters (receive all events)
129
-
Drinkup.Jetstream.update_options(MyJetstream, %{
197
+
Drinkup.Jetstream.update_options(:my_jetstream, %{
130
198
wanted_collections: [],
131
199
wanted_dids: []
132
200
})
···
140
208
Invalid updates will result in the connection being closed by the server.
141
209
"""
142
210
@spec update_options(atom(), update_opts()) :: :ok | {:error, term()}
143
-
def update_options(name \\ Drinkup.Jetstream, opts) when is_map(opts) do
211
+
def update_options(name, opts) when is_atom(name) and is_map(opts) do
144
212
case find_connection(name) do
145
213
{:ok, {conn, stream}} ->
146
214
message = build_options_update_message(opts)
···
153
221
{:error, reason}
154
222
end
155
223
end
156
-
157
-
# Private functions
158
224
159
225
@spec find_connection(atom()) :: {:ok, {pid(), :gun.stream_ref()}} | {:error, :not_connected}
160
226
defp find_connection(name) do
+2
-52
lib/jetstream/consumer.ex
+2
-52
lib/jetstream/consumer.ex
···
1
1
defmodule Drinkup.Jetstream.Consumer do
2
2
@moduledoc """
3
-
Consumer behaviour for handling Jetstream events.
4
-
5
-
Implement this behaviour to process events from a Jetstream instance.
6
-
Events are dispatched asynchronously via `Task.Supervisor`.
7
-
8
-
Unlike Tap, Jetstream does not require event acknowledgments. Events are
9
-
processed in a fire-and-forget manner.
10
-
11
-
## Example
12
-
13
-
defmodule MyJetstreamConsumer do
14
-
@behaviour Drinkup.Jetstream.Consumer
15
-
16
-
def handle_event(%Drinkup.Jetstream.Event.Commit{operation: :create} = event) do
17
-
# Handle new record creation
18
-
IO.inspect(event, label: "New record")
19
-
:ok
20
-
end
21
-
22
-
def handle_event(%Drinkup.Jetstream.Event.Commit{operation: :delete} = event) do
23
-
# Handle record deletion
24
-
IO.inspect(event, label: "Deleted record")
25
-
:ok
26
-
end
27
-
28
-
def handle_event(%Drinkup.Jetstream.Event.Identity{} = event) do
29
-
# Handle identity changes
30
-
IO.inspect(event, label: "Identity update")
31
-
:ok
32
-
end
33
-
34
-
def handle_event(%Drinkup.Jetstream.Event.Account{active: false} = event) do
35
-
# Handle account deactivation
36
-
IO.inspect(event, label: "Account inactive")
37
-
:ok
38
-
end
3
+
Behaviour for handling Jetstream events.
39
4
40
-
def handle_event(_event), do: :ok
41
-
end
42
-
43
-
## Event Types
44
-
45
-
The consumer will receive one of three event types:
46
-
47
-
- `Drinkup.Jetstream.Event.Commit` - Repository commits (create, update, delete)
48
-
- `Drinkup.Jetstream.Event.Identity` - Identity updates (handle changes, etc.)
49
-
- `Drinkup.Jetstream.Event.Account` - Account status changes (active, taken down, etc.)
50
-
51
-
## Error Handling
52
-
53
-
If your `handle_event/1` implementation raises an exception, it will be logged
54
-
but will not affect the stream. The error is caught and logged by the event
55
-
dispatcher.
5
+
Implemented by `Drinkup.Jetstream`, you'll likely want to be using that instead.
56
6
"""
57
7
58
8
alias Drinkup.Jetstream.Event
+159
-61
lib/tap.ex
+159
-61
lib/tap.ex
···
1
1
defmodule Drinkup.Tap do
2
2
@moduledoc """
3
-
Supervisor and HTTP API for Tap indexer/backfill service.
3
+
Module for handling events from a
4
+
[Tap](https://github.com/bluesky-social/indigo/tree/main/cmd/tap) instance.
4
5
5
-
Tap simplifies AT sync by handling the firehose connection, verification,
6
-
backfill, and filtering. Your application connects to a Tap service and
7
-
receives simple JSON events for only the repos and collections you care about.
6
+
Tap is a complete sync and backfill solution which handles the firehose
7
+
connection itself, and automatically searches for repositories to backfill
8
+
from based on the options given to it. It's great for building an app that
9
+
wants all of a certain set of records within the AT Protocol network.
10
+
11
+
This module requires you to be running a properly configured Tap instance, it
12
+
doesn't spawn one for itself.
8
13
9
14
## Usage
10
15
11
-
Add Tap to your supervision tree:
16
+
defmodule MyTapConsumer do
17
+
use Drinkup.Tap,
18
+
name: :my_tap,
19
+
host: "http://localhost:2480",
20
+
admin_password: System.get_env("TAP_PASSWORD")
12
21
13
-
children = [
14
-
{Drinkup.Tap, %{
15
-
consumer: MyTapConsumer,
16
-
name: MyTap,
17
-
host: "http://localhost:2480",
18
-
admin_password: "secret" # optional
19
-
}}
20
-
]
22
+
@impl true
23
+
def handle_event(event) do
24
+
# Process event
25
+
:ok
26
+
end
27
+
end
28
+
29
+
# In your application supervision tree:
30
+
children = [MyTapConsumer]
21
31
22
-
Then interact with the Tap HTTP API:
32
+
You can also interact with the Tap HTTP API to manually start tracking
33
+
specific repositories or get information about what's going on.
23
34
24
35
# Add repos to track (triggers backfill)
25
-
Drinkup.Tap.add_repos(MyTap, ["did:plc:abc123"])
36
+
Drinkup.Tap.add_repos(:my_tap, ["did:plc:abc123"])
26
37
27
38
# Get stats
28
-
{:ok, count} = Drinkup.Tap.get_repo_count(MyTap)
29
-
30
-
## Configuration
31
-
32
-
Tap itself is configured via environment variables. See the Tap documentation
33
-
for details on configuring collection filters, signal collections, and other
34
-
operational settings:
35
-
https://github.com/bluesky-social/indigo/blob/main/cmd/tap/README.md
39
+
{:ok, count} = Drinkup.Tap.get_repo_count(:my_tap)
36
40
"""
37
41
38
-
use Supervisor
39
42
alias Drinkup.Tap.Options
40
43
41
-
@dialyzer nowarn_function: {:init, 1}
42
-
@impl true
43
-
def init({%Options{name: name} = drinkup_options, supervisor_options}) do
44
-
# Register options in Registry for HTTP API access
45
-
Registry.register(Drinkup.Registry, {name, TapOptions}, drinkup_options)
44
+
defmacro __using__(opts) do
45
+
quote location: :keep, bind_quoted: [opts: opts] do
46
+
use Supervisor
47
+
@behaviour Drinkup.Tap.Consumer
46
48
47
-
children = [
48
-
{Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, TapTasks}}}},
49
-
{Drinkup.Tap.Socket, drinkup_options}
50
-
]
49
+
alias Drinkup.Tap.Options
50
+
51
+
# Store compile-time options as module attributes
52
+
@name Keyword.get(opts, :name)
53
+
@host Keyword.get(opts, :host, "http://localhost:2480")
54
+
@admin_password Keyword.get(opts, :admin_password)
55
+
@disable_acks Keyword.get(opts, :disable_acks, false)
56
+
57
+
@doc """
58
+
Starts the Tap consumer supervisor.
59
+
60
+
Accepts optional runtime configuration that overrides compile-time options.
61
+
"""
62
+
def start_link(runtime_opts \\ []) do
63
+
opts = build_options(runtime_opts)
64
+
Supervisor.start_link(__MODULE__, opts, name: via_tuple(opts.name))
65
+
end
66
+
67
+
@impl true
68
+
def init(%Options{name: name} = options) do
69
+
# Register options in Registry for HTTP API access
70
+
Registry.register(Drinkup.Registry, {name, TapOptions}, options)
51
71
52
-
Supervisor.start_link(
53
-
children,
54
-
supervisor_options ++ [name: {:via, Registry, {Drinkup.Registry, {name, TapSupervisor}}}]
55
-
)
56
-
end
72
+
children = [
73
+
{Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, TapTasks}}}},
74
+
{Drinkup.Tap.Socket, options}
75
+
]
57
76
58
-
@spec child_spec(Options.options()) :: Supervisor.child_spec()
59
-
def child_spec(%{} = options), do: child_spec({options, [strategy: :one_for_one]})
77
+
Supervisor.init(children, strategy: :one_for_one)
78
+
end
60
79
61
-
@spec child_spec({Options.options(), Keyword.t()}) :: Supervisor.child_spec()
62
-
def child_spec({drinkup_options, supervisor_options}) do
63
-
%{
64
-
id: Map.get(drinkup_options, :name, __MODULE__),
65
-
start: {__MODULE__, :init, [{Options.from(drinkup_options), supervisor_options}]},
66
-
type: :supervisor,
67
-
restart: :permanent,
68
-
shutdown: 500
69
-
}
80
+
@doc """
81
+
Returns a child spec for adding this consumer to a supervision tree.
82
+
83
+
Runtime options override compile-time options.
84
+
"""
85
+
def child_spec(runtime_opts) when is_list(runtime_opts) do
86
+
opts = build_options(runtime_opts)
87
+
88
+
%{
89
+
id: opts.name,
90
+
start: {__MODULE__, :start_link, [runtime_opts]},
91
+
type: :supervisor,
92
+
restart: :permanent,
93
+
shutdown: 500
94
+
}
95
+
end
96
+
97
+
def child_spec(_opts) do
98
+
raise ArgumentError, "child_spec expects a keyword list of options"
99
+
end
100
+
101
+
defoverridable child_spec: 1
102
+
103
+
# Build Options struct from compile-time and runtime options
104
+
defp build_options(runtime_opts) do
105
+
compile_opts = [
106
+
name: @name || __MODULE__,
107
+
host: @host,
108
+
admin_password: @admin_password,
109
+
disable_acks: @disable_acks
110
+
]
111
+
112
+
merged =
113
+
compile_opts
114
+
|> Keyword.merge(runtime_opts)
115
+
|> Enum.reject(fn {_k, v} -> is_nil(v) end)
116
+
|> Map.new()
117
+
|> Map.put(:consumer, __MODULE__)
118
+
119
+
Options.from(merged)
120
+
end
121
+
122
+
defp via_tuple(name) do
123
+
{:via, Registry, {Drinkup.Registry, {name, TapSupervisor}}}
124
+
end
125
+
end
70
126
end
71
127
72
128
# HTTP API Functions
···
76
132
77
133
Triggers backfill for the specified DIDs. Historical events will be fetched
78
134
from each repo's PDS, followed by live events from the firehose.
135
+
136
+
## Parameters
137
+
138
+
- `name` - The name of the Tap consumer (the `:name` option passed to `use Drinkup.Tap`)
139
+
- `dids` - List of DID strings to add
79
140
"""
80
141
@spec add_repos(atom(), [String.t()]) :: {:ok, term()} | {:error, term()}
81
-
def add_repos(name \\ Drinkup.Tap, dids) when is_list(dids) do
142
+
def add_repos(name, dids) when is_atom(name) and is_list(dids) do
82
143
with {:ok, options} <- get_options(name),
83
144
{:ok, response} <- make_request(options, :post, "/repos/add", %{dids: dids}) do
84
145
{:ok, response}
···
90
151
91
152
Stops syncing the specified repos and deletes tracked repo metadata. Does not
92
153
delete buffered events in the outbox.
154
+
155
+
## Parameters
156
+
157
+
- `name` - The name of the Tap consumer (the `:name` option passed to `use Drinkup.Tap`)
158
+
- `dids` - List of DID strings to remove
93
159
"""
94
160
@spec remove_repos(atom(), [String.t()]) :: {:ok, term()} | {:error, term()}
95
-
def remove_repos(name \\ Drinkup.Tap, dids) when is_list(dids) do
161
+
def remove_repos(name, dids) when is_atom(name) and is_list(dids) do
96
162
with {:ok, options} <- get_options(name),
97
163
{:ok, response} <- make_request(options, :post, "/repos/remove", %{dids: dids}) do
98
164
{:ok, response}
···
101
167
102
168
@doc """
103
169
Resolve a DID to its DID document.
170
+
171
+
## Parameters
172
+
173
+
- `name` - The name of the Tap consumer
174
+
- `did` - DID string to resolve
104
175
"""
105
176
@spec resolve_did(atom(), String.t()) :: {:ok, term()} | {:error, term()}
106
-
def resolve_did(name \\ Drinkup.Tap, did) when is_binary(did) do
177
+
def resolve_did(name, did) when is_atom(name) and is_binary(did) do
107
178
with {:ok, options} <- get_options(name),
108
179
{:ok, response} <- make_request(options, :get, "/resolve/#{did}") do
109
180
{:ok, response}
···
114
185
Get info about a tracked repo.
115
186
116
187
Returns repo state, repo rev, record count, error info, and retry count.
188
+
189
+
## Parameters
190
+
191
+
- `name` - The name of the Tap consumer
192
+
- `did` - DID string to get info for
117
193
"""
118
194
@spec get_repo_info(atom(), String.t()) :: {:ok, term()} | {:error, term()}
119
-
def get_repo_info(name \\ Drinkup.Tap, did) when is_binary(did) do
195
+
def get_repo_info(name, did) when is_atom(name) and is_binary(did) do
120
196
with {:ok, options} <- get_options(name),
121
197
{:ok, response} <- make_request(options, :get, "/info/#{did}") do
122
198
{:ok, response}
···
125
201
126
202
@doc """
127
203
Get the total number of tracked repos.
204
+
205
+
## Parameters
206
+
207
+
- `name` - The name of the Tap consumer
128
208
"""
129
209
@spec get_repo_count(atom()) :: {:ok, integer()} | {:error, term()}
130
-
def get_repo_count(name \\ Drinkup.Tap) do
210
+
def get_repo_count(name) when is_atom(name) do
131
211
with {:ok, options} <- get_options(name),
132
212
{:ok, response} <- make_request(options, :get, "/stats/repo-count") do
133
213
{:ok, response}
···
136
216
137
217
@doc """
138
218
Get the total number of tracked records.
219
+
220
+
## Parameters
221
+
222
+
- `name` - The name of the Tap consumer
139
223
"""
140
224
@spec get_record_count(atom()) :: {:ok, integer()} | {:error, term()}
141
-
def get_record_count(name \\ Drinkup.Tap) do
225
+
def get_record_count(name) when is_atom(name) do
142
226
with {:ok, options} <- get_options(name),
143
227
{:ok, response} <- make_request(options, :get, "/stats/record-count") do
144
228
{:ok, response}
···
147
231
148
232
@doc """
149
233
Get the number of events in the outbox buffer.
234
+
235
+
## Parameters
236
+
237
+
- `name` - The name of the Tap consumer
150
238
"""
151
239
@spec get_outbox_buffer(atom()) :: {:ok, integer()} | {:error, term()}
152
-
def get_outbox_buffer(name \\ Drinkup.Tap) do
240
+
def get_outbox_buffer(name) when is_atom(name) do
153
241
with {:ok, options} <- get_options(name),
154
242
{:ok, response} <- make_request(options, :get, "/stats/outbox-buffer") do
155
243
{:ok, response}
···
158
246
159
247
@doc """
160
248
Get the number of events in the resync buffer.
249
+
250
+
## Parameters
251
+
252
+
- `name` - The name of the Tap consumer
161
253
"""
162
254
@spec get_resync_buffer(atom()) :: {:ok, integer()} | {:error, term()}
163
-
def get_resync_buffer(name \\ Drinkup.Tap) do
255
+
def get_resync_buffer(name) when is_atom(name) do
164
256
with {:ok, options} <- get_options(name),
165
257
{:ok, response} <- make_request(options, :get, "/stats/resync-buffer") do
166
258
{:ok, response}
···
169
261
170
262
@doc """
171
263
Get current firehose and list repos cursors.
264
+
265
+
## Parameters
266
+
267
+
- `name` - The name of the Tap consumer
172
268
"""
173
269
@spec get_cursors(atom()) :: {:ok, map()} | {:error, term()}
174
-
def get_cursors(name \\ Drinkup.Tap) do
270
+
def get_cursors(name) when is_atom(name) do
175
271
with {:ok, options} <- get_options(name),
176
272
{:ok, response} <- make_request(options, :get, "/stats/cursors") do
177
273
{:ok, response}
···
182
278
Check Tap health status.
183
279
184
280
Returns `{:ok, %{"status" => "ok"}}` if healthy.
281
+
282
+
## Parameters
283
+
284
+
- `name` - The name of the Tap consumer
185
285
"""
186
286
@spec health(atom()) :: {:ok, map()} | {:error, term()}
187
-
def health(name \\ Drinkup.Tap) do
287
+
def health(name) when is_atom(name) do
188
288
with {:ok, options} <- get_options(name),
189
289
{:ok, response} <- make_request(options, :get, "/health") do
190
290
{:ok, response}
191
291
end
192
292
end
193
-
194
-
# Private Functions
195
293
196
294
@spec get_options(atom()) :: {:ok, Options.t()} | {:error, :not_found}
197
295
defp get_options(name) do
+2
-37
lib/tap/consumer.ex
+2
-37
lib/tap/consumer.ex
···
1
1
defmodule Drinkup.Tap.Consumer do
2
2
@moduledoc """
3
-
Consumer behaviour for handling Tap events.
4
-
5
-
Implement this behaviour to process events from a Tap indexer/backfill service.
6
-
Events are dispatched asynchronously via `Task.Supervisor` and acknowledged
7
-
to Tap based on the return value of `handle_event/1`.
8
-
9
-
## Event Acknowledgment
10
-
11
-
By default, events are acknowledged to Tap based on your return value:
12
-
13
-
- `:ok`, `{:ok, any()}`, or `nil` → Success, event is acked to Tap
14
-
- `{:error, reason}` → Failure, event is NOT acked (Tap will retry after timeout)
15
-
- Exception raised → Failure, event is NOT acked (Tap will retry after timeout)
16
-
17
-
Any other value will log a warning and acknowledge the event anyway.
3
+
Behaviour for handling Tap events.
18
4
19
-
If you set `disable_acks: true` in your Tap options, no acks are sent regardless
20
-
of the return value. This matches Tap's `TAP_DISABLE_ACKS` environment variable.
21
-
22
-
## Example
23
-
24
-
defmodule MyTapConsumer do
25
-
@behaviour Drinkup.Tap.Consumer
26
-
27
-
def handle_event(%Drinkup.Tap.Event.Record{action: :create} = record) do
28
-
# Handle new record creation
29
-
case save_to_database(record) do
30
-
:ok -> :ok # Success - event will be acked
31
-
{:error, reason} -> {:error, reason} # Failure - Tap will retry
32
-
end
33
-
end
34
-
35
-
def handle_event(%Drinkup.Tap.Event.Identity{} = identity) do
36
-
# Handle identity changes
37
-
update_identity(identity)
38
-
:ok # Success - event will be acked
39
-
end
40
-
end
5
+
Implemented by `Drinkup.Tap`, you'll likely want to be using that instead.
41
6
"""
42
7
43
8
alias Drinkup.Tap.Event