logs/internal/ingest/alert_outbox.go

package ingest

import (
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"strings"
	"time"

	"git.apinb.com/ops/logs/internal/config"
	"git.apinb.com/ops/logs/internal/impl"
	"git.apinb.com/ops/logs/internal/models"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

const (
	outboxStatusPending    = "pending"
	outboxStatusProcessing = "processing"
	outboxStatusRetrying   = "retrying"
	outboxStatusSent       = "sent"
	outboxStatusDead       = "dead"
)

func enqueueAlertWithDB(db *gorm.DB, logEventID uint, body AlertReceiveBody) (uint, error) {
	payload, err := json.Marshal(body)
	if err != nil {
		return 0, err
	}
	return enqueuePayloadWithDB(db, logEventID, string(payload))
}

func enqueueRawEventWithDB(db *gorm.DB, logEventID uint, body AlertReceiveBody, parseStatus string) (uint, error) {
	payload, err := json.Marshal(buildRawEventIngestBody(body, parseStatus))
	if err != nil {
		return 0, err
	}
	return enqueuePayloadWithDB(db, logEventID, string(payload))
}

func enqueuePayloadWithDB(db *gorm.DB, logEventID uint, payloadJSON string) (uint, error) {
	if db == nil {
		return 0, fmt.Errorf("database is not initialized")
	}
	now, err := databaseNow(db)
	if err != nil {
		return 0, err
	}
	row := models.AlertOutbox{
		LogEventID:  logEventID,
		PayloadJSON: payloadJSON,
		Status:      outboxStatusPending,
		RetryCount:  0,
		NextRetryAt: now,
		LastError:   "",
	}
	if err := db.Create(&row).Error; err != nil {
		return 0, err
	}
	return row.ID, nil
}

func StartAlertDispatcher() {
	owner := dispatcherOwner()
	go func() {
		ticker := time.NewTicker(2 * time.Second)
		defer ticker.Stop()
		for range ticker.C {
			if _, err := ProcessAlertOutboxBatch(impl.DBService, 20, owner); err != nil {
				log.Printf("logs: alert outbox dispatch: %v", err)
			}
		}
	}()
}

func dispatcherOwner() string {
	host, _ := os.Hostname()
	return fmt.Sprintf("%s:%d:%d", host, os.Getpid(), time.Now().UnixNano())
}

// ProcessAlertOutboxBatch 原子领取并处理一批任务，可安全用于多实例 worker。
func ProcessAlertOutboxBatch(db *gorm.DB, limit int, owner string) (int, error) {
	if db == nil {
		return 0, fmt.Errorf("database is not initialized")
	}
	if limit <= 0 {
		limit = 20
	}
	owner = strings.TrimSpace(owner)
	if owner == "" {
		return 0, fmt.Errorf("outbox lease owner is required")
	}
	processed := 0
	for processed < limit {
		rows, err := claimAlertOutboxBatch(db, 1, owner)
		if err != nil {
			return processed, err
		}
		if len(rows) == 0 {
			break
		}
		if err := processOneOutbox(db, rows[0]); err != nil {
			return processed, err
		}
		processed++
	}
	return processed, nil
}

func claimAlertOutboxBatch(db *gorm.DB, limit int, owner string) ([]models.AlertOutbox, error) {
	var rows []models.AlertOutbox
	now, err := databaseNow(db)
	if err != nil {
		return nil, err
	}
	leaseUntil := now.Add(30 * time.Second)
	err = db.Transaction(func(tx *gorm.DB) error {
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}).
			Where("(status IN ? AND next_retry_at <= ?) OR (status = ? AND lease_until IS NOT NULL AND lease_until <= ?)", []string{outboxStatusPending, outboxStatusRetrying}, now, outboxStatusProcessing, now).
			Order("id asc").Limit(limit).Find(&rows).Error; err != nil {
			return err
		}
		if len(rows) == 0 {
			return nil
		}
		ids := make([]uint, 0, len(rows))
		for i := range rows {
			ids = append(ids, rows[i].ID)
		}
		if err := tx.Model(&models.AlertOutbox{}).Where("id IN ?", ids).Updates(map[string]interface{}{
			"status":      outboxStatusProcessing,
			"lease_until": leaseUntil,
			"lease_owner": owner,
		}).Error; err != nil {
			return err
		}
		for i := range rows {
			rows[i].Status = outboxStatusProcessing
			rows[i].LeaseUntil = &leaseUntil
			rows[i].LeaseOwner = owner
		}
		return nil
	})
	return rows, err
}

func processOneOutbox(db *gorm.DB, row models.AlertOutbox) error {
	var body AlertReceiveBody
	if err := json.Unmarshal([]byte(row.PayloadJSON), &body); err != nil {
		now, nowErr := databaseNow(db)
		if nowErr != nil {
			return nowErr
		}
		return markOutboxDead(db, row, row.RetryCount, "invalid_payload: "+err.Error(), now)
	}
	forwardErr := forwardOutboxPayload(row.PayloadJSON, body)
	now, err := databaseNow(db)
	if err != nil {
		return err
	}
	if forwardErr != nil {
		if errors.Is(forwardErr, errAlertForwardDisabled) {
			return markOutboxDead(db, row, row.RetryCount, forwardErr.Error(), now)
		}
		return markOutboxRetry(db, row, forwardErr.Error(), now)
	}
	return markOutboxSent(db, row, now)
}

