<commit_before>package migrations

import (
	"fmt"
	"io"
	"os"
	"runtime"
	"strings"

	"github.com/cozy/cozy-stack/pkg/config"
	"github.com/cozy/cozy-stack/pkg/consts"
	"github.com/cozy/cozy-stack/pkg/couchdb"
	"github.com/cozy/cozy-stack/pkg/instance"
	"github.com/cozy/cozy-stack/pkg/jobs"
	"github.com/cozy/cozy-stack/pkg/vfs/vfsswift"
	"github.com/cozy/swift"
	multierror "github.com/hashicorp/go-multierror"
)

const swiftV1ContainerPrefixCozy = "cozy-"
const swiftV1ContainerPrefixData = "data-"
const swiftV2ContainerPrefixCozy = "cozy-v2-"
const swiftV2ContainerPrefixData = "data-v2"
const versionSuffix = "-version"
const dirContentType = "directory"

func init() {
	jobs.AddWorker(&jobs.WorkerConfig{
		WorkerType:   "migrations",
		Concurrency:  runtime.NumCPU(),
		MaxExecCount: 2,
		WorkerFunc:   worker,
		WorkerCommit: commit,
	})
}

const swiftV1ToV2 = "swift-v1-to-v2"

type message struct {
	Type    string `json:"type"`
	Cluster int    `json:"cluster"`
}

func worker(ctx *jobs.WorkerContext) error {
	domain := ctx.Domain()

	var msg message
	if err := ctx.UnmarshalMessage(&msg); err != nil {
		return err
	}

	switch msg.Type {
	case swiftV1ToV2:
		return migrateSwiftV1ToV2(domain)
	default:
		return fmt.Errorf("unknown migration type %q", msg.Type)
	}
}

func commit(ctx *jobs.WorkerContext, err error) error {
	if err != nil {
		return nil
	}
	domain := ctx.Domain()

	var msg message
	if err := ctx.UnmarshalMessage(&msg); err != nil {
		return err
	}

	switch msg.Type {
	case swiftV1ToV2:
		return commitSwiftV1ToV2(domain, msg.Cluster)
	default:
		return fmt.Errorf("unknown migration type %q", msg.Type)
	}
}

type object struct {
	obj          swift.Object
	containerSrc string
	containerDst string
}

func migrateSwiftV1ToV2(domain string) error {
	c := config.GetSwiftConnection()
	inst, err := instance.Get(domain)
	if err != nil {
		return err
	}
	if inst.SwiftCluster > 0 {
		return nil
	}

	containerV1 := swiftV1ContainerPrefixCozy + domain
	containerV2 := swiftV2ContainerPrefixCozy + domain

	// container containing thumbnails
	containerV1Data := swiftV1ContainerPrefixData + domain
	containerV2Data := swiftV2ContainerPrefixData + domain

	err = c.VersionContainerCreate(containerV2, containerV2+versionSuffix)
	if err != nil {
		return err
	}

	objc := make(chan object)
	errc := make(chan error)

	go func() {
		errc <- readObjects(c, objc, containerV1, containerV2)
		errc <- readObjects(c, objc, containerV1Data, containerV2Data)
		close(objc)
	}()

	const N = 4

	for i := 0; i < N; i++ {
		go copyObjects(c, inst, objc, errc)
	}

	var errm error
	done := N
	for {
		err := <-errc
		if err != nil {
			errm = multierror.Append(errm, err)
		} else {
			done--
		}
		if done == 0 {
			break
		}
	}

	return errm
}

func commitSwiftV1ToV2(domain string, swiftCluster int) error {
	inst, err := instance.Get(domain)
	if err != nil {
		return err
	}
	if swiftCluster == 0 {
		swiftCluster = 1
	}
	inst.SwiftCluster = swiftCluster
	return instance.Update(inst)
}

