package service

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/1024XEngineer/xinfra/server/internal/config"
	"github.com/1024XEngineer/xinfra/server/internal/model"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

const (
	postgresqlPortPoolStart = 15432
	postgresqlPortPoolEnd   = 15999
	postgresqlServiceType   = "postgresql"
)

var supportedPostgreSQLVersions = map[string]bool{"15": true, "16": true}
var supportedPostgreSQLTopologies = map[string]bool{"standalone": true, "primary_replica": true}
var postgresqlNamePattern = regexp.MustCompile(`^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?$`)
var errPostgreSQLTaskHandled = errors.New("postgresql task validation completed without dispatch")

type PostgreSQLDeliveryInput struct {
	BusinessLineID    uint64   `json:"business_line_id" binding:"required"`
	TargetID          uint64   `json:"target_id" binding:"required"`
	Namespace         string   `json:"namespace" binding:"required"`
	ClusterName       string   `json:"cluster_name" binding:"required"`
	VersionMajor      string   `json:"version_major"`
	PostgreSQLVersion string   `json:"postgresql_version,omitempty"`
	Topology          string   `json:"topology"`
	ReplicaCount      int      `json:"replica_count"`
	TargetHosts       []string `json:"target_hosts,omitempty"`
	CPUMilli          int64    `json:"cpu_milli" binding:"required"`
	MemoryMi          int64    `json:"memory_mi" binding:"required"`
	StorageGi         int64    `json:"storage_gi" binding:"required"`
	DataRoot          string   `json:"data_root"`
	MaxConnections    int      `json:"max_connections"`
}

type postgresqlDeliveryPayload struct {
	PostgreSQLDeliveryInput
	TargetType string `json:"target_type"`
}

func validatePostgreSQLDeliveryInput(input PostgreSQLDeliveryInput, dataDisks []string) error {
	_ = dataDisks
	if input.VersionMajor == "" {
		input.VersionMajor = input.PostgreSQLVersion
	}
	if input.PostgreSQLVersion != "" && input.VersionMajor != "" && input.PostgreSQLVersion != input.VersionMajor {
		return fmt.Errorf("version_major and postgresql_version must match")
	}
	if len(input.Namespace) > 63 || !dnsLabelPattern.MatchString(input.Namespace) {
		return fmt.Errorf("namespace must be a valid Kubernetes DNS label")
	}
	if len(input.ClusterName) > 63 || !postgresqlNamePattern.MatchString(input.ClusterName) {
		return fmt.Errorf("cluster_name must be a valid DNS label")
	}
	if !supportedPostgreSQLVersions[input.VersionMajor] {
		return fmt.Errorf("unsupported version_major %q, supported: 15, 16", input.VersionMajor)
	}
	topology := input.Topology
	if topology == "" {
		topology = "standalone"
	}
	if !supportedPostgreSQLTopologies[topology] {
		return fmt.Errorf("unsupported topology %q, supported: standalone, primary_replica", topology)
	}
	if topology == "standalone" && input.ReplicaCount != 0 {
		return fmt.Errorf("standalone topology cannot have replicas")
	}
	if topology == "primary_replica" && (input.ReplicaCount < 1 || input.ReplicaCount > 7) {
		return fmt.Errorf("replica_count must be between 1 and 7 for primary_replica")
	}
	if input.CPUMilli < 100 || input.CPUMilli > 64000 || input.MemoryMi < 2048 || input.MemoryMi > 65536 || input.StorageGi < 20 || input.StorageGi > 2000 {
		return fmt.Errorf("requested resources are outside the supported range (memory: 2048-65536 MiB, storage: 20-2000 GiB)")
	}
	nodes := 1
	if topology == "primary_replica" {
		nodes += input.ReplicaCount
	}
	if len(input.TargetHosts) > 0 && len(input.TargetHosts) != nodes {
		return fmt.Errorf("target_hosts must contain exactly %d distinct hosts", nodes)
	}
	seen := map[string]bool{}
	for _, host := range input.TargetHosts {
		if len(host) > 253 || !hostNamePattern.MatchString(host) || seen[host] {
			return fmt.Errorf("target_hosts must contain unique valid inventory host names")
		}
		seen[host] = true
	}
	if input.DataRoot != "" {
		if input.DataRoot != "/data/postgresql" {
			return fmt.Errorf("data_root is fixed to /data/postgresql in the first release")
		}
	}
	if input.MaxConnections < 0 || input.MaxConnections > 10000 {
		return fmt.Errorf("max_connections must be between 0 and 10000")
	}
	return nil
}

func allocatePostgreSQLPort(requested int, used []int) (int, error) {
	taken := make(map[int]bool, len(used))
	for _, port := range used {
		taken[port] = true
	}
	if requested != 0 {
		if requested < postgresqlPortPoolStart || requested > postgresqlPortPoolEnd {
			return 0, fmt.Errorf("postgresql_port must be within %d-%d", postgresqlPortPoolStart, postgresqlPortPoolEnd)
		}
		if taken[requested] {
			return 0, fmt.Errorf("postgresql port %d is already allocated on the target host", requested)
		}
		return requested, nil
	}
	for port := postgresqlPortPoolStart; port <= postgresqlPortPoolEnd; port++ {
		if !taken[port] {
			return port, nil
		}
	}
	return 0, fmt.Errorf("postgresql port pool %d-%d is exhausted on the target host", postgresqlPortPoolStart, postgresqlPortPoolEnd)
}

type postgresqlPortProbe func(context.Context, string, int) bool

func allocateReachablePostgreSQLPort(ctx context.Context, host string, used []int, inUse postgresqlPortProbe) (int, error) {
	occupied := append([]int(nil), used...)
	for {
		port, err := allocatePostgreSQLPort(0, occupied)
		if err != nil {
			return 0, err
		}
		if !inUse(ctx, host, port) {
			return port, nil
		}
		occupied = append(occupied, port)
	}
}

func postgresqlPortInUse(ctx context.Context, host string, port int) bool {
	probeCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
	defer cancel()
	conn, err := (&net.Dialer{}).DialContext(probeCtx, "tcp", net.JoinHostPort(host, strconv.Itoa(port)))
	if err != nil {
		return false
	}
	_ = conn.Close()
	return true
}

type PostgreSQLDeliveryService struct {
	db     *gorm.DB
	cfg    config.Config
	awx    *AWXClient
	common *DeliveryService
}

func NewPostgreSQLDeliveryService(cfg config.Config, db *gorm.DB, common *DeliveryService) *PostgreSQLDeliveryService {
	return &PostgreSQLDeliveryService{db: db, cfg: cfg, awx: NewAWXClient(cfg.AWXBaseURL, cfg.AWXToken, cfg.AWXUsername, cfg.AWXPassword), common: common}
}

func (s *PostgreSQLDeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin bool, idempotencyKey string, input PostgreSQLDeliveryInput) (*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 := validatePostgreSQLDeliveryInput(input, s.cfg.DeliveryDataDisks); err != nil {
		return nil, false, err
	}
	if input.VersionMajor == "" {
		input.VersionMajor = input.PostgreSQLVersion
	}
	var existing model.DeliveryTask
	if err := s.db.WithContext(ctx).Where("idempotency_key = ?", idempotencyKey).First(&existing).Error; err == nil {
		if existing.RequestedBy != userID || existing.ServiceType != postgresqlServiceType {
			return nil, false, fmt.Errorf("idempotency key is already in use")
		}
		return &existing, true, nil
	} else if !errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, false, err
	}
	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.Topology == "" {
		input.Topology = "standalone"
	}
	if input.DataRoot == "" {
		input.DataRoot = "/data/postgresql"
	}
	target, err := getPostgreSQLTarget(ctx, s.awx, input.TargetID)
	if err != nil {
		return nil, false, err
	}
	payload := postgresqlDeliveryPayload{PostgreSQLDeliveryInput: 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, Component: postgresqlServiceType, TargetType: target.TargetType, ServiceType: postgresqlServiceType, TargetID: input.TargetID, Namespace: input.Namespace, InstanceName: input.ClusterName, 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: "postgresql delivery task created"}).Error
	return &task, false, nil
}

