remark42/backend/app/store/engine/mongo.go

package engine

import (
	"time"

	"github.com/globalsign/mgo"
	"github.com/globalsign/mgo/bson"
	"github.com/go-pkgz/mongo"
	multierror "github.com/hashicorp/go-multierror"
	"github.com/pkg/errors"

	"github.com/umputun/remark/backend/app/store"
)

// Mongo implements engine interface
type Mongo struct {
	conn       *mongo.Connection
	postWriter mongo.BufferedWriter
}

const (
	mongoPosts     = "posts"
	mongoMetaPosts = "meta_posts"
	mongoMetaUsers = "meta_users"
)

type metaPost struct {
	ID       string `bson:"_id"` // url
	SiteID   string `bson:"site"`
	ReadOnly bool   `bson:"read_only"`
}

type metaUser struct {
	ID           string    `bson:"_id"` // user_id
	SiteID       string    `bson:"site"`
	Verified     bool      `bson:"verified"`
	Blocked      bool      `bson:"blocked"`
	BlockedUntil time.Time `bson:"blocked_until"`
}

// NewMongo makes mongo engine. bufferSize denies how many records will be buffered, 0 turns buffering off.
// flushDuration triggers automatic flus (write from buffer), 0 disables it and will flush as buffer size reached.
// important! don't use flushDuration=0 for production use as it can leave records in-fly state for long or even unlimited time.
func NewMongo(conn *mongo.Connection, bufferSize int, flushDuration time.Duration) (*Mongo, error) {
	writer := mongo.NewBufferedWriter(bufferSize, conn).WithCollection(mongoPosts).WithAutoFlush(flushDuration)
	result := Mongo{conn: conn, postWriter: writer}
	err := result.prepare()
	return &result, errors.Wrap(err, "failed to prepare mongo")
}

// Create new comment, write can be buffered and delayed.
func (m *Mongo) Create(comment store.Comment) (commentID string, err error) {
	// err = m.postWriter.Write(comment)
	err = m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		return coll.Insert(&comment)
	})
	return comment.ID, err
}

// Find returns all comments for post and sorts results
func (m *Mongo) Find(locator store.Locator, sortFld string) (comments []store.Comment, err error) {
	comments = []store.Comment{}
	err = m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		query := bson.M{"locator.site": locator.SiteID, "locator.url": locator.URL}
		return coll.Find(query).Sort(sortFld).All(&comments)
	})
	return comments, err
}

// Get returns comment for locator.URL and commentID string
func (m *Mongo) Get(locator store.Locator, commentID string) (comment store.Comment, err error) {
	err = m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		query := bson.M{"_id": commentID, "locator.site": locator.SiteID, "locator.url": locator.URL}
		return coll.Find(query).One(&comment)
	})
	return comment, err
}

// Put updates comment for locator.URL with mutable part of comment
func (m *Mongo) Put(locator store.Locator, comment store.Comment) error {
	return m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		return coll.Update(bson.M{"_id": comment.ID, "locator.site": locator.SiteID, "locator.url": locator.URL},
			bson.M{"$set": bson.M{
				"text":    comment.Text,
				"orig":    comment.Orig,
				"score":   comment.Score,
				"votes":   comment.Votes,
				"pin":     comment.Pin,
				"deleted": comment.Deleted,
			}})
	})
}

// Last returns up to max last comments for given siteID
func (m *Mongo) Last(siteID string, max int) (comments []store.Comment, err error) {
	comments = []store.Comment{}
	if max > lastLimit || max == 0 {
		max = lastLimit
	}
	err = m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		query := bson.M{"locator.site": siteID, "delete": false}
		return coll.Find(query).Sort("-time").Limit(max).All(&comments)
	})
	return comments, err
}

// Count returns number of comments for locator
func (m *Mongo) Count(locator store.Locator) (count int, err error) {

	e := m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		query := bson.M{"locator.site": locator.SiteID, "locator.url": locator.URL, "delete": false}
		count, err = coll.Find(query).Count()
		return err
	})
	return count, e
}

// List returns list of all commented posts with counters
func (m *Mongo) List(siteID string, limit, skip int) (list []store.PostInfo, err error) {
	list = []store.PostInfo{}

	if limit <= 0 {
		limit = 1000
	}
	if skip < 0 {
		skip = 0
	}

	err = m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		pipeline := coll.Pipe([]bson.M{
			{"$match": bson.M{"locator.site": siteID}},
			{"$project": bson.M{"locator.site": 1, "locator.url": 1, "time": 1}},
			{"$group": bson.M{"_id": "$locator.url", "url": bson.M{"$first": "$locator.url"}, "count": bson.M{"$sum": 1},
				"first_time": bson.M{"$min": "$time"}, "last_time": bson.M{"$max": "$time"}}},
			{"$skip": skip},
			{"$limit": limit},
		})
		return errors.Wrap(pipeline.AllowDiskUse().All(&list), "list pipeline failed")
	})
	return list, errors.Wrap(err, "can't get list")
}

