-136

knotserver/ingester.go

··· 7 7 "io" 8 8 "net/http" 9 9 "net/url" 10 10 - "path/filepath" 11 10 "strings" 12 11 13 12 comatproto "github.com/bluesky-social/indigo/api/atproto" ··· 17 16 securejoin "github.com/cyphar/filepath-securejoin" 18 17 "tangled.org/core/api/tangled" 19 18 "tangled.org/core/knotserver/db" 20 20 - "tangled.org/core/knotserver/git" 21 19 "tangled.org/core/log" 22 20 "tangled.org/core/rbac" 23 23 - "tangled.org/core/workflow" 24 21 ) 25 22 26 23 func (h *Knot) processPublicKey(ctx context.Context, event *models.Event) error { ··· 85 82 return nil 86 83 } 87 84 88 88 - func (h *Knot) processPull(ctx context.Context, event *models.Event) error { 89 89 - raw := json.RawMessage(event.Commit.Record) 90 90 - did := event.Did 91 91 - 92 92 - var record tangled.RepoPull 93 93 - if err := json.Unmarshal(raw, &record); err != nil { 94 94 - return fmt.Errorf("failed to unmarshal record: %w", err) 95 95 - } 96 96 - 97 97 - l := log.FromContext(ctx) 98 98 - l = l.With("handler", "processPull") 99 99 - l = l.With("did", did) 100 100 - 101 101 - if record.Target == nil { 102 102 - return fmt.Errorf("ignoring pull record: target repo is nil") 103 103 - } 104 104 - 105 105 - l = l.With("target_repo", record.Target.Repo) 106 106 - l = l.With("target_branch", record.Target.Branch) 107 107 - 108 108 - if record.Source == nil { 109 109 - return fmt.Errorf("ignoring pull record: not a branch-based pull request") 110 110 - } 111 111 - 112 112 - if record.Source.Repo != nil { 113 113 - return fmt.Errorf("ignoring pull record: fork based pull") 114 114 - } 115 115 - 116 116 - repoAt, err := syntax.ParseATURI(record.Target.Repo) 117 117 - if err != nil { 118 118 - return fmt.Errorf("failed to parse ATURI: %w", err) 119 119 - } 120 120 - 121 121 - // resolve this aturi to extract the repo record 122 122 - ident, err := h.resolver.ResolveIdent(ctx, repoAt.Authority().String()) 123 123 - if err != nil || ident.Handle.IsInvalidHandle() { 124 124 - return fmt.Errorf("failed to resolve handle: %w", err) 125 125 - } 126 126 - 127 127 - xrpcc := xrpc.Client{ 128 128 - Host: ident.PDSEndpoint(), 129 129 - } 130 130 - 131 131 - resp, err := comatproto.RepoGetRecord(ctx, &xrpcc, "", tangled.RepoNSID, repoAt.Authority().String(), repoAt.RecordKey().String()) 132 132 - if err != nil { 133 133 - return fmt.Errorf("failed to resolver repo: %w", err) 134 134 - } 135 135 - 136 136 - repo := resp.Value.Val.(*tangled.Repo) 137 137 - 138 138 - if repo.Knot != h.c.Server.Hostname { 139 139 - return fmt.Errorf("rejected pull record: not this knot, %s != %s", repo.Knot, h.c.Server.Hostname) 140 140 - } 141 141 - 142 142 - didSlashRepo, err := securejoin.SecureJoin(ident.DID.String(), repo.Name) 143 143 - if err != nil { 144 144 - return fmt.Errorf("failed to construct relative repo path: %w", err) 145 145 - } 146 146 - 147 147 - repoPath, err := securejoin.SecureJoin(h.c.Repo.ScanPath, didSlashRepo) 148 148 - if err != nil { 149 149 - return fmt.Errorf("failed to construct absolute repo path: %w", err) 150 150 - } 151 151 - 152 152 - gr, err := git.Open(repoPath, record.Source.Sha) 153 153 - if err != nil { 154 154 - return fmt.Errorf("failed to open git repository: %w", err) 155 155 - } 156 156 - 157 157 - workflowDir, err := gr.FileTree(ctx, workflow.WorkflowDir) 158 158 - if err != nil { 159 159 - return fmt.Errorf("failed to open workflow directory: %w", err) 160 160 - } 161 161 - 162 162 - var pipeline workflow.RawPipeline 163 163 - for _, e := range workflowDir { 164 164 - if !e.IsFile() { 165 165 - continue 166 166 - } 167 167 - 168 168 - fpath := filepath.Join(workflow.WorkflowDir, e.Name) 169 169 - contents, err := gr.RawContent(fpath) 170 170 - if err != nil { 171 171 - continue 172 172 - } 173 173 - 174 174 - pipeline = append(pipeline, workflow.RawWorkflow{ 175 175 - Name: e.Name, 176 176 - Contents: contents, 177 177 - }) 178 178 - } 179 179 - 180 180 - trigger := tangled.Pipeline_PullRequestTriggerData{ 181 181 - Action: "create", 182 182 - SourceBranch: record.Source.Branch, 183 183 - SourceSha: record.Source.Sha, 184 184 - TargetBranch: record.Target.Branch, 185 185 - } 186 186 - 187 187 - compiler := workflow.Compiler{ 188 188 - Trigger: tangled.Pipeline_TriggerMetadata{ 189 189 - Kind: string(workflow.TriggerKindPullRequest), 190 190 - PullRequest: &trigger, 191 191 - Repo: &tangled.Pipeline_TriggerRepo{ 192 192 - Did: ident.DID.String(), 193 193 - Knot: repo.Knot, 194 194 - Repo: repo.Name, 195 195 - }, 196 196 - }, 197 197 - } 198 198 - 199 199 - cp := compiler.Compile(compiler.Parse(pipeline)) 200 200 - eventJson, err := json.Marshal(cp) 201 201 - if err != nil { 202 202 - return fmt.Errorf("failed to marshal pipeline event: %w", err) 203 203 - } 204 204 - 205 205 - // do not run empty pipelines 206 206 - if cp.Workflows == nil { 207 207 - return nil 208 208 - } 209 209 - 210 210 - ev := db.Event{ 211 211 - Rkey: TID(), 212 212 - Nsid: tangled.PipelineNSID, 213 213 - EventJson: string(eventJson), 214 214 - } 215 215 - 216 216 - return h.db.InsertEvent(ev, h.n) 217 217 - } 218 218 - 219 85 // duplicated from add collaborator 220 86 func (h *Knot) processCollaborator(ctx context.Context, event *models.Event) error { 221 87 raw := json.RawMessage(event.Commit.Record) ··· 338 204 err = h.processPublicKey(ctx, event) 339 205 case tangled.KnotMemberNSID: 340 206 err = h.processKnotMember(ctx, event) 341 341 - case tangled.RepoPullNSID: 342 342 - err = h.processPull(ctx, event) 343 207 case tangled.RepoCollaboratorNSID: 344 208 err = h.processCollaborator(ctx, event) 345 209 }

-109

knotserver/internal.go

··· 23 23 "tangled.org/core/log" 24 24 "tangled.org/core/notifier" 25 25 "tangled.org/core/rbac" 26 26 - "tangled.org/core/workflow" 27 26 ) 28 27 29 28 type InternalHandle struct { ··· 188 187 l.Error("failed to reply with compare link", "err", err, "line", line, "did", gitUserDid, "repo", gitRelativeDir) 189 188 // non-fatal 190 189 } 191 191 - 192 192 - err = h.triggerPipeline(&resp.Messages, line, gitUserDid, repoDid, repoName, pushOptions) 193 193 - if err != nil { 194 194 - l.Error("failed to trigger pipeline", "err", err, "line", line, "did", gitUserDid, "repo", gitRelativeDir) 195 195 - // non-fatal 196 196 - } 197 190 } 198 191 199 192 writeJSON(w, resp) ··· 242 235 } 243 236 244 237 return errors.Join(errs, h.db.InsertEvent(event, h.n)) 245 245 - } 246 246 - 247 247 - func (h *InternalHandle) triggerPipeline( 248 248 - clientMsgs *[]string, 249 249 - line git.PostReceiveLine, 250 250 - gitUserDid string, 251 251 - repoDid string, 252 252 - repoName string, 253 253 - pushOptions PushOptions, 254 254 - ) error { 255 255 - if pushOptions.skipCi { 256 256 - return nil 257 257 - } 258 258 - 259 259 - didSlashRepo, err := securejoin.SecureJoin(repoDid, repoName) 260 260 - if err != nil { 261 261 - return err 262 262 - } 263 263 - 264 264 - repoPath, err := securejoin.SecureJoin(h.c.Repo.ScanPath, didSlashRepo) 265 265 - if err != nil { 266 266 - return err 267 267 - } 268 268 - 269 269 - gr, err := git.Open(repoPath, line.Ref) 270 270 - if err != nil { 271 271 - return err 272 272 - } 273 273 - 274 274 - workflowDir, err := gr.FileTree(context.Background(), workflow.WorkflowDir) 275 275 - if err != nil { 276 276 - return err 277 277 - } 278 278 - 279 279 - var pipeline workflow.RawPipeline 280 280 - for _, e := range workflowDir { 281 281 - if !e.IsFile() { 282 282 - continue 283 283 - } 284 284 - 285 285 - fpath := filepath.Join(workflow.WorkflowDir, e.Name) 286 286 - contents, err := gr.RawContent(fpath) 287 287 - if err != nil { 288 288 - continue 289 289 - } 290 290 - 291 291 - pipeline = append(pipeline, workflow.RawWorkflow{ 292 292 - Name: e.Name, 293 293 - Contents: contents, 294 294 - }) 295 295 - } 296 296 - 297 297 - trigger := tangled.Pipeline_PushTriggerData{ 298 298 - Ref: line.Ref, 299 299 - OldSha: line.OldSha.String(), 300 300 - NewSha: line.NewSha.String(), 301 301 - } 302 302 - 303 303 - compiler := workflow.Compiler{ 304 304 - Trigger: tangled.Pipeline_TriggerMetadata{ 305 305 - Kind: string(workflow.TriggerKindPush), 306 306 - Push: &trigger, 307 307 - Repo: &tangled.Pipeline_TriggerRepo{ 308 308 - Did: repoDid, 309 309 - Knot: h.c.Server.Hostname, 310 310 - Repo: repoName, 311 311 - }, 312 312 - }, 313 313 - } 314 314 - 315 315 - cp := compiler.Compile(compiler.Parse(pipeline)) 316 316 - eventJson, err := json.Marshal(cp) 317 317 - if err != nil { 318 318 - return err 319 319 - } 320 320 - 321 321 - for _, e := range compiler.Diagnostics.Errors { 322 322 - *clientMsgs = append(*clientMsgs, e.String()) 323 323 - } 324 324 - 325 325 - if pushOptions.verboseCi { 326 326 - if compiler.Diagnostics.IsEmpty() { 327 327 - *clientMsgs = append(*clientMsgs, "success: pipeline compiled with no diagnostics") 328 328 - } 329 329 - 330 330 - for _, w := range compiler.Diagnostics.Warnings { 331 331 - *clientMsgs = append(*clientMsgs, w.String()) 332 332 - } 333 333 - } 334 334 - 335 335 - // do not run empty pipelines 336 336 - if cp.Workflows == nil { 337 337 - return nil 338 338 - } 339 339 - 340 340 - event := db.Event{ 341 341 - Rkey: TID(), 342 342 - Nsid: tangled.PipelineNSID, 343 343 - EventJson: string(eventJson), 344 344 - } 345 345 - 346 346 - return h.db.InsertEvent(event, h.n) 347 238 } 348 239 349 240 func (h *InternalHandle) emitCompareLink(

-1

knotserver/server.go

··· 79 79 jc, err := jetstream.NewJetstreamClient(c.Server.JetstreamEndpoint, "knotserver", []string{ 80 80 tangled.PublicKeyNSID, 81 81 tangled.KnotMemberNSID, 82 82 - tangled.RepoPullNSID, 83 82 tangled.RepoCollaboratorNSID, 84 83 }, nil, log.SubLogger(logger, "jetstream"), db, true, c.Server.LogDids) 85 84 if err != nil {

+1 -40

spindle/server.go

··· 325 325 func (s *Spindle) processKnotStream(ctx context.Context, src eventconsumer.Source, msg eventconsumer.Message) error { 326 326 l := log.FromContext(ctx).With("handler", "processKnotStream") 327 327 l = l.With("src", src.Key(), "msg.Nsid", msg.Nsid, "msg.Rkey", msg.Rkey) 328 328 - if msg.Nsid == tangled.PipelineNSID { 329 329 - return nil 330 330 - tpl := tangled.Pipeline{} 331 331 - err := json.Unmarshal(msg.EventJson, &tpl) 332 332 - if err != nil { 333 333 - fmt.Println("error unmarshalling", err) 334 334 - return err 335 335 - } 336 336 - 337 337 - if tpl.TriggerMetadata == nil { 338 338 - return fmt.Errorf("no trigger metadata found") 339 339 - } 340 340 - 341 341 - if tpl.TriggerMetadata.Repo == nil { 342 342 - return fmt.Errorf("no repo data found") 343 343 - } 344 344 - 345 345 - if src.Key() != tpl.TriggerMetadata.Repo.Knot { 346 346 - return fmt.Errorf("repo knot does not match event source: %s != %s", src.Key(), tpl.TriggerMetadata.Repo.Knot) 347 347 - } 348 348 - 349 349 - // filter by repos 350 350 - _, err = s.db.GetRepoWithName( 351 351 - syntax.DID(tpl.TriggerMetadata.Repo.Did), 352 352 - tpl.TriggerMetadata.Repo.Repo, 353 353 - ) 354 354 - if err != nil { 355 355 - return fmt.Errorf("failed to get repo: %w", err) 356 356 - } 357 357 - 358 358 - pipelineId := models.PipelineId{ 359 359 - Knot: src.Key(), 360 360 - Rkey: msg.Rkey, 361 361 - } 362 362 - 363 363 - err = s.processPipeline(ctx, tpl, pipelineId) 364 364 - if err != nil { 365 365 - return err 366 366 - } 367 367 - } else if msg.Nsid == tangled.GitRefUpdateNSID { 328 328 + if msg.Nsid == tangled.GitRefUpdateNSID { 368 329 event := tangled.GitRefUpdate{} 369 330 if err := json.Unmarshal(msg.EventJson, &event); err != nil { 370 331 l.Error("error unmarshalling", "err", err)