Agent-System-Backend/internal/store/messages.go

package store

import (
	"crypto/rand"
	"database/sql"
	"encoding/hex"
	"time"

	core "agent-system-core"
)

// ---- 消息 ----

func (s *Store) AppendMessage(m *core.Message) error {
	if m.ID == "" {
		m.ID = newID("msg")
	}
	m.CreatedAt = m.CreatedAt.UTC()
	var next int
	if err := s.db.QueryRow(`SELECT COALESCE(MAX(seq),0)+1 FROM messages WHERE conversation_id=?`, m.ConversationID).Scan(&next); err != nil {
		return err
	}
	_, err := s.db.Exec(`INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at) VALUES(?,?,?,?,?,?,?)`,
		m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), next, nowISO())
	if err == nil {
		s.TouchConversation(m.ConversationID)
	}
	return err
}

func (s *Store) ListMessages(conversationID string) ([]core.Message, error) {
	rows, err := s.db.Query(`SELECT id,conversation_id,run_id,role,parts,created_at FROM messages WHERE conversation_id=? ORDER BY seq`, conversationID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []core.Message
	for rows.Next() {
		var m core.Message
		var parts, ca string
		if err := rows.Scan(&m.ID, &m.ConversationID, &m.RunID, &m.Role, &parts, &ca); err != nil {
			return nil, err
		}
		m.Role = core.Role(m.Role)
		m.Parts, _ = unmarshalJSON[[]core.Part](parts)
		m.CreatedAt, _ = parseTime(ca)
		out = append(out, m)
	}
	return out, rows.Err()
}

func (s *Store) LastMessages(conversationID string, n int) ([]core.Message, error) {
	all, err := s.ListMessages(conversationID)
	if err != nil {
		return nil, err
	}
	if len(all) > n {
		all = all[len(all)-n:]
	}
	return all, nil
}

// ---- Runs ----

func (s *Store) CreateRun(r *core.Run) error {
	if r.ID == "" {
		r.ID = newID("run")
	}
	r.CreatedAt = r.CreatedAt.UTC()
	_, err := s.db.Exec(`INSERT INTO runs(id,parent_run_id,conversation_id,project_id,kind,profile_id,status,phase,agent_name,created_at) VALUES(?,?,?,?,?,?,?,?,?,?)`,
		r.ID, r.ParentRunID, r.ConversationID, r.ProjectID, string(r.Kind), r.ProfileID, string(r.Status), string(r.Phase), r.AgentName, nowISO())
	return err
}

func (s *Store) UpdateRun(r *core.Run) error {
	var ended any
	if r.EndedAt != nil {
		ended = nowISO()
	}
	_, err := s.db.Exec(`UPDATE runs SET status=?, phase=?, usage=?, error=?, ended_at=? WHERE id=?`,
		string(r.Status), string(r.Phase), marshalJSON(r.Usage), marshalJSON(r.Error), ended, r.ID)
	return err
}

const runColumns = `id,parent_run_id,conversation_id,project_id,kind,profile_id,status,phase,agent_name,created_at,ended_at,usage,error`

func scanRun(scan func(...any) error) (core.Run, error) {
	var r core.Run
	var parent, profile, phase, agent, ended, usage, errJSON *string
	var kind, status string
	var ca string
	if err := scan(&r.ID, &parent, &r.ConversationID, &r.ProjectID, &kind, &profile, &status, &phase, &agent, &ca, &ended, &usage, &errJSON); err != nil {
		return r, err
	}
	if parent != nil {
		r.ParentRunID = *parent
	}
	if profile != nil {
		r.ProfileID = *profile
	}
	r.Kind = core.RunKind(kind)
	r.Status = core.RunStatus(status)
	if phase != nil {
		r.Phase = core.RunPhase(*phase)
	}
	if agent != nil {
		r.AgentName = *agent
	}
	r.CreatedAt, _ = parseTime(ca)
	if ended != nil {
		t, _ := parseTime(*ended)
		r.EndedAt = &t
	}
	if usage != nil && *usage != "" && *usage != "null" {
		r.Usage, _ = unmarshalJSON[*core.Usage](*usage)
	}
	if errJSON != nil && *errJSON != "" && *errJSON != "null" {
		r.Error, _ = unmarshalJSON[*core.ErrorPart](*errJSON)
	}
	return r, nil
}

func (s *Store) GetRun(id string) (core.Run, error) {
	r, err := scanRun(func(dst ...any) error {
		return s.db.QueryRow(`SELECT `+runColumns+` FROM runs WHERE id=?`, id).Scan(dst...)
	})
	if err != nil {
		if err.Error() == "sql: no rows" {
			return r, ErrNotFound
		}
		return r, err
	}
	return r, nil
}

func (s *Store) ListRunsByConversation(conversationID string) ([]core.Run, error) {
	rows, err := s.db.Query(`SELECT `+runColumns+` FROM runs WHERE conversation_id=? ORDER BY created_at`, conversationID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []core.Run
	for rows.Next() {
		r, err := scanRun(rows.Scan)
		if err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

// ChildRuns 返回直接子 run（任务树）。
func (s *Store) ChildRuns(runID string) ([]core.Run, error) {
	rows, err := s.db.Query(`SELECT `+runColumns+` FROM runs WHERE parent_run_id=? ORDER BY created_at`, runID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []core.Run
	for rows.Next() {
		r, err := scanRun(rows.Scan)
		if err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

// ---- 事件（seq 游标权威）----

func (s *Store) AppendEvent(e *core.StreamEvent, payload any) (int64, error) {
	var payloadStr any
	if payload != nil {
		payloadStr = marshalJSON(payload)
	}
	res, err := s.db.Exec(`INSERT INTO events(conversation_id,run_id,ts,v,type,payload) VALUES(?,?,?,?,?,?)`,
		e.ConversationID, e.RunID, e.TS, e.V, string(e.Type), payloadStr)
	if err != nil {
		return 0, err
	}
	return res.LastInsertId()
}

func (s *Store) EventsAfter(conversationID string, since int64, limit int) ([]core.StreamEvent, error) {
	rows, err := s.db.Query(`SELECT seq,run_id,ts,v,type,payload FROM events WHERE conversation_id=? AND seq>? ORDER BY seq LIMIT ?`,
		conversationID, since, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []core.StreamEvent
	for rows.Next() {
		var e core.StreamEvent
		var typ string
		var payload []byte
		if err := rows.Scan(&e.Seq, &e.RunID, &e.TS, &e.V, &typ, &payload); err != nil {
			return nil, err
		}
		e.ConversationID = conversationID
		e.Type = core.StreamEventType(typ)
		if len(payload) > 0 {
			e.Payload = payload
		}
		out = append(out, e)
	}
	return out, rows.Err()
}

func (s *Store) LastEventSeq(conversationID string) (int64, error) {
	var seq sql.NullInt64
	err := s.db.QueryRow(`SELECT MAX(seq) FROM events WHERE conversation_id=?`, conversationID).Scan(&seq)
	if err != nil {
		return 0, err
	}
	return seq.Int64, nil
}

func newID(kind string) string {
	b := make([]byte, 6)
	rand.Read(b)
	return kind + "_" + hex.EncodeToString(b)
}

func parseTime(s string) (time.Time, error) {
	return time.Parse(time.RFC3339Nano, s)
}

// NewID 生成带前缀的随机 ID（导出供 runmgr/server 使用）。
func NewID(kind string) string { return newID(kind) }