package spindle

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"tangled.org/core/api/org_tangled"
	"tangled.org/core/rbac"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/knotclient"
	"tangled.org/core/spindle/models"
	"tangled.org/core/tapc"
	"tangled.org/core/tid"
	"tangled.org/core/workflow"
)

const pullEventBatch = 16

// a pull request is a ticket with a target branch plus the patches that name it
type knotTicket struct {
	org_tangled.TrackTicket
	Editor syntax.DID `json:"x-tngl-editor"`
}

type knotPatch struct {
	org_tangled.ReviewPatch
	Editor syntax.DID `json:"x-tngl-editor"`
}

const (
	pullOpen   = "open"
	pullClosed = "closed"
	pullMerged = "completed"
)

var pullStates = []string{pullOpen, pullClosed, pullMerged}

var pullActions = map[models.WebhookEvent]string{
	models.WebhookEventPullRequestCreated:     workflow.PullRequestActionOpened,
	models.WebhookEventPullRequestResubmitted: workflow.PullRequestActionSynchronize,
	models.WebhookEventPullRequestReopened:    workflow.PullRequestActionReopened,
	models.WebhookEventPullRequestClosed:      workflow.PullRequestActionClosed,
	models.WebhookEventPullRequestMerged:      workflow.PullRequestActionMerged,
}

func (s *Spindle) onTicket(ctx context.Context, rec *tapc.RecordEventData) error {
	return s.onPullRecord(ctx, rec, s.applyTicket)
}

func (s *Spindle) onPatch(ctx context.Context, rec *tapc.RecordEventData) error {
	return s.onPullRecord(ctx, rec, s.applyPatch)
}

// a live event won't come again, so failing before anything got stored puts the record in
// the retry set, and anything after that is the stored event's to retry
func (s *Spindle) onPullRecord(ctx context.Context, rec *tapc.RecordEventData, apply func(context.Context, *db.Repo, *tapc.RecordEventData) error) error {
	if markOnly(ctx) {
		return s.markRecord(rec)
	}
	retry := db.RecordRetry{RepoDid: rec.Did, Collection: rec.Collection.String(), Rkey: rec.Rkey.String()}
	err := s.applyPullRecord(ctx, rec, retry, apply)
	if err != nil && rec.Live && ctx.Err() == nil {
		s.failRecordMark(retry)
	}
	return err
}

func (s *Spindle) applyPullRecord(ctx context.Context, rec *tapc.RecordEventData, retry db.RecordRetry, apply func(context.Context, *db.Repo, *tapc.RecordEventData) error) error {
	repo, err := s.servedRepo(rec.Did)
	if err != nil || repo == nil {
		return err
	}
	// a mark set during apply bumps the epoch, which makes the clear skip it
	epoch, err := s.db.RecordRetryEpoch(retry)
	if err != nil {
		return err
	}
	if err := apply(ctx, repo, rec); err != nil {
		return err
	}
	if err := s.db.ClearRecordRetryAt(retry, epoch); err != nil {
		s.l.Error("clearing a record retry", "repo", rec.Did, "rkey", retry.Rkey, "err", err)
	}
	s.retries.jobs.kick(repo.RepoDid)
	return nil
}

// events only say that something changed, so every handler rereads the knot
// and derives what happened from the state it stored last time
func (s *Spindle) applyTicket(ctx context.Context, repo *db.Repo, rec *tapc.RecordEventData) error {
	rkey := rec.Rkey.String()
	if rec.Action == tapc.RecordDeleteAction {
		return s.db.ForgetPullTicket(repo.RepoDid, rkey)
	}
	k, err := s.knots.KnotOf(ctx, repo.RepoDid)
	if err != nil {
		return fmt.Errorf("finding the knot of %s: %w", repo.RepoDid, err)
	}
	_, _, err = s.syncPullTicket(ctx, k, repo, rkey)
	return err
}

