fix: audit events schema

package.json

@@ -71,7 +71,7 @@
     "@segment/analytics-node": "2.1.0",
     "@sendgrid/mail": "8.0.0",
     "@sentry/node": "7.64.0",
-    "@sinclair/typebox": "0.31.28",
+    "@sinclair/typebox": "0.32.35",
     "@slack/web-api": "7.0.4",
     "@socket.io/redis-adapter": "8.2.1",
     "@supabase/supabase-js": "2.33.1",

packages/ee/shared/src/lib/audit-events/index.ts

@@ -32,6 +32,7 @@ const BaseAuditEventProps = {
     ...BaseModelSchema,
     platformId: Type.String(),
     projectId: Type.Optional(Type.String()),
+    projectDisplayName: Type.Optional(Type.String()),
     userId: Type.Optional(Type.String()),
     userEmail: Type.Optional(Type.String()),
     ip: Type.Optional(Type.String()),

packages/server/api/src/app/database/migration/postgres/1723489038729-MigrateAuditEventSchema.ts

@@ -0,0 +1,42 @@
+import { MigrationInterface, QueryRunner } from 'typeorm'
+
+export class MigrateAuditEventSchema1723489038729 implements MigrationInterface {
+    name = 'MigrateAuditEventSchema1723489038729'
+
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`
+            DROP INDEX "public"."audit_event_platform_id_project_id_user_id_idx"
+        `)
+        await queryRunner.query(`
+            ALTER TABLE "audit_event"
+            ALTER COLUMN "userEmail" DROP NOT NULL
+        `)
+        await queryRunner.query(`
+            ALTER TABLE "audit_event"
+            ALTER COLUMN "userId" DROP NOT NULL
+        `)
+        await queryRunner.query(`
+            CREATE INDEX "audit_event_platform_id_project_id_user_id_idx" ON "audit_event" ("platformId", "projectId", "userId")
+        `)
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`
+            DROP INDEX "public"."audit_event_platform_id_project_id_user_id_idx"
+        `)
+        await queryRunner.query(`
+            ALTER TABLE "audit_event"
+            ALTER COLUMN "userId"
+            SET NOT NULL
+        `)
+        await queryRunner.query(`
+            ALTER TABLE "audit_event"
+            ALTER COLUMN "userEmail"
+            SET NOT NULL
+        `)
+        await queryRunner.query(`
+            CREATE INDEX "audit_event_platform_id_project_id_user_id_idx" ON "audit_event" ("platformId", "projectId", "userId")
+        `)
+    }
+
+}

packages/server/api/src/app/database/postgres-connection.ts

@@ -139,6 +139,7 @@ import { AddPremiumPiecesColumnPostgres1717370717678 } from './migration/postgre
 import { AddUserInvitation1717960689650 } from './migration/postgres/1717960689650-AddUserInvitation'
 import { ModifyProjectMembers1717961669938 } from './migration/postgres/1717961669938-ModifyProjectMembers'
 import { AddWorkerMachine1720101280025 } from './migration/postgres/1720101280025-AddWorkerMachine'
