Monorepo for Tangled
tangled.org
1package pulls
2
3import (
4 "context"
5 "iter"
6 "net/url"
7 "slices"
8 "time"
9
10 "tangled.org/core/api/tangled"
11 "tangled.org/core/appview/db"
12 "tangled.org/core/appview/models"
13 "tangled.org/core/orm"
14 "tangled.org/core/tid"
15
16 comatproto "github.com/bluesky-social/indigo/api/atproto"
17 "github.com/bluesky-social/indigo/atproto/atclient"
18 "github.com/bluesky-social/indigo/atproto/syntax"
19 lexutil "github.com/bluesky-social/indigo/lex/util"
20)
21
22func (s *Pulls) pullLabelDefs(repo *models.Repo) (map[string]*models.LabelDefinition, error) {
23 defs, err := db.GetLabelDefinitions(
24 s.db,
25 orm.FilterIn("at_uri", repo.Labels),
26 orm.FilterContains("scope", tangled.RepoPullNSID),
27 )
28 if err != nil {
29 return nil, err
30 }
31
32 out := make(map[string]*models.LabelDefinition, len(defs))
33 for i := range defs {
34 d := defs[i]
35 if !slices.Contains(d.Scope, tangled.RepoPullNSID) {
36 continue
37 }
38 out[d.AtUri().String()] = &d
39 }
40 return out, nil
41}
42
43func formLabelEntries(form url.Values, defs map[string]*models.LabelDefinition) iter.Seq2[string, string] {
44 return func(yield func(string, string) bool) {
45 for key := range defs {
46 for _, v := range form[key] {
47 if v == "" {
48 continue
49 }
50 if !yield(key, v) {
51 return
52 }
53 }
54 }
55 }
56}
57
58func labelStateFromForm(form url.Values, defs map[string]*models.LabelDefinition) models.LabelState {
59 state := models.NewLabelState()
60 actx := &models.LabelApplicationCtx{Defs: defs}
61 for key, val := range formLabelEntries(form, defs) {
62 _ = actx.ApplyLabelOp(state, models.LabelOp{
63 Operation: models.LabelOperationAdd,
64 OperandKey: key,
65 OperandValue: val,
66 })
67 }
68 return state
69}
70
71func buildCreationLabelOps(
72 userDid syntax.DID,
73 subject syntax.ATURI,
74 rkey string,
75 form url.Values,
76 defs map[string]*models.LabelDefinition,
77 performedAt time.Time,
78) []models.LabelOp {
79 var ops []models.LabelOp
80 for key, val := range formLabelEntries(form, defs) {
81 ops = append(ops, models.LabelOp{
82 Did: userDid.String(),
83 Rkey: rkey,
84 Subject: subject,
85 Operation: models.LabelOperationAdd,
86 OperandKey: key,
87 OperandValue: val,
88 PerformedAt: performedAt,
89 })
90 }
91 return ops
92}
93
94func (s *Pulls) applyCreationLabels(
95 ctx context.Context,
96 client *atclient.APIClient,
97 userDid syntax.DID,
98 pulls []*models.Pull,
99 form url.Values,
100 repo *models.Repo,
101) {
102 l := s.logger.With("handler", "applyCreationLabels", "user", userDid)
103
104 defs, err := s.pullLabelDefs(repo)
105 if err != nil {
106 l.Warn("failed to fetch label defs", "err", err)
107 return
108 }
109 if len(defs) == 0 {
110 return
111 }
112
113 perCidForms := parseStackLabelForms(form)
114
115 applyAll := form.Get("applyLabelsToAll") == "on"
116 var firstStackForm url.Values
117 if applyAll && len(pulls) > 0 && len(pulls[0].Submissions) > 0 {
118 if firstCid := pulls[0].Submissions[0].ChangeId(); firstCid != "" {
119 if f, ok := perCidForms[firstCid]; ok {
120 firstStackForm = f
121 }
122 }
123 }
124
125 performedAt := time.Now()
126 for _, pull := range pulls {
127 labelForm := form
128 if firstStackForm != nil {
129 labelForm = firstStackForm
130 } else if len(perCidForms) > 0 && len(pull.Submissions) > 0 {
131 if cid := pull.Submissions[0].ChangeId(); cid != "" {
132 if perForm, ok := perCidForms[cid]; ok {
133 labelForm = perForm
134 }
135 }
136 }
137 rkey := tid.TID()
138 raw := buildCreationLabelOps(userDid, pull.AtUri(), rkey, labelForm, defs, performedAt)
139
140 valid := make([]models.LabelOp, 0, len(raw))
141 for _, op := range raw {
142 def := defs[op.OperandKey]
143 if err := s.validator.ValidateLabelOp(def, repo, &op); err != nil {
144 l.Warn("invalid label op", "err", err, "subject", op.Subject, "key", op.OperandKey)
145 continue
146 }
147 valid = append(valid, op)
148 }
149 if len(valid) == 0 {
150 continue
151 }
152
153 record := models.LabelOpsAsRecord(valid)
154 if _, err := comatproto.RepoPutRecord(ctx, client, &comatproto.RepoPutRecord_Input{
155 Collection: tangled.LabelOpNSID,
156 Repo: userDid.String(),
157 Rkey: rkey,
158 Record: &lexutil.LexiconTypeDecoder{Val: &record},
159 }); err != nil {
160 l.Warn("failed to write label ops to PDS", "err", err, "subject", pull.AtUri())
161 continue
162 }
163
164 if err := s.indexLabelOps(ctx, valid); err != nil {
165 l.Warn("failed to index label ops", "err", err, "subject", pull.AtUri())
166 if _, err := comatproto.RepoDeleteRecord(context.Background(), client, &comatproto.RepoDeleteRecord_Input{
167 Collection: tangled.LabelOpNSID,
168 Repo: userDid.String(),
169 Rkey: rkey,
170 }); err != nil {
171 l.Warn("failed to rollback label ops record from PDS", "err", err, "subject", pull.AtUri())
172 }
173 continue
174 }
175
176 s.notifier.NewPullLabelOp(ctx, pull)
177 }
178}
179
180func (s *Pulls) indexLabelOps(ctx context.Context, ops []models.LabelOp) error {
181 tx, err := s.db.BeginTx(ctx, nil)
182 if err != nil {
183 return err
184 }
185 defer tx.Rollback()
186 for _, op := range ops {
187 if _, err := db.AddLabelOp(tx, &op); err != nil {
188 return err
189 }
190 }
191 return tx.Commit()
192}