Saving Aggregate with entity list

[maranqz]

Hello Guys,

Initially, I want to say "Thank you" for your articles and the book.

Problem

Updating message is the most blur process for me.

Context

I have Chat (Aggregate) and Message (Entity, which knows nothing about Chat).

When I send a message, I want to create Chat if it doesn't exist. Then, a Message is attached to the Chat; change observing and message saving are tricky for me.

I track domain events and can handle them to update Chat and Message. *sqlx.Tx is passed in an observer to save transactionality; however, it looks complicated and hard to optimize the command with a few similar events. For example, to mark a bunch of messages as read.

Another way is to compare an old value with a new one and generate SQL for updating.

Questions

• Could you give me another way or advice, which way is better?
• May I ask an approachable lib with an observable pattern?
• Could you advise a lib for comparing? I saw go-cmp, but maybe there is a more convenient lib.
• What do you think about additional repository argument ([]Option)? Should it be moved in context?

Code

// /domain/chat.go

// Key is an Identity Field
type Key struct {
    ID int64 // Artificial ID
    ProductID ItemID // Each product can have several chats 
    SellerID UserID
    BuyerID UserID
}

type Chat struct {
    Key
    IsDeleted bool
    CreatedAt time.Time
    DeletedAt time.Time
    messages []*Message
    events
}

func (c *Chat) SendMessage(m *Message) error {
    if err := m.Validation(); err != nil {
        return err
    }
    
    c.raise(SendMessageEvent{
        event:   event{chat: c},
        message: m,
    })
    
    c.messages = append(c.messages, m)

    return nil
}

// /domain/event.go

type Event interface {
    isEvent()
    Entity() *Chat
}

type events interface {
    Events() []Event
    raise(e Event)
}


type SendMessageEvent struct {
    Event
    message *Message
}

// /domain/message.go

type Message struct {
    Text      string
    AuthorID  UserID  
    ReplyID   int64      // if it is a reply on the previous message
    IsRead    bool   
    IsDeleted bool
    CreatedAt time.Time
    DeletedAt *time.Time
}

// /adapters/mysql_repository.go

type observable interface {
    Dispatch(event Event) error
}

type mysqlRepository struct{
	obs observer
}

func (c *mysqlRepository) getOrCreate(ctx context.Context, key chat.Key, opts ...chat.Option) (*chat.Chat, error) {
    ...
    var cfg config
    for _, opt := range opts {
        opt(&cfg)
    }

    if row.version != nil && cfg.version != 0 && cfg.version <= *row.version {
        return fmt.Errorf("chat is already updated: %w", chat.ErrDBUnRetryable)
    }
    ...
}

func (c *mysqlRepository) Update(
    ctx context.Context,
    key chat.Key,
    updateFn func(entity *chat.Chat) error,
    opts ...chat.Option,
) {
    // beginTransaction
    // defer closeTransaction

	entity, err := c.getOrCreate(ctx, key, opts) 
	if err != nil {
       return err
    }

    if err = sendFn(entity); err != nil {
        return err
    }

    if err = c.upsert(ctx, entity); err != nil {
        return err
    }

    // TODO MESSAGE SAVING

    return c.processEvents(ctx, tx, entity, opts)
}

func (c *mysqlRepository) processEvents(ctx context.Context, tx *sqlx.Tx, entity *chat.Chat, opts []chat.Option) (err error) {
    for _, e := range entity.Events() {
        if err = c.obs.Dispatch(e); err != nil {
            return err
        }
    }
    
    return nil
}

[m110]

Hey @maranqz.

I think I don't understand the context fully, but here's some thoughts:

• Consider if all messages need to be part of the same aggregate. Keeping them like this means they need to be transactionally consistent — is this really a requirement? If not, it'll be much easier to update each message separately.
• Rather than a single Update method, consider smaller, specialized methods that work with the aggregate. For example, MarkMessageAsRead.

May I ask an approachable lib with an observable pattern?

Are you using a message queue? You might find this example useful.

Another way is to compare an old value with a new one and generate SQL for updating.

I wouldn't recommend that. It means you have to load the entire aggregate every time you update it. It'll be slow and error-prone.

What do you think about additional repository argument ([]Option)? Should it be moved in context?

What are the options? It seems to me like something you'd pass to the repository's constructor, not to its methods.

Hopefully that's helpful. :) Let me know.

[maranqz]

@m110, thank for the detailed answer.

they need to be transactionally consistent — is this really a requirement?

It's just an example and I want them to be in a transaction.

consider smaller, specialized methods that work with the aggregate. For example, MarkMessageAsRead

I though there is a more elegant solution without creating many small methods on each operation.

observable pattern? message queue

Is it ok to put Pub/Sub in domain or better add in repository after saving?

[]Option

I use []Option to add additional data. For example, I want to check version of a record and understand can or can't update the record. Also I use []Options to send additional non-domain data in event handlers. For analytics, I want know a User-Agent. I get this information in a controller/http handler and put in []Options. Event handler gets the User-Agent and send data about event and User-Agent in another system.

[m110]

I though there is a more elegant solution without creating many small methods on each operation.

I would say it's a trade-off. :) You can have some sort of balance, don't need to go all the way into one huge Save or multiple tiny methods. Depends on what you need, how fast it needs to be, etc. "Elegant" is probably too subjective to be useful, try to understand what works best for your scenario.

Is it ok to put Pub/Sub in domain or better add in repository after saving?

The Pub/Sub implementation shouldn't interfere with your domain but you could have events as a domain concept.

If you plan to publish events together with saving the aggregate, keep in mind transaction boundaries. Here are some examples:

• https://threedots.tech/post/when-sql-database-makes-great-pub-sub/
• https://github.com/ThreeDotsLabs/watermill/tree/master/_examples/real-world-examples/transactional-events
• https://github.com/ThreeDotsLabs/watermill/tree/master/_examples/real-world-examples/transactional-events-forwarder

I use []Option to add additional data. For example, I want to check version of a record and understand can or can't update the record. Also I use []Options to send additional non-domain data in event handlers. For analytics, I want know a User-Agent. I get this information in a controller/http handler and put in []Options. Event handler gets the User-Agent and send data about event and User-Agent in another system.

This seems to me like a mix of a few different concepts. "Can update" seems like a key logic of the repository. I would choose to have it more explicit. Options seem like an optional thing you might forget to pass.

On the other hand, the UserAgent could be a good fit for passing in Context. However, it's also very implicit. It's hard to tell the repository uses it by looking just at the method signature. You could also consider having some sort of Metadata structure tied to the aggreagte.