Merge branch 'main' into chore/update-ghjk

Author: Yohe-Am

docs/metatype.dev/docs/reference/typegate/requests/index.mdx

@@ -0,0 +1,29 @@
+# Requests
+
+## Idempotency
+
+An idempotent request is a request that always returns the same result when retried with the same input.
+This is useful to prevent duplicate operations, such as charging a customer twice if a network timeout or retry occurs.
+
+You can enforce idempotency on any request to a typegraph, you just have to set the `Idempotency-Key` header with a unique value (such as a UUID).
+
+```curl
+curl https://localhost:7890/awesome_typegraph \
+  -X POST \
+  -H "Authorization: Bearer ...." \
+  -H "Content-Type: application/json" \
+  -H "Idempotency-Key: 123abc" \
+  -d '{
+    "query": "mutation CreatePost($title: String!) { createPost(title: $title) { id title } }",
+    "variables": { "title": "My First Post" }
+  }'
+```
+
+:::info Notes
+
+- Keys expire happens after 24 hours.
+- A key must match the original request exactly during its lifetime.
+  - Reusing the same key for different requests will result in error status 422.
+- An empty string explicitly disables idempotency, it is a no-op.
+- The maximum key length is 255 characters.
+  :::

src/typegate/src/services/graphql_service.ts

@@ -13,14 +13,22 @@ import {
   findOperation,
   type FragmentDefs,
 } from "../transports/graphql/graphql.ts";
-import { forceAnyToOption } from "../utils.ts";
+import {
+  type CachedResponse,
+  computeRequestSignature,
+  forceAnyToOption,
+  toResponse,
+  toSerializableResponse,
+} from "../utils.ts";
 import type { QueryEngine } from "../engine/query_engine.ts";
 import type * as ast from "graphql/ast";
 import { BadContext, ResolverError } from "../errors.ts";
 import { badRequest, jsonError, jsonOk } from "./responses.ts";
 import { BaseError, ErrorKind } from "../errors.ts";
+import type { Register } from "@metatype/typegate/typegate/register.ts";
 
 const logger = getLogger(import.meta);
+const IDEMPOTENCY_HEADER = "Idempotency-Key";
 
 class InvalidQuery extends BaseError {
   constructor(message: string) {
@@ -41,23 +49,14 @@ export function isIntrospectionQuery(
   return operation.name?.value === "IntrospectionQuery";
 }
 
-export async function handleGraphQL(
-  request: Request,
+export async function handleGraphQLHelper(
+  content: Operations,
   engine: QueryEngine,
   context: Context,
   info: Info,
   limit: RateLimit | null,
   headers: Headers,
 ): Promise<Response> {
-  let content: Operations | null = null;
-  try {
-    content = await parseRequest(request);
-  } catch (err: any) {
-    if (err instanceof BaseError) {
-      return err.toResponse(headers);
-    }
-    return badRequest(err.message);
-  }
   const { query, operationName: operationNameRaw, variables } = content;
   const operationName = forceAnyToOption(operationNameRaw);
 
@@ -145,3 +144,93 @@ export async function handleGraphQL(
     }
   }
 }
+
+export async function handleGraphQL(
+  register: Register,
+  request: Request,
+  engine: QueryEngine,
+  context: Context,
+  info: Info,
+  limit: RateLimit | null,
+  headers: Headers,
+): Promise<Response> {
+  const key = request.headers.get(IDEMPOTENCY_HEADER);
+  let content: Operations | null = null;
+  try {
+    content = await parseRequest(key ? request.clone() : request);
+  } catch (err: any) {
+    if (err instanceof BaseError) {
+      return err.toResponse(headers);
+    }
+    return badRequest(err.message);
+  }
+
+  if (key) {
+    if (key.length > 255) {
+      return jsonError({
+        status: 422,
+        message:
+          `'${IDEMPOTENCY_HEADER}' value should not exceed 255 characters`,
+        headers,
+      });
+    }
+
+    const userRequestHash = await computeRequestSignature(request, [
+      IDEMPOTENCY_HEADER,
+    ]);
+    const now = Date.now();
+    const memoized = await register.getResponse(key);
+
+    if (memoized) {
+      const { response, expiryMillis, requestHash: savedHash } = memoized;
+
+      if (now < expiryMillis) {
+        if (userRequestHash != savedHash) {
+          return jsonError({
+            status: 422,
+            message:
+              `The request associated with key "${key}" has changed. Please use a new key or ensure the request matches the original.`,
+            headers,
+          });
+        }
+
+        logger.debug(`Idempotent request key "${key}" replayed`);
+        return toResponse(response);
+      } else {
+        await register.deleteResponse(key);
+      }
+    }
+
+    const response = await handleGraphQLHelper(
+      content,
+      engine,
+      context,
+      info,
+      limit,
+      headers,
+    );
+
+    const oneDay = 24 * 3600 * 1000;
+    const expiryMillis = now + oneDay;
+    await register.addResponse(
+      key,
+      {
+        response: await toSerializableResponse(response.clone()),
+        expiryMillis,
+        requestHash: userRequestHash,
+      } satisfies CachedResponse,
+    );
+
+    logger.warn(`Idempotent request key "${key}" renewed`);
+    return response;
+  }
+
+  return await handleGraphQLHelper(
+    content,
+    engine,
+    context,
+    info,
+    limit,
+    headers,
+  );
+}

src/typegate/src/typegate/memory_register.ts

@@ -3,9 +3,11 @@
 
 import type { QueryEngine } from "../engine/query_engine.ts";
 import { Register } from "../typegate/register.ts";
+import type { CachedResponse } from "@metatype/typegate/utils.ts";
 
 export class MemoryRegister extends Register {
   private map = new Map<string, QueryEngine>();
+  private responseMap = new Map<string, CachedResponse>();
 
   constructor() {
     super();
@@ -18,6 +20,7 @@ export class MemoryRegister extends Register {
       ),
     );
     this.map.clear();
+    this.responseMap.clear();
   }
 
   async add(engine: QueryEngine): Promise<void> {
@@ -47,4 +50,18 @@ export class MemoryRegister extends Register {
   has(name: string): boolean {
     return this.map.has(name);
   }
+
+  addResponse(key: string, response: CachedResponse): Promise<void> {
+    this.responseMap.set(key, response);
+    return Promise.resolve();
+  }
+
+  deleteResponse(key: string): Promise<void> {
+    this.responseMap.delete(key);
+    return Promise.resolve();
+  }
+
+  getResponse(key: string): CachedResponse | undefined {
+    return this.responseMap.get(key);
+  }
 }

src/typegate/src/typegate/mod.ts

@@ -174,6 +174,16 @@ export class Typegate implements AsyncDisposable {
       });
       register.startSync(lastSync);
 
+      const lastSyncResponses = await register.historySyncResponses().catch(
+        (err) => {
+          logger.error(err);
+          throw new Error(
+            `failed to load response history at boot, aborting: ${err.message}`,
+          );
+        },
+      );
+      register.startSyncResponses(lastSyncResponses);
+
       return typegate;
     }
   }
