Feature/Add Audit and Deletion

packages/components/src/storage/AzureBlobStorageProvider.ts

@@ -2,9 +2,9 @@ import { BlobServiceClient, ContainerClient, StorageSharedKeyCredential } from '
 import multer from 'multer'
 import { MulterAzureStorage } from 'multer-azure-blob-storage'
 import { v4 as uuidv4 } from 'uuid'
+import { winstonAzureBlob } from 'winston-azure-blob'
 import { BaseStorageProvider } from './BaseStorageProvider'
 import { FileInfo, StorageResult, StorageSizeResult } from './IStorageProvider'
-import { winstonAzureBlob } from 'winston-azure-blob'
 
 /**
  * Extends MulterAzureStorage to set file.path from file.blobName after upload.
@@ -281,7 +281,7 @@ export class AzureBlobStorageProvider extends BaseStorageProvider {
         return multer({ storage: azureStorage })
     }
 
-    getLoggerTransports(logType: 'server' | 'error' | 'requests'): any[] {
+    getLoggerTransports(logType: 'server' | 'error' | 'requests' | 'audit'): any[] {
         const connectionString = process.env.AZURE_BLOB_STORAGE_CONNECTION_STRING
         const accountName = process.env.AZURE_BLOB_STORAGE_ACCOUNT_NAME
         const accountKey = process.env.AZURE_BLOB_STORAGE_ACCOUNT_KEY
@@ -325,7 +325,18 @@ export class AzureBlobStorageProvider extends BaseStorageProvider {
                     level: 'debug'
                 })
             ]
+        } else if (logType === 'audit') {
+            return [
+                winstonAzureBlob({
+                    ...baseConfig,
+                    blobName: 'logs/audit/audit',
+                    rotatePeriod: 'YYYY-MM-DD',
+                    extension: '.log.jsonl',
+                    level: 'info'
+                })
+            ]
         }
+
         return []
     }
 }

packages/components/src/storage/BaseStorageProvider.ts

@@ -1,9 +1,9 @@
+import fs from 'node:fs'
 import path from 'node:path'
-import { IStorageProvider, FileInfo, StorageResult, StorageSizeResult } from './IStorageProvider'
 import sanitize from 'sanitize-filename'
 import { getUserHome } from '../utils'
 import { isPathTraversal, isUnsafeFilePath, isValidUUID } from '../validator'
-import fs from 'node:fs'
+import { FileInfo, IStorageProvider, StorageResult, StorageSizeResult } from './IStorageProvider'
 
 export abstract class BaseStorageProvider implements IStorageProvider {
     protected storagePath: string
@@ -38,7 +38,7 @@ export abstract class BaseStorageProvider implements IStorageProvider {
     abstract removeFolderFromStorage(...paths: string[]): Promise<StorageSizeResult>
     abstract getStorageSize(orgId: string): Promise<number>
     abstract getMulterStorage(): any
-    abstract getLoggerTransports(logType: 'server' | 'error' | 'requests', config?: any): any[]
+    abstract getLoggerTransports(logType: 'server' | 'error' | 'requests' | 'audit', config?: any): any[]
 
     /**
      * Shared utility for sanitizing filenames to prevent path traversal and other issues

packages/components/src/storage/GCSStorageProvider.ts

@@ -1,5 +1,5 @@
-import { Storage, Bucket } from '@google-cloud/storage'
 import { LoggingWinston } from '@google-cloud/logging-winston'
+import { Bucket, Storage } from '@google-cloud/storage'
 import multer from 'multer'
 import { v4 as uuidv4 } from 'uuid'
 import { BaseStorageProvider } from './BaseStorageProvider'
@@ -336,7 +336,7 @@ export class GCSStorageProvider extends BaseStorageProvider {
         })
     }
 
-    getLoggerTransports(logType: 'server' | 'error' | 'requests'): any[] {
+    getLoggerTransports(logType: 'server' | 'error' | 'requests' | 'audit'): any[] {
         const gcsConfig = {
             projectId: this.projectId,
             keyFilename: this.keyFilename,
@@ -368,7 +368,15 @@ export class GCSStorageProvider extends BaseStorageProvider {
                     logName: 'requests'
                 })
             ]
+        } else if (logType === 'audit') {
+            return [
+                new LoggingWinston({
+                    ...gcsConfig,
+                    logName: 'audit'
+                })
+            ]
         }
+
         return []
     }
 }

packages/components/src/storage/IStorageProvider.ts

@@ -94,5 +94,5 @@ export interface IStorageProvider {
     /**
      * Get the Winston logger transports for this provider
      */
