package syncrecord import ( "context" "fmt" "stat/common/log" "stat/common/mongos" "time" "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options" "go.mongodb.org/mongo-driver/bson" ) // 初始化model func Init() { // 获取数据库表 db = mongos.New(mongos.MgoManager, CollSyncRecord) indexes := []mongo.IndexModel{ { Keys: bson.D{{Key: "job", Value: 1}, {Key: "appid", Value: 1}}, Options: options.Index().SetUnique(true), }, } if _, err := db.CreateIndex(context.Background(), indexes); err != nil { panic(fmt.Sprintf("Model %s InitIndex error %+v", CollSyncRecord, err)) } } // LastRecordTimeMap 查询统计记录 func LastRecordTimeMap(ctx context.Context) (list []SyncRecord, err error) { list = make([]SyncRecord, 0) err = db.Find(ctx, &list, bson.M{}) if err != nil { log.Error("SyncRecord LastRecordTimeMap err", log.E(err)) return nil, err } return } // UpsertItemRecord 插入统计成功项记录 func UpsertItemRecord(ctx context.Context, job string, appid int32, userid uint64, platformId string, successAt time.Time) (err error) { filter := bson.M{ "job": job, "appid": appid, } exists, err := db.Exists(ctx, filter) if err != nil { return err } if exists { _, err = db.UpdateOne(ctx, filter, bson.M{ "$set": bson.M{ "userId": userid, "successAt": successAt, "platformId": platformId, "updatedAt": time.Now(), }}) } else { insertDoc := bson.M{ "job": job, "appid": appid, "userId": userid, "platformId": platformId, "successAt": successAt, "createdAt": time.Now(), } _, err = db.InsertOne(ctx, insertDoc) } return } // LastOneRecordTime 查询上一次统计时间 func LastOneRecordTime(ctx context.Context, job string, appid int32) (record SyncRecord, err error) { err = db.FindOne(ctx, &record, bson.M{"job": job, "appid": appid}) if err != nil { log.Error("SyncRecord LastRecordTimeMap err", log.E(err)) return } return }