1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package ports
4
5import (
6 "context"
7 "errors"
8 "strconv"
9 "time"
10
11 "bnema.dev/ermite/internal/domain"
12)
13
14// CIRunDraft is what CreateRun stores: a run with its source, its jobs and
15// their checks, built once the store has chosen the ids.
16type CIRunDraft struct {
17 Run domain.CIRun
18 Source domain.CIRunSource
19 Jobs []domain.CIJob
20 Checks []domain.CICheck
21}
22
23// CIWrite is a set of records written in one atomic batch. A run or a job with
24// Version 0 must not exist yet; any other must be unchanged since it was read.
25// Checks are derived from the jobs and upserted without a precondition.
26// Apply returns domain.ErrConflict otherwise and domain.ErrNotFound when the
27// repository no longer exists.
28type CIWrite struct {
29 Runs []domain.CIRun
30 Jobs []domain.CIJob
31 Checks []domain.CICheck
32}
33
34// CIStore keeps CI runs, jobs and checks. The store derives its own indexes
35// (the queue of ready jobs, the active jobs, the open runs, the job tokens)
36// from the records it writes.
37type CIStore interface {
38 // CreateRun takes the next run id and the next number of the workflow in
39 // the repository (both never reused), calls build with them and stores
40 // the draft it returns, in one batch that fails with domain.ErrNotFound
41 // once the repository is deleted. The records come back with their
42 // versions.
43 CreateRun(ctx context.Context, repo domain.RepoName, workflowPath string, build func(id, number int64) CIRunDraft) (CIRunDraft, error)
44 // Apply writes the records in one batch and returns the version they all
45 // have afterwards.
46 Apply(ctx context.Context, w CIWrite) (version uint64, err error)
47 // GetRun returns the run, or domain.ErrNotFound.
48 GetRun(ctx context.Context, repo domain.RepoName, id int64) (domain.CIRun, error)
49 // GetRunSource returns what the run was planned from, or
50 // domain.ErrNotFound. It is how a run is found from its id alone.
51 GetRunSource(ctx context.Context, runID int64) (domain.CIRunSource, error)
52 // ListRuns returns the runs of the repository with an id below before (0
53 // for no bound), newest first, at most limit.
54 ListRuns(ctx context.Context, repo domain.RepoName, before int64, limit int) ([]domain.CIRun, error)
55 // RunIDs returns the ids of the runs of the repository.
56 RunIDs(ctx context.Context, repo domain.RepoName) ([]int64, error)
57 // ListOpenRuns returns the runs that have not completed.
58 ListOpenRuns(ctx context.Context) ([]domain.CIRun, error)
59 // GetJob returns a job of a run at an attempt of the run, or
60 // domain.ErrNotFound.
61 GetJob(ctx context.Context, runID int64, runAttempt int32, jobID string) (domain.CIJob, error)
62 // ListJobs returns the jobs of a run at an attempt, ordered by index then
63 // combination.
64 ListJobs(ctx context.Context, runID int64, runAttempt int32) ([]domain.CIJob, error)
65 // FindJobByToken returns the job whose current token has that hash and
66 // whose token is still valid at now, or domain.ErrNotFound.
67 FindJobByToken(ctx context.Context, hash domain.TokenHash, now time.Time) (domain.CIJob, error)
68 // ListReadyJobs returns the jobs that wait for a runner, oldest first, at
69 // most limit.
70 ListReadyJobs(ctx context.Context, limit int) ([]domain.CIJob, error)
71 // ListActiveJobs returns the jobs given to a runner and not over.
72 ListActiveJobs(ctx context.Context) ([]domain.CIJob, error)
73 // ListChecks returns the checks of a commit, at most limit; more is true
74 // when there are more.
75 ListChecks(ctx context.Context, repo domain.RepoName, sha string, limit int) (checks []domain.CICheck, more bool, err error)
76 // Sweep removes the runs completed before the time, with their jobs,
77 // checks and sources (not their numbers), and returns their ids so that
78 // the logs can go too.
79 Sweep(ctx context.Context, before time.Time) ([]int64, error)
80 // CreateRunner stores the runner and the index of its token. A name in use
81 // is domain.ErrAlreadyExists, or with replace the old runner is removed in
82 // the same batch and returned.
83 CreateRunner(ctx context.Context, r domain.CIRunner, replace bool) (removed []domain.CIRunner, err error)
84 // GetRunner returns the runner, or domain.ErrNotFound.
85 GetRunner(ctx context.Context, id string) (domain.CIRunner, error)
86 // FindRunnerByToken returns the runner whose token has that hash, or
87 // domain.ErrNotFound.
88 FindRunnerByToken(ctx context.Context, hash domain.TokenHash) (domain.CIRunner, error)
89 // ListRunners returns every runner ordered by name.
90 ListRunners(ctx context.Context) ([]domain.CIRunner, error)
91 // RemoveRunner removes the runner and its token, or returns
92 // domain.ErrNotFound.
93 RemoveRunner(ctx context.Context, id string) (domain.CIRunner, error)
94}
95
96// CIRunCleaner is what repository deletion needs from CI: the runs of the
97// repository before its records go, and the removal of their logs after.
98// *usecase.CI implements it.
99type CIRunCleaner interface {
100 RunIDs(ctx context.Context, repo domain.RepoName) ([]int64, error)
101 RemoveLogs(ctx context.Context, runIDs []int64) error
102}
103
104// CIChecksReader is the read side of the checks, for the merge rule and the
105// pages that show them. *usecase.CI implements it.
106type CIChecksReader interface {
107 // CommitChecks returns the checks of a commit, without access control (the
108 // caller has checked it). More than 1,000 is domain.ErrResourceExhausted,
109 // not a part of them: a rule must not decide on some of the checks.
110 CommitChecks(ctx context.Context, repo domain.RepoName, sha string) ([]domain.CICheck, error)
111}
112
113// CILogRef names the log of one execution of a job.
114type CILogRef struct {
115 RunID int64
116 JobID string
117 Attempt int32
118}
119
120// ErrCILogLimit is returned by CILogStore.Append when the lines do not fit in
121// the log limit; nothing was appended.
122var ErrCILogLimit = errors.New("log limit reached")
123
124// ErrCILogGap wraps domain.ErrOutOfRange when an offset is above the count.
125type ErrCILogGap struct{ Count int64 }
126
127// Error implements error.
128func (e *ErrCILogGap) Error() string {
129 return "out of range: the log holds lines up to the offset of " + strconv.FormatInt(e.Count, 10)
130}
131
132// Is makes errors.Is(err, domain.ErrOutOfRange) true.
133func (e *ErrCILogGap) Is(target error) bool { return target == domain.ErrOutOfRange }
134
135// CILogStore keeps the logs of the jobs: append-only files, one line per log
136// line, never in the metadata store. Lines are numbered from 0.
137type CILogStore interface {
138 // Append appends the lines at offset (the number of the first line), with
139 // the idempotence of AppendLog: lines already stored are skipped, a gap is
140 // an *ErrCILogGap. It returns the number of lines after the call and how
141 // many of the request were new. A log that would exceed limit bytes gets
142 // nothing and ErrCILogLimit.
143 Append(ctx context.Context, ref CILogRef, offset int64, lines []domain.CILogLine, limit int64) (next int64, added int, err error)
144 // AppendForced appends lines whatever the limit (the last line of a log
145 // that reached it).
146 AppendForced(ctx context.Context, ref CILogRef, lines []domain.CILogLine) (next int64, err error)
147 // Count returns the number of lines held, 0 for a log that does not exist.
148 Count(ctx context.Context, ref CILogRef) (int64, error)
149 // Read returns lines from the one numbered from, at most maxLines lines and
150 // about maxBytes of text. It returns domain.ErrNotFound when the log does
151 // not exist and domain.ErrOutOfRange when from is above the count.
152 Read(ctx context.Context, ref CILogRef, from int64, maxLines, maxBytes int) ([]domain.CILogLine, error)
153 // Changed returns a channel that is closed at the next append to the log.
154 Changed(ref CILogRef) <-chan struct{}
155 // Reset empties the log.
156 Reset(ctx context.Context, ref CILogRef) error
157 // RemoveRuns deletes the logs of the runs.
158 RemoveRuns(ctx context.Context, runIDs []int64) error
159 // RunIDs lists the runs that have a log directory, for the orphan sweep.
160 RunIDs(ctx context.Context) ([]int64, error)
161}
162
163// WorkflowTrigger starts the workflows that match pushed refs. OnPush only
164// queues the work and returns: planning happens in the background and its
165// errors never reach the push.
166type WorkflowTrigger interface {
167 OnPush(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, pusher domain.Principal)
168}
169
170// CIGit is what planning needs from Git beyond RepoReader and MergeGit.
171type CIGit interface {
172 // ChangedPaths returns the paths changed from base to head (a zero base
173 // means the empty tree), in the order of the diff and without renames
174 // detection, at most max; truncated is true when there were more.
175 ChangedPaths(ctx context.Context, repo domain.RepoName, base, head domain.ObjectID, max int) (paths []string, truncated bool, err error)
176 // CommitsOffBranches returns the commits reachable from head and from no
177 // branch (refs/heads/*; tags do not count), newest first, at most limit
178 // (1 to MaxCommitLimit); more is true when there are others. It is empty
179 // for a commit that a branch already has.
180 CommitsOffBranches(ctx context.Context, repo domain.RepoName, head domain.ObjectID, limit int) (commits []domain.Commit, more bool, err error)
181}
182
183// --- inbound ---
184
185// RegisterRunnerParams describe a runner to register.
186type RegisterRunnerParams struct {
187 Name string
188 Labels []string
189 Version string
190 Isolation domain.CIIsolation
191 Arch, OS string
192 Replace bool
193}
194
195// RegisteredRunner is a new runner with its secret, shown once.
196type RegisteredRunner struct {
197 Runner domain.CIRunner
198 Token string
199}
200
201// RunnerJob is a job given to a runner.
202type RunnerJob struct {
203 Job domain.CIJob
204 Token string
205 Repo domain.RepoName
206 Sha, Ref string
207 WorkflowPath string
208 WorkflowYAML string
209}
210
211// FetchJobParams selects the job a runner asks for.
212type FetchJobParams struct {
213 Labels []string
214 PollID string
215}
216
217// FetchJobResult is the answer of a poll.
218type FetchJobResult struct {
219 Job *RunnerJob
220 // RetryAfter is how long the runner waits before polling again.
221 RetryAfter time.Duration
222}
223
224// StepUpdate is the state of one step as a runner reports it.
225type StepUpdate struct {
226 Number int32
227 Name string
228 State domain.CIStepState
229 Conclusion domain.CIConclusion
230 StartedAt time.Time
231 CompletedAt time.Time
232}
233
234// CompleteJobParams end a job.
235type CompleteJobParams struct {
236 JobID string
237 Conclusion domain.CIConclusion
238 Outputs map[string]string
239 RunnerShutdown bool
240 Steps []StepUpdate
241}
242
243// CompleteJobResult is the stored result of a job.
244type CompleteJobResult struct {
245 Conclusion domain.CIConclusion
246 Error string
247 Applied bool
248}
249
250// JobSignal is one message of WatchJob.
251type JobSignal struct {
252 Keepalive bool
253 // Cancel is set for a cancel request.
254 Cancel *JobCancel
255 // Ended is set when the job has ended.
256 Ended *domain.JobEndedError
257}
258
259// JobCancel is a cancel request for a running job.
260type JobCancel struct {
261 Reason domain.CICancelReason
262 Message string
263 RequestedAt time.Time
264}
265
266// RunnerService is what runners call. The secret is the bearer secret of the
267// call: the runner token for FetchJob, the job token for the other calls.
268// A bad secret is domain.ErrUnauthenticated; a job token of another job than
269// the one named is domain.ErrForbidden. *usecase.CI implements it.
270type RunnerService interface {
271 Register(ctx context.Context, p domain.Principal, params RegisterRunnerParams) (RegisteredRunner, error)
272 FetchJob(ctx context.Context, secret string, params FetchJobParams) (FetchJobResult, error)
273 UpdateJob(ctx context.Context, secret, jobID string, steps []StepUpdate) error
274 AppendLog(ctx context.Context, secret, jobID string, step int32, offset int64, lines []domain.CILogLine) (next int64, err error)
275 Heartbeat(ctx context.Context, secret, jobID string) error
276 // WatchJob calls ready once the job token is valid and the stream is
277 // registered, before any message: the adapter flushes the response headers
278 // then, because a client's call returns only when they arrive.
279 WatchJob(ctx context.Context, secret, jobID string, ready func(), send func(JobSignal) error) error
280 CompleteJob(ctx context.Context, secret string, params CompleteJobParams) (CompleteJobResult, error)
281}
282
283// JobView is a job as owners see it.
284type JobView struct {
285 domain.CIJob
286 // Lines is the number of lines of its log now.
287 Lines int64
288 // RunnerGone is set when the runner of the job was removed since.
289 RunnerGone bool
290}
291
292// StepRanges returns where the lines of every step are in the log.
293func (v JobView) StepRanges() []domain.StepRange { return domain.StepRanges(v.Steps, v.Lines) }
294
295// ListRunsParams select a page of runs.
296type ListRunsParams struct {
297 PageSize int
298 PageToken string
299 // Workflow restricts the result to a workflow file: its path or its file
300 // name. Branch restricts it to the runs of a branch, by short name. Status
301 // restricts it to the runs of that status. Empty means no restriction.
302 Workflow string
303 Branch string
304 Status domain.CIStatus
305}
306
307// RunPage is a page of runs, newest first.
308type RunPage struct {
309 Runs []domain.CIRun
310 NextPageToken string
311}
312
313// RunDetail is a run with its jobs.
314type RunDetail struct {
315 Run domain.CIRun
316 Jobs []JobView
317 LatestAttempt int32
318}
319
320// LogBatch is a message of a log stream: consecutive lines from Offset.
321type LogBatch struct {
322 Offset int64
323 Lines []JobLogLine
324}
325
326// JobLogLine is a log line with its step.
327type JobLogLine struct {
328 Step int32
329 domain.CILogLine
330}
331
332// GetJobLogParams select the log to stream.
333type GetJobLogParams struct {
334 JobID string
335 // Attempt is the attempt of the job, 0 for the latest.
336 Attempt int32
337 Offset int64
338 Follow bool
339 // Ready, when not nil, is called once the access is checked and the
340 // stream is counted, before any batch: the adapter flushes the response
341 // headers then (a follow on a job with no line yet would keep the client's
342 // call waiting for the first one).
343 Ready func()
344 // Client identifies where the call comes from, for the limit of streams
345 // per client; empty counts against the total only.
346 Client string
347}
348
349// CheckView is a check with what owners need to read it.
350type CheckView struct {
351 domain.CICheck
352 // WaitingForRunner is set for a queued check whose job no registered
353 // runner can take (no runner has all its labels).
354 WaitingForRunner bool
355}
356
357// ChecksResult are the checks of a commit.
358type ChecksResult struct {
359 Checks []CheckView
360 Truncated bool
361}
362
363// RunnerPage is a page of runners.
364type RunnerPage struct {
365 Runners []RunnerView
366 NextPageToken string
367}
368
369// RunnerView is a runner with what is known of it now.
370type RunnerView struct {
371 domain.CIRunner
372 LastSeen time.Time
373 Busy bool
374}
375
376// RunService is the owner side of CI. *usecase.CI implements it.
377type RunService interface {
378 ListRuns(ctx context.Context, p domain.Principal, repo domain.RepoName, params ListRunsParams) (RunPage, error)
379 GetRun(ctx context.Context, p domain.Principal, repo domain.RepoName, runID int64, attempt int32) (RunDetail, error)
380 GetJob(ctx context.Context, p domain.Principal, repo domain.RepoName, jobID string, attempt int32) (JobView, error)
381 GetJobLog(ctx context.Context, p domain.Principal, repo domain.RepoName, params GetJobLogParams, send func(LogBatch) error) error
382 CancelRun(ctx context.Context, p domain.Principal, repo domain.RepoName, runID int64) (domain.CIRun, error)
383 RerunRun(ctx context.Context, p domain.Principal, repo domain.RepoName, runID int64, failedOnly bool) (domain.CIRun, error)
384 ListChecks(ctx context.Context, p domain.Principal, repo domain.RepoName, sha string) (ChecksResult, error)
385 ListRunners(ctx context.Context, p domain.Principal, size int, token string) (RunnerPage, error)
386 RemoveRunner(ctx context.Context, p domain.Principal, id string) error
387}