Skip to content

feat(attachments): use filesystem for gmail, outlook triggers to save attachments #1631

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
icecrasher321 merged 10 commits into from
Oct 15, 2025
Merged

feat(attachments): use filesystem for gmail, outlook triggers to save attachments #1631

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
dfefe2a
feat(outlook): add include attachment feature to outlook
icecrasher321 Oct 14, 2025
ff2d891
add include attachments to gmail trigger
icecrasher321 Oct 15, 2025
8ba36de
add gmail trigger, outlook block include attachments
icecrasher321 Oct 15, 2025
c5478c9
fix rendering issue
icecrasher321 Oct 15, 2025
3a07855
remove comment
icecrasher321 Oct 15, 2025
887b3a5
fix architecture
icecrasher321 Oct 15, 2025
bcdf843
fix redeploy
icecrasher321 Oct 15, 2025
1a88ab5
pass files to logging session to surface in logs
icecrasher321 Oct 15, 2025
22842eb
fix gmail block parsing attachments
icecrasher321 Oct 15, 2025
0908b34
fix reads
icecrasher321 Oct 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 55 additions & 95 deletions apps/sim/app/api/workflows/[id]/deploy/route.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,7 @@
import { apiKey, db, workflow, workflowDeploymentVersion } from '@sim/db'
import { and, desc, eq, sql } from 'drizzle-orm'
import type { NextRequest } from 'next/server'
import { type NextRequest, NextResponse } from 'next/server'
import { v4 as uuidv4 } from 'uuid'
import { generateApiKey } from '@/lib/api-key/service'
import { createLogger } from '@/lib/logs/console/logger'
import { generateRequestId } from '@/lib/utils'
import { loadWorkflowFromNormalizedTables } from '@/lib/workflows/db-helpers'
Expand Down Expand Up @@ -64,26 +63,7 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
.orderBy(desc(apiKey.lastUsed), desc(apiKey.createdAt))
.limit(1)

if (userApiKey.length === 0) {
try {
const newApiKeyVal = generateApiKey()
const keyName = 'Default API Key'
await db.insert(apiKey).values({
id: uuidv4(),
userId: workflowData.userId,
workspaceId: null,
name: keyName,
key: newApiKeyVal,
type: 'personal',
createdAt: new Date(),
updatedAt: new Date(),
})
keyInfo = { name: keyName, type: 'personal' }
logger.info(`[${requestId}] Generated new API key for user: ${workflowData.userId}`)
} catch (keyError) {
logger.error(`[${requestId}] Failed to generate API key:`, keyError)
}
} else {
if (userApiKey.length > 0) {
keyInfo = { name: userApiKey[0].name, type: userApiKey[0].type as 'personal' | 'workspace' }
}
}
Expand Down Expand Up @@ -190,34 +170,6 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
const deployedAt = new Date()
logger.debug(`[${requestId}] Proceeding with deployment at ${deployedAt.toISOString()}`)

const userApiKey = await db
.select({
key: apiKey.key,
})
.from(apiKey)
.where(and(eq(apiKey.userId, userId), eq(apiKey.type, 'personal')))
.orderBy(desc(apiKey.lastUsed), desc(apiKey.createdAt))
.limit(1)

if (userApiKey.length === 0) {
try {
const newApiKey = generateApiKey()
await db.insert(apiKey).values({
id: uuidv4(),
userId,
workspaceId: null,
name: 'Default API Key',
key: newApiKey,
type: 'personal',
createdAt: new Date(),
updatedAt: new Date(),
})
logger.info(`[${requestId}] Generated new API key for user: ${userId}`)
} catch (keyError) {
logger.error(`[${requestId}] Failed to generate API key:`, keyError)
}
}

let keyInfo: { name: string; type: 'personal' | 'workspace' } | null = null
let matchedKey: {
id: string
Expand All @@ -226,13 +178,50 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
type: 'personal' | 'workspace'
} | null = null

