Logs and posts are now on the new architecture.

database/database.go

@@ -10,12 +10,14 @@ import (
 )
 
 // ErrDriverUnrecognized is returned if the driver is not recognized
-var ErrDriverUnrecognized = errors.New("Driver not recognized, check your config or update rpdata")
+var ErrDriverUnrecognized = errors.New("driver not recognized, check installed version or your configuration")
 
 type Database interface {
 	Changes() repositories.ChangeRepository
 	Characters() repositories.CharacterRepository
 	Tags() repositories.TagRepository
+	Logs() repositories.LogRepository
+	Posts() repositories.PostRepository
 
 	Close(ctx context.Context) error
 }

database/mongodb/characters.go

@@ -99,7 +99,7 @@ func (r *characterRepository) List(ctx context.Context, filter models.CharacterF
 	}
 
 	characters := make([]*models.Character, 0, 32)
-	err := r.characters.Find(query).All(&characters)
+	err := r.characters.Find(query).Limit(filter.Limit).All(&characters)
 	if err != nil {
 		if err == mgo.ErrNotFound {
 			return characters, nil

database/mongodb/db.go

@@ -17,6 +17,8 @@ type MongoDB struct {
 	changes    repositories.ChangeRepository
 	characters repositories.CharacterRepository
 	tags       repositories.TagRepository
+	logs       *logRepository
+	posts      *postRepository
 }
 
 func (m *MongoDB) Changes() repositories.ChangeRepository {
@@ -31,6 +33,14 @@ func (m *MongoDB) Tags() repositories.TagRepository {
 	return m.tags
 }
 
+func (m *MongoDB) Logs() repositories.LogRepository {
+	return m.logs
+}
+
+func (m *MongoDB) Posts() repositories.PostRepository {
+	return m.posts
+}
+
 func (m *MongoDB) Close(ctx context.Context) error {
 	m.session.Close()
 	return nil
@@ -70,12 +80,28 @@ func Init(cfg config.Database) (*MongoDB, error) {
 		return nil, err
 	}
 
+	logs, err := newLogRepository(db)
+	if err != nil {
+		session.Close()
+		return nil, err
+	}
+
+	posts, err := newPostRepository(db)
+	if err != nil {
+		session.Close()
+		return nil, err
+	}
+
+	go posts.fixPositions(logs)
+
 	return &MongoDB{
 		session: session,
 
 		changes:    changes,
 		characters: characters,
 		tags:       newTagRepository(db),
+		logs:       logs,
+		posts:      posts,
 	}, nil
 }
 

database/mongodb/logs.go

@@ -0,0 +1,193 @@
+package mongodb
+
+import (
+	"context"
+	"errors"
+	"git.aiterp.net/rpdata/api/internal/generate"
+	"git.aiterp.net/rpdata/api/models"
+	"github.com/globalsign/mgo"
+	"github.com/globalsign/mgo/bson"
+	"strconv"
+	"sync"
+)
+
+type logRepository struct {
+	openMutex sync.Mutex
+
+	logs           *mgo.Collection
+	posts          *mgo.Collection
+	shortIdCounter *counter
+}
+
+func newLogRepository(db *mgo.Database) (*logRepository, error) {
+	logs := db.C("logbot3.logs")
+	posts := db.C("logbot3.posts")
+
+	err := logs.EnsureIndexKey("date")
+	if err != nil {
+		return nil, err
+	}
+	err = logs.EnsureIndexKey("channel")
+	if err != nil {
+		return nil, err
+	}
+	err = logs.EnsureIndexKey("characterIds")
+	if err != nil {
+		return nil, err
+	}
+	err = logs.EnsureIndexKey("event")
+	if err != nil {
+		return nil, err
+	}
+	err = logs.EnsureIndex(mgo.Index{Key: []string{"channel", "open"}})
+	if err != nil {
+		return nil, err
+	}
+
+	err = logs.EnsureIndex(mgo.Index{
+		Key:      []string{"shortId"},
+		Unique:   true,
+		DropDups: true,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &logRepository{
+		logs:           logs,
+		posts:          posts,
+		shortIdCounter: newCounter(db, "auto_increment", "Log"),
+	}, nil
+}
+
+func (r *logRepository) Find(ctx context.Context, id string) (*models.Log, error) {
+	log := new(models.Log)
+	err := r.logs.Find(bson.M{"$or": []bson.M{{"_id": id}, {"shortId": id}}}).One(log)
+	if err != nil {
+		return nil, err
+	}
+
+	return log, nil
+}
+
+func (r *logRepository) List(ctx context.Context, filter models.LogFilter) ([]*models.Log, error) {
+	query := bson.M{}
+	if filter.Search != nil {
+		searchQuery := bson.M{
+			"$text": bson.M{"$search": *filter.Search},
+			"logId": bson.M{"$ne": nil},
+		}
+
+		logIds := make([]string, 0, 64)
+		err := r.posts.Find(searchQuery).Distinct("logId", &logIds)
+		if err != nil {
+			return nil, err
+		}
+
+		query["shortId"] = bson.M{"$in": logIds}
+	}
+	if filter.Open != nil {
+		r.openMutex.Lock()
+		defer r.openMutex.Unlock()
+
+		query["open"] = filter.Open
+	}
+	if len(filter.Characters) > 0 {
+		query["characterIds"] = bson.M{"$in": filter.Characters}
+	}
+	if len(filter.Channels) > 0 {
+		query["channel"] = bson.M{"$in": filter.Channels}
+	}
+	if len(filter.Events) > 0 {
+		query["event"] = bson.M{"$in": filter.Events}
+	}
+
+	logs := make([]*models.Log, 0, 32)
+	err := r.logs.Find(query).Sort("-date").Limit(filter.Limit).All(&logs)
+	if err != nil {
+		if err == mgo.ErrNotFound {
+			return logs, nil
+		}
+
+		return nil, err
+	}
+
+	return logs, nil
+}
+
+func (r *logRepository) Insert(ctx context.Context, log models.Log) (*models.Log, error) {
+	nextShortId, err := r.shortIdCounter.Increment(1)
+	if err != nil {
+		return nil, err
+	}
+
+	log.ID = generate.LogID(log)
+	log.ShortID = "L" + strconv.Itoa(nextShortId)
+
+	if log.Open {
+		// There can be only one open log in the same channel.
+		r.openMutex.Lock()
+		defer r.openMutex.Unlock()
+
+		_, err = r.logs.UpdateAll(bson.M{"channel": log.ChannelName, "open": true}, bson.M{"$set": bson.M{"open": false}})
+		if err != nil {
+			return nil, errors.New("Cannot close other logs: " + err.Error())
+		}
+	}
+
+	err = r.logs.Insert(&log)
+	if err != nil {
+		return nil, err
+	}
+
+	return &log, nil
+}
+
+func (r *logRepository) Update(ctx context.Context, log models.Log, update models.LogUpdate) (*models.Log, error) {
+	updateBson := bson.M{}
+	if update.Open != nil {
+		if *update.Open == true {
+			// There can be only one open log in the same channel.
+			r.openMutex.Lock()
+			defer r.openMutex.Unlock()
+
+			_, err := r.logs.UpdateAll(bson.M{"channel": log.ChannelName, "open": true}, bson.M{"$set": bson.M{"open": false}})
+			if err != nil {
+				return nil, errors.New("Cannot close other logs: " + err.Error())
+			}
+		}
+
+		updateBson["open"] = *update.Open
+		log.Open = *update.Open
+	}
+	if update.Title != nil {
+		updateBson["title"] = *update.Title
+		log.Title = *update.Title
+	}
+	if update.Description != nil {
+		updateBson["description"] = *update.Description
+		log.Description = *update.Description
+	}
+	if update.EventName != nil {
+		updateBson["event"] = *update.EventName
+		log.EventName = *update.EventName
+	}
+
+	err := r.logs.UpdateId(log.ID, bson.M{"$set": updateBson})
+	if err != nil {
+		return nil, err
+	}
+
+	return &log, nil
+}
+
+func (r *logRepository) Delete(ctx context.Context, log models.Log) error {
+	err := r.logs.RemoveId(log.ID)
+	if err != nil {
+		return err
+	}
+
+	_, _ = r.posts.RemoveAll(bson.M{"logId": log.ShortID})
+
+	return nil
+}

database/mongodb/posts.go

@@ -0,0 +1,345 @@
+package mongodb
+
+import (
+	"context"
+	"git.aiterp.net/rpdata/api/internal/generate"
+	"git.aiterp.net/rpdata/api/models"
+	"git.aiterp.net/rpdata/api/repositories"
+	"github.com/globalsign/mgo"
+	"github.com/globalsign/mgo/bson"
+	"log"
+	"strings"
+	"sync"
+	"time"
+)
+
+type postRepository struct {
+	logs  *mgo.Collection
+	posts *mgo.Collection
+
+	orderMutex sync.Mutex
+}
+
+func newPostRepository(db *mgo.Database) (*postRepository, error) {
+	posts := db.C("logbot3.posts")
+
+	err := posts.EnsureIndexKey("logId")
+	if err != nil {
+		return nil, err
+	}
+	err = posts.EnsureIndexKey("time")
+	if err != nil {
+		return nil, err
+	}
+	err = posts.EnsureIndexKey("kind")
+	if err != nil {
+		return nil, err
+	}
+	err = posts.EnsureIndexKey("position")
+	if err != nil {
+		return nil, err
+	}
+
+	err = posts.EnsureIndex(mgo.Index{
+		Key: []string{"$text:text"},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &postRepository{
+		posts: posts,
+		logs:  db.C("logbot3.logs"),
+	}, nil
+}
+
+func (r *postRepository) Find(ctx context.Context, id string) (*models.Post, error) {
+	post := new(models.Post)
+	err := r.posts.FindId(id).One(post)
+	if err != nil {
+		return nil, err
+	}
+
+	return post, nil
+}
+
+func (r *postRepository) List(ctx context.Context, filter models.PostFilter) ([]*models.Post, error) {
+	query := bson.M{}
+	if filter.LogID != nil && *filter.LogID != "" {
+		logId := *filter.LogID
+		if !strings.HasPrefix(logId, "L") {
+			// Resolve long id to short id
+			log := new(models.Log)
+			err := r.logs.FindId(logId).Select(bson.M{"logId": 1, "_id": 1}).One(log)
+			if err != nil {
+				return nil, err
+			}
+
+			logId = log.ShortID
+		}
+
+		query["logId"] = logId
+	}
+	if len(filter.IDs) > 0 {
+		query["_id"] = bson.M{"$in": filter.IDs}
+	}
+	if len(filter.Kinds) > 0 {
+		query["kind"] = bson.M{"$in": filter.Kinds}
+	}
+	if filter.Search != nil {
+		query["$text"] = bson.M{"$search": *filter.Search}
+	}
+
+	posts := make([]*models.Post, 0, 32)
+	err := r.posts.Find(query).Sort("-logId", "position").Limit(filter.Limit).All(&posts)
+	if err != nil {
+		if err == mgo.ErrNotFound {
+			return []*models.Post{}, nil
+		}
+
+		return nil, err
+	}
+
+	return posts, nil
+}
+
+func (r *postRepository) Insert(ctx context.Context, post models.Post) (*models.Post, error) {
+	r.orderMutex.Lock()
+	defer r.orderMutex.Unlock()
+
+	lastPost := new(models.Post)
+	err := r.posts.Find(bson.M{"logId": post.LogID}).Sort("-position").One(lastPost)
+	if err != nil && err != mgo.ErrNotFound {
+		return nil, err
+	}
+
+	post.ID = generate.PostID()
+	post.Position = lastPost.Position + 1 // Position 1 is first position, so this is safe.
+
+	err = r.posts.Insert(post)
+	if err != nil {
+		return nil, err
+	}
+
+	return &post, nil
+}
+
+func (r *postRepository) InsertMany(ctx context.Context, posts ...*models.Post) ([]*models.Post, error) {
+	if len(posts) == 0 {
+		return []*models.Post{}, nil
+	}
+
+	logId := posts[0].LogID
+	for _, post := range posts[1:] {
+		if post.LogID != logId {
+			return nil, repositories.ErrParentMismatch
+		}
+
+		post.ID = generate.PostID()
+	}
+
+	r.orderMutex.Lock()
+	defer r.orderMutex.Unlock()
+
+	lastPost := new(models.Post)
+	err := r.posts.Find(bson.M{"logId": posts[0].LogID}).Sort("-position").One(lastPost)
+	if err != nil && err != mgo.ErrNotFound {
+		return nil, err
+	}
+
+	docs := make([]interface{}, len(posts))
+	for i := range posts {
+		posts[i].Position = lastPost.Position + 1 + i
+		docs[i] = posts[i]
+	}
+
+	err = r.posts.Insert(docs...)
+	if err != nil && err != mgo.ErrNotFound {
+		return nil, err
+	}
+
+	return posts, nil
+}
+
+func (r *postRepository) Update(ctx context.Context, post models.Post, update models.PostUpdate) (*models.Post, error) {
+	updateBson := bson.M{}
+	if update.Time != nil {
+		updateBson["time"] = *update.Time
+		post.Time = *update.Time
+	}
+	if update.Kind != nil {
+		updateBson["kind"] = *update.Kind
+		post.Kind = *update.Kind
+	}
+	if update.Nick != nil {
+		updateBson["nick"] = *update.Nick
+		post.Nick = *update.Nick
+	}
+	if update.Text != nil {
+		updateBson["text"] = *update.Text
+		post.Text = *update.Text
+	}
+
+	err := r.posts.UpdateId(post.ID, bson.M{"$set": updateBson})
+	if err != nil {
+		return nil, err
+	}
+
+	return &post, nil
+}
+
+func (r *postRepository) Move(ctx context.Context, post models.Post, position int) ([]*models.Post, error) {
+	// If only MongoDB transactions weren't awful, this function would have been safe.
+	// Since it isn't, then good luck.
+
+	// Validate lower position bound.
+	if position < 1 {
+		return nil, repositories.ErrInvalidPosition
+	}
+
+	// Determine the operations on adjacent posts
+	var resultFilter bson.M
+	var pushFilter bson.M
+	var increment int
+	if post.Position > position {
+		pushFilter = bson.M{"position": bson.M{
+			"$gte": position,
+			"$lt":  post.Position,
+		}}
+		increment = 1
+		resultFilter = bson.M{"position": bson.M{
+			"$lte": post.Position,
+			"$gte": position,
+		}}
+	} else {
+		pushFilter = bson.M{"position": bson.M{
+			"$lte": position,
+			"$gt":  post.Position,
+		}}
+		increment = -1
+		resultFilter = bson.M{"position": bson.M{
+			"$lte": position,
+			"$gte": post.Position,
+		}}
+	}
+	pushFilter["logId"] = post.LogID
+	resultFilter["logId"] = post.LogID
+
+	// From here on out, sync is required
+	r.orderMutex.Lock()
+	defer r.orderMutex.Unlock()
+
+	// Detect ninja shenanigans
+	post2 := new(models.Post)
+	err := r.posts.FindId(post.ID).One(post2)
+	if err != nil && post2.Position != post.Position {
+		return nil, repositories.ErrNotFound
+	}
+
+	// Validate upper position bound
+	lastPost := new(models.Post)
+	err = r.posts.Find(bson.M{"logId": post.LogID}).Sort("-position").One(lastPost)
+	if err != nil && err != mgo.ErrNotFound {
+		return nil, err
+	}
+	if position > lastPost.Position {
+		return nil, repositories.ErrInvalidPosition
+	}
+
+	// Move the posts
+	changeInfo, err := r.posts.UpdateAll(pushFilter, bson.M{"$inc": bson.M{"position": increment}})
+	if err != nil && err != mgo.ErrNotFound {
+		return nil, err
+	}
+	err = r.posts.UpdateId(post.ID, bson.M{"$set": bson.M{"position": position}})
+	if err != nil {
+		// Try to undo it
+		_, err := r.posts.UpdateAll(pushFilter, bson.M{"$inc": bson.M{"position": -increment}})
+		if err != nil && err != mgo.ErrNotFound {
+			return nil, err
+		}
+
+		return nil, err
+	}
+
+	results := make([]*models.Post, 0, changeInfo.Matched+1)
+	err = r.posts.Find(resultFilter).All(&results)
+	if err != nil {
+		return nil, err
+	}
+
+	return results, nil
+}
+
+func (r *postRepository) Delete(ctx context.Context, post models.Post) error {
+	r.orderMutex.Lock()
+	defer r.orderMutex.Unlock()
+
+	err := r.posts.RemoveId(post.ID)
+	if err != nil {
+		return err
+	}
+
+	_, _ = r.posts.UpdateAll(bson.M{"logId": post.LogID, "position": bson.M{"$gt": post.Position}}, bson.M{"$inc": bson.M{"position": -1}})
+
+	return nil
+}
+
+func (r *postRepository) fixPositions(logRepo *logRepository) {
+	disorders := make([]int, 0, 16)
+	diffs := make([]int, 0, 16)
+	startTime := time.Now()
+
+	timeout, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+
+	logs, err := logRepo.List(timeout, models.LogFilter{})
+	if err != nil {
+		log.Println("Failed to get logs for position fix:", err)
+	}
+
+	log.Println("Starting log position fixing, this should not take longer than 10 seconds.")
+
+	for _, l := range logs {
+		r.orderMutex.Lock()
+
+		posts, err := r.List(timeout, models.PostFilter{LogID: &l.ShortID})
+		if err != nil {
+			r.orderMutex.Unlock()
+			continue
+		}
+
+		disorders = disorders[:0]
+		diffs = diffs[:0]
+		for i, post := range posts {
+			if post.Position != (i + 1) {
+				disorders = append(disorders, i)
+				diffs = append(diffs, post.Position-(i+1))
+			}
+		}
+
+		if len(disorders) > 0 {
+			log.Println(len(disorders), "order errors detected in", l.ID)
+
+			ops := 0
+
+			for i, post := range posts {
+				if (i + 1) != posts[i].Position {
+					ops++
+
+					err := r.posts.UpdateId(post.ID, bson.M{"$set": bson.M{"position": i + 1}})
+					if err != nil {
+						log.Println(l.ShortID, "fix failed after", ops, "ops:", err)
+						break
+					}
+				}
+			}
+
+			log.Println(l.ShortID, "fixed after", ops, "ops.")
+		}
+
+		r.orderMutex.Unlock()
+	}
+
+	log.Println("Log position fixing finished in", time.Since(startTime))
+}