func (s *Spindle) applyPatch(ctx context.Context, repo *db.Repo, rec *tapc.RecordEventData) error {
	rkey := rec.Rkey.String()
	if rec.Action == tapc.RecordDeleteAction {
		return s.db.ForgetPullPatch(repo.RepoDid, rkey)
	}
	k, err := s.knots.KnotOf(ctx, repo.RepoDid)
	if err != nil {
		return fmt.Errorf("finding the knot of %s: %w", repo.RepoDid, err)
	}
	var patch knotPatch
	err = s.knots.GetRecord(ctx, k, repo.RepoDid, org_tangled.ReviewPatchNSID, rkey, &patch)
	if errors.Is(err, knotclient.ErrNotFound) {
		return s.db.ForgetPullPatch(repo.RepoDid, rkey)
	}
	if err != nil {
		return fmt.Errorf("reading patch %s/%s: %w", repo.RepoDid, rkey, err)
	}
	if patch.Head == nil {
		s.l.Warn("skipping patch without a head", "repo", repo.RepoDid, "rkey", rkey)
		return nil
	}

	ticket, row, err := s.syncPullTicket(ctx, k, repo, patch.Ticket)
	if err != nil || ticket == nil {
		return err
	}
	// the event is created or resubmitted, which only the stored patches can tell
	ev, err := newPullEvent(repo, patch.Ticket, "", ticket, row.Author, patch.Editor)
	if err != nil {
		return err
	}
	stored := db.PullPatch{Rkey: rkey, Ticket: patch.Ticket, Head: patch.Head.Oid, CreatedAt: patch.CreatedAt}
	if patch.Source != nil {
		stored.SourceRepo, stored.SourceBranch = patch.Source.Repo, patch.Source.Branch
	}
	return s.whileServed(repo, func() error { return s.db.AddPullPatch(repo.RepoDid, stored, ev) })
}

// holding the repo lock means a wipe can't land between the check and the write
func (s *Spindle) whileServed(repo *db.Repo, write func() error) error {
	defer s.lockRepo(repo.RepoDid).Unlock()
	if served, err := s.servedRepo(repo.RepoDid); err != nil || served == nil {
		return err
	}
	return write()
}

// a nil ticket means there is no pull request to speak of, an issue or a record the knot lacks
func (s *Spindle) syncPullTicket(ctx context.Context, k knotclient.Knot, repo *db.Repo, rkey string) (*knotTicket, *db.PullTicket, error) {
	var ticket knotTicket
	err := s.knots.GetRecord(ctx, k, repo.RepoDid, org_tangled.TrackTicketNSID, rkey, &ticket)
	if errors.Is(err, knotclient.ErrNotFound) {
		return nil, nil, s.db.ForgetPullTicket(repo.RepoDid, rkey)
	}
	if err != nil {
		return nil, nil, fmt.Errorf("reading ticket %s/%s: %w", repo.RepoDid, rkey, err)
	}
	if ticket.TargetBranch == nil {
		return nil, nil, nil
	}
	state := pullOpen
	if ticket.State != nil {
		state = *ticket.State
	}
	if !slices.Contains(pullStates, state) {
		s.l.Warn("skipping ticket with an unknown state", "repo", repo.RepoDid, "rkey", rkey, "state", state)
		return nil, nil, nil
	}
	ticket.State = &state

	row, err := s.db.GetPullTicket(repo.RepoDid, rkey)
	if errors.Is(err, sql.ErrNoRows) {
		versions, err := s.knots.ListRecordVersions(ctx, k, repo.RepoDid, org_tangled.TrackTicketNSID, rkey)
		if errors.Is(err, knotclient.ErrNotFound) {
			return nil, nil, s.db.ForgetPullTicket(repo.RepoDid, rkey)
		}
		if err != nil {
			return nil, nil, fmt.Errorf("finding who opened %s/%s: %w", repo.RepoDid, rkey, err)
		}
		// nothing to compare a first sighting with, so it fires nothing until its patch lands
		row = &db.PullTicket{Author: versions[0].Editor, State: state, Version: versions[len(versions)-1].Cid}
		err = s.whileServed(repo, func() error { return s.db.AddPullTicket(repo.RepoDid, rkey, *row) })
		return &ticket, row, err
	}
	if err != nil {
		return nil, nil, err
	}
	if err := s.walkPullTicket(ctx, k, repo, rkey, &ticket, row); err != nil {
		return nil, nil, err
	}
	return &ticket, row, nil
}

const maxPullVersionWalk = 64

