Add lastReadAt property to Message and update logic for batch message headers

- Introduce the `lastReadAt` property in the `Message` class to track the last read timestamp.
- Update `send_batch` SQL method to handle per-message headers correctly.
- Adjust `MessageRowMapper` to include support for the new `lastReadAt` column.

Author: smyrgeorge

sqlx4k-postgres-pgmq/src/commonMain/kotlin/io/github/smyrgeorge/sqlx4k/postgres/pgmq/Message.kt

@@ -8,6 +8,7 @@ import kotlin.time.Instant
  * @property msgId Unique identifier for the message.
  * @property readCt Indicates how many times the message has been read.
  * @property enqueuedAt The timestamp when the message was added to the queue.
+ * @property lastReadAt The timestamp when the message was last read, or null if never read.
  * @property vt The visibility timeout of the message, indicating when it becomes visible for processing.
  * @property message The content of the message, expected to be in valid JSON format.
  * @property headers A map containing metadata key-value pairs associated with the message,
@@ -17,6 +18,7 @@ data class Message(
     val msgId: Long,
     val readCt: Int,
     val enqueuedAt: Instant,
+    val lastReadAt: Instant?,
     val vt: Instant,
     val message: String, // Should be a valid JSON.
     val headers: Map<String, String> // Will be converted to JSON.

sqlx4k-postgres-pgmq/src/commonMain/kotlin/io/github/smyrgeorge/sqlx4k/postgres/pgmq/PgmqClient.kt

@@ -352,11 +352,12 @@ class PgmqClient(
         delay: Duration = 0.seconds
     ): Result<List<Long>> {
         // language=PostgreSQL
-        val sql = "SELECT pgmq.send_batch(queue_name := ?, msgs := ARRAY[?]::jsonb[], headers := ?, delay := ?)"
+        // headers must be jsonb[] with one entry per message — replicate the shared headers map for each
+        val sql = "SELECT pgmq.send_batch(queue_name := ?, msgs := ARRAY[?]::jsonb[], headers := ARRAY[?]::jsonb[], delay := ?)"
         val statement = Statement.create(sql)
             .bind(0, queue)
             .bind(1, NoWrappingTuple(messages))
-            .bind(2, headers.toJsonString())
+            .bind(2, NoWrappingTuple(messages.map { headers.toJsonString() }))
             .bind(3, delay.inWholeSeconds.toInt())
         return db.fetchAll(statement, LongRowMapper) // returns the message-ids.
     }

sqlx4k-postgres-pgmq/src/commonMain/kotlin/io/github/smyrgeorge/sqlx4k/postgres/pgmq/impl/mappers/MessageRowMapper.kt

@@ -4,20 +4,23 @@ import io.github.smyrgeorge.sqlx4k.ResultSet
 import io.github.smyrgeorge.sqlx4k.RowMapper
 import io.github.smyrgeorge.sqlx4k.ValueEncoderRegistry
 import io.github.smyrgeorge.sqlx4k.impl.extensions.asInstant
+import io.github.smyrgeorge.sqlx4k.impl.extensions.asInstantOrNull
 import io.github.smyrgeorge.sqlx4k.impl.extensions.asInt
 import io.github.smyrgeorge.sqlx4k.impl.extensions.asLong
 import io.github.smyrgeorge.sqlx4k.postgres.pgmq.Message
 import io.github.smyrgeorge.sqlx4k.postgres.pgmq.impl.extensions.toStringMap
 
 internal object MessageRowMapper : RowMapper<Message> {
+    // Column order (pgmq >= 1.5): msg_id(0), read_ct(1), enqueued_at(2), last_read_at(3), vt(4), message(5), headers(6)
     override fun map(row: ResultSet.Row, converters: ValueEncoderRegistry): Message {
         return Message(
             msgId = row.get(0).asLong(),
             readCt = row.get(1).asInt(),
             enqueuedAt = row.get(2).asInstant(),
-            vt = row.get(3).asInstant(),
-            message = row.get(4).asString(),
-            headers = row.get(5).asStringOrNull()?.toStringMap() ?: emptyMap(),
+            lastReadAt = row.get(3).asInstantOrNull(),
+            vt = row.get(4).asInstant(),
+            message = row.get(5).asString(),
+            headers = row.get(6).asStringOrNull()?.toStringMap() ?: emptyMap(),
         )
     }
 }