open-im-server-deploy/internal/tools/cron/meeting.go

package cron

import (
	"context"
	"fmt"
	"os"
	"time"

	"git.imall.cloud/openim/open-im-server-deploy/pkg/common/storage/model"
	"git.imall.cloud/openim/open-im-server-deploy/pkg/common/webhook"
	"github.com/openimsdk/tools/errs"
	"github.com/openimsdk/tools/log"
	"github.com/openimsdk/tools/mcontext"
)

// meetingPagination 简单的分页实现，供定时任务批量扫描使用
type meetingPagination struct {
	pageNumber int32
	showNumber int32
}

func (p *meetingPagination) GetPageNumber() int32 {
	if p.pageNumber <= 0 {
		return 1
	}
	return p.pageNumber
}

func (p *meetingPagination) GetShowNumber() int32 {
	if p.showNumber <= 0 {
		return 200
	}
	return p.showNumber
}

// dismissMeetingGroups 解散已结束超过10分钟的会议群聊
func (c *cronServer) dismissMeetingGroups() {
	now := time.Now()
	// 计算10分钟前的时间
	beforeTime := now.Add(-10 * time.Minute)
	operationID := fmt.Sprintf("cron_dismiss_meeting_groups_%d_%d", os.Getpid(), now.UnixMilli())
	ctx := mcontext.SetOperationID(c.ctx, operationID)

	log.ZDebug(ctx, "Start dismissing meeting groups", "beforeTime", beforeTime)

	// 先将已过期但状态仍为已预约/进行中的会议标记为已结束
	c.finishExpiredMeetings(ctx, now)

	// 查询已结束且结束时间在10分钟前的会议
	// 结束时间 = scheduledTime + duration（分钟）
	meetings, err := c.meetingDB.FindFinishedMeetingsBefore(ctx, beforeTime)
	if err != nil {
		log.ZError(ctx, "Failed to find finished meetings", err)
		return
	}

	if len(meetings) == 0 {
		log.ZDebug(ctx, "No finished meetings to dismiss groups")
		return
	}

	log.ZInfo(ctx, "Found finished meetings to dismiss groups", "count", len(meetings))

	dismissedCount := 0
	failedCount := 0

	for _, meeting := range meetings {
		if meeting.GroupID == "" {
			log.ZWarn(ctx, "Meeting has no group ID, skip", nil, "meetingID", meeting.MeetingID)
			continue
		}

		// 计算会议结束时间
		endTime := meeting.ScheduledTime.Add(time.Duration(meeting.Duration) * time.Minute)
		// 检查是否已经超过10分钟
		if now.Sub(endTime) < 10*time.Minute {
			log.ZDebug(ctx, "Meeting ended less than 10 minutes ago, skip", "meetingID", meeting.MeetingID, "endTime", endTime)
			continue
		}

		// 解散群聊，deleteMember设为true表示删除所有成员
		ctx := mcontext.SetOperationID(c.ctx, fmt.Sprintf("%s_%s", operationID, meeting.MeetingID))
		err := c.groupClient.DismissGroup(ctx, meeting.GroupID, true)
		if err != nil {
			// 如果群不存在或找不到群主（RecordNotFoundError），说明群可能已经被解散或数据不一致
			if errs.ErrRecordNotFound.Is(err) {
				log.ZWarn(ctx, "Group not found or owner not found, may already be dismissed, clear groupID", nil, "meetingID", meeting.MeetingID, "groupID", meeting.GroupID)
				// 清空groupID，避免下次重复处理
				if updateErr := c.meetingDB.Update(ctx, meeting.MeetingID, map[string]any{"group_id": ""}); updateErr != nil {
					log.ZWarn(ctx, "Failed to clear groupID after group not found", updateErr, "meetingID", meeting.MeetingID)
				}
				// 不增加失败计数，因为这不是真正的失败
				continue
			}
			log.ZError(ctx, "Failed to dismiss meeting group", err, "meetingID", meeting.MeetingID, "groupID", meeting.GroupID)
			failedCount++
			continue
		}

		// 从webhook配置的attentionIds中移除会议群ID
		if c.systemConfigDB != nil {
			if err := webhook.UpdateAttentionIds(ctx, c.systemConfigDB, meeting.GroupID, false); err != nil {
				log.ZWarn(ctx, "dismissMeetingGroups: failed to remove groupID from webhook attentionIds", err, "meetingID", meeting.MeetingID, "groupID", meeting.GroupID)
			}
		}

		// 解散群成功后，清空会议的groupID，避免下次重复处理
		if updateErr := c.meetingDB.Update(ctx, meeting.MeetingID, map[string]any{"group_id": ""}); updateErr != nil {
			log.ZWarn(ctx, "Failed to clear groupID after dismissing group", updateErr, "meetingID", meeting.MeetingID, "groupID", meeting.GroupID)
		} else {
			log.ZInfo(ctx, "Successfully dismissed meeting group and cleared groupID", "meetingID", meeting.MeetingID, "groupID", meeting.GroupID, "endTime", endTime)
		}
		dismissedCount++
	}

	log.ZInfo(ctx, "Finished dismissing meeting groups", "total", len(meetings), "dismissed", dismissedCount, "failed", failedCount, "duration", time.Since(now))
}

// finishExpiredMeetings 将已过结束时间的会议状态更新为已结束
func (c *cronServer) finishExpiredMeetings(ctx context.Context, now time.Time) {
	statuses := []int32{model.MeetingStatusScheduled, model.MeetingStatusOngoing}
	for _, status := range statuses {
		page := int32(1)
		for {
			total, meetings, err := c.meetingDB.FindByStatus(ctx, status, &meetingPagination{pageNumber: page, showNumber: 200})
			if err != nil {
				log.ZWarn(ctx, "finishExpiredMeetings: failed to list meetings", err, "status", status, "page", page)
				break
			}
			if len(meetings) == 0 {
				break
			}

			for _, meeting := range meetings {
				endTime := meeting.ScheduledTime.Add(time.Duration(meeting.Duration) * time.Minute)
				if now.After(endTime) {
					if err := c.meetingDB.UpdateStatus(ctx, meeting.MeetingID, model.MeetingStatusFinished); err != nil {
						log.ZWarn(ctx, "finishExpiredMeetings: failed to update status", err, "meetingID", meeting.MeetingID)
						continue
					}
					log.ZInfo(ctx, "finishExpiredMeetings: meeting marked finished", "meetingID", meeting.MeetingID, "endTime", endTime)
				}
			}

			// 分页结束条件
			if int64(page*200) >= total {
				break
			}
			page++
		}
	}
}