From c050815900e5572ec49d622b960a5a7c7f66228a Mon Sep 17 00:00:00 2001 From: Daniel Noulet Date: Sat, 19 Sep 2026 22:22:42 +0200 Subject: [PATCH] feat: sync interview content in real time 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> --- .github/copilot-instructions.md | 9 +- README.md | 14 +-- backend/internal/eventbus/amqp.go | 24 ++--- backend/internal/eventbus/eventbus.go | 22 +++-- backend/internal/integration/api_test.go | 58 +++++++++-- backend/internal/service/interviews.go | 80 +++++++++++++-- backend/internal/service/services_test.go | 63 +++++++++++- backend/internal/web/handler/handler.go | 28 +++++- backend/internal/web/router/router.go | 1 + backend/internal/web/router/router_test.go | 7 +- frontend/README.md | 12 ++- frontend/nginx.conf | 3 + frontend/src/api/client.ts | 110 ++++++++++++++------- frontend/src/app/App.test.tsx | 34 ++++++- frontend/src/features/interviews/pages.tsx | 62 ++++++++++-- frontend/src/types/index.ts | 12 +++ 16 files changed, 438 insertions(+), 101 deletions(-) diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index cf7cd3d..4547459 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -33,10 +33,11 @@ The required toolchain is Go 1.27.1 and React 19.3 with TypeScript/Vite. Use `internal/database/{repository,mapper,entity}`. Never expose GORM entities from handlers. - `backend/cmd/seed` applies embedded environment-aware fixtures. -- `backend/internal/eventbus` publishes persisted interview notes to the durable - `go-guess.interview-notes` RabbitMQ fanout exchange. Each API instance consumes - through its own exclusive queue and distributes events to authenticated SSE - subscribers. +- `backend/internal/eventbus` publishes note and shared-document events to the + durable `go-guess.interview-events` RabbitMQ fanout exchange. Each API instance + consumes through its own exclusive queue and distributes interview-filtered + events to authenticated interviewer SSE streams or token-authorized participant + streams. - `frontend/src/app` owns application routing and providers, `src/features` groups feature pages and components, `src/components` contains reusable UI, `src/api/client.ts` is the typed HTTP boundary, `src/types` contains client diff --git a/README.md b/README.md index 9097176..b7b24ca 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ timed assessments followed by collaborative interviews. - Go 1.27.1 HTTP API using Chi, GORM, PostgreSQL, Goose, bcrypt, and signed JWTs - React 19.3, TypeScript, and Vite - PostgreSQL 18 -- RabbitMQ 4 for live interview-note fanout +- RabbitMQ 4 for live interview event fanout ## Run locally @@ -62,7 +62,7 @@ cd frontend && npm test -- src/app/App.test.tsx -t "filters jobs by title" Docker must be available for `make test`. The integration suite starts PostgreSQL 18 and RabbitMQ 4 in disposable Testcontainers, applies every Goose migration, loads deterministic test fixtures, exercises login and the complete assessment and -interview workflow, verifies live note fanout, and removes both containers +interview workflow, verifies live note and document fanout, and removes both containers afterward. Start an isolated application stack populated with test data for manual UI testing: @@ -92,11 +92,11 @@ Completed assessments can be marked passed or failed. A passed candidate can be scheduled for an interview with a date, location or meeting link, shared document, and selected co-interviewers. Each interviewer has an inbox for invitations and a calendar for accepted interviews. During a started interview, attendees can add -private notes; the API persists each note and publishes it through a durable -RabbitMQ fanout exchange so connected interviewer sessions receive it over an -authenticated server-sent-event stream. The candidate receives a separate opaque -meeting link that exposes the schedule and shared document, but never private notes -or assessment reference answers. +private notes. Notes and shared-document changes are published through a durable +RabbitMQ fanout exchange and delivered over server-sent-event streams. Authenticated +interviewer sessions receive both event types; the candidate's opaque meeting link +uses a token-authorized stream that receives shared-document updates but never +private notes or assessment reference answers. Development fixtures include `co-interviewer@go-guess.local` with password `admin123`. Additional co-interviewers can be created from the Users screen. diff --git a/backend/internal/eventbus/amqp.go b/backend/internal/eventbus/amqp.go index a3d39d7..e3f56b1 100644 --- a/backend/internal/eventbus/amqp.go +++ b/backend/internal/eventbus/amqp.go @@ -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() diff --git a/backend/internal/eventbus/eventbus.go b/backend/internal/eventbus/eventbus.go index 9d040fd..349966c 100644 --- a/backend/internal/eventbus/eventbus.go +++ b/backend/internal/eventbus/eventbus.go @@ -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() { diff --git a/backend/internal/integration/api_test.go b/backend/internal/integration/api_test.go index 1ecb894..3d29387 100644 --- a/backend/internal/integration/api_test.go +++ b/backend/internal/integration/api_test.go @@ -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":`)) { diff --git a/backend/internal/service/interviews.go b/backend/internal/service/interviews.go index 6480626..a0e77ac 100644 --- a/backend/internal/service/interviews.go +++ b/backend/internal/service/interviews.go @@ -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 { diff --git a/backend/internal/service/services_test.go b/backend/internal/service/services_test.go index e83d412..9d050b5 100644 --- a/backend/internal/service/services_test.go +++ b/backend/internal/service/services_test.go @@ -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 ") diff --git a/backend/internal/web/handler/handler.go b/backend/internal/web/handler/handler.go index 9e541df..13bd539 100644 --- a/backend/internal/web/handler/handler.go +++ b/backend/internal/web/handler/handler.go @@ -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() diff --git a/backend/internal/web/router/router.go b/backend/internal/web/router/router.go index fe59852..3dfc11c 100644 --- a/backend/internal/web/router/router.go +++ b/backend/internal/web/router/router.go @@ -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) diff --git a/backend/internal/web/router/router_test.go b/backend/internal/web/router/router_test.go index 9a70111..23cf473 100644 --- a/backend/internal/web/router/router_test.go +++ b/backend/internal/web/router/router_test.go @@ -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 { diff --git a/frontend/README.md b/frontend/README.md index a259c1c..d9456b9 100644 --- a/frontend/README.md +++ b/frontend/README.md @@ -71,6 +71,7 @@ Supported routes: - `GET|POST /api/scheduled-interviews/:id/notes` - `GET /api/scheduled-interviews/:id/events` (authenticated fetch-based SSE) - `GET /api/participant-meetings/:token` (public) +- `GET /api/participant-meetings/:token/events` (token-authorized SSE) Participant creation uses multipart fields `firstName`, `lastName`, `birthday`, `email`, `contactInfo`, `photo`, and `cv`. @@ -96,9 +97,9 @@ The authenticated workspace also includes: - **Inbox** for accepting or declining interview assignments. - **Calendar** for interviews grouped by date. - **Interviewer sessions** for changing status, editing prominent shared documentation, and - posting private notes while an interview is started. Notes arrive live through an authenticated - streaming `fetch`; the app intentionally does not use `EventSource`, because the request needs - the JWT authorization header. + posting private notes while an interview is started. Notes and shared-document changes arrive + live through an authenticated streaming `fetch`; the app intentionally does not use + `EventSource`, because the request needs the JWT authorization header. Participants open the generated `/participant/:invitationId` URL without signing in. The opaque invitation ID authorizes the existing `/api/interviews/:token` API. Code-review questions are shown @@ -108,8 +109,9 @@ derived from the backend `acceptedAt` timestamp plus the job duration so refresh it. Questions stay hidden until the participant explicitly accepts the invitation. Scheduled candidates use `/participant/meeting/:token`. This public view shows the interview date, -location, status, and read-only shared documentation. It never renders interviewer notes or -assessment reference answers. +location, status, and read-only shared documentation. A token-authorized event stream updates the +document immediately when an interviewer saves it in another browser. The participant view never +receives interviewer notes or assessment reference answers. ## Container diff --git a/frontend/nginx.conf b/frontend/nginx.conf index 416a992..957980e 100644 --- a/frontend/nginx.conf +++ b/frontend/nginx.conf @@ -12,6 +12,9 @@ server { proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; + proxy_buffering off; + proxy_cache off; + proxy_read_timeout 1h; } location / { diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 75ea02c..51b07fa 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -6,6 +6,7 @@ import type { CreateQuestionInput, CreateScheduledInterviewInput, Interview, + InterviewEvent, Invitation, Job, JobStatus, @@ -97,6 +98,56 @@ async function requestBlob(path: string): Promise { return response.blob() } +async function subscribeToInterviewEvents( + path: string, + onEvent: (event: InterviewEvent) => void, + signal: AbortSignal, + protectedRequest: boolean, + onConnectionError?: (error: unknown) => void, +) { + while (!signal.aborted) { + try { + const token = authStorage.token() + const response = await fetch(`/api${path}`, { + headers: { + Accept: 'text/event-stream', + ...(protectedRequest && token ? { Authorization: 'Bearer ' + token } : {}), + }, + signal, + }) + if (!response.ok || !response.body) + throw new ApiError(`Live updates failed with status ${response.status}`, response.status) + const reader = response.body.getReader() + const decoder = new TextDecoder() + let buffer = '' + while (!signal.aborted) { + const { done, value } = await reader.read() + if (done) break + buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n') + let boundary = buffer.indexOf('\n\n') + while (boundary >= 0) { + const message = buffer.slice(0, boundary) + buffer = buffer.slice(boundary + 2) + const data = message + .split('\n') + .filter((line) => line.startsWith('data:')) + .map((line) => line.slice(5).trimStart()) + .join('\n') + if (data) onEvent(JSON.parse(data) as InterviewEvent) + boundary = buffer.indexOf('\n\n') + } + } + } catch (error) { + if (signal.aborted) return + if (error instanceof ApiError && error.status >= 400 && error.status < 500) throw error + onConnectionError?.(error) + } + if (!signal.aborted) { + await new Promise((resolve) => window.setTimeout(resolve, 1000)) + } + } +} + export const api = { login: (email: string, password: string) => request( @@ -226,44 +277,35 @@ export const api = { method: 'POST', body: JSON.stringify({ body }), }), - subscribe: async (id: number, onNote: (note: InterviewNote) => void, signal: AbortSignal) => { - const token = authStorage.token() - const response = await fetch(`/api/scheduled-interviews/${id}/events`, { - headers: { - Accept: 'text/event-stream', - ...(token ? { Authorization: 'Bearer ' + token } : {}), - }, + subscribe: ( + id: number, + onEvent: (event: InterviewEvent) => void, + signal: AbortSignal, + onConnectionError?: (error: unknown) => void, + ) => + subscribeToInterviewEvents( + `/scheduled-interviews/${id}/events`, + onEvent, signal, - }) - if (!response.ok || !response.body) - throw new ApiError(`Live updates failed with status ${response.status}`, response.status) - const reader = response.body.getReader() - const decoder = new TextDecoder() - let buffer = '' - while (!signal.aborted) { - const { done, value } = await reader.read() - if (done) break - buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n') - let boundary = buffer.indexOf('\n\n') - while (boundary >= 0) { - const message = buffer.slice(0, boundary) - buffer = buffer.slice(boundary + 2) - const data = message - .split('\n') - .filter((line) => line.startsWith('data:')) - .map((line) => line.slice(5).trimStart()) - .join('\n') - if (data) { - const event = JSON.parse(data) as InterviewNote | { note: InterviewNote } - onNote('note' in event ? event.note : event) - } - boundary = buffer.indexOf('\n\n') - } - } - }, + true, + onConnectionError, + ), }, participantMeetings: { get: (token: string) => request(`/participant-meetings/${token}`, {}, false), + subscribe: ( + token: string, + onEvent: (event: InterviewEvent) => void, + signal: AbortSignal, + onConnectionError?: (error: unknown) => void, + ) => + subscribeToInterviewEvents( + `/participant-meetings/${token}/events`, + onEvent, + signal, + false, + onConnectionError, + ), }, } diff --git a/frontend/src/app/App.test.tsx b/frontend/src/app/App.test.tsx index 2eaf2b5..041f3d7 100644 --- a/frontend/src/app/App.test.tsx +++ b/frontend/src/app/App.test.tsx @@ -853,7 +853,17 @@ describe('Go Guess frontend', () => { const stream = new ReadableStream({ start(controller) { controller.enqueue( - new TextEncoder().encode(`event: note\ndata: ${JSON.stringify(liveNote)}\n\n`), + new TextEncoder().encode( + `event: interview-event\ndata: ${JSON.stringify({ + type: 'note.created', + interviewId: 80, + note: liveNote, + })}\n\nevent: interview-event\ndata: ${JSON.stringify({ + type: 'document.updated', + interviewId: 80, + sharedDocument: 'Remote agenda', + })}\n\n`, + ), ) controller.close() }, @@ -867,6 +877,7 @@ describe('Go Guess frontend', () => { }) renderApp('/scheduled-interviews/80') expect(await screen.findByText('Strong system design answer.')).toBeInTheDocument() + expect(screen.getByLabelText('Shared documentation')).toHaveValue('Remote agenda') await user.click(screen.getByRole('button', { name: 'Copy participant interview link' })) expect(writeText).toHaveBeenCalledWith('/participant/meeting/meeting-token') expect( @@ -903,11 +914,30 @@ describe('Go Guess frontend', () => { status: 'scheduled', sharedDocument: 'Please prepare the architecture exercise.', }) + if (String(input) === '/api/participant-meetings/meeting-token/events') { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + `event: interview-event\ndata: ${JSON.stringify({ + type: 'document.updated', + interviewId: 80, + sharedDocument: 'The architecture exercise changed live.', + })}\n\n`, + ), + ) + controller.close() + }, + }) + return Promise.resolve( + new Response(stream, { status: 200, headers: { 'Content-Type': 'text/event-stream' } }), + ) + } throw new Error(`Unexpected request: ${String(input)}`) }) renderApp('/participant/meeting/meeting-token') expect(await screen.findByRole('heading', { name: job.title })).toBeInTheDocument() - expect(screen.getByText('Please prepare the architecture exercise.')).toBeInTheDocument() + expect(await screen.findByText('The architecture exercise changed live.')).toBeInTheDocument() expect(screen.queryByText(/live notes/i)).not.toBeInTheDocument() expect(screen.queryByText(question.referenceAnswer)).not.toBeInTheDocument() }) diff --git a/frontend/src/features/interviews/pages.tsx b/frontend/src/features/interviews/pages.tsx index d51336e..5feab26 100644 --- a/frontend/src/features/interviews/pages.tsx +++ b/frontend/src/features/interviews/pages.tsx @@ -146,6 +146,7 @@ export function ScheduledInterviewPage() { const [error, setError] = useState('') const [saving, setSaving] = useState(false) const [linkCopied, setLinkCopied] = useState(false) + const loadedInterviewID = interview?.id useEffect(() => { let active = true @@ -163,23 +164,34 @@ export function ScheduledInterviewPage() { }, [id]) useEffect(() => { - if (interview?.status !== 'started') return + if (!loadedInterviewID) return const controller = new AbortController() void api.scheduledInterviews .subscribe( id, - (note) => { - setNotes((current) => - current.some((item) => item.id === note.id) ? current : [...current, note], - ) + (event) => { + if (event.type === 'note.created') { + setNotes((current) => + current.some((item) => item.id === event.note.id) + ? current + : [...current, event.note], + ) + } + if (event.type === 'document.updated') { + setDocument(event.sharedDocument) + setInterview((current) => + current ? { ...current, sharedDocument: event.sharedDocument } : current, + ) + } }, controller.signal, + (value) => setError(message(value)), ) .catch((value) => { if (!controller.signal.aborted) setError(message(value)) }) return () => controller.abort() - }, [id, interview?.status]) + }, [id, loadedInterviewID]) async function setStatus(status: string) { setError('') @@ -363,10 +375,44 @@ export function ParticipantMeetingPage() { const [meeting, setMeeting] = useState() const [error, setError] = useState('') useEffect(() => { + let active = true + let latestDocument: string | undefined + const controller = new AbortController() + void api.participantMeetings + .subscribe( + token, + (event) => { + if (event.type !== 'document.updated') return + latestDocument = event.sharedDocument + setMeeting((current) => + current ? { ...current, sharedDocument: event.sharedDocument } : current, + ) + }, + controller.signal, + (value) => { + if (active) setError(message(value)) + }, + ) + .catch((value) => { + if (active && !controller.signal.aborted) setError(message(value)) + }) void api.participantMeetings .get(token) - .then(setMeeting) - .catch((value) => setError(message(value))) + .then((value) => { + if (active) { + setMeeting({ + ...value, + sharedDocument: latestDocument ?? value.sharedDocument, + }) + } + }) + .catch((value) => { + if (active) setError(message(value)) + }) + return () => { + active = false + controller.abort() + } }, [token]) return (
diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 4d8f47c..8e7bbe8 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -90,6 +90,18 @@ export interface InterviewNote { createdAt: string } +export type InterviewEvent = + | { + type: 'note.created' + interviewId: number + note: InterviewNote + } + | { + type: 'document.updated' + interviewId: number + sharedDocument: string + } + export interface ScheduledInterview { id: number jobId: number