very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
rust
fjall
at-protocol
atproto
indexer
1#!/usr/bin/env nu
2use common.nu *
3
4def run-auth-test [did: string, password: string, pds_url: string, relays: string, port: int] {
5 let url = $"http://localhost:($port)"
6 let ws_url = $"ws://localhost:($port)/stream"
7 let db_path = (mktemp -d -t hydrant_auth_test.XXXXXX)
8
9 # 1. authenticate
10 print $"authenticating with ($pds_url)..."
11 let session = authenticate $pds_url $did $password
12 let jwt = $session.accessJwt
13 print "authentication successful"
14
15 # 2. start hydrant
16 print $"starting hydrant on port ($port) with relays: ($relays)..."
17 let binary = "target/debug/hydrant" # already built in main
18 let instance = (with-env { HYDRANT_RELAY_HOSTS: $relays } {
19 start-hydrant $binary $db_path $port
20 })
21
22 mut test_passed = false
23
24 if (wait-for-api $url) {
25 # 3. start listener (live stream)
26 let output_file = $"($db_path)/stream_output.txt"
27 print $"starting stream listener -> ($output_file)"
28 # use websocat to capture output.
29 let stream_pid = (bash -c $"websocat '($ws_url)' > '($output_file)' & echo $!" | str trim | into int)
30 print $"listener pid: ($stream_pid)"
31
32 # 4. add repo to hydrant (backfill trigger)
33 print $"adding repo ($did) to tracking..."
34 try {
35 http put -t application/json $"($url)/repos" [ { did: ($did) } ]
36 } catch {
37 print "warning: failed to add repo (might already be tracked), continuing..."
38 }
39 wait-for-backfill $url
40
41 # 5. perform actions
42 let collection = "app.bsky.feed.post"
43 let timestamp = (date now | format date "%Y-%m-%dT%H:%M:%SZ")
44 let record_data = {
45 "$type": "app.bsky.feed.post",
46 text: $"hydrant integration test ($timestamp)",
47 createdAt: $timestamp
48 }
49
50 print "--- action: create ---"
51 let create_res = create-record $pds_url $jwt $did $collection $record_data
52 print $"created uri: ($create_res.uri)"
53 print $"created cid: ($create_res.cid)"
54 let rkey = ($create_res.uri | split row "/" | last)
55
56 print "--- action: update ---"
57 let update_data = ($record_data | update text $"updated text ($timestamp)")
58
59 try {
60 http post -t application/json -H ["Authorization" $"Bearer ($jwt)"] $"($pds_url)/xrpc/com.atproto.repo.putRecord" {
61 repo: $did,
62 collection: $collection,
63 rkey: $rkey,
64 record: $update_data,
65 }
66 print "updated record"
67 } catch { |err|
68 print $"update failed: ($err)"
69 # try to continue to delete
70 }
71
72 print "--- action: delete ---"
73 delete-record $pds_url $jwt $did $collection $rkey
74 print "deleted record"
75
76 print "--- action: deactivate ---"
77 deactivate-account $pds_url $jwt
78
79 sleep 1sec
80
81 # we might need to re-auth if session was killed by deactivation
82 print "re-authenticating..."
83 let session = authenticate $pds_url $did $password
84 let jwt = $session.accessJwt
85
86 sleep 1sec
87
88 print "--- action: activate ---"
89 activate-account $pds_url $jwt
90
91 # 6. verify
92 sleep 3sec
93
94 print "stopping listener..."
95 try { kill -9 $stream_pid }
96
97 if ($output_file | path exists) {
98 let content = (open $output_file | str trim)
99 if ($content | is-empty) {
100 print "failed: no events captured"
101 } else {
102 # parse json lines
103 let events = ($content | lines | each { |it| $it | from json })
104 let display_events = ($events | each { |e|
105 let value = if $e.type == "record" { $e | get -o record } else if $e.type == "account" { $e | get -o account } else { $e | get -o identity }
106 $e | select id type | insert value $value
107 })
108 print $"captured ($events | length) events"
109 $display_events | to text | print
110
111 # filter live events for the relevant entities
112 let relevant_events = ($events | where { |it|
113 if $it.type == "record" {
114 if ($it.record | get -o live) == false {
115 return false
116 }
117 }
118 true
119 })
120
121 let checks = [
122 { |e| $e.type == "account" and $e.account.active == true },
123 { |e| $e.type == "record" and $e.record.action == "create" },
124 { |e| $e.type == "record" and $e.record.action == "update" },
125 { |e| $e.type == "record" and $e.record.action == "delete" },
126 { |e| $e.type == "account" and $e.account.active == false },
127 { |e| $e.type == "account" and $e.account.active == true }
128 ]
129
130 if ($relevant_events | length) != ($checks | length) {
131 print $"verification failed: expected ($checks | length) events, got ($relevant_events | length)"
132 $test_passed = false
133 } else {
134
135 mut failed = false
136 for i in 0..(($relevant_events | length) - 1) {
137 let event = ($relevant_events | get $i)
138 let check = ($checks | get $i)
139 if not (do $check $event) {
140 print $"verification failed at event #($i + 1)"
141 print $"event: ($event)"
142 $failed = true
143 break
144 }
145 }
146
147 if not $failed {
148 print "test success!"
149 $test_passed = true
150 } else {
151 $test_passed = false
152 }
153 }
154 }
155 } else {
156 print "failed: output file missing"
157 }
158
159 } else {
160 print "hydrant failed to start"
161 }
162
163 # cleanup
164 print "cleaning up..."
165 try { kill -9 $instance.pid }
166
167 $test_passed
168}
169
170def main [] {
171 let env_vars = load-env-file
172 let did = ($env_vars | get --optional TEST_REPO)
173 let password = ($env_vars | get --optional TEST_PASSWORD)
174
175 if ($did | is-empty) or ($password | is-empty) {
176 print "SKIP: TEST_REPO and TEST_PASSWORD not set in .env"
177 exit 0
178 }
179
180 let pds_url = resolve-pds $did
181
182 # ensure build
183 build-hydrant | ignore
184
185 let port = resolve-test-port 3005
186
187 print "=== running single-relay test ==="
188 let relay1 = "wss://relay.fire.hose.cam"
189 let success1 = run-auth-test $did $password $pds_url $relay1 $port
190
191 print ""
192 print "=== running multi-relay test ==="
193 let relay_multi = "wss://relay.fire.hose.cam,wss://relay3.fr.hose.cam,wss://relay1.us-west.bsky.network,wss://relay1.us-east.bsky.network"
194 let success2 = run-auth-test $did $password $pds_url $relay_multi $port
195
196 if $success1 and $success2 {
197 print ""
198 print "ALL AUTHENTICATED STREAM TESTS PASSED"
199 exit 0
200 } else {
201 print ""
202 print $"TESTS FAILED: single=($success1), multi=($success2)"
203 exit 1
204 }
205}