iotlcm/internal/notify/service.go

package notify

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"sort"
	"time"

	"git.5iot.com/iotlcm-public/iotlcm-bot/internal/archive"
	"git.5iot.com/iotlcm-public/iotlcm-bot/internal/render"
	"git.5iot.com/iotlcm-public/iotlcm-bot/internal/state"
)

type Service struct {
	archiveDir, environment string
	stateDir                string
	pollInterval            time.Duration
	store                   state.Store
	client                  Sender
	logger                  *slog.Logger
	// warned 记录已经报过错的归档，避免坏文件每轮扫描都刷一条日志。
	// 键为「路径 + mtime + 大小」：文件一旦被重写就会重新告警。
	warned map[string]struct{}
	// transient 记录临时错误的重报节奏与累计次数。
	// 临时错误必须持续可见（一次网络抖动不能被吞掉），但 bot 被移出群这类
	// 长期故障不能每秒刷一条，否则日志被淹、真告警失联。
	transient map[string]*transientWarn
}

// Sender 是发送端抽象：*feishu.Client 满足它，测试可注入替身。
type Sender interface {
	Send(ctx context.Context, card []byte) error
	SendText(ctx context.Context, text string) error
}

func New(archiveDir, environment string, pollInterval time.Duration, store state.Store, client Sender, logger *slog.Logger) Service {
	if logger == nil {
		logger = DefaultLogger()
	}
	return Service{archiveDir: archiveDir, environment: environment, stateDir: store.Dir(), pollInterval: pollInterval, store: store, client: client, logger: logger, warned: map[string]struct{}{}, transient: map[string]*transientWarn{}}
}
func (s Service) Render(path string) (string, error) {
	a, err := archive.Load(path)
	if err != nil {
		return "", err
	}
	return render.Text(a, s.environment), nil
}

func (s Service) SendFile(ctx context.Context, path string, force bool) error {
	if err := s.store.Prepare(); err != nil {
		return err
	}
	return s.process(ctx, path, force)
}

func (s Service) runFiles() ([]string, error) {
	files, err := filepath.Glob(filepath.Join(s.archiveDir, "run-*.json"))
	if err != nil {
		return nil, err
	}
	sort.Strings(files)
	return files, nil
}

// skipKey 生成归档文件的「内容版本」标识，用于去重告警。
// 读不到元数据时退化为路径本身，宁可少报警也不要刷屏。
func skipKey(path string) string {
	info, err := os.Stat(path)
	if err != nil {
		return path
	}
	return fmt.Sprintf("%s|%d|%d", path, info.ModTime().UnixNano(), info.Size())
}

// archiveError 区分「这个归档永远不会成功」与「这次发送碰上了临时故障」。
// 前者（格式不对、旧版本、rounds 为空）是常态，只该报一次；
// 后者（飞书超时、鉴权失败、写状态失败）必须每轮重试并持续可见，
// 否则一次网络抖动就会被去重逻辑吞掉，通知静默丢失。
type archiveError struct {
	err       error
	permanent bool
}

func (e *archiveError) Error() string { return e.err.Error() }
func (e *archiveError) Unwrap() error { return e.err }

// permanentErr 标记「重试也不会好」的错误。
func permanentErr(err error) error { return &archiveError{err: err, permanent: true} }

// isPermanent 判断错误是否属于「归档本身无效」。
func isPermanent(err error) bool {
	var ae *archiveError
	return errors.As(err, &ae) && ae.permanent
}

// warnOnce 对同一个归档文件的重报做节流。
//
// 永久错误（归档格式坏、旧版本）只记一次：重试也不会好。
// 临时错误（飞书超时、bot 不在群）按退避节奏重报：
//   - 前 transientWarnFastPhase 次每次都报，保证瞬时抖动可见；
//   - 之后按指数退避（1min→2min→4min…上限 transientWarnMaxInterval）；
//   - 超过 transientWarnMaxCount 次后只记一条汇总并停止重报，避免日志风暴。
//
// 历史上 230002（bot 不在群）持续 10 小时刷了 18,266 条 ERROR，占日志 99.9%，
// 把真正的告警完全淹没——这正是本函数要防的情况。
func (s Service) warnOnce(path string, err error) {
	if !isPermanent(err) {
		s.warnTransient(path, err)
		return
	}
	key := skipKey(path)
	if s.warned != nil {
		if _, seen := s.warned[key]; seen {
			return
		}
		s.warned[key] = struct{}{}
	}
	s.logger.Error("skip archive", "path", path, "error", err)
}

const (
	// transientWarnFastPhase 是「每次都报」的次数，覆盖瞬时抖动。
	transientWarnFastPhase = 5
	// transientWarnMaxInterval 是退避重报的间隔上限。
	transientWarnMaxInterval = 10 * time.Minute
	// transientWarnMaxCount 是同一归档同一错误的最大重报次数，超过后静默。
	transientWarnMaxCount = 30
)

// transientWarn 记录一个归档的临时错误重报状态。
type transientWarn struct {
	count      int
	lastLog    time.Time
	suppressed bool
}

// warnTransient 按退避节奏记录临时错误。
func (s Service) warnTransient(path string, err error) {
	key := skipKey(path)
	now := time.Now()

	if s.transient == nil {
		s.transient = map[string]*transientWarn{}
	}
	st, ok := s.transient[key]
	if !ok {
		st = &transientWarn{}
		s.transient[key] = st
	}

	if st.suppressed {
		return
	}

	// 前 fastPhase 次每次都报。
	if st.count < transientWarnFastPhase {
		st.count++
		st.lastLog = now
		s.logger.Error("process archive failed", "path", path, "error", err, "attempt", st.count)
		return
	}

	// 之后按指数退避。
	interval := time.Minute << uint(st.count-transientWarnFastPhase)
	if interval > transientWarnMaxInterval {
		interval = transientWarnMaxInterval
	}
	if now.Sub(st.lastLog) < interval {
		return
	}

	st.count++
	st.lastLog = now
	if st.count > transientWarnMaxCount {
		st.suppressed = true
		s.logger.Error("process archive failed, suppressing further identical alerts",
			"path", path, "error", err, "attempts", st.count-1,
			"hint", "this is likely a persistent configuration problem (e.g. bot removed from chat), fix the cause to resume notifications")
		return
	}
	s.logger.Error("process archive failed", "path", path, "error", err, "attempt", st.count)
}

// Scan 扫描归档目录并发送尚未通知过的 Run。
//
// 单个归档出错（旧格式、写了一半、权限不足）只跳过并记日志：
// 过去这里直接 return，一个坏文件会挡住后面所有归档，通知会静默丢失。
// 坏文件是常态（例如 manual 部分同步的 Run 天然没有 rounds），所以只记一次。
func (s Service) Scan(ctx context.Context) error {
	if err := s.store.Prepare(); err != nil {
		return err
	}
	files, err := s.runFiles()
	if err != nil {
		return err
	}
	for _, path := range files {
		if err := s.process(ctx, path, false); err != nil {
			s.warnOnce(path, err)
		}
	}
	return nil
}
func (s Service) Watch(ctx context.Context) error {
	if s.pollInterval <= 0 {
		s.pollInterval = 10 * time.Second
	}
	var emptySince time.Time
	for {
		h := s.Health()
		if h.Degraded {
			// 目录已被宿主删除等硬故障：告警节流状态落盘，避免退出重启后
			// 把同一次故障反复推送成飞书骚扰。
			now := time.Now()
			st := loadAlertState(s.stateDir)
			if st.shouldAlert(h.Reason, now, DegradedAlertInterval) {
				s.logger.Error("archive directory is not usable", "path", s.archiveDir, "reason", h.Reason, "entries", h.Entries)
				s.alertMount(ctx, h.Reason, st.ExitCount+1)
				st.LastDegradedAlert = now
				st.LastDegradedReason = h.Reason
				if err := saveAlertState(s.stateDir, st); err != nil {
					s.logger.Error("save alert state failed", "error", err)
				}
			}
			if ExitOnDegraded {
				// 退出并交给 restart policy 重新拉起：重启会重新解析 bind mount，
				// 从而重新挂载到宿主当前的目录（详见 ExitOnDegraded 注释）。
				st.ExitCount++
				if err := saveAlertState(s.stateDir, st); err != nil {
					s.logger.Error("save alert state failed", "error", err)
				}
				if st.ExitCount > DegradedMaxExits {
					// 已经反复重启仍未恢复（例如宿主把目录删了且不再重建）：
					// 继续退出只会变成重启风暴，改为留在原地持续观测+告警。
					s.logger.Error("archive mount still unusable after repeated restarts, staying up to keep reporting",
						"path", s.archiveDir, "reason", h.Reason, "exitCount", st.ExitCount)
				} else {
					// 等一会再退：restart 策略是立即拉起的，立刻退出会造成几秒一次的崩溃循环。
					s.logger.Error("exit to rebind archive mount", "path", s.archiveDir, "reason", h.Reason, "exitCount", st.ExitCount, "delay", DegradedExitDelay)
					timer := time.NewTimer(DegradedExitDelay)
					select {
					case <-ctx.Done():
						timer.Stop()
						return ctx.Err()
					case <-timer.C:
					}
					return fmt.Errorf("archive directory unusable, restarting to rebind mount: %s", h.Reason)
				}
			}
		} else if h.Entries == 0 {
			if emptySince.IsZero() {
				emptySince = time.Now()
			}
		} else {
			emptySince = time.Time{}
		}

		if err := s.Scan(ctx); err != nil {
			s.logger.Error("scan failed", "error", err)
		}

		timer := time.NewTimer(s.Interval(emptySince))
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case <-timer.C:
		}
	}
}

// alert 在飞书客户端可用时发一条文本告警；失败只记日志，不影响主循环。
func (s Service) alert(ctx context.Context, title, reason string) {
	if s.client == nil {
		return
	}
	text := fmt.Sprintf("[iotlcm-bot][%s] %s: %s", s.environment, title, reason)
	if err := s.client.SendText(ctx, text); err != nil {
		s.logger.Error("send alert failed", "error", err)
	}
}

// alertMount 发送「归档目录不可用」告警卡片。
//
// 用卡片而非纯文本：text 消息在飞书里是灰底等宽块，长错误串会把气泡撑得很宽；
// 卡片带红色 header 与分区字段，一眼能看出是故障，且与同步结果卡片风格一致。
func (s Service) alertMount(ctx context.Context, reason string, exitCount int) {
	if s.client == nil {
		return
	}
	summary := "正在重启以重新挂载"
	if exitCount > DegradedMaxExits {
		summary = "反复重启仍未恢复，已改为持续监控"
	}
	card, err := render.AlertCard(render.Alert{
		Environment: s.environment,
		Title:       "归档目录不可用",
		Summary:     summary,
		Reason:      render.FriendlyReason(reason),
		ExitCount:   exitCount,
		Path:        s.archiveDir,
	})
	if err != nil {
		s.logger.Error("render alert card failed", "error", err)
		return
	}
	if err := s.client.Send(ctx, card); err != nil {
		s.logger.Error("send alert failed", "error", err)
	}
}
func (s Service) process(ctx context.Context, path string, force bool) error {
	a, err := archive.Load(path)
	if err != nil {
		// 归档本身无效（格式/版本/rounds 为空）：重试也不会好。
		return permanentErr(err)
	}
	if !force {
		sent, err := s.store.Has(a.Run.RunID)
		if err != nil {
			return err
		}
		if sent {
			return nil
		}
	}
	card, err := render.Card(a, s.environment)
	if err != nil {
		return permanentErr(err)
	}
	if s.client == nil {
		return fmt.Errorf("feishu client is not configured")
	}
	if err := s.client.Send(ctx, card); err != nil {
		return err
	}
	if err := s.store.Mark(a.Run.RunID); err != nil {
		return err
	}
	s.logger.Info("run notification sent", "runId", a.Run.RunID, "path", path)
	return nil
}

func DefaultLogger() *slog.Logger {
	return slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
}