stock-analysis/internal/storage/sqlite/store.go

// Package sqlite 实现 storage.KlineStore 接口，使用 SQLite 数据库。
// 表结构与 DML 使用通用 SQL 语法（ON CONFLICT 为 SQLite 3.24+ 与 PostgreSQL 共有的标准写法），
// 迁移到 PostgreSQL 时仅需调整连接与参数占位符。
package sqlite

import (
	"context"
	"database/sql"
	"fmt"
	"time"

	_ "modernc.org/sqlite" // 纯 Go 驱动，无 CGO 依赖

	"stock-analysis/internal/model"
	"stock-analysis/internal/storage"
)

// 通用 SQL：表结构。
const schemaSQL = `
CREATE TABLE IF NOT EXISTS convertible_kline (
    code          TEXT    NOT NULL,
    date          TEXT    NOT NULL,
    open          REAL    NOT NULL,
    close         REAL    NOT NULL,
    high          REAL    NOT NULL,
    low           REAL    NOT NULL,
    volume        INTEGER NOT NULL DEFAULT 0,
    amount        REAL    NOT NULL DEFAULT 0,
    convert_price REAL    NOT NULL DEFAULT 0,
    convert_value REAL    NOT NULL DEFAULT 0,
    premium_rate  REAL    NOT NULL DEFAULT 0,
    PRIMARY KEY (code, date)
);
CREATE INDEX IF NOT EXISTS idx_kline_code_date ON convertible_kline (code, date);

CREATE TABLE IF NOT EXISTS convert_bond (
    code                  TEXT PRIMARY KEY,
    name                  TEXT NOT NULL,
    stock_code            TEXT NOT NULL DEFAULT '',
    initial_convert_price REAL NOT NULL DEFAULT 0,
    list_date             TEXT NOT NULL DEFAULT '',
    delist_date           TEXT NOT NULL DEFAULT ''
);

CREATE TABLE IF NOT EXISTS convert_price_history (
	    code           TEXT NOT NULL,
	    effective_date TEXT NOT NULL,
	    price          REAL NOT NULL,
	    PRIMARY KEY (code, effective_date)
	);

	CREATE TABLE IF NOT EXISTS stock_fundamental (
		    code        TEXT PRIMARY KEY,
		    name        TEXT    NOT NULL DEFAULT '',
		    roe         REAL    NOT NULL DEFAULT 0,
		    debt_ratio  REAL    NOT NULL DEFAULT 0,
		    net_profit  REAL    NOT NULL DEFAULT 0,
		    status      INTEGER NOT NULL DEFAULT 1,
		    total_asset REAL    NOT NULL DEFAULT 0,
		    total_debt  REAL    NOT NULL DEFAULT 0,
		    update_date TEXT    NOT NULL DEFAULT ''
		);

		CREATE TABLE IF NOT EXISTS user (
		    id            INTEGER PRIMARY KEY AUTOINCREMENT,
		    username      TEXT    NOT NULL DEFAULT '',
		    email         TEXT    NOT NULL UNIQUE,
		    password_hash TEXT    NOT NULL,
		    created_at    TEXT    NOT NULL DEFAULT '',
		    updated_at    TEXT    NOT NULL DEFAULT ''
		);

		CREATE TABLE IF NOT EXISTS verify_code (
		    id         INTEGER PRIMARY KEY AUTOINCREMENT,
		    email      TEXT    NOT NULL,
		    code       TEXT    NOT NULL,
		    purpose    TEXT    NOT NULL DEFAULT 'register',
		    expires_at INTEGER NOT NULL,
		    used       INTEGER NOT NULL DEFAULT 0,
		    created_at TEXT    NOT NULL DEFAULT CURRENT_TIMESTAMP
		);
		CREATE INDEX IF NOT EXISTS idx_verify_code_email ON verify_code (email, purpose);

		CREATE TABLE IF NOT EXISTS ingest_task (
		    id           TEXT PRIMARY KEY,
		    type         TEXT NOT NULL,
		    status       TEXT NOT NULL,
		    total        INTEGER NOT NULL DEFAULT 0,
		    done         INTEGER NOT NULL DEFAULT 0,
		    success      INTEGER NOT NULL DEFAULT 0,
		    failed       INTEGER NOT NULL DEFAULT 0,
		    failed_codes TEXT NOT NULL DEFAULT '',
		    params       TEXT NOT NULL DEFAULT '',
		    error        TEXT NOT NULL DEFAULT '',
		    started_at   TEXT NOT NULL,
		    finished_at  TEXT NOT NULL DEFAULT ''
		);
		CREATE INDEX IF NOT EXISTS idx_ingest_task_status ON ingest_task (status);
		CREATE INDEX IF NOT EXISTS idx_ingest_task_started ON ingest_task (started_at);
		`

// 通用 SQL：幂等写入（存在则跳过）。
const upsertSQL = `
INSERT INTO convertible_kline (code, date, open, close, high, low, volume, amount, convert_price, convert_value, premium_rate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (code, date) DO NOTHING;
`

const (
	querySQL    = `SELECT code, date, open, close, high, low, volume, amount, convert_price, convert_value, premium_rate FROM convertible_kline WHERE code = ? ORDER BY date;`
	countSQL    = `SELECT COUNT(*) FROM convertible_kline WHERE code = ?;`
	lastDateSQL = `SELECT MAX(date) FROM convertible_kline WHERE code = ?;`
	allKlineSQL = `SELECT code, date, open, close, high, low, volume, amount, convert_price, convert_value, premium_rate FROM convertible_kline ORDER BY code, date;`

	saveBondSQL = `
INSERT INTO convert_bond (code, name, stock_code, initial_convert_price, list_date, delist_date)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT (code) DO UPDATE SET
    name = excluded.name,
    stock_code = excluded.stock_code,
    initial_convert_price = excluded.initial_convert_price,
    list_date = excluded.list_date,
    delist_date = excluded.delist_date;`

	saveConvertPriceSQL = `
INSERT INTO convert_price_history (code, effective_date, price)
VALUES (?, ?, ?)
ON CONFLICT (code, effective_date) DO UPDATE SET price = excluded.price;`

	updatePremiumSQL = `UPDATE convertible_kline SET convert_price = ?, convert_value = ?, premium_rate = ? WHERE code = ? AND date = ?;`

	saveFundamentalSQL = `
	INSERT INTO stock_fundamental (code, name, roe, debt_ratio, net_profit, status, total_asset, total_debt, update_date)
	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
	ON CONFLICT (code) DO UPDATE SET
	    name = excluded.name,
	    roe = excluded.roe,
	    debt_ratio = excluded.debt_ratio,
	    net_profit = excluded.net_profit,
	    status = excluded.status,
	    total_asset = excluded.total_asset,
	    total_debt = excluded.total_debt,
	    update_date = excluded.update_date;`

	getFundamentalSQL = `SELECT code, name, roe, debt_ratio, net_profit, status, total_asset, total_debt, update_date FROM stock_fundamental WHERE code = ?;`

	allFundamentalSQL = `SELECT code, name, roe, debt_ratio, net_profit, status, total_asset, total_debt, update_date FROM stock_fundamental;`
)

// PremiumCount 返回指定转债已计算转股溢价率的K线数量（步骤3断点续跑依据）。
func (s *Store) PremiumCount(ctx context.Context, code string) (int, error) {
	var n int
	err := s.db.QueryRowContext(ctx,
		"SELECT COUNT(*) FROM convertible_kline WHERE code = ? AND premium_rate <> 0", code).Scan(&n)
	return n, err
}

// PremiumRow 表示单日溢价率计算结果（用于批量更新）。
type PremiumRow struct {
	Date         string
	ConvertPrice float64
	ConvertValue float64
	PremiumRate  float64
}

// Store 是 SQLite 存储实现，实现 storage.KlineStore。
type Store struct {
	db   *sql.DB
	path string
}

// Path 返回数据库文件路径。
func (s *Store) Path() string { return s.path }

// Open 打开（或创建）SQLite 数据库文件并初始化表结构。
func Open(path string) (*Store, error) {
	db, err := sql.Open("sqlite", path)
	if err != nil {
		return nil, fmt.Errorf("open sqlite %s: %w", path, err)
	}
	// SQLite 同一时刻仅允许一个写者，限制连接数避免写锁竞争。
	db.SetMaxOpenConns(1)

	if err := initSchema(db); err != nil {
		db.Close()
		return nil, err
	}
	if err := migrateKlineColumns(db); err != nil {
		db.Close()
		return nil, err
	}
	if err := migrateUserColumns(db); err != nil {
		db.Close()
		return nil, err
	}
	// 关闭 journal，避免在受限文件系统上创建 journal 文件导致写库失败（SQLite_IOERR）。
	if _, err := db.Exec("PRAGMA journal_mode=OFF"); err != nil {
		db.Close()
		return nil, fmt.Errorf("set journal mode: %w", err)
	}
	// 禁用 mmap 读取：modernc 驱动默认内存映射数据库文件，在受限文件系统上会报 IOERR_READ。
	if _, err := db.Exec("PRAGMA mmap_size=0"); err != nil {
		db.Close()
		return nil, fmt.Errorf("set mmap size: %w", err)
	}
	// 等待锁：避免其他进程（杀软/索引）短暂占用文件时报错。
	if _, err := db.Exec("PRAGMA busy_timeout=8000"); err != nil {
		db.Close()
		return nil, fmt.Errorf("set busy timeout: %w", err)
	}
	return &Store{db: db, path: path}, nil
}

// initSchema 执行建表语句。
func initSchema(db *sql.DB) error {
	if _, err := db.Exec(schemaSQL); err != nil {
		return fmt.Errorf("init schema: %w", err)
	}
	return nil
}

// migrateKlineColumns 为旧库的 convertible_kline 表补齐溢价率相关列。
func migrateKlineColumns(db *sql.DB) error {
	rows, err := db.Query("PRAGMA table_info(convertible_kline)")
	if err != nil {
		return fmt.Errorf("table info: %w", err)
	}
	defer rows.Close()

	have := make(map[string]bool)
	for rows.Next() {
		var (
			cid       int
			name, typ string
			notnull   int
			dflt      any
			pk        int
		)
		if err := rows.Scan(&cid, &name, &typ, &notnull, &dflt, &pk); err != nil {
			return fmt.Errorf("scan table info: %w", err)
		}
		have[name] = true
	}
	if err := rows.Err(); err != nil {
		return err
	}

	for _, col := range []string{"convert_price", "convert_value", "premium_rate"} {
		if have[col] {
			continue
		}
		if _, err := db.Exec(fmt.Sprintf("ALTER TABLE convertible_kline ADD COLUMN %s REAL NOT NULL DEFAULT 0", col)); err != nil {
			return fmt.Errorf("add column %s: %w", col, err)
		}
	}
	return nil
}

// migrateUserColumns 为旧库的 user 表补齐 username 列并创建索引。
func migrateUserColumns(db *sql.DB) error {
	rows, err := db.Query("PRAGMA table_info(user)")
	if err != nil {
		return fmt.Errorf("table info user: %w", err)
	}
	hasUsername := false
	for rows.Next() {
		var cid int
		var name, colType string
		var notNull, pk int
		var dflt sql.NullString
		if err := rows.Scan(&cid, &name, &colType, &notNull, &dflt, &pk); err != nil {
			rows.Close()
			return fmt.Errorf("scan column: %w", err)
		}
		if name == "username" {
			hasUsername = true
		}
	}
	rows.Close()

	if !hasUsername {
		if _, err := db.Exec("ALTER TABLE user ADD COLUMN username TEXT NOT NULL DEFAULT ''"); err != nil {
			return fmt.Errorf("add username column: %w", err)
		}
	}
	// 始终创建索引（幂等）。
	if _, err := db.Exec("CREATE UNIQUE INDEX IF NOT EXISTS idx_user_username ON user (username) WHERE username != ''"); err != nil {
		return fmt.Errorf("create username index: %w", err)
	}
	return nil
}

// SaveKlines 幂等批量写入K线。
// 长进程中数据库文件可能被系统安全软件短暂占用导致 IOERR，此处带有限次重试。
func (s *Store) SaveKlines(ctx context.Context, code string, klines []model.Kline) error {
	if len(klines) == 0 {
		return nil
	}
	const maxAttempt = 3
	var lastErr error
	for attempt := 0; attempt < maxAttempt; attempt++ {
		if attempt > 0 {
			time.Sleep(2 * time.Second)
		}
		if err := s.saveOnce(ctx, code, klines); err != nil {
			lastErr = err
			continue
		}
		return nil
	}
	return lastErr
}

