2026-07-23 12:04:37 +08:00
package service
import (
"bytes"
"context"
"crypto/rand"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"net"
"net/http"
"regexp"
"strings"
"sync"
"time"
"github.com/1024XEngineer/xinfra/server/internal/config"
"github.com/1024XEngineer/xinfra/server/internal/model"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
var dnsLabelPattern = regexp . MustCompile ( `^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?$` )
type MySQLDeliveryInput struct {
BusinessLineID uint64 `json:"business_line_id" binding:"required"`
TargetID uint64 `json:"target_id" binding:"required"`
Namespace string `json:"namespace" binding:"required"`
InstanceName string `json:"instance_name" binding:"required"`
MySQLVersion string `json:"mysql_version"`
CPUMilli int64 `json:"cpu_milli" binding:"required"`
MemoryMi int64 `json:"memory_mi" binding:"required"`
StorageGi int64 `json:"storage_gi" binding:"required"`
}
type deliveryPayload struct {
MySQLDeliveryInput
TargetType string `json:"target_type"`
}
// targetMetadata describes the native VM候选节点池以及部署形态，存储在 DeploymentTarget.Metadata (JSON)。
type targetMetadata struct {
Topology string `json:"topology"`
MySQLPort int `json:"mysql_port"`
Hosts [] targetHost `json:"hosts"`
}
type targetHost struct {
Name string `json:"name"`
IP string `json:"ip"`
}
func parseTargetMetadata ( raw string ) targetMetadata {
meta := targetMetadata {}
if strings . TrimSpace ( raw ) != "" {
_ = json . Unmarshal ([] byte ( raw ), & meta )
}
if meta . Topology == "" {
meta . Topology = "standalone"
}
if meta . MySQLPort == 0 {
meta . MySQLPort = 3307
}
return meta
}
// firstFreeHost 返回候选池中第一个未被占用的节点。
func firstFreeHost ( hosts [] targetHost , occupied [] string ) * targetHost {
taken := make ( map [ string ] bool , len ( occupied ))
for _ , h := range occupied {
taken [ h ] = true
}
for i := range hosts {
if ! taken [ hosts [ i ]. Name ] {
return & hosts [ i ]
}
}
return nil
}
type DeliveryService struct {
db * gorm . DB
cfg config . Config
awx * AWXClient
audit * AuditService
executionMu sync . Mutex
}
func ( s * DeliveryService ) DB () * gorm . DB { return s . db }
func NewDeliveryService ( cfg config . Config , db * gorm . DB , audit * AuditService ) * DeliveryService {
return & DeliveryService { db : db , cfg : cfg , awx : NewAWXClient ( cfg . AWXBaseURL , cfg . AWXToken ), audit : audit }
}
func ( s * DeliveryService ) CreateTask ( ctx context . Context , userID uint64 , isAdmin bool , idempotencyKey string , input MySQLDeliveryInput ) ( * model . DeliveryTask , bool , error ) {
idempotencyKey = strings . TrimSpace ( idempotencyKey )
if idempotencyKey == "" || len ( idempotencyKey ) > 128 {
return nil , false , fmt . Errorf ( "Idempotency-Key header is required and must not exceed 128 characters" )
}
if err := validateDeliveryInput ( input ); err != nil {
return nil , false , err
}
var existing model . DeliveryTask
if err := s . db . WithContext ( ctx ). Where ( "idempotency_key = ?" , idempotencyKey ). First ( & existing ). Error ; err == nil {
2026-07-23 14:29:32 +08:00
if existing . RequestedBy != userID {
return nil , false , fmt . Errorf ( "idempotency key is already in use by another user" )
}
2026-07-23 12:04:37 +08:00
return & existing , true , nil
} else if ! errors . Is ( err , gorm . ErrRecordNotFound ) {
return nil , false , err
}
var target model . DeploymentTarget
if err := s . db . WithContext ( ctx ). First ( & target , "id = ? AND enabled = ?" , input . TargetID , true ). Error ; err != nil {
return nil , false , fmt . Errorf ( "deployment target is unavailable: %w" , err )
}
if target . TargetType != "k8s" {
return nil , false , fmt . Errorf ( "target type %q is not supported in the MVP" , target . TargetType )
}
if ! isAdmin {
var count int64
if err := s . db . WithContext ( ctx ). Model ( & model . BusinessLineUser {}).
Where ( "business_line_id = ? AND user_id = ?" , input . BusinessLineID , userID ). Count ( & count ). Error ; err != nil {
return nil , false , err
}
if count == 0 {
return nil , false , fmt . Errorf ( "user is not authorized for this business line" )
}
}
if input . MySQLVersion == "" {
input . MySQLVersion = "8.0"
}
payload := deliveryPayload { MySQLDeliveryInput : input , TargetType : target . TargetType }
raw , err := json . Marshal ( payload )
if err != nil {
return nil , false , err
}
digest := sha256 . Sum256 ( raw )
task := model . DeliveryTask {
ID : randomUUID (),
BusinessLineID : input . BusinessLineID ,
RequestedBy : userID ,
TargetType : target . TargetType ,
TargetID : target . ID ,
Namespace : input . Namespace ,
InstanceName : input . InstanceName ,
Status : model . TaskPending ,
ImmutablePayload : string ( raw ),
PayloadHash : hex . EncodeToString ( digest [:]),
IdempotencyKey : idempotencyKey ,
}
if err := s . db . WithContext ( ctx ). Create ( & task ). Error ; err != nil {
if lookupErr := s . db . WithContext ( ctx ). Where ( "idempotency_key = ?" , idempotencyKey ). First ( & existing ). Error ; lookupErr == nil {
return & existing , true , nil
}
return nil , false , err
}
_ = s . db . WithContext ( ctx ). Create ( & model . TaskEvent { TaskID : task . ID , ToState : model . TaskPending , Message : "delivery task created" }). Error
return & task , false , nil
}
2026-07-23 14:29:32 +08:00
var supportedMySQLVersions = map [ string ] bool { "8.0" : true }
2026-07-23 12:04:37 +08:00
func validateDeliveryInput ( input MySQLDeliveryInput ) error {
if len ( input . Namespace ) > 63 || ! dnsLabelPattern . MatchString ( input . Namespace ) {
return fmt . Errorf ( "namespace must be a valid Kubernetes DNS label" )
}
if len ( input . InstanceName ) > 63 || ! dnsLabelPattern . MatchString ( input . InstanceName ) {
return fmt . Errorf ( "instance_name must be a valid Kubernetes DNS label" )
}
2026-07-23 14:29:32 +08:00
if input . CPUMilli < 100 || input . CPUMilli > 64000 || input . MemoryMi < 1024 || input . MemoryMi > 4096 || input . StorageGi < 10 || input . StorageGi > 100 {
return fmt . Errorf ( "requested resources are outside the supported range (memory: 1024-4096 MiB, storage: 10-100 GiB)" )
}
if input . MySQLVersion != "" && ! supportedMySQLVersions [ input . MySQLVersion ] {
return fmt . Errorf ( "unsupported mysql_version %q, supported: 8.0" , input . MySQLVersion )
2026-07-23 12:04:37 +08:00
}
return nil
}
func ( s * DeliveryService ) ListTasks ( ctx context . Context , userID uint64 , isAdmin bool , businessLineID uint64 ) ([] model . DeliveryTask , error ) {
query := s . db . WithContext ( ctx ). Order ( "created_at DESC" )
if businessLineID != 0 {
query = query . Where ( "business_line_id = ?" , businessLineID )
}
if ! isAdmin {
query = query . Where ( "business_line_id IN (?)" , s . db . Model ( & model . BusinessLineUser {}). Select ( "business_line_id" ). Where ( "user_id = ?" , userID ))
}
var tasks [] model . DeliveryTask
return tasks , query . Find ( & tasks ). Error
}
func ( s * DeliveryService ) GetTask ( ctx context . Context , taskID string , userID uint64 , isAdmin bool ) ( * model . DeliveryTask , [] model . TaskEvent , error ) {
query := s . db . WithContext ( ctx ). Where ( "id = ?" , taskID )
if ! isAdmin {
query = query . Where ( "business_line_id IN (?)" , s . db . Model ( & model . BusinessLineUser {}). Select ( "business_line_id" ). Where ( "user_id = ?" , userID ))
}
var task model . DeliveryTask
if err := query . First ( & task ). Error ; err != nil {
return nil , nil , err
}
var events [] model . TaskEvent
if err := s . db . WithContext ( ctx ). Where ( "task_id = ?" , taskID ). Order ( "id ASC" ). Find ( & events ). Error ; err != nil {
return nil , nil , err
}
return & task , events , nil
}
func ( s * DeliveryService ) Cancel ( ctx context . Context , taskID string , userID uint64 , isAdmin bool ) error {
task , _ , err := s . GetTask ( ctx , taskID , userID , isAdmin )
if err != nil {
return err
}
if task . Status == model . TaskPending {
return s . transition ( ctx , task , model . TaskCanceled , "canceled before dispatch" , "" )
}
if task . Status != model . TaskRunning && task . Status != model . TaskDispatching {
return fmt . Errorf ( "task in state %q cannot be canceled" , task . Status )
}
var job model . ExecutionJob
if err := s . db . WithContext ( ctx ). Where ( "task_id = ?" , task . ID ). First ( & job ). Error ; err != nil {
return err
}
if err := s . awx . Cancel ( ctx , job . ExecutorJobID ); err != nil {
return err
}
return s . transition ( ctx , task , model . TaskCanceling , "cancel requested in AWX" , "" )
}
func ( s * DeliveryService ) claimAndReserve ( ctx context . Context ) ( * model . DeliveryTask , error ) {
var task model . DeliveryTask
var payload deliveryPayload
var target model . DeploymentTarget
dispatchable := false
err := s . db . WithContext ( ctx ). Transaction ( func ( tx * gorm . DB ) error {
if err := tx . Clauses ( clause . Locking { Strength : "UPDATE" , Options : "SKIP LOCKED" }). Where ( "status = ?" , model . TaskPending ). Order ( "created_at ASC" ). First ( & task ). Error ; err != nil {
return err
}
if err := tx . Clauses ( clause . Locking { Strength : "UPDATE" }). First ( & target , "id = ? AND enabled = ?" , task . TargetID , true ). Error ; err != nil {
return s . failInTransaction ( tx , & task , model . TaskValidationFailed , "deployment target is unavailable" )
}
if err := json . Unmarshal ([] byte ( task . ImmutablePayload ), & payload ); err != nil {
return s . failInTransaction ( tx , & task , model . TaskValidationFailed , "stored deployment payload is invalid" )
}
activeStates := [] string { model . TaskValidating , model . TaskDispatching , model . TaskRunning , model . TaskCanceling }
checks := [] struct {
query string
args [] any
limit int
message string
}{
{ "status IN ?" , [] any { activeStates }, s . cfg . DeliveryGlobalLimit , "global concurrency limit reached" },
{ "status IN ? AND target_id = ?" , [] any { activeStates , task . TargetID }, s . cfg . DeliveryTargetLimit , "target concurrency limit reached" },
{ "status IN ? AND business_line_id = ?" , [] any { activeStates , task . BusinessLineID }, s . cfg . DeliveryBusinessLimit , "business line concurrency limit reached" },
{ "status IN ? AND target_id = ? AND namespace = ?" , [] any { activeStates , task . TargetID , task . Namespace }, 1 , "namespace already has an active MySQL delivery" },
}
for _ , check := range checks {
var count int64
if check . limit > 0 {
if err := tx . Model ( & model . DeliveryTask {}). Where ( check . query , check . args ... ). Count ( & count ). Error ; err != nil {
return err
}
if count >= int64 ( check . limit ) {
return fmt . Errorf ( "defer: %s" , check . message )
}
}
}
var quota model . ResourceQuota
if err := tx . Clauses ( clause . Locking { Strength : "UPDATE" }). Where ( "business_line_id = ? AND target_id = ?" , task . BusinessLineID , task . TargetID ). First ( & quota ). Error ; err != nil {
return s . failInTransaction ( tx , & task , model . TaskValidationFailed , "resource quota is not configured" )
}
type totals struct { CPU , Memory , Storage , Instances int64 }
var used , reserved totals
if err := tx . Model ( & model . ResourceUsage {}). Select ( "COALESCE(SUM(cpu_milli),0) cpu, COALESCE(SUM(memory_mi),0) memory, COALESCE(SUM(storage_gi),0) storage, COALESCE(SUM(instance_count),0) instances" ). Where ( "business_line_id = ? AND target_id = ? AND status = ?" , task . BusinessLineID , task . TargetID , "active" ). Scan ( & used ). Error ; err != nil {
return err
}
2026-07-23 14:29:32 +08:00
if err := tx . Model ( & model . ResourceReservation {}). Select ( "COALESCE(SUM(cpu_milli),0) cpu, COALESCE(SUM(memory_mi),0) memory, COALESCE(SUM(storage_gi),0) storage, COALESCE(SUM(instance_count),0) instances" ). Where ( "business_line_id = ? AND target_id = ? AND status = ? AND expires_at > ?" , task . BusinessLineID , task . TargetID , "reserved" , time . Now ()). Scan ( & reserved ). Error ; err != nil {
2026-07-23 12:04:37 +08:00
return err
}
if used . CPU + reserved . CPU + payload . CPUMilli > quota . CPUMilli || used . Memory + reserved . Memory + payload . MemoryMi > quota . MemoryMi || used . Storage + reserved . Storage + payload . StorageGi > quota . StorageGi || used . Instances + reserved . Instances + 1 > quota . InstanceLimit {
return s . failInTransaction ( tx , & task , model . TaskValidationFailed , "resource quota is insufficient" )
}
meta := parseTargetMetadata ( target . Metadata )
if len ( meta . Hosts ) == 0 {
return s . failInTransaction ( tx , & task , model . TaskValidationFailed , "deployment target has no candidate hosts" )
}
var occupied [] string
occupiedExclude := [] string { model . TaskExecutionFailed , model . TaskValidationFailed , model . TaskCanceled }
if err := tx . Model ( & model . DeliveryTask {}). Where ( "target_id = ? AND target_host <> ? AND status NOT IN ?" , task . TargetID , "" , occupiedExclude ). Pluck ( "target_host" , & occupied ). Error ; err != nil {
return err
}
host := firstFreeHost ( meta . Hosts , occupied )
if host == nil {
return fmt . Errorf ( "defer: no free host available on target" )
}
reservation := model . ResourceReservation { TaskID : task . ID , BusinessLineID : task . BusinessLineID , TargetID : task . TargetID , CPUMilli : payload . CPUMilli , MemoryMi : payload . MemoryMi , StorageGi : payload . StorageGi , InstanceCount : 1 , Status : "reserved" , ExpiresAt : time . Now (). Add ( time . Duration ( s . cfg . ReservationTTLMinutes ) * time . Minute )}
if err := tx . Create ( & reservation ). Error ; err != nil {
return err
}
if err := tx . Model ( & model . DeliveryTask {}). Where ( "id = ?" , task . ID ). Updates ( map [ string ] any { "target_host" : host . Name , "target_host_ip" : host . IP , "mysql_port" : meta . MySQLPort }). Error ; err != nil {
return err
}
task . TargetHost = host . Name
task . TargetHostIP = host . IP
task . MySQLPort = meta . MySQLPort
if err := s . transitionTx ( tx , & task , model . TaskDispatching , "resources reserved" , "" ); err != nil {
return err
}
dispatchable = true
return nil
})
if err == nil && ! dispatchable {
err = gorm . ErrRecordNotFound
}
return & task , err
}
func ( s * DeliveryService ) failInTransaction ( tx * gorm . DB , task * model . DeliveryTask , status , message string ) error {
if err := s . transitionTx ( tx , task , status , message , message ); err != nil {
return err
}
return nil
}
func ( s * DeliveryService ) transition ( ctx context . Context , task * model . DeliveryTask , status , message , errorMessage string ) error {
return s . db . WithContext ( ctx ). Transaction ( func ( tx * gorm . DB ) error { return s . transitionTx ( tx , task , status , message , errorMessage ) })
}
func ( s * DeliveryService ) transitionTx ( tx * gorm . DB , task * model . DeliveryTask , status , message , errorMessage string ) error {
from := task . Status
updates := map [ string ] any { "status" : status , "error_message" : errorMessage }
now := time . Now ()
if status == model . TaskRunning {
updates [ "started_at" ] = now
}
if status == model . TaskFinished || status == model . TaskExecutionFailed || status == model . TaskValidationFailed || status == model . TaskCanceled || status == model . TaskRegisterFailed {
updates [ "finished_at" ] = now
}
result := tx . Model ( & model . DeliveryTask {}). Where ( "id = ? AND status = ?" , task . ID , from ). Updates ( updates )
if result . Error != nil {
return result . Error
}
if result . RowsAffected != 1 {
return fmt . Errorf ( "task %s changed concurrently" , task . ID )
}
task . Status = status
task . ErrorMessage = errorMessage
return tx . Create ( & model . TaskEvent { TaskID : task . ID , FromState : from , ToState : status , Message : message }). Error
}
func ( s * DeliveryService ) releaseReservation ( tx * gorm . DB , taskID string ) error {
return tx . Model ( & model . ResourceReservation {}). Where ( "task_id = ? AND status = ?" , taskID , "reserved" ). Update ( "status" , "released" ). Error
}
func randomUUID () string {
b := make ([] byte , 16 )
if _ , err := rand . Read ( b ); err != nil {
panic ( err )
}
b [ 6 ] = ( b [ 6 ] & 0x0f ) | 0x40
b [ 8 ] = ( b [ 8 ] & 0x3f ) | 0x80
return fmt . Sprintf ( "%08x-%04x-%04x-%04x-%012x" , b [ 0 : 4 ], b [ 4 : 6 ], b [ 6 : 8 ], b [ 8 : 10 ], b [ 10 : 16 ])
}
func mysqlReady ( ctx context . Context , address string ) error {
dialer := net . Dialer { Timeout : 5 * time . Second }
conn , err := dialer . DialContext ( ctx , "tcp" , address )
if err != nil {
return err
}
return conn . Close ()
}
func ( s * DeliveryService ) DispatchOnce ( ctx context . Context ) error {
task , err := s . claimAndReserve ( ctx )
if err != nil {
if errors . Is ( err , gorm . ErrRecordNotFound ) || strings . HasPrefix ( err . Error (), "defer:" ) {
return nil
}
return err
}
_ , _ , err = s . CreateExecution ( ctx , task . ID , task . PayloadHash , task . IdempotencyKey )
if err != nil {
return s . failTask ( ctx , task , model . TaskExecutionFailed , err . Error ())
}
return nil
}
func ( s * DeliveryService ) CreateExecution ( ctx context . Context , taskID , payloadHash , idempotencyKey string ) ( * model . ExecutionJob , bool , error ) {
s . executionMu . Lock ()
defer s . executionMu . Unlock ()
var existing model . ExecutionJob
if err := s . db . WithContext ( ctx ). Where ( "task_id = ? OR idempotency_key = ?" , taskID , idempotencyKey ). First ( & existing ). Error ; err == nil {
if existing . TaskID != taskID || existing . IdempotencyKey != idempotencyKey {
return nil , false , fmt . Errorf ( "idempotency key is already bound to another task" )
}
return & existing , true , nil
} else if ! errors . Is ( err , gorm . ErrRecordNotFound ) {
return nil , false , err
}
var task model . DeliveryTask
if err := s . db . WithContext ( ctx ). First ( & task , "id = ?" , taskID ). Error ; err != nil {
return nil , false , err
}
if task . PayloadHash != payloadHash || task . IdempotencyKey != idempotencyKey {
return nil , false , fmt . Errorf ( "execution request does not match the immutable task payload" )
}
if task . Status != model . TaskDispatching {
return nil , false , fmt . Errorf ( "task in state %q is not ready for execution" , task . Status )
}
var target model . DeploymentTarget
if err := s . db . WithContext ( ctx ). First ( & target , "id = ? AND enabled = ?" , task . TargetID , true ). Error ; err != nil {
return nil , false , err
}
var payload deliveryPayload
if err := json . Unmarshal ([] byte ( task . ImmutablePayload ), & payload ); err != nil {
return nil , false , err
}
meta := parseTargetMetadata ( target . Metadata )
2026-07-23 14:29:32 +08:00
// Persist execution record BEFORE launching AWX to ensure crash recovery.
now := time . Now ()
execution := model . ExecutionJob { TaskID : task . ID , IdempotencyKey : task . IdempotencyKey , ExecutorJobID : "pending" , Status : "launching" , StartedAt : & now }
if err := s . db . WithContext ( ctx ). Create ( & execution ). Error ; err != nil {
return nil , false , err
}
2026-07-23 12:04:37 +08:00
job , err := s . awx . Launch ( ctx , target . AWXTemplateID , AWXLaunchRequest { InventoryID : target . AWXInventoryID , Limit : task . TargetHost , ExtraVars : map [ string ] any {
"task_id" : task . ID , "payload_hash" : task . PayloadHash ,
"target_hosts" : task . TargetHost , "topology" : meta . Topology ,
"instance_name" : payload . InstanceName , "mysql_port" : task . MySQLPort ,
"memory_mb" : payload . MemoryMi , "storage_gb" : payload . StorageGi ,
2026-07-23 14:29:32 +08:00
"mysql_version" : payload . MySQLVersion ,
2026-07-23 12:04:37 +08:00
}})
if err != nil {
2026-07-23 14:29:32 +08:00
_ = s . db . WithContext ( ctx ). Model ( & execution ). Updates ( map [ string ] any { "status" : "failed" , "finished_at" : time . Now ()})
2026-07-23 12:04:37 +08:00
return nil , false , err
}
if err := s . db . WithContext ( ctx ). Transaction ( func ( tx * gorm . DB ) error {
2026-07-23 14:29:32 +08:00
if err := tx . Model ( & execution ). Updates ( map [ string ] any { "executor_job_id" : fmt . Sprint ( job . ID ), "status" : "running" }). Error ; err != nil {
2026-07-23 12:04:37 +08:00
return err
}
return s . transitionTx ( tx , & task , model . TaskRunning , "AWX job started" , "" )
}); err != nil {
return nil , false , err
}
return & execution , false , nil
}
func ( s * DeliveryService ) PollOnce ( ctx context . Context ) error {
var jobs [] model . ExecutionJob
if err := s . db . WithContext ( ctx ). Where ( "status = ?" , "running" ). Find ( & jobs ). Error ; err != nil {
return err
}
for _ , execution := range jobs {
job , err := s . awx . GetJob ( ctx , execution . ExecutorJobID )
if err != nil {
continue
}
switch strings . ToLower ( job . Status ) {
case "pending" , "waiting" , "running" , "new" :
continue
case "successful" :
2026-07-23 14:29:32 +08:00
s . finishExecution ( ctx , & execution , "successful" )
2026-07-23 12:04:37 +08:00
if err := s . completeTask ( ctx , execution . TaskID ); err != nil {
_ = s . failTask ( ctx , & model . DeliveryTask { ID : execution . TaskID , Status : model . TaskRunning }, model . TaskValidationFailed , err . Error ())
}
2026-07-23 14:29:32 +08:00
case "canceled" :
s . finishExecution ( ctx , & execution , "canceled" )
_ = s . failTask ( ctx , & model . DeliveryTask { ID : execution . TaskID , Status : model . TaskRunning }, model . TaskCanceled , "AWX job was canceled" )
2026-07-23 12:04:37 +08:00
default :
2026-07-23 14:29:32 +08:00
s . finishExecution ( ctx , & execution , "failed" )
2026-07-23 12:04:37 +08:00
_ = s . failTask ( ctx , & model . DeliveryTask { ID : execution . TaskID , Status : model . TaskRunning }, model . TaskExecutionFailed , "AWX job finished with status " + job . Status )
}
}
return nil
}
2026-07-23 14:29:32 +08:00
func ( s * DeliveryService ) finishExecution ( ctx context . Context , execution * model . ExecutionJob , status string ) {
now := time . Now ()
_ = s . db . WithContext ( ctx ). Model ( execution ). Updates ( map [ string ] any { "status" : status , "finished_at" : now })
}
2026-07-23 12:04:37 +08:00
func ( s * DeliveryService ) completeTask ( ctx context . Context , taskID string ) error {
var task model . DeliveryTask
if err := s . db . WithContext ( ctx ). First ( & task , "id = ?" , taskID ). Error ; err != nil {
return err
}
var payload deliveryPayload
if err := json . Unmarshal ([] byte ( task . ImmutablePayload ), & payload ); err != nil {
return err
}
addr := fmt . Sprintf ( "%s:%d" , task . TargetHostIP , task . MySQLPort )
if err := mysqlReady ( ctx , addr ); err != nil {
return fmt . Errorf ( "MySQL health check failed: %w" , err )
}
now := time . Now ()
if err := s . db . WithContext ( ctx ). Transaction ( func ( tx * gorm . DB ) error {
if err := s . transitionTx ( tx , & task , model . TaskRegistering , "AWX succeeded and MySQL health check passed" , "" ); err != nil {
return err
}
instance := model . MySQLInstance { TaskID : task . ID , BusinessLineID : task . BusinessLineID , TargetID : task . TargetID , Namespace : payload . Namespace , Name : payload . InstanceName , NodeName : task . TargetHost , Host : task . TargetHostIP , Port : task . MySQLPort , Version : payload . MySQLVersion , Status : "active" }
if err := tx . Create ( & instance ). Error ; err != nil {
return err
}
if err := tx . Create ( & model . ResourceUsage { TaskID : task . ID , InstanceID : instance . ID , BusinessLineID : task . BusinessLineID , TargetID : task . TargetID , CPUMilli : payload . CPUMilli , MemoryMi : payload . MemoryMi , StorageGi : payload . StorageGi , InstanceCount : 1 , Status : "active" }). Error ; err != nil {
return err
}
if err := tx . Model ( & model . ResourceReservation {}). Where ( "task_id = ? AND status = ?" , task . ID , "reserved" ). Update ( "status" , "consumed" ). Error ; err != nil {
return err
}
if err := tx . Model ( & model . ExecutionJob {}). Where ( "task_id = ?" , task . ID ). Updates ( map [ string ] any { "status" : "successful" , "finished_at" : now }). Error ; err != nil {
return err
}
return nil
}); err != nil {
return err
}
if err := s . RegisterCloudDM ( ctx , task . ID ); err != nil {
return s . transition ( ctx , & task , model . TaskRegisterFailed , "CloudDM registration failed" , err . Error ())
}
return s . transition ( ctx , & task , model . TaskFinished , "MySQL delivery completed" , "" )
}
func ( s * DeliveryService ) RegisterCloudDM ( ctx context . Context , taskID string ) error {
if s . cfg . CloudDMRegisterURL == "" {
return nil
}
var instance model . MySQLInstance
if err := s . db . WithContext ( ctx ). Where ( "task_id = ?" , taskID ). First ( & instance ). Error ; err != nil {
return err
}
body := map [ string ] any { "name" : instance . Name , "host" : instance . Host , "port" : instance . Port , "username" : "root" , "database_type" : "mysql" }
raw , _ := json . Marshal ( body )
req , err := http . NewRequestWithContext ( ctx , http . MethodPost , s . cfg . CloudDMRegisterURL , bytes . NewReader ( raw ))
if err != nil {
return err
}
req . Header . Set ( "Content-Type" , "application/json" )
if s . cfg . CloudDMAPIToken != "" {
req . Header . Set ( "Authorization" , "Bearer " + s . cfg . CloudDMAPIToken )
}
resp , err := http . DefaultClient . Do ( req )
if err != nil {
return err
}
defer resp . Body . Close ()
if resp . StatusCode < 200 || resp . StatusCode >= 300 {
return fmt . Errorf ( "CloudDM returned %s" , resp . Status )
}
return nil
}
func ( s * DeliveryService ) failTask ( ctx context . Context , task * model . DeliveryTask , status , message string ) error {
return s . db . WithContext ( ctx ). Transaction ( func ( tx * gorm . DB ) error {
var current model . DeliveryTask
if err := tx . First ( & current , "id = ?" , task . ID ). Error ; err != nil {
return err
}
2026-07-23 14:29:32 +08:00
if current . Status != model . TaskPending && current . Status != model . TaskDispatching && current . Status != model . TaskRunning && current . Status != model . TaskRegistering && current . Status != model . TaskCanceling {
2026-07-23 12:04:37 +08:00
return nil
}
if err := s . transitionTx ( tx , & current , status , message , message ); err != nil {
return err
}
if status == model . TaskExecutionFailed || status == model . TaskValidationFailed || status == model . TaskCanceled {
return s . releaseReservation ( tx , current . ID )
}
return nil
})
}
func ( s * DeliveryService ) Run ( ctx context . Context ) {
interval := time . Duration ( s . cfg . DeliveryPollSeconds ) * time . Second
if interval < time . Second {
interval = time . Second
}
ticker := time . NewTicker ( interval )
defer ticker . Stop ()
for {
select {
case <- ctx . Done ():
return
case <- ticker . C :
_ = s . DispatchOnce ( ctx )
_ = s . PollOnce ( ctx )
}
}
}
