WIP [ci skip]

go.mod

@@ -5,6 +5,7 @@ go 1.22
 require (
 	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/dogmatiq/configkit v0.13.0
+	github.com/dogmatiq/dapper v0.5.3
 	github.com/dogmatiq/discoverkit v0.1.2
 	github.com/dogmatiq/dogma v0.13.0
 	github.com/dogmatiq/enginekit v0.10.2
@@ -27,10 +28,10 @@ require (
 
 require (
 	github.com/dogmatiq/cosyne v0.2.0 // indirect
-	github.com/dogmatiq/dapper v0.5.3 // indirect
 	github.com/dogmatiq/dyad v1.0.0 // indirect
 	github.com/dogmatiq/iago v0.4.0 // indirect
 	github.com/dogmatiq/interopspec v0.5.3 // indirect
+	github.com/dogmatiq/jumble v0.1.0 // indirect
 	github.com/dogmatiq/linger v1.1.0 // indirect
 	github.com/dogmatiq/projectionkit v0.6.5 // indirect
 	github.com/fxamacker/cbor/v2 v2.4.0 // indirect

internal/eventstream/append.go

@@ -78,9 +78,9 @@ func (w *worker) handleAppend(
 			return res, nil
 		}
 
-		res, err = w.writeEventsToJournal(ctx, req)
+		pos, rec, err := w.writeEventsToJournal(ctx, req)
 		if err == nil {
-			w.publishEvents(res.BeginOffset, req.Events)
+			w.publishEvents(pos, rec)
 			return res, nil
 		}
 
@@ -171,41 +171,32 @@ func (w *worker) findPriorAppend(
 func (w *worker) writeEventsToJournal(
 	ctx context.Context,
 	req AppendRequest,
-) (AppendResponse, error) {
-	before := w.nextOffset
-	after := w.nextOffset + Offset(len(req.Events))
-
-	if err := w.Journal.Append(
-		ctx,
-		w.nextPos,
-		eventstreamjournal.
-			NewRecordBuilder().
-			WithStreamOffsetBefore(uint64(before)).
-			WithStreamOffsetAfter(uint64(after)).
-			WithEventsAppended(&eventstreamjournal.EventsAppended{
-				Events: req.Events,
-			}).
-			Build(),
-	); err != nil {
-		return AppendResponse{}, err
+) (journal.Position, *eventstreamjournal.Record, error) {
+	pos := w.nextPos
+	rec := eventstreamjournal.
+		NewRecordBuilder().
+		WithStreamOffsetBefore(uint64(w.nextOffset)).
+		WithStreamOffsetAfter(uint64(w.nextOffset) + uint64(len(req.Events))).
+		WithEventsAppended(&eventstreamjournal.EventsAppended{Events: req.Events}).
+		Build()
+
+	if err := w.Journal.Append(ctx, pos, rec); err != nil {
+		return 0, nil, err
 	}
 
-	for index, event := range req.Events {
+	w.nextPos++
+
+	for _, event := range req.Events {
 		w.Logger.Info(
 			"appended event to the stream",
-			slog.Uint64("journal_position", uint64(w.nextPos)),
-			slog.Uint64("stream_offset", uint64(before)+uint64(index)),
+			slog.Uint64("journal_position", uint64(pos)),
+			slog.Uint64("stream_offset", uint64(w.nextOffset)),
 			slog.String("message_id", event.MessageId.AsString()),
 			slog.String("description", event.Description),
 		)
-	}
 
-	w.nextPos++
-	w.nextOffset = after
+		w.nextOffset++
+	}
 
-	return AppendResponse{
-		BeginOffset:            before,
-		EndOffset:              after,
-		AppendedByPriorAttempt: false,
-	}, nil
+	return pos, rec, nil
 }