-    getLoggerTransports(logType: 'server' | 'error' | 'requests', config?: any): any[]
+    getLoggerTransports(logType: 'server' | 'error' | 'requests' | 'audit', config?: any): any[]
 }

packages/components/src/storage/LocalStorageProvider.ts

@@ -1,8 +1,8 @@
-import path from 'node:path'
 import fs from 'fs'
 import multer from 'multer'
-import DailyRotateFile from 'winston-daily-rotate-file'
+import path from 'node:path'
 import { transports } from 'winston'
+import DailyRotateFile from 'winston-daily-rotate-file'
 import { BaseStorageProvider } from './BaseStorageProvider'
 import { FileInfo, StorageResult, StorageSizeResult } from './IStorageProvider'
 
@@ -350,7 +350,7 @@ export class LocalStorageProvider extends BaseStorageProvider {
         return process.env.HOME || process.env.USERPROFILE || process.env.HOMEPATH || ''
     }
 
-    getLoggerTransports(logType: 'server' | 'error' | 'requests', config?: any): any[] {
+    getLoggerTransports(logType: 'server' | 'error' | 'requests' | 'audit', config?: any): any[] {
         const logDir = config?.logging?.dir || path.join(this.getUserHome(), '.flowise', 'logs')
 
         if (!fs.existsSync(logDir)) {
@@ -373,6 +373,15 @@ export class LocalStorageProvider extends BaseStorageProvider {
                     level: config?.logging?.express?.level ?? 'debug'
                 })
             ]
+        } else if (logType === 'audit') {
+            return [
+                new DailyRotateFile({
+                    filename: path.join(logDir, 'audit-%DATE%.log.jsonl'),
+                    datePattern: 'YYYY-MM-DD-HH',
+                    maxSize: '20m',
+                    level: 'info'
+                })
+            ]
         }
 
         // For 'error' type, return empty array (handled by exceptionHandlers in logger.ts)

packages/components/src/storage/S3StorageProvider.ts

@@ -7,11 +7,11 @@ import {
     S3Client,
     S3ClientConfig
 } from '@aws-sdk/client-s3'
-import { Readable } from 'node:stream'
 import multer from 'multer'
 import multerS3 from 'multer-s3'
-import { transports } from 'winston'
+import { Readable } from 'node:stream'
 import { v4 as uuidv4 } from 'uuid'
+import { transports } from 'winston'
 import { BaseStorageProvider } from './BaseStorageProvider'
 import { FileInfo, StorageResult, StorageSizeResult } from './IStorageProvider'
 
@@ -507,7 +507,7 @@ export class S3StorageProvider extends BaseStorageProvider {
         })
     }
 
