mirror of
https://github.com/simstudioai/sim.git
synced 2026-10-30 19:32:46 +08:00
refactor(tiktok): align webhook routing with shared dispatcher (#6261)
* refactor(tiktok): route webhooks by account key * refactor(tiktok): remove custom webhook ingress jobs * fix(tiktok): constrain webhook routing migration * test(tiktok): remove redundant assertions * fix(tiktok): preserve routing backfill state * fix(tiktok): preserve legacy webhook routing during rollout * fix(tiktok): preserve webhook delivery guarantees * refactor(tiktok): remove rollout compatibility * refactor(tiktok): remove unused webhook backfill --------- Co-authored-by: Bill Leoutsakos <billleoutsakos@Bills-MacBook-Pro.local>
This commit is contained in:
Bill Leoutsakos
co-authored by
Bill Leoutsakos
Bill Leoutsakos
parent
9883543eb2
commit
fb6e18f5db
@@ -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,12 +62,13 @@ 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(() => {
|
||||
@@ -66,25 +76,64 @@ describe('TikTok webhook ingress route', () => {
|
||||
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('returns a retryable response when a target cannot be dispatched', async () => {
|
||||
mockFindWebhooksByRoutingKey.mockResolvedValue([target('webhook-1')])
|
||||
mockDispatchResolvedWebhookTarget.mockResolvedValue({
|
||||
outcome: 'failed',
|
||||
reason: 'queue-failed',
|
||||
})
|
||||
|
||||
const response = await POST(signedRequest())
|
||||
|
||||
expect(response.status).toBe(503)
|
||||
})
|
||||
|
||||
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()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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,34 @@ 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
|
||||
let failed = 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
|
||||
if (result.outcome === 'failed') failed += 1
|
||||
}
|
||||
const jobId = await enqueueTikTokWebhookIngress(payload)
|
||||
|
||||
logger.info(`[${requestId}] Accepted TikTok webhook delivery`, {
|
||||
logger.info(`[${requestId}] Processed TikTok webhook delivery`, {
|
||||
dispatched,
|
||||
failed,
|
||||
event: envelope.event,
|
||||
jobId,
|
||||
targetCount: webhooks.length,
|
||||
userOpenIdPrefix: envelope.user_openid.slice(0, 12),
|
||||
})
|
||||
|
||||
if (failed > 0) {
|
||||
return NextResponse.json({ error: 'Temporarily unable to accept webhook' }, { status: 503 })
|
||||
}
|
||||
|
||||
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 })
|
||||
|
||||
@@ -1,173 +0,0 @@
|
||||
/**
|
||||
* @vitest-environment node
|
||||
*/
|
||||
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const { mockDispatchResolvedWebhookTarget, mockEnqueue, mockFindTikTokWebhookTargetPage } =
|
||||
vi.hoisted(() => ({
|
||||
mockDispatchResolvedWebhookTarget: vi.fn(),
|
||||
mockEnqueue: vi.fn(),
|
||||
mockFindTikTokWebhookTargetPage: vi.fn(),
|
||||
}))
|
||||
|
||||
vi.mock('@trigger.dev/sdk', () => ({
|
||||
task: vi.fn((config: unknown) => config),
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/webhooks/processor', () => ({
|
||||
dispatchResolvedWebhookTarget: mockDispatchResolvedWebhookTarget,
|
||||
}))
|
||||
|
||||
vi.mock('@/background/tiktok-webhook-targets', () => ({
|
||||
findTikTokWebhookTargetPage: mockFindTikTokWebhookTargetPage,
|
||||
}))
|
||||
vi.mock('@/lib/core/async-jobs', () => ({
|
||||
getJobQueue: vi.fn(async () => ({ enqueue: mockEnqueue })),
|
||||
}))
|
||||
|
||||
import {
|
||||
enqueueTikTokWebhookIngress,
|
||||
executeTikTokWebhookIngress,
|
||||
type TikTokWebhookIngressPayload,
|
||||
} from '@/background/tiktok-webhook-ingress'
|
||||
|
||||
const payload: TikTokWebhookIngressPayload = {
|
||||
envelope: {
|
||||
client_key: 'client-key',
|
||||
event: 'post.publish.complete',
|
||||
create_time: 1_725_000_000,
|
||||
user_openid: 'act.user',
|
||||
content: '{"publish_id":"publish-1"}',
|
||||
},
|
||||
headers: { 'content-type': 'application/json' },
|
||||
requestId: 'request-1',
|
||||
receivedAt: 1_725_000_000_000,
|
||||
}
|
||||
|
||||
const firstTarget = {
|
||||
webhook: { id: 'webhook-1', path: 'tiktok', provider: 'tiktok' },
|
||||
workflow: { id: 'workflow-1' },
|
||||
}
|
||||
const secondTarget = {
|
||||
webhook: { id: 'webhook-2', path: 'tiktok', provider: 'tiktok' },
|
||||
workflow: { id: 'workflow-2' },
|
||||
}
|
||||
|
||||
describe('executeTikTokWebhookIngress', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
mockEnqueue.mockResolvedValue('ingress-job-1')
|
||||
})
|
||||
|
||||
it('acknowledges deliveries without active targets', async () => {
|
||||
mockFindTikTokWebhookTargetPage.mockResolvedValue({
|
||||
hasMore: false,
|
||||
nextCursor: null,
|
||||
targets: [],
|
||||
})
|
||||
|
||||
await expect(executeTikTokWebhookIngress(payload)).resolves.toEqual({
|
||||
ignored: 0,
|
||||
processed: 0,
|
||||
targetCount: 0,
|
||||
})
|
||||
expect(mockDispatchResolvedWebhookTarget).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('dispatches one bounded page and returns its continuation cursor', async () => {
|
||||
const events: string[] = []
|
||||
mockFindTikTokWebhookTargetPage.mockImplementationOnce(async () => {
|
||||
events.push('page-1')
|
||||
return {
|
||||
hasMore: true,
|
||||
nextCursor: 'webhook-1',
|
||||
targets: [firstTarget],
|
||||
}
|
||||
})
|
||||
mockDispatchResolvedWebhookTarget.mockImplementationOnce(async () => {
|
||||
events.push('dispatch-1')
|
||||
return {
|
||||
outcome: 'queued',
|
||||
reason: 'queued',
|
||||
response: new Response(null, { status: 200 }),
|
||||
}
|
||||
})
|
||||
|
||||
await expect(executeTikTokWebhookIngress(payload)).resolves.toEqual({
|
||||
ignored: 0,
|
||||
nextCursor: 'webhook-1',
|
||||
processed: 1,
|
||||
targetCount: 1,
|
||||
})
|
||||
expect(events).toEqual(['page-1', 'dispatch-1'])
|
||||
expect(mockFindTikTokWebhookTargetPage).toHaveBeenNthCalledWith(
|
||||
1,
|
||||
'act.user',
|
||||
'request-1',
|
||||
undefined
|
||||
)
|
||||
})
|
||||
|
||||
it('enqueues the next page only after the current page succeeds', async () => {
|
||||
mockFindTikTokWebhookTargetPage.mockResolvedValue({
|
||||
hasMore: true,
|
||||
nextCursor: 'webhook-1',
|
||||
targets: [firstTarget],
|
||||
})
|
||||
mockDispatchResolvedWebhookTarget.mockResolvedValue({
|
||||
outcome: 'queued',
|
||||
reason: 'queued',
|
||||
response: new Response(null, { status: 200 }),
|
||||
})
|
||||
|
||||
await enqueueTikTokWebhookIngress(payload)
|
||||
const options = mockEnqueue.mock.calls[0][2] as { runner: () => Promise<void> }
|
||||
await options.runner()
|
||||
|
||||
expect(mockEnqueue).toHaveBeenNthCalledWith(
|
||||
2,
|
||||
'tiktok-webhook-ingress',
|
||||
expect.objectContaining({ afterWebhookId: 'webhook-1' }),
|
||||
expect.objectContaining({
|
||||
jobId: 'tiktok-webhook-ingress:request-1:webhook-1',
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
it('throws after a target failure so the durable ingress job retries', async () => {
|
||||
mockFindTikTokWebhookTargetPage.mockResolvedValueOnce({
|
||||
hasMore: false,
|
||||
nextCursor: 'webhook-2',
|
||||
targets: [firstTarget, secondTarget],
|
||||
})
|
||||
mockDispatchResolvedWebhookTarget
|
||||
.mockResolvedValueOnce({
|
||||
outcome: 'failed',
|
||||
reason: 'queue-failed',
|
||||
response: new Response(null, { status: 500 }),
|
||||
})
|
||||
.mockResolvedValueOnce({
|
||||
outcome: 'queued',
|
||||
reason: 'queued',
|
||||
response: new Response(null, { status: 200 }),
|
||||
})
|
||||
|
||||
await expect(executeTikTokWebhookIngress(payload)).rejects.toThrow(
|
||||
'Failed to dispatch 1 of 2 TikTok webhook targets'
|
||||
)
|
||||
expect(mockDispatchResolvedWebhookTarget).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('fails closed if a full target page does not provide a forward cursor', async () => {
|
||||
mockFindTikTokWebhookTargetPage.mockResolvedValue({
|
||||
hasMore: true,
|
||||
nextCursor: null,
|
||||
targets: [],
|
||||
})
|
||||
|
||||
await expect(executeTikTokWebhookIngress(payload)).rejects.toThrow(
|
||||
'TikTok webhook target pagination did not advance'
|
||||
)
|
||||
})
|
||||
})
|
||||
@@ -1,148 +0,0 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { task } from '@trigger.dev/sdk'
|
||||
import { NextRequest } from 'next/server'
|
||||
import type { TikTokWebhookEnvelope } from '@/lib/api/contracts/webhooks'
|
||||
import { getJobQueue } from '@/lib/core/async-jobs'
|
||||
import { dispatchResolvedWebhookTarget } from '@/lib/webhooks/processor'
|
||||
import { findTikTokWebhookTargetPage } from '@/background/tiktok-webhook-targets'
|
||||
|
||||
const logger = createLogger('TikTokWebhookIngressTask')
|
||||
|
||||
export const TIKTOK_WEBHOOK_INGRESS_CONCURRENCY_LIMIT = 50
|
||||
export const TIKTOK_WEBHOOK_INGRESS_MAX_ATTEMPTS = 3
|
||||
|
||||
export interface TikTokWebhookIngressPayload {
|
||||
afterWebhookId?: string
|
||||
envelope: TikTokWebhookEnvelope
|
||||
headers: {
|
||||
'content-type': string
|
||||
}
|
||||
requestId: string
|
||||
receivedAt: number
|
||||
}
|
||||
|
||||
export interface TikTokWebhookIngressResult {
|
||||
ignored: number
|
||||
nextCursor?: string
|
||||
processed: number
|
||||
targetCount: number
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves and dispatches one fixed-size keyset page for a verified TikTok delivery. Each page is a
|
||||
* separate durable job, so retries replay at most one bounded page and successful pages continue
|
||||
* from their last webhook ID.
|
||||
*/
|
||||
export async function executeTikTokWebhookIngress(
|
||||
payload: TikTokWebhookIngressPayload
|
||||
): Promise<TikTokWebhookIngressResult> {
|
||||
const request = new NextRequest('http://internal/api/webhooks/tiktok', {
|
||||
method: 'POST',
|
||||
headers: payload.headers,
|
||||
body: JSON.stringify(payload.envelope),
|
||||
})
|
||||
|
||||
const page = await findTikTokWebhookTargetPage(
|
||||
payload.envelope.user_openid,
|
||||
payload.requestId,
|
||||
payload.afterWebhookId
|
||||
)
|
||||
const nextCursor = page.hasMore ? page.nextCursor : null
|
||||
if (page.hasMore && (!nextCursor || nextCursor === payload.afterWebhookId)) {
|
||||
throw new Error('TikTok webhook target pagination did not advance')
|
||||
}
|
||||
|
||||
let ignored = 0
|
||||
let processed = 0
|
||||
let failed = 0
|
||||
const targetCount = page.targets.length
|
||||
|
||||
for (const { webhook, workflow } of page.targets) {
|
||||
const result = await dispatchResolvedWebhookTarget(
|
||||
webhook,
|
||||
workflow,
|
||||
payload.envelope,
|
||||
request,
|
||||
{
|
||||
requestId: payload.requestId,
|
||||
path: webhook.path ?? undefined,
|
||||
receivedAt: payload.receivedAt,
|
||||
triggerTimestampMs: payload.envelope.create_time * 1000,
|
||||
}
|
||||
)
|
||||
|
||||
if (result.outcome === 'queued') {
|
||||
processed += 1
|
||||
} else if (result.outcome === 'ignored') {
|
||||
ignored += 1
|
||||
} else {
|
||||
failed += 1
|
||||
}
|
||||
}
|
||||
|
||||
if (failed > 0) {
|
||||
throw new Error(`Failed to dispatch ${failed} of ${targetCount} TikTok webhook targets`)
|
||||
}
|
||||
|
||||
if (targetCount === 0) {
|
||||
logger.info(`[${payload.requestId}] No TikTok webhook targets found in page`, {
|
||||
event: payload.envelope.event,
|
||||
userOpenIdPrefix: payload.envelope.user_openid.slice(0, 12),
|
||||
})
|
||||
return {
|
||||
ignored: 0,
|
||||
processed: 0,
|
||||
targetCount: 0,
|
||||
...(nextCursor ? { nextCursor } : {}),
|
||||
}
|
||||
}
|
||||
|
||||
logger.info(`[${payload.requestId}] TikTok webhook fanout page completed`, {
|
||||
event: payload.envelope.event,
|
||||
ignored,
|
||||
nextCursor,
|
||||
processed,
|
||||
targetCount,
|
||||
})
|
||||
|
||||
return { ignored, processed, targetCount, ...(nextCursor ? { nextCursor } : {}) }
|
||||
}
|
||||
|
||||
async function runTikTokWebhookIngressJob(payload: TikTokWebhookIngressPayload): Promise<void> {
|
||||
const result = await executeTikTokWebhookIngress(payload)
|
||||
if (!result.nextCursor) return
|
||||
|
||||
await enqueueTikTokWebhookIngress({
|
||||
...payload,
|
||||
afterWebhookId: result.nextCursor,
|
||||
})
|
||||
}
|
||||
|
||||
/** Enqueues one bounded TikTok webhook fanout page with stable continuation identity. */
|
||||
export async function enqueueTikTokWebhookIngress(
|
||||
payload: TikTokWebhookIngressPayload
|
||||
): Promise<string> {
|
||||
const jobQueue = await getJobQueue()
|
||||
return jobQueue.enqueue('tiktok-webhook-ingress', payload, {
|
||||
jobId: `tiktok-webhook-ingress:${payload.requestId}:${payload.afterWebhookId ?? 'root'}`,
|
||||
maxAttempts: TIKTOK_WEBHOOK_INGRESS_MAX_ATTEMPTS,
|
||||
concurrencyKey: 'tiktok-webhook-ingress',
|
||||
concurrencyLimit: TIKTOK_WEBHOOK_INGRESS_CONCURRENCY_LIMIT,
|
||||
runner: async () => runTikTokWebhookIngressJob(payload),
|
||||
})
|
||||
}
|
||||
|
||||
export const tiktokWebhookIngressTask = task({
|
||||
id: 'tiktok-webhook-ingress',
|
||||
machine: 'small-1x',
|
||||
retry: {
|
||||
maxAttempts: TIKTOK_WEBHOOK_INGRESS_MAX_ATTEMPTS,
|
||||
factor: 2,
|
||||
minTimeoutInMs: 1000,
|
||||
maxTimeoutInMs: 10_000,
|
||||
},
|
||||
queue: {
|
||||
concurrencyLimit: TIKTOK_WEBHOOK_INGRESS_CONCURRENCY_LIMIT,
|
||||
},
|
||||
run: async (payload: TikTokWebhookIngressPayload) => runTikTokWebhookIngressJob(payload),
|
||||
})
|
||||
@@ -1,181 +0,0 @@
|
||||
/**
|
||||
* @vitest-environment node
|
||||
*/
|
||||
|
||||
import { dbChainMock, dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing'
|
||||
import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const { mockAsc, mockCredentialExpression, mockEq, mockGt, mockLike, tables } = vi.hoisted(() => ({
|
||||
mockAsc: vi.fn((value: unknown) => ({ asc: value })),
|
||||
mockCredentialExpression: vi.fn(() => 'webhook.credentialId'),
|
||||
mockEq: vi.fn((left: unknown, right: unknown) => ({ left, right })),
|
||||
mockGt: vi.fn((left: unknown, right: unknown) => ({ gt: [left, right] })),
|
||||
mockLike: vi.fn((left: unknown, right: unknown) => ({ left, right })),
|
||||
/** Table-qualified column names keep the eq/like assertions unambiguous. */
|
||||
tables: {
|
||||
account: {
|
||||
id: 'account.id',
|
||||
accountId: 'account.accountId',
|
||||
providerId: 'account.providerId',
|
||||
},
|
||||
credential: {
|
||||
id: 'credential.id',
|
||||
accountId: 'credential.accountId',
|
||||
providerId: 'credential.providerId',
|
||||
type: 'credential.type',
|
||||
workspaceId: 'credential.workspaceId',
|
||||
},
|
||||
webhook: {
|
||||
deploymentVersionId: 'webhook.deploymentVersionId',
|
||||
isActive: 'webhook.isActive',
|
||||
id: 'webhook.id',
|
||||
archivedAt: 'webhook.archivedAt',
|
||||
provider: 'webhook.provider',
|
||||
providerConfig: 'webhook.providerConfig',
|
||||
workflowId: 'webhook.workflowId',
|
||||
},
|
||||
workflow: {
|
||||
id: 'workflow.id',
|
||||
workspaceId: 'workflow.workspaceId',
|
||||
archivedAt: 'workflow.archivedAt',
|
||||
},
|
||||
workflowDeploymentVersion: {
|
||||
id: 'workflowDeploymentVersion.id',
|
||||
workflowId: 'workflowDeploymentVersion.workflowId',
|
||||
isActive: 'workflowDeploymentVersion.isActive',
|
||||
},
|
||||
},
|
||||
}))
|
||||
|
||||
vi.mock('@sim/db', () => ({
|
||||
...dbChainMock,
|
||||
...tables,
|
||||
webhookCredentialIdExpression: mockCredentialExpression,
|
||||
}))
|
||||
|
||||
vi.mock('drizzle-orm', () => ({
|
||||
and: vi.fn((...conditions: unknown[]) => conditions),
|
||||
asc: mockAsc,
|
||||
eq: mockEq,
|
||||
gt: mockGt,
|
||||
isNull: vi.fn((value: unknown) => ({ isNull: value })),
|
||||
like: mockLike,
|
||||
or: vi.fn((...conditions: unknown[]) => conditions),
|
||||
}))
|
||||
|
||||
import {
|
||||
findTikTokWebhookTargetPage,
|
||||
TIKTOK_WEBHOOK_TARGET_PAGE_SIZE,
|
||||
} from '@/background/tiktok-webhook-targets'
|
||||
|
||||
const ACCOUNT_UUID = '11111111-2222-3333-4444-555555555555'
|
||||
|
||||
/** Queues one page of joined rows, clipped like the SQL LIMIT would. */
|
||||
function queuePageRows(
|
||||
rows: Array<{
|
||||
accountId: string
|
||||
webhookId: string
|
||||
webhook: Record<string, unknown>
|
||||
workflow: Record<string, unknown>
|
||||
}>
|
||||
) {
|
||||
queueTableRows(tables.account, rows.slice(0, TIKTOK_WEBHOOK_TARGET_PAGE_SIZE))
|
||||
}
|
||||
|
||||
describe('findTikTokWebhookTargetPage', () => {
|
||||
afterAll(() => {
|
||||
resetDbChainMock()
|
||||
})
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
resetDbChainMock()
|
||||
})
|
||||
|
||||
it('returns only rows whose stored account ID exactly matches user_openid', async () => {
|
||||
queuePageRows([
|
||||
{
|
||||
accountId: `act.user-${ACCOUNT_UUID}`,
|
||||
webhookId: 'webhook-1',
|
||||
webhook: { id: 'webhook-1' },
|
||||
workflow: { id: 'workflow-1' },
|
||||
},
|
||||
{
|
||||
accountId: `act.user-other-${ACCOUNT_UUID}`,
|
||||
webhookId: 'webhook-2',
|
||||
webhook: { id: 'webhook-2' },
|
||||
workflow: { id: 'workflow-2' },
|
||||
},
|
||||
])
|
||||
|
||||
const page = await findTikTokWebhookTargetPage('act.user', 'request-1')
|
||||
|
||||
expect(page).toEqual({
|
||||
hasMore: false,
|
||||
nextCursor: 'webhook-2',
|
||||
targets: [
|
||||
{
|
||||
webhook: { id: 'webhook-1' },
|
||||
workflow: { id: 'workflow-1' },
|
||||
},
|
||||
],
|
||||
})
|
||||
})
|
||||
|
||||
it('enforces provider and workspace bindings in the database query', async () => {
|
||||
await findTikTokWebhookTargetPage('act.user', 'request-2')
|
||||
|
||||
expect(mockEq).toHaveBeenCalledWith('credential.providerId', 'tiktok')
|
||||
expect(mockEq).toHaveBeenCalledWith('webhook.provider', 'tiktok')
|
||||
expect(mockEq).toHaveBeenCalledWith('workflow.workspaceId', 'credential.workspaceId')
|
||||
expect(mockCredentialExpression).toHaveBeenCalledWith('webhook.providerConfig')
|
||||
expect(mockEq).toHaveBeenCalledWith('webhook.credentialId', 'credential.id')
|
||||
})
|
||||
|
||||
it('uses a fixed-size webhook ID keyset in ascending order', async () => {
|
||||
await findTikTokWebhookTargetPage('act.user', 'request-3', 'webhook-100')
|
||||
|
||||
expect(mockGt).toHaveBeenCalledWith('webhook.id', 'webhook-100')
|
||||
expect(mockAsc).toHaveBeenCalledWith('webhook.id')
|
||||
expect(dbChainMockFns.orderBy).toHaveBeenCalledWith({ asc: 'webhook.id' })
|
||||
expect(dbChainMockFns.limit).toHaveBeenCalledWith(TIKTOK_WEBHOOK_TARGET_PAGE_SIZE)
|
||||
})
|
||||
|
||||
it('returns a continuation cursor when the fixed-size page is full', async () => {
|
||||
queuePageRows(
|
||||
Array.from({ length: TIKTOK_WEBHOOK_TARGET_PAGE_SIZE + 1 }, (_, index) => {
|
||||
const webhookId = `webhook-${String(index).padStart(3, '0')}`
|
||||
return {
|
||||
accountId: `act.user-${ACCOUNT_UUID}`,
|
||||
webhookId,
|
||||
webhook: { id: webhookId },
|
||||
workflow: { id: `workflow-${index}` },
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
const page = await findTikTokWebhookTargetPage('act.user', 'request-4')
|
||||
|
||||
expect(page.targets).toHaveLength(TIKTOK_WEBHOOK_TARGET_PAGE_SIZE)
|
||||
expect(page.hasMore).toBe(true)
|
||||
expect(page.nextCursor).toBe('webhook-099')
|
||||
})
|
||||
|
||||
it('escapes user_openid wildcard characters in the account lookup', async () => {
|
||||
await findTikTokWebhookTargetPage('act_%', 'request-5')
|
||||
|
||||
expect(mockLike).toHaveBeenCalledWith(
|
||||
'account.accountId',
|
||||
'act\\_\\%-________-____-____-____-____________'
|
||||
)
|
||||
})
|
||||
|
||||
it('does not query for an empty user_openid', async () => {
|
||||
expect(await findTikTokWebhookTargetPage('', 'request-6')).toEqual({
|
||||
hasMore: false,
|
||||
nextCursor: null,
|
||||
targets: [],
|
||||
})
|
||||
expect(dbChainMockFns.select).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
@@ -1,122 +0,0 @@
|
||||
import {
|
||||
account,
|
||||
credential,
|
||||
db,
|
||||
webhook,
|
||||
webhookCredentialIdExpression,
|
||||
workflow,
|
||||
workflowDeploymentVersion,
|
||||
} from '@sim/db'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { and, asc, eq, gt, isNull, like, or } from 'drizzle-orm'
|
||||
import { deliverableWebhookPredicate } from '@/lib/webhooks/delivery-predicate'
|
||||
|
||||
const logger = createLogger('TikTokWebhookTargets')
|
||||
const ACCOUNT_ID_UUID_SUFFIX = /-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i
|
||||
const ACCOUNT_ID_UUID_LIKE_SUFFIX = '________-____-____-____-____________'
|
||||
|
||||
export const TIKTOK_WEBHOOK_TARGET_PAGE_SIZE = 100
|
||||
|
||||
export interface TikTokWebhookTarget {
|
||||
webhook: typeof webhook.$inferSelect
|
||||
workflow: typeof workflow.$inferSelect
|
||||
}
|
||||
|
||||
export interface TikTokWebhookTargetPage {
|
||||
hasMore: boolean
|
||||
nextCursor: string | null
|
||||
targets: TikTokWebhookTarget[]
|
||||
}
|
||||
|
||||
function escapeLikePattern(value: string): string {
|
||||
return value.replace(/\\/g, '\\\\').replace(/%/g, '\\%').replace(/_/g, '\\_')
|
||||
}
|
||||
|
||||
function openIdFromAccountId(accountId: string): string {
|
||||
return accountId.replace(ACCOUNT_ID_UUID_SUFFIX, '')
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves one deterministic page of active webhook targets for the TikTok background ingress.
|
||||
* The workflow-workspace equality prevents cross-tenant event routing, while the webhook ID cursor
|
||||
* keeps each query and retained result set bounded without offset drift.
|
||||
*/
|
||||
export async function findTikTokWebhookTargetPage(
|
||||
userOpenId: string,
|
||||
requestId: string,
|
||||
afterWebhookId?: string
|
||||
): Promise<TikTokWebhookTargetPage> {
|
||||
if (!userOpenId) {
|
||||
return { hasMore: false, nextCursor: null, targets: [] }
|
||||
}
|
||||
|
||||
const rows = await db
|
||||
.select({
|
||||
accountId: account.accountId,
|
||||
webhookId: webhook.id,
|
||||
webhook,
|
||||
workflow,
|
||||
})
|
||||
.from(account)
|
||||
.innerJoin(
|
||||
credential,
|
||||
and(
|
||||
eq(credential.accountId, account.id),
|
||||
eq(credential.type, 'oauth'),
|
||||
eq(credential.providerId, 'tiktok')
|
||||
)
|
||||
)
|
||||
.innerJoin(
|
||||
webhook,
|
||||
and(
|
||||
eq(webhookCredentialIdExpression(webhook.providerConfig), credential.id),
|
||||
eq(webhook.provider, 'tiktok'),
|
||||
deliverableWebhookPredicate(webhook)
|
||||
)
|
||||
)
|
||||
.innerJoin(
|
||||
workflow,
|
||||
and(
|
||||
eq(workflow.id, webhook.workflowId),
|
||||
eq(workflow.workspaceId, credential.workspaceId),
|
||||
isNull(workflow.archivedAt)
|
||||
)
|
||||
)
|
||||
.leftJoin(
|
||||
workflowDeploymentVersion,
|
||||
and(
|
||||
eq(workflowDeploymentVersion.workflowId, workflow.id),
|
||||
eq(workflowDeploymentVersion.isActive, true)
|
||||
)
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
eq(account.providerId, 'tiktok'),
|
||||
like(account.accountId, `${escapeLikePattern(userOpenId)}-${ACCOUNT_ID_UUID_LIKE_SUFFIX}`),
|
||||
or(
|
||||
eq(webhook.deploymentVersionId, workflowDeploymentVersion.id),
|
||||
and(isNull(workflowDeploymentVersion.id), isNull(webhook.deploymentVersionId))
|
||||
),
|
||||
afterWebhookId ? gt(webhook.id, afterWebhookId) : undefined
|
||||
)
|
||||
)
|
||||
.orderBy(asc(webhook.id))
|
||||
.limit(TIKTOK_WEBHOOK_TARGET_PAGE_SIZE)
|
||||
|
||||
const targets = rows
|
||||
.filter((row) => openIdFromAccountId(row.accountId) === userOpenId)
|
||||
.map(({ webhook: webhookRecord, workflow: workflowRecord }) => ({
|
||||
webhook: webhookRecord,
|
||||
workflow: workflowRecord,
|
||||
}))
|
||||
const nextCursor = rows.at(-1)?.webhookId ?? null
|
||||
const hasMore = rows.length === TIKTOK_WEBHOOK_TARGET_PAGE_SIZE
|
||||
|
||||
logger.info(`[${requestId}] Resolved TikTok webhook target page`, {
|
||||
hasMore,
|
||||
userOpenIdPrefix: userOpenId.slice(0, 12),
|
||||
targetCount: targets.length,
|
||||
})
|
||||
|
||||
return { hasMore, nextCursor, targets }
|
||||
}
|
||||
@@ -43,10 +43,4 @@ describe('TikTokBlock', () => {
|
||||
expect(finalInputs.file).toEqual(file)
|
||||
expect(finalInputs.videoIds).toBeUndefined()
|
||||
})
|
||||
|
||||
it('declares file-like outputs with canonical block output types', () => {
|
||||
expect(TikTokBlock.outputs.avatarFile.type).toBe('file')
|
||||
expect(TikTokBlock.outputs.videos.type).toBe('array')
|
||||
expect(TikTokBlock.outputs.publiclyAvailablePostId.type).toBe('array')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,38 +0,0 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { tiktokUploadVideoDraftBodySchema } from '@/lib/api/contracts/tiktok-tools'
|
||||
|
||||
const file = {
|
||||
key: 'workspace/test/video.mp4',
|
||||
name: 'video.mp4',
|
||||
size: 1024,
|
||||
type: 'video/mp4',
|
||||
}
|
||||
|
||||
describe('tiktokUploadVideoDraftBodySchema', () => {
|
||||
it('accepts a draft upload body', () => {
|
||||
const parsed = tiktokUploadVideoDraftBodySchema.safeParse({
|
||||
accessToken: 'token',
|
||||
file,
|
||||
})
|
||||
expect(parsed.success).toBe(true)
|
||||
if (parsed.success) {
|
||||
expect(parsed.data).toEqual({ accessToken: 'token', file })
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects a body without an access token', () => {
|
||||
expect(
|
||||
tiktokUploadVideoDraftBodySchema.safeParse({
|
||||
file,
|
||||
}).success
|
||||
).toBe(false)
|
||||
})
|
||||
|
||||
it('rejects a body without a file', () => {
|
||||
expect(
|
||||
tiktokUploadVideoDraftBodySchema.safeParse({
|
||||
accessToken: 'token',
|
||||
}).success
|
||||
).toBe(false)
|
||||
})
|
||||
})
|
||||
@@ -176,7 +176,6 @@ const JOB_TYPE_TO_TASK_ID: Record<JobType, string> = {
|
||||
'workflow-execution': 'workflow-execution',
|
||||
'schedule-execution': 'schedule-execution',
|
||||
'webhook-execution': 'webhook-execution',
|
||||
'tiktok-webhook-ingress': 'tiktok-webhook-ingress',
|
||||
'resume-execution': 'resume-execution',
|
||||
'workflow-group-cell': 'workflow-group-cell',
|
||||
'cleanup-logs': 'cleanup-logs',
|
||||
|
||||
@@ -33,7 +33,6 @@ export type JobType =
|
||||
| 'workflow-execution'
|
||||
| 'schedule-execution'
|
||||
| 'webhook-execution'
|
||||
| 'tiktok-webhook-ingress'
|
||||
| 'resume-execution'
|
||||
| 'workflow-group-cell'
|
||||
| 'cleanup-logs'
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
/**
|
||||
* @vitest-environment node
|
||||
*/
|
||||
|
||||
import {
|
||||
dbChainMock,
|
||||
dbChainMockFns,
|
||||
drizzleOrmMock,
|
||||
resetDbChainMock,
|
||||
schemaMock,
|
||||
} from '@sim/testing'
|
||||
import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
vi.mock('@sim/db', () => ({ ...dbChainMock, ...schemaMock }))
|
||||
vi.mock('@sim/db/schema', () => schemaMock)
|
||||
vi.mock('drizzle-orm', () => drizzleOrmMock)
|
||||
|
||||
import { clearCredentialRefs } from '@/lib/credentials/deletion'
|
||||
|
||||
describe('credential-bound webhook deactivation', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
resetDbChainMock()
|
||||
})
|
||||
|
||||
afterAll(() => {
|
||||
resetDbChainMock()
|
||||
})
|
||||
|
||||
it('deactivates TikTok and Slack webhook registrations when a credential is removed', async () => {
|
||||
await clearCredentialRefs('credential-1', 'workspace-1')
|
||||
|
||||
expect(dbChainMockFns.update).toHaveBeenCalledWith(schemaMock.webhook)
|
||||
expect(dbChainMockFns.set).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ isActive: false, updatedAt: expect.any(Date) })
|
||||
)
|
||||
expect(drizzleOrmMock.eq).toHaveBeenCalledWith(schemaMock.webhook.provider, 'tiktok')
|
||||
expect(drizzleOrmMock.eq).toHaveBeenCalledWith(schemaMock.webhook.provider, 'slack_app')
|
||||
expect(drizzleOrmMock.eq).toHaveBeenCalledWith(schemaMock.webhook.provider, 'slack')
|
||||
})
|
||||
})
|
||||
@@ -90,18 +90,18 @@ export async function clearCredentialRefs(
|
||||
clearInPausedExecutions(credentialId, workspaceId, needle),
|
||||
clearInWorkflowCheckpoints(credentialId, workspaceId, needle),
|
||||
clearInKnowledgeConnectors(credentialId),
|
||||
deactivateSlackAppWebhooks(credentialId),
|
||||
deactivateCredentialBoundWebhooks(credentialId),
|
||||
])
|
||||
}
|
||||
|
||||
/**
|
||||
* Deactivates Slack trigger webhooks bound to this credential so inbound
|
||||
* events stop routing once the account is disconnected: native (`slack_app`)
|
||||
* rows reference it via `providerConfig.credentialId`, custom-bot (`slack`)
|
||||
* rows via `routingKey` = the bot credential id. Neither is a foreign key, so
|
||||
* Deactivates app-level trigger webhooks bound to this credential so inbound
|
||||
* events stop routing once the account is disconnected. Native Slack and
|
||||
* TikTok rows reference it via `providerConfig.credentialId`; custom-bot Slack
|
||||
* rows use `routingKey` = the bot credential id. Neither is a foreign key, so
|
||||
* neither is covered by CASCADE.
|
||||
*/
|
||||
async function deactivateSlackAppWebhooks(credentialId: string): Promise<void> {
|
||||
async function deactivateCredentialBoundWebhooks(credentialId: string): Promise<void> {
|
||||
await db
|
||||
.update(schema.webhook)
|
||||
.set({ isActive: false, updatedAt: new Date() })
|
||||
@@ -113,6 +113,10 @@ async function deactivateSlackAppWebhooks(credentialId: string): Promise<void> {
|
||||
eq(schema.webhook.provider, 'slack_app'),
|
||||
sql`${schema.webhook.providerConfig}->>'credentialId' = ${credentialId}`
|
||||
),
|
||||
and(
|
||||
eq(schema.webhook.provider, 'tiktok'),
|
||||
sql`${schema.webhook.providerConfig}->>'credentialId' = ${credentialId}`
|
||||
),
|
||||
and(eq(schema.webhook.provider, 'slack'), eq(schema.webhook.routingKey, credentialId))
|
||||
)
|
||||
)
|
||||
|
||||
@@ -97,6 +97,15 @@ const slackTrigger = trigger([
|
||||
},
|
||||
])
|
||||
|
||||
const tiktokTrigger = trigger([
|
||||
{
|
||||
id: 'triggerCredentials',
|
||||
mode: 'trigger',
|
||||
serviceId: 'tiktok',
|
||||
required: true,
|
||||
},
|
||||
])
|
||||
|
||||
function makeBlock(
|
||||
type: string,
|
||||
subBlockValues: Record<string, unknown>,
|
||||
@@ -364,3 +373,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')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -40,6 +40,7 @@ import { SIM_SUBSCRIBED_EVENTS } from '@/triggers/slack/shared'
|
||||
import { resolveBlockTriggerId } from '@/triggers/webhook-url'
|
||||
|
||||
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
|
||||
@@ -325,9 +326,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
|
||||
@@ -528,6 +528,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 {
|
||||
|
||||
@@ -1,12 +0,0 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { tiktokGetPostStatusTool } from '@/tools/tiktok/get_post_status'
|
||||
|
||||
describe('TikTok Get Post Status OAuth scopes', () => {
|
||||
it('requires the draft-upload scope used by the supported posting flow', () => {
|
||||
expect(tiktokGetPostStatusTool.oauth).toEqual({
|
||||
required: true,
|
||||
provider: 'tiktok',
|
||||
requiredScopes: ['video.upload'],
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,9 @@
|
||||
-- TikTok now uses the shared routing-key lookup. No deployed TikTok webhooks
|
||||
-- exist to backfill, so only remove the obsolete provider-specific index.
|
||||
COMMIT;
|
||||
--> statement-breakpoint
|
||||
SET lock_timeout = 0;
|
||||
--> statement-breakpoint
|
||||
DROP INDEX CONCURRENTLY IF EXISTS "webhook_tiktok_credential_id_idx";
|
||||
--> statement-breakpoint
|
||||
SET lock_timeout = '5s';
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1996,6 +1996,13 @@
|
||||
"when": 1786217700133,
|
||||
"tag": "0285_workspace_inbox_provider_id_idx",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 286,
|
||||
"version": "7",
|
||||
"when": 1786226773546,
|
||||
"tag": "0286_tiktok_routing_key",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
+5
-15
@@ -838,11 +838,6 @@ export const jobExecutionLogs = pgTable(
|
||||
})
|
||||
)
|
||||
|
||||
/** Extracts the canonical credential ID persisted in webhook provider configuration. */
|
||||
export function webhookCredentialIdExpression(column: AnyPgColumn): SQL<string> {
|
||||
return sql<string>`((${column})::jsonb ->> 'credentialId')`
|
||||
}
|
||||
|
||||
export const webhook = pgTable(
|
||||
'webhook',
|
||||
{
|
||||
@@ -861,15 +856,15 @@ export const webhook = pgTable(
|
||||
blockId: text('block_id'),
|
||||
/**
|
||||
* URL-addressable webhook path. NULL for shared-app providers (e.g. the
|
||||
* native Slack OAuth trigger) whose events arrive on a single shared
|
||||
* native Slack and TikTok triggers) whose events arrive on a single shared
|
||||
* endpoint and route by `routingKey` instead of a per-workflow path.
|
||||
*/
|
||||
path: text('path'),
|
||||
/**
|
||||
* Tenant routing key for shared-app providers. For `provider='slack_app'`
|
||||
* this is the Slack `team_id`, derived server-side from the connected
|
||||
* credential at deploy time — never user input. Inbound events match on
|
||||
* this after HMAC verification.
|
||||
* Tenant routing key for shared-app providers, such as Slack `team_id` or
|
||||
* TikTok `open_id`, derived server-side from the connected credential at
|
||||
* deploy time — never user input. Inbound events match on this after HMAC
|
||||
* verification.
|
||||
*/
|
||||
routingKey: text('routing_key'),
|
||||
provider: text('provider'), // e.g., "whatsapp", "github", etc.
|
||||
@@ -901,11 +896,6 @@ export const webhook = pgTable(
|
||||
providerActiveWorkflowDeploymentIdx: index(
|
||||
'idx_webhook_on_provider_is_active_workflow_id_deploym_bdeed5468'
|
||||
).on(table.provider, table.isActive, table.workflowId, table.deploymentVersionId),
|
||||
tiktokCredentialIdIdx: index('webhook_tiktok_credential_id_idx')
|
||||
.on(webhookCredentialIdExpression(table.providerConfig))
|
||||
.where(
|
||||
sql`${table.provider} = 'tiktok' AND ${table.isActive} = true AND ${table.archivedAt} IS NULL`
|
||||
),
|
||||
workflowBlockUpdatedDescIdx: index('idx_webhook_on_workflow_id_block_id_updated_at_desc').on(
|
||||
table.workflowId,
|
||||
table.blockId,
|
||||
|
||||
Reference in New Issue
Block a user