3.2 Пул соединений: pgxpool

pgx.Conn — одно соединение, не потокобезопасное. В сервисе, где хендлеры летят параллельно, нужен пул:

package main

import (
	"context"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
)

func newPool(ctx context.Context, dsn string) (*pgxpool.Pool, error) {
	cfg, err := pgxpool.ParseConfig(dsn)
	// dsn: "postgres://user:pass@localhost:5432/mydb?sslmode=disable"
	//       или в стиле keyword: "host=... user=... dbname=..."
	if err != nil {
		return nil, err
	}

	cfg.MaxConns = 10                       // потолок; по умолчанию max(4, NumCPU)
	cfg.MinConns = 2                        // держать тёплыми
	cfg.MaxConnLifetime = time.Hour         // ротация соединений
	cfg.MaxConnIdleTime = 30 * time.Minute
	cfg.HealthCheckPeriod = time.Minute

	pool, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		return nil, err
	}
	if err := pool.Ping(ctx); err != nil { // проверяем сразу, а не на первом запросе
		pool.Close()
		return nil, err
	}
	return pool, nil
}

Пул потокобезопасен: создаётся один раз в main, передаётся в слои хранилища, закрывается pool.Close() при shutdown (после остановки HTTP-сервера — помните порядок «от входа к выходу»). Каждый Query/Exec сам берёт соединение из пула и возвращает его.

Про размер: MaxConns — это не «чем больше, тем лучше». У Postgres соединения дороги; десятки соединений на инстанс сервиса × десятки инстансов легко упираются в max_connections сервера. Считайте от общего бюджета.

3.3 Запросы: Exec, QueryRow, Query

type Store struct {
	pool *pgxpool.Pool
}

// Exec — запросы без результата (INSERT/UPDATE/DELETE/DDL).
func (s *Store) DeactivateUser(ctx context.Context, id int64) error {
	tag, err := s.pool.Exec(ctx,
		`UPDATE users SET active = false WHERE id = $1`, id)
	if err != nil {
		return fmt.Errorf("deactivate user %d: %w", id, err)
	}
	if tag.RowsAffected() == 0 {
		return ErrNotFound // UPDATE «прошёл», но никого не задел — это часто бизнес-ошибка
	}
	return nil
}

// QueryRow — ровно одна строка.
func (s *Store) GetUser(ctx context.Context, id int64) (*User, error) {
	var u User
	err := s.pool.QueryRow(ctx,
		`SELECT id, name, email, created_at FROM users WHERE id = $1`, id,
	).Scan(&u.ID, &u.Name, &u.Email, &u.CreatedAt)

	if errors.Is(err, pgx.ErrNoRows) {
		return nil, ErrNotFound // доменная ошибка вместо драйверной
	}
	if err != nil {
		return nil, fmt.Errorf("get user %d: %w", id, err)
	}
	return &u, nil
}

Заповеди:

• Только плейсхолдеры $1, $2... для данных. Конкатенация строк = SQL-инъекция, без исключений. (Имена таблиц/колонок плейсхолдером передать нельзя — если они динамические, валидируйте по белому списку.)
• ctx — в каждый вызов. Отмена запроса клиентом или таймаут реально прервут запрос на стороне Postgres.
• pgx.ErrNoRows — не авария, а «не найдено»; конвертируйте в доменную ошибку на границе слоя хранилища, чтобы хендлеры не импортировали pgx.

3.4 Множество строк: Query и щедрый сахар CollectRows

Классический способ:

func (s *Store) ListActive(ctx context.Context) ([]User, error) {
	rows, err := s.pool.Query(ctx,
		`SELECT id, name, email, created_at FROM users WHERE active ORDER BY id`)
	if err != nil {
		return nil, err
	}
	defer rows.Close() // обязательно: возвращает соединение в пул

	var users []User
	for rows.Next() {
		var u User
		if err := rows.Scan(&u.ID, &u.Name, &u.Email, &u.CreatedAt); err != nil {
			return nil, err
		}
		users = append(users, u)
	}
	return users, rows.Err() // ошибки итерации всплывают ЗДЕСЬ — не забывать!
}

rows.Err() после цикла обязателен: обрыв соединения посреди выборки не вернёт ошибку из Next() (он просто вернёт false), она ждёт вас в Err(). Забытая проверка = молча усечённые данные.

pgx v5 добавил дженерик-хелперы, убирающие всю эту шарманку:

// Строки → структуры по именам полей (сопоставление name → Name, created_at → CreatedAt)
users, err := pgx.CollectRows(rows, pgx.RowToStructByName[User])

// Ровно одна строка:
u, err := pgx.CollectOneRow(rows, pgx.RowToStructByName[User])

// Одна колонка в срез:
ids, err := pgx.CollectRows(rows, pgx.RowTo[int64])

// В одну строку целиком:
users, err := pgx.CollectRows(
	mustQuery(s.pool.Query(ctx, `SELECT ... FROM users`)),
	pgx.RowToStructByName[User],
)

RowToStructByName требует совпадения набора колонок и полей (защита от рассинхрона запроса и структуры), RowToStructByNameLax разрешает лишние поля в структуре. Эти хелперы закрывают 90% болей, ради которых люди тащили ORM.

3.5 Транзакции

func (s *Store) TransferMoney(ctx context.Context, from, to int64, amount int64) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	// Страховка: если выйдем раньше Commit (ошибка, паника) — откат.
	// Rollback после успешного Commit безвреден: вернёт ErrTxClosed, игнорируем.
	defer tx.Rollback(ctx)

	if _, err := tx.Exec(ctx,
		`UPDATE accounts SET balance = balance - $1 WHERE id = $2 AND balance >= $1`,
		amount, from); err != nil {
		return fmt.Errorf("списание: %w", err)
	}
	if _, err := tx.Exec(ctx,
		`UPDATE accounts SET balance = balance + $1 WHERE id = $2`,
		amount, to); err != nil {
		return fmt.Errorf("зачисление: %w", err)
	}

	return tx.Commit(ctx)
}

Паттерн defer tx.Rollback(ctx) сразу после Begin — обязательный рефлекс: любой ранний return безопасен, транзакция не повиснет. Всё внутри транзакции выполняется через tx, не через pool — иначе запрос уедет в другое соединение и окажется вне транзакции (тихий и страшный баг).

Для «выполни функцию в транзакции с авто-commit/rollback» есть хелпер:

err := pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
	// return nil → Commit; return err или паника → Rollback
	...
	return nil
})

3.6 Ошибки Postgres: pgconn.PgError

Уникальные индексы и констрейнты — это не «ошибка БД», это бизнес-события. Различаем их по SQLSTATE:

import "github.com/jackc/pgx/v5/pgconn"

func (s *Store) CreateUser(ctx context.Context, name, email string) (int64, error) {
	var id int64
	err := s.pool.QueryRow(ctx,
		`INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id`,
		name, email).Scan(&id)

	var pgErr *pgconn.PgError
	if errors.As(err, &pgErr) {
		switch pgErr.Code {
		case "23505": // unique_violation
			return 0, ErrEmailTaken
		case "23503": // foreign_key_violation
			return 0, ErrRelatedMissing
		}
	}
	if err != nil {
		return 0, fmt.Errorf("create user: %w", err)
	}
	return id, nil
}

Обратите внимание на RETURNING id — постгресовый способ получить сгенерированный ключ одним запросом, вместо двух round-trip'ов.

3.7 Batch: пачка запросов за один round-trip

Сетевая задержка × количество запросов — главный убийца производительности «болтливого» кода. Batch склеивает запросы:

