Custom commands & events

This page shows how to add your own event types next to libspiffy and register them so they load after a restart, and how a new wallet command and event are added to libspiffy itself.

Two audiences, two different jobs:

You are	You can	Where the code lives
An app using libspiffy	Drive the coordinator from your own actors, keep your own event-sourced state, register your own event types	Your app
A contributor to libspiffy	Add commands and events to the wallet aggregate, its projection and the coordinator	A fork or a pull request to libspiffy

An app cannot add a command to the wallet itself. BitcoinWalletAggregate decides which commands it handles in a switch inside lib/src/core/bitcoin_wallet_aggregate.dart, and WalletStateBuilder, which applies events to the wallet’s state, is not exported. Extending the wallet means changing libspiffy.

For apps

Drive libspiffy from your own actors

If your app already runs a Dactor actor system, pass it to libspiffy so its actors join yours, then spawn your own actors that talk to libspiffy’s coordinator:

// A sketch: PaymentProcessorActor is your own actor, not part of libspiffy.
import 'package:dactor/dactor.dart';
import 'package:libspiffy/libspiffy.dart';

final hostSystem = LocalActorSystem(ActorSystemConfig());
final libspiffy = LibSpiffyActorSystem();
await libspiffy.initialize(actorSystem: hostSystem, dataDirectory: './wallet-data');

final processor = await hostSystem.spawn(
  'payment-processor',
  () => PaymentProcessorActor(coordinator: libspiffy.coordinator),
);

// libspiffy.shutdown() leaves your actor system running; shut it down yourself afterwards.

libspiffy.coordinator is a WalletCoordinator, not an ActorRef, so give your actor a field of that type. Your actor sends a command with ask and awaits its reply, and follows events nobody requested with on<E>():

// A sketch: PaymentProcessorActor and ProcessOrder are your own, not part of libspiffy.
import 'package:dactor/dactor.dart';
import 'package:libspiffy/coordinator.dart';

class ProcessOrder {
  final String walletId;
  final BigInt amount;
  ProcessOrder(this.walletId, this.amount);
}

class PaymentProcessorActor extends Actor {
  final WalletCoordinator coordinator;

  PaymentProcessorActor({required this.coordinator});

  @override
  Future<void> preStart() async {
    coordinator.on<InvoicePaidEvent>().listen((paid) => print('paid: ${paid.invoiceId}'));
  }

  @override
  Future<void> onMessage(dynamic message) async {
    if (message is ProcessOrder) {
      try {
        final invoice = await coordinator.ask(
            CreateInvoiceCommand(walletId: message.walletId, amount: message.amount));
        print('invoice ${invoice.invoiceId}: pay ${invoice.addresses.first}');
      } on CoordinatorFailure catch (failure) {
        print('order refused: ${failure.message}');
      }
    }
  }
}

An actor that awaits inside onMessage handles its next message only when the reply has arrived. If your actor must stay responsive, start the ask without awaiting it there.

libspiffy.ownsActorSystem tells you whether libspiffy created the actor system or joined yours.

Register your own event types

libspiffy stores events with eventador, which serializes them to CBOR. After a restart, eventador rebuilds each stored event from its type name through EventRegistry. An event type that is not registered fails with ArgumentError: Event type XYZ not registered.

libspiffy registers its own wallet, invoice and payment channel events during initialize() (LibSpiffyActorSystem.registerEventTypes()). If your app keeps its own event-sourced state with eventador, register your event types before you initialize libspiffy:

import 'package:eventador/eventador.dart';

void registerMyEvents() {
  EventRegistry.register<LoyaltyPointsAwarded>(
    LoyaltyPointsAwarded.stableTypeName,
    LoyaltyPointsAwarded.fromMap,
  );
}

void main() async {
  registerMyEvents();
  await libspiffy.initialize(dataDirectory: './wallet-data');
}

Follow the same rules libspiffy follows for its own events:

• Give each event a stable typeName. eventador stores an event under Event.typeName, which defaults to the Dart class name. A rename, or a build with --obfuscate, changes that name and orphans the stored events. Override it with a constant and register under the same string.
• Keep your names out of libspiffy’s namespaces. EventRegistry is one registry for the whole process, and libspiffy’s events are named wallet.…, invoice.… and channel.…. Use your own prefix.
• Renamed a type that is already stored? Register the old name as an alias: EventRegistry.register<T>(newName, fromMap, aliases: const ['OldName']), or EventRegistry.registerAlias(oldName, newName).

// A sketch of an app event on eventador's base class; see eventador's docs for aggregates and projections.
class LoyaltyPointsAwarded extends Event with SerializableEvent {
  static const String stableTypeName = 'myapp.loyalty.points_awarded';

  @override
  String get typeName => stableTypeName;

  final String customerId;
  final int points;

  LoyaltyPointsAwarded({required this.customerId, required this.points, super.eventId, super.timestamp, super.version});

  @override
  Map<String, dynamic> getEventData() => {'customerId': customerId, 'points': points};

