Go client

tephra-go is the official Go client. It is a single, concurrent-safe Client that multiplexes requests over a control socket and a pool of bulk read sockets, and it is safe to share across goroutines.

go get github.com/tephradb/tephra-go

Requires Go 1.23 or newer (it uses iter.Seq2) and a Tephra server on 0.4 or above, which speaks the mandatory Hello handshake this client opens with.

Quick start

Dial connects and runs the handshake on every socket; Close shuts them all down. A context.Context bounds the dial, not the later operations.

ctx := context.Background()

client, err := tephra.Dial(ctx, "127.0.0.1:9000")
if err != nil {
  log.Fatal(err)
}
defer client.Close()

event, err := tephra.NewEvent("CourseOpened", []string{"course:c1"}, []byte(`{"seats":30}`))
if err != nil {
  log.Fatal(err)
}
if _, err := client.Append(ctx, []tephra.Event{event}, nil); err != nil {
  log.Fatal(err)
}

events, watermark, err := client.ReadAll(ctx, tephra.QueryAll(), tephra.Zero, nil)
if err != nil {
  log.Fatal(err)
}
for _, e := range events {
  fmt.Println(e.Position, e.Type())
}
_ = watermark

NewEvent validates the type and tags exactly as the server does. Zero is the start cursor and a nil limit reads the whole matching history. The event and query model is the same across every client; Core concepts is the precise definition.

Append with a condition

An AppendCondition guards the append with two checks, OR’d. The boundary check is a dynamic consistency boundary: it rejects the append if any event matching FailIfEventsMatch already exists after its After position. Leave After at Zero for the uniqueness-guard pattern, “fail if this has ever happened”.

item, err := tephra.WithTags("email:a@example.com")
if err != nil {
  log.Fatal(err)
}
event, err := tephra.NewEvent("Registered", []string{"email:a@example.com"}, []byte(`{}`))
if err != nil {
  log.Fatal(err)
}

cond := tephra.NewAppendCondition(tephra.QueryItems(item))
_, err = client.Append(ctx, []tephra.Event{event}, &cond)

var serverErr *tephra.ServerError
if errors.As(err, &serverErr) && serverErr.Code == tephra.ErrCodeConflict {
  fmt.Println("email already registered")
}

Set the After field when you have already read up to a position and only need to guard against events since. See Handling conflicts for the retry contract on a same-batch versus a durable conflict.

Idempotent appends

The optional existence check, FailIfExists, rejects the append if any event matches its query anywhere in the log, independent of the boundary’s After. It is the idempotency/dedupe guard: a single After cannot be both a moving decision boundary and a whole-log uniqueness assertion, so this is the second, separate check. Its conflict comes back as ErrCodeAlreadyExists, distinct from a boundary ErrCodeConflict, so a retried command can be treated as “already applied” (a no-op) rather than “rebuild and retry”. Use ExistsOnly for the pure-dedupe case, or set FailIfExists alongside a boundary query to assert both in one append.

key, err := tephra.WithTags("cmd:order-42")
if err != nil {
  log.Fatal(err)
}
event, err := tephra.NewEvent("OrderPlaced", []string{"cmd:order-42"}, []byte(`{}`))
if err != nil {
  log.Fatal(err)
}

cond := tephra.ExistsOnly(tephra.QueryItems(key))
_, err = client.Append(ctx, []tephra.Event{event}, &cond)

var serverErr *tephra.ServerError
switch {
case err == nil:
  fmt.Println("order placed")
case errors.As(err, &serverErr) && serverErr.Code == tephra.ErrCodeAlreadyExists:
  fmt.Println("order already placed, treating as a no-op")
default:
  log.Fatal(err)
}

See Idempotency and deduplication for the full pattern.

Reads and pagination

Read returns a ReadStream driven by Next/Event, with Err checked at the end and Watermark available once it closes. ReadAll drains one into a slice. ReadBack and ReadAllBack are the newest-first duals, taking a Before upper bound (Max starts at the tip).

after (exclusive) and a limit compose into a stateless pagination cursor:

