diff --git a/src/database/profile-blobs.repository.ts b/src/database/profile-blobs.repository.ts new file mode 100644 index 0000000..859d5d1 --- /dev/null +++ b/src/database/profile-blobs.repository.ts @@ -0,0 +1,109 @@ +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; +} + +export class ProfileBlobsRepository { + constructor(private db: Database) {} + + async insert(blob: ProfileBlob): Promise { + return new Promise((resolve, reject) => { + this.db.prepare( + ` + INSERT INTO profile_blobs (did, blob_type, blob_cid, sha256, phash, storage_path, mimetype) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (did, blob_type) DO UPDATE SET + blob_cid = EXCLUDED.blob_cid, + sha256 = EXCLUDED.sha256, + phash = EXCLUDED.phash, + storage_path = EXCLUDED.storage_path, + mimetype = EXCLUDED.mimetype + `, + (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, + (err) => { + if (err) { + logger.error({ err, blob }, "Failed to insert profile blob"); + reject(err); + return; + } + resolve(); + } + ); + } + ); + }); + } + + async findByDid(did: string): Promise { + return new Promise((resolve, reject) => { + this.db.all( + `SELECT * FROM profile_blobs WHERE did = $1`, + did, + (err, rows: ProfileBlob[]) => { + if (err) { + logger.error({ err, did }, "Failed to find profile blobs by DID"); + reject(err); + return; + } + resolve(rows || []); + } + ); + }); + } + + async findBySha256(sha256: string): Promise { + 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 { + 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 || []); + } + ); + }); + } +} diff --git a/src/database/schema.ts b/src/database/schema.ts index 0eb3c98..f3424c0 100644 --- a/src/database/schema.ts +++ b/src/database/schema.ts @@ -39,7 +39,7 @@ CREATE TABLE IF NOT EXISTS profiles ( banner_cid TEXT ); --- Blobs table: stores information about image blobs found in posts and profiles +-- Blobs table: stores information about image blobs found in posts CREATE TABLE IF NOT EXISTS blobs ( post_uri TEXT NOT NULL, blob_cid TEXT NOT NULL, @@ -47,7 +47,21 @@ CREATE TABLE IF NOT EXISTS blobs ( phash TEXT, storage_path TEXT, mimetype TEXT, - PRIMARY KEY (post_uri, blob_cid) + PRIMARY KEY (post_uri, blob_cid), + 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, + PRIMARY KEY (did, blob_type), + FOREIGN KEY (did) REFERENCES profiles(did) ); -- Indexes for performance @@ -57,6 +71,8 @@ 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_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 { @@ -105,49 +121,6 @@ async function migrateProfilesTable(): Promise { }); } -async function migrateBlobsTableConstraint(): Promise { - const db = getDatabase(); - - return new Promise((resolve, reject) => { - db.all( - `SELECT constraint_name FROM information_schema.table_constraints - WHERE table_name = 'blobs' AND constraint_type = 'FOREIGN KEY'`, - (err, rows: any[]) => { - if (err) { - logger.error({ err }, "Failed to check blobs table constraints"); - reject(err); - return; - } - - if (rows && rows.length > 0) { - logger.info("Migrating blobs table to remove foreign key constraint"); - - const migration = ` - CREATE TABLE blobs_new AS SELECT * FROM blobs; - DROP TABLE blobs; - ALTER TABLE blobs_new RENAME TO blobs; - CREATE INDEX IF NOT EXISTS idx_blobs_sha256 ON blobs(sha256); - CREATE INDEX IF NOT EXISTS idx_blobs_phash ON blobs(phash); - `; - - db.exec(migration, (err) => { - if (err) { - logger.error({ err }, "Failed to migrate blobs table"); - reject(err); - return; - } - logger.info("Blobs table migration completed"); - resolve(); - }); - } else { - logger.debug("Blobs table already has no foreign key constraint"); - resolve(); - } - } - ); - }); -} - export async function initializeSchema(): Promise { const db = getDatabase(); @@ -162,7 +135,6 @@ export async function initializeSchema(): Promise { try { await migrateProfilesTable(); - await migrateBlobsTableConstraint(); resolve(); } catch (migrationErr) { reject(migrationErr); diff --git a/src/hydration/profiles.service.ts b/src/hydration/profiles.service.ts index cbe5a3a..efae263 100644 --- a/src/hydration/profiles.service.ts +++ b/src/hydration/profiles.service.ts @@ -1,7 +1,7 @@ import { AtpAgent } from "@atproto/api"; import { Database } from "duckdb"; import { ProfilesRepository } from "../database/profiles.repository.js"; -import { BlobsRepository } from "../database/blobs.repository.js"; +import { ProfileBlobsRepository } from "../database/profile-blobs.repository.js"; import { computeBlobHashes } from "../blobs/hasher.js"; import { pRateLimit } from "p-ratelimit"; import { withRetry, isRateLimitError, isNetworkError, isServerError } from "../utils/retry.js"; @@ -11,13 +11,13 @@ import { config } from "../config/index.js"; export class ProfileHydrationService { private agent: AtpAgent; private profilesRepo: ProfilesRepository; - private blobsRepo: BlobsRepository; + private profileBlobsRepo: ProfileBlobsRepository; private limit: ReturnType; constructor(db: Database) { this.agent = new AtpAgent({ service: `https://${config.bsky.pds}` }); this.profilesRepo = new ProfilesRepository(db); - this.blobsRepo = new BlobsRepository(db); + this.profileBlobsRepo = new ProfileBlobsRepository(db); this.limit = pRateLimit({ interval: 300000, rate: 3000, @@ -187,10 +187,10 @@ export class ProfileHydrationService { cid: string, type: "avatar" | "banner" ): Promise { - const postUri = `profile://${did}/${type}`; - const existing = await this.blobsRepo.findByPostUri(postUri); + const existing = await this.profileBlobsRepo.findByDid(did); + const existingBlob = existing.find(b => b.blob_type === type && b.blob_cid === cid); - if (existing.length > 0 && existing.some(b => b.blob_cid === cid)) { + if (existingBlob) { logger.debug({ did, cid, type }, "Blob already processed, skipping"); return; } @@ -212,8 +212,9 @@ export class ProfileHydrationService { const blobData = Buffer.from(await blobResponse.arrayBuffer()); const hashes = await computeBlobHashes(blobData, "image/jpeg"); - await this.blobsRepo.insert({ - post_uri: postUri, + await this.profileBlobsRepo.insert({ + did, + blob_type: type, blob_cid: cid, sha256: hashes.sha256, phash: hashes.phash,