Merge pull request #193 from supermemoryai/kush/be-queue

Kush/be queue

3 files changed, 167 insertions, 0 deletions

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