// DecodeIndex loads and unserializes an index from rd.
func DecodeIndex(rd io.Reader) (idx *Index, err error) {
	debug.Log("Index.DecodeIndex", "Start decoding index")
	idxJSON := jsonIndex{}

	dec := json.NewDecoder(rd)
	err = dec.Decode(&idxJSON)
	if err != nil {
		debug.Log("Index.DecodeIndex", "Error %v", err)

		if isErrOldIndex(err) {
			debug.Log("Index.DecodeIndex", "index is probably old format, trying that")
			err = ErrOldIndexFormat
		}

		return nil, err
	}

	idx = NewIndex()
	for _, pack := range idxJSON.Packs {
		for _, blob := range pack.Blobs {
			idx.store(PackedBlob{
				Type:   blob.Type,
				ID:     blob.ID,
				Offset: blob.Offset,
				Length: blob.Length,
				PackID: pack.ID,
			})
		}
	}
	idx.supersedes = idxJSON.Supersedes
	idx.final = true

	debug.Log("Index.DecodeIndex", "done")
	return idx, err
}

func (res *Restorer) restoreNodeTo(node *Node, dir string, dst string) error {
	debug.Log("Restorer.restoreNodeTo", "node %v, dir %v, dst %v", node.Name, dir, dst)
	dstPath := filepath.Join(dst, dir, node.Name)

	err := node.CreateAt(dstPath, res.repo)
	if err != nil {
		debug.Log("Restorer.restoreNodeTo", "node.CreateAt(%s) error %v", dstPath, err)
	}

	// Did it fail because of ENOENT?
	if err != nil && os.IsNotExist(errors.Cause(err)) {
		debug.Log("Restorer.restoreNodeTo", "create intermediate paths")

		// Create parent directories and retry
		err = os.MkdirAll(filepath.Dir(dstPath), 0700)
		if err == nil || err == os.ErrExist {
			err = node.CreateAt(dstPath, res.repo)
		}
	}

	if err != nil {
		debug.Log("Restorer.restoreNodeTo", "error %v", err)
		err = res.Error(dstPath, node, errors.Annotate(err, "create node"))
		if err != nil {
			return err
		}
	}

	debug.Log("Restorer.restoreNodeTo", "successfully restored %v", node.Name)

	return nil
}

// loadIndex loads the index id and merges it with the currently used index.
func (s *Repository) loadIndex(id string) error {
	debug.Log("Repo.loadIndex", "Loading index %v", id[:8])
	before := len(s.idx.pack)

	rd, err := s.be.Get(backend.Index, id)
	defer rd.Close()
	if err != nil {
		return err
	}

	// decrypt
	decryptRd, err := crypto.DecryptFrom(s.key, rd)
	defer decryptRd.Close()
	if err != nil {
		return err
	}

	idx, err := DecodeIndex(decryptRd)
	if err != nil {
		debug.Log("Repo.loadIndex", "error while decoding index %v: %v", id, err)
		return err
	}

	s.idx.Merge(idx)

	after := len(s.idx.pack)
	debug.Log("Repo.loadIndex", "Loaded index %v, added %v blobs", id[:8], after-before)

	return nil
}

// savePacker stores p in the backend.
func (s *Repository) savePacker(p *pack.Packer) error {
	debug.Log("Repo.savePacker", "save packer with %d blobs\n", p.Count())
	_, err := p.Finalize()
	if err != nil {
		return err
	}

	// move file to the final location
	sid := p.ID()
	err = p.Writer().(backend.Blob).Finalize(backend.Data, sid.String())
	if err != nil {
		debug.Log("Repo.savePacker", "blob Finalize() error: %v", err)
		return err
	}

	debug.Log("Repo.savePacker", "saved as %v", sid.Str())

	// update blobs in the index
	for _, b := range p.Blobs() {
		debug.Log("Repo.savePacker", "  updating blob %v to pack %v", b.ID.Str(), sid.Str())
		s.idx.Store(b.Type, b.ID, sid, b.Offset, uint(b.Length))
	}

	return nil
}

// Stale returns true if the lock is stale. A lock is stale if the timestamp is
// older than 30 minutes or if it was created on the current machine and the
// process isn't alive any more.
func (l *Lock) Stale() bool {
	debug.Log("Lock.Stale", "testing if lock %v for process %d is stale", l.lockID.Str(), l.PID)
	if time.Now().Sub(l.Time) > staleTimeout {
		debug.Log("Lock.Stale", "lock is stale, timestamp is too old: %v\n", l.Time)
		return true
	}

	hn, err := os.Hostname()
	if err != nil {
		debug.Log("Lock.Stale", "unable to find current hostnanme: %v", err)
		// since we cannot find the current hostname, assume that the lock is
		// not stale.
		return false
	}

	if hn != l.Hostname {
		// lock was created on a different host, assume the lock is not stale.
		return false
	}

	// check if we can reach the process retaining the lock
	exists := l.processExists()
	if !exists {
		debug.Log("Lock.Stale", "could not reach process, %d, lock is probably stale\n", l.PID)
		return true
	}

	debug.Log("Lock.Stale", "lock not stale\n")
	return false
}

