client.go

package jetstream

import (
	"context"
	"fmt"
	"io"
	"iter"
	"log/slog"
	"net"
	"net/url"
	"strings"
	"sync"
	"sync/atomic"

	"github.com/jcalabro/atmos"
)

// Client is a Jetstream replay consumer. Construct one with Subscribe. A Client
// drives at most one Events iteration at a time; create separate Clients for
// concurrent streams. Close releases its resources and is safe to call
// concurrently with a running Events (the natural way to stop a live tail) and
// to call more than once.
type Client struct {
	host   string // normalized base URL, e.g. "https://host"
	engine engine

	closed    atomic.Bool
	closeOnce sync.Once
	closeErr  error
}

// Batch is a group of events delivered together by the Events iterator. It
// amortizes per-event overhead (notably cursor persistence): handle the whole
// batch, then save LastCursor to your database of choice once.
type Batch struct {
	events []Event
}

// Events returns the events in this batch. The slice is owned by the caller
// for the lifetime of the loop iteration; do not retain it past the next
// iteration without copying.
func (b *Batch) Events() []Event { return b.events }

// LastCursor returns the highest Seq in the batch, suitable for persisting as
// a resume point. Returns 0 for an empty batch.
func (b *Batch) LastCursor() uint64 {
	var max uint64
	for i := range b.events {
		if b.events[i].Seq > max {
			max = b.events[i].Seq
		}
	}
	return max
}

// Stats is a point-in-time snapshot of replay-loop progress, returned by
// Client.Stats. The Jetstream client library has no metrics registry, so this
// accessor is how a caller observes how far a replay has progressed and the
// residual gap a sustained-ingest stream is still closing before cutover.
type Stats struct {
	// Pages is the number of planSnapshot pages downloaded across the replay thus far.
	Pages uint64
	// SealedTip is the most recently learned sealed-archive tip (the pinned
	// pagination goal of the current sweep).
	SealedTip uint64
	// PlannedThrough is the continuation cursor reached so far: the highest
	// sealed seq accounted for. Equals SealedTip once a sweep completes.
	PlannedThrough uint64
	// ResidualGap is SealedTip - PlannedThrough: the sealed seqs still to
	// download before cutover. Zero once the sweep has consumed the archive.
	ResidualGap uint64
}

// Stats returns a snapshot of replay progress. It is safe to call from
// another goroutine while Events is running (e.g. a monitoring ticker), and
// after it returns. A zero-value Client (not built by Subscribe) reports a zero
// snapshot.
func (c *Client) Stats() Stats {
	if c == nil || c.engine == nil {
		return Stats{}
	}
	return c.engine.stats()
}

// Subscribe creates a Client for the given Jetstream host. host may be a bare
// hostname ("jetstream.us-west.bsky.network"), a host:port, or a full
// http(s):// URL; the scheme defaults to https.
//
// With no replay bound, the Client live-tails from the current tip or resumes
// from WithLiveCursor. WithAfterSeq starts by replaying sealed history and then
// cuts over to live. WithSnapshotOnly stops after the sealed range;
// WithBeforeSeq bounds that snapshot and requires WithSnapshotOnly.
func Subscribe(host string, opts ...Option) (*Client, error) {
	cfg := defaultConfig()
	for i, opt := range opts {
		if opt == nil {
			return nil, fmt.Errorf("jetstream: option %d is nil", i)
		}
		opt(&cfg)
	}
	if cfg.logger == nil {
		cfg.logger = slog.New(slog.NewTextHandler(io.Discard, nil))
	}

	base, err := normalizeHost(host)
	if err != nil {
		return nil, err
	}
	if err := validateConfig(&cfg); err != nil {
		return nil, err
	}

	return &Client{host: base, engine: newEngine(base, cfg)}, nil
}

// Format renders a bounded public summary without recursively formatting the
// engine, whose authenticated XRPC clients may retain bearer credentials.
// Client intentionally implements fmt.Formatter so debugging verbs such as
// %v, %+v, and %#v cannot expose constructor secrets.
func (c *Client) Format(s fmt.State, verb rune) {
	if c == nil {
		_, _ = io.WriteString(s, "<nil>")
		return
	}
	if verb != 'v' {
		_, _ = fmt.Fprintf(s, "%%!%c(*jetstream.Client)", verb)
		return
	}
	if s.Flag('#') {
		_, _ = fmt.Fprintf(s, "jetstream.Client{Host:%q, Closed:%t}", c.host, c.closed.Load())
		return
	}
	_, _ = fmt.Fprintf(s, "jetstream.Client{host:%q closed:%t}", c.host, c.closed.Load())
}

