feat: add collaborative interview scheduling
Add candidate decisions, co-interviewer accounts, scheduled interview inbox and calendar views, candidate-safe meeting links, shared documents, and private live notes backed by RabbitMQ. Extend the disposable integration environment to validate PostgreSQL migrations, seeded users, interview lifecycle behavior, and RabbitMQ fanout. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Daniel Noulet
parent
fd06fbbb5d
commit
75e3c73a61
35 changed files with 3621 additions and 58 deletions
212
backend/internal/eventbus/amqp.go
Normal file
212
backend/internal/eventbus/amqp.go
Normal file
|
|
@ -0,0 +1,212 @@
|
|||
package eventbus
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
amqp "github.com/rabbitmq/amqp091-go"
|
||||
)
|
||||
|
||||
const interviewExchange = "go-guess.interview-notes"
|
||||
|
||||
type AMQPBus struct {
|
||||
url string
|
||||
mu sync.Mutex
|
||||
connection *amqp.Connection
|
||||
publisher *amqp.Channel
|
||||
consumer *amqp.Channel
|
||||
subscribers map[chan InterviewNoteEvent]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{}),
|
||||
}
|
||||
deliveries, err := bus.connect()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
go bus.dispatch(deliveries)
|
||||
return bus, nil
|
||||
}
|
||||
|
||||
func (b *AMQPBus) connect() (<-chan amqp.Delivery, error) {
|
||||
connection, err := amqp.Dial(b.url)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("connect RabbitMQ: %w", err)
|
||||
}
|
||||
publisher, err := connection.Channel()
|
||||
if err != nil {
|
||||
_ = connection.Close()
|
||||
return nil, fmt.Errorf("open RabbitMQ publisher: %w", err)
|
||||
}
|
||||
consumer, err := connection.Channel()
|
||||
if err != nil {
|
||||
_ = publisher.Close()
|
||||
_ = connection.Close()
|
||||
return nil, fmt.Errorf("open RabbitMQ consumer: %w", err)
|
||||
}
|
||||
for _, channel := range []*amqp.Channel{publisher, consumer} {
|
||||
if err := channel.ExchangeDeclare(interviewExchange, "fanout", true, false, false, false, nil); err != nil {
|
||||
_ = consumer.Close()
|
||||
_ = publisher.Close()
|
||||
_ = connection.Close()
|
||||
return nil, fmt.Errorf("declare RabbitMQ exchange: %w", err)
|
||||
}
|
||||
}
|
||||
if err := publisher.Confirm(false); err != nil {
|
||||
_ = consumer.Close()
|
||||
_ = publisher.Close()
|
||||
_ = connection.Close()
|
||||
return nil, fmt.Errorf("enable RabbitMQ publisher confirms: %w", err)
|
||||
}
|
||||
queue, err := consumer.QueueDeclare("", false, true, true, false, nil)
|
||||
if err != nil {
|
||||
_ = consumer.Close()
|
||||
_ = publisher.Close()
|
||||
_ = connection.Close()
|
||||
return nil, fmt.Errorf("declare RabbitMQ queue: %w", err)
|
||||
}
|
||||
if err := consumer.QueueBind(queue.Name, "", interviewExchange, false, nil); err != nil {
|
||||
_ = consumer.Close()
|
||||
_ = publisher.Close()
|
||||
_ = connection.Close()
|
||||
return nil, fmt.Errorf("bind RabbitMQ queue: %w", err)
|
||||
}
|
||||
deliveries, err := consumer.Consume(queue.Name, "", true, true, false, false, nil)
|
||||
if err != nil {
|
||||
_ = consumer.Close()
|
||||
_ = publisher.Close()
|
||||
_ = connection.Close()
|
||||
return nil, fmt.Errorf("consume RabbitMQ events: %w", err)
|
||||
}
|
||||
b.connection, b.publisher, b.consumer = connection, publisher, consumer
|
||||
return deliveries, nil
|
||||
}
|
||||
|
||||
func (b *AMQPBus) PublishInterviewNote(ctx context.Context, event InterviewNoteEvent) error {
|
||||
body, err := json.Marshal(event)
|
||||
if err != nil {
|
||||
return fmt.Errorf("encode interview note event: %w", err)
|
||||
}
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if b.closed {
|
||||
return ErrClosed
|
||||
}
|
||||
err = b.publishConfirmed(ctx, body)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if b.consumer != nil {
|
||||
_ = b.consumer.Close()
|
||||
}
|
||||
if b.publisher != nil {
|
||||
_ = b.publisher.Close()
|
||||
}
|
||||
if b.connection != nil {
|
||||
_ = b.connection.Close()
|
||||
}
|
||||
deliveries, reconnectErr := b.connect()
|
||||
if reconnectErr != nil {
|
||||
return fmt.Errorf("publish interview note: %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 nil
|
||||
}
|
||||
|
||||
func (b *AMQPBus) publishConfirmed(ctx context.Context, body []byte) error {
|
||||
confirmation, err := b.publisher.PublishWithDeferredConfirmWithContext(
|
||||
ctx, interviewExchange, "", false, false, amqp.Publishing{
|
||||
ContentType: "application/json", DeliveryMode: amqp.Persistent, Body: body,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
acknowledged, err := confirmation.WaitContext(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !acknowledged {
|
||||
return errors.New("RabbitMQ rejected interview note")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *AMQPBus) SubscribeInterviewNotes(ctx context.Context) (<-chan InterviewNoteEvent, error) {
|
||||
b.mu.Lock()
|
||||
if b.closed {
|
||||
b.mu.Unlock()
|
||||
return nil, ErrClosed
|
||||
}
|
||||
ch := make(chan InterviewNoteEvent, 16)
|
||||
b.subscribers[ch] = struct{}{}
|
||||
b.mu.Unlock()
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
b.mu.Lock()
|
||||
if _, ok := b.subscribers[ch]; ok {
|
||||
delete(b.subscribers, ch)
|
||||
close(ch)
|
||||
}
|
||||
b.mu.Unlock()
|
||||
}()
|
||||
return ch, nil
|
||||
}
|
||||
|
||||
func (b *AMQPBus) dispatch(deliveries <-chan amqp.Delivery) {
|
||||
for {
|
||||
select {
|
||||
case <-b.done:
|
||||
return
|
||||
case delivery, ok := <-deliveries:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var event InterviewNoteEvent
|
||||
if json.Unmarshal(delivery.Body, &event) != nil {
|
||||
continue
|
||||
}
|
||||
b.mu.Lock()
|
||||
for subscriber := range b.subscribers {
|
||||
select {
|
||||
case subscriber <- event:
|
||||
default:
|
||||
}
|
||||
}
|
||||
b.mu.Unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (b *AMQPBus) Close() error {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if b.closed {
|
||||
return nil
|
||||
}
|
||||
b.closed = true
|
||||
close(b.done)
|
||||
for subscriber := range b.subscribers {
|
||||
close(subscriber)
|
||||
}
|
||||
clear(b.subscribers)
|
||||
if b.consumer != nil {
|
||||
_ = b.consumer.Close()
|
||||
}
|
||||
if b.publisher != nil {
|
||||
_ = b.publisher.Close()
|
||||
}
|
||||
if b.connection != nil {
|
||||
return b.connection.Close()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
86
backend/internal/eventbus/eventbus.go
Normal file
86
backend/internal/eventbus/eventbus.go
Normal file
|
|
@ -0,0 +1,86 @@
|
|||
package eventbus
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type InterviewNoteEvent struct {
|
||||
InterviewID uint `json:"interviewId"`
|
||||
ID uint `json:"id"`
|
||||
AuthorUserID uint `json:"authorUserId"`
|
||||
AuthorName string `json:"authorName"`
|
||||
Body string `json:"body"`
|
||||
CreatedAt time.Time `json:"createdAt"`
|
||||
}
|
||||
|
||||
type IEventBus interface {
|
||||
PublishInterviewNote(context.Context, InterviewNoteEvent) error
|
||||
SubscribeInterviewNotes(context.Context) (<-chan InterviewNoteEvent, error)
|
||||
Close() error
|
||||
}
|
||||
|
||||
var ErrClosed = errors.New("event bus is closed")
|
||||
|
||||
type MemoryBus struct {
|
||||
mu sync.RWMutex
|
||||
subscribers map[chan InterviewNoteEvent]struct{}
|
||||
closed bool
|
||||
}
|
||||
|
||||
func NewMemoryBus() *MemoryBus {
|
||||
return &MemoryBus{subscribers: make(map[chan InterviewNoteEvent]struct{})}
|
||||
}
|
||||
|
||||
func (b *MemoryBus) PublishInterviewNote(ctx context.Context, event InterviewNoteEvent) error {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
if b.closed {
|
||||
return ErrClosed
|
||||
}
|
||||
for subscriber := range b.subscribers {
|
||||
select {
|
||||
case subscriber <- event:
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *MemoryBus) SubscribeInterviewNotes(ctx context.Context) (<-chan InterviewNoteEvent, error) {
|
||||
b.mu.Lock()
|
||||
if b.closed {
|
||||
b.mu.Unlock()
|
||||
return nil, ErrClosed
|
||||
}
|
||||
ch := make(chan InterviewNoteEvent, 16)
|
||||
b.subscribers[ch] = struct{}{}
|
||||
b.mu.Unlock()
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
b.mu.Lock()
|
||||
if _, ok := b.subscribers[ch]; ok {
|
||||
delete(b.subscribers, ch)
|
||||
close(ch)
|
||||
}
|
||||
b.mu.Unlock()
|
||||
}()
|
||||
return ch, nil
|
||||
}
|
||||
|
||||
func (b *MemoryBus) Close() error {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if b.closed {
|
||||
return nil
|
||||
}
|
||||
b.closed = true
|
||||
for subscriber := range b.subscribers {
|
||||
close(subscriber)
|
||||
}
|
||||
clear(b.subscribers)
|
||||
return nil
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue