package events

// 本文件归属**事件存储核心**（design/26 §4.1/§4.3；切片 E1）：构造、写入路径
// （`Record` 的逐条落行 + 计数归属）与两个只读查询（`Count` / `List`）。
//
// 三条取舍写在这里：
//
//  1. **T2/T3/unknown 同步逐条落行，失败上抛**（§4.3）：这些行有审计价值、丢不起，
//     它们的失败必须让调用方看见（E3 据此记 WARN，但**不**因此丢帧 —— 留存与
//     「进不进上下文」是两个独立动作）。
//  2. **T1 只累加计数**（§4.3）：心跳/在线状态是高频遥测，逐帧写库是纯浪费。
//     两个**例外**逐条落行（见 persistsAsRow）：`unknown`/`decode_error`（§3.7 明令
//     「留痕 + 计数」）与 §3.6 的链路事件（`link`，它回答的正是「链路当时是不是
//     活的」——design/26 §0.2 事故定案所依赖的那份证据）。
//  3. **时间戳口径与既有迁移逐字同形**（`strftime('%Y-%m-%dT%H:%M:%fZ','now')`，
//     UTC 毫秒 + 字面量 Z）：定宽 ⇒ 字典序比较等价于时间序，时间窗筛选因此能直接
//     吃上 `idx_channel_events_created_at`（与 `runtime/model` 的 `llm/usage` 同口径）。

import (
	"context"
	"database/sql"
	"fmt"
	"io"
	"log/slog"
	"strings"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
)

const (
	// DefaultFlushInterval 是 `CHANNEL_EVENT_COUNTER_FLUSH` 的默认值（design/26 §4.3）：
	// 内存计数每 30s 落盘一次。30s 是「心跳不把库写胖」与「crash 后还能答上近期
	// 有没有推」之间的取舍值。
	DefaultFlushInterval = 30 * time.Second

	// DefaultMaxPendingKeys 是计数缓冲的**键数上界**（design/26 §4.3 的背压）：
	// 不同 `(platform, account_id, platform_type)` 键超过它就整批落盘。有了它，
	// 内存占用与「DB 长时间不可用」时长脱钩 —— 绝不无界增长。
	DefaultMaxPendingKeys = 512

	// DefaultListLimit 是 List 的缺省页大小；MaxListLimit 是它的硬上界
	// （超过即钳住，绝不静默返回全表）。
	DefaultListLimit = 100
	MaxListLimit     = 500

	// sqliteTimeLayout 与迁移 0021 的默认值 `strftime('%Y-%m-%dT%H:%M:%fZ','now')`
	// 逐字同形（UTC、毫秒、字面量 Z）。写入与解析共用它，使「我们写进去的」与
	// 「SQLite 默认生成的」可以直接比较（与 `documents` 同一手法）。
	sqliteTimeLayout = "2006-01-02T15:04:05.000"

	// eventsLogSource 是全部日志的 source 标签（AGENTS §10：禁无来源裸日志）。
	eventsLogSource = "runtime/events"
)

// Config 是事件存储的接线参数。时间与开关口径由调用方（组合根）给出：本包**不读
// 环境变量**、不猜数据目录布局（与 `documents.StoreConfig` / `documents.GCConfig`
// 同一理由）。
type Config struct {
	// DB 是活连接（组合根注入既有 SQLite 连接池；测试注入临时库）。为 nil 时
	// NewStore 返回 ErrNilDB —— 计数的 fail-visible 降级（§4.3「store 为 nil ⇒
	// 只记 DEBUG + 内存计数」）由桥接层（E3）负责，不在本包伪造替身。
	DB *sql.DB
	// Log 是基础 logger；nil 时替换为标记来源 runtime/events 的丢弃 logger。
	Log *slog.Logger
	// Now 是可注入时钟（测试用）；nil 时用 time.Now。
	Now func() time.Time
	// FlushInterval 是内存计数的定期落盘间隔（`CHANNEL_EVENT_COUNTER_FLUSH`）。
	// **≤0 取 DefaultFlushInterval**（与 GC 不同：计数 flush 不是破坏性动作，
	// 「配了 0 就永远不落盘」只会悄悄丢掉计数证据，因此这里不做「显式关闭」）。
	FlushInterval time.Duration
	// MaxPendingKeys 是计数缓冲的键数上界；≤0 取 DefaultMaxPendingKeys。
	MaxPendingKeys int
	// GCInterval 是保留期清扫的间隔（`CHANNEL_EVENT_GC_INTERVAL`）。
	// **≤0 = 关闭 GC**（含启动期清扫），默认 DefaultGCInterval。
	GCInterval time.Duration
	// Retention 是 `channel_events` 的保留期（`CHANNEL_EVENT_RETENTION`）。
	// **≤0 = 关闭 GC**，默认 DefaultRetention。0 刻意**不**解释成「立即清扫」：
	// 那会让一次笔误把整张事件表清空 —— 而它正是「上游推过没有」的唯一证据。
	Retention time.Duration
}

