coderd/provisionerjobs.go
Rama main · réplica de lectura del repositorio original.
package coderd
import (
"context"
"database/sql"
"encoding/json"
"errors"
"io"
"net/http"
"sort"
"strconv"
"github.com/google/uuid"
"golang.org/x/xerrors"
"cdr.dev/slog/v3"
"github.com/coder/coder/v2/coderd/database"
"github.com/coder/coder/v2/coderd/database/db2sdk"
"github.com/coder/coder/v2/coderd/database/dbauthz"
"github.com/coder/coder/v2/coderd/database/pubsub"
"github.com/coder/coder/v2/coderd/httpapi"
"github.com/coder/coder/v2/coderd/httpmw"
"github.com/coder/coder/v2/coderd/httpmw/loggermw"
"github.com/coder/coder/v2/coderd/rbac"
"github.com/coder/coder/v2/coderd/rbac/policy"
"github.com/coder/coder/v2/coderd/util/slice"
"github.com/coder/coder/v2/codersdk"
"github.com/coder/coder/v2/codersdk/wsjson"
"github.com/coder/coder/v2/provisionersdk"
"github.com/coder/websocket"
)
// @Summary Get provisioner job
// @ID get-provisioner-job
// @Security CoderSessionToken
// @Produce json
// @Tags Organizations
// @Param organization path string true "Organization ID" format(uuid)
// @Param job path string true "Job ID" format(uuid)
// @Success 200 {object} codersdk.ProvisionerJob
// @Router /api/v2/organizations/{organization}/provisionerjobs/{job} [get]
func (api *API) provisionerJob(rw http.ResponseWriter, r *http.Request) {
ctx := r.Context()
jobID, ok := httpmw.ParseUUIDParam(rw, r, "job")
if !ok {
return
}
jobs, ok := api.handleAuthAndFetchProvisionerJobs(rw, r, []uuid.UUID{jobID})
if !ok {
return
}
if len(jobs) == 0 {
httpapi.ResourceNotFound(rw)
return
}
if len(jobs) > 1 || jobs[0].ProvisionerJob.ID != jobID {
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Internal error fetching provisioner job.",
Detail: "Database returned an unexpected job.",
})
return
}
httpapi.Write(ctx, rw, http.StatusOK, convertProvisionerJobWithQueuePosition(jobs[0]))
}
// @Summary Get provisioner jobs
// @ID get-provisioner-jobs
// @Security CoderSessionToken
// @Produce json
// @Tags Organizations
// @Param organization path string true "Organization ID" format(uuid)
// @Param limit query int false "Page limit"
// @Param ids query []string false "Filter results by job IDs (UUIDs)"
// @Param status query []codersdk.ProvisionerJobStatus false "Filter results by status"
// @Param tags query string false "Provisioner tags to filter by (JSON of the form `{'tag1':'value1','tag2':'value2'}`)"
// @Param initiator query string false "Filter results by initiator" format(uuid)
// @Success 200 {array} codersdk.ProvisionerJob
// @Router /api/v2/organizations/{organization}/provisionerjobs [get]
func (api *API) provisionerJobs(rw http.ResponseWriter, r *http.Request) {
ctx := r.Context()
jobs, ok := api.handleAuthAndFetchProvisionerJobs(rw, r, nil)
if !ok {
return
}
httpapi.Write(ctx, rw, http.StatusOK, slice.List(jobs, convertProvisionerJobWithQueuePosition))
}
// handleAuthAndFetchProvisionerJobs is an internal method shared by
// provisionerJob and provisionerJobs. If ok is false the caller should
// return immediately because the response has already been written.
func (api *API) handleAuthAndFetchProvisionerJobs(rw http.ResponseWriter, r *http.Request, ids []uuid.UUID) (_ []database.GetProvisionerJobsByOrganizationAndStatusWithQueuePositionAndProvisionerRow, ok bool) {
ctx := r.Context()
org := httpmw.OrganizationParam(r)
// For now, only owners and template admins can access provisioner jobs.
if !api.Authorize(r, policy.ActionRead, rbac.ResourceProvisionerJobs.InOrg(org.ID)) {
httpapi.ResourceNotFound(rw)
return nil, false
}
qp := r.URL.Query()
p := httpapi.NewQueryParamParser()
limit := p.PositiveInt32(qp, 50, "limit")
status := p.Strings(qp, nil, "status")
if ids == nil {
ids = p.UUIDs(qp, nil, "ids")
}
tags := p.JSONStringMap(qp, database.StringMap{}, "tags")
initiatorID := p.UUID(qp, uuid.Nil, "initiator")
p.ErrorExcessParams(qp)
if len(p.Errors) > 0 {
httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
Message: "Invalid query parameters.",
Validations: p.Errors,
})
return nil, false
}
jobs, err := api.Database.GetProvisionerJobsByOrganizationAndStatusWithQueuePositionAndProvisioner(ctx, database.GetProvisionerJobsByOrganizationAndStatusWithQueuePositionAndProvisionerParams{
OrganizationID: org.ID,
Status: slice.StringEnums[database.ProvisionerJobStatus](status),
Limit: sql.NullInt32{Int32: limit, Valid: limit > 0},
IDs: ids,
Tags: tags,
InitiatorID: initiatorID,
})
if err != nil {
if httpapi.Is404Error(err) {
httpapi.ResourceNotFound(rw)
return nil, false
}
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Internal error fetching provisioner jobs.",
Detail: err.Error(),
})
return nil, false
}
return jobs, true
}
// Returns provisioner logs based on query parameters.
// The intended usage for a client to stream all logs (with JS API):
// GET /logs
// GET /logs?after=<id>&follow
// The combination of these responses should provide all current logs
// to the consumer, and future logs are streamed in the follow request.
func (api *API) provisionerJobLogs(rw http.ResponseWriter, r *http.Request, job database.ProvisionerJob) {
var (
ctx = r.Context()
logger = api.Logger.With(slog.F("job_id", job.ID))
follow = r.URL.Query().Has("follow")
afterRaw = r.URL.Query().Get("after")
format = r.URL.Query().Get("format")
)
// Validate format parameter.
if format == "" {
format = "json"
}
if format != "json" && format != "text" {
httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
Message: "Invalid format parameter.",
Detail: "Allowed values are \"json\" and \"text\".",
})
return
}
// Text format is not supported with streaming.
if format == "text" && follow {
httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
Message: "Text format is not supported with follow mode.",
Detail: "Use format=json or omit the follow parameter.",
})
return
}
var after int64
// Only fetch logs created after the time provided.
if afterRaw != "" {
var err error
after, err = strconv.ParseInt(afterRaw, 10, 64)
if err != nil {
httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
Message: "Query param \"after\" must be an integer.",
Validations: []codersdk.ValidationError{
{Field: "after", Detail: "Must be an integer"},
},
})
return
}
}
if !follow {
fetchAndWriteLogs(ctx, api.Database, job.ID, after, rw, format)
return
}
follower := newLogFollower(ctx, logger, api.Database, api.Pubsub, api.wsWatcher, rw, r, job, after)
api.WebsocketWaitMutex.Lock()
api.WebsocketWaitGroup.Add(1)
api.WebsocketWaitMutex.Unlock()
defer api.WebsocketWaitGroup.Done()
follower.follow()
}
func (api *API) provisionerJobResources(rw http.ResponseWriter, r *http.Request, job database.ProvisionerJob) {
ctx := r.Context()
if !job.CompletedAt.Valid {
httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
Message: "Job hasn't completed!",
})
return
}
// nolint:gocritic // GetWorkspaceResourcesByJobID is a system function.
resources, err := api.Database.GetWorkspaceResourcesByJobID(dbauthz.AsSystemRestricted(ctx), job.ID)
if errors.Is(err, sql.ErrNoRows) {
err = nil
}
if err != nil {
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Internal error fetching job resources.",
Detail: err.Error(),
})
return
}
resourceIDs := make([]uuid.UUID, 0)
for _, resource := range resources {
resourceIDs = append(resourceIDs, resource.ID)
}
// nolint:gocritic // GetWorkspaceAgentsByResourceIDs is a system function.
resourceAgents, err := api.Database.GetWorkspaceAgentsByResourceIDs(dbauthz.AsSystemRestricted(ctx), resourceIDs)
if errors.Is(err, sql.ErrNoRows) {
err = nil
}
if err != nil {
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Internal error fetching workspace agent.",
Detail: err.Error(),
})
return
}
resourceAgentIDs := make([]uuid.UUID, 0)
for _, agent := range resourceAgents {
resourceAgentIDs = append(resourceAgentIDs, agent.ID)
}
// nolint:gocritic // GetWorkspaceAppsByAgentIDs is a system function.
apps, err := api.Database.GetWorkspaceAppsByAgentIDs(dbauthz.AsSystemRestricted(ctx), resourceAgentIDs)
if errors.Is(err, sql.ErrNoRows) {
err = nil
}
if err != nil {
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Internal error fetching workspace applications.",
Detail: err.Error(),
})
return
}
// nolint:gocritic // GetWorkspaceAgentScriptsByAgentIDs is a system function.
scripts, err := api.Database.GetWorkspaceAgentScriptsByAgentIDs(dbauthz.AsSystemRestricted(ctx), resourceAgentIDs)
if errors.Is(err, sql.ErrNoRows) {
err = nil
}
if err != nil {
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Internal error fetching workspace agent scripts.",
Detail: err.Error(),
})
return
}
// nolint:gocritic // GetWorkspaceAgentLogSourcesByAgentIDs is a system function.
logSources, err := api.Database.GetWorkspaceAgentLogSourcesByAgentIDs(dbauthz.AsSystemRestricted(ctx), resourceAgentIDs)
if errors.Is(err, sql.ErrNoRows) {
err = nil
}
if err != nil {
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Internal error fetching workspace agent log sources.",
Detail: err.Error(),
})
return
}
// nolint:gocritic // GetWorkspaceResourceMetadataByResourceIDs is a system function.
resourceMetadata, err := api.Database.GetWorkspaceResourceMetadataByResourceIDs(dbauthz.AsSystemRestricted(ctx), resourceIDs)
if err != nil {
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Internal error fetching workspace metadata.",
Detail: err.Error(),
})
return
}
apiResources := make([]codersdk.WorkspaceResource, 0)
for _, resource := range resources {
agents := make([]codersdk.WorkspaceAgent, 0)
for _, agent := range resourceAgents {
if agent.ResourceID != resource.ID {
continue
}
dbApps := make([]database.WorkspaceApp, 0)
for _, app := range apps {
if app.AgentID == agent.ID {
dbApps = append(dbApps, app)
}
}
dbScripts := make([]database.GetWorkspaceAgentScriptsByAgentIDsRow, 0)
for _, script := range scripts {
if script.WorkspaceAgentID == agent.ID {
dbScripts = append(dbScripts, script)
}
}
dbLogSources := make([]database.WorkspaceAgentLogSource, 0)
for _, logSource := range logSources {
if logSource.WorkspaceAgentID == agent.ID {
dbLogSources = append(dbLogSources, logSource)
}
}
apiScripts, err := convertScripts(dbScripts)
if err != nil {
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Internal error reading workspace agent scripts.",
Detail: err.Error(),
})
return
}
apiAgent, err := db2sdk.WorkspaceAgent(
api.DERPMap(), *api.TailnetCoordinator.Load(), agent, convertProvisionedApps(dbApps), apiScripts, convertLogSources(dbLogSources), api.AgentInactiveDisconnectTimeout,
api.DeploymentValues.AgentFallbackTroubleshootingURL.String(),
)
if err != nil {
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Internal error reading job agent.",
Detail: err.Error(),
})
return
}
agents = append(agents, apiAgent)
}
metadata := make([]database.WorkspaceResourceMetadatum, 0)
for _, field := range resourceMetadata {
if field.WorkspaceResourceID == resource.ID {
metadata = append(metadata, field)
}
}
apiResources = append(apiResources, convertWorkspaceResource(resource, agents, metadata))
}
sort.Slice(apiResources, func(i, j int) bool {
return apiResources[i].Name < apiResources[j].Name
})
httpapi.Write(ctx, rw, http.StatusOK, apiResources)
}
func convertProvisionerJobLogs(provisionerJobLogs []database.ProvisionerJobLog) []codersdk.ProvisionerJobLog {
sdk := make([]codersdk.ProvisionerJobLog, 0, len(provisionerJobLogs))
for _, log := range provisionerJobLogs {
sdk = append(sdk, convertProvisionerJobLog(log))
}
return sdk
}
func convertProvisionerJobLog(provisionerJobLog database.ProvisionerJobLog) codersdk.ProvisionerJobLog {
return codersdk.ProvisionerJobLog{
ID: provisionerJobLog.ID,
CreatedAt: provisionerJobLog.CreatedAt,
Source: codersdk.LogSource(provisionerJobLog.Source),
Level: codersdk.LogLevel(provisionerJobLog.Level),
Stage: provisionerJobLog.Stage,
Output: provisionerJobLog.Output,
}
}
func convertProvisionerJob(pj database.GetProvisionerJobsByIDsWithQueuePositionRow) codersdk.ProvisionerJob {
provisionerJob := pj.ProvisionerJob
job := codersdk.ProvisionerJob{
ID: provisionerJob.ID,
OrganizationID: provisionerJob.OrganizationID,
InitiatorID: provisionerJob.InitiatorID,
CreatedAt: provisionerJob.CreatedAt,
Type: codersdk.ProvisionerJobType(provisionerJob.Type),
Error: provisionerJob.Error.String,
ErrorCode: codersdk.JobErrorCode(provisionerJob.ErrorCode.String),
FileID: provisionerJob.FileID,
Tags: provisionerJob.Tags,
QueuePosition: int(pj.QueuePosition),
QueueSize: int(pj.QueueSize),
LogsOverflowed: provisionerJob.LogsOverflowed,
}
// Applying values optional to the struct.
if provisionerJob.StartedAt.Valid {
job.StartedAt = &provisionerJob.StartedAt.Time
}
if provisionerJob.CompletedAt.Valid {
job.CompletedAt = &provisionerJob.CompletedAt.Time
}
if provisionerJob.CanceledAt.Valid {
job.CanceledAt = &provisionerJob.CanceledAt.Time
}
if provisionerJob.WorkerID.Valid {
job.WorkerID = &provisionerJob.WorkerID.UUID
}
job.Status = codersdk.ProvisionerJobStatus(pj.ProvisionerJob.JobStatus)
// Only unmarshal input if it exists, this should only be zero in testing.
if len(provisionerJob.Input) > 0 {
if err := json.Unmarshal(provisionerJob.Input, &job.Input); err != nil {
job.Input.Error = xerrors.Errorf("decode input %s: %w", provisionerJob.Input, err).Error()
}
}
return job
}
func convertProvisionerJobWithQueuePosition(pj database.GetProvisionerJobsByOrganizationAndStatusWithQueuePositionAndProvisionerRow) codersdk.ProvisionerJob {
job := convertProvisionerJob(database.GetProvisionerJobsByIDsWithQueuePositionRow{
ProvisionerJob: pj.ProvisionerJob,
QueuePosition: pj.QueuePosition,
QueueSize: pj.QueueSize,
})
job.WorkerName = pj.WorkerName
job.AvailableWorkers = pj.AvailableWorkers
job.Metadata = codersdk.ProvisionerJobMetadata{
TemplateVersionName: pj.TemplateVersionName,
TemplateID: pj.TemplateID.UUID,
TemplateName: pj.TemplateName,
TemplateDisplayName: pj.TemplateDisplayName,
TemplateIcon: pj.TemplateIcon,
WorkspaceName: pj.WorkspaceName,
}
if pj.WorkspaceID.Valid {
job.Metadata.WorkspaceID = &pj.WorkspaceID.UUID
}
if pj.WorkspaceBuildTransition.Valid {
job.Metadata.WorkspaceBuildTransition = codersdk.WorkspaceTransition(pj.WorkspaceBuildTransition.WorkspaceTransition)
}
return job
}
func fetchAndWriteLogs(ctx context.Context, db database.Store, jobID uuid.UUID, after int64, rw http.ResponseWriter, format string) {
logs, err := db.GetProvisionerLogsAfterID(ctx, database.GetProvisionerLogsAfterIDParams{
JobID: jobID,
CreatedAfter: after,
})
if err != nil && !errors.Is(err, sql.ErrNoRows) {
httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
Message: "Internal error fetching provisioner logs.",
Detail: err.Error(),
})
return
}
if logs == nil {
logs = []database.ProvisionerJobLog{}
}
if format == "text" {
rw.Header().Set("Content-Type", "text/plain; charset=utf-8")
rw.WriteHeader(http.StatusOK)
for _, log := range logs {
_, _ = rw.Write([]byte(db2sdk.ProvisionerJobLog(log).Text()))
_, _ = rw.Write([]byte("\n"))
}
return
}
httpapi.Write(ctx, rw, http.StatusOK, convertProvisionerJobLogs(logs))
}
func jobIsComplete(logger slog.Logger, job database.ProvisionerJob) bool {
status := codersdk.ProvisionerJobStatus(job.JobStatus)
switch status {
case codersdk.ProvisionerJobCanceled:
return true
case codersdk.ProvisionerJobFailed:
return true
case codersdk.ProvisionerJobSucceeded:
return true
case codersdk.ProvisionerJobPending:
return false
case codersdk.ProvisionerJobCanceling:
return false
case codersdk.ProvisionerJobRunning:
return false
default:
logger.Error(context.Background(),
"can't convert the provisioner job status",
slog.F("job_id", job.ID), slog.F("status", status))
return false
}
}
type logFollower struct {
ctx context.Context
logger slog.Logger
db database.Store
pubsub pubsub.Pubsub
wsWatcher *httpapi.WSWatcher
r *http.Request
rw http.ResponseWriter
conn *websocket.Conn
enc *wsjson.Encoder[codersdk.ProvisionerJobLog]
jobID uuid.UUID
after int64
complete bool
notifications chan provisionersdk.ProvisionerJobLogsNotifyMessage
errors chan error
}
func newLogFollower(
ctx context.Context, logger slog.Logger, db database.Store, ps pubsub.Pubsub,
wsWatcher *httpapi.WSWatcher, rw http.ResponseWriter, r *http.Request,
job database.ProvisionerJob, after int64,
) *logFollower {
return &logFollower{
ctx: ctx,
logger: logger,
db: db,
pubsub: ps,
wsWatcher: wsWatcher,
r: r,
rw: rw,
jobID: job.ID,
after: after,
complete: jobIsComplete(logger, job),
notifications: make(chan provisionersdk.ProvisionerJobLogsNotifyMessage),
errors: make(chan error),
}
}
func (f *logFollower) follow() {
var cancel context.CancelFunc
f.ctx, cancel = context.WithCancel(f.ctx)
defer cancel()
// note that we only need to subscribe to updates if the job is not yet
// complete.
if !f.complete {
subCancel, err := f.pubsub.SubscribeWithErr(
provisionersdk.ProvisionerJobLogsNotifyChannel(f.jobID),
f.listener,
)
if err != nil {
httpapi.Write(f.ctx, f.rw, http.StatusInternalServerError, codersdk.Response{
Message: "failed to subscribe to job updates",
Detail: err.Error(),
})
return
}
defer subCancel()
// Move cancel up the stack so it happens before unsubscribing,
// otherwise we can end up in a deadlock due to how the
// in-memory pubsub does mutex locking on send/unsubscribe.
defer cancel()
// we were provided `complete` prior to starting this subscription, so
// we also need to check whether the job is now complete, in case the
// job completed between the last time we queried the job and the start
// of the subscription. If the job completes after this, we will get
// a notification on the subscription.
job, err := f.db.GetProvisionerJobByID(f.ctx, f.jobID)
if err != nil {
httpapi.Write(f.ctx, f.rw, http.StatusInternalServerError, codersdk.Response{
Message: "failed to query job",
Detail: err.Error(),
})
return
}
f.complete = jobIsComplete(f.logger, job)
f.logger.Debug(f.ctx, "queried job after subscribe", slog.F("complete", f.complete))
}
var err error
f.conn, err = websocket.Accept(f.rw, f.r, nil)
if err != nil {
httpapi.Write(f.ctx, f.rw, http.StatusBadRequest, codersdk.Response{
Message: "Failed to accept websocket.",
Detail: err.Error(),
})
return
}
defer f.conn.Close(websocket.StatusNormalClosure, "done")
// Do not reassign f.ctx here; the listener method reads
// f.ctx on the pubsub goroutine concurrently. Use a local
// variable instead. The watched context is a child of f.ctx,
// so canceling f.ctx still cascades.
watchCtx := f.wsWatcher.Watch(f.ctx, f.logger, f.conn)
f.enc = wsjson.NewEncoder[codersdk.ProvisionerJobLog](f.conn, websocket.MessageText)
// query for logs once right away, so we can get historical data from before
// subscription
if err := f.query(watchCtx); err != nil {
if watchCtx.Err() == nil && !xerrors.Is(err, io.EOF) {
// neither context expiry, nor EOF, close and log
f.logger.Error(watchCtx, "failed to query logs", slog.Error(err))
err = f.conn.Close(websocket.StatusInternalError, err.Error())
if err != nil {
f.logger.Warn(watchCtx, "failed to close websocket", slog.Error(err))
}
}
return
}
// Log the request immediately instead of after it completes.
if rl := loggermw.RequestLoggerFromContext(watchCtx); rl != nil {
rl.WriteLog(watchCtx, http.StatusAccepted)
}
// no need to wait if the job is done
if f.complete {
return
}
for {
select {
case err := <-f.errors:
// we've dropped at least one notification. This can happen if we
// lose database connectivity. We don't know whether the job is
// now complete since we could have missed the end of logs message.
// We could soldier on and retry, but loss of database connectivity
// is fairly serious, so instead just 500 and bail out. Client
// can retry and hopefully find a healthier node.
f.logger.Error(watchCtx, "dropped or corrupted notification", slog.Error(err))
err = f.conn.Close(websocket.StatusInternalError, err.Error())
if err != nil {
f.logger.Warn(watchCtx, "failed to close websocket", slog.Error(err))
}
return
case <-watchCtx.Done():
// client disconnect or probe failure
return
case n := <-f.notifications:
if n.EndOfLogs {
// safe to return here because we started the subscription,
// and then queried at least once, so we will have already
// gotten all logs prior to the start of our subscription.
return
}
err = f.query(watchCtx)
if err != nil {
if watchCtx.Err() == nil && !xerrors.Is(err, io.EOF) {
// neither context expiry, nor EOF, close and log
f.logger.Error(watchCtx, "failed to query logs", slog.Error(err))
err = f.conn.Close(websocket.StatusInternalError, httpapi.WebsocketCloseSprintf("%s", err.Error()))
if err != nil {
f.logger.Warn(watchCtx, "failed to close websocket", slog.Error(err))
}
}
return
}
}
}
}
func (f *logFollower) listener(_ context.Context, message []byte, err error) {
// in this function we always pair writes to channels with a select on the context
// otherwise we could block a goroutine if the follow() method exits.
if err != nil {
select {
case <-f.ctx.Done():
case f.errors <- err:
}
return
}
var n provisionersdk.ProvisionerJobLogsNotifyMessage
err = json.Unmarshal(message, &n)
if err != nil {
select {
case <-f.ctx.Done():
case f.errors <- err:
}
return
}
select {
case <-f.ctx.Done():
case f.notifications <- n:
}
}
// query fetches the latest job logs from the database and writes them to the
// connection.
func (f *logFollower) query(watchCtx context.Context) error {
f.logger.Debug(watchCtx, "querying logs", slog.F("after", f.after))
logs, err := f.db.GetProvisionerLogsAfterID(watchCtx, database.GetProvisionerLogsAfterIDParams{
JobID: f.jobID,
CreatedAfter: f.after,
})
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return xerrors.Errorf("error fetching logs: %w", err)
}
for _, log := range logs {
err := f.enc.Encode(convertProvisionerJobLog(log))
if err != nil {
return xerrors.Errorf("error writing to websocket: %w", err)
}
f.after = log.ID
f.logger.Debug(watchCtx, "wrote log to websocket", slog.F("id", log.ID))
}
return nil
}