grendervill
8619f37cd5
- упоминания в сообщении увеличивают счётчик упоминаний комнаты у упомянутых (store.BumpMentionCount) и отправляют им READ_STATE_UPDATE на все устройства; - отправка сообщения отмечает его прочитанным для автора; - подтверждение прочтения сбрасывает упоминания; - суточная квота 10 приглашений на пользователя (AGENT.md 8.6), администратор инстанса её обходит; httpx.NewRateLimiterWindow для произвольных окон; - тест на рост и сброс счётчика упоминаний.
420 lines
13 KiB
Go
420 lines
13 KiB
Go
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
|
|
}
|
|
|
|
// 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: права проверяет вызывающий код, хранилище ограничивает выборку).
|
|
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
|
|
}
|