From 279e71b74f6645a7d12d4f55d174ff04f645c267 Mon Sep 17 00:00:00 2001 From: bryan newbold Date: Mon, 1 Dec 2025 13:50:13 -0800 Subject: [PATCH] relay: migrate from xrpc pkg to atclient --- cmd/relay/handlers.go | 40 ++++++++++++++++----------------- cmd/relay/pull.go | 8 +++---- cmd/relay/relay/host_checker.go | 29 +++++++++++++----------- cmd/relay/stubs.go | 24 ++++++++++---------- 4 files changed, 51 insertions(+), 50 deletions(-) diff --git a/cmd/relay/handlers.go b/cmd/relay/handlers.go index 6ae91770..99e31868 100644 --- a/cmd/relay/handlers.go +++ b/cmd/relay/handlers.go @@ -8,10 +8,10 @@ import ( "strings" comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/atclient" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/cmd/relay/relay" "github.com/bluesky-social/indigo/cmd/relay/relay/models" - "github.com/bluesky-social/indigo/xrpc" "github.com/labstack/echo/v4" ) @@ -20,23 +20,23 @@ func (s *Service) handleComAtprotoSyncRequestCrawl(c echo.Context, body *comatpr ctx := c.Request().Context() if s.config.DisableRequestCrawl && !admin { - return c.JSON(http.StatusForbidden, xrpc.XRPCError{ErrStr: "Forbidden", Message: "public requestCrawl not allowed on this relay"}) + return c.JSON(http.StatusForbidden, atclient.ErrorBody{Name: "Forbidden", Message: "public requestCrawl not allowed on this relay"}) } hostname, noSSL, err := relay.ParseHostname(body.Hostname) if err != nil { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("hostname field empty or invalid: %s", body.Hostname)}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("hostname field empty or invalid: %s", body.Hostname)}) } if noSSL && !s.config.AllowInsecureHosts && !admin { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: "this relay requires host SSL"}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: "this relay requires host SSL"}) } // TODO: could ensure that query and path are empty if strings.HasPrefix(hostname, "localhost:") { if !admin { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: "can not configure localhost via public endpoint"}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: "can not configure localhost via public endpoint"}) } // else, allowed } else { @@ -45,7 +45,7 @@ func (s *Service) handleComAtprotoSyncRequestCrawl(c echo.Context, body *comatpr return nil } if banned { - return c.JSON(http.StatusUnauthorized, xrpc.XRPCError{ErrStr: "DomainBan", Message: "host domain is banned"}) + return c.JSON(http.StatusUnauthorized, atclient.ErrorBody{Name: "DomainBan", Message: "host domain is banned"}) } } @@ -55,7 +55,7 @@ func (s *Service) handleComAtprotoSyncRequestCrawl(c echo.Context, body *comatpr } if err := s.relay.HostChecker.CheckHost(ctx, hostURL); err != nil { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "HostNotFound", Message: fmt.Sprintf("host server unreachable: %s", err)}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "HostNotFound", Message: fmt.Sprintf("host server unreachable: %s", err)}) } // forward on to any sibling instances (note that sometimes is, sometimes isn't an admin request) @@ -73,7 +73,7 @@ func (s *Service) handleComAtprotoSyncListHosts(c echo.Context, cursor int64, li hosts, err := s.relay.ListHosts(ctx, cursor, limit, true) if err != nil { - return nil, c.JSON(http.StatusInternalServerError, xrpc.XRPCError{ErrStr: "DatabaseError", Message: "failed to list hosts"}) + return nil, c.JSON(http.StatusInternalServerError, atclient.ErrorBody{Name: "DatabaseError", Message: "failed to list hosts"}) } if len(hosts) == 0 { @@ -112,9 +112,9 @@ func (s *Service) handleComAtprotoSyncGetHostStatus(c echo.Context, hostname str if err != nil { if errors.Is(err, relay.ErrHostNotFound) { // TODO: test that not found DID is a 404 - return nil, c.JSON(http.StatusNotFound, xrpc.XRPCError{ErrStr: "HostNotFound", Message: "host not found"}) + return nil, c.JSON(http.StatusNotFound, atclient.ErrorBody{Name: "HostNotFound", Message: "host not found"}) } - return nil, c.JSON(http.StatusInternalServerError, xrpc.XRPCError{ErrStr: "DatabaseError", Message: "looking up host information"}) + return nil, c.JSON(http.StatusInternalServerError, atclient.ErrorBody{Name: "DatabaseError", Message: "looking up host information"}) } out := &comatproto.SyncGetHostStatus_Output{ @@ -133,7 +133,7 @@ func (s *Service) handleComAtprotoSyncListRepos(c echo.Context, cursor int64, li accounts, err := s.relay.ListAccountsDetailed(ctx, cursor, limit) if err != nil { s.logger.Error("failed to query accounts", "err", err) - return nil, c.JSON(http.StatusInternalServerError, xrpc.XRPCError{ErrStr: "DatabaseError", Message: "failed to list accounts (repos)"}) + return nil, c.JSON(http.StatusInternalServerError, atclient.ErrorBody{Name: "DatabaseError", Message: "failed to list accounts (repos)"}) } if len(accounts) == 0 { @@ -174,9 +174,9 @@ func (s *Service) handleComAtprotoSyncGetRepoStatus(c echo.Context, did syntax.D if err != nil { if errors.Is(err, relay.ErrAccountNotFound) { // TODO: test that not found DID is a 404 - return nil, c.JSON(http.StatusNotFound, xrpc.XRPCError{ErrStr: "RepoNotFound", Message: "account not found"}) + return nil, c.JSON(http.StatusNotFound, atclient.ErrorBody{Name: "RepoNotFound", Message: "account not found"}) } - return nil, c.JSON(http.StatusInternalServerError, xrpc.XRPCError{ErrStr: "DatabaseError", Message: "looking up account information"}) + return nil, c.JSON(http.StatusInternalServerError, atclient.ErrorBody{Name: "DatabaseError", Message: "looking up account information"}) } out := &comatproto.SyncGetRepoStatus_Output{ @@ -205,28 +205,28 @@ func (s *Service) handleComAtprotoSyncGetLatestCommit(c echo.Context, did syntax if err != nil { if errors.Is(err, relay.ErrAccountNotFound) { // TODO: test that not found DID is a 404 - return nil, c.JSON(http.StatusNotFound, xrpc.XRPCError{ErrStr: "RepoNotFound", Message: "account not found"}) + return nil, c.JSON(http.StatusNotFound, atclient.ErrorBody{Name: "RepoNotFound", Message: "account not found"}) } - return nil, c.JSON(http.StatusInternalServerError, xrpc.XRPCError{ErrStr: "DatabaseError", Message: "looking up account information"}) + return nil, c.JSON(http.StatusInternalServerError, atclient.ErrorBody{Name: "DatabaseError", Message: "looking up account information"}) } switch acc.AccountStatus() { case models.AccountStatusTakendown, models.AccountStatusSuspended: - return nil, c.JSON(http.StatusForbidden, xrpc.XRPCError{ErrStr: "RepoTakendown", Message: "account not active (takendown)"}) + return nil, c.JSON(http.StatusForbidden, atclient.ErrorBody{Name: "RepoTakendown", Message: "account not active (takendown)"}) case models.AccountStatusDeactivated: - return nil, c.JSON(http.StatusForbidden, xrpc.XRPCError{ErrStr: "RepoDeactivated", Message: "account not active (deactivated)"}) + return nil, c.JSON(http.StatusForbidden, atclient.ErrorBody{Name: "RepoDeactivated", Message: "account not active (deactivated)"}) case models.AccountStatusDeleted: - return nil, c.JSON(http.StatusForbidden, xrpc.XRPCError{ErrStr: "RepoDeleted", Message: "account not active (deleted)"}) + return nil, c.JSON(http.StatusForbidden, atclient.ErrorBody{Name: "RepoDeleted", Message: "account not active (deleted)"}) case models.AccountStatusActive: // pass default: - return nil, c.JSON(http.StatusForbidden, xrpc.XRPCError{ErrStr: "RepoInactive", Message: fmt.Sprintf("account not active: %s", acc.AccountStatus())}) + return nil, c.JSON(http.StatusForbidden, atclient.ErrorBody{Name: "RepoInactive", Message: fmt.Sprintf("account not active: %s", acc.AccountStatus())}) } repo, err := s.relay.GetAccountRepo(ctx, acc.UID) if err != nil { if errors.Is(err, relay.ErrAccountRepoNotFound) { - return nil, c.JSON(http.StatusNotFound, xrpc.XRPCError{ErrStr: "RepoNotSynchronized", Message: "do not know current repo state for account"}) + return nil, c.JSON(http.StatusNotFound, atclient.ErrorBody{Name: "RepoNotSynchronized", Message: "do not know current repo state for account"}) } return nil, err } diff --git a/cmd/relay/pull.go b/cmd/relay/pull.go index 6c3911c3..3f917857 100644 --- a/cmd/relay/pull.go +++ b/cmd/relay/pull.go @@ -6,11 +6,11 @@ import ( "fmt" comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/atclient" "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/cmd/relay/relay" "github.com/bluesky-social/indigo/cmd/relay/relay/models" "github.com/bluesky-social/indigo/util/cliutil" - "github.com/bluesky-social/indigo/xrpc" "github.com/urfave/cli/v3" ) @@ -64,9 +64,7 @@ func runPullHosts(ctx context.Context, cmd *cli.Command) error { return fmt.Errorf("unexpected arguments") } - client := xrpc.Client{ - Host: cmd.String("relay-host"), - } + client := atclient.NewAPIClient(cmd.String("relay-host")) skipHostChecks := cmd.Bool("skip-host-checks") @@ -93,7 +91,7 @@ func runPullHosts(ctx context.Context, cmd *cli.Command) error { cursor := "" size := cmd.Int64("batch-size") for { - resp, err := comatproto.SyncListHosts(ctx, &client, cursor, size) + resp, err := comatproto.SyncListHosts(ctx, client, cursor, size) if err != nil { return err } diff --git a/cmd/relay/relay/host_checker.go b/cmd/relay/relay/host_checker.go index a99b46e5..156d26ce 100644 --- a/cmd/relay/relay/host_checker.go +++ b/cmd/relay/relay/host_checker.go @@ -7,9 +7,9 @@ import ( "time" comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/atclient" "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/util/ssrf" - "github.com/bluesky-social/indigo/xrpc" ) // Simple interface for doing host and account status checks. @@ -42,14 +42,21 @@ func NewHostClient(userAgent string) *HostClient { } } -func (hc *HostClient) CheckHost(ctx context.Context, host string) error { - xrpcc := xrpc.Client{ - Client: hc.Client, - UserAgent: &hc.UserAgent, - Host: host, +func (hc *HostClient) apiClient(host string) *atclient.APIClient { + client := atclient.APIClient{ + Client: hc.Client, + Host: host, + Headers: map[string][]string{ + "User-Agent": []string{hc.UserAgent}, + }, } + return &client +} - _, err := comatproto.ServerDescribeServer(ctx, &xrpcc) +func (hc *HostClient) CheckHost(ctx context.Context, host string) error { + + client := hc.apiClient(host) + _, err := comatproto.ServerDescribeServer(ctx, client) if err != nil { return fmt.Errorf("%w: %w", ErrHostNotPDS, err) } @@ -62,13 +69,9 @@ func (hc *HostClient) FetchAccountStatus(ctx context.Context, ident *identity.Id return "", fmt.Errorf("account does not declare a PDS: %s", ident.DID) } - xrpcc := xrpc.Client{ - Client: hc.Client, - UserAgent: &hc.UserAgent, - Host: pdsEndpoint, - } + client := hc.apiClient(pdsEndpoint) - info, err := comatproto.SyncGetRepoStatus(ctx, &xrpcc, ident.DID.String()) + info, err := comatproto.SyncGetRepoStatus(ctx, client, ident.DID.String()) if err != nil { return "", err } diff --git a/cmd/relay/stubs.go b/cmd/relay/stubs.go index b75163e7..0f873041 100644 --- a/cmd/relay/stubs.go +++ b/cmd/relay/stubs.go @@ -6,9 +6,9 @@ import ( "strconv" comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/atclient" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/cmd/relay/relay" - "github.com/bluesky-social/indigo/xrpc" "github.com/labstack/echo/v4" "go.opentelemetry.io/otel" @@ -22,7 +22,7 @@ func (s *Service) HandleComAtprotoSyncSubscribeRepos(c echo.Context) error { if cursorQuery != "" { cval, err := strconv.ParseInt(cursorQuery, 10, 64) if err != nil || cval < 0 { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("cursor parameter invalid: %s", cursorQuery)}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("cursor parameter invalid: %s", cursorQuery)}) } cursor = &cval } @@ -37,7 +37,7 @@ func (s *Service) HandleComAtprotoSyncRequestCrawl(c echo.Context) error { var body comatproto.SyncRequestCrawl_Input if err := c.Bind(&body); err != nil { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("invalid body: %s", err)}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("invalid body: %s", err)}) } // func (s *Service) handleComAtprotoSyncRequestCrawl(ctx context.Context,body *comatproto.SyncRequestCrawl_Input) error @@ -58,7 +58,7 @@ func (s *Service) HandleComAtprotoSyncListHosts(c echo.Context) error { if limitQuery != "" { limit, err = strconv.Atoi(limitQuery) if err != nil || limit < 1 || limit > 1000 { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("limit parameter invalid or out of range: %s", limitQuery)}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("limit parameter invalid or out of range: %s", limitQuery)}) } } @@ -66,7 +66,7 @@ func (s *Service) HandleComAtprotoSyncListHosts(c echo.Context) error { if cursorQuery != "" { cursor, err = strconv.ParseInt(cursorQuery, 10, 64) if err != nil || cursor < 0 { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("cursor parameter invalid: %s", cursorQuery)}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("cursor parameter invalid: %s", cursorQuery)}) } } @@ -103,7 +103,7 @@ func (s *Service) HandleComAtprotoSyncListRepos(c echo.Context) error { if limitQuery != "" { limit, err = strconv.Atoi(limitQuery) if err != nil || limit < 1 || limit > 1000 { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("limit parameter invalid: %s", limitQuery)}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("limit parameter invalid: %s", limitQuery)}) } } @@ -111,7 +111,7 @@ func (s *Service) HandleComAtprotoSyncListRepos(c echo.Context) error { if cursorQuery != "" { cursor, err = strconv.ParseInt(cursorQuery, 10, 64) if err != nil || cursor < 0 { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("limit parameter invalid cursor: %s", cursorQuery)}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("limit parameter invalid cursor: %s", cursorQuery)}) } } @@ -133,17 +133,17 @@ func (s *Service) HandleComAtprotoSyncGetRepo(c echo.Context) error { did, err := syntax.ParseDID(didQuery) if err != nil { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("missing or invalid DID parameter: %s", err)}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("missing or invalid DID parameter: %s", err)}) } ident, err := s.relay.Dir.LookupDID(ctx, did) if err != nil { // TODO: could handle lookup errors more granularly - return c.JSON(http.StatusNotFound, xrpc.XRPCError{ErrStr: "RepoNotFound", Message: fmt.Sprintf("could not resolve DID: %s", err)}) + return c.JSON(http.StatusNotFound, atclient.ErrorBody{Name: "RepoNotFound", Message: fmt.Sprintf("could not resolve DID: %s", err)}) } pdsHost, _, err := relay.ParseHostname(ident.PDSEndpoint()) if err != nil { - return c.JSON(http.StatusNotFound, xrpc.XRPCError{ErrStr: "RepoNotFound", Message: "DID document has no valid atproto PDS endpoint"}) + return c.JSON(http.StatusNotFound, atclient.ErrorBody{Name: "RepoNotFound", Message: "DID document has no valid atproto PDS endpoint"}) } u := c.Request().URL @@ -165,7 +165,7 @@ func (s *Service) HandleComAtprotoSyncGetRepoStatus(c echo.Context) error { did, err := syntax.ParseDID(didQuery) if err != nil { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("missing or invalid DID parameter: %s", err)}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("missing or invalid DID parameter: %s", err)}) } out, handleErr := s.handleComAtprotoSyncGetRepoStatus(c, did) @@ -183,7 +183,7 @@ func (s *Service) HandleComAtprotoSyncGetLatestCommit(c echo.Context) error { did, err := syntax.ParseDID(didQuery) if err != nil { - return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("missing or invalid DID parameter: %s", err)}) + return c.JSON(http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("missing or invalid DID parameter: %s", err)}) } var out *comatproto.SyncGetLatestCommit_Output -- 2.51.2