diff options
Diffstat (limited to 'apps/cf-ai-backend/src/queueConsumer')
9 files changed, 703 insertions, 5 deletions
diff --git a/apps/cf-ai-backend/src/queueConsumer/chunkers/chonker.ts b/apps/cf-ai-backend/src/queueConsumer/chunkers/chonker.ts new file mode 100644 index 00000000..18788dab --- /dev/null +++ b/apps/cf-ai-backend/src/queueConsumer/chunkers/chonker.ts @@ -0,0 +1,47 @@ +import nlp from "compromise"; + +/** + * Split text into chunks of specified max size with some overlap for continuity. + */ +export default function chunkText( + text: string, + maxChunkSize: number, + overlap: number = 0.2, +): string[] { + const sentences = nlp(text).sentences().out("array"); + const chunks = []; + let currentChunk: string[] = []; + let currentSize = 0; + + for (let i = 0; i < sentences.length; i++) { + const sentence = sentences[i]; + currentChunk.push(sentence); + currentSize += sentence.length; + + if (currentSize >= maxChunkSize) { + // Calculate overlap + const overlapSize = Math.floor(currentChunk.length * overlap); + const chunkText = currentChunk.join(" "); + chunks.push({ + text: chunkText, + start: i - currentChunk.length + 1, + end: i, + }); + + // Prepare the next chunk with overlap + currentChunk = currentChunk.slice(-overlapSize); + currentSize = currentChunk.reduce((sum, s) => sum + s.length, 0); + } + } + + if (currentChunk.length > 0) { + const chunkText = currentChunk.join(" "); + chunks.push({ + text: chunkText, + start: sentences.length - currentChunk.length, + end: sentences.length, + }); + } + + return chunks.map((chunk) => chunk.text); +} diff --git a/apps/cf-ai-backend/src/queueConsumer/chunkers/chunkPageOrNotes.ts b/apps/cf-ai-backend/src/queueConsumer/chunkers/chunkPageOrNotes.ts new file mode 100644 index 00000000..0da01c3f --- /dev/null +++ b/apps/cf-ai-backend/src/queueConsumer/chunkers/chunkPageOrNotes.ts @@ -0,0 +1,13 @@ +import chunkText from "./chonker"; +import { PageOrNoteChunks } from "../../types"; +export function chunkPage(pageContent: string): PageOrNoteChunks { + const chunks = chunkText(pageContent, 1536); + + return { type: "page", chunks: chunks }; +} + +export function chunkNote(noteContent: string): PageOrNoteChunks { + const chunks = chunkText(noteContent, 1536); + + return { type: "note", chunks: chunks }; +} diff --git a/apps/cf-ai-backend/src/queueConsumer/chunkers/chunkTweet.ts b/apps/cf-ai-backend/src/queueConsumer/chunkers/chunkTweet.ts index ae1b18c6..46a56410 100644 --- a/apps/cf-ai-backend/src/queueConsumer/chunkers/chunkTweet.ts +++ b/apps/cf-ai-backend/src/queueConsumer/chunkers/chunkTweet.ts @@ -22,10 +22,18 @@ export interface ThreadTweetData { } export function chunkThread(threadText: string): TweetChunks { - const thread = JSON.parse(threadText); - if (typeof thread == "string") { - console.log("DA WORKER FAILED DO SOMEHTING FIX DA WROKER"); + let thread = threadText; + + try { + thread = JSON.parse(threadText); + } catch (e) { + console.log("error: thread is not json.", e); + } + + if (typeof threadText == "string") { + console.log("DA WORKER FAILED DO SOMEHTING FIX DA WROKER", thread); const rawTweet = getRawTweet(thread); + console.log(rawTweet); const parsedTweet: any = JSON.parse(rawTweet); const chunkedTweet = chunkText(parsedTweet.text, 1536); @@ -48,8 +56,8 @@ export function chunkThread(threadText: string): TweetChunks { return { type: "tweet", chunks }; } else { - console.log(JSON.stringify(thread)); - const chunkedTweets = thread.map((tweet: Tweet) => { + console.log("thread in else statement", JSON.stringify(thread)); + const chunkedTweets = (thread as any).map((tweet: Tweet) => { const chunkedTweet = chunkText(tweet.text, 1536); const metadata = { diff --git a/apps/cf-ai-backend/src/queueConsumer/helpers/processNotes.ts b/apps/cf-ai-backend/src/queueConsumer/helpers/processNotes.ts new file mode 100644 index 00000000..466690cc --- /dev/null +++ b/apps/cf-ai-backend/src/queueConsumer/helpers/processNotes.ts @@ -0,0 +1,36 @@ +import { Result, Ok, Err } from "../../errors/results"; +import { BaseError } from "../../errors/baseError"; +import { Metadata } from "../utils/get-metadata"; + +class ProcessNotesError extends BaseError { + constructor(message?: string, source?: string) { + super("[Note Processing Error]", message, source); + } +} + +type ProcessNoteResult = { + noteContent: { noteId: number; noteContent: string }; + metadata: Metadata; +}; + +export function processNote( + content: string, +): Result<ProcessNoteResult, ProcessNotesError> { + try { + const pageContent = content; + const noteId = new Date().getTime(); + + const metadata = { + baseUrl: `https://supermemory.ai/note/${noteId}`, + description: `Note created at ${new Date().toLocaleString()}`, + image: "https://supermemory.ai/logo.png", + title: `${pageContent.slice(0, 20)} ${pageContent.length > 20 ? "..." : ""}`, + }; + + const noteContent = { noteId: noteId, noteContent: pageContent }; + return Ok({ noteContent, metadata }); + } catch (e) { + console.error("[Note Processing Error]", e); + return Err(new ProcessNotesError((e as Error).message, "processNote")); + } +} diff --git a/apps/cf-ai-backend/src/queueConsumer/helpers/processPage.ts b/apps/cf-ai-backend/src/queueConsumer/helpers/processPage.ts new file mode 100644 index 00000000..9a50d701 --- /dev/null +++ b/apps/cf-ai-backend/src/queueConsumer/helpers/processPage.ts @@ -0,0 +1,43 @@ +import { Result, Ok, Err, isErr } from "../../errors/results"; +import { BaseError } from "../../errors/baseError"; +import { getMetaData, Metadata } from "../utils/get-metadata"; + +class ProcessPageError extends BaseError { + constructor(message?: string, source?: string) { + super("[Page Proceessing Error]", message, source); + } +} + +type PageProcessResult = { pageContent: string; metadata: Metadata }; + +export async function processPage(input: { + url: string; + securityKey: string; +}): Promise<Result<PageProcessResult, ProcessPageError>> { + try { + const response = await fetch("https://md.dhr.wtf/?url=" + input.url, { + headers: { + Authorization: "Bearer " + input.securityKey, + }, + }); + const pageContent = await response.text(); + if (!pageContent) { + return Err( + new ProcessPageError( + "Failed to get response form markdowner", + "processPage", + ), + ); + } + const metadataResult = await getMetaData(input.url); + if (isErr(metadataResult)) { + throw metadataResult.error; + } + const metadata = metadataResult.value; + console.log("[this is the metadata]", metadata); + return Ok({ pageContent, metadata }); + } catch (e) { + console.error("[Page Processing Error]", e); + return Err(new ProcessPageError((e as Error).message, "processPage")); + } +} diff --git a/apps/cf-ai-backend/src/queueConsumer/helpers/processTweet.ts b/apps/cf-ai-backend/src/queueConsumer/helpers/processTweet.ts new file mode 100644 index 00000000..8d83f2dc --- /dev/null +++ b/apps/cf-ai-backend/src/queueConsumer/helpers/processTweet.ts @@ -0,0 +1,88 @@ +import { Tweet } from "react-tweet/api"; +import { Result, Ok, Err, isErr } from "../../errors/results"; +import { BaseError } from "../../errors/baseError"; +import { getMetaData, Metadata } from "../utils/get-metadata"; +import { tweetToMd } from "@repo/shared-types/utils"; // can I do this? +import { Env } from "../../types"; + +class ProcessTweetError extends BaseError { + constructor(message?: string, source?: string) { + super("[Tweet Proceessing Error]", message, source); + } +} + +type GetTweetResult = Tweet; + +export const getTweetData = async ( + tweetID: string, +): Promise<Result<GetTweetResult, ProcessTweetError>> => { + try { + console.log("is fetch defined here?"); + const url = `https://cdn.syndication.twimg.com/tweet-result?id=${tweetID}&lang=en&features=tfw_timeline_list%3A%3Btfw_follower_count_sunset%3Atrue%3Btfw_tweet_edit_backend%3Aon%3Btfw_refsrc_session%3Aon%3Btfw_fosnr_soft_interventions_enabled%3Aon%3Btfw_show_birdwatch_pivots_enabled%3Aon%3Btfw_show_business_verified_badge%3Aon%3Btfw_duplicate_scribes_to_settings%3Aon%3Btfw_use_profile_image_shape_enabled%3Aon%3Btfw_show_blue_verified_badge%3Aon%3Btfw_legacy_timeline_sunset%3Atrue%3Btfw_show_gov_verified_badge%3Aon%3Btfw_show_business_affiliate_badge%3Aon%3Btfw_tweet_edit_frontend%3Aon&token=4c2mmul6mnh`; + + const resp = await fetch(url, { + headers: { + "User-Agent": + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3", + Accept: "application/json", + "Accept-Language": "en-US,en;q=0.5", + "Accept-Encoding": "gzip, deflate, br", + Connection: "keep-alive", + "Upgrade-Insecure-Requests": "1", + "Cache-Control": "max-age=0", + TE: "Trailers", + }, + }); + console.log(resp.status); + + const data = (await resp.json()) as Tweet; + + return Ok(data); + } catch (e) { + console.error("[Tweet Proceessing Error]", e); + return Err(new ProcessTweetError(e, "getTweetData")); + } +}; + +export const getThreadData = async (input: { + tweetUrl: string; + env: Env; +}): Promise<Result<any, ProcessTweetError>> => { + try { + // const threadRequest = await fetch(input.cf_thread_endpoint, { + // method: "POST", + // headers: { + // "Content-Type": "application/json", + // Authorization: input.authKey, + // }, + // body: JSON.stringify({ url: input.tweetUrl }), + // }); + // if (threadRequest.status !== 200) { + // console.log(await threadRequest.text()); + // console.log(input.tweetUrl); + // return Err( + // new ProcessTweetError( + // `Failed to fetch the thread: ${input.tweetUrl}, Reason: ${threadRequest.statusText}`, + // "getThreadData", + // ), + // ); + // } + //@ts-ignore + const thread = await input.env.THREAD.processTweets(input.tweetUrl); + console.log("[thread response]", thread); + + if (!thread.length) { + console.log("Thread is an empty array"); + return Err( + new ProcessTweetError( + "[THREAD FETCHING SERVICE] Got no content form thread worker", + "getThreadData", + ), + ); + } + return Ok(thread); + } catch (e) { + console.error("[Thread Processing Error]", e); + return Err(new ProcessTweetError((e as Error).message, "getThreadData")); + } +}; diff --git a/apps/cf-ai-backend/src/queueConsumer/index.ts b/apps/cf-ai-backend/src/queueConsumer/index.ts new file mode 100644 index 00000000..393f1fbf --- /dev/null +++ b/apps/cf-ai-backend/src/queueConsumer/index.ts @@ -0,0 +1,372 @@ +import { Env, PageOrNoteChunks, TweetChunks, vectorObj } from "../types"; +import { typeDecider } from "./utils/typeDecider"; +import { isErr, wrap } from "../errors/results"; +import { processNote } from "./helpers/processNotes"; +import { processPage } from "./helpers/processPage"; +import { getThreadData, getTweetData } from "./helpers/processTweet"; +import { tweetToMd } from "@repo/shared-types/utils"; +import { chunkNote, chunkPage } from "./chunkers/chunkPageOrNotes"; +import { chunkThread } from "./chunkers/chunkTweet"; +import { batchCreateChunksAndEmbeddings, initQuery } from "../helper"; +import { z } from "zod"; +import { Metadata } from "./utils/get-metadata"; +import { BaseError } from "../errors/baseError"; +import { database } from "../db"; +import { + storedContent, + space, + contentToSpace, + users, + jobs, + Job, +} from "@repo/db/schema"; +import { and, eq, inArray, sql } from "drizzle-orm"; + +class VectorInsertError extends BaseError { + constructor(message?: string, source?: string) { + super("[Vector Insert Error]", message, source); + } +} +const vectorErrorFactory = (err: Error) => new VectorInsertError(err.message); + +class D1InsertError extends BaseError { + constructor(message?: string, source?: string) { + super("[D1 Insert Error]", message, source); + } +} + +const d1ErrorFactory = (err: Error, source: string) => + new D1InsertError(err.message, source); + +const calculateExponentialBackoff = ( + attempts: number, + baseDelaySeconds: number, +) => { + return baseDelaySeconds ** attempts; +}; + +const BASE_DELAY_SECONDS = 5; +export async function queue( + batch: MessageBatch<{ + content: string; + space: Array<number>; + user: string; + type: string; + }>, + env: Env, +): Promise<void> { + const db = database(env); + console.log(env.CF_ACCOUNT_ID, env.CF_KV_AUTH_TOKEN); + for (let message of batch.messages) { + console.log(env.CF_ACCOUNT_ID, env.CF_KV_AUTH_TOKEN); + console.log("is thie even running?", message.body); + const body = message.body; + + const type = body.type; + const userExists = await wrap( + db.select().from(users).where(eq(users.id, body.user)).limit(1), + d1ErrorFactory, + "Error when trying to verify user", + ); + + if (isErr(userExists)) { + throw userExists.error; + } + + //check if this is a retry job.. by checking if the combination of the userId and the url already exists on the queue + let jobId; + const existingJob = await wrap( + db + .select() + .from(jobs) + .where( + and( + eq(jobs.userId, userExists.value[0].id), + eq(jobs.url, body.content), + ), + ) + .limit(1), + d1ErrorFactory, + "Error when checking for existing job", + ); + + if (isErr(existingJob)) { + throw existingJob.error; + } + + if (existingJob.value.length > 0) { + jobId = existingJob.value[0].id; + await wrap( + db + .update(jobs) + .set({ + attempts: existingJob.value[0].attempts + 1, + updatedAt: new Date(), + status: "Processing", + }) + .where(eq(jobs.id, jobId)), + d1ErrorFactory, + "Error when updating job attempts", + ); + } else { + const job = await wrap( + db + .insert(jobs) + .values({ + userId: userExists.value[0].id as string, + url: body.content, + status: "Processing", + attempts: 1, + createdAt: new Date(), + updatedAt: new Date(), + }) + .returning({ jobId: jobs.id }), + d1ErrorFactory, + "Error When inserting into jobs table", + ); + if (isErr(job)) { + throw job.error; + } + jobId = job.value[0].jobId; + } + + let pageContent: string; + let vectorData: string; + let metadata: Metadata; + let storeToSpaces = body.space; + let chunks: TweetChunks | PageOrNoteChunks; + let noteId = 0; + switch (type) { + case "note": { + console.log("note hit"); + const note = processNote(body.content); + if (isErr(note)) { + throw note.error; + } + pageContent = note.value.noteContent.noteContent; + noteId = note.value.noteContent.noteId; + metadata = note.value.metadata; + vectorData = pageContent; + chunks = chunkNote(pageContent); + break; + } + case "page": { + console.log("page hit"); + const page = await processPage({ + url: body.content, + securityKey: env.MD_SEC_KEY, + }); + if (isErr(page)) { + console.log("there is a page error here"); + throw page.error; + } + pageContent = page.value.pageContent; + metadata = page.value.metadata; + vectorData = pageContent; + chunks = chunkPage(pageContent); + break; + } + + case "tweet": { + console.log("tweet hit"); + const tweet = await getTweetData(body.content.split("/").pop()); + console.log(env.THREAD_CF_WORKER, env.THREAD_CF_AUTH); + const thread = await getThreadData({ + tweetUrl: body.content, + env: env, + }); + console.log("[This is the thread]", thread); + if (isErr(tweet)) { + throw tweet.error; + } + pageContent = tweetToMd(tweet.value); + console.log(pageContent); + metadata = { + baseUrl: body.content, + description: tweet.value.text.slice(0, 200), + image: tweet.value.user.profile_image_url_https, + title: `Tweet by ${tweet.value.user.name}`, + }; + if (isErr(thread)) { + console.log("Thread worker is down!"); + vectorData = JSON.stringify(pageContent); + console.error(thread.error); + } else { + console.log("thread worker is fine"); + vectorData = thread.value; + } + chunks = chunkThread(vectorData); + break; + } + } + + //add to mem0, abstract + + // const mem0Response = fetch('https://api.mem0.ai/v1/memories/', { + // method: 'POST', + // headers: { + // 'Content-Type': 'application/json', + // Authorization: `Token ${process.env.MEM0_API_KEY}`, + // }, + // body: JSON.stringify({ + // messages: [ + // { + // role: 'user', + // content: query, + // }, + // ], + // user_id: user?.user?.email, + // }), + // }); + + // see what's up with the storedToSpaces in this block + const { store } = await initQuery(env); + + type body = z.infer<typeof vectorObj>; + + const Chunkbody: body = { + pageContent: pageContent, + spaces: storeToSpaces.map((spaceId) => spaceId.toString()), + user: body.user, + type: type, + url: metadata.baseUrl, + description: metadata.description, + title: metadata.description, + }; + + try { + const vectorResult = await wrap( + batchCreateChunksAndEmbeddings({ + store: store, + body: Chunkbody, + chunks: chunks, + env: env, + }), + vectorErrorFactory, + "Error when Inserting into vector database", + ); + + if (isErr(vectorResult)) { + await db + .update(jobs) + .set({ error: vectorResult.error.message, status: "error" }) + .where(eq(jobs.id, jobId)); + message.retry({ + delaySeconds: calculateExponentialBackoff( + message.attempts, + BASE_DELAY_SECONDS, + ), + }); + throw vectorResult.error; + } + + const saveToDbUrl = + (metadata.baseUrl.split("#supermemory-user-")[0] ?? metadata.baseUrl) + + "#supermemory-user-" + + body.user; + let contentId: number; + + const insertResponse = await wrap( + db + .insert(storedContent) + .values({ + content: pageContent as string, + title: metadata.title, + description: metadata.description, + url: saveToDbUrl, + baseUrl: saveToDbUrl, + image: metadata.image, + savedAt: new Date(), + userId: body.user, + type: type, + noteId: noteId, + }) + .returning({ id: storedContent.id }), + d1ErrorFactory, + "Error when inserting into storedContent", + ); + + if (isErr(insertResponse)) { + await db + .update(jobs) + .set({ error: insertResponse.error.message, status: "error" }) + .where(eq(jobs.id, jobId)); + message.retry({ + delaySeconds: calculateExponentialBackoff( + message.attempts, + BASE_DELAY_SECONDS, + ), + }); + throw insertResponse.error; + } + console.log(JSON.stringify(insertResponse)); + contentId = insertResponse[0]?.id; + console.log("this is the content Id", contentId); + if (storeToSpaces.length > 0) { + // Adding the many-to-many relationship between content and spaces + const spaceData = await wrap( + db + .select() + .from(space) + .where( + and(inArray(space.id, storeToSpaces), eq(space.user, body.user)), + ) + .all(), + d1ErrorFactory, + "Error when getting data from spaces", + ); + + if (isErr(spaceData)) { + throw spaceData.error; + } + try { + await Promise.all( + spaceData.value.map(async (s) => { + try { + await db + .insert(contentToSpace) + .values({ contentId: contentId, spaceId: s.id }); + + await db.update(space).set({ numItems: s.numItems + 1 }); + } catch (e) { + console.error(`Error updating space ${s.id}:`, e); + throw e; + } + }), + ); + } catch (e) { + console.error("Error in updateSpacesWithContent:", e); + throw new Error(`Failed to update spaces: ${e.message}`); + } + } + } catch (e) { + console.error("Error in simulated transaction", e.message); + + message.retry({ + delaySeconds: calculateExponentialBackoff( + message.attempts, + BASE_DELAY_SECONDS, + ), + }); + throw new D1InsertError( + "Error when inserting into d1", + "D1 stuff after the vectorize", + ); + } + + // After the d1 and vectories suceeds then finally update the jobs table to indicate that the job has completed + + await db + .update(jobs) + .set({ status: "Processed" }) + .where(eq(jobs.id, jobId)); + + return; + } +} + +/* +To do: +Figure out rate limits!! + +*/ diff --git a/apps/cf-ai-backend/src/queueConsumer/utils/get-metadata.ts b/apps/cf-ai-backend/src/queueConsumer/utils/get-metadata.ts new file mode 100644 index 00000000..95916506 --- /dev/null +++ b/apps/cf-ai-backend/src/queueConsumer/utils/get-metadata.ts @@ -0,0 +1,57 @@ +import * as cheerio from "cheerio"; +import { Result, Ok, Err } from "../../errors/results"; +import { BaseError } from "../../errors/baseError"; + +class GetMetadataError extends BaseError { + constructor(message?: string, source?: string) { + super("[Fetch Metadata Error]", message, source); + } +} +export type Metadata = { + title: string; + description: string; + image: string; + baseUrl: string; +}; +// TODO: THIS SHOULD PROBABLY ALSO FETCH THE OG-IMAGE +export async function getMetaData( + url: string, +): Promise<Result<Metadata, GetMetadataError>> { + try { + const response = await fetch(url); + const html = await response.text(); + + const $ = cheerio.load(html); + + // Extract the base URL + const baseUrl = url; + + // Extract title + const title = $("title").text().trim(); + + const description = $("meta[name=description]").attr("content") ?? ""; + + const _favicon = + $("link[rel=icon]").attr("href") ?? "https://supermemory.dhr.wtf/web.svg"; + + let favicon = + _favicon.trim().length > 0 + ? _favicon.trim() + : "https://supermemory.dhr.wtf/web.svg"; + if (favicon.startsWith("/")) { + favicon = baseUrl + favicon; + } else if (favicon.startsWith("./")) { + favicon = baseUrl + favicon.slice(1); + } + + return Ok({ + title, + description, + image: favicon, + baseUrl, + }); + } catch (e) { + console.error("[Metadata Fetch Error]", e); + return Err(new GetMetadataError((e as Error).message, "getMetaData")); + } +} diff --git a/apps/cf-ai-backend/src/queueConsumer/utils/typeDecider.ts b/apps/cf-ai-backend/src/queueConsumer/utils/typeDecider.ts new file mode 100644 index 00000000..037ab40c --- /dev/null +++ b/apps/cf-ai-backend/src/queueConsumer/utils/typeDecider.ts @@ -0,0 +1,34 @@ +import { Result, Ok, Err } from "../../errors/results"; +import { BaseError } from "../../errors/baseError"; + +export type contentType = "page" | "tweet" | "note"; + +class GetTypeError extends BaseError { + constructor(message?: string, source?: string) { + super("[Decide Type Error]", message, source); + } +} +export const typeDecider = ( + content: string, +): Result<contentType, GetTypeError> => { + try { + // if the content is a URL, then it's a page. if its a URL with https://x.com/user/status/123, then it's a tweet. else, it's a note. + // do strict checking with regex + if ( + content.match(/https?:\/\/(x\.com|twitter\.com)\/[\w]+\/[\w]+\/[\d]+/) + ) { + return Ok("tweet"); + } else if ( + content.match( + /^(https?:\/\/)?(www\.)?[a-z0-9]+([-.]{1}[a-z0-9]+)*\.[a-z]{2,5}(\/.*)?$/i, + ) + ) { + return Ok("page"); + } else { + return Ok("note"); + } + } catch (e) { + console.error("[Decide Type Error]", e); + return Err(new GetTypeError((e as Error).message, "typeDecider")); + } +}; |