xinfra/server/internal/service/deployment.go

package service

import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/1024XEngineer/xinfra/server/internal/config"
	"github.com/1024XEngineer/xinfra/server/internal/model"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

const (
	DeploymentStatusPending   = "pending"
	DeploymentStatusRunning   = "running"
	DeploymentStatusSuccess   = "success"
	DeploymentStatusFailed    = "failed"
	DeploymentStatusCanceling = "canceling"
	DeploymentStatusCanceled  = "canceled"

	DeploymentEventLog    = "log"
	DeploymentEventStatus = "status"
	DeploymentEventResult = "result"
	DeploymentEventDone   = "done"
	DeploymentEventError  = "error"
)

var (
	ErrDeploymentNotConfigured = errors.New("ansible deployment service is not configured")
	ErrDeploymentNotFound      = errors.New("deployment not found")
	ErrDeploymentForbidden     = errors.New("deployment permission denied")
	ErrDeploymentInvalidState  = errors.New("deployment state does not allow this operation")
)

type DeploymentCreateRequest struct {
	Component      string         `json:"component"`
	BusinessLineID uint64         `json:"business_line_id"`
	Params         map[string]any `json:"params"`
}

type DeploymentEventRequest struct {
	Type    string         `json:"type"`
	Level   string         `json:"level"`
	Status  string         `json:"status"`
	Seq     uint64         `json:"seq"`
	Message string         `json:"message"`
	Payload map[string]any `json:"payload"`
}

type DeploymentFinishRequest struct {
	Status   string         `json:"status"`
	ExitCode int            `json:"exit_code"`
	Error    string         `json:"error"`
	Summary  map[string]any `json:"summary"`
}

type DeploymentEventEnvelope struct {
	Event model.DeploymentEvent
	Data  map[string]any
}

type DeploymentService struct {
	cfg    config.Config
	db     *gorm.DB
	client *http.Client

	mu          sync.Mutex
	subscribers map[string]map[chan DeploymentEventEnvelope]struct{}
}

func NewDeploymentService(cfg config.Config, db *gorm.DB) *DeploymentService {
	return &DeploymentService{
		cfg:         cfg,
		db:          db,
		client:      &http.Client{Timeout: 8 * time.Second},
		subscribers: make(map[string]map[chan DeploymentEventEnvelope]struct{}),
	}
}

func (s *DeploymentService) Create(ctx context.Context, req DeploymentCreateRequest, actorID uint64, actorName string) (model.Deployment, error) {
	component := strings.TrimSpace(strings.ToLower(req.Component))
	if !isSupportedDeploymentComponent(component) {
		return model.Deployment{}, fmt.Errorf("unsupported deployment component: %s", req.Component)
	}
	if req.BusinessLineID == 0 {
		return model.Deployment{}, errors.New("business_line_id is required")
	}
	payload, err := json.Marshal(req)
	if err != nil {
		return model.Deployment{}, err
	}

	var businessLine model.BusinessLine
	if err := s.db.WithContext(ctx).First(&businessLine, req.BusinessLineID).Error; err != nil {
		return model.Deployment{}, err
	}

	deployment := model.Deployment{
		DeploymentID:   newDeploymentID(),
		Component:      component,
		BusinessLineID: req.BusinessLineID,
		BusinessLine:   businessLine.Name,
		Status:         DeploymentStatusPending,
		RequestPayload: string(payload),
		CreatedBy:      actorID,
		CreatedByName:  actorName,
	}

	if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := tx.Create(&deployment).Error; err != nil {
			return err
		}
		_, err := s.appendEventTx(ctx, tx, deployment.DeploymentID, DeploymentEventStatus, "info", "deployment created", map[string]any{"status": DeploymentStatusPending})
		return err
	}); err != nil {
		return model.Deployment{}, err
	}

	if err := s.startPythonDeployment(ctx, deployment, req.Params); err != nil {
		_ = s.FailStart(ctx, deployment.DeploymentID, err)
		return deployment, err
	}
	return deployment, nil
}

func (s *DeploymentService) Get(ctx context.Context, deploymentID string) (model.Deployment, error) {
	var deployment model.Deployment
	if err := s.db.WithContext(ctx).Where("deployment_id = ?", deploymentID).First(&deployment).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return deployment, ErrDeploymentNotFound
		}
		return deployment, err
	}
	return deployment, nil
}

func (s *DeploymentService) Events(ctx context.Context, deploymentID string) ([]model.DeploymentEvent, error) {
	var events []model.DeploymentEvent
	err := s.db.WithContext(ctx).Where("deployment_id = ?", deploymentID).Order("seq ASC").Find(&events).Error
	return events, err
}

func (s *DeploymentService) AppendEvent(ctx context.Context, deploymentID string, req DeploymentEventRequest) (model.DeploymentEvent, error) {
	eventType := normalizeDeploymentEventType(req.Type)
	level := strings.TrimSpace(req.Level)
	if level == "" {
		level = "info"
	}
	payload := req.Payload
	if payload == nil {
		payload = map[string]any{}
	}
	if req.Status != "" {
		payload["status"] = normalizeDeploymentStatus(req.Status)
	}
	event, err := s.appendEvent(ctx, deploymentID, eventType, level, req.Message, payload)
	if err != nil {
		return event, err
	}
	if status, _ := payload["status"].(string); status != "" {
		_ = s.updateStatus(ctx, deploymentID, status, "")
	}
	return event, nil
}

func (s *DeploymentService) Finish(ctx context.Context, deploymentID string, req DeploymentFinishRequest) (model.Deployment, error) {
	status := normalizeDeploymentStatus(req.Status)
	if status == "" {
		status = DeploymentStatusFailed
	}
	payload := map[string]any{
		"status":    status,
		"exit_code": req.ExitCode,
		"summary":   req.Summary,
	}
	if req.Error != "" {
		payload["error"] = req.Error
	}
	result, _ := json.Marshal(payload)
	now := time.Now()
	var deployment model.Deployment
	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("deployment_id = ?", deploymentID).First(&deployment).Error; err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return ErrDeploymentNotFound
			}
			return err
		}
		updates := map[string]any{"status": status, "result_payload": string(result), "finished_at": &now}
		if err := tx.Model(&model.Deployment{}).Where("deployment_id = ?", deploymentID).Updates(updates).Error; err != nil {
			return err
		}
		eventType := DeploymentEventResult
		if status == DeploymentStatusFailed || status == DeploymentStatusCanceled {
			eventType = DeploymentEventError
		}
		if _, err := s.appendEventTx(ctx, tx, deploymentID, eventType, eventLevelForStatus(status), finishMessage(status, req.Error), payload); err != nil {
			return err
		}
		_, err := s.appendEventTx(ctx, tx, deploymentID, DeploymentEventDone, eventLevelForStatus(status), status, map[string]any{"status": status})
		return err
	})
	if err != nil {
		return deployment, err
	}
	deployment.Status = status
	deployment.ResultPayload = string(result)
	deployment.FinishedAt = &now
	return deployment, nil
}

func (s *DeploymentService) Cancel(ctx context.Context, deploymentID string) error {
	deployment, err := s.Get(ctx, deploymentID)
	if err != nil {
		return err
	}
	if !deploymentCancelable(deployment.Status) {
		return ErrDeploymentInvalidState
	}
	if err := s.updateStatus(ctx, deploymentID, DeploymentStatusCanceling, "cancel requested"); err != nil {
		return err
	}
	if s.cfg.AnsibleServiceBaseURL == "" {
		return ErrDeploymentNotConfigured
	}
	body, _ := json.Marshal(map[string]any{"deployment_id": deploymentID})
	path := s.cfg.AnsibleServiceBaseURL + "/internal/ansible/deployments/" + deploymentID + "/cancel"
	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, path, bytes.NewReader(body))
	if err != nil {
		return err
	}
	httpReq.Header.Set("Content-Type", "application/json")
	if s.cfg.AnsibleInternalToken != "" {
		httpReq.Header.Set("Authorization", "Bearer "+s.cfg.AnsibleInternalToken)
	}
	resp, err := s.client.Do(httpReq)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		data, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))
		return fmt.Errorf("ansible cancel failed: status %d: %s", resp.StatusCode, strings.TrimSpace(string(data)))
	}
	return nil
}

