diff --git a/backend/docs/_index.md b/backend/docs/_index.md index 0862d73..a75c592 100644 --- a/backend/docs/_index.md +++ b/backend/docs/_index.md @@ -6,18 +6,18 @@ The `docs` agent reads this index first to locate the right file before diving i | Topic | File | Source files covered | |---|---|---| | Startup lifecycle & dependency initialisation | [bootstrap.md](bootstrap.md) | `internal/bootstrap/bootstrap.go`, `internal/server/server.go`, `cmd/api/main.go` | -| Database connection & query patterns | [database.md](database.md) | `internal/infrastructure/database/postgres/db.go`, `internal/infrastructure/database/postgres/health_repository.go`, `internal/domain/health.go`, `internal/usecase/health_usecase.go` | +| Database connection & query patterns | [database.md](database.md) | `internal/infrastructure/database/postgres/db.go`, `internal/infrastructure/database/postgres/health_repository.go`, `internal/infrastructure/database/postgres/user_repository.go`, `internal/domain/health.go`, `internal/domain/user.go`, `internal/domain/pagination.go`, `internal/usecase/health_usecase.go`, `internal/usecase/user.go` | | Schema migrations (goose) | [migrations.md](migrations.md) | `cmd/migrate/main.go`, `internal/infrastructure/database/migrations/`, `Makefile` | -| HTTP routing & handler patterns | [routing.md](routing.md) | `internal/transport/handlers/handler.go`, `internal/transport/handlers/routes.go`, `internal/transport/handlers/hello_handler.go`, `internal/transport/handlers/health_handler.go`, `internal/transport/middleware/logger.go`, `internal/server/server.go` | +| HTTP routing & handler patterns | [routing.md](routing.md) | `internal/transport/handlers/handler.go`, `internal/transport/handlers/routes.go`, `internal/transport/handlers/hello_handler.go`, `internal/transport/handlers/health_handler.go`, `internal/transport/handlers/me_handler.go`, `internal/transport/handlers/validation.go`, `internal/transport/middleware/logger.go`, `internal/server/server.go` | | Testing patterns (unit, handler, Redis, bootstrap) | [testing.md](testing.md) | `internal/infrastructure/database/postgres/health_repository_test.go`, `internal/transport/handlers/hello_handler_test.go`, `internal/transport/handlers/health_handler_test.go`, `internal/transport/middleware/logger_test.go`, `internal/usecase/health_usecase_test.go`, `internal/infrastructure/cache/redis/cache_test.go`, `internal/bootstrap/bootstrap_test.go` | -| Error handling conventions | [error-handling.md](error-handling.md) | `internal/infrastructure/database/postgres/health_repository.go`, `internal/transport/handlers/health_handler.go`, `cmd/api/main.go` | +| Error handling conventions | [error-handling.md](error-handling.md) | `internal/infrastructure/database/postgres/health_repository.go`, `internal/transport/handlers/health_handler.go`, `internal/transport/handlers/validation.go`, `cmd/api/main.go` | | Environment variables | [environment.md](environment.md) | `.env`, `internal/bootstrap/bootstrap.go`, `internal/infrastructure/database/postgres/db.go` | | Middleware (logger, rate limiter) | [middleware.md](middleware.md) | `internal/transport/middleware/logger.go`, `internal/transport/middleware/ratelimit.go`, `internal/transport/handlers/routes.go` | | Firebase Auth (token verification, middleware, MeHandler) | [auth.md](auth.md) | `internal/usecase/auth_usecase.go`, `internal/transport/middleware/auth.go`, `internal/transport/handlers/auth_handler.go`, `pkg/firebase/admin.go`, `internal/bootstrap/bootstrap.go` | | Observability (Sentry error tracking) | [observability.md](observability.md) | `internal/transport/middleware/sentry.go`, `internal/bootstrap/bootstrap.go`, `internal/transport/handlers/routes.go` | -| WebSocket (Hub, client, GET /ws, auth, wiring) | [websocket.md](websocket.md) | `internal/infrastructure/ws/`, `internal/transport/handlers/ws_handler.go`, `internal/transport/handlers/routes.go`, `internal/server/server.go`, `cmd/api/main.go` | -| Background job queue (Asynq, task definitions, worker, Asynqmon UI) | [queue.md](queue.md) | `internal/usecase/enqueuer.go`, `internal/infrastructure/queue/tasks.go`, `internal/infrastructure/queue/client.go`, `internal/infrastructure/queue/worker.go`, `internal/infrastructure/queue/handlers.go`, `internal/transport/handlers/routes.go`, `cmd/api/main.go` | -| Redis Streams event fan-out (producer, consumer, consumer groups) | [streams.md](streams.md) | `internal/infrastructure/streams/events.go`, `internal/infrastructure/streams/producer.go`, `internal/infrastructure/streams/consumer.go` | +| WebSocket (Hub, client, GET /ws, auth, bidirectional inbound handling, semaphore) | [websocket.md](websocket.md) | `internal/infrastructure/ws/`, `internal/transport/handlers/ws_handler.go`, `internal/transport/handlers/routes.go`, `internal/server/server.go`, `cmd/api/main.go` | +| Background job queue (Asynq, task definitions, worker, Asynqmon UI, NewHandle* constructor pattern) | [queue.md](queue.md) | `internal/usecase/enqueuer.go`, `internal/infrastructure/queue/tasks.go`, `internal/infrastructure/queue/client.go`, `internal/infrastructure/queue/worker.go`, `internal/infrastructure/queue/handlers.go`, `internal/transport/handlers/routes.go`, `cmd/api/main.go` | +| Redis Streams event fan-out (producer, consumer, consumer groups) | [streams.md](streams.md) | `internal/infrastructure/streams/events.go`, `internal/infrastructure/streams/producer.go`, `internal/infrastructure/streams/consumer.go`, `internal/usecase/streams.go` | | Firebase Cloud Messaging — token storage, send API, FCM endpoints | [fcm.md](fcm.md) | `internal/domain/fcm_token.go`, `internal/usecase/notification.go`, `internal/infrastructure/database/postgres/fcm_token_repository.go`, `internal/transport/handlers/fcm_handler.go`, `pkg/firebase/app.go`, `pkg/firebase/messaging.go` | | Transactional email (Mailjet) — EmailSender interface, MailjetSender, sandbox mode, templates | [email.md](email.md) | `internal/usecase/email.go`, `internal/infrastructure/email/mailjet.go`, `internal/infrastructure/email/templates/welcome.html`, `internal/bootstrap/bootstrap.go`, `internal/server/server.go`, `internal/transport/handlers/handler.go` | | Object storage (Cloudflare R2) — StorageService interface, R2 implementation, presign/delete endpoints | [storage.md](storage.md) | `internal/usecase/storage.go`, `internal/infrastructure/storage/r2/storage.go`, `internal/transport/handlers/storage_handler.go`, `internal/transport/handlers/routes.go`, `internal/bootstrap/bootstrap.go` | diff --git a/backend/docs/database.md b/backend/docs/database.md index 50e8bce..23918a3 100644 --- a/backend/docs/database.md +++ b/backend/docs/database.md @@ -1,11 +1,15 @@ --- topic: database -last_verified: 2026-06-15 +last_verified: 2026-06-23 sources: - internal/domain/health.go + - internal/domain/user.go + - internal/domain/pagination.go - internal/usecase/health_usecase.go + - internal/usecase/user.go - internal/infrastructure/database/postgres/db.go - internal/infrastructure/database/postgres/health_repository.go + - internal/infrastructure/database/postgres/user_repository.go --- # Database @@ -45,11 +49,14 @@ Env vars are loaded via `_ "github.com/joho/godotenv/autoload"` blank import in ``` internal/domain/health.go — HealthStats type +internal/domain/user.go — User type +internal/domain/pagination.go — Page[T], CursorPage[T], PageRequest internal/usecase/health_usecase.go — HealthReader interface (repo contract), HealthUseCase interface + impl -internal/infrastructure/database/postgres/ — HealthRepository: implements HealthReader against *sql.DB +internal/usecase/user.go — UserRepository interface +internal/infrastructure/database/postgres/ — HealthRepository, UserRepository: implement interfaces against *sql.DB ``` -The `HealthReader` interface is defined in the `usecase` package (Dependency Inversion — the use case owns the interface it depends on): +Repository interfaces are defined in the `usecase` package (Dependency Inversion — the use case owns the interface it depends on): ```go // usecase/health_usecase.go @@ -60,8 +67,75 @@ type HealthReader interface { type HealthUseCase interface { GetHealth(ctx context.Context) (domain.HealthStats, error) } + +// usecase/user.go +type UserRepository interface { + Upsert(ctx context.Context, u *domain.User) (*domain.User, error) + DeleteByFirebaseUID(ctx context.Context, firebaseUID string) error +} ``` +## Domain types + +```go +// internal/domain/user.go +type User struct { + ID int64 `json:"id"` + FirebaseUID string `json:"firebase_uid"` + Name string `json:"name"` + Email string `json:"email"` + PhotoURL string `json:"photo_url"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// internal/domain/pagination.go +type Page[T any] struct { + Items []T `json:"items"` + Total int `json:"total"` + Page int `json:"page"` + PageSize int `json:"page_size"` + HasMore bool `json:"has_more"` +} + +type CursorPage[T any] struct { + Items []T `json:"items"` + NextCursor string `json:"next_cursor,omitempty"` + HasMore bool `json:"has_more"` +} + +type PageRequest struct { + Page int `form:"page" binding:"min=0"` + PageSize int `form:"page_size" binding:"min=0,max=100"` +} + +func (p *PageRequest) Defaults() // fills zero values with page=1, size=20 +func (p PageRequest) Offset() int // returns SQL OFFSET value +``` + +## Users table schema +Added via migration `20260623063851_add_user_profile.sql`: + +| Column | Type | Constraints | +|---|---|---| +| `id` | `BIGSERIAL` | PRIMARY KEY | +| `firebase_uid` | `TEXT` | NOT NULL, UNIQUE | +| `name` | `TEXT` | — | +| `email` | `TEXT` | — | +| `photo_url` | `TEXT` | — | +| `created_at` | `TIMESTAMPTZ` | NOT NULL DEFAULT now() | +| `updated_at` | `TIMESTAMPTZ` | NOT NULL DEFAULT now() | + +## UserRepository +`internal/infrastructure/database/postgres/user_repository.go` — constructor returns the interface, not the concrete type: + +```go +func NewUserRepository(db *sql.DB) usecase.UserRepository +``` + +`Upsert` uses `INSERT ... ON CONFLICT (firebase_uid) DO UPDATE` and returns the full row via `RETURNING`. +`DeleteByFirebaseUID` deletes by `firebase_uid`; does not error when the row does not exist (DELETE is idempotent). + ## Repository pattern Each repository is a struct that holds `*sql.DB` and is constructed with a `New*` function. diff --git a/backend/docs/email.md b/backend/docs/email.md index b0925cc..b8e27c7 100644 --- a/backend/docs/email.md +++ b/backend/docs/email.md @@ -1,11 +1,15 @@ --- -title: Email (Mailjet) +topic: email last_verified: 2026-06-23 sources: - internal/usecase/email.go - internal/infrastructure/email/mailjet.go - internal/infrastructure/email/templates/welcome.html + - internal/infrastructure/queue/handlers.go + - internal/infrastructure/queue/tasks.go - internal/bootstrap/bootstrap.go + - internal/server/server.go + - internal/transport/handlers/handler.go --- # Email (Mailjet) @@ -52,6 +56,21 @@ var templateFS embed.FS `templates/welcome.html` is a Go `html/template` file. The only template data value currently used is `{{.Name}}` (the recipient's display name). `renderWelcomeTemplate` parses and executes the template on each call and returns the rendered HTML string. +### How welcome email is triggered + +The welcome email is sent via the Asynq queue (not directly from a handler). The Asynq +task handler `NewHandleWelcomeEmail(sender)` in `internal/infrastructure/queue/handlers.go` +calls `sender.SendWelcomeEmail` with the name from `WelcomeEmailPayload.Name`. +When `WelcomeEmailPayload.Name` is empty the handler falls back to `WelcomeEmailPayload.Email` +as the display name. + +```go +// Triggered when a TypeWelcomeEmail task is dequeued. +func NewHandleWelcomeEmail(sender usecase.EmailSender) asynq.HandlerFunc +``` + +`sender` is nil when `MAILJET_API_KEY` is not set; the handler no-ops gracefully in that case. + ### Wiring (bootstrap) `bootstrap.Run` constructs the sender when both `MAILJET_API_KEY` and `MAILJET_SECRET_KEY` are non-empty: diff --git a/backend/docs/error-handling.md b/backend/docs/error-handling.md index 4cd6004..c0ab667 100644 --- a/backend/docs/error-handling.md +++ b/backend/docs/error-handling.md @@ -4,6 +4,7 @@ last_verified: 2026-06-23 sources: - internal/infrastructure/database/postgres/health_repository.go - internal/transport/handlers/health_handler.go + - internal/transport/handlers/validation.go - cmd/api/main.go --- @@ -71,16 +72,25 @@ func (h *Handler) getItemHandler(c *gin.Context) { Never expose internal error messages to clients. Log the original error server-side. ## Request binding errors -Always validate and return 400 on bad input: +Use the shared helpers in `internal/transport/handlers/validation.go` — do not call `c.ShouldBindJSON` / `c.ShouldBindQuery` directly in handlers. ```go -var input MyRequest -if err := c.ShouldBindJSON(&input); err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) +// internal/transport/handlers/validation.go +func bindJSON(c *gin.Context, dst any) bool // writes 400 {"error":"invalid request body"} on failure +func bindQuery(c *gin.Context, dst any) bool // writes 400 {"error":"invalid query parameters"} on failure +``` + +Both helpers return `false` and write the 400 response on failure, so the handler just returns immediately: + +```go +var req updateMeRequest +if !bindJSON(c, &req) { return } ``` +The stable message strings (`"invalid request body"`, `"invalid query parameters"`) never expose raw validation errors to the client. + ## Error wrapping Use `fmt.Errorf("context: %w", err)` when adding context to returned errors so callers can use `errors.Is` / `errors.As`. diff --git a/backend/docs/migrations.md b/backend/docs/migrations.md index 39f9b7e..0365023 100644 --- a/backend/docs/migrations.md +++ b/backend/docs/migrations.md @@ -1,6 +1,6 @@ --- topic: migrations -last_verified: 2026-06-15 +last_verified: 2026-06-23 sources: - cmd/migrate/main.go - internal/infrastructure/database/migrations/ @@ -46,10 +46,32 @@ DROP TABLE users; Rules: - Every statement must end with `;` -- DDL only — no application data mutations in migrations +- DDL only — no application data mutations in migrations, with one exception: `UPDATE` backfills are permitted as a step between `ADD COLUMN` (nullable) and `SET NOT NULL` (see pattern below) - For multi-statement blocks (PL/pgSQL), wrap with `-- +goose StatementBegin` / `-- +goose StatementEnd` - Avoid `-- +goose NO TRANSACTION` unless the statement genuinely cannot run in a transaction (e.g. `CREATE INDEX CONCURRENTLY`) +## Adding a NOT NULL column to an existing table +When adding a `NOT NULL` column to a table that may already have rows, follow the three-step pattern from `20260623063851_add_user_profile.sql`: + +```sql +-- +goose Up +-- Step 1: add as nullable +ALTER TABLE users ADD COLUMN IF NOT EXISTS firebase_uid TEXT; + +-- Step 2: backfill existing rows so SET NOT NULL will succeed +UPDATE users SET firebase_uid = 'legacy-' || id::text WHERE firebase_uid IS NULL; + +-- Step 3: apply the constraint and unique index +ALTER TABLE users ALTER COLUMN firebase_uid SET NOT NULL; +ALTER TABLE users ADD CONSTRAINT users_firebase_uid_key UNIQUE (firebase_uid); + +-- +goose Down +ALTER TABLE users DROP CONSTRAINT IF EXISTS users_firebase_uid_key; +ALTER TABLE users DROP COLUMN IF EXISTS firebase_uid; +``` + +This avoids the `ERROR: column contains null values` failure that occurs when you add a `NOT NULL` column with no default and the table is non-empty. + ## Workflow for any new table 1. `make migrate-create name=add_` — generates the timestamped file 2. Fill in `CREATE TABLE` (Up) and `DROP TABLE` (Down) diff --git a/backend/docs/queue.md b/backend/docs/queue.md index abc6129..0e4158a 100644 --- a/backend/docs/queue.md +++ b/backend/docs/queue.md @@ -1,6 +1,6 @@ --- topic: queue -last_verified: 2026-06-15 +last_verified: 2026-06-23 sources: - internal/usecase/enqueuer.go - internal/infrastructure/queue/tasks.go @@ -36,9 +36,12 @@ const ( type WelcomeEmailPayload struct { UserID string `json:"user_id"` Email string `json:"email"` + Name string `json:"name"` } ``` +`Name` is used by `NewHandleWelcomeEmail` as the display name in the welcome email. When empty, the handler falls back to `Email`. + Both the enqueuer (caller side) and the handler (worker side) import these constants so the string is never duplicated. @@ -116,7 +119,7 @@ if app.Config.RedisURL != "" { workerCancel = wCancel worker, err := queue.NewWorker(app.Config.RedisURL) // ... - worker.Register(queue.TypeWelcomeEmail, asynq.HandlerFunc(queue.HandleWelcomeEmail)) + worker.Register(queue.TypeWelcomeEmail, queue.NewHandleWelcomeEmail(app.EmailSender)) go func() { if err := worker.Run(workerCtx); err != nil { slog.Error("queue: worker error", "err", err) @@ -139,31 +142,32 @@ can still reach the hub during drain. ## Task handlers -Handler functions have the signature `func(context.Context, *asynq.Task) error` and live -in `internal/infrastructure/queue/handlers.go`: +Handler functions live in `internal/infrastructure/queue/handlers.go`. +They are constructed via a `New*` function that closes over dependencies (e.g. `EmailSender`), +rather than being plain `func(context.Context, *asynq.Task) error` functions. ```go -func HandleWelcomeEmail(_ context.Context, t *asynq.Task) error { - var p WelcomeEmailPayload - if err := json.Unmarshal(t.Payload(), &p); err != nil { - return fmt.Errorf("welcome email: unmarshal payload: %w", err) - } - slog.Info("queue: welcome email task received", "user_id", p.UserID, "email", p.Email) - return nil -} +// NewHandleWelcomeEmail returns an asynq.HandlerFunc that sends a welcome email via sender. +// If sender is nil, the task is acknowledged without sending (graceful degradation). +func NewHandleWelcomeEmail(sender usecase.EmailSender) asynq.HandlerFunc ``` +Internally it unmarshals `WelcomeEmailPayload`, falls back to `p.Email` when `p.Name` is +empty, then calls `sender.SendWelcomeEmail`. A nil `sender` is accepted silently so the +worker starts cleanly even when Mailjet is not configured. + Returning a non-nil error causes Asynq to retry the task (up to its configured retry limit). ## Adding a new task 1. Add a `Type = ":"` constant and `Payload` struct to `internal/infrastructure/queue/tasks.go`. -2. Write a `Handle(ctx context.Context, t *asynq.Task) error` function in - `internal/infrastructure/queue/handlers.go`. +2. Write a `NewHandle(dep SomeDep) asynq.HandlerFunc` constructor in + `internal/infrastructure/queue/handlers.go`. Close over any dependencies (e.g. `EmailSender`). + The returned `HandlerFunc` should gracefully no-op when a nil dependency is passed. 3. Register the handler in `cmd/api/main.go`: ```go - worker.Register(queue.Type, asynq.HandlerFunc(queue.Handle)) + worker.Register(queue.Type, queue.NewHandle(dep)) ``` 4. Enqueue from the relevant use case or handler using `app.Enqueuer.Enqueue(ctx, queue.Type, payload)`. @@ -189,12 +193,13 @@ The UI is not mounted in staging or production (`gin.ReleaseMode`). ## Testing -**Unit tests** call handler functions directly with `asynq.NewTask` — no Redis required: +**Unit tests** call the constructed handler directly with `asynq.NewTask` — no Redis required: ```go -payload, _ := json.Marshal(queue.WelcomeEmailPayload{UserID: "u1", Email: "a@b.com"}) +payload, _ := json.Marshal(queue.WelcomeEmailPayload{UserID: "u1", Email: "a@b.com", Name: "Alice"}) task := asynq.NewTask(queue.TypeWelcomeEmail, payload) -err := queue.HandleWelcomeEmail(context.Background(), task) +handler := queue.NewHandleWelcomeEmail(mockSender) // or nil to test graceful degradation +err := handler(context.Background(), task) // assert err == nil ``` diff --git a/backend/docs/routing.md b/backend/docs/routing.md index 37df7ed..84a1e12 100644 --- a/backend/docs/routing.md +++ b/backend/docs/routing.md @@ -7,6 +7,8 @@ sources: - internal/transport/handlers/hello_handler.go - internal/transport/handlers/health_handler.go - internal/transport/handlers/auth_handler.go + - internal/transport/handlers/me_handler.go + - internal/transport/handlers/validation.go - internal/transport/middleware/logger.go - internal/server/server.go - cmd/api/main.go @@ -18,14 +20,18 @@ sources: ```go // internal/transport/handlers/handler.go type Handler struct { - healthUC usecase.HealthUseCase - verifier usecase.FirebaseTokenVerifier // nil disables auth (dev only) - hub *ws.Hub - enqueuer usecase.Enqueuer // nil when REDIS_URL is not set - queueUI http.Handler // nil disables /admin/queues route - fcmSender usecase.NotificationSender // nil when Firebase is not configured - fcmTokenRepo usecase.FCMTokenRepository // nil when Firebase is not configured - emailSender usecase.EmailSender // nil when MAILJET_API_KEY/SECRET_KEY are not set + healthUC usecase.HealthUseCase + verifier usecase.FirebaseTokenVerifier // nil disables auth (dev only) + hub *ws.Hub + enqueuer usecase.Enqueuer // nil when REDIS_URL is not set + queueUI http.Handler // nil disables /admin/queues route + fcmSender usecase.NotificationSender // nil when Firebase is not configured + fcmTokenRepo usecase.FCMTokenRepository // nil when Firebase is not configured + emailSender usecase.EmailSender // nil when MAILJET_API_KEY is not set + storageService usecase.StorageService // nil when R2_ACCOUNT_ID is not set + geoLocator usecase.GeoLocator // nil when geo client is not configured + streamProducer usecase.StreamProducer // nil when REDIS_URL is not set + userRepo usecase.UserRepository // nil when not wired } func NewHandler( @@ -37,9 +43,13 @@ func NewHandler( fcmSender usecase.NotificationSender, fcmTokenRepo usecase.FCMTokenRepository, emailSender usecase.EmailSender, + storageService usecase.StorageService, + geoLocator usecase.GeoLocator, + streamProducer usecase.StreamProducer, + userRepo usecase.UserRepository, ) *Handler ``` -The `Handler` struct holds use case interfaces and infrastructure dependencies — not `*sql.DB` directly. `verifier` is stored on the struct (not passed to `RegisterRoutes`) so the WebSocket handler can read it inline for query-param auth. `fcmTokenRepo` and `fcmSender` are nil when `FIREBASE_PROJECT_ID` is not set; their routes are only registered when non-nil. +The `Handler` struct holds use case interfaces and infrastructure dependencies — not `*sql.DB` directly. `verifier` is stored on the struct (not passed to `RegisterRoutes`) so the WebSocket handler can read it inline for query-param auth. `fcmTokenRepo` and `fcmSender` are nil when `FIREBASE_PROJECT_ID` is not set; their routes are only registered when non-nil. `userRepo` is always wired (constructed unconditionally in `server.NewServer`). ## Wiring (server.go) `internal/server/server.go` contains `NewServer(app *bootstrap.App, hub *ws.Hub) (*http.Server, error)` — wiring only, no logic. @@ -54,12 +64,14 @@ if app.Firebase != nil { fcmTokenRepo = postgres.NewFCMTokenRepository(app.DB) } +userRepo := postgres.NewUserRepository(app.DB) + var queueUI http.Handler if app.Config.RedisURL != "" { // parse URL and build asynqmon.New(...) } -h := handlers.NewHandler(healthUC, app.Firebase, hub, app.Enqueuer, queueUI, app.FCMSender, fcmTokenRepo, app.EmailSender) +h := handlers.NewHandler(healthUC, app.Firebase, hub, app.Enqueuer, queueUI, app.FCMSender, fcmTokenRepo, app.EmailSender, app.StorageService, app.GeoLocator, app.StreamProducer, userRepo) // Register DB pool metrics collector (AlreadyRegisteredError is silenced). prometheus.Register(postgres.NewDBStatsCollector(app.DB)) @@ -119,13 +131,25 @@ func (h *Handler) RegisterRoutes(rps float64, burst int, sentryDSN string) http. if h.verifier != nil { api.Use(middleware.FirebaseAuth(h.verifier)) } + if h.geoLocator != nil { + api.Use(middleware.GeoFromRequest(h.geoLocator)) + } api.GET("/me", h.MeHandler) + if h.userRepo != nil { + api.PATCH("/me", h.UpdateMeHandler) + api.DELETE("/me", h.DeleteMeHandler) + } if h.fcmTokenRepo != nil { api.POST("/fcm/register", h.RegisterFCMToken) api.DELETE("/fcm/unregister", h.UnregisterFCMToken) } + if h.storageService != nil { + api.POST("/storage/presign", h.PresignHandler) + api.DELETE("/storage/:key", h.DeleteObjectHandler) + } + return r } ``` @@ -160,10 +184,16 @@ Allowed methods: GET, POST, PUT, DELETE, OPTIONS, PATCH. | GET | `/swagger/*any` | none | Swagger UI | `routes.go` | | GET | `/admin/queues` | none (debug mode only) | Asynqmon job-monitoring UI | `routes.go` | | GET | `/api/v1/me` | FirebaseAuth header | `MeHandler` — returns verified `FirebaseToken` claims | `auth_handler.go` | +| PATCH | `/api/v1/me` | FirebaseAuth header | `UpdateMeHandler` — upserts user profile; returns `domain.User` | `me_handler.go` | +| DELETE | `/api/v1/me` | FirebaseAuth header | `DeleteMeHandler` — deletes user profile record; 204 on success | `me_handler.go` | | POST | `/api/v1/fcm/register` | FirebaseAuth header | `RegisterFCMToken` — stores device FCM token | `fcm_handler.go` | | DELETE | `/api/v1/fcm/unregister` | FirebaseAuth header | `UnregisterFCMToken` — removes device FCM token | `fcm_handler.go` | +| POST | `/api/v1/storage/presign` | FirebaseAuth header | `PresignHandler` — returns a presigned R2 upload URL | `storage_handler.go` | +| DELETE | `/api/v1/storage/:key` | FirebaseAuth header | `DeleteObjectHandler` — deletes an R2 object | `storage_handler.go` | FCM routes are only registered when `h.fcmTokenRepo != nil` (i.e., `FIREBASE_PROJECT_ID` is set). +Storage routes are only registered when `h.storageService != nil` (i.e., `R2_ACCOUNT_ID` is set). +`PATCH /api/v1/me` and `DELETE /api/v1/me` are registered when `h.userRepo != nil` (always wired). ## Graceful shutdown Wired in `cmd/api/main.go` via `signal.NotifyContext` for SIGINT/SIGTERM. @@ -181,5 +211,5 @@ Do not add shutdown logic to `internal/` — it belongs in `cmd/`. 6. Register the route in `RegisterRoutes()`: `r.METHOD("/path", h.handlerName)`. 7. Add handler as `func (h *Handler) handlerName(c *gin.Context)` in its own file. 8. Always pass `c.Request.Context()` to use case calls. -9. For request body: bind with `c.ShouldBindJSON(&input)`, return 400 on error. -10. For path params: `c.Param("id")`. For query params: `c.Query("key")`. +9. For request body: use the `bindJSON(c, &input) bool` helper (returns `false` and writes 400 on error). For query params: use `bindQuery(c, &input) bool`. Both helpers are defined in `internal/transport/handlers/validation.go` and return stable error messages ("invalid request body" / "invalid query parameters") rather than exposing raw validation errors. +10. For path params: `c.Param("id")`. diff --git a/backend/docs/streams.md b/backend/docs/streams.md index b2b4c36..8830be7 100644 --- a/backend/docs/streams.md +++ b/backend/docs/streams.md @@ -1,10 +1,12 @@ --- topic: streams -last_verified: 2026-06-15 +last_verified: 2026-06-23 sources: - internal/infrastructure/streams/events.go - internal/infrastructure/streams/producer.go - internal/infrastructure/streams/consumer.go + - internal/usecase/streams.go + - cmd/api/main.go --- # Redis Streams @@ -24,6 +26,45 @@ Streams complement Asynq rather than replace it: All Streams code lives in `internal/infrastructure/streams/`. +## StreamProducer interface + +`internal/usecase/streams.go` defines the interface used by handlers and use cases: + +```go +type StreamProducer interface { + Publish(ctx context.Context, stream string, event any) error + Close() error +} +``` + +`*streams.Producer` implements this interface. `app.StreamProducer` is closed during graceful +shutdown in `cmd/api/main.go` after the HTTP server and worker have stopped. + +## Runtime wiring status + +**Producer:** wired — `app.StreamProducer` is set when `REDIS_URL` is non-empty. It is +passed to `handlers.NewHandler` and is available for use in handlers and use cases. + +**Consumer:** intentionally NOT started at runtime. The consumer infrastructure is fully +implemented but no consumer goroutine is registered in `cmd/api/main.go`. The commented-out +wiring block in `main.go` serves as the canonical example for adding a consumer: + +```go +// streamCtx, streamCancel := context.WithCancel(context.Background()) +// consumer, err := streams.NewConsumer(app.Config.RedisURL, streams.StreamUserCreated, "api", "api-1") +// if err != nil { ... } +// go func() { +// _ = consumer.Run(streamCtx, func(ctx context.Context, data []byte) error { +// var evt streams.UserCreatedEvent +// if err := json.Unmarshal(data, &evt); err != nil { return err } +// payload, _ := json.Marshal(queue.WelcomeEmailPayload{UserID: evt.UserID, Email: evt.Email, Name: evt.Name}) +// return app.Enqueuer.Enqueue(ctx, queue.TypeWelcomeEmail, payload) +// }) +// _ = consumer.Close() +// }() +// // In shutdown: streamCancel() +``` + ## Stream names Stream name constants and event payload structs are defined in @@ -38,6 +79,7 @@ const ( type UserCreatedEvent struct { UserID string `json:"user_id"` Email string `json:"email"` + Name string `json:"name"` } ``` @@ -74,6 +116,7 @@ producer, err := streams.NewProducer(redisURL) err = producer.Publish(ctx, streams.StreamUserCreated, streams.UserCreatedEvent{ UserID: user.ID, Email: user.Email, + Name: user.Name, }) ``` @@ -108,24 +151,31 @@ group name, and a unique consumer name within the group. 7. On `redis.Nil` (timeout with no messages): continues the loop. 8. On other Redis errors: logs and continues. +### Idle-connection hardening + +`NewConsumer` sets `MaxIdleConns=1` and `ConnMaxIdleTime=8s` on the Redis client options. +This proactively recycles idle connections before managed Redis providers (Upstash, Redis +Cloud) kill them at their own idle timeout (~10–60 s). Without this, long-blocking +`XReadGroup` calls surface as `i/o timeout` errors from the pool. + +`Run` also handles `net.Error` timeouts explicitly: it logs a warning and backs off 2 s +before reconnecting, rather than exiting or spinning tightly. + ### Example — registering a consumer goroutine in main.go +See the commented-out wiring block in `cmd/api/main.go` (reproduced in the Runtime wiring +status section above). The key pattern: + ```go -consumer, err := streams.NewConsumer(app.Config.RedisURL, - streams.StreamUserCreated, "api-group", "api-consumer-1") -if err != nil { - // handle -} +streamCtx, streamCancel := context.WithCancel(context.Background()) +consumer, _ := streams.NewConsumer(redisURL, streams.StreamUserCreated, "api", "api-1") go func() { - if err := consumer.Run(ctx, handleUserCreated); err != nil { - slog.Error("streams: consumer error", "err", err) - } + _ = consumer.Run(streamCtx, func(ctx context.Context, data []byte) error { ... }) + _ = consumer.Close() }() +// In shutdown sequence: streamCancel() ``` -Cancel `ctx` (the same child context used for the worker) to stop the consumer during -shutdown, then call `consumer.Close()` to release the connection. - ## Adding a new event type 1. Add a `Stream = "stream:."` constant to @@ -146,7 +196,7 @@ Typical test flow: ```go producer, _ := streams.NewProducer(redisURL) producer.Publish(ctx, streams.StreamUserCreated, streams.UserCreatedEvent{ - UserID: "u1", Email: "a@b.com", + UserID: "u1", Email: "a@b.com", Name: "Alice", }) producer.Close() diff --git a/backend/docs/websocket.md b/backend/docs/websocket.md index 3eeed84..53d0706 100644 --- a/backend/docs/websocket.md +++ b/backend/docs/websocket.md @@ -21,59 +21,80 @@ A `Hub` runs as a long-lived goroutine and fans out messages to all connected cl The `GET /ws` endpoint upgrades HTTP connections; a Firebase ID token is required as a query parameter. -## Message envelope +## Message types -All messages use a typed JSON envelope defined in `internal/infrastructure/ws/message.go`: +`internal/infrastructure/ws/message.go` defines the wire format and inbound handler types: ```go +// Outbound (server → client) and inbound (client → server) wire format. type Envelope struct { Type string `json:"type"` Payload json.RawMessage `json:"payload"` } + +// InboundMessage wraps an Envelope with the originating client ID. +type InboundMessage struct { + ClientID string + Msg Envelope +} + +// InboundHandler processes an inbound WebSocket message. +type InboundHandler func(ctx context.Context, msg InboundMessage) error ``` `Type` is a dot-separated event name (e.g. `"job.completed"`). `Payload` is arbitrary JSON whose shape is determined by `Type`. +The same `Envelope` format is used for both directions — clients send JSON frames in this +shape, and the server broadcasts frames in this shape. + ## Hub `internal/infrastructure/ws/hub.go` ```go -type Hub struct { - clients map[*Client]struct{} - broadcast chan []byte // buffered, capacity 256 - Register chan *Client - Unregister chan *Client -} - func NewHub() *Hub -func (h *Hub) Run(ctx context.Context) // blocking; cancel ctx to stop -func (h *Hub) Publish(msgType string, payload any) error +func (h *Hub) Run(ctx context.Context) // blocking; cancel ctx to stop +func (h *Hub) Publish(msgType string, payload any) error // server → all clients +func (h *Hub) OnMessage(msgType string, handler InboundHandler) // register inbound handler ``` `Run` must be called in its own goroutine and runs until `ctx` is cancelled. -`Publish` marshals `payload` into an `Envelope` and queues it for broadcast — -safe to call from any goroutine (e.g. an Asynq worker in future #18). +`Publish` marshals `payload` into an `Envelope` and queues it for broadcast — safe to call +from any goroutine. +`OnMessage` registers a handler for a specific inbound message type. Safe to call before +or after `Run` starts (protected by `sync.RWMutex`). ### Goroutine model -Each WebSocket connection spawns two goroutines: `ReadPump` and `WritePump` (on `Client`). -The Hub serialises all mutations (register / unregister / broadcast) through a `select` loop -so no locking is needed on its internal `clients` map. +The Hub is now fully bidirectional. Each WebSocket connection spawns `ReadPump` and `WritePump`. +`ReadPump` parses each inbound frame as an `Envelope` and forwards it to `hub.inbound`. +`Run` dispatches inbound messages to registered handlers via a 64-slot semaphore-bounded +goroutine pool so slow handlers do not stall the event loop. ```text + WebSocket conn ──► client.ReadPump goroutine ──► hub.inbound chan + │ + hub.Run goroutine ──► InboundHandler goroutine (≤64 concurrent) + caller goroutine │ hub.Publish(...) ▼ hub.broadcast chan │ hub.Run goroutine ──► client.Send chan ──► client.WritePump goroutine ──► WebSocket conn - client.ReadPump goroutine ──► (discards incoming / handles pings) ``` -Slow clients are dropped: if `client.Send` is full, the Hub closes the channel and -removes the client without blocking the broadcast loop. +Slow clients are dropped: if `client.Send` is full, the Hub closes the channel and removes +the client without blocking the broadcast loop. + +When `hub.inbound` is full, the message is dropped and a warning is logged (non-blocking). +When the semaphore is full (64 in-flight handlers), additional inbound messages are dropped. + +### Lifecycle context + +The `ctx` passed to `hub.Run` is forwarded to each `InboundHandler` goroutine. This means +handlers can respect cancellation and use it for downstream calls (e.g. `db.QueryContext`). ## Client @@ -109,7 +130,8 @@ cannot set `Authorization` headers. 2. If `h.verifier == nil` (development): skips auth — connects immediately. After successful auth, the connection is upgraded and `ReadPump` / `WritePump` are -started in separate goroutines. +started in separate goroutines. `ReadPump` now parses each inbound frame as an `Envelope` +and routes it to the Hub — malformed frames are logged and discarded (the connection stays open). ## Wiring in server.go and main.go @@ -135,9 +157,9 @@ hubCancel() // stop hub after server drains connections The canonical `Handler` struct definition and `NewHandler` signature (including all fields beyond `hub` and `verifier`) are documented in `backend/docs/routing.md`. -## Publishing events from workers (future #18) +## Publishing events from workers -Call `hub.Publish` from any goroutine: +Call `hub.Publish` from any goroutine (Asynq workers, Redis Streams consumers, etc.): ```go hub.Publish("job.completed", map[string]any{ @@ -146,8 +168,19 @@ hub.Publish("job.completed", map[string]any{ }) ``` -When #18 (Asynq + Redis Streams) lands, the Asynq task handlers and Redis Streams -consumers will call this method to push domain events to connected clients. +## Handling inbound messages + +Register an `InboundHandler` before or after calling `hub.Run`: + +```go +hub.OnMessage("ping", func(ctx context.Context, msg ws.InboundMessage) error { + // msg.ClientID is the remote address of the originating client + // msg.Msg.Payload contains the raw JSON payload from the client + return hub.Publish("pong", map[string]any{"client": msg.ClientID}) +}) +``` + +Handlers must be non-blocking or return quickly; the 64-slot semaphore limits concurrency. ## Testing @@ -156,6 +189,7 @@ Unit tests in `internal/infrastructure/ws/hub_test.go` cover: - `TestHub_UnregisterRemovesClient` — Send channel is closed on unregister - `TestHub_ContextCancelClosesSendChannels` — all channels closed on ctx cancel - `TestHub_ConcurrentClientsAndBroadcast` — 10 concurrent clients all receive +- `TestHub_OnMessage` — `InboundHandler` is invoked with correct `InboundMessage` - `TestHub_Publish` — JSON marshalling and delivery Tests inject `*Client` with a nil `conn` and a buffered `Send` channel; the Hub only diff --git a/web/AGENTS.md b/web/AGENTS.md index 47c2275..3033cd6 100644 --- a/web/AGENTS.md +++ b/web/AGENTS.md @@ -24,14 +24,26 @@ pnpm test:ui # Vitest browser UI ## Project structure ``` -app/ # Next.js App Router — routes only - layout.tsx # root layout: Geist fonts, global CSS, / - page.tsx # home page (Server Component) - globals.css # Tailwind v4 import + CSS variable definitions -public/ # static assets -components/ # shared UI components (create when needed) -lib/ # utilities and non-React helpers (create when needed) -types/ # shared TypeScript type definitions (create when needed) +app/ # Next.js App Router — routes only (no business logic) + layout.tsx # root layout: Geist fonts, global CSS, / + page.tsx # home page (Server Component) + globals.css # Tailwind v4 import + CSS variable definitions + (auth)/ # route group for unauthenticated pages (login, register…) + (dashboard)/ # route group for authenticated/protected pages +components/ + layout/ # AppShell, Sidebar, Navbar, Footer + common/ # reusable cross-feature UI (Avatar, Badge, Spinner…) + home/ # landing / home page components + ui/ # shadcn primitives — populated by CLI, do not hand-create +features/ # one subfolder per domain feature (auth/, notifications/…) + / + components/ # feature-scoped UI + hooks/ # feature-scoped React hooks + types.ts # feature-scoped TypeScript types +lib/ # pure utilities and non-React helpers +public/ # static assets +server/ # tRPC routers and server-side logic +types/ # shared TypeScript interfaces used across features ``` --- diff --git a/web/app/(auth)/.gitkeep b/web/app/(auth)/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/web/app/(dashboard)/.gitkeep b/web/app/(dashboard)/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/web/components/common/.gitkeep b/web/components/common/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/web/components/home/.gitkeep b/web/components/home/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/web/components/layout/.gitkeep b/web/components/layout/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/web/components/ui/.gitkeep b/web/components/ui/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/web/features/.gitkeep b/web/features/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/web/server/.gitkeep b/web/server/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/web/types/.gitkeep b/web/types/.gitkeep new file mode 100644 index 0000000..e69de29