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.
102 lines
2.6 KiB
Go
102 lines
2.6 KiB
Go
package events
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
|
|
"git.pengzhan.dev/haopengzhan/rsspipe/internal/logger"
|
|
"git.pengzhan.dev/haopengzhan/rsspipe/internal/models"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
type EventType string
|
|
|
|
const (
|
|
EventPostCreated EventType = "on_post_created"
|
|
EventPostUpdated EventType = "on_post_updated"
|
|
EventVariantUpdated EventType = "on_variant_updated"
|
|
EventEntryTransition EventType = "on_entry_transition"
|
|
EventManualTrigger EventType = "manual"
|
|
EventJobProgress EventType = "job_progress"
|
|
EventJobCompleted EventType = "job_completed"
|
|
EventJobFailed EventType = "job_failed"
|
|
EventPreferencesSync EventType = "preferences_updated"
|
|
)
|
|
|
|
type Event struct {
|
|
Type EventType
|
|
Entry models.FeedEntry
|
|
OldEntry *models.FeedEntry
|
|
NewEntry *models.FeedEntry
|
|
Diff *models.EntryDiff
|
|
Job *models.Job
|
|
VariantKey string // populated when Type == EventVariantUpdated
|
|
Depth int
|
|
DirectAttrs map[string]interface{}
|
|
}
|
|
|
|
type EventHandler func(ctx context.Context, evt Event) error
|
|
|
|
type EventBus struct {
|
|
mu sync.RWMutex
|
|
subscribers map[EventType][]EventHandler
|
|
log *zap.Logger
|
|
maxDepth int
|
|
}
|
|
|
|
func NewEventBus() *EventBus {
|
|
return &EventBus{
|
|
subscribers: make(map[EventType][]EventHandler),
|
|
log: logger.Named("eventbus"),
|
|
maxDepth: 5,
|
|
}
|
|
}
|
|
|
|
func (b *EventBus) Subscribe(eventType EventType, handler EventHandler) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.subscribers[eventType] = append(b.subscribers[eventType], handler)
|
|
}
|
|
|
|
func (b *EventBus) Publish(ctx context.Context, evt Event) {
|
|
if evt.Depth > b.maxDepth {
|
|
b.log.Warn("Event dropped: recursion depth limit exceeded",
|
|
zap.String("type", string(evt.Type)),
|
|
zap.String("entry_id", evt.Entry.ID),
|
|
zap.Int("depth", evt.Depth),
|
|
)
|
|
return
|
|
}
|
|
|
|
b.mu.RLock()
|
|
handlers := append([]EventHandler{}, b.subscribers[evt.Type]...)
|
|
wildcardHandlers := append([]EventHandler{}, b.subscribers["*"]...)
|
|
b.mu.RUnlock()
|
|
|
|
b.log.Debug("Event published",
|
|
zap.String("type", string(evt.Type)),
|
|
zap.String("entry_id", evt.Entry.ID),
|
|
zap.String("variant_key", evt.VariantKey),
|
|
zap.Int("depth", evt.Depth),
|
|
zap.Int("subscriber_count", len(handlers)+len(wildcardHandlers)),
|
|
)
|
|
|
|
allHandlers := append(handlers, wildcardHandlers...)
|
|
for _, h := range allHandlers {
|
|
go func(fn EventHandler) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
b.log.Error("Event handler panic recovered", zap.Any("panic", r))
|
|
}
|
|
}()
|
|
if err := fn(ctx, evt); err != nil {
|
|
b.log.Error("Event handler error",
|
|
zap.String("type", string(evt.Type)),
|
|
zap.String("entry_id", evt.Entry.ID),
|
|
zap.Error(err),
|
|
)
|
|
}
|
|
}(h)
|
|
}
|
|
}
|