Merge branch 'main' into hailey/codegen-2

api/agnostic/actorgetPreferences.go

···
 import (
 	"context"
 
-	"github.com/bluesky-social/indigo/xrpc"
+	"github.com/bluesky-social/indigo/lex/util"
 )
 
 // ActorGetPreferences_Output is the output of a app.bsky.actor.getPreferences call.
···
 }
 
 // ActorGetPreferences calls the XRPC method "app.bsky.actor.getPreferences".
-func ActorGetPreferences(ctx context.Context, c *xrpc.Client) (*ActorGetPreferences_Output, error) {
+func ActorGetPreferences(ctx context.Context, c util.LexClient) (*ActorGetPreferences_Output, error) {
 	var out ActorGetPreferences_Output
 
 	params := map[string]interface{}{}
-	if err := c.Do(ctx, xrpc.Query, "", "app.bsky.actor.getPreferences", params, nil, &out); err != nil {
+	if err := c.LexDo(ctx, util.Query, "", "app.bsky.actor.getPreferences", params, nil, &out); err != nil {
 		return nil, err
 	}
 

api/agnostic/actorputPreferences.go

···
 import (
 	"context"
 
-	"github.com/bluesky-social/indigo/xrpc"
+	"github.com/bluesky-social/indigo/lex/util"
 )
 
 // ActorPutPreferences_Input is the input argument to a app.bsky.actor.putPreferences call.
···
 }
 
 // ActorPutPreferences calls the XRPC method "app.bsky.actor.putPreferences".
-func ActorPutPreferences(ctx context.Context, c *xrpc.Client, input *ActorPutPreferences_Input) error {
-	if err := c.Do(ctx, xrpc.Procedure, "application/json", "app.bsky.actor.putPreferences", nil, input, nil); err != nil {
+func ActorPutPreferences(ctx context.Context, c util.LexClient, input *ActorPutPreferences_Input) error {
+	if err := c.LexDo(ctx, util.Procedure, "application/json", "app.bsky.actor.putPreferences", nil, input, nil); err != nil {
 		return err
 	}
 

api/agnostic/identitygetRecommendedDidCredentials.go

···
 	"context"
 	"encoding/json"
 
-	"github.com/bluesky-social/indigo/xrpc"
+	"github.com/bluesky-social/indigo/lex/util"
 )
 
 // IdentityGetRecommendedDidCredentials calls the XRPC method "com.atproto.identity.getRecommendedDidCredentials".
-func IdentityGetRecommendedDidCredentials(ctx context.Context, c *xrpc.Client) (*json.RawMessage, error) {
+func IdentityGetRecommendedDidCredentials(ctx context.Context, c util.LexClient) (*json.RawMessage, error) {
 	var out json.RawMessage
 
-	if err := c.Do(ctx, xrpc.Query, "", "com.atproto.identity.getRecommendedDidCredentials", nil, nil, &out); err != nil {
+	if err := c.LexDo(ctx, util.Query, "", "com.atproto.identity.getRecommendedDidCredentials", nil, nil, &out); err != nil {
 		return nil, err
 	}
 

api/agnostic/identitysignPlcOperation.go

···
 	"context"
 	"encoding/json"
 
-	"github.com/bluesky-social/indigo/xrpc"
+	"github.com/bluesky-social/indigo/lex/util"
 )
 
 // IdentitySignPlcOperation_Input is the input argument to a com.atproto.identity.signPlcOperation call.
···
 }
 
 // IdentitySignPlcOperation calls the XRPC method "com.atproto.identity.signPlcOperation".
-func IdentitySignPlcOperation(ctx context.Context, c *xrpc.Client, input *IdentitySignPlcOperation_Input) (*IdentitySignPlcOperation_Output, error) {
+func IdentitySignPlcOperation(ctx context.Context, c util.LexClient, input *IdentitySignPlcOperation_Input) (*IdentitySignPlcOperation_Output, error) {
 	var out IdentitySignPlcOperation_Output
-	if err := c.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.identity.signPlcOperation", nil, input, &out); err != nil {
+	if err := c.LexDo(ctx, util.Procedure, "application/json", "com.atproto.identity.signPlcOperation", nil, input, &out); err != nil {
 		return nil, err
 	}
 

api/agnostic/identitysubmitPlcOperation.go

···
 	"context"
 	"encoding/json"
 
-	"github.com/bluesky-social/indigo/xrpc"
+	"github.com/bluesky-social/indigo/lex/util"
 )
 
 // IdentitySubmitPlcOperation_Input is the input argument to a com.atproto.identity.submitPlcOperation call.
···
 }
 
 // IdentitySubmitPlcOperation calls the XRPC method "com.atproto.identity.submitPlcOperation".
-func IdentitySubmitPlcOperation(ctx context.Context, c *xrpc.Client, input *IdentitySubmitPlcOperation_Input) error {
-	if err := c.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.identity.submitPlcOperation", nil, input, nil); err != nil {
+func IdentitySubmitPlcOperation(ctx context.Context, c util.LexClient, input *IdentitySubmitPlcOperation_Input) error {
+	if err := c.LexDo(ctx, util.Procedure, "application/json", "com.atproto.identity.submitPlcOperation", nil, input, nil); err != nil {
 		return err
 	}
 

api/agnostic/repoapplyWrites.go

···
 	"fmt"
 
 	"github.com/bluesky-social/indigo/lex/util"
-	"github.com/bluesky-social/indigo/xrpc"
 )
 
 // RepoApplyWrites_Create is a "create" in the com.atproto.repo.applyWrites schema.
···
 }
 
 // RepoApplyWrites calls the XRPC method "com.atproto.repo.applyWrites".
-func RepoApplyWrites(ctx context.Context, c *xrpc.Client, input *RepoApplyWrites_Input) (*RepoApplyWrites_Output, error) {
+func RepoApplyWrites(ctx context.Context, c util.LexClient, input *RepoApplyWrites_Input) (*RepoApplyWrites_Output, error) {
 	var out RepoApplyWrites_Output
-	if err := c.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.applyWrites", nil, input, &out); err != nil {
+	if err := c.LexDo(ctx, util.Procedure, "application/json", "com.atproto.repo.applyWrites", nil, input, &out); err != nil {
 		return nil, err
 	}
 

api/agnostic/repocreateRecord.go

···
 import (
 	"context"
 
-	"github.com/bluesky-social/indigo/xrpc"
+	"github.com/bluesky-social/indigo/lex/util"
 )
 
 // RepoDefs_CommitMeta is a "commitMeta" in the com.atproto.repo.defs schema.
···
 }
 
 // RepoCreateRecord calls the XRPC method "com.atproto.repo.createRecord".
-func RepoCreateRecord(ctx context.Context, c *xrpc.Client, input *RepoCreateRecord_Input) (*RepoCreateRecord_Output, error) {
+func RepoCreateRecord(ctx context.Context, c util.LexClient, input *RepoCreateRecord_Input) (*RepoCreateRecord_Output, error) {
 	var out RepoCreateRecord_Output
-	if err := c.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.createRecord", nil, input, &out); err != nil {
+	if err := c.LexDo(ctx, util.Procedure, "application/json", "com.atproto.repo.createRecord", nil, input, &out); err != nil {
 		return nil, err
 	}
 

api/agnostic/repogetRecord.go

···
 	"context"
 	"encoding/json"
 
-	"github.com/bluesky-social/indigo/xrpc"
+	"github.com/bluesky-social/indigo/lex/util"
 )
 
 // RepoGetRecord_Output is the output of a com.atproto.repo.getRecord call.
···
 // collection: The NSID of the record collection.
 // repo: The handle or DID of the repo.
 // rkey: The Record Key.
-func RepoGetRecord(ctx context.Context, c *xrpc.Client, cid string, collection string, repo string, rkey string) (*RepoGetRecord_Output, error) {
+func RepoGetRecord(ctx context.Context, c util.LexClient, cid string, collection string, repo string, rkey string) (*RepoGetRecord_Output, error) {
 	var out RepoGetRecord_Output
 
 	params := map[string]interface{}{
···
 		"repo":       repo,
 		"rkey":       rkey,
 	}
-	if err := c.Do(ctx, xrpc.Query, "", "com.atproto.repo.getRecord", params, nil, &out); err != nil {
+	if err := c.LexDo(ctx, util.Query, "", "com.atproto.repo.getRecord", params, nil, &out); err != nil {
 		return nil, err
 	}
 

api/agnostic/repolistRecords.go

···
 	"context"
 	"encoding/json"
 
-	"github.com/bluesky-social/indigo/xrpc"
+	"github.com/bluesky-social/indigo/lex/util"
 )
 
 // RepoListRecords_Output is the output of a com.atproto.repo.listRecords call.
···
 // limit: The number of records to return.
 // repo: The handle or DID of the repo.
 // reverse: Flag to reverse the order of the returned records.
-func RepoListRecords(ctx context.Context, c *xrpc.Client, collection string, cursor string, limit int64, repo string, reverse bool) (*RepoListRecords_Output, error) {
+func RepoListRecords(ctx context.Context, c util.LexClient, collection string, cursor string, limit int64, repo string, reverse bool) (*RepoListRecords_Output, error) {
 	var out RepoListRecords_Output
 
 	params := map[string]interface{}{
···
 		"repo":       repo,
 		"reverse":    reverse,
 	}
-	if err := c.Do(ctx, xrpc.Query, "", "com.atproto.repo.listRecords", params, nil, &out); err != nil {
+	if err := c.LexDo(ctx, util.Query, "", "com.atproto.repo.listRecords", params, nil, &out); err != nil {
 		return nil, err
 	}
 

api/agnostic/repoputRecord.go

···
 import (
 	"context"
 
-	"github.com/bluesky-social/indigo/xrpc"
+	"github.com/bluesky-social/indigo/lex/util"
 )
 
 // RepoPutRecord_Input is the input argument to a com.atproto.repo.putRecord call.
···
 }
 
 // RepoPutRecord calls the XRPC method "com.atproto.repo.putRecord".
-func RepoPutRecord(ctx context.Context, c *xrpc.Client, input *RepoPutRecord_Input) (*RepoPutRecord_Output, error) {
+func RepoPutRecord(ctx context.Context, c util.LexClient, input *RepoPutRecord_Input) (*RepoPutRecord_Output, error) {
 	var out RepoPutRecord_Output
-	if err := c.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.putRecord", nil, input, &out); err != nil {
+	if err := c.LexDo(ctx, util.Procedure, "application/json", "com.atproto.repo.putRecord", nil, input, &out); err != nil {
 		return nil, err
 	}
 

api/bsky/feeddefs.go

···
 type FeedDefs_ReasonRepost struct {
 	LexiconTypeID string                      `json:"$type,const=app.bsky.feed.defs#reasonRepost" cborgen:"$type,const=app.bsky.feed.defs#reasonRepost"`
 	By            *ActorDefs_ProfileViewBasic `json:"by" cborgen:"by"`
+	Cid           *string                     `json:"cid,omitempty" cborgen:"cid,omitempty"`
 	IndexedAt     string                      `json:"indexedAt" cborgen:"indexedAt"`
+	Uri           *string                     `json:"uri,omitempty" cborgen:"uri,omitempty"`
 }
 
 // FeedDefs_ReplyRef is a "replyRef" in the app.bsky.feed.defs schema.

api/bsky/feedlike.go

···
 	LexiconTypeID string                         `json:"$type,const=app.bsky.feed.like" cborgen:"$type,const=app.bsky.feed.like"`
 	CreatedAt     string                         `json:"createdAt" cborgen:"createdAt"`
 	Subject       *comatprototypes.RepoStrongRef `json:"subject" cborgen:"subject"`
+	Via           *comatprototypes.RepoStrongRef `json:"via,omitempty" cborgen:"via,omitempty"`
 }

api/bsky/feedrepost.go

···
 	LexiconTypeID string                         `json:"$type,const=app.bsky.feed.repost" cborgen:"$type,const=app.bsky.feed.repost"`
 	CreatedAt     string                         `json:"createdAt" cborgen:"createdAt"`
 	Subject       *comatprototypes.RepoStrongRef `json:"subject" cborgen:"subject"`
+	Via           *comatprototypes.RepoStrongRef `json:"via,omitempty" cborgen:"via,omitempty"`
 }

api/bsky/notificationlistNotifications.go

···
 	IndexedAt string                             `json:"indexedAt" cborgen:"indexedAt"`
 	IsRead    bool                               `json:"isRead" cborgen:"isRead"`
 	Labels    []*comatprototypes.LabelDefs_Label `json:"labels,omitempty" cborgen:"labels,omitempty"`
-	// reason: Expected values are 'like', 'repost', 'follow', 'mention', 'reply', 'quote', 'starterpack-joined', 'verified', and 'unverified'.
+	// reason: The reason why this notification was delivered - e.g. your post was liked, or you received a new follower.
 	Reason        string                   `json:"reason" cborgen:"reason"`
 	ReasonSubject *string                  `json:"reasonSubject,omitempty" cborgen:"reasonSubject,omitempty"`
 	Record        *util.LexiconTypeDecoder `json:"record" cborgen:"record"`

atproto/crypto/jwk.go

···
+package crypto
+
+import (
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"math/big"
+
+	secp256k1 "gitlab.com/yawning/secp256k1-voi"
+	secp256k1secec "gitlab.com/yawning/secp256k1-voi/secec"
+)
+
+// Representation of a JSON Web Key (JWK), as relevant to the keys supported by this package.
+//
+// Expected to be marshalled/unmarshalled as JSON.
+type JWK struct {
+	KeyType string  `json:"kty"`
+	Curve   string  `json:"crv"`
+	X       string  `json:"x"` // base64url, no padding
+	Y       string  `json:"y"` // base64url, no padding
+	Use     string  `json:"use,omitempty"`
+	KeyID   *string `json:"kid,omitempty"`
+}
+
+// Loads a [PublicKey] from JWK (serialized as JSON bytes)
+func ParsePublicJWKBytes(jwkBytes []byte) (PublicKey, error) {
+	var jwk JWK
+	if err := json.Unmarshal(jwkBytes, &jwk); err != nil {
+		return nil, fmt.Errorf("parsing JWK JSON: %w", err)
+	}
+	return ParsePublicJWK(jwk)
+}
+
+// Loads a [PublicKey] from JWK struct.
+func ParsePublicJWK(jwk JWK) (PublicKey, error) {
+
+	if jwk.KeyType != "EC" {
+		return nil, fmt.Errorf("unsupported JWK key type: %s", jwk.KeyType)
+	}
+
+	// base64url with no encoding
+	xbuf, err := base64.RawURLEncoding.DecodeString(jwk.X)
+	if err != nil {
+		return nil, fmt.Errorf("invalid JWK base64 encoding: %w", err)
+	}
+	ybuf, err := base64.RawURLEncoding.DecodeString(jwk.Y)
+	if err != nil {
+		return nil, fmt.Errorf("invalid JWK base64 encoding: %w", err)
+	}
+
+	switch jwk.Curve {
+	case "P-256":
+		curve := elliptic.P256()
+
+		var x, y big.Int
+		x.SetBytes(xbuf)
+		y.SetBytes(ybuf)
+
+		if !curve.Params().IsOnCurve(&x, &y) {
+			return nil, fmt.Errorf("invalid P-256 public key (not on curve)")
+		}
+		pubECDSA := &ecdsa.PublicKey{
+			Curve: curve,
+			X:     &x,
+			Y:     &y,
+		}
+		pub := PublicKeyP256{pubP256: *pubECDSA}
+		err := pub.checkCurve()
+		if err != nil {
+			return nil, err
+		}
+		return &pub, nil
+	case "secp256k1": // K-256
+		if len(xbuf) != 32 || len(ybuf) != 32 {
+			return nil, fmt.Errorf("invalid K-256 coordinates")
+		}
+		xarr := ([32]byte)(xbuf[:32])
+		yarr := ([32]byte)(ybuf[:32])
+		p, err := secp256k1.NewPointFromCoords(&xarr, &yarr)
+		if err != nil {
+			return nil, fmt.Errorf("invalid K-256 coordinates: %w", err)
+		}
+		pubK, err := secp256k1secec.NewPublicKeyFromPoint(p)
+		if err != nil {
+			return nil, fmt.Errorf("invalid K-256/secp256k1 public key: %w", err)
+		}
+		pub := PublicKeyK256{pubK256: pubK}
+		err = pub.ensureBytes()
+		if err != nil {
+			return nil, err
+		}
+		return &pub, nil
+	default:
+		return nil, fmt.Errorf("unsupported JWK cryptography: %s", jwk.Curve)
+	}
+}
+
+func (k *PublicKeyP256) JWK() (*JWK, error) {
+	jwk := JWK{
+		KeyType: "EC",
+		Curve:   "P-256",
+		X:       base64.RawURLEncoding.EncodeToString(k.pubP256.X.Bytes()),
+		Y:       base64.RawURLEncoding.EncodeToString(k.pubP256.Y.Bytes()),
+	}
+	return &jwk, nil
+}
+
+func (k *PublicKeyK256) JWK() (*JWK, error) {
+	raw := k.UncompressedBytes()
+	if len(raw) != 65 {
+		return nil, fmt.Errorf("unexpected K-256 bytes size")
+	}
+	xbytes := raw[1:33]
+	ybytes := raw[33:65]
+	jwk := JWK{
+		KeyType: "EC",
+		Curve:   "secp256k1",
+		X:       base64.RawURLEncoding.EncodeToString(xbytes),
+		Y:       base64.RawURLEncoding.EncodeToString(ybytes),
+	}
+	return &jwk, nil
+}

atproto/crypto/jwk_test.go

···
+package crypto
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseJWK(t *testing.T) {
+	assert := assert.New(t)
+
+	jwkTestFixtures := []string{
+		// https://datatracker.ietf.org/doc/html/rfc7517#appendix-A.1
+		`{
+			"kty":"EC",
+          	"crv":"P-256",
+          	"x":"MKBCTNIcKUSDii11ySs3526iDZ8AiTo7Tu6KPAqv7D4",
+          	"y":"4Etl6SRW2YiLUrN5vfvVHuhp7x8PxltmWWlbbM4IFyM",
+          	"d":"870MB6gfuTJ4HtUnUvYMyJpr5eUZNP4Bk43bVdj3eAE",
+          	"use":"enc",
+          	"kid":"1"
+      	}`,
+		// https://openid.net/specs/draft-jones-json-web-key-03.html; with kty in addition to alg
+		`{
+			"alg":"EC",
+			"kty":"EC",
+			"crv":"P-256",
+			"x":"MKBCTNIcKUSDii11ySs3526iDZ8AiTo7Tu6KPAqv7D4",
+			"y":"4Etl6SRW2YiLUrN5vfvVHuhp7x8PxltmWWlbbM4IFyM",
+			"use":"enc",
+			"kid":"1"
+		}`,
+		// https://w3c-ccg.github.io/lds-ecdsa-secp256k1-2019/; with kty in addition to alg
+		`{
+    		"alg": "EC",
+    		"kty": "EC",
+    		"crv": "secp256k1",
+    		"kid": "JUvpllMEYUZ2joO59UNui_XYDqxVqiFLLAJ8klWuPBw",
+    		"x": "dWCvM4fTdeM0KmloF57zxtBPXTOythHPMm1HCLrdd3A",
+    		"y": "36uMVGM7hnw-N6GnjFcihWE3SkrhMLzzLCdPMXPEXlA"
+  		}`,
+	}
+
+	for _, jwkBytes := range jwkTestFixtures {
+		_, err := ParsePublicJWKBytes([]byte(jwkBytes))
+		assert.NoError(err)
+	}
+}
+
+func TestP256GenJWK(t *testing.T) {
+	assert := assert.New(t)
+
+	priv, err := GeneratePrivateKeyP256()
+	if err != nil {
+		t.Fatal(err)
+	}
+	pub, err := priv.PublicKey()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pk, ok := pub.(*PublicKeyP256)
+	if !ok {
+		t.Fatal()
+	}
+	jwk, err := pk.JWK()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = ParsePublicJWK(*jwk)
+	assert.NoError(err)
+}
+
+func TestK256GenJWK(t *testing.T) {
+	assert := assert.New(t)
+
+	priv, err := GeneratePrivateKeyK256()
+	if err != nil {
+		t.Fatal(err)
+	}
+	pub, err := priv.PublicKey()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pk, ok := pub.(*PublicKeyK256)
+	if !ok {
+		t.Fatal()
+	}
+	jwk, err := pk.JWK()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = ParsePublicJWK(*jwk)
+	assert.NoError(err)
+}

atproto/crypto/keys.go

···
 	// For systems with no compressed/uncompressed distinction, returns the same
 	// value as Bytes().
 	UncompressedBytes() []byte
+
+	// Serialization as JWK struct (which can be marshalled to JSON)
+	JWK() (*JWK, error)
 }
 
 var ErrInvalidSignature = errors.New("crytographic signature invalid")

atproto/data/parse.go

···
 
 func parseFloat(f float64) (int64, error) {
 	if f != float64(int64(f)) {
-		return 0, fmt.Errorf("number was is not a safe integer: %f", f)
+		return 0, fmt.Errorf("number is not a safe integer: %f", f)
 	}
 	return int64(f), nil
 }

cmd/collectiondir/serve.go

···
 	"context"
 	"encoding/csv"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log/slog"
 	"net"
 	"net/http"
+	_ "net/http/pprof"
 	"net/url"
 	"os"
 	"os/signal"
···
 	if err != nil {
 		return fmt.Errorf("lru init, %w", err)
 	}
-	cs.wg.Add(1)
-	go cs.ingestReceiver()
 	cs.log = log
 	cs.ctx = cctx.Context
 	cs.AdminToken = cctx.String("admin-token")
 	cs.ExepctedAuthHeader = "Bearer " + cs.AdminToken
+	cs.wg.Add(1)
+	go cs.ingestReceiver()
 	pebblePath := cctx.String("pebble")
 	cs.pcd = &PebbleCollectionDirectory{
 		log: cs.log,
···
 		}
 	}
 	cs.statsCacheFresh.L = &cs.statsCacheLock
-	errchan := make(chan error, 3)
-	apiAddr := cctx.String("api-listen")
+
+	apiServerEcho, err := cs.createApiServer(cctx.Context, cctx.String("api-listen"))
+	if err != nil {
+		return err
+	}
 	cs.wg.Add(1)
-	go func() {
-		errchan <- cs.StartApiServer(cctx.Context, apiAddr)
-	}()
-	metricsAddr := cctx.String("metrics-listen")
+	go func() { cs.StartApiServer(cctx.Context, apiServerEcho) }()
+
+	cs.createMetricsServer(cctx.String("metrics-listen"))
 	cs.wg.Add(1)
-	go func() {
-		errchan <- cs.StartMetricsServer(cctx.Context, metricsAddr)
-	}()
+	go func() { cs.StartMetricsServer(cctx.Context) }()
 
 	upstream := cctx.String("upstream")
 	if upstream != "" {
···
 		go cs.handleFirehose(fhevents)
 	}
 
-	select {
-	case <-signals:
-		log.Info("received shutdown signal")
-		go errchanlog(cs.log, "server error", errchan)
-		return cs.Shutdown()
-	case err := <-errchan:
-		if err != nil {
-			log.Error("server error", "err", err)
-			go errchanlog(cs.log, "server error", errchan)
-			return cs.Shutdown()
-		}
-	}
-	return nil
+	<-signals
+	log.Info("received shutdown signal")
+	return cs.Shutdown()
 }
 
 func (cs *collectionServer) openDau() error {
···
 	return nil
 }
 
-func errchanlog(log *slog.Logger, msg string, errchan <-chan error) {
-	for err := range errchan {
-		log.Error(msg, "err", err)
-	}
-}
-
 func (cs *collectionServer) Shutdown() error {
 	close(cs.shutdown)
-	go func() {
+
+	func() {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
 		cs.log.Info("metrics shutdown start")
-		sherr := cs.metricsServer.Shutdown(context.Background())
+		sherr := cs.metricsServer.Shutdown(ctx)
 		cs.log.Info("metrics shutdown", "err", sherr)
 	}()
-	cs.log.Info("api shutdown start...")
-	err := cs.apiServer.Shutdown(context.Background())
-	cs.log.Info("api shutdown, thread wait...", "err", err)
-	cs.wg.Wait()
+
+	func() {
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+
+		cs.log.Info("api shutdown start...")
+		err := cs.apiServer.Shutdown(ctx)
+		cs.log.Info("api shutdown, thread wait...", "err", err)
+	}()
+
 	cs.log.Info("threads done, db close...")
-	ee := cs.pcd.Close()
-	if ee != nil {
-		cs.log.Error("failed to shutdown pebble", "err", ee)
+	err := cs.pcd.Close()
+	if err != nil {
+		cs.log.Error("failed to shutdown pebble", "err", err)
 	}
 	cs.log.Info("db done. done.")
+	cs.wg.Wait()
 	return err
 }
 
···
 	}
 }
 
-func (cs *collectionServer) StartMetricsServer(ctx context.Context, addr string) error {
-	defer cs.wg.Done()
-	defer cs.log.Info("metrics server exit")
+func (cs *collectionServer) createMetricsServer(addr string) {
+	e := echo.New()
+	e.GET("/metrics", echo.WrapHandler(promhttp.Handler()))
+	e.Any("/debug/pprof/*", echo.WrapHandler(http.DefaultServeMux))
+
 	cs.metricsServer = &http.Server{
 		Addr:    addr,
-		Handler: promhttp.Handler(),
+		Handler: e,
 	}
-	return cs.metricsServer.ListenAndServe()
 }
 
-func (cs *collectionServer) StartApiServer(ctx context.Context, addr string) error {
+func (cs *collectionServer) StartMetricsServer(ctx context.Context) {
 	defer cs.wg.Done()
-	defer cs.log.Info("api server exit")
+	defer cs.log.Info("metrics server exit")
+
+	err := cs.metricsServer.ListenAndServe()
+	if err != nil && !errors.Is(err, http.ErrServerClosed) {
+		slog.Error("error in metrics server", "err", err)
+		os.Exit(1)
+	}
+}
+
+func (cs *collectionServer) createApiServer(ctx context.Context, addr string) (*echo.Echo, error) {
 	var lc net.ListenConfig
 	li, err := lc.Listen(ctx, "tcp", addr)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	e := echo.New()
 	e.HideBanner = true
···
 		Handler: e,
 	}
 	cs.apiServer = srv
-	return srv.Serve(li)
+	return e, nil
+}
+
+func (cs *collectionServer) StartApiServer(ctx context.Context, e *echo.Echo) {
+	defer cs.wg.Done()
+	defer cs.log.Info("api server exit")
+	err := cs.apiServer.Serve(e.Listener)
+	if err != nil && !errors.Is(err, http.ErrServerClosed) {
+		slog.Error("error in api server", "err", err)
+		os.Exit(1)
+	}
 }
 
 const statsCacheDuration = time.Second * 300

events/dbpersist/dbpersist.go

···
 		switch {
 		case e.RepoCommit != nil:
 			e.RepoCommit.Seq = int64(item.Seq)
+		case e.RepoSync != nil:
+			e.RepoSync.Seq = int64(item.Seq)
 		case e.RepoIdentity != nil:
 			e.RepoIdentity.Seq = int64(item.Seq)
 		case e.RepoAccount != nil:
···
 	switch {
 	case e.RepoCommit != nil:
 		rer, err = p.RecordFromRepoCommit(ctx, e.RepoCommit)
+		if err != nil {
+			return err
+		}
+	case e.RepoSync != nil:
+		rer, err = p.RecordFromRepoSync(ctx, e.RepoSync)
 		if err != nil {
 			return err
 		}
···
 	return &rer, nil
 }
 
+func (p *DbPersistence) RecordFromRepoSync(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Sync) (*RepoEventRecord, error) {
+
+	uid, err := p.uidForDid(ctx, evt.Did)
+	if err != nil {
+		return nil, err
+	}
+
+	t, err := time.Parse(util.ISO8601, evt.Time)
+	if err != nil {
+		return nil, err
+	}
+
+	rer := RepoEventRecord{
+		Repo: uid,
+		Type: "repo_sync",
+		Time: t,
+		Rev:  evt.Rev,
+	}
+
+	return &rer, nil
+}
+
 func (p *DbPersistence) Playback(ctx context.Context, since int64, cb func(*events.XRPCStreamEvent) error) error {
 	pageSize := 1000
 
···
 			switch {
 			case record.Commit != nil:
 				streamEvent, err = p.hydrateCommit(ctx, record)
+			case record.Type == "repo_sync":
+				streamEvent, err = p.hydrateSyncEvent(ctx, record)
 			case record.Type == "repo_identity":
 				streamEvent, err = p.hydrateIdentityEvent(ctx, record)
 			case record.Type == "repo_account":
···
 	}
 
 	return &events.XRPCStreamEvent{RepoCommit: out}, nil
+}
+
+func (p *DbPersistence) hydrateSyncEvent(ctx context.Context, rer *RepoEventRecord) (*events.XRPCStreamEvent, error) {
+
+	did, err := p.didForUid(ctx, rer.Repo)
+	if err != nil {
+		return nil, err
+	}
+
+	evt := &comatproto.SyncSubscribeRepos_Sync{
+		Seq:  int64(rer.Seq),
+		Did:  did,
+		Time: rer.Time.Format(util.ISO8601),
+		Rev:  rer.Rev,
+	}
+
+	cs, err := p.readCarSlice(ctx, rer)
+	if err != nil {
+		return nil, fmt.Errorf("read car slice: %w", err)
+	}
+	evt.Blocks = cs
+
+	return &events.XRPCStreamEvent{RepoSync: evt}, nil
 }
 
 func (p *DbPersistence) readCarSlice(ctx context.Context, rer *RepoEventRecord) ([]byte, error) {

events/diskpersist/diskpersist.go

···
 	evtKindTombstone = 3
 	evtKindIdentity  = 4
 	evtKindAccount   = 5
+	evtKindSync      = 6
 )
 
 var emptyHeader = make([]byte, headerSize)
···
 	switch {
 	case e.RepoCommit != nil:
 		e.RepoCommit.Seq = seq
+	case e.RepoSync != nil:
+		e.RepoSync.Seq = seq
 	case e.RepoIdentity != nil:
 		e.RepoIdentity.Seq = seq
 	case e.RepoAccount != nil:
···
 		evtKind = evtKindCommit
 		did = e.RepoCommit.Repo
 		if err := e.RepoCommit.MarshalCBOR(cw); err != nil {
+			return fmt.Errorf("failed to marshal: %w", err)
+		}
+	case e.RepoSync != nil:
+		evtKind = evtKindSync
+		did = e.RepoSync.Did
+		if err := e.RepoSync.MarshalCBOR(cw); err != nil {
 			return fmt.Errorf("failed to marshal: %w", err)
 		}
 	case e.RepoIdentity != nil:
···
 			}
 			evt.Seq = h.Seq
 			if err := cb(&events.XRPCStreamEvent{RepoCommit: &evt}); err != nil {
+				return nil, err
+			}
+		case evtKindSync:
+			var evt atproto.SyncSubscribeRepos_Sync
+			if err := evt.UnmarshalCBOR(io.LimitReader(bufr, h.Len64())); err != nil {
+				return nil, err
+			}
+			evt.Seq = h.Seq
+			if err := cb(&events.XRPCStreamEvent{RepoSync: &evt}); err != nil {
 				return nil, err
 			}
 		case evtKindIdentity:

events/yolopersist/yolopersist.go

···
 	switch {
 	case e.RepoCommit != nil:
 		e.RepoCommit.Seq = yp.seq
+	case e.RepoSync != nil:
+		e.RepoSync.Seq = yp.seq
 	case e.RepoIdentity != nil:
 		e.RepoIdentity.Seq = yp.seq
 	case e.RepoAccount != nil:

lex/gen.go

···
 	pf("\t\"fmt\"\n")
 	pf("\t\"encoding/json\"\n")
 	pf("\tcbg \"github.com/whyrusleeping/cbor-gen\"\n")
-	pf("\t\"github.com/bluesky-social/indigo/xrpc\"\n")
 	pf("\t\"github.com/bluesky-social/indigo/lex/util\"\n")
 	for _, xpkg := range packages {
 		if xpkg.Prefix != pkg.Prefix {

lex/type_schema.go

···
 	pf := printerf(w)
 	fname := typename
 
-	params := "ctx context.Context, c *xrpc.Client"
+	params := "ctx context.Context, c util.LexClient"
 	inpvar := "nil"
 	inpenc := ""
 
···
 	var reqtype string
 	switch s.Type {
 	case "procedure":
-		reqtype = "xrpc.Procedure"
+		reqtype = "util.Procedure"
 	case "query":
-		reqtype = "xrpc.Query"
+		reqtype = "util.Query"
 	default:
 		return fmt.Errorf("can only generate RPC for Query or Procedure (got %s)", s.Type)
 	}
 
-	pf("\tif err := c.Do(ctx, %s, %q, \"%s\", %s, %s, %s); err != nil {\n", reqtype, inpenc, s.id, queryparams, inpvar, outvar)
+	pf("\tif err := c.LexDo(ctx, %s, %q, \"%s\", %s, %s, %s); err != nil {\n", reqtype, inpenc, s.id, queryparams, inpvar, outvar)
 	pf("\t\treturn %s\n", errRet)
 	pf("\t}\n\n")
 	pf("\treturn %s\n", outRet)

lex/util/client.go

···
+package util
+
+import (
+	"context"
+	"net/http"
+)
+
+const (
+	Query     = http.MethodGet
+	Procedure = http.MethodPost
+)
+
+// API client interface used in lexgen.
+//
+// 'method' is the HTTP method type. 'inputEncoding' is the Content-Type for bodyData in Procedure calls. 'params' are query parameters. 'bodyData' should be either 'nil', an [io.Reader], or a type which can be marshalled to JSON. 'out' is optional; if not nil it should be a pointer to a type which can be un-Marshaled as JSON, for the response body.
+type LexClient interface {
+	LexDo(ctx context.Context, method string, inputEncoding string, endpoint string, params map[string]any, bodyData any, out any) error
+}

xrpc/xrpc.go

···
 	return c.Client
 }
 
-type XRPCRequestType int
+var (
+	Query     = http.MethodGet
+	Procedure = http.MethodPost
+)
 
 type AuthInfo struct {
 	AccessJwt  string `json:"accessJwt"`
···
 	Reset     time.Time
 }
 
-const (
-	Query = XRPCRequestType(iota)
-	Procedure
-)
-
 // makeParams converts a map of string keys and any values into a URL-encoded string.
 // If a value is a slice of strings, it will be joined with commas.
 // Generally the values will be strings, numbers, booleans, or slices of strings
···
 	return params.Encode()
 }
 
-func (c *Client) Do(ctx context.Context, kind XRPCRequestType, inpenc string, method string, params map[string]interface{}, bodyobj interface{}, out interface{}) error {
+func (c *Client) Do(ctx context.Context, kind string, inpenc string, method string, params map[string]interface{}, bodyobj interface{}, out interface{}) error {
 	var body io.Reader
 	if bodyobj != nil {
 		if rr, ok := bodyobj.(io.Reader); ok {
···
 	case Procedure:
 		m = "POST"
 	default:
-		return fmt.Errorf("unsupported request kind: %d", kind)
+		return fmt.Errorf("unsupported request kind: %s", kind)
 	}
 
 	var paramStr string
···
 
 	return nil
 }
+
+func (c *Client) LexDo(ctx context.Context, method string, inputEncoding string, endpoint string, params map[string]any, bodyData any, out any) error {
+	return c.Do(ctx, method, inputEncoding, endpoint, params, bodyData, out)
+}