func (s *DeploymentService) Subscribe(deploymentID string) (chan DeploymentEventEnvelope, func()) {
	ch := make(chan DeploymentEventEnvelope, 64)
	s.mu.Lock()
	if s.subscribers[deploymentID] == nil {
		s.subscribers[deploymentID] = make(map[chan DeploymentEventEnvelope]struct{})
	}
	s.subscribers[deploymentID][ch] = struct{}{}
	s.mu.Unlock()
	return ch, func() {
		s.mu.Lock()
		if subscribers := s.subscribers[deploymentID]; subscribers != nil {
			delete(subscribers, ch)
			if len(subscribers) == 0 {
				delete(s.subscribers, deploymentID)
			}
		}
		s.mu.Unlock()
		close(ch)
	}
}

func (s *DeploymentService) FailStart(ctx context.Context, deploymentID string, cause error) error {
	_, err := s.Finish(ctx, deploymentID, DeploymentFinishRequest{
		Status: DeploymentStatusFailed,
		Error:  cause.Error(),
	})
	return err
}

func (s *DeploymentService) startPythonDeployment(ctx context.Context, deployment model.Deployment, params map[string]any) error {
	if s.cfg.AnsibleServiceBaseURL == "" {
		return ErrDeploymentNotConfigured
	}
	callbackBaseURL := strings.TrimRight(s.cfg.DeploymentCallbackBaseURL, "/")
	body, err := json.Marshal(map[string]any{
		"deployment_id": deployment.DeploymentID,
		"component":     deployment.Component,
		"callback_url":  callbackBaseURL + "/auth/internal/deployments/" + deployment.DeploymentID + "/events",
		"finish_url":    callbackBaseURL + "/auth/internal/deployments/" + deployment.DeploymentID + "/finish",
		"params":        params,
	})
	if err != nil {
		return err
	}
	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, s.cfg.AnsibleServiceBaseURL+"/internal/ansible/deploy", bytes.NewReader(body))
	if err != nil {
		return err
	}
	httpReq.Header.Set("Content-Type", "application/json")
	if s.cfg.AnsibleInternalToken != "" {
		httpReq.Header.Set("Authorization", "Bearer "+s.cfg.AnsibleInternalToken)
	}
	resp, err := s.client.Do(httpReq)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		data, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))
		return fmt.Errorf("ansible deploy failed: status %d: %s", resp.StatusCode, strings.TrimSpace(string(data)))
	}
	return nil
}

func (s *DeploymentService) appendEvent(ctx context.Context, deploymentID, eventType, level, message string, payload map[string]any) (model.DeploymentEvent, error) {
	var event model.DeploymentEvent
	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		var err error
		event, err = s.appendEventTx(ctx, tx, deploymentID, eventType, level, message, payload)
		return err
	})
	return event, err
}

func (s *DeploymentService) appendEventTx(ctx context.Context, tx *gorm.DB, deploymentID, eventType, level, message string, payload map[string]any) (model.DeploymentEvent, error) {
	var deployment model.Deployment
	if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("deployment_id = ?", deploymentID).First(&deployment).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return model.DeploymentEvent{}, ErrDeploymentNotFound
		}
		return model.DeploymentEvent{}, err
	}
	var maxSeq uint64
	if err := tx.Model(&model.DeploymentEvent{}).Where("deployment_id = ?", deploymentID).Select("COALESCE(MAX(seq), 0)").Scan(&maxSeq).Error; err != nil {
		return model.DeploymentEvent{}, err
	}
	payloadBody, _ := json.Marshal(payload)
	event := model.DeploymentEvent{
		DeploymentID: deploymentID,
		Seq:          maxSeq + 1,
		Type:         eventType,
		Level:        level,
		Message:      message,
		Payload:      string(payloadBody),
	}
	if err := tx.Create(&event).Error; err != nil {
		return model.DeploymentEvent{}, err
	}
	go s.broadcast(event, payload)
	return event, nil
}

