fanOutOne handles a single leased event: read enabled subscribers, apply filter matching, insert delivery rows, mark outbox row processed — all in one transaction so partial fan-out is impossible.
(ctx context.Context, ev leasedEvent)
| 303 | // apply filter matching, insert delivery rows, mark outbox row |
| 304 | // processed — all in one transaction so partial fan-out is impossible. |
| 305 | func (w *OutboxWorker) fanOutOne(ctx context.Context, ev leasedEvent) { |
| 306 | webhooks, err := w.identityStore.ListEnabledWebhooksForRouting(ctx, ev.userID, ev.eventType) |
| 307 | if err != nil { |
| 308 | w.recordFailure(ctx, ev.id, fmt.Sprintf("list subscribers: %v", err)) |
| 309 | return |
| 310 | } |
| 311 | |
| 312 | // Apply filter matching. Need to reconstruct an Event-shaped |
| 313 | // struct from the leasedEvent to feed `matches`. |
| 314 | eventForMatching := Event{ |
| 315 | Type: ev.eventType, |
| 316 | UserID: ev.userID, |
| 317 | AgentID: derefString(ev.agentID), |
| 318 | ConversationID: derefString(ev.conversationID), |
| 319 | // Labels: not currently tracked on webhook_events; deferred. |
| 320 | MessageID: derefString(ev.messageID), |
| 321 | } |
| 322 | |
| 323 | // matched starts as an empty slice (not nil) so pgx serializes it |
| 324 | // as the empty Postgres array '{}', not NULL. The column is |
| 325 | // matched_webhook_ids TEXT[] NOT NULL DEFAULT '{}' — a NULL would |
| 326 | // fail the NOT NULL constraint and the UPDATE would error. |
| 327 | matched := make([]string, 0, len(webhooks)) |
| 328 | for _, w := range webhooks { |
| 329 | if matches(eventForMatching, w.Filters) { |
| 330 | matched = append(matched, w.ID) |
| 331 | } |
| 332 | } |
| 333 | |
| 334 | if len(matched) == 0 { |
| 335 | w.metrics.OutboxEventsNoMatch(ev.eventType) |
| 336 | } else { |
| 337 | w.metrics.OutboxEventsFanOut(ev.eventType, len(matched)) |
| 338 | } |
| 339 | err = poolBeginFunc(ctx, w.pool, func(tx pgx.Tx) error { |
| 340 | if len(matched) > 0 { |
| 341 | if err := insertPendingBatchTx(ctx, tx, ev.id, matched, ev.eventType, ev.messageID, ev.envelope); err != nil { |
| 342 | return err |
| 343 | } |
| 344 | } |
| 345 | // Conditional UPDATE: if another worker already processed |
| 346 | // this event row (e.g. our lease expired during a long |
| 347 | // fan-out and replica B took over and finished), the |
| 348 | // status='pending' predicate matches zero rows and our |
| 349 | // UPDATE no-ops. Lease-vs-fanout race fix from §4.4. |
| 350 | newStatus := "processed" |
| 351 | if len(matched) == 0 { |
| 352 | newStatus = "no_match" |
| 353 | } |
| 354 | _, err := tx.Exec(ctx, |
| 355 | `UPDATE webhook_events |
| 356 | SET status = $1, processed_at = now(), matched_webhook_ids = $3 |
| 357 | WHERE id = $2 AND status = 'pending'`, |
| 358 | newStatus, ev.id, matched, |
| 359 | ) |
| 360 | return err |
| 361 | }) |
| 362 | if err != nil { |
no test coverage detected