# `Issho.Streams`
[🔗](https://github.com/hattoriai/issho/blob/v0.4.0/lib/issho/streams.ex#L1)

Streams of a room: output that a process writes a piece at a time (a
model's answer, an agent's report, an import's log), that people see as
it comes, and find again when they come back: from another page,
another tab, after their connection dropped, on another node.

    {:ok, stream} =
      Issho.Streams.start(MyApp.Issho, "notebook:7",
        kind: "summary",
        for: ada,
        place: %{"doc" => "notebook:7/intro", "at" => %{"node" => block_id}},
        by: claude
      )

    :ok = Issho.Streams.append(stream, "## Summary\n\n")
    :ok = Issho.Streams.append(stream, "The plan holds…")
    :ok = Issho.Streams.finish(stream)

Issho keeps a stream's chunks and sends them on; it does not read them.
What a stream is, each component or app decides: a suggestion in a
text, a card waiting on a canvas, a chat's answer, a log.

## Who sees it

`:for` is required: a person (an actor, or a key), who sees it, and so
do their agents (`on_behalf_of`); `:room`, everyone the policy lets
read the room; or `:policy`, whomever the policy allows the action
`:read_stream` in the room. Its writer (`:by`) sees it too. Every reader
may read the room (`:read`).

## Seen as it comes, found again

`subscribe/4` gives the stream as it is now, its content so far, and
from then on sends the subscriber what comes, in order, with nothing
missed or sent twice:

  * `{Issho, :stream, id, {:chunk, data}}`
  * `{Issho, :stream, id, :finished}`
  * `{Issho, :stream, id, {:failed, reason}}`
  * `{Issho, :stream, id, {:closed, outcome}}`
  * `{Issho, :stream, id, :gone}`: kept no longer.

`watch/3` subscribes to every stream of a room the caller may see, and
to the news of new ones: `{Issho, :stream_started, room, stream}`, on
which the watcher calls `subscribe/4` (it is refused a stream it may not
see). A LiveView does it with `Issho.Live.streams/1`, which does both.

## Its life

A stream lives while it is written. Finished (`finish/1`) or failed
(`fail/2`), it is kept for `:keep` milliseconds (the instance's
`:stream_keep`, five minutes), for those who come back, then goes.
`close/2` ends it at once, with an outcome the app names (`"accepted"`,
`"rejected"`, `"stopped"`, its own): a suggestion taken or turned down.

Its writer is the process that starts it (or `:writer`): when it exits
while the stream is written, the stream fails (`"writer_exited"`), or
finishes with `on_exit: :finish`. No stream is left being written.

Each start, finish, failure and close is an event of the room
(`Issho.Event`): `"stream.started"`, `"stream.finished"`,
`"stream.failed"` and `"stream.closed"`, with `data` `%{stream: id,
kind: kind, for: for}` and the `reason` or `outcome`; so an agent learns
whether its draft was taken. Their content is not in them.

## Limits

A stream holds at most the instance's `:stream_max_bytes` (1 MB): one
that would hold more fails (`"too_large"`). A room has at most
`:room_max_streams` (32) at once. Streams are in memory, as presence
is: a stream does not outlive its node.

Each stream is a process (`:gen_statem`), named in the whole cluster
(`:global`), so it is found from any node; the streams of a room are a
process group (`:pg`).

# `ref`

```elixir
@type ref() :: Issho.Stream.t() | {atom(), String.t()}
```

# `append`

```elixir
@spec append(ref(), iodata()) :: :ok | {:error, term()}
```

Adds a chunk (a binary) to the stream. `{:error, :not_writing}` once it
is done; `{:error, :too_large}` when the stream would hold more than its
limit, and it fails.

# `close`

```elixir
@spec close(ref(), String.t() | atom()) :: :ok | {:error, term()}
```

Closes the stream now, with an outcome the app names (`:accepted`,
`:rejected`, `:stopped`, its own): it goes, and its subscribers hear
how. From a person's page, `Issho.Live.close_stream/3` checks they may
see it first.

# `fail`

```elixir
@spec fail(ref(), String.t() | atom()) :: :ok | {:error, term()}
```

Fails the stream, with a reason (a string or an atom).

# `finish`

```elixir
@spec finish(ref()) :: :ok | {:error, term()}
```

Finishes the stream: it is complete, and kept for those who come back.

# `get`

```elixir
@spec get(atom(), String.t()) :: {:ok, Issho.Stream.t()} | {:error, :not_found}
```

The stream as it is now: `{:ok, stream}` or `{:error, :not_found}`.

# `list`

```elixir
@spec list(atom(), String.t(), Issho.Actor.t() | Issho.Identity.t()) :: [
  Issho.Stream.t()
]
```

The streams of a room that `who` (an actor or an identity) may see, as
they are now, oldest first.

# `start`

```elixir
@spec start(atom(), String.t(), keyword()) ::
  {:ok, Issho.Stream.t()} | {:error, term()}
```

Starts a stream in a room, written by the caller (or `:writer`).
Options: `:for` (required), `:kind`, `:place`, `:meta`, `:by` (an actor
or an identity: who writes it, as the room sees them), `:writer`,
`:keep` (milliseconds), `:on_exit` (`:fail` or `:finish`).

# `subscribe`

```elixir
@spec subscribe(
  atom(),
  String.t() | Issho.Stream.t(),
  Issho.Actor.t() | Issho.Identity.t(),
  pid()
) ::
  {:ok, Issho.Stream.t()} | {:error, :forbidden | :not_found}
```

Subscribes `pid` (the caller) to a stream that `who` may see: gives the
stream as it is now, then sends what comes (see above).
`{:error, :forbidden}` for one they may not see.

# `unsubscribe`

```elixir
@spec unsubscribe(ref(), pid()) :: :ok
```

Stops sending a stream to `pid` (the caller).

# `unwatch`

```elixir
@spec unwatch(atom(), String.t()) :: :ok
```

Stops the news of a room's new streams (`watch/3`).

# `visible?`

```elixir
@spec visible?(Issho.Stream.t(), Issho.Identity.t()) :: boolean()
```

Whether `who` (an identity) may see the stream: they may read its room,
and it is for them (or for whom they act), for the room, for whom the
policy allows `:read_stream`, or they write it.

# `watch`

```elixir
@spec watch(atom(), String.t(), Issho.Actor.t() | Issho.Identity.t()) ::
  {:ok, [Issho.Stream.t()]}
```

Subscribes the caller to every stream of the room that `who` may see,
and to the news of new ones (`{Issho, :stream_started, room, stream}`).
Gives the streams, as they are now.

---

*Consult [api-reference.md](api-reference.md) for complete listing*