atarchive

Reusable Go packages for AT Protocol archiving, plus a reference application that archives the entire Bluesky network.

Packages

All packages live under pkg/ and are designed for independent use. Each accepts optional metrics interfaces (pass nil to disable) so they work standalone without Prometheus.

Package	Description	Dependencies
pkg/tid	AT Protocol TID parser (base32-sortkey to Unix microseconds)	None
pkg/tap	Tap WebSocket client with auto-reconnect and per-event acks	gorilla/websocket
pkg/did	DID document resolver (did:plc and did:web)	None (stdlib only)
pkg/eventlog	Append-only JSONL.zst writer with hourly rotation	klauspost/compress
pkg/circuitbreaker	3-state circuit breaker	None (stdlib only)
pkg/diskmon	Disk space monitor with backpressure	None (stdlib only)

Usage Examples

Parse a TID:

import "github.com/orpheuslummis/atarchive/pkg/tid"

us, err := tid.ToMicroseconds("3lbvll55z2s2i") // Unix microseconds

Connect to Tap:

import "github.com/orpheuslummis/atarchive/pkg/tap"

events := make(chan *tap.Event, 10_000)
client := tap.NewClient("ws://localhost:2480/channel", 0, events, logger, nil)
// With custom dialer:
// client := tap.NewClient(url, 0, events, logger, nil, tap.WithDialer(myDialer))
go client.Run(ctx)

Resolve a DID:

import "github.com/orpheuslummis/atarchive/pkg/did"

resolver := did.NewResolver("https://plc.directory", logger)
// With custom HTTP client:
// resolver := did.NewResolver(url, logger, did.WithHTTPClient(myClient))
doc, signingKey, err := resolver.Resolve(ctx, "did:plc:ewvi7nxzyoun6zhxrhs64oiz")

Write an event log:

import "github.com/orpheuslummis/atarchive/pkg/eventlog"

w := eventlog.NewWriter("/data/eventlog")
defer w.Close() // Critical: finalizes zstd frame

entries := []eventlog.LogEntry{
    {TapID: 1, Type: "record", Record: recordJSON},
}
n, err := w.WriteBatch(entries)

Circuit breaker:

import "github.com/orpheuslummis/atarchive/pkg/circuitbreaker"

cb := circuitbreaker.New(5, 30*time.Second, nil) // trips after 5 failures
if cb.Allow() {
    err := doWork()
    if err != nil {
        cb.RecordFailure()
    } else {
        cb.RecordSuccess()
    }
}

Running the Archive Service

The internal/ directory contains a reference application that uses these packages to archive the entire Bluesky network.

Architecture

Tap WebSocket (verified AT Protocol events)
       |
  tap.Client  (WebSocket, per-event acks, auto-reconnect)
       |
  chan *Event  (buffered 10k)
       |
  store.Batcher  (flush every 1000 events or 250ms)
       |
  +----+----+
  | Phase 1 |  eventlog.Writer -> JSONL.zst (source of truth)
  | Phase 2 |  store.StoreBatch -> PostgreSQL (derived index)
  |    .    |  onFlush -> did.Capturer (async DID doc capture)
  | Phase 3 |  store.SaveCursor
  | Phase 4 |  tap.Client.Ack (per-event acks back to Tap)
  +---------+

The two-phase flush is critical: the event log is written first. If PostgreSQL fails, the event log has the data and Tap will re-deliver (events are only acked after both phases succeed).

Quick Start

# Build
go build -o bin/atarchive ./cmd/atarchive

# Run tests (no PostgreSQL needed for unit tests)
go test ./...

# Docker
cp .env.example .env  # edit passwords
docker build -t atarchive .
docker compose up -d

Configuration

All configuration is via environment variables:

Variable	Default	Description
DATABASE_URL	(required)	PostgreSQL connection string
TAP_WS_URL	ws://localhost:2480/channel	Tap WebSocket URL
EVENT_LOG_DIR	/data/eventlog	Directory for JSONL.zst event logs
BATCH_SIZE	1000	Events per batch
BATCH_TIMEOUT	250ms	Max time before flushing a partial batch
LOG_LEVEL	info	Log level (debug, info, warn, error)
HTTP_ADDR	:9090	HTTP server address (health/metrics)
PLC_DIRECTORY_URL	https://plc.directory	PLC directory for DID resolution
SHUTDOWN_TIMEOUT	30s	Hard shutdown deadline
DISK_MIN_FREE_BYTES	5000000000	Pause ingestion below this free space
DISK_RESUME_FREE_BYTES	10000000000	Resume ingestion above this free space

Database Schema

Migrations are in migrations/ and applied automatically by the migrate init container in compose.

• record_versions - Append-only, partitioned by LIST on collection, sub-partitioned by RANGE on time_us
• records_current - Trigger-maintained latest state per (collection, did, rkey)
• identity_events - Append-only identity changes
• did_documents - DID document snapshots with extracted signing key
• event_gaps - Detected gaps in Tap event ID sequence (live mode only)
• cursor - Single-row table tracking last processed event ID

Monitoring

• GET /healthz - Liveness (always 200)
• GET /readyz - Readiness (200 when WS connected and last flush succeeded)
• GET /metrics - Prometheus metrics

License

MIT