// SQLStore 是基于 SQLite 的事件存储实现。它可安全并发使用：连接池本身串行化写
// （`db.Open` 的 SetMaxOpenConns(1)），内存计数缓冲由自身互斥量保护。
type SQLStore struct {
	db            *sql.DB
	log           *slog.Logger
	now           func() time.Time
	counters      *counterBuffer
	flushInterval time.Duration
	gcInterval    time.Duration
	retention     time.Duration
	gcEnabled     bool
}

// NewStore 构造事件存储。唯一的失败是 DB 为 nil（ErrNilDB）；GC 的关闭**不是**
// 构造失败，它是可见的正常状态（Enabled() 为 false + 一条记明原因的 WARN）。
func NewStore(cfg Config) (*SQLStore, error) {
	if cfg.DB == nil {
		return nil, fmt.Errorf("events: new store: %w", ErrNilDB)
	}
	log := sourcedLog(cfg.Log)
	now := cfg.Now
	if now == nil {
		now = time.Now
	}
	flush := cfg.FlushInterval
	if flush <= 0 {
		flush = DefaultFlushInterval
	}
	s := &SQLStore{
		db:            cfg.DB,
		log:           log,
		now:           now,
		counters:      newCounterBuffer(cfg.MaxPendingKeys),
		flushInterval: flush,
		gcInterval:    cfg.GCInterval,
		retention:     cfg.Retention,
	}
	switch {
	case cfg.GCInterval <= 0:
		log.Warn("channel event gc disabled: interval <= 0", "interval", cfg.GCInterval.String())
	case cfg.Retention <= 0:
		log.Warn("channel event gc disabled: retention <= 0", "retention", cfg.Retention.String())
	default:
		s.gcEnabled = true
	}
	log.Info("channel event store ready",
		"flush_interval", s.flushInterval.String(),
		"max_pending_keys", s.counters.max,
		"gc_enabled", s.gcEnabled,
		"gc_interval", s.gcInterval.String(),
		"retention", s.retention.String())
	return s, nil
}

// sourcedLog 把基础 logger 标上 source=runtime/events；nil 时给出丢弃 logger，
// 使调用方永远可以安全记日志（与 `documents` / `history` 同一手法）。
func sourcedLog(l *slog.Logger) *slog.Logger {
	if l == nil {
		l = slog.New(slog.NewTextHandler(io.Discard, nil))
	}
	return logger.WithSource(l, eventsLogSource)
}

// FlushInterval 返回生效的计数落盘间隔（组合根的常驻 ticker 用它）。
func (s *SQLStore) FlushInterval() time.Duration {
	if s == nil {
		return 0
	}
	return s.flushInterval
}

// PendingKeys 返回内存里待落盘的计数的键数。它有上界（MaxPendingKeys），因此可以
// 直接作为健康指标暴露给面板：只涨不落 = DB 出了问题。
func (s *SQLStore) PendingKeys() int {
	if s == nil {
		return 0
	}
	return s.counters.len()
}

// Record 记录一帧事件（design/26 §4.3 的唯一写入口）。
//
// 判定顺序刻意固定：**先校验 → 再归一化 → 再按档落行 → 最后计数**。这样即便逐条
// 落行失败（返回错误），调用方也能明确知道「这一帧没能留下可查痕迹」，而不会拿到
// 一个「成功了但其实什么都没落」的假象。
//
// 计数侧的错误**不上抛**：计数是尽力而为（§4.3），绝不因为计数落盘失败而让入站
// 流水线失败或丢帧。缓冲满了先整批落盘，失败则记 WARN。
func (s *SQLStore) Record(ctx context.Context, ev Event) error {
	if strings.TrimSpace(ev.Platform) == "" {
		return fmt.Errorf("events: record: %w", ErrNoPlatform)
	}
	ev = s.normalize(ev)
	if persistsAsRow(ev.Tier, ev.Category) {
		if err := s.insertEvent(ctx, ev); err != nil {
			return fmt.Errorf("events: record 落行: %w", err)
		}
	}
	s.recordCounter(ctx, ev)
	return nil
}

