alice.pds.demo.boltless.dev
This repository has no description
1package migration
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "time"
8
9 comatproto "github.com/bluesky-social/indigo/api/atproto"
10 "github.com/bluesky-social/indigo/atproto/atclient"
11 "github.com/bluesky-social/indigo/atproto/syntax"
12 lexutil "github.com/bluesky-social/indigo/lex/util"
13 "github.com/samber/lo"
14 cbg "github.com/whyrusleeping/cbor-gen"
15
16 "tangled.org/core/api/tangled"
17 "tangled.org/core/appview/db"
18 "tangled.org/core/appview/models"
19)
20
21func (s *Migration) backfillEntityState(ctx context.Context, client *atclient.APIClient, did syntax.DID, _ syntax.ATURI) error {
22 l := s.logger.With("migration", db.EntityStateBackfillName, "owner", did)
23
24 subjects, err := db.ColumnOnlyClosedSubjectsForOwner(ctx, s.db, did)
25 if err != nil {
26 return fmt.Errorf("db: %w", err)
27 }
28 if len(subjects) == 0 {
29 l.Info("no column-only closed entity state to backfill")
30 return nil
31 }
32
33 results := lo.Map(subjects, func(subj db.BackfillSubject, _ int) error {
34 if err := s.writeBackfillStateRecord(ctx, client, did, subj); err != nil {
35 l.Error("failed to backfill entity state record", "subject", subj.Subject, "err", err)
36 return err
37 }
38 return nil
39 })
40
41 written := lo.CountBy(results, func(err error) bool { return err == nil })
42 l.Info("backfilled entity state", "written", written, "total", len(subjects))
43
44 return errors.Join(results...)
45}
46
47func (s *Migration) writeBackfillStateRecord(ctx context.Context, client *atclient.APIClient, owner syntax.DID, subj db.BackfillSubject) error {
48 createdAt, err := time.Parse(time.RFC3339, subj.CreatedAt)
49 if err != nil {
50 createdAt = time.Unix(0, 0).UTC()
51 }
52
53 collection, record, err := backfillRecord(subj, createdAt)
54 if err != nil {
55 return err
56 }
57
58 _, err = comatproto.RepoPutRecord(ctx, client, &comatproto.RepoPutRecord_Input{
59 Repo: owner.String(),
60 Collection: collection,
61 Rkey: subj.Subject.RecordKey().String(),
62 Record: &lexutil.LexiconTypeDecoder{Val: record},
63 })
64 return err
65}
66
67func backfillRecord(subj db.BackfillSubject, createdAt time.Time) (string, cbg.CBORMarshaler, error) {
68 switch subj.Subject.Collection().String() {
69 case tangled.RepoIssueNSID:
70 record, err := models.AsIssueStateRecord(subj.Subject, subj.Value, createdAt)
71 if err != nil {
72 return "", nil, err
73 }
74 return tangled.RepoIssueStateNSID, &record, nil
75 case tangled.RepoPullNSID:
76 records, err := models.AsPullStatusRecords([]syntax.ATURI{subj.Subject}, subj.Value, createdAt)
77 if err != nil {
78 return "", nil, err
79 }
80 return tangled.RepoPullStatusNSID, &records[0], nil
81 default:
82 return "", nil, fmt.Errorf("unexpected backfill subject collection: %s", subj.Subject.Collection())
83 }
84}