oci-portal/internal/service/logevent.go

package service

import (
	"context"
	"crypto/rand"
	"crypto/subtle"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

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

	"oci-portal/internal/model"
	"oci-portal/internal/oci"
)

// 日志回传的保留策略与后台节奏(方案A §2.4/§3)。
const (
	logEventRetention   = 90 * 24 * time.Hour // 过期删除阈值
	logEventMaxRows     = 20000               // 总量兜底上限(Payload 大,低于系统日志的 5 万)
	logEventCleanupTick = 24 * time.Hour      // 周期清理间隔
	logEventParseTick   = 30 * time.Second    // 解析器轮询间隔
	logEventParseBatch  = 200                 // 单轮解析行数上限
	logEventConfirmWait = 10 * time.Second    // 订阅确认 GET 超时
)

// 日志回传查询分页默认值与上限。
const (
	logEventDefaultPageSize = 20
	logEventMaxPageSize     = 100
)

// logWebhookSecretPrefix 是每租户回传 secret 的 Setting 键前缀,后接 cfgID。
const logWebhookSecretPrefix = "log_webhook_secret:"

// LogEventService 承接 OCI 日志回传:secret 管理、事件入库、异步解析与清理,
// 以及 P1 引导创建(SetRelayDeps 注入)与 P2 告警联动(SetNotifier 注入)。
type LogEventService struct {
	db      *gorm.DB
	wg      sync.WaitGroup
	confirm func(ctx context.Context, url string) error // 订阅确认 GET,测试可注入

	configs     *OciConfigService // 凭据来源(P1)
	relayClient oci.Client        // 云端链路操作(P1)
	publicURL   string            // 面板公网基址,拼接回调 endpoint(P1)
	notifier    *Notifier         // 关键事件推送(P2)
	settings    *SettingService   // log_event 事件开关(P2)

	relayPollTick    time.Duration // 订阅确认轮询间隔,零值用默认
	relayPollTimeout time.Duration // 订阅确认轮询上限,零值用默认
}

// NewLogEventService 组装依赖;调用 StartParser / StartCleanup 后台协程后生效。
func NewLogEventService(db *gorm.DB) *LogEventService {
	s := &LogEventService{db: db}
	s.confirm = s.confirmSubscription
	return s
}

// secretKey 拼接指定配置的 Setting 键。
func secretKey(cfgID uint) string {
	return fmt.Sprintf("%s%d", logWebhookSecretPrefix, cfgID)
}

// LogWebhookInfo 是回传回调地址视图;Path 供前端以公网域名拼接完整 URL。
type LogWebhookInfo struct {
	Path      string    `json:"path"`
	Secret    string    `json:"secret"`
	CreatedAt time.Time `json:"createdAt"`
}

// EnsureSecret 为配置生成(或幂等返回)回传 secret。
func (s *LogEventService) EnsureSecret(ctx context.Context, cfgID uint) (LogWebhookInfo, error) {
	if err := s.requireConfig(ctx, cfgID); err != nil {
		return LogWebhookInfo{}, err
	}
	if info, ok, err := s.SecretInfo(ctx, cfgID); err != nil || ok {
		return info, err
	}
	buf := make([]byte, 32)
	if _, err := rand.Read(buf); err != nil {
		return LogWebhookInfo{}, fmt.Errorf("generate webhook secret: %w", err)
	}
	secret := hex.EncodeToString(buf)
	st := model.Setting{Key: secretKey(cfgID), Value: secret, UpdatedAt: time.Now()}
	if err := s.db.WithContext(ctx).Save(&st).Error; err != nil {
		return LogWebhookInfo{}, fmt.Errorf("save webhook secret: %w", err)
	}
	return webhookInfo(secret, st.UpdatedAt), nil
}

// requireConfig 校验配置存在,不存在透传 gorm.ErrRecordNotFound(api 层映射 404)。
func (s *LogEventService) requireConfig(ctx context.Context, cfgID uint) error {
	var cfg model.OciConfig
	if err := s.db.WithContext(ctx).Select("id").First(&cfg, cfgID).Error; err != nil {
		return fmt.Errorf("find oci config %d: %w", cfgID, err)
	}
	return nil
}

// SecretInfo 查询配置是否已生成 secret;未生成时 ok 为 false。
func (s *LogEventService) SecretInfo(ctx context.Context, cfgID uint) (LogWebhookInfo, bool, error) {
	var st model.Setting
	err := s.db.WithContext(ctx).First(&st, "key = ?", secretKey(cfgID)).Error
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return LogWebhookInfo{}, false, nil
		}
		return LogWebhookInfo{}, false, fmt.Errorf("get webhook secret: %w", err)
	}
	if st.Value == "" {
		return LogWebhookInfo{}, false, nil
	}
	return webhookInfo(st.Value, st.UpdatedAt), true, nil
}

// webhookInfo 由 secret 组装回调地址视图。
func webhookInfo(secret string, at time.Time) LogWebhookInfo {
	return LogWebhookInfo{
		Path:      "/api/v1/webhooks/oci-logs/" + secret,
		Secret:    secret,
		CreatedAt: at,
	}
}

// RevokeSecret 撤销配置的回传 secret;之后旧回调地址一律 404。
func (s *LogEventService) RevokeSecret(ctx context.Context, cfgID uint) error {
	err := s.db.WithContext(ctx).Delete(&model.Setting{}, "key = ?", secretKey(cfgID)).Error
	if err != nil {
		return fmt.Errorf("revoke webhook secret: %w", err)
	}
	return nil
}

// ResolveSecret 由 secret 反查归属配置;比对使用常数时间比较,防时序侧信道。
func (s *LogEventService) ResolveSecret(ctx context.Context, secret string) (uint, bool) {
	if secret == "" {
		return 0, false
	}
	var rows []model.Setting
	err := s.db.WithContext(ctx).
		Where("key LIKE ?", logWebhookSecretPrefix+"%").Find(&rows).Error
	if err != nil {
		log.Printf("resolve webhook secret: %v", err)
		return 0, false
	}
	for _, row := range rows {
		if subtle.ConstantTimeCompare([]byte(row.Value), []byte(secret)) == 1 {
			id, err := strconv.ParseUint(strings.TrimPrefix(row.Key, logWebhookSecretPrefix), 10, 64)
			if err != nil {
				return 0, false
			}
			return uint(id), true
		}
	}
	return 0, false
}

// Ingest 落一条回传事件;MessageID 唯一索引冲突即静默忽略(at-least-once 幂等)。
func (s *LogEventService) Ingest(ctx context.Context, cfgID uint, messageID string, payload []byte, truncated bool) error {
	event := model.LogEvent{
		OciConfigID: cfgID,
		MessageID:   messageID,
		Payload:     string(payload),
		Truncated:   truncated,
		ReceivedAt:  time.Now(),
	}
	err := s.db.WithContext(ctx).
		Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "message_id"}}, DoNothing: true}).
		Create(&event).Error
	if err != nil {
		return fmt.Errorf("ingest log event: %w", err)
	}
	return nil
}

// LogEventQuery 是回传事件查询参数;CfgID 为 0 表示全部租户。
type LogEventQuery struct {
	CfgID    uint
	Page     int
	PageSize int
}

// normalize 补分页默认值并钳制上限。
func (q LogEventQuery) normalize() LogEventQuery {
	if q.Page < 1 {
		q.Page = 1
	}
	if q.PageSize < 1 {
		q.PageSize = logEventDefaultPageSize
	}
	if q.PageSize > logEventMaxPageSize {
		q.PageSize = logEventMaxPageSize
	}
	return q
}

// List 按接收顺序倒序分页查询回传事件。
func (s *LogEventService) List(ctx context.Context, q LogEventQuery) ([]model.LogEvent, int64, error) {
	q = q.normalize()
	tx := s.db.WithContext(ctx).Model(&model.LogEvent{})
	if q.CfgID > 0 {
		tx = tx.Where("oci_config_id = ?", q.CfgID)
	}
	var total int64
	if err := tx.Count(&total).Error; err != nil {
		return nil, 0, fmt.Errorf("count log events: %w", err)
	}
	items := make([]model.LogEvent, 0, q.PageSize)
	err := tx.Order("id DESC").Offset((q.Page - 1) * q.PageSize).Limit(q.PageSize).Find(&items).Error
	if err != nil {
		return nil, 0, fmt.Errorf("list log events: %w", err)
	}
	return items, total, nil
}

