feat: add im home and message action flows

This commit is contained in:
richjohnrambo
2026-08-12 10:36:41 +08:00
parent d239e37f56
commit 5bd35f97a6
21 changed files with 3087 additions and 61 deletions
+480 -8
View File
@@ -4,7 +4,10 @@
package mysql
import (
"context"
"crypto/sha256"
"database/sql"
"fmt"
"sort"
"strconv"
"strings"
@@ -1344,9 +1347,10 @@ func (a *adapter) TenantMessageSave(id t.TenantID, msg *t.Message) error {
if cancel != nil {
defer cancel()
}
clientMsgID := sql.NullString{String: msg.ClientMsgID, Valid: msg.ClientMsgID != ""}
res, err := a.db.ExecContext(ctx,
"INSERT INTO messages(tenant_id,createdat,updatedat,seqid,topic,`from`,head,content) VALUES(?,?,?,?,?,?,?,?)",
id, msg.CreatedAt, msg.UpdatedAt, msg.SeqId, msg.Topic, decodeUID(t.ParseUid(msg.From)), msg.Head, common.ToJSON(msg.Content))
"INSERT INTO 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 err == nil {
lastID, _ := res.LastInsertId()
msg.SetUid(t.Uid(lastID))
@@ -1354,6 +1358,128 @@ func (a *adapter) TenantMessageSave(id t.TenantID, msg *t.Message) error {
return err
}
func (a *adapter) TenantMessageGetByClientID(id t.TenantID, from t.Uid, clientMsgID string) (*t.Message, error) {
if err := validTenant(id); err != nil || from.IsZero() || clientMsgID == "" {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
var msg t.Message
err := a.db.QueryRowxContext(ctx,
"SELECT id,tenant_id,createdat,updatedat,deletedat,delid,seqid,topic,`from`,client_msg_id,version,recalled_at,recalled_by,head,content "+
"FROM messages WHERE tenant_id=? AND `from`=? AND client_msg_id=?",
id, decodeUID(from), clientMsgID).StructScan(&msg)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
msg.From = common.EncodeUidString(msg.From).String()
msg.Content = common.FromJSON(msg.Content)
return &msg, nil
}
func (a *adapter) TenantMessageGetOne(id t.TenantID, topic string, seq int) (*t.Message, error) {
if err := validTenant(id); err != nil || topic == "" || seq <= 0 {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
msg, err := tenantMessageGetOne(a.db, ctx, id, topic, seq, "")
if err == sql.ErrNoRows {
return nil, t.ErrNotFound
}
return msg, err
}
func (a *adapter) TenantMessageRecall(id t.TenantID, topic string, seq int, actor t.Uid, expectedVersion int, recalledAt time.Time) (*t.Message, error) {
if err := validTenant(id); err != nil || topic == "" || seq <= 0 || actor.IsZero() || expectedVersion <= 0 {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContextForTx()
if cancel != nil {
defer cancel()
}
tx, err := a.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer func() {
if err != nil {
tx.Rollback()
}
}()
msg, err := tenantMessageGetOne(tx, ctx, id, topic, seq, " FOR UPDATE")
if err != nil {
if err == sql.ErrNoRows {
return nil, t.ErrNotFound
}
return nil, err
}
if msg.Version != expectedVersion || msg.RecalledAt != nil || msg.DeletedAt != nil {
return nil, t.ErrDuplicate
}
_, err = tx.ExecContext(ctx,
"UPDATE messages SET updatedat=?,version=version+1,recalled_at=?,recalled_by=?,head=NULL,content=NULL WHERE tenant_id=? AND topic=? AND seqid=? AND version=?",
recalledAt, recalledAt, decodeUID(actor), id, topic, seq, expectedVersion)
if err != nil {
return nil, err
}
if err = tx.Commit(); err != nil {
return nil, err
}
return a.TenantMessageGetOne(id, topic, seq)
}
func (a *adapter) TenantMessageEdit(id t.TenantID, topic string, seq int, actor t.Uid, expectedVersion int, head t.KVMap, content any, reason string, editedAt time.Time) (*t.Message, error) {
if err := validTenant(id); err != nil || topic == "" || seq <= 0 || actor.IsZero() || expectedVersion <= 0 {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContextForTx()
if cancel != nil {
defer cancel()
}
tx, err := a.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer func() {
if err != nil {
tx.Rollback()
}
}()
msg, err := tenantMessageGetOne(tx, ctx, id, topic, seq, " FOR UPDATE")
if err != nil {
if err == sql.ErrNoRows {
return nil, t.ErrNotFound
}
return nil, err
}
if msg.Version != expectedVersion || msg.RecalledAt != nil || msg.DeletedAt != nil {
return nil, t.ErrDuplicate
}
if _, err = tx.ExecContext(ctx,
"INSERT INTO im_message_revisions(tenant_id,createdat,updatedat,topic,seqid,version,actor,head,content,reason) VALUES(?,?,?,?,?,?,?,?,?,?)",
id, editedAt, editedAt, topic, seq, msg.Version, decodeUID(actor), msg.Head, common.ToJSON(msg.Content), strings.TrimSpace(reason)); err != nil {
return nil, err
}
_, err = tx.ExecContext(ctx,
"UPDATE messages SET updatedat=?,version=version+1,head=?,content=? WHERE tenant_id=? AND topic=? AND seqid=? AND version=?",
editedAt, head, common.ToJSON(content), id, topic, seq, expectedVersion)
if err != nil {
return nil, err
}
if err = tx.Commit(); err != nil {
return nil, err
}
return a.TenantMessageGetOne(id, topic, seq)
}
func (a *adapter) TenantMessageGetAll(id t.TenantID, topic string, uid t.Uid, opts *t.QueryOpt) ([]t.Message, error) {
if err := validTenant(id); err != nil || topic == "" {
return nil, t.ErrMalformed
@@ -1390,7 +1516,7 @@ func (a *adapter) TenantMessageGetAll(id t.TenantID, topic string, uid t.Uid, op
defer cancel()
}
rows, err := a.db.QueryxContext(ctx,
"SELECT m.tenant_id,m.createdat,m.updatedat,m.deletedat,m.delid,m.seqid,m.topic,m.`from`,m.head,m.content "+
"SELECT m.tenant_id,m.createdat,m.updatedat,m.deletedat,m.delid,m.seqid,m.topic,m.`from`,m.client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content "+
"FROM messages AS m LEFT JOIN dellog AS d ON d.tenant_id=m.tenant_id AND d.topic=m.topic "+
"AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=? WHERE m.tenant_id=? AND m.delid=0 AND m.topic=? "+
seq+" AND d.deletedfor IS NULL ORDER BY m.seqid DESC LIMIT ?",
@@ -1412,6 +1538,34 @@ func (a *adapter) TenantMessageGetAll(id t.TenantID, topic string, uid t.Uid, op
return msgs, rows.Err()
}
func tenantMessageGetOne(q sqlx.QueryerContext, ctx context.Context, tenantID t.TenantID, topic string, seq int, suffix string) (*t.Message, error) {
row := struct {
t.Message
RecalledBy sql.NullInt64 `db:"recalled_by"`
}{}
err := sqlx.GetContext(ctx, q, &row,
"SELECT id,tenant_id,createdat,updatedat,deletedat,delid,seqid,topic,`from`,client_msg_id,version,recalled_at,recalled_by,head,content "+
"FROM messages WHERE tenant_id=? AND topic=? AND seqid=?"+suffix,
tenantID, topic, seq)
if err != nil {
return nil, err
}
msg := row.Message
msg.From = common.EncodeUidString(msg.From).String()
if row.RecalledBy.Valid {
msg.RecalledBy = common.EncodeUidString(strconv.FormatInt(row.RecalledBy.Int64, 10)).String()
}
msg.Content = common.FromJSON(msg.Content)
return &msg, nil
}
func messageVersion(version int) int {
if version <= 0 {
return 1
}
return version
}
func (a *adapter) TenantMessageDeleteList(id t.TenantID, topic string, del *t.DelMessage) error {
if err := validTenant(id); err != nil || topic == "" || (del != nil && del.TenantID != id) {
return t.ErrMalformed
@@ -1867,10 +2021,10 @@ func (a *adapter) TenantVerificationCreate(id t.TenantID, vc *t.VerificationCode
vc.UpdatedAt = now
_, err = tx.ExecContext(ctx,
`INSERT INTO im_verification_codes
(tenant_id,challenge_id,scene,phone_hash,phone_mask,target_user,code_hash,state,attempts,max_attempts,
(tenant_id,challenge_id,scene,action,phone_hash,phone_mask,target_user,code_hash,state,attempts,max_attempts,
expires_at,resend_after,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
id, vc.ChallengeID, vc.Scene, vc.PhoneHash, vc.PhoneMask, nullableUID(vc.TargetUser), vc.CodeHash, vc.State, vc.Attempts,
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
id, vc.ChallengeID, vc.Scene, vc.Action, vc.PhoneHash, vc.PhoneMask, nullableUID(vc.TargetUser), vc.CodeHash, vc.State, vc.Attempts,
vc.MaxAttempts, vc.ExpiresAt, vc.ResendAfter, vc.CreatedAt, vc.UpdatedAt)
if isDupe(err) {
return t.ErrDuplicate
@@ -1891,7 +2045,7 @@ func (a *adapter) TenantVerificationGet(id t.TenantID, challengeID string) (*t.V
}
var vc t.VerificationCode
err := a.db.GetContext(ctx, &vc,
`SELECT id,tenant_id,challenge_id,scene,phone_hash,phone_mask,IFNULL(target_user,0) AS target_user,code_hash,verification_token_hash,
`SELECT id,tenant_id,challenge_id,scene,action,phone_hash,phone_mask,IFNULL(target_user,0) AS target_user,code_hash,verification_token_hash,
state,attempts,max_attempts,expires_at,resend_after,verified_at,consumed_at,created_at,updated_at
FROM im_verification_codes
WHERE tenant_id=? AND challenge_id=? LIMIT 1`, id, challengeID)
@@ -1914,7 +2068,7 @@ func (a *adapter) TenantVerificationGetActive(id t.TenantID, scene, phoneHash st
}
var vc t.VerificationCode
err := a.db.GetContext(ctx, &vc,
`SELECT id,tenant_id,challenge_id,scene,phone_hash,phone_mask,IFNULL(target_user,0) AS target_user,code_hash,verification_token_hash,
`SELECT id,tenant_id,challenge_id,scene,action,phone_hash,phone_mask,IFNULL(target_user,0) AS target_user,code_hash,verification_token_hash,
state,attempts,max_attempts,expires_at,resend_after,verified_at,consumed_at,created_at,updated_at
FROM im_verification_codes
WHERE tenant_id=? AND scene=? AND phone_hash=? AND state=? AND expires_at>?
@@ -2058,9 +2212,327 @@ func (a *adapter) TenantUserSecurityBumpAuthVersion(id t.TenantID, uid t.Uid, up
return version, nil
}
func (a *adapter) TenantRealNameGet(id t.TenantID, uid t.Uid) (*t.RealNameVerification, error) {
if err := validTenant(id); err != nil || uid.IsZero() {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
var rv t.RealNameVerification
err := a.db.GetContext(ctx, &rv,
`SELECT id,tenant_id,user_id,submission_id,status,name_cipher,id_type,id_number_cipher,id_number_hash,
masked_name,masked_id_number,review_reason,verified_at,created_at,updated_at
FROM im_realname_verifications
WHERE tenant_id=? AND user_id=?
ORDER BY created_at DESC,id DESC LIMIT 1`,
id, decodeUID(uid))
if err == sql.ErrNoRows {
return nil, t.ErrNotFound
}
if err != nil {
return nil, err
}
return &rv, nil
}
func (a *adapter) TenantRealNameCreate(id t.TenantID, rv *t.RealNameVerification) error {
if err := validTenant(id); err != nil || rv == nil || rv.TenantID != id || rv.UserID.IsZero() ||
rv.SubmissionID == "" || rv.Status == "" || rv.NameCipher == "" || rv.IDType == "" ||
rv.IDNumberCipher == "" || rv.IDNumberHash == "" || rv.MaskedName == "" || rv.MaskedIDNumber == "" {
return t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
now := t.TimeNow()
rv.CreatedAt = now
rv.UpdatedAt = now
_, err := a.db.ExecContext(ctx,
`INSERT INTO im_realname_verifications
(tenant_id,user_id,submission_id,status,name_cipher,id_type,id_number_cipher,id_number_hash,
masked_name,masked_id_number,review_reason,verified_at,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
id, decodeUID(rv.UserID), rv.SubmissionID, rv.Status, rv.NameCipher, rv.IDType, rv.IDNumberCipher, rv.IDNumberHash,
rv.MaskedName, rv.MaskedIDNumber, rv.ReviewReason, rv.VerifiedAt, rv.CreatedAt, rv.UpdatedAt)
if isDupe(err) {
return t.ErrDuplicate
}
return err
}
func (a *adapter) TenantActionTokenCreate(id t.TenantID, token *t.ActionToken) error {
if err := validTenant(id); err != nil || token == nil || token.TenantID != id || token.UserID.IsZero() ||
token.TokenID == "" || token.Action == "" || token.TokenHash == "" || token.ExpiresAt.IsZero() {
return t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
now := t.TimeNow()
token.CreatedAt = now
token.UpdatedAt = now
_, err := a.db.ExecContext(ctx,
`INSERT INTO im_action_tokens
(tenant_id,user_id,token_id,action,target,token_hash,state,expires_at,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?)`,
id, decodeUID(token.UserID), token.TokenID, token.Action, token.Target, token.TokenHash, token.State,
token.ExpiresAt, token.CreatedAt, token.UpdatedAt)
if isDupe(err) {
return t.ErrDuplicate
}
return err
}
func (a *adapter) TenantActionTokenConsume(id t.TenantID, uid t.Uid, token, action, target string, now time.Time) (*t.ActionToken, error) {
if err := validTenant(id); err != nil || uid.IsZero() || token == "" || action == "" {
return nil, t.ErrMalformed
}
if now.IsZero() {
now = t.TimeNow()
}
tokenID, secret, ok := strings.Cut(token, ".")
if !ok || tokenID == "" || secret == "" {
return nil, t.ErrMalformed
}
tokenHash := securityTokenHash(id, uid, action, target, tokenID, secret)
ctx, cancel := a.getContextForTx()
if cancel != nil {
defer cancel()
}
tx, err := a.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer func() {
if err != nil {
tx.Rollback()
}
}()
var at t.ActionToken
err = tx.GetContext(ctx, &at,
`SELECT id,tenant_id,user_id,token_id,action,target,token_hash,state,expires_at,consumed_at,created_at,updated_at
FROM im_action_tokens
WHERE tenant_id=? AND user_id=? AND token_id=? AND action=? AND target=? LIMIT 1`,
id, decodeUID(uid), tokenID, action, target)
if err == sql.ErrNoRows {
return nil, t.ErrNotFound
}
if err != nil {
return nil, err
}
if at.State != 0 || at.TokenHash != tokenHash || !now.Before(at.ExpiresAt) {
return nil, t.ErrNotFound
}
res, err := tx.ExecContext(ctx,
`UPDATE im_action_tokens SET state=1,consumed_at=?,updated_at=?
WHERE tenant_id=? AND user_id=? AND token_id=? AND action=? AND target=? AND state=0 AND expires_at>?`,
now, now, id, decodeUID(uid), tokenID, action, target, now)
if err != nil {
return nil, err
}
if rows, _ := res.RowsAffected(); rows == 0 {
return nil, t.ErrNotFound
}
if err = tx.Commit(); err != nil {
return nil, err
}
return &at, nil
}
func (a *adapter) TenantIndependentPasswordGet(id t.TenantID, uid t.Uid, purpose string) (*t.IndependentPassword, error) {
if err := validTenant(id); err != nil || uid.IsZero() || purpose == "" {
return nil, t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
var pass t.IndependentPassword
err := a.db.GetContext(ctx, &pass,
`SELECT tenant_id,user_id,purpose,password_hash,failed_attempts,locked_until,created_at,updated_at
FROM im_youth_mode_settings
WHERE tenant_id=? AND user_id=? AND purpose=?`,
id, decodeUID(uid), purpose)
if err == sql.ErrNoRows {
return nil, t.ErrNotFound
}
if err != nil {
return nil, err
}
return &pass, nil
}
func (a *adapter) TenantIndependentPasswordUpsert(id t.TenantID, pass *t.IndependentPassword) error {
if err := validTenant(id); err != nil || pass == nil || pass.TenantID != id || pass.UserID.IsZero() ||
pass.Purpose == "" || len(pass.PasswordHash) == 0 {
return t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
now := t.TimeNow()
pass.CreatedAt = now
pass.UpdatedAt = now
_, err := a.db.ExecContext(ctx,
`INSERT INTO im_youth_mode_settings
(tenant_id,user_id,purpose,password_hash,failed_attempts,locked_until,created_at,updated_at)
VALUES(?,?,?,?,0,NULL,?,?)
ON DUPLICATE KEY UPDATE password_hash=VALUES(password_hash),failed_attempts=0,locked_until=NULL,updated_at=VALUES(updated_at)`,
id, decodeUID(pass.UserID), pass.Purpose, pass.PasswordHash, pass.CreatedAt, pass.UpdatedAt)
return err
}
func (a *adapter) TenantIndependentPasswordUpdateAttempts(id t.TenantID, uid t.Uid, purpose string, attempts int, lockedUntil *time.Time, updatedAt time.Time) error {
if err := validTenant(id); err != nil || uid.IsZero() || purpose == "" || attempts < 0 {
return t.ErrMalformed
}
if updatedAt.IsZero() {
updatedAt = t.TimeNow()
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
_, err := a.db.ExecContext(ctx,
`UPDATE im_youth_mode_settings SET failed_attempts=?,locked_until=?,updated_at=?
WHERE tenant_id=? AND user_id=? AND purpose=?`,
attempts, lockedUntil, updatedAt, id, decodeUID(uid), purpose)
return err
}
func (a *adapter) TenantSystemNotificationCreate(id t.TenantID, notification *t.SystemNotification) error {
if err := validTenant(id); err != nil || notification == nil || notification.TenantID != id ||
notification.UserID.IsZero() || notification.EventID == "" || notification.EventType == "" ||
notification.Title == "" {
return t.ErrMalformed
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
now := t.TimeNow()
notification.CreatedAt = now
notification.UpdatedAt = now
_, err := a.db.ExecContext(ctx,
`INSERT INTO im_system_notifications
(tenant_id,user_id,event_id,event_type,title,summary,action,read_at,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE
event_type=VALUES(event_type),
title=VALUES(title),
summary=VALUES(summary),
action=VALUES(action),
updated_at=VALUES(updated_at)`,
id, decodeUID(notification.UserID), notification.EventID, notification.EventType, notification.Title,
notification.Summary, common.ToJSON(notification.Action), notification.ReadAt, notification.CreatedAt, notification.UpdatedAt)
return err
}
func (a *adapter) TenantSystemNotificationList(id t.TenantID, uid t.Uid, cursor string, limit int) ([]t.SystemNotification, string, error) {
if err := validTenant(id); err != nil || uid.IsZero() {
return nil, "", t.ErrMalformed
}
if limit <= 0 || limit > 100 {
limit = 20
}
lastID := int64(0)
if cursor != "" {
parsed, err := strconv.ParseInt(cursor, 10, 64)
if err != nil || parsed < 0 {
return nil, "", t.ErrMalformed
}
lastID = parsed
}
query := `SELECT id,tenant_id,user_id,event_id,event_type,title,summary,action,read_at,created_at,updated_at
FROM im_system_notifications
WHERE tenant_id=? AND user_id=?`
args := []any{id, decodeUID(uid)}
if lastID > 0 {
query += " AND id<?"
args = append(args, lastID)
}
query += " ORDER BY id DESC LIMIT ?"
args = append(args, limit+1)
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
rows, err := a.db.QueryxContext(ctx, query, args...)
if err != nil {
return nil, "", err
}
defer rows.Close()
notifications := make([]t.SystemNotification, 0, limit)
nextCursor := ""
for rows.Next() {
var notification t.SystemNotification
if err = rows.StructScan(&notification); err != nil {
return nil, "", err
}
if action, ok := common.FromJSON(notification.Action).(map[string]any); ok {
notification.Action = t.KVMap(action)
}
if len(notifications) == limit {
nextCursor = strconv.FormatInt(notification.ID, 10)
continue
}
notifications = append(notifications, notification)
}
if err = rows.Err(); err != nil {
return nil, "", err
}
return notifications, nextCursor, nil
}
func (a *adapter) TenantSystemNotificationMarkRead(id t.TenantID, uid t.Uid, eventID string, readAt time.Time) error {
if err := validTenant(id); err != nil || uid.IsZero() || eventID == "" {
return t.ErrMalformed
}
if readAt.IsZero() {
readAt = t.TimeNow()
}
ctx, cancel := a.getContext()
if cancel != nil {
defer cancel()
}
res, err := a.db.ExecContext(ctx,
`UPDATE im_system_notifications
SET read_at=COALESCE(read_at,?),updated_at=?
WHERE tenant_id=? AND user_id=? AND event_id=?`,
readAt, readAt, id, decodeUID(uid), eventID)
if err != nil {
return err
}
if rows, _ := res.RowsAffected(); rows == 0 {
return t.ErrNotFound
}
return nil
}
func nullableUID(uid t.Uid) any {
if uid.IsZero() {
return nil
}
return store.DecodeUid(uid)
}
func securityTokenHash(tenantID t.TenantID, uid t.Uid, action, target, tokenID, secret string) string {
h := sha256.New()
h.Write([]byte(strconv.FormatInt(int64(tenantID), 10)))
h.Write([]byte{0})
h.Write([]byte(strconv.FormatInt(store.DecodeUid(uid), 10)))
h.Write([]byte{0})
h.Write([]byte(action))
h.Write([]byte{0})
h.Write([]byte(target))
h.Write([]byte{0})
h.Write([]byte(tokenID))
h.Write([]byte{0})
h.Write([]byte(secret))
return fmt.Sprintf("%x", h.Sum(nil))
}