[OSO-1497] Move publish notebook workflow to async infra

apps/frontend/app/api/v1/osograph/route.ts

@@ -7,6 +7,7 @@ import { gql } from "graphql-tag";
 import { getSystemCredentials, getUser } from "@/lib/auth/auth";
 import { buildSubgraphSchema } from "@apollo/subgraph";
 import { resolvers } from "@/app/api/v1/osograph/schema/resolvers";
+import { GraphQLContext } from "@/app/api/v1/osograph/types/context";
 
 function loadSchemas(): string {
   const schemaDir = path.join(
@@ -44,7 +45,7 @@ const apolloHandler = startServerAndCreateNextHandler<NextRequest>(server, {
   context: async (req) => {
     const user = await getUser(req);
     const systemCredentials = await getSystemCredentials(req);
-    return { req, user, systemCredentials };
+    return { req, user, systemCredentials } satisfies GraphQLContext;
   },
 });
 

apps/frontend/app/api/v1/osograph/schema/graphql/notebook.graphql

@@ -2,6 +2,7 @@ type Notebook {
   id: ID!
   orgId: ID!
   name: String!
+  data: String!
   description: String
   creatorId: ID!
   preview: String
@@ -59,7 +60,19 @@ extend type Mutation {
   """
   Save notebook preview image
   """
-  saveNotebookPreview(input: SaveNotebookPreviewInput!): SaveNotebookPreviewPayload!
+  saveNotebookPreview(
+    input: SaveNotebookPreviewInput!
+  ): SaveNotebookPreviewPayload!
+
+  """
+  Publish the notebook HTML to the public
+  """
+  publishNotebook(input: PublishNotebookInput!): PublishNotebookPayload!
+
+  """  
+  Unpublish the notebook
+  """
+  unpublishNotebook(notebookId: ID!): UnpublishNotebookPayload!
 }
 
 input CreateNotebookInput {
@@ -97,3 +110,18 @@ type SaveNotebookPreviewPayload {
   previewUrl: String
   message: String
 }
+
+input PublishNotebookInput {
+  notebookId: ID!
+}
+
+type PublishNotebookPayload {
+  success: Boolean!
+  message: String
+  run: Run!
+}
+
+type UnpublishNotebookPayload {
+  success: Boolean!
+  message: String
+}

apps/frontend/app/api/v1/osograph/schema/graphql/system.graphql

@@ -2,7 +2,6 @@
 Special type for system related reads that are only allowed by authenticated
 service accounts that have system privileges.
 """
-
 type SystemResolvedTableReference {
   reference: String!
   fqn: String!
@@ -25,7 +24,7 @@ extend type Mutation {
   System only. Mark a run as started.
   """
   startRun(input: StartRunInput!): StartRunPayload!
-  
+
   """
   System only. Update run metadata. This can be called at any time
   """
@@ -49,7 +48,16 @@ extend type Mutation {
   """
   System only. Create a materialization for a step
   """
-  createMaterialization(input: CreateMaterializationInput!): CreateMaterializationPayload!
+  createMaterialization(
+    input: CreateMaterializationInput!
+  ): CreateMaterializationPayload!
+
+  """
+  System only. Save the generated published notebook HTML to object storage
+  """
+  savePublishedNotebookHtml(
+    input: SavePublishedNotebookHtmlInput!
+  ): SavePublishedNotebookHtmlPayload!
 }
 
 extend type Query {
@@ -135,3 +143,13 @@ type CreateMaterializationPayload {
   message: String
   materialization: Materialization!
 }
+
+input SavePublishedNotebookHtmlInput {
+  notebookId: ID!
+  htmlContent: String!
+}
+
+type SavePublishedNotebookHtmlPayload {
+  success: Boolean!
+  message: String
+}

apps/frontend/app/api/v1/osograph/schema/resolvers/notebook.ts

@@ -28,6 +28,10 @@ import {
   validateInput,
 } from "@/app/api/v1/osograph/utils/validation";
 import { queryWithPagination } from "@/app/api/v1/osograph/utils/query-helpers";
+import { signOsoJwt } from "@/lib/auth/auth";
+import { createQueueService } from "@/lib/services/queue/factory";
+import { PublishNotebookRunRequest } from "@opensource-observer/osoprotobufs/publish-notebook";
+import { revalidateTag } from "next/cache";
 
 const PREVIEWS_BUCKET = "notebook-previews";
 const SIGNED_URL_EXPIRY = 900;
@@ -186,6 +190,126 @@ export const notebookResolvers: GraphQLResolverModule<GraphQLContext> = {
         throw ServerErrors.storage("Failed to save notebook preview");
       }
     },
+
+    publishNotebook: async (
+      _: unknown,
+      args: { input: { notebookId: string } },
+      context: GraphQLContext,
+    ) => {
+      const authenticatedUser = requireAuthentication(context.user);
+      const { notebookId } = args.input;
+
+      const supabase = createAdminClient();
+
+      const { data: notebook } = await supabase
+        .from("notebooks")
+        .select("id, organizations!inner(id, org_name)")
+        .eq("id", notebookId)
+        .single();
+
+      if (!notebook) {
+        throw NotebookErrors.notFound();
+      }
+
+      await requireOrgMembership(
+        authenticatedUser.userId,
+        notebook.organizations.id,
+      );
+
+      const osoToken = await signOsoJwt(authenticatedUser, {
+        orgId: notebook.organizations.id,
+        orgName: notebook.organizations.org_name,
+      });
+
+      const { data: queuedRun, error: queuedRunError } = await supabase
+        .from("run")
+        .insert({
+          org_id: notebook.organizations.id,
+          run_type: "manual",
+          requested_by: authenticatedUser.userId,
+        })
+        .select()
+        .single();
+      if (queuedRunError || !queuedRun) {
+        logger.error(
+          `Error creating run for notebook ${notebook.id}: ${queuedRunError?.message}`,
+        );
+        throw ServerErrors.database("Failed to create run request");
+      }
+
+      const queueService = createQueueService();
+
+      const runIdBuffer = Buffer.from(queuedRun.id.replace(/-/g, ""), "hex");
+      const publishMessage: PublishNotebookRunRequest = {
+        runId: new Uint8Array(runIdBuffer),
+        notebookId: notebook.id,
+        osoApiKey: osoToken,
+      };
+
+      const result = await queueService.queueMessage({
+        queueName: "publish_notebook_run_requests",
+        message: publishMessage,
+        encoder: PublishNotebookRunRequest,
+      });
+      if (!result.success) {
+        logger.error(
+          `Failed to publish message to queue: ${result.error?.message}`,
+        );
+        throw ServerErrors.queueError(
+          result.error?.message || "Failed to publish to queue",
+        );
+      }
+
+      return {
+        success: true,
+        run: queuedRun,
+        message: "Notebook publish run queued successfully",
+      };
+    },
+    unpublishNotebook: async (
+      _: unknown,
+      args: { notebookId: string },
+      context: GraphQLContext,
+    ) => {
+      const authenticatedUser = requireAuthentication(context.user);
+      const { notebookId } = args;
+
+      const supabase = createAdminClient();
+
+      const { data: publishedNotebook, error } = await supabase
+        .from("published_notebooks")
+        .select("*")
+        .eq("notebook_id", notebookId)
+        .single();
+      if (error) {
+        logger.log("Failed to find published notebook:", error);
+        throw NotebookErrors.notFound();
+      }
+      const { error: deleteError } = await supabase.storage
+        .from("published-notebooks")
+        .remove([publishedNotebook.data_path]);
+      if (deleteError) {
+        logger.log("Failed to delete notebook file:", deleteError);
+        throw ServerErrors.database("Failed to delete notebook file");
+      }
+      const { error: updateError } = await supabase
+        .from("published_notebooks")
+        .update({
+          deleted_at: new Date().toISOString(),
+          updated_by: authenticatedUser.userId,
+        })
+        .eq("id", publishedNotebook.id);
+      if (updateError) {
+        logger.log("Failed to delete notebook file:", updateError);
+        throw ServerErrors.database("Failed to delete notebook file");
+      }
+
+      revalidateTag(publishedNotebook.id);
+      return {
+        success: true,
+        message: "Notebook unpublished successfully",
+      };
+    },
   },
 
   Notebook: {

apps/frontend/app/api/v1/osograph/schema/resolvers/system.ts

@@ -22,10 +22,13 @@ import {
   StartStepSchema,
   validateInput,
   UpdateMetadataSchema,
+  SavePublishedNotebookHtmlSchema,
 } from "@/app/api/v1/osograph/utils/validation";
 import z from "zod";
 import { logger } from "@/lib/logger";
 import { Json } from "@/lib/types/supabase";
+import { generatePublishedNotebookPath } from "@/lib/notebook/utils";
+import { revalidateTag } from "next/cache";
 
 type SystemMutationOptions<T extends z.ZodTypeAny, O> = {
   inputSchema: T;
@@ -48,6 +51,7 @@ function systemMutation<T extends z.ZodTypeAny, O>({
     if (!context.systemCredentials) {
       throw AuthenticationErrors.notAuthorized();
     }
+
     const validatedInput = validateInput(inputSchema, args.input);
     return resolver(validatedInput, context);
   };
@@ -393,6 +397,76 @@ export const systemResolvers: GraphQLResolverModule<GraphQLContext> = {
         };
       },
     }),
+    savePublishedNotebookHtml: systemMutation({
+      inputSchema: SavePublishedNotebookHtmlSchema,
+      resolver: async (input) => {
+        const supabase = createAdminClient();
+
+        const { notebookId, htmlContent } = input;
+
+        console.log("Encoded", htmlContent);
+
+        // Decode base64 content
+        const byteArray = Buffer.from(htmlContent, "base64");
+
+        const { data: notebook } = await supabase
+          .from("notebooks")
+          .select("org_id")
+          .eq("id", notebookId)
+          .single();
+        if (!notebook) {
+          throw ResourceErrors.notFound(`Notebook ${notebookId} not found`);
+        }
+        const filePath = generatePublishedNotebookPath(
+          notebookId,
+          notebook.org_id,
+        );
+        // Save the HTML content to Supabase Storage
+        const { data: uploadData, error: uploadError } = await supabase.storage
+          .from("published-notebooks")
+          .upload(filePath, byteArray, {
+            upsert: true,
+            contentType: "text/html",
+            headers: {
+              "Content-Encoding": "gzip",
+            },
+            // 5 Minute CDN cache. We will also cache on Vercel side to control it with revalidateTag
+            cacheControl: "300",
+          });
+        if (uploadError || !uploadData) {
+          throw ServerErrors.internal(
+            `Failed to upload published notebook HTML for notebook ${notebookId}: ${uploadError.message}`,
+          );
+        }
+
+        // Update the published_notebooks table with the new data path
+        const { data: publishedNotebook, error: upsertError } = await supabase
+          .from("published_notebooks")
+          .upsert(
+            {
+              notebook_id: notebookId,
+              data_path: filePath,
+              updated_at: new Date().toISOString(),
+              deleted_at: null,
+            },
+            { onConflict: "notebook_id" },
+          )
+          .select("id")
+          .single();
+
+        if (upsertError || !publishedNotebook) {
+          throw ServerErrors.internal(
+            `Failed to update published_notebooks for notebook ${notebookId}: ${upsertError.message}`,
+          );
+        }
+
+        revalidateTag(publishedNotebook.id);
+        return {
+          message: "Saved published notebook HTML",
+          success: true,
+        };
+      },
+    }),
   },
   Query: {
     system: async (_: unknown, _args: unknown, context: GraphQLContext) => {

apps/frontend/app/api/v1/osograph/utils/validation.ts

@@ -307,6 +307,11 @@ export const CreateMaterializationSchema = z.object({
   schema: z.array(DataModelColumnSchema),
 });
 
+export const SavePublishedNotebookHtmlSchema = z.object({
+  notebookId: z.string().uuid(),
+  htmlContent: z.string(),
+});
+
 export const ResolveTablesSchema = z.object({
   references: z.array(z.string()).min(1, "At least one reference is required"),
   metadata: z