Update

Author: Raezil

README.MD

@@ -185,6 +185,8 @@ type Event struct {
 type Transaction struct {
     store  *EventStore
     events []Event
+    startHead uint64 // head position when transaction began
+
 }
 ```
 

transaction.go

@@ -27,26 +27,86 @@ func (tx *Transaction) Publish(e Event) {
 // Commit enqueues all buffered events and processes them immediately.
 // It returns the first error encountered from Subscribe or handler execution.
 func (tx *Transaction) Commit(ctx context.Context) error {
-	// (same as before)…
-	// at the very end we clear tx.events
-	atomic.StoreUint64(&tx.store.tail, atomic.LoadUint64(&tx.store.head))
+	// 1) Enqueue
+	for _, e := range tx.events {
+		e.Ctx = ctx
+		if err := tx.store.Subscribe(ctx, e); err != nil {
+			return err
+		}
+	}
+
+	// 2) Process synchronously
+	head := atomic.LoadUint64(&tx.store.head)
+	tail := atomic.LoadUint64(&tx.store.tail)
+	disp := *tx.store.dispatcher
+	mask := tx.store.size - 1
+
+	for i := tail; i < head; i++ {
+		ptr := atomic.LoadPointer(&tx.store.buf[i&mask])
+		if ptr == nil {
+			continue
+		}
+		ev := *(*Event)(ptr)
+		if handler, ok := disp[ev.Projection]; ok {
+			// pick up recorded context
+			cctx := ev.Ctx
+			if c2, ok2 := ev.Args["__ctx"].(context.Context); ok2 && c2 != nil {
+				cctx = c2
+			}
+
+			// before hooks
+			for _, hook := range tx.store.beforeHooks {
+				hook(cctx, ev)
+			}
+
+			// wrap middlewares
+			wrapped := handler
+			for j := len(tx.store.middlewares) - 1; j >= 0; j-- {
+				wrapped = tx.store.middlewares[j](wrapped)
+			}
+
+			// invoke handler
+			res, err := wrapped(cctx, ev.Args)
+			atomic.AddUint64(&tx.store.processedCount, 1)
+
+			// after hooks
+			for _, hook := range tx.store.afterHooks {
+				hook(cctx, ev, res, err)
+			}
+
+			if err != nil {
+				atomic.AddUint64(&tx.store.errorCount, 1)
+				for _, hook := range tx.store.errorHooks {
+					hook(cctx, ev, err)
+				}
+				// advance tail and exit on first handler error
+				atomic.StoreUint64(&tx.store.tail, head)
+				return err
+			}
+		}
+	}
+
+	// 3) Advance tail past processed events
+	atomic.StoreUint64(&tx.store.tail, head)
+
+	// 4) Clear transaction buffer
 	tx.events = tx.events[:0]
 	return nil
 }
 
 // Rollback clears the local buffer *and* any events that have already
 // been pushed into the store’s ring-buffer since the transaction began.
 func (tx *Transaction) Rollback() {
-	// 1) clear our local buffer
+	// clear our local buffer
 	tx.events = tx.events[:0]
 
-	// 2) remove any partial enqueues from the store
-	//    by zeroing pointers in the buffer slots we touched…
+	// remove any partial enqueues from the store
 	currHead := atomic.LoadUint64(&tx.store.head)
 	mask := tx.store.size - 1
 	for i := tx.startHead; i < currHead; i++ {
 		atomic.StorePointer(&tx.store.buf[i&mask], nil)
 	}
-	// 3) restore the head pointer
+
+	// restore the head pointer
 	atomic.StoreUint64(&tx.store.head, tx.startHead)
 }