support for transactions

[pgt502]

Hello,

We recently started using GOES and it looks really good.
One thing that we would like to be able to do is to atomically perform a Save operation with another operation. In case, any of the operations fail, they should both be rolled back. I looked at the code and couldnt find a way of achieving this - if it is possible, please point me in the right direction (I saw BeforeInsert and AfterInsert functions but they are not part of the transaction).
Otherwise, we were planning to register a handler (callback function) that would be called during the save operation - inside of the mongo transaction.

Thank you

[bounoable]

Hello,

Thank you for reaching out, and I'm pleased to know you're finding value in the library. Out of curiosity, could you share some of your use cases for goes? I'm interested in how others might use this library.

To address your question about atomic Save operations: While the current implementation of the aggregate repository in the repository package doesn't support this, I believe it's a viable feature to introduce.

Idea

Internally, the repository.Repository.Save() method relies on event.Store.Insert() to add aggregate events to the event store. This insertion is managed within a transaction, ensuring that updates to the states collection remain consistent when inserting events. (Refer to this code for a clearer picture.)

Here's an idea:

1. Introduce Manual Transactions: We could extend the event store to allow users to initiate and manage transactions manually. It might look something like this:

package example

import "github.com/modernice/goes/backend/mongo"

func example(store *mongo.EventStore) {
  tx := store.Transaction(context.TODO())

  err := tx.Execute(func() error {
    // perform some operations
    return nil
  })

  err = tx.Execute(func() error {
    // perform some more operations
    return nil
  })

  if err == nil {
    err = tx.Commit()
  } else {
    err = tx.Abort()
  }
}

2. Enhance Repository Usage: The repository.Repository.Use() method could then leverage the event store's BeginTransaction() (when available) to encapsulate commands within the transaction. If there are nested calls to Use(), the same transaction could be shared, committing only when the outermost Use() invocation returns. This could be used as:

package example

import "github.com/modernice/goes/aggregate"

func example(repo *repository.Repository, foo, bar aggregate.Aggregate) {
  err := repo.Use(context.TODO(), foo, func() {
    // Operations with foo

    return repo.Use(context.TODO(), bar, func() {
      // Operations with both foo and bar
      return nil
    })
  })
}

Do you think this approach would meet your requirements?

Best

[pgt502]

Hi,

Thanks for a quick response!
I am in fintech and am exploring the library.

Thank you for the example. The problem with the repo.Use is that it operates on an aggregate (not events) and doesnt allow me to access the transaction. My use case is, I need to be able to save data in another mongo collection whilst saving the aggregate - all in one transaction. Ideally, I would have access to the events so I know what changed in the aggregate.

Let me share an example of what I implemented (it is similar to BeforeInsert and AfterInsert).

.../mongo/store.go

type EventHandler func(context.Context, ...event.Event) error

type EventStore struct {
        // ...
        txEventHandlers     []EventHandler
        // ...
}

func WithTxEventHandler(th EventHandler) EventStoreOption {
	return func(s *EventStore) {
		s.transactions = true
		s.txEventHandlers = append(s.txEventHandlers, th)
	}
}

func (s *EventStore) Insert(ctx context.Context, events ...event.Event) error {
        // ...
        beginTx(...)
        insert(ctx, events)
        // ...
        for _, fn := range s.txEventHandlers {
		if handlerErr := fn(ctx, events...); handlerErr != nil {
			fmt.Println("aborting transaction...")
			if abortError := ctx.AbortTransaction(ctx); abortError != nil {
				return fmt.Errorf("abort transaction: %w", abortError)
			}
			return fmt.Errorf("tx event handler: %w", handlerErr)
		}
	}
       commitTx(...)
}

package example

func example(){
        handler := mongo.EventHandler(func(ctx context.Context, events ...event.Event) error {
		
		// Do something as part of the transaction...

		return nil
	})

        eventStore := mongotest.NewEventStore(		
		// ...
		mongo.WithTxEventHandler(handler),
	)
}

Let me know what you think.
Thanks!

[bounoable]

Okay I get it now. What you're aiming for is to execute actions within a single aggregate's transaction.

After giving this some thought, I think I found a potential solution:

