From c79a3cfe68b919819ce7f1d8af529b7897511579 Mon Sep 17 00:00:00 2001 From: vantablackTea Date: Sun, 27 Sep 2026 19:42:49 +0300 Subject: [PATCH] backend barebone done, big chunk of work done --- .github/workflows/ci.yml | 16 ++ README.md | 48 +++- deploy/.env.example | 1 + deploy/docker-compose.yml | 2 + server/cmd/api/main.go | 30 +- server/go.mod | 21 ++ server/go.sum | 49 ++++ server/internal/api/api.go | 67 +++++ server/internal/api/auth.go | 172 ++++++++++++ server/internal/api/contacts.go | 93 +++++++ server/internal/api/conversations.go | 172 ++++++++++++ server/internal/api/middleware.go | 74 +++++ server/internal/api/respond.go | 33 +++ server/internal/auth/password.go | 103 +++++++ server/internal/auth/password_test.go | 49 ++++ server/internal/auth/refresh.go | 27 ++ server/internal/auth/refresh_test.go | 39 +++ server/internal/auth/tokens.go | 57 ++++ server/internal/auth/tokens_test.go | 78 ++++++ server/internal/platform/config/config.go | 10 + .../internal/platform/config/config_test.go | 13 + server/internal/realtime/client.go | 136 +++++++++ server/internal/realtime/events.go | 70 +++++ server/internal/realtime/hub.go | 163 +++++++++++ server/internal/realtime/hub_test.go | 262 ++++++++++++++++++ server/internal/server/server.go | 13 +- server/internal/store/contacts.go | 69 +++++ server/internal/store/conversations.go | 171 ++++++++++++ server/internal/store/errors.go | 11 + server/internal/store/integration_test.go | 111 ++++++++ server/internal/store/messages.go | 83 ++++++ server/internal/store/migrate.go | 89 ++++++ .../internal/store/migrations/0001_init.sql | 56 ++++ server/internal/store/sessions.go | 67 +++++ server/internal/store/store.go | 71 +++++ server/internal/store/users.go | 98 +++++++ 36 files changed, 2607 insertions(+), 17 deletions(-) create mode 100644 server/internal/api/api.go create mode 100644 server/internal/api/auth.go create mode 100644 server/internal/api/contacts.go create mode 100644 server/internal/api/conversations.go create mode 100644 server/internal/api/middleware.go create mode 100644 server/internal/api/respond.go create mode 100644 server/internal/auth/password.go create mode 100644 server/internal/auth/password_test.go create mode 100644 server/internal/auth/refresh.go create mode 100644 server/internal/auth/refresh_test.go create mode 100644 server/internal/auth/tokens.go create mode 100644 server/internal/auth/tokens_test.go create mode 100644 server/internal/realtime/client.go create mode 100644 server/internal/realtime/events.go create mode 100644 server/internal/realtime/hub.go create mode 100644 server/internal/realtime/hub_test.go create mode 100644 server/internal/store/contacts.go create mode 100644 server/internal/store/conversations.go create mode 100644 server/internal/store/errors.go create mode 100644 server/internal/store/integration_test.go create mode 100644 server/internal/store/messages.go create mode 100644 server/internal/store/migrate.go create mode 100644 server/internal/store/migrations/0001_init.sql create mode 100644 server/internal/store/sessions.go create mode 100644 server/internal/store/store.go create mode 100644 server/internal/store/users.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3bc69f7..39b6499 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -9,6 +9,22 @@ jobs: server: name: Go (build + vet + test) runs-on: ubuntu-latest + services: + postgres: + image: postgres:17-alpine + env: + POSTGRES_USER: messenger + POSTGRES_PASSWORD: messenger + POSTGRES_DB: messenger + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U messenger" + --health-interval 5s + --health-timeout 3s + --health-retries 10 + env: + MESSENGER_TEST_DATABASE_URL: postgres://messenger:messenger@localhost:5432/messenger?sslmode=disable defaults: run: working-directory: server diff --git a/README.md b/README.md index d84479f..9748b24 100644 --- a/README.md +++ b/README.md @@ -3,19 +3,51 @@ Self-hosted messenger for family and friends. Built from scratch: Go backend, React PWA frontend, PostgreSQL, WebRTC calls. Deployed on a home-lab mini PC. -> **Status:** Phase 0 — project skeleton. No functionality yet, but everything -> builds, tests pass, and the deployment shape is in place. +> **Status:** Phase 1 — auth (JWT + refresh in httpOnly cookies, argon2id), +> contacts, direct-message chats, real-time delivery over WebSocket +> (messages, typing, presence, read receipts). Backend is complete and tested; +> the frontend still shows the Phase 0 status screen. ## Repository layout ``` -server/ Go API — config, logging, HTTP server, health endpoints +server/ Go API + cmd/api/ entrypoint (wiring, migrations, graceful shutdown) + internal/api/ HTTP handlers: auth, users, contacts, conversations, messages + internal/auth/ argon2id hashing, JWT access tokens, refresh tokens + internal/realtime/ WebSocket hub: events, presence, typing, read receipts + internal/store/ PostgreSQL access (pgx), embedded migrations + internal/platform/ config, structured logging web/ React + TypeScript + Vite frontend (PWA-ready) deploy/ docker-compose and operational config docs/ architecture, ADR, OpenAPI .github/ CI workflows ``` +## API (Phase 1) + +All endpoints under `/api/v1`. Auth via httpOnly cookies (`messenger_access` += JWT 15 min, `messenger_refresh` = opaque token 30 days, rotated on refresh). + +| Method | Path | Description | +|--------|-----------------------------------|---------------------------------| +| POST | `/api/v1/auth/register` | Create account (sets cookies) | +| POST | `/api/v1/auth/login` | Sign in (sets cookies) | +| POST | `/api/v1/auth/refresh` | Rotate refresh token | +| POST | `/api/v1/auth/logout` | Revoke session, clear cookies | +| GET | `/api/v1/users/me` | Current user | +| GET | `/api/v1/users?q=` | Search users by username/name | +| GET | `/api/v1/contacts` | Contact list | +| POST | `/api/v1/contacts` | `{user_id}` — add contact | +| DELETE | `/api/v1/contacts/{user_id}` | Remove contact | +| GET | `/api/v1/conversations` | List chats + unread counts | +| POST | `/api/v1/conversations` | `{user_id}` — create/find DM | +| GET | `/api/v1/conversations/{id}/messages` | History, keyset `?before=` | +| POST | `/api/v1/conversations/{id}/messages` | `{body, temp_id}` — send | +| GET | `/ws` | WebSocket (auth via cookie) | + +WebSocket event types: `message.created`, `typing`, `presence`, `read`. + ## Prerequisites - Go 1.27+ @@ -24,8 +56,11 @@ docs/ architecture, ADR, OpenAPI ## Local development +Requires a local PostgreSQL (e.g. `sudo -u postgres psql -c "CREATE ROLE messenger LOGIN PASSWORD 'messenger';"` and +`sudo -u postgres psql -c "CREATE DATABASE messenger OWNER messenger;"`). + ```bash -# Backend +# Backend (migrations run automatically on start) cd server go run ./cmd/api # listen on :8080 @@ -58,8 +93,9 @@ frontend (`/app/web`). PostgreSQL runs alongside with a healthcheck. ## Roadmap -- **Phase 0** — skeleton, CI, Docker (current) -- Phase 1 — auth (JWT + refresh in httpOnly cookie, argon2id), contacts, DMs, WebSocket delivery, receipts, typing +- **Phase 0** — skeleton, CI, Docker ✅ +- **Phase 1** — auth (JWT + refresh in httpOnly cookie, argon2id), contacts, DMs, WebSocket delivery, receipts, typing ✅ (backend) +- Phase 1.5 — frontend: auth screens, chat UI, realtime client (PWA) - Phase 2 — groups, reactions, media upload + thumbnails (S3-like storage interface) - Phase 3 — search (pg_trgm), Web Push (VAPID), PWA offline - Phase 4 — P2P WebRTC calls via our WebSocket signalling + coturn TURN fallback diff --git a/deploy/.env.example b/deploy/.env.example index e6ab23f..d677ea1 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -2,6 +2,7 @@ MESSENGER_ENV=production MESSENGER_ALLOWED_ORIGINS=https://msg.example.com +MESSENGER_JWT_SECRET=generate-a-long-random-string POSTGRES_USER=messenger POSTGRES_PASSWORD=change-me diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index a8d93cd..e82985f 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -9,6 +9,8 @@ services: MESSENGER_ENV: ${MESSENGER_ENV:-production} MESSENGER_HTTP_ADDR: ":8080" MESSENGER_ALLOWED_ORIGINS: ${MESSENGER_ALLOWED_ORIGINS:-} + MESSENGER_DATABASE_URL: postgres://${POSTGRES_USER:-messenger}:${POSTGRES_PASSWORD:-messenger}@postgres:5432/${POSTGRES_DB:-messenger}?sslmode=disable + MESSENGER_JWT_SECRET: ${MESSENGER_JWT_SECRET:?MESSENGER_JWT_SECRET must be set in .env} ports: - "8080:8080" depends_on: diff --git a/server/cmd/api/main.go b/server/cmd/api/main.go index 85b1c74..52f01c4 100644 --- a/server/cmd/api/main.go +++ b/server/cmd/api/main.go @@ -13,9 +13,14 @@ import ( "syscall" "time" + "messenger/internal/auth" "messenger/internal/platform/config" "messenger/internal/platform/logger" + "messenger/internal/realtime" "messenger/internal/server" + "messenger/internal/store" + + "github.com/jackc/pgx/v5/pgxpool" ) func main() { @@ -27,7 +32,27 @@ func main() { os.Exit(1) } - httpServer := server.New(cfg, log) + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + pool, err := pgxpool.New(ctx, cfg.DatabaseURL) + if err != nil { + log.Error("failed to create db pool", "error", err) + os.Exit(1) + } + defer pool.Close() + + st := store.New(pool) + if err := st.Migrate(ctx); err != nil { + log.Error("database migration failed", "error", err) + os.Exit(1) + } + log.Info("database ready") + + tm := auth.NewTokenManager(cfg.JWTSecret) + hub := realtime.NewHub(st) + + httpServer := server.New(cfg, log, st, tm, hub) // Run the server in a goroutine so that shutdown can be triggered // from the signal handler below. @@ -44,9 +69,6 @@ func main() { }() // Wait for a termination signal, then drain in-flight requests. - ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) - defer stop() - <-ctx.Done() log.Info("shutting down") diff --git a/server/go.mod b/server/go.mod index 7d1fc78..6118d8b 100644 --- a/server/go.mod +++ b/server/go.mod @@ -1,3 +1,24 @@ module messenger go 1.27 + +require ( + github.com/coder/websocket v1.8.15 + github.com/golang-jwt/jwt/v5 v5.3.1 + github.com/jackc/pgconn v1.14.3 + github.com/jackc/pgerrcode v0.0.0-20250907135507-afb5586c32a6 + github.com/jackc/pgx/v5 v5.11.0 + golang.org/x/crypto v0.57.0 +) + +require ( + github.com/jackc/chunkreader/v2 v2.0.1 // indirect + github.com/jackc/pgio v1.0.0 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgproto3/v2 v2.3.3 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + golang.org/x/sync v0.23.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/text v0.42.0 // indirect +) diff --git a/server/go.sum b/server/go.sum index e69de29..250ce2c 100644 --- a/server/go.sum +++ b/server/go.sum @@ -0,0 +1,49 @@ +github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= +github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= +github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= +github.com/jackc/chunkreader/v2 v2.0.0/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk= +github.com/jackc/chunkreader/v2 v2.0.1 h1:i+RDz65UE+mmpjTfyz0MoVTnzeYxroil2G82ki7MGG8= +github.com/jackc/chunkreader/v2 v2.0.1/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk= +github.com/jackc/pgconn v1.14.3 h1:bVoTr12EGANZz66nZPkMInAV/KHD2TxH9npjXXgiB3w= +github.com/jackc/pgconn v1.14.3/go.mod h1:RZbme4uasqzybK2RK5c65VsHxoyaml09lx3tXOcO/VM= +github.com/jackc/pgerrcode v0.0.0-20250907135507-afb5586c32a6 h1:D/V0gu4zQ3cL2WKeVNVM4r2gLxGGf6McLwgXzRTo2RQ= +github.com/jackc/pgerrcode v0.0.0-20250907135507-afb5586c32a6/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds= +github.com/jackc/pgio v1.0.0 h1:g12B9UwVnzGhueNavwioyEEpAmqMe1E/BN9ES+8ovkE= +github.com/jackc/pgio v1.0.0/go.mod h1:oP+2QK2wFfUWgr+gxjoBH9KGBb31Eio69xUb0w5bYf8= +github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65 h1:DadwsjnMwFjfWc9y5Wi/+Zz7xoE5ALHsRQlOctkOiHc= +github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65/go.mod h1:5R2h2EEX+qri8jOWMbJCtaPWkrrNc7OHwsp2TCqp7ak= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgproto3/v2 v2.3.3 h1:1HLSx5H+tXR9pW3in3zaztoEwQYRC9SQaYUHjTSUOag= +github.com/jackc/pgproto3/v2 v2.3.3/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg= +github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/server/internal/api/api.go b/server/internal/api/api.go new file mode 100644 index 0000000..6b0a887 --- /dev/null +++ b/server/internal/api/api.go @@ -0,0 +1,67 @@ +package api + +import ( + "log/slog" + "net/http" + + "messenger/internal/auth" + "messenger/internal/platform/config" + "messenger/internal/realtime" + "messenger/internal/store" +) + +type apiHandler struct { + store *store.Store + tokens *auth.TokenManager + hub *realtime.Hub + log *slog.Logger + secure bool +} + +// NewHandler builds the /api/v1 route tree with all middleware applied. +func NewHandler(st *store.Store, tm *auth.TokenManager, hub *realtime.Hub, log *slog.Logger, cfg *config.Config) http.Handler { + h := &apiHandler{ + store: st, + tokens: tm, + hub: hub, + log: log, + secure: cfg.Environment == "production", + } + + mux := http.NewServeMux() + authHandler := requireAuth(tm, http.HandlerFunc(h.websocket)) + + mux.HandleFunc("GET /api/v1/ping", func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]string{"message": "pong"}) + }) + + mux.HandleFunc("POST /api/v1/auth/register", h.register) + mux.HandleFunc("POST /api/v1/auth/login", h.login) + mux.HandleFunc("POST /api/v1/auth/refresh", h.refresh) + mux.HandleFunc("POST /api/v1/auth/logout", h.logout) + + mux.Handle("GET /api/v1/users/me", requireAuth(tm, h.me)) + mux.Handle("GET /api/v1/users", requireAuth(tm, h.searchUsers)) + + mux.Handle("GET /api/v1/contacts", requireAuth(tm, h.listContacts)) + mux.Handle("POST /api/v1/contacts", requireAuth(tm, h.addContact)) + mux.Handle("DELETE /api/v1/contacts/{user_id}", requireAuth(tm, h.removeContact)) + + mux.Handle("GET /api/v1/conversations", requireAuth(tm, h.listConversations)) + mux.Handle("POST /api/v1/conversations", requireAuth(tm, h.createDM)) + mux.Handle("GET /api/v1/conversations/{id}/messages", requireAuth(tm, h.listMessages)) + mux.Handle("POST /api/v1/conversations/{id}/messages", requireAuth(tm, h.sendMessage)) + + mux.Handle("GET /ws", authHandler) + + return mux +} + +func (h *apiHandler) websocket(w http.ResponseWriter, r *http.Request) { + userID, ok := userIDFrom(r.Context()) + if !ok { + writeError(w, http.StatusUnauthorized, "unauthenticated") + return + } + h.hub.HandleWS(w, r, userID) +} diff --git a/server/internal/api/auth.go b/server/internal/api/auth.go new file mode 100644 index 0000000..721fec6 --- /dev/null +++ b/server/internal/api/auth.go @@ -0,0 +1,172 @@ +package api + +import ( + "encoding/json" + "errors" + "net/http" + "regexp" + "time" + + "messenger/internal/auth" + "messenger/internal/store" +) + +var usernameRe = regexp.MustCompile(`^[a-z0-9_]{3,32}$`) + +type userJSON struct { + ID int64 `json:"id"` + Username string `json:"username"` + DisplayName string `json:"display_name"` + Avatar string `json:"avatar"` + CreatedAt time.Time `json:"created_at"` +} + +func toUserJSON(u *store.User) userJSON { + return userJSON{ + ID: u.ID, + Username: u.Username, + DisplayName: u.DisplayName, + Avatar: u.Avatar, + CreatedAt: u.CreatedAt, + } +} + +type credentialsRequest struct { + Username string `json:"username"` + DisplayName string `json:"display_name"` + Password string `json:"password"` +} + +func (h *apiHandler) register(w http.ResponseWriter, r *http.Request) { + var req credentialsRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid json") + return + } + if !usernameRe.MatchString(req.Username) { + writeError(w, http.StatusBadRequest, "username must be 3-32 chars of [a-z0-9_]") + return + } + if req.DisplayName == "" || len(req.DisplayName) > 64 { + writeError(w, http.StatusBadRequest, "display_name must be 1-64 chars") + return + } + if len(req.Password) < 8 || len(req.Password) > 128 { + writeError(w, http.StatusBadRequest, "password must be 8-128 chars") + return + } + + hash, err := auth.HashPassword(req.Password) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + user, err := h.store.CreateUser(r.Context(), req.Username, req.DisplayName, hash) + if err != nil { + if errors.Is(err, store.ErrDuplicate) { + writeError(w, http.StatusConflict, "username already taken") + return + } + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + h.issueSession(w, r, user.ID) + writeJSON(w, http.StatusCreated, toUserJSON(user)) +} + +func (h *apiHandler) login(w http.ResponseWriter, r *http.Request) { + var req credentialsRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid json") + return + } + user, err := h.store.GetUserByUsername(r.Context(), req.Username) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + writeError(w, http.StatusUnauthorized, "invalid credentials") + return + } + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + ok, err := auth.VerifyPassword(user.PasswordHash, req.Password) + if err != nil || !ok { + writeError(w, http.StatusUnauthorized, "invalid credentials") + return + } + h.issueSession(w, r, user.ID) + writeJSON(w, http.StatusOK, toUserJSON(user)) +} + +func (h *apiHandler) refresh(w http.ResponseWriter, r *http.Request) { + cookie, err := r.Cookie(refreshCookieName) + if err != nil { + writeError(w, http.StatusUnauthorized, "missing refresh token") + return + } + session, err := h.store.GetSessionByRefreshHash(r.Context(), auth.HashRefreshToken(cookie.Value)) + if err != nil { + writeError(w, http.StatusUnauthorized, "invalid refresh token") + return + } + if session.RevokedAt != nil { + writeError(w, http.StatusUnauthorized, "session revoked") + return + } + if time.Now().After(session.ExpiresAt) { + writeError(w, http.StatusUnauthorized, "session expired") + return + } + + // Rotation: revoke the used session and issue a fresh pair. + if err := h.store.RevokeSession(r.Context(), session.ID); err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + h.issueSession(w, r, session.UserID) + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) +} + +func (h *apiHandler) logout(w http.ResponseWriter, r *http.Request) { + if cookie, err := r.Cookie(refreshCookieName); err == nil { + session, err := h.store.GetSessionByRefreshHash(r.Context(), auth.HashRefreshToken(cookie.Value)) + if err == nil { + _ = h.store.RevokeSession(r.Context(), session.ID) + } + } + clearAuthCookies(w) + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) +} + +func (h *apiHandler) me(w http.ResponseWriter, r *http.Request) { + userID, ok := userIDFrom(r.Context()) + if !ok { + writeError(w, http.StatusUnauthorized, "unauthenticated") + return + } + user, err := h.store.GetUserByID(r.Context(), userID) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + writeJSON(w, http.StatusOK, toUserJSON(user)) +} + +func (h *apiHandler) issueSession(w http.ResponseWriter, r *http.Request, userID int64) { + access, err := h.tokens.IssueAccessToken(userID, accessTTL) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + rawRefresh, refreshHash, err := auth.GenerateRefreshToken() + if err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + _, err = h.store.CreateSession(r.Context(), userID, refreshHash, r.UserAgent(), r.RemoteAddr, time.Now().Add(refreshTTL)) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + setAuthCookies(w, access, rawRefresh, h.secure) +} diff --git a/server/internal/api/contacts.go b/server/internal/api/contacts.go new file mode 100644 index 0000000..4332e3e --- /dev/null +++ b/server/internal/api/contacts.go @@ -0,0 +1,93 @@ +package api + +import ( + "encoding/json" + "errors" + "net/http" + "strconv" + + "messenger/internal/store" +) + +func (h *apiHandler) searchUsers(w http.ResponseWriter, r *http.Request) { + query := r.URL.Query().Get("q") + users, err := h.store.SearchUsers(r.Context(), query, 20) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + out := make([]userJSON, 0, len(users)) + for i := range users { + out = append(out, toUserJSON(&users[i])) + } + writeJSON(w, http.StatusOK, out) +} + +type contactJSON struct { + User userJSON `json:"user"` + AddedAt string `json:"added_at"` +} + +func (h *apiHandler) listContacts(w http.ResponseWriter, r *http.Request) { + userID, _ := userIDFrom(r.Context()) + contacts, err := h.store.ListContacts(r.Context(), userID) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + out := make([]contactJSON, 0, len(contacts)) + for _, c := range contacts { + u, err := h.store.GetUserByID(r.Context(), c.ContactID) + if err != nil { + continue + } + out = append(out, contactJSON{User: toUserJSON(u), AddedAt: c.CreatedAt.Format("2006-01-02T15:04:05Z07:00")}) + } + writeJSON(w, http.StatusOK, out) +} + +func (h *apiHandler) addContact(w http.ResponseWriter, r *http.Request) { + userID, _ := userIDFrom(r.Context()) + var req struct { + UserID int64 `json:"user_id"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.UserID <= 0 { + writeError(w, http.StatusBadRequest, "invalid user_id") + return + } + if req.UserID == userID { + writeError(w, http.StatusBadRequest, "cannot add yourself") + return + } + if _, err := h.store.GetUserByID(r.Context(), req.UserID); err != nil { + if errors.Is(err, store.ErrNotFound) { + writeError(w, http.StatusNotFound, "user not found") + return + } + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + if err := h.store.AddContact(r.Context(), userID, req.UserID); err != nil { + if errors.Is(err, store.ErrDuplicate) { + writeJSON(w, http.StatusOK, map[string]string{"status": "already a contact"}) + return + } + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + writeJSON(w, http.StatusCreated, map[string]string{"status": "ok"}) +} + +func (h *apiHandler) removeContact(w http.ResponseWriter, r *http.Request) { + userID, _ := userIDFrom(r.Context()) + contactID, err := strconv.ParseInt(r.PathValue("user_id"), 10, 64) + if err != nil || contactID <= 0 { + writeError(w, http.StatusBadRequest, "invalid user_id") + return + } + if err := h.store.RemoveContact(r.Context(), userID, contactID); err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) +} diff --git a/server/internal/api/conversations.go b/server/internal/api/conversations.go new file mode 100644 index 0000000..61e3d7d --- /dev/null +++ b/server/internal/api/conversations.go @@ -0,0 +1,172 @@ +package api + +import ( + "encoding/json" + "net/http" + "strconv" + "time" + + "messenger/internal/realtime" + "messenger/internal/store" +) + +type conversationJSON struct { + ID int64 `json:"id"` + Type string `json:"type"` + CreatedAt time.Time `json:"created_at"` + LastMessage *messageJSON `json:"last_message,omitempty"` + UnreadCount int64 `json:"unread_count"` +} + +type messageJSON struct { + ID int64 `json:"id"` + ConversationID int64 `json:"conversation_id"` + SenderID int64 `json:"sender_id"` + Body string `json:"body"` + CreatedAt time.Time `json:"created_at"` + EditedAt *time.Time `json:"edited_at,omitempty"` + DeletedAt *time.Time `json:"deleted_at,omitempty"` +} + +func toMessageJSON(m *store.Message) messageJSON { + return messageJSON{ + ID: m.ID, + ConversationID: m.ConversationID, + SenderID: m.SenderID, + Body: m.Body, + CreatedAt: m.CreatedAt, + EditedAt: m.EditedAt, + DeletedAt: m.DeletedAt, + } +} + +func (h *apiHandler) listConversations(w http.ResponseWriter, r *http.Request) { + userID, _ := userIDFrom(r.Context()) + summaries, err := h.store.ListConversationsForUser(r.Context(), userID) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + out := make([]conversationJSON, 0, len(summaries)) + for _, s := range summaries { + item := conversationJSON{ + ID: s.ID, + Type: s.Type, + CreatedAt: s.CreatedAt, + UnreadCount: s.UnreadCount, + } + if s.LastMessage != nil { + msg := toMessageJSON(s.LastMessage) + item.LastMessage = &msg + } + out = append(out, item) + } + writeJSON(w, http.StatusOK, out) +} + +func (h *apiHandler) createDM(w http.ResponseWriter, r *http.Request) { + userID, _ := userIDFrom(r.Context()) + var req struct { + UserID int64 `json:"user_id"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.UserID <= 0 { + writeError(w, http.StatusBadRequest, "invalid user_id") + return + } + if req.UserID == userID { + writeError(w, http.StatusBadRequest, "cannot chat with yourself") + return + } + conv, err := h.store.CreateDM(r.Context(), userID, req.UserID) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + writeJSON(w, http.StatusCreated, conversationJSON{ID: conv.ID, Type: conv.Type, CreatedAt: conv.CreatedAt}) +} + +func (h *apiHandler) listMessages(w http.ResponseWriter, r *http.Request) { + userID, _ := userIDFrom(r.Context()) + convID, err := strconv.ParseInt(r.PathValue("id"), 10, 64) + if err != nil || convID <= 0 { + writeError(w, http.StatusBadRequest, "invalid conversation id") + return + } + member, err := h.store.IsMember(r.Context(), convID, userID) + if err != nil || !member { + writeError(w, http.StatusForbidden, "not a member") + return + } + + limit := 50 + if v := r.URL.Query().Get("limit"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 100 { + limit = n + } + } + var before int64 + if v := r.URL.Query().Get("before"); v != "" { + before, _ = strconv.ParseInt(v, 10, 64) + } + + messages, err := h.store.ListMessages(r.Context(), convID, before, limit) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + out := make([]messageJSON, 0, len(messages)) + for i := range messages { + out = append(out, toMessageJSON(&messages[i])) + } + writeJSON(w, http.StatusOK, out) +} + +type sendMessageRequest struct { + Body string `json:"body"` + TempID string `json:"temp_id"` +} + +func (h *apiHandler) sendMessage(w http.ResponseWriter, r *http.Request) { + userID, _ := userIDFrom(r.Context()) + convID, err := strconv.ParseInt(r.PathValue("id"), 10, 64) + if err != nil || convID <= 0 { + writeError(w, http.StatusBadRequest, "invalid conversation id") + return + } + member, err := h.store.IsMember(r.Context(), convID, userID) + if err != nil || !member { + writeError(w, http.StatusForbidden, "not a member") + return + } + + var req sendMessageRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid json") + return + } + if req.Body == "" { + writeError(w, http.StatusBadRequest, "body is required") + return + } + if len(req.Body) > 4000 { + writeError(w, http.StatusBadRequest, "body too long (max 4000)") + return + } + + msg, err := h.store.InsertMessage(r.Context(), convID, userID, req.Body) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal server error") + return + } + + h.hub.PublishMessageCreated(convID, realtime.MessageCreated{ + TempID: req.TempID, + ConversationID: convID, + MessageID: msg.ID, + SenderID: userID, + Body: msg.Body, + CreatedAt: msg.CreatedAt, + }) + + writeJSON(w, http.StatusCreated, toMessageJSON(msg)) +} diff --git a/server/internal/api/middleware.go b/server/internal/api/middleware.go new file mode 100644 index 0000000..45fcef4 --- /dev/null +++ b/server/internal/api/middleware.go @@ -0,0 +1,74 @@ +package api + +import ( + "context" + "net/http" + "time" + + "messenger/internal/auth" +) + +const ( + accessCookieName = "messenger_access" + refreshCookieName = "messenger_refresh" + accessTTL = 15 * time.Minute + refreshTTL = 30 * 24 * time.Hour +) + +type ctxKey string + +// ctxKeyUserID is the context key carrying the authenticated user ID. +const ctxKeyUserID ctxKey = "messenger.userID" + +func userIDFrom(ctx context.Context) (int64, bool) { + id, ok := ctx.Value(ctxKeyUserID).(int64) + return id, ok +} + +func withUserID(ctx context.Context, userID int64) context.Context { + return context.WithValue(ctx, ctxKeyUserID, userID) +} + +// requireAuth parses the access-token cookie and injects the user ID +// into the request context. +func requireAuth(tm *auth.TokenManager, next http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + cookie, err := r.Cookie(accessCookieName) + if err != nil { + writeError(w, http.StatusUnauthorized, "missing access token") + return + } + userID, err := tm.ParseAccessToken(cookie.Value) + if err != nil { + writeError(w, http.StatusUnauthorized, "invalid access token") + return + } + next(w, r.WithContext(withUserID(r.Context(), userID))) + } +} + +func setAuthCookies(w http.ResponseWriter, accessToken, refreshToken string, secure bool) { + http.SetCookie(w, &http.Cookie{ + Name: accessCookieName, + Value: accessToken, + Path: "/", + HttpOnly: true, + Secure: secure, + SameSite: http.SameSiteLaxMode, + MaxAge: int(accessTTL.Seconds()), + }) + http.SetCookie(w, &http.Cookie{ + Name: refreshCookieName, + Value: refreshToken, + Path: "/", + HttpOnly: true, + Secure: secure, + SameSite: http.SameSiteLaxMode, + MaxAge: int(refreshTTL.Seconds()), + }) +} + +func clearAuthCookies(w http.ResponseWriter) { + http.SetCookie(w, &http.Cookie{Name: accessCookieName, Path: "/", MaxAge: -1}) + http.SetCookie(w, &http.Cookie{Name: refreshCookieName, Path: "/", MaxAge: -1}) +} diff --git a/server/internal/api/respond.go b/server/internal/api/respond.go new file mode 100644 index 0000000..dab93d7 --- /dev/null +++ b/server/internal/api/respond.go @@ -0,0 +1,33 @@ +// Package api implements the HTTP handlers for the messenger REST API. +package api + +import ( + "encoding/json" + "errors" + "net/http" + + "messenger/internal/store" +) + +func writeJSON(w http.ResponseWriter, status int, body any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + if err := json.NewEncoder(w).Encode(body); err != nil { + http.Error(w, `{"error":"failed to encode response"}`, http.StatusInternalServerError) + } +} + +func writeError(w http.ResponseWriter, status int, message string) { + writeJSON(w, status, map[string]string{"error": message}) +} + +func writeStoreError(w http.ResponseWriter, err error) { + switch { + case errors.Is(err, store.ErrNotFound): + writeError(w, http.StatusNotFound, "not found") + case errors.Is(err, store.ErrDuplicate): + writeError(w, http.StatusConflict, "already exists") + default: + writeError(w, http.StatusInternalServerError, "internal server error") + } +} diff --git a/server/internal/auth/password.go b/server/internal/auth/password.go new file mode 100644 index 0000000..b8c4022 --- /dev/null +++ b/server/internal/auth/password.go @@ -0,0 +1,103 @@ +// Package auth provides cryptographic core for authentication. +package auth + +import ( + "crypto/rand" + "crypto/subtle" + "encoding/base64" + "errors" + "fmt" + "strconv" + "strings" + + "golang.org/x/crypto/argon2" +) + +// argon2Memory, argon2Time, argon2Threads, argon2KeyLen, and saltLen are parameters +// recommended by OWASP for Argon2id (memory=65536 KiB, time=3 iterations, parallelism=2). +const ( + argon2Memory = 65536 + argon2Time = 3 + argon2Threads = 2 + argon2KeyLen = 32 + saltLen = 16 +) + +// HashPassword generates an Argon2id hash of the provided password and returns it +// in PHC string format: $argon2id$v=19$m=...,t=...,p=...$$. +func HashPassword(password string) (string, error) { + salt := make([]byte, saltLen) + if _, err := rand.Read(salt); err != nil { + return "", fmt.Errorf("auth: generate salt: %w", err) + } + hash := argon2.IDKey([]byte(password), salt, argon2Time, argon2Memory, argon2Threads, argon2KeyLen) + b64Salt := base64.RawStdEncoding.EncodeToString(salt) + b64Hash := base64.RawStdEncoding.EncodeToString(hash) + phc := fmt.Sprintf("$argon2id$v=%d$m=%d,t=%d,p=%d$%s$%s", argon2.Version, argon2Memory, argon2Time, argon2Threads, b64Salt, b64Hash) + return phc, nil +} + +// VerifyPassword checks whether the provided password matches the given Argon2id PHC hash. +// It returns (true, nil) on match, (false, nil) on mismatch, or (false, error) on invalid format. +func VerifyPassword(phc, password string) (bool, error) { + parts := strings.Split(phc, "$") + if len(parts) != 6 || parts[1] != "argon2id" { + return false, errors.New("auth: invalid hash format") + } + // Version + vPart := parts[2] + if !strings.HasPrefix(vPart, "v=") { + return false, errors.New("auth: invalid version format") + } + v, err := strconv.Atoi(strings.TrimPrefix(vPart, "v=")) + if err != nil { + return false, fmt.Errorf("auth: parse version: %w", err) + } + if v != argon2.Version { + return false, errors.New("auth: incompatible version") + } + // Parameters + mem, t, p := uint32(0), uint32(0), uint32(0) + for _, kv := range strings.Split(parts[3], ",") { + pair := strings.SplitN(kv, "=", 2) + if len(pair) != 2 { + return false, errors.New("auth: invalid parameter format") + } + key, val := pair[0], pair[1] + switch key { + case "m": + x, err := strconv.ParseUint(val, 10, 32) + if err != nil { + return false, fmt.Errorf("auth: parse memory: %w", err) + } + mem = uint32(x) + case "t": + x, err := strconv.ParseUint(val, 10, 32) + if err != nil { + return false, fmt.Errorf("auth: parse time: %w", err) + } + t = uint32(x) + case "p": + x, err := strconv.ParseUint(val, 10, 32) + if err != nil { + return false, fmt.Errorf("auth: parse threads: %w", err) + } + p = uint32(x) + default: + return false, fmt.Errorf("auth: unknown parameter %q", key) + } + } + salt, err := base64.RawStdEncoding.DecodeString(parts[4]) + if err != nil { + return false, fmt.Errorf("auth: decode salt: %w", err) + } + expected, err := base64.RawStdEncoding.DecodeString(parts[5]) + if err != nil { + return false, fmt.Errorf("auth: decode hash: %w", err) + } + computed := argon2.IDKey([]byte(password), salt, t, mem, uint8(p), uint32(len(expected))) + if subtle.ConstantTimeCompare(computed, expected) == 1 { + return true, nil + } + return false, nil +} diff --git a/server/internal/auth/password_test.go b/server/internal/auth/password_test.go new file mode 100644 index 0000000..b9abb6e --- /dev/null +++ b/server/internal/auth/password_test.go @@ -0,0 +1,49 @@ +// Package auth_test tests password hashing and verification. +package auth_test + +import ( + "testing" + + "messenger/internal/auth" +) + +func TestHashAndVerify(t *testing.T) { + pwd := "secret" + hash, err := auth.HashPassword(pwd) + if err != nil { + t.Fatalf("HashPassword error: %v", err) + } + ok, err := auth.VerifyPassword(hash, pwd) + if err != nil { + t.Fatalf("VerifyPassword error: %v", err) + } + if !ok { + t.Fatal("expected password to verify") + } +} + +func TestWrongPassword(t *testing.T) { + hash, _ := auth.HashPassword("correct") + ok, err := auth.VerifyPassword(hash, "wrong") + if err != nil { + t.Fatalf("VerifyPassword error on wrong password: %v", err) + } + if ok { + t.Fatal("wrong password should not verify") + } +} + +func TestInvalidHashFormat(t *testing.T) { + _, err := auth.VerifyPassword("invalidformat", "pwd") + if err == nil { + t.Fatal("expected error on invalid hash format") + } +} + +func TestDifferentSalts(t *testing.T) { + h1, _ := auth.HashPassword("pass") + h2, _ := auth.HashPassword("pass") + if h1 == h2 { + t.Fatal("expected different hashes for same password due to salt") + } +} diff --git a/server/internal/auth/refresh.go b/server/internal/auth/refresh.go new file mode 100644 index 0000000..137a3d0 --- /dev/null +++ b/server/internal/auth/refresh.go @@ -0,0 +1,27 @@ +// Package auth provides cryptographic core for authentication. +package auth + +import ( + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "fmt" +) + +// GenerateRefreshToken generates a new opaque refresh token (raw) and its SHA256 hash. +// The raw token is 32 random bytes hex-encoded; the hash is hex(SHA256(raw)). +func GenerateRefreshToken() (raw, hash string, err error) { + b := make([]byte, 32) + if _, err = rand.Read(b); err != nil { + return "", "", fmt.Errorf("auth: generate refresh token: %w", err) + } + raw = hex.EncodeToString(b) + hash = HashRefreshToken(raw) + return raw, hash, nil +} + +// HashRefreshToken returns the SHA256 hash (hex) of the raw refresh token. +func HashRefreshToken(raw string) string { + sum := sha256.Sum256([]byte(raw)) + return hex.EncodeToString(sum[:]) +} diff --git a/server/internal/auth/refresh_test.go b/server/internal/auth/refresh_test.go new file mode 100644 index 0000000..ca5485f --- /dev/null +++ b/server/internal/auth/refresh_test.go @@ -0,0 +1,39 @@ +// Package auth_test tests refresh token generation and hashing. +package auth_test + +import ( + "testing" + + "messenger/internal/auth" +) + +func TestGenerateAndHash(t *testing.T) { + raw, hash, err := auth.GenerateRefreshToken() + if err != nil { + t.Fatalf("GenerateRefreshToken error: %v", err) + } + if raw == hash { + t.Fatal("raw and hash should differ") + } + if auth.HashRefreshToken(raw) != hash { + t.Fatal("HashRefreshToken mismatch") + } +} + +func TestGenerateRandomness(t *testing.T) { + r1, _, _ := auth.GenerateRefreshToken() + r2, _, _ := auth.GenerateRefreshToken() + if r1 == r2 { + t.Fatal("expected two different raw tokens") + } +} + +func TestRawLength(t *testing.T) { + raw, _, err := auth.GenerateRefreshToken() + if err != nil { + t.Fatalf("GenerateRefreshToken error: %v", err) + } + if len(raw) != 64 { + t.Fatalf("expected raw length 64, got %d", len(raw)) + } +} diff --git a/server/internal/auth/tokens.go b/server/internal/auth/tokens.go new file mode 100644 index 0000000..88a56c1 --- /dev/null +++ b/server/internal/auth/tokens.go @@ -0,0 +1,57 @@ +// Package auth provides cryptographic core for authentication. +package auth + +import ( + "errors" + "strconv" + "time" + + "github.com/golang-jwt/jwt/v5" +) + +// ErrInvalidToken is returned when a token is invalid or cannot be parsed. +var ErrInvalidToken = errors.New("auth: invalid token") + +// TokenManager handles issuing and parsing JWT access tokens. +type TokenManager struct { + secret []byte +} + +// NewTokenManager creates a TokenManager using the provided secret key. +func NewTokenManager(secret string) *TokenManager { + return &TokenManager{secret: []byte(secret)} +} + +// IssueAccessToken issues a signed JWT access token for the given userID with TTL. +func (t *TokenManager) IssueAccessToken(userID int64, ttl time.Duration) (string, error) { + now := time.Now() + claims := jwt.RegisteredClaims{ + Subject: strconv.FormatInt(userID, 10), + IssuedAt: jwt.NewNumericDate(now), + ExpiresAt: jwt.NewNumericDate(now.Add(ttl)), + } + token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims) + return token.SignedString(t.secret) +} + +// ParseAccessToken parses and validates the JWT access token, returning the userID. +func (t *TokenManager) ParseAccessToken(tokenStr string) (int64, error) { + token, err := jwt.ParseWithClaims(tokenStr, &jwt.RegisteredClaims{}, func(token *jwt.Token) (interface{}, error) { + if token.Method != jwt.SigningMethodHS256 { + return nil, ErrInvalidToken + } + return t.secret, nil + }) + if err != nil { + return 0, ErrInvalidToken + } + claims, ok := token.Claims.(*jwt.RegisteredClaims) + if !ok || !token.Valid { + return 0, ErrInvalidToken + } + userID, err := strconv.ParseInt(claims.Subject, 10, 64) + if err != nil { + return 0, ErrInvalidToken + } + return userID, nil +} diff --git a/server/internal/auth/tokens_test.go b/server/internal/auth/tokens_test.go new file mode 100644 index 0000000..12c141d --- /dev/null +++ b/server/internal/auth/tokens_test.go @@ -0,0 +1,78 @@ +// Package auth_test tests access token issuance and parsing. +package auth_test + +import ( + "errors" + "testing" + "time" + + "github.com/golang-jwt/jwt/v5" + "messenger/internal/auth" +) + +func TestIssueParseRoundTrip(t *testing.T) { + secret := "0123456789abcdef0123456789abcdef" + tm := auth.NewTokenManager(secret) + token, err := tm.IssueAccessToken(42, time.Minute) + if err != nil { + t.Fatalf("IssueAccessToken error: %v", err) + } + uid, err := tm.ParseAccessToken(token) + if err != nil { + t.Fatalf("ParseAccessToken error: %v", err) + } + if uid != 42 { + t.Fatalf("expected userID 42, got %d", uid) + } +} + +func TestInvalidSignature(t *testing.T) { + tm1 := auth.NewTokenManager("secret1secret1secret1secret1sec1") + tm2 := auth.NewTokenManager("secret2secret2secret2secret2sec2") + token, err := tm1.IssueAccessToken(1, time.Minute) + if err != nil { + t.Fatalf("IssueAccessToken error: %v", err) + } + _, err = tm2.ParseAccessToken(token) + if !errors.Is(err, auth.ErrInvalidToken) { + t.Fatalf("expected ErrInvalidToken, got %v", err) + } +} + +func TestExpiredToken(t *testing.T) { + tm := auth.NewTokenManager("secretsecretsecretsecretsecretse") + token, err := tm.IssueAccessToken(1, -time.Minute) + if err != nil { + t.Fatalf("IssueAccessToken error: %v", err) + } + _, err = tm.ParseAccessToken(token) + if !errors.Is(err, auth.ErrInvalidToken) { + t.Fatalf("expected ErrInvalidToken for expired, got %v", err) + } +} + +func TestNoneAlgorithmToken(t *testing.T) { + claims := jwt.RegisteredClaims{ + Subject: "1", + IssuedAt: jwt.NewNumericDate(time.Now()), + ExpiresAt: jwt.NewNumericDate(time.Now().Add(time.Minute)), + } + token := jwt.NewWithClaims(jwt.SigningMethodNone, claims) + tokenStr, err := token.SignedString(jwt.UnsafeAllowNoneSignatureType) + if err != nil { + t.Fatalf("create none token: %v", err) + } + tm := auth.NewTokenManager("longsecretlongsecretlongsecretlongsec") + _, err = tm.ParseAccessToken(tokenStr) + if !errors.Is(err, auth.ErrInvalidToken) { + t.Fatalf("expected ErrInvalidToken for none alg, got %v", err) + } +} + +func TestGarbageToken(t *testing.T) { + tm := auth.NewTokenManager("secretsecretsecretsecretsecretse") + _, err := tm.ParseAccessToken("not a token") + if !errors.Is(err, auth.ErrInvalidToken) { + t.Fatalf("expected ErrInvalidToken for garbage token, got %v", err) + } +} diff --git a/server/internal/platform/config/config.go b/server/internal/platform/config/config.go index f976638..861901a 100644 --- a/server/internal/platform/config/config.go +++ b/server/internal/platform/config/config.go @@ -20,6 +20,10 @@ type Config struct { // AllowedOrigins is a comma-separated list of CORS origins // allowed to call the API from a browser. AllowedOrigins []string + // DatabaseURL is the PostgreSQL DSN. + DatabaseURL string + // JWTSecret signs access tokens. Must be set in production. + JWTSecret string } // Load reads configuration from the environment, falling back to @@ -29,6 +33,8 @@ func Load() (*Config, error) { HTTPAddr: fromEnv("MESSENGER_HTTP_ADDR", ":8080"), StaticDir: os.Getenv("MESSENGER_STATIC_DIR"), Environment: fromEnv("MESSENGER_ENV", "development"), + DatabaseURL: fromEnv("MESSENGER_DATABASE_URL", "postgres://messenger:messenger@localhost:5432/messenger?sslmode=disable"), + JWTSecret: fromEnv("MESSENGER_JWT_SECRET", "dev-secret-change-me"), } for _, origin := range strings.Split(fromEnv("MESSENGER_ALLOWED_ORIGINS", ""), ",") { @@ -43,6 +49,10 @@ func Load() (*Config, error) { return nil, fmt.Errorf("MESSENGER_ENV must be one of development|staging|production, got %q", cfg.Environment) } + if cfg.Environment == "production" && cfg.JWTSecret == "dev-secret-change-me" { + return nil, fmt.Errorf("MESSENGER_JWT_SECRET must be set in production") + } + return cfg, nil } diff --git a/server/internal/platform/config/config_test.go b/server/internal/platform/config/config_test.go index e0c4857..16cba1a 100644 --- a/server/internal/platform/config/config_test.go +++ b/server/internal/platform/config/config_test.go @@ -24,6 +24,7 @@ func TestLoadFromEnv(t *testing.T) { t.Setenv("MESSENGER_HTTP_ADDR", ":9000") t.Setenv("MESSENGER_ENV", "production") t.Setenv("MESSENGER_ALLOWED_ORIGINS", "https://a.example, https://b.example") + t.Setenv("MESSENGER_JWT_SECRET", "prod-secret-at-least-32-chars-long!!") cfg, err := Load() if err != nil { @@ -38,6 +39,9 @@ func TestLoadFromEnv(t *testing.T) { if len(cfg.AllowedOrigins) != 2 || cfg.AllowedOrigins[1] != "https://b.example" { t.Errorf("AllowedOrigins = %v, want [https://a.example https://b.example]", cfg.AllowedOrigins) } + if cfg.DatabaseURL == "" { + t.Error("DatabaseURL = empty, want default DSN") + } } func TestLoadInvalidEnv(t *testing.T) { @@ -47,3 +51,12 @@ func TestLoadInvalidEnv(t *testing.T) { t.Fatal("Load() error = nil, want validation error") } } + +func TestLoadRequiresJWTSecretInProduction(t *testing.T) { + t.Setenv("MESSENGER_ENV", "production") + t.Setenv("MESSENGER_JWT_SECRET", "") + + if _, err := Load(); err == nil { + t.Fatal("Load() error = nil, want missing JWT secret error") + } +} diff --git a/server/internal/realtime/client.go b/server/internal/realtime/client.go new file mode 100644 index 0000000..7cc7fb4 --- /dev/null +++ b/server/internal/realtime/client.go @@ -0,0 +1,136 @@ +package realtime + +import ( + "context" + "encoding/json" + "time" + + "github.com/coder/websocket" +) + +const sendBufferSize = 64 + +// client wraps a single WebSocket connection belonging to a user. +type client struct { + hub *Hub + conn *websocket.Conn + userID int64 + send chan []byte + done chan struct{} +} + +func newClient(hub *Hub, conn *websocket.Conn, userID int64) *client { + return &client{ + hub: hub, + conn: conn, + userID: userID, + send: make(chan []byte, sendBufferSize), + done: make(chan struct{}), + } +} + +// enqueue marshals env and queues it for delivery; drops the message if the +// client cannot keep up. +func (c *client) enqueue(env envelope) { + payload, err := json.Marshal(env) + if err != nil { + return + } + c.enqueueRaw(payload) +} + +func (c *client) enqueueRaw(payload []byte) { + select { + case <-c.done: + case c.send <- payload: + default: + c.closeSlow() + } +} + +// closeSlow terminates a connection that cannot keep up with its send queue. +func (c *client) closeSlow() { + c.hub.mu.Lock() + defer c.hub.mu.Unlock() + if conns := c.hub.conns[c.userID]; conns != nil { + if _, ok := conns[c]; ok { + c.conn.Close(websocket.StatusPolicyViolation, "client too slow") + } + } +} + +// writeLoop drains the send queue onto the connection. +func (c *client) writeLoop() { + ctx := context.Background() + for { + select { + case <-c.done: + return + case msg := <-c.send: + writeCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + err := c.conn.Write(writeCtx, websocket.MessageText, msg) + cancel() + if err != nil { + return + } + } + } +} + +// readLoop processes inbound client events until the connection closes. +func (c *client) readLoop() error { + ctx := context.Background() + for { + _, data, err := c.conn.Read(ctx) + if err != nil { + return err + } + var env envelope + if err := json.Unmarshal(data, &env); err != nil { + c.conn.Close(websocket.StatusInvalidFramePayloadData, "bad json") + return err + } + + switch env.Type { + case EventTyping: + var msg clientTyping + if err := json.Unmarshal(env.Data, &msg); err != nil { + continue + } + if err := c.handleTyping(ctx, msg); err != nil { + return err + } + case EventRead: + var msg clientRead + if err := json.Unmarshal(env.Data, &msg); err != nil { + continue + } + if err := c.handleRead(ctx, msg); err != nil { + return err + } + default: + return websocket.CloseError{Code: websocket.StatusUnsupportedData, Reason: "unsupported event type"} + } + } +} + +func (c *client) handleTyping(ctx context.Context, msg clientTyping) error { + ok, err := c.hub.members.IsMember(ctx, msg.ConversationID, c.userID) + if err != nil || !ok { + return nil + } + c.hub.PublishTyping(msg.ConversationID, c.userID, msg.IsTyping) + return nil +} + +func (c *client) handleRead(ctx context.Context, msg clientRead) error { + ok, err := c.hub.members.IsMember(ctx, msg.ConversationID, c.userID) + if err != nil || !ok { + return nil + } + if err := c.hub.members.UpdateLastRead(ctx, msg.ConversationID, c.userID); err != nil { + return err + } + c.hub.PublishRead(msg.ConversationID, c.userID, msg.MessageID) + return nil +} diff --git a/server/internal/realtime/events.go b/server/internal/realtime/events.go new file mode 100644 index 0000000..e526d19 --- /dev/null +++ b/server/internal/realtime/events.go @@ -0,0 +1,70 @@ +// Package realtime implements the WebSocket hub: real-time delivery of +// conversation events (messages, typing, presence, read receipts). +package realtime + +import ( + "encoding/json" + "time" +) + +// Event type constants sent over the wire. +const ( + EventMessageCreated = "message.created" + EventTyping = "typing" + EventPresence = "presence" + EventRead = "read" +) + +// Presence status values. +const ( + StatusOnline = "online" + StatusOffline = "offline" +) + +// MessageCreated is published when a message is stored. +type MessageCreated struct { + TempID string `json:"temp_id,omitempty"` + ConversationID int64 `json:"conversation_id"` + MessageID int64 `json:"message_id"` + SenderID int64 `json:"sender_id"` + Body string `json:"body"` + CreatedAt time.Time `json:"created_at"` +} + +// Typing is published when a user starts or stops typing in a conversation. +type Typing struct { + ConversationID int64 `json:"conversation_id"` + UserID int64 `json:"user_id"` + IsTyping bool `json:"is_typing"` +} + +// Presence is published when a user comes online or goes offline. +type Presence struct { + UserID int64 `json:"user_id"` + Status string `json:"status"` +} + +// Read is published when a user marks messages in a conversation as read. +type Read struct { + ConversationID int64 `json:"conversation_id"` + UserID int64 `json:"user_id"` + MessageID int64 `json:"message_id"` +} + +// envelope is the wire format for every message, both inbound and outbound. +type envelope struct { + Type string `json:"type"` + Data json.RawMessage `json:"data"` +} + +// clientTyping is sent by the client to signal typing state. +type clientTyping struct { + ConversationID int64 `json:"conversation_id"` + IsTyping bool `json:"is_typing"` +} + +// clientRead is sent by the client to mark messages as read. +type clientRead struct { + ConversationID int64 `json:"conversation_id"` + MessageID int64 `json:"message_id"` +} diff --git a/server/internal/realtime/hub.go b/server/internal/realtime/hub.go new file mode 100644 index 0000000..229576b --- /dev/null +++ b/server/internal/realtime/hub.go @@ -0,0 +1,163 @@ +package realtime + +import ( + "context" + "encoding/json" + "log/slog" + "net/http" + "sync" + + "github.com/coder/websocket" +) + +// MembershipProvider resolves conversation membership and read state. +type MembershipProvider interface { + ConversationIDsForUser(ctx context.Context, userID int64) ([]int64, error) + IsMember(ctx context.Context, conversationID, userID int64) (bool, error) + UpdateLastRead(ctx context.Context, conversationID, userID int64) error +} + +// Hub tracks WebSocket connections and fans out events to the +// users subscribed to a conversation. +type Hub struct { + mu sync.Mutex + conns map[int64]map[*client]struct{} + subs map[int64]map[int64]struct{} + members MembershipProvider +} + +// NewHub creates a hub that resolves membership through members. +func NewHub(members MembershipProvider) *Hub { + return &Hub{ + conns: make(map[int64]map[*client]struct{}), + subs: make(map[int64]map[int64]struct{}), + members: members, + } +} + +// HandleWS upgrades the HTTP request to a WebSocket connection and +// runs the client loops until the connection closes. +func (h *Hub) HandleWS(w http.ResponseWriter, r *http.Request, userID int64) { + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{OriginPatterns: []string{"*"}}) + if err != nil { + slog.Error("ws upgrade failed", "error", err) + return + } + conn.SetReadLimit(32 << 10) + + ctx := r.Context() + conversationIDs, err := h.members.ConversationIDsForUser(ctx, userID) + if err != nil { + slog.Error("ws: load memberships", "user_id", userID, "error", err) + conn.Close(websocket.StatusInternalError, "membership lookup failed") + return + } + + c := newClient(h, conn, userID) + h.register(c, conversationIDs) + defer h.unregister(c, conversationIDs) + + h.broadcastPresence(userID, conversationIDs, StatusOnline) + defer h.broadcastPresence(userID, conversationIDs, StatusOffline) + + go c.writeLoop() + if err := c.readLoop(); err != nil { + slog.Debug("ws connection closed", "user_id", userID, "error", err) + } +} + +// PublishMessageCreated delivers a stored message to everyone subscribed +// to the conversation, including the sender. +func (h *Hub) PublishMessageCreated(conversationID int64, evt MessageCreated) { + h.sendToConversation(conversationID, envelope{Type: EventMessageCreated, Data: mustMarshal(evt)}, 0) +} + +// PublishTyping delivers a typing state change to everyone except the sender. +func (h *Hub) PublishTyping(conversationID, userID int64, isTyping bool) { + evt := Typing{ConversationID: conversationID, UserID: userID, IsTyping: isTyping} + h.sendToConversation(conversationID, envelope{Type: EventTyping, Data: mustMarshal(evt)}, userID) +} + +// PublishRead delivers a read receipt to everyone except the reader. +func (h *Hub) PublishRead(conversationID, userID, messageID int64) { + evt := Read{ConversationID: conversationID, UserID: userID, MessageID: messageID} + h.sendToConversation(conversationID, envelope{Type: EventRead, Data: mustMarshal(evt)}, userID) +} + +func (h *Hub) register(c *client, conversationIDs []int64) { + h.mu.Lock() + defer h.mu.Unlock() + if h.conns[c.userID] == nil { + h.conns[c.userID] = make(map[*client]struct{}) + } + h.conns[c.userID][c] = struct{}{} + for _, cid := range conversationIDs { + if h.subs[cid] == nil { + h.subs[cid] = make(map[int64]struct{}) + } + h.subs[cid][c.userID] = struct{}{} + } +} + +func (h *Hub) unregister(c *client, conversationIDs []int64) { + h.mu.Lock() + defer h.mu.Unlock() + if conns := h.conns[c.userID]; conns != nil { + delete(conns, c) + if len(conns) == 0 { + delete(h.conns, c.userID) + } + } + for _, cid := range conversationIDs { + if users := h.subs[cid]; users != nil { + delete(users, c.userID) + if len(users) == 0 { + delete(h.subs, cid) + } + } + } + close(c.done) +} + +func (h *Hub) broadcastPresence(userID int64, conversationIDs []int64, status string) { + evt := Presence{UserID: userID, Status: status} + env := envelope{Type: EventPresence, Data: mustMarshal(evt)} + h.mu.Lock() + defer h.mu.Unlock() + for _, cid := range conversationIDs { + for uid := range h.subs[cid] { + if uid == userID { + continue + } + for c := range h.conns[uid] { + c.enqueue(env) + } + } + } +} + +func (h *Hub) sendToConversation(conversationID int64, env envelope, exceptUserID int64) { + payload, err := json.Marshal(env) + if err != nil { + slog.Error("ws: marshal event", "error", err) + return + } + h.mu.Lock() + defer h.mu.Unlock() + for uid := range h.subs[conversationID] { + if uid == exceptUserID { + continue + } + for c := range h.conns[uid] { + c.enqueueRaw(payload) + } + } +} + +func mustMarshal(v any) json.RawMessage { + data, err := json.Marshal(v) + if err != nil { + panic(err) + } + return data +} diff --git a/server/internal/realtime/hub_test.go b/server/internal/realtime/hub_test.go new file mode 100644 index 0000000..8bfac4a --- /dev/null +++ b/server/internal/realtime/hub_test.go @@ -0,0 +1,262 @@ +package realtime + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/coder/websocket" +) + +type fakeMembers struct { + mu sync.Mutex + byUser map[int64][]int64 + member map[[2]int64]bool + lastReads [][2]int64 +} + +func newFakeMembers() *fakeMembers { + return &fakeMembers{ + byUser: make(map[int64][]int64), + member: make(map[[2]int64]bool), + } +} + +func (f *fakeMembers) ConversationIDsForUser(_ context.Context, userID int64) ([]int64, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.byUser[userID], nil +} + +func (f *fakeMembers) IsMember(_ context.Context, conversationID, userID int64) (bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.member[[2]int64{conversationID, userID}], nil +} + +func (f *fakeMembers) UpdateLastRead(_ context.Context, conversationID, userID int64) error { + f.mu.Lock() + defer f.mu.Unlock() + f.lastReads = append(f.lastReads, [2]int64{conversationID, userID}) + return nil +} + +func (f *fakeMembers) readCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.lastReads) +} + +func (f *fakeMembers) readAt(i int) [2]int64 { + f.mu.Lock() + defer f.mu.Unlock() + return f.lastReads[i] +} + +func waitFor(t *testing.T, cond func() bool, what string) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", what) +} + +func startHub(t *testing.T, fm *fakeMembers) (*Hub, *httptest.Server) { + t.Helper() + hub := NewHub(fm) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + userID := int64(0) + switch r.URL.Query().Get("user") { + case "1": + userID = 1 + case "2": + userID = 2 + } + hub.HandleWS(w, r, userID) + })) + t.Cleanup(srv.Close) + return hub, srv +} + +func dialWS(t *testing.T, srv *httptest.Server, user string) *websocket.Conn { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + url := "ws" + strings.TrimPrefix(srv.URL, "http") + "?user=" + user + conn, _, err := websocket.Dial(ctx, url, nil) + if err != nil { + t.Fatalf("websocket dial: %v", err) + } + t.Cleanup(func() { conn.Close(websocket.StatusNormalClosure, "") }) + time.Sleep(50 * time.Millisecond) + return conn +} + +func readEnvelope(t *testing.T, conn *websocket.Conn) envelope { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _, data, err := conn.Read(ctx) + if err != nil { + t.Fatalf("ws read: %v", err) + } + var env envelope + if err := json.Unmarshal(data, &env); err != nil { + t.Fatalf("unmarshal envelope: %v", err) + } + return env +} + +// readEventOfType reads events until one of the wanted type arrives, +// skipping unrelated events such as presence. +func readEventOfType(t *testing.T, conn *websocket.Conn, wantType string) envelope { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + env := readEnvelope(t, conn) + if env.Type == wantType { + return env + } + } + t.Fatalf("timed out waiting for event %q", wantType) + return envelope{} +} + +func sendEnvelope(t *testing.T, conn *websocket.Conn, env envelope) { + t.Helper() + data, err := json.Marshal(env) + if err != nil { + t.Fatalf("marshal envelope: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := conn.Write(ctx, websocket.MessageText, data); err != nil { + t.Fatalf("ws write: %v", err) + } +} + +func TestMessageCreatedReachesSubscribers(t *testing.T) { + fm := newFakeMembers() + fm.byUser[1] = []int64{10} + fm.byUser[2] = []int64{10} + fm.member[[2]int64{10, 1}] = true + fm.member[[2]int64{10, 2}] = true + + hub, srv := startHub(t, fm) + conn1 := dialWS(t, srv, "1") + conn2 := dialWS(t, srv, "2") + + evt := MessageCreated{ConversationID: 10, MessageID: 55, SenderID: 1, Body: "hello"} + hub.PublishMessageCreated(10, evt) + + for _, conn := range []*websocket.Conn{conn1, conn2} { + env := readEventOfType(t, conn, EventMessageCreated) + var got MessageCreated + if err := json.Unmarshal(env.Data, &got); err != nil { + t.Fatalf("unmarshal data: %v", err) + } + if got.MessageID != 55 || got.Body != "hello" { + t.Errorf("data = %+v, want MessageID=55 Body=hello", got) + } + } +} + +func TestNonMemberTypingIsIgnored(t *testing.T) { + fm := newFakeMembers() + fm.byUser[1] = []int64{10} + fm.byUser[2] = []int64{10} + fm.member[[2]int64{10, 2}] = true + + hub, srv := startHub(t, fm) + conn1 := dialWS(t, srv, "1") + conn2 := dialWS(t, srv, "2") + + sendEnvelope(t, conn1, envelope{ + Type: EventTyping, + Data: mustMarshal(clientTyping{ConversationID: 10, IsTyping: true}), + }) + + evt := MessageCreated{ConversationID: 10, MessageID: 1, SenderID: 2, Body: "x"} + hub.PublishMessageCreated(10, evt) + env := readEventOfType(t, conn2, EventMessageCreated) + if env.Type != EventMessageCreated { + t.Errorf("member got %q, want message.created (no typing leak)", env.Type) + } +} + +func TestPresenceOnConnectAndClose(t *testing.T) { + fm := newFakeMembers() + fm.byUser[1] = []int64{10} + fm.byUser[2] = []int64{10} + + _, srv := startHub(t, fm) + conn1 := dialWS(t, srv, "1") + conn2 := dialWS(t, srv, "2") + + env := readEnvelope(t, conn1) + if env.Type != EventPresence { + t.Fatalf("first event type = %q, want presence", env.Type) + } + var p Presence + if err := json.Unmarshal(env.Data, &p); err != nil { + t.Fatalf("unmarshal presence: %v", err) + } + if p.UserID != 2 || p.Status != StatusOnline { + t.Errorf("presence = %+v, want user 2 online", p) + } + + conn2.Close(websocket.StatusNormalClosure, "") + env = readEnvelope(t, conn1) + if env.Type != EventPresence { + t.Fatalf("closing event type = %q, want presence", env.Type) + } + if err := json.Unmarshal(env.Data, &p); err != nil { + t.Fatalf("unmarshal presence: %v", err) + } + if p.UserID != 2 || p.Status != StatusOffline { + t.Errorf("presence = %+v, want user 2 offline", p) + } +} + +func TestReadEventUpdatesLastReadAndBroadcasts(t *testing.T) { + fm := newFakeMembers() + fm.byUser[1] = []int64{10} + fm.byUser[2] = []int64{10} + fm.member[[2]int64{10, 1}] = true + fm.member[[2]int64{10, 2}] = true + + _, srv := startHub(t, fm) + conn1 := dialWS(t, srv, "1") + conn2 := dialWS(t, srv, "2") + + sendEnvelope(t, conn1, envelope{ + Type: EventRead, + Data: mustMarshal(clientRead{ConversationID: 10, MessageID: 99}), + }) + + waitFor(t, func() bool { return fm.readCount() == 1 }, "UpdateLastRead call") + if got := fm.readAt(0); got != [2]int64{10, 1} { + t.Errorf("lastReads[0] = %v, want [10 1]", got) + } + + env := readEnvelope(t, conn2) + if env.Type != EventRead { + t.Fatalf("type = %q, want read", env.Type) + } + var r Read + if err := json.Unmarshal(env.Data, &r); err != nil { + t.Fatalf("unmarshal read: %v", err) + } + if r.UserID != 1 || r.MessageID != 99 || r.ConversationID != 10 { + t.Errorf("read = %+v, want user 1, message 99, conv 10", r) + } +} diff --git a/server/internal/server/server.go b/server/internal/server/server.go index 0fb9b4e..3b3d2bc 100644 --- a/server/internal/server/server.go +++ b/server/internal/server/server.go @@ -8,11 +8,15 @@ import ( "net/http" "time" + "messenger/internal/api" + "messenger/internal/auth" "messenger/internal/platform/config" + "messenger/internal/realtime" + "messenger/internal/store" ) // New builds an *http.Server with all routes and middleware applied. -func New(cfg *config.Config, log *slog.Logger) *http.Server { +func New(cfg *config.Config, log *slog.Logger, st *store.Store, tm *auth.TokenManager, hub *realtime.Hub) *http.Server { mux := http.NewServeMux() mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { @@ -23,12 +27,7 @@ func New(cfg *config.Config, log *slog.Logger) *http.Server { writeJSON(w, http.StatusOK, map[string]string{"status": "ready"}) }) - mux.HandleFunc("GET /api/v1/ping", func(w http.ResponseWriter, r *http.Request) { - writeJSON(w, http.StatusOK, map[string]string{ - "message": "pong", - "time": time.Now().UTC().Format(time.RFC3339), - }) - }) + mux.Handle("/api/v1/", api.NewHandler(st, tm, hub, log, cfg)) if cfg.StaticDir != "" { mux.Handle("/", spaHandler{staticDir: cfg.StaticDir, log: log}) diff --git a/server/internal/store/contacts.go b/server/internal/store/contacts.go new file mode 100644 index 0000000..e05b31c --- /dev/null +++ b/server/internal/store/contacts.go @@ -0,0 +1,69 @@ +package store + +import ( + "context" + "errors" + + "github.com/jackc/pgconn" + "github.com/jackc/pgerrcode" +) + +// AddContact creates a contact relation or returns ErrDuplicate if it exists. +func (s *Store) AddContact(ctx context.Context, ownerID, contactID int64) error { + _, err := s.pool.Exec(ctx, ` + INSERT INTO contacts(owner_id, contact_id) + VALUES($1, $2) + `, ownerID, contactID) + if err != nil { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == pgerrcode.UniqueViolation { + return ErrDuplicate + } + } + return err +} + +// RemoveContact deletes a contact relation; no error if not present. +func (s *Store) RemoveContact(ctx context.Context, ownerID, contactID int64) error { + _, err := s.pool.Exec(ctx, ` + DELETE FROM contacts WHERE owner_id=$1 AND contact_id=$2 + `, ownerID, contactID) + return err +} + +// ListContacts returns all contacts for an owner ordered by created_at descending. +func (s *Store) ListContacts(ctx context.Context, ownerID int64) ([]Contact, error) { + rows, err := s.pool.Query(ctx, ` + SELECT owner_id, contact_id, created_at + FROM contacts WHERE owner_id=$1 + ORDER BY created_at DESC + `, ownerID) + if err != nil { + return nil, err + } + defer rows.Close() + + var list []Contact + for rows.Next() { + var c Contact + if err := rows.Scan(&c.OwnerID, &c.ContactID, &c.CreatedAt); err != nil { + return nil, err + } + list = append(list, c) + } + return list, nil +} + +// IsContact reports whether a contact relation exists. +func (s *Store) IsContact(ctx context.Context, ownerID, contactID int64) (bool, error) { + var exists bool + err := s.pool.QueryRow(ctx, ` + SELECT EXISTS( + SELECT 1 FROM contacts WHERE owner_id=$1 AND contact_id=$2 + ) + `, ownerID, contactID).Scan(&exists) + if err != nil { + return false, err + } + return exists, nil +} diff --git a/server/internal/store/conversations.go b/server/internal/store/conversations.go new file mode 100644 index 0000000..efa2d53 --- /dev/null +++ b/server/internal/store/conversations.go @@ -0,0 +1,171 @@ +package store + +import ( + "context" + "errors" + "sort" + + "github.com/jackc/pgx/v5" +) + +// CreateDM returns an existing direct-message conversation between two users or creates it. +func (s *Store) CreateDM(ctx context.Context, userA, userB int64) (*Conversation, error) { + // Check existing DM + var conv Conversation + err := s.pool.QueryRow(ctx, ` + SELECT c.id, c.type, c.created_at + FROM conversations c + JOIN conversation_members m1 ON m1.conversation_id=c.id AND m1.user_id=$1 + JOIN conversation_members m2 ON m2.conversation_id=c.id AND m2.user_id=$2 + WHERE c.type='dm' + LIMIT 1 + `, userA, userB).Scan(&conv.ID, &conv.Type, &conv.CreatedAt) + if err == nil { + return &conv, nil + } + if !errors.Is(err, pgx.ErrNoRows) { + return nil, err + } + + conn, err := s.pool.Acquire(ctx) + if err != nil { + return nil, err + } + defer conn.Release() + + tx, err := conn.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + err = tx.QueryRow(ctx, ` + INSERT INTO conversations(type) VALUES('dm') + RETURNING id, type, created_at + `).Scan(&conv.ID, &conv.Type, &conv.CreatedAt) + if err != nil { + return nil, err + } + // Add members + members := []int64{userA, userB} + for _, uid := range members { + if _, err := tx.Exec(ctx, ` + INSERT INTO conversation_members(conversation_id, user_id) + VALUES($1, $2) + `, conv.ID, uid); err != nil { + return nil, err + } + } + + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return &conv, nil +} + +// GetConversationByID retrieves a conversation by ID or returns ErrNotFound. +func (s *Store) GetConversationByID(ctx context.Context, id int64) (*Conversation, error) { + var conv Conversation + err := s.pool.QueryRow(ctx, ` + SELECT id, type, created_at FROM conversations WHERE id=$1 + `, id).Scan(&conv.ID, &conv.Type, &conv.CreatedAt) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return nil, err + } + return &conv, nil +} + +// IsMember reports whether a user is a member of a conversation. +func (s *Store) IsMember(ctx context.Context, conversationID, userID int64) (bool, error) { + var exists bool + err := s.pool.QueryRow(ctx, ` + SELECT EXISTS( + SELECT 1 FROM conversation_members WHERE conversation_id=$1 AND user_id=$2 + ) + `, conversationID, userID).Scan(&exists) + if err != nil { + return false, err + } + return exists, nil +} + +// ConversationIDsForUser returns all conversation IDs the user is a member of. +func (s *Store) ConversationIDsForUser(ctx context.Context, userID int64) ([]int64, error) { + rows, err := s.pool.Query(ctx, ` + SELECT conversation_id FROM conversation_members WHERE user_id=$1 + `, userID) + if err != nil { + return nil, err + } + defer rows.Close() + + var ids []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + return ids, nil +} + +// ListConversationsForUser returns summaries of conversations for a user. +func (s *Store) ListConversationsForUser(ctx context.Context, userID int64) ([]ConversationSummary, error) { + rows, err := s.pool.Query(ctx, ` + SELECT c.id, c.type, c.created_at, + m.id, m.conversation_id, m.sender_id, m.body, m.created_at, m.edited_at, m.deleted_at, + uc.unread_count + FROM conversation_members cm + JOIN conversations c ON c.id = cm.conversation_id + LEFT JOIN LATERAL ( + SELECT id, conversation_id, sender_id, body, created_at, edited_at, deleted_at + FROM messages WHERE conversation_id = cm.conversation_id + ORDER BY id DESC LIMIT 1 + ) m ON true + LEFT JOIN LATERAL ( + SELECT COUNT(*) AS unread_count + FROM messages msg + WHERE msg.conversation_id = cm.conversation_id + AND msg.sender_id <> $1 + AND (cm.last_read_at IS NULL OR msg.created_at > cm.last_read_at) + ) uc ON true + WHERE cm.user_id = $1 + ORDER BY m.id DESC NULLS LAST + `, userID) + if err != nil { + return nil, err + } + defer rows.Close() + + var list []ConversationSummary + for rows.Next() { + var cs ConversationSummary + var msg Message + if err := rows.Scan( + &cs.ID, &cs.Type, &cs.CreatedAt, + &msg.ID, &msg.ConversationID, &msg.SenderID, &msg.Body, &msg.CreatedAt, &msg.EditedAt, &msg.DeletedAt, + &cs.UnreadCount, + ); err != nil { + return nil, err + } + if msg.ID != 0 { + cs.LastMessage = &msg + } + list = append(list, cs) + } + return list, nil +} + +// UpdateLastRead sets the last_read_at to now for a user's conversation. +func (s *Store) UpdateLastRead(ctx context.Context, conversationID, userID int64) error { + _, err := s.pool.Exec(ctx, ` + UPDATE conversation_members SET last_read_at = now() + WHERE conversation_id=$1 AND user_id=$2 + `, conversationID, userID) + return err +} diff --git a/server/internal/store/errors.go b/server/internal/store/errors.go new file mode 100644 index 0000000..4108e18 --- /dev/null +++ b/server/internal/store/errors.go @@ -0,0 +1,11 @@ +package store + +import ( + "errors" +) + +// ErrNotFound is returned when a record is not found in the store. +var ErrNotFound = errors.New("store: not found") + +// ErrDuplicate is returned when attempting to create a record that already exists. +var ErrDuplicate = errors.New("store: duplicate") diff --git a/server/internal/store/integration_test.go b/server/internal/store/integration_test.go new file mode 100644 index 0000000..18c470d --- /dev/null +++ b/server/internal/store/integration_test.go @@ -0,0 +1,111 @@ +package store + +import ( + "context" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// TestIntegration exercises basic CRUD operations against a real database. +func TestIntegration(t *testing.T) { + url := os.Getenv("MESSENGER_TEST_DATABASE_URL") + if url == "" { + t.Skip("skipping integration test; MESSENGER_TEST_DATABASE_URL not set") + } + ctx := context.Background() + pool, err := pgxpool.New(ctx, url) + if err != nil { + t.Fatalf("unable to connect to db: %v", err) + } + defer pool.Close() + + db := New(pool) + if err := db.Migrate(ctx); err != nil { + t.Fatalf("Migrate failed: %v", err) + } + + // Users + u1, err := db.CreateUser(ctx, "user1", "User One", "hash1") + if err != nil { + t.Fatalf("CreateUser u1: %v", err) + } + u2, err := db.CreateUser(ctx, "user2", "User Two", "hash2") + if err != nil { + t.Fatalf("CreateUser u2: %v", err) + } + if _, err := db.GetUserByUsername(ctx, u1.Username); err != nil { + t.Fatalf("GetUserByUsername: %v", err) + } + if _, err := db.GetUserByID(ctx, u2.ID); err != nil { + t.Fatalf("GetUserByID: %v", err) + } + + // Sessions + exp := time.Now().Add(time.Hour) + sess, err := db.CreateSession(ctx, u1.ID, "rh1", "ua", "127.0.0.1", exp) + if err != nil { + t.Fatalf("CreateSession: %v", err) + } + if _, err := db.GetSessionByRefreshHash(ctx, sess.RefreshHash); err != nil { + t.Fatalf("GetSessionByRefreshHash: %v", err) + } + if err := db.RevokeSession(ctx, sess.ID); err != nil { + t.Fatalf("RevokeSession: %v", err) + } + if err := db.RevokeAllSessions(ctx, u1.ID); err != nil { + t.Fatalf("RevokeAllSessions: %v", err) + } + + // Contacts + if err := db.AddContact(ctx, u1.ID, u2.ID); err != nil { + t.Fatalf("AddContact: %v", err) + } + if ok, err := db.IsContact(ctx, u1.ID, u2.ID); err != nil || !ok { + t.Fatalf("IsContact: %v, %v", ok, err) + } + list, err := db.ListContacts(ctx, u1.ID) + if err != nil || len(list) != 1 { + t.Fatalf("ListContacts: got %v, err %v", list, err) + } + if err := db.RemoveContact(ctx, u1.ID, u2.ID); err != nil { + t.Fatalf("RemoveContact: %v", err) + } + + // Conversations and messages + conv, err := db.CreateDM(ctx, u1.ID, u2.ID) + if err != nil { + t.Fatalf("CreateDM: %v", err) + } + if conv2, err := db.CreateDM(ctx, u1.ID, u2.ID); err != nil || conv2.ID != conv.ID { + t.Fatalf("CreateDM idempotent: %v, %v", err, conv2) + } + if ok, err := db.IsMember(ctx, conv.ID, u1.ID); err != nil || !ok { + t.Fatalf("IsMember: %v, %v", ok, err) + } + ids, err := db.ConversationIDsForUser(ctx, u1.ID) + if err != nil || len(ids) == 0 { + t.Fatalf("ConversationIDsForUser: %v, %v", ids, err) + } + sums, err := db.ListConversationsForUser(ctx, u1.ID) + if err != nil || len(sums) != 1 { + t.Fatalf("ListConversationsForUser: %v, %v", sums, err) + } + + msg, err := db.InsertMessage(ctx, conv.ID, u1.ID, "hello") + if err != nil { + t.Fatalf("InsertMessage: %v", err) + } + if _, err := db.GetMessageByID(ctx, msg.ID); err != nil { + t.Fatalf("GetMessageByID: %v", err) + } + msgs, err := db.ListMessages(ctx, conv.ID, 0, 10) + if err != nil || len(msgs) != 1 { + t.Fatalf("ListMessages: %v, %v", msgs, err) + } + if err := db.UpdateLastRead(ctx, conv.ID, u2.ID); err != nil { + t.Fatalf("UpdateLastRead: %v", err) + } +} diff --git a/server/internal/store/messages.go b/server/internal/store/messages.go new file mode 100644 index 0000000..427ccc4 --- /dev/null +++ b/server/internal/store/messages.go @@ -0,0 +1,83 @@ +package store + +import ( + "context" + "errors" + + "github.com/jackc/pgx/v5" +) + +// InsertMessage creates a new message in a conversation and returns it. +func (s *Store) InsertMessage(ctx context.Context, conversationID, senderID int64, body string) (*Message, error) { + var m Message + err := s.pool.QueryRow(ctx, ` + INSERT INTO messages(conversation_id, sender_id, body) + VALUES($1, $2, $3) + RETURNING id, conversation_id, sender_id, body, created_at, edited_at, deleted_at + `, conversationID, senderID, body).Scan( + &m.ID, &m.ConversationID, &m.SenderID, &m.Body, + &m.CreatedAt, &m.EditedAt, &m.DeletedAt, + ) + if err != nil { + return nil, err + } + return &m, nil +} + +// GetMessageByID retrieves a message by ID or returns ErrNotFound. +func (s *Store) GetMessageByID(ctx context.Context, id int64) (*Message, error) { + var m Message + err := s.pool.QueryRow(ctx, ` + SELECT id, conversation_id, sender_id, body, created_at, edited_at, deleted_at + FROM messages WHERE id=$1 + `, id).Scan( + &m.ID, &m.ConversationID, &m.SenderID, &m.Body, + &m.CreatedAt, &m.EditedAt, &m.DeletedAt, + ) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return nil, err + } + return &m, nil +} + +// ListMessages returns messages for a conversation in reverse chronological order. +// If before > 0, only messages with ID less than before are returned. +func (s *Store) ListMessages(ctx context.Context, conversationID int64, before int64, limit int) ([]Message, error) { + var rows pgx.Rows + var err error + if before > 0 { + rows, err = s.pool.Query(ctx, ` + SELECT id, conversation_id, sender_id, body, created_at, edited_at, deleted_at + FROM messages + WHERE conversation_id=$1 AND id<$2 + ORDER BY id DESC LIMIT $3 + `, conversationID, before, limit) + } else { + rows, err = s.pool.Query(ctx, ` + SELECT id, conversation_id, sender_id, body, created_at, edited_at, deleted_at + FROM messages + WHERE conversation_id=$1 + ORDER BY id DESC LIMIT $2 + `, conversationID, limit) + } + if err != nil { + return nil, err + } + defer rows.Close() + + var msgs []Message + for rows.Next() { + var m Message + if err := rows.Scan( + &m.ID, &m.ConversationID, &m.SenderID, &m.Body, + &m.CreatedAt, &m.EditedAt, &m.DeletedAt, + ); err != nil { + return nil, err + } + msgs = append(msgs, m) + } + return msgs, nil +} diff --git a/server/internal/store/migrate.go b/server/internal/store/migrate.go new file mode 100644 index 0000000..0ec0a0d --- /dev/null +++ b/server/internal/store/migrate.go @@ -0,0 +1,89 @@ +package store + +import ( + "context" + "embed" + "sort" + "strconv" + "strings" +) + +//go:embed migrations/*.sql +var migrationFiles embed.FS + +// Migrate applies any pending SQL migrations in order within a single transaction. +func (s *Store) Migrate(ctx context.Context) error { + conn, err := s.pool.Acquire(ctx) + if err != nil { + return err + } + defer conn.Release() + + tx, err := conn.Begin(ctx) + if err != nil { + return err + } + defer func() { + if tx != nil { + tx.Rollback(ctx) + } + }() + + _, err = tx.Exec(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations ( + version INT PRIMARY KEY + )`) + if err != nil { + return err + } + + entries, err := migrationFiles.ReadDir("migrations") + if err != nil { + return err + } + + type mig struct { + ver int + name string + } + var migs []mig + for _, e := range entries { + name := e.Name() + if !strings.HasSuffix(name, ".sql") { + continue + } + parts := strings.SplitN(name, "_", 2) + v, err := strconv.Atoi(parts[0]) + if err != nil { + continue + } + migs = append(migs, mig{ver: v, name: name}) + } + sort.Slice(migs, func(i, j int) bool { return migs[i].ver < migs[j].ver }) + + for _, m := range migs { + var exists bool + err = tx.QueryRow(ctx, "SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version=$1)", m.ver).Scan(&exists) + if err != nil { + return err + } + if exists { + continue + } + data, err := migrationFiles.ReadFile("migrations/" + m.name) + if err != nil { + return err + } + if _, err = tx.Exec(ctx, string(data)); err != nil { + return err + } + if _, err = tx.Exec(ctx, "INSERT INTO schema_migrations(version) VALUES($1)", m.ver); err != nil { + return err + } + } + + if err = tx.Commit(ctx); err != nil { + return err + } + tx = nil + return nil +} diff --git a/server/internal/store/migrations/0001_init.sql b/server/internal/store/migrations/0001_init.sql new file mode 100644 index 0000000..ab7e2ae --- /dev/null +++ b/server/internal/store/migrations/0001_init.sql @@ -0,0 +1,56 @@ +-- 0001_init.sql +CREATE TABLE IF NOT EXISTS users ( + id BIGSERIAL PRIMARY KEY, + username TEXT NOT NULL UNIQUE, + display_name TEXT NOT NULL DEFAULT '', + password_hash TEXT NOT NULL, + avatar TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS sessions ( + id BIGSERIAL PRIMARY KEY, + user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + refresh_hash TEXT NOT NULL, + user_agent TEXT NOT NULL DEFAULT '', + ip TEXT NOT NULL DEFAULT '', + expires_at TIMESTAMPTZ NOT NULL, + revoked_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id); +CREATE INDEX IF NOT EXISTS idx_sessions_refresh_hash ON sessions(refresh_hash); + +CREATE TABLE IF NOT EXISTS contacts ( + owner_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + contact_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY(owner_id, contact_id), + CHECK(owner_id <> contact_id) +); + +CREATE TABLE IF NOT EXISTS conversations ( + id BIGSERIAL PRIMARY KEY, + type TEXT NOT NULL CHECK(type IN ('dm','group')), + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS conversation_members ( + conversation_id BIGINT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE, + user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + last_read_at TIMESTAMPTZ, + joined_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY(conversation_id, user_id) +); +CREATE INDEX IF NOT EXISTS idx_members_user ON conversation_members(user_id); + +CREATE TABLE IF NOT EXISTS messages ( + id BIGSERIAL PRIMARY KEY, + conversation_id BIGINT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE, + sender_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + body TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + edited_at TIMESTAMPTZ, + deleted_at TIMESTAMPTZ +); +CREATE INDEX IF NOT EXISTS idx_messages_conversation_id ON messages(conversation_id, id DESC); diff --git a/server/internal/store/sessions.go b/server/internal/store/sessions.go new file mode 100644 index 0000000..3d78faf --- /dev/null +++ b/server/internal/store/sessions.go @@ -0,0 +1,67 @@ +package store + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgconn" + "github.com/jackc/pgerrcode" + "github.com/jackc/pgx/v5" +) + +// CreateSession adds a new session record. +func (s *Store) CreateSession(ctx context.Context, userID int64, refreshHash, userAgent, ip string, expiresAt time.Time) (*Session, error) { + var sess Session + err := s.pool.QueryRow(ctx, ` + INSERT INTO sessions(user_id, refresh_hash, user_agent, ip, expires_at) + VALUES($1, $2, $3, $4, $5) + RETURNING id, user_id, refresh_hash, expires_at, revoked_at, created_at, user_agent, ip + `, userID, refreshHash, userAgent, ip, expiresAt).Scan( + &sess.ID, &sess.UserID, &sess.RefreshHash, &sess.ExpiresAt, + &sess.RevokedAt, &sess.CreatedAt, &sess.UserAgent, &sess.IP, + ) + if err != nil { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == pgerrcode.UniqueViolation { + return nil, ErrDuplicate + } + return nil, err + } + return &sess, nil +} + +// GetSessionByRefreshHash returns a session by its refresh hash or ErrNotFound. +func (s *Store) GetSessionByRefreshHash(ctx context.Context, hash string) (*Session, error) { + var sess Session + err := s.pool.QueryRow(ctx, ` + SELECT id, user_id, refresh_hash, expires_at, revoked_at, created_at, user_agent, ip + FROM sessions WHERE refresh_hash=$1 + `, hash).Scan( + &sess.ID, &sess.UserID, &sess.RefreshHash, &sess.ExpiresAt, + &sess.RevokedAt, &sess.CreatedAt, &sess.UserAgent, &sess.IP, + ) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return nil, err + } + return &sess, nil +} + +// RevokeSession sets revoked_at to now() for the given session ID. +func (s *Store) RevokeSession(ctx context.Context, id int64) error { + _, err := s.pool.Exec(ctx, ` + UPDATE sessions SET revoked_at = now() WHERE id = $1 + `, id) + return err +} + +// RevokeAllSessions revokes all sessions for a given user. +func (s *Store) RevokeAllSessions(ctx context.Context, userID int64) error { + _, err := s.pool.Exec(ctx, ` + UPDATE sessions SET revoked_at = now() WHERE user_id = $1 + `, userID) + return err +} diff --git a/server/internal/store/store.go b/server/internal/store/store.go new file mode 100644 index 0000000..f105282 --- /dev/null +++ b/server/internal/store/store.go @@ -0,0 +1,71 @@ +package store + +import ( + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// Store provides database access via a connection pool. +type Store struct { + pool *pgxpool.Pool +} + +// New creates a new Store with the given pgx pool. +func New(pool *pgxpool.Pool) *Store { + return &Store{pool: pool} +} + +// User represents a user account. +type User struct { + ID int64 + Username string + DisplayName string + PasswordHash string + Avatar string + CreatedAt time.Time +} + +// Session represents a user session for refresh tokens. +type Session struct { + ID int64 + UserID int64 + RefreshHash string + ExpiresAt time.Time + CreatedAt time.Time + RevokedAt *time.Time + UserAgent string + IP string +} + +// Conversation represents a chat conversation. +type Conversation struct { + ID int64 + Type string + CreatedAt time.Time +} + +// ConversationSummary provides a conversation with its last message and unread count. +type ConversationSummary struct { + Conversation + LastMessage *Message + UnreadCount int64 +} + +// Message represents a message sent within a conversation. +type Message struct { + ID int64 + ConversationID int64 + SenderID int64 + Body string + CreatedAt time.Time + EditedAt *time.Time + DeletedAt *time.Time +} + +// Contact represents a contact relationship between users. +type Contact struct { + OwnerID int64 + ContactID int64 + CreatedAt time.Time +} diff --git a/server/internal/store/users.go b/server/internal/store/users.go new file mode 100644 index 0000000..e165d2e --- /dev/null +++ b/server/internal/store/users.go @@ -0,0 +1,98 @@ +package store + +import ( + "context" + "errors" + + "github.com/jackc/pgconn" + "github.com/jackc/pgerrcode" + "github.com/jackc/pgx/v5" +) + +// CreateUser inserts a new user returning the created record or ErrDuplicate on conflict. +func (s *Store) CreateUser(ctx context.Context, username, displayName, passwordHash string) (*User, error) { + var u User + err := s.pool.QueryRow(ctx, ` + INSERT INTO users(username, display_name, password_hash) + VALUES($1, $2, $3) + RETURNING id, username, display_name, password_hash, avatar, created_at + `, username, displayName, passwordHash).Scan( + &u.ID, &u.Username, &u.DisplayName, &u.PasswordHash, &u.Avatar, &u.CreatedAt, + ) + if err != nil { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == pgerrcode.UniqueViolation { + return nil, ErrDuplicate + } + return nil, err + } + return &u, nil +} + +// GetUserByUsername retrieves a user by username or returns ErrNotFound. +func (s *Store) GetUserByUsername(ctx context.Context, username string) (*User, error) { + var u User + err := s.pool.QueryRow(ctx, ` + SELECT id, username, display_name, password_hash, avatar, created_at + FROM users WHERE username=$1 + `, username).Scan( + &u.ID, &u.Username, &u.DisplayName, &u.PasswordHash, &u.Avatar, &u.CreatedAt, + ) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return nil, err + } + return &u, nil +} + +// GetUserByID retrieves a user by ID or returns ErrNotFound. +func (s *Store) GetUserByID(ctx context.Context, id int64) (*User, error) { + var u User + err := s.pool.QueryRow(ctx, ` + SELECT id, username, display_name, password_hash, avatar, created_at + FROM users WHERE id=$1 + `, id).Scan( + &u.ID, &u.Username, &u.DisplayName, &u.PasswordHash, &u.Avatar, &u.CreatedAt, + ) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return nil, err + } + return &u, nil +} + +// SearchUsers returns users matching query in username or display_name via ILIKE. +// Empty query returns an empty slice. +func (s *Store) SearchUsers(ctx context.Context, query string, limit int) ([]User, error) { + if query == "" { + return []User{}, nil + } + pattern := "%" + query + "%" + rows, err := s.pool.Query(ctx, ` + SELECT id, username, display_name, password_hash, avatar, created_at + FROM users + WHERE username ILIKE $1 OR display_name ILIKE $1 + ORDER BY username + LIMIT $2 + `, pattern, limit) + 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.Username, &u.DisplayName, &u.PasswordHash, &u.Avatar, &u.CreatedAt, + ); err != nil { + return nil, err + } + users = append(users, u) + } + return users, nil +}