diff --git a/apps/server/package.json b/apps/server/package.json index d928dbd7..2a8fb6bf 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -39,6 +39,7 @@ "date-fns": "^4.4.0", "effect": "4.0.0-beta.100", "grammy": "catalog:", + "jszip": "^3.10.1", "pg": "^8.22.0", "pino": "^10.3.1", "pino-pretty": "^13.1.3", diff --git a/apps/server/src/crypto/cookie-encryption.test.ts b/apps/server/src/crypto/cookie-encryption.test.ts index e275f3b0..84a8d495 100644 --- a/apps/server/src/crypto/cookie-encryption.test.ts +++ b/apps/server/src/crypto/cookie-encryption.test.ts @@ -195,6 +195,35 @@ describe("CookieEncryption", () => { }); describe("key derivation", () => { + test("decrypts scoped ciphertext without marking it as legacy", () => { + const encrypted = encryption.encryptScoped( + testCookieData, + testUserId, + "provider:twitter:cookies:v1", + ); + const decrypted = encryption.decryptScopedOrLegacy( + encrypted, + testUserId, + "provider:twitter:cookies:v1", + "legacy-telegram-id", + ); + + expect(decrypted).toEqual({ data: testCookieData, usedLegacyEncryption: false }); + }); + + test("decrypts legacy user-scoped ciphertext for migration", () => { + const legacyTelegramId = "123456"; + const encrypted = encryption.encrypt(testCookieData, legacyTelegramId); + const decrypted = encryption.decryptScopedOrLegacy( + encrypted, + testUserId, + "provider:twitter:cookies:v1", + legacyTelegramId, + ); + + expect(decrypted).toEqual({ data: testCookieData, usedLegacyEncryption: true }); + }); + test("should produce consistent results for same inputs", () => { const encrypted1 = encryption.encrypt(testCookieData, testUserId); const decrypted1 = encryption.decrypt(encrypted1, testUserId); @@ -243,41 +272,4 @@ describe("CookieEncryption", () => { expect(() => encryption2.decrypt(encrypted1, testUserId)).toThrow(); }); }); - - describe("performance characteristics", () => { - test("should encrypt and decrypt efficiently", () => { - const start = performance.now(); - - // Perform multiple operations - for (let i = 0; i < 100; i++) { - const encrypted = encryption.encrypt(testCookieData, testUserId); - const decrypted = encryption.decrypt(encrypted, testUserId); - expect(decrypted).toBe(testCookieData); - } - - const end = performance.now(); - const duration = end - start; - - // Should complete 100 encrypt/decrypt cycles in under 1 second - expect(duration).toBeLessThan(1000); - }); - - test("should handle concurrent operations", async () => { - const operations = Array.from({ length: 50 }, (_, i) => - Promise.resolve().then(() => { - const userId = `user_${i}`; - const data = `${testCookieData}_${i}`; - const encrypted = encryption.encrypt(data, userId); - const decrypted = encryption.decrypt(encrypted, userId); - return { original: data, decrypted, userId }; - }), - ); - - const results = await Promise.all(operations); - - for (const result of results) { - expect(result.decrypted).toBe(result.original); - } - }); - }); }); diff --git a/apps/server/src/handlers/image.ts b/apps/server/src/handlers/image.ts index 425bd592..ff77b09d 100644 --- a/apps/server/src/handlers/image.ts +++ b/apps/server/src/handlers/image.ts @@ -1,24 +1,43 @@ import { FormattedString } from "@grammyjs/parse-mode"; -import { CookieEncryption } from "@starlight/crypto"; import { EmbeddingsService } from "@starlight/api/services/embeddings"; +import { hasTwitterCookies } from "@starlight/api/services/twitter-credential"; import { env, isTwitterUrl, Prisma, prisma } from "@starlight/utils"; import { Composer, InlineKeyboard, InlineQueryResultBuilder } from "grammy"; -import { webAppKeyboard } from "@/bot"; import type { Logger } from "@/logger"; -import { RETRY } from "@/queue/absurd"; -import { getScheduledScrapperGeneration, scrapperApp } from "@/queue/scrapper"; import { runtime } from "@/services/runtime"; -import { Cookies } from "@/storage"; import type { Context } from "@/types"; const INLINE_QUERY_PAGE_SIZE = 50; const INLINE_QUERY_CANDIDATE_MULTIPLIER = 8; const INLINE_QUERY_AUTHOR_REGEX = /(^|\s)@([A-Za-z0-9_]+)/g; +const createInlineImageResultId = ( + provider: string, + externalMediaId: string, + userId: string, +): string => { + const hasher = new Bun.CryptoHasher("sha256"); + hasher.update(JSON.stringify(["v1", provider, externalMediaId, userId])); + return `m_${hasher.digest("base64url")}`; +}; + +const createInlineImageDedupeKey = ( + provider: string, + externalMediaId: string, + userId: string, + perceptualHash: string | null, +): string => + perceptualHash?.trim() + ? JSON.stringify(["hash", provider, perceptualHash.trim(), userId]) + : JSON.stringify(["identity", provider, externalMediaId, userId]); + type InlineImageSearchResult = { photo_id: string; + photo_provider: string; + photo_user_id: string; s3_path: string; tweet_id: string; + source_url: string; username: string | null; height: number | null; width: number | null; @@ -72,54 +91,62 @@ async function searchInlineImagesWithLegacyQuery( while (allPhotos.length < photoOffset + pageQueryLimit) { const { authors, textQuery } = parseInlineImageQuery(query); - const whereClause: Prisma.TweetWhereInput = {}; + const whereClause: Prisma.PostWhereInput = {}; if (authors.length > 0 && textQuery) { whereClause.AND = [ { OR: authors.map((author) => ({ - username: { contains: author, mode: "insensitive" }, + authorUsername: { contains: author, mode: "insensitive" }, })), }, - { tweetText: { contains: textQuery, mode: "insensitive" } }, + { + OR: [ + { text: { contains: textQuery, mode: "insensitive" } }, + { title: { contains: textQuery, mode: "insensitive" } }, + { tags: { has: textQuery } }, + ], + }, ]; } else if (authors.length > 0) { whereClause.OR = authors.map((author) => ({ - username: { contains: author, mode: "insensitive" }, + authorUsername: { contains: author, mode: "insensitive" }, })); } else if (textQuery) { - whereClause.tweetText = { contains: textQuery, mode: "insensitive" }; + whereClause.OR = [ + { text: { contains: textQuery, mode: "insensitive" } }, + { title: { contains: textQuery, mode: "insensitive" } }, + { tags: { has: textQuery } }, + ]; } const tweets = await runInlineImageQuery( logger, { searchMode: "legacy", userId, tweetSkip, pageSize: INLINE_QUERY_PAGE_SIZE }, () => - prisma.tweet.findMany({ + prisma.post.findMany({ where: { userId, - photos: { + media: { some: { deletedAt: null, + kind: "image", s3Path: { not: null }, }, }, ...whereClause, }, include: { - photos: { + media: { where: { deletedAt: null, + kind: "image", s3Path: { not: null }, }, - orderBy: { - createdAt: "desc", - }, + orderBy: [{ createdAt: "desc" }, { provider: "desc" }, { id: "desc" }], }, }, - orderBy: { - createdAt: "desc", - }, + orderBy: [{ createdAt: "desc" }, { provider: "desc" }, { id: "desc" }], take: INLINE_QUERY_PAGE_SIZE, skip: tweetSkip, }), @@ -130,8 +157,13 @@ async function searchInlineImagesWithLegacyQuery( } for (const tweet of tweets) { - for (const photo of tweet.photos) { - const dedupeKey = photo.perceptualHash?.trim() || photo.id; + for (const photo of tweet.media) { + const dedupeKey = createInlineImageDedupeKey( + photo.provider, + photo.id, + photo.userId, + photo.perceptualHash, + ); if (seenPhotoKeys.has(dedupeKey)) { continue; @@ -140,9 +172,12 @@ async function searchInlineImagesWithLegacyQuery( seenPhotoKeys.add(dedupeKey); allPhotos.push({ photo_id: photo.id, + photo_provider: photo.provider, + photo_user_id: photo.userId, s3_path: photo.s3Path as string, tweet_id: tweet.id, - username: tweet.username, + source_url: tweet.sourceUrl, + username: tweet.authorUsername ?? tweet.username, height: photo.height, width: photo.width, final_score: 0, @@ -198,11 +233,6 @@ async function getInlineQueryEmbedding(query: string) { return text; } -const cookieEncryption = new CookieEncryption( - env.COOKIE_ENCRYPTION_KEY, - env.COOKIE_ENCRYPTION_SALT, -); - const composer = new Composer(); const privateChat = composer.chatType("private"); @@ -225,13 +255,19 @@ composer.on("inline_query").filter( 200, ); const queryTime = new Date().toISOString(); - const photoDedupeKey = Prisma.sql`COALESCE(NULLIF(p.perceptual_hash, ''), p.id)`; + const photoDedupeKey = Prisma.sql`jsonb_build_array( + CASE WHEN NULLIF(p.perceptual_hash, '') IS NULL THEN 'identity' ELSE 'hash' END, + p.provider, + COALESCE(NULLIF(p.perceptual_hash, ''), p.external_id), + p.user_id + )::text`; const authorFilter = authors.length > 0 ? Prisma.sql`AND (${Prisma.join( authors.map( - (author) => Prisma.sql`strpos(lower(COALESCE(t.username, '')), ${author}) > 0`, + (author) => + Prisma.sql`strpos(lower(COALESCE(t.author_username, t.username, '')), ${author}) > 0`, ), " OR ", )})` @@ -243,9 +279,9 @@ composer.on("inline_query").filter( authors.map( (author) => Prisma.sql`CASE - WHEN lower(COALESCE(t.username, '')) = ${author} THEN 1.0 - WHEN strpos(lower(COALESCE(t.username, '')), ${author}) = 1 THEN 0.88 - WHEN strpos(lower(COALESCE(t.username, '')), ${author}) > 0 THEN 0.76 + WHEN lower(COALESCE(t.author_username, t.username, '')) = ${author} THEN 1.0 + WHEN strpos(lower(COALESCE(t.author_username, t.username, '')), ${author}) = 1 THEN 0.88 + WHEN strpos(lower(COALESCE(t.author_username, t.username, '')), ${author}) > 0 THEN 0.76 ELSE 0.0 END`, ), @@ -271,13 +307,16 @@ composer.on("inline_query").filter( OR lower(general_tag.value) LIKE ${queryStartsWith} OR lower(general_tag.value) LIKE ${queryContains} ) - OR lower(COALESCE(t.tweet_text, '')) LIKE ${queryContains} + OR lower(COALESCE(t.text, '')) LIKE ${queryContains} + OR lower(COALESCE(t.title, '')) LIKE ${queryContains} + OR lower(COALESCE(t.author_name, '')) LIKE ${queryContains} + OR lower(COALESCE(t.author_username, '')) LIKE ${queryContains} OR EXISTS ( SELECT 1 - FROM jsonb_array_elements_text(COALESCE(t.tweet_data->'hashtags', '[]'::jsonb)) AS hashtag(value) - WHERE lower(hashtag.value) = ${queryLower} - OR lower(hashtag.value) LIKE ${queryStartsWith} - OR lower(hashtag.value) LIKE ${queryContains} + FROM unnest(t.tags) AS post_tag(value) + WHERE lower(post_tag.value) = ${queryLower} + OR lower(post_tag.value) LIKE ${queryStartsWith} + OR lower(post_tag.value) LIKE ${queryContains} ) ) ` @@ -322,27 +361,32 @@ composer.on("inline_query").filter( ` : Prisma.sql`0.0`; - const hashtagScore = hasTextQuery + const postTagScore = hasTextQuery ? Prisma.sql` COALESCE( ( SELECT MAX( CASE - WHEN lower(hashtag.value) = ${queryLower} THEN 0.76 - WHEN lower(hashtag.value) LIKE ${queryStartsWith} THEN 0.62 - WHEN lower(hashtag.value) LIKE ${queryContains} THEN 0.5 + WHEN lower(post_tag.value) = ${queryLower} THEN 0.76 + WHEN lower(post_tag.value) LIKE ${queryStartsWith} THEN 0.62 + WHEN lower(post_tag.value) LIKE ${queryContains} THEN 0.5 ELSE 0.0 END ) - FROM jsonb_array_elements_text(COALESCE(t.tweet_data->'hashtags', '[]'::jsonb)) AS hashtag(value) + FROM unnest(t.tags) AS post_tag(value) ), 0.0 ) ` : Prisma.sql`0.0`; - const tweetTextScore = hasTextQuery - ? Prisma.sql`CASE WHEN lower(COALESCE(t.tweet_text, '')) LIKE ${queryContains} THEN 0.34 ELSE 0.0 END` + const postTextScore = hasTextQuery + ? Prisma.sql`GREATEST( + CASE WHEN lower(COALESCE(t.text, '')) LIKE ${queryContains} THEN 0.34 ELSE 0.0 END, + CASE WHEN lower(COALESCE(t.title, '')) LIKE ${queryContains} THEN 0.34 ELSE 0.0 END, + CASE WHEN lower(COALESCE(t.author_name, '')) LIKE ${queryContains} THEN 0.3 ELSE 0.0 END, + CASE WHEN lower(COALESCE(t.author_username, '')) LIKE ${queryContains} THEN 0.3 ELSE 0.0 END + )` : Prisma.sql`0.0`; let rankedPhotos: InlineImageSearchResult[] = []; @@ -353,7 +397,8 @@ composer.on("inline_query").filter( authors.length > 0 ? Prisma.sql`AND (${Prisma.join( authors.map( - (author) => Prisma.sql`strpos(lower(COALESCE(t.username, '')), ${author}) > 0`, + (author) => + Prisma.sql`strpos(lower(COALESCE(t.author_username, t.username, '')), ${author}) > 0`, ), " OR ", )})` @@ -366,35 +411,42 @@ composer.on("inline_query").filter( prisma.$queryRaw(Prisma.sql` WITH ranked AS ( SELECT - p.id AS photo_id, + p.external_id AS photo_id, + p.provider AS photo_provider, + p.user_id AS photo_user_id, p.s3_path, - t.id AS tweet_id, - t.username, + t.external_id AS tweet_id, + t.source_url, + COALESCE(t.author_username, t.username) AS username, p.height, p.width, p.created_at AS photo_created_at, ROW_NUMBER() OVER ( PARTITION BY ${photoDedupeKey} - ORDER BY p.created_at DESC, p.id DESC + ORDER BY p.created_at DESC, p.provider DESC, p.external_id DESC, p.user_id DESC ) AS duplicate_rank - FROM photos p - JOIN tweets t ON t.id = p.tweet_id AND t.user_id = p.user_id + FROM media p + JOIN posts t ON t.external_id = p.post_external_id AND t.user_id = p.user_id AND t.provider = p.provider WHERE p.user_id = ${userId} AND p.deleted_at IS NULL + AND p.kind = 'image' AND p.s3_path IS NOT NULL ${recencyAuthorFilter} ) SELECT photo_id, + photo_provider, + photo_user_id, s3_path, tweet_id, + source_url, username, height, width, 0.0 AS final_score FROM ranked WHERE duplicate_rank = 1 - ORDER BY photo_created_at DESC, photo_id DESC + ORDER BY photo_created_at DESC, photo_provider DESC, photo_id DESC, photo_user_id DESC OFFSET ${photoOffset} LIMIT ${pageQueryLimit} `), @@ -435,81 +487,110 @@ composer.on("inline_query").filter( () => prisma.$queryRaw(Prisma.sql` WITH image_candidates AS ( - SELECT p.id, p.user_id - FROM photos p - JOIN tweets t ON t.id = p.tweet_id AND t.user_id = p.user_id + SELECT p.external_id AS id, p.user_id, p.provider + FROM media p + JOIN posts t ON t.external_id = p.post_external_id AND t.user_id = p.user_id AND t.provider = p.provider WHERE p.user_id = ${userId} AND p.deleted_at IS NULL + AND p.kind = 'image' AND p.s3_path IS NOT NULL AND p.classification IS NOT NULL AND p.image_vec IS NOT NULL AND p.tag_vec IS NOT NULL ${authorFilter} - ORDER BY p.image_vec <=> ${textVector}::vector + ORDER BY p.image_vec <=> ${textVector}::vector, p.provider DESC, p.external_id DESC, p.user_id DESC LIMIT ${candidateLimit} ), tag_candidates AS ( - SELECT p.id, p.user_id - FROM photos p - JOIN tweets t ON t.id = p.tweet_id AND t.user_id = p.user_id + SELECT p.external_id AS id, p.user_id, p.provider + FROM media p + JOIN posts t ON t.external_id = p.post_external_id AND t.user_id = p.user_id AND t.provider = p.provider WHERE p.user_id = ${userId} AND p.deleted_at IS NULL + AND p.kind = 'image' AND p.s3_path IS NOT NULL AND p.classification IS NOT NULL AND p.image_vec IS NOT NULL AND p.tag_vec IS NOT NULL ${authorFilter} - ORDER BY p.tag_vec <=> ${textVector}::vector + ORDER BY p.tag_vec <=> ${textVector}::vector, p.provider DESC, p.external_id DESC, p.user_id DESC LIMIT ${candidateLimit} ), lexical_candidates AS ( - SELECT p.id, p.user_id - FROM photos p - JOIN tweets t ON t.id = p.tweet_id AND t.user_id = p.user_id + SELECT p.external_id AS id, p.user_id, p.provider + FROM media p + JOIN posts t ON t.external_id = p.post_external_id AND t.user_id = p.user_id AND t.provider = p.provider + CROSS JOIN LATERAL ( + SELECT COALESCE(MAX( + CASE + WHEN lower(lexical_value.value) = ${queryLower} THEN 3 + WHEN lower(lexical_value.value) LIKE ${queryStartsWith} THEN 2 + WHEN lower(lexical_value.value) LIKE ${queryContains} THEN 1 + ELSE 0 + END + ), 0) AS lexical_score + FROM jsonb_array_elements_text( + COALESCE(p.classification->'characters', '[]'::jsonb) + || COALESCE(p.classification->'tags', '[]'::jsonb) + || to_jsonb(COALESCE(t.tags, ARRAY[]::text[])) + || jsonb_build_array( + COALESCE(t.text, ''), COALESCE(t.title, ''), + COALESCE(t.author_name, ''), COALESCE(t.author_username, '') + ) + ) AS lexical_value(value) + ) lexical_rank WHERE p.user_id = ${userId} AND p.deleted_at IS NULL + AND p.kind = 'image' AND p.s3_path IS NOT NULL AND ${lexicalMatch} ${authorFilter} + ORDER BY lexical_rank.lexical_score DESC, p.provider DESC, p.external_id DESC, p.user_id DESC LIMIT ${candidateLimit} ), candidate_pool AS ( - SELECT DISTINCT id, user_id + SELECT DISTINCT id, user_id, provider FROM ( - SELECT id, user_id FROM image_candidates + SELECT id, user_id, provider FROM image_candidates UNION ALL - SELECT id, user_id FROM tag_candidates + SELECT id, user_id, provider FROM tag_candidates UNION ALL - SELECT id, user_id FROM lexical_candidates + SELECT id, user_id, provider FROM lexical_candidates ) candidates ), scored AS ( SELECT - p.id AS photo_id, + p.external_id AS photo_id, + p.provider AS photo_provider, + p.user_id AS photo_user_id, ${photoDedupeKey} AS dedupe_key, p.s3_path, p.height, p.width, - t.username, - t.id AS tweet_id, + COALESCE(t.author_username, t.username) AS username, + t.external_id AS tweet_id, + t.source_url, t.created_at AS tweet_created_at, COALESCE(1.0 - (p.image_vec <=> ${textVector}::vector), 0.0) AS s_image, COALESCE(1.0 - (p.tag_vec <=> ${textVector}::vector), 0.0) AS s_tag_semantic, ${characterScore} AS s_character, ${tagLexicalScore} AS s_tag_lexical, - ${hashtagScore} AS s_hashtag, - ${tweetTextScore} AS s_tweet_text, + ${postTagScore} AS s_post_tag, + ${postTextScore} AS s_post_text, ${authorScore} AS s_author FROM candidate_pool c - JOIN photos p ON p.id = c.id AND p.user_id = c.user_id - JOIN tweets t ON t.id = p.tweet_id AND t.user_id = p.user_id + JOIN media p ON p.external_id = c.id AND p.user_id = c.user_id AND p.provider = c.provider + JOIN posts t ON t.external_id = p.post_external_id AND t.user_id = p.user_id AND t.provider = p.provider ), fused AS ( SELECT photo_id, + photo_provider, + photo_user_id, dedupe_key, s3_path, tweet_id, + source_url, username, tweet_created_at, height, @@ -517,7 +598,7 @@ composer.on("inline_query").filter( ( (s_character * 0.4) + (GREATEST(s_tag_semantic, s_tag_lexical) * 0.24) + - (GREATEST(s_hashtag, s_tweet_text) * 0.12) + + (GREATEST(s_post_tag, s_post_text) * 0.12) + (s_image * 0.1) + (s_author * 0.08) + (0.02 * EXP(LN(0.5) * (EXTRACT(EPOCH FROM (${queryTime}::timestamptz - tweet_created_at)) / (180.0 * 24 * 3600.0)))) @@ -527,22 +608,25 @@ composer.on("inline_query").filter( deduped AS ( SELECT photo_id, + photo_provider, + photo_user_id, s3_path, tweet_id, + source_url, username, height, width, final_score, ROW_NUMBER() OVER ( PARTITION BY dedupe_key - ORDER BY final_score DESC NULLS LAST, tweet_created_at DESC, photo_id DESC + ORDER BY final_score DESC NULLS LAST, tweet_created_at DESC, photo_provider DESC, photo_id DESC, photo_user_id DESC ) AS duplicate_rank FROM fused ) - SELECT photo_id, s3_path, tweet_id, username, height, width, final_score + SELECT photo_id, photo_provider, photo_user_id, s3_path, tweet_id, source_url, username, height, width, final_score FROM deduped WHERE duplicate_rank = 1 - ORDER BY final_score DESC NULLS LAST, photo_id DESC + ORDER BY final_score DESC NULLS LAST, photo_provider DESC, photo_id DESC, photo_user_id DESC OFFSET ${photoOffset} LIMIT ${pageQueryLimit} `), @@ -557,23 +641,27 @@ composer.on("inline_query").filter( prisma.$queryRaw(Prisma.sql` WITH scored AS ( SELECT - p.id AS photo_id, + p.external_id AS photo_id, + p.provider AS photo_provider, + p.user_id AS photo_user_id, ${photoDedupeKey} AS dedupe_key, p.s3_path, p.height, p.width, - t.username, - t.id AS tweet_id, + COALESCE(t.author_username, t.username) AS username, + t.external_id AS tweet_id, + t.source_url, t.created_at AS tweet_created_at, ${characterScore} AS s_character, ${tagLexicalScore} AS s_tag_lexical, - ${hashtagScore} AS s_hashtag, - ${tweetTextScore} AS s_tweet_text, + ${postTagScore} AS s_post_tag, + ${postTextScore} AS s_post_text, ${authorScore} AS s_author - FROM photos p - JOIN tweets t ON t.id = p.tweet_id AND t.user_id = p.user_id + FROM media p + JOIN posts t ON t.external_id = p.post_external_id AND t.user_id = p.user_id AND t.provider = p.provider WHERE p.user_id = ${userId} AND p.deleted_at IS NULL + AND p.kind = 'image' AND p.s3_path IS NOT NULL ${authorFilter} ${lexicalFilter} @@ -581,9 +669,12 @@ composer.on("inline_query").filter( fused AS ( SELECT photo_id, + photo_provider, + photo_user_id, dedupe_key, s3_path, tweet_id, + source_url, username, tweet_created_at, height, @@ -592,7 +683,7 @@ composer.on("inline_query").filter( (s_author * 0.56) + (s_character * 0.22) + (s_tag_lexical * 0.12) + - (GREATEST(s_hashtag, s_tweet_text) * 0.08) + + (GREATEST(s_post_tag, s_post_text) * 0.08) + (0.02 * EXP(LN(0.5) * (EXTRACT(EPOCH FROM (NOW() - tweet_created_at)) / (180.0 * 24 * 3600.0)))) ) AS final_score FROM scored @@ -600,22 +691,25 @@ composer.on("inline_query").filter( deduped AS ( SELECT photo_id, + photo_provider, + photo_user_id, s3_path, tweet_id, + source_url, username, height, width, final_score, ROW_NUMBER() OVER ( PARTITION BY dedupe_key - ORDER BY final_score DESC NULLS LAST, tweet_created_at DESC, photo_id DESC + ORDER BY final_score DESC NULLS LAST, tweet_created_at DESC, photo_provider DESC, photo_id DESC, photo_user_id DESC ) AS duplicate_rank FROM fused ) - SELECT photo_id, s3_path, tweet_id, username, height, width, final_score + SELECT photo_id, photo_provider, photo_user_id, s3_path, tweet_id, source_url, username, height, width, final_score FROM deduped WHERE duplicate_rank = 1 - ORDER BY final_score DESC NULLS LAST, photo_id DESC + ORDER BY final_score DESC NULLS LAST, photo_provider DESC, photo_id DESC, photo_user_id DESC OFFSET ${photoOffset} LIMIT ${pageQueryLimit} `), @@ -626,7 +720,7 @@ composer.on("inline_query").filter( const photosForThisPage = rankedPhotos.slice(0, INLINE_QUERY_PAGE_SIZE); - if (photosForThisPage.length === 0 && !ctx.user?.cookies) { + if (photosForThisPage.length === 0 && (!ctx.user || !(await hasTwitterCookies(ctx.user.id)))) { // User didn't setup the bot yet await ctx.answerInlineQuery( [ @@ -648,16 +742,20 @@ composer.on("inline_query").filter( const results = photosForThisPage.map((photo) => { const photoUrl = `${env.BASE_CDN_URL}/${photo.s3_path}`; const caption = photo.username - ? FormattedString.link(`@${photo.username}`, `https://x.com/i/status/${photo.tweet_id}`) - : new FormattedString(`https://x.com/i/status/${photo.tweet_id}`); - - return InlineQueryResultBuilder.photo(photo.photo_id, photoUrl, { - caption: caption.caption, - caption_entities: caption.caption_entities, - thumbnail_url: photoUrl, - photo_height: photo.height ?? undefined, - photo_width: photo.width ?? undefined, - }); + ? FormattedString.link(`@${photo.username}`, photo.source_url) + : new FormattedString(photo.source_url); + + return InlineQueryResultBuilder.photo( + createInlineImageResultId(photo.photo_provider, photo.photo_id, photo.photo_user_id), + photoUrl, + { + caption: caption.caption, + caption_entities: caption.caption_entities, + thumbnail_url: photoUrl, + photo_height: photo.height ?? undefined, + photo_width: photo.width ?? undefined, + }, + ); }); // Calculate next offset for pagination @@ -674,109 +772,15 @@ composer.on("inline_query").filter( }, ); -privateChat.command("cookies").filter( - async (ctx) => !ctx.user?.cookies, - async (ctx) => { - const keyboard = new InlineKeyboard().webApp("Set cookies", { - url: `${env.BASE_FRONTEND_URL}/settings`, - }); - - await ctx.reply("No cookies found. Please set your cookies first.", { - reply_markup: keyboard, - }); - }, -); - -privateChat.command("cookies").filter( - async (ctx) => Boolean(ctx.user?.cookies), - async (ctx) => { - try { - const userCookies = ctx.user?.cookies; - - if (!(userCookies && ctx.user)) { - await ctx.reply("No cookies found."); - return; - } - - const cookiesJson = cookieEncryption.safeDecrypt(userCookies, ctx.user.telegramId.toString()); - - const cookies = Cookies.fromJSON(cookiesJson); - const cookiesString = cookies.toString(); - - await ctx.reply(`Your cookies:\n\n${cookiesString}`); - } catch (error) { - ctx.logger.error({ error }, "Failed to decrypt cookies"); - await ctx.reply("Failed to decrypt cookies. Please try setting them again."); - } - }, -); - -privateChat.command("scrapper").filter( - async (ctx) => !ctx.user?.cookies, - async (ctx) => { - const keyboard = new InlineKeyboard().webApp("Set cookies", { - url: `${env.BASE_FRONTEND_URL}/cookies`, - }); - - await ctx.reply( - "Beep boop, you need to give me your cookies before I can send you daily images.", - { reply_markup: keyboard }, - ); - }, -); - -privateChat.command("scrapper").filter( - async (ctx) => Boolean(ctx.user?.cookies), - async (ctx) => { - const user = ctx.user!; - const generation = getScheduledScrapperGeneration(); - - const scheduledJob = await scrapperApp.spawn( - "scheduled-feed-scrapper", - { - generation, - userId: user.id, - limit: 300, - }, - { - idempotencyKey: `scheduled-scrapper-${user.id}-${generation}`, - maxAttempts: 3, - retryStrategy: RETRY.scrapper, - }, - ); - - if (scheduledJob.created) { - ctx.logger.debug({ userId: user.id }, "Scheduled scrapper"); - - await scrapperApp.spawn( - "feed-scrapper", - { userId: user.id, count: 0, limit: 300 }, - { - maxAttempts: 3, - retryStrategy: RETRY.scrapper, - }, - ); - - await ctx.reply( - "You placed in the queue (runs every 6 hours). You can check your images in a few minutes in your gallery.\n\nYou can start the job anytime by sending /scrapper command again.", - { - reply_markup: webAppKeyboard("app", "View gallery"), - }, - ); - return; - } - - await scrapperApp.spawn( - "feed-scrapper", - { userId: user.id, count: 0, limit: 100 }, - { - maxAttempts: 3, - retryStrategy: RETRY.scrapper, - }, - ); - - await ctx.reply("Starting to collect images, check back in a few minutes."); - }, -); +privateChat.command("cookies", async (ctx) => { + const connected = ctx.user ? await hasTwitterCookies(ctx.user.id) : false; + const keyboard = new InlineKeyboard().webApp(connected ? "Manage Twitter" : "Set cookies", { + url: `${env.BASE_FRONTEND_URL}/settings`, + }); + const message = connected + ? "Twitter is connected. Use settings to replace or delete your cookies." + : "No cookies found. Please set your cookies first."; + await ctx.reply(message, { reply_markup: keyboard }); +}); export default composer; diff --git a/apps/server/src/handlers/pixiv.ts b/apps/server/src/handlers/pixiv.ts new file mode 100644 index 00000000..70c1b92d --- /dev/null +++ b/apps/server/src/handlers/pixiv.ts @@ -0,0 +1,98 @@ +import { withPixivClient } from "@starlight/api/services/pixiv-credential"; +import { Composer, InputFile } from "grammy"; +import { readResponseBounded } from "@/services/media-download"; +import { + convertUgoira, + extractUgoiraZip, + MAX_PIXIV_DOWNLOAD_BYTES, + parsePixivArtworkUrl, +} from "@/services/pixiv-media"; +import type { Context } from "@/types"; + +const MAX_MANGA_BYTES = 150_000_000; +const pixivHandler = new Composer(); + +const createTemporaryDirectory = async () => { + const directory = `${Bun.env.TMPDIR ?? "/tmp"}/starlight-pixiv-${Bun.randomUUIDv7()}`; + const child = Bun.spawn(["mkdir", "-m", "700", directory], { + stdout: "ignore", + stderr: "ignore", + }); + if ((await child.exited) !== 0) { + throw new Error("Failed to create Pixiv temporary directory"); + } + return directory; +}; + +const removeTemporaryDirectory = async (directory: string) => { + const child = Bun.spawn(["rm", "-rf", directory], { stdout: "ignore", stderr: "ignore" }); + await child.exited; +}; + +pixivHandler.on("message:text").filter( + (ctx) => parsePixivArtworkUrl(ctx.message.text.trim()) !== null, + async (ctx) => { + const id = parsePixivArtworkUrl(ctx.message.text.trim())!; + const user = ctx.user!; + const directory = await createTemporaryDirectory(); + try { + const handled = await withPixivClient(user.id, async (client) => { + const artwork = await client.artwork(id); + if (artwork.type === "ugoira") { + const metadata = await client.ugoira(id); + const archive = await readResponseBounded( + await client.fetchMedia(metadata.zipUrls.medium), + ); + const extracted = await extractUgoiraZip(archive, metadata.frames); + try { + const output = `${extracted.directory}/ugoira.mp4`; + await convertUgoira(extracted.concatPath, output); + await ctx.replyWithVideo(new InputFile(output), { + caption: artwork.title, + }); + } finally { + await removeTemporaryDirectory(extracted.directory); + } + return true; + } + + let aggregateBytes = 0; + const files: InputFile[] = []; + for (const [position, url] of artwork.mediaUrls.entries()) { + const bytes = await readResponseBounded( + await client.fetchMedia(url), + Math.min(MAX_PIXIV_DOWNLOAD_BYTES, MAX_MANGA_BYTES - aggregateBytes), + ); + aggregateBytes += bytes.byteLength; + const extension = new URL(url).pathname.split(".").at(-1) ?? "jpg"; + const path = `${directory}/${position}.${extension}`; + await Bun.write(path, bytes); + files.push(new InputFile(path)); + } + for (let offset = 0; offset < files.length; offset += 10) { + const chunk = files.slice(offset, offset + 10); + const single = chunk.at(0); + if (chunk.length === 1 && single) { + await ctx.replyWithDocument(single, { caption: artwork.title }); + } else { + await ctx.replyWithMediaGroup( + chunk.map((file, index) => ({ + type: "document" as const, + media: file, + caption: index === 0 ? artwork.title : undefined, + })), + ); + } + } + return true; + }); + if (handled === undefined) { + await ctx.reply("Connect Pixiv in Settings first."); + } + } finally { + await removeTemporaryDirectory(directory); + } + }, +); + +export default pixivHandler; diff --git a/apps/server/src/handlers/scrapper.ts b/apps/server/src/handlers/scrapper.ts new file mode 100644 index 00000000..0ce631eb --- /dev/null +++ b/apps/server/src/handlers/scrapper.ts @@ -0,0 +1,223 @@ +import { hasTwitterCookies } from "@starlight/api/services/twitter-credential"; +import { env, prisma } from "@starlight/utils"; +import { Composer, InlineKeyboard } from "grammy"; +import { webAppKeyboard } from "@/bot"; +import type { Logger } from "@/logger"; +import { RETRY } from "@/queue/absurd"; +import { getScheduledPixivGeneration, pixivApp } from "@/queue/pixiv"; +import { getScheduledScrapperGeneration, scrapperApp } from "@/queue/scrapper"; +import type { Context } from "@/types"; + +type ProviderCollectionResult = { + immediateStarted: boolean; + scheduleReady: boolean; +}; + +type ProviderConnection = "connected" | "disconnected" | "lookup-failed"; + +type ScrapperConnections = { + pixiv: ProviderConnection; + twitter: ProviderConnection; +}; + +const composer = new Composer(); +const privateChat = composer.chatType("private"); + +async function getScrapperConnections(ctx: Context): Promise { + const user = ctx.user!; + const [twitter, pixivCredential] = await Promise.allSettled([ + hasTwitterCookies(user.id), + prisma.providerCredential.findUnique({ + where: { userId_provider: { userId: user.id, provider: "pixiv" } }, + select: { credentialType: true }, + }), + ]); + if (twitter.status === "rejected") { + ctx.logger.warn( + { error: twitter.reason, userId: user.id, provider: "twitter" }, + "Failed to check provider connection", + ); + } + if (pixivCredential.status === "rejected") { + ctx.logger.warn( + { error: pixivCredential.reason, userId: user.id, provider: "pixiv" }, + "Failed to check provider connection", + ); + } + + return { + twitter: + twitter.status === "rejected" + ? "lookup-failed" + : twitter.value + ? "connected" + : "disconnected", + pixiv: + pixivCredential.status === "rejected" + ? "lookup-failed" + : pixivCredential.value?.credentialType === "refresh_token" + ? "connected" + : "disconnected", + }; +} + +async function startTwitterCollection( + userId: string, + updateId: number, + logger: Logger, +): Promise { + const generation = getScheduledScrapperGeneration(); + let scheduleReady = false; + let firstSchedule = false; + + try { + const scheduledJob = await scrapperApp.spawn( + "scheduled-feed-scrapper", + { generation, userId, limit: 300 }, + { + idempotencyKey: `scheduled-scrapper-${userId}-${generation}`, + maxAttempts: 3, + retryStrategy: RETRY.scrapper, + }, + ); + scheduleReady = true; + firstSchedule = scheduledJob.created; + if (firstSchedule) { + logger.debug({ userId, provider: "twitter" }, "Scheduled collector"); + } + } catch (error) { + logger.warn({ error, userId, provider: "twitter" }, "Failed to schedule collector"); + } + + let immediateStarted = false; + try { + await scrapperApp.spawn( + "feed-scrapper", + { userId, count: 0, limit: firstSchedule ? 300 : 100 }, + { + idempotencyKey: `manual-scrapper-${userId}-${updateId}`, + maxAttempts: 3, + retryStrategy: RETRY.scrapper, + }, + ); + immediateStarted = true; + } catch (error) { + logger.warn({ error, userId, provider: "twitter" }, "Failed to start collector"); + } + + return { immediateStarted, scheduleReady }; +} + +async function startPixivCollection( + userId: string, + updateId: number, + logger: Logger, +): Promise { + const generation = getScheduledPixivGeneration(); + const runId = `manual-${userId}-${updateId}`; + const [schedule, immediate] = await Promise.allSettled([ + pixivApp.spawn( + "scheduled-pixiv-bookmarks", + { generation, userId, limit: 300 }, + { + idempotencyKey: `scheduled-pixiv-${userId}-${generation}`, + maxAttempts: 3, + retryStrategy: RETRY.pixiv, + }, + ), + pixivApp.spawn( + "pixiv-bookmarks", + { userId, runId, count: 0, limit: 300 }, + { + idempotencyKey: `pixiv-${userId}-${runId}`, + maxAttempts: 3, + retryStrategy: RETRY.pixiv, + }, + ), + ]); + + if (schedule.status === "rejected") { + logger.warn( + { error: schedule.reason, userId, provider: "pixiv" }, + "Failed to schedule collector", + ); + } + if (immediate.status === "rejected") { + logger.warn( + { error: immediate.reason, userId, provider: "pixiv" }, + "Failed to start collector", + ); + } + + return { + immediateStarted: immediate.status === "fulfilled", + scheduleReady: schedule.status === "fulfilled", + }; +} + +privateChat.command("scrapper", async (ctx) => { + const user = ctx.user!; + const connections = await getScrapperConnections(ctx); + const hasConnectedProvider = + connections.twitter === "connected" || connections.pixiv === "connected"; + + if (!hasConnectedProvider) { + const lookupFailures = [ + connections.twitter === "lookup-failed" && "• Twitter: connection check failed.", + connections.pixiv === "lookup-failed" && "• Pixiv: connection check failed.", + ].filter((line) => line !== false); + const keyboard = new InlineKeyboard().webApp("Connect providers", { + url: `${env.BASE_FRONTEND_URL}/settings`, + }); + await ctx.reply( + lookupFailures.length > 0 + ? [ + "Collection was not started:", + ...lookupFailures, + "Try /scrapper again. Connect any disconnected providers in Settings.", + ].join("\n") + : "Connect Twitter or Pixiv in Settings before starting collection.", + { reply_markup: keyboard }, + ); + } else { + const [twitter, pixiv] = await Promise.all([ + connections.twitter === "connected" + ? startTwitterCollection(user.id, ctx.update.update_id, ctx.logger) + : Promise.resolve(null), + connections.pixiv === "connected" + ? startPixivCollection(user.id, ctx.update.update_id, ctx.logger) + : Promise.resolve(null), + ]); + const results = [ + twitter && { provider: "Twitter", ...twitter }, + pixiv && { provider: "Pixiv", ...pixiv }, + ].filter((result) => result !== null); + const lookupFailures = [ + connections.twitter === "lookup-failed" && + "• Twitter: connection check failed; collection was not started.", + connections.pixiv === "lookup-failed" && + "• Pixiv: connection check failed; collection was not started.", + ].filter((line) => line !== false); + const complete = + lookupFailures.length === 0 && + results.every((result) => result.immediateStarted && result.scheduleReady); + const lines = results.map( + (result) => + `• ${result.provider}: sync ${result.immediateStarted ? "started" : "failed"}; recurring schedule ${result.scheduleReady ? "ready" : "failed"}.`, + ); + + await ctx.reply( + [ + complete ? "Collection started:" : "Collection was only partially started:", + ...lines, + ...lookupFailures, + results.some((result) => result.immediateStarted) + ? "Check your gallery in a few minutes." + : "No immediate sync started. Try /scrapper again.", + ].join("\n"), + { reply_markup: webAppKeyboard("app", "View gallery") }, + ); + } +}); + +export default composer; diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index c87ba430..86b7bd95 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -5,6 +5,8 @@ import "@/services/runtime"; import chatMemberHandler from "@/handlers/chat-member"; import imageHandler from "@/handlers/image"; import messageHandler from "@/handlers/message"; +import pixivHandler from "@/handlers/pixiv"; +import scrapperHandler from "@/handlers/scrapper"; import startHandler from "@/handlers/start"; import tweetImageHandler from "@/handlers/tweet-image"; import videoHandler from "@/handlers/video"; @@ -12,8 +14,9 @@ import { logger } from "@/logger"; import { QUEUES } from "@/queue/absurd"; import { classificationApp } from "@/queue/classification"; import { embeddingsApp } from "@/queue/embeddings"; -import { imagesApp } from "@/queue/image-collector"; +import { mediaCollectorApp } from "@/queue/media-collector"; import { memoryApp } from "@/queue/memory"; +import { pixivApp } from "@/queue/pixiv"; import { scrapperApp } from "@/queue/scrapper"; initTelemetry(); @@ -31,27 +34,31 @@ const boundary = bot.errorBoundary((error) => { }); boundary.use(videoHandler); +boundary.use(pixivHandler); boundary.use(tweetImageHandler); +boundary.use(scrapperHandler); boundary.use(imageHandler); boundary.use(messageHandler); boundary.use(startHandler); boundary.use(chatMemberHandler); await Promise.all( - [imagesApp, classificationApp, embeddingsApp, scrapperApp, memoryApp].map(async (app) => { - await app.createQueue(); - await app.setQueuePolicy(undefined, { - cleanupLimit: 2000, - cleanupTtl: "1 day", - }); - }), + [mediaCollectorApp, classificationApp, embeddingsApp, scrapperApp, memoryApp, pixivApp].map( + async (app) => { + await app.createQueue(); + await app.setQueuePolicy(undefined, { + cleanupLimit: 2000, + cleanupTtl: "1 day", + }); + }, + ), ); const workers = await Promise.all([ - imagesApp.startWorker({ + mediaCollectorApp.startWorker({ batchSize: 3, concurrency: 3, - onError: (error) => logger.error({ err: error }, "Images worker error"), - workerId: QUEUES.images, + onError: (error) => logger.error({ err: error }, "Media collector worker error"), + workerId: QUEUES.media, }), classificationApp.startWorker({ batchSize: 1, @@ -80,8 +87,21 @@ const workers = await Promise.all([ onError: (error) => logger.error({ err: error }, "Memory worker error"), workerId: QUEUES.memory, }), + pixivApp.startWorker({ + batchSize: 1, + concurrency: 1, + onError: (error) => logger.error({ err: error }, "Pixiv worker error"), + workerId: QUEUES.pixiv, + }), ]); -const queueApps = [imagesApp, classificationApp, embeddingsApp, scrapperApp, memoryApp]; +const queueApps = [ + mediaCollectorApp, + classificationApp, + embeddingsApp, + scrapperApp, + memoryApp, + pixivApp, +]; const runner = run(bot); logger.info("Bot is running..."); diff --git a/apps/server/src/queue/absurd.ts b/apps/server/src/queue/absurd.ts index 9a0e9842..669ceabe 100644 --- a/apps/server/src/queue/absurd.ts +++ b/apps/server/src/queue/absurd.ts @@ -40,15 +40,18 @@ export const absurdLogger = { export const QUEUES = { classification: "classification", embeddings: "embeddings", - images: "images-collector", + // Persisted queue/task name; keep stable for jobs created before the provider-neutral rename. + media: "images-collector", memory: "chat-memory", + pixiv: "pixiv-bookmarks", scrapper: "feed-scrapper", } as const; export const RETRY = { classification: { kind: "exponential", baseSeconds: 30, factor: 2 } satisfies RetryStrategy, embeddings: { kind: "exponential", baseSeconds: 30, factor: 2 } satisfies RetryStrategy, - images: { kind: "exponential", baseSeconds: 10, factor: 2 } satisfies RetryStrategy, + media: { kind: "exponential", baseSeconds: 10, factor: 2 } satisfies RetryStrategy, memory: { kind: "exponential", baseSeconds: 20, factor: 2 } satisfies RetryStrategy, + pixiv: { kind: "exponential", baseSeconds: 150, factor: 2 } satisfies RetryStrategy, scrapper: { kind: "exponential", baseSeconds: 150, factor: 2 } satisfies RetryStrategy, } as const; diff --git a/apps/server/src/queue/classification-recovery.ts b/apps/server/src/queue/classification-recovery.ts new file mode 100644 index 00000000..3a7c7dad --- /dev/null +++ b/apps/server/src/queue/classification-recovery.ts @@ -0,0 +1,66 @@ +interface ClassificationQueue { + spawn( + name: string, + data: { photoId: string; provider: string; userId: string }, + options: { + idempotencyKey: string; + maxAttempts: number; + retryStrategy: unknown; + }, + ): Promise<{ created: boolean; taskID: string }>; + fetchTaskResult(taskID: string): Promise<{ state: string } | null | undefined>; + retryTask(taskID: string): Promise; +} + +interface ClassificationRecoveryDependencies { + classificationApp: ClassificationQueue; + retryStrategy: unknown; + logger: { + info(context: Record, message: string): void; + }; +} + +export async function enqueueClassification( + { classificationApp, retryStrategy, logger }: ClassificationRecoveryDependencies, + photoId: string, + provider: string, + userId: string, +) { + const idempotencyKey = `classify-${provider}-${userId}-${photoId}`; + const task = await classificationApp.spawn( + "classification", + { photoId, provider, userId }, + { + idempotencyKey, + maxAttempts: 5, + retryStrategy, + }, + ); + + if (task.created) { + return; + } + + const result = await classificationApp.fetchTaskResult(task.taskID); + if (result?.state !== "failed") { + return; + } + + // retryTask atomically changes the failed task back to pending. This keeps + // its idempotency key and lets every later terminal failure recover again. + // Omitting maxAttempts adds one attempt beyond the terminal task's count. + try { + await classificationApp.retryTask(task.taskID); + } catch (error) { + // Another collector may have recovered this task after our snapshot. Once + // it is no longer failed, that recovery is the coalesced outcome. + const latestResult = await classificationApp.fetchTaskResult(task.taskID); + if (!latestResult || latestResult.state === "failed") { + throw error; + } + } + logger.info( + { photoId, provider, userId, taskId: task.taskID }, + "Classification recovery enqueued", + ); +} diff --git a/apps/server/src/queue/classification.ts b/apps/server/src/queue/classification.ts index 0c353348..00e5d4a0 100644 --- a/apps/server/src/queue/classification.ts +++ b/apps/server/src/queue/classification.ts @@ -8,6 +8,7 @@ import type { Classification } from "@/types"; interface ClassificationJobData { photoId: string; + provider?: string; requestId?: string; userId: string; } @@ -27,6 +28,7 @@ classificationApp.registerTask( } const { photoId, userId, requestId: incomingRequestId } = params; + const provider = params.provider ?? "twitter"; const requestId = incomingRequestId || Bun.randomUUIDv7(); if (!(env.ML_BASE_URL && env.ML_API_TOKEN)) { @@ -37,8 +39,8 @@ classificationApp.registerTask( logger.info({ photoId, userId, requestId }, "Classifying photo"); // Fetch photo record to get URL - const photo = await prisma.photo.findUnique({ - where: { photoId: { id: photoId, userId } }, + const photo = await prisma.media.findUnique({ + where: { mediaId: { id: photoId, userId, provider } }, select: { id: true, userId: true, @@ -101,16 +103,16 @@ classificationApp.registerTask( throw error; } - await prisma.photo.update({ - where: { photoId: { id: photoId, userId } }, + await prisma.media.update({ + where: { mediaId: { id: photoId, userId, provider } }, data: { classification: data }, }); await embeddingsApp.spawn( "embeddings", - { photoId, userId, requestId }, + { photoId, provider, userId, requestId }, { - idempotencyKey: `embed-${photoId}-${userId}`, + idempotencyKey: `embed-${provider}-${photoId}-${userId}`, maxAttempts: 5, retryStrategy: RETRY.embeddings, }, diff --git a/apps/server/src/queue/embeddings.ts b/apps/server/src/queue/embeddings.ts index 538c52dc..fdd5ed09 100644 --- a/apps/server/src/queue/embeddings.ts +++ b/apps/server/src/queue/embeddings.ts @@ -7,6 +7,7 @@ import { runtime } from "@/services/runtime"; interface ClassificationJobData { photoId: string; + provider?: string; requestId?: string; userId: string; } @@ -24,6 +25,7 @@ embeddingsApp.registerTask({ name: "embeddings" }, async } const { photoId, userId, requestId: incomingRequestId } = params; + const provider = params.provider ?? "twitter"; const requestId = incomingRequestId || Bun.randomUUIDv7(); if (!(env.ML_BASE_URL && env.ML_API_TOKEN)) { @@ -33,9 +35,9 @@ embeddingsApp.registerTask({ name: "embeddings" }, async logger.info({ photoId, userId, requestId }, "Generating photo embeddings"); - const photo = await prisma.photo.findUnique({ + const photo = await prisma.media.findUnique({ where: { - photoId: { id: photoId, userId }, + mediaId: { id: photoId, userId, provider }, classification: { not: DbNull }, }, select: { @@ -76,7 +78,7 @@ embeddingsApp.registerTask({ name: "embeddings" }, async const imageVecStr = `[${(result.image ?? []).join(",")}]`; await prisma.$executeRaw( - Prisma.sql`UPDATE photos SET tag_vec = ${textVecStr}::vector, image_vec = ${imageVecStr}::vector WHERE id = ${photoId} AND user_id = ${userId}`, + Prisma.sql`UPDATE media SET tag_vec = ${textVecStr}::vector, image_vec = ${imageVecStr}::vector WHERE external_id = ${photoId} AND user_id = ${userId} AND provider = ${provider}`, ); logger.info({ photoId, userId, requestId }, "Photo embeddings generated"); diff --git a/apps/server/src/queue/image-collector.ts b/apps/server/src/queue/image-collector.ts deleted file mode 100644 index bfc2ed77..00000000 --- a/apps/server/src/queue/image-collector.ts +++ /dev/null @@ -1,176 +0,0 @@ -import { Absurd } from "absurd-sdk"; -import { env, prisma } from "@starlight/utils"; -import { http } from "@starlight/utils/http"; -import type { Tweet } from "@the-convocation/twitter-scraper"; -import UserAgent from "user-agents"; -import { logger } from "@/logger"; -import { absurdLogger, QUEUES, RETRY } from "@/queue/absurd"; -import { classificationApp } from "@/queue/classification"; -import { findDuplicatesByImageContent } from "@/services/duplicate-detection"; -import { calculatePerceptualHash } from "@/services/image"; -import { s3 } from "@/storage"; - -export const imagesApp = new Absurd({ - db: env.DATABASE_URL, - log: absurdLogger, - queueName: QUEUES.images, -}); - -export interface ImageCollectorJobData { - tweet: Tweet; - // From database - userId: string; -} - -imagesApp.registerTask({ name: "images-collector" }, async (data) => { - const { tweet, userId } = data; - - // Tweet guaranteed to have IDs, fucking types - const id = tweet.id!; - - logger.info({ tweetId: tweet.id, userId }, "Processing tweet"); - - if (tweet.photos.length === 0) { - logger.debug({ tweetId: tweet.id, userId }, "Tweet has no photos, skipping job"); - return; - } - - const userAgent = new UserAgent(); - - // We can safely update Tweet record here, because we created Tweet object in scrapper queue - const tweetRecord = await prisma.tweet.update({ - where: { tweetId: { userId, id } }, - data: { - tweetData: tweet, - photos: { - createMany: { - data: tweet.photos.map((photo) => ({ - id: photo.id, - originalUrl: photo.url, - })), - // Guaranteed that if we'll restart a job then we won't have additional photos in Tweet relation - skipDuplicates: true, - }, - }, - }, - include: { - photos: true, - }, - }); - - logger.info( - { tweetId: tweet.id, userId, photos: tweetRecord.photos.length }, - "Tweet upserted with photos", - ); - - const refreshedPhotoIds = new Set(); - const refreshTimestamp = new Date(); - - for (const photo of tweetRecord.photos) { - if (photo.s3Path && photo.perceptualHash) { - logger.debug( - { - tweetId: tweet.id, - photoId: photo.id, - userId, - }, - "Photo already downloaded; skipping", - ); - continue; - } - - const response = await http(photo.originalUrl, { - headers: { - "User-Agent": userAgent.toString(), - }, - }); - - if (!response.ok) { - logger.error( - { - tweetId: tweet.id, - photoUrl: photo.originalUrl, - status: response.status, - userId, - }, - "Failed to fetch photo", - ); - throw new Error(`Failed to fetch photo ${photo.originalUrl}`); - } - - const imageBuffer = await response.arrayBuffer(); - - const similarPhotos = await findDuplicatesByImageContent(imageBuffer); - - if (similarPhotos.length > 0) { - const existingPhoto = similarPhotos.find((similarPhoto) => similarPhoto.userId === userId); - - if (existingPhoto && !refreshedPhotoIds.has(existingPhoto.id)) { - await prisma.photo.update({ - where: { photoId: { id: existingPhoto.id, userId } }, - data: { updatedAt: refreshTimestamp }, - }); - refreshedPhotoIds.add(existingPhoto.id); - } - - logger.info( - { - tweetId: tweet.id, - photoId: photo.id, - userId, - refreshedPhotoId: existingPhoto?.id, - similarPhotos, - }, - "Found similar photos, skipping saving photo", - ); - continue; - } - - const extension = photo.originalUrl.split(".").pop() ?? "jpg"; - - const photoName = `${photo.externalId}.${extension}`; - - const [, hash, metadata] = await Promise.all([ - s3.write(`media/${photoName}`, imageBuffer), - calculatePerceptualHash(imageBuffer), - new Bun.Image(imageBuffer).metadata().catch(() => ({ height: null, width: null })), - ]); - - await prisma.photo.update({ - where: { photoId: { id: photo.id, userId } }, - data: { - perceptualHash: hash, - s3Path: `media/${photoName}`, - height: metadata.height, - width: metadata.width, - }, - }); - - // Enqueue classification job - try { - await classificationApp.spawn( - "classification", - { photoId: photo.id, userId }, - { - idempotencyKey: `classify-${photo.id}-${userId}`, - maxAttempts: 5, - retryStrategy: RETRY.classification, - }, - ); - } catch (error) { - logger.error( - { err: error, photoId: photo.id, userId }, - "Failed to enqueue classification job", - ); - } - - logger.info( - { - tweetId: tweet.id, - photoId: photo.id, - userId, - }, - "Photo saved to S3", - ); - } -}); diff --git a/apps/server/src/queue/media-collector.test.ts b/apps/server/src/queue/media-collector.test.ts new file mode 100644 index 00000000..e49d74f1 --- /dev/null +++ b/apps/server/src/queue/media-collector.test.ts @@ -0,0 +1,46 @@ +import { beforeEach, describe, expect, mock, test } from "bun:test"; +import { enqueueClassification } from "@/queue/classification-recovery"; + +const classificationSpawn = mock(); +const fetchTaskResult = mock(); +const retryTask = mock(); +const loggerInfo = mock(); + +describe("media collector classification recovery", () => { + beforeEach(() => { + classificationSpawn.mockReset(); + fetchTaskResult.mockReset(); + retryTask.mockReset(); + loggerInfo.mockReset(); + }); + + test("recovers each terminal classification failure", async () => { + classificationSpawn.mockResolvedValue({ taskID: "failed-task", created: false }); + fetchTaskResult + .mockResolvedValueOnce({ state: "failed", failure: null }) + .mockResolvedValueOnce({ state: "failed", failure: null }) + .mockResolvedValueOnce({ state: "pending" }); + retryTask.mockResolvedValue({ taskID: "failed-task", created: false }); + + const dependencies = { + classificationApp: { spawn: classificationSpawn, fetchTaskResult, retryTask }, + retryStrategy: { kind: "exponential", baseSeconds: 30, factor: 2 }, + logger: { info: loggerInfo }, + }; + await enqueueClassification(dependencies, "media-1", "twitter", "user-1"); + // The first in-place recovery terminally fails; the next collection + // creates one more recovery on that same task. + await enqueueClassification(dependencies, "media-1", "twitter", "user-1"); + // Once that recovery is pending, further collectors only coalesce onto it. + await enqueueClassification(dependencies, "media-1", "twitter", "user-1"); + + expect(classificationSpawn).toHaveBeenCalledTimes(3); + expect(retryTask).toHaveBeenCalledTimes(2); + expect(retryTask).toHaveBeenNthCalledWith(1, "failed-task"); + expect(retryTask).toHaveBeenNthCalledWith(2, "failed-task"); + expect(classificationSpawn.mock.calls[0]?.[2]).toMatchObject({ + idempotencyKey: "classify-twitter-user-1-media-1", + maxAttempts: 5, + }); + }); +}); diff --git a/apps/server/src/queue/media-collector.ts b/apps/server/src/queue/media-collector.ts new file mode 100644 index 00000000..86649ce9 --- /dev/null +++ b/apps/server/src/queue/media-collector.ts @@ -0,0 +1,192 @@ +import { Absurd } from "absurd-sdk"; +import { env, prisma } from "@starlight/utils"; +import sharp from "sharp"; +import { logger } from "@/logger"; +import { absurdLogger, QUEUES, RETRY } from "@/queue/absurd"; +import { classificationApp } from "@/queue/classification"; +import { enqueueClassification } from "@/queue/classification-recovery"; +import { findSimilarPhotos } from "@/services/duplicate-detection"; +import { calculatePerceptualHash } from "@/services/image"; +import { normalizeCollectorTags } from "@/services/collector-tags"; +import { isMediaResolved, resolveMediaFromAsset } from "@/services/media-resolution"; +import { + MAX_MEDIA_DOWNLOAD_BYTES, + MAX_POST_DOWNLOAD_BYTES, + readResponseBounded, +} from "@/services/media-download"; +import { s3 } from "@/storage"; + +export const mediaCollectorApp = new Absurd({ + db: env.DATABASE_URL, + log: absurdLogger, + queueName: QUEUES.media, +}); + +export interface MediaCollectorJobData { + userId: string; + post: { + provider: string; + externalId: string; + sourceUrl: string; + authorExternalId?: string; + authorName?: string; + authorUsername?: string; + title?: string; + text?: string; + tags?: string[]; + providerPayload: object; + media: Array<{ + externalId: string; + url: string; + kind?: string; + position: number; + fetchHeaders?: Record; + }>; + }; +} + +mediaCollectorApp.registerTask( + { name: "images-collector" }, + async (data) => { + const { post, userId } = data; + const tags = normalizeCollectorTags(post.provider, post.tags, post.providerPayload); + let downloadedBytes = 0; + const postRecord = await prisma.post.upsert({ + where: { postId: { id: post.externalId, userId, provider: post.provider } }, + create: { + id: post.externalId, + userId, + provider: post.provider, + sourceUrl: post.sourceUrl, + authorExternalId: post.authorExternalId, + authorName: post.authorName, + authorUsername: post.authorUsername, + title: post.title, + text: post.text, + tags, + username: post.authorUsername, + providerPayload: post.providerPayload, + media: { + createMany: { + data: post.media.map((media) => ({ + id: media.externalId, + position: media.position, + kind: media.kind ?? "image", + originalUrl: media.url, + })), + skipDuplicates: true, + }, + }, + }, + update: { + sourceUrl: post.sourceUrl, + authorExternalId: post.authorExternalId, + authorName: post.authorName, + authorUsername: post.authorUsername, + title: post.title, + text: post.text, + tags, + username: post.authorUsername, + providerPayload: post.providerPayload, + media: { + createMany: { + data: post.media.map((media) => ({ + id: media.externalId, + position: media.position, + kind: media.kind ?? "image", + originalUrl: media.url, + })), + skipDuplicates: true, + }, + }, + }, + include: { media: true }, + }); + + for (const media of postRecord.media) { + if (isMediaResolved(media)) { + if (media.kind === "image" && media.classification === null) { + await enqueueClassification( + { classificationApp, retryStrategy: RETRY.classification, logger }, + media.id, + post.provider, + userId, + ); + } + continue; + } + const input = post.media.find((item) => item.externalId === media.id); + if (!input) { + throw new Error(`Media ${media.id} is missing from collector payload`); + } + const remainingBytes = MAX_POST_DOWNLOAD_BYTES - downloadedBytes; + if (remainingBytes <= 0) { + throw new Error("Post media is too large"); + } + const bytes = await readResponseBounded( + await fetch(media.originalUrl, { headers: input.fetchHeaders }), + Math.min(MAX_MEDIA_DOWNLOAD_BYTES, remainingBytes), + ); + downloadedBytes += bytes.byteLength; + const extension = new URL(media.originalUrl).pathname.split(".").at(-1) ?? "jpg"; + const mediaPath = `media/${post.provider}/${userId}/${media.id}.${extension}`; + + if (media.kind !== "image") { + await s3.write(mediaPath, bytes); + await prisma.media.update({ + where: { mediaId: { id: media.id, userId, provider: post.provider } }, + data: { s3Path: mediaPath }, + }); + continue; + } + + const hash = await calculatePerceptualHash(bytes); + const duplicates = await findSimilarPhotos(hash); + if (duplicates.length > 0) { + const asset = duplicates[0]!; + await prisma.media.update({ + where: { mediaId: { id: media.id, userId, provider: post.provider } }, + data: resolveMediaFromAsset(asset), + }); + logger.info( + { + mediaId: media.id, + provider: post.provider, + userId, + assetMediaId: asset.id, + assetUserId: asset.userId, + }, + "Duplicate media resolved from existing asset", + ); + await enqueueClassification( + { classificationApp, retryStrategy: RETRY.classification, logger }, + media.id, + post.provider, + userId, + ); + continue; + } + const [, metadata] = await Promise.all([ + s3.write(mediaPath, bytes), + sharp(bytes) + .metadata() + .catch(() => ({ height: null, width: null })), + ]); + await prisma.media.update({ + where: { mediaId: { id: media.id, userId, provider: post.provider } }, + data: { + perceptualHash: hash, + s3Path: mediaPath, + height: metadata.height, + width: metadata.width, + }, + }); + await enqueueClassification( + { classificationApp, retryStrategy: RETRY.classification, logger }, + media.id, + post.provider, + userId, + ); + } + }, +); diff --git a/apps/server/src/queue/pixiv.ts b/apps/server/src/queue/pixiv.ts new file mode 100644 index 00000000..028c3713 --- /dev/null +++ b/apps/server/src/queue/pixiv.ts @@ -0,0 +1,182 @@ +import { Absurd } from "absurd-sdk"; +import { withPixivClient } from "@starlight/api/services/pixiv-credential"; +import { env, prisma } from "@starlight/utils"; +import { absurdLogger, QUEUES, RETRY } from "@/queue/absurd"; +import { mediaCollectorApp, type MediaCollectorJobData } from "@/queue/media-collector"; +import { mediaResolvedWhere } from "@/services/media-resolution"; + +const CONSECUTIVE_THRESHOLD = 15; +const SCHEDULE_INTERVAL_SECONDS = 60 * 60 * 6; + +export const pixivApp = new Absurd({ + db: env.DATABASE_URL, + log: absurdLogger, + queueName: QUEUES.pixiv, +}); + +export interface PixivCrawlJobData { + userId: string; + runId: string; + count: number; + limit: number; + cursor?: number; + visibility?: "public" | "private"; +} + +export interface ScheduledPixivJobData { + generation: number; + limit: number; + userId: string; +} + +export const getScheduledPixivGeneration = (date = new Date()) => + Math.floor(date.getTime() / (SCHEDULE_INTERVAL_SECONDS * 1000)); + +pixivApp.registerTask( + { name: "scheduled-pixiv-bookmarks" }, + async (data, ctx) => { + await ctx.sleepFor("next-run", SCHEDULE_INTERVAL_SECONDS); + try { + const credential = await prisma.providerCredential.findUnique({ + where: { userId_provider: { userId: data.userId, provider: "pixiv" } }, + select: { credentialType: true }, + }); + if (credential?.credentialType === "refresh_token") { + await pixivApp.spawn( + "pixiv-bookmarks", + { + count: 0, + limit: data.limit, + runId: `scheduled-${data.generation}`, + userId: data.userId, + }, + { + idempotencyKey: `scheduled-pixiv-run-${data.userId}-${data.generation}`, + maxAttempts: 3, + retryStrategy: RETRY.pixiv, + }, + ); + } + } finally { + const nextGeneration = data.generation + 1; + await pixivApp.spawn( + "scheduled-pixiv-bookmarks", + { ...data, generation: nextGeneration }, + { + idempotencyKey: `scheduled-pixiv-${data.userId}-${nextGeneration}`, + maxAttempts: 3, + retryStrategy: RETRY.pixiv, + }, + ); + } + }, +); + +pixivApp.registerTask({ name: "pixiv-bookmarks" }, async (data) => { + if (!data.visibility) { + const user = await prisma.user.findUnique({ + where: { id: data.userId }, + select: { + pixivIncludePrivate: true, + providerCredentials: { + where: { provider: "pixiv", credentialType: "refresh_token" }, + }, + }, + }); + if (!user?.providerCredentials.length) { + return; + } + const visibilities: Array<"public" | "private"> = ["public"]; + if (user.pixivIncludePrivate) { + visibilities.push("private"); + } + await Promise.all( + visibilities.map((visibility) => + pixivApp.spawn( + "pixiv-bookmarks", + { ...data, count: 0, cursor: undefined, visibility }, + { + idempotencyKey: `pixiv-${data.userId}-${data.runId}-${visibility}-start`, + maxAttempts: 3, + retryStrategy: RETRY.pixiv, + }, + ), + ), + ); + return; + } + + const page = await withPixivClient(data.userId, (client) => + client.bookmarks({ cursor: data.cursor, visibility: data.visibility! }), + ); + if (!page) { + return; + } + const known = new Set( + ( + await prisma.post.findMany({ + where: { + userId: data.userId, + provider: "pixiv", + id: { in: page.artworks.map((artwork) => artwork.id) }, + media: { every: mediaResolvedWhere }, + }, + select: { id: true }, + }) + ).map((post) => post.id), + ); + + let consecutiveKnown = 0; + const jobs: MediaCollectorJobData[] = []; + for (const artwork of page.artworks) { + consecutiveKnown = known.has(artwork.id) ? consecutiveKnown + 1 : 0; + jobs.push({ + userId: data.userId, + post: { + provider: "pixiv", + externalId: artwork.id, + sourceUrl: artwork.sourceUrl, + authorExternalId: artwork.author.id, + authorName: artwork.author.name, + authorUsername: artwork.author.username, + title: artwork.title, + text: artwork.caption, + tags: artwork.tags, + providerPayload: { starlightMediaType: artwork.type }, + media: artwork.mediaUrls.map((url, position) => ({ + externalId: `${artwork.id}:${position}`, + url, + position, + kind: artwork.type === "ugoira" ? "animation-preview" : "image", + fetchHeaders: { Referer: "https://www.pixiv.net/" }, + })), + }, + }); + if (consecutiveKnown >= CONSECUTIVE_THRESHOLD) { + break; + } + } + await Promise.all( + jobs.map((job) => + mediaCollectorApp.spawn("images-collector", job, { + idempotencyKey: `media-pixiv-${data.userId}-${job.post.externalId}`, + maxAttempts: 3, + retryStrategy: RETRY.media, + }), + ), + ); + + const count = data.count + page.artworks.length; + if (consecutiveKnown >= CONSECUTIVE_THRESHOLD || count >= data.limit || !page.nextCursor) { + return; + } + await pixivApp.spawn( + "pixiv-bookmarks", + { ...data, count, cursor: page.nextCursor }, + { + idempotencyKey: `pixiv-${data.userId}-${data.runId}-${data.visibility}-${page.nextCursor}`, + maxAttempts: 3, + retryStrategy: RETRY.pixiv, + }, + ); +}); diff --git a/apps/server/src/queue/scrapper.ts b/apps/server/src/queue/scrapper.ts index 7b09bf2f..610cb18c 100644 --- a/apps/server/src/queue/scrapper.ts +++ b/apps/server/src/queue/scrapper.ts @@ -1,19 +1,17 @@ import { Absurd } from "absurd-sdk"; -import { CookieEncryption } from "@starlight/crypto"; +import { getTwitterCookies, hasTwitterCookies } from "@starlight/api/services/twitter-credential"; import type { User } from "@starlight/utils"; import { env, prisma } from "@starlight/utils"; -import { type QueryTweetsResponse, Scraper, type Tweet } from "@the-convocation/twitter-scraper"; +import { type QueryTweetsResponse, Scraper } from "@the-convocation/twitter-scraper"; import { bot } from "@/bot"; import { logger } from "@/logger"; import { absurdLogger, QUEUES, RETRY } from "@/queue/absurd"; -import { imagesApp } from "@/queue/image-collector"; +import { mediaCollectorApp } from "@/queue/media-collector"; +import type { MediaCollectorJobData } from "@/queue/media-collector"; +import { mediaResolvedWhere } from "@/services/media-resolution"; +import { normalizeTwitterTags } from "@/services/twitter-tags"; import { Cookies } from "@/storage"; -const cookieEncryption = new CookieEncryption( - env.COOKIE_ENCRYPTION_KEY, - env.COOKIE_ENCRYPTION_SALT, -); - export const SCHEDULED_SCRAPPER_INTERVAL_SECONDS = 60 * 60 * 6; export function getScheduledScrapperGeneration(date = new Date()) { @@ -64,12 +62,7 @@ scrapperApp.registerTask( await ctx.sleepFor("next-run", SCHEDULED_SCRAPPER_INTERVAL_SECONDS); try { - const user = await prisma.user.findUnique({ - where: { id: data.userId }, - select: { cookies: true }, - }); - - if (!user?.cookies) { + if (!(await hasTwitterCookies(data.userId))) { logger.info({ userId: data.userId }, "Skipping scheduled scrapper: user has no cookies"); } else { await ctx.step("spawn-run", () => @@ -111,7 +104,6 @@ scrapperApp.registerTask({ name: "feed-scrapper" }, async (data logger.info({ userId, cursor: data.cursor, jobData: data }, "Scraping timeline"); let user: User; - try { user = await prisma.user.findUniqueOrThrow({ where: { @@ -123,7 +115,13 @@ scrapperApp.registerTask({ name: "feed-scrapper" }, async (data throw error; } - const userCookies = user.cookies; + let userCookies: string | undefined; + try { + userCookies = await getTwitterCookies(user.id); + } catch (error) { + logger.error({ err: error, userId }, "Failed to decrypt Twitter cookies"); + throw new Error("Failed to decrypt Twitter cookies"); + } if (!userCookies) { logger.error({ userId }, "User cookies not found"); @@ -136,16 +134,7 @@ scrapperApp.registerTask({ name: "feed-scrapper" }, async (data return; } - // Decrypt cookies with migration support - let cookiesJson: string; - try { - cookiesJson = cookieEncryption.safeDecrypt(userCookies, user.telegramId.toString()); - } catch (error) { - logger.error({ err: error, userId }, "Failed to decrypt user cookies"); - throw new Error("Failed to decrypt user cookies"); - } - - const cookies = Cookies.fromJSON(cookiesJson); + const cookies = Cookies.fromJSON(userCookies); const twid = cookies.userId(); @@ -189,11 +178,12 @@ scrapperApp.registerTask({ name: "feed-scrapper" }, async (data const existingTweetMap = new Map( ( - await prisma.tweet.findMany({ + await prisma.post.findMany({ where: { userId, + provider: "twitter", id: { in: tweetIds }, - photos: { every: { s3Path: { not: null } } }, + media: { every: mediaResolvedWhere }, }, select: { id: true, createdAt: true }, }) @@ -201,13 +191,7 @@ scrapperApp.registerTask({ name: "feed-scrapper" }, async (data ); // Step 2: Process tweets and build batch operations - const newTweets: Array<{ - id: string; - userId: string; - tweetData: Tweet; - }> = []; - const updatedTweets: Array<{ id: string; tweetData: Tweet }> = []; - const tweetsToQueue: Array<{ tweet: Tweet; userId: string }> = []; + const tweetsToQueue: MediaCollectorJobData[] = []; let consecutiveKnownTweets = 0; let newTweetsInBatch = 0; @@ -222,22 +206,32 @@ scrapperApp.registerTask({ name: "feed-scrapper" }, async (data if (isNewTweet) { consecutiveKnownTweets = 0; newTweetsInBatch++; - newTweets.push({ - id: tweet.id, - userId, - tweetData: tweet, - }); } else { consecutiveKnownTweets++; - updatedTweets.push({ - id: tweet.id, - tweetData: tweet, - }); } // Only queue tweets with photos for image processing if (tweet.photos.length > 0) { - tweetsToQueue.push({ tweet, userId }); + tweetsToQueue.push({ + userId, + post: { + provider: "twitter", + externalId: tweet.id, + sourceUrl: `https://x.com/i/status/${tweet.id}`, + authorExternalId: tweet.userId, + authorName: tweet.name, + authorUsername: tweet.username, + text: tweet.text, + tags: normalizeTwitterTags(tweet), + providerPayload: tweet, + media: tweet.photos.map((photo, position) => ({ + externalId: photo.id, + url: photo.url, + position, + kind: "image", + })), + }, + }); } // Stop if we've seen too many consecutive known tweets (unless force is enabled) @@ -255,37 +249,14 @@ scrapperApp.registerTask({ name: "feed-scrapper" }, async (data } } - // Step 3: Execute batch operations in transaction - await prisma.$transaction(async (tx) => { - // Batch create new tweets - if (newTweets.length > 0) { - await tx.tweet.createMany({ - data: newTweets, - skipDuplicates: true, - }); - } - - // Batch update existing tweets - if (updatedTweets.length > 0) { - await Promise.all( - updatedTweets.map((tweet) => - tx.tweet.update({ - where: { tweetId: { userId, id: tweet.id } }, - data: { tweetData: tweet.tweetData }, - }), - ), - ); - } - }); - // Queue image processing jobs for tweets with photos if (tweetsToQueue.length > 0) { await Promise.all( tweetsToQueue.map((job) => - imagesApp.spawn("images-collector", job, { - idempotencyKey: `post-${job.tweet.id}-${job.userId}`, + mediaCollectorApp.spawn("images-collector", job, { + idempotencyKey: `media-twitter-${job.userId}-${job.post.externalId}`, maxAttempts: 3, - retryStrategy: RETRY.images, + retryStrategy: RETRY.media, }), ), ); diff --git a/apps/server/src/scripts/classification.ts b/apps/server/src/scripts/classification.ts index a6826536..62ef9d2d 100644 --- a/apps/server/src/scripts/classification.ts +++ b/apps/server/src/scripts/classification.ts @@ -48,14 +48,14 @@ async function main() { } const photos = ALL_PICTURES - ? await prisma.photo.findMany({ + ? await prisma.media.findMany({ where: { deletedAt: null, s3Path: { not: null } }, - select: { id: true, userId: true }, + select: { id: true, provider: true, userId: true }, orderBy: { id: "asc" }, }) - : await prisma.$queryRaw<{ id: string; userId: string }[]>` - SELECT id, user_id as "userId" - FROM photos + : await prisma.$queryRaw<{ id: string; provider: string; userId: string }[]>` + SELECT external_id AS id, provider, user_id as "userId" + FROM media WHERE deleted_at IS NULL AND s3_path IS NOT NULL AND ( @@ -74,7 +74,7 @@ async function main() { if (!DRY_RUN && photos.length > 0) { await Promise.all( photos.map((photo) => { - const data = { photoId: photo.id, userId: photo.userId }; + const data = { photoId: photo.id, provider: photo.provider, userId: photo.userId }; return FORCE ? classificationApp.spawn("classification", data, { @@ -82,7 +82,7 @@ async function main() { retryStrategy: RETRY.classification, }) : classificationApp.spawn("classification", data, { - idempotencyKey: `classify-${data.photoId}-${data.userId}`, + idempotencyKey: `classify-${data.provider}-${data.photoId}-${data.userId}`, maxAttempts: 5, retryStrategy: RETRY.classification, }); diff --git a/apps/server/src/scripts/embeddings.ts b/apps/server/src/scripts/embeddings.ts index b1fd290b..975d51bc 100644 --- a/apps/server/src/scripts/embeddings.ts +++ b/apps/server/src/scripts/embeddings.ts @@ -43,16 +43,16 @@ async function main() { logger.info("Embeddings queue dropped and recreated"); } - const photos = await prisma.$queryRaw<{ id: string; userId: string }[]>` - SELECT id, user_id as "userId" - FROM photos + const photos = await prisma.$queryRaw<{ id: string; provider: string; userId: string }[]>` + SELECT external_id AS id, provider, user_id as "userId" + FROM media WHERE deleted_at IS NULL AND s3_path IS NOT NULL AND ( image_vec IS NULL OR tag_vec IS NULL ) - ORDER BY id ASC + ORDER BY external_id ASC `; let enqueued = 0; @@ -60,7 +60,7 @@ async function main() { if (!DRY_RUN && photos.length > 0) { await Promise.all( photos.map((photo) => { - const data = { photoId: photo.id, userId: photo.userId }; + const data = { photoId: photo.id, provider: photo.provider, userId: photo.userId }; return FORCE ? embeddingsApp.spawn("embeddings", data, { @@ -68,7 +68,7 @@ async function main() { retryStrategy: RETRY.embeddings, }) : embeddingsApp.spawn("embeddings", data, { - idempotencyKey: `embed-${data.photoId}-${data.userId}`, + idempotencyKey: `embed-${data.provider}-${data.photoId}-${data.userId}`, maxAttempts: 5, retryStrategy: RETRY.embeddings, }); diff --git a/apps/server/src/scripts/update-photo-dimensions.ts b/apps/server/src/scripts/update-media-dimensions.ts similarity index 56% rename from apps/server/src/scripts/update-photo-dimensions.ts rename to apps/server/src/scripts/update-media-dimensions.ts index 4d1ee491..08a6dbb9 100644 --- a/apps/server/src/scripts/update-photo-dimensions.ts +++ b/apps/server/src/scripts/update-media-dimensions.ts @@ -2,8 +2,8 @@ import { prisma } from "@starlight/utils"; import { logger } from "@/logger"; import { s3 } from "@/storage"; -// Manual script: update height/width for photos missing dimensions -// Usage: bun run apps/server/src/scripts/update-photo-dimensions.ts +// Manual script: update height/width for media missing dimensions +// Usage: bun run apps/server/src/scripts/update-media-dimensions.ts // Optional env vars: // DRY_RUN=1 (only log, do not update) // BATCH_SIZE=100 (batch size for processing, default: 50) @@ -17,11 +17,11 @@ async function main() { dryRun: DRY_RUN, batchSize: BATCH_SIZE, }, - "Starting photo dimensions update", + "Starting media dimensions update", ); - // Find photos with null height or width that have s3Path - const photos = await prisma.photo.findMany({ + // Find media with null height or width that have s3Path + const media = await prisma.media.findMany({ where: { deletedAt: null, s3Path: { not: null }, @@ -30,6 +30,7 @@ async function main() { select: { id: true, userId: true, + provider: true, s3Path: true, height: true, width: true, @@ -37,10 +38,10 @@ async function main() { orderBy: { createdAt: "asc" }, }); - logger.info({ count: photos.length }, "Found photos missing dimensions"); + logger.info({ count: media.length }, "Found media missing dimensions"); - if (photos.length === 0) { - logger.info("No photos need dimension updates"); + if (media.length === 0) { + logger.info("No media need dimension updates"); return; } @@ -48,27 +49,27 @@ async function main() { let failed = 0; // Process in batches - for (let i = 0; i < photos.length; i += BATCH_SIZE) { - const batch = photos.slice(i, i + BATCH_SIZE); + for (let i = 0; i < media.length; i += BATCH_SIZE) { + const batch = media.slice(i, i + BATCH_SIZE); logger.info( { batch: Math.floor(i / BATCH_SIZE) + 1, - totalBatches: Math.ceil(photos.length / BATCH_SIZE), + totalBatches: Math.ceil(media.length / BATCH_SIZE), }, "Processing batch", ); await Promise.allSettled( - batch.map(async (photo) => { + batch.map(async (mediaItem) => { try { - if (!photo.s3Path) { - logger.warn({ photoId: photo.id, userId: photo.userId }, "Photo has no s3Path"); + if (!mediaItem.s3Path) { + logger.warn({ mediaId: mediaItem.id, userId: mediaItem.userId }, "Media has no s3Path"); return; } // Download image from S3 - const imageBuffer = await s3.file(photo.s3Path).arrayBuffer(); + const imageBuffer = await s3.file(mediaItem.s3Path).arrayBuffer(); const metadata = await new Bun.Image(imageBuffer) .metadata() @@ -76,7 +77,7 @@ async function main() { if (!(metadata.height && metadata.width)) { logger.warn( - { photoId: photo.id, userId: photo.userId, metadata }, + { mediaId: mediaItem.id, userId: mediaItem.userId, metadata }, "Failed to extract dimensions", ); failed++; @@ -84,9 +85,15 @@ async function main() { } if (!DRY_RUN) { - // Update photo with dimensions - await prisma.photo.update({ - where: { photoId: { id: photo.id, userId: photo.userId } }, + // Update media with dimensions + await prisma.media.update({ + where: { + mediaId: { + id: mediaItem.id, + userId: mediaItem.userId, + provider: mediaItem.provider, + }, + }, data: { height: metadata.height, width: metadata.width, @@ -96,20 +103,20 @@ async function main() { logger.debug( { - photoId: photo.id, - userId: photo.userId, + mediaId: mediaItem.id, + userId: mediaItem.userId, height: metadata.height, width: metadata.width, dryRun: DRY_RUN, }, - "Updated photo dimensions", + "Updated media dimensions", ); updated++; } catch (error) { logger.error( - { error, photoId: photo.id, userId: photo.userId }, - "Failed to update photo dimensions", + { error, mediaId: mediaItem.id, userId: mediaItem.userId }, + "Failed to update media dimensions", ); failed++; } @@ -119,19 +126,19 @@ async function main() { logger.info( { - total: photos.length, + total: media.length, updated, failed, dryRun: DRY_RUN, batchSize: BATCH_SIZE, }, - "Finished photo dimensions update", + "Finished media dimensions update", ); } main() .catch((error) => { - logger.error({ error }, "Photo dimensions update script failed"); + logger.error({ error }, "Media dimensions update script failed"); process.exitCode = 1; }) .finally(async () => { diff --git a/apps/server/src/services/collector-tags.ts b/apps/server/src/services/collector-tags.ts new file mode 100644 index 00000000..408b0482 --- /dev/null +++ b/apps/server/src/services/collector-tags.ts @@ -0,0 +1,18 @@ +import { normalizeTags } from "@/services/tag-normalization"; + +export const normalizeCollectorTags = ( + provider: string, + tags: unknown, + providerPayload: object, +): string[] => { + if (Array.isArray(tags)) { + return normalizeTags(tags.filter((tag): tag is string => typeof tag === "string")); + } + if (tags !== undefined || provider !== "twitter") { + return []; + } + const hashtags = (providerPayload as { hashtags?: unknown }).hashtags; + return Array.isArray(hashtags) + ? normalizeTags(hashtags.filter((tag): tag is string => typeof tag === "string")) + : []; +}; diff --git a/apps/server/src/services/duplicate-detection.test.ts b/apps/server/src/services/duplicate-detection.test.ts new file mode 100644 index 00000000..c900c3bd --- /dev/null +++ b/apps/server/src/services/duplicate-detection.test.ts @@ -0,0 +1,48 @@ +import { beforeEach, describe, expect, mock, test } from "bun:test"; + +const findMany = mock(); + +mock.module("@starlight/utils", () => ({ + prisma: { media: { findMany } }, +})); +mock.module("@/logger", () => ({ + logger: { debug: mock() }, +})); + +const { findSimilarPhotos } = await import("@/services/duplicate-detection"); + +const targetHash = "0000000000000000"; + +describe("findSimilarPhotos", () => { + beforeEach(() => { + findMany.mockReset(); + }); + + test("returns matching assets from a full candidate bucket", async () => { + findMany + .mockResolvedValueOnce( + Array.from({ length: 50 }, (_, index) => ({ + id: `media-${index}`, + userId: "user", + perceptualHash: targetHash, + s3Path: `media/twitter/user/${index}.jpg`, + originalUrl: `https://example.test/${index}.jpg`, + postId: `post-${index}`, + height: 100, + width: 100, + post: { sourceUrl: "https://example.test/post" }, + })), + ) + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([]); + + const matches = await findSimilarPhotos(targetHash); + + expect(matches).toHaveLength(50); + expect(matches[0]).toMatchObject({ + perceptualHash: targetHash, + s3Path: "media/twitter/user/0.jpg", + }); + expect(findMany.mock.calls.map(([query]) => query.take)).toEqual([50, 200, 1000]); + }); +}); diff --git a/apps/server/src/services/duplicate-detection.ts b/apps/server/src/services/duplicate-detection.ts index d89891f5..f921d669 100644 --- a/apps/server/src/services/duplicate-detection.ts +++ b/apps/server/src/services/duplicate-detection.ts @@ -7,8 +7,11 @@ interface SimilarPhoto { id: string; originalUrl: string; perceptualHash: string; - s3Path?: string; - tweetId: string; + s3Path: string; + height: number | null; + width: number | null; + sourceUrl: string; + postId: string; userId: string; } @@ -23,16 +26,18 @@ export async function findSimilarPhotos( { len: 8, field: "hashBucket8" as const, maxCandidates: 200 }, { len: 4, field: "hashBucket4" as const, maxCandidates: 1000 }, ]; + const similarPhotos = new Map(); for (const { len, field, maxCandidates } of buckets) { const prefix = targetHash.substring(0, len); logger.debug({ prefix, field, maxCandidates }, "Searching for similar photos"); - const candidates = await prisma.photo.findMany({ + const candidates = await prisma.media.findMany({ where: { [field]: prefix, perceptualHash: { not: null }, + s3Path: { not: null }, deletedAt: null, NOT: excludePhotoId && excludeUserId @@ -47,7 +52,10 @@ export async function findSimilarPhotos( perceptualHash: true, s3Path: true, originalUrl: true, - tweetId: true, + postId: true, + height: true, + width: true, + post: { select: { sourceUrl: true } }, }, take: maxCandidates, }); @@ -56,32 +64,28 @@ export async function findSimilarPhotos( continue; } - // If we got results and didn't hit the limit, process them - if (candidates.length < maxCandidates) { - const similarPhotos: SimilarPhoto[] = []; + for (const candidate of candidates) { + const distance = calculateHashDistance(targetHash, candidate.perceptualHash!); - for (const candidate of candidates) { - const distance = calculateHashDistance(targetHash, candidate.perceptualHash!); - - if (distance <= maxDistance) { - similarPhotos.push({ - id: candidate.id, - userId: candidate.userId, - perceptualHash: candidate.perceptualHash!, - distance, - s3Path: candidate.s3Path || undefined, - originalUrl: candidate.originalUrl, - tweetId: candidate.tweetId, - }); - } + if (distance <= maxDistance) { + const similarPhoto = { + id: candidate.id, + userId: candidate.userId, + perceptualHash: candidate.perceptualHash!, + distance, + s3Path: candidate.s3Path!, + originalUrl: candidate.originalUrl, + postId: candidate.postId, + sourceUrl: candidate.post.sourceUrl, + height: candidate.height, + width: candidate.width, + }; + similarPhotos.set(`${candidate.id}:${candidate.userId}:${candidate.s3Path}`, similarPhoto); } - - // Sort by distance (most similar first) - return similarPhotos.sort((a, b) => a.distance - b.distance); } } - return []; + return [...similarPhotos.values()].sort((a, b) => a.distance - b.distance); } export async function findDuplicatesByImageContent( diff --git a/apps/server/src/services/media-download.ts b/apps/server/src/services/media-download.ts new file mode 100644 index 00000000..3acf701a --- /dev/null +++ b/apps/server/src/services/media-download.ts @@ -0,0 +1,42 @@ +export const MAX_MEDIA_DOWNLOAD_BYTES = 50_000_000; +export const MAX_POST_DOWNLOAD_BYTES = 200_000_000; + +export const readResponseBounded = async (response: Response, limit = MAX_MEDIA_DOWNLOAD_BYTES) => { + if (!response.ok) { + throw new Error(`Media request failed (${response.status})`); + } + const declaredLength = Number(response.headers.get("content-length")); + if (Number.isFinite(declaredLength) && declaredLength > limit) { + await response.body?.cancel(); + throw new Error("Media is too large"); + } + if (!response.body) { + throw new Error("Media response had no body"); + } + const reader = response.body.getReader(); + const chunks: Uint8Array[] = []; + let total = 0; + try { + while (true) { + const chunk = await reader.read(); + if (chunk.done) { + break; + } + total += chunk.value.byteLength; + if (total > limit) { + await reader.cancel("Media is too large"); + throw new Error("Media is too large"); + } + chunks.push(chunk.value); + } + } finally { + reader.releaseLock(); + } + const output = new Uint8Array(total); + let offset = 0; + for (const chunk of chunks) { + output.set(chunk, offset); + offset += chunk.byteLength; + } + return output; +}; diff --git a/apps/server/src/services/media-resolution.ts b/apps/server/src/services/media-resolution.ts new file mode 100644 index 00000000..0aaae438 --- /dev/null +++ b/apps/server/src/services/media-resolution.ts @@ -0,0 +1,24 @@ +import { Prisma, type Media } from "@starlight/utils"; + +export const mediaResolvedWhere = { + s3Path: { not: null }, + OR: [{ kind: { not: "image" } }, { perceptualHash: { not: null } }], +} satisfies Prisma.MediaWhereInput; + +export function isMediaResolved(media: Pick): boolean { + return media.s3Path !== null && (media.kind !== "image" || media.perceptualHash !== null); +} + +export function resolveMediaFromAsset( + asset: Pick & { + perceptualHash: string; + s3Path: string; + }, +) { + return { + s3Path: asset.s3Path, + perceptualHash: asset.perceptualHash, + height: asset.height, + width: asset.width, + }; +} diff --git a/apps/server/src/services/pixiv-media.test.ts b/apps/server/src/services/pixiv-media.test.ts new file mode 100644 index 00000000..9be88bb4 --- /dev/null +++ b/apps/server/src/services/pixiv-media.test.ts @@ -0,0 +1,79 @@ +import { describe, expect, test } from "bun:test"; +import JSZip from "jszip"; +import { readResponseBounded } from "@/services/media-download"; +import { buildFfmpegConcat, extractUgoiraZip, parsePixivArtworkUrl } from "@/services/pixiv-media"; + +describe("parsePixivArtworkUrl", () => { + test("accepts only canonical HTTPS artwork URLs", () => { + expect(parsePixivArtworkUrl("https://www.pixiv.net/artworks/12345")).toBe("12345"); + expect(parsePixivArtworkUrl("https://pixiv.net/artworks/12345/")).toBe("12345"); + expect(parsePixivArtworkUrl("http://www.pixiv.net/artworks/12345")).toBeNull(); + expect(parsePixivArtworkUrl("https://pixiv.net.evil.test/artworks/12345")).toBeNull(); + expect(parsePixivArtworkUrl("https://www.pixiv.net/users/12345")).toBeNull(); + }); +}); + +describe("ugoira conversion input", () => { + test("preserves variable frame delays", () => { + expect( + buildFfmpegConcat([ + { file: "a.jpg", delay: 40 }, + { file: "b.jpg", delay: 125 }, + ]), + ).toContain("duration 0.04\nfile 'b.jpg'\nduration 0.125"); + }); + + test("rejects traversal in frame metadata", async () => { + const zip = new JSZip(); + zip.file("frame.jpg", "data"); + const archive = await zip.generateAsync({ type: "arraybuffer" }); + expect(extractUgoiraZip(archive, [{ file: "../frame.jpg", delay: 100 }])).rejects.toThrow( + "Unsafe ugoira archive path", + ); + }); + + test("bounds declared and streamed download sizes", async () => { + const declared = new Response("small", { + headers: { "content-length": "100" }, + }); + await expect(readResponseBounded(declared, 10)).rejects.toThrow("too large"); + const streamed = new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array(8)); + controller.enqueue(new Uint8Array(8)); + controller.close(); + }, + }), + ); + await expect(readResponseBounded(streamed, 10)).rejects.toThrow("too large"); + }); + + test("bounds cumulative extraction and removes failed temp directories", async () => { + const glob = new Bun.Glob("starlight-ugoira-*"); + const temporaryDirectory = Bun.env.TMPDIR ?? "/tmp"; + const before = new Set(await Array.fromAsync(glob.scan({ cwd: temporaryDirectory }))); + const zip = new JSZip(); + zip.file("a.jpg", new Uint8Array(8)); + zip.file("b.jpg", new Uint8Array(8)); + const archive = await zip.generateAsync({ type: "arraybuffer" }); + expect( + extractUgoiraZip( + archive, + [ + { file: "a.jpg", delay: 100 }, + { file: "b.jpg", delay: 100 }, + ], + { uncompressed: 10 }, + ), + ).rejects.toThrow("too large"); + const after = await Array.fromAsync(glob.scan({ cwd: temporaryDirectory })); + const leakedDirectories: string[] = []; + for (const entry of after) { + if (!before.has(entry)) { + leakedDirectories.push(entry); + } + } + expect(leakedDirectories).toEqual([]); + }); +}); diff --git a/apps/server/src/services/pixiv-media.ts b/apps/server/src/services/pixiv-media.ts new file mode 100644 index 00000000..091ce38c --- /dev/null +++ b/apps/server/src/services/pixiv-media.ts @@ -0,0 +1,189 @@ +import JSZip from "jszip"; +import { MAX_MEDIA_DOWNLOAD_BYTES } from "@/services/media-download"; + +const PIXIV_HOSTS = new Set(["pixiv.net", "www.pixiv.net"]); +export const MAX_PIXIV_DOWNLOAD_BYTES = MAX_MEDIA_DOWNLOAD_BYTES; +export const MAX_UGOIRA_UNCOMPRESSED_BYTES = 200_000_000; +const MAX_UGOIRA_FRAMES = 1000; +const FFMPEG_TIMEOUT_MS = 120_000; + +const createTemporaryDirectory = async () => { + const directory = `${Bun.env.TMPDIR ?? "/tmp"}/starlight-ugoira-${Bun.randomUUIDv7()}`; + const child = Bun.spawn(["mkdir", "-m", "700", directory], { + stdout: "ignore", + stderr: "ignore", + }); + if ((await child.exited) !== 0) { + throw new Error("Failed to create ugoira temporary directory"); + } + return directory; +}; + +const removeTemporaryDirectory = async (directory: string) => { + const child = Bun.spawn(["rm", "-rf", directory], { stdout: "ignore", stderr: "ignore" }); + await child.exited; +}; + +const writeBoundedEntry = async ( + entry: JSZip.JSZipObject, + path: string, + limit: number, + total: { value: number }, +) => { + const stream = entry.nodeStream("nodebuffer"); + const sink = Bun.file(path).writer(); + await new Promise((resolve, reject) => { + let finished = false; + const finish = (error?: Error) => { + if (finished) { + return; + } + finished = true; + void Promise.resolve(sink.end(error)).then(() => (error ? reject(error) : resolve()), reject); + }; + + stream.on("data", (chunk: Uint8Array) => { + if (finished) { + return; + } + total.value += chunk.byteLength; + if (total.value > limit) { + finish(new Error("Ugoira is too large")); + return; + } + sink.write(chunk); + }); + stream.on("error", finish); + stream.on("end", () => finish()); + }); +}; + +export const parsePixivArtworkUrl = (value: string) => { + let url: URL; + try { + url = new URL(value); + } catch { + return null; + } + if (url.protocol !== "https:" || !PIXIV_HOSTS.has(url.hostname.toLowerCase())) { + return null; + } + const match = /^\/artworks\/(\d+)\/?$/.exec(url.pathname); + return match?.[1] ?? null; +}; + +export const buildFfmpegConcat = (frames: Array<{ file: string; delay: number }>) => { + if (frames.length === 0 || frames.some((frame) => frame.delay <= 0)) { + throw new Error("Invalid ugoira frame timing"); + } + const lines: string[] = []; + for (const frame of frames) { + if (frame.file.includes("'") || frame.file.includes("\n")) { + throw new Error("Invalid ugoira frame filename"); + } + lines.push(`file '${frame.file}'`, `duration ${frame.delay / 1000}`); + } + lines.push(`file '${frames.at(-1)?.file}'`); + return `${lines.join("\n")}\n`; +}; + +export const extractUgoiraZip = async ( + archive: ArrayBuffer | Uint8Array, + frames: Array<{ file: string; delay: number }>, + limits: { compressed?: number; uncompressed?: number; frames?: number } = {}, +) => { + const compressedLimit = limits.compressed ?? MAX_PIXIV_DOWNLOAD_BYTES; + const uncompressedLimit = limits.uncompressed ?? MAX_UGOIRA_UNCOMPRESSED_BYTES; + const frameLimit = limits.frames ?? MAX_UGOIRA_FRAMES; + if (archive.byteLength > compressedLimit || frames.length > frameLimit) { + throw new Error("Ugoira is too large"); + } + for (const frame of frames) { + if (frame.file.includes("/") || frame.file.includes("\\")) { + throw new Error("Unsafe ugoira archive path"); + } + } + const zip = await JSZip.loadAsync(archive); + for (const entry of Object.values(zip.files)) { + if (entry.unsafeOriginalName && entry.unsafeOriginalName !== entry.name) { + throw new Error("Unsafe ugoira archive path"); + } + } + const directory = await createTemporaryDirectory(); + const total = { value: 0 }; + try { + for (const frame of frames) { + const entry = zip.file(frame.file); + if (!entry || entry.dir) { + throw new Error(`Missing ugoira frame ${frame.file}`); + } + await writeBoundedEntry(entry, `${directory}/${frame.file}`, uncompressedLimit, total); + } + const concatPath = `${directory}/frames.txt`; + await Bun.write( + concatPath, + buildFfmpegConcat(frames.map((frame) => ({ ...frame, file: `${directory}/${frame.file}` }))), + ); + return { directory, concatPath }; + } catch (error) { + await removeTemporaryDirectory(directory); + throw error; + } +}; + +export const convertUgoira = async (concatPath: string, output: string) => { + // biome-ignore lint/correctness/noUndeclaredVariables: Server runtime is Bun. + const child = Bun.spawn( + [ + "ffmpeg", + "-y", + "-f", + "concat", + "-safe", + "0", + "-i", + concatPath, + "-vf", + "scale=trunc(iw/2)*2:trunc(ih/2)*2", + "-c:v", + "libx264", + "-pix_fmt", + "yuv420p", + "-movflags", + "+faststart", + "-fs", + String(MAX_PIXIV_DOWNLOAD_BYTES), + output, + ], + { stdout: "ignore", stderr: "pipe" }, + ); + const stderr = new Response(child.stderr).arrayBuffer(); + let timeout: ReturnType | undefined; + try { + const result = await Promise.race([ + child.exited, + new Promise<"timeout">((resolve) => { + timeout = setTimeout(() => { + resolve("timeout"); + }, FFMPEG_TIMEOUT_MS); + }), + ]); + if (result === "timeout") { + child.kill(); + await child.exited; + await stderr; + throw new Error("Ugoira conversion timed out"); + } + await stderr; + if (result !== 0) { + throw new Error("Failed to convert ugoira"); + } + if ((await Bun.file(output).stat()).size > MAX_PIXIV_DOWNLOAD_BYTES) { + throw new Error("Converted ugoira is too large"); + } + } finally { + if (timeout) { + clearTimeout(timeout); + } + } +}; diff --git a/apps/server/src/services/render/fonts.ts b/apps/server/src/services/render/fonts.ts index efc45788..54308f1d 100644 --- a/apps/server/src/services/render/fonts.ts +++ b/apps/server/src/services/render/fonts.ts @@ -1,4 +1,3 @@ -import path from "node:path"; import { GlobalFonts } from "@napi-rs/canvas"; import { logger } from "@/logger"; @@ -15,22 +14,19 @@ export function registerFonts(): void { } try { - const fontPath = path.join(process.cwd(), "assets", "fonts"); + const fontPath = `${process.cwd()}/assets/fonts`; - GlobalFonts.registerFromPath(path.join(fontPath, "Inter-Regular.ttf"), "Inter"); + GlobalFonts.registerFromPath(`${fontPath}/Inter-Regular.ttf`, "Inter"); - GlobalFonts.registerFromPath(path.join(fontPath, "Inter-Bold.ttf"), "Inter"); + GlobalFonts.registerFromPath(`${fontPath}/Inter-Bold.ttf`, "Inter"); - GlobalFonts.registerFromPath( - path.join(fontPath, "NotoSans-Regular.ttf"), - UNICODE_FALLBACK_FAMILY, - ); + GlobalFonts.registerFromPath(`${fontPath}/NotoSans-Regular.ttf`, UNICODE_FALLBACK_FAMILY); - GlobalFonts.registerFromPath(path.join(fontPath, "NotoSansCJKsc-Regular.otf"), CJK_FONT_FAMILY); + GlobalFonts.registerFromPath(`${fontPath}/NotoSansCJKsc-Regular.otf`, CJK_FONT_FAMILY); - GlobalFonts.registerFromPath(path.join(fontPath, "NotoSansMath-Regular.ttf"), MATH_FONT_FAMILY); + GlobalFonts.registerFromPath(`${fontPath}/NotoSansMath-Regular.ttf`, MATH_FONT_FAMILY); - GlobalFonts.registerFromPath(path.join(fontPath, "NotoColorEmoji.ttf"), EMOJI_FONT_FAMILY); + GlobalFonts.registerFromPath(`${fontPath}/NotoColorEmoji.ttf`, EMOJI_FONT_FAMILY); fontsRegistered = true; logger.info("Registered Inter, Noto Sans, Noto Sans CJK, Noto Sans Math, and emoji fonts"); diff --git a/apps/server/src/services/tag-normalization.ts b/apps/server/src/services/tag-normalization.ts new file mode 100644 index 00000000..5d1ac837 --- /dev/null +++ b/apps/server/src/services/tag-normalization.ts @@ -0,0 +1,12 @@ +export const normalizeTags = (tags: readonly string[]): string[] => { + const normalized: string[] = []; + const seen = new Set(); + for (const tag of tags) { + const value = tag.trim(); + if (value && !seen.has(value)) { + seen.add(value); + normalized.push(value); + } + } + return normalized; +}; diff --git a/apps/server/src/services/twitter-tags.ts b/apps/server/src/services/twitter-tags.ts new file mode 100644 index 00000000..9f89bce9 --- /dev/null +++ b/apps/server/src/services/twitter-tags.ts @@ -0,0 +1,5 @@ +import type { Tweet } from "@the-convocation/twitter-scraper"; +import { normalizeTags } from "@/services/tag-normalization"; + +export const normalizeTwitterTags = (tweet: Pick): string[] => + normalizeTags(tweet.hashtags); diff --git a/apps/server/src/services/video.ts b/apps/server/src/services/video.ts index 83d31804..d5cb154e 100644 --- a/apps/server/src/services/video.ts +++ b/apps/server/src/services/video.ts @@ -1,4 +1,3 @@ -import path from "node:path"; import { env } from "@starlight/utils"; import { http } from "@starlight/utils/http"; import { create } from "youtube-dl-exec"; @@ -17,9 +16,7 @@ export interface VideoInformation { } async function createVideoInformation(filePath: string): Promise { - const parsedPath = path.parse(filePath); - - const infoJsonPath = path.join(parsedPath.dir, `${parsedPath.name}.info.json`); + const infoJsonPath = filePath.replace(/\.mp4$/, ".info.json"); logger.debug({ infoJsonPath }, "Creating video information"); @@ -45,7 +42,7 @@ export async function downloadVideoFromUrl( metadata: VideoMetadata = {}, ): Promise { const uuid = Bun.randomUUIDv7(); - const filePath = path.join(folder, `${uuid}.mp4`); + const filePath = `${folder}/${uuid}.mp4`; logger.debug({ url }, "Downloading video directly from URL"); @@ -87,7 +84,7 @@ export async function downloadVideo(url: string, folder: string): Promise new Cookie(mapToRFC6265Cookie(cookie)))); + return new Cookies(parseTwitterCookies(data).map((cookie) => new Cookie(cookie))); } userId() { - const twidValue = this.cookies.find((cookie) => cookie.key === "twid")?.value; - - if (!twidValue) { - return; - } - - const decoded = decodeURIComponent(twidValue); - const match = decoded.match(TWID_REGEX); - return match ? match[1] : undefined; + return getTwitterUserId( + this.cookies.map((cookie) => ({ + domain: cookie.domain ?? "", + key: cookie.key, + value: cookie.value, + })), + ); } } @@ -45,16 +33,3 @@ export const s3 = new Bun.S3Client({ secretAccessKey: env.AWS_SECRET_ACCESS_KEY, endpoint: env.AWS_ENDPOINT, }); - -function extractDomain(hostRaw: string): string { - const match = hostRaw.match(DOMAIN_REGEX); - return match?.[1] ?? "x.com"; -} - -export function mapToRFC6265Cookie(firefoxCookie: any): RFC6265Cookie { - return { - key: firefoxCookie["Name raw"], - value: firefoxCookie["Content raw"], - domain: extractDomain(firefoxCookie["Host raw"]), - }; -} diff --git a/apps/web/src/components/tweet-image-grid.tsx b/apps/web/src/components/post-media-grid.tsx similarity index 87% rename from apps/web/src/components/tweet-image-grid.tsx rename to apps/web/src/components/post-media-grid.tsx index 83162f39..9abd3c75 100644 --- a/apps/web/src/components/tweet-image-grid.tsx +++ b/apps/web/src/components/post-media-grid.tsx @@ -1,4 +1,4 @@ -import type { TweetData } from "@starlight/api/src/types/tweets"; +import type { PostData } from "@starlight/api/src/types/posts"; import { X } from "lucide-react"; import type { UIElementData } from "photoswipe"; import type { PhotoSwipe } from "photoswipe/lightbox"; @@ -17,19 +17,19 @@ import { import { Button } from "@/components/ui/button"; import { Carousel } from "@/components/ui/skiper-ui/carousel"; -interface TweetImageGridProps { - onDeleteImage?: (photoId: string) => void; +interface PostMediaGridProps { + onDeleteMedia?: (mediaId: string) => void; showActions?: boolean; showArtistOnHover?: boolean; - tweet: TweetData; + post: PostData; } -export function TweetImageGrid({ - tweet, +export function PostMediaGrid({ + post, showActions = false, showArtistOnHover = false, - onDeleteImage, -}: TweetImageGridProps) { + onDeleteMedia, +}: PostMediaGridProps) { const [isImageLoading, setIsImageLoading] = useState<{ [key: string]: boolean; }>({}); @@ -42,7 +42,7 @@ export function TweetImageGrid({ const handleArtistClick = (e: React.MouseEvent) => { e.stopPropagation(); - window.open(tweet.sourceUrl, "_blank", "noopener,noreferrer"); + window.open(post.sourceUrl, "_blank", "noopener,noreferrer"); }; const handleGalleryOpenKeyDown = (event: React.KeyboardEvent) => { @@ -98,8 +98,8 @@ export function TweetImageGrid({ }, ]; - if (tweet.photos.length === 1) { - const photo = tweet.photos[0]; + if (post.media.length === 1) { + const media = post.media[0]; return ( {({ ref, open }) => ( // biome-ignore lint/a11y/useSemanticElements: wrapper must stay a non-button container because it contains nested interactive controls @@ -126,22 +126,22 @@ export function TweetImageGrid({ role="button" tabIndex={0} > - {isImageLoading[photo.id] && ( + {isImageLoading[media.id] && (
)} {/* biome-ignore lint/a11y/noNoninteractiveElementInteractions: onLoad and onLoadStart are used only for image loading state */} {photo.alt} handleImageLoad(photo.id, false)} - onLoadStart={() => handleImageLoad(photo.id, true)} - src={photo.url} - width={photo.width || 400} + height={media.height || 400} + onLoad={() => handleImageLoad(media.id, false)} + onLoadStart={() => handleImageLoad(media.id, true)} + src={media.url} + width={media.width || 400} />
@@ -159,15 +159,15 @@ export function TweetImageGrid({ onClick={(e) => handleArtistClick(e)} type="button" > - {tweet.artist} + {post.artist}
- {showActions && onDeleteImage && ( + {showActions && onDeleteMedia && (
- {showActions && onDeleteImage && ( + {showActions && onDeleteMedia && (
)} @@ -186,7 +186,7 @@ function TwitterArtViewer() { tweet.id} + itemKey={(post) => post.id} items={displayItems} onRender={currentInfiniteLoader} overscanBy={MASONRY_OVERSCAN_BY} @@ -247,18 +247,18 @@ export const Route = createFileRoute("/app")({ await queryClient.fetchQuery(profileOptions); await queryClient.fetchInfiniteQuery( - orpc.tweets.list.infiniteOptions({ + orpc.posts.list.infiniteOptions({ input: (pageParam: string | undefined) => ({ cursor: pageParam, limit: 30, }), - queryKey: ["tweets", { username: undefined }], + queryKey: ["posts", { username: undefined }], initialPageParam: undefined, - getNextPageParam: (lastPage: TweetsPageResult) => lastPage.nextCursor ?? undefined, + getNextPageParam: (lastPage: PostsPageResult) => lastPage.nextCursor ?? undefined, retry: false, gcTime: 10 * 60 * 1000, }), ); }, - component: TwitterArtViewer, + component: MediaGallery, }); diff --git a/apps/web/src/routes/index.tsx b/apps/web/src/routes/index.tsx index 03be4b0b..68bc1948 100644 --- a/apps/web/src/routes/index.tsx +++ b/apps/web/src/routes/index.tsx @@ -1,12 +1,12 @@ -import type { TweetData } from "@starlight/api/src/types/tweets"; +import type { PostData } from "@starlight/api/src/types/posts"; import { useQuery } from "@tanstack/react-query"; import { createFileRoute } from "@tanstack/react-router"; import { Search } from "lucide-react"; import { Masonry, useInfiniteLoader } from "masonic"; import { parseAsString, useQueryState } from "nuqs"; import { useCallback, useEffect, useMemo, useState, lazy, Suspense } from "react"; -const TweetImageGrid = lazy(() => - import("@/components/tweet-image-grid").then((m) => ({ default: m.TweetImageGrid })), +const PostMediaGrid = lazy(() => + import("@/components/post-media-grid").then((m) => ({ default: m.PostMediaGrid })), ); import { Button } from "@/components/ui/button"; import { Input } from "@/components/ui/input"; @@ -52,14 +52,14 @@ export default function DiscoverPage() { }); const randomQuery = useQuery({ - ...orpc.tweets.random.queryOptions({ retry: false }), - queryKey: ["tweets-random"], + ...orpc.posts.random.queryOptions({ retry: false }), + queryKey: ["posts-random"], enabled: true, staleTime: Number.POSITIVE_INFINITY, gcTime: Number.POSITIVE_INFINITY, }); - const randomImages: TweetData[] = randomQuery.data || []; + const randomPosts: PostData[] = randomQuery.data || []; const handleSearch = (e: React.FormEvent) => { e.preventDefault(); @@ -90,21 +90,21 @@ export default function DiscoverPage() { const renderMasonryItem = useCallback( ({ data, width }: { data: any; width: number }) => (
- +
), [], ); - // Generate non-overlapping positions for random images + // Generate non-overlapping positions for random posts const placedData = useMemo(() => { - if (randomImages.length === 0 || typeof window === "undefined") { + if (randomPosts.length === 0 || typeof window === "undefined") { return []; } const layout = new LayoutManager(100, 100); - return layout.placeTweets(randomImages); - }, [randomImages]); + return layout.placePosts(randomPosts); + }, [randomPosts]); useEffect(() => { if (placedData.length > 0 && randomQuery.isSuccess && !isLoading && results.length === 0) { @@ -135,7 +135,7 @@ export default function DiscoverPage() { tweet.id} + itemKey={(post) => post.id} items={results} onRender={infiniteLoader} overscanBy={MASONRY_OVERSCAN_BY} @@ -144,7 +144,7 @@ export default function DiscoverPage() { ) : ( - // Hero Section with centered search and floating images on large screen + // Hero Section with centered search and floating media on large screen
@@ -229,7 +229,7 @@ export default function DiscoverPage() {
{placedData.map(({ position, index }) => { - const tweet = randomImages[index]; + const post = randomPosts[index]; const isVisible = visibleIndices.includes(index); return (
- +
); })} @@ -292,8 +292,8 @@ export default function DiscoverPage() { export const Route = createFileRoute("/")({ loader: ({ context: { queryClient } }) => { queryClient.prefetchQuery({ - ...orpc.tweets.random.queryOptions({ retry: false }), - queryKey: ["tweets-random"], + ...orpc.posts.random.queryOptions({ retry: false }), + queryKey: ["posts-random"], }); }, component: DiscoverPage, diff --git a/apps/web/src/routes/profile/$slug.tsx b/apps/web/src/routes/profile/$slug.tsx index 635e4461..e7aaec0e 100644 --- a/apps/web/src/routes/profile/$slug.tsx +++ b/apps/web/src/routes/profile/$slug.tsx @@ -1,10 +1,10 @@ -import type { TweetData, TweetsPageResult } from "@starlight/api/src/types/tweets"; +import type { PostData, PostsPageResult } from "@starlight/api/src/types/posts"; import { createFileRoute, useParams } from "@tanstack/react-router"; import { Masonry, useInfiniteLoader } from "masonic"; import { useCallback, lazy, Suspense } from "react"; import { NotFound } from "@/components/not-found"; -const TweetImageGrid = lazy(() => import("@/components/tweet-image-grid").then((m) => ({ default: m.TweetImageGrid }))); -import { useTweets } from "@/hooks/use-tweets"; +const PostMediaGrid = lazy(() => import("@/components/post-media-grid").then((m) => ({ default: m.PostMediaGrid }))); +import { usePosts } from "@/hooks/use-posts"; import { orpc } from "@/utils/orpc"; const MASONRY_ITEM_HEIGHT_ESTIMATE = 360; @@ -13,7 +13,7 @@ const MASONRY_OVERSCAN_BY = 1.25; function SharedProfileViewer() { const { slug } = useParams({ from: "/profile/$slug" }); - const { tweets, isLoading, isFetchingNextPage, hasNextPage, error, fetchNextPage } = useTweets({ + const { posts, isLoading, isFetchingNextPage, hasNextPage, error, fetchNextPage } = usePosts({ username: slug, }); @@ -31,9 +31,9 @@ function SharedProfileViewer() { ); const renderMasonryItem = useCallback( - ({ data, width }: { data: TweetData; width: number }) => ( + ({ data, width }: { data: PostData; width: number }) => (
- +
), [], @@ -58,7 +58,7 @@ function SharedProfileViewer() { return (
- {!isLoading && tweets.length === 0 && ( + {!isLoading && posts.length === 0 && (
)} - {tweets.length > 0 && ( + {posts.length > 0 && (
tweet.id} - items={tweets} + itemKey={(post) => post.id} + items={posts} onRender={infiniteLoader} overscanBy={MASONRY_OVERSCAN_BY} render={renderMasonryItem} @@ -96,14 +96,14 @@ export const Route = createFileRoute("/profile/$slug")({ } await queryClient.fetchInfiniteQuery( - orpc.tweets.list.infiniteOptions({ + orpc.posts.list.infiniteOptions({ input: (pageParam: string | undefined) => ({ cursor: pageParam, limit: 30, }), - queryKey: ["tweets", { username: slug }], + queryKey: ["posts", { username: slug }], initialPageParam: undefined, - getNextPageParam: (lastPage: TweetsPageResult) => lastPage.nextCursor ?? undefined, + getNextPageParam: (lastPage: PostsPageResult) => lastPage.nextCursor ?? undefined, retry: false, gcTime: 10 * 60 * 1000, }), diff --git a/apps/web/src/routes/settings.tsx b/apps/web/src/routes/settings.tsx index d9372683..69f60c7b 100644 --- a/apps/web/src/routes/settings.tsx +++ b/apps/web/src/routes/settings.tsx @@ -1,7 +1,7 @@ import type { ProfileResult } from "@starlight/api/routers/index"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { createFileRoute, Link } from "@tanstack/react-router"; -import { AlertCircle, Cookie, Trash2 } from "lucide-react"; +import { AlertCircle, Cookie, KeyRound, Trash2 } from "lucide-react"; import { useState } from "react"; import { Alert, AlertDescription, AlertTitle } from "@/components/ui/alert"; import { Button } from "@/components/ui/button"; @@ -32,6 +32,7 @@ export const Route = createFileRoute("/settings")({ function RouteComponent() { const [newCookies, setNewCookies] = useState(""); + const [pixivToken, setPixivToken] = useState(""); const [displayError, setDisplayError] = useState(null); const { rawInitData } = useTelegramContext(); @@ -93,6 +94,38 @@ function RouteComponent() { }), ); + const savePixivMutation = useMutation( + orpc.pixiv.save.mutationOptions({ + onSuccess: () => { + queryClient.setQueryData(["profile"], (old: ProfileResult) => ({ + ...old, + hasPixivCredential: true, + })); + setPixivToken(""); + }, + }), + ); + const deletePixivMutation = useMutation( + orpc.pixiv.delete.mutationOptions({ + onSuccess: () => { + queryClient.setQueryData(["profile"], (old: ProfileResult) => ({ + ...old, + hasPixivCredential: false, + })); + }, + }), + ); + const pixivPrivateMutation = useMutation( + orpc.pixiv.privateBookmarks.mutationOptions({ + onSuccess: (_data, variables) => { + queryClient.setQueryData(["profile"], (old: ProfileResult) => ({ + ...old, + pixivIncludePrivate: variables.enabled, + })); + }, + }), + ); + if (isLoading && !profile) { return (
@@ -148,29 +181,38 @@ function RouteComponent() { const isSubmitting = saveCookiesMutation.isPending || deleteCookiesMutation.isPending || - visibilityMutation.isPending; + visibilityMutation.isPending || + savePixivMutation.isPending || + deletePixivMutation.isPending || + pixivPrivateMutation.isPending; + const twitterError = profile?.hasValidCookies + ? deleteCookiesMutation.error?.message + : displayError; + const pixivError = profile?.hasPixivCredential + ? (deletePixivMutation.error?.message ?? pixivPrivateMutation.error?.message) + : savePixivMutation.error?.message; return (
- {/* Cookie Management Section */} + {/* Twitter Cookie Management Section */}

- Authentication Cookies + Twitter

{/* Cookie Success/Error Messages */} - {cookieError && ( + {(cookieError || twitterError) && ( - {cookieError.message} + {cookieError?.message ?? twitterError} )} {profile?.hasValidCookies ? ( - Authentication cookies are saved. + Twitter is connected.
{profile?.hasValidCookies && ( + + + + ) : ( +
{ + event.preventDefault(); + savePixivMutation.mutate({ refreshToken: pixivToken }); + }} + > + + + + )} +
+ {/* Profile Visibility Section */} -
+
+ {visibilityMutation.error && ( + + + {visibilityMutation.error.message} + + )}