queues so far

Co-authored-by: Dhravya Shah <hello@dhravya.dev>

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

@@ -0,0 +1,7 @@
+import { drizzle } from "drizzle-orm/d1";
+import { Env } from "../types";
+
+import * as schema from "./schema";
+
+export const database = (env: Env) =>
+	drizzle(env.DATABASE, { schema, logger: true });

apps/cf-ai-backend/src/db/schema.ts

@@ -0,0 +1,93 @@
+import {
+	index,
+	int,
+	primaryKey,
+	sqliteTableCreator,
+	text,
+	integer,
+} from "drizzle-orm/sqlite-core";
+
+export const createTable = sqliteTableCreator((name) => `${name}`);
+
+export const users = createTable(
+	"user",
+	{
+		id: text("id")
+			.primaryKey()
+			.$defaultFn(() => crypto.randomUUID()),
+		name: text("name"),
+		email: text("email").notNull(),
+		emailVerified: integer("emailVerified", { mode: "timestamp_ms" }),
+		image: text("image"),
+		telegramId: text("telegramId"),
+		hasOnboarded: integer("hasOnboarded", { mode: "boolean" }).default(false),
+	},
+	(user) => ({
+		emailIdx: index("users_email_idx").on(user.email),
+		telegramIdx: index("users_telegram_idx").on(user.telegramId),
+		idIdx: index("users_id_idx").on(user.id),
+	}),
+);
+
+export type User = typeof users.$inferSelect;
+
+export const storedContent = createTable(
+	"storedContent",
+	{
+		id: integer("id").notNull().primaryKey({ autoIncrement: true }),
+		content: text("content").notNull(),
+		title: text("title", { length: 255 }),
+		description: text("description", { length: 255 }),
+		url: text("url").notNull(),
+		savedAt: int("savedAt", { mode: "timestamp" }).notNull(),
+		baseUrl: text("baseUrl", { length: 255 }).unique(),
+		ogImage: text("ogImage", { length: 255 }),
+		type: text("type").default("page"),
+		image: text("image", { length: 255 }),
+		userId: text("user").references(() => users.id, {
+			onDelete: "cascade",
+		}),
+		noteId: integer("noteId"),
+	},
+	(sc) => ({
+		urlIdx: index("storedContent_url_idx").on(sc.url),
+		savedAtIdx: index("storedContent_savedAt_idx").on(sc.savedAt),
+		titleInx: index("storedContent_title_idx").on(sc.title),
+		userIdx: index("storedContent_user_idx").on(sc.userId),
+	}),
+);
+
+export type Content = typeof storedContent.$inferSelect;
+
+
+export const contentToSpace = createTable(
+	"contentToSpace",
+	{
+		contentId: integer("contentId")
+			.notNull()
+			.references(() => storedContent.id, { onDelete: "cascade" }),
+		spaceId: integer("spaceId")
+			.notNull()
+			.references(() => space.id, { onDelete: "cascade" }),
+	},
+	(cts) => ({
+		compoundKey: primaryKey({ columns: [cts.contentId, cts.spaceId] }),
+	}),
+);
+
+export const space = createTable(
+	"space",
+	{
+		id: integer("id").notNull().primaryKey({ autoIncrement: true }),
+		name: text("name").notNull().unique().default("none"),
+		user: text("user", { length: 255 }).references(() => users.id, {
+			onDelete: "cascade",
+		}),
+		createdAt: int("createdAt", { mode: "timestamp" }).notNull(),
+		numItems: integer("numItems").notNull().default(0),
+	},
+	(space) => ({
+		nameIdx: index("spaces_name_idx").on(space.name),
+		userIdx: index("spaces_user_idx").on(space.user),
+	}),
+);

apps/cf-ai-backend/src/errors/baseError.ts

@@ -0,0 +1,46 @@
+export class BaseHttpError extends Error {
+    public status: number;
+    public message: string;
+  
+    constructor(status: number, message: string) {
+      super(message);
+      this.status = status;
+      this.message = message;
+      Object.setPrototypeOf(this, new.target.prototype); // Restore prototype chain
+    }
+  }
+  
+  
+  export class BaseError extends Error {
+    type: string;
+    message: string;
+    source: string;
+    ignoreLog: boolean;
+  
+    constructor(
+      type: string,
+      message?: string,
+      source?: string,
+      ignoreLog = false
+    ) {
+      super();
+  
+      Object.setPrototypeOf(this, new.target.prototype);
+  
+      this.type = type;
+      this.message =
+        message ??
+        "An unknown error occurred. If this persists, please contact us.";
+      this.source = source ?? "unspecified";
+      this.ignoreLog = ignoreLog;
+    }
+  
+    toJSON(): Record<PropertyKey, string> {
+      return {
+        type: this.type,
+        message: this.message,
+        source: this.source,
+      };
+    }
+  }
+  

apps/cf-ai-backend/src/errors/results.ts

