README

PHP client for TamarackDB, an event store compliant with the DCB specification.

It covers the whole HTTP API: transactions, reading and writing events, projections, projection rebuilds, and pauses. It is a low-level client, meant to be used by an event sourcing framework or directly by an application. It is tested against TamarackDB v0.43.0.

Requirements

PHP 8.5 or later, with the curl and json extensions.
A running TamarackDB server, over TCP or its unix socket.

Installation

composer require tamarackdb/tamarackdb-php

Connecting

use TamarackDB\Client;

$client = Client::http('http://127.0.0.1:8085');
$client = Client::unixSocket('/run/tamarackdb/tamarackdb.sock');

// With authTokenFile set on the server, and your own limits:
$client = Client::http('http://127.0.0.1:8085', token: 'secret', timeout: 120.0);

timeout is how long the client waits for a response (60 seconds by default). It includes the time a request waits for its turn in the server's queue: a commit, writeProjections(), the bulk deletes, pause(), resume(), and optimize(). Past it, the client throws a TimeoutException.

Handling a command

A command runs in one transaction, which lives in the server. Each decision reads events, then writes its events, or none. Your event handlers react, your projections are read and written, and the commit writes everything at once, or nothing. How it works is in Transactions.

use TamarackDB\Event\NewEvent;
use TamarackDB\Projection\TxProjectionWrites;
use TamarackDB\Query\Identifier;
use TamarackDB\Query\Query;

$tx = $client->beginTransaction();
try {
    // One decision: one read, then one write.
    foreach ($tx->readEvents(new Query(Identifier::is('userId', $userId))) as $event) {
        // Build your decision model from $event.
    }
    $result = $tx->appendEvents([
        new NewEvent('user-renamed', ['userId' => $userId], ['tenantId' => 'acme'], json_encode(['name' => $name])),
    ]);
    // Give $result->time to the events before your event handlers react to them.

    // Projections, once the events are written. A replace needs a read.
    $tx->getProjection('user-profile', $userId);
    $tx->writeProjections(
        new TxProjectionWrites()->replace('user-profile', $userId, json_encode(['name' => $name])),
    );

    $tx->commit();
} catch (\Throwable $e) {
    $tx->rollback();
    throw $e;
}

beginTransaction() returns the transaction. $client->getTransaction() returns it too, while it's active, and $client->inTransaction() tells whether there is one. A client holds at most one active transaction.
After a ConcurrencyException or a TransactionNotFoundException, run the whole command again, in a new transaction.
Any server error ends the transaction, except a missing projection and a TransactionBusyException. $tx->isActive() then returns false, and every call but rollback() throws a NoActiveTransactionException. The same goes once the transaction is committed or rolled back.
A transport failure leaves the transaction active on the client, since the call may not have reached the server. Call rollback().
rollback() never throws, and does nothing on a transaction that is already over: it's safe in any error handler.
If a commit's response is lost, the commit can't be sent again. See A lost response.

Reading events

In a transaction

$tx->readEvents() returns a list: every committed event that matches (Event), then every event written earlier in the transaction that matches (PendingEvent). The whole response is read before it returns.

foreach ($tx->readEvents($query) as $event) {
    $event->time;                   // DateTimeImmutable, UTC
    $event->type;                   // string
    $event->identifiers;            // ['userId' => '123', 'courseId' => ['a', 'b']]
    $event->metadata;               // ['tenantId' => 'acme']
    $event->payload;                // string, exactly as written
    if ($event instanceof Event) {
        $event->sequence;           // int; a PendingEvent gets its own at commit
    }
}

The next call after a read must be appendEvents(), with the events of the decision or an empty list. A decision that rests on no event reads new NoEvents() first. A response cut short, or one the client can't read, abandons the transaction and throws: run the command again.

backwards: true reads the newest events first: the pending events, then the committed ones. limit is the most events returned, pending and committed together. The read's condition still covers every event that matches the query. To give an invoice the next number:

use TamarackDB\Query\EventType;

$tx = $client->beginTransaction();
$last = $tx->readEvents(new Query(EventType::in('invoice-created')), backwards: true, limit: 1);
$number = $last === [] ? 1 : (int) $last[0]->identifiers['invoice'] + 1;
$tx->appendEvents([new NewEvent('invoice-created', ['invoice' => (string) $number])]);
$tx->commit(); // ConcurrencyException when another invoice was created since the read

Outside a transaction

$client->readEvents() reads committed events, for a projector that catches up on its own or a projection rebuild. It returns Events, which you iterate once. Pages are fetched as you iterate, and a page cut short is resumed after the last event received, so no event is skipped or repeated.

use TamarackDB\Query\AllEvents;

$events = $client->readEvents(new AllEvents(), afterSequence: $last, pageSize: 500);
foreach ($events as $event) {
    // Every event is an Event, with its sequence.
    $last = $event->sequence;
}

To follow new events, keep the last Sequence Position you read, and pass it to the next read.

backwards: true reads the newest events first, in a single page of at most pageSize events. It takes no afterSequence.

Queries

Both reads take a Query, new AllEvents(), or new NoEvents().

use TamarackDB\Query\EventType;
use TamarackDB\Query\Identifier;
use TamarackDB\Query\Metadata;
use TamarackDB\Query\Query;

new Query(
    EventType::in('user-created', 'user-updated'),
    Identifier::is('userId', '123'),
)->or(
    EventType::in('some-other-event'),
    Metadata::is('tenantId', 'acme'),
);

