Merge branch 'main' of github.com:skywatch-bsky/skywatch-tail

.env.example

···
 PDS=bsky.social
 WSS_URL=wss://your-labeler-service.com/xrpc/com.atproto.label.subscribeLabels
 
+# PLC Directory (for DID resolution)
+PLC_ENDPOINT=https://plc.wtf
+
 # Blob & Image Handling
 HYDRATE_BLOBS=false # Set to true to download images/videos
 BLOB_STORAGE_TYPE=local # 'local' or 's3'

docs/profile-blob-hydration.md

···
+# Profile Blob Hydration - Implementation Notes
+
+## Overview
+
+This document captures key learnings from implementing avatar and banner blob hydration for Bluesky profiles.
+
+## Key Discoveries
+
+### 1. CID Deserialization in @atproto/api
+
+The `@atproto/api` library deserializes blob references from their JSON `$link` representation into CID class objects.
+
+**Raw JSON from API:**
+```json
+{
+  "avatar": {
+    "$type": "blob",
+    "ref": {
+      "$link": "bafkreigg3s6plegjncmxubeufbohj3qasbm4r23q2x7zlivdhccfqfypve"
+    },
+    "mimeType": "image/jpeg",
+    "size": 101770
+  }
+}
+```
+
+**What you get in TypeScript:**
+```typescript
+record.avatar.ref // CID object with { code, version, hash, ... }
+```
+
+**Solution:**
+```typescript
+const cid = record.avatar.ref.toString(); // "bafkrei..."
+```
+
+### 2. PDS Endpoint Resolution
+
+Users can be on different Personal Data Servers (PDS), not just `bsky.social`. Blobs must be fetched from the user's actual PDS.
+
+**Process:**
+1. Query PLC directory for DID document: `https://plc.wtf/${did}`
+2. Find service with `id: "#atproto_pds"` and `type: "AtprotoPersonalDataServer"`
+3. Extract `serviceEndpoint` URL
+4. Use that endpoint for `com.atproto.sync.getBlob`
+
+**Example:**
+```typescript
+const didDoc = await fetch(`https://plc.wtf/${did}`).then(r => r.json());
+const pdsService = didDoc.service?.find(s =>
+  s.id === "#atproto_pds" && s.type === "AtprotoPersonalDataServer"
+);
+const pdsEndpoint = pdsService.serviceEndpoint; // e.g., "https://waxcap.us-west.host.bsky.network"
+```
+
+### 3. Correct Blob Fetching
+
+**Don't use CDN paths** - they don't work reliably for all blobs and require authentication context.
+
+**Use the AT Protocol API:**
+```typescript
+const blobUrl = `${pdsEndpoint}/xrpc/com.atproto.sync.getBlob?did=${did}&cid=${cid}`;
+const response = await fetch(blobUrl);
+const blobData = Buffer.from(await response.arrayBuffer());
+```
+
+### 4. Database Schema Design
+
+**Separate tables for different blob types:**
+
+- `blobs` table: Post images with FK to `posts(uri)`
+- `profile_blobs` table: Avatars/banners with FK to `profiles(did)`
+
+This allows proper relational queries and analysis.
+
+**Profile blobs schema:**
+```sql
+CREATE TABLE profile_blobs (
+  did TEXT NOT NULL,
+  blob_type TEXT NOT NULL CHECK (blob_type IN ('avatar', 'banner')),
+  blob_cid TEXT NOT NULL,
+  sha256 TEXT NOT NULL,
+  phash TEXT,
+  storage_path TEXT,
+  mimetype TEXT,
+  captured_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (did, blob_type, captured_at),
+  FOREIGN KEY (did) REFERENCES profiles(did)
+);
+```
+
+### 5. Change Tracking
+
+Including `captured_at` in the primary key allows tracking when users change their avatars/banners.
+
+**Query latest state:**
+```sql
+SELECT * FROM profile_blobs
+WHERE did = ? AND blob_type = ?
+ORDER BY captured_at DESC
+LIMIT 1
+```
+
+**Only insert if changed:**
+```typescript
+const latest = await findLatestByDidAndType(did, type);
+if (latest && latest.blob_cid === cid) {
+  return; // No change, skip
+}
+// Insert new row with current timestamp
+```
+
+### 6. Sentinel Values for Missing Data
+
+Use empty string (`""`) to distinguish "we checked, user has no avatar" from NULL "we haven't checked yet".
+
+```typescript
+if (record.avatar?.ref) {
+  avatarCid = record.avatar.ref.toString();
+} else {
+  avatarCid = ""; // Explicitly checked, not present
+}
+```
+
+This prevents infinite re-hydration loops for profiles without avatars.
+
+### 7. Profile Re-hydration Logic
+
+```typescript
+const existingProfile = await findByDid(did);
+const needsRehydration = existingProfile &&
+  (existingProfile.avatar_cid === null || existingProfile.banner_cid === null);
+
+if (existingProfile && !needsRehydration) {
+  return; // Skip
+}
+```
+
+## Configuration
+
+- `PLC_ENDPOINT`: DID resolution endpoint (default: `https://plc.wtf`)
+  - Can be changed to `https://plc.directory` or custom instance
+  - plc.wtf is faster but unofficial
+
+## Common Errors
+
+### "RepoNotFound"
+- **Cause:** Querying wrong PDS endpoint
+- **Solution:** Resolve correct PDS from DID document
+
+### Foreign Key Constraint Violation
+- **Cause:** Trying to insert profile blobs into `blobs` table
+- **Solution:** Use separate `profile_blobs` table
+
+### Missing CIDs Despite API Returning Them
+- **Cause:** Trying to access `ref.$link` when ref is a CID object
+- **Solution:** Call `.toString()` on the CID object
+
+## Related Files
+
+- `src/hydration/profiles.service.ts` - Main hydration logic
+- `src/database/profile-blobs.repository.ts` - Profile blob persistence
+- `src/database/schema.ts` - Table definitions
+- `src/config/index.ts` - PLC endpoint configuration