// Events streams event batches in delivery order until ctx is cancelled or a
// terminal error occurs. It is a Go range-over-func iterator:
//
//	for batch, err := range client.Events(ctx) {
//		...
//	}
//
// A non-nil err is yielded for recoverable problems; iteration continues so
// the caller may log and move on. A terminal failure is yielded as an error
// satisfying errors.Is(err, ErrFatal), after which the stream aborts and the
// iterator returns no further events — callers should stop and surface a
// failure rather than treat it as recoverable. When ctx is done or the stream
// ends, the iterator returns. Events must not be called concurrently on the
// same Client.
func (c *Client) Events(ctx context.Context) iter.Seq2[*Batch, error] {
	return func(yield func(*Batch, error) bool) {
		if c == nil || c.engine == nil {
			yield(nil, errClientNotInitialized)
			return
		}
		if c.closed.Load() {
			yield(nil, fmt.Errorf("jetstream: client is closed"))
			return
		}
		c.engine.run(ctx, yield)
	}
}

// Close releases the Client's resources. It is safe to call after Events
// returns, concurrently with a running Events (to stop a live tail), and more
// than once; the underlying engine is closed exactly once and every call
// returns that same result. Calling Events after Close yields an error.
func (c *Client) Close() error {
	if c == nil || c.engine == nil {
		return errClientNotInitialized
	}
	c.closeOnce.Do(func() {
		c.closed.Store(true)
		c.closeErr = c.engine.close()
	})
	return c.closeErr
}

// errClientNotInitialized is returned (rather than panicking) when a method is
// called on a zero-value or nil Client. Subscribe is the only constructor; a
// Client built any other way has a nil engine. Failing deterministically keeps
// API misuse from surfacing as a nil-pointer panic during cleanup or iteration.
var errClientNotInitialized = fmt.Errorf("jetstream: client not initialized (use Subscribe)")

// engine is the narrow seam between the public iterator and the replay
// implementation. Tests substitute small scripted engines through it.
type engine interface {
	// run drives the stream, invoking yield for each batch or recoverable
	// error. It returns when ctx is done, the stream ends, or yield returns
	// false.
	run(ctx context.Context, yield func(*Batch, error) bool)
	// stats returns a snapshot of backfill-loop progress for the Stats accessor.
	stats() Stats
	close() error
}

// normalizeHost turns a bare host, host:port, or URL into a normalized
// "scheme://host[:port]" base URL with no path.
//
// When no scheme is given, it defaults to https — except for loopback hosts
// (localhost, 127.0.0.0/8, ::1), which default to http since a local dev
// server almost never terminates TLS. An explicit scheme is always honored.
func normalizeHost(host string) (string, error) {
	raw := strings.TrimSpace(host)
	if raw == "" {
		return "", fmt.Errorf("jetstream: host is required")
	}
	schemeless := !strings.Contains(raw, "://")
	if schemeless {
		// Parse with a placeholder scheme so url.Parse populates Host/Hostname,
		// then pick the real default from whether the host is loopback.
		probe, err := url.Parse("https://" + raw)
		if err != nil {
			return "", fmt.Errorf("jetstream: parse host: %w", err)
		}
		scheme := "https"
		if isLoopbackHost(probe.Hostname()) {
			scheme = "http"
		}
		raw = scheme + "://" + raw
	}
	u, err := url.Parse(raw)
	if err != nil {
		return "", fmt.Errorf("jetstream: parse host: %w", err)
	}
	switch u.Scheme {
	case "http", "https":
	case "ws":
		u.Scheme = "http"
	case "wss":
		u.Scheme = "https"
	default:
		return "", fmt.Errorf("jetstream: unsupported host scheme %q", u.Scheme)
	}
	if u.Host == "" {
		return "", fmt.Errorf("jetstream: host is required")
	}
	return u.Scheme + "://" + u.Host, nil
}

// isLoopbackHost reports whether host (a hostname with no port) refers to the
// local machine: the literal "localhost" (or any *.localhost name, per RFC
// 6761) or a loopback IP literal (127.0.0.0/8, ::1).
func isLoopbackHost(host string) bool {
	if host == "" {
		return false
	}
	h := strings.ToLower(host)
	if h == "localhost" || strings.HasSuffix(h, ".localhost") {
		return true
	}
	if ip := net.ParseIP(host); ip != nil {
		return ip.IsLoopback()
	}
	return false
}

