Add xid8 (#45)

Author: m110

.github/workflows/master.yml

@@ -8,3 +8,4 @@ jobs:
     uses: ThreeDotsLabs/watermill/.github/workflows/tests.yml@master
     with:
       stress-tests: true
+      runs-on: ubuntu-latest-16core

docker-compose.yml

@@ -5,7 +5,7 @@ services:
   mysql:
     image: mysql:8.0
     restart: unless-stopped
-    command: [ "--max_connections=5000" ]
+    command: [ "--max_connections=50000" ]
     ports:
       - 3306:3306
     environment:
@@ -15,7 +15,7 @@ services:
   postgres:
     image: postgres:15.3
     restart: unless-stopped
-    command: postgres -c 'max_connections=5000'
+    command: postgres -c 'max_connections=50000'
     ports:
       - 5432:5432
     environment:

pkg/sql/schema_adapter_postgresql.go

@@ -2,13 +2,15 @@ package sql
 
 import (
 	"database/sql"
+	"database/sql/driver"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"hash/fnv"
+	"strconv"
 	"strings"
 
 	"github.com/ThreeDotsLabs/watermill/message"
-	"github.com/jackc/pgx/v5/pgtype"
 )
 
 // DefaultPostgreSQLSchema is a default implementation of SchemaAdapter based on PostgreSQL.
@@ -178,7 +180,7 @@ func (s DefaultPostgreSQLSchema) SelectQuery(params SelectQueryParams) (Query, e
 			` + nextOffsetQuery.Query + `
 		)
 
-		SELECT "offset", transaction_id, uuid, payload, metadata FROM ` + s.MessagesTable(params.Topic) + `
+		SELECT "offset", transaction_id::text, uuid, payload, metadata FROM ` + s.MessagesTable(params.Topic) + `
 
 		WHERE 
 		(
@@ -203,7 +205,7 @@ func (s DefaultPostgreSQLSchema) SelectQuery(params SelectQueryParams) (Query, e
 
 func (s DefaultPostgreSQLSchema) UnmarshalMessage(params UnmarshalMessageParams) (Row, error) {
 	r := Row{}
-	var transactionID pgtype.Uint64
+	var transactionID XID8
 
 	err := params.Row.Scan(&r.Offset, &transactionID, &r.UUID, &r.Payload, &r.Metadata)
 	if err != nil {
@@ -259,3 +261,34 @@ func DefaultSchemaInitializationLock(appName string) int {
 	// PostgreSQL advisory locks use int32 internally
 	return int(h.Sum32() & 0x7fffffff)
 }
+
+type XID8 uint64
+
+func (x *XID8) Scan(src interface{}) error {
+	if src == nil {
+		return errors.New("cannot scan nil value into XID8")
+	}
+
+	switch v := src.(type) {
+	case string:
+		val, err := strconv.ParseUint(v, 10, 64)
+		if err != nil {
+			return err
+		}
+		*x = XID8(val)
+		return nil
+	case []byte:
+		val, err := strconv.ParseUint(string(v), 10, 64)
+		if err != nil {
+			return err
+		}
+		*x = XID8(val)
+		return nil
+	default:
+		return errors.New("unsupported Scan value type for XID8")
+	}
+}
+
+func (x *XID8) Value() (driver.Value, error) {
+	return uint64(*x), nil
+}