Custom Triggers

Overview

Custom triggers allow you to invoke functions in response to external events like cron schedules, webhooks, message queues, or database changes. The SDK provides registerTriggerType to define reusable trigger types that can be registered multiple times with different configurations.

Trigger Architecture

Define the trigger type once with a unique ID and handler logic.

Create multiple trigger instances with different configurations.

Handle events

When an event occurs, your handler invokes the configured function via iii.call().

TriggerHandler Interface

import type { TriggerHandler } from 'iii-sdk'

type TriggerConfig<TConfig> = {
  id: string           // Unique trigger instance ID
  function_id: string  // Function to invoke when triggered
  config: TConfig      // Custom configuration for this trigger
}

type TriggerHandler<TConfig> = {
  registerTrigger(config: TriggerConfig<TConfig>): Promise<void>
  unregisterTrigger(config: TriggerConfig<TConfig>): Promise<void>
}

Example: Cron Trigger

import { init } from 'iii-sdk'
import type { TriggerHandler } from 'iii-sdk'
import { CronJob } from 'cron'

type CronConfig = {
  schedule: string  // e.g., '*/5 * * * *' for every 5 minutes
  timezone?: string
}

const iii = init('ws://localhost:49134')

// Store active cron jobs
const cronJobs = new Map<string, CronJob>()

const cronHandler: TriggerHandler<CronConfig> = {
  async registerTrigger(config) {
    const { id, function_id, config: { schedule, timezone } } = config
    
    if (!schedule) {
      throw new Error('schedule is required')
    }
    
    console.log(`Registering cron trigger ${id}: ${schedule} → ${function_id}`)
    
    const job = new CronJob(
      schedule,
      async () => {
        console.log(`Cron triggered: ${id}`)
        try {
          await iii.call(function_id, { triggerId: id, timestamp: Date.now() })
        } catch (error) {
          console.error(`Cron execution failed for ${id}:`, error)
        }
      },
      null,
      true,  // Start immediately
      timezone
    )
    
    cronJobs.set(id, job)
  },
  
  async unregisterTrigger(config) {
    const job = cronJobs.get(config.id)
    if (job) {
      job.stop()
      cronJobs.delete(config.id)
      console.log(`Unregistered cron trigger ${config.id}`)
    }
  }
}

// Register the trigger type
iii.registerTriggerType(
  { id: 'cron', description: 'Cron-based scheduled execution' },
  cronHandler
)

// Now register specific cron triggers
iii.registerFunction({ id: 'cleanup::daily' }, async (data) => {
  console.log('Running daily cleanup...')
  return { cleaned: true }
})

iii.registerTrigger({
  type: 'cron',
  function_id: 'cleanup::daily',
  config: { schedule: '0 0 * * *' }  // Daily at midnight
})

iii.registerTrigger({
  type: 'cron',
  function_id: 'cleanup::daily',
  config: { schedule: '*/5 * * * *' }  // Every 5 minutes
})

Example: Webhook Trigger

import express from 'express'
import crypto from 'crypto'
import type { TriggerHandler } from 'iii-sdk'

type WebhookConfig = {
  path: string      // e.g., '/webhooks/stripe'
  secret?: string   // Optional HMAC secret for verification
}

const app = express()
app.use(express.json())

const webhookRoutes = new Map<string, { functionId: string; secret?: string }>()

const webhookHandler: TriggerHandler<WebhookConfig> = {
  async registerTrigger(config) {
    const { id, function_id, config: { path, secret } } = config
    
    if (!path) {
      throw new Error('path is required')
    }
    
    console.log(`Registering webhook trigger ${id}: ${path} → ${function_id}`)
    
    webhookRoutes.set(path, { functionId: function_id, secret })
    
    // Register Express route if not exists
    if (!app._router.stack.find(layer => layer.route?.path === path)) {
      app.post(path, async (req, res) => {
        const route = webhookRoutes.get(path)
        if (!route) {
          return res.status(404).json({ error: 'Webhook not found' })
        }
        
        // Verify signature if secret is configured
        if (route.secret) {
          const signature = req.headers['x-webhook-signature'] as string
          const expectedSig = crypto
            .createHmac('sha256', route.secret)
            .update(JSON.stringify(req.body))
            .digest('hex')
          
          if (signature !== expectedSig) {
            return res.status(401).json({ error: 'Invalid signature' })
          }
        }
        
        try {
          const result = await iii.call(route.functionId, {
            headers: req.headers,
            body: req.body,
            path: req.path
          })
          res.json(result)
        } catch (error) {
          console.error('Webhook execution failed:', error)
          res.status(500).json({ error: 'Internal server error' })
        }
      })
    }
  },
  
  async unregisterTrigger(config) {
    const { config: { path } } = config
    webhookRoutes.delete(path)
    console.log(`Unregistered webhook trigger ${config.id}`)
  }
}