-    getLoggerTransports(logType: 'server' | 'error' | 'requests'): any[] {
+    getLoggerTransports(logType: 'server' | 'error' | 'requests' | 'audit'): any[] {
         if (logType === 'server') {
             const s3ServerStream = new S3StreamLogger({
                 bucket: this.bucket,
@@ -532,7 +532,17 @@ export class S3StorageProvider extends BaseStorageProvider {
                 config: this.s3Config
             })
             return [new transports.Stream({ stream: s3ServerReqStream })]
+        } else if (logType === 'audit') {
+            const instance = process.env.HOSTNAME || process.env.POD_NAME || String(process.pid)
+            const s3AuditStream = new S3StreamLogger({
+                bucket: this.bucket,
+                folder: 'logs/audit',
+                name_format: `audit-%Y-%m-%d-%H-%M-%S-%L-${instance}.log.jsonl`,
+                config: this.s3Config
+            })
+            return [new transports.Stream({ stream: s3AuditStream })]
         }
+
         return []
     }
 }

packages/server/package.json

@@ -109,6 +109,7 @@
         "flowise-components": "workspace:^",
         "flowise-nim-container-manager": "^1.0.11",
         "flowise-ui": "workspace:^",
+        "geoip-lite": "^1.4.10",
         "global-agent": "^3.0.0",
         "gulp": "^4.0.2",
         "handlebars": "^4.7.8",
@@ -158,6 +159,7 @@
         "@types/cors": "^2.8.12",
         "@types/crypto-js": "^4.1.1",
         "@types/express-session": "^1.18.0",
+        "@types/geoip-lite": "^1.4.4",
         "@types/jest": "^29.5.14",
         "@types/jsonwebtoken": "^9.0.6",
         "@types/multer": "^1.4.7",

packages/server/src/IdentityManager.ts

@@ -16,6 +16,7 @@ import * as fs from 'fs'
 import { StatusCodes } from 'http-status-codes'
 import jwt from 'jsonwebtoken'
 import path from 'path'
+import Stripe from 'stripe'
 import { LoginMethodStatus } from './enterprise/database/entities/login-method.entity'
 import { ErrorMessage, LoggedInUser } from './enterprise/Interface.Enterprise'
 import { Permissions } from './enterprise/rbac/Permissions'
@@ -33,7 +34,6 @@ import { UsageCacheManager } from './UsageCacheManager'
 import { GeneralErrorMessage, LICENSE_QUOTAS } from './utils/constants'
 import { getRunningExpressApp } from './utils/getRunningExpressApp'
 import { ENTERPRISE_FEATURE_FLAGS } from './utils/quotaUsage'
-import Stripe from 'stripe'
 
 const allSSOProviders = ['azure', 'google', 'auth0', 'github']
 export class IdentityManager {
@@ -526,4 +526,24 @@ export class IdentityManager {
             throw error
         }
     }
+
+    /**
+     * Cancels a Stripe subscription and syncs the result to the usage cache.
+     *
+     * Requests cancellation via Stripe, then updates the subscription data in cache
+     * so usage and billing state stay consistent. Throws if the Stripe manager
+     * is not initialized.
+     *
+     * @param subscriptionId - The Stripe subscription ID to cancel
+     * @returns The cancelled Stripe subscription object
+     */
+    public async cancelSubscription(subscriptionId: string) {
+        if (!this.stripeManager) throw new Error('Stripe manager is not initialized')
+        const subscription = await this.stripeManager.cancelSubscription(subscriptionId)
+        const cacheManager = await UsageCacheManager.getInstance()
+        await cacheManager.updateSubscriptionDataToCache(subscriptionId, {
+            subsriptionDetails: this.stripeManager.getSubscriptionObject(subscription)
+        })
+        return subscription
+    }
 }

packages/server/src/StripeManager.ts

@@ -1,7 +1,7 @@
-import Stripe from 'stripe'
 import { Request } from 'express'
-import { UsageCacheManager } from './UsageCacheManager'
+import Stripe from 'stripe'
 import { UserPlan } from './Interface'
+import { UsageCacheManager } from './UsageCacheManager'
 import { LICENSE_QUOTAS } from './utils/constants'
 
 export class StripeManager {
@@ -611,4 +611,18 @@ export class StripeManager {
             throw error
         }
     }