1. Introduce Manual Transactions in the Event Store:
   This will allow actions to be performed within the context of a single transaction.

   package example
   
   func example(store *mongo.EventStore) {
     tx, err := store.Transaction(context.TODO())
     err = tx.Execute(func(store *mongo.EventStore) error { ... })
     err = tx.Execute(func(store *mongo.EventStore) error { ... })
     err = tx.Commit(context.TODO())
     // Optional rollback: err = tx.Rollback(context.TODO())
   }

2. Introduce New Interfaces for Transactional Event Stores:
   These interfaces will support stores that are capable of handling transactions.

   // eventstore/transactions.go # new file
   package eventstore
   
   type TransactionalStore[Store event.Store] interface {
     Transaction(context.Context) (Transaction[Store], error)
   }
   
   type Transaction[Store event.Store] interface {
     Execute(func(Store) error) error
     Commit(context.Context) error
     Rollback(context.Context) error
   }

3. Add a PreCommitter Interface:
   Aggregates that require actions before committing can implement this interface.

   package repository
   
   type PreCommitter interface {
     BeforeCommit(CommitContext) error
   }
   
   type CommitContext interface {
     context.Context
     EventStore()     event.Store
     InsertedEvents()  []event.Event
   }

   In the repository, we'll integrate this interface:

   // aggregate/repository/repository.go
   package repository
   
   type Repository[Store event.Store] struct {
     store Store
     // ...
   }
   
   func (repo *Repository[Store]) Save(ctx context.Context, a aggregate.Aggregate) error {
     // ...
     
     txStore, supportsTransactions := repo.store.(eventstore.TransactionalStore[Store])
     pc, isPreCommitter := a.(PreCommitter)
     
     if supportsTransactions {
       tx, err := repo.store.Transaction(ctx)
       
       events := a.AggregateChanges()
       err = repo.store.Insert(ctx, events...)
       
       if isPreCommitter {
           err = tx.Execute(func(store Store) error {
             return pc.BeforeCommit(newCommitContext(ctx, store, events))
           })
   
           if err != nil {
             return tx.Rollback(ctx)
           }
       }
       
       return tx.Commit(ctx)
     }
   }

   And here's an example of how aggregates can utilize the BeforeCommit hook:

   package example
   
   import (
     "github.com/modernice/goes/aggregate"
     "github.com/modernice/goes/aggregate/repository"
     "github.com/modernice/goes/backend/mongo"
   )
   
   type TodoList struct {
     *aggregate.Base
   }
   
   // BeforeCommit is executed during the transaction
   func (list *TodoList) BeforeCommit(ctx repository.CommitContext) error {
     store := ctx.EventStore()
   
     // these events have not been committed yet
     events := ctx.InsertedEvents()
     
     // Work with the event store inside the transaction
     // store.Query(ctx, query.New(...))
     // store.Insert(ctx, ...)
     // store.Delete(ctx, ...)
     
     // Alternatively, directly access MongoDB
     eventDB := ctx.EventStore().(*mongo.EventStore).Database()
     eventCollection := ctx.EventStore().(*mongo.EventStore).Collection()
   
     // we can return an error to abort/rollback the transaction
     // return errors.New("aborted")
   
     return nil
   }

Do you think this design would cover your use cases or is it too restricting?

[pgt502]

What you're aiming for is to execute actions within a single aggregate's transaction

Yes, that's right.

I wouldnt want to have this logic in the aggregate. In our application, aggregates belong to the domain layer which should not be aware of any infrastructure technologies like mongo event store.

Thanks

[pgt502]

hello, any update on this?
Is there a chance you would accept the given proposal? Or provide feedback on how it could be amended to get your approval?

[bounoable]

sorry, I'm very busy at the moment so I hadn't had the time to check on this again.

I think for your use case where you just want to generally hook into all transactions, a simple options for the event store like the one in your example should work out. I would change the signature of the transaction handler a bit and add a flag to specify when to call these hooks:

package mongo

var (
  // Call hook before inserting events into the event store.
  PreInsert = TransactionHook("pre:insert")
  
  // Call hook after inserting events into the event store.
  PostInsert = TransactionHook("post:insert")
)

type TransactionHook string

