Sprint 3: chat with WebSocket
- POST /api/v1/chats (ensure chat between users, normalized pair) - GET /api/v1/chats (list with last message + unread) - GET /api/v1/chats/:id/messages - POST /api/v1/chats/:id/messages (send text/photo, push via WS) - PUT /api/v1/chats/:id/read (mark all read) - PUT /api/v1/messages/:msgId (edit, owner only) - DELETE /api/v1/messages/:msgId (soft delete) - POST /api/v1/blocks, DELETE /api/v1/blocks/:id - POST /api/v1/reports - WS /ws?token= (real-time message push, ping/pong) - migrations 0003: chats, messages, blocks, reports
This commit is contained in:
ga
parent
d4997e85ef
commit
4533ceaf2d
@ -16,6 +16,7 @@ import (
|
||||
|
||||
"github.com/buhapp/backend/internal/audit"
|
||||
"github.com/buhapp/backend/internal/auth"
|
||||
"github.com/buhapp/backend/internal/chat"
|
||||
"github.com/buhapp/backend/internal/config"
|
||||
"github.com/buhapp/backend/internal/consent"
|
||||
"github.com/buhapp/backend/internal/db"
|
||||
@ -25,6 +26,7 @@ import (
|
||||
"github.com/buhapp/backend/internal/redis"
|
||||
"github.com/buhapp/backend/internal/storage"
|
||||
"github.com/buhapp/backend/internal/users"
|
||||
"github.com/buhapp/backend/internal/ws"
|
||||
)
|
||||
|
||||
func main() {
|
||||
@ -64,20 +66,22 @@ func main() {
|
||||
log.Println("minio: connected")
|
||||
}
|
||||
|
||||
// Repos
|
||||
usersRepo := users.NewRepo(pg.Pool)
|
||||
consentRepo := consent.NewRepo(pg.Pool)
|
||||
auditRepo := audit.NewRepo(pg.Pool)
|
||||
prefsRepo := preferences.NewRepo(pg.Pool)
|
||||
locsRepo := locations.NewRepo(pg.Pool)
|
||||
chatRepo := chat.NewRepo(pg.Pool)
|
||||
|
||||
authSvc := auth.NewService(cfg.JWTSecret, cfg.JwTAccessTTL(), cfg.JwtRefreshTTL())
|
||||
_ = authSvc
|
||||
_ = rdb
|
||||
_ = st
|
||||
hub := ws.NewHub()
|
||||
|
||||
authH := handlers.NewAuthHandler(cfg, usersRepo, consentRepo, auditRepo, authSvc)
|
||||
h := handlers.New(cfg, usersRepo, prefsRepo, locsRepo, consentRepo, auditRepo, authSvc)
|
||||
chatH := handlers.NewChatHandlers(chatRepo, auditRepo, hub)
|
||||
|
||||
_ = rdb
|
||||
_ = st
|
||||
|
||||
app := fiber.New(fiber.Config{
|
||||
AppName: "buhapp-api",
|
||||
@ -96,7 +100,7 @@ func main() {
|
||||
return c.JSON(fiber.Map{
|
||||
"status": "ok",
|
||||
"time": time.Now().UTC().Format(time.RFC3339),
|
||||
"version": "0.2.0",
|
||||
"version": "0.3.0",
|
||||
"db": pg.Pool != nil,
|
||||
"redis": rdb != nil,
|
||||
"storage": st != nil,
|
||||
@ -120,16 +124,31 @@ func main() {
|
||||
protected := api.Group("", auth.Middleware(authSvc))
|
||||
protected.Get("/me", h.GetMe)
|
||||
protected.Put("/me", h.UpdateMe)
|
||||
|
||||
protected.Get("/me/prefs", h.GetPrefs)
|
||||
protected.Put("/me/prefs", h.UpdatePrefs)
|
||||
|
||||
protected.Put("/me/location", h.UpdateLocation)
|
||||
protected.Put("/me/visibility", h.SetVisibility)
|
||||
|
||||
protected.Get("/search/nearby", h.SearchNearby)
|
||||
protected.Get("/users/:id", h.GetUserPublic)
|
||||
|
||||
// Chat
|
||||
protected.Post("/chats", chatH.EnsureChat)
|
||||
protected.Get("/chats", chatH.ListChats)
|
||||
protected.Get("/chats/:id/messages", chatH.ListMessages)
|
||||
protected.Post("/chats/:id/messages", chatH.SendMessage)
|
||||
protected.Put("/chats/:id/read", chatH.MarkRead)
|
||||
|
||||
protected.Put("/messages/:msgId", chatH.EditMessage)
|
||||
protected.Delete("/messages/:msgId", chatH.DeleteMessage)
|
||||
|
||||
protected.Post("/blocks", chatH.Block)
|
||||
protected.Delete("/blocks/:id", chatH.Unblock)
|
||||
|
||||
protected.Post("/reports", chatH.Report)
|
||||
|
||||
// WebSocket — авторизация по ?token=
|
||||
app.Get("/ws", ws.AuthMiddleware(authSvc), ws.Upgrade, ws.Handler(hub))
|
||||
|
||||
go func() {
|
||||
sigCh := make(chan os.Signal, 1)
|
||||
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
|
||||
|
||||
1
go.mod
1
go.mod
@ -5,6 +5,7 @@ go 1.22
|
||||
require (
|
||||
github.com/gofiber/fiber/v2 v2.52.5
|
||||
github.com/gofiber/contrib/jwt v1.0.10
|
||||
github.com/gofiber/contrib/websocket v1.3.0
|
||||
github.com/golang-jwt/jwt/v5 v5.2.1
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/jackc/pgx/v5 v5.7.1
|
||||
|
||||
263
internal/chat/repo.go
Normal file
263
internal/chat/repo.go
Normal file
@ -0,0 +1,263 @@
|
||||
package chat
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
type Chat struct {
|
||||
ID uuid.UUID
|
||||
UserA uuid.UUID
|
||||
UserB uuid.UUID
|
||||
CreatedAt time.Time
|
||||
LastMsgAt time.Time
|
||||
}
|
||||
|
||||
type Message struct {
|
||||
ID uuid.UUID
|
||||
ChatID uuid.UUID
|
||||
SenderID uuid.UUID
|
||||
Body string
|
||||
PhotoURL string
|
||||
Read bool
|
||||
Edited bool
|
||||
Deleted bool
|
||||
CreatedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
type Repo struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewRepo(pool *pgxpool.Pool) *Repo {
|
||||
return &Repo{pool: pool}
|
||||
}
|
||||
|
||||
// EnsureChat — нормализуем user_a < user_b, чтобы избежать дублей
|
||||
func (r *Repo) EnsureChat(ctx context.Context, me, other uuid.UUID) (*Chat, error) {
|
||||
a, b := me, other
|
||||
if a.String() > b.String() {
|
||||
a, b = b, a
|
||||
}
|
||||
// ищем существующий
|
||||
c := &Chat{}
|
||||
err := r.pool.QueryRow(ctx, `
|
||||
SELECT id, user_a, user_b, created_at, last_msg_at
|
||||
FROM chats WHERE user_a=$1 AND user_b=$2`, a, b,
|
||||
).Scan(&c.ID, &c.UserA, &c.UserB, &c.CreatedAt, &c.LastMsgAt)
|
||||
if err == nil {
|
||||
return c, nil
|
||||
}
|
||||
if !errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, err
|
||||
}
|
||||
// создаём
|
||||
err = r.pool.QueryRow(ctx, `
|
||||
INSERT INTO chats (user_a, user_b) VALUES ($1, $2)
|
||||
RETURNING id, created_at, last_msg_at`,
|
||||
a, b,
|
||||
).Scan(&c.ID, &c.CreatedAt, &c.LastMsgAt)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.UserA, c.UserB = a, b
|
||||
return c, nil
|
||||
}
|
||||
|
||||
func (r *Repo) GetByID(ctx context.Context, id uuid.UUID) (*Chat, error) {
|
||||
c := &Chat{}
|
||||
err := r.pool.QueryRow(ctx, `
|
||||
SELECT id, user_a, user_b, created_at, last_msg_at
|
||||
FROM chats WHERE id=$1`, id,
|
||||
).Scan(&c.ID, &c.UserA, &c.UserB, &c.CreatedAt, &c.LastMsgAt)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// ListChats — список чатов пользователя с инфой по последнему сообщению
|
||||
type ChatListItem struct {
|
||||
Chat Chat
|
||||
OtherID uuid.UUID
|
||||
OtherName string
|
||||
OtherPhoto string
|
||||
LastMessage string
|
||||
LastMsgAt time.Time
|
||||
UnreadCount int
|
||||
}
|
||||
|
||||
func (r *Repo) ListChats(ctx context.Context, me uuid.UUID, limit, offset int) ([]ChatListItem, error) {
|
||||
if limit <= 0 || limit > 200 {
|
||||
limit = 50
|
||||
}
|
||||
rows, err := r.pool.Query(ctx, `
|
||||
SELECT c.id, c.user_a, c.user_b, c.created_at, c.last_msg_at,
|
||||
CASE WHEN c.user_a = $1 THEN c.user_b ELSE c.user_a END AS other_id,
|
||||
u.name,
|
||||
u.photo_url,
|
||||
(SELECT COALESCE(body, '') FROM messages m
|
||||
WHERE m.chat_id = c.id AND m.deleted = FALSE
|
||||
ORDER BY m.created_at DESC LIMIT 1) AS last_body,
|
||||
(SELECT COUNT(*) FROM messages m
|
||||
WHERE m.chat_id = c.id AND m.read = FALSE
|
||||
AND m.sender_id != $1) AS unread
|
||||
FROM chats c
|
||||
JOIN users u ON u.id = CASE WHEN c.user_a = $1 THEN c.user_b ELSE c.user_a END
|
||||
WHERE c.user_a = $1 OR c.user_b = $1
|
||||
ORDER BY c.last_msg_at DESC
|
||||
LIMIT $2 OFFSET $3`,
|
||||
me, limit, offset,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []ChatListItem
|
||||
for rows.Next() {
|
||||
var c ChatListItem
|
||||
var photo *string
|
||||
if err := rows.Scan(&c.Chat.ID, &c.Chat.UserA, &c.Chat.UserB, &c.Chat.CreatedAt, &c.Chat.LastMsgAt,
|
||||
&c.OtherID, &c.OtherName, &photo, &c.LastMessage, &c.UnreadCount); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if photo != nil {
|
||||
c.OtherPhoto = *photo
|
||||
}
|
||||
c.LastMsgAt = c.Chat.LastMsgAt
|
||||
out = append(out, c)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (r *Repo) Send(ctx context.Context, chatID, senderID uuid.UUID, body, photoURL string) (*Message, error) {
|
||||
m := &Message{}
|
||||
err := r.pool.QueryRow(ctx, `
|
||||
INSERT INTO messages (chat_id, sender_id, body, photo_url)
|
||||
VALUES ($1, $2, NULLIF($3, ''), NULLIF($4, ''))
|
||||
RETURNING id, chat_id, sender_id, COALESCE(body, ''), COALESCE(photo_url, ''),
|
||||
read, edited, deleted, created_at, updated_at`,
|
||||
chatID, senderID, body, photoURL,
|
||||
).Scan(&m.ID, &m.ChatID, &m.SenderID, &m.Body, &m.PhotoURL,
|
||||
&m.Read, &m.Edited, &m.Deleted, &m.CreatedAt, &m.UpdatedAt)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, _ = r.pool.Exec(ctx, `UPDATE chats SET last_msg_at = NOW() WHERE id=$1`, chatID)
|
||||
return m, nil
|
||||
}
|
||||
|
||||
func (r *Repo) ListMessages(ctx context.Context, chatID uuid.UUID, limit, offset int) ([]Message, error) {
|
||||
if limit <= 0 || limit > 200 {
|
||||
limit = 50
|
||||
}
|
||||
rows, err := r.pool.Query(ctx, `
|
||||
SELECT id, chat_id, sender_id, COALESCE(body, ''), COALESCE(photo_url, ''),
|
||||
read, edited, deleted, created_at, updated_at
|
||||
FROM messages
|
||||
WHERE chat_id=$1
|
||||
ORDER BY created_at ASC
|
||||
LIMIT $2 OFFSET $3`, chatID, limit, offset)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []Message
|
||||
for rows.Next() {
|
||||
var m Message
|
||||
if err := rows.Scan(&m.ID, &m.ChatID, &m.SenderID, &m.Body, &m.PhotoURL,
|
||||
&m.Read, &m.Edited, &m.Deleted, &m.CreatedAt, &m.UpdatedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, m)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (r *Repo) EditMessage(ctx context.Context, msgID, senderID uuid.UUID, newBody string) error {
|
||||
res, err := r.pool.Exec(ctx, `
|
||||
UPDATE messages SET body=$1, edited=TRUE, updated_at=NOW()
|
||||
WHERE id=$2 AND sender_id=$3 AND deleted=FALSE`,
|
||||
newBody, msgID, senderID,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if res.RowsAffected() == 0 {
|
||||
return errors.New("message not found or not yours")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Repo) DeleteMessage(ctx context.Context, msgID, senderID uuid.UUID) error {
|
||||
res, err := r.pool.Exec(ctx, `
|
||||
UPDATE messages SET deleted=TRUE, body=NULL, photo_url=NULL, updated_at=NOW()
|
||||
WHERE id=$1 AND sender_id=$2`,
|
||||
msgID, senderID,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if res.RowsAffected() == 0 {
|
||||
return errors.New("message not found or not yours")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Repo) MarkRead(ctx context.Context, chatID, meID uuid.UUID) error {
|
||||
_, err := r.pool.Exec(ctx, `
|
||||
UPDATE messages SET read=TRUE
|
||||
WHERE chat_id=$1 AND sender_id != $2 AND read=FALSE`,
|
||||
chatID, meID,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// Block / Unblock
|
||||
func (r *Repo) Block(ctx context.Context, me, other uuid.UUID) error {
|
||||
_, err := r.pool.Exec(ctx, `
|
||||
INSERT INTO blocks (blocker_id, blocked_id) VALUES ($1, $2)
|
||||
ON CONFLICT DO NOTHING`, me, other)
|
||||
return err
|
||||
}
|
||||
|
||||
func (r *Repo) Unblock(ctx context.Context, me, other uuid.UUID) error {
|
||||
_, err := r.pool.Exec(ctx, `DELETE FROM blocks WHERE blocker_id=$1 AND blocked_id=$2`, me, other)
|
||||
return err
|
||||
}
|
||||
|
||||
func (r *Repo) IsBlockedEither(ctx context.Context, a, b uuid.UUID) (bool, error) {
|
||||
var blocked bool
|
||||
err := r.pool.QueryRow(ctx, `
|
||||
SELECT EXISTS (
|
||||
SELECT 1 FROM blocks WHERE
|
||||
(blocker_id=$1 AND blocked_id=$2) OR (blocker_id=$2 AND blocked_id=$1)
|
||||
)`, a, b).Scan(&blocked)
|
||||
return blocked, err
|
||||
}
|
||||
|
||||
// Report
|
||||
type Report struct {
|
||||
ReporterID uuid.UUID
|
||||
TargetType string
|
||||
TargetID string
|
||||
Reason string
|
||||
}
|
||||
|
||||
func (r *Repo) AddReport(ctx context.Context, rep *Report) error {
|
||||
_, err := r.pool.Exec(ctx, `
|
||||
INSERT INTO reports (reporter_id, target_type, target_id, reason)
|
||||
VALUES ($1, $2, $3, NULLIF($4, ''))`,
|
||||
rep.ReporterID, rep.TargetType, rep.TargetID, rep.Reason)
|
||||
return err
|
||||
}
|
||||
329
internal/handlers/chat.go
Normal file
329
internal/handlers/chat.go
Normal file
@ -0,0 +1,329 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gofiber/fiber/v2"
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/buhapp/backend/internal/audit"
|
||||
"github.com/buhapp/backend/internal/chat"
|
||||
"github.com/buhapp/backend/internal/ws"
|
||||
)
|
||||
|
||||
type ChatHandlers struct {
|
||||
Chat *chat.Repo
|
||||
Audit *audit.Repo
|
||||
Hub *ws.Hub
|
||||
}
|
||||
|
||||
func NewChatHandlers(c *chat.Repo, a *audit.Repo, hub *ws.Hub) *ChatHandlers {
|
||||
return &ChatHandlers{Chat: c, Audit: a, Hub: hub}
|
||||
}
|
||||
|
||||
type ensureChatRequest struct {
|
||||
OtherID string `json:"other_id"`
|
||||
}
|
||||
|
||||
func (h *ChatHandlers) EnsureChat(c *fiber.Ctx) error {
|
||||
me, err := userID(c)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "no user"})
|
||||
}
|
||||
var req ensureChatRequest
|
||||
if err := c.BodyParser(&req); err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "invalid JSON"})
|
||||
}
|
||||
other, err := uuid.Parse(req.OtherID)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "bad other_id"})
|
||||
}
|
||||
if me == other {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "cannot chat with yourself"})
|
||||
}
|
||||
blocked, err := h.Chat.IsBlockedEither(c.UserContext(), me, other)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "db error"})
|
||||
}
|
||||
if blocked {
|
||||
return c.Status(fiber.StatusForbidden).JSON(fiber.Map{"error": "user blocked"})
|
||||
}
|
||||
chatObj, err := h.Chat.EnsureChat(c.UserContext(), me, other)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "db error"})
|
||||
}
|
||||
return c.JSON(chatObj)
|
||||
}
|
||||
|
||||
func (h *ChatHandlers) ListChats(c *fiber.Ctx) error {
|
||||
me, err := userID(c)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "no user"})
|
||||
}
|
||||
limit := c.QueryInt("limit", 50)
|
||||
offset := c.QueryInt("offset", 0)
|
||||
items, err := h.Chat.ListChats(c.UserContext(), me, limit, offset)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "db error"})
|
||||
}
|
||||
// mark online
|
||||
for i := range items {
|
||||
items[i].OtherID = items[i].OtherID // already set
|
||||
// online info
|
||||
}
|
||||
return c.JSON(fiber.Map{
|
||||
"chats": items,
|
||||
"count": len(items),
|
||||
})
|
||||
}
|
||||
|
||||
type sendMessageRequest struct {
|
||||
Body string `json:"body"`
|
||||
PhotoURL string `json:"photo_url"`
|
||||
}
|
||||
|
||||
func (h *ChatHandlers) SendMessage(c *fiber.Ctx) error {
|
||||
me, err := userID(c)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "no user"})
|
||||
}
|
||||
chatID, err := uuid.Parse(c.Params("id"))
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "bad chat id"})
|
||||
}
|
||||
var req sendMessageRequest
|
||||
if err := c.BodyParser(&req); err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "invalid JSON"})
|
||||
}
|
||||
body := strings.TrimSpace(req.Body)
|
||||
photo := strings.TrimSpace(req.PhotoURL)
|
||||
if body == "" && photo == "" {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "body or photo_url required"})
|
||||
}
|
||||
if len(body) > 4000 {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "body too long (max 4000)"})
|
||||
}
|
||||
|
||||
// Проверим что я участник чата
|
||||
ch, err := h.Chat.GetByID(c.UserContext(), chatID)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "db error"})
|
||||
}
|
||||
if ch == nil {
|
||||
return c.Status(fiber.StatusNotFound).JSON(fiber.Map{"error": "chat not found"})
|
||||
}
|
||||
if ch.UserA != me && ch.UserB != me {
|
||||
return c.Status(fiber.StatusForbidden).JSON(fiber.Map{"error": "not a chat member"})
|
||||
}
|
||||
|
||||
// Проверим блок
|
||||
other := ch.UserA
|
||||
if other == me {
|
||||
other = ch.UserB
|
||||
}
|
||||
blocked, _ := h.Chat.IsBlockedEither(c.UserContext(), me, other)
|
||||
if blocked {
|
||||
return c.Status(fiber.StatusForbidden).JSON(fiber.Map{"error": "user blocked"})
|
||||
}
|
||||
|
||||
m, err := h.Chat.Send(c.UserContext(), chatID, me, body, photo)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "db error"})
|
||||
}
|
||||
|
||||
_ = h.Audit.Log(c.UserContext(), &audit.Event{
|
||||
UserID: &me, Action: "message.send",
|
||||
TargetType: "chat", TargetID: chatID.String(),
|
||||
IP: c.IP(), UserAgent: c.Get("User-Agent"),
|
||||
})
|
||||
|
||||
// push to recipient via WebSocket
|
||||
h.Hub.SendTo(other, ws.Outgoing{
|
||||
Type: "message",
|
||||
Payload: fiber.Map{
|
||||
"id": m.ID,
|
||||
"chat_id": m.ChatID,
|
||||
"sender_id": m.SenderID,
|
||||
"body": m.Body,
|
||||
"photo_url": m.PhotoURL,
|
||||
"created_at": m.CreatedAt,
|
||||
},
|
||||
Time: time.Now(),
|
||||
})
|
||||
// echo to sender too
|
||||
h.Hub.SendTo(me, ws.Outgoing{
|
||||
Type: "message",
|
||||
Payload: fiber.Map{
|
||||
"id": m.ID,
|
||||
"chat_id": m.ChatID,
|
||||
"sender_id": m.SenderID,
|
||||
"body": m.Body,
|
||||
"photo_url": m.PhotoURL,
|
||||
"created_at": m.CreatedAt,
|
||||
},
|
||||
Time: time.Now(),
|
||||
})
|
||||
|
||||
return c.Status(fiber.StatusCreated).JSON(m)
|
||||
}
|
||||
|
||||
func (h *ChatHandlers) ListMessages(c *fiber.Ctx) error {
|
||||
me, err := userID(c)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "no user"})
|
||||
}
|
||||
chatID, err := uuid.Parse(c.Params("id"))
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "bad chat id"})
|
||||
}
|
||||
ch, err := h.Chat.GetByID(c.UserContext(), chatID)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "db error"})
|
||||
}
|
||||
if ch == nil {
|
||||
return c.Status(fiber.StatusNotFound).JSON(fiber.Map{"error": "chat not found"})
|
||||
}
|
||||
if ch.UserA != me && ch.UserB != me {
|
||||
return c.Status(fiber.StatusForbidden).JSON(fiber.Map{"error": "not a chat member"})
|
||||
}
|
||||
limit := c.QueryInt("limit", 50)
|
||||
offset := c.QueryInt("offset", 0)
|
||||
msgs, err := h.Chat.ListMessages(c.UserContext(), chatID, limit, offset)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "db error"})
|
||||
}
|
||||
return c.JSON(fiber.Map{"messages": msgs, "count": len(msgs)})
|
||||
}
|
||||
|
||||
type editRequest struct {
|
||||
Body string `json:"body"`
|
||||
}
|
||||
|
||||
func (h *ChatHandlers) EditMessage(c *fiber.Ctx) error {
|
||||
me, err := userID(c)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "no user"})
|
||||
}
|
||||
msgID, err := uuid.Parse(c.Params("msgId"))
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "bad msg id"})
|
||||
}
|
||||
var req editRequest
|
||||
if err := c.BodyParser(&req); err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "invalid JSON"})
|
||||
}
|
||||
body := strings.TrimSpace(req.Body)
|
||||
if body == "" || len(body) > 4000 {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "body required, max 4000"})
|
||||
}
|
||||
if err := h.Chat.EditMessage(c.UserContext(), msgID, me, body); err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": err.Error()})
|
||||
}
|
||||
return c.JSON(fiber.Map{"ok": true})
|
||||
}
|
||||
|
||||
func (h *ChatHandlers) DeleteMessage(c *fiber.Ctx) error {
|
||||
me, err := userID(c)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "no user"})
|
||||
}
|
||||
msgID, err := uuid.Parse(c.Params("msgId"))
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "bad msg id"})
|
||||
}
|
||||
if err := h.Chat.DeleteMessage(c.UserContext(), msgID, me); err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": err.Error()})
|
||||
}
|
||||
return c.JSON(fiber.Map{"ok": true})
|
||||
}
|
||||
|
||||
func (h *ChatHandlers) MarkRead(c *fiber.Ctx) error {
|
||||
me, err := userID(c)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "no user"})
|
||||
}
|
||||
chatID, err := uuid.Parse(c.Params("id"))
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "bad chat id"})
|
||||
}
|
||||
if err := h.Chat.MarkRead(c.UserContext(), chatID, me); err != nil {
|
||||
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "db error"})
|
||||
}
|
||||
return c.JSON(fiber.Map{"ok": true})
|
||||
}
|
||||
|
||||
type blockRequest struct {
|
||||
UserID string `json:"user_id"`
|
||||
}
|
||||
|
||||
func (h *ChatHandlers) Block(c *fiber.Ctx) error {
|
||||
me, err := userID(c)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "no user"})
|
||||
}
|
||||
var req blockRequest
|
||||
if err := c.BodyParser(&req); err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "invalid JSON"})
|
||||
}
|
||||
other, err := uuid.Parse(req.UserID)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "bad user_id"})
|
||||
}
|
||||
if err := h.Chat.Block(c.UserContext(), me, other); err != nil {
|
||||
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "db error"})
|
||||
}
|
||||
_ = h.Audit.Log(c.UserContext(), &audit.Event{
|
||||
UserID: &me, Action: "user.block", TargetType: "user", TargetID: other.String(),
|
||||
IP: c.IP(), UserAgent: c.Get("User-Agent"),
|
||||
})
|
||||
return c.JSON(fiber.Map{"ok": true})
|
||||
}
|
||||
|
||||
func (h *ChatHandlers) Unblock(c *fiber.Ctx) error {
|
||||
me, err := userID(c)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "no user"})
|
||||
}
|
||||
other, err := uuid.Parse(c.Params("id"))
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "bad id"})
|
||||
}
|
||||
if err := h.Chat.Unblock(c.UserContext(), me, other); err != nil {
|
||||
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "db error"})
|
||||
}
|
||||
return c.JSON(fiber.Map{"ok": true})
|
||||
}
|
||||
|
||||
type reportRequest struct {
|
||||
TargetType string `json:"target_type"`
|
||||
TargetID string `json:"target_id"`
|
||||
Reason string `json:"reason"`
|
||||
}
|
||||
|
||||
func (h *ChatHandlers) Report(c *fiber.Ctx) error {
|
||||
me, err := userID(c)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "no user"})
|
||||
}
|
||||
var req reportRequest
|
||||
if err := c.BodyParser(&req); err != nil {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "invalid JSON"})
|
||||
}
|
||||
if req.TargetType != "user" && req.TargetType != "message" && req.TargetType != "chat" {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "bad target_type"})
|
||||
}
|
||||
if req.TargetID == "" {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "target_id required"})
|
||||
}
|
||||
if err := h.Chat.AddReport(c.UserContext(), &chat.Report{
|
||||
ReporterID: me, TargetType: req.TargetType, TargetID: req.TargetID, Reason: req.Reason,
|
||||
}); err != nil {
|
||||
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "db error"})
|
||||
}
|
||||
_ = h.Audit.Log(c.UserContext(), &audit.Event{
|
||||
UserID: &me, Action: "report.create", TargetType: req.TargetType, TargetID: req.TargetID,
|
||||
IP: c.IP(), UserAgent: c.Get("User-Agent"),
|
||||
})
|
||||
return c.JSON(fiber.Map{"ok": true})
|
||||
}
|
||||
55
internal/ws/handler.go
Normal file
55
internal/ws/handler.go
Normal file
@ -0,0 +1,55 @@
|
||||
package ws
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/gofiber/contrib/websocket"
|
||||
"github.com/gofiber/fiber/v2"
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/buhapp/backend/internal/auth"
|
||||
)
|
||||
|
||||
// AuthMiddleware — авторизация по ?token=... (для WS)
|
||||
func AuthMiddleware(svc *auth.Service) fiber.Handler {
|
||||
return func(c *fiber.Ctx) error {
|
||||
tok := c.Query("token")
|
||||
if tok == "" {
|
||||
h := c.Get("Authorization")
|
||||
if strings.HasPrefix(h, "Bearer ") {
|
||||
tok = strings.TrimPrefix(h, "Bearer ")
|
||||
}
|
||||
}
|
||||
if tok == "" {
|
||||
return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "token required"})
|
||||
}
|
||||
claims, err := svc.Parse(c.UserContext(), tok)
|
||||
if err != nil {
|
||||
return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "invalid token"})
|
||||
}
|
||||
c.Locals("user_id", claims.UserID)
|
||||
return c.Next()
|
||||
}
|
||||
}
|
||||
|
||||
// Upgrade — проверка что соединение upgradeable
|
||||
func Upgrade(c *fiber.Ctx) error {
|
||||
if websocket.IsWebSocketUpgrade(c) {
|
||||
c.Locals("allowed", true)
|
||||
return c.Next()
|
||||
}
|
||||
return fiber.ErrUpgradeRequired
|
||||
}
|
||||
|
||||
// Handler — отдаёт hub'у новое WS-соединение
|
||||
func Handler(hub *Hub) fiber.Handler {
|
||||
return websocket.New(func(c *websocket.Conn) {
|
||||
v := c.Locals("user_id")
|
||||
uid, ok := v.(uuid.UUID)
|
||||
if !ok {
|
||||
_ = c.Close()
|
||||
return
|
||||
}
|
||||
hub.Handle(uid, c)
|
||||
})
|
||||
}
|
||||
112
internal/ws/hub.go
Normal file
112
internal/ws/hub.go
Normal file
@ -0,0 +1,112 @@
|
||||
package ws
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gofiber/contrib/websocket"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// Hub — WebSocket connection hub
|
||||
type Hub struct {
|
||||
mu sync.RWMutex
|
||||
conns map[uuid.UUID]map[*websocket.Conn]struct{} // user_id -> set of conns
|
||||
}
|
||||
|
||||
func NewHub() *Hub {
|
||||
return &Hub{conns: make(map[uuid.UUID]map[*websocket.Conn]struct{})}
|
||||
}
|
||||
|
||||
func (h *Hub) add(userID uuid.UUID, c *websocket.Conn) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if h.conns[userID] == nil {
|
||||
h.conns[userID] = make(map[*websocket.Conn]struct{})
|
||||
}
|
||||
h.conns[userID][c] = struct{}{}
|
||||
}
|
||||
|
||||
func (h *Hub) remove(userID uuid.UUID, c *websocket.Conn) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if set, ok := h.conns[userID]; ok {
|
||||
delete(set, c)
|
||||
if len(set) == 0 {
|
||||
delete(h.conns, userID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Hub) IsOnline(userID uuid.UUID) bool {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
_, ok := h.conns[userID]
|
||||
return ok
|
||||
}
|
||||
|
||||
// SendTo — отправить сообщение всем коннектам пользователя
|
||||
func (h *Hub) SendTo(userID uuid.UUID, payload any) {
|
||||
data, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
log.Printf("ws: marshal: %v", err)
|
||||
return
|
||||
}
|
||||
h.mu.RLock()
|
||||
conns := make([]*websocket.Conn, 0, 4)
|
||||
for c := range h.conns[userID] {
|
||||
conns = append(conns, c)
|
||||
}
|
||||
h.mu.RUnlock()
|
||||
for _, c := range conns {
|
||||
if err := c.WriteMessage(websocket.TextMessage, data); err != nil {
|
||||
log.Printf("ws: write: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ClientMessage — входящее от клиента
|
||||
type ClientMessage struct {
|
||||
Type string `json:"type"` // "ping", "typing", "read"
|
||||
Payload json.RawMessage `json:"payload,omitempty"`
|
||||
}
|
||||
|
||||
type Outgoing struct {
|
||||
Type string `json:"type"`
|
||||
Payload any `json:"payload"`
|
||||
Time time.Time `json:"time"`
|
||||
}
|
||||
|
||||
// Handle — обработчик WebSocket-соединения
|
||||
func (h *Hub) Handle(userID uuid.UUID, c *websocket.Conn) {
|
||||
h.add(userID, c)
|
||||
defer func() {
|
||||
h.remove(userID, c)
|
||||
_ = c.Close()
|
||||
}()
|
||||
|
||||
// уведомим контактов что онлайн? для MVP — пропустим
|
||||
_ = h
|
||||
|
||||
// отправим pong при подключении
|
||||
_ = c.WriteJSON(Outgoing{Type: "connected", Payload: map[string]any{"user_id": userID}, Time: time.Now()})
|
||||
|
||||
for {
|
||||
_, msg, err := c.ReadMessage()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
var cm ClientMessage
|
||||
if err := json.Unmarshal(msg, &cm); err != nil {
|
||||
continue
|
||||
}
|
||||
switch cm.Type {
|
||||
case "ping":
|
||||
_ = c.WriteJSON(Outgoing{Type: "pong", Time: time.Now()})
|
||||
case "typing":
|
||||
// можно broadcastить — но для MVP отправим обратно клиенту для подтверждения
|
||||
}
|
||||
}
|
||||
}
|
||||
4
migrations/0003_chat.down.sql
Normal file
4
migrations/0003_chat.down.sql
Normal file
@ -0,0 +1,4 @@
|
||||
DROP TABLE IF EXISTS reports;
|
||||
DROP TABLE IF EXISTS blocks;
|
||||
DROP TABLE IF EXISTS messages;
|
||||
DROP TABLE IF EXISTS chats;
|
||||
49
migrations/0003_chat.up.sql
Normal file
49
migrations/0003_chat.up.sql
Normal file
@ -0,0 +1,49 @@
|
||||
-- Чаты и сообщения
|
||||
CREATE TABLE IF NOT EXISTS chats (
|
||||
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
|
||||
user_a UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
user_b UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
created_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
last_msg_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
UNIQUE (user_a, user_b),
|
||||
CHECK (user_a < user_b)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_chats_user_a ON chats(user_a, last_msg_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS idx_chats_user_b ON chats(user_b, last_msg_at DESC);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS messages (
|
||||
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
|
||||
chat_id UUID NOT NULL REFERENCES chats(id) ON DELETE CASCADE,
|
||||
sender_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
body TEXT,
|
||||
photo_url TEXT,
|
||||
read BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
edited BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
deleted BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
created_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
updated_at TIMESTAMPTZ DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_messages_chat ON messages(chat_id, created_at);
|
||||
|
||||
-- Блокировки
|
||||
CREATE TABLE IF NOT EXISTS blocks (
|
||||
blocker_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
blocked_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
created_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
PRIMARY KEY (blocker_id, blocked_id)
|
||||
);
|
||||
|
||||
-- Жалобы
|
||||
CREATE TABLE IF NOT EXISTS reports (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
reporter_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
target_type TEXT NOT NULL CHECK (target_type IN ('user','message','chat')),
|
||||
target_id TEXT NOT NULL,
|
||||
reason TEXT,
|
||||
metadata JSONB,
|
||||
created_at TIMESTAMPTZ DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_reports_target ON reports(target_type, target_id);
|
||||
Loading…
Reference in New Issue
Block a user