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}
177
178// --- inbound ---
179
180// RegisterRunnerParams describe a runner to register.
181type RegisterRunnerParams struct {
182 Name string
183 Labels []string
184 Version string
185 Isolation domain.CIIsolation
186 Arch, OS string
187 Replace bool
188}
189
190// RegisteredRunner is a new runner with its secret, shown once.
191type RegisteredRunner struct {
192 Runner domain.CIRunner
193 Token string
194}
195
196// RunnerJob is a job given to a runner.
197type RunnerJob struct {
198 Job domain.CIJob
199 Token string
200 Repo domain.RepoName
201 Sha, Ref string
202 WorkflowPath string
203 WorkflowYAML string
204}
205
206// FetchJobParams selects the job a runner asks for.
207type FetchJobParams struct {
208 Labels []string
209 PollID string
210}
211
212// FetchJobResult is the answer of a poll.
213type FetchJobResult struct {
214 Job *RunnerJob
215 // RetryAfter is how long the runner waits before polling again.
216 RetryAfter time.Duration
217}
218
219// StepUpdate is the state of one step as a runner reports it.
220type StepUpdate struct {
221 Number int32
222 Name string
223 State domain.CIStepState
224 Conclusion domain.CIConclusion
225 StartedAt time.Time
226 CompletedAt time.Time
227}
228
229// CompleteJobParams end a job.
230type CompleteJobParams struct {
231 JobID string
232 Conclusion domain.CIConclusion
233 Outputs map[string]string
234 RunnerShutdown bool
235 Steps []StepUpdate
236}
237
238// CompleteJobResult is the stored result of a job.
239type CompleteJobResult struct {
240 Conclusion domain.CIConclusion
241 Error string
242 Applied bool
243}
244
245// JobSignal is one message of WatchJob.
246type JobSignal struct {
247 Keepalive bool
248 // Cancel is set for a cancel request.
249 Cancel *JobCancel
250 // Ended is set when the job has ended.
251 Ended *domain.JobEndedError
252}
253
254// JobCancel is a cancel request for a running job.
255type JobCancel struct {
256 Reason domain.CICancelReason
257 Message string
258 RequestedAt time.Time
259}
260
261// RunnerService is what runners call. The secret is the bearer secret of the
262// call: the runner token for FetchJob, the job token for the other calls.
263// A bad secret is domain.ErrUnauthenticated; a job token of another job than
264// the one named is domain.ErrForbidden. *usecase.CI implements it.
265type RunnerService interface {
266 Register(ctx context.Context, p domain.Principal, params RegisterRunnerParams) (RegisteredRunner, error)
267 FetchJob(ctx context.Context, secret string, params FetchJobParams) (FetchJobResult, error)
268 UpdateJob(ctx context.Context, secret, jobID string, steps []StepUpdate) error
269 AppendLog(ctx context.Context, secret, jobID string, step int32, offset int64, lines []domain.CILogLine) (next int64, err error)
270 Heartbeat(ctx context.Context, secret, jobID string) error
271 // WatchJob calls ready once the job token is valid and the stream is
272 // registered, before any message: the adapter flushes the response headers
273 // then, because a client's call returns only when they arrive.
274 WatchJob(ctx context.Context, secret, jobID string, ready func(), send func(JobSignal) error) error
275 CompleteJob(ctx context.Context, secret string, params CompleteJobParams) (CompleteJobResult, error)
276}
277
278// JobView is a job as owners see it.
279type JobView struct {
280 domain.CIJob
281 // Lines is the number of lines of its log now.
282 Lines int64
283 // RunnerGone is set when the runner of the job was removed since.
284 RunnerGone bool
285}
286
287// StepRanges returns where the lines of every step are in the log.
288func (v JobView) StepRanges() []domain.StepRange { return domain.StepRanges(v.Steps, v.Lines) }
289
290// ListRunsParams select a page of runs.
291type ListRunsParams struct {
292 PageSize int
293 PageToken string
294 // Workflow restricts the result to a workflow file: its path or its file
295 // name. Branch restricts it to the runs of a branch, by short name. Status
296 // restricts it to the runs of that status. Empty means no restriction.
297 Workflow string
298 Branch string
299 Status domain.CIStatus
300}
301
302// RunPage is a page of runs, newest first.
303type RunPage struct {
304 Runs []domain.CIRun
305 NextPageToken string
306}
307
308// RunDetail is a run with its jobs.
309type RunDetail struct {
310 Run domain.CIRun
311 Jobs []JobView
312 LatestAttempt int32
313}
314
315// LogBatch is a message of a log stream: consecutive lines from Offset.
316type LogBatch struct {
317 Offset int64
318 Lines []JobLogLine
319}
320
321// JobLogLine is a log line with its step.
322type JobLogLine struct {
323 Step int32
324 domain.CILogLine
325}
326
327// GetJobLogParams select the log to stream.
328type GetJobLogParams struct {
329 JobID string
330 // Attempt is the attempt of the job, 0 for the latest.
331 Attempt int32
332 Offset int64
333 Follow bool
334 // Ready, when not nil, is called once the access is checked and the
335 // stream is counted, before any batch: the adapter flushes the response
336 // headers then (a follow on a job with no line yet would keep the client's
337 // call waiting for the first one).
338 Ready func()
339 // Client identifies where the call comes from, for the limit of streams
340 // per client; empty counts against the total only.
341 Client string
342}
343
344// CheckView is a check with what owners need to read it.
345type CheckView struct {
346 domain.CICheck
347 // WaitingForRunner is set for a queued check whose job no registered
348 // runner can take (no runner has all its labels).
349 WaitingForRunner bool
350}
351
352// ChecksResult are the checks of a commit.
353type ChecksResult struct {
354 Checks []CheckView
355 Truncated bool
356}
357
358// RunnerPage is a page of runners.
359type RunnerPage struct {
360 Runners []RunnerView
361 NextPageToken string
362}
363
364// RunnerView is a runner with what is known of it now.
365type RunnerView struct {
366 domain.CIRunner
367 LastSeen time.Time
368 Busy bool
369}
370
371// RunService is the owner side of CI. *usecase.CI implements it.
372type RunService interface {
373 ListRuns(ctx context.Context, p domain.Principal, repo domain.RepoName, params ListRunsParams) (RunPage, error)
374 GetRun(ctx context.Context, p domain.Principal, repo domain.RepoName, runID int64, attempt int32) (RunDetail, error)
375 GetJob(ctx context.Context, p domain.Principal, repo domain.RepoName, jobID string, attempt int32) (JobView, error)
376 GetJobLog(ctx context.Context, p domain.Principal, repo domain.RepoName, params GetJobLogParams, send func(LogBatch) error) error
377 CancelRun(ctx context.Context, p domain.Principal, repo domain.RepoName, runID int64) (domain.CIRun, error)
378 RerunRun(ctx context.Context, p domain.Principal, repo domain.RepoName, runID int64, failedOnly bool) (domain.CIRun, error)
379 ListChecks(ctx context.Context, p domain.Principal, repo domain.RepoName, sha string) (ChecksResult, error)
380 ListRunners(ctx context.Context, p domain.Principal, size int, token string) (RunnerPage, error)
381 RemoveRunner(ctx context.Context, p domain.Principal, id string) error
382}