Add background-sync service
This commit is contained in:
Sosthene
parent
1408b2ddf3
commit
34a65ec079
14
CHANGELOG.md
14
CHANGELOG.md
@ -4,6 +4,20 @@ Toutes les modifications notables de ce projet seront documentées ici.
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Added
|
||||
- Service de Background Sync pour la détection automatique des données manquantes
|
||||
- Surveillance périodique des `pcd_commitment` (toutes les 30 secondes)
|
||||
- API WebSocket pour contrôler le background sync (`FORCE_DATA_SCAN`, `GET_BACKGROUND_SYNC_STATUS`)
|
||||
- Gestion automatique des entrées `diff` pour le tracking des données manquantes
|
||||
- Récupération automatique des données manquantes auprès des pairs
|
||||
- Scripts de test et validation du service de background sync
|
||||
- Documentation complète dans `docs/BACKGROUND_SYNC.md`
|
||||
|
||||
### Changed
|
||||
- Intégration du background sync dans le service principal
|
||||
- Démarrage automatique du background sync avec le serveur
|
||||
- Arrêt propre du background sync lors de l'arrêt du serveur
|
||||
|
||||
## [0.1.1] - 2025-08-26
|
||||
- Bump version package.json à 0.1.1
|
||||
- Documentation déploiement mise à jour (exemples tag)
|
||||
|
||||
352
src/background-sync.service.ts
Normal file
352
src/background-sync.service.ts
Normal file
@ -0,0 +1,352 @@
|
||||
// Service de surveillance en arrière-plan pour Node.js
|
||||
// Équivalent du service worker pour la détection des données manquantes
|
||||
|
||||
import { Service } from './service';
|
||||
import Database from './database.service';
|
||||
import { config } from './config';
|
||||
import { retrieveData } from './storage.service';
|
||||
|
||||
export class BackgroundSyncService {
|
||||
private static instance: BackgroundSyncService;
|
||||
private service: Service;
|
||||
private db: Database | null = null;
|
||||
private scanInterval: NodeJS.Timeout | null = null;
|
||||
private isRunning: boolean = false;
|
||||
private readonly SCAN_INTERVAL_MS = 30000; // 30 secondes
|
||||
private readonly EMPTY32BYTES = '0'.repeat(64);
|
||||
|
||||
private constructor() {
|
||||
this.service = Service.getInstance();
|
||||
console.log('🔧 BackgroundSyncService initialized');
|
||||
}
|
||||
|
||||
static async getInstance(): Promise<BackgroundSyncService> {
|
||||
if (!BackgroundSyncService.instance) {
|
||||
BackgroundSyncService.instance = new BackgroundSyncService();
|
||||
await BackgroundSyncService.instance.init();
|
||||
}
|
||||
return BackgroundSyncService.instance;
|
||||
}
|
||||
|
||||
private async init(): Promise<void> {
|
||||
this.db = await Database.getInstance();
|
||||
}
|
||||
|
||||
/**
|
||||
* Démarre le service de surveillance en arrière-plan
|
||||
*/
|
||||
public async start(): Promise<void> {
|
||||
if (this.isRunning) {
|
||||
console.log('⚠️ BackgroundSyncService already running');
|
||||
return;
|
||||
}
|
||||
|
||||
console.log('🚀 Starting background sync service...');
|
||||
this.isRunning = true;
|
||||
|
||||
// Scan immédiat au démarrage
|
||||
await this.scanMissingData();
|
||||
|
||||
// Puis scan périodique
|
||||
this.scanInterval = setInterval(async () => {
|
||||
try {
|
||||
await this.scanMissingData();
|
||||
} catch (error) {
|
||||
console.error('❌ Error in background scan:', error);
|
||||
}
|
||||
}, this.SCAN_INTERVAL_MS);
|
||||
|
||||
console.log('✅ Background sync service started');
|
||||
}
|
||||
|
||||
/**
|
||||
* Arrête le service de surveillance
|
||||
*/
|
||||
public stop(): void {
|
||||
if (!this.isRunning) {
|
||||
return;
|
||||
}
|
||||
|
||||
console.log('🛑 Stopping background sync service...');
|
||||
this.isRunning = false;
|
||||
|
||||
if (this.scanInterval) {
|
||||
clearInterval(this.scanInterval);
|
||||
this.scanInterval = null;
|
||||
}
|
||||
|
||||
console.log('✅ Background sync service stopped');
|
||||
}
|
||||
|
||||
/**
|
||||
* Scan manuel des données manquantes (équivalent à scanMissingData du service worker)
|
||||
*/
|
||||
public async scanMissingData(): Promise<string[]> {
|
||||
console.log('🔍 Scanning for missing data...');
|
||||
|
||||
try {
|
||||
const myProcesses = await this.getMyProcesses();
|
||||
if (!myProcesses || myProcesses.length === 0) {
|
||||
console.log('No processes to scan');
|
||||
return [];
|
||||
}
|
||||
|
||||
const toDownload = new Set<string>();
|
||||
|
||||
for (const processId of myProcesses) {
|
||||
const process = await this.service.getProcess(processId);
|
||||
if (!process) continue;
|
||||
|
||||
for (const state of process.states) {
|
||||
if (state.state_id === this.EMPTY32BYTES) continue;
|
||||
|
||||
// Vérifier chaque pcd_commitment
|
||||
for (const [field, hash] of Object.entries(state.pcd_commitment)) {
|
||||
// Ignorer les champs publics
|
||||
if (state.public_data[field] !== undefined || field === 'roles') continue;
|
||||
|
||||
// Vérifier que hash est une string
|
||||
if (typeof hash !== 'string') continue;
|
||||
|
||||
// Vérifier si on a déjà les données
|
||||
const existingData = await this.getBlob(hash);
|
||||
if (!existingData) {
|
||||
toDownload.add(hash);
|
||||
// Ajouter une entrée diff si elle n'existe pas
|
||||
await this.addDiff(processId, state.state_id, hash, state.roles, field);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const missingHashes = Array.from(toDownload);
|
||||
|
||||
if (missingHashes.length > 0) {
|
||||
console.log(`📥 Found ${missingHashes.length} missing data hashes:`, missingHashes);
|
||||
await this.requestMissingData(missingHashes);
|
||||
} else {
|
||||
console.log('✅ No missing data found');
|
||||
}
|
||||
|
||||
return missingHashes;
|
||||
} catch (error) {
|
||||
console.error('❌ Error scanning missing data:', error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Récupère les processus de l'utilisateur
|
||||
*/
|
||||
private async getMyProcesses(): Promise<string[]> {
|
||||
try {
|
||||
return await this.service.getMyProcesses() || [];
|
||||
} catch (error) {
|
||||
console.error('Error getting my processes:', error);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Vérifie si un blob existe en base
|
||||
*/
|
||||
private async getBlob(hash: string): Promise<Buffer | null> {
|
||||
try {
|
||||
return await this.service.getBufferFromDb(hash);
|
||||
} catch (error) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Ajoute une entrée diff pour tracking
|
||||
*/
|
||||
private async addDiff(
|
||||
processId: string,
|
||||
stateId: string,
|
||||
hash: string,
|
||||
roles: any,
|
||||
field: string
|
||||
): Promise<void> {
|
||||
try {
|
||||
if (!this.db) {
|
||||
console.error('Database not initialized');
|
||||
return;
|
||||
}
|
||||
|
||||
const existingDiff = await this.db.getObject('diffs', hash);
|
||||
|
||||
if (!existingDiff) {
|
||||
const newDiff = {
|
||||
process_id: processId,
|
||||
state_id: stateId,
|
||||
value_commitment: hash,
|
||||
roles: roles,
|
||||
field: field,
|
||||
description: null,
|
||||
previous_value: null,
|
||||
new_value: null,
|
||||
notify_user: false,
|
||||
need_validation: false,
|
||||
validation_status: 'None'
|
||||
};
|
||||
|
||||
await this.db.addObject({
|
||||
storeName: 'diffs',
|
||||
object: newDiff,
|
||||
key: hash
|
||||
});
|
||||
|
||||
console.log(`📝 Added diff entry for hash: ${hash}`);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Error adding diff:', error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Demande les données manquantes aux pairs
|
||||
*/
|
||||
private async requestMissingData(hashes: string[]): Promise<void> {
|
||||
try {
|
||||
console.log('🔄 Requesting missing data from peers...');
|
||||
|
||||
// Récupérer tous les processus pour déterminer les rôles
|
||||
const myProcesses = await this.getMyProcesses();
|
||||
const processesToRequest: Record<string, any> = {};
|
||||
|
||||
for (const processId of myProcesses) {
|
||||
const process = await this.service.getProcess(processId);
|
||||
if (process) {
|
||||
processesToRequest[processId] = process;
|
||||
}
|
||||
}
|
||||
|
||||
// Pour chaque hash manquant, essayer de le récupérer
|
||||
for (const hash of hashes) {
|
||||
try {
|
||||
// Trouver le diff correspondant
|
||||
const diffs = await this.service.getDiffsFromDb();
|
||||
const diff = Object.values(diffs).find((d: any) => d.value_commitment === hash);
|
||||
|
||||
if (diff) {
|
||||
const processId = diff.process_id;
|
||||
const stateId = diff.state_id;
|
||||
const roles = diff.roles;
|
||||
|
||||
if (processesToRequest[processId]) {
|
||||
// D'abord essayer de récupérer depuis les serveurs de stockage
|
||||
const retrievedFromStorage = await this.tryRetrieveFromStorage(hash, roles);
|
||||
|
||||
if (retrievedFromStorage) {
|
||||
console.log(`✅ Data retrieved from storage for hash ${hash}`);
|
||||
// Sauvegarder les données récupérées en base
|
||||
await this.service.saveBufferToDb(hash, Buffer.from(retrievedFromStorage));
|
||||
continue; // Passer au hash suivant
|
||||
}
|
||||
|
||||
// Si pas trouvé en storage, demander aux pairs
|
||||
console.log(`🔄 Requesting data for hash ${hash} from process ${processId}`);
|
||||
await this.service.requestDataFromPeers(processId, [stateId], [roles]);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`Error requesting data for hash ${hash}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
} catch (error) {
|
||||
console.error('Error requesting missing data:', error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Essaie de récupérer les données depuis les serveurs de stockage
|
||||
*/
|
||||
private async tryRetrieveFromStorage(hash: string, roles: any): Promise<ArrayBuffer | null> {
|
||||
try {
|
||||
// Extraire les URLs de stockage depuis les rôles
|
||||
const storageUrls = this.extractStorageUrls(roles);
|
||||
|
||||
if (storageUrls.length === 0) {
|
||||
console.log(`No storage URLs found for hash ${hash}`);
|
||||
return null;
|
||||
}
|
||||
|
||||
console.log(`🔍 Trying to retrieve hash ${hash} from storage servers:`, storageUrls);
|
||||
|
||||
// Essayer de récupérer depuis les serveurs de stockage
|
||||
const data = await retrieveData(storageUrls, hash);
|
||||
|
||||
if (data) {
|
||||
console.log(`✅ Successfully retrieved data for hash ${hash} from storage`);
|
||||
return data;
|
||||
} else {
|
||||
console.log(`❌ Data not found in storage for hash ${hash}`);
|
||||
return null;
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`Error retrieving data from storage for hash ${hash}:`, error);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Extrait les URLs de stockage depuis les rôles
|
||||
*/
|
||||
private extractStorageUrls(roles: any): string[] {
|
||||
const storageUrls = new Set<string>();
|
||||
|
||||
try {
|
||||
if (roles && typeof roles === 'object') {
|
||||
for (const role of Object.values(roles)) {
|
||||
if (role && typeof role === 'object' && 'storages' in role && Array.isArray((role as any).storages)) {
|
||||
for (const storageUrl of (role as any).storages) {
|
||||
if (typeof storageUrl === 'string' && storageUrl.trim()) {
|
||||
storageUrls.add(storageUrl.trim());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Error extracting storage URLs from roles:', error);
|
||||
}
|
||||
|
||||
return Array.from(storageUrls);
|
||||
}
|
||||
|
||||
/**
|
||||
* Force un scan immédiat (pour tests ou usage manuel)
|
||||
*/
|
||||
public async forceScan(): Promise<string[]> {
|
||||
console.log('🔄 Forcing immediate scan...');
|
||||
return await this.scanMissingData();
|
||||
}
|
||||
|
||||
/**
|
||||
* Obtient le statut du service
|
||||
*/
|
||||
public getStatus(): { isRunning: boolean; scanInterval: number } {
|
||||
return {
|
||||
isRunning: this.isRunning,
|
||||
scanInterval: this.SCAN_INTERVAL_MS
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Met à jour l'intervalle de scan
|
||||
*/
|
||||
public setScanInterval(intervalMs: number): void {
|
||||
if (this.isRunning && this.scanInterval) {
|
||||
clearInterval(this.scanInterval);
|
||||
this.scanInterval = setInterval(async () => {
|
||||
try {
|
||||
await this.scanMissingData();
|
||||
} catch (error) {
|
||||
console.error('❌ Error in background scan:', error);
|
||||
}
|
||||
}, intervalMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -37,6 +37,11 @@ export enum MessageType {
|
||||
// Account management
|
||||
ADD_DEVICE = 'ADD_DEVICE',
|
||||
DEVICE_ADDED = 'DEVICE_ADDED',
|
||||
// Background sync
|
||||
FORCE_DATA_SCAN = 'FORCE_DATA_SCAN',
|
||||
DATA_SCAN_RESULT = 'DATA_SCAN_RESULT',
|
||||
GET_BACKGROUND_SYNC_STATUS = 'GET_BACKGROUND_SYNC_STATUS',
|
||||
BACKGROUND_SYNC_STATUS = 'BACKGROUND_SYNC_STATUS',
|
||||
}
|
||||
|
||||
// Re-export AnkFlag from WASM for relay message typing
|
||||
|
||||
@ -16,6 +16,7 @@ export class Service {
|
||||
private membersList: any = {};
|
||||
private relayManager: RelayManager;
|
||||
private storages: string[] = []; // storage urls
|
||||
private backgroundSync: any = null; // BackgroundSyncService
|
||||
|
||||
private constructor() {
|
||||
console.log('🔧 Service initialized');
|
||||
@ -1448,4 +1449,54 @@ export class Service {
|
||||
throw new Error(`Failed to dump device: ${e}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Démarre le service de surveillance en arrière-plan
|
||||
*/
|
||||
public async startBackgroundSync(): Promise<void> {
|
||||
if (!this.backgroundSync) {
|
||||
const { BackgroundSyncService } = await import('./background-sync.service');
|
||||
this.backgroundSync = await BackgroundSyncService.getInstance();
|
||||
}
|
||||
await this.backgroundSync.start();
|
||||
}
|
||||
|
||||
/**
|
||||
* Arrête le service de surveillance
|
||||
*/
|
||||
public stopBackgroundSync(): void {
|
||||
if (this.backgroundSync) {
|
||||
this.backgroundSync.stop();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Force un scan manuel des données manquantes
|
||||
*/
|
||||
public async forceDataScan(): Promise<string[]> {
|
||||
if (!this.backgroundSync) {
|
||||
const { BackgroundSyncService } = await import('./background-sync.service');
|
||||
this.backgroundSync = await BackgroundSyncService.getInstance();
|
||||
}
|
||||
return await this.backgroundSync.forceScan();
|
||||
}
|
||||
|
||||
/**
|
||||
* Obtient le statut du service de background sync
|
||||
*/
|
||||
public getBackgroundSyncStatus(): { isRunning: boolean; scanInterval: number } | null {
|
||||
if (!this.backgroundSync) {
|
||||
return null;
|
||||
}
|
||||
return this.backgroundSync.getStatus();
|
||||
}
|
||||
|
||||
/**
|
||||
* Configure l'intervalle de scan du background sync
|
||||
*/
|
||||
public setBackgroundSyncInterval(intervalMs: number): void {
|
||||
if (this.backgroundSync) {
|
||||
this.backgroundSync.setScanInterval(intervalMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -321,6 +321,57 @@ class SimpleProcessHandlers {
|
||||
}
|
||||
}
|
||||
|
||||
async handleForceDataScan(event: ServerMessageEvent): Promise<ServerResponse> {
|
||||
if (event.data.type !== MessageType.FORCE_DATA_SCAN) {
|
||||
throw new Error('Invalid message type');
|
||||
}
|
||||
|
||||
try {
|
||||
const { apiKey } = event.data;
|
||||
|
||||
if (!apiKey || !this.validateApiKey(apiKey)) {
|
||||
throw new Error('Invalid API key');
|
||||
}
|
||||
|
||||
const missingHashes = await this.service.forceDataScan();
|
||||
|
||||
return {
|
||||
type: MessageType.DATA_SCAN_RESULT,
|
||||
missingHashes,
|
||||
count: missingHashes.length,
|
||||
messageId: event.data.messageId
|
||||
};
|
||||
} catch (e) {
|
||||
const errorMessage = e instanceof Error ? e.message : String(e || 'Unknown error');
|
||||
throw new Error(errorMessage);
|
||||
}
|
||||
}
|
||||
|
||||
async handleGetBackgroundSyncStatus(event: ServerMessageEvent): Promise<ServerResponse> {
|
||||
if (event.data.type !== MessageType.GET_BACKGROUND_SYNC_STATUS) {
|
||||
throw new Error('Invalid message type');
|
||||
}
|
||||
|
||||
try {
|
||||
const { apiKey } = event.data;
|
||||
|
||||
if (!apiKey || !this.validateApiKey(apiKey)) {
|
||||
throw new Error('Invalid API key');
|
||||
}
|
||||
|
||||
const status = this.service.getBackgroundSyncStatus();
|
||||
|
||||
return {
|
||||
type: MessageType.BACKGROUND_SYNC_STATUS,
|
||||
status,
|
||||
messageId: event.data.messageId
|
||||
};
|
||||
} catch (e) {
|
||||
const errorMessage = e instanceof Error ? e.message : String(e || 'Unknown error');
|
||||
throw new Error(errorMessage);
|
||||
}
|
||||
}
|
||||
|
||||
async handleMessage(event: ServerMessageEvent): Promise<ServerResponse> {
|
||||
try {
|
||||
switch (event.data.type) {
|
||||
@ -336,6 +387,10 @@ class SimpleProcessHandlers {
|
||||
return await this.handleGetMyProcesses(event);
|
||||
case MessageType.GET_PAIRING_ID:
|
||||
return await this.handleGetPairingId(event);
|
||||
case MessageType.FORCE_DATA_SCAN:
|
||||
return await this.handleForceDataScan(event);
|
||||
case MessageType.GET_BACKGROUND_SYNC_STATUS:
|
||||
return await this.handleGetBackgroundSyncStatus(event);
|
||||
default:
|
||||
throw new Error(`Unhandled message type: ${event.data.type}`);
|
||||
}
|
||||
@ -433,10 +488,20 @@ export class Server {
|
||||
// Connect to relays
|
||||
await service.connectToRelaysAndWaitForHandshake();
|
||||
|
||||
// Start background sync service for missing data detection
|
||||
try {
|
||||
await service.startBackgroundSync();
|
||||
console.log('🔄 Background sync service started');
|
||||
} catch (error) {
|
||||
console.error('❌ Failed to start background sync service:', error);
|
||||
// Don't exit, continue without background sync
|
||||
}
|
||||
|
||||
console.log(`✅ Simple server running on port ${this.wss.options.port}`);
|
||||
console.log('📋 Supported operations: UPDATE_PROCESS, NOTIFY_UPDATE, VALIDATE_STATE');
|
||||
console.log('🔑 Authentication: API key required for all operations');
|
||||
console.log('🔧 Services: Integrated with SimpleService protocol logic');
|
||||
console.log('🔄 Background sync: Automatic missing data detection enabled');
|
||||
|
||||
} catch (error) {
|
||||
console.error('❌ Failed to initialize server:', error);
|
||||
@ -509,6 +574,15 @@ export class Server {
|
||||
public shutdown() {
|
||||
console.log('🛑 Shutting down server...');
|
||||
|
||||
// Stop background sync service
|
||||
try {
|
||||
const service = Service.getInstance();
|
||||
service.stopBackgroundSync();
|
||||
console.log('🔄 Background sync service stopped');
|
||||
} catch (error) {
|
||||
console.error('❌ Error stopping background sync service:', error);
|
||||
}
|
||||
|
||||
// Close all active client connections first
|
||||
for (const [ws, clientId] of this.clients.entries()) {
|
||||
console.log(`🔌 Closing connection to ${clientId}...`);
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user