angelandClaude Sonnet 5
984a4465f8
Dispatching an event to the listener previously happened from an un-awaited goroutine, with only a log line on failure - once processed_at was set, a dropped or failed notification was permanently and silently lost, with no way to tell it had happened. Replaces the processed/processed_successfully booleans with a notified_at/processed_at pair (migration 000031): the dispatcher sets notified_at and hands the event to MockEventListener.Notify, which now also receives an ack callback the listener calls whenever it's truly done, synchronously or arbitrarily later. Anything still "notified" but unacked past notifyRetryAfter (a tunable field, not a stored per-row timestamp) gets notified again on the dispatcher's normal poll/reactive loop - no new retry mechanism needed. ack is idempotent, since a late ack from an earlier attempt and one from a retry can both eventually fire for the same event. Dispatch is deferred until after the transaction that recorded notified_at has actually committed, so ack's independent write can't race a still-open transaction it implicitly depends on being visible. The real SSE listener (server/sse/db_event_publisher.go) acks inline, since its work is synchronous. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XEDaCB7C2NEBgyvqEtZuxY
52 lines
1.3 KiB
Go
52 lines
1.3 KiB
Go
package sse
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"ruben/inventory2/domains/raw_events"
|
|
)
|
|
|
|
type (
|
|
DBEventPublisher struct {
|
|
queue sender
|
|
getAccountID func(context.Context, raw_events.Event) (int64, error)
|
|
getEventType func(acctID int64, e raw_events.Event) (string, error)
|
|
}
|
|
)
|
|
|
|
func (q *Queue) NewDBEventPublisher(
|
|
getAccountID func(context.Context, raw_events.Event) (int64, error),
|
|
getEventType func(acctID int64, e raw_events.Event) (string, error),
|
|
) *DBEventPublisher {
|
|
return &DBEventPublisher{
|
|
queue: q,
|
|
getAccountID: getAccountID,
|
|
getEventType: getEventType,
|
|
}
|
|
}
|
|
|
|
// Notify publishes e to the SSE queue. The publish itself is quick and
|
|
// synchronous, so it acks inline once it succeeds - there's no separate
|
|
// async completion to wait for here.
|
|
func (p *DBEventPublisher) Notify(ctx context.Context, e raw_events.Event, ack func(context.Context) error) error {
|
|
acctID, err := p.getAccountID(ctx, e)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to get account id: %w", err)
|
|
}
|
|
|
|
et, err := p.getEventType(acctID, e)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to compute event type: %w", err)
|
|
}
|
|
|
|
if err := p.queue.Send(ctx, StandardEvent(acctID, et)); err != nil {
|
|
return fmt.Errorf("failed to send sse event to listener: %w", err)
|
|
}
|
|
|
|
if err := ack(ctx); err != nil {
|
|
return fmt.Errorf("failed to ack event: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|