Add SSE event ack support to prevent channel backup

- Track lastHeardEventId and lastAcknowledgedEventId
- Send ack every 20 events (threshold recommended by Urbit)
- Parse event ID from SSE 'id:' lines
- This should fix dropped messages issue
This commit is contained in:
Hunter Miller 2026-01-29 21:17:27 -06:00
parent a41d83c8f2
commit 40e8f37022

View File

@ -43,6 +43,11 @@ export class UrbitSSEClient {
isConnected = false; isConnected = false;
logger: UrbitSseLogger; logger: UrbitSseLogger;
// Event ack tracking - must ack every ~50 events to keep channel healthy
private lastHeardEventId = -1;
private lastAcknowledgedEventId = -1;
private readonly ackThreshold = 20;
constructor(url: string, cookie: string, options: UrbitSseOptions = {}) { constructor(url: string, cookie: string, options: UrbitSseOptions = {}) {
this.url = url; this.url = url;
this.cookie = cookie.split(";")[0]; this.cookie = cookie.split(";")[0];
@ -214,15 +219,31 @@ export class UrbitSSEClient {
processEvent(eventData: string) { processEvent(eventData: string) {
const lines = eventData.split("\n"); const lines = eventData.split("\n");
let data: string | null = null; let data: string | null = null;
let eventId: number | null = null;
for (const line of lines) { for (const line of lines) {
if (line.startsWith("data: ")) { if (line.startsWith("data: ")) {
data = line.substring(6); data = line.substring(6);
} }
if (line.startsWith("id: ")) {
eventId = parseInt(line.substring(4), 10);
}
} }
if (!data) return; if (!data) return;
// Track event ID and send ack if needed
if (eventId !== null && !isNaN(eventId)) {
if (eventId > this.lastHeardEventId) {
this.lastHeardEventId = eventId;
if (eventId - this.lastAcknowledgedEventId > this.ackThreshold) {
this.ack(eventId).catch((err) => {
this.logger.error?.(`Failed to ack event ${eventId}: ${String(err)}`);
});
}
}
}
try { try {
const parsed = JSON.parse(data) as { id?: number; json?: unknown; response?: string }; const parsed = JSON.parse(data) as { id?: number; json?: unknown; response?: string };
@ -249,6 +270,30 @@ export class UrbitSSEClient {
} }
} }
private async ack(eventId: number): Promise<void> {
this.lastAcknowledgedEventId = eventId;
const ackData = {
action: "ack",
"event-id": eventId,
};
const response = await fetch(this.channelUrl, {
method: "PUT",
headers: {
"Content-Type": "application/json",
Cookie: this.cookie,
},
body: JSON.stringify([ackData]),
});
if (!response.ok && response.status !== 204) {
throw new Error(`Ack failed: ${response.status}`);
}
this.logger.log?.(`[SSE] Acked event ${eventId}`);
}
async poke(params: { app: string; mark: string; json: unknown }) { async poke(params: { app: string; mark: string; json: unknown }) {
const pokeId = Date.now(); const pokeId = Date.now();
const pokeData = { const pokeData = {