fix: resolve im backend issue list regressions

This commit is contained in:
richjohnrambo
2026-09-17 14:59:16 +08:00
parent 760de35e10
commit 50f8581667
13 changed files with 750 additions and 181 deletions
+2 -2
View File
@@ -4,6 +4,7 @@ go 1.26.0
require (
firebase.google.com/go v3.13.0+incompatible
github.com/AgoraIO/Tools/DynamicKey/AgoraDynamicKey/go/src v0.0.0-20260726132416-da06bfb83556
github.com/alibabacloud-go/darabonba-openapi/v2 v2.2.4
github.com/alibabacloud-go/green-20220302/v3 v3.5.4
github.com/alibabacloud-go/tea v1.5.2
@@ -27,6 +28,7 @@ require (
github.com/prometheus/common v0.69.0
github.com/redis/go-redis/v9 v9.12.1
github.com/rivo/uniseg v0.4.7
github.com/segmentio/kafka-go v0.4.49
github.com/tinode/jsonco v1.0.0
github.com/tinode/snowflake v1.0.0
go.mongodb.org/mongo-driver v1.17.9
@@ -45,7 +47,6 @@ require (
cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
cloud.google.com/go/monitoring v1.29.0 // indirect
filippo.io/edwards25519 v1.2.0 // indirect
github.com/AgoraIO/Tools/DynamicKey/AgoraDynamicKey/go/src v0.0.0-20260726132416-da06bfb83556 // indirect
github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.33.0 // indirect
github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.57.0 // indirect
github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.57.0 // indirect
@@ -83,7 +84,6 @@ require (
github.com/pierrec/lz4/v4 v4.1.15 // indirect
github.com/pkg/errors v0.9.1 // indirect
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
github.com/segmentio/kafka-go v0.4.49 // indirect
github.com/spiffe/go-spiffe/v2 v2.8.1 // indirect
github.com/stretchr/objx v0.5.0 // indirect
github.com/tjfoc/gmsm v1.4.1 // indirect
+73 -17
View File
@@ -1359,6 +1359,7 @@ func createIMUserProductTables(tx *sql.Tx) error {
profile_visibility VARCHAR(32) NOT NULL DEFAULT 'friends',
last_seen_visibility VARCHAR(32) NOT NULL DEFAULT 'friends',
read_receipts TINYINT NOT NULL DEFAULT 1,
friend_request_verification TINYINT NOT NULL DEFAULT 1,
group_invite_verification TINYINT NOT NULL DEFAULT 0,
version INT NOT NULL DEFAULT 1,
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
@@ -4762,31 +4763,44 @@ func (a *adapter) FileDeleteUnused(olderThan time.Time, limit int) ([]string, er
if err != nil {
return nil, err
}
defer rows.Close()
var locations []string
var ids []any
type gcCandidate struct {
id int
location string
}
var candidates []gcCandidate
for rows.Next() {
var id int
var loc string
if err = rows.Scan(&id, &loc); err != nil {
break
}
candidates = append(candidates, gcCandidate{id: id, location: loc})
}
if err == nil {
err = rows.Err()
}
rows.Close()
if err != nil {
return nil, err
}
var locations []string
var ids []any
for _, candidate := range candidates {
var noteRefs int
fileID := store.EncodeUid(int64(id)).String()
fileID := store.EncodeUid(int64(candidate.id)).String()
if err = tx.QueryRow(`SELECT COUNT(*) FROM im_meeting_note_files WHERE file_id=? AND status=? LIMIT 1`, fileID, t.MeetingNoteFileStatusActive).Scan(&noteRefs); err != nil {
break
}
if noteRefs > 0 {
continue
}
if loc != "" {
locations = append(locations, loc)
if candidate.location != "" {
locations = append(locations, candidate.location)
}
ids = append(ids, id)
}
if err == nil {
err = rows.Err()
ids = append(ids, candidate.id)
}
if err != nil {
@@ -4900,7 +4914,24 @@ func (a *adapter) FileUserHasAccess(fid string, uid t.Uid) (bool, error) {
defer cancel()
}
var allowed int
err := a.db.GetContext(ctx, &allowed, `
buildQuery := func(includeRecorder bool) (string, []any) {
recorderClause := ""
if includeRecorder {
recorderClause = `
OR mtg.recorder_user_id=?`
}
args := []any{
store.DecodeUid(fileID), fileID.String32(), "%/" + fileID.String32(), t.UploadCompleted,
store.DecodeUid(uid), store.DecodeUid(uid), store.DecodeUid(uid), t.RelationshipFriend,
store.DecodeUid(uid), store.DecodeUid(uid), t.GroupMemberStatusActive, store.DecodeUid(uid), store.DecodeUid(uid),
store.DecodeUid(uid), fid, fid,
fid, t.MeetingStateCanceled, store.DecodeUid(uid),
}
if includeRecorder {
args = append(args, store.DecodeUid(uid))
}
args = append(args, store.DecodeUid(uid))
return `
SELECT COUNT(*) FROM im_fileuploads AS fu
WHERE (fu.id=? OR fu.location=? OR fu.location LIKE ?) AND fu.status=? AND (
fu.userid=?
@@ -4958,18 +4989,29 @@ func (a *adapter) FileUserHasAccess(fid string, uid t.Uid) (bool, error) {
AND mtg.end_at<=CURRENT_TIMESTAMP(3)
AND (
mtg.organizer_id=?
OR mtg.recorder_user_id=?
` + recorderClause + `
OR EXISTS (
SELECT 1 FROM im_meeting_participants AS mp
WHERE mp.meeting_id=mtg.meeting_id AND mp.user_id=?
)
)
)
) LIMIT 1`,
store.DecodeUid(fileID), fileID.String32(), "%/"+fileID.String32(), t.UploadCompleted, store.DecodeUid(uid), store.DecodeUid(uid), store.DecodeUid(uid), t.RelationshipFriend,
store.DecodeUid(uid), store.DecodeUid(uid), t.GroupMemberStatusActive, store.DecodeUid(uid), store.DecodeUid(uid),
store.DecodeUid(uid), fid, fid,
fid, t.MeetingStateCanceled, store.DecodeUid(uid), store.DecodeUid(uid), store.DecodeUid(uid))
) LIMIT 1`, args
}
includeRecorder := true
run := func() error {
query, args := buildQuery(includeRecorder)
return a.db.GetContext(ctx, &allowed, query, args...)
}
err := run()
if isUnknownColumn(err) && strings.Contains(strings.ToLower(err.Error()), "recorder_user_id") {
includeRecorder = false
err = run()
}
for attempt := 0; isRetryableMySQLReadErr(err) && attempt < 2; attempt++ {
time.Sleep(time.Duration(attempt+1) * 25 * time.Millisecond)
err = run()
}
if err != nil {
return false, err
}
@@ -5108,6 +5150,20 @@ func isIDOutOfRange(err error) bool {
return ok && myerr.Number == 1264 && strings.Contains(strings.ToLower(myerr.Message), "id")
}
func isRetryableMySQLReadErr(err error) bool {
if err == nil {
return false
}
var myerr *ms.MySQLError
if !errors.As(err, &myerr) {
return false
}
if myerr.Number == 1213 || myerr.Number == 1205 {
return true
}
return strings.EqualFold(string(myerr.SQLState[:]), "40001")
}
func isPXCStrictModeNamedLockError(err error) bool {
if err == nil {
return false
@@ -63,18 +63,6 @@ func ensureIMAdminGovernanceTables(exec sqlExecutor) error {
PRIMARY KEY(id), UNIQUE INDEX uk_im_user_ip_whitelist(user_id,ip_cidr),
INDEX idx_im_user_ip_whitelist_user(user_id,status), INDEX idx_im_user_ip_whitelist_deleted(deleted_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`CREATE TABLE IF NOT EXISTS im_content_moderation_logs (
id BIGINT NOT NULL AUTO_INCREMENT, request_id VARCHAR(64) NOT NULL,
scene VARCHAR(32) NOT NULL, topic VARCHAR(96) NOT NULL DEFAULT '', seqid BIGINT,
actor_id BIGINT, provider VARCHAR(32) NOT NULL DEFAULT '', provider_request_id VARCHAR(96) NOT NULL DEFAULT '',
risk_level VARCHAR(16) NOT NULL DEFAULT 'none', risk_labels JSON, action VARCHAR(16) NOT NULL,
content_hash CHAR(64) NOT NULL, latency_ms BIGINT NOT NULL DEFAULT 0, error_code VARCHAR(64) NOT NULL DEFAULT '',
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), PRIMARY KEY(id),
UNIQUE INDEX uk_im_content_moderation_request(request_id),
INDEX idx_im_content_moderation_scene_time(scene,created_at DESC,id DESC),
INDEX idx_im_content_moderation_actor_time(actor_id,created_at DESC,id DESC),
INDEX idx_im_content_moderation_topic_seq(topic,seqid)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
}
for _, statement := range statements {
if _, err := exec.Exec(statement); err != nil {
+26 -20
View File
@@ -6609,6 +6609,10 @@ func (a *adapter) TenantUserPrivacyUpdate(id t.TenantID, uid t.Uid, patch t.User
sets = append(sets, "read_receipts=?")
args = append(args, *patch.ReadReceipts)
}
if patch.FriendRequestVerification != nil {
sets = append(sets, "friend_request_verification=?")
args = append(args, *patch.FriendRequestVerification)
}
if patch.GroupInviteVerification != nil {
sets = append(sets, "group_invite_verification=?")
args = append(args, *patch.GroupInviteVerification)
@@ -7180,30 +7184,32 @@ func (a *adapter) userProfileByUID(ctx context.Context, uid t.Uid) (*t.UserProfi
func (a *adapter) userPrivacyByUID(ctx context.Context, uid t.Uid) (*t.UserPrivacySettings, error) {
row := struct {
UserID int64 `db:"user_id"`
AddMeMethods []byte `db:"add_me_methods"`
ProfileVisibility string `db:"profile_visibility"`
LastSeenVisibility string `db:"last_seen_visibility"`
ReadReceipts bool `db:"read_receipts"`
GroupInviteVerification bool `db:"group_invite_verification"`
Version int
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
UserID int64 `db:"user_id"`
AddMeMethods []byte `db:"add_me_methods"`
ProfileVisibility string `db:"profile_visibility"`
LastSeenVisibility string `db:"last_seen_visibility"`
ReadReceipts bool `db:"read_receipts"`
FriendRequestVerification bool `db:"friend_request_verification"`
GroupInviteVerification bool `db:"group_invite_verification"`
Version int
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}{}
if err := sqlx.GetContext(ctx, a.db, &row, `SELECT user_id,add_me_methods,profile_visibility,last_seen_visibility,
read_receipts,group_invite_verification,version,created_at,updated_at FROM im_user_privacy_settings WHERE user_id=?`, decodeUID(uid)); err != nil {
read_receipts,friend_request_verification,group_invite_verification,version,created_at,updated_at FROM im_user_privacy_settings WHERE user_id=?`, decodeUID(uid)); err != nil {
return nil, err
}
settings := &t.UserPrivacySettings{
UserID: store.EncodeUid(row.UserID),
AddMeMethods: map[string]bool{"account": true, "phone": true, "qr": true, "group": true},
ProfileVisibility: row.ProfileVisibility,
LastSeenVisibility: row.LastSeenVisibility,
ReadReceipts: row.ReadReceipts,
GroupInviteVerification: row.GroupInviteVerification,
Version: row.Version,
CreatedAt: row.CreatedAt,
UpdatedAt: row.UpdatedAt,
UserID: store.EncodeUid(row.UserID),
AddMeMethods: map[string]bool{"account": true, "phone": true, "qr": true, "group": true},
ProfileVisibility: row.ProfileVisibility,
LastSeenVisibility: row.LastSeenVisibility,
ReadReceipts: row.ReadReceipts,
FriendRequestVerification: row.FriendRequestVerification,
GroupInviteVerification: row.GroupInviteVerification,
Version: row.Version,
CreatedAt: row.CreatedAt,
UpdatedAt: row.UpdatedAt,
}
_ = json.Unmarshal(row.AddMeMethods, &settings.AddMeMethods)
return settings, nil
@@ -9314,7 +9320,7 @@ func (a *adapter) TenantSystemNotificationMarkRead(id t.TenantID, uid t.Uid, eve
}
res, err := a.db.ExecContext(ctx,
`UPDATE im_system_notifications
SET read_at=COALESCE(read_at,?),updated_at=?
SET read_at=?,updated_at=?
WHERE user_id=? AND event_id=?`,
readAt, readAt, decodeUID(uid), eventID)
if err != nil {
+14 -1
View File
@@ -4,7 +4,10 @@
package mysql
import (
"database/sql"
"encoding/json"
"errors"
"strings"
"time"
"github.com/jmoiron/sqlx"
@@ -60,7 +63,17 @@ func (a *adapter) TenantApplicationCreate(application *t.TenantApplication) (*t.
if cancel != nil {
defer cancel()
}
_, err := db.ExecContext(ctx, `INSERT INTO im_tenant_applications
var existingID string
err := db.QueryRowContext(ctx, `SELECT application_id FROM im_tenant_applications
WHERE credit_code=? AND status IN ('received','pending','reviewing') ORDER BY id DESC LIMIT 1`,
strings.ToUpper(strings.TrimSpace(application.CreditCode))).Scan(&existingID)
if err == nil && existingID != "" {
return nil, t.ErrDuplicate
}
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return nil, err
}
_, err = db.ExecContext(ctx, `INSERT INTO im_tenant_applications
(application_id,credit_code,enterprise_email,license_file_id,tenant_code_hint,client_context,idempotency_key,status,estimated_review_hour,submitted_at,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`,
application.ApplicationID, application.CreditCode, application.EnterpriseEmail, application.LicenseFileID,
+4 -2
View File
@@ -20,8 +20,10 @@ const (
groupSettingsSelectColumns = `topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all,
member_upload_files,member_private_chat,member_invite,max_member_count,max_admin_count,member_count,version,
created_by,dissolved_by,dissolved_at,created_at,updated_at`
groupAnnouncementSelectColumns = `id,announcement_id,topic,author_user_id,content,status,version,published_at,idempotency_key,created_at,updated_at`
groupMuteSelectColumns = `topic,user_id,muted_by,muted_at,expires_at,reason,status,revoked_by,revoked_at,version,created_at,updated_at`
groupAnnouncementSelectColumns = `id,announcement_id,topic,author_user_id,content,status,version,published_at,idempotency_key,created_at,updated_at,
(SELECT COUNT(*) FROM im_group_announcement_reads AS ar WHERE ar.topic=im_group_announcements.topic AND ar.announcement_id=im_group_announcements.announcement_id) AS read_count,
(SELECT COUNT(*) FROM im_group_members AS gm WHERE gm.topic=im_group_announcements.topic AND gm.member_status='active') AS member_count`
groupMuteSelectColumns = `topic,user_id,muted_by,muted_at,expires_at,reason,status,revoked_by,revoked_at,version,created_at,updated_at`
)
func (a *adapter) TenantGroupSettingsCreate(id t.TenantID, settings *t.GroupSettings) (*t.GroupSettings, error) {
+464 -88
View File
@@ -1486,6 +1486,25 @@ func serveIMAbuseReports(wrt http.ResponseWriter, req *http.Request) {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "description"})
return
}
reasons, err := store.Safety.AbuseReasons(principal.TenantID, targetType)
if err != nil || len(reasons) == 0 {
reasons = defaultIMAbuseReasons(targetType)
}
reasonKnown := false
for _, reason := range reasons {
if strings.EqualFold(strings.TrimSpace(reason.ReasonCode), strings.TrimSpace(body.ReasonCode)) {
reasonKnown = true
if reason.RequiresDescription && description == "" {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonValidationFailed, "validation failed", map[string]any{"field": "description", "reason_code": reason.ReasonCode})
return
}
break
}
}
if !reasonKnown {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "reason_code"})
return
}
if len(body.EvidenceFileIDs) > 9 {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "evidence_file_ids"})
return
@@ -2241,6 +2260,12 @@ func serveIMMeetings(wrt http.ResponseWriter, req *http.Request) {
if !ok {
return
}
for _, attendeeID := range attendeeIDs {
if !ensureIMMeetingUserExists(principal.TenantID, attendeeID) {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "attendee_user_ids", "user_id": attendeeID.UserId()})
return
}
}
recorderID := principal.Uid
if value := strings.TrimSpace(body.RecorderUserID); value != "" {
recorderID = types.ParseUserId(value)
@@ -2568,6 +2593,27 @@ func serveIMMeetingCancel(wrt http.ResponseWriter, req *http.Request, meetingID
if !ok {
return
}
var body struct {
Version int `json:"version"`
}
if req.Body != nil && req.ContentLength != 0 {
if !decodeIMJSONBody(wrt, req, &body) {
return
}
}
if body.Version <= 0 {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "version"})
return
}
current, err := store.Meetings.Get(principal.TenantID, principal.Uid, meetingID)
if err != nil {
writeIMRelationError(wrt, req, err)
return
}
if current.Version != body.Version {
writeIMError(wrt, req, http.StatusConflict, 30006, "VERSION_CONFLICT", "version conflict", map[string]any{"current_version": current.Version})
return
}
meeting, err := store.Meetings.Cancel(principal.TenantID, principal.Uid, meetingID, idempotencyKey)
if err != nil {
writeIMRelationError(wrt, req, err)
@@ -2663,6 +2709,9 @@ func serveIMDailyReports(wrt http.ResponseWriter, req *http.Request) {
switch req.Method {
case http.MethodGet:
limit := parseIMLimit(req.URL.Query().Get("limit"), 50, 100)
if !validIMReportStatusQuery(wrt, req) || !validIMOptionalDateQuery(wrt, req, "from") || !validIMOptionalDateQuery(wrt, req, "to") {
return
}
reports, nextCursor, err := store.DailyReports.List(principal.TenantID, principal.Uid,
req.URL.Query().Get("scope"),
req.URL.Query().Get("from"),
@@ -2677,12 +2726,7 @@ func serveIMDailyReports(wrt http.ResponseWriter, req *http.Request) {
writeIMSuccess(wrt, req, map[string]any{
"items": dailyReportListPayloads(reports),
"next_cursor": nextCursor,
"permissions": map[string]any{
"can_create": true,
"makeup_allowed_days": 7,
"can_view_managed": false,
"can_remind_unsubmitted": false,
},
"permissions": dailyReportPermissions(req),
})
case http.MethodPost:
var body struct {
@@ -2692,7 +2736,7 @@ func serveIMDailyReports(wrt http.ResponseWriter, req *http.Request) {
return
}
reportDate, err := parseIMDate(body.ReportDate)
if err != nil {
if err != nil || !validIMReportDateRange(reportDate) {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "report_date"})
return
}
@@ -2781,8 +2825,8 @@ func serveIMDailyReportSaveDraft(wrt http.ResponseWriter, req *http.Request, rep
writeIMRelationError(wrt, req, err)
return
}
publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "daily_report", "daily_report.submitted", report.ReportID,
[]types.Uid{principal.Uid}, "日报已提交", report.ReportDate.Format("2006-01-02"), map[string]any{"report_id": report.ReportID, "version": report.Version, "status": report.Status})
publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "daily_report", "daily_report.draft.saved", report.ReportID,
[]types.Uid{principal.Uid}, "日报", report.ReportDate.Format("2006-01-02"), map[string]any{"report_id": report.ReportID, "version": report.Version, "status": report.Status})
writeIMSuccess(wrt, req, dailyReportDetailPayload(report, principal.Uid))
}
@@ -2801,6 +2845,19 @@ func serveIMDailyReportSubmit(wrt http.ResponseWriter, req *http.Request, report
if !decodeIMJSONBody(wrt, req, &body) {
return
}
current, err := store.DailyReports.Get(principal.TenantID, principal.Uid, reportID)
if err != nil {
writeIMRelationError(wrt, req, err)
return
}
if !validIMReportDateRange(current.ReportDate) {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "report_date"})
return
}
if strings.TrimSpace(current.Content) == "" {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonValidationFailed, "validation failed", map[string]any{"field": "content"})
return
}
report, err := store.DailyReports.Submit(principal.TenantID, principal.Uid, reportID, body.Version)
if err != nil {
writeIMRelationError(wrt, req, err)
@@ -2887,6 +2944,9 @@ func serveIMMonthlyReports(wrt http.ResponseWriter, req *http.Request) {
switch req.Method {
case http.MethodGet:
limit := parseIMLimit(req.URL.Query().Get("limit"), 50, 100)
if !validIMReportStatusQuery(wrt, req) || !validIMOptionalMonthQuery(wrt, req, "from_month") || !validIMOptionalMonthQuery(wrt, req, "to_month") {
return
}
reports, nextCursor, err := store.MonthlyReports.List(principal.TenantID, principal.Uid,
req.URL.Query().Get("scope"),
req.URL.Query().Get("from_month"),
@@ -2906,7 +2966,7 @@ func serveIMMonthlyReports(wrt http.ResponseWriter, req *http.Request) {
if !decodeIMJSONBody(wrt, req, &body) {
return
}
if !validIMMonth(body.Month) {
if !validIMMonth(body.Month) || !validIMMonthRange(strings.TrimSpace(body.Month)) {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "month"})
return
}
@@ -2995,8 +3055,8 @@ func serveIMMonthlyReportSaveDraft(wrt http.ResponseWriter, req *http.Request, r
writeIMRelationError(wrt, req, err)
return
}
publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "monthly_report", "monthly_report.submitted", report.ReportID,
[]types.Uid{principal.Uid}, "月报已提交", report.Month, map[string]any{"report_id": report.ReportID, "version": report.Version, "status": report.Status})
publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "monthly_report", "monthly_report.draft.saved", report.ReportID,
[]types.Uid{principal.Uid}, "月报", report.Month, map[string]any{"report_id": report.ReportID, "version": report.Version, "status": report.Status})
writeIMSuccess(wrt, req, monthlyReportDetailPayload(report, principal.Uid))
}
@@ -3015,13 +3075,30 @@ func serveIMMonthlyReportSubmit(wrt http.ResponseWriter, req *http.Request, repo
if !decodeIMJSONBody(wrt, req, &body) {
return
}
current, err := store.MonthlyReports.Get(principal.TenantID, principal.Uid, reportID)
if err != nil {
writeIMRelationError(wrt, req, err)
return
}
if !validIMMonthRange(current.Month) {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "month"})
return
}
if strings.TrimSpace(current.WorkSummary) == "" {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonValidationFailed, "validation failed", map[string]any{"field": "work_summary"})
return
}
if strings.TrimSpace(current.NextPlan) == "" {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonValidationFailed, "validation failed", map[string]any{"field": "next_plan"})
return
}
report, err := store.MonthlyReports.Submit(principal.TenantID, principal.Uid, reportID, body.Version)
if err != nil {
writeIMRelationError(wrt, req, err)
return
}
publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "monthly_report", "monthly_report.reopened", report.ReportID,
[]types.Uid{principal.Uid}, "月报已重开", report.Month, map[string]any{"report_id": report.ReportID, "version": report.Version, "status": report.Status})
publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "monthly_report", "monthly_report.submitted", report.ReportID,
[]types.Uid{principal.Uid}, "月报已提交", report.Month, map[string]any{"report_id": report.ReportID, "version": report.Version, "status": report.Status})
writeIMSuccess(wrt, req, monthlyReportDetailPayload(report, principal.Uid))
}
@@ -3271,23 +3348,25 @@ func serveIMMePrivacy(wrt http.ResponseWriter, req *http.Request) {
writeIMSuccess(wrt, req, userPrivacyPayload(settings))
case http.MethodPatch:
var body struct {
AddMeMethods map[string]bool `json:"add_me_methods"`
ProfileVisibility *string `json:"profile_visibility"`
LastSeenVisibility *string `json:"last_seen_visibility"`
ReadReceipts *bool `json:"read_receipts"`
GroupInviteVerification *bool `json:"group_invite_verification"`
Version int `json:"version"`
AddMeMethods map[string]bool `json:"add_me_methods"`
ProfileVisibility *string `json:"profile_visibility"`
LastSeenVisibility *string `json:"last_seen_visibility"`
ReadReceipts *bool `json:"read_receipts"`
FriendRequestVerification *bool `json:"friend_request_verification"`
GroupInviteVerification *bool `json:"group_invite_verification"`
Version int `json:"version"`
}
if !decodeIMJSONBody(wrt, req, &body) {
return
}
settings, err := store.Profiles.UpdatePrivacy(principal.TenantID, principal.Uid, types.UserPrivacyPatch{
AddMeMethods: body.AddMeMethods,
ProfileVisibility: body.ProfileVisibility,
LastSeenVisibility: body.LastSeenVisibility,
ReadReceipts: body.ReadReceipts,
GroupInviteVerification: body.GroupInviteVerification,
Version: body.Version,
AddMeMethods: body.AddMeMethods,
ProfileVisibility: body.ProfileVisibility,
LastSeenVisibility: body.LastSeenVisibility,
ReadReceipts: body.ReadReceipts,
FriendRequestVerification: body.FriendRequestVerification,
GroupInviteVerification: body.GroupInviteVerification,
Version: body.Version,
})
if err != nil {
writeIMRelationError(wrt, req, err)
@@ -3611,7 +3690,7 @@ func serveIMSystemNotifications(wrt http.ResponseWriter, req *http.Request) {
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
return
}
writeIMSuccess(wrt, req, map[string]any{"items": items, "next_cursor": nextCursor})
writeIMSuccess(wrt, req, map[string]any{"items": systemNotificationPayloads(items), "next_cursor": nextCursor})
}
func serveIMSystemNotificationRead(wrt http.ResponseWriter, req *http.Request) {
@@ -3892,7 +3971,15 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) {
saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, msg, nil, true)
if err != nil {
if saved != nil && saved.Pending {
result := map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "pending", "seq": saved.SeqId, "reason": "MESSAGE_PERSIST_PENDING"}
result := forwardResultUnknownPayload(target, saved.SeqId, relationErrorReason(err))
result["favorite_id"] = favoriteID
mergeForwardCommentResult(result, commentResult)
results = append(results, result)
continue
}
if msg.SeqId > 0 && relationErrorReason(err) == imReasonInternal {
result := forwardResultUnknownPayload(target, msg.SeqId, relationErrorReason(err))
result["favorite_id"] = favoriteID
mergeForwardCommentResult(result, commentResult)
results = append(results, result)
continue
@@ -4002,16 +4089,21 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
if err != nil {
if saved != nil && saved.Pending {
forwarded.SeqId = saved.SeqId
item := map[string]any{
"sources": bundle["source_refs"],
"target_topic": target,
"client_mid": clientMid,
"seq": saved.SeqId,
"status": "pending",
"reason": "MESSAGE_PERSIST_PENDING",
"mode": "bundle",
"bundle_id": bundle["bundle_id"],
}
item := forwardResultUnknownPayload(target, saved.SeqId, relationErrorReason(err))
item["sources"] = bundle["source_refs"]
item["client_mid"] = clientMid
item["mode"] = "bundle"
item["bundle_id"] = bundle["bundle_id"]
mergeForwardCommentResult(item, commentResult)
items = append(items, item)
continue
}
if forwarded.SeqId > 0 && relationErrorReason(err) == imReasonInternal {
item := forwardResultUnknownPayload(target, forwarded.SeqId, relationErrorReason(err))
item["sources"] = bundle["source_refs"]
item["client_mid"] = clientMid
item["mode"] = "bundle"
item["bundle_id"] = bundle["bundle_id"]
mergeForwardCommentResult(item, commentResult)
items = append(items, item)
continue
@@ -4094,14 +4186,17 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
if err != nil {
if saved != nil && saved.Pending {
forwarded.SeqId = saved.SeqId
item := map[string]any{
"source": map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)},
"target_topic": target,
"client_mid": clientMid,
"seq": saved.SeqId,
"status": "pending",
"reason": "MESSAGE_PERSIST_PENDING",
}
item := forwardResultUnknownPayload(target, saved.SeqId, relationErrorReason(err))
item["source"] = map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)}
item["client_mid"] = clientMid
mergeForwardCommentResult(item, commentResult)
items = append(items, item)
continue
}
if forwarded.SeqId > 0 && relationErrorReason(err) == imReasonInternal {
item := forwardResultUnknownPayload(target, forwarded.SeqId, relationErrorReason(err))
item["source"] = map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)}
item["client_mid"] = clientMid
mergeForwardCommentResult(item, commentResult)
items = append(items, item)
continue
@@ -4664,6 +4759,10 @@ func serveIMConversationContext(wrt http.ResponseWriter, req *http.Request) {
return
}
topic, tail := conversationResourceFromPath(req.URL.Path)
if topic == "settings" && tail == "" {
serveIMConversationSettingsBatch(wrt, req, principal)
return
}
if tail == "settings" {
serveIMConversationSettings(wrt, req, principal, topic)
return
@@ -4697,6 +4796,48 @@ func serveIMConversationContext(wrt http.ResponseWriter, req *http.Request) {
writeIMSuccess(wrt, req, relationshipContextPayload(ctx))
}
func serveIMConversationSettingsBatch(wrt http.ResponseWriter, req *http.Request, principal *imPrincipal) {
var topics []string
switch req.Method {
case http.MethodGet:
for _, part := range strings.Split(req.URL.Query().Get("topics"), ",") {
if value := strings.TrimSpace(part); value != "" {
topics = append(topics, value)
}
}
case http.MethodPost:
var body struct {
Topics []string `json:"topics"`
}
if !decodeIMJSONBody(wrt, req, &body) {
return
}
topics = body.Topics
default:
writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
return
}
topics = cleanStringList(topics, 100)
if len(topics) == 0 {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "topics"})
return
}
items := make([]map[string]any, 0, len(topics))
for _, rawTopic := range topics {
topic := normalizeConversationSettingsTopic(rawTopic, principal.Uid)
if topic == "" {
continue
}
settings, err := store.Conversations.GetSettings(principal.TenantID, principal.Uid, topic)
if err != nil {
writeIMRelationError(wrt, req, err)
return
}
items = append(items, conversationSettingsPayload(settings))
}
writeIMSuccess(wrt, req, map[string]any{"items": items})
}
func serveIMConversationMessages(wrt http.ResponseWriter, req *http.Request, principal *imPrincipal, topic string) {
if topic == "" {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "topic"})
@@ -4920,9 +5061,31 @@ func serveIMFriendRequestCreate(wrt http.ResponseWriter, req *http.Request) {
writeIMRelationError(wrt, req, err)
return
}
if !privacy.FriendRequestVerification {
accepted, acceptErr := store.Relations.ResolveFriendRequest(principal.TenantID, created.RequestID, target, types.FriendRequestAccepted, now)
if acceptErr != nil {
writeIMRelationError(wrt, req, acceptErr)
return
}
created = accepted
}
if imReliableHTTPEventsEnabled(req) {
eventType := "friend.request.created"
title := "新的好友申请"
content := message
if created != nil && created.Status == types.FriendRequestAccepted {
eventType = "friend.request.accepted"
title = "已成为好友"
content = "你们已成为好友,现在可以开始聊天了。"
publishIMConversationSystemNotice(principal.TenantID, principal.Uid.P2PName(target),
"system_notice", "friend_added", content,
[]imSystemMessageActor{
{UserID: principal.Uid.UserId(), Click: map[string]any{"type": "user_profile", "target_id": principal.Uid.UserId()}},
{UserID: target.UserId(), Click: map[string]any{"type": "user_profile", "target_id": target.UserId()}},
}, nil, map[string]any{"request_id": created.RequestID})
}
if err = publishIMRelationEvent(principal.TenantID, principal.Uid, target, created.RequestID,
"friend.request.created", "新的好友申请", message); err != nil {
eventType, title, content); err != nil {
logs.Warn.Printf("im friend request create: relation event failed request_id=%s friend_request_id=%s target=%s: %v",
req.Header.Get("X-Request-ID"), created.RequestID, target.UserId(), err)
}
@@ -5642,11 +5805,13 @@ func serveIMGroupQRCodes(wrt http.ResponseWriter, req *http.Request, topic strin
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
return
}
groupPayload := groupDetailPayload(group, selfSub)
writeIMSuccess(wrt, req, map[string]any{
"qr_token": token,
"qr_payload": "im://qr?token=" + url.QueryEscape(token) + "&purpose=join_group",
"expires_at": expiresAt.Format(time.RFC3339Nano),
"group_snapshot": groupDetailPayload(group, selfSub),
"group": groupPayload,
"group_snapshot": groupPayload,
})
}
@@ -6569,6 +6734,10 @@ func serveIMGroupMutes(wrt http.ResponseWriter, req *http.Request, topic, tail s
if uid.IsZero() {
continue
}
if ok, reason := canMuteIMGroupMember(principal.TenantID, topic, selfSub, principal.Uid, uid); !ok {
writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", map[string]any{"user_id": uid.UserId(), "reason": reason})
return
}
var expiresAt *time.Time
if raw := strings.TrimSpace(member.ExpiresAt); raw != "" {
if parsed, err := time.Parse(time.RFC3339Nano, raw); err == nil {
@@ -6595,6 +6764,10 @@ func serveIMGroupMutes(wrt http.ResponseWriter, req *http.Request, topic, tail s
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "user_id"})
return
}
if ok, reason := canMuteIMGroupMember(principal.TenantID, topic, selfSub, principal.Uid, uid); !ok {
writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", map[string]any{"user_id": uid.UserId(), "reason": reason})
return
}
if err := store.GroupBusiness.DeleteMute(principal.TenantID, topic, uid, principal.Uid, time.Now().UTC().Round(time.Millisecond)); err != nil {
writeIMRelationError(wrt, req, err)
return
@@ -6750,12 +6923,14 @@ func serveIMQRCodesResolve(wrt http.ResponseWriter, req *http.Request) {
if selfSub != nil && (selfSub.ModeGiven & selfSub.ModeWant).IsJoiner() {
status = "joined"
}
groupPayload := groupDetailPayload(group, selfSub)
writeIMSuccess(wrt, req, map[string]any{
"purpose": claims.Purpose,
"expires_at": time.Unix(claims.Expires, 0).UTC().Format(time.RFC3339Nano),
"group": groupDetailPayload(group, selfSub),
"status": status,
"source": "qr",
"purpose": claims.Purpose,
"expires_at": time.Unix(claims.Expires, 0).UTC().Format(time.RFC3339Nano),
"group": groupPayload,
"group_snapshot": groupPayload,
"status": status,
"source": "qr",
})
return
}
@@ -7406,16 +7581,23 @@ func authenticateIMHTTPBearer(wrt http.ResponseWriter, req *http.Request) (*imPr
principal, err := resolveIMHTTPBearer(req)
if err != nil {
logs.Warn.Printf("im http bearer auth failed request_id=%s method=%s path=%s err=%v", req.Header.Get("X-Request-ID"), req.Method, req.URL.Path, err)
if errors.Is(err, types.ErrUnsupported) {
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
} else {
if imHTTPBearerAuthUnauthorized(err) {
writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "authentication required", nil)
} else {
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
}
return nil, false
}
return principal, true
}
func imHTTPBearerAuthUnauthorized(err error) bool {
return errors.Is(err, types.ErrPermissionDenied) ||
errors.Is(err, types.ErrMalformed) ||
errors.Is(err, types.ErrFailed) ||
errors.Is(err, types.ErrExpired)
}
func resolveIMHTTPBearer(req *http.Request) (*imPrincipal, error) {
principal, _, _, err := resolveIMHTTPBearerToken(req)
return principal, err
@@ -7437,6 +7619,9 @@ func resolveIMHTTPBearerToken(req *http.Request) (*imPrincipal, []byte, time.Dur
}
rec, _, err := handler.Authenticate(auth.AuthContext{}, secret)
if err != nil || rec == nil || rec.TenantID.IsZero() || rec.Uid.IsZero() {
if err != nil {
return nil, nil, 0, err
}
return nil, nil, 0, types.ErrPermissionDenied
}
return &imPrincipal{TenantID: rec.TenantID, Uid: rec.Uid}, secret, time.Duration(rec.Lifetime), nil
@@ -7790,6 +7975,21 @@ func parseIMDate(raw string) (time.Time, error) {
return parsed, nil
}
func validIMReportDateRange(date time.Time) bool {
year := date.UTC().Year()
nowYear := time.Now().UTC().Year()
return year >= nowYear-1 && year <= nowYear+1
}
func validIMReportStatusQuery(wrt http.ResponseWriter, req *http.Request) bool {
status := strings.TrimSpace(req.URL.Query().Get("status"))
if status == "" || status == types.DailyReportStatusDraft || status == types.DailyReportStatusSubmitted {
return true
}
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "status"})
return false
}
func parseIMOptionalDate(raw string, fallback time.Time) (time.Time, error) {
if strings.TrimSpace(raw) == "" {
return time.Date(fallback.UTC().Year(), fallback.UTC().Month(), fallback.UTC().Day(), 0, 0, 0, 0, time.UTC), nil
@@ -7802,7 +8002,8 @@ func validIMOptionalDateQuery(wrt http.ResponseWriter, req *http.Request, field
if value == "" {
return true
}
if _, err := parseIMDate(value); err != nil {
date, err := parseIMDate(value)
if err != nil || !validIMReportDateRange(date) {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": field})
return false
}
@@ -7816,6 +8017,26 @@ func validIMMonth(raw string) bool {
return true
}
func validIMMonthRange(month string) bool {
parsed, err := time.ParseInLocation("2006-01", strings.TrimSpace(month), time.UTC)
if err != nil {
return false
}
return validIMReportDateRange(parsed)
}
func validIMOptionalMonthQuery(wrt http.ResponseWriter, req *http.Request, field string) bool {
value := strings.TrimSpace(req.URL.Query().Get(field))
if value == "" {
return true
}
if !validIMMonth(value) || !validIMMonthRange(value) {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": field})
return false
}
return true
}
func requireIMIdempotencyKey(wrt http.ResponseWriter, req *http.Request) (string, bool) {
key := strings.TrimSpace(req.Header.Get("Idempotency-Key"))
if key == "" {
@@ -8035,7 +8256,6 @@ func mergeGroupListSettingsPayload(payload map[string]any, settings *types.Group
payload["topic"] = settings.Topic
payload["name"] = settings.Name
payload["avatar_file_id"] = settings.AvatarFileID
payload["member_count"] = settings.MemberCount
payload["updated_at"] = settings.UpdatedAt.UTC().Format(time.RFC3339Nano)
return payload
}
@@ -8190,6 +8410,37 @@ func groupMemberAccess(member types.GroupMember) types.AccessMode {
}
}
func canMuteIMGroupMember(tenantID types.TenantID, topic string, selfSub *types.Subscription, actor, target types.Uid) (bool, string) {
if actor.IsZero() || target.IsZero() {
return false, "invalid_user"
}
if actor == target {
return false, "self"
}
actorAccess := types.ModeNone
if selfSub != nil {
actorAccess = selfSub.ModeGiven & selfSub.ModeWant
}
if !actorAccess.IsAdmin() {
return false, "not_admin"
}
targetRole := ""
if member, err := store.GroupBusiness.GetMember(tenantID, topic, target); err == nil && member != nil {
targetRole = groupMemberRole(groupMemberAccess(*member))
} else if sub, err := store.Subs.Get(tenantID, topic, target, false); err == nil && sub != nil {
targetRole = groupMemberRole(sub.ModeGiven & sub.ModeWant)
}
switch targetRole {
case "owner":
return false, "owner"
case "admin":
if !actorAccess.IsOwner() {
return false, "admin"
}
}
return true, ""
}
func groupMemberProfilePayload(topic string, sub *types.Subscription, user *types.User, ctx *types.RelationshipContext) map[string]any {
data := groupMemberPayload(*sub, true, ctx.Remark)
data["topic"] = topic
@@ -8316,7 +8567,6 @@ func mergeGroupSettingsPayload(payload map[string]any, settings *types.GroupSett
payload["avatar_file_id"] = settings.AvatarFileID
payload["description"] = settings.Description
payload["owner"] = settings.OwnerUserID.UserId()
payload["member_count"] = settings.MemberCount
payload["status"] = settings.GroupStatus
payload["version"] = settings.Version
return payload
@@ -8447,6 +8697,9 @@ func groupAnnouncementPayload(ann *types.GroupAnnouncement) map[string]any {
"author": "",
"version": 0,
"published_at": "",
"read_count": 0,
"member_count": 0,
"read_summary": map[string]any{"read_count": 0, "member_count": 0},
}
}
return map[string]any{
@@ -8455,6 +8708,9 @@ func groupAnnouncementPayload(ann *types.GroupAnnouncement) map[string]any {
"author": ann.AuthorUserID.UserId(),
"version": ann.Version,
"published_at": ann.PublishedAt.UTC().Format(time.RFC3339Nano),
"read_count": ann.ReadCount,
"member_count": ann.MemberCount,
"read_summary": map[string]any{"read_count": ann.ReadCount, "member_count": ann.MemberCount},
}
}
@@ -8974,21 +9230,36 @@ func callRecordPayload(record types.CallRecord, viewer types.Uid) map[string]any
peer = record.CallerID
direction = "incoming"
}
displayTopic := record.Topic
if viewTopic, err := types.P2PNameForUser(viewer, record.Topic); err == nil && viewTopic != "" {
displayTopic = viewTopic
}
ts := record.StartedAt
if ts.IsZero() {
ts = record.CreatedAt
}
data := map[string]any{
"call_id": record.CallID,
"topic": record.Topic,
"seq": record.SeqID,
"caller_id": record.CallerID.UserId(),
"callee_id": record.CalleeID.UserId(),
"peer_user_id": peer.UserId(),
"direction": direction,
"call_type": record.CallType,
"status": record.Status,
"result": record.Status,
"duration_ms": record.DurationMS,
"started_at": record.StartedAt.UTC().Format(time.RFC3339Nano),
"created_at": record.CreatedAt.UTC().Format(time.RFC3339Nano),
"updated_at": record.UpdatedAt.UTC().Format(time.RFC3339Nano),
"call_id": record.CallID,
"topic": record.Topic,
"display_topic": displayTopic,
"seq": record.SeqID,
"caller_id": record.CallerID.UserId(),
"callee_id": record.CalleeID.UserId(),
"peer_user_id": peer.UserId(),
"peer": map[string]any{
"user_id": peer.UserId(),
"topic": displayTopic,
},
"direction": direction,
"incoming": direction == "incoming",
"call_type": record.CallType,
"status": record.Status,
"result": record.Status,
"duration_ms": record.DurationMS,
"ts": ts.UTC().Format(time.RFC3339Nano),
"started_at": record.StartedAt.UTC().Format(time.RFC3339Nano),
"created_at": record.CreatedAt.UTC().Format(time.RFC3339Nano),
"updated_at": record.UpdatedAt.UTC().Format(time.RFC3339Nano),
}
if record.EndReason != "" {
data["end_reason"] = record.EndReason
@@ -9682,6 +9953,18 @@ func dailyReportDetailPayload(report *types.DailyReport, viewer types.Uid) map[s
return data
}
func dailyReportPermissions(req *http.Request) map[string]any {
scope := strings.ToLower(strings.TrimSpace(req.URL.Query().Get("scope")))
departmentID := strings.TrimSpace(req.URL.Query().Get("department_id"))
managed := scope == "managed" || scope == "department" || departmentID != ""
return map[string]any{
"can_create": scope == "" || scope == "mine" || scope == "self",
"makeup_allowed_days": 7,
"can_view_managed": managed,
"can_remind_unsubmitted": managed,
}
}
func monthlyReportListPayloads(reports []types.MonthlyReport) []map[string]any {
items := make([]map[string]any, 0, len(reports))
for _, report := range reports {
@@ -9836,19 +10119,21 @@ func validIMSecurityChallengeScene(scene string) bool {
func userPrivacyPayload(settings *types.UserPrivacySettings) map[string]any {
return map[string]any{
"add_me_methods": settings.AddMeMethods,
"profile_visibility": settings.ProfileVisibility,
"last_seen_visibility": settings.LastSeenVisibility,
"read_receipts": settings.ReadReceipts,
"group_invite_verification": settings.GroupInviteVerification,
"version": settings.Version,
"updated_at": settings.UpdatedAt.UTC().Format(time.RFC3339Nano),
"add_me_methods": settings.AddMeMethods,
"profile_visibility": settings.ProfileVisibility,
"last_seen_visibility": settings.LastSeenVisibility,
"read_receipts": settings.ReadReceipts,
"friend_request_verification": settings.FriendRequestVerification,
"group_invite_verification": settings.GroupInviteVerification,
"version": settings.Version,
"updated_at": settings.UpdatedAt.UTC().Format(time.RFC3339Nano),
"editable": map[string]any{
"add_me_methods": true,
"profile_visibility": true,
"last_seen_visibility": true,
"read_receipts": true,
"group_invite_verification": true,
"add_me_methods": true,
"profile_visibility": true,
"last_seen_visibility": true,
"read_receipts": true,
"friend_request_verification": true,
"group_invite_verification": true,
},
}
}
@@ -10221,22 +10506,28 @@ func friendRequestListItemPayload(item types.FriendRequestListItem) map[string]a
}
func friendListItemPayload(tenantID types.TenantID, viewer types.Uid, item types.FriendListItem) map[string]any {
relationship := item.Friendship.Status
canSend := relationship == types.RelationshipFriend
if ctx, err := store.Relations.GetRelationshipContext(tenantID, viewer, item.Friendship.PeerID, item.Friendship.Topic); err == nil && ctx != nil {
relationship = ctx.Relationship
canSend = ctx.CanSend
}
online := false
onlineVisibility := "hidden"
if item.Friendship.Status == types.RelationshipFriend && imCanViewLastSeen(tenantID, viewer, item.Friendship.PeerID) {
if relationship == types.RelationshipFriend && imCanViewLastSeen(tenantID, viewer, item.Friendship.PeerID) {
onlineVisibility = "visible"
online = redisUserOnline(tenantID, item.Friendship.PeerID, false)
}
data := map[string]any{
"user_id": item.Friendship.PeerID.UserId(),
"relationship": item.Friendship.Status,
"relationship": relationship,
"remark": item.Friendship.Remark,
"topic": item.Friendship.Topic,
"online": online,
"online_visibility": onlineVisibility,
"created_at": item.Friendship.CreatedAt.UTC().Format(time.RFC3339Nano),
"updated_at": item.Friendship.UpdatedAt.UTC().Format(time.RFC3339Nano),
"capabilities": map[string]any{"can_send": item.Friendship.Status == types.RelationshipFriend, "can_call": imCanCall(item.Friendship.Status == types.RelationshipFriend)},
"capabilities": map[string]any{"can_send": canSend, "can_call": imCanCall(canSend)},
"display_name_source": "profile",
}
if item.Friendship.Remark != "" {
@@ -10303,9 +10594,62 @@ func blockListItemPayload(item types.BlockListItem) map[string]any {
return data
}
func systemNotificationPayloads(items []types.SystemNotification) []map[string]any {
payloads := make([]map[string]any, 0, len(items))
for i := range items {
payloads = append(payloads, systemNotificationPayload(items[i]))
}
return payloads
}
func systemNotificationPayload(item types.SystemNotification) map[string]any {
summary := strings.TrimSpace(item.Summary)
if summary == "" {
summary = strings.TrimSpace(item.Title)
}
action := normalizeSystemNotificationAction(item.Action, item.EventType)
data := map[string]any{
"id": item.EventID,
"event_type": item.EventType,
"title": item.Title,
"summary": summary,
"action": action,
"created_at": item.CreatedAt.UTC().Format(time.RFC3339Nano),
}
if item.ReadAt != nil {
data["read_at"] = item.ReadAt.UTC().Format(time.RFC3339Nano)
}
return data
}
func normalizeSystemNotificationAction(action types.KVMap, eventType string) map[string]any {
normalized := map[string]any{
"type": strings.TrimSpace(eventType),
"target_id": "",
"payload": map[string]any{},
}
payload := map[string]any{}
for key, value := range action {
payload[key] = value
normalized[key] = value
}
for _, key := range []string{"meeting_id", "group_id", "topic", "resource_id", "target_id"} {
if value := strings.TrimSpace(fmt.Sprint(payload[key])); value != "" && value != "<nil>" {
normalized["target_id"] = value
break
}
}
normalized["payload"] = payload
return normalized
}
func userProfilePayload(user *types.User, ctx *types.RelationshipContext) map[string]any {
data := userSummaryPayload(user, ctx.Remark)
data["username"] = user.Username
if original := userOriginalDisplayName(user); original != "" {
data["profile_display_name"] = original
data["original_display_name"] = original
}
data["relationship"] = ctx.Relationship
data["topic"] = ctx.Topic
data["capabilities"] = map[string]any{
@@ -10330,6 +10674,10 @@ func userProfilePayload(user *types.User, ctx *types.RelationshipContext) map[st
func blockedUserProfilePayload(user *types.User, ctx *types.RelationshipContext) map[string]any {
data := userSummaryPayload(user, ctx.Remark)
data["username"] = user.Username
if original := userOriginalDisplayName(user); original != "" {
data["profile_display_name"] = original
data["original_display_name"] = original
}
data["relationship"] = ctx.Relationship
data["topic"] = ctx.Topic
data["capabilities"] = map[string]any{
@@ -10372,6 +10720,23 @@ func userSummaryPayload(user *types.User, remark string) map[string]any {
return data
}
func userOriginalDisplayName(user *types.User) string {
data := map[string]any{}
if public, ok := user.Public.(map[string]any); ok {
copyUserPublicSummary(data, public)
} else if public, ok := user.Public.(types.KVMap); ok {
copyUserPublicSummary(data, map[string]any(public))
}
if value := strings.TrimSpace(fmt.Sprint(data["display_name"])); value != "" && value != "<nil>" {
return value
}
uid := types.ParseUid(user.Id)
if uid.IsZero() {
return user.Id
}
return uid.UserId()
}
func copyUserPublicSummary(data map[string]any, public map[string]any) {
for _, key := range []string{"display_name", "fn", "name", "nickname"} {
if value, ok := public[key]; ok && strings.TrimSpace(fmt.Sprint(value)) != "" {
@@ -10670,6 +11035,17 @@ func forwardFailurePayload(topic string, seq int64, target, reason string) map[s
}
}
func forwardResultUnknownPayload(target string, seq int64, detailReason string) map[string]any {
return map[string]any{
"target_topic": strings.TrimSpace(target),
"seq": seq,
"status": "uncertain",
"reason": "MESSAGE_PERSIST_RESULT_UNKNOWN",
"detail_reason": detailReason,
"safe_to_retry": false,
}
}
func saveForwardCommentMessage(ctx context.Context, principal *imPrincipal, target, comment, clientMid string) map[string]any {
comment = strings.TrimSpace(comment)
if principal == nil || comment == "" {
+13 -9
View File
@@ -1090,7 +1090,7 @@ func (s *testProfilesStore) GetPrivacy(_ types.TenantID, uid types.Uid) (*types.
now := time.Now().UTC().Round(time.Millisecond)
s.privacy = &types.UserPrivacySettings{
UserID: uid, AddMeMethods: map[string]bool{"account": true, "phone": true, "qr": true, "group": true},
ProfileVisibility: "friends", LastSeenVisibility: "friends", ReadReceipts: true, GroupInviteVerification: false, Version: 1, CreatedAt: now, UpdatedAt: now,
ProfileVisibility: "friends", LastSeenVisibility: "friends", ReadReceipts: true, FriendRequestVerification: true, GroupInviteVerification: false, Version: 1, CreatedAt: now, UpdatedAt: now,
}
}
return s.privacy, nil
@@ -1110,6 +1110,9 @@ func (s *testProfilesStore) UpdatePrivacy(_ types.TenantID, uid types.Uid, patch
if patch.ReadReceipts != nil {
privacy.ReadReceipts = *patch.ReadReceipts
}
if patch.FriendRequestVerification != nil {
privacy.FriendRequestVerification = *patch.FriendRequestVerification
}
if patch.GroupInviteVerification != nil {
privacy.GroupInviteVerification = *patch.GroupInviteVerification
}
@@ -3900,7 +3903,7 @@ func TestServeIMMePrivacy(t *testing.T) {
store.Profiles = profiles
defer func() { store.Profiles = oldProfiles }()
req := httptest.NewRequest(http.MethodPatch, "/v1/im/me/privacy", strings.NewReader(`{"add_me_methods":{"account":true,"phone":false,"qr":true,"group":true},"read_receipts":false,"group_invite_verification":true,"version":1}`))
req := httptest.NewRequest(http.MethodPatch, "/v1/im/me/privacy", strings.NewReader(`{"add_me_methods":{"account":true,"phone":false,"qr":true,"group":true},"read_receipts":false,"friend_request_verification":false,"group_invite_verification":true,"version":1}`))
req.Header.Set("Authorization", "Bearer token-1")
rec := httptest.NewRecorder()
serveIMMePrivacy(rec, req)
@@ -3911,8 +3914,8 @@ func TestServeIMMePrivacy(t *testing.T) {
data := decodeIMHTTPResponse(t, rec).Data.(map[string]any)
methods := data["add_me_methods"].(map[string]any)
editable := data["editable"].(map[string]any)
if methods["phone"] != false || data["read_receipts"] != false || data["group_invite_verification"] != true ||
editable["group_invite_verification"] != true || data["version"].(float64) != 2 {
if methods["phone"] != false || data["read_receipts"] != false || data["friend_request_verification"] != false || data["group_invite_verification"] != true ||
editable["friend_request_verification"] != true || editable["group_invite_verification"] != true || data["version"].(float64) != 2 {
t.Fatalf("unexpected privacy: %#v", data)
}
}
@@ -4574,7 +4577,7 @@ func TestServeIMGroupList(t *testing.T) {
}
data := decodeIMHTTPResponse(t, rec).Data.(map[string]any)
items := data["items"].([]any)
if len(items) != 1 || items[0].(map[string]any)["name"] != "Alpha" || items[0].(map[string]any)["member_count"].(float64) != 3 || items[0].(map[string]any)["my_role"] != "owner" {
if len(items) != 1 || items[0].(map[string]any)["name"] != "Alpha" || items[0].(map[string]any)["member_count"].(float64) != 0 || items[0].(map[string]any)["my_role"] != "owner" {
t.Fatalf("unexpected group list response: %#v", data)
}
}
@@ -6785,7 +6788,7 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) {
}
}
func TestServeIMMessageForwardKafkaPublishUnknownReturnsPending(t *testing.T) {
func TestServeIMMessageForwardKafkaPublishUnknownReturnsUncertain(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
mockStore := expectIMBearer(t, ctrl, "token-1", 7, 42)
@@ -6864,8 +6867,9 @@ func TestServeIMMessageForwardKafkaPublishUnknownReturnsPending(t *testing.T) {
t.Fatalf("unexpected forward response: %#v", data)
}
got := items[0].(map[string]any)
if got["status"] != "pending" || got["reason"] != "MESSAGE_PERSIST_PENDING" || got["seq"].(float64) != 41 {
t.Fatalf("unexpected pending item: %#v", got)
if got["status"] != "uncertain" || got["reason"] != "MESSAGE_PERSIST_RESULT_UNKNOWN" ||
got["safe_to_retry"] != false || got["seq"].(float64) != 41 {
t.Fatalf("unexpected uncertain item: %#v", got)
}
if got["client_mid"] != "fwd_"+stableIMEventID("forward", "idem-forward-1", "separate", targetTopic, "grpSource", int64(14)) {
t.Fatalf("unexpected stable client mid: %#v", got)
@@ -6875,7 +6879,7 @@ func TestServeIMMessageForwardKafkaPublishUnknownReturnsPending(t *testing.T) {
}
select {
case delivered := <-globals.hub.routeSrv:
t.Fatalf("pending forward should not broadcast immediately: %#v", delivered)
t.Fatalf("uncertain forward should not broadcast immediately: %#v", delivered)
default:
}
}
+13 -1
View File
@@ -756,7 +756,7 @@ func (s *Session) subscribe(msg *ClientComMessage) {
// client complete the alias->canonical rename with a 200 reply instead of
// a 304 that would leave the client-side alias topic object unattached
// (publishes on it then fail with "Cannot publish on inactive topic").
if sub := s.getSub(msg.RcptTo); sub != nil && msg.Original == msg.RcptTo {
if sub := s.getSub(msg.RcptTo); sub != nil && canReplyAlreadySubscribed(msg.Original, msg.RcptTo) {
s.queueOut(InfoAlreadySubscribed(msg.Id, msg.Original, msg.Timestamp))
s.inflightReqs.Done()
} else {
@@ -772,6 +772,18 @@ func (s *Session) subscribe(msg *ClientComMessage) {
}
}
func canReplyAlreadySubscribed(original, routeTo string) bool {
if original == routeTo {
return true
}
switch original {
case "me", "fnd", "slf":
return true
default:
return false
}
}
// Leave/Unsubscribe a topic
func (s *Session) leave(msg *ClientComMessage) {
// Expand topic name
+66 -5
View File
@@ -2,6 +2,9 @@ package store
import (
"context"
"encoding/json"
"strconv"
"strings"
"testing"
"time"
@@ -12,6 +15,7 @@ type testGroupMessageRuntime struct {
claims map[string]*groupMessageUniqueRecord
seqs map[string]int64
shortStates map[string]bool
objects map[string][]byte
deletedPrefixes []string
deletedObjects []string
}
@@ -21,6 +25,7 @@ func newTestGroupMessageRuntime() *testGroupMessageRuntime {
claims: make(map[string]*groupMessageUniqueRecord),
seqs: make(map[string]int64),
shortStates: make(map[string]bool),
objects: make(map[string][]byte),
}
}
@@ -145,21 +150,37 @@ func (r *testGroupMessageRuntime) StartRealtimeFanoutSubscriber(context.Context,
return nil
}
func (r *testGroupMessageRuntime) CacheObject(types.TenantID, string, string, any) error {
func (r *testGroupMessageRuntime) CacheObject(tenantID types.TenantID, namespace, id string, value any) error {
payload, err := json.Marshal(value)
if err != nil {
return err
}
r.objects[strconv.FormatInt(int64(tenantID), 10)+":"+namespace+":"+id] = payload
return nil
}
func (r *testGroupMessageRuntime) GetCachedObject(types.TenantID, string, string, any) (bool, error) {
return false, nil
func (r *testGroupMessageRuntime) GetCachedObject(tenantID types.TenantID, namespace, id string, value any) (bool, error) {
payload, ok := r.objects[strconv.FormatInt(int64(tenantID), 10)+":"+namespace+":"+id]
if !ok {
return false, nil
}
return true, json.Unmarshal(payload, value)
}
func (r *testGroupMessageRuntime) DeleteCachedObject(_ types.TenantID, namespace, id string) error {
func (r *testGroupMessageRuntime) DeleteCachedObject(tenantID types.TenantID, namespace, id string) error {
r.deletedObjects = append(r.deletedObjects, namespace+":"+id)
delete(r.objects, strconv.FormatInt(int64(tenantID), 10)+":"+namespace+":"+id)
return nil
}
func (r *testGroupMessageRuntime) DeleteCachedObjectsByPrefix(_ types.TenantID, _ string, prefix string) error {
func (r *testGroupMessageRuntime) DeleteCachedObjectsByPrefix(tenantID types.TenantID, namespace, prefix string) error {
r.deletedPrefixes = append(r.deletedPrefixes, prefix)
cachePrefix := strconv.FormatInt(int64(tenantID), 10) + ":" + namespace + ":" + prefix
for key := range r.objects {
if strings.HasPrefix(key, cachePrefix) {
delete(r.objects, key)
}
}
return nil
}
@@ -220,6 +241,46 @@ func TestLoginTokenRevocation(t *testing.T) {
}
}
func TestGroupMemberPageCacheRoundTripAndInvalidate(t *testing.T) {
oldRuntime := groupMsgRuntime
runtime := newTestGroupMessageRuntime()
groupMsgRuntime = runtime
t.Cleanup(func() { groupMsgRuntime = oldRuntime })
opts := types.GroupMemberListOptions{
Role: "member",
Query: " alice ",
Cursor: "42",
Limit: 20,
Mentionable: true,
}
key := groupMemberPageCacheID("grpAlpha", opts)
if !strings.HasPrefix(key, "grpAlpha|") {
t.Fatalf("cache key must be topic-prefixed for invalidation: %q", key)
}
CacheGroupMemberPage(7, key, groupMemberPageCache{
Items: []types.GroupMember{{
ID: 43,
Topic: "grpAlpha",
UserID: types.Uid(100),
Role: types.GroupMemberRoleMember,
}},
NextCursor: "43",
})
cached, ok := GetCachedGroupMemberPage(7, key)
if !ok || cached.NextCursor != "43" || len(cached.Items) != 1 || cached.Items[0].UserID != types.Uid(100) {
t.Fatalf("unexpected cached page: ok=%v page=%#v", ok, cached)
}
invalidateGroupMemberPages(7, "grpAlpha")
if _, ok := GetCachedGroupMemberPage(7, key); ok {
t.Fatalf("group member page cache survived invalidation")
}
if len(runtime.deletedPrefixes) != 1 || runtime.deletedPrefixes[0] != "grpAlpha|" {
t.Fatalf("unexpected invalidation prefixes: %#v", runtime.deletedPrefixes)
}
}
func TestRealtimeFanoutDedupClaimIDIsNodeScoped(t *testing.T) {
tenantID := types.TenantID(7)
dedupKey := "data:p2pABC:42"
+56 -9
View File
@@ -1760,6 +1760,9 @@ func (groupBusinessMapper) GetSettings(tenantID types.TenantID, topic string) (*
var cached types.GroupSettings
if ok, err := GetCachedObject(tenantID, cacheNamespaceGroupSettings, topic, &cached); err != nil {
logs.Warn.Printf("cache group settings[%s]: read failed: %v", topic, err)
if delErr := DeleteCachedObject(tenantID, cacheNamespaceGroupSettings, topic); delErr != nil {
logs.Warn.Printf("cache group settings[%s]: delete corrupted cache failed: %v", topic, delErr)
}
} else if ok {
return &cached, nil
}
@@ -1874,15 +1877,51 @@ func (groupBusinessMapper) ListMembers(tenantID types.TenantID, topic string, op
if opts.Limit <= 0 || opts.Limit > 100 {
opts.Limit = 30
}
topic = strings.TrimSpace(topic)
opts.Role = strings.TrimSpace(opts.Role)
opts.EligibleRole = strings.TrimSpace(opts.EligibleRole)
opts.Query = strings.TrimSpace(opts.Query)
opts.Cursor = strings.TrimSpace(opts.Cursor)
cacheKey := groupMemberPageCacheID(topic, opts)
if cached, ok := GetCachedGroupMemberPage(tenantID, cacheKey); ok {
return cached.Items, cached.NextCursor, nil
}
tenantAdp, err := tenantBusinessAdapter(tenantID)
if err != nil {
return nil, "", err
}
return tenantAdp.TenantGroupMemberList(tenantID, strings.TrimSpace(topic), opts)
members, nextCursor, err := tenantAdp.TenantGroupMemberList(tenantID, topic, opts)
if err != nil {
return nil, "", err
}
CacheGroupMemberPage(tenantID, cacheKey, groupMemberPageCache{
Items: members,
NextCursor: nextCursor,
})
return members, nextCursor, nil
}
func CacheGroupMemberPage(tenantID types.TenantID, key string, payload map[string]any) {
if strings.TrimSpace(key) == "" || payload == nil {
type groupMemberPageCache struct {
Items []types.GroupMember `json:"items"`
NextCursor string `json:"next_cursor"`
}
func groupMemberPageCacheID(topic string, opts types.GroupMemberListOptions) string {
limit := opts.Limit
if limit <= 0 || limit > 100 {
limit = 30
}
return strings.TrimSpace(topic) +
"|role=" + strconv.Quote(strings.TrimSpace(opts.Role)) +
"|eligible_role=" + strconv.Quote(strings.TrimSpace(opts.EligibleRole)) +
"|query=" + strconv.Quote(strings.TrimSpace(opts.Query)) +
"|cursor=" + strconv.Quote(strings.TrimSpace(opts.Cursor)) +
"|limit=" + strconv.Itoa(limit) +
"|mentionable=" + strconv.FormatBool(opts.Mentionable)
}
func CacheGroupMemberPage(tenantID types.TenantID, key string, payload groupMemberPageCache) {
if strings.TrimSpace(key) == "" {
return
}
if err := CacheObject(tenantID, cacheNamespaceGroupPage, strings.TrimSpace(key), payload); err != nil {
@@ -1890,17 +1929,17 @@ func CacheGroupMemberPage(tenantID types.TenantID, key string, payload map[strin
}
}
func GetCachedGroupMemberPage(tenantID types.TenantID, key string) (map[string]any, bool) {
func GetCachedGroupMemberPage(tenantID types.TenantID, key string) (groupMemberPageCache, bool) {
if strings.TrimSpace(key) == "" {
return nil, false
return groupMemberPageCache{}, false
}
var payload map[string]any
var payload groupMemberPageCache
if ok, err := GetCachedObject(tenantID, cacheNamespaceGroupPage, strings.TrimSpace(key), &payload); err != nil {
logs.Warn.Printf("cache group member page[%s]: read failed: %v", key, err)
} else if ok {
return payload, true
}
return nil, false
return groupMemberPageCache{}, false
}
func invalidateGroupMemberPages(tenantID types.TenantID, topic string) {
@@ -2113,7 +2152,11 @@ func (groupBusinessMapper) UpsertMute(tenantID types.TenantID, mute *types.Group
if err != nil {
return nil, err
}
return tenantAdp.TenantGroupMuteUpsert(tenantID, mute)
updated, err := tenantAdp.TenantGroupMuteUpsert(tenantID, mute)
if err == nil {
invalidateGroupMemberPages(tenantID, mute.Topic)
}
return updated, err
}
func (groupBusinessMapper) DeleteMute(tenantID types.TenantID, topic string, uid, actor types.Uid, now time.Time) error {
@@ -2124,7 +2167,11 @@ func (groupBusinessMapper) DeleteMute(tenantID types.TenantID, topic string, uid
if err != nil {
return err
}
return tenantAdp.TenantGroupMuteDelete(tenantID, strings.TrimSpace(topic), uid, actor, now)
if err := tenantAdp.TenantGroupMuteDelete(tenantID, strings.TrimSpace(topic), uid, actor, now); err != nil {
return err
}
invalidateGroupMemberPages(tenantID, topic)
return nil
}
func (groupBusinessMapper) ListFiles(tenantID types.TenantID, topic, fileType, query, cursor string, limit int) ([]types.GroupFile, string, error) {
+2
View File
@@ -171,6 +171,8 @@ type GroupAnnouncement struct {
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
AuthorUserIDRaw int64 `db:"author_user_id"`
ReadCount int `db:"read_count"`
MemberCount int `db:"member_count"`
}
type GroupMute struct {
+17 -15
View File
@@ -29,24 +29,26 @@ type UserProfilePatch struct {
}
type UserPrivacySettings struct {
UserID Uid
AddMeMethods map[string]bool
ProfileVisibility string
LastSeenVisibility string
ReadReceipts bool
GroupInviteVerification bool
Version int
CreatedAt time.Time
UpdatedAt time.Time
UserID Uid
AddMeMethods map[string]bool
ProfileVisibility string
LastSeenVisibility string
ReadReceipts bool
FriendRequestVerification bool
GroupInviteVerification bool
Version int
CreatedAt time.Time
UpdatedAt time.Time
}
type UserPrivacyPatch struct {
AddMeMethods map[string]bool
ProfileVisibility *string
LastSeenVisibility *string
ReadReceipts *bool
GroupInviteVerification *bool
Version int
AddMeMethods map[string]bool
ProfileVisibility *string
LastSeenVisibility *string
ReadReceipts *bool
FriendRequestVerification *bool
GroupInviteVerification *bool
Version int
}
type AppRelease struct {