feat(#195): SSE realtime Cartes (stream + emit audience)

GET /api/v1/cards/stream (Bearer ou ?access_token=).
Events card.created|updated|deleted, response.added + heartbeat.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
2026-09-24 11:50:30 +02:00
co-authored by Cursor
parent dfac075a74
commit 823ea6cd22
8 changed files with 313 additions and 8 deletions
@@ -0,0 +1,94 @@
import {
Controller,
Get,
Headers,
MessageEvent,
Query,
Sse,
UnauthorizedException,
UseGuards,
} from '@nestjs/common';
import {
ApiBearerAuth,
ApiOperation,
ApiQuery,
ApiTags,
} from '@nestjs/swagger';
import { ConfigService } from '@nestjs/config';
import { JwtService } from '@nestjs/jwt';
import { Observable } from 'rxjs';
import { AuthGuard } from 'src/common/guards/auth.guard';
import { RolesGuard } from 'src/common/guards/roles.guard';
import { Roles } from 'src/common/decorators/roles.decorator';
import { Public } from 'src/common/decorators/public.decorator';
import { User } from 'src/common/decorators/user.decorator';
import { RoleType } from 'src/entities/users.entity';
import { CardsRealtimeService } from './cards-realtime.service';
/**
* SSE Cartes — #195
* Auth : Bearer (recommandé) ou `?access_token=` (EventSource navigateur).
*/
@ApiTags('Cartes')
@Controller('cards')
export class CardsRealtimeController {
constructor(
private readonly realtime: CardsRealtimeService,
private readonly jwtService: JwtService,
private readonly configService: ConfigService,
) {}
@Sse('stream')
@Public()
@ApiOperation({
summary: 'Flux SSE bulles (card.created|updated|deleted, response.added) — #195',
})
@ApiQuery({
name: 'access_token',
required: false,
description: 'JWT si pas de header Authorization (EventSource)',
})
@ApiBearerAuth('access-token')
async stream(
@Headers('authorization') authorization?: string,
@Query('access_token') accessToken?: string,
): Promise<Observable<MessageEvent>> {
const userId = await this.resolveUserId(authorization, accessToken);
return this.realtime.streamFor(userId);
}
/** Variante gardée (clients qui envoient Bearer correctement). */
@Get('stream/info')
@UseGuards(AuthGuard, RolesGuard)
@Roles(RoleType.PARENT, RoleType.ASSISTANTE_MATERNELLE)
@ApiBearerAuth('access-token')
@ApiOperation({ summary: 'Debug : abonnés SSE pour mon user — #195' })
info(@User('id') userId: string) {
return {
user_id: userId,
subscribers: this.realtime.subscriberCount(userId),
};
}
private async resolveUserId(
authorization?: string,
accessToken?: string,
): Promise<string> {
let token = accessToken?.trim();
if (!token && authorization?.startsWith('Bearer ')) {
token = authorization.slice(7).trim();
}
if (!token) {
throw new UnauthorizedException('Token manquant (Bearer ou access_token)');
}
try {
const payload = await this.jwtService.verifyAsync<{ sub: string }>(token, {
secret: this.configService.get<string>('jwt.accessSecret'),
});
if (!payload?.sub) throw new UnauthorizedException('Token invalide');
return payload.sub;
} catch {
throw new UnauthorizedException('Token invalide ou expiré');
}
}
}
@@ -0,0 +1,47 @@
import { MessageEvent } from '@nestjs/common';
import { CardsRealtimeService } from './cards-realtime.service';
describe('CardsRealtimeService (#195)', () => {
let service: CardsRealtimeService;
beforeEach(() => {
service = new CardsRealtimeService();
});
afterEach(() => {
service.onModuleDestroy();
});
it('émet card.created aux abonnés du user', async () => {
const events: MessageEvent[] = [];
const sub = service.streamFor('user-a').subscribe((e) => events.push(e));
await new Promise((r) => setTimeout(r, 20));
expect(events[0]?.type).toBe('heartbeat');
expect(service.subscriberCount('user-a')).toBe(1);
service.emitToUsers(['user-a', 'user-b'], 'card.created', 'card-1', {
id: 'card-1',
});
await new Promise((r) => setTimeout(r, 20));
const created = events.find((e) => e.type === 'card.created');
expect(created).toBeDefined();
expect((created!.data as { card_id: string }).card_id).toBe('card-1');
expect(service.subscriberCount('user-b')).toBe(0);
sub.unsubscribe();
expect(service.subscriberCount('user-a')).toBe(0);
});
it('ne diffuse pas aux users non abonnés', async () => {
const events: MessageEvent[] = [];
const sub = service.streamFor('user-a').subscribe((e) => events.push(e));
await new Promise((r) => setTimeout(r, 20));
const before = events.length;
service.emitToUsers(['user-b'], 'card.updated', 'x');
await new Promise((r) => setTimeout(r, 20));
expect(events.length).toBe(before);
sub.unsubscribe();
});
});
@@ -0,0 +1,111 @@
import { Injectable, MessageEvent, OnModuleDestroy } from '@nestjs/common';
import { Observable, Subject, interval, merge, takeUntil } from 'rxjs';
import { map } from 'rxjs/operators';
export type CardRealtimeEventType =
| 'card.created'
| 'card.updated'
| 'card.deleted'
| 'response.added'
| 'heartbeat';
export interface CardRealtimePayload {
event: CardRealtimeEventType;
card_id?: string;
data?: unknown;
at: string;
}
/**
* Bus SSE in-memory par utilisateur (V1 mono-instance).
* Rooms = userId (audience carte).
*/
@Injectable()
export class CardsRealtimeService implements OnModuleDestroy {
private readonly byUser = new Map<string, Set<Subject<CardRealtimePayload>>>();
private readonly destroy$ = new Subject<void>();
onModuleDestroy(): void {
this.destroy$.next();
this.destroy$.complete();
for (const set of this.byUser.values()) {
for (const s of set) s.complete();
}
this.byUser.clear();
}
/** Flux SSE pour un utilisateur authentifié. */
streamFor(userId: string): Observable<MessageEvent> {
const subject = new Subject<CardRealtimePayload>();
let set = this.byUser.get(userId);
if (!set) {
set = new Set();
this.byUser.set(userId, set);
}
set.add(subject);
const heartbeat$ = interval(25_000).pipe(
map(
(): CardRealtimePayload => ({
event: 'heartbeat',
at: new Date().toISOString(),
}),
),
);
return new Observable<MessageEvent>((observer) => {
const sub = merge(subject, heartbeat$)
.pipe(takeUntil(this.destroy$))
.subscribe({
next: (payload) =>
observer.next({
type: payload.event,
data: payload,
} as MessageEvent),
error: (err) => observer.error(err),
complete: () => observer.complete(),
});
// ping initial
subject.next({
event: 'heartbeat',
at: new Date().toISOString(),
});
return () => {
sub.unsubscribe();
set!.delete(subject);
subject.complete();
if (set!.size === 0) this.byUser.delete(userId);
};
});
}
emitToUsers(
userIds: string[],
event: Exclude<CardRealtimeEventType, 'heartbeat'>,
cardId: string | undefined,
data?: unknown,
): void {
const unique = [...new Set(userIds.filter(Boolean))];
const payload: CardRealtimePayload = {
event,
card_id: cardId,
data,
at: new Date().toISOString(),
};
for (const uid of unique) {
const set = this.byUser.get(uid);
if (!set) continue;
for (const s of set) s.next(payload);
}
}
/** Test / debug : nombre d’abonnés actifs. */
subscriberCount(userId?: string): number {
if (userId) return this.byUser.get(userId)?.size ?? 0;
let n = 0;
for (const s of this.byUser.values()) n += s.size;
return n;
}
}
+5 -3
View File
@@ -10,7 +10,9 @@ import { CardResponse } from 'src/entities/card_responses.entity';
import { AmChildren } from 'src/entities/am_children.entity';
import { ParentsChildren } from 'src/entities/parents_children.entity';
import { CardsController } from './cards.controller';
import { CardsRealtimeController } from './cards-realtime.controller';
import { CardsService } from './cards.service';
import { CardsRealtimeService } from './cards-realtime.service';
@Module({
imports: [
@@ -32,8 +34,8 @@ import { CardsService } from './cards.service';
inject: [ConfigService],
}),
],
controllers: [CardsController],
providers: [CardsService],
exports: [CardsService],
controllers: [CardsController, CardsRealtimeController],
providers: [CardsService, CardsRealtimeService],
exports: [CardsService, CardsRealtimeService],
})
export class CardsModule {}
@@ -2,6 +2,7 @@ import { Test, TestingModule } from '@nestjs/testing';
import { getRepositoryToken } from '@nestjs/typeorm';
import { ForbiddenException } from '@nestjs/common';
import { CardsService } from './cards.service';
import { CardsRealtimeService } from './cards-realtime.service';
import { AbsencesGardeService } from '../absences-garde/absences-garde.service';
import { CardType, CardResponseModeType } from 'src/entities/card_types.entity';
import {
@@ -32,6 +33,7 @@ describe('CardsService (#194)', () => {
create: jest.fn((x) => x),
save: jest.fn(),
findOne: jest.fn(),
find: jest.fn(),
};
const responsesRepo = {
create: jest.fn((x) => x),
@@ -44,6 +46,7 @@ describe('CardsService (#194)', () => {
maj: jest.fn(),
supprimer: jest.fn(),
};
const realtime = { emitToUsers: jest.fn() };
beforeEach(async () => {
const module: TestingModule = await Test.createTestingModule({
@@ -56,6 +59,7 @@ describe('CardsService (#194)', () => {
{ provide: getRepositoryToken(AmChildren), useValue: amChildrenRepo },
{ provide: getRepositoryToken(ParentsChildren), useValue: parentsChildrenRepo },
{ provide: AbsencesGardeService, useValue: absencesService },
{ provide: CardsRealtimeService, useValue: realtime },
],
}).compile();
service = module.get(CardsService);
@@ -112,7 +116,11 @@ describe('CardsService (#194)', () => {
retention_days: 14,
},
responses: [],
audience: [],
audience: [
{ id_utilisateur: 'am-1' },
{ id_utilisateur: 'p-1' },
{ id_utilisateur: 'p-2' },
],
});
const res = await service.creer('am-1', RoleType.ASSISTANTE_MATERNELLE, {
@@ -124,6 +132,12 @@ describe('CardsService (#194)', () => {
expect(absencesService.creer).toHaveBeenCalled();
expect(audienceRepo.save).toHaveBeenCalled();
expect(realtime.emitToUsers).toHaveBeenCalledWith(
['am-1', 'p-1', 'p-2'],
'card.created',
'card-1',
expect.any(Object),
);
expect(res.type_code).toBe('conge_am');
expect(res.statut).toBe(CardInstanceStatutType.OUVERTE);
});
+39 -3
View File
@@ -32,6 +32,7 @@ import {
MajCarteDto,
RepondreCarteDto,
} from './dto/cards.dto';
import { CardsRealtimeService } from './cards-realtime.service';
@Injectable()
export class CardsService {
@@ -49,6 +50,7 @@ export class CardsService {
@InjectRepository(ParentsChildren)
private readonly parentsChildrenRepo: Repository<ParentsChildren>,
private readonly absencesService: AbsencesGardeService,
private readonly realtime: CardsRealtimeService,
) {}
async listerTypes(role: RoleType): Promise<CardType[]> {
@@ -164,7 +166,9 @@ export class CardsService {
await this.buildAudience(saved, type, placement, userId, role);
const full = await this.loadCard(saved.id);
return this.toDto(full, userId);
const dtoOut = this.toDto(full, userId);
this.emitAudience(full, 'card.created', dtoOut);
return dtoOut;
}
async repondre(
@@ -228,7 +232,14 @@ export class CardsService {
card.purge_at = this.purgeAt(card.type.retention_days, card.statut);
await this.cardsRepo.save(card);
return this.toDto(await this.loadCard(cardId), userId);
const full = await this.loadCard(cardId);
const dtoOut = this.toDto(full, userId);
this.emitAudience(full, 'response.added', {
action: dto.action,
card: dtoOut,
});
this.emitAudience(full, 'card.updated', dtoOut);
return dtoOut;
}
async majEnAttente(
@@ -273,7 +284,10 @@ export class CardsService {
card.statut = CardInstanceStatutType.OUVERTE;
card.purge_at = this.purgeAt(card.type.retention_days, card.statut);
await this.cardsRepo.save(card);
return this.toDto(await this.loadCard(cardId), userId);
const full = await this.loadCard(cardId);
const dtoOut = this.toDto(full, userId);
this.emitAudience(full, 'card.updated', dtoOut);
return dtoOut;
}
async supprimer(
@@ -285,6 +299,7 @@ export class CardsService {
if (card.cree_par !== userId) {
throw new ForbiddenException('Seul le créateur peut supprimer cette carte');
}
const audienceIds = await this.audienceUserIds(card);
if (card.id_absence) {
const absStatut =
card.statut === CardInstanceStatutType.TRAITEE
@@ -300,6 +315,27 @@ export class CardsService {
void absStatut;
}
await this.cardsRepo.delete({ id: cardId });
this.realtime.emitToUsers(audienceIds, 'card.deleted', cardId, {
id: cardId,
});
}
private emitAudience(
card: CardInstance,
event: 'card.created' | 'card.updated' | 'response.added',
data: unknown,
): void {
const ids = (card.audience ?? []).map((a) => a.id_utilisateur);
if (ids.length === 0) return;
this.realtime.emitToUsers(ids, event, card.id, data);
}
private async audienceUserIds(card: CardInstance): Promise<string[]> {
if (card.audience?.length) {
return card.audience.map((a) => a.id_utilisateur);
}
const rows = await this.audienceRepo.find({ where: { id_card: card.id } });
return rows.map((r) => r.id_utilisateur);
}
private mapAbsenceType(typeCode: string): TypeAbsenceGardeType {
+1
View File
@@ -1,2 +1,3 @@
export { CardsModule } from './cards.module';
export { CardsService } from './cards.service';
export { CardsRealtimeService } from './cards-realtime.service';