diff --git a/scripts/apply_hotgo_tenant_db.go b/scripts/apply_hotgo_tenant_db.go index fb0a0c4..daa0d65 100644 --- a/scripts/apply_hotgo_tenant_db.go +++ b/scripts/apply_hotgo_tenant_db.go @@ -57,6 +57,29 @@ func main() { CONSTRAINT fk_im_tenant_config_tenant FOREIGN KEY(tenant_id) REFERENCES hotgo.im_tenant(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, "CREATE DATABASE IF NOT EXISTS tinode CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci", + `CREATE TABLE IF NOT EXISTS tinode.im_call_records( + id BIGINT NOT NULL AUTO_INCREMENT, + call_id VARCHAR(64) NOT NULL, + topic CHAR(25) NOT NULL, + seq_id INT NOT NULL, + caller_id BIGINT NOT NULL, + callee_id BIGINT NOT NULL, + call_type VARCHAR(16) NOT NULL, + status VARCHAR(32) NOT NULL, + end_reason VARCHAR(64) NOT NULL DEFAULT '', + duration_ms BIGINT NOT NULL DEFAULT 0, + started_at DATETIME(3) NOT NULL, + accepted_at DATETIME(3), + finished_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_call_records_call(call_id), + UNIQUE INDEX uk_im_call_records_topic_seq(topic, seq_id), + INDEX idx_im_call_records_caller(caller_id, id DESC), + INDEX idx_im_call_records_callee(callee_id, id DESC), + INDEX idx_im_call_records_status(status, updated_at) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, } for _, stmt := range statements { if _, err = db.Exec(stmt); err != nil { @@ -88,6 +111,9 @@ func main() { if _, err = db.Exec("UPDATE tinode.im_tenant SET db_name='tinode' WHERE code='default' AND (db_name='' OR db_name IS NULL)"); err != nil { fatal(err) } + if err = dropTenantIDColumns(db, "tinode"); err != nil { + fatal(err) + } var code, dbName string var state int @@ -101,9 +127,107 @@ func main() { fatal(err) } fmt.Printf("tinode.im_tenant routing columns: %d/3\n", columnCount) + + rows, err := db.Query(`SELECT table_name FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA='tinode' AND COLUMN_NAME='tenant_id' + ORDER BY table_name`) + if err != nil { + fatal(err) + } + defer rows.Close() + var tables []string + for rows.Next() { + var table string + if err = rows.Scan(&table); err != nil { + fatal(err) + } + tables = append(tables, table) + } + if err = rows.Err(); err != nil { + fatal(err) + } + if len(tables) == 0 { + fmt.Println("tinode tenant_id columns: none") + } else { + fmt.Printf("tinode tenant_id columns: %s\n", strings.Join(tables, ",")) + } } func fatal(v any) { fmt.Fprintln(os.Stderr, v) os.Exit(1) } + +func dropTenantIDColumns(db *sql.DB, schema string) error { + fks, err := queryPairs(db, `SELECT DISTINCT table_name, constraint_name + FROM information_schema.KEY_COLUMN_USAGE + WHERE TABLE_SCHEMA=? AND COLUMN_NAME='tenant_id' AND REFERENCED_TABLE_NAME IS NOT NULL`, + schema) + if err != nil { + return err + } + for _, fk := range fks { + if _, err = db.Exec("ALTER TABLE " + quoteIdent(schema) + "." + quoteIdent(fk[0]) + " DROP FOREIGN KEY " + quoteIdent(fk[1])); err != nil { + return fmt.Errorf("drop foreign key %s.%s: %w", fk[0], fk[1], err) + } + } + + indexes, err := queryPairs(db, `SELECT DISTINCT table_name, index_name + FROM information_schema.STATISTICS + WHERE TABLE_SCHEMA=? AND COLUMN_NAME='tenant_id' AND INDEX_NAME<>'PRIMARY'`, + schema) + if err != nil { + return err + } + for _, idx := range indexes { + if _, err = db.Exec("ALTER TABLE " + quoteIdent(schema) + "." + quoteIdent(idx[0]) + " DROP INDEX " + quoteIdent(idx[1])); err != nil { + return fmt.Errorf("drop index %s.%s: %w", idx[0], idx[1], err) + } + } + + rows, err := db.Query(`SELECT table_name FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA=? AND COLUMN_NAME='tenant_id' + ORDER BY table_name`, schema) + if err != nil { + return err + } + defer rows.Close() + var tables []string + for rows.Next() { + var table string + if err = rows.Scan(&table); err != nil { + return err + } + tables = append(tables, table) + } + if err = rows.Err(); err != nil { + return err + } + for _, table := range tables { + if _, err = db.Exec("ALTER TABLE " + quoteIdent(schema) + "." + quoteIdent(table) + " DROP COLUMN tenant_id"); err != nil { + return fmt.Errorf("drop column %s.tenant_id: %w", table, err) + } + } + return nil +} + +func queryPairs(db *sql.DB, query string, args ...any) ([][2]string, error) { + rows, err := db.Query(query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var pairs [][2]string + for rows.Next() { + var first, second string + if err = rows.Scan(&first, &second); err != nil { + return nil, err + } + pairs = append(pairs, [2]string{first, second}) + } + return pairs, rows.Err() +} + +func quoteIdent(name string) string { + return "`" + strings.ReplaceAll(name, "`", "``") + "`" +} diff --git a/server/calls.go b/server/calls.go index 9c7b110..14d625c 100644 --- a/server/calls.go +++ b/server/calls.go @@ -12,9 +12,11 @@ import ( "fmt" "os" "strconv" + "strings" "time" "github.com/tinode/chat/server/logs" + "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" jcr "github.com/tinode/jsonco" ) @@ -45,6 +47,9 @@ const ( constCallMsgMissed = "missed" // Call is declined (the callee hung up before picking up). constCallMsgDeclined = "declined" + + constCallTypeAudio = "audio" + constCallTypeVideo = "video" ) type callConfig struct { @@ -86,10 +91,127 @@ type videoCall struct { content any // Call message content mime type. contentMime any + // Product-level call identifier. + callID string + // Product-level call type: audio or video. + callType string // Time when the call was accepted. acceptedAt time.Time } +func normalizeCallInvite(head map[string]any, content any) (map[string]any, any, error) { + if head == nil { + return nil, nil, types.ErrMalformed + } + state, _ := head["webrtc"].(string) + if state != "started" { + return nil, nil, types.ErrMalformed + } + call, ok := content.(map[string]any) + if !ok { + if typed, typedOK := content.(types.KVMap); typedOK { + call = map[string]any(typed) + } + } + if len(call) == 0 { + return nil, nil, types.ErrMalformed + } + callID, _ := call["callId"].(string) + callType, _ := call["callType"].(string) + if callType == "" { + callType, _ = call["media_type"].(string) + } + callType = strings.ToLower(strings.TrimSpace(callType)) + if strings.TrimSpace(callID) == "" || (callType != constCallTypeAudio && callType != constCallTypeVideo) { + return nil, nil, types.ErrMalformed + } + normalizedHead := make(map[string]any, len(head)+2) + for key, value := range head { + normalizedHead[key] = value + } + normalizedHead["webrtc"] = "started" + normalizedHead["x-im-type"] = "call" + if _, ok := normalizedHead["mime"]; !ok { + normalizedHead["mime"] = "application/json" + } + normalizedContent := make(map[string]any, len(call)+3) + for key, value := range call { + normalizedContent[key] = value + } + normalizedContent["callId"] = strings.TrimSpace(callID) + normalizedContent["callType"] = callType + normalizedContent["audio"] = true + normalizedContent["video"] = callType == constCallTypeVideo + return normalizedHead, normalizedContent, nil +} + +func callPayloadString(payload any, key string) string { + data, ok := payload.(map[string]any) + if !ok { + if typed, typedOK := payload.(types.KVMap); typedOK { + data = map[string]any(typed) + } + } + if len(data) == 0 { + return "" + } + value, _ := data[key].(string) + return strings.TrimSpace(value) +} + +func callContentString(content any, key string) string { + return callPayloadString(content, key) +} + +func (call *videoCall) matchesPayload(payload any) bool { + payloadCallID := callPayloadString(payload, "callId") + return payloadCallID == "" || payloadCallID == call.callID +} + +func (t *Topic) auditCallCreate(call *videoCall, caller types.Uid) { + if store.Calls == nil || call == nil || call.callID == "" || caller.IsZero() { + return + } + callee := t.p2pOtherUser(caller) + if callee.IsZero() { + return + } + now := types.TimeNow() + if err := store.Calls.Create(t.tenantID, &types.CallRecord{ + TenantID: t.tenantID, + CallID: call.callID, + Topic: t.name, + SeqID: call.seq, + CallerID: caller, + CalleeID: callee, + CallType: call.callType, + Status: types.CallStatusStarted, + StartedAt: now, + CreatedAt: now, + UpdatedAt: now, + }); err != nil { + logs.Warn.Printf("topic[%s]: failed to audit call start %s: %v", t.name, call.callID, err) + } +} + +func (t *Topic) auditCallAccept(call *videoCall, acceptedAt time.Time) { + if store.Calls == nil || call == nil || call.callID == "" { + return + } + if err := store.Calls.Accept(t.tenantID, call.callID, call.seq, acceptedAt); err != nil { + logs.Warn.Printf("topic[%s]: failed to audit call accept %s: %v", t.name, call.callID, err) + } +} + +func (t *Topic) auditCallFinish(call *videoCall, status string, durationMS int64, finishedAt time.Time) { + if store.Calls == nil || call == nil || call.callID == "" { + return + } + if err := store.Calls.Finish(t.tenantID, call.callID, call.seq, status, durationMS, finishedAt); err != nil { + logs.Warn.Printf("topic[%s]: failed to audit call finish %s: %v", t.name, call.callID, err) + } +} + // callPartySession returns a session to be stored in the call party data. func callPartySession(sess *Session) *Session { if sess.isProxy() { @@ -226,12 +348,15 @@ func (t *Topic) handleCallInvite(msg *ClientComMessage, asUid types.Uid) { seq: t.lastID, content: msg.Pub.Content, contentMime: msg.Pub.Head["mime"], + callID: callContentString(msg.Pub.Content, "callId"), + callType: callContentString(msg.Pub.Content, "callType"), } t.currentCall.parties[msg.sess.sid] = callPartyData{ uid: asUid, isOriginator: true, sess: callPartySession(msg.sess), } + t.auditCallCreate(t.currentCall, asUid) // Wait for constCallEstablishmentTimeout for the other side to accept the call. t.callEstablishmentTimer.Reset(time.Duration(globals.callEstablishmentTimeout) * time.Second) } @@ -255,6 +380,10 @@ func (t *Topic) handleCallEvent(msg *ClientComMessage) { logs.Info.Printf("topic[%s]: invalid seq id - current call (%d) vs received (%d)", t.name, t.currentCall.seq, call.SeqId) return } + if !t.currentCall.matchesPayload(call.Payload) { + logs.Info.Printf("topic[%s]: invalid call id for seq %d", t.name, t.currentCall.seq) + return + } asUid := types.ParseUserId(msg.AsUser) @@ -307,6 +436,7 @@ func (t *Topic) handleCallEvent(msg *ClientComMessage) { sess: callPartySession(msg.sess), } t.currentCall.acceptedAt = time.Now() + t.auditCallAccept(t.currentCall, t.currentCall.acceptedAt) // Notify other clients that the call has been accepted. t.infoCallSubsOffline(msg.AsUser, asUid, call.Event, t.currentCall.seq, call.Payload, msg.sess.sid, false) @@ -382,10 +512,11 @@ func (t *Topic) maybeEndCallInProgress(from string, msg *ClientComMessage, callD originatorUid, _ := t.getCallOriginator() var replaceWith string var callDuration int64 + finishedAt := time.Now() if from != "" && len(t.currentCall.parties) == 2 { // This is a call in progress. replaceWith = constCallMsgFinished - callDuration = time.Since(t.currentCall.acceptedAt).Milliseconds() + callDuration = finishedAt.Sub(t.currentCall.acceptedAt).Milliseconds() } else { if from != "" { // User originated hang-up. @@ -418,6 +549,7 @@ func (t *Topic) maybeEndCallInProgress(from string, msg *ClientComMessage, callD if err := t.saveAndBroadcastMessage(&msgCopy, originatorUid, false, nil, head, t.currentCall.content); err != nil { logs.Err.Printf("topic[%s]: failed to write finalizing message for call seq id %d - '%s'", t.name, t.currentCall.seq, err) } + t.auditCallFinish(t.currentCall, callFinalStatus(replaceWith), callDuration, finishedAt) // Send {info} hangup event to the subscribed sessions. t.broadcastToSessions(t.currentCall.infoMessage(constCallEventHangUp)) @@ -429,6 +561,19 @@ func (t *Topic) maybeEndCallInProgress(from string, msg *ClientComMessage, callD t.currentCall = nil } +func callFinalStatus(replaceWith string) string { + switch replaceWith { + case constCallMsgFinished: + return types.CallStatusFinished + case constCallMsgMissed: + return types.CallStatusMissed + case constCallMsgDeclined: + return types.CallStatusDeclined + default: + return types.CallStatusDisconnected + } +} + // Server initiated call termination. func (t *Topic) terminateCallInProgress(callDidTimeout bool) { if t.currentCall == nil { diff --git a/server/db/adapter.go b/server/db/adapter.go index 849f7b6..992b1e3 100644 --- a/server/db/adapter.go +++ b/server/db/adapter.go @@ -289,6 +289,10 @@ type TenantBusinessAdapter interface { TenantMessageGetDeleted(t.TenantID, string, t.Uid, *t.QueryOpt) ([]t.DelMessage, error) TenantMessageSearch(t.TenantID, string, t.Uid, string, string, string, int) ([]t.Message, string, error) + TenantCallCreate(t.TenantID, *t.CallRecord) error + TenantCallAccept(t.TenantID, string, int, time.Time) error + TenantCallFinish(t.TenantID, string, int, string, int64, time.Time) error + TenantFavoriteCreate(t.TenantID, *t.FavoriteItem) (*t.FavoriteItem, error) TenantFavoriteList(t.TenantID, t.Uid, string, string, string, int) ([]t.FavoriteItem, string, error) TenantFavoriteGet(t.TenantID, t.Uid, string) (*t.FavoriteItem, error) diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index 2e3984f..a35c565 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -791,6 +791,9 @@ func (a *adapter) CreateDb(reset bool) error { if err = createIMRelationTables(tx); err != nil { return err } + if err = createIMCallTables(tx); err != nil { + return err + } // Deletion log if _, err = tx.Exec( @@ -919,6 +922,34 @@ func createIMNotificationTables(tx *sql.Tx) error { return err } +func createIMCallTables(tx *sql.Tx) error { + _, err := tx.Exec( + `CREATE TABLE im_call_records( + id BIGINT NOT NULL AUTO_INCREMENT, + call_id VARCHAR(64) NOT NULL, + topic CHAR(25) NOT NULL, + seq_id INT NOT NULL, + caller_id BIGINT NOT NULL, + callee_id BIGINT NOT NULL, + call_type VARCHAR(16) NOT NULL, + status VARCHAR(32) NOT NULL, + end_reason VARCHAR(64) NOT NULL DEFAULT '', + duration_ms BIGINT NOT NULL DEFAULT 0, + started_at DATETIME(3) NOT NULL, + accepted_at DATETIME(3), + finished_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_call_records_call(call_id), + UNIQUE INDEX uk_im_call_records_topic_seq(topic, seq_id), + INDEX idx_im_call_records_caller(caller_id, id DESC), + INDEX idx_im_call_records_callee(callee_id, id DESC), + INDEX idx_im_call_records_status(status, updated_at) + )`) + return err +} + func ensureTenantRoutingColumns(a *adapter) error { statements := []string{ "ALTER TABLE im_tenant ADD COLUMN db_name VARCHAR(128) NOT NULL DEFAULT '' AFTER tenant_desc", @@ -1258,7 +1289,7 @@ func (a *adapter) TenantAuthDelScheme(tenantID t.TenantID, uid t.Uid, scheme str if cancel != nil { defer cancel() } - _, err := a.db.ExecContext(ctx, "DELETE FROM auth WHERE tenant_id=? AND userid=? AND scheme=?", + _, err = a.db.ExecContext(ctx, "DELETE FROM auth WHERE tenant_id=? AND userid=? AND scheme=?", tenantID, store.DecodeUid(uid), scheme) return err } @@ -3942,7 +3973,7 @@ func (a *adapter) TenantCredFail(tenantID t.TenantID, uid t.Uid, method string) if cancel != nil { defer cancel() } - _, err := a.db.ExecContext(ctx, "UPDATE credentials SET updatedat=?,retries=retries+1 "+ + _, err = a.db.ExecContext(ctx, "UPDATE credentials SET updatedat=?,retries=retries+1 "+ "WHERE tenant_id=? AND userid=? AND method=? AND done=FALSE", t.TimeNow(), tenantID, store.DecodeUid(uid), method) return err @@ -3983,7 +4014,7 @@ func (a *adapter) TenantCredGetActive(tenantID t.TenantID, uid t.Uid, method str defer cancel() } var cred t.Credential - err := a.db.GetContext(ctx, &cred, "SELECT createdat,updatedat,method,value,resp,done,retries "+ + err = a.db.GetContext(ctx, &cred, "SELECT createdat,updatedat,method,value,resp,done,retries "+ "FROM credentials WHERE tenant_id=? AND userid=? AND deletedat IS NULL AND method=? AND done=FALSE", tenantID, store.DecodeUid(uid), method) if err == sql.ErrNoRows { @@ -4292,6 +4323,57 @@ func (a *adapter) FileLinkAttachments(topic string, userId, msgId t.Uid, fids [] return tx.Commit() } +func (a *adapter) FileUserHasAccess(fid string, uid t.Uid) (bool, error) { + if uid.IsZero() { + return false, t.ErrMalformed + } + fileID := t.ParseUid(fid) + if fileID.IsZero() { + return false, t.ErrMalformed + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + var allowed int + err := a.db.GetContext(ctx, &allowed, ` + SELECT COUNT(*) FROM fileuploads AS fu + WHERE fu.id=? AND fu.status=? AND ( + fu.userid=? + OR EXISTS ( + SELECT 1 FROM filemsglinks AS fml + WHERE fml.fileid=fu.id AND fml.userid=? + ) + OR EXISTS ( + SELECT 1 FROM filemsglinks AS fml + INNER JOIN subscriptions AS s + ON s.topic=fml.topic AND s.userid=? + WHERE fml.fileid=fu.id AND fml.topic IS NOT NULL + AND s.deletedat IS NULL AND INSTR(s.modewant,'R')>0 AND INSTR(s.modegiven,'R')>0 + ) + OR EXISTS ( + SELECT 1 FROM filemsglinks AS fml + INNER JOIN messages AS m + ON m.id=fml.msgid + INNER JOIN subscriptions AS s + ON s.topic=m.topic AND s.userid=? + WHERE fml.fileid=fu.id AND fml.msgid IS NOT NULL + AND m.deletedat IS NULL AND m.recalled_at IS NULL + AND s.deletedat IS NULL AND INSTR(s.modewant,'R')>0 AND INSTR(s.modegiven,'R')>0 + AND NOT EXISTS ( + SELECT 1 FROM dellog AS d + WHERE d.topic=m.topic AND d.deletedfor=? + AND d.low<=m.seqid AND (d.hi=0 OR d.hi>m.seqid) + ) + ) + ) LIMIT 1`, + store.DecodeUid(fileID), t.UploadCompleted, store.DecodeUid(uid), store.DecodeUid(uid), store.DecodeUid(uid), store.DecodeUid(uid), store.DecodeUid(uid)) + if err != nil { + return false, err + } + return allowed > 0, nil +} + // PCacheGet reads a persistet cache entry. func (a *adapter) PCacheGet(key string) (string, error) { ctx, cancel := a.getContext() diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index 122fc88..0e99c8e 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -127,6 +127,9 @@ func (a *adapter) TenantUserCreate(id t.TenantID, user *t.User) error { if err != nil || user == nil || user.TenantID != id { return t.ErrMalformed } + if tenantAdp.tenantScopedDB { + return tenantAdp.UserCreate(user) + } return tenantAdp.UserCreate(user) } @@ -139,12 +142,16 @@ func (a *adapter) TenantUserGet(id t.TenantID, uid t.Uid) (*t.User, error) { user, err := tenantAdp.UserGet(uid) return stampTenantUser(id, user), err } + if a.tenantScopedDB { + user, err := a.UserGet(uid) + return stampTenantUser(id, user), err + } ctx, cancel := a.getContext() if cancel != nil { defer cancel() } var user t.User - err := a.db.GetContext(ctx, &user, "SELECT * FROM users WHERE tenant_id=? AND id=? AND state!=?", + err = a.db.GetContext(ctx, &user, "SELECT * FROM users WHERE tenant_id=? AND id=? AND state!=?", id, decodeUID(uid), t.StateDeleted) if err == sql.ErrNoRows { return nil, nil @@ -165,6 +172,10 @@ func (a *adapter) TenantUserGetAll(id t.TenantID, uids ...t.Uid) ([]t.User, erro users, err := tenantAdp.UserGetAll(uids...) return stampTenantUsers(id, users), err } + if a.tenantScopedDB { + users, err := a.UserGetAll(uids...) + return stampTenantUsers(id, users), err + } ctx, cancel := a.getContext() if cancel != nil { defer cancel() @@ -185,7 +196,13 @@ func (a *adapter) tenantUserGetAllTx(ctx context.Context, tx *sqlx.Tx, id t.Tena if len(args) == 0 { return nil, nil } - q, params, _ := sqlx.In("SELECT * FROM users WHERE tenant_id=? AND id IN (?) AND state!=?", id, args, t.StateDeleted) + var q string + var params []any + if a.tenantScopedDB { + q, params, _ = sqlx.In("SELECT * FROM users WHERE id IN (?) AND state!=?", args, t.StateDeleted) + } else { + q, params, _ = sqlx.In("SELECT * FROM users WHERE tenant_id=? AND id IN (?) AND state!=?", id, args, t.StateDeleted) + } query := a.db.Rebind(q) var rows *sqlx.Rows var err error @@ -219,6 +236,9 @@ func (a *adapter) TenantUserDelete(id t.TenantID, uid t.Uid, hard bool) error { if tenantAdp != a { return tenantAdp.UserDelete(uid, hard) } + if a.tenantScopedDB { + return a.UserDelete(uid, hard) + } decoded := decodeUID(uid) ctx, cancel := a.getContextForTx() if cancel != nil { @@ -351,6 +371,9 @@ func (a *adapter) TenantUserUpdate(id t.TenantID, uid t.Uid, update map[string]a if tenantAdp != a { return tenantAdp.UserUpdate(uid, update) } + if a.tenantScopedDB { + return a.UserUpdate(uid, update) + } ctx, cancel := a.getContextForTx() if cancel != nil { defer cancel() @@ -399,6 +422,9 @@ func (a *adapter) TenantUserUpdateTags(id t.TenantID, uid t.Uid, add, remove, re if tenantAdp != a { return tenantAdp.UserUpdateTags(uid, add, remove, reset) } + if a.tenantScopedDB { + return a.UserUpdateTags(uid, add, remove, reset) + } ctx, cancel := a.getContextForTx() if cancel != nil { defer cancel() @@ -446,6 +472,9 @@ func (a *adapter) TenantUserUnreadCount(id t.TenantID, uids ...t.Uid) (map[t.Uid if tenantAdp != a { return tenantAdp.UserUnreadCount(uids...) } + if a.tenantScopedDB { + return a.UserUnreadCount(uids...) + } counts := make(map[t.Uid]int, len(uids)) var decoded []any for _, uid := range uids { @@ -489,6 +518,9 @@ func (a *adapter) TenantUserGetUnvalidated(id t.TenantID, before time.Time, limi if tenantAdp != a { return tenantAdp.UserGetUnvalidated(before, limit) } + if a.tenantScopedDB { + return a.UserGetUnvalidated(before, limit) + } ctx, cancel := a.getContext() if cancel != nil { defer cancel() @@ -537,6 +569,9 @@ func (a *adapter) TenantTopicCreate(id t.TenantID, topic *t.Topic) error { if tenantAdp != a { return tenantAdp.TopicCreate(topic) } + if a.tenantScopedDB { + return a.TopicCreate(topic) + } ctx, cancel := a.getContextForTx() if cancel != nil { defer cancel() @@ -564,6 +599,9 @@ func (a *adapter) TenantTopicCreateP2P(id t.TenantID, one, two *t.Subscription) if tenantAdp != a { return tenantAdp.TopicCreateP2P(one, two) } + if a.tenantScopedDB { + return a.TopicCreateP2P(one, two) + } ctx, cancel := a.getContextForTx() if cancel != nil { defer cancel() @@ -601,12 +639,16 @@ func (a *adapter) TenantTopicGet(id t.TenantID, topic string) (*t.Topic, error) top, err := tenantAdp.TopicGet(topic) return stampTenantTopic(id, top), err } + if a.tenantScopedDB { + top, err := a.TopicGet(topic) + return stampTenantTopic(id, top), err + } ctx, cancel := a.getContext() if cancel != nil { defer cancel() } var top t.Topic - err := a.db.GetContext(ctx, &top, + err = a.db.GetContext(ctx, &top, "SELECT tenant_id,createdat,updatedat,state,stateat,touchedat,name AS id,usebt,access,owner,seqid,delid,subcnt,public,trusted,tags,aux "+ "FROM topics WHERE tenant_id=? AND name=?", id, topic) if err == sql.ErrNoRows { @@ -642,6 +684,10 @@ func (a *adapter) TenantTopicsForUser(id t.TenantID, uid t.Uid, keepDeleted bool subs, err := tenantAdp.TopicsForUser(uid, keepDeleted, opts) return stampTenantSubs(id, subs), err } + if a.tenantScopedDB { + subs, err := a.TopicsForUser(uid, keepDeleted, opts) + return stampTenantSubs(id, subs), err + } q := `SELECT tenant_id,createdat,updatedat,deletedat,topic,delid,recvseqid, readseqid,modewant,modegiven,private FROM subscriptions WHERE tenant_id=? AND userid=?` args := []any{id, decodeUID(uid)} @@ -805,6 +851,10 @@ func (a *adapter) TenantUsersForTopic(id t.TenantID, topic string, keepDeleted b subs, err := tenantAdp.UsersForTopic(topic, keepDeleted, opts) return stampTenantSubs(id, subs), err } + if a.tenantScopedDB { + subs, err := a.UsersForTopic(topic, keepDeleted, opts) + return stampTenantSubs(id, subs), err + } tcat := t.GetTopicCat(topic) q := `SELECT s.tenant_id,s.createdat,s.updatedat,s.deletedat,s.userid,s.topic,s.delid,s.recvseqid, s.readseqid,s.modewant,s.modegiven,u.public,u.trusted,u.lastseen,u.useragent,s.private @@ -924,6 +974,9 @@ func (a *adapter) TenantOwnTopics(id t.TenantID, uid t.Uid) ([]string, error) { if tenantAdp != a { return tenantAdp.OwnTopics(uid) } + if a.tenantScopedDB { + return a.OwnTopics(uid) + } return a.tenantTopicNames("SELECT name FROM topics WHERE tenant_id=? AND owner=? AND state!=?", id, decodeUID(uid), t.StateDeleted) } @@ -936,6 +989,9 @@ func (a *adapter) TenantChannelsForUser(id t.TenantID, uid t.Uid) ([]string, err if tenantAdp != a { return tenantAdp.ChannelsForUser(uid) } + if a.tenantScopedDB { + return a.ChannelsForUser(uid) + } return a.tenantTopicNames("SELECT topic FROM subscriptions WHERE tenant_id=? AND userid=? AND topic LIKE 'chn%' "+ "AND INSTR(modewant,'P')>0 AND INSTR(modegiven,'P')>0 AND deletedat IS NULL", id, decodeUID(uid)) } @@ -953,6 +1009,9 @@ func (a *adapter) TenantTopicShare(id t.TenantID, topic string, subs []*t.Subscr if tenantAdp != a { return tenantAdp.TopicShare(topic, subs) } + if a.tenantScopedDB { + return a.TopicShare(topic, subs) + } ctx, cancel := a.getContextForTx() if cancel != nil { defer cancel() @@ -1046,6 +1105,9 @@ func (a *adapter) TenantTopicDelete(id t.TenantID, topic string, isChan, hard bo if tenantAdp != a { return tenantAdp.TopicDelete(topic, isChan, hard) } + if a.tenantScopedDB { + return a.TopicDelete(topic, isChan, hard) + } topics := []any{topic} if isChan { topics = append(topics, t.GrpToChn(topic)) @@ -1099,11 +1161,14 @@ func (a *adapter) TenantTopicUpdateOnMessage(id t.TenantID, topic string, msg *t if tenantAdp != a { return tenantAdp.TopicUpdateOnMessage(topic, msg) } + if a.tenantScopedDB { + return a.TopicUpdateOnMessage(topic, msg) + } ctx, cancel := a.getContext() if cancel != nil { defer cancel() } - _, err := a.db.ExecContext(ctx, "UPDATE topics SET seqid=?,touchedat=? WHERE tenant_id=? AND name=?", + _, err = a.db.ExecContext(ctx, "UPDATE topics SET seqid=?,touchedat=? WHERE tenant_id=? AND name=?", msg.SeqId, msg.CreatedAt, id, topic) return err } @@ -1116,11 +1181,14 @@ func (a *adapter) TenantTopicUpdateSubCnt(id t.TenantID, topic string) error { if tenantAdp != a { return tenantAdp.TopicUpdateSubCnt(topic) } + if a.tenantScopedDB { + return a.TopicUpdateSubCnt(topic) + } ctx, cancel := a.getContext() if cancel != nil { defer cancel() } - _, err := a.db.ExecContext(ctx, + _, err = a.db.ExecContext(ctx, "UPDATE topics SET subcnt=(SELECT COUNT(*) FROM subscriptions WHERE tenant_id=? AND topic IN (?,?) AND deletedat IS NULL) WHERE tenant_id=? AND name=?", id, topic, t.GrpToChn(topic), id, topic) return err @@ -1134,6 +1202,9 @@ func (a *adapter) TenantTopicUpdate(id t.TenantID, topic string, update map[stri if tenantAdp != a { return tenantAdp.TopicUpdate(topic, update) } + if a.tenantScopedDB { + return a.TopicUpdate(topic, update) + } ctx, cancel := a.getContextForTx() if cancel != nil { defer cancel() @@ -1178,11 +1249,14 @@ func (a *adapter) TenantTopicOwnerChange(id t.TenantID, topic string, owner t.Ui if tenantAdp != a { return tenantAdp.TopicOwnerChange(topic, owner) } + if a.tenantScopedDB { + return a.TopicOwnerChange(topic, owner) + } ctx, cancel := a.getContext() if cancel != nil { defer cancel() } - _, err := a.db.ExecContext(ctx, "UPDATE topics SET owner=? WHERE tenant_id=? AND name=?", decodeUID(owner), id, topic) + _, err = a.db.ExecContext(ctx, "UPDATE topics SET owner=? WHERE tenant_id=? AND name=?", decodeUID(owner), id, topic) return err } @@ -1198,6 +1272,13 @@ func (a *adapter) TenantSubscriptionGet(id t.TenantID, topic string, uid t.Uid, } return sub, err } + if a.tenantScopedDB { + sub, err := a.SubscriptionGet(topic, uid, keepDeleted) + if sub != nil { + sub.TenantID = id + } + return sub, err + } query := `SELECT tenant_id,createdat,updatedat,deletedat,userid AS user,topic,delid,recvseqid, readseqid,modewant,modegiven,private FROM subscriptions WHERE tenant_id=? AND topic=? AND userid=?` if !keepDeleted { @@ -1208,7 +1289,7 @@ func (a *adapter) TenantSubscriptionGet(id t.TenantID, topic string, uid t.Uid, defer cancel() } var sub t.Subscription - err := a.db.GetContext(ctx, &sub, query, id, topic, decodeUID(uid)) + err = a.db.GetContext(ctx, &sub, query, id, topic, decodeUID(uid)) if err == sql.ErrNoRows { return nil, nil } @@ -1229,6 +1310,10 @@ func (a *adapter) TenantSubsForUser(id t.TenantID, uid t.Uid) ([]t.Subscription, subs, err := tenantAdp.SubsForUser(uid) return stampTenantSubs(id, subs), err } + if a.tenantScopedDB { + subs, err := a.SubsForUser(uid) + return stampTenantSubs(id, subs), err + } rows, err := a.db.Queryx(`SELECT tenant_id,createdat,updatedat,deletedat,userid AS user,topic,delid,recvseqid, readseqid,modewant,modegiven FROM subscriptions WHERE tenant_id=? AND userid=? AND deletedat IS NULL`, id, decodeUID(uid)) @@ -1257,6 +1342,10 @@ func (a *adapter) TenantSubsForTopic(id t.TenantID, topic string, keepDeleted bo subs, err := tenantAdp.SubsForTopic(topic, keepDeleted, opts) return stampTenantSubs(id, subs), err } + if a.tenantScopedDB { + subs, err := a.SubsForTopic(topic, keepDeleted, opts) + return stampTenantSubs(id, subs), err + } q := `SELECT tenant_id,createdat,updatedat,deletedat,userid AS user,topic,delid,recvseqid, readseqid,modewant,modegiven,private FROM subscriptions WHERE tenant_id=? AND topic=?` args := []any{id, topic} @@ -1305,6 +1394,9 @@ func (a *adapter) TenantSubsUpdate(id t.TenantID, topic string, uid t.Uid, updat if tenantAdp != a { return tenantAdp.SubsUpdate(topic, uid, update) } + if a.tenantScopedDB { + return a.SubsUpdate(topic, uid, update) + } cols, args := common.UpdateByMap(update) q := "UPDATE subscriptions SET " + strings.Join(cols, ",") + " WHERE tenant_id=? AND topic=?" args = append(args, id, topic) @@ -1316,7 +1408,7 @@ func (a *adapter) TenantSubsUpdate(id t.TenantID, topic string, uid t.Uid, updat if cancel != nil { defer cancel() } - _, err := a.db.ExecContext(ctx, q, args...) + _, err = a.db.ExecContext(ctx, q, args...) return err } @@ -1328,6 +1420,9 @@ func (a *adapter) TenantSubsDelete(id t.TenantID, topic string, uid t.Uid) error if tenantAdp != a { return tenantAdp.SubsDelete(topic, uid) } + if a.tenantScopedDB { + return a.SubsDelete(topic, uid) + } ctx, cancel := a.getContextForTx() if cancel != nil { defer cancel() @@ -1411,6 +1506,10 @@ func (a *adapter) TenantFind(id t.TenantID, caller, promoPrefix string, req [][] subs, err := tenantAdp.Find(caller, promoPrefix, req, opt, activeOnly) return stampTenantSubs(id, subs), err } + if a.tenantScopedDB { + subs, err := a.Find(caller, promoPrefix, req, opt, activeOnly) + return stampTenantSubs(id, subs), err + } allReq := t.FlattenDoubleSlice(req) allTags := append(append([]string{}, allReq...), opt...) if len(allTags) == 0 { @@ -1513,6 +1612,9 @@ func (a *adapter) TenantFindOne(id t.TenantID, tag string) (string, error) { if tenantAdp != a { return tenantAdp.FindOne(tag) } + if a.tenantScopedDB { + return a.FindOne(tag) + } ctx, cancel := a.getContext() if cancel != nil { defer cancel() @@ -1546,6 +1648,9 @@ func (a *adapter) TenantMessageSave(id t.TenantID, msg *t.Message) error { if tenantAdp != a { return tenantAdp.MessageSave(msg) } + if a.tenantScopedDB { + return a.MessageSave(msg) + } ctx, cancel := a.getContext() if cancel != nil { defer cancel() @@ -1729,6 +1834,10 @@ func (a *adapter) TenantMessageGetAll(id t.TenantID, topic string, uid t.Uid, op msgs, err := tenantAdp.MessageGetAll(topic, uid, opts) return stampTenantMessages(id, msgs), err } + if a.tenantScopedDB { + msgs, err := a.MessageGetAll(topic, uid, opts) + return stampTenantMessages(id, msgs), err + } limit := a.maxMessageResults args := []any{id, topic, id, decodeUID(uid)} seq := "" @@ -1856,6 +1965,95 @@ func (a *adapter) TenantMessageSearch(id t.TenantID, topic string, uid t.Uid, qu return msgs, nextCursor, nil } +func (a *adapter) TenantCallCreate(id t.TenantID, record *t.CallRecord) error { + if !a.tenantScopedDB { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil { + return err + } + if tenantAdp != a { + return tenantAdp.TenantCallCreate(id, record) + } + } + if err := validTenant(id); err != nil || record == nil || record.TenantID != id || record.CallID == "" || + record.Topic == "" || record.SeqID <= 0 || record.CallerID.IsZero() || record.CalleeID.IsZero() || + record.CallType == "" { + return t.ErrMalformed + } + now := t.TimeNow() + if record.StartedAt.IsZero() { + record.StartedAt = now + } + record.CreatedAt = now + record.UpdatedAt = now + if record.Status == "" { + record.Status = t.CallStatusStarted + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + _, err := a.db.ExecContext(ctx, + `INSERT INTO im_call_records + (call_id,topic,seq_id,caller_id,callee_id,call_type,status,end_reason,duration_ms,started_at,accepted_at,finished_at,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON DUPLICATE KEY UPDATE updated_at=VALUES(updated_at)`, + record.CallID, record.Topic, record.SeqID, decodeUID(record.CallerID), decodeUID(record.CalleeID), + record.CallType, record.Status, record.EndReason, record.DurationMS, record.StartedAt, record.AcceptedAt, + record.FinishedAt, record.CreatedAt, record.UpdatedAt) + return err +} + +func (a *adapter) TenantCallAccept(id t.TenantID, callID string, seq int, acceptedAt time.Time) error { + if !a.tenantScopedDB { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil { + return err + } + if tenantAdp != a { + return tenantAdp.TenantCallAccept(id, callID, seq, acceptedAt) + } + } + if err := validTenant(id); err != nil || callID == "" || seq <= 0 || acceptedAt.IsZero() { + return t.ErrMalformed + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + _, err := a.db.ExecContext(ctx, + `UPDATE im_call_records + SET status=?,accepted_at=?,updated_at=? + WHERE call_id=? AND seq_id=? AND status=?`, + t.CallStatusAccepted, acceptedAt, acceptedAt, callID, seq, t.CallStatusStarted) + return err +} + +func (a *adapter) TenantCallFinish(id t.TenantID, callID string, seq int, status string, durationMS int64, finishedAt time.Time) error { + if !a.tenantScopedDB { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil { + return err + } + if tenantAdp != a { + return tenantAdp.TenantCallFinish(id, callID, seq, status, durationMS, finishedAt) + } + } + if err := validTenant(id); err != nil || callID == "" || seq <= 0 || finishedAt.IsZero() || durationMS < 0 { + return t.ErrMalformed + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + _, err := a.db.ExecContext(ctx, + `UPDATE im_call_records + SET status=?,end_reason=?,duration_ms=?,finished_at=?,updated_at=? + WHERE call_id=? AND seq_id=?`, + status, status, durationMS, finishedAt, finishedAt, callID, seq) + return err +} + func tenantMessageGetOne(q sqlx.QueryerContext, ctx context.Context, tenantID t.TenantID, topic string, seq int, suffix string) (*t.Message, error) { row := struct { t.Message @@ -2071,6 +2269,9 @@ func (a *adapter) TenantMessageDeleteList(id t.TenantID, topic string, del *t.De if tenantAdp != a { return tenantAdp.MessageDeleteList(topic, del) } + if a.tenantScopedDB { + return a.MessageDeleteList(topic, del) + } ctx, cancel := a.getContextForTx() if cancel != nil { defer cancel() @@ -2102,6 +2303,13 @@ func (a *adapter) TenantMessageGetDeleted(id t.TenantID, topic string, uid t.Uid } return deleted, err } + if a.tenantScopedDB { + deleted, err := a.MessageGetDeleted(topic, uid, opts) + for i := range deleted { + deleted[i].TenantID = id + } + return deleted, err + } limit, lower, upper := a.maxResults, 0, 1<<31-1 if opts != nil { if opts.Since > 0 { @@ -2312,7 +2520,7 @@ func (a *adapter) TenantFileStartUpload(id t.TenantID, fd *t.FileDef) error { if cancel != nil { defer cancel() } - _, err := a.db.ExecContext(ctx, + _, err = a.db.ExecContext(ctx, "INSERT INTO fileuploads(tenant_id,id,createdat,updatedat,userid,status,mimetype,size,etag,location) VALUES(?,?,?,?,?,?,?,?,?,?)", id, decodeUID(fd.Uid()), fd.CreatedAt, fd.UpdatedAt, user, fd.Status, fd.MimeType, fd.Size, fd.ETag, fd.Location) return err @@ -2384,7 +2592,7 @@ func (a *adapter) TenantFileGet(id t.TenantID, fid string) (*t.FileDef, error) { defer cancel() } var fd t.FileDef - err := a.db.GetContext(ctx, &fd, + err = a.db.GetContext(ctx, &fd, "SELECT tenant_id,id,createdat,updatedat,userid AS user,status,mimetype,size,IFNULL(etag,'') AS etag,location "+ "FROM fileuploads WHERE tenant_id=? AND id=?", id, decodeUID(fileID)) if err == sql.ErrNoRows { @@ -2544,7 +2752,7 @@ func (a *adapter) TenantFileUserHasAccess(id t.TenantID, fid string, uid t.Uid) defer cancel() } var allowed int - err := a.db.GetContext(ctx, &allowed, ` + err = a.db.GetContext(ctx, &allowed, ` SELECT COUNT(*) FROM fileuploads AS fu WHERE fu.tenant_id=? AND fu.id=? AND fu.status=? AND ( fu.userid=? @@ -3638,7 +3846,12 @@ func (a *adapter) TenantFriendRequestAccept(id t.TenantID, requestID string, act return nil, err } topicRow := &t.Topic{TenantID: id, ObjHeader: t.ObjHeader{Id: topic, CreatedAt: handledAt, UpdatedAt: handledAt}, TouchedAt: handledAt} - if err = tenantTopicCreate(tx, topicRow); err != nil { + if a.tenantScopedDB { + err = a.topicCreate(tx, topicRow) + } else { + err = tenantTopicCreate(tx, topicRow) + } + if err != nil { if !isDupe(err) { return nil, err } diff --git a/server/store/store.go b/server/store/store.go index e454f4a..4b6d53d 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -1014,6 +1014,56 @@ func (messagesMapper) Edit(tenantID types.TenantID, topic string, seq int, actor return tenantAdp.TenantMessageEdit(tenantID, topic, seq, actor, expectedVersion, head, content, reason, editedAt) } +type CallsPersistenceInterface interface { + Create(tenantID types.TenantID, record *types.CallRecord) error + Accept(tenantID types.TenantID, callID string, seq int, acceptedAt time.Time) error + Finish(tenantID types.TenantID, callID string, seq int, status string, durationMS int64, finishedAt time.Time) error +} + +type callsMapper struct{} + +var Calls CallsPersistenceInterface + +func (callsMapper) Create(tenantID types.TenantID, record *types.CallRecord) error { + if record == nil || record.TenantID != tenantID || strings.TrimSpace(record.CallID) == "" || + record.Topic == "" || record.SeqID <= 0 || record.CallerID.IsZero() || record.CalleeID.IsZero() || + record.CallType == "" { + return types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return err + } + return tenantAdp.TenantCallCreate(tenantID, record) +} + +func (callsMapper) Accept(tenantID types.TenantID, callID string, seq int, acceptedAt time.Time) error { + if strings.TrimSpace(callID) == "" || seq <= 0 || acceptedAt.IsZero() { + return types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return err + } + return tenantAdp.TenantCallAccept(tenantID, callID, seq, acceptedAt) +} + +func (callsMapper) Finish(tenantID types.TenantID, callID string, seq int, status string, durationMS int64, finishedAt time.Time) error { + switch status { + case types.CallStatusFinished, types.CallStatusMissed, types.CallStatusDeclined, types.CallStatusDisconnected: + default: + return types.ErrMalformed + } + if strings.TrimSpace(callID) == "" || seq <= 0 || finishedAt.IsZero() || durationMS < 0 { + return types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return err + } + return tenantAdp.TenantCallFinish(tenantID, callID, seq, status, durationMS, finishedAt) +} + type RelationsPersistenceInterface interface { CreateFriendRequest(tenantID types.TenantID, req *types.FriendRequest) (*types.FriendRequest, error) ListFriendRequests(tenantID types.TenantID, uid types.Uid, direction, status, cursor string, limit int) ([]types.FriendRequestListItem, string, error) @@ -2181,6 +2231,7 @@ func init() { Topics = topicsMapper{} Subs = subsMapper{} Messages = messagesMapper{} + Calls = callsMapper{} Devices = deviceMapper{} Files = fileMapper{} PCache = pcacheMapper{} diff --git a/server/store/types/calls.go b/server/store/types/calls.go new file mode 100644 index 0000000..4cfc657 --- /dev/null +++ b/server/store/types/calls.go @@ -0,0 +1,30 @@ +package types + +import "time" + +const ( + CallStatusStarted = "started" + CallStatusAccepted = "accepted" + CallStatusFinished = "finished" + CallStatusMissed = "missed" + CallStatusDeclined = "declined" + CallStatusDisconnected = "disconnected" +) + +type CallRecord struct { + TenantID TenantID `db:"tenant_id" json:"-" bson:"tenant_id"` + CallID string `db:"call_id"` + Topic string + SeqID int `db:"seq_id"` + CallerID Uid `db:"caller_id"` + CalleeID Uid `db:"callee_id"` + CallType string `db:"call_type"` + Status string + EndReason string `db:"end_reason"` + DurationMS int64 `db:"duration_ms"` + StartedAt time.Time `db:"started_at"` + AcceptedAt *time.Time `db:"accepted_at"` + FinishedAt *time.Time `db:"finished_at"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` +} diff --git a/server/topic.go b/server/topic.go index 4874cda..ee033e4 100644 --- a/server/topic.go +++ b/server/topic.go @@ -1186,6 +1186,13 @@ func (t *Topic) handlePubBroadcast(msg *ClientComMessage) { msg.sess.queueOut(ErrCallBusyReply(msg, types.TimeNow())) return } + head, content, err := normalizeCallInvite(msg.Pub.Head, msg.Pub.Content) + if err != nil { + msg.sess.queueOut(ErrMalformedReply(msg, types.TimeNow())) + return + } + msg.Pub.Head = head + msg.Pub.Content = content } // Save to DB at master topic. diff --git a/server/topic_test.go b/server/topic_test.go index a2d6cc4..a355276 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -351,7 +351,7 @@ func TestHandleBroadcastCall(t *testing.T) { Pub: &MsgClientPub{ Topic: "p2p", Head: map[string]any{"webrtc": "started"}, - Content: "test", + Content: map[string]any{"callId": "call-1", "callType": "audio"}, NoEcho: true, }, sess: helper.sessions[0], @@ -383,11 +383,15 @@ func TestHandleBroadcastCall(t *testing.T) { if r.Data.Topic != from { t.Errorf("Response[0] topic: expected '%s', got '%s'", from, r.Data.Topic) } - if r.Data.Content.(string) != "test" { - t.Errorf("Response[0] content: expected 'test', got '%s'", r.Data.Content.(string)) + content := r.Data.Content.(map[string]any) + if content["callId"] != "call-1" || content["callType"] != "audio" || content["video"] != false { + t.Errorf("Response[0] content: unexpected call payload: %#v", content) } if r.Data.Head == nil || r.Data.Head["webrtc"].(string) != "started" { - t.Errorf("Response[0] head: expected {'webrtc': 'started'}', got '%s'", r.Data.Content.(string)) + t.Errorf("Response[0] head: expected {'webrtc': 'started'}', got %#v", r.Data.Head) + } + if r.Data.Head["x-im-type"] != "call" || r.Data.Head["mime"] != "application/json" { + t.Errorf("Response[0] head: expected normalized call head, got %#v", r.Data.Head) } if r.Data.From != from { t.Errorf("Response[0] from: expected '%s', got '%s'", from, r.Data.From) @@ -445,6 +449,37 @@ func TestHandleBroadcastCall(t *testing.T) { } } +func TestHandleBroadcastCallRequiresStructuredInvite(t *testing.T) { + numUsers := 2 + helper := TopicTestHelper{} + helper.setUp(t, numUsers, types.TopicCatP2P, "p2p-test" /*attach=*/, true) + globals.iceServers = []iceServer{{Username: "dummy"}} + defer func() { + globals.iceServers = nil + helper.tearDown() + }() + + from := helper.uids[0].UserId() + msg := &ClientComMessage{ + AsUser: from, + Original: from, + Pub: &MsgClientPub{ + Topic: "p2p", + Head: map[string]any{"webrtc": "started"}, + Content: "test", + NoEcho: true, + }, + sess: helper.sessions[0], + } + helper.topic.handleClientMsg(msg) + helper.finish() + + verifyResponseCodes(helper.results[0], []int{http.StatusBadRequest}, t) + if len(helper.results[1].messages) != 0 { + t.Fatalf("peer: expected 0 messages, got %d", len(helper.results[1].messages)) + } +} + func TestHandleBroadcastDataGroup(t *testing.T) { topicName := "grp-test" numUsers := 4