diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index 6a9eb65..b9fea07 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -1055,6 +1055,9 @@ func (a *adapter) UpgradeDb() error { if err := a.ensureMessageTopicSeqIndex(); err != nil { return err } + if err := a.ensureTopicSeqMatchesMessages(); err != nil { + return err + } if a.version == adpVersion { return nil @@ -1080,6 +1083,61 @@ func (a *adapter) ensureMessageTopicSeqIndex() error { return err } +func (a *adapter) ensureTopicSeqMatchesMessages() error { + ctx := context.Background() + conn, err := a.db.Connx(ctx) + if err != nil { + return err + } + defer conn.Close() + + topicsScoped, err := identityColumnExists(ctx, conn, "im_topics", "tenant_id") + if err != nil { + return err + } + messagesScoped, err := identityColumnExists(ctx, conn, "im_messages", "tenant_id") + if err != nil { + return err + } + if topicsScoped && messagesScoped { + if _, err = conn.ExecContext(ctx, `UPDATE im_topics AS t + JOIN (SELECT tenant_id,topic,MAX(seqid) AS max_seq FROM im_messages GROUP BY tenant_id,topic) AS m + ON m.tenant_id=t.tenant_id AND m.topic COLLATE utf8mb4_unicode_ci=t.name + SET t.seqid=GREATEST(t.seqid,m.max_seq) + WHERE t.seqid