# inngest-workflows

> Inngest workflow patterns for OCR Chandra. Step functions, retry logic, event handling, and pipeline orchestration.

- Author: Long0308
- Repository: Long0308/OCR-Chandra
- Version: 20260207140355
- Stars: 0
- Forks: 0
- Last Updated: 2026-02-07
- Source: https://github.com/Long0308/OCR-Chandra
- Web: https://mule.run/skillshub/@@Long0308/OCR-Chandra~inngest-workflows:20260207140355

---

---
name: inngest-workflows
description: Inngest workflow patterns for OCR Chandra. Step functions, retry logic, event handling, and pipeline orchestration.
allowed-tools: Read, Write, Edit, Glob, Grep
---

# Inngest Workflows Skill

> Workflow orchestration patterns with Inngest.
> **Design for resilience, not just functionality.**

## 🎯 Selective Reading Rule

**Read ONLY files relevant to the request!**

---

## 📑 Content Map

| File | Description | When to Read |
|------|-------------|--------------|
| `setup.md` | Inngest Python setup | Initial configuration |
| `steps.md` | Step function patterns | Writing workflow steps |
| `events.md` | Event handling | Event triggers |
| `debugging.md` | Debugging workflows | Troubleshooting |

---

## 🔧 Inngest Setup

### FastAPI Integration

```python
# app/main.py
from fastapi import FastAPI
from inngest.fast_api import serve
from app.inngest.client import inngest_client
from app.inngest import functions  # Import all functions

app = FastAPI()

serve(
    app,
    inngest_client,
    [functions.process_document, functions.sync_sharepoint],
)
```

### Client Configuration

```python
# app/inngest/client.py
from inngest import Inngest
import os

inngest_client = Inngest(
    app_id="ocr-chandra",
    is_production=os.getenv("ENVIRONMENT") == "production",
)
```

---

## 📝 Step Patterns

### Basic Step

```python
result = await step.run("step-name", lambda: my_function())
```

### Parallel Steps

```python
# Process pages in parallel
results = []
for i, page in enumerate(pages):
    result = await step.run(f"ocr-page-{i}", lambda p=page: process_page(p))
    results.append(result)
```

### Sleep

```python
await step.sleep("wait", duration="30s")
```

### Send Event

```python
await step.send_event("notify", {
    "name": "ocr/job.completed",
    "data": {"job_id": job_id}
})
```

---

## 🔄 Error Handling

### Retriable Errors

```python
# Will be retried up to fn retry limit
async def my_step():
    if external_service_down():
        raise Exception("Service unavailable")  # Retriable
```

### Non-Retriable Errors

```python
from inngest import NonRetriableError

async def validate():
    if not valid:
        raise NonRetriableError("Invalid input")  # Won't retry
```

---

## 📊 Event Types for OCR Chandra

| Event | Trigger | Data |
|-------|---------|------|
| `ocr/document.uploaded` | File uploaded | `job_id`, `filename`, `path` |
| `ocr/page.processed` | Page OCR done | `job_id`, `page`, `result` |
| `ocr/job.completed` | All pages done | `job_id`, `outputs` |
| `ocr/job.failed` | Processing failed | `job_id`, `error` |
| `sync/sharepoint.requested` | Sync triggered | `folder_path` |

---

## 🐛 Debugging

### Local Dev Server

```bash
npx inngest-cli@latest dev -u http://localhost:8000/api/inngest
```

Dashboard: http://localhost:8288

### Logging

```python
import structlog
logger = structlog.get_logger()

@inngest_client.create_function(...)
async def my_function(ctx, step):
    logger.info("Starting", job_id=ctx.event.data["job_id"])
    # ...
    logger.info("Completed", duration_ms=elapsed)
```

---

## ✅ Checklist

- [ ] Steps have unique, descriptive names
- [ ] Retry count appropriate for operation type
- [ ] NonRetriableError for validation failures
- [ ] Logging in each major step
- [ ] Events properly typed

---

## 🔗 Related Skills

| Need | Skill |
|------|-------|
| OCR | `@[skills/ocr-processing]` |
| Storage | `@[skills/supabase-patterns]` |
| API | `@[skills/api-patterns]` |