// normalize 补齐一条待写事件：时钟零值填当前时刻；tier 取值非法时归一为 TierT1
// 并记 WARN —— 这是 fail-safe（§1.5）：看不懂的档**只**降级成「只计数」，绝不让
// 一个字段写错演变成「这一帧什么都没有留下」。
func (s *SQLStore) normalize(ev Event) Event {
	switch ev.Tier {
	case TierT1, TierT2, TierT3:
	default:
		s.log.Warn("channel event tier invalid, falling back to t1",
			"platform", ev.Platform, "platform_type", ev.PlatformType, "tier", string(ev.Tier))
		ev.Tier = TierT1
	}
	if ev.CreatedAt.IsZero() {
		ev.CreatedAt = s.now()
	}
	return ev
}

// persistsAsRow 判定一帧是否要逐条落 `channel_events`（design/26 §4.1/§4.3/§3.6/§3.7）：
//
//   - T2 / T3：逐条（有审计价值，丢不起）；
//   - unknown / decode_error：逐条（§3.7「落库 + 计数 + 脱敏结构留痕」，且
//     **绝不进上下文**——兜底不等于放行）；
//   - link：逐条（§3.6 明写「落 channel_events（category=link）」；链路事件是
//     §0.2 事故里「当时链路活没活」的唯一答案，即便它是 T1 也不能只留一个计数）。
//
// 它是纯函数（按 AGENTS §10 不引入包级可变状态）。
func persistsAsRow(tier Tier, category Category) bool {
	switch tier {
	case TierT2, TierT3:
		return true
	}
	switch category {
	case CategoryUnknown, CategoryDecodeError, CategoryLink:
		return true
	default:
		return false
	}
}

// insertEventSQL 是逐条落行的唯一语句：列名显式列出（禁 `SELECT *` 的同一条纪律
// 在写侧的表现），值全部参数化。updated_at 与 created_at 同值：事件行不可变，
// 两列同形使「读回时的 updated_at」不会比事实更晚。
const insertEventSQL = `INSERT INTO channel_events
    (platform, account_id, tier, category, platform_type, channel_id, channel_type,
     group_id, actor_id, actor_role, target_id, self_related, message_id, summary,
     created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`

// insertEvent 写一行 `channel_events`。
func (s *SQLStore) insertEvent(ctx context.Context, ev Event) error {
	stamp := sqliteTimeStamp(ev.CreatedAt)
	if _, err := s.db.ExecContext(ctx, insertEventSQL,
		ev.Platform, ev.AccountID, string(ev.Tier), string(ev.Category), ev.PlatformType,
		ev.ChannelID, ev.ChannelType, ev.GroupID, ev.ActorID, ev.ActorRole, ev.TargetID,
		boolToInt(ev.SelfRelated), ev.MessageID, ev.Summary, stamp, stamp); err != nil {
		return fmt.Errorf("events: 落事件行(platform=%s type=%s): %w", ev.Platform, ev.PlatformType, err)
	}
	return nil
}

// Count 统计符合条件的事件行数。
func (s *SQLStore) Count(ctx context.Context, f Filter) (int64, error) {
	where, args := f.where()
	var n int64
	if err := s.db.QueryRowContext(ctx, "SELECT count(*) FROM channel_events"+where, args...).Scan(&n); err != nil {
		return 0, fmt.Errorf("events: 统计事件行: %w", err)
	}
	return n, nil
}

// listEventsSQL 的列显式列出（与迁移 0021 同序），便于与 scanEvent 对齐核对。
const listEventsSQL = `SELECT id, platform, account_id, tier, category, platform_type,
    channel_id, channel_type, group_id, actor_id, actor_role, target_id, self_related,
    message_id, summary, created_at, updated_at
FROM channel_events`

// List 按 `created_at DESC, id DESC`（倒序，design/26 §4.5）返回符合条件的事件行。
// id 参与排序是为了让同一毫秒内的多行有确定顺序，分页因此不会重复或漏行。
func (s *SQLStore) List(ctx context.Context, f Filter) ([]Event, error) {
	where, args := f.where()
	limit, offset := f.pageLimit(), f.pageOffset()
	query := listEventsSQL + where + " ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?"
	args = append(args, limit, offset)

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

	out := make([]Event, 0, limit)
	for rows.Next() {
		ev, err := scanEvent(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, ev)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("events: 遍历事件行: %w", err)
	}
	return out, nil
}

