flxn-app/src/features/telemetry/server.ts

import { createServerFn } from "@tanstack/react-start";
import { z } from "zod";
import { superTokensAdminFunctionMiddleware } from "@/utils/supertokens";
import { pbAdmin } from "@/lib/pocketbase/client";
import { toServerResult } from "@/lib/tanstack-query/utils/to-server-result";
import {
  transformClientError,
  transformClientEvent,
} from "@/lib/pocketbase/util/transform-types";
import { ensureTelemetryScheduler, getSchedulerStatus } from "@/lib/telemetry/scheduler.server";
import { getHealthSnapshot } from "@/lib/telemetry/health.server";
import { getActiveConnectionCount } from "@/lib/events/emitter";
import type {
  ClientError,
  ClientErrorGroup,
  ClientEvent,
  RollupRow,
  TelemetryAlert,
  TelemetryListResult,
  TelemetryRuntimeStatus,
} from "./types";

const clientEventSearchSchema = z.object({
  page: z.number().optional(),
  perPage: z.number().max(200).optional(),
  name: z.string().optional(),
  player: z.string().optional(),
  sessionId: z.string().optional(),
  path: z.string().optional(),
  from: z.string().optional(),
  to: z.string().optional(),
  sortBy: z.string().optional(),
});

export const searchClientEvents = createServerFn()
  .validator(clientEventSearchSchema)
  .middleware([superTokensAdminFunctionMiddleware])
  .handler(async ({ data }) =>
    toServerResult<TelemetryListResult<ClientEvent>>(async () => {
      const result = await pbAdmin.searchClientEvents(data);
      return { ...result, items: result.items.map(transformClientEvent) };
    })
  );

const clientErrorSearchSchema = z.object({
  page: z.number().optional(),
  perPage: z.number().max(200).optional(),
  groupHash: z.string().optional(),
  resolved: z.boolean().optional(),
  player: z.string().optional(),
  from: z.string().optional(),
  to: z.string().optional(),
  sortBy: z.string().optional(),
});

export const searchClientErrors = createServerFn()
  .validator(clientErrorSearchSchema)
  .middleware([superTokensAdminFunctionMiddleware])
  .handler(async ({ data }) =>
    toServerResult<TelemetryListResult<ClientError>>(async () => {
      const result = await pbAdmin.searchClientErrors(data);
      return { ...result, items: result.items.map(transformClientError) };
    })
  );

const errorGroupsSchema = z.object({
  from: z.string(),
  to: z.string(),
  includeResolved: z.boolean().optional(),
  limit: z.number().max(100).optional(),
});

export const getClientErrorGroups = createServerFn()
  .validator(errorGroupsSchema)
  .middleware([superTokensAdminFunctionMiddleware])
  .handler(async ({ data }) =>
    toServerResult<ClientErrorGroup[]>(async () => {
      const { from, to, includeResolved = false, limit = 50 } = data;

      const rollups = await pbAdmin.getRollupRange({
        metrics: ["client_error.count"],
        from,
        to,
      });

      const countByGroup = new Map<string, number>();
      for (const rollup of rollups) {
        if (!rollup.dim) continue;
        countByGroup.set(rollup.dim, (countByGroup.get(rollup.dim) ?? 0) + rollup.value);
      }

      const topGroups = [...countByGroup.entries()]
        .sort((a, b) => b[1] - a[1])
        .slice(0, limit);

      const groups: ClientErrorGroup[] = [];
      for (const [groupHash, count] of topGroups) {
        const latest = await pbAdmin.searchClientErrors({ groupHash, perPage: 1 });
        const sample = latest.items[0];
        if (!sample) continue;

        // A group counts as resolved when its most recent occurrence is
        // resolved; a recurrence arrives resolved=false and surfaces again.
        const resolved = !!sample.resolved;
        if (resolved && !includeResolved) continue;

        groups.push({
          group_hash: groupHash,
          count,
          lastSeen: sample.created,
          resolved,
          sample: {
            message: sample.message,
            stack: sample.stack,
            path: sample.path,
            source: sample.source,
          },
        });
      }

      return groups;
    })
  );

export const resolveClientErrorGroup = createServerFn({ method: "POST" })
  .validator(z.object({ groupHash: z.string(), resolved: z.boolean().optional() }))
  .middleware([superTokensAdminFunctionMiddleware])
  .handler(async ({ data }) =>
    toServerResult<{ updated: number }>(async () => {
      const updated = await pbAdmin.resolveErrorsByGroup(data.groupHash, data.resolved ?? true);
      return { updated };
    })
  );

const rollupRangeSchema = z.object({
  metrics: z.array(z.string()).min(1).max(8),
  from: z.string().regex(/^\d{4}-\d{2}-\d{2}$/),
  to: z.string().regex(/^\d{4}-\d{2}-\d{2}$/),
  dim: z.string().optional(),
});

export const getRollupRange = createServerFn()
  .validator(rollupRangeSchema)
  .middleware([superTokensAdminFunctionMiddleware])
  .handler(async ({ data }) =>
    toServerResult<RollupRow[]>(async () => pbAdmin.getRollupRange(data))
  );

export const listTelemetryAlerts = createServerFn()
  .validator(
    z.object({ page: z.number().optional(), perPage: z.number().max(200).optional() })
  )
  .middleware([superTokensAdminFunctionMiddleware])
  .handler(async ({ data }) =>
    toServerResult<TelemetryListResult<TelemetryAlert>>(async () =>
      pbAdmin.listTelemetryAlerts(data.page, data.perPage)
    )
  );

export const getTelemetryRuntimeStatus = createServerFn()
  .middleware([superTokensAdminFunctionMiddleware])
  .handler(async () =>
    toServerResult<TelemetryRuntimeStatus>(async () => {
      // Dev has no k8s probes hitting /api/health, so opening the dashboard
      // is the scheduler's fallback starter.
      ensureTelemetryScheduler();
      const health = await getHealthSnapshot();
      return {
        health: { status: health.status, checks: health.checks },
        scheduler: getSchedulerStatus(),
        activeSseConnections: getActiveConnectionCount(),
      };
    })
  );