Exosphere.ATProto.Firehose.Message (Exosphere v0.2.0)

Parse firehose message payloads into structured events.

Message types from com.atproto.sync.subscribeRepos:

#commit - Repository commit with record operations
#sync - Asserts the current repository state (commit + CAR blocks)
#identity - Identity update (DID document and/or handle change)
#account - Account hosting status change (active / takendown / suspended / deleted / deactivated)
#info - Informational message

The following event types are deprecated by the current sync spec and retained only for backwards compatibility with older relays:

#handle - Handle change (superseded by #identity)
#tombstone - Repository deletion (superseded by #account)

Summary

Types

account()

commit()

handle()

identity()

message()

operation()

sync()

Functions

decode(type, payload)

Decode a message payload based on its type.

extract_records(arg1)

Extract records from a commit's CAR blocks.

filter_by_collection(map, collection)

Filter operations by collection prefix.

has_collection?(map, collection)

Check if a commit contains operations for a specific collection.

Types

account()

@type account() :: %{
  type: :account,
  seq: integer(),
  did: String.t(),
  active: boolean(),
  status: String.t() | nil,
  time: String.t()
}

commit()

@type commit() :: %{
  type: :commit,
  seq: integer(),
  repo: String.t(),
  commit: Exosphere.ATProto.CID.t(),
  rev: String.t(),
  since: String.t() | nil,
  prev_data: Exosphere.ATProto.CID.t() | nil,
  ops: [operation()],
  blocks: binary(),
  time: String.t()
}

handle()

@type handle() :: %{
  type: :handle,
  seq: integer(),
  did: String.t(),
  handle: String.t(),
  time: String.t()
}

identity()

@type identity() :: %{
  type: :identity,
  seq: integer(),
  did: String.t(),
  handle: String.t() | nil,
  time: String.t()
}

message()

@type message() :: commit() | sync() | identity() | account() | handle() | map()

operation()

@type operation() :: %{
  action: :create | :update | :delete,
  path: String.t(),
  cid: Exosphere.ATProto.CID.t() | nil,
  prev: Exosphere.ATProto.CID.t() | nil
}

sync()

@type sync() :: %{
  type: :sync,
  seq: integer(),
  did: String.t(),
  rev: String.t() | nil,
  blocks: binary(),
  time: String.t()
}

Functions

decode(type, payload)

@spec decode(String.t(), map()) :: {:ok, message()}

Decode a message payload based on its type.

extract_records(arg1)

@spec extract_records(commit()) :: {:ok, [map()]} | {:error, term()}

Extract records from a commit's CAR blocks.

Parses the embedded CAR file to extract the actual record data. Returns a list of records with their collection, rkey, cid, and parsed record data.

filter_by_collection(map, collection)

@spec filter_by_collection(commit(), String.t()) :: [operation()]

Filter operations by collection prefix.

has_collection?(map, collection)

@spec has_collection?(commit(), String.t()) :: boolean()

Check if a commit contains operations for a specific collection.