// Walk starts walking the tree given by id. When the channel done is closed,
// processing stops.
func (tw *TreeWalker) Walk(path string, id backend.ID, done chan struct{}) {
	debug.Log("TreeWalker.Walk", "starting on tree %v for %v", id.Str(), path)
	defer debug.Log("TreeWalker.Walk", "done walking tree %v for %v", id.Str(), path)

	resCh := make(chan loadTreeResult, 1)
	tw.ch <- loadTreeJob{
		id:  id,
		res: resCh,
	}

	res := <-resCh
	if res.err != nil {
		select {
		case tw.out <- WalkTreeJob{Path: path, Error: res.err}:
		case <-done:
			return
		}
		return
	}

	tw.walk(path, res.tree, done)

	select {
	case tw.out <- WalkTreeJob{Path: path, Tree: res.tree}:
	case <-done:
		return
	}
}

// DecodeOldIndex loads and unserializes an index in the old format from rd.
func DecodeOldIndex(rd io.Reader) (idx *Index, err error) {
	debug.Log("Index.DecodeOldIndex", "Start decoding old index")
	list := []*packJSON{}

	dec := json.NewDecoder(rd)
	err = dec.Decode(&list)
	if err != nil {
		debug.Log("Index.DecodeOldIndex", "Error %#v", err)
		return nil, err
	}

	idx = NewIndex()
	for _, pack := range list {
		for _, blob := range pack.Blobs {
			idx.store(PackedBlob{
				Type:   blob.Type,
				ID:     blob.ID,
				PackID: pack.ID,
				Offset: blob.Offset,
				Length: blob.Length,
			})
		}
	}
	idx.final = true

	debug.Log("Index.DecodeOldIndex", "done")
	return idx, err
}

// savePacker stores p in the backend.
func (r *Repository) savePacker(p *pack.Packer) error {
	debug.Log("Repo.savePacker", "save packer with %d blobs\n", p.Count())
	data, err := p.Finalize()
	if err != nil {
		return err
	}

	id := backend.Hash(data)
	h := backend.Handle{Type: backend.Data, Name: id.String()}

	err = r.be.Save(h, data)
	if err != nil {
		debug.Log("Repo.savePacker", "Save(%v) error: %v", h, err)
		return err
	}

	debug.Log("Repo.savePacker", "saved as %v", h)

	// update blobs in the index
	for _, b := range p.Blobs() {
		debug.Log("Repo.savePacker", "  updating blob %v to pack %v", b.ID.Str(), id.Str())
		r.idx.Current().Store(PackedBlob{
			Type:   b.Type,
			ID:     b.ID,
			PackID: id,
			Offset: b.Offset,
			Length: uint(b.Length),
		})
	}

	return nil
}

func refreshLocks(wg *sync.WaitGroup, done <-chan struct{}) {
	debug.Log("main.refreshLocks", "start")
	defer func() {
		wg.Done()
		globalLocks.Lock()
		globalLocks.cancelRefresh = nil
		globalLocks.Unlock()
	}()

	ticker := time.NewTicker(refreshInterval)

	for {
		select {
		case <-done:
			debug.Log("main.refreshLocks", "terminate")
			return
		case <-ticker.C:
			debug.Log("main.refreshLocks", "refreshing locks")
			globalLocks.Lock()
			for _, lock := range globalLocks.locks {
				err := lock.Refresh()
				if err != nil {
					fmt.Fprintf(os.Stderr, "unable to refresh lock: %v\n", err)
				}
			}
			globalLocks.Unlock()
		}
	}
}

// Save stores data in the backend at the handle.
func (be s3) Save(h backend.Handle, p []byte) (err error) {
	if err := h.Valid(); err != nil {
		return err
	}

	debug.Log("s3.Save", "%v bytes at %d", len(p), h)

	path := s3path(h.Type, h.Name)

	// Check key does not already exist
	_, err = be.client.StatObject(be.bucketname, path)
	if err == nil {
		debug.Log("s3.blob.Finalize()", "%v already exists", h)
		return errors.New("key already exists")
	}

	<-be.connChan
	defer func() {
		be.connChan <- struct{}{}
	}()

	debug.Log("s3.Save", "PutObject(%v, %v, %v, %v)",
		be.bucketname, path, int64(len(p)), "binary/octet-stream")
	n, err := be.client.PutObject(be.bucketname, path, bytes.NewReader(p), "binary/octet-stream")
	debug.Log("s3.Save", "%v -> %v bytes, err %#v", path, n, err)

	return err
}

