Files
glchat/internal/store/messages.go
T

613 lines
20 KiB
Go
Raw Normal View History

package store
import (
"context"
"database/sql"
"encoding/json"
"strings"
"time"
)
// MessageType — тип сообщения (AGENT.md 7.6): обычное, системное или шёпот.
type MessageType string
const (
MessageDefault MessageType = "default"
MessageSystem MessageType = "system"
MessageWhisper MessageType = "whisper"
// MessageAction — блок действия (/me), MessageScream — громкий блок,
// MessagePrivate — личное сообщение (/ls): видно только автору и адресатам
// (AGENT.md 7.6).
MessageAction MessageType = "action"
MessageScream MessageType = "scream"
MessagePrivate MessageType = "private"
)
// IsPrivate сообщает, что сообщение видно только автору и адресатам.
func (m MessageType) IsPrivate() bool { return m == MessagePrivate }
// Message — сообщение комнаты.
type Message struct {
ID uint64
ChannelID uint64
AuthorID *uint64
Content string
ReplyToID *uint64
Type MessageType
EditedAt *time.Time
Pinned bool
Attachments []Attachment
Mentions []uint64
CreatedAt time.Time
}
// Attachment — метаданные вложения: файл регистрируется в таблице files.
type Attachment struct {
FileID uint64 `json:"file_id"`
Filename string `json:"filename"`
ContentType string `json:"content_type"`
SizeBytes int64 `json:"size_bytes"`
Width int `json:"width,omitempty"`
Height int `json:"height,omitempty"`
}
// Reaction — агрегированная реакция: эмодзи и кто её поставил.
type Reaction struct {
Emoji string
Count int
UserIDs []uint64
Me bool
}
// ReadState — состояние прочтения комнаты пользователем.
type ReadState struct {
UserID uint64
ChannelID uint64
LastMessageID uint64
MentionCount int
UpdatedAt time.Time
}
const messageColumns = `id, channel_id, author_id, content, reply_to_id, type,
attachments_json, mentions_json, edited_at, pinned, created_at, content_lower`
// CreateMessageParams — параметры нового сообщения.
type CreateMessageParams struct {
ID uint64
ChannelID uint64
AuthorID uint64
Content string
ReplyToID *uint64
Type MessageType
Attachments []Attachment
Mentions []uint64
}
// CreateMessage сохраняет сообщение; пустое содержимое без вложений запрещено
// на уровне API, здесь только запись (AGENT.md 7.6).
func (s *Store) CreateMessage(ctx context.Context, params CreateMessageParams) (*Message, error) {
if params.ID == 0 {
params.ID = s.NextID()
}
if params.Type == "" {
params.Type = MessageDefault
}
attachments, err := json.Marshal(orEmptyAttachments(params.Attachments))
if err != nil {
return nil, err
}
mentions, err := json.Marshal(orEmptyIDs(params.Mentions))
if err != nil {
return nil, err
}
var author any
if params.AuthorID != 0 {
author = int64(params.AuthorID)
}
_, err = s.writer.ExecContext(ctx, `
INSERT INTO messages (id, channel_id, author_id, content, reply_to_id, type,
attachments_json, mentions_json, pinned, created_at, content_lower)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)`,
int64(params.ID), int64(params.ChannelID), author, params.Content,
nullableID(params.ReplyToID), string(params.Type), string(attachments), string(mentions),
s.Now(), strings.ToLower(params.Content))
if err != nil {
return nil, mapError(err)
}
return s.GetMessage(ctx, params.ID)
}
func (s *Store) GetMessage(ctx context.Context, id uint64) (*Message, error) {
row := s.reader.QueryRowContext(ctx, `SELECT `+messageColumns+` FROM messages WHERE id = ?`, int64(id))
return scanMessage(row)
}
// ListMessages возвращает сообщения комнаты от новых к старым. beforeID
// используется для подгрузки истории вверх, viewerID — чтобы не отдавать
// личные сообщения (/ls) посторонним (AGENT.md 7.6).
func (s *Store) ListMessages(ctx context.Context, channelID, beforeID, viewerID uint64, limit int) ([]Message, error) {
if limit <= 0 || limit > 100 {
limit = 50
}
query := `SELECT ` + messageColumns + ` FROM messages WHERE channel_id = ?`
args := []any{int64(channelID)}
if beforeID > 0 {
query += ` AND id < ?`
args = append(args, int64(beforeID))
}
query += ` AND (` + privateVisibilityClause + `)`
args = append(args, int64(viewerID), int64(viewerID))
query += ` ORDER BY id DESC LIMIT ?`
args = append(args, limit)
rows, err := s.reader.QueryContext(ctx, query, args...)
if err != nil {
return nil, err
}
defer rows.Close()
messages := make([]Message, 0, limit)
for rows.Next() {
message, err := scanMessage(rows)
if err != nil {
return nil, err
}
messages = append(messages, *message)
}
return messages, rows.Err()
}
// privateVisibilityClause пропускает обычные сообщения и личные, если зритель
// автор или адресат (адресаты хранятся в mentions_json).
const privateVisibilityClause = `type <> 'private'
OR author_id = ?
OR EXISTS (SELECT 1 FROM json_each(messages.mentions_json) WHERE json_each.value = ?)`
// ListPinnedMessages возвращает закреплённые сообщения комнаты.
func (s *Store) ListPinnedMessages(ctx context.Context, channelID uint64, limit int) ([]Message, error) {
if limit <= 0 || limit > 100 {
limit = 50
}
rows, err := s.reader.QueryContext(ctx, `
SELECT `+messageColumns+` FROM messages
WHERE channel_id = ? AND pinned = 1 ORDER BY id DESC LIMIT ?`, int64(channelID), limit)
if err != nil {
return nil, err
}
defer rows.Close()
messages := make([]Message, 0, limit)
for rows.Next() {
message, err := scanMessage(rows)
if err != nil {
return nil, err
}
messages = append(messages, *message)
}
return messages, rows.Err()
}
// UpdateMessageContent меняет текст сообщения и фиксирует время правки.
func (s *Store) UpdateMessageContent(ctx context.Context, id uint64, content string) (*Message, error) {
result, err := s.writer.ExecContext(ctx,
`UPDATE messages SET content = ?, content_lower = ?, edited_at = ? WHERE id = ?`,
content, strings.ToLower(content), s.Now(), int64(id))
if err != nil {
return nil, err
}
if affected, err := result.RowsAffected(); err == nil && affected == 0 {
return nil, ErrNotFound
}
return s.GetMessage(ctx, id)
}
// SetMessagePinned закрепляет или открепляет сообщение (AGENT.md 7.6).
func (s *Store) SetMessagePinned(ctx context.Context, id uint64, pinned bool) error {
result, err := s.writer.ExecContext(ctx,
`UPDATE messages SET pinned = ? WHERE id = ?`, boolToInt(pinned), int64(id))
if err != nil {
return err
}
if affected, err := result.RowsAffected(); err == nil && affected == 0 {
return ErrNotFound
}
return nil
}
func (s *Store) DeleteMessage(ctx context.Context, id uint64) error {
result, err := s.writer.ExecContext(ctx, `DELETE FROM messages WHERE id = ?`, int64(id))
if err != nil {
return err
}
if affected, err := result.RowsAffected(); err == nil && affected == 0 {
return ErrNotFound
}
return nil
}
// AddReaction ставит реакцию; повторная установка идемпотентна.
func (s *Store) AddReaction(ctx context.Context, messageID, userID uint64, emoji string) error {
_, err := s.writer.ExecContext(ctx, `
INSERT INTO message_reactions (message_id, user_id, emoji, created_at)
VALUES (?, ?, ?, ?) ON CONFLICT (message_id, user_id, emoji) DO NOTHING`,
int64(messageID), int64(userID), emoji, s.Now())
return err
}
// RemoveReaction снимает реакцию пользователя.
func (s *Store) RemoveReaction(ctx context.Context, messageID, userID uint64, emoji string) error {
_, err := s.writer.ExecContext(ctx,
`DELETE FROM message_reactions WHERE message_id = ? AND user_id = ? AND emoji = ?`,
int64(messageID), int64(userID), emoji)
return err
}
// ListReactions отдаёт реакции сообщения, сгруппированные по эмодзи.
func (s *Store) ListReactions(ctx context.Context, messageID, viewerID uint64) ([]Reaction, error) {
rows, err := s.reader.QueryContext(ctx, `
SELECT emoji, user_id FROM message_reactions
WHERE message_id = ? ORDER BY created_at`, int64(messageID))
if err != nil {
return nil, err
}
defer rows.Close()
order := make([]string, 0, 8)
grouped := map[string]*Reaction{}
for rows.Next() {
var (
emoji string
userID uint64
)
if err := rows.Scan(&emoji, &userID); err != nil {
return nil, err
}
reaction, ok := grouped[emoji]
if !ok {
reaction = &Reaction{Emoji: emoji}
grouped[emoji] = reaction
order = append(order, emoji)
}
reaction.Count++
reaction.UserIDs = append(reaction.UserIDs, userID)
if userID == viewerID {
reaction.Me = true
}
}
if err := rows.Err(); err != nil {
return nil, err
}
reactions := make([]Reaction, 0, len(order))
for _, emoji := range order {
reactions = append(reactions, *grouped[emoji])
}
return reactions, nil
}
// SetReadState сохраняет позицию прочтения и счётчик упоминаний.
func (s *Store) SetReadState(ctx context.Context, userID, channelID, lastMessageID uint64, mentionCount int) error {
_, err := s.writer.ExecContext(ctx, `
INSERT INTO channel_read_states (user_id, channel_id, last_message_id, mention_count, updated_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT (user_id, channel_id) DO UPDATE SET
last_message_id = excluded.last_message_id,
mention_count = excluded.mention_count,
updated_at = excluded.updated_at`,
int64(userID), int64(channelID), int64(lastMessageID), mentionCount, s.Now())
return err
}
// BumpMentionCount увеличивает счётчик упоминаний пользователя в комнате и
// возвращает новое значение (AGENT.md 7.16).
func (s *Store) BumpMentionCount(ctx context.Context, userID, channelID uint64) (int, error) {
var count int
err := s.writer.QueryRowContext(ctx, `
INSERT INTO channel_read_states (user_id, channel_id, last_message_id, mention_count, updated_at)
VALUES (?, ?, 0, 1, ?)
ON CONFLICT (user_id, channel_id) DO UPDATE SET
mention_count = channel_read_states.mention_count + 1,
updated_at = excluded.updated_at
RETURNING mention_count`,
int64(userID), int64(channelID), s.Now()).Scan(&count)
if err != nil {
return 0, mapError(err)
}
return count, nil
}
// ListReadStates отдаёт состояния прочтения пользователя (для READY).
func (s *Store) ListReadStates(ctx context.Context, userID uint64) ([]ReadState, error) {
rows, err := s.reader.QueryContext(ctx, `
SELECT user_id, channel_id, last_message_id, mention_count, updated_at
FROM channel_read_states WHERE user_id = ?`, int64(userID))
if err != nil {
return nil, err
}
defer rows.Close()
states := make([]ReadState, 0, 16)
for rows.Next() {
var (
state ReadState
updatedAt string
)
if err := rows.Scan(&state.UserID, &state.ChannelID, &state.LastMessageID, &state.MentionCount, &updatedAt); err != nil {
return nil, err
}
state.UpdatedAt = parseTimestamp(updatedAt)
states = append(states, state)
}
return states, rows.Err()
}
// SearchMessages ищет сообщения по тексту с учётом списка доступных комнат
// (AGENT.md 7.15: права проверяет вызывающий код, хранилище ограничивает выборку).
// Поиск нечёткий: FTS5 находит слова по началу («пр» → «Привет»), а подстрочные
// совпадения добираются по нормализованной копии текста без учёта регистра.
func (s *Store) SearchMessages(ctx context.Context, channelIDs []uint64, viewerID uint64, query string, limit int) ([]Message, error) {
query = strings.TrimSpace(query)
if len(channelIDs) == 0 || query == "" {
return []Message{}, nil
}
if limit <= 0 || limit > 100 {
limit = 25
}
found := make([]Message, 0, limit)
seen := map[uint64]bool{}
if ftsQuery := toFTSPrefixQuery(query); ftsQuery != "" {
prefixMatches, err := s.searchMessagesFTS(ctx, channelIDs, viewerID, ftsQuery, limit)
if err != nil {
return nil, err
}
for _, message := range prefixMatches {
if !seen[message.ID] {
seen[message.ID] = true
found = append(found, message)
}
}
}
if len(found) < limit {
substringMatches, err := s.searchMessagesSubstring(ctx, channelIDs, viewerID, strings.ToLower(query), limit-len(found))
if err != nil {
return nil, err
}
for _, message := range substringMatches {
if !seen[message.ID] {
seen[message.ID] = true
found = append(found, message)
}
}
}
return found, nil
}
// toFTSPrefixQuery превращает пользовательский ввод в безопасный запрос FTS5:
// слова соединяются по AND, последнее ищется по началу («пр» → «Привет»),
// служебные символы вырезаются.
func toFTSPrefixQuery(input string) string {
fields := strings.FieldsFunc(input, func(r rune) bool {
return r == ' ' || r == '\t' || r == '\n'
})
terms := make([]string, 0, len(fields))
for index, field := range fields {
cleaned := strings.Map(func(r rune) rune {
switch r {
case '"', '\'', '*', '(', ')', ':', '^', '-', '+':
return -1
}
return r
}, field)
if cleaned == "" {
continue
}
term := `"` + cleaned + `"`
if index == len(fields)-1 {
// Префиксный поиск по последнему слову: нечёткий ввод по мере набора.
term += "*"
}
terms = append(terms, term)
}
return strings.Join(terms, " AND ")
}
// searchMessagesFTS ищет по индексу FTS5 (слова и их начала).
func (s *Store) searchMessagesFTS(ctx context.Context, channelIDs []uint64, viewerID uint64, match string, limit int) ([]Message, error) {
placeholders := strings.TrimSuffix(strings.Repeat("?,", len(channelIDs)), ",")
args := make([]any, 0, len(channelIDs)+4)
args = append(args, match)
for _, id := range channelIDs {
args = append(args, int64(id))
}
args = append(args, int64(viewerID), int64(viewerID), limit)
rows, err := s.reader.QueryContext(ctx, `
SELECT `+prefixedMessageColumns("m")+`
FROM messages_fts f
JOIN messages m ON m.id = f.rowid
WHERE messages_fts MATCH ? AND m.channel_id IN (`+placeholders+`)
AND (`+privateVisibilityClauseFor("m")+`)
ORDER BY m.id DESC LIMIT ?`, args...)
if err != nil {
return nil, err
}
return scanMessages(rows, limit)
}
// searchMessagesSubstring ищет вхождения подстроки без учёта регистра.
func (s *Store) searchMessagesSubstring(ctx context.Context, channelIDs []uint64, viewerID uint64, query string, limit int) ([]Message, error) {
placeholders := strings.TrimSuffix(strings.Repeat("?,", len(channelIDs)), ",")
args := make([]any, 0, len(channelIDs)+4)
args = append(args, "%"+escapeLike(query)+"%")
for _, id := range channelIDs {
args = append(args, int64(id))
}
args = append(args, int64(viewerID), int64(viewerID), limit)
rows, err := s.reader.QueryContext(ctx, `
SELECT `+messageColumns+` FROM messages
WHERE content_lower LIKE ? ESCAPE '\' AND channel_id IN (`+placeholders+`)
AND (`+privateVisibilityClause+`)
ORDER BY id DESC LIMIT ?`, args...)
if err != nil {
return nil, err
}
return scanMessages(rows, limit)
}
func scanMessages(rows *sql.Rows, limit int) ([]Message, error) {
defer rows.Close()
messages := make([]Message, 0, limit)
for rows.Next() {
message, err := scanMessage(rows)
if err != nil {
return nil, err
}
messages = append(messages, *message)
}
return messages, rows.Err()
}
// privateVisibilityClauseFor — то же правило видимости для таблицы с алиасом.
func privateVisibilityClauseFor(alias string) string {
return alias + `.type <> 'private'
OR ` + alias + `.author_id = ?
OR EXISTS (SELECT 1 FROM json_each(` + alias + `.mentions_json) WHERE json_each.value = ?)`
}
// CountPinnedMessages считает закрепления комнаты (лимит 50, AGENT.md 7.6).
func (s *Store) CountPinnedMessages(ctx context.Context, channelID uint64) (int, error) {
var count int
err := s.reader.QueryRowContext(ctx,
`SELECT COUNT(*) FROM messages WHERE channel_id = ? AND pinned = 1`, int64(channelID)).Scan(&count)
return count, err
}
// CountReactions считает уникальные эмодзи сообщения (лимит 20, AGENT.md 7.6).
func (s *Store) CountReactions(ctx context.Context, messageID uint64) (int, error) {
var count int
err := s.reader.QueryRowContext(ctx,
`SELECT COUNT(DISTINCT emoji) FROM message_reactions WHERE message_id = ?`, int64(messageID)).Scan(&count)
return count, err
}
// prefixedMessageColumns добавляет префикс таблицы к списку колонок.
func prefixedMessageColumns(alias string) string {
columns := strings.Split(strings.ReplaceAll(messageColumns, "\n", " "), ",")
for i, column := range columns {
columns[i] = alias + "." + strings.TrimSpace(column)
}
return strings.Join(columns, ", ")
}
// searchBackfillItem — сообщение, которому нужно заполнить content_lower.
type searchBackfillItem struct {
id uint64
content string
}
// BackfillSearchIndex заполняет нормализованный текст у старых сообщений:
// нужен для поиска подстроки без учёта регистра (кириллица в SQLite lower()
// не приводится). Вызывается один раз при старте, порциями.
func (s *Store) BackfillSearchIndex(ctx context.Context, limit int) (int, error) {
if limit <= 0 || limit > 5000 {
limit = 500
}
rows, err := s.reader.QueryContext(ctx,
`SELECT id, content FROM messages WHERE content_lower = '' AND content <> '' LIMIT ?`, limit)
if err != nil {
return 0, err
}
items, err := collectSearchBackfill(rows, limit)
if err != nil {
return 0, err
}
updated := 0
for _, item := range items {
if _, err := s.writer.ExecContext(ctx,
`UPDATE messages SET content_lower = ? WHERE id = ?`,
strings.ToLower(item.content), int64(item.id)); err != nil {
return updated, err
}
updated++
}
return updated, nil
}
// collectSearchBackfill читает порцию сообщений и закрывает набор строк.
func collectSearchBackfill(rows *sql.Rows, limit int) ([]searchBackfillItem, error) {
defer func() { _ = rows.Close() }()
items := make([]searchBackfillItem, 0, limit)
for rows.Next() {
var item searchBackfillItem
if err := rows.Scan(&item.id, &item.content); err != nil {
return nil, err
}
items = append(items, item)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
func scanMessage(scanner interface{ Scan(...any) error }) (*Message, error) {
var (
message Message
authorID sql.NullInt64
replyToID sql.NullInt64
attachments string
mentions string
editedAt sql.NullString
pinned int
createdAt string
contentLow string
)
err := scanner.Scan(&message.ID, &message.ChannelID, &authorID, &message.Content, &replyToID,
&message.Type, &attachments, &mentions, &editedAt, &pinned, &createdAt, &contentLow)
if err != nil {
return nil, mapError(err)
}
if authorID.Valid {
value := uint64(authorID.Int64)
message.AuthorID = &value
}
if replyToID.Valid {
value := uint64(replyToID.Int64)
message.ReplyToID = &value
}
if err := json.Unmarshal([]byte(attachments), &message.Attachments); err != nil {
message.Attachments = nil
}
if err := json.Unmarshal([]byte(mentions), &message.Mentions); err != nil {
message.Mentions = nil
}
if editedAt.Valid {
value := parseTimestamp(editedAt.String)
message.EditedAt = &value
}
message.Pinned = pinned == 1
message.CreatedAt = parseTimestamp(createdAt)
return &message, nil
}
func orEmptyAttachments(values []Attachment) []Attachment {
if values == nil {
return []Attachment{}
}
return values
}
func orEmptyIDs(values []uint64) []uint64 {
if values == nil {
return []uint64{}
}
return values
}