> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/iii-hq/sdk/llms.txt
> Use this file to discover all available pages before exploring further.

# Custom Triggers

> Create custom trigger types with registerTriggerType for scheduled jobs, webhooks, and event-driven architectures

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

<Steps>
  <Step title="Register a trigger type">
    Define the trigger type once with a unique ID and handler logic.
  </Step>

  <Step title="Register trigger instances">
    Create multiple trigger instances with different configurations.
  </Step>

  <Step title="Handle events">
    When an event occurs, your handler invokes the configured function via `iii.call()`.
  </Step>
</Steps>

## TriggerHandler Interface

<CodeGroup>
  ```typescript Node.js theme={null}
  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>
  }
  ```

  ```python Python theme={null}
  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
  ```
</CodeGroup>

## Example: Cron Trigger

<CodeGroup>
  ```typescript Node.js theme={null}
  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
  })
  ```

  ```python Python theme={null}
  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
  })
  ```
</CodeGroup>

## Example: Webhook Trigger

<CodeGroup>
  ```typescript Node.js theme={null}
  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 }
  })
  ```

  ```python Python theme={null}
  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)
  ```
</CodeGroup>

## Example: Message Queue Trigger

<CodeGroup>
  ```typescript Node.js theme={null}
  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' }
  })
  ```

  ```python Python theme={null}
  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())
  ```
</CodeGroup>

## Unregistering Triggers

<CodeGroup>
  ```typescript Node.js theme={null}
  // 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
  ```

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

  # Later: unregister
  trigger.unregister()  # Calls unregister_trigger() on handler
  ```
</CodeGroup>

## Error Handling

<Warning>
  If `registerTrigger()` throws an error, it's communicated back to the engine:

  <CodeGroup>
    ```typescript Node.js theme={null}
    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) {}
    }
    ```

    ```python Python theme={null}
    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'}}
    ```
  </CodeGroup>
</Warning>

## Best Practices

<Tip>
  **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
</Tip>