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

fix: improve conversation and report stability

Browse Source
This commit is contained in:
richjohnrambo
2026-08-26 14:12:58 +08:00
parent dc80aa5104
commit a0fc553c6f
6 changed files with 138 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/db/mysql/adapter.go
+44 -3
View File
@@ -2541,7 +2541,35 @@ func createSubscription(tx *sqlx.Tx, sub *t.Subscription, undelete bool) error {
jpriv := common.ToJSON(sub.Private)
decoded_uid := store.DecodeUid(t.ParseUid(sub.User))
_, err := tx.Exec(
existing := struct {
ID int64 `db:"id"`
DeletedAt sql.NullTime `db:"deletedat"`
}{}
err := tx.Get(&existing, "SELECT id,deletedat FROM im_subscriptions WHERE topic=? AND userid=? ORDER BY deletedat IS NULL DESC,id ASC LIMIT 1",
sub.Topic, decoded_uid)
if err != nil && err != sql.ErrNoRows {
return err
}
if err == nil {
if existing.DeletedAt.Valid {
_, err = tx.Exec("UPDATE im_subscriptions SET createdat=?,updatedat=?,deletedat=NULL,modeWant=?,modeGiven=?,"+
"delid=0,recvseqid=0,readseqid=0,private=? WHERE id=?",
sub.CreatedAt, sub.UpdatedAt, sub.ModeWant.String(), sub.ModeGiven.String(), jpriv, existing.ID)
} else {
_, err = tx.Exec("UPDATE im_subscriptions SET updatedat=?,modeWant=?,modeGiven=?,private=? WHERE id=?",
sub.UpdatedAt, sub.ModeWant.String(), sub.ModeGiven.String(), jpriv, existing.ID)
}
if err != nil {
return err
}
if isOwner {
_, err = tx.Exec("UPDATE im_topics SET owner=? WHERE name=?", decoded_uid, sub.Topic)
}
return err
}
_, err = tx.Exec(
"INSERT INTO im_subscriptions(createdat,updatedat,deletedat,userid,topic,modeWant,modeGiven,private) "+
"VALUES(?,?,NULL,?,?,?,?,?)",
sub.CreatedAt, sub.UpdatedAt, decoded_uid, sub.Topic, sub.ModeWant.String(), sub.ModeGiven.String(), jpriv)
@@ -2596,8 +2624,18 @@ func (a *adapter) TopicCreateP2P(initiator, invited *t.Subscription) error {
topic := &t.Topic{ObjHeader: t.ObjHeader{Id: initiator.Topic}}
topic.ObjHeader.MergeTimes(&initiator.ObjHeader)
topic.TouchedAt = initiator.GetTouchedAt()
err = a.topicCreate(tx, topic)
if err != nil {
var existingTopicID int64
err = tx.Get(&existingTopicID, "SELECT id FROM im_topics WHERE name=? ORDER BY id ASC LIMIT 1", topic.Id)
if err != nil && err != sql.ErrNoRows {
return err
}
if err == sql.ErrNoRows {
err = a.topicCreate(tx, topic)
if err != nil {
return err
}
} else if _, err = tx.Exec("UPDATE im_topics SET updatedat=?,touchedat=?,state=? WHERE id=?",
topic.UpdatedAt, topic.TouchedAt, t.StateOK, existingTopicID); err != nil {
return err
}
@@ -3624,6 +3662,9 @@ func (a *adapter) MessageSave(msg *t.Message) error {
"INSERT INTO im_messages(createdAt,updatedAt,seqid,topic,`from`,client_msg_id,version,head,content) VALUES(?,?,?,?,?,?,?,?,?)",
msg.CreatedAt, msg.UpdatedAt, msg.SeqId, msg.Topic,
store.DecodeUid(t.ParseUid(msg.From)), clientMsgID, messageVersion(msg.Version), msg.Head, common.ToJSON(msg.Content))
if isDupe(err) {
return t.ErrDuplicate
}
if err == nil {
id, _ := res.LastInsertId()
// Replacing ID given by store by ID given by the DB.
server/db/mysql/tenant_adapter.go
+14 -1
View File
@@ -682,7 +682,17 @@ func (a *adapter) TenantTopicCreateP2P(id t.TenantID, one, two *t.Subscription)
topic := &t.Topic{TenantID: id, ObjHeader: t.ObjHeader{Id: one.Topic}}
topic.ObjHeader.MergeTimes(&one.ObjHeader)
topic.TouchedAt = one.GetTouchedAt()
if err = tenantTopicCreate(tx, topic); err != nil {
var existingTopicID int64
err = tx.Get(&existingTopicID, "SELECT id FROM im_topics WHERE name=? ORDER BY id ASC LIMIT 1", topic.Id)
if err != nil && err != sql.ErrNoRows {
return err
}
if err == sql.ErrNoRows {
if err = tenantTopicCreate(tx, topic); err != nil {
return err
}
} else if _, err = tx.ExecContext(ctx, "UPDATE im_topics SET updatedat=?,touchedat=?,state=? WHERE id=?",
topic.UpdatedAt, topic.TouchedAt, t.StateOK, existingTopicID); err != nil {
return err
}
return tx.Commit()
@@ -1757,6 +1767,9 @@ func (a *adapter) TenantMessageSave(id t.TenantID, msg *t.Message) error {
res, err := a.db.ExecContext(ctx,
"INSERT INTO im_messages(tenant_id,createdat,updatedat,seqid,topic,`from`,client_msg_id,version,head,content) VALUES(?,?,?,?,?,?,?,?,?,?)",
id, msg.CreatedAt, msg.UpdatedAt, msg.SeqId, msg.Topic, decodeUID(t.ParseUid(msg.From)), clientMsgID, messageVersion(msg.Version), msg.Head, common.ToJSON(msg.Content))
if isDupe(err) {
return t.ErrDuplicate
}
if err == nil {
lastID, _ := res.LastInsertId()
msg.SetUid(t.Uid(lastID))
server/im_http.go
+20 -8
View File
@@ -68,6 +68,7 @@ const (
imQRCodeMaxExpiresInSeconds = 7 * 86400
imPublicUploadMaxBytes = 10 << 20
groupMaxAdminCount = 10
monthlyReportPreviewMaxRunes = 120
)
type imHTTPResponse struct {
@@ -7206,14 +7207,16 @@ func monthlyReportListPayload(report types.MonthlyReport) map[string]any {
action = "edit"
}
data := map[string]any{
"report_id": report.ReportID,
"author_id": report.AuthorID.UserId(),
"month": report.Month,
"status": report.Status,
"read_state": report.ReadState,
"version": report.Version,
"action": action,
"updated_at": report.UpdatedAt.UTC().Format(time.RFC3339Nano),
"report_id": report.ReportID,
"author_id": report.AuthorID.UserId(),
"month": report.Month,
"work_summary_short": monthlyReportPreview(report.WorkSummary),
"next_plan_short": monthlyReportPreview(report.NextPlan),
"status": report.Status,
"read_state": report.ReadState,
"version": report.Version,
"action": action,
"updated_at": report.UpdatedAt.UTC().Format(time.RFC3339Nano),
}
if report.SubmittedAt != nil {
data["submitted_at"] = report.SubmittedAt.UTC().Format(time.RFC3339Nano)
@@ -7221,6 +7224,15 @@ func monthlyReportListPayload(report types.MonthlyReport) map[string]any {
return data
}
func monthlyReportPreview(text string) string {
text = strings.Join(strings.Fields(text), " ")
runes := []rune(text)
if len(runes) <= monthlyReportPreviewMaxRunes {
return text
}
return string(runes[:monthlyReportPreviewMaxRunes]) + "..."
}
func monthlyReportDetailPayload(report *types.MonthlyReport, viewer types.Uid) map[string]any {
data := monthlyReportListPayload(*report)
data["work_summary"] = report.WorkSummary
server/im_http_test.go
+40
View File
@@ -2442,6 +2442,46 @@ func TestServeIMMonthlyReportsCreate(t *testing.T) {
}
}
func TestServeIMMonthlyReportsListIncludesPreviews(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
expectIMBearer(t, ctrl, "token-1", 7, 42)
oldMonthlyReports := store.MonthlyReports
now := time.Now().UTC().Round(time.Millisecond)
reports := &testMonthlyReportsStore{report: &types.MonthlyReport{
ReportID: "mr_test",
AuthorID: 42,
Month: "2026-08",
WorkSummary: strings.Repeat("工作", 80),
NextPlan: " 下月计划一 \n 下月计划二 ",
Status: types.DailyReportStatusDraft,
ReadState: "unread",
Version: 1,
CreatedAt: now,
UpdatedAt: now,
}}
store.MonthlyReports = reports
defer func() { store.MonthlyReports = oldMonthlyReports }()
req := httptest.NewRequest(http.MethodGet, "/v1/im/reports/monthly?scope=mine", nil)
req.Header.Set("Authorization", "Bearer token-1")
rec := httptest.NewRecorder()
serveIMMonthlyReports(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
}
data := decodeIMHTTPResponse(t, rec).Data.(map[string]any)
items := data["items"].([]any)
item := items[0].(map[string]any)
if item["work_summary_short"] != strings.Repeat("工作", 60)+"..." {
t.Fatalf("unexpected work summary preview: %#v", item["work_summary_short"])
}
if item["next_plan_short"] != "下月计划一 下月计划二" {
t.Fatalf("unexpected next plan preview: %#v", item["next_plan_short"])
}
}
func TestServeIMMonthlyReportDraftAndSubmit(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
server/store/store.go
+18
View File
@@ -3136,6 +3136,9 @@ func (messagesMapper) Save(tenantID types.TenantID, msg *types.Message, attachme
err = tenantAdp.TenantGroupMessageSave(tenantID, msg)
} else {
err = tenantAdp.TenantMessageSave(tenantID, msg)
if err == types.ErrDuplicate && clientMsgID == "" {
err = retryTenantMessageSaveAfterSeqConflict(tenantAdp, tenantID, msg)
}
}
if err != nil {
if isGroupMessage && err == types.ErrDuplicate && clientMsgID != "" {
@@ -3217,6 +3220,21 @@ func (messagesMapper) Save(tenantID types.TenantID, msg *types.Message, attachme
return &types.MessageSaveResult{SeqId: msg.SeqId, MarkedReadBySender: markedReadBySender}, nil
}
func retryTenantMessageSaveAfterSeqConflict(tenantAdp adapter.TenantBusinessAdapter, tenantID types.TenantID, msg *types.Message) error {
latest, err := tenantAdp.TenantMessageGetAll(tenantID, msg.Topic, types.ZeroUid, &types.QueryOpt{Before: 1 << 62, Limit: 1})
if err != nil {
return err
}
if len(latest) == 0 || latest[0].SeqId < msg.SeqId {
return types.ErrDuplicate
}
msg.SeqId = latest[0].SeqId + 1
if err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, msg.Topic, msg); err != nil {
return err
}
return tenantAdp.TenantMessageSave(tenantID, msg)
}
func normalizeClientMsgID(head types.KVMap) (string, error) {
if head == nil {
return "", nil