Merge pull request #571 from streamplace/eli/handle-changes

atproto: index handle changes

Makefile

···
  		&& sed -i.bak 's/AppBskyGraphBlock\.Main/AppBskyGraphBlock\.Record/' $$(find ./js/streamplace/src/lexicons/types/place/stream -type f) \
  		&& sed -i.bak 's/PlaceStreamChatProfile\.Main/PlaceStreamChatProfile\.Record/' $$(find ./js/streamplace/src/lexicons/types/place/stream -type f) \
 		&& for x in $$(find ./js/streamplace/src/lexicons -type f -name '*.ts'); do \
-			echo 'import { AppBskyRichtextFacet, AppBskyGraphBlock, ComAtprotoRepoStrongRef, AppBskyActorDefs, ComAtprotoSyncListRepos, AppBskyActorGetProfile, AppBskyFeedGetFeedSkeleton, ComAtprotoIdentityResolveHandle, ComAtprotoModerationCreateReport, ComAtprotoRepoCreateRecord, ComAtprotoRepoDeleteRecord, ComAtprotoRepoDescribeRepo, ComAtprotoRepoGetRecord, ComAtprotoRepoListRecords, ComAtprotoRepoPutRecord, ComAtprotoRepoUploadBlob, ComAtprotoServerDescribeServer, ComAtprotoSyncGetRecord, ComAtprotoSyncListReposComAtprotoRepoCreateRecord, ComAtprotoRepoDeleteRecord, ComAtprotoRepoGetRecord, ComAtprotoRepoListRecords } from "@atproto/api"' >> $$x; \
+			echo 'import { AppBskyRichtextFacet, AppBskyGraphBlock, ComAtprotoRepoStrongRef, AppBskyActorDefs, ComAtprotoSyncListRepos, AppBskyActorGetProfile, AppBskyFeedGetFeedSkeleton, ComAtprotoIdentityResolveHandle, ComAtprotoModerationCreateReport, ComAtprotoRepoCreateRecord, ComAtprotoRepoDeleteRecord, ComAtprotoRepoDescribeRepo, ComAtprotoRepoGetRecord, ComAtprotoRepoListRecords, ComAtprotoRepoPutRecord, ComAtprotoRepoUploadBlob, ComAtprotoServerDescribeServer, ComAtprotoSyncGetRecord, ComAtprotoSyncListReposComAtprotoRepoCreateRecord, ComAtprotoRepoDeleteRecord, ComAtprotoRepoGetRecord, ComAtprotoRepoListRecords, ComAtprotoIdentityRefreshIdentity } from "@atproto/api"' >> $$x; \
 		done \
 		&& npx prettier --write $$(find ./js/streamplace/src/lexicons -type f -name '*.ts') \
 		&& find . | grep bak$$ | xargs rm

js/docs/src/content/docs/lex-reference/openapi.json

···
         }
       }
     },
+    "/xrpc/com.atproto.identity.refreshIdentity": {
+      "post": {
+        "summary": "Request that the server re-resolve an identity (DID and handle). The server may ignore this request, or require authentication, depending on the role, implementation, and policy of the server.",
+        "operationId": "com.atproto.identity.refreshIdentity",
+        "tags": ["com.atproto.identity"],
+        "responses": {
+          "200": {
+            "description": "Success",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/com.atproto.identity.defs_identityInfo"
+                }
+              }
+            }
+          },
+          "400": {
+            "description": "Bad Request",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "object",
+                  "required": ["error", "message"],
+                  "properties": {
+                    "error": {
+                      "type": "string",
+                      "oneOf": [
+                        {
+                          "const": "HandleNotFound"
+                        },
+                        {
+                          "const": "DidNotFound"
+                        },
+                        {
+                          "const": "DidDeactivated"
+                        }
+                      ]
+                    },
+                    "message": {
+                      "type": "string"
+                    }
+                  }
+                }
+              }
+            }
+          }
+        },
+        "requestBody": {
+          "required": true,
+          "content": {
+            "application/json": {
+              "schema": {
+                "type": "object",
+                "properties": {
+                  "identifier": {
+                    "type": "string",
+                    "format": "at-identifier"
+                  }
+                },
+                "required": ["identifier"]
+              }
+            }
+          }
+        }
+      }
+    },
     "/xrpc/com.atproto.identity.resolveHandle": {
       "get": {
         "summary": "Resolves an atproto handle (hostname) to a DID. Does not necessarily bi-directionally verify against the the DID document.",
···
           }
         },
         "required": ["name"]
