alice.pds.demo.boltless.dev
This repository has no description
1package state
2
3import (
4 "context"
5 "database/sql"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "slices"
10 "strings"
11 "time"
12
13 "tangled.org/core/appview/cloudflare"
14 "tangled.org/core/appview/notify"
15
16 "tangled.org/core/api/tangled"
17 "tangled.org/core/appview/cache"
18 "tangled.org/core/appview/config"
19 "tangled.org/core/appview/db"
20 "tangled.org/core/appview/models"
21 "tangled.org/core/appview/sites"
22 ec "tangled.org/core/eventconsumer"
23 "tangled.org/core/eventconsumer/cursor"
24 knotdb "tangled.org/core/knotserver/db"
25 "tangled.org/core/log"
26 "tangled.org/core/orm"
27 "tangled.org/core/rbac"
28 "tangled.org/core/workflow"
29
30 "github.com/bluesky-social/indigo/atproto/syntax"
31 "github.com/go-git/go-git/v5/plumbing"
32 "github.com/posthog/posthog-go"
33)
34
35func Knotstream(ctx context.Context, c *config.Config, d *db.DB, enforcer *rbac.Enforcer, posthog posthog.Client, notifier notify.Notifier, cfClient *cloudflare.Client) (*ec.Consumer, error) {
36 logger := log.FromContext(ctx)
37 logger = log.SubLogger(logger, "knotstream")
38
39 knots, err := db.GetRegistrations(
40 d,
41 orm.FilterIsNot("registered", "null"),
42 )
43 if err != nil {
44 return nil, err
45 }
46
47 srcs := make(map[ec.Source]struct{})
48 for _, k := range knots {
49 s := ec.NewKnotSource(k.Domain)
50 srcs[s] = struct{}{}
51 }
52
53 cache := cache.New(c.Redis.Addr)
54 cursorStore := cursor.NewRedisCursorStore(cache)
55
56 cfg := ec.ConsumerConfig{
57 Sources: srcs,
58 ProcessFunc: knotIngester(d, enforcer, posthog, notifier, c.Core.Dev, c, cfClient),
59 RetryInterval: c.Knotstream.RetryInterval,
60 MaxRetryInterval: c.Knotstream.MaxRetryInterval,
61 ConnectionTimeout: c.Knotstream.ConnectionTimeout,
62 WorkerCount: c.Knotstream.WorkerCount,
63 QueueSize: c.Knotstream.QueueSize,
64 Logger: logger,
65 Dev: c.Core.Dev,
66 CursorStore: &cursorStore,
67 }
68
69 return ec.NewConsumer(cfg), nil
70}
71
72func resolveRepo(d *db.DB, repoDid *string, ownerDid, repoName string) (*models.Repo, error) {
73 if repoDid != nil && *repoDid != "" {
74 return db.GetRepoByDid(d, *repoDid)
75 }
76 repos, err := db.GetRepos(d, orm.FilterEq("did", ownerDid), orm.FilterEq("rkey", strings.ToLower(repoName)))
77 if err != nil {
78 return nil, err
79 }
80 if len(repos) == 0 {
81 return nil, sql.ErrNoRows
82 }
83 return &repos[0], nil
84}
85
86func knotIngester(d *db.DB, enforcer *rbac.Enforcer, posthog posthog.Client, notifier notify.Notifier, dev bool, c *config.Config, cfClient *cloudflare.Client) ec.ProcessFunc {
87 return func(ctx context.Context, source ec.Source, msg ec.Message) error {
88 switch msg.Nsid {
89 case tangled.GitRefUpdateNSID:
90 return ingestRefUpdate(ctx, d, enforcer, posthog, notifier, dev, c, cfClient, source, msg)
91 case tangled.PipelineNSID:
92 return ingestPipeline(d, source, msg)
93 case knotdb.RepoDIDAssignNSID:
94 return ingestDIDAssign(d, enforcer, source, msg, ctx)
95 }
96
97 return nil
98 }
99}
100
101// TODO(boltless): remove this. knotmirror should do all sort of indexing
102func ingestRefUpdate(ctx context.Context, d *db.DB, enforcer *rbac.Enforcer, pc posthog.Client, notifier notify.Notifier, dev bool, c *config.Config, cfClient *cloudflare.Client, source ec.Source, msg ec.Message) error {
103 logger := log.FromContext(ctx)
104
105 var record tangled.GitRefUpdate
106 err := json.Unmarshal(msg.EventJson, &record)
107 if err != nil {
108 return err
109 }
110
111 knownKnots, err := enforcer.GetKnotsForUser(record.CommitterDid)
112 if err != nil {
113 return err
114 }
115 if !slices.Contains(knownKnots, source.Key()) {
116 return fmt.Errorf("%s does not belong to %s, something is fishy", record.CommitterDid, source.Key())
117 }
118
119 if record.Repo == "" {
120 return fmt.Errorf("gitRefUpdate from %s missing repo", source.Key())
121 }
122
123 repo, lookupErr := db.GetRepoByDid(d, record.Repo)
124 if lookupErr != nil {
125 return fmt.Errorf("failed to look up repo: %w", lookupErr)
126 }
127
128 logger.Info("processing gitRefUpdate event",
129 "repo", repo.RepoIdentifier(),
130 "ref", record.Ref,
131 "old_sha", record.OldSha,
132 "new_sha", record.NewSha)
133
134 notifier.Push(ctx, repo, record.Ref, record.OldSha, record.NewSha, record.CommitterDid)
135
136 errPunchcard := populatePunchcard(d, record)
137 errLanguages := updateRepoLanguages(d, record)
138
139 var errPosthog error
140 if !dev && record.CommitterDid != "" {
141 errPosthog = pc.Enqueue(posthog.Capture{
142 DistinctId: record.CommitterDid,
143 Event: "git_ref_update",
144 })
145 }
146
147 // Trigger a sites redeploy if this push is to the configured sites branch.
148 if cfClient.Enabled() {
149 go triggerSitesDeployIfNeeded(ctx, d, cfClient, c, record, source)
150 }
151
152 return errors.Join(errPunchcard, errLanguages, errPosthog)
153}
154
155// triggerSitesDeployIfNeeded checks whether the pushed ref matches the sites
156// branch configured for this repo and, if so, syncs the site to R2
157func triggerSitesDeployIfNeeded(ctx context.Context, d *db.DB, cfClient *cloudflare.Client, cfg *config.Config, record tangled.GitRefUpdate, source ec.Source) {
158 logger := log.FromContext(ctx)
159
160 ref := plumbing.ReferenceName(record.Ref)
161 if !ref.IsBranch() {
162 return
163 }
164 pushedBranch := ref.Short()
165
166 repo, err := db.GetRepoByDid(d, record.Repo)
167 if err != nil {
168 return
169 }
170
171 siteConfig, err := db.GetRepoSiteConfig(d, repo.RepoDid)
172 if err != nil || siteConfig == nil {
173 return
174 }
175 if siteConfig.Branch != pushedBranch {
176 return
177 }
178
179 deploy := &models.SiteDeploy{
180 RepoDid: syntax.DID(repo.RepoDid),
181 Branch: siteConfig.Branch,
182 Dir: siteConfig.Dir,
183 CommitSHA: record.NewSha,
184 Trigger: models.SiteDeployTriggerPush,
185 }
186
187 deployErr := sites.Deploy(ctx, cfClient, cfg, repo, siteConfig.Branch, siteConfig.Dir)
188 if deployErr != nil {
189 logger.Error("sites: R2 sync failed on push", "repo", repo.RepoIdentifier(), "err", deployErr)
190 deploy.Status = models.SiteDeployStatusFailure
191 deploy.Error = deployErr.Error()
192 } else {
193 deploy.Status = models.SiteDeployStatusSuccess
194 }
195
196 if err := db.AddSiteDeploy(d, deploy); err != nil {
197 logger.Error("sites: failed to record deploy", "repo", repo.RepoIdentifier(), "err", err)
198 }
199
200 if deployErr == nil {
201 logger.Info("site deployed to r2", "repo", repo.RepoIdentifier())
202 }
203}
204
205func populatePunchcard(d *db.DB, record tangled.GitRefUpdate) error {
206 if record.CommitterDid == "" {
207 return nil
208 }
209
210 knownEmails, err := db.GetAllEmails(d, record.CommitterDid)
211 if err != nil {
212 return err
213 }
214
215 count := 0
216 for _, ke := range knownEmails {
217 if record.Meta == nil {
218 continue
219 }
220 if record.Meta.CommitCount == nil {
221 continue
222 }
223 for _, ce := range record.Meta.CommitCount.ByEmail {
224 if ce == nil {
225 continue
226 }
227 if ce.Email == ke.Address || ce.Email == record.CommitterDid {
228 count += int(ce.Count)
229 }
230 }
231 }
232
233 punch := models.Punch{
234 Did: record.CommitterDid,
235 Date: time.Now(),
236 Count: count,
237 }
238 return db.AddPunch(d, punch)
239}
240
241func updateRepoLanguages(d *db.DB, record tangled.GitRefUpdate) error {
242 if record.Meta == nil || record.Meta.LangBreakdown == nil || record.Meta.LangBreakdown.Inputs == nil {
243 return fmt.Errorf("empty language data for repo: %s", record.Repo)
244 }
245
246 r, lookupErr := db.GetRepoByDid(d, record.Repo)
247 if lookupErr != nil {
248 return fmt.Errorf("failed to look up repo: %w", lookupErr)
249 }
250 repo := *r
251
252 ref := plumbing.ReferenceName(record.Ref)
253 if !ref.IsBranch() {
254 return fmt.Errorf("%s is not a valid reference name", ref)
255 }
256
257 var langs []models.RepoLanguage
258 for _, l := range record.Meta.LangBreakdown.Inputs {
259 if l == nil {
260 continue
261 }
262
263 langs = append(langs, models.RepoLanguage{
264 RepoDid: syntax.DID(repo.RepoDid),
265 Ref: ref.Short(),
266 IsDefaultRef: record.Meta.IsDefaultRef,
267 Language: l.Lang,
268 Bytes: l.Size,
269 })
270 }
271
272 tx, err := d.Begin()
273 if err != nil {
274 return err
275 }
276 defer tx.Rollback()
277
278 // update appview's cache
279 err = db.UpdateRepoLanguages(tx, syntax.DID(repo.RepoDid), ref.Short(), langs)
280 if err != nil {
281 fmt.Printf("failed; %s\n", err)
282 // non-fatal
283 }
284
285 return tx.Commit()
286}
287
288func ingestPipeline(d *db.DB, source ec.Source, msg ec.Message) error {
289 var record tangled.Pipeline
290 err := json.Unmarshal(msg.EventJson, &record)
291 if err != nil {
292 return err
293 }
294
295 if record.TriggerMetadata == nil {
296 return fmt.Errorf("empty trigger metadata: nsid %s, rkey %s", msg.Nsid, msg.Rkey)
297 }
298
299 if record.TriggerMetadata.Repo == nil {
300 return fmt.Errorf("empty repo: nsid %s, rkey %s", msg.Nsid, msg.Rkey)
301 }
302
303 repoName := ""
304 if record.TriggerMetadata.Repo.Repo != nil {
305 repoName = *record.TriggerMetadata.Repo.Repo
306 }
307
308 repo, lookupErr := resolveRepo(d, record.TriggerMetadata.Repo.RepoDid, record.TriggerMetadata.Repo.Did, repoName)
309 if lookupErr != nil {
310 return fmt.Errorf("failed to look up repo: %w", lookupErr)
311 }
312 if repo.Spindle == "" {
313 return fmt.Errorf("repo does not have a spindle configured yet: nsid %s, rkey %s", msg.Nsid, msg.Rkey)
314 }
315
316 // trigger info
317 var trigger models.Trigger
318 var sha string
319 trigger.Kind = workflow.TriggerKind(record.TriggerMetadata.Kind)
320 switch trigger.Kind {
321 case workflow.TriggerKindPush:
322 trigger.PushRef = &record.TriggerMetadata.Push.Ref
323 trigger.PushNewSha = &record.TriggerMetadata.Push.NewSha
324 trigger.PushOldSha = &record.TriggerMetadata.Push.OldSha
325 sha = *trigger.PushNewSha
326 case workflow.TriggerKindPullRequest:
327 trigger.PRSourceBranch = &record.TriggerMetadata.PullRequest.SourceBranch
328 trigger.PRTargetBranch = &record.TriggerMetadata.PullRequest.TargetBranch
329 trigger.PRSourceSha = &record.TriggerMetadata.PullRequest.SourceSha
330 trigger.PRAction = &record.TriggerMetadata.PullRequest.Action
331 sha = *trigger.PRSourceSha
332 }
333
334 tx, err := d.Begin()
335 if err != nil {
336 return fmt.Errorf("failed to start txn: %w", err)
337 }
338
339 triggerId, err := db.AddTrigger(tx, trigger)
340 if err != nil {
341 return fmt.Errorf("failed to add trigger entry: %w", err)
342 }
343
344 pipeline := models.Pipeline{
345 Rkey: msg.Rkey,
346 Knot: source.Key(),
347 RepoOwner: syntax.DID(record.TriggerMetadata.Repo.Did),
348 RepoName: repoName,
349 RepoDid: repo.RepoDid,
350 TriggerId: int(triggerId),
351 Sha: sha,
352 }
353
354 err = db.AddPipeline(tx, pipeline)
355 if err != nil {
356 return fmt.Errorf("failed to add pipeline: %w", err)
357 }
358
359 err = tx.Commit()
360 if err != nil {
361 return fmt.Errorf("failed to commit txn: %w", err)
362 }
363
364 return nil
365}
366
367func ingestDIDAssign(d *db.DB, enforcer *rbac.Enforcer, source ec.Source, msg ec.Message, ctx context.Context) error {
368 logger := log.FromContext(ctx)
369
370 var record knotdb.RepoDIDAssign
371 if err := json.Unmarshal(msg.EventJson, &record); err != nil {
372 return fmt.Errorf("unmarshal didAssign: %w", err)
373 }
374
375 if record.RepoDid == "" || record.OwnerDid == "" || record.RepoName == "" {
376 return fmt.Errorf("didAssign missing required fields: repoDid=%q ownerDid=%q repoName=%q",
377 record.RepoDid, record.OwnerDid, record.RepoName)
378 }
379
380 logger.Info("processing didAssign event",
381 "repo_did", record.RepoDid,
382 "owner_did", record.OwnerDid,
383 "repo_name", record.RepoName)
384
385 repos, err := db.GetRepos(d,
386 orm.FilterEq("did", record.OwnerDid),
387 orm.FilterEq("rkey", strings.ToLower(record.RepoName)),
388 )
389 if err != nil || len(repos) == 0 {
390 logger.Warn("didAssign for unknown repo, skipping",
391 "owner_did", record.OwnerDid,
392 "repo_name", record.RepoName)
393 return nil
394 }
395 repo := repos[0]
396 knot := source.Key()
397
398 if repo.Knot != knot {
399 return fmt.Errorf("didAssign from %s for repo hosted on %s, rejecting", knot, repo.Knot)
400 }
401
402 repoAtUri := repo.RepoAt().String()
403 legacyResource := record.OwnerDid + "/" + record.RepoName
404
405 if repo.RepoDid != record.RepoDid {
406 tx, err := d.Begin()
407 if err != nil {
408 return fmt.Errorf("begin didAssign txn: %w", err)
409 }
410 defer tx.Rollback()
411
412 if err := db.CascadeRepoDid(tx, repoAtUri, record.RepoDid); err != nil {
413 return fmt.Errorf("cascade repo_did: %w", err)
414 }
415
416 if err := db.EnqueuePdsRewritesForRepo(tx, record.RepoDid, repoAtUri); err != nil {
417 return fmt.Errorf("enqueue pds rewrites: %w", err)
418 }
419
420 if err := tx.Commit(); err != nil {
421 return fmt.Errorf("commit didAssign txn: %w", err)
422 }
423 }
424
425 if err := enforcer.RemoveRepo(record.OwnerDid, knot, legacyResource); err != nil {
426 return fmt.Errorf("remove legacy RBAC policies for %s: %w", legacyResource, err)
427 }
428 if err := enforcer.AddRepo(record.OwnerDid, knot, record.RepoDid); err != nil {
429 return fmt.Errorf("add RBAC policies for %s: %w", record.RepoDid, err)
430 }
431
432 collabs, collabErr := db.GetCollaborators(d, orm.FilterEq("repo_did", record.RepoDid))
433 if collabErr != nil {
434 return fmt.Errorf("get collaborators for RBAC update: %w", collabErr)
435 }
436 for _, c := range collabs {
437 collabDid := c.SubjectDid.String()
438 if err := enforcer.RemoveCollaborator(collabDid, knot, legacyResource); err != nil {
439 return fmt.Errorf("remove collaborator RBAC for %s: %w", collabDid, err)
440 }
441 if err := enforcer.AddCollaborator(collabDid, knot, record.RepoDid); err != nil {
442 return fmt.Errorf("add collaborator RBAC for %s: %w", collabDid, err)
443 }
444 }
445
446 if err := enforcer.E.SavePolicy(); err != nil {
447 return fmt.Errorf("save RBAC policies after didAssign: %w", err)
448 }
449
450 logger.Info("didAssign processed successfully",
451 "repo_did", record.RepoDid,
452 "owner_did", record.OwnerDid,
453 "repo_name", record.RepoName)
454
455 return nil
456}