cleaning of the files event system
This commit is contained in:
Pierre
+38
-186
@@ -1,223 +1,75 @@
|
||||
package watcher
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/metazla/meta-core/internal/storage"
|
||||
)
|
||||
|
||||
const (
|
||||
// MaxRetries is the maximum number of delivery retries
|
||||
MaxRetries = 3
|
||||
// RetryDelay is the delay between retries
|
||||
RetryDelay = 5 * time.Second
|
||||
// WebhookTimeout is the timeout for webhook requests
|
||||
WebhookTimeout = 10 * time.Second
|
||||
// MaxFailCount is the max failures before removing subscriber
|
||||
MaxFailCount = 10
|
||||
// EventsStream is the Redis Stream name for file events
|
||||
EventsStream = "file:events"
|
||||
// StreamMaxLen is the approximate max length for the stream
|
||||
StreamMaxLen = 10000
|
||||
)
|
||||
|
||||
// Dispatcher sends file events to subscribers
|
||||
// Dispatcher sends file events to Redis Stream
|
||||
type Dispatcher struct {
|
||||
subscribers map[string]*Subscriber
|
||||
sseClients map[chan FileEvent]bool
|
||||
mu sync.RWMutex
|
||||
httpClient *http.Client
|
||||
mu sync.RWMutex
|
||||
storageClient *storage.Client
|
||||
}
|
||||
|
||||
// NewDispatcher creates a new event dispatcher
|
||||
func NewDispatcher() *Dispatcher {
|
||||
return &Dispatcher{
|
||||
subscribers: make(map[string]*Subscriber),
|
||||
sseClients: make(map[chan FileEvent]bool),
|
||||
httpClient: &http.Client{
|
||||
Timeout: WebhookTimeout,
|
||||
},
|
||||
}
|
||||
return &Dispatcher{}
|
||||
}
|
||||
|
||||
// Subscribe registers a webhook subscriber
|
||||
func (d *Dispatcher) Subscribe(url string, eventTypes []string) error {
|
||||
// SetStorageClient sets the storage client for Redis Stream publishing
|
||||
func (d *Dispatcher) SetStorageClient(client *storage.Client) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
|
||||
d.subscribers[url] = &Subscriber{
|
||||
URL: url,
|
||||
RegisteredAt: NowMS(),
|
||||
EventTypes: eventTypes,
|
||||
FailCount: 0,
|
||||
}
|
||||
|
||||
log.Printf("[Dispatcher] Subscribed webhook: %s", url)
|
||||
return nil
|
||||
d.storageClient = client
|
||||
log.Println("[Dispatcher] Storage client set for Redis Stream publishing")
|
||||
}
|
||||
|
||||
// Unsubscribe removes a webhook subscriber
|
||||
func (d *Dispatcher) Unsubscribe(url string) error {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
|
||||
delete(d.subscribers, url)
|
||||
log.Printf("[Dispatcher] Unsubscribed webhook: %s", url)
|
||||
return nil
|
||||
}
|
||||
|
||||
// ListSubscribers returns all subscribers
|
||||
func (d *Dispatcher) ListSubscribers() []Subscriber {
|
||||
d.mu.RLock()
|
||||
defer d.mu.RUnlock()
|
||||
|
||||
result := make([]Subscriber, 0, len(d.subscribers))
|
||||
for _, sub := range d.subscribers {
|
||||
result = append(result, *sub)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// AddSSEClient adds an SSE client channel
|
||||
func (d *Dispatcher) AddSSEClient(ch chan FileEvent) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
d.sseClients[ch] = true
|
||||
}
|
||||
|
||||
// RemoveSSEClient removes an SSE client channel
|
||||
func (d *Dispatcher) RemoveSSEClient(ch chan FileEvent) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
delete(d.sseClients, ch)
|
||||
close(ch)
|
||||
}
|
||||
|
||||
// Dispatch sends an event to all subscribers
|
||||
// Dispatch sends an event to Redis Stream
|
||||
func (d *Dispatcher) Dispatch(event FileEvent) {
|
||||
// Dispatch to webhooks
|
||||
go d.dispatchToWebhooks(event)
|
||||
|
||||
// Dispatch to SSE clients
|
||||
d.dispatchToSSE(event)
|
||||
go d.dispatchToStream(event)
|
||||
}
|
||||
|
||||
// dispatchToWebhooks sends event to webhook subscribers
|
||||
func (d *Dispatcher) dispatchToWebhooks(event FileEvent) {
|
||||
// dispatchToStream publishes event to Redis Stream for service consumption
|
||||
func (d *Dispatcher) dispatchToStream(event FileEvent) {
|
||||
d.mu.RLock()
|
||||
subscribers := make([]*Subscriber, 0, len(d.subscribers))
|
||||
for _, sub := range d.subscribers {
|
||||
subscribers = append(subscribers, sub)
|
||||
}
|
||||
client := d.storageClient
|
||||
d.mu.RUnlock()
|
||||
|
||||
for _, sub := range subscribers {
|
||||
// Check event type filter
|
||||
if len(sub.EventTypes) > 0 {
|
||||
found := false
|
||||
for _, et := range sub.EventTypes {
|
||||
if et == string(event.Type) {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
go d.deliverToWebhook(sub.URL, event)
|
||||
}
|
||||
}
|
||||
|
||||
// deliverToWebhook delivers an event to a webhook with retries
|
||||
func (d *Dispatcher) deliverToWebhook(url string, event FileEvent) {
|
||||
body, err := json.Marshal(event)
|
||||
if err != nil {
|
||||
log.Printf("[Dispatcher] Failed to marshal event: %v", err)
|
||||
if client == nil || !client.IsConnected() {
|
||||
return
|
||||
}
|
||||
|
||||
var lastErr error
|
||||
for attempt := 0; attempt < MaxRetries; attempt++ {
|
||||
if attempt > 0 {
|
||||
time.Sleep(RetryDelay)
|
||||
}
|
||||
|
||||
req, err := http.NewRequest("POST", url, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
lastErr = err
|
||||
continue
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("X-Event-Type", string(event.Type))
|
||||
|
||||
resp, err := d.httpClient.Do(req)
|
||||
if err != nil {
|
||||
lastErr = err
|
||||
continue
|
||||
}
|
||||
|
||||
resp.Body.Close()
|
||||
|
||||
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
|
||||
// Success
|
||||
d.mu.Lock()
|
||||
if sub, ok := d.subscribers[url]; ok {
|
||||
sub.LastDelivery = NowMS()
|
||||
sub.FailCount = 0
|
||||
}
|
||||
d.mu.Unlock()
|
||||
return
|
||||
}
|
||||
|
||||
lastErr = &httpError{StatusCode: resp.StatusCode}
|
||||
// Build stream fields
|
||||
fields := map[string]interface{}{
|
||||
"type": string(event.Type),
|
||||
"path": event.Path,
|
||||
"timestamp": fmt.Sprintf("%d", event.Timestamp),
|
||||
}
|
||||
|
||||
// All retries failed
|
||||
log.Printf("[Dispatcher] Failed to deliver to %s after %d retries: %v", url, MaxRetries, lastErr)
|
||||
|
||||
d.mu.Lock()
|
||||
if sub, ok := d.subscribers[url]; ok {
|
||||
sub.FailCount++
|
||||
if sub.FailCount >= MaxFailCount {
|
||||
delete(d.subscribers, url)
|
||||
log.Printf("[Dispatcher] Removed webhook %s after %d failures", url, MaxFailCount)
|
||||
}
|
||||
// Add optional fields
|
||||
if event.Size > 0 {
|
||||
fields["size"] = fmt.Sprintf("%d", event.Size)
|
||||
}
|
||||
d.mu.Unlock()
|
||||
}
|
||||
|
||||
// dispatchToSSE sends event to SSE clients
|
||||
func (d *Dispatcher) dispatchToSSE(event FileEvent) {
|
||||
d.mu.RLock()
|
||||
clients := make([]chan FileEvent, 0, len(d.sseClients))
|
||||
for ch := range d.sseClients {
|
||||
clients = append(clients, ch)
|
||||
if event.PartialHash != "" {
|
||||
fields["partialHash"] = event.PartialHash
|
||||
}
|
||||
if event.OldPath != "" {
|
||||
fields["oldPath"] = event.OldPath
|
||||
}
|
||||
d.mu.RUnlock()
|
||||
|
||||
for _, ch := range clients {
|
||||
select {
|
||||
case ch <- event:
|
||||
default:
|
||||
// Channel full, skip
|
||||
}
|
||||
// Publish to stream
|
||||
_, err := client.XAdd(EventsStream, StreamMaxLen, fields)
|
||||
if err != nil {
|
||||
log.Printf("[Dispatcher] Failed to publish to stream: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// SSEClientCount returns the number of connected SSE clients
|
||||
func (d *Dispatcher) SSEClientCount() int {
|
||||
d.mu.RLock()
|
||||
defer d.mu.RUnlock()
|
||||
return len(d.sseClients)
|
||||
}
|
||||
|
||||
// httpError represents an HTTP error
|
||||
type httpError struct {
|
||||
StatusCode int
|
||||
}
|
||||
|
||||
func (e *httpError) Error() string {
|
||||
return http.StatusText(e.StatusCode)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user