feat: Add profile images
This commit is contained in:
438
toju-app/src/app/store/users/user-avatar.effects.ts
Normal file
438
toju-app/src/app/store/users/user-avatar.effects.ts
Normal file
@@ -0,0 +1,438 @@
|
||||
import { Injectable, inject } from '@angular/core';
|
||||
import {
|
||||
Actions,
|
||||
createEffect,
|
||||
ofType
|
||||
} from '@ngrx/effects';
|
||||
import { Action, Store } from '@ngrx/store';
|
||||
import {
|
||||
EMPTY,
|
||||
from,
|
||||
of
|
||||
} from 'rxjs';
|
||||
import {
|
||||
mergeMap,
|
||||
tap,
|
||||
withLatestFrom
|
||||
} from 'rxjs/operators';
|
||||
import { ProfileAvatarFacade } from '../../domains/profile-avatar';
|
||||
import {
|
||||
ChatEvent,
|
||||
P2P_BASE64_CHUNK_SIZE_BYTES,
|
||||
User,
|
||||
decodeBase64,
|
||||
iterateBlobChunks
|
||||
} from '../../shared-kernel';
|
||||
import { RealtimeSessionFacade } from '../../core/realtime';
|
||||
import { DatabaseService } from '../../infrastructure/persistence';
|
||||
import { UsersActions } from './users.actions';
|
||||
import {
|
||||
selectAllUsers,
|
||||
selectCurrentUser
|
||||
} from './users.selectors';
|
||||
import { selectCurrentRoom, selectSavedRooms } from '../rooms/rooms.selectors';
|
||||
import { RoomsActions } from '../rooms/rooms.actions';
|
||||
import { findRoomMember } from '../rooms/room-members.helpers';
|
||||
|
||||
interface PendingAvatarTransfer {
|
||||
displayName: string;
|
||||
mime: string;
|
||||
oderId: string;
|
||||
total: number;
|
||||
updatedAt: number;
|
||||
username: string;
|
||||
chunks: (string | undefined)[];
|
||||
hash?: string;
|
||||
}
|
||||
|
||||
type AvatarVersionState = Pick<User, 'avatarUrl' | 'avatarHash' | 'avatarUpdatedAt'> | undefined;
|
||||
|
||||
function shouldAcceptAvatarPayload(
|
||||
existingUser: AvatarVersionState,
|
||||
incomingUpdatedAt: number,
|
||||
incomingHash?: string
|
||||
): boolean {
|
||||
const localUpdatedAt = existingUser?.avatarUpdatedAt ?? 0;
|
||||
|
||||
if (incomingUpdatedAt > localUpdatedAt) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (incomingUpdatedAt < localUpdatedAt || incomingUpdatedAt === 0) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!existingUser?.avatarUrl) {
|
||||
return true;
|
||||
}
|
||||
|
||||
return !!incomingHash && incomingHash !== existingUser.avatarHash;
|
||||
}
|
||||
|
||||
export function shouldRequestAvatarData(
|
||||
existingUser: AvatarVersionState,
|
||||
incomingAvatar: Pick<ChatEvent, 'avatarHash' | 'avatarUpdatedAt'>
|
||||
): boolean {
|
||||
return shouldAcceptAvatarPayload(existingUser, incomingAvatar.avatarUpdatedAt ?? 0, incomingAvatar.avatarHash);
|
||||
}
|
||||
|
||||
export function shouldApplyAvatarTransfer(
|
||||
existingUser: AvatarVersionState,
|
||||
transfer: Pick<PendingAvatarTransfer, 'hash' | 'updatedAt'>
|
||||
): boolean {
|
||||
return shouldAcceptAvatarPayload(existingUser, transfer.updatedAt, transfer.hash);
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class UserAvatarEffects {
|
||||
private readonly actions$ = inject(Actions);
|
||||
private readonly store = inject(Store);
|
||||
private readonly webrtc = inject(RealtimeSessionFacade);
|
||||
private readonly db = inject(DatabaseService);
|
||||
private readonly avatars = inject(ProfileAvatarFacade);
|
||||
|
||||
private readonly pendingTransfers = new Map<string, PendingAvatarTransfer>();
|
||||
|
||||
persistCurrentAvatar$ = createEffect(
|
||||
() =>
|
||||
this.actions$.pipe(
|
||||
ofType(UsersActions.updateCurrentUserAvatar),
|
||||
withLatestFrom(this.store.select(selectCurrentUser)),
|
||||
tap(([, currentUser]) => {
|
||||
if (currentUser) {
|
||||
this.db.saveUser(currentUser);
|
||||
}
|
||||
})
|
||||
),
|
||||
{ dispatch: false }
|
||||
);
|
||||
|
||||
persistRemoteAvatar$ = createEffect(
|
||||
() =>
|
||||
this.actions$.pipe(
|
||||
ofType(UsersActions.upsertRemoteUserAvatar),
|
||||
withLatestFrom(this.store.select(selectAllUsers)),
|
||||
tap(([{ user }, allUsers]) => {
|
||||
const mergedUser = allUsers.find((entry) => entry.id === user.id || entry.oderId === user.oderId);
|
||||
const avatarUrl = mergedUser?.avatarUrl ?? user.avatarUrl;
|
||||
|
||||
if (!avatarUrl) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (mergedUser) {
|
||||
this.db.saveUser(mergedUser);
|
||||
}
|
||||
|
||||
void this.avatars.persistAvatarDataUrl({
|
||||
id: mergedUser?.id || user.id,
|
||||
username: mergedUser?.username || user.username,
|
||||
displayName: mergedUser?.displayName || user.displayName
|
||||
}, avatarUrl);
|
||||
})
|
||||
),
|
||||
{ dispatch: false }
|
||||
);
|
||||
|
||||
syncRoomMemberAvatars$ = createEffect(() =>
|
||||
this.actions$.pipe(
|
||||
ofType(UsersActions.updateCurrentUserAvatar, UsersActions.upsertRemoteUserAvatar),
|
||||
withLatestFrom(
|
||||
this.store.select(selectCurrentUser),
|
||||
this.store.select(selectCurrentRoom),
|
||||
this.store.select(selectSavedRooms)
|
||||
),
|
||||
mergeMap(([
|
||||
action,
|
||||
currentUser,
|
||||
currentRoom,
|
||||
savedRooms
|
||||
]) => {
|
||||
const avatarOwner = action.type === UsersActions.updateCurrentUserAvatar.type
|
||||
? currentUser
|
||||
: ('user' in action ? action.user : null);
|
||||
|
||||
if (!avatarOwner) {
|
||||
return EMPTY;
|
||||
}
|
||||
|
||||
const actions = this.buildRoomAvatarActions(avatarOwner, currentRoom, savedRooms);
|
||||
|
||||
return actions.length > 0 ? actions : EMPTY;
|
||||
})
|
||||
)
|
||||
);
|
||||
|
||||
broadcastCurrentAvatarSummary$ = createEffect(
|
||||
() =>
|
||||
this.actions$.pipe(
|
||||
ofType(UsersActions.updateCurrentUserAvatar),
|
||||
withLatestFrom(this.store.select(selectCurrentUser)),
|
||||
tap(([, currentUser]) => {
|
||||
if (!currentUser?.avatarUpdatedAt) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.webrtc.broadcastMessage(this.buildAvatarSummary(currentUser));
|
||||
})
|
||||
),
|
||||
{ dispatch: false }
|
||||
);
|
||||
|
||||
peerConnectedAvatarSummary$ = createEffect(
|
||||
() =>
|
||||
this.webrtc.onPeerConnected.pipe(
|
||||
withLatestFrom(this.store.select(selectCurrentUser)),
|
||||
tap(([peerId, currentUser]) => {
|
||||
if (!currentUser?.avatarUpdatedAt) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.webrtc.sendToPeer(peerId, this.buildAvatarSummary(currentUser));
|
||||
})
|
||||
),
|
||||
{ dispatch: false }
|
||||
);
|
||||
|
||||
incomingAvatarEvents$ = createEffect(() =>
|
||||
this.webrtc.onMessageReceived.pipe(
|
||||
withLatestFrom(this.store.select(selectAllUsers), this.store.select(selectCurrentUser)),
|
||||
mergeMap(([
|
||||
event,
|
||||
allUsers,
|
||||
currentUser
|
||||
]) => {
|
||||
switch (event.type) {
|
||||
case 'user-avatar-summary':
|
||||
return this.handleAvatarSummary(event, allUsers);
|
||||
|
||||
case 'user-avatar-request':
|
||||
return this.handleAvatarRequest(event, currentUser ?? null);
|
||||
|
||||
case 'user-avatar-full':
|
||||
return this.handleAvatarFull(event);
|
||||
|
||||
case 'user-avatar-chunk':
|
||||
return this.handleAvatarChunk(event, allUsers);
|
||||
|
||||
default:
|
||||
return EMPTY;
|
||||
}
|
||||
})
|
||||
)
|
||||
);
|
||||
|
||||
private buildAvatarSummary(user: Pick<User, 'oderId' | 'id' | 'username' | 'displayName' | 'avatarHash' | 'avatarMime' | 'avatarUpdatedAt'>): ChatEvent {
|
||||
return {
|
||||
type: 'user-avatar-summary',
|
||||
oderId: user.oderId || user.id,
|
||||
username: user.username,
|
||||
displayName: user.displayName,
|
||||
avatarHash: user.avatarHash,
|
||||
avatarMime: user.avatarMime,
|
||||
avatarUpdatedAt: user.avatarUpdatedAt || 0
|
||||
};
|
||||
}
|
||||
|
||||
private handleAvatarSummary(event: ChatEvent, allUsers: User[]) {
|
||||
if (!event.fromPeerId || !event.oderId || !event.avatarUpdatedAt) {
|
||||
return EMPTY;
|
||||
}
|
||||
|
||||
const existingUser = allUsers.find((user) => user.id === event.oderId || user.oderId === event.oderId);
|
||||
|
||||
if (!shouldRequestAvatarData(existingUser, event)) {
|
||||
return EMPTY;
|
||||
}
|
||||
|
||||
this.webrtc.sendToPeer(event.fromPeerId, {
|
||||
type: 'user-avatar-request',
|
||||
oderId: event.oderId
|
||||
});
|
||||
|
||||
return EMPTY;
|
||||
}
|
||||
|
||||
private handleAvatarRequest(event: ChatEvent, currentUser: User | null) {
|
||||
const currentUserKey = currentUser?.oderId || currentUser?.id;
|
||||
|
||||
if (!event.fromPeerId || !currentUser || !currentUserKey || event.oderId !== currentUserKey || !currentUser.avatarUrl) {
|
||||
return EMPTY;
|
||||
}
|
||||
|
||||
return from(this.sendAvatarToPeer(event.fromPeerId, currentUser)).pipe(mergeMap(() => EMPTY));
|
||||
}
|
||||
|
||||
private handleAvatarFull(event: ChatEvent) {
|
||||
if (!event.oderId || !event.avatarMime || typeof event.total !== 'number' || event.total < 1) {
|
||||
return EMPTY;
|
||||
}
|
||||
|
||||
this.pendingTransfers.set(event.oderId, {
|
||||
chunks: new Array<string | undefined>(event.total),
|
||||
displayName: event.displayName || 'User',
|
||||
mime: event.avatarMime,
|
||||
oderId: event.oderId,
|
||||
total: event.total,
|
||||
updatedAt: event.avatarUpdatedAt || Date.now(),
|
||||
username: event.username || (event.displayName || 'User').toLowerCase().replace(/\s+/g, '_'),
|
||||
hash: event.avatarHash
|
||||
});
|
||||
|
||||
return EMPTY;
|
||||
}
|
||||
|
||||
private handleAvatarChunk(event: ChatEvent, allUsers: User[]) {
|
||||
if (!event.oderId || typeof event.index !== 'number' || typeof event.total !== 'number' || typeof event.data !== 'string') {
|
||||
return EMPTY;
|
||||
}
|
||||
|
||||
const transfer = this.pendingTransfers.get(event.oderId);
|
||||
|
||||
if (!transfer || transfer.total !== event.total || event.index < 0 || event.index >= transfer.total) {
|
||||
return EMPTY;
|
||||
}
|
||||
|
||||
transfer.chunks[event.index] = event.data;
|
||||
|
||||
if (transfer.chunks.some((chunk) => !chunk)) {
|
||||
return EMPTY;
|
||||
}
|
||||
|
||||
this.pendingTransfers.delete(event.oderId);
|
||||
|
||||
return from(this.buildRemoteAvatarAction(transfer, allUsers)).pipe(
|
||||
mergeMap((action) => action ? of(action) : EMPTY)
|
||||
);
|
||||
}
|
||||
|
||||
private async buildRemoteAvatarAction(transfer: PendingAvatarTransfer, allUsers: User[]): Promise<Action | null> {
|
||||
const existingUser = allUsers.find((user) => user.id === transfer.oderId || user.oderId === transfer.oderId);
|
||||
|
||||
if (!shouldApplyAvatarTransfer(existingUser, transfer)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const blob = new Blob(transfer.chunks.map((chunk) => this.decodeBase64ToArrayBuffer(chunk!)), { type: transfer.mime });
|
||||
const dataUrl = await this.readBlobAsDataUrl(blob);
|
||||
|
||||
return UsersActions.upsertRemoteUserAvatar({
|
||||
user: {
|
||||
id: existingUser?.id || transfer.oderId,
|
||||
oderId: existingUser?.oderId || transfer.oderId,
|
||||
username: existingUser?.username || transfer.username,
|
||||
displayName: existingUser?.displayName || transfer.displayName,
|
||||
avatarUrl: dataUrl,
|
||||
avatarHash: transfer.hash,
|
||||
avatarMime: transfer.mime,
|
||||
avatarUpdatedAt: transfer.updatedAt
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private buildRoomAvatarActions(
|
||||
avatarOwner: Pick<User, 'id' | 'oderId' | 'avatarUrl' | 'avatarHash' | 'avatarMime' | 'avatarUpdatedAt'>,
|
||||
currentRoom: ReturnType<typeof selectCurrentRoom['projector']> | null,
|
||||
savedRooms: ReturnType<typeof selectSavedRooms['projector']>
|
||||
): Action[] {
|
||||
const rooms = [currentRoom, ...savedRooms.filter((room) => room.id !== currentRoom?.id)].filter((room): room is NonNullable<typeof currentRoom> => !!room);
|
||||
const roomActions: Action[] = [];
|
||||
const avatarOwnerId = avatarOwner.oderId || avatarOwner.id;
|
||||
|
||||
for (const room of rooms) {
|
||||
const member = findRoomMember(room.members ?? [], avatarOwnerId);
|
||||
|
||||
if (!member) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const nextMembers = (room.members ?? []).map((roomMember) => {
|
||||
if (roomMember.id !== member.id && roomMember.oderId !== member.oderId) {
|
||||
return roomMember;
|
||||
}
|
||||
|
||||
return {
|
||||
...roomMember,
|
||||
avatarUrl: avatarOwner.avatarUrl,
|
||||
avatarHash: avatarOwner.avatarHash,
|
||||
avatarMime: avatarOwner.avatarMime,
|
||||
avatarUpdatedAt: avatarOwner.avatarUpdatedAt
|
||||
};
|
||||
});
|
||||
|
||||
roomActions.push(RoomsActions.updateRoom({
|
||||
roomId: room.id,
|
||||
changes: { members: nextMembers }
|
||||
}));
|
||||
}
|
||||
|
||||
return roomActions;
|
||||
}
|
||||
|
||||
private async sendAvatarToPeer(targetPeerId: string, user: User): Promise<void> {
|
||||
if (!user.avatarUrl) {
|
||||
return;
|
||||
}
|
||||
|
||||
const blob = await this.dataUrlToBlob(user.avatarUrl, user.avatarMime || 'image/webp');
|
||||
const total = Math.ceil(blob.size / P2P_BASE64_CHUNK_SIZE_BYTES);
|
||||
const userKey = user.oderId || user.id;
|
||||
|
||||
this.webrtc.sendToPeer(targetPeerId, {
|
||||
type: 'user-avatar-full',
|
||||
oderId: userKey,
|
||||
username: user.username,
|
||||
displayName: user.displayName,
|
||||
avatarHash: user.avatarHash,
|
||||
avatarMime: user.avatarMime || blob.type || 'image/webp',
|
||||
avatarUpdatedAt: user.avatarUpdatedAt || Date.now(),
|
||||
total
|
||||
});
|
||||
|
||||
for await (const chunk of iterateBlobChunks(blob, P2P_BASE64_CHUNK_SIZE_BYTES)) {
|
||||
await this.webrtc.sendToPeerBuffered(targetPeerId, {
|
||||
type: 'user-avatar-chunk',
|
||||
oderId: userKey,
|
||||
avatarHash: user.avatarHash,
|
||||
avatarMime: user.avatarMime || blob.type || 'image/webp',
|
||||
avatarUpdatedAt: user.avatarUpdatedAt || Date.now(),
|
||||
index: chunk.index,
|
||||
total: chunk.total,
|
||||
data: chunk.base64
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private dataUrlToBlob(dataUrl: string, mimeType: string): Promise<Blob> {
|
||||
const base64 = dataUrl.split(',', 2)[1] ?? '';
|
||||
|
||||
return Promise.resolve(new Blob([this.decodeBase64ToArrayBuffer(base64)], { type: mimeType }));
|
||||
}
|
||||
|
||||
private decodeBase64ToArrayBuffer(base64: string): ArrayBuffer {
|
||||
const decodedBytes = decodeBase64(base64);
|
||||
|
||||
return decodedBytes.buffer.slice(
|
||||
decodedBytes.byteOffset,
|
||||
decodedBytes.byteOffset + decodedBytes.byteLength
|
||||
) as ArrayBuffer;
|
||||
}
|
||||
|
||||
private readBlobAsDataUrl(blob: Blob): Promise<string> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const reader = new FileReader();
|
||||
|
||||
reader.onload = () => {
|
||||
if (typeof reader.result === 'string') {
|
||||
resolve(reader.result);
|
||||
return;
|
||||
}
|
||||
|
||||
reject(new Error('Failed to encode avatar image'));
|
||||
};
|
||||
|
||||
reader.onerror = () => reject(reader.error ?? new Error('Failed to read avatar image'));
|
||||
reader.readAsDataURL(blob);
|
||||
});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user