src/blobs/processor.ts

···
     }
   }
 
+  private parseBlobUri(uri: string): { did: string; type: 'post' | 'avatar' | 'banner' } {
+    if (uri.startsWith("profile://")) {
+      const match = uri.match(/^profile:\/\/([^/]+)\/(avatar|banner)$/);
+      if (match) {
+        return { did: match[1], type: match[2] as 'avatar' | 'banner' };
+      }
+    }
+
+    const [, did] = uri.replace("at://", "").split("/");
+    return { did, type: 'post' };
+  }
+
   private async processBlob(
     postUri: string,
     ref: BlobReference
   ): Promise<void> {
-    const existing = await this.blobsRepo.findBySha256(ref.cid);
+    const existing = await this.blobsRepo.findByCid(ref.cid);
     if (existing) {
+      await this.blobsRepo.insert({
+        post_uri: postUri,
+        blob_cid: ref.cid,
+        sha256: existing.sha256,
+        phash: existing.phash,
+        storage_path: existing.storage_path,
+        mimetype: existing.mimetype,
+      });
       logger.debug(
         { postUri, cid: ref.cid },
-        "Blob already processed, skipping"
+        "Blob already processed, reusing hashes"
       );
       return;
     }
 
-    const [, did] = postUri.replace("at://", "").split("/");
+    const { did, type } = this.parseBlobUri(postUri);
+    const pds = `https://${config.bsky.pds}`;
+    const blobUrl = `${pds}/xrpc/com.atproto.sync.getBlob?did=${did}&cid=${ref.cid}`;
 
     try {
-      const response = await fetch(
-        `https://cdn.bsky.app/img/feed_thumbnail/plain/${did}/${ref.cid}@jpeg`,
-        { method: "HEAD" }
-      );
+      const response = await fetch(blobUrl);
 
       if (!response.ok) {
         logger.warn(
-          { postUri, cid: ref.cid, status: response.status },
-          "Failed to fetch blob metadata"
+          { postUri, cid: ref.cid, status: response.status, did },
+          "Failed to fetch blob"
         );
         return;
       }
 
-      let blobData: Buffer | null = null;
-      let storagePath: string | undefined;
+      const blobData = Buffer.from(await response.arrayBuffer());
 
+      let storagePath: string | undefined;
       if (this.storage && config.blobs.hydrateBlobs) {
-        const fullResponse = await fetch(
-          `https://cdn.bsky.app/img/feed_fullsize/plain/${did}/${ref.cid}@jpeg`
+        storagePath = await this.storage.store(
+          ref.cid,
+          blobData,
+          ref.mimeType
         );
-
-        if (fullResponse.ok) {
-          blobData = Buffer.from(
-            await fullResponse.arrayBuffer()
-          );
-          storagePath = await this.storage.store(
-            ref.cid,
-            blobData,
-            ref.mimeType
-          );
-        }
-      } else {
-        const thumbResponse = await fetch(
-          `https://cdn.bsky.app/img/feed_thumbnail/plain/${did}/${ref.cid}@jpeg`
-        );
-
-        if (thumbResponse.ok) {
-          blobData = Buffer.from(
-            await thumbResponse.arrayBuffer()
-          );
-        }
-      }
-
-      if (!blobData) {
-        logger.warn(
-          { postUri, cid: ref.cid },
-          "Could not fetch blob data"
-        );
-        return;
       }
 
       const hashes = await computeBlobHashes(
···
       });
 
       logger.info(
-        { postUri, cid: ref.cid, sha256: hashes.sha256 },
+        { postUri, cid: ref.cid, sha256: hashes.sha256, type },
         "Blob processed successfully"
       );
     } catch (error) {

src/config/index.ts

···
     password: z.string().min(1, "BSKY_PASSWORD is required"),
     pds: z.string().default("bsky.social"),
   }),
