Fix content sniffing on streams (bluesky-social#570)

fix content sniffing on streamsx

packages/pds/src/services/repo/blobs.ts

@@ -16,15 +16,15 @@ export class RepoBlobs {
   constructor(public db: Database, public blobstore: BlobStore) {}
 
   async addUntetheredBlob(
-    mimeType: string,
+    userSuggestedMime: string,
     blobStream: stream.Readable,
   ): Promise<CID> {
-    const [tempKey, size, sha256, imgInfo, fileType] = await Promise.all([
+    const [tempKey, size, sha256, imgInfo, sniffedMime] = await Promise.all([
       this.blobstore.putTemp(cloneStream(blobStream)),
       streamSize(cloneStream(blobStream)),
       sha256Stream(cloneStream(blobStream)),
       img.maybeGetInfo(cloneStream(blobStream)),
-      fileTypeFromStream(blobStream),
+      mimeTypeFromStream(cloneStream(blobStream)),
     ])
 
     const cid = sha256RawToCid(sha256)
@@ -33,7 +33,7 @@ export class RepoBlobs {
       .insertInto('blob')
       .values({
         cid: cid.toString(),
-        mimeType: fileType?.mime || mimeType,
+        mimeType: sniffedMime || userSuggestedMime,
         size,
         tempKey,
         width: imgInfo?.width || null,
@@ -151,6 +151,14 @@ export class CidNotFound extends Error {
   }
 }
 
+async function mimeTypeFromStream(
+  blobStream: stream.Readable,
+): Promise<string | undefined> {
+  const fileType = await fileTypeFromStream(blobStream)
+  blobStream.destroy()
+  return fileType?.mime
+}
+
 function acceptedMime(mime: string, accepted: string[]): boolean {
   if (accepted.includes('*/*')) return true
   const globs = accepted.filter((a) => a.endsWith('/*'))

packages/pds/tests/file-uploads.test.ts

@@ -254,6 +254,23 @@ describe('file uploads', () => {
     expect(found?.height).toBe(742)
   })
 
+  it('handles pngs', async () => {
+    const file = await fs.readFile('tests/image/fixtures/at.png')
+    const res = await aliceAgent.api.com.atproto.blob.upload(file, {
+      encoding: 'image/png',
+    })
+
+    const found = await db.db
+      .selectFrom('blob')
+      .selectAll()
+      .where('cid', '=', res.data.cid.toString())
+      .executeTakeFirst()
+
+    expect(found?.mimeType).toBe('image/png')
+    expect(found?.width).toBe(554)
+    expect(found?.height).toBe(532)
+  })
+
   it('handles unknown mimetypes', async () => {
     const file = await randomBytes(20000)
     const res = await aliceAgent.api.com.atproto.blob.upload(file, {