add zod validation for internal Request and Response message

mod.ts

@@ -1,2 +1,4 @@
 export * from "./src/service.ts";
 export * from "./src/adaptors/nats.ts";
+export * from "./src/error.ts";
+export type { Request, Response } from "./src/messages.ts";

src/messages.ts

@@ -1,13 +1,23 @@
+import { z } from "zod";
+
 export interface Request<T> {
   service: string;
   subject: string;
   data?: T;
 }
 
-export interface Message<T> {
+export const InternalRequestSchema = z.object({
-  from: string;
+  from: z.string(),
-  data?: T;
+  data: z.optional(z.any()),
-}
+});
+
+export const InternalResponseSchema = z.object({
+  data: z.optional(z.any()),
+  statusCode: z.number(),
+});
+
+export type InternalRequest = z.infer<typeof InternalRequestSchema>;
+export type InternalResponse = z.infer<typeof InternalResponseSchema>;
 
 export interface Response<T> {
   data?: T;

src/service.ts

@@ -1,7 +1,13 @@
-import { ServiceError } from "nats";
 import Adaptor from "./adaptors/adaptor.ts";
 import { RequestError } from "./error.ts";
-import { Message, Request, Response } from "./messages.ts";
+import {
+  InternalRequest,
+  InternalRequestSchema,
+  InternalResponse,
+  InternalResponseSchema,
+  Request,
+  Response,
+} from "./messages.ts";
 import { RouteSubscribeTypeFn } from "./types.ts";
 import { z } from "zod";
 
@@ -28,40 +34,62 @@ export default class Service {
     this.adaptors[adaptor].subscribe(
       `${this.name}.${subject}`,
       async (rawReq) => {
-        const msg: Request<T> = JSON.parse(rawReq);
+        const rawReqJson = JSON.parse(rawReq);
+        const internalRequestJson = InternalRequestSchema.safeParse(rawReqJson);
+        if (!internalRequestJson.success) {
+          return JSON.stringify(
+            {
+              statusCode: 400,
+              data: "bad request structure",
+            } satisfies InternalResponse,
+          );
+        }
 
-        const message: Message<z.infer<T>> = { from: this.name };
+        const internalRequest = internalRequestJson.data;
+        const req = {
+          service: internalRequest.from,
+          subject: subject,
+        } as Request<z.infer<T>>;
 
-        if (msg.data && schema) {
+        if (internalRequest.data && schema) {
-          const validate = schema.safeParse(msg.data);
+          const validate = schema.safeParse(internalRequest.data);
           if (!validate.success) {
             return JSON.stringify(
               {
                 statusCode: 400,
-                data: validate.error.toString(),
+                data: validate.error,
-              } satisfies Response<string>,
+              } satisfies InternalResponse,
             );
           } else {
-            message.data = validate.data;
+            req.data = validate.data;
           }
         }
 
-        let res;
         try {
-          res = await fn(message);
+          const res = await fn(req);
+          const internalResponse = {
+            statusCode: res.statusCode,
+            data: res.data,
+          } satisfies InternalResponse;
+
+          return JSON.stringify(internalResponse);
         } catch (err) {
-          if (err instanceof ServiceError) {
+          if (err instanceof RequestError) {
-            res = { data: err.message, statusCode: err.code } as Response<
+            return JSON.stringify(
-              string
+              {
-            >;
+                statusCode: err.statusCode,
+                data: err.message,
+              } satisfies InternalResponse,
+            );
           } else {
-            res = { data: "unknow error append", statusCode: 500 } as Response<
+            return JSON.stringify(
-              string
+              {
-            >;
+                statusCode: 500,
+                data: err?.message || "unknow error apend",
+              } satisfies InternalResponse,
+            );
           }
         }
-
-        return JSON.stringify(res);
       },
     );
   }
@@ -75,24 +103,40 @@ export default class Service {
       throw new Error(`${adaptor} adaptor not exist`);
     }
 
-    const rawReq = JSON.stringify(req);
+    const internalRequest = {
+      from: this.name,
+      data: req.data,
+    } satisfies InternalRequest;
+    const internalRequestJson = JSON.stringify(internalRequest);
 
     try {
       const rawRes = await this.adaptors[adaptor].request(
         `${req.service}.${req.subject}`,
-        rawReq,
+        internalRequestJson,
       );
-      const res: Response<z.infer<U>> = JSON.parse(rawRes);
+      const rawResJson: unknown = JSON.parse(rawRes);
-
+      const internalResponseJson = InternalResponseSchema.safeParse(rawResJson);
-      if (res.statusCode < 200 || res.statusCode >= 299) {
+      if (!internalResponseJson.success) {
-        throw new RequestError("error while request", res.statusCode);
+        throw new RequestError(internalResponseJson.error.toString(), 500);
       }
 
-      if (res.data && schema) {
+      const internalResponse = internalResponseJson.data;
-        const validate = schema.safeParse(res.data);
+      if (
+        internalResponse.statusCode < 200 || internalResponse.statusCode >= 299
+      ) {
+        throw new RequestError(
+          internalResponse.data,
+          internalResponse.statusCode,
+        );
+      }
+      const res: Response<z.infer<U>> = {
+        statusCode: internalResponse.statusCode,
+      };
+      if (internalResponse.data && schema) {
+        const validate = schema.safeParse(internalResponse.data);
 
         if (!validate.success) {
-          throw new ServiceError(400, validate.error.toString());
+          throw new RequestError(validate.error.message, 400);
         } else {
           res.data = validate.data;
         }

src/types.ts

@@ -1,6 +1,6 @@
-import { Message, Response } from "./messages.ts";
+import { Request, Response } from "./messages.ts";
 
 export type AdaptorSubscribeTypeFn = (msg: string) => Promise<string>;
 export type RouteSubscribeTypeFn<T, U> = (
-  msg: Message<T>,
+  msg: Request<T>,
 ) => Promise<Response<U>>;

tests/service.test.ts

@@ -1,9 +1,9 @@
 import { z } from "zod";
 import NatsAdaptor from "../src/adaptors/nats.ts";
-import { Message } from "../src/messages.ts";
 import Service from "../src/service.ts";
 import { assertEquals, assertRejects, assertThrows } from "std/assert/mod.ts";
 import { afterEach, beforeEach, it } from "std/testing/bdd.ts";
+import { RequestError } from "../src/error.ts";
 
 let srv!: Service;
 
@@ -55,7 +55,7 @@ it("request error", {
 
   srv.addAdaptor(adaptorName, new NatsAdaptor({ servers: [natsServer] }));
   srv.subscribe(adaptorName, subject, async (msg) => {
-    return { data: msg.data, statusCode: statusCodeExpected };
+    throw new RequestError("request error", 500);
   }, z.string());
 
   await srv.listen();