H-419: run auto-assignment concurrency on PostgreSQL (#77)
Co-authored-by: Rogee <rogee@ipao.vip>
This commit is contained in:
@@ -3,6 +3,8 @@ package autoassignment
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@@ -14,11 +16,23 @@ import (
|
||||
"github.com/gochat/gochat/internal/model"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/driver/postgres"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/logger"
|
||||
)
|
||||
|
||||
func migrateFullAADB_Cov9(t *testing.T, db *gorm.DB) {
|
||||
t.Helper()
|
||||
require.NoError(t, db.AutoMigrate(
|
||||
&model.Account{}, &model.User{}, &model.AccountUser{},
|
||||
&model.Inbox{}, &model.InboxMember{}, &model.Contact{}, &model.ContactInbox{},
|
||||
&model.Conversation{}, &model.Message{},
|
||||
&AssignmentPolicy{}, &InboxAssignmentPolicy{},
|
||||
&model.AgentCapacityPolicy{}, &model.InboxCapacityLimit{},
|
||||
))
|
||||
}
|
||||
|
||||
func setupFullAADB_Cov9(t *testing.T) (*gorm.DB, *redis.Client) {
|
||||
t.Helper()
|
||||
db, err := gorm.Open(sqlite.Open(fmt.Sprintf("file:%s?mode=memory&cache=shared&_busy_timeout=5000", strings.ReplaceAll(t.Name(), "/", "_"))), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
|
||||
@@ -26,13 +40,49 @@ func setupFullAADB_Cov9(t *testing.T) (*gorm.DB, *redis.Client) {
|
||||
sqlDB, err := db.DB()
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = sqlDB.Close() })
|
||||
require.NoError(t, db.AutoMigrate(
|
||||
&model.Account{}, &model.User{}, &model.AccountUser{},
|
||||
&model.Inbox{}, &model.InboxMember{}, &model.Contact{}, &model.ContactInbox{},
|
||||
&model.Conversation{}, &model.Message{},
|
||||
&AssignmentPolicy{}, &InboxAssignmentPolicy{},
|
||||
&model.AgentCapacityPolicy{}, &model.InboxCapacityLimit{},
|
||||
))
|
||||
migrateFullAADB_Cov9(t, db)
|
||||
mr := miniredis.RunT(t)
|
||||
rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
|
||||
t.Cleanup(func() { _ = rdb.Close() })
|
||||
return db, rdb
|
||||
}
|
||||
|
||||
func setupFullAAPostgresDB_Cov9(t *testing.T) (*gorm.DB, *redis.Client) {
|
||||
t.Helper()
|
||||
dsn := os.Getenv("GOCHAT_TEST_DB_URL")
|
||||
if dsn == "" {
|
||||
dsn = "postgres://postgres:postgres@localhost:5432/gochat_test?sslmode=disable"
|
||||
}
|
||||
admin, err := gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
|
||||
require.NoError(t, err)
|
||||
adminDB, err := admin.DB()
|
||||
require.NoError(t, err)
|
||||
schema := fmt.Sprintf("autoassignment_%d", time.Now().UnixNano())
|
||||
require.NoError(t, admin.Exec("CREATE SCHEMA "+schema).Error)
|
||||
t.Cleanup(func() {
|
||||
_ = admin.Exec("DROP SCHEMA " + schema + " CASCADE").Error
|
||||
_ = adminDB.Close()
|
||||
})
|
||||
|
||||
if strings.HasPrefix(dsn, "postgres://") || strings.HasPrefix(dsn, "postgresql://") {
|
||||
dsnURL, parseErr := url.Parse(dsn)
|
||||
require.NoError(t, parseErr)
|
||||
query := dsnURL.Query()
|
||||
query.Set("search_path", schema)
|
||||
dsnURL.RawQuery = query.Encode()
|
||||
dsn = dsnURL.String()
|
||||
} else {
|
||||
dsn += " search_path=" + schema
|
||||
}
|
||||
db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{
|
||||
Logger: logger.Default.LogMode(logger.Silent),
|
||||
DisableForeignKeyConstraintWhenMigrating: true,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
sqlDB, err := db.DB()
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = sqlDB.Close() })
|
||||
migrateFullAADB_Cov9(t, db)
|
||||
mr := miniredis.RunT(t)
|
||||
rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
|
||||
t.Cleanup(func() { _ = rdb.Close() })
|
||||
@@ -168,6 +218,19 @@ func TestAssignmentServiceAutoAssignmentDoesNotOverwriteNewConversationState(t *
|
||||
|
||||
func TestAssignmentServiceOnlyOneConcurrentWorkerWins(t *testing.T) {
|
||||
db, rdb := setupFullAADB_Cov9(t)
|
||||
testAssignmentServiceOnlyOneConcurrentWorkerWins(t, db, rdb)
|
||||
}
|
||||
|
||||
func TestAssignmentServiceOnlyOneConcurrentWorkerWinsPostgres(t *testing.T) {
|
||||
if os.Getenv("GOCHAT_TEST_DB") != "postgres" {
|
||||
t.Skip("PostgreSQL concurrency test")
|
||||
}
|
||||
db, rdb := setupFullAAPostgresDB_Cov9(t)
|
||||
testAssignmentServiceOnlyOneConcurrentWorkerWins(t, db, rdb)
|
||||
}
|
||||
|
||||
func testAssignmentServiceOnlyOneConcurrentWorkerWins(t *testing.T, db *gorm.DB, rdb *redis.Client) {
|
||||
t.Helper()
|
||||
account, agent, inbox, conversation := seedAssignableConversation_Cov9(t, db)
|
||||
db = db.Session(&gorm.Session{SkipDefaultTransaction: true})
|
||||
sqlDB, err := db.DB()
|
||||
|
||||
Reference in New Issue
Block a user