postgreSQLProjection handle always has events array with only last event

[rianzietsman]

I am using postgreSQLProjection to save data to my Prisma database.

To my understanding, handle: async events => {) should have all events with the same stream_id, so that one would arrive at the latest state by reducing those with the evolve function. Instead, it only ever contains the last added event.

I can confirm that my evolve function operates as expected in that it builds the correct state when executing a command.

Is there perhaps some metadata that I am not aware of that needs to be set?

Using:
"@event-driven-io/emmett": "^0.38.5",
"@event-driven-io/emmett-postgresql": "^0.38.5",

[oskardudycz]

@rianzietsman, could you show me how you're using postgreSQLProjection? The handler will get only newly appended events that are the outcome of the command handling logic. Then, if you want to update your read model, you should load its current state (so the record you want to update) and apply the event on top of it.

See more:

• how projections are handled -

  emmett/src/packages/emmett-postgresql/src/eventStore/projections/postgreSQLProjection.ts

  Line 74 in c1f0d45

  for (const projection of projections) {

• how pongo projection is doing it -

  emmett/src/packages/emmett-postgresql/src/eventStore/projections/pongo/pongoProjections.ts

  Line 154 in c1f0d45

  await collection.handle(getDocumentId(event), async (document) => {

If you're open to contribution, I could consider adding the Prisma package to Emmett to make easier handling such migrations.

[rianzietsman]

@oskardudycz I realise now that this is the intended behaviour. This gave me 2 options: either fetch the Prisma record that was created with this projection and update it with the latest event as supplied in handle: async events , or find the other events in that stream, replay them as well as the latest one.

I opted for the second option as it seemed more in line with this architecture. So in the end, my handle function looks like this:

handle: async events => {
     const [firstEvent] = events;
      const id = firstEvent.metadata.streamName;
    
      const allEvents = [
        ...(await EventSourcingService.getEventStore().readStream(id)).events,
        firstEvent,
      ];
    
      const state = (allEvents as (typeof firstEvent)[]).reduce((currentState, event) => {
        return evolve(currentState, event);
      }, initialState());

      // use state to update the Prisma table
  },

Works nicely.

[oskardudycz]

What you did can be a good enough solution, but my recommendation would be not to read all events each time you're updating the projection, as that's potentially doubling the reads of events. Normally, for projections, you're just doing a left fold: so taking the current state and updating it, then storing the result.

Any reason why you opted out of doing so?

Technically, I could expand Emmett to pass all events to inline projection, but I can't guarantee that all events will be available, as projection can also be used in async projections, then all events won't be available, as I'll be continuously polling next from the database, not reading all.

[rianzietsman]

taking the current state is what I was not able to figure out how to do. Is the current state available inside the handle function? What is the best way to obtain that?

[oskardudycz]

@rianzietsman you can have a look on how Pongo handles it: https://github.com/event-driven-io/Pongo/blob/bf3f51bc1f2fb127e83d20e6771102387537ba05/src/packages/pongo/src/core/collection/pongoCollection.ts#L361

Essentially, it loads the record from the database by the id taken from the event, then it does an upsert, or deletes it if the evolve function returned null.

[rianzietsman]

Ah, I see. What I perhaps did not point out earlier is that my Prisma record/table does not necessarily look the same as the output of the evolve function. Therefor I cannot guarantee that the result of a fetch of that record from Prisma is identical to the current state before applying the new event. As I understand it, Pongo saves the data in a NoSQL kind of way, so you could reliably hydrate the state with it and then just apply the new event.

For the moment, I am sticking to my approach I mentioned before - it's good enough for the moment.