yao/event/README.md

# event — Yao In-Process Event Bus

Global event service for async/sync event routing, serial queue processing, and real-time subscriptions. All operations are goroutine-safe.

## Import

```go
import (
    "github.com/yaoapp/yao/event"
    "github.com/yaoapp/yao/event/types"
)
```

## Core Concepts

| Concept | Description |
|---|---|
| **Push** | Async fire-and-forget delivery. Returns event ID immediately. |
| **Call** | Sync request-response. Blocks until handler writes to `resp`. |
| **Handler** | One per prefix (e.g. `"trace"`). Processes `Push` and `Call` events. |
| **Queue** | FIFO serial processing per entity (e.g. per traceID). Events in same queue never run concurrently. |
| **Listener** | Persistent background consumer (registered at startup). Gets a copy of every matching event. |
| **Subscriber** | Dynamic subscription (e.g. SSE/WebSocket). Non-blocking; skips if channel full. |

## Lifecycle

```go
// 1. Register handlers and listeners (before Start, typically in init())
event.Register("trace", traceHandler, event.MaxWorkers(512), event.ReservedWorkers(20))
event.Register("job", jobHandler)
event.Listen("trace.*", traceListener)

// 2. Start
event.Start()

// 3. Use (from any goroutine)
event.Push(ctx, "trace.add", payload, event.Queue(traceQueueID))
id, data, err := event.Call(ctx, "trace.get", req, event.Queue(traceQueueID))

// 4. Stop (during shutdown)
event.Stop(ctx)
```

## Handler

Implement `types.Handler`:

```go
type TraceHandler struct{}

func (h *TraceHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) {
    var p TracePayload
    if err := ev.Should(&p); err != nil {
        if ev.IsCall { resp <- types.Result{Err: err} }
        return
    }
    // ... business logic ...
    if ev.IsCall {
        resp <- types.Result{Data: result}
    }
}

func (h *TraceHandler) Shutdown(ctx context.Context) error { return nil }
```

- `ctx`: non-cancellable for Push; caller's context for Call.
- `resp`: always non-nil. Write exactly once for Call; ignore for Push.
- `ev.Should(&target)`: type-safe payload extraction.
- Panics are recovered automatically; `ErrHandlerPanic` is returned to Call.

## Queue

```go
queueID, err := event.QueueCreate("trace")           // auto-generated ID
queueID, err := event.QueueCreate("trace", "my-id")  // custom ID

event.Push(ctx, "trace.add", data, event.Queue(queueID))    // serial
event.Call(ctx, "trace.get", req, event.Queue(queueID))      // serial, same queue

event.QueueRelease(queueID)  // graceful: drain pending, reject new
event.QueueAbort(queueID)    // forceful: discard pending, reject new
```

## Listener

Implement `types.Listener`:

```go
type MailListener struct{}
func (l *MailListener) OnEvent(ev *types.Event) { /* ... */ }
func (l *MailListener) Shutdown(ctx context.Context) error { return nil }

// Register before Start
event.Listen("mail.*", &MailListener{}, event.Filter(fn), event.BufferSize(4096))
```

- Each listener runs in its own goroutine.
- Non-blocking: if buffer full, event is skipped (logged as warning).

## Subscriber

```go
ch := make(chan *types.Event, 256)
subID := event.Subscribe("trace.*", ch, event.Filter(fn))
defer event.Unsubscribe(subID)

for ev := range ch {
    // push to SSE / WebSocket
}
```

- Non-blocking: if `ch` full, event is skipped silently.
- Call `Unsubscribe` when client disconnects.

## Context Propagation

```go
ctx = event.WithSID(ctx, sessionID)
ctx = event.WithAuth(ctx, &types.AuthorizedInfo{UserID: "u-1"})

// Inside handler:
sid := ev.SID
auth := ev.Auth  // may be nil
```

SID and Auth are extracted from `ctx` automatically when calling `Push`/`Call`.

## Pattern Matching

Used by `Listen` and `Subscribe`:

| Pattern | Matches |
|---|---|
| `"*"` | Everything |
| `"trace.*"` | `"trace.add"`, `"trace.get"`, etc. |
| `"trace.add"` | Exact match only |

## Handler Options

| Option | Default | Description |
|---|---|---|
| `MaxWorkers(n)` | 512 | Max concurrent goroutines for this handler |
| `ReservedWorkers(n)` | 10 | Slots reserved for Call (Push can use Max−Reserved) |
| `QueueSize(n)` | 8192 | Per-queue buffered channel capacity |

## Errors

| Error | When |
|---|---|
| `ErrNotStarted` | Push/Call before Start or after Stop |
| `ErrNoHandler` | No handler registered for event prefix |
| `ErrQueueFull` | Queue buffer at capacity |
| `ErrQueueNotFound` | Queue ID never created |
| `ErrQueueReleased` | Queue already released/aborted |
| `ErrQueueExists` | QueueCreate with duplicate ID |
| `ErrHandlerPanic` | Handler panicked (recovered) |

## Performance (M2 Max, 12 cores)

| Metric | Value |
|---|---|
| Push (no queue) | ~860K ops/sec, 456 B/op |
| Call (no queue) | ~1.2M ops/sec, 440 B/op |
| Push (with queue) | ~2.9M ops/sec, 341 B/op |
| 1000-user scenario (2000 queues, 27K events) | ~100K events/sec, 280ms total |
| Steady-state memory (1000 users) | ~27 MB |
| Goroutine leaks | Zero |

## File Structure

```
event/
├── types/
│   ├── types.go        # Event, Result, HandlerEntry, FilterEntry, options
│   └── interfaces.go   # Handler, Listener interfaces
├── service.go          # Register, Start, Stop, Reload, global state
├── bus.go              # Push, Call, QueueCreate/Release/Abort
├── queue.go            # FIFO queue + queue manager
├── worker.go           # Worker pool (two-tier semaphore)
├── listener.go         # Listener manager + pattern matching
├── sub.go              # Subscriber manager
├── option.go           # Option functions
└── README.md
```