migrate the post pipeline write and reply lookup to the clients

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/lib/api/index.ts

@@ -1,22 +1,11 @@
-import {
-  type $Typed,
-  type AppBskyEmbedExternal,
-  type AppBskyEmbedGallery,
-  type AppBskyEmbedImages,
-  type AppBskyEmbedRecord,
-  type AppBskyEmbedRecordWithMedia,
-  type AppBskyEmbedVideo,
-  AppBskyFeedPost,
-  type AtpAgent,
-  type 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'
@@ -39,11 +28,11 @@ import {
   type PostDraft,
   type ThreadDraft,
 } from '#/view/com/composer/state/composer'
-import {type app} from '#/lexicons'
+import {app, com} from '#/lexicons'
 import * as bsky from '#/types/bsky'
 import {createGIFDescription} from '../gif-alt-text'
 import {computeCid} from './computeCid'
-import {toLegacyBlobRef} from './legacy-blob'
+import {fromLegacyBlobRef} from './legacy-blob'
 import {uploadBlob} from './upload-blob'
 
 export {uploadBlob}
@@ -54,17 +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.
+   * 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
   /*
-   * Record blobs (images, gallery items, link thumbnails, video captions)
-   * upload to the account's own PDS, never the appview.
+   * 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,
@@ -74,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
@@ -88,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()
@@ -107,7 +102,7 @@ export async function post(
       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',
@@ -120,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,
@@ -169,7 +164,7 @@ export async function post(
         value: {
           ...thread.postgate,
           $type: 'app.bsky.feed.postgate',
-          createdAt: now.toISOString(),
+          createdAt: toDatetimeString(now),
           post: uri,
         },
       })
@@ -177,13 +172,7 @@ export async function post(
 
     // Prepare a ref to the current post for the next post in the thread.
     const ref = {
-      /*
-       * `computeCid` is typed against the lex record. The pipeline still builds
-       * the legacy `AppBskyFeedPost.Record`; the two shapes are structurally
-       * equivalent for hashing, so bridge them until the pipeline itself moves
-       * to the lex record type.
-       */
-      cid: await computeCid(record as unknown as app.bsky.feed.post.Main),
+      cid: await computeCid(record),
       uri,
     }
     replyPromise = {
@@ -193,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,
     })
@@ -235,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) {
@@ -248,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
     }
@@ -273,15 +257,7 @@ async function resolveEmbed(
   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, pdsClient, queryClient, draft.embed, onStateChange),
@@ -321,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,
       }
     }
   }
@@ -335,10 +316,10 @@ async function resolveMedia(
   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') {
@@ -347,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(
@@ -357,7 +338,7 @@ async function resolveMedia(
         logger.debug(`Uploading image #${i}`)
         const res = await uploadBlob(pdsClient, path, mime)
         return {
-          image: toLegacyBlobRef(res.blob),
+          image: res.blob,
           alt: image.alt,
           aspectRatio: {width, height},
         }
@@ -374,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(
@@ -385,7 +366,7 @@ async function resolveMedia(
         const res = await uploadBlob(pdsClient, path, mime)
         return {
           $type: 'app.bsky.embed.gallery#image' as const,
-          image: toLegacyBlobRef(res.blob),
+          image: res.blob,
           alt: image.alt,
           aspectRatio: {width, height},
         }
@@ -405,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}
         }),
     )
 
@@ -428,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,
@@ -443,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(pdsClient, path, mime)
-      blob = toLegacyBlobRef(response.blob)
+      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,
@@ -467,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(pdsClient, path, mime)
-        blob = toLegacyBlobRef(response.blob)
+        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,
         },
       }
     }
@@ -492,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}`,
         },
@@ -502,14 +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
+  return resolvedLink.record as com.atproto.repo.strongRef.Main
 }

src/lib/api/legacy-blob.ts

@@ -14,3 +14,18 @@ import {type BlobRef as LexBlobRef} from '@atproto/lex'
 export function toLegacyBlobRef(blob: LexBlobRef): BlobRef {
   return BlobRef.fromJsonRef(blob as Parameters<typeof BlobRef.fromJsonRef>[0])
 }
+
+/**
+ * Normalize a legacy `BlobRef` class instance to the plain-JSON lex blob shape.
+ *
+ * Required for any blob that reaches a lex write: the lex serializer walks
+ * plain objects, so a class instance goes on the wire with its internal
+ * `original` field and no `$type`. `ipld()` yields exactly the lex shape, and
+ * hashes identically (see `computeCid.test.ts` case 2b).
+ *
+ * Only the video pipeline still needs this - it reads its blob off the legacy
+ * agent (`app.bsky.video.getJobStatus`). Drop it when the video client moves.
+ */
+export function fromLegacyBlobRef(blob: BlobRef): LexBlobRef {
+  return blob.ipld()
+}