alice.pds.demo.boltless.dev
This repository has no description
1package spindle
2
3import (
4 "context"
5 "database/sql"
6 "encoding/json"
7 "errors"
8 "fmt"
9
10 "tangled.org/core/api/tangled"
11 "tangled.org/core/spindle/db"
12 "tangled.org/core/tapc"
13
14 "github.com/bluesky-social/indigo/atproto/syntax"
15 "github.com/bluesky-social/jetstream/pkg/models"
16)
17
18type Ingester func(ctx context.Context, e *models.Event) error
19
20func (s *Spindle) ingest() Ingester {
21 return func(ctx context.Context, e *models.Event) error {
22 if e.Kind != models.EventKindCommit {
23 return nil
24 }
25
26 var err error
27 switch e.Commit.Collection {
28 case tangled.SpindleMemberNSID:
29 err = s.ingestMember(ctx, e)
30 case tangled.RepoNSID, tangled.RepoCollaboratorNSID:
31 if evt, ok := jetstreamToTapEvent(e); ok {
32 err = s.tap.processEvent(ctx, evt)
33 }
34 }
35
36 if err != nil {
37 s.l.Warn("failed to process message, skipping", "nsid", e.Commit.Collection, "did", e.Did, "rkey", e.Commit.RKey, "err", err)
38 }
39
40 return nil
41 }
42}
43
44func jetstreamToTapEvent(e *models.Event) (tapc.Event, bool) {
45 if e.Commit == nil {
46 return tapc.Event{}, false
47 }
48 did, err := syntax.ParseDID(e.Did)
49 if err != nil {
50 return tapc.Event{}, false
51 }
52 var action tapc.RecordAction
53 switch e.Commit.Operation {
54 case models.CommitOperationCreate:
55 action = tapc.RecordCreateAction
56 case models.CommitOperationUpdate:
57 action = tapc.RecordUpdateAction
58 case models.CommitOperationDelete:
59 action = tapc.RecordDeleteAction
60 default:
61 return tapc.Event{}, false
62 }
63 return tapc.Event{
64 Type: tapc.EvtRecord,
65 Record: &tapc.RecordEventData{
66 Did: did,
67 Rkey: syntax.RecordKey(e.Commit.RKey),
68 Collection: syntax.NSID(e.Commit.Collection),
69 Action: action,
70 Record: e.Commit.Record,
71 },
72 }, true
73}
74
75func (s *Spindle) ingestMember(ctx context.Context, e *models.Event) error {
76 did := e.Did
77 rkey := e.Commit.RKey
78 l := s.l.With("component", "ingester", "record", tangled.SpindleMemberNSID, "did", did, "rkey", rkey)
79
80 switch e.Commit.Operation {
81 case models.CommitOperationCreate, models.CommitOperationUpdate:
82 raw := e.Commit.Record
83 record := tangled.SpindleMember{}
84 if err := json.Unmarshal(raw, &record); err != nil {
85 return fmt.Errorf("invalid record: %w", err)
86 }
87
88 domain := s.cfg.Server.Hostname
89 recordInstance := record.Instance
90
91 if recordInstance != domain {
92 return fmt.Errorf("domain mismatch: %s != %s", record.Instance, domain)
93 }
94
95 subject, err := syntax.ParseDID(record.Subject)
96 if err != nil {
97 return fmt.Errorf("invalid subject DID %q: %w", record.Subject, err)
98 }
99
100 ok, err := s.e.IsSpindleInviteAllowed(did, rbacDomain)
101 if err != nil {
102 return fmt.Errorf("failed to enforce permissions: %w", err)
103 }
104 if !ok {
105 return fmt.Errorf("permission denied for %s", did)
106 }
107
108 sqlTx, err := s.db.BeginTx(ctx, nil)
109 if err != nil {
110 return fmt.Errorf("failed to start txn: %w", err)
111 }
112 committed := false
113 defer func() {
114 if !committed {
115 sqlTx.Rollback()
116 }
117 }()
118
119 existing, err := db.GetSpindleMember(sqlTx, did, rkey)
120 if err != nil && !errors.Is(err, sql.ErrNoRows) {
121 return fmt.Errorf("failed to look up existing member: %w", err)
122 }
123
124 var staleSubject string
125 if existing != nil && existing.Subject != subject {
126 staleSubject = existing.Subject.String()
127 if err := db.RemoveSpindleMember(sqlTx, did, rkey); err != nil {
128 return fmt.Errorf("failed to remove stale member row: %w", err)
129 }
130 }
131
132 if err := db.AddSpindleMember(sqlTx, db.SpindleMember{
133 Did: syntax.DID(did),
134 Rkey: rkey,
135 Instance: recordInstance,
136 Subject: subject,
137 }); err != nil {
138 return fmt.Errorf("failed to add member: %w", err)
139 }
140
141 if err := db.AddDid(sqlTx, subject.String()); err != nil {
142 return fmt.Errorf("failed to add did: %w", err)
143 }
144
145 dropStaleAcl := false
146 var staleDidDropped bool
147 if staleSubject != "" {
148 remaining, err := db.CountSpindleMembersBySubject(sqlTx, staleSubject)
149 if err != nil {
150 return fmt.Errorf("failed to count stale subject rows: %w", err)
151 }
152 if remaining == 0 {
153 dropStaleAcl = true
154 stillNeeded, err := s.e.WouldHaveAnyPolicyExcludingSpindleMember(staleSubject, rbacDomain)
155 if err != nil {
156 return fmt.Errorf("failed to check residual policies for stale subject: %w", err)
157 }
158 if !stillNeeded {
159 if err := db.RemoveDid(sqlTx, staleSubject); err != nil {
160 return fmt.Errorf("failed to remove stale did: %w", err)
161 }
162 staleDidDropped = true
163 }
164 }
165 l.Info("replaced stale spindle member", "old_subject", staleSubject, "new_subject", subject, "stale_did_dropped", staleDidDropped)
166 }
167
168 if err := sqlTx.Commit(); err != nil {
169 return fmt.Errorf("failed to commit txn: %w", err)
170 }
171 committed = true
172
173 if dropStaleAcl {
174 if _, err := s.e.TryRemoveSpindleMember(rbacDomain, staleSubject); err != nil {
175 l.Error("post-commit: failed to remove stale ACL", "subject", staleSubject, "err", err)
176 }
177 }
178 if _, err := s.e.TryAddSpindleMember(rbacDomain, subject.String()); err != nil {
179 l.Error("post-commit: failed to add member ACL", "subject", subject, "err", err)
180 }
181
182 if staleDidDropped {
183 s.jc.RemoveDid(staleSubject)
184 }
185 s.jc.AddDid(subject.String())
186 l.Info("added member from firehose", "member", subject)
187 return nil
188
189 case models.CommitOperationDelete:
190 sqlTx, err := s.db.BeginTx(ctx, nil)
191 if err != nil {
192 return fmt.Errorf("failed to start txn: %w", err)
193 }
194 committed := false
195 defer func() {
196 if !committed {
197 sqlTx.Rollback()
198 }
199 }()
200
201 record, err := db.GetSpindleMember(sqlTx, did, rkey)
202 if errors.Is(err, sql.ErrNoRows) {
203 l.Info("spindle member already removed")
204 return nil
205 }
206 if err != nil {
207 return fmt.Errorf("failed to find member: %w", err)
208 }
209
210 staleSubject := record.Subject.String()
211
212 if err := db.RemoveSpindleMember(sqlTx, did, rkey); err != nil {
213 return fmt.Errorf("failed to remove member: %w", err)
214 }
215
216 remaining, err := db.CountSpindleMembersBySubject(sqlTx, staleSubject)
217 if err != nil {
218 return fmt.Errorf("failed to count remaining member rows: %w", err)
219 }
220
221 dropAcl := false
222 var staleDidDropped bool
223 if remaining == 0 {
224 dropAcl = true
225 stillNeeded, err := s.e.WouldHaveAnyPolicyExcludingSpindleMember(staleSubject, rbacDomain)
226 if err != nil {
227 return fmt.Errorf("failed to check residual policies: %w", err)
228 }
229 if !stillNeeded {
230 if err := db.RemoveDid(sqlTx, staleSubject); err != nil {
231 return fmt.Errorf("failed to remove did: %w", err)
232 }
233 staleDidDropped = true
234 }
235 }
236
237 if err := sqlTx.Commit(); err != nil {
238 return fmt.Errorf("failed to commit txn: %w", err)
239 }
240 committed = true
241
242 if dropAcl {
243 if _, err := s.e.TryRemoveSpindleMember(rbacDomain, staleSubject); err != nil {
244 l.Error("post-commit: failed to remove member ACL", "subject", staleSubject, "err", err)
245 }
246 }
247
248 if staleDidDropped {
249 s.jc.RemoveDid(staleSubject)
250 }
251 l.Info("removed member from firehose", "member", record.Subject, "remaining_rows", remaining, "stale_did_dropped", staleDidDropped)
252 }
253 return nil
254}