Exosphere.ATProto.Firehose.Message (Exosphere v0.4.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()

info()

message()

operation()

sync()

tombstone()

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.

verify_commit(arg1)

Verify a #commit message's blocks against its signed MST root.

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()
}

info()

@type info() :: %{type: :info, name: String.t() | nil, message: String.t() | nil}

message()

@type message() ::
  commit()
  | sync()
  | identity()
  | account()
  | handle()
  | tombstone()
  | info()
  | 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()
}

tombstone()

@type tombstone() :: %{
  type: :tombstone,
  seq: integer(),
  did: String.t(),
  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.

verify_commit(arg1)

@spec verify_commit(commit()) ::
  {:ok, %{required(String.t()) => Exosphere.ATProto.CID.t()}} | {:error, term()}

Verify a #commit message's blocks against its signed MST root.

Decodes the embedded CAR, confirms the CAR's root is the commit the message points at, and checks the commit's record set against its signed data root via Exosphere.ATProto.Repo.Commit.verify_checkout/2. On success the full path => CID record set is returned.

Firehose commit CARs are incremental: they only carry the blocks new in that commit, so unchanged MST subtrees are referenced but not included. This function therefore succeeds only when every referenced node is present — typically an initial snapshot commit (no since). For steady-state verification, maintain the record set across commits and rebuild via MST.build/1 + Commit.verify_data/2, or verify a complete snapshot with Exosphere.ATProto.Repo.verify_checkout/3.

This checks structure only; authenticate the signer with Exosphere.ATProto.Repo.Commit.verify/3 and the repo's DID document.

Returns {:ok, records}, {:error, :not_a_commit} for other message types, {:error, :commit_not_in_blocks} when the root block is absent, {:error, :root_mismatch} when the CAR root disagrees with the message's commit link, or any error from CAR decoding / verify_checkout/2 (e.g. {:error, {:missing_block, cid}} for an incremental CAR).