Sessions now update live across all users and devices

PROGRESS.md

@@ -72,13 +72,13 @@
 24. **Drag-and-Drop Stability:** Fixed a bug where tasks would "disappear" if dropped in an invalid area. Tasks now remain visible at their original position if a drop is cancelled.
 25. **User Deletion Safety:** Implemented a backend check to prevent deleting users who have assigned tasks or notes. Upgraded the frontend `ApiService` to correctly parse and display these descriptive error messages from the backend.
 26. **Enhanced User Management:** Admins can now manage full team profiles (Name, Role, Email, and Phone) during both user creation and inline editing in the Workspace settings.
+27. **Real-time Updates (SSE):** Implemented a global real-time notification system using Server-Sent Events (SSE). Any change made by one user (moving tasks, adding notes, updating profiles) is now instantly broadcast to all other connected clients. Switched from WebSockets to SSE to ensure 100% compatibility with Nginx and reverse proxies without extra configuration.
 
 ### Phase 3: Advanced Features
-- **Real-time Notifications:** Explore WebSockets for task assignments.
 - **iMessage Integration:** Develop the "Molty" bridge for phone-to-task creation.
 - **File Uploads:** Support for attaching photos/documents to tasks.
 
 ---
 
 **Last Updated:** Wednesday, May 13, 2026
-**Status:** Phase 2 Complete / Ready for Phase 3
+**Status:** Phase 3 in Progress / Real-time Updates Active

api.js