func databaseNow(db *gorm.DB) (time.Time, error) {
	var now time.Time
	if err := db.Raw("SELECT CURRENT_TIMESTAMP").Scan(&now).Error; err != nil {
		return time.Time{}, err
	}
	return now.UTC(), nil
}

func forwardOutboxPayload(payloadJSON string, legacyBody AlertReceiveBody) error {
	var rawEvent RawEventIngestBody
	if err := json.Unmarshal([]byte(payloadJSON), &rawEvent); err == nil && rawEvent.SourceType != "" && len(rawEvent.RawPayload) > 0 {
		return forwardRawEvent(rawEvent)
	}
	return forwardAlert(legacyBody)
}

func forwardRawEvent(body RawEventIngestBody) error {
	cfg := config.Spec.AlertForward
	if cfg == nil || !cfg.Enabled || cfg.BaseURL == "" {
		return errAlertForwardDisabled
	}
	if len(body.RawPayload) == 0 {
		return fmt.Errorf("raw_payload 不能为空")
	}
	raw, err := json.Marshal(body)
	if err != nil {
		return err
	}
	result, err := postAlertPayload(cfg, "/Alert/v1/raw-events/ingest", raw)
	if err != nil {
		return err
	}
	var details struct {
		ID            uint   `json:"id"`
		RawEventID    uint   `json:"raw_event_id"`
		AlertRecordID uint   `json:"alert_record_id"`
		IncidentID    uint   `json:"incident_id"`
		ParseStatus   string `json:"parse_status"`
		Status        string `json:"status"`
		Fingerprint   string `json:"fingerprint"`
	}
	if err := json.Unmarshal(result.Details, &details); err != nil {
		return fmt.Errorf("Alert 原始事件响应 details 无效：%v；请稍后重试", err)
	}
	if body.ParseStatus == "unparsed" {
		if details.ID == 0 || (details.ParseStatus != "unparsed" && details.ParseStatus != "replayed") {
			return fmt.Errorf("Alert 原始事件响应缺少持久化结果，无法确认转发成功；请稍后重试")
		}
		return nil
	}
	rawEventID := details.RawEventID
	if rawEventID == 0 {
		rawEventID = details.ID
	}
	if rawEventID == 0 || details.AlertRecordID == 0 || details.IncidentID == 0 ||
		(details.ID != 0 && details.RawEventID != 0 && details.ID != details.RawEventID) ||
		(details.Status != "firing" && details.Status != "resolved") || !validFingerprint(details.Fingerprint) {
		return fmt.Errorf("Alert 原始事件响应缺少完整处理结果，无法确认转发成功；请稍后重试")
	}
	return nil
}

func markOutboxSent(db *gorm.DB, row models.AlertOutbox, now time.Time) error {
	return db.Transaction(func(tx *gorm.DB) error {
		result := tx.Model(&models.AlertOutbox{}).
			Where("id = ? AND status = ? AND lease_owner = ?", row.ID, outboxStatusProcessing, row.LeaseOwner).
			Updates(map[string]interface{}{"status": outboxStatusSent, "last_error": "", "next_retry_at": now, "lease_until": nil, "lease_owner": ""})
		if result.Error != nil {
			return result.Error
		}
		if result.RowsAffected == 0 {
			return nil
		}
		return tx.Model(&models.LogEvent{}).Where("id = ? AND dispatch_outbox_id = ?", row.LogEventID, row.ID).Updates(map[string]interface{}{
			"alert_sent": true, "dispatch_status": "sent",
		}).Error
	})
}

func markOutboxRetry(db *gorm.DB, row models.AlertOutbox, msg string, now time.Time) error {
	retry := row.RetryCount + 1
	const maxRetry = 5
	if retry > maxRetry {
		return markOutboxDead(db, row, retry, msg, now)
	}
	backoff := time.Duration(retry*retry) * time.Second
	if backoff > 60*time.Second {
		backoff = 60 * time.Second
	}
	return updateClaimedOutbox(db, row, map[string]interface{}{
		"status": outboxStatusRetrying, "retry_count": retry, "next_retry_at": now.Add(backoff),
		"last_error": truncateError(msg, 1024), "lease_until": nil, "lease_owner": "",
	}, "retrying")
}

func markOutboxDead(db *gorm.DB, row models.AlertOutbox, retry int, msg string, now time.Time) error {
	return updateClaimedOutbox(db, row, map[string]interface{}{
		"status": outboxStatusDead, "retry_count": retry, "next_retry_at": now,
		"last_error": truncateError(msg, 1024), "lease_until": nil, "lease_owner": "",
	}, "dead")
}

func updateClaimedOutbox(db *gorm.DB, row models.AlertOutbox, updates map[string]interface{}, eventStatus string) error {
	return db.Transaction(func(tx *gorm.DB) error {
		result := tx.Model(&models.AlertOutbox{}).
			Where("id = ? AND status = ? AND lease_owner = ?", row.ID, outboxStatusProcessing, row.LeaseOwner).
			Updates(updates)
		if result.Error != nil {
			return result.Error
		}
		if result.RowsAffected == 0 {
			return nil
		}
		return tx.Model(&models.LogEvent{}).Where("id = ? AND dispatch_outbox_id = ?", row.LogEventID, row.ID).Update("dispatch_status", eventStatus).Error
	})
}

func truncateError(s string, n int) string {
	s = strings.TrimSpace(s)
	if n <= 0 {
		return ""
	}
	runes := []rune(s)
	if len(runes) <= n {
		return s
	}
	return string(runes[:n])
}