+
+    /**
+     * Cancels a Stripe subscription immediately without proration or an immediate invoice.
+     *
+     * Calls the Stripe API to cancel the subscription with `prorate: false` and
+     * `invoice_now: false`. Throws if the Stripe client is not initialized.
+     *
+     * @param subscriptionId - The Stripe subscription ID to cancel
+     * @returns A promise resolving to the Stripe API response containing the cancelled subscription
+     */
+    public async cancelSubscription(subscriptionId: string): Promise<Stripe.Response<Stripe.Subscription>> {
+        if (!this.stripe) throw new Error('Stripe is not initialized')
+        return await this.stripe.subscriptions.cancel(subscriptionId, { prorate: false, invoice_now: false })
+    }
 }

packages/server/src/enterprise/controllers/account.controller.ts

@@ -1,6 +1,10 @@
 import { NextFunction, Request, Response } from 'express'
 import { StatusCodes } from 'http-status-codes'
+import { QueryRunner } from 'typeorm'
+import { InternalFlowiseError } from '../../errors/internalFlowiseError'
+import { GeneralErrorMessage } from '../../utils/constants'
 import { getRunningExpressApp } from '../../utils/getRunningExpressApp'
+import { emitEvent, TelemetryEventCategory, TelemetryEventResult } from '../../utils/telemetry'
 import { Organization } from '../database/entities/organization.entity'
 import { User } from '../database/entities/user.entity'
 import { AccountDTO, AccountService } from '../services/account.service'
@@ -146,6 +150,45 @@ export class AccountController {
             return res.json({ message: 'Authentication failed' })
         }
     }
+
+    public async delete(req: Request, res: Response, next: NextFunction) {
+        let queryRunner: QueryRunner | undefined
+        try {
+            const { confirmationText } = req.body
+            if (confirmationText !== 'permanently delete') {
+                throw new InternalFlowiseError(StatusCodes.BAD_REQUEST, 'Confirmation text must match "permanently delete"')
+            }
+
+            queryRunner = getRunningExpressApp().AppDataSource.createQueryRunner()
+            await queryRunner.connect()
+            if (!req.user || !req.ip) throw new InternalFlowiseError(StatusCodes.UNAUTHORIZED, GeneralErrorMessage.UNAUTHORIZED)
+
+            const accountService = new AccountService()
+            await accountService.delete(queryRunner, req.user, req.ip)
+
+            return res.status(StatusCodes.OK).json({ message: 'Account deleted' })
+        } catch (error) {
+            if (queryRunner && queryRunner.isTransactionActive) await queryRunner.rollbackTransaction()
+
+            await emitEvent({
+                category: TelemetryEventCategory.AUDIT,
+                eventType: 'account-deleted',
+                actionType: 'delete',
+                userId: req.user?.id ?? 'unknown',
+                orgId: req.user?.activeOrganizationId ?? 'unknown',
+                resourceId: req.user?.id ?? 'unknown',
+                ipAddress: req.ip,
+                result: TelemetryEventResult.FAILED,
+                metadata: {
+                    failureReason: error instanceof InternalFlowiseError ? error.message : 'internal_error'
+                }
+            })
+
+            next(error)
+        } finally {
+            if (queryRunner && !queryRunner.isReleased) await queryRunner.release()
+        }
+    }
 }
 
 function sanitizeRegistrationDTO(data: AccountDTO): AccountDTO {

packages/server/src/enterprise/routes/account.route.ts

@@ -1,6 +1,6 @@
 import express from 'express'
-import { AccountController } from '../controllers/account.controller'
 import { IdentityManager } from '../../IdentityManager'
+import { AccountController } from '../controllers/account.controller'
 import { checkAnyPermission } from '../rbac/PermissionCheck'
 
 const router = express.Router()
@@ -36,4 +36,6 @@ router.get('/basic-auth', accountController.getBasicAuth)
 
 router.post('/basic-auth', accountController.checkBasicAuth)
 
+router.delete('/delete', accountController.delete)
+
 export default router