// validateConfig rejects internally inconsistent option combinations.
func validateConfig(c *config) error {
	if c.hasAPIKey && c.apiKey == "" {
		return fmt.Errorf("jetstream: API key cannot be empty")
	}
	if err := canonicalizeFilters(c); err != nil {
		return err
	}
	if c.hasAfterSeq && c.hasBeforeSeq && c.beforeSeq <= c.afterSeq {
		return fmt.Errorf("jetstream: beforeSeq (%d) must be greater than afterSeq (%d)", c.beforeSeq, c.afterSeq)
	}
	if c.snapshotOnly && !c.backfillRequested() {
		return fmt.Errorf("jetstream: WithSnapshotOnly requires a replay bound (WithAfterSeq and/or WithBeforeSeq)")
	}
	// WithBeforeSeq is an ARCHIVE upper bound, enforced by the row matcher's
	// inclusive beforeSeq. The live cutover tail reuses that same matcher, so a
	// backfill-then-live subscription with a beforeSeq would silently drop every
	// live event with seq > beforeSeq — after the brief (S, beforeSeq] window the
	// tail runs forever delivering nothing, a silent loss of in-scope data the
	// server is actively serving (CLAUDE.md: crash over silent corruption). A
	// beforeSeq is only coherent as a bounded snapshot, so require
	// WithSnapshotOnly.
	if c.hasBeforeSeq && !c.snapshotOnly {
		return fmt.Errorf("jetstream: WithBeforeSeq requires WithSnapshotOnly (beforeSeq is an archive-snapshot upper bound; on a replay that continues live it would silently drop every later event)")
	}
	return nil
}

const (
	maxClientKinds       = 4
	maxClientDIDs        = 10000
	maxClientCollections = 100
)

func canonicalizeFilters(c *config) error {
	kinds := make([]Kind, 0, len(c.kinds))
	seenKinds := make(map[Kind]struct{}, min(len(c.kinds), maxClientKinds))
	for _, kind := range c.kinds {
		switch kind {
		case KindCommit, KindIdentity, KindAccount, KindSync:
		default:
			return fmt.Errorf("jetstream: invalid kind %q", kind)
		}
		if _, ok := seenKinds[kind]; ok {
			continue
		}
		seenKinds[kind] = struct{}{}
		kinds = append(kinds, kind)
	}
	if len(kinds) > maxClientKinds {
		return fmt.Errorf("jetstream: too many kinds: %d > %d", len(kinds), maxClientKinds)
	}

	dids := make([]string, 0, min(len(c.dids), maxClientDIDs))
	seenDIDs := make(map[string]struct{}, min(len(c.dids), maxClientDIDs))
	for _, raw := range c.dids {
		if _, ok := seenDIDs[raw]; ok {
			continue
		}
		did, err := atmos.ParseDID(raw)
		if err != nil {
			return fmt.Errorf("jetstream: invalid DID %q: %w", raw, err)
		}
		seenDIDs[raw] = struct{}{}
		dids = append(dids, string(did))
		if len(dids) > maxClientDIDs {
			return fmt.Errorf("jetstream: too many DIDs: %d > %d", len(dids), maxClientDIDs)
		}
	}

	collections := make([]string, 0, min(len(c.collections), maxClientCollections))
	seenCollections := make(map[string]struct{}, min(len(c.collections), maxClientCollections))
	for _, raw := range c.collections {
		if _, ok := seenCollections[raw]; ok {
			continue
		}
		if head, ok := strings.CutSuffix(raw, ".*"); ok {
			if _, err := atmos.ParseNSID(head + ".x"); err != nil {
				return fmt.Errorf("jetstream: invalid collection wildcard %q: %w", raw, err)
			}
		} else if _, err := atmos.ParseNSID(raw); err != nil {
			return fmt.Errorf("jetstream: invalid collection %q: %w", raw, err)
		}
		seenCollections[raw] = struct{}{}
		collections = append(collections, raw)
		if len(collections) > maxClientCollections {
			return fmt.Errorf("jetstream: too many collections: %d > %d", len(collections), maxClientCollections)
		}
	}

	if len(collections) > 0 && len(kinds) > 0 {
		hasCommit := false
		for _, kind := range kinds {
			hasCommit = hasCommit || kind == KindCommit
		}
		if !hasCommit {
			return fmt.Errorf("jetstream: collections filter can never apply: kinds excludes commit")
		}
	}
	c.kinds = kinds
	c.dids = dids
	c.collections = collections
	return nil
}