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
}