| 351 | } |
| 352 | |
| 353 | private async onSubscribe(msg: SubscribeMessage): Promise<void> { |
| 354 | const { session_ids, cursors, watch_fs } = msg.payload; |
| 355 | this.logger.info( |
| 356 | { sessionIds: session_ids, cursors, hasWatchFs: !!watch_fs }, |
| 357 | 'ws subscribe', |
| 358 | ); |
| 359 | |
| 360 | const sync = await this.syncSessions(session_ids, cursors); |
| 361 | |
| 362 | if (watch_fs && this.fsWatchHandler !== undefined) { |
| 363 | for (const [sid, cfg] of Object.entries(watch_fs)) { |
| 364 | if (cfg.paths.length === 0) continue; |
| 365 | const handler = this.fsWatchHandler; |
| 366 | void handler |
| 367 | .add(sid, this.id, cfg.paths) |
| 368 | .then((result) => { |
| 369 | if (!result.ok) { |
| 370 | this.logger.warn( |
| 371 | { sid, code: result.code, msg: result.msg }, |
| 372 | 'subscribe.watch_fs add failed; client should retry via watch_fs_add', |
| 373 | ); |
| 374 | } |
| 375 | }) |
| 376 | .catch((err: unknown) => { |
| 377 | this.logger.warn( |
| 378 | { sid, err: String(err) }, |
| 379 | 'subscribe.watch_fs add threw', |
| 380 | ); |
| 381 | }); |
| 382 | } |
| 383 | } |
| 384 | |
| 385 | this.send( |
| 386 | buildAck(msg.id, 0, 'success', { |
| 387 | accepted: sync.accepted, |
| 388 | not_found: [], |
| 389 | resync_required: sync.resyncRequired, |
| 390 | cursors: sync.serverCursors, |
| 391 | }), |
| 392 | ); |
| 393 | } |
| 394 | |
| 395 | private onUnsubscribe(msg: UnsubscribeMessage): void { |
| 396 | const { session_ids } = msg.payload; |