mirror of
https://github.com/FlowiseAI/Flowise.git
synced 2026-10-12 02:02:17 +00:00
Add support for azure blob storage and storage refactor (#5604)
* storageUtils functions azure storage * interface, base and local storage provider * s3 storage provider * gcs, azure storage provider * refactor final changes * fix gcs path bug * gemini and lint fixes * documentation changes * update pnpm lock * update azure winston to ESM imports --------- Co-authored-by: Henry <[email protected]>
This commit is contained in:
chiragjagga
+44
-40
@@ -120,46 +120,50 @@ Flowise has 3 different modules in a single mono repository.
|
||||
|
||||
Flowise support different environment variables to configure your instance. You can specify the following variables in the `.env` file inside `packages/server` folder. Read [more](https://docs.flowiseai.com/environment-variables)
|
||||
|
||||
| Variable | Description | Type | Default |
|
||||
| ---------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------ | ----------------------------------- |
|
||||
| PORT | The HTTP port Flowise runs on | Number | 3000 |
|
||||
| CORS_ORIGINS | The allowed origins for all cross-origin HTTP calls | String | |
|
||||
| IFRAME_ORIGINS | The allowed origins for iframe src embedding | String | |
|
||||
| FLOWISE_FILE_SIZE_LIMIT | Upload File Size Limit | String | 50mb |
|
||||
| DEBUG | Print logs from components | Boolean | |
|
||||
| LOG_PATH | Location where log files are stored | String | `your-path/Flowise/logs` |
|
||||
| LOG_LEVEL | Different levels of logs | Enum String: `error`, `info`, `verbose`, `debug` | `info` |
|
||||
| LOG_JSON_SPACES | Spaces to beautify JSON logs | | 2 |
|
||||
| TOOL_FUNCTION_BUILTIN_DEP | NodeJS built-in modules to be used for Custom Tool or Function | String | |
|
||||
| TOOL_FUNCTION_EXTERNAL_DEP | External modules to be used for Custom Tool or Function | String | |
|
||||
| ALLOW_BUILTIN_DEP | Allow project dependencies to be used for Custom Tool or Function | Boolean | false |
|
||||
| DATABASE_TYPE | Type of database to store the flowise data | Enum String: `sqlite`, `mysql`, `postgres` | `sqlite` |
|
||||
| DATABASE_PATH | Location where database is saved (When DATABASE_TYPE is sqlite) | String | `your-home-dir/.flowise` |
|
||||
| DATABASE_HOST | Host URL or IP address (When DATABASE_TYPE is not sqlite) | String | |
|
||||
| DATABASE_PORT | Database port (When DATABASE_TYPE is not sqlite) | String | |
|
||||
| DATABASE_USER | Database username (When DATABASE_TYPE is not sqlite) | String | |
|
||||
| DATABASE_PASSWORD | Database password (When DATABASE_TYPE is not sqlite) | String | |
|
||||
| DATABASE_NAME | Database name (When DATABASE_TYPE is not sqlite) | String | |
|
||||
| DATABASE_SSL_KEY_BASE64 | Database SSL client cert in base64 (takes priority over DATABASE_SSL) | Boolean | false |
|
||||
| DATABASE_SSL | Database connection overssl (When DATABASE_TYPE is postgre) | Boolean | false |
|
||||
| SECRETKEY_PATH | Location where encryption key (used to encrypt/decrypt credentials) is saved | String | `your-path/Flowise/packages/server` |
|
||||
| FLOWISE_SECRETKEY_OVERWRITE | Encryption key to be used instead of the key stored in SECRETKEY_PATH | String | |
|
||||
| MODEL_LIST_CONFIG_JSON | File path to load list of models from your local config file | String | `/your_model_list_config_file_path` |
|
||||
| STORAGE_TYPE | Type of storage for uploaded files. default is `local` | Enum String: `s3`, `local`, `gcs` | `local` |
|
||||
| BLOB_STORAGE_PATH | Local folder path where uploaded files are stored when `STORAGE_TYPE` is `local` | String | `your-home-dir/.flowise/storage` |
|
||||
| S3_STORAGE_BUCKET_NAME | Bucket name to hold the uploaded files when `STORAGE_TYPE` is `s3` | String | |
|
||||
| S3_STORAGE_ACCESS_KEY_ID | AWS Access Key | String | |
|
||||
| S3_STORAGE_SECRET_ACCESS_KEY | AWS Secret Key | String | |
|
||||
| S3_STORAGE_REGION | Region for S3 bucket | String | |
|
||||
| S3_ENDPOINT_URL | Custom Endpoint for S3 | String | |
|
||||
| S3_FORCE_PATH_STYLE | Set this to true to force the request to use path-style addressing | Boolean | false |
|
||||
| GOOGLE_CLOUD_STORAGE_PROJ_ID | The GCP project id for cloud storage & logging when `STORAGE_TYPE` is `gcs` | String | |
|
||||
| GOOGLE_CLOUD_STORAGE_CREDENTIAL | The credential key file path when `STORAGE_TYPE` is `gcs` | String | |
|
||||
| GOOGLE_CLOUD_STORAGE_BUCKET_NAME | Bucket name to hold the uploaded files when `STORAGE_TYPE` is `gcs` | String | |
|
||||
| GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS | Enable uniform bucket level access when `STORAGE_TYPE` is `gcs` | Boolean | true |
|
||||
| SHOW_COMMUNITY_NODES | Show nodes created by community | Boolean | |
|
||||
| DISABLED_NODES | Hide nodes from UI (comma separated list of node names) | String | |
|
||||
| TRUST_PROXY | Configure proxy trust settings for proper IP detection. Values: 'true' (trust all), 'false' (disable), number (hop count), or Express proxy values (e.g., 'loopback', 'linklocal', 'uniquelocal', IP addresses). [Learn More](https://expressjs.com/en/guide/behind-proxies.html) | Boolean/String/Number | true |
|
||||
| Variable | Description | Type | Default |
|
||||
| ------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------ | ----------------------------------- |
|
||||
| PORT | The HTTP port Flowise runs on | Number | 3000 |
|
||||
| CORS_ORIGINS | The allowed origins for all cross-origin HTTP calls | String | |
|
||||
| IFRAME_ORIGINS | The allowed origins for iframe src embedding | String | |
|
||||
| FLOWISE_FILE_SIZE_LIMIT | Upload File Size Limit | String | 50mb |
|
||||
| DEBUG | Print logs from components | Boolean | |
|
||||
| LOG_PATH | Location where log files are stored | String | `your-path/Flowise/logs` |
|
||||
| LOG_LEVEL | Different levels of logs | Enum String: `error`, `info`, `verbose`, `debug` | `info` |
|
||||
| LOG_JSON_SPACES | Spaces to beautify JSON logs | | 2 |
|
||||
| TOOL_FUNCTION_BUILTIN_DEP | NodeJS built-in modules to be used for Custom Tool or Function | String | |
|
||||
| TOOL_FUNCTION_EXTERNAL_DEP | External modules to be used for Custom Tool or Function | String | |
|
||||
| ALLOW_BUILTIN_DEP | Allow project dependencies to be used for Custom Tool or Function | Boolean | false |
|
||||
| DATABASE_TYPE | Type of database to store the flowise data | Enum String: `sqlite`, `mysql`, `postgres` | `sqlite` |
|
||||
| DATABASE_PATH | Location where database is saved (When DATABASE_TYPE is sqlite) | String | `your-home-dir/.flowise` |
|
||||
| DATABASE_HOST | Host URL or IP address (When DATABASE_TYPE is not sqlite) | String | |
|
||||
| DATABASE_PORT | Database port (When DATABASE_TYPE is not sqlite) | String | |
|
||||
| DATABASE_USER | Database username (When DATABASE_TYPE is not sqlite) | String | |
|
||||
| DATABASE_PASSWORD | Database password (When DATABASE_TYPE is not sqlite) | String | |
|
||||
| DATABASE_NAME | Database name (When DATABASE_TYPE is not sqlite) | String | |
|
||||
| DATABASE_SSL_KEY_BASE64 | Database SSL client cert in base64 (takes priority over DATABASE_SSL) | Boolean | false |
|
||||
| DATABASE_SSL | Database connection overssl (When DATABASE_TYPE is postgre) | Boolean | false |
|
||||
| SECRETKEY_PATH | Location where encryption key (used to encrypt/decrypt credentials) is saved | String | `your-path/Flowise/packages/server` |
|
||||
| FLOWISE_SECRETKEY_OVERWRITE | Encryption key to be used instead of the key stored in SECRETKEY_PATH | String | |
|
||||
| MODEL_LIST_CONFIG_JSON | File path to load list of models from your local config file | String | `/your_model_list_config_file_path` |
|
||||
| STORAGE_TYPE | Type of storage for uploaded files. default is `local` | Enum String: `s3`, `local`, `gcs` ,`azure` | `local` |
|
||||
| BLOB_STORAGE_PATH | Local folder path where uploaded files are stored when `STORAGE_TYPE` is `local` | String | `your-home-dir/.flowise/storage` |
|
||||
| S3_STORAGE_BUCKET_NAME | Bucket name to hold the uploaded files when `STORAGE_TYPE` is `s3` | String | |
|
||||
| S3_STORAGE_ACCESS_KEY_ID | AWS Access Key | String | |
|
||||
| S3_STORAGE_SECRET_ACCESS_KEY | AWS Secret Key | String | |
|
||||
| S3_STORAGE_REGION | Region for S3 bucket | String | |
|
||||
| S3_ENDPOINT_URL | Custom Endpoint for S3 | String | |
|
||||
| S3_FORCE_PATH_STYLE | Set this to true to force the request to use path-style addressing | Boolean | false |
|
||||
| GOOGLE_CLOUD_STORAGE_PROJ_ID | The GCP project id for cloud storage & logging when `STORAGE_TYPE` is `gcs` | String | |
|
||||
| GOOGLE_CLOUD_STORAGE_CREDENTIAL | The credential key file path when `STORAGE_TYPE` is `gcs` | String | |
|
||||
| GOOGLE_CLOUD_STORAGE_BUCKET_NAME | Bucket name to hold the uploaded files when `STORAGE_TYPE` is `gcs` | String | |
|
||||
| GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS | Enable uniform bucket level access when `STORAGE_TYPE` is `gcs` | Boolean | true |
|
||||
| AZURE_BLOB_STORAGE_CONNECTION_STRING | Azure Blob Storage connection string when `STORAGE_TYPE` is `azure`. Either this or account name + key is required | String | |
|
||||
| AZURE_BLOB_STORAGE_ACCOUNT_NAME | Azure storage account name when `STORAGE_TYPE` is `azure`. Required if connection string is not provided | String | |
|
||||
| AZURE_BLOB_STORAGE_ACCOUNT_KEY | Azure storage account key when `STORAGE_TYPE` is `azure`. Required if connection string is not provided | String | |
|
||||
| AZURE_BLOB_STORAGE_CONTAINER_NAME | Container name to hold the uploaded files when `STORAGE_TYPE` is `azure` | String | |
|
||||
| SHOW_COMMUNITY_NODES | Show nodes created by community | Boolean | |
|
||||
| DISABLED_NODES | Hide nodes from UI (comma separated list of node names) | String | |
|
||||
| TRUST_PROXY | Configure proxy trust settings for proper IP detection. Values: 'true' (trust all), 'false' (disable), number (hop count), or Express proxy values (e.g., 'loopback', 'linklocal', 'uniquelocal', IP addresses). [Learn More](https://expressjs.com/en/guide/behind-proxies.html) | Boolean/String/Number | true |
|
||||
|
||||
You can also specify the env variables when using `npx`. For example:
|
||||
|
||||
|
||||
+7
-1
@@ -49,7 +49,7 @@ PORT=3000
|
||||
############################################## STORAGE #####################################################
|
||||
############################################################################################################
|
||||
|
||||
# STORAGE_TYPE=local (local | s3 | gcs)
|
||||
# STORAGE_TYPE=local (local | s3 | gcs | azure)
|
||||
# BLOB_STORAGE_PATH=/your_storage_path/.flowise/storage
|
||||
# S3_STORAGE_BUCKET_NAME=flowise
|
||||
# S3_STORAGE_ACCESS_KEY_ID=<your-access-key>
|
||||
@@ -61,6 +61,12 @@ PORT=3000
|
||||
# GOOGLE_CLOUD_STORAGE_PROJ_ID=<your-gcp-project-id>
|
||||
# GOOGLE_CLOUD_STORAGE_BUCKET_NAME=<the-bucket-name>
|
||||
# GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS=true
|
||||
# Azure Blob Storage (when STORAGE_TYPE=azure)
|
||||
# Provide EITHER a connection string OR account name + key
|
||||
# AZURE_BLOB_STORAGE_CONNECTION_STRING=<your-connection-string>
|
||||
# AZURE_BLOB_STORAGE_ACCOUNT_NAME=<your-account-name>
|
||||
# AZURE_BLOB_STORAGE_ACCOUNT_KEY=<your-account-key>
|
||||
# AZURE_BLOB_STORAGE_CONTAINER_NAME=<your-container-name>
|
||||
|
||||
|
||||
############################################################################################################
|
||||
|
||||
@@ -67,6 +67,11 @@ services:
|
||||
- GOOGLE_CLOUD_STORAGE_PROJ_ID=${GOOGLE_CLOUD_STORAGE_PROJ_ID}
|
||||
- GOOGLE_CLOUD_STORAGE_BUCKET_NAME=${GOOGLE_CLOUD_STORAGE_BUCKET_NAME}
|
||||
- GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS=${GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS}
|
||||
# Azure Blob Storage (provide EITHER connection string OR account name + key)
|
||||
- AZURE_BLOB_STORAGE_CONNECTION_STRING=${AZURE_BLOB_STORAGE_CONNECTION_STRING}
|
||||
- AZURE_BLOB_STORAGE_ACCOUNT_NAME=${AZURE_BLOB_STORAGE_ACCOUNT_NAME}
|
||||
- AZURE_BLOB_STORAGE_ACCOUNT_KEY=${AZURE_BLOB_STORAGE_ACCOUNT_KEY}
|
||||
- AZURE_BLOB_STORAGE_CONTAINER_NAME=${AZURE_BLOB_STORAGE_CONTAINER_NAME}
|
||||
|
||||
# SETTINGS
|
||||
- NUMBER_OF_PROXIES=${NUMBER_OF_PROXIES}
|
||||
|
||||
@@ -52,6 +52,11 @@ services:
|
||||
- GOOGLE_CLOUD_STORAGE_PROJ_ID=${GOOGLE_CLOUD_STORAGE_PROJ_ID}
|
||||
- GOOGLE_CLOUD_STORAGE_BUCKET_NAME=${GOOGLE_CLOUD_STORAGE_BUCKET_NAME}
|
||||
- GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS=${GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS}
|
||||
# Azure Blob Storage (provide EITHER connection string OR account name + key)
|
||||
- AZURE_BLOB_STORAGE_CONNECTION_STRING=${AZURE_BLOB_STORAGE_CONNECTION_STRING}
|
||||
- AZURE_BLOB_STORAGE_ACCOUNT_NAME=${AZURE_BLOB_STORAGE_ACCOUNT_NAME}
|
||||
- AZURE_BLOB_STORAGE_ACCOUNT_KEY=${AZURE_BLOB_STORAGE_ACCOUNT_KEY}
|
||||
- AZURE_BLOB_STORAGE_CONTAINER_NAME=${AZURE_BLOB_STORAGE_CONTAINER_NAME}
|
||||
|
||||
# SETTINGS
|
||||
- NUMBER_OF_PROXIES=${NUMBER_OF_PROXIES}
|
||||
|
||||
@@ -49,7 +49,7 @@ WORKER_PORT=5566
|
||||
############################################## STORAGE #####################################################
|
||||
############################################################################################################
|
||||
|
||||
# STORAGE_TYPE=local (local | s3 | gcs)
|
||||
# STORAGE_TYPE=local (local | s3 | gcs | azure)
|
||||
# BLOB_STORAGE_PATH=/your_storage_path/.flowise/storage
|
||||
# S3_STORAGE_BUCKET_NAME=flowise
|
||||
# S3_STORAGE_ACCESS_KEY_ID=<your-access-key>
|
||||
@@ -61,6 +61,12 @@ WORKER_PORT=5566
|
||||
# GOOGLE_CLOUD_STORAGE_PROJ_ID=<your-gcp-project-id>
|
||||
# GOOGLE_CLOUD_STORAGE_BUCKET_NAME=<the-bucket-name>
|
||||
# GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS=true
|
||||
# Azure Blob Storage (when STORAGE_TYPE=azure)
|
||||
# Provide EITHER a connection string OR account name + key
|
||||
# AZURE_BLOB_STORAGE_CONNECTION_STRING=<your-connection-string>
|
||||
# AZURE_BLOB_STORAGE_ACCOUNT_NAME=<your-account-name>
|
||||
# AZURE_BLOB_STORAGE_ACCOUNT_KEY=<your-account-key>
|
||||
# AZURE_BLOB_STORAGE_CONTAINER_NAME=<your-container-name>
|
||||
|
||||
|
||||
############################################################################################################
|
||||
|
||||
@@ -52,6 +52,11 @@ services:
|
||||
- GOOGLE_CLOUD_STORAGE_PROJ_ID=${GOOGLE_CLOUD_STORAGE_PROJ_ID}
|
||||
- GOOGLE_CLOUD_STORAGE_BUCKET_NAME=${GOOGLE_CLOUD_STORAGE_BUCKET_NAME}
|
||||
- GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS=${GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS}
|
||||
# Azure Blob Storage (provide EITHER connection string OR account name + key)
|
||||
- AZURE_BLOB_STORAGE_CONNECTION_STRING=${AZURE_BLOB_STORAGE_CONNECTION_STRING}
|
||||
- AZURE_BLOB_STORAGE_ACCOUNT_NAME=${AZURE_BLOB_STORAGE_ACCOUNT_NAME}
|
||||
- AZURE_BLOB_STORAGE_ACCOUNT_KEY=${AZURE_BLOB_STORAGE_ACCOUNT_KEY}
|
||||
- AZURE_BLOB_STORAGE_CONTAINER_NAME=${AZURE_BLOB_STORAGE_CONTAINER_NAME}
|
||||
|
||||
# SETTINGS
|
||||
- NUMBER_OF_PROXIES=${NUMBER_OF_PROXIES}
|
||||
|
||||
+207
-207
@@ -1,207 +1,207 @@
|
||||
# @flowiseai/agentflow
|
||||
|
||||
[](https://www.npmjs.com/package/@flowiseai/agentflow)
|
||||
[](https://github.com/FlowiseAI/Flowise/blob/main/LICENSE.md)
|
||||
|
||||
> Embeddable React component for building and visualizing AI agent workflows
|
||||
|
||||
## ⚠️ Status
|
||||
|
||||
**This package is currently under active development.**
|
||||
|
||||
- 🚧 Components are not yet fully functional
|
||||
- ❌ End-to-end functionality is not complete
|
||||
- 🔄 Features are still being implemented and tested
|
||||
- ⚡ APIs may change before stable release
|
||||
- 📝 Documentation is being updated as development progresses
|
||||
|
||||
**Cannot be used in production. For development and testing purposes only.**
|
||||
|
||||
## Overview
|
||||
|
||||
`@flowiseai/agentflow` is a React-based flow editor for creating AI agent workflows. It provides a visual canvas built on ReactFlow for connecting AI agents, LLMs, tools, and logic nodes.
|
||||
|
||||
## Installation
|
||||
|
||||
```bash
|
||||
pnpm add @flowiseai/agentflow
|
||||
```
|
||||
|
||||
**Peer Dependencies:**
|
||||
|
||||
```bash
|
||||
pnpm add react react-dom @mui/material @mui/icons-material @emotion/react @emotion/styled reactflow
|
||||
```
|
||||
|
||||
## Basic Usage
|
||||
|
||||
```tsx
|
||||
import { Agentflow } from '@flowiseai/agentflow'
|
||||
|
||||
import '@flowiseai/agentflow/flowise.css'
|
||||
|
||||
export default function App() {
|
||||
return (
|
||||
<div style={{ width: '100vw', height: '100vh' }}>
|
||||
<Agentflow apiBaseUrl='http://localhost:3000' token='your-api-key' />
|
||||
</div>
|
||||
)
|
||||
}
|
||||
```
|
||||
|
||||
### With Initial Flow Data and Callbacks
|
||||
|
||||
```tsx
|
||||
import { useRef } from 'react'
|
||||
|
||||
import { Agentflow, type AgentFlowInstance, type FlowData } from '@flowiseai/agentflow'
|
||||
|
||||
import '@flowiseai/agentflow/flowise.css'
|
||||
|
||||
export default function App() {
|
||||
const ref = useRef<AgentFlowInstance>(null)
|
||||
|
||||
const initialFlow: FlowData = {
|
||||
nodes: [
|
||||
{
|
||||
id: 'startAgentflow_0',
|
||||
type: 'agentflowNode',
|
||||
position: { x: 100, y: 100 },
|
||||
data: {
|
||||
id: 'startAgentflow_0',
|
||||
name: 'startAgentflow',
|
||||
label: 'Start',
|
||||
color: '#7EE787',
|
||||
hideInput: true,
|
||||
outputAnchors: [{ id: 'startAgentflow_0-output-0', name: 'start', label: 'Start', type: 'start' }]
|
||||
}
|
||||
}
|
||||
],
|
||||
edges: [],
|
||||
viewport: { x: 0, y: 0, zoom: 1 }
|
||||
}
|
||||
|
||||
return (
|
||||
<div style={{ width: '100vw', height: '100vh' }}>
|
||||
<Agentflow
|
||||
ref={ref}
|
||||
apiBaseUrl='http://localhost:3000'
|
||||
token='your-api-key'
|
||||
initialFlow={initialFlow}
|
||||
onFlowChange={(flow) => console.log('Flow changed:', flow)}
|
||||
onSave={(flow) => console.log('Flow saved:', flow)}
|
||||
/>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
```
|
||||
|
||||
## Props
|
||||
|
||||
| Prop | Type | Default | Description |
|
||||
| -------------------- | ------------------------------------------ | -------------- | --------------------------------------------------------------- |
|
||||
| `apiBaseUrl` | `string` | **(required)** | Flowise API server endpoint |
|
||||
| `token` | `string` | — | Authentication token for API calls |
|
||||
| `initialFlow` | `FlowData` | — | Initial flow data to render (uncontrolled — only used on mount) |
|
||||
| `components` | `string[]` | — | Restrict which node types appear in the palette |
|
||||
| `onFlowChange` | `(flow: FlowData) => void` | — | Called when the flow changes (node/edge add, remove, move) |
|
||||
| `onSave` | `(flow: FlowData) => void` | — | Called when the user triggers a save |
|
||||
| `onFlowGenerated` | `(flow: FlowData) => void` | — | Called when a flow is generated via AI |
|
||||
| `isDarkMode` | `boolean` | `false` | Use dark mode theme |
|
||||
| `readOnly` | `boolean` | `false` | Disable editing (nodes not draggable/connectable) |
|
||||
| `showDefaultHeader` | `boolean` | `true` | Show built-in header (ignored if `renderHeader` provided) |
|
||||
| `showDefaultPalette` | `boolean` | `true` | Show built-in node palette |
|
||||
| `enableGenerator` | `boolean` | `true` | Show the AI flow generator button |
|
||||
| `renderHeader` | `(props: HeaderRenderProps) => ReactNode` | — | Custom header renderer |
|
||||
| `renderNodePalette` | `(props: PaletteRenderProps) => ReactNode` | — | Custom node palette renderer |
|
||||
|
||||
### Imperative Methods (via `ref`)
|
||||
|
||||
| Method | Return Type | Description |
|
||||
| ------------------------ | ------------------------- | ------------------------------------- |
|
||||
| `getFlow()` | `FlowData` | Get current flow data |
|
||||
| `toJSON()` | `string` | Export flow as JSON string |
|
||||
| `validate()` | `ValidationResult` | Validate the current flow |
|
||||
| `fitView()` | `void` | Fit all nodes into view |
|
||||
| `clear()` | `void` | Remove all nodes and edges |
|
||||
| `addNode(nodeData)` | `void` | Add a node (`Partial<FlowNode>`) |
|
||||
| `getReactFlowInstance()` | `ReactFlowInstance\|null` | Get the underlying ReactFlow instance |
|
||||
|
||||
### Design Note
|
||||
|
||||
`<Agentflow>` is an **uncontrolled component**. The `initialFlow` prop seeds the canvas state on mount, but the component owns its own state afterward. Use the `ref` for imperative access and `onFlowChange` to observe changes.
|
||||
|
||||
## Development
|
||||
|
||||
```bash
|
||||
# Install dependencies
|
||||
pnpm install
|
||||
|
||||
# Build the package
|
||||
pnpm build
|
||||
|
||||
# Run tests
|
||||
pnpm test
|
||||
|
||||
# Run examples
|
||||
cd examples && pnpm install && pnpm dev
|
||||
```
|
||||
|
||||
Visit the [examples](./examples) directory for more usage patterns. See [TESTS.md](./TESTS.md) for the full test plan and coverage status.
|
||||
|
||||
## Publishing
|
||||
|
||||
### Version Update
|
||||
|
||||
Bump the version in `package.json` before publishing. Use `npm version` to update the version and create a git tag:
|
||||
|
||||
```bash
|
||||
# Prerelease (for testing)
|
||||
npm version prerelease --preid=dev # 0.0.0-dev.1 → 0.0.0-dev.2
|
||||
|
||||
# Patch / Minor / Major (for stable releases)
|
||||
npm version patch # 0.0.1
|
||||
npm version minor # 0.1.0
|
||||
npm version major # 1.0.0
|
||||
```
|
||||
|
||||
### Verify Before Publishing
|
||||
|
||||
```bash
|
||||
# Build and check the tarball contents
|
||||
pnpm build
|
||||
npm pack --dry-run
|
||||
|
||||
# Full publish dry-run (runs prepublishOnly + simulates upload)
|
||||
npm publish --dry-run
|
||||
```
|
||||
|
||||
### Publish
|
||||
|
||||
```bash
|
||||
# Prerelease — tagged so `npm install @flowiseai/agentflow` won't pick it up
|
||||
npm publish --tag dev
|
||||
|
||||
# Stable release — gets the `latest` tag
|
||||
npm publish
|
||||
```
|
||||
|
||||
> The `prepublishOnly` script automatically runs `clean` and `build` before every publish, so stale dist files are never uploaded.
|
||||
|
||||
## Documentation
|
||||
|
||||
- [ARCHITECTURE.md](./ARCHITECTURE.md) - Internal architecture and design patterns
|
||||
- [TESTS.md](./TESTS.md) - Test plan, coverage tiers, and configuration
|
||||
- [Examples](./examples/README.md) - Usage examples and demos
|
||||
|
||||
## Contributing
|
||||
|
||||
This package follows a feature-based architecture with clear separation of concerns. See [ARCHITECTURE.md](./ARCHITECTURE.md) for details on the project structure and development guidelines.
|
||||
|
||||
## License
|
||||
|
||||
Apache-2.0 — see the repository root [LICENSE.md](https://github.com/FlowiseAI/Flowise/blob/main/LICENSE.md) for details.
|
||||
|
||||
---
|
||||
|
||||
Part of the [Flowise](https://github.com/FlowiseAI/Flowise) ecosystem
|
||||
# @flowiseai/agentflow
|
||||
|
||||
[](https://www.npmjs.com/package/@flowiseai/agentflow)
|
||||
[](https://github.com/FlowiseAI/Flowise/blob/main/LICENSE.md)
|
||||
|
||||
> Embeddable React component for building and visualizing AI agent workflows
|
||||
|
||||
## ⚠️ Status
|
||||
|
||||
**This package is currently under active development.**
|
||||
|
||||
- 🚧 Components are not yet fully functional
|
||||
- ❌ End-to-end functionality is not complete
|
||||
- 🔄 Features are still being implemented and tested
|
||||
- ⚡ APIs may change before stable release
|
||||
- 📝 Documentation is being updated as development progresses
|
||||
|
||||
**Cannot be used in production. For development and testing purposes only.**
|
||||
|
||||
## Overview
|
||||
|
||||
`@flowiseai/agentflow` is a React-based flow editor for creating AI agent workflows. It provides a visual canvas built on ReactFlow for connecting AI agents, LLMs, tools, and logic nodes.
|
||||
|
||||
## Installation
|
||||
|
||||
```bash
|
||||
pnpm add @flowiseai/agentflow
|
||||
```
|
||||
|
||||
**Peer Dependencies:**
|
||||
|
||||
```bash
|
||||
pnpm add react react-dom @mui/material @mui/icons-material @emotion/react @emotion/styled reactflow
|
||||
```
|
||||
|
||||
## Basic Usage
|
||||
|
||||
```tsx
|
||||
import { Agentflow } from '@flowiseai/agentflow'
|
||||
|
||||
import '@flowiseai/agentflow/flowise.css'
|
||||
|
||||
export default function App() {
|
||||
return (
|
||||
<div style={{ width: '100vw', height: '100vh' }}>
|
||||
<Agentflow apiBaseUrl='http://localhost:3000' token='your-api-key' />
|
||||
</div>
|
||||
)
|
||||
}␍
|
||||
```
|
||||
|
||||
### With Initial Flow Data and Callbacks
|
||||
|
||||
```tsx
|
||||
import { useRef } from 'react'
|
||||
|
||||
import { Agentflow, type AgentFlowInstance, type FlowData } from '@flowiseai/agentflow'
|
||||
|
||||
import '@flowiseai/agentflow/flowise.css'
|
||||
|
||||
export default function App() {
|
||||
const ref = useRef<AgentFlowInstance>(null)
|
||||
|
||||
const initialFlow: FlowData = {
|
||||
nodes: [
|
||||
{
|
||||
id: 'startAgentflow_0',
|
||||
type: 'agentflowNode',
|
||||
position: { x: 100, y: 100 },
|
||||
data: {
|
||||
id: 'startAgentflow_0',
|
||||
name: 'startAgentflow',
|
||||
label: 'Start',
|
||||
color: '#7EE787',
|
||||
hideInput: true,
|
||||
outputAnchors: [{ id: 'startAgentflow_0-output-0', name: 'start', label: 'Start', type: 'start' }]
|
||||
}
|
||||
}
|
||||
],
|
||||
edges: [],
|
||||
viewport: { x: 0, y: 0, zoom: 1 }
|
||||
}
|
||||
|
||||
return (
|
||||
<div style={{ width: '100vw', height: '100vh' }}>
|
||||
<Agentflow
|
||||
ref={ref}
|
||||
apiBaseUrl='http://localhost:3000'
|
||||
token='your-api-key'
|
||||
initialFlow={initialFlow}
|
||||
onFlowChange={(flow) => console.log('Flow changed:', flow)}
|
||||
onSave={(flow) => console.log('Flow saved:', flow)}
|
||||
/>
|
||||
</div>
|
||||
)
|
||||
}␍
|
||||
```
|
||||
|
||||
## Props
|
||||
|
||||
| Prop | Type | Default | Description |
|
||||
| -------------------- | ------------------------------------------ | -------------- | --------------------------------------------------------------- |
|
||||
| `apiBaseUrl` | `string` | **(required)** | Flowise API server endpoint |
|
||||
| `token` | `string` | — | Authentication token for API calls |
|
||||
| `initialFlow` | `FlowData` | — | Initial flow data to render (uncontrolled — only used on mount) |
|
||||
| `components` | `string[]` | — | Restrict which node types appear in the palette |
|
||||
| `onFlowChange` | `(flow: FlowData) => void` | — | Called when the flow changes (node/edge add, remove, move) |
|
||||
| `onSave` | `(flow: FlowData) => void` | — | Called when the user triggers a save |
|
||||
| `onFlowGenerated` | `(flow: FlowData) => void` | — | Called when a flow is generated via AI |
|
||||
| `isDarkMode` | `boolean` | `false` | Use dark mode theme |
|
||||
| `readOnly` | `boolean` | `false` | Disable editing (nodes not draggable/connectable) |
|
||||
| `showDefaultHeader` | `boolean` | `true` | Show built-in header (ignored if `renderHeader` provided) |
|
||||
| `showDefaultPalette` | `boolean` | `true` | Show built-in node palette |
|
||||
| `enableGenerator` | `boolean` | `true` | Show the AI flow generator button |
|
||||
| `renderHeader` | `(props: HeaderRenderProps) => ReactNode` | — | Custom header renderer |
|
||||
| `renderNodePalette` | `(props: PaletteRenderProps) => ReactNode` | — | Custom node palette renderer |
|
||||
|
||||
### Imperative Methods (via `ref`)
|
||||
|
||||
| Method | Return Type | Description |
|
||||
| ------------------------ | ------------------------- | ------------------------------------- |
|
||||
| `getFlow()` | `FlowData` | Get current flow data |
|
||||
| `toJSON()` | `string` | Export flow as JSON string |
|
||||
| `validate()` | `ValidationResult` | Validate the current flow |
|
||||
| `fitView()` | `void` | Fit all nodes into view |
|
||||
| `clear()` | `void` | Remove all nodes and edges |
|
||||
| `addNode(nodeData)` | `void` | Add a node (`Partial<FlowNode>`) |
|
||||
| `getReactFlowInstance()` | `ReactFlowInstance\|null` | Get the underlying ReactFlow instance |
|
||||
|
||||
### Design Note
|
||||
|
||||
`<Agentflow>` is an **uncontrolled component**. The `initialFlow` prop seeds the canvas state on mount, but the component owns its own state afterward. Use the `ref` for imperative access and `onFlowChange` to observe changes.
|
||||
|
||||
## Development
|
||||
|
||||
```bash
|
||||
# Install dependencies
|
||||
pnpm install
|
||||
|
||||
# Build the package
|
||||
pnpm build
|
||||
|
||||
# Run tests
|
||||
pnpm test
|
||||
|
||||
# Run examples
|
||||
cd examples && pnpm install && pnpm dev
|
||||
```
|
||||
|
||||
Visit the [examples](./examples) directory for more usage patterns. See [TESTS.md](./TESTS.md) for the full test plan and coverage status.
|
||||
|
||||
## Publishing
|
||||
|
||||
### Version Update
|
||||
|
||||
Bump the version in `package.json` before publishing. Use `npm version` to update the version and create a git tag:
|
||||
|
||||
```bash
|
||||
# Prerelease (for testing)
|
||||
npm version prerelease --preid=dev # 0.0.0-dev.1 → 0.0.0-dev.2
|
||||
|
||||
# Patch / Minor / Major (for stable releases)
|
||||
npm version patch # 0.0.1
|
||||
npm version minor # 0.1.0
|
||||
npm version major # 1.0.0
|
||||
```
|
||||
|
||||
### Verify Before Publishing
|
||||
|
||||
```bash
|
||||
# Build and check the tarball contents
|
||||
pnpm build
|
||||
npm pack --dry-run
|
||||
|
||||
# Full publish dry-run (runs prepublishOnly + simulates upload)
|
||||
npm publish --dry-run
|
||||
```
|
||||
|
||||
### Publish
|
||||
|
||||
```bash
|
||||
# Prerelease — tagged so `npm install @flowiseai/agentflow` won't pick it up
|
||||
npm publish --tag dev
|
||||
|
||||
# Stable release — gets the `latest` tag
|
||||
npm publish
|
||||
```
|
||||
|
||||
> The `prepublishOnly` script automatically runs `clean` and `build` before every publish, so stale dist files are never uploaded.
|
||||
|
||||
## Documentation
|
||||
|
||||
- [ARCHITECTURE.md](./ARCHITECTURE.md) - Internal architecture and design patterns
|
||||
- [TESTS.md](./TESTS.md) - Test plan, coverage tiers, and configuration
|
||||
- [Examples](./examples/README.md) - Usage examples and demos
|
||||
|
||||
## Contributing
|
||||
|
||||
This package follows a feature-based architecture with clear separation of concerns. See [ARCHITECTURE.md](./ARCHITECTURE.md) for details on the project structure and development guidelines.
|
||||
|
||||
## License
|
||||
|
||||
Apache-2.0 — see the repository root [LICENSE.md](https://github.com/FlowiseAI/Flowise/blob/main/LICENSE.md) for details.
|
||||
|
||||
---
|
||||
|
||||
Part of the [Flowise](https://github.com/FlowiseAI/Flowise) ecosystem
|
||||
|
||||
@@ -42,6 +42,7 @@
|
||||
"@gomomento/sdk-core": "^1.51.1",
|
||||
"@google-ai/generativelanguage": "^2.5.0",
|
||||
"@google-cloud/storage": "^7.15.2",
|
||||
"@azure/storage-blob": "^12.29.0",
|
||||
"@google/generative-ai": "^0.24.0",
|
||||
"@grpc/grpc-js": "^1.10.10",
|
||||
"@huggingface/inference": "^4.13.2",
|
||||
|
||||
@@ -9,6 +9,7 @@ export * from './utils'
|
||||
export * from './speechToText'
|
||||
export * from './textToSpeech'
|
||||
export * from './storageUtils'
|
||||
export * from './storage'
|
||||
export * from './handler'
|
||||
export * from '../evaluation/EvaluationRunner'
|
||||
export * from './followUpPrompts'
|
||||
|
||||
@@ -0,0 +1,331 @@
|
||||
import { BlobServiceClient, ContainerClient, StorageSharedKeyCredential } from '@azure/storage-blob'
|
||||
import multer from 'multer'
|
||||
import { MulterAzureStorage } from 'multer-azure-blob-storage'
|
||||
import { v4 as uuidv4 } from 'uuid'
|
||||
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.
|
||||
* The server expects file.path (local/GCS) or file.key (S3) but multer-azure-blob-storage
|
||||
* only sets file.blobName. This subclass bridges that gap.
|
||||
*/
|
||||
class MulterAzureStorageWithPath extends MulterAzureStorage {
|
||||
_handleFile(req: any, file: any, cb: (error?: any, info?: any) => void): Promise<void> {
|
||||
return super._handleFile(req, file, (err: any, info: any) => {
|
||||
if (!err && info) {
|
||||
info.path = info.blobName
|
||||
}
|
||||
cb(err, info)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
export class AzureBlobStorageProvider extends BaseStorageProvider {
|
||||
private containerClient: ContainerClient
|
||||
private containerName: string
|
||||
|
||||
constructor() {
|
||||
super()
|
||||
const config = this.initAzureConfig()
|
||||
this.containerClient = config.containerClient
|
||||
this.containerName = config.containerName
|
||||
}
|
||||
|
||||
private initAzureConfig(): {
|
||||
containerClient: ContainerClient
|
||||
containerName: string
|
||||
} {
|
||||
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
|
||||
const containerName = process.env.AZURE_BLOB_STORAGE_CONTAINER_NAME
|
||||
|
||||
if (!containerName || containerName.trim() === '') {
|
||||
throw new Error('AZURE_BLOB_STORAGE_CONTAINER_NAME env variable is required')
|
||||
}
|
||||
|
||||
let blobServiceClient: BlobServiceClient
|
||||
|
||||
// Authenticate either using connection string or account name and key
|
||||
if (connectionString && connectionString.trim() !== '') {
|
||||
blobServiceClient = BlobServiceClient.fromConnectionString(connectionString)
|
||||
} else if (accountName && accountName.trim() !== '' && accountKey && accountKey.trim() !== '') {
|
||||
const sharedKeyCredential = new StorageSharedKeyCredential(accountName, accountKey)
|
||||
blobServiceClient = new BlobServiceClient(`https://${accountName}.blob.core.windows.net`, sharedKeyCredential)
|
||||
} else {
|
||||
throw new Error(
|
||||
'Azure Blob Storage configuration is missing. Provide AZURE_BLOB_STORAGE_CONNECTION_STRING or AZURE_BLOB_STORAGE_ACCOUNT_NAME + AZURE_BLOB_STORAGE_ACCOUNT_KEY'
|
||||
)
|
||||
}
|
||||
|
||||
const containerClient = blobServiceClient.getContainerClient(containerName)
|
||||
return { containerClient, containerName }
|
||||
}
|
||||
|
||||
getStorageType(): string {
|
||||
return 'azure'
|
||||
}
|
||||
|
||||
getConfig(): any {
|
||||
return {
|
||||
containerName: this.containerName
|
||||
}
|
||||
}
|
||||
|
||||
async addBase64FilesToStorage(fileBase64: string, chatflowid: string, fileNames: string[], orgId: string): Promise<StorageResult> {
|
||||
// Validate chatflowid
|
||||
this.validateChatflowId(chatflowid)
|
||||
this.validatePathSecurity(chatflowid)
|
||||
|
||||
const splitDataURI = fileBase64.split(',')
|
||||
const filename = splitDataURI.pop()?.split(':')[1] ?? ''
|
||||
const bf = Buffer.from(splitDataURI.pop() || '', 'base64')
|
||||
const mime = splitDataURI[0].split(':')[1].split(';')[0]
|
||||
|
||||
const sanitizedFilename = this.sanitizeFilename(filename)
|
||||
const blobName = `${orgId}/${chatflowid}/${sanitizedFilename}`
|
||||
|
||||
const blockBlobClient = this.containerClient.getBlockBlobClient(blobName)
|
||||
await blockBlobClient.upload(bf, bf.length, {
|
||||
blobHTTPHeaders: { blobContentType: mime }
|
||||
})
|
||||
|
||||
fileNames.push(sanitizedFilename)
|
||||
const totalSize = await this.getStorageSize(orgId)
|
||||
|
||||
return { path: 'FILE-STORAGE::' + JSON.stringify(fileNames), totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async addArrayFilesToStorage(
|
||||
mime: string,
|
||||
bf: Buffer,
|
||||
fileName: string,
|
||||
fileNames: string[],
|
||||
...paths: string[]
|
||||
): Promise<StorageResult> {
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
|
||||
let blobName = paths.reduce((acc, cur) => acc + '/' + cur, '') + '/' + sanitizedFilename
|
||||
if (blobName.startsWith('/')) {
|
||||
blobName = blobName.substring(1)
|
||||
}
|
||||
|
||||
const blockBlobClient = this.containerClient.getBlockBlobClient(blobName)
|
||||
await blockBlobClient.upload(bf, bf.length, {
|
||||
blobHTTPHeaders: { blobContentType: mime }
|
||||
})
|
||||
fileNames.push(sanitizedFilename)
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
|
||||
return { path: 'FILE-STORAGE::' + JSON.stringify(fileNames), totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async addSingleFileToStorage(mime: string, bf: Buffer, fileName: string, ...paths: string[]): Promise<StorageResult> {
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
|
||||
let blobName = paths.reduce((acc, cur) => acc + '/' + cur, '') + '/' + sanitizedFilename
|
||||
if (blobName.startsWith('/')) {
|
||||
blobName = blobName.substring(1)
|
||||
}
|
||||
|
||||
const blockBlobClient = this.containerClient.getBlockBlobClient(blobName)
|
||||
await blockBlobClient.upload(bf, bf.length, {
|
||||
blobHTTPHeaders: { blobContentType: mime }
|
||||
})
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
|
||||
return { path: 'FILE-STORAGE::' + sanitizedFilename, totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async getFileFromUpload(filePath: string): Promise<Buffer> {
|
||||
let blobName = filePath
|
||||
// remove the first '/' if it exists
|
||||
if (blobName.startsWith('/')) {
|
||||
blobName = blobName.substring(1)
|
||||
}
|
||||
|
||||
const blockBlobClient = this.containerClient.getBlockBlobClient(blobName)
|
||||
const downloadResponse = await blockBlobClient.downloadToBuffer()
|
||||
return downloadResponse
|
||||
}
|
||||
|
||||
async getFileFromStorage(file: string, ...paths: string[]): Promise<Buffer> {
|
||||
const sanitizedFilename = this.sanitizeFilename(file)
|
||||
|
||||
let blobName = paths.reduce((acc, cur) => acc + '/' + cur, '') + '/' + sanitizedFilename
|
||||
if (blobName.startsWith('/')) {
|
||||
blobName = blobName.substring(1)
|
||||
}
|
||||
|
||||
const blockBlobClient = this.containerClient.getBlockBlobClient(blobName)
|
||||
const buffer = await blockBlobClient.downloadToBuffer()
|
||||
return buffer
|
||||
}
|
||||
|
||||
async streamStorageFile(chatflowId: string, chatId: string, fileName: string, orgId: string): Promise<Buffer | undefined> {
|
||||
// Validate chatflowId and chatId
|
||||
this.validateChatflowId(chatflowId)
|
||||
this.validatePathSecurity(chatflowId, chatId)
|
||||
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
const blobName = `${orgId}/${chatflowId}/${chatId}/${sanitizedFilename}`
|
||||
|
||||
const blockBlobClient = this.containerClient.getBlockBlobClient(blobName)
|
||||
const buffer = await blockBlobClient.downloadToBuffer()
|
||||
return buffer
|
||||
}
|
||||
|
||||
async getFilesListFromStorage(...paths: string[]): Promise<FileInfo[]> {
|
||||
let prefix = paths.reduce((acc, cur) => acc + '/' + cur, '')
|
||||
if (prefix.startsWith('/')) {
|
||||
prefix = prefix.substring(1)
|
||||
}
|
||||
|
||||
const filesList: FileInfo[] = []
|
||||
for await (const blob of this.containerClient.listBlobsFlat({ prefix })) {
|
||||
filesList.push({
|
||||
name: blob.name.split('/').pop() || '',
|
||||
path: blob.name,
|
||||
size: blob.properties.contentLength || 0
|
||||
})
|
||||
}
|
||||
return filesList
|
||||
}
|
||||
|
||||
async removeFilesFromStorage(...paths: string[]): Promise<StorageSizeResult> {
|
||||
let prefix = paths.reduce((acc, cur) => acc + '/' + cur, '')
|
||||
if (prefix.startsWith('/')) {
|
||||
prefix = prefix.substring(1)
|
||||
}
|
||||
|
||||
// Delete all blobs with the prefix
|
||||
for await (const blob of this.containerClient.listBlobsFlat({ prefix })) {
|
||||
await this.containerClient.deleteBlob(blob.name)
|
||||
}
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async removeSpecificFileFromUpload(filePath: string): Promise<void> {
|
||||
let blobName = filePath
|
||||
if (blobName.startsWith('/')) {
|
||||
blobName = blobName.substring(1)
|
||||
}
|
||||
|
||||
await this.containerClient.deleteBlob(blobName)
|
||||
}
|
||||
|
||||
async removeSpecificFileFromStorage(...paths: string[]): Promise<StorageSizeResult> {
|
||||
let blobName = paths.reduce((acc, cur) => acc + '/' + cur, '')
|
||||
if (blobName.startsWith('/')) {
|
||||
blobName = blobName.substring(1)
|
||||
}
|
||||
|
||||
const blockBlobClient = this.containerClient.getBlockBlobClient(blobName)
|
||||
if (await blockBlobClient.exists()) {
|
||||
await blockBlobClient.delete()
|
||||
}
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async removeFolderFromStorage(...paths: string[]): Promise<StorageSizeResult> {
|
||||
let prefix = paths.reduce((acc, cur) => acc + '/' + cur, '')
|
||||
if (prefix.startsWith('/')) {
|
||||
prefix = prefix.substring(1)
|
||||
}
|
||||
|
||||
// Delete all blobs with the prefix
|
||||
for await (const blob of this.containerClient.listBlobsFlat({ prefix })) {
|
||||
await this.containerClient.deleteBlob(blob.name)
|
||||
}
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async getStorageSize(orgId: string): Promise<number> {
|
||||
if (!orgId) return 0
|
||||
|
||||
let totalSize = 0
|
||||
for await (const blob of this.containerClient.listBlobsFlat({ prefix: orgId })) {
|
||||
totalSize += blob.properties.contentLength || 0
|
||||
}
|
||||
|
||||
return totalSize
|
||||
}
|
||||
|
||||
getMulterStorage(): multer.Multer {
|
||||
const connectionString = process.env.AZURE_BLOB_STORAGE_CONNECTION_STRING
|
||||
|
||||
let storageConfig: any = {
|
||||
containerName: this.containerName,
|
||||
blobName: async (_req: any, file: any) => `uploads/${uuidv4()}/${file.originalname}`
|
||||
}
|
||||
|
||||
// Use connection string if available, otherwise use account name/key
|
||||
if (connectionString && connectionString.trim() !== '') {
|
||||
storageConfig.connectionString = connectionString
|
||||
} else {
|
||||
storageConfig.accountName = process.env.AZURE_BLOB_STORAGE_ACCOUNT_NAME
|
||||
storageConfig.accessKey = process.env.AZURE_BLOB_STORAGE_ACCOUNT_KEY
|
||||
}
|
||||
|
||||
const azureStorage = new MulterAzureStorageWithPath(storageConfig)
|
||||
return multer({ storage: azureStorage })
|
||||
}
|
||||
|
||||
getLoggerTransports(logType: 'server' | 'error' | 'requests'): 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
|
||||
|
||||
let baseConfig: any = { containerName: this.containerName }
|
||||
|
||||
// Support both connection string and account name/key authentication
|
||||
if (connectionString && connectionString.trim() !== '') {
|
||||
baseConfig.account = { connectionString }
|
||||
} else {
|
||||
baseConfig.account = { name: accountName, key: accountKey }
|
||||
}
|
||||
|
||||
if (logType === 'server') {
|
||||
return [
|
||||
winstonAzureBlob({
|
||||
...baseConfig,
|
||||
blobName: 'logs/server/server',
|
||||
rotatePeriod: 'YYYY-MM-DD',
|
||||
extension: '.log',
|
||||
level: 'info'
|
||||
})
|
||||
]
|
||||
} else if (logType === 'error') {
|
||||
return [
|
||||
winstonAzureBlob({
|
||||
...baseConfig,
|
||||
blobName: 'logs/error/server-error',
|
||||
rotatePeriod: 'YYYY-MM-DD',
|
||||
extension: '.log',
|
||||
level: 'error'
|
||||
})
|
||||
]
|
||||
} else if (logType === 'requests') {
|
||||
return [
|
||||
winstonAzureBlob({
|
||||
...baseConfig,
|
||||
blobName: 'logs/requests/server-requests',
|
||||
rotatePeriod: 'YYYY-MM-DD',
|
||||
extension: '.log.jsonl',
|
||||
level: 'debug'
|
||||
})
|
||||
]
|
||||
}
|
||||
return []
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
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'
|
||||
|
||||
export abstract class BaseStorageProvider implements IStorageProvider {
|
||||
protected storagePath: string
|
||||
|
||||
constructor() {
|
||||
this.storagePath = this.getStoragePath()
|
||||
}
|
||||
|
||||
abstract getStorageType(): string
|
||||
abstract getConfig(): any
|
||||
abstract addBase64FilesToStorage(fileBase64: string, chatflowid: string, fileNames: string[], orgId: string): Promise<StorageResult>
|
||||
abstract addArrayFilesToStorage(
|
||||
mime: string,
|
||||
bf: Buffer,
|
||||
fileName: string,
|
||||
fileNames: string[],
|
||||
...paths: string[]
|
||||
): Promise<StorageResult>
|
||||
abstract addSingleFileToStorage(mime: string, bf: Buffer, fileName: string, ...paths: string[]): Promise<StorageResult>
|
||||
abstract getFileFromUpload(filePath: string): Promise<Buffer>
|
||||
abstract getFileFromStorage(file: string, ...paths: string[]): Promise<Buffer>
|
||||
abstract getFilesListFromStorage(...paths: string[]): Promise<FileInfo[]>
|
||||
abstract streamStorageFile(
|
||||
chatflowId: string,
|
||||
chatId: string,
|
||||
fileName: string,
|
||||
orgId: string
|
||||
): Promise<fs.ReadStream | Buffer | undefined>
|
||||
abstract removeFilesFromStorage(...paths: string[]): Promise<StorageSizeResult>
|
||||
abstract removeSpecificFileFromUpload(filePath: string): Promise<void>
|
||||
abstract removeSpecificFileFromStorage(...paths: string[]): Promise<StorageSizeResult>
|
||||
abstract removeFolderFromStorage(...paths: string[]): Promise<StorageSizeResult>
|
||||
abstract getStorageSize(orgId: string): Promise<number>
|
||||
abstract getMulterStorage(): any
|
||||
abstract getLoggerTransports(logType: 'server' | 'error' | 'requests', config?: any): any[]
|
||||
|
||||
/**
|
||||
* Shared utility for sanitizing filenames to prevent path traversal and other issues
|
||||
*/
|
||||
protected sanitizeFilename(filename: string): string {
|
||||
if (!filename || isUnsafeFilePath(filename)) {
|
||||
throw new Error('Invalid or unsafe fileName detected')
|
||||
}
|
||||
const sanitizedFilename = sanitize(filename)
|
||||
// Remove leading dots to prevent hidden files or relative path jumps
|
||||
const cleaned = sanitizedFilename.replace(/^\.+/, '')
|
||||
if (!cleaned || cleaned.includes('/') || cleaned.includes('\\')) {
|
||||
throw new Error('Invalid filename after sanitization')
|
||||
}
|
||||
return cleaned
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared utility for getting the base storage path
|
||||
*/
|
||||
protected getStoragePath(): string {
|
||||
const storagePath = process.env.BLOB_STORAGE_PATH
|
||||
? path.join(process.env.BLOB_STORAGE_PATH)
|
||||
: path.join(getUserHome(), '.flowise', 'storage')
|
||||
|
||||
if (!fs.existsSync(storagePath)) {
|
||||
fs.mkdirSync(storagePath, { recursive: true })
|
||||
}
|
||||
return storagePath
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared utility for validating chatflowId format (UUID)
|
||||
*/
|
||||
protected validateChatflowId(chatflowId: string): void {
|
||||
if (!chatflowId || !isValidUUID(chatflowId)) {
|
||||
throw new Error('Invalid chatflowId format - must be a valid UUID')
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared utility for checking path traversal attempts
|
||||
*/
|
||||
protected validatePathSecurity(...paths: string[]): void {
|
||||
for (const p of paths) {
|
||||
if (p && isPathTraversal(p)) {
|
||||
throw new Error('Invalid path characters detected')
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared utility for building a storage path from components
|
||||
*/
|
||||
protected buildPath(...paths: string[]): string {
|
||||
const sanitizedPaths = paths.filter((p) => p && typeof p === 'string').map((p) => this.sanitizeFilename(p))
|
||||
return path.join(this.storagePath, ...sanitizedPaths)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,374 @@
|
||||
import { Storage, Bucket } from '@google-cloud/storage'
|
||||
import { LoggingWinston } from '@google-cloud/logging-winston'
|
||||
import multer from 'multer'
|
||||
import { v4 as uuidv4 } from 'uuid'
|
||||
import { BaseStorageProvider } from './BaseStorageProvider'
|
||||
import { FileInfo, StorageResult, StorageSizeResult } from './IStorageProvider'
|
||||
|
||||
import MulterGoogleCloudStorage from 'multer-cloud-storage'
|
||||
|
||||
export class GCSStorageProvider extends BaseStorageProvider {
|
||||
private bucket: Bucket
|
||||
private bucketName: string
|
||||
private projectId: string | undefined
|
||||
private keyFilename: string | undefined
|
||||
|
||||
constructor() {
|
||||
super()
|
||||
const config = this.initGCSConfig()
|
||||
|
||||
this.bucket = config.bucket
|
||||
this.bucketName = config.bucketName
|
||||
this.projectId = config.projectId
|
||||
this.keyFilename = config.keyFilename
|
||||
}
|
||||
|
||||
private initGCSConfig(): {
|
||||
bucket: Bucket
|
||||
bucketName: string
|
||||
projectId: string | undefined
|
||||
keyFilename: string | undefined
|
||||
} {
|
||||
const keyFilename = process.env.GOOGLE_CLOUD_STORAGE_CREDENTIAL
|
||||
const projectId = process.env.GOOGLE_CLOUD_STORAGE_PROJ_ID
|
||||
const bucketName = process.env.GOOGLE_CLOUD_STORAGE_BUCKET_NAME
|
||||
|
||||
if (!bucketName) {
|
||||
throw new Error('GOOGLE_CLOUD_STORAGE_BUCKET_NAME env variable is required')
|
||||
}
|
||||
|
||||
const storageConfig = {
|
||||
...(keyFilename ? { keyFilename } : {}),
|
||||
...(projectId ? { projectId } : {})
|
||||
}
|
||||
|
||||
const storage = new Storage(storageConfig)
|
||||
const bucket = storage.bucket(bucketName)
|
||||
return { bucket, bucketName, projectId, keyFilename }
|
||||
}
|
||||
|
||||
getStorageType(): string {
|
||||
return 'gcs'
|
||||
}
|
||||
|
||||
getConfig(): any {
|
||||
return {
|
||||
bucketName: this.bucketName,
|
||||
projectId: this.projectId
|
||||
}
|
||||
}
|
||||
|
||||
private normalizePath(p: string): string {
|
||||
return p.replace(/\\/g, '/')
|
||||
}
|
||||
|
||||
async addBase64FilesToStorage(fileBase64: string, chatflowid: string, fileNames: string[], orgId: string): Promise<StorageResult> {
|
||||
// Validate chatflowid
|
||||
this.validateChatflowId(chatflowid)
|
||||
this.validatePathSecurity(chatflowid)
|
||||
|
||||
const splitDataURI = fileBase64.split(',')
|
||||
const filename = splitDataURI.pop()?.split(':')[1] ?? ''
|
||||
const bf = Buffer.from(splitDataURI.pop() || '', 'base64')
|
||||
const mime = splitDataURI[0].split(':')[1].split(';')[0]
|
||||
|
||||
const sanitizedFilename = this.sanitizeFilename(filename)
|
||||
const normalizedChatflowid = this.normalizePath(chatflowid)
|
||||
const normalizedFilename = this.normalizePath(sanitizedFilename)
|
||||
const filePath = `${orgId}/${normalizedChatflowid}/${normalizedFilename}`
|
||||
|
||||
const file = this.bucket.file(filePath)
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
file.createWriteStream({ contentType: mime, metadata: { contentEncoding: 'base64' } })
|
||||
.on('error', (err) => reject(err))
|
||||
.on('finish', () => resolve())
|
||||
.end(bf)
|
||||
})
|
||||
|
||||
fileNames.push(sanitizedFilename)
|
||||
const totalSize = await this.getStorageSize(orgId)
|
||||
|
||||
return { path: 'FILE-STORAGE::' + JSON.stringify(fileNames), totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async addArrayFilesToStorage(
|
||||
mime: string,
|
||||
bf: Buffer,
|
||||
fileName: string,
|
||||
fileNames: string[],
|
||||
...paths: string[]
|
||||
): Promise<StorageResult> {
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
const normalizedPaths = paths.map((p) => this.normalizePath(p))
|
||||
const normalizedFilename = this.normalizePath(sanitizedFilename)
|
||||
const filePath = [...normalizedPaths, normalizedFilename].join('/')
|
||||
|
||||
const file = this.bucket.file(filePath)
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
file.createWriteStream()
|
||||
.on('error', (err) => reject(err))
|
||||
.on('finish', () => resolve())
|
||||
.end(bf)
|
||||
})
|
||||
|
||||
fileNames.push(sanitizedFilename)
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
|
||||
return { path: 'FILE-STORAGE::' + JSON.stringify(fileNames), totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async addSingleFileToStorage(mime: string, bf: Buffer, fileName: string, ...paths: string[]): Promise<StorageResult> {
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
const normalizedPaths = paths.map((p) => this.normalizePath(p))
|
||||
const normalizedFilename = this.normalizePath(sanitizedFilename)
|
||||
const filePath = [...normalizedPaths, normalizedFilename].join('/')
|
||||
|
||||
const file = this.bucket.file(filePath)
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
file.createWriteStream({ contentType: mime, metadata: { contentEncoding: 'base64' } })
|
||||
.on('error', (err) => reject(err))
|
||||
.on('finish', () => resolve())
|
||||
.end(bf)
|
||||
})
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { path: 'FILE-STORAGE::' + sanitizedFilename, totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async getFileFromUpload(filePath: string): Promise<Buffer> {
|
||||
const file = this.bucket.file(filePath)
|
||||
const [buffer] = await file.download()
|
||||
return buffer
|
||||
}
|
||||
|
||||
async getFileFromStorage(file: string, ...paths: string[]): Promise<Buffer> {
|
||||
const sanitizedFilename = this.sanitizeFilename(file)
|
||||
const normalizedPaths = paths.map((p) => this.normalizePath(p))
|
||||
const normalizedFilename = this.normalizePath(sanitizedFilename)
|
||||
const filePath = [...normalizedPaths, normalizedFilename].join('/')
|
||||
|
||||
try {
|
||||
const gcsFile = this.bucket.file(filePath)
|
||||
const [buffer] = await gcsFile.download()
|
||||
return buffer
|
||||
} catch (error) {
|
||||
// Fallback: Check if file exists without the first path element (likely orgId)
|
||||
if (normalizedPaths.length > 1) {
|
||||
const fallbackPaths = normalizedPaths.slice(1)
|
||||
const fallbackPath = [...fallbackPaths, normalizedFilename].join('/')
|
||||
|
||||
try {
|
||||
const fallbackFile = this.bucket.file(fallbackPath)
|
||||
const [buffer] = await fallbackFile.download()
|
||||
|
||||
// Move to correct location with orgId
|
||||
const gcsFile = this.bucket.file(filePath)
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
gcsFile
|
||||
.createWriteStream()
|
||||
.on('error', (err) => reject(err))
|
||||
.on('finish', () => resolve())
|
||||
.end(buffer)
|
||||
})
|
||||
|
||||
// Delete the old file
|
||||
await fallbackFile.delete()
|
||||
|
||||
// Check if the directory is empty and delete recursively if needed
|
||||
if (fallbackPaths.length > 0) {
|
||||
await this.cleanEmptyGCSFolders(fallbackPaths[0])
|
||||
}
|
||||
|
||||
return buffer
|
||||
} catch (fallbackError) {
|
||||
throw error
|
||||
}
|
||||
} else {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async streamStorageFile(chatflowId: string, chatId: string, fileName: string, orgId: string): Promise<Buffer | undefined> {
|
||||
// Validate chatflowId and chatId
|
||||
this.validateChatflowId(chatflowId)
|
||||
this.validatePathSecurity(chatflowId, chatId)
|
||||
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
const normalizedChatflowId = this.normalizePath(chatflowId)
|
||||
const normalizedChatId = this.normalizePath(chatId)
|
||||
const normalizedFilename = this.normalizePath(sanitizedFilename)
|
||||
const filePath = `${orgId}/${normalizedChatflowId}/${normalizedChatId}/${normalizedFilename}`
|
||||
|
||||
try {
|
||||
const [buffer] = await this.bucket.file(filePath).download()
|
||||
return buffer
|
||||
} catch (error) {
|
||||
// Fallback: Check if file exists without orgId
|
||||
const fallbackPath = `${normalizedChatflowId}/${normalizedChatId}/${normalizedFilename}`
|
||||
try {
|
||||
const fallbackFile = this.bucket.file(fallbackPath)
|
||||
const [buffer] = await fallbackFile.download()
|
||||
|
||||
// If found, copy to correct location with orgId
|
||||
if (buffer) {
|
||||
const file = this.bucket.file(filePath)
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
file.createWriteStream()
|
||||
.on('error', (err) => reject(err))
|
||||
.on('finish', () => resolve())
|
||||
.end(buffer)
|
||||
})
|
||||
|
||||
// Delete the old file
|
||||
await fallbackFile.delete()
|
||||
|
||||
// Check if the directory is empty and delete recursively if needed
|
||||
await this.cleanEmptyGCSFolders(normalizedChatflowId)
|
||||
|
||||
return buffer
|
||||
}
|
||||
} catch (fallbackError) {
|
||||
throw new Error(`File ${fileName} not found`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async getFilesListFromStorage(...paths: string[]): Promise<FileInfo[]> {
|
||||
const normalizedPaths = paths.map((p) => this.normalizePath(p))
|
||||
const prefix = normalizedPaths.join('/')
|
||||
|
||||
const [files] = await this.bucket.getFiles({ prefix })
|
||||
|
||||
return files.map((file) => ({
|
||||
name: file.name.split('/').pop() || '',
|
||||
path: file.name,
|
||||
size: typeof file.metadata.size === 'string' ? parseInt(file.metadata.size, 10) || 0 : file.metadata.size || 0
|
||||
}))
|
||||
}
|
||||
|
||||
async removeFilesFromStorage(...paths: string[]): Promise<StorageSizeResult> {
|
||||
const normalizedPath = paths.map((p) => this.normalizePath(p)).join('/')
|
||||
await this.bucket.deleteFiles({ prefix: `${normalizedPath}/` })
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async removeSpecificFileFromUpload(filePath: string): Promise<void> {
|
||||
await this.bucket.file(filePath).delete()
|
||||
}
|
||||
|
||||
async removeSpecificFileFromStorage(...paths: string[]): Promise<StorageSizeResult> {
|
||||
const fileName = paths.pop()
|
||||
if (fileName) {
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
paths.push(sanitizedFilename)
|
||||
}
|
||||
const normalizedPath = paths.map((p) => this.normalizePath(p)).join('/')
|
||||
await this.bucket.file(normalizedPath).delete()
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async removeFolderFromStorage(...paths: string[]): Promise<StorageSizeResult> {
|
||||
const normalizedPath = paths.map((p) => this.normalizePath(p)).join('/')
|
||||
await this.bucket.deleteFiles({ prefix: `${normalizedPath}/` })
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
private async cleanEmptyGCSFolders(prefix: string): Promise<void> {
|
||||
try {
|
||||
if (!prefix) return
|
||||
|
||||
const [files] = await this.bucket.getFiles({
|
||||
prefix: prefix + '/',
|
||||
delimiter: '/'
|
||||
})
|
||||
|
||||
if (files.length === 0) {
|
||||
try {
|
||||
await this.bucket.file(prefix + '/').delete()
|
||||
} catch (err) {
|
||||
// Folder marker might not exist, ignore
|
||||
}
|
||||
|
||||
const parentPrefix = prefix.substring(0, prefix.lastIndexOf('/'))
|
||||
if (parentPrefix) {
|
||||
await this.cleanEmptyGCSFolders(parentPrefix)
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Error cleaning empty GCS folders:', error)
|
||||
}
|
||||
}
|
||||
|
||||
async getStorageSize(orgId: string): Promise<number> {
|
||||
if (!orgId) return 0
|
||||
|
||||
const [files] = await this.bucket.getFiles({ prefix: orgId })
|
||||
let totalSize = 0
|
||||
|
||||
for (const file of files) {
|
||||
const size = file.metadata.size
|
||||
if (typeof size === 'string') {
|
||||
totalSize += parseInt(size, 10) || 0
|
||||
} else if (typeof size === 'number') {
|
||||
totalSize += size
|
||||
}
|
||||
}
|
||||
|
||||
return totalSize
|
||||
}
|
||||
|
||||
getMulterStorage(): multer.Multer {
|
||||
return multer({
|
||||
storage: new MulterGoogleCloudStorage({
|
||||
projectId: this.projectId,
|
||||
bucket: this.bucketName,
|
||||
keyFilename: this.keyFilename,
|
||||
uniformBucketLevelAccess: Boolean(process.env.GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS) ?? true,
|
||||
destination: `uploads/${uuidv4()}`
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
getLoggerTransports(logType: 'server' | 'error' | 'requests'): any[] {
|
||||
const gcsConfig = {
|
||||
projectId: this.projectId,
|
||||
keyFilename: this.keyFilename,
|
||||
defaultCallback: (err: any) => {
|
||||
if (err) {
|
||||
console.error('Error logging to GCS: ' + err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (logType === 'server') {
|
||||
return [
|
||||
new LoggingWinston({
|
||||
...gcsConfig,
|
||||
logName: 'server'
|
||||
})
|
||||
]
|
||||
} else if (logType === 'error') {
|
||||
return [
|
||||
new LoggingWinston({
|
||||
...gcsConfig,
|
||||
logName: 'error'
|
||||
})
|
||||
]
|
||||
} else if (logType === 'requests') {
|
||||
return [
|
||||
new LoggingWinston({
|
||||
...gcsConfig,
|
||||
logName: 'requests'
|
||||
})
|
||||
]
|
||||
}
|
||||
return []
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
import fs from 'node:fs'
|
||||
|
||||
export interface FileInfo {
|
||||
name: string
|
||||
path: string
|
||||
size: number
|
||||
}
|
||||
|
||||
export interface StorageResult {
|
||||
path: string
|
||||
totalSize: number
|
||||
}
|
||||
|
||||
export interface StorageSizeResult {
|
||||
totalSize: number
|
||||
}
|
||||
|
||||
export interface IStorageProvider {
|
||||
/**
|
||||
* Get the storage type (e.g., 'local', 's3', 'gcs', 'azure')
|
||||
*/
|
||||
getStorageType(): string
|
||||
|
||||
/**
|
||||
* Get the provider configuration
|
||||
*/
|
||||
getConfig(): any
|
||||
|
||||
/**
|
||||
* Add files provided as base64 strings to storage
|
||||
*/
|
||||
addBase64FilesToStorage(fileBase64: string, chatflowid: string, fileNames: string[], orgId: string): Promise<StorageResult>
|
||||
|
||||
/**
|
||||
* Add file buffer to storage as part of an array
|
||||
*/
|
||||
addArrayFilesToStorage(mime: string, bf: Buffer, fileName: string, fileNames: string[], ...paths: string[]): Promise<StorageResult>
|
||||
|
||||
/**
|
||||
* Add a single file buffer to storage
|
||||
*/
|
||||
addSingleFileToStorage(mime: string, bf: Buffer, fileName: string, ...paths: string[]): Promise<StorageResult>
|
||||
|
||||
/**
|
||||
* Retrieve a file from the upload directory
|
||||
*/
|
||||
getFileFromUpload(filePath: string): Promise<Buffer>
|
||||
|
||||
/**
|
||||
* Retrieve a file from generic storage
|
||||
*/
|
||||
getFileFromStorage(file: string, ...paths: string[]): Promise<Buffer>
|
||||
|
||||
/**
|
||||
* List files in a storage path
|
||||
*/
|
||||
getFilesListFromStorage(...paths: string[]): Promise<FileInfo[]>
|
||||
|
||||
/**
|
||||
* Stream a file from storage
|
||||
*/
|
||||
streamStorageFile(chatflowId: string, chatId: string, fileName: string, orgId: string): Promise<fs.ReadStream | Buffer | undefined>
|
||||
|
||||
/**
|
||||
* Remove multiple files from storage
|
||||
*/
|
||||
removeFilesFromStorage(...paths: string[]): Promise<StorageSizeResult>
|
||||
|
||||
/**
|
||||
* Remove a specific file from the upload directory
|
||||
*/
|
||||
removeSpecificFileFromUpload(filePath: string): Promise<void>
|
||||
|
||||
/**
|
||||
* Remove a specific file from storage
|
||||
*/
|
||||
removeSpecificFileFromStorage(...paths: string[]): Promise<StorageSizeResult>
|
||||
|
||||
/**
|
||||
* Remove a complete folder and its contents from storage
|
||||
*/
|
||||
removeFolderFromStorage(...paths: string[]): Promise<StorageSizeResult>
|
||||
|
||||
/**
|
||||
* Calculate the total storage size for an organization
|
||||
*/
|
||||
getStorageSize(orgId: string): Promise<number>
|
||||
|
||||
/**
|
||||
* Get the Multer storage engine for this provider
|
||||
*/
|
||||
getMulterStorage(): any
|
||||
|
||||
/**
|
||||
* Get the Winston logger transports for this provider
|
||||
*/
|
||||
getLoggerTransports(logType: 'server' | 'error' | 'requests', config?: any): any[]
|
||||
}
|
||||
@@ -0,0 +1,381 @@
|
||||
import path from 'node:path'
|
||||
import fs from 'fs'
|
||||
import multer from 'multer'
|
||||
import DailyRotateFile from 'winston-daily-rotate-file'
|
||||
import { transports } from 'winston'
|
||||
import { BaseStorageProvider } from './BaseStorageProvider'
|
||||
import { FileInfo, StorageResult, StorageSizeResult } from './IStorageProvider'
|
||||
|
||||
export class LocalStorageProvider extends BaseStorageProvider {
|
||||
constructor() {
|
||||
super()
|
||||
}
|
||||
|
||||
getStorageType(): string {
|
||||
return 'local'
|
||||
}
|
||||
|
||||
getConfig(): any {
|
||||
return {
|
||||
storagePath: this.storagePath
|
||||
}
|
||||
}
|
||||
|
||||
async addBase64FilesToStorage(fileBase64: string, chatflowid: string, fileNames: string[], orgId: string): Promise<StorageResult> {
|
||||
// Validate chatflowid
|
||||
this.validateChatflowId(chatflowid)
|
||||
this.validatePathSecurity(chatflowid)
|
||||
|
||||
const dir = this.buildPath(orgId, chatflowid)
|
||||
if (!fs.existsSync(dir)) {
|
||||
fs.mkdirSync(dir, { recursive: true })
|
||||
}
|
||||
|
||||
const splitDataURI = fileBase64.split(',')
|
||||
const filename = splitDataURI.pop()?.split(':')[1] ?? ''
|
||||
const bf = Buffer.from(splitDataURI.pop() || '', 'base64')
|
||||
const sanitizedFilename = this.sanitizeFilename(filename)
|
||||
|
||||
const filePath = path.join(dir, sanitizedFilename)
|
||||
fs.writeFileSync(filePath, bf)
|
||||
fileNames.push(sanitizedFilename)
|
||||
|
||||
const totalSize = await this.getStorageSize(orgId)
|
||||
return { path: 'FILE-STORAGE::' + JSON.stringify(fileNames), totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async addArrayFilesToStorage(
|
||||
mime: string,
|
||||
bf: Buffer,
|
||||
fileName: string,
|
||||
fileNames: string[],
|
||||
...paths: string[]
|
||||
): Promise<StorageResult> {
|
||||
const dir = this.buildPath(...paths)
|
||||
if (!fs.existsSync(dir)) {
|
||||
fs.mkdirSync(dir, { recursive: true })
|
||||
}
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
const filePath = path.join(dir, sanitizedFilename)
|
||||
fs.writeFileSync(filePath, bf)
|
||||
fileNames.push(sanitizedFilename)
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { path: 'FILE-STORAGE::' + JSON.stringify(fileNames), totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async addSingleFileToStorage(mime: string, bf: Buffer, fileName: string, ...paths: string[]): Promise<StorageResult> {
|
||||
const dir = this.buildPath(...paths)
|
||||
if (!fs.existsSync(dir)) {
|
||||
fs.mkdirSync(dir, { recursive: true })
|
||||
}
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
const filePath = path.join(dir, sanitizedFilename)
|
||||
fs.writeFileSync(filePath, bf)
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { path: 'FILE-STORAGE::' + sanitizedFilename, totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async getFileFromUpload(filePath: string): Promise<Buffer> {
|
||||
return fs.readFileSync(filePath)
|
||||
}
|
||||
|
||||
async getFileFromStorage(file: string, ...paths: string[]): Promise<Buffer> {
|
||||
const sanitizedFilename = this.sanitizeFilename(file)
|
||||
const fileInStorage = this.buildPath(...paths.map((p) => this.sanitizeFilename(p)), sanitizedFilename)
|
||||
try {
|
||||
return fs.readFileSync(fileInStorage)
|
||||
} catch (error) {
|
||||
// Fallback: Check if file exists without the first path element (likely orgId)
|
||||
if (paths.length > 1) {
|
||||
const fallbackPaths = paths.slice(1)
|
||||
const fallbackPath = this.buildPath(...fallbackPaths.map((p) => this.sanitizeFilename(p)), sanitizedFilename)
|
||||
|
||||
if (fs.existsSync(fallbackPath)) {
|
||||
// Create directory if it doesn't exist
|
||||
const targetPath = fileInStorage
|
||||
const dir = path.dirname(targetPath)
|
||||
if (!fs.existsSync(dir)) {
|
||||
fs.mkdirSync(dir, { recursive: true })
|
||||
}
|
||||
|
||||
// Copy file to correct location with orgId
|
||||
fs.copyFileSync(fallbackPath, targetPath)
|
||||
|
||||
// Delete the old file
|
||||
fs.unlinkSync(fallbackPath)
|
||||
|
||||
// Clean up empty directories recursively
|
||||
if (fallbackPaths.length > 0) {
|
||||
this.cleanEmptyLocalFolders(this.buildPath(...fallbackPaths.map((p) => this.sanitizeFilename(p)).slice(0, -1)))
|
||||
}
|
||||
|
||||
return fs.readFileSync(targetPath)
|
||||
}
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
async streamStorageFile(
|
||||
chatflowId: string,
|
||||
chatId: string,
|
||||
fileName: string,
|
||||
orgId: string
|
||||
): Promise<fs.ReadStream | Buffer | undefined> {
|
||||
// Validate chatflowId and chatId
|
||||
this.validateChatflowId(chatflowId)
|
||||
this.validatePathSecurity(chatflowId, chatId)
|
||||
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
const filePath = this.buildPath(orgId, chatflowId, chatId, sanitizedFilename)
|
||||
|
||||
//raise error if file path is not absolute
|
||||
if (!path.isAbsolute(filePath)) throw new Error(`Invalid file path`)
|
||||
//raise error if file path contains '..'
|
||||
if (filePath.includes('..')) throw new Error(`Invalid file path`)
|
||||
//only return from the storage folder
|
||||
if (!filePath.startsWith(this.storagePath)) throw new Error(`Invalid file path`)
|
||||
|
||||
if (fs.existsSync(filePath)) {
|
||||
return fs.createReadStream(filePath)
|
||||
} else {
|
||||
// Fallback: Check if file exists without orgId
|
||||
const fallbackPath = this.buildPath(chatflowId, chatId, sanitizedFilename)
|
||||
|
||||
if (fs.existsSync(fallbackPath)) {
|
||||
// Create directory if it doesn't exist
|
||||
const dir = path.dirname(filePath)
|
||||
if (!fs.existsSync(dir)) {
|
||||
fs.mkdirSync(dir, { recursive: true })
|
||||
}
|
||||
|
||||
// Copy file to correct location with orgId
|
||||
fs.copyFileSync(fallbackPath, filePath)
|
||||
|
||||
// Delete the old file
|
||||
fs.unlinkSync(fallbackPath)
|
||||
|
||||
// Clean up empty directories recursively
|
||||
this.cleanEmptyLocalFolders(path.dirname(fallbackPath))
|
||||
|
||||
return fs.createReadStream(filePath)
|
||||
} else {
|
||||
throw new Error(`File ${fileName} not found`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async getFilesListFromStorage(...paths: string[]): Promise<FileInfo[]> {
|
||||
const directory = this.buildPath(...paths)
|
||||
return this.getFilePaths(directory)
|
||||
}
|
||||
|
||||
private getFilePaths(dir: string): FileInfo[] {
|
||||
let results: FileInfo[] = []
|
||||
|
||||
const readDirectory = (directory: string) => {
|
||||
try {
|
||||
if (!fs.existsSync(directory)) {
|
||||
console.warn(`Directory does not exist: ${directory}`)
|
||||
return
|
||||
}
|
||||
|
||||
const list = fs.readdirSync(directory)
|
||||
list.forEach((file) => {
|
||||
const filePath = path.join(directory, file)
|
||||
try {
|
||||
const stat = fs.statSync(filePath)
|
||||
if (stat && stat.isDirectory()) {
|
||||
readDirectory(filePath)
|
||||
} else {
|
||||
results.push({ name: file, path: filePath, size: stat.size })
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`Error processing file ${filePath}:`, error)
|
||||
}
|
||||
})
|
||||
} catch (error) {
|
||||
console.error(`Error reading directory ${directory}:`, error)
|
||||
}
|
||||
}
|
||||
|
||||
readDirectory(dir)
|
||||
return results
|
||||
}
|
||||
|
||||
private cleanEmptyLocalFolders(dirPath: string) {
|
||||
try {
|
||||
// Stop if we reach the storage root
|
||||
if (dirPath === this.storagePath) return
|
||||
|
||||
// Check if directory exists
|
||||
if (!fs.existsSync(dirPath)) return
|
||||
|
||||
// Read directory contents
|
||||
const files = fs.readdirSync(dirPath)
|
||||
|
||||
// If directory is empty, delete it and check parent
|
||||
if (files.length === 0) {
|
||||
fs.rmdirSync(dirPath)
|
||||
// Recursively check parent directory
|
||||
this.cleanEmptyLocalFolders(path.dirname(dirPath))
|
||||
}
|
||||
} catch (error) {
|
||||
// Ignore errors during cleanup
|
||||
console.error('Error cleaning empty folders:', error)
|
||||
}
|
||||
}
|
||||
|
||||
async removeFilesFromStorage(...paths: string[]): Promise<StorageSizeResult> {
|
||||
const directory = this.buildPath(...paths.map((p) => this.sanitizeFilename(p)))
|
||||
await this.deleteLocalFolderRecursive(directory)
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async removeSpecificFileFromUpload(filePath: string): Promise<void> {
|
||||
if (fs.existsSync(filePath)) {
|
||||
fs.unlinkSync(filePath)
|
||||
}
|
||||
}
|
||||
|
||||
async removeSpecificFileFromStorage(...paths: string[]): Promise<StorageSizeResult> {
|
||||
const fileName = paths.pop()
|
||||
if (fileName) {
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
paths.push(sanitizedFilename)
|
||||
}
|
||||
const file = this.buildPath(...paths.map((p) => this.sanitizeFilename(p)))
|
||||
|
||||
// check if file exists, if not skip delete
|
||||
const stat = fs.statSync(file, { throwIfNoEntry: false })
|
||||
if (stat && stat.isFile()) {
|
||||
fs.unlinkSync(file)
|
||||
}
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async removeFolderFromStorage(...paths: string[]): Promise<StorageSizeResult> {
|
||||
const directory = this.buildPath(...paths.map((p) => this.sanitizeFilename(p)))
|
||||
await this.deleteLocalFolderRecursive(directory, true)
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
private async deleteLocalFolderRecursive(directory: string, deleteParentChatflowFolder?: boolean): Promise<void> {
|
||||
try {
|
||||
// Check if the path exists
|
||||
await fs.promises.access(directory)
|
||||
|
||||
if (deleteParentChatflowFolder) {
|
||||
await fs.promises.rm(directory, { recursive: true })
|
||||
return
|
||||
}
|
||||
|
||||
// Get stats of the path to determine if it's a file or directory
|
||||
const stats = await fs.promises.stat(directory)
|
||||
|
||||
if (stats.isDirectory()) {
|
||||
// Read all directory contents
|
||||
const files = await fs.promises.readdir(directory)
|
||||
|
||||
// Recursively delete all contents
|
||||
for (const file of files) {
|
||||
const currentPath = path.join(directory, file)
|
||||
await this.deleteLocalFolderRecursive(currentPath)
|
||||
}
|
||||
|
||||
// Delete the directory itself after emptying it
|
||||
await fs.promises.rm(directory, { recursive: true })
|
||||
} else {
|
||||
// If it's a file, delete it directly
|
||||
await fs.promises.unlink(directory)
|
||||
}
|
||||
} catch (error) {
|
||||
// Error handling - ignore if file/directory doesn't exist
|
||||
}
|
||||
}
|
||||
|
||||
async getStorageSize(orgId: string): Promise<number> {
|
||||
if (!orgId) return 0
|
||||
const directory = this.buildPath(orgId)
|
||||
return this.dirSize(directory)
|
||||
}
|
||||
|
||||
private async dirSize(directoryPath: string): Promise<number> {
|
||||
let totalSize = 0
|
||||
|
||||
const calculateSize = async (itemPath: string) => {
|
||||
try {
|
||||
const stats = await fs.promises.stat(itemPath)
|
||||
|
||||
if (stats.isFile()) {
|
||||
totalSize += stats.size
|
||||
} else if (stats.isDirectory()) {
|
||||
const files = await fs.promises.readdir(itemPath)
|
||||
for (const file of files) {
|
||||
await calculateSize(path.join(itemPath, file))
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// Ignore missing files/dirs during calculation
|
||||
}
|
||||
}
|
||||
|
||||
await calculateSize(directoryPath)
|
||||
return totalSize
|
||||
}
|
||||
|
||||
getMulterStorage(): multer.Multer {
|
||||
const uploadPath = this.getUploadPath()
|
||||
if (!fs.existsSync(uploadPath)) {
|
||||
fs.mkdirSync(uploadPath, { recursive: true })
|
||||
}
|
||||
return multer({ dest: uploadPath })
|
||||
}
|
||||
|
||||
private getUploadPath(): string {
|
||||
return process.env.BLOB_STORAGE_PATH
|
||||
? path.join(process.env.BLOB_STORAGE_PATH, 'uploads')
|
||||
: path.join(this.getUserHome(), '.flowise', 'uploads')
|
||||
}
|
||||
|
||||
private getUserHome(): string {
|
||||
return process.env.HOME || process.env.USERPROFILE || process.env.HOMEPATH || ''
|
||||
}
|
||||
|
||||
getLoggerTransports(logType: 'server' | 'error' | 'requests', config?: any): any[] {
|
||||
const logDir = config?.logging?.dir || path.join(this.getUserHome(), '.flowise', 'logs')
|
||||
|
||||
if (!fs.existsSync(logDir)) {
|
||||
fs.mkdirSync(logDir, { recursive: true })
|
||||
}
|
||||
|
||||
if (logType === 'server') {
|
||||
return [
|
||||
new DailyRotateFile({
|
||||
filename: path.join(logDir, config?.logging?.server?.filename ?? 'server-%DATE%.log'),
|
||||
datePattern: 'YYYY-MM-DD-HH',
|
||||
maxSize: '20m',
|
||||
level: config?.logging?.server?.level ?? 'info'
|
||||
})
|
||||
]
|
||||
} else if (logType === 'requests') {
|
||||
return [
|
||||
new transports.File({
|
||||
filename: path.join(logDir, config?.logging?.express?.filename ?? 'server-requests.log.jsonl'),
|
||||
level: config?.logging?.express?.level ?? 'debug'
|
||||
})
|
||||
]
|
||||
}
|
||||
|
||||
// For 'error' type, return empty array (handled by exceptionHandlers in logger.ts)
|
||||
return []
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,538 @@
|
||||
import {
|
||||
DeleteObjectsCommand,
|
||||
GetObjectCommand,
|
||||
ListObjectsCommand,
|
||||
ListObjectsV2Command,
|
||||
PutObjectCommand,
|
||||
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 { v4 as uuidv4 } from 'uuid'
|
||||
import { BaseStorageProvider } from './BaseStorageProvider'
|
||||
import { FileInfo, StorageResult, StorageSizeResult } from './IStorageProvider'
|
||||
|
||||
const { S3StreamLogger } = require('s3-streamlogger')
|
||||
|
||||
export class S3StorageProvider extends BaseStorageProvider {
|
||||
private s3Client: S3Client
|
||||
private bucket: string
|
||||
private s3Config: S3ClientConfig
|
||||
|
||||
constructor() {
|
||||
super()
|
||||
const config = this.initS3Config()
|
||||
this.s3Client = config.s3Client
|
||||
this.bucket = config.bucket
|
||||
this.s3Config = config.s3Config
|
||||
}
|
||||
|
||||
private initS3Config(): { s3Client: S3Client; bucket: string; s3Config: S3ClientConfig } {
|
||||
const accessKeyId = process.env.S3_STORAGE_ACCESS_KEY_ID
|
||||
const secretAccessKey = process.env.S3_STORAGE_SECRET_ACCESS_KEY
|
||||
const region = process.env.S3_STORAGE_REGION
|
||||
const bucket = process.env.S3_STORAGE_BUCKET_NAME
|
||||
const customURL = process.env.S3_ENDPOINT_URL
|
||||
const forcePathStyle = process.env.S3_FORCE_PATH_STYLE === 'true'
|
||||
|
||||
if (!region || region.trim() === '' || !bucket || bucket.trim() === '') {
|
||||
throw new Error('S3 storage configuration is missing')
|
||||
}
|
||||
|
||||
const s3Config: S3ClientConfig = {
|
||||
region: region,
|
||||
forcePathStyle: forcePathStyle
|
||||
}
|
||||
|
||||
// Only include endpoint if customURL is not empty
|
||||
if (customURL && customURL.trim() !== '') {
|
||||
s3Config.endpoint = customURL
|
||||
}
|
||||
|
||||
if (accessKeyId && accessKeyId.trim() !== '' && secretAccessKey && secretAccessKey.trim() !== '') {
|
||||
s3Config.credentials = {
|
||||
accessKeyId: accessKeyId,
|
||||
secretAccessKey: secretAccessKey
|
||||
}
|
||||
}
|
||||
|
||||
const s3Client = new S3Client(s3Config)
|
||||
return { s3Client, bucket, s3Config }
|
||||
}
|
||||
|
||||
getStorageType(): string {
|
||||
return 's3'
|
||||
}
|
||||
|
||||
getConfig(): any {
|
||||
return {
|
||||
bucket: this.bucket,
|
||||
region: process.env.S3_STORAGE_REGION,
|
||||
endpoint: process.env.S3_ENDPOINT_URL
|
||||
}
|
||||
}
|
||||
|
||||
async addBase64FilesToStorage(fileBase64: string, chatflowid: string, fileNames: string[], orgId: string): Promise<StorageResult> {
|
||||
// Validate chatflowid
|
||||
this.validateChatflowId(chatflowid)
|
||||
this.validatePathSecurity(chatflowid)
|
||||
|
||||
const splitDataURI = fileBase64.split(',')
|
||||
const filename = splitDataURI.pop()?.split(':')[1] ?? ''
|
||||
const bf = Buffer.from(splitDataURI.pop() || '', 'base64')
|
||||
const mime = splitDataURI[0].split(':')[1].split(';')[0]
|
||||
|
||||
const sanitizedFilename = this.sanitizeFilename(filename)
|
||||
const Key = orgId + '/' + chatflowid + '/' + sanitizedFilename
|
||||
|
||||
const putObjCmd = new PutObjectCommand({
|
||||
Bucket: this.bucket,
|
||||
Key,
|
||||
ContentEncoding: 'base64',
|
||||
ContentType: mime,
|
||||
Body: bf
|
||||
})
|
||||
await this.s3Client.send(putObjCmd)
|
||||
|
||||
fileNames.push(sanitizedFilename)
|
||||
const totalSize = await this.getStorageSize(orgId)
|
||||
|
||||
return { path: 'FILE-STORAGE::' + JSON.stringify(fileNames), totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async addArrayFilesToStorage(
|
||||
mime: string,
|
||||
bf: Buffer,
|
||||
fileName: string,
|
||||
fileNames: string[],
|
||||
...paths: string[]
|
||||
): Promise<StorageResult> {
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
|
||||
let Key = paths.reduce((acc, cur) => acc + '/' + cur, '') + '/' + sanitizedFilename
|
||||
if (Key.startsWith('/')) {
|
||||
Key = Key.substring(1)
|
||||
}
|
||||
|
||||
const putObjCmd = new PutObjectCommand({
|
||||
Bucket: this.bucket,
|
||||
Key,
|
||||
ContentEncoding: 'base64',
|
||||
ContentType: mime,
|
||||
Body: bf
|
||||
})
|
||||
await this.s3Client.send(putObjCmd)
|
||||
fileNames.push(sanitizedFilename)
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { path: 'FILE-STORAGE::' + JSON.stringify(fileNames), totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async addSingleFileToStorage(mime: string, bf: Buffer, fileName: string, ...paths: string[]): Promise<StorageResult> {
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
|
||||
let Key = paths.reduce((acc, cur) => acc + '/' + cur, '') + '/' + sanitizedFilename
|
||||
if (Key.startsWith('/')) {
|
||||
Key = Key.substring(1)
|
||||
}
|
||||
|
||||
const putObjCmd = new PutObjectCommand({
|
||||
Bucket: this.bucket,
|
||||
Key,
|
||||
ContentEncoding: 'base64',
|
||||
ContentType: mime,
|
||||
Body: bf
|
||||
})
|
||||
await this.s3Client.send(putObjCmd)
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { path: 'FILE-STORAGE::' + sanitizedFilename, totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async getFileFromUpload(filePath: string): Promise<Buffer> {
|
||||
// For S3, the filePath is the S3 key
|
||||
let Key = filePath
|
||||
if (Key.startsWith('/')) {
|
||||
Key = Key.substring(1)
|
||||
}
|
||||
|
||||
const getParams = {
|
||||
Bucket: this.bucket,
|
||||
Key
|
||||
}
|
||||
|
||||
const response = await this.s3Client.send(new GetObjectCommand(getParams))
|
||||
const body = response.Body
|
||||
if (body instanceof Readable) {
|
||||
const streamToString = await body.transformToString('base64')
|
||||
if (streamToString) {
|
||||
return Buffer.from(streamToString, 'base64')
|
||||
}
|
||||
}
|
||||
const byteArray = await response.Body!.transformToByteArray()
|
||||
return Buffer.from(byteArray)
|
||||
}
|
||||
|
||||
async getFileFromStorage(file: string, ...paths: string[]): Promise<Buffer> {
|
||||
const sanitizedFilename = this.sanitizeFilename(file)
|
||||
|
||||
let Key = paths.reduce((acc, cur) => acc + '/' + cur, '') + '/' + sanitizedFilename
|
||||
if (Key.startsWith('/')) {
|
||||
Key = Key.substring(1)
|
||||
}
|
||||
|
||||
try {
|
||||
const getParams = {
|
||||
Bucket: this.bucket,
|
||||
Key
|
||||
}
|
||||
|
||||
const response = await this.s3Client.send(new GetObjectCommand(getParams))
|
||||
const body = response.Body
|
||||
if (body instanceof Readable) {
|
||||
const streamToString = await body.transformToString('base64')
|
||||
if (streamToString) {
|
||||
return Buffer.from(streamToString, 'base64')
|
||||
}
|
||||
}
|
||||
const byteArray = await response.Body!.transformToByteArray()
|
||||
return Buffer.from(byteArray)
|
||||
} catch (error) {
|
||||
// Fallback: Check if file exists without the first path element (likely orgId)
|
||||
if (paths.length > 1) {
|
||||
const fallbackPaths = paths.slice(1)
|
||||
let fallbackKey = fallbackPaths.reduce((acc, cur) => acc + '/' + cur, '') + '/' + sanitizedFilename
|
||||
if (fallbackKey.startsWith('/')) {
|
||||
fallbackKey = fallbackKey.substring(1)
|
||||
}
|
||||
|
||||
try {
|
||||
const fallbackParams = {
|
||||
Bucket: this.bucket,
|
||||
Key: fallbackKey
|
||||
}
|
||||
const fallbackResponse = await this.s3Client.send(new GetObjectCommand(fallbackParams))
|
||||
const fallbackBody = fallbackResponse.Body
|
||||
|
||||
// Get the file content
|
||||
let fileContent: Buffer
|
||||
if (fallbackBody instanceof Readable) {
|
||||
const streamToString = await fallbackBody.transformToString('base64')
|
||||
if (streamToString) {
|
||||
fileContent = Buffer.from(streamToString, 'base64')
|
||||
} else {
|
||||
const byteArray = await fallbackBody.transformToByteArray()
|
||||
fileContent = Buffer.from(byteArray)
|
||||
}
|
||||
} else {
|
||||
const byteArray = await fallbackBody!.transformToByteArray()
|
||||
fileContent = Buffer.from(byteArray)
|
||||
}
|
||||
|
||||
// Move to correct location with orgId
|
||||
const putObjCmd = new PutObjectCommand({
|
||||
Bucket: this.bucket,
|
||||
Key,
|
||||
Body: fileContent
|
||||
})
|
||||
await this.s3Client.send(putObjCmd)
|
||||
|
||||
// Delete the old file
|
||||
await this.s3Client.send(
|
||||
new DeleteObjectsCommand({
|
||||
Bucket: this.bucket,
|
||||
Delete: {
|
||||
Objects: [{ Key: fallbackKey }],
|
||||
Quiet: false
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
// Check if the directory is empty and delete recursively if needed
|
||||
if (fallbackPaths.length > 0) {
|
||||
await this.cleanEmptyS3Folders(fallbackPaths[0])
|
||||
}
|
||||
|
||||
return fileContent
|
||||
} catch (fallbackError) {
|
||||
throw error
|
||||
}
|
||||
} else {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async streamStorageFile(chatflowId: string, chatId: string, fileName: string, orgId: string): Promise<Buffer | undefined> {
|
||||
// Validate chatflowId and chatId
|
||||
this.validateChatflowId(chatflowId)
|
||||
this.validatePathSecurity(chatflowId, chatId)
|
||||
|
||||
const sanitizedFilename = this.sanitizeFilename(fileName)
|
||||
const Key = orgId + '/' + chatflowId + '/' + chatId + '/' + sanitizedFilename
|
||||
|
||||
const getParams = {
|
||||
Bucket: this.bucket,
|
||||
Key
|
||||
}
|
||||
|
||||
try {
|
||||
const response = await this.s3Client.send(new GetObjectCommand(getParams))
|
||||
const body = response.Body
|
||||
if (body instanceof Readable) {
|
||||
const blob = await body.transformToByteArray()
|
||||
return Buffer.from(blob)
|
||||
}
|
||||
} catch (error) {
|
||||
// Fallback: Check if file exists without orgId
|
||||
const fallbackKey = chatflowId + '/' + chatId + '/' + sanitizedFilename
|
||||
try {
|
||||
const fallbackParams = {
|
||||
Bucket: this.bucket,
|
||||
Key: fallbackKey
|
||||
}
|
||||
const fallbackResponse = await this.s3Client.send(new GetObjectCommand(fallbackParams))
|
||||
const fallbackBody = fallbackResponse.Body
|
||||
|
||||
// If found, copy to correct location with orgId
|
||||
if (fallbackBody) {
|
||||
let fileContent: Buffer
|
||||
if (fallbackBody instanceof Readable) {
|
||||
const blob = await fallbackBody.transformToByteArray()
|
||||
fileContent = Buffer.from(blob)
|
||||
} else {
|
||||
const byteArray = await fallbackBody.transformToByteArray()
|
||||
fileContent = Buffer.from(byteArray)
|
||||
}
|
||||
|
||||
// Move to correct location with orgId
|
||||
const putObjCmd = new PutObjectCommand({
|
||||
Bucket: this.bucket,
|
||||
Key,
|
||||
Body: fileContent
|
||||
})
|
||||
await this.s3Client.send(putObjCmd)
|
||||
|
||||
// Delete the old file
|
||||
await this.s3Client.send(
|
||||
new DeleteObjectsCommand({
|
||||
Bucket: this.bucket,
|
||||
Delete: {
|
||||
Objects: [{ Key: fallbackKey }],
|
||||
Quiet: false
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
// Check if the directory is empty and delete recursively if needed
|
||||
await this.cleanEmptyS3Folders(chatflowId)
|
||||
|
||||
return fileContent
|
||||
}
|
||||
} catch (fallbackError) {
|
||||
throw new Error(`File ${fileName} not found`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async getFilesListFromStorage(...paths: string[]): Promise<FileInfo[]> {
|
||||
let Key = paths.reduce((acc, cur) => acc + '/' + cur, '')
|
||||
if (Key.startsWith('/')) {
|
||||
Key = Key.substring(1)
|
||||
}
|
||||
|
||||
const listCommand = new ListObjectsV2Command({
|
||||
Bucket: this.bucket,
|
||||
Prefix: Key
|
||||
})
|
||||
const list = await this.s3Client.send(listCommand)
|
||||
|
||||
if (list.Contents && list.Contents.length > 0) {
|
||||
return list.Contents.map((item) => ({
|
||||
name: item.Key?.split('/').pop() || '',
|
||||
path: item.Key ?? '',
|
||||
size: item.Size || 0
|
||||
}))
|
||||
} else {
|
||||
return []
|
||||
}
|
||||
}
|
||||
|
||||
async removeFilesFromStorage(...paths: string[]): Promise<StorageSizeResult> {
|
||||
let Key = paths.reduce((acc, cur) => acc + '/' + cur, '')
|
||||
if (Key.startsWith('/')) {
|
||||
Key = Key.substring(1)
|
||||
}
|
||||
|
||||
await this.deleteS3Folder(Key)
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async removeSpecificFileFromUpload(filePath: string): Promise<void> {
|
||||
let Key = filePath
|
||||
if (Key.startsWith('/')) {
|
||||
Key = Key.substring(1)
|
||||
}
|
||||
await this.deleteS3Folder(Key)
|
||||
}
|
||||
|
||||
async removeSpecificFileFromStorage(...paths: string[]): Promise<StorageSizeResult> {
|
||||
let Key = paths.reduce((acc, cur) => acc + '/' + cur, '')
|
||||
if (Key.startsWith('/')) {
|
||||
Key = Key.substring(1)
|
||||
}
|
||||
await this.deleteS3Folder(Key)
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
async removeFolderFromStorage(...paths: string[]): Promise<StorageSizeResult> {
|
||||
let Key = paths.reduce((acc, cur) => acc + '/' + cur, '')
|
||||
if (Key.startsWith('/')) {
|
||||
Key = Key.substring(1)
|
||||
}
|
||||
await this.deleteS3Folder(Key)
|
||||
|
||||
const totalSize = await this.getStorageSize(paths[0])
|
||||
return { totalSize: totalSize / 1024 / 1024 }
|
||||
}
|
||||
|
||||
private async deleteS3Folder(location: string): Promise<string> {
|
||||
let count = 0
|
||||
|
||||
const recursiveS3Delete = async (token?: string): Promise<string> => {
|
||||
const listCommand = new ListObjectsV2Command({
|
||||
Bucket: this.bucket,
|
||||
Prefix: location,
|
||||
ContinuationToken: token
|
||||
})
|
||||
const list = await this.s3Client.send(listCommand)
|
||||
|
||||
if (list.KeyCount) {
|
||||
const deleteCommand = new DeleteObjectsCommand({
|
||||
Bucket: this.bucket,
|
||||
Delete: {
|
||||
Objects: list.Contents?.map((item) => ({ Key: item.Key })),
|
||||
Quiet: false
|
||||
}
|
||||
})
|
||||
const deleted = await this.s3Client.send(deleteCommand)
|
||||
// @ts-ignore
|
||||
count += deleted.Deleted?.length || 0
|
||||
|
||||
if (deleted.Errors) {
|
||||
deleted.Errors.map((error: any) => console.error(`${error.Key} could not be deleted - ${error.Code}`))
|
||||
}
|
||||
}
|
||||
|
||||
if (list.NextContinuationToken) {
|
||||
return recursiveS3Delete(list.NextContinuationToken)
|
||||
}
|
||||
|
||||
return `${count} files deleted from S3`
|
||||
}
|
||||
|
||||
return recursiveS3Delete()
|
||||
}
|
||||
|
||||
private async cleanEmptyS3Folders(prefix: string): Promise<void> {
|
||||
try {
|
||||
if (!prefix) return
|
||||
|
||||
const listCmd = new ListObjectsV2Command({
|
||||
Bucket: this.bucket,
|
||||
Prefix: prefix + '/',
|
||||
Delimiter: '/'
|
||||
})
|
||||
|
||||
const response = await this.s3Client.send(listCmd)
|
||||
|
||||
if (
|
||||
(response.Contents?.length === 0 || !response.Contents) &&
|
||||
(response.CommonPrefixes?.length === 0 || !response.CommonPrefixes)
|
||||
) {
|
||||
await this.s3Client.send(
|
||||
new DeleteObjectsCommand({
|
||||
Bucket: this.bucket,
|
||||
Delete: {
|
||||
Objects: [{ Key: prefix + '/' }],
|
||||
Quiet: true
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
const parentPrefix = prefix.substring(0, prefix.lastIndexOf('/'))
|
||||
if (parentPrefix) {
|
||||
await this.cleanEmptyS3Folders(parentPrefix)
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Error cleaning empty S3 folders:', error)
|
||||
}
|
||||
}
|
||||
|
||||
async getStorageSize(orgId: string): Promise<number> {
|
||||
if (!orgId) return 0
|
||||
|
||||
const getCmd = new ListObjectsCommand({
|
||||
Bucket: this.bucket,
|
||||
Prefix: orgId
|
||||
})
|
||||
const headObj = await this.s3Client.send(getCmd)
|
||||
let totalSize = 0
|
||||
for (const obj of headObj.Contents || []) {
|
||||
totalSize += obj.Size || 0
|
||||
}
|
||||
return totalSize
|
||||
}
|
||||
|
||||
getMulterStorage(): multer.Multer {
|
||||
return multer({
|
||||
storage: multerS3({
|
||||
s3: this.s3Client,
|
||||
bucket: this.bucket,
|
||||
metadata: function (req: any, file: any, cb: any) {
|
||||
cb(null, { fieldName: file.fieldname, originalName: file.originalname })
|
||||
},
|
||||
key: function (req: any, file: any, cb: any) {
|
||||
cb(null, `${uuidv4()}`)
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
getLoggerTransports(logType: 'server' | 'error' | 'requests'): any[] {
|
||||
if (logType === 'server') {
|
||||
const s3ServerStream = new S3StreamLogger({
|
||||
bucket: this.bucket,
|
||||
folder: 'logs/server',
|
||||
name_format: `server-%Y-%m-%d-%H-%M-%S-%L.log`,
|
||||
config: this.s3Config
|
||||
})
|
||||
return [new transports.Stream({ stream: s3ServerStream })]
|
||||
} else if (logType === 'error') {
|
||||
const s3ErrorStream = new S3StreamLogger({
|
||||
bucket: this.bucket,
|
||||
folder: 'logs/error',
|
||||
name_format: `server-error-%Y-%m-%d-%H-%M-%S-%L.log`,
|
||||
config: this.s3Config
|
||||
})
|
||||
return [new transports.Stream({ stream: s3ErrorStream })]
|
||||
} else if (logType === 'requests') {
|
||||
const s3ServerReqStream = new S3StreamLogger({
|
||||
bucket: this.bucket,
|
||||
folder: 'logs/requests',
|
||||
name_format: `server-requests-%Y-%m-%d-%H-%M-%S-%L.log.jsonl`,
|
||||
config: this.s3Config
|
||||
})
|
||||
return [new transports.Stream({ stream: s3ServerReqStream })]
|
||||
}
|
||||
return []
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
import { IStorageProvider } from './IStorageProvider'
|
||||
import { LocalStorageProvider } from './LocalStorageProvider'
|
||||
import { S3StorageProvider } from './S3StorageProvider'
|
||||
import { GCSStorageProvider } from './GCSStorageProvider'
|
||||
import { AzureBlobStorageProvider } from './AzureBlobStorageProvider'
|
||||
|
||||
/**
|
||||
* Factory for creating and managing storage provider instances.
|
||||
* Uses singleton pattern to ensure only one provider instance exists.
|
||||
*/
|
||||
export class StorageProviderFactory {
|
||||
private static instance: IStorageProvider | null = null
|
||||
|
||||
/**
|
||||
* Get the storage provider instance based on STORAGE_TYPE environment variable.
|
||||
* Creates a new instance if one doesn't exist.
|
||||
*/
|
||||
static getProvider(): IStorageProvider {
|
||||
if (!StorageProviderFactory.instance) {
|
||||
const storageType = process.env.STORAGE_TYPE || 'local'
|
||||
|
||||
switch (storageType) {
|
||||
case 's3':
|
||||
StorageProviderFactory.instance = new S3StorageProvider()
|
||||
break
|
||||
case 'gcs':
|
||||
StorageProviderFactory.instance = new GCSStorageProvider()
|
||||
break
|
||||
case 'azure':
|
||||
StorageProviderFactory.instance = new AzureBlobStorageProvider()
|
||||
break
|
||||
case 'local':
|
||||
default:
|
||||
StorageProviderFactory.instance = new LocalStorageProvider()
|
||||
break
|
||||
}
|
||||
}
|
||||
return StorageProviderFactory.instance
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
// Interfaces and types
|
||||
export * from './IStorageProvider'
|
||||
|
||||
// Base provider
|
||||
export { BaseStorageProvider } from './BaseStorageProvider'
|
||||
|
||||
// Provider implementations
|
||||
export { LocalStorageProvider } from './LocalStorageProvider'
|
||||
export { S3StorageProvider } from './S3StorageProvider'
|
||||
export { GCSStorageProvider } from './GCSStorageProvider'
|
||||
export { AzureBlobStorageProvider } from './AzureBlobStorageProvider'
|
||||
|
||||
// Factory
|
||||
export { StorageProviderFactory } from './StorageProviderFactory'
|
||||
File diff suppressed because it is too large
Load Diff
@@ -49,7 +49,7 @@ PORT=3000
|
||||
############################################## STORAGE #####################################################
|
||||
############################################################################################################
|
||||
|
||||
# STORAGE_TYPE=local (local | s3 | gcs)
|
||||
# STORAGE_TYPE=local (local | s3 | gcs | azure)
|
||||
# BLOB_STORAGE_PATH=/your_storage_path/.flowise/storage
|
||||
# S3_STORAGE_BUCKET_NAME=flowise
|
||||
# S3_STORAGE_ACCESS_KEY_ID=<your-access-key>
|
||||
@@ -61,7 +61,12 @@ PORT=3000
|
||||
# GOOGLE_CLOUD_STORAGE_PROJ_ID=<your-gcp-project-id>
|
||||
# GOOGLE_CLOUD_STORAGE_BUCKET_NAME=<the-bucket-name>
|
||||
# GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS=true
|
||||
|
||||
# Azure Blob Storage (when STORAGE_TYPE=azure)
|
||||
# Provide EITHER a connection string OR account name + key
|
||||
# AZURE_BLOB_STORAGE_CONNECTION_STRING=<your-connection-string>
|
||||
# AZURE_BLOB_STORAGE_ACCOUNT_NAME=<your-account-name>
|
||||
# AZURE_BLOB_STORAGE_ACCOUNT_KEY=<your-account-key>
|
||||
# AZURE_BLOB_STORAGE_CONTAINER_NAME=<your-container-name>
|
||||
|
||||
############################################################################################################
|
||||
############################################## SETTINGS ####################################################
|
||||
|
||||
@@ -120,6 +120,7 @@
|
||||
"moment": "^2.29.3",
|
||||
"moment-timezone": "^0.5.34",
|
||||
"multer": "^2.0.2",
|
||||
"multer-azure-blob-storage": "^1.2.0",
|
||||
"multer-cloud-storage": "^4.0.0",
|
||||
"multer-s3": "^3.0.1",
|
||||
"mysql2": "^3.11.3",
|
||||
@@ -147,6 +148,7 @@
|
||||
"typeorm": "^0.3.6",
|
||||
"uuid": "^9.0.1",
|
||||
"winston": "^3.9.0",
|
||||
"winston-azure-blob": "^1.5.0",
|
||||
"winston-daily-rotate-file": "^5.0.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
|
||||
@@ -68,6 +68,10 @@ export abstract class BaseCommand extends Command {
|
||||
GOOGLE_CLOUD_STORAGE_PROJ_ID: Flags.string(),
|
||||
GOOGLE_CLOUD_STORAGE_BUCKET_NAME: Flags.string(),
|
||||
GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS: Flags.string(),
|
||||
AZURE_BLOB_STORAGE_CONNECTION_STRING: Flags.string(),
|
||||
AZURE_BLOB_STORAGE_ACCOUNT_NAME: Flags.string(),
|
||||
AZURE_BLOB_STORAGE_ACCOUNT_KEY: Flags.string(),
|
||||
AZURE_BLOB_STORAGE_CONTAINER_NAME: Flags.string(),
|
||||
|
||||
// Credentials / Secret Keys
|
||||
SECRETKEY_STORAGE_TYPE: Flags.string(),
|
||||
|
||||
@@ -38,13 +38,11 @@ import {
|
||||
IMessage,
|
||||
FlowiseMemory,
|
||||
IFileUpload,
|
||||
getS3Config
|
||||
StorageProviderFactory
|
||||
} from 'flowise-components'
|
||||
import { randomBytes } from 'crypto'
|
||||
import { AES, enc } from 'crypto-js'
|
||||
import multer from 'multer'
|
||||
import multerS3 from 'multer-s3'
|
||||
import MulterGoogleCloudStorage from 'multer-cloud-storage'
|
||||
|
||||
import { ChatFlow } from '../database/entities/ChatFlow'
|
||||
import { ChatMessage } from '../database/entities/ChatMessage'
|
||||
import { Credential } from '../database/entities/Credential'
|
||||
@@ -1997,38 +1995,8 @@ export function generateId() {
|
||||
}
|
||||
|
||||
export const getMulterStorage = () => {
|
||||
const storageType = process.env.STORAGE_TYPE ? process.env.STORAGE_TYPE : 'local'
|
||||
|
||||
if (storageType === 's3') {
|
||||
const s3Client = getS3Config().s3Client
|
||||
const Bucket = getS3Config().Bucket
|
||||
|
||||
const upload = multer({
|
||||
storage: multerS3({
|
||||
s3: s3Client,
|
||||
bucket: Bucket,
|
||||
metadata: function (req, file, cb) {
|
||||
cb(null, { fieldName: file.fieldname, originalName: file.originalname })
|
||||
},
|
||||
key: function (req, file, cb) {
|
||||
cb(null, `${generateId()}`)
|
||||
}
|
||||
})
|
||||
})
|
||||
return upload
|
||||
} else if (storageType === 'gcs') {
|
||||
return multer({
|
||||
storage: new MulterGoogleCloudStorage({
|
||||
projectId: process.env.GOOGLE_CLOUD_STORAGE_PROJ_ID,
|
||||
bucket: process.env.GOOGLE_CLOUD_STORAGE_BUCKET_NAME,
|
||||
keyFilename: process.env.GOOGLE_CLOUD_STORAGE_CREDENTIAL,
|
||||
uniformBucketLevelAccess: Boolean(process.env.GOOGLE_CLOUD_UNIFORM_BUCKET_ACCESS) ?? true,
|
||||
destination: `uploads/${generateId()}`
|
||||
})
|
||||
})
|
||||
} else {
|
||||
return multer({ dest: getUploadPath() })
|
||||
}
|
||||
const provider = StorageProviderFactory.getProvider()
|
||||
return provider.getMulterStorage()
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -1,100 +1,18 @@
|
||||
import * as path from 'path'
|
||||
import * as fs from 'fs'
|
||||
import config from './config' // should be replaced by node-config or similar
|
||||
import { createLogger, transports, format } from 'winston'
|
||||
import { NextFunction, Request, Response } from 'express'
|
||||
import DailyRotateFile from 'winston-daily-rotate-file'
|
||||
import { S3ClientConfig } from '@aws-sdk/client-s3'
|
||||
import { LoggingWinston } from '@google-cloud/logging-winston'
|
||||
|
||||
const { S3StreamLogger } = require('s3-streamlogger')
|
||||
import { StorageProviderFactory } from 'flowise-components'
|
||||
|
||||
const { combine, timestamp, printf, errors } = format
|
||||
|
||||
let s3ServerStream: any
|
||||
let s3ErrorStream: any
|
||||
let s3ServerReqStream: any
|
||||
|
||||
let gcsServerStream: any
|
||||
let gcsErrorStream: any
|
||||
let gcsServerReqStream: any
|
||||
|
||||
let requestLogger: any
|
||||
|
||||
if (process.env.STORAGE_TYPE === 's3') {
|
||||
const accessKeyId = process.env.S3_STORAGE_ACCESS_KEY_ID
|
||||
const secretAccessKey = process.env.S3_STORAGE_SECRET_ACCESS_KEY
|
||||
const region = process.env.S3_STORAGE_REGION
|
||||
const s3Bucket = process.env.S3_STORAGE_BUCKET_NAME
|
||||
const customURL = process.env.S3_ENDPOINT_URL
|
||||
const forcePathStyle = process.env.S3_FORCE_PATH_STYLE === 'true'
|
||||
const provider = StorageProviderFactory.getProvider()
|
||||
const serverTransports = provider.getLoggerTransports('server', config)
|
||||
const errorTransports = provider.getLoggerTransports('error', config)
|
||||
const requestTransports = provider.getLoggerTransports('requests', config)
|
||||
|
||||
if (!region || region.trim() === '' || !s3Bucket || s3Bucket.trim() === '') {
|
||||
throw new Error('S3 storage configuration is missing')
|
||||
}
|
||||
|
||||
const s3Config: S3ClientConfig = {
|
||||
region: region,
|
||||
forcePathStyle: forcePathStyle
|
||||
}
|
||||
|
||||
// Only include endpoint if customURL is not empty
|
||||
if (customURL && customURL.trim() !== '') {
|
||||
s3Config.endpoint = customURL
|
||||
}
|
||||
|
||||
if (accessKeyId && accessKeyId.trim() !== '' && secretAccessKey && secretAccessKey.trim() !== '') {
|
||||
s3Config.credentials = {
|
||||
accessKeyId: accessKeyId,
|
||||
secretAccessKey: secretAccessKey
|
||||
}
|
||||
}
|
||||
|
||||
s3ServerStream = new S3StreamLogger({
|
||||
bucket: s3Bucket,
|
||||
folder: 'logs/server',
|
||||
name_format: `server-%Y-%m-%d-%H-%M-%S-%L.log`,
|
||||
config: s3Config
|
||||
})
|
||||
|
||||
s3ErrorStream = new S3StreamLogger({
|
||||
bucket: s3Bucket,
|
||||
folder: 'logs/error',
|
||||
name_format: `server-error-%Y-%m-%d-%H-%M-%S-%L.log`,
|
||||
config: s3Config
|
||||
})
|
||||
|
||||
s3ServerReqStream = new S3StreamLogger({
|
||||
bucket: s3Bucket,
|
||||
folder: 'logs/requests',
|
||||
name_format: `server-requests-%Y-%m-%d-%H-%M-%S-%L.log.jsonl`,
|
||||
config: s3Config
|
||||
})
|
||||
}
|
||||
|
||||
if (process.env.STORAGE_TYPE === 'gcs') {
|
||||
const config = {
|
||||
projectId: process.env.GOOGLE_CLOUD_STORAGE_PROJ_ID,
|
||||
keyFilename: process.env.GOOGLE_CLOUD_STORAGE_CREDENTIAL,
|
||||
defaultCallback: (err: any) => {
|
||||
if (err) {
|
||||
console.error('Error logging to GCS: ' + err)
|
||||
}
|
||||
}
|
||||
}
|
||||
gcsServerStream = new LoggingWinston({
|
||||
...config,
|
||||
logName: 'server'
|
||||
})
|
||||
gcsErrorStream = new LoggingWinston({
|
||||
...config,
|
||||
logName: 'error'
|
||||
})
|
||||
gcsServerReqStream = new LoggingWinston({
|
||||
...config,
|
||||
logName: 'requests'
|
||||
})
|
||||
}
|
||||
// expect the log dir be relative to the projects root
|
||||
const logDir = config.logging.dir
|
||||
|
||||
@@ -117,52 +35,13 @@ const logger = createLogger({
|
||||
package: 'server'
|
||||
},
|
||||
exitOnError: false,
|
||||
transports: [
|
||||
new transports.Console(),
|
||||
...(!process.env.STORAGE_TYPE || process.env.STORAGE_TYPE === 'local'
|
||||
? [
|
||||
new DailyRotateFile({
|
||||
filename: path.join(logDir, config.logging.server.filename ?? 'server-%DATE%.log'),
|
||||
datePattern: 'YYYY-MM-DD-HH',
|
||||
maxSize: '20m',
|
||||
level: config.logging.server.level ?? 'info'
|
||||
})
|
||||
]
|
||||
: []),
|
||||
...(process.env.STORAGE_TYPE === 's3'
|
||||
? [
|
||||
new transports.Stream({
|
||||
stream: s3ServerStream
|
||||
})
|
||||
]
|
||||
: []),
|
||||
...(process.env.STORAGE_TYPE === 'gcs' ? [gcsServerStream] : [])
|
||||
],
|
||||
exceptionHandlers: [
|
||||
...(process.env.DEBUG && process.env.DEBUG === 'true' ? [new transports.Console()] : []),
|
||||
...(process.env.STORAGE_TYPE === 's3'
|
||||
? [
|
||||
new transports.Stream({
|
||||
stream: s3ErrorStream
|
||||
})
|
||||
]
|
||||
: []),
|
||||
...(process.env.STORAGE_TYPE === 'gcs' ? [gcsErrorStream] : [])
|
||||
],
|
||||
transports: [new transports.Console(), ...serverTransports],
|
||||
exceptionHandlers: [...(process.env.DEBUG && process.env.DEBUG === 'true' ? [new transports.Console()] : []), ...errorTransports],
|
||||
rejectionHandlers: [
|
||||
...(process.env.DEBUG && process.env.DEBUG === 'true' ? [new transports.Console()] : []),
|
||||
...(process.env.STORAGE_TYPE === 's3'
|
||||
? [
|
||||
new transports.Stream({
|
||||
stream: s3ErrorStream
|
||||
})
|
||||
]
|
||||
: []),
|
||||
...(process.env.STORAGE_TYPE === 'gcs' ? [gcsErrorStream] : []),
|
||||
...errorTransports,
|
||||
// Always provide a fallback rejection handler when no other handlers are configured
|
||||
...((!process.env.DEBUG || process.env.DEBUG !== 'true') && process.env.STORAGE_TYPE !== 's3' && process.env.STORAGE_TYPE !== 'gcs'
|
||||
? [new transports.Console()]
|
||||
: [])
|
||||
...((!process.env.DEBUG || process.env.DEBUG !== 'true') && errorTransports.length === 0 ? [new transports.Console()] : [])
|
||||
]
|
||||
})
|
||||
|
||||
@@ -171,25 +50,7 @@ requestLogger = createLogger({
|
||||
defaultMeta: {
|
||||
package: 'server'
|
||||
},
|
||||
transports: [
|
||||
...(process.env.DEBUG && process.env.DEBUG === 'true' ? [new transports.Console()] : []),
|
||||
...(!process.env.STORAGE_TYPE || process.env.STORAGE_TYPE === 'local'
|
||||
? [
|
||||
new transports.File({
|
||||
filename: path.join(logDir, config.logging.express.filename ?? 'server-requests.log.jsonl'),
|
||||
level: config.logging.express.level ?? 'debug'
|
||||
})
|
||||
]
|
||||
: []),
|
||||
...(process.env.STORAGE_TYPE === 's3'
|
||||
? [
|
||||
new transports.Stream({
|
||||
stream: s3ServerReqStream
|
||||
})
|
||||
]
|
||||
: []),
|
||||
...(process.env.STORAGE_TYPE === 'gcs' ? [gcsServerReqStream] : [])
|
||||
]
|
||||
transports: [...(process.env.DEBUG && process.env.DEBUG === 'true' ? [new transports.Console()] : []), ...requestTransports]
|
||||
})
|
||||
|
||||
function getSensitiveBodyFields(): string[] {
|
||||
|
||||
Generated
+528
-14
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user