+      },
+      "com.atproto.identity.defs_identityInfo": {
+        "type": "object",
+        "properties": {
+          "did": {
+            "type": "string",
+            "format": "did"
+          },
+          "handle": {
+            "type": "string",
+            "description": "The validated handle of the account; or 'handle.invalid' if the handle did not bi-directionally match the DID document.",
+            "format": "handle"
+          },
+          "didDoc": {
+            "description": "The complete DID document for the identity."
+          }
+        },
+        "required": ["did", "handle", "didDoc"]
       },
       "app.bsky.feed.defs_skeletonFeedPost": {
         "type": "object",

lexicons/com/atproto/identity/refreshIdentity.json

···
+{
+  "lexicon": 1,
+  "id": "com.atproto.identity.refreshIdentity",
+  "defs": {
+    "main": {
+      "type": "procedure",
+      "description": "Request that the server re-resolve an identity (DID and handle). The server may ignore this request, or require authentication, depending on the role, implementation, and policy of the server.",
+      "input": {
+        "encoding": "application/json",
+        "schema": {
+          "type": "object",
+          "required": ["identifier"],
+          "properties": {
+            "identifier": {
+              "type": "string",
+              "format": "at-identifier"
+            }
+          }
+        }
+      },
+      "output": {
+        "encoding": "application/json",
+        "schema": {
+          "type": "ref",
+          "ref": "com.atproto.identity.defs#identityInfo"
+        }
+      },
+      "errors": [
+        {
+          "name": "HandleNotFound",
+          "description": "The resolution process confirmed that the handle does not resolve to any DID."
+        },
+        {
+          "name": "DidNotFound",
+          "description": "The DID resolution process confirmed that there is no current DID."
+        },
+        {
+          "name": "DidDeactivated",
+          "description": "The DID previously existed, but has been deactivated."
+        }
+      ]
+    }
+  }
+}

pkg/atproto/atproto.go

···
 }
 
 func (atsync *ATProtoSynchronizer) SyncBlueskyRepo(ctx context.Context, handle string, mod model.Model) (*model.Repo, error) {
-	ident, err := atsync.resolveIdent(ctx, handle)
+	ident, err := atsync.resolveIdent(ctx, handle, true)
 	if err != nil {
 		return nil, fmt.Errorf("failed to resolve Bluesky handle %s: %w", handle, err)
 	}
···
 	return &newRepo, nil
 }
 
