Merge pull request #35 from makeopensource/dev

adde getStatus

Author: RA341

spec/leviathan_node/src/generated/jobs/v1/jobs_pb.ts

@@ -12,7 +12,7 @@ import type { Message } from "@bufbuild/protobuf";
  * Describes the file jobs/v1/jobs.proto.
  */
 export const file_jobs_v1_jobs: GenFile = /*@__PURE__*/
-  fileDesc("ChJqb2JzL3YxL2pvYnMucHJvdG8SB2pvYnMudjEizAEKDU5ld0pvYlJlcXVlc3QSJgoIam9iRmlsZXMYASADKAsyFC50eXBlcy52MS5GaWxlVXBsb2FkEigKCmRvY2tlckZpbGUYAiABKAsyFC50eXBlcy52MS5GaWxlVXBsb2FkEhEKCWltYWdlTmFtZRgDIAEoCRIbChNqb2JUaW1lb3V0SW5TZWNvbmRzGAQgASgEEhAKCGVudHJ5Q21kGAUgASgJEicKBmxpbWl0cxgGIAEoCzIXLnR5cGVzLnYxLk1hY2hpbmVMaW1pdHMiHwoOTmV3Sm9iUmVzcG9uc2USDQoFam9iSWQYASABKAkiIQoQQ2FuY2VsSm9iUmVxdWVzdBINCgVqb2JJZBgBIAEoCSITChFDYW5jZWxKb2JSZXNwb25zZSIeCg1Kb2JMb2dSZXF1ZXN0Eg0KBWpvYklkGAEgASgJIkQKD0pvYkxvZ3NSZXNwb25zZRIjCgdqb2JJbmZvGAEgASgLMhIuam9icy52MS5Kb2JTdGF0dXMSDAoEbG9ncxgCIAEoCSJDCglKb2JTdGF0dXMSDgoGam9iX2lkGAEgASgJEg4KBnN0YXR1cxgCIAEoCRIWCg5zdGF0dXNfbWVzc2FnZRgDIAEoCTLVAQoKSm9iU2VydmljZRI7CgZOZXdKb2ISFi5qb2JzLnYxLk5ld0pvYlJlcXVlc3QaFy5qb2JzLnYxLk5ld0pvYlJlc3BvbnNlIgASRAoMU3RyZWFtU3RhdHVzEhYuam9icy52MS5Kb2JMb2dSZXF1ZXN0Ghguam9icy52MS5Kb2JMb2dzUmVzcG9uc2UiADABEkQKCUNhbmNlbEpvYhIZLmpvYnMudjEuQ2FuY2VsSm9iUmVxdWVzdBoaLmpvYnMudjEuQ2FuY2VsSm9iUmVzcG9uc2UiAEKMAQoLY29tLmpvYnMudjFCCUpvYnNQcm90b1ABWjVnaXRodWIuY29tL21ha2VvcGVuc291cmNlL2xldmlhdGhhbi9nZW5lcmF0ZWQvam9icy92MaICA0pYWKoCB0pvYnMuVjHKAgdKb2JzXFYx4gITSm9ic1xWMVxHUEJNZXRhZGF0YeoCCEpvYnM6OlYxYgZwcm90bzM", [file_types_v1_types]);
+  fileDesc("ChJqb2JzL3YxL2pvYnMucHJvdG8SB2pvYnMudjEizAEKDU5ld0pvYlJlcXVlc3QSJgoIam9iRmlsZXMYASADKAsyFC50eXBlcy52MS5GaWxlVXBsb2FkEigKCmRvY2tlckZpbGUYAiABKAsyFC50eXBlcy52MS5GaWxlVXBsb2FkEhEKCWltYWdlTmFtZRgDIAEoCRIbChNqb2JUaW1lb3V0SW5TZWNvbmRzGAQgASgEEhAKCGVudHJ5Q21kGAUgASgJEicKBmxpbWl0cxgGIAEoCzIXLnR5cGVzLnYxLk1hY2hpbmVMaW1pdHMiHwoOTmV3Sm9iUmVzcG9uc2USDQoFam9iSWQYASABKAkiIQoQQ2FuY2VsSm9iUmVxdWVzdBINCgVqb2JJZBgBIAEoCSITChFDYW5jZWxKb2JSZXNwb25zZSIeCg1Kb2JMb2dSZXF1ZXN0Eg0KBWpvYklkGAEgASgJIkQKD0pvYkxvZ3NSZXNwb25zZRIjCgdqb2JJbmZvGAEgASgLMhIuam9icy52MS5Kb2JTdGF0dXMSDAoEbG9ncxgCIAEoCSJDCglKb2JTdGF0dXMSDgoGam9iX2lkGAEgASgJEg4KBnN0YXR1cxgCIAEoCRIWCg5zdGF0dXNfbWVzc2FnZRgDIAEoCTKWAgoKSm9iU2VydmljZRI7CgZOZXdKb2ISFi5qb2JzLnYxLk5ld0pvYlJlcXVlc3QaFy5qb2JzLnYxLk5ld0pvYlJlc3BvbnNlIgASPwoJR2V0U3RhdHVzEhYuam9icy52MS5Kb2JMb2dSZXF1ZXN0Ghguam9icy52MS5Kb2JMb2dzUmVzcG9uc2UiABJECgxTdHJlYW1TdGF0dXMSFi5qb2JzLnYxLkpvYkxvZ1JlcXVlc3QaGC5qb2JzLnYxLkpvYkxvZ3NSZXNwb25zZSIAMAESRAoJQ2FuY2VsSm9iEhkuam9icy52MS5DYW5jZWxKb2JSZXF1ZXN0Ghouam9icy52MS5DYW5jZWxKb2JSZXNwb25zZSIAQowBCgtjb20uam9icy52MUIJSm9ic1Byb3RvUAFaNWdpdGh1Yi5jb20vbWFrZW9wZW5zb3VyY2UvbGV2aWF0aGFuL2dlbmVyYXRlZC9qb2JzL3YxogIDSlhYqgIHSm9icy5WMcoCB0pvYnNcVjHiAhNKb2JzXFYxXEdQQk1ldGFkYXRh6gIISm9iczo6VjFiBnByb3RvMw", [file_types_v1_types]);
 
 /**
  * @generated from message jobs.v1.NewJobRequest
@@ -182,6 +182,18 @@ export const JobService: GenService<{
     output: typeof NewJobResponseSchema;
   },
   /**
+   * Gets job status at call time, whatever it may be
+   *
+   * @generated from rpc jobs.v1.JobService.GetStatus
+   */
+  getStatus: {
+    methodKind: "unary";
+    input: typeof JobLogRequestSchema;
+    output: typeof JobLogsResponseSchema;
+  },
+  /**
+   * Streams job status until it is complete
+   *
    * @generated from rpc jobs.v1.JobService.StreamStatus
    */
   streamStatus: {

spec/proto/jobs/v1/jobs.proto

@@ -8,6 +8,9 @@ import "types/v1/types.proto";
 
 service JobService {
   rpc NewJob(NewJobRequest) returns (NewJobResponse) {}
+  // Gets job status at call time, whatever it may be
+  rpc GetStatus (JobLogRequest) returns (JobLogsResponse) {}
+  // Streams job status until it is complete
   rpc StreamStatus(JobLogRequest) returns (stream JobLogsResponse) {}
   rpc CancelJob(CancelJobRequest) returns (CancelJobResponse) {}
 }

src/api/v1/job_impl.go

@@ -63,15 +63,24 @@ func (job *JobServer) NewJob(ctx context.Context, req *connect.Request[v1.NewJob
 	return res, nil
 }
 
+func (job *JobServer) GetStatus(_ context.Context, req *connect.Request[v1.JobLogRequest]) (*connect.Response[v1.JobLogsResponse], error) {
+	status, logs, err := job.srv.GetJobStatusAndLogs(req.Msg.GetJobId())
+	if err != nil {
+		return nil, err
+	}
+
+	res := connect.NewResponse(&v1.JobLogsResponse{
+		JobInfo: status.ToProto(),
+		Logs:    logs,
+	})
+	return res, nil
+}
+
 func (job *JobServer) StreamStatus(ctx context.Context, req *connect.Request[v1.JobLogRequest], stream *connect.ServerStream[v1.JobLogsResponse]) error {
 	streamFunc := func(jobInfo *models.Job, logs string) error {
 		return stream.Send(&v1.JobLogsResponse{
-			JobInfo: &v1.JobStatus{
-				JobId:         jobInfo.JobId,
-				Status:        string(jobInfo.Status),
-				StatusMessage: jobInfo.StatusMessage,
-			},
-			Logs: logs,
+			JobInfo: jobInfo.ToProto(),
+			Logs:    logs,
 		})
 	}
 

src/generated/jobs/v1/jobs.pb.go

@@ -3,6 +3,7 @@ package models
 import (
 	"context"
 	"fmt"
+	v1 "github.com/makeopensource/leviathan/generated/jobs/v1"
 	"github.com/rs/zerolog/log"
 	"gorm.io/gorm"
 )
@@ -63,6 +64,14 @@ type Job struct {
 	JobCtx           context.Context `gorm:"-"`
 }
 
+func (j *Job) ToProto() *v1.JobStatus {
+	return &v1.JobStatus{
+		JobId:         j.JobId,
+		Status:        string(j.Status),
+		StatusMessage: j.StatusMessage,
+	}
+}
+
 // ValidateForQueue checks for fields required before sending job to queue
 func (j *Job) ValidateForQueue() error {
 	if j == nil {

src/generated/jobs/v1/v1connect/jobs.connect.go

@@ -249,6 +249,15 @@ func (job *JobService) StreamJobAndLogs(
 	}
 }
 
+// GetJobStatusAndLogs gets the status once whatever it may be and current logs
+func (job *JobService) GetJobStatusAndLogs(jobUuid string) (*models.Job, string, error) {
+	jobInfo, _, logs, err := job.checkJob(jobUuid)
+	if err != nil {
+		return nil, "", err
+	}
+	return jobInfo, logs, nil
+}
+
 func (job *JobService) ListenToJobLogs(ctx context.Context, jobInfo *models.Job) chan string {
 	logChannel := make(chan string, 2)
 	go func(ctx context.Context) {