func readObjects(c *swift.Connection, objc chan object,
	containerSrc, containerDst string) error {
	return c.ObjectsWalk(containerSrc, nil, func(opts *swift.ObjectsOpts) (interface{}, error) {
		objs, err := c.Objects(containerSrc, opts)<commit_msg>Set metadata on migrated containers<commit_after>package migrations

import (
	"fmt"
	"io"
	"os"
	"runtime"
	"strings"

	"github.com/cozy/cozy-stack/pkg/config"
	"github.com/cozy/cozy-stack/pkg/consts"
	"github.com/cozy/cozy-stack/pkg/couchdb"
	"github.com/cozy/cozy-stack/pkg/instance"
	"github.com/cozy/cozy-stack/pkg/jobs"
	"github.com/cozy/cozy-stack/pkg/vfs/vfsswift"
	"github.com/cozy/swift"
	multierror "github.com/hashicorp/go-multierror"
)

const swiftV1ContainerPrefixCozy = "cozy-"
const swiftV1ContainerPrefixData = "data-"
const swiftV2ContainerPrefixCozy = "cozy-v2-"
const swiftV2ContainerPrefixData = "data-v2"
const versionSuffix = "-version"
const dirContentType = "directory"

func init() {
	jobs.AddWorker(&jobs.WorkerConfig{
		WorkerType:   "migrations",
		Concurrency:  runtime.NumCPU(),
		MaxExecCount: 2,
		WorkerFunc:   worker,
		WorkerCommit: commit,
	})
}

const swiftV1ToV2 = "swift-v1-to-v2"

type message struct {
	Type    string `json:"type"`
	Cluster int    `json:"cluster"`
}

func worker(ctx *jobs.WorkerContext) error {
	domain := ctx.Domain()

	var msg message
	if err := ctx.UnmarshalMessage(&msg); err != nil {
		return err
	}

	switch msg.Type {
	case swiftV1ToV2:
		return migrateSwiftV1ToV2(domain)
	default:
		return fmt.Errorf("unknown migration type %q", msg.Type)
	}
}

func commit(ctx *jobs.WorkerContext, err error) error {
	if err != nil {
		return nil
	}
	domain := ctx.Domain()

	var msg message
	if err := ctx.UnmarshalMessage(&msg); err != nil {
		return err
	}

	switch msg.Type {
	case swiftV1ToV2:
		return commitSwiftV1ToV2(domain, msg.Cluster)
	default:
		return fmt.Errorf("unknown migration type %q", msg.Type)
	}
}

type object struct {
	obj          swift.Object
	containerSrc string
	containerDst string
}

func migrateSwiftV1ToV2(domain string) error {
	c := config.GetSwiftConnection()
	inst, err := instance.Get(domain)
	if err != nil {
		return err
	}
	if inst.SwiftCluster > 0 {
		return nil
	}

	containerV1 := swiftV1ContainerPrefixCozy + domain
	containerV2 := swiftV2ContainerPrefixCozy + domain

	// container containing thumbnails
	containerV1Data := swiftV1ContainerPrefixData + domain
	containerV2Data := swiftV2ContainerPrefixData + domain

	err = c.VersionContainerCreate(containerV2, containerV2+versionSuffix)
	if err != nil {
		return err
	}

	objc := make(chan object)
	errc := make(chan error)

	go func() {
		errc <- readObjects(c, objc, containerV1, containerV2)
		errc <- readObjects(c, objc, containerV1Data, containerV2Data)
		close(objc)
	}()

	const N = 4

	for i := 0; i < N; i++ {
		go copyObjects(c, inst, objc, errc)
	}

	var errm error
	done := N
	for {
		err := <-errc
		if err != nil {
			errm = multierror.Append(errm, err)
		} else {
			done--
		}
		if done == 0 {
			break
		}
	}

	return errm
}

func commitSwiftV1ToV2(domain string, swiftCluster int) error {
	inst, err := instance.Get(domain)
	if err != nil {
		return err
	}

	c := config.GetSwiftConnection()

	containerName := swiftV1ContainerPrefixCozy + domain
	containerMeta := &swift.Metadata{"cozy-v1-migrated": "1"}
	err = c.ContainerUpdate(containerName, containerMeta.ContainerHeaders())
	if err != nil {
		return err
	}

	if swiftCluster == 0 {
		swiftCluster = 1
	}
	inst.SwiftCluster = swiftCluster
	return instance.Update(inst)
}

func readObjects(c *swift.Connection, objc chan object,
	containerSrc, containerDst string) error {
	return c.ObjectsWalk(containerSrc, nil, func(opts *swift.ObjectsOpts) (interface{}, error) {
		objs, err := c.Objects(containerSrc, opts)
{'fraction_non_alphanumeric': 0.12203739143235684, 'fraction_numerical': 0.013101722361254233, 'mean_word_length': 4.085329341317365, 'pattern_counts': {'":': 1, '<': 11, '<?xml version=': 0, '>': 5, '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': '1670965', 'n_tokens_mistral': 2604, 'n_tokens_neox': 2438, 'n_words': 571}