@@ -0,0 +1,28 @@
+import { BaseError } from "./baseError";
+
+export type Result<T, E extends Error> =
+	| { ok: true; value: T }
+	| { ok: false; error: E };
+
+export const Ok = <T>(data: T): Result<T, never> => {
+	return { ok: true, value: data };
+};
+
+export const Err = <E extends BaseError>(error: E): Result<never, E> => {
+	return { ok: false, error };
+};
+
+export async function wrap<T, E extends BaseError>(
+	p: Promise<T>,
+	errorFactory: (err: Error) => E,
+): Promise<Result<T, E>> {
+	try {
+		return Ok(await p);
+	} catch (e) {
+		return Err(errorFactory(e as Error));
+	}
+}
+
+export function isErr<T, E extends Error>(result: Result<T, E>): result is { ok: false; error: E } {
+    return !result.ok;
+  }

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

@@ -132,12 +132,12 @@ export async function batchCreateChunksAndEmbeddings({
 	store,
 	body,
 	chunks,
-	context,
+	env: env,
 }: {
 	store: CloudflareVectorizeStore;
 	body: z.infer<typeof vectorObj>;
 	chunks: Chunks;
-	context: Context<{ Bindings: Env }>;
+	env: Env;
 }) {
 	//! NOTE that we use #supermemory-web to ensure that
 	//! If a user saves it through the extension, we don't want other users to be able to see it.
@@ -149,7 +149,7 @@ export async function batchCreateChunksAndEmbeddings({
 		random().toString(36).substring(2, 15) +
 		random().toString(36).substring(2, 15);
 
-	const allIds = await context.env.KV.list({ prefix: uuid });
+	const allIds = await env.KV.list({ prefix: uuid });
 
 	// If some chunks for that content already exist, we'll just update the metadata to include
 	// the user.
@@ -159,7 +159,7 @@ export async function batchCreateChunksAndEmbeddings({
 		//Search in a batch of 20
 		for (let i = 0; i < savedVectorIds.length; i += 20) {
 			const batch = savedVectorIds.slice(i, i + 20);
-			const batchVectors = await context.env.VECTORIZE_INDEX.getByIds(batch);
+			const batchVectors = await env.VECTORIZE_INDEX.getByIds(batch);
 			vectors.push(...batchVectors);
 		}
 		console.log(
@@ -192,7 +192,7 @@ export async function batchCreateChunksAndEmbeddings({
 
 		await Promise.all(
 			results.map((result) => {
-				return context.env.VECTORIZE_INDEX.upsert(result);
+				return env.VECTORIZE_INDEX.upsert(result);
 			}),
 		);
 		return;
@@ -243,8 +243,7 @@ export async function batchCreateChunksAndEmbeddings({
 				});
 				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;
+				const { CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID } = env;
 				await bulkInsertKv(
 					{ CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID },
 					{ chunkIds: ids, urlid: ourID },
@@ -281,8 +280,7 @@ export async function batchCreateChunksAndEmbeddings({
 
 				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;
+				const { CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID } = env;
 				await bulkInsertKv(
 					{ CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID },
 					{ chunkIds: ids, urlid: ourID },
@@ -319,8 +317,7 @@ export async function batchCreateChunksAndEmbeddings({
 
 				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;
+				const { CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID } = env;
 				await bulkInsertKv(
 					{ CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID },
 					{ chunkIds: ids, urlid: ourID },
@@ -355,7 +352,7 @@ export async function batchCreateChunksAndEmbeddings({
 
 			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;
+			const { CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID } = env;
 			await bulkInsertKv(
 				{ CF_KV_AUTH_TOKEN, CF_ACCOUNT_ID, KV_NAMESPACE_ID },
 				{ chunkIds: ids, urlid: ourID },

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

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { boolean, z } from "zod";
 import { Hono } from "hono";
 import { CoreMessage, generateText, streamText, tool } from "ai";
 import {
@@ -9,6 +9,7 @@ import {
 	PageOrNoteChunks,
 	TweetChunks,
 	vectorObj,
+	vectorBody,
 } from "./types";
 import {
 	batchCreateChunksAndEmbeddings,
@@ -20,11 +21,15 @@ import { logger } from "hono/logger";
 import { poweredBy } from "hono/powered-by";
 import { bearerAuth } from "hono/bearer-auth";
 import { zValidator } from "@hono/zod-validator";
-import chunkText from "./utils/chonker";
+import chunkText from "./queueConsumer/chunkers/chonker";
 import { systemPrompt, template } from "./prompts/prompt1";
 import { swaggerUI } from "@hono/swagger-ui";
-import { chunkThread } from "./utils/chunkTweet";
-import { chunkNote, chunkPage } from "./utils/chunkPageOrNotes";
+// import { chunkThread } from "./utils/chunkTweet";
+import {
+	chunkNote,
+	chunkPage,
+} from "./queueConsumer/chunkers/chunkPageOrNotes";
+import { queue } from "./queueConsumer";
 
 const app = new Hono<{ Bindings: Env }>();
 
@@ -68,38 +73,45 @@ app.get("/api/health", (c) => {
 	return c.json({ status: "ok" });
 });
 
-app.post("/api/add", zValidator("json", vectorObj), async (c) => {
+app.post("/api/add", zValidator("json", vectorBody), async (c) => {
 	try {
+		// console.log("api/add hit!!!!");
 		const body = c.req.valid("json");
-
-		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, "");
-
-		switch (body.type) {
-			case "tweet":
-				chunks = chunkThread(body.pageContent);
-				break;
-
-			case "page":
-				chunks = chunkPage(body.pageContent);
-				break;
-
-			case "note":
-				chunks = chunkNote(body.pageContent);
-				break;
-		}
-
-		await batchCreateChunksAndEmbeddings({
-			store,
-			body,
-			chunks: chunks,
-			context: c,
+		const spaceNumbers = body.spaces.map((s: string) => Number(s));
+		await c.env.EMBEDCHUNKS_QUEUE.send({
+			content: body.url,
+			user: body.user,
+			space: spaceNumbers,
 		});
 
+		// 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, "");
+
+		// switch (body.type) {
+		// 	case "tweet":
+		// 		chunks = chunkThread(body.pageContent);
+		// 		break;
+
+		// 	case "page":
+		// 		chunks = chunkPage(body.pageContent);
+		// 		break;
+
+		// 	case "note":
+		// 		chunks = chunkNote(body.pageContent);
+		// 		break;
+		// }
+
+		// await batchCreateChunksAndEmbeddings({
+		// 	store,
+		// 	body,
+		// 	chunks: chunks,
+		// 	env: c,
+		// });
+
 		return c.json({ status: "ok" });
 	} catch (error) {
 		console.error("Error processing request:", error);
@@ -180,7 +192,7 @@ app.post(
 				title: "Image content from the web",
 			},
 			chunks: chunks,
-			context: c,
+			env: c.env,
 		});
 
 		return c.json({ status: "ok" });
@@ -330,7 +342,7 @@ app.post(
 					title: `${addString.slice(0, 30)}... (Added from chatbot)`,
 				},
 				chunks: vectorContent,
-				context: c,
+				env: c.env,
 			});
 
 			return c.json({
@@ -664,4 +676,7 @@ app.get(
 	},
 );
 
-export default app;
+export default {
+	fetch: app.fetch,
+	queue,
+};

apps/cf-ai-backend/src/queueConsumer/chunkers/chunkPageOrNotes.ts

@@ -1,5 +1,5 @@
 import chunkText from "./chonker";
-import { PageOrNoteChunks } from "../types";
+import { PageOrNoteChunks } from "../../types";
 export function chunkPage(pageContent: string): PageOrNoteChunks {
 	const chunks = chunkText(pageContent, 1536);
 

apps/cf-ai-backend/src/queueConsumer/chunkers/chunkTweet.ts

@@ -1,4 +1,4 @@
-import { TweetChunks } from "../types";
+import { TweetChunks } from "../../types";
 import chunkText from "./chonker";
 import { getRawTweet } from "@repo/shared-types/utils";
 

apps/cf-ai-backend/src/queueConsumer/helpers/initQuery.ts

@@ -0,0 +1,58 @@
+import { Env } from "../../types";
+import { OpenAIEmbeddings } from "../../utils/OpenAIEmbedder";
+import { CloudflareVectorizeStore } from "@langchain/cloudflare";
+import { createOpenAI } from "@ai-sdk/openai";
+import { createGoogleGenerativeAI } from "@ai-sdk/google";
+import { createAnthropic } from "@ai-sdk/anthropic";
+
+export async function initQQuery(
+	env: Env,
+	model: string = "gpt-4o",
+) {
+	const embeddings = new OpenAIEmbeddings({
+		apiKey: env.OPENAI_API_KEY,
+		modelName: "text-embedding-3-small",
+	});
+
+	const store = new CloudflareVectorizeStore(embeddings, {
+		index: env.VECTORIZE_INDEX,
+	});
+
+	let selectedModel:
+		| ReturnType<ReturnType<typeof createOpenAI>>
+		| ReturnType<ReturnType<typeof createGoogleGenerativeAI>>
+		| ReturnType<ReturnType<typeof createAnthropic>>;
+
+	switch (model) {
+		case "claude-3-opus":
+			const anthropic = createAnthropic({
+				apiKey: env.ANTHROPIC_API_KEY,
+				baseURL:
+					"https://gateway.ai.cloudflare.com/v1/47c2b4d598af9d423c06fc9f936226d5/supermemory/anthropic",
+			});
+			selectedModel = anthropic.chat("claude-3-opus-20240229");
+			console.log("Selected model: ", selectedModel);
+			break;
+		case "gemini-1.5-pro":
+			const googleai = createGoogleGenerativeAI({
+				apiKey: env.GOOGLE_AI_API_KEY,
+				baseURL:
+					"https://gateway.ai.cloudflare.com/v1/47c2b4d598af9d423c06fc9f936226d5/supermemory/google-vertex-ai",
+			});
+			selectedModel = googleai.chat("models/gemini-1.5-pro-latest");
+			console.log("Selected model: ", selectedModel);
+			break;
+		case "gpt-4o":
+		default:
+			const openai = createOpenAI({
+				apiKey: env.OPENAI_API_KEY,
+				baseURL:
+					"https://gateway.ai.cloudflare.com/v1/47c2b4d598af9d423c06fc9f936226d5/supermemory/openai",
+				compatibility: "strict",
+			});
+			selectedModel = openai.chat("gpt-4o-mini");
+			break;
+	}
+
+	return { store, model: selectedModel };
+}

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"));
+	}
+}

apps/cf-ai-backend/src/queueConsumer/helpers/processPage.ts

@@ -0,0 +1,42 @@
+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(
+	url: string,
+): Promise<Result<PageProcessResult, ProcessPageError>> {
+	try {
+		const response = await fetch("https://md.dhr.wtf/?url=" + url, {
+			headers: {
+				Authorization: "Bearer " + process.env.BACKEND_SECURITY_KEY,
+			},
+		});
+		const pageContent = await response.text();
+		if (!pageContent) {
+			return Err(
+				new ProcessPageError(
+					"Failed to get response form markdowner",
+					"processPage",
+				),
+			);
+		}
+		console.log("[This is the page content]", pageContent);
+		const metadataResult = await getMetaData(url);
+		if (isErr(metadataResult)) {
+			throw metadataResult.error;
+		}
+		const metadata = metadataResult.value;
+		return Ok({ pageContent, metadata });
+	} catch (e) {
+		console.error("[Page Processing Error]", e);
+		return Err(new ProcessPageError((e as Error).message, "processPage"));
+	}
+}

apps/cf-ai-backend/src/queueConsumer/helpers/processTweet.ts

@@ -0,0 +1,81 @@
+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?
+
+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 (
+	tweetUrl: string,
+	cf_thread_endpoint: string,
+	authKey: string,
+): Promise<Result<string, ProcessTweetError>> => {
+	const threadRequest = await fetch(cf_thread_endpoint, {
+		method: "POST",
+		headers: {
+			"Content-Type": "application/json",
+			Authorization: authKey,
+		},
+		body: JSON.stringify({ url: tweetUrl }),
+	});
+	if (threadRequest.status !== 200) {
+		return Err(
+			new ProcessTweetError(
+				`Failed to fetch the thread: ${tweetUrl}, Reason: ${threadRequest.statusText}`,
+				"getThreadData",
+			),
+		);
+	}
+
+	const thread = await threadRequest.text();
+	console.log("[thread response]");
+
+	if (thread.trim().length === 2) {
+		console.log("Thread is an empty array");
+		return Err(
+			new ProcessTweetError(
+				"[THREAD FETCHING SERVICE] Got no content form thread worker",
+				"getThreadData",
+			),
+		);
+	}
+	return Ok(thread);
+};

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

@@ -0,0 +1,204 @@
+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 { initQQuery } from "./helpers/initQuery";
+import { chunkNote, chunkPage } from "./chunkers/chunkPageOrNotes";
+import { chunkThread } from "./chunkers/chunkTweet";
+import { batchCreateChunksAndEmbeddings } 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 } from "../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);
+	}
+}
+
+export async function queue(
+	batch: MessageBatch<{ content: string; space: Array<number>; user: string }>,
+	env: Env,
+): Promise<void> {
+	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;
+		console.log("v got shit in the queue", body);
+
+		const typeResult = typeDecider(body.content);
+
+		if (isErr(typeResult)) {
+			throw typeResult.error;
+		}
+		console.log(typeResult.value);
+		const type = typeResult.value;
+
+		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(body.content);
+				if (isErr(page)) {
+					throw page.error;
+				}
+				pageContent = page.value.pageContent;
+				metadata = page.value.metadata;
+				vectorData = pageContent;
+				chunks = chunkPage(pageContent);
+				break;
+			}
+
+			case "tweet": {
+				console.log("tweet hit");
+				console.log(body.content.split("/").pop());
+				const tweet = await getTweetData(body.content.split("/").pop());
+				console.log(tweet);
+				const thread = await getThreadData(
+					body.content,
+					env.THREAD_CF_WORKER,
+					env.THREAD_CF_AUTH,
+				);
+
+				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 {
+					vectorData = thread.value;
+				}
+				chunks = chunkThread(vectorData);
+				break;
+			}
+		}
+
+		// see what's up with the storedToSpaces in this block
+		const { store } = await initQQuery(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,
+		};
+		const vectorResult = await wrap(
+			batchCreateChunksAndEmbeddings({
+				store: store,
+				body: Chunkbody,
+				chunks: chunks,
+				env: env,
+			}),
+			vectorErrorFactory,
+		);
+
+		if (isErr(vectorResult)) {
+			throw vectorResult.error;
+		}
+		const saveToDbUrl =
+			(metadata.baseUrl.split("#supermemory-user-")[0] ?? metadata.baseUrl) +
+			"#supermemory-user-" +
+			body.user;
+		let contentId: number;
+		const db = database(env);
+		const insertResponse = await 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 });
+
+		if (!insertResponse[0]?.id) {
+			throw new D1InsertError(
+				"something went worng when inserting to database",
+				"inresertResponse",
+			);
+		}
+		contentId = insertResponse[0]?.id;
+		if (storeToSpaces.length > 0) {
+			// Adding the many-to-many relationship between content and spaces
+			const spaceData = await db
+				.select()
+				.from(space)
+				.where(and(inArray(space.id, storeToSpaces), eq(space.user, body.user)))
+				.all();
+
+			await Promise.all(
+				spaceData.map(async (s) => {
+					await db
+						.insert(contentToSpace)
+						.values({ contentId: contentId, spaceId: s.id });
+
+					await db.update(space).set({ numItems: s.numItems + 1 });
+				}),
+			);
+		}
+	}
+}
+
+/*
+To do: 
+1. Abstract and shitft the entrie creatememory function to the queue consumer --> Hopefully done
+2. Make the front end use that instead of whatever khichidi is going on right now 
+3. remove getMetada form the lib file as it's not being used anywhere else 
+4. Figure out the limit stuff ( server action for that seems fine because no use in limiting after they already in the queue rigth? )
+5. Figure out the initQuery stuff ( ;( ) --> This is a bad way of doing stuff :0 
+6. How do I hande the content already exists wala use case?
+7. Figure out retry and not add shit to the vectirze over and over again on failure
+*/

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"));
+	}
+}

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"));
+	}
+};

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

@@ -1,6 +1,6 @@
 import { sourcesZod } from "@repo/shared-types";
 import { z } from "zod";
-import { ThreadTweetData } from "./utils/chunkTweet";
+import { ThreadTweetData } from "./queueConsumer/chunkers/chunkTweet";
 
 export type Env = {
 	VECTORIZE_INDEX: VectorizeIndex;
@@ -11,13 +11,23 @@ export type Env = {
 	CF_KV_AUTH_TOKEN: string;
 	KV_NAMESPACE_ID: string;
 	CF_ACCOUNT_ID: string;
+	DATABASE: D1Database;
 	MY_QUEUE: Queue<TweetData[]>;
 	KV: KVNamespace;
+	EMBEDCHUNKS_QUEUE: Queue<JobData>;
 	MYBROWSER: unknown;
 	ANTHROPIC_API_KEY: string;
+	THREAD_CF_AUTH: string;
+	THREAD_CF_WORKER: string;
 	NODE_ENV: string;
 };
 
+export interface JobData {
+	content: string;
+	space: Array<number>;
+	user: string;
+}
+
 export interface TweetData {
 	tweetText: string;
 	postUrl: string;
@@ -80,3 +90,8 @@ export const vectorObj = z.object({
 	user: z.string(),
 	type: z.string().optional().default("page"),
 });
+export const vectorBody = z.object({
+	spaces: z.array(z.string()).optional(),
+	url: z.string(),
+	user: z.string(),
+});

apps/web/app/actions/doers.ts

@@ -15,7 +15,7 @@ import {
 import { ServerActionReturnType } from "./types";
 import { auth } from "../../server/auth";
 import { Tweet } from "react-tweet/api";
-import { getMetaData } from "@/lib/get-metadata";
+// import { getMetaData } from "@/lib/get-metadata";
 import { and, eq, inArray, sql } from "drizzle-orm";
 import { LIMITS } from "@/lib/constants";
 import { ChatHistory } from "@repo/shared-types";
@@ -197,122 +197,16 @@ export const createMemory = async (input: {
 		return { error: "Not authenticated", success: false };
 	}
 
-	const type = typeDecider(input.content);
-
-	let pageContent = input.content;
-	let metadata: Awaited<ReturnType<typeof getMetaData>>;
-	let vectorData: string;
-
-	if (!(await limit(data.user.id, type))) {
-		return {
-			success: false,
-			data: 0,
-			error: `You have exceeded the limit of ${LIMITS[type as keyof typeof LIMITS]} ${type}s.`,
-		};
-	}
-
-	let noteId = 0;
-
-	if (type === "page") {
-		const response = await fetch("https://md.dhr.wtf/?url=" + input.content, {
-			headers: {
-				Authorization: "Bearer " + process.env.BACKEND_SECURITY_KEY,
-			},
-		});
-		pageContent = await response.text();
-		vectorData = pageContent;
-		try {
-			metadata = await getMetaData(input.content);
-		} catch (e) {
-			return {
-				success: false,
-				error: "Failed to fetch metadata for the page. Please try again later.",
-			};
-		}
-	} 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();
-			if (thread.trim().length === 2) {
-				console.log("Thread is an empty array");
-				throw new Error(
-					"[THREAD FETCHING SERVICE] Got no content form thread worker",
-				);
-			}
-		} 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 ? JSON.stringify(pageContent) : thread;
-		metadata = {
-			baseUrl: input.content,
-			description: tweet.text.slice(0, 200),
-			image: tweet.user.profile_image_url_https,
-			title: `Tweet by ${tweet.user.name}`,
-		};
-	} else if (type === "note") {
-		pageContent = input.content;
-		vectorData = pageContent;
-		noteId = new Date().getTime();
-		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 ? "..." : ""}`,
-		};
-	} else {
-		return {
-			success: false,
-			data: 0,
-			error: "Invalid type",
-		};
-	}
-
-	let storeToSpaces = input.spaces;
-
-	if (!storeToSpaces) {
-		storeToSpaces = [];
-	}
-
-	const vectorSaveResponse = await fetch(
+	
+	// make the backend reqeust for the queue here
+	const vectorSaveResponses = await fetch(
 		`${process.env.BACKEND_BASE_URL}/api/add`,
 		{
 			method: "POST",
 			body: JSON.stringify({
-				pageContent: vectorData,
-				title: metadata.title,
-				description: metadata.description,
-				url: metadata.baseUrl,
-				spaces: storeToSpaces.map((spaceId) => spaceId.toString()),
+				url: input.content,
+				spaces: input.spaces,
 				user: data.user.id,
-				type,
 			}),
 			headers: {
 				"Content-Type": "application/json",
@@ -321,125 +215,249 @@ export const createMemory = async (input: {
 		},
 	);
 
-	if (!vectorSaveResponse.ok) {
-		const errorData = await vectorSaveResponse.text();
-		console.error(errorData);
-		return {
-			success: false,
-			data: 0,
-			error: `Failed to save to vector store. Backend returned error: ${errorData}`,
-		};
-	}
+// 	const type = typeDecider(input.content);
 
-	let contentId: number;
+// 	let pageContent = input.content;
+// 	let metadata: Awaited<ReturnType<typeof getMetaData>>;
+// 	let vectorData: string;
 
-	const response = (await vectorSaveResponse.json()) as {
-		status: string;
-		chunkedInput: string;
-		message?: string;
-	};
+// 	if (!(await limit(data.user.id, type))) {
+// 		return {
+// 			success: false,
+// 			data: 0,
+// 			error: `You have exceeded the limit of ${LIMITS[type as keyof typeof LIMITS]} ${type}s.`,
+// 		};
+// 	}
 
-	try {
-		if (response.status !== "ok") {
-			if (response.status === "error") {
-				return {
-					success: false,
-					data: 0,
-					error: response.message,
-				};
-			} else {
-				return {
-					success: false,
-					data: 0,
-					error: `Failed to save to vector store. Backend returned error: ${response.message}`,
-				};
-			}
-		}
-	} catch (e) {
-		return {
-			success: false,
-			data: 0,
-			error: `Failed to save to vector store. Backend returned error: ${e}`,
-		};
-	}
+// 	let noteId = 0;
 
-	const saveToDbUrl =
-		(metadata.baseUrl.split("#supermemory-user-")[0] ?? metadata.baseUrl) +
-		"#supermemory-user-" +
-		data.user.id;
+// 	if (type === "page") {
+// 		const response = await fetch("https://md.dhr.wtf/?url=" + input.content, {
+// 			headers: {
+// 				Authorization: "Bearer " + process.env.BACKEND_SECURITY_KEY,
+// 			},
+// 		});
+// 		pageContent = await response.text();
+// 		vectorData = pageContent;
+// 		try {
+// 			metadata = await getMetaData(input.content);
+// 		} catch (e) {
+// 			return {
+// 				success: false,
+// 				error: "Failed to fetch metadata for the page. Please try again later.",
+// 			};
+// 		}
+// 	} else if (type === "tweet") {
+// 		//Request the worker for the entire thread
 
-	// Insert into database
-	try {
-		const insertResponse = await db
-			.insert(storedContent)
-			.values({
-				content: pageContent,
-				title: metadata.title,
-				description: metadata.description,
-				url: saveToDbUrl,
-				baseUrl: saveToDbUrl,
-				image: metadata.image,
-				savedAt: new Date(),
-				userId: data.user.id,
-				type,
-				noteId,
-			})
-			.returning({ id: storedContent.id });
-		revalidatePath("/memories");
-		revalidatePath("/home");
+// 		let thread: string;
+// 		let errorOccurred: boolean = false;
 
-		if (!insertResponse[0]?.id) {
-			return {
-				success: false,
-				data: 0,
-				error: "Something went wrong while saving the document to the database",
-			};
-		}
+// 		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 }),
+// 			});
 
-		contentId = insertResponse[0]?.id;
-	} catch (e) {
-		const error = e as Error;
-		console.log("Error: ", error.message);
+// 			if (threadRequest.status !== 200) {
+// 				throw new Error(
+// 					`Failed to fetch the thread: ${input.content}, Reason: ${threadRequest.statusText}`,
+// 				);
+// 			}
 
-		if (
-			error.message.includes(
-				"D1_ERROR: UNIQUE constraint failed: storedContent.baseUrl",
-			)
-		) {
-			return {
-				success: false,
-				data: 0,
-				error: "Content already exists",
-			};
-		}
+// 			thread = await threadRequest.text();
+// 			if (thread.trim().length === 2) {
+// 				console.log("Thread is an empty array");
+// 				throw new Error(
+// 					"[THREAD FETCHING SERVICE] Got no content form thread worker",
+// 				);
+// 			}
+// 		} catch (e) {
+// 			console.log("[THREAD FETCHING SERVICE] Failed to fetch the thread", e);
+// 			errorOccurred = true;
+// 		}
 
-		return {
-			success: false,
-			data: 0,
-			error: "Failed to save to database with error: " + error.message,
-		};
-	}
+// 		const tweet = await getTweetData(input.content.split("/").pop() as string);
 
-	if (storeToSpaces.length > 0) {
-		// Adding the many-to-many relationship between content and spaces
-		const spaceData = await db
-			.select()
-			.from(space)
-			.where(
-				and(inArray(space.id, storeToSpaces), eq(space.user, data.user.id)),
-			)
-			.all();
+// 		pageContent = tweetToMd(tweet);
+// 		console.log("THis ishte page content!!", pageContent);
+// 		//@ts-ignore
+// 		vectorData = errorOccurred ? JSON.stringify(pageContent) : thread;
+// 		metadata = {
+// 			baseUrl: input.content,
+// 			description: tweet.text.slice(0, 200),
+// 			image: tweet.user.profile_image_url_https,
+// 			title: `Tweet by ${tweet.user.name}`,
+// 		};
+// 	} else if (type === "note") {
+// 		pageContent = input.content;
+// 		vectorData = pageContent;
+// 		noteId = new Date().getTime();
+// 		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 ? "..." : ""}`,
+// 		};
+// 	} else {
+// 		return {
+// 			success: false,
+// 			data: 0,
+// 			error: "Invalid type",
+// 		};
+// 	}
 
-		await Promise.all(
-			spaceData.map(async (s) => {
-				await db
-					.insert(contentToSpace)
-					.values({ contentId: contentId, spaceId: s.id });
+// 	let storeToSpaces = input.spaces;
 
-				await db.update(space).set({ numItems: s.numItems + 1 });
-			}),
-		);
-	}
+// 	if (!storeToSpaces) {
+// 		storeToSpaces = [];
+// 	}
+
+// 	const vectorSaveResponse = await fetch(
+// 		`${process.env.BACKEND_BASE_URL}/api/add`,
+// 		{
+// 			method: "POST",
+// 			body: JSON.stringify({
+// 				pageContent: vectorData,
+// 				title: metadata.title,
+// 				description: metadata.description,
+// 				url: metadata.baseUrl,
+// 				spaces: storeToSpaces.map((spaceId) => spaceId.toString()),
+// 				user: data.user.id,
+// 				type,
+// 			}),
+// 			headers: {
+// 				"Content-Type": "application/json",
+// 				Authorization: "Bearer " + process.env.BACKEND_SECURITY_KEY,
+// 			},
+// 		},
+// 	);
+
+// 	if (!vectorSaveResponse.ok) {
+// 		const errorData = await vectorSaveResponse.text();
+// 		console.error(errorData);
+// 		return {
+// 			success: false,
+// 			data: 0,
+// 			error: `Failed to save to vector store. Backend returned error: ${errorData}`,
+// 		};
+// 	}
+
+// 	let contentId: number;
+
+// 	const response = (await vectorSaveResponse.json()) as {
+// 		status: string;
+// 		chunkedInput: string;
+// 		message?: string;
+// 	};
+
+// 	try {
+// 		if (response.status !== "ok") {
+// 			if (response.status === "error") {
+// 				return {
+// 					success: false,
+// 					data: 0,
+// 					error: response.message,
+// 				};
+// 			} else {
+// 				return {
+// 					success: false,
+// 					data: 0,
+// 					error: `Failed to save to vector store. Backend returned error: ${response.message}`,
+// 				};
+// 			}
+// 		}
+// 	} catch (e) {
+// 		return {
+// 			success: false,
+// 			data: 0,
+// 			error: `Failed to save to vector store. Backend returned error: ${e}`,
+// 		};
+// 	}
+
+// 	const saveToDbUrl =
+// 		(metadata.baseUrl.split("#supermemory-user-")[0] ?? metadata.baseUrl) +
+// 		"#supermemory-user-" +
+// 		data.user.id;
+
+// 	// Insert into database
+// 	try {
+// 		const insertResponse = await db
+// 			.insert(storedContent)
+// 			.values({
+// 				content: pageContent,
+// 				title: metadata.title,
+// 				description: metadata.description,
+// 				url: saveToDbUrl,
+// 				baseUrl: saveToDbUrl,
+// 				image: metadata.image,
+// 				savedAt: new Date(),
+// 				userId: data.user.id,
+// 				type,
+// 				noteId,
+// 			})
+// 			.returning({ id: storedContent.id });
+// 		revalidatePath("/memories");
+// 		revalidatePath("/home");
+
+// 		if (!insertResponse[0]?.id) {
+// 			return {
+// 				success: false,
+// 				data: 0,
+// 				error: "Something went wrong while saving the document to the database",
+// 			};
+// 		}
+
+// 		contentId = insertResponse[0]?.id;
+// 	} catch (e) {
+// 		const error = e as Error;
+// 		console.log("Error: ", error.message);
+
+// 		if (
+// 			error.message.includes(
+// 				"D1_ERROR: UNIQUE constraint failed: storedContent.baseUrl",
+// 			)
+// 		) {
+// 			return {
+// 				success: false,
+// 				data: 0,
+// 				error: "Content already exists",
+// 			};
+// 		}
+
+// 		return {
+// 			success: false,
+// 			data: 0,
+// 			error: "Failed to save to database with error: " + error.message,
+// 		};
+// 	}
+
+// 	if (storeToSpaces.length > 0) {
+// 		// Adding the many-to-many relationship between content and spaces
+// 		const spaceData = await db
+// 			.select()
+// 			.from(space)
+// 			.where(
+// 				and(inArray(space.id, storeToSpaces), eq(space.user, data.user.id)),
+// 			)
+// 			.all();
+
+// 		await Promise.all(
+// 			spaceData.map(async (s) => {
+// 				await db
+// 					.insert(contentToSpace)
+// 					.values({ contentId: contentId, spaceId: s.id });
+
+// 				await db.update(space).set({ numItems: s.numItems + 1 });
+// 			}),
+// 		);
+// 	}
 
 	return {
 		success: true,
@@ -457,6 +475,7 @@ export const createChatThread = async (
 		return { error: "Not authenticated", success: false };
 	}
 
+	
 	const thread = await db
 		.insert(chatThreads)
 		.values({

apps/web/lib/get-metadata.ts

@@ -1,40 +0,0 @@
-"use server";
-import * as cheerio from "cheerio";
-
-// TODO: THIS SHOULD PROBABLY ALSO FETCH THE OG-IMAGE
-export async function getMetaData(url: string) {
-	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);
-	}
-
-	// Prepare the metadata object
-	const metadata = {
-		title,
-		description,
-		image: favicon,
-		baseUrl,
-	};
-	return metadata;
-}

apps/web/migrations/0000_omniscient_stick.sql

@@ -43,7 +43,7 @@ CREATE TABLE `chatHistory` (
 	`answerParts` text,
 	`answerSources` text,
 	`answerJustification` text,
-	`createdAt` integer DEFAULT '"2024-07-25T22:31:50.848Z"' NOT NULL,
+	`createdAt` integer DEFAULT '"2024-07-29T17:06:56.122Z"' NOT NULL,
 	FOREIGN KEY (`threadId`) REFERENCES `chatThread`(`id`) ON UPDATE no action ON DELETE cascade
 );
 --> statement-breakpoint
@@ -62,6 +62,19 @@ CREATE TABLE `contentToSpace` (
 	FOREIGN KEY (`spaceId`) REFERENCES `space`(`id`) ON UPDATE no action ON DELETE cascade
 );
 --> statement-breakpoint
+CREATE TABLE `jobs` (
+	`id` integer PRIMARY KEY AUTOINCREMENT NOT NULL,
+	`userId` text NOT NULL,
+	`url` text NOT NULL,
+	`status` text NOT NULL,
+	`attempts` integer DEFAULT 0 NOT NULL,
+	`lastAttemptAt` integer,
+	`error` blob,
+	`createdAt` integer NOT NULL,
+	`updatedAt` integer NOT NULL,
+	FOREIGN KEY (`userId`) REFERENCES `user`(`id`) ON UPDATE no action ON DELETE cascade
+);
+--> statement-breakpoint
 CREATE TABLE `session` (
 	`sessionToken` text PRIMARY KEY NOT NULL,
 	`userId` text NOT NULL,
@@ -122,6 +135,10 @@ CREATE UNIQUE INDEX `authenticator_credentialID_unique` ON `authenticator` (`cre
 CREATE INDEX `canvas_user_userId` ON `canvas` (`userId`);--> statement-breakpoint
 CREATE INDEX `chatHistory_thread_idx` ON `chatHistory` (`threadId`);--> statement-breakpoint
 CREATE INDEX `chatThread_user_idx` ON `chatThread` (`userId`);--> statement-breakpoint
+CREATE INDEX `jobs_userId_idx` ON `jobs` (`userId`);--> statement-breakpoint
+CREATE INDEX `jobs_status_idx` ON `jobs` (`status`);--> statement-breakpoint
+CREATE INDEX `jobs_createdAt_idx` ON `jobs` (`createdAt`);--> statement-breakpoint
+CREATE INDEX `jobs_url_idx` ON `jobs` (`url`);--> statement-breakpoint
 CREATE UNIQUE INDEX `space_name_unique` ON `space` (`name`);--> statement-breakpoint
 CREATE INDEX `spaces_name_idx` ON `space` (`name`);--> statement-breakpoint
 CREATE INDEX `spaces_user_idx` ON `space` (`user`);--> statement-breakpoint

apps/web/migrations/meta/_journal.json

@@ -1,13 +1,13 @@
 {
-	"version": "6",
-	"dialect": "sqlite",
-	"entries": [
-		{
-			"idx": 0,
-			"version": "6",
-			"when": 1721946710900,
-			"tag": "0000_steep_moira_mactaggert",
-			"breakpoints": true
-		}
-	]
-}
+  "version": "6",
+  "dialect": "sqlite",
+  "entries": [
+    {
+      "idx": 0,
+      "version": "6",
+      "when": 1722272816127,
+      "tag": "0000_omniscient_stick",
+      "breakpoints": true
+    }
+  ]
+}

apps/web/server/db/schema.ts

@@ -7,6 +7,7 @@ import {
 	sqliteTableCreator,
 	text,
 	integer,
+	blob,
 } from "drizzle-orm/sqlite-core";
 import type { AdapterAccountType } from "next-auth/adapters";
 
@@ -242,3 +243,28 @@ export const canvas = createTable(
 
 export type ChatThread = typeof chatThreads.$inferSelect;
 export type ChatHistory = typeof chatHistory.$inferSelect;
+
+export const jobs = createTable(
+	"jobs",
+	{
+		id: integer("id").notNull().primaryKey({ autoIncrement: true }),
+		userId: text("userId")
+			.notNull()
+			.references(() => users.id, { onDelete: "cascade" }),
+		url: text("url").notNull(),
+		status: text("status").notNull(),
+		attempts: integer("attempts").notNull().default(0),
+		lastAttemptAt: integer("lastAttemptAt"),
+		error: blob("error"),
+		createdAt: integer("createdAt").notNull(),
+		updatedAt: integer("updatedAt").notNull(),
+	},
+	(job) => ({
+		userIdx: index("jobs_userId_idx").on(job.userId),
+		statusIdx: index("jobs_status_idx").on(job.status),
+		createdAtIdx: index("jobs_createdAt_idx").on(job.createdAt),
+		urlIdx: index("jobs_url_idx").on(job.url),
+	}),
+);
+
+export type Job = typeof jobs.$inferSelect;