pro/im_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add im relationship flows

Browse Source
This commit is contained in:
richjohnrambo
2026-08-12 11:31:16 +08:00
parent 5bd35f97a6
commit d02041659e
11 changed files with 1100 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/db/mysql/adapter.go
+74 -1
View File
@@ -41,7 +41,7 @@ type adapter struct {
}
const (
adpVersion = 122
adpVersion = 123
adapterName = "mysql"
defaultDSN = "root:@tcp(localhost:3306)/tinode?parseTime=true"
@@ -637,6 +637,10 @@ func (a *adapter) CreateDb(reset bool) error {
return err
}
if err = createIMRelationTables(tx); err != nil {
return err
}
// Deletion log
if _, err = tx.Exec(
`CREATE TABLE dellog(
@@ -1149,6 +1153,15 @@ func (a *adapter) UpgradeDb() error {
}
}
if a.version == 122 {
if err := createIMRelationTables(a.db.DB); err != nil {
return err
}
if err := bumpVersion(a, 123); err != nil {
return err
}
}
if a.version != adpVersion {
return errors.New("Failed to perform database upgrade to version " + strconv.Itoa(adpVersion) +
". DB is still at " + strconv.Itoa(a.version))
@@ -1208,6 +1221,66 @@ func upgradeIMMessageActions(a *adapter) error {
return nil
}
type sqlExecutor interface {
Exec(query string, args ...any) (sql.Result, error)
}
func createIMRelationTables(exec sqlExecutor) error {
statements := []string{
`CREATE TABLE im_friend_requests(
id BIGINT NOT NULL AUTO_INCREMENT,
tenant_id BIGINT NOT NULL,
request_id VARCHAR(64) NOT NULL,
applicant_id BIGINT NOT NULL,
target_id BIGINT NOT NULL,
message VARCHAR(200) NOT NULL DEFAULT '',
source VARCHAR(32) NOT NULL DEFAULT '',
source_topic VARCHAR(64) NOT NULL DEFAULT '',
status VARCHAR(16) NOT NULL,
idempotency_key VARCHAR(128) NOT NULL DEFAULT '',
expires_at DATETIME(3) NOT NULL,
handled_at DATETIME(3),
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id),
UNIQUE INDEX uk_im_friend_requests_request(tenant_id, request_id),
UNIQUE INDEX uk_im_friend_requests_idempotency(tenant_id, applicant_id, idempotency_key),
INDEX idx_im_friend_requests_pair(tenant_id, applicant_id, target_id, status, created_at),
INDEX idx_im_friend_requests_inbox(tenant_id, target_id, status, created_at),
CONSTRAINT fk_im_friend_requests_tenant FOREIGN KEY(tenant_id) REFERENCES im_tenant(id) ON DELETE CASCADE
)`,
`CREATE TABLE im_friendships(
tenant_id BIGINT NOT NULL,
user_id BIGINT NOT NULL,
peer_id BIGINT NOT NULL,
status VARCHAR(16) NOT NULL,
remark VARCHAR(64) NOT NULL DEFAULT '',
topic CHAR(25) NOT NULL DEFAULT '',
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(tenant_id, user_id, peer_id),
INDEX idx_im_friendships_peer(tenant_id, peer_id, status),
CONSTRAINT fk_im_friendships_tenant FOREIGN KEY(tenant_id) REFERENCES im_tenant(id) ON DELETE CASCADE
)`,
`CREATE TABLE im_user_blocks(
tenant_id BIGINT NOT NULL,
blocker_id BIGINT NOT NULL,
blocked_id BIGINT NOT NULL,
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(tenant_id, blocker_id, blocked_id),
INDEX idx_im_user_blocks_blocked(tenant_id, blocked_id),
CONSTRAINT fk_im_user_blocks_tenant FOREIGN KEY(tenant_id) REFERENCES im_tenant(id) ON DELETE CASCADE
)`,
}
for _, stmt := range statements {
if _, err := exec.Exec(stmt); err != nil {
return err
}
}
return nil
}
func createIMSecurityTables(tx *sql.Tx) error {
if _, err := tx.Exec(
`CREATE TABLE im_realname_verifications(
server/db/mysql/schema.sql
+52 -1
View File
@@ -84,7 +84,7 @@ CREATE TABLE kvmeta(
INDEX kvmeta_createdat_key(createdat, `key`)
);
INSERT INTO kvmeta(`key`, `value`) VALUES("version", "122");
INSERT INTO kvmeta(`key`, `value`) VALUES("version", "123");
CREATE TABLE users(
id BIGINT NOT NULL,
@@ -195,6 +195,57 @@ CREATE TABLE im_system_notifications(
CONSTRAINT fk_im_system_notifications_tenant FOREIGN KEY(tenant_id) REFERENCES im_tenant(id) ON DELETE CASCADE
);
CREATE TABLE im_friend_requests(
id BIGINT NOT NULL AUTO_INCREMENT,
tenant_id BIGINT NOT NULL,
request_id VARCHAR(64) NOT NULL,
applicant_id BIGINT NOT NULL,
target_id BIGINT NOT NULL,
message VARCHAR(200) NOT NULL DEFAULT '',
source VARCHAR(32) NOT NULL DEFAULT '',
source_topic VARCHAR(64) NOT NULL DEFAULT '',
status VARCHAR(16) NOT NULL,
idempotency_key VARCHAR(128) NOT NULL DEFAULT '',
expires_at DATETIME(3) NOT NULL,
handled_at DATETIME(3),
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(id),
UNIQUE INDEX uk_im_friend_requests_request(tenant_id, request_id),
UNIQUE INDEX uk_im_friend_requests_idempotency(tenant_id, applicant_id, idempotency_key),
INDEX idx_im_friend_requests_pair(tenant_id, applicant_id, target_id, status, created_at),
INDEX idx_im_friend_requests_inbox(tenant_id, target_id, status, created_at),
CONSTRAINT fk_im_friend_requests_tenant FOREIGN KEY(tenant_id) REFERENCES im_tenant(id) ON DELETE CASCADE
);
CREATE TABLE im_friendships(
tenant_id BIGINT NOT NULL,
user_id BIGINT NOT NULL,
peer_id BIGINT NOT NULL,
status VARCHAR(16) NOT NULL,
remark VARCHAR(64) NOT NULL DEFAULT '',
topic CHAR(25) NOT NULL DEFAULT '',
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(tenant_id, user_id, peer_id),
INDEX idx_im_friendships_peer(tenant_id, peer_id, status),
CONSTRAINT fk_im_friendships_tenant FOREIGN KEY(tenant_id) REFERENCES im_tenant(id) ON DELETE CASCADE
);
CREATE TABLE im_user_blocks(
tenant_id BIGINT NOT NULL,
blocker_id BIGINT NOT NULL,
blocked_id BIGINT NOT NULL,
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY(tenant_id, blocker_id, blocked_id),
INDEX idx_im_user_blocks_blocked(tenant_id, blocked_id),
CONSTRAINT fk_im_user_blocks_tenant FOREIGN KEY(tenant_id) REFERENCES im_tenant(id) ON DELETE CASCADE
);
# Indexed user tags.
CREATE TABLE usertags(
id INT NOT NULL AUTO_INCREMENT,
server/db/mysql/tenant_adapter.go
+246
View File
@@ -2514,6 +2514,252 @@ func (a *adapter) TenantSystemNotificationMarkRead(id t.TenantID, uid t.Uid, eve
return nil
}
func (a *adapter) TenantFriendRequestCreate(id t.TenantID, req *t.FriendRequest) (*t.FriendRequest, error) {
if err := validTenant(id); err != nil || req == nil || req.TenantID != id || req.ApplicantID.IsZero() || req.TargetID.IsZero() {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
_, err := a.db.ExecContext(ctx,
"INSERT INTO im_friend_requests(tenant_id,request_id,applicant_id,target_id,message,source,source_topic,status,idempotency_key,expires_at,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",
id, req.RequestID, decodeUID(req.ApplicantID), decodeUID(req.TargetID), req.Message, req.Source, req.SourceTopic, req.Status, req.IdempotencyKey, req.ExpiresAt, req.CreatedAt, req.UpdatedAt)
if err != nil {
if isDupe(err) && req.IdempotencyKey != "" {
return a.friendRequestByIdempotency(ctx, id, req.ApplicantID, req.IdempotencyKey)
}
return nil, err
}
return a.friendRequestByRequestID(ctx, id, req.RequestID)
}
func (a *adapter) TenantFriendRequestLatestBetween(id t.TenantID, uid1, uid2 t.Uid) (*t.FriendRequest, error) {
if err := validTenant(id); err != nil || uid1.IsZero() || uid2.IsZero() {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
return a.friendRequestByQuery(ctx,
"SELECT * FROM im_friend_requests WHERE tenant_id=? AND ((applicant_id=? AND target_id=?) OR (applicant_id=? AND target_id=?)) ORDER BY created_at DESC,id DESC LIMIT 1",
id, decodeUID(uid1), decodeUID(uid2), decodeUID(uid2), decodeUID(uid1))
}
func (a *adapter) TenantFriendRequestResolve(id t.TenantID, requestID string, actor t.Uid, status string, handledAt time.Time) (*t.FriendRequest, error) {
if err := validTenant(id); err != nil || requestID == "" || actor.IsZero() {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
req, err := a.friendRequestByRequestID(ctx, id, requestID)
if err != nil || req == nil {
return req, err
}
if req.TargetID != actor {
return nil, t.ErrPermissionDenied
}
if req.Status != t.FriendRequestPending {
return req, nil
}
if _, err = a.db.ExecContext(ctx,
"UPDATE im_friend_requests SET status=?,handled_at=?,updated_at=? WHERE tenant_id=? AND request_id=? AND status=?",
status, handledAt, handledAt, id, requestID, t.FriendRequestPending); err != nil {
return nil, err
}
return a.friendRequestByRequestID(ctx, id, requestID)
}
func (a *adapter) TenantFriendshipUpsertPair(id t.TenantID, uid1, uid2 t.Uid, topic string, now time.Time) error {
if err := validTenant(id); err != nil || uid1.IsZero() || uid2.IsZero() || uid1 == uid2 {
return t.ErrMalformed
}
if topic == "" {
topic = uid1.P2PName(uid2)
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
for _, pair := range [][2]t.Uid{{uid1, uid2}, {uid2, uid1}} {
if _, err := a.db.ExecContext(ctx,
"INSERT INTO im_friendships(tenant_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)",
id, decodeUID(pair[0]), decodeUID(pair[1]), t.RelationshipFriend, topic, now, now); err != nil {
return err
}
}
return nil
}
func (a *adapter) TenantFriendshipGet(id t.TenantID, userID, peerID t.Uid) (*t.Friendship, error) {
if err := validTenant(id); err != nil || userID.IsZero() || peerID.IsZero() {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
row := struct {
TenantID t.TenantID `db:"tenant_id"`
UserID int64 `db:"user_id"`
PeerID int64 `db:"peer_id"`
Status string
Remark string
Topic string
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}{}
err := a.db.GetContext(ctx, &row, "SELECT * FROM im_friendships WHERE tenant_id=? AND user_id=? AND peer_id=?",
id, decodeUID(userID), decodeUID(peerID))
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
return &t.Friendship{
TenantID: row.TenantID,
UserID: store.EncodeUid(row.UserID),
PeerID: store.EncodeUid(row.PeerID),
Status: row.Status,
Remark: row.Remark,
Topic: row.Topic,
CreatedAt: row.CreatedAt,
UpdatedAt: row.UpdatedAt,
}, nil
}
func (a *adapter) TenantFriendshipDeletePair(id t.TenantID, uid1, uid2 t.Uid, now time.Time) error {
if err := validTenant(id); err != nil || uid1.IsZero() || uid2.IsZero() {
return t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
_, err := a.db.ExecContext(ctx,
"UPDATE im_friendships SET status=?,updated_at=? WHERE tenant_id=? AND ((user_id=? AND peer_id=?) OR (user_id=? AND peer_id=?))",
t.RelationshipNone, now, id, decodeUID(uid1), decodeUID(uid2), decodeUID(uid2), decodeUID(uid1))
return err
}
func (a *adapter) TenantUserBlockGet(id t.TenantID, blockerID, blockedID t.Uid) (*t.UserBlock, error) {
if err := validTenant(id); err != nil || blockerID.IsZero() || blockedID.IsZero() {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
row := struct {
TenantID t.TenantID `db:"tenant_id"`
BlockerID int64 `db:"blocker_id"`
BlockedID int64 `db:"blocked_id"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}{}
err := a.db.GetContext(ctx, &row, "SELECT * FROM im_user_blocks WHERE tenant_id=? AND blocker_id=? AND blocked_id=?",
id, decodeUID(blockerID), decodeUID(blockedID))
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
return &t.UserBlock{
TenantID: row.TenantID,
BlockerID: store.EncodeUid(row.BlockerID),
BlockedID: store.EncodeUid(row.BlockedID),
CreatedAt: row.CreatedAt,
UpdatedAt: row.UpdatedAt,
}, nil
}
func (a *adapter) TenantUserBlockUpsert(id t.TenantID, blockerID, blockedID t.Uid, now time.Time) (*t.UserBlock, error) {
if err := validTenant(id); err != nil || blockerID.IsZero() || blockedID.IsZero() {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
_, err := a.db.ExecContext(ctx,
"INSERT INTO im_user_blocks(tenant_id,blocker_id,blocked_id,created_at,updated_at) VALUES(?,?,?,?,?) "+
"ON DUPLICATE KEY UPDATE updated_at=VALUES(updated_at)",
id, decodeUID(blockerID), decodeUID(blockedID), now, now)
if err != nil {
return nil, err
}
return a.TenantUserBlockGet(id, blockerID, blockedID)
}
func (a *adapter) TenantUserBlockDelete(id t.TenantID, blockerID, blockedID t.Uid) error {
if err := validTenant(id); err != nil || blockerID.IsZero() || blockedID.IsZero() {
return t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
_, err := a.db.ExecContext(ctx, "DELETE FROM im_user_blocks WHERE tenant_id=? AND blocker_id=? AND blocked_id=?",
id, decodeUID(blockerID), decodeUID(blockedID))
return err
}
func (a *adapter) friendRequestByRequestID(ctx context.Context, tenantID t.TenantID, requestID string) (*t.FriendRequest, error) {
return a.friendRequestByQuery(ctx, "SELECT * FROM im_friend_requests WHERE tenant_id=? AND request_id=?", tenantID, requestID)
}
func (a *adapter) friendRequestByIdempotency(ctx context.Context, tenantID t.TenantID, applicantID t.Uid, key string) (*t.FriendRequest, error) {
return a.friendRequestByQuery(ctx, "SELECT * FROM im_friend_requests WHERE tenant_id=? AND applicant_id=? AND idempotency_key=?", tenantID, decodeUID(applicantID), key)
}
func (a *adapter) friendRequestByQuery(ctx context.Context, query string, args ...any) (*t.FriendRequest, error) {
row := struct {
ID int64
TenantID t.TenantID `db:"tenant_id"`
RequestID string `db:"request_id"`
ApplicantID int64 `db:"applicant_id"`
TargetID int64 `db:"target_id"`
Message string
Source string
SourceTopic string `db:"source_topic"`
Status string
IdempotencyKey string `db:"idempotency_key"`
ExpiresAt time.Time `db:"expires_at"`
HandledAt *time.Time `db:"handled_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}{}
err := a.db.GetContext(ctx, &row, query, args...)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
return &t.FriendRequest{
ID: row.ID,
TenantID: row.TenantID,
RequestID: row.RequestID,
ApplicantID: store.EncodeUid(row.ApplicantID),
TargetID: store.EncodeUid(row.TargetID),
Message: row.Message,
Source: row.Source,
SourceTopic: row.SourceTopic,
Status: row.Status,
IdempotencyKey: row.IdempotencyKey,
ExpiresAt: row.ExpiresAt,
HandledAt: row.HandledAt,
CreatedAt: row.CreatedAt,
UpdatedAt: row.UpdatedAt,
}, nil
}
func nullableUID(uid t.Uid) any {
if uid.IsZero() {
return nil