# Streamed mutations

> A mutation that returns an async iterable fills in as it streams, with chunksOptions(), and stops on unmount or on demand; long jobs pair a mutation with a subscription.

A mutation that returns an async iterable (an AI reply, say) can fill in as it streams:

```ts
const send = useMutation(trpc.chat.send.streamedMutationOptions());
const chunks = useQuery(trpc.chat.send.chunksOptions());

send.mutate({ text });
// chunks.data: ['ok'] → ['ok', 'hi'], then send.data is ['ok', 'hi']
```

`chunksOptions()` never fetches; it reads what the mutation writes to `streamedMutationKey()`. Each `mutate` starts the chunks over. Pass `onChunk(chunk, chunks)` to act on each one.

## Stopping the stream

Resetting the mutation doesn't stop the stream on the server. Two options do, and in both `data` is the chunks so far:

```ts
// stop when the last chunksOptions() reader unmounts
trpc.chat.send.streamedMutationOptions({ abortWhenUnobserved: true });

// stop on demand: a function gives each mutate() its own signal
const stop = useRef<AbortController>(undefined);
trpc.chat.send.streamedMutationOptions({
  signal: () => (stop.current = new AbortController()).signal,
});
stop.current?.abort();
```

## Long jobs

A stream tied to one screen fits a streamed mutation. A job that should outlive the screen, be resumed after a reload, or be watched from several tabs fits a mutation that starts it, a `tracked()` subscription for its progress, and a mutation that cancels it:

```ts
const job = {
  start: t.procedure.input(Params).mutation((opts) => jobs.start(opts.input)), // { id }
  progress: t.procedure
    .input(z.object({ id: z.string() }))
    .subscription(async function* (opts) {
      const steps = jobs.watch(opts.input.id, { after: opts.lastEventId });
      for await (const step of steps) yield tracked(step.id, step);
    }),
  cancel: t.procedure
    .input(z.object({ id: z.string() }))
    .mutation((opts) => jobs.cancel(opts.input.id)),
};

const start = useMutation(trpc.job.start.mutationOptions());
const progress = useSubscription(
  trpc.job.progress.subscriptionOptions({
    input: start.data ? { id: start.data.id } : skipToken,
  }),
);
```

Unsubscribing stops watching, not the job. Keep jobs in a [Publisher](https://v12.trpc.io/docs/subscriptions/publisher.md) so any instance can stream them.