feat(messages): сообщения, реакции, пины, typing, read states и поиск

Бэкенд текстовой связи (AGENT.md 7.6, 7.15, 7.16, 8.6): - POST/GET/PATCH/DELETE /channels/{id}/messages, ответы и упоминания, окно правки 24 часа, права автора или MANAGE_MESSAGES; - реакции (PUT/DELETE .../reactions/{emoji}) с агрегацией и отметкой «моя»; - закрепления (GET/PUT/DELETE .../pins) под MANAGE_MESSAGES; - typing с лимитом 1/3 с и адресной рассылкой всем, кроме автора; - read states: POST /channels/{id}/ack, синхронизация READ_STATE_UPDATE между устройствами пользователя; - поиск FTS5 по комнате (10/мин) с экранированием запроса; - лимиты отправки 5/5 с (burst 10) и slowmode комнаты; администратор инстанса обходит и то, и другое (AGENT.md 7.19); - Gateway: DispatchToChannel/DispatchToChannelExcept доставляют события комнат только тем, кто видит комнату (VIEW_CHANNEL), права считает общий движок; - store: messages, message_reactions, channel_read_states, files (загрузка файлов появится вместе с вложениями). Тесты: жизненный цикл сообщения, видимость скрытой комнаты (404 участнику, 200 админу), slowmode, упоминания и ответы, typing и read state, фильтрация событий комнаты между двумя WS-клиентами.
2026-09-19 23:32:11 +03:00
parent 4550bea6cc
commit e4ffbb05a5
30 changed files with 3182 additions and 106 deletions
@@ -0,0 +1,139 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"time"
+)
+
+// File — загруженный файл: метаданные в БД, содержимое на диске (AGENT.md 7.7).
+type File struct {
+	ID          uint64
+	UploaderID  *uint64
+	GuildID     *uint64
+	ChannelID   *uint64
+	MessageID   *uint64
+	Filename    string
+	ContentType string
+	SizeBytes   int64
+	Width       int
+	Height      int
+	StoragePath string
+	SHA256      string
+	CreatedAt   time.Time
+}
+
+// CreateFileParams — параметры регистрации файла.
+type CreateFileParams struct {
+	ID          uint64
+	UploaderID  uint64
+	GuildID     *uint64
+	ChannelID   *uint64
+	Filename    string
+	ContentType string
+	SizeBytes   int64
+	Width       int
+	Height      int
+	StoragePath string
+	SHA256      string
+}
+
+const fileColumns = `id, uploader_id, guild_id, channel_id, message_id, filename,
+	content_type, size_bytes, width, height, storage_path, sha256, created_at`
+
+// CreateFile регистрирует загруженный файл.
+func (s *Store) CreateFile(ctx context.Context, params CreateFileParams) (*File, error) {
+	if params.ID == 0 {
+		params.ID = s.NextID()
+	}
+	_, err := s.writer.ExecContext(ctx, `
+		INSERT INTO files (id, uploader_id, guild_id, channel_id, message_id, filename,
+			content_type, size_bytes, width, height, storage_path, sha256, created_at)
+		VALUES (?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		int64(params.ID), int64(params.UploaderID), nullableID(params.GuildID), nullableID(params.ChannelID),
+		params.Filename, params.ContentType, params.SizeBytes, params.Width, params.Height,
+		params.StoragePath, params.SHA256, s.Now())
+	if err != nil {
+		return nil, mapError(err)
+	}
+	return s.GetFile(ctx, params.ID)
+}
+
+func (s *Store) GetFile(ctx context.Context, id uint64) (*File, error) {
+	row := s.reader.QueryRowContext(ctx, `SELECT `+fileColumns+` FROM files WHERE id = ?`, int64(id))
+	return scanFile(row)
+}
+
+// AttachFileToMessage связывает файл с сообщением после отправки.
+func (s *Store) AttachFileToMessage(ctx context.Context, fileID, messageID uint64) error {
+	result, err := s.writer.ExecContext(ctx,
+		`UPDATE files SET message_id = ? WHERE id = ?`, int64(messageID), int64(fileID))
+	if err != nil {
+		return err
+	}
+	if affected, err := result.RowsAffected(); err == nil && affected == 0 {
+		return ErrNotFound
+	}
+	return nil
+}
+
+// DeleteFile удаляет запись о файле (содержимое чистит вызывающий код).
+func (s *Store) DeleteFile(ctx context.Context, id uint64) error {
+	result, err := s.writer.ExecContext(ctx, `DELETE FROM files WHERE id = ?`, int64(id))
+	if err != nil {
+		return err
+	}
+	if affected, err := result.RowsAffected(); err == nil && affected == 0 {
+		return ErrNotFound
+	}
+	return nil
+}
+
+// ListOrphanFiles возвращает файлы без сообщения старше указанного времени:
+// их удаляет обслуживание (AGENT.md 7.7).
+func (s *Store) ListOrphanFiles(ctx context.Context, olderThan time.Time, limit int) ([]File, error) {
+	if limit <= 0 || limit > 500 {
+		limit = 100
+	}
+	rows, err := s.reader.QueryContext(ctx, `
+		SELECT `+fileColumns+` FROM files
+		WHERE message_id IS NULL AND created_at < ? ORDER BY id LIMIT ?`,
+		s.Timestamp(olderThan), limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	files := make([]File, 0, limit)
+	for rows.Next() {
+		file, err := scanFile(rows)
+		if err != nil {
+			return nil, err
+		}
+		files = append(files, *file)
+	}
+	return files, rows.Err()
+}
+
+func scanFile(scanner interface{ Scan(...any) error }) (*File, error) {
+	var (
+		file       File
+		uploaderID sql.NullInt64
+		guildID    sql.NullInt64
+		channelID  sql.NullInt64
+		messageID  sql.NullInt64
+		createdAt  string
+	)
+	err := scanner.Scan(&file.ID, &uploaderID, &guildID, &channelID, &messageID, &file.Filename,
+		&file.ContentType, &file.SizeBytes, &file.Width, &file.Height, &file.StoragePath,
+		&file.SHA256, &createdAt)
+	if err != nil {
+		return nil, mapError(err)
+	}
+	file.UploaderID = optionalID(uploaderID)
+	file.GuildID = optionalID(guildID)
+	file.ChannelID = optionalID(channelID)
+	file.MessageID = optionalID(messageID)
+	file.CreatedAt = parseTimestamp(createdAt)
+	return &file, nil
+}
@@ -0,0 +1,401 @@
+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"
+)
+
+// 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`
+
+// 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
+	}
+	_, 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)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?)`,
+		int64(params.ID), int64(params.ChannelID), int64(params.AuthorID), params.Content,
+		nullableID(params.ReplyToID), string(params.Type), string(attachments), string(mentions), s.Now())
+	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
+// используется для подгрузки истории вверх (AGENT.md 7.6).
+func (s *Store) ListMessages(ctx context.Context, channelID, beforeID 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 += ` 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()
+}
+
+// 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 = ?, edited_at = ? WHERE id = ?`, 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
+}
+
+// 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: права проверяет вызывающий код, хранилище ограничивает выборку).
+func (s *Store) SearchMessages(ctx context.Context, channelIDs []uint64, query string, limit int) ([]Message, error) {
+	if len(channelIDs) == 0 || strings.TrimSpace(query) == "" {
+		return []Message{}, nil
+	}
+	if limit <= 0 || limit > 100 {
+		limit = 25
+	}
+	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(channelIDs)), ",")
+	args := make([]any, 0, len(channelIDs)+2)
+	args = append(args, query)
+	for _, id := range channelIDs {
+		args = append(args, int64(id))
+	}
+	args = append(args, 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+`)
+		ORDER BY m.id DESC LIMIT ?`, 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()
+}
+
+// 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, ", ")
+}
+
+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
+	)
+	err := scanner.Scan(&message.ID, &message.ChannelID, &authorID, &message.Content, &replyToID,
+		&message.Type, &attachments, &mentions, &editedAt, &pinned, &createdAt)
+	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
+}