vole
Clone
Read only, no credentials:
https://rodent.software/vole.git
Write over SSH with a registered key:
git clone ssh://git@rodent.software:2222/vole.git
A free and open-source forge that lives in your Git repo.
Add event, schedule and manual workflow triggers
eb07ff5696e0d83201215966e3ea1e6a24467581 · Keenan Nemetz · 8d ago · parent 1cde8b9e
Add event, schedule and manual workflow triggers Workflows can now react to change requests, issues, comments, cron schedules and manual runs, and a push trigger can filter by the paths it changed. Branch and tag patterns gain ! exclusion. Event runs are pinned with a JSON payload and the runner exposes it as VOLE_EVENT, VOLE_EVENT_ACTION and VOLE_EVENT_PATH. Change-request runs check out the base tip and carry the head and changed files, so an event never executes an unapproved contributor commit. Workflow dispatch resolves typed inputs and is available from the workflows page and `vole workflow dispatch`. Deduplication now includes an occurrence so the same commit can run for a later event or schedule slot while a push remains idempotent. Part of: #138
server/events.go +634
| @@ -0,0 +1,634 @@ | |||
| 1 | package server | ||
| 2 | |||
| 3 | // Event triggers: change requests, issues, comments, schedules, and manual runs. | ||
| 4 | // | ||
| 5 | // A push is a ref, so it can be reconciled from the refs. An event is a change to | ||
| 6 | // the metadata branch, so it is reconciled from that branch's commits: the cursor is | ||
| 7 | // the last commit whose events were applied, and every commit after it is read in | ||
| 8 | // order. That includes writes made by the browser, which never go through a Git | ||
| 9 | // push, and it keeps the spool a latency optimisation rather than the only path. | ||
| 10 | // | ||
| 11 | // A metadata event queues a run at the trusted branch tip (the change request's base, | ||
| 12 | // or the default branch). A change request's head is described in the payload and | ||
| 13 | // its changed files are listed there, but the head's code is not checked out: an | ||
| 14 | // event run must not be a way to execute an unapproved contributor tip. | ||
| 15 | |||
| 16 | import ( | ||
| 17 | "context" | ||
| 18 | "encoding/json" | ||
| 19 | "fmt" | ||
| 20 | "strings" | ||
| 21 | "time" | ||
| 22 | |||
| 23 | "github.com/rodent-software/vole/ci" | ||
| 24 | "github.com/rodent-software/vole/core/git" | ||
| 25 | "github.com/rodent-software/vole/core/id" | ||
| 26 | "github.com/rodent-software/vole/forge/access" | ||
| 27 | "github.com/rodent-software/vole/forge/repo" | ||
| 28 | "github.com/rodent-software/vole/metadata" | ||
| 29 | ) | ||
| 30 | |||
| 31 | const ( | ||
| 32 | metadataSeconds = 5 * time.Second | ||
| 33 | maxPayloadBody = 16 << 10 | ||
| 34 | ) | ||
| 35 | |||
| 36 | // eventPayload is the JSON a triggered step reads from VOLE_EVENT_PATH. It is | ||
| 37 | // deliberately data: the only place user text reaches a step is this file, not an | ||
| 38 | // interpolated command. | ||
| 39 | type eventPayload struct { | ||
| 40 | Event string `json:"event"` | ||
| 41 | Action string `json:"action"` | ||
| 42 | Repository string `json:"repository"` | ||
| 43 | Ref string `json:"ref,omitempty"` | ||
| 44 | Branch string `json:"branch,omitempty"` | ||
| 45 | Commit string `json:"commit,omitempty"` | ||
| 46 | Sender string `json:"sender,omitempty"` | ||
| 47 | BaseSHA string `json:"base_sha,omitempty"` | ||
| 48 | HeadSHA string `json:"head_sha,omitempty"` | ||
| 49 | ChangedFiles []string `json:"changed_files,omitempty"` | ||
| 50 | Issue *payloadIssue `json:"issue,omitempty"` | ||
| 51 | Change *payloadRequest `json:"change_request,omitempty"` | ||
| 52 | Comment *payloadComment `json:"comment,omitempty"` | ||
| 53 | CommentOn string `json:"comment_on,omitempty"` | ||
| 54 | Inputs map[string]string `json:"inputs,omitempty"` | ||
| 55 | Schedule string `json:"schedule,omitempty"` | ||
| 56 | } | ||
| 57 | |||
| 58 | type payloadIssue struct { | ||
| 59 | ID string `json:"id"` | ||
| 60 | Title string `json:"title"` | ||
| 61 | State string `json:"state"` | ||
| 62 | Author string `json:"author"` | ||
| 63 | Labels []string `json:"labels,omitempty"` | ||
| 64 | Assignees []string `json:"assignees,omitempty"` | ||
| 65 | Body string `json:"body,omitempty"` | ||
| 66 | BodyTruncated bool `json:"body_truncated,omitempty"` | ||
| 67 | URL string `json:"url"` | ||
| 68 | } | ||
| 69 | |||
| 70 | type payloadRequest struct { | ||
| 71 | ID string `json:"id"` | ||
| 72 | Title string `json:"title"` | ||
| 73 | State string `json:"state"` | ||
| 74 | Author string `json:"author"` | ||
| 75 | Base string `json:"base"` | ||
| 76 | Head string `json:"head"` | ||
| 77 | HeadRef string `json:"head_ref,omitempty"` | ||
| 78 | Body string `json:"body,omitempty"` | ||
| 79 | BodyTruncated bool `json:"body_truncated,omitempty"` | ||
| 80 | URL string `json:"url"` | ||
| 81 | } | ||
| 82 | |||
| 83 | type payloadComment struct { | ||
| 84 | ID string `json:"id"` | ||
| 85 | Author string `json:"author"` | ||
| 86 | Body string `json:"body,omitempty"` | ||
| 87 | BodyTruncated bool `json:"body_truncated,omitempty"` | ||
| 88 | ReplyTo string `json:"reply_to,omitempty"` | ||
| 89 | Deleted bool `json:"deleted,omitempty"` | ||
| 90 | URL string `json:"url"` | ||
| 91 | } | ||
| 92 | |||
| 93 | // watchMetadata applies metadata events at startup and then on a timer. A browser | ||
| 94 | // write also pokes the wake channel, so a comment does not wait for the tick. | ||
| 95 | func (s *Server) watchMetadata(ctx context.Context) { | ||
| 96 | s.reconcileMetadata(true) | ||
| 97 | t := time.NewTicker(metadataSeconds) | ||
| 98 | defer t.Stop() | ||
| 99 | for { | ||
| 100 | select { | ||
| 101 | case <-ctx.Done(): | ||
| 102 | return | ||
| 103 | case <-t.C: | ||
| 104 | s.reconcileMetadata(false) | ||
| 105 | case <-s.metadataWake: | ||
| 106 | s.reconcileMetadata(false) | ||
| 107 | } | ||
| 108 | } | ||
| 109 | } | ||
| 110 | |||
| 111 | func (s *Server) wakeMetadata(repo string) { | ||
| 112 | select { | ||
| 113 | case s.metadataWake <- struct{}{}: | ||
| 114 | default: | ||
| 115 | } | ||
| 116 | } | ||
| 117 | |||
| 118 | // reconcileMetadata moves each repository's metadata cursor forward one commit at a | ||
| 119 | // time. A repository whose metadata branch has not been seen yet is baselined: an | ||
| 120 | // instance that starts on an old repository must not replay every issue it ever had. | ||
| 121 | func (s *Server) reconcileMetadata(baseline bool) { | ||
| 122 | repos, err := s.Repos.List() | ||
| 123 | if err != nil { | ||
| 124 | s.Log.Warn("ci: cannot list repositories for metadata events", "err", err) | ||
| 125 | return | ||
| 126 | } | ||
| 127 | now := time.Now().UTC() | ||
| 128 | for _, m := range repos { | ||
| 129 | tip, err := m.Git().RevParse(metadata.Ref) | ||
| 130 | if err != nil || tip == "" { | ||
| 131 | continue | ||
| 132 | } | ||
| 133 | cursors, err := s.Runs.Cursors(m.Name) | ||
| 134 | if err != nil { | ||
| 135 | s.Log.Warn("ci: cannot read metadata cursor", "repo", m.Name, "err", err) | ||
| 136 | continue | ||
| 137 | } | ||
| 138 | old, seen := cursors[metadata.Ref] | ||
| 139 | if !seen { | ||
| 140 | if baseline { | ||
| 141 | // Start from here rather than firing every event the repository | ||
| 142 | // accumulated before this instance first looked. A branch created later | ||
| 143 | // has no earlier events to lose. | ||
| 144 | if err := s.Runs.SetCursor(m.Name, metadata.Ref, tip, now); err != nil { | ||
| 145 | s.Log.Warn("ci: cannot baseline the metadata cursor", "repo", m.Name, "err", err) | ||
| 146 | } | ||
| 147 | continue | ||
| 148 | } | ||
| 149 | old = "" | ||
| 150 | } | ||
| 151 | if old == tip { | ||
| 152 | continue | ||
| 153 | } | ||
| 154 | if old != "" { | ||
| 155 | ancestor, err := m.Git().IsAncestor(old, tip) | ||
| 156 | if err != nil || !ancestor { | ||
| 157 | // A rewritten metadata branch is not a sequence of events to replay. | ||
| 158 | if err := s.Runs.SetCursor(m.Name, metadata.Ref, tip, now); err != nil { | ||
| 159 | s.Log.Warn("ci: cannot reset the metadata cursor", "repo", m.Name, "err", err) | ||
| 160 | } | ||
| 161 | continue | ||
| 162 | } | ||
| 163 | } | ||
| 164 | commits, err := m.Git().LogRange(old, tip) | ||
| 165 | if err != nil { | ||
| 166 | s.Log.Warn("ci: cannot read metadata commits", "repo", m.Name, "err", err) | ||
| 167 | continue | ||
| 168 | } | ||
| 169 | for _, c := range commits { | ||
| 170 | s.processMetadataCommit(m, c) | ||
| 171 | if err := s.Runs.SetCursor(m.Name, metadata.Ref, c.SHA, now); err != nil { | ||
| 172 | s.Log.Warn("ci: cannot record the metadata cursor", "repo", m.Name, "commit", c.SHA, "err", err) | ||
| 173 | } | ||
| 174 | } | ||
| 175 | } | ||
| 176 | } | ||
| 177 | |||
| 178 | // processMetadataRange applies the commits of one metadata push immediately, so a | ||
| 179 | // CLI comment does not wait for the reconciler's tick. The cursor is left to the | ||
| 180 | // reconciler: this is an optimisation, and the reconciler is what makes the range | ||
| 181 | // complete when a spool entry was never written. | ||
| 182 | func (s *Server) processMetadataRange(repoName, old, next string) { | ||
| 183 | if next == "" || next == git.ZeroSHA { | ||
| 184 | return | ||
| 185 | } | ||
| 186 | m, err := s.Repos.Open(repoName) | ||
| 187 | if err != nil { | ||
| 188 | return | ||
| 189 | } | ||
| 190 | if old != "" && old != git.ZeroSHA { | ||
| 191 | ancestor, err := m.Git().IsAncestor(old, next) | ||
| 192 | if err != nil || !ancestor { | ||
| 193 | return | ||
| 194 | } | ||
| 195 | } | ||
| 196 | commits, err := m.Git().LogRange(old, next) | ||
| 197 | if err != nil { | ||
| 198 | s.Log.Warn("ci: cannot read metadata commits from the spool", "repo", repoName, "err", err) | ||
| 199 | return | ||
| 200 | } | ||
| 201 | for _, c := range commits { | ||
| 202 | s.processMetadataCommit(m, c) | ||
| 203 | } | ||
| 204 | } | ||
| 205 | |||
| 206 | func (s *Server) processMetadataCommit(m *repo.Meta, c git.Commit) { | ||
| 207 | parents, err := m.Git().CommitParents(c.SHA) | ||
| 208 | if err != nil { | ||
| 209 | s.Log.Warn("ci: cannot read metadata commit parents", "repo", m.Name, "commit", c.SHA, "err", err) | ||
| 210 | return | ||
| 211 | } | ||
| 212 | old := "" | ||
| 213 | if len(parents) > 0 { | ||
| 214 | old = parents[0] | ||
| 215 | } | ||
| 216 | changed, err := changedPaths(m.Git(), old, c.SHA) | ||
| 217 | if err != nil { | ||
| 218 | s.Log.Warn("ci: cannot read metadata changes", "repo", m.Name, "commit", c.SHA, "err", err) | ||
| 219 | return | ||
| 220 | } | ||
| 221 | acl, _ := m.ACL() | ||
| 222 | seen := map[string]bool{} | ||
| 223 | for _, path := range changed { | ||
| 224 | issueID, commentID, kind, ok := metadataPathIDs(path) | ||
| 225 | if !ok { | ||
| 226 | continue | ||
| 227 | } | ||
| 228 | where := c.SHA + ":" + kind + ":" + issueID + ":" + commentID | ||
| 229 | if seen[where] { | ||
| 230 | continue | ||
| 231 | } | ||
| 232 | seen[where] = true | ||
| 233 | s.processMetadataPath(m, old, c, acl, issueID, commentID, kind) | ||
| 234 | } | ||
| 235 | } | ||
| 236 | |||
| 237 | // metadataPathIDs names the entity a metadata path belongs to. The paths are the | ||
| 238 | // public format, so this is the one place the server decides what an issue, a | ||
| 239 | // request, an approval and a comment look like on disk. | ||
| 240 | func metadataPathIDs(path string) (entityID, commentID, kind string, ok bool) { | ||
| 241 | parts := strings.Split(path, "/") | ||
| 242 | switch { | ||
| 243 | case len(parts) == 3 && parts[0] == metadata.IssuesDir && parts[2] == "issue.md": | ||
| 244 | if id.Valid(parts[1]) { | ||
| 245 | return parts[1], "", "issue", true | ||
| 246 | } | ||
| 247 | case len(parts) == 4 && parts[0] == metadata.IssuesDir && parts[2] == "comments" && | ||
| 248 | strings.HasSuffix(parts[3], ".md"): | ||
| 249 | cid := strings.TrimSuffix(parts[3], ".md") | ||
| 250 | if id.Valid(parts[1]) && id.Valid(cid) { | ||
| 251 | return parts[1], cid, "issue_comment", true | ||
| 252 | } | ||
| 253 | case len(parts) == 3 && parts[0] == metadata.RequestsDir && parts[2] == "request.md": | ||
| 254 | if id.Valid(parts[1]) { | ||
| 255 | return parts[1], "", "request", true | ||
| 256 | } | ||
| 257 | case len(parts) == 4 && parts[0] == metadata.RequestsDir && parts[2] == "comments" && | ||
| 258 | strings.HasSuffix(parts[3], ".md"): | ||
| 259 | cid := strings.TrimSuffix(parts[3], ".md") | ||
| 260 | if id.Valid(parts[1]) && id.Valid(cid) { | ||
| 261 | return parts[1], cid, "request_comment", true | ||
| 262 | } | ||
| 263 | case len(parts) == 4 && parts[0] == metadata.RequestsDir && parts[2] == "approvals": | ||
| 264 | if id.Valid(parts[1]) { | ||
| 265 | return parts[1], "", "approval", true | ||
| 266 | } | ||
| 267 | } | ||
| 268 | return "", "", "", false | ||
| 269 | } | ||
| 270 | |||
| 271 | func (s *Server) processMetadataPath(m *repo.Meta, old string, c git.Commit, acl *access.ACL, entityID, commentID, kind string) { | ||
| 272 | switch kind { | ||
| 273 | case "issue": | ||
| 274 | s.issueEvent(m, old, c, entityID) | ||
| 275 | case "issue_comment": | ||
| 276 | s.commentEvent(m, old, c, entityID, commentID, false) | ||
| 277 | case "request_comment": | ||
| 278 | s.commentEvent(m, old, c, entityID, commentID, true) | ||
| 279 | case "request", "approval": | ||
| 280 | s.changeRequestEvent(m, old, c, acl, entityID, kind == "request") | ||
| 281 | } | ||
| 282 | } | ||
| 283 | |||
| 284 | func (s *Server) issueEvent(m *repo.Meta, old string, c git.Commit, issueID string) { | ||
| 285 | oldIssue, _, _ := metadata.LoadIssueAtCommit(m.Git(), old, issueID) | ||
| 286 | newIssue, exists, err := metadata.LoadIssueAtCommit(m.Git(), c.SHA, issueID) | ||
| 287 | if err != nil || !exists { | ||
| 288 | return | ||
| 289 | } | ||
| 290 | var action string | ||
| 291 | switch { | ||
| 292 | case oldIssue == nil: | ||
| 293 | action = "opened" | ||
| 294 | case oldIssue.State != newIssue.State: | ||
| 295 | if newIssue.State == "closed" { | ||
| 296 | action = "closed" | ||
| 297 | } else { | ||
| 298 | action = "reopened" | ||
| 299 | } | ||
| 300 | default: | ||
| 301 | action = "edited" | ||
| 302 | } | ||
| 303 | ref, branch, commit, err := s.defaultBranchCommit(m) | ||
| 304 | if err != nil { | ||
| 305 | s.Log.Warn("ci: cannot resolve the default branch for an issue event", "repo", m.Name, "err", err) | ||
| 306 | return | ||
| 307 | } | ||
| 308 | body, clipped := clipBody(newIssue.Description) | ||
| 309 | payload := eventPayload{ | ||
| 310 | Event: "issue", Action: action, Repository: m.Name, Ref: ref, Branch: branch, | ||
| 311 | Commit: commit, Sender: c.Author, | ||
| 312 | Issue: &payloadIssue{ | ||
| 313 | ID: newIssue.ID, Title: newIssue.Title, State: newIssue.State, Author: newIssue.Author, | ||
| 314 | Labels: newIssue.Labels, Assignees: newIssue.Assignees, | ||
| 315 | Body: body, BodyTruncated: clipped, | ||
| 316 | URL: "/r/" + m.Name + "/issues/" + newIssue.ID, | ||
| 317 | }, | ||
| 318 | } | ||
| 319 | s.queuePayload(m, "issue", action, ref, branch, "", commit, payload, c.SHA+":issue:"+action+":"+issueID) | ||
| 320 | } | ||
| 321 | |||
| 322 | func (s *Server) commentEvent(m *repo.Meta, old string, c git.Commit, entityID, commentID string, request bool) { | ||
| 323 | var server string | ||
| 324 | if request { | ||
| 325 | server = "request" | ||
| 326 | } else { | ||
| 327 | server = "issue" | ||
| 328 | } | ||
| 329 | oldComment := findComment(m, old, entityID, commentID, request) | ||
| 330 | newComment := findComment(m, c.SHA, entityID, commentID, request) | ||
| 331 | if oldComment == nil && newComment == nil { | ||
| 332 | return | ||
| 333 | } | ||
| 334 | action := "edited" | ||
| 335 | switch { | ||
| 336 | case oldComment == nil: | ||
| 337 | action = "created" | ||
| 338 | case newComment == nil || newComment.Deleted && !oldComment.Deleted: | ||
| 339 | action = "deleted" | ||
| 340 | } | ||
| 341 | shown := newComment | ||
| 342 | if shown == nil { | ||
| 343 | shown = oldComment | ||
| 344 | } | ||
| 345 | ref, branch, commit, err := s.defaultBranchCommit(m) | ||
| 346 | if err != nil { | ||
| 347 | s.Log.Warn("ci: cannot resolve the default branch for a comment event", "repo", m.Name, "err", err) | ||
| 348 | return | ||
| 349 | } | ||
| 350 | var issue *payloadIssue | ||
| 351 | var change *payloadRequest | ||
| 352 | var url string | ||
| 353 | if request { | ||
| 354 | if req, ok, _ := metadata.LoadRequestAt(m.Git(), c.SHA, entityID); ok && req != nil { | ||
| 355 | url = "/r/" + m.Name + "/requests/" + req.ID | ||
| 356 | body, clipped := clipBody(req.Description) | ||
| 357 | change = &payloadRequest{ID: req.ID, Title: req.Title, State: req.State, Author: req.Author, | ||
| 358 | Base: req.Base, Head: req.Head, HeadRef: req.HeadRef, Body: body, | ||
| 359 | BodyTruncated: clipped, URL: url} | ||
| 360 | } | ||
| 361 | } else if iss, ok, _ := metadata.LoadIssueAtCommit(m.Git(), c.SHA, entityID); ok && iss != nil { | ||
| 362 | url = "/r/" + m.Name + "/issues/" + iss.ID | ||
| 363 | body, clipped := clipBody(iss.Description) | ||
| 364 | issue = &payloadIssue{ID: iss.ID, Title: iss.Title, State: iss.State, Author: iss.Author, | ||
| 365 | Labels: iss.Labels, Assignees: iss.Assignees, Body: body, BodyTruncated: clipped, URL: url} | ||
| 366 | } | ||
| 367 | body, clipped := clipBody(shown.Body) | ||
| 368 | payload := eventPayload{ | ||
| 369 | Event: "comment", Action: action, Repository: m.Name, Ref: ref, Branch: branch, | ||
| 370 | Commit: commit, Sender: c.Author, CommentOn: server, Issue: issue, Change: change, | ||
| 371 | Comment: &payloadComment{ID: shown.ID, Author: shown.Author, Body: body, | ||
| 372 | BodyTruncated: clipped, ReplyTo: shown.ReplyTo, Deleted: shown.Deleted, | ||
| 373 | URL: url + "#c-" + shown.ID}, | ||
| 374 | } | ||
| 375 | s.queuePayload(m, "comment", action, ref, branch, "", commit, payload, | ||
| 376 | c.SHA+":comment:"+action+":"+entityID+":"+commentID) | ||
| 377 | } | ||
| 378 | |||
| 379 | func findComment(m *repo.Meta, commit, entityID, commentID string, request bool) *metadata.Comment { | ||
| 380 | if commit == "" { | ||
| 381 | return nil | ||
| 382 | } | ||
| 383 | if request { | ||
| 384 | req, ok, _ := metadata.LoadRequestAt(m.Git(), commit, entityID) | ||
| 385 | if !ok || req == nil { | ||
| 386 | return nil | ||
| 387 | } | ||
| 388 | for i := range req.Comments { | ||
| 389 | if req.Comments[i].ID == commentID { | ||
| 390 | return &req.Comments[i] | ||
| 391 | } | ||
| 392 | } | ||
| 393 | return nil | ||
| 394 | } | ||
| 395 | iss, ok, _ := metadata.LoadIssueAtCommit(m.Git(), commit, entityID) | ||
| 396 | if !ok || iss == nil { | ||
| 397 | return nil | ||
| 398 | } | ||
| 399 | for i := range iss.Comments { | ||
| 400 | if iss.Comments[i].ID == commentID { | ||
| 401 | return &iss.Comments[i] | ||
| 402 | } | ||
| 403 | } | ||
| 404 | return nil | ||
| 405 | } | ||
| 406 | |||
| 407 | func (s *Server) changeRequestEvent(m *repo.Meta, old string, c git.Commit, acl *access.ACL, requestID string, requestFile bool) { | ||
| 408 | oldReq, _, _ := metadata.LoadRequestAt(m.Git(), old, requestID) | ||
| 409 | newReq, exists, err := metadata.LoadRequestAt(m.Git(), c.SHA, requestID) | ||
| 410 | if err != nil || !exists || newReq == nil { | ||
| 411 | return | ||
| 412 | } | ||
| 413 | baseRef := "refs/heads/" + newReq.Base | ||
| 414 | base, err := m.Git().RevParse(baseRef) | ||
| 415 | if err != nil || base == "" { | ||
| 416 | s.Log.Warn("ci: cannot resolve a change's base for an event", "repo", m.Name, "request", requestID, "base", newReq.Base, "err", err) | ||
| 417 | return | ||
| 418 | } | ||
| 419 | changed, err := changedPaths(m.Git(), base, newReq.Head) | ||
| 420 | if err != nil { | ||
| 421 | s.Log.Warn("ci: cannot read a change's files for an event", "repo", m.Name, "request", requestID, "err", err) | ||
| 422 | return | ||
| 423 | } | ||
| 424 | body, clipped := clipBody(newReq.Description) | ||
| 425 | payload := eventPayload{ | ||
| 426 | Event: "change_request", Repository: m.Name, Ref: baseRef, Branch: newReq.Base, | ||
| 427 | Commit: base, Sender: c.Author, BaseSHA: base, HeadSHA: newReq.Head, | ||
| 428 | ChangedFiles: changed, | ||
| 429 | Change: &payloadRequest{ID: newReq.ID, Title: newReq.Title, State: newReq.State, | ||
| 430 | Author: newReq.Author, Base: newReq.Base, Head: newReq.Head, HeadRef: newReq.HeadRef, | ||
| 431 | Body: body, BodyTruncated: clipped, URL: "/r/" + m.Name + "/requests/" + newReq.ID}, | ||
| 432 | } | ||
| 433 | actions := []string{} | ||
| 434 | switch { | ||
| 435 | case oldReq == nil: | ||
| 436 | actions = append(actions, "opened") | ||
| 437 | case oldReq.State != newReq.State: | ||
| 438 | switch newReq.State { | ||
| 439 | case "merged": | ||
| 440 | actions = append(actions, "merged") | ||
| 441 | case "closed": | ||
| 442 | actions = append(actions, "closed") | ||
| 443 | case "open": | ||
| 444 | actions = append(actions, "updated") | ||
| 445 | } | ||
| 446 | case requestFile && (oldReq.Head != newReq.Head || oldReq.HeadRef != newReq.HeadRef || | ||
| 447 | oldReq.Base != newReq.Base || oldReq.Title != newReq.Title || | ||
| 448 | oldReq.Description != newReq.Description): | ||
| 449 | actions = append(actions, "updated") | ||
| 450 | } | ||
| 451 | required := 0 | ||
| 452 | if acl != nil { | ||
| 453 | required = acl.RequiredApprovals() | ||
| 454 | } | ||
| 455 | if newReq.State == "open" && required > 0 && acl != nil { | ||
| 456 | nowReady := newReq.CurrentApprovalsBy(acl.EligibleApprover) >= required | ||
| 457 | wasReady := oldReq != nil && oldReq.State == "open" && | ||
| 458 | oldReq.CurrentApprovalsBy(acl.EligibleApprover) >= required | ||
| 459 | if nowReady && !wasReady { | ||
| 460 | actions = append(actions, "ready") | ||
| 461 | } | ||
| 462 | } | ||
| 463 | for _, action := range actions { | ||
| 464 | s.queuePayload(m, "change_request", action, baseRef, newReq.Base, "", base, payload, | ||
| 465 | c.SHA+":change_request:"+action+":"+requestID) | ||
| 466 | } | ||
| 467 | } | ||
| 468 | |||
| 469 | func (s *Server) defaultBranchCommit(m *repo.Meta) (ref, branch, commit string, err error) { | ||
| 470 | branch = strings.TrimSpace(m.DefaultBranch) | ||
| 471 | if branch == "" { | ||
| 472 | branch = "main" | ||
| 473 | } | ||
| 474 | ref = "refs/heads/" + branch | ||
| 475 | commit, err = m.Git().RevParse(ref) | ||
| 476 | if err != nil { | ||
| 477 | return "", "", "", err | ||
| 478 | } | ||
| 479 | if commit == "" { | ||
| 480 | return "", "", "", fmt.Errorf("%s has no commits", ref) | ||
| 481 | } | ||
| 482 | return ref, branch, commit, nil | ||
| 483 | } | ||
| 484 | |||
| 485 | func (s *Server) queuePayload(m *repo.Meta, trigger, action, ref, branch, tag, commit string, payload eventPayload, occurrence string) { | ||
| 486 | payload.Event = trigger | ||
| 487 | payload.Action = action | ||
| 488 | body, err := json.Marshal(payload) | ||
| 489 | if err != nil { | ||
| 490 | s.Log.Warn("ci: cannot build an event payload", "repo", m.Name, "trigger", trigger, "err", err) | ||
| 491 | return | ||
| 492 | } | ||
| 493 | if len(body) > ci.MaxEventPayloadBytes { | ||
| 494 | s.Log.Warn("ci: event payload is too large", "repo", m.Name, "trigger", trigger, | ||
| 495 | "bytes", len(body), "limit", ci.MaxEventPayloadBytes) | ||
| 496 | return | ||
| 497 | } | ||
| 498 | n := s.queue(m, m.Name, queueSpec{ | ||
| 499 | ref: ref, branch: branch, tag: tag, commit: commit, | ||
| 500 | trigger: trigger, action: action, payload: string(body), occurrence: occurrence, | ||
| 501 | }) | ||
| 502 | if n > 0 { | ||
| 503 | s.Log.Info("ci: queued an event run", "repo", m.Name, "trigger", trigger, "action", action, "runs", n) | ||
| 504 | } | ||
| 505 | } | ||
| 506 | |||
| 507 | func clipBody(body string) (string, bool) { | ||
| 508 | if len(body) <= maxPayloadBody { | ||
| 509 | return body, false | ||
| 510 | } | ||
| 511 | return body[:maxPayloadBody], true | ||
| 512 | } | ||
| 513 | |||
| 514 | // watchSchedules checks each repository's cron entries once a minute. The occurrence | ||
| 515 | // is the minute itself, so a second check inside the same minute is a no-op and a | ||
| 516 | // server that was down simply misses that slot rather than replaying it. | ||
| 517 | func (s *Server) watchSchedules(ctx context.Context) { | ||
| 518 | t := time.NewTicker(20 * time.Second) | ||
| 519 | defer t.Stop() | ||
| 520 | for { | ||
| 521 | select { | ||
| 522 | case <-ctx.Done(): | ||
| 523 | return | ||
| 524 | case <-t.C: | ||
| 525 | s.runSchedules(time.Now().UTC()) | ||
| 526 | } | ||
| 527 | } | ||
| 528 | } | ||
| 529 | |||
| 530 | func (s *Server) runSchedules(now time.Time) { | ||
| 531 | slot := now.Truncate(time.Minute) | ||
| 532 | repos, err := s.Repos.List() | ||
| 533 | if err != nil { | ||
| 534 | s.Log.Warn("ci: cannot list repositories for schedules", "err", err) | ||
| 535 | return | ||
| 536 | } | ||
| 537 | for _, m := range repos { | ||
| 538 | ref, branch, commit, err := s.defaultBranchCommit(m) | ||
| 539 | if err != nil { | ||
| 540 | continue | ||
| 541 | } | ||
| 542 | wfs, problems := ci.At(m.Git(), commit) | ||
| 543 | if len(problems) > 0 { | ||
| 544 | continue | ||
| 545 | } | ||
| 546 | for _, wf := range wfs { | ||
| 547 | for _, schedule := range wf.Schedules { | ||
| 548 | cron, err := ci.ParseCron(schedule.Cron) | ||
| 549 | if err != nil || !cron.Matches(slot) { | ||
| 550 | continue | ||
| 551 | } | ||
| 552 | payload := eventPayload{ | ||
| 553 | Event: "schedule", Action: "scheduled", Repository: m.Name, | ||
| 554 | Ref: ref, Branch: branch, Commit: commit, Sender: "schedule", | ||
| 555 | Schedule: schedule.Cron, | ||
| 556 | } | ||
| 557 | s.queue(m, m.Name, queueSpec{ | ||
| 558 | ref: ref, branch: branch, commit: commit, deployPages: true, | ||
| 559 | trigger: "schedule", action: "scheduled", payload: mustJSON(payload), | ||
| 560 | occurrence: "schedule:" + slot.Format(time.RFC3339), | ||
| 561 | }) | ||
| 562 | } | ||
| 563 | } | ||
| 564 | } | ||
| 565 | } | ||
| 566 | |||
| 567 | // dispatchWorkflow queues one manual run of a workflow that declared | ||
| 568 | // workflow-dispatch. The caller has already checked the maintainer capability; this | ||
| 569 | // re-checks it so the SSH and browser surfaces cannot disagree. | ||
| 570 | func (s *Server) dispatchWorkflow(repoName, workflowPath, refName string, inputs map[string]string, actor string) (int, error) { | ||
| 571 | m, err := s.Repos.Open(repoName) | ||
| 572 | if err != nil { | ||
| 573 | return 0, err | ||
| 574 | } | ||
| 575 | acl, err := m.ACL() | ||
| 576 | if err != nil || acl == nil { | ||
| 577 | return 0, fmt.Errorf("cannot read access rules for %s", repoName) | ||
| 578 | } | ||
| 579 | if !acl.CanTriage(actor) && !(s.Identity != nil && s.Identity.IsAdmin(actor)) { | ||
| 580 | return 0, fmt.Errorf("%s may not run workflows in %s", actor, repoName) | ||
| 581 | } | ||
| 582 | refName = strings.TrimPrefix(strings.TrimSpace(refName), "refs/heads/") | ||
| 583 | if refName == "" { | ||
| 584 | refName = m.DefaultBranch | ||
| 585 | } | ||
| 586 | if refName == "" || !m.Git().ValidBranch(refName) { | ||
| 587 | return 0, fmt.Errorf("%s is not a branch in %s", refName, repoName) | ||
| 588 | } | ||
| 589 | ref := "refs/heads/" + refName | ||
| 590 | commit, err := m.Git().RevParse(ref) | ||
| 591 | if err != nil || commit == "" { | ||
| 592 | return 0, fmt.Errorf("%s does not exist", ref) | ||
| 593 | } | ||
| 594 | wfs, problems := ci.At(m.Git(), commit) | ||
| 595 | if len(problems) > 0 { | ||
| 596 | return 0, fmt.Errorf("the workflows at %s do not validate", refName) | ||
| 597 | } | ||
| 598 | var wf *ci.Workflow | ||
| 599 | for _, candidate := range wfs { | ||
| 600 | if candidate.Path == workflowPath { | ||
| 601 | wf = candidate | ||
| 602 | break | ||
| 603 | } | ||
| 604 | } | ||
| 605 | if wf == nil { | ||
| 606 | return 0, fmt.Errorf("%s is not a workflow at %s", workflowPath, refName) | ||
| 607 | } | ||
| 608 | if wf.Dispatch == nil { | ||
| 609 | return 0, fmt.Errorf("%s does not declare workflow-dispatch", workflowPath) | ||
| 610 | } | ||
| 611 | resolved, problems := wf.ResolveDispatchInputs(inputs) | ||
| 612 | if len(problems) > 0 { | ||
| 613 | return 0, fmt.Errorf("%s", strings.Join(problems, "; ")) | ||
| 614 | } | ||
| 615 | runID, err := id.New(time.Now()) | ||
| 616 | if err != nil { | ||
| 617 | return 0, err | ||
| 618 | } | ||
| 619 | payload := eventPayload{ | ||
| 620 | Event: "workflow_dispatch", Action: "dispatched", Repository: m.Name, | ||
| 621 | Ref: ref, Branch: refName, Commit: commit, Sender: actor, Inputs: resolved, | ||
| 622 | } | ||
| 623 | n := s.queue(m, m.Name, queueSpec{ | ||
| 624 | ref: ref, branch: refName, commit: commit, deployPages: true, | ||
| 625 | trigger: "workflow_dispatch", action: "dispatched", payload: mustJSON(payload), | ||
| 626 | occurrence: "dispatch:" + runID, | ||
| 627 | }) | ||
| 628 | return n, nil | ||
| 629 | } | ||
| 630 | |||
| 631 | func mustJSON(payload eventPayload) string { | ||
| 632 | body, _ := json.Marshal(payload) | ||
| 633 | return string(body) | ||
| 634 | } | ||