MATTERMOST_OPENCLAW_LIVESTATUS/plugin/server/api.go

package main

import (
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strings"

	"github.com/mattermost/mattermost/server/public/model"
	"github.com/mattermost/mattermost/server/public/plugin"
)

// ServeHTTP handles HTTP requests to the plugin.
func (p *Plugin) ServeHTTP(c *plugin.Context, w http.ResponseWriter, r *http.Request) {
	// Auth middleware: validate shared secret
	config := p.getConfiguration()
	if config.SharedSecret != "" {
		auth := r.Header.Get("Authorization")
		expected := "Bearer " + config.SharedSecret
		if auth != expected {
			http.Error(w, `{"error": "unauthorized"}`, http.StatusUnauthorized)
			return
		}
	}

	path := r.URL.Path

	switch {
	case path == "/api/v1/health" && r.Method == http.MethodGet:
		p.handleHealth(w, r)
	case path == "/api/v1/sessions" && r.Method == http.MethodGet:
		p.handleListSessions(w, r)
	case path == "/api/v1/sessions" && r.Method == http.MethodPost:
		p.handleCreateSession(w, r)
	case strings.HasPrefix(path, "/api/v1/sessions/") && r.Method == http.MethodPut:
		sessionKey := strings.TrimPrefix(path, "/api/v1/sessions/")
		p.handleUpdateSession(w, r, sessionKey)
	case strings.HasPrefix(path, "/api/v1/sessions/") && r.Method == http.MethodDelete:
		sessionKey := strings.TrimPrefix(path, "/api/v1/sessions/")
		p.handleDeleteSession(w, r, sessionKey)
	default:
		http.NotFound(w, r)
	}
}

// handleHealth returns plugin health status.
func (p *Plugin) handleHealth(w http.ResponseWriter, r *http.Request) {
	sessions, err := p.store.ListActiveSessions()
	count := 0
	if err == nil {
		count = len(sessions)
	}

	writeJSON(w, http.StatusOK, map[string]interface{}{
		"status":          "healthy",
		"active_sessions": count,
		"plugin_id":       "com.openclaw.livestatus",
	})
}

// handleListSessions returns all active sessions.
func (p *Plugin) handleListSessions(w http.ResponseWriter, r *http.Request) {
	sessions, err := p.store.ListActiveSessions()
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
		return
	}
	writeJSON(w, http.StatusOK, sessions)
}

// CreateSessionRequest is the request body for creating a new session.
type CreateSessionRequest struct {
	SessionKey string `json:"session_key"`
	ChannelID  string `json:"channel_id"`
	RootID     string `json:"root_id,omitempty"`
	AgentID    string `json:"agent_id"`
	BotUserID  string `json:"bot_user_id,omitempty"`
}

// handleCreateSession creates a new custom_livestatus post and starts tracking.
func (p *Plugin) handleCreateSession(w http.ResponseWriter, r *http.Request) {
	var req CreateSessionRequest
	if err := readJSON(r, &req); err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
		return
	}

	if req.SessionKey == "" || req.ChannelID == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "session_key and channel_id required"})
		return
	}

	// Check max active sessions
	config := p.getConfiguration()
	sessions, _ := p.store.ListActiveSessions()
	if len(sessions) >= config.MaxActiveSessions {
		writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "max active sessions reached"})
		return
	}

	// Create the custom post — UserId is required
	// Use the bot user ID passed in the request, or fall back to plugin bot
	userID := req.BotUserID
	if userID == "" {
		// Try to get plugin's own bot
		userID = p.getBotUserID()
	}
	if userID == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "bot_user_id required (no plugin bot available)"})
		return
	}

	post := &model.Post{
		UserId:    userID,
		ChannelId: req.ChannelID,
		RootId:    req.RootID,
		Type:      "custom_livestatus",
		Message:   "", // Custom post types don't need a message
	}
	post.AddProp("session_key", req.SessionKey)
	post.AddProp("agent_id", req.AgentID)
	post.AddProp("status", "active")

	createdPost, appErr := p.API.CreatePost(post)
	if appErr != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": appErr.Error()})
		return
	}

	// Store session data
	sessionData := SessionData{
		SessionKey: req.SessionKey,
		PostID:     createdPost.Id,
		ChannelID:  req.ChannelID,
		RootID:     req.RootID,
		AgentID:    req.AgentID,
		Status:     "active",
		Lines:      []string{},
	}
	if err := p.store.SaveSession(req.SessionKey, sessionData); err != nil {
		p.API.LogWarn("Failed to save session", "error", err.Error())
	}

	// Broadcast initial state
	p.broadcastUpdate(req.ChannelID, sessionData)

	writeJSON(w, http.StatusCreated, map[string]string{
		"post_id":     createdPost.Id,
		"session_key": req.SessionKey,
	})
}

