feat: add application batch endpoint

moonrailgun · moonrailgun · commit d1ff4e6f7b78 · 2025-06-07T00:35:36.000+08:00
diff --git a/src/server/model/application/index.ts b/src/server/model/application/index.ts
@@ -19,15 +19,20 @@ export interface ApplicationEventPayload {
   url?: string;
   application: string;
   name?: string;
+  version?: string;
+  sdkVersion?: string;
 }
 
-export async function findSession(req: Request): Promise<
+export async function findSession(
+  req: Request,
+  body: any
+): Promise<
   ApplicationSession & {
     workspaceId: string;
   }
 > {
   // Verify payload
-  const { payload } = req.body;
+  const { payload } = body;
 
   // Check if cache token is passed
   const cacheToken = req.headers['x-tianji-cache'] as string;
@@ -44,6 +49,8 @@ export async function findSession(req: Request): Promise<
     application: applicationId,
     os,
     language,
+    version,
+    sdkVersion,
   } = payload as ApplicationEventPayload;
 
   if (!isCuid(applicationId)) {
@@ -84,6 +91,8 @@ export async function findSession(req: Request): Promise<
           os,
           language,
           ip,
+          version,
+          sdkVersion,
           country,
           subdivision1,
           subdivision2,
diff --git a/src/server/router/application.ts b/src/server/router/application.ts
@@ -12,6 +12,54 @@ import { isWorkspacePaused } from '../model/billing/workspace.js';
 
 export const applicationRouter = Router();
 
+// Shared validation schema for event payload
+const eventPayloadSchema = yup.object().shape({
+  data: yup.object(),
+  language: yup.string().max(35),
+  os: yup.string().max(20),
+  application: yup.string().required(),
+  name: yup.string().max(50),
+  screen: yup.string().max(500),
+  params: yup.object(),
+});
+
+// Shared function to process a single event
+async function processEvent(type: string, payload: any, session: any) {
+  const {
+    name: eventName,
+    data: eventData,
+    screen: screenName,
+    params: screenParams,
+  } = payload;
+
+  if (type === COLLECTION_TYPE.event) {
+    await saveApplicationEvent({
+      eventName,
+      eventData,
+      screenName,
+      screenParams,
+      ...session,
+      sessionId: session.id,
+    });
+    return { status: 'success', type: 'event' };
+  }
+
+  if (type === COLLECTION_TYPE.identify) {
+    if (!eventData) {
+      throw new Error('Data required for identify event');
+    }
+
+    await saveApplicationSessionData({
+      ...session,
+      sessionData: eventData,
+      sessionId: session.id,
+    });
+    return { status: 'success', type: 'identify' };
+  }
+
+  throw new Error(`Unsupported event type: ${type}`);
+}
+
 applicationRouter.post(
   '/send',
   validate(
@@ -20,20 +68,7 @@ applicationRouter.post(
       .withMessage('payload should be existed')
       .isObject()
       .custom(async (input) => {
-        return yup
-          .object()
-          .shape({
-            data: yup.object(),
-            language: yup.string().max(35),
-            os: yup.string().max(20),
-            url: yup.string().max(500),
-            application: yup.string().required(),
-            name: yup.string().max(50),
-            screen: yup.string().max(500),
-            params: yup.object(),
-          })
-          .required()
-          .validate(input);
+        return eventPayloadSchema.required().validate(input);
       }),
     body('type')
       .exists()
@@ -43,46 +78,86 @@ applicationRouter.post(
   ),
   async (req, res) => {
     const { type, payload } = req.body;
-    const {
-      url,
-      name: eventName,
-      data: eventData,
-      screen: screenName,
-      params: screenParams,
-    } = payload;
 
-    const session = await findSession(req);
+    const session = await findSession(req, req.body);
 
     if (await isWorkspacePaused(session.workspaceId)) {
       res.status(403).send('Workspace is paused.');
       return;
     }
 
-    if (type === COLLECTION_TYPE.event) {
-      await saveApplicationEvent({
-        eventName,
-        eventData,
-        screenName,
-        screenParams,
-        ...session,
-        sessionId: session.id,
+    try {
+      await processEvent(type, payload, session);
+      const token = createToken(session);
+      res.send(token);
+    } catch (error) {
+      res.status(400).json({
+        error: error instanceof Error ? error.message : 'Unknown error',
       });
     }
+  }
+);
 
-    if (type === COLLECTION_TYPE.identify) {
-      if (!eventData) {
-        throw new Error('Data required');
-      }
+applicationRouter.post(
+  '/batch',
+  validate(
+    body('events')
+      .exists()
+      .withMessage('events should be existed')
+      .isArray({ min: 1, max: 100 })
+      .withMessage('events should be an array with 1-100 items')
+      .custom(async (events) => {
+        // Validate each event in the array
+        const eventSchema = yup.object().shape({
+          type: yup
+            .string()
+            .required()
+            .matches(/event|identify/i),
+          payload: eventPayloadSchema.required(),
+        });
 
-      await saveApplicationSessionData({
-        ...session,
-        sessionData: eventData,
-        sessionId: session.id,
-      });
+        for (const event of events) {
+          await eventSchema.validate(event);
+        }
+        return true;
+      })
+  ),
+  async (req, res) => {
+    const { events } = req.body;
+
+    const session = await findSession(req, events[0]);
+
+    if (await isWorkspacePaused(session.workspaceId)) {
+      res.status(403).send('Workspace is paused.');
+      return;
+    }
+
+    const results = [];
+    const errors = [];
+
+    // Process each event in the batch
+    for (let i = 0; i < events.length; i++) {
+      const event = events[i];
+      try {
+        const result = await processEvent(event.type, event.payload, session);
+        results.push({ index: i, ...result });
+      } catch (error) {
+        errors.push({
+          index: i,
+          status: 'error',
+          message: error instanceof Error ? error.message : 'Unknown error',
+        });
+      }
     }
 
     const token = createToken(session);
 
-    res.send(token);
+    res.json({
+      token,
+      processed: results.length,
+      errors: errors.length,
+      results,
+      errorDetails: errors.length > 0 ? errors : undefined,
+    });
   }
 );