+import { MigrateAuditEventSchema1723489038729 } from './migration/postgres/1723489038729-MigrateAuditEventSchema'
 
 const getSslConfig = (): boolean | TlsOptions => {
     const useSsl = system.get(AppSystemProp.POSTGRES_USE_SSL)
@@ -301,6 +302,7 @@ const getMigrations = (): (new () => MigrationInterface)[] => {
 
                 // New Migration After Unifying
                 ModifyProjectMembers1717961669938,
+                MigrateAuditEventSchema1723489038729,
             )
             break
         case ApEdition.COMMUNITY:

packages/server/api/src/app/ee/audit-logs/audit-event-entity.ts

@@ -23,6 +23,11 @@ export const AuditEventEntity = new EntitySchema<AuditEventSchema>({
         },
         userEmail: {
             type: String,
+            nullable: true,
+        },
+        projectDisplayName: {
+            type: String,
+            nullable: true,
         },
         data: {
             type: JSONB_COLUMN_TYPE,

packages/server/api/src/app/ee/audit-logs/audit-event-service.ts

@@ -4,12 +4,11 @@ import {
 import { logger, networkUtls, rejectedPromiseHandler } from '@activepieces/server-shared'
 import {
     apId,
-    assertEqual,
     Cursor,
     PrincipalType,
     SeekPage,
 } from '@activepieces/shared'
-import Ajv from 'ajv'
+import { Value } from '@sinclair/typebox/value'
 import { FastifyRequest } from 'fastify'
 import { repoFactory } from '../../core/db/repo-factory'
 import { AuditEventParam } from '../../helper/application-events'
@@ -19,11 +18,8 @@ import { platformService } from '../../platform/platform.service'
 import { projectService } from '../../project/project-service'
 import { userService } from '../../user/user-service'
 import { AuditEventEntity } from './audit-event-entity'
-
 const auditLogRepo = repoFactory(AuditEventEntity)
 
-const ajv = new Ajv({ removeAdditional: 'all' })
-const eventSchema = ajv.compile<ApplicationEvent>(ApplicationEvent)
 
 export const auditLogService = {
     sendUserEvent(request: FastifyRequest, params: AuditEventParam): void {
@@ -84,7 +80,9 @@ async function saveEvent(info: MetaInformation, rawEvent: AuditEventParam): Prom
         created: new Date().toISOString(),
         updated: new Date().toISOString(),
         userId: info.userId,
+        userEmail: user?.email,
         projectId: info.projectId,
+        projectDisplayName: project?.displayName,
         platformId: info.platformId,
         ip: info.ip,
         data: {
@@ -94,12 +92,15 @@ async function saveEvent(info: MetaInformation, rawEvent: AuditEventParam): Prom
         },
         action: rawEvent.action,
     }
-    const valid = eventSchema(eventToSave)
-    assertEqual(valid, true, 'Event validation', 'true')
-    const appEvent = await auditLogRepo().save(eventToSave as ApplicationEvent)
+
+    // The event may contain Date objects, so we serialize it to convert dates back to strings as per the schema.
+    const clonedAndSerializedDates = JSON.parse(JSON.stringify(eventToSave))
+    const cleanedEvent = Value.Clean(ApplicationEvent, clonedAndSerializedDates) as ApplicationEvent
+
+    const savedEvent = await auditLogRepo().save(cleanedEvent)
     logger.info({
         message: '[AuditEventService#saveEvent] Audit event saved',
-        appEvent,
+        appEvent: savedEvent,
     })
 }
 

packages/server/api/src/app/pieces/piece-metadata-service/db-piece-metadata-service.ts

@@ -1,6 +1,6 @@
 
 import { PieceMetadataModel, PieceMetadataModelSummary } from '@activepieces/pieces-framework'
-import { ActivepiecesError, apId, assertNotNullOrUndefined, ErrorCode, EXACT_VERSION_PATTERN, isNil, ListVersionsResponse, PieceType } from '@activepieces/shared'
+import { ActivepiecesError, apId, assertNotNullOrUndefined, ErrorCode, EXACT_VERSION_REGEX, isNil, ListVersionsResponse, PieceType } from '@activepieces/shared'
 import dayjs from 'dayjs'
 import semVer from 'semver'
 import { IsNull } from 'typeorm'
@@ -112,7 +112,7 @@ export const FastDbPieceMetadataService = (): PieceMetadataService => {
             })
         },
         async getExactPieceVersion({ name, version, projectId }): Promise<string> {
-            const isExactVersion = EXACT_VERSION_PATTERN.test(version)
+            const isExactVersion = EXACT_VERSION_REGEX.test(version)
 
             if (isExactVersion) {
                 return version

packages/server/api/src/app/pieces/piece-metadata-service/file-piece-metadata-service.ts

@@ -7,7 +7,7 @@ import {
     ActivepiecesError,
     ApEdition,
     ErrorCode,
-    EXACT_VERSION_PATTERN,
+    EXACT_VERSION_REGEX,
     extractPieceFromModule,
     isNil,
     ListVersionsResponse,
@@ -170,7 +170,7 @@ export const FilePieceMetadataService = (): PieceMetadataService => {
         },
 
         async getExactPieceVersion({ projectId, name, version }): Promise<string> {
-            const isExactVersion = EXACT_VERSION_PATTERN.test(version)
+            const isExactVersion = EXACT_VERSION_REGEX.test(version)
 
             if (isExactVersion) {
                 return version

packages/server/api/src/app/workers/redis/redis-consumer.ts

@@ -60,12 +60,7 @@ export const redisConsumer: ConsumerManager = {
         await Promise.all(sharedConsumers)
     },
     async close(): Promise<void> {
-        const promises = Object.values(consumer).map((consumerGroup) => {
-            return Promise.all(Object.values(consumerGroup).map(async (consumer) => {
-                await consumer.drain()
-                await consumer.close()
-            }))
-        })
+        const promises = Object.values(consumer).map(consumer => consumer.close())
         await Promise.all(promises)
     },
 }

packages/server/shared/package.json

@@ -9,7 +9,7 @@
     "@activepieces/shared": "*",
     "fastify": "4.12.0",
     "async-mutex": "0.4.0",
-    "@sinclair/typebox": "0.31.28"
+    "@sinclair/typebox": "0.32.35"
   },
   "type": "commonjs",
   "main": "./src/index.js",

packages/server/worker/src/lib/engine/flow-enginer-util.ts

@@ -1,4 +1,4 @@
-import { Action, ActionType, assertEqual, CodeAction, EXACT_VERSION_PATTERN, flowHelper, FlowVersion, PackageType, PieceActionSettings, PiecePackage, PieceTriggerSettings, PieceType, Trigger, TriggerType } from '@activepieces/shared'
+import { Action, ActionType, assertEqual, CodeAction, EXACT_VERSION_REGEX, flowHelper, FlowVersion, PackageType, PieceActionSettings, PiecePackage, PieceTriggerSettings, PieceType, Trigger, TriggerType } from '@activepieces/shared'
 import { engineApiService } from '../api/server-api.service'
 import { CodeArtifact } from './engine-runner'
 
@@ -62,7 +62,7 @@ export const pieceEngineUtil = {
                 }
             }
             case PackageType.REGISTRY: {
-                const exactVersion = EXACT_VERSION_PATTERN.test(pieceVersion)
+                const exactVersion = EXACT_VERSION_REGEX.test(pieceVersion)
                 const version = exactVersion ? pieceVersion : (await engineApiService(engineToken).getPiece(pieceName, {
                     version: pieceVersion,
                 })).version

packages/shared/src/lib/flow-run/flow-run.ts

@@ -34,7 +34,7 @@ export const FlowRun = Type.Object({
     status: Type.Enum(FlowRunStatus),
     duration: Type.Optional(Type.Number()),
     startTime: Type.String(),
-    finishTime: Type.String(),
+    finishTime: Type.Optional(Type.String()),
     environment: Type.Enum(RunEnvironment),
     pauseMetadata: Type.Optional(PauseMetadata),
     steps: Type.Record(Type.String(), Type.Unknown()),

packages/shared/src/lib/pieces/dto/piece-requests.ts

@@ -2,12 +2,16 @@ import { Static, Type } from '@sinclair/typebox'
 import { ApEdition } from '../../flag/flag'
 import { PackageType, PieceCategory, PieceType } from '../piece'
 
-export const EXACT_VERSION_PATTERN = /^[0-9]+\.[0-9]+\.[0-9]+$/
-export const VERSION_PATTERN = /^([~^])?[0-9]+\.[0-9]+\.[0-9]+$/
-
-export const ExactVersionType = Type.RegExp(EXACT_VERSION_PATTERN)
-export const VersionType = Type.RegExp(VERSION_PATTERN)
+export const EXACT_VERSION_PATTERN = '^[0-9]+\\.[0-9]+\\.[0-9]+$'
+export const EXACT_VERSION_REGEX = new RegExp(EXACT_VERSION_PATTERN)
+const VERSION_PATTERN = '^([~^])?[0-9]+\\.[0-9]+\\.[0-9]+$'
 
+export const ExactVersionType = Type.String({
+    pattern: EXACT_VERSION_PATTERN,
+})
+export const VersionType = Type.String({
+    pattern: VERSION_PATTERN,
+})
 export enum SuggestionType {
     ACTION = 'ACTION',
     TRIGGER = 'TRIGGER',