Closing Streams

Event streams can be permanently closed to prevent further appends. Once a stream is closed, any attempt to append events to it will result in a constraint violation of type StreamClosed.

Why close a stream?

Closing a stream is useful when a logical unit of work is complete and no further events should be recorded for that stream. Examples include:

Finalizing an invoice — once issued, no further line items should be added.
Archiving a case — the case is resolved and the event history is sealed.
Completing an order — the order lifecycle has ended and further mutations are disallowed.

How to close a stream

Call CompleteStream on the event log with the stream type and stream identifier you want to close:

using Cratis.Chronicle.EventSequences;

public class ClosingStreamsInvoiceCloser(IEventLog eventLog)
{
    public async Task CloseInvoiceStream(EventStreamId invoiceStreamId)
    {
        var result = await eventLog.CompleteStream(new EventStreamType("invoices"), invoiceStreamId);

        result.Switch(
            sequenceNumber => Console.WriteLine($"Stream closed at sequence number {sequenceNumber}"),
            error => Console.WriteLine($"Failed to close stream: {error}"));
    }
}

import io.cratis.chronicle.IEventStore
import io.cratis.chronicle.eventSequences.CompleteStreamResult

/**
 * Closes the invoice stream so no further line items can be appended to it once the invoice
 * has been issued.
 */
suspend fun closeInvoiceStream(store: IEventStore, invoiceStreamId: String) {
    when (val result = store.eventLog.completeStream("invoices", invoiceStreamId)) {
        is CompleteStreamResult.Success -> println("Stream closed at sequence ${result.sequenceNumber.value}")
        CompleteStreamResult.AlreadyCompleted -> println("Failed to close stream: already completed")
        CompleteStreamResult.DefaultStreamCannotBeCompleted -> println("Failed to close stream: the default stream cannot be completed")
    }
}

import io.cratis.chronicle.EventStore;

import io.cratis.chronicle.java.EventLogJavaBridge;

class ClosingStreamsIndexCompleteStream {
    // Closes the invoice stream so no further line items can be appended to it. Returns false
    // when the stream was already completed, or when it is the default stream (which can never
    // be completed).
    boolean closeInvoiceStream(EventStore store, String invoiceStreamId) {
        return EventLogJavaBridge.completeStream(store.getEventLog(), "invoices", invoiceStreamId);
    }
}

defmodule MyApp.ClosingStreamsInvoiceCloser do
  alias Chronicle.EventSequences.EventLog

  def close_invoice_stream(invoice_stream_id) do
    case EventLog.complete_stream("invoices", invoice_stream_id) do
      {:ok, sequence_number} ->
        IO.puts("Stream closed at sequence number #{sequence_number}")

      {:error, reason} ->
        IO.puts("Failed to close stream: #{inspect(reason)}")
    end
  end
end

import { IEventLog } from '@cratis/chronicle';

async function closeInvoiceStream(eventLog: IEventLog, invoiceStreamId: string): Promise<void> {
    const result = await eventLog.completeStream('invoices', invoiceStreamId);

    if (result.isSuccess) {
        console.log(`Stream closed at sequence number ${result.sequenceNumber.value}`);
    } else {
        console.log(`Failed to close stream: ${result.error}`);
    }
}

The method returns Result<EventSequenceNumber, CompleteStreamError>.

Error cases

Error	Meaning
`AlreadyCompleted`	The stream has already been closed.
`DefaultStreamCannotBeCompleted`	The default stream (`EventStreamType.All` / `EventStreamId.Default`) cannot be closed.

What happens after closing

After a stream is closed, any append targeting that stream is rejected with a StreamClosed constraint violation:

using Cratis.Chronicle.Events;
using Cratis.Chronicle.Events.Constraints;
using Cratis.Chronicle.EventSequences;
using System.Linq;

[EventType]
public record ClosingStreamsInvoiceLineAdded(string Description, decimal Amount);

public class ClosingStreamsInvoiceLineAppender(IEventLog eventLog)
{
    public async Task<bool> TryAppendLine(EventSourceId invoiceId)
    {
        var appendResult = await eventLog.Append(
            invoiceId,
            new ClosingStreamsInvoiceLineAdded("Consulting", 500m),
            new EventStreamType("invoices"),
            new EventStreamId("invoice-42"));

        if (!appendResult.IsSuccess)
        {
            var violation = appendResult.ConstraintViolations
                .FirstOrDefault(v => v.ConstraintType == ConstraintType.StreamClosed);
            return violation is null;
        }

        return true;
    }
}

