package browse

import (
	"context"
	"errors"
	"fmt"

	"github.com/dolthub/dolt/go/libraries/doltcore/diff"
	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
)

// TableDiff summarizes how one table changed in a commit relative to its first
// parent. Row-level diffs are out of scope for v1; only the per-table shape and
// exact row counts are reported.
type TableDiff struct {
	Name          string
	Added         bool
	Dropped       bool
	SchemaChanged bool
	// RowsAdded, RowsRemoved and RowsModified are exact counts derived from the
	// prolly diff. They are 0 when counts could not be computed (e.g. the
	// primary key set changed, in which case SchemaChanged is true).
	RowsAdded    int64
	RowsRemoved  int64
	RowsModified int64
}

// CommitDiff is the per-table summary of a single commit versus its first
// parent (or the empty root, for the initial commit).
type CommitDiff struct {
	Hash   string
	Tables []TableDiff
}

// CommitSummary computes the per-table diff summary of the commit identified by
// hashStr (a commit hash, or any ref resolveCommit accepts) against its first
// parent. For the initial commit the comparison is against an empty root, so
// every table shows as added.
func (db *DB) CommitSummary(ctx context.Context, hashStr string) (*CommitDiff, error) {
	commit, err := db.resolveCommit(ctx, hashStr)
	if err != nil {
		return nil, err
	}

	toRoot, err := commit.GetRootValue(ctx)
	if err != nil {
		return nil, fmt.Errorf("browse: root value of %q: %w", hashStr, err)
	}

	fromRoot, err := db.firstParentRoot(ctx, commit)
	if err != nil {
		return nil, err
	}

	deltas, err := diff.GetTableDeltas(ctx, fromRoot, toRoot)
	if err != nil {
		return nil, fmt.Errorf("browse: table deltas for %q: %w", hashStr, err)
	}

	h, err := commit.HashOf()
	if err != nil {
		return nil, fmt.Errorf("browse: hash of %q: %w", hashStr, err)
	}

	out := &CommitDiff{Hash: h.String(), Tables: make([]TableDiff, 0, len(deltas))}
	for _, delta := range deltas {
		td, err := tableDiff(ctx, delta)
		if err != nil {
			return nil, err
		}
		out.Tables = append(out.Tables, td)
	}

	return out, nil
}

// firstParentRoot returns the root value of the commit's first parent, or an
// empty root when the commit has no parents (the initial commit).
func (db *DB) firstParentRoot(ctx context.Context, commit *doltdb.Commit) (doltdb.RootValue, error) {
	parents, err := commit.ParentHashes(ctx)
	if err != nil {
		return nil, fmt.Errorf("browse: parent hashes: %w", err)
	}

	if len(parents) == 0 {
		root, err := doltdb.EmptyRootValue(ctx, db.ddb.ValueReadWriter(), db.ddb.NodeStore())
		if err != nil {
			return nil, fmt.Errorf("browse: empty root: %w", err)
		}
		return root, nil
	}

	oc, err := db.ddb.ResolveHash(ctx, parents[0])
	if err != nil {
		return nil, fmt.Errorf("browse: resolve parent %s: %w", parents[0].String(), err)
	}
	parent, ok := oc.ToCommit()
	if !ok {
		return nil, fmt.Errorf("browse: parent commit %s is not resolvable (ghost)", parents[0].String())
	}
	root, err := parent.GetRootValue(ctx)
	if err != nil {
		return nil, fmt.Errorf("browse: parent root %s: %w", parents[0].String(), err)
	}
	return root, nil
}

// tableDiff summarizes a single TableDelta.
func tableDiff(ctx context.Context, delta diff.TableDelta) (TableDiff, error) {
	name := delta.ToName.Name
	if delta.IsDrop() {
		name = delta.FromName.Name
	}

	td := TableDiff{
		Name:    name,
		Added:   delta.IsAdd(),
		Dropped: delta.IsDrop(),
	}

	schemaChanged, err := delta.HasSchemaChanged(ctx)
	if err != nil {
		return TableDiff{}, fmt.Errorf("browse: schema-changed check for %q: %w", name, err)
	}
	td.SchemaChanged = schemaChanged

	added, removed, modified, err := tableRowStat(ctx, delta)
	if err != nil {
		if errors.Is(err, diff.ErrPrimaryKeySetChanged) {
			// Row counts are undefined when the PK set changed; the schema
			// change is already reflected. Leave counts at 0.
			return td, nil
		}
		return TableDiff{}, fmt.Errorf("browse: row stat for %q: %w", name, err)
	}
	td.RowsAdded = added
	td.RowsRemoved = removed
	td.RowsModified = modified

	return td, nil
}

// tableRowStat drains diff.StatForTableDelta into exact added/removed/modified
// row counts. StatForTableDelta streams one progress message per changed row,
// so we accumulate the per-message Adds/Removes/Changes.
func tableRowStat(ctx context.Context, delta diff.TableDelta) (added, removed, modified int64, err error) {
	ch := make(chan diff.DiffStatProgress, 128)
	done := make(chan struct{})
	var statErr error
	go func() {
		defer close(done)
		statErr = diff.StatForTableDelta(ctx, ch, delta)
		close(ch)
	}()

	var a, r, m uint64
	for p := range ch {
		a += p.Adds
		r += p.Removes
		m += p.Changes
	}
	<-done

	if statErr != nil {
		return 0, 0, 0, statErr
	}
	return int64(a), int64(r), int64(m), nil
}