diff --git a/.env.example b/.env.example index 334977c..eeb865a 100644 --- a/.env.example +++ b/.env.example @@ -37,3 +37,21 @@ DATABASE_URL=postgres://postgres:postgres@db:5432/guestbook?sslmode=disable # DB_PASSWORD=postgres # DB_NAME=guestbook # DB_SSL_MODE=disable + +# === Scaling / live updates ======== +# Redis for cross-replica live updates and presence. Without it each replica +# only pushes to its own clients. On Miabi, attach a managed Redis with the +# "REDIS" env prefix and REDIS_DATABASE_URL is injected (also accepted). +# REDIS_URL=redis://redis:6379 + +# === Admin console ================= +# Enables /admin. Store as a secret. Generate with: openssl rand -hex 16 +# ADMIN_TOKEN= + +# CIDRs whose X-Forwarded-For is trusted for the signer IP shown in /admin. +# Unset trusts the header as-is (fine behind the Miabi gateway). +# TRUSTED_PROXIES=10.0.0.0/8 + +# Expose /api/debug/* failure injection (crash, slow, 5xx, CPU, memory, logs). +# Still requires admin auth. Demo environments only. +# DEBUG_ENDPOINTS=false diff --git a/DEMO.md b/DEMO.md new file mode 100644 index 0000000..926d4f9 --- /dev/null +++ b/DEMO.md @@ -0,0 +1,146 @@ +# Guestbook demo runbook + +A step-by-step script for showing Miabi with the guestbook. Each act builds on +the previous one and takes 3–8 minutes. Skip acts freely: everything after +act 2 only assumes the app is deployed. + +Keep two windows open: the **wall** (`https:///`) and the **admin +console** (`https:///admin`). A third tab on the Miabi app page is +handy for logs, events and scaling. + +| Act | Shows | Miabi features | +|-----|-------|----------------| +| 1 | Deploy from Git / image | pipeline, build, domain + TLS | +| 2 | Managed Postgres | databases, env injection | +| 3 | Secrets + admin console | secret env vars | +| 4 | Scale out → broken live updates | replicas | +| 5 | Attach Redis → fixed | managed Redis, env prefix | +| 6 | Break it on purpose | restarts, health gating, alerts, logs, analytics | +| 7 | Canary v1 → v2 | canary weights, promote | +| 8 | Scheduled cleanup | cron jobs | + +--- + +## 0. Prepare (before the session) + +```bash +docker build --build-arg VERSION=1.0.0 -t /guestbook:1.0.0 . +docker build --build-arg VERSION=2.0.0 -t /guestbook:2.0.0 . +docker push /guestbook:1.0.0 && docker push /guestbook:2.0.0 +``` + +- Pick an admin token: `openssl rand -hex 16`. +- Create an empty Postgres and a Redis in the workspace ahead of time, since provisioning is slower than talking. +- Open `scripts/hammer.sh` in a terminal (see act 7). + +## 1. Deploy + +1. **New application**, from the Git repo (the `.miabi/pipeline.yaml` builds, scans and deploys) or from image `guestbook:1.0.0`. +2. Port `8080`. **Health check**: HTTP, path `/healthz`. (The default is *none*, and without it act 6 loses its punch.) +3. Add a domain. TLS is issued automatically. +4. Open the wall. It runs on **SQLite**, as the first seeded signature says. + +> Talking point: zero config. With no database attached the app falls back +> to a local SQLite file. Mount a volume at `/data` if you want it to survive +> redeploys. + +## 2. Attach Postgres + +1. App → **Databases** → attach the Postgres, **no prefix**. +2. Miabi injects `DATABASE_URL` (as a secret reference) plus `DB_*` and flags the app for redeploy. +3. Redeploy. The app migrates the schema and seeds the wall on first boot. The welcome signature now says *PostgreSQL*. + +## 3. Secrets and the admin console + +1. App → **Environment** → add `ADMIN_TOKEN` = your token, marked **secret**. Add `DEBUG_ENDPOINTS=true` too (used in act 6). +2. Redeploy, open `/admin` and sign in with the token. +3. Tour it: + - **Overview**: signatures per day, replicas, and *this replica*'s runtime (host, DB, uptime, memory). + - **Moderation**: each signature shows the signer's IP, forwarded by the gateway. Click one to filter by it. Pin a signature. It jumps to the top of every open wall, live. Hide one and it disappears. + - **Settings**: set a banner ("Welcome to the Miabi training 👋"), then toggle **Pause signing**. The wall goes read-only instantly in every tab. + +> Talking point: the token never appears in the app's config in plain text. +> Rotating it (edit + redeploy) signs every admin out, because sessions are +> HMAC-signed with it. + +Ask the audience to open the wall on their phones and sign it. The +**online** counter climbs live. + +## 4. Scale out, and watch it break + +1. App → **Replicas** → scale to **3**. +2. In the admin **Overview**, the *served by* chip changes as you refresh, but the **Replicas** table only ever lists one host. The hint says *in-process broker*. +3. Open the wall in two browsers (or phones) until their clock cards show **different hosts**. Sign in one: **the other doesn't update.** Each replica only pushes to its own SSE clients. +4. Also note the `via replica-x` tag on new signatures: each one records which replica accepted it. + +> Talking point: this is the classic "works on one box" bug. Stateless apps +> need shared state for fan-out. Scaling is a platform feature; being +> scalable is an app property. + +## 5. Attach Redis, and it's fixed + +1. App → **Databases** → attach the Redis with prefix **`REDIS`**. Miabi injects `REDIS_DATABASE_URL`, which the app reads (so does `REDIS_URL`). +2. Redeploy. Startup logs show `live updates ready broker=redis`. +3. Repeat the two-browser test: signatures now appear everywhere. The **Replicas** table lists all 3 hosts with their client counts, and the wall header says *3 replicas*. +4. **Chaos lab → Traffic probe** → *Run probe*: the bars show requests spread across the three hosts. + +## 6. Break it on purpose + +All buttons are in the admin **Chaos lab**. Each acts on *the replica that +receives the request*, and the activity log shows which one. + +| Button | What happens | Where to look in Miabi | +|--------|--------------|------------------------| +| 💥 Crash it | process exits 1; the restart policy restarts it; the page reports when it's back | app **Events**, **Notifications** (crash-loop if repeated) | +| 🩺 Unhealthy | `/healthz` → 503 on that replica | `app_unhealthy` alert; a deploy made now fails the health gate | +| 🔥 CPU burn | all cores busy for N s | app metrics | +| 🧠 Hold memory | allocates N MiB. Set a **Memory limit** of 256 MB first, then hold 400 MiB → OOM kill → restart | `app_oom` alert | +| 📜 Log burst | N error lines | **Logs** tab (Live) | +| Probe `/api/debug/slow` | 400 ms latency | **Analytics → Performance** | +| Probe `/api/debug/error` | 5xx responses | **Analytics → HTTP Traffic** | + +The same endpoints work from a shell: + +```bash +T=; D=https:// +curl -XPOST -H "Authorization: Bearer $T" $D/api/debug/crash +curl -XPOST -H "Authorization: Bearer $T" -H 'Content-Type: application/json' \ + -d '{"healthy":false}' $D/api/debug/health +curl -XPOST -H "Authorization: Bearer $T" "$D/api/debug/memory?mb=400&seconds=30" +``` + +Remember to click **Healthy** afterwards, or redeploy. + +## 7. Canary v1 → v2 + +1. In a terminal: `scripts/hammer.sh https://`. It prints a live count and, on Ctrl-C, a tally of `status version host`. +2. Set **deploy strategy** to *canary* (initial weight 10%), and deploy image `2.0.0`. +3. On the wall, reload a few times: ~1 in 10 loads shows the **v2 badge** in a different colour. In the admin **Traffic probe**, the bars split by version. +4. **Promote now.** The probe goes 100% v2, and `hammer.sh` should show `0 failed`. + +> The v2 build is the same code with a different `VERSION`. Nothing needs to +> differ for the demo to work. + +## 8. Scheduled cleanup + +The binary has a `cleanup` subcommand that permanently removes deleted +signatures and optionally expires old ones: + +```bash +guestbook cleanup --purge-after 168h # default: purge deletes older than 7 days +guestbook cleanup --max-age 720h --dry-run # preview expiring >30-day-old, unpinned entries +``` + +1. **Jobs** → *New cronjob*, image blank (uses the app's image), command `/app/guestbook cleanup --purge-after 1h`, schedule `*/15 * * * *`. +2. Delete a signature in **Moderation**, then **Run now**. The job log shows `cleanup finished purged=… remaining=…`. + +`guestbook migrate` works the same way as a one-off **Job**, for teams that +want migrations as an explicit step rather than on boot. + +--- + +## Reset between sessions + +- Admin → Settings: clear the banner, un-pause. +- Chaos lab: **Healthy**. +- To wipe the wall: detach/recreate the database, or `DELETE FROM entries;` and restart (the seed runs when the table is empty). diff --git a/README.md b/README.md index 2748d55..52837b6 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,7 @@ # Miabi Guestbook — example app +[![Running on Miabi](https://miabi.io/badges/running-on-miabi-dark.svg)](https://miabi.io?ref=badge) + A tiny, self-contained web app used as a **reference for deploying to [Miabi](https://github.com/miabi-io/miabi)**. It is a classic guestbook: visitors leave a name and a message, and the wall shows every signature. @@ -28,6 +30,19 @@ miniature: - **`AutoMigrate` on startup** with retry (tolerates the DB coming up a beat later) - **First-run seeder** — populates the wall when the table is empty (`SEED=false` to skip) - **`/healthz` readiness probe** that returns `503` until the database is reachable +- **Scales horizontally** — attach a Redis and live updates + presence fan out + across every replica via pub/sub; without it, the in-process broker makes the + classic "works on one replica" bug easy to demonstrate +- **Admin console** at `/admin`, guarded by a single `ADMIN_TOKEN` secret: + overview (stats, replicas, runtime), moderation (pin / hide / delete, signer + IP, CSV export), wall settings (pause signing, announcement banner) — all pushed live +- **Chaos lab** (`DEBUG_ENDPOINTS=true`): crash, fail health checks, burn CPU, + hold memory, burst logs, inject latency/5xx, plus an in-browser traffic probe + that shows load-balancing and canary splits +- **Scheduled-job friendly CLI** — `guestbook cleanup` (purge/expire) and + `guestbook migrate` subcommands for Miabi cron jobs and one-off jobs + +**Presenting it?** [`DEMO.md`](DEMO.md) is a step-by-step training runbook. - **Single static binary** — the whole UI is embedded with `go:embed`, so the image is one distroless layer with **no Node build step** @@ -45,14 +60,44 @@ and keeps the container tiny. | Method | Path | Description | |----------|----------------------|------------------------------------| -| `GET` | `/healthz` | Liveness/readiness (checks the DB), reports `version` | -| `GET` | `/api/info` | App name, serving `version`, `host`, connected-client count | +| `GET` | `/healthz` | Liveness/readiness (checks the DB), reports `version`, `host`, `redis` | +| `GET` | `/api/info` | App name, serving `version`, `host`, online count, `broker` mode, `replicas` | | `GET` | `/api/time` | Current server time + `version` + `host` (v2 — canary probe) | -| `GET` | `/api/entries` | List entries + total; paginated via `?limit=&offset=` | -| `POST` | `/api/entries` | Create `{ "name", "message" }` (broadcast live) | -| `DELETE` | `/api/entries/{id}` | Delete an entry (broadcast live) | -| `GET` | `/api/stream` | **SSE** stream: `welcome`, `created`, `deleted`, `presence`, `tick` events | -| `GET` | `/` · `/all` | The web UI (home · all-signatures page) | +| `GET` | `/api/settings` | Public wall settings: `signing_paused`, `banner` | +| `GET` | `/api/entries` | Visible entries (pinned first) + total; paginated via `?limit=&offset=` | +| `POST` | `/api/entries` | Create `{ "name", "message" }` (broadcast live; `503` while paused) | +| `GET` | `/api/stream` | **SSE** stream: `welcome`, `created`, `updated`, `deleted`, `settings`, `presence`, `tick` events | +| `GET` | `/` · `/all` · `/admin` | The web UI (home · all signatures · admin console) | + +### Admin API + +Authenticated with the session cookie from `POST /api/admin/login {"token"}` +or an `Authorization: Bearer ` header. Disabled (`404`) when +`ADMIN_TOKEN` is unset. + +| Method | Path | Description | +|----------|-------------------------------|-------------| +| `GET` | `/api/admin/overview` | Stats (14-day histogram), replicas, runtime of the answering replica | +| `GET` | `/api/admin/entries` | All entries; `?status=visible\|hidden\|pinned&q=&limit=&offset=` | +| `PATCH` | `/api/admin/entries/{id}` | `{ "pinned"?: bool, "hidden"?: bool }` | +| `DELETE` | `/api/admin/entries/{id}` | Soft-delete an entry | +| `GET`·`PUT` | `/api/admin/settings` | `{ "signing_paused": bool, "banner": string }` | +| `GET` | `/api/admin/export.csv` | Every entry as CSV | + +### Debug API (chaos) + +Requires admin auth **and** `DEBUG_ENDPOINTS=true`. Each call affects only the +replica that receives it and echoes its `host`. + +| Method | Path | Effect | +|--------|------|--------| +| `POST` | `/api/debug/crash` | Exit with status 1 | +| `POST` | `/api/debug/health` | `{ "healthy": false }` makes `/healthz` return 503 | +| `GET` | `/api/debug/slow?ms=2000` | Sleep before answering (max 30 s) | +| `GET` | `/api/debug/error?status=500` | Answer with a 4xx/5xx | +| `POST` | `/api/debug/cpu?seconds=30` | Saturate all cores (max 120 s) | +| `POST` | `/api/debug/memory?mb=256&seconds=30` | Hold resident memory (max 2048 MiB) | +| `POST` | `/api/debug/logs?count=20&level=error` | Write a burst of log lines | ### Live updates (SSE) @@ -64,6 +109,11 @@ the live server time (with `version` + `host`). The UI uses these to update the wall in real time, show a live **“N online”** indicator, and drive the **server-time card** — all without polling. +With `REDIS_URL` set, `created`/`updated`/`deleted`/`settings` events go through +a Redis pub/sub channel so every replica delivers them, and each replica +advertises its client count under a short-TTL key so presence is cluster-wide. +`tick` events stay per-replica on purpose: the clock shows who served you. + ## Configuration | Variable | Default | Purpose | @@ -81,6 +131,10 @@ wall in real time, show a live **“N online”** indicator, and drive the | `DB_PASSWORD` | `postgres` | Database password | | `DB_NAME` | `guestbook` | Database name | | `DB_SSL_MODE` | `disable` | Postgres `sslmode` | +| `REDIS_URL` | — | Redis URL (`redis://:pass@host:6379`). Enables cross-replica live updates. Falls back to `REDIS_DATABASE_URL`, which is what Miabi injects when you attach a managed Redis with the `REDIS` prefix. | +| `ADMIN_TOKEN` | — | Enables `/admin`. Store it as a **secret**. | +| `TRUSTED_PROXIES` | — | Comma-separated CIDRs whose `X-Forwarded-For` is trusted when recording a signer's IP (e.g. `10.0.0.0/8`). Unset: the header is taken as-is — fine behind the Miabi gateway, spoofable if the app is exposed directly. | +| `DEBUG_ENDPOINTS` | `false` | `true` enables the `/api/debug/*` chaos endpoints (still admin-only). | > **On Miabi:** attach a managed database to the app and Miabi injects these > variables automatically — `DATABASE_URL` for the full DSN, plus the discrete @@ -102,6 +156,8 @@ SQLite file, so this just works: go run . # runs the default "server" command go run . server -p 9000 # okapicli flags: choose the port go run . --help # list commands and flags +go run . migrate # apply migrations and exit +go run . cleanup --purge-after 168h --max-age 720h --dry-run ``` To point it at your own Postgres, set `DATABASE_URL` (or the `DB_*` vars) first: @@ -159,5 +215,7 @@ docker build --build-arg VERSION=2.0.0 -t miabi/guestbook:2.0.0 . 3. **Attach the database** — Miabi injects the connection as `DATABASE_URL`. 4. **Set the health check** to `GET /healthz` and the port to `8080`. 5. **Connect a domain** — Miabi issues automatic SSL and routes traffic. +6. **Optional:** add `ADMIN_TOKEN` as a secret, scale replicas, and attach a + managed Redis with the `REDIS` prefix. See [`DEMO.md`](DEMO.md). The app migrates its schema on first boot, so there is no separate release step. diff --git a/admin.go b/admin.go new file mode 100644 index 0000000..9d30dc4 --- /dev/null +++ b/admin.go @@ -0,0 +1,210 @@ +package main + +import ( + "encoding/csv" + "errors" + "net/http" + "runtime" + "strconv" + "strings" + "time" + + "github.com/jkaninda/okapi" + "gorm.io/gorm" +) + +const maxBannerLen = 200 + +type loginRequest struct { + Token string `json:"token"` +} + +// adminEntry exposes the fields public responses omit. +type adminEntry struct { + Entry + IP string `json:"ip,omitempty"` +} + +func toAdmin(e Entry) adminEntry { return adminEntry{Entry: e, IP: e.IP} } + +type flagsRequest struct { + Pinned *bool `json:"pinned"` + Hidden *bool `json:"hidden"` +} + +// AdminSession reports whether the console is enabled and the caller signed in. +func (h *Handler) AdminSession(c *okapi.Context) error { + return c.OK(okapi.M{ + "enabled": h.auth.Enabled(), + "authenticated": h.auth.Authenticated(c), + }) +} + +// AdminLogin exchanges ADMIN_TOKEN for a session cookie. +func (h *Handler) AdminLogin(c *okapi.Context) error { + if !h.auth.Enabled() { + return c.JSON(http.StatusNotFound, okapi.M{"error": "admin console is disabled (set ADMIN_TOKEN)"}) + } + var req loginRequest + if err := c.Bind(&req); err != nil { + return c.JSON(http.StatusBadRequest, okapi.M{"error": "invalid JSON body"}) + } + if !h.auth.CheckToken(strings.TrimSpace(req.Token)) { + // Slows down guessing without needing shared rate-limit state. + time.Sleep(700 * time.Millisecond) + return c.JSON(http.StatusUnauthorized, okapi.M{"error": "invalid admin token"}) + } + h.auth.SetSession(c, h.auth.Issue(time.Now())) + return c.OK(okapi.M{"authenticated": true}) +} + +// AdminLogout clears the session cookie. +func (h *Handler) AdminLogout(c *okapi.Context) error { + h.auth.SetSession(c, "") + return c.NoContent() +} + +// AdminOverview returns wall statistics plus runtime details of the replica +// that answered, which is what makes scaling visible in the console. +func (h *Handler) AdminOverview(c *okapi.Context) error { + ctx := c.Request().Context() + stats, err := h.store.Stats(ctx, 14) + if err != nil { + return c.JSON(http.StatusInternalServerError, okapi.M{"error": "could not compute stats"}) + } + settings, _ := h.store.Settings(ctx) + + var mem runtime.MemStats + runtime.ReadMemStats(&mem) + + return c.OK(okapi.M{ + "stats": stats, + "settings": settings, + "online": h.broker.Online(), + "replicas": h.broker.Replicas(), + "debug": h.debug, + "runtime": okapi.M{ + "app": h.appName, + "version": h.version, + "host": h.host, + "db": dbDisplayName(h.dbDriver), + "broker": h.broker.Mode(), + "redis": h.broker.RedisStatus(ctx), + "healthy": !h.unhealthy.Load(), + "started_at": h.started.UTC().Format(time.RFC3339), + "uptime_s": int64(time.Since(h.started).Seconds()), + "go": runtime.Version(), + "cpus": runtime.NumCPU(), + "goroutines": runtime.NumGoroutine(), + "heap_mb": float64(mem.HeapAlloc) / (1 << 20), + "sys_mb": float64(mem.Sys) / (1 << 20), + }, + }) +} + +// AdminListEntries lists every entry, hidden ones included, with filters. +func (h *Handler) AdminListEntries(c *okapi.Context) error { + limit := atoiDefault(c.Query("limit"), 25) + if limit <= 0 || limit > 200 { + limit = 25 + } + offset := max(atoiDefault(c.Query("offset"), 0), 0) + + entries, total, err := h.store.AdminList(c.Request().Context(), EntryFilter{ + Status: c.Query("status"), + Query: c.Query("q"), + Limit: limit, + Offset: offset, + }) + if err != nil { + return c.JSON(http.StatusInternalServerError, okapi.M{"error": "could not load entries"}) + } + out := make([]adminEntry, len(entries)) + for i, e := range entries { + out[i] = toAdmin(e) + } + return c.OK(okapi.M{"entries": out, "total": total, "limit": limit, "offset": offset}) +} + +// AdminUpdateEntry pins/unpins or hides/unhides an entry and broadcasts it. +func (h *Handler) AdminUpdateEntry(c *okapi.Context) error { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + return c.JSON(http.StatusBadRequest, okapi.M{"error": "invalid id"}) + } + var req flagsRequest + if err := c.Bind(&req); err != nil { + return c.JSON(http.StatusBadRequest, okapi.M{"error": "invalid JSON body"}) + } + entry, err := h.store.SetFlags(c.Request().Context(), id, req.Pinned, req.Hidden) + if errors.Is(err, gorm.ErrRecordNotFound) { + return c.JSON(http.StatusNotFound, okapi.M{"error": "entry not found"}) + } + if err != nil { + return c.JSON(http.StatusInternalServerError, okapi.M{"error": "could not update entry"}) + } + h.broker.Publish(Event{Type: "updated", Entry: &entry}) + return c.OK(toAdmin(entry)) +} + +// AdminSettings returns the shared wall settings. +func (h *Handler) AdminSettings(c *okapi.Context) error { + st, err := h.store.Settings(c.Request().Context()) + if err != nil { + return c.JSON(http.StatusInternalServerError, okapi.M{"error": "could not load settings"}) + } + return c.OK(st) +} + +// AdminSaveSettings persists the wall settings and broadcasts them so every +// open tab on every replica reacts immediately. +func (h *Handler) AdminSaveSettings(c *okapi.Context) error { + var st Settings + if err := c.Bind(&st); err != nil { + return c.JSON(http.StatusBadRequest, okapi.M{"error": "invalid JSON body"}) + } + st.Banner = strings.TrimSpace(st.Banner) + if len(st.Banner) > maxBannerLen { + return c.JSON(http.StatusBadRequest, okapi.M{"error": "banner is too long (max 200 characters)"}) + } + if err := h.store.SaveSettings(c.Request().Context(), st); err != nil { + return c.JSON(http.StatusInternalServerError, okapi.M{"error": "could not save settings"}) + } + h.broker.Publish(Event{Type: "settings", Settings: &st}) + return c.OK(st) +} + +// AdminExport streams every entry as CSV. +func (h *Handler) AdminExport(c *okapi.Context) error { + entries, err := h.store.All(c.Request().Context()) + if err != nil { + return c.JSON(http.StatusInternalServerError, okapi.M{"error": "could not export entries"}) + } + w := c.ResponseWriter() + w.Header().Set("Content-Type", "text/csv; charset=utf-8") + w.Header().Set("Content-Disposition", `attachment; filename="guestbook-`+time.Now().UTC().Format("20060102-150405")+`.csv"`) + cw := csv.NewWriter(w) + _ = cw.Write([]string{"id", "name", "message", "pinned", "hidden", "replica", "ip", "created_at"}) + for _, e := range entries { + _ = cw.Write([]string{ + strconv.FormatUint(uint64(e.ID), 10), + csvSafe(e.Name), + csvSafe(e.Message), + strconv.FormatBool(e.Pinned), + strconv.FormatBool(e.Hidden), + e.Replica, + e.IP, + e.CreatedAt.UTC().Format(time.RFC3339), + }) + } + cw.Flush() + return cw.Error() +} + +// csvSafe neutralises values a spreadsheet would evaluate as a formula. +func csvSafe(s string) string { + if s != "" && strings.ContainsRune("=+-@\t\r", rune(s[0])) { + return "'" + s + } + return s +} diff --git a/auth.go b/auth.go new file mode 100644 index 0000000..103861f --- /dev/null +++ b/auth.go @@ -0,0 +1,107 @@ +package main + +import ( + "crypto/hmac" + "crypto/sha256" + "crypto/subtle" + "encoding/hex" + "net/http" + "strconv" + "strings" + "time" + + "github.com/jkaninda/okapi" +) + +const ( + sessionCookie = "gb_admin" + sessionTTL = 12 * time.Hour +) + +// AdminAuth guards the admin console with a single shared ADMIN_TOKEN. +// +// Sessions are stateless HMAC-signed cookies keyed on the token, so any +// replica can verify a session issued by another, and rotating ADMIN_TOKEN +// logs everyone out. +type AdminAuth struct { + token string + key []byte +} + +// NewAdminAuth returns an AdminAuth; an empty token disables the console. +func NewAdminAuth(token string) *AdminAuth { + sum := sha256.Sum256([]byte("guestbook-admin-session:" + token)) + return &AdminAuth{token: token, key: sum[:]} +} + +// Enabled reports whether an admin token is configured. +func (a *AdminAuth) Enabled() bool { return a.token != "" } + +// CheckToken compares a candidate against ADMIN_TOKEN in constant time. +func (a *AdminAuth) CheckToken(candidate string) bool { + return a.Enabled() && subtle.ConstantTimeCompare([]byte(candidate), []byte(a.token)) == 1 +} + +// Issue returns a signed session value expiring after sessionTTL. +func (a *AdminAuth) Issue(now time.Time) string { + exp := strconv.FormatInt(now.Add(sessionTTL).Unix(), 10) + return exp + "." + a.sign(exp) +} + +// Verify reports whether a session value is authentic and unexpired. +func (a *AdminAuth) Verify(value string, now time.Time) bool { + if !a.Enabled() { + return false + } + exp, sig, ok := strings.Cut(value, ".") + if !ok || !hmac.Equal([]byte(sig), []byte(a.sign(exp))) { + return false + } + unix, err := strconv.ParseInt(exp, 10, 64) + return err == nil && now.Unix() < unix +} + +func (a *AdminAuth) sign(payload string) string { + m := hmac.New(sha256.New, a.key) + m.Write([]byte(payload)) + return hex.EncodeToString(m.Sum(nil)) +} + +// Authenticated accepts either a session cookie (the console) or an +// "Authorization: Bearer " header (curl during a demo). +func (a *AdminAuth) Authenticated(c *okapi.Context) bool { + if bearer, ok := strings.CutPrefix(c.Header("Authorization"), "Bearer "); ok { + return a.CheckToken(strings.TrimSpace(bearer)) + } + v, err := c.Cookie(sessionCookie) + return err == nil && a.Verify(v, time.Now()) +} + +// Require is middleware that rejects unauthenticated requests. +func (a *AdminAuth) Require(c *okapi.Context) error { + if !a.Enabled() { + return c.JSON(http.StatusNotFound, okapi.M{"error": "admin console is disabled (set ADMIN_TOKEN)"}) + } + if !a.Authenticated(c) { + return c.JSON(http.StatusUnauthorized, okapi.M{"error": "authentication required"}) + } + return c.Next() +} + +// SetSession writes (or, with an empty value, clears) the session cookie. +func (a *AdminAuth) SetSession(c *okapi.Context, value string) { + maxAge := int(sessionTTL.Seconds()) + if value == "" { + maxAge = -1 + } + r := c.Request() + http.SetCookie(c.ResponseWriter(), &http.Cookie{ + Name: sessionCookie, + Value: value, + Path: "/", + MaxAge: maxAge, + HttpOnly: true, + Secure: r.TLS != nil || r.Header.Get("X-Forwarded-Proto") == "https", + SameSite: http.SameSiteStrictMode, + }) +} diff --git a/auth_test.go b/auth_test.go new file mode 100644 index 0000000..d8d90f7 --- /dev/null +++ b/auth_test.go @@ -0,0 +1,32 @@ +package main + +import ( + "testing" + "time" +) + +func TestAdminSession(t *testing.T) { + a := NewAdminAuth("s3cret") + now := time.Now() + v := a.Issue(now) + + if !a.Verify(v, now) { + t.Fatal("fresh session should verify") + } + if a.Verify(v, now.Add(sessionTTL+time.Second)) { + t.Fatal("expired session should not verify") + } + if NewAdminAuth("other").Verify(v, now) { + t.Fatal("session signed with another token should not verify") + } + if a.Verify("9999999999."+v[len(v)-64:], now) { + t.Fatal("tampered expiry should not verify") + } +} + +func TestAdminDisabled(t *testing.T) { + a := NewAdminAuth("") + if a.Enabled() || a.CheckToken("") || a.Verify(a.Issue(time.Now()), time.Now()) { + t.Fatal("empty token must disable all admin access") + } +} diff --git a/broker.go b/broker.go index 4fa1ffd..a10285f 100644 --- a/broker.go +++ b/broker.go @@ -1,53 +1,158 @@ package main -import "sync" +import ( + "context" + "encoding/json" + "sort" + "sync" + "time" + + "github.com/jkaninda/logger" + "github.com/redis/go-redis/v9" +) // Event is a server-to-client message pushed over SSE. // // Type selects the shape of the payload: // - "created" → Entry is set (a new signature) +// - "updated" → Entry is set (pinned/hidden by a moderator) // - "deleted" → ID is set (a removed signature) -// - "presence" → Online is set (connected-client count changed) +// - "settings" → Settings is set (signing paused, banner changed) +// - "presence" → Online/Replicas are set (connected clients changed) // - "tick" → Time/Version/Host are set (live server clock; new in v2) type Event struct { - Type string `json:"type"` - Entry *Entry `json:"entry,omitempty"` - ID uint `json:"id,omitempty"` - Online int `json:"online,omitempty"` - Time string `json:"time,omitempty"` - Version string `json:"version,omitempty"` - Host string `json:"host,omitempty"` + Type string `json:"type"` + Entry *Entry `json:"entry,omitempty"` + ID uint `json:"id,omitempty"` + Settings *Settings `json:"settings,omitempty"` + Online int `json:"online,omitempty"` + Replicas int `json:"replicas,omitempty"` + Time string `json:"time,omitempty"` + Version string `json:"version,omitempty"` + Host string `json:"host,omitempty"` } -// Broker is a tiny in-process pub/sub hub. Each connected SSE client holds one -// subscription; handlers Publish entry events, and the broker itself emits a -// "presence" event whenever the number of connected clients changes. +// Replica describes one running instance of the app and its SSE clients. +type Replica struct { + Host string `json:"host"` + Version string `json:"version"` + Online int `json:"online"` + Seen time.Time `json:"seen"` +} + +const ( + eventsChannel = "guestbook:events" + replicaKeyPrefix = "guestbook:replica:" + replicaTTL = 15 * time.Second + heartbeatEvery = 5 * time.Second + syncEvent = "sync" +) + +// Broker fans events out to the SSE clients connected to this replica. +// +// Without Redis it is purely in-process, so a signature posted on replica A +// never reaches viewers on replica B. With Redis, published events go through +// a pub/sub channel that every replica listens to, and each replica advertises +// its client count under a TTL'd key so presence is cluster-wide. type Broker struct { mu sync.RWMutex subs map[chan Event]struct{} + + host, version string + rdb *redis.Client + dirty chan struct{} + + cmu sync.RWMutex + replicas []Replica + online int } -// NewBroker creates an empty Broker. -func NewBroker() *Broker { - return &Broker{subs: make(map[chan Event]struct{})} +// NewBroker creates an in-process Broker. Call UseRedis to make it cluster-wide. +func NewBroker(host, version string) *Broker { + return &Broker{ + subs: make(map[chan Event]struct{}), + host: host, + version: version, + dirty: make(chan struct{}, 1), + } } -// Subscribe registers a new client and returns its event channel. It also -// broadcasts the updated presence count to everyone. +// UseRedis switches the broker to Redis pub/sub. The client reconnects on its +// own, so an unreachable Redis at startup is not fatal. +func (b *Broker) UseRedis(url string) error { + opts, err := redis.ParseURL(url) + if err != nil { + return err + } + b.rdb = redis.NewClient(opts) + return nil +} + +// Mode reports "redis" when events are shared across replicas, else "local". +func (b *Broker) Mode() string { + if b.rdb != nil { + return "redis" + } + return "local" +} + +// RedisStatus reports "disabled", "up" or "down". +func (b *Broker) RedisStatus(ctx context.Context) string { + if b.rdb == nil { + return "disabled" + } + ctx, cancel := context.WithTimeout(ctx, time.Second) + defer cancel() + if err := b.rdb.Ping(ctx).Err(); err != nil { + return "down" + } + return "up" +} + +// Run drives the Redis subscription and presence heartbeat until ctx is done. +// It is a no-op in local mode. +func (b *Broker) Run(ctx context.Context) { + if b.rdb == nil { + return + } + go b.listen(ctx) + + t := time.NewTicker(heartbeatEvery) + defer t.Stop() + b.announce(ctx, true) + for { + select { + case <-ctx.Done(): + return + case <-b.dirty: + b.announce(ctx, true) + case <-t.C: + b.announce(ctx, false) + } + } +} + +// Close withdraws this replica from the presence set. +func (b *Broker) Close(ctx context.Context) { + if b.rdb == nil { + return + } + _ = b.rdb.Del(ctx, replicaKeyPrefix+b.host).Err() + _ = b.publishRaw(ctx, Event{Type: syncEvent}) + _ = b.rdb.Close() +} + +// Subscribe registers a new client and returns its event channel. func (b *Broker) Subscribe() chan Event { ch := make(chan Event, 16) - b.mu.Lock() b.subs[ch] = struct{}{} - n := len(b.subs) b.mu.Unlock() - - b.emit(Event{Type: "presence", Online: n}) + b.presenceChanged() return ch } -// Unsubscribe removes a client, closes its channel, and broadcasts the updated -// presence count. +// Unsubscribe removes a client and closes its channel. func (b *Broker) Unsubscribe(ch chan Event) { b.mu.Lock() if _, ok := b.subs[ch]; !ok { @@ -56,25 +161,158 @@ func (b *Broker) Unsubscribe(ch chan Event) { } delete(b.subs, ch) close(ch) - n := len(b.subs) b.mu.Unlock() - - b.emit(Event{Type: "presence", Online: n}) + b.presenceChanged() } -// Publish broadcasts an event to every connected client. -func (b *Broker) Publish(ev Event) { b.emit(ev) } +// Publish broadcasts an event to every client on every replica. +func (b *Broker) Publish(ev Event) { + if b.rdb == nil { + b.emit(ev) + return + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := b.publishRaw(ctx, ev); err != nil { + logger.Warn("redis publish failed, delivering locally only", "type", ev.Type, "error", err) + b.emit(ev) + } +} -// Online returns the current number of connected clients. +// PublishLocal broadcasts an event to this replica's clients only. +func (b *Broker) PublishLocal(ev Event) { b.emit(ev) } + +// Online returns the number of connected clients across all replicas. func (b *Broker) Online() int { + if b.rdb == nil { + return b.localCount() + } + b.cmu.RLock() + defer b.cmu.RUnlock() + return b.online +} + +// Replicas lists the running replicas and their client counts. +func (b *Broker) Replicas() []Replica { + if b.rdb == nil { + return []Replica{b.self()} + } + b.cmu.RLock() + defer b.cmu.RUnlock() + return append([]Replica(nil), b.replicas...) +} + +func (b *Broker) self() Replica { + return Replica{Host: b.host, Version: b.version, Online: b.localCount(), Seen: time.Now().UTC()} +} + +func (b *Broker) localCount() int { b.mu.RLock() defer b.mu.RUnlock() return len(b.subs) } -// emit fans an event out to all subscribers. Sends are non-blocking: a slow -// client that has filled its buffer simply misses the event rather than -// stalling the broker. +func (b *Broker) presenceChanged() { + if b.rdb == nil { + b.emit(Event{Type: "presence", Online: b.localCount(), Replicas: 1}) + return + } + select { + case b.dirty <- struct{}{}: + default: + } +} + +func (b *Broker) publishRaw(ctx context.Context, ev Event) error { + payload, err := json.Marshal(ev) + if err != nil { + return err + } + return b.rdb.Publish(ctx, eventsChannel, payload).Err() +} + +func (b *Broker) listen(ctx context.Context) { + ps := b.rdb.Subscribe(ctx, eventsChannel) + defer func() { _ = ps.Close() }() + ch := ps.Channel() + for { + select { + case <-ctx.Done(): + return + case msg, ok := <-ch: + if !ok { + return + } + var ev Event + if err := json.Unmarshal([]byte(msg.Payload), &ev); err != nil { + continue + } + if ev.Type == syncEvent { + b.refresh(ctx) + continue + } + b.emit(ev) + } + } +} + +// announce writes this replica's heartbeat and, when notify is set, tells the +// other replicas to recount presence. +func (b *Broker) announce(ctx context.Context, notify bool) { + payload, _ := json.Marshal(b.self()) + if err := b.rdb.Set(ctx, replicaKeyPrefix+b.host, payload, replicaTTL).Err(); err != nil { + logger.Warn("redis heartbeat failed", "error", err) + return + } + if notify { + _ = b.publishRaw(ctx, Event{Type: syncEvent}) + } + b.refresh(ctx) +} + +// refresh recomputes cluster presence from the replica keys and pushes a +// presence event to local clients when it changed. +func (b *Broker) refresh(ctx context.Context) { + var keys []string + iter := b.rdb.Scan(ctx, 0, replicaKeyPrefix+"*", 100).Iterator() + for iter.Next(ctx) { + keys = append(keys, iter.Val()) + } + if iter.Err() != nil || len(keys) == 0 { + return + } + vals, err := b.rdb.MGet(ctx, keys...).Result() + if err != nil { + return + } + replicas := make([]Replica, 0, len(vals)) + total := 0 + for _, v := range vals { + s, ok := v.(string) + if !ok { + continue + } + var r Replica + if json.Unmarshal([]byte(s), &r) == nil { + replicas = append(replicas, r) + total += r.Online + } + } + sort.Slice(replicas, func(i, j int) bool { return replicas[i].Host < replicas[j].Host }) + + b.cmu.Lock() + changed := total != b.online || len(replicas) != len(b.replicas) + b.online, b.replicas = total, replicas + b.cmu.Unlock() + + if changed { + b.emit(Event{Type: "presence", Online: total, Replicas: len(replicas)}) + } +} + +// emit fans an event out to local subscribers. Sends are non-blocking: a slow +// client that has filled its buffer misses the event rather than stalling +// the broker. func (b *Broker) emit(ev Event) { b.mu.RLock() defer b.mu.RUnlock() diff --git a/broker_test.go b/broker_test.go new file mode 100644 index 0000000..103dc11 --- /dev/null +++ b/broker_test.go @@ -0,0 +1,33 @@ +package main + +import ( + "testing" + "time" +) + +func TestLocalBroker(t *testing.T) { + b := NewBroker("host-a", "1.0.0") + ch := b.Subscribe() + defer b.Unsubscribe(ch) + + if ev := <-ch; ev.Type != "presence" || ev.Online != 1 { + t.Fatalf("want presence=1 on subscribe, got %+v", ev) + } + + b.Publish(Event{Type: "deleted", ID: 7}) + select { + case ev := <-ch: + if ev.Type != "deleted" || ev.ID != 7 { + t.Fatalf("unexpected event %+v", ev) + } + case <-time.After(time.Second): + t.Fatal("published event not delivered") + } + + if r := b.Replicas(); len(r) != 1 || r[0].Host != "host-a" || r[0].Online != 1 { + t.Fatalf("unexpected replicas %+v", r) + } + if b.Mode() != "local" { + t.Fatalf("mode = %s", b.Mode()) + } +} diff --git a/compose.yaml b/compose.yaml index 54d33b0..c0b3d77 100644 --- a/compose.yaml +++ b/compose.yaml @@ -1,5 +1,5 @@ -# Local development: Postgres + the app. Mirrors what Miabi wires up for you -# (a managed Postgres attached to the app via DATABASE_URL). +# Local development: Postgres + Redis + the app. Mirrors what Miabi wires up +# for you (managed databases attached to the app via env vars). services: db: image: postgres:17-alpine @@ -17,14 +17,22 @@ services: volumes: - db-data:/var/lib/postgresql/data + redis: + image: redis:7-alpine + app: build: . environment: PORT: 8080 DATABASE_URL: postgres://postgres:postgres@db:5432/guestbook?sslmode=disable + REDIS_URL: redis://redis:6379 + ADMIN_TOKEN: ${ADMIN_TOKEN:-change-me} + DEBUG_ENDPOINTS: "true" depends_on: db: condition: service_healthy + redis: + condition: service_started ports: - "8080:8080" diff --git a/config.go b/config.go index ca17579..0507e8b 100644 --- a/config.go +++ b/config.go @@ -18,6 +18,15 @@ type Config struct { AppName string Version string Seed bool + + // RedisURL enables the cross-replica broker; empty keeps it in-process. + RedisURL string + // AdminToken enables the /admin console; empty disables it. + AdminToken string + // DebugEndpoints exposes the /api/debug failure-injection endpoints. + DebugEndpoints bool + // TrustedProxies limits whose X-Forwarded-For is believed (CIDRs). + TrustedProxies []string } // LoadConfig builds the Config from the environment. @@ -30,6 +39,11 @@ func LoadConfig() (Config, error) { Seed: goutils.Env("SEED", "true") == "true", DBDriver: driver, DatabaseDSN: dsn, + + RedisURL: goutils.Env("REDIS_URL", goutils.Env("REDIS_DATABASE_URL", "")), + AdminToken: goutils.Env("ADMIN_TOKEN", ""), + DebugEndpoints: goutils.Env("DEBUG_ENDPOINTS", "false") == "true", + TrustedProxies: splitList(goutils.Env("TRUSTED_PROXIES", "")), }, nil } @@ -80,3 +94,13 @@ func postgresDSN() string { goutils.Env("DB_SSL_MODE", "disable"), ) } + +func splitList(s string) []string { + var out []string + for _, p := range strings.Split(s, ",") { + if p = strings.TrimSpace(p); p != "" { + out = append(out, p) + } + } + return out +} diff --git a/debug.go b/debug.go new file mode 100644 index 0000000..8e0568d --- /dev/null +++ b/debug.go @@ -0,0 +1,159 @@ +package main + +import ( + "net/http" + "os" + "runtime" + "strconv" + "sync/atomic" + "time" + + "github.com/jkaninda/logger" + "github.com/jkaninda/okapi" +) + +// Failure injection so restarts, health gating, analytics and alerts have +// something to react to during a demo. See DEMO.md. + +const ( + maxSlow = 30 * time.Second + maxBurn = 120 * time.Second + maxBallastMB = 2048 + maxLogBurst = 1000 + defaultBurnS = 30 + defaultMemMB = 256 +) + +type chaosState struct { + cpuBusy atomic.Bool + ballastMB atomic.Int64 +} + +// RequireDebug rejects requests unless DEBUG_ENDPOINTS=true. +func (h *Handler) RequireDebug(c *okapi.Context) error { + if !h.debug { + return c.JSON(http.StatusNotFound, okapi.M{"error": "debug endpoints are disabled (set DEBUG_ENDPOINTS=true)"}) + } + return c.Next() +} + +// DebugCrash exits the process with status 1 after answering. +func (h *Handler) DebugCrash(c *okapi.Context) error { + logger.Error("crash requested via /api/debug/crash, exiting", "host", h.host) + go func() { + time.Sleep(300 * time.Millisecond) + os.Exit(1) + }() + return c.JSON(http.StatusAccepted, okapi.M{"status": "crashing", "host": h.host}) +} + +// DebugSlow sleeps for ?ms= milliseconds (default 2000) before answering. +func (h *Handler) DebugSlow(c *okapi.Context) error { + d := min(time.Duration(atoiDefault(c.Query("ms"), 2000))*time.Millisecond, maxSlow) + select { + case <-time.After(d): + case <-c.Request().Context().Done(): + return nil + } + return c.OK(okapi.M{"slept_ms": d.Milliseconds(), "host": h.host}) +} + +// DebugError answers with ?status= (default 500). +func (h *Handler) DebugError(c *okapi.Context) error { + status := atoiDefault(c.Query("status"), http.StatusInternalServerError) + if status < 400 || status > 599 { + status = http.StatusInternalServerError + } + logger.Error("injected error via /api/debug/error", "status", status, "host", h.host) + return c.JSON(status, okapi.M{"error": "injected failure", "status": status, "host": h.host}) +} + +// DebugHealth forces /healthz on this replica to fail (or recover). +func (h *Handler) DebugHealth(c *okapi.Context) error { + var req struct { + Healthy bool `json:"healthy"` + } + if err := c.Bind(&req); err != nil { + return c.JSON(http.StatusBadRequest, okapi.M{"error": "invalid JSON body"}) + } + h.unhealthy.Store(!req.Healthy) + logger.Warn("health overridden via /api/debug/health", "healthy", req.Healthy, "host", h.host) + return c.OK(okapi.M{"healthy": req.Healthy, "host": h.host}) +} + +// DebugCPU saturates every core for ?seconds= (default 30). +func (h *Handler) DebugCPU(c *okapi.Context) error { + d := min(time.Duration(atoiDefault(c.Query("seconds"), defaultBurnS))*time.Second, maxBurn) + if !h.chaos.cpuBusy.CompareAndSwap(false, true) { + return c.JSON(http.StatusConflict, okapi.M{"error": "a CPU burn is already running"}) + } + workers := runtime.GOMAXPROCS(0) + deadline := time.Now().Add(d) + done := make(chan struct{}, workers) + for range workers { + go func() { + x := 0 + for time.Now().Before(deadline) { + for i := range 1_000_000 { + x += i + } + } + _ = x + done <- struct{}{} + }() + } + go func() { + for range workers { + <-done + } + h.chaos.cpuBusy.Store(false) + logger.Info("cpu burn finished", "host", h.host) + }() + logger.Warn("cpu burn started", "seconds", d.Seconds(), "workers", workers, "host", h.host) + return c.JSON(http.StatusAccepted, okapi.M{"seconds": d.Seconds(), "workers": workers, "host": h.host}) +} + +// DebugMemory holds ?mb= MiB (default 256) of touched memory for ?seconds= +// (default 30). Useful to show memory limits and OOM restarts. +func (h *Handler) DebugMemory(c *okapi.Context) error { + mb := min(atoiDefault(c.Query("mb"), defaultMemMB), maxBallastMB) + d := min(time.Duration(atoiDefault(c.Query("seconds"), defaultBurnS))*time.Second, maxBurn) + if mb <= 0 { + return c.JSON(http.StatusBadRequest, okapi.M{"error": "mb must be positive"}) + } + if !h.chaos.ballastMB.CompareAndSwap(0, int64(mb)) { + return c.JSON(http.StatusConflict, okapi.M{"error": "memory is already being held"}) + } + go func() { + ballast := make([]byte, mb<<20) + // Touch every page so the memory is actually resident, not just reserved. + for i := 0; i < len(ballast); i += 4096 { + ballast[i] = 1 + } + time.Sleep(d) + runtime.KeepAlive(ballast) + h.chaos.ballastMB.Store(0) + logger.Info("memory released", "mb", mb, "host", h.host) + }() + logger.Warn("holding memory", "mb", mb, "seconds", d.Seconds(), "host", h.host) + return c.JSON(http.StatusAccepted, okapi.M{"mb": mb, "seconds": d.Seconds(), "host": h.host}) +} + +// DebugLogs writes ?count= (default 20) log lines at ?level= (default error). +func (h *Handler) DebugLogs(c *okapi.Context) error { + count := min(max(atoiDefault(c.Query("count"), 20), 1), maxLogBurst) + level := c.Query("level") + for i := range count { + n := strconv.Itoa(i + 1) + switch level { + case "info": + logger.Info("demo log line", "n", n, "host", h.host) + case "warn": + logger.Warn("demo log line", "n", n, "host", h.host) + default: + level = "error" + logger.Error("demo log line", "n", n, "host", h.host) + } + } + return c.OK(okapi.M{"count": count, "level": level, "host": h.host}) +} diff --git a/go.mod b/go.mod index 1fc7129..b8a82e9 100644 --- a/go.mod +++ b/go.mod @@ -7,11 +7,14 @@ require ( github.com/jkaninda/go-utils v0.1.4 github.com/jkaninda/logger v0.0.5 github.com/jkaninda/okapi v1.0.0 + github.com/redis/go-redis/v9 v9.9.0 gorm.io/driver/postgres v1.6.0 gorm.io/gorm v1.31.1 ) require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/getkin/kin-openapi v0.145.0 // indirect github.com/glebarez/go-sqlite v1.21.2 // indirect diff --git a/go.sum b/go.sum index e1b36e7..ae901d1 100644 --- a/go.sum +++ b/go.sum @@ -1,4 +1,12 @@ +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= github.com/dlclark/regexp2 v1.11.0 h1:G/nrcoOa7ZXlpoa/91N3X7mM3r8eIlMBBJZvsz/mxKI= github.com/dlclark/regexp2 v1.11.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= @@ -54,6 +62,8 @@ github.com/oasdiff/yaml v0.1.1/go.mod h1:EYJNoyktvWMJ0Hmhx+6qTaqMOsalUaRGT8Sj1hN github.com/oasdiff/yaml3 v0.0.14 h1:aLJee3hxBK2H5wdXd9iPcIXb93Nty1Ge0pT171eHtkw= github.com/oasdiff/yaml3 v0.0.14/go.mod h1:csto2xfDjYccdUn/yw/bPjj/cYTdp6HtFA0J4TWG+gg= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.9.0 h1:URbPQ4xVQSQhZ27WMQVmZSo3uT3pL+4IdHVcYq2nVfM= +github.com/redis/go-redis/v9 v9.9.0/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw= github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= diff --git a/handlers.go b/handlers.go index b4265c5..e98f967 100644 --- a/handlers.go +++ b/handlers.go @@ -3,21 +3,31 @@ package main import ( "context" "errors" + "net" "net/http" "strconv" "strings" + "sync/atomic" "time" + "unicode/utf8" "github.com/jkaninda/okapi" ) // Handler holds the dependencies for the HTTP layer. type Handler struct { - store *Store - broker *Broker - appName string - version string - host string + store *Store + broker *Broker + auth *AdminAuth + appName string + version string + host string + dbDriver string + debug bool + started time.Time + + unhealthy atomic.Bool + chaos chaosState } // serverTime is the shape of the live clock payload (v2 feature). It carries @@ -45,6 +55,14 @@ const ( // Health reports liveness/readiness. Returns 503 when the database is // unreachable so Miabi's health checks hold traffic until the DB is up. func (h *Handler) Health(c *okapi.Context) error { + if h.unhealthy.Load() { + return c.JSON(http.StatusServiceUnavailable, okapi.M{ + "status": "unavailable", + "reason": "forced unhealthy via /api/debug/health", + "version": h.version, + "host": h.host, + }) + } if err := h.store.Ping(c.Request().Context()); err != nil { return c.JSON(http.StatusServiceUnavailable, okapi.M{ "status": "unavailable", @@ -52,7 +70,14 @@ func (h *Handler) Health(c *okapi.Context) error { "version": h.version, }) } - return c.OK(okapi.M{"status": "ok", "db": "up", "app": h.appName, "version": h.version}) + return c.OK(okapi.M{ + "status": "ok", + "db": "up", + "redis": h.broker.RedisStatus(c.Request().Context()), + "app": h.appName, + "version": h.version, + "host": h.host, + }) } // Info returns app metadata for the UI: name, the serving version (useful for @@ -60,20 +85,32 @@ func (h *Handler) Health(c *okapi.Context) error { // are currently connected. func (h *Handler) Info(c *okapi.Context) error { return c.OK(okapi.M{ - "app": h.appName, - "version": h.version, - "online": h.broker.Online(), - "host": h.host, + "app": h.appName, + "version": h.version, + "online": h.broker.Online(), + "host": h.host, + "broker": h.broker.Mode(), + "replicas": h.broker.Replicas(), + "admin": h.auth.Enabled(), }) } +// PublicSettings returns the wall state visitors need: paused flag and banner. +func (h *Handler) PublicSettings(c *okapi.Context) error { + st, err := h.store.Settings(c.Request().Context()) + if err != nil { + return c.JSON(http.StatusInternalServerError, okapi.M{"error": "could not load settings"}) + } + return c.OK(st) +} + // Time returns the current server time (plus version + host). Handy for probing // a canary rollout from the shell: `watch curl -s .../api/time`. func (h *Handler) Time(c *okapi.Context) error { return c.OK(h.serverTime()) } -// ListEntries returns guestbook entries, newest first, with simple pagination. +// ListEntries returns visible entries, pinned first, with simple pagination. // Query params: limit (default 100, max 200) and offset (default 0). The home // page requests a small limit; the "All signatures" page pages through them. func (h *Handler) ListEntries(c *okapi.Context) error { @@ -92,7 +129,7 @@ func (h *Handler) ListEntries(c *okapi.Context) error { if err != nil { return c.JSON(http.StatusInternalServerError, okapi.M{"error": "could not load entries"}) } - count, _ := h.store.Count(ctx) + count, _ := h.store.CountVisible(ctx) return c.OK(okapi.M{ "entries": entries, "total": count, @@ -115,6 +152,12 @@ func atoiDefault(s string, def int) int { // CreateEntry adds a new signature to the wall and broadcasts it live. func (h *Handler) CreateEntry(c *okapi.Context) error { + ctx := c.Request().Context() + // Read from the DB rather than a cache so a pause applies on every replica. + if st, err := h.store.Settings(ctx); err == nil && st.SigningPaused { + return c.JSON(http.StatusServiceUnavailable, okapi.M{"error": "signing is paused by a moderator"}) + } + var req createEntryRequest if err := c.Bind(&req); err != nil { return c.JSON(http.StatusBadRequest, okapi.M{"error": "invalid JSON body"}) @@ -126,19 +169,19 @@ func (h *Handler) CreateEntry(c *okapi.Context) error { if name == "" { name = "Anonymous" } - if len(name) > maxNameLen { - name = name[:maxNameLen] + if r := []rune(name); len(r) > maxNameLen { + name = string(r[:maxNameLen]) } if message == "" { return c.JSON(http.StatusBadRequest, okapi.M{"error": "message is required"}) } - if len(message) > maxMessageLen { + if utf8.RuneCountInString(message) > maxMessageLen { return c.JSON(http.StatusBadRequest, okapi.M{ "error": "message is too long (max 500 characters)", }) } - entry, err := h.store.Create(c.Request().Context(), name, message) + entry, err := h.store.Create(ctx, name, message, h.host, clientIP(c)) if err != nil { return c.JSON(http.StatusInternalServerError, okapi.M{"error": "could not save entry"}) } @@ -147,7 +190,7 @@ func (h *Handler) CreateEntry(c *okapi.Context) error { return c.Created(entry) } -// DeleteEntry removes an entry by id and broadcasts the removal live. +// DeleteEntry soft-deletes an entry by id and broadcasts the removal live. func (h *Handler) DeleteEntry(c *okapi.Context) error { id, err := strconv.ParseUint(c.Param("id"), 10, 64) if err != nil { @@ -180,9 +223,10 @@ func (h *Handler) Stream(c *okapi.Context) error { defer close(msgs) welcome := okapi.Message{Event: "welcome", Data: okapi.M{ - "version": h.version, - "online": h.broker.Online(), - "host": h.host, + "version": h.version, + "online": h.broker.Online(), + "replicas": len(h.broker.Replicas()), + "host": h.host, }} select { case msgs <- welcome: @@ -217,3 +261,16 @@ func (h *Handler) Stream(c *okapi.Context) error { } return err } + +// clientIP returns the caller's address as seen through the gateway. Without +// TRUSTED_PROXIES the forwarded headers are taken at face value, which is +// fine for display but spoofable. +func clientIP(c *okapi.Context) string { + if ip := c.RealIP(); net.ParseIP(ip) != nil { + return ip + } + if host, _, err := net.SplitHostPort(c.Request().RemoteAddr); err == nil { + return host + } + return "" +} diff --git a/main.go b/main.go index c78353e..531ec2b 100644 --- a/main.go +++ b/main.go @@ -5,17 +5,19 @@ // It is a classic "guestbook" — visitors leave a name and a message, and the // wall shows every signature. Everything a real Miabi deployment exercises is // here in miniature: env-based config (DATABASE_URL), an attached managed -// database, a GORM AutoMigrate on startup, a first-run seeder, a /healthz -// readiness probe, and a single static binary that ships the whole frontend -// embedded — no Node build step. +// database, an optional Redis for multi-replica live updates, a secret-guarded +// admin console, failure-injection endpoints, a scheduled cleanup job, a +// /healthz readiness probe, and a single static binary that ships the whole +// frontend embedded — no Node build step. // -// The HTTP server is driven by okapicli (a "server" subcommand with graceful -// lifecycle hooks), and the embedded UI is served with app.WebFS. +// Subcommands (okapicli): server (default), migrate, cleanup. package main import ( "context" "embed" + "io/fs" + "net/http" "os" "time" @@ -33,77 +35,195 @@ func main() { logger.Fatal("invalid configuration", "error", err) } - ctx := context.Background() + opts := []okapi.OptionFunc{okapi.WithPort(cfg.Port)} + if len(cfg.TrustedProxies) > 0 { + opts = append(opts, okapi.WithTrustedProxies(cfg.TrustedProxies...)) + } + app := okapi.New(opts...) + cli := okapicli.New(app, "guestbook") - store, err := NewStore(cfg.DBDriver, cfg.DatabaseDSN) + cli.Command("server", "Start the Guestbook HTTP server", func(cmd *okapicli.Command) error { + return runServer(cmd, cfg) + }).Int("port", "p", cfg.Port, "HTTP server port") + + cli.Command("migrate", "Apply database migrations and exit (e.g. as a pre-deploy step)", func(cmd *okapicli.Command) error { + store, err := openStore(cfg) + if err != nil { + return err + } + defer func() { _ = store.Close() }() + logger.Info("migrations applied", "db", dbDisplayName(cfg.DBDriver)) + return nil + }) + + cli.Command("cleanup", "Purge deleted entries and optionally expire old ones (run as a scheduled job)", func(cmd *okapicli.Command) error { + return runCleanup(cmd, cfg) + }). + Duration("purge-after", "", 7*24*time.Hour, "Permanently remove entries deleted longer ago than this"). + Duration("max-age", "", 0, "Also delete unpinned entries older than this (0 = keep forever)"). + Bool("dry-run", "", false, "Report what would be removed without changing anything") + + cli.DefaultCommand("server") + + if err := cli.Execute(); err != nil { + logger.Fatal("command failed", "error", err) + } +} + +func runServer(cmd *okapicli.Command, cfg Config) error { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + store, err := openStore(cfg) if err != nil { - logger.Fatal("failed to open database", "error", err) + return err } - - if err := migrateWithRetry(store, 30, 2*time.Second); err != nil { - logger.Fatal("failed to run migrations", "error", err) - } - logger.Info("schema ready", "app", cfg.AppName) - - // Seed sample data on first run (empty table only). if cfg.Seed { if err := Seed(ctx, store, cfg.DBDriver); err != nil { - logger.Fatal("failed to seed database", "error", err) + return err } } host, _ := os.Hostname() - broker := NewBroker() - h := &Handler{store: store, broker: broker, appName: cfg.AppName, version: cfg.Version, host: host} + broker := NewBroker(host, cfg.Version) + if cfg.RedisURL != "" { + if err := broker.UseRedis(cfg.RedisURL); err != nil { + return err + } + } + go broker.Run(ctx) + go publishTicks(ctx, broker, cfg.Version, host) + logger.Info("live updates ready", "broker", broker.Mode()) - // Broadcast a live server-time "tick" to all SSE clients once a second - // (the v2 clock card). Stopped on shutdown. - tickCtx, stopTicks := context.WithCancel(ctx) - go publishTicks(tickCtx, broker, cfg.Version, host) + h := &Handler{ + store: store, + broker: broker, + auth: NewAdminAuth(cfg.AdminToken), + appName: cfg.AppName, + version: cfg.Version, + host: host, + dbDriver: cfg.DBDriver, + debug: cfg.DebugEndpoints, + started: time.Now(), + } + if !h.auth.Enabled() { + logger.Warn("admin console disabled: set ADMIN_TOKEN to enable /admin") + } + if h.debug { + logger.Warn("debug endpoints enabled under /api/debug (admin auth required)") + } - app := okapi.New(okapi.WithPort(cfg.Port)) + app := cmd.Okapi() + app.WithPort(cmd.GetInt("port")) app.WithDebug() app.Use(okapi.LoggerMiddleware, okapi.RequestID()) + registerRoutes(app, h) - app.Get("/healthz", h.Health) - api := app.Group("/api") - api.Get("/info", h.Info) - api.Get("/time", h.Time) // current server time (v2 clock card) - api.Get("/entries", h.ListEntries) - api.Post("/entries", h.CreateEntry) - api.Delete("/entries/{id:int}", h.DeleteEntry) - api.Get("/stream", h.Stream) // Server-Sent Events: live entries + presence - - // Serve the embedded single-page UI. Okapi serves real files directly and - // falls back to index.html for any other path so a client-side router can - // take over. - app.WebFS("/", webFS, okapi.WebConfig{Root: "web"}) - - cli := okapicli.New(app, "guestbook") - cli.Command("server", "Start the Guestbook HTTP server", func(cmd *okapicli.Command) error { - cmd.Okapi().WithPort(cmd.GetInt("port")) - return cmd.CLI().RunServer(&okapicli.RunOptions{ - ShutdownTimeout: 10 * time.Second, - OnStarted: func() { - logger.Info("server started", "app", cfg.AppName, "version", cfg.Version, "port", cmd.GetInt("port")) - }, - OnShutdown: func() { - logger.Info("shutting down") - stopTicks() - _ = store.Close() - }, - }) - }).Int("port", "p", cfg.Port, "HTTP server port") - cli.DefaultCommand("server") - - if err := cli.Execute(); err != nil { - logger.Fatal("server error", "error", err) - } + return cmd.CLI().RunServer(&okapicli.RunOptions{ + ShutdownTimeout: 10 * time.Second, + OnStarted: func() { + logger.Info("server started", "app", cfg.AppName, "version", cfg.Version, "port", cmd.GetInt("port")) + }, + OnShutdown: func() { + logger.Info("shutting down") + closeCtx, done := context.WithTimeout(context.Background(), 2*time.Second) + defer done() + cancel() + broker.Close(closeCtx) + _ = store.Close() + }, + }) } -// publishTicks broadcasts a live server-time "tick" event to all connected SSE -// clients once a second. Each replica sends its own time/host, so under a -// canary rollout the clock card reveals exactly which build served you. +func registerRoutes(app *okapi.Okapi, h *Handler) { + app.Get("/healthz", h.Health) + + api := app.Group("/api") + api.Get("/info", h.Info) + api.Get("/time", h.Time) + api.Get("/settings", h.PublicSettings) + api.Get("/entries", h.ListEntries) + api.Post("/entries", h.CreateEntry) + api.Get("/stream", h.Stream) + + api.Get("/admin/session", h.AdminSession) + api.Post("/admin/login", h.AdminLogin) + api.Post("/admin/logout", h.AdminLogout) + + admin := app.Group("/api/admin", h.auth.Require) + admin.Get("/overview", h.AdminOverview) + admin.Get("/entries", h.AdminListEntries) + admin.Patch("/entries/{id:int}", h.AdminUpdateEntry) + admin.Delete("/entries/{id:int}", h.DeleteEntry) + admin.Get("/settings", h.AdminSettings) + admin.Put("/settings", h.AdminSaveSettings) + admin.Get("/export.csv", h.AdminExport) + + debug := app.Group("/api/debug", h.auth.Require, h.RequireDebug) + debug.Post("/crash", h.DebugCrash) + debug.Get("/slow", h.DebugSlow) + debug.Get("/error", h.DebugError) + debug.Post("/health", h.DebugHealth) + debug.Post("/cpu", h.DebugCPU) + debug.Post("/memory", h.DebugMemory) + debug.Post("/logs", h.DebugLogs) + + adminPage, err := fs.ReadFile(webFS, "web/admin.html") + if err != nil { + logger.Fatal("admin page missing from embedded assets", "error", err) + } + app.Get("/admin", func(c *okapi.Context) error { + return c.Data(http.StatusOK, "text/html; charset=utf-8", adminPage) + }) + + // Okapi serves real files directly and falls back to index.html for any + // other path so the client-side router can take over. + app.WebFS("/", webFS, okapi.WebConfig{Root: "web"}) +} + +func runCleanup(cmd *okapicli.Command, cfg Config) error { + ctx := context.Background() + store, err := openStore(cfg) + if err != nil { + return err + } + defer func() { _ = store.Close() }() + + dryRun := cmd.GetBool("dry-run") + now := time.Now() + + purged, err := store.Purge(ctx, now.Add(-cmd.GetDuration("purge-after")), dryRun) + if err != nil { + return err + } + var expired int64 + if maxAge := cmd.GetDuration("max-age"); maxAge > 0 { + if expired, err = store.Expire(ctx, now.Add(-maxAge), dryRun); err != nil { + return err + } + } + remaining, _ := store.Count(ctx) + logger.Info("cleanup finished", "purged", purged, "expired", expired, "remaining", remaining, "dry_run", dryRun) + return nil +} + +// openStore connects to the database and applies migrations, retrying while +// the database comes up. +func openStore(cfg Config) (*Store, error) { + store, err := NewStore(cfg.DBDriver, cfg.DatabaseDSN) + if err != nil { + return nil, err + } + if err := migrateWithRetry(store, 30, 2*time.Second); err != nil { + _ = store.Close() + return nil, err + } + return store, nil +} + +// publishTicks sends each replica's own clock to its local clients once a +// second, so under a canary rollout the clock card reveals which build served +// you. Ticks stay local on purpose, even with Redis. func publishTicks(ctx context.Context, b *Broker, version, host string) { t := time.NewTicker(time.Second) defer t.Stop() @@ -112,7 +232,7 @@ func publishTicks(ctx context.Context, b *Broker, version, host string) { case <-ctx.Done(): return case now := <-t.C: - b.Publish(Event{ + b.PublishLocal(Event{ Type: "tick", Time: now.Format(time.RFC3339), Version: version, diff --git a/scripts/hammer.sh b/scripts/hammer.sh new file mode 100755 index 0000000..d722bd7 --- /dev/null +++ b/scripts/hammer.sh @@ -0,0 +1,46 @@ +#!/usr/bin/env bash +# Hit an endpoint in a loop during a deploy and report any non-2xx responses, +# plus which version/host answered. Proves zero-downtime rollouts. +# +# scripts/hammer.sh https://guestbook.example.com # /api/time, 10 req/s +# scripts/hammer.sh https://guestbook.example.com 20 /healthz +set -u + +BASE="${1:?usage: hammer.sh [req-per-sec] [path]}" +RATE="${2:-10}" +PATH_="${3:-/api/time}" +URL="${BASE%/}${PATH_}" +SLEEP=$(awk "BEGIN { print 1 / $RATE }") + +ok=0 fail=0 total=0 +# A temp file instead of an associative array: macOS still ships bash 3.2. +seen=$(mktemp) + +summary() { + echo + echo "── $total requests · $ok ok · $fail failed ──" + sort "$seen" | uniq -c | sort -rn + rm -f "$seen" + exit 0 +} +trap summary INT TERM + +echo "hammering $URL at ~${RATE} req/s — Ctrl-C for the summary" +while true; do + resp=$(curl -s -m 5 -w '\n%{http_code}' "$URL" 2>/dev/null) + code="${resp##*$'\n'}" + body="${resp%$'\n'*}" + total=$((total + 1)) + version=$(printf '%s' "$body" | sed -n 's/.*"version":"\([^"]*\)".*/\1/p') + host=$(printf '%s' "$body" | sed -n 's/.*"host":"\([^"]*\)".*/\1/p') + key="${code} v${version:-?} ${host:-?}" + echo "$key" >> "$seen" + if [[ "$code" =~ ^2 ]]; then + ok=$((ok + 1)) + printf '\r%s ok=%d fail=%d last: %-40s' "$(date +%T)" "$ok" "$fail" "$key" + else + fail=$((fail + 1)) + printf '\n%s FAIL %s\n' "$(date +%T)" "${code:-000}" + fi + sleep "$SLEEP" +done diff --git a/seed.go b/seed.go index f11c580..16b4854 100644 --- a/seed.go +++ b/seed.go @@ -42,22 +42,24 @@ func seedEntries(driver string) []struct{ Name, Message string } { } } +// Seed populates an empty wall with sample signatures. func Seed(ctx context.Context, store *Store, driver string) error { - count, err := store.Count(ctx) - if err != nil { - return err - } - if count > 0 { - logger.Info("seed skipped, entries already present", "count", count) - return nil - } - - entries := seedEntries(driver) - for _, e := range entries { - if _, err := store.Create(ctx, e.Name, e.Message); err != nil { + return store.Exclusive(ctx, func(tx *Store) error { + count, err := tx.Count(ctx) + if err != nil { return err } - } - logger.Info("database seeded", "entries", len(entries)) - return nil + if count > 0 { + logger.Info("seed skipped, entries already present", "count", count) + return nil + } + entries := seedEntries(driver) + for _, e := range entries { + if _, err := tx.Create(ctx, e.Name, e.Message, "seed", ""); err != nil { + return err + } + } + logger.Info("database seeded", "entries", len(entries)) + return nil + }) } diff --git a/store.go b/store.go index 58c8e5c..b6f35f2 100644 --- a/store.go +++ b/store.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "fmt" "os" "path/filepath" @@ -19,11 +20,54 @@ type Entry struct { ID uint `gorm:"primaryKey" json:"id"` Name string `gorm:"type:varchar(60);not null" json:"name"` Message string `gorm:"type:varchar(500);not null" json:"message"` + Pinned bool `gorm:"not null;default:false;index" json:"pinned"` + Hidden bool `gorm:"not null;default:false;index" json:"hidden"` + Replica string `gorm:"type:varchar(128)" json:"replica,omitempty"` + IP string `gorm:"type:varchar(64)" json:"-"` CreatedAt time.Time `gorm:"index" json:"created_at"` UpdatedAt time.Time `json:"-"` DeletedAt gorm.DeletedAt `gorm:"index" json:"-"` } +// Setting is a key/value row holding app-wide state shared by all replicas. +type Setting struct { + Key string `gorm:"primaryKey;type:varchar(64)"` + Value string `gorm:"type:text;not null"` + UpdatedAt time.Time +} + +// Settings is the moderator-controlled state of the wall. +type Settings struct { + SigningPaused bool `json:"signing_paused"` + Banner string `json:"banner"` +} + +// EntryFilter selects entries for the admin console. +type EntryFilter struct { + Status string // "", "visible", "hidden" or "pinned" + Query string + Limit int + Offset int +} + +// DayCount is the number of signatures created on a given UTC day. +type DayCount struct { + Day string `json:"day"` + Count int `json:"count"` +} + +// Stats summarises the wall for the admin overview. +type Stats struct { + Total int64 `json:"total"` + Visible int64 `json:"visible"` + Hidden int64 `json:"hidden"` + Pinned int64 `json:"pinned"` + Today int64 `json:"today"` + Daily []DayCount `json:"daily"` +} + +const settingsKey = "settings" + type Store struct { db *gorm.DB } @@ -92,11 +136,15 @@ func (s *Store) Ping(ctx context.Context) error { return sqlDB.PingContext(ctx) } +// Migrate applies the schema under the same lock as seeding, since +// concurrent AutoMigrate calls race on Postgres catalog inserts. func (s *Store) Migrate() error { - return s.db.AutoMigrate(&Entry{}) + return s.Exclusive(context.Background(), func(tx *Store) error { + return tx.db.AutoMigrate(&Entry{}, &Setting{}) + }) } -// List returns entries newest first, with limit/offset pagination. +// List returns the public wall: visible entries, pinned first, then newest. func (s *Store) List(ctx context.Context, limit, offset int) ([]Entry, error) { if limit <= 0 || limit > 200 { limit = 100 @@ -106,6 +154,8 @@ func (s *Store) List(ctx context.Context, limit, offset int) ([]Entry, error) { } var entries []Entry err := s.db.WithContext(ctx). + Where("hidden = ?", false). + Order("pinned DESC"). Order("created_at DESC"). Limit(limit). Offset(offset). @@ -114,8 +164,8 @@ func (s *Store) List(ctx context.Context, limit, offset int) ([]Entry, error) { } // Create inserts a new entry and returns it with its generated id/timestamp. -func (s *Store) Create(ctx context.Context, name, message string) (Entry, error) { - e := Entry{Name: name, Message: message} +func (s *Store) Create(ctx context.Context, name, message, replica, ip string) (Entry, error) { + e := Entry{Name: name, Message: message, Replica: replica, IP: ip} err := s.db.WithContext(ctx).Create(&e).Error return e, err } @@ -126,9 +176,169 @@ func (s *Store) Delete(ctx context.Context, id uint64) (bool, error) { return res.RowsAffected > 0, res.Error } -// Count returns the total number of (non-deleted) entries. +// schemaLockID is an arbitrary Postgres advisory-lock key. +const schemaLockID = 7_406_381 + +// Exclusive runs fn in a transaction holding a Postgres advisory lock, so +// replicas booting together migrate and seed one at a time. +func (s *Store) Exclusive(ctx context.Context, fn func(tx *Store) error) error { + return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if tx.Name() == "postgres" { + if err := tx.Exec("SELECT pg_advisory_xact_lock(?)", schemaLockID).Error; err != nil { + return err + } + } + return fn(&Store{db: tx}) + }) +} + +// Count returns the total number of (non-deleted) entries, hidden included. func (s *Store) Count(ctx context.Context) (int64, error) { var n int64 err := s.db.WithContext(ctx).Model(&Entry{}).Count(&n).Error return n, err } + +// CountVisible returns the number of entries shown on the public wall. +func (s *Store) CountVisible(ctx context.Context) (int64, error) { + var n int64 + err := s.db.WithContext(ctx).Model(&Entry{}).Where("hidden = ?", false).Count(&n).Error + return n, err +} + +// Get returns a single entry by id. +func (s *Store) Get(ctx context.Context, id uint64) (Entry, error) { + var e Entry + err := s.db.WithContext(ctx).First(&e, id).Error + return e, err +} + +// AdminList returns entries matching f, newest first, and the match count. +func (s *Store) AdminList(ctx context.Context, f EntryFilter) ([]Entry, int64, error) { + q := s.db.WithContext(ctx).Model(&Entry{}) + switch f.Status { + case "visible": + q = q.Where("hidden = ?", false) + case "hidden": + q = q.Where("hidden = ?", true) + case "pinned": + q = q.Where("pinned = ?", true) + } + if term := strings.TrimSpace(f.Query); term != "" { + like := "%" + strings.ToLower(term) + "%" + q = q.Where("LOWER(name) LIKE ? OR LOWER(message) LIKE ? OR ip LIKE ?", like, like, like) + } + + var total int64 + if err := q.Count(&total).Error; err != nil { + return nil, 0, err + } + var entries []Entry + err := q.Order("created_at DESC").Limit(f.Limit).Offset(f.Offset).Find(&entries).Error + return entries, total, err +} + +// SetFlags updates the moderation flags that are non-nil and returns the entry. +func (s *Store) SetFlags(ctx context.Context, id uint64, pinned, hidden *bool) (Entry, error) { + updates := map[string]any{} + if pinned != nil { + updates["pinned"] = *pinned + } + if hidden != nil { + updates["hidden"] = *hidden + } + e, err := s.Get(ctx, id) + if err != nil || len(updates) == 0 { + return e, err + } + if err := s.db.WithContext(ctx).Model(&e).Updates(updates).Error; err != nil { + return e, err + } + return s.Get(ctx, id) +} + +// All returns every entry, oldest first, for export. +func (s *Store) All(ctx context.Context) ([]Entry, error) { + var entries []Entry + err := s.db.WithContext(ctx).Order("created_at ASC").Find(&entries).Error + return entries, err +} + +// Stats computes the admin overview, including per-day counts for the last +// `days` days. Bucketing happens in Go so it works the same on both drivers. +func (s *Store) Stats(ctx context.Context, days int) (Stats, error) { + var st Stats + db := s.db.WithContext(ctx).Model(&Entry{}) + if err := db.Count(&st.Total).Error; err != nil { + return st, err + } + s.db.WithContext(ctx).Model(&Entry{}).Where("hidden = ?", true).Count(&st.Hidden) + s.db.WithContext(ctx).Model(&Entry{}).Where("pinned = ?", true).Count(&st.Pinned) + st.Visible = st.Total - st.Hidden + + now := time.Now().UTC() + today := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC) + since := today.AddDate(0, 0, -(days - 1)) + + var stamps []time.Time + if err := s.db.WithContext(ctx).Model(&Entry{}). + Where("created_at >= ?", since). + Pluck("created_at", &stamps).Error; err != nil { + return st, err + } + buckets := make(map[string]int, days) + for _, t := range stamps { + buckets[t.UTC().Format("2006-01-02")]++ + } + for d := since; !d.After(today); d = d.AddDate(0, 0, 1) { + key := d.Format("2006-01-02") + st.Daily = append(st.Daily, DayCount{Day: key, Count: buckets[key]}) + } + st.Today = int64(buckets[today.Format("2006-01-02")]) + return st, nil +} + +// Settings loads the shared wall settings, returning defaults when unset. +func (s *Store) Settings(ctx context.Context) (Settings, error) { + var row Setting + var st Settings + err := s.db.WithContext(ctx).Where(&Setting{Key: settingsKey}).Limit(1).Find(&row).Error + if err != nil || row.Value == "" { + return st, err + } + err = json.Unmarshal([]byte(row.Value), &st) + return st, err +} + +// SaveSettings persists the shared wall settings. +func (s *Store) SaveSettings(ctx context.Context, st Settings) error { + payload, err := json.Marshal(st) + if err != nil { + return err + } + return s.db.WithContext(ctx).Save(&Setting{Key: settingsKey, Value: string(payload)}).Error +} + +// Purge permanently removes entries soft-deleted before cutoff. +func (s *Store) Purge(ctx context.Context, cutoff time.Time, dryRun bool) (int64, error) { + q := s.db.WithContext(ctx).Unscoped().Where("deleted_at IS NOT NULL AND deleted_at < ?", cutoff) + if dryRun { + var n int64 + err := q.Model(&Entry{}).Count(&n).Error + return n, err + } + res := q.Delete(&Entry{}) + return res.RowsAffected, res.Error +} + +// Expire soft-deletes unpinned entries created before cutoff. +func (s *Store) Expire(ctx context.Context, cutoff time.Time, dryRun bool) (int64, error) { + q := s.db.WithContext(ctx).Where("pinned = ? AND created_at < ?", false, cutoff) + if dryRun { + var n int64 + err := q.Model(&Entry{}).Count(&n).Error + return n, err + } + res := q.Delete(&Entry{}) + return res.RowsAffected, res.Error +} diff --git a/store_test.go b/store_test.go new file mode 100644 index 0000000..eb21a51 --- /dev/null +++ b/store_test.go @@ -0,0 +1,126 @@ +package main + +import ( + "context" + "encoding/json" + "path/filepath" + "strings" + "testing" + "time" +) + +func newTestStore(t *testing.T) *Store { + t.Helper() + s, err := NewStore("sqlite", filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatal(err) + } + if err := s.Migrate(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = s.Close() }) + return s +} + +func TestModeration(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + + a, _ := s.Create(ctx, "a", "first", "r1", "") + b, _ := s.Create(ctx, "b", "second", "r2", "") + c, _ := s.Create(ctx, "c", "third", "r1", "") + + yes := true + if _, err := s.SetFlags(ctx, uint64(a.ID), &yes, nil); err != nil { + t.Fatal(err) + } + if _, err := s.SetFlags(ctx, uint64(b.ID), nil, &yes); err != nil { + t.Fatal(err) + } + + wall, err := s.List(ctx, 10, 0) + if err != nil { + t.Fatal(err) + } + if len(wall) != 2 || wall[0].ID != a.ID || wall[1].ID != c.ID { + t.Fatalf("want pinned a then c (b hidden), got %+v", wall) + } + if n, _ := s.CountVisible(ctx); n != 2 { + t.Fatalf("visible = %d, want 2", n) + } + + hidden, total, err := s.AdminList(ctx, EntryFilter{Status: "hidden", Limit: 10}) + if err != nil || total != 1 || hidden[0].ID != b.ID { + t.Fatalf("hidden filter: total=%d err=%v", total, err) + } + found, total, _ := s.AdminList(ctx, EntryFilter{Query: "THIRD", Limit: 10}) + if total != 1 || found[0].ID != c.ID { + t.Fatalf("search should be case-insensitive, got total=%d", total) + } + + st, err := s.Stats(ctx, 7) + if err != nil { + t.Fatal(err) + } + if st.Total != 3 || st.Hidden != 1 || st.Pinned != 1 || st.Today != 3 || len(st.Daily) != 7 { + t.Fatalf("unexpected stats %+v", st) + } +} + +func TestSettingsRoundTrip(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + + if st, err := s.Settings(ctx); err != nil || st.SigningPaused || st.Banner != "" { + t.Fatalf("defaults: %+v %v", st, err) + } + want := Settings{SigningPaused: true, Banner: "hello"} + if err := s.SaveSettings(ctx, want); err != nil { + t.Fatal(err) + } + if err := s.SaveSettings(ctx, want); err != nil { + t.Fatal("saving twice should upsert:", err) + } + if got, _ := s.Settings(ctx); got != want { + t.Fatalf("got %+v, want %+v", got, want) + } +} + +func TestCleanup(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + + old, _ := s.Create(ctx, "old", "old", "r", "") + pinned, _ := s.Create(ctx, "pin", "pinned", "r", "") + gone, _ := s.Create(ctx, "gone", "deleted", "r", "") + past := time.Now().Add(-48 * time.Hour) + s.db.Model(&Entry{}).Where("id IN ?", []uint{old.ID, pinned.ID}).Update("created_at", past) + yes := true + _, _ = s.SetFlags(ctx, uint64(pinned.ID), &yes, nil) + _, _ = s.Delete(ctx, uint64(gone.ID)) + + if n, _ := s.Expire(ctx, time.Now().Add(-24*time.Hour), true); n != 1 { + t.Fatalf("dry-run expire = %d, want 1 (pinned is kept)", n) + } + if n, _ := s.Expire(ctx, time.Now().Add(-24*time.Hour), false); n != 1 { + t.Fatalf("expire = %d, want 1", n) + } + if n, _ := s.Purge(ctx, time.Now().Add(time.Minute), false); n != 2 { + t.Fatalf("purge = %d, want 2 (expired + deleted)", n) + } + if n, _ := s.Count(ctx); n != 1 { + t.Fatalf("remaining = %d, want 1", n) + } +} + +func TestIPIsAdminOnly(t *testing.T) { + e := Entry{ID: 1, Name: "a", Message: "m", IP: "203.0.113.7"} + pub, _ := json.Marshal(e) + if strings.Contains(string(pub), "203.0.113.7") { + t.Fatalf("public JSON leaks IP: %s", pub) + } + adm, _ := json.Marshal(toAdmin(e)) + if !strings.Contains(string(adm), `"ip":"203.0.113.7"`) || !strings.Contains(string(adm), `"name":"a"`) { + t.Fatalf("admin JSON missing fields: %s", adm) + } +} diff --git a/web/admin.html b/web/admin.html new file mode 100644 index 0000000..f06d4db --- /dev/null +++ b/web/admin.html @@ -0,0 +1,707 @@ + + + + + + Guestbook Admin + + + + +
+
+
Guestbook Admin
+
+ served by … + v… + ← Wall + +
+
+ + + + + + +
+ +
+ + + + diff --git a/web/badges/running-on-miabi-dark.svg b/web/badges/running-on-miabi-dark.svg new file mode 100644 index 0000000..3d494b6 --- /dev/null +++ b/web/badges/running-on-miabi-dark.svg @@ -0,0 +1,21 @@ + + Running on Miabi + + + + + + + + + + + + + + + + + Running on Miabi + + diff --git a/web/index.html b/web/index.html index 08b2005..ecee4b4 100644 --- a/web/index.html +++ b/web/index.html @@ -143,13 +143,28 @@ } .entry .who { font-weight: 600; } .entry .when { color: var(--muted); } - .del { - position: absolute; top: 14px; right: 14px; cursor: pointer; - background: transparent; border: 0; color: var(--muted); font-size: 17px; line-height: 1; - padding: 4px 7px; border-radius: 8px; opacity: 0; transition: opacity .15s, color .15s, background .15s; + .entry .via { + margin-left: auto; font-size: 11.5px; color: var(--muted); font-family: ui-monospace, SFMono-Regular, Menlo, monospace; + background: rgba(0,0,0,.2); border: 1px solid var(--card-border); padding: 2px 8px; border-radius: 999px; + max-width: 180px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } - li.entry:hover .del { opacity: 1; } - .del:hover { color: var(--danger); background: rgba(244, 63, 94, 0.12); } + li.entry.pinned { border-color: rgba(139, 92, 246, 0.55); background: rgba(139, 92, 246, 0.08); } + .pin { + position: absolute; top: 14px; right: 16px; font-size: 11px; font-weight: 700; letter-spacing: .06em; + text-transform: uppercase; color: var(--accent-2); + } + + .banner { + display: flex; gap: 10px; align-items: flex-start; padding: 13px 18px; margin-bottom: 18px; + border-color: rgba(245, 158, 11, 0.45); background: rgba(245, 158, 11, 0.1); font-size: 14.5px; line-height: 1.5; + } + .banner .icon { flex: 0 0 auto; } + .paused { + text-align: center; padding: 22px 20px; margin-bottom: 14px; color: var(--muted); font-size: 15px; + } + .paused b { color: var(--text); } + .replicas { font-size: 12.5px; color: var(--muted); padding: 5px 12px; border-radius: 999px; + background: var(--card); border: 1px solid var(--card-border); white-space: nowrap; } .empty, .loading { text-align: center; color: var(--muted); padding: 40px 20px; font-size: 15px; } @@ -170,6 +185,8 @@ footer { text-align: center; margin-top: 46px; color: var(--muted); font-size: 13px; line-height: 1.8; } footer a { color: var(--accent-2); text-decoration: none; } footer a:hover { text-decoration: underline; } + .miabi-badge { display: block; width: fit-content; margin: 0 auto 16px; line-height: 0; border-radius: 16px; transition: transform .15s, box-shadow .15s; } + .miabi-badge:hover { transform: translateY(-1px); box-shadow: 0 8px 24px -8px rgba(139, 92, 246, 0.6); } .stack { display: inline-flex; gap: 6px; flex-wrap: wrap; justify-content: center; } .pill { background: var(--card); border: 1px solid var(--card-border); padding: 3px 10px; border-radius: 999px; } @@ -189,6 +206,7 @@
v… + 0 online
@@ -202,6 +220,8 @@ + +
@@ -213,6 +233,7 @@
+
@@ -249,14 +270,19 @@
@@ -271,9 +297,9 @@ const api = { info: () => fetch('/api/info').then(r => r.json()), time: () => fetch('/api/time').then(r => r.json()), + settings: () => fetch('/api/settings').then(r => r.json()), list: (limit, offset) => fetch(`/api/entries?limit=${limit}&offset=${offset}`).then(r => r.json()), create: (body) => fetch('/api/entries', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body) }), - remove: (id) => fetch('/api/entries/' + id, { method: 'DELETE' }), }; let total = 0; // total signatures (kept in sync via SSE) @@ -300,13 +326,16 @@ } function cardHTML(e) { - return `
  • - + const via = e.replica && e.replica !== 'seed' + ? `via ${esc(e.replica)}` : ''; + return `
  • + ${e.pinned ? '📌 Pinned' : ''}
    ${esc(e.message)}
    ${esc(initials(e.name))} ${esc(e.name)} · ${timeAgo(e.created_at)} + ${via}
  • `; } @@ -332,7 +361,20 @@ let h = 0; for (const c of String(v)) h = (h * 31 + c.charCodeAt(0)) >>> 0; el.style.background = `hsl(${h % 360} 70% 52%)`; } - function setOnline(n) { $('online').textContent = n; } + function setOnline(n) { $('online').textContent = n || 0; } + function setReplicas(n) { + if (!n) return; + $('replicas').hidden = false; + $('replicas').textContent = n === 1 ? '1 replica' : n + ' replicas'; + } + function applySettings(st) { + const banner = (st && st.banner) || ''; + $('banner').hidden = !banner; + $('bannerText').textContent = banner; + const paused = !!(st && st.signing_paused); + $('paused').hidden = !paused; + $('form').hidden = paused; + } // ---- views -------------------------------------------------------------- async function loadHome() { @@ -383,21 +425,25 @@ window.addEventListener('popstate', route); // ---- live SSE events ---------------------------------------------------- + // Pinned entries stay on top, so a new signature goes right after them. function onCreated(entry) { setTotals(total + 1); if (currentView() === 'home') { const ul = $('entries'); if (ul.querySelector('.empty')) ul.innerHTML = ''; - ul.insertAdjacentHTML('afterbegin', cardHTML(entry)); + const pins = ul.querySelectorAll('.entry.pinned'); + const html = cardHTML(entry); + if (pins.length) pins[pins.length - 1].insertAdjacentHTML('afterend', html); + else ul.insertAdjacentHTML('afterbegin', html); while (ul.children.length > HOME_SIZE) ul.lastElementChild.remove(); } else if (allOffset === 0) { loadAll(); } } + function reloadView() { currentView() === 'all' ? loadAll() : loadHome(); } function onDeleted(id) { - setTotals(Math.max(0, total - 1)); document.querySelectorAll(`.entry[data-id="${id}"]`).forEach((el) => el.remove()); - if (currentView() === 'all') loadAll(); + reloadView(); } // ---- live server clock (v2) -------------------------------------------- @@ -415,12 +461,17 @@ clockStaleTimer = setTimeout(() => $('clock').classList.add('stale'), 4000); } + let connectedOnce = false; function connectSSE() { const es = new EventSource('/api/stream'); - es.addEventListener('welcome', (e) => { const d = JSON.parse(e.data); setVersion(d.version); setOnline(d.online); }); - es.addEventListener('presence', (e) => { const d = JSON.parse(e.data); setOnline(d.online); }); - es.addEventListener('created', (e) => { const d = JSON.parse(e.data); if (d.entry) onCreated(d.entry); }); + es.addEventListener('welcome', (e) => { const d = JSON.parse(e.data); setVersion(d.version); setOnline(d.online); setReplicas(d.replicas); }); + es.addEventListener('presence', (e) => { const d = JSON.parse(e.data); setOnline(d.online); setReplicas(d.replicas); }); + es.addEventListener('created', (e) => { const d = JSON.parse(e.data); if (d.entry && !d.entry.hidden) onCreated(d.entry); }); + es.addEventListener('updated', () => reloadView()); es.addEventListener('deleted', (e) => { const d = JSON.parse(e.data); onDeleted(d.id); }); + es.addEventListener('settings', (e) => { const d = JSON.parse(e.data); applySettings(d.settings); }); + // Resync after a reconnect: events sent while disconnected were missed. + es.addEventListener('open', () => { if (connectedOnce) { reloadView(); api.settings().then(applySettings).catch(() => {}); } connectedOnce = true; }); es.addEventListener('tick', (e) => { const d = JSON.parse(e.data); showTime(d); setVersion(d.version); }); // EventSource auto-reconnects on error; nothing to do here. } @@ -442,22 +493,16 @@ finally { btn.disabled = false; } }); - document.addEventListener('click', async (e) => { - const btn = e.target.closest('.del'); - if (!btn) return; - const li = btn.closest('.entry'); - const id = li.dataset.id; - li.style.opacity = '.4'; - const res = await api.remove(id); - if (!(res.ok || res.status === 404)) { li.style.opacity = '1'; toast('Could not delete'); } - // Removal is reflected via SSE ("deleted"). - }); - $('prev').addEventListener('click', () => { allOffset = Math.max(0, allOffset - PAGE_SIZE); loadAll(); }); $('next').addEventListener('click', () => { allOffset += PAGE_SIZE; loadAll(); }); // ---- boot --------------------------------------------------------------- - api.info().then((d) => { setVersion(d.version); setOnline(d.online); }).catch(() => {}); + api.info().then((d) => { + setVersion(d.version); setOnline(d.online); setReplicas((d.replicas || []).length); + $('adminLink').hidden = !d.admin; + if (d.broker) { $('brokerPill').hidden = false; $('brokerPill').textContent = d.broker === 'redis' ? 'Redis pub/sub' : 'In-process broker'; } + }).catch(() => {}); + api.settings().then(applySettings).catch(() => {}); api.time().then(showTime).catch(() => {}); // immediate clock; SSE ticks keep it live route(); connectSSE();