@@ -165,26 +165,16 @@ func (server *provisionerdServer) AcquireJob(ctx context.Context, _ *proto.Empty
165165 return xerrors .Errorf ("request job was invalidated: %s" , errorMessage )
166166 }
167167
168- project , err := server .Database .GetProjectByID (ctx , job .ProjectID )
169- if err != nil {
170- return nil , failJob (fmt .Sprintf ("get project: %s" , err ))
171- }
172- organization , err := server .Database .GetOrganizationByID (ctx , project .OrganizationID )
173- if err != nil {
174- return nil , failJob (fmt .Sprintf ("get organization: %s" , err ))
175- }
176168 user , err := server .Database .GetUserByID (ctx , job .InitiatorID )
177169 if err != nil {
178170 return nil , failJob (fmt .Sprintf ("get user: %s" , err ))
179171 }
180172
181173 protoJob := & proto.AcquiredJob {
182- JobId : job .ID .String (),
183- CreatedAt : job .CreatedAt .UnixMilli (),
184- Provisioner : string (job .Provisioner ),
185- OrganizationName : organization .Name ,
186- ProjectName : project .Name ,
187- UserName : user .Username ,
174+ JobId : job .ID .String (),
175+ CreatedAt : job .CreatedAt .UnixMilli (),
176+ Provisioner : string (job .Provisioner ),
177+ UserName : user .Username ,
188178 }
189179 var projectVersion database.ProjectVersion
190180 switch job .Type {
@@ -206,6 +196,14 @@ func (server *provisionerdServer) AcquireJob(ctx context.Context, _ *proto.Empty
206196 if err != nil {
207197 return nil , failJob (fmt .Sprintf ("get project version: %s" , err ))
208198 }
199+ project , err := server .Database .GetProjectByID (ctx , projectVersion .ProjectID )
200+ if err != nil {
201+ return nil , failJob (fmt .Sprintf ("get project: %s" , err ))
202+ }
203+ organization , err := server .Database .GetOrganizationByID (ctx , project .OrganizationID )
204+ if err != nil {
205+ return nil , failJob (fmt .Sprintf ("get organization: %s" , err ))
206+ }
209207
210208 // Compute parameters for the workspace to consume.
211209 parameters , err := projectparameter .Compute (ctx , server .Database , projectparameter.Scope {
@@ -251,8 +249,8 @@ func (server *provisionerdServer) AcquireJob(ctx context.Context, _ *proto.Empty
251249
252250 protoJob .Type = & proto.AcquiredJob_ProjectImport_ {
253251 ProjectImport : & proto.AcquiredJob_ProjectImport {
254- ProjectVersionId : projectVersion . ID . String (),
255- ProjectVersionName : projectVersion . Name ,
252+ // This will be replaced once the project import has been refactored.
253+ ProjectName : "placeholder" ,
256254 },
257255 }
258256 }
@@ -294,85 +292,36 @@ func (server *provisionerdServer) UpdateJob(stream proto.DRPCProvisionerDaemon_U
294292 if err != nil {
295293 return xerrors .Errorf ("update job: %w" , err )
296294 }
297- switch job .Type {
298- case database .ProvisionerJobTypeProjectImport :
299- if len (update .ProjectImportLogs ) == 0 {
300- continue
301- }
302- var input projectImportJob
303- err = json .Unmarshal (job .Input , & input )
304- if err != nil {
305- return xerrors .Errorf ("unmarshal job input %q: %s" , job .Input , err )
306- }
307- insertParams := database.InsertProjectVersionLogsParams {
308- ProjectVersionID : input .ProjectVersionID ,
309- }
310- for _ , log := range update .ProjectImportLogs {
311- logLevel , err := convertLogLevel (log .Level )
312- if err != nil {
313- return xerrors .Errorf ("convert log level: %w" , err )
314- }
315- logSource , err := convertLogSource (log .Source )
316- if err != nil {
317- return xerrors .Errorf ("convert log source: %w" , err )
318- }
319- insertParams .ID = append (insertParams .ID , uuid .New ())
320- insertParams .CreatedAt = append (insertParams .CreatedAt , time .UnixMilli (log .CreatedAt ))
321- insertParams .Level = append (insertParams .Level , logLevel )
322- insertParams .Source = append (insertParams .Source , logSource )
323- insertParams .Output = append (insertParams .Output , log .Output )
324- }
325- logs , err := server .Database .InsertProjectVersionLogs (stream .Context (), insertParams )
326- if err != nil {
327- return xerrors .Errorf ("insert project logs: %w" , err )
328- }
329- data , err := json .Marshal (logs )
330- if err != nil {
331- return xerrors .Errorf ("marshal project log: %w" , err )
332- }
333- err = server .Pubsub .Publish (projectVersionLogsChannel (input .ProjectVersionID ), data )
334- if err != nil {
335- return xerrors .Errorf ("publish history log: %w" , err )
336- }
337- case database .ProvisionerJobTypeWorkspaceProvision :
338- if len (update .WorkspaceProvisionLogs ) == 0 {
339- continue
340- }
341- var input workspaceProvisionJob
342- err = json .Unmarshal (job .Input , & input )
343- if err != nil {
344- return xerrors .Errorf ("unmarshal job input %q: %s" , job .Input , err )
345- }
346- insertParams := database.InsertWorkspaceHistoryLogsParams {
347- WorkspaceHistoryID : input .WorkspaceHistoryID ,
348- }
349- for _ , log := range update .WorkspaceProvisionLogs {
350- logLevel , err := convertLogLevel (log .Level )
351- if err != nil {
352- return xerrors .Errorf ("convert log level: %w" , err )
353- }
354- logSource , err := convertLogSource (log .Source )
355- if err != nil {
356- return xerrors .Errorf ("convert log source: %w" , err )
357- }
358- insertParams .ID = append (insertParams .ID , uuid .New ())
359- insertParams .CreatedAt = append (insertParams .CreatedAt , time .UnixMilli (log .CreatedAt ))
360- insertParams .Level = append (insertParams .Level , logLevel )
361- insertParams .Source = append (insertParams .Source , logSource )
362- insertParams .Output = append (insertParams .Output , log .Output )
363- }
364- logs , err := server .Database .InsertWorkspaceHistoryLogs (stream .Context (), insertParams )
365- if err != nil {
366- return xerrors .Errorf ("insert workspace logs: %w" , err )
367- }
368- data , err := json .Marshal (logs )
295+ insertParams := database.InsertProvisionerJobLogsParams {
296+ JobID : parsedID ,
297+ }
298+ for _ , log := range update .Logs {
299+ logLevel , err := convertLogLevel (log .Level )
369300 if err != nil {
370- return xerrors .Errorf ("marshal project log: %w" , err )
301+ return xerrors .Errorf ("convert log level : %w" , err )
371302 }
372- err = server . Pubsub . Publish ( workspaceHistoryLogsChannel ( input . WorkspaceHistoryID ), data )
303+ logSource , err := convertLogSource ( log . Source )
373304 if err != nil {
374- return xerrors .Errorf ("publish history log: %w" , err )
305+ return xerrors .Errorf ("convert log source : %w" , err )
375306 }
307+ insertParams .ID = append (insertParams .ID , uuid .New ())
308+ insertParams .CreatedAt = append (insertParams .CreatedAt , time .UnixMilli (log .CreatedAt ))
309+ insertParams .Level = append (insertParams .Level , logLevel )
310+ insertParams .Source = append (insertParams .Source , logSource )
311+ insertParams .Output = append (insertParams .Output , log .Output )
312+ }
313+ logs , err := server .Database .InsertProvisionerJobLogs (stream .Context (), insertParams )
314+ if err != nil {
315+ server .Logger .Error (stream .Context (), "insert provisioner job logs" , slog .Error (err ))
316+ return xerrors .Errorf ("insert job logs: %w" , err )
317+ }
318+ data , err := json .Marshal (logs )
319+ if err != nil {
320+ return xerrors .Errorf ("marshal job log: %w" , err )
321+ }
322+ err = server .Pubsub .Publish (provisionerJobLogsChannel (parsedID ), data )
323+ if err != nil {
324+ return xerrors .Errorf ("publish job log: %w" , err )
376325 }
377326 }
378327}
@@ -382,6 +331,13 @@ func (server *provisionerdServer) CancelJob(ctx context.Context, cancelJob *prot
382331 if err != nil {
383332 return nil , xerrors .Errorf ("parse job id: %w" , err )
384333 }
334+ job , err := server .Database .GetProvisionerJobByID (ctx , jobID )
335+ if err != nil {
336+ return nil , xerrors .Errorf ("get provisioner job: %w" , err )
337+ }
338+ if job .CompletedAt .Valid {
339+ return nil , xerrors .Errorf ("job already completed" )
340+ }
385341 err = server .Database .UpdateProvisionerJobWithCompleteByID (ctx , database.UpdateProvisionerJobWithCompleteByIDParams {
386342 ID : jobID ,
387343 CompletedAt : sql.NullTime {
0 commit comments