<commit_before>package controller

import (
	"errors"
	"fmt"
	mongomodels "koding/db/models"
	"koding/db/mongodb/modelhelper"
	"socialapi/models"
	"time"

	"github.com/koding/logging"
	"labix.org/v2/mgo"
)

var (
	ErrMigrated     = errors.New("already migrated")
	kodingChannelId int64
)

type MigratorWorkerController struct {
	log logging.Logger
}

func NewMigratorWorkerController(log logging.Logger) (*MigratorWorkerController, error) {
	wc := &MigratorWorkerController{
		log: log,
	}

	return wc, nil
}

func (mwc *MigratorWorkerController) Start() error {
	o := modelhelper.Options{
		Sort: "meta.createdAt",
	}
	s := modelhelper.Selector{
		"socialMessageId": modelhelper.Selector{"$exists": false},
	}
	kodingChannel, err := createGroupChannel("koding", "koding")
	if err != nil {
		return fmt.Errorf("Koding channel cannot be created: %s", err)
	}
	kodingChannelId = kodingChannel.Id

	errCount := 0

	handleError := func(su *mongomodels.StatusUpdate, err error) {
		mwc.log.Error("an error occured for %s: %s", su.Id.Hex(), err)
		errCount++
	}

	for {
		o.Skip = errCount
		su, err := modelhelper.GetStatusUpdate(s, o)
		if err != nil {
			if err == mgo.ErrNotFound {
				mwc.log.Info("Migration completed")
				return nil
			}
			return fmt.Errorf("status update cannot be fetched: %s", err)
		}

		// create channel message
		cm, err := createChannelMessage(&su)
		if err != nil {
			handleError(&su, err)
			continue
		}

		// create reply messages
		cm.CreatedAt = su.Meta.CreatedAt
		cm.UpdatedAt = su.Meta.ModifiedAt
		cm.Update()

		// update mongo status update channelMessageId field
		if err := completePostMigration(&su, cm); err != nil {
			handleError(&su, err)
			continue
		}

		fmt.Printf("\n\nStatus update var %+v \n\n", su)
	}

	return nil
}

func createGroupChannel(name, groupName string) (*models.Channel, error) {
	c := models.NewChannel()
	c.Name = name
	c.GroupName = groupName
	c.TypeConstant = models.Channel_TYPE_GROUP

	group, err := modelhelper.GetGroup(groupName)
	if err != nil {
		return nil, err
	}

	// TODO check it causes any error
	c.PrivacyConstant = group.Privacy

	// find group owner
	creatorId, err := fetchGroupOwnerId(group)
	if err != nil {
		return nil, err
	}

	c.CreatorId = creatorId
	// create channel
	if err := c.Create(); err != nil {
		return nil, err
	}

	return c, nil
}

func fetchGroupOwnerId(g *mongomodels.Group) (int64, error) {
	// fetch owner relationship
	s := modelhelper.Selector{
		"targetName": "JAccount",
		"as":         "owner",
		"sourceId":   g.Id,
	}
	r, err := modelhelper.GetRelationship(s)
	if err != nil {
		return 0, err
	}

	a := models.NewAccount()
	a.OldId = r.TargetId.Hex()
	if err := a.FetchOrCreate(); err != nil {
		return 0, err
	}

	return a.Id, nil
}

func createChannelMessage(su *mongomodels.StatusUpdate) (*models.ChannelMessage, error) {
	cm := models.NewChannelMessage()
	cm.Slug = su.Slug
	cm.Body = su.Body
	cm.TypeConstant = models.ChannelMessage_TYPE_POST
	// cm.CreatedAt = su.Meta.CreatedAt
	cm.CreatedAt = time.Now().Add(-48 * time.Hour)
	cm.UpdatedAt = su.Meta.ModifiedAt

	if err := prepareMessageAccount(cm, su); err != nil {
		return nil, err
	}

	if err := prepareMessageChannel(cm, su); err != nil {
		return nil, err
	}

	if err := cm.CreateRaw(); err != nil {
		return nil, err
	}

	return cm, nil
}

