# Write outbox rows

> Insert an outbox row in the transaction of your business write, and set its key, headers, schedule and retry ceiling.

This guide shows how an application hands a message to QueueBox. The application runs one `INSERT` into the `outbox` table, inside the transaction of its business write. QueueBox reads the table, routes each row to a destination, and delivers it.

[The outbox table](/reference/outbox-table/) lists every column, its type and its default. This guide shows the columns an application writes.

<Aside>
`app/src/test/kotlin/docs/IntegrationDocSqlTest.kt` runs every SQL block on this page. The PostgreSQL blocks run against a PostgreSQL container, and the SQL Server blocks run against a SQL Server container. The running poller then delivers every inserted row to an HTTP destination.
</Aside>

## The insert shares the business transaction

Open one transaction, write the business rows, insert the outbox row, then commit.

This shared transaction is the reason for the outbox pattern. The database commits the message and the business state together, so the two never disagree.

- If the transaction commits, the business row and the outbox row both exist. QueueBox then delivers the message at least once.
- If the transaction rolls back, neither row exists. No message announces an order that does not exist.

An application that publishes to a broker outside the transaction has no such guarantee. The commit can fail after the publish, and the message then announces an order that does not exist. The publish can fail after the commit, and the message is lost. No order of the two writes removes the failure, because they go to two systems.

Two more rules follow from this one:

- Do not open a second connection for the outbox insert. A second connection is a second transaction, and the guarantee is gone.
- Keep the transaction short. QueueBox sees the row only after the commit.

## Insert a row

The examples use one business table, `orders`. The examples create it, because the test runs every statement. Your application has its own table already.

<Tabs syncKey="dialect">
<TabItem label="PostgreSQL">

```sql postgres
CREATE TABLE IF NOT EXISTS orders (
    id          UUID PRIMARY KEY,
    customer_id VARCHAR(64) NOT NULL,
    amount      NUMERIC(12, 2) NOT NULL,
    created_at  TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP
);
```

`payload` and `headers` are `JSONB`, so cast each string literal.

```sql postgres
BEGIN;

INSERT INTO orders (id, customer_id, amount)
VALUES ('11111111-1111-1111-1111-111111111111', 'cust-42', 99.99);

INSERT INTO outbox (topic, key, payload, headers, aggregate_type)
VALUES (
    'order.created',
    'cust-42',
    '{"orderId":"11111111-1111-1111-1111-111111111111","amount":99.99}'::jsonb,
    '{"X-Tenant":"acme"}'::jsonb,
    'order'
);

COMMIT;
```

</TabItem>
<TabItem label="SQL Server">

```sql sqlserver
IF OBJECT_ID('orders', 'U') IS NULL
CREATE TABLE orders (
    id          UNIQUEIDENTIFIER PRIMARY KEY,
    customer_id NVARCHAR(64) NOT NULL,
    amount      DECIMAL(12, 2) NOT NULL,
    created_at  DATETIME2 NOT NULL DEFAULT GETUTCDATE()
);
```

`payload` and `headers` are `NVARCHAR(MAX)`. `key` is a reserved word, so put it in brackets.

```sql sqlserver
BEGIN TRANSACTION;

INSERT INTO orders (id, customer_id, amount)
VALUES ('11111111-1111-1111-1111-111111111111', N'cust-42', 99.99);

INSERT INTO outbox (topic, [key], payload, headers, aggregate_type)
VALUES (
    N'order.created',
    N'cust-42',
    N'{"orderId":"11111111-1111-1111-1111-111111111111","amount":99.99}',
    N'{"X-Tenant":"acme"}',
    N'order'
);

COMMIT TRANSACTION;
```

</TabItem>
</Tabs>

The insert names five columns:

