Something went wrong. Try again.
forked from tangled.org/core
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273package db
import ( "database/sql" "errors" "fmt" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/spindle/models")
type PullTicket struct { Author syntax.DID State string // empty on rows written before versions were tracked Version string}
type PullPatch struct { Rkey string Ticket string Head string // where the patch came from, empty for patches written before the knot recorded it SourceRepo string SourceBranch string CreatedAt string}
// the newest patch of a tickettype PullHead struct { Sha string SourceRepo string SourceBranch string PatchRkey syntax.RecordKey}
// a derived pull request event that hasn't fully run yet, it carries everything// it needs so a retry describes the pull as it was, not as it is nowtype PullEvent struct { ID int64 RepoDid syntax.DID Ticket string Event models.WebhookEvent Author syntax.DID Sender syntax.DID // the webhook's pull request without the round and source, which come from the patches Pull []byte Rounds int Head PullHead Pipeline models.PipelineId WebhookFired bool Attempts int}
// an unknown ticket comes back as sql.ErrNoRowsfunc (d *DB) GetPullTicket(repoDid syntax.DID, rkey string) (*PullTicket, error) { var t PullTicket var author string if err := d.QueryRow( `select author, state, version from pull_tickets where repo_did = ? and rkey = ?`, repoDid.String(), rkey, ).Scan(&author, &t.State, &t.Version); err != nil { return nil, err } t.Author = syntax.DID(author) return &t, nil}
func (d *DB) AddPullTicket(repoDid syntax.DID, rkey string, ticket PullTicket) error { _, err := d.Exec( `insert or ignore into pull_tickets (repo_did, rkey, author, state, version) values (?, ?, ?, ?, ?)`, repoDid.String(), rkey, ticket.Author.String(), ticket.State, ticket.Version, ) return err}
// the ticket moved since it was read, re-read the row and retryvar ErrPullRaced = errors.New("the ticket changed since it was read")
func movedOne(res sql.Result, err error) error { if err != nil { return err } if n, err := res.RowsAffected(); err != nil || n == 0 { return errors.Join(err, ErrPullRaced) } return nil}
func (d *DB) AdvancePullTicket(repoDid syntax.DID, rkey, prev, version, state string, ev *PullEvent) error { tx, err := d.Begin() if err != nil { return err } defer tx.Rollback() if err := movedOne(tx.Exec( `update pull_tickets set state = ?, version = ? where repo_did = ? and rkey = ? and version = ?`, state, version, repoDid.String(), rkey, prev, )); err != nil { return err } if ev != nil { if err := addPullEvent(tx, *ev); err != nil { return err } } return tx.Commit()}
// the state and its event commit together, or a crash between them loses the event for goodfunc (d *DB) SetPullTicketState(repoDid syntax.DID, rkey, prev, state string, ev PullEvent) error { tx, err := d.Begin() if err != nil { return err } defer tx.Rollback() if err := movedOne(tx.Exec( `update pull_tickets set state = ? where repo_did = ? and rkey = ? and state = ?`, state, repoDid.String(), rkey, prev, )); err != nil { return err } if err := addPullEvent(tx, ev); err != nil { return err } return tx.Commit()}
// a patch that was already seen is a replay, so it stores no eventfunc (d *DB) AddPullPatch(repoDid syntax.DID, patch PullPatch, ev PullEvent) error { tx, err := d.Begin() if err != nil { return err } defer tx.Rollback() res, err := tx.Exec( `insert or ignore into pull_patches (repo_did, rkey, ticket, head, source_repo, source_branch, created_at) values (?, ?, ?, ?, ?, ?, ?)`, repoDid.String(), patch.Rkey, patch.Ticket, patch.Head, patch.SourceRepo, patch.SourceBranch, patch.CreatedAt, ) if err != nil { return fmt.Errorf("failed to record pull patch: %w", err) } if n, err := res.RowsAffected(); err != nil || n == 0 { return err } ev.Event = models.WebhookEventPullRequestResubmitted if rounds, _, err := pullRounds(tx, repoDid, patch.Ticket); err != nil { return err } else if rounds == 1 { ev.Event = models.WebhookEventPullRequestCreated } if err := addPullEvent(tx, ev); err != nil { return err } return tx.Commit()}
// head is empty when the ticket has no patch yetfunc (d *DB) PullRounds(repoDid syntax.DID, ticket string) (rounds int, head PullHead, err error) { return pullRounds(d, repoDid, ticket)}
func pullRounds(q DBTX, repoDid syntax.DID, ticket string) (rounds int, head PullHead, err error) { if err := q.QueryRow( `select count(*) from pull_patches where repo_did = ? and ticket = ?`, repoDid.String(), ticket, ).Scan(&rounds); err != nil || rounds == 0 { return 0, PullHead{}, err } err = q.QueryRow( `select head, source_repo, source_branch, rkey from pull_patches where repo_did = ? and ticket = ? order by created_at desc, rowid desc limit 1`, repoDid.String(), ticket, ).Scan(&head.Sha, &head.SourceRepo, &head.SourceBranch, &head.PatchRkey) return rounds, head, err}
// rounds and head are read in the same tx, so they're the ones the event sawfunc addPullEvent(tx *sql.Tx, ev PullEvent) error { rounds, head, err := pullRounds(tx, ev.RepoDid, ev.Ticket) if err != nil { return err } _, err = tx.Exec( `insert into pull_events (repo_did, ticket, event, author, sender, pull, rounds, head, source_repo, source_branch, patch, pipeline_id, due) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ev.RepoDid.String(), ev.Ticket, string(ev.Event), ev.Author.String(), ev.Sender.String(), ev.Pull, rounds, head.Sha, head.SourceRepo, head.SourceBranch, head.PatchRkey, string(ev.Pipeline), time.Now().Unix(), ) return err}
// oldest first, so a pull's webhooks go out in the order its events happenedfunc (d *DB) DuePullEvents(repoDid syntax.DID, now time.Time, limit int) ([]PullEvent, error) { rows, err := d.Query( `select id, ticket, event, author, sender, pull, rounds, head, source_repo, source_branch, patch, pipeline_id, webhook_fired, attempts from pull_events where repo_did = ? and due <= ? order by id limit ?`, repoDid.String(), now.Unix(), limit, ) if err != nil { return nil, err } defer rows.Close() var out []PullEvent for rows.Next() { ev := PullEvent{RepoDid: repoDid} var event, author, sender, pipeline string if err := rows.Scan(&ev.ID, &ev.Ticket, &event, &author, &sender, &ev.Pull, &ev.Rounds, &ev.Head.Sha, &ev.Head.SourceRepo, &ev.Head.SourceBranch, &ev.Head.PatchRkey, &pipeline, &ev.WebhookFired, &ev.Attempts); err != nil { return nil, err } ev.Event, ev.Author, ev.Sender, ev.Pipeline = models.WebhookEvent(event), syntax.DID(author), syntax.DID(sender), models.PipelineId(pipeline) out = append(out, ev) } return out, rows.Err()}
func (d *DB) DuePullEventRepos(now time.Time) ([]syntax.DID, error) { return queryDIDs(d, `select distinct repo_did from pull_events where due <= ?`, now.Unix())}
func (d *DB) MarkPullEventWebhookFired(id int64) error { _, err := d.Exec(`update pull_events set webhook_fired = 1 where id = ?`, id) return err}
func (d *DB) PutPullEventRetry(id int64, attempts int, due time.Time) error { _, err := d.Exec(`update pull_events set attempts = ?, due = ? where id = ?`, attempts, due.Unix(), id) return err}
func (d *DB) DeletePullEvent(id int64) error { _, err := d.Exec(`delete from pull_events where id = ?`, id) return err}
func (d *DB) DeletePullEventsByRepo(repoDid syntax.DID) error { _, err := d.Exec(`delete from pull_events where repo_did = ?`, repoDid.String()) return err}
func (d *DB) PipelineExists(id models.PipelineId) (bool, error) { var one int err := d.QueryRow(`select 1 from pipelines where pipeline_id = ?`, string(id)).Scan(&one) if errors.Is(err, sql.ErrNoRows) { return false, nil } return err == nil, err}
func (d *DB) ForgetPullPatch(repoDid syntax.DID, rkey string) error { _, err := d.Exec(`delete from pull_patches where repo_did = ? and rkey = ?`, repoDid.String(), rkey) return err}
func (d *DB) ForgetPullTicket(repoDid syntax.DID, rkey string) error { if _, err := d.Exec(`delete from pull_patches where repo_did = ? and ticket = ?`, repoDid.String(), rkey); err != nil { return err } _, err := d.Exec(`delete from pull_tickets where repo_did = ? and rkey = ?`, repoDid.String(), rkey) return err}
func (d *DB) DeletePullsByRepo(repoDid syntax.DID) error { for _, table := range []string{"pull_patches", "pull_tickets", "pull_events", "record_retries"} { if _, err := d.Exec(`delete from `+table+` where repo_did = ?`, repoDid.String()); err != nil { return err } } return nil}