// saveOnce 单次执行批量写入事务。
func (s *Store) saveOnce(ctx context.Context, code string, klines []model.Kline) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback() //nolint:errcheck // 提交成功后回滚为无操作

	stmt, err := tx.PrepareContext(ctx, upsertSQL)
	if err != nil {
		return fmt.Errorf("prepare upsert: %w", err)
	}
	defer stmt.Close()

	for _, k := range klines {
		if _, err := stmt.ExecContext(ctx, code, k.Date, k.Open, k.Close, k.High, k.Low, k.Volume, k.Amount, k.ConvertPrice, k.ConvertValue, k.PremiumRate); err != nil {
			return fmt.Errorf("insert kline %s %s: %w", code, k.Date, err)
		}
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit tx: %w", err)
	}
	return nil
}

// GetKlines 按时间升序返回指定标的的全部K线。
func (s *Store) GetKlines(ctx context.Context, code string) ([]model.Kline, error) {
	rows, err := s.db.QueryContext(ctx, querySQL, code)
	if err != nil {
		return nil, fmt.Errorf("query klines: %w", err)
	}
	defer rows.Close()

	var klines []model.Kline
	for rows.Next() {
		k, code, err := scanKline(rows)
		if err != nil {
			return nil, err
		}
		_ = code
		klines = append(klines, k)
	}
	return klines, rows.Err()
}

// ListCodes 返回库中全部标的代码（升序）。
func (s *Store) ListCodes(ctx context.Context) ([]string, error) {
	rows, err := s.db.QueryContext(ctx, "SELECT DISTINCT code FROM convertible_kline ORDER BY code")
	if err != nil {
		return nil, fmt.Errorf("list codes: %w", err)
	}
	defer rows.Close()

	var codes []string
	for rows.Next() {
		var code string
		if err := rows.Scan(&code); err != nil {
			return nil, fmt.Errorf("scan code: %w", err)
		}
		codes = append(codes, code)
	}
	return codes, rows.Err()
}

// LastKlineDates 一次性返回所有已入库 code 的最新K线日期（YYYY-MM-DD）。
// 用于增量同步预加载，替代逐只 LastKlineDate 查询，避免占用唯一写连接。
func (s *Store) LastKlineDates(ctx context.Context) (map[string]string, error) {
	rows, err := s.db.QueryContext(ctx, "SELECT code, MAX(date) FROM convertible_kline GROUP BY code")
	if err != nil {
		return nil, fmt.Errorf("last kline dates: %w", err)
	}
	defer rows.Close()

	out := make(map[string]string)
	for rows.Next() {
		var code, date string
		if err := rows.Scan(&code, &date); err != nil {
			return nil, fmt.Errorf("scan last date: %w", err)
		}
		out[code] = date
	}
	return out, rows.Err()
}

// SaveKlinesBatch 单事务批量写入多只转债K线（collector 调用）。
// 接收 map[转债代码]K线切片，在一个事务内遍历写入，带 3 次重试。
func (s *Store) SaveKlinesBatch(ctx context.Context, batch map[string][]model.Kline) error {
	if len(batch) == 0 {
		return nil
	}
	const maxAttempt = 3
	var lastErr error
	for attempt := 0; attempt < maxAttempt; attempt++ {
		if attempt > 0 {
			time.Sleep(2 * time.Second)
		}
		if err := s.saveBatchOnce(ctx, batch); err != nil {
			lastErr = err
			continue
		}
		return nil
	}
	return lastErr
}

// saveBatchOnce 单次执行批量写入事务。
func (s *Store) saveBatchOnce(ctx context.Context, batch map[string][]model.Kline) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback() //nolint:errcheck

	stmt, err := tx.PrepareContext(ctx, upsertSQL)
	if err != nil {
		return fmt.Errorf("prepare upsert: %w", err)
	}
	defer stmt.Close()

	for code, klines := range batch {
		for _, k := range klines {
			if _, err := stmt.ExecContext(ctx, code, k.Date, k.Open, k.Close, k.High, k.Low, k.Volume, k.Amount, k.ConvertPrice, k.ConvertValue, k.PremiumRate); err != nil {
				return fmt.Errorf("insert kline %s %s: %w", code, k.Date, err)
			}
		}
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit tx: %w", err)
	}
	return nil
}