@@ -326,7 +336,15 @@ export class Typegate implements AsyncDisposable {
         return methodNotAllowed();
       }
 
-      return handleGraphQL(request, engine, context, info, limit, headers);
+      return handleGraphQL(
+        this.register,
+        request,
+        engine,
+        context,
+        info,
+        limit,
+        headers,
+      );
     } catch (e) {
       Sentry.captureException(e);
       console.error(e);

src/typegate/src/typegate/register.ts

@@ -11,6 +11,10 @@ import {
 } from "../typegraph/versions.ts";
 import { typegraphIdSchema, type TypegraphStore } from "../sync/typegraph.ts";
 import { RedisReplicatedMap } from "../sync/replicated_map.ts";
+import {
+  type CachedResponse,
+  CachedResponseSchema,
+} from "@metatype/typegate/utils.ts";
 
 export interface MessageEntry {
   type: "info" | "warning" | "error";
@@ -33,6 +37,12 @@ export abstract class Register implements AsyncDisposable {
 
   abstract has(name: string): boolean;
 
+  abstract addResponse(key: string, response: CachedResponse): Promise<void>;
+
+  abstract deleteResponse(key: string): Promise<void>;
+
+  abstract getResponse(key: string): CachedResponse | undefined;
+
   abstract [Symbol.asyncDispose](): Promise<void>;
 }
 
@@ -83,15 +93,35 @@ export class ReplicatedRegister extends Register {
       },
     );
 
-    return new ReplicatedRegister(replicatedMap);
+    const replicatedResponseMap = await RedisReplicatedMap.init<CachedResponse>(
+      "typegraph_responses",
+      redisConfig,
+      {
+        serialize(data: CachedResponse) {
+          return JSON.stringify(data);
+        },
+        deserialize(json: string, _: boolean) {
+          const raw = JSON.parse(json);
+          const cachedResponse = CachedResponseSchema.parse(raw);
+          return cachedResponse;
+        },
+        async terminate(_: CachedResponse) {},
+      },
+    );
+
+    return new ReplicatedRegister(replicatedMap, replicatedResponseMap);
   }
 
-  constructor(public replicatedMap: RedisReplicatedMap<QueryEngine>) {
+  constructor(
+    public replicatedMap: RedisReplicatedMap<QueryEngine>,
+    public replicatedResponseMap: RedisReplicatedMap<CachedResponse>,
+  ) {
     super();
   }
 
   async [Symbol.asyncDispose](): Promise<void> {
     await this.replicatedMap[Symbol.asyncDispose]();
+    await this.replicatedResponseMap[Symbol.asyncDispose]();
     await Promise.all(this.list().map((e) => e[Symbol.asyncDispose]()));
   }
 
@@ -133,7 +163,27 @@ export class ReplicatedRegister extends Register {
     return this.replicatedMap.historySync();
   }
 
+  historySyncResponses(): Promise<XIdInput> {
+    return this.replicatedResponseMap.historySync();
+  }
+
   startSync(xid: XIdInput): void {
     void this.replicatedMap.startSync(xid);
   }
+
+  startSyncResponses(xid: XIdInput): void {
+    void this.replicatedResponseMap.startSync(xid);
+  }
+
+  addResponse(key: string, response: CachedResponse): Promise<void> {
+    return this.replicatedResponseMap.set(key, response);
+  }
+
+  deleteResponse(key: string): Promise<void> {
+    return this.replicatedResponseMap.delete(key);
+  }
+
+  getResponse(key: string): CachedResponse | undefined {
+    return this.replicatedResponseMap.get(key);
+  }
 }