// Info returns time range and count for locator
func (m *Mongo) Info(locator store.Locator, readOnlyAge int) (info store.PostInfo, err error) {
	list := []store.PostInfo{}
	err = m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		pipeline := coll.Pipe([]bson.M{
			{"$match": bson.M{"locator.site": locator.SiteID, "locator.url": locator.URL}},
			{"$project": bson.M{"locator.site": 1, "locator.url": 1, "time": 1}},
			{"$group": bson.M{"_id": "$locator.url", "url": bson.M{"$first": "$locator.url"}, "count": bson.M{"$sum": 1},
				"first_time": bson.M{"$min": "$time"}, "last_time": bson.M{"$max": "$time"}}},
		})
		return errors.Wrap(pipeline.AllowDiskUse().All(&list), "list pipeline failed")
	})
	if err != nil {
		return info, err
	}
	if len(list) == 0 {
		return info, errors.Errorf("can't load info for %s", locator.URL)
	}
	info = list[0]
	// set read-only from age and manual bucket
	info.ReadOnly = readOnlyAge > 0 && !info.FirstTS.IsZero() && info.FirstTS.AddDate(0, 0, readOnlyAge).Before(time.Now())
	if m.IsReadOnly(locator) {
		info.ReadOnly = true
	}
	return info, nil
}

// User extracts all comments for given site and given userID
func (m *Mongo) User(siteID, userID string, limit, skip int) (comments []store.Comment, err error) {
	comments = []store.Comment{}
	err = m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		query := bson.M{"locator.site": siteID, "user.id": userID}
		return m.setLimitAndSkip(coll.Find(query).Sort("-time"), limit, skip).All(&comments)
	})
	return comments, errors.Wrapf(err, "can't get comments for user %s", userID)
}

// UserCount returns number of comments for user
func (m *Mongo) UserCount(siteID, userID string) (count int, err error) {
	err = m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		var e error
		count, e = coll.Find(bson.M{"locator.site": siteID, "user.id": userID}).Count()
		return e
	})
	return count, errors.Wrapf(err, "can't get comments count for user %s", userID)
}

// SetReadOnly makes post read-only or reset the ro flag
func (m *Mongo) SetReadOnly(locator store.Locator, status bool) (err error) {
	return m.conn.WithCustomCollection(mongoMetaPosts, func(coll *mgo.Collection) error {
		_, e := coll.Upsert(bson.M{"_id": locator.URL, "site": locator.SiteID}, bson.M{"$set": bson.M{"read_only": status}})
		return e
	})
}

// IsReadOnly checks if post in RO
func (m *Mongo) IsReadOnly(locator store.Locator) (ro bool) {
	meta := metaPost{}
	err := m.conn.WithCustomCollection(mongoMetaPosts, func(coll *mgo.Collection) error {
		return coll.Find(bson.M{"_id": locator.URL, "site": locator.SiteID}).One(&meta)
	})
	return err == nil && meta.ReadOnly
}

// SetVerified makes user verified or reset the flag
func (m *Mongo) SetVerified(siteID string, userID string, status bool) error {
	return m.conn.WithCustomCollection(mongoMetaUsers, func(coll *mgo.Collection) error {
		_, e := coll.Upsert(bson.M{"_id": userID, "site": siteID}, bson.M{"$set": bson.M{"verified": status}})
		return e
	})
}

// IsVerified checks if user verified
func (m *Mongo) IsVerified(siteID string, userID string) (verified bool) {
	meta := metaUser{}
	err := m.conn.WithCustomCollection(mongoMetaUsers, func(coll *mgo.Collection) error {
		return coll.Find(bson.M{"_id": userID, "site": siteID}).One(&meta)
	})
	return err == nil && meta.Verified
}

// Verified returns list of verified user IDs
func (m *Mongo) Verified(siteID string) (ids []string, err error) {
	metas := []metaUser{}
	err = m.conn.WithCustomCollection(mongoMetaUsers, func(coll *mgo.Collection) error {
		return coll.Find(bson.M{"site": siteID, "verified": true}).All(&metas)
	})
	if err != nil {
		return nil, err
	}
	for _, meta := range metas {
		ids = append(ids, meta.ID)
	}
	return ids, nil
}

// SetBlock blocks/unblocks user for given site. ttl defines for for how long, 0 - permanent
// block uses blocksBucketName with key=userID and val=TTL+now
func (m *Mongo) SetBlock(siteID string, userID string, status bool, ttl time.Duration) error {

	until := time.Time{}
	if status {
		until = time.Now().AddDate(100, 0, 0) // permanent is 50year
		if ttl > 0 {
			until = time.Now().Add(ttl)
		}
	}
	return m.conn.WithCustomCollection(mongoMetaUsers, func(coll *mgo.Collection) error {
		_, e := coll.Upsert(bson.M{"_id": userID, "site": siteID},
			bson.M{"$set": bson.M{"blocked": status, "blocked_until": until}})
		return errors.Wrapf(e, "failed to set block for %s", userID)
	})
}

