Auto-indexing service and GraphQL API for AT Protocol Records
mv jetstream to a new pkg called goose
+21
-499
-24
jetstream/README.md
-24
jetstream/README.md
···
1
-
# jetstream
2
-
3
-
[](https://hex.pm/packages/jetstream)
4
-
[](https://hexdocs.pm/jetstream/)
5
-
6
-
```sh
7
-
gleam add jetstream@1
8
-
```
9
-
```gleam
10
-
import jetstream
11
-
12
-
pub fn main() -> Nil {
13
-
// TODO: An example of the project in use
14
-
}
15
-
```
16
-
17
-
Further documentation can be found at <https://hexdocs.pm/jetstream>.
18
-
19
-
## Development
20
-
21
-
```sh
22
-
gleam run # Run the project
23
-
gleam test # Run the tests
24
-
```
-23
jetstream/gleam.toml
-23
jetstream/gleam.toml
···
1
-
name = "jetstream"
2
-
version = "1.0.0"
3
-
4
-
# Fill out these fields if you intend to generate HTML documentation or publish
5
-
# your project to the Hex package manager.
6
-
#
7
-
# description = ""
8
-
# licences = ["Apache-2.0"]
9
-
# repository = { type = "github", user = "", repo = "" }
10
-
# links = [{ title = "Website", href = "" }]
11
-
#
12
-
# For a full reference of all the available options, you can have a look at
13
-
# https://gleam.run/writing-gleam/gleam-toml/.
14
-
15
-
[dependencies]
16
-
gleam_stdlib = ">= 0.44.0 and < 2.0.0"
17
-
gleam_erlang = ">= 1.0.0 and < 2.0.0"
18
-
gleam_http = ">= 4.0.0 and < 5.0.0"
19
-
gleam_json = ">= 3.0.2 and < 4.0.0"
20
-
gun = ">= 2.2.0 and < 3.0.0"
21
-
22
-
[dev-dependencies]
23
-
gleeunit = ">= 1.0.0 and < 2.0.0"
-20
jetstream/manifest.toml
-20
jetstream/manifest.toml
···
1
-
# This file was generated by Gleam
2
-
# You typically do not need to edit this file
3
-
4
-
packages = [
5
-
{ name = "cowlib", version = "2.16.0", build_tools = ["make", "rebar3"], requirements = [], otp_app = "cowlib", source = "hex", outer_checksum = "7F478D80D66B747344F0EA7708C187645CFCC08B11AA424632F78E25BF05DB51" },
6
-
{ name = "gleam_erlang", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_erlang", source = "hex", outer_checksum = "1124AD3AA21143E5AF0FC5CF3D9529F6DB8CA03E43A55711B60B6B7B3874375C" },
7
-
{ name = "gleam_http", version = "4.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_http", source = "hex", outer_checksum = "82EA6A717C842456188C190AFB372665EA56CE13D8559BF3B1DD9E40F619EE0C" },
8
-
{ name = "gleam_json", version = "3.0.2", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_json", source = "hex", outer_checksum = "874FA3C3BB6E22DD2BB111966BD40B3759E9094E05257899A7C08F5DE77EC049" },
9
-
{ name = "gleam_stdlib", version = "0.65.0", build_tools = ["gleam"], requirements = [], otp_app = "gleam_stdlib", source = "hex", outer_checksum = "7C69C71D8C493AE11A5184828A77110EB05A7786EBF8B25B36A72F879C3EE107" },
10
-
{ name = "gleeunit", version = "1.6.1", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "FDC68A8C492B1E9B429249062CD9BAC9B5538C6FBF584817205D0998C42E1DAC" },
11
-
{ name = "gun", version = "2.2.0", build_tools = ["make", "rebar3"], requirements = ["cowlib"], otp_app = "gun", source = "hex", outer_checksum = "76022700C64287FEB4DF93A1795CFF6741B83FB37415C40C34C38D2A4645261A" },
12
-
]
13
-
14
-
[requirements]
15
-
gleam_erlang = { version = ">= 1.0.0 and < 2.0.0" }
16
-
gleam_http = { version = ">= 4.0.0 and < 5.0.0" }
17
-
gleam_json = { version = ">= 3.0.2 and < 4.0.0" }
18
-
gleam_stdlib = { version = ">= 0.44.0 and < 2.0.0" }
19
-
gleeunit = { version = ">= 1.0.0 and < 2.0.0" }
20
-
gun = { version = ">= 2.2.0 and < 3.0.0" }
-233
jetstream/src/jetstream.gleam
-233
jetstream/src/jetstream.gleam
···
1
-
import gleam/dynamic.{type Dynamic}
2
-
import gleam/dynamic/decode
3
-
import gleam/erlang/process.{type Pid}
4
-
import gleam/io
5
-
import gleam/json
6
-
import gleam/list
7
-
import gleam/option.{type Option}
8
-
import gleam/string
9
-
10
-
/// Jetstream event types
11
-
pub type JetstreamEvent {
12
-
CommitEvent(did: String, time_us: Int, commit: CommitData)
13
-
IdentityEvent(did: String, time_us: Int, identity: IdentityData)
14
-
AccountEvent(did: String, time_us: Int, account: AccountData)
15
-
UnknownEvent(raw: String)
16
-
}
17
-
18
-
pub type CommitData {
19
-
CommitData(
20
-
rev: String,
21
-
operation: String,
22
-
collection: String,
23
-
rkey: String,
24
-
record: Option(Dynamic),
25
-
cid: Option(String),
26
-
)
27
-
}
28
-
29
-
pub type IdentityData {
30
-
IdentityData(did: String, handle: String, seq: Int, time: String)
31
-
}
32
-
33
-
pub type AccountData {
34
-
AccountData(active: Bool, did: String, seq: Int, time: String)
35
-
}
36
-
37
-
/// Configuration for Jetstream consumer
38
-
pub type JetstreamConfig {
39
-
JetstreamConfig(
40
-
endpoint: String,
41
-
wanted_collections: List(String),
42
-
wanted_dids: List(String),
43
-
)
44
-
}
45
-
46
-
/// Create a default configuration for US East endpoint
47
-
pub fn default_config() -> JetstreamConfig {
48
-
JetstreamConfig(
49
-
endpoint: "wss://jetstream2.us-east.bsky.network/subscribe",
50
-
wanted_collections: [],
51
-
wanted_dids: [],
52
-
)
53
-
}
54
-
55
-
/// Build the WebSocket URL with query parameters
56
-
pub fn build_url(config: JetstreamConfig) -> String {
57
-
let base = config.endpoint
58
-
let mut_params = []
59
-
60
-
// Add wanted collections (each as a separate query parameter)
61
-
let mut_params = case config.wanted_collections {
62
-
[] -> mut_params
63
-
collections -> {
64
-
let collection_params =
65
-
list.map(collections, fn(col) { "wantedCollections=" <> col })
66
-
list.append(collection_params, mut_params)
67
-
}
68
-
}
69
-
70
-
// Add wanted DIDs (each as a separate query parameter)
71
-
let mut_params = case config.wanted_dids {
72
-
[] -> mut_params
73
-
dids -> {
74
-
let did_params = list.map(dids, fn(did) { "wantedDids=" <> did })
75
-
list.append(did_params, mut_params)
76
-
}
77
-
}
78
-
79
-
case mut_params {
80
-
[] -> base
81
-
params -> base <> "?" <> string.join(list.reverse(params), "&")
82
-
}
83
-
}
84
-
85
-
/// Connect to Jetstream WebSocket using Erlang gun library
86
-
@external(erlang, "jetstream_ws_ffi", "connect")
87
-
pub fn connect(url: String, handler_pid: Pid) -> Result(Pid, Dynamic)
88
-
89
-
/// Start consuming the Jetstream feed
90
-
pub fn start_consumer(
91
-
config: JetstreamConfig,
92
-
on_event: fn(String) -> Nil,
93
-
) -> Nil {
94
-
let url = build_url(config)
95
-
io.println("🔗 Jetstream URL: " <> url)
96
-
let self = process.self()
97
-
let result = connect(url, self)
98
-
99
-
case result {
100
-
Ok(_conn_pid) -> {
101
-
receive_loop(on_event)
102
-
}
103
-
Error(err) -> {
104
-
io.println("Failed to connect to Jetstream")
105
-
io.println_error(string.inspect(err))
106
-
}
107
-
}
108
-
}
109
-
110
-
/// Receive loop for WebSocket messages
111
-
fn receive_loop(on_event: fn(String) -> Nil) -> Nil {
112
-
// Call Erlang to receive one message
113
-
case receive_ws_message() {
114
-
Ok(text) -> {
115
-
on_event(text)
116
-
receive_loop(on_event)
117
-
}
118
-
Error(_) -> {
119
-
// Timeout or error, continue loop
120
-
receive_loop(on_event)
121
-
}
122
-
}
123
-
}
124
-
125
-
/// Receive a WebSocket message from the message queue
126
-
@external(erlang, "jetstream_ffi", "receive_ws_message")
127
-
fn receive_ws_message() -> Result(String, Nil)
128
-
129
-
/// Parse a JSON event string into a JetstreamEvent
130
-
pub fn parse_event(json_string: String) -> JetstreamEvent {
131
-
// Try to parse as commit event first
132
-
case json.parse(json_string, commit_event_decoder()) {
133
-
Ok(event) -> event
134
-
Error(_) -> {
135
-
// Try identity event
136
-
case json.parse(json_string, identity_event_decoder()) {
137
-
Ok(event) -> event
138
-
Error(_) -> {
139
-
// Try account event
140
-
case json.parse(json_string, account_event_decoder()) {
141
-
Ok(event) -> event
142
-
Error(_) -> UnknownEvent(json_string)
143
-
}
144
-
}
145
-
}
146
-
}
147
-
}
148
-
}
149
-
150
-
/// Decoder for commit events
151
-
fn commit_event_decoder() {
152
-
use did <- decode.field("did", decode.string)
153
-
use time_us <- decode.field("time_us", decode.int)
154
-
use commit <- decode.field("commit", commit_data_decoder())
155
-
decode.success(CommitEvent(did: did, time_us: time_us, commit: commit))
156
-
}
157
-
158
-
/// Decoder for commit data - handles both create/update (with record) and delete (without)
159
-
fn commit_data_decoder() {
160
-
// Try decoder with record and cid fields first (for create/update)
161
-
// If that fails, try without (for delete)
162
-
decode.one_of(commit_with_record_decoder(), or: [
163
-
commit_without_record_decoder(),
164
-
])
165
-
}
166
-
167
-
/// Decoder for commit with record (create/update operations)
168
-
fn commit_with_record_decoder() {
169
-
use rev <- decode.field("rev", decode.string)
170
-
use operation <- decode.field("operation", decode.string)
171
-
use collection <- decode.field("collection", decode.string)
172
-
use rkey <- decode.field("rkey", decode.string)
173
-
use record <- decode.field("record", decode.dynamic)
174
-
use cid <- decode.field("cid", decode.string)
175
-
decode.success(CommitData(
176
-
rev: rev,
177
-
operation: operation,
178
-
collection: collection,
179
-
rkey: rkey,
180
-
record: option.Some(record),
181
-
cid: option.Some(cid),
182
-
))
183
-
}
184
-
185
-
/// Decoder for commit without record (delete operations)
186
-
fn commit_without_record_decoder() {
187
-
use rev <- decode.field("rev", decode.string)
188
-
use operation <- decode.field("operation", decode.string)
189
-
use collection <- decode.field("collection", decode.string)
190
-
use rkey <- decode.field("rkey", decode.string)
191
-
decode.success(CommitData(
192
-
rev: rev,
193
-
operation: operation,
194
-
collection: collection,
195
-
rkey: rkey,
196
-
record: option.None,
197
-
cid: option.None,
198
-
))
199
-
}
200
-
201
-
/// Decoder for identity events
202
-
fn identity_event_decoder() {
203
-
use did <- decode.field("did", decode.string)
204
-
use time_us <- decode.field("time_us", decode.int)
205
-
use identity <- decode.field("identity", identity_data_decoder())
206
-
decode.success(IdentityEvent(did: did, time_us: time_us, identity: identity))
207
-
}
208
-
209
-
/// Decoder for identity data
210
-
fn identity_data_decoder() {
211
-
use did <- decode.field("did", decode.string)
212
-
use handle <- decode.field("handle", decode.string)
213
-
use seq <- decode.field("seq", decode.int)
214
-
use time <- decode.field("time", decode.string)
215
-
decode.success(IdentityData(did: did, handle: handle, seq: seq, time: time))
216
-
}
217
-
218
-
/// Decoder for account events
219
-
fn account_event_decoder() {
220
-
use did <- decode.field("did", decode.string)
221
-
use time_us <- decode.field("time_us", decode.int)
222
-
use account <- decode.field("account", account_data_decoder())
223
-
decode.success(AccountEvent(did: did, time_us: time_us, account: account))
224
-
}
225
-
226
-
/// Decoder for account data
227
-
fn account_data_decoder() {
228
-
use active <- decode.field("active", decode.bool)
229
-
use did <- decode.field("did", decode.string)
230
-
use seq <- decode.field("seq", decode.int)
231
-
use time <- decode.field("time", decode.string)
232
-
decode.success(AccountData(active: active, did: did, seq: seq, time: time))
233
-
}
-25
jetstream/src/jetstream_ffi.erl
-25
jetstream/src/jetstream_ffi.erl
···
1
-
-module(jetstream_ffi).
2
-
-export([receive_ws_message/0]).
3
-
4
-
%% Receive a WebSocket text message from the process mailbox
5
-
receive_ws_message() ->
6
-
receive
7
-
%% Handle gun_ws messages directly (they're coming to us, not the handler)
8
-
{gun_ws, _ConnPid, _StreamRef, {text, Text}} ->
9
-
{ok, Text};
10
-
{gun_ws, _ConnPid, _StreamRef, {binary, _Binary}} ->
11
-
%% Ignore binary messages, try again
12
-
receive_ws_message();
13
-
{gun_ws, _ConnPid, _StreamRef, close} ->
14
-
{error, nil};
15
-
{gun_down, _ConnPid, _Protocol, _Reason, _KilledStreams} ->
16
-
{error, nil};
17
-
{gun_error, _ConnPid, _StreamRef, _Reason} ->
18
-
{error, nil};
19
-
_Other ->
20
-
%% Ignore unexpected messages
21
-
receive_ws_message()
22
-
after 60000 ->
23
-
%% Timeout - return error to continue loop
24
-
{error, nil}
25
-
end.
-142
jetstream/src/jetstream_ws_ffi.erl
-142
jetstream/src/jetstream_ws_ffi.erl
···
1
-
-module(jetstream_ws_ffi).
2
-
-export([connect/2]).
3
-
4
-
%% Connect to WebSocket using gun
5
-
connect(Url, HandlerPid) ->
6
-
%% Start gun application and dependencies
7
-
application:ensure_all_started(ssl),
8
-
application:ensure_all_started(gun),
9
-
10
-
%% Parse URL using uri_string
11
-
UriMap = uri_string:parse(Url),
12
-
#{scheme := SchemeStr, host := Host, path := Path} = UriMap,
13
-
14
-
%% Get query string if present and append to path
15
-
Query = maps:get(query, UriMap, undefined),
16
-
PathWithQuery = case Query of
17
-
undefined -> Path;
18
-
<<>> -> Path;
19
-
Q -> <<Path/binary, "?", Q/binary>>
20
-
end,
21
-
22
-
%% Get port, use defaults if not specified
23
-
Port = maps:get(port, uri_string:parse(Url),
24
-
case SchemeStr of
25
-
<<"wss">> -> 443;
26
-
<<"ws">> -> 80;
27
-
_ -> 443
28
-
end),
29
-
30
-
%% Determine transport
31
-
Transport = case SchemeStr of
32
-
<<"wss">> -> tls;
33
-
<<"ws">> -> tcp;
34
-
_ -> tls
35
-
end,
36
-
37
-
%% TLS options for secure connections
38
-
TlsOpts = [{verify, verify_none}], %% For simplicity, disable cert verification
39
-
%% In production, use proper CA certs
40
-
41
-
%% Connection options
42
-
Opts = case Transport of
43
-
tls ->
44
-
#{
45
-
transport => tls,
46
-
tls_opts => TlsOpts,
47
-
protocols => [http],
48
-
retry => 10,
49
-
retry_timeout => 1000
50
-
};
51
-
tcp ->
52
-
#{
53
-
transport => tcp,
54
-
protocols => [http],
55
-
retry => 10,
56
-
retry_timeout => 1000
57
-
}
58
-
end,
59
-
60
-
%% Convert host to list if needed
61
-
HostStr = case is_binary(Host) of
62
-
true -> binary_to_list(Host);
63
-
false -> Host
64
-
end,
65
-
66
-
%% Ensure path with query is binary
67
-
PathBin = case is_binary(PathWithQuery) of
68
-
true -> PathWithQuery;
69
-
false -> list_to_binary(PathWithQuery)
70
-
end,
71
-
72
-
%% Open connection
73
-
case gun:open(HostStr, Port, Opts) of
74
-
{ok, ConnPid} ->
75
-
%% Monitor the connection
76
-
MRef = monitor(process, ConnPid),
77
-
78
-
%% Wait for connection
79
-
receive
80
-
{gun_up, ConnPid, _Protocol} ->
81
-
%% Upgrade to WebSocket
82
-
StreamRef = gun:ws_upgrade(ConnPid, binary_to_list(PathBin), []),
83
-
84
-
%% Wait for upgrade
85
-
receive
86
-
{gun_upgrade, ConnPid, StreamRef, [<<"websocket">>], _Headers} ->
87
-
%% Spawn a handler process to listen for WebSocket frames
88
-
spawn(fun() -> handle_messages(ConnPid, StreamRef, HandlerPid) end),
89
-
%% Return immediately so Gleam can continue
90
-
{ok, ConnPid};
91
-
{gun_response, ConnPid, _, _, Status, Headers} ->
92
-
gun:close(ConnPid),
93
-
{error, {upgrade_failed, Status, Headers}};
94
-
{gun_error, ConnPid, _StreamRef, Reason} ->
95
-
gun:close(ConnPid),
96
-
{error, {gun_error, Reason}};
97
-
{'DOWN', MRef, process, ConnPid, Reason} ->
98
-
{error, {connection_down, Reason}};
99
-
_Other ->
100
-
gun:close(ConnPid),
101
-
{error, unexpected_message}
102
-
after 30000 ->
103
-
gun:close(ConnPid),
104
-
{error, upgrade_timeout}
105
-
end;
106
-
{'DOWN', MRef, process, ConnPid, Reason} ->
107
-
{error, {connection_failed, Reason}};
108
-
_Other ->
109
-
gun:close(ConnPid),
110
-
{error, unexpected_message}
111
-
after 30000 ->
112
-
gun:close(ConnPid),
113
-
{error, connection_timeout}
114
-
end;
115
-
{error, Reason} ->
116
-
{error, {open_failed, Reason}}
117
-
end.
118
-
119
-
%% Handle incoming WebSocket messages
120
-
handle_messages(ConnPid, StreamRef, HandlerPid) ->
121
-
receive
122
-
{gun_ws, ConnPid, StreamRef, {text, Text}} ->
123
-
HandlerPid ! {ws_text, Text},
124
-
handle_messages(ConnPid, StreamRef, HandlerPid);
125
-
{gun_ws, ConnPid, StreamRef, {binary, Binary}} ->
126
-
HandlerPid ! {ws_binary, Binary},
127
-
handle_messages(ConnPid, StreamRef, HandlerPid);
128
-
{gun_ws, ConnPid, StreamRef, close} ->
129
-
HandlerPid ! {ws_closed, normal},
130
-
gun:close(ConnPid);
131
-
{gun_down, ConnPid, _Protocol, Reason, _KilledStreams} ->
132
-
HandlerPid ! {ws_error, Reason},
133
-
gun:close(ConnPid);
134
-
{gun_error, ConnPid, StreamRef, Reason} ->
135
-
HandlerPid ! {ws_error, Reason},
136
-
handle_messages(ConnPid, StreamRef, HandlerPid);
137
-
stop ->
138
-
gun:close(ConnPid)
139
-
after 30000 ->
140
-
%% Heartbeat every 30 seconds to keep connection alive
141
-
handle_messages(ConnPid, StreamRef, HandlerPid)
142
-
end.
-13
jetstream/test/jetstream_test.gleam
-13
jetstream/test/jetstream_test.gleam
+1
-1
server/gleam.toml
+1
-1
server/gleam.toml
···
13
13
# https://gleam.run/writing-gleam/gleam-toml/.
14
14
15
15
[dependencies]
16
-
jetstream = { path = "../jetstream" }
17
16
lexicon = { path = "../lexicon" }
18
17
graphql = { path = "../graphql" }
19
18
lexicon_graphql = { path = "../lexicon_graphql" }
···
34
33
envoy = ">= 1.0.2 and < 2.0.0"
35
34
dotenv_gleam = ">= 2.0.1 and < 3.0.0"
36
35
thoas = ">= 1.0.0 and < 2.0.0"
36
+
goose = ">= 1.0.0 and < 2.0.0"
37
37
38
38
[dev-dependencies]
39
39
gleeunit = ">= 1.0.0 and < 2.0.0"
+3
-2
server/manifest.toml
+3
-2
server/manifest.toml
···
9
9
{ name = "envoy", version = "1.0.2", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "envoy", source = "hex", outer_checksum = "95FD059345AA982E89A0B6E2A3BF1CF43E17A7048DCD85B5B65D3B9E4E39D359" },
10
10
{ name = "esqlite", version = "0.9.0", build_tools = ["rebar3"], requirements = [], otp_app = "esqlite", source = "hex", outer_checksum = "CCF72258A4EE152EC7AD92AA9A03552EB6CA1B06B65C93AD5B6E55C302E05855" },
11
11
{ name = "exception", version = "2.1.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "exception", source = "hex", outer_checksum = "329D269D5C2A314F7364BD2711372B6F2C58FA6F39981572E5CA68624D291F8C" },
12
+
{ name = "ezstd", version = "1.2.3", build_tools = ["rebar3"], requirements = [], otp_app = "ezstd", source = "hex", outer_checksum = "DE32E0B41BA36A9ED46DB8215DA74777D2F141BB75F67BFC05DBB4B7C3386DEE" },
12
13
{ name = "filepath", version = "1.1.2", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "filepath", source = "hex", outer_checksum = "B06A9AF0BF10E51401D64B98E4B627F1D2E48C154967DA7AF4D0914780A6D40A" },
13
14
{ name = "gleam_crypto", version = "1.5.1", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_crypto", source = "hex", outer_checksum = "50774BAFFF1144E7872814C566C5D653D83A3EBF23ACC3156B757A1B6819086E" },
14
15
{ name = "gleam_erlang", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_erlang", source = "hex", outer_checksum = "1124AD3AA21143E5AF0FC5CF3D9529F6DB8CA03E43A55711B60B6B7B3874375C" },
···
21
22
{ name = "gleam_yielder", version = "1.1.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_yielder", source = "hex", outer_checksum = "8E4E4ECFA7982859F430C57F549200C7749823C106759F4A19A78AEA6687717A" },
22
23
{ name = "gleeunit", version = "1.6.1", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "FDC68A8C492B1E9B429249062CD9BAC9B5538C6FBF584817205D0998C42E1DAC" },
23
24
{ name = "glisten", version = "8.0.1", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_otp", "gleam_stdlib", "logging", "telemetry"], otp_app = "glisten", source = "hex", outer_checksum = "534BB27C71FB9E506345A767C0D76B17A9E9199934340C975DC003C710E3692D" },
25
+
{ name = "goose", version = "1.0.0", build_tools = ["gleam"], requirements = ["ezstd", "gleam_erlang", "gleam_http", "gleam_json", "gleam_stdlib", "gun"], otp_app = "goose", source = "hex", outer_checksum = "A983AB3E25C6E8F3CC8BBB63A649AFCAFF6A23AC232AF28612D9B31643F07053" },
24
26
{ name = "gramps", version = "6.0.0", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_erlang", "gleam_http", "gleam_stdlib"], otp_app = "gramps", source = "hex", outer_checksum = "8B7195978FBFD30B43DF791A8A272041B81E45D245314D7A41FC57237AA882A0" },
25
27
{ name = "graphql", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], source = "local", path = "../graphql" },
26
28
{ name = "gun", version = "2.2.0", build_tools = ["make", "rebar3"], requirements = ["cowlib"], otp_app = "gun", source = "hex", outer_checksum = "76022700C64287FEB4DF93A1795CFF6741B83FB37415C40C34C38D2A4645261A" },
27
29
{ name = "houdini", version = "1.2.0", build_tools = ["gleam"], requirements = [], otp_app = "houdini", source = "hex", outer_checksum = "5DB1053F1AF828049C2B206D4403C18970ABEF5C18671CA3C2D2ED0DD64F6385" },
28
30
{ name = "hpack_erl", version = "0.3.0", build_tools = ["rebar3"], requirements = [], otp_app = "hpack", source = "hex", outer_checksum = "D6137D7079169D8C485C6962DFE261AF5B9EF60FBC557344511C1E65E3D95FB0" },
29
-
{ name = "jetstream", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_http", "gleam_json", "gleam_stdlib", "gun"], source = "local", path = "../jetstream" },
30
31
{ name = "jose", version = "1.11.10", build_tools = ["mix", "rebar3"], requirements = [], otp_app = "jose", source = "hex", outer_checksum = "0D6CD36FF8BA174DB29148FC112B5842186B68A90CE9FC2B3EC3AFE76593E614" },
31
32
{ name = "lexicon", version = "0.1.0", build_tools = ["gleam"], requirements = ["gleam_json", "gleam_stdlib"], source = "local", path = "../lexicon" },
32
33
{ name = "lexicon_graphql", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_json", "gleam_stdlib", "graphql"], source = "local", path = "../lexicon_graphql" },
···
54
55
gleam_stdlib = { version = ">= 0.60.0 and < 1.0.0" }
55
56
gleam_time = { version = ">= 1.4.0 and < 2.0.0" }
56
57
gleeunit = { version = ">= 1.0.0 and < 2.0.0" }
58
+
goose = { version = ">= 1.0.0 and < 2.0.0" }
57
59
graphql = { path = "../graphql" }
58
-
jetstream = { path = "../jetstream" }
59
60
jose = { version = ">= 1.11.10 and < 2.0.0" }
60
61
lexicon = { path = "../lexicon" }
61
62
lexicon_graphql = { path = "../lexicon_graphql" }
+4
-4
server/src/event_handler.gleam
+4
-4
server/src/event_handler.gleam
···
5
5
import gleam/list
6
6
import gleam/option
7
7
import gleam/string
8
-
import jetstream
8
+
import goose
9
9
import lexicon
10
10
import sqlight
11
11
···
41
41
pub fn handle_commit_event(
42
42
db: sqlight.Connection,
43
43
did: String,
44
-
commit: jetstream.CommitData,
44
+
commit: goose.CommitData,
45
45
) -> Nil {
46
46
let uri = "at://" <> did <> "/" <> commit.collection <> "/" <> commit.rkey
47
47
···
150
150
/// Handle an identity event (update actor handle)
151
151
pub fn handle_identity_event(
152
152
db: sqlight.Connection,
153
-
identity: jetstream.IdentityData,
153
+
identity: goose.IdentityData,
154
154
) -> Nil {
155
155
case database.upsert_actor(db, identity.did, identity.handle) {
156
156
Ok(_) -> {
···
172
172
/// Handle an account event
173
173
pub fn handle_account_event(
174
174
_db: sqlight.Connection,
175
-
account: jetstream.AccountData,
175
+
account: goose.AccountData,
176
176
) -> Nil {
177
177
// For now, just log account events - we could extend this in the future
178
178
let status = case account.active {
+13
-8
server/src/jetstream_consumer.gleam
+13
-8
server/src/jetstream_consumer.gleam
···
5
5
import gleam/int
6
6
import gleam/io
7
7
import gleam/list
8
+
import gleam/option
8
9
import gleam/string
9
-
import jetstream
10
+
import goose
10
11
import sqlight
11
12
12
13
/// Start the Jetstream consumer in a background process
···
44
45
45
46
// Create Jetstream config
46
47
let config =
47
-
jetstream.JetstreamConfig(
48
+
goose.JetstreamConfig(
48
49
endpoint: jetstream_url,
49
50
wanted_collections: collection_ids,
50
51
wanted_dids: [],
52
+
cursor: option.None,
53
+
max_message_size_bytes: option.None,
54
+
compress: True,
55
+
require_hello: False,
51
56
)
52
57
53
58
io.println("")
···
59
64
// Start the Jetstream consumer in a separate process
60
65
// This will run independently and call our event handler callback
61
66
process.spawn_unlinked(fn() {
62
-
jetstream.start_consumer(config, fn(event_json) {
67
+
goose.start_consumer(config, fn(event_json) {
63
68
handle_jetstream_event(db, event_json)
64
69
})
65
70
})
···
79
84
80
85
/// Handle a raw Jetstream event JSON string
81
86
fn handle_jetstream_event(db: sqlight.Connection, event_json: String) -> Nil {
82
-
case jetstream.parse_event(event_json) {
83
-
jetstream.CommitEvent(did, _time_us, commit) -> {
87
+
case goose.parse_event(event_json) {
88
+
goose.CommitEvent(did, _time_us, commit) -> {
84
89
event_handler.handle_commit_event(db, did, commit)
85
90
}
86
-
jetstream.IdentityEvent(_did, _time_us, _identity) -> {
91
+
goose.IdentityEvent(_did, _time_us, _identity) -> {
87
92
// Silently ignore identity events
88
93
Nil
89
94
}
90
-
jetstream.AccountEvent(_did, _time_us, _account) -> {
95
+
goose.AccountEvent(_did, _time_us, _account) -> {
91
96
// Silently ignore account events
92
97
Nil
93
98
}
94
-
jetstream.UnknownEvent(_raw) -> {
99
+
goose.UnknownEvent(_raw) -> {
95
100
// Silently ignore unknown events
96
101
Nil
97
102
}