func (s *Store) GetDashboard(ctx context.Context, userID int64) (*Dashboard, error) {
	b := &pgx.Batch{}
	b.Queue(`SELECT id, name, email, created_at FROM users WHERE id = $1`, userID)
	b.Queue(`SELECT id, total, status FROM orders WHERE user_id = $1 ORDER BY id DESC LIMIT 10`, userID)
	b.Queue(`SELECT count(*) FROM notifications WHERE user_id = $1 AND NOT read`, userID)

	res := s.pool.SendBatch(ctx, b) // ОДИН поход в сеть
	defer res.Close()               // обязательно, в порядке очереди

	var d Dashboard
	var err error

	if d.User, err = pgx.CollectOneRow(mustQ(res.Query()), pgx.RowToStructByName[User]); err != nil {
		return nil, err
	}
	if d.Orders, err = pgx.CollectRows(mustQ(res.Query()), pgx.RowToStructByName[Order]); err != nil {
		return nil, err
	}
	if err = res.QueryRow().Scan(&d.UnreadCount); err != nil {
		return nil, err
	}
	return &d, nil
}

Результаты вычитываются строго в порядке постановки в очередь. Batch внутри транзакции (tx.SendBatch) — тоже работает.

3.8 CopyFrom: массовая вставка

Вставить 100 000 строк INSERT'ами — минуты; протоколом COPY — секунды:

rows := make([][]any, 0, len(events))
for _, e := range events {
	rows = append(rows, []any{e.UserID, e.Type, e.Payload, e.At})
}

n, err := s.pool.CopyFrom(ctx,
	pgx.Identifier{"events"},                       // таблица
	[]string{"user_id", "type", "payload", "at"},   // колонки
	pgx.CopyFromRows(rows),
)
// n — сколько строк залито

Для стриминга без материализации среза есть pgx.CopyFromFunc / собственная реализация CopyFromSource.

3.9 LISTEN/NOTIFY: пуш-события из БД

Postgres умеет присылать события — удобно для инвалидации кешей и лёгкого pub/sub без брокера:

func listenEvents(ctx context.Context, pool *pgxpool.Pool) error {
	conn, err := pool.Acquire(ctx) // выделенное соединение — LISTEN живёт на соединении
	if err != nil {
		return err
	}
	defer conn.Release()

	if _, err := conn.Exec(ctx, `LISTEN user_events`); err != nil {
		return err
	}
	for {
		n, err := conn.Conn().WaitForNotification(ctx) // блокируется до события или отмены ctx
		if err != nil {
			return err // ctx отменён при shutdown — штатный выход
		}
		log.Printf("канал=%s payload=%s", n.Channel, n.Payload)
	}
}

// Слать из SQL: NOTIFY user_events, '{"user_id":42}';
// или из триггера: PERFORM pg_notify('user_events', row_to_json(NEW)::text);

Эта штука идеально стыкуется с WebSocket из следующей главы: БД нотифицирует сервис, сервис рассылает клиентам.

---

4. WebSocket

4.1 Что это и когда нужно

HTTP — запрос-ответ: сервер молчит, пока его не спросят. WebSocket — постоянное двунаправленное соединение поверх TCP: любая сторона пишет в любой момент. Нужен для чатов, нотификаций, live-дашбордов, курсов/котировок, многопользовательских редакторов, игр.

Протокол начинается как HTTP: клиент шлёт GET с заголовками Upgrade: websocket, Sec-WebSocket-Key; сервер отвечает 101 Switching Protocols — и дальше по тому же TCP-соединению летают уже не HTTP-запросы, а фреймы: текстовые, бинарные и служебные (ping/pong/close).

В стандартной библиотеке Go веб-сокетов нет. Два основных пакета:

• github.com/gorilla/websocket — ветеран, огромная база примеров, низкоуровневый контроль. Возьмём его за основу — на нём проще всего показать механику.
• github.com/coder/websocket (бывший nhooyr.io/websocket) — современный, компактный API с ctx во всех операциях, поддержка Wasm. Хороший выбор для нового кода; всё концептуальное из этой главы применимо и к нему.

