feat(api): add procedure for upserting weaveVM blob storage references

.changeset/hot-bobcats-train.md

@@ -0,0 +1,5 @@
+---
+"@blobscan/api": minor
+---
+
+Added a protected procedure to upsert weaveVM blob storage references

.env.test

@@ -26,7 +26,7 @@ SWARM_BATCH_ID=f89e63edf757f06e89933761d6d46592d03026efb9871f9d244f34da86b6c242
 
 FILE_SYSTEM_STORAGE_PATH=test-blobscan-blobs
 
-
+WEAVEVM_API_KEY=weavevm-api-key
 # Blob Propagator
 
 BLOB_PROPAGATOR_TMP_BLOB_STORAGE=FILE_SYSTEM

apps/docs/src/app/docs/environment/page.md

@@ -57,6 +57,7 @@ nextjs:
 | `FILE_SYSTEM_STORAGE_PATH`           | Store blobs in this path                                                                           | No                              | `/tmp/blobscan-blobs`      |
 | `WEAVEVM_STORAGE_ENABLED`            | Weavevm storage usage                                                                              | No                              | `false`                    |
 | `WEAVEVM_STORAGE_API_BASE_URL`       | Weavevm API base url                                                                               | No                              | (empty)                    |
+| `WEAVEVM_API_KEY`                    | API key required to authenticate requests to the WeaveVM blob storage reference updater endpoint   | No                              | (empty)                    |
 | `STATS_SYNCER_DAILY_CRON_PATTERN`    | Cron pattern for the daily stats job                                                               | No                              | `30 0 * * * *`             |
 | `STATS_SYNCER_OVERALL_CRON_PATTERN`  | Cron pattern for the overall stats job                                                             | No                              | `*/15 * * * *`             |
 | `SWARM_STAMP_CRON_PATTERN`           | Cron pattern for swarm job                                                                         | No                              | `*/15 * * * *`             |

packages/api/src/context.ts

@@ -7,7 +7,6 @@ import type {
   NodeHTTPResponse,
 } from "@trpc/server/adapters/node-http";
 import cookie from "cookie";
-import jwt from "jsonwebtoken";
 
 import type { BlobPropagator } from "@blobscan/blob-propagator";
 import { getBlobPropagator } from "@blobscan/blob-propagator";
@@ -17,6 +16,8 @@ import { prisma } from "@blobscan/db";
 import { env } from "@blobscan/env";
 
 import { PostHogClient, shouldIgnoreURL } from "./posthog";
+import type { APIClient } from "./utils";
+import { retrieveAPIClient } from "./utils";
 
 type NextHTTPRequest = CreateNextContextOptions["req"];
 
@@ -25,38 +26,14 @@ export type CreateContextOptions =
   | CreateNextContextOptions;
 
 type CreateInnerContextOptions = Partial<CreateContextOptions> & {
-  apiClient: string | null;
+  apiClient?: APIClient;
 };
 
