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