appview, knotserver: add didAssign event and PDS rewrite on login #1143

+142 -2

appview/db/repos.go

··· 577 577 return GetRepo(e, orm.FilterEq("repo_did", repoDid)) 578 578 } 579 579 580 580 + func EnqueuePdsRewritesForRepo(tx *sql.Tx, repoDid, repoAtUri string) error { 581 581 + type record struct { 582 582 + userDidCol string 583 583 + table string 584 584 + nsid string 585 585 + fkCol string 586 586 + } 587 587 + sources := []record{ 588 588 + {"did", "repos", "sh.tangled.repo", "at_uri"}, 589 589 + {"did", "issues", "sh.tangled.repo.issue", "repo_at"}, 590 590 + {"owner_did", "pulls", "sh.tangled.repo.pull", "repo_at"}, 591 591 + {"did", "collaborators", "sh.tangled.repo.collaborator", "repo_at"}, 592 592 + {"did", "artifacts", "sh.tangled.repo.artifact", "repo_at"}, 593 593 + {"did", "stars", "sh.tangled.feed.star", "subject_at"}, 594 594 + } 595 595 + 596 596 + for _, src := range sources { 597 597 + rows, err := tx.Query( 598 598 + fmt.Sprintf(`SELECT %s, rkey FROM %s WHERE %s = ?`, src.userDidCol, src.table, src.fkCol), 599 599 + repoAtUri, 600 600 + ) 601 601 + if err != nil { 602 602 + return fmt.Errorf("query %s for pds rewrites: %w", src.table, err) 603 603 + } 604 604 + 605 605 + var pairs []struct{ did, rkey string } 606 606 + for rows.Next() { 607 607 + var d, r string 608 608 + if scanErr := rows.Scan(&d, &r); scanErr != nil { 609 609 + rows.Close() 610 610 + return fmt.Errorf("scan %s for pds rewrites: %w", src.table, scanErr) 611 611 + } 612 612 + pairs = append(pairs, struct{ did, rkey string }{d, r}) 613 613 + } 614 614 + rows.Close() 615 615 + if rowsErr := rows.Err(); rowsErr != nil { 616 616 + return fmt.Errorf("iterate %s for pds rewrites: %w", src.table, rowsErr) 617 617 + } 618 618 + 619 619 + for _, p := range pairs { 620 620 + if err := EnqueuePdsRewrite(tx, p.did, repoDid, src.nsid, p.rkey, repoAtUri); err != nil { 621 621 + return fmt.Errorf("enqueue pds rewrite for %s/%s: %w", src.table, p.rkey, err) 622 622 + } 623 623 + } 624 624 + } 625 625 + 626 626 + profileRows, err := tx.Query( 627 627 + `SELECT DISTINCT did FROM profile_pinned_repositories WHERE at_uri = ?`, 628 628 + repoAtUri, 629 629 + ) 630 630 + if err != nil { 631 631 + return fmt.Errorf("query profile_pinned_repositories for pds rewrites: %w", err) 632 632 + } 633 633 + var profileDids []string 634 634 + for profileRows.Next() { 635 635 + var d string 636 636 + if scanErr := profileRows.Scan(&d); scanErr != nil { 637 637 + profileRows.Close() 638 638 + return fmt.Errorf("scan profile_pinned_repositories for pds rewrites: %w", scanErr) 639 639 + } 640 640 + profileDids = append(profileDids, d) 641 641 + } 642 642 + profileRows.Close() 643 643 + if profileRowsErr := profileRows.Err(); profileRowsErr != nil { 644 644 + return fmt.Errorf("iterate profile_pinned_repositories for pds rewrites: %w", profileRowsErr) 645 645 + } 646 646 + 647 647 + for _, d := range profileDids { 648 648 + if err := EnqueuePdsRewrite(tx, d, repoDid, "sh.tangled.actor.profile", "self", repoAtUri); err != nil { 649 649 + return fmt.Errorf("enqueue pds rewrite for profile/%s: %w", d, err) 650 650 + } 651 651 + } 652 652 + 653 653 + return nil 654 654 + } 655 655 + 656 656 + type PdsRewrite struct { 657 657 + Id int 658 658 + RepoDid string 659 659 + RecordNsid string 660 660 + RecordRkey string 661 661 + OldRepoAt string 662 662 + } 663 663 + 664 664 + func GetPendingPdsRewrites(e Execer, userDid string) ([]PdsRewrite, error) { 665 665 + rows, err := e.Query( 666 666 + `SELECT id, repo_did, record_nsid, record_rkey, old_repo_at 667 667 + FROM pds_rewrite_status 668 668 + WHERE user_did = ? AND status = 'pending'`, 669 669 + userDid, 670 670 + ) 671 671 + if err != nil { 672 672 + return nil, err 673 673 + } 674 674 + defer rows.Close() 675 675 + 676 676 + var rewrites []PdsRewrite 677 677 + for rows.Next() { 678 678 + var r PdsRewrite 679 679 + if err := rows.Scan(&r.Id, &r.RepoDid, &r.RecordNsid, &r.RecordRkey, &r.OldRepoAt); err != nil { 680 680 + return nil, err 681 681 + } 682 682 + rewrites = append(rewrites, r) 683 683 + } 684 684 + return rewrites, rows.Err() 685 685 + } 686 686 + 687 687 + func CompletePdsRewrite(e Execer, id int) error { 688 688 + _, err := e.Exec( 689 689 + `UPDATE pds_rewrite_status SET status = 'done', updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?`, 690 690 + id, 691 691 + ) 692 692 + return err 693 693 + } 694 694 + 580 695 func EnqueuePdsRewrite(e Execer, userDid, repoDid, recordNsid, recordRkey, oldRepoAt string) error { 581 696 _, err := e.Exec( 582 582 - `INSERT OR IGNORE INTO pds_rewrite_status 697 697 + `INSERT INTO pds_rewrite_status 583 698 (user_did, repo_did, record_nsid, record_rkey, old_repo_at, status) 584 584 - VALUES (?, ?, ?, ?, ?, 'pending')`, 699 699 + VALUES (?, ?, ?, ?, ?, 'pending') 700 700 + ON CONFLICT(user_did, record_nsid, record_rkey) DO UPDATE SET 701 701 + status = 'pending', 702 702 + repo_did = excluded.repo_did, 703 703 + old_repo_at = excluded.old_repo_at, 704 704 + updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')`, 585 705 userDid, repoDid, recordNsid, recordRkey, oldRepoAt, 586 706 ) 587 707 return err 588 708 } 589 709 710 710 + func CascadeRepoDid(tx *sql.Tx, repoAtUri, repoDid string) error { 711 711 + _, err := tx.Exec( 712 712 + `UPDATE repos SET repo_did = ? WHERE at_uri = ?`, 713 713 + repoDid, repoAtUri, 714 714 + ) 715 715 + if err != nil { 716 716 + return fmt.Errorf("cascade repo_did to repos: %w", err) 717 717 + } 718 718 + 719 719 + _, err = tx.Exec( 720 720 + `UPDATE repos SET source = ? WHERE source = ?`, 721 721 + repoDid, repoAtUri, 722 722 + ) 723 723 + if err != nil { 724 724 + return fmt.Errorf("cascade repo_did to repos.source: %w", err) 725 725 + } 726 726 + 727 727 + return nil 728 728 + } 729 729 + 590 730 func UpdateDescription(e Execer, repoAt, newDescription string) error { 591 731 _, err := e.Exec( 592 732 `update repos set description = ? where at_uri = ?`, newDescription, repoAt)

+134

appview/oauth/handler.go

··· 14 14 15 15 comatproto "github.com/bluesky-social/indigo/api/atproto" 16 16 "github.com/bluesky-social/indigo/atproto/auth/oauth" 17 17 + atpclient "github.com/bluesky-social/indigo/atproto/client" 17 18 lexutil "github.com/bluesky-social/indigo/lex/util" 18 19 xrpc "github.com/bluesky-social/indigo/xrpc" 19 20 "github.com/go-chi/chi/v5" ··· 95 96 go o.addToDefaultSpindle(sessData.AccountDID.String()) 96 97 go o.ensureTangledProfile(sessData) 97 98 go o.autoClaimTnglShDomain(sessData.AccountDID.String()) 99 99 + go o.drainPdsRewrites(sessData) 98 100 99 101 if !o.Config.Core.Dev { 100 102 err = o.Posthog.Enqueue(posthog.Capture{ ··· 273 275 l.Debug("successfully created empty Tangled profile on PDS and DB") 274 276 } 275 277 278 278 + func (o *OAuth) drainPdsRewrites(sessData *oauth.ClientSessionData) { 279 279 + ctx := context.Background() 280 280 + did := sessData.AccountDID.String() 281 281 + l := o.Logger.With("did", did, "handler", "drainPdsRewrites") 282 282 + 283 283 + rewrites, err := db.GetPendingPdsRewrites(o.Db, did) 284 284 + if err != nil { 285 285 + l.Error("failed to get pending rewrites", "err", err) 286 286 + return 287 287 + } 288 288 + if len(rewrites) == 0 { 289 289 + return 290 290 + } 291 291 + 292 292 + l.Info("draining pending PDS rewrites", "count", len(rewrites)) 293 293 + 294 294 + sess, err := o.ClientApp.ResumeSession(ctx, sessData.AccountDID, sessData.SessionID) 295 295 + if err != nil { 296 296 + l.Error("failed to resume session for PDS rewrites", "err", err) 297 297 + return 298 298 + } 299 299 + client := sess.APIClient() 300 300 + 301 301 + for _, rw := range rewrites { 302 302 + if err := o.rewritePdsRecord(ctx, client, did, rw); err != nil { 303 303 + l.Error("failed to rewrite PDS record", 304 304 + "nsid", rw.RecordNsid, 305 305 + "rkey", rw.RecordRkey, 306 306 + "repo_did", rw.RepoDid, 307 307 + "err", err) 308 308 + continue 309 309 + } 310 310 + 311 311 + if err := db.CompletePdsRewrite(o.Db, rw.Id); err != nil { 312 312 + l.Error("failed to mark rewrite complete", "id", rw.Id, "err", err) 313 313 + } 314 314 + } 315 315 + } 316 316 + 317 317 + func (o *OAuth) rewritePdsRecord(ctx context.Context, client *atpclient.APIClient, userDid string, rw db.PdsRewrite) error { 318 318 + ex, err := comatproto.RepoGetRecord(ctx, client, "", rw.RecordNsid, userDid, rw.RecordRkey) 319 319 + if err != nil { 320 320 + return fmt.Errorf("get record: %w", err) 321 321 + } 322 322 + 323 323 + val := ex.Value.Val 324 324 + repoDid := rw.RepoDid 325 325 + 326 326 + switch rw.RecordNsid { 327 327 + case tangled.RepoNSID: 328 328 + rec, ok := val.(*tangled.Repo) 329 329 + if !ok { 330 330 + return fmt.Errorf("unexpected type for repo record") 331 331 + } 332 332 + rec.RepoDid = &repoDid 333 333 + 334 334 + case tangled.RepoIssueNSID: 335 335 + rec, ok := val.(*tangled.RepoIssue) 336 336 + if !ok { 337 337 + return fmt.Errorf("unexpected type for issue record") 338 338 + } 339 339 + rec.RepoDid = &repoDid 340 340 + 341 341 + case tangled.RepoPullNSID: 342 342 + rec, ok := val.(*tangled.RepoPull) 343 343 + if !ok { 344 344 + return fmt.Errorf("unexpected type for pull record") 345 345 + } 346 346 + if rec.Target != nil { 347 347 + rec.Target.RepoDid = &repoDid 348 348 + } 349 349 + if rec.Source != nil && rec.Source.Repo != nil && *rec.Source.Repo == rw.OldRepoAt { 350 350 + rec.Source.RepoDid = &repoDid 351 351 + } 352 352 + 353 353 + case tangled.RepoCollaboratorNSID: 354 354 + rec, ok := val.(*tangled.RepoCollaborator) 355 355 + if !ok { 356 356 + return fmt.Errorf("unexpected type for collaborator record") 357 357 + } 358 358 + rec.RepoDid = &repoDid 359 359 + 360 360 + case tangled.RepoArtifactNSID: 361 361 + rec, ok := val.(*tangled.RepoArtifact) 362 362 + if !ok { 363 363 + return fmt.Errorf("unexpected type for artifact record") 364 364 + } 365 365 + rec.RepoDid = &repoDid 366 366 + 367 367 + case tangled.FeedStarNSID: 368 368 + rec, ok := val.(*tangled.FeedStar) 369 369 + if !ok { 370 370 + return fmt.Errorf("unexpected type for star record") 371 371 + } 372 372 + rec.SubjectDid = &repoDid 373 373 + 374 374 + case tangled.ActorProfileNSID: 375 375 + rec, ok := val.(*tangled.ActorProfile) 376 376 + if !ok { 377 377 + return fmt.Errorf("unexpected type for profile record") 378 378 + } 379 379 + var dids []string 380 380 + var remaining []string 381 381 + for _, pinUri := range rec.PinnedRepositories { 382 382 + repo, repoErr := db.GetRepoByAtUri(o.Db, pinUri) 383 383 + if repoErr != nil || repo.RepoDid == "" { 384 384 + remaining = append(remaining, pinUri) 385 385 + continue 386 386 + } 387 387 + dids = append(dids, repo.RepoDid) 388 388 + } 389 389 + rec.PinnedRepositoryDids = append(rec.PinnedRepositoryDids, dids...) 390 390 + rec.PinnedRepositories = remaining 391 391 + 392 392 + default: 393 393 + return fmt.Errorf("unsupported NSID for PDS rewrite: %s", rw.RecordNsid) 394 394 + } 395 395 + 396 396 + _, err = comatproto.RepoPutRecord(ctx, client, &comatproto.RepoPutRecord_Input{ 397 397 + Collection: rw.RecordNsid, 398 398 + Repo: userDid, 399 399 + Rkey: rw.RecordRkey, 400 400 + SwapRecord: ex.Cid, 401 401 + Record: &lexutil.LexiconTypeDecoder{Val: val}, 402 402 + }) 403 403 + if err != nil { 404 404 + return fmt.Errorf("put record: %w", err) 405 405 + } 406 406 + 407 407 + return nil 408 408 + } 409 409 + 276 410 // create a AppPasswordSession using apppasswords 277 411 type AppPasswordSession struct { 278 412 AccessJwt string `json:"accessJwt"`

+90

appview/state/knotstream.go

··· 20 20 "tangled.org/core/appview/sites" 21 21 ec "tangled.org/core/eventconsumer" 22 22 "tangled.org/core/eventconsumer/cursor" 23 23 + knotdb "tangled.org/core/knotserver/db" 23 24 "tangled.org/core/log" 24 25 "tangled.org/core/orm" 25 26 "tangled.org/core/rbac" ··· 88 89 return ingestRefUpdate(ctx, d, enforcer, posthog, notifier, dev, c, cfClient, source, msg) 89 90 case tangled.PipelineNSID: 90 91 return ingestPipeline(d, source, msg) 92 92 + case knotdb.RepoDIDAssignNSID: 93 93 + return ingestDIDAssign(d, enforcer, source, msg, ctx) 91 94 } 92 95 93 96 return nil ··· 375 378 376 379 return nil 377 380 } 381 381 + 382 382 + func ingestDIDAssign(d *db.DB, enforcer *rbac.Enforcer, source ec.Source, msg ec.Message, ctx context.Context) error { 383 383 + logger := log.FromContext(ctx) 384 384 + 385 385 + var record knotdb.RepoDIDAssign 386 386 + if err := json.Unmarshal(msg.EventJson, &record); err != nil { 387 387 + return fmt.Errorf("unmarshal didAssign: %w", err) 388 388 + } 389 389 + 390 390 + if record.RepoDid == "" || record.OwnerDid == "" || record.RepoName == "" { 391 391 + return fmt.Errorf("didAssign missing required fields: repoDid=%q ownerDid=%q repoName=%q", 392 392 + record.RepoDid, record.OwnerDid, record.RepoName) 393 393 + } 394 394 + 395 395 + logger.Info("processing didAssign event", 396 396 + "repo_did", record.RepoDid, 397 397 + "owner_did", record.OwnerDid, 398 398 + "repo_name", record.RepoName) 399 399 + 400 400 + repos, err := db.GetRepos(d, 1, 401 401 + orm.FilterEq("did", record.OwnerDid), 402 402 + orm.FilterEq("name", record.RepoName), 403 403 + ) 404 404 + if err != nil || len(repos) == 0 { 405 405 + logger.Warn("didAssign for unknown repo, skipping", 406 406 + "owner_did", record.OwnerDid, 407 407 + "repo_name", record.RepoName) 408 408 + return nil 409 409 + } 410 410 + repo := repos[0] 411 411 + knot := source.Key() 412 412 + 413 413 + if repo.Knot != knot { 414 414 + return fmt.Errorf("didAssign from %s for repo hosted on %s, rejecting", knot, repo.Knot) 415 415 + } 416 416 + 417 417 + repoAtUri := repo.RepoAt().String() 418 418 + legacyResource := record.OwnerDid + "/" + record.RepoName 419 419 + 420 420 + if repo.RepoDid != record.RepoDid { 421 421 + tx, err := d.Begin() 422 422 + if err != nil { 423 423 + return fmt.Errorf("begin didAssign txn: %w", err) 424 424 + } 425 425 + defer tx.Rollback() 426 426 + 427 427 + if err := db.CascadeRepoDid(tx, repoAtUri, record.RepoDid); err != nil { 428 428 + return fmt.Errorf("cascade repo_did: %w", err) 429 429 + } 430 430 + 431 431 + if err := db.EnqueuePdsRewritesForRepo(tx, record.RepoDid, repoAtUri); err != nil { 432 432 + return fmt.Errorf("enqueue pds rewrites: %w", err) 433 433 + } 434 434 + 435 435 + if err := tx.Commit(); err != nil { 436 436 + return fmt.Errorf("commit didAssign txn: %w", err) 437 437 + } 438 438 + } 439 439 + 440 440 + if err := enforcer.RemoveRepo(record.OwnerDid, knot, legacyResource); err != nil { 441 441 + return fmt.Errorf("remove legacy RBAC policies for %s: %w", legacyResource, err) 442 442 + } 443 443 + if err := enforcer.AddRepo(record.OwnerDid, knot, record.RepoDid); err != nil { 444 444 + return fmt.Errorf("add RBAC policies for %s: %w", record.RepoDid, err) 445 445 + } 446 446 + 447 447 + collabs, collabErr := db.GetCollaborators(d, orm.FilterEq("repo_at", repoAtUri)) 448 448 + if collabErr != nil { 449 449 + return fmt.Errorf("get collaborators for RBAC update: %w", collabErr) 450 450 + } 451 451 + for _, c := range collabs { 452 452 + collabDid := c.SubjectDid.String() 453 453 + if err := enforcer.RemoveCollaborator(collabDid, knot, legacyResource); err != nil { 454 454 + return fmt.Errorf("remove collaborator RBAC for %s: %w", collabDid, err) 455 455 + } 456 456 + if err := enforcer.AddCollaborator(collabDid, knot, record.RepoDid); err != nil { 457 457 + return fmt.Errorf("add collaborator RBAC for %s: %w", collabDid, err) 458 458 + } 459 459 + } 460 460 + 461 461 + logger.Info("didAssign processed successfully", 462 462 + "repo_did", record.RepoDid, 463 463 + "owner_did", record.OwnerDid, 464 464 + "repo_name", record.RepoName) 465 465 + 466 466 + return nil 467 467 + }

+10

knotserver/db/didassign.go

··· 1 1 + package db 2 2 + 3 3 + const RepoDIDAssignNSID = "sh.tangled.repo.didAssign" 4 4 + 5 5 + type RepoDIDAssign struct { 6 6 + OwnerDid string `json:"ownerDid"` 7 7 + RepoName string `json:"repoName"` 8 8 + RepoDid string `json:"repoDid"` 9 9 + OldRepoAt string `json:"oldRepoAt,omitempty"` 10 10 + }

+22

knotserver/db/events.go

··· 1 1 package db 2 2 3 3 import ( 4 4 + "encoding/json" 4 5 "fmt" 5 6 "time" 6 7 7 8 "tangled.org/core/notifier" 9 9 + "tangled.org/core/tid" 8 10 ) 9 11 10 12 type Event struct { ··· 29 31 return err 30 32 } 31 33 34 34 + func (d *DB) EmitDIDAssign(n *notifier.Notifier, ownerDid, repoName, repoDid, oldRepoAt string) error { 35 35 + payload := RepoDIDAssign{ 36 36 + OwnerDid: ownerDid, 37 37 + RepoName: repoName, 38 38 + RepoDid: repoDid, 39 39 + OldRepoAt: oldRepoAt, 40 40 + } 41 41 + 42 42 + eventJson, err := json.Marshal(payload) 43 43 + if err != nil { 44 44 + return fmt.Errorf("marshal didAssign event: %w", err) 45 45 + } 46 46 + 47 47 + return d.InsertEvent(Event{ 48 48 + Rkey: tid.TID(), 49 49 + Nsid: RepoDIDAssignNSID, 50 50 + EventJson: string(eventJson), 51 51 + }, n) 52 52 + } 53 53 + 32 54 func (d *DB) GetEvents(cursor int64) ([]Event, error) { 33 55 whereClause := "" 34 56 args := []any{}

+26 -14

knotserver/events.go

··· 54 54 cursor = defaultCursor 55 55 } 56 56 57 57 - // complete backfill first before going to live data 58 57 l.Debug("going through backfill", "cursor", cursor) 59 59 - if err := h.streamOps(conn, &cursor); err != nil { 58 58 + if err := h.drainBackfill(conn, &cursor, 10_000); err != nil { 60 59 l.Error("failed to backfill", "err", err) 61 60 return 62 61 } ··· 68 67 l.Debug("stopping stream: client closed connection") 69 68 return 70 69 case <-ch: 71 71 - // we have been notified of new data 72 70 l.Debug("going through live data", "cursor", cursor) 73 73 - if err := h.streamOps(conn, &cursor); err != nil { 71 71 + if _, err := h.streamOps(conn, &cursor); err != nil { 74 72 l.Error("failed to stream", "err", err) 75 73 return 76 74 } ··· 83 81 } 84 82 } 85 83 86 86 - func (h *Knot) streamOps(conn *websocket.Conn, cursor *int64) error { 84 84 + func (h *Knot) drainBackfill(conn *websocket.Conn, cursor *int64, maxBatches int) error { 85 85 + for range maxBatches { 86 86 + n, err := h.streamOps(conn, cursor) 87 87 + if err != nil { 88 88 + return err 89 89 + } 90 90 + if n < 100 { 91 91 + return nil 92 92 + } 93 93 + } 94 94 + h.l.Warn("backfill hit batch limit", "maxBatches", maxBatches, "cursor", *cursor) 95 95 + return nil 96 96 + } 97 97 + 98 98 + func (h *Knot) streamOps(conn *websocket.Conn, cursor *int64) (int, error) { 87 99 events, err := h.db.GetEvents(*cursor) 88 100 if err != nil { 89 101 h.l.Error("failed to fetch events from db", "err", err, "cursor", cursor) 90 90 - return err 102 102 + return 0, err 91 103 } 92 104 93 105 for _, event := range events { 94 94 - // first extract the inner json into a map 95 106 var eventJson map[string]any 96 107 err := json.Unmarshal([]byte(event.EventJson), &eventJson) 97 108 if err != nil { 98 109 h.l.Error("failed to unmarshal event", "err", err) 99 99 - return err 110 110 + return 0, err 100 111 } 101 112 102 113 jsonMsg, err := json.Marshal(map[string]any{ 103 103 - "rkey": event.Rkey, 104 104 - "nsid": event.Nsid, 105 105 - "event": eventJson, 114 114 + "rkey": event.Rkey, 115 115 + "nsid": event.Nsid, 116 116 + "event": eventJson, 117 117 + "created": event.Created, 106 118 }) 107 119 if err != nil { 108 120 h.l.Error("failed to marshal record", "err", err) 109 109 - return err 121 121 + return 0, err 110 122 } 111 123 112 124 if err := conn.WriteMessage(websocket.TextMessage, jsonMsg); err != nil { 113 125 h.l.Debug("err", "err", err) 114 114 - return err 126 126 + return 0, err 115 127 } 116 128 *cursor = event.Created 117 129 } 118 130 119 119 - return nil 131 131 + return len(events), nil 120 132 }

+14 -5

knotserver/git.go

··· 6 6 "io" 7 7 "net/http" 8 8 "os" 9 9 + "path/filepath" 9 10 "strings" 10 11 12 12 + securejoin "github.com/cyphar/filepath-securejoin" 11 13 "github.com/go-chi/chi/v5" 12 14 "tangled.org/core/knotserver/git/service" 13 15 ) ··· 25 27 } 26 28 27 29 repoDid, err := h.db.GetRepoDid(did, name) 28 28 - if err != nil { 29 29 - return "", "", fmt.Errorf("repo not found: %w", err) 30 30 + if err == nil { 31 31 + repoPath, _, _, resolveErr := h.db.ResolveRepoDIDOnDisk(h.c.Repo.ScanPath, repoDid) 32 32 + if resolveErr == nil { 33 33 + return repoPath, name, nil 34 34 + } 30 35 } 31 31 - repoPath, _, _, err := h.db.ResolveRepoDIDOnDisk(h.c.Repo.ScanPath, repoDid) 32 32 - if err != nil { 33 33 - return "", "", fmt.Errorf("repo not found: %w", err) 36 36 + 37 37 + repoPath, joinErr := securejoin.SecureJoin(h.c.Repo.ScanPath, filepath.Join(did, name)) 38 38 + if joinErr != nil { 39 39 + return "", "", fmt.Errorf("repo not found: %w", joinErr) 40 40 + } 41 41 + if _, statErr := os.Stat(repoPath); statErr != nil { 42 42 + return "", "", fmt.Errorf("repo not found: %w", statErr) 34 43 } 35 44 return repoPath, name, nil 36 45 }

+28 -13

knotserver/internal.go

··· 6 6 "fmt" 7 7 "log/slog" 8 8 "net/http" 9 9 + "os" 9 10 "path/filepath" 10 11 "strings" 11 12 13 13 + securejoin "github.com/cyphar/filepath-securejoin" 12 14 "github.com/go-chi/chi/v5" 13 15 "github.com/go-chi/chi/v5/middleware" 14 16 "github.com/go-git/go-git/v5/plumbing" ··· 124 126 ownerDid := repoOwnerIdent.DID.String() 125 127 repoName := components[1] 126 128 repoDid, didErr := h.db.GetRepoDid(ownerDid, repoName) 127 127 - if didErr != nil { 128 128 - w.WriteHeader(http.StatusNotFound) 129 129 - l.Error("repo DID not found", "owner", ownerDid, "name", repoName, "err", didErr) 130 130 - fmt.Fprintln(w, "repo not found") 131 131 - return 132 132 - } 133 133 - repoPath, _, _, lookupErr := h.db.ResolveRepoDIDOnDisk(h.c.Repo.ScanPath, repoDid) 134 134 - if lookupErr != nil { 135 135 - w.WriteHeader(http.StatusNotFound) 136 136 - l.Error("repo not found on disk", "repoDid", repoDid, "err", lookupErr) 137 137 - fmt.Fprintln(w, "repo not found") 138 138 - return 129 129 + var repoPath string 130 130 + if didErr == nil { 131 131 + var lookupErr error 132 132 + repoPath, _, _, lookupErr = h.db.ResolveRepoDIDOnDisk(h.c.Repo.ScanPath, repoDid) 133 133 + if lookupErr != nil { 134 134 + w.WriteHeader(http.StatusNotFound) 135 135 + l.Error("repo not found on disk", "repoDid", repoDid, "err", lookupErr) 136 136 + fmt.Fprintln(w, "repo not found") 137 137 + return 138 138 + } 139 139 + rbacResource = repoDid 140 140 + } else { 141 141 + legacyPath, joinErr := securejoin.SecureJoin(h.c.Repo.ScanPath, filepath.Join(ownerDid, repoName)) 142 142 + if joinErr != nil { 143 143 + w.WriteHeader(http.StatusNotFound) 144 144 + fmt.Fprintln(w, "repo not found") 145 145 + return 146 146 + } 147 147 + if _, statErr := os.Stat(legacyPath); statErr != nil { 148 148 + w.WriteHeader(http.StatusNotFound) 149 149 + l.Error("repo not found on disk (legacy)", "owner", ownerDid, "name", repoName) 150 150 + fmt.Fprintln(w, "repo not found") 151 151 + return 152 152 + } 153 153 + repoPath = legacyPath 154 154 + rbacResource = ownerDid + "/" + repoName 139 155 } 140 140 - rbacResource = repoDid 141 156 rel, relErr := filepath.Rel(h.c.Repo.ScanPath, repoPath) 142 157 if relErr != nil { 143 158 w.WriteHeader(http.StatusInternalServerError)

+2 -2

rbac/rbac.go

··· 34 34 ) 35 35 36 36 type Enforcer struct { 37 37 - E *casbin.Enforcer 37 37 + E *casbin.SyncedEnforcer 38 38 } 39 39 40 40 func NewEnforcer(path string) (*Enforcer, error) { ··· 53 53 return nil, err 54 54 } 55 55 56 56 - e, err := casbin.NewEnforcer(m, a) 56 56 + e, err := casbin.NewSyncedEnforcer(m, a) 57 57 if err != nil { 58 58 return nil, err 59 59 }

+1 -1

rbac/rbac_test.go

··· 23 23 m, err := model.NewModelFromString(rbac.Model) 24 24 assert.NoError(t, err) 25 25 26 26 - e, err := casbin.NewEnforcer(m, a) 26 26 + e, err := casbin.NewSyncedEnforcer(m, a) 27 27 assert.NoError(t, err) 28 28 29 29 e.EnableAutoSave(false)