open-im-server-deploy/pkg/common/storage/database/mgo/meeting.go

// Copyright © 2023 OpenIM. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package mgo

import (
	"context"
	"time"

	"git.imall.cloud/openim/open-im-server-deploy/pkg/common/storage/database"
	"git.imall.cloud/openim/open-im-server-deploy/pkg/common/storage/model"
	"github.com/openimsdk/tools/db/mongoutil"
	"github.com/openimsdk/tools/db/pagination"
	"github.com/openimsdk/tools/errs"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

// MeetingMgo implements Meeting using MongoDB as the storage backend.
type MeetingMgo struct {
	coll *mongo.Collection
}

// NewMeetingMongo creates a new instance of MeetingMgo with the provided MongoDB database.
func NewMeetingMongo(db *mongo.Database) (database.Meeting, error) {
	coll := db.Collection(database.MeetingName)
	_, err := coll.Indexes().CreateMany(context.Background(), []mongo.IndexModel{
		{
			Keys:    bson.D{{Key: "meeting_id", Value: 1}},
			Options: options.Index().SetUnique(true),
		},
		{
			Keys: bson.D{{Key: "creator_user_id", Value: 1}},
		},
		{
			Keys: bson.D{{Key: "status", Value: 1}},
		},
		{
			Keys: bson.D{{Key: "scheduled_time", Value: 1}},
		},
		{
			Keys: bson.D{{Key: "create_time", Value: -1}},
		},
		{
			Keys: bson.D{{Key: "update_time", Value: -1}},
		},
		{
			Keys: bson.D{{Key: "subject", Value: "text"}, {Key: "description", Value: "text"}},
		},
	})
	if err != nil {
		return nil, errs.Wrap(err)
	}
	return &MeetingMgo{coll: coll}, nil
}

// Create creates a new meeting record.
func (m *MeetingMgo) Create(ctx context.Context, meeting *model.Meeting) error {
	if meeting.CreateTime.IsZero() {
		meeting.CreateTime = time.Now()
	}
	if meeting.UpdateTime.IsZero() {
		meeting.UpdateTime = time.Now()
	}
	return mongoutil.InsertOne(ctx, m.coll, meeting)
}

// Take retrieves a meeting by meeting ID. Returns an error if not found.
func (m *MeetingMgo) Take(ctx context.Context, meetingID string) (*model.Meeting, error) {
	return mongoutil.FindOne[*model.Meeting](ctx, m.coll, bson.M{"meeting_id": meetingID})
}

// Update updates meeting information.
func (m *MeetingMgo) Update(ctx context.Context, meetingID string, data map[string]any) error {
	data["update_time"] = time.Now()
	update := bson.M{"$set": data}
	return mongoutil.UpdateOne(ctx, m.coll, bson.M{"meeting_id": meetingID}, update, false)
}

// UpdateStatus updates the status of a meeting.
func (m *MeetingMgo) UpdateStatus(ctx context.Context, meetingID string, status int32) error {
	return m.Update(ctx, meetingID, map[string]any{"status": status})
}

// Find finds meetings by meeting IDs.
func (m *MeetingMgo) Find(ctx context.Context, meetingIDs []string) ([]*model.Meeting, error) {
	if len(meetingIDs) == 0 {
		return []*model.Meeting{}, nil
	}
	filter := bson.M{"meeting_id": bson.M{"$in": meetingIDs}}
	return mongoutil.Find[*model.Meeting](ctx, m.coll, filter)
}

// FindByCreator finds meetings created by a specific user.
func (m *MeetingMgo) FindByCreator(ctx context.Context, creatorUserID string, pagination pagination.Pagination) (total int64, meetings []*model.Meeting, err error) {
	filter := bson.M{"creator_user_id": creatorUserID}
	return mongoutil.FindPage[*model.Meeting](ctx, m.coll, filter, pagination, &options.FindOptions{
		Sort: bson.D{{Key: "scheduled_time", Value: -1}},
	})
}

// FindAll finds all meetings with pagination.
func (m *MeetingMgo) FindAll(ctx context.Context, pagination pagination.Pagination) (total int64, meetings []*model.Meeting, err error) {
	return mongoutil.FindPage[*model.Meeting](ctx, m.coll, bson.M{}, pagination, &options.FindOptions{
		Sort: bson.D{{Key: "scheduled_time", Value: -1}},
	})
}

// Search searches meetings by keyword (subject, description).
func (m *MeetingMgo) Search(ctx context.Context, keyword string, pagination pagination.Pagination) (total int64, meetings []*model.Meeting, err error) {
	filter := bson.M{
		"$or": []bson.M{
			{"subject": bson.M{"$regex": keyword, "$options": "i"}},
			{"description": bson.M{"$regex": keyword, "$options": "i"}},
		},
	}
	return mongoutil.FindPage[*model.Meeting](ctx, m.coll, filter, pagination, &options.FindOptions{
		Sort: bson.D{{Key: "scheduled_time", Value: -1}},
	})
}

// FindByStatus finds meetings by status.
func (m *MeetingMgo) FindByStatus(ctx context.Context, status int32, pagination pagination.Pagination) (total int64, meetings []*model.Meeting, err error) {
	filter := bson.M{"status": status}
	return mongoutil.FindPage[*model.Meeting](ctx, m.coll, filter, pagination, &options.FindOptions{
		Sort: bson.D{{Key: "scheduled_time", Value: -1}},
	})
}

// FindByScheduledTimeRange finds meetings within a scheduled time range.
func (m *MeetingMgo) FindByScheduledTimeRange(ctx context.Context, startTime, endTime int64, pagination pagination.Pagination) (total int64, meetings []*model.Meeting, err error) {
	filter := bson.M{
		"scheduled_time": bson.M{
			"$gte": time.UnixMilli(startTime),
			"$lte": time.UnixMilli(endTime),
		},
	}
	return mongoutil.FindPage[*model.Meeting](ctx, m.coll, filter, pagination, &options.FindOptions{
		Sort: bson.D{{Key: "scheduled_time", Value: 1}},
	})
}

// FindFinishedMeetingsBefore finds finished meetings that ended before the specified time.
// A meeting is considered finished if its status is 3 (Finished) and its end time (scheduledTime + duration) is before beforeTime.
// Only returns meetings with a non-empty group_id to avoid processing meetings that have already been handled.
func (m *MeetingMgo) FindFinishedMeetingsBefore(ctx context.Context, beforeTime time.Time) ([]*model.Meeting, error) {
	// 查询状态为3（已结束）且group_id不为空的会议
	// 结束时间 = scheduledTime + duration（分钟）
	// 需要计算：scheduledTime + duration * 60秒 <= beforeTime
	filter := bson.M{
		"status": 3, // 已结束
		"group_id": bson.M{"$ne": ""}, // 只查询group_id不为空的会议，避免重复处理已清空groupID的会议
		"$expr": bson.M{
			"$lte": []interface{}{
				bson.M{
					"$add": []interface{}{
						"$scheduled_time",
						bson.M{"$multiply": []interface{}{"$duration", int64(60)}}, // duration是分钟，转换为秒
					},
				},
				beforeTime,
			},
		},
	}
	return mongoutil.Find[*model.Meeting](ctx, m.coll, filter)
}

// Delete deletes a meeting by meeting ID.
func (m *MeetingMgo) Delete(ctx context.Context, meetingID string) error {
	return mongoutil.DeleteOne(ctx, m.coll, bson.M{"meeting_id": meetingID})
}