acetheking987/bsky-social-app
1
0
Fork 0
Code Issues Actions Packages Projects Releases 1 Wiki Activity

[SDK] Migrate the repo writes cluster (#11379)

Browse Source 
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Samuel Newman
2026-08-13 22:26:20 +03:00
committed by GitHub
parent ca226f589d
commit 75e4180800
22 changed files with 488 additions and 399 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/screens/Onboarding/StepFinished/index.tsx
+11 -3
View File
@@ -8,6 +8,7 @@ import {
type Un$Typed,
} from '@atproto/api'
import {TID} from '@atproto/common-web'
import {type AtUriString} from '@atproto/syntax'
import {msg} from '@lingui/core/macro'
import {useLingui} from '@lingui/react'
import {Trans} from '@lingui/react/macro'
@@ -26,7 +27,7 @@ import {useSetHasCheckedForStarterPack} from '#/state/preferences/used-starter-p
import {getAllListMembers} from '#/state/queries/list-members'
import {preferencesQueryKey} from '#/state/queries/preferences'
import {RQKEY as profileRQKey} from '#/state/queries/profile'
import {useAgent, useAppviewClient} from '#/state/session'
import {useAgent, useAppviewClient, usePdsClient} from '#/state/session'
import {useOnboardingDispatch} from '#/state/shell'
import {
useActiveStarterPack,
@@ -59,6 +60,7 @@ export function StepFinished() {
const queryClient = useQueryClient()
const agent = useAgent()
const appviewClient = useAppviewClient()
const pdsClient = usePdsClient()
const requestNotificationsPermission = useRequestNotificationsPermission()
const activeStarterPack = useActiveStarterPack()
const setActiveStarterPack = useSetActiveStarterPack()
@@ -102,10 +104,15 @@ export function StepFinished() {
await Promise.all([
bulkWriteFollows(
agent,
pdsClient,
appviewClient,
[BSKY_APP_ACCOUNT_DID, ...(listItems?.map(i => i.subject.did) ?? [])],
starterPack
? {uri: starterPack.uri, cid: starterPack.cid}
? // the starter pack view is still legacy-typed
{
uri: starterPack.uri as AtUriString,
cid: starterPack.cid,
}
: undefined,
),
(async () => {
@@ -236,6 +243,7 @@ export function StepFinished() {
queryClient,
agent,
appviewClient,
pdsClient,
dispatch,
onboardDispatch,
activeStarterPack,
src/screens/Onboarding/StepSuggestedAccounts/index.tsx
+7 -3
View File
@@ -14,7 +14,7 @@ import {logger} from '#/logger'
import {updateProfileShadow} from '#/state/cache/profile-shadow'
import {useLanguagePrefs} from '#/state/preferences'
import {useModerationOpts} from '#/state/preferences/moderation-opts'
import {useAgent, useSession} from '#/state/session'
import {useAppviewClient, usePdsClient, useSession} from '#/state/session'
import {
OnboardingControls,
OnboardingPosition,
@@ -42,7 +42,8 @@ export function StepSuggestedAccounts() {
const t = useTheme()
const {gtMobile} = useBreakpoints()
const moderationOpts = useModerationOpts()
const agent = useAgent()
const appviewClient = useAppviewClient()
const pdsClient = usePdsClient()
const {currentAccount} = useSession()
const queryClient = useQueryClient()
@@ -119,7 +120,10 @@ export function StepSuggestedAccounts() {
followingUri: 'pending',
})
}
const uris = await wait(1e3, bulkWriteFollows(agent, followableDids))
const uris = await wait(
1e3,
bulkWriteFollows(pdsClient, appviewClient, followableDids),
)
for (const did of followableDids) {
const uri = uris.get(did)
updateProfileShadow(queryClient, did, {
src/screens/Onboarding/StepSuggestedStarterpacks/StarterPackCard.tsx
+6 -4
View File
@@ -1,6 +1,7 @@
import {useState} from 'react'
import {View} from 'react-native'
import {type AppBskyGraphDefs, AppBskyGraphStarterpack} from '@atproto/api'
import {type AtUriString} from '@atproto/syntax'
import {msg} from '@lingui/core/macro'
import {useLingui} from '@lingui/react'
import {Trans} from '@lingui/react/macro'
@@ -11,7 +12,7 @@ import {isBlockedOrBlocking, isMuted} from '#/lib/moderation/blocked-and-muted'
import {logger} from '#/logger'
import {updateProfileShadow} from '#/state/cache/profile-shadow'
import {getAllListMembers} from '#/state/queries/list-members'
import {useAgent, useAppviewClient, useSession} from '#/state/session'
import {useAppviewClient, usePdsClient, useSession} from '#/state/session'
import {bulkWriteFollows} from '#/screens/Onboarding/util'
import {AvatarStack} from '#/screens/Search/components/StarterPackCard'
import {atoms as a, useBreakpoints, useTheme, web} from '#/alf'
@@ -35,8 +36,8 @@ export function StarterPackCard({
const ax = useAnalytics()
const {currentAccount} = useSession()
const {gtPhone} = useBreakpoints()
const agent = useAgent()
const appviewClient = useAppviewClient()
const pdsClient = usePdsClient()
const queryClient = useQueryClient()
const record = view.record
const [isProcessing, setIsProcessing] = useState(false)
@@ -74,8 +75,9 @@ export function StarterPackCard({
let followUris: Map<string, string>
try {
followUris = await bulkWriteFollows(agent, dids, {
uri: view.uri,
followUris = await bulkWriteFollows(pdsClient, appviewClient, dids, {
// the starter pack view is still legacy-typed
uri: view.uri as AtUriString,
cid: view.cid,
})
} catch (e) {
src/screens/Onboarding/util.ts
+24 -27
View File
@@ -1,37 +1,34 @@
import {
type $Typed,
type AppBskyGraphFollow,
type AppBskyGraphGetFollows,
type AtpAgent,
type ComAtprotoRepoApplyWrites,
type ComAtprotoRepoStrongRef,
} from '@atproto/api'
import {TID} from '@atproto/common-web'
import {type $Typed, type Client} from '@atproto/lex'
import {
type AtIdentifierString,
type DidString,
toDatetimeString,
} from '@atproto/syntax'
import chunk from 'lodash.chunk'
import {until} from '#/lib/async/until'
import {app, com} from '#/lexicons'
export async function bulkWriteFollows(
agent: AtpAgent,
pdsClient: Client,
appviewClient: Client,
dids: string[],
via?: ComAtprotoRepoStrongRef.Main,
via?: com.atproto.repo.strongRef.Main,
) {
const session = agent.session
const did = pdsClient.assertDid
if (!session) {
throw new Error(`bulkWriteFollows failed: no session`)
}
const followRecords: $Typed<AppBskyGraphFollow.Record>[] = dids.map(did => {
const followRecords: $Typed<app.bsky.graph.follow.Main>[] = dids.map(did => {
return {
$type: 'app.bsky.graph.follow',
subject: did,
createdAt: new Date().toISOString(),
// callers hold plain dids read off legacy-typed views
subject: did as DidString,
createdAt: toDatetimeString(new Date()),
via,
}
})
const followWrites: $Typed<ComAtprotoRepoApplyWrites.Create>[] =
const followWrites: $Typed<com.atproto.repo.applyWrites.Create>[] =
followRecords.map(r => ({
$type: 'com.atproto.repo.applyWrites#create',
collection: 'app.bsky.graph.follow',
@@ -41,35 +38,35 @@ export async function bulkWriteFollows(
const chunks = chunk(followWrites, 50)
for (const chunk of chunks) {
await agent.com.atproto.repo.applyWrites({
repo: session.did,
await pdsClient.call(com.atproto.repo.applyWrites, {
repo: did,
writes: chunk,
})
}
await whenFollowsIndexed(agent, session.did, res => !!res.data.follows.length)
await whenFollowsIndexed(appviewClient, did, res => !!res.follows.length)
const followUris = new Map<string, string>()
for (const r of followWrites) {
followUris.set(
r.value.subject as string,
`at://${session.did}/app.bsky.graph.follow/${r.rkey}`,
`at://${did}/app.bsky.graph.follow/${r.rkey}`,
)
}
return followUris
}
async function whenFollowsIndexed(
agent: AtpAgent,
appviewClient: Client,
actor: string,
fn: (res: AppBskyGraphGetFollows.Response) => boolean,
fn: (res: app.bsky.graph.getFollows.$OutputBody) => boolean,
) {
await until(
5, // 5 tries
1e3, // 1s delay between tries
fn,
() =>
agent.app.bsky.graph.getFollows({
actor,
appviewClient.call(app.bsky.graph.getFollows, {
actor: actor as AtIdentifierString,
limit: 1,
}),
)
src/screens/Profile/components/GermButton.tsx
+28 -23
View File
@@ -1,10 +1,8 @@
import {Platform, View} from 'react-native'
import {Image} from 'expo-image'
import {
type AppBskyActorDefs,
type AppBskyActorGetProfile,
type AtpAgent,
} from '@atproto/api'
import {type AppBskyActorDefs} from '@atproto/api'
import {type Client} from '@atproto/lex'
import {type DidString} from '@atproto/syntax'
import {msg} from '@lingui/core/macro'
import {useLingui} from '@lingui/react'
import {Trans} from '@lingui/react/macro'
@@ -13,7 +11,7 @@ import {useMutation, useQueryClient} from '@tanstack/react-query'
import {until} from '#/lib/async/until'
import {isNetworkError} from '#/lib/strings/errors'
import {RQKEY} from '#/state/queries/profile'
import {useAgent, useSession} from '#/state/session'
import {useAppviewClient, usePdsClient, useSession} from '#/state/session'
import {atoms as a, useTheme, web} from '#/alf'
import {Button, ButtonIcon, ButtonText} from '#/components/Button'
import * as Dialog from '#/components/Dialog'
@@ -24,6 +22,7 @@ import {Loader} from '#/components/Loader'
import * as Toast from '#/components/Toast'
import {Text} from '#/components/Typography'
import {useAnalytics} from '#/analytics'
import {app, com} from '#/lexicons'
import type * as bsky from '#/types/bsky'
export function GermButton({
@@ -119,25 +118,27 @@ function GermSelfButton({did}: {did: string}) {
const ax = useAnalytics()
const {_} = useLingui()
const selfExplanationDialogControl = Dialog.useDialogControl()
const agent = useAgent()
const appviewClient = useAppviewClient()
const pdsClient = usePdsClient()
const queryClient = useQueryClient()
const {mutate: deleteDeclaration, isPending} = useMutation({
mutationFn: async () => {
const previousRecord = await agent.com.germnetwork.declaration
.get({
repo: did,
const previousRecord = await pdsClient
// the profile view is still legacy-typed, so its did is unbranded
.get(com.germnetwork.declaration, {
repo: did as DidString,
rkey: 'self',
})
.then(res => res.value)
.catch(() => null)
await agent.com.germnetwork.declaration.delete({
repo: did,
await pdsClient.delete(com.germnetwork.declaration, {
repo: did as DidString,
rkey: 'self',
})
await whenAppViewReady(agent, did, res => !res.data.associated?.germ)
await whenAppViewReady(appviewClient, did, res => !res.associated?.germ)
return previousRecord
},
@@ -147,14 +148,15 @@ function GermSelfButton({did}: {did: string}) {
async function undo() {
if (!previousRecord) return
try {
await agent.com.germnetwork.declaration.put(
{
repo: did,
rkey: 'self',
},
previousRecord,
await pdsClient.put(com.germnetwork.declaration, previousRecord, {
repo: did as DidString,
rkey: 'self',
})
await whenAppViewReady(
appviewClient,
did,
res => !!res.associated?.germ,
)
await whenAppViewReady(agent, did, res => !!res.data.associated?.germ)
await queryClient.refetchQueries({queryKey: RQKEY(did)})
Toast.show(_(msg`Germ DM reconnected`))
@@ -323,14 +325,17 @@ function platform() {
}
async function whenAppViewReady(
agent: AtpAgent,
appviewClient: Client,
actor: string,
fn: (res: AppBskyActorGetProfile.Response) => boolean,
fn: (res: app.bsky.actor.getProfile.$OutputBody) => boolean,
) {
await until(
5, // 5 tries
1e3, // 1s delay between tries
fn,
() => agent.app.bsky.actor.getProfile({actor}),
() =>
appviewClient.call(app.bsky.actor.getProfile, {
actor: actor as DidString,
}),
)
}
src/screens/Settings/FindContactsSettings.tsx
+6 -3
View File
@@ -29,7 +29,7 @@ import {
useContactsMatchesQuery,
useContactsSyncStatusQuery,
} from '#/state/queries/find-contacts'
import {useAgent, useAppviewClient, useSession} from '#/state/session'
import {useAppviewClient, usePdsClient, useSession} from '#/state/session'
import {ErrorScreen} from '#/view/com/util/error/ErrorScreen'
import {List} from '#/view/com/util/List'
import {atoms as a, tokens, useGutters, useTheme} from '#/alf'
@@ -370,7 +370,7 @@ function StatusHeader({
}) {
const {_} = useLingui()
const ax = useAnalytics()
const agent = useAgent()
const pdsClient = usePdsClient()
const client = useAppviewClient()
const queryClient = useQueryClient()
const {currentAccount} = useSession()
@@ -406,7 +406,10 @@ function StatusHeader({
followCount: didsToFollow.length,
})
const uris = await wait(500, bulkWriteFollows(agent, didsToFollow))
const uris = await wait(
500,
bulkWriteFollows(pdsClient, client, didsToFollow),
)
for (const did of didsToFollow) {
const uri = uris.get(did)
src/screens/StarterPack/StarterPackScreen.tsx
+6 -4
View File
@@ -7,6 +7,7 @@ import {
AtUri,
type ModerationOpts,
} from '@atproto/api'
import {type AtUriString} from '@atproto/syntax'
import {RichText as RichTextAPI} from '@bsky.app/sdk/richtext'
import {msg} from '@lingui/core/macro'
import {useLingui} from '@lingui/react'
@@ -37,7 +38,7 @@ import {
useDeleteStarterPackMutation,
useStarterPackQuery,
} from '#/state/queries/starter-packs'
import {useAgent, useAppviewClient, useSession} from '#/state/session'
import {useAppviewClient, usePdsClient, useSession} from '#/state/session'
import {useSetActiveStarterPack} from '#/state/shell/landing'
import {useLoggedOutViewControls} from '#/state/shell/logged-out'
import {
@@ -311,8 +312,8 @@ function Header({
const {_} = useLingui()
const t = useTheme()
const {currentAccount, hasSession} = useSession()
const agent = useAgent()
const appviewClient = useAppviewClient()
const pdsClient = usePdsClient()
const queryClient = useQueryClient()
const setActiveStarterPack = useSetActiveStarterPack()
const {requestSwitchToAccount} = useLoggedOutViewControls()
@@ -379,8 +380,9 @@ function Header({
let followUris: Map<string, string>
try {
followUris = await bulkWriteFollows(agent, dids, {
uri: starterPack.uri,
followUris = await bulkWriteFollows(pdsClient, appviewClient, dids, {
// the starter pack view is still legacy-typed
uri: starterPack.uri as AtUriString,
cid: starterPack.cid,
})
} catch (e) {