feat: add thread support for twitter; add segregation in types of chunks and methods to process those chunks

apps/cf-ai-backend/src/helper.ts

@@ -1,5 +1,5 @@
 import { Context } from "hono";
-import { Env, vectorObj } from "./types";
+import { Env, vectorObj, Chunks } from "./types";
 import { CloudflareVectorizeStore } from "@langchain/cloudflare";
 import { OpenAIEmbeddings } from "./utils/OpenAIEmbedder";
 import { createOpenAI } from "@ai-sdk/openai";
@@ -7,6 +7,7 @@ import { createGoogleGenerativeAI } from "@ai-sdk/google";
 import { createAnthropic } from "@ai-sdk/anthropic";
 import { z } from "zod";
 import { seededRandom } from "./utils/seededRandom";
+import { bulkInsertKv } from "./utils/kvBulkInsert";
 
 export async function initQuery(
 	c: Context<{ Bindings: Env }>,
@@ -135,7 +136,7 @@ export async function batchCreateChunksAndEmbeddings({
 }: {
 	store: CloudflareVectorizeStore;
 	body: z.infer<typeof vectorObj>;
-	chunks: string[];
+	chunks: Chunks;
 	context: Context<{ Bindings: Env }>;
 }) {
 	//! NOTE that we use #supermemory-web to ensure that
@@ -150,7 +151,6 @@ export async function batchCreateChunksAndEmbeddings({
 
 	const allIds = await context.env.KV.list({ prefix: uuid });
 
-	let pageContent = "";
 	// If some chunks for that content already exist, we'll just update the metadata to include
 	// the user.
 	if (allIds.keys.length > 0) {
@@ -169,51 +169,169 @@ export async function batchCreateChunksAndEmbeddings({
 					return acc;
 				}, {}),
 			};
-			const content =
-				vector.metadata.content.toString().split("Content: ")[1] ||
-				vector.metadata.content;
-			pageContent += `<---chunkId: ${vector.id}\n${content}\n---->`;
 			return vector;
 		});
 
 		await context.env.VECTORIZE_INDEX.upsert(newVectors);
-		return pageContent; //Return the page content that goes to d1 db
+		return;
 	}
 
-	for (let i = 0; i < chunks.length; i++) {
-		const chunk = chunks[i];
-		const chunkId = `${uuid}-${i}`;
-
-		const newPageContent = `Title: ${body.title}\nDescription: ${body.description}\nURL: ${body.url}\nContent: ${chunk}`;
-
-		const docs = await store.addDocuments(
-			[
-				{
-					pageContent: newPageContent,
-					metadata: {
-						title: body.title?.slice(0, 50) ?? "",
-						description: body.description ?? "",
-						url: body.url,
-						type: body.type ?? "page",
-						content: newPageContent,
-
-						[sanitizeKey(`user-${body.user}`)]: 1,
-						...body.spaces?.reduce((acc, space) => {
-							acc[`space-${body.user}-${space}`] = 1;
-							return acc;
-						}, {}),
-					},
-				},
-			],
+	switch (chunks.type) {
+		case "tweet":
 			{
-				ids: [chunkId],
-			},
-		);
+				const commonMetaData = {
+					type: body.type ?? "tweet",
+					title: body.title,
+					description: body.description ?? "",
+					url: body.url,
+					[sanitizeKey(`user-${body.user}`)]: 1,
+				};
+				const spaceMetadata = body.spaces?.reduce((acc, space) => {
+					acc[`space-${body.user}-${space}`] = 1;
+					return acc;
+				}, {});
 
-		console.log("Docs added: ", docs);
+				const ids = [];
+				const preparedDocuments = chunks.chunks
+					.map((tweet, i) => {
+						return tweet.chunkedTweet.map((chunk) => {
+							const id = `${uuid}-${i}`;
+							ids.push(id);
+							const { tweetLinks, tweetVids, tweetId } = tweet.metadata;
+							return {
+								pageContent: chunk,
+								metadata: {
+									links: tweetLinks,
+									videos: tweetVids,
+									tweetId: tweetId,
+									...commonMetaData,
+									...spaceMetadata,
+								},
+							};
+						});
+					})
+					.flat();
 
-		await context.env.KV.put(chunkId, ourID);
-		pageContent += `<---chunkId: ${chunkId}\n${chunk}\n---->`;
+				const docs = await store.addDocuments(preparedDocuments, {
+					ids: ids,
+				});
+				console.log("these are the doucment ids", ids);
+				console.log("Docs added:", docs);
+				const { CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID } =
+					context.env;
+				await bulkInsertKv(
+					{ CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID },
+					{ chunkIds: ids, urlid: ourID },
+				);
+			}
+			break;
+		case "page":
+			{
+				const commonMetaData = {
+					type: body.type ?? "page",
+					title: body.title,
+					description: body.description ?? "",
+					url: body.url,
+					[sanitizeKey(`user-${body.user}`)]: 1,
+				};
+				const spaceMetadata = body.spaces?.reduce((acc, space) => {
+					acc[`space-${body.user}-${space}`] = 1;
+					return acc;
+				}, {});
+
+				const ids = [];
+				const preparedDocuments = chunks.chunks.map((chunk, i) => {
+					const id = `${uuid}-${i}`;
+					ids.push(id);
+					return {
+						pageContent: chunk,
+						metadata: {
+							...commonMetaData,
+							...spaceMetadata,
+						},
+					};
+				});
+
+				const docs = await store.addDocuments(preparedDocuments, { ids: ids });
+				console.log("Docs added:", docs);
+				const { CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID } =
+					context.env;
+				await bulkInsertKv(
+					{ CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID },
+					{ chunkIds: ids, urlid: ourID },
+				);
+			}
+			break;
+		case "note":
+			{
+				const commonMetaData = {
+					type: body.type ?? "page",
+					description: body.description ?? "",
+					url: body.url,
+					[sanitizeKey(`user-${body.user}`)]: 1,
+				};
+				const spaceMetadata = body.spaces?.reduce((acc, space) => {
+					acc[`space-${body.user}-${space}`] = 1;
+					return acc;
+				}, {});
+
+				const ids = [];
+				const preparedDocuments = chunks.chunks.map((chunk, i) => {
+					const id = `${uuid}-${i}`;
+					ids.push(id);
+					return {
+						pageContent: chunk,
+						metadata: {
+							...commonMetaData,
+							...spaceMetadata,
+						},
+					};
+				});
+
+				const docs = await store.addDocuments(preparedDocuments, { ids: ids });
+				console.log("Docs added:", docs);
+				const { CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID } =
+					context.env;
+				await bulkInsertKv(
+					{ CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID },
+					{ chunkIds: ids, urlid: ourID },
+				);
+			}
+			break;
+		case "image": {
+			const commonMetaData = {
+				type: body.type ?? "image",
+				title: body.title,
+				description: body.description ?? "",
+				url: body.url,
+				[sanitizeKey(`user-${body.user}`)]: 1,
+			};
+			const spaceMetadata = body.spaces?.reduce((acc, space) => {
+				acc[`space-${body.user}-${space}`] = 1;
+				return acc;
+			}, {});
+
+			const ids = [];
+			const preparedDocuments = chunks.chunks.map((chunk, i) => {
+				const id = `${uuid}-${i}`;
+				ids.push(id);
+				return {
+					pageContent: chunk,
+					metadata: {
+						...commonMetaData,
+						...spaceMetadata,
+					},
+				};
+			});
+
+			const docs = await store.addDocuments(preparedDocuments, { ids: ids });
+			console.log("Docs added:", docs);
+			const { CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID } = context.env;
+			await bulkInsertKv(
+				{ CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID },
+				{ chunkIds: ids, urlid: ourID },
+			);
+		}
 	}
-	return pageContent; // Return the pageContent  that goes to the d1 db
+	return;
 }