func getPostgreSQLTarget(ctx context.Context, awx *AWXClient, templateID uint64) (DeliveryTarget, error) {
	template, err := awx.GetJobTemplate(ctx, templateID)
	if err != nil {
		return DeliveryTarget{}, fmt.Errorf("deployment target is unavailable: %w", err)
	}
	templateText := strings.ToLower(template.Name + " " + template.Description)
	if !strings.Contains(templateText, "postgresql") || strings.Contains(templateText, "rollback") {
		return DeliveryTarget{}, fmt.Errorf("AWX job template %d is not a PostgreSQL target", templateID)
	}
	hosts, err := awx.ListInventoryHosts(ctx, template.Inventory)
	if err != nil {
		return DeliveryTarget{}, err
	}
	meta := targetMetadata{Topology: "standalone"}
	for _, host := range hosts {
		if host.Enabled {
			meta.Hosts = append(meta.Hosts, targetHost{Name: host.Name, IP: AWXHostIP(host)})
		}
	}
	raw, err := json.Marshal(meta)
	if err != nil {
		return DeliveryTarget{}, err
	}
	return DeliveryTarget{ID: template.ID, Name: template.Name, TargetType: "host_pool", AWXInventoryID: template.Inventory, AWXTemplateID: template.ID, Enabled: true, Metadata: string(raw)}, nil
}

func selectPostgreSQLHosts(hosts []targetHost, requested []string, count int) ([]targetHost, error) {
	if len(requested) == 0 {
		if len(hosts) < count {
			return nil, fmt.Errorf("deployment target has %d hosts, but %d PostgreSQL instances are required", len(hosts), count)
		}
		return append([]targetHost(nil), hosts[:count]...), nil
	}
	byName := map[string]targetHost{}
	for _, host := range hosts {
		byName[host.Name] = host
	}
	selected := make([]targetHost, 0, len(requested))
	for _, name := range requested {
		host, ok := byName[name]
		if !ok {
			return nil, fmt.Errorf("target host %q is not in the candidate host pool", name)
		}
		selected = append(selected, host)
	}
	return selected, nil
}