// repackBlob loads a single blob from src and saves it in dst.
func repackBlob(src, dst *repository.Repository, id backend.ID) error {
	blob, err := src.Index().Lookup(id)
	if err != nil {
		return err
	}

	debug.Log("RepackBlobs", "repacking blob %v, len %v", id.Str(), blob.PlaintextLength())

	buf := make([]byte, 0, blob.PlaintextLength())
	buf, err = src.LoadBlob(blob.Type, id, buf)
	if err != nil {
		return err
	}

	if uint(len(buf)) != blob.PlaintextLength() {
		debug.Log("RepackBlobs", "repack blob %v: len(buf) isn't equal to length: %v = %v", id.Str(), len(buf), blob.PlaintextLength())
		return errors.New("LoadBlob returned wrong data, len() doesn't match")
	}

	_, err = dst.SaveAndEncrypt(blob.Type, buf, &id)
	if err != nil {
		return err
	}

	return nil
}

// Save stores data in the backend at the handle.
func (r *SFTP) Save(h backend.Handle, p []byte) (err error) {
	if err := h.Valid(); err != nil {
		return err
	}

	filename, tmpfile, err := r.tempFile()
	debug.Log("sftp.Save", "save %v (%d bytes) to %v", h, len(p), filename)

	n, err := tmpfile.Write(p)
	if err != nil {
		return err
	}

	if n != len(p) {
		return errors.New("not all bytes writen")
	}

	err = tmpfile.Close()
	if err != nil {
		return err
	}

	err = r.renameFile(filename, h.Type, h.Name)
	debug.Log("sftp.Save", "save %v: rename %v: %v",
		h, filepath.Base(filename), err)
	if err != nil {
		return fmt.Errorf("sftp: renameFile: %v", err)
	}

	return nil
}

func packIDTester(repo *repository.Repository, inChan <-chan backend.ID, errChan chan<- error, wg *sync.WaitGroup, done <-chan struct{}) {
	debug.Log("Checker.testPackID", "worker start")
	defer debug.Log("Checker.testPackID", "worker done")

	defer wg.Done()

	for id := range inChan {
		ok, err := repo.Backend().Test(backend.Data, id.String())
		if err != nil {
			err = PackError{ID: id, Err: err}
		} else {
			if !ok {
				err = PackError{ID: id, Err: errors.New("does not exist")}
			}
		}

		if err != nil {
			debug.Log("Checker.testPackID", "error checking for pack %s: %v", id.Str(), err)
			select {
			case <-done:
				return
			case errChan <- err:
			}

			continue
		}

		debug.Log("Checker.testPackID", "pack %s exists", id.Str())
	}
}

// Create the backend specified by URI.
func create(s string) (backend.Backend, error) {
	debug.Log("open", "parsing location %v", s)
	loc, err := location.Parse(s)
	if err != nil {
		return nil, err
	}

	switch loc.Scheme {
	case "local":
		debug.Log("open", "create local repository at %#v", loc.Config)
		return local.Create(loc.Config.(string))
	case "sftp":
		debug.Log("open", "create sftp repository at %#v", loc.Config)
		return sftp.CreateWithConfig(loc.Config.(sftp.Config))
	case "s3":
		cfg := loc.Config.(s3.Config)
		if cfg.KeyID == "" {
			cfg.KeyID = os.Getenv("AWS_ACCESS_KEY_ID")

		}
		if cfg.Secret == "" {
			cfg.Secret = os.Getenv("AWS_SECRET_ACCESS_KEY")
		}

		debug.Log("open", "create s3 repository at %#v", loc.Config)
		return s3.Open(cfg)
	}

	debug.Log("open", "invalid repository scheme: %v", s)
	return nil, fmt.Errorf("invalid scheme %q", loc.Scheme)
}

func walkTree(repo *repository.Repository, path string, treeID backend.ID, done chan struct{}, jobCh chan<- WalkTreeJob) {
	debug.Log("walkTree", "start on %q (%v)", path, treeID.Str())

	t, err := LoadTree(repo, treeID)
	if err != nil {
		select {
		case jobCh <- WalkTreeJob{Path: path, Error: err}:
		case <-done:
			return
		}
		return
	}

	for _, node := range t.Nodes {
		p := filepath.Join(path, node.Name)
		if node.Type == "dir" {
			walkTree(repo, p, *node.Subtree, done, jobCh)
		} else {
			select {
			case jobCh <- WalkTreeJob{Path: p, Node: node}:
			case <-done:
				return
			}
		}
	}

	select {
	case jobCh <- WalkTreeJob{Path: path, Tree: t}:
	case <-done:
		return
	}

	debug.Log("walkTree", "done for %q (%v)", path, treeID.Str())
}

