feat: enhance chat service with Redis pub/sub for multi-instance message handling, add instance ID for deduplication
This commit is contained in:
1 parent
d991d6b6c7
commit
7244c1d658
4 files changed
+90
-19
No files matched your search
@@ -1,5 +1,7 @@
|
|||||||
# BACKLOG
|
# BACKLOG
|
||||||
|
|
||||||
|
Status reflects 2026-08-06 code scan + implementation batches.
|
||||||
|
|
||||||
## Playout & Delivery
|
## Playout & Delivery
|
||||||
|
|
||||||
- [x] Implement FFmpeg + MediaMTX/SRS continuous HLS playout from playlist
|
- [x] Implement FFmpeg + MediaMTX/SRS continuous HLS playout from playlist
|
||||||
@@ -55,15 +57,15 @@
|
|||||||
## Infrastructure — Docker
|
## Infrastructure — Docker
|
||||||
|
|
||||||
- [x] Create Dockerfiles for playout, chat, nginx services
|
- [x] Create Dockerfiles for playout, chat, nginx services
|
||||||
- [x] Create docker-compose.yml for all services
|
- [x] Create docker-compose.yml for all services (11 containers)
|
||||||
- [x] Create docker-compose.prod.yml override for production
|
- [x] Create docker-compose.prod.yml override for production
|
||||||
|
|
||||||
## Infrastructure — CI/CD (Gitea Actions)
|
## Infrastructure — CI/CD (Gitea Actions)
|
||||||
|
|
||||||
- [x] Implement CI workflow (lint, build, test, push images)
|
- [x] Implement CI workflow (lint, build, vet, push images)
|
||||||
- [x] Implement CD workflow (pull, deploy, health checks — IP-agnostic)
|
- [x] Implement CD workflow (pull, deploy, health checks — IP-agnostic)
|
||||||
- [x] Generate .env from Gitea secrets in CD workflow
|
- [x] Generate .env from Gitea secrets in CD workflow
|
||||||
- [x] Seed Prometheus config into named volume during deploy
|
- [x] Seed Prometheus config + alerts into named volume during deploy
|
||||||
- [x] Implement health check validation with retry in CD pipeline
|
- [x] Implement health check validation with retry in CD pipeline
|
||||||
|
|
||||||
## Cloudflare Configuration
|
## Cloudflare Configuration
|
||||||
@@ -79,10 +81,10 @@
|
|||||||
|
|
||||||
## Monitoring & Observability
|
## Monitoring & Observability
|
||||||
|
|
||||||
- [x] Set up Prometheus with chat scrape target
|
- [x] Set up Prometheus with chat/playout/nginx/redis/postgres/node scrape targets
|
||||||
- [ ] Add nginx/redis/postgres/node exporters
|
- [x] Add nginx-exporter, redis-exporter, postgres-exporter, node-exporter
|
||||||
- [x] Import Grafana app + operational dashboards
|
- [x] Import Grafana app + operational dashboards
|
||||||
- [ ] Define Prometheus alerting rules
|
- [x] Define Prometheus alerting rules (8 rules: ChatDown, PlayoutDown, RedisDown, etc.)
|
||||||
- [ ] Set up UptimeRobot external monitoring
|
- [ ] Set up UptimeRobot external monitoring
|
||||||
|
|
||||||
## Security
|
## Security
|
||||||
@@ -90,15 +92,18 @@
|
|||||||
- [x] Implement admin cookie-session auth (HttpOnly, SameSite=Strict, Secure when TLS)
|
- [x] Implement admin cookie-session auth (HttpOnly, SameSite=Strict, Secure when TLS)
|
||||||
- [x] Implement server-side session expiration
|
- [x] Implement server-side session expiration
|
||||||
- [ ] Enforce HTTPS/WSS at edge and origin
|
- [ ] Enforce HTTPS/WSS at edge and origin
|
||||||
- [x] Implement rate limiting for chat messages with Redis windowing
|
- [x] Implement rate limiting for chat messages (Redis windowing)
|
||||||
|
- [x] Implement WebSocket connection rate limiting (5/IP/60s)
|
||||||
|
- [x] Implement admin login rate limiting (5/IP/5min)
|
||||||
|
|
||||||
## Data & Storage
|
## Data & Storage
|
||||||
|
|
||||||
- [x] Set up PostgreSQL/SQLite for chat history and moderation data
|
- [x] Set up PostgreSQL/SQLite for chat history and moderation data
|
||||||
- [x] Set up Redis with append-only persistence + maxmemory policy
|
- [x] Set up Redis with append-only persistence + maxmemory policy
|
||||||
- [ ] Implement daily/weekly/monthly Postgres backups
|
- [x] Implement daily Postgres backups to R2 (14-day retention)
|
||||||
- [ ] Implement backup integrity verification
|
- [x] Implement weekly full backup snapshots (8-week retention)
|
||||||
- [ ] Document and rehearse recovery runbook
|
- [x] Implement backup integrity verification (restore test)
|
||||||
|
- [ ] Document and rehearse recovery runbook (quarterly drill)
|
||||||
|
|
||||||
## Scaling & Performance
|
## Scaling & Performance
|
||||||
|
|
||||||
@@ -117,5 +122,5 @@
|
|||||||
- [ ] Document deployment runbook in DEPLOYMENT.md
|
- [ ] Document deployment runbook in DEPLOYMENT.md
|
||||||
- [ ] Document monitoring setup in MONITORING.md
|
- [ ] Document monitoring setup in MONITORING.md
|
||||||
- [ ] Document security model in SECURITY.md
|
- [ ] Document security model in SECURITY.md
|
||||||
- [ ] Create incident runbook
|
- [x] Create incident runbook (docs/incident-runbook.md)
|
||||||
- [ ] Write Level 2/3 building block views
|
- [ ] Write Level 2/3 building block views
|
||||||
+6
-6
@@ -4,12 +4,12 @@ Prometheus + Grafana monitoring for Linear Void.
|
|||||||
|
|
||||||
## Architecture
|
## Architecture
|
||||||
|
|
||||||
```
|
```text
|
||||||
┌─────────────┐ ┌──────────────┐ ┌───────────┐
|
┌──────────────┐ ┌──────────────┐ ┌───────────┐
|
||||||
│ Exporters │───▶│ Prometheus │───▶│ Grafana │
|
│ Exporters │───> │ Prometheus │───>│ Grafana │
|
||||||
│ (7 targets) │ │ :9090 │ │ :3001 │
|
│ (7 targets) │ │ :9090 │ │ :3001 │
|
||||||
└─────────────┘ └──────┬───────┘ └───────────┘
|
└──────────────┘ └───────┬──────┘ └───────────┘
|
||||||
│
|
│
|
||||||
alerts.yml (8 rules)
|
alerts.yml (8 rules)
|
||||||
```
|
```
|
||||||
|
|
||||||
|
|||||||
@@ -14,7 +14,7 @@
|
|||||||
|
|
||||||
The architecture (detailed in [`docs/architecture/README.md`](docs/architecture/README.md)) uses:
|
The architecture (detailed in [`docs/architecture/README.md`](docs/architecture/README.md)) uses:
|
||||||
|
|
||||||
- Hetzner Ashburn VPS for playout (FFmpeg + MediaMTX/SRS) and chat *(planned production; currently test lab at 192.168.88.40)*.
|
- Hetzner Ashburn VPS for playout (FFmpeg + MediaMTX/SRS) and chat _(planned production; currently test lab at 192.168.88.40)_.
|
||||||
- Cloudflare R2 + CDN with aggressive immutable caching of HLS `.ts` segments ($0 egress).
|
- Cloudflare R2 + CDN with aggressive immutable caching of HLS `.ts` segments ($0 egress).
|
||||||
- Lightweight static frontend and WebSocket chat backend.
|
- Lightweight static frontend and WebSocket chat backend.
|
||||||
- Gitea self-hosted CI/CD (Actions) and Container Registry.
|
- Gitea self-hosted CI/CD (Actions) and Container Registry.
|
||||||
|
|||||||
@@ -41,8 +41,17 @@ var (
|
|||||||
// Rate limit config (env-overridable)
|
// Rate limit config (env-overridable)
|
||||||
rateLimitMax = 10
|
rateLimitMax = 10
|
||||||
rateLimitWindow = 10 * time.Second
|
rateLimitWindow = 10 * time.Second
|
||||||
|
|
||||||
|
// Instance ID for multi-instance pub/sub deduplication
|
||||||
|
instanceID = fmt.Sprintf("chat-%s", generateShortID())
|
||||||
)
|
)
|
||||||
|
|
||||||
|
func generateShortID() string {
|
||||||
|
b := make([]byte, 6)
|
||||||
|
rand.Read(b)
|
||||||
|
return hex.EncodeToString(b)
|
||||||
|
}
|
||||||
|
|
||||||
type clientInfo struct {
|
type clientInfo struct {
|
||||||
Username string
|
Username string
|
||||||
Room string
|
Room string
|
||||||
@@ -2161,7 +2170,7 @@ func handleMetrics(w http.ResponseWriter, r *http.Request) {
|
|||||||
|
|
||||||
fmt.Fprintln(w, "# HELP linearvoid_chat_build_info Static build info for chat service.")
|
fmt.Fprintln(w, "# HELP linearvoid_chat_build_info Static build info for chat service.")
|
||||||
fmt.Fprintln(w, "# TYPE linearvoid_chat_build_info gauge")
|
fmt.Fprintln(w, "# TYPE linearvoid_chat_build_info gauge")
|
||||||
fmt.Fprintln(w, "linearvoid_chat_build_info{service=\"chat\",database_driver=\""+databaseDriverName+"\"} 1")
|
fmt.Fprintln(w, "linearvoid_chat_build_info{service=\"chat\",database_driver=\""+databaseDriverName+"\",instance_id=\""+instanceID+"\"} 1")
|
||||||
}
|
}
|
||||||
|
|
||||||
// broadcast sends a message to all connected clients in a room and saves to DB
|
// broadcast sends a message to all connected clients in a room and saves to DB
|
||||||
@@ -2172,12 +2181,23 @@ func broadcast(msg Message) {
|
|||||||
// Persist to configured SQL backend.
|
// Persist to configured SQL backend.
|
||||||
db.Create(&msg)
|
db.Create(&msg)
|
||||||
|
|
||||||
|
// Local fan-out
|
||||||
for client := range clients {
|
for client := range clients {
|
||||||
if err := client.WriteJSON(msg); err != nil {
|
if err := client.WriteJSON(msg); err != nil {
|
||||||
client.Close()
|
client.Close()
|
||||||
delete(clients, client)
|
delete(clients, client)
|
||||||
|
delete(clientMeta, client)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Redis pub/sub for multi-instance fan-out
|
||||||
|
if redisClient != nil && msg.Room != "" {
|
||||||
|
payload, _ := json.Marshal(map[string]any{
|
||||||
|
"instance": instanceID,
|
||||||
|
"msg": msg,
|
||||||
|
})
|
||||||
|
redisClient.Publish(context.Background(), fmt.Sprintf("room:%s:messages", msg.Room), payload)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// handleConnection handles WebSocket connections for chat (public rooms, private DMs, moderation)
|
// handleConnection handles WebSocket connections for chat (public rooms, private DMs, moderation)
|
||||||
@@ -2333,6 +2353,47 @@ func pushChatHistory(conn *websocket.Conn, room string, limit int) {
|
|||||||
// dmsSent tracks total private messages for Prometheus metrics (set from handleMetrics).
|
// dmsSent tracks total private messages for Prometheus metrics (set from handleMetrics).
|
||||||
var dmsSent int64
|
var dmsSent int64
|
||||||
|
|
||||||
|
// subscribeRoomMessages listens for Redis pub/sub messages and fans out locally.
|
||||||
|
func subscribeRoomMessages() {
|
||||||
|
if redisClient == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
pubsub := redisClient.PSubscribe(context.Background(), "room:*:messages")
|
||||||
|
defer pubsub.Close()
|
||||||
|
|
||||||
|
log.Printf("Redis pub/sub subscriber started (instance %s)", instanceID)
|
||||||
|
|
||||||
|
for {
|
||||||
|
msg, err := pubsub.ReceiveMessage(context.Background())
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("Redis pub/sub error: %v — retrying in 5s", err)
|
||||||
|
time.Sleep(5 * time.Second)
|
||||||
|
pubsub = redisClient.PSubscribe(context.Background(), "room:*:messages")
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
var envelope struct {
|
||||||
|
Instance string `json:"instance"`
|
||||||
|
Msg Message `json:"msg"`
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal([]byte(msg.Payload), &envelope); err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
// Skip messages originating from this instance
|
||||||
|
if envelope.Instance == instanceID {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
// Fan out to local clients
|
||||||
|
mutex.Lock()
|
||||||
|
for client := range clients {
|
||||||
|
client.WriteJSON(envelope.Msg)
|
||||||
|
}
|
||||||
|
mutex.Unlock()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// handlePrivateDM routes private messages to the specific user pair
|
// handlePrivateDM routes private messages to the specific user pair
|
||||||
func handlePrivateDM(msg Message) {
|
func handlePrivateDM(msg Message) {
|
||||||
msg.Type = "private"
|
msg.Type = "private"
|
||||||
@@ -2640,6 +2701,11 @@ func main() {
|
|||||||
initRedis()
|
initRedis()
|
||||||
initDB()
|
initDB()
|
||||||
|
|
||||||
|
// Subscribe to Redis pub/sub for multi-instance message fan-out
|
||||||
|
go subscribeRoomMessages()
|
||||||
|
|
||||||
|
log.Printf("Instance ID: %s", instanceID)
|
||||||
|
|
||||||
http.HandleFunc("/ws", handleConnection)
|
http.HandleFunc("/ws", handleConnection)
|
||||||
http.HandleFunc("/api/admin/login", handleAdminLogin)
|
http.HandleFunc("/api/admin/login", handleAdminLogin)
|
||||||
http.HandleFunc("/api/admin/logout", handleAdminLogout)
|
http.HandleFunc("/api/admin/logout", handleAdminLogout)
|
||||||
|
|||||||
Reference in new issue
Block a user