cursor := tephra.Zero
for {
  page, _, err := client.ReadAll(ctx, query, cursor, tephra.Limit(100))
  if err != nil {
    log.Fatal(err)
  }
  if len(page) == 0 {
    break
  }
  for _, e := range page {
    handle(e)
  }
  cursor = page[len(page)-1].Position // next page resumes here, no gap or duplicate
}

A range-over-func helper streams without a manual loop:

for event, err := range client.ReadSeq(ctx, tephra.QueryAll(), tephra.Zero, nil) {
  if err != nil {
    log.Fatal(err)
  }
  fmt.Println(event.Position, event.Type())
}

Subscriptions

Subscribe catches up on matching events, then tails new ones live, delivering a caught-up marker each time it reaches the live edge.

sub, err := client.Subscribe(ctx, tephra.QueryAll(), tephra.Zero)
if err != nil {
  log.Fatal(err)
}
defer sub.Close()

for sub.Next() {
  item := sub.Item()
  if item.IsCaughtUp() {
    continue
  }
  handle(item.Event)
}
if err := sub.Err(); err != nil {
  log.Fatal(err)
}

Cancel a stream by calling Close or by cancelling the context.Context you passed in. Either sends a best-effort cancel so the server stops producing frames.

Server stats

Stats returns a point-in-time snapshot: the event, segment, and on-disk-byte counts, uptime, and the live connection and subscription counts.

stats, err := client.Stats(ctx)
if err != nil {
  log.Fatal(err)
}
fmt.Printf("%d events across %d segments\n", stats.EventCount, stats.SegmentCount)

TLS

The server can serve TLS 1.3 (server-authenticated). Enable it with WithTLS, passing a standard *tls.Config. ServerName defaults to the dial host, so a hostname certificate needs no extra configuration.

import "crypto/tls"

// Verify against the system roots (a public CA):
client, err := tephra.Dial(ctx, "tephra.example.com:9000", tephra.WithTLS(&tls.Config{}))

// Or trust a private CA, and optionally present a client certificate for mutual TLS:
client, err = tephra.Dial(ctx, "tephra.internal:9000", tephra.WithTLS(&tls.Config{
  RootCAs:      privateCAs,
  Certificates: []tls.Certificate{clientCert},
}))

Authentication

When the server requires bearer tokens, present one with WithAuthToken. Each socket authenticates in its own Hello, so a rejected token fails Dial with a *ServerError whose Code is ErrCodeUnauthenticated, rather than on the first request. Pair it with WithTLS so the token is not sent in the clear; a plaintext token is accepted only by a server explicitly configured to allow it.

client, err := tephra.Dial(ctx, "tephra.internal:9000",
  tephra.WithTLS(&tls.Config{RootCAs: privateCAs}),
  tephra.WithAuthToken(os.Getenv("TEPHRA_TOKEN")),
)

Options

Dial takes functional options; the defaults match the reference Rust client.

Option	Default	Meaning
WithBulkConnections(n)	4	Dedicated bulk sockets for reads and subscriptions. 0 folds reads onto the control socket.
WithMaxInflightRequests(n)	1024	Outstanding requests per socket before backpressure.
WithRequestQueueDepth(n)	256	Outbound queue depth per socket.
WithMaxFrameLen(n)	16 MiB	Largest frame accepted or produced.
WithDialer(d)	standard	A custom net.Dialer; TCP_NODELAY is always set.
WithTLS(cfg)	off	Wrap each connection in a TLS client session.
WithAuthToken(token)	off	Bearer token presented in each socket’s Hello.

Errors

The client does no automatic retries or reconnection: it surfaces a typed error and leaves the policy to you. Match with errors.As.

• *ServerError: the server returned an error. Code is an ErrorCode (ErrCodeConflict for a boundary conflict, ErrCodeAlreadyExists for an existence-clause match), Retryable marks an advisory same-batch append conflict, and ConflictPosition is set for a durable one.
• *ProtocolError: the peer sent something outside the protocol.
• *ConnError: the connection failed with requests in flight; every one is failed with it rather than left hanging. It unwraps to the underlying cause.
• ErrClosed: the client was closed.