// Wait 等待在途后台协程(确认/解析/清理)退出,供进程收尾与测试同步。
func (s *LogEventService) Wait() {
	s.wg.Wait()
}

// ConfirmAsync 异步 GET 订阅确认链接完成激活;失败仅记系统日志可人工重发,
// URL 白名单校验由 api 层完成,此处不再信任外部输入以外的假设。
func (s *LogEventService) ConfirmAsync(confirmURL string) {
	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		ctx, cancel := context.WithTimeout(context.Background(), logEventConfirmWait)
		defer cancel()
		if err := s.confirm(ctx, confirmURL); err != nil {
			log.Printf("confirm ons subscription: %v", err)
		}
	}()
}

// confirmSubscription 执行确认 GET;重定向只允许留在 Oracle 域内,响应体丢弃。
func (s *LogEventService) confirmSubscription(ctx context.Context, confirmURL string) error {
	client := &http.Client{
		CheckRedirect: func(req *http.Request, _ []*http.Request) error {
			if !IsOracleHost(req.URL.Host) {
				return fmt.Errorf("redirect outside oracle domain")
			}
			return nil
		},
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, confirmURL, nil)
	if err != nil {
		return fmt.Errorf("build confirm request: %w", err)
	}
	resp, err := client.Do(req)
	if err != nil {
		return fmt.Errorf("confirm request: %w", sanitizeURLError(err))
	}
	defer resp.Body.Close()
	if resp.StatusCode >= http.StatusBadRequest {
		return fmt.Errorf("confirm request: status %d", resp.StatusCode)
	}
	return nil
}

// IsOracleHost 判定 host 是否属于 Oracle 云域(订阅确认/验签证书源白名单)。
func IsOracleHost(host string) bool {
	if h, _, err := net.SplitHostPort(host); err == nil {
		host = h
	}
	host = strings.ToLower(host)
	return host == "oraclecloud.com" || strings.HasSuffix(host, ".oraclecloud.com")
}

// StartParser 启动解析协程:周期消费未解析事件,回填类型/来源/事件时间。
func (s *LogEventService) StartParser(ctx context.Context) {
	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		ticker := time.NewTicker(logEventParseTick)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				s.parseOnce(ctx)
			}
		}
	}()
}

// parseOnce 消费一批未解析事件;失败只记日志,不中断周期调度。
func (s *LogEventService) parseOnce(ctx context.Context) {
	var events []model.LogEvent
	err := s.db.WithContext(ctx).
		Where("processed = ?", false).Order("id").Limit(logEventParseBatch).
		Find(&events).Error
	if err != nil {
		log.Printf("log event parse load: %v", err)
		return
	}
	for i := range events {
		e := &events[i]
		parsed := parseLogEvent([]byte(e.Payload))
		e.EventType, e.Source, e.SourceIP, e.EventTime =
			parsed.EventType, parsed.Source, parsed.SourceIP, parsed.EventTime
		e.Processed = true
		if err := s.db.WithContext(ctx).Save(e).Error; err != nil {
			log.Printf("log event parse save %d: %v", e.ID, err)
			return
		}
		s.notifyCritical(ctx, e, parsed)
	}
}

// onsEnvelope 覆盖 ONS 消息与 CloudEvents 审计事件的常见字段;
// 真实格式以联调实测为准,提不出字段时只置 Processed 不回填。
type onsEnvelope struct {
	EventType    string          `json:"eventType"`
	Type         string          `json:"type"`
	Source       string          `json:"source"`
	EventTime    string          `json:"eventTime"`
	ResourceName string          `json:"resourceName"`
	Data         json.RawMessage `json:"data"`
	Identity     *onsIdentity    `json:"identity"`
}

// onsIdentity 是 Audit 事件 data.identity 中与展示相关的字段。
type onsIdentity struct {
	IPAddress string `json:"ipAddress"`
}