go get github.com/gorilla/websocket

4.2 Upgrade: превращаем HTTP в WebSocket

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
	// ЗАЩИТА ОТ CSWSH (cross-site WebSocket hijacking):
	// браузер шлёт Origin, и мы обязаны проверить, что это НАШ сайт.
	// Вернуть true не глядя = любой чужой сайт сможет открыть сокет
	// от имени залогиненного пользователя (куки уедут автоматически!).
	CheckOrigin: func(r *http.Request) bool {
		return r.Header.Get("Origin") == "https://myapp.example.com"
	},
}

func wsHandler(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		// Upgrade сам ответил клиенту ошибкой — просто логируем
		log.Println("upgrade:", err)
		return
	}
	defer conn.Close()

	// эхо-сервер: читаем сообщение — отправляем обратно
	for {
		msgType, data, err := conn.ReadMessage()
		if err != nil {
			// сюда попадаем и при штатном закрытии клиентом
			if websocket.IsUnexpectedCloseError(err,
				websocket.CloseGoingAway, websocket.CloseNormalClosure) {
				log.Println("read:", err)
			}
			return
		}
		if err := conn.WriteMessage(msgType, data); err != nil {
			return
		}
	}
}

Аутентификация: браузерный WebSocket API не позволяет ставить произвольные заголовки, так что Authorization: Bearer ... не пройдёт. Варианты: кука сессии (едет автоматически — потому и критичен CheckOrigin), токен в query-параметре (попадает в логи — осторожно), или первый-же кадр после подключения с токеном внутри. Проверяйте auth до Upgrade — после 101 отвечать HTTP-статусами уже нельзя.

4.3 Золотое правило конкурентности WebSocket

Вот то, из-за чего падает большинство самописных реализаций:

> Соединение поддерживает максимум одного одновременного читателя и одного одновременного писателя. Две горутины, пишущие в один conn параллельно, — это перемешанные фреймы и паника/коррупция протокола.

А писать в соединение хотят многие: бизнес-логика шлёт сообщения, пингер шлёт ping, hub рассылает broadcast... Каноническое решение — вся запись идёт через один канал, который вычитывает единственная горутина-писатель (writePump). Никто не трогает conn напрямую. Знакомо? Это буквально «не общайтесь через разделяемую память — разделяйте память через общение».

4.4 Каноническая архитектура: Client + Hub

Схема из официального примера gorilla (chat), выдержавшая годы продакшена. Три сущности:

                     ┌──────────────── Hub ────────────────┐
browser ⇄ readPump ─▶ inbound     register/unregister      │
browser ⇄ writePump ◀─ client.send ◀── broadcast ──────────┘

• readPump (горутина на клиента): единственный читатель conn; принятое отдаёт хабу.
• writePump (горутина на клиента): единственный писатель conn; берёт из канала send, плюс шлёт ping по тикеру.
• Hub (одна горутина на приложение): владеет множеством клиентов, обрабатывает register/unregister/broadcast. Поскольку с мапой клиентов работает только горутина хаба, мьютекс не нужен — синхронизация каналами.

package ws

import (
	"log"
	"time"

	"github.com/gorilla/websocket"
)

const (
	writeWait  = 10 * time.Second     // дедлайн на одну запись
	pongWait   = 60 * time.Second     // ждём pong не дольше этого
	pingPeriod = (pongWait * 9) / 10  // пингуем чаще, чем ждём (54s)
	maxMsgSize = 4096                 // лимит входящего сообщения
)

type Hub struct {
	clients    map[*Client]struct{}
	register   chan *Client
	unregister chan *Client
	broadcast  chan []byte
}

func NewHub() *Hub {
	return &Hub{
		clients:    make(map[*Client]struct{}),
		register:   make(chan *Client),
		unregister: make(chan *Client),
		broadcast:  make(chan []byte, 256),
	}
}

