dnoulet/go-guess
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: sync interview content in real time

Browse source 
Publish shared-document and note events through RabbitMQ, stream token-scoped document updates to participants, and update connected interviewer and participant browsers through resilient SSE subscriptions.

Add snapshots, heartbeats, reconnection behavior, proxy streaming configuration, privacy filtering, and end-to-end RabbitMQ coverage.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Daniel Noulet 2026-09-19 22:22:42 +02:00
parent e4fee06e2f
commit c050815900
16 changed files with 438 additions and 101 deletions
Download patch file Download diff file Expand all files Collapse all files

24
backend/internal/eventbus/amqp.go
View file

@ -10,7 +10,7 @@ import (
amqp "github.com/rabbitmq/amqp091-go"
)
const interviewExchange = "go-guess.interview-notes"
const interviewExchange = "go-guess.interview-events"
type AMQPBus struct {
url string
@ -18,14 +18,14 @@ type AMQPBus struct {
connection *amqp.Connection
publisher *amqp.Channel
consumer *amqp.Channel
subscribers map[chan InterviewNoteEvent]struct{}
subscribers map[chan InterviewEvent]struct{}
done chan struct{}
closed bool
}
func NewAMQPBus(url string) (*AMQPBus, error) {
bus := &AMQPBus{
url: url, subscribers: make(map[chan InterviewNoteEvent]struct{}), done: make(chan struct{}),
url: url, subscribers: make(map[chan InterviewEvent]struct{}), done: make(chan struct{}),
}
deliveries, err := bus.connect()
if err != nil {
@ -89,10 +89,10 @@ func (b *AMQPBus) connect() (<-chan amqp.Delivery, error) {
return deliveries, nil
}
func (b *AMQPBus) PublishInterviewNote(ctx context.Context, event InterviewNoteEvent) error {
func (b *AMQPBus) PublishInterviewEvent(ctx context.Context, event InterviewEvent) error {
body, err := json.Marshal(event)
if err != nil {
return fmt.Errorf("encode interview note event: %w", err)
return fmt.Errorf("encode interview event: %w", err)
}
b.mu.Lock()
defer b.mu.Unlock()
@ -114,11 +114,11 @@ func (b *AMQPBus) PublishInterviewNote(ctx context.Context, event InterviewNoteE
}
deliveries, reconnectErr := b.connect()
if reconnectErr != nil {
return fmt.Errorf("publish interview note: %w; reconnect: %v", err, reconnectErr)
return fmt.Errorf("publish interview event: %w; reconnect: %v", err, reconnectErr)
}
go b.dispatch(deliveries)
if err := b.publishConfirmed(ctx, body); err != nil {
return fmt.Errorf("publish interview note after reconnect: %w", err)
return fmt.Errorf("publish interview event after reconnect: %w", err)
}
return nil
}
@ -136,18 +136,18 @@ func (b *AMQPBus) publishConfirmed(ctx context.Context, body []byte) error {
return err
}
if !acknowledged {
return errors.New("RabbitMQ rejected interview note")
return errors.New("RabbitMQ rejected interview event")
}
return nil
}
func (b *AMQPBus) SubscribeInterviewNotes(ctx context.Context) (<-chan InterviewNoteEvent, error) {
func (b *AMQPBus) SubscribeInterviewEvents(ctx context.Context) (<-chan InterviewEvent, error) {
b.mu.Lock()
if b.closed {
b.mu.Unlock()
return nil, ErrClosed
}
ch := make(chan InterviewNoteEvent, 16)
ch := make(chan InterviewEvent, 16)
b.subscribers[ch] = struct{}{}
b.mu.Unlock()
go func() {
@ -171,7 +171,7 @@ func (b *AMQPBus) dispatch(deliveries <-chan amqp.Delivery) {
if !ok {
return
}
var event InterviewNoteEvent
var event InterviewEvent
if json.Unmarshal(delivery.Body, &event) != nil {
continue
}
@ -180,6 +180,8 @@ func (b *AMQPBus) dispatch(deliveries <-chan amqp.Delivery) {
select {
case subscriber <- event:
default:
delete(b.subscribers, subscriber)
close(subscriber)
}
}
b.mu.Unlock()

22
backend/internal/eventbus/eventbus.go
View file

@ -8,7 +8,6 @@ import (
)
type InterviewNoteEvent struct {
InterviewID uint `json:"interviewId"`
ID uint `json:"id"`
AuthorUserID uint `json:"authorUserId"`
AuthorName string `json:"authorName"`
@ -16,9 +15,16 @@ type InterviewNoteEvent struct {
CreatedAt time.Time `json:"createdAt"`
}
type InterviewEvent struct {
Type string `json:"type"`
InterviewID uint `json:"interviewId"`
Note *InterviewNoteEvent `json:"note,omitempty"`
SharedDocument string `json:"sharedDocument,omitempty"`
}
type IEventBus interface {
PublishInterviewNote(context.Context, InterviewNoteEvent) error
SubscribeInterviewNotes(context.Context) (<-chan InterviewNoteEvent, error)
PublishInterviewEvent(context.Context, InterviewEvent) error
SubscribeInterviewEvents(context.Context) (<-chan InterviewEvent, error)
Close() error
}
@ -26,15 +32,15 @@ var ErrClosed = errors.New("event bus is closed")
type MemoryBus struct {
mu sync.RWMutex
subscribers map[chan InterviewNoteEvent]struct{}
subscribers map[chan InterviewEvent]struct{}
closed bool
}
func NewMemoryBus() *MemoryBus {
return &MemoryBus{subscribers: make(map[chan InterviewNoteEvent]struct{})}
return &MemoryBus{subscribers: make(map[chan InterviewEvent]struct{})}
}
func (b *MemoryBus) PublishInterviewNote(ctx context.Context, event InterviewNoteEvent) error {
func (b *MemoryBus) PublishInterviewEvent(ctx context.Context, event InterviewEvent) error {
b.mu.RLock()
defer b.mu.RUnlock()
if b.closed {
@ -50,13 +56,13 @@ func (b *MemoryBus) PublishInterviewNote(ctx context.Context, event InterviewNot
return nil
}
func (b *MemoryBus) SubscribeInterviewNotes(ctx context.Context) (<-chan InterviewNoteEvent, error) {
func (b *MemoryBus) SubscribeInterviewEvents(ctx context.Context) (<-chan InterviewEvent, error) {
b.mu.Lock()
if b.closed {
b.mu.Unlock()
return nil, ErrClosed
}
ch := make(chan InterviewNoteEvent, 16)
ch := make(chan InterviewEvent, 16)
b.subscribers[ch] = struct{}{}
b.mu.Unlock()
go func() {

58
backend/internal/integration/api_test.go
View file

@ -3,6 +3,7 @@
package integration
import (
"bufio"
"bytes"
"context"
"database/sql"
@ -313,13 +314,55 @@ func TestCompleteInterviewWorkflow(t *testing.T) {
requestJSON(t, http.MethodPatch, fmt.Sprintf("/api/scheduled-interviews/%d/status", scheduled.ID),
login.Token, map[string]any{"status": "started"}, http.StatusOK, &scheduled)
subscriptionContext, cancelSubscriptions := context.WithCancel(context.Background())
defer cancelSubscriptions()
firstEvents, err := testBus.SubscribeInterviewNotes(subscriptionContext)
streamContext, cancelStream := context.WithTimeout(context.Background(), 5*time.Second)
defer cancelStream()
streamRequest, err := http.NewRequestWithContext(
streamContext, http.MethodGet,
testServer.URL+"/api/participant-meetings/"+scheduled.CandidateToken+"/events", nil,
)
if err != nil {
t.Fatal(err)
}
secondEvents, err := testBus.SubscribeInterviewNotes(subscriptionContext)
streamResponse, err := http.DefaultClient.Do(streamRequest)
if err != nil {
t.Fatal(err)
}
defer streamResponse.Body.Close()
if streamResponse.StatusCode != http.StatusOK {
t.Fatalf("candidate event stream returned %d", streamResponse.StatusCode)
}
const updatedDocument = "Updated collaborative exercise"
requestJSON(t, http.MethodPatch, fmt.Sprintf("/api/scheduled-interviews/%d/document", scheduled.ID),
login.Token, map[string]any{"sharedDocument": updatedDocument}, http.StatusOK, &scheduled)
var documentEvent eventbus.InterviewEvent
scanner := bufio.NewScanner(streamResponse.Body)
for scanner.Scan() {
if strings.HasPrefix(scanner.Text(), "data: ") {
if err := json.Unmarshal([]byte(strings.TrimPrefix(scanner.Text(), "data: ")), &documentEvent); err != nil {
t.Fatal(err)
}
if documentEvent.Type == "document.updated" &&
documentEvent.SharedDocument == updatedDocument {
break
}
}
}
if err := scanner.Err(); err != nil {
t.Fatal(err)
}
if documentEvent.Type != "document.updated" ||
documentEvent.InterviewID != scheduled.ID ||
documentEvent.SharedDocument != updatedDocument {
t.Fatalf("candidate received unexpected document event: %#v", documentEvent)
}
subscriptionContext, cancelSubscriptions := context.WithCancel(context.Background())
defer cancelSubscriptions()
firstEvents, err := testBus.SubscribeInterviewEvents(subscriptionContext)
if err != nil {
t.Fatal(err)
}
secondEvents, err := testBus.SubscribeInterviewEvents(subscriptionContext)
if err != nil {
t.Fatal(err)
}
@ -331,10 +374,11 @@ func TestCompleteInterviewWorkflow(t *testing.T) {
if note.AuthorUserID != coInterviewer.ID || note.AuthorName != "Co Interviewer" {
t.Fatalf("unexpected note response: %#v", note)
}
for index, events := range []<-chan eventbus.InterviewNoteEvent{firstEvents, secondEvents} {
for index, events := range []<-chan eventbus.InterviewEvent{firstEvents, secondEvents} {
select {
case received := <-events:
if received.ID != note.ID || received.AuthorUserID != coInterviewer.ID {
if received.Type != "note.created" || received.Note == nil ||
received.Note.ID != note.ID || received.Note.AuthorUserID != coInterviewer.ID {
t.Fatalf("subscriber %d received unexpected RabbitMQ event: %#v", index+1, received)
}
encoded, err := json.Marshal(received)
@ -358,7 +402,7 @@ func TestCompleteInterviewWorkflow(t *testing.T) {
var meeting webmodel.ParticipantMeetingResponse
body := requestJSON(t, http.MethodGet, "/api/participant-meetings/"+scheduled.CandidateToken,
"", nil, http.StatusOK, &meeting)
if meeting.SharedDocument != "Collaborative notes" ||
if meeting.SharedDocument != updatedDocument ||
bytes.Contains(body, []byte("participantEmail")) ||
bytes.Contains(body, []byte("referenceAnswer")) ||
bytes.Contains(body, []byte(`"notes":`)) {

80
backend/internal/service/interviews.go
View file

@ -30,7 +30,8 @@ type IScheduledInterviewService interface {
ListInbox(context.Context, uint) ([]model.ScheduledInterview, error)
UpdateInbox(context.Context, uint, uint, string) (model.ScheduledInterview, error)
ListCalendar(context.Context, uint) ([]model.ScheduledInterview, error)
Subscribe(context.Context, uint, uint) (<-chan eventbus.InterviewNoteEvent, error)
Subscribe(context.Context, uint, uint) (<-chan eventbus.InterviewEvent, error)
SubscribeParticipant(context.Context, string) (<-chan eventbus.InterviewEvent, error)
}
type UserService struct {
@ -174,7 +175,16 @@ func (s *ScheduledInterviewService) UpdateDocument(
if _, err := s.Get(ctx, userID, interviewID); err != nil {
return model.ScheduledInterview{}, err
}
return s.store.UpdateScheduledInterviewDocument(ctx, interviewID, document)
updated, err := s.store.UpdateScheduledInterviewDocument(ctx, interviewID, document)
if err != nil {
return model.ScheduledInterview{}, err
}
if err := s.bus.PublishInterviewEvent(ctx, eventbus.InterviewEvent{
Type: "document.updated", InterviewID: interviewID, SharedDocument: updated.SharedDocument,
}); err != nil {
return model.ScheduledInterview{}, fmt.Errorf("publish persisted interview document: %w", err)
}
return updated, nil
}
func (s *ScheduledInterviewService) ListNotes(
@ -209,9 +219,12 @@ func (s *ScheduledInterviewService) CreateNote(
}
return model.InterviewNote{}, err
}
if err := s.bus.PublishInterviewNote(ctx, eventbus.InterviewNoteEvent{
InterviewID: note.InterviewID, ID: note.ID, AuthorUserID: note.AuthorID,
AuthorName: note.Author.DisplayName, Body: note.Body, CreatedAt: note.CreatedAt,
if err := s.bus.PublishInterviewEvent(ctx, eventbus.InterviewEvent{
Type: "note.created", InterviewID: note.InterviewID,
Note: &eventbus.InterviewNoteEvent{
ID: note.ID, AuthorUserID: note.AuthorID, AuthorName: note.Author.DisplayName,
Body: note.Body, CreatedAt: note.CreatedAt,
},
}); err != nil {
return model.InterviewNote{}, fmt.Errorf("publish persisted interview note: %w", err)
}
@ -244,11 +257,64 @@ func (s *ScheduledInterviewService) ListCalendar(
func (s *ScheduledInterviewService) Subscribe(
ctx context.Context, userID, interviewID uint,
) (<-chan eventbus.InterviewNoteEvent, error) {
) (<-chan eventbus.InterviewEvent, error) {
if _, err := s.Get(ctx, userID, interviewID); err != nil {
return nil, err
}
return s.bus.SubscribeInterviewNotes(ctx)
return s.subscribeToInterview(ctx, interviewID, false)
}
func (s *ScheduledInterviewService) SubscribeParticipant(
ctx context.Context, token string,
) (<-chan eventbus.InterviewEvent, error) {
interview, err := s.GetParticipantMeeting(ctx, token)
if err != nil {
return nil, err
}
return s.subscribeToInterview(ctx, interview.ID, true)
}
func (s *ScheduledInterviewService) subscribeToInterview(
ctx context.Context, interviewID uint, documentsOnly bool,
) (<-chan eventbus.InterviewEvent, error) {
events, err := s.bus.SubscribeInterviewEvents(ctx)
if err != nil {
return nil, err
}
current, err := s.store.GetScheduledInterview(ctx, interviewID)
if err != nil {
return nil, err
}
filtered := make(chan eventbus.InterviewEvent, 16)
filtered <- eventbus.InterviewEvent{
Type: "document.updated", InterviewID: interviewID,
SharedDocument: current.SharedDocument,
}
go func() {
defer close(filtered)
for {
select {
case <-ctx.Done():
return
case event, ok := <-events:
if !ok {
return
}
if event.InterviewID != interviewID {
continue
}
if documentsOnly && event.Type != "document.updated" {
continue
}
select {
case filtered <- event:
case <-ctx.Done():
return
}
}
}
}()
return filtered, nil
}
func isAttendee(value model.ScheduledInterview, userID uint) bool {

63
backend/internal/service/services_test.go
View file

@ -391,7 +391,7 @@ func TestCreateNotePersistsThenPublishesSafeEvent(t *testing.T) {
},
}
bus := eventbus.NewMemoryBus()
events, err := bus.SubscribeInterviewNotes(context.Background())
events, err := bus.SubscribeInterviewEvents(context.Background())
if err != nil {
t.Fatal(err)
}
@ -401,11 +401,70 @@ func TestCreateNotePersistsThenPublishesSafeEvent(t *testing.T) {
t.Fatal(err)
}
event := <-events
if note.Body != "useful note" || event.ID != note.ID || event.AuthorName != "Ada" {
if note.Body != "useful note" || event.Type != "note.created" || event.Note == nil ||
event.Note.ID != note.ID || event.Note.AuthorName != "Ada" {
t.Fatalf("unexpected note/event: %#v %#v", note, event)
}
}
func TestUpdateDocumentPublishesEvent(t *testing.T) {
store := &scheduledStore{
fakeStore: &fakeStore{},
interview: model.ScheduledInterview{
ID: 9, Status: "scheduled", Attendees: []model.InterviewAttendee{{UserID: 1}},
},
}
bus := eventbus.NewMemoryBus()
events, err := bus.SubscribeInterviewEvents(context.Background())
if err != nil {
t.Fatal(err)
}
interviews := NewScheduledInterviewService(store, bus)
updated, err := interviews.UpdateDocument(context.Background(), 1, 9, "New agenda")
if err != nil {
t.Fatal(err)
}
event := <-events
if updated.SharedDocument != "New agenda" || event.Type != "document.updated" ||
event.InterviewID != 9 || event.SharedDocument != "New agenda" {
t.Fatalf("unexpected document/event: %#v %#v", updated, event)
}
}
func TestParticipantSubscriptionFiltersPrivateNotes(t *testing.T) {
bus := eventbus.NewMemoryBus()
interviews := NewScheduledInterviewService(&fakeStore{}, bus)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
events, err := interviews.SubscribeParticipant(ctx, "candidate-token")
if err != nil {
t.Fatal(err)
}
initial := <-events
if initial.Type != "document.updated" {
t.Fatalf("participant did not receive initial document snapshot: %#v", initial)
}
if err := bus.PublishInterviewEvent(ctx, eventbus.InterviewEvent{
Type: "note.created", InterviewID: 0,
Note: &eventbus.InterviewNoteEvent{ID: 4, Body: "private"},
}); err != nil {
t.Fatal(err)
}
if err := bus.PublishInterviewEvent(ctx, eventbus.InterviewEvent{
Type: "document.updated", InterviewID: 0, SharedDocument: "candidate-safe",
}); err != nil {
t.Fatal(err)
}
select {
case event := <-events:
if event.Type != "document.updated" || event.SharedDocument != "candidate-safe" {
t.Fatalf("participant received unexpected event: %#v", event)
}
case <-time.After(time.Second):
t.Fatal("participant did not receive document event")
}
}
func TestUserCreationHashesPasswordAndUsesCoInterviewerRole(t *testing.T) {
store := &fakeStore{}
user, err := NewUserService(store).Create(context.Background(), " CO@Example.com ", "password1", " Ada ")

28
backend/internal/web/handler/handler.go
View file

@ -15,6 +15,7 @@ import (
"github.com/go-chi/chi/v5"
"github.com/danyel/go-guess/backend/internal/database/repository"
"github.com/danyel/go-guess/backend/internal/eventbus"
"github.com/danyel/go-guess/backend/internal/security"
"github.com/danyel/go-guess/backend/internal/service"
servicemodel "github.com/danyel/go-guess/backend/internal/service/model"
@ -511,6 +512,20 @@ func (h *Handler) InterviewEvents(w http.ResponseWriter, r *http.Request) {
if handleServiceError(w, r, err) {
return
}
streamInterviewEvents(w, r, events)
}
func (h *Handler) ParticipantMeetingEvents(w http.ResponseWriter, r *http.Request) {
events, err := h.interviews.SubscribeParticipant(r.Context(), chi.URLParam(r, "token"))
if handleServiceError(w, r, err) {
return
}
streamInterviewEvents(w, r, events)
}
func streamInterviewEvents(
w http.ResponseWriter, r *http.Request, events <-chan eventbus.InterviewEvent,
) {
flusher, ok := w.(http.Flusher)
if !ok {
internalError(w, r, errors.New("streaming is unsupported"))
@ -519,24 +534,29 @@ func (h *Handler) InterviewEvents(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
w.Header().Set("X-Accel-Buffering", "no")
w.WriteHeader(http.StatusOK)
flusher.Flush()
heartbeat := time.NewTicker(20 * time.Second)
defer heartbeat.Stop()
for {
select {
case <-r.Context().Done():
return
case <-heartbeat.C:
if _, err := io.WriteString(w, ": keepalive\n\n"); err != nil {
return
}
flusher.Flush()
case event, ok := <-events:
if !ok {
return
}
if event.InterviewID != idParam(r) {
continue
}
payload, err := json.Marshal(event)
if err != nil {
return
}
if _, err := fmt.Fprintf(w, "event: interview-note\ndata: %s\n\n", payload); err != nil {
if _, err := fmt.Fprintf(w, "event: interview-event\ndata: %s\n\n", payload); err != nil {
return
}
flusher.Flush()

1
backend/internal/web/router/router.go
View file

@ -23,6 +23,7 @@ func New(h *handler.Handler, tokens security.ITokenManager, frontendURL string)
router.Put("/api/interviews/{token}/answers/{questionId}", h.SaveAnswer)
router.Post("/api/interviews/{token}/finish", h.FinishInterview)
router.Get("/api/participant-meetings/{token}", h.GetParticipantMeeting)
router.Get("/api/participant-meetings/{token}/events", h.ParticipantMeetingEvents)
router.Group(func(protected chi.Router) {
protected.Use(authenticate(tokens))
protected.Get("/api/jobs", h.ListJobs)

7
backend/internal/web/router/router_test.go
View file

@ -128,8 +128,11 @@ func (interviewStub) UpdateInbox(context.Context, uint, uint, string) (model.Sch
func (interviewStub) ListCalendar(context.Context, uint) ([]model.ScheduledInterview, error) {
return nil, nil
}
func (interviewStub) Subscribe(context.Context, uint, uint) (<-chan eventbus.InterviewNoteEvent, error) {
return make(chan eventbus.InterviewNoteEvent), nil
func (interviewStub) Subscribe(context.Context, uint, uint) (<-chan eventbus.InterviewEvent, error) {
return make(chan eventbus.InterviewEvent), nil
}
func (interviewStub) SubscribeParticipant(context.Context, string) (<-chan eventbus.InterviewEvent, error) {
return make(chan eventbus.InterviewEvent), nil
}
type interviewCapture struct {