apps/cf-ai-backend/src/index.ts

@@ -1,7 +1,15 @@
 import { z } from "zod";
 import { Hono } from "hono";
 import { CoreMessage, generateText, streamText, tool } from "ai";
-import { chatObj, Env, vectorObj } from "./types";
+import {
+	chatObj,
+	Chunks,
+	Env,
+	ImageChunks,
+	PageOrNoteChunks,
+	TweetChunks,
+	vectorObj,
+} from "./types";
 import {
 	batchCreateChunksAndEmbeddings,
 	deleteDocument,
@@ -15,6 +23,8 @@ import { zValidator } from "@hono/zod-validator";
 import chunkText from "./utils/chonker";
 import { systemPrompt, template } from "./prompts/prompt1";
 import { swaggerUI } from "@hono/swagger-ui";
+import { chunkThread } from "./utils/chunkTweet";
+import { chunkNote, chunkPage } from "./utils/chunkPageOrNotes";
 
 const app = new Hono<{ Bindings: Env }>();
 
@@ -64,26 +74,41 @@ app.post("/api/add", zValidator("json", vectorObj), async (c) => {
 	const { store } = await initQuery(c);
 
 	console.log(body.spaces);
-
+	let chunks: TweetChunks | PageOrNoteChunks;
 	// remove everything in <raw> tags
 	const newPageContent = body.pageContent?.replace(/<raw>.*?<\/raw>/g, "");
 
-	const chunks = chunkText(newPageContent, 1536);
-	if (chunks.length > 20) {
+	switch (body.type) {
+		case "tweet":
+			chunks = chunkThread(newPageContent);
+			break;
+
+		case "page":
+			chunks = chunkPage(newPageContent);
+			break;
+
+		case "note":
+			chunks = chunkNote(newPageContent);
+			break;
+	}
+
+	console.log(JSON.stringify(chunks));
+
+	if (chunks.chunks.length > 20) {
 		return c.json({
 			status: "error",
 			message:
 				"We are unable to process documents this size just yet, try something smaller",
 		});
 	}
-	const chunkedInput = await batchCreateChunksAndEmbeddings({
+	await batchCreateChunksAndEmbeddings({
 		store,
 		body,
 		chunks: chunks,
 		context: c,
 	});
 
-	return c.json({ status: "ok", chunkedInput });
+	return c.json({ status: "ok" });
 });
 
 app.post(
@@ -136,6 +161,13 @@ app.post(
 		);
 
 		const imageDescriptions = await Promise.all(imagePromises);
+		const chunks: ImageChunks = {
+			type: "image",
+			chunks: [
+				imageDescriptions,
+				...(body.text ? chunkText(body.text, 1536) : []),
+			].flat(),
+		};
 
 		await batchCreateChunksAndEmbeddings({
 			store,
@@ -151,10 +183,7 @@ app.post(
 				pageContent: imageDescriptions.join("\n"),
 				title: "Image content from the web",
 			},
-			chunks: [
-				imageDescriptions,
-				...(body.text ? chunkText(body.text, 1536) : []),
-			].flat(),
+			chunks: chunks,
 			context: c,
 		});
 
@@ -252,7 +281,7 @@ app.post(
 		// This is a "router". this finds out if the user wants to add a document, or chat with the AI to get a response.
 		const routerQuery = await generateText({
 			model: model,
-			system: `You are Supermemory chatbot. You can either add a document to the supermemory database, or return a chat response. Based on this query, 
+			system: `You are Supermemory chatbot. You can either add a document to the supermemory database, or return a chat response. Based on this query,
         You must determine what to do. Basically if it feels like a "question", then you should intiate a chat. If it feels like a "command" or feels like something that could be forwarded to the AI, then you should add a document.
         You must also extract the "thing" to add and what type of thing it is.`,
 			prompt: `Question from user: ${query}`,
@@ -278,7 +307,9 @@ app.post(
 
 		if ((task as string) === "add") {
 			// addString is the plaintext string that the user wants to add to the database
+			//chunk the note
 			let addString: string = addContent;
+			let vectorContent: Chunks = chunkNote(addContent);
 
 			if (thingToAdd === "page") {
 				// TODO: Sometimes this query hangs, and errors out. we need to do proper error management here.
@@ -289,6 +320,7 @@ app.post(
 				});
 
 				addString = await response.text();
+				vectorContent = chunkPage(addString);
 			}
 
 			// At this point, we can just go ahead and create the embeddings!
@@ -301,7 +333,7 @@ app.post(
 					pageContent: addString,
 					title: `${addString.slice(0, 30)}... (Added from chatbot)`,
 				},
-				chunks: chunkText(addString, 1536),
+				chunks: vectorContent,
 				context: c,
 			});
 
@@ -494,7 +526,6 @@ app.post(
 				);
 
 				const metadata = normalizedData.map((datapoint) => datapoint.metadata);
-
 				return c.json({ ids: storedContent, metadata, normalizedData });
 			}
 		}

apps/cf-ai-backend/src/types.ts

@@ -1,5 +1,6 @@
 import { sourcesZod } from "@repo/shared-types";
 import { z } from "zod";
+import { ThreadTweetData } from "./utils/chunkTweet";
 
 export type Env = {
 	VECTORIZE_INDEX: VectorizeIndex;
@@ -7,6 +8,9 @@ export type Env = {
 	SECURITY_KEY: string;
 	OPENAI_API_KEY: string;
 	GOOGLE_AI_API_KEY: string;
+	CF_KV_AUTH_TOKEN: string;
+	KV_NAMESPACE_ID: string;
+	CF_ACCOUNT_ID: string;
 	MY_QUEUE: Queue<TweetData[]>;
 	KV: KVNamespace;
 	MYBROWSER: unknown;
@@ -23,6 +27,32 @@ export interface TweetData {
 	saveToUser: string;
 }
 
+interface BaseChunks {
+	type: "tweet" | "page" | "note" | "image";
+}
+
+export interface TweetChunks extends BaseChunks {
+	type: "tweet";
+	chunks: Array<ThreadTweetData>;
+}
+
+export interface PageOrNoteChunks extends BaseChunks {
+	type: "page" | "note";
+	chunks: string[];
+}
+export interface ImageChunks extends BaseChunks {
+	type: "image";
+	chunks: string[];
+}
+
+export type Chunks = TweetChunks | PageOrNoteChunks | ImageChunks;
+
+export interface KVBulkItem {
+	key: string;
+	value: string;
+	base64: boolean;
+}
+
 export const contentObj = z.object({
 	role: z.string(),
 	parts: z

apps/cf-ai-backend/src/utils/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 };
+}

apps/cf-ai-backend/src/utils/chunkTweet.ts

@@ -0,0 +1,39 @@
+import { TweetChunks } from "../types";
+import chunkText from "./chonker";
+
+interface Tweet {
+	id: string;
+	text: string;
+	links: Array<string>;
+	images: Array<string>;
+	videos: Array<string>;
+}
+interface Metadata {
+	tweetId: string;
+	tweetLinks: any[];
+	tweetVids: any[];
+}
+
+export interface ThreadTweetData {
+	chunkedTweet: string[];
+	metadata: Metadata;
+}
+
+
+export function chunkThread(threadText: string): TweetChunks {
+	const thread = JSON.parse(threadText);
+
+	const chunkedTweets = thread.map((tweet: Tweet) => {
+		const chunkedTweet = chunkText(tweet.text, 1536);
+
+		const metadata = {
+			tweetId: tweet.id,
+			tweetLinks: tweet.links,
+			tweetVids: tweet.videos,
+		};
+
+		return { chunkedTweet, metadata };
+	});
+
+	return { type: "tweet", chunks: chunkedTweets };
+}

apps/cf-ai-backend/src/utils/kvBulkInsert.ts

@@ -0,0 +1,43 @@
+import { KVBulkItem } from "../types";
+
+export const bulkInsertKv = async (
+	credentials: {
+		CF_KV_AUTH_TOKEN: string;
+		KV_NAMESPACE_ID: string;
+		CF_ACCOUNT_ID: string;
+	},
+	keyData: {
+		chunkIds: Array<string>;
+		urlid: string;
+	},
+) => {
+	const data: Array<KVBulkItem> = keyData.chunkIds.map((chunkId) => ({
+		key: chunkId,
+		value: keyData.urlid,
+		base64: false,
+	}));
+
+	try {
+		const response = await fetch(
+			`https://api.cloudflare.com/client/v4/accounts/${credentials.CF_ACCOUNT_ID}/storage/kv/namespaces/${credentials.KV_NAMESPACE_ID}/bulk`,
+			{
+				method: "PUT",
+				headers: {
+					Authorization: `Bearer ${credentials.CF_KV_AUTH_TOKEN}`,
+					"Content-Type": "application/json",
+				},
+				body: JSON.stringify(data),
+			},
+		);
+
+		if (!response.ok) {
+			throw new Error(
+				`can't insert bulk to kv because ${response.status} ${response.statusText} ${JSON.stringify(response.body)}`,
+			);
+		}
+		return await response.json();
+	} catch (e) {
+		//dosomething
+		throw e;
+	}
+};

apps/web/app/actions/doers.ts

@@ -60,7 +60,7 @@ export const createSpace = async (
 	}
 };
 
-const typeDecider = (content: string) => {
+const typeDecider = (content: string): "page" | "tweet" | "note" => {
 	// 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]+/)) {
@@ -171,6 +171,7 @@ export const createMemory = async (input: {
 
 	let pageContent = input.content;
 	let metadata: Awaited<ReturnType<typeof getMetaData>>;
+	let vectorData: string;
 
 	if (!(await limit(data.user.id, type))) {
 		return {
@@ -189,7 +190,7 @@ export const createMemory = async (input: {
 			},
 		});
 		pageContent = await response.text();
-
+		vectorData = pageContent;
 		try {
 			metadata = await getMetaData(input.content);
 		} catch (e) {
@@ -199,8 +200,42 @@ export const createMemory = async (input: {
 			};
 		}
 	} else if (type === "tweet") {
+		//Request the worker for the entire thread
+
+		let thread: string;
+		let errorOccurred: boolean = false;
+
+		try {
+			const cf_thread_endpoint = process.env.THREAD_CF_WORKER;
+			const authKey = process.env.THREAD_CF_AUTH;
+
+			const threadRequest = await fetch(cf_thread_endpoint, {
+				method: "POST",
+				headers: {
+					"Content-Type": "application/json",
+					Authorization: authKey,
+				},
+				body: JSON.stringify({ url: input.content }),
+			});
+
+			if (threadRequest.status !== 200) {
+				throw new Error(
+					`Failed to fetch the thread: ${input.content}, Reason: ${threadRequest.statusText}`,
+				);
+			}
+
+			thread = await threadRequest.text();
+		} catch (e) {
+			console.log("[THREAD FETCHING SERVICE] Failed to fetch the thread", e);
+			errorOccurred = true;
+		}
+
 		const tweet = await getTweetData(input.content.split("/").pop() as string);
+
 		pageContent = tweetToMd(tweet);
+		console.log("THis ishte page content!!", pageContent);
+		//@ts-ignore
+		vectorData = errorOccurred ? pageContent : thread;
 		metadata = {
 			baseUrl: input.content,
 			description: tweet.text.slice(0, 200),
@@ -209,6 +244,7 @@ export const createMemory = async (input: {
 		};
 	} else if (type === "note") {
 		pageContent = input.content;
+		vectorData = pageContent;
 		noteId = new Date().getTime();
 		metadata = {
 			baseUrl: `https://supermemory.ai/note/${noteId}`,
@@ -235,7 +271,7 @@ export const createMemory = async (input: {
 		{
 			method: "POST",
 			body: JSON.stringify({
-				pageContent,
+				pageContent: vectorData,
 				title: metadata.title,
 				description: metadata.description,
 				url: metadata.baseUrl,

apps/web/cf-env.d.ts

@@ -18,6 +18,9 @@ declare global {
 			CLOUDFLARE_DATABASE_ID: string;
 			CLOUDFLARE_D1_TOKEN: string;
 
+			THREAD_CF_WORKER: string;
+			THREAD_CF_AUTH: string;
+
 			MOBILE_TRUST_TOKEN: string;
 		}
 	}