Something went wrong. Try again.
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
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288# print a failure message, kill any running hydrant pids, and exit with code 1.export def fail [msg: string, ...pids: int] { print $" FAILED: ($msg)" for pid in $pids { try { kill $pid } } exit 1}
# pipe a full http response (obtained via -f -e flags) through this to assert an expected# status code. on mismatch, prints the status, response body, kills any supplied pids, and exits.# on success, returns the response record so callers can inspect fields further.export def assert-status [expected: int, label: string, ...pids: int] { let resp = $in if $resp.status != $expected { print $" ($label): expected status ($expected), got ($resp.status)" try { print $" body: ($resp.body)" } for pid in $pids { try { kill $pid } } exit 1 } $resp}
# resolve the api port this test instance should use.export def resolve-test-port [default: int] { $env | get --optional HYDRANT_API_PORT | default ($default | into string) | into int}
# resolve the debug port this test instance should use.export def resolve-test-debug-port [default: int] { $env | get --optional HYDRANT_DEBUG_PORT | default ($default | into string) | into int}
# resolve the mock relay port for tests that need one.export def resolve-test-mock-port [default: int] { $env | get --optional HYDRANT_TEST_MOCK_PORT | default ($default | into string) | into int}
export def resolve-binary [default: string] { $env | get --optional HYDRANT_BINARY | default $default}
export def load-env-file [] { if (".env" | path exists) { let content = (open .env) $content | lines | where { |x| ($x | str trim | is-empty) == false and ($x | str trim | str starts-with "#") == false } | each { |x| let parts = ($x | split row "=" -n 2) { key: ($parts.0 | str trim), value: ($parts.1 | str trim | str trim -c '"' | str trim -c "'") } } | reduce -f {} { |it, acc| $acc | insert $it.key $it.value } } else { {} }}
# plc directories to use, best first. the mirror is unthrottled but has gone down for days,# so a dead mirror falls back to plc.directory instead of failing every test that resolves.# set HYDRANT_TEST_PLC_URL (comma-separated) to pick your own.export def plc-urls [] { let candidates = ( $env | get --optional HYDRANT_TEST_PLC_URL | default "https://plc.klbr.net,https://plc.directory" | split row "," | each { str trim | str trim -r -c "/" } | where { is-not-empty } ) for c in ($candidates | enumerate) { if (try { http get --max-time 5sec $"($c.item)/_health"; true } catch { false }) { return ($candidates | skip $c.index) } } # nothing answers, so keep them all and let the real request say why $candidates}
export def resolve-pds [did: string] { mut errors = [] for plc in (plc-urls) { let attempt = try { let doc = (http get --raw $"($plc)/($did)" | from json) {pds: ($doc.service | where type == "AtprotoPersonalDataServer" | first).serviceEndpoint} } catch { |e| {error: $"($plc): ($e.msg)"} } if "pds" in $attempt { return $attempt.pds } $errors = ($errors | append $attempt.error) } error make {msg: $"could not resolve ($did):\n($errors | str join "\n")"}}
export def authenticate [pds_url: string, identifier: string, password: string] { http post -t application/json $"($pds_url)/xrpc/com.atproto.server.createSession" { identifier: $identifier, password: $password }}
export def create-record [pds_url: string, jwt: string, repo: string, collection: string, record: any] { http post -t application/json -H ["Authorization" $"Bearer ($jwt)"] $"($pds_url)/xrpc/com.atproto.repo.createRecord" { repo: $repo, collection: $collection, record: $record }}
export def delete-record [pds_url: string, jwt: string, repo: string, collection: string, rkey: string] { http post -t application/json -H ["Authorization" $"Bearer ($jwt)"] $"($pds_url)/xrpc/com.atproto.repo.deleteRecord" { repo: $repo, collection: $collection, rkey: $rkey }}
export def deactivate-account [pds_url: string, jwt: string] { http post -t application/json -H ["Authorization" $"Bearer ($jwt)"] $"($pds_url)/xrpc/com.atproto.server.deactivateAccount" {}}
export def activate-account [pds_url: string, jwt: string] { mut last_message = "" for attempt in 1..5 { let result = (curl --fail-with-body --silent --show-error --max-time 20 -X POST -H "Content-Type: application/json" -H $"Authorization: Bearer ($jwt)" $"($pds_url)/xrpc/com.atproto.server.activateAccount" | complete) if $result.exit_code == 0 { return $result.stdout }
$last_message = $"($result.stdout)\n($result.stderr)" | str trim if $attempt < 5 { sleep 1sec } }
error make {msg: $"activateAccount failed: ($last_message)"}}
# extract the hydrant executable path from cargo's json build outputdef parse-hydrant-executable [output: string] { let items = ($output | lines | each { |line| try { $line | from json } catch { null } } | compact | where { |r| $r.reason? == "compiler-artifact" and $r.executable? != null and ($r.target?.name? == "hydrant") })
if ($items | is-empty) { error make {msg: "cargo succeeded without reporting a hydrant executable"} }
$items | last | get executable}
def require-build-success [out: record] { if $out.exit_code != 0 { let diagnostics = ([$out.stdout $out.stderr] | str join "\n" | str trim) error make {msg: $"cargo build failed:\n($diagnostics)"} }
parse-hydrant-executable $out.stdout}
# build the hydrant binaryexport def build-hydrant [] { if ($env | get --optional HYDRANT_BINARY | is-not-empty) { return $env.HYDRANT_BINARY } print "building hydrant..." let out = (^cargo build --message-format json | complete) require-build-success $out}
# build the hydrant binary with extra cargo features (space-separated string)export def build-hydrant-features [features: string] { if ($env | get --optional HYDRANT_BINARY | is-not-empty) { return $env.HYDRANT_BINARY } print $"building hydrant with features: ($features)..." let out = (^cargo build --features $features --message-format json | complete) require-build-success $out}
# build the hydrant binary for relay only (no default features)export def build-hydrant-relay [] { if ($env | get --optional HYDRANT_BINARY | is-not-empty) { return $env.HYDRANT_BINARY } print "building hydrant for relay only..." let out = (^cargo build --no-default-features --features relay --message-format json | complete) require-build-success $out}
# build the hydrant binary for relay with jetstream support (no default features)export def build-hydrant-relay-jetstream [] { if ($env | get --optional HYDRANT_BINARY | is-not-empty) { return $env.HYDRANT_BINARY } print "building hydrant for relay with jetstream..." let out = (^cargo build --no-default-features --features relay,jetstream --message-format json | complete) require-build-success $out}
# start a websocket client in the background that writes each message as a line of out_file,# and return its pid. stderr goes to out_file.err so it never reads as a message.# websocat's async stdio fails with EINVAL on darwin when stdin is /dev/null, which it is# for a background job, hence --no-async-stdio.export def start-ws-listener [url: string, out_file: string, --ping] { # with --ping the client gives up once a pong is 4s late, so a dead keepalive ends it let flags = if $ping { "--ping-interval 1 --ping-timeout 4" } else { "" } bash -c $"websocat --no-async-stdio -n ($flags) '($url)' > '($out_file)' 2> '($out_file).err' & echo $!" | str trim | into int}
# start hydrant in the backgroundexport def start-hydrant [binary: string, db_path: string, port: int] { let occupied = (^bash -c $"echo > /dev/tcp/127.0.0.1/($port)" | complete) if $occupied.exit_code == 0 { error make {msg: $"cannot start test hydrant: port ($port) is already in use"} }
let log_file = $"($db_path)/hydrant.log" print $"starting hydrant - logs at ($log_file)..."
let hydrant_vars = ($env | transpose k v | where k =~ "HYDRANT_" | reduce -f {} { |it, acc| $acc | upsert $it.k $it.v }) let env_vars = { HYDRANT_DATABASE_PATH: ($db_path), HYDRANT_FULL_NETWORK: "false", HYDRANT_API_BIND: $"127.0.0.1:($port)", # tests drive the management api over tcp, but it's unix-only by default HYDRANT_API_TCP_MANAGEMENT: "true", HYDRANT_ENABLE_DEBUG: "true", HYDRANT_DEBUG_PORT: (resolve-test-debug-port ($port + 1) | into string), HYDRANT_PLC_URL: (plc-urls | str join ","), RUST_LOG: "debug,hyper=error,tokio=error,h2=error,tower=error,rustls=error" } | merge $hydrant_vars
let pid = (with-env $env_vars { sh -c $"($binary) >($log_file) 2>&1 & echo $!" | str trim | into int })
print $"hydrant started with pid: ($pid)" { pid: $pid, log: $log_file }}
# wait for the api to become responsiveexport def wait-for-api [url: string] { print "waiting for api to be ready..." for i in 1..30 { try { http get $"($url)/stats" return true } catch { sleep 1sec } } false}
# poll stats until backfill is complete or failsexport def wait-for-backfill [url: string] { print "waiting for backfill to complete..." let attempts = 240 for i in 1..$attempts { let stats = (http get $"($url)/stats").counts let pending = ($stats.pending | into int) let records = ($stats.records | into int) let repos = ($stats.repos | into int) let resync = ($stats.resync | into int)
print $"[($i)/($attempts)] pending: ($pending), records: ($records), repos: ($repos), resync: ($resync)"
if $resync > 0 { print "resync state detected (failure or gone)!" print ($stats | table) return false }
if ($pending == 0) and ($repos > 0) and ($records > 0) { print "backfill complete." return true }
sleep 1sec } false}