| Column | What to write |
|--------|---------------|
| `topic` | Required. The name that routes the row, for example `order.created`. |
| `payload` | Required. The message body, as a JSON object. |
| `key` | Optional. The unit of order. See [Key and order](#key-and-order). |
| `headers` | Optional. A JSON object of headers that travel with the message. See [Headers](#headers). |
| `aggregate_type` | Optional. The kind of business entity, for example `order`. |

`aggregate_type` lets a broker destination choose its address from the row. A RabbitMQ exchange, a Kafka topic or a NATS subject can be a template such as `public.orders.{{ aggregateType }}.v1`. The destination can also read the column through `exchangeFrom`, `topicFrom` or `subjectFrom`. See [Bridge a broker](/how-to/bridge-a-broker/).

The shortest legal insert names only `topic` and `payload`. Every other column has a default or accepts null.

<Tabs syncKey="dialect">
<TabItem label="PostgreSQL">

```sql postgres
BEGIN;

INSERT INTO orders (id, customer_id, amount)
VALUES ('22222222-2222-2222-2222-222222222222', 'cust-7', 12.00);

INSERT INTO outbox (topic, payload)
VALUES ('order.paid', '{"orderId":"22222222-2222-2222-2222-222222222222"}'::jsonb);

COMMIT;
```

</TabItem>
<TabItem label="SQL Server">

```sql sqlserver
BEGIN TRANSACTION;

INSERT INTO orders (id, customer_id, amount)
VALUES ('22222222-2222-2222-2222-222222222222', N'cust-7', 12.00);

INSERT INTO outbox (topic, payload)
VALUES (N'order.paid', N'{"orderId":"22222222-2222-2222-2222-222222222222"}');

COMMIT TRANSACTION;
```

</TabItem>
</Tabs>

### Columns the application must not write

QueueBox owns `state`, `attempt`, `claimed_at`, `claim_token`, `lease_expires_at` and `last_error`. The database fills `sequence`. Write `state` only with the value `pending`, which is the default. A row that starts in another state can stall, or it can deliver twice.

## Choose a topic

QueueBox matches `topic` against the `topicPattern` of each route, in configuration order. The first route that matches wins. A topic that matches no route goes to `dead`.

Use a dotted, lower case topic, for example `order.created`. The column holds 255 characters at most. [Fan out over HTTP](/how-to/fan-out-over-http/#match-topics-with-patterns) shows how a pattern matches a topic.

## Key and order

The rows of one non-empty `key` arrive in insert order, one row at a time. Rows with a null or empty `key`, and rows of different keys, have no order between them.

Use the identifier of the entity whose events must stay in order, for example the order ID or the customer ID. The two rows below share a key, so QueueBox delivers `order.created` before `order.paid`.

<Tabs syncKey="dialect">
<TabItem label="PostgreSQL">

```sql postgres
BEGIN;

INSERT INTO outbox (topic, key, payload, aggregate_type)
VALUES (
    'order.created',
    'order-33333333',
    '{"orderId":"33333333-3333-3333-3333-333333333333","step":1}'::jsonb,
    'order'
);

INSERT INTO outbox (topic, key, payload, aggregate_type)
VALUES (
    'order.paid',
    'order-33333333',
    '{"orderId":"33333333-3333-3333-3333-333333333333","step":2}'::jsonb,
    'order'
);

COMMIT;
```

</TabItem>
<TabItem label="SQL Server">

```sql sqlserver
BEGIN TRANSACTION;

INSERT INTO outbox (topic, [key], payload, aggregate_type)
VALUES (
    N'order.created',
    N'order-33333333',
    N'{"orderId":"33333333-3333-3333-3333-333333333333","step":1}',
    N'order'
);

INSERT INTO outbox (topic, [key], payload, aggregate_type)
VALUES (
    N'order.paid',
    N'order-33333333',
    N'{"orderId":"33333333-3333-3333-3333-333333333333","step":2}',
    N'order'
);

COMMIT TRANSACTION;
```

</TabItem>
</Tabs>

These rules decide what a key costs and what it gives:

- The database fills `sequence` on insert, and the claim orders the rows of one key by it. Two rows that one transaction writes keep their insert order.
- A row that waits for a retry holds back the later rows of its key. A dead row releases its key.
- One row of a key is in flight at a time. The throughput of one key is one publish round trip per row.
- The order holds for one writer per key. Two transactions that insert rows of one key at the same time can commit out of `sequence` order.

Spread a busy stream over more keys only when the consumer needs no order across them. [Ordering](/concepts/ordering/#order-and-the-key) explains the rule in full.

## Headers

`headers` holds one JSON object. Each key is a header name, and each value is a string. The column is `NOT NULL` with the default `'{}'`.

- To send no header, omit the column. An explicit `NULL` makes the insert fail.
- An HTTP destination sends each row header as an HTTP header. A row header wins over a static destination header and over the authentication header of the same name.
- A broker destination sends the row headers as message headers.
- Put `X-Correlation-Id` in `headers` to follow one message through every log line.

[Headers](/reference/headers/) lists the headers that QueueBox adds itself.

## Delay a message

Set `scheduled_at` to delay a message. QueueBox claims a row only when `scheduled_at` has passed. The default is the moment of the insert.

`max_attempts` sets the retry ceiling of one row. The block below delays a reminder by five seconds and gives it ten attempts.

<Tabs syncKey="dialect">
<TabItem label="PostgreSQL">

```sql postgres
BEGIN;

INSERT INTO outbox (topic, payload, scheduled_at, max_attempts)
VALUES (
    'order.reminder',
    '{"orderId":"22222222-2222-2222-2222-222222222222"}'::jsonb,
    CURRENT_TIMESTAMP + INTERVAL '5 seconds',
    10
);

COMMIT;
```

</TabItem>
<TabItem label="SQL Server">

```sql sqlserver
BEGIN TRANSACTION;

INSERT INTO outbox (topic, payload, scheduled_at, max_attempts)
VALUES (
    N'order.reminder',
    N'{"orderId":"22222222-2222-2222-2222-222222222222"}',
    DATEADD(SECOND, 5, SYSUTCDATETIME()),
    10
);

COMMIT TRANSACTION;
```

</TabItem>
</Tabs>

On SQL Server, `scheduled_at` is `DATETIME2` and carries no time zone. QueueBox compares it with the wall clock of the host that runs QueueBox. Run every QueueBox instance in UTC, and write UTC values.

## Set the retry ceiling

QueueBox compares the `attempt` column of a row with the `max_attempts` column of the same row. `attempt` is `0` on the first delivery, and each failed delivery raises it. The ceiling comes from the first of these that applies:

1. The `max_attempts` value that the application writes on the row.
2. `outbox.maxAttempts`, which QueueBox writes on every row it creates itself, for example a row that the inbox relay forwards. `inbox.relay.maxAttempts` overrides it for relayed rows.
3. The column default, `5`.

Set `max_attempts` on the insert to give a slow destination more attempts than the rest of the system. A row that reaches its ceiling goes to `dead`. See [Dead letters](/operations/dead-letters/).

## Write the row from application code

Any database library can write the row, because the row is one insert into one table. QueueBox publishes no library for the outbox side. Check two things in your code:

1. The insert joins the transaction of the business write.
2. `payload` and `headers` hold a JSON object. A JSON string that holds another JSON string breaks a transform and a routing key template.

<Tabs syncKey="lang">
<TabItem label="Go">

```go
func createOrder(ctx context.Context, db *sql.DB, order Order) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	if _, err := tx.ExecContext(ctx,
		"INSERT INTO orders (id, customer_id, amount) VALUES ($1, $2, $3)",
		order.ID, order.CustomerID, order.Amount); err != nil {
		return err
	}

	payload, err := json.Marshal(map[string]any{"orderId": order.ID, "amount": order.Amount})
	if err != nil {
		return err
	}

	if _, err := tx.ExecContext(ctx,
		`INSERT INTO outbox (topic, key, payload, headers, aggregate_type)
		 VALUES ($1, $2, $3::jsonb, $4::jsonb, $5)`,
		"order.created", order.CustomerID, string(payload), `{"X-Tenant":"acme"}`, "order"); err != nil {
		return err
	}

	return tx.Commit()
}
```

</TabItem>
<TabItem label="TypeScript">

```ts

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });

export async function createOrder(order: { id: string; customerId: string; amount: number }) {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    await client.query(
      "INSERT INTO orders (id, customer_id, amount) VALUES ($1, $2, $3)",
      [order.id, order.customerId, order.amount],
    );
    await client.query(
      "INSERT INTO outbox (topic, key, payload, headers, aggregate_type) VALUES ($1, $2, $3, $4, $5)",
      [
        "order.created",
        order.customerId,
        JSON.stringify({ orderId: order.id, amount: order.amount }),
        JSON.stringify({ "X-Tenant": "acme" }),
        "order",
      ],
    );
    await client.query("COMMIT");
  } catch (error) {
    await client.query("ROLLBACK");
    throw error;
  } finally {
    client.release();
  }
}
```

</TabItem>
<TabItem label="C#">

```csharp
var orderId = Guid.NewGuid();

await using var transaction = await db.Database.BeginTransactionAsync(cancellationToken);

db.Orders.Add(new Order { Id = orderId, CustomerId = "cust-42", Amount = 99.99m });
db.OutboxMessages.Add(new OutboxMessage
{
    Id = Guid.NewGuid(),
    Topic = "order.created",
    Key = "cust-42",
    Payload = JsonSerializer.Serialize(new { orderId, amount = 99.99m }),
    Headers = """{"X-Tenant":"acme"}""",
    AggregateType = "order",
});

await db.SaveChangesAsync(cancellationToken);
await transaction.CommitAsync(cancellationToken);
```

The next section holds the `OutboxMessage` entity and its mapping.

</TabItem>
</Tabs>

### The Entity Framework Core mapping

Entity Framework Core writes the properties of an entity, not the columns of a table. An entity without a `Headers` property or an `AggregateType` property never writes those columns. The default convention also does not turn `AggregateType` into `aggregate_type`, so name every column explicitly.

```csharp
public class OutboxMessage
{
    public Guid Id { get; set; }
    public string Topic { get; set; } = default!;
    public string? Key { get; set; }
    public string Payload { get; set; } = default!;
    public string Headers { get; set; } = "{}";
    public string? AggregateType { get; set; }
}

public class OutboxMessageConfiguration : IEntityTypeConfiguration<OutboxMessage>
{
    public void Configure(EntityTypeBuilder<OutboxMessage> builder)
    {
        builder.ToTable("outbox");
        builder.Property(m => m.Id).HasColumnName("id");
        builder.Property(m => m.Topic).HasColumnName("topic");
        builder.Property(m => m.Key).HasColumnName("key");
        builder.Property(m => m.Payload).HasColumnName("payload").HasColumnType("jsonb");
        builder.Property(m => m.Headers).HasColumnName("headers").HasColumnType("jsonb");
        builder.Property(m => m.AggregateType).HasColumnName("aggregate_type");
    }
}
```

- `HasColumnType("jsonb")` applies to PostgreSQL only. On SQL Server, use `nvarchar(max)` for `Payload` and `Headers`, and `nvarchar(255)` for `AggregateType`.
- Set `Headers` to a JSON object string, never to `null`. The column is `NOT NULL`.
- Map `AggregateType` even when no destination reads it today. An unmapped property leaves the column empty for a destination that reads it later.
- QueueBox owns and migrates the `outbox` table. Do not generate an Entity Framework Core migration from this entity.

[Use QueueBox from an EF Core app](/how-to/use-entity-framework-core/) shows the SQL Server mapping, and the inbox side with `InboxDbContextFactory`.

## Next steps

- [Fan out over HTTP](/how-to/fan-out-over-http/) routes the rows to HTTP endpoints.
- [Bridge a broker](/how-to/bridge-a-broker/) publishes the rows to RabbitMQ, Kafka or NATS.
- [Delivery semantics](/concepts/delivery-semantics/) states what QueueBox promises for a committed row.