// scanEvent 读出一行 `channel_events`。时间戳读不出来时返回错误而不是零值时间：
// 一个读不懂的时间戳意味着这一行被谁改坏了，「如实报错」比「静默给一个 1970 年」
// 更接近 design/26 的诚实口径（与 `documents` GC 的「读不懂的时间绝不当成已超期」
// 同一个保守方向）。
func scanEvent(rows *sql.Rows) (Event, error) {
	var (
		ev               Event
		tier, category   string
		selfRelated      int64
		created, updated string
	)
	if err := rows.Scan(&ev.ID, &ev.Platform, &ev.AccountID, &tier, &category, &ev.PlatformType,
		&ev.ChannelID, &ev.ChannelType, &ev.GroupID, &ev.ActorID, &ev.ActorRole, &ev.TargetID,
		&selfRelated, &ev.MessageID, &ev.Summary, &created, &updated); err != nil {
		return Event{}, fmt.Errorf("events: 读事件行: %w", err)
	}
	ev.Tier = Tier(tier)
	ev.Category = Category(category)
	ev.SelfRelated = selfRelated != 0
	at, ok := parseSQLiteTime(created)
	if !ok {
		return Event{}, fmt.Errorf("events: 事件行 %d 的 created_at 不可解析(%q): %w", ev.ID, created, ErrBadTimestamp)
	}
	ev.CreatedAt = at
	if ts, ok := parseSQLiteTime(updated); ok {
		ev.UpdatedAt = ts
	} else {
		ev.UpdatedAt = at
	}
	return ev, nil
}

// where 把 Filter 编译成参数化的 WHERE 子句（列名是常量，值全部走占位符）。
// 有意的取舍：**只拼列名与运算符，绝不拼值**（SQL 全参数化，AGENTS §10）。
func (f Filter) where() (string, []any) {
	clauses := make([]string, 0, 6)
	args := make([]any, 0, 6)
	if platform := strings.TrimSpace(f.Platform); platform != "" {
		clauses = append(clauses, "platform = ?")
		args = append(args, platform)
	}
	if f.Tier != "" {
		clauses = append(clauses, "tier = ?")
		args = append(args, string(f.Tier))
	}
	if f.Category != "" {
		clauses = append(clauses, "category = ?")
		args = append(args, string(f.Category))
	}
	if channelID := strings.TrimSpace(f.ChannelID); channelID != "" {
		clauses = append(clauses, "channel_id = ?")
		args = append(args, channelID)
	}
	if !f.Since.IsZero() {
		clauses = append(clauses, "created_at >= ?")
		args = append(args, sqliteTimeStamp(f.Since))
	}
	if !f.Until.IsZero() {
		clauses = append(clauses, "created_at <= ?")
		args = append(args, sqliteTimeStamp(f.Until))
	}
	if len(clauses) == 0 {
		return "", nil
	}
	return " WHERE " + strings.Join(clauses, " AND "), args
}

// pageLimit 把 Limit 归一：≤0 取缺省、超过硬上界即钳住。
func (f Filter) pageLimit() int {
	switch {
	case f.Limit <= 0:
		return DefaultListLimit
	case f.Limit > MaxListLimit:
		return MaxListLimit
	default:
		return f.Limit
	}
}

// pageOffset 把 Offset 归一到 ≥0。
func (f Filter) pageOffset() int {
	if f.Offset < 0 {
		return 0
	}
	return f.Offset
}

// sqliteTimeStamp 把时刻格式化成库里的时间戳口径（UTC 毫秒 + 字面量 Z）。
func sqliteTimeStamp(t time.Time) string {
	return t.UTC().Format(sqliteTimeLayout) + "Z"
}

// parseSQLiteTime 解析库里的时间戳列。它先试本包写出的口径，再退回 RFC3339
// （覆盖带偏移或非毫秒精度的历史值）；两者都不成立时返回 false，调用方据此保守
// 处理（绝不把读不懂的时间当成一个真实时刻）。
func parseSQLiteTime(raw string) (time.Time, bool) {
	trimmed := strings.TrimSpace(raw)
	if trimmed == "" {
		return time.Time{}, false
	}
	if t, err := time.Parse(sqliteTimeLayout+"Z", trimmed); err == nil {
		return t.UTC(), true
	}
	if t, err := time.Parse(time.RFC3339Nano, trimmed); err == nil {
		return t.UTC(), true
	}
	return time.Time{}, false
}

// boolToInt 把布尔列写成 SQLite 的 0/1（列本身是 INTEGER NOT NULL DEFAULT 0）。
func boolToInt(v bool) int {
	if v {
		return 1
	}
	return 0
}