// UpdateSessionRequest is the request body for updating a session.
type UpdateSessionRequest struct {
	Status      string        `json:"status"`
	Lines       []string      `json:"lines"`
	ElapsedMs   int64         `json:"elapsed_ms"`
	TokenCount  int           `json:"token_count"`
	Children    []SessionData `json:"children,omitempty"`
	StartTimeMs int64         `json:"start_time_ms"`
}

// handleUpdateSession updates session data and broadcasts via WebSocket.
// Critically: does NOT call any Mattermost post API — no "(edited)" label.
func (p *Plugin) handleUpdateSession(w http.ResponseWriter, r *http.Request, sessionKey string) {
	var req UpdateSessionRequest
	if err := readJSON(r, &req); err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
		return
	}

	// Get existing session
	existing, err := p.store.GetSession(sessionKey)
	if err != nil || existing == nil {
		writeJSON(w, http.StatusNotFound, map[string]string{"error": "session not found"})
		return
	}

	// Update fields
	existing.Status = req.Status
	existing.Lines = req.Lines
	existing.ElapsedMs = req.ElapsedMs
	existing.TokenCount = req.TokenCount
	existing.Children = req.Children
	if req.StartTimeMs > 0 {
		existing.StartTimeMs = req.StartTimeMs
	}

	// Save to KV store
	if err := p.store.SaveSession(sessionKey, *existing); err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
		return
	}

	// Broadcast update via WebSocket (no Mattermost post API call!)
	p.broadcastUpdate(existing.ChannelID, *existing)

	writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
}

// handleDeleteSession marks a session as complete.
func (p *Plugin) handleDeleteSession(w http.ResponseWriter, r *http.Request, sessionKey string) {
	existing, err := p.store.GetSession(sessionKey)
	if err != nil || existing == nil {
		writeJSON(w, http.StatusNotFound, map[string]string{"error": "session not found"})
		return
	}

	// Mark as done
	existing.Status = "done"

	// Update the post props to reflect completion (one final API call)
	post, appErr := p.API.GetPost(existing.PostID)
	if appErr == nil && post != nil {
		post.AddProp("status", "done")
		post.AddProp("final_lines", existing.Lines)
		post.AddProp("elapsed_ms", existing.ElapsedMs)
		post.AddProp("token_count", existing.TokenCount)
		_, _ = p.API.UpdatePost(post)
	}

	// Broadcast final state
	p.broadcastUpdate(existing.ChannelID, *existing)

	// Clean up KV store
	_ = p.store.DeleteSession(sessionKey)

	writeJSON(w, http.StatusOK, map[string]string{"status": "done"})
}

// Helper functions

func readJSON(r *http.Request, v interface{}) error {
	body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20)) // 1MB limit
	if err != nil {
		return fmt.Errorf("read body: %w", err)
	}
	defer r.Body.Close()
	return json.Unmarshal(body, v)
}

func writeJSON(w http.ResponseWriter, status int, v interface{}) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	json.NewEncoder(w).Encode(v)
}