// IsBlocked checks if user blocked
func (m *Mongo) IsBlocked(siteID string, userID string) (blocked bool) {
	meta := metaUser{}
	err := m.conn.WithCustomCollection(mongoMetaUsers, func(coll *mgo.Collection) error {
		return coll.Find(bson.M{"_id": userID, "site": siteID}).One(&meta)
	})
	return err == nil && meta.Blocked && meta.BlockedUntil.After(time.Now())
}

// Blocked get lists of blocked users for given site
func (m *Mongo) Blocked(siteID string) (users []store.BlockedUser, err error) {
	users = []store.BlockedUser{}
	metas := []metaUser{}
	err = m.conn.WithCustomCollection(mongoMetaUsers, func(coll *mgo.Collection) error {
		return coll.Find(bson.M{"site": siteID,
			"blocked": true, "blocked_until": bson.M{"$gt": time.Now()}}).All(&metas)
	})
	if err != nil {
		return users, errors.Wrapf(err, "can't get blocked users for site for %s", siteID)
	}

	for _, mu := range metas {
		blockedUser := store.BlockedUser{ID: mu.ID, Until: mu.BlockedUntil}
		if ucc, e := m.User(siteID, mu.ID, 1, 0); e == nil && len(ucc) > 0 {
			blockedUser.Name = ucc[0].User.Name
		}
		users = append(users, blockedUser)
	}
	return users, nil
}

// Delete removes comment, by locator from the store.
// Posts collection only sets status to deleted and clear fields in order to prevent breaking trees of replies.
func (m *Mongo) Delete(locator store.Locator, commentID string, mode store.DeleteMode) error {
	comment := store.Comment{}
	err := m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		e := coll.Find(bson.M{"locator.site": locator.SiteID, "locator.url": locator.URL, "_id": commentID}).One(&comment)
		if e != nil {
			return e
		}
		comment.SetDeleted(mode)
		return coll.Update(bson.M{"locator.site": locator.SiteID, "locator.url": locator.URL, "_id": commentID}, comment)
	})
	return errors.Wrapf(err, "can't delete %s", commentID)
}

// DeleteAll removes all info about siteID
func (m *Mongo) DeleteAll(siteID string) error {
	err := m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		_, e := coll.RemoveAll(bson.M{"locator.site": siteID})
		return e
	})
	return errors.Wrapf(err, "can't delete site %s", siteID)
}

// DeleteUser removes all comments for given user. Everything will be market as deleted
// and user name and userID will be changed to "deleted".
func (m *Mongo) DeleteUser(siteID string, userID string) error {
	comments := []store.Comment{}
	return m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		e := coll.Find(bson.M{"locator.site": siteID, "user.id": userID}).All(&comments)
		if e != nil {
			return e
		}
		for _, c := range comments {
			if e = m.Delete(c.Locator, c.ID, store.HardDelete); e != nil {
				return e
			}
		}
		return nil
	})
}

// Close boltdb store
func (m *Mongo) Close() error {
	if m.postWriter != nil {
		return m.postWriter.Close()
	}
	return nil
}

// prepare collections with all indexes
func (m *Mongo) prepare() error {
	errs := new(multierror.Error)
	e := m.conn.WithCustomCollection(mongoPosts, func(coll *mgo.Collection) error {
		errs = multierror.Append(errs, coll.EnsureIndexKey("user.id", "locator.site", "time"))
		errs = multierror.Append(errs, coll.EnsureIndexKey("locator.url", "locator.site", "time"))
		errs = multierror.Append(errs, coll.EnsureIndexKey("locator.site", "time"))
		errs = multierror.Append(errs, coll.EnsureIndexKey("locator.url", "locator.site", "score"))
		return errors.Wrapf(errs.ErrorOrNil(), "can't create index for %s", mongoPosts)
	})
	if e != nil {
		return e
	}

	e = m.conn.WithCustomCollection(mongoMetaPosts, func(coll *mgo.Collection) error {
		errs = multierror.Append(errs, coll.EnsureIndexKey("_id", "site"))
		errs = multierror.Append(errs, coll.EnsureIndexKey("site", "read_only"))
		return errors.Wrapf(errs.ErrorOrNil(), "can't create index for %s", mongoMetaPosts)
	})
	if e != nil {
		return e
	}

	return m.conn.WithCustomCollection(mongoMetaUsers, func(coll *mgo.Collection) error {
		errs = multierror.Append(errs, coll.EnsureIndexKey("_id", "site"))
		errs = multierror.Append(errs, coll.EnsureIndexKey("site", "blocked"))
		errs = multierror.Append(errs, coll.EnsureIndexKey("site", "verified"))
		return errors.Wrapf(errs.ErrorOrNil(), "can't create index for %s", mongoMetaUsers)
	})
}

func (m *Mongo) setLimitAndSkip(q *mgo.Query, limit, skip int) *mgo.Query {
	if limit <= 0 {
		limit = 1000
	}
	if skip < 0 {
		skip = 0
	}
	return q.Skip(skip).Limit(limit)
}