next-tjwater-frontend/src/features/workbench/api/scada-replay-client.ts

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

const ruleSchema = z.object({
  rule_id: z.string(), device_ids: z.array(z.string()), kind: z.enum(["relative_change", "bounds"]),
  direction: z.enum(["both", "increase", "decrease"]).optional(), threshold: z.number().nullable().optional(),
  lower: z.number().nullable().optional(), upper: z.number().nullable().optional(), unit: z.string(), metric_name: z.string().nullable().optional(),
  minimum_samples: z.number().optional(), window_samples: z.number().optional(),
  consecutive_samples: z.number().optional(), recovery_samples: z.number().optional(), minimum_devices: z.number().optional()
});
const readingSchema = z.object({
  time: z.string().datetime({ offset: true }), device_id: z.string(),
  monitored_value: z.number().finite().nullable(), cleaned_value: z.number().finite().nullable(),
  quality: z.enum(["valid", "corrected", "missing", "invalid", "unvalidated"]), reason: z.string()
});
const eventSchema = z.object({
  event_id: z.string(), rule_id: z.string(), device_ids: z.array(z.string()),
  occurred_at: z.string(), last_observed_at: z.string(), status: z.enum(["active", "resolved"]),
  detail: z.string(), task_id: z.string().nullable(), category: z.string().optional(),
  change: z.enum(["raised", "resolved"]).optional(), resolved_at: z.string().optional()
});
const monitoringWeightsSchema = z.object({
  burst_detection: z.number(), pressure: z.number(), outlet_flow: z.number()
});
const monitoringConfigSchema = z.object({
  rule_version: z.string(), decision_policy: z.enum(["weighted", "all_signals"]),
  pressure_device_ids: z.array(z.string()), outlet_flow_device_ids: z.array(z.string()),
  weights: monitoringWeightsSchema, alert_threshold: z.number()
});
const calibrationSchema = z.object({
  version: z.string().optional(), diagnostic_policy: z.string().optional(),
  status: z.enum(["ready", "insufficient_data"]),
  knowledge_cutoff: z.string(), history_start: z.string(),
  pressure: z.record(z.string(), z.object({ threshold_percent: z.number() })),
  outlet_flow: z.record(z.string(), z.object({ threshold_percent: z.number() })),
  burst_detection: z.object({ threshold: z.number().optional() }).optional()
});
const monitoringSchema = z.object({
  monitoring_id: z.string().uuid(),
  period_start: z.string(), period_end: z.string(), knowledge_cutoff: z.string(),
  batch_sequence: z.number().int(), rule_version: z.string(),
  status: z.enum(["normal", "attention", "insufficient_data", "suspected_burst"]),
  detail: z.string(), task_id: z.string().uuid().nullable(),
  pressure: z.record(z.string(), z.unknown()),
  outlet_flow: z.record(z.string(), z.unknown()),
  burst_detection: z.record(z.string(), z.unknown()),
  anomaly_signals: z.array(z.object({
    event_id: z.string(), rule_id: z.string(), device_ids: z.array(z.string()), detail: z.string()
  })).optional(),
  weighted: z.object({
    score: z.number(), threshold: z.number(),
    weights: monitoringWeightsSchema
  }).passthrough().nullable().optional(),
  incident: z.object({
    incident_id: z.string().uuid(), status: z.enum(["active", "resolved"]),
    action: z.enum(["raised", "ongoing", "resolved"]),
    started_at: z.string(), last_observed_at: z.string(), recovery_count: z.number().int()
  }).nullable().optional()
}).passthrough();
export const scadaReplaySchema = z.object({
  replay_id: z.string().uuid(), project_id: z.string().uuid(), revision: z.number().int(),
  status: z.enum(["ready", "paused", "completed", "stopped", "failed"]),
  start_time: z.string(), end_time: z.string(), virtual_now: z.string(),
  monitoring_interval_seconds: z.number().int().min(60).max(31_536_000).nullable().default(null),
  step_pending: z.boolean().default(false),
  monitoring_count: z.number().int().default(0),
  last_monitoring: monitoringSchema.nullable().default(null),
  monitoring_config: monitoringConfigSchema.nullable().optional(),
  monitoring_calibration: calibrationSchema.nullable().optional(),
  delivered_until: z.string().nullable(), processed_until: z.string().nullable(),
  processed_count: z.number(), anomaly_count: z.number(),
  devices: z.array(z.object({ device_id: z.string(), measurement_unit: z.string(), device_type: z.string() })),
  rules: z.array(ruleSchema), latest: z.array(readingSchema), active_events: z.array(eventSchema),
  error: z.string().nullable(), created_at: z.string(), updated_at: z.string()
});
const capabilitiesSchema = z.object({
  available: z.boolean(), can_write: z.boolean(), reason: z.string().nullable(), current: scadaReplaySchema.nullable(),
  monitoring_interval_seconds: z.number().int().min(60).max(31_536_000).nullable().default(null),
  burst_monitoring_ready: z.boolean().default(false),
  burst_monitoring_reason: z.string().nullable().default(null),
  monitoring_config: monitoringConfigSchema.nullable().optional()
});
export type ScadaMonitoringConfig = z.infer<typeof monitoringConfigSchema>;
export type ScadaMonitoringCalibration = z.infer<typeof calibrationSchema>;
export type ScadaReplay = z.infer<typeof scadaReplaySchema>;
export type ScadaReplayRule = z.infer<typeof ruleSchema>;
export type ScadaReplayEvent = z.infer<typeof eventSchema>;
export type ScadaMonitoringResult = z.infer<typeof monitoringSchema>;
export type ScadaReplayCapabilities = z.infer<typeof capabilitiesSchema>;
export type ScadaReplayAction = "step" | "stop" | "retry";

export class ScadaReplayClient {
  private projectId: Promise<string> | null = null;
  constructor(private readonly access?: WorkbenchAccess) {}

  async capabilities(signal?: AbortSignal) {
    return capabilitiesSchema.parse(await this.request("/capabilities", undefined, signal));
  }
  async get(id: string, signal?: AbortSignal) {
    return scadaReplaySchema.parse(await this.request(`/${encodeURIComponent(id)}`, undefined, signal));
  }
  async events(id: string, signal?: AbortSignal) {
    return z.object({ items: z.array(eventSchema), next_offset: z.number().nullable() })
      .parse(await this.request(`/${encodeURIComponent(id)}/events?limit=100&offset=0`, undefined, signal)).items;
  }
  async observations(id: string, through: string, signal?: AbortSignal) {
    return z.object({ items: z.array(readingSchema), through: z.string() }).parse(
      await this.request(`/${encodeURIComponent(id)}/observations?${new URLSearchParams({ through })}`, undefined, signal)
    ).items;
  }
  async create(input: { start_time: string; end_time: string; device_ids: string[]; rules: ScadaReplayRule[] }) {
    return scadaReplaySchema.parse(await this.request("", { request_id: crypto.randomUUID(), ...input }));
  }
  async command(replay: ScadaReplay, action: ScadaReplayAction) {
    return scadaReplaySchema.parse(await this.request(`/${replay.replay_id}/commands`, {
      request_id: crypto.randomUUID(), expected_revision: replay.revision, action
    }));
  }
  async monitorAt(replay: ScadaReplay, atTime: string) {
    return scadaReplaySchema.parse(await this.request(`/${replay.replay_id}/commands`, {
      request_id: crypto.randomUUID(), expected_revision: replay.revision,
      action: "monitor", at_time: atTime
    }));
  }
  private async request(path: string, body?: unknown, signal?: AbortSignal) {
    this.projectId ??= findOwnedWorkbenchProject(this.access).then(({ project }) => project.project_id)
      .catch((error) => { this.projectId = null; throw error; });
    const [projectId, token] = await Promise.all([this.projectId, getWorkbenchAccessToken(this.access)]);
    const headers = createWorkbenchHeaders(token, projectId);
    if (body !== undefined) headers.set("Content-Type", "application/json");
    const response = await fetch(`${env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, "")}/api/v1/scada-replays${path}`, {
      method: body === undefined ? "GET" : "POST", headers, signal,
      ...(body === undefined ? {} : { body: JSON.stringify(body) })
    });
    const payload: unknown = await response.json().catch(() => null);
    if (!response.ok) throw new Error(getServerError(payload, response.status, "SCADA 回放"));
    return payload;
  }
}