501 lines
16 KiB
Go
501 lines
16 KiB
Go
//go:build mysql
|
|
// +build mysql
|
|
|
|
package mysql
|
|
|
|
import (
|
|
"database/sql"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/jmoiron/sqlx"
|
|
"github.com/tinode/chat/server/store"
|
|
t "github.com/tinode/chat/server/store/types"
|
|
)
|
|
|
|
func (a *adapter) routedRegistrationInviteAdapter(id t.TenantID) (*adapter, error) {
|
|
if err := validTenant(id); err != nil {
|
|
return nil, err
|
|
}
|
|
if a.tenantScopedDB {
|
|
return a, nil
|
|
}
|
|
tenantAdp, err := a.routedTenantAdapter(id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return tenantAdp, nil
|
|
}
|
|
|
|
func (a *adapter) registrationInviteControlDB() *sqlx.DB {
|
|
if a.registryDB != nil {
|
|
return a.registryDB
|
|
}
|
|
return a.db
|
|
}
|
|
|
|
func (a *adapter) TenantRegistrationInviteCreate(id t.TenantID, code *t.RegistrationInviteCode) error {
|
|
tenantAdp, err := a.routedRegistrationInviteAdapter(id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tenantAdp != a {
|
|
return tenantAdp.TenantRegistrationInviteCreate(id, code)
|
|
}
|
|
if code == nil || len(code.Code) != 5 || code.ExpiresAt.IsZero() || code.CreatedBy <= 0 {
|
|
return t.ErrMalformed
|
|
}
|
|
if code.ID == 0 {
|
|
code.ID, err = nextSnowflakeID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
now := t.TimeNow()
|
|
code.TenantID = id
|
|
code.Status = t.RegistrationInviteActive
|
|
code.CreatedAt, code.UpdatedAt = now, now
|
|
ctx, cancel := a.getContext()
|
|
if cancel != nil {
|
|
defer cancel()
|
|
}
|
|
_, err = a.registrationInviteControlDB().ExecContext(ctx, `INSERT INTO im_registration_invite_codes
|
|
(id,tenant_id,batch_id,code,status,expires_at,max_uses,usage_count,last_used_at,created_by,remark,created_at,updated_at)
|
|
VALUES(?,?,?,?,?,?,1,0,NULL,?,?,?,?)`, code.ID, id, code.BatchID, code.Code, code.Status, code.ExpiresAt,
|
|
code.CreatedBy, code.Remark, code.CreatedAt, code.UpdatedAt)
|
|
if isDupe(err) {
|
|
return t.ErrDuplicate
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (a *adapter) TenantRegistrationInviteCreateBatch(id t.TenantID, batch *t.RegistrationInviteBatch, codes []t.RegistrationInviteCode) (err error) {
|
|
tenantAdp, err := a.routedRegistrationInviteAdapter(id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tenantAdp != a {
|
|
return tenantAdp.TenantRegistrationInviteCreateBatch(id, batch, codes)
|
|
}
|
|
if batch == nil || batch.Quantity != len(codes) || len(codes) == 0 || strings.TrimSpace(batch.IdempotencyKey) == "" ||
|
|
batch.ExpiresAt.IsZero() || batch.CreatedBy <= 0 {
|
|
return t.ErrMalformed
|
|
}
|
|
if batch.ID == 0 {
|
|
batch.ID, err = nextSnowflakeID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
now := t.TimeNow()
|
|
batch.TenantID = id
|
|
batch.CreatedAt, batch.UpdatedAt = now, now
|
|
ctx, cancel := a.getContextForTx()
|
|
if cancel != nil {
|
|
defer cancel()
|
|
}
|
|
tx, err := a.registrationInviteControlDB().BeginTxx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() {
|
|
if err != nil {
|
|
_ = tx.Rollback()
|
|
}
|
|
}()
|
|
if _, err = tx.ExecContext(ctx, `INSERT INTO im_registration_invite_batches
|
|
(id,tenant_id,quantity,idempotency_key,expires_at,created_by,remark,created_at,updated_at)
|
|
VALUES(?,?,?,?,?,?,?,?,?)`, batch.ID, id, batch.Quantity, batch.IdempotencyKey, batch.ExpiresAt,
|
|
batch.CreatedBy, batch.Remark, batch.CreatedAt, batch.UpdatedAt); err != nil {
|
|
if isDupe(err) {
|
|
return t.ErrDuplicate
|
|
}
|
|
return err
|
|
}
|
|
for i := range codes {
|
|
if len(codes[i].Code) != 5 {
|
|
return t.ErrMalformed
|
|
}
|
|
codes[i].ID, err = nextSnowflakeID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
codes[i].BatchID = &batch.ID
|
|
codes[i].TenantID = id
|
|
codes[i].Status = t.RegistrationInviteActive
|
|
codes[i].ExpiresAt = batch.ExpiresAt
|
|
codes[i].CreatedBy = batch.CreatedBy
|
|
codes[i].Remark = batch.Remark
|
|
codes[i].CreatedAt, codes[i].UpdatedAt = now, now
|
|
if _, err = tx.ExecContext(ctx, `INSERT INTO im_registration_invite_codes
|
|
(id,tenant_id,batch_id,code,status,expires_at,max_uses,usage_count,last_used_at,created_by,remark,created_at,updated_at)
|
|
VALUES(?,?,?,?,?,?,0,0,NULL,?,?,?,?)`, codes[i].ID, id, batch.ID, codes[i].Code, codes[i].Status,
|
|
codes[i].ExpiresAt, codes[i].CreatedBy, codes[i].Remark, now, now); err != nil {
|
|
if isDupe(err) {
|
|
return t.ErrDuplicate
|
|
}
|
|
return err
|
|
}
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (a *adapter) TenantRegistrationInviteGetByCode(id t.TenantID, value string) (*t.RegistrationInviteCode, error) {
|
|
tenantAdp, err := a.routedRegistrationInviteAdapter(id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if tenantAdp != a {
|
|
return tenantAdp.TenantRegistrationInviteGetByCode(id, value)
|
|
}
|
|
if len(value) != 5 {
|
|
return nil, t.ErrInviteInvalid
|
|
}
|
|
ctx, cancel := a.getContext()
|
|
if cancel != nil {
|
|
defer cancel()
|
|
}
|
|
var code t.RegistrationInviteCode
|
|
err = a.registrationInviteControlDB().GetContext(ctx, &code, `SELECT c.id,c.tenant_id,c.batch_id,c.code,c.status,c.expires_at,c.max_uses,c.bound_user_id,
|
|
c.created_by,c.remark,c.created_at,c.updated_at,c.usage_count,c.last_used_at
|
|
FROM im_registration_invite_codes AS c
|
|
WHERE c.tenant_id=? AND c.code=? LIMIT 1`, id, value)
|
|
if err == sql.ErrNoRows {
|
|
return nil, t.ErrInviteInvalid
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if code.Status == t.RegistrationInviteActive && !t.TimeNow().Before(code.ExpiresAt) {
|
|
code.Status = t.RegistrationInviteExpired
|
|
}
|
|
return &code, nil
|
|
}
|
|
|
|
func (a *adapter) TenantRegistrationInviteList(id t.TenantID, opts t.RegistrationInviteListOptions) (*t.RegistrationInvitePage, error) {
|
|
tenantAdp, err := a.routedRegistrationInviteAdapter(id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if tenantAdp != a {
|
|
return tenantAdp.TenantRegistrationInviteList(id, opts)
|
|
}
|
|
if opts.Limit <= 0 || opts.Limit > 100 {
|
|
opts.Limit = 20
|
|
}
|
|
if opts.Offset < 0 {
|
|
opts.Offset = 0
|
|
}
|
|
ctx, cancel := a.getContext()
|
|
if cancel != nil {
|
|
defer cancel()
|
|
}
|
|
now := t.TimeNow()
|
|
controlDB := a.registrationInviteControlDB()
|
|
_, _ = controlDB.ExecContext(ctx, `UPDATE im_registration_invite_codes SET status=?,updated_at=?
|
|
WHERE tenant_id=? AND status=? AND expires_at<=?`, t.RegistrationInviteExpired, now, id, t.RegistrationInviteActive, now)
|
|
where := []string{"c.tenant_id=?"}
|
|
args := []any{id}
|
|
if opts.Code != "" {
|
|
where, args = append(where, "c.code=?"), append(args, opts.Code)
|
|
}
|
|
if opts.Status != "" {
|
|
where, args = append(where, "c.status=?"), append(args, opts.Status)
|
|
}
|
|
if opts.BatchID > 0 {
|
|
where, args = append(where, "c.batch_id=?"), append(args, opts.BatchID)
|
|
}
|
|
if opts.CreatedBy > 0 {
|
|
where, args = append(where, "c.created_by=?"), append(args, opts.CreatedBy)
|
|
}
|
|
if opts.CreatedAtStart != nil {
|
|
where, args = append(where, "c.created_at>=?"), append(args, *opts.CreatedAtStart)
|
|
}
|
|
if opts.CreatedAtEnd != nil {
|
|
where, args = append(where, "c.created_at<=?"), append(args, *opts.CreatedAtEnd)
|
|
}
|
|
if opts.ExpiresAtStart != nil {
|
|
where, args = append(where, "c.expires_at>=?"), append(args, *opts.ExpiresAtStart)
|
|
}
|
|
if opts.ExpiresAtEnd != nil {
|
|
where, args = append(where, "c.expires_at<=?"), append(args, *opts.ExpiresAtEnd)
|
|
}
|
|
filter := strings.Join(where, " AND ")
|
|
var total int
|
|
if err = controlDB.GetContext(ctx, &total, "SELECT COUNT(*) FROM im_registration_invite_codes AS c WHERE "+filter, args...); err != nil {
|
|
return nil, err
|
|
}
|
|
queryArgs := append(append([]any{}, args...), opts.Limit, opts.Offset)
|
|
var items []t.RegistrationInviteCode
|
|
err = controlDB.SelectContext(ctx, &items, `SELECT c.id,c.tenant_id,c.batch_id,c.code,c.status,c.expires_at,c.max_uses,c.bound_user_id,
|
|
c.created_by,c.remark,c.created_at,c.updated_at,c.usage_count,c.last_used_at
|
|
FROM im_registration_invite_codes AS c
|
|
WHERE `+filter+` ORDER BY c.created_at DESC,c.id DESC LIMIT ? OFFSET ?`, queryArgs...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &t.RegistrationInvitePage{Items: items, Total: total}, nil
|
|
}
|
|
|
|
func (a *adapter) registrationInviteBatchGet(id t.TenantID, where string, arg any) (*t.RegistrationInviteBatch, error) {
|
|
tenantAdp, err := a.routedRegistrationInviteAdapter(id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if tenantAdp != a {
|
|
if where == "b.id=?" {
|
|
return tenantAdp.TenantRegistrationInviteBatchGet(id, arg.(int64))
|
|
}
|
|
return tenantAdp.TenantRegistrationInviteBatchGetByIdempotency(id, arg.(string))
|
|
}
|
|
ctx, cancel := a.getContext()
|
|
if cancel != nil {
|
|
defer cancel()
|
|
}
|
|
var batch t.RegistrationInviteBatch
|
|
err = a.registrationInviteControlDB().GetContext(ctx, &batch, `SELECT b.id,b.tenant_id,b.quantity,b.idempotency_key,b.expires_at,b.created_by,b.remark,b.created_at,b.updated_at,
|
|
COALESCE(SUM(CASE WHEN c.status='active' AND c.expires_at>? THEN 1 ELSE 0 END),0) AS active_count,
|
|
COALESCE(SUM(c.usage_count),0) AS usage_count,
|
|
COALESCE(SUM(CASE WHEN (c.status='active' AND c.expires_at<=?) OR c.status='expired' THEN 1 ELSE 0 END),0) AS expired_count,
|
|
COALESCE(SUM(CASE WHEN c.status='revoked' THEN 1 ELSE 0 END),0) AS revoked_count
|
|
FROM im_registration_invite_batches AS b
|
|
LEFT JOIN im_registration_invite_codes AS c ON c.tenant_id=b.tenant_id AND c.batch_id=b.id
|
|
WHERE b.tenant_id=? AND `+where+` GROUP BY b.id`, t.TimeNow(), t.TimeNow(), id, arg)
|
|
if err == sql.ErrNoRows {
|
|
return nil, t.ErrNotFound
|
|
}
|
|
return &batch, err
|
|
}
|
|
|
|
func (a *adapter) TenantRegistrationInviteBatchGet(id t.TenantID, batchID int64) (*t.RegistrationInviteBatch, error) {
|
|
if batchID <= 0 {
|
|
return nil, t.ErrMalformed
|
|
}
|
|
return a.registrationInviteBatchGet(id, "b.id=?", batchID)
|
|
}
|
|
|
|
func (a *adapter) TenantRegistrationInviteBatchGetByIdempotency(id t.TenantID, key string) (*t.RegistrationInviteBatch, error) {
|
|
if strings.TrimSpace(key) == "" {
|
|
return nil, t.ErrMalformed
|
|
}
|
|
return a.registrationInviteBatchGet(id, "b.idempotency_key=?", key)
|
|
}
|
|
|
|
func (a *adapter) TenantRegistrationInviteRevoke(id t.TenantID, codeID int64, now time.Time) error {
|
|
tenantAdp, err := a.routedRegistrationInviteAdapter(id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tenantAdp != a {
|
|
return tenantAdp.TenantRegistrationInviteRevoke(id, codeID, now)
|
|
}
|
|
if codeID <= 0 || now.IsZero() {
|
|
return t.ErrMalformed
|
|
}
|
|
ctx, cancel := a.getContext()
|
|
if cancel != nil {
|
|
defer cancel()
|
|
}
|
|
res, err := a.registrationInviteControlDB().ExecContext(ctx, `UPDATE im_registration_invite_codes SET status=?,updated_at=?
|
|
WHERE tenant_id=? AND id=? AND status=? AND expires_at>?`, t.RegistrationInviteRevoked, now,
|
|
id, codeID, t.RegistrationInviteActive, now)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if count, _ := res.RowsAffected(); count == 0 {
|
|
return t.ErrNotFound
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (a *adapter) TenantRegistrationInviteBatchRevoke(id t.TenantID, batchID int64, now time.Time) (int, error) {
|
|
tenantAdp, err := a.routedRegistrationInviteAdapter(id)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if tenantAdp != a {
|
|
return tenantAdp.TenantRegistrationInviteBatchRevoke(id, batchID, now)
|
|
}
|
|
if batchID <= 0 || now.IsZero() {
|
|
return 0, t.ErrMalformed
|
|
}
|
|
ctx, cancel := a.getContext()
|
|
if cancel != nil {
|
|
defer cancel()
|
|
}
|
|
res, err := a.registrationInviteControlDB().ExecContext(ctx, `UPDATE im_registration_invite_codes SET status=?,updated_at=?
|
|
WHERE tenant_id=? AND batch_id=? AND status=? AND expires_at>?`, t.RegistrationInviteRevoked,
|
|
now, id, batchID, t.RegistrationInviteActive, now)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
count, _ := res.RowsAffected()
|
|
return int(count), nil
|
|
}
|
|
|
|
func (a *adapter) TenantRegistrationInviteRecordUse(id t.TenantID, value string, uid t.Uid, requestID string, now time.Time) (err error) {
|
|
tenantAdp, err := a.routedRegistrationInviteAdapter(id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tenantAdp != a {
|
|
return tenantAdp.TenantRegistrationInviteRecordUse(id, value, uid, requestID, now)
|
|
}
|
|
if len(value) != 5 || uid.IsZero() || strings.TrimSpace(requestID) == "" || now.IsZero() {
|
|
return t.ErrMalformed
|
|
}
|
|
var row struct {
|
|
ID int64 `db:"id"`
|
|
Status t.RegistrationInviteStatus `db:"status"`
|
|
ExpiresAt time.Time `db:"expires_at"`
|
|
MaxUses int `db:"max_uses"`
|
|
BoundUserID *int64 `db:"bound_user_id"`
|
|
}
|
|
ctx, cancel := a.getContext()
|
|
if cancel != nil {
|
|
defer cancel()
|
|
}
|
|
err = a.registrationInviteControlDB().GetContext(ctx, &row, `SELECT id,status,expires_at,max_uses,bound_user_id
|
|
FROM im_registration_invite_codes WHERE tenant_id=? AND code=? LIMIT 1`, id, value)
|
|
if err == sql.ErrNoRows {
|
|
return t.ErrInviteInvalid
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if row.Status == t.RegistrationInviteRevoked {
|
|
return t.ErrInviteRevoked
|
|
}
|
|
if row.Status == t.RegistrationInviteExpired || !now.Before(row.ExpiresAt) {
|
|
return t.ErrInviteExpired
|
|
}
|
|
if row.Status != t.RegistrationInviteActive {
|
|
return t.ErrInviteInvalid
|
|
}
|
|
|
|
txCtx, txCancel := a.getContextForTx()
|
|
if txCancel != nil {
|
|
defer txCancel()
|
|
}
|
|
tx, err := a.db.BeginTxx(txCtx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() {
|
|
if err != nil {
|
|
_ = tx.Rollback()
|
|
}
|
|
}()
|
|
if row.MaxUses > 0 {
|
|
var usedCount int
|
|
if err = tx.GetContext(txCtx, &usedCount,
|
|
"SELECT COUNT(*) FROM im_registration_invite_uses WHERE invite_code_id=?", row.ID); err != nil {
|
|
return err
|
|
}
|
|
if usedCount >= row.MaxUses {
|
|
return t.ErrInviteInvalid
|
|
}
|
|
}
|
|
useID, err := nextSnowflakeID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err = tx.ExecContext(txCtx, `INSERT INTO im_registration_invite_uses
|
|
(id,invite_code_id,user_id,request_id,used_at) VALUES(?,?,?,?,?)`, useID, row.ID,
|
|
store.DecodeUid(uid), requestID, now); err != nil {
|
|
if isDupe(err) {
|
|
var existingCodeID int64
|
|
if lookupErr := tx.GetContext(txCtx, &existingCodeID,
|
|
"SELECT invite_code_id FROM im_registration_invite_uses WHERE user_id=?", store.DecodeUid(uid)); lookupErr == nil && existingCodeID == row.ID {
|
|
err = nil
|
|
} else {
|
|
return t.ErrDuplicate
|
|
}
|
|
} else {
|
|
return err
|
|
}
|
|
}
|
|
if row.BoundUserID != nil && *row.BoundUserID > 0 {
|
|
boundUID := t.ParseUserId(strconv.FormatInt(*row.BoundUserID, 10))
|
|
if !boundUID.IsZero() && boundUID != uid {
|
|
if err = upsertFriendshipPairTx(tx, uid, boundUID, uid.P2PName(boundUID), now); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
eventID := t.StableEventID("registration-invite-use", id, row.ID, store.DecodeUid(uid))
|
|
if err = insertBusinessEventTx(tx, eventID, "user", uid.UserId(), uid, "registration.invite_used", map[string]any{
|
|
"tenant_id": id, "invite_code_id": row.ID, "user_id": store.DecodeUid(uid), "bound_user_id": row.BoundUserID,
|
|
"request_id": requestID, "used_at": now,
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func upsertFriendshipPairTx(tx *sqlx.Tx, uid1, uid2 t.Uid, topic string, now time.Time) error {
|
|
if uid1.IsZero() || uid2.IsZero() || uid1 == uid2 {
|
|
return t.ErrMalformed
|
|
}
|
|
if topic == "" {
|
|
topic = uid1.P2PName(uid2)
|
|
}
|
|
for _, pair := range [][2]t.Uid{{uid1, uid2}, {uid2, uid1}} {
|
|
friendshipID, err := nextSnowflakeID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err = tx.Exec(
|
|
"INSERT INTO im_friendships(id,user_id,peer_id,status,topic,created_at,updated_at) VALUES(?,?,?,?,?,?,?) "+
|
|
"ON DUPLICATE KEY UPDATE status=VALUES(status),topic=VALUES(topic),updated_at=VALUES(updated_at)",
|
|
friendshipID, decodeUID(pair[0]), decodeUID(pair[1]), t.RelationshipFriend, topic, now, now); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// TenantRegistrationInviteSyncUsage projects the tenant usage aggregate to the
|
|
// control database. The overwrite is idempotent, so outbox retries are safe.
|
|
func (a *adapter) TenantRegistrationInviteSyncUsage(id t.TenantID, codeID int64) error {
|
|
tenantAdp, err := a.routedRegistrationInviteAdapter(id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tenantAdp != a {
|
|
return tenantAdp.TenantRegistrationInviteSyncUsage(id, codeID)
|
|
}
|
|
if codeID <= 0 {
|
|
return t.ErrMalformed
|
|
}
|
|
ctx, cancel := a.getContext()
|
|
if cancel != nil {
|
|
defer cancel()
|
|
}
|
|
var aggregate struct {
|
|
UsageCount int `db:"usage_count"`
|
|
LastUsedAt *time.Time `db:"last_used_at"`
|
|
}
|
|
if err = a.db.GetContext(ctx, &aggregate, `SELECT COUNT(*) AS usage_count,MAX(used_at) AS last_used_at
|
|
FROM im_registration_invite_uses WHERE invite_code_id=?`, codeID); err != nil {
|
|
return err
|
|
}
|
|
res, err := a.registrationInviteControlDB().ExecContext(ctx, `UPDATE im_registration_invite_codes
|
|
SET usage_count=?,last_used_at=? WHERE tenant_id=? AND id=?`, aggregate.UsageCount, aggregate.LastUsedAt, id, codeID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if count, _ := res.RowsAffected(); count == 0 {
|
|
var exists int
|
|
if lookupErr := a.registrationInviteControlDB().GetContext(ctx, &exists,
|
|
"SELECT 1 FROM im_registration_invite_codes WHERE tenant_id=? AND id=?", id, codeID); lookupErr != nil {
|
|
if lookupErr == sql.ErrNoRows {
|
|
return t.ErrNotFound
|
|
}
|
|
return lookupErr
|
|
}
|
|
}
|
|
return nil
|
|
}
|