处理工作手机 和 本地版 过来的联系人

internal/lock/wxid_lock.go

@@ -0,0 +1,73 @@
+package lock
+
+import (
+	"sync"
+)
+
+// WxIdLockManager 管理微信号级别的锁
+type WxIdLockManager struct {
+	lockMap map[string]*sync.Mutex
+	mapMu   sync.Mutex
+}
+
+var (
+	wxIdLockManager *WxIdLockManager
+	once            sync.Once
+)
+
+// GetWxIdLockManager 获取单例锁管理器
+func GetWxIdLockManager() *WxIdLockManager {
+	once.Do(func() {
+		wxIdLockManager = &WxIdLockManager{
+			lockMap: make(map[string]*sync.Mutex),
+		}
+	})
+	return wxIdLockManager
+}
+
+// getLock 获取指定微信号的锁
+func (m *WxIdLockManager) getLock(wxid string) *sync.Mutex {
+	m.mapMu.Lock()
+	defer m.mapMu.Unlock()
+
+	if _, ok := m.lockMap[wxid]; !ok {
+		m.lockMap[wxid] = &sync.Mutex{}
+	}
+	return m.lockMap[wxid]
+}
+
+// RunWithLock 只对某个微信号加锁并执行指定函数
+func (m *WxIdLockManager) RunWithLock(wxid string, fn func()) {
+	lock := m.getLock(wxid)
+	lock.Lock()
+	defer lock.Unlock()
+	fn()
+}
+
+var (
+	lockMap = make(map[string]*sync.Mutex)
+	mapLock = sync.Mutex{}
+)
+
+func LockWxId(wxid string) {
+	mapLock.Lock()
+	mu, ok := lockMap[wxid]
+	if !ok {
+		mu = &sync.Mutex{}
+		lockMap[wxid] = mu
+	}
+	mapLock.Unlock()
+
+	mu.Lock()
+}
+
+// UnlockWxId 解锁指定微信号
+func UnlockWxId(wxid string) {
+	mapLock.Lock()
+	mu, ok := lockMap[wxid]
+	mapLock.Unlock()
+
+	if ok {
+		mu.Unlock()
+	}
+}

internal/service/MessageHandlers/contact_Label_info_notice.go