-func (atsync *ATProtoSynchronizer) resolveIdent(ctx context.Context, arg string) (*identity.Identity, error) {
+func (atsync *ATProtoSynchronizer) RefreshIdentity(ctx context.Context, did string) (*identity.Identity, error) {
+	id, err := atsync.resolveIdent(ctx, did, false)
+	if err != nil {
+		return nil, fmt.Errorf("failed to resolve ident: %w", err)
+	}
+	newRepo := model.Repo{
+		DID:    id.DID.String(),
+		PDS:    id.PDSEndpoint(),
+		Handle: id.Handle.String(),
+	}
+	err = atsync.Model.UpdateRepo(&newRepo)
+	if err != nil {
+		return nil, fmt.Errorf("failed to update repo: %w", err)
+	}
+	return id, nil
+}
+
+func (atsync *ATProtoSynchronizer) resolveIdent(ctx context.Context, arg string, cached bool) (*identity.Identity, error) {
 	if atsync.PLCDirectory == nil {
 		atsync.PLCDirectory = CustomDirectory(atsync.CLI.PLCURL)
 	}
+	if atsync.CachedPLCDirectory == nil {
+		cachedDir := identity.NewCacheDirectory(atsync.PLCDirectory, 250_000, time.Hour*24, time.Minute*2, time.Minute*5)
+		atsync.CachedPLCDirectory = &cachedDir
+	}
+	dir := atsync.PLCDirectory
+	if cached {
+		dir = atsync.CachedPLCDirectory
+	}
 	id, err := syntax.ParseAtIdentifier(arg)
 	if err != nil {
 		return nil, err
 	}
 
-	return atsync.PLCDirectory.Lookup(ctx, *id)
-}
+	resolvedID, err := dir.Lookup(ctx, *id)
+	if err != nil {
+		return nil, err
+	}
+	log.Log(ctx, "resolved ident", "id", resolvedID.DID.String(), "handle", resolvedID.Handle.String())
 
+	return resolvedID, nil
+}
 func CustomDirectory(plcURL string) identity.Directory {
 	base := identity.BaseDirectory{
 		PLCURL:     plcURL,
···
 		// primary Bluesky PDS instance only supports HTTP resolution method
 		SkipDNSDomainSuffixes: []string{".bsky.social"},
 	}
-	cached := identity.NewCacheDirectory(&base, 250_000, time.Hour*24, time.Minute*2, time.Minute*5)
-	return &cached
+	return &base
 }
 
 func DIDDoc(host string) map[string]any {

pkg/atproto/firehose.go

···
 )
 
 type ATProtoSynchronizer struct {
-	CLI          *config.CLI
-	Model        model.Model
-	StatefulDB   *statedb.StatefulDB
-	LastSeen     time.Time
-	LastEvent    time.Time
-	Noter        notificationpkg.FirebaseNotifier
-	Bus          *bus.Bus
-	PLCDirectory identity.Directory
+	CLI                *config.CLI
+	Model              model.Model
+	StatefulDB         *statedb.StatefulDB
+	LastSeen           time.Time
+	LastEvent          time.Time
+	Noter              notificationpkg.FirebaseNotifier
+	Bus                *bus.Bus
+	PLCDirectory       identity.Directory
+	CachedPLCDirectory identity.Directory
 }
 
 func (atsync *ATProtoSynchronizer) StartFirehose(ctx context.Context) error {
···
 	rsc := &events.RepoStreamCallbacks{
 		RepoCommit: func(evt *comatproto.SyncSubscribeRepos_Commit) error {
 			go atsync.handleCommitEventOps(ctx, evt)
+			return nil
+		},
+		RepoIdentity: func(evt *comatproto.SyncSubscribeRepos_Identity) error {
+			go atsync.handleIdentityEventOps(ctx, evt)
 			return nil
 		},
 		Error: func(evt *events.ErrorFrame) error {
···
 		}
 	}
 }
+
+func (atsync *ATProtoSynchronizer) handleIdentityEventOps(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Identity) {
+	handle := ""
+	if evt.Handle != nil {
+		handle = *evt.Handle
+	}
+	ctx = log.WithLogValues(ctx, "event", "identity", "did", evt.Did, "handle", handle, "func", "handleIdentityEventOps")
+	r, err := atsync.Model.GetRepo(evt.Did)
+	if err != nil {
+		log.Error(ctx, "failed to get repo", "err", err)
+		return
+	}
+	if r == nil {
+		log.Debug(ctx, "no repo found for identity", "did", evt.Did)
+		return
+	}
+	_, err = atsync.RefreshIdentity(ctx, evt.Did)
+	if err != nil {
+		log.Error(ctx, "failed to refresh ident", "err", err)
+		return
+	}
+}

pkg/atproto/handle_test.go

