package 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 ticket
type 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 now
type 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.ErrNoRows
func (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 retry
var 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 good
func (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 event
func (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 yet
func (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 saw
func 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 happened
func (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
}