usaa.ma / core
Monorepo for Tangled
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

appview: DID-based routing, state/handler/middleware updates

Signed-off-by: Lewis <lewis@tangled.org>
Lewis: May this revision serve well! <lewis@tangled.org>

authored by

Lewis and committed by tangled.org (Apr 2, 2026, 9:16 AM UTC) b1d5ce3c 118dda16

+469 -251
+74 -18
appview/ingester.go
··· 2 2 3 3 import ( 4 4 "context" 5 + "database/sql" 5 6 "encoding/json" 7 + "errors" 6 8 "fmt" 7 9 "log/slog" 8 10 "maps" ··· 116 118 return err 117 119 } 118 120 119 - subjectUri, err = syntax.ParseATURI(record.Subject) 120 - if err != nil { 121 - l.Error("invalid record", "err", err) 122 - return err 121 + star := &models.Star{ 122 + Did: did, 123 + Rkey: e.Commit.RKey, 123 124 } 124 - err = db.AddStar(i.Db, &models.Star{ 125 - Did: did, 126 - RepoAt: subjectUri, 127 - Rkey: e.Commit.RKey, 128 - }) 125 + 126 + switch { 127 + case record.SubjectDid != nil: 128 + repo, repoErr := db.GetRepo(i.Db, orm.FilterEq("repo_did", *record.SubjectDid)) 129 + if repoErr == nil { 130 + subjectUri = repo.RepoAt() 131 + star.RepoAt = subjectUri 132 + } 133 + case record.Subject != nil: 134 + subjectUri, err = syntax.ParseATURI(*record.Subject) 135 + if err != nil { 136 + l.Error("invalid record", "err", err) 137 + return err 138 + } 139 + star.RepoAt = subjectUri 140 + repo, repoErr := db.GetRepoByAtUri(i.Db, subjectUri.String()) 141 + if repoErr == nil && repo.RepoDid != "" { 142 + if enqErr := db.EnqueuePdsRewrite(i.Db, did, repo.RepoDid, tangled.FeedStarNSID, e.Commit.RKey, *record.Subject); enqErr != nil { 143 + l.Warn("failed to enqueue PDS rewrite for star", "err", enqErr, "did", did, "repoDid", repo.RepoDid) 144 + } 145 + } 146 + default: 147 + l.Error("star record has neither subject nor subjectDid") 148 + return fmt.Errorf("star record has neither subject nor subjectDid") 149 + } 150 + err = db.AddStar(i.Db, star) 129 151 case jmodels.CommitOperationDelete: 130 152 err = db.DeleteStarByRkey(i.Db, did, e.Commit.RKey) 131 153 } ··· 220 242 return err 221 243 } 222 244 223 - repoAt, err := syntax.ParseATURI(record.Repo) 224 - if err != nil { 225 - return err 245 + var repo *models.Repo 246 + if record.RepoDid != nil && *record.RepoDid != "" { 247 + repo, err = db.GetRepoByDid(i.Db, *record.RepoDid) 248 + if err != nil && !errors.Is(err, sql.ErrNoRows) { 249 + return fmt.Errorf("failed to look up repo by DID %s: %w", *record.RepoDid, err) 250 + } 251 + } 252 + if repo == nil && record.Repo != nil { 253 + repoAt, parseErr := syntax.ParseATURI(*record.Repo) 254 + if parseErr != nil { 255 + return parseErr 256 + } 257 + repo, err = db.GetRepoByAtUri(i.Db, repoAt.String()) 258 + if err != nil { 259 + return err 260 + } 261 + } 262 + if repo == nil { 263 + return fmt.Errorf("artifact record has neither valid repoDid nor repo field") 226 264 } 227 265 228 - repo, err := db.GetRepoByAtUri(i.Db, repoAt.String()) 229 - if err != nil { 266 + ok, err := i.Enforcer.E.Enforce(did, repo.Knot, repo.RepoIdentifier(), "repo:push") 267 + if err != nil || !ok { 230 268 return err 231 269 } 232 270 233 - ok, err := i.Enforcer.E.Enforce(did, repo.Knot, repo.DidSlashRepo(), "repo:push") 234 - if err != nil || !ok { 235 - return err 271 + repoDid := repo.RepoDid 272 + if repoDid == "" && record.RepoDid != nil { 273 + repoDid = *record.RepoDid 274 + } 275 + if repoDid != "" && (record.RepoDid == nil || *record.RepoDid == "") && record.Repo != nil { 276 + if enqErr := db.EnqueuePdsRewrite(i.Db, did, repoDid, tangled.RepoArtifactNSID, e.Commit.RKey, *record.Repo); enqErr != nil { 277 + l.Warn("failed to enqueue PDS rewrite for artifact", "err", enqErr, "did", did, "repoDid", repoDid) 278 + } 236 279 } 237 280 238 281 createdAt, err := time.Parse(time.RFC3339, record.CreatedAt) ··· 243 286 artifact := models.Artifact{ 244 287 Did: did, 245 288 Rkey: e.Commit.RKey, 246 - RepoAt: repoAt, 289 + RepoAt: repo.RepoAt(), 247 290 Tag: plumbing.Hash(record.Tag), 248 291 CreatedAt: createdAt, 249 292 BlobCid: cid.Cid(record.Artifact.Ref), ··· 834 877 835 878 issue := models.IssueFromRecord(did, rkey, record) 836 879 880 + if issue.RepoAt == "" { 881 + return fmt.Errorf("issue record has no repo field") 882 + } 883 + 837 884 if err := i.Validator.ValidateIssue(&issue); err != nil { 838 885 return fmt.Errorf("failed to validate issue: %w", err) 886 + } 887 + 888 + if record.Repo != nil { 889 + repo, repoErr := db.GetRepoByAtUri(i.Db, *record.Repo) 890 + if repoErr == nil && repo.RepoDid != "" { 891 + if enqErr := db.EnqueuePdsRewrite(i.Db, did, repo.RepoDid, tangled.RepoIssueNSID, rkey, *record.Repo); enqErr != nil { 892 + l.Warn("failed to enqueue PDS rewrite for issue", "err", enqErr, "did", did, "repoDid", repo.RepoDid) 893 + } 894 + } 839 895 } 840 896 841 897 tx, err := ddb.BeginTx(ctx, nil)
+2 -2
appview/issues/issues.go
··· 309 309 return 310 310 } 311 311 312 - roles := repoinfo.RolesInRepo{Roles: rp.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.DidSlashRepo())} 312 + roles := repoinfo.RolesInRepo{Roles: rp.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.RepoIdentifier())} 313 313 isRepoOwner := roles.IsOwner() 314 314 isCollaborator := roles.IsCollaborator() 315 315 isIssueOwner := user.Active.Did == issue.Did ··· 357 357 return 358 358 } 359 359 360 - roles := repoinfo.RolesInRepo{Roles: rp.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.DidSlashRepo())} 360 + roles := repoinfo.RolesInRepo{Roles: rp.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.RepoIdentifier())} 361 361 isRepoOwner := roles.IsOwner() 362 362 isCollaborator := roles.IsCollaborator() 363 363 isIssueOwner := user.Active.Did == issue.Did
+9 -5
appview/middleware/middleware.go
··· 18 18 "tangled.org/core/appview/pages" 19 19 "tangled.org/core/appview/pagination" 20 20 "tangled.org/core/appview/reporesolver" 21 + "tangled.org/core/appview/state/userutil" 21 22 "tangled.org/core/idresolver" 22 23 "tangled.org/core/orm" 23 24 "tangled.org/core/rbac" ··· 162 163 return 163 164 } 164 165 165 - ok, err := mw.enforcer.E.Enforce(actor.Active.Did, f.Knot, f.DidSlashRepo(), requiredPerm) 166 + ok, err := mw.enforcer.E.Enforce(actor.Active.Did, f.Knot, f.RepoIdentifier(), requiredPerm) 166 167 if err != nil || !ok { 167 - log.Printf("%s does not have perms of a %s in repo %s", actor.Active.Did, requiredPerm, f.DidSlashRepo()) 168 + log.Printf("%s does not have perms of a %s in repo %s", actor.Active.Did, requiredPerm, f.RepoIdentifier()) 168 169 http.Error(w, "Forbiden", http.StatusUnauthorized) 169 170 return 170 171 } ··· 195 196 } 196 197 } 197 198 } 198 - // invalid did or handle 199 199 if err != nil { 200 200 log.Printf("failed to resolve did/handle '%s': %s\n", didOrHandle, err) 201 201 mw.pages.Error404(w) ··· 342 342 343 343 if r.Header.Get("User-Agent") == "Go-http-client/1.1" { 344 344 if r.URL.Query().Get("go-get") == "1" { 345 + modulePath := userutil.FlattenDid(fullName) 346 + if strings.Contains(modulePath, ":") { 347 + modulePath = userutil.FlattenDid(f.Did) + "/" + f.Name 348 + } 345 349 html := fmt.Sprintf( 346 350 `<meta name="go-import" content="tangled.sh/%s git https://tangled.sh/%s"/> 347 351 <meta name="go-import" content="tangled.org/%s git https://tangled.org/%s"/>`, 348 - fullName, fullName, 349 - fullName, fullName, 352 + modulePath, fullName, 353 + modulePath, fullName, 350 354 ) 351 355 w.Header().Set("Content-Type", "text/html") 352 356 w.Write([]byte(html))
+5 -1
appview/models/issue.go
··· 45 45 references[i] = string(uri) 46 46 } 47 47 repoAtStr := i.RepoAt.String() 48 - return tangled.RepoIssue{ 48 + rec := tangled.RepoIssue{ 49 49 Repo: &repoAtStr, 50 50 Title: i.Title, 51 51 Body: &i.Body, ··· 53 53 References: references, 54 54 CreatedAt: i.Created.Format(time.RFC3339), 55 55 } 56 + if i.Repo != nil && i.Repo.RepoDid != "" { 57 + rec.RepoDid = &i.Repo.RepoDid 58 + } 59 + return rec 56 60 } 57 61 58 62 func (i *Issue) State() string {
+26 -20
appview/pulls/pulls.go
··· 408 408 } 409 409 410 410 // user can only delete branch if they are a collaborator in the repo that the branch belongs to 411 - perms := s.enforcer.GetPermissionsInRepo(user.Active.Did, repo.Knot, repo.DidSlashRepo()) 411 + perms := s.enforcer.GetPermissionsInRepo(user.Active.Did, repo.Knot, repo.RepoIdentifier()) 412 412 if !slices.Contains(perms, "repo:push") { 413 413 return nil 414 414 } ··· 432 432 433 433 var sourceRepo syntax.ATURI 434 434 if pull.PullSource.RepoAt != nil { 435 - // fork-based pulls 436 435 sourceRepo = *pull.PullSource.RepoAt 437 436 } else { 438 - // pulls within the same repo 439 437 sourceRepo = repo.RepoAt() 440 438 } 441 439 ··· 929 927 } 930 928 931 929 // Determine PR type based on input parameters 932 - roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.DidSlashRepo())} 930 + roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.RepoIdentifier())} 933 931 isPushAllowed := roles.IsPushAllowed() 934 932 isBranchBased := isPushAllowed && sourceBranch != "" && fromFork == "" 935 933 isForkBased := fromFork != "" && sourceBranch != "" ··· 1045 1043 Host: host, 1046 1044 } 1047 1045 1048 - didSlashRepo := fmt.Sprintf("%s/%s", repo.Did, repo.Name) 1049 - xrpcBytes, err := tangled.RepoCompare(r.Context(), xrpcc, didSlashRepo, targetBranch, sourceBranch) 1046 + xrpcBytes, err := tangled.RepoCompare(r.Context(), xrpcc, repo.RepoIdentifier(), targetBranch, sourceBranch) 1050 1047 if err != nil { 1051 1048 if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { 1052 1049 s.logger.Error("failed to call XRPC repo.compare", "err", xrpcerr) ··· 1155 1152 Host: forkHost, 1156 1153 } 1157 1154 1158 - forkRepoId := fmt.Sprintf("%s/%s", fork.Did, fork.Name) 1159 - forkXrpcBytes, err := tangled.RepoCompare(r.Context(), forkXrpcc, forkRepoId, hiddenRef, sourceBranch) 1155 + forkXrpcBytes, err := tangled.RepoCompare(r.Context(), forkXrpcc, fork.RepoIdentifier(), hiddenRef, sourceBranch) 1160 1156 if err != nil { 1161 1157 if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { 1162 1158 s.logger.Error("failed to call XRPC repo.compare for fork", "err", xrpcerr) ··· 1197 1193 Repo: &forkAtUriStr, 1198 1194 Sha: sourceRev, 1199 1195 } 1196 + if fork.RepoDid != "" { 1197 + recordPullSource.RepoDid = &fork.RepoDid 1198 + } 1200 1199 1201 1200 s.createPullRequest(w, r, repo, user, title, body, targetBranch, patch, combined, sourceRev, pullSource, recordPullSource, isStacked) 1202 1201 } ··· 1313 1312 Rkey: rkey, 1314 1313 Record: &lexutil.LexiconTypeDecoder{ 1315 1314 Val: &tangled.RepoPull{ 1316 - Title: title, 1317 - Target: &tangled.RepoPull_Target{ 1318 - Repo: string(repo.RepoAt()), 1319 - Branch: targetBranch, 1320 - }, 1315 + Title: title, 1316 + Target: repoPullTarget(repo, targetBranch), 1321 1317 PatchBlob: blob.Blob, 1322 1318 Source: recordPullSource, 1323 1319 CreatedAt: time.Now().Format(time.RFC3339), ··· 1707 1703 return 1708 1704 } 1709 1705 1710 - roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.DidSlashRepo())} 1706 + roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.RepoIdentifier())} 1711 1707 if !roles.IsPushAllowed() { 1712 1708 s.logger.Warn("unauthorized user") 1713 1709 w.WriteHeader(http.StatusUnauthorized) ··· 1723 1719 Host: host, 1724 1720 } 1725 1721 1726 - repo := fmt.Sprintf("%s/%s", f.Did, f.Name) 1727 - xrpcBytes, err := tangled.RepoCompare(r.Context(), xrpcc, repo, pull.TargetBranch, pull.PullSource.Branch) 1722 + xrpcBytes, err := tangled.RepoCompare(r.Context(), xrpcc, f.RepoIdentifier(), pull.TargetBranch, pull.PullSource.Branch) 1728 1723 if err != nil { 1729 1724 if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { 1730 1725 s.logger.Error("failed to call XRPC repo.compare", "err", xrpcerr) ··· 1817 1812 forkScheme = "https" 1818 1813 } 1819 1814 forkHost := fmt.Sprintf("%s://%s", forkScheme, forkRepo.Knot) 1820 - forkRepoId := fmt.Sprintf("%s/%s", forkRepo.Did, forkRepo.Name) 1821 - forkXrpcBytes, err := tangled.RepoCompare(r.Context(), &indigoxrpc.Client{Host: forkHost}, forkRepoId, hiddenRef, pull.PullSource.Branch) 1815 + forkXrpcBytes, err := tangled.RepoCompare(r.Context(), &indigoxrpc.Client{Host: forkHost}, forkRepo.RepoIdentifier(), hiddenRef, pull.PullSource.Branch) 1822 1816 if err != nil { 1823 1817 if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { 1824 1818 s.logger.Error("failed to call XRPC repo.compare for fork", "err", xrpcerr) ··· 2296 2290 } 2297 2291 2298 2292 // auth filter: only owner or collaborators can close 2299 - roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.DidSlashRepo())} 2293 + roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.RepoIdentifier())} 2300 2294 isOwner := roles.IsOwner() 2301 2295 isCollaborator := roles.IsCollaborator() 2302 2296 isPullAuthor := user.Active.Did == pull.OwnerDid ··· 2370 2364 } 2371 2365 2372 2366 // auth filter: only owner or collaborators can close 2373 - roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.DidSlashRepo())} 2367 + roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.RepoIdentifier())} 2374 2368 isOwner := roles.IsOwner() 2375 2369 isCollaborator := roles.IsCollaborator() 2376 2370 isPullAuthor := user.Active.Did == pull.OwnerDid ··· 2495 2489 } 2496 2490 2497 2491 func ptrPullState(s models.PullState) *models.PullState { return &s } 2492 + 2493 + func repoPullTarget(repo *models.Repo, branch string) *tangled.RepoPull_Target { 2494 + s := string(repo.RepoAt()) 2495 + t := &tangled.RepoPull_Target{ 2496 + Branch: branch, 2497 + Repo: &s, 2498 + } 2499 + if repo.RepoDid != "" { 2500 + t.RepoDid = &repo.RepoDid 2501 + } 2502 + return t 2503 + }
+1 -1
appview/repo/archive.go
··· 60 60 if link := resp.Header.Get("Link"); link != "" { 61 61 if resolvedRef, err := extractImmutableLink(link); err == nil { 62 62 newLink := fmt.Sprintf("<%s/%s/archive/%s.tar.gz>; rel=\"immutable\"", 63 - rp.config.Core.BaseUrl(), f.DidSlashRepo(), resolvedRef) 63 + rp.config.Core.BaseUrl(), f.RepoIdentifier(), resolvedRef) 64 64 w.Header().Set("Link", newLink) 65 65 } 66 66 }
+16 -7
appview/repo/artifact.go
··· 80 80 Repo: user.Active.Did, 81 81 Rkey: rkey, 82 82 Record: &lexutil.LexiconTypeDecoder{ 83 - Val: &tangled.RepoArtifact{ 84 - Artifact: uploadBlobResp.Blob, 85 - CreatedAt: createdAt.Format(time.RFC3339), 86 - Name: header.Filename, 87 - Repo: f.RepoAt().String(), 88 - Tag: tag.Tag.Hash[:], 89 - }, 83 + Val: repoArtifactRecord(f, uploadBlobResp.Blob, createdAt, header.Filename, tag.Tag.Hash[:]), 90 84 }, 91 85 }) 92 86 if err != nil { ··· 350 344 351 345 return tag, nil 352 346 } 347 + 348 + func repoArtifactRecord(f *models.Repo, blob *lexutil.LexBlob, createdAt time.Time, name string, tag []byte) *tangled.RepoArtifact { 349 + rec := &tangled.RepoArtifact{ 350 + Artifact: blob, 351 + CreatedAt: createdAt.Format(time.RFC3339), 352 + Name: name, 353 + Tag: tag, 354 + } 355 + s := f.RepoAt().String() 356 + rec.Repo = &s 357 + if f.RepoDid != "" { 358 + rec.RepoDid = &f.RepoDid 359 + } 360 + return rec 361 + }
+3 -4
appview/repo/blob.go
··· 58 58 xrpcc := &indigoxrpc.Client{ 59 59 Host: host, 60 60 } 61 - repo := fmt.Sprintf("%s/%s", f.Did, f.Name) 62 - resp, err := tangled.RepoBlob(r.Context(), xrpcc, filePath, false, ref, repo) 61 + resp, err := tangled.RepoBlob(r.Context(), xrpcc, filePath, false, ref, f.RepoIdentifier()) 63 62 if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { 64 63 l.Error("failed to call XRPC repo.blob", "err", xrpcerr) 65 64 rp.pages.Error503(w) ··· 139 138 if !rp.config.Core.Dev { 140 139 scheme = "https" 141 140 } 142 - repo := f.DidSlashRepo() 141 + repo := f.RepoIdentifier() 143 142 baseURL := &url.URL{ 144 143 Scheme: scheme, 145 144 Host: f.Knot, ··· 290 289 scheme = "https" 291 290 } 292 291 293 - repoName := fmt.Sprintf("%s/%s", repo.Did, repo.Name) 292 + repoName := repo.RepoIdentifier() 294 293 baseURL := &url.URL{ 295 294 Scheme: scheme, 296 295 Host: repo.Knot,
+4 -4
appview/repo/compare.go
··· 141 141 Host: host, 142 142 } 143 143 144 - repo := fmt.Sprintf("%s/%s", f.Did, f.Name) 144 + repoId := f.RepoIdentifier() 145 145 146 - branchBytes, err := tangled.RepoBranches(r.Context(), xrpcc, "", 0, repo) 146 + branchBytes, err := tangled.RepoBranches(r.Context(), xrpcc, "", 0, repoId) 147 147 if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { 148 148 l.Error("failed to call XRPC repo.branches", "err", xrpcerr) 149 149 rp.pages.Error503(w) ··· 157 157 return 158 158 } 159 159 160 - tagBytes, err := tangled.RepoTags(r.Context(), xrpcc, "", 0, repo) 160 + tagBytes, err := tangled.RepoTags(r.Context(), xrpcc, "", 0, repoId) 161 161 if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { 162 162 l.Error("failed to call XRPC repo.tags", "err", xrpcerr) 163 163 rp.pages.Error503(w) ··· 171 171 return 172 172 } 173 173 174 - compareBytes, err := tangled.RepoCompare(r.Context(), xrpcc, repo, base, head) 174 + compareBytes, err := tangled.RepoCompare(r.Context(), xrpcc, repoId, base, head) 175 175 if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { 176 176 l.Error("failed to call XRPC repo.compare", "err", xrpcerr) 177 177 rp.pages.Error503(w)
-1
appview/repo/index.go
··· 239 239 func (rp *Repo) buildIndexResponse(ctx context.Context, repo *models.Repo, ref string) (*types.RepoIndexResponse, error) { 240 240 xrpcc := &indigoxrpc.Client{Host: rp.config.KnotMirror.Url} 241 241 242 - // first get branches to determine the ref if not specified 243 242 branchesBytes, err := tangled.GitTempListBranches(ctx, xrpcc, "", 0, repo.RepoAt().String()) 244 243 if err != nil { 245 244 return nil, fmt.Errorf("calling knotmirror git.listBranches: %w", err)
+1 -2
appview/repo/log.go
··· 164 164 Host: host, 165 165 } 166 166 167 - repo := fmt.Sprintf("%s/%s", f.Did, f.Name) 168 - xrpcBytes, err := tangled.RepoDiff(r.Context(), xrpcc, ref, repo) 167 + xrpcBytes, err := tangled.RepoDiff(r.Context(), xrpcc, ref, f.RepoIdentifier()) 169 168 if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { 170 169 l.Error("failed to call XRPC repo.diff", "err", xrpcerr) 171 170 rp.pages.Error503(w)
+120 -55
appview/repo/repo.go
··· 36 36 "github.com/bluesky-social/indigo/atproto/atclient" 37 37 "github.com/bluesky-social/indigo/atproto/syntax" 38 38 lexutil "github.com/bluesky-social/indigo/lex/util" 39 - securejoin "github.com/cyphar/filepath-securejoin" 39 + 40 40 "github.com/go-chi/chi/v5" 41 41 ) 42 42 ··· 318 318 return 319 319 } 320 320 321 - err = db.SubscribeLabel(tx, &models.RepoLabel{ 321 + if err = db.SubscribeLabel(tx, &models.RepoLabel{ 322 322 RepoAt: f.RepoAt(), 323 323 LabelAt: label.AtUri(), 324 - }) 324 + }); err != nil { 325 + fail("Failed to subscribe to label.", err) 326 + return 327 + } 325 328 326 329 err = tx.Commit() 327 330 if err != nil { ··· 755 758 Repo: currentUser.Active.Did, 756 759 Rkey: rkey, 757 760 Record: &lexutil.LexiconTypeDecoder{ 758 - Val: &tangled.RepoCollaborator{ 759 - Subject: collaboratorIdent.DID.String(), 760 - Repo: string(f.RepoAt()), 761 - CreatedAt: createdAt.Format(time.RFC3339), 762 - }}, 761 + Val: repoCollaboratorRecord(f, collaboratorIdent.DID.String(), createdAt), 762 + }, 763 763 }) 764 764 // invalid record 765 765 if err != nil { ··· 794 794 } 795 795 defer rollback() 796 796 797 - err = rp.enforcer.AddCollaborator(collaboratorIdent.DID.String(), f.Knot, f.DidSlashRepo()) 797 + err = rp.enforcer.AddCollaborator(collaboratorIdent.DID.String(), f.Knot, f.RepoIdentifier()) 798 798 if err != nil { 799 799 fail("Failed to add collaborator permissions.", err) 800 800 return ··· 900 900 }() 901 901 902 902 // remove collaborator RBAC 903 - repoCollaborators, err := rp.enforcer.E.GetImplicitUsersForResourceByDomain(f.DidSlashRepo(), f.Knot) 903 + repoCollaborators, err := rp.enforcer.E.GetImplicitUsersForResourceByDomain(f.RepoIdentifier(), f.Knot) 904 904 if err != nil { 905 905 rp.pages.Notice(w, noticeId, "Failed to remove collaborators") 906 906 return 907 907 } 908 908 for _, c := range repoCollaborators { 909 909 did := c[0] 910 - rp.enforcer.RemoveCollaborator(did, f.Knot, f.DidSlashRepo()) 910 + rp.enforcer.RemoveCollaborator(did, f.Knot, f.RepoIdentifier()) 911 911 } 912 912 l.Info("removed collaborators") 913 913 914 914 // remove repo RBAC 915 - err = rp.enforcer.RemoveRepo(f.Did, f.Knot, f.DidSlashRepo()) 915 + err = rp.enforcer.RemoveRepo(f.Did, f.Knot, f.RepoIdentifier()) 916 916 if err != nil { 917 917 rp.pages.Notice(w, noticeId, "Failed to update RBAC rules") 918 918 return ··· 1067 1067 uri = "http" 1068 1068 } 1069 1069 1070 - forkSourceUrl := fmt.Sprintf("%s://%s/%s/%s", uri, f.Knot, f.Did, f.Name) 1070 + forkSourceUrl := fmt.Sprintf("%s://%s/%s", uri, f.Knot, f.RepoIdentifier()) 1071 1071 l = l.With("cloneUrl", forkSourceUrl) 1072 1072 1073 - sourceAt := f.RepoAt().String() 1074 - 1075 - // create an atproto record for this fork 1076 1073 rkey := tid.TID() 1074 + 1075 + // TODO: this could coordinate better with the knot to recieve a clone status 1076 + client, err := rp.oauth.ServiceClient( 1077 + r, 1078 + oauth.WithService(targetKnot), 1079 + oauth.WithLxm(tangled.RepoCreateNSID), 1080 + oauth.WithDev(rp.config.Core.Dev), 1081 + oauth.WithTimeout(time.Second*20), 1082 + ) 1083 + if err != nil { 1084 + l.Error("could not create service client", "err", err) 1085 + rp.pages.Notice(w, "repo", "Failed to connect to knot server.") 1086 + return 1087 + } 1088 + 1089 + forkInput := &tangled.RepoCreate_Input{ 1090 + Rkey: rkey, 1091 + Name: forkName, 1092 + Source: &forkSourceUrl, 1093 + } 1094 + createResp, createErr := tangled.RepoCreate( 1095 + r.Context(), 1096 + client, 1097 + forkInput, 1098 + ) 1099 + if err := xrpcclient.HandleXrpcErr(createErr); err != nil { 1100 + rp.pages.Notice(w, "repo", err.Error()) 1101 + return 1102 + } 1103 + 1104 + var repoDid string 1105 + if createResp != nil && createResp.RepoDid != nil { 1106 + repoDid = *createResp.RepoDid 1107 + } 1108 + if repoDid == "" { 1109 + l.Error("knot returned empty repo DID for fork") 1110 + rp.pages.Notice(w, "repo", "Knot failed to mint a repo DID. The knot may need to be upgraded.") 1111 + return 1112 + } 1113 + 1114 + forkSource := f.RepoAt().String() 1115 + if f.RepoDid != "" { 1116 + forkSource = f.RepoDid 1117 + } 1118 + 1077 1119 repo := &models.Repo{ 1078 1120 Did: user.Active.Did, 1079 1121 Name: forkName, 1080 1122 Knot: targetKnot, 1081 1123 Rkey: rkey, 1082 - Source: sourceAt, 1124 + Source: forkSource, 1083 1125 Description: f.Description, 1084 1126 Created: time.Now(), 1085 1127 Labels: rp.config.Label.DefaultLabelDefs, 1128 + RepoDid: repoDid, 1086 1129 } 1087 1130 record := repo.AsRecord() 1088 1131 1132 + cleanupKnot := func() { 1133 + go func() { 1134 + delays := []time.Duration{0, 2 * time.Second, 5 * time.Second} 1135 + for attempt, delay := range delays { 1136 + time.Sleep(delay) 1137 + deleteClient, dErr := rp.oauth.ServiceClient( 1138 + r, 1139 + oauth.WithService(targetKnot), 1140 + oauth.WithLxm(tangled.RepoDeleteNSID), 1141 + oauth.WithDev(rp.config.Core.Dev), 1142 + ) 1143 + if dErr != nil { 1144 + l.Error("failed to create delete client for knot cleanup", "attempt", attempt+1, "err", dErr) 1145 + continue 1146 + } 1147 + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) 1148 + if dErr := tangled.RepoDelete(ctx, deleteClient, &tangled.RepoDelete_Input{ 1149 + Did: user.Active.Did, 1150 + Name: forkName, 1151 + Rkey: rkey, 1152 + }); dErr != nil { 1153 + cancel() 1154 + l.Error("failed to clean up fork on knot after rollback", "attempt", attempt+1, "err", dErr) 1155 + continue 1156 + } 1157 + cancel() 1158 + l.Info("successfully cleaned up fork on knot after rollback", "attempt", attempt+1) 1159 + return 1160 + } 1161 + l.Error("exhausted retries for knot cleanup, fork may be orphaned", 1162 + "did", user.Active.Did, "fork", forkName, "knot", targetKnot) 1163 + }() 1164 + } 1165 + 1089 1166 atpClient, err := rp.oauth.AuthorizedClient(r) 1090 1167 if err != nil { 1091 1168 l.Error("failed to create xrpcclient", "err", err) 1169 + cleanupKnot() 1092 1170 rp.pages.Notice(w, "repo", "Failed to fork repository.") 1093 1171 return 1094 1172 } ··· 1103 1181 }) 1104 1182 if err != nil { 1105 1183 l.Error("failed to write to PDS", "err", err) 1184 + cleanupKnot() 1106 1185 rp.pages.Notice(w, "repo", "Failed to announce repository creation.") 1107 1186 return 1108 1187 } ··· 1118 1197 return 1119 1198 } 1120 1199 1121 - // The rollback function reverts a few things on failure: 1122 - // - the pending txn 1123 - // - the ACLs 1124 - // - the atproto record created 1125 1200 rollback := func() { 1126 1201 err1 := tx.Rollback() 1127 1202 err2 := rp.enforcer.E.LoadPolicy() 1128 1203 err3 := rollbackRecord(context.Background(), aturi, atpClient) 1129 1204 1130 - // ignore txn complete errors, this is okay 1131 1205 if errors.Is(err1, sql.ErrTxDone) { 1132 1206 err1 = nil 1133 1207 } 1134 1208 1135 1209 if errs := errors.Join(err1, err2, err3); errs != nil { 1136 1210 l.Error("failed to rollback changes", "errs", errs) 1137 - return 1211 + } 1212 + 1213 + if aturi != "" { 1214 + cleanupKnot() 1138 1215 } 1139 1216 } 1140 1217 defer rollback() 1141 1218 1142 - // TODO: this could coordinate better with the knot to recieve a clone status 1143 - client, err := rp.oauth.ServiceClient( 1144 - r, 1145 - oauth.WithService(targetKnot), 1146 - oauth.WithLxm(tangled.RepoCreateNSID), 1147 - oauth.WithDev(rp.config.Core.Dev), 1148 - oauth.WithTimeout(time.Second*20), // big repos take time to clone 1149 - ) 1150 - if err != nil { 1151 - l.Error("could not create service client", "err", err) 1152 - rp.pages.Notice(w, "repo", "Failed to connect to knot server.") 1153 - return 1154 - } 1155 - 1156 - err = tangled.RepoCreate( 1157 - r.Context(), 1158 - client, 1159 - &tangled.RepoCreate_Input{ 1160 - Rkey: rkey, 1161 - Source: &forkSourceUrl, 1162 - }, 1163 - ) 1164 - if err := xrpcclient.HandleXrpcErr(err); err != nil { 1165 - rp.pages.Notice(w, "repo", err.Error()) 1166 - return 1167 - } 1168 - 1169 1219 err = db.AddRepo(tx, repo) 1170 1220 if err != nil { 1171 1221 l.Error("failed to AddRepo", "err", err) ··· 1173 1223 return 1174 1224 } 1175 1225 1176 - // acls 1177 - p, _ := securejoin.SecureJoin(user.Active.Did, forkName) 1178 - err = rp.enforcer.AddRepo(user.Active.Did, targetKnot, p) 1226 + rbacPath := repo.RepoIdentifier() 1227 + err = rp.enforcer.AddRepo(user.Active.Did, targetKnot, rbacPath) 1179 1228 if err != nil { 1180 1229 l.Error("failed to add ACLs", "err", err) 1181 1230 rp.pages.Notice(w, "repo", "Failed to set up repository permissions.") ··· 1196 1245 return 1197 1246 } 1198 1247 1199 - // reset the ATURI because the transaction completed successfully 1200 1248 aturi = "" 1201 1249 1202 1250 rp.notifier.NewRepo(r.Context(), repo) 1203 - rp.pages.HxLocation(w, fmt.Sprintf("/%s/%s", user.Active.Did, forkName)) 1251 + if repoDid != "" { 1252 + rp.pages.HxLocation(w, fmt.Sprintf("/%s", repoDid)) 1253 + } else { 1254 + rp.pages.HxLocation(w, fmt.Sprintf("/%s/%s", user.Active.Did, forkName)) 1255 + } 1204 1256 } 1205 1257 } 1206 1258 ··· 1225 1277 }) 1226 1278 return err 1227 1279 } 1280 + 1281 + func repoCollaboratorRecord(f *models.Repo, subject string, createdAt time.Time) *tangled.RepoCollaborator { 1282 + rec := &tangled.RepoCollaborator{ 1283 + Subject: subject, 1284 + CreatedAt: createdAt.Format(time.RFC3339), 1285 + } 1286 + s := string(f.RepoAt()) 1287 + rec.Repo = &s 1288 + if f.RepoDid != "" { 1289 + rec.RepoDid = &f.RepoDid 1290 + } 1291 + return rec 1292 + }
+7 -7
appview/repo/settings.go
··· 293 293 // Skip entirely if there is no active domain claim — the site cannot be served anyway. 294 294 ownerClaim, _ := db.GetActiveDomainClaimForDid(rp.db, f.Did) 295 295 if ownerClaim == nil { 296 - rp.logger.Info("skipping deploy: no active domain claim", "repo", f.DidSlashRepo()) 296 + rp.logger.Info("skipping deploy: no active domain claim", "repo", f.RepoIdentifier()) 297 297 } else if rp.cfClient.Enabled() { 298 298 scheme := "http" 299 299 if !rp.config.Core.Dev { ··· 313 313 314 314 deployErr := sites.Deploy(ctx, rp.cfClient, knotHost, f.Did, f.Name, branch, dir) 315 315 if deployErr != nil { 316 - l.Error("sites: initial R2 sync failed", "repo", f.DidSlashRepo(), "err", deployErr) 316 + l.Error("sites: initial R2 sync failed", "repo", f.RepoIdentifier(), "err", deployErr) 317 317 deploy.Status = models.SiteDeployStatusFailure 318 318 deploy.Error = deployErr.Error() 319 319 } else { ··· 321 321 } 322 322 323 323 if err := db.AddSiteDeploy(rp.db, deploy); err != nil { 324 - l.Error("sites: failed to record deploy", "repo", f.DidSlashRepo(), "err", err) 324 + l.Error("sites: failed to record deploy", "repo", f.RepoIdentifier(), "err", err) 325 325 } 326 326 327 327 if deployErr == nil { 328 328 if err := sites.PutDomainMapping(ctx, rp.cfClient, ownerClaim.Domain, f.Did, f.Name, isIndex); err != nil { 329 329 l.Error("sites: KV write failed", "domain", ownerClaim.Domain, "err", err) 330 330 } 331 - rp.logger.Info("site deployed to r2", "repo", f.DidSlashRepo(), "is_index", isIndex) 331 + rp.logger.Info("site deployed to r2", "repo", f.RepoIdentifier(), "is_index", isIndex) 332 332 } 333 333 }() 334 334 } else { 335 - rp.logger.Warn("cloudflare integration is disabled; site won't be deployed", "repo", f.DidSlashRepo()) 335 + rp.logger.Warn("cloudflare integration is disabled; site won't be deployed", "repo", f.RepoIdentifier()) 336 336 } 337 337 338 338 rp.pages.HxRefresh(w) ··· 367 367 go func() { 368 368 ctx := context.Background() 369 369 if err := sites.Delete(ctx, rp.cfClient, f.Did, f.Name); err != nil { 370 - l.Error("sites: R2 delete failed", "repo", f.DidSlashRepo(), "err", err) 370 + l.Error("sites: R2 delete failed", "repo", f.RepoIdentifier(), "err", err) 371 371 } 372 372 if ownerClaim != nil { 373 373 if err := sites.DeleteDomainMapping(ctx, rp.cfClient, ownerClaim.Domain, f.Name); err != nil { ··· 459 459 user := rp.oauth.GetMultiAccountUser(r) 460 460 461 461 collaborators, err := func(repo *models.Repo) ([]pages.Collaborator, error) { 462 - repoCollaborators, err := rp.enforcer.E.GetImplicitUsersForResourceByDomain(repo.DidSlashRepo(), repo.Knot) 462 + repoCollaborators, err := rp.enforcer.E.GetImplicitUsersForResourceByDomain(repo.RepoIdentifier(), repo.Knot) 463 463 if err != nil { 464 464 return nil, err 465 465 }
+13 -6
appview/reporesolver/resolver.go
··· 36 36 37 37 // NOTE: this... should not even be here. the entire package will be removed in future refactor 38 38 func GetBaseRepoPath(r *http.Request, repo *models.Repo) string { 39 + if repo.RepoDid != "" { 40 + return repo.RepoDid 41 + } 39 42 var ( 40 43 user = chi.URLParam(r, "user") 41 44 name = chi.URLParam(r, "repo") 42 45 ) 43 46 if user == "" || name == "" { 44 - return repo.DidSlashRepo() 47 + return repo.RepoIdentifier() 45 48 } 46 49 return path.Join(user, name) 47 50 } ··· 77 80 roles := repoinfo.RolesInRepo{} 78 81 if user != nil && user.Active != nil { 79 82 isStarred = db.GetStarStatus(rr.execer, user.Active.Did, repoAt) 80 - roles.Roles = rr.enforcer.GetPermissionsInRepo(user.Active.Did, repo.Knot, repo.DidSlashRepo()) 83 + roles.Roles = rr.enforcer.GetPermissionsInRepo(user.Active.Did, repo.Knot, repo.RepoIdentifier()) 81 84 } 82 85 83 86 stats := repo.RepoStats 84 87 if stats == nil { 85 - starCount, err := db.GetStarCount(rr.execer, repoAt) 86 - if err != nil { 88 + starCount, starErr := db.GetStarCount(rr.execer, repoAt) 89 + if starErr != nil { 87 90 log.Println("failed to get star count for ", repoAt) 88 91 } 89 92 issueCount, err := db.GetIssueCount(rr.execer, repoAt) ··· 104 107 var sourceRepo *models.Repo 105 108 var err error 106 109 if repo.Source != "" { 107 - sourceRepo, err = db.GetRepoByAtUri(rr.execer, repo.Source) 110 + if strings.HasPrefix(repo.Source, "did:") { 111 + sourceRepo, err = db.GetRepoByDid(rr.execer, repo.Source) 112 + } else { 113 + sourceRepo, err = db.GetRepoByAtUri(rr.execer, repo.Source) 114 + } 108 115 if err != nil { 109 - log.Println("failed to get repo by at uri", err) 116 + log.Println("failed to get source repo", err) 110 117 } 111 118 } 112 119
+7 -22
appview/state/git_http.go
··· 37 37 } 38 38 39 39 func (s *State) InfoRefs(w http.ResponseWriter, r *http.Request) { 40 - user := r.Context().Value("resolvedId").(identity.Identity) 41 40 repo := r.Context().Value("repo").(*models.Repo) 42 41 43 42 scheme := "https" ··· 45 44 scheme = "http" 46 45 } 47 46 48 - // check for the 'service' url param 49 47 service := r.URL.Query().Get("service") 50 48 var contentType string 51 49 switch service { 52 50 case "git-receive-pack": 53 51 contentType = "application/x-git-receive-pack-advertisement" 54 52 default: 55 - // git-upload-pack is the default service for git-clone / git-fetch. 56 53 contentType = "application/x-git-upload-pack-advertisement" 57 54 } 58 55 59 - targetURL := fmt.Sprintf("%s://%s/%s/%s/info/refs?%s", scheme, repo.Knot, user.DID, repo.Name, r.URL.RawQuery) 56 + targetURL := fmt.Sprintf("%s://%s/%s/info/refs?%s", scheme, repo.Knot, repo.RepoIdentifier(), r.URL.RawQuery) 60 57 s.proxyRequest(w, r, targetURL, contentType) 61 58 } 62 59 63 60 func (s *State) UploadArchive(w http.ResponseWriter, r *http.Request) { 64 - user, ok := r.Context().Value("resolvedId").(identity.Identity) 65 - if !ok { 66 - http.Error(w, "failed to resolve user", http.StatusInternalServerError) 67 - return 68 - } 69 61 repo := r.Context().Value("repo").(*models.Repo) 70 62 71 63 scheme := "https" ··· 73 65 scheme = "http" 74 66 } 75 67 76 - targetURL := fmt.Sprintf("%s://%s/%s/%s/git-upload-archive?%s", scheme, repo.Knot, user.DID, repo.Name, r.URL.RawQuery) 68 + targetURL := fmt.Sprintf("%s://%s/%s/git-upload-archive?%s", scheme, repo.Knot, repo.RepoIdentifier(), r.URL.RawQuery) 77 69 s.proxyRequest(w, r, targetURL, "application/x-git-upload-archive-result") 78 70 } 79 71 80 72 func (s *State) UploadPack(w http.ResponseWriter, r *http.Request) { 81 - user, ok := r.Context().Value("resolvedId").(identity.Identity) 82 - if !ok { 83 - http.Error(w, "failed to resolve user", http.StatusInternalServerError) 84 - return 85 - } 86 73 repo := r.Context().Value("repo").(*models.Repo) 87 74 88 75 scheme := "https" ··· 90 77 scheme = "http" 91 78 } 92 79 93 - targetURL := fmt.Sprintf("%s://%s/%s/%s/git-upload-pack?%s", scheme, repo.Knot, user.DID, repo.Name, r.URL.RawQuery) 80 + targetURL := fmt.Sprintf("%s://%s/%s/git-upload-pack?%s", scheme, repo.Knot, repo.RepoIdentifier(), r.URL.RawQuery) 94 81 s.proxyRequest(w, r, targetURL, "application/x-git-upload-pack-result") 95 82 } 96 83 97 84 func (s *State) ReceivePack(w http.ResponseWriter, r *http.Request) { 98 - user, ok := r.Context().Value("resolvedId").(identity.Identity) 99 - if !ok { 100 - http.Error(w, "failed to resolve user", http.StatusInternalServerError) 101 - return 102 - } 103 85 repo := r.Context().Value("repo").(*models.Repo) 104 86 105 87 scheme := "https" ··· 107 89 scheme = "http" 108 90 } 109 91 110 - targetURL := fmt.Sprintf("%s://%s/%s/%s/git-receive-pack?%s", scheme, repo.Knot, user.DID, repo.Name, r.URL.RawQuery) 92 + targetURL := fmt.Sprintf("%s://%s/%s/git-receive-pack?%s", scheme, repo.Knot, repo.RepoIdentifier(), r.URL.RawQuery) 111 93 s.proxyRequest(w, r, targetURL, "application/x-git-receive-pack-result") 112 94 } 113 95 ··· 123 105 proxyReq.Header = r.Header.Clone() 124 106 125 107 repoOwnerHandle := chi.URLParam(r, "user") 108 + if id, ok := r.Context().Value("resolvedId").(identity.Identity); ok && !id.Handle.IsInvalidHandle() { 109 + repoOwnerHandle = id.Handle.String() 110 + } 126 111 proxyReq.Header.Set("x-tangled-repo-owner-handle", repoOwnerHandle) 127 112 128 113 resp, err := client.Do(proxyReq)
+53 -50
appview/state/knotstream.go
··· 2 2 3 3 import ( 4 4 "context" 5 + "database/sql" 5 6 "encoding/json" 6 7 "errors" 7 8 "fmt" ··· 66 67 return ec.NewConsumer(cfg), nil 67 68 } 68 69 70 + func resolveRepo(d *db.DB, repoDid *string, ownerDid, repoName string) (*models.Repo, error) { 71 + if repoDid != nil && *repoDid != "" { 72 + return db.GetRepoByDid(d, *repoDid) 73 + } 74 + repos, err := db.GetRepos(d, orm.FilterEq("did", ownerDid), orm.FilterEq("name", repoName)) 75 + if err != nil { 76 + return nil, err 77 + } 78 + if len(repos) == 0 { 79 + return nil, sql.ErrNoRows 80 + } 81 + return &repos[0], nil 82 + } 83 + 69 84 func knotIngester(d *db.DB, enforcer *rbac.Enforcer, posthog posthog.Client, notifier notify.Notifier, dev bool, c *config.Config, cfClient *cloudflare.Client) ec.ProcessFunc { 70 85 return func(ctx context.Context, source ec.Source, msg ec.Message) error { 71 86 switch msg.Nsid { ··· 96 111 return fmt.Errorf("%s does not belong to %s, something is fishy", record.CommitterDid, source.Key()) 97 112 } 98 113 99 - repo, err := db.GetRepo( 100 - d, 101 - orm.FilterEq("did", record.RepoDid), 102 - orm.FilterEq("name", record.RepoName), 103 - orm.FilterEq("knot", source.Key()), 104 - ) 105 - if err != nil { 106 - return fmt.Errorf("repo %s/%s on knot %s not found", record.RepoDid, record.RepoName, source.Key()) 114 + ownerDid := "" 115 + if record.OwnerDid != nil { 116 + ownerDid = *record.OwnerDid 117 + } 118 + 119 + repo, lookupErr := resolveRepo(d, record.RepoDid, ownerDid, record.RepoName) 120 + if lookupErr != nil { 121 + return fmt.Errorf("failed to look up repo: %w", lookupErr) 107 122 } 108 123 109 124 logger.Info("processing gitRefUpdate event", 110 - "repo_did", record.RepoDid, 111 - "repo_name", record.RepoName, 125 + "repo", repo.RepoIdentifier(), 112 126 "ref", record.Ref, 113 127 "old_sha", record.OldSha, 114 128 "new_sha", record.NewSha) ··· 145 159 } 146 160 pushedBranch := ref.Short() 147 161 148 - repos, err := db.GetRepos( 149 - d, 150 - orm.FilterEq("did", record.RepoDid), 151 - orm.FilterEq("name", record.RepoName), 152 - ) 153 - if err != nil || len(repos) != 1 { 162 + ownerDid := "" 163 + if record.OwnerDid != nil { 164 + ownerDid = *record.OwnerDid 165 + } 166 + 167 + repo, err := resolveRepo(d, record.RepoDid, ownerDid, record.RepoName) 168 + if err != nil { 154 169 return 155 170 } 156 - repo := repos[0] 157 171 158 172 siteConfig, err := db.GetRepoSiteConfig(d, repo.RepoAt().String()) 159 173 if err != nil || siteConfig == nil { ··· 177 191 Trigger: models.SiteDeployTriggerPush, 178 192 } 179 193 180 - deployErr := sites.Deploy(ctx, cfClient, knotHost, record.RepoDid, record.RepoName, siteConfig.Branch, siteConfig.Dir) 194 + deployErr := sites.Deploy(ctx, cfClient, knotHost, repo.RepoIdentifier(), record.RepoName, siteConfig.Branch, siteConfig.Dir) 181 195 if deployErr != nil { 182 - logger.Error("sites: R2 sync failed on push", "repo", record.RepoDid+"/"+record.RepoName, "err", deployErr) 196 + logger.Error("sites: R2 sync failed on push", "repo", repo.RepoIdentifier(), "err", deployErr) 183 197 deploy.Status = models.SiteDeployStatusFailure 184 198 deploy.Error = deployErr.Error() 185 199 } else { ··· 187 201 } 188 202 189 203 if err := db.AddSiteDeploy(d, deploy); err != nil { 190 - logger.Error("sites: failed to record deploy", "repo", record.RepoDid+"/"+record.RepoName, "err", err) 204 + logger.Error("sites: failed to record deploy", "repo", repo.RepoIdentifier(), "err", err) 191 205 } 192 206 193 207 if deployErr == nil { 194 - logger.Info("site deployed to r2", "repo", record.RepoDid+"/"+record.RepoName) 208 + logger.Info("site deployed to r2", "repo", repo.RepoIdentifier()) 195 209 } 196 210 } 197 211 ··· 233 247 234 248 func updateRepoLanguages(d *db.DB, record tangled.GitRefUpdate) error { 235 249 if record.Meta == nil || record.Meta.LangBreakdown == nil || record.Meta.LangBreakdown.Inputs == nil { 236 - return fmt.Errorf("empty language data for repo: %s/%s", record.RepoDid, record.RepoName) 250 + return fmt.Errorf("empty language data for repo: %v/%s", record.OwnerDid, record.RepoName) 237 251 } 238 252 239 - repos, err := db.GetRepos( 240 - d, 241 - orm.FilterEq("did", record.RepoDid), 242 - orm.FilterEq("name", record.RepoName), 243 - ) 244 - if err != nil { 245 - return fmt.Errorf("failed to look for repo in DB (%s/%s): %w", record.RepoDid, record.RepoName, err) 253 + ownerDid := "" 254 + if record.OwnerDid != nil { 255 + ownerDid = *record.OwnerDid 246 256 } 247 - if len(repos) != 1 { 248 - return fmt.Errorf("incorrect number of repos returned: %d (expected 1)", len(repos)) 257 + 258 + r, lookupErr := resolveRepo(d, record.RepoDid, ownerDid, record.RepoName) 259 + if lookupErr != nil { 260 + return fmt.Errorf("failed to look up repo: %w", lookupErr) 249 261 } 250 - repo := repos[0] 262 + repo := *r 251 263 252 264 ref := plumbing.ReferenceName(record.Ref) 253 265 if !ref.IsBranch() { ··· 300 312 return fmt.Errorf("empty repo: nsid %s, rkey %s", msg.Nsid, msg.Rkey) 301 313 } 302 314 303 - repo, err := db.GetRepo( 304 - d, 305 - orm.FilterEq("did", record.TriggerMetadata.Repo.Did), 306 - orm.FilterEq("name", record.TriggerMetadata.Repo.Repo), 307 - orm.FilterEq("knot", source.Key()), 308 - ) 309 - if err != nil { 310 - return fmt.Errorf( 311 - "failed to look for repo in DB: nsid %s, rkey %s, %s/%s, knot %s, %w", 312 - msg.Nsid, 313 - msg.Rkey, 314 - record.TriggerMetadata.Repo.Did, 315 - record.TriggerMetadata.Repo.Did, 316 - source.Key(), 317 - err, 318 - ) 315 + repoName := "" 316 + if record.TriggerMetadata.Repo.Repo != nil { 317 + repoName = *record.TriggerMetadata.Repo.Repo 319 318 } 320 319 321 - // does this repo have a spindle configured? 320 + repo, lookupErr := resolveRepo(d, record.TriggerMetadata.Repo.RepoDid, record.TriggerMetadata.Repo.Did, repoName) 321 + if lookupErr != nil { 322 + return fmt.Errorf("failed to look up repo: %w", lookupErr) 323 + } 322 324 if repo.Spindle == "" { 323 325 return fmt.Errorf("repo does not have a spindle configured yet: nsid %s, rkey %s", msg.Nsid, msg.Rkey) 324 326 } ··· 355 357 Rkey: msg.Rkey, 356 358 Knot: source.Key(), 357 359 RepoOwner: syntax.DID(record.TriggerMetadata.Repo.Did), 358 - RepoName: record.TriggerMetadata.Repo.Repo, 360 + RepoName: repoName, 361 + RepoDid: repo.RepoDid, 359 362 TriggerId: int(triggerId), 360 363 Sha: sha, 361 364 }
+26 -2
appview/state/router.go
··· 1 1 package state 2 2 3 3 import ( 4 + "database/sql" 5 + "errors" 4 6 "net/http" 5 7 "strings" 6 8 7 9 "github.com/go-chi/chi/v5" 10 + "tangled.org/core/appview/db" 8 11 "tangled.org/core/appview/issues" 9 12 "tangled.org/core/appview/knots" 10 13 "tangled.org/core/appview/labels" ··· 46 49 if len(pathParts) > 0 { 47 50 firstPart := pathParts[0] 48 51 49 - // if using a DID or handle, just continue as per usual 50 - if userutil.IsDid(firstPart) || userutil.IsHandle(firstPart) { 52 + if userutil.IsDid(firstPart) { 53 + repo, err := db.GetRepoByDid(s.db, firstPart) 54 + switch { 55 + case err == nil: 56 + remaining := "" 57 + if len(pathParts) > 1 { 58 + remaining = "/" + pathParts[1] 59 + } 60 + rewritten := "/" + repo.Did + "/" + repo.Name + remaining 61 + r2 := r.Clone(r.Context()) 62 + r2.URL.Path = rewritten 63 + r2.URL.RawPath = rewritten 64 + userRouter.ServeHTTP(w, r2) 65 + case errors.Is(err, sql.ErrNoRows): 66 + userRouter.ServeHTTP(w, r) 67 + default: 68 + s.logger.Error("db error looking up repo DID", "repoDid", firstPart, "err", err) 69 + http.Error(w, "internal server error", http.StatusInternalServerError) 70 + } 71 + return 72 + } 73 + 74 + if userutil.IsHandle(firstPart) { 51 75 userRouter.ServeHTTP(w, r) 52 76 return 53 77 }
+13 -5
appview/state/star.go
··· 12 12 "tangled.org/core/appview/db" 13 13 "tangled.org/core/appview/models" 14 14 "tangled.org/core/appview/pages" 15 + "tangled.org/core/orm" 15 16 "tangled.org/core/tid" 16 17 ) 17 18 ··· 40 41 case http.MethodPost: 41 42 createdAt := time.Now().Format(time.RFC3339) 42 43 rkey := tid.TID() 44 + 45 + subjectStr := subjectUri.String() 46 + starRecord := &tangled.FeedStar{ 47 + CreatedAt: createdAt, 48 + Subject: &subjectStr, 49 + } 50 + repo, err := db.GetRepo(s.db, orm.FilterEq("at_uri", subjectUri.String())) 51 + if err == nil && repo.RepoDid != "" { 52 + starRecord.SubjectDid = &repo.RepoDid 53 + } 54 + 43 55 resp, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ 44 56 Collection: tangled.FeedStarNSID, 45 57 Repo: currentUser.Active.Did, 46 58 Rkey: rkey, 47 - Record: &lexutil.LexiconTypeDecoder{ 48 - Val: &tangled.FeedStar{ 49 - Subject: subjectUri.String(), 50 - CreatedAt: createdAt, 51 - }}, 59 + Record: &lexutil.LexiconTypeDecoder{Val: starRecord}, 52 60 }) 53 61 if err != nil { 54 62 log.Println("failed to create atproto record", err)
+88 -38
appview/state/state.go
··· 42 42 "github.com/bluesky-social/indigo/atproto/syntax" 43 43 lexutil "github.com/bluesky-social/indigo/lex/util" 44 44 "github.com/bluesky-social/indigo/xrpc" 45 - securejoin "github.com/cyphar/filepath-securejoin" 45 + 46 46 "github.com/go-chi/chi/v5" 47 47 "github.com/posthog/posthog-go" 48 48 ) ··· 457 457 return 458 458 } 459 459 460 - // create atproto record for this repo 461 460 rkey := tid.TID() 461 + 462 + client, err := s.oauth.ServiceClient( 463 + r, 464 + oauth.WithService(domain), 465 + oauth.WithLxm(tangled.RepoCreateNSID), 466 + oauth.WithDev(s.config.Core.Dev), 467 + ) 468 + if err != nil { 469 + l.Error("service auth failed", "err", err) 470 + s.pages.Notice(w, "repo", "Failed to reach knot server.") 471 + return 472 + } 473 + 474 + input := &tangled.RepoCreate_Input{ 475 + Rkey: rkey, 476 + Name: repoName, 477 + DefaultBranch: &defaultBranch, 478 + } 479 + createResp, xe := tangled.RepoCreate( 480 + r.Context(), 481 + client, 482 + input, 483 + ) 484 + if err := xrpcclient.HandleXrpcErr(xe); err != nil { 485 + l.Error("xrpc error", "xe", xe) 486 + s.pages.Notice(w, "repo", err.Error()) 487 + return 488 + } 489 + 490 + var repoDid string 491 + if createResp != nil && createResp.RepoDid != nil { 492 + repoDid = *createResp.RepoDid 493 + } 494 + if repoDid == "" { 495 + l.Error("knot returned empty repo DID") 496 + s.pages.Notice(w, "repo", "Knot failed to mint a repo DID. The knot may need to be upgraded.") 497 + return 498 + } 499 + 462 500 repo := &models.Repo{ 463 501 Did: user.Active.Did, 464 502 Name: repoName, ··· 467 505 Description: description, 468 506 Created: time.Now(), 469 507 Labels: s.config.Label.DefaultLabelDefs, 508 + RepoDid: repoDid, 470 509 } 471 510 record := repo.AsRecord() 472 511 512 + cleanupKnot := func() { 513 + go func() { 514 + delays := []time.Duration{0, 2 * time.Second, 5 * time.Second} 515 + for attempt, delay := range delays { 516 + time.Sleep(delay) 517 + deleteClient, dErr := s.oauth.ServiceClient( 518 + r, 519 + oauth.WithService(domain), 520 + oauth.WithLxm(tangled.RepoDeleteNSID), 521 + oauth.WithDev(s.config.Core.Dev), 522 + ) 523 + if dErr != nil { 524 + l.Error("failed to create delete client for knot cleanup", "attempt", attempt+1, "err", dErr) 525 + continue 526 + } 527 + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) 528 + if dErr := tangled.RepoDelete(ctx, deleteClient, &tangled.RepoDelete_Input{ 529 + Did: user.Active.Did, 530 + Name: repoName, 531 + Rkey: rkey, 532 + }); dErr != nil { 533 + cancel() 534 + l.Error("failed to clean up repo on knot after rollback", "attempt", attempt+1, "err", dErr) 535 + continue 536 + } 537 + cancel() 538 + l.Info("successfully cleaned up repo on knot after rollback", "attempt", attempt+1) 539 + return 540 + } 541 + l.Error("exhausted retries for knot cleanup, repo may be orphaned", 542 + "did", user.Active.Did, "repo", repoName, "knot", domain) 543 + }() 544 + } 545 + 473 546 atpClient, err := s.oauth.AuthorizedClient(r) 474 547 if err != nil { 475 548 l.Info("PDS write failed", "err", err) 549 + cleanupKnot() 476 550 s.pages.Notice(w, "repo", "Failed to write record to PDS.") 477 551 return 478 552 } ··· 487 561 }) 488 562 if err != nil { 489 563 l.Info("PDS write failed", "err", err) 564 + cleanupKnot() 490 565 s.pages.Notice(w, "repo", "Failed to announce repository creation.") 491 566 return 492 567 } ··· 502 577 return 503 578 } 504 579 505 - // The rollback function reverts a few things on failure: 506 - // - the pending txn 507 - // - the ACLs 508 - // - the atproto record created 509 580 rollback := func() { 510 581 err1 := tx.Rollback() 511 582 err2 := s.enforcer.E.LoadPolicy() 512 583 err3 := rollbackRecord(context.Background(), aturi, atpClient) 513 584 514 - // ignore txn complete errors, this is okay 515 585 if errors.Is(err1, sql.ErrTxDone) { 516 586 err1 = nil 517 587 } 518 588 519 589 if errs := errors.Join(err1, err2, err3); errs != nil { 520 590 l.Error("failed to rollback changes", "errs", errs) 521 - return 522 591 } 523 - } 524 - defer rollback() 525 592 526 - client, err := s.oauth.ServiceClient( 527 - r, 528 - oauth.WithService(domain), 529 - oauth.WithLxm(tangled.RepoCreateNSID), 530 - oauth.WithDev(s.config.Core.Dev), 531 - ) 532 - if err != nil { 533 - l.Error("service auth failed", "err", err) 534 - s.pages.Notice(w, "repo", "Failed to reach PDS.") 535 - return 593 + if aturi != "" { 594 + cleanupKnot() 595 + } 536 596 } 537 - 538 - xe := tangled.RepoCreate( 539 - r.Context(), 540 - client, 541 - &tangled.RepoCreate_Input{ 542 - Rkey: rkey, 543 - }, 544 - ) 545 - if err := xrpcclient.HandleXrpcErr(xe); err != nil { 546 - l.Error("xrpc error", "xe", xe) 547 - s.pages.Notice(w, "repo", err.Error()) 548 - return 549 - } 597 + defer rollback() 550 598 551 599 err = db.AddRepo(tx, repo) 552 600 if err != nil { ··· 555 603 return 556 604 } 557 605 558 - // acls 559 - p, _ := securejoin.SecureJoin(user.Active.Did, repoName) 560 - err = s.enforcer.AddRepo(user.Active.Did, domain, p) 606 + rbacPath := repo.RepoIdentifier() 607 + err = s.enforcer.AddRepo(user.Active.Did, domain, rbacPath) 561 608 if err != nil { 562 609 l.Error("acl setup failed", "err", err) 563 610 s.pages.Notice(w, "repo", "Failed to set up repository permissions.") ··· 578 625 return 579 626 } 580 627 581 - // reset the ATURI because the transaction completed successfully 582 628 aturi = "" 583 629 584 630 s.notifier.NewRepo(r.Context(), repo) 585 - s.pages.HxLocation(w, fmt.Sprintf("/%s/%s", user.Active.Did, repoName)) 631 + if repoDid != "" { 632 + s.pages.HxLocation(w, fmt.Sprintf("/%s", repoDid)) 633 + } else { 634 + s.pages.HxLocation(w, fmt.Sprintf("/%s/%s", user.Active.Did, repoName)) 635 + } 586 636 } 587 637 } 588 638
+1 -1
appview/validator/label.go
··· 109 109 // validate permissions: only collaborators can apply labels currently 110 110 // 111 111 // TODO: introduce a repo:triage permission 112 - ok, err := v.enforcer.IsPushAllowed(labelOp.Did, repo.Knot, repo.DidSlashRepo()) 112 + ok, err := v.enforcer.IsPushAllowed(labelOp.Did, repo.Knot, repo.RepoIdentifier()) 113 113 if err != nil { 114 114 return fmt.Errorf("failed to enforce permissions: %w", err) 115 115 }