bee in action

main.go

router.Get("/stream/{id}", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(200)
		id := chi.URLParam(r, "id")
		sse := datastar.NewSSE(w, r)
		_ = sse

		ctx := bee.WithJetStream(r.Context(), js)
		ctx = bee.WithNats(ctx, nc)

		agg := &Aggregate{}
		updates := bee.ReplayAndSubscribe(ctx, users.Aggregate, id, agg)
		go func() {
			for {
				select {
				case <-r.Context().Done():
					return
				case update := <-updates:
					sse.MergeFragmentTempl(partials.History(update.History))
				}
			}
		}()
		<-r.Context().Done()
	})

z.go

type userModel struct {
	Name    string `json:"name"`
	Country string `json:"country"`
}

type Aggregate struct {
	History []string
}

func (a *Aggregate) ApplyEvent(event *gen.EventEnvelope) error {
	switch event.EventType {
	case "created":
		var user userModel
		if err := json.Unmarshal(event.Payload, &user); err != nil {
			return err
		}
		a.History = append(a.History, "User created: "+user.Name+" from "+user.Country)
	case "updated":
		var user userModel
		if err := json.Unmarshal(event.Payload, &user); err != nil {
			return err
		}
		a.History = append(a.History, "User updated: "+user.Name+" from "+user.Country)
	default:
		return nil // Ignore other event types
	}
	return nil
}