if (providedApiKey) {
let isValidKey = false
// Use provided API key, or fall back to existing pinned API key for redeployment
const apiKeyToUse = providedApiKey || workflowData!.pinnedApiKeyId

if (!apiKeyToUse) {
return NextResponse.json(
{ error: 'API key is required. Please create or select an API key before deploying.' },
{ status: 400 }
)
}

let isValidKey = false

const currentUserId = session?.user?.id

if (currentUserId) {
const [personalKey] = await db
.select({
id: apiKey.id,
key: apiKey.key,
name: apiKey.name,
expiresAt: apiKey.expiresAt,
})
.from(apiKey)
.where(
and(
eq(apiKey.id, apiKeyToUse),
eq(apiKey.userId, currentUserId),
eq(apiKey.type, 'personal')
)
)
.limit(1)

const currentUserId = session?.user?.id
if (personalKey) {
if (!personalKey.expiresAt || personalKey.expiresAt >= new Date()) {
matchedKey = { ...personalKey, type: 'personal' }
isValidKey = true
keyInfo = { name: personalKey.name, type: 'personal' }
}
}
}

if (currentUserId) {
const [personalKey] = await db
if (!isValidKey) {
if (workflowData!.workspaceId) {
const [workspaceKey] = await db
.select({
id: apiKey.id,
key: apiKey.key,
Expand All @@ -242,55 +231,26 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
.from(apiKey)
.where(
and(
eq(apiKey.id, providedApiKey),
eq(apiKey.userId, currentUserId),
eq(apiKey.type, 'personal')
eq(apiKey.id, apiKeyToUse),
eq(apiKey.workspaceId, workflowData!.workspaceId),
eq(apiKey.type, 'workspace')
)
)
.limit(1)

if (personalKey) {
if (!personalKey.expiresAt || personalKey.expiresAt >= new Date()) {
matchedKey = { ...personalKey, type: 'personal' }
if (workspaceKey) {
if (!workspaceKey.expiresAt || workspaceKey.expiresAt >= new Date()) {
matchedKey = { ...workspaceKey, type: 'workspace' }
isValidKey = true
keyInfo = { name: personalKey.name, type: 'personal' }
}
}
}

if (!isValidKey) {
if (workflowData!.workspaceId) {
const [workspaceKey] = await db
.select({
id: apiKey.id,
key: apiKey.key,
name: apiKey.name,
expiresAt: apiKey.expiresAt,
})
.from(apiKey)
.where(
and(
eq(apiKey.id, providedApiKey),
eq(apiKey.workspaceId, workflowData!.workspaceId),
eq(apiKey.type, 'workspace')
)
)
.limit(1)

if (workspaceKey) {
if (!workspaceKey.expiresAt || workspaceKey.expiresAt >= new Date()) {
matchedKey = { ...workspaceKey, type: 'workspace' }
isValidKey = true
keyInfo = { name: workspaceKey.name, type: 'workspace' }
}
keyInfo = { name: workspaceKey.name, type: 'workspace' }
}
}
}
}

if (!isValidKey) {
logger.warn(`[${requestId}] Invalid API key ID provided for workflow deployment: ${id}`)
return createErrorResponse('Invalid API key provided', 400)
}
if (!isValidKey) {
logger.warn(`[${requestId}] Invalid API key ID provided for workflow deployment: ${id}`)
return createErrorResponse('Invalid API key provided', 400)
}

// Attribution: this route is UI-only; require session user as actor
Expand Down
2 changes: 2 additions & 0 deletions ...rkspaceId]/w/[workflowId]/components/control-bar/components/deploy-modal/deploy-modal.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -361,6 +361,8 @@ export function DeployModal({
await fetchVersions()
} catch (error: unknown) {
logger.error('Error deploying workflow:', { error })
const errorMessage = error instanceof Error ? error.message : 'Failed to deploy workflow'
setApiDeployError(errorMessage)
} finally {
setIsSubmitting(false)
}
Expand Down
16 changes: 14 additions & 2 deletions ...m/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/workflow-block.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -432,6 +432,17 @@ export const WorkflowBlock = memo(
}
}, [id, blockHeight, blockWidth, updateBlockLayoutMetrics, updateNodeInternals, debounce])

// Subscribe to this block's subblock values to track changes for conditional rendering
const blockSubBlockValues = useSubBlockStore(
useCallback(
(state) => {
if (!activeWorkflowId) return {}
return state.workflowValues[activeWorkflowId]?.[id] || {}
},
[activeWorkflowId, id]
)
)

// Memoized SubBlock layout management - only recalculate when dependencies change
const subBlockRows = useMemo(() => {
const rows: SubBlockConfig[][] = []
Expand All @@ -450,8 +461,7 @@ export const WorkflowBlock = memo(
} else {
// In normal mode, use merged state
const blocks = useWorkflowStore.getState().blocks
const activeWorkflowId = useWorkflowRegistry.getState().activeWorkflowId || undefined
const mergedState = mergeSubblockState(blocks, activeWorkflowId, id)[id]
const mergedState = mergeSubblockState(blocks, activeWorkflowId || undefined, id)[id]
stateToUse = mergedState?.subBlocks || {}
}

Expand Down Expand Up @@ -552,6 +562,8 @@ export const WorkflowBlock = memo(
data.subBlockValues,
currentWorkflow.isDiffMode,
currentBlock,
blockSubBlockValues,
activeWorkflowId,
])

// Name editing handlers
Expand Down
86 changes: 86 additions & 0 deletions apps/sim/background/webhook-execution.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import { createLogger } from '@/lib/logs/console/logger'
import { LoggingSession } from '@/lib/logs/execution/logging-session'
import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
import { decryptSecret } from '@/lib/utils'
import { WebhookAttachmentProcessor } from '@/lib/webhooks/attachment-processor'
import { fetchAndProcessAirtablePayloads, formatWebhookInput } from '@/lib/webhooks/utils'
import {
loadDeployedWorkflowState,
Expand All @@ -20,9 +21,66 @@ import { Executor } from '@/executor'
import type { ExecutionResult } from '@/executor/types'
import { Serializer } from '@/serializer'
import { mergeSubblockState } from '@/stores/workflows/server-utils'
import { getTrigger } from '@/triggers'

const logger = createLogger('TriggerWebhookExecution')

/**
* Process trigger outputs based on their schema definitions
* Finds outputs marked as 'file' or 'file[]' and uploads them to execution storage
*/
async function processTriggerFileOutputs(
input: any,
triggerOutputs: Record<string, any>,
context: {
workspaceId: string
workflowId: string
executionId: string
requestId: string
},
path = ''
): Promise<any> {
if (!input || typeof input !== 'object') {
return input
}

const processed: any = Array.isArray(input) ? [] : {}

for (const [key, value] of Object.entries(input)) {
const currentPath = path ? `${path}.${key}` : key
const outputDef = triggerOutputs[key]
const val: any = value

// If this field is marked as file or file[], process it
if (outputDef?.type === 'file[]' && Array.isArray(val)) {
try {
processed[key] = await WebhookAttachmentProcessor.processAttachments(val as any, context)
} catch (error) {
processed[key] = []
}
} else if (outputDef?.type === 'file' && val) {
try {
const [processedFile] = await WebhookAttachmentProcessor.processAttachments(
[val as any],
context
)
processed[key] = processedFile
} catch (error) {
logger.error(`[${context.requestId}] Error processing ${currentPath}:`, error)
processed[key] = val
}
} else if (outputDef && typeof outputDef === 'object' && !outputDef.type) {
// Nested object in schema - recurse with the nested schema
processed[key] = await processTriggerFileOutputs(val, outputDef, context, currentPath)
} else {
// Not a file output - keep as is
processed[key] = val
}
}

return processed
}

export type WebhookExecutionPayload = {
webhookId: string
workflowId: string
Expand Down Expand Up @@ -250,6 +308,7 @@ async function executeWebhookJobInternal(
totalDurationMs: totalDuration || 0,
finalOutput: executionResult.output || {},
traceSpans: traceSpans as any,
workflowInput: airtableInput,
})

return {
Expand Down Expand Up @@ -312,6 +371,32 @@ async function executeWebhookJobInternal(
}
}

// Process trigger file outputs based on schema
if (input && payload.blockId && blocks[payload.blockId]) {
try {
const triggerBlock = blocks[payload.blockId]
const triggerId = triggerBlock?.subBlocks?.triggerId?.value

if (triggerId && typeof triggerId === 'string') {
const triggerConfig = getTrigger(triggerId)

if (triggerConfig?.outputs) {
logger.debug(`[${requestId}] Processing trigger ${triggerId} file outputs`)
const processedInput = await processTriggerFileOutputs(input, triggerConfig.outputs, {
workspaceId: workspaceId || '',
workflowId: payload.workflowId,
executionId,
requestId,
})
Object.assign(input, processedInput)
}
}
} catch (error) {
logger.error(`[${requestId}] Error processing trigger file outputs:`, error)
// Continue without processing attachments rather than failing execution
}
}

// Create executor and execute
const executor = new Executor({
workflow: serializedWorkflow,
Expand Down Expand Up @@ -367,6 +452,7 @@ async function executeWebhookJobInternal(
totalDurationMs: totalDuration || 0,
finalOutput: executionResult.output || {},
traceSpans: traceSpans as any,
workflowInput: input,
})

return {
Expand Down
12 changes: 12 additions & 0 deletions apps/sim/blocks/blocks/outlook.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,6 +173,13 @@ export const OutlookBlock: BlockConfig<OutlookResponse> = {
placeholder: 'Number of emails to retrieve (default: 1, max: 10)',
condition: { field: 'operation', value: 'read_outlook' },
},
{
id: 'includeAttachments',
title: 'Include Attachments',
type: 'switch',
layout: 'full',
condition: { field: 'operation', value: 'read_outlook' },
},
// TRIGGER MODE: Trigger configuration (only shown when trigger mode is active)
{
id: 'triggerConfig',
Expand Down Expand Up @@ -231,6 +238,7 @@ export const OutlookBlock: BlockConfig<OutlookResponse> = {
folder: { type: 'string', description: 'Email folder' },
manualFolder: { type: 'string', description: 'Manual folder name' },
maxResults: { type: 'number', description: 'Maximum emails' },
includeAttachments: { type: 'boolean', description: 'Include email attachments' },
},
outputs: {
// Common outputs
Expand All @@ -255,6 +263,10 @@ export const OutlookBlock: BlockConfig<OutlookResponse> = {
receivedDateTime: { type: 'string', description: 'Email received timestamp' },
sentDateTime: { type: 'string', description: 'Email sent timestamp' },
hasAttachments: { type: 'boolean', description: 'Whether email has attachments' },
attachments: {
type: 'json',
description: 'Email attachments (if includeAttachments is enabled)',
},
isRead: { type: 'boolean', description: 'Whether email is read' },
importance: { type: 'string', description: 'Email importance level' },
// Trigger outputs
Expand Down
Loading