iii.registerTriggerType(
  { id: 'webhook', description: 'HTTP webhook trigger' },
  webhookHandler
)

app.listen(3000, () => console.log('Webhook server listening on :3000'))

// Usage
iii.registerFunction({ id: 'stripe::payment' }, async (data) => {
  console.log('Stripe webhook received:', data.body)
  return { received: true }
})

iii.registerTrigger({
  type: 'webhook',
  function_id: 'stripe::payment',
  config: { path: '/webhooks/stripe', secret: process.env.STRIPE_SECRET }
})

Example: Message Queue Trigger

import { Kafka } from 'kafkajs'
import type { TriggerHandler } from 'iii-sdk'

type KafkaConfig = {
  topic: string
  groupId: string
}

const kafka = new Kafka({ brokers: ['localhost:9092'] })
const consumers = new Map<string, any>()

const kafkaHandler: TriggerHandler<KafkaConfig> = {
  async registerTrigger(config) {
    const { id, function_id, config: { topic, groupId } } = config
    
    if (!topic || !groupId) {
      throw new Error('topic and groupId are required')
    }
    
    console.log(`Registering Kafka trigger ${id}: ${topic} → ${function_id}`)
    
    const consumer = kafka.consumer({ groupId })
    await consumer.connect()
    await consumer.subscribe({ topic })
    
    await consumer.run({
      eachMessage: async ({ message }) => {
        try {
          const value = message.value?.toString()
          const data = value ? JSON.parse(value) : {}
          
          await iii.call(function_id, {
            triggerId: id,
            topic,
            offset: message.offset,
            data
          })
        } catch (error) {
          console.error(`Kafka message processing failed for ${id}:`, error)
        }
      }
    })
    
    consumers.set(id, consumer)
  },
  
  async unregisterTrigger(config) {
    const consumer = consumers.get(config.id)
    if (consumer) {
      await consumer.disconnect()
      consumers.delete(config.id)
      console.log(`Unregistered Kafka trigger ${config.id}`)
    }
  }
}

iii.registerTriggerType(
  { id: 'kafka', description: 'Kafka message queue trigger' },
  kafkaHandler
)

// Usage
iii.registerFunction({ id: 'orders::process' }, async (data) => {
  console.log('Processing order from Kafka:', data)
  return { processed: true }
})

iii.registerTrigger({
  type: 'kafka',
  function_id: 'orders::process',
  config: { topic: 'orders', groupId: 'order-processor' }
})

Unregistering Triggers

// Register and get trigger reference
const trigger = iii.registerTrigger({
  type: 'cron',
  function_id: 'cleanup::hourly',
  config: { schedule: '0 * * * *' }
})

// Later: unregister
trigger.unregister()  // Calls unregisterTrigger() on handler

Error Handling

If registerTrigger() throws an error, it’s communicated back to the engine:

const cronHandler: TriggerHandler<CronConfig> = {
  async registerTrigger(config) {
    if (!config.config.schedule) {
      throw new Error('schedule is required')
    }
    // Error sent as TriggerRegistrationResult:
    // { error: { code: 'trigger_registration_failed', message: 'schedule is required' } }
  },
  async unregisterTrigger(config) {}
}

Best Practices

Custom Trigger Checklist:

✅ Validate config in registerTrigger() and throw descriptive errors
✅ Store trigger state (jobs, connections) in a Map keyed by id
✅ Clean up resources in unregisterTrigger() (stop jobs, close connections)
✅ Handle errors when calling iii.call() - don’t let trigger crashes stop the process
✅ Use callVoid for fire-and-forget triggers, call when you need results
✅ Log trigger events for debugging (registration, execution, errors)
✅ Consider idempotency - triggers may fire multiple times for the same event

Get Started

Core Concepts

Guides

Custom Triggers

Overview

Trigger Architecture

TriggerHandler Interface

Example: Cron Trigger

Example: Webhook Trigger

Example: Message Queue Trigger

Unregistering Triggers

Error Handling

Best Practices

Build docs developers (and LLMs) love

Get Started

Core Concepts

Guides

Documentation Index

​Overview

​Trigger Architecture

​TriggerHandler Interface

​Example: Cron Trigger

​Example: Webhook Trigger

​Example: Message Queue Trigger

​Unregistering Triggers

​Error Handling

​Best Practices

Build docs developers (and LLMs) love

Overview

Trigger Architecture

TriggerHandler Interface

Example: Cron Trigger

Example: Webhook Trigger

Example: Message Queue Trigger

Unregistering Triggers

Error Handling

Best Practices