diff --git a/cmd/server/main.go b/cmd/server/main.go index f7a9ec7..89da1fe 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -4,7 +4,7 @@ import ( "context" "fmt" "io" - "log" + "log/slog" "os" "os/signal" "syscall" @@ -12,8 +12,8 @@ import ( "github.com/gofiber/fiber/v2" "github.com/gofiber/fiber/v2/middleware/cors" + flog "github.com/gofiber/fiber/v2/middleware/logger" "github.com/gofiber/fiber/v2/middleware/limiter" - "github.com/gofiber/fiber/v2/middleware/logger" "github.com/gofiber/fiber/v2/middleware/recover" "github.com/minio/minio-go/v7" @@ -35,40 +35,47 @@ import ( ) func main() { + // структурный логгер — JSON для prod + logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})) + slog.SetDefault(logger) + cfg, err := config.Load() if err != nil { - log.Fatalf("config: %v", err) + logger.Error("config load failed", "err", err) + os.Exit(1) } ctx := context.Background() pg, err := db.New(ctx, cfg.PostgresDSN()) if err != nil { - log.Fatalf("postgres: %v", err) + logger.Error("postgres connect failed", "err", err) + os.Exit(1) } defer pg.Close() - log.Println("postgres: connected") + logger.Info("postgres connected") migDir := os.Getenv("MIGRATIONS_DIR") if migDir == "" { migDir = "./migrations" } if err := db.RunMigrations(ctx, pg, migDir); err != nil { - log.Fatalf("migrations: %v", err) + logger.Error("migrations failed", "err", err) + os.Exit(1) } rdb, err := redis.New(ctx, cfg.RedisAddr(), cfg.RedisPassword, cfg.RedisDB) if err != nil { - log.Printf("redis warning: %v", err) + logger.Warn("redis not available", "err", err) } else { - log.Println("redis: connected") + logger.Info("redis connected") } st, err := storage.New(ctx, cfg.MinIOEndpoint, cfg.MinIOAccessKey, cfg.MinIOSecretKey, cfg.MinIOBucket, cfg.MinIOUseSSL) if err != nil { - log.Printf("minio warning: %v", err) + logger.Warn("minio not available", "err", err) } else { - log.Println("minio: connected") + logger.Info("minio connected") } usersRepo := users.NewRepo(pg.Pool) @@ -87,9 +94,9 @@ func main() { if cfg.TelegramBotToken != "" { bot = telegram.NewBot(cfg.TelegramBotToken) tgH = handlers.NewTGHandler(bot, usersRepo, authSvc, consentRepo, auditRepo, cfg.TelegramWebhookSecret) - log.Println("telegram bot: configured") + logger.Info("telegram bot configured") } else { - log.Println("telegram bot: NOT configured (set TELEGRAM_BOT_TOKEN)") + logger.Warn("telegram bot NOT configured") } authH := handlers.NewAuthHandler(cfg, usersRepo, consentRepo, auditRepo, authSvc) @@ -107,7 +114,7 @@ func main() { WriteTimeout: 15 * time.Second, }) app.Use(recover.New()) - app.Use(logger.New()) + app.Use(flog.New()) app.Use(cors.New(cors.Config{ AllowOrigins: "https://buhapp.mygoodservice.ru,https://t.me,https://web.telegram.org", AllowHeaders: "Origin, Content-Type, Accept, Authorization, X-Telegram-Bot-Api-Secret-Token", @@ -135,7 +142,7 @@ func main() { return c.JSON(fiber.Map{ "status": "ok", "time": time.Now().UTC().Format(time.RFC3339), - "version": "0.5.0", + "version": "0.6.0", "db": pg.Pool != nil, "redis": rdb != nil, "storage": st != nil, @@ -152,7 +159,6 @@ func main() { return c.JSON(fiber.Map{"version": cfg.DisclaimerVersion, "url": "/legal/DISCLAIMER.md"}) }) - // публичная раздача MinIO через бэкенд (MinIO в docker network, не на хосте) if st != nil { app.Get("/storage/*", func(c *fiber.Ctx) error { key := c.Params("*") @@ -224,7 +230,7 @@ func main() { sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) <-sigCh - log.Println("shutdown...") + logger.Info("shutdown initiated") _ = app.ShutdownWithTimeout(10 * time.Second) }() @@ -232,13 +238,13 @@ func main() { webhookURL := cfg.TelegramWebhookURL if webhookURL != "" { if err := bot.SetWebhook(webhookURL, cfg.TelegramWebhookSecret); err != nil { - log.Printf("telegram setWebhook failed: %v (fallback to long poll)", err) + logger.Warn("telegram setWebhook failed, will use long poll", "err", err) } else { - log.Printf("telegram bot: webhook set to %s", webhookURL) + logger.Info("telegram bot webhook set", "url", webhookURL) } } else { go func() { - log.Println("telegram bot: starting long-poll loop") + logger.Info("telegram bot starting long-poll loop") bot.RunLoop(func(u telegram.Update) { tgH.ProcessUpdate(u) }) @@ -247,8 +253,9 @@ func main() { } addr := fmt.Sprintf(":%d", cfg.AppPort) - log.Printf("listening on %s", addr) + logger.Info("listening", "addr", addr) if err := app.Listen(addr); err != nil { - log.Fatalf("listen: %v", err) + logger.Error("listen failed", "err", err) + os.Exit(1) } } diff --git a/internal/audit/repo.go b/internal/audit/repo.go index cd31acf..e9551da 100644 --- a/internal/audit/repo.go +++ b/internal/audit/repo.go @@ -30,9 +30,14 @@ func NewRepo(pool *pgxpool.Pool) *Repo { func (r *Repo) Log(ctx context.Context, e *Event) error { meta := []byte("null") - if e.Metadata != nil { - meta = []byte(`{}`) // упрощённо - // NOTE: для prod надо marshal JSON; пропускаем пока + if len(e.Metadata) > 0 { + // Если Metadata уже валидный JSON — пишем как есть + // Иначе (хэштег, текст) — обернём в JSON-объект + meta = e.Metadata + // если начинается не с { или [, оборачиваем + if len(meta) == 0 || (meta[0] != '{' && meta[0] != '[' && meta[0] != '"') { + meta = []byte(`{"data":` + string(meta) + `}`) + } } return r.pool.QueryRow(ctx, ` INSERT INTO audit_log (user_id, action, target_type, target_id, metadata, ip, user_agent) diff --git a/internal/chat/repo.go b/internal/chat/repo.go index 4b6a297..e6cf4a5 100644 --- a/internal/chat/repo.go +++ b/internal/chat/repo.go @@ -25,6 +25,7 @@ type Message struct { Body string `json:"body"` PhotoURL string `json:"photo_url,omitempty"` Read bool `json:"read"` + ReadAt *time.Time `json:"read_at,omitempty"` Edited bool `json:"edited"` Deleted bool `json:"deleted"` CreatedAt time.Time `json:"created_at"` @@ -39,34 +40,24 @@ func NewRepo(pool *pgxpool.Pool) *Repo { return &Repo{pool: pool} } -// EnsureChat — нормализуем user_a < user_b, чтобы избежать дублей +// EnsureChat — race-safe: ON CONFLICT DO NOTHING + RETURNING. +// Нормализуем 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`, + ON CONFLICT (user_a, user_b) DO UPDATE + SET last_msg_at = chats.last_msg_at -- noop, чтобы вернулось + RETURNING id, user_a, user_b, created_at, last_msg_at`, a, b, - ).Scan(&c.ID, &c.CreatedAt, &c.LastMsgAt) + ).Scan(&c.ID, &c.UserA, &c.UserB, &c.CreatedAt, &c.LastMsgAt) if err != nil { return nil, err } - c.UserA, c.UserB = a, b return c, nil } @@ -140,20 +131,34 @@ func (r *Repo) ListChats(ctx context.Context, me uuid.UUID, limit, offset int) ( return out, rows.Err() } +// Send — атомарно: INSERT message + UPDATE chat.last_msg_at в одной транзакции 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) + tx, err := r.pool.Begin(ctx) if err != nil { return nil, err } - _, _ = r.pool.Exec(ctx, `UPDATE chats SET last_msg_at = NOW() WHERE id=$1`, chatID) + defer tx.Rollback(ctx) + + err = tx.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, read_at, edited, deleted, created_at, updated_at`, + chatID, senderID, body, photoURL, + ).Scan(&m.ID, &m.ChatID, &m.SenderID, &m.Body, &m.PhotoURL, + &m.Read, &m.ReadAt, &m.Edited, &m.Deleted, &m.CreatedAt, &m.UpdatedAt) + if err != nil { + return nil, err + } + + if _, err := tx.Exec(ctx, `UPDATE chats SET last_msg_at = NOW() WHERE id=$1`, chatID); err != nil { + return nil, err + } + + if err := tx.Commit(ctx); err != nil { + return nil, err + } return m, nil } @@ -163,7 +168,7 @@ func (r *Repo) ListMessages(ctx context.Context, chatID uuid.UUID, limit, offset } rows, err := r.pool.Query(ctx, ` SELECT id, chat_id, sender_id, COALESCE(body, ''), COALESCE(photo_url, ''), - read, edited, deleted, created_at, updated_at + read, read_at, edited, deleted, created_at, updated_at FROM messages WHERE chat_id=$1 ORDER BY created_at ASC @@ -176,7 +181,7 @@ func (r *Repo) ListMessages(ctx context.Context, chatID uuid.UUID, limit, offset 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 { + &m.Read, &m.ReadAt, &m.Edited, &m.Deleted, &m.CreatedAt, &m.UpdatedAt); err != nil { return nil, err } out = append(out, m) @@ -203,10 +208,10 @@ func (r *Repo) GetMessage(ctx context.Context, msgID uuid.UUID) (*Message, error m := &Message{} err := r.pool.QueryRow(ctx, ` SELECT id, chat_id, sender_id, COALESCE(body, ''), COALESCE(photo_url, ''), - read, edited, deleted, created_at, updated_at + read, read_at, edited, deleted, created_at, updated_at FROM messages WHERE id=$1`, msgID, ).Scan(&m.ID, &m.ChatID, &m.SenderID, &m.Body, &m.PhotoURL, - &m.Read, &m.Edited, &m.Deleted, &m.CreatedAt, &m.UpdatedAt) + &m.Read, &m.ReadAt, &m.Edited, &m.Deleted, &m.CreatedAt, &m.UpdatedAt) if errors.Is(err, pgx.ErrNoRows) { return nil, nil } @@ -231,13 +236,34 @@ func (r *Repo) DeleteMessage(ctx context.Context, msgID, senderID uuid.UUID) err return nil } +// MarkRead — атомарно с проверкой членства. Возвращает chat_id для WS broadcast. 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`, + // member-check + update в одном UPDATE с подзапросом + res, err := r.pool.Exec(ctx, ` + UPDATE messages SET read=TRUE, read_at=NOW() + WHERE chat_id=$1 AND sender_id != $2 AND read=FALSE + AND EXISTS (SELECT 1 FROM chats c + WHERE c.id=$1 AND (c.user_a=$2 OR c.user_b=$2))`, chatID, meID, ) - return err + if err != nil { + return err + } + if res.RowsAffected() == 0 { + // не ошибка, просто либо нет сообщений, либо юзер не в чате + // проверим явно membership для возврата ошибки + var isMember bool + if err := r.pool.QueryRow(ctx, + `SELECT EXISTS(SELECT 1 FROM chats WHERE id=$1 AND (user_a=$2 OR user_b=$2))`, + chatID, meID, + ).Scan(&isMember); err != nil { + return err + } + if !isMember { + return errors.New("not a member of this chat") + } + } + return nil } // Block / Unblock diff --git a/internal/handlers/chat.go b/internal/handlers/chat.go index 60d60c2..7af5273 100644 --- a/internal/handlers/chat.go +++ b/internal/handlers/chat.go @@ -147,6 +147,9 @@ func (h *ChatHandlers) SendMessage(c *fiber.Ctx) error { "sender_id": m.SenderID, "body": m.Body, "photo_url": m.PhotoURL, + "read": m.Read, + "read_at": m.ReadAt, + "edited": m.Edited, "created_at": m.CreatedAt, }, Time: time.Now(), @@ -160,6 +163,9 @@ func (h *ChatHandlers) SendMessage(c *fiber.Ctx) error { "sender_id": m.SenderID, "body": m.Body, "photo_url": m.PhotoURL, + "read": m.Read, + "read_at": m.ReadAt, + "edited": m.Edited, "created_at": m.CreatedAt, }, Time: time.Now(), @@ -278,7 +284,24 @@ func (h *ChatHandlers) MarkRead(c *fiber.Ctx) error { 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.Status(fiber.StatusForbidden).JSON(fiber.Map{"error": err.Error()}) + } + // оповестим отправителя через WS что сообщения прочитаны + chat, _ := h.Chat.GetByID(c.UserContext(), chatID) + if chat != nil { + other := chat.UserA + if other == me { + other = chat.UserB + } + h.Hub.SendTo(other, ws.Outgoing{ + Type: "message_read", + Payload: fiber.Map{ + "chat_id": chatID, + "reader": me, + "read_at": time.Now(), + }, + Time: time.Now(), + }) } return c.JSON(fiber.Map{"ok": true}) } diff --git a/internal/locations/repo.go b/internal/locations/repo.go index 7820e8c..e65893a 100644 --- a/internal/locations/repo.go +++ b/internal/locations/repo.go @@ -76,6 +76,8 @@ func (r *Repo) Get(ctx context.Context, userID uuid.UUID) (*Location, error) { } // Nearby — пользователи в радиусе radius км (по haversine), visible, не старше 1 часа +// Оптимизация: фильтруем сначала по bounding-box (lat/lng), потом haversine только для отфильтрованных. +// Это устраняет двойной расчёт haversine и даёт индекс если добавить GiST/btree на lat,lng. type Nearby struct { UserID uuid.UUID Name string @@ -89,28 +91,43 @@ func (r *Repo) Nearby(ctx context.Context, myID uuid.UUID, lat, lng, radiusKm fl if limit <= 0 || limit > 500 { limit = 100 } + // bounding box: 1° lat ≈ 111 км, 1° lng ≈ 111 км * cos(lat) + latDelta := radiusKm / 111.0 + lngDelta := radiusKm / (111.0 * math.Cos(lat*math.Pi/180)) + if latDelta < 0.001 { + latDelta = 0.001 + } + if lngDelta < 0.001 { + lngDelta = 0.001 + } rows, err := r.pool.Query(ctx, ` - SELECT u.id, u.name, u.photo_url, ul.lat, ul.lng, + WITH bbox AS ( + SELECT u.id, u.name, u.photo_url, ul.lat, ul.lng + FROM user_locations ul + JOIN users u ON u.id = ul.user_id + WHERE ul.visible = TRUE + AND u.is_blocked = FALSE + AND u.id != $3 + AND ul.lat IS NOT NULL AND ul.lng IS NOT NULL + AND ul.updated_at > NOW() - INTERVAL '1 hour' + AND ul.lat BETWEEN $1 - $6 AND $1 + $6 + AND ul.lng BETWEEN $2 - $7 AND $2 + $7 + ) + SELECT id, name, photo_url, lat, lng, 2 * 6371 * asin(sqrt( - sin(radians((ul.lat - $1) / 2))^2 + - cos(radians($1)) * cos(radians(ul.lat)) * - sin(radians((ul.lng - $2) / 2))^2 - )) AS dist_km - FROM user_locations ul - JOIN users u ON u.id = ul.user_id - WHERE ul.visible = TRUE - AND u.is_blocked = FALSE - AND u.id != $3 - AND ul.lat IS NOT NULL AND ul.lng IS NOT NULL - AND ul.updated_at > NOW() - INTERVAL '1 hour' - AND 2 * 6371 * asin(sqrt( - sin(radians((ul.lat - $1) / 2))^2 + - cos(radians($1)) * cos(radians(ul.lat)) * - sin(radians((ul.lng - $2) / 2))^2 + sin(radians((lat - $1) / 2))^2 + + cos(radians($1)) * cos(radians(lat)) * + sin(radians((lng - $2) / 2))^2 + )) * 1000 AS dist_m + FROM bbox + WHERE 2 * 6371 * asin(sqrt( + sin(radians((lat - $1) / 2))^2 + + cos(radians($1)) * cos(radians(lat)) * + sin(radians((lng - $2) / 2))^2 )) <= $4 - ORDER BY dist_km + ORDER BY dist_m LIMIT $5`, - lat, lng, myID, radiusKm, limit, + lat, lng, myID, radiusKm, limit, latDelta, lngDelta, ) if err != nil { return nil, err @@ -127,7 +144,6 @@ func (r *Repo) Nearby(ctx context.Context, myID uuid.UUID, lat, lng, radiusKm fl if photo != nil { n.PhotoURL = *photo } - n.DistanceM = n.DistanceM * 1000 out = append(out, n) } return out, rows.Err() diff --git a/migrations/0006_read_at.down.sql b/migrations/0006_read_at.down.sql new file mode 100644 index 0000000..a5fd449 --- /dev/null +++ b/migrations/0006_read_at.down.sql @@ -0,0 +1 @@ +ALTER TABLE messages DROP COLUMN IF EXISTS read_at; \ No newline at end of file diff --git a/migrations/0006_read_at.up.sql b/migrations/0006_read_at.up.sql new file mode 100644 index 0000000..baeaa7e --- /dev/null +++ b/migrations/0006_read_at.up.sql @@ -0,0 +1,3 @@ +-- read_at timestamp для галочек "прочитано" +ALTER TABLE messages ADD COLUMN IF NOT EXISTS read_at TIMESTAMPTZ; +UPDATE messages SET read_at = updated_at WHERE read = TRUE AND read_at IS NULL; \ No newline at end of file