// events don't say what changed, and a close then reopen only shows up in the version walk
func (s *Spindle) walkPullTicket(ctx context.Context, k knotclient.Knot, repo *db.Repo, rkey string, current *knotTicket, row *db.PullTicket) error {
	versions, err := s.knots.ListRecordVersions(ctx, k, repo.RepoDid, org_tangled.TrackTicketNSID, rkey)
	if errors.Is(err, knotclient.ErrNotFound) {
		return s.applyPullState(repo, rkey, current, row, "", current.Editor)
	}
	if err != nil {
		// using the newest state here would skip the middle transitions for good
		return fmt.Errorf("listing the versions of ticket %s/%s: %w", repo.RepoDid, rkey, err)
	}
	newest := len(versions) - 1
	start := newest
	if i := slices.IndexFunc(versions, func(v knotclient.RecordVersion) bool { return v.Cid == row.Version }); i >= 0 {
		start = i + 1
	}
	start = max(start, len(versions)-maxPullVersionWalk)
	for _, v := range versions[start:] {
		var at knotTicket
		err := s.knots.GetRecordAt(ctx, k, repo.RepoDid, org_tangled.TrackTicketNSID, rkey, v.Cid, &at)
		if errors.Is(err, knotclient.ErrNotFound) {
			continue
		}
		if err != nil {
			return fmt.Errorf("reading ticket %s/%s at %s: %w", repo.RepoDid, rkey, v.Cid, err)
		}
		state := pullOpen
		if at.State != nil {
			state = *at.State
		}
		if !slices.Contains(pullStates, state) {
			continue
		}
		at.State = &state
		if at.TargetBranch == nil {
			at.TargetBranch = current.TargetBranch
		}
		if err := s.applyPullState(repo, rkey, &at, row, v.Cid, v.Editor); err != nil {
			return err
		}
	}
	return nil
}

func (s *Spindle) applyPullState(repo *db.Repo, rkey string, at *knotTicket, row *db.PullTicket, version string, sender syntax.DID) error {
	state := *at.State
	event, moved := pullTransition(row.State, state)
	if !moved {
		if version == "" || version == row.Version {
			return nil
		}
		advance := func() error { return s.db.AdvancePullTicket(repo.RepoDid, rkey, row.Version, version, state, nil) }
		if err := s.whileServed(repo, advance); err != nil {
			return err
		}
		row.Version = version
		return nil
	}
	ev, err := newPullEvent(repo, rkey, event, at, row.Author, sender)
	if err != nil {
		return err
	}
	err = s.whileServed(repo, func() error {
		if version == "" {
			return s.db.SetPullTicketState(repo.RepoDid, rkey, row.State, state, ev)
		}
		return s.db.AdvancePullTicket(repo.RepoDid, rkey, row.Version, version, state, &ev)
	})
	if err != nil {
		return err
	}
	row.State, row.Version = state, version
	return nil
}

func pullTransition(from, to string) (models.WebhookEvent, bool) {
	switch {
	case from == to:
		return "", false
	case to == pullMerged:
		return models.WebhookEventPullRequestMerged, true
	case to == pullClosed:
		return models.WebhookEventPullRequestClosed, true
	default:
		return models.WebhookEventPullRequestReopened, true
	}
}

// sender is whoever wrote the version that caused the event, so not the author on a merge
func newPullEvent(repo *db.Repo, rkey string, event models.WebhookEvent, ticket *knotTicket, author, sender syntax.DID) (db.PullEvent, error) {
	pr := models.WebhookPullRequest{
		State:        "open",
		TargetBranch: *ticket.TargetBranch,
		Owner:        models.WebhookUser{Did: author.String()},
		CreatedAt:    ticket.CreatedAt,
	}
	switch *ticket.State {
	case pullClosed:
		pr.State = "closed"
	case pullMerged:
		pr.State = "merged"
	}
	if ticket.Title != nil && ticket.Title.MarkupMarkdown_Inline != nil {
		pr.Title = ticket.Title.MarkupMarkdown_Inline.Text
	}
	if ticket.Body != nil && ticket.Body.MarkupMarkdown != nil {
		pr.Body = ticket.Body.MarkupMarkdown.Text
	}
	pull, err := json.Marshal(pr)
	if err != nil {
		return db.PullEvent{}, err
	}
	return db.PullEvent{
		RepoDid:  repo.RepoDid,
		Ticket:   rkey,
		Event:    event,
		Author:   author,
		Sender:   sender,
		Pull:     pull,
		Pipeline: models.PipelineId(tid.TID()),
	}, nil
}