func (s *PostgreSQLDeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryTask, error) {
	var task model.DeliveryTask
	var payload postgresqlDeliveryPayload
	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}).Where("status = ? AND service_type = ?", model.TaskPending, postgresqlServiceType).Order("created_at ASC").First(&task).Error; err != nil {
			return err
		}
		if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
			return s.common.transitionTx(tx, &task, model.TaskValidationFailed, "stored PostgreSQL deployment payload is invalid", "stored PostgreSQL deployment payload is invalid")
		}
		activeStates := []string{model.TaskValidating, model.TaskDispatching, model.TaskRunning, model.TaskCanceling}
		checks := []struct {
			query string
			args  []any
			limit int
		}{
			{"status IN ?", []any{activeStates}, s.cfg.DeliveryGlobalLimit},
			{"status IN ? AND target_id = ?", []any{activeStates, task.TargetID}, s.cfg.DeliveryTargetLimit},
			{"status IN ? AND business_line_id = ?", []any{activeStates, task.BusinessLineID}, s.cfg.DeliveryBusinessLimit},
		}
		for _, check := range checks {
			if check.limit <= 0 {
				continue
			}
			var count int64
			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: delivery concurrency limit reached")
			}
		}
		target, err := getPostgreSQLTarget(ctx, s.awx, task.TargetID)
		if err != nil {
			return s.common.transitionTx(tx, &task, model.TaskValidationFailed, err.Error(), err.Error())
		}
		meta := parseTargetMetadata(target.Metadata)
		nodes := 1
		if payload.Topology == "primary_replica" {
			nodes += payload.ReplicaCount
		}
		selected, err := selectPostgreSQLHosts(meta.Hosts, payload.TargetHosts, nodes)
		if err != nil {
			return s.common.transitionTx(tx, &task, model.TaskValidationFailed, err.Error(), err.Error())
		}
		quotaOK, err := checkPostgreSQLResourceQuota(tx, task.BusinessLineID, task.TargetID, payload, int64(nodes))
		if err != nil {
			return err
		}
		if !quotaOK {
			return s.common.transitionTx(tx, &task, model.TaskValidationFailed, "resource quota is insufficient", "resource quota is insufficient")
		}
		var existingCluster model.PostgreSQLCluster
		if err := tx.Where("name = ?", payload.ClusterName).First(&existingCluster).Error; err == nil {
			return s.common.transitionTx(tx, &task, model.TaskValidationFailed, "PostgreSQL cluster name is already in use", "PostgreSQL cluster name is already in use")
		} else if !errors.Is(err, gorm.ErrRecordNotFound) {
			return err
		}
		cluster := model.PostgreSQLCluster{TaskID: task.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, Name: payload.ClusterName, VersionMajor: payload.VersionMajor, Topology: payload.Topology, ReplicationMode: "async", FailoverMode: "manual", Status: "provisioning", BackupStatus: "not_configured", MonitoringStatus: "not_configured"}
		if err := tx.Create(&cluster).Error; err != nil {
			return err
		}
		primaryPort := 0
		for index, host := range selected {
			var usedPorts []int
			if err := tx.Model(&model.PostgreSQLInstance{}).Where("hostname = ? AND status IN ?", host.Name, []string{"provisioning", "active", "quarantined"}).Pluck("port", &usedPorts).Error; err != nil {
				return err
			}
			port, err := allocateReachablePostgreSQLPort(ctx, host.IP, usedPorts, postgresqlPortInUse)
			if err != nil {
				if cleanupErr := tx.Where("task_id = ?", task.ID).Delete(&model.PostgreSQLInstance{}).Error; cleanupErr != nil {
					return cleanupErr
				}
				if cleanupErr := tx.Delete(&cluster).Error; cleanupErr != nil {
					return cleanupErr
				}
				return s.common.transitionTx(tx, &task, model.TaskValidationFailed, err.Error(), err.Error())
			}
			role := "replica"
			instanceID := fmt.Sprintf("%s-replica-%d", payload.ClusterName, index)
			upstream := payload.ClusterName + "-primary"
			slot := fmt.Sprintf("xinfra_%s_replica_%d", strings.ReplaceAll(payload.ClusterName, "-", "_"), index)
			if index == 0 {
				role = "primary"
				instanceID = payload.ClusterName + "-primary"
				upstream = ""
				slot = ""
				primaryPort = port
			}
			if payload.Topology == "standalone" {
				role = "standalone"
				instanceID = payload.ClusterName
				upstream = ""
				slot = ""
			}
			root := strings.TrimRight(payload.DataRoot, "/") + "/" + instanceID
			instance := model.PostgreSQLInstance{TaskID: task.ID, ClusterID: cluster.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, InstanceID: instanceID, Hostname: host.Name, HostIP: host.IP, Port: port, DataDir: root + "/data", ConfigDir: root + "/conf", LogDir: root + "/log", SystemdUnit: "postgresql-xinfra@" + instanceID + ".service", VersionMajor: payload.VersionMajor, Role: role, HAComponentRole: "database", UpstreamInstanceID: upstream, ReplicationSlotName: slot, Status: "provisioning", BackupStatus: "not_configured", MonitoringStatus: "not_configured"}
			if err := tx.Create(&instance).Error; err != nil {
				return err
			}
			if index == 0 {
				cluster.PrimaryInstanceID = instance.ID
			}
		}
		if err := tx.Save(&cluster).Error; err != nil {
			return err
		}
		reservation := model.ResourceReservation{TaskID: task.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, CPUMilli: payload.CPUMilli * int64(nodes), MemoryMi: payload.MemoryMi * int64(nodes), StorageGi: payload.StorageGi * int64(nodes), InstanceCount: int64(nodes), Status: "reserved", ExpiresAt: time.Now().Add(time.Duration(s.cfg.ReservationTTLMinutes) * time.Minute)}
		if err := tx.Create(&reservation).Error; err != nil {
			return err
		}
		hostNames := make([]string, 0, len(selected))
		for _, host := range selected {
			hostNames = append(hostNames, host.Name)
		}
		task.TargetHost = strings.Join(hostNames, ",")
		task.TargetHostIP = selected[0].IP
		task.PostgreSQLPort = primaryPort
		if err := tx.Model(&model.DeliveryTask{}).Where("id = ?", task.ID).Updates(map[string]any{"target_host": task.TargetHost, "target_host_ip": task.TargetHostIP, "postgresql_port": primaryPort}).Error; err != nil {
			return err
		}
		return s.common.transitionTx(tx, &task, model.TaskDispatching, "PostgreSQL resources, ports and directories reserved", "")
	})
	if err == nil && task.Status != model.TaskDispatching {
		return nil, errPostgreSQLTaskHandled
	}
	return &task, err
}