type Transaction interface {
  Session() mongo.SessionContext

  // EventStore() returns the event store itself to allow inserting (or deleting)
  // more events within the same transaction.
  EventStore() *EventStore
  
  // Returns all events that were inserted during this transaction.
  // They won't be inserted if the transaction is aborted.
  // When calling EventStore().Insert(), the inserted events will be added to this list.
  InsertedEvents() []event.Event
}

// mongo.SessionContext uses the same pattern, so I would like to stick to that.
type TransactionContext interface {
  context.Context
  Transaction
}

func WithTransactionHook(hook TransactionHook, fn func(TransactionContext) error) EventStoreOption {...}

package example

import (
  "github.com/modernice/goes/backend/mongo"
)

func example() {
  store := mongo.NewEventStore(
    codec.New(),
    mongo.WithTransactionHook(mongo.PostInsert, func(ctx mongo.TransactionContext) {
      // insert events
      err := ctx.EventStore().Insert(ctx, ...)

      // or use the mongo collection directly for more advanced use cases.
      // insertions won't be tracked by Transaction.InsertedEvents()
      col := ctx.EventStore().Collection()
    }),
  )
}

The event store would need to persist the mongo.SessionContext during the Insert() call such that additional calls to Insert() will use the same transaction.

I don't see a reason to not add this feature to the event store. Unfortunately I'm pretty loaded with work at the moment so I don't know whether I'm able to implement this soon. If you'd like to create a PR for this, I would gladly accept that. Otherwise I'll implement this when I got some free time on hand 👍🏼

[pgt502]

no worries, thank you for your reply.
Ok, that makes sense to me. I just wanted to understand the use-case for pre-insert hook. For me, it doesnt matter which action happens first as long as both are part of the same transaction. Is there a specific scenario you're trying to address?

Also, it should be possible to register multiple hooks, right?

I was thinking, I could implement the simplified version of this (without EventStore() and InsertedEvents) which would satisfy my requirements. Then, in the future, one of us can implement the rest? Would you be happy with that and accept such a PR?

Thanks

[bounoable]

Is there a specific scenario you're trying to address?

I don't have a specific scenario in mind but the PreInsert hook could be useful in scenarios where the order of the insertions matter. The event store could for example connect to an event bus that publishes the events in the same order as they were inserted, so I think this could provide flexibility for more advanced use cases.

Also, we could provide additional hooks in the future to hook into other methods of the event store by introducing for example delete:pre and delete:post hooks.

Also, it should be possible to register multiple hooks, right?

Yes, this should be possible. Each registered hook should be called sequentially and each one should be able to make the transaction fail.

If you implement a simplified version of this, I would be happy to merge the PR 👍🏼
You can just use the following for the Transaction interface and we will extend the API afterwards:

type Transaction interface {
  Session() mongo.SessionContext
}

And thank you for your contribution 🙂

[pgt502]

thanks for the quick response.
Ok, that makes sense - I will add the hook type.
I just realised that the signature of the hook doesnt have the events which I will need so I will have to implement the InsertedEvents() method too and might as well return the event store.

One question that popped into my mind now is: if we have different hooks, happening one by one and they all insert events, each of them should have access to InsertedEvents at the time a given hook is executed, right? For example:
hook2 should have access to events which were inserted by hook1 given order: hook1 -> hook2 and if they are post-insert they should also have access to the original events that were inserted outside of the hook. Hope that makes sense.

thanks

[bounoable]

yes that's exactly right 👍🏼 All event that were inserted during the transaction should be returned by the InsertedEvents() method, so the following holds:

Given: [0: "insert:pre", 1: "insert:pre", 2: "insert:post", 3: "insert:post"] hooks, then

• for hook 0: InsertedEvents() returns empty slice
• for hook 1: InsertedEvents() returns events that were inserted by hook 0
• for hook 2: InsertedEvents() returns events that were inserted by hook 0 and 1, and the "main" events
• for hook 3: InsertedEvents() returns events that were inserted by hook 0 and 1, the "main" events, and the events inserted by hook 2

[pgt502]

hey, I've created a draft PR for you to take a look: #146
I know I need to update the workflow to run on self-hosted but let me know if there is anything else you want me to change.

[pgt502]

implemented in #148