The filters given together form one item, and an event must match all of them. or() adds another item, and an event matching any item matches the query. Within EventType::in(), any of the types matches. Give Identifier::is() or Metadata::is() twice with the same name to require both values. To add a filter to every item of a query, use map() and with(): $query->map(fn (QueryItem $item) => $item->with(Metadata::is('tenantId', 'acme'))).

In identifiers and metadata, a name with one value maps to a string, a name with several values to a list. NewEvent and QueryItem expose them the same way.

Writing events

$result = $tx->appendEvents([
    new NewEvent('user-created', ['userId' => '123'], ['tenantId' => 'acme'], '{"name":"Ada"}'),
]);

The write closes the read before it. appendEvents([]) is the decision to write nothing, and the commit still checks it.
$result->time is the time every event of the write carries, and keeps once committed. The events get their Sequence Position at commit.
The payload is an opaque string: encode it as you like (JSON, XML, ...).

Projections

A projection is an opaque payload identified by type and id. In a transaction, it's written with the events it's computed from (see Projections).

use TamarackDB\Projection\TxProjectionWrites;

$profile = $tx->getProjection('user-profile', '123'); // null when missing
$profile?->payload;
$tx->getProjection('user-list-entry', '456');

$tx->writeProjections(
    new TxProjectionWrites()
        ->create('user-list-entry', '789', '{"name":"Grace"}')
        ->replace('user-profile', '123', '{"name":"Ada Lovelace"}')
        ->delete('user-list-entry', '456'),
);

getProjection() sees the changes the transaction made. A missing projection doesn't end the transaction. Its version is always null: the server keeps the version read.
No write carries a version. A create needs no read: your code knows the projection doesn't exist yet. A replace or a delete needs a read of the projection in the transaction, or a create earlier in it.
The server refuses reads and writes of projections while a read of events waits for its write, so handle projections once the events are written.
Empty writes send nothing.
At commit, a projection changed by another write since it was read, or a created one that already exists, gets a ConcurrencyException.

Projection rebuilds

Outside a transaction, the client reads committed projections with their version, and writes them with writeProjections(), all or nothing:

use TamarackDB\Projection\ProjectionWrites;

$profile = $client->getProjection('user-profile', '123'); // with $profile->version

$result = $client->writeProjections(
    new ProjectionWrites()
        ->create('user-list-entry', '789', '{"name":"Grace"}')
        ->replace('user-profile', '123', $profile->version, '{"name":"Ada Lovelace"}')
        ->delete('user-list-entry', '456', $entryVersion),
);
$result->createVersions;  // new versions, in order
$result->replaceVersions;

replace and delete carry the version read. When it no longer matches, or a created projection already exists, the call throws a ConcurrencyException.

A rebuild is your application's job. The client gives you the calls it needs:

$client->deleteProjectionsByType('user-profile'); // or deleteAllProjections()

foreach ($client->readEvents(new AllEvents()) as $event) {
    // Run your projectors, and every so often:
    // $client->writeProjections($writes);
}

writeProjections() and the bulk deletes wait for their turn in the server's queue. See rebuilds for how to run one.

Pause

$point = $client->pause();  // returns once the pause is in place
$point->lastSequence;       // int

$client->resume();

While transactions are still open, pause() asks the server again every $retryAfter milliseconds (1000 by default): $client->pause(retryAfter: 200). See Pause.

Errors

Every exception implements TamarackDB\Exception\TamarackDBException.

Exception	When
`ConcurrencyException`	409: a commit whose reads or projections changed since, or a projection version that doesn't match
`TransactionBusyException`	409: another call still runs on the transaction, for example one that timed out. The transaction goes on
`InvalidRequestException`	400: the server rejected the request, or a call that breaks a rule of transactions
`UnauthorizedException`	401: missing or wrong token
`TransactionNotFoundException`	404: the transaction expired, or an error ended it
`PayloadTooLargeException`	413: an event, a projection, or the request is too large
`InternalErrorException`	500
`WriteQueueFullException`	503: too many requests are waiting in the server's queue
`ShuttingDownException`	503: the server is shutting down
`PausedException`	503: `beginTransaction()` while a pause is requested or in place
`TooManyTransactionsException`	503: `beginTransaction()` while the server has as many open transactions as it allows
`UnavailableException`	503: `health()` only, storage is unreachable
`TransactionAlreadyActiveException`	`beginTransaction()` while the client has an active transaction
`NoActiveTransactionException`	a call on a transaction that is over, or `getTransaction()` without one
`TransportException`	no full response: server unreachable, connection dropped
`TimeoutException`	a `TransportException`: the client stopped waiting
`ProtocolException`	a response this client can't make sense of
`InvalidArgumentException`	a value that breaks an API rule, caught before sending

Every server error extends ServerException, with $statusCode, $errorCode (such as "ConcurrencyException"), and $detail.

Server state

$client->health();    // Health { status, paused, version }
$client->optimize();  // refreshes the statistics SQLite plans queries with

In development mode, a test suite empties the store with:

$client->deleteAllEvents();       // Sequence Positions go on: the next event doesn't get 1
$client->deleteAllProjections();

Development

composer install
composer test:unit
composer analyse
composer cs

The integration tests start their own tamarackdb-server processes, with devMode on and a new database, on a free port and on a unix socket. They are skipped unless TAMARACKDB_SERVER_BIN points to a server binary, with tamarackdb-init in the same directory. Build both from a clone of the server repo, at the version stated above, and check it with -version:

TAMARACKDB_SERVER_BIN=/path/to/tamarackdb-server composer test

License

MIT, see LICENSE.

tamarackdb / tamarackdb-php

Package info

Security

Statistics