func (h *Hub) Run(ctx context.Context) {
	for {
		select {
		case c := <-h.register:
			h.clients[c] = struct{}{}

		case c := <-h.unregister:
			if _, ok := h.clients[c]; ok {
				delete(h.clients, c)
				close(c.send) // сигнал writePump'у клиента: закругляйся
			}

		case msg := <-h.broadcast:
			for c := range h.clients {
				select {
				case c.send <- msg:
				default:
					// Буфер клиента полон = клиент слишком медленный.
					// НЕ блокируем весь хаб из-за одного тормоза — отключаем его.
					delete(h.clients, c)
					close(c.send)
				}
			}

		case <-ctx.Done(): // graceful shutdown
			for c := range h.clients {
				close(c.send)
			}
			return
		}
	}
}

type Client struct {
	hub  *Hub
	conn *websocket.Conn
	send chan []byte // очередь исходящих; писатель в conn — только writePump
}

func (c *Client) readPump() {
	defer func() {
		c.hub.unregister <- c
		c.conn.Close()
	}()

	c.conn.SetReadLimit(maxMsgSize)
	c.conn.SetReadDeadline(time.Now().Add(pongWait))
	c.conn.SetPongHandler(func(string) error {
		// живой клиент ответил на ping — продлеваем дедлайн чтения
		c.conn.SetReadDeadline(time.Now().Add(pongWait))
		return nil
	})

	for {
		_, msg, err := c.conn.ReadMessage()
		if err != nil {
			return // любой разрыв → defer: unregister + close
		}
		c.hub.broadcast <- msg // в чате: входящее → всем
	}
}

func (c *Client) writePump() {
	ticker := time.NewTicker(pingPeriod)
	defer func() {
		ticker.Stop()
		c.conn.Close() // закрытие conn разбудит readPump с ошибкой — симметричный демонтаж
	}()

	for {
		select {
		case msg, ok := <-c.send:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if !ok {
				// хаб закрыл канал — вежливо прощаемся
				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
				return
			}
			if err := c.conn.WriteMessage(websocket.TextMessage, msg); err != nil {
				return
			}

		case <-ticker.C:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return
			}
		}
	}
}

// HTTP-хендлер, порождающий клиента:
func ServeWS(hub *Hub, w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		return
	}
	c := &Client{hub: hub, conn: conn, send: make(chan []byte, 64)}
	hub.register <- c

	go c.writePump()
	go c.readPump()
	// хендлер завершается — соединение живёт в двух горутинах клиента
}

4.5 Почему именно так: разбор решений

Ping/pong — это детектор смерти. TCP может «молча» умереть (клиент ушёл в метро, роутер перезагрузился) — без трафика вы узнаете об этом через десятки минут ОС-таймаутов, а горутины будут висеть. Схема: сервер шлёт ping каждые 54с; живой браузер отвечает pong автоматически; pong продлевает ReadDeadline на 60с. Нет понга — ReadMessage падает по дедлайну → демонтаж. Мёртвые соединения гарантированно убираются за ~минуту.

Сброс медленных клиентов (default в broadcast) — защита всей системы. Один клиент на GPRS, не успевающий читать, забьёт свой send-буфер. Блокироваться на нём — значит остановить рассылку всем. Политика «не влез в буфер — отключаем» жестока, но честна; альтернативы — копить в персональной очереди с лимитом или отправлять клиенту только последний снапшот (для дашбордов).

Демонтаж всегда симметричен. Умер readPump → unregister → хаб делает close(c.send) → writePump видит !ok → закрывает conn. Умер writePump → закрывает conn → readPump падает на чтении → unregister. Любая смерть с любой стороны сворачивает обе горутины и вычищает клиента из хаба. Проверяйте свою реализацию именно на это: «а что, если умрёт вот эта половина?»

