feat: move telemetry out of core into a usage statistics plugin

This commit is contained in:
LukeGus committed 2026-09-30 23:54:38 -05:00
1 parent f8edb5f299
commit 6217b91ae2
56 files changed
+2064 -564

No files matched your search

+140
View File
@@ -0,0 +1,140 @@
import fs from "node:fs";
import { randomUUID } from "node:crypto";
import { sql } from "drizzle-orm";
import type { PluginContext } from "@termix/plugin-sdk/backend";
import { envOverride } from "./config.js";
import { usageProperty, type UsageCounts, type UsageStore } from "./usage.js";
/* eslint-disable @typescript-eslint/no-explicit-any */
// ctx.db hands back the server's own drizzle handle and tables, untyped.
type Table = any;
type Drizzle = any;
/* eslint-enable @typescript-eslint/no-explicit-any */
export interface TelemetrySettings {
enabled: boolean;
includePlatform: boolean;
includeFeatureUsage: boolean;
includePlugins: boolean;
instanceId: string;
}
export interface TelemetryPayload {
event: "instance_heartbeat";
distinct_id: string;
properties: Record<string, unknown>;
}
type Env = Record<string, string | undefined>;
export async function readSettings(
ctx: PluginContext,
): Promise<TelemetrySettings> {
const all = await ctx.settings.getAll("admin");
return {
enabled: all.enabled !== false,
includePlatform: all.includePlatform !== false,
includeFeatureUsage: all.includeFeatureUsage !== false,
includePlugins: all.includePlugins !== false,
instanceId: typeof all.instanceId === "string" ? all.instanceId : "",
};
}
/** The env variable decides when set, the admin switch otherwise. */
export function isEnabled(settings: TelemetrySettings, env?: Env): boolean {
return envOverride(env) ?? settings.enabled;
}
/**
* Made on first use rather than on activate, so the boot copy of a 2.8
* instance id is never shadowed by a new one.
*/
export async function ensureInstanceId(ctx: PluginContext): Promise<string> {
const existing = await ctx.settings.get<string>("instanceId");
if (existing) return existing;
const id = randomUUID();
await ctx.settings.set("instanceId", id);
return id;
}
export function detectDeployment(
env: Env = process.env,
exists: (path: string) => boolean = fs.existsSync,
): "desktop" | "docker" | "server" {
if (env.ELECTRON_EMBEDDED === "true") return "desktop";
if (exists("/.dockerenv")) return "docker";
return "server";
}
async function countRows(ctx: PluginContext): Promise<{
userCount: number;
hostCount: number;
}> {
const { users, hosts } = await ctx.db.refs<{ users: Table; hosts: Table }>();
const db = await ctx.db.client<Drizzle>();
const count = async (table: Table) => {
const rows = await db.select({ count: sql<number>`count(*)` }).from(table);
return Number(rows[0]?.count ?? 0);
};
const [userCount, hostCount] = await Promise.all([
count(users),
count(hosts),
]);
return { userCount, hostCount };
}
export interface BuiltPayload {
payload: TelemetryPayload;
/** The counts the payload carries, to subtract once it is sent. */
usage: UsageCounts;
}
export async function buildPayload(
ctx: PluginContext,
usage: UsageStore,
env: Env = process.env,
): Promise<BuiltPayload> {
const settings = await readSettings(ctx);
const distinctId = await ensureInstanceId(ctx);
const { userCount, hostCount } = await countRows(ctx);
const properties: Record<string, unknown> = {
version: env.VERSION || "unknown",
user_count: userCount,
host_count: hostCount,
};
if (settings.includePlatform) {
properties.os = process.platform;
properties.arch = process.arch;
properties.node_version = process.versions.node;
properties.db_dialect = ctx.db.dialect;
properties.deployment = detectDeployment(env);
}
let carried: UsageCounts = {};
if (settings.includeFeatureUsage) {
carried = await usage.read();
for (const [name, count] of Object.entries(carried)) {
properties[usageProperty(name)] = count;
}
}
if (settings.includePlugins) {
const running = (await ctx.plugins.list())
.filter((plugin) => plugin.state === "active")
.map((plugin) => plugin.id)
.sort();
properties.plugins_enabled = running;
properties.plugin_count = running.length;
}
return {
payload: {
event: "instance_heartbeat",
distinct_id: distinctId,
properties,
},
usage: carried,
};
}
+26
View File
@@ -0,0 +1,26 @@
/** Public project key. Only lets a client send events, never read them. */
export const DEFAULT_POSTHOG_API_KEY =
"phc_xM8UznirsFxUkGE68gH4jzeqevf4kh76wGw7Ci7hH2dd";
export const DEFAULT_POSTHOG_HOST = "https://us.i.posthog.com";
type Env = Record<string, string | undefined>;
/** ENABLE_TELEMETRY wins over the admin setting when set. */
export function envOverride(env: Env = process.env): boolean | null {
const value = env.ENABLE_TELEMETRY?.trim().toLowerCase();
if (!value) return null;
return value === "true";
}
export function posthogConfig(env: Env = process.env): {
apiKey: string;
host: string;
} {
return {
apiKey: env.POSTHOG_API_KEY?.trim() || DEFAULT_POSTHOG_API_KEY,
host: (env.POSTHOG_HOST?.trim() || DEFAULT_POSTHOG_HOST).replace(
/\/+$/,
"",
),
};
}
+65
View File
@@ -0,0 +1,65 @@
import type { Router } from "express";
import type { PluginContext } from "@termix/plugin-sdk/backend";
import { envOverride } from "./config.js";
import { isEnabled, readSettings } from "./collect.js";
import { CHECK_INTERVAL_MS, createReporter } from "./reporter.js";
import { registerRoutes } from "./routes.js";
import { createUsageStore } from "./usage.js";
/** First check a few minutes after boot, once core's boot copies ran. */
const FIRST_CHECK_DELAY_MS = 3 * 60 * 1000;
type Env = Record<string, string | undefined>;
export async function activateWith(ctx: PluginContext, env: Env) {
const usage = createUsageStore(ctx.kv);
const reporter = createReporter(ctx, usage, env);
const check = async () => {
try {
await reporter.send();
} catch {
// Logged by the reporter; the next check tries again.
}
};
ctx.schedule.after(FIRST_CHECK_DELAY_MS, check);
ctx.schedule.every(CHECK_INTERVAL_MS, check);
const countEvent = (feature: string) => async () => {
const settings = await readSettings(ctx);
if (isEnabled(settings, env) && settings.includeFeatureUsage) {
await usage.add({ [feature]: 1 });
}
};
const onLogin = countEvent("ssh_login");
const onConnected = countEvent("host_connected");
ctx.events.on("host.login", () => void onLogin().catch(() => undefined));
ctx.events.on("host.session.status", (payload) => {
if ((payload as { online?: boolean } | null)?.online) {
void onConnected().catch(() => undefined);
}
});
ctx.settings.onValidate("admin", async (values) => {
const locked = envOverride(env);
if (locked === null || values.enabled === undefined) return;
const stored = (await ctx.settings.get<boolean>("enabled")) !== false;
if (values.enabled !== stored) {
return {
enabled:
"This is set by the ENABLE_TELEMETRY environment variable and cannot be changed here.",
};
}
});
registerRoutes(ctx, ctx.http.router<Router>(), reporter, usage, env);
return { reporter, usage };
}
export async function activate(ctx: PluginContext) {
await activateWith(ctx, process.env);
}
export async function deactivate() {
// Everything above was registered through ctx and is disposed by core.
}
+117
View File
@@ -0,0 +1,117 @@
import { randomUUID } from "node:crypto";
import type { PluginContext } from "@termix/plugin-sdk/backend";
import { envOverride, posthogConfig } from "./config.js";
import { buildPayload, isEnabled, readSettings } from "./collect.js";
import type { UsageStore } from "./usage.js";
export const REPORT_INTERVAL_MS = 24 * 60 * 60 * 1000;
/** Checked hourly, so a restart never skips or doubles a day. */
export const CHECK_INTERVAL_MS = 60 * 60 * 1000;
const STATE_KEY = "state";
export interface ReporterState {
lastSentAt: string | null;
lastAttemptAt: string | null;
lastError: string | null;
}
export type SendResult =
{ sent: true } | { sent: false; reason: "disabled" | "not-due" };
type Env = Record<string, string | undefined>;
export function createReporter(
ctx: PluginContext,
usage: UsageStore,
env: Env = process.env,
now: () => number = Date.now,
) {
async function state(): Promise<ReporterState> {
const stored = (await ctx.kv.get(STATE_KEY)) as
Partial<ReporterState> | undefined;
return {
lastSentAt: stored?.lastSentAt ?? null,
lastAttemptAt: stored?.lastAttemptAt ?? null,
lastError: stored?.lastError ?? null,
};
}
async function saveState(patch: Partial<ReporterState>) {
await ctx.kv.set(STATE_KEY, { ...(await state()), ...patch });
}
async function nextDueAt(): Promise<string | null> {
const { lastSentAt } = await state();
if (!lastSentAt) return null;
return new Date(Date.parse(lastSentAt) + REPORT_INTERVAL_MS).toISOString();
}
async function send(options: { force?: boolean } = {}): Promise<SendResult> {
if (!isEnabled(await readSettings(ctx), env)) {
return { sent: false, reason: "disabled" };
}
if (!options.force) {
const due = await nextDueAt();
if (due && Date.parse(due) > now()) {
return { sent: false, reason: "not-due" };
}
}
const attemptAt = new Date(now()).toISOString();
const { apiKey, host } = posthogConfig(env);
try {
const { payload, usage: carried } = await buildPayload(ctx, usage, env);
const response = await ctx.fetch(`${host}/capture/`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ api_key: apiKey, ...payload }),
timeoutMs: 10_000,
});
if (!response.ok) {
throw new Error(`PostHog answered ${response.status}`);
}
await usage.subtract(carried);
await saveState({
lastSentAt: attemptAt,
lastAttemptAt: attemptAt,
lastError: null,
});
ctx.log.info("Sent daily usage statistics");
return { sent: true };
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
await saveState({ lastAttemptAt: attemptAt, lastError: message });
ctx.log.warn(`Could not send usage statistics: ${message}`);
throw error;
}
}
return {
state,
send,
async status() {
const settings = await readSettings(ctx);
return {
enabled: isEnabled(settings, env),
locked: envOverride(env) !== null,
instanceId: settings.instanceId || null,
nextDueAt: await nextDueAt(),
...(await state()),
};
},
async preview() {
return (await buildPayload(ctx, usage, env)).payload;
},
async resetInstanceId(): Promise<string> {
const id = randomUUID();
await ctx.settings.set("instanceId", id);
return id;
},
};
}
export type Reporter = ReturnType<typeof createReporter>;
+148
View File
@@ -0,0 +1,148 @@
import type { Request, RequestHandler, Response, Router } from "express";
import type { PluginContext } from "@termix/plugin-sdk/backend";
import { isEnabled, readSettings } from "./collect.js";
import type { Reporter } from "./reporter.js";
import { sanitizeUsage, type UsageStore } from "./usage.js";
/** Whether the signed-in user's feature usage should be counted. */
export async function shouldTrackUser(
ctx: PluginContext,
userId: string | undefined,
env?: Record<string, string | undefined>,
): Promise<boolean> {
if (!userId) return false;
const settings = await readSettings(ctx);
if (!isEnabled(settings, env) || !settings.includeFeatureUsage) return false;
return (await ctx.settings.getUser(userId, "shareFeatureUsage")) !== false;
}
export function registerRoutes(
ctx: PluginContext,
router: Router,
reporter: Reporter,
usage: UsageStore,
env?: Record<string, string | undefined>,
): void {
const manage = ctx.rbac.require("manage") as unknown as RequestHandler;
/**
* @openapi
* /plugin-api/telemetry/status:
* get:
* summary: Usage statistics status
* description: Whether usage statistics are sent, whether ENABLE_TELEMETRY locks that, the instance ID and the last attempt.
* tags: [Usage Statistics]
* responses:
* 200:
* description: Status.
* 403:
* description: Missing the telemetry.manage permission.
*/
router.get("/status", manage, async (_req: Request, res: Response) => {
res.json(await reporter.status());
});
/**
* @openapi
* /plugin-api/telemetry/preview:
* get:
* summary: Preview the next report
* description: The exact event that would be sent right now, without the API key.
* tags: [Usage Statistics]
* responses:
* 200:
* description: The event.
* 403:
* description: Missing the telemetry.manage permission.
*/
router.get("/preview", manage, async (_req: Request, res: Response) => {
res.json(await reporter.preview());
});
/**
* @openapi
* /plugin-api/telemetry/send:
* post:
* summary: Send usage statistics now
* description: Sends the report straight away. Does nothing while usage statistics are turned off.
* tags: [Usage Statistics]
* responses:
* 200:
* description: Whether a report was sent.
* 403:
* description: Missing the telemetry.manage permission.
* 502:
* description: The report could not be delivered.
*/
router.post("/send", manage, async (_req: Request, res: Response) => {
try {
res.json(await reporter.send({ force: true }));
} catch (error) {
res.status(502).json({
error: error instanceof Error ? error.message : String(error),
});
}
});
/**
* @openapi
* /plugin-api/telemetry/reset-id:
* post:
* summary: Reset the instance ID
* description: Picks a new random instance ID, so later reports look like a new install.
* tags: [Usage Statistics]
* responses:
* 200:
* description: The new ID.
* 403:
* description: Missing the telemetry.manage permission.
*/
router.post("/reset-id", manage, async (_req: Request, res: Response) => {
res.json({ instanceId: await reporter.resetInstanceId() });
});
/**
* @openapi
* /plugin-api/telemetry/usage/config:
* get:
* summary: Whether to count this user's feature usage
* description: False when usage statistics or feature usage are off, or the user turned off their own part.
* tags: [Usage Statistics]
* responses:
* 200:
* description: "{ track: boolean }"
*/
router.get("/usage/config", async (_req: Request, res: Response) => {
res.json({ track: await shouldTrackUser(ctx, ctx.currentActor(), env) });
});
/**
* @openapi
* /plugin-api/telemetry/usage:
* post:
* summary: Add feature usage counts
* description: Counts from the signed-in user's browser, added to the next report. Ignored when this user is not counted. Names and counts are checked and capped.
* tags: [Usage Statistics]
* requestBody:
* required: true
* content:
* application/json:
* schema:
* type: object
* properties:
* features:
* type: object
* additionalProperties:
* type: integer
* responses:
* 204:
* description: Recorded or ignored.
*/
router.post("/usage", async (req: Request, res: Response) => {
if (await shouldTrackUser(ctx, ctx.currentActor(), env)) {
const counts = sanitizeUsage(req.body?.features);
if (Object.keys(counts).length > 0) await usage.add(counts);
}
res.status(204).end();
});
}
+75
View File
@@ -0,0 +1,75 @@
import type { PluginKeyValue } from "@termix/plugin-sdk/backend";
const USAGE_KEY = "usage";
const FEATURE_NAME = /^[a-z][a-z0-9_.-]{0,63}$/;
export const MAX_FEATURES = 100;
const MAX_COUNT = 100_000;
export type UsageCounts = Record<string, number>;
/** Keeps valid names with positive whole counts, capped. */
export function sanitizeUsage(input: unknown): UsageCounts {
const clean: UsageCounts = {};
if (!input || typeof input !== "object" || Array.isArray(input)) {
return clean;
}
for (const [name, raw] of Object.entries(input)) {
if (!FEATURE_NAME.test(name)) continue;
const count = Math.floor(Number(raw));
if (!Number.isFinite(count) || count <= 0) continue;
clean[name] = Math.min(count, MAX_COUNT);
if (Object.keys(clean).length >= MAX_FEATURES) break;
}
return clean;
}
export function createUsageStore(kv: PluginKeyValue) {
// One write at a time, so two requests never drop each other's counts.
let queue: Promise<unknown> = Promise.resolve();
const serial = <T>(fn: () => Promise<T>): Promise<T> => {
const next = queue.then(fn, fn);
queue = next.catch(() => undefined);
return next;
};
const read = async (): Promise<UsageCounts> =>
sanitizeUsage(await kv.get(USAGE_KEY));
return {
read,
add: (counts: UsageCounts) =>
serial(async () => {
const current = await read();
for (const [name, count] of Object.entries(sanitizeUsage(counts))) {
if (
!(name in current) &&
Object.keys(current).length >= MAX_FEATURES
) {
continue;
}
current[name] = Math.min((current[name] ?? 0) + count, MAX_COUNT);
}
await kv.set(USAGE_KEY, current);
}),
/** Drops what a successful report already carried. */
subtract: (sent: UsageCounts) =>
serial(async () => {
const current = await read();
for (const [name, count] of Object.entries(sent)) {
const left = (current[name] ?? 0) - count;
if (left > 0) current[name] = left;
else delete current[name];
}
await kv.set(USAGE_KEY, current);
}),
};
}
export type UsageStore = ReturnType<typeof createUsageStore>;
/** PostHog property name for a feature: used_tab_terminal. */
export function usageProperty(name: string): string {
return `used_${name.replace(/[.-]/g, "_")}`;
}
@@ -0,0 +1,191 @@
import { useCallback, useEffect, useState } from "react";
import { toast } from "sonner";
import {
usePluginApi,
useTranslation,
type SettingsComponentProps,
} from "@termix/plugin-sdk/frontend";
import {
Button,
Dialog,
DialogContent,
DialogDescription,
DialogFooter,
DialogHeader,
DialogTitle,
useConfirmation,
} from "@termix/plugin-sdk/ui";
export interface TelemetryStatus {
enabled: boolean;
locked: boolean;
instanceId: string | null;
lastSentAt: string | null;
lastAttemptAt: string | null;
lastError: string | null;
nextDueAt: string | null;
}
function errorMessage(error: unknown): string {
const data = (error as { response?: { data?: { error?: string } } })?.response
?.data;
if (data?.error) return data.error;
return error instanceof Error ? error.message : String(error);
}
export function TelemetryStatusSetting({ running }: SettingsComponentProps) {
const { t } = useTranslation();
const api = usePluginApi();
const { confirmWithToast } = useConfirmation();
const [status, setStatus] = useState<TelemetryStatus | null>(null);
const [sending, setSending] = useState(false);
const [preview, setPreview] = useState<unknown>(null);
const load = useCallback(async () => {
try {
const { data } = await api.get<TelemetryStatus>("/status");
setStatus(data);
} catch {
setStatus(null);
toast.error(t("status.loadFailed"));
}
}, [api, t]);
useEffect(() => {
if (running) void load();
}, [running, load]);
async function sendNow() {
setSending(true);
try {
const { data } = await api.post<{ sent: boolean }>("/send");
if (data.sent) toast.success(t("status.sent"));
else toast.info(t("status.notSent"));
} catch (error) {
toast.error(errorMessage(error));
} finally {
setSending(false);
await load();
}
}
async function showPreview() {
try {
const { data } = await api.get("/preview");
setPreview(data);
} catch (error) {
toast.error(errorMessage(error));
}
}
function resetId() {
void confirmWithToast(t("status.resetIdConfirm"), async () => {
try {
await api.post("/reset-id");
toast.success(t("status.resetIdDone"));
await load();
} catch (error) {
toast.error(errorMessage(error));
}
});
}
let summary = t("status.off");
if (status?.locked) {
summary = status.enabled ? t("status.lockedOn") : t("status.lockedOff");
} else if (status?.enabled) {
summary = t("status.on");
}
const muted = "text-[10px] text-muted-foreground";
return (
<div className="flex flex-col gap-2 border border-border bg-background/50 p-2">
<span className="text-xs font-medium">{t("status.title")}</span>
{status && (
<>
<span className={muted}>{summary}</span>
<span className={`${muted} break-all`}>
{status.instanceId
? t("status.instanceId", { id: status.instanceId })
: t("status.noInstanceId")}
</span>
<span className={muted}>
{status.lastSentAt
? t("status.lastSent", {
date: new Date(status.lastSentAt).toLocaleString(),
})
: t("status.neverSent")}
</span>
{status.enabled && status.nextDueAt && (
<span className={muted}>
{t("status.nextSend", {
date: new Date(status.nextDueAt).toLocaleString(),
})}
</span>
)}
{status.lastError && (
<span className="text-[10px] text-destructive">
{t("status.lastError", { error: status.lastError })}
</span>
)}
</>
)}
<div className="flex flex-wrap gap-2">
<Button
type="button"
variant="outline"
className="rounded-none"
disabled={!running}
onClick={showPreview}
>
{t("status.preview")}
</Button>
<Button
type="button"
variant="outline"
className="rounded-none"
disabled={!running || sending || !status?.enabled}
onClick={sendNow}
>
{sending ? t("status.sending") : t("status.sendNow")}
</Button>
<Button
type="button"
variant="outline"
className="rounded-none"
disabled={!running}
onClick={resetId}
>
{t("status.resetId")}
</Button>
</div>
<Dialog
open={preview !== null}
onOpenChange={(open) => !open && setPreview(null)}
>
<DialogContent className="rounded-none sm:max-w-2xl">
<DialogHeader>
<DialogTitle>{t("status.previewTitle")}</DialogTitle>
<DialogDescription>
{t("status.previewDescription")}
</DialogDescription>
</DialogHeader>
<pre className="max-h-[60vh] overflow-auto border border-border bg-muted/40 p-2 text-[11px]">
{JSON.stringify(preview, null, 2)}
</pre>
<DialogFooter>
<Button
type="button"
variant="outline"
className="rounded-none"
onClick={() => setPreview(null)}
>
{t("status.close")}
</Button>
</DialogFooter>
</DialogContent>
</Dialog>
</div>
);
}
+12
View File
@@ -0,0 +1,12 @@
import type { TermixApp } from "@termix/plugin-sdk/frontend";
import { TelemetryStatusSetting } from "./TelemetryStatusSetting";
import { startTabTracker } from "./tracker";
export function activate(app: TermixApp): void {
app.registerSettingsComponent("status", TelemetryStatusSetting);
if (!app.guest) app.onDispose(startTabTracker(app));
}
export function deactivate(): void {
// Registrations through app are disposed by core.
}
+109
View File
@@ -0,0 +1,109 @@
import type { TermixApp } from "@termix/plugin-sdk/frontend";
export const FLUSH_INTERVAL_MS = 5 * 60 * 1000;
type Counts = Record<string, number>;
/** Open tabs per type. Slot ids change on every read, so only types count. */
export function countTabTypes(layout: unknown): Counts {
const tabs = (layout as { tabs?: unknown } | null)?.tabs;
const counts: Counts = {};
if (!Array.isArray(tabs)) return counts;
for (const tab of tabs) {
const type = (tab as { type?: unknown })?.type;
if (typeof type === "string" && type)
counts[type] = (counts[type] ?? 0) + 1;
}
return counts;
}
/** Tabs opened between two snapshots, keyed as tab.<type>. */
export function openedSince(before: Counts, after: Counts): Counts {
const opened: Counts = {};
for (const [type, count] of Object.entries(after)) {
const added = count - (before[type] ?? 0);
if (added > 0) opened[`tab.${type}`] = added;
}
return opened;
}
/**
* Counts which kinds of tabs the user opens and sends the totals to the
* plugin's backend every few minutes. Does nothing unless the backend says
* this user is counted.
*/
export function startTabTracker(
app: TermixApp,
flushMs: number = FLUSH_INTERVAL_MS,
): () => void {
let track = false;
let ready = false;
let previous: Counts = {};
let pending: Counts = {};
const snapshot = () => countTabTypes(app.tabs.getLayout());
const refreshConfig = async () => {
try {
const { data } = await app.api.get<{ track?: boolean }>("/usage/config");
track = data?.track === true;
} catch {
track = false;
}
if (!track) pending = {};
};
const flush = (keepalive = false) => {
if (!track || Object.keys(pending).length === 0) return;
const features = pending;
pending = {};
void app
.fetch("/usage", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ features }),
keepalive,
})
.catch(() => undefined);
};
const disposers: Array<() => void> = [];
disposers.push(
app.tabs.onReady(() => {
// Tabs restored after login were opened in an earlier session.
previous = snapshot();
ready = true;
}),
);
disposers.push(
app.tabs.onChange(() => {
const current = snapshot();
if (ready && track) {
for (const [name, count] of Object.entries(
openedSince(previous, current),
)) {
pending[name] = (pending[name] ?? 0) + count;
}
}
previous = current;
}),
);
disposers.push(app.onSettingsChanged(() => void refreshConfig()));
const timer = setInterval(() => flush(), flushMs);
disposers.push(() => clearInterval(timer));
const onPageHide = () => flush(true);
window.addEventListener("pagehide", onPageHide);
disposers.push(() => window.removeEventListener("pagehide", onPageHide));
void refreshConfig();
return () => {
flush(true);
for (const dispose of disposers) dispose();
};
}