func (s *Spindle) runDuePullEvents(ctx context.Context, did syntax.DID) {
	events, err := s.db.DuePullEvents(did, s.retries.clock(), pullEventBatch)
	if err != nil {
		s.l.Error("reading the pull events", "repo", did, "err", err)
		return
	}
	if len(events) == 0 {
		return
	}
	repo, err := s.servedRepo(did)
	if err != nil {
		s.l.Error("looking up a repo with pull events", "repo", did, "err", err)
		return
	}
	// an unserved repo's events would stay due forever unless dropped
	if repo == nil {
		if err := s.db.DeletePullEventsByRepo(did); err != nil {
			s.l.Error("dropping the pull events of a repo no longer served", "repo", did, "err", err)
		}
		return
	}
	if s.runPullEvents(ctx, repo, events) && len(events) == pullEventBatch {
		// kick to the back of the queue, or one long backlog can hog a worker
		s.retries.jobs.kick(did)
	}
}

// one notify slot per event, otherwise a backlog pins the pool
func (s *Spindle) runPullEvents(ctx context.Context, repo *db.Repo, events []db.PullEvent) bool {
	for _, ev := range events {
		release, ok := s.acquireNotifySlot(repo.Knot)
		if !ok {
			return false
		}
		// only a shutdown skips the failure mark, a budget timeout still counts as one
		work, cancel := context.WithTimeout(ctx, notifyWorkBudget)
		err := s.runPullEvent(work, repo, ev)
		cancel()
		release()
		if err != nil && ctx.Err() == nil {
			s.l.Warn("a pull request event failed", "repo", repo.RepoDid, "pull", ev.Ticket, "event", ev.Event, "err", err)
			s.failPullEvent(ev)
		}
	}
	return true
}

func (s *Spindle) bannedInPull(ev db.PullEvent) (bool, error) {
	for _, did := range []syntax.DID{ev.Author, ev.Sender} {
		if ban, err := s.db.GetBan(did); err != nil || ban != nil {
			return ban != nil, err
		}
	}
	return false, nil
}

func (s *Spindle) runPullEvent(ctx context.Context, repo *db.Repo, ev db.PullEvent) error {
	if banned, err := s.bannedInPull(ev); err != nil {
		return fmt.Errorf("checking bans: %w", err)
	} else if banned {
		s.l.Warn("dropping a pull request event of a banned user", "repo", repo.RepoDid, "pull", ev.Ticket, "author", ev.Author, "sender", ev.Sender)
		return s.db.DeletePullEvent(ev.ID)
	}
	var pr models.WebhookPullRequest
	if err := json.Unmarshal(ev.Pull, &pr); err != nil {
		return err
	}
	pr.RoundNumber = ev.Rounds
	if ev.Head.Sha != "" {
		pr.Source = &models.WebhookPullRequestSource{Sha: ev.Head.Sha, Branch: ev.Head.SourceBranch, Repo: ev.Head.SourceRepo}
	}
	// marked before the handoff, so a crash in between loses the webhook instead of sending it twice
	if !ev.WebhookFired {
		if err := s.db.MarkPullEventWebhookFired(ev.ID); err != nil {
			return err
		}
		s.wh.FirePullRequest(ctx, ev.Event, repo, pr, ev.Sender)
	}
	if err := s.triggerPullPipeline(ctx, repo, ev, pr.TargetBranch); err != nil {
		return err
	}
	return s.db.DeletePullEvent(ev.ID)
}

// only an error is worth retrying, everything that would end the same way again returns nil
func (s *Spindle) triggerPullPipeline(ctx context.Context, repo *db.Repo, ev db.PullEvent, targetBranch string) error {
	l := s.l.With("repo", repo.RepoDid, "pull", ev.Ticket, "event", ev.Event)
	// the pipeline id was minted with the event, so finding it means a retry after the enqueue
	if exists, err := s.db.PipelineExists(ev.Pipeline); err != nil || exists {
		return err
	}
	if ev.Head.Sha == "" {
		return nil
	}
	if ban, err := s.db.IsBanned(repo.RepoDid, repo.Owner); err != nil {
		return fmt.Errorf("checking bans: %w", err)
	} else if ban != nil {
		l.Warn("dropping pull trigger for banned repo")
		return nil
	}
	allowed, err := s.mayTriggerPull(repo.RepoDid, ev.Author, ev.Sender)
	if err != nil {
		return fmt.Errorf("authorizing pull-triggered pipeline: %w", err)
	}

	triggerRepo, err := s.buildTriggerRepo(ctx, repo)
	if err != nil {
		return fmt.Errorf("building trigger repo: %w", err)
	}
	cloneUri, repoPath := s.resolveCheckout(repo)
	trigger := pullTrigger(triggerRepo, ev, targetBranch)
	if !allowed {
		l.Info("blocking pull-triggered pipeline until approved", "actor", ev.Sender, "author", ev.Author)
	}
	pipelineId, err := s.runPipeline(ctx, ev.Pipeline, repo.RepoDid, trigger, nil, cloneUri, repoPath, ev.Head.Sha, nil, triggerRepo, !allowed)
	switch {
	case errors.Is(err, errRevRefused), errors.Is(err, db.ErrBannedSubject):
		l.Warn("dropping pull trigger", "sha", ev.Head.Sha, "err", err)
		return nil
	case err != nil:
		return err
	case pipelineId == "":
		l.Info("no workflow matched 'pull_request' trigger, skipping the event")
		return nil
	}
	l.Info("pipeline triggered", "pipeline", pipelineId)
	return nil
}

