alice.pds.demo.boltless.dev
This repository has no description
1package db
2
3import (
4 "context"
5 "database/sql"
6 "fmt"
7 "maps"
8 "slices"
9 "sort"
10 "strings"
11 "time"
12
13 "github.com/bluesky-social/indigo/atproto/syntax"
14 "tangled.org/core/appview/models"
15 "tangled.org/core/appview/pagination"
16 "tangled.org/core/orm"
17)
18
19func PutPull(ctx context.Context, tx *sql.Tx, pull *models.Pull, references []syntax.ATURI) error {
20 // ensure sequence exists
21 _, err := tx.ExecContext(ctx, `
22 insert or ignore into repo_pull_seqs (repo_did, next_pull_id)
23 values (?, 1)
24 `, pull.RepoDid)
25 if err != nil {
26 return err
27 }
28
29 var exists bool
30 if err := tx.QueryRowContext(ctx,
31 `select exists (select 1 from pulls where at_uri = ?)`,
32 pull.AtUri(),
33 ).Scan(&exists); err != nil {
34 return err
35 }
36
37 if !exists {
38 // assign new ID for a PR
39 if err := tx.QueryRowContext(ctx,
40 `update repo_pull_seqs
41 set next_pull_id = next_pull_id + 1
42 where repo_did = ?
43 returning next_pull_id - 1`,
44 pull.RepoDid,
45 ).Scan(&pull.PullId); err != nil {
46 return err
47 }
48 }
49
50 result, err := tx.ExecContext(ctx,
51 `insert into pulls (
52 owner_did,
53 rkey,
54 cid,
55 repo_did,
56 pull_id,
57 title,
58 body,
59 target_branch,
60 source_repo_did,
61 source_branch,
62 created,
63 state
64 )
65 values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
66 on conflict(at_uri) do update set
67 cid = excluded.cid,
68 repo_did = excluded.repo_did,
69 title = excluded.title,
70 body = excluded.body,
71 target_branch = excluded.target_branch,
72 source_repo_did = excluded.source_repo_did,
73 source_branch = excluded.source_branch,
74 created = excluded.created,
75 state = excluded.state
76 where pulls.cid is not excluded.cid`,
77 pull.OwnerDid,
78 pull.Rkey,
79 pull.Cid,
80 pull.RepoDid,
81 pull.PullId,
82 pull.Title,
83 pull.Body,
84 pull.TargetBranch,
85 pull.SourceRepo,
86 pull.SourceBranch,
87 pull.Created.Format(time.RFC3339),
88 pull.State,
89 )
90 if err != nil {
91 return fmt.Errorf("inserting pr: %w", err)
92 }
93
94 id, err := result.LastInsertId()
95 if err != nil {
96 return err
97 }
98 pull.ID = id
99
100 // delete all existing versions
101 if _, err := tx.ExecContext(ctx,
102 `delete from pull_versions where pull_at = ?`,
103 pull.AtUri(),
104 ); err != nil {
105 return fmt.Errorf("deleting old pr versions: %w", err)
106 }
107
108 // re-create all versions
109 if len(pull.Versions) > 0 {
110 pullAt := pull.AtUri()
111 var sb strings.Builder
112 sb.WriteString(`insert into pull_versions (pull_at, id, head, base, created) values `)
113 args := make([]any, 0, len(pull.Versions)*5)
114 for i, v := range pull.Versions {
115 if i > 0 {
116 sb.WriteString(", ")
117 }
118 sb.WriteString("(?, ?, ?, ?, ?)")
119 args = append(args, pullAt, v.ID, v.Head, v.Base, v.Created.Format(time.RFC3339))
120 }
121 if _, err := tx.ExecContext(ctx, sb.String(), args...); err != nil {
122 return fmt.Errorf("inserting pr versions: %w", err)
123 }
124 }
125
126 // update references when comment is updated
127 if err := putReferences(tx, pull.AtUri(), references); err != nil {
128 return fmt.Errorf("put reference_links: %w", err)
129 }
130
131 return nil
132}
133
134func SubmitPullVersion(ctx context.Context, q Execer, pullAt syntax.ATURI, version models.PullVersion) error {
135 _, err := q.ExecContext(ctx,
136 `insert into pull_versions (pull_at, id, head, base, created)
137 values (?, ?, ?, ?, ?)`,
138 pullAt,
139 version.ID,
140 version.Head,
141 version.Base,
142 version.Created.Format(time.RFC3339),
143 )
144 return err
145}
146
147func GetPull(ctx context.Context, q Execer, filters ...orm.Filter) (*models.Pull, error) {
148 pulls, err := GetPullsPaginated(ctx, q, pagination.Page{Limit: 1}, filters...)
149 if err != nil {
150 return nil, err
151 }
152 if len(pulls) == 0 {
153 return nil, sql.ErrNoRows
154 }
155 return pulls[0], nil
156}
157
158func GetPullsPaginated(ctx context.Context, q Execer, page pagination.Page, filters ...orm.Filter) ([]*models.Pull, error) {
159 pulls := make(map[syntax.ATURI]*models.Pull)
160
161 var conditions []string
162 var args []any
163 for _, filter := range filters {
164 conditions = append(conditions, filter.Condition())
165 args = append(args, filter.Arg()...)
166 }
167
168 whereClause := ""
169 if conditions != nil {
170 whereClause = " where " + strings.Join(conditions, " and ")
171 }
172 pageClause := ""
173 if page.Limit != 0 {
174 pageClause = fmt.Sprintf(
175 " limit %d offset %d ",
176 page.Limit,
177 page.Offset,
178 )
179 }
180
181 query := fmt.Sprintf(`
182 select
183 id,
184 owner_did,
185 rkey,
186 cid,
187 repo_did,
188 pull_id,
189 title,
190 body,
191 target_branch,
192 source_repo_did,
193 source_branch,
194 created,
195 state
196 from
197 pulls
198 %s
199 order by
200 created desc
201 %s
202 `, whereClause, pageClause)
203
204 rows, err := q.QueryContext(ctx, query, args...)
205 if err != nil {
206 return nil, err
207 }
208 defer rows.Close()
209
210 for rows.Next() {
211 var pull models.Pull
212 var createdAt string
213 var sourceRepo, sourceBranch sql.NullString
214 err := rows.Scan(
215 &pull.ID,
216 &pull.OwnerDid,
217 &pull.Rkey,
218 &pull.Cid,
219 &pull.RepoDid,
220 &pull.PullId,
221 &pull.Title,
222 &pull.Body,
223 &pull.TargetBranch,
224 &sourceRepo,
225 &sourceBranch,
226 &createdAt,
227 &pull.State,
228 )
229 if err != nil {
230 return nil, fmt.Errorf("scanning row: %w", err)
231 }
232
233 createdTime, err := time.Parse(time.RFC3339, createdAt)
234 if err != nil {
235 return nil, fmt.Errorf("parsing created: %w", err)
236 }
237 pull.Created = createdTime
238
239 if sourceRepo.Valid {
240 pull.SourceRepo = syntax.DID(sourceRepo.String)
241 } else {
242 // fallback to pull.target.repo
243 pull.SourceRepo = pull.RepoDid
244 }
245
246 if sourceBranch.Valid {
247 pull.SourceBranch = &sourceBranch.String
248 }
249
250 pulls[pull.AtUri()] = &pull
251 }
252 if err := rows.Err(); err != nil {
253 return nil, fmt.Errorf("scanning rows: %w", err)
254 }
255
256 pullAts := slices.Collect(maps.Keys(pulls))
257
258 versionsMap, err := ListVersions(ctx, q, pullAts)
259 if err != nil {
260 return nil, fmt.Errorf("querying versions: %w", err)
261 }
262
263 for pullAt, p := range pulls {
264 if versions, ok := versionsMap[pullAt]; ok {
265 p.Versions = versions
266 } else {
267 return nil, fmt.Errorf("find 0 versions for PR %s", pullAt)
268 }
269 }
270
271 // collect reverse repos
272 {
273 repoDids := make([]string, 0, len(pulls))
274 for _, issue := range pulls {
275 repoDids = append(repoDids, string(issue.RepoDid))
276 }
277
278 repos, err := GetRepos(q, orm.FilterIn("repo_did", repoDids))
279 if err != nil {
280 return nil, fmt.Errorf("failed to build repo mappings: %w", err)
281 }
282 repoMap := make(map[syntax.DID]*models.Repo)
283 for i := range repos {
284 repoMap[syntax.DID(repos[i].RepoDid)] = &repos[i]
285 }
286
287 for pullAt, p := range pulls {
288 if r, ok := repoMap[p.RepoDid]; ok {
289 p.Repo = r
290 } else {
291 delete(pulls, pullAt)
292 }
293 }
294 }
295
296 // collect allLabels for each PR
297 {
298 allLabels, err := GetLabels(q, orm.FilterIn("subject", pullAts))
299 if err != nil {
300 return nil, fmt.Errorf("failed to query labels: %w", err)
301 }
302 for pullAt, labels := range allLabels {
303 if pull, ok := pulls[pullAt]; ok {
304 pull.Labels = labels
305 }
306 }
307 }
308
309 orderedById := []*models.Pull{}
310 for _, p := range pulls {
311 orderedById = append(orderedById, p)
312 }
313 sort.Slice(orderedById, func(i, j int) bool {
314 return orderedById[i].PullId > orderedById[j].PullId
315 })
316
317 return orderedById, nil
318}
319
320// mapping from pull -> pull submissions
321func ListVersions(ctx context.Context, q Execer, pullAts []syntax.ATURI) (map[syntax.ATURI][]models.PullVersion, error) {
322 filter := orm.FilterIn("pull_at", pullAts)
323
324 query := fmt.Sprintf(`
325 select
326 pull_at,
327 id,
328 head,
329 base,
330 created
331 from pull_versions
332 where %s
333 order by id asc
334 `, filter.Condition())
335
336 rows, err := q.QueryContext(ctx, query, filter.Arg()...)
337 if err != nil {
338 return nil, fmt.Errorf("failed to query: %w", err)
339 }
340 defer rows.Close()
341
342 versionsMap := make(map[syntax.ATURI][]models.PullVersion)
343
344 for rows.Next() {
345 var version models.PullVersion
346 var pullAt syntax.ATURI
347 var createdAt string
348 err := rows.Scan(
349 &pullAt,
350 &version.ID,
351 &version.Head,
352 &version.Base,
353 &createdAt,
354 )
355 if err != nil {
356 return nil, fmt.Errorf("scanning row: %w", err)
357 }
358
359 createdTime, err := time.Parse(time.RFC3339, createdAt)
360 if err != nil {
361 return nil, fmt.Errorf("parsing created: %w", err)
362 }
363 version.Created = createdTime
364
365 versionsMap[pullAt] = append(versionsMap[pullAt], version)
366 }
367 if err := rows.Err(); err != nil {
368 return nil, fmt.Errorf("scanning rows: %w", err)
369 }
370
371 comments, err := GetComments(q, orm.FilterIn("subject_uri", pullAts))
372 if err != nil {
373 return nil, fmt.Errorf("failed to get pull comments: %w", err)
374 }
375 for _, comment := range comments {
376 if comment.PullRoundIdx == nil {
377 continue
378 }
379 versionIdx := *comment.PullRoundIdx
380 if versions, ok := versionsMap[syntax.ATURI(comment.Subject.Uri)]; ok {
381 if versionIdx >= len(versions) {
382 continue
383 }
384 versions[versionIdx].Comments = append(versions[versionIdx].Comments, comment)
385 }
386 }
387
388 // TODO: reverse-map version.Comments
389
390 return versionsMap, nil
391}
392
393// timeframe here is directly passed into the sql query filter, and any
394// timeframe in the past should be negative; e.g.: "-3 months"
395func GetPullsByOwnerDid(e Execer, did syntax.DID, timeframe string) ([]models.Pull, error) {
396 var pulls []models.Pull
397
398 rows, err := e.Query(`
399 select
400 p.owner_did,
401 p.repo_did,
402 p.pull_id,
403 p.created,
404 p.title,
405 p.state,
406 r.did,
407 r.name,
408 r.knot,
409 r.rkey,
410 r.created
411 from
412 pulls p
413 join
414 repos r on p.repo_did = r.repo_did
415 where
416 p.owner_did = ? and p.created >= date ('now', ?)
417 order by
418 p.created desc`, did, timeframe)
419 if err != nil {
420 return nil, err
421 }
422 defer rows.Close()
423
424 for rows.Next() {
425 var pull models.Pull
426 var repo models.Repo
427 var pullCreatedAt, repoCreatedAt string
428 err := rows.Scan(
429 &pull.OwnerDid,
430 &pull.RepoDid,
431 &pull.PullId,
432 &pullCreatedAt,
433 &pull.Title,
434 &pull.State,
435 &repo.Did,
436 &repo.Name,
437 &repo.Knot,
438 &repo.Rkey,
439 &repoCreatedAt,
440 )
441 if err != nil {
442 return nil, err
443 }
444
445 pullCreatedTime, err := time.Parse(time.RFC3339, pullCreatedAt)
446 if err != nil {
447 return nil, err
448 }
449 pull.Created = pullCreatedTime
450
451 repoCreatedTime, err := time.Parse(time.RFC3339, repoCreatedAt)
452 if err != nil {
453 return nil, err
454 }
455 repo.Created = repoCreatedTime
456
457 pull.Repo = &repo
458
459 pulls = append(pulls, pull)
460 }
461
462 if err := rows.Err(); err != nil {
463 return nil, err
464 }
465
466 return pulls, nil
467}
468
469// use with transaction
470func setPullsState(e Execer, pullState models.PullState, filters ...orm.Filter) error {
471 var conditions []string
472 var args []any
473
474 args = append(args, pullState)
475 for _, filter := range filters {
476 conditions = append(conditions, filter.Condition())
477 args = append(args, filter.Arg()...)
478 }
479 args = append(args, models.PullAbandoned) // only update state of non-deleted pulls
480 args = append(args, models.PullMerged) // only update state of non-merged pulls
481
482 whereClause := ""
483 if conditions != nil {
484 whereClause = " where " + strings.Join(conditions, " and ")
485 }
486
487 query := fmt.Sprintf("update pulls set state = ? %s and state <> ? and state <> ?", whereClause)
488
489 _, err := e.Exec(query, args...)
490 return err
491}
492
493func ClosePulls(e Execer, filters ...orm.Filter) error {
494 return setPullsState(e, models.PullClosed, filters...)
495}
496
497func ReopenPulls(e Execer, filters ...orm.Filter) error {
498 return setPullsState(e, models.PullOpen, filters...)
499}
500
501func MergePulls(e Execer, filters ...orm.Filter) error {
502 return setPullsState(e, models.PullMerged, filters...)
503}
504
505func AbandonPulls(e Execer, filters ...orm.Filter) error {
506 return setPullsState(e, models.PullAbandoned, filters...)
507}
508
509func GetPullCount(e Execer, repoDid string) (models.PullCount, error) {
510 row := e.QueryRow(`
511 select
512 count(case when state = ? then 1 end) as open_count,
513 count(case when state = ? then 1 end) as merged_count,
514 count(case when state = ? then 1 end) as closed_count,
515 count(case when state = ? then 1 end) as deleted_count
516 from pulls
517 where repo_did = ?`,
518 models.PullOpen,
519 models.PullMerged,
520 models.PullClosed,
521 models.PullAbandoned,
522 repoDid,
523 )
524
525 var count models.PullCount
526 if err := row.Scan(&count.Open, &count.Merged, &count.Closed, &count.Deleted); err != nil {
527 return models.PullCount{Open: 0, Merged: 0, Closed: 0, Deleted: 0}, err
528 }
529
530 return count, nil
531}