atmoq

Documentation

Overview

Package atmoq is a Go client for the atproto firehose carried over MoQ transport (https://github.com/streamplace/atmoq).

It speaks kixelated's moq-lite protocol (draft 03/04) directly over raw QUIC — no WebTransport/HTTP3 layer — which is what `atmoq serve` and public moq-lite relays accept alongside h3. A consumer subscribes to a broadcast's track and receives a stream of "groups", each carrying length-prefixed frames. For atmoq, every frame is one at-sync firehose message: the exact same bytes (a DRISL header object followed by a DRISL payload object) that com.atproto.sync.subscribeRepos delivers over WebSocket. So a frame can be fed straight into indigo's existing event decoder. See ValidateDrisl for atmoq's stack-wide DRISL strictness.

This package implements the consumer (subscribe) path only.

Index

• Constants
• func ValidateDrisl(data []byte, offset int) (int, error)
• func ValidateDrislExact(data []byte) error
• type DrislError
  • func (e *DrislError) Error() string
• type Options
• type Session
  • func Dial(ctx context.Context, rawURL string, opts *Options) (*Session, error)
  • func (s *Session) Close() error
  • func (s *Session) Subscribe(ctx context.Context, broadcast, track string) (*Subscription, error)
  • func (s *Session) SubscribeFrom(ctx context.Context, broadcast, track string, startGroup uint64) (*Subscription, error)
  • func (s *Session) Version() string
• type Subscription
  • func (sub *Subscription) Close() error
  • func (sub *Subscription) ReadFrame(ctx context.Context) (data []byte, group uint64, err error)

Constants

const (
	DefaultBroadcast = "atproto"
	DefaultTrack     = "atproto"
)

Defaults matching `atmoq serve`: a single broadcast named "atproto" carrying a single track also named "atproto".

Functions

func ValidateDrisl

func ValidateDrisl(data []byte, offset int) (int, error)

ValidateDrisl validates one complete DRISL item starting at offset and returns the offset just past the item.

func ValidateDrislExact

func ValidateDrislExact(data []byte) error

ValidateDrislExact validates that data is exactly one complete DRISL item — no trailing bytes.

Types

type DrislError

type DrislError struct {
	Offset  int
	Message string
}

DrislError is a DRISL violation, with the byte offset where it was found.

func (*DrislError) Error

func (e *DrislError) Error() string

type Options

type Options struct {
	// TLSConfig, if set, is used as-is except that the moq-lite ALPNs and the
	// ServerName (from the dial URL host) are filled in when empty.
	TLSConfig *tls.Config
	// Insecure disables TLS certificate verification. Useful for self-signed
	// dev servers; ignored when TLSConfig is provided.
	Insecure bool
	// Log receives debug/info logs. Defaults to slog.Default().
	Log *slog.Logger
}

Options configures a Dial.

type Session

type Session struct {
	// contains filtered or unexported fields
}

Session is a live moq-lite connection to a relay or server.

func Dial

func Dial(ctx context.Context, rawURL string, opts *Options) (*Session, error)

Dial establishes a moq-lite session over QUIC. The URL host (and optional port, defaulting to 443) identifies the relay; the canonical scheme is "moqt" (the MoQ Transport URI scheme), with "moql"/"moq"/"moqs" accepted as aliases, or the scheme may be omitted. The scheme is only a client-side hint to use raw-QUIC MoQ — the server negotiates the actual protocol via ALPN. Any path is ignored: broadcasts are addressed by name in Subscribe, not by URL path.

func (*Session) Close

func (s *Session) Close() error

Close tears down the session. All subscriptions are failed first, so readers blocked in ReadFrame (and internal goroutines blocked delivering to a full buffer) unblock immediately rather than waiting on transport timeouts.

func (*Session) Subscribe

func (s *Session) Subscribe(ctx context.Context, broadcast, track string) (*Subscription, error)

Subscribe requests every future frame of the given track within the given broadcast and returns a Subscription to read them from. The subscription starts at the publisher's latest group (the live edge), matching the default `goat firehose` tail behavior.

func (*Session) SubscribeFrom

func (s *Session) SubscribeFrom(ctx context.Context, broadcast, track string, startGroup uint64) (*Subscription, error)

SubscribeFrom is like Subscribe but requests replay starting at the given MoQ group sequence rather than the live edge. The relay serves from startGroup if it is still within the relay's retention window, otherwise from the oldest group it still retains that is >= startGroup (a forward jump). The caller detects that jump — and the resulting gap — by comparing the first delivered group/at-seq against what it expected; deeper recovery is out of scope for the transport (re-sync from the PDS).

The group sequence to pass is one previously observed from ReadFrame: a consumer resumes by remembering the last group it fully processed.

func (*Session) Version

func (s *Session) Version() string

Version returns the negotiated moq-lite ALPN (e.g. "moq-lite-04").

type Subscription

type Subscription struct {
	// contains filtered or unexported fields
}

Subscription is a stream of frames for one subscribed track.

func (*Subscription) Close

func (sub *Subscription) Close() error

Close ends the subscription and releases its resources.

func (*Subscription) ReadFrame

func (sub *Subscription) ReadFrame(ctx context.Context) (data []byte, group uint64, err error)

ReadFrame returns the next frame's raw bytes and the sequence number of the group it belongs to. For an atmoq firehose, the bytes are a complete at-sync message (CBOR header object + CBOR payload object), identical to a subscribeRepos WebSocket message. It blocks until a frame is available, the context is cancelled, or the subscription ends.