mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-24 15:45:35 +08:00
improvement(webhooks): add trigger-age instrumentation + guard env decryption (#5236)
- Add dispatch-latency / trigger-age instrumentation: capture webhook receipt
time + Slack x-slack-request-timestamp at the route and log structured
dispatchLatencyMs + triggerAgeMs before execution, surfacing the pre-execution
latency that per-block timings cannot see (Slack trigger_id expires at 3s).
- Guard the effective-env fetch in verifyProviderAuth: only fetch+decrypt when
the handler verifies auth AND the providerConfig references env vars ({{VAR}}),
avoiding a needless DB read/decrypt on the synchronous pre-ack path. The guard
scope exactly matches resolveProviderConfigEnvVars, so resolution is identical.
This commit is contained in:
@@ -69,6 +69,17 @@ async function handleWebhookPost(
|
||||
request: NextRequest,
|
||||
context: { params: Promise<{ path: string }> }
|
||||
): Promise<NextResponse> {
|
||||
const receivedAt = Date.now()
|
||||
/**
|
||||
* Slack signs every interactive request with the originating interaction time.
|
||||
* Capturing it lets the executor surface the true trigger_id age (the window
|
||||
* that expires at 3s) instead of only the in-workflow block timings.
|
||||
*/
|
||||
const slackRequestTimestamp = request.headers.get('x-slack-request-timestamp')
|
||||
const triggerTimestampMs = slackRequestTimestamp
|
||||
? Number(slackRequestTimestamp) * 1000
|
||||
: undefined
|
||||
|
||||
const requestId = generateRequestId()
|
||||
const parsed = await parseRequest(webhookTriggerPostContract, request, context)
|
||||
if (!parsed.success) return parsed.response
|
||||
@@ -200,6 +211,8 @@ async function handleWebhookPost(
|
||||
actorUserId: preprocessResult.actorUserId,
|
||||
executionId: preprocessResult.executionId,
|
||||
correlation: preprocessResult.correlation,
|
||||
receivedAt,
|
||||
triggerTimestampMs: Number.isFinite(triggerTimestampMs) ? triggerTimestampMs : undefined,
|
||||
})
|
||||
responses.push(response)
|
||||
}
|
||||
|
||||
@@ -237,6 +237,10 @@ export type WebhookExecutionPayload = {
|
||||
blockId?: string
|
||||
workspaceId?: string
|
||||
credentialId?: string
|
||||
/** Epoch ms when the webhook HTTP request was first received (for dispatch-latency metrics). */
|
||||
webhookReceivedAt?: number
|
||||
/** Epoch ms of the originating provider interaction (e.g. Slack x-slack-request-timestamp). */
|
||||
triggerTimestampMs?: number
|
||||
}
|
||||
|
||||
export async function executeWebhookJob(payload: WebhookExecutionPayload) {
|
||||
@@ -565,6 +569,24 @@ async function executeWebhookJobInternal(
|
||||
|
||||
const triggerInput = input || {}
|
||||
|
||||
/**
|
||||
* Surface the pre-execution latency that per-block timings cannot see: the
|
||||
* gap between webhook receipt and the first block running, and — for
|
||||
* trigger_id-bound providers like Slack — the true age of the interaction
|
||||
* against its 3s expiry window. Logged structured so it is queryable/alarmable.
|
||||
*/
|
||||
if (payload.webhookReceivedAt !== undefined || payload.triggerTimestampMs !== undefined) {
|
||||
const now = Date.now()
|
||||
logger.info(`[${requestId}] Webhook dispatch latency`, {
|
||||
workflowId: payload.workflowId,
|
||||
provider: payload.provider,
|
||||
dispatchLatencyMs:
|
||||
payload.webhookReceivedAt !== undefined ? now - payload.webhookReceivedAt : undefined,
|
||||
triggerAgeMs:
|
||||
payload.triggerTimestampMs !== undefined ? now - payload.triggerTimestampMs : undefined,
|
||||
})
|
||||
}
|
||||
|
||||
const snapshot = new ExecutionSnapshot(
|
||||
metadata,
|
||||
workflowRecord,
|
||||
|
||||
@@ -39,6 +39,10 @@ export interface WebhookProcessorOptions {
|
||||
actorUserId?: string
|
||||
executionId?: string
|
||||
correlation?: AsyncExecutionCorrelation
|
||||
/** Epoch ms when the webhook HTTP request was first received (for dispatch-latency metrics). */
|
||||
receivedAt?: number
|
||||
/** Epoch ms of the originating provider interaction (e.g. Slack x-slack-request-timestamp). */
|
||||
triggerTimestampMs?: number
|
||||
}
|
||||
|
||||
export interface WebhookPreprocessingResult {
|
||||
@@ -406,6 +410,16 @@ function resolveEnvVars(value: string, envVars: Record<string, string>): string
|
||||
return resolveEnvVarReferences(value, envVars) as string
|
||||
}
|
||||
|
||||
/** True when any string value in the provider config contains an env-var reference (`{{VAR}}`). */
|
||||
function providerConfigReferencesEnvVars(config: Record<string, unknown>): boolean {
|
||||
for (const value of Object.values(config)) {
|
||||
if (typeof value === 'string' && value.includes('{{')) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
function resolveProviderConfigEnvVars(
|
||||
config: Record<string, unknown>,
|
||||
envVars: Record<string, string>
|
||||
@@ -432,22 +446,30 @@ export async function verifyProviderAuth(
|
||||
rawBody: string,
|
||||
requestId: string
|
||||
): Promise<NextResponse | null> {
|
||||
const handler = getProviderHandler(foundWebhook.provider)
|
||||
const rawProviderConfig = (foundWebhook.providerConfig as Record<string, unknown>) || {}
|
||||
|
||||
/**
|
||||
* Only fetch + decrypt the effective env when there is auth to verify AND the
|
||||
* provider config actually references env vars (`{{VAR}}`). This avoids a DB
|
||||
* read and decryption on the synchronous pre-ack path for the common case.
|
||||
*/
|
||||
let decryptedEnvVars: Record<string, string> = {}
|
||||
try {
|
||||
decryptedEnvVars = await getEffectiveDecryptedEnv(
|
||||
foundWorkflow.userId,
|
||||
foundWorkflow.workspaceId
|
||||
)
|
||||
} catch (error) {
|
||||
logger.error(`[${requestId}] Failed to fetch environment variables`, {
|
||||
error,
|
||||
})
|
||||
if (handler.verifyAuth && providerConfigReferencesEnvVars(rawProviderConfig)) {
|
||||
try {
|
||||
decryptedEnvVars = await getEffectiveDecryptedEnv(
|
||||
foundWorkflow.userId,
|
||||
foundWorkflow.workspaceId
|
||||
)
|
||||
} catch (error) {
|
||||
logger.error(`[${requestId}] Failed to fetch environment variables`, {
|
||||
error,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
const rawProviderConfig = (foundWebhook.providerConfig as Record<string, unknown>) || {}
|
||||
const providerConfig = resolveProviderConfigEnvVars(rawProviderConfig, decryptedEnvVars)
|
||||
|
||||
const handler = getProviderHandler(foundWebhook.provider)
|
||||
if (handler.verifyAuth) {
|
||||
const authResult = await handler.verifyAuth({
|
||||
webhook: foundWebhook,
|
||||
@@ -611,6 +633,10 @@ export async function queueWebhookExecution(
|
||||
blockId: foundWebhook.blockId,
|
||||
workspaceId: foundWorkflow.workspaceId,
|
||||
...(credentialId ? { credentialId } : {}),
|
||||
...(options.receivedAt !== undefined ? { webhookReceivedAt: options.receivedAt } : {}),
|
||||
...(options.triggerTimestampMs !== undefined
|
||||
? { triggerTimestampMs: options.triggerTimestampMs }
|
||||
: {}),
|
||||
}
|
||||
|
||||
const isPolling = isPollingWebhookProvider(payload.provider)
|
||||
|
||||
Reference in New Issue
Block a user