› async-matrix › Source › Async › Matrix › Schema › Registry

class Registry

Loads and indexes Matrix event schemas from the official matrix-org/matrix-spec YAML files bundled in data/matrix-spec/event-schemas/schema/.

Schemas are loaded lazily on first access and cached for the lifetime of the process. Each schema is a JSONSchemer::Schema instance that resolves relative $ref paths (e.g. core-event-schema/room_event.yaml) via a file-based resolver.

registry = Async::Matrix::Schema::Registry.instance registry["m.room.message"] # => JSONSchemer::Schema registry.variant("m.room.message", "m.text") # => JSONSchemer::Schema registry.event_types # => ["m.accepted_terms", ...]

Nested

Definitions

MATRIX_FORMATS

Custom format validators for Matrix-specific format hints.

Implementation

 
MATRIX_FORMATS = {
	"mx-user-id"        => ->(value, _schema) { value.is_a?(String) && value.match?(/\A@[^:]+:.+\z/) },
	"mx-room-id"        => ->(value, _schema) { value.is_a?(String) && value.match?(/\A![^:]+:.+\z/) },
	"mx-event-id"       => ->(value, _schema) { value.is_a?(String) && value.match?(/\A\$/) },
	"mx-room-alias"     => ->(value, _schema) { value.is_a?(String) && value.match?(/\A#[^:]+:.+\z/) },
	"mx-server-name"    => ->(value, _schema) { value.is_a?(String) && !value.empty? },
	"mx-unpadded-base64" => ->(value, _schema) { value.is_a?(String) && value.match?(/\A[A-Za-z0-9+\/]*\z/) },
}.freeze

def [](event_type)

Look up a schema by Matrix event type (e.g. "m.room.message"). Returns a JSONSchemer::Schema or nil if no schema exists for that type.

Implementation

 
def [](event_type)
	load_schemas! unless @loaded
	@schemas[event_type]
end

def variant(event_type, subtype)

Look up a variant schema (e.g. variant("m.room.message", "m.text")). Returns a JSONSchemer::Schema or nil.

Implementation

 
def variant(event_type, subtype)
	load_schemas! unless @loaded
	@variants[[event_type, subtype]]
end

def event_types

All known base event types (excludes variant subtypes).

Implementation

 
def event_types
	load_schemas! unless @loaded
	@schemas.keys.sort
end

def variant_types

All known variant keys as [event_type, subtype] pairs.

Implementation

 
def variant_types
	load_schemas! unless @loaded
	@variants.keys.sort
end

def size

Total number of schemas loaded (base + variants).

Implementation

 
def size
	load_schemas! unless @loaded
	@schemas.size + @variants.size
end

def validate(event_hash)

Validate an event hash against its schema. Returns an array of error hashes (empty if valid). If no schema exists for the event type, returns empty (lenient).

Implementation

 
def validate(event_hash)
	event_type = event_hash["type"]
	return [] unless event_type

	base = self[event_type]
	return [] unless base

	errors = base.validate(event_hash).to_a

	# Also validate against variant if applicable
	variant_key = detect_variant_key(event_type, event_hash)
	if variant_key
		variant_schema = variant(event_type, variant_key)
		if variant_schema
			errors.concat(variant_schema.validate(event_hash).to_a)
		end
	end

	errors
end

def valid?(event_hash)

Boolean validation.

Implementation

 
def valid?(event_hash)
	validate(event_hash).empty?
end

def content_properties(event_type)

Content properties defined by the schema for a given event type. Returns an array of property name strings, or empty array if unknown.

Implementation

 
def content_properties(event_type)
	schema = self[event_type]
	return [] unless schema

	extract_content_properties(schema.value)
end

def ref_resolver

Ref resolver that loads YAML/JSON files from disk. json_schemer calls this with a URI for each $ref it encounters.

Implementation

 
def ref_resolver
	@ref_resolver ||= proc do |uri|
		path = uri.path
		if path && File.exist?(path)
			content = File.read(path)
			if path.end_with?(".yaml", ".yml")
				YAML.safe_load(content, permitted_classes: [Symbol])
			else
				JSON.parse(content)
			end
		else
			raise JSONSchemer::UnknownRef, uri.to_s
		end
	end
end

def build_schemer(parsed, file_path)

Build a JSONSchemer::Schema from a parsed YAML hash, anchored at the given file path so relative $ref paths resolve correctly.

Implementation

 
def build_schemer(parsed, file_path)
	uri = URI("file://#{File.expand_path(file_path)}")
	JSONSchemer.schema(
		parsed,
		base_uri: uri,
		ref_resolver: ref_resolver,
		insert_property_defaults: false,
		formats: MATRIX_FORMATS
	)
end

def load_schemas!

Scan the schema directory and load all event schemas.

Implementation

 
def load_schemas!
	return if @loaded

	Dir.glob(File.join(SCHEMA_DIR, "*.yaml")).each do |path|
		load_schema_file(path)
	end

	# Also load JSON schema files (a few older schemas are JSON)
	Dir.glob(File.join(SCHEMA_DIR, "*.json")).each do |path|
		load_schema_file(path)
	end

	@loaded = true
end

def detect_variant_key(event_type, event_hash)

Detect the variant subtype key from an event hash. For m.room.message, the variant is the msgtype. For m.room.encrypted, the variant is the algorithm. For m.key.verification.start, the variant is the method.

Implementation

 
def detect_variant_key(event_type, event_hash)
	content = event_hash["content"]
	return nil unless content.is_a?(Hash)

	case event_type
	when "m.room.message"
		content["msgtype"]
	when "m.room.encrypted"
		content["algorithm"]
	when "m.key.verification.start"
		content["method"]
	end
end

def extract_content_properties(schema_hash)

Extract content property names from a schema's parsed YAML.

Implementation

 
def extract_content_properties(schema_hash)
	props = schema_hash.dig("properties", "content", "properties")
	return [] unless props.is_a?(Hash)
	props.keys
end