P4-B: notifications Web Push (VAPID) sur passage en waiting
- db: migration id=4 `push_subscriptions` (liées au token d'auth) ; clés VAPID en settings - PushService: bootstrap idempotent des clés VAPID, subscribe/unsubscribe/count, notify() best-effort (purge des abonnements 410/404 Gone) ; sender injectable pour les tests - routes/push.ts: GET vapid-public-key, POST subscribe/unsubscribe/test (toutes sous auth globale) - pty-manager: déclencheur push sur FRONT MONTANT vers waiting, débouncé 1500ms et annulable (faux positif ignoré) ; câblage app/index/config (--vapid-contact) - shared/api.ts: types VapidKeyResponse / PushSubscribeRequest / PushUnsubscribeRequest - deps: web-push (+ @types/web-push) côté serveur - tests: push-service (idempotence, UPSERT, 410-purge, payload) + trigger pty-manager (169 verts) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -2,17 +2,20 @@ import { EventEmitter } from 'node:events';
|
||||
import { existsSync, statSync } from 'node:fs';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { homedir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { basename, join } from 'node:path';
|
||||
import pty from '@homebridge/node-pty-prebuilt-multiarch';
|
||||
import { FLOW, REPLAY_TAIL_BYTES, type SessionSummary } from '@arboretum/shared';
|
||||
import { FLOW, REPLAY_TAIL_BYTES, type SessionActivity, type SessionSummary } from '@arboretum/shared';
|
||||
import { RingBuffer } from './ring-buffer.js';
|
||||
import { buildSpawnSpec } from './claude-launcher.js';
|
||||
import { findByPid } from './session-registry.js';
|
||||
import { SessionActivityTracker } from './claude-adapter.js';
|
||||
import type { PushService } from './push-service.js';
|
||||
import type { Db } from '../db/index.js';
|
||||
|
||||
const RING_CAPACITY = 2 * 1024 * 1024;
|
||||
const KILL_GRACE_MS = 5000;
|
||||
/** Délai avant envoi d'une notif push sur passage en `waiting` : annulé si la session repart (faux positif). */
|
||||
const NOTIFY_DEBOUNCE_MS = 1500;
|
||||
/** Capture du claudeSessionId après spawn : poll du registre par pid (waitReady validé S1). */
|
||||
const CLAUDE_ID_POLL_MS = 400;
|
||||
const CLAUDE_ID_TIMEOUT_MS = 60_000;
|
||||
@@ -47,6 +50,10 @@ interface ManagedSession {
|
||||
claudeSessionId: string | null;
|
||||
/** détection d'état fin (busy/waiting/idle + dialogue) ; null pour bash (P3-B). */
|
||||
tracker: SessionActivityTracker | null;
|
||||
/** dernière activité notifiée (détection du front montant vers `waiting` pour le push P4-B). */
|
||||
prevActivity: SessionActivity | null;
|
||||
/** timer de notif push débouncée (annulé si la session quitte `waiting` avant l'échéance). */
|
||||
notifyTimer: NodeJS.Timeout | null;
|
||||
}
|
||||
|
||||
export interface PtyManagerEvents {
|
||||
@@ -60,6 +67,7 @@ export class PtyManager extends EventEmitter<PtyManagerEvents> {
|
||||
constructor(
|
||||
private readonly db: Db,
|
||||
private readonly sessionsDir: string = join(homedir(), '.claude', 'sessions'),
|
||||
private readonly push: PushService | null = null,
|
||||
) {
|
||||
super();
|
||||
}
|
||||
@@ -94,11 +102,16 @@ export class PtyManager extends EventEmitter<PtyManagerEvents> {
|
||||
killTimer: null,
|
||||
claudeSessionId: null,
|
||||
tracker: null,
|
||||
prevActivity: null,
|
||||
notifyTimer: null,
|
||||
};
|
||||
// Détection d'état fin (P3-B) : uniquement pour claude (bash n'a pas de registre).
|
||||
if (command === 'claude') {
|
||||
session.tracker = new SessionActivityTracker(proc.pid, this.sessionsDir, () => {
|
||||
if (!session.exited) this.emit('session_update', this.summarize(session));
|
||||
if (session.exited) return;
|
||||
const summary = this.summarize(session);
|
||||
this.maybeNotify(session, summary.activity ?? null);
|
||||
this.emit('session_update', summary);
|
||||
});
|
||||
}
|
||||
this.live.set(id, session);
|
||||
@@ -308,6 +321,36 @@ export class PtyManager extends EventEmitter<PtyManagerEvents> {
|
||||
|
||||
// ---- interne ----
|
||||
|
||||
/**
|
||||
* Push P4-B : notifie sur le FRONT MONTANT vers `waiting` uniquement (le tracker réémet
|
||||
* souvent le même état), avec un debounce annulable — un `waiting` ultra-bref (Claude répond
|
||||
* tout seul) ne déclenche pas de notif. Cible tous les abonnements (un seul utilisateur).
|
||||
*/
|
||||
private maybeNotify(s: ManagedSession, next: SessionActivity | null): void {
|
||||
const prev = s.prevActivity;
|
||||
s.prevActivity = next;
|
||||
if (!this.push) return;
|
||||
if (next === 'waiting' && prev !== 'waiting') {
|
||||
if (s.notifyTimer) clearTimeout(s.notifyTimer);
|
||||
s.notifyTimer = setTimeout(() => {
|
||||
s.notifyTimer = null;
|
||||
const act = s.tracker?.snapshot();
|
||||
if (s.exited || act?.activity !== 'waiting') return; // faux positif : annulé
|
||||
void this.push?.notify({
|
||||
sessionId: s.id,
|
||||
title: basename(s.cwd) || s.cwd,
|
||||
body: act.dialog?.waitingFor ?? act.waitingFor ?? 'waiting for your input',
|
||||
kind: act.dialog?.kind ?? null,
|
||||
url: `/sessions/${s.id}`,
|
||||
});
|
||||
}, NOTIFY_DEBOUNCE_MS);
|
||||
s.notifyTimer.unref();
|
||||
} else if (next !== 'waiting' && s.notifyTimer) {
|
||||
clearTimeout(s.notifyTimer);
|
||||
s.notifyTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
private handleOutput(s: ManagedSession, chunk: Buffer): void {
|
||||
s.ring.write(chunk);
|
||||
s.tracker?.feed(chunk);
|
||||
@@ -350,6 +393,7 @@ export class PtyManager extends EventEmitter<PtyManagerEvents> {
|
||||
s.tracker?.dispose();
|
||||
s.tracker = null;
|
||||
if (s.killTimer) clearTimeout(s.killTimer);
|
||||
if (s.notifyTimer) clearTimeout(s.notifyTimer);
|
||||
const endedAt = new Date().toISOString();
|
||||
this.db.prepare('UPDATE sessions SET ended_at = ?, exit_code = ? WHERE id = ?').run(endedAt, exitCode, s.id);
|
||||
for (const c of s.clients) c.onDetached('session_exit');
|
||||
|
||||
116
packages/server/src/core/push-service.ts
Normal file
116
packages/server/src/core/push-service.ts
Normal file
@@ -0,0 +1,116 @@
|
||||
// PushService : notifications Web Push (VAPID). Clés VAPID générées une fois au bootstrap
|
||||
// et stockées dans `settings` (comme le server_secret de l'auth) ; abonnements liés au token
|
||||
// d'auth (token_id). Quand une session managée passe en `waiting`, pty-manager appelle notify().
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { createRequire } from 'node:module';
|
||||
import { type Db, getSetting, setSetting } from '../db/index.js';
|
||||
|
||||
// web-push est publié en CommonJS : on le charge via require (verbatimModuleSyntax + NodeNext),
|
||||
// typé par l'import type — même pattern que @xterm/headless dans screen-reader.ts.
|
||||
const require = createRequire(import.meta.url);
|
||||
const webpush = require('web-push') as typeof import('web-push');
|
||||
|
||||
export interface PushSubscriptionInput {
|
||||
endpoint: string;
|
||||
keys: { p256dh: string; auth: string };
|
||||
}
|
||||
|
||||
/** Charge utile JSON poussée au service worker (cf. packages/web/src/sw.ts). */
|
||||
export interface PushPayload {
|
||||
sessionId: string;
|
||||
title: string;
|
||||
body: string;
|
||||
kind: string | null;
|
||||
url: string;
|
||||
}
|
||||
|
||||
interface SubRow {
|
||||
id: string;
|
||||
endpoint: string;
|
||||
p256dh: string;
|
||||
auth: string;
|
||||
}
|
||||
|
||||
/** Envoi d'une notif à un abonnement — injectable pour les tests ; défaut = web-push réel. */
|
||||
export type PushSender = (
|
||||
subscription: { endpoint: string; keys: { p256dh: string; auth: string } },
|
||||
payload: string,
|
||||
options: unknown,
|
||||
) => Promise<unknown>;
|
||||
|
||||
export class PushService {
|
||||
private readonly vapidPublic: string;
|
||||
private readonly vapidPrivate: string;
|
||||
private readonly send: PushSender;
|
||||
|
||||
constructor(
|
||||
private readonly db: Db,
|
||||
private readonly contact: string = 'mailto:arboretum@localhost',
|
||||
sender?: PushSender,
|
||||
) {
|
||||
this.send = sender ?? ((sub, payload, options) => webpush.sendNotification(sub, payload, options as Parameters<typeof webpush.sendNotification>[2]));
|
||||
let pub = getSetting(db, 'vapid_public');
|
||||
let priv = getSetting(db, 'vapid_private');
|
||||
if (!pub || !priv) {
|
||||
const keys = webpush.generateVAPIDKeys();
|
||||
pub = keys.publicKey;
|
||||
priv = keys.privateKey;
|
||||
setSetting(db, 'vapid_public', pub);
|
||||
setSetting(db, 'vapid_private', priv);
|
||||
}
|
||||
this.vapidPublic = pub;
|
||||
this.vapidPrivate = priv;
|
||||
}
|
||||
|
||||
/** Clé publique VAPID — sûre à exposer (applicationServerKey côté navigateur). */
|
||||
publicKey(): string {
|
||||
return this.vapidPublic;
|
||||
}
|
||||
|
||||
/** Enregistre (ou ré-associe) un abonnement, lié au token authentifié. */
|
||||
subscribe(tokenId: string, sub: PushSubscriptionInput, userAgent: string | null): void {
|
||||
this.db
|
||||
.prepare(
|
||||
`INSERT INTO push_subscriptions (id, token_id, endpoint, p256dh, auth, user_agent, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(endpoint) DO UPDATE SET
|
||||
token_id = excluded.token_id, p256dh = excluded.p256dh,
|
||||
auth = excluded.auth, user_agent = excluded.user_agent`,
|
||||
)
|
||||
.run(randomUUID(), tokenId, sub.endpoint, sub.keys.p256dh, sub.keys.auth, userAgent, new Date().toISOString());
|
||||
}
|
||||
|
||||
unsubscribe(endpoint: string): void {
|
||||
this.db.prepare('DELETE FROM push_subscriptions WHERE endpoint = ?').run(endpoint);
|
||||
}
|
||||
|
||||
count(): number {
|
||||
return (this.db.prepare('SELECT COUNT(*) AS n FROM push_subscriptions').get() as { n: number }).n;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pousse une notification à TOUS les abonnements (un seul utilisateur, plusieurs appareils).
|
||||
* Un abonnement expiré (404/410 Gone) est purgé ; les autres erreurs sont ignorées (best-effort,
|
||||
* un appareil injoignable ne doit pas bloquer les autres).
|
||||
*/
|
||||
async notify(payload: PushPayload): Promise<void> {
|
||||
const rows = this.db.prepare('SELECT id, endpoint, p256dh, auth FROM push_subscriptions').all() as unknown as SubRow[];
|
||||
if (rows.length === 0) return;
|
||||
const body = JSON.stringify(payload);
|
||||
const options = {
|
||||
vapidDetails: { subject: this.contact, publicKey: this.vapidPublic, privateKey: this.vapidPrivate },
|
||||
TTL: 60,
|
||||
};
|
||||
await Promise.all(
|
||||
rows.map(async (r) => {
|
||||
try {
|
||||
await this.send({ endpoint: r.endpoint, keys: { p256dh: r.p256dh, auth: r.auth } }, body, options);
|
||||
this.db.prepare('UPDATE push_subscriptions SET last_ok_at = ? WHERE id = ?').run(new Date().toISOString(), r.id);
|
||||
} catch (err) {
|
||||
const code = (err as { statusCode?: number }).statusCode;
|
||||
if (code === 404 || code === 410) this.db.prepare('DELETE FROM push_subscriptions WHERE id = ?').run(r.id);
|
||||
}
|
||||
}),
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user