func checkPostgreSQLResourceQuota(tx *gorm.DB, businessLineID, targetID uint64, payload postgresqlDeliveryPayload, nodes int64) (bool, error) {
	var quota model.ResourceQuota
	if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("business_line_id = ? AND target_id = ?", businessLineID, targetID).First(&quota).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return true, nil
		}
		return false, err
	}
	type totals struct{ CPU, Memory, Storage, Instances int64 }
	var mysqlUsed, postgresUsed, 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 = ?", businessLineID, targetID, "active").Scan(&mysqlUsed).Error; err != nil {
		return false, err
	}
	if err := tx.Model(&model.PostgreSQLResourceUsage{}).Select("COALESCE(SUM(cpu_milli),0) cpu, COALESCE(SUM(memory_mi),0) memory, COALESCE(SUM(storage_gi),0) storage, COUNT(*) instances").Where("business_line_id = ? AND target_id = ? AND status = ?", businessLineID, targetID, "active").Scan(&postgresUsed).Error; err != nil {
		return false, err
	}
	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 > ?", businessLineID, targetID, "reserved", time.Now()).Scan(&reserved).Error; err != nil {
		return false, err
	}
	requestedCPU := payload.CPUMilli * nodes
	requestedMemory := payload.MemoryMi * nodes
	requestedStorage := payload.StorageGi * nodes
	return mysqlUsed.CPU+postgresUsed.CPU+reserved.CPU+requestedCPU <= quota.CPUMilli &&
		mysqlUsed.Memory+postgresUsed.Memory+reserved.Memory+requestedMemory <= quota.MemoryMi &&
		mysqlUsed.Storage+postgresUsed.Storage+reserved.Storage+requestedStorage <= quota.StorageGi &&
		mysqlUsed.Instances+postgresUsed.Instances+reserved.Instances+nodes <= quota.InstanceLimit, nil
}

func (s *PostgreSQLDeliveryService) CreateExecution(ctx context.Context, task *model.DeliveryTask) error {
	var payload postgresqlDeliveryPayload
	if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
		return err
	}
	var instances []model.PostgreSQLInstance
	if err := s.db.WithContext(ctx).Where("task_id = ?", task.ID).Order("id ASC").Find(&instances).Error; err != nil {
		return err
	}
	instanceVars := make(map[string]any, len(instances))
	for _, instance := range instances {
		instanceVars[instance.Hostname] = map[string]any{"instance_id": instance.InstanceID, "port": instance.Port, "data_dir": instance.DataDir, "config_dir": instance.ConfigDir, "log_dir": instance.LogDir, "role": instance.Role, "replication_slot": instance.ReplicationSlotName}
	}
	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 err
	}
	target, err := getPostgreSQLTarget(ctx, s.awx, task.TargetID)
	if err != nil {
		return err
	}
	extraVars := map[string]any{"task_id": task.ID, "payload_hash": task.PayloadHash, "target_hosts": task.TargetHost, "cluster_name": payload.ClusterName, "postgresql_version": payload.VersionMajor, "postgresql_port": task.PostgreSQLPort, "topology": payload.Topology, "replica_count": payload.ReplicaCount, "postgresql_instances": instanceVars, "memory_mb": payload.MemoryMi, "storage_gb": payload.StorageGi}
	if payload.MaxConnections > 0 {
		extraVars["max_connections"] = payload.MaxConnections
	}
	job, err := s.awx.Launch(ctx, target.AWXTemplateID, AWXLaunchRequest{InventoryID: target.AWXInventoryID, Limit: task.TargetHost, ExtraVars: extraVars})
	if err != nil {
		_ = s.db.WithContext(ctx).Model(&execution).Updates(map[string]any{"status": "failed", "finished_at": time.Now()})
		return err
	}
	return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := tx.Model(&execution).Updates(map[string]any{"executor_job_id": fmt.Sprint(job.ID), "status": "running"}).Error; err != nil {
			return err
		}
		return s.common.transitionTx(tx, task, model.TaskRunning, "PostgreSQL AWX job started", "")
	})
}

func (s *PostgreSQLDeliveryService) DispatchOnce(ctx context.Context) error {
	task, err := s.claimAndReserve(ctx)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) || errors.Is(err, errPostgreSQLTaskHandled) || strings.HasPrefix(err.Error(), "defer:") {
			return nil
		}
		return err
	}
	if err := s.CreateExecution(ctx, task); err != nil {
		return s.beginRollback(ctx, task.ID, "PostgreSQL deployment could not be started: "+err.Error())
	}
	return nil
}