// LoadAllKlines 一次性加载全部标的的K线（按代码、日期升序）。
// 用于回测等需要全量数据在内存的场景。
func (s *Store) LoadAllKlines(ctx context.Context) (map[string][]model.Kline, error) {
	rows, err := s.db.QueryContext(ctx, allKlineSQL)
	if err != nil {
		return nil, fmt.Errorf("load all klines: %w", err)
	}
	defer rows.Close()

	out := make(map[string][]model.Kline)
	for rows.Next() {
		k, code, err := scanKline(rows)
		if err != nil {
			return nil, err
		}
		out[code] = append(out[code], k)
	}
	return out, rows.Err()
}

// scanKline 从查询行扫描一条K线记录。
func scanKline(rows *sql.Rows) (model.Kline, string, error) {
	var (
		k          model.Kline
		code, date string
	)
	if err := rows.Scan(&code, &date, &k.Open, &k.Close, &k.High, &k.Low, &k.Volume, &k.Amount, &k.ConvertPrice, &k.ConvertValue, &k.PremiumRate); err != nil {
		return model.Kline{}, "", fmt.Errorf("scan kline: %w", err)
	}
	k.Date = date
	return k, code, nil
}

// CountKlines 返回指定标的的K线条数。
func (s *Store) CountKlines(ctx context.Context, code string) (int, error) {
	var n int
	if err := s.db.QueryRowContext(ctx, countSQL, code).Scan(&n); err != nil {
		return 0, fmt.Errorf("count klines: %w", err)
	}
	return n, nil
}

// LastKlineDate 返回指定标的的最新K线日期（YYYY-MM-DD），无记录返回空串。
func (s *Store) LastKlineDate(ctx context.Context, code string) (string, error) {
	var date sql.NullString
	if err := s.db.QueryRowContext(ctx, lastDateSQL, code).Scan(&date); err != nil {
		return "", fmt.Errorf("last kline date: %w", err)
	}
	if !date.Valid {
		return "", nil
	}
	return date.String, nil
}

// SaveBonds 幂等写入转债基础信息（含正股代码/初始转股价）。
func (s *Store) SaveBonds(ctx context.Context, bonds []model.ConvertibleBond) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback() //nolint:errcheck

	stmt, err := tx.PrepareContext(ctx, saveBondSQL)
	if err != nil {
		return fmt.Errorf("prepare save bond: %w", err)
	}
	defer stmt.Close()

	for _, b := range bonds {
		if _, err := stmt.ExecContext(ctx, b.Code, b.Name, b.StockCode, b.InitialConvertPrice, b.ListDate, b.DelistDate); err != nil {
			return fmt.Errorf("save bond %s: %w", b.Code, err)
		}
	}
	return tx.Commit()
}

// ListBonds 返回全部转债基础信息。
func (s *Store) ListBonds(ctx context.Context) ([]model.ConvertibleBond, error) {
	rows, err := s.db.QueryContext(ctx, "SELECT code, name, stock_code, initial_convert_price, list_date, delist_date FROM convert_bond ORDER BY code")
	if err != nil {
		return nil, fmt.Errorf("list bonds: %w", err)
	}
	defer rows.Close()

	var bonds []model.ConvertibleBond
	for rows.Next() {
		var b model.ConvertibleBond
		if err := rows.Scan(&b.Code, &b.Name, &b.StockCode, &b.InitialConvertPrice, &b.ListDate, &b.DelistDate); err != nil {
			return nil, fmt.Errorf("scan bond: %w", err)
		}
		bonds = append(bonds, b)
	}
	return bonds, rows.Err()
}

