alice.pds.demo.boltless.dev
This repository has no description
1package db
2
3import (
4 "context"
5 "database/sql"
6 "errors"
7 "fmt"
8 "log"
9 "slices"
10 "strings"
11 "time"
12
13 "github.com/bluesky-social/indigo/atproto/syntax"
14 "tangled.org/core/api/tangled"
15 "tangled.org/core/appview/models"
16 "tangled.org/core/appview/pagination"
17 "tangled.org/core/orm"
18)
19
20func GetRepos(e Execer, filters ...orm.Filter) ([]models.Repo, error) {
21 return GetReposPaginated(e, pagination.Page{}, filters...)
22}
23
24func GetReposPaginated(e Execer, page pagination.Page, filters ...orm.Filter) ([]models.Repo, error) {
25 var conditions []string
26 var args []any
27 for _, filter := range filters {
28 conditions = append(conditions, filter.Condition())
29 args = append(args, filter.Arg()...)
30 }
31
32 whereClause := ""
33 if conditions != nil {
34 whereClause = " where " + strings.Join(conditions, " and ")
35 }
36
37 pageClause := ""
38 if page.Limit != 0 {
39 pageClause = fmt.Sprintf(" limit %d offset %d", page.Limit, page.Offset)
40 }
41
42 // main query to get repos with pagination
43 query := fmt.Sprintf(`
44 select
45 id,
46 did,
47 name,
48 knot,
49 rkey,
50 created,
51 description,
52 website,
53 topics,
54 source,
55 spindle,
56 repo_did
57 from repos
58 %s
59 order by created desc
60 %s
61 `, whereClause, pageClause)
62
63 rows, err := e.Query(query, args...)
64 if err != nil {
65 return nil, err
66 }
67 defer rows.Close()
68
69 repoMap := make(map[syntax.ATURI]*models.Repo)
70 for rows.Next() {
71 var repo models.Repo
72 var createdAt string
73 var description, website, topicStr, source, spindle, repoDid sql.NullString
74
75 err := rows.Scan(
76 &repo.Id,
77 &repo.Did,
78 &repo.Name,
79 &repo.Knot,
80 &repo.Rkey,
81 &createdAt,
82 &description,
83 &website,
84 &topicStr,
85 &source,
86 &spindle,
87 &repoDid,
88 )
89 if err != nil {
90 return nil, err
91 }
92
93 // parse created timestamp
94 if t, err := time.Parse(time.RFC3339, createdAt); err == nil {
95 repo.Created = t
96 }
97
98 // handle nullable fields
99 if description.Valid {
100 repo.Description = description.String
101 }
102 if website.Valid {
103 repo.Website = website.String
104 }
105 if topicStr.Valid {
106 repo.Topics = strings.Fields(topicStr.String)
107 }
108 if source.Valid {
109 repo.Source = source.String
110 }
111 if spindle.Valid {
112 repo.Spindle = spindle.String
113 }
114 if repoDid.Valid {
115 repo.RepoDid = repoDid.String
116 }
117
118 repo.RepoStats = &models.RepoStats{}
119 repoMap[repo.RepoAt()] = &repo
120 }
121
122 if err = rows.Err(); err != nil {
123 return nil, err
124 }
125
126 // if no repos, return early
127 if len(repoMap) == 0 {
128 return nil, nil
129 }
130
131 // build IN clause for related queries
132 inClause := strings.TrimSuffix(strings.Repeat("?, ", len(repoMap)), ", ")
133 args = make([]any, len(repoMap))
134 i := 0
135 for _, r := range repoMap {
136 args[i] = r.RepoAt()
137 i++
138 }
139
140 // get labels for all repos
141 labelsQuery := fmt.Sprintf(
142 `select repo_at, label_at from repo_labels where repo_at in (%s)`,
143 inClause,
144 )
145
146 rows, err = e.Query(labelsQuery, args...)
147 if err != nil {
148 return nil, err
149 }
150 defer rows.Close()
151
152 for rows.Next() {
153 var repoat, labelat string
154 if err := rows.Scan(&repoat, &labelat); err != nil {
155 continue
156 }
157 if r, ok := repoMap[syntax.ATURI(repoat)]; ok {
158 r.Labels = append(r.Labels, labelat)
159 }
160 }
161
162 // get primary language for all repos
163 languageQuery := fmt.Sprintf(`
164 select repo_at, language
165 from (
166 select
167 repo_at, language,
168 row_number() over (
169 partition by repo_at
170 order by bytes desc
171 ) as rn
172 from repo_languages
173 where repo_at in (%s)
174 and is_default_ref = 1
175 and language <> ''
176 )
177 where rn = 1
178 `, inClause)
179
180 rows, err = e.Query(languageQuery, args...)
181 if err != nil {
182 return nil, fmt.Errorf("failed to execute lang query: %w", err)
183 }
184 defer rows.Close()
185
186 for rows.Next() {
187 var repoat, lang string
188 if err := rows.Scan(&repoat, &lang); err != nil {
189 log.Println("err", "err", err)
190 continue
191 }
192 if r, ok := repoMap[syntax.ATURI(repoat)]; ok {
193 r.RepoStats.Language = lang
194 }
195 }
196 if err = rows.Err(); err != nil {
197 return nil, fmt.Errorf("failed to execute lang query: %w", err)
198 }
199
200 // get star counts
201 starCountQuery := fmt.Sprintf(
202 `select subject_at, count(1) from stars where subject_at in (%s) group by subject_at`,
203 inClause,
204 )
205
206 rows, err = e.Query(starCountQuery, args...)
207 if err != nil {
208 return nil, fmt.Errorf("failed to execute star-count query: %w", err)
209 }
210 defer rows.Close()
211
212 for rows.Next() {
213 var repoat string
214 var count int
215 if err := rows.Scan(&repoat, &count); err != nil {
216 log.Println("err", "err", err)
217 continue
218 }
219 if r, ok := repoMap[syntax.ATURI(repoat)]; ok {
220 r.RepoStats.StarCount = count
221 }
222 }
223 if err = rows.Err(); err != nil {
224 return nil, fmt.Errorf("failed to execute star-count query: %w", err)
225 }
226
227 // get issue counts
228 issueCountQuery := fmt.Sprintf(`
229 select
230 repo_at,
231 count(case when open = 1 then 1 end) as open_count,
232 count(case when open = 0 then 1 end) as closed_count
233 from issues
234 where repo_at in (%s)
235 group by repo_at
236 `, inClause)
237
238 rows, err = e.Query(issueCountQuery, args...)
239 if err != nil {
240 return nil, fmt.Errorf("failed to execute issue-count query: %w", err)
241 }
242 defer rows.Close()
243
244 for rows.Next() {
245 var repoat string
246 var open, closed int
247 if err := rows.Scan(&repoat, &open, &closed); err != nil {
248 log.Println("err", "err", err)
249 continue
250 }
251 if r, ok := repoMap[syntax.ATURI(repoat)]; ok {
252 r.RepoStats.IssueCount.Open = open
253 r.RepoStats.IssueCount.Closed = closed
254 }
255 }
256 if err = rows.Err(); err != nil {
257 return nil, fmt.Errorf("failed to execute issue-count query: %w", err)
258 }
259
260 // get pull counts
261 pullCountQuery := fmt.Sprintf(`
262 select
263 repo_at,
264 count(case when state = ? then 1 end) as open_count,
265 count(case when state = ? then 1 end) as merged_count,
266 count(case when state = ? then 1 end) as closed_count,
267 count(case when state = ? then 1 end) as deleted_count
268 from pulls
269 where repo_at in (%s)
270 group by repo_at
271 `, inClause)
272
273 pullArgs := append([]any{
274 models.PullOpen,
275 models.PullMerged,
276 models.PullClosed,
277 models.PullAbandoned,
278 }, args...)
279
280 rows, err = e.Query(pullCountQuery, pullArgs...)
281 if err != nil {
282 return nil, fmt.Errorf("failed to execute pulls-count query: %w", err)
283 }
284 defer rows.Close()
285
286 for rows.Next() {
287 var repoat string
288 var open, merged, closed, deleted int
289 if err := rows.Scan(&repoat, &open, &merged, &closed, &deleted); err != nil {
290 log.Println("err", "err", err)
291 continue
292 }
293 if r, ok := repoMap[syntax.ATURI(repoat)]; ok {
294 r.RepoStats.PullCount.Open = open
295 r.RepoStats.PullCount.Merged = merged
296 r.RepoStats.PullCount.Closed = closed
297 r.RepoStats.PullCount.Deleted = deleted
298 }
299 }
300 if err = rows.Err(); err != nil {
301 return nil, fmt.Errorf("failed to execute pulls-count query: %w", err)
302 }
303
304 var repos []models.Repo
305 for _, r := range repoMap {
306 repos = append(repos, *r)
307 }
308
309 // sort by created timestamp (desc)
310 slices.SortFunc(repos, func(a, b models.Repo) int {
311 if a.Created.After(b.Created) {
312 return -1
313 }
314 return 1
315 })
316
317 return repos, nil
318}
319
320// helper to get exactly one repo
321func GetRepo(e Execer, filters ...orm.Filter) (*models.Repo, error) {
322 repos, err := GetReposPaginated(e, pagination.Page{Limit: 1}, filters...)
323 if err != nil {
324 return nil, err
325 }
326
327 if repos == nil {
328 return nil, sql.ErrNoRows
329 }
330
331 if len(repos) != 1 {
332 return nil, fmt.Errorf("too few rows returned")
333 }
334
335 return &repos[0], nil
336}
337
338func CountRepos(e Execer, filters ...orm.Filter) (int64, error) {
339 var conditions []string
340 var args []any
341 for _, filter := range filters {
342 conditions = append(conditions, filter.Condition())
343 args = append(args, filter.Arg()...)
344 }
345
346 whereClause := ""
347 if conditions != nil {
348 whereClause = " where " + strings.Join(conditions, " and ")
349 }
350
351 repoQuery := fmt.Sprintf(`select count(1) from repos %s`, whereClause)
352 var count int64
353 err := e.QueryRow(repoQuery, args...).Scan(&count)
354
355 if !errors.Is(err, sql.ErrNoRows) && err != nil {
356 return 0, err
357 }
358
359 return count, nil
360}
361
362func GetRepoByAtUri(e Execer, atUri string) (*models.Repo, error) {
363 var repo models.Repo
364 var nullableDescription sql.NullString
365 var nullableWebsite sql.NullString
366 var nullableTopicStr sql.NullString
367 var nullableRepoDid sql.NullString
368 var nullableSource sql.NullString
369 var nullableSpindle sql.NullString
370
371 row := e.QueryRow(`select id, did, name, knot, created, rkey, description, website, topics, source, spindle, repo_did from repos where at_uri = ?`, atUri)
372
373 var createdAt string
374 if err := row.Scan(&repo.Id, &repo.Did, &repo.Name, &repo.Knot, &createdAt, &repo.Rkey, &nullableDescription, &nullableWebsite, &nullableTopicStr, &nullableSource, &nullableSpindle, &nullableRepoDid); err != nil {
375 return nil, err
376 }
377 createdAtTime, _ := time.Parse(time.RFC3339, createdAt)
378 repo.Created = createdAtTime
379
380 if nullableDescription.Valid {
381 repo.Description = nullableDescription.String
382 }
383 if nullableWebsite.Valid {
384 repo.Website = nullableWebsite.String
385 }
386 if nullableTopicStr.Valid {
387 repo.Topics = strings.Fields(nullableTopicStr.String)
388 }
389 if nullableSource.Valid {
390 repo.Source = nullableSource.String
391 }
392 if nullableSpindle.Valid {
393 repo.Spindle = nullableSpindle.String
394 }
395 if nullableRepoDid.Valid {
396 repo.RepoDid = nullableRepoDid.String
397 }
398
399 return &repo, nil
400}
401
402func PutRepo(tx *sql.Tx, repo models.Repo) error {
403 var repoDid *string
404 if repo.RepoDid != "" {
405 repoDid = &repo.RepoDid
406 }
407 _, err := tx.Exec(
408 `update repos
409 set knot = ?, description = ?, website = ?, topics = ?, repo_did = coalesce(?, repo_did)
410 where did = ? and rkey = ?
411 `,
412 repo.Knot, repo.Description, repo.Website, repo.TopicStr(), repoDid, repo.Did, repo.Rkey,
413 )
414 return err
415}
416
417func AddRepo(tx *sql.Tx, repo *models.Repo) error {
418 var repoDid *string
419 if repo.RepoDid != "" {
420 repoDid = &repo.RepoDid
421 }
422 result, err := tx.Exec(
423 `insert into repos
424 (did, name, knot, rkey, at_uri, description, website, topics, source, repo_did)
425 values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
426 repo.Did, repo.Name, repo.Knot, repo.Rkey, repo.RepoAt().String(), repo.Description, repo.Website, repo.TopicStr(), repo.Source, repoDid,
427 )
428 if err != nil {
429 return fmt.Errorf("failed to insert repo: %w", err)
430 }
431
432 id, err := result.LastInsertId()
433 if err != nil {
434 return fmt.Errorf("failed to get last insert id: %w", err)
435 }
436 repo.Id = id
437
438 for _, dl := range repo.Labels {
439 if err := SubscribeLabel(tx, &models.RepoLabel{
440 RepoAt: repo.RepoAt(),
441 LabelAt: syntax.ATURI(dl),
442 }); err != nil {
443 return fmt.Errorf("failed to subscribe to label: %w", err)
444 }
445 }
446
447 return nil
448}
449
450func RemoveRepo(e Execer, did, name string) error {
451 _, err := e.Exec(`delete from repos where did = ? and name = ?`, did, name)
452 return err
453}
454
455func GetRepoSource(e Execer, repoAt syntax.ATURI) (string, error) {
456 var nullableSource sql.NullString
457 err := e.QueryRow(`select source from repos where at_uri = ?`, repoAt).Scan(&nullableSource)
458 if err != nil {
459 return "", err
460 }
461 return nullableSource.String, nil
462}
463
464func GetRepoSourceRepo(e Execer, repoAt syntax.ATURI) (*models.Repo, error) {
465 source, err := GetRepoSource(e, repoAt)
466 if source == "" || errors.Is(err, sql.ErrNoRows) {
467 return nil, nil
468 }
469 if err != nil {
470 return nil, err
471 }
472 if strings.HasPrefix(source, "did:") {
473 return GetRepoByDid(e, source)
474 }
475 return GetRepoByAtUri(e, source)
476}
477
478func GetForksByDid(e Execer, did string) ([]models.Repo, error) {
479 var repos []models.Repo
480
481 rows, err := e.Query(
482 `select distinct r.id, r.did, r.name, r.knot, r.rkey, r.description, r.website, r.created, r.source, r.repo_did
483 from repos r
484 left join collaborators c on r.at_uri = c.repo_at
485 where (r.did = ? or c.subject_did = ?)
486 and r.source is not null
487 and r.source != ''
488 order by r.created desc`,
489 did, did,
490 )
491 if err != nil {
492 return nil, err
493 }
494 defer rows.Close()
495
496 for rows.Next() {
497 var repo models.Repo
498 var createdAt string
499 var nullableDescription sql.NullString
500 var nullableWebsite sql.NullString
501 var nullableSource sql.NullString
502 var nullableRepoDid sql.NullString
503
504 err := rows.Scan(&repo.Id, &repo.Did, &repo.Name, &repo.Knot, &repo.Rkey, &nullableDescription, &nullableWebsite, &createdAt, &nullableSource, &nullableRepoDid)
505 if err != nil {
506 return nil, err
507 }
508
509 if nullableDescription.Valid {
510 repo.Description = nullableDescription.String
511 }
512 if nullableWebsite.Valid {
513 repo.Website = nullableWebsite.String
514 }
515
516 if nullableSource.Valid {
517 repo.Source = nullableSource.String
518 }
519 if nullableRepoDid.Valid {
520 repo.RepoDid = nullableRepoDid.String
521 }
522
523 createdAtTime, err := time.Parse(time.RFC3339, createdAt)
524 if err != nil {
525 repo.Created = time.Now()
526 } else {
527 repo.Created = createdAtTime
528 }
529
530 repos = append(repos, repo)
531 }
532
533 if err := rows.Err(); err != nil {
534 return nil, err
535 }
536
537 return repos, nil
538}
539
540func GetForkByDid(e Execer, did string, name string) (*models.Repo, error) {
541 var repo models.Repo
542 var createdAt string
543 var nullableDescription sql.NullString
544 var nullableWebsite sql.NullString
545 var nullableTopicStr sql.NullString
546 var nullableSource sql.NullString
547 var nullableRepoDid sql.NullString
548
549 row := e.QueryRow(
550 `select id, did, name, knot, rkey, description, website, topics, created, source, repo_did
551 from repos
552 where did = ? and name = ? and source is not null and source != ''`,
553 did, name,
554 )
555
556 err := row.Scan(&repo.Id, &repo.Did, &repo.Name, &repo.Knot, &repo.Rkey, &nullableDescription, &nullableWebsite, &nullableTopicStr, &createdAt, &nullableSource, &nullableRepoDid)
557 if err != nil {
558 return nil, err
559 }
560
561 if nullableDescription.Valid {
562 repo.Description = nullableDescription.String
563 }
564
565 if nullableWebsite.Valid {
566 repo.Website = nullableWebsite.String
567 }
568
569 if nullableTopicStr.Valid {
570 repo.Topics = strings.Fields(nullableTopicStr.String)
571 }
572
573 if nullableSource.Valid {
574 repo.Source = nullableSource.String
575 }
576 if nullableRepoDid.Valid {
577 repo.RepoDid = nullableRepoDid.String
578 }
579
580 createdAtTime, err := time.Parse(time.RFC3339, createdAt)
581 if err != nil {
582 repo.Created = time.Now()
583 } else {
584 repo.Created = createdAtTime
585 }
586
587 return &repo, nil
588}
589
590func GetRepoByDid(e Execer, repoDid string) (*models.Repo, error) {
591 return GetRepo(e, orm.FilterEq("repo_did", repoDid))
592}
593
594// TODO: just queue every legacy records regardless of target repo has a DID or not.
595// doable after we have `repo_did` column in db for each tables.
596func EnqueuePdsRewritesForRepo(tx *sql.Tx, repoDid, repoAtUri string) error {
597 type record struct {
598 userDidCol string
599 table string
600 nsid syntax.NSID
601 fkCol string
602 }
603 sources := []record{
604 {"did", "repos", tangled.RepoNSID, "at_uri"},
605 {"did", "issues", tangled.RepoIssueNSID, "repo_at"},
606 {"owner_did", "pulls", tangled.RepoPullNSID, "repo_at"},
607 {"did", "collaborators", tangled.RepoCollaboratorNSID, "repo_at"},
608 {"did", "artifacts", tangled.RepoArchiveNSID, "repo_at"},
609 {"did", "stars", tangled.FeedStarNSID, "subject_at"},
610 }
611
612 for _, src := range sources {
613 rows, err := tx.Query(
614 fmt.Sprintf(`SELECT %s, rkey FROM %s WHERE %s = ?`, src.userDidCol, src.table, src.fkCol),
615 repoAtUri,
616 )
617 if err != nil {
618 return fmt.Errorf("query %s for pds rewrites: %w", src.table, err)
619 }
620
621 var pairs []struct{ did, rkey string }
622 for rows.Next() {
623 var d, r string
624 if scanErr := rows.Scan(&d, &r); scanErr != nil {
625 rows.Close()
626 return fmt.Errorf("scan %s for pds rewrites: %w", src.table, scanErr)
627 }
628 pairs = append(pairs, struct{ did, rkey string }{d, r})
629 }
630 rows.Close()
631 if rowsErr := rows.Err(); rowsErr != nil {
632 return fmt.Errorf("iterate %s for pds rewrites: %w", src.table, rowsErr)
633 }
634
635 for _, p := range pairs {
636 if err := EnqueuePdsRecordMigration(context.Background(), tx, "add-repo-did", syntax.DID(p.did), src.nsid, syntax.RecordKey(p.rkey)); err != nil {
637 return fmt.Errorf("enqueue pds rewrite for %s/%s: %w", src.table, p.rkey, err)
638 }
639 }
640 }
641
642 profileRows, err := tx.Query(
643 `SELECT DISTINCT did FROM profile_pinned_repositories WHERE pin = ?`,
644 repoAtUri,
645 )
646 if err != nil {
647 return fmt.Errorf("query profile_pinned_repositories for pds rewrites: %w", err)
648 }
649 var profileDids []string
650 for profileRows.Next() {
651 var d string
652 if scanErr := profileRows.Scan(&d); scanErr != nil {
653 profileRows.Close()
654 return fmt.Errorf("scan profile_pinned_repositories for pds rewrites: %w", scanErr)
655 }
656 profileDids = append(profileDids, d)
657 }
658 profileRows.Close()
659 if profileRowsErr := profileRows.Err(); profileRowsErr != nil {
660 return fmt.Errorf("iterate profile_pinned_repositories for pds rewrites: %w", profileRowsErr)
661 }
662
663 for _, d := range profileDids {
664 if err := EnqueuePdsRecordMigration(context.Background(), tx, "add-repo-did", syntax.DID(d), tangled.ActorProfileNSID, "self"); err != nil {
665 return fmt.Errorf("enqueue pds rewrite for profile/%s: %w", d, err)
666 }
667 }
668
669 return nil
670}
671
672func CascadeRepoDid(tx *sql.Tx, repoAtUri, repoDid string) error {
673 _, err := tx.Exec(
674 `UPDATE repos SET repo_did = ? WHERE at_uri = ?`,
675 repoDid, repoAtUri,
676 )
677 if err != nil {
678 return fmt.Errorf("cascade repo_did to repos: %w", err)
679 }
680
681 _, err = tx.Exec(
682 `UPDATE repos SET source = ? WHERE source = ?`,
683 repoDid, repoAtUri,
684 )
685 if err != nil {
686 return fmt.Errorf("cascade repo_did to repos.source: %w", err)
687 }
688
689 return nil
690}
691
692func UpdateDescription(e Execer, repoAt, newDescription string) error {
693 _, err := e.Exec(
694 `update repos set description = ? where at_uri = ?`, newDescription, repoAt)
695 return err
696}
697
698func UpdateSpindle(e Execer, repoAt string, spindle *string) error {
699 _, err := e.Exec(
700 `update repos set spindle = ? where at_uri = ?`, spindle, repoAt)
701 return err
702}
703
704func SubscribeLabel(e Execer, rl *models.RepoLabel) error {
705 query := `insert or ignore into repo_labels (repo_at, label_at) values (?, ?)`
706
707 _, err := e.Exec(query, rl.RepoAt.String(), rl.LabelAt.String())
708 return err
709}
710
711func UnsubscribeLabel(e Execer, filters ...orm.Filter) error {
712 var conditions []string
713 var args []any
714 for _, filter := range filters {
715 conditions = append(conditions, filter.Condition())
716 args = append(args, filter.Arg()...)
717 }
718
719 whereClause := ""
720 if conditions != nil {
721 whereClause = " where " + strings.Join(conditions, " and ")
722 }
723
724 query := fmt.Sprintf(`delete from repo_labels %s`, whereClause)
725 _, err := e.Exec(query, args...)
726 return err
727}
728
729func GetRepoLabels(e Execer, filters ...orm.Filter) ([]models.RepoLabel, error) {
730 var conditions []string
731 var args []any
732 for _, filter := range filters {
733 conditions = append(conditions, filter.Condition())
734 args = append(args, filter.Arg()...)
735 }
736
737 whereClause := ""
738 if conditions != nil {
739 whereClause = " where " + strings.Join(conditions, " and ")
740 }
741
742 query := fmt.Sprintf(`select id, repo_at, label_at from repo_labels %s`, whereClause)
743
744 rows, err := e.Query(query, args...)
745 if err != nil {
746 return nil, err
747 }
748 defer rows.Close()
749
750 var labels []models.RepoLabel
751 for rows.Next() {
752 var label models.RepoLabel
753
754 err := rows.Scan(&label.Id, &label.RepoAt, &label.LabelAt)
755 if err != nil {
756 return nil, err
757 }
758
759 labels = append(labels, label)
760 }
761
762 if err = rows.Err(); err != nil {
763 return nil, err
764 }
765
766 return labels, nil
767}