feat: partykit provider 🎈 (#146)

* feat: partykit provider https://www.loom.com/share/937d20d923d346119cf805071bf9c9b9?sid=5bea5447-6f99-462f-82a1-e899dc1d71b0 * changeset added
badass-courses · Apr 6, 2024 · c56c4f9 · c56c4f9
1 parent 5894844
commit c56c4f9
Show file tree

Hide file tree

Showing 16 changed files with 170 additions and 125 deletions.
diff --git a/.changeset/yellow-adults-talk.md b/.changeset/yellow-adults-talk.md
@@ -0,0 +1,5 @@
+---
+"@coursebuilder/core": patch
+---
+
+adds a partykit provider to core
diff --git a/apps/course-builder-web/src/inngest/functions/cloudinary/image-resource-created.ts b/apps/course-builder-web/src/inngest/functions/cloudinary/image-resource-created.ts
@@ -7,20 +7,15 @@ export const imageResourceCreated = inngest.createFunction(
 	{
 		event: IMAGE_RESOURCE_CREATED_EVENT,
 	},
-	async ({ event, step }) => {
+	async ({ event, step, partyProvider }) => {
 		await step.run('announce asset created', async () => {
-			await fetch(
-				`${env.NEXT_PUBLIC_PARTY_KIT_URL}/party/${env.NEXT_PUBLIC_PARTYKIT_ROOM_NAME}`,
-				{
-					method: 'POST',
-					body: JSON.stringify({
-						body: event.data,
-						requestId: event.data.resourceId,
-						name: 'image.resource.created',
-					}),
+			await partyProvider.broadcastMessage({
+				body: {
+					body: event.data,
+					requestId: event.data.resourceId,
+					name: 'image.resource.created',
 				},
-			).catch((e) => {
-				console.error(e)
+				roomId: env.NEXT_PUBLIC_PARTYKIT_ROOM_NAME,
 			})
 		})
 

diff --git a/apps/course-builder-web/src/inngest/functions/ocr/ocr-code-extractor.ts b/apps/course-builder-web/src/inngest/functions/ocr/ocr-code-extractor.ts
@@ -12,7 +12,7 @@ export const performCodeExtraction = inngest.createFunction(
 	{
 		event: OCR_WEBHOOK_EVENT,
 	},
-	async ({ event, step }) => {
+	async ({ event, step, partyProvider }) => {
 		const screenshotUrl = event.data.ocrWebhookEvent.screenshotUrl as string
 		const resourceId = event.data.ocrWebhookEvent.resourceId as string
 
@@ -33,26 +33,17 @@ export const performCodeExtraction = inngest.createFunction(
 		)
 
 		await step.run(`partykit broadcast [${resourceId}]`, async () => {
-			return await fetch(
-				`${env.NEXT_PUBLIC_PARTY_KIT_URL}/party/${resourceId}`,
-				{
-					method: 'POST',
-					body: JSON.stringify({
-						body: {
-							content: extractedCode.codeString,
-							role: 'assistant',
-						},
-						requestId: resourceId,
-						name: 'code.extraction.completed',
-					}),
+			return await partyProvider.broadcastMessage({
+				body: {
+					body: {
+						content: extractedCode.codeString,
+						role: 'assistant',
+					},
+					requestId: resourceId,
+					name: 'code.extraction.completed',
 				},
-			)
-				.then((res) => {
-					return res.text()
-				})
-				.catch((e) => {
-					console.error(e)
-				})
+				roomId: resourceId,
+			})
 		})
 
 		return extractedCode.codeString

diff --git a/apps/course-builder-web/src/inngest/inngest.server.ts b/apps/course-builder-web/src/inngest/inngest.server.ts
@@ -27,6 +27,7 @@ import {
 import { CourseBuilderCoreEvents } from '@coursebuilder/core/inngest/video-processing/events'
 import DeepgramProvider from '@coursebuilder/core/providers/deepgram'
 import OpenAIProvider from '@coursebuilder/core/providers/openai'
+import PartykitProvider from '@coursebuilder/core/providers/partykit'
 
 import {
 	CONCEPT_SELECTED,
@@ -62,12 +63,14 @@ const callbackBase =
 const middleware = createInngestMiddleware({
 	db: courseBuilderAdapter,
 	siteRootUrl: env.NEXT_PUBLIC_URL,
-	partyKitRootUrl: env.NEXT_PUBLIC_PARTY_KIT_URL,
 	mediaUploadProvider: new UTApi(),
 	openaiProvider: OpenAIProvider({
 		apiKey: env.OPENAI_API_KEY,
 		partyUrlBase: env.NEXT_PUBLIC_PARTY_KIT_URL,
 	}),
+	partyProvider: PartykitProvider({
+		partyUrlBase: env.NEXT_PUBLIC_PARTY_KIT_URL,
+	}),
 	transcriptProvider: DeepgramProvider({
 		apiKey: env.DEEPGRAM_API_KEY,
 		callbackUrl: `${callbackBase}/api/coursebuilder/webhook/deepgram`,

diff --git a/apps/pro-aws/src/inngest/functions/cloudinary/image-resource-created.ts b/apps/pro-aws/src/inngest/functions/cloudinary/image-resource-created.ts
@@ -7,20 +7,15 @@ export const imageResourceCreated = inngest.createFunction(
 	{
 		event: IMAGE_RESOURCE_CREATED_EVENT,
 	},
-	async ({ event, step }) => {
+	async ({ event, step, partyProvider }) => {
 		await step.run('announce asset created', async () => {
-			await fetch(
-				`${env.NEXT_PUBLIC_PARTY_KIT_URL}/party/${env.NEXT_PUBLIC_PARTYKIT_ROOM_NAME}`,
-				{
-					method: 'POST',
-					body: JSON.stringify({
-						body: event.data,
-						requestId: event.data.resourceId,
-						name: 'image.resource.created',
-					}),
+			await partyProvider.broadcastMessage({
+				body: {
+					body: event.data,
+					requestId: event.data.resourceId,
+					name: 'image.resource.created',
 				},
-			).catch((e) => {
-				console.error(e)
+				roomId: env.NEXT_PUBLIC_PARTYKIT_ROOM_NAME,
 			})
 		})
 

diff --git a/apps/pro-aws/src/inngest/functions/ocr/ocr-code-extractor.ts b/apps/pro-aws/src/inngest/functions/ocr/ocr-code-extractor.ts
@@ -12,7 +12,7 @@ export const performCodeExtraction = inngest.createFunction(
 	{
 		event: OCR_WEBHOOK_EVENT,
 	},
-	async ({ event, step }) => {
+	async ({ event, step, partyProvider }) => {
 		const screenshotUrl = event.data.ocrWebhookEvent.screenshotUrl as string
 		const resourceId = event.data.ocrWebhookEvent.resourceId as string
 
@@ -33,26 +33,17 @@ export const performCodeExtraction = inngest.createFunction(
 		)
 
 		await step.run(`partykit broadcast [${resourceId}]`, async () => {
-			return await fetch(
-				`${env.NEXT_PUBLIC_PARTY_KIT_URL}/party/${resourceId}`,
-				{
-					method: 'POST',
-					body: JSON.stringify({
-						body: {
-							content: extractedCode.codeString,
-							role: 'assistant',
-						},
-						requestId: resourceId,
-						name: 'code.extraction.completed',
-					}),
+			return await partyProvider.broadcastMessage({
+				body: {
+					body: {
+						content: extractedCode.codeString,
+						role: 'assistant',
+					},
+					requestId: resourceId,
+					name: 'code.extraction.completed',
 				},
-			)
-				.then((res) => {
-					return res.text()
-				})
-				.catch((e) => {
-					console.error(e)
-				})
+				roomId: resourceId,
+			})
 		})
 
 		return extractedCode.codeString

diff --git a/apps/pro-aws/src/inngest/inngest.server.ts b/apps/pro-aws/src/inngest/inngest.server.ts
@@ -1,4 +1,4 @@
-import { courseBuilderAdapter, db } from '@/db'
+import { courseBuilderAdapter } from '@/db'
 import { env } from '@/env.mjs'
 import {
 	EMAIL_SEND_BROADCAST,
@@ -27,6 +27,7 @@ import {
 import { CourseBuilderCoreEvents } from '@coursebuilder/core/inngest/video-processing/events'
 import DeepgramProvider from '@coursebuilder/core/providers/deepgram'
 import OpenAIProvider from '@coursebuilder/core/providers/openai'
+import PartykitProvider from '@coursebuilder/core/providers/partykit'
 
 import {
 	CONCEPT_SELECTED,
@@ -62,12 +63,14 @@ const callbackBase =
 const middleware = createInngestMiddleware({
 	db: courseBuilderAdapter,
 	siteRootUrl: env.NEXT_PUBLIC_URL,
-	partyKitRootUrl: env.NEXT_PUBLIC_PARTY_KIT_URL,
 	mediaUploadProvider: new UTApi(),
 	openaiProvider: OpenAIProvider({
 		apiKey: env.OPENAI_API_KEY,
 		partyUrlBase: env.NEXT_PUBLIC_PARTY_KIT_URL,
 	}),
+	partyProvider: PartykitProvider({
+		partyUrlBase: env.NEXT_PUBLIC_PARTY_KIT_URL,
+	}),
 	transcriptProvider: DeepgramProvider({
 		apiKey: env.DEEPGRAM_API_KEY,
 		callbackUrl: `${callbackBase}/api/coursebuilder/webhook/deepgram`,

diff --git a/packages/core/src/inngest/co-gardener/resource-chat.ts b/packages/core/src/inngest/co-gardener/resource-chat.ts
@@ -8,6 +8,7 @@ import { z } from 'zod'
 
 import { CourseBuilderAdapter } from '../../adapters'
 import { LlmProviderConfig } from '../../providers/openai'
+import { PartyProviderConfig } from '../../providers/partykit'
 import { User } from '../../schemas'
 import {
 	CoreInngestFunctionInput,
@@ -62,6 +63,7 @@ export const resourceChat: CoreInngestHandler = async ({
 	event,
 	step,
 	openaiProvider,
+	partyProvider,
 	db,
 }: CoreInngestFunctionInput) => {
 	const resourceId = event.data.resourceId
@@ -82,6 +84,7 @@ export const resourceChat: CoreInngestHandler = async ({
 	const messages = await resourceChatWorkflowExecutor({
 		db,
 		openaiProvider,
+		partyProvider,
 		step,
 		workflowTrigger,
 		resourceId,
@@ -117,9 +120,11 @@ export async function resourceChatWorkflowExecutor({
 	resource,
 	user,
 	openaiProvider,
+	partyProvider,
 	db,
 }: {
 	openaiProvider: LlmProviderConfig
+	partyProvider: PartyProviderConfig
 	db: CourseBuilderAdapter
 	resource: ChatResource
 	step: any
@@ -216,16 +221,14 @@ export async function resourceChatWorkflowExecutor({
 		await step.run(
 			`partykit broadcast user prompt [${resourceId}]`,
 			async () => {
-				await fetch(`${openaiProvider.partyUrlBase}/party/${resourceId}`, {
-					method: 'POST',
-					body: JSON.stringify({
+				await partyProvider.broadcastMessage({
+					body: {
 						body: currentUserMessage.content,
 						requestId: resourceId,
 						name: 'resource.chat.prompted',
 						userId: user.id,
-					}),
-				}).catch((e) => {
-					console.error(e)
+					},
+					roomId: resourceId,
 				})
 			},
 		)
@@ -255,20 +258,14 @@ export async function resourceChatWorkflowExecutor({
 	})
 
 	await step.run(`partykit broadcast [${resourceId}]`, async () => {
-		return await fetch(`${openaiProvider.partyUrlBase}/party/${resourceId}`, {
-			method: 'POST',
-			body: JSON.stringify({
+		return await partyProvider.broadcastMessage({
+			body: {
 				body: messages,
 				requestId: resourceId,
 				name: 'resource.chat.completed',
-			}),
+			},
+			roomId: resourceId,
 		})
-			.then((res) => {
-				return res.text()
-			})
-			.catch((e) => {
-				console.error(e)
-			})
 	})
 
 	return messages

diff --git a/packages/core/src/inngest/create-inngest-middleware.ts b/packages/core/src/inngest/create-inngest-middleware.ts
@@ -17,14 +17,18 @@ import {
 	type TranscriptionConfig,
 } from '../providers'
 import { LlmProviderConfig, MockOpenAIProvider } from '../providers/openai'
+import {
+	MockPartykitProvider,
+	PartyProviderConfig,
+} from '../providers/partykit'
 import { CourseBuilderCoreEvents } from './video-processing/events'
 
 export interface CoreInngestContext {
 	db: CourseBuilderAdapter
 	siteRootUrl: string
-	partyKitRootUrl: string
 	transcriptProvider: TranscriptionConfig
 	openaiProvider: LlmProviderConfig
+	partyProvider: PartyProviderConfig
 	mediaUploadProvider: {
 		deleteFiles: (fileKey: string) => Promise<{ success: boolean }>
 	}
@@ -74,6 +78,7 @@ export const coreInngest = new Inngest({
 			partyKitRootUrl: '',
 			transcriptProvider: MockTranscriptionProvider,
 			openaiProvider: MockOpenAIProvider,
+			partyProvider: MockPartykitProvider,
 			mediaUploadProvider: {
 				deleteFiles: async (_) => Promise.resolve({ success: true }),
 			},

diff --git a/packages/core/src/inngest/video-processing/functions/generate-transcript-with-screnshots.ts b/packages/core/src/inngest/video-processing/functions/generate-transcript-with-screnshots.ts
@@ -17,7 +17,7 @@ export const generateTranscriptWithScreenshotsTrigger: CoreInngestTrigger = {
 }
 
 export const generateTranscriptWithScreenshotsHandler: CoreInngestHandler =
-	async ({ event, step, db, partyKitRootUrl }: CoreInngestFunctionInput) => {
+	async ({ event, step, db, partyProvider }: CoreInngestFunctionInput) => {
 		const videoResourceId = event.data.videoResourceId
 
 		if (!videoResourceId) {
@@ -65,15 +65,13 @@ export const generateTranscriptWithScreenshotsHandler: CoreInngestHandler =
 		})
 
 		await step.run('send the transcript to the party', async () => {
-			await fetch(`${partyKitRootUrl}/party/${event.data.videoResourceId}`, {
-				method: 'POST',
-				body: JSON.stringify({
+			return await partyProvider.broadcastMessage({
+				body: {
 					body: transcriptWithScreenshots,
 					requestId: event.data.videoResourceId,
 					name: 'transcriptWithScreenshots.ready',
-				}),
-			}).catch((e) => {
-				console.error(e)
+				},
+				roomId: event.data.videoResourceId,
 			})
 		})
 

diff --git a/packages/core/src/inngest/video-processing/functions/transcript-ready.ts b/packages/core/src/inngest/video-processing/functions/transcript-ready.ts
@@ -16,7 +16,7 @@ const transcriptReadyTrigger: CoreInngestTrigger = {
 const transcriptReadyHandler: CoreInngestHandler = async ({
 	event,
 	step,
-	partyKitRootUrl,
+	partyProvider,
 	db,
 }: CoreInngestFunctionInput) => {
 	const videoResourceId = event.data.videoResourceId
@@ -52,15 +52,13 @@ const transcriptReadyHandler: CoreInngestHandler = async ({
 	}
 
 	await step.run('send the transcript to the party', async () => {
-		await fetch(`${partyKitRootUrl}/party/${videoResourceId}`, {
-			method: 'POST',
-			body: JSON.stringify({
+		return await partyProvider.broadcastMessage({
+			body: {
 				body: transcript,
 				requestId: videoResourceId,
 				name: 'transcript.ready',
-			}),
-		}).catch((e) => {
-			console.error(e)
+			},
+			roomId: videoResourceId,
 		})
 	})