-export function getJWTFromRequest(req: NodeHTTPRequest | NextHTTPRequest) {
-  const authHeader = req.headers.authorization;
-  if (!authHeader) {
-    return null;
-  }
-
-  try {
-    const [type, token] = authHeader.split(" ");
-    if (type !== "Bearer" || !token) {
-      return null;
-    }
-
-    const decoded = jwt.verify(token, env.SECRET_KEY) as string;
-
-    return decoded;
-  } catch (err) {
-    if (err instanceof jwt.JsonWebTokenError) {
-      return null;
-    }
-
-    throw new TRPCError({ code: "BAD_REQUEST" });
-  }
-}
-
 export type TRPCInnerContext = {
   prisma: typeof prisma;
   blobStorageManager: BlobStorageManager;
-  blobPropagator: BlobPropagator | undefined;
-  apiClient: string | null | undefined;
+  blobPropagator?: BlobPropagator;
+  apiClient?: APIClient;
 };
 
 export async function createTRPCInnerContext(
@@ -92,7 +69,7 @@ export function createTRPCContext(
 ) {
   return async (opts: CreateContextOptions) => {
     try {
-      const apiClient = getJWTFromRequest(opts.req);
+      const apiClient = retrieveAPIClient(opts.req);
 
       const innerContext = await createTRPCInnerContext({
         apiClient,

packages/api/src/middlewares/withAuthed.ts

@@ -0,0 +1,15 @@
+import { TRPCError } from "@trpc/server";
+
+import { t } from "../trpc-client";
+import type { APIClientType } from "../utils";
+
+export const withAuthed = (expectedApiClientType: APIClientType) =>
+  t.middleware(({ ctx, next }) => {
+    if (ctx.apiClient?.type !== expectedApiClientType) {
+      throw new TRPCError({ code: "UNAUTHORIZED" });
+    }
+
+    return next({
+      ctx,
+    });
+  });

packages/api/src/procedures/authed.ts

@@ -0,0 +1,6 @@
+import { withAuthed } from "../middlewares/withAuthed";
+import type { APIClientType } from "../utils";
+import { publicProcedure } from "./public";
+
+export const createAuthedProcedure = (apiClientType: APIClientType) =>
+  publicProcedure.use(withAuthed(apiClientType));

packages/api/src/procedures/index.ts

@@ -1,2 +1,2 @@
-export * from "./jwt-authed";
+export * from "./authed";
 export * from "./public";

packages/api/src/routers/blob/index.ts

@@ -3,10 +3,12 @@ import { getAll } from "./getAll";
 import { getBlobDataByBlobId } from "./getBlobDataByBlobId";
 import { getByBlobId } from "./getByBlobId";
 import { getCount } from "./getCount";
+import { upsertWeaveVMReferences } from "./upsertWeaveVMReferences";
 
 export const blobRouter = t.router({
   getAll,
   getByBlobId,
   getBlobDataByBlobId,
   getCount,
+  upsertWeaveVMReferences,
 });

packages/api/src/routers/blob/upsertWeaveVMReferences.ts

@@ -0,0 +1,64 @@
+import { TRPCError } from "@trpc/server";
+
+import type { BlobDataStorageReference } from "@blobscan/db";
+import { BlobStorage } from "@blobscan/db/prisma/enums";
+import { z } from "@blobscan/zod";
+
+import { createAuthedProcedure } from "../../procedures";
+
+const inputSchema = z.object({
+  blobHashes: z.array(z.string()),
+});
+
+export const upsertWeaveVMReferences = createAuthedProcedure("weavevm")
+  .meta({
+    openapi: {
+      method: "PUT",
+      path: "/blobs/weavevm-references",
+      tags: ["blobs"],
+      summary: "upserts weaveVM blob references.",
+    },
+  })
+  .input(inputSchema)
+  .output(z.void())
+  .mutation(async ({ ctx: { prisma }, input: { blobHashes } }) => {
+    if (!blobHashes.length) {
+      return;
+    }
+
+    const blobDataStorageReferences = blobHashes.map<BlobDataStorageReference>(
+      (hash) => ({
+        blobHash: hash,
+        dataReference: hash,
+        blobStorage: BlobStorage.WEAVEVM,
+      })
+    );
+
+    const dbVersionedHashes = await prisma.blob
+      .findMany({
+        select: {
+          versionedHash: true,
+        },
+        where: {
+          versionedHash: {
+            in: blobHashes,
+          },
+        },
+      })
+      .then((blobs) => blobs.map((b) => b.versionedHash));
+
+    const missingHashes = blobHashes
+      .filter((hash) => !dbVersionedHashes.includes(hash))
+      .map((hash) => `"${hash}"`);
+
+    if (missingHashes.length > 0) {
+      throw new TRPCError({
+        code: "BAD_REQUEST",
+        message: `Couldn't found the following blobs: ${missingHashes.join(
+          ", "
+        )}`,
+      });
+    }
+
+    await prisma.blobDataStorageReference.upsertMany(blobDataStorageReferences);
+  });

packages/api/src/routers/blockchain-sync-state/updateState.ts

@@ -2,7 +2,7 @@ import { TRPCError } from "@trpc/server";
 
 import { z } from "@blobscan/zod";
 
-import { jwtAuthedProcedure } from "../../procedures";
+import { createAuthedProcedure } from "../../procedures";
 import { BASE_PATH } from "./common";
 
 export const inputSchema = z.object({
@@ -13,7 +13,7 @@ export const inputSchema = z.object({
 
 export const outputSchema = z.void();
 
-export const updateState = jwtAuthedProcedure
+export const updateState = createAuthedProcedure("indexer")
   .meta({
     openapi: {
       method: "PUT",

packages/api/src/routers/indexer/handleReorgedSlots.ts

@@ -2,7 +2,7 @@ import type { Prisma } from "@blobscan/db";
 import { z } from "@blobscan/zod";
 
 import type { TRPCInnerContext } from "../../context";
-import { jwtAuthedProcedure } from "../../procedures";
+import { createAuthedProcedure } from "../../procedures";
 import { INDEXER_PATH } from "./common";
 
 const inputSchema = z.object({
@@ -121,7 +121,7 @@ async function generateBlockCleanupOperations(
   return referenceRemovalOps;
 }
 
-export const handleReorgedSlots = jwtAuthedProcedure
+export const handleReorgedSlots = createAuthedProcedure("indexer")
   .meta({
     openapi: {
       method: "PUT",

packages/api/src/routers/indexer/indexData.ts

@@ -3,7 +3,7 @@ import { TRPCError } from "@trpc/server";
 import type { BlobDataStorageReference } from "@blobscan/db";
 import { toBigIntSchema, z } from "@blobscan/zod";
 
-import { jwtAuthedProcedure } from "../../procedures";
+import { createAuthedProcedure } from "../../procedures";
 import { INDEXER_PATH } from "./common";
 import {
   createDBAddresses,
@@ -58,7 +58,7 @@ export type IndexDataInput = z.input<typeof inputSchema>;
 
 export type IndexDataFormattedInput = z.output<typeof inputSchema>;
 
-export const indexData = jwtAuthedProcedure
+export const indexData = createAuthedProcedure("indexer")
   .meta({
     openapi: {
       method: "PUT",

packages/api/src/utils/auth.ts

@@ -0,0 +1,56 @@
+import { TRPCError } from "@trpc/server";
+import type { CreateNextContextOptions } from "@trpc/server/adapters/next";
+import type { NodeHTTPRequest } from "@trpc/server/adapters/node-http";
+import jwt from "jsonwebtoken";
+
+import { env } from "@blobscan/env";
+
+type NextHTTPRequest = CreateNextContextOptions["req"];
+
+type HTTPRequest = NodeHTTPRequest | NextHTTPRequest;
+
+export type APIClientType = "indexer" | "weavevm";
+
+export type APIClient = {
+  type: APIClientType;
+};
+
+function verifyIndexerClient(token: string) {
+  const decoded = jwt.verify(token, env.SECRET_KEY) as string;
+
+  return decoded;
+}
+
+function verifyWeaveVMClient(token: string) {
+  return token === env.WEAVEVM_API_KEY;
+}
+
+export function retrieveAPIClient(req: HTTPRequest): APIClient | undefined {
+  const authHeader = req.headers.authorization;
+
+  if (!authHeader) {
+    return;
+  }
+
+  const [type, token] = authHeader.split(" ");
+
+  if (type !== "Bearer" || !token) {
+    return;
+  }
+
+  try {
+    if (verifyWeaveVMClient(token)) {
+      return { type: "weavevm" };
+    }
+
+    if (verifyIndexerClient(token)) {
+      return { type: "indexer" };
+    }
+  } catch (err) {
+    if (err instanceof jwt.JsonWebTokenError) {
+      return;
+    }
+
+    throw new TRPCError({ code: "BAD_REQUEST" });
+  }
+}

packages/api/src/utils/index.ts

@@ -1,3 +1,4 @@
+export * from "./auth";
 export * from "./blob";
 export * from "./identifiers";
 export * from "./schemas";

packages/api/test/__snapshots__/blob.test.ts.snap

@@ -2623,3 +2623,7 @@ exports[`Blob router > getByBlobId > should get a blob by versioned hash 1`] = `
   "versionedHash": "blobHash004",
 }
 `;
+
+exports[`Blob router > upsertWeaveVMReferences > when authorized > should fail when one or more provided blobs do not exist 1`] = `"Couldn't found the following blobs: \\"nonExistingBlobHash\\""`;
+
+exports[`Blob router > upsertWeaveVMReferences > when authorized > should fail when one or more provided blobs do not exist 2`] = `undefined`;