// random users' pipelines wait for approval, so running right away needs the author to
// have push access, and whoever caused the event to be the author or have push access too
func (s *Spindle) mayTriggerPull(repoDid, author, actor syntax.DID) (bool, error) {
	allowed, err := s.e.IsPushAllowed(author.String(), rbac.ThisServer, repoDid.String())
	if err != nil || !allowed || actor == author {
		return allowed, err
	}
	return s.e.IsPushAllowed(actor.String(), rbac.ThisServer, repoDid.String())
}

// the head is always checked out from the target repo, and a fork's branch name
// would pass for the target's own branch in refs and cache scopes, so only a
// branch of the target repo becomes the source branch
func pullTrigger(repo *workflow.TriggerRepo, ev db.PullEvent, targetBranch string) workflow.TriggerMetadata {
	action := pullActions[ev.Event]
	sourceBranch := ""
	if ev.Head.SourceRepo == ev.RepoDid.String() {
		sourceBranch = ev.Head.SourceBranch
	}
	pull := fmt.Sprintf("at://%s/%s/%s", ev.RepoDid, org_tangled.TrackTicketNSID, ev.Ticket)
	var patch *string
	if ev.Head.PatchRkey != "" {
		uri := fmt.Sprintf("at://%s/%s/%s", ev.RepoDid, org_tangled.ReviewPatchNSID, ev.Head.PatchRkey)
		patch = &uri
	}
	actor := ev.Sender.String()
	return workflow.TriggerMetadata{
		Kind: string(workflow.TriggerKindPullRequest),
		PullRequest: &workflow.PullRequestTriggerData{
			Action:       &action,
			Pull:         &pull,
			Patch:        patch,
			SourceBranch: sourceBranch,
			SourceSha:    ev.Head.Sha,
			TargetBranch: targetBranch,
		},
		Repo:  repo,
		Actor: &actor,
	}
}

func (s *Spindle) failPullEvent(ev db.PullEvent) {
	retained, err := s.retries.failWith(
		func() (int, error) { return ev.Attempts, nil },
		func(attempts int, due time.Time) error { return s.db.PutPullEventRetry(ev.ID, attempts, due) },
		func() error { return s.db.DeletePullEvent(ev.ID) },
	)
	switch {
	case err != nil:
		s.l.Error("saving a pull event retry", "repo", ev.RepoDid, "pull", ev.Ticket, "err", err)
	case !retained:
		s.l.Error("giving up on a pull request event", "repo", ev.RepoDid, "pull", ev.Ticket, "event", ev.Event)
	}
}

func (s *Spindle) failRecordMark(r db.RecordRetry) {
	retained, err := s.retries.failWith(
		func() (int, error) { return s.db.RecordRetryAttempts(r) },
		func(attempts int, due time.Time) error { return s.db.PutRecordRetry(r, attempts, due) },
		func() error { return s.db.DeleteRecordRetry(r) },
	)
	switch {
	case err != nil:
		s.l.Error("saving a record retry", "repo", r.RepoDid, "collection", r.Collection, "rkey", r.Rkey, "err", err)
	case !retained:
		s.l.Error("giving up on a pull request record", "repo", r.RepoDid, "collection", r.Collection, "rkey", r.Rkey)
	}
}

func (s *Spindle) servedRepo(repoDid syntax.DID) (*db.Repo, error) {
	repo, err := s.db.GetRepoByDid(repoDid)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return repo, err
}