[MM-54589] Implement job status API (#542)

* Implement job status API

* Add job status check

* Start recordings when bot posts status update

* Bump calls-recorder version

* Bump recorder

go.mod

@@ -16,8 +16,9 @@ require (
 	github.com/mattermost/calls-offloader v0.3.2
 	github.com/mattermost/calls-recorder v0.4.2
 	github.com/mattermost/logr/v2 v2.0.16
+	github.com/mattermost/mattermost-plugin-calls/server/public v0.0.1
 	github.com/mattermost/mattermost/server/public v0.0.9-0.20230824163353-69c11cfe1403
-	github.com/mattermost/rtcd v0.11.1
+	github.com/mattermost/rtcd v0.11.3-0.20230913232654-bdaaa43e3e1c
 	github.com/mattermost/squirrel v0.2.0
 	github.com/pion/interceptor v0.1.12
 	github.com/pion/rtp v1.7.13
@@ -29,6 +30,8 @@ require (
 
 replace github.com/pion/interceptor v0.1.12 => github.com/streamer45/interceptor v0.0.0-20230202152215-57f3ac9e7696
 
+replace github.com/mattermost/mattermost-plugin-calls/server/public => ./server/public
+
 require (
 	git.mills.io/prologic/bitcask v1.0.2 // indirect
 	github.com/abcum/lcp v0.0.0-20201209214815-7a3f3840be81 // indirect
@@ -76,7 +79,7 @@ require (
 	github.com/prometheus/client_model v0.4.0 // indirect
 	github.com/prometheus/common v0.44.0 // indirect
 	github.com/prometheus/procfs v0.11.0 // indirect
-	github.com/pyroscope-io/godeltaprof v0.1.1 // indirect
+	github.com/pyroscope-io/godeltaprof v0.1.2 // indirect
 	github.com/segmentio/backo-go v1.0.1 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/tidwall/gjson v1.14.4 // indirect

go.sum

@@ -300,8 +300,8 @@ github.com/mattermost/logr/v2 v2.0.16 h1:jnePX4cPskC3WDFvUardh/xZfxNdsFXbEERJQ1k
 github.com/mattermost/logr/v2 v2.0.16/go.mod h1:1dm/YhTpozsqANXxo5Pi5zYLBsal2xY0pX+JZNbzYJY=
 github.com/mattermost/mattermost/server/public v0.0.9-0.20230824163353-69c11cfe1403 h1:/rxsEaisu+Rb5mWfoIEnbFqscJeKVkspj+BWzchUAfs=
 github.com/mattermost/mattermost/server/public v0.0.9-0.20230824163353-69c11cfe1403/go.mod h1:sgXQrYzs+IJy51mB8E8OBljagk2u3YwQRoYlBH5goiw=
-github.com/mattermost/rtcd v0.11.1 h1:xaP/s0/WX8rDqHq05l8b4QLLJuRZXucr9Qh6cHTQSHk=
-github.com/mattermost/rtcd v0.11.1/go.mod h1:ketmoC7+9IOjynE5YJgR6GrFTG1b78byVlkVsLcCDa0=
+github.com/mattermost/rtcd v0.11.3-0.20230913232654-bdaaa43e3e1c h1:I869qzi119z/b5hEcsFGrbDTkS9f8HAVg+IO56Q6diw=
+github.com/mattermost/rtcd v0.11.3-0.20230913232654-bdaaa43e3e1c/go.mod h1:7C412PFWeVSZK0E8Ruht5ArHP+zxhSr/sMDJvnUFDe4=
 github.com/mattermost/squirrel v0.2.0 h1:8ZWeyf+MWQ2cL7hu9REZgLtz2IJi51qqZEovI3T3TT8=
 github.com/mattermost/squirrel v0.2.0/go.mod h1:NPPtk+CdpWre4GxMGoOpzEVFVc0ZoEFyJBZGCtn9nSU=
 github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
@@ -433,8 +433,8 @@ github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7z
 github.com/prometheus/procfs v0.11.0 h1:5EAgkfkMl659uZPbe9AS2N68a7Cc1TJbPEuGzFuRbyk=
 github.com/prometheus/procfs v0.11.0/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPHWJq+XBB/FM=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
-github.com/pyroscope-io/godeltaprof v0.1.1 h1:+Mmi+b9gR3s/qufuQSxOBjyXZR1fmvS/C12Q73PIPvw=
-github.com/pyroscope-io/godeltaprof v0.1.1/go.mod h1:psMITXp90+8pFenXkKIpNhrfmI9saQnPbba27VIaiQE=
+github.com/pyroscope-io/godeltaprof v0.1.2 h1:MdlEmYELd5w+lvIzmZvXGNMVzW2Qc9jDMuJaPOR75g4=
+github.com/pyroscope-io/godeltaprof v0.1.2/go.mod h1:psMITXp90+8pFenXkKIpNhrfmI9saQnPbba27VIaiQE=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=

plugin.json

@@ -199,6 +199,6 @@
   "props": {
     "min_rtcd_version": "v0.10.1",
     "min_offloader_version": "v0.3.2",
-    "calls_recorder_version": "v0.4.2"
+    "calls_recorder_version": "v0.5.1"
   }
 }

server/bot_api.go

@@ -11,13 +11,16 @@ import (
 	"regexp"
 	"time"
 
+	"github.com/mattermost/mattermost-plugin-calls/server/public"
+
 	"github.com/mattermost/mattermost/server/public/model"
 )
 
 var botChRE = regexp.MustCompile(`^\/bot\/channels\/([a-z0-9]+)$`)
 var botUserImageRE = regexp.MustCompile(`^\/bot\/users\/([a-z0-9]+)\/image$`)
 var botUploadsRE = regexp.MustCompile(`^\/bot\/uploads\/?([a-z0-9]+)?$`)
 var botRecordingsRE = regexp.MustCompile(`^\/bot\/calls\/([a-z0-9]+)\/recordings$`)
+var botJobsStatusRE = regexp.MustCompile(`^\/bot\/calls\/([a-z0-9]+)\/jobs\/([a-z0-9]+)\/status$`)
 
 func (p *Plugin) getBotID() string {
 	if p.botSession != nil {
@@ -245,6 +248,88 @@ func (p *Plugin) handleBotPostRecordings(w http.ResponseWriter, r *http.Request,
 	res.Msg = "success"
 }
 
+func (p *Plugin) handleBotPostJobsStatus(w http.ResponseWriter, r *http.Request, callID, jobID string) {
+	var res httpResponse
+	defer p.httpAudit("handleBotPostJobsStatus", &res, w, r)
+
+	var status public.JobStatus
+	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, requestBodyMaxSizeBytes)).Decode(&status); err != nil {
+		res.Err = "failed to decode request body: " + err.Error()
+		res.Code = http.StatusBadRequest
+		return
+	}
+
+	state, err := p.lockCall(callID)
+	if err != nil {
+		res.Err = fmt.Errorf("failed to lock call: %w", err).Error()
+		res.Code = http.StatusInternalServerError
+		return
+	}
+	defer p.unlockCall(callID)
+
+	if state == nil || state.Call == nil {
+		res.Err = "no call ongoing"
+		res.Code = http.StatusBadRequest
+		return
+	}
+
+	if status.JobType == public.JobTypeRecording {
+		if state.Call.Recording == nil {
+			res.Err = "no recording ongoing"
+			res.Code = http.StatusBadRequest
+			return
+		}
+
+		if state.Call.Recording.ID != jobID {
+			res.Err = "invalid recording job ID"
+			res.Code = http.StatusBadRequest
+			return
+		}
+
+		if state.Call.Recording.EndAt > 0 {
+			res.Err = "recording has ended"
+			res.Code = http.StatusBadRequest
+			return
+		}
+
+		if status.Status == public.JobStatusTypeFailed {
+			p.LogDebug("recording has failed", "jobID", jobID)
+			state.Call.Recording.EndAt = time.Now().UnixMilli()
+			state.Call.Recording.Err = status.Error
+		} else if status.Status == public.JobStatusTypeStarted {
+			if state.Call.Recording.StartAt > 0 {
+				res.Err = "recording has already started"
+				res.Code = http.StatusBadRequest
+				return
+			}
+			p.LogDebug("recording has started", "jobID", jobID)
+			state.Call.Recording.StartAt = time.Now().UnixMilli()
+		} else {
+			res.Err = "unsupported status type"
+			res.Code = http.StatusBadRequest
+			return
+		}
+
+		if err := p.kvSetChannelState(callID, state); err != nil {
+			res.Err = fmt.Errorf("failed to set channel state: %w", err).Error()
+			res.Code = http.StatusInternalServerError
+			return
+		}
+
+		p.publishWebSocketEvent(wsEventCallRecordingState, map[string]interface{}{
+			"callID":   callID,
+			"recState": state.Call.Recording.getClientState().toMap(),
+		}, &model.WebsocketBroadcast{ChannelId: callID, ReliableClusterSend: true})
+
+		res.Code = http.StatusOK
+		res.Msg = "success"
+		return
+	}
+
+	res.Err = "bad request"
+	res.Code = http.StatusBadRequest
+}
+
 func (p *Plugin) handleBotAPI(w http.ResponseWriter, r *http.Request) {
 	if !p.licenseChecker.RecordingsAllowed() {
 		http.Error(w, "Forbidden", http.StatusForbidden)
@@ -286,6 +371,11 @@ func (p *Plugin) handleBotAPI(w http.ResponseWriter, r *http.Request) {
 			p.handleBotPostRecordings(w, r, matches[1])
 			return
 		}
+
+		if matches := botJobsStatusRE.FindStringSubmatch(r.URL.Path); len(matches) == 3 {
+			p.handleBotPostJobsStatus(w, r, matches[1], matches[2])
+			return
+		}
 	}
 
 	http.NotFound(w, r)

server/public/go.mod

@@ -0,0 +1,3 @@
+module github.com/mattermost/mattermost-plugin-calls/server/public
+
+go 1.19

server/public/job.go

@@ -0,0 +1,20 @@
+package public
+
+type JobType string
+
+const (
+	JobTypeRecording JobType = "recording"
+)
+
+type JobStatusType string
+
+const (
+	JobStatusTypeStarted JobStatusType = "started"
+	JobStatusTypeFailed                = "failed"
+)
+
+type JobStatus struct {
+	JobType JobType
+	Status  JobStatusType
+	Error   string `json:"omitempty"`
+}

server/session.go

@@ -118,15 +118,16 @@ func (p *Plugin) addUserSession(state *channelState, userID, connID, channelID,
 		return nil, fmt.Errorf("user cannot join because of limits")
 	}
 
-	// When the bot joins the call it means the recording has started.
+	// When the bot joins the call it means the recording is starting. The actual
+	// start time is when the bot sends the status update through the API.
 	if userID == p.getBotID() {
-		if state.Call.Recording != nil && state.Call.Recording.StartAt == 0 {
+		if state.Call.Recording != nil && state.Call.Recording.StartAt == 0 && state.Call.Recording.BotConnID == "" {
 			if state.Call.Recording.ID != jobID {
 				return nil, fmt.Errorf("invalid job ID for recording")
 			}
-			state.Call.Recording.StartAt = time.Now().UnixMilli()
+			p.LogDebug("bot joined, recording is starting", "jobID", jobID)
 			state.Call.Recording.BotConnID = connID
-		} else if state.Call.Recording == nil || state.Call.Recording.StartAt > 0 {
+		} else if state.Call.Recording == nil || state.Call.Recording.StartAt > 0 || state.Call.Recording.BotConnID != "" {
 			// In this case we should fail to prevent the bot from recording
 			// without consent.
 			return nil, fmt.Errorf("recording not in progress or already started")