refactor(tiktok): route webhooks by account key

Author: Bill Leoutsakos

apps/sim/app/api/webhooks/tiktok/route.test.ts

@@ -7,13 +7,17 @@ import { requestUtilsMockFns, resetEnvMock, setEnv } from '@sim/testing'
 import { NextRequest } from 'next/server'
 import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { mockEnqueueTikTokWebhookIngress, mockRelease } = vi.hoisted(() => ({
-  mockEnqueueTikTokWebhookIngress: vi.fn(),
-  mockRelease: vi.fn(),
-}))
+const { mockDispatchResolvedWebhookTarget, mockFindWebhooksByRoutingKey, mockRelease } = vi.hoisted(
+  () => ({
+    mockDispatchResolvedWebhookTarget: vi.fn(),
+    mockFindWebhooksByRoutingKey: vi.fn(),
+    mockRelease: vi.fn(),
+  })
+)
 
-vi.mock('@/background/tiktok-webhook-ingress', () => ({
-  enqueueTikTokWebhookIngress: mockEnqueueTikTokWebhookIngress,
+vi.mock('@/lib/webhooks/processor', () => ({
+  dispatchResolvedWebhookTarget: mockDispatchResolvedWebhookTarget,
+  findWebhooksByRoutingKey: mockFindWebhooksByRoutingKey,
 }))
 
 vi.mock('@/lib/core/admission/gate', () => ({
@@ -29,12 +33,17 @@ vi.mock('@/lib/core/utils/with-route-handler', () => ({
 
 import { POST } from '@/app/api/webhooks/tiktok/route'
 
-function signedRequest(overrides?: { clientKey?: string }): NextRequest {
+const target = (id: string) => ({
+  webhook: { id, path: null, provider: 'tiktok' },
+  workflow: { id: `workflow-${id}` },
+})
+
+function signedRequest(overrides?: { clientKey?: string; userOpenId?: string }): NextRequest {
   const body = JSON.stringify({
     client_key: overrides?.clientKey ?? 'client-key',
     event: 'post.publish.complete',
     create_time: 1_725_000_000,
-    user_openid: 'act.user',
+    user_openid: overrides?.userOpenId ?? 'act.user',
     content: '{"publish_id":"publish-1"}',
   })
   const timestamp = String(Math.floor(Date.now() / 1000))
@@ -53,38 +62,78 @@ function signedRequest(overrides?: { clientKey?: string }): NextRequest {
   })
 }
 
-describe('TikTok webhook ingress route', () => {
+describe('TikTok app webhook route', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     setEnv({ TIKTOK_CLIENT_ID: 'client-key', TIKTOK_CLIENT_SECRET: 'client-secret' })
     requestUtilsMockFns.mockGenerateRequestId.mockReturnValue('request-1')
-    mockEnqueueTikTokWebhookIngress.mockResolvedValue('ingress-job-1')
+    mockFindWebhooksByRoutingKey.mockResolvedValue([])
+    mockDispatchResolvedWebhookTarget.mockResolvedValue({ outcome: 'queued', reason: 'queued' })
   })
 
   afterAll(() => {
     resetEnvMock()
     requestUtilsMockFns.mockGenerateRequestId.mockReset()
   })
 
-  it('returns 200 only after the verified delivery is accepted by the job queue', async () => {
-    const response = await POST(signedRequest())
+  it('routes a verified delivery by user_openid on the TikTok provider', async () => {
+    mockFindWebhooksByRoutingKey.mockResolvedValue([target('webhook-1')])
+
+    const response = await POST(signedRequest({ userOpenId: 'user-open-id' }))
 
     expect(response.status).toBe(200)
     await expect(response.json()).resolves.toEqual({ ok: true })
-    expect(mockEnqueueTikTokWebhookIngress).toHaveBeenCalledWith(
+    expect(mockFindWebhooksByRoutingKey).toHaveBeenCalledWith('user-open-id', 'request-1', 'tiktok')
+    expect(mockDispatchResolvedWebhookTarget).toHaveBeenCalledWith(
+      expect.objectContaining({ id: 'webhook-1' }),
+      expect.objectContaining({ id: 'workflow-webhook-1' }),
+      expect.objectContaining({ user_openid: 'user-open-id' }),
+      expect.any(NextRequest),
       expect.objectContaining({
-        envelope: expect.objectContaining({
-          client_key: 'client-key',
-          user_openid: 'act.user',
-        }),
         requestId: 'request-1',
+        triggerTimestampMs: 1_725_000_000_000,
       })
     )
     expect(mockRelease).toHaveBeenCalledOnce()
   })
 
-  it('returns 503 when durable acceptance fails so TikTok retries', async () => {
-    mockEnqueueTikTokWebhookIngress.mockRejectedValue(new Error('queue unavailable'))
+  it('acknowledges a verified delivery when no workflow targets match', async () => {
+    const response = await POST(signedRequest())
+
+    expect(response.status).toBe(200)
+    expect(mockDispatchResolvedWebhookTarget).not.toHaveBeenCalled()
+  })
+
+  it('dispatches matching workflows sequentially', async () => {
+    mockFindWebhooksByRoutingKey.mockResolvedValue([target('webhook-1'), target('webhook-2')])
+    const order: string[] = []
+    mockDispatchResolvedWebhookTarget.mockImplementation(async (webhook: { id: string }) => {
+      order.push(`start:${webhook.id}`)
+      await Promise.resolve()
+      order.push(`end:${webhook.id}`)
+      return { outcome: 'queued', reason: 'queued' }
+    })
+
+    const response = await POST(signedRequest())
+
+    expect(response.status).toBe(200)
+    expect(order).toEqual(['start:webhook-1', 'end:webhook-1', 'start:webhook-2', 'end:webhook-2'])
+  })
+
+  it('acknowledges provider-local dispatch failures like the Slack app route', async () => {
+    mockFindWebhooksByRoutingKey.mockResolvedValue([target('webhook-1')])
+    mockDispatchResolvedWebhookTarget.mockResolvedValue({
+      outcome: 'failed',
+      reason: 'queue-failed',
+    })
+
+    const response = await POST(signedRequest())
+
+    expect(response.status).toBe(200)
+  })
+
+  it('returns 503 when target lookup fails', async () => {
+    mockFindWebhooksByRoutingKey.mockRejectedValue(new Error('database unavailable'))
 
     const response = await POST(signedRequest())
 
@@ -96,6 +145,6 @@ describe('TikTok webhook ingress route', () => {
     const response = await POST(signedRequest({ clientKey: 'other-client-key' }))
 
     expect(response.status).toBe(401)
-    expect(mockEnqueueTikTokWebhookIngress).not.toHaveBeenCalled()
+    expect(mockFindWebhooksByRoutingKey).not.toHaveBeenCalled()
   })
 })

apps/sim/app/api/webhooks/tiktok/route.ts

@@ -12,13 +12,10 @@ import {
 } from '@/lib/core/utils/stream-limits'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { WEBHOOK_MAX_BODY_BYTES } from '@/lib/webhooks/constants'
+import { dispatchResolvedWebhookTarget, findWebhooksByRoutingKey } from '@/lib/webhooks/processor'
 import { verifyTikTokSignature } from '@/lib/webhooks/providers/tiktok'
-import {
-  enqueueTikTokWebhookIngress,
-  type TikTokWebhookIngressPayload,
-} from '@/background/tiktok-webhook-ingress'
 
-const logger = createLogger('TikTokWebhookIngress')
+const logger = createLogger('TikTokAppWebhookAPI')
 
 const TIKTOK_BODY_LABEL = 'TikTok webhook body'
 
@@ -38,7 +35,7 @@ async function readTikTokBody(req: Request): Promise<string> {
 /**
  * App-level TikTok webhook Callback URL.
  * Portal: `{APP_URL}/api/webhooks/tiktok` (e.g. https://www.sim.ai/api/webhooks/tiktok).
- * Verifies TikTok-Signature and durably accepts the delivery before background target fanout.
+ * Verifies TikTok-Signature and routes the delivery by TikTok `user_openid`.
  */
 export const POST = withRouteHandler(async (request: NextRequest) => {
   const ticket = tryAdmit()
@@ -96,25 +93,27 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
       return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
     }
 
-    const payload: TikTokWebhookIngressPayload = {
-      envelope,
-      headers: {
-        'content-type': request.headers.get('content-type') ?? 'application/json',
-      },
-      requestId,
-      receivedAt,
+    const webhooks = await findWebhooksByRoutingKey(envelope.user_openid, requestId, 'tiktok')
+    let dispatched = 0
+    for (const { webhook, workflow } of webhooks) {
+      const result = await dispatchResolvedWebhookTarget(webhook, workflow, envelope, request, {
+        requestId,
+        receivedAt,
+        triggerTimestampMs: envelope.create_time * 1000,
+      })
+      if (result.outcome === 'queued') dispatched += 1
     }
-    const jobId = await enqueueTikTokWebhookIngress(payload)
 
-    logger.info(`[${requestId}] Accepted TikTok webhook delivery`, {
+    logger.info(`[${requestId}] Processed TikTok webhook delivery`, {
+      dispatched,
       event: envelope.event,
-      jobId,
+      targetCount: webhooks.length,
       userOpenIdPrefix: envelope.user_openid.slice(0, 12),
     })
 
     return NextResponse.json({ ok: true })
   } catch (error) {
-    logger.error(`[${requestId}] TikTok webhook ingress error`, {
+    logger.error(`[${requestId}] TikTok webhook processing error`, {
       error: getErrorMessage(error, 'Unknown error'),
     })
     return NextResponse.json({ error: 'Temporarily unable to accept webhook' }, { status: 503 })

apps/sim/lib/webhooks/deploy.test.ts

@@ -92,6 +92,15 @@ const slackTrigger = trigger([
   },
 ])
 
+const tiktokTrigger = trigger([
+  {
+    id: 'triggerCredentials',
+    mode: 'trigger',
+    serviceId: 'tiktok',
+    required: true,
+  },
+])
+
 function makeBlock(
   type: string,
   subBlockValues: Record<string, unknown>,
@@ -359,3 +368,63 @@ describe('resolveWebhookConfigForBlock — slack_oauth routing', () => {
     expect(mockFetchSlackTeamId).not.toHaveBeenCalled()
   })
 })
+
+describe('resolveWebhookConfigForBlock — TikTok routing', () => {
+  const tiktokTriggerDef = {
+    provider: 'tiktok',
+    name: 'TikTok',
+    subBlocks: tiktokTrigger.subBlocks,
+  }
+
+  function resolveTikTok(
+    credentialReference = 'credential-1',
+    workflow: Record<string, unknown> = { workspaceId: 'ws-1' }
+  ) {
+    ;(getBlock as unknown as Mock).mockReturnValue({ category: 'triggers' })
+    ;(getTrigger as unknown as Mock).mockReturnValue(tiktokTriggerDef)
+    return resolveWebhookConfigForBlock({
+      block: makeBlock('tiktok', { triggerCredentials: credentialReference }),
+      workflow,
+      userId: 'deployer-1',
+      requestId: 'req-1',
+    })
+  }
+
+  it('routes a canonical workspace credential by its TikTok open_id', async () => {
+    queueTableRows(credential, [{ id: 'credential-1' }])
+    mockResolveOAuthAccountId.mockResolvedValue({ accountId: 'account-1' })
+    queueTableRows(account, [
+      { accountId: 'open-id-with-hyphens-12345678-1234-1234-1234-123456789abc' },
+    ])
+
+    const result = await resolveTikTok()
+
+    expect(result?.success).toBe(true)
+    if (!result?.success) throw new Error('expected success')
+    expect(result.config.provider).toBe('tiktok')
+    expect(result.config.routingKey).toBe('open-id-with-hyphens')
+    expect(result.config.triggerPath).toBeNull()
+    expect(result.config.providerConfig.credentialId).toBe('credential-1')
+  })
+
+  it('rejects a TikTok credential not available in the workflow workspace', async () => {
+    const result = await resolveTikTok('foreign-credential')
+
+    expect(result?.success).toBe(false)
+    if (result?.success) throw new Error('expected failure')
+    expect(result?.error.message).toContain('not available in this workspace')
+    expect(mockResolveOAuthAccountId).not.toHaveBeenCalled()
+  })
+
+  it('rejects a malformed TikTok account identity', async () => {
+    queueTableRows(credential, [{ id: 'credential-1' }])
+    mockResolveOAuthAccountId.mockResolvedValue({ accountId: 'account-1' })
+    queueTableRows(account, [{ accountId: 'missing-generated-uuid' }])
+
+    const result = await resolveTikTok()
+
+    expect(result?.success).toBe(false)
+    if (result?.success) throw new Error('expected failure')
+    expect(result?.error.message).toContain('Reconnect')
+  })
+})

apps/sim/lib/webhooks/deploy.ts

@@ -40,6 +40,7 @@ import { SYSTEM_SUBBLOCK_IDS } from '@/triggers/constants'
 import { SIM_SUBSCRIBED_EVENTS } from '@/triggers/slack/shared'
 
 const logger = createLogger('DeployWebhookSync')
+const TIKTOK_ACCOUNT_UUID_SUFFIX = /-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i
 
 interface TriggerSaveError {
   message: string
@@ -362,9 +363,8 @@ export async function resolveTriggerCredentialId(
 }
 
 /**
- * Resolves a trigger block to its persisted webhook config, including the
- * Slack-specific routing branch. Exported for unit testing that branch; not part
- * of the public deploy API.
+ * Resolves a trigger block to its persisted webhook config, including app-level
+ * provider routing. Exported for focused unit testing; not part of the public deploy API.
  */
 export async function resolveWebhookConfigForBlock(input: {
   block: BlockState
@@ -565,6 +565,36 @@ export async function resolveWebhookConfigForBlock(input: {
       // (`slack_app`) rows on providerConfig.credentialId.
       providerConfig.credentialId = resolvedCredentialId
     }
+  } else if (triggerDef.provider === 'tiktok') {
+    if (!credentialId) {
+      return {
+        success: false,
+        error: { message: 'Select a TikTok account for the trigger.', status: 400 },
+      }
+    }
+
+    const resolvedAccount = await resolveOAuthAccountId(credentialId)
+    const [tiktokAccount] = resolvedAccount?.accountId
+      ? await db
+          .select({ accountId: account.accountId })
+          .from(account)
+          .where(and(eq(account.id, resolvedAccount.accountId), eq(account.providerId, 'tiktok')))
+          .limit(1)
+      : []
+    const openId = tiktokAccount?.accountId.replace(TIKTOK_ACCOUNT_UUID_SUFFIX, '')
+
+    if (!openId || openId === tiktokAccount?.accountId) {
+      return {
+        success: false,
+        error: {
+          message: 'Could not verify the connected TikTok account. Reconnect it and try again.',
+          status: 400,
+        },
+      }
+    }
+
+    effectivePath = null
+    routingKey = openId
   }
 
   return {