func (s *PostgreSQLDeliveryService) PollOnce(ctx context.Context) error {
	var jobs []model.ExecutionJob
	if err := s.db.WithContext(ctx).Joins("JOIN delivery_tasks ON delivery_tasks.id = execution_jobs.task_id").Where("execution_jobs.status = ? AND delivery_tasks.service_type = ?", "running", postgresqlServiceType).Find(&jobs).Error; err != nil {
		return err
	}
	for _, execution := range jobs {
		job, err := s.awx.GetJob(ctx, execution.ExecutorJobID)
		if err != nil {
			// A temporary AWX failure does not mean the deployment failed. Retry on
			// the next scheduler tick while the execution remains running.
			continue
		}
		switch strings.ToLower(job.Status) {
		case "pending", "waiting", "running", "new":
			continue
		case "canceled":
			_ = s.beginRollback(ctx, execution.TaskID, "PostgreSQL AWX job was canceled")
		case "successful":
			if err := s.complete(ctx, execution.TaskID); err != nil {
				_ = s.beginRollback(ctx, execution.TaskID, err.Error())
			}
		default:
			_ = s.beginRollback(ctx, execution.TaskID, "PostgreSQL AWX job finished with status "+job.Status)
		}
	}
	return nil
}

func (s *PostgreSQLDeliveryService) complete(ctx context.Context, taskID string) error {
	var task model.DeliveryTask
	if err := s.db.WithContext(ctx).First(&task, "id = ? AND service_type = ?", taskID, postgresqlServiceType).Error; err != nil {
		return err
	}
	var payload postgresqlDeliveryPayload
	if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
		return err
	}
	var instances []model.PostgreSQLInstance
	if err := s.db.WithContext(ctx).Where("task_id = ?", taskID).Order("id ASC").Find(&instances).Error; err != nil {
		return err
	}
	if len(instances) == 0 {
		return fmt.Errorf("PostgreSQL task has no planned instances")
	}
	for _, instance := range instances {
		if err := postgresReady(ctx, net.JoinHostPort(instance.HostIP, fmt.Sprint(instance.Port))); err != nil {
			return fmt.Errorf("postgresql health check failed for %s: %w", instance.InstanceID, err)
		}
	}
	now := time.Now()
	if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := s.common.transitionTx(tx, &task, model.TaskRegistering, "PostgreSQL health checks passed", ""); err != nil {
			return err
		}
		for _, instance := range instances {
			if err := tx.Model(&instance).Updates(map[string]any{"status": "active", "version_full": instance.VersionMajor}).Error; err != nil {
				return err
			}
			if err := tx.Create(&model.PostgreSQLResourceUsage{TaskID: task.ID, InstanceID: instance.ID, ClusterID: instance.ClusterID, BusinessLineID: instance.BusinessLineID, TargetID: instance.TargetID, CPUMilli: payload.CPUMilli, MemoryMi: payload.MemoryMi, StorageGi: payload.StorageGi, Port: instance.Port, DataDir: instance.DataDir, 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.PostgreSQLCluster{}).Where("task_id = ?", task.ID).Update("status", "active").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 {
		message := fmt.Sprintf("PostgreSQL delivered; CloudDM registration failed: %v", err)
		if transitionErr := s.common.transition(ctx, &task, model.TaskRegisterFailed, "PostgreSQL delivered; CloudDM registration failed and can be retried", message); transitionErr != nil {
			return fmt.Errorf("%s; cannot record register_failed: %w", message, transitionErr)
		}
		return nil
	}
	message := "PostgreSQL delivery completed; CloudDM registration was skipped because no PostgreSQL endpoint is configured"
	if strings.TrimSpace(s.cfg.CloudDMPostgreSQLRegisterURL) != "" {
		message = "PostgreSQL delivery completed, registered and recorded in the resource ledger"
	}
	return s.common.transition(ctx, &task, model.TaskFinished, message, "")
}

func postgresReady(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 *PostgreSQLDeliveryService) 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)
			_ = s.PollRollbackOnce(ctx)
		}
	}
}