import { z } from "zod";
import type { WorkbenchAccess } from "./request-context";
import { env } from "@/shared/config/env";
import type { PipeFlowCalculation } from "../map/flow-overlay";
import { createWorkbenchHeaders, findOwnedWorkbenchProject, getServerError } from "./project-client";

const realtimeLinkSchema = z.object({
  time: z.string().datetime({ offset: true }),
  link_id: z.union([z.string().trim().min(1), z.number()]),
  velocity: z.number().finite().nullable()
});

const realtimeLinksSchema = z.array(realtimeLinkSchema);
const SERVER_API_BASE_URL = env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, "");
const FLOW_LOOKBACK_MS = 15 * 60 * 1000;

export type RealtimePipeFlowSnapshot = {
  observedAt: string;
  records: PipeFlowCalculation[];
};

export async function fetchRealtimePipeFlows(
  getAccessToken?: WorkbenchAccess,
  signal?: AbortSignal,
  now = new Date()
): Promise<RealtimePipeFlowSnapshot> {
  const { project, accessToken } = await findOwnedWorkbenchProject(getAccessToken, signal);
  const query = new URLSearchParams({
    start_time: new Date(now.getTime() - FLOW_LOOKBACK_MS).toISOString(),
    end_time: now.toISOString()
  });
  const response = await fetch(
    `${SERVER_API_BASE_URL}/api/v1/timeseries/realtime/links?${query.toString()}`,
    {
      headers: createWorkbenchHeaders(accessToken, project.project_id),
      signal
    }
  );
  const payload: unknown = await response.json().catch(() => null);
  if (!response.ok) {
    throw new Error(getServerError(payload, response.status, "实时管道流速请求"));
  }

  const rows = realtimeLinksSchema.parse(payload);
  const latestByPipe = new Map<string, z.infer<typeof realtimeLinkSchema>>();
  rows.forEach((row) => {
    if (row.velocity === null) return;
    const pipeId = String(row.link_id);
    const current = latestByPipe.get(pipeId);
    if (!current || Date.parse(row.time) >= Date.parse(current.time)) {
      latestByPipe.set(pipeId, row);
    }
  });
  const latestRows = [...latestByPipe.values()];
  return {
    observedAt: latestRows.reduce(
      (latest, row) => Date.parse(row.time) > Date.parse(latest) ? row.time : latest,
      latestRows[0]?.time ?? now.toISOString()
    ),
    records: latestRows.map((row) => ({
      pipeId: row.link_id,
      velocity: row.velocity as number
    }))
  };
}