From d02041659e712ccfb36a2d4fd65d16e2e9bde30b Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Wed, 12 Aug 2026 11:31:16 +0800 Subject: [PATCH] feat: add im relationship flows --- server/db/adapter.go | 10 ++ server/db/mysql/adapter.go | 75 +++++++- server/db/mysql/schema.sql | 53 +++++- server/db/mysql/tenant_adapter.go | 246 ++++++++++++++++++++++++++ server/im_http.go | 279 ++++++++++++++++++++++++++++++ server/im_http_test.go | 123 +++++++++++++ server/main.go | 5 + server/store/store.go | 152 ++++++++++++++++ server/store/types/relations.go | 65 +++++++ server/topic.go | 29 ++++ server/topic_test.go | 75 ++++++-- 11 files changed, 1100 insertions(+), 12 deletions(-) create mode 100644 server/store/types/relations.go diff --git a/server/db/adapter.go b/server/db/adapter.go index c911c01..95b98f9 100644 --- a/server/db/adapter.go +++ b/server/db/adapter.go @@ -317,4 +317,14 @@ type TenantBusinessAdapter interface { TenantSystemNotificationCreate(t.TenantID, *t.SystemNotification) error TenantSystemNotificationList(t.TenantID, t.Uid, string, int) ([]t.SystemNotification, string, error) TenantSystemNotificationMarkRead(t.TenantID, t.Uid, string, time.Time) error + + TenantFriendRequestCreate(t.TenantID, *t.FriendRequest) (*t.FriendRequest, error) + TenantFriendRequestLatestBetween(t.TenantID, t.Uid, t.Uid) (*t.FriendRequest, error) + TenantFriendRequestResolve(t.TenantID, string, t.Uid, string, time.Time) (*t.FriendRequest, error) + TenantFriendshipUpsertPair(t.TenantID, t.Uid, t.Uid, string, time.Time) error + TenantFriendshipGet(t.TenantID, t.Uid, t.Uid) (*t.Friendship, error) + TenantFriendshipDeletePair(t.TenantID, t.Uid, t.Uid, time.Time) error + TenantUserBlockGet(t.TenantID, t.Uid, t.Uid) (*t.UserBlock, error) + TenantUserBlockUpsert(t.TenantID, t.Uid, t.Uid, time.Time) (*t.UserBlock, error) + TenantUserBlockDelete(t.TenantID, t.Uid, t.Uid) error } diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index 8d6b8b5..727b3dd 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -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( diff --git a/server/db/mysql/schema.sql b/server/db/mysql/schema.sql index aa34d78..a8377e8 100644 --- a/server/db/mysql/schema.sql +++ b/server/db/mysql/schema.sql @@ -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, diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index d64cbc4..74901ee 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -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 diff --git a/server/im_http.go b/server/im_http.go index 5c06bfb..4c490ea 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -127,6 +127,13 @@ type imMessageRevisionRequest struct { Reason string `json:"reason,omitempty"` } +type imFriendRequestCreateRequest struct { + TargetUserID string `json:"target_user_id"` + Message string `json:"message,omitempty"` + Source string `json:"source,omitempty"` + SourceTopic string `json:"source_topic,omitempty"` +} + func serveIMTenantResolve(wrt http.ResponseWriter, req *http.Request) { if req.Method != http.MethodPost { writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) @@ -983,6 +990,163 @@ func serveIMMessageRevision(wrt http.ResponseWriter, req *http.Request) { writeIMSuccess(wrt, req, messageMutationPayload(updated, "edited")) } +func serveIMConversationContext(wrt http.ResponseWriter, req *http.Request) { + if req.Method != http.MethodGet { + writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) + return + } + principal, ok := authenticateIMHTTPBearer(wrt, req) + if !ok { + return + } + topic := conversationContextTopicFromPath(req.URL.Path) + peer := peerFromP2PTopic(topic, principal.Uid) + if topic == "" || peer.IsZero() { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "topic"}) + return + } + ctx, err := store.Relations.GetRelationshipContext(principal.TenantID, principal.Uid, peer, topic) + if err != nil { + writeIMRelationError(wrt, req, err) + return + } + writeIMSuccess(wrt, req, relationshipContextPayload(ctx)) +} + +func serveIMFriendRequests(wrt http.ResponseWriter, req *http.Request) { + if req.Method != http.MethodPost { + writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) + return + } + principal, ok := authenticateIMHTTPBearer(wrt, req) + if !ok { + return + } + var body imFriendRequestCreateRequest + if !decodeIMJSONBody(wrt, req, &body) { + return + } + target := types.ParseUserId(body.TargetUserID) + message := strings.TrimSpace(body.Message) + if target.IsZero() || target == principal.Uid || len([]rune(message)) > 50 { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil) + return + } + idempotencyKey := strings.TrimSpace(req.Header.Get("Idempotency-Key")) + if idempotencyKey == "" { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "Idempotency-Key is required", map[string]any{"field": "Idempotency-Key"}) + return + } + now := time.Now().UTC().Round(time.Millisecond) + requestID := "frq_" + store.Store.GetUidString() + created, err := store.Relations.CreateFriendRequest(principal.TenantID, &types.FriendRequest{ + TenantID: principal.TenantID, + RequestID: requestID, + ApplicantID: principal.Uid, + TargetID: target, + Message: message, + Source: strings.TrimSpace(body.Source), + SourceTopic: strings.TrimSpace(body.SourceTopic), + Status: types.FriendRequestPending, + IdempotencyKey: idempotencyKey, + ExpiresAt: now.Add(7 * 24 * time.Hour), + CreatedAt: now, + UpdatedAt: now, + }) + if err != nil { + writeIMRelationError(wrt, req, err) + return + } + writeIMSuccess(wrt, req, friendRequestPayload(created)) +} + +func serveIMFriendRequestAction(wrt http.ResponseWriter, req *http.Request) { + if req.Method != http.MethodPost { + writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) + return + } + principal, ok := authenticateIMHTTPBearer(wrt, req) + if !ok { + return + } + requestID, action := friendRequestActionFromPath(req.URL.Path) + status := "" + switch action { + case "accept": + status = types.FriendRequestAccepted + case "reject": + status = types.FriendRequestRejected + default: + writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "friend request action not found", nil) + return + } + updated, err := store.Relations.ResolveFriendRequest(principal.TenantID, requestID, principal.Uid, status, time.Now().UTC().Round(time.Millisecond)) + if err != nil { + writeIMRelationError(wrt, req, err) + return + } + writeIMSuccess(wrt, req, friendRequestPayload(updated)) +} + +func serveIMBlockUser(wrt http.ResponseWriter, req *http.Request) { + principal, ok := authenticateIMHTTPBearer(wrt, req) + if !ok { + return + } + peer := userIDFromTail(req.URL.Path, "/v1/im/blocks/") + if peer.IsZero() || peer == principal.Uid { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil) + return + } + switch req.Method { + case http.MethodPut: + if _, err := store.Relations.BlockUser(principal.TenantID, principal.Uid, peer, time.Now().UTC().Round(time.Millisecond)); err != nil { + writeIMRelationError(wrt, req, err) + return + } + case http.MethodDelete: + if err := store.Relations.UnblockUser(principal.TenantID, principal.Uid, peer); err != nil { + writeIMRelationError(wrt, req, err) + return + } + default: + writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) + return + } + ctx, err := store.Relations.GetRelationshipContext(principal.TenantID, principal.Uid, peer, principal.Uid.P2PName(peer)) + if err != nil { + writeIMRelationError(wrt, req, err) + return + } + writeIMSuccess(wrt, req, relationshipContextPayload(ctx)) +} + +func serveIMFriendResource(wrt http.ResponseWriter, req *http.Request) { + if req.Method != http.MethodDelete { + writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) + return + } + principal, ok := authenticateIMHTTPBearer(wrt, req) + if !ok { + return + } + peer := userIDFromTail(req.URL.Path, "/v1/im/friends/") + if peer.IsZero() || peer == principal.Uid { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil) + return + } + if err := store.Relations.DeleteFriend(principal.TenantID, principal.Uid, peer, time.Now().UTC().Round(time.Millisecond)); err != nil { + writeIMRelationError(wrt, req, err) + return + } + ctx, err := store.Relations.GetRelationshipContext(principal.TenantID, principal.Uid, peer, principal.Uid.P2PName(peer)) + if err != nil { + writeIMRelationError(wrt, req, err) + return + } + writeIMSuccess(wrt, req, relationshipContextPayload(ctx)) +} + func serveIMIndependentPassword(wrt http.ResponseWriter, req *http.Request) { if req.Method != http.MethodPost { writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) @@ -1413,6 +1577,68 @@ func messageMutationPayload(msg *types.Message, status string) map[string]any { return data } +func relationshipContextPayload(ctx *types.RelationshipContext) map[string]any { + data := map[string]any{ + "topic": ctx.Topic, + "peer_user_id": ctx.PeerID.UserId(), + "relationship": ctx.Relationship, + "can_send": ctx.CanSend, + "capabilities": map[string]any{ + "can_send_text": ctx.CanSend, + "can_send_media": ctx.CanSend, + "can_send_card": ctx.CanSend, + "can_call": ctx.CanSend, + "can_add_friend": ctx.Relationship == types.RelationshipNone, + "can_handle_request": ctx.Relationship == types.RelationshipPendingInbound, + }, + } + if ctx.RequestID != "" { + data["request_id"] = ctx.RequestID + } + if ctx.Peer != nil { + data["peer"] = map[string]any{ + "user_id": ctx.PeerID.UserId(), + "public": ctx.Peer.Public, + "trusted": ctx.Peer.Trusted, + } + } + return data +} + +func friendRequestPayload(req *types.FriendRequest) map[string]any { + data := map[string]any{ + "request_id": req.RequestID, + "applicant_id": req.ApplicantID.UserId(), + "target_user_id": req.TargetID.UserId(), + "message": req.Message, + "source": req.Source, + "status": req.Status, + "created_at": req.CreatedAt.UTC().Format(time.RFC3339Nano), + "expires_at": req.ExpiresAt.UTC().Format(time.RFC3339Nano), + "topic": req.ApplicantID.P2PName(req.TargetID), + } + if req.SourceTopic != "" { + data["source_topic"] = req.SourceTopic + } + if req.HandledAt != nil { + data["handled_at"] = req.HandledAt.UTC().Format(time.RFC3339Nano) + } + return data +} + +func writeIMRelationError(wrt http.ResponseWriter, req *http.Request, err error) { + switch { + case errors.Is(err, types.ErrNotFound): + writeIMError(wrt, req, http.StatusNotFound, 20004, imReasonMalformedRequest, "resource not found", nil) + case errors.Is(err, types.ErrPermissionDenied): + writeIMError(wrt, req, http.StatusForbidden, 20005, imReasonPermissionDenied, "permission denied", nil) + case errors.Is(err, types.ErrMalformed): + writeIMError(wrt, req, http.StatusBadRequest, 20000, imReasonMalformedRequest, "malformed request", nil) + default: + writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) + } +} + func validIMRealName(name, idType, idNumber string) bool { if len([]rune(name)) < 2 || len([]rune(name)) > 32 || idType != "cn_id" { return false @@ -1690,3 +1916,56 @@ func messageActionPath(path, action string) (string, int, bool) { } return topic, seq, true } + +func conversationContextTopicFromPath(path string) string { + const prefix = "/v1/im/conversations/" + const suffix = "/context" + if !strings.Contains(path, prefix) || !strings.HasSuffix(path, suffix) { + return "" + } + idx := strings.Index(path, prefix) + return strings.TrimSuffix(strings.TrimPrefix(path[idx:], prefix), suffix) +} + +func friendRequestActionFromPath(path string) (string, string) { + const prefix = "/v1/im/friend-requests/" + idx := strings.Index(path, prefix) + if idx < 0 { + return "", "" + } + rest := strings.TrimPrefix(path[idx:], prefix) + requestID, action, ok := strings.Cut(rest, "/") + if !ok { + return "", "" + } + return strings.TrimSpace(requestID), strings.TrimSpace(action) +} + +func userIDFromTail(path, prefix string) types.Uid { + idx := strings.Index(path, prefix) + if idx < 0 { + return types.ZeroUid + } + return types.ParseUserId(strings.TrimSpace(strings.TrimPrefix(path[idx:], prefix))) +} + +func peerFromP2PTopic(topic string, uid types.Uid) types.Uid { + if types.GetTopicCat(topic) != types.TopicCatP2P || uid.IsZero() { + return types.ZeroUid + } + prefix := "p2p" + body := strings.TrimPrefix(topic, prefix) + uidLen := len(uid.String()) + if len(body) < 2*uidLen { + return types.ZeroUid + } + left := types.ParseUid(body[:uidLen]) + right := types.ParseUid(body[uidLen:]) + if left == uid { + return right + } + if right == uid { + return left + } + return types.ZeroUid +} diff --git a/server/im_http_test.go b/server/im_http_test.go index bce490d..df74cfe 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -33,6 +33,70 @@ type testVerificationStore struct { tokenHash string } +type testRelationsStore struct { + context *types.RelationshipContext + created *types.FriendRequest + blocked *types.UserBlock + deleted bool + canSend *bool + reason string +} + +func (s *testRelationsStore) CreateFriendRequest(_ types.TenantID, req *types.FriendRequest) (*types.FriendRequest, error) { + s.created = req + return req, nil +} + +func (s *testRelationsStore) ResolveFriendRequest(_ types.TenantID, requestID string, actor types.Uid, status string, handledAt time.Time) (*types.FriendRequest, error) { + return &types.FriendRequest{ + TenantID: 7, + RequestID: requestID, + ApplicantID: 1, + TargetID: actor, + Status: status, + HandledAt: &handledAt, + CreatedAt: handledAt, + UpdatedAt: handledAt, + ExpiresAt: handledAt.Add(time.Hour), + }, nil +} + +func (s *testRelationsStore) GetRelationshipContext(_ types.TenantID, userID, peerID types.Uid, topic string) (*types.RelationshipContext, error) { + if s.context != nil { + return s.context, nil + } + return &types.RelationshipContext{ + TenantID: 7, + UserID: userID, + PeerID: peerID, + Topic: topic, + Relationship: types.RelationshipFriend, + CanSend: true, + }, nil +} + +func (s *testRelationsStore) CanSendP2P(_ types.TenantID, _, _ types.Uid) (bool, string, error) { + if s.canSend != nil { + return *s.canSend, s.reason, nil + } + return true, types.RelationshipFriend, nil +} + +func (s *testRelationsStore) BlockUser(tenantID types.TenantID, blockerID, blockedID types.Uid, now time.Time) (*types.UserBlock, error) { + s.blocked = &types.UserBlock{TenantID: tenantID, BlockerID: blockerID, BlockedID: blockedID, CreatedAt: now, UpdatedAt: now} + return s.blocked, nil +} + +func (s *testRelationsStore) UnblockUser(_ types.TenantID, _, _ types.Uid) error { + s.blocked = nil + return nil +} + +func (s *testRelationsStore) DeleteFriend(_ types.TenantID, _, _ types.Uid, _ time.Time) error { + s.deleted = true + return nil +} + func (s *testVerificationStore) Create(_ types.TenantID, vc *types.VerificationCode) error { s.created = vc return s.err @@ -861,6 +925,65 @@ func TestServeIMMessageRevision(t *testing.T) { } } +func TestServeIMConversationContext(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + expectIMBearer(t, ctrl, "token-1", 7, 1) + oldRelations := store.Relations + peer := types.Uid(2) + topic := types.Uid(1).P2PName(peer) + store.Relations = &testRelationsStore{context: &types.RelationshipContext{ + TenantID: 7, + UserID: 1, + PeerID: peer, + Topic: topic, + Relationship: types.RelationshipFriend, + CanSend: true, + }} + defer func() { store.Relations = oldRelations }() + + req := httptest.NewRequest(http.MethodGet, "/v1/im/conversations/"+topic+"/context", nil) + req.Header.Set("Authorization", "Bearer token-1") + rec := httptest.NewRecorder() + + serveIMConversationContext(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String()) + } + data := decodeIMHTTPResponse(t, rec).Data.(map[string]any) + if data["relationship"] != types.RelationshipFriend || data["can_send"] != true { + t.Fatalf("unexpected context response: %#v", data) + } +} + +func TestServeIMFriendRequestCreate(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + mockStore := expectIMBearer(t, ctrl, "token-1", 7, 1) + mockStore.EXPECT().GetUidString().Return("request-1") + oldRelations := store.Relations + relations := &testRelationsStore{} + store.Relations = relations + defer func() { store.Relations = oldRelations }() + + req := httptest.NewRequest(http.MethodPost, "/v1/im/friend-requests", + bytes.NewBufferString(`{"target_user_id":"`+types.Uid(2).UserId()+`","message":"你好","source":"search"}`)) + req.Header.Set("Authorization", "Bearer token-1") + req.Header.Set("Idempotency-Key", "fr-1") + rec := httptest.NewRecorder() + + serveIMFriendRequests(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String()) + } + data := decodeIMHTTPResponse(t, rec).Data.(map[string]any) + if data["status"] != types.FriendRequestPending || relations.created == nil || relations.created.IdempotencyKey != "fr-1" { + t.Fatalf("unexpected friend request response: %#v created=%#v", data, relations.created) + } +} + func ptrTime(t time.Time) *time.Time { return &t } diff --git a/server/main.go b/server/main.go index 11d175e..5ce2136 100644 --- a/server/main.go +++ b/server/main.go @@ -793,6 +793,11 @@ func main() { mux.HandleFunc(config.ApiPath+"v1/im/system-notifications", serveIMSystemNotifications) mux.HandleFunc(config.ApiPath+"v1/im/system-notifications/", serveIMSystemNotificationRead) mux.HandleFunc(config.ApiPath+"v1/im/topics/", serveIMTopicMessageAction) + mux.HandleFunc(config.ApiPath+"v1/im/conversations/", serveIMConversationContext) + mux.HandleFunc(config.ApiPath+"v1/im/friend-requests", serveIMFriendRequests) + mux.HandleFunc(config.ApiPath+"v1/im/friend-requests/", serveIMFriendRequestAction) + mux.HandleFunc(config.ApiPath+"v1/im/blocks/", serveIMBlockUser) + mux.HandleFunc(config.ApiPath+"v1/im/friends/", serveIMFriendResource) // Handle long polling clients. Enable compression. mux.Handle(config.ApiPath+"v0/channels/lp", gh.CompressHandler(http.HandlerFunc(serveLongPoll))) if config.Media != nil { diff --git a/server/store/store.go b/server/store/store.go index f565831..794bf56 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -995,6 +995,157 @@ func (messagesMapper) Edit(tenantID types.TenantID, topic string, seq int, actor return tenantAdp.TenantMessageEdit(tenantID, topic, seq, actor, expectedVersion, head, content, reason, editedAt) } +type RelationsPersistenceInterface interface { + CreateFriendRequest(tenantID types.TenantID, req *types.FriendRequest) (*types.FriendRequest, error) + ResolveFriendRequest(tenantID types.TenantID, requestID string, actor types.Uid, status string, handledAt time.Time) (*types.FriendRequest, error) + GetRelationshipContext(tenantID types.TenantID, userID, peerID types.Uid, topic string) (*types.RelationshipContext, error) + CanSendP2P(tenantID types.TenantID, userID, peerID types.Uid) (bool, string, error) + BlockUser(tenantID types.TenantID, blockerID, blockedID types.Uid, now time.Time) (*types.UserBlock, error) + UnblockUser(tenantID types.TenantID, blockerID, blockedID types.Uid) error + DeleteFriend(tenantID types.TenantID, userID, peerID types.Uid, now time.Time) error +} + +type relationsMapper struct{} + +var Relations RelationsPersistenceInterface + +func (relationsMapper) CreateFriendRequest(tenantID types.TenantID, req *types.FriendRequest) (*types.FriendRequest, error) { + if req == nil || req.TenantID != tenantID || req.ApplicantID.IsZero() || req.TargetID.IsZero() || req.ApplicantID == req.TargetID { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantFriendRequestCreate(tenantID, req) +} + +func (relationsMapper) ResolveFriendRequest(tenantID types.TenantID, requestID string, actor types.Uid, status string, handledAt time.Time) (*types.FriendRequest, error) { + if strings.TrimSpace(requestID) == "" || actor.IsZero() || (status != types.FriendRequestAccepted && status != types.FriendRequestRejected) { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + req, err := tenantAdp.TenantFriendRequestResolve(tenantID, requestID, actor, status, handledAt) + if err != nil { + return nil, err + } + if status == types.FriendRequestAccepted { + if err := tenantAdp.TenantFriendshipUpsertPair(tenantID, req.ApplicantID, req.TargetID, "", handledAt); err != nil { + return nil, err + } + } + return req, nil +} + +func (relationsMapper) GetRelationshipContext(tenantID types.TenantID, userID, peerID types.Uid, topic string) (*types.RelationshipContext, error) { + if userID.IsZero() || peerID.IsZero() || userID == peerID { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return relationshipContext(tenantAdp, tenantID, userID, peerID, topic) +} + +func (relationsMapper) CanSendP2P(tenantID types.TenantID, userID, peerID types.Uid) (bool, string, error) { + ctx, err := (relationsMapper{}).GetRelationshipContext(tenantID, userID, peerID, "") + if err != nil { + return false, "", err + } + if ctx.Relationship == types.RelationshipFriend { + return true, ctx.Relationship, nil + } + return false, ctx.Relationship, nil +} + +func (relationsMapper) BlockUser(tenantID types.TenantID, blockerID, blockedID types.Uid, now time.Time) (*types.UserBlock, error) { + if blockerID.IsZero() || blockedID.IsZero() || blockerID == blockedID { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantUserBlockUpsert(tenantID, blockerID, blockedID, now) +} + +func (relationsMapper) UnblockUser(tenantID types.TenantID, blockerID, blockedID types.Uid) error { + if blockerID.IsZero() || blockedID.IsZero() || blockerID == blockedID { + return types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return err + } + return tenantAdp.TenantUserBlockDelete(tenantID, blockerID, blockedID) +} + +func (relationsMapper) DeleteFriend(tenantID types.TenantID, userID, peerID types.Uid, now time.Time) error { + if userID.IsZero() || peerID.IsZero() || userID == peerID { + return types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return err + } + return tenantAdp.TenantFriendshipDeletePair(tenantID, userID, peerID, now) +} + +func relationshipContext(tenantAdp adapter.TenantBusinessAdapter, tenantID types.TenantID, userID, peerID types.Uid, topic string) (*types.RelationshipContext, error) { + ctx := &types.RelationshipContext{ + TenantID: tenantID, + UserID: userID, + PeerID: peerID, + Topic: topic, + Relationship: types.RelationshipNone, + } + peer, err := tenantAdp.TenantUserGet(tenantID, peerID) + if err != nil { + return nil, err + } + ctx.Peer = peer + if block, err := tenantAdp.TenantUserBlockGet(tenantID, userID, peerID); err != nil { + return nil, err + } else if block != nil { + ctx.Relationship = types.RelationshipBlockedByMe + return ctx, nil + } + if block, err := tenantAdp.TenantUserBlockGet(tenantID, peerID, userID); err != nil { + return nil, err + } else if block != nil { + ctx.Relationship = types.RelationshipBlockedByPeer + return ctx, nil + } + if friendship, err := tenantAdp.TenantFriendshipGet(tenantID, userID, peerID); err != nil { + return nil, err + } else if friendship != nil && friendship.Status == types.RelationshipFriend { + ctx.Relationship = types.RelationshipFriend + ctx.CanSend = true + if ctx.Topic == "" { + ctx.Topic = friendship.Topic + } + return ctx, nil + } + req, err := tenantAdp.TenantFriendRequestLatestBetween(tenantID, userID, peerID) + if err != nil { + return nil, err + } + if req != nil && req.Status == types.FriendRequestPending && req.ExpiresAt.After(time.Now()) { + ctx.RequestID = req.RequestID + ctx.PendingStatus = req.Status + if req.ApplicantID == userID { + ctx.Relationship = types.RelationshipPendingOutbound + } else { + ctx.Relationship = types.RelationshipPendingInbound + } + } + return ctx, nil +} + // Save message func (messagesMapper) Save(tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) (*types.MessageSaveResult, error) { if msg == nil || msg.TenantID != tenantID { @@ -1814,4 +1965,5 @@ func init() { PCache = pcacheMapper{} Verifications = verificationMapper{} Security = userSecurityMapper{} + Relations = relationsMapper{} } diff --git a/server/store/types/relations.go b/server/store/types/relations.go new file mode 100644 index 0000000..037ff0a --- /dev/null +++ b/server/store/types/relations.go @@ -0,0 +1,65 @@ +package types + +import "time" + +const ( + FriendRequestPending = "pending" + FriendRequestAccepted = "accepted" + FriendRequestRejected = "rejected" + FriendRequestExpired = "expired" + + RelationshipNone = "none" + RelationshipFriend = "friend" + RelationshipPendingInbound = "pending_inbound" + RelationshipPendingOutbound = "pending_outbound" + RelationshipBlockedByMe = "blocked_by_me" + RelationshipBlockedByPeer = "blocked_by_peer" +) + +type FriendRequest struct { + ID int64 + TenantID TenantID `db:"tenant_id" json:"-" bson:"tenant_id"` + RequestID string `db:"request_id"` + ApplicantID Uid `db:"applicant_id"` + TargetID Uid `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"` +} + +type Friendship struct { + TenantID TenantID `db:"tenant_id" json:"-" bson:"tenant_id"` + UserID Uid `db:"user_id"` + PeerID Uid `db:"peer_id"` + Status string + Remark string + Topic string + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` +} + +type UserBlock struct { + TenantID TenantID `db:"tenant_id" json:"-" bson:"tenant_id"` + BlockerID Uid `db:"blocker_id"` + BlockedID Uid `db:"blocked_id"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` +} + +type RelationshipContext struct { + TenantID TenantID `json:"-"` + UserID Uid `json:"-"` + PeerID Uid `json:"-"` + Topic string `json:"topic"` + Relationship string `json:"relationship"` + CanSend bool `json:"can_send"` + RequestID string `json:"request_id,omitempty"` + PendingStatus string `json:"pending_status,omitempty"` + Peer *User `json:"peer,omitempty"` +} diff --git a/server/topic.go b/server/topic.go index aae0923..33b4881 100644 --- a/server/topic.go +++ b/server/topic.go @@ -1013,6 +1013,16 @@ func (t *Topic) sendSubNotifications(asUid types.Uid, sid, userAgent string) { // in response to a client request (msg, asUid) and broadcasts it to the attached sessions. func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid, noEcho bool, attachments []string, head map[string]any, content any) error { pud, userFound := t.perUser[asUid] + if !userFound && msg.sess != nil { + asUid = msg.sess.uid + pud, userFound = t.perUser[asUid] + } + if !userFound && msg.sess != nil { + if psd, ok := t.sessions[msg.sess]; ok { + asUid = psd.uid + pud, userFound = t.perUser[asUid] + } + } if t.cat == types.TopicCatSys { if msg.sess != nil && msg.sess.authLvl != auth.LevelRoot { msg.sess.queueOut(ErrPermissionDenied(msg.Id, t.original(asUid), msg.Timestamp)) @@ -1042,6 +1052,22 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid, return types.ErrPermissionDenied } } + if t.cat == types.TopicCatP2P && msg.sess != nil && msg.sess.authLvl != auth.LevelRoot && + store.Relations != nil && store.Store != nil && store.Store.IsOpen() { + peer := t.p2pOtherUser(asUid) + if !peer.IsZero() { + canSend, _, err := store.Relations.CanSendP2P(t.tenantID, asUid, peer) + if err != nil { + logs.Warn.Printf("topic[%s]: failed to check p2p relationship: %v", t.name, err) + msg.sess.queueOut(ErrUnknown(msg.Id, t.original(asUid), msg.Timestamp)) + return err + } + if !canSend { + msg.sess.queueOut(ErrPermissionDenied(msg.Id, t.original(asUid), msg.Timestamp)) + return types.ErrPermissionDenied + } + } + } saveResult, err := store.Messages.Save(t.tenantID, &types.Message{ @@ -1124,6 +1150,9 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid, // This is a NON-proxy broadcast. func (t *Topic) handlePubBroadcast(msg *ClientComMessage) { asUid := types.ParseUserId(msg.AsUser) + if asUid.IsZero() && msg.sess != nil { + asUid = msg.sess.uid + } if t.isInactive() { // Ignore broadcast - topic is paused or being deleted. msg.sess.queueOut(ErrLocked(msg.Id, t.original(asUid), msg.Timestamp)) diff --git a/server/topic_test.go b/server/topic_test.go index a44b888..72dd72e 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -49,6 +49,13 @@ type TopicTestHelper struct { uu *mock_store.MockUsersPersistenceInterface tt *mock_store.MockTopicsPersistenceInterface ss *mock_store.MockSubsPersistenceInterface + + oldMessages store.MessagesPersistenceInterface + oldUsers store.UsersPersistenceInterface + oldTopics store.TopicsPersistenceInterface + oldSubs store.SubsPersistenceInterface + oldRelations store.RelationsPersistenceInterface + oldModeCP2P types.AccessMode } func (b *TopicTestHelper) finish() { @@ -67,12 +74,14 @@ func (b *TopicTestHelper) finish() { func (b *TopicTestHelper) newSession(sid string, uid types.Uid) (*Session, *responses) { s := &Session{ - tenantID: 1, - sid: sid, - uid: uid, - subs: make(map[string]*Subscription), - send: make(chan any, 10), - detach: make(chan string, 10), + tenantID: 1, + tenantCode: "test", + sid: sid, + uid: uid, + authLvl: auth.LevelAuth, + subs: make(map[string]*Subscription), + send: make(chan any, 10), + detach: make(chan string, 10), } r := &responses{} b.sessWg.Add(1) @@ -95,10 +104,18 @@ func (b *TopicTestHelper) setUp(t *testing.T, numUsers int, cat types.TopicCat, b.uu = mock_store.NewMockUsersPersistenceInterface(b.ctrl) b.tt = mock_store.NewMockTopicsPersistenceInterface(b.ctrl) b.ss = mock_store.NewMockSubsPersistenceInterface(b.ctrl) + b.oldMessages = store.Messages + b.oldUsers = store.Users + b.oldTopics = store.Topics + b.oldSubs = store.Subs store.Messages = b.mm store.Users = b.uu store.Topics = b.tt store.Subs = b.ss + b.oldRelations = store.Relations + store.Relations = nil + b.oldModeCP2P = globals.typesModeCP2P + globals.typesModeCP2P = types.ModeCP2P // Sessions. b.sessions = make([]*Session, b.numUsers) b.results = make([]*responses, b.numUsers) @@ -162,10 +179,12 @@ func (b *TopicTestHelper) setUp(t *testing.T, numUsers int, cat types.TopicCat, func (b *TopicTestHelper) tearDown() { globals.hub = nil - store.Messages = nil - store.Users = nil - store.Topics = nil - store.Subs = nil + store.Messages = b.oldMessages + store.Users = b.oldUsers + store.Topics = b.oldTopics + store.Subs = b.oldSubs + store.Relations = b.oldRelations + globals.typesModeCP2P = b.oldModeCP2P b.ctrl.Finish() } @@ -279,6 +298,42 @@ func TestHandleBroadcastDataP2P(t *testing.T) { } } +func TestHandleBroadcastDataP2PRequiresRelationship(t *testing.T) { + numUsers := 2 + helper := TopicTestHelper{} + helper.setUp(t, numUsers, types.TopicCatP2P, "p2p-test" /*attach=*/, true) + defer helper.tearDown() + + oldStore := store.Store + mockStore := mock_store.NewMockPersistentStorageInterface(helper.ctrl) + mockStore.EXPECT().IsOpen().Return(true) + store.Store = mockStore + canSend := false + store.Relations = &testRelationsStore{canSend: &canSend, reason: types.RelationshipNone} + defer func() { + store.Store = oldStore + }() + + from := helper.uids[0].UserId() + msg := &ClientComMessage{ + AsUser: from, + Original: from, + Pub: &MsgClientPub{ + Topic: "p2p", + Content: "test", + NoEcho: true, + }, + sess: helper.sessions[0], + } + helper.topic.handleClientMsg(msg) + helper.finish() + + verifyResponseCodes(helper.results[0], []int{http.StatusForbidden}, t) + if len(helper.results[1].messages) != 0 { + t.Fatalf("peer: expected 0 messages, got %d", len(helper.results[1].messages)) + } +} + func TestHandleBroadcastCall(t *testing.T) { numUsers := 2 helper := TopicTestHelper{}