[SDK] Migrate the post pipeline and blob uploads (#11380)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

src/lib/api/index.ts

@@ -1,28 +1,14 @@
-import {
-  type $Typed,
-  type AppBskyEmbedExternal,
-  type AppBskyEmbedGallery,
-  type AppBskyEmbedImages,
-  type AppBskyEmbedRecord,
-  type AppBskyEmbedRecordWithMedia,
-  type AppBskyEmbedVideo,
-  AppBskyFeedPost,
-  type AtpAgent,
-  BlobRef,
-  ChatBskyGroupDefs,
-  type ComAtprotoLabelDefs,
-  type ComAtprotoRepoApplyWrites,
-  type ComAtprotoRepoStrongRef,
-} from '@atproto/api'
+import {type AtpAgent, ChatBskyGroupDefs} from '@atproto/api'
 import {TID} from '@atproto/common-web'
-import {type Client} from '@atproto/lex'
-import {toDatetimeString} from '@atproto/syntax'
+import {type $Typed, type Client} from '@atproto/lex'
+import {
+  type AtUriString,
+  toDatetimeString,
+  type UriString,
+} from '@atproto/syntax'
 import {RichText} from '@bsky.app/sdk/richtext'
 import {t} from '@lingui/core/macro'
 import {type QueryClient} from '@tanstack/react-query'
-import {sha256} from 'js-sha256'
-import {CID} from 'multiformats/cid'
-import * as Hasher from 'multiformats/hashes/hasher'
 
 import {IMAGE_SIZE_CONFIG_POSTS} from '#/lib/constants'
 import {isNetworkError} from '#/lib/strings/errors'
@@ -42,8 +28,11 @@ import {
   type PostDraft,
   type ThreadDraft,
 } from '#/view/com/composer/state/composer'
+import {app, com} from '#/lexicons'
 import * as bsky from '#/types/bsky'
 import {createGIFDescription} from '../gif-alt-text'
+import {computeCid} from './computeCid'
+import {fromLegacyBlobRef} from './legacy-blob'
 import {uploadBlob} from './upload-blob'
 
 export {uploadBlob}
@@ -54,13 +43,23 @@ interface PostOpts {
   onStateChange?: (state: string) => void
   langs?: string[]
   /*
-   * Facet/mention resolution is an appview job - it resolves handles through
-   * the appview, and the public fallback keeps it working when logged out.
-   * The rest of this pipeline still writes through the agent.
+   * Facet/mention resolution and reply-root lookup are appview jobs - they
+   * resolve handles and read posts through the appview, and the public
+   * fallback keeps facet detection working when logged out.
    */
   appviewClient: Client
+  /*
+   * The repo write itself (applyWrites) plus every record blob (images,
+   * gallery items, link thumbnails, video captions) goes to the account's own
+   * PDS, never the appview.
+   */
+  pdsClient: Client
 }
 
+/**
+ * The `agent` is only still here for the link/gif resolvers in `resolve.ts`,
+ * which read through it; drop the parameter when those move to the clients.
+ */
 export async function post(
   agent: AtpAgent,
   queryClient: QueryClient,
@@ -70,12 +69,12 @@ export async function post(
   opts.onStateChange?.(t`Processing...`)
 
   let replyPromise:
-    | Promise<AppBskyFeedPost.Record['reply']>
-    | AppBskyFeedPost.Record['reply']
+    | Promise<app.bsky.feed.post.Main['reply']>
+    | app.bsky.feed.post.Main['reply']
     | undefined
   if (opts.replyTo) {
     // Not awaited to avoid waterfalls.
-    replyPromise = resolveReply(agent, opts.replyTo)
+    replyPromise = resolveReply(opts.appviewClient, opts.replyTo)
   }
 
   // add top 3 languages from user preferences if langs is provided
@@ -84,8 +83,8 @@ export async function post(
     langs = opts.langs.slice(0, 3)
   }
 
-  const did = agent.assertDid
-  const writes: $Typed<ComAtprotoRepoApplyWrites.Create>[] = []
+  const did = opts.pdsClient.assertDid
+  const writes: com.atproto.repo.applyWrites.$InputBody['writes'] = []
   const uris: string[] = []
 
   let now = new Date()
@@ -98,11 +97,12 @@ export async function post(
     const rtPromise = resolveRT(opts.appviewClient, draft.richtext)
     const embedPromise = resolveEmbed(
       agent,
+      opts.pdsClient,
       queryClient,
       draft,
       opts.onStateChange,
     )
-    let labels: $Typed<ComAtprotoLabelDefs.SelfLabels> | undefined
+    let labels: $Typed<com.atproto.label.defs.SelfLabels> | undefined
     if (draft.labels.length) {
       labels = {
         $type: 'com.atproto.label.defs#selfLabels',
@@ -115,17 +115,17 @@ export async function post(
     now.setMilliseconds(now.getMilliseconds() + 1)
     tid = TID.next(tid)
     const rkey = tid.toString()
-    const uri = `at://${did}/app.bsky.feed.post/${rkey}`
+    const uri = `at://${did}/app.bsky.feed.post/${rkey}` as AtUriString
     uris.push(uri)
 
     const rt = await rtPromise
     const embed = await embedPromise
     const reply = await replyPromise
-    const record: AppBskyFeedPost.Record = {
+    const record: app.bsky.feed.post.Main = {
       // IMPORTANT: $type has to exist, CID is calculated with the `$type` field
       // present and will produce the wrong CID if you omit it.
       $type: 'app.bsky.feed.post',
-      createdAt: now.toISOString(),
+      createdAt: toDatetimeString(now),
       text: rt.text,
       facets: rt.facets,
       reply,
@@ -164,7 +164,7 @@ export async function post(
         value: {
           ...thread.postgate,
           $type: 'app.bsky.feed.postgate',
-          createdAt: now.toISOString(),
+          createdAt: toDatetimeString(now),
           post: uri,
         },
       })
@@ -182,8 +182,8 @@ export async function post(
   }
 
   try {
-    await agent.com.atproto.repo.applyWrites({
-      repo: agent.assertDid,
+    await opts.pdsClient.call(com.atproto.repo.applyWrites, {
+      repo: did,
       writes: writes,
       validate: true,
     })
@@ -224,9 +224,9 @@ export class ReplyDeletedError extends Error {
   }
 }
 
-async function resolveReply(agent: AtpAgent, replyTo: string) {
-  const {data} = await agent.app.bsky.feed.getPosts({
-    uris: [replyTo],
+async function resolveReply(appviewClient: Client, replyTo: string) {
+  const data = await appviewClient.call(app.bsky.feed.getPosts, {
+    uris: [replyTo as AtUriString],
   })
   const parentPost = data.posts[0]
   if (!parentPost) {
@@ -237,14 +237,9 @@ async function resolveReply(agent: AtpAgent, replyTo: string) {
     uri: parentPost.uri,
     cid: parentPost.cid,
   }
-  let rootRef = parentRef
+  let rootRef: com.atproto.repo.strongRef.Main = parentRef
 
-  if (
-    bsky.dangerousIsType<AppBskyFeedPost.Record>(
-      parentPost.record,
-      AppBskyFeedPost.isRecord,
-    )
-  ) {
+  if (bsky.isType(app.bsky.feed.post, parentPost.record)) {
     if (parentPost.record.reply) {
       rootRef = parentPost.record.reply.root
     }
@@ -258,21 +253,14 @@ async function resolveReply(agent: AtpAgent, replyTo: string) {
 
 async function resolveEmbed(
   agent: AtpAgent,
+  pdsClient: Client,
   queryClient: QueryClient,
   draft: PostDraft,
   onStateChange: ((state: string) => void) | undefined,
-): Promise<
-  | $Typed<AppBskyEmbedImages.Main>
-  | $Typed<AppBskyEmbedGallery.Main>
-  | $Typed<AppBskyEmbedVideo.Main>
-  | $Typed<AppBskyEmbedExternal.Main>
-  | $Typed<AppBskyEmbedRecord.Main>
-  | $Typed<AppBskyEmbedRecordWithMedia.Main>
-  | undefined
-> {
+): Promise<app.bsky.feed.post.Main['embed']> {
   if (draft.embed.quote) {
     const [resolvedMedia, resolvedQuote] = await Promise.all([
-      resolveMedia(agent, queryClient, draft.embed, onStateChange),
+      resolveMedia(agent, pdsClient, queryClient, draft.embed, onStateChange),
       resolveRecord(agent, queryClient, draft.embed.quote.uri),
     ])
     if (resolvedMedia) {
@@ -292,6 +280,7 @@ async function resolveEmbed(
   }
   const resolvedMedia = await resolveMedia(
     agent,
+    pdsClient,
     queryClient,
     draft.embed,
     onStateChange,
@@ -308,7 +297,12 @@ async function resolveEmbed(
     if (resolvedLink.type === 'record') {
       return {
         $type: 'app.bsky.embed.record',
-        record: resolvedLink.record,
+        /*
+         * `resolve.ts` is still legacy-typed - its strong refs and URIs carry
+         * unbranded strings. Assert at the boundary until it moves to the
+         * clients.
+         */
+        record: resolvedLink.record as com.atproto.repo.strongRef.Main,
       }
     }
   }
@@ -317,14 +311,15 @@ async function resolveEmbed(
 
 async function resolveMedia(
   agent: AtpAgent,
+  pdsClient: Client,
   queryClient: QueryClient,
   embedDraft: EmbedDraft,
   onStateChange: ((state: string) => void) | undefined,
 ): Promise<
-  | $Typed<AppBskyEmbedExternal.Main>
-  | $Typed<AppBskyEmbedImages.Main>
-  | $Typed<AppBskyEmbedGallery.Main>
-  | $Typed<AppBskyEmbedVideo.Main>
+  | $Typed<app.bsky.embed.external.Main>
+  | $Typed<app.bsky.embed.images.Main>
+  | $Typed<app.bsky.embed.gallery.Main>
+  | $Typed<app.bsky.embed.video.Main>
   | undefined
 > {
   if (embedDraft.media?.type === 'images') {
@@ -333,7 +328,7 @@ async function resolveMedia(
       count: imagesDraft.length,
     })
     onStateChange?.(t`Uploading images...`)
-    const images: AppBskyEmbedImages.Image[] = await Promise.all(
+    const images: app.bsky.embed.images.Image[] = await Promise.all(
       imagesDraft.map(async (image, i) => {
         logger.debug(`Compressing image #${i}`)
         const {path, width, height, mime} = await compressImage(
@@ -341,9 +336,9 @@ async function resolveMedia(
           IMAGE_SIZE_CONFIG_POSTS,
         )
         logger.debug(`Uploading image #${i}`)
-        const res = await uploadBlob(agent, path, mime)
+        const res = await uploadBlob(pdsClient, path, mime)
         return {
-          image: res.data.blob,
+          image: res.blob,
           alt: image.alt,
           aspectRatio: {width, height},
         }
@@ -360,7 +355,7 @@ async function resolveMedia(
       count: imagesDraft.length,
     })
     onStateChange?.(t`Uploading images...`)
-    const items: $Typed<AppBskyEmbedGallery.Image>[] = await Promise.all(
+    const items: $Typed<app.bsky.embed.gallery.Image>[] = await Promise.all(
       imagesDraft.map(async (image, i) => {
         logger.debug(`Compressing image #${i}`)
         const {path, width, height, mime} = await compressImage(
@@ -368,10 +363,10 @@ async function resolveMedia(
           IMAGE_SIZE_CONFIG_POSTS,
         )
         logger.debug(`Uploading image #${i}`)
-        const res = await uploadBlob(agent, path, mime)
+        const res = await uploadBlob(pdsClient, path, mime)
         return {
           $type: 'app.bsky.embed.gallery#image' as const,
-          image: res.data.blob,
+          image: res.blob,
           alt: image.alt,
           aspectRatio: {width, height},
         }
@@ -391,10 +386,10 @@ async function resolveMedia(
       videoDraft.captions
         .filter(caption => caption.lang !== '')
         .map(async caption => {
-          const {data} = await agent.uploadBlob(caption.file, {
+          const res = await pdsClient.uploadBlob(caption.file, {
             encoding: 'text/vtt',
           })
-          return {lang: caption.lang, file: data.blob}
+          return {lang: caption.lang, file: res.body.blob}
         }),
     )
 
@@ -414,7 +409,11 @@ async function resolveMedia(
 
     return {
       $type: 'app.bsky.embed.video',
-      video: videoDraft.pendingPublish.blobRef,
+      /*
+       * The video pipeline still reads its blob off the legacy agent, so
+       * normalize it to the lex shape before it reaches the lex write.
+       */
+      video: fromLegacyBlobRef(videoDraft.pendingPublish.blobRef),
       alt: videoDraft.altText || undefined,
       captions: captions.length === 0 ? undefined : captions,
       aspectRatio,
@@ -429,17 +428,17 @@ async function resolveMedia(
       agent,
       gifDraft.gif,
     )
-    let blob: BlobRef | undefined
+    let blob: app.bsky.embed.external.External['thumb']
     if (resolvedGif.thumb) {
       onStateChange?.(t`Uploading link thumbnail...`)
       const {path, mime} = resolvedGif.thumb.source
-      const response = await uploadBlob(agent, path, mime)
-      blob = response.data.blob
+      const response = await uploadBlob(pdsClient, path, mime)
+      blob = response.blob
     }
     return {
       $type: 'app.bsky.embed.external',
       external: {
-        uri: resolvedGif.uri,
+        uri: resolvedGif.uri as UriString,
         title: resolvedGif.title,
         description: createGIFDescription(resolvedGif.title, gifDraft.alt),
         thumb: blob,
@@ -453,21 +452,23 @@ async function resolveMedia(
       embedDraft.link.uri,
     )
     if (resolvedLink.type === 'external') {
-      let blob: BlobRef | undefined
+      let blob: app.bsky.embed.external.External['thumb']
       if (resolvedLink.thumb) {
         onStateChange?.(t`Uploading link thumbnail...`)
         const {path, mime} = resolvedLink.thumb.source
-        const response = await uploadBlob(agent, path, mime)
-        blob = response.data.blob
+        const response = await uploadBlob(pdsClient, path, mime)
+        blob = response.blob
       }
       return {
         $type: 'app.bsky.embed.external',
         external: {
-          uri: resolvedLink.uri,
+          uri: resolvedLink.uri as UriString,
           title: resolvedLink.title,
           description: resolvedLink.description,
           thumb: blob,
-          associatedRefs: resolvedLink.associatedRefs,
+          associatedRefs: resolvedLink.associatedRefs as
+            | com.atproto.repo.strongRef.Main[]
+            | undefined,
         },
       }
     }
@@ -478,7 +479,7 @@ async function resolveMedia(
       return {
         $type: 'app.bsky.embed.external',
         external: {
-          uri: resolvedLink.uri,
+          uri: resolvedLink.uri as UriString,
           title: resolvedLink.view.name,
           description: `${resolvedLink.view.memberCount}/${resolvedLink.view.memberLimit}`,
         },
@@ -488,101 +489,18 @@ async function resolveMedia(
   return undefined
 }
 
+/*
+ * `resolve.ts` still resolves through the agent and returns legacy-typed refs;
+ * assert at the boundary until it moves to the clients.
+ */
 async function resolveRecord(
   agent: AtpAgent,
   queryClient: QueryClient,
   uri: string,
-): Promise<ComAtprotoRepoStrongRef.Main> {
+): Promise<com.atproto.repo.strongRef.Main> {
   const resolvedLink = await fetchResolveLinkQuery(queryClient, agent, uri)
   if (resolvedLink.type !== 'record') {
     throw Error(t`Expected uri to resolve to a record`)
   }
-  return resolvedLink.record
-}
-
-// The built-in hashing functions from multiformats (`multiformats/hashes/sha2`)
-// are meant for Node.js, this is the cross-platform equivalent.
-const mf_sha256 = Hasher.from({
-  name: 'sha2-256',
-  code: 0x12,
-  encode: input => {
-    const digest = sha256.arrayBuffer(input)
-    return new Uint8Array(digest)
-  },
-})
-
-async function computeCid(record: AppBskyFeedPost.Record): Promise<string> {
-  /*
-   * Lazily loaded since it's only needed when posting a thread, and its
-   * `cborg` dependency is ~190KB that would otherwise be in the initial
-   * web bundle.
-   */
-  const dcbor = await import('@ipld/dag-cbor')
-  // IMPORTANT: `prepareObject` prepares the record to be hashed by removing
-  // fields with undefined value, and converting BlobRef instances to the
-  // right IPLD representation.
-  const prepared = prepareForHashing(record)
-  // 1. Encode the record into DAG-CBOR format
-  const encoded = dcbor.encode(prepared)
-  // 2. Hash the record in SHA-256 (code 0x12)
-  const digest = await mf_sha256.digest(encoded)
-  // 3. Create a CIDv1, specifying DAG-CBOR as content (code 0x71)
-  const cid = CID.createV1(0x71, digest)
-  // 4. Get the Base32 representation of the CID (`b` prefix)
-  return cid.toString()
-}
-
-// Returns a transformed version of the object for use in DAG-CBOR.
-// eslint-disable-next-line @typescript-eslint/no-explicit-any
-function prepareForHashing(v: any): any {
-  // IMPORTANT: BlobRef#ipld() returns the correct object we need for hashing,
-  // the API client will convert this for you but we're hashing in the client,
-  // so we need it *now*.
-  if (v instanceof BlobRef) {
-    return v.ipld()
-  }
-
-  // Walk through arrays
-  if (Array.isArray(v)) {
-    let pure = true
-    const mapped = v.map(value => {
-      if (value !== (value = prepareForHashing(value))) {
-        pure = false
-      }
-      return value
-    })
-    return pure ? v : mapped
-  }
-
-  // Walk through plain objects
-  if (isPlainObject(v)) {
-    const obj: Record<string, unknown> = {}
-    let pure = true
-    for (const key in v) {
-      // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
-      let value = v[key]
-      // `value` is undefined
-      if (value === undefined) {
-        pure = false
-        continue
-      }
-      // `prepareObject` returned a value that's different from what we had before
-      if (value !== (value = prepareForHashing(value))) {
-        pure = false
-      }
-      obj[key] = value
-    }
-    // Return as is if we haven't needed to tamper with anything
-    return pure ? v : obj
-  }
-  return v
-}
-
-// eslint-disable-next-line @typescript-eslint/no-explicit-any
-function isPlainObject(v: any): boolean {
-  if (typeof v !== 'object' || v === null) {
-    return false
-  }
-  const proto = Object.getPrototypeOf(v)
-  return proto === Object.prototype || proto === null
+  return resolvedLink.record as com.atproto.repo.strongRef.Main
 }