// SaveConvertPrices 幂等写入转股价历史。
func (s *Store) SaveConvertPrices(ctx context.Context, code string, prices []model.ConvertPrice) error {
	if len(prices) == 0 {
		return nil
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback() //nolint:errcheck

	stmt, err := tx.PrepareContext(ctx, saveConvertPriceSQL)
	if err != nil {
		return fmt.Errorf("prepare save convert price: %w", err)
	}
	defer stmt.Close()

	for _, p := range prices {
		if _, err := stmt.ExecContext(ctx, code, p.EffectiveDate, p.Price); err != nil {
			return fmt.Errorf("save convert price %s %s: %w", code, p.EffectiveDate, err)
		}
	}
	return tx.Commit()
}

// GetConvertPrices 返回指定转债的转股价历史（按生效日期升序）。
func (s *Store) GetConvertPrices(ctx context.Context, code string) ([]model.ConvertPrice, error) {
	rows, err := s.db.QueryContext(ctx, "SELECT effective_date, price FROM convert_price_history WHERE code = ? ORDER BY effective_date", code)
	if err != nil {
		return nil, fmt.Errorf("get convert prices: %w", err)
	}
	defer rows.Close()

	var prices []model.ConvertPrice
	for rows.Next() {
		var p model.ConvertPrice
		if err := rows.Scan(&p.EffectiveDate, &p.Price); err != nil {
			return nil, fmt.Errorf("scan convert price: %w", err)
		}
		p.Code = code
		prices = append(prices, p)
	}
	return prices, rows.Err()
}

// UpdatePremiumBatch 批量更新指定转债每日的转股价/转股价值/溢价率。
func (s *Store) UpdatePremiumBatch(ctx context.Context, code string, rowsData []PremiumRow) error {
	if len(rowsData) == 0 {
		return nil
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback() //nolint:errcheck

	stmt, err := tx.PrepareContext(ctx, updatePremiumSQL)
	if err != nil {
		return fmt.Errorf("prepare update premium: %w", err)
	}
	defer stmt.Close()

	for _, r := range rowsData {
		if _, err := stmt.ExecContext(ctx, r.ConvertPrice, r.ConvertValue, r.PremiumRate, code, r.Date); err != nil {
			return fmt.Errorf("update premium %s %s: %w", code, r.Date, err)
		}
	}
	return tx.Commit()
}

// SaveStockFundamental 保存单条正股基本面数据。
func (s *Store) SaveStockFundamental(ctx context.Context, f model.StockFundamental) error {
	_, err := s.db.ExecContext(ctx, saveFundamentalSQL,
		f.Code, f.Name, f.ROE, f.DebtRatio, f.NetProfit, f.Status, f.TotalAsset, f.TotalDebt, f.UpdateDate)
	return err
}

// GetStockFundamental 获取指定正股的基本面数据。
func (s *Store) GetStockFundamental(ctx context.Context, code string) (*model.StockFundamental, error) {
	row := s.db.QueryRowContext(ctx, getFundamentalSQL, code)
	var f model.StockFundamental
	if err := row.Scan(&f.Code, &f.Name, &f.ROE, &f.DebtRatio, &f.NetProfit, &f.Status, &f.TotalAsset, &f.TotalDebt, &f.UpdateDate); err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, fmt.Errorf("get fundamental %s: %w", code, err)
	}
	return &f, nil
}

// LoadAllFundamentals 加载全部正股基本面数据。
func (s *Store) LoadAllFundamentals(ctx context.Context) (map[string]*model.StockFundamental, error) {
	rows, err := s.db.QueryContext(ctx, allFundamentalSQL)
	if err != nil {
		return nil, fmt.Errorf("load all fundamentals: %w", err)
	}
	defer rows.Close()

	out := make(map[string]*model.StockFundamental)
	for rows.Next() {
		var f model.StockFundamental
		if err := rows.Scan(&f.Code, &f.Name, &f.ROE, &f.DebtRatio, &f.NetProfit, &f.Status, &f.TotalAsset, &f.TotalDebt, &f.UpdateDate); err != nil {
			return nil, fmt.Errorf("scan fundamental: %w", err)
		}
		out[f.Code] = &f
	}
	return out, rows.Err()
}

// SaveUser 保存新注册用户。
func (s *Store) SaveUser(ctx context.Context, u model.User) (int64, error) {
	res, err := s.db.ExecContext(ctx,
		`INSERT INTO user (username, email, password_hash, created_at, updated_at) VALUES (?, ?, ?, ?, ?)`,
		u.Username, u.Email, u.PasswordHash, u.CreatedAt, u.UpdatedAt)
	if err != nil {
		return 0, fmt.Errorf("save user: %w", err)
	}
	return res.LastInsertId()
}

// GetUserByID 按 ID 查询用户。
func (s *Store) GetUserByID(ctx context.Context, id int64) (*model.User, error) {
	row := s.db.QueryRowContext(ctx,
		`SELECT id, username, email, password_hash, created_at, updated_at FROM user WHERE id = ?`, id)
	var u model.User
	if err := row.Scan(&u.ID, &u.Username, &u.Email, &u.PasswordHash, &u.CreatedAt, &u.UpdatedAt); err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, fmt.Errorf("get user by id: %w", err)
	}
	return &u, nil
}

