this repo has no description
tangled.org
1package spindle
2
3import (
4 "context"
5 "encoding/json"
6 "fmt"
7 "path/filepath"
8
9 "tangled.sh/tangled.sh/core/api/tangled"
10 "tangled.sh/tangled.sh/core/eventconsumer"
11 "tangled.sh/tangled.sh/core/idresolver"
12 "tangled.sh/tangled.sh/core/rbac"
13
14 comatproto "github.com/bluesky-social/indigo/api/atproto"
15 "github.com/bluesky-social/indigo/atproto/identity"
16 "github.com/bluesky-social/indigo/atproto/syntax"
17 "github.com/bluesky-social/indigo/xrpc"
18 "github.com/bluesky-social/jetstream/pkg/models"
19 securejoin "github.com/cyphar/filepath-securejoin"
20)
21
22type Ingester func(ctx context.Context, e *models.Event) error
23
24func (s *Spindle) ingest() Ingester {
25 return func(ctx context.Context, e *models.Event) error {
26 var err error
27 defer func() {
28 eventTime := e.TimeUS
29 lastTimeUs := eventTime + 1
30 if err := s.db.SaveLastTimeUs(lastTimeUs); err != nil {
31 err = fmt.Errorf("(deferred) failed to save last time us: %w", err)
32 }
33 }()
34
35 if e.Kind != models.EventKindCommit {
36 return nil
37 }
38
39 switch e.Commit.Collection {
40 case tangled.SpindleMemberNSID:
41 s.ingestMember(ctx, e)
42 case tangled.RepoNSID:
43 s.ingestRepo(ctx, e)
44 case tangled.RepoCollaboratorNSID:
45 s.ingestCollaborator(ctx, e)
46 }
47
48 return err
49 }
50}
51
52func (s *Spindle) ingestMember(_ context.Context, e *models.Event) error {
53 did := e.Did
54 var err error
55
56 l := s.l.With("component", "ingester", "record", tangled.SpindleMemberNSID)
57
58 switch e.Commit.Operation {
59 case models.CommitOperationCreate, models.CommitOperationUpdate:
60 raw := e.Commit.Record
61 record := tangled.SpindleMember{}
62 err = json.Unmarshal(raw, &record)
63 if err != nil {
64 l.Error("invalid record", "error", err)
65 return err
66 }
67
68 domain := s.cfg.Server.Hostname
69 if s.cfg.Server.Dev {
70 domain = s.cfg.Server.ListenAddr
71 }
72 recordInstance := record.Instance
73
74 if recordInstance != domain {
75 l.Error("domain mismatch", "domain", recordInstance, "expected", domain)
76 return fmt.Errorf("domain mismatch: %s != %s", record.Instance, domain)
77 }
78
79 ok, err := s.e.IsSpindleInviteAllowed(did, rbacDomain)
80 if err != nil || !ok {
81 l.Error("failed to add member", "did", did, "error", err)
82 return fmt.Errorf("failed to enforce permissions: %w", err)
83 }
84
85 if err := s.e.AddSpindleMember(rbacDomain, record.Subject); err != nil {
86 l.Error("failed to add member", "error", err)
87 return fmt.Errorf("failed to add member: %w", err)
88 }
89 l.Info("added member from firehose", "member", record.Subject)
90
91 if err := s.db.AddDid(record.Subject); err != nil {
92 l.Error("failed to add did", "error", err)
93 return fmt.Errorf("failed to add did: %w", err)
94 }
95 s.jc.AddDid(record.Subject)
96
97 return nil
98
99 }
100 return nil
101}
102
103func (s *Spindle) ingestRepo(_ context.Context, e *models.Event) error {
104 var err error
105
106 l := s.l.With("component", "ingester", "record", tangled.RepoNSID)
107
108 l.Info("ingesting repo record")
109
110 switch e.Commit.Operation {
111 case models.CommitOperationCreate, models.CommitOperationUpdate:
112 raw := e.Commit.Record
113 record := tangled.Repo{}
114 err = json.Unmarshal(raw, &record)
115 if err != nil {
116 l.Error("invalid record", "error", err)
117 return err
118 }
119
120 domain := s.cfg.Server.Hostname
121
122 // no spindle configured for this repo
123 if record.Spindle == nil {
124 l.Info("no spindle configured", "did", record.Owner, "name", record.Name)
125 return nil
126 }
127
128 // this repo did not want this spindle
129 if *record.Spindle != domain {
130 l.Info("different spindle configured", "did", record.Owner, "name", record.Name, "spindle", *record.Spindle, "domain", domain)
131 return nil
132 }
133
134 // add this repo to the watch list
135 if err := s.db.AddRepo(record.Knot, record.Owner, record.Name); err != nil {
136 l.Error("failed to add repo", "error", err)
137 return fmt.Errorf("failed to add repo: %w", err)
138 }
139
140 // add repo to rbac
141 if err := s.e.AddRepo(record.Owner, rbac.ThisServer, filepath.Join(record.Owner, record.Name)); err != nil {
142 l.Error("failed to add repo to enforcer", "error", err)
143 return fmt.Errorf("failed to add repo: %w", err)
144 }
145
146 // add this knot to the event consumer
147 src := eventconsumer.NewKnotSource(record.Knot)
148 s.ks.AddSource(context.Background(), src)
149
150 return nil
151
152 }
153 return nil
154}
155
156func (s *Spindle) ingestCollaborator(ctx context.Context, e *models.Event) error {
157 var err error
158
159 l := s.l.With("component", "ingester", "record", tangled.RepoCollaboratorNSID, "did", e.Did)
160
161 l.Info("ingesting collaborator record")
162
163 switch e.Commit.Operation {
164 case models.CommitOperationCreate, models.CommitOperationUpdate:
165 raw := e.Commit.Record
166 record := tangled.RepoCollaborator{}
167 err = json.Unmarshal(raw, &record)
168 if err != nil {
169 l.Error("invalid record", "error", err)
170 return err
171 }
172
173 resolver := idresolver.DefaultResolver()
174
175 subjectId, err := resolver.ResolveIdent(ctx, record.Subject)
176 if err != nil || subjectId.Handle.IsInvalidHandle() {
177 return err
178 }
179
180 repoAt, err := syntax.ParseATURI(record.Repo)
181 if err != nil {
182 l.Info("rejecting record, invalid repoAt", "repoAt", record.Repo)
183 return nil
184 }
185
186 // TODO: get rid of this entirely
187 // resolve this aturi to extract the repo record
188 owner, err := resolver.ResolveIdent(ctx, repoAt.Authority().String())
189 if err != nil || owner.Handle.IsInvalidHandle() {
190 return fmt.Errorf("failed to resolve handle: %w", err)
191 }
192
193 xrpcc := xrpc.Client{
194 Host: owner.PDSEndpoint(),
195 }
196
197 resp, err := comatproto.RepoGetRecord(ctx, &xrpcc, "", tangled.RepoNSID, repoAt.Authority().String(), repoAt.RecordKey().String())
198 if err != nil {
199 return err
200 }
201
202 repo := resp.Value.Val.(*tangled.Repo)
203 didSlashRepo, _ := securejoin.SecureJoin(owner.DID.String(), repo.Name)
204
205 // check perms for this user
206 if ok, err := s.e.IsCollaboratorInviteAllowed(owner.DID.String(), rbac.ThisServer, didSlashRepo); !ok || err != nil {
207 return fmt.Errorf("insufficient permissions: %w", err)
208 }
209
210 // add collaborator to rbac
211 if err := s.e.AddCollaborator(record.Subject, rbac.ThisServer, didSlashRepo); err != nil {
212 l.Error("failed to add repo to enforcer", "error", err)
213 return fmt.Errorf("failed to add repo: %w", err)
214 }
215
216 return nil
217 }
218 return nil
219}
220
221func (s *Spindle) fetchAndAddCollaborators(ctx context.Context, owner *identity.Identity, didSlashRepo string) error {
222 l := s.l.With("component", "ingester", "handler", "fetchAndAddCollaborators")
223
224 l.Info("fetching and adding existing collaborators")
225
226 xrpcc := xrpc.Client{
227 Host: owner.PDSEndpoint(),
228 }
229
230 resp, err := comatproto.RepoListRecords(ctx, &xrpcc, tangled.RepoCollaboratorNSID, "", 50, owner.DID.String(), false)
231 if err != nil {
232 return err
233 }
234
235 var errs error
236 for _, r := range resp.Records {
237 if r == nil {
238 continue
239 }
240 record := r.Value.Val.(*tangled.RepoCollaborator)
241
242 if err := s.e.AddCollaborator(record.Subject, rbac.ThisServer, didSlashRepo); err != nil {
243 l.Error("failed to add repo to enforcer", "error", err)
244 errors.Join(errs, fmt.Errorf("failed to add repo: %w", err))
245 }
246 }
247
248 return errs
249}