func prepareMessageAccount(cm *models.ChannelMessage, su *mongomodels.StatusUpdate) error {
	a := models.NewAccount()
	a.OldId = su.OriginId.Hex()<commit_msg>Migration: CreatedAt must be lte UpdatedAt constraint violation is handled<commit_after>package controller

import (
	"errors"
	"fmt"
	mongomodels "koding/db/models"
	"koding/db/mongodb/modelhelper"
	"socialapi/models"

	"github.com/koding/logging"
	"labix.org/v2/mgo"
)

var (
	ErrMigrated     = errors.New("already migrated")
	kodingChannelId int64
)

type MigratorWorkerController struct {
	log logging.Logger
}

func NewMigratorWorkerController(log logging.Logger) (*MigratorWorkerController, error) {
	wc := &MigratorWorkerController{
		log: log,
	}

	return wc, nil
}

func (mwc *MigratorWorkerController) Start() error {
	o := modelhelper.Options{
		Sort: "meta.createdAt",
	}
	s := modelhelper.Selector{
		"socialMessageId": modelhelper.Selector{"$exists": false},
	}
	kodingChannel, err := createGroupChannel("koding", "koding")
	if err != nil {
		return fmt.Errorf("Koding channel cannot be created: %s", err)
	}
	kodingChannelId = kodingChannel.Id

	errCount := 0

	handleError := func(su *mongomodels.StatusUpdate, err error) {
		mwc.log.Error("an error occured for %s: %s", su.Id.Hex(), err)
		errCount++
	}

	for {
		o.Skip = errCount
		su, err := modelhelper.GetStatusUpdate(s, o)
		if err != nil {
			if err == mgo.ErrNotFound {
				mwc.log.Notice("Migration completed")
				return nil
			}
			return fmt.Errorf("status update cannot be fetched: %s", err)
		}

		// create channel message
		cm, err := createChannelMessage(&su)
		if err != nil {
			handleError(&su, err)
			continue
		}

		// create reply messages

		// update mongo status update channelMessageId field
		if err := completePostMigration(&su, cm); err != nil {
			handleError(&su, err)
			continue
		}

		fmt.Printf("\n\nStatus update var %+v \n\n", su)
	}

	return nil
}

func createGroupChannel(name, groupName string) (*models.Channel, error) {
	c := models.NewChannel()
	c.Name = name
	c.GroupName = groupName
	c.TypeConstant = models.Channel_TYPE_GROUP

	group, err := modelhelper.GetGroup(groupName)
	if err != nil {
		return nil, err
	}

	// TODO check it causes any error
	c.PrivacyConstant = group.Privacy

	// find group owner
	creatorId, err := fetchGroupOwnerId(group)
	if err != nil {
		return nil, err
	}

	c.CreatorId = creatorId
	// create channel
	if err := c.Create(); err != nil {
		return nil, err
	}

	return c, nil
}

func fetchGroupOwnerId(g *mongomodels.Group) (int64, error) {
	// fetch owner relationship
	s := modelhelper.Selector{
		"targetName": "JAccount",
		"as":         "owner",
		"sourceId":   g.Id,
	}
	r, err := modelhelper.GetRelationship(s)
	if err != nil {
		return 0, err
	}

	a := models.NewAccount()
	a.OldId = r.TargetId.Hex()
	if err := a.FetchOrCreate(); err != nil {
		return 0, err
	}

	return a.Id, nil
}

func createChannelMessage(su *mongomodels.StatusUpdate) (*models.ChannelMessage, error) {
	cm := models.NewChannelMessage()
	cm.Slug = su.Slug
	cm.Body = su.Body
	cm.TypeConstant = models.ChannelMessage_TYPE_POST
	cm.CreatedAt = su.Meta.CreatedAt
	// this is added because status update->modified at field is before createdAt
	if cm.CreatedAt.After(su.Meta.ModifiedAt) {
		cm.UpdatedAt = cm.CreatedAt
	} else {
		cm.UpdatedAt = su.Meta.ModifiedAt
	}

	// cm.CreatedAt = time.Now().Add(-48 * time.Hour)

	if err := prepareMessageAccount(cm, su); err != nil {
		return nil, err
	}

	if err := prepareMessageChannel(cm, su); err != nil {
		return nil, err
	}

	if err := cm.CreateRaw(); err != nil {
		return nil, err
	}

	return cm, nil
}

func prepareMessageAccount(cm *models.ChannelMessage, su *mongomodels.StatusUpdate) error {
	a := models.NewAccount()
	a.OldId = su.OriginId.Hex()
{'fraction_non_alphanumeric': 0.13332394762776292, 'fraction_numerical': 0.002815711671124877, 'mean_word_length': 3.720265780730897, 'pattern_counts': {'":': 10, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5213858', 'n_tokens_mistral': 2656, 'n_tokens_neox': 2485, 'n_words': 636}