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

feat: add chapter eight call records

Browse Source
This commit is contained in:
richjohnrambo
2026-08-13 16:14:57 +08:00
parent 92d0066a02
commit dfcccbb761
9 changed files with 711 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/calls.go
+146 -1
View File
@@ -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 {
server/db/adapter.go
+4
View File
@@ -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)
server/db/mysql/adapter.go
+85 -3
View File
@@ -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()
server/db/mysql/tenant_adapter.go
+225 -12
View File
@@ -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
}
server/store/store.go
+51
View File
@@ -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{}
server/store/types/calls.go
+30
View File
@@ -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"`
}
server/topic.go
+7
View File
@@ -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.
server/topic_test.go
+39 -4
View File
@@ -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