Hub без мьютекса — сознательный выбор. Мапа клиентов конфайнится в одной горутине, все команды — через каналы. Можно и наоборот: мапа под sync.RWMutex, методы Register/Broadcast. Оба варианта валидны; канальный лучше масштабируется по сложности (легко добавить комнаты, приоритеты), мьютексный — короче для простых случаев.

4.6 Клиент на Go

WebSocket-клиенты пишут не только браузеры:

conn, resp, err := websocket.DefaultDialer.DialContext(ctx,
	"wss://api.example.com/ws", http.Header{
		"Authorization": []string{"Bearer " + token}, // из Go заголовки МОЖНО
	})
if err != nil {
	log.Fatal("dial:", err, resp)
}
defer conn.Close()

go func() { // читатель
	for {
		_, msg, err := conn.ReadMessage()
		if err != nil {
			return
		}
		log.Printf("← %s", msg)
	}
}()

conn.WriteJSON(map[string]any{"type": "subscribe", "channel": "orders"})

WriteJSON/ReadJSON — удобные обёртки для JSON-сообщений (в проде обычно определяют типизированный конверт: {"type": "...", "payload": {...}} и switch по type).

4.7 Продакшен-заметки

• wss:// всегда. TLS обычно терминируется на балансировщике/ингрессе; проследите, чтобы прокси пробрасывал Upgrade (в nginx: proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; proxy_read_timeout больше pingPeriod).
• Горизонтальное масштабирование: клиенты одного «чата» окажутся на разных инстансах — хабам нужен общий шинный слой: Redis Pub/Sub, NATS или тот самый LISTEN/NOTIFY из 3.9.
• Reconnect — забота клиента: экспоненциальный backoff + повторная подписка + докачка пропущенного (храните события с ID).
• Метрики: число открытых соединений, длина send-буферов, частота сбросов медленных клиентов — три графика, по которым видно здоровье реалтайма.

---

5. Скелет сервиса

Финал: соберём все четыре главы в структуру реального сервиса — как это обычно раскладывают по файлам.

myservice/
├── cmd/server/main.go        # сборка и запуск
├── internal/
│   ├── api/                  # chi-роутер, хендлеры, middleware
│   │   ├── router.go
│   │   ├── users.go
│   │   └── ws.go
│   ├── store/                # pgx: всё общение с БД
│   │   └── store.go
│   └── ws/                   # hub, client
│       └── hub.go
└── migrations/               # SQL-миграции (goose / golang-migrate / tern)

// cmd/server/main.go — дирижёр
func main() {
	ctx, stop := signal.NotifyContext(context.Background(),
		syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	pool, err := store.NewPool(ctx, os.Getenv("DATABASE_URL")) // глава 3
	if err != nil {
		log.Fatal(err)
	}
	defer pool.Close()

	st := store.New(pool)

	hub := ws.NewHub() // глава 4
	go hub.Run(ctx)

	router := api.NewRouter(st, hub) // глава 2: роутер принимает зависимости

	srv := &http.Server{ // глава 1: явный сервер с таймаутами
		Addr:              ":8080",
		Handler:           router,
		ReadHeaderTimeout: 5 * time.Second,
	}

	go func() {
		log.Println("listening :8080")
		if err := srv.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
			log.Fatal(err)
		}
	}()

	<-ctx.Done() // ждём сигнал

	shCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := srv.Shutdown(shCtx); err != nil {
		log.Println("shutdown:", err)
	}
	// порядок: HTTP остановлен → хаб получил ctx.Done() и закрыл клиентов →
	// defer pool.Close() последним, когда запросов в БД уже нет
}

Зависимости текут в одну сторону: main → api → store, main → ws. store не знает про HTTP, api не знает про pgx (только про интерфейс store) — благодаря этому каждый слой тестируется отдельно: хендлеры с фейковым store через httptest, store — интеграционно на Postgres в