Add files using upload-large-folder tool

4e23b01 verified about 1 month ago

11.4 kB

import { randomBytes } from 'node:crypto';
import {
  chmodSync,
  mkdirSync,
  readdirSync,
  readFileSync,
  statSync,
  unlinkSync,
  writeFileSync,
} from 'node:fs';
import { join } from 'node:path';

import type { EnrichedTelemetryEvent, TelemetryPrimitive } from './types';
import { isTelemetryPrimitive } from './types';

// Mainland-China telemetry endpoint, mirroring
// `KIMI_REGION_PROFILES['mainland-cn'].telemetryEndpoint` in
// `@moonshot-ai/kimi-code-oauth` (the region source of truth). This package
// deliberately has no dependency on it — region-aware callers pass `endpoint`
// explicitly (e.g. through `initializeTelemetry`).
export const TELEMETRY_ENDPOINT = 'https://telemetry-logs.kimi.com/v1/event';
export const SERVER_EVENT_PREFIX = 'kfc_';
export const USER_ID_PREFIX = 'kfc_device_id_';
export const DISK_EVENT_MAX_AGE_MS = 7 * 24 * 60 * 60 * 1000;
export const RETRY_BACKOFFS_MS = [1_000, 4_000, 16_000] as const;

export interface AsyncTransportOptions {
  readonly homeDir: string;
  readonly deviceId: string;
  /** Static endpoint, or a resolver invoked per flush so an in-process region
      switch (login/logout) takes effect without rebuilding the transport. */
  readonly endpoint?: string | (() => string);
  readonly getAccessToken?: () => string | null | Promise<string | null>;
  readonly fetchImpl?: typeof fetch;
  readonly retryBackoffsMs?: readonly number[];
  readonly requestTimeoutMs?: number;
  readonly sleep?: (ms: number, signal?: AbortSignal) => Promise<void>;
  readonly now?: () => number;
}

export interface TelemetryPayload {
  readonly user_id: string;
  readonly events: readonly Record<string, TelemetryPrimitive>[];
}

const DEFAULT_REQUEST_TIMEOUT_MS = 10_000;

export class AsyncTransport {
  private readonly homeDir: string;
  private readonly deviceId: string;
  private readonly endpoint: string | (() => string);
  private readonly getAccessToken: (() => string | null | Promise<string | null>) | null;
  private readonly fetchImpl: typeof fetch;
  private readonly retryBackoffsMs: readonly number[];
  private readonly requestTimeoutMs: number;
  private readonly sleepImpl: (ms: number, signal?: AbortSignal) => Promise<void>;
  private readonly now: () => number;

  constructor(options: AsyncTransportOptions) {
    this.homeDir = options.homeDir;
    this.deviceId = options.deviceId;
    this.endpoint = options.endpoint ?? TELEMETRY_ENDPOINT;
    this.getAccessToken = options.getAccessToken ?? null;
    this.fetchImpl = options.fetchImpl ?? globalThis.fetch.bind(globalThis);
    this.retryBackoffsMs = options.retryBackoffsMs ?? RETRY_BACKOFFS_MS;
    this.requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS;
    this.sleepImpl = options.sleep ?? abortableSleep;
    this.now = options.now ?? Date.now;
  }

  async send(events: readonly EnrichedTelemetryEvent[], signal?: AbortSignal): Promise<void> {
    if (events.length === 0) return;
    let savedToDisk = false;
    const saveEventsToDisk = (): void => {
      if (savedToDisk) return;
      this.saveToDisk(events);
      savedToDisk = true;
    };
    if (signal?.aborted === true) {
      saveEventsToDisk();
      throw abortError();
    }

    let payload: TelemetryPayload;
    try {
      payload = buildPayload(events, this.deviceId);
    } catch {
      return;
    }

    try {
      for (let attempt = 0; attempt <= this.retryBackoffsMs.length; attempt++) {
        try {
          await this.sendHttp(payload, signal);
          return;
        } catch (error) {
          if (isSignalAborted(signal) || isAbortError(error)) {
            saveEventsToDisk();
            throw error;
          }
          if (!(error instanceof TransientTelemetryError)) {
            break;
          }
          const backoff = this.retryBackoffsMs[attempt];
          if (backoff === undefined) break;
          await this.sleepImpl(backoff, signal);
        }
      }
    } catch (error) {
      if (isSignalAborted(signal) || isAbortError(error)) {
        saveEventsToDisk();
        throw error;
      }
    }

    saveEventsToDisk();
  }

  saveToDisk(events: readonly EnrichedTelemetryEvent[]): void {
    if (events.length === 0) return;
    const path = join(this.telemetryDir(), `failed_${randomBytes(6).toString('hex')}.jsonl`);
    const text = events.map((event) => JSON.stringify(event)).join('\n') + '\n';
    writeFileSync(path, text, { encoding: 'utf-8', mode: 0o600, flag: 'wx' });
    try {
      chmodSync(path, 0o600);
    } catch {
      // best effort on platforms that do not support chmod.
    }
  }

  async retryDiskEvents(): Promise<void> {
    let entries: string[];
    try {
      entries = readdirSync(this.telemetryDir());
    } catch {
      return;
    }

    const now = this.now();
    for (const entry of entries) {
      if (!entry.startsWith('failed_') || !entry.endsWith('.jsonl')) continue;
      const path = join(this.telemetryDir(), entry);
      try {
        const stat = statSync(path);
        if (now - stat.mtimeMs > DISK_EVENT_MAX_AGE_MS) {
          unlinkSync(path);
          continue;
        }
      } catch {
        continue;
      }

      let events: EnrichedTelemetryEvent[];
      let payload: TelemetryPayload;
      try {
        events = readJsonl(path);
        payload = buildPayload(events, this.deviceId);
      } catch (error) {
        if (error instanceof SyntaxError || error instanceof TypeError) {
          try {
            unlinkSync(path);
          } catch {
            // best effort cleanup.
          }
        }
        continue;
      }

      try {
        await this.sendHttp(payload);
        unlinkSync(path);
      } catch (error) {
        if (error instanceof TransientTelemetryError) continue;
      }
    }
  }

  private async sendHttp(payload: TelemetryPayload, signal?: AbortSignal): Promise<void> {
    const token = this.getAccessToken === null ? null : await this.getAccessToken();
    const headers: Record<string, string> = {
      'Content-Type': 'application/json',
    };
    if (token !== null && token.length > 0) {
      headers['Authorization'] = `Bearer ${token}`;
    }

    const response = await this.post(payload, headers, signal);
    if (response.status === 401 && headers['Authorization'] !== undefined) {
      delete headers['Authorization'];
      const retry = await this.post(payload, headers, signal);
      handleStatus(retry.status);
      return;
    }
    handleStatus(response.status);
  }

  private async post(
    payload: TelemetryPayload,
    headers: Record<string, string>,
    signal?: AbortSignal,
  ): Promise<Response> {
    try {
      const endpoint = typeof this.endpoint === 'function' ? this.endpoint() : this.endpoint;
      return await fetchWithTimeout(
        this.fetchImpl,
        endpoint,
        {
          method: 'POST',
          headers: { ...headers },
          body: JSON.stringify(payload),
        },
        this.requestTimeoutMs,
        signal,
      );
    } catch (error) {
      if (signal?.aborted === true || isAbortError(error)) throw error;
      throw new TransientTelemetryError(String(error));
    }
  }

  private telemetryDir(): string {
    const path = join(this.homeDir, 'telemetry');
    mkdirSync(path, { recursive: true, mode: 0o700 });
    try {
      chmodSync(path, 0o700);
    } catch {
      // best effort on platforms that do not support chmod.
    }
    return path;
  }
}

export class TransientTelemetryError extends Error {
  override readonly name = 'TransientTelemetryError';
}

export function buildUserId(deviceId: string): string {
  return USER_ID_PREFIX + deviceId;
}

export function buildPayload(
  events: readonly EnrichedTelemetryEvent[],
  deviceId: string,
): TelemetryPayload {
  return {
    user_id: buildUserId(deviceId),
    events: events.map((event) => flattenEvent(applyServerPrefix(event))),
  };
}

export function applyServerPrefix(event: EnrichedTelemetryEvent): EnrichedTelemetryEvent {
  const name: unknown = event.event;
  if (typeof name !== 'string' || name.length === 0 || name.startsWith(SERVER_EVENT_PREFIX)) {
    return event;
  }
  return { ...event, event: SERVER_EVENT_PREFIX + name };
}

export function flattenEvent(event: EnrichedTelemetryEvent): Record<string, TelemetryPrimitive> {
  const out: Record<string, TelemetryPrimitive> = {};
  for (const [key, value] of Object.entries(event)) {
    if (key === 'properties') {
      flattenNested(out, 'property', value);
    } else if (key === 'context') {
      flattenNested(out, 'context', value);
    } else {
      assertPrimitive(key, value);
      if (value !== null) {
        out[key] = value;
      }
    }
  }
  return out;
}

function flattenNested(target: Record<string, TelemetryPrimitive>, prefix: string, value: unknown) {
  if (value === null || typeof value !== 'object' || Array.isArray(value)) return;
  for (const [key, nestedValue] of Object.entries(value)) {
    assertPrimitive(`${prefix}.${key}`, nestedValue);
    if (nestedValue !== null) {
      target[`${prefix}_${key}`] = nestedValue;
    }
  }
}

function assertPrimitive(key: string, value: unknown): asserts value is TelemetryPrimitive {
  if (isTelemetryPrimitive(value)) return;
  throw new TypeError(`telemetry ${key} must be primitive`);
}

function handleStatus(status: number): void {
  if (status >= 500 || status === 429) {
    throw new TransientTelemetryError(`HTTP ${String(status)}`);
  }
  if (status >= 400) {
    return;
  }
}

function readJsonl(path: string): EnrichedTelemetryEvent[] {
  const text = readFileSync(path, 'utf-8');
  const events: EnrichedTelemetryEvent[] = [];
  for (const line of text.split('\n')) {
    const trimmed = line.trim();
    if (trimmed.length === 0) continue;
    events.push(JSON.parse(trimmed) as EnrichedTelemetryEvent);
  }
  return events;
}

async function fetchWithTimeout(
  fetchImpl: typeof fetch,
  url: string,
  init: RequestInit,
  timeoutMs: number,
  externalSignal?: AbortSignal,
): Promise<Response> {
  const controller = new AbortController();
  const abortFromExternal = (): void => {
    controller.abort(externalSignal?.reason);
  };
  const timeout = setTimeout(() => {
    controller.abort(new Error('telemetry request timed out'));
  }, timeoutMs);
  timeout.unref?.();
  if (externalSignal?.aborted === true) abortFromExternal();
  externalSignal?.addEventListener('abort', abortFromExternal, { once: true });
  try {
    return await fetchImpl(url, {
      ...init,
      signal: controller.signal,
    });
  } finally {
    clearTimeout(timeout);
    externalSignal?.removeEventListener('abort', abortFromExternal);
  }
}

function abortableSleep(ms: number, signal?: AbortSignal): Promise<void> {
  if (signal?.aborted === true) return Promise.reject(abortError());
  return new Promise((resolve, reject) => {
    const timer = setTimeout(resolve, ms);
    timer.unref?.();
    const onAbort = (): void => {
      clearTimeout(timer);
      reject(abortError());
    };
    signal?.addEventListener('abort', onAbort, { once: true });
  });
}

function isAbortError(error: unknown): boolean {
  return error instanceof Error && error.name === 'AbortError';
}

function isSignalAborted(signal?: AbortSignal): boolean {
  return signal?.aborted === true;
}

function abortError(): DOMException {
  return new DOMException('The operation was aborted.', 'AbortError');
}

	import { randomBytes } from 'node:crypto';
	import {
	chmodSync,
	mkdirSync,
	readdirSync,
	readFileSync,
	statSync,
	unlinkSync,
	writeFileSync,
	} from 'node:fs';
	import { join } from 'node:path';

	import type { EnrichedTelemetryEvent, TelemetryPrimitive } from './types';
	import { isTelemetryPrimitive } from './types';

	// Mainland-China telemetry endpoint, mirroring
	// `KIMI_REGION_PROFILES['mainland-cn'].telemetryEndpoint` in
	// `@moonshot-ai/kimi-code-oauth` (the region source of truth). This package
	// deliberately has no dependency on it — region-aware callers pass `endpoint`
	// explicitly (e.g. through `initializeTelemetry`).
	export const TELEMETRY_ENDPOINT = 'https://telemetry-logs.kimi.com/v1/event';
	export const SERVER_EVENT_PREFIX = 'kfc_';
	export const USER_ID_PREFIX = 'kfc_device_id_';
	export const DISK_EVENT_MAX_AGE_MS = 7 * 24 * 60 * 60 * 1000;
	export const RETRY_BACKOFFS_MS = [1_000, 4_000, 16_000] as const;

	export interface AsyncTransportOptions {
	readonly homeDir: string;
	readonly deviceId: string;
	/** Static endpoint, or a resolver invoked per flush so an in-process region
	switch (login/logout) takes effect without rebuilding the transport. */
	readonly endpoint?: string \| (() => string);
	readonly getAccessToken?: () => string \| null \| Promise<string \| null>;
	readonly fetchImpl?: typeof fetch;
	readonly retryBackoffsMs?: readonly number[];
	readonly requestTimeoutMs?: number;
	readonly sleep?: (ms: number, signal?: AbortSignal) => Promise<void>;
	readonly now?: () => number;
	}

	export interface TelemetryPayload {
	readonly user_id: string;
	readonly events: readonly Record<string, TelemetryPrimitive>[];
	}

	const DEFAULT_REQUEST_TIMEOUT_MS = 10_000;

	export class AsyncTransport {
	private readonly homeDir: string;
	private readonly deviceId: string;
	private readonly endpoint: string \| (() => string);
	private readonly getAccessToken: (() => string \| null \| Promise<string \| null>) \| null;
	private readonly fetchImpl: typeof fetch;
	private readonly retryBackoffsMs: readonly number[];
	private readonly requestTimeoutMs: number;
	private readonly sleepImpl: (ms: number, signal?: AbortSignal) => Promise<void>;
	private readonly now: () => number;

	constructor(options: AsyncTransportOptions) {
	this.homeDir = options.homeDir;
	this.deviceId = options.deviceId;
	this.endpoint = options.endpoint ?? TELEMETRY_ENDPOINT;
	this.getAccessToken = options.getAccessToken ?? null;
	this.fetchImpl = options.fetchImpl ?? globalThis.fetch.bind(globalThis);
	this.retryBackoffsMs = options.retryBackoffsMs ?? RETRY_BACKOFFS_MS;
	this.requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS;
	this.sleepImpl = options.sleep ?? abortableSleep;
	this.now = options.now ?? Date.now;
	}

	async send(events: readonly EnrichedTelemetryEvent[], signal?: AbortSignal): Promise<void> {
	if (events.length === 0) return;
	let savedToDisk = false;
	const saveEventsToDisk = (): void => {
	if (savedToDisk) return;
	this.saveToDisk(events);
	savedToDisk = true;
	};
	if (signal?.aborted === true) {
	saveEventsToDisk();
	throw abortError();
	}

	let payload: TelemetryPayload;
	try {
	payload = buildPayload(events, this.deviceId);
	} catch {
	return;
	}

	try {
	for (let attempt = 0; attempt <= this.retryBackoffsMs.length; attempt++) {
	try {
	await this.sendHttp(payload, signal);
	return;
	} catch (error) {
	if (isSignalAborted(signal) \|\| isAbortError(error)) {
	saveEventsToDisk();
	throw error;
	}
	if (!(error instanceof TransientTelemetryError)) {
	break;
	}
	const backoff = this.retryBackoffsMs[attempt];
	if (backoff === undefined) break;
	await this.sleepImpl(backoff, signal);
	}
	}
	} catch (error) {
	if (isSignalAborted(signal) \|\| isAbortError(error)) {
	saveEventsToDisk();
	throw error;
	}
	}

	saveEventsToDisk();
	}

	saveToDisk(events: readonly EnrichedTelemetryEvent[]): void {
	if (events.length === 0) return;
	const path = join(this.telemetryDir(), `failed_${randomBytes(6).toString('hex')}.jsonl`);
	const text = events.map((event) => JSON.stringify(event)).join('\n') + '\n';
	writeFileSync(path, text, { encoding: 'utf-8', mode: 0o600, flag: 'wx' });
	try {
	chmodSync(path, 0o600);
	} catch {
	// best effort on platforms that do not support chmod.
	}
	}

	async retryDiskEvents(): Promise<void> {
	let entries: string[];
	try {
	entries = readdirSync(this.telemetryDir());
	} catch {
	return;
	}

	const now = this.now();
	for (const entry of entries) {
	if (!entry.startsWith('failed_') \|\| !entry.endsWith('.jsonl')) continue;
	const path = join(this.telemetryDir(), entry);
	try {
	const stat = statSync(path);
	if (now - stat.mtimeMs > DISK_EVENT_MAX_AGE_MS) {
	unlinkSync(path);
	continue;
	}
	} catch {
	continue;
	}

	let events: EnrichedTelemetryEvent[];
	let payload: TelemetryPayload;
	try {
	events = readJsonl(path);
	payload = buildPayload(events, this.deviceId);
	} catch (error) {
	if (error instanceof SyntaxError \|\| error instanceof TypeError) {
	try {
	unlinkSync(path);
	} catch {
	// best effort cleanup.
	}
	}
	continue;
	}

	try {
	await this.sendHttp(payload);
	unlinkSync(path);
	} catch (error) {
	if (error instanceof TransientTelemetryError) continue;
	}
	}
	}

	private async sendHttp(payload: TelemetryPayload, signal?: AbortSignal): Promise<void> {
	const token = this.getAccessToken === null ? null : await this.getAccessToken();
	const headers: Record<string, string> = {
	'Content-Type': 'application/json',
	};
	if (token !== null && token.length > 0) {
	headers['Authorization'] = `Bearer ${token}`;
	}

	const response = await this.post(payload, headers, signal);
	if (response.status === 401 && headers['Authorization'] !== undefined) {
	delete headers['Authorization'];
	const retry = await this.post(payload, headers, signal);
	handleStatus(retry.status);
	return;
	}
	handleStatus(response.status);
	}

	private async post(
	payload: TelemetryPayload,
	headers: Record<string, string>,
	signal?: AbortSignal,
	): Promise<Response> {
	try {
	const endpoint = typeof this.endpoint === 'function' ? this.endpoint() : this.endpoint;
	return await fetchWithTimeout(
	this.fetchImpl,
	endpoint,
	{
	method: 'POST',
	headers: { ...headers },
	body: JSON.stringify(payload),
	},
	this.requestTimeoutMs,
	signal,
	);
	} catch (error) {
	if (signal?.aborted === true \|\| isAbortError(error)) throw error;
	throw new TransientTelemetryError(String(error));
	}
	}

	private telemetryDir(): string {
	const path = join(this.homeDir, 'telemetry');
	mkdirSync(path, { recursive: true, mode: 0o700 });
	try {
	chmodSync(path, 0o700);
	} catch {
	// best effort on platforms that do not support chmod.
	}
	return path;
	}
	}

	export class TransientTelemetryError extends Error {
	override readonly name = 'TransientTelemetryError';
	}

	export function buildUserId(deviceId: string): string {
	return USER_ID_PREFIX + deviceId;
	}

	export function buildPayload(
	events: readonly EnrichedTelemetryEvent[],
	deviceId: string,
	): TelemetryPayload {
	return {
	user_id: buildUserId(deviceId),
	events: events.map((event) => flattenEvent(applyServerPrefix(event))),
	};
	}

	export function applyServerPrefix(event: EnrichedTelemetryEvent): EnrichedTelemetryEvent {
	const name: unknown = event.event;
	if (typeof name !== 'string' \|\| name.length === 0 \|\| name.startsWith(SERVER_EVENT_PREFIX)) {
	return event;
	}
	return { ...event, event: SERVER_EVENT_PREFIX + name };
	}

	export function flattenEvent(event: EnrichedTelemetryEvent): Record<string, TelemetryPrimitive> {
	const out: Record<string, TelemetryPrimitive> = {};
	for (const [key, value] of Object.entries(event)) {
	if (key === 'properties') {
	flattenNested(out, 'property', value);
	} else if (key === 'context') {
	flattenNested(out, 'context', value);
	} else {
	assertPrimitive(key, value);
	if (value !== null) {
	out[key] = value;
	}
	}
	}
	return out;
	}

	function flattenNested(target: Record<string, TelemetryPrimitive>, prefix: string, value: unknown) {
	if (value === null \|\| typeof value !== 'object' \|\| Array.isArray(value)) return;
	for (const [key, nestedValue] of Object.entries(value)) {
	assertPrimitive(`${prefix}.${key}`, nestedValue);
	if (nestedValue !== null) {
	target[`${prefix}_${key}`] = nestedValue;
	}
	}
	}

	function assertPrimitive(key: string, value: unknown): asserts value is TelemetryPrimitive {
	if (isTelemetryPrimitive(value)) return;
	throw new TypeError(`telemetry ${key} must be primitive`);
	}

	function handleStatus(status: number): void {
	if (status >= 500 \|\| status === 429) {
	throw new TransientTelemetryError(`HTTP ${String(status)}`);
	}
	if (status >= 400) {
	return;
	}
	}

	function readJsonl(path: string): EnrichedTelemetryEvent[] {
	const text = readFileSync(path, 'utf-8');
	const events: EnrichedTelemetryEvent[] = [];
	for (const line of text.split('\n')) {
	const trimmed = line.trim();
	if (trimmed.length === 0) continue;
	events.push(JSON.parse(trimmed) as EnrichedTelemetryEvent);
	}
	return events;
	}

	async function fetchWithTimeout(
	fetchImpl: typeof fetch,
	url: string,
	init: RequestInit,
	timeoutMs: number,
	externalSignal?: AbortSignal,
	): Promise<Response> {
	const controller = new AbortController();
	const abortFromExternal = (): void => {
	controller.abort(externalSignal?.reason);
	};
	const timeout = setTimeout(() => {
	controller.abort(new Error('telemetry request timed out'));
	}, timeoutMs);
	timeout.unref?.();
	if (externalSignal?.aborted === true) abortFromExternal();
	externalSignal?.addEventListener('abort', abortFromExternal, { once: true });
	try {
	return await fetchImpl(url, {
	...init,
	signal: controller.signal,
	});
	} finally {
	clearTimeout(timeout);
	externalSignal?.removeEventListener('abort', abortFromExternal);
	}
	}

	function abortableSleep(ms: number, signal?: AbortSignal): Promise<void> {
	if (signal?.aborted === true) return Promise.reject(abortError());
	return new Promise((resolve, reject) => {
	const timer = setTimeout(resolve, ms);
	timer.unref?.();
	const onAbort = (): void => {
	clearTimeout(timer);
	reject(abortError());
	};
	signal?.addEventListener('abort', onAbort, { once: true });
	});
	}

	function isAbortError(error: unknown): boolean {
	return error instanceof Error && error.name === 'AbortError';
	}

	function isSignalAborted(signal?: AbortSignal): boolean {
	return signal?.aborted === true;
	}

	function abortError(): DOMException {
	return new DOMException('The operation was aborted.', 'AbortError');
	}