···
+package atproto
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	comatproto "github.com/bluesky-social/indigo/api/atproto"
+	lexutil "github.com/bluesky-social/indigo/lex/util"
+	"github.com/bluesky-social/indigo/util"
+	"github.com/stretchr/testify/require"
+	"stream.place/streamplace/pkg/bus"
+	"stream.place/streamplace/pkg/config"
+	"stream.place/streamplace/pkg/devenv"
+	"stream.place/streamplace/pkg/log"
+	"stream.place/streamplace/pkg/model"
+	"stream.place/streamplace/pkg/statedb"
+	"stream.place/streamplace/pkg/streamplace"
+)
+
+func TestHandleChange(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	dev := devenv.WithDevEnv(t)
+	t.Logf("dev: %+v", dev)
+	cli := config.CLI{
+		PublicHost: "example.com",
+		DBURL:      ":memory:",
+		RelayHost:  strings.ReplaceAll(dev.PDSURL, "http://", "ws://"),
+		PLCURL:     dev.PLCURL,
+	}
+
+	t.Logf("cli: %+v", cli)
+	b := bus.NewBus()
+	cli.DataDir = t.TempDir()
+	mod, err := model.MakeDB(":memory:")
+	require.NoError(t, err)
+	state, err := statedb.MakeDB(context.Background(), &cli, nil, mod)
+	require.NoError(t, err)
+	atsync := &ATProtoSynchronizer{
+		CLI:          &cli,
+		StatefulDB:   state,
+		Model:        mod,
+		Bus:          b,
+		PLCDirectory: dev.TestDirectory(),
+	}
+
+	done := make(chan struct{})
+
+	go func() {
+		err := atsync.StartFirehose(ctx)
+		require.NoError(t, err)
+		close(done)
+	}()
+
+	user := dev.CreateAccount(t)
+
+	msg := &streamplace.ChatMessage{
+		LexiconTypeID: "place.stream.chat.message",
+		Text:          "Hello, world!",
+		CreatedAt:     time.Now().Add(-time.Second).Format(util.ISO8601),
+		Streamer:      user.DID,
+	}
+
+	_, err = comatproto.RepoCreateRecord(ctx, user.XRPC, &comatproto.RepoCreateRecord_Input{
+		Collection: "place.stream.chat.message",
+		Repo:       user.DID,
+		Record:     &lexutil.LexiconTypeDecoder{Val: msg},
+	})
+	require.NoError(t, err)
+
+	var message *streamplace.ChatDefs_MessageView
+	err = untilNoErrors(t, func() error {
+		messages, err := mod.MostRecentChatMessages(user.DID)
+		if err != nil {
+			return err
+		}
+		if len(messages) != 1 {
+			return fmt.Errorf("expected 2 messages, got %d", len(messages))
+		}
+		message = messages[0]
+		return nil
+	})
+	require.NoError(t, err)
+	require.Equal(t, user.Handle, message.Author.Handle)
+
+	log.Log(ctx, "updating handle", "handle", "new-handle.test")
+	err = comatproto.IdentityUpdateHandle(context.Background(), user.XRPC, &comatproto.IdentityUpdateHandle_Input{
+		Handle: "new-handle.test",
+	})
+	require.NoError(t, err)
+
+	err = untilNoErrors(t, func() error {
+		messages, err := mod.MostRecentChatMessages(user.DID)
+		if err != nil {
+			return err
+		}
+		if len(messages) != 1 {
+			return fmt.Errorf("expected 2 messages, got %d", len(messages))
+		}
+		message = messages[0]
+		if message.Author.Handle != "new-handle.test" {
+			return fmt.Errorf("expected new handle, got %s", message.Author.Handle)
+		}
+		return nil
+	})
+	require.NoError(t, err)
+	require.Equal(t, message.Author.Handle, "new-handle.test")
+}

pkg/atproto/labeler_firehose.go

···
 func (atsync *ATProtoSynchronizer) StartLabelerFirehoseRetry(ctx context.Context, did string) error {
 	ctx = log.WithLogValues(ctx, "func", "StartLabelerFirehose")
 
-	ident, err := atsync.resolveIdent(ctx, did)
+	ident, err := atsync.resolveIdent(ctx, did, true)
 	if err != nil {
 		return fmt.Errorf("failed to resolve DID %s: %w", did, err)
 	}

pkg/config/config.go

···
 	slogGorm.WithHandler(tint.NewHandler(os.Stderr, &tint.Options{
 		TimeFormat: time.RFC3339,
 	})),
-	slogGorm.WithTraceAll(),
+	// slogGorm.WithTraceAll(),
 )
 
 func (cli *CLI) Parse(fs *flag.FlagSet, args []string) error {

pkg/devenv/devenv.go

···
 	"context"
 	"encoding/json"
 	"fmt"
+	"net"
+	"net/http"
 	"os/exec"
 	"path/filepath"
 	"runtime"
+	"strings"
 	"testing"
+	"time"
 
 	comatproto "github.com/bluesky-social/indigo/api/atproto"
+	"github.com/bluesky-social/indigo/atproto/identity"
 	"github.com/bluesky-social/indigo/xrpc"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
···
 		XRPC:     xrpcc,
 	}
 }