// findPacker returns a packer for a new blob of size bytes. Either a new one is
// created or one is returned that already has some blobs.
func (r *packerManager) findPacker(size uint) (*pack.Packer, error) {
	r.pm.Lock()
	defer r.pm.Unlock()

	// search for a suitable packer
	if len(r.packs) > 0 {
		debug.Log("Repo.findPacker", "searching packer for %d bytes\n", size)
		for i, p := range r.packs {
			if p.Size()+size < maxPackSize {
				debug.Log("Repo.findPacker", "found packer %v", p)
				// remove from list
				r.packs = append(r.packs[:i], r.packs[i+1:]...)
				return p, nil
			}
		}
	}

	// no suitable packer found, return new
	blob, err := r.be.Create()
	if err != nil {
		return nil, err
	}
	debug.Log("Repo.findPacker", "create new pack %p for %d bytes", blob, size)
	return pack.NewPacker(r.key, blob), nil
}

// savePacker stores p in the backend.
func (r *Repository) savePacker(p *pack.Packer) error {
	debug.Log("Repo.savePacker", "save packer with %d blobs\n", p.Count())
	_, err := p.Finalize()
	if err != nil {
		return err
	}

	// move file to the final location
	sid := p.ID()
	err = p.Writer().(backend.Blob).Finalize(backend.Data, sid.String())
	if err != nil {
		debug.Log("Repo.savePacker", "blob Finalize() error: %v", err)
		return err
	}

	debug.Log("Repo.savePacker", "saved as %v", sid.Str())

	// update blobs in the index
	for _, b := range p.Blobs() {
		debug.Log("Repo.savePacker", "  updating blob %v to pack %v", b.ID.Str(), sid.Str())
		r.idx.Current().Store(PackedBlob{
			Type:   b.Type,
			ID:     b.ID,
			PackID: sid,
			Offset: b.Offset,
			Length: uint(b.Length),
		})
		r.idx.RemoveFromInFlight(b.ID)
	}

	return nil
}

func (node *Node) isNewer(path string, fi os.FileInfo) bool {
	if node.Type != "file" {
		debug.Log("node.isNewer", "node %v is newer: not file", path)
		return true
	}

	tpe := nodeTypeFromFileInfo(fi)
	if node.Name != fi.Name() || node.Type != tpe {
		debug.Log("node.isNewer", "node %v is newer: name or type changed", path)
		return true
	}

	extendedStat := fi.Sys().(*syscall.Stat_t)
	inode := extendedStat.Ino
	size := uint64(extendedStat.Size)

	if node.ModTime != fi.ModTime() ||
		node.ChangeTime != changeTime(extendedStat) ||
		node.Inode != uint64(inode) ||
		node.Size != size {
		debug.Log("node.isNewer", "node %v is newer: timestamp, size or inode changed", path)
		return true
	}

	debug.Log("node.isNewer", "node %v is not newer", path)
	return false
}

// WalkTree walks the tree specified by id recursively and sends a job for each
// file and directory it finds. When the channel done is closed, processing
// stops.
func WalkTree(repo TreeLoader, id backend.ID, done chan struct{}, jobCh chan<- WalkTreeJob) {
	debug.Log("WalkTree", "start on %v, start workers", id.Str())

	load := func(id backend.ID) (*Tree, error) {
		tree := &Tree{}
		err := repo.LoadJSONPack(pack.Tree, id, tree)
		if err != nil {
			return nil, err
		}
		return tree, nil
	}

	ch := make(chan loadTreeJob)

	var wg sync.WaitGroup
	for i := 0; i < loadTreeWorkers; i++ {
		wg.Add(1)
		go loadTreeWorker(&wg, ch, load, done)
	}

	tw := NewTreeWalker(ch, jobCh)
	tw.Walk("", id, done)
	close(jobCh)

	close(ch)
	wg.Wait()

	debug.Log("WalkTree", "done")
}

func (j archiveJob) Copy() pipe.Job {
	if !j.hasOld {
		return j.new
	}

	// handle files
	if isRegularFile(j.new.Info()) {
		debug.Log("archiveJob.Copy", "   job %v is file", j.new.Path())

		// if type has changed, return new job directly
		if j.old.Node == nil {
			return j.new
		}

		// if file is newer, return the new job
		if j.old.Node.isNewer(j.new.Fullpath(), j.new.Info()) {
			debug.Log("archiveJob.Copy", "   job %v is newer", j.new.Path())
			return j.new
		}

		debug.Log("archiveJob.Copy", "   job %v add old data", j.new.Path())
		// otherwise annotate job with old data
		e := j.new.(pipe.Entry)
		e.Node = j.old.Node
		return e
	}

	// dirs and other types are just returned
	return j.new
}