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>
}

from iii import TriggerHandler, TriggerConfig
from typing import TypeVar, Generic

TConfig = TypeVar('TConfig')

class TriggerHandler(Generic[TConfig]):
    async def register_trigger(self, config: TriggerConfig[TConfig]) -> None:
        """Called when a new trigger instance is registered"""
        ...
    
    async def unregister_trigger(self, config: TriggerConfig[TConfig]) -> None:
        """Called when a trigger instance is unregistered"""
        ...

# TriggerConfig structure:
class TriggerConfig:
    id: str              # Unique trigger instance ID
    function_id: str     # Function to invoke when triggered
    config: dict         # Custom configuration for this trigger

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
})

import asyncio
from datetime import datetime
from typing import Any
from iii import III, TriggerHandler, TriggerConfig
from aiocron import crontab  # pip install aiocron

class CronConfig:
    schedule: str
    timezone: str | None = None

iii = III('ws://localhost:49134')

# Store active cron jobs
cron_jobs: dict[str, Any] = {}

class CronHandler(TriggerHandler[dict]):
    async def register_trigger(self, config: TriggerConfig) -> None:
        trigger_id = config.id
        function_id = config.function_id
        schedule = config.config.get('schedule')
        
        if not schedule:
            raise ValueError('schedule is required')
        
        print(f'Registering cron trigger {trigger_id}: {schedule} → {function_id}')
        
        async def job_func():
            print(f'Cron triggered: {trigger_id}')
            try:
                await iii.call(function_id, {
                    'trigger_id': trigger_id,
                    'timestamp': datetime.now().timestamp()
                })
            except Exception as error:
                print(f'Cron execution failed for {trigger_id}: {error}')
        
        job = crontab(schedule, func=job_func)
        cron_jobs[trigger_id] = job
    
    async def unregister_trigger(self, config: TriggerConfig) -> None:
        job = cron_jobs.pop(config.id, None)
        if job:
            job.stop()
            print(f'Unregistered cron trigger {config.id}')

# Register the trigger type
iii.register_trigger_type('cron', 'Cron-based scheduled execution', CronHandler())

# Now register specific cron triggers
async def daily_cleanup(data: dict) -> dict:
    print('Running daily cleanup...')
    return {'cleaned': True}

iii.register_function('cleanup::daily', daily_cleanup)

iii.register_trigger('cron', 'cleanup::daily', {
    'schedule': '0 0 * * *'  # Daily at midnight
})

iii.register_trigger('cron', 'cleanup::daily', {
    '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 }
})

from aiohttp import web
import hmac
import hashlib
from typing import Any
from iii import TriggerHandler, TriggerConfig

class WebhookConfig:
    path: str
    secret: str | None = None

webhook_routes: dict[str, dict[str, Any]] = {}

class WebhookHandler(TriggerHandler[dict]):
    async def register_trigger(self, config: TriggerConfig) -> None:
        trigger_id = config.id
        function_id = config.function_id
        path = config.config.get('path')
        secret = config.config.get('secret')
        
        if not path:
            raise ValueError('path is required')
        
        print(f'Registering webhook trigger {trigger_id}: {path} → {function_id}')
        webhook_routes[path] = {'function_id': function_id, 'secret': secret}
    
    async def unregister_trigger(self, config: TriggerConfig) -> None:
        path = config.config.get('path')
        if path:
            webhook_routes.pop(path, None)
            print(f'Unregistered webhook trigger {config.id}')

iii.register_trigger_type('webhook', 'HTTP webhook trigger', WebhookHandler())

async def webhook_handler(request: web.Request) -> web.Response:
    path = request.path
    route = webhook_routes.get(path)
    
    if not route:
        return web.json_response({'error': 'Webhook not found'}, status=404)
    
    body = await request.json()
    
    # Verify signature if secret is configured
    if route['secret']:
        signature = request.headers.get('X-Webhook-Signature', '')
        expected_sig = hmac.new(
            route['secret'].encode(),
            str(body).encode(),
            hashlib.sha256
        ).hexdigest()
        
        if signature != expected_sig:
            return web.json_response({'error': 'Invalid signature'}, status=401)
    
    try:
        result = await iii.call(route['function_id'], {
            'headers': dict(request.headers),
            'body': body,
            'path': request.path
        })
        return web.json_response(result)
    except Exception as error:
        print(f'Webhook execution failed: {error}')
        return web.json_response({'error': 'Internal server error'}, status=500)

app = web.Application()
app.router.add_post('/webhooks/{path:.*}', webhook_handler)
web.run_app(app, port=3000)

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' }
})

from aiokafka import AIOKafkaConsumer
import json
from iii import TriggerHandler, TriggerConfig

consumers: dict[str, AIOKafkaConsumer] = {}

class KafkaHandler(TriggerHandler[dict]):
    async def register_trigger(self, config: TriggerConfig) -> None:
        trigger_id = config.id
        function_id = config.function_id
        topic = config.config.get('topic')
        group_id = config.config.get('group_id')
        
        if not topic or not group_id:
            raise ValueError('topic and group_id are required')
        
        print(f'Registering Kafka trigger {trigger_id}: {topic} → {function_id}')
        
        consumer = AIOKafkaConsumer(
            topic,
            bootstrap_servers='localhost:9092',
            group_id=group_id
        )
        await consumer.start()
        
        async def consume():
            try:
                async for message in consumer:
                    try:
                        data = json.loads(message.value.decode())
                        await iii.call(function_id, {
                            'trigger_id': trigger_id,
                            'topic': topic,
                            'offset': message.offset,
                            'data': data
                        })
                    except Exception as error:
                        print(f'Kafka message processing failed for {trigger_id}: {error}')
            finally:
                await consumer.stop()
        
        consumers[trigger_id] = consumer
        asyncio.create_task(consume())
    
    async def unregister_trigger(self, config: TriggerConfig) -> None:
        consumer = consumers.pop(config.id, None)
        if consumer:
            await consumer.stop()
            print(f'Unregistered Kafka trigger {config.id}')

iii.register_trigger_type('kafka', 'Kafka message queue trigger', KafkaHandler())

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

# Register and get trigger reference
trigger = iii.register_trigger('cron', 'cleanup::hourly', {
    'schedule': '0 * * * *'
})

# Later: unregister
trigger.unregister()  # Calls unregister_trigger() 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) {}
}

class CronHandler(TriggerHandler):
    async def register_trigger(self, config: TriggerConfig) -> None:
        if not config.config.get('schedule'):
            raise ValueError('schedule is required')
        # Error sent as TriggerRegistrationResult:
        # {'error': {'code': 'trigger_registration_failed', 'message': 'schedule is required'}}

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

​Overview

​Trigger Architecture

​TriggerHandler Interface

​Example: Cron Trigger

​Example: Webhook Trigger

​Example: Message Queue Trigger

​Unregistering Triggers

​Error Handling

​Best Practices

Overview

Trigger Architecture

TriggerHandler Interface

Example: Cron Trigger

Example: Webhook Trigger

Example: Message Queue Trigger

Unregistering Triggers

Error Handling

Best Practices