+
+// Custom RoundTripper for intercepting .test domain requests
+type TestRoundTripper struct {
+	DevEnv *DevEnv
+}
+
+func (d *DevEnv) TestHTTPClient() *http.Client {
+	return &http.Client{
+		Transport: d.TestRoundTripper(),
+	}
+}
+
+func (d *DevEnv) TestRoundTripper() *TestRoundTripper {
+	return &TestRoundTripper{DevEnv: d}
+}
+
+func (rt *TestRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	if strings.HasSuffix(req.URL.Hostname(), ".test") {
+		log.Log(context.Background(), "intercepting .test domain request", "url", req.URL.String())
+		upstreamURL := fmt.Sprintf("%s%s", rt.DevEnv.PDSURL, req.URL.Path)
+		upstreamReq, err := http.NewRequest(req.Method, upstreamURL, req.Body)
+		if err != nil {
+			return nil, err
+		}
+		upstreamReq.Header = req.Header
+		upstreamReq.Host = req.URL.Hostname()
+		upstreamResp, err := http.DefaultTransport.RoundTrip(upstreamReq)
+		if err != nil {
+			return nil, err
+		}
+		return upstreamResp, nil
+	}
+	// For non-.test domains, use the default transport
+	return http.DefaultTransport.RoundTrip(req)
+}
+
+func (d *DevEnv) TestDirectory() identity.Directory {
+	// We need to create a new directory with our custom client
+	base := identity.BaseDirectory{
+		PLCURL:     d.PLCURL,
+		HTTPClient: *d.TestHTTPClient(),
+		Resolver: net.Resolver{
+			Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
+				d := net.Dialer{Timeout: time.Second * 3}
+				return d.DialContext(ctx, network, address)
+			},
+		},
+		TryAuthoritativeDNS:   true,
+		SkipDNSDomainSuffixes: []string{".bsky.social"},
+	}
+	return &base
+}

pkg/spxrpc/com_atproto_identity.go

···
 	}
 	return &comatprototypes.IdentityResolveHandle_Output{Did: did}, nil
 }
+
+func (s *Server) handleComAtprotoIdentityRefreshIdentity(ctx context.Context, body *comatprototypes.IdentityRefreshIdentity_Input) (*comatprototypes.IdentityDefs_IdentityInfo, error) {
+	ident, err := s.ATSync.RefreshIdentity(ctx, body.Identifier)
+	if err != nil {
+		return nil, err
+	}
+	return &comatprototypes.IdentityDefs_IdentityInfo{
+		Did:    ident.DID.String(),
+		Handle: ident.Handle.String(),
+		DidDoc: ident.DIDDocument(),
+	}, nil
+}

pkg/spxrpc/stubs.go

···
 }
 
 func (s *Server) RegisterHandlersComAtproto(e *echo.Echo) error {
+	e.POST("/xrpc/com.atproto.identity.refreshIdentity", s.HandleComAtprotoIdentityRefreshIdentity)
 	e.GET("/xrpc/com.atproto.identity.resolveHandle", s.HandleComAtprotoIdentityResolveHandle)
 	e.POST("/xrpc/com.atproto.moderation.createReport", s.HandleComAtprotoModerationCreateReport)
 	e.GET("/xrpc/com.atproto.repo.describeRepo", s.HandleComAtprotoRepoDescribeRepo)
···
 	e.GET("/xrpc/com.atproto.sync.getRecord", s.HandleComAtprotoSyncGetRecord)
 	e.GET("/xrpc/com.atproto.sync.listRepos", s.HandleComAtprotoSyncListRepos)
 	return nil
+}
+
+func (s *Server) HandleComAtprotoIdentityRefreshIdentity(c echo.Context) error {
+	ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandleComAtprotoIdentityRefreshIdentity")
+	defer span.End()
+
+	var body comatprototypes.IdentityRefreshIdentity_Input
+	if err := c.Bind(&body); err != nil {
+		return err
+	}
+	var out *comatprototypes.IdentityDefs_IdentityInfo
+	var handleErr error
+	// func (s *Server) handleComAtprotoIdentityRefreshIdentity(ctx context.Context,body *comatprototypes.IdentityRefreshIdentity_Input) (*comatprototypes.IdentityDefs_IdentityInfo, error)
+	out, handleErr = s.handleComAtprotoIdentityRefreshIdentity(ctx, &body)
+	if handleErr != nil {
+		return handleErr
+	}
+	return c.JSON(200, out)
 }
 
 func (s *Server) HandleComAtprotoIdentityResolveHandle(c echo.Context) error {