An upsert combines insertion and updating: it inserts a record when no matching data exists, or updates the existing record otherwise. In a projection, this lets aggregated event data move from a journal into a separate database while reading and writing remain decoupled. section 10.2.1
Counting cargo per container
A repository can upsert a container’s cargo count. The first CargoAdded event creates (containerId, 1); later events for the same container update the row by adding one to its existing count. The statement runs inside a database transaction, which can commit or roll back. section 10.3.2
A projection can apply the operation through a small processing pipeline:
- Read events from the journal and wrap them in event envelopes. section 10.2.1
- Match CargoAdded events and delegate storage to a repository. section 10.3.2
- Insert a missing row or update the matching row with the new aggregate value. section 10.3.2
Listing 10.7 Repository Implementation
trait CargosPerContainerRepository { #1
def addCargo(containerId: String, session: ScalikeJdbcSession): Unit
}
class CargosPerContainerRepositoryImpl extends CargosPerContainerRepository {
override def addCargo(
containerId: String,
session: ScalikeJdbcSession): Unit = {
session.db.withinTx { implicit dbSession => #2
sql"""INSERT INTO cargos_per_container (containerId, carg #3
VALUES ($containerId, 1)
ON CONFLICT (containerId) DO
UPDATE SET cargos = cargos_per_container.cargos + 1
""".executeUpdate().apply()
}
}
Browse the repository implementation to see the INSERT ... ON CONFLICT statement that creates or increments a container record. section 10.3.2