@@ -12,6 +12,7 @@ import (
 	"wechat-api/ent/label"
 	"wechat-api/ent/labellog"
 	"wechat-api/ent/wx"
+	"wechat-api/internal/lock"
 	"wechat-api/internal/pkg/wechat_ws"
 	"wechat-api/internal/svc"
 	"wechat-api/workphone"
@@ -90,7 +91,8 @@ func (f *ContactLabelInfoNotice) Handle(ctx context.Context, msg *wechat_ws.MsgJ
 		tsInt, err := strconv.ParseInt(labelWx.CreateTime, 10, 64)
 		if err != nil {
 			logx.Errorf("时间戳转换失败: %v (Label ID: %d)", err, labelWx.LabelId)
-			continue
+			//continue
+			tsInt = time.Now().UnixMilli()
 		}
 
 		// 插入 LabelLog
@@ -122,6 +124,8 @@ func (f *ContactLabelInfoNotice) Handle(ctx context.Context, msg *wechat_ws.MsgJ
 		}
 	}
 
+	lock.LockWxId(message.WeChatId)
+	defer lock.UnlockWxId(message.WeChatId)
 	// 批量插入 LabelLog
 	if len(bulkLabelLogs) > 0 {
 		err := svcCtx.DB.LabelLog.CreateBulk(bulkLabelLogs...).

internal/service/MessageHandlers/friend_push_notice.go

@@ -7,12 +7,14 @@ import (
 	"github.com/zeromicro/go-zero/core/logx"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 	"wechat-api/ent"
 	"wechat-api/ent/label"
 	"wechat-api/ent/labellog"
 	"wechat-api/ent/labelrelationship"
 	"wechat-api/ent/wx"
+	"wechat-api/internal/lock"
 	"wechat-api/internal/pkg/wechat_ws"
 	"wechat-api/internal/svc"
 	"wechat-api/workphone"
@@ -76,7 +78,8 @@ func (f *FriendPushNoticeHandler) Handler(msg *wechat_ws.MsgJsonObject) error {
 }
 
 type FriendPushNoticeTypeHandler struct {
-	svcCtx *svc.ServiceContext
+	svcCtx  *svc.ServiceContext
+	lockMap sync.Map // 微信号 -> *sync.Mutex
 }
 
 func NewFriendPushNoticeTypeHandler(svcCtx *svc.ServiceContext) *FriendPushNoticeTypeHandler {
@@ -106,6 +109,13 @@ func (f *FriendPushNoticeTypeHandler) Handle(ctx context.Context, msg *wechat_ws
 	for _, friend := range message.Friends {
 		var friendId uint64
 		friendType := 1
+		if friend.Type == 1 {
+			friendType = 4
+		} else {
+			friendType = 1
+		}
+
+		//修改拉黑后的状态被重置
 		friendId, err = svcCtx.DB.Contact.Create().
 			SetWxWxid(message.WeChatId).
 			SetType(friendType).
@@ -116,9 +126,14 @@ func (f *FriendPushNoticeTypeHandler) Handle(ctx context.Context, msg *wechat_ws
 			SetHeadimg(friend.Avatar).
 			SetOrganizationID(wxInfo.OrganizationID).
 			OnConflict().
-			UpdateNewValues().
-			SetType(friendType).
-			SetOrganizationID(wxInfo.OrganizationID).
+			UpdateWxWxid().
+			UpdateWxid().
+			UpdateType().
+			UpdateAccount().
+			UpdateNickname().
+			UpdateMarkname().
+			UpdateHeadimg().
+			UpdateOrganizationID().
 			ID(ctx)
 		if err != nil {
 			logx.Errorf("Contact.Create 失败, OrgID=%d, err=%v", wxInfo.OrganizationID, err)
@@ -164,23 +179,6 @@ func (f *FriendPushNoticeTypeHandler) Handle(ctx context.Context, msg *wechat_ws
 				logx.Error("label not found.fail: ", wxInfo.OrganizationID)
 				continue
 			}
-
-			//svcCtx.DB.LabelRelationship.Create().
-			//	SetOrganizationID(wxInfo.OrganizationID).
-			//	SetContactID(friendId).
-			//	SetLabelID(labelInfo.ID).
-			//	SetAccount(friend.FriendNo).
-			//	SetNickname(friend.FriendNick).
-			//	SetMarkname(friend.Memo).
-			//	SetHeadimg(friend.Avatar).
-			//	SetOrganizationID(wxInfo.OrganizationID).
-			//	OnConflict().
-			//	UpdateNewValues().
-			//	SetType(friendType).
-			//	SetOrganizationID(wxInfo.OrganizationID).
-			//	ID(ctx)
-
-			//生成批量的关系数据 待插入
 			labelRelationshipCreates = append(labelRelationshipCreates,
 				svcCtx.DB.LabelRelationship.Create().
 					//SetID(int(label.LabelId)).
@@ -195,14 +193,15 @@ func (f *FriendPushNoticeTypeHandler) Handle(ctx context.Context, msg *wechat_ws
 		}
 	}
 	if len(labelRelationshipCreates) > 0 {
-		errShip := svcCtx.DB.LabelRelationship.CreateBulk(labelRelationshipCreates...).
-			OnConflict(
-				sql.ConflictColumns(labelrelationship.FieldLabelID, labelrelationship.FieldContactID),
-			).DoNothing().Exec(ctx)
-		if errShip != nil {
-			logx.Error("label_relationship.create.fail: ", wxInfo.OrganizationID, labelRelationshipCreates)
-			return err
-		}
+		lock.GetWxIdLockManager().RunWithLock(message.WeChatId, func() {
+			errShip := svcCtx.DB.LabelRelationship.CreateBulk(labelRelationshipCreates...).
+				OnConflict(
+					sql.ConflictColumns(labelrelationship.FieldLabelID, labelrelationship.FieldContactID),
+				).DoNothing().Exec(ctx)
+			if errShip != nil {
+				logx.Error("label_relationship.create.fail: ", wxInfo.OrganizationID, labelRelationshipCreates)
+			}
+		})
 	}
 	return nil
 }
@@ -227,3 +226,8 @@ func ParseCSVToIntSlice(csv string) ([]int, error) {
 	}
 	return result, nil
 }
+
+func (f *FriendPushNoticeTypeHandler) getWxLock(wxid string) *sync.Mutex {
+	actual, _ := f.lockMap.LoadOrStore(wxid, &sync.Mutex{})
+	return actual.(*sync.Mutex)
+}