Build a Self-Hosted Zapier Alternative

Triggers (webhook/cron/poll)
    ↓
Inngest Queue (durable execution)
    ↓
Step Runner (HTTP | Transform | Filter | Delay)
    ↓
Prisma (workflow definitions + execution logs)
    ↓
Notification (Slack/email on failure)

// schema.prisma
model Workflow {
  id          String   @id @default(cuid())
  name        String
  description String?
  enabled     Boolean  @default(true)
  trigger     Json     // { type: "webhook"|"schedule"|"poll", config: {...} }
  steps       Json     // Array of step definitions
  createdAt   DateTime @default(now())
  updatedAt   DateTime @updatedAt
  executions  WorkflowExecution[]
}

model WorkflowExecution {
  id         String   @id @default(cuid())
  workflowId String
  workflow   Workflow @relation(fields: [workflowId], references: [id])
  status     String   // running | success | failed | dead
  triggerData Json
  stepResults Json     @default("[]")
  error      String?
  startedAt  DateTime @default(now())
  finishedAt DateTime?
}

// app/api/webhooks/[workflowId]/route.ts
import { inngest } from '@/lib/inngest'
import { prisma } from '@/lib/prisma'

export async function POST(
  req: Request,
  { params }: { params: { workflowId: string } }
) {
  const workflow = await prisma.workflow.findUnique({
    where: { id: params.workflowId, enabled: true }
  })
  if (!workflow) return new Response('Not found', { status: 404 })

  const body = await req.json()

  await inngest.send({
    name: 'workflow/execute',
    data: { workflowId: workflow.id, triggerData: body }
  })

  return Response.json({ queued: true })
}

// inngest/functions/executeWorkflow.ts
import { inngest } from '@/lib/inngest'
import { prisma } from '@/lib/prisma'
import { runStep } from '@/lib/stepRunner'

export const executeWorkflow = inngest.createFunction(
  {
    id: 'workflow-execute',
    retries: 3,
    onFailure: async ({ event, error }) => {
      await prisma.workflowExecution.update({
        where: { id: event.data.executionId },
        data: { status: 'dead', error: error.message, finishedAt: new Date() }
      })
    }
  },
  { event: 'workflow/execute' },
  async ({ event, step }) => {
    const { workflowId, triggerData } = event.data

    const workflow = await prisma.workflow.findUniqueOrThrow({
      where: { id: workflowId }
    })

    const execution = await prisma.workflowExecution.create({
      data: { workflowId, status: 'running', triggerData }
    })

    let context = { trigger: triggerData, steps: {} as Record<string, unknown> }

    for (const stepDef of workflow.steps as any[]) {
      const result = await step.run(`step-${stepDef.id}`, async () => {
        return runStep(stepDef, context)
      })
      context.steps[stepDef.id] = result

      if (stepDef.type === 'delay') {
        await step.sleep(`delay-${stepDef.id}`, stepDef.config.duration)
      }
    }

    await prisma.workflowExecution.update({
      where: { id: execution.id },
      data: { status: 'success', stepResults: context.steps, finishedAt: new Date() }
    })

    return context
  }
)

// lib/stepRunner.ts
export async function runStep(stepDef: any, context: any): Promise<any> {
  switch (stepDef.type) {
    case 'http': {
      const { url, method, headers, body } = resolveTemplates(stepDef.config, context)
      const res = await fetch(url, {
        method,
        headers: { 'Content-Type': 'application/json', ...headers },
        body: body ? JSON.stringify(body) : undefined
      })
      return { status: res.status, body: await res.json() }
    }

    case 'transform': {
      // Simple JSONata-style transform
      const { mapping } = stepDef.config
      return Object.fromEntries(
        Object.entries(mapping).map(([key, template]) => [
          key,
          resolveTemplate(template as string, context)
        ])
      )
    }

    case 'filter': {
      const { condition } = stepDef.config
      const passes = evaluateCondition(condition, context)
      if (!passes) throw new Error('FILTER_BLOCKED')
      return { passed: true }
    }

    default:
      throw new Error(`Unknown step type: ${stepDef.type}`)
  }
}

function resolveTemplate(template: string, context: any): string {
  return template.replace(/\{\{(.+?)\}\}/g, (_, path) => {
    return path.split('.').reduce((obj: any, key: string) => obj?.[key], context) ?? ''
  })
}

// inngest/functions/triggerScheduled.ts
export const triggerScheduled = inngest.createFunction(
  { id: 'trigger-scheduled' },
  { cron: '*/5 * * * *' }, // Every 5 minutes
  async ({ step }) => {
    const workflows = await prisma.workflow.findMany({
      where: {
        enabled: true,
        trigger: { path: ['type'], equals: 'schedule' }
      }
    })

    for (const wf of workflows) {
      const trigger = wf.trigger as any
      if (shouldRunNow(trigger.config.cron)) {
        await inngest.send({
          name: 'workflow/execute',
          data: { workflowId: wf.id, triggerData: { scheduledAt: new Date() } }
        })
      }
    }
  }
)