  static LoyaltyPointsAwarded fromMap(Map<String, dynamic> map) => LoyaltyPointsAwarded(
        customerId: map['customerId'] as String,
        points: map['points'] as int,
        eventId: map['eventId'] as String?,
        version: map['version'] as int?,
      );
}

For contributors to libspiffy

A wallet feature follows the CQRS path: a command goes to BitcoinWalletAggregate, which checks its rules and returns events; the events are journaled, applied to the aggregate’s state, and projected into the read model. Never write to storage from an aggregate or a coordinator.

1. The command

Add it to lib/src/core/wallet_commands.dart:

class MyNewCommand extends WalletCommand {
  final String someParameter;

  MyNewCommand({
    required super.walletId,
    required this.someParameter,
    super.commandId,
    super.timestamp,
    super.metadata,
  });

  @override
  String get commandType => 'MyNewCommand';
}

2. The event

Add it to lib/src/core/wallet_events.dart. Every wallet event declares a stableTypeName that never changes, and a static fromMap for loading it after a restart.

class MyNewEvent extends WalletEvent {
  /// Journal identifier of this event type. Never change it.
  static const String stableTypeName = 'wallet.my_new';

  @override
  String get typeName => stableTypeName;

  final String someData;

  MyNewEvent({required super.walletId, required this.someData, super.eventId, super.timestamp, super.version});

  @override
  Map<String, dynamic> getWalletEventData() => {'someData': someData};

  static MyNewEvent fromMap(Map<String, dynamic> map) => MyNewEvent(
        walletId: map['walletId'] as String,
        someData: map['someData'] as String,
        eventId: map['eventId'] as String?,
        timestamp: map['timestamp'] is String
            ? DateTime.parse(map['timestamp'] as String)
            : map['timestamp'] as DateTime?,
        version: map['version'] as int?,
      );
}

3. Register it

Add a line to LibSpiffyActorSystem.registerEventTypes() in lib/src/actors/libspiffy_actor_system.dart:

EventRegistry.register<MyNewEvent>(MyNewEvent.stableTypeName, MyNewEvent.fromMap);

4. Handle the command

In BitcoinWalletAggregate.handleCommand (lib/src/core/bitcoin_wallet_aggregate.dart), add a case that checks the rules against the current state and returns events. It changes no state itself.

case final MyNewCommand cmd:
  if (!currentState.isCreated) throw StateError('Wallet not yet created');
  return [MyNewEvent(walletId: cmd.walletId, someData: cmd.someParameter, version: currentState.version + 1)];

5. Apply the event

In BitcoinWalletAggregate.applyEvent, the current state is never modified: the event is applied to a draft from current.toBuilder(), and build() returns the new state. Add the new field to WalletState and WalletStateBuilder, and a case that fills it in.

6. Project it, if the read model needs it

In lib/src/projections/wallet_projection.dart, add the event to interestedEventTypes and a case to handle() that updates the read model through ReadModelStorage. A new kind of row needs a method on the ReadModelStorage interface and on every backend (in-memory, Isar, Postgres).

7. Expose it to apps

Apps reach the wallet only through the coordinator, in lib/src/actors/coordinator_messages.dart:

• The reply extends CoordinatorReply. It holds a nullable requestId, and overrides failure to return why the request failed, or null on success. WalletCoordinator.ask throws CoordinatorFailure when failure is not null.
• The command extends CoordinatorRequest<R> with that reply as R, takes super.requestId, and overrides replyTimeout when the work takes longer than CoordinatorRequest.defaultTimeout (1 minute).

class MyNewCoordinatorCommand extends CoordinatorRequest<MyNewDoneEvent> {
  final String walletId;
  final String someParameter;

  MyNewCoordinatorCommand({required this.walletId, required this.someParameter, super.requestId});

  @override
  Map<String, dynamic> get metadata => {'walletId': walletId};
}

class MyNewDoneEvent extends CoordinatorReply {
  @override
  final String walletId;
  @override
  final String? requestId;
  final bool success;
  final String? error;

  MyNewDoneEvent({required this.walletId, required this.success, this.error, this.requestId});

  @override
  String? get failure => success ? null : error ?? 'The request failed';
}

Then dispatch the command in WalletCoordinatorActor.onMessage (lib/src/actors/wallet_coordinator_actor.dart) and send the wallet command to the wallet manager as a WalletCommandMessage. Answer every request with exactly one reply carrying its requestId, on success and on every failure path. Emit it once the read model shows the result, so an app that queries on hearing it sees the change.

Actor messages

Messages between libspiffy’s actors implement Dactor’s Message:

class MyNewMessage implements Message {
  final String data;

  MyNewMessage(this.data);

  @override
  String get correlationId => 'my-new-message-$data';
  @override
  Map<String, dynamic> get metadata => {'data': data};
  @override
  ActorRef? get replyTo => null;
  @override
  DateTime get timestamp => DateTime.now();
}

The receiving actor handles it in its onMessage and may answer with context.sender?.tell(...).

Related

• Script plugins: teaching libspiffy new script types needs no fork
• API reference