postgres/me.ahoo.wow.postgres.subscription/PostgresCheckpointStore

PostgresCheckpointStore

class PostgresCheckpointStore(dataSource: DataSource, dispatcher: CoroutineDispatcher = VirtualThreads.dispatcher, ddl: Ddl = Ddl.AUTO) : CheckpointStore

The CheckpointStore on PostgreSQL: one row per (group, subscription, partition) in wow_checkpoint.

A save is one autocommitted upsert whose update applies only when the new position is higher, so concurrent saves leave the highest and a lower one changes nothing. JDBC calls run on dispatcher, virtual threads by default.

Constructors

constructor(dataSource: DataSource, dispatcher: CoroutineDispatcher = VirtualThreads.dispatcher, ddl: Ddl = Ddl.AUTO)

Functions

createSchema

fun createSchema()

Claims the table's format (PostgresFormat) and creates it when missing.

list

open suspend override fun list(group: String): List<Checkpoint>

load

open suspend override fun load(group: String, subscription: String, partition: Int): Long?

save

open suspend override fun save(group: String, subscription: String, partition: Int, position: Long)

saveAll

open suspend override fun saveAll(group: String, checkpoints: List<Checkpoint>)

The upserts save makes one by one, as one multi-row upsert in one transaction: one statement, whose round trip also carries the transaction's bound, and one commit, one WAL flush (in autocommit each upsert of a batch was a transaction of its own). A key saved twice keeps its furthest position, as one statement may touch a row once. The rows go in key order, so two batches touching the same rows lock them in the same order and never wait on each other in a cycle; past MAX_ROWS they are split into several statements of the one transaction. A failed batch commits none of them, and the caller saves them again, as it saves any batch that failed: each upsert only moves a position forward.