import io.cratis.chronicle.IEventStore
import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.eventSequences.AppendOptions

@EventType
data class ClosingStreamsInvoiceLineAdded(val description: String = "", val amount: Double = 0.0)

/**
 * Appends a line item to an invoice stream. Once the stream has been closed, the append is
 * rejected with a "StreamClosed" constraint violation and no further lines can be added.
 */
suspend fun tryAppendLine(store: IEventStore, invoiceId: String): Boolean {
    val result = store.eventLog.append(
        invoiceId,
        ClosingStreamsInvoiceLineAdded("Consulting", 500.0),
        AppendOptions(eventStreamType = "invoices", eventStreamId = "invoice-42")
    )

    if (!result.isSuccess) {
        return result.constraintViolations.none { it.constraintId == "StreamClosed" }
    }

    return true
}

import io.cratis.chronicle.EventStore;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.eventSequences.AppendOptions;
import io.cratis.chronicle.eventSequences.AppendResult;
import io.cratis.chronicle.eventSequences.ConstraintViolation;

import io.cratis.chronicle.java.AppendOptionsBuilder;
import io.cratis.chronicle.java.EventLogJavaBridge;

@EventType
record ClosingStreamsInvoiceLineAdded(String description, double amount) {}

class ClosingStreamsIndexAppendRejected {
    // Appends a line item to an invoice stream. Once the stream has been closed, the append is
    // rejected with a "StreamClosed" constraint violation and no further lines can be added.
    boolean tryAppendLine(EventStore store, String invoiceId) {
        AppendOptions options = new AppendOptionsBuilder()
            .eventStreamType("invoices")
            .eventStreamId("invoice-42")
            .build();

        AppendResult result = EventLogJavaBridge.append(
            store.getEventLog(), invoiceId, new ClosingStreamsInvoiceLineAdded("Consulting", 500.0), options);

        if (!result.isSuccess()) {
            for (ConstraintViolation violation : result.getConstraintViolations()) {
                if (violation.getConstraintId().equals("StreamClosed")) {
                    return false;
                }
            }
        }

        return true;
    }
}

defmodule MyApp.Events.ClosingStreamsInvoiceLineAdded do
  use Chronicle.Events.EventType, id: "closing-streams-invoice-line-added"

  defstruct [:description, :amount]
end

defmodule MyApp.ClosingStreamsInvoiceLineAppender do
  alias Chronicle.EventSequences.EventLog
  alias MyApp.Events.ClosingStreamsInvoiceLineAdded

  def try_append_line(invoice_id) do
    case EventLog.append(
           invoice_id,
           %ClosingStreamsInvoiceLineAdded{description: "Consulting", amount: 500},
           event_stream_type: "invoices",
           event_stream_id: "invoice-42"
         ) do
      :ok ->
        true

      {:error, {:constraint_violations, violations}} ->
        not Enum.any?(violations, &stream_closed?/1)

      {:error, _reason} ->
        true
    end
  end

  # Mirrors the wire Constraint's Type field (seen as :Unique / :UniqueEventType
  # when registering constraints) — a rejection caused by a closed stream comes
  # back as a violation whose type is :StreamClosed.
  defp stream_closed?(violation) when is_map(violation) do
    Map.get(violation, :Type) == :StreamClosed or Map.get(violation, :type) == :stream_closed
  end

  defp stream_closed?(_violation), do: false
end

import { eventType, IEventLog } from '@cratis/chronicle';

@eventType()
class ClosingStreamsInvoiceLineAdded {
    constructor(readonly description: string = '', readonly amount: number = 0) {}
}

async function tryAppendLine(eventLog: IEventLog, invoiceId: string): Promise<boolean> {
    const [appendResult] = await eventLog.appendMany([{
        eventSourceId: invoiceId,
        event: new ClosingStreamsInvoiceLineAdded('Consulting', 500),
        eventStreamType: 'invoices',
        eventStreamId: 'invoice-42'
    }]);

    if (!appendResult.isSuccess) {
        const wasStreamClosed = appendResult.constraintViolations.some(violation => violation.constraintId === 'StreamClosed');
        return !wasStreamClosed;
    }

    return true;
}

The rejection is enforced by the ClosedStreamConstraintValidator which is automatically active for every event sequence — no additional configuration is required.