# Task Creation from Email or Chat Messages in Macro: The Complete Workflow

> Learn how Macro converts emails and chat messages into structured tasks via parsing, action persistence, dispatch, and storage. Master the complete task creation workflow.

- Repository: [Macro/macro](https://github.com/macro-inc/macro)
- Tags: how-to-guide
- Published: 2026-08-15

---

**Macro converts plain emails and chat messages into structured task objects through a multi-stage pipeline involving email parsing, scheduled action persistence, background dispatch, and final document storage.**

The **macro-inc/macro** repository implements an end-to-end workflow for **task creation from email or chat messages**. Incoming communications are parsed, transformed into JSON payloads, persisted to PostgreSQL, and ultimately stored as documents that downstream services can process. This article breaks down each stage with source code references from the actual implementation.

---

## How Email Messages Trigger Task Creation

The email flow begins when an inbound message arrives at Macro's email service. The system parses MIME payloads and extracts structured task data.

### Email Reception and Parsing

In [`email_service/src/util/sync_contacts.rs`](https://github.com/macro-inc/macro/blob/main/email_service/src/util/sync_contacts.rs), the email service handles inbound webhooks. The parser scans email bodies for task indicators—typically a `Task:` block or specially formatted commands—and constructs a JSON `task` object containing the title, description, due date, and owner information.

This extracted payload is then forwarded to the scheduled action system for persistence and execution.

### Scheduled Action API Entry Point

The parsed task JSON arrives at the HTTP endpoint defined in [`scheduled_action/src/inbound/axum_router.rs`](https://github.com/macro-inc/macro/blob/main/scheduled_action/src/inbound/axum_router.rs). The `POST /scheduled_action/create` handler receives a request body with:

- `owner`: The task assignee (e.g., `macro|user@user.com`)
- `name`: Display name for the scheduled action
- `schedule`: ISO 8601 timestamp for execution
- `kind`: Action type identifier (set to `"task"` for task creation)
- `timezone`: Scheduling timezone
- `task`: The nested task definition object

```http
POST /scheduled_action/create
Content-Type: application/json

{
  "owner": "macro|user@user.com",
  "name": "Follow‑up on Q3 report",
  "schedule": "2024-09-01T09:00:00Z",
  "kind": "task",
  "timezone": "America/Los_Angeles",
  "task": {
    "title": "Review Q3 financials",
    "description": "Ask finance team for the latest numbers",
    "due_at": "2024-09-05T17:00:00Z"
  }
}

```

---

## Database Persistence and Dispatch

Once the API receives the payload, the system persists it for reliable, at-least-once execution.

### Scheduled Action Repository

The [`scheduled_action/src/outbound/pg_scheduled_action_repo.rs`](https://github.com/macro-inc/macro/blob/main/scheduled_action/src/outbound/pg_scheduled_action_repo.rs) module handles database operations. It executes an `INSERT` into the `scheduled_action` table, storing:

- Raw JSON payload
- Owner identity
- Schedule metadata
- The serialized `task` column

This persistence decouples receipt from execution, ensuring tasks survive service restarts.

### Background Polling Dispatcher

The [`scheduled_action/src/outbound/pg_polling_dispatcher.rs`](https://github.com/macro-inc/macro/blob/main/scheduled_action/src/outbound/pg_polling_dispatcher.rs) runs continuously, querying for pending `scheduled_action` rows. For each due record, it spawns a Tokio task via `TaskTracker` to execute the payload asynchronously.

The dispatcher determines execution path based on the `kind` field:

- **Worker tasks**: Trigger external ECS containers
- **Document tasks**: Call internal document service APIs

---

## Task Execution Paths

Macro supports two execution modes depending on task requirements.

### External Worker Execution (ECS)

For tasks requiring isolated compute resources, the dispatcher invokes [`worker_trigger/src/service/ecs/run_task.rs`](https://github.com/macro-inc/macro/blob/main/worker_trigger/src/service/ecs/run_task.rs). This module calls AWS ECS `RunTask` with a `task_definition` ARN, launching a containerized worker that processes the payload.

This path suits resource-intensive or security-sensitive operations.

### Document Service Task Creation

Most email-derived tasks follow the internal path to [`documents/src/inbound/axum_router/create_task.rs`](https://github.com/macro-inc/macro/blob/main/documents/src/inbound/axum_router/create_task.rs). This handler:

1. Validates the `CreateTaskRequest`
2. Generates a task ID
3. Inserts a row into the `documents` table via [`documents/src/outbound/pg_document_repo.rs`](https://github.com/macro-inc/macro/blob/main/documents/src/outbound/pg_document_repo.rs)
4. Returns the created task identifier

```http
POST /documents/create_task
Content-Type: application/json
Authorization: Bearer <jwt>

{
  "title": "Update product roadmap",
  "description": "Add Q4 milestones",
  "owner": "macro|user@user.com",
  "metadata": { "priority": "high" }
}

```

The `documents` table serves as the canonical store for all tasks, enabling downstream services like search indexing and notification systems to react to new entries.

---

## Chat Message Task Creation

Chat platforms like Slack follow a parallel but unified path through the notification service.

### Notification Service Event Handling

When a chat message contains a task assignment, [`notification_service/src/api/user_notification.rs`](https://github.com/macro-inc/macro/blob/main/notification_service/src/api/user_notification.rs) receives a `task_assigned` event. The handler:

- Persists the event to the `notifications` table
- Spawns an in-process executor task

### In-Process Agent Task Execution

The [`scheduled_action/src/outbound/inprocess_executor/agent_task.rs`](https://github.com/macro-inc/macro/blob/main/scheduled_action/src/outbound/inprocess_executor/agent_task.rs) module extracts task descriptions from chat payloads and constructs API calls to the Documents service. This Rust code illustrates the pattern:

```rust
let resp = client
    .post("/documents/create_task")
    .json(&CreateTaskRequest {
        title: "Follow‑up on demo".into(),
        description: Some("Schedule a follow‑up call".into()),
        owner: "macro|user@user.com".into(),
        metadata: None,
    })
    .send()
    .await?;

```

This design reuses the same `create_task` endpoint regardless of origin—email or chat—ensuring consistent validation, persistence, and downstream processing.

---

## Architecture Comparison: Email vs. Chat Flows

| Aspect | Email Flow | Chat Flow |
|--------|-----------|-----------|
| **Entry point** | `email_service` webhook | `notification_service` event |
| **Parsing location** | [`sync_contacts.rs`](https://github.com/macro-inc/macro/blob/main/sync_contacts.rs) | [`user_notification.rs`](https://github.com/macro-inc/macro/blob/main/user_notification.rs) |
| **Initial storage** | `scheduled_action` table | `notifications` table |
| **Dispatcher** | [`pg_polling_dispatcher.rs`](https://github.com/macro-inc/macro/blob/main/pg_polling_dispatcher.rs) | In-process executor |
| **Task creation** | Documents API (async) | Documents API (sync/async hybrid) |
| **Latency** | Minutes (scheduled) | Seconds (real-time) |

Both paths converge at the Documents service, maintaining a single source of truth for task state.

---

## Summary

- **Email parsing** happens in [`email_service/src/util/sync_contacts.rs`](https://github.com/macro-inc/macro/blob/main/email_service/src/util/sync_contacts.rs), extracting task JSON from MIME bodies.
- **Scheduled actions** provide durable persistence via [`pg_scheduled_action_repo.rs`](https://github.com/macro-inc/macro/blob/main/pg_scheduled_action_repo.rs) and reliable execution through [`pg_polling_dispatcher.rs`](https://github.com/macro-inc/macro/blob/main/pg_polling_dispatcher.rs).
- **Worker tasks** launch ECS containers through [`worker_trigger/src/service/ecs/run_task.rs`](https://github.com/macro-inc/macro/blob/main/worker_trigger/src/service/ecs/run_task.rs).
- **Document tasks** persist to the `documents` table via [`documents/src/inbound/axum_router/create_task.rs`](https://github.com/macro-inc/macro/blob/main/documents/src/inbound/axum_router/create_task.rs).
- **Chat messages** route through `notification_service` and execute via [`inprocess_executor/agent_task.rs`](https://github.com/macro-inc/macro/blob/main/inprocess_executor/agent_task.rs), reusing the same Documents API.

---

## Frequently Asked Questions

### How does Macro identify tasks inside email bodies?

The email service in [`sync_contacts.rs`](https://github.com/macro-inc/macro/blob/main/sync_contacts.rs) scans for structured markers like `Task:` prefixes or command syntax. It parses the surrounding text into JSON fields including title, description, and due dates. The parser handles plain text and HTML MIME parts.

### What happens if the scheduled action dispatcher crashes?

The [`pg_polling_dispatcher.rs`](https://github.com/macro-inc/macro/blob/main/pg_polling_dispatcher.rs) uses PostgreSQL row-level locking with `SKIP LOCKED` to ensure exactly-once processing. Unacknowledged tasks remain in the table and are retried on restart. The `scheduled_action` table persistence guarantees no task loss.

### Why are chat tasks handled differently from email tasks?

Chat requires lower latency. The notification service uses an in-process executor in [`agent_task.rs`](https://github.com/macro-inc/macro/blob/main/agent_task.rs) rather than polling, enabling sub-second task creation. Email tolerates minutes of delay, making the polling dispatcher more efficient for batch processing.

### Can external systems trigger this same workflow?

Yes. Any source can `POST` to `/scheduled_action/create` with a valid task payload. The API in [`axum_router.rs`](https://github.com/macro-inc/macro/blob/main/axum_router.rs) accepts authenticated requests from integrations, not just internal services.