152 lines
3.6 KiB
Go
152 lines
3.6 KiB
Go
// Package mysql 实现 room-service 的 MySQL 持久化边界。
|
||
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
mysqlerr "github.com/go-sql-driver/mysql"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
)
|
||
|
||
const (
|
||
roomOutboxTable = "room_outbox"
|
||
roomRobotOutboxTable = "room_robot_outbox"
|
||
)
|
||
|
||
// Repository 把 room-service 的权威状态持久化到 MySQL。
|
||
type Repository struct {
|
||
// db 是标准库连接池，所有 repository 方法都通过它访问 MySQL。
|
||
db *sql.DB
|
||
}
|
||
|
||
// Open 建立 MySQL 连接池，并验证数据库可用。
|
||
func Open(ctx context.Context, dsn string, maxOpenConns int, maxIdleConns int) (*Repository, error) {
|
||
// sql.Open 只校验 driver 和 DSN 形态，真实连通性由后续 PingContext 验证。
|
||
db, err := sql.Open("mysql", dsn)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if maxOpenConns > 0 {
|
||
// 只在配置显式给出正数时覆盖标准库默认值。
|
||
db.SetMaxOpenConns(maxOpenConns)
|
||
}
|
||
|
||
if maxIdleConns > 0 {
|
||
// 空闲连接数需要小于等于最大连接数，由配置层负责合理设置。
|
||
db.SetMaxIdleConns(maxIdleConns)
|
||
}
|
||
|
||
// 长连接定期轮换，避免云数据库或网络层回收导致长期坏连接驻留。
|
||
db.SetConnMaxLifetime(30 * time.Minute)
|
||
|
||
if err := db.PingContext(ctx); err != nil {
|
||
// 启动探测失败时必须关闭连接池，避免返回半可用 repository。
|
||
_ = db.Close()
|
||
return nil, err
|
||
}
|
||
|
||
return &Repository{db: db}, nil
|
||
}
|
||
|
||
// Close 关闭底层 MySQL 连接池。
|
||
func (r *Repository) Close() error {
|
||
// App.Close 调用该方法释放所有空闲连接，并阻止后续新查询。
|
||
return r.db.Close()
|
||
}
|
||
|
||
// Ping 验证 MySQL 连接池当前仍能在调用方 deadline 内拿到可用连接。
|
||
func (r *Repository) Ping(ctx context.Context) error {
|
||
// ready 探针通过该方法确认 command log/snapshot/outbox 的持久化来源可用。
|
||
return r.db.PingContext(ctx)
|
||
}
|
||
|
||
const (
|
||
roomPresenceStatusActiveSQL = "active"
|
||
roomPresenceStatusLeftSQL = "left"
|
||
roomFollowStatusActiveSQL = "active"
|
||
roomFollowStatusCancelledSQL = "cancelled"
|
||
)
|
||
|
||
func escapeRoomListLike(value string) string {
|
||
var builder strings.Builder
|
||
for _, r := range value {
|
||
if r == '\\' || r == '%' || r == '_' {
|
||
builder.WriteRune('\\')
|
||
}
|
||
builder.WriteRune(r)
|
||
}
|
||
return builder.String()
|
||
}
|
||
|
||
func nullableString(value string) sql.Null[string] {
|
||
// 空错误写 NULL，便于 SQL 查询区分没有错误和空字符串。
|
||
return sql.Null[string]{
|
||
V: value,
|
||
Valid: value != "",
|
||
}
|
||
}
|
||
|
||
func normalizedRecordAppCode(ctx context.Context, value string) string {
|
||
value = strings.TrimSpace(value)
|
||
if value == "" {
|
||
return appcode.FromContext(ctx)
|
||
}
|
||
|
||
return appcode.Normalize(value)
|
||
}
|
||
|
||
func boolToInt(value bool) int {
|
||
if value {
|
||
return 1
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func firstPositiveInt64(values ...int64) int64 {
|
||
for _, value := range values {
|
||
if value > 0 {
|
||
return value
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func uniquePositiveInt64(values []int64) []int64 {
|
||
seen := make(map[int64]bool, len(values))
|
||
out := make([]int64, 0, len(values))
|
||
for _, value := range values {
|
||
if value <= 0 || seen[value] {
|
||
continue
|
||
}
|
||
seen[value] = true
|
||
out = append(out, value)
|
||
}
|
||
sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
|
||
return out
|
||
}
|
||
|
||
func mapRoomMetaDuplicateError(err error) error {
|
||
if err == nil {
|
||
return nil
|
||
}
|
||
|
||
var mysqlErr *mysqlerr.MySQLError
|
||
if !errors.As(err, &mysqlErr) || mysqlErr.Number != 1062 {
|
||
return err
|
||
}
|
||
|
||
message := mysqlErr.Message
|
||
if strings.Contains(message, "uk_rooms_owner_user") || strings.Contains(message, "owner_user_id") {
|
||
return xerr.New(xerr.Conflict, "owner already has room")
|
||
}
|
||
|
||
return xerr.New(xerr.Conflict, "room already exists")
|
||
}
|