// Copyright 2019 Dolthub, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// This file incorporates work covered by the following copyright and
// permission notice:
//
// Copyright 2016 Attic Labs, Inc. All rights reserved.
// Licensed under the Apache License, version 2.0:
// http://www.apache.org/licenses/LICENSE-2.0

package datas

import (
	"container/heap"
	"context"
	"errors"
	"fmt"
	"sort"
	"time"

	flatbuffers "github.com/dolthub/flatbuffers/v23/go"

	"github.com/dolthub/dolt/go/gen/fb/serial"
	"github.com/dolthub/dolt/go/store/chunks"
	"github.com/dolthub/dolt/go/store/hash"
	"github.com/dolthub/dolt/go/store/prolly/tree"
	"github.com/dolthub/dolt/go/store/types"
)

const (
	parentsField = "parents"
	// Added in July, 2020. Commits created with versions before this was
	// added have only a Set of parents. Commits created after this was
	// added carry a List of parents, because parent order can matter.
	// `"parents"` is still written as a Set as well, so that commits
	// created with newer versions of still usable by older versions.
	parentsListField = "parents_list"
	// Added in October, 2021. Stores a Ref<Value(Map<Tuple,List>)>.
	// The key of the map is a Tuple<Height, InlineBlob(Hash)>, reffable to
	// a Commit ref. The value of the map is a List of Ref<Value>s pointing
	// to the parents of the Commit which corresponds to the key.
	//
	// This structure is a materialized closure of a commit's parents. It
	// is used for pull/fetch/push commit graph fan-out and for sub-O(n)
	// FindCommonAncestor calculations.
	parentsClosureField = "parents_closure"
	valueField          = "value"
	commitMetaField     = "meta"
	commitName          = "Commit"
)

var ErrCommitNotFound = errors.New("target commit not found")
var ErrNotACommit = errors.New("value is not a commit")

type Commit struct {
	val    types.Value
	addr   hash.Hash
	height uint64
}

func (c *Commit) NomsValue() types.Value {
	return c.val
}

func (c *Commit) IsGhost() bool {
	_, ok := c.val.(types.GhostValue)
	return ok
}

func (c *Commit) Height() uint64 {
	return c.height
}

func (c *Commit) Addr() hash.Hash {
	return c.addr
}

func NewCommitForValue(ctx context.Context, cs chunks.ChunkStore, vrw types.ValueReadWriter, ns tree.NodeStore, v types.Value, opts CommitOptions) (*Commit, error) {
	if opts.Parents == nil || len(opts.Parents) == 0 {
		return nil, errors.New("cannot create commit without parents")
	}

	return newCommitForValue(ctx, cs, vrw, ns, v, opts)
}

// createStringIfDiffers writes |s| to |b| and returns its offset when |s| differs from |ref|.
// When they match it returns a zero offset so the field is omitted from the serialized message.
func createStringIfDiffers(b *flatbuffers.Builder, s, ref string) flatbuffers.UOffsetT {
	if s == ref {
		return 0
	}
	return b.CreateString(s)
}

func commit_flatbuffer(vaddr hash.Hash, opts CommitOptions, heights []uint64, parentsClosureAddr hash.Hash) (serial.Message, uint64) {
	builder := flatbuffers.NewBuilder(1024)
	vaddroff := builder.CreateByteVector(vaddr[:])

	hashsz := 20
	hashessz := len(opts.Parents) * hashsz
	builder.Prep(flatbuffers.SizeUOffsetT, hashessz)
	stop := int(builder.Head())
	start := stop - hashessz
	for i := 0; i < len(opts.Parents); i++ {
		copy(builder.Bytes[start:stop], opts.Parents[i][:])
		start += hashsz
	}
	start = stop - hashessz
	parentaddrsoff := builder.CreateByteVector(builder.Bytes[start:stop])

	maxheight := uint64(0)
	for _, h := range heights {
		if h > maxheight {
			maxheight = h
		}
	}

	pcaddroff := builder.CreateByteVector(parentsClosureAddr[:])

	nameoff := builder.CreateString(opts.Meta.Author.Name)
	emailoff := builder.CreateString(opts.Meta.Author.Email)
	descoff := builder.CreateString(opts.Meta.Description)

	var sigoff flatbuffers.UOffsetT
	if len(opts.Meta.Signature) != 0 {
		sigoff = builder.CreateString(opts.Meta.Signature)
	}

	// Committer name and email are only written when they differ from the author, so commits where
	// author and committer are the same produce the same hash as before committer identity was added.
	committerNameOff := createStringIfDiffers(builder, opts.Meta.Committer.Name, opts.Meta.Author.Name)
	committerEmailOff := createStringIfDiffers(builder, opts.Meta.Committer.Email, opts.Meta.Author.Email)

	serial.CommitStart(builder)
	serial.CommitAddRoot(builder, vaddroff)
	serial.CommitAddHeight(builder, maxheight+1)
	serial.CommitAddParentAddrs(builder, parentaddrsoff)
	serial.CommitAddParentClosure(builder, pcaddroff)
	serial.CommitAddName(builder, nameoff)
	serial.CommitAddEmail(builder, emailoff)
	serial.CommitAddDescription(builder, descoff)
	serial.CommitAddTimestampMillis(builder, opts.Meta.TimestampMillis())
	serial.CommitAddUserTimestampMillis(builder, opts.Meta.UserTimestampMillis())
	serial.CommitAddSignature(builder, sigoff)
	serial.CommitAddCommitterName(builder, committerNameOff)
	serial.CommitAddCommitterEmail(builder, committerEmailOff)

	bytes := serial.FinishMessage(builder, serial.CommitEnd(builder), []byte(serial.CommitFileID))
	return bytes, maxheight + 1
}

func newCommitForValue(ctx context.Context, cs chunks.ChunkStore, vrw types.ValueReadWriter, ns tree.NodeStore, v types.Value, opts CommitOptions) (*Commit, error) {
	now := CommitNow()
	opts.Meta.Author.Date = opts.Meta.Author.Date.Or(now)
	opts.Meta.Committer.Date = opts.Meta.Committer.Date.Or(now)

	if opts.Signer != nil {
		payload := SignaturePayloadV2(opts.DBName, opts.Meta, opts.HeadHash.String(), opts.StagedHash.String())
		signature, err := opts.Signer.Sign(ctx, []byte(payload))
		if err != nil {
			return nil, fmt.Errorf("failed to sign commit: %w", err)
		}
		opts.Meta.Signature = string(signature)
	}

	r, err := vrw.WriteValue(ctx, v)
	if err != nil {
		return nil, err
	}
	parents := make([]*serial.Commit, len(opts.Parents))
	heights := make([]uint64, len(opts.Parents))
	parentValues, err := vrw.ReadManyValues(ctx, opts.Parents)
	if err != nil {
		return nil, err
	}
	for i := range heights {
		parents[i], err = serial.TryGetRootAsCommit([]byte(parentValues[i].(types.SerialMessage)), serial.MessagePrefixSz)
		if err != nil {
			return nil, err
		}
		heights[i] = parents[i].Height()
	}
	parentClosureAddr, err := writeFbCommitParentClosure(ctx, cs, vrw, ns, parents, opts.Parents)
	if err != nil {
		return nil, err
	}
	bs, height := commit_flatbuffer(r.TargetHash(), opts, heights, parentClosureAddr)

	v = types.SerialMessage(bs)
	addr, err := v.Hash(vrw.Format())
	if err != nil {
		return nil, err
	}
	return &Commit{v, addr, height}, nil
}

func commitPtr(nbf *types.NomsBinFormat, v types.Value, r *types.Ref) (*Commit, error) {
	bs := []byte(v.(types.SerialMessage))
	var cm serial.Commit
	err := serial.InitCommitRoot(&cm, bs, serial.MessagePrefixSz)
	if err != nil {
		return nil, err
	}
	var addr hash.Hash
	if r != nil {
		addr = r.TargetHash()
	} else {
		addr, err = v.Hash(nbf)
		if err != nil {
			return nil, err
		}
	}
	return &Commit{
		val:    v,
		height: cm.Height(),
		addr:   addr,
	}, nil
}

// CommitFromValue deserializes a types.Value into a Commit.
func CommitFromValue(nbf *types.NomsBinFormat, v types.Value) (*Commit, error) {
	if g, ok := v.(types.GhostValue); ok {
		return &Commit{val: g}, nil
	}

	isCommit, err := IsCommit(v)
	if err != nil {
		return nil, err
	}
	if !isCommit {
		return nil, ErrNotACommit
	}
	return commitPtr(nbf, v, nil)
}

func LoadCommitRef(ctx context.Context, vr types.ValueReader, r types.Ref) (*Commit, error) {
	v, err := vr.ReadValue(ctx, r.TargetHash())
	if err != nil {
		return nil, err
	}
	if v == nil {
		return nil, ErrCommitNotFound
	}
	return commitPtr(vr.Format(), v, &r)
}

func LoadCommitAddr(ctx context.Context, vr types.ValueReader, addr hash.Hash) (*Commit, error) {
	v, err := vr.ReadValue(ctx, addr)
	if err != nil {
		return nil, err
	}
	if v == nil {
		return nil, ErrCommitNotFound
	}
	return CommitFromValue(vr.Format(), v)
}

func findCommonAncestorUsingParentsList(ctx context.Context, c1, c2 *Commit, vr1, vr2 types.ValueReader, ns1, ns2 tree.NodeStore) (hash.Hash, bool, error) {
	c1Q, c2Q := CommitByHeightHeap{c1}, CommitByHeightHeap{c2}
	for !c1Q.Empty() && !c2Q.Empty() {
		c1Ht, c2Ht := c1Q.MaxHeight(), c2Q.MaxHeight()
		if c1Ht == c2Ht {
			c1Parents, c2Parents := c1Q.PopCommitsOfHeight(c1Ht), c2Q.PopCommitsOfHeight(c2Ht)
			if common, ok := findCommonCommit(c1Parents, c2Parents); ok {
				return common.Addr(), true, nil
			}
			err := parentsToQueue(ctx, c1Parents, &c1Q, vr1)
			if err != nil {
				return hash.Hash{}, false, err
			}
			err = parentsToQueue(ctx, c2Parents, &c2Q, vr2)
			if err != nil {
				return hash.Hash{}, false, err
			}
		} else if c1Ht > c2Ht {
			err := parentsToQueue(ctx, c1Q.PopCommitsOfHeight(c1Ht), &c1Q, vr1)
			if err != nil {
				return hash.Hash{}, false, err
			}
		} else {
			err := parentsToQueue(ctx, c2Q.PopCommitsOfHeight(c2Ht), &c2Q, vr2)
			if err != nil {
				return hash.Hash{}, false, err
			}
		}
	}

	return hash.Hash{}, false, nil
}

// FindCommonAncestor returns the most recent common ancestor of c1 and c2, if
// one exists, setting ok to true. If there is no common ancestor, ok is set
// to false. Refs of |c1| are dereferenced through |vr1|, while refs of |c2|
// are dereferenced through |vr2|.
//
// This implementation makes use of the parents_closure field on the commit
// struct.  If the commit does not have a materialized parents_closure, this
// implementation delegates to findCommonAncestorUsingParentsList.
func FindCommonAncestor(ctx context.Context, c1, c2 *Commit, vr1, vr2 types.ValueReader, ns1, ns2 tree.NodeStore) (hash.Hash, bool, error) {
	pi1, err := newParentsClosureIterator(ctx, c1, vr1, ns1)
	if err != nil {
		return hash.Hash{}, false, err
	}
	if pi1 == nil {
		return findCommonAncestorUsingParentsList(ctx, c1, c2, vr1, vr2, ns1, ns2)
	}

	pi2, err := newParentsClosureIterator(ctx, c2, vr2, ns2)
	if err != nil {
		return hash.Hash{}, false, err
	}
	if pi2 == nil {
		return findCommonAncestorUsingParentsList(ctx, c1, c2, vr1, vr2, ns1, ns2)
	}

	for {
		h1, h2 := pi1.Hash(), pi2.Hash()
		if h1 == h2 {
			if err := firstError(pi1.Err(), pi2.Err()); err != nil {
				return hash.Hash{}, false, err
			}
			return h1, true, nil
		}
		if pi1.Less(ctx, vr1.Format(), pi2) {
			// TODO: Should pi2.Seek(pi1.curr), but MapIterator does not expose Seek yet.
			if !pi2.Next(ctx) {
				return hash.Hash{}, false, firstError(pi1.Err(), pi2.Err())
			}
		} else {
			if !pi1.Next(ctx) {
				return hash.Hash{}, false, firstError(pi1.Err(), pi2.Err())
			}
		}
	}
}

// FindClosureCommonAncestor returns the most recent common ancestor of |cl| and |cm|,
// where |cl| is the transitive closure of one or more refs. If a common ancestor
// exists, |ok| is set to true, else false.
func FindClosureCommonAncestor(ctx context.Context, cl CommitClosure, cm *Commit, vr types.ValueReader) (a hash.Hash, ok bool, err error) {
	q := &CommitByHeightHeap{cm}
	var curr []*Commit

	for !q.Empty() {
		curr = q.PopCommitsOfHeight(q.MaxHeight())

		for _, r := range curr {
			ok, err = cl.Contains(ctx, r)
			if err != nil {
				return hash.Hash{}, false, err
			}
			if ok {
				return r.Addr(), ok, nil
			}
		}

		err = parentsToQueue(ctx, curr, q, vr)
		if err != nil {
			return hash.Hash{}, false, err
		}
	}

	return hash.Hash{}, false, nil
}

// GetCommitParents returns |Ref|s to the parents of the commit.
func GetCommitParents(ctx context.Context, vr types.ValueReader, cv types.Value) ([]*Commit, error) {
	_, ok := cv.(types.GhostValue)
	if ok {
		// Not using the common error here because they are in the doltdb package which results in a cycle.
		return nil, fmt.Errorf("runtime exception. GetCommitParents called with GhostCommit.")
	}

	sm, ok := cv.(types.SerialMessage)
	if !ok {
		return nil, fmt.Errorf("expected SerialMessage type for commit value, found %T", cv)
	}

	data := []byte(sm)
	if serial.GetFileID(data) != serial.CommitFileID {
		return nil, errors.New("GetCommitParents: provided value is not a commit.")
	}
	addrs, err := types.SerialCommitParentAddrs(vr.Format(), sm)
	if err != nil {
		return nil, err
	}

	vals, err := vr.ReadManyValues(ctx, addrs)
	if err != nil {
		return nil, err
	}
	res := make([]*Commit, len(vals))
	for i, v := range vals {
		if v == nil {
			return nil, fmt.Errorf("GetCommitParents: Did not find parent Commit in ValueReader: %s", addrs[i].String())
		}

		if g, ok := v.(types.GhostValue); ok {
			res[i] = &Commit{
				val:  g,
				addr: addrs[i],
			}
		} else {
			var csm serial.Commit
			err := serial.InitCommitRoot(&csm, []byte(v.(types.SerialMessage)), serial.MessagePrefixSz)
			if err != nil {
				return nil, err
			}
			res[i] = &Commit{
				val:    v,
				height: csm.Height(),
				addr:   addrs[i],
			}
		}
	}
	return res, nil
}

// GetCommitMeta extracts the CommitMeta field from a commit. Returns |nil,
// nil| if there is no metadata for the commit.
func GetCommitMeta(ctx context.Context, cv types.Value) (*CommitMeta, error) {
	sm, ok := cv.(types.SerialMessage)
	if !ok {
		return nil, fmt.Errorf("expected SerialMessage type for commit value, found %T", cv)
	}

	data := []byte(sm)
	if serial.GetFileID(data) != serial.CommitFileID {
		return nil, errors.New("GetCommitMeta: provided value is not a commit.")
	}
	var serializedCommit serial.Commit
	err := serial.InitCommitRoot(&serializedCommit, data, serial.MessagePrefixSz)
	if err != nil {
		return nil, err
	}
	ret := &CommitMeta{
		Author: CommitIdent{
			Name:  string(serializedCommit.Name()),
			Email: string(serializedCommit.Email()),
			Date:  CommitDateAt(time.UnixMilli(serializedCommit.UserTimestampMillis())),
		},
		Description: string(serializedCommit.Description()),
		Signature:   string(serializedCommit.Signature()),
	}
	// Committer fields default to the author when the stored commit has no separate committer identity.
	ret.Committer.Name = ret.Author.Name
	if b := serializedCommit.CommitterName(); b != nil {
		ret.Committer.Name = string(b)
	}
	ret.Committer.Email = ret.Author.Email
	if b := serializedCommit.CommitterEmail(); b != nil {
		ret.Committer.Email = string(b)
	}
	ret.Committer.Date = CommitDateAt(time.UnixMilli(int64(serializedCommit.TimestampMillis())))
	return ret, nil
}

// GetCommittedValue returns the value of a commit. If the commit isn't found, nil is returned.
func GetCommittedValue(ctx context.Context, vr types.ValueReader, cv types.Value) (types.Value, error) {
	sm, ok := cv.(types.SerialMessage)
	if !ok {
		return nil, fmt.Errorf("expected SerialMessage type for commit value, found %T", cv)
	}

	data := []byte(sm)
	if serial.GetFileID(data) != serial.CommitFileID {
		return nil, errors.New("GetCommittedValue: provided value is not a commit.")
	}
	var cmsg serial.Commit
	err := serial.InitCommitRoot(&cmsg, data, serial.MessagePrefixSz)
	if err != nil {
		return nil, err
	}
	var roothash hash.Hash
	copy(roothash[:], cmsg.RootBytes())
	return vr.ReadValue(ctx, roothash)
}

func GetCommitRootHash(cv types.Value) (hash.Hash, error) {
	sm, ok := cv.(types.SerialMessage)
	if !ok {
		return hash.Hash{}, fmt.Errorf("expected SerialMessage type for commit value, found %T", cv)
	}

	data := []byte(sm)
	if serial.GetFileID(data) != serial.CommitFileID {
		return hash.Hash{}, errors.New("GetCommitRootHash: provided value is not a commit.")
	}
	var cmsg serial.Commit
	err := serial.InitCommitRoot(&cmsg, data, serial.MessagePrefixSz)
	if err != nil {
		return hash.Hash{}, err
	}
	var roothash hash.Hash
	copy(roothash[:], cmsg.RootBytes())
	return roothash, nil
}

func parentsToQueue(ctx context.Context, commits []*Commit, q *CommitByHeightHeap, vr types.ValueReader) error {
	seen := make(map[hash.Hash]struct{})
	for _, c := range commits {
		if _, ok := seen[c.Addr()]; ok {
			continue
		}
		seen[c.Addr()] = struct{}{}

		parents, err := GetCommitParents(ctx, vr, c.NomsValue())
		if err != nil {
			return err
		}
		for _, r := range parents {
			heap.Push(q, r)
		}
	}

	return nil
}

func findCommonCommit(a, b []*Commit) (*Commit, bool) {
	toAddrMap := func(s []*Commit) map[hash.Hash]*Commit {
		out := map[hash.Hash]*Commit{}
		for _, r := range s {
			out[r.Addr()] = r
		}
		return out
	}

	as, bs := toAddrMap(a), toAddrMap(b)
	var common []*Commit
	for s, c := range as {
		if _, present := bs[s]; present {
			common = append(common, c)
		}
	}
	if len(common) == 0 {
		return nil, false
	}
	// Sort by address to ensure deterministic selection when multiple
	// common ancestors exist at the same height (e.g., criss-cross merges).
	sort.Slice(common, func(i, j int) bool {
		return common[i].Addr().Less(common[j].Addr())
	})
	return common[0], true
}

func firstError(l, r error) error {
	if l != nil {
		return l
	}
	return r
}

func IsCommit(v types.Value) (bool, error) {
	if sm, ok := v.(types.SerialMessage); ok {
		data := []byte(sm)
		return serial.GetFileID(data) == serial.CommitFileID, nil
	} else {
		return false, nil
	}
}

type CommitByHeightHeap []*Commit

func (r CommitByHeightHeap) Less(i, j int) bool {
	if r[i].Height() == r[j].Height() {
		return r[i].Addr().Less(r[j].Addr())
	}
	return r[i].Height() > r[j].Height()
}

func (r CommitByHeightHeap) Swap(i, j int) {
	r[i], r[j] = r[j], r[i]
}

func (r CommitByHeightHeap) Len() int {
	return len(r)
}

func (r *CommitByHeightHeap) Push(x interface{}) {
	*r = append(*r, x.(*Commit))
}

func (r *CommitByHeightHeap) Pop() interface{} {
	old := *r
	ret := old[len(old)-1]
	*r = old[:len(old)-1]
	return ret
}

func (r CommitByHeightHeap) Empty() bool {
	return len(r) == 0
}

func (r CommitByHeightHeap) MaxHeight() uint64 {
	return r[0].Height()
}

func (r *CommitByHeightHeap) PopCommitsOfHeight(h uint64) []*Commit {
	var ret []*Commit
	for !r.Empty() && r.MaxHeight() == h {
		ret = append(ret, heap.Pop(r).(*Commit))
	}
	return ret
}