// parsedEvent 是从消息原文提取的展示字段集;ResourceName 仅供 P2 推送文案,不落库。
type parsedEvent struct {
	EventType    string
	Source       string
	SourceIP     string
	ResourceName string
	EventTime    *time.Time
}

// parseLogEvent 从消息原文宽松提取事件字段;非 JSON 返回零值。
func parseLogEvent(payload []byte) parsedEvent {
	var env onsEnvelope
	if err := json.Unmarshal(payload, &env); err != nil {
		var batch []onsEnvelope
		if err := json.Unmarshal(payload, &batch); err != nil || len(batch) == 0 {
			return parsedEvent{}
		}
		env = batch[0]
	}
	if len(env.Data) > 0 {
		var inner onsEnvelope
		if err := json.Unmarshal(env.Data, &inner); err == nil {
			return envelopeFields(mergeEnvelope(env, inner))
		}
	}
	return envelopeFields(env)
}

// mergeEnvelope 外层缺失字段时以 data 内层补齐(Audit 的 identity 在内层)。
func mergeEnvelope(outer, inner onsEnvelope) onsEnvelope {
	if outer.EventType == "" {
		outer.EventType = inner.EventType
	}
	if outer.Type == "" {
		outer.Type = inner.Type
	}
	if outer.Source == "" {
		outer.Source = inner.Source
	}
	if outer.EventTime == "" {
		outer.EventTime = inner.EventTime
	}
	if outer.ResourceName == "" {
		outer.ResourceName = inner.ResourceName
	}
	if outer.Identity == nil {
		outer.Identity = inner.Identity
	}
	return outer
}

// envelopeFields 收敛字段别名并解析事件时间。
func envelopeFields(env onsEnvelope) parsedEvent {
	eventType := env.EventType
	if eventType == "" {
		eventType = env.Type
	}
	var eventTime *time.Time
	if env.EventTime != "" {
		if ts, err := time.Parse(time.RFC3339, env.EventTime); err == nil {
			eventTime = &ts
		}
	}
	ip := ""
	if env.Identity != nil {
		ip = env.Identity.IPAddress
	}
	return parsedEvent{
		EventType:    clip(eventType, 128),
		Source:       clip(env.Source, 64),
		SourceIP:     clip(ip, 64),
		ResourceName: clip(env.ResourceName, 128),
		EventTime:    eventTime,
	}
}

// clip 按模型列宽截断解析出的字段。
func clip(s string, max int) string {
	if len(s) > max {
		return s[:max]
	}
	return s
}

// StartCleanup 启动周期清理:启动即清一次,之后每 24h 一次,随 ctx 取消退出。
func (s *LogEventService) StartCleanup(ctx context.Context) {
	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		s.cleanupOnce(ctx)
		ticker := time.NewTicker(logEventCleanupTick)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				s.cleanupOnce(ctx)
			}
		}
	}()
}

// cleanupOnce 执行一轮清理,失败只记日志、不中断周期调度。
func (s *LogEventService) cleanupOnce(ctx context.Context) {
	if err := s.cleanup(ctx, logEventRetention, logEventMaxRows); err != nil {
		log.Printf("log event cleanup: %v", err)
	}
}

// cleanup 先删过期记录,再对超量部分删最旧;阈值参数化便于测试。
func (s *LogEventService) cleanup(ctx context.Context, retention time.Duration, maxRows int) error {
	cutoff := time.Now().Add(-retention)
	if err := s.db.WithContext(ctx).Where("received_at < ?", cutoff).Delete(&model.LogEvent{}).Error; err != nil {
		return fmt.Errorf("delete expired log events: %w", err)
	}
	var total int64
	if err := s.db.WithContext(ctx).Model(&model.LogEvent{}).Count(&total).Error; err != nil {
		return fmt.Errorf("count log events: %w", err)
	}
	overflow := int(total) - maxRows
	if overflow <= 0 {
		return nil
	}
	oldest := s.db.Model(&model.LogEvent{}).Select("id").Order("id ASC").Limit(overflow)
	if err := s.db.WithContext(ctx).Where("id IN (?)", oldest).Delete(&model.LogEvent{}).Error; err != nil {
		return fmt.Errorf("trim log events over cap: %w", err)
	}
	return nil
}