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[string]*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.RepoDid] = &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.RepoDid
137 i++
138 }
139
140 // get labels for all repos
141 labelsQuery := fmt.Sprintf(
142 `select repo_did, label_at from repo_labels where repo_did 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 repoDid, labelat string
154 if err := rows.Scan(&repoDid, &labelat); err != nil {
155 continue
156 }
157 if r, ok := repoMap[repoDid]; 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_did, language
165 from (
166 select
167 repo_did, language,
168 row_number() over (
169 partition by repo_did
170 order by bytes desc
171 ) as rn
172 from repo_languages
173 where repo_did 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 repoDid, lang string
188 if err := rows.Scan(&repoDid, &lang); err != nil {
189 log.Println("err", "err", err)
190 continue
191 }
192 if r, ok := repoMap[repoDid]; 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, count(1) from stars where subject_type = 'repo' and subject in (%s) group by subject`,
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 repoDid string
214 var count int
215 if err := rows.Scan(&repoDid, &count); err != nil {
216 log.Println("err", "err", err)
217 continue
218 }
219 if r, ok := repoMap[repoDid]; 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_did,
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_did in (%s)
235 group by repo_did
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 repoDid string
246 var open, closed int
247 if err := rows.Scan(&repoDid, &open, &closed); err != nil {
248 log.Println("err", "err", err)
249 continue
250 }
251 if r, ok := repoMap[repoDid]; 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_did,
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_did in (%s)
270 group by repo_did
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 repoDid string
288 var open, merged, closed, deleted int
289 if err := rows.Scan(&repoDid, &open, &merged, &closed, &deleted); err != nil {
290 log.Println("err", "err", err)
291 continue
292 }
293 if r, ok := repoMap[repoDid]; 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 name = ?, knot = ?, description = ?, website = ?, topics = ?, repo_did = coalesce(?, repo_did)
410 where did = ? and rkey = ?
411 `,
412 repo.Name, 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 RepoDid: syntax.DID(repo.RepoDid),
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, rkey string) error {
451 _, err := e.Exec(`delete from repos where did = ? and rkey = ?`, did, rkey)
452 return err
453}
454
455func GetRepoSource(e Execer, repoDid string) (string, error) {
456 var nullableSource sql.NullString
457 err := e.QueryRow(`select source from repos where repo_did = ?`, repoDid).Scan(&nullableSource)
458 if err != nil {
459 return "", err
460 }
461 return nullableSource.String, nil
462}
463
464func GetRepoSourceRepo(e Execer, repoDid string) (*models.Repo, error) {
465 source, err := GetRepoSource(e, repoDid)
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.repo_did = c.repo_did
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, rkey 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 rkey = ? and source is not null and source != ''`,
553 did, rkey,
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 fkVal string
603 }
604 sources := []record{
605 {"did", "repos", tangled.RepoNSID, "at_uri", repoAtUri},
606 {"did", "issues", tangled.RepoIssueNSID, "repo_did", repoDid},
607 {"owner_did", "pulls", tangled.RepoPullNSID, "repo_did", repoDid},
608 {"did", "collaborators", tangled.RepoCollaboratorNSID, "repo_did", repoDid},
609 {"did", "artifacts", tangled.RepoArchiveNSID, "repo_did", repoDid},
610 {"did", "stars", tangled.FeedStarNSID, "subject", repoDid},
611 }
612
613 for _, src := range sources {
614 rows, err := tx.Query(
615 fmt.Sprintf(`SELECT %s, rkey FROM %s WHERE %s = ?`, src.userDidCol, src.table, src.fkCol),
616 src.fkVal,
617 )
618 if err != nil {
619 return fmt.Errorf("query %s for pds rewrites: %w", src.table, err)
620 }
621
622 var pairs []struct{ did, rkey string }
623 for rows.Next() {
624 var d, r string
625 if scanErr := rows.Scan(&d, &r); scanErr != nil {
626 rows.Close()
627 return fmt.Errorf("scan %s for pds rewrites: %w", src.table, scanErr)
628 }
629 pairs = append(pairs, struct{ did, rkey string }{d, r})
630 }
631 rows.Close()
632 if rowsErr := rows.Err(); rowsErr != nil {
633 return fmt.Errorf("iterate %s for pds rewrites: %w", src.table, rowsErr)
634 }
635
636 for _, p := range pairs {
637 if err := EnqueuePdsRecordMigration(context.Background(), tx, "add-repo-did", syntax.DID(p.did), src.nsid, syntax.RecordKey(p.rkey)); err != nil {
638 return fmt.Errorf("enqueue pds rewrite for %s/%s: %w", src.table, p.rkey, err)
639 }
640 }
641 }
642
643 profileRows, err := tx.Query(
644 `SELECT DISTINCT did FROM profile_pinned_repositories WHERE pin = ?`,
645 repoAtUri,
646 )
647 if err != nil {
648 return fmt.Errorf("query profile_pinned_repositories for pds rewrites: %w", err)
649 }
650 var profileDids []string
651 for profileRows.Next() {
652 var d string
653 if scanErr := profileRows.Scan(&d); scanErr != nil {
654 profileRows.Close()
655 return fmt.Errorf("scan profile_pinned_repositories for pds rewrites: %w", scanErr)
656 }
657 profileDids = append(profileDids, d)
658 }
659 profileRows.Close()
660 if profileRowsErr := profileRows.Err(); profileRowsErr != nil {
661 return fmt.Errorf("iterate profile_pinned_repositories for pds rewrites: %w", profileRowsErr)
662 }
663
664 for _, d := range profileDids {
665 if err := EnqueuePdsRecordMigration(context.Background(), tx, "add-repo-did", syntax.DID(d), tangled.ActorProfileNSID, "self"); err != nil {
666 return fmt.Errorf("enqueue pds rewrite for profile/%s: %w", d, err)
667 }
668 }
669
670 return nil
671}
672
673func CascadeRepoDid(tx *sql.Tx, repoAtUri, repoDid string) error {
674 _, err := tx.Exec(
675 `UPDATE repos SET repo_did = ? WHERE at_uri = ?`,
676 repoDid, repoAtUri,
677 )
678 if err != nil {
679 return fmt.Errorf("cascade repo_did to repos: %w", err)
680 }
681
682 _, err = tx.Exec(
683 `UPDATE repos SET source = ? WHERE source = ?`,
684 repoDid, repoAtUri,
685 )
686 if err != nil {
687 return fmt.Errorf("cascade repo_did to repos.source: %w", err)
688 }
689
690 return nil
691}
692
693func UpdateDescription(e Execer, repoDid, newDescription string) error {
694 _, err := e.Exec(
695 `update repos set description = ? where repo_did = ?`, newDescription, repoDid)
696 return err
697}
698
699func UpdateSpindle(e Execer, repoDid string, spindle *string) error {
700 _, err := e.Exec(
701 `update repos set spindle = ? where repo_did = ?`, spindle, repoDid)
702 return err
703}
704
705func SubscribeLabel(e Execer, rl *models.RepoLabel) error {
706 query := `insert or ignore into repo_labels (repo_did, label_at) values (?, ?)`
707
708 _, err := e.Exec(query, string(rl.RepoDid), rl.LabelAt.String())
709 return err
710}
711
712func UnsubscribeLabel(e Execer, filters ...orm.Filter) error {
713 var conditions []string
714 var args []any
715 for _, filter := range filters {
716 conditions = append(conditions, filter.Condition())
717 args = append(args, filter.Arg()...)
718 }
719
720 whereClause := ""
721 if conditions != nil {
722 whereClause = " where " + strings.Join(conditions, " and ")
723 }
724
725 query := fmt.Sprintf(`delete from repo_labels %s`, whereClause)
726 _, err := e.Exec(query, args...)
727 return err
728}
729
730func GetRepoLabels(e Execer, filters ...orm.Filter) ([]models.RepoLabel, error) {
731 var conditions []string
732 var args []any
733 for _, filter := range filters {
734 conditions = append(conditions, filter.Condition())
735 args = append(args, filter.Arg()...)
736 }
737
738 whereClause := ""
739 if conditions != nil {
740 whereClause = " where " + strings.Join(conditions, " and ")
741 }
742
743 query := fmt.Sprintf(`select id, repo_did, label_at from repo_labels %s`, whereClause)
744
745 rows, err := e.Query(query, args...)
746 if err != nil {
747 return nil, err
748 }
749 defer rows.Close()
750
751 var labels []models.RepoLabel
752 for rows.Next() {
753 var label models.RepoLabel
754
755 err := rows.Scan(&label.Id, &label.RepoDid, &label.LabelAt)
756 if err != nil {
757 return nil, err
758 }
759
760 labels = append(labels, label)
761 }
762
763 if err = rows.Err(); err != nil {
764 return nil, err
765 }
766
767 return labels, nil
768}