diff --git a/server/calls.go b/server/calls.go index 4e3538c..1e8c6b6 100644 --- a/server/calls.go +++ b/server/calls.go @@ -298,6 +298,24 @@ func callContentString(content any, key string) string { return callPayloadString(content, key) } +func normalizeCallEvent(event string) string { + switch strings.ToLower(strings.TrimSpace(event)) { + case "hangup", "hang_up", "hang-up", "finish", "finished", "end", "ended", "disconnect", "disconnected": + return constCallEventHangUp + default: + return strings.TrimSpace(event) + } +} + +func isDetachedCallEvent(event string) bool { + switch normalizeCallEvent(event) { + case constCallEventRinging, constCallEventHangUp, constCallEventAccept: + return true + default: + return false + } +} + func (call *videoCall) matchesPayload(payload any) bool { payloadCallID := callPayloadString(payload, "callId") return payloadCallID == "" || payloadCallID == call.callID @@ -536,7 +554,8 @@ func (t *Topic) handleCallEvent(msg *ClientComMessage) { return } - switch call.Event { + event := normalizeCallEvent(call.Event) + switch event { case constCallEventRinging, constCallEventAccept: // Invariants: // 1. Call has been initiated but not been established yet. @@ -554,10 +573,10 @@ func (t *Topic) handleCallEvent(msg *ClientComMessage) { return } // Prepare a {info} message to forward to the call originator. - forwardMsg := t.currentCall.infoMessage(call.Event) + forwardMsg := t.currentCall.infoMessage(event) forwardMsg.Info.From = msg.AsUser forwardMsg.Info.Topic = t.original(originatorUid) - if call.Event == constCallEventAccept { + if event == constCallEventAccept { // The call has been accepted. // Send a replacement {data} message to the topic. msgCopy := *msg @@ -582,7 +601,7 @@ func (t *Topic) handleCallEvent(msg *ClientComMessage) { 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) + t.infoCallSubsOffline(msg.AsUser, asUid, event, t.currentCall.seq, call.Payload, msg.sess.sid, false) t.callEstablishmentTimer.Stop() } originator.queueOut(forwardMsg) diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index fc6c3a4..46a8658 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -1983,8 +1983,12 @@ func (a *adapter) AuthAddRecord(uid t.Uid, scheme, unique string, authLvl auth.L } else if unique != storedUsername { return t.ErrMalformed } - if _, err := a.db.ExecContext(ctx, "INSERT INTO im_auth(userid,authLvl,secret,expires) VALUES(?,?,?,?)", - store.DecodeUid(uid), authLvl, secret, exp); err != nil { + authID, err := nextSnowflakeID() + if err != nil { + return err + } + if _, err := a.db.ExecContext(ctx, "INSERT INTO im_auth(id,userid,authLvl,secret,expires) VALUES(?,?,?,?,?)", + authID, store.DecodeUid(uid), authLvl, secret, exp); err != nil { if isDupe(err) { return t.ErrDuplicate } @@ -2637,10 +2641,14 @@ func createSubscription(tx *sqlx.Tx, sub *t.Subscription, undelete bool) error { return err } + subID, err := nextSnowflakeID() + if err != nil { + return err + } _, err = tx.Exec( - "INSERT INTO im_subscriptions(createdat,updatedat,deletedat,userid,topic,modeWant,modeGiven,private) "+ - "VALUES(?,?,NULL,?,?,?,?,?)", - sub.CreatedAt, sub.UpdatedAt, decoded_uid, sub.Topic, sub.ModeWant.String(), sub.ModeGiven.String(), jpriv) + "INSERT INTO im_subscriptions(id,createdat,updatedat,deletedat,userid,topic,modeWant,modeGiven,private) "+ + "VALUES(?,?,?,NULL,?,?,?,?,?)", + subID, sub.CreatedAt, sub.UpdatedAt, decoded_uid, sub.Topic, sub.ModeWant.String(), sub.ModeGiven.String(), jpriv) if err != nil && isDupe(err) { if undelete { @@ -4262,9 +4270,13 @@ func (a *adapter) CredUpsert(cred *t.Credential) (bool, error) { } } // Add new record. - _, err = tx.Exec("INSERT INTO im_credentials(createdat,updatedat,method,value,synthetic,userid,resp,done) "+ - "VALUES(?,?,?,?,?,?,?,?)", - cred.CreatedAt, cred.UpdatedAt, cred.Method, cred.Value, synth, userId, cred.Resp, cred.Done) + credID, err := nextSnowflakeID() + if err != nil { + return false, err + } + _, err = tx.Exec("INSERT INTO im_credentials(id,createdat,updatedat,method,value,synthetic,userid,resp,done) "+ + "VALUES(?,?,?,?,?,?,?,?,?)", + credID, cred.CreatedAt, cred.UpdatedAt, cred.Method, cred.Value, synth, userId, cred.Resp, cred.Done) if err != nil { if isDupe(err) { return true, t.ErrDuplicate diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index 945df5e..8a377eb 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -2725,6 +2725,9 @@ func (a *adapter) TenantCallList(id t.TenantID, uid t.Uid, topic, cursor string, if cancel != nil { defer cancel() } + if err := a.finalizeStaleCalls(ctx); err != nil { + return nil, "", err + } rows, err := a.db.QueryxContext(ctx, `SELECT id,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 @@ -2786,6 +2789,23 @@ func (a *adapter) TenantCallList(id t.TenantID, uid t.Uid, topic, cursor string, return records, nextCursor, nil } +func (a *adapter) finalizeStaleCalls(ctx context.Context) error { + now := t.TimeNow() + startedBefore := now.Add(-2 * time.Minute) + acceptedBefore := now.Add(-12 * time.Hour) + if _, err := a.db.ExecContext(ctx, `UPDATE im_call_records + SET status=?,end_reason=?,duration_ms=0,finished_at=?,updated_at=? + WHERE status=? AND started_at'dissolved'`, decodeUID(actor), now, now, strings.TrimSpace(topic)) if err != nil { return nil, err @@ -207,6 +212,21 @@ func (a *adapter) TenantGroupDissolve(id t.TenantID, topic string, actor t.Uid, if affected, _ := res.RowsAffected(); affected == 0 { return nil, t.ErrNotFound } + if _, err = tx.ExecContext(ctx, `UPDATE im_subscriptions + SET updatedat=?,deletedat=? + WHERE topic IN (?,?) AND deletedat IS NULL`, now, now, strings.TrimSpace(topic), t.GrpToChn(strings.TrimSpace(topic))); err != nil { + return nil, err + } + if _, err = tx.ExecContext(ctx, `UPDATE im_group_members + SET member_status=?,left_at=?,removed_by=?,version=version+1,updated_at=? + WHERE topic=? AND member_status=?`, t.GroupMemberStatusLeft, now, decodeUID(actor), now, strings.TrimSpace(topic), t.GroupMemberStatusActive); err != nil { + return nil, err + } + if _, err = tx.ExecContext(ctx, `UPDATE im_topics + SET updatedat=?,touchedat=?,state=?,stateat=?,subcnt=0 + WHERE name=?`, now, now, t.StateDeleted, now, strings.TrimSpace(topic)); err != nil { + return nil, err + } if err = tx.Commit(); err != nil { return nil, err } @@ -573,7 +593,7 @@ func (a *adapter) TenantGroupInvitationAccept(id t.TenantID, invitationID string if err = upsertGroupMemberTx(tx, member); err != nil { return nil, err } - if err = recountGroupMembersTx(tx, inv.Topic); err != nil { + if err = recountGroupMembersAndBumpVersionTx(tx, inv.Topic); err != nil { return nil, err } if err = updateGroupTopicSubCountTx(tx, inv.Topic); err != nil { @@ -757,11 +777,13 @@ func (a *adapter) TenantGroupJoinRequestResolve(id t.TenantID, topic, requestID WHERE topic=? AND request_id=? AND status=?`, status, rejectReason, decodeUID(actor), now, now, topic, requestID, t.GroupJoinRequestStatusPending); err != nil { return nil, err } - if err = recountGroupMembersTx(tx, topic); err != nil { - return nil, err - } - if err = updateGroupTopicSubCountTx(tx, topic); err != nil { - return nil, err + if action == "accept" { + if err = recountGroupMembersAndBumpVersionTx(tx, topic); err != nil { + return nil, err + } + if err = updateGroupTopicSubCountTx(tx, topic); err != nil { + return nil, err + } } if err = tx.Commit(); err != nil { return nil, err @@ -1284,7 +1306,7 @@ func (a *adapter) TenantGroupInviteMember(id t.TenantID, topic string, inviter, if err = upsertGroupMemberTx(tx, member); err != nil { return false, err } - if err = recountGroupMembersTx(tx, topic); err != nil { + if err = recountGroupMembersAndBumpVersionTx(tx, topic); err != nil { return false, err } if err = updateGroupTopicSubCountTx(tx, topic); err != nil { @@ -1322,7 +1344,7 @@ func (a *adapter) TenantGroupLeave(id t.TenantID, topic string, uid t.Uid, now t WHERE topic=? AND user_id=? AND member_status=?`, t.GroupMemberStatusLeft, now, now, topic, decodeUID(uid), t.GroupMemberStatusActive); err != nil { return err } - if err = recountGroupMembersTx(tx, topic); err != nil { + if err = recountGroupMembersAndBumpVersionTx(tx, topic); err != nil { return err } if err = updateGroupTopicSubCountTx(tx, topic); err != nil { @@ -1485,12 +1507,33 @@ func recountGroupMembersTx(tx *sqlx.Tx, topic string) error { return err } +func recountGroupMembersAndBumpVersionTx(tx *sqlx.Tx, topic string) error { + _, err := tx.Exec(`UPDATE im_group_settings SET member_count=(SELECT COUNT(*) FROM im_group_members + WHERE topic=? AND member_status=?),version=version+1,updated_at=CURRENT_TIMESTAMP(3) WHERE topic=?`, topic, t.GroupMemberStatusActive, topic) + return err +} + func updateGroupTopicSubCountTx(tx *sqlx.Tx, topic string) error { _, err := tx.Exec(`UPDATE im_topics SET subcnt=(SELECT COUNT(*) FROM im_subscriptions WHERE topic IN (?,?) AND deletedat IS NULL) WHERE name=?`, topic, t.GrpToChn(topic), topic) return err } +func syncGroupMuteAllTx(tx *sqlx.Tx, topic string, muteAll bool, now time.Time) error { + mode := t.ModeCPublic + if muteAll { + mode = t.ModeCPublic &^ t.ModeWrite + } + _, err := tx.Exec(`UPDATE im_subscriptions AS s + JOIN im_group_members AS gm ON gm.topic=s.topic AND gm.user_id=s.userid + SET s.modewant=?,s.modegiven=?,s.updatedat=?, + gm.version=gm.version+1,gm.updated_at=? + WHERE s.topic=? AND s.deletedat IS NULL + AND gm.member_status=? AND gm.role=?`, + mode.String(), mode.String(), now, now, topic, t.GroupMemberStatusActive, t.GroupMemberRoleMember) + return err +} + func (a *adapter) routeGroupTenant(id t.TenantID) (*adapter, error) { if !a.tenantScopedDB { tenantAdp, err := a.routedTenantAdapter(id) diff --git a/server/im_http.go b/server/im_http.go index 5fc9afc..9c88209 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -6210,6 +6210,14 @@ func serveIMGroupOwnerTransfer(wrt http.ResponseWriter, req *http.Request, topic } if err = store.GroupBusiness.TransferOwner(principal.TenantID, topic, principal.Uid, target, body.Version, time.Now().UTC().Round(time.Millisecond)); err != nil { + if errors.Is(err, types.ErrInvalidResponse) { + currentVersion := 0 + if settings, settingsErr := store.GroupBusiness.GetSettings(principal.TenantID, topic); settingsErr == nil && settings != nil { + currentVersion = settings.Version + } + writeIMError(wrt, req, http.StatusConflict, 30006, "VERSION_CONFLICT", "version conflict", map[string]any{"current_version": currentVersion}) + return + } writeIMRelationError(wrt, req, err) return } @@ -7996,9 +8004,8 @@ func mergeGroupSettingsPayload(payload map[string]any, settings *types.GroupSett payload["description"] = settings.Description payload["owner"] = settings.OwnerUserID.UserId() payload["member_count"] = settings.MemberCount - if _, exists := payload["version"]; !exists { - payload["version"] = settings.Version - } + payload["status"] = settings.GroupStatus + payload["version"] = settings.Version return payload } diff --git a/server/im_http_test.go b/server/im_http_test.go index b91fea5..9b12084 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -5843,11 +5843,11 @@ func TestNormalizeIMUserVisibleTopicSelf(t *testing.T) { } } -func TestMergeGroupSettingsPayloadPreservesTopicVersion(t *testing.T) { +func TestMergeGroupSettingsPayloadUsesSettingsVersion(t *testing.T) { payload := map[string]any{"version": 7} mergeGroupSettingsPayload(payload, &types.GroupSettings{Version: 2}) - if payload["version"] != 7 { - t.Fatalf("topic version was overwritten: %#v", payload) + if payload["version"] != 2 { + t.Fatalf("settings version was not used: %#v", payload) } } diff --git a/server/session.go b/server/session.go index 029c76f..3737a39 100644 --- a/server/session.go +++ b/server/session.go @@ -1602,7 +1602,7 @@ func (s *Session) note(msg *ClientComMessage) { s.queueOut(ErrServiceUnavailableReply(msg, msg.Timestamp)) logs.Err.Println("s.note: sub.broacast channel full, topic ", msg.RcptTo, s.sid) } - } else if msg.Note.What == "recv" || (msg.Note.What == "call" && (msg.Note.Event == "ringing" || msg.Note.Event == "hang-up" || msg.Note.Event == "accept")) { + } else if msg.Note.What == "recv" || (msg.Note.What == "call" && isDetachedCallEvent(msg.Note.Event)) { // One of the following events happened: // 1. Client received a pres notification about a new message, initiated a fetch // from the server (and detached from the topic) and acknowledges receipt. diff --git a/server/store/store.go b/server/store/store.go index 2947c8f..8eec489 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -1716,6 +1716,11 @@ func (groupBusinessMapper) UpdateSettings(tenantID types.TenantID, topic string, } else { _ = DeleteCachedObject(tenantID, cacheNamespaceGroupSettings, strings.TrimSpace(topic)) } + if patch.MuteAll != nil { + invalidateGroupMemberPages(tenantID, topic) + invalidateGroupMemberObjects(tenantID, topic) + invalidateTopicSubscriptions(tenantID, topic) + } } return settings, err } @@ -1735,6 +1740,9 @@ func (groupBusinessMapper) Dissolve(tenantID types.TenantID, topic string, actor } else { _ = DeleteCachedObject(tenantID, cacheNamespaceGroupSettings, strings.TrimSpace(topic)) } + invalidateGroupMemberPages(tenantID, topic) + invalidateGroupMemberObjects(tenantID, topic) + invalidateTopicSubscriptions(tenantID, topic) } return settings, err } @@ -1825,6 +1833,31 @@ func invalidateGroupMemberPages(tenantID types.TenantID, topic string) { } } +func invalidateGroupMemberObjects(tenantID types.TenantID, topic string) { + topic = strings.TrimSpace(topic) + if topic == "" { + return + } + if err := DeleteCachedObjectsByPrefix(tenantID, cacheNamespaceGroupMember, topic+":"); err != nil { + logs.Warn.Printf("cache group members[%s]: delete failed: %v", topic, err) + } +} + +func invalidateTopicSubscriptions(tenantID types.TenantID, topic string) { + topic = strings.TrimSpace(topic) + if topic == "" { + return + } + if err := DeleteCachedObjectsByPrefix(tenantID, cacheNamespaceSub, topic+":"); err != nil { + logs.Warn.Printf("cache subscriptions[%s]: delete failed: %v", topic, err) + } + if channel := types.GrpToChn(topic); channel != topic { + if err := DeleteCachedObjectsByPrefix(tenantID, cacheNamespaceSub, channel+":"); err != nil { + logs.Warn.Printf("cache subscriptions[%s]: delete failed: %v", channel, err) + } + } +} + func (groupBusinessMapper) UpdateMember(tenantID types.TenantID, topic string, uid types.Uid, patch types.GroupMemberPatch) (*types.GroupMember, error) { if strings.TrimSpace(topic) == "" || uid.IsZero() { return nil, types.ErrMalformed diff --git a/server/topic_test.go b/server/topic_test.go index 2870ab4..af7e2d0 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -613,6 +613,26 @@ func TestVideoCallStaleUnanswered(t *testing.T) { } } +func TestNormalizeCallEventAliases(t *testing.T) { + for _, event := range []string{"hang-up", "hangup", "hang_up", "finish", "finished", "end", "ended", "disconnect", "disconnected"} { + if got := normalizeCallEvent(event); got != constCallEventHangUp { + t.Fatalf("normalizeCallEvent(%q) = %q, want %q", event, got, constCallEventHangUp) + } + if !isDetachedCallEvent(event) { + t.Fatalf("isDetachedCallEvent(%q) = false, want true", event) + } + } + if got := normalizeCallEvent(constCallEventAccept); got != constCallEventAccept { + t.Fatalf("normalizeCallEvent(%q) = %q, want %q", constCallEventAccept, got, constCallEventAccept) + } + if !isDetachedCallEvent(constCallEventAccept) { + t.Fatalf("isDetachedCallEvent(%q) = false, want true", constCallEventAccept) + } + if isDetachedCallEvent("unknown") { + t.Fatal("isDetachedCallEvent(\"unknown\") = true, want false") + } +} + func TestHandleBroadcastDataGroup(t *testing.T) { topicName := "grp-test" numUsers := 4