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