// GetUserByEmail 按邮箱查询用户。
func (s *Store) GetUserByEmail(ctx context.Context, email string) (*model.User, error) {
	row := s.db.QueryRowContext(ctx,
		`SELECT id, username, email, password_hash, created_at, updated_at FROM user WHERE email = ?`, email)
	var u model.User
	if err := row.Scan(&u.ID, &u.Username, &u.Email, &u.PasswordHash, &u.CreatedAt, &u.UpdatedAt); err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, fmt.Errorf("get user: %w", err)
	}
	return &u, nil
}

// GetUserByUsername 按用户名查询用户。
func (s *Store) GetUserByUsername(ctx context.Context, username string) (*model.User, error) {
	row := s.db.QueryRowContext(ctx,
		`SELECT id, username, email, password_hash, created_at, updated_at FROM user WHERE username = ?`, username)
	var u model.User
	if err := row.Scan(&u.ID, &u.Username, &u.Email, &u.PasswordHash, &u.CreatedAt, &u.UpdatedAt); err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, fmt.Errorf("get user by username: %w", err)
	}
	return &u, nil
}

// SaveVerifyCode 保存验证码。
func (s *Store) SaveVerifyCode(ctx context.Context, v model.VerifyCode) error {
	_, err := s.db.ExecContext(ctx,
		`INSERT INTO verify_code (email, code, purpose, expires_at, used) VALUES (?, ?, ?, ?, 0)`,
		v.Email, v.Code, v.Purpose, v.ExpiresAt)
	if err != nil {
		return fmt.Errorf("save verify code: %w", err)
	}
	return nil
}

// GetLatestVerifyCode 获取某邮箱某用途最新未使用验证码。
func (s *Store) GetLatestVerifyCode(ctx context.Context, email, purpose string) (*model.VerifyCode, error) {
	row := s.db.QueryRowContext(ctx,
		`SELECT id, email, code, purpose, expires_at, used FROM verify_code
		 WHERE email = ? AND purpose = ? AND used = 0
		 ORDER BY id DESC LIMIT 1`, email, purpose)
	var v model.VerifyCode
	var used int
	var id int64
	if err := row.Scan(&id, &v.Email, &v.Code, &v.Purpose, &v.ExpiresAt, &used); err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, fmt.Errorf("get verify code: %w", err)
	}
	v.Used = used != 0
	return &v, nil
}

// MarkVerifyCodeUsed 将验证码标记为已使用。
func (s *Store) MarkVerifyCodeUsed(ctx context.Context, email, code, purpose string) error {
	_, err := s.db.ExecContext(ctx,
		`UPDATE verify_code SET used = 1 WHERE email = ? AND code = ? AND purpose = ?`,
		email, code, purpose)
	if err != nil {
		return fmt.Errorf("mark verify code used: %w", err)
	}
	return nil
}

// GetStockBondMap 返回正股代码到转债代码的映射。
func (s *Store) GetStockBondMap(ctx context.Context) (map[string]string, error) {
	rows, err := s.db.QueryContext(ctx, "SELECT stock_code, code FROM convert_bond WHERE stock_code != ''")
	if err != nil {
		return nil, fmt.Errorf("get stock bond map: %w", err)
	}
	defer rows.Close()

	m := make(map[string]string)
	for rows.Next() {
		var stock, bond string
		if err := rows.Scan(&stock, &bond); err != nil {
			return nil, fmt.Errorf("scan stock bond: %w", err)
		}
		if m[stock] == "" {
			m[stock] = bond
		}
	}
	return m, rows.Err()
}

// Close 关闭数据库连接。
func (s *Store) Close() error {
	return s.db.Close()
}

// 编译期断言：确保 Store 实现 storage.KlineStore。
var _ storage.KlineStore = (*Store)(nil)