haopengzhan
ebcaa5d72a
- Implement atomic state transitions: add EntryService.PatchEntry() with EntryDiff computation and EventEntryTransition event emission. - Add persistent SQLite task queue (JobEngine): add jobs table with crash recovery, exponential backoff retries, concurrency control, and idempotency key deduplication. - Decompose monolithic HTTP handlers: split handlers.go into domain handlers (auth, feeds, entries, jobs, settings, rules, sse). - Implement real-time Server-Sent Events (/api/events/stream) for multi-tab/device state synchronization. - Fix SQLite concurrency: apply busy_timeout(10000) and journal_mode(WAL) via DSN query parameters to resolve SQLITE_BUSY (5). - Fix AI settings & prompt loading: unmarshal JSON from ai_settings, apply configured prompt templates, and enforce strict enabled/API key validations. - Fix undo share state: remove omitempty from ShareUUID to serialize null for state resets. - Fix middle list counter: compute true unread/total stats from DB and clean switcher button labels. - Adapt mobile UI for iOS: add viewport-fit=cover in index.html and calibrate safe-area insets for the bottom navigation and article reader toolbar. - Add architectural documentation and worklog in docs/009_worklog.md.
105 lines
2.2 KiB
Go
105 lines
2.2 KiB
Go
package api
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
type SSEBroadcaster struct {
|
|
mu sync.RWMutex
|
|
clients map[chan []byte]struct{}
|
|
}
|
|
|
|
func NewSSEBroadcaster() *SSEBroadcaster {
|
|
return &SSEBroadcaster{
|
|
clients: make(map[chan []byte]struct{}),
|
|
}
|
|
}
|
|
|
|
func (b *SSEBroadcaster) Subscribe() chan []byte {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
ch := make(chan []byte, 32)
|
|
b.clients[ch] = struct{}{}
|
|
return ch
|
|
}
|
|
|
|
func (b *SSEBroadcaster) Unsubscribe(ch chan []byte) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if _, ok := b.clients[ch]; ok {
|
|
delete(b.clients, ch)
|
|
close(ch)
|
|
}
|
|
}
|
|
|
|
func (b *SSEBroadcaster) Broadcast(event string, data interface{}) {
|
|
payload, err := json.Marshal(data)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
msg := fmt.Sprintf("event: %s\ndata: %s\n\n", event, string(payload))
|
|
msgBytes := []byte(msg)
|
|
|
|
b.mu.RLock()
|
|
defer b.mu.RUnlock()
|
|
for ch := range b.clients {
|
|
select {
|
|
case ch <- msgBytes:
|
|
default:
|
|
// Non-blocking drop if client buffer is full
|
|
}
|
|
}
|
|
}
|
|
|
|
// StreamEvents handles GET /api/events/stream
|
|
func (h *Handler) StreamEvents(w http.ResponseWriter, r *http.Request) {
|
|
flusher, ok := w.(http.Flusher)
|
|
if !ok {
|
|
http.Error(w, "Streaming unsupported", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "text/event-stream")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
w.Header().Set("Connection", "keep-alive")
|
|
w.Header().Set("X-Accel-Buffering", "no")
|
|
w.WriteHeader(http.StatusOK)
|
|
|
|
// Send initial connection packet
|
|
initialMsg := fmt.Sprintf("event: connected\ndata: {\"status\":\"connected\",\"time\":\"%s\"}\n\n", time.Now().UTC().Format(time.RFC3339))
|
|
_, _ = w.Write([]byte(initialMsg))
|
|
flusher.Flush()
|
|
|
|
clientCh := h.broadcaster.Subscribe()
|
|
defer h.broadcaster.Unsubscribe(clientCh)
|
|
|
|
heartbeat := time.NewTicker(25 * time.Second)
|
|
defer heartbeat.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-r.Context().Done():
|
|
return
|
|
case <-heartbeat.C:
|
|
pingMsg := fmt.Sprintf("event: ping\ndata: {\"time\":\"%s\"}\n\n", time.Now().UTC().Format(time.RFC3339))
|
|
if _, err := w.Write([]byte(pingMsg)); err != nil {
|
|
return
|
|
}
|
|
flusher.Flush()
|
|
case msg, ok := <-clientCh:
|
|
if !ok {
|
|
return
|
|
}
|
|
if _, err := w.Write(msg); err != nil {
|
|
return
|
|
}
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|