func (s *DeploymentService) updateStatus(ctx context.Context, deploymentID string, status string, message string) error {
	status = normalizeDeploymentStatus(status)
	if status == "" {
		return nil
	}
	now := time.Now()
	updates := map[string]any{"status": status}
	if status == DeploymentStatusRunning {
		updates["started_at"] = &now
	}
	if isDeploymentTerminal(status) {
		updates["finished_at"] = &now
	}
	if err := s.db.WithContext(ctx).Model(&model.Deployment{}).Where("deployment_id = ?", deploymentID).Updates(updates).Error; err != nil {
		return err
	}
	if message != "" {
		_, err := s.appendEvent(ctx, deploymentID, DeploymentEventStatus, eventLevelForStatus(status), message, map[string]any{"status": status})
		return err
	}
	return nil
}

func (s *DeploymentService) broadcast(event model.DeploymentEvent, data map[string]any) {
	envelope := DeploymentEventEnvelope{Event: event, Data: eventData(event, data)}
	s.mu.Lock()
	subscribers := make([]chan DeploymentEventEnvelope, 0, len(s.subscribers[event.DeploymentID]))
	for ch := range s.subscribers[event.DeploymentID] {
		subscribers = append(subscribers, ch)
	}
	s.mu.Unlock()
	for _, ch := range subscribers {
		select {
		case ch <- envelope:
		default:
		}
	}
}

func eventData(event model.DeploymentEvent, data map[string]any) map[string]any {
	out := map[string]any{
		"deployment_id": event.DeploymentID,
		"seq":           event.Seq,
		"type":          event.Type,
		"level":         event.Level,
		"message":       event.Message,
		"created_at":    event.CreatedAt,
	}
	for key, value := range data {
		out[key] = value
	}
	return out
}

func DeploymentEventData(event model.DeploymentEvent) map[string]any {
	payload := map[string]any{}
	if strings.TrimSpace(event.Payload) != "" {
		_ = json.Unmarshal([]byte(event.Payload), &payload)
	}
	return eventData(event, payload)
}

func isSupportedDeploymentComponent(component string) bool {
	switch component {
	case "mysql", "openresty":
		return true
	default:
		return false
	}
}

func normalizeDeploymentEventType(value string) string {
	switch strings.TrimSpace(strings.ToLower(value)) {
	case DeploymentEventStatus:
		return DeploymentEventStatus
	case DeploymentEventResult:
		return DeploymentEventResult
	case DeploymentEventDone:
		return DeploymentEventDone
	case DeploymentEventError:
		return DeploymentEventError
	default:
		return DeploymentEventLog
	}
}

func normalizeDeploymentStatus(value string) string {
	switch strings.TrimSpace(strings.ToLower(value)) {
	case DeploymentStatusPending, DeploymentStatusRunning, DeploymentStatusSuccess, DeploymentStatusFailed, DeploymentStatusCanceling, DeploymentStatusCanceled:
		return strings.TrimSpace(strings.ToLower(value))
	default:
		return ""
	}
}

func isDeploymentTerminal(status string) bool {
	return status == DeploymentStatusSuccess || status == DeploymentStatusFailed || status == DeploymentStatusCanceled
}

func deploymentCancelable(status string) bool {
	return status == DeploymentStatusPending || status == DeploymentStatusRunning || status == DeploymentStatusCanceling
}

func eventLevelForStatus(status string) string {
	if status == DeploymentStatusFailed || status == DeploymentStatusCanceled {
		return "error"
	}
	return "info"
}

func finishMessage(status, fallback string) string {
	if fallback != "" {
		return fallback
	}
	switch status {
	case DeploymentStatusSuccess:
		return "deployment completed"
	case DeploymentStatusCanceled:
		return "deployment canceled"
	default:
		return "deployment failed"
	}
}

func newDeploymentID() string {
	now := time.Now()
	buf := make([]byte, 3)
	if _, err := rand.Read(buf); err != nil {
		return fmt.Sprintf("CMP-%s-%d", now.Format("20060102"), now.UnixNano()%1000000)
	}
	return fmt.Sprintf("CMP-%s-%s", now.Format("20060102"), strings.ToUpper(hex.EncodeToString(buf)))
}

func SortedDeploymentEvents(events []model.DeploymentEvent) {
	sort.Slice(events, func(i, j int) bool {
		return events[i].Seq < events[j].Seq
	})
}