@@ -3,6 +3,28 @@ class ApiService {
     this.baseUrl = '/api';
     this.token = localStorage.getItem('dashy_token');
     this.subscribers = new Set();
+    this.connectSSE();
+  }
+
+  connectSSE() {
+    const url = `${this.baseUrl}/stream`;
+    this.sse = new EventSource(url);
+
+    this.sse.onmessage = (event) => {
+      try {
+        const data = JSON.parse(event.data);
+        if (data.type === 'refresh') {
+          this.notify();
+        }
+      } catch (e) {
+        // Fallback for non-JSON messages or just generic refresh
+        this.notify();
+      }
+    };
+
+    this.sse.onerror = (err) => {
+      console.error('SSE connection error. EventSource will auto-reconnect.', err);
+    };
   }
 
   subscribe(fn) {

backend/main.py

@@ -1,15 +1,37 @@
-from fastapi import FastAPI, Depends, HTTPException, status
+from fastapi import FastAPI, Depends, HTTPException, status, Request
+from fastapi.responses import StreamingResponse
 from fastapi.middleware.cors import CORSMiddleware
 from sqlalchemy.orm import Session
 from sqlalchemy.sql import func
-from typing import List
+from typing import List, Dict
 import uuid
+import json
+import asyncio
 
 from . import models, schemas, auth, database
 from .database import engine, get_db
 
 models.Base.metadata.create_all(bind=engine)
 
+class EventNotifier:
+    def __init__(self):
+        self.queues = []
+
+    def subscribe(self):
+        q = asyncio.Queue()
+        self.queues.append(q)
+        return q
+
+    def unsubscribe(self, q):
+        if q in self.queues:
+            self.queues.remove(q)
+
+    async def broadcast(self, message: str):
+        for q in self.queues:
+            await q.put(message)
+
+manager = EventNotifier()
+
 app = FastAPI(title="Dashy API")
 
 app.add_middleware(
@@ -20,6 +42,26 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+@app.get("/stream")
+async def message_stream(request: Request):
+    async def event_generator():
+        q = manager.subscribe()
+        try:
+            while True:
+                if await request.is_disconnected():
+                    break
+                # use a timeout so we can periodically check for disconnects
+                try:
+                    msg = await asyncio.wait_for(q.get(), timeout=2.0)
+                    yield f"data: {msg}\n\n"
+                except asyncio.TimeoutError:
+                    # just a keep-alive ping
+                    yield ": keepalive\n\n"
+        finally:
+            manager.unsubscribe(q)
+            
+    return StreamingResponse(event_generator(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "Connection": "keep-alive"})
+
 @app.post("/token", response_model=schemas.Token)
 async def login_for_access_token(form_data: schemas.UserLogin, db: Session = Depends(get_db)):
     # Search by ID or Name
@@ -42,7 +84,7 @@ def read_users(db: Session = Depends(get_db)):
     return db.query(models.User).all()
 
 @app.post("/users", response_model=schemas.User)
-def create_user(user: schemas.UserCreate, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
+async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
     db_user = models.User(
         id=user.id,
         name=user.name,
@@ -57,10 +99,11 @@ def create_user(user: schemas.UserCreate, db: Session = Depends(get_db), current
     db.add(db_user)
     db.commit()
     db.refresh(db_user)
+    await manager.broadcast(json.dumps({"type": "refresh"}))
     return db_user
 
 @app.patch("/users/{user_id}", response_model=schemas.User)
-def update_user(user_id: str, user_update: schemas.UserUpdate, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
+async def update_user(user_id: str, user_update: schemas.UserUpdate, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
     if current_user.account_type != "admin" and current_user.id != user_id:
         raise HTTPException(status_code=403, detail="Not enough permissions")
     
@@ -74,10 +117,11 @@ def update_user(user_id: str, user_update: schemas.UserUpdate, db: Session = Dep
     
     db.commit()
     db.refresh(db_user)
+    await manager.broadcast(json.dumps({"type": "refresh"}))
     return db_user
 
 @app.post("/users/{user_id}/password")
-def change_password(user_id: str, pwd_data: schemas.PasswordChange, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
+async def change_password(user_id: str, pwd_data: schemas.PasswordChange, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
     if current_user.id != user_id:
         raise HTTPException(status_code=403, detail="Cannot change another user's password")
     
@@ -86,10 +130,11 @@ def change_password(user_id: str, pwd_data: schemas.PasswordChange, db: Session
     
     current_user.password_hash = auth.get_password_hash(pwd_data.new_password)
     db.commit()
+    await manager.broadcast(json.dumps({"type": "refresh"}))
     return {"message": "Password updated successfully"}
 
 @app.delete("/users/{user_id}")
-def delete_user(user_id: str, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
+async def delete_user(user_id: str, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
     if current_user.account_type != "admin":
         raise HTTPException(status_code=403, detail="Not enough permissions")
     
@@ -112,6 +157,7 @@ def delete_user(user_id: str, db: Session = Depends(get_db), current_user: model
     
     db.delete(db_user)
     db.commit()
+    await manager.broadcast(json.dumps({"type": "refresh"}))
     return {"message": "User deleted"}
 
 @app.get("/tasks", response_model=List[schemas.Task])
@@ -125,7 +171,7 @@ def read_deleted_tasks(db: Session = Depends(get_db), current_user: models.User
     return db.query(models.Task).filter(models.Task.deleted_at != None).order_by(models.Task.position.asc()).all()
 
 @app.post("/tasks", response_model=schemas.Task)
-def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
+async def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
     task_id = task.id or f"t_{uuid.uuid4().hex[:8]}"
     
     # Calculate position (max in column + 1000)
@@ -155,10 +201,11 @@ def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db), current
     
     db.commit()
     db.refresh(db_task)
+    await manager.broadcast(json.dumps({"type": "refresh"}))
     return db_task
 
 @app.patch("/tasks/{task_id}", response_model=schemas.Task)
-def update_task(task_id: str, task_update: schemas.TaskUpdate, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
+async def update_task(task_id: str, task_update: schemas.TaskUpdate, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
     db_task = db.query(models.Task).filter(models.Task.id == task_id).first()
     if not db_task:
         raise HTTPException(status_code=404, detail="Task not found")
@@ -169,16 +216,18 @@ def update_task(task_id: str, task_update: schemas.TaskUpdate, db: Session = Dep
 
     db.commit()
     db.refresh(db_task)
+    await manager.broadcast(json.dumps({"type": "refresh"}))
     return db_task
 
 @app.delete("/tasks/{task_id}")
-def delete_task(task_id: str, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
+async def delete_task(task_id: str, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
     db_task = db.query(models.Task).filter(models.Task.id == task_id).first()
     if not db_task:
         raise HTTPException(status_code=404, detail="Task not found")
 
     db_task.deleted_at = func.now()
     db.commit()
+    await manager.broadcast(json.dumps({"type": "refresh"}))
     return {"message": "Task moved to trash"}
 
 @app.get("/tasks/{task_id}/notes", response_model=List[schemas.TaskNote])
@@ -186,7 +235,7 @@ def read_task_notes(task_id: str, db: Session = Depends(get_db), current_user: m
     return db.query(models.TaskNote).filter(models.TaskNote.task_id == task_id).order_by(models.TaskNote.created_at.desc()).all()
 
 @app.post("/tasks/{task_id}/notes", response_model=schemas.TaskNote)
-def create_task_note(task_id: str, note: schemas.TaskNoteBase, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
+async def create_task_note(task_id: str, note: schemas.TaskNoteBase, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
     db_note = models.TaskNote(
         task_id=task_id,
         author_id=current_user.id,
@@ -195,10 +244,11 @@ def create_task_note(task_id: str, note: schemas.TaskNoteBase, db: Session = Dep
     db.add(db_note)
     db.commit()
     db.refresh(db_note)
+    await manager.broadcast(json.dumps({"type": "refresh"}))
     return db_note
 
 @app.post("/tasks/{task_id}/restore", response_model=schemas.Task)
-def restore_task(task_id: str, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
+async def restore_task(task_id: str, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
     if current_user.account_type != "admin":
         raise HTTPException(status_code=403, detail="Not enough permissions")
     
@@ -209,20 +259,22 @@ def restore_task(task_id: str, db: Session = Depends(get_db), current_user: mode
     db_task.deleted_at = None
     db.commit()
     db.refresh(db_task)
+    await manager.broadcast(json.dumps({"type": "refresh"}))
     return db_task
 
 @app.get("/workspace", response_model=schemas.Workspace)
-def read_workspace(db: Session = Depends(get_db)):
+async def read_workspace(db: Session = Depends(get_db)):
     ws = db.query(models.Workspace).first()
     if not ws:
         ws = models.Workspace(id="default", name="murchison-auto", timezone="Pacific/Auckland")
         db.add(ws)
         db.commit()
         db.refresh(ws)
+        await manager.broadcast(json.dumps({"type": "refresh"}))
     return ws
 
 @app.patch("/workspace", response_model=schemas.Workspace)
-def update_workspace(ws_update: schemas.WorkspaceUpdate, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
+async def update_workspace(ws_update: schemas.WorkspaceUpdate, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
     if current_user.account_type != "admin":
         raise HTTPException(status_code=403, detail="Not enough permissions")
     
@@ -237,6 +289,7 @@ def update_workspace(ws_update: schemas.WorkspaceUpdate, db: Session = Depends(g
     
     db.commit()
     db.refresh(ws)
+    await manager.broadcast(json.dumps({"type": "refresh"}))
     return ws
 
 @app.get("/audit", response_model=List[schemas.AuditLog])
@@ -244,7 +297,7 @@ def read_audit(db: Session = Depends(get_db), current_user: models.User = Depend
     return db.query(models.AuditLog).order_by(models.AuditLog.at.desc()).all()
 
 @app.post("/audit", response_model=schemas.AuditLog)
-def create_audit(audit: schemas.AuditLogBase, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
+async def create_audit(audit: schemas.AuditLogBase, db: Session = Depends(get_db), current_user: models.User = Depends(auth.get_current_user)):
     audit_id = f"a_{uuid.uuid4().hex[:8]}"
     db_audit = models.AuditLog(
         id=audit_id,
@@ -256,4 +309,5 @@ def create_audit(audit: schemas.AuditLogBase, db: Session = Depends(get_db), cur
     db.add(db_audit)
     db.commit()
     db.refresh(db_audit)
+    await manager.broadcast(json.dumps({"type": "refresh"}))
     return db_audit

plumbing-dashy.service

@@ -0,0 +1,16 @@
+[Unit]
+Description=Plumbing Dashy Uvicorn API
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+WorkingDirectory=/opt/plumbing-dashy
+Environment=DASHY_DB_PATH=/opt/plumbing-dashy/dashy.db
+ExecStart=/opt/plumbing-dashy/venv/bin/uvicorn backend.main:app --host 127.0.0.1 --port 24024
+Restart=on-failure
+RestartSec=5s
+TimeoutStopSec=30s
+
+[Install]
+WantedBy=default.target