Cyrene_v2/internal/runtime/events/counters.go

package events

// 本文件归属**计数面**（design/26 §4.1 表二 + §4.3；切片 E1）：内存聚合 + 有界
// 缓冲 + 定期 flush + 逐类汇总读取。
//
// 为什么计数不逐帧落行：心跳 / 在线状态是**高频遥测**，逐帧插一行会把库写胖，而
// 问题「上游到底推过没有」只要求「某个平台某类事件推过几次」。按
// `(platform, account_id, platform_type)` upsert 之后，行数上界是「平台数 × 事件
// 类型数」，与帧数无关（§4.1 表二）。
//
// 三条不变量（每条都有用例钉住）：
//
//  1. **有界**：待落盘的键数绝不超过 `Config.MaxPendingKeys`。缓冲满了先整批落盘，
//     绝不无界增长（§4.3 的背压；对齐 AGENTS §10 的压测纪律：有界，不靠堆进程）。
//  2. **尽力而为**：落盘失败只记 WARN / 丢键记 DEBUG，**绝不**让 `Record` 失败 ——
//     入站流水线不该因为计数写不进去而失败或丢帧（§4.3）。
//  3. **首次权威**：同一个键的 tier/category 只认第一次（upsert 不刷写这两列），
//     与 `documents` 的 `ON CONFLICT(name) DO NOTHING` 同口径。

import (
	"context"
	"fmt"
	"sync"
	"time"
)

// counterKey 是计数聚合的键，**逐列对齐**迁移 0021 的唯一索引
// `idx_channel_event_counters_key (platform, account_id, platform_type)`。
// 键里刻意不含 tier/category：它们由「平台原生类型」决定，不该让同一类帧分裂成
// 两行（那会让计数上界失控，也让「推过几次」的答案分叉）。
type counterKey struct {
	platform     string
	accountID    string
	platformType string
}

// counterEntry 是一个键的累计值与归类。
type counterEntry struct {
	tier      Tier
	category  Category
	count     int64
	firstSeen time.Time
	lastSeen  time.Time
}

// counterBuffer 是内存计数聚合。它由互斥量保护（Record 可能来自多条渠道 goroutine）。
//
// 锁的纪律（AGENTS §六）：**锁内绝不做 IO** —— drain 在锁内只做「搬运 + 清空」，
// 真正的落盘在锁外进行；失败时的回填也走一次短暂的加锁。
type counterBuffer struct {
	mu      sync.Mutex
	max     int
	pending map[counterKey]counterEntry
}

// newCounterBuffer 构造聚合缓冲；最大键数 ≤0 时取 DefaultMaxPendingKeys。
func newCounterBuffer(max int) *counterBuffer {
	if max <= 0 {
		max = DefaultMaxPendingKeys
	}
	return &counterBuffer{max: max, pending: make(map[counterKey]counterEntry, max)}
}

// add 把一帧的计数累加进缓冲。返回值报告**这次计数是否被吞掉**：缓冲已达最大键数
// 且这是一个新键时为 false（该键不落内存）。它是「有界」在代码里的落点 ——
// 无论落盘成功与否，`len()` 都绝不超过 `max`。
func (b *counterBuffer) add(key counterKey, tier Tier, category Category, at time.Time) bool {
	b.mu.Lock()
	defer b.mu.Unlock()
	if entry, ok := b.pending[key]; ok {
		entry.count++
		entry.lastSeen = at
		b.pending[key] = entry
		return true
	}
	if len(b.pending) >= b.max {
		return false
	}
	b.pending[key] = counterEntry{tier: tier, category: category, count: 1, firstSeen: at, lastSeen: at}
	return true
}

// len 返回待落盘的键数（≤ max）。
func (b *counterBuffer) len() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return len(b.pending)
}

// drain 取走全部待落盘项并清空缓冲（锁内只搬运，不 IO）。
func (b *counterBuffer) drain() map[counterKey]counterEntry {
	b.mu.Lock()
	defer b.mu.Unlock()
	if len(b.pending) == 0 {
		return nil
	}
	out := b.pending
	b.pending = make(map[counterKey]counterEntry, b.max)
	return out
}

// restore 把一次**落盘失败**的项回填进缓冲，使计数不在瞬时 DB 故障里丢掉。
// 回填同样受 `max` 约束：装不下的项被丢弃并返回丢弃数（调用方记 WARN）——
// 「尽力而为」与「有界」冲突时，**有界**优先（§4.3 的背压是硬约束）。
func (b *counterBuffer) restore(entries map[counterKey]counterEntry) int {
	b.mu.Lock()
	defer b.mu.Unlock()
	dropped := 0
	for key, entry := range entries {
		if existing, ok := b.pending[key]; ok {
			// 落盘失败期间同一键可能又累加了新计数：合并而不是覆盖。
			existing.count += entry.count
			if entry.lastSeen.After(existing.lastSeen) {
				existing.lastSeen = entry.lastSeen
			}
			if entry.firstSeen.Before(existing.firstSeen) {
				existing.firstSeen = entry.firstSeen
			}
			b.pending[key] = existing
			continue
		}
		if len(b.pending) >= b.max {
			dropped++
			continue
		}
		b.pending[key] = entry
	}
	return dropped
}

// recordCounter 把一帧计入内存聚合（design/26 §4.3：T1 与**全部帧**都计数）。
//
// 它**不返回错误**：计数是尽力而为，绝不让入站流水线因计数而失败。缓冲满时先尝试
// 整批落盘；落盘失败记 WARN（DB 出问题时每次 Record 都会试一次，这也让「DB 恢复后
// 自动收敛」不需要额外的重试器）。被吞掉的新键记 DEBUG —— 单个键的丢弃是上述 WARN
// 的后果，不在这里重复刷 WARN（避免 DB 长时间不可用时日志洪泛）。
func (s *SQLStore) recordCounter(ctx context.Context, ev Event) {
	if s.counters.len() >= s.counters.max {
		if err := s.flushCounters(ctx); err != nil {
			s.log.Warn("channel event counter flush failed",
				"err", err, "pending", s.counters.len(), "max_pending", s.counters.max)
		}
	}
	key := counterKey{platform: ev.Platform, accountID: ev.AccountID, platformType: ev.PlatformType}
	if !s.counters.add(key, ev.Tier, ev.Category, ev.CreatedAt) {
		s.log.Debug("channel event counter key dropped: buffer full",
			"platform", ev.Platform, "platform_type", ev.PlatformType)
	}
}

// Flush 把内存里待落盘的计数写进库（组合根的常驻 ticker 与退出前的最后一次调用
// 都走它）。没有待落盘项时是空操作。
func (s *SQLStore) Flush(ctx context.Context) error {
	return s.flushCounters(ctx)
}

// flushCounters 是 Flush 的实现：取走 → 落盘 → 失败回填。
func (s *SQLStore) flushCounters(ctx context.Context) error {
	pending := s.counters.drain()
	if len(pending) == 0 {
		return nil
	}
	if err := s.writeCounters(ctx, pending); err != nil {
		if dropped := s.counters.restore(pending); dropped > 0 {
			s.log.Warn("channel event counter entries dropped after failed flush",
				"dropped", dropped, "max_pending", s.counters.max)
		}
		return fmt.Errorf("events: flush 计数: %w", err)
	}
	s.log.Debug("channel event counters flushed", "keys", len(pending))
	return nil
}

// upsertCounterSQL 是批量落盘的唯一语句：单事务 + 同一条语句反复执行，避免 N 次
// 往返以外的组织开销（design/26 §4.3）。`count = count + excluded.count` 使同键
// 累加；tier/category 只在**插入**时写入，冲突时保持首次的归类（首次权威）。
const upsertCounterSQL = `INSERT INTO channel_event_counters
    (platform, account_id, tier, platform_type, category, count, first_seen_at, last_seen_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(platform, account_id, platform_type) DO UPDATE SET
    count = count + excluded.count,
    last_seen_at = excluded.last_seen_at`

// writeCounters 在一个事务里 upsert 整批计数：要么全落，要么全不落（失败时由
// flushCounters 回填缓冲，交给下一次 flush 重试）。
func (s *SQLStore) writeCounters(ctx context.Context, pending map[counterKey]counterEntry) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("events: 计数落盘开事务: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	for key, entry := range pending {
		if _, err := tx.ExecContext(ctx, upsertCounterSQL,
			key.platform, key.accountID, string(entry.tier), key.platformType, string(entry.category),
			entry.count, sqliteTimeStamp(entry.firstSeen), sqliteTimeStamp(entry.lastSeen)); err != nil {
			return fmt.Errorf("events: 计数落盘(platform=%s type=%s): %w", key.platform, key.platformType, err)
		}
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("events: 计数落盘提交: %w", err)
	}
	return nil
}

// listCountersSQL 的列显式列出（禁 `SELECT *`），排序固定，便于客户端做稳定展示。
const listCountersSQL = `SELECT platform, account_id, tier, platform_type, category, count,
    first_seen_at, last_seen_at
FROM channel_event_counters`

// Counters 返回逐类计数汇总，按 `(platform, account_id, platform_type)` 升序。
// platform 为空表示不过滤（返回全部平台）。
//
// 它读的是**已落盘**的表，不含内存里尚未 flush 的部分 —— 这正是「计数为近似值」
// 的来源，查询面必须如实标注（design/26 §4.3/§4.5）。需要新鲜数据时先 Flush。
func (s *SQLStore) Counters(ctx context.Context, platform string) ([]Counter, error) {
	query := listCountersSQL
	args := make([]any, 0, 1)
	if platform != "" {
		query += " WHERE platform = ?"
		args = append(args, platform)
	}
	query += " ORDER BY platform, account_id, platform_type"

	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("events: 查计数行: %w", err)
	}
	defer func() { _ = rows.Close() }()

	out := make([]Counter, 0, 16)
	for rows.Next() {
		var (
			c              Counter
			tier, category string
			first, last    string
		)
		if err := rows.Scan(&c.Platform, &c.AccountID, &tier, &c.PlatformType, &category,
			&c.Count, &first, &last); err != nil {
			return nil, fmt.Errorf("events: 读计数行: %w", err)
		}
		c.Tier = Tier(tier)
		c.Category = Category(category)
		at, ok := parseSQLiteTime(first)
		if !ok {
			return nil, fmt.Errorf("events: 计数行 %s/%s 的 first_seen_at 不可解析(%q): %w",
				c.Platform, c.PlatformType, first, ErrBadTimestamp)
		}
		c.FirstSeenAt = at
		if ts, ok := parseSQLiteTime(last); ok {
			c.LastSeenAt = ts
		} else {
			c.LastSeenAt = at
		}
		out = append(out, c)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("events: 遍历计数行: %w", err)
	}
	return out, nil
}