+  plc: z.object({
+    endpoint: z.string().url().default("https://plc.wtf"),
+  }),
   labeler: z.object({
     wssUrl: z.string().url("WSS_URL must be a valid URL"),
   }),
···
       handle: process.env.BSKY_HANDLE,
       password: process.env.BSKY_PASSWORD,
       pds: process.env.PDS,
+    },
+    plc: {
+      endpoint: process.env.PLC_ENDPOINT,
     },
     labeler: {
       wssUrl: process.env.WSS_URL,

src/database/blobs.repository.ts

···
       );
     });
   }
+
+  async findByCid(cid: string): Promise<Blob | null> {
+    return new Promise((resolve, reject) => {
+      this.db.all(
+        `SELECT * FROM blobs WHERE blob_cid = $1 LIMIT 1`,
+        cid,
+        (err, rows: Blob[]) => {
+          if (err) {
+            logger.error({ err, cid }, "Failed to find blob by CID");
+            reject(err);
+            return;
+          }
+          resolve(rows?.[0] || null);
+        }
+      );
+    });
+  }
 }

src/database/profile-blobs.repository.ts

···
+import { Database } from "duckdb";
+import { logger } from "../logger/index.js";
+
+export interface ProfileBlob {
+  did: string;
+  blob_type: "avatar" | "banner";
+  blob_cid: string;
+  sha256: string;
+  phash?: string;
+  storage_path?: string;
+  mimetype?: string;
+  captured_at?: Date;
+}
+
+export class ProfileBlobsRepository {
+  constructor(private db: Database) {}
+
+  async insert(blob: ProfileBlob): Promise<void> {
+    return new Promise((resolve, reject) => {
+      this.db.prepare(
+        `
+        INSERT INTO profile_blobs (did, blob_type, blob_cid, sha256, phash, storage_path, mimetype, captured_at)
+        VALUES ($1, $2, $3, $4, $5, $6, $7, COALESCE($8, CURRENT_TIMESTAMP))
+      `,
+        (err, stmt) => {
+          if (err) {
+            logger.error({ err }, "Failed to prepare profile blob insert statement");
+            reject(err);
+            return;
+          }
+
+          stmt.run(
+            blob.did,
+            blob.blob_type,
+            blob.blob_cid,
+            blob.sha256,
+            blob.phash || null,
+            blob.storage_path || null,
+            blob.mimetype || null,
+            blob.captured_at || null,
+            (err) => {
+              if (err) {
+                logger.error({ err, blob }, "Failed to insert profile blob");
+                reject(err);
+                return;
+              }
+              resolve();
+            }
+          );
+        }
+      );
+    });
+  }
+
+  async findByDid(did: string): Promise<ProfileBlob[]> {
+    return new Promise((resolve, reject) => {
+      this.db.all(
+        `SELECT * FROM profile_blobs WHERE did = $1 ORDER BY captured_at DESC`,
+        did,
+        (err, rows: ProfileBlob[]) => {
+          if (err) {
+            logger.error({ err, did }, "Failed to find profile blobs by DID");
+            reject(err);
+            return;
+          }
+          resolve(rows || []);
+        }
+      );
+    });
+  }
+
+  async findLatestByDidAndType(did: string, blobType: "avatar" | "banner"): Promise<ProfileBlob | null> {
+    return new Promise((resolve, reject) => {
+      this.db.all(
+        `SELECT * FROM profile_blobs WHERE did = $1 AND blob_type = $2 ORDER BY captured_at DESC LIMIT 1`,
+        did,
+        blobType,
+        (err, rows: ProfileBlob[]) => {
+          if (err) {
+            logger.error({ err, did, blobType }, "Failed to find latest profile blob");
+            reject(err);
+            return;
+          }
+          resolve(rows?.[0] || null);
+        }
+      );
+    });
+  }
+
+  async findBySha256(sha256: string): Promise<ProfileBlob | null> {
+    return new Promise((resolve, reject) => {
+      this.db.all(
+        `SELECT * FROM profile_blobs WHERE sha256 = $1 LIMIT 1`,
+        sha256,
+        (err, rows: ProfileBlob[]) => {
+          if (err) {
+            logger.error({ err, sha256 }, "Failed to find profile blob by SHA256");
+            reject(err);
+            return;
+          }
+          resolve(rows?.[0] || null);
+        }
+      );
+    });
+  }
+
+  async findByPhash(phash: string): Promise<ProfileBlob[]> {
+    return new Promise((resolve, reject) => {
+      this.db.all(
+        `SELECT * FROM profile_blobs WHERE phash = $1`,
+        phash,
+        (err, rows: ProfileBlob[]) => {
+          if (err) {
+            logger.error({ err, phash }, "Failed to find profile blobs by pHash");
+            reject(err);
+            return;
+          }
+          resolve(rows || []);
+        }
+      );
+    });
+  }
+}

src/database/profiles.repository.ts

···
   handle?: string;
   display_name?: string;
   description?: string;
+  avatar_cid?: string;
+  banner_cid?: string;
 }
 
 export class ProfilesRepository {
···
     return new Promise((resolve, reject) => {
       this.db.prepare(
         `
-        INSERT INTO profiles (did, handle, display_name, description)
-        VALUES ($1, $2, $3, $4)
+        INSERT INTO profiles (did, handle, display_name, description, avatar_cid, banner_cid)
+        VALUES ($1, $2, $3, $4, $5, $6)
         ON CONFLICT (did) DO UPDATE SET
           handle = EXCLUDED.handle,
           display_name = EXCLUDED.display_name,
-          description = EXCLUDED.description
+          description = EXCLUDED.description,
+          avatar_cid = EXCLUDED.avatar_cid,
+          banner_cid = EXCLUDED.banner_cid
       `,
         (err, stmt) => {
           if (err) {
···
             profile.handle || null,
             profile.display_name || null,
             profile.description || null,
+            profile.avatar_cid || null,
+            profile.banner_cid || null,
             (err) => {
               if (err) {
                 logger.error({ err, profile }, "Failed to insert profile");

src/database/schema.ts

···
   did TEXT PRIMARY KEY,
   handle TEXT,
   display_name TEXT,
-  description TEXT
+  description TEXT,
+  avatar_cid TEXT,
+  banner_cid TEXT
 );
 
 -- Blobs table: stores information about image blobs found in posts
···
   FOREIGN KEY (post_uri) REFERENCES posts(uri)
 );
 
+-- Profile blobs table: stores avatar and banner blobs for profiles
+CREATE TABLE IF NOT EXISTS profile_blobs (
+  did TEXT NOT NULL,
+  blob_type TEXT NOT NULL CHECK (blob_type IN ('avatar', 'banner')),
+  blob_cid TEXT NOT NULL,
+  sha256 TEXT NOT NULL,
+  phash TEXT,
+  storage_path TEXT,
+  mimetype TEXT,
+  captured_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (did, blob_type, captured_at),
+  FOREIGN KEY (did) REFERENCES profiles(did)
+);
+
 -- Indexes for performance
 CREATE INDEX IF NOT EXISTS idx_labels_uri ON labels(uri);
 CREATE INDEX IF NOT EXISTS idx_labels_val ON labels(val);
 CREATE INDEX IF NOT EXISTS idx_labels_cts ON labels(cts);
 CREATE INDEX IF NOT EXISTS idx_posts_did ON posts(did);
+CREATE INDEX IF NOT EXISTS idx_blobs_cid ON blobs(blob_cid);
 CREATE INDEX IF NOT EXISTS idx_blobs_sha256 ON blobs(sha256);
 CREATE INDEX IF NOT EXISTS idx_blobs_phash ON blobs(phash);
+CREATE INDEX IF NOT EXISTS idx_profile_blobs_sha256 ON profile_blobs(sha256);
+CREATE INDEX IF NOT EXISTS idx_profile_blobs_phash ON profile_blobs(phash);
 `;
 
+async function migrateProfilesTable(): Promise<void> {
+  const db = getDatabase();
+
+  return new Promise((resolve, reject) => {
+    db.all(
+      "SELECT column_name FROM information_schema.columns WHERE table_name = 'profiles'",
+      (err, rows: any[]) => {
+        if (err) {
+          logger.error({ err }, "Failed to check profiles table columns");
+          reject(err);
+          return;
+        }
+
+        const columnNames = rows.map((row) => row.column_name);
+        const hasAvatarCid = columnNames.includes("avatar_cid");
+        const hasBannerCid = columnNames.includes("banner_cid");
+
+        if (!hasAvatarCid || !hasBannerCid) {
+          logger.info("Migrating profiles table to add avatar_cid and banner_cid columns");
+
+          const migrations: string[] = [];
+          if (!hasAvatarCid) {
+            migrations.push("ALTER TABLE profiles ADD COLUMN avatar_cid TEXT");
+          }
+          if (!hasBannerCid) {
+            migrations.push("ALTER TABLE profiles ADD COLUMN banner_cid TEXT");
+          }
+
+          db.exec(migrations.join("; "), (err) => {
+            if (err) {
+              logger.error({ err }, "Failed to migrate profiles table");
+              reject(err);
+              return;
+            }
+            logger.info("Profiles table migration completed");
+            resolve();
+          });
+        } else {
+          logger.debug("Profiles table already has avatar_cid and banner_cid columns");
+          resolve();
+        }
+      }
+    );
+  });
+}
+
 export async function initializeSchema(): Promise<void> {
   const db = getDatabase();
 
   return new Promise((resolve, reject) => {
-    db.exec(SCHEMA_SQL, (err) => {
+    db.exec(SCHEMA_SQL, async (err) => {
       if (err) {
         logger.error({ err }, "Failed to initialize schema");
         reject(err);
         return;
       }
       logger.info("Database schema initialized");
-      resolve();
+
+      try {
+        await migrateProfilesTable();
+        resolve();
+      } catch (migrationErr) {
+        reject(migrationErr);
+      }
     });
   });
 }

src/hydration/profiles.service.ts

···
 import { AtpAgent } from "@atproto/api";
 import { Database } from "duckdb";
 import { ProfilesRepository } from "../database/profiles.repository.js";
+import { ProfileBlobsRepository } from "../database/profile-blobs.repository.js";
+import { computeBlobHashes } from "../blobs/hasher.js";
+import { LocalBlobStorage } from "../blobs/storage/local.js";
+import { S3BlobStorage } from "../blobs/storage/s3.js";
+import { BlobStorage } from "../blobs/processor.js";
 import { pRateLimit } from "p-ratelimit";
 import { withRetry, isRateLimitError, isNetworkError, isServerError, isRecordNotFoundError } from "../utils/retry.js";
 import { logger } from "../logger/index.js";
···
 export class ProfileHydrationService {
   private agent: AtpAgent;
   private profilesRepo: ProfilesRepository;
+  private profileBlobsRepo: ProfileBlobsRepository;
+  private storage: BlobStorage | null = null;
   private limit: ReturnType<typeof pRateLimit>;
 
   constructor(db: Database) {
     this.agent = new AtpAgent({ service: `https://${config.bsky.pds}` });
     this.profilesRepo = new ProfilesRepository(db);
+    this.profileBlobsRepo = new ProfileBlobsRepository(db);
+
+    if (config.blobs.hydrateBlobs) {
+      if (config.blobs.storage.type === "s3") {
+        this.storage = new S3BlobStorage(
+          config.blobs.storage.s3Bucket!,
+          config.blobs.storage.s3Region!
+        );
+      } else {
+        this.storage = new LocalBlobStorage(
+          config.blobs.storage.localPath
+        );
+      }
+    }
+
     this.limit = pRateLimit({
       interval: 300000,
       rate: 3000,
···
   async hydrateProfile(did: string): Promise<void> {
     try {
       const existingProfile = await this.profilesRepo.findByDid(did);
-      if (existingProfile) {
-        logger.debug({ did }, "Profile already hydrated, skipping");
+      const needsRehydration = existingProfile && (existingProfile.avatar_cid === null || existingProfile.banner_cid === null);
+
+      if (existingProfile && !needsRehydration) {
+        logger.debug({ did }, "Profile already fully hydrated, skipping");
         return;
+      }
+
+      if (needsRehydration) {
+        logger.debug({ did }, "Re-hydrating profile to fetch avatar/banner CIDs");
       }
 
       const profileResponse = await this.limit(() =>
···
 
       let displayName: string | undefined;
       let description: string | undefined;
+      let avatarCid: string | undefined;
+      let bannerCid: string | undefined;
 
       if (profileResponse.success && profileResponse.data.value) {
         const record = profileResponse.data.value as any;
         displayName = record.displayName;
         description = record.description;
+
+        if (record.avatar?.ref) {
+          avatarCid = record.avatar.ref.toString();
+        } else {
+          avatarCid = "";
+        }
+
+        if (record.banner?.ref) {
+          bannerCid = record.banner.ref.toString();
+        } else {
+          bannerCid = "";
+        }
+
+        logger.debug({ did, avatarCid, bannerCid, hasAvatar: !!record.avatar, hasBanner: !!record.banner }, "Extracted CIDs from profile record");
       }
 
       const profileLookup = await this.limit(() =>
···
         handle,
         display_name: displayName,
         description,
+        avatar_cid: avatarCid,
+        banner_cid: bannerCid,
       });
 
-      logger.info({ did, handle }, "Profile hydrated successfully");
+      if (avatarCid && avatarCid !== "") {
+        try {
+          await this.processProfileBlob(did, avatarCid, "avatar");
+        } catch (error) {
+          logger.warn({ error, did, avatarCid }, "Failed to process avatar blob");
+        }
+      }
+
+      if (bannerCid && bannerCid !== "") {
+        try {
+          await this.processProfileBlob(did, bannerCid, "banner");
+        } catch (error) {
+          logger.warn({ error, did, bannerCid }, "Failed to process banner blob");
+        }
+      }
+
+      logger.info({ did, handle, avatarCid, bannerCid }, "Profile hydrated successfully");
     } catch (error) {
       if (isRecordNotFoundError(error)) {
         logger.warn({ did }, "Profile record not found, skipping");
···
       logger.error({ error, did }, "Failed to hydrate profile");
       throw error;
     }
+  }
+
+  private async resolvePds(did: string): Promise<string | null> {
+    try {
+      const didDocResponse = await fetch(`${config.plc.endpoint}/${did}`);
+      if (!didDocResponse.ok) {
+        logger.warn({ did, status: didDocResponse.status }, "Failed to fetch DID document");
+        return null;
+      }
+
+      const didDoc = await didDocResponse.json();
+      const pdsService = didDoc.service?.find((s: any) =>
+        s.id === "#atproto_pds" && s.type === "AtprotoPersonalDataServer"
+      );
+
+      if (!pdsService?.serviceEndpoint) {
+        logger.warn({ did }, "No PDS endpoint found in DID document");
+        return null;
+      }
+
+      return pdsService.serviceEndpoint;
+    } catch (error) {
+      logger.error({ error, did }, "Failed to resolve PDS from DID");
+      return null;
+    }
+  }
+
+  private async processProfileBlob(
+    did: string,
+    cid: string,
+    type: "avatar" | "banner"
+  ): Promise<void> {
+    const latestBlob = await this.profileBlobsRepo.findLatestByDidAndType(did, type);
+
+    if (latestBlob && latestBlob.blob_cid === cid) {
+      logger.debug({ did, cid, type }, "Latest blob already has same CID, skipping");
+      return;
+    }
+
+    const pdsEndpoint = await this.resolvePds(did);
+    if (!pdsEndpoint) {
+      logger.warn({ did, cid, type }, "Cannot fetch blob without PDS endpoint");
+      return;
+    }
+
+    const blobUrl = `${pdsEndpoint}/xrpc/com.atproto.sync.getBlob?did=${did}&cid=${cid}`;
+    const blobResponse = await fetch(blobUrl);
+
+    if (!blobResponse.ok) {
+      logger.warn({ did, cid, type, pdsEndpoint, status: blobResponse.status }, "Failed to fetch blob from PDS");
+      return;
+    }
+
+    const blobData = Buffer.from(await blobResponse.arrayBuffer());
+
+    let storagePath: string | undefined;
+    if (this.storage && config.blobs.hydrateBlobs) {
+      storagePath = await this.storage.store(cid, blobData, "image/jpeg");
+    }
+
+    const hashes = await computeBlobHashes(blobData, "image/jpeg");
+
+    await this.profileBlobsRepo.insert({
+      did,
+      blob_type: type,
+      blob_cid: cid,
+      sha256: hashes.sha256,
+      phash: hashes.phash,
+      storage_path: storagePath,
+      mimetype: "image/jpeg",
+    });
+
+    logger.info({ did, cid, type, sha256: hashes.sha256, pdsEndpoint, storagePath }, "Profile blob processed successfully");
   }
 }

tests/integration/database.test.ts

···
           did TEXT PRIMARY KEY,
           handle TEXT,
           display_name TEXT,
-          description TEXT
+          description TEXT,
+          avatar_cid TEXT,
+          banner_cid TEXT
         );
 
         CREATE TABLE IF NOT EXISTS blobs (
···
       expect(found).not.toBeNull();
       expect(found?.did).toBe("did:plc:testuser");
     });
+
+    test("should insert and retrieve profile with avatar and banner", async () => {
+      const profile = {
+        did: "did:plc:testuser2",
+        handle: "testuser2.bsky.social",
+        display_name: "Test User 2",
+        description: "A test user with avatar",
+        avatar_cid: "bafyavatartest",
+        banner_cid: "bafybannertest",
+      };
+
+      await profilesRepo.insert(profile);
+      const found = await profilesRepo.findByDid(profile.did);
+
+      expect(found).not.toBeNull();
+      expect(found?.avatar_cid).toBe("bafyavatartest");
+      expect(found?.banner_cid).toBe("bafybannertest");
+    });
   });
 
   describe("BlobsRepository", () => {
···
     test("should find blobs by pHash", async () => {
       const found = await blobsRepo.findByPhash("deadbeef");
       expect(found.length).toBeGreaterThan(0);
+    });
+
+    test("should find blob by CID", async () => {
+      const found = await blobsRepo.findByCid("bafytest123");
+      expect(found).not.toBeNull();
+      expect(found?.sha256).toBe("abc123def456");
     });
   });
 });