Something went wrong. Try again.
An ATProto Lexicon validator for Gleam.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251// Example: Validating xyz.statusphere.status records from Jetstream using honk//// This example connects to Bluesky's Jetstream firehose, filters for// xyz.statusphere.status records, and validates them in real-time using honk.
import gleam/dynamic/decodeimport gleam/ioimport gleam/jsonimport gleam/optionimport gleam/stringimport gooseimport honkimport honk/errors.{DataValidation, InvalidSchema, LexiconNotFound}
pub fn main() { io.println("🦢 Honk + Goose: Jetstream Validation Example") io.println("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━") io.println("") io.println("Connecting to Jetstream...") io.println("Filtering for: xyz.statusphere.status") io.println("Validating records with honk...") io.println("")
// Define the xyz.statusphere.status lexicon let lexicon = create_statusphere_lexicon()
// Configure goose to connect to Jetstream let config = goose.JetstreamConfig( endpoint: "wss://jetstream2.us-west.bsky.network/subscribe", wanted_collections: ["xyz.statusphere.status"], wanted_dids: [], cursor: option.None, max_message_size_bytes: option.None, compress: True, require_hello: False, )
// Start consuming events (this blocks forever) goose.start_consumer(config, handle_event(_, lexicon))}
/// Handles each Jetstream eventfn handle_event(json_event: String, lexicon: json.Json) -> Nil { let event = goose.parse_event(json_event)
case event { // Handle commit events (create/update/delete) goose.CommitEvent(did, time_us, commit) -> { case commit.operation { "create" -> handle_create(did, time_us, commit, lexicon) "update" -> handle_update(did, time_us, commit, lexicon) "delete" -> handle_delete(did, time_us, commit) _ -> Nil } }
// Ignore identity and account events for this example goose.IdentityEvent(_, _, _) -> Nil goose.AccountEvent(_, _, _) -> Nil goose.UnknownEvent(raw) -> { io.println("⚠️ Unknown event: " <> raw) } }}
/// Handles create operations - validates the new recordfn handle_create( did: String, _time_us: Int, commit: goose.CommitData, lexicon: json.Json,) -> Nil { case commit.record { option.Some(record_dynamic) -> { // Convert Dynamic to JSON for honk validation let record_json = dynamic_to_json(record_dynamic)
// Validate the record using honk case honk.validate_record([lexicon], "xyz.statusphere.status", record_json) { Ok(_) -> { // Extract status emoji for display let status_emoji = extract_status(record_dynamic) io.println( "✓ VALID | " <> truncate_did(did) <> " | " <> status_emoji <> " | " <> commit.rkey, ) } Error(err) -> { io.println( "✗ INVALID | " <> truncate_did(did) <> " | " <> format_error(err) <> " | " <> commit.rkey, ) } } } option.None -> { io.println("⚠️ CREATE event without record data") } }}
/// Handles update operations - validates the updated recordfn handle_update( did: String, _time_us: Int, commit: goose.CommitData, lexicon: json.Json,) -> Nil { case commit.record { option.Some(record_dynamic) -> { let record_json = dynamic_to_json(record_dynamic)
case honk.validate_record([lexicon], "xyz.statusphere.status", record_json) { Ok(_) -> { let status_emoji = extract_status(record_dynamic) io.println( "✓ UPDATED | " <> truncate_did(did) <> " | " <> status_emoji <> " | " <> commit.rkey, ) } Error(err) -> { io.println( "✗ INVALID | " <> truncate_did(did) <> " | " <> format_error(err) <> " | " <> commit.rkey, ) } } } option.None -> { io.println("⚠️ UPDATE event without record data") } }}
/// Handles delete operations - no validation neededfn handle_delete(did: String, _time_us: Int, commit: goose.CommitData) -> Nil { io.println("🗑️ DELETED | " <> truncate_did(did) <> " | " <> commit.rkey)}
/// Creates the xyz.statusphere.status lexicon definitionfn create_statusphere_lexicon() -> json.Json { json.object([ #("lexicon", json.int(1)), #("id", json.string("xyz.statusphere.status")), #( "defs", json.object([ #( "main", json.object([ #("type", json.string("record")), #("key", json.string("tid")), #( "record", json.object([ #("type", json.string("object")), #( "required", json.preprocessed_array([ json.string("status"), json.string("createdAt"), ]), ), #( "properties", json.object([ #( "status", json.object([ #("type", json.string("string")), #("minLength", json.int(1)), #("maxGraphemes", json.int(1)), #("maxLength", json.int(32)), ]), ), #( "createdAt", json.object([ #("type", json.string("string")), #("format", json.string("datetime")), ]), ), ]), ), ]), ), ]), ), ]), ), ])}
/// Converts Dynamic to Json (they're the same underlying type)@external(erlang, "gleam@dynamic", "unsafe_coerce")fn dynamic_to_json(value: decode.Dynamic) -> json.Json
/// Extracts the status emoji from a record for displayfn extract_status(record: decode.Dynamic) -> String { let decoder = { use status <- decode.field("status", decode.string) decode.success(status) } case decode.run(record, decoder) { Ok(status) -> status Error(_) -> "�" }}
/// Formats a validation error for displayfn format_error(err: honk.ValidationError) -> String { case err { InvalidSchema(msg) -> "Schema: " <> msg DataValidation(msg) -> "Data: " <> msg LexiconNotFound(id) -> "Not found: " <> id }}
/// Truncates a DID for cleaner displayfn truncate_did(did: String) -> String { case string.split(did, ":") { [_, _, suffix] -> case string.length(suffix) > 12 { True -> string.slice(suffix, 0, 12) <> "..." False -> suffix } _ -> did }}