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 ( require (
firebase.google.com/go v3.13.0+incompatible 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/darabonba-openapi/v2 v2.2.4
github.com/alibabacloud-go/green-20220302/v3 v3.5.4 github.com/alibabacloud-go/green-20220302/v3 v3.5.4
github.com/alibabacloud-go/tea v1.5.2 github.com/alibabacloud-go/tea v1.5.2
@@ -27,6 +28,7 @@ require (
github.com/prometheus/common v0.69.0 github.com/prometheus/common v0.69.0
github.com/redis/go-redis/v9 v9.12.1 github.com/redis/go-redis/v9 v9.12.1
github.com/rivo/uniseg v0.4.7 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/jsonco v1.0.0
github.com/tinode/snowflake v1.0.0 github.com/tinode/snowflake v1.0.0
go.mongodb.org/mongo-driver v1.17.9 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/auth/oauth2adapt v0.2.8 // indirect
cloud.google.com/go/monitoring v1.29.0 // indirect cloud.google.com/go/monitoring v1.29.0 // indirect
filippo.io/edwards25519 v1.2.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/detectors/gcp v1.33.0 // indirect
github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.57.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 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/pierrec/lz4/v4 v4.1.15 // indirect
github.com/pkg/errors v0.9.1 // indirect github.com/pkg/errors v0.9.1 // indirect
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // 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/spiffe/go-spiffe/v2 v2.8.1 // indirect
github.com/stretchr/objx v0.5.0 // indirect github.com/stretchr/objx v0.5.0 // indirect
github.com/tjfoc/gmsm v1.4.1 // 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', profile_visibility VARCHAR(32) NOT NULL DEFAULT 'friends',
last_seen_visibility VARCHAR(32) NOT NULL DEFAULT 'friends', last_seen_visibility VARCHAR(32) NOT NULL DEFAULT 'friends',
read_receipts TINYINT NOT NULL DEFAULT 1, read_receipts TINYINT NOT NULL DEFAULT 1,
friend_request_verification TINYINT NOT NULL DEFAULT 1,
group_invite_verification TINYINT NOT NULL DEFAULT 0, group_invite_verification TINYINT NOT NULL DEFAULT 0,
version INT NOT NULL DEFAULT 1, version INT NOT NULL DEFAULT 1,
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), 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 { if err != nil {
return nil, err return nil, err
} }
defer rows.Close()
var locations []string type gcCandidate struct {
var ids []any id int
location string
}
var candidates []gcCandidate
for rows.Next() { for rows.Next() {
var id int var id int
var loc string var loc string
if err = rows.Scan(&id, &loc); err != nil { if err = rows.Scan(&id, &loc); err != nil {
break 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 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 { 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 break
} }
if noteRefs > 0 { if noteRefs > 0 {
continue continue
} }
if loc != "" { if candidate.location != "" {
locations = append(locations, loc) locations = append(locations, candidate.location)
} }
ids = append(ids, id) ids = append(ids, candidate.id)
}
if err == nil {
err = rows.Err()
} }
if err != nil { if err != nil {
@@ -4900,7 +4914,24 @@ func (a *adapter) FileUserHasAccess(fid string, uid t.Uid) (bool, error) {
defer cancel() defer cancel()
} }
var allowed int 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 SELECT COUNT(*) FROM im_fileuploads AS fu
WHERE (fu.id=? OR fu.location=? OR fu.location LIKE ?) AND fu.status=? AND ( WHERE (fu.id=? OR fu.location=? OR fu.location LIKE ?) AND fu.status=? AND (
fu.userid=? 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.end_at<=CURRENT_TIMESTAMP(3)
AND ( AND (
mtg.organizer_id=? mtg.organizer_id=?
OR mtg.recorder_user_id=? ` + recorderClause + `
OR EXISTS ( OR EXISTS (
SELECT 1 FROM im_meeting_participants AS mp SELECT 1 FROM im_meeting_participants AS mp
WHERE mp.meeting_id=mtg.meeting_id AND mp.user_id=? WHERE mp.meeting_id=mtg.meeting_id AND mp.user_id=?
) )
) )
) )
) LIMIT 1`, ) LIMIT 1`, args
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), includeRecorder := true
store.DecodeUid(uid), fid, fid, run := func() error {
fid, t.MeetingStateCanceled, store.DecodeUid(uid), store.DecodeUid(uid), store.DecodeUid(uid)) 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 { if err != nil {
return false, err return false, err
} }
@@ -5108,6 +5150,20 @@ func isIDOutOfRange(err error) bool {
return ok && myerr.Number == 1264 && strings.Contains(strings.ToLower(myerr.Message), "id") 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 { func isPXCStrictModeNamedLockError(err error) bool {
if err == nil { if err == nil {
return false 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), 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) 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`, ) 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 { for _, statement := range statements {
if _, err := exec.Exec(statement); err != nil { if _, err := exec.Exec(statement); err != nil {
+8 -2
View File
@@ -6609,6 +6609,10 @@ func (a *adapter) TenantUserPrivacyUpdate(id t.TenantID, uid t.Uid, patch t.User
sets = append(sets, "read_receipts=?") sets = append(sets, "read_receipts=?")
args = append(args, *patch.ReadReceipts) args = append(args, *patch.ReadReceipts)
} }
if patch.FriendRequestVerification != nil {
sets = append(sets, "friend_request_verification=?")
args = append(args, *patch.FriendRequestVerification)
}
if patch.GroupInviteVerification != nil { if patch.GroupInviteVerification != nil {
sets = append(sets, "group_invite_verification=?") sets = append(sets, "group_invite_verification=?")
args = append(args, *patch.GroupInviteVerification) args = append(args, *patch.GroupInviteVerification)
@@ -7185,13 +7189,14 @@ func (a *adapter) userPrivacyByUID(ctx context.Context, uid t.Uid) (*t.UserPriva
ProfileVisibility string `db:"profile_visibility"` ProfileVisibility string `db:"profile_visibility"`
LastSeenVisibility string `db:"last_seen_visibility"` LastSeenVisibility string `db:"last_seen_visibility"`
ReadReceipts bool `db:"read_receipts"` ReadReceipts bool `db:"read_receipts"`
FriendRequestVerification bool `db:"friend_request_verification"`
GroupInviteVerification bool `db:"group_invite_verification"` GroupInviteVerification bool `db:"group_invite_verification"`
Version int Version int
CreatedAt time.Time `db:"created_at"` CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_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, 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 return nil, err
} }
settings := &t.UserPrivacySettings{ settings := &t.UserPrivacySettings{
@@ -7200,6 +7205,7 @@ func (a *adapter) userPrivacyByUID(ctx context.Context, uid t.Uid) (*t.UserPriva
ProfileVisibility: row.ProfileVisibility, ProfileVisibility: row.ProfileVisibility,
LastSeenVisibility: row.LastSeenVisibility, LastSeenVisibility: row.LastSeenVisibility,
ReadReceipts: row.ReadReceipts, ReadReceipts: row.ReadReceipts,
FriendRequestVerification: row.FriendRequestVerification,
GroupInviteVerification: row.GroupInviteVerification, GroupInviteVerification: row.GroupInviteVerification,
Version: row.Version, Version: row.Version,
CreatedAt: row.CreatedAt, CreatedAt: row.CreatedAt,
@@ -9314,7 +9320,7 @@ func (a *adapter) TenantSystemNotificationMarkRead(id t.TenantID, uid t.Uid, eve
} }
res, err := a.db.ExecContext(ctx, res, err := a.db.ExecContext(ctx,
`UPDATE im_system_notifications `UPDATE im_system_notifications
SET read_at=COALESCE(read_at,?),updated_at=? SET read_at=?,updated_at=?
WHERE user_id=? AND event_id=?`, WHERE user_id=? AND event_id=?`,
readAt, readAt, decodeUID(uid), eventID) readAt, readAt, decodeUID(uid), eventID)
if err != nil { if err != nil {
+14 -1
View File
@@ -4,7 +4,10 @@
package mysql package mysql
import ( import (
"database/sql"
"encoding/json" "encoding/json"
"errors"
"strings"
"time" "time"
"github.com/jmoiron/sqlx" "github.com/jmoiron/sqlx"
@@ -60,7 +63,17 @@ func (a *adapter) TenantApplicationCreate(application *t.TenantApplication) (*t.
if cancel != nil { if cancel != nil {
defer cancel() 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) (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(?,?,?,?,?,?,?,?,?,?,?,?)`, VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`,
application.ApplicationID, application.CreditCode, application.EnterpriseEmail, application.LicenseFileID, application.ApplicationID, application.CreditCode, application.EnterpriseEmail, application.LicenseFileID,
+3 -1
View File
@@ -20,7 +20,9 @@ const (
groupSettingsSelectColumns = `topic,owner_user_id,name,avatar_file_id,description,group_status,join_approval,mute_all, 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, 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` 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` 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` groupMuteSelectColumns = `topic,user_id,muted_by,muted_at,expires_at,reason,status,revoked_by,revoked_at,version,created_at,updated_at`
) )
+420 -44
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"}) writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "description"})
return 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 { if len(body.EvidenceFileIDs) > 9 {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "evidence_file_ids"}) writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "evidence_file_ids"})
return return
@@ -2241,6 +2260,12 @@ func serveIMMeetings(wrt http.ResponseWriter, req *http.Request) {
if !ok { if !ok {
return 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 recorderID := principal.Uid
if value := strings.TrimSpace(body.RecorderUserID); value != "" { if value := strings.TrimSpace(body.RecorderUserID); value != "" {
recorderID = types.ParseUserId(value) recorderID = types.ParseUserId(value)
@@ -2568,6 +2593,27 @@ func serveIMMeetingCancel(wrt http.ResponseWriter, req *http.Request, meetingID
if !ok { if !ok {
return 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) meeting, err := store.Meetings.Cancel(principal.TenantID, principal.Uid, meetingID, idempotencyKey)
if err != nil { if err != nil {
writeIMRelationError(wrt, req, err) writeIMRelationError(wrt, req, err)
@@ -2663,6 +2709,9 @@ func serveIMDailyReports(wrt http.ResponseWriter, req *http.Request) {
switch req.Method { switch req.Method {
case http.MethodGet: case http.MethodGet:
limit := parseIMLimit(req.URL.Query().Get("limit"), 50, 100) 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, reports, nextCursor, err := store.DailyReports.List(principal.TenantID, principal.Uid,
req.URL.Query().Get("scope"), req.URL.Query().Get("scope"),
req.URL.Query().Get("from"), req.URL.Query().Get("from"),
@@ -2677,12 +2726,7 @@ func serveIMDailyReports(wrt http.ResponseWriter, req *http.Request) {
writeIMSuccess(wrt, req, map[string]any{ writeIMSuccess(wrt, req, map[string]any{
"items": dailyReportListPayloads(reports), "items": dailyReportListPayloads(reports),
"next_cursor": nextCursor, "next_cursor": nextCursor,
"permissions": map[string]any{ "permissions": dailyReportPermissions(req),
"can_create": true,
"makeup_allowed_days": 7,
"can_view_managed": false,
"can_remind_unsubmitted": false,
},
}) })
case http.MethodPost: case http.MethodPost:
var body struct { var body struct {
@@ -2692,7 +2736,7 @@ func serveIMDailyReports(wrt http.ResponseWriter, req *http.Request) {
return return
} }
reportDate, err := parseIMDate(body.ReportDate) 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"}) writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "report_date"})
return return
} }
@@ -2781,8 +2825,8 @@ func serveIMDailyReportSaveDraft(wrt http.ResponseWriter, req *http.Request, rep
writeIMRelationError(wrt, req, err) writeIMRelationError(wrt, req, err)
return return
} }
publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "daily_report", "daily_report.submitted", report.ReportID, 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}) []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)) 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) { if !decodeIMJSONBody(wrt, req, &body) {
return 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) report, err := store.DailyReports.Submit(principal.TenantID, principal.Uid, reportID, body.Version)
if err != nil { if err != nil {
writeIMRelationError(wrt, req, err) writeIMRelationError(wrt, req, err)
@@ -2887,6 +2944,9 @@ func serveIMMonthlyReports(wrt http.ResponseWriter, req *http.Request) {
switch req.Method { switch req.Method {
case http.MethodGet: case http.MethodGet:
limit := parseIMLimit(req.URL.Query().Get("limit"), 50, 100) 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, reports, nextCursor, err := store.MonthlyReports.List(principal.TenantID, principal.Uid,
req.URL.Query().Get("scope"), req.URL.Query().Get("scope"),
req.URL.Query().Get("from_month"), req.URL.Query().Get("from_month"),
@@ -2906,7 +2966,7 @@ func serveIMMonthlyReports(wrt http.ResponseWriter, req *http.Request) {
if !decodeIMJSONBody(wrt, req, &body) { if !decodeIMJSONBody(wrt, req, &body) {
return 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"}) writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "month"})
return return
} }
@@ -2995,8 +3055,8 @@ func serveIMMonthlyReportSaveDraft(wrt http.ResponseWriter, req *http.Request, r
writeIMRelationError(wrt, req, err) writeIMRelationError(wrt, req, err)
return return
} }
publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "monthly_report", "monthly_report.submitted", report.ReportID, 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}) []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)) 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) { if !decodeIMJSONBody(wrt, req, &body) {
return 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) report, err := store.MonthlyReports.Submit(principal.TenantID, principal.Uid, reportID, body.Version)
if err != nil { if err != nil {
writeIMRelationError(wrt, req, err) writeIMRelationError(wrt, req, err)
return return
} }
publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "monthly_report", "monthly_report.reopened", report.ReportID, 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}) []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)) writeIMSuccess(wrt, req, monthlyReportDetailPayload(report, principal.Uid))
} }
@@ -3275,6 +3352,7 @@ func serveIMMePrivacy(wrt http.ResponseWriter, req *http.Request) {
ProfileVisibility *string `json:"profile_visibility"` ProfileVisibility *string `json:"profile_visibility"`
LastSeenVisibility *string `json:"last_seen_visibility"` LastSeenVisibility *string `json:"last_seen_visibility"`
ReadReceipts *bool `json:"read_receipts"` ReadReceipts *bool `json:"read_receipts"`
FriendRequestVerification *bool `json:"friend_request_verification"`
GroupInviteVerification *bool `json:"group_invite_verification"` GroupInviteVerification *bool `json:"group_invite_verification"`
Version int `json:"version"` Version int `json:"version"`
} }
@@ -3286,6 +3364,7 @@ func serveIMMePrivacy(wrt http.ResponseWriter, req *http.Request) {
ProfileVisibility: body.ProfileVisibility, ProfileVisibility: body.ProfileVisibility,
LastSeenVisibility: body.LastSeenVisibility, LastSeenVisibility: body.LastSeenVisibility,
ReadReceipts: body.ReadReceipts, ReadReceipts: body.ReadReceipts,
FriendRequestVerification: body.FriendRequestVerification,
GroupInviteVerification: body.GroupInviteVerification, GroupInviteVerification: body.GroupInviteVerification,
Version: body.Version, Version: body.Version,
}) })
@@ -3611,7 +3690,7 @@ func serveIMSystemNotifications(wrt http.ResponseWriter, req *http.Request) {
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
return 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) { 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) saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, msg, nil, true)
if err != nil { if err != nil {
if saved != nil && saved.Pending { 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) mergeForwardCommentResult(result, commentResult)
results = append(results, result) results = append(results, result)
continue continue
@@ -4002,16 +4089,21 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
if err != nil { if err != nil {
if saved != nil && saved.Pending { if saved != nil && saved.Pending {
forwarded.SeqId = saved.SeqId forwarded.SeqId = saved.SeqId
item := map[string]any{ item := forwardResultUnknownPayload(target, saved.SeqId, relationErrorReason(err))
"sources": bundle["source_refs"], item["sources"] = bundle["source_refs"]
"target_topic": target, item["client_mid"] = clientMid
"client_mid": clientMid, item["mode"] = "bundle"
"seq": saved.SeqId, item["bundle_id"] = bundle["bundle_id"]
"status": "pending", mergeForwardCommentResult(item, commentResult)
"reason": "MESSAGE_PERSIST_PENDING", items = append(items, item)
"mode": "bundle", continue
"bundle_id": bundle["bundle_id"],
} }
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) mergeForwardCommentResult(item, commentResult)
items = append(items, item) items = append(items, item)
continue continue
@@ -4094,14 +4186,17 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
if err != nil { if err != nil {
if saved != nil && saved.Pending { if saved != nil && saved.Pending {
forwarded.SeqId = saved.SeqId forwarded.SeqId = saved.SeqId
item := map[string]any{ item := forwardResultUnknownPayload(target, saved.SeqId, relationErrorReason(err))
"source": map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)}, item["source"] = map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)}
"target_topic": target, item["client_mid"] = clientMid
"client_mid": clientMid, mergeForwardCommentResult(item, commentResult)
"seq": saved.SeqId, items = append(items, item)
"status": "pending", continue
"reason": "MESSAGE_PERSIST_PENDING",
} }
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) mergeForwardCommentResult(item, commentResult)
items = append(items, item) items = append(items, item)
continue continue
@@ -4664,6 +4759,10 @@ func serveIMConversationContext(wrt http.ResponseWriter, req *http.Request) {
return return
} }
topic, tail := conversationResourceFromPath(req.URL.Path) topic, tail := conversationResourceFromPath(req.URL.Path)
if topic == "settings" && tail == "" {
serveIMConversationSettingsBatch(wrt, req, principal)
return
}
if tail == "settings" { if tail == "settings" {
serveIMConversationSettings(wrt, req, principal, topic) serveIMConversationSettings(wrt, req, principal, topic)
return return
@@ -4697,6 +4796,48 @@ func serveIMConversationContext(wrt http.ResponseWriter, req *http.Request) {
writeIMSuccess(wrt, req, relationshipContextPayload(ctx)) 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) { func serveIMConversationMessages(wrt http.ResponseWriter, req *http.Request, principal *imPrincipal, topic string) {
if topic == "" { if topic == "" {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "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) writeIMRelationError(wrt, req, err)
return 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) { 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, 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", 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) 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) writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
return return
} }
groupPayload := groupDetailPayload(group, selfSub)
writeIMSuccess(wrt, req, map[string]any{ writeIMSuccess(wrt, req, map[string]any{
"qr_token": token, "qr_token": token,
"qr_payload": "im://qr?token=" + url.QueryEscape(token) + "&purpose=join_group", "qr_payload": "im://qr?token=" + url.QueryEscape(token) + "&purpose=join_group",
"expires_at": expiresAt.Format(time.RFC3339Nano), "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() { if uid.IsZero() {
continue 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 var expiresAt *time.Time
if raw := strings.TrimSpace(member.ExpiresAt); raw != "" { if raw := strings.TrimSpace(member.ExpiresAt); raw != "" {
if parsed, err := time.Parse(time.RFC3339Nano, raw); err == nil { 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"}) writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "user_id"})
return 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 { if err := store.GroupBusiness.DeleteMute(principal.TenantID, topic, uid, principal.Uid, time.Now().UTC().Round(time.Millisecond)); err != nil {
writeIMRelationError(wrt, req, err) writeIMRelationError(wrt, req, err)
return return
@@ -6750,10 +6923,12 @@ func serveIMQRCodesResolve(wrt http.ResponseWriter, req *http.Request) {
if selfSub != nil && (selfSub.ModeGiven & selfSub.ModeWant).IsJoiner() { if selfSub != nil && (selfSub.ModeGiven & selfSub.ModeWant).IsJoiner() {
status = "joined" status = "joined"
} }
groupPayload := groupDetailPayload(group, selfSub)
writeIMSuccess(wrt, req, map[string]any{ writeIMSuccess(wrt, req, map[string]any{
"purpose": claims.Purpose, "purpose": claims.Purpose,
"expires_at": time.Unix(claims.Expires, 0).UTC().Format(time.RFC3339Nano), "expires_at": time.Unix(claims.Expires, 0).UTC().Format(time.RFC3339Nano),
"group": groupDetailPayload(group, selfSub), "group": groupPayload,
"group_snapshot": groupPayload,
"status": status, "status": status,
"source": "qr", "source": "qr",
}) })
@@ -7406,16 +7581,23 @@ func authenticateIMHTTPBearer(wrt http.ResponseWriter, req *http.Request) (*imPr
principal, err := resolveIMHTTPBearer(req) principal, err := resolveIMHTTPBearer(req)
if err != nil { 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) 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) { if imHTTPBearerAuthUnauthorized(err) {
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
} else {
writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "authentication required", nil) 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 nil, false
} }
return principal, true 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) { func resolveIMHTTPBearer(req *http.Request) (*imPrincipal, error) {
principal, _, _, err := resolveIMHTTPBearerToken(req) principal, _, _, err := resolveIMHTTPBearerToken(req)
return principal, err return principal, err
@@ -7437,6 +7619,9 @@ func resolveIMHTTPBearerToken(req *http.Request) (*imPrincipal, []byte, time.Dur
} }
rec, _, err := handler.Authenticate(auth.AuthContext{}, secret) rec, _, err := handler.Authenticate(auth.AuthContext{}, secret)
if err != nil || rec == nil || rec.TenantID.IsZero() || rec.Uid.IsZero() { 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 nil, nil, 0, types.ErrPermissionDenied
} }
return &imPrincipal{TenantID: rec.TenantID, Uid: rec.Uid}, secret, time.Duration(rec.Lifetime), nil 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 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) { func parseIMOptionalDate(raw string, fallback time.Time) (time.Time, error) {
if strings.TrimSpace(raw) == "" { if strings.TrimSpace(raw) == "" {
return time.Date(fallback.UTC().Year(), fallback.UTC().Month(), fallback.UTC().Day(), 0, 0, 0, 0, time.UTC), nil 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 == "" { if value == "" {
return true 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}) writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": field})
return false return false
} }
@@ -7816,6 +8017,26 @@ func validIMMonth(raw string) bool {
return true 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) { func requireIMIdempotencyKey(wrt http.ResponseWriter, req *http.Request) (string, bool) {
key := strings.TrimSpace(req.Header.Get("Idempotency-Key")) key := strings.TrimSpace(req.Header.Get("Idempotency-Key"))
if key == "" { if key == "" {
@@ -8035,7 +8256,6 @@ func mergeGroupListSettingsPayload(payload map[string]any, settings *types.Group
payload["topic"] = settings.Topic payload["topic"] = settings.Topic
payload["name"] = settings.Name payload["name"] = settings.Name
payload["avatar_file_id"] = settings.AvatarFileID payload["avatar_file_id"] = settings.AvatarFileID
payload["member_count"] = settings.MemberCount
payload["updated_at"] = settings.UpdatedAt.UTC().Format(time.RFC3339Nano) payload["updated_at"] = settings.UpdatedAt.UTC().Format(time.RFC3339Nano)
return payload 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 { func groupMemberProfilePayload(topic string, sub *types.Subscription, user *types.User, ctx *types.RelationshipContext) map[string]any {
data := groupMemberPayload(*sub, true, ctx.Remark) data := groupMemberPayload(*sub, true, ctx.Remark)
data["topic"] = topic data["topic"] = topic
@@ -8316,7 +8567,6 @@ func mergeGroupSettingsPayload(payload map[string]any, settings *types.GroupSett
payload["avatar_file_id"] = settings.AvatarFileID payload["avatar_file_id"] = settings.AvatarFileID
payload["description"] = settings.Description payload["description"] = settings.Description
payload["owner"] = settings.OwnerUserID.UserId() payload["owner"] = settings.OwnerUserID.UserId()
payload["member_count"] = settings.MemberCount
payload["status"] = settings.GroupStatus payload["status"] = settings.GroupStatus
payload["version"] = settings.Version payload["version"] = settings.Version
return payload return payload
@@ -8447,6 +8697,9 @@ func groupAnnouncementPayload(ann *types.GroupAnnouncement) map[string]any {
"author": "", "author": "",
"version": 0, "version": 0,
"published_at": "", "published_at": "",
"read_count": 0,
"member_count": 0,
"read_summary": map[string]any{"read_count": 0, "member_count": 0},
} }
} }
return map[string]any{ return map[string]any{
@@ -8455,6 +8708,9 @@ func groupAnnouncementPayload(ann *types.GroupAnnouncement) map[string]any {
"author": ann.AuthorUserID.UserId(), "author": ann.AuthorUserID.UserId(),
"version": ann.Version, "version": ann.Version,
"published_at": ann.PublishedAt.UTC().Format(time.RFC3339Nano), "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,18 +9230,33 @@ func callRecordPayload(record types.CallRecord, viewer types.Uid) map[string]any
peer = record.CallerID peer = record.CallerID
direction = "incoming" 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{ data := map[string]any{
"call_id": record.CallID, "call_id": record.CallID,
"topic": record.Topic, "topic": record.Topic,
"display_topic": displayTopic,
"seq": record.SeqID, "seq": record.SeqID,
"caller_id": record.CallerID.UserId(), "caller_id": record.CallerID.UserId(),
"callee_id": record.CalleeID.UserId(), "callee_id": record.CalleeID.UserId(),
"peer_user_id": peer.UserId(), "peer_user_id": peer.UserId(),
"peer": map[string]any{
"user_id": peer.UserId(),
"topic": displayTopic,
},
"direction": direction, "direction": direction,
"incoming": direction == "incoming",
"call_type": record.CallType, "call_type": record.CallType,
"status": record.Status, "status": record.Status,
"result": record.Status, "result": record.Status,
"duration_ms": record.DurationMS, "duration_ms": record.DurationMS,
"ts": ts.UTC().Format(time.RFC3339Nano),
"started_at": record.StartedAt.UTC().Format(time.RFC3339Nano), "started_at": record.StartedAt.UTC().Format(time.RFC3339Nano),
"created_at": record.CreatedAt.UTC().Format(time.RFC3339Nano), "created_at": record.CreatedAt.UTC().Format(time.RFC3339Nano),
"updated_at": record.UpdatedAt.UTC().Format(time.RFC3339Nano), "updated_at": record.UpdatedAt.UTC().Format(time.RFC3339Nano),
@@ -9682,6 +9953,18 @@ func dailyReportDetailPayload(report *types.DailyReport, viewer types.Uid) map[s
return data 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 { func monthlyReportListPayloads(reports []types.MonthlyReport) []map[string]any {
items := make([]map[string]any, 0, len(reports)) items := make([]map[string]any, 0, len(reports))
for _, report := range reports { for _, report := range reports {
@@ -9840,6 +10123,7 @@ func userPrivacyPayload(settings *types.UserPrivacySettings) map[string]any {
"profile_visibility": settings.ProfileVisibility, "profile_visibility": settings.ProfileVisibility,
"last_seen_visibility": settings.LastSeenVisibility, "last_seen_visibility": settings.LastSeenVisibility,
"read_receipts": settings.ReadReceipts, "read_receipts": settings.ReadReceipts,
"friend_request_verification": settings.FriendRequestVerification,
"group_invite_verification": settings.GroupInviteVerification, "group_invite_verification": settings.GroupInviteVerification,
"version": settings.Version, "version": settings.Version,
"updated_at": settings.UpdatedAt.UTC().Format(time.RFC3339Nano), "updated_at": settings.UpdatedAt.UTC().Format(time.RFC3339Nano),
@@ -9848,6 +10132,7 @@ func userPrivacyPayload(settings *types.UserPrivacySettings) map[string]any {
"profile_visibility": true, "profile_visibility": true,
"last_seen_visibility": true, "last_seen_visibility": true,
"read_receipts": true, "read_receipts": true,
"friend_request_verification": true,
"group_invite_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 { 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 online := false
onlineVisibility := "hidden" 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" onlineVisibility = "visible"
online = redisUserOnline(tenantID, item.Friendship.PeerID, false) online = redisUserOnline(tenantID, item.Friendship.PeerID, false)
} }
data := map[string]any{ data := map[string]any{
"user_id": item.Friendship.PeerID.UserId(), "user_id": item.Friendship.PeerID.UserId(),
"relationship": item.Friendship.Status, "relationship": relationship,
"remark": item.Friendship.Remark, "remark": item.Friendship.Remark,
"topic": item.Friendship.Topic, "topic": item.Friendship.Topic,
"online": online, "online": online,
"online_visibility": onlineVisibility, "online_visibility": onlineVisibility,
"created_at": item.Friendship.CreatedAt.UTC().Format(time.RFC3339Nano), "created_at": item.Friendship.CreatedAt.UTC().Format(time.RFC3339Nano),
"updated_at": item.Friendship.UpdatedAt.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", "display_name_source": "profile",
} }
if item.Friendship.Remark != "" { if item.Friendship.Remark != "" {
@@ -10303,9 +10594,62 @@ func blockListItemPayload(item types.BlockListItem) map[string]any {
return data 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 { func userProfilePayload(user *types.User, ctx *types.RelationshipContext) map[string]any {
data := userSummaryPayload(user, ctx.Remark) data := userSummaryPayload(user, ctx.Remark)
data["username"] = user.Username data["username"] = user.Username
if original := userOriginalDisplayName(user); original != "" {
data["profile_display_name"] = original
data["original_display_name"] = original
}
data["relationship"] = ctx.Relationship data["relationship"] = ctx.Relationship
data["topic"] = ctx.Topic data["topic"] = ctx.Topic
data["capabilities"] = map[string]any{ 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 { func blockedUserProfilePayload(user *types.User, ctx *types.RelationshipContext) map[string]any {
data := userSummaryPayload(user, ctx.Remark) data := userSummaryPayload(user, ctx.Remark)
data["username"] = user.Username data["username"] = user.Username
if original := userOriginalDisplayName(user); original != "" {
data["profile_display_name"] = original
data["original_display_name"] = original
}
data["relationship"] = ctx.Relationship data["relationship"] = ctx.Relationship
data["topic"] = ctx.Topic data["topic"] = ctx.Topic
data["capabilities"] = map[string]any{ data["capabilities"] = map[string]any{
@@ -10372,6 +10720,23 @@ func userSummaryPayload(user *types.User, remark string) map[string]any {
return data 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) { func copyUserPublicSummary(data map[string]any, public map[string]any) {
for _, key := range []string{"display_name", "fn", "name", "nickname"} { for _, key := range []string{"display_name", "fn", "name", "nickname"} {
if value, ok := public[key]; ok && strings.TrimSpace(fmt.Sprint(value)) != "" { 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 { func saveForwardCommentMessage(ctx context.Context, principal *imPrincipal, target, comment, clientMid string) map[string]any {
comment = strings.TrimSpace(comment) comment = strings.TrimSpace(comment)
if principal == nil || 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) now := time.Now().UTC().Round(time.Millisecond)
s.privacy = &types.UserPrivacySettings{ s.privacy = &types.UserPrivacySettings{
UserID: uid, AddMeMethods: map[string]bool{"account": true, "phone": true, "qr": true, "group": true}, 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 return s.privacy, nil
@@ -1110,6 +1110,9 @@ func (s *testProfilesStore) UpdatePrivacy(_ types.TenantID, uid types.Uid, patch
if patch.ReadReceipts != nil { if patch.ReadReceipts != nil {
privacy.ReadReceipts = *patch.ReadReceipts privacy.ReadReceipts = *patch.ReadReceipts
} }
if patch.FriendRequestVerification != nil {
privacy.FriendRequestVerification = *patch.FriendRequestVerification
}
if patch.GroupInviteVerification != nil { if patch.GroupInviteVerification != nil {
privacy.GroupInviteVerification = *patch.GroupInviteVerification privacy.GroupInviteVerification = *patch.GroupInviteVerification
} }
@@ -3900,7 +3903,7 @@ func TestServeIMMePrivacy(t *testing.T) {
store.Profiles = profiles store.Profiles = profiles
defer func() { store.Profiles = oldProfiles }() 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") req.Header.Set("Authorization", "Bearer token-1")
rec := httptest.NewRecorder() rec := httptest.NewRecorder()
serveIMMePrivacy(rec, req) serveIMMePrivacy(rec, req)
@@ -3911,8 +3914,8 @@ func TestServeIMMePrivacy(t *testing.T) {
data := decodeIMHTTPResponse(t, rec).Data.(map[string]any) data := decodeIMHTTPResponse(t, rec).Data.(map[string]any)
methods := data["add_me_methods"].(map[string]any) methods := data["add_me_methods"].(map[string]any)
editable := data["editable"].(map[string]any) editable := data["editable"].(map[string]any)
if methods["phone"] != false || data["read_receipts"] != false || data["group_invite_verification"] != true || if methods["phone"] != false || data["read_receipts"] != false || data["friend_request_verification"] != false || data["group_invite_verification"] != true ||
editable["group_invite_verification"] != true || data["version"].(float64) != 2 { editable["friend_request_verification"] != true || editable["group_invite_verification"] != true || data["version"].(float64) != 2 {
t.Fatalf("unexpected privacy: %#v", data) t.Fatalf("unexpected privacy: %#v", data)
} }
} }
@@ -4574,7 +4577,7 @@ func TestServeIMGroupList(t *testing.T) {
} }
data := decodeIMHTTPResponse(t, rec).Data.(map[string]any) data := decodeIMHTTPResponse(t, rec).Data.(map[string]any)
items := data["items"].([]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) 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) ctrl := gomock.NewController(t)
defer ctrl.Finish() defer ctrl.Finish()
mockStore := expectIMBearer(t, ctrl, "token-1", 7, 42) mockStore := expectIMBearer(t, ctrl, "token-1", 7, 42)
@@ -6864,8 +6867,9 @@ func TestServeIMMessageForwardKafkaPublishUnknownReturnsPending(t *testing.T) {
t.Fatalf("unexpected forward response: %#v", data) t.Fatalf("unexpected forward response: %#v", data)
} }
got := items[0].(map[string]any) got := items[0].(map[string]any)
if got["status"] != "pending" || got["reason"] != "MESSAGE_PERSIST_PENDING" || got["seq"].(float64) != 41 { if got["status"] != "uncertain" || got["reason"] != "MESSAGE_PERSIST_RESULT_UNKNOWN" ||
t.Fatalf("unexpected pending item: %#v", got) 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)) { if got["client_mid"] != "fwd_"+stableIMEventID("forward", "idem-forward-1", "separate", targetTopic, "grpSource", int64(14)) {
t.Fatalf("unexpected stable client mid: %#v", got) t.Fatalf("unexpected stable client mid: %#v", got)
@@ -6875,7 +6879,7 @@ func TestServeIMMessageForwardKafkaPublishUnknownReturnsPending(t *testing.T) {
} }
select { select {
case delivered := <-globals.hub.routeSrv: 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: 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 // client complete the alias->canonical rename with a 200 reply instead of
// a 304 that would leave the client-side alias topic object unattached // a 304 that would leave the client-side alias topic object unattached
// (publishes on it then fail with "Cannot publish on inactive topic"). // (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.queueOut(InfoAlreadySubscribed(msg.Id, msg.Original, msg.Timestamp))
s.inflightReqs.Done() s.inflightReqs.Done()
} else { } 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 // Leave/Unsubscribe a topic
func (s *Session) leave(msg *ClientComMessage) { func (s *Session) leave(msg *ClientComMessage) {
// Expand topic name // Expand topic name
+65 -4
View File
@@ -2,6 +2,9 @@ package store
import ( import (
"context" "context"
"encoding/json"
"strconv"
"strings"
"testing" "testing"
"time" "time"
@@ -12,6 +15,7 @@ type testGroupMessageRuntime struct {
claims map[string]*groupMessageUniqueRecord claims map[string]*groupMessageUniqueRecord
seqs map[string]int64 seqs map[string]int64
shortStates map[string]bool shortStates map[string]bool
objects map[string][]byte
deletedPrefixes []string deletedPrefixes []string
deletedObjects []string deletedObjects []string
} }
@@ -21,6 +25,7 @@ func newTestGroupMessageRuntime() *testGroupMessageRuntime {
claims: make(map[string]*groupMessageUniqueRecord), claims: make(map[string]*groupMessageUniqueRecord),
seqs: make(map[string]int64), seqs: make(map[string]int64),
shortStates: make(map[string]bool), shortStates: make(map[string]bool),
objects: make(map[string][]byte),
} }
} }
@@ -145,21 +150,37 @@ func (r *testGroupMessageRuntime) StartRealtimeFanoutSubscriber(context.Context,
return nil 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 return nil
} }
func (r *testGroupMessageRuntime) GetCachedObject(types.TenantID, string, string, any) (bool, error) { 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 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) r.deletedObjects = append(r.deletedObjects, namespace+":"+id)
delete(r.objects, strconv.FormatInt(int64(tenantID), 10)+":"+namespace+":"+id)
return nil 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) 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 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) { func TestRealtimeFanoutDedupClaimIDIsNodeScoped(t *testing.T) {
tenantID := types.TenantID(7) tenantID := types.TenantID(7)
dedupKey := "data:p2pABC:42" dedupKey := "data:p2pABC:42"
+56 -9
View File
@@ -1760,6 +1760,9 @@ func (groupBusinessMapper) GetSettings(tenantID types.TenantID, topic string) (*
var cached types.GroupSettings var cached types.GroupSettings
if ok, err := GetCachedObject(tenantID, cacheNamespaceGroupSettings, topic, &cached); err != nil { if ok, err := GetCachedObject(tenantID, cacheNamespaceGroupSettings, topic, &cached); err != nil {
logs.Warn.Printf("cache group settings[%s]: read failed: %v", topic, err) 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 { } else if ok {
return &cached, nil return &cached, nil
} }
@@ -1874,15 +1877,51 @@ func (groupBusinessMapper) ListMembers(tenantID types.TenantID, topic string, op
if opts.Limit <= 0 || opts.Limit > 100 { if opts.Limit <= 0 || opts.Limit > 100 {
opts.Limit = 30 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) tenantAdp, err := tenantBusinessAdapter(tenantID)
if err != nil { if err != nil {
return nil, "", err 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) { type groupMemberPageCache struct {
if strings.TrimSpace(key) == "" || payload == nil { 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 return
} }
if err := CacheObject(tenantID, cacheNamespaceGroupPage, strings.TrimSpace(key), payload); err != nil { 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) == "" { 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 { 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) logs.Warn.Printf("cache group member page[%s]: read failed: %v", key, err)
} else if ok { } else if ok {
return payload, true return payload, true
} }
return nil, false return groupMemberPageCache{}, false
} }
func invalidateGroupMemberPages(tenantID types.TenantID, topic string) { func invalidateGroupMemberPages(tenantID types.TenantID, topic string) {
@@ -2113,7 +2152,11 @@ func (groupBusinessMapper) UpsertMute(tenantID types.TenantID, mute *types.Group
if err != nil { if err != nil {
return nil, err 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 { 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 { if err != nil {
return err 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) { 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"` CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"` UpdatedAt time.Time `db:"updated_at"`
AuthorUserIDRaw int64 `db:"author_user_id"` AuthorUserIDRaw int64 `db:"author_user_id"`
ReadCount int `db:"read_count"`
MemberCount int `db:"member_count"`
} }
type GroupMute struct { type GroupMute struct {
+2
View File
@@ -34,6 +34,7 @@ type UserPrivacySettings struct {
ProfileVisibility string ProfileVisibility string
LastSeenVisibility string LastSeenVisibility string
ReadReceipts bool ReadReceipts bool
FriendRequestVerification bool
GroupInviteVerification bool GroupInviteVerification bool
Version int Version int
CreatedAt time.Time CreatedAt time.Time
@@ -45,6 +46,7 @@ type UserPrivacyPatch struct {
ProfileVisibility *string ProfileVisibility *string
LastSeenVisibility *string LastSeenVisibility *string
ReadReceipts *bool ReadReceipts *bool
FriendRequestVerification *bool
GroupInviteVerification *bool GroupInviteVerification *bool
Version int Version int
} }