~james/core

core/spindle/db/pulls.go -rw-r--r-- 8.4 KiB
452f17b7 — james chore(import): snapshot james.dad.town/core at f70c71c7696a 14 hours ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
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
}