package githttp

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os/exec"
	"strings"
	"time"

	"github.com/pingcap/agent-git-service/internal/db"
	"github.com/pingcap/agent-git-service/internal/rest/transform"
	"github.com/pingcap/agent-git-service/internal/service"
)

const zeroGitSHA = "0000000000000000000000000000000000000000"

const (
	maxPushCommitDetailBytes      = 256 << 10
	maxPushCommitDetailFiles      = 1024
	maxPushCommitPathRecordBytes  = 4096
	maxPushCommitOperationBytes   = 16 << 20
	maxPushCommitOperationOIDs    = 1024 * 20
	maxPushWebhookEventBytes      = 8 << 20
	maxPushWebhookOperationBytes  = 16 << 20
	maxPushWebhookOperationEvents = 2048
)

type pushRefChange struct {
	Ref        string
	Before     string
	After      string
	Created    bool
	Deleted    bool
	Forced     bool
	CommitSHAs []string
}

func (h *Handler) handlePostPushWebhooks(ctx context.Context, repoFullName, repoPath string, updates []service.RepositoryReceivePackRefUpdate, occurredAt time.Time, operationID string) error {
	repo, err := h.Svc.GetRepo(ctx, repoFullName)
	if err != nil {
		return err
	}
	beforeRefs := make(map[string]string, len(updates))
	afterRefs := make(map[string]string, len(updates))
	commitsByRef := make(map[string][]string, len(updates))
	for _, update := range updates {
		if update.Before != "" {
			beforeRefs[update.Ref] = update.Before
		}
		if update.After != "" {
			afterRefs[update.Ref] = update.After
		}
		commitsByRef[update.Ref] = append([]string(nil), update.CommitSHAs...)
	}
	changes, err := diffPushedRefs(ctx, repoPath, beforeRefs, afterRefs)
	if err != nil {
		return err
	}
	for index := range changes {
		changes[index].CommitSHAs = commitsByRef[changes[index].Ref]
	}
	if len(changes) == 0 {
		return nil
	}
	commitDetails, err := pushCommitDetailsBatch(ctx, repoPath, repo.FullName, h.Svc.HTMLBaseURL(), changes)
	if err != nil {
		return err
	}

	if err := h.Svc.UpdateRepositoryPushedAt(ctx, repo.ID, occurredAt); err != nil {
		return err
	}
	repo, err = h.Svc.GetRepo(ctx, repoFullName)
	if err != nil {
		return err
	}
	operationBytes := 0
	operationEvents := 0
	webhookEvents := make([]service.RepositoryEvent, 0, len(changes))
	dispatch := func(repositoryID uint, event, action string, payload any, eventKey string) error {
		encoded, err := json.Marshal(payload)
		if err != nil {
			return err
		}
		operationEvents++
		operationBytes += len(encoded)
		if len(encoded) > maxPushWebhookEventBytes || operationBytes > maxPushWebhookOperationBytes || operationEvents > maxPushWebhookOperationEvents {
			return fmt.Errorf("post-push webhook materialization exceeds limit")
		}
		webhookEvents = append(webhookEvents, service.RepositoryEvent{
			RepositoryID: repositoryID, Event: event, Action: action, EventKey: eventKey, PayloadJSON: encoded,
		})
		return nil
	}

	for _, change := range changes {
		payload := h.buildPushWebhookPayload(ctx, repo, change, commitDetails)
		if err := dispatch(repo.ID, "push", "", payload, "push:"+change.Ref); err != nil {
			return err
		}
		if !strings.HasPrefix(change.Ref, "refs/heads/") || change.Deleted {
			continue
		}
		branch := strings.TrimPrefix(change.Ref, "refs/heads/")
		prs, err := h.Svc.ListOpenPRsByHead(ctx, repo.ID, branch)
		if err != nil {
			return err
		}
		for _, pr := range prs {
			updatedPR, err := h.Svc.SyncPRHeadAfterPush(ctx, pr.ID, repo.FullName)
			if err != nil {
				return err
			}
			if err := dispatch(updatedPR.RepositoryID, "pull_request", "synchronize", map[string]any{
				"action":       "synchronize",
				"number":       updatedPR.Number,
				"pull_request": transform.PR(updatedPR, nil, transform.AuthorAssociationChecks{}, nil),
				"repository":   transform.Repo(updatedPR.Repository),
				"sender":       pushSenderPayload(ctx),
			}, fmt.Sprintf("pull_request:%d:synchronize", updatedPR.ID)); err != nil {
				return err
			}
		}
	}
	return h.Svc.DispatchWebhookOperationIdempotent(ctx, operationID, webhookEvents)
}

func (h *Handler) buildPushWebhookPayload(ctx context.Context, repo db.Repository, change pushRefChange, details map[string]map[string]any) map[string]any {
	commits, headCommit := pushCommits(change, details)
	var compareURL string
	if change.Before != zeroGitSHA && change.After != zeroGitSHA {
		compareURL = fmt.Sprintf("%s/%s/compare/%s...%s", h.Svc.HTMLBaseURL(), repo.FullName, change.Before, change.After)
	}
	return map[string]any{
		"ref":         change.Ref,
		"before":      change.Before,
		"after":       change.After,
		"created":     change.Created,
		"deleted":     change.Deleted,
		"forced":      change.Forced,
		"base_ref":    nil,
		"compare":     compareURL,
		"commits":     commits,
		"head_commit": headCommit,
		"repository":  transform.Repo(repo),
		"pusher":      pushPusherPayload(ctx),
		"sender":      pushSenderPayload(ctx),
	}
}

func pushSenderPayload(ctx context.Context) any {
	user, ok := service.UserFromContext(ctx)
	if !ok {
		return nil
	}
	return transform.User(user)
}

func pushPusherPayload(ctx context.Context) map[string]any {
	user, ok := service.UserFromContext(ctx)
	if !ok {
		return map[string]any{"name": "", "email": ""}
	}
	return map[string]any{
		"name":  user.Login,
		"email": user.Email,
	}
}

func diffPushedRefs(ctx context.Context, repoPath string, before, after map[string]string) ([]pushRefChange, error) {
	seen := make(map[string]struct{}, len(before)+len(after))
	var changes []pushRefChange
	for ref, beforeSHA := range before {
		seen[ref] = struct{}{}
		afterSHA := after[ref]
		change, ok, err := buildPushRefChange(ctx, repoPath, ref, beforeSHA, afterSHA)
		if err != nil {
			return nil, err
		}
		if ok {
			changes = append(changes, change)
		}
	}
	for ref, afterSHA := range after {
		if _, ok := seen[ref]; ok {
			continue
		}
		change, ok, err := buildPushRefChange(ctx, repoPath, ref, "", afterSHA)
		if err != nil {
			return nil, err
		}
		if ok {
			changes = append(changes, change)
		}
	}
	return changes, nil
}

func buildPushRefChange(ctx context.Context, repoPath, ref, beforeSHA, afterSHA string) (pushRefChange, bool, error) {
	if !strings.HasPrefix(ref, "refs/heads/") && !strings.HasPrefix(ref, "refs/tags/") {
		return pushRefChange{}, false, nil
	}
	if beforeSHA == afterSHA {
		return pushRefChange{}, false, nil
	}
	change := pushRefChange{
		Ref:    ref,
		Before: beforeSHA,
		After:  afterSHA,
	}
	if change.Before == "" {
		change.Before = zeroGitSHA
		change.Created = true
	}
	if change.After == "" {
		change.After = zeroGitSHA
		change.Deleted = true
	}
	if !change.Created && !change.Deleted {
		ancestor, err := gitIsAncestor(ctx, repoPath, change.Before, change.After)
		if err != nil {
			return pushRefChange{}, false, fmt.Errorf("classify pushed ref %s: %w", ref, err)
		}
		change.Forced = !ancestor
	}
	return change, true, nil
}

func gitIsAncestor(ctx context.Context, repoPath, beforeSHA, afterSHA string) (bool, error) {
	cmd := exec.CommandContext(ctx, "git", "-C", repoPath, "merge-base", "--is-ancestor", beforeSHA, afterSHA)
	err := cmd.Run()
	if err == nil {
		return true, nil
	}
	var exitErr *exec.ExitError
	if errors.As(err, &exitErr) && exitErr.ExitCode() == 1 {
		return false, nil
	}
	return false, err
}

func pushCommits(change pushRefChange, details map[string]map[string]any) ([]map[string]any, any) {
	if change.Deleted || change.After == zeroGitSHA {
		return []map[string]any{}, nil
	}

	var commits []map[string]any
	for _, sha := range change.CommitSHAs {
		sha = strings.TrimSpace(sha)
		if sha == "" {
			continue
		}
		commit, ok := details[sha]
		if !ok {
			return []map[string]any{}, nil
		}
		commits = append(commits, commit)
	}
	if len(commits) == 0 {
		return []map[string]any{}, nil
	}
	return commits, commits[len(commits)-1]
}

func pushCommitDetailsBatch(ctx context.Context, repoPath, repoFullName, htmlBaseURL string, changes []pushRefChange) (map[string]map[string]any, error) {
	seen := make(map[string]bool)
	ordered := make([]string, 0)
	for _, change := range changes {
		for _, sha := range change.CommitSHAs {
			sha = strings.TrimSpace(sha)
			if sha != "" && !seen[sha] {
				seen[sha] = true
				ordered = append(ordered, sha)
			}
		}
	}
	if len(ordered) == 0 {
		return map[string]map[string]any{}, nil
	}
	if len(ordered) > maxPushCommitOperationOIDs {
		return nil, fmt.Errorf("push commit hydration exceeds commit limit")
	}
	cmd := exec.CommandContext(ctx, "git", "-C", repoPath, "log", "--no-walk=unsorted", "--stdin", "--format=%x1e%H%x00%T%x00%an%x00%ae%x00%cn%x00%ce%x00%aI%x00%s", "--name-status", "--find-renames", "--find-copies", "--no-ext-diff", "--no-color")
	cmd.Stdin = strings.NewReader(strings.Join(ordered, "\n") + "\n")
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return nil, err
	}
	var stderr bytes.Buffer
	cmd.Stderr = &stderr
	if err := cmd.Start(); err != nil {
		return nil, err
	}
	limited := &countedReader{reader: io.LimitReader(stdout, maxPushCommitOperationBytes+1)}
	scanner := bufio.NewScanner(limited)
	scanner.Split(splitPushCommitRecords)
	scanner.Buffer(make([]byte, 64<<10), maxPushCommitDetailBytes+1)
	result := make(map[string]map[string]any, len(ordered))
	var parseErr error
	for scanner.Scan() {
		record := bytes.Trim(scanner.Bytes(), "\n")
		if len(record) == 0 {
			continue
		}
		if len(record) > maxPushCommitDetailBytes {
			parseErr = fmt.Errorf("push commit detail exceeds byte limit")
			break
		}
		lines := strings.Split(string(record), "\n")
		if len(lines)-1 > maxPushCommitDetailFiles {
			parseErr = fmt.Errorf("push commit detail exceeds file limit")
			break
		}
		for _, line := range lines[1:] {
			if len(line) > maxPushCommitPathRecordBytes {
				parseErr = fmt.Errorf("push commit path record exceeds byte limit")
				break
			}
		}
		if parseErr != nil {
			break
		}
		commit, sha, ok := parsePushCommitRecord(lines, repoFullName, htmlBaseURL)
		if !ok {
			parseErr = fmt.Errorf("parse push commit details")
			break
		}
		result[sha] = commit
	}
	if err := scanner.Err(); parseErr == nil {
		parseErr = err
	}
	if limited.read > maxPushCommitOperationBytes && parseErr == nil {
		parseErr = fmt.Errorf("push commit hydration exceeds operation byte limit")
	}
	_ = stdout.Close()
	if parseErr != nil {
		_ = cmd.Process.Kill()
	}
	waitErr := cmd.Wait()
	if parseErr != nil {
		return nil, parseErr
	}
	if waitErr != nil {
		return nil, fmt.Errorf("hydrate push commits: %w: %s", waitErr, strings.TrimSpace(stderr.String()))
	}
	for _, sha := range ordered {
		if _, ok := result[sha]; !ok {
			return nil, fmt.Errorf("push commit %s is missing", sha)
		}
	}
	return result, nil
}

type countedReader struct {
	reader io.Reader
	read   int64
}

func (reader *countedReader) Read(buffer []byte) (int, error) {
	count, err := reader.reader.Read(buffer)
	reader.read += int64(count)
	return count, err
}

func splitPushCommitRecords(data []byte, atEOF bool) (int, []byte, error) {
	if len(data) != 0 && data[0] == 0x1e {
		return 1, nil, nil
	}
	if index := bytes.IndexByte(data, 0x1e); index >= 0 {
		return index, data[:index], nil
	}
	if atEOF && len(data) != 0 {
		return len(data), data, nil
	}
	return 0, nil, nil
}

func parsePushCommitRecord(lines []string, repoFullName, htmlBaseURL string) (map[string]any, string, bool) {
	if len(lines) == 0 {
		return nil, "", false
	}
	parts := strings.Split(lines[0], "\x00")
	if len(parts) != 8 {
		return nil, "", false
	}

	added := make([]string, 0)
	removed := make([]string, 0)
	modified := make([]string, 0)
	for _, line := range lines[1:] {
		if line == "" {
			continue
		}
		statusParts := strings.Split(line, "\t")
		if len(statusParts) < 2 || statusParts[0] == "" {
			continue
		}
		switch statusParts[0][0] {
		case 'A', 'C':
			added = append(added, statusParts[len(statusParts)-1])
		case 'D':
			removed = append(removed, statusParts[len(statusParts)-1])
		case 'R':
			if len(statusParts) >= 3 {
				removed = append(removed, statusParts[1])
				added = append(added, statusParts[2])
			}
		default:
			modified = append(modified, statusParts[len(statusParts)-1])
		}
	}

	return map[string]any{
		"id":        parts[0],
		"tree_id":   parts[1],
		"message":   parts[7],
		"timestamp": parts[6],
		"url":       fmt.Sprintf("%s/%s/commit/%s", htmlBaseURL, repoFullName, parts[0]),
		"distinct":  true,
		"author": map[string]any{
			"name":  parts[2],
			"email": parts[3],
		},
		"committer": map[string]any{
			"name":  parts[4],
			"email": parts[5],
		},
		"added":    added,
		"removed":  removed,
		"modified": modified,
	}, parts[0], true
}