Compare commits

...
Author SHA1 Message Date
QiuSW b9b067213f feat: 建立 Brain 可替换视频解码流水线 (#13) 2026-08-28 22:27:04 +08:00
ila 52b368068e feat: 实现 Brain 合成与本地视频输入 (#11)
合入 dev,#11 保持待验收。
2026-08-28 22:20:04 +08:00
QiuSW 7964d61cab feat: 实现 Brain 合成与本地视频输入 (#11) 2026-08-28 22:19:22 +08:00
ila 4a605f6482 Merge pull request #123 from feat/sense-79-ops-alerts
feat: 实现 Sense 运维告警隔离 (#79)
2026-08-28 18:25:33 +08:00
QiuSW d61e6d5ee1 feat: 实现 Sense 运维告警隔离 (#79) 2026-08-28 18:25:07 +08:00
ila 7274bd42f5 Merge PR #122: Sense 可靠投递状态机
Implements #78. Issue remains open for user acceptance.
2026-08-28 17:23:01 +08:00
QiuSW 02a5af5e3b feat: 建立 Sense 可靠投递状态机 (#78) 2026-08-28 17:22:44 +08:00
ila e06904272a Merge PR #121: [SEN] MediaMTX 分片分配与故障范围可观察
实现工单 #77;保持工单待人工验收。
2026-08-28 16:10:21 +08:00
QiuSW d681fd1345 feat: 实现 MediaMTX 分片分配与故障范围可观察 (#77) 2026-08-28 16:09:52 +08:00
ila bdd78523b5 Merge PR #120: Sense 边缘节点资产与离线状态管理
Implements #76; remains pending user acceptance.
2026-08-28 14:29:55 +08:00
QiuSW 2db49fc955 feat: 实现边缘节点资产与离线状态管理 (#76) 2026-08-28 14:25:32 +08:00
ila a9fd8e8a47 Merge PR #119: Sense 容量与配额安全闸
合入 dev,工单 #75 保持待验收。
2026-08-28 11:55:51 +08:00
99 changed files with 5945 additions and 20 deletions
@@ -0,0 +1,27 @@
"""Brain-internal configuration models.
These types are deliberately not a cross-project contract. Adapters for a
future versioned Sense/Brain contract belong in a coordination task.
"""
from .models import (
AreaRule,
BrainInputConfig,
ConfigError,
DirectionalLineRule,
Point,
SourceConfig,
VideoProfile,
parse_input_config,
)
__all__ = [
"AreaRule",
"BrainInputConfig",
"ConfigError",
"DirectionalLineRule",
"Point",
"SourceConfig",
"VideoProfile",
"parse_input_config",
]
+206
View File
@@ -0,0 +1,206 @@
"""Validated project-internal input configuration."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Sequence
INTERNAL_SCHEMA = "brain.internal.input/v1"
_SECRET_KEYS = {"credential", "password", "secret", "token", "username"}
class ConfigError(ValueError):
"""Raised when Brain's project-internal test/runtime config is invalid."""
@dataclass(frozen=True, slots=True)
class Point:
x: float
y: float
@dataclass(frozen=True, slots=True)
class AreaRule:
rule_id: str
points: tuple[Point, ...]
@dataclass(frozen=True, slots=True)
class DirectionalLineRule:
rule_id: str
start: Point
end: Point
trigger_direction: str
@dataclass(frozen=True, slots=True)
class VideoProfile:
profile_id: str
width: int
height: int
fps: float
@dataclass(frozen=True, slots=True)
class SourceConfig:
kind: str
seed: int | None = None
frame_count: int | None = None
path: Path | None = None
chunk_size: int = 64 * 1024
@dataclass(frozen=True, slots=True)
class BrainInputConfig:
schema: str
logical_device_id: str
profile: VideoProfile
source: SourceConfig
areas: tuple[AreaRule, ...] = ()
directional_lines: tuple[DirectionalLineRule, ...] = ()
def _mapping(value: object, field: str) -> Mapping[str, Any]:
if not isinstance(value, Mapping):
raise ConfigError(f"{field} must be an object")
return value
def _text(value: object, field: str) -> str:
if not isinstance(value, str) or not value.strip():
raise ConfigError(f"{field} must be a non-empty string")
return value.strip()
def _integer(value: object, field: str, *, minimum: int = 1) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
raise ConfigError(f"{field} must be an integer >= {minimum}")
return value
def _number(value: object, field: str, *, minimum: float = 0.0) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ConfigError(f"{field} must be a number")
result = float(value)
if result <= minimum:
raise ConfigError(f"{field} must be greater than {minimum}")
return result
def _reject_secrets(value: object, field: str = "config") -> None:
if isinstance(value, Mapping):
for key, child in value.items():
normalized = str(key).strip().lower()
if normalized in _SECRET_KEYS or any(
marker in normalized for marker in ("password", "secret", "token")
):
raise ConfigError(f"{field} must not contain credential field {key!r}")
_reject_secrets(child, f"{field}.{key}")
elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
for index, child in enumerate(value):
_reject_secrets(child, f"{field}[{index}]")
def _point(value: object, field: str) -> Point:
if (
not isinstance(value, Sequence)
or isinstance(value, (str, bytes, bytearray))
or len(value) != 2
):
raise ConfigError(f"{field} must be [x, y]")
x, y = value
if isinstance(x, bool) or isinstance(y, bool) or not isinstance(x, (int, float)) or not isinstance(y, (int, float)):
raise ConfigError(f"{field} coordinates must be numbers")
point = Point(float(x), float(y))
if not 0.0 <= point.x <= 1.0 or not 0.0 <= point.y <= 1.0:
raise ConfigError(f"{field} coordinates must be normalized to 0..1")
return point
def parse_input_config(raw: Mapping[str, Any], *, base_dir: Path | None = None) -> BrainInputConfig:
"""Parse the explicitly versioned Brain-internal input configuration."""
_reject_secrets(raw)
schema = _text(raw.get("schema"), "schema")
if schema != INTERNAL_SCHEMA:
raise ConfigError(f"schema must be {INTERNAL_SCHEMA!r}")
profile_raw = _mapping(raw.get("profile"), "profile")
profile = VideoProfile(
profile_id=_text(profile_raw.get("id"), "profile.id"),
width=_integer(profile_raw.get("width"), "profile.width"),
height=_integer(profile_raw.get("height"), "profile.height"),
fps=_number(profile_raw.get("fps"), "profile.fps"),
)
source_raw = _mapping(raw.get("source"), "source")
kind = _text(source_raw.get("kind"), "source.kind")
if kind == "synthetic":
seed = source_raw.get("seed", 0)
if isinstance(seed, bool) or not isinstance(seed, int):
raise ConfigError("source.seed must be an integer")
source = SourceConfig(
kind=kind,
seed=seed,
frame_count=_integer(source_raw.get("frame_count"), "source.frame_count"),
)
elif kind == "local_file":
configured_path = Path(_text(source_raw.get("path"), "source.path"))
if not configured_path.is_absolute() and base_dir is not None:
configured_path = base_dir / configured_path
source = SourceConfig(
kind=kind,
path=configured_path,
chunk_size=_integer(source_raw.get("chunk_size", 64 * 1024), "source.chunk_size"),
)
else:
raise ConfigError("source.kind must be 'synthetic' or 'local_file'")
areas_raw = raw.get("areas", [])
if not isinstance(areas_raw, list):
raise ConfigError("areas must be an array")
areas: list[AreaRule] = []
for index, item in enumerate(areas_raw):
area = _mapping(item, f"areas[{index}]")
points_raw = area.get("points")
if not isinstance(points_raw, list) or len(points_raw) < 3:
raise ConfigError(f"areas[{index}].points must contain at least three points")
areas.append(
AreaRule(
rule_id=_text(area.get("id"), f"areas[{index}].id"),
points=tuple(_point(point, f"areas[{index}].points[{point_index}]") for point_index, point in enumerate(points_raw)),
)
)
lines_raw = raw.get("directional_lines", [])
if not isinstance(lines_raw, list):
raise ConfigError("directional_lines must be an array")
lines: list[DirectionalLineRule] = []
for index, item in enumerate(lines_raw):
line = _mapping(item, f"directional_lines[{index}]")
direction = _text(line.get("trigger_direction"), f"directional_lines[{index}].trigger_direction")
if direction not in {"left_to_right", "right_to_left"}:
raise ConfigError(
f"directional_lines[{index}].trigger_direction must be left_to_right or right_to_left"
)
lines.append(
DirectionalLineRule(
rule_id=_text(line.get("id"), f"directional_lines[{index}].id"),
start=_point(line.get("start"), f"directional_lines[{index}].start"),
end=_point(line.get("end"), f"directional_lines[{index}].end"),
trigger_direction=direction,
)
)
identifiers = [area.rule_id for area in areas] + [line.rule_id for line in lines]
if len(set(identifiers)) != len(identifiers):
raise ConfigError("rule ids must be unique")
return BrainInputConfig(
schema=schema,
logical_device_id=_text(raw.get("logical_device_id"), "logical_device_id"),
profile=profile,
source=source,
areas=tuple(areas),
directional_lines=tuple(lines),
)
@@ -0,0 +1,16 @@
"""Replaceable Brain-internal video decode pipeline."""
from .models import DecodedFrame, DecoderBackend, DecoderError
from .pipeline import DecoderPipeline, decode_packets
from .raw_rgb import RawRGBDecoder
from .y4m import Y4MDecoder
__all__ = [
"DecodedFrame",
"DecoderBackend",
"DecoderError",
"DecoderPipeline",
"RawRGBDecoder",
"Y4MDecoder",
"decode_packets",
]
+35
View File
@@ -0,0 +1,35 @@
"""Decode-layer ports and frame model."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterable, Iterator, Protocol
from yovision_brain.input import CancellationToken, InputPacket
class DecoderError(RuntimeError):
"""A safe and actionable decode failure."""
@dataclass(frozen=True, slots=True)
class DecodedFrame:
sequence: int
timestamp_ns: int
logical_device_id: str
profile_id: str
width: int
height: int
pixel_format: str
payload: bytes
dimensions_changed: bool = False
class DecoderBackend(Protocol):
media_formats: frozenset[str]
def decode(
self,
packets: Iterable[InputPacket],
cancellation: CancellationToken | None = None,
) -> Iterator[DecodedFrame]: ...
@@ -0,0 +1,47 @@
"""Decoder selection independent of concrete codec libraries."""
from __future__ import annotations
from collections.abc import Iterable, Iterator
from itertools import chain
from yovision_brain.input import CancellationToken, InputPacket
from .models import DecodedFrame, DecoderBackend, DecoderError
from .raw_rgb import RawRGBDecoder
from .y4m import Y4MDecoder
class DecoderPipeline:
def __init__(self, backends: Iterable[DecoderBackend] | None = None) -> None:
selected = tuple(backends) if backends is not None else (RawRGBDecoder(), Y4MDecoder())
self._backends: dict[str, DecoderBackend] = {}
for backend in selected:
for media_format in backend.media_formats:
if media_format in self._backends:
raise ValueError(f"duplicate decoder for media format {media_format!r}")
self._backends[media_format] = backend
def decode(
self,
packets: Iterable[InputPacket],
cancellation: CancellationToken | None = None,
) -> Iterator[DecodedFrame]:
iterator = iter(packets)
if cancellation is not None and cancellation.cancelled:
return
try:
first = next(iterator)
except StopIteration:
return
backend = self._backends.get(first.media_format)
if backend is None:
raise DecoderError(f"no decoder registered for media format {first.media_format!r}")
yield from backend.decode(chain((first,), iterator), cancellation)
def decode_packets(
packets: Iterable[InputPacket],
cancellation: CancellationToken | None = None,
) -> Iterator[DecodedFrame]:
return DecoderPipeline().decode(packets, cancellation)
@@ -0,0 +1,45 @@
"""Pass-through decoder for deterministic RGB24 synthetic frames."""
from __future__ import annotations
from collections.abc import Iterable, Iterator
from yovision_brain.input import CancellationToken, InputPacket
from .models import DecodedFrame, DecoderError
class RawRGBDecoder:
media_formats = frozenset({"rgb24"})
def decode(
self,
packets: Iterable[InputPacket],
cancellation: CancellationToken | None = None,
) -> Iterator[DecodedFrame]:
previous_dimensions: tuple[int, int] | None = None
for packet in packets:
if cancellation is not None and cancellation.cancelled:
return
if packet.media_format != "rgb24":
raise DecoderError(f"raw RGB decoder does not support {packet.media_format!r}")
expected = packet.width * packet.height * 3
if len(packet.payload) != expected:
raise DecoderError(
f"RGB24 frame {packet.sequence} has {len(packet.payload)} bytes; expected {expected}"
)
if packet.timestamp_ns is None:
raise DecoderError(f"RGB24 frame {packet.sequence} has no source timestamp")
dimensions = (packet.width, packet.height)
yield DecodedFrame(
sequence=packet.sequence,
timestamp_ns=packet.timestamp_ns,
logical_device_id=packet.logical_device_id,
profile_id=packet.profile_id,
width=packet.width,
height=packet.height,
pixel_format="rgb24",
payload=packet.payload,
dimensions_changed=previous_dimensions is not None and dimensions != previous_dimensions,
)
previous_dimensions = dimensions
+167
View File
@@ -0,0 +1,167 @@
"""Minimal streaming YUV4MPEG2 decoder for anonymous local fixtures.
The backend intentionally supports only uncompressed C444 streams. Production
codecs and RTSP belong behind the same decoder port in later tasks.
"""
from __future__ import annotations
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from yovision_brain.input import CancellationToken, InputPacket
from .models import DecodedFrame, DecoderError
_MAX_HEADER_BYTES = 4096
_MAX_FRAME_BYTES = 256 * 1024 * 1024
class _Cancelled(Exception):
pass
class _PacketReader:
def __init__(
self,
packets: Iterable[InputPacket],
cancellation: CancellationToken | None,
) -> None:
self._packets = iter(packets)
self._cancellation = cancellation
self._buffer = bytearray()
self._ended = False
self.first_packet: InputPacket | None = None
def _fill(self) -> bool:
if self._cancellation is not None and self._cancellation.cancelled:
raise _Cancelled
if self._ended:
return False
try:
packet = next(self._packets)
except StopIteration:
self._ended = True
return False
if packet.media_format != "container-bytes":
raise DecoderError(f"Y4M decoder does not support {packet.media_format!r}")
if self.first_packet is None:
self.first_packet = packet
else:
first = self.first_packet
if (packet.logical_device_id, packet.profile_id) != (
first.logical_device_id,
first.profile_id,
):
raise DecoderError("input identity changed inside one local video stream")
self._buffer.extend(packet.payload)
return True
def line(self, *, allow_clean_eof: bool = False) -> bytes | None:
while True:
newline = self._buffer.find(b"\n")
if newline >= 0:
result = bytes(self._buffer[:newline])
del self._buffer[: newline + 1]
return result
if len(self._buffer) > _MAX_HEADER_BYTES:
raise DecoderError("Y4M header exceeds the safe size limit")
if not self._fill():
if not self._buffer and allow_clean_eof:
return None
raise DecoderError("truncated Y4M header")
def exact(self, size: int) -> bytes:
while len(self._buffer) < size:
if not self._fill():
raise DecoderError("truncated Y4M frame payload")
result = bytes(self._buffer[:size])
del self._buffer[:size]
return result
@dataclass(frozen=True, slots=True)
class _Header:
width: int
height: int
fps_numerator: int
fps_denominator: int
def _positive_int(value: bytes, field: str) -> int:
try:
result = int(value)
except ValueError as exc:
raise DecoderError(f"invalid Y4M {field}") from exc
if result <= 0:
raise DecoderError(f"invalid Y4M {field}")
return result
def _parse_header(line: bytes) -> _Header:
parts = line.split()
if not parts or parts[0] != b"YUV4MPEG2":
raise DecoderError("unsupported local video format; expected YUV4MPEG2")
fields = {part[:1]: part[1:] for part in parts[1:] if len(part) > 1}
if fields.get(b"C", b"444") not in {b"444", b"444jpeg"}:
raise DecoderError("unsupported Y4M chroma; only C444 is supported")
width = _positive_int(fields.get(b"W", b""), "width")
height = _positive_int(fields.get(b"H", b""), "height")
fps_parts = fields.get(b"F", b"").split(b":", 1)
if len(fps_parts) != 2:
raise DecoderError("invalid Y4M frame rate")
header = _Header(
width=width,
height=height,
fps_numerator=_positive_int(fps_parts[0], "frame rate numerator"),
fps_denominator=_positive_int(fps_parts[1], "frame rate denominator"),
)
if header.width * header.height * 3 > _MAX_FRAME_BYTES:
raise DecoderError("Y4M frame exceeds the safe size limit")
return header
class Y4MDecoder:
media_formats = frozenset({"container-bytes"})
def decode(
self,
packets: Iterable[InputPacket],
cancellation: CancellationToken | None = None,
) -> Iterator[DecodedFrame]:
reader = _PacketReader(packets, cancellation)
try:
header_line = reader.line()
assert header_line is not None
header = _parse_header(header_line)
first = reader.first_packet
if first is None:
raise DecoderError("local video input is empty")
if (first.width, first.height) != (header.width, header.height):
raise DecoderError(
"Y4M dimensions do not match the configured input profile "
f"({header.width}x{header.height} != {first.width}x{first.height})"
)
interval_ns = round(1_000_000_000 * header.fps_denominator / header.fps_numerator)
frame_size = header.width * header.height * 3
sequence = 0
while True:
frame_header = reader.line(allow_clean_eof=True)
if frame_header is None:
return
if frame_header != b"FRAME":
raise DecoderError(f"invalid Y4M frame header at frame {sequence}")
payload = reader.exact(frame_size)
yield DecodedFrame(
sequence=sequence,
timestamp_ns=sequence * interval_ns,
logical_device_id=first.logical_device_id,
profile_id=first.profile_id,
width=header.width,
height=header.height,
pixel_format="yuv444p",
payload=payload,
)
sequence += 1
except _Cancelled:
return
@@ -0,0 +1,16 @@
"""Brain-internal video input adapters."""
from .factory import build_input_source
from .local_file import LocalFileInput
from .models import CancellationToken, InputError, InputPacket, InputSource
from .synthetic import SyntheticInput
__all__ = [
"CancellationToken",
"InputError",
"InputPacket",
"InputSource",
"LocalFileInput",
"SyntheticInput",
"build_input_source",
]
+17
View File
@@ -0,0 +1,17 @@
"""Construct the configured Brain-internal input adapter."""
from __future__ import annotations
from yovision_brain.config import BrainInputConfig
from .local_file import LocalFileInput
from .models import InputSource
from .synthetic import SyntheticInput
def build_input_source(config: BrainInputConfig) -> InputSource:
if config.source.kind == "synthetic":
return SyntheticInput(config)
if config.source.kind == "local_file":
return LocalFileInput(config)
raise ValueError(f"unsupported Brain input source kind: {config.source.kind}")
@@ -0,0 +1,54 @@
"""Explicit local-file input adapter with safe error reporting."""
from __future__ import annotations
from collections.abc import Iterator
from pathlib import Path
from yovision_brain.config import BrainInputConfig
from .models import CancellationToken, InputError, InputPacket
class LocalFileInput:
def __init__(self, config: BrainInputConfig) -> None:
if config.source.kind != "local_file" or config.source.path is None:
raise ValueError("LocalFileInput requires a local_file source config")
self._config = config
self._path = config.source.path
@property
def source_label(self) -> str:
"""Return a safe label rather than exposing the internal absolute path."""
return self._path.name
def packets(self, cancellation: CancellationToken | None = None) -> Iterator[InputPacket]:
profile = self._config.profile
source = self._config.source
try:
stream = self._path.open("rb")
except FileNotFoundError as exc:
raise InputError(f"local video source not found: {self.source_label}") from exc
except OSError as exc:
raise InputError(f"local video source cannot be opened: {self.source_label}: {exc.strerror}") from exc
with stream:
sequence = 0
while cancellation is None or not cancellation.cancelled:
try:
payload = stream.read(source.chunk_size)
except OSError as exc:
raise InputError(f"local video source read failed: {self.source_label}: {exc.strerror}") from exc
if not payload:
return
yield InputPacket(
sequence=sequence,
timestamp_ns=None,
logical_device_id=self._config.logical_device_id,
profile_id=profile.profile_id,
width=profile.width,
height=profile.height,
media_format="container-bytes",
payload=payload,
)
sequence += 1
+43
View File
@@ -0,0 +1,43 @@
"""Common project-internal input types."""
from __future__ import annotations
from dataclasses import dataclass
from threading import Event
from typing import Iterator, Protocol
class InputError(RuntimeError):
"""A safe, actionable input adapter error."""
class CancellationToken:
"""Thread-safe cooperative cancellation without platform dependencies."""
def __init__(self) -> None:
self._event = Event()
def cancel(self) -> None:
self._event.set()
@property
def cancelled(self) -> bool:
return self._event.is_set()
@dataclass(frozen=True, slots=True)
class InputPacket:
sequence: int
timestamp_ns: int | None
logical_device_id: str
profile_id: str
width: int
height: int
media_format: str
payload: bytes
class InputSource(Protocol):
"""Replaceable source boundary consumed by the future decode layer."""
def packets(self, cancellation: CancellationToken | None = None) -> Iterator[InputPacket]: ...
@@ -0,0 +1,38 @@
"""Deterministic synthetic RGB input for isolated tests and smoke runs."""
from __future__ import annotations
import random
from collections.abc import Iterator
from yovision_brain.config import BrainInputConfig
from .models import CancellationToken, InputPacket
class SyntheticInput:
def __init__(self, config: BrainInputConfig) -> None:
if config.source.kind != "synthetic":
raise ValueError("SyntheticInput requires a synthetic source config")
self._config = config
def packets(self, cancellation: CancellationToken | None = None) -> Iterator[InputPacket]:
source = self._config.source
assert source.seed is not None and source.frame_count is not None
randomizer = random.Random(source.seed)
profile = self._config.profile
frame_size = profile.width * profile.height * 3
interval_ns = round(1_000_000_000 / profile.fps)
for sequence in range(source.frame_count):
if cancellation is not None and cancellation.cancelled:
return
yield InputPacket(
sequence=sequence,
timestamp_ns=sequence * interval_ns,
logical_device_id=self._config.logical_device_id,
profile_id=profile.profile_id,
width=profile.width,
height=profile.height,
media_format="rgb24",
payload=randomizer.randbytes(frame_size),
)
+72
View File
@@ -0,0 +1,72 @@
from __future__ import annotations
from pathlib import Path
import pytest
from yovision_brain.config import ConfigError, parse_input_config
def valid_config() -> dict[str, object]:
return {
"schema": "brain.internal.input/v1",
"logical_device_id": "synthetic-camera-01",
"profile": {"id": "main", "width": 4, "height": 3, "fps": 5},
"source": {"kind": "synthetic", "seed": 17, "frame_count": 3},
"areas": [{"id": "danger-yard", "points": [[0.1, 0.1], [0.9, 0.1], [0.5, 0.8]]}],
"directional_lines": [
{
"id": "gate-line",
"start": [0.2, 0.5],
"end": [0.8, 0.5],
"trigger_direction": "left_to_right",
}
],
}
def test_parse_versioned_internal_config() -> None:
parsed = parse_input_config(valid_config())
assert parsed.schema == "brain.internal.input/v1"
assert parsed.logical_device_id == "synthetic-camera-01"
assert parsed.profile.width == 4
assert parsed.areas[0].rule_id == "danger-yard"
assert parsed.directional_lines[0].trigger_direction == "left_to_right"
def test_relative_local_path_is_bound_to_explicit_base(tmp_path: Path) -> None:
raw = valid_config()
raw["source"] = {"kind": "local_file", "path": "fixture.bin", "chunk_size": 8}
parsed = parse_input_config(raw, base_dir=tmp_path)
assert parsed.source.path == tmp_path / "fixture.bin"
@pytest.mark.parametrize(
("change", "message"),
[
({"schema": "shared.source/v1"}, "schema must be"),
({"logical_device_id": ""}, "logical_device_id"),
({"source": {"kind": "synthetic", "seed": 1, "frame_count": 0}}, "frame_count"),
({"password": "must-not-be-accepted"}, "credential field"),
],
)
def test_invalid_or_secret_config_is_rejected(change: dict[str, object], message: str) -> None:
raw = valid_config()
raw.update(change)
with pytest.raises(ConfigError, match=message):
parse_input_config(raw)
def test_coordinates_and_rule_ids_are_validated() -> None:
raw = valid_config()
raw["areas"] = [{"id": "same", "points": [[0, 0], [2, 0], [0, 1]]}]
with pytest.raises(ConfigError, match="normalized"):
parse_input_config(raw)
raw = valid_config()
raw["areas"] = [{"id": "same", "points": [[0, 0], [1, 0], [0, 1]]}]
raw["directional_lines"] = [
{"id": "same", "start": [0, 0], "end": [1, 1], "trigger_direction": "left_to_right"}
]
with pytest.raises(ConfigError, match="unique"):
parse_input_config(raw)
+104
View File
@@ -0,0 +1,104 @@
from __future__ import annotations
from pathlib import Path
import pytest
from yovision_brain.config import parse_input_config
from yovision_brain.decode import DecoderError, DecoderPipeline, decode_packets
from yovision_brain.input import CancellationToken, InputPacket, LocalFileInput, SyntheticInput
def synthetic_packets():
config = parse_input_config(
{
"schema": "brain.internal.input/v1",
"logical_device_id": "synthetic-01",
"profile": {"id": "main", "width": 2, "height": 1, "fps": 5},
"source": {"kind": "synthetic", "seed": 3, "frame_count": 2},
}
)
return SyntheticInput(config).packets()
def local_packets(path: Path, *, width: int = 2, height: int = 1, chunk_size: int = 5):
config = parse_input_config(
{
"schema": "brain.internal.input/v1",
"logical_device_id": "local-01",
"profile": {"id": "archive", "width": width, "height": height, "fps": 25},
"source": {"kind": "local_file", "path": str(path), "chunk_size": chunk_size},
}
)
return LocalFileInput(config).packets()
def test_rgb24_pipeline_preserves_order_timestamps_and_metadata() -> None:
frames = list(decode_packets(synthetic_packets()))
assert [frame.sequence for frame in frames] == [0, 1]
assert [frame.timestamp_ns for frame in frames] == [0, 200_000_000]
assert all(frame.logical_device_id == "synthetic-01" for frame in frames)
assert all(frame.profile_id == "main" for frame in frames)
assert all((frame.width, frame.height, frame.pixel_format) == (2, 1, "rgb24") for frame in frames)
def test_rgb24_dimension_change_is_explicit() -> None:
packets = [
InputPacket(0, 0, "camera", "main", 1, 1, "rgb24", b"abc"),
InputPacket(1, 1, "camera", "main", 2, 1, "rgb24", b"abcdef"),
]
frames = list(decode_packets(packets))
assert [frame.dimensions_changed for frame in frames] == [False, True]
def test_invalid_rgb_payload_and_unsupported_format_are_clear() -> None:
bad = [InputPacket(0, 0, "camera", "main", 2, 2, "rgb24", b"short")]
with pytest.raises(DecoderError, match="expected 12"):
list(decode_packets(bad))
unknown = [InputPacket(0, 0, "camera", "main", 1, 1, "opaque", b"data")]
with pytest.raises(DecoderError, match="no decoder registered"):
list(DecoderPipeline().decode(unknown))
def test_y4m_local_video_decodes_across_input_chunks(tmp_path: Path) -> None:
video = tmp_path / "anonymous.y4m"
first, second = b"abcdef", b"ghijkl"
video.write_bytes(b"YUV4MPEG2 W2 H1 F25:1 C444\nFRAME\n" + first + b"FRAME\n" + second)
frames = list(decode_packets(local_packets(video)))
assert [frame.payload for frame in frames] == [first, second]
assert [frame.timestamp_ns for frame in frames] == [0, 40_000_000]
assert all(frame.pixel_format == "yuv444p" for frame in frames)
assert all((frame.width, frame.height) == (2, 1) for frame in frames)
assert all(frame.profile_id == "archive" for frame in frames)
def test_y4m_clean_eof_and_cancellation_are_normal(tmp_path: Path) -> None:
video = tmp_path / "empty.y4m"
video.write_bytes(b"YUV4MPEG2 W2 H1 F25:1 C444\n")
assert list(decode_packets(local_packets(video))) == []
token = CancellationToken()
token.cancel()
assert list(decode_packets(local_packets(video), token)) == []
@pytest.mark.parametrize(
("payload", "message"),
[
(b"not-video\n", "expected YUV4MPEG2"),
(b"YUV4MPEG2 W2 H1 F25:1 C420\n", "only C444"),
(b"YUV4MPEG2 W2 H1 F25:1 C444\nFRAME\nabc", "truncated Y4M frame"),
],
)
def test_y4m_damage_and_unsupported_content_are_clear(tmp_path: Path, payload: bytes, message: str) -> None:
video = tmp_path / "broken.y4m"
video.write_bytes(payload)
with pytest.raises(DecoderError, match=message):
list(decode_packets(local_packets(video)))
def test_y4m_profile_dimension_mismatch_is_rejected(tmp_path: Path) -> None:
video = tmp_path / "mismatch.y4m"
video.write_bytes(b"YUV4MPEG2 W2 H1 F25:1 C444\n")
with pytest.raises(DecoderError, match="do not match"):
list(decode_packets(local_packets(video, width=3)))
+5
View File
@@ -0,0 +1,5 @@
# Brain decode fixtures
Decode tests generate tiny anonymous YUV4MPEG2 streams at runtime. Do not add
customer recordings, camera credentials, machine-specific codec paths, or
large model/media artifacts to this directory.
+7
View File
@@ -0,0 +1,7 @@
# Brain input fixtures
This directory may contain only synthetic or anonymous fixtures. Do not add
customer video, camera credentials, personal data, or machine-specific paths.
The current tests generate their tiny local-file payload at runtime so the
repository does not carry a file that could be mistaken for customer media.
+83
View File
@@ -0,0 +1,83 @@
from __future__ import annotations
from pathlib import Path
import pytest
from yovision_brain.config import parse_input_config
from yovision_brain.input import CancellationToken, InputError, LocalFileInput, SyntheticInput, build_input_source
def synthetic_config(seed: int = 9, frame_count: int = 3):
return parse_input_config(
{
"schema": "brain.internal.input/v1",
"logical_device_id": "camera-lab-01",
"profile": {"id": "main", "width": 3, "height": 2, "fps": 4},
"source": {"kind": "synthetic", "seed": seed, "frame_count": frame_count},
}
)
def local_config(path: Path, *, chunk_size: int = 4):
return parse_input_config(
{
"schema": "brain.internal.input/v1",
"logical_device_id": "local-video-01",
"profile": {"id": "archive", "width": 1920, "height": 1080, "fps": 25},
"source": {"kind": "local_file", "path": str(path), "chunk_size": chunk_size},
}
)
def test_synthetic_input_is_deterministic_and_carries_metadata() -> None:
first = list(SyntheticInput(synthetic_config()).packets())
second = list(build_input_source(synthetic_config()).packets())
different = list(SyntheticInput(synthetic_config(seed=10)).packets())
assert first == second
assert [packet.sequence for packet in first] == [0, 1, 2]
assert [packet.timestamp_ns for packet in first] == [0, 250_000_000, 500_000_000]
assert first[0].logical_device_id == "camera-lab-01"
assert first[0].profile_id == "main"
assert first[0].media_format == "rgb24"
assert len(first[0].payload) == 3 * 2 * 3
assert first[0].payload != different[0].payload
def test_synthetic_input_honors_cancellation() -> None:
token = CancellationToken()
packets = SyntheticInput(synthetic_config(frame_count=10)).packets(token)
assert next(packets).sequence == 0
token.cancel()
assert list(packets) == []
def test_local_file_input_reads_chunks_and_finishes_at_eof(tmp_path: Path) -> None:
video = tmp_path / "anonymous-fixture.bin"
video.write_bytes(b"abcdefghij")
source = LocalFileInput(local_config(video))
packets = list(source.packets())
assert source.source_label == "anonymous-fixture.bin"
assert [packet.payload for packet in packets] == [b"abcd", b"efgh", b"ij"]
assert [packet.sequence for packet in packets] == [0, 1, 2]
assert all(packet.timestamp_ns is None for packet in packets)
assert all(packet.media_format == "container-bytes" for packet in packets)
def test_local_file_input_honors_cancellation(tmp_path: Path) -> None:
video = tmp_path / "anonymous-fixture.bin"
video.write_bytes(b"abcdefghij")
token = CancellationToken()
packets = LocalFileInput(local_config(video)).packets(token)
assert next(packets).payload == b"abcd"
token.cancel()
assert list(packets) == []
def test_missing_file_error_is_actionable_without_absolute_path(tmp_path: Path) -> None:
missing = tmp_path / "missing-video.mp4"
with pytest.raises(InputError, match="local video source not found: missing-video.mp4") as caught:
list(LocalFileInput(local_config(missing)).packets())
assert str(tmp_path) not in str(caught.value)
@@ -0,0 +1,9 @@
[
{
"id": "secondary",
"name": "备用媒体分片",
"mode": "external",
"controlAPI": "http://127.0.0.1:19997",
"capacity": 24
}
]
+2
View File
@@ -13,6 +13,8 @@ SENSE_MEDIAMTX_MODE=disabled
SENSE_MEDIAMTX_BINARY=
SENSE_MEDIAMTX_CONFIG=
SENSE_MEDIAMTX_API=http://127.0.0.1:9997
SENSE_MEDIAMTX_CAPACITY=
SENSE_MEDIAMTX_SHARDS_FILE=
SENSE_WEB_ROOT=web
SENSE_AUTO_MIGRATE=true
SENSE_POSTGRES_BIN=
+4
View File
@@ -13,6 +13,10 @@ SENSE_MEDIAMTX_MODE=managed
SENSE_MEDIAMTX_BINARY=bin\mediamtx.exe
SENSE_MEDIAMTX_CONFIG=config\mediamtx.yml
SENSE_MEDIAMTX_API=http://127.0.0.1:9997
# Leave capacity empty to use the current database quota. Optional extra
# shards are read from a repository-external JSON file and must use loopback APIs.
SENSE_MEDIAMTX_CAPACITY=
SENSE_MEDIAMTX_SHARDS_FILE=
SENSE_WEB_ROOT=web
SENSE_AUTO_MIGRATE=true
SENSE_POSTGRES_BIN=
+1
View File
@@ -81,6 +81,7 @@ try {
Copy-Item -LiteralPath (Join-Path $senseRoot 'config\sense.demo.env.example') -Destination (Join-Path $staging 'config\sense.demo.env.example')
Copy-Item -LiteralPath (Join-Path $senseRoot 'config\sense.demo.env.example') -Destination (Join-Path $staging 'config\sense.demo.env')
Copy-Item -LiteralPath (Join-Path $senseRoot 'config\mediamtx.yml') -Destination (Join-Path $staging 'config\mediamtx.yml')
Copy-Item -LiteralPath (Join-Path $senseRoot 'config\mediamtx-shards.example.json') -Destination (Join-Path $staging 'config\mediamtx-shards.example.json')
Copy-Item -LiteralPath (Join-Path $serverRoot 'config\db.sql') -Destination (Join-Path $staging 'config\db.sql')
Copy-Item -LiteralPath (Join-Path $serverRoot 'config\pg.sql') -Destination (Join-Path $staging 'config\pg.sql')
Copy-Item -LiteralPath (Join-Path $senseRoot 'README-WINDOWS.md') -Destination (Join-Path $staging 'README-WINDOWS.md')
+1 -1
View File
@@ -9,7 +9,7 @@ $required = @(
'migrate-sense.bat', 'backup-sense.bat', 'restore-sense.bat',
'initialize-admin.bat', 'README-WINDOWS.md', 'config\sense.env',
'config\sense.env.example', 'config\sense.demo.env',
'config\mediamtx.yml', 'config\db.sql', 'config\pg.sql',
'config\mediamtx.yml', 'config\mediamtx-shards.example.json', 'config\db.sql', 'config\pg.sql',
'web\index.html', 'scripts\runtime\sense-common.ps1'
)
foreach ($relative in $required) {
+7 -1
View File
@@ -7,7 +7,8 @@ $script:SenseAllowedEnvironment = @(
'SENSE_CREDENTIAL_KEY', 'SENSE_ONVIF_DISCOVERY_IP',
'SENSE_ONVIF_ALLOWED_CIDRS', 'SENSE_MEDIAMTX_MODE',
'SENSE_MEDIAMTX_BINARY', 'SENSE_MEDIAMTX_CONFIG',
'SENSE_MEDIAMTX_API', 'SENSE_WEB_ROOT', 'SENSE_AUTO_MIGRATE',
'SENSE_MEDIAMTX_API', 'SENSE_MEDIAMTX_CAPACITY',
'SENSE_MEDIAMTX_SHARDS_FILE', 'SENSE_WEB_ROOT', 'SENSE_AUTO_MIGRATE',
'SENSE_POSTGRES_BIN'
)
@@ -185,6 +186,7 @@ function Initialize-SenseRuntime {
}
$mediaBinary = Resolve-SenseConfiguredPath -PackageRoot $PackageRoot -Value (Get-SenseEnvironmentValue -Name 'SENSE_MEDIAMTX_BINARY')
$mediaConfig = Resolve-SenseConfiguredPath -PackageRoot $PackageRoot -Value (Get-SenseEnvironmentValue -Name 'SENSE_MEDIAMTX_CONFIG')
$mediaShardsFile = Resolve-SenseConfiguredPath -PackageRoot $PackageRoot -Value (Get-SenseEnvironmentValue -Name 'SENSE_MEDIAMTX_SHARDS_FILE')
if ($mediaMode -eq 'managed') {
if (-not (Test-Path -LiteralPath $mediaBinary -PathType Leaf)) { throw 'Managed MediaMTX binary not found. Set SENSE_MEDIAMTX_BINARY to mediamtx.exe.' }
if (-not (Test-Path -LiteralPath $mediaConfig -PathType Leaf)) { throw 'Managed MediaMTX configuration not found. Set SENSE_MEDIAMTX_CONFIG.' }
@@ -192,6 +194,9 @@ function Initialize-SenseRuntime {
if ($mediaMode -eq 'external' -and -not (Test-SenseTcpEndpoint -HostName $apiUri.Host -Port $apiUri.Port)) {
throw "External MediaMTX Control API is unreachable at $($apiUri.Host):$($apiUri.Port)."
}
if (-not [string]::IsNullOrWhiteSpace($mediaShardsFile) -and -not (Test-Path -LiteralPath $mediaShardsFile -PathType Leaf)) {
throw 'SENSE_MEDIAMTX_SHARDS_FILE does not exist.'
}
$webRoot = Resolve-SenseConfiguredPath -PackageRoot $PackageRoot -Value (Get-SenseEnvironmentValue -Name 'SENSE_WEB_ROOT' -Default 'web')
if (-not (Test-Path -LiteralPath (Join-Path $webRoot 'index.html') -PathType Leaf)) { throw 'Sense web assets are missing. Rebuild or replace the delivery package.' }
@@ -205,6 +210,7 @@ function Initialize-SenseRuntime {
[Environment]::SetEnvironmentVariable('SENSE_MEDIAMTX_CONFIG', '', 'Process')
}
[Environment]::SetEnvironmentVariable('SENSE_MEDIAMTX_API', $mediaAPI, 'Process')
[Environment]::SetEnvironmentVariable('SENSE_MEDIAMTX_SHARDS_FILE', $mediaShardsFile, 'Process')
$runtimeDir = Join-Path $PackageRoot 'data\runtime'
$logDir = Join-Path $PackageRoot 'logs'
@@ -0,0 +1,19 @@
package router
import (
"github.com/gin-gonic/gin"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/edge_node"
"git.ilapage.cn/ila/yovision/Sense/server/common/actions"
"git.ilapage.cn/ila/yovision/Sense/server/common/middleware"
)
func init() { routerCheckRole = append(routerCheckRole, registerSenseEdgeNodeRouter) }
func registerSenseEdgeNodeRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
api := &edge_node.API{}
r := v1.Group("/edge-nodes").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole()).Use(actions.PermissionAction())
r.GET("", api.List)
r.GET("/:id", api.Get)
}
@@ -0,0 +1,20 @@
package router
import (
"github.com/gin-gonic/gin"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
"git.ilapage.cn/ila/yovision/Sense/server/common/actions"
"git.ilapage.cn/ila/yovision/Sense/server/common/middleware"
)
func init() { routerCheckRole = append(routerCheckRole, registerSenseMediaShardRouter) }
func registerSenseMediaShardRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
api := &media_shard.API{}
r := v1.Group("/media-shards").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole()).Use(actions.PermissionAction())
r.GET("", api.List)
r.GET("/:id", api.Get)
r.GET("/:id/migration-preflight", api.Preflight)
}
@@ -0,0 +1,22 @@
package router
import (
"github.com/gin-gonic/gin"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/ops_alert"
"git.ilapage.cn/ila/yovision/Sense/server/common/actions"
"git.ilapage.cn/ila/yovision/Sense/server/common/middleware"
)
func init() { routerCheckRole = append(routerCheckRole, registerSenseOpsAlertRouter) }
func registerSenseOpsAlertRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
api := &ops_alert.API{}
r := v1.Group("/ops-alerts").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole()).Use(actions.PermissionAction())
r.GET("", api.List)
r.GET("/:id", api.Get)
r.POST("/evaluate", api.Evaluate)
r.POST("/:id/acknowledge", api.Acknowledge)
r.POST("/:id/recover", api.Recover)
}
@@ -0,0 +1,32 @@
package router
import (
"github.com/gin-gonic/gin"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
"net/http"
"testing"
)
func TestSenseOpsAlertRoutes(t *testing.T) {
gin.SetMode(gin.TestMode)
engine := gin.New()
registerSenseOpsAlertRouter(engine.Group("/api/v1"), &jwt.GinJWTMiddleware{})
wanted := map[string]bool{
http.MethodGet + " /api/v1/ops-alerts": false,
http.MethodGet + " /api/v1/ops-alerts/:id": false,
http.MethodPost + " /api/v1/ops-alerts/evaluate": false,
http.MethodPost + " /api/v1/ops-alerts/:id/acknowledge": false,
http.MethodPost + " /api/v1/ops-alerts/:id/recover": false,
}
for _, route := range engine.Routes() {
key := route.Method + " " + route.Path
if _, ok := wanted[key]; ok {
wanted[key] = true
}
}
for route, found := range wanted {
if !found {
t.Fatalf("route not registered: %s", route)
}
}
}
@@ -0,0 +1,19 @@
package router
import (
"github.com/gin-gonic/gin"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
"git.ilapage.cn/ila/yovision/Sense/server/common/actions"
"git.ilapage.cn/ila/yovision/Sense/server/common/middleware"
)
func init() { routerCheckRole = append(routerCheckRole, registerSenseOutboxRouter) }
func registerSenseOutboxRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
api := &outbox.API{}
r := v1.Group("/outbox").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole()).Use(actions.PermissionAction())
r.GET("", api.List)
r.GET("/:id", api.Get)
r.POST("/:id/requeue", api.Requeue)
}
@@ -0,0 +1,26 @@
package router
import (
"github.com/gin-gonic/gin"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
"net/http"
"testing"
)
func TestSenseOutboxRoutes(t *testing.T) {
gin.SetMode(gin.TestMode)
engine := gin.New()
registerSenseOutboxRouter(engine.Group("/api/v1"), &jwt.GinJWTMiddleware{})
wanted := map[string]bool{http.MethodGet + " /api/v1/outbox": false, http.MethodGet + " /api/v1/outbox/:id": false, http.MethodPost + " /api/v1/outbox/:id/requeue": false}
for _, route := range engine.Routes() {
key := route.Method + " " + route.Path
if _, ok := wanted[key]; ok {
wanted[key] = true
}
}
for route, found := range wanted {
if !found {
t.Fatalf("route not registered: %s", route)
}
}
}
+81
View File
@@ -0,0 +1,81 @@
package edge_node
import (
"errors"
"net/http"
"time"
"github.com/gin-gonic/gin"
"github.com/go-admin-team/go-admin-core/sdk/api"
"github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth/user"
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
"git.ilapage.cn/ila/yovision/Sense/server/common"
)
const auditSuccess, auditFailure = "1", "2"
type API struct{ api.Api }
func (e *API) service(c *gin.Context) (*Service, error) {
base := coreService.Service{}
if err := e.MakeContext(c).MakeOrm().MakeService(&base).Errors; err != nil {
return nil, err
}
return NewService(base.Orm), nil
}
func (e *API) List(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
request := PageRequest{}
if err = e.MakeContext(c).Bind(&request).Errors; err != nil {
e.audit(c, service, "List", auditFailure, "边缘节点查询条件格式不正确")
e.Error(http.StatusBadRequest, err, "查询条件格式不正确")
return
}
response, err := service.List(request)
if err != nil {
e.audit(c, service, "List", auditFailure, "边缘节点列表查询失败")
e.writeError(err)
return
}
e.audit(c, service, "List", auditSuccess, "读取边缘节点列表")
e.OK(response, "查询成功")
}
func (e *API) Get(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
item, err := service.Get(c.Param("id"))
if err != nil {
e.audit(c, service, "Get", auditFailure, "边缘节点详情查询失败")
e.writeError(err)
return
}
e.audit(c, service, "Get", auditSuccess, "读取边缘节点详情 "+item.ID)
e.OK(item, "查询成功")
}
func (e *API) audit(c *gin.Context, service *Service, action, status, remark string) {
if err := WriteAudit(service.Orm, Audit{Action: action, Method: c.Request.Method, Status: status, Username: user.GetUserName(c), UserID: user.GetUserId(c), ClientIP: common.GetClientIP(c), Route: c.FullPath(), Remark: remark, At: time.Now()}); err != nil {
api.GetRequestLogger(c).Errorf("edge node audit failed: %s", err.Error())
}
}
func (e *API) writeError(err error) {
switch {
case errors.Is(err, ErrInvalidQuery):
e.Error(http.StatusBadRequest, err, err.Error())
case errors.Is(err, ErrNotFound):
e.Error(http.StatusNotFound, err, err.Error())
default:
e.Error(http.StatusInternalServerError, err, "边缘节点查询失败")
}
}
+21
View File
@@ -0,0 +1,21 @@
package edge_node
import (
"time"
"gorm.io/gorm"
adminModels "git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
)
type Audit struct {
Action, Method, Status, Username, ClientIP, Route, Remark string
UserID int
At time.Time
}
func WriteAudit(db *gorm.DB, input Audit) error {
model := adminModels.SysOperaLog{Title: "边缘节点", BusinessType: "other", Method: "edge_node.API." + input.Action, RequestMethod: input.Method, OperatorType: "1", OperName: input.Username, OperUrl: input.Route, OperIp: input.ClientIP, Status: input.Status, OperTime: input.At.UTC(), Remark: input.Remark, CreatedAt: input.At.UTC(), UpdatedAt: input.At.UTC()}
model.CreateBy, model.UpdateBy = input.UserID, input.UserID
return db.Create(&model).Error
}
+65
View File
@@ -0,0 +1,65 @@
package edge_node
import (
"time"
commonDto "git.ilapage.cn/ila/yovision/Sense/server/common/dto"
)
const HeartbeatTimeout = 90 * time.Second
type PageRequest struct {
commonDto.Pagination `search:"-"`
}
type Summary struct {
Total int `json:"total"`
Online int `json:"online"`
Offline int `json:"offline"`
Recovering int `json:"recovering"`
LoadUsed int `json:"loadUsed"`
LoadCapacity int `json:"loadCapacity"`
BackfillQueueDepth int `json:"backfillQueueDepth"`
HeartbeatTimeout int64 `json:"heartbeatTimeoutSeconds"`
}
type Response struct {
ID string `json:"id"`
Name string `json:"name"`
Location string `json:"location"`
Status string `json:"status"`
RuntimeVersion string `json:"runtimeVersion"`
ModelVersion string `json:"modelVersion"`
StartedAt time.Time `json:"startedAt"`
UptimeSeconds int64 `json:"uptimeSeconds"`
LastHeartbeatAt time.Time `json:"lastHeartbeatAt"`
LastCollectedAt time.Time `json:"lastCollectedAt"`
FreshnessSeconds int64 `json:"freshnessSeconds"`
Stale bool `json:"stale"`
LoadUsed int `json:"loadUsed"`
LoadCapacity int `json:"loadCapacity"`
ControlTunnelStatus string `json:"controlTunnelStatus"`
ControlTunnelDetail string `json:"controlTunnelDetail"`
ControlLatencyMs int `json:"controlLatencyMs"`
VideoPlaneStatus string `json:"videoPlaneStatus"`
VideoPlaneDetail string `json:"videoPlaneDetail"`
ActiveStreamCount int `json:"activeStreamCount"`
BackfillStatus string `json:"backfillStatus"`
BackfillQueueDepth int `json:"backfillQueueDepth"`
RecoveryPhase string `json:"recoveryPhase"`
ProjectionVersion int64 `json:"projectionVersion"`
Events []Event `json:"events,omitempty"`
}
type PageResponse struct {
Summary Summary `json:"summary"`
List []Response `json:"list"`
Count int64 `json:"count"`
}
// HeartbeatSample is an internal Sense adapter input. It is intentionally not
// exposed as an HTTP contract by this task.
type HeartbeatSample struct {
Node
CollectedAt time.Time
}
@@ -0,0 +1,20 @@
package edge_node
import (
"time"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
// SeedSyntheticFixture is explicit test/development data. Production startup
// and migrations never call it.
func SeedSyntheticFixture(db *gorm.DB, now time.Time) error {
now = now.UTC().Truncate(time.Second)
items := []Node{
{ID: "SEN-EDGE-01", Name: "主楼边缘节点", Location: "主楼机房", RuntimeVersion: "sense-edge 0.1.0", ModelVersion: "detector-v3", StartedAt: now.Add(-7 * 24 * time.Hour), LastHeartbeatAt: now.Add(-18 * time.Second), LastCollectedAt: now.Add(-18 * time.Second), LoadUsed: 6, LoadCapacity: 16, ControlTunnelStatus: ChannelReady, ControlTunnelDetail: "控制隧道正常", ControlLatencyMs: 24, VideoPlaneStatus: ChannelReady, VideoPlaneDetail: "6 路视频正常", ActiveStreamCount: 6, BackfillStatus: BackfillIdle, ProjectionVersion: 1},
{ID: "SEN-EDGE-02", Name: "仓库边缘节点", Location: "仓库弱电间", RuntimeVersion: "sense-edge 0.1.0", ModelVersion: "detector-v3", StartedAt: now.Add(-3 * 24 * time.Hour), LastHeartbeatAt: now.Add(-4 * time.Minute), LastCollectedAt: now.Add(-4 * time.Minute), LoadUsed: 4, LoadCapacity: 16, ControlTunnelStatus: ChannelReady, ControlTunnelDetail: "最后已知:控制隧道正常", ControlLatencyMs: 31, VideoPlaneStatus: ChannelReady, VideoPlaneDetail: "最后已知:4 路视频正常", ActiveStreamCount: 4, BackfillStatus: BackfillPending, BackfillQueueDepth: 12, ProjectionVersion: 1},
{ID: "SEN-EDGE-03", Name: "南门边缘节点", Location: "南门岗亭", RuntimeVersion: "sense-edge 0.1.0", ModelVersion: "detector-v3", StartedAt: now.Add(-10 * time.Hour), LastHeartbeatAt: now.Add(-12 * time.Second), LastCollectedAt: now.Add(-12 * time.Second), LoadUsed: 2, LoadCapacity: 8, ControlTunnelStatus: ChannelReady, ControlTunnelDetail: "控制隧道已恢复", ControlLatencyMs: 46, VideoPlaneStatus: ChannelUnavailable, VideoPlaneDetail: "视频数据面恢复中", ActiveStreamCount: 0, BackfillStatus: BackfillPending, BackfillQueueDepth: 5, RecoveryPhase: "video_reconnecting", ProjectionVersion: 2},
}
return db.Clauses(clause.OnConflict{DoNothing: true}).Create(&items).Error
}
@@ -0,0 +1,61 @@
package edge_node
import (
"time"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
)
const (
StatusOnline = "online"
StatusOffline = "offline"
StatusRecovering = "recovering"
ChannelReady = "ready"
ChannelUnavailable = "unavailable"
BackfillIdle = "idle"
BackfillPending = "pending"
)
// Node is the Sense-owned last-known projection of one edge runtime. It is
// deliberately not a shared identity or a cross-product protocol model.
type Node struct {
ID string `gorm:"size:64;primaryKey" json:"id"`
Name string `gorm:"size:128;not null" json:"name"`
Location string `gorm:"size:256;not null;default:''" json:"location"`
RuntimeVersion string `gorm:"size:64;not null;default:''" json:"runtimeVersion"`
ModelVersion string `gorm:"size:64;not null;default:''" json:"modelVersion"`
StartedAt time.Time `gorm:"not null" json:"startedAt"`
LastHeartbeatAt time.Time `gorm:"not null;index" json:"lastHeartbeatAt"`
LastCollectedAt time.Time `gorm:"not null" json:"lastCollectedAt"`
LoadUsed int `gorm:"not null;default:0" json:"loadUsed"`
LoadCapacity int `gorm:"not null;default:0" json:"loadCapacity"`
ControlTunnelStatus string `gorm:"size:32;not null;default:'unavailable'" json:"controlTunnelStatus"`
ControlTunnelDetail string `gorm:"size:256;not null;default:''" json:"controlTunnelDetail"`
ControlLatencyMs int `gorm:"not null;default:0" json:"controlLatencyMs"`
VideoPlaneStatus string `gorm:"size:32;not null;default:'unavailable'" json:"videoPlaneStatus"`
VideoPlaneDetail string `gorm:"size:256;not null;default:''" json:"videoPlaneDetail"`
ActiveStreamCount int `gorm:"not null;default:0" json:"activeStreamCount"`
BackfillStatus string `gorm:"size:32;not null;default:'idle'" json:"backfillStatus"`
BackfillQueueDepth int `gorm:"not null;default:0" json:"backfillQueueDepth"`
RecoveryPhase string `gorm:"size:32;not null;default:''" json:"recoveryPhase"`
ProjectionVersion int64 `gorm:"not null;default:1" json:"projectionVersion"`
common.ControlBy
common.ModelTime
}
func (Node) TableName() string { return "sense_edge_nodes" }
// Event records projection state transitions without storing credentials or
// raw heartbeat payloads.
type Event struct {
ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"`
NodeID string `gorm:"size:64;not null;index" json:"nodeId"`
EventType string `gorm:"size:32;not null;index" json:"eventType"`
FromStatus string `gorm:"size:32;not null;default:''" json:"fromStatus"`
ToStatus string `gorm:"size:32;not null" json:"toStatus"`
Detail string `gorm:"size:256;not null;default:''" json:"detail"`
OccurredAt time.Time `gorm:"not null;index" json:"occurredAt"`
}
func (Event) TableName() string { return "sense_edge_node_events" }
+167
View File
@@ -0,0 +1,167 @@
package edge_node
import (
"context"
"errors"
"fmt"
"strings"
"time"
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
var (
ErrInvalidQuery = errors.New("边缘节点查询条件不符合要求")
ErrNotFound = errors.New("边缘节点不存在")
ErrInvalidSample = errors.New("边缘节点心跳样本不符合要求")
)
type Service struct {
coreService.Service
Now func() time.Time
}
func NewService(db *gorm.DB) *Service {
return &Service{Service: coreService.Service{Orm: db}, Now: time.Now}
}
func (s *Service) List(request PageRequest) (PageResponse, error) {
if request.GetPageSize() > 100 {
return PageResponse{}, ErrInvalidQuery
}
var count int64
if err := s.Orm.Model(&Node{}).Count(&count).Error; err != nil {
return PageResponse{}, fmt.Errorf("count edge nodes: %w", err)
}
var models []Node
if err := s.Orm.Order("name ASC, id ASC").Limit(request.GetPageSize()).Offset((request.GetPageIndex() - 1) * request.GetPageSize()).Find(&models).Error; err != nil {
return PageResponse{}, fmt.Errorf("list edge nodes: %w", err)
}
result := PageResponse{List: make([]Response, 0, len(models)), Count: count}
result.Summary.HeartbeatTimeout = int64(HeartbeatTimeout / time.Second)
var all []Node
if err := s.Orm.Find(&all).Error; err != nil {
return PageResponse{}, fmt.Errorf("summarize edge nodes: %w", err)
}
result.Summary.Total = len(all)
for _, item := range all {
status := s.status(item)
switch status {
case StatusOffline:
result.Summary.Offline++
case StatusRecovering:
result.Summary.Recovering++
default:
result.Summary.Online++
}
result.Summary.LoadUsed += item.LoadUsed
result.Summary.LoadCapacity += item.LoadCapacity
result.Summary.BackfillQueueDepth += item.BackfillQueueDepth
}
for _, item := range models {
result.List = append(result.List, s.response(item, nil))
}
return result, nil
}
func (s *Service) Get(id string) (Response, error) {
id = strings.TrimSpace(id)
if id == "" || len(id) > 64 {
return Response{}, ErrInvalidQuery
}
var model Node
if err := s.Orm.First(&model, "id = ?", id).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return Response{}, ErrNotFound
}
return Response{}, fmt.Errorf("get edge node: %w", err)
}
var events []Event
if err := s.Orm.Where("node_id = ?", id).Order("occurred_at DESC, id DESC").Limit(20).Find(&events).Error; err != nil {
return Response{}, fmt.Errorf("list edge node events: %w", err)
}
return s.response(model, events), nil
}
// ApplyHeartbeat updates the Sense-owned projection atomically and records
// only state transitions. Callers are internal adapters, not public APIs.
func (s *Service) ApplyHeartbeat(ctx context.Context, sample HeartbeatSample) error {
now := sample.CollectedAt.UTC()
if now.IsZero() {
now = s.now()
}
sample.ID = strings.TrimSpace(sample.ID)
if sample.ID == "" || len(sample.ID) > 64 || strings.TrimSpace(sample.Name) == "" || sample.LastHeartbeatAt.IsZero() || sample.StartedAt.IsZero() || sample.LoadUsed < 0 || sample.LoadCapacity < 0 || sample.LoadUsed > sample.LoadCapacity || sample.BackfillQueueDepth < 0 {
return ErrInvalidSample
}
sample.LastHeartbeatAt = sample.LastHeartbeatAt.UTC()
sample.LastCollectedAt = now
return s.Orm.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var previous Node
err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&previous, "id = ?", sample.ID).Error
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
return fmt.Errorf("lock edge node projection: %w", err)
}
from := ""
if err == nil {
from = statusAt(previous, now)
sample.CreatedAt = previous.CreatedAt
sample.ProjectionVersion = previous.ProjectionVersion + 1
} else {
sample.ProjectionVersion = 1
}
if err = tx.Save(&sample.Node).Error; err != nil {
return fmt.Errorf("save edge node projection: %w", err)
}
to := statusAt(sample.Node, now)
if from == to && from != "" {
return nil
}
if from == StatusOffline {
timeoutAt := previous.LastHeartbeatAt.UTC().Add(HeartbeatTimeout)
if err = tx.Create(&Event{NodeID: sample.ID, EventType: "heartbeat_timeout", FromStatus: StatusOnline, ToStatus: StatusOffline, Detail: "节点心跳超过 90 秒未更新", OccurredAt: timeoutAt}).Error; err != nil {
return err
}
}
eventType, detail := "heartbeat_received", "收到节点首次心跳"
if from == StatusOffline {
eventType, detail = "node_recovered", "节点心跳恢复,通道状态继续独立收敛"
} else if to == StatusRecovering {
eventType, detail = "recovery_started", "节点进入恢复阶段"
}
return tx.Create(&Event{NodeID: sample.ID, EventType: eventType, FromStatus: from, ToStatus: to, Detail: detail, OccurredAt: now}).Error
})
}
func (s *Service) response(model Node, events []Event) Response {
now := s.now()
freshness := int64(now.Sub(model.LastHeartbeatAt.UTC()).Seconds())
if freshness < 0 {
freshness = 0
}
uptime := int64(model.LastHeartbeatAt.UTC().Sub(model.StartedAt.UTC()).Seconds())
if uptime < 0 {
uptime = 0
}
return Response{ID: model.ID, Name: model.Name, Location: model.Location, Status: statusAt(model, now), RuntimeVersion: model.RuntimeVersion, ModelVersion: model.ModelVersion, StartedAt: model.StartedAt, UptimeSeconds: uptime, LastHeartbeatAt: model.LastHeartbeatAt, LastCollectedAt: model.LastCollectedAt, FreshnessSeconds: freshness, Stale: freshness > int64(HeartbeatTimeout/time.Second), LoadUsed: model.LoadUsed, LoadCapacity: model.LoadCapacity, ControlTunnelStatus: model.ControlTunnelStatus, ControlTunnelDetail: model.ControlTunnelDetail, ControlLatencyMs: model.ControlLatencyMs, VideoPlaneStatus: model.VideoPlaneStatus, VideoPlaneDetail: model.VideoPlaneDetail, ActiveStreamCount: model.ActiveStreamCount, BackfillStatus: model.BackfillStatus, BackfillQueueDepth: model.BackfillQueueDepth, RecoveryPhase: model.RecoveryPhase, ProjectionVersion: model.ProjectionVersion, Events: events}
}
func (s *Service) status(model Node) string { return statusAt(model, s.now()) }
func (s *Service) now() time.Time {
if s.Now != nil {
return s.Now().UTC()
}
return time.Now().UTC()
}
func statusAt(model Node, now time.Time) string {
if now.UTC().Sub(model.LastHeartbeatAt.UTC()) > HeartbeatTimeout {
return StatusOffline
}
if strings.TrimSpace(model.RecoveryPhase) != "" && model.RecoveryPhase != "converged" {
return StatusRecovering
}
return StatusOnline
}
@@ -0,0 +1,103 @@
package edge_node
import (
"context"
"errors"
"testing"
"time"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
func edgeNodeTestService(t *testing.T) (*Service, *gorm.DB, time.Time) {
t.Helper()
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&Node{}, &Event{}); err != nil {
t.Fatal(err)
}
now := time.Date(2026, 8, 28, 4, 0, 0, 0, time.UTC)
service := NewService(db)
service.Now = func() time.Time { return now }
return service, db, now
}
func TestListClassifiesOnlineOfflineAndRecoveringWithoutErasingLastKnownState(t *testing.T) {
service, db, now := edgeNodeTestService(t)
if err := SeedSyntheticFixture(db, now); err != nil {
t.Fatal(err)
}
page, err := service.List(PageRequest{})
if err != nil {
t.Fatal(err)
}
if page.Count != 3 || page.Summary.Online != 1 || page.Summary.Offline != 1 || page.Summary.Recovering != 1 || page.Summary.BackfillQueueDepth != 17 {
t.Fatalf("unexpected summary: %+v", page.Summary)
}
var offline Response
for _, item := range page.List {
if item.Status == StatusOffline {
offline = item
}
}
if !offline.Stale || offline.ControlTunnelStatus != ChannelReady || offline.VideoPlaneStatus != ChannelReady || offline.BackfillQueueDepth != 12 {
t.Fatalf("offline projection lost last-known state: %+v", offline)
}
}
func TestApplyHeartbeatRecordsRecoveryAndPreservesIndependentChannelConvergence(t *testing.T) {
service, db, now := edgeNodeTestService(t)
old := Node{ID: "EDGE-1", Name: "旧节点", StartedAt: now.Add(-time.Hour), LastHeartbeatAt: now.Add(-5 * time.Minute), LastCollectedAt: now.Add(-5 * time.Minute), LoadCapacity: 16, ControlTunnelStatus: ChannelReady, VideoPlaneStatus: ChannelReady, BackfillStatus: BackfillPending, BackfillQueueDepth: 8, ProjectionVersion: 3}
if err := db.Create(&old).Error; err != nil {
t.Fatal(err)
}
sample := HeartbeatSample{Node: Node{ID: "EDGE-1", Name: "旧节点", StartedAt: old.StartedAt, LastHeartbeatAt: now.Add(-5 * time.Second), LoadUsed: 4, LoadCapacity: 16, ControlTunnelStatus: ChannelReady, VideoPlaneStatus: ChannelUnavailable, BackfillStatus: BackfillPending, BackfillQueueDepth: 3, RecoveryPhase: "video_reconnecting"}, CollectedAt: now}
if err := service.ApplyHeartbeat(context.Background(), sample); err != nil {
t.Fatal(err)
}
item, err := service.Get("EDGE-1")
if err != nil {
t.Fatal(err)
}
if item.Status != StatusRecovering || item.ProjectionVersion != 4 || item.VideoPlaneStatus != ChannelUnavailable || item.BackfillQueueDepth != 3 {
t.Fatalf("unexpected recovery projection: %+v", item)
}
if len(item.Events) != 2 || item.Events[0].EventType != "node_recovered" || item.Events[0].FromStatus != StatusOffline || item.Events[0].ToStatus != StatusRecovering || item.Events[1].EventType != "heartbeat_timeout" || item.Events[1].ToStatus != StatusOffline {
t.Fatalf("unexpected events: %+v", item.Events)
}
}
func TestApplyHeartbeatRejectsInvalidSampleAndDoesNotPersist(t *testing.T) {
service, db, now := edgeNodeTestService(t)
err := service.ApplyHeartbeat(context.Background(), HeartbeatSample{Node: Node{ID: "EDGE-2", Name: "节点", StartedAt: now.Add(-time.Hour), LastHeartbeatAt: now, LoadUsed: 17, LoadCapacity: 16}, CollectedAt: now})
if !errors.Is(err, ErrInvalidSample) {
t.Fatalf("err=%v", err)
}
var count int64
db.Model(&Node{}).Count(&count)
if count != 0 {
t.Fatalf("count=%d", count)
}
}
func TestGetRejectsMissingOrMalformedID(t *testing.T) {
service, _, _ := edgeNodeTestService(t)
if _, err := service.Get(""); !errors.Is(err, ErrInvalidQuery) {
t.Fatalf("err=%v", err)
}
if _, err := service.Get("missing"); !errors.Is(err, ErrNotFound) {
t.Fatalf("err=%v", err)
}
}
func TestListRejectsOversizedPage(t *testing.T) {
service, _, _ := edgeNodeTestService(t)
request := PageRequest{}
request.PageSize = 101
if _, err := service.List(request); !errors.Is(err, ErrInvalidQuery) {
t.Fatalf("err=%v", err)
}
}
@@ -0,0 +1,29 @@
package local_event
import (
"context"
"encoding/json"
"fmt"
"time"
"gorm.io/gorm"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
)
// CreateWithOutbox commits the local candidate and its internal delivery
// record atomically. The payload remains Sense-internal and is not a Bell or
// Brain contract.
func CreateWithOutbox(ctx context.Context, db *gorm.DB, candidate EventCandidate, payload map[string]interface{}, now time.Time) error {
encoded, err := json.Marshal(payload)
if err != nil {
return fmt.Errorf("encode local event outbox payload: %w", err)
}
return db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Create(&candidate).Error; err != nil {
return fmt.Errorf("create local event candidate: %w", err)
}
_, err = outbox.Enqueue(tx, outbox.EnqueueInput{InternalType: "local_event_candidate", BusinessRef: candidate.ID, IdempotencyKey: "local-event:" + candidate.ID + ":v1", PayloadJSON: encoded}, now)
return err
})
}
@@ -0,0 +1,46 @@
package local_event
import (
"context"
"testing"
"time"
"github.com/google/uuid"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
)
func TestCreateWithOutboxCommitsAndRollsBackAtomically(t *testing.T) {
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
sqlDB, _ := db.DB()
sqlDB.SetMaxOpenConns(1)
if err = db.AutoMigrate(&EventCandidate{}, &outbox.Message{}, &outbox.DeliveryRecord{}, &outbox.Attempt{}); err != nil {
t.Fatal(err)
}
now := time.Date(2026, 8, 28, 10, 0, 0, 0, time.UTC)
candidate := EventCandidate{ID: uuid.NewString(), OccurredAt: now, SourceRef: "SEN-CAM-01", RuleRef: "rule-1", RuleName: "区域闯入", CandidateState: CandidateStateCandidate, EvidenceState: EvidenceStatePending, RetainUntil: now.Add(24 * time.Hour)}
if err = CreateWithOutbox(context.Background(), db, candidate, map[string]interface{}{"eventId": candidate.ID}, now); err != nil {
t.Fatal(err)
}
var candidates, messages int64
db.Model(&EventCandidate{}).Count(&candidates)
db.Model(&outbox.Message{}).Count(&messages)
if candidates != 1 || messages != 1 {
t.Fatalf("candidates=%d messages=%d", candidates, messages)
}
duplicate := candidate
duplicate.ID = candidate.ID
if err = CreateWithOutbox(context.Background(), db, duplicate, map[string]interface{}{"eventId": duplicate.ID}, now); err == nil {
t.Fatal("expected duplicate transaction failure")
}
db.Model(&EventCandidate{}).Count(&candidates)
db.Model(&outbox.Message{}).Count(&messages)
if candidates != 1 || messages != 1 {
t.Fatalf("atomic rollback failed candidates=%d messages=%d", candidates, messages)
}
}
+23
View File
@@ -9,6 +9,8 @@ import (
"time"
"gorm.io/gorm"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
)
var runtimeState struct {
@@ -41,6 +43,27 @@ func StartRuntime(parent context.Context, db *gorm.DB) error {
return err
}
}
shards := media_shard.NewService(db, func(probeCtx context.Context, endpoint string) error {
probe, probeErr := NewHTTPController(endpoint)
if probeErr != nil {
return probeErr
}
return probe.Health(probeCtx)
})
specs, err := media_shard.SpecsFromEnvironment(db, mode, config.APIBase)
if err != nil {
cancel()
return err
}
if err = shards.SyncSpecs(ctx, specs); err != nil {
cancel()
return err
}
if err = shards.RefreshAll(ctx); err != nil {
cancel()
return err
}
service.WithShardService(shards)
runtimeState.Lock()
if runtimeState.cancel != nil {
runtimeState.cancel()
+64 -7
View File
@@ -13,6 +13,7 @@ import (
"gorm.io/gorm/clause"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/credential"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/reconcile"
)
@@ -26,10 +27,16 @@ type Service struct {
controller Controller
process *Supervisor
config RuntimeConfig
shards *media_shard.Service
now func() time.Time
mu sync.Mutex
}
func (s *Service) WithShardService(shards *media_shard.Service) *Service {
s.shards = shards
return s
}
func NewService(db *gorm.DB, controller Controller, process *Supervisor, config RuntimeConfig) *Service {
return &Service{db: db, controller: controller, process: process, config: config, now: time.Now}
}
@@ -76,11 +83,19 @@ func (s *Service) ensureDevice(ctx context.Context, deviceID string, reactivate
return err
}
}
if s.shards != nil {
if _, err := s.shards.EnsureAssignment(ctx, id); err != nil {
return fmt.Errorf("assign media route to shard: %w", err)
}
}
}
return nil
}
func (s *Service) Run(ctx context.Context) {
if s.shards != nil {
_ = s.shards.RefreshAll(ctx)
}
_ = s.EnsureAllVerified(ctx)
_ = s.ReconcileDue(ctx)
interval := s.config.PollInterval
@@ -94,6 +109,9 @@ func (s *Service) Run(ctx context.Context) {
case <-ctx.Done():
return
case <-ticker.C:
if s.shards != nil {
_ = s.shards.RefreshAll(ctx)
}
_ = s.ReconcileDue(ctx)
}
}
@@ -127,10 +145,11 @@ func (s *Service) Refresh(ctx context.Context, id string) (RouteResponse, error)
if err := s.db.WithContext(ctx).First(&route, "id = ?", id).Error; err != nil {
return RouteResponse{}, ErrNotFound
}
if err := s.ensureControl(ctx); err != nil {
controller, err := s.controllerForRoute(ctx, route)
if err != nil {
return s.saveFailure(ctx, route, "process_unavailable", "MediaMTX 未启动或 Control API 未就绪", err)
}
status, err := s.controller.Status(ctx, route.Path)
status, err := controller.Status(ctx, route.Path)
if err != nil {
return s.saveFailure(ctx, route, "status_unavailable", "尚未取得媒体路径状态", err)
}
@@ -159,12 +178,13 @@ func (s *Service) Reconcile(ctx context.Context, id string) (RouteResponse, erro
return RouteResponse{}, err
}
if route.Desired == DesiredStopped {
if s.controller != nil {
_ = s.controller.Delete(ctx, route.Path)
if controller, err := s.controllerForRoute(ctx, route); err == nil {
_ = controller.Delete(ctx, route.Path)
}
return s.saveSuccess(ctx, route, "stopped", false, 0, "媒体路径已停止")
}
if err := s.ensureControl(ctx); err != nil {
controller, err := s.controllerForRoute(ctx, route)
if err != nil {
return s.saveFailure(ctx, route, "process_unavailable", "MediaMTX 未启动或 Control API 未就绪", err)
}
var profile admissionProfile
@@ -175,10 +195,10 @@ func (s *Service) Reconcile(ctx context.Context, id string) (RouteResponse, erro
if err != nil {
return s.saveFailure(ctx, route, "credential_unavailable", "RTSP 凭据不可用", err)
}
if err = s.controller.Apply(ctx, Source{Path: route.Path, URI: profile.StreamURI, Username: value.Username, Password: value.Password}); err != nil {
if err = controller.Apply(ctx, Source{Path: route.Path, URI: profile.StreamURI, Username: value.Username, Password: value.Password}); err != nil {
return s.saveFailure(ctx, route, "apply_failed", "媒体路径配置失败", err)
}
status, err := s.controller.Status(ctx, route.Path)
status, err := controller.Status(ctx, route.Path)
if err != nil {
return s.saveFailure(ctx, route, "status_unavailable", "尚未取得媒体路径状态", err)
}
@@ -191,6 +211,43 @@ func (s *Service) Reconcile(ctx context.Context, id string) (RouteResponse, erro
return s.saveSuccess(ctx, route, "waiting", false, status.Readers, "等待播放器连接并按需拉流")
}
func (s *Service) controllerForRoute(ctx context.Context, route Route) (Controller, error) {
if s.shards == nil {
if err := s.ensureControl(ctx); err != nil {
return nil, err
}
return s.controller, nil
}
shard, err := s.shards.ResolveRoute(ctx, route.ID)
if errors.Is(err, media_shard.ErrNotFound) {
shard, err = s.shards.EnsureAssignment(ctx, route.ID)
}
if err != nil {
return nil, err
}
if shard.Status != media_shard.StatusRunning {
return nil, ErrRuntimeUnavailable
}
if shard.ID == "primary" {
if err = s.ensureControl(ctx); err != nil {
return nil, err
}
return s.controller, nil
}
endpoint, err := s.shards.ControlAPI(ctx, shard.ID)
if err != nil {
return nil, err
}
controller, err := NewHTTPController(endpoint)
if err != nil {
return nil, err
}
if err = controller.Health(ctx); err != nil {
return nil, err
}
return controller, nil
}
func (s *Service) ensureControl(ctx context.Context) error {
if s.controller == nil {
return ErrRuntimeUnavailable
@@ -0,0 +1,91 @@
package media_shard
import (
"errors"
"net/http"
"time"
"github.com/gin-gonic/gin"
"github.com/go-admin-team/go-admin-core/sdk/api"
"github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth/user"
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
"git.ilapage.cn/ila/yovision/Sense/server/common"
)
const auditSuccess, auditFailure = "1", "2"
type API struct{ api.Api }
func (e *API) service(c *gin.Context) (*Service, error) {
base := coreService.Service{}
if err := e.MakeContext(c).MakeOrm().MakeService(&base).Errors; err != nil {
return nil, err
}
return NewService(base.Orm, nil), nil
}
func (e *API) List(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
result, err := service.List(c.Request.Context())
if err != nil {
e.audit(c, service, "List", auditFailure, "媒体分片列表查询失败")
e.writeError(err)
return
}
e.audit(c, service, "List", auditSuccess, "读取媒体分片列表")
e.OK(result, "查询成功")
}
func (e *API) Get(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
result, err := service.Get(c.Request.Context(), c.Param("id"))
if err != nil {
e.audit(c, service, "Get", auditFailure, "媒体分片详情查询失败")
e.writeError(err)
return
}
e.audit(c, service, "Get", auditSuccess, "读取媒体分片详情 "+result.ID)
e.OK(result, "查询成功")
}
func (e *API) Preflight(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
result, err := service.Preflight(c.Request.Context(), c.Param("id"))
if err != nil {
e.audit(c, service, "Preflight", auditFailure, "媒体分片迁移预检失败")
e.writeError(err)
return
}
e.audit(c, service, "Preflight", auditSuccess, "执行媒体分片只读迁移预检 "+result.SourceShardID)
e.OK(result, "预检完成")
}
func (e *API) audit(c *gin.Context, service *Service, action, status, remark string) {
if err := WriteAudit(service.DB, Audit{Action: action, Method: c.Request.Method, Status: status, Username: user.GetUserName(c), UserID: user.GetUserId(c), ClientIP: common.GetClientIP(c), Route: c.FullPath(), Remark: remark, At: time.Now()}); err != nil {
api.GetRequestLogger(c).Errorf("media shard audit failed: %s", err.Error())
}
}
func (e *API) writeError(err error) {
switch {
case errors.Is(err, ErrInvalidID), errors.Is(err, ErrInvalidConfig):
e.Error(http.StatusBadRequest, err, err.Error())
case errors.Is(err, ErrNotFound):
e.Error(http.StatusNotFound, err, err.Error())
default:
e.Error(http.StatusInternalServerError, err, "媒体分片查询失败")
}
}
@@ -0,0 +1,19 @@
package media_shard
import (
"time"
adminModels "git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
"gorm.io/gorm"
)
type Audit struct {
Action, Method, Status, Username, ClientIP, Route, Remark string
UserID int
At time.Time
}
func WriteAudit(db *gorm.DB, input Audit) error {
model := adminModels.SysOperaLog{Title: "媒体分片", BusinessType: "other", Method: "media_shard.API." + input.Action, RequestMethod: input.Method, OperatorType: "1", OperName: input.Username, OperUrl: input.Route, OperIp: input.ClientIP, Status: input.Status, OperTime: input.At.UTC(), Remark: input.Remark, CreatedAt: input.At.UTC(), UpdatedAt: input.At.UTC()}
return db.Create(&model).Error
}
@@ -0,0 +1,80 @@
package media_shard
import (
"encoding/json"
"errors"
"fmt"
"net"
"net/url"
"os"
"strconv"
"strings"
"gorm.io/gorm"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/quota"
)
var ErrInvalidConfig = errors.New("MediaMTX 分片配置不符合要求")
func SpecsFromEnvironment(db *gorm.DB, mode, primaryAPI string) ([]Spec, error) {
capacity, err := quota.ReadLimit(db)
if raw := strings.TrimSpace(os.Getenv("SENSE_MEDIAMTX_CAPACITY")); raw != "" {
capacity, err = strconv.Atoi(raw)
}
if err != nil || capacity < 1 {
return nil, fmt.Errorf("%w: primary capacity", ErrInvalidConfig)
}
primary := Spec{ID: "primary", Name: "主媒体分片", Mode: mode, ControlAPI: strings.TrimSpace(primaryAPI), Capacity: capacity}
if err = validateSpec(primary, true); err != nil {
return nil, err
}
result := []Spec{primary}
path := strings.TrimSpace(os.Getenv("SENSE_MEDIAMTX_SHARDS_FILE"))
if path == "" {
return result, nil
}
data, err := os.ReadFile(path)
if err != nil {
return nil, fmt.Errorf("read MediaMTX shards file: %w", err)
}
var extra []Spec
if err = json.Unmarshal(data, &extra); err != nil {
return nil, fmt.Errorf("%w: shards JSON", ErrInvalidConfig)
}
seen := map[string]bool{"primary": true}
for _, item := range extra {
item.ID, item.Name, item.Mode, item.ControlAPI = strings.TrimSpace(item.ID), strings.TrimSpace(item.Name), strings.ToLower(strings.TrimSpace(item.Mode)), strings.TrimSpace(item.ControlAPI)
if item.Mode == "" {
item.Mode = "external"
}
if seen[item.ID] || item.Mode != "external" {
return nil, fmt.Errorf("%w: duplicate id or non-external extra shard", ErrInvalidConfig)
}
if err = validateSpec(item, false); err != nil {
return nil, err
}
seen[item.ID] = true
result = append(result, item)
}
return result, nil
}
func validateSpec(spec Spec, primary bool) error {
if spec.ID == "" || len(spec.ID) > 64 || spec.Name == "" || len([]rune(spec.Name)) > 128 || spec.Capacity < 1 || spec.Capacity > 100000 {
return fmt.Errorf("%w: shard identity or capacity", ErrInvalidConfig)
}
if primary && spec.Mode != "managed" && spec.Mode != "external" {
return fmt.Errorf("%w: primary mode", ErrInvalidConfig)
}
parsed, err := url.Parse(spec.ControlAPI)
if err != nil || parsed.Scheme != "http" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" || parsed.Path != "" {
return fmt.Errorf("%w: control API", ErrInvalidConfig)
}
host := parsed.Hostname()
ip := net.ParseIP(host)
if !strings.EqualFold(host, "localhost") && (ip == nil || !ip.IsLoopback()) {
return fmt.Errorf("%w: control API must be loopback", ErrInvalidConfig)
}
return nil
}
@@ -0,0 +1,46 @@
package media_shard
import (
"os"
"path/filepath"
"testing"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/quota"
)
func TestSpecsFromEnvironmentUsesDatabaseQuotaAndExternalFile(t *testing.T) {
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&quota.Setting{}); err != nil {
t.Fatal(err)
}
if err = db.Create(&quota.Setting{ID: quota.SettingID, Limit: 23, Source: "database", Version: 1}).Error; err != nil {
t.Fatal(err)
}
path := filepath.Join(t.TempDir(), "shards.json")
if err = os.WriteFile(path, []byte(`[{"id":"secondary","name":"备用分片","controlAPI":"http://127.0.0.1:19997","capacity":11}]`), 0600); err != nil {
t.Fatal(err)
}
t.Setenv("SENSE_MEDIAMTX_CAPACITY", "")
t.Setenv("SENSE_MEDIAMTX_SHARDS_FILE", path)
specs, err := SpecsFromEnvironment(db, "managed", "http://127.0.0.1:9997")
if err != nil {
t.Fatal(err)
}
if len(specs) != 2 || specs[0].Capacity != 23 || specs[1].Capacity != 11 || specs[1].Mode != "external" {
t.Fatalf("unexpected specs: %+v", specs)
}
}
func TestSpecsRejectsNonLoopbackControlAPI(t *testing.T) {
t.Setenv("SENSE_MEDIAMTX_CAPACITY", "5")
t.Setenv("SENSE_MEDIAMTX_SHARDS_FILE", "")
if _, err := SpecsFromEnvironment(nil, "external", "http://192.0.2.1:9997"); err == nil {
t.Fatal("expected unsafe control API to be rejected")
}
}
+72
View File
@@ -0,0 +1,72 @@
package media_shard
import "time"
type Spec struct {
ID string `json:"id"`
Name string `json:"name"`
Mode string `json:"mode"`
ControlAPI string `json:"controlAPI"`
Capacity int `json:"capacity"`
}
type Summary struct {
Total int `json:"total"`
Available int `json:"available"`
Failed int `json:"failed"`
Configured int `json:"configuredCapacity"`
Assigned int `json:"assignedPaths"`
Impacted int `json:"impactedPaths"`
}
type Response struct {
ID string `json:"id"`
Name string `json:"name"`
Mode string `json:"mode"`
Status string `json:"status"`
Detail string `json:"detail"`
Capacity int `json:"capacity"`
AssignedPaths int `json:"assignedPaths"`
Remaining int `json:"remaining"`
LastProbeAt *time.Time `json:"lastProbeAt,omitempty"`
Stale bool `json:"stale"`
}
type Impact struct {
RouteID string `json:"routeId"`
DeviceID string `json:"deviceId"`
DeviceName string `json:"deviceName"`
Location string `json:"location"`
ProfileToken string `json:"profileToken"`
Path string `json:"path"`
Desired string `json:"desired"`
Actual string `json:"actual"`
}
type DetailResponse struct {
Response
Impact []Impact `json:"impact"`
}
type PageResponse struct {
Summary Summary `json:"summary"`
List []Response `json:"list"`
Count int64 `json:"count"`
}
type Check struct {
Name string `json:"name"`
Passed bool `json:"passed"`
Detail string `json:"detail"`
}
type PreflightResponse struct {
SourceShardID string `json:"sourceShardId"`
TargetShardID string `json:"targetShardId,omitempty"`
TargetShardName string `json:"targetShardName,omitempty"`
ImpactedPaths int `json:"impactedPaths"`
Ready bool `json:"ready"`
ExecutionAuthorized bool `json:"executionAuthorized"`
RequiresIssue bool `json:"requiresSeparateHighRiskIssue"`
Checks []Check `json:"checks"`
}
@@ -0,0 +1,59 @@
package media_shard
import "time"
const (
StatusUnknown = "unknown"
StatusRunning = "running"
StatusFailed = "failed"
StatusDisabled = "disabled"
)
// Shard stores only a loopback control endpoint. It is deliberately omitted
// from API response DTOs so credentials and control-plane topology cannot leak.
type Shard struct {
ID string `gorm:"size:64;primaryKey"`
Name string `gorm:"size:128;not null"`
Mode string `gorm:"size:16;not null"`
ControlAPI string `gorm:"size:512;not null"`
Capacity int `gorm:"not null"`
Status string `gorm:"size:16;not null;index"`
Detail string `gorm:"size:256;not null"`
LastProbeAt *time.Time `gorm:"index"`
ConfigVersion int64 `gorm:"not null;default:1"`
CreatedAt time.Time
UpdatedAt time.Time
}
func (Shard) TableName() string { return "sense_media_shards" }
// Assignment is the stable, Sense-owned mapping between an existing media
// route and a MediaMTX shard. Failures never rewrite this record automatically.
type Assignment struct {
RouteID string `gorm:"size:96;primaryKey"`
ShardID string `gorm:"size:64;not null;index"`
Algorithm string `gorm:"size:32;not null"`
CreatedAt time.Time `gorm:"not null"`
UpdatedAt time.Time `gorm:"not null"`
}
func (Assignment) TableName() string { return "sense_media_shard_assignments" }
type routeProjection struct {
ID string `gorm:"column:id"`
DeviceID string `gorm:"column:device_id"`
ProfileToken string `gorm:"column:profile_token"`
Path string `gorm:"column:path"`
Desired string `gorm:"column:desired"`
Actual string `gorm:"column:actual"`
}
func (routeProjection) TableName() string { return "sense_media_routes" }
type deviceProjection struct {
ID string `gorm:"column:id"`
Name string `gorm:"column:name"`
Location string `gorm:"column:location"`
}
func (deviceProjection) TableName() string { return "sense_devices" }
@@ -0,0 +1,259 @@
package media_shard
import (
"context"
"crypto/sha256"
"encoding/binary"
"errors"
"fmt"
"sort"
"strings"
"time"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
var (
ErrNotFound = errors.New("媒体分片不存在")
ErrNoCapacity = errors.New("没有可用的媒体分片容量")
ErrInvalidID = errors.New("媒体分片标识不符合要求")
)
const staleAfter = 30 * time.Second
type Probe func(context.Context, string) error
type Service struct {
DB *gorm.DB
Probe Probe
Now func() time.Time
}
func NewService(db *gorm.DB, probe Probe) *Service {
return &Service{DB: db, Probe: probe, Now: time.Now}
}
func (s *Service) SyncSpecs(ctx context.Context, specs []Spec) error {
return s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
for _, spec := range specs {
var current Shard
err := tx.First(&current, "id = ?", spec.ID).Error
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
if errors.Is(err, gorm.ErrRecordNotFound) {
current = Shard{ID: spec.ID, Status: StatusUnknown, Detail: "等待健康探测", ConfigVersion: 1}
} else if current.Name != spec.Name || current.Mode != spec.Mode || current.ControlAPI != spec.ControlAPI || current.Capacity != spec.Capacity {
current.ConfigVersion++
}
current.Name, current.Mode, current.ControlAPI, current.Capacity = spec.Name, spec.Mode, spec.ControlAPI, spec.Capacity
if err = tx.Save(&current).Error; err != nil {
return err
}
}
ids := make([]string, 0, len(specs))
for _, spec := range specs {
ids = append(ids, spec.ID)
}
return tx.Model(&Shard{}).Where("id NOT IN ?", ids).Updates(map[string]any{"status": StatusDisabled, "detail": "分片已从当前运行配置移除", "updated_at": s.now()}).Error
})
}
func (s *Service) RefreshAll(ctx context.Context) error {
var shards []Shard
if err := s.DB.WithContext(ctx).Where("status <> ?", StatusDisabled).Find(&shards).Error; err != nil {
return err
}
var result error
for _, shard := range shards {
status, detail := StatusRunning, "Control API 正常"
if s.Probe == nil || s.Probe(ctx, shard.ControlAPI) != nil {
status, detail = StatusFailed, "Control API 不可用"
}
now := s.now()
if err := s.DB.WithContext(ctx).Model(&Shard{}).Where("id = ?", shard.ID).Updates(map[string]any{"status": status, "detail": detail, "last_probe_at": &now, "updated_at": now}).Error; err != nil {
result = errors.Join(result, err)
}
}
return result
}
func (s *Service) EnsureAssignment(ctx context.Context, routeID string) (Shard, error) {
routeID = strings.TrimSpace(routeID)
if routeID == "" {
return Shard{}, ErrInvalidID
}
var selected Shard
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var existing Assignment
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&existing, "route_id = ?", routeID).Error; err == nil {
return tx.First(&selected, "id = ?", existing.ShardID).Error
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
var candidates []Shard
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("status IN ?", []string{StatusRunning, StatusUnknown}).Find(&candidates).Error; err != nil {
return err
}
counts, err := assignmentCounts(tx)
if err != nil {
return err
}
eligible := candidates[:0]
for _, item := range candidates {
if counts[item.ID] < int64(item.Capacity) {
eligible = append(eligible, item)
}
}
if len(eligible) == 0 {
return ErrNoCapacity
}
sort.SliceStable(eligible, func(i, j int) bool { return rendezvous(routeID, eligible[i].ID) > rendezvous(routeID, eligible[j].ID) })
selected = eligible[0]
now := s.now()
assignment := Assignment{RouteID: routeID, ShardID: selected.ID, Algorithm: "rendezvous-v1", CreatedAt: now, UpdatedAt: now}
if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&assignment).Error; err != nil {
return err
}
if assignment.RouteID != "" { // reload also handles a concurrent winner on PostgreSQL.
if err := tx.First(&existing, "route_id = ?", routeID).Error; err != nil {
return err
}
return tx.First(&selected, "id = ?", existing.ShardID).Error
}
return nil
})
return selected, err
}
func (s *Service) ResolveRoute(ctx context.Context, routeID string) (Shard, error) {
var shard Shard
err := s.DB.WithContext(ctx).Table("sense_media_shards AS s").Select("s.*").Joins("JOIN sense_media_shard_assignments a ON a.shard_id = s.id").Where("a.route_id = ?", routeID).Scan(&shard).Error
if err != nil {
return Shard{}, err
}
if shard.ID == "" {
return Shard{}, ErrNotFound
}
return shard, nil
}
func (s *Service) List(ctx context.Context) (PageResponse, error) {
var shards []Shard
if err := s.DB.WithContext(ctx).Order("name ASC, id ASC").Find(&shards).Error; err != nil {
return PageResponse{}, err
}
counts, err := assignmentCounts(s.DB.WithContext(ctx))
if err != nil {
return PageResponse{}, err
}
result := PageResponse{List: make([]Response, 0, len(shards)), Count: int64(len(shards))}
for _, shard := range shards {
item := s.response(shard, int(counts[shard.ID]))
result.List = append(result.List, item)
result.Summary.Total++
result.Summary.Configured += item.Capacity
result.Summary.Assigned += item.AssignedPaths
if item.Status == StatusRunning {
result.Summary.Available++
}
if item.Status == StatusFailed {
result.Summary.Failed++
result.Summary.Impacted += item.AssignedPaths
}
}
return result, nil
}
func (s *Service) Get(ctx context.Context, id string) (DetailResponse, error) {
id = strings.TrimSpace(id)
if id == "" || len(id) > 64 {
return DetailResponse{}, ErrInvalidID
}
var shard Shard
if err := s.DB.WithContext(ctx).First(&shard, "id = ?", id).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return DetailResponse{}, ErrNotFound
}
return DetailResponse{}, err
}
impact, err := s.impact(ctx, id)
if err != nil {
return DetailResponse{}, err
}
return DetailResponse{Response: s.response(shard, len(impact)), Impact: impact}, nil
}
func (s *Service) Preflight(ctx context.Context, id string) (PreflightResponse, error) {
detail, err := s.Get(ctx, id)
if err != nil {
return PreflightResponse{}, err
}
result := PreflightResponse{SourceShardID: id, ImpactedPaths: len(detail.Impact), ExecutionAuthorized: false, RequiresIssue: true}
result.Checks = append(result.Checks, Check{Name: "源分片状态", Passed: detail.Status == StatusRunning && !detail.Stale, Detail: detail.Detail})
var candidates []Shard
if err = s.DB.WithContext(ctx).Where("id <> ? AND status = ?", id, StatusRunning).Find(&candidates).Error; err != nil {
return PreflightResponse{}, err
}
counts, err := assignmentCounts(s.DB.WithContext(ctx))
if err != nil {
return PreflightResponse{}, err
}
for _, item := range candidates {
if item.Capacity-int(counts[item.ID]) >= len(detail.Impact) && (result.TargetShardID == "" || rendezvous(id, item.ID) > rendezvous(id, result.TargetShardID)) {
result.TargetShardID, result.TargetShardName = item.ID, item.Name
}
}
result.Checks = append(result.Checks, Check{Name: "目标容量", Passed: result.TargetShardID != "", Detail: map[bool]string{true: "存在可容纳全部受影响路径的健康目标分片", false: "没有可容纳全部受影响路径的健康目标分片"}[result.TargetShardID != ""]})
result.Checks = append(result.Checks, Check{Name: "实施授权", Passed: false, Detail: "本工单只提供只读预检;实际迁移必须另建高风险工单并人工确认"})
result.Ready = result.Checks[0].Passed && result.Checks[1].Passed
return result, nil
}
func (s *Service) impact(ctx context.Context, id string) ([]Impact, error) {
var rows []Impact
err := s.DB.WithContext(ctx).Table("sense_media_shard_assignments AS a").Select("r.id AS route_id, r.device_id, COALESCE(d.name, '') AS device_name, COALESCE(d.location, '') AS location, r.profile_token, r.path, r.desired, r.actual").Joins("JOIN sense_media_routes r ON r.id = a.route_id").Joins("LEFT JOIN sense_devices d ON d.id = r.device_id").Where("a.shard_id = ?", id).Order("d.name ASC, r.profile_token ASC").Scan(&rows).Error
return rows, err
}
func (s *Service) response(shard Shard, assigned int) Response {
remaining := shard.Capacity - assigned
if remaining < 0 {
remaining = 0
}
stale := shard.LastProbeAt == nil || s.now().Sub(shard.LastProbeAt.UTC()) > staleAfter
return Response{ID: shard.ID, Name: shard.Name, Mode: shard.Mode, Status: shard.Status, Detail: shard.Detail, Capacity: shard.Capacity, AssignedPaths: assigned, Remaining: remaining, LastProbeAt: shard.LastProbeAt, Stale: stale}
}
func assignmentCounts(db *gorm.DB) (map[string]int64, error) {
type row struct {
ShardID string
Count int64
}
var rows []row
err := db.Model(&Assignment{}).Select("shard_id, COUNT(*) AS count").Group("shard_id").Scan(&rows).Error
out := map[string]int64{}
for _, r := range rows {
out[r.ShardID] = r.Count
}
return out, err
}
func rendezvous(routeID, shardID string) uint64 {
sum := sha256.Sum256([]byte(routeID + "\x00" + shardID))
return binary.BigEndian.Uint64(sum[:8])
}
func (s *Service) now() time.Time {
if s.Now != nil {
return s.Now().UTC()
}
return time.Now().UTC()
}
func (s *Service) ControlAPI(ctx context.Context, id string) (string, error) {
var shard Shard
if err := s.DB.WithContext(ctx).First(&shard, "id = ?", id).Error; err != nil {
return "", fmt.Errorf("get media shard control endpoint: %w", err)
}
return shard.ControlAPI, nil
}
@@ -0,0 +1,130 @@
package media_shard
import (
"context"
"errors"
"testing"
"time"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
type testRoute struct{ ID, DeviceID, ProfileToken, Path, Desired, Actual string }
func (testRoute) TableName() string { return "sense_media_routes" }
type testDevice struct{ ID, Name, Location string }
func (testDevice) TableName() string { return "sense_devices" }
func testService(t *testing.T) (*Service, *gorm.DB) {
t.Helper()
db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&Shard{}, &Assignment{}, &testRoute{}, &testDevice{}); err != nil {
t.Fatal(err)
}
now := time.Date(2026, 8, 28, 8, 0, 0, 0, time.UTC)
service := NewService(db, func(context.Context, string) error { return nil })
service.Now = func() time.Time { return now }
return service, db
}
func TestStableAssignmentUsesConfiguredCapacity(t *testing.T) {
service, _ := testService(t)
ctx := context.Background()
if err := service.SyncSpecs(ctx, []Spec{{ID: "alpha", Name: "A", Mode: "external", ControlAPI: "http://127.0.0.1:9997", Capacity: 3}, {ID: "beta", Name: "B", Mode: "external", ControlAPI: "http://127.0.0.1:19997", Capacity: 2}}); err != nil {
t.Fatal(err)
}
if err := service.RefreshAll(ctx); err != nil {
t.Fatal(err)
}
first, err := service.EnsureAssignment(ctx, "route-1")
if err != nil {
t.Fatal(err)
}
second, err := service.EnsureAssignment(ctx, "route-1")
if err != nil {
t.Fatal(err)
}
if first.ID != second.ID {
t.Fatalf("assignment changed from %s to %s", first.ID, second.ID)
}
for _, id := range []string{"route-2", "route-3", "route-4", "route-5"} {
if _, err = service.EnsureAssignment(ctx, id); err != nil {
t.Fatalf("assign %s: %v", id, err)
}
}
if _, err = service.EnsureAssignment(ctx, "route-6"); !errors.Is(err, ErrNoCapacity) {
t.Fatalf("expected configured capacity error, got %v", err)
}
}
func TestFailureKeepsOwnershipAndLocatesImpact(t *testing.T) {
service, db := testService(t)
ctx := context.Background()
if err := service.SyncSpecs(ctx, []Spec{{ID: "failed", Name: "故障分片", Mode: "external", ControlAPI: "http://127.0.0.1:9997", Capacity: 7}}); err != nil {
t.Fatal(err)
}
if err := service.RefreshAll(ctx); err != nil {
t.Fatal(err)
}
if err := db.Create(&testDevice{ID: "device-1", Name: "东门摄像机", Location: "东门"}).Error; err != nil {
t.Fatal(err)
}
if err := db.Create(&testRoute{ID: "device-1:main", DeviceID: "device-1", ProfileToken: "main", Path: "sense_abc", Desired: "running", Actual: "ready"}).Error; err != nil {
t.Fatal(err)
}
if _, err := service.EnsureAssignment(ctx, "device-1:main"); err != nil {
t.Fatal(err)
}
service.Probe = func(context.Context, string) error { return errors.New("offline") }
if err := service.RefreshAll(ctx); err != nil {
t.Fatal(err)
}
detail, err := service.Get(ctx, "failed")
if err != nil {
t.Fatal(err)
}
if detail.Status != StatusFailed || len(detail.Impact) != 1 || detail.Impact[0].DeviceName != "东门摄像机" || detail.Impact[0].ProfileToken != "main" {
t.Fatalf("unexpected detail: %+v", detail)
}
resolved, err := service.ResolveRoute(ctx, "device-1:main")
if err != nil || resolved.ID != "failed" {
t.Fatalf("failure rewrote ownership: %+v %v", resolved, err)
}
}
func TestMigrationPreflightIsReadOnly(t *testing.T) {
service, db := testService(t)
ctx := context.Background()
if err := service.SyncSpecs(ctx, []Spec{{ID: "source", Name: "源", Mode: "external", ControlAPI: "http://127.0.0.1:9997", Capacity: 1}, {ID: "target", Name: "目标", Mode: "external", ControlAPI: "http://127.0.0.1:19997", Capacity: 4}}); err != nil {
t.Fatal(err)
}
if err := service.RefreshAll(ctx); err != nil {
t.Fatal(err)
}
if err := db.Create(&testRoute{ID: "route", DeviceID: "d", ProfileToken: "p", Path: "path", Desired: "running", Actual: "ready"}).Error; err != nil {
t.Fatal(err)
}
if err := db.Create(&Assignment{RouteID: "route", ShardID: "source", Algorithm: "rendezvous-v1", CreatedAt: service.now(), UpdatedAt: service.now()}).Error; err != nil {
t.Fatal(err)
}
before := Assignment{}
db.First(&before, "route_id = ?", "route")
result, err := service.Preflight(ctx, "source")
if err != nil {
t.Fatal(err)
}
after := Assignment{}
db.First(&after, "route_id = ?", "route")
if !result.Ready || result.ExecutionAuthorized || !result.RequiresIssue || result.TargetShardID != "target" {
t.Fatalf("unexpected preflight: %+v", result)
}
if before.ShardID != after.ShardID || !before.UpdatedAt.Equal(after.UpdatedAt) {
t.Fatalf("preflight mutated assignment: before=%+v after=%+v", before, after)
}
}
+121
View File
@@ -0,0 +1,121 @@
package ops_alert
import (
"errors"
"git.ilapage.cn/ila/yovision/Sense/server/common"
"github.com/gin-gonic/gin"
"github.com/gin-gonic/gin/binding"
"github.com/go-admin-team/go-admin-core/sdk/api"
"github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth/user"
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
"net/http"
"time"
)
const auditSuccess, auditFailure = "1", "2"
type API struct{ api.Api }
func (e *API) service(c *gin.Context) (*Service, error) {
base := coreService.Service{}
if err := e.MakeContext(c).MakeOrm().MakeService(&base).Errors; err != nil {
return nil, err
}
return NewService(base.Orm), nil
}
func (e *API) List(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
request := PageRequest{}
if err = e.MakeContext(c).Bind(&request).Errors; err != nil {
e.audit(c, service, "List", auditFailure, "告警查询条件格式不正确")
e.Error(http.StatusBadRequest, err, "查询条件格式不正确")
return
}
response, err := service.List(request)
if err != nil {
e.audit(c, service, "List", auditFailure, "告警列表查询失败")
e.writeError(err)
return
}
e.audit(c, service, "List", auditSuccess, "读取运维告警列表")
e.OK(response, "查询成功")
}
func (e *API) Get(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
response, err := service.Get(c.Param("id"))
if err != nil {
e.audit(c, service, "Get", auditFailure, "告警详情查询失败")
e.writeError(err)
return
}
e.audit(c, service, "Get", auditSuccess, "读取运维告警详情 "+response.Alert.ID)
e.OK(response, "查询成功")
}
func (e *API) Evaluate(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
response, err := service.EvaluateSources(c.Request.Context())
if err != nil {
e.audit(c, service, "Evaluate", auditFailure, "健康事实刷新失败")
e.writeError(err)
return
}
e.audit(c, service, "Evaluate", auditSuccess, "刷新 Sense 内部健康事实")
e.OK(response, "状态已刷新")
}
func (e *API) Acknowledge(c *gin.Context) { e.action(c, "Acknowledge") }
func (e *API) Recover(c *gin.Context) { e.action(c, "Recover") }
func (e *API) action(c *gin.Context, action string) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
request := ActionRequest{}
if err = e.MakeContext(c).Bind(&request, binding.JSON).Errors; err != nil {
e.audit(c, service, action, auditFailure, "告警操作请求格式不正确")
e.Error(http.StatusBadRequest, err, "请求格式不正确")
return
}
var item Alert
if action == "Acknowledge" {
item, err = service.Acknowledge(c.Request.Context(), c.Param("id"), request, user.GetUserId(c))
} else {
item, err = service.Recover(c.Request.Context(), c.Param("id"), request, user.GetUserId(c))
}
if err != nil {
e.audit(c, service, action, auditFailure, "运维告警状态操作被拒绝")
e.writeError(err)
return
}
e.audit(c, service, action, auditSuccess, action+" "+item.ID)
e.OK(item, "操作成功")
}
func (e *API) audit(c *gin.Context, service *Service, action, status, remark string) {
if err := WriteAudit(service.DB, Audit{Action: action, Method: c.Request.Method, Status: status, Username: user.GetUserName(c), UserID: user.GetUserId(c), ClientIP: common.GetClientIP(c), Route: c.FullPath(), Remark: remark, At: time.Now()}); err != nil {
api.GetRequestLogger(c).Errorf("ops alert audit failed: %s", err.Error())
}
}
func (e *API) writeError(err error) {
switch {
case errors.Is(err, ErrInvalidFilter), errors.Is(err, ErrInvalidReason):
e.Error(http.StatusBadRequest, err, err.Error())
case errors.Is(err, ErrVersionConflict), errors.Is(err, ErrInvalidTransition), errors.Is(err, ErrRecoveryWindowOpen):
e.Error(http.StatusConflict, err, err.Error())
case errors.Is(err, ErrAlertNotFound):
e.Error(http.StatusNotFound, err, err.Error())
default:
e.Error(http.StatusInternalServerError, err, "运维告警操作失败")
}
}
+19
View File
@@ -0,0 +1,19 @@
package ops_alert
import (
adminModels "git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
"gorm.io/gorm"
"time"
)
type Audit struct {
Action, Method, Status, Username, ClientIP, Route, Remark string
UserID int
At time.Time
}
func WriteAudit(db *gorm.DB, input Audit) error {
model := adminModels.SysOperaLog{Title: "运维告警", BusinessType: "other", Method: "ops_alert.API." + input.Action, RequestMethod: input.Method, OperatorType: "1", OperName: input.Username, OperUrl: input.Route, OperIp: input.ClientIP, Status: input.Status, OperTime: input.At.UTC(), Remark: input.Remark, CreatedAt: input.At.UTC(), UpdatedAt: input.At.UTC()}
model.CreateBy, model.UpdateBy = input.UserID, input.UserID
return db.Create(&model).Error
}
+37
View File
@@ -0,0 +1,37 @@
package ops_alert
import commonDto "git.ilapage.cn/ila/yovision/Sense/server/common/dto"
type PageRequest struct {
commonDto.Pagination `search:"-"`
AlertType string `form:"alertType"`
State string `form:"state"`
Keyword string `form:"keyword"`
}
type Summary struct {
Active int64 `json:"active"`
Unacknowledged int64 `json:"unacknowledged"`
Recovering int64 `json:"recovering"`
}
type PageResponse struct {
Summary Summary `json:"summary"`
List []Alert `json:"list"`
Count int64 `json:"count"`
}
type DetailResponse struct {
Alert Alert `json:"alert"`
Transitions []Transition `json:"transitions"`
}
type ActionRequest struct {
ExpectedVersion int64 `json:"expectedVersion" binding:"required,min=1"`
Reason string `json:"reason" binding:"required"`
}
type EvaluateResponse struct {
Observed int `json:"observed"`
Abnormal int `json:"abnormal"`
}
@@ -0,0 +1,62 @@
package ops_alert
import "time"
const (
StateUnacknowledged = "unacknowledged"
StateAcknowledged = "acknowledged"
StateRecovering = "recovering"
StateRecovered = "recovered"
TypeDeviceOffline = "device_offline"
TypeAuthentication = "authentication_failed"
TypeClockDrift = "clock_drift"
TypeReconciliation = "reconciliation_failed"
TypeMediaShard = "media_shard_failed"
TypeControlTunnel = "control_tunnel_failed"
)
type Alert struct {
ID string `gorm:"size:36;primaryKey" json:"id"`
Fingerprint string `gorm:"size:255;not null;uniqueIndex" json:"fingerprint"`
AlertType string `gorm:"size:32;not null;index" json:"alertType"`
Severity string `gorm:"size:16;not null;index" json:"severity"`
ObjectType string `gorm:"size:32;not null;index" json:"objectType"`
ObjectID string `gorm:"size:96;not null;index" json:"objectId"`
ObjectName string `gorm:"size:128;not null" json:"objectName"`
Location string `gorm:"size:255;not null;default:''" json:"location"`
State string `gorm:"size:24;not null;index" json:"state"`
Title string `gorm:"size:160;not null" json:"title"`
Detail string `gorm:"size:512;not null" json:"detail"`
NextAction string `gorm:"size:255;not null" json:"nextAction"`
LastSourceVersion string `gorm:"size:128;not null;default:''" json:"-"`
OccurrenceCount int `gorm:"not null;default:1" json:"occurrenceCount"`
Cycle int `gorm:"not null;default:1" json:"cycle"`
FirstSeenAt time.Time `gorm:"not null;index" json:"firstSeenAt"`
LastSeenAt time.Time `gorm:"not null;index" json:"lastSeenAt"`
HealthySince *time.Time `json:"healthySince,omitempty"`
AcknowledgedAt *time.Time `json:"acknowledgedAt,omitempty"`
AcknowledgedBy int `gorm:"not null;default:0" json:"acknowledgedBy,omitempty"`
RecoveredAt *time.Time `json:"recoveredAt,omitempty"`
RecoveredBy int `gorm:"not null;default:0" json:"recoveredBy,omitempty"`
OperationalOnly bool `gorm:"not null;default:true" json:"operationalOnly"`
Version int64 `gorm:"not null;default:1" json:"version"`
CreatedAt time.Time `gorm:"not null" json:"createdAt"`
UpdatedAt time.Time `gorm:"not null" json:"updatedAt"`
}
func (Alert) TableName() string { return "sense_ops_alerts" }
type Transition struct {
ID string `gorm:"size:36;primaryKey" json:"id"`
AlertID string `gorm:"size:36;not null;index" json:"alertId"`
Cycle int `gorm:"not null" json:"cycle"`
Action string `gorm:"size:32;not null;index" json:"action"`
FromState string `gorm:"size:24;not null;default:''" json:"fromState"`
ToState string `gorm:"size:24;not null" json:"toState"`
Reason string `gorm:"size:512;not null;default:''" json:"reason"`
ActorUserID int `gorm:"not null;default:0" json:"actorUserId"`
CreatedAt time.Time `gorm:"not null;index" json:"createdAt"`
}
func (Transition) TableName() string { return "sense_ops_alert_transitions" }
+326
View File
@@ -0,0 +1,326 @@
package ops_alert
import (
"context"
"errors"
"fmt"
"strings"
"time"
"unicode/utf8"
"github.com/google/uuid"
"gorm.io/gorm"
"gorm.io/gorm/clause"
deviceModels "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/edge_node"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
)
var (
ErrAlertNotFound = errors.New("运维告警不存在")
ErrInvalidFilter = errors.New("告警筛选条件不正确")
ErrInvalidReason = errors.New("处理说明需为 6 至 256 个字符")
ErrInvalidTransition = errors.New("当前告警状态不允许此操作")
ErrVersionConflict = errors.New("告警状态已变化,请刷新后重试")
ErrRecoveryWindowOpen = errors.New("健康恢复观察窗口尚未结束")
)
const defaultRecoveryWindow = 5 * time.Minute
type Service struct {
DB *gorm.DB
Now func() time.Time
RecoveryWindow time.Duration
}
func NewService(db *gorm.DB) *Service {
return &Service{DB: db, Now: time.Now, RecoveryWindow: defaultRecoveryWindow}
}
func (s *Service) List(request PageRequest) (PageResponse, error) {
if !validOptional(request.AlertType, alertTypes()) || !validOptional(request.State, states()) {
return PageResponse{}, ErrInvalidFilter
}
query := s.DB.Model(&Alert{})
if request.AlertType != "" {
query = query.Where("alert_type = ?", request.AlertType)
}
if request.State != "" {
query = query.Where("state = ?", request.State)
}
if keyword := strings.TrimSpace(request.Keyword); keyword != "" {
like := "%" + keyword + "%"
query = query.Where("object_name LIKE ? OR location LIKE ? OR title LIKE ? OR detail LIKE ?", like, like, like, like)
}
response := PageResponse{}
if err := query.Count(&response.Count).Error; err != nil {
return response, err
}
if err := query.Order("CASE WHEN state = 'unacknowledged' THEN 0 WHEN state = 'acknowledged' THEN 1 WHEN state = 'recovering' THEN 2 ELSE 3 END, last_seen_at DESC").Offset((request.GetPageIndex() - 1) * request.GetPageSize()).Limit(request.GetPageSize()).Find(&response.List).Error; err != nil {
return response, err
}
active := []string{StateUnacknowledged, StateAcknowledged, StateRecovering}
if err := s.DB.Model(&Alert{}).Where("state IN ?", active).Count(&response.Summary.Active).Error; err != nil {
return response, err
}
if err := s.DB.Model(&Alert{}).Where("state = ?", StateUnacknowledged).Count(&response.Summary.Unacknowledged).Error; err != nil {
return response, err
}
if err := s.DB.Model(&Alert{}).Where("state = ?", StateRecovering).Count(&response.Summary.Recovering).Error; err != nil {
return response, err
}
return response, nil
}
func (s *Service) Get(id string) (DetailResponse, error) {
response := DetailResponse{Transitions: []Transition{}}
if err := s.DB.First(&response.Alert, "id = ?", id).Error; errors.Is(err, gorm.ErrRecordNotFound) {
return response, ErrAlertNotFound
} else if err != nil {
return response, err
}
if err := s.DB.Where("alert_id = ?", id).Order("created_at DESC").Find(&response.Transitions).Error; err != nil {
return response, err
}
return response, nil
}
func (s *Service) Acknowledge(ctx context.Context, id string, request ActionRequest, actor int) (Alert, error) {
return s.transition(ctx, id, request, actor, StateUnacknowledged, StateAcknowledged, "acknowledge")
}
func (s *Service) Recover(ctx context.Context, id string, request ActionRequest, actor int) (Alert, error) {
return s.transition(ctx, id, request, actor, StateRecovering, StateRecovered, "recover")
}
func (s *Service) transition(ctx context.Context, id string, request ActionRequest, actor int, expectedState, targetState, action string) (Alert, error) {
reason := strings.TrimSpace(request.Reason)
if utf8.RuneCountInString(reason) < 6 || utf8.RuneCountInString(reason) > 256 {
return Alert{}, ErrInvalidReason
}
if request.ExpectedVersion < 1 {
return Alert{}, ErrVersionConflict
}
now := s.Now().UTC()
result := Alert{}
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var current Alert
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&current, "id = ?", id).Error; errors.Is(err, gorm.ErrRecordNotFound) {
return ErrAlertNotFound
} else if err != nil {
return err
}
if current.Version != request.ExpectedVersion {
return ErrVersionConflict
}
if current.State != expectedState {
return ErrInvalidTransition
}
if action == "recover" && (current.HealthySince == nil || now.Sub(current.HealthySince.UTC()) < s.RecoveryWindow) {
return ErrRecoveryWindowOpen
}
updates := map[string]any{"state": targetState, "version": current.Version + 1, "updated_at": now}
if action == "acknowledge" {
updates["acknowledged_at"], updates["acknowledged_by"] = now, actor
} else {
updates["recovered_at"], updates["recovered_by"] = now, actor
}
write := tx.Model(&Alert{}).Where("id = ? AND version = ? AND state = ?", id, current.Version, expectedState).Updates(updates)
if write.Error != nil {
return write.Error
}
if write.RowsAffected != 1 {
return ErrVersionConflict
}
if err := tx.Create(&Transition{ID: uuid.NewString(), AlertID: id, Cycle: current.Cycle, Action: action, FromState: expectedState, ToState: targetState, Reason: reason, ActorUserID: actor, CreatedAt: now}).Error; err != nil {
return err
}
return tx.First(&result, "id = ?", id).Error
})
return result, err
}
type observation struct {
Fingerprint, AlertType, Severity, ObjectType, ObjectID, ObjectName, Location, Title, Detail, NextAction, SourceVersion string
Abnormal bool
}
func (s *Service) EvaluateSources(ctx context.Context) (EvaluateResponse, error) {
items, err := s.sourceObservations(ctx)
if err != nil {
return EvaluateResponse{}, err
}
response := EvaluateResponse{Observed: len(items)}
for _, item := range items {
if item.Abnormal {
response.Abnormal++
}
if err = s.observe(ctx, item); err != nil {
return response, err
}
}
return response, nil
}
func (s *Service) observe(ctx context.Context, item observation) error {
now := s.Now().UTC()
return s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var current Alert
err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("fingerprint = ?", item.Fingerprint).First(&current).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
if !item.Abnormal {
return nil
}
current = Alert{ID: uuid.NewString(), Fingerprint: item.Fingerprint, AlertType: item.AlertType, Severity: item.Severity, ObjectType: item.ObjectType, ObjectID: item.ObjectID, ObjectName: item.ObjectName, Location: item.Location, State: StateUnacknowledged, Title: item.Title, Detail: item.Detail, NextAction: item.NextAction, LastSourceVersion: item.SourceVersion, OccurrenceCount: 1, Cycle: 1, FirstSeenAt: now, LastSeenAt: now, OperationalOnly: true, Version: 1, CreatedAt: now, UpdatedAt: now}
if err = tx.Create(&current).Error; err != nil {
return err
}
return tx.Create(&Transition{ID: uuid.NewString(), AlertID: current.ID, Cycle: 1, Action: "detected", ToState: StateUnacknowledged, Reason: "健康事实首次满足告警规则", CreatedAt: now}).Error
}
if err != nil {
return err
}
if item.Abnormal {
updates := map[string]any{"severity": item.Severity, "object_name": item.ObjectName, "location": item.Location, "title": item.Title, "detail": item.Detail, "next_action": item.NextAction, "last_seen_at": now, "updated_at": now, "healthy_since": nil}
if item.SourceVersion != current.LastSourceVersion {
updates["last_source_version"], updates["occurrence_count"] = item.SourceVersion, current.OccurrenceCount+1
}
if current.State == StateRecovered {
updates["state"], updates["cycle"], updates["first_seen_at"], updates["occurrence_count"], updates["acknowledged_at"], updates["acknowledged_by"], updates["recovered_at"], updates["recovered_by"] = StateUnacknowledged, current.Cycle+1, now, 1, nil, 0, nil, 0
if err = tx.Create(&Transition{ID: uuid.NewString(), AlertID: current.ID, Cycle: current.Cycle + 1, Action: "reopened", FromState: StateRecovered, ToState: StateUnacknowledged, Reason: "健康事实再次异常", CreatedAt: now}).Error; err != nil {
return err
}
} else if current.State == StateRecovering {
target := StateUnacknowledged
if current.AcknowledgedAt != nil {
target = StateAcknowledged
}
updates["state"] = target
if err = tx.Create(&Transition{ID: uuid.NewString(), AlertID: current.ID, Cycle: current.Cycle, Action: "relapsed", FromState: StateRecovering, ToState: target, Reason: "恢复观察期间再次异常", CreatedAt: now}).Error; err != nil {
return err
}
}
updates["version"] = current.Version + 1
return tx.Model(&Alert{}).Where("id = ? AND version = ?", current.ID, current.Version).Updates(updates).Error
}
if current.State == StateUnacknowledged || current.State == StateAcknowledged {
if err = tx.Create(&Transition{ID: uuid.NewString(), AlertID: current.ID, Cycle: current.Cycle, Action: "health_restored", FromState: current.State, ToState: StateRecovering, Reason: "健康事实已恢复,进入稳定观察窗口", CreatedAt: now}).Error; err != nil {
return err
}
return tx.Model(&Alert{}).Where("id = ? AND version = ?", current.ID, current.Version).Updates(map[string]any{"state": StateRecovering, "healthy_since": now, "version": current.Version + 1, "updated_at": now}).Error
}
return nil
})
}
type admissionProjection struct {
DeviceID, Status, Detail string
CheckedAt time.Time
}
func (admissionProjection) TableName() string { return "sense_admission_results" }
func (s *Service) sourceObservations(ctx context.Context) ([]observation, error) {
now := s.Now().UTC()
result := []observation{}
var devices []deviceModels.Device
if err := s.DB.WithContext(ctx).Find(&devices).Error; err != nil {
return nil, err
}
var admissions []admissionProjection
if err := s.DB.WithContext(ctx).Find(&admissions).Error; err != nil && !isMissingTable(err) {
return nil, err
}
byDevice := map[string]admissionProjection{}
for _, item := range admissions {
byDevice[item.DeviceID] = item
}
for _, device := range devices {
if device.Status == deviceModels.StatusDisabled {
continue
}
state, detail, observed := device.AdapterStatus, "尚未获得设备接入结果", device.UpdatedAt
if item, ok := byDevice[device.ID]; ok {
state, detail, observed = item.Status, item.Detail, item.CheckedAt
}
text := strings.ToLower(state + " " + detail)
version := fmt.Sprintf("%d:%d", device.Version, observed.UnixNano())
base := observation{ObjectType: "device", ObjectID: device.ID, ObjectName: device.Name, Location: device.Location, SourceVersion: version}
result = append(result,
withRule(base, TypeDeviceOffline, "high", "设备离线", detail, "检查设备供电、网络和地址后重新发现", containsAny(text, "offline", "unreachable", "timeout", "离线", "无法连接", "超时")),
withRule(base, TypeAuthentication, "high", "设备认证失败", detail, "更新设备凭据后重新执行接入验证", containsAny(text, "authentication_failed", "unauthorized", "auth", "认证", "凭据")),
withRule(base, TypeClockDrift, "medium", "设备时间漂移", detail, "校准设备时间后重新执行接入验证", containsAny(text, "clock_skew", "clock_drift", "time drift", "时间漂移", "时钟")),
)
}
var routes []media.Route
if err := s.DB.WithContext(ctx).Find(&routes).Error; err != nil && !isMissingTable(err) {
return nil, err
}
for _, route := range routes {
converged := (route.Desired == media.DesiredRunning && (route.Actual == "ready" || route.Actual == "waiting")) || (route.Desired == media.DesiredStopped && route.Actual == "stopped")
base := observation{ObjectType: "media_route", ObjectID: route.ID, ObjectName: route.Path, SourceVersion: fmt.Sprintf("%d", route.Version)}
result = append(result, withRule(base, TypeReconciliation, "medium", "媒体状态未收敛", route.Detail, "在运维中心核对期望态并执行受控重试", !converged))
}
var shards []media_shard.Shard
if err := s.DB.WithContext(ctx).Find(&shards).Error; err != nil && !isMissingTable(err) {
return nil, err
}
for _, shard := range shards {
if shard.Status == media_shard.StatusDisabled {
continue
}
stale := shard.LastProbeAt == nil || now.Sub(shard.LastProbeAt.UTC()) > 30*time.Second
base := observation{ObjectType: "media_shard", ObjectID: shard.ID, ObjectName: shard.Name, SourceVersion: fmt.Sprintf("%d:%d", shard.ConfigVersion, timeValue(shard.LastProbeAt))}
result = append(result, withRule(base, TypeMediaShard, "high", "媒体分片异常", shard.Detail, "检查 MediaMTX 进程和 Control API 后重新探测", shard.Status == media_shard.StatusFailed || stale))
}
var nodes []edge_node.Node
if err := s.DB.WithContext(ctx).Find(&nodes).Error; err != nil && !isMissingTable(err) {
return nil, err
}
for _, node := range nodes {
base := observation{ObjectType: "edge_node", ObjectID: node.ID, ObjectName: node.Name, Location: node.Location, SourceVersion: fmt.Sprintf("%d:%d", node.ProjectionVersion, node.LastHeartbeatAt.UnixNano())}
result = append(result,
withRule(base, TypeDeviceOffline, "high", "边缘节点离线", "心跳超过 90 秒未更新", "检查节点进程、网络和机器身份后等待心跳恢复", now.Sub(node.LastHeartbeatAt.UTC()) > edge_node.HeartbeatTimeout),
withRule(base, TypeControlTunnel, "high", "控制隧道异常", node.ControlTunnelDetail, "检查节点到 Sense 的控制通道,不影响业务预警数据", node.ControlTunnelStatus != edge_node.ChannelReady),
)
}
return result, nil
}
func withRule(base observation, typ, severity, title, detail, next string, abnormal bool) observation {
base.Fingerprint = typ + ":" + base.ObjectType + ":" + base.ObjectID
base.AlertType = typ
base.Severity = severity
base.Title = title
base.Detail = detail
base.NextAction = next
base.Abnormal = abnormal
return base
}
func containsAny(value string, needles ...string) bool {
for _, needle := range needles {
if strings.Contains(value, needle) {
return true
}
}
return false
}
func isMissingTable(err error) bool {
text := strings.ToLower(err.Error())
return strings.Contains(text, "no such table") || strings.Contains(text, "does not exist")
}
func timeValue(value *time.Time) int64 {
if value == nil {
return 0
}
return value.UnixNano()
}
func validOptional(value string, allowed map[string]bool) bool { return value == "" || allowed[value] }
func states() map[string]bool {
return map[string]bool{StateUnacknowledged: true, StateAcknowledged: true, StateRecovering: true, StateRecovered: true}
}
func alertTypes() map[string]bool {
return map[string]bool{TypeDeviceOffline: true, TypeAuthentication: true, TypeClockDrift: true, TypeReconciliation: true, TypeMediaShard: true, TypeControlTunnel: true}
}
@@ -0,0 +1,198 @@
package ops_alert
import (
"context"
"errors"
"testing"
"time"
"github.com/google/uuid"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
adminModels "git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
deviceModels "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/edge_node"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
)
func opsAlertTestService(t *testing.T) (*Service, *gorm.DB, time.Time) {
t.Helper()
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&Alert{}, &Transition{}, &deviceModels.Device{}, &admissionProjection{}, &media.Route{}, &media_shard.Shard{}, &edge_node.Node{}, &adminModels.SysOperaLog{}); err != nil {
t.Fatal(err)
}
now := time.Date(2026, 8, 28, 8, 0, 0, 0, time.UTC)
service := NewService(db)
service.Now = func() time.Time { return now }
return service, db, now
}
func TestEvaluateSourcesGeneratesSixTypesAndDeduplicates(t *testing.T) {
service, db, now := opsAlertTestService(t)
devices := []deviceModels.Device{
{ID: "offline", Name: "东门摄像机", Location: "东门", Modality: "video", CapabilitiesJSON: "[]", Status: "pending", AdapterStatus: "verification_failed", Version: 1},
{ID: "auth", Name: "仓库摄像机", Location: "仓库", Modality: "video", CapabilitiesJSON: "[]", Status: "pending", AdapterStatus: "verification_failed", Version: 2},
{ID: "clock", Name: "南门摄像机", Location: "南门", Modality: "video", CapabilitiesJSON: "[]", Status: "pending", AdapterStatus: "verification_failed", Version: 3},
}
if err := db.Create(&devices).Error; err != nil {
t.Fatal(err)
}
results := []admissionProjection{
{DeviceID: "offline", Status: "device_offline", Detail: "设备离线,无法连接", CheckedAt: now},
{DeviceID: "auth", Status: "authentication_failed", Detail: "设备认证失败", CheckedAt: now},
{DeviceID: "clock", Status: "clock_skew", Detail: "设备时间漂移", CheckedAt: now},
}
if err := db.Create(&results).Error; err != nil {
t.Fatal(err)
}
if err := db.Create(&media.Route{ID: "route-1", DeviceID: "auth", ProfileToken: "main", Path: "sense_auth", Desired: "running", Actual: "apply_failed", Detail: "配置未收敛", Version: 4, UpdatedAt: now}).Error; err != nil {
t.Fatal(err)
}
probe := now
if err := db.Create(&media_shard.Shard{ID: "shard-1", Name: "媒体服务 A", Mode: "local", ControlAPI: "http://127.0.0.1:9997", Capacity: 16, Status: media_shard.StatusFailed, Detail: "Control API 不可用", LastProbeAt: &probe, ConfigVersion: 1, CreatedAt: now, UpdatedAt: now}).Error; err != nil {
t.Fatal(err)
}
if err := db.Create(&edge_node.Node{ID: "node-1", Name: "主楼边缘节点", Location: "机房", StartedAt: now.Add(-time.Hour), LastHeartbeatAt: now.Add(-10 * time.Second), LastCollectedAt: now.Add(-10 * time.Second), ControlTunnelStatus: edge_node.ChannelUnavailable, ControlTunnelDetail: "TLS 隧道断开", VideoPlaneStatus: edge_node.ChannelReady, ProjectionVersion: 7}).Error; err != nil {
t.Fatal(err)
}
response, err := service.EvaluateSources(context.Background())
if err != nil {
t.Fatal(err)
}
if response.Abnormal != 6 {
t.Fatalf("abnormal=%d want 6", response.Abnormal)
}
var alerts []Alert
if err = db.Find(&alerts).Error; err != nil {
t.Fatal(err)
}
if len(alerts) != 6 {
t.Fatalf("alerts=%d want 6", len(alerts))
}
types := map[string]bool{}
for _, item := range alerts {
types[item.AlertType] = true
if !item.OperationalOnly {
t.Fatal("alert crossed operational boundary")
}
}
for _, typ := range []string{TypeDeviceOffline, TypeAuthentication, TypeClockDrift, TypeReconciliation, TypeMediaShard, TypeControlTunnel} {
if !types[typ] {
t.Fatalf("missing type %s", typ)
}
}
if _, err = service.EvaluateSources(context.Background()); err != nil {
t.Fatal(err)
}
var count int64
db.Model(&Alert{}).Count(&count)
if count != 6 {
t.Fatalf("duplicate active alerts: %d", count)
}
var offline Alert
if err = db.First(&offline, "fingerprint = ?", TypeDeviceOffline+":device:offline").Error; err != nil {
t.Fatal(err)
}
if offline.OccurrenceCount != 1 {
t.Fatalf("unchanged source counted twice: %d", offline.OccurrenceCount)
}
}
func TestAcknowledgeRecoverAndReopenPreserveOneAlert(t *testing.T) {
service, db, now := opsAlertTestService(t)
device := deviceModels.Device{ID: "auth", Name: "仓库摄像机", Modality: "video", CapabilitiesJSON: "[]", Status: "pending", AdapterStatus: "verification_failed", Version: 1}
if err := db.Create(&device).Error; err != nil {
t.Fatal(err)
}
result := admissionProjection{DeviceID: "auth", Status: "authentication_failed", Detail: "认证失败", CheckedAt: now}
if err := db.Create(&result).Error; err != nil {
t.Fatal(err)
}
if _, err := service.EvaluateSources(context.Background()); err != nil {
t.Fatal(err)
}
var alert Alert
if err := db.First(&alert, "alert_type = ?", TypeAuthentication).Error; err != nil {
t.Fatal(err)
}
ack, err := service.Acknowledge(context.Background(), alert.ID, ActionRequest{ExpectedVersion: alert.Version, Reason: "已安排现场人员更新设备凭据"}, 7)
if err != nil {
t.Fatal(err)
}
if ack.State != StateAcknowledged || ack.AcknowledgedBy != 7 {
t.Fatalf("unexpected acknowledgement: %+v", ack)
}
if _, err = service.Acknowledge(context.Background(), alert.ID, ActionRequest{ExpectedVersion: alert.Version, Reason: "并发重复确认不应成功"}, 8); !errors.Is(err, ErrVersionConflict) {
t.Fatalf("want version conflict, got %v", err)
}
if err = db.Model(&admissionProjection{}).Where("device_id = ?", "auth").Updates(map[string]any{"status": "ready", "detail": "接入验证完成", "checked_at": now.Add(time.Minute)}).Error; err != nil {
t.Fatal(err)
}
service.Now = func() time.Time { return now.Add(time.Minute) }
if _, err = service.EvaluateSources(context.Background()); err != nil {
t.Fatal(err)
}
if err = db.First(&alert, "id = ?", alert.ID).Error; err != nil {
t.Fatal(err)
}
if alert.State != StateRecovering || alert.HealthySince == nil {
t.Fatalf("want recovering: %+v", alert)
}
if _, err = service.Recover(context.Background(), alert.ID, ActionRequest{ExpectedVersion: alert.Version, Reason: "恢复窗口尚未结束,不能关闭"}, 7); !errors.Is(err, ErrRecoveryWindowOpen) {
t.Fatalf("want recovery window error, got %v", err)
}
service.Now = func() time.Time { return now.Add(7 * time.Minute) }
recovered, err := service.Recover(context.Background(), alert.ID, ActionRequest{ExpectedVersion: alert.Version, Reason: "设备已稳定在线超过恢复观察窗口"}, 7)
if err != nil {
t.Fatal(err)
}
if recovered.State != StateRecovered {
t.Fatalf("state=%s", recovered.State)
}
if _, err = service.EvaluateSources(context.Background()); err != nil {
t.Fatal(err)
}
if err = db.Model(&admissionProjection{}).Where("device_id = ?", "auth").Updates(map[string]any{"status": "authentication_failed", "detail": "认证再次失败", "checked_at": now.Add(8 * time.Minute)}).Error; err != nil {
t.Fatal(err)
}
service.Now = func() time.Time { return now.Add(8 * time.Minute) }
if _, err = service.EvaluateSources(context.Background()); err != nil {
t.Fatal(err)
}
var count int64
db.Model(&Alert{}).Where("fingerprint = ?", recovered.Fingerprint).Count(&count)
if count != 1 {
t.Fatalf("reopen duplicated alert: %d", count)
}
if err = db.First(&alert, "id = ?", alert.ID).Error; err != nil {
t.Fatal(err)
}
if alert.State != StateUnacknowledged || alert.Cycle != 2 {
t.Fatalf("unexpected reopened alert: %+v", alert)
}
var transitions int64
db.Model(&Transition{}).Where("alert_id = ?", alert.ID).Count(&transitions)
if transitions < 5 {
t.Fatalf("transitions=%d", transitions)
}
}
func TestAuditIsStoredOnlyInGoAdminOperationLog(t *testing.T) {
_, db, now := opsAlertTestService(t)
if err := WriteAudit(db, Audit{Action: "Acknowledge", Method: "POST", Status: "1", Username: "operator", UserID: 9, ClientIP: "127.0.0.1", Route: "/api/v1/ops-alerts/:id/acknowledge", Remark: "确认运维告警", At: now}); err != nil {
t.Fatal(err)
}
var log adminModels.SysOperaLog
if err := db.First(&log).Error; err != nil {
t.Fatal(err)
}
if log.Title != "运维告警" || log.CreateBy != 9 {
t.Fatalf("unexpected audit: %+v", log)
}
}
+103
View File
@@ -0,0 +1,103 @@
package outbox
import (
"errors"
"net/http"
"time"
"github.com/gin-gonic/gin"
"github.com/gin-gonic/gin/binding"
"github.com/go-admin-team/go-admin-core/sdk/api"
"github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth/user"
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
"git.ilapage.cn/ila/yovision/Sense/server/common"
)
type API struct{ api.Api }
func (e *API) service(c *gin.Context) (*Service, error) {
base := coreService.Service{}
if err := e.MakeContext(c).MakeOrm().MakeService(&base).Errors; err != nil {
return nil, err
}
return NewService(base.Orm), nil
}
func (e *API) List(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
request := PageRequest{}
if err = e.MakeContext(c).Bind(&request).Errors; err != nil {
e.audit(c, service, "List", auditFailure, "可靠投递查询条件格式不正确")
e.Error(http.StatusBadRequest, err, "查询条件格式不正确")
return
}
response, err := service.List(request)
if err != nil {
e.audit(c, service, "List", auditFailure, "可靠投递列表查询失败")
e.writeError(err)
return
}
e.audit(c, service, "List", auditSuccess, "读取可靠投递列表")
e.OK(response, "查询成功")
}
func (e *API) Get(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
response, err := service.Get(c.Param("id"))
if err != nil {
e.audit(c, service, "Get", auditFailure, "可靠投递详情查询失败")
e.writeError(err)
return
}
e.audit(c, service, "Get", auditSuccess, "读取可靠投递详情 "+response.Message.ID)
e.OK(response, "查询成功")
}
func (e *API) Requeue(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
request := RequeueRequest{}
if err = e.MakeContext(c).Bind(&request, binding.JSON).Errors; err != nil {
e.audit(c, service, "Requeue", auditFailure, "死信重新排队请求格式不正确")
e.Error(http.StatusBadRequest, err, "请求格式不正确")
return
}
message, err := service.Requeue(c.Param("id"), request, user.GetUserId(c))
if err != nil {
e.audit(c, service, "Requeue", auditFailure, "死信重新排队被拒绝")
e.writeError(err)
return
}
e.audit(c, service, "Requeue", auditSuccess, "死信已重新排队 "+message.ID)
e.OK(message, "已重新排队")
}
func (e *API) audit(c *gin.Context, service *Service, action, status, remark string) {
if err := WriteAudit(service.Orm, Audit{Action: action, Method: c.Request.Method, Status: status, Username: user.GetUserName(c), UserID: user.GetUserId(c), ClientIP: common.GetClientIP(c), Route: c.FullPath(), Remark: remark, At: time.Now()}); err != nil {
api.GetRequestLogger(c).Errorf("outbox audit failed: %s", err.Error())
}
}
func (e *API) writeError(err error) {
switch {
case errors.Is(err, ErrInvalidInput):
e.Error(http.StatusBadRequest, err, err.Error())
case errors.Is(err, ErrNotFound):
e.Error(http.StatusNotFound, err, err.Error())
case errors.Is(err, ErrNotDead), errors.Is(err, ErrVersionConflict):
e.Error(http.StatusConflict, err, err.Error())
default:
e.Error(http.StatusInternalServerError, err, "可靠投递操作失败")
}
}
+23
View File
@@ -0,0 +1,23 @@
package outbox
import (
"time"
"gorm.io/gorm"
adminModels "git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
)
const auditSuccess, auditFailure = "1", "2"
type Audit struct {
Action, Method, Status, Username, ClientIP, Route, Remark string
UserID int
At time.Time
}
func WriteAudit(db *gorm.DB, input Audit) error {
model := adminModels.SysOperaLog{Title: "可靠投递", BusinessType: "other", Method: "outbox.API." + input.Action, RequestMethod: input.Method, OperatorType: "1", OperName: input.Username, OperUrl: input.Route, OperIp: input.ClientIP, Status: input.Status, OperTime: input.At.UTC(), Remark: input.Remark, CreatedAt: input.At.UTC(), UpdatedAt: input.At.UTC()}
model.CreateBy, model.UpdateBy = input.UserID, input.UserID
return db.Create(&model).Error
}
+41
View File
@@ -0,0 +1,41 @@
package outbox
import commonDTO "git.ilapage.cn/ila/yovision/Sense/server/common/dto"
type PageRequest struct {
commonDTO.Pagination `search:"-"`
State string `form:"state"`
InternalType string `form:"internalType"`
Keyword string `form:"keyword"`
}
type Summary struct {
Pending int64 `json:"pending"`
Retry int64 `json:"retry"`
Processing int64 `json:"processing"`
Dead int64 `json:"dead"`
}
type PageResponse struct {
List []Message `json:"list"`
Count int64 `json:"count"`
Summary Summary `json:"summary"`
}
type DetailResponse struct {
Message Message `json:"message"`
Attempts []Attempt `json:"attempts"`
}
type RequeueRequest struct {
ExpectedVersion int64 `json:"expectedVersion" binding:"required,min=1"`
Reason string `json:"reason" binding:"required"`
}
type EnqueueInput struct {
InternalType string
BusinessRef string
IdempotencyKey string
PayloadJSON []byte
MaxAttempts int
}
+58
View File
@@ -0,0 +1,58 @@
package outbox
import "time"
const (
StatePending = "pending"
StateProcessing = "processing"
StateRetry = "retry"
StateDead = "dead"
StateDelivered = "delivered"
)
// Message is a Sense-internal delivery record. PayloadJSON is deliberately
// excluded from management APIs and is not a cross-product contract.
type Message struct {
ID string `gorm:"size:36;primaryKey" json:"id"`
InternalType string `gorm:"size:64;not null;index" json:"internalType"`
BusinessRef string `gorm:"size:128;not null;index" json:"businessRef"`
IdempotencyKey string `gorm:"size:191;not null;uniqueIndex" json:"idempotencyKey"`
PayloadJSON string `gorm:"column:payload;type:jsonb;not null" json:"-"`
State string `gorm:"size:24;not null;index" json:"state"`
AttemptCount int `gorm:"not null;default:0" json:"attemptCount"`
MaxAttempts int `gorm:"not null;default:12" json:"maxAttempts"`
AvailableAt time.Time `gorm:"not null;index" json:"availableAt"`
LeaseOwner string `gorm:"size:128;not null;default:''" json:"leaseOwner,omitempty"`
LeaseUntil *time.Time `gorm:"index" json:"leaseUntil,omitempty"`
LastError string `gorm:"size:512;not null;default:''" json:"lastError,omitempty"`
DeliveredAt *time.Time `json:"deliveredAt,omitempty"`
Version int64 `gorm:"not null;default:1" json:"version"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
func (Message) TableName() string { return "sense_outbox_messages" }
// DeliveryRecord is permanent idempotency evidence. It is never deleted by
// queue cleanup and prevents a delivered key from being processed again.
type DeliveryRecord struct {
IdempotencyKey string `gorm:"size:191;primaryKey" json:"idempotencyKey"`
MessageID string `gorm:"size:36;not null;uniqueIndex" json:"messageId"`
DeliveredAt time.Time `gorm:"not null" json:"deliveredAt"`
CreatedAt time.Time `json:"createdAt"`
}
func (DeliveryRecord) TableName() string { return "sense_outbox_deliveries" }
type Attempt struct {
ID uint `gorm:"primaryKey;autoIncrement" json:"id"`
MessageID string `gorm:"size:36;not null;index" json:"messageId"`
Number int `gorm:"not null" json:"number"`
Outcome string `gorm:"size:32;not null" json:"outcome"`
Detail string `gorm:"size:512;not null;default:''" json:"detail"`
Worker string `gorm:"size:128;not null;default:''" json:"worker,omitempty"`
ActorUserID int `gorm:"not null;default:0" json:"actorUserId,omitempty"`
CreatedAt time.Time `json:"createdAt"`
}
func (Attempt) TableName() string { return "sense_outbox_attempts" }
@@ -0,0 +1,92 @@
package outbox
import (
"fmt"
"net/url"
"os"
"strings"
"sync"
"testing"
"time"
"gorm.io/driver/postgres"
"gorm.io/gorm"
)
func TestPostgresConcurrentWorkersDoNotClaimSameMessage(t *testing.T) {
dsn := os.Getenv("SENSE_OUTBOX_TEST_DATABASE_URL")
if dsn == "" {
t.Skip("set SENSE_OUTBOX_TEST_DATABASE_URL to run the PostgreSQL multi-worker test")
}
base, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
schema := fmt.Sprintf("sense_outbox_78_%d", time.Now().UnixNano())
if err = base.Exec("CREATE SCHEMA " + schema).Error; err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = base.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE").Error })
scoped, err := gorm.Open(postgres.Open(withSearchPath(dsn, schema)), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = scoped.AutoMigrate(&Message{}, &DeliveryRecord{}, &Attempt{}); err != nil {
t.Fatal(err)
}
now := time.Date(2026, 8, 28, 12, 0, 0, 0, time.UTC)
for index := 0; index < 20; index++ {
if _, err = Enqueue(scoped, EnqueueInput{InternalType: "local_event_candidate", BusinessRef: fmt.Sprintf("event-%d", index), IdempotencyKey: fmt.Sprintf("event:%d:v1", index), PayloadJSON: []byte(`{}`)}, now); err != nil {
t.Fatal(err)
}
}
workers := []string{"worker-a", "worker-b"}
results := make(chan []Message, len(workers))
errorsCh := make(chan error, len(workers))
var group sync.WaitGroup
for _, worker := range workers {
group.Add(1)
go func(name string) {
defer group.Done()
relay := NewRelay(scoped)
relay.Now = func() time.Time { return now }
items, claimErr := relay.Claim(name, 20)
if claimErr != nil {
errorsCh <- claimErr
return
}
results <- items
}(worker)
}
group.Wait()
close(results)
close(errorsCh)
for claimErr := range errorsCh {
t.Fatal(claimErr)
}
seen := map[string]string{}
for batch := range results {
for _, item := range batch {
if owner, exists := seen[item.ID]; exists {
t.Fatalf("message %s claimed by %s and %s", item.ID, owner, item.LeaseOwner)
}
seen[item.ID] = item.LeaseOwner
}
}
if len(seen) != 20 {
t.Fatalf("claimed=%d want=20", len(seen))
}
}
func withSearchPath(dsn, schema string) string {
if strings.Contains(dsn, "://") {
parsed, err := url.Parse(dsn)
if err == nil {
query := parsed.Query()
query.Set("search_path", schema)
parsed.RawQuery = query.Encode()
return parsed.String()
}
}
return strings.TrimSpace(dsn) + " search_path=" + schema
}
+177
View File
@@ -0,0 +1,177 @@
package outbox
import (
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"github.com/google/uuid"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
var (
ErrDuplicateIdempotency = errors.New("可靠投递幂等键已存在")
ErrLeaseLost = errors.New("可靠投递租约已失效")
)
func Enqueue(tx *gorm.DB, input EnqueueInput, now time.Time) (Message, error) {
input.InternalType, input.BusinessRef, input.IdempotencyKey = strings.TrimSpace(input.InternalType), strings.TrimSpace(input.BusinessRef), strings.TrimSpace(input.IdempotencyKey)
if input.InternalType == "" || len(input.InternalType) > 64 || input.BusinessRef == "" || len(input.BusinessRef) > 128 || input.IdempotencyKey == "" || len(input.IdempotencyKey) > 191 || !json.Valid(input.PayloadJSON) {
return Message{}, ErrInvalidInput
}
if input.MaxAttempts == 0 {
input.MaxAttempts = 12
}
if input.MaxAttempts < 1 || input.MaxAttempts > 100 {
return Message{}, ErrInvalidInput
}
now = now.UTC()
message := Message{ID: uuid.NewString(), InternalType: input.InternalType, BusinessRef: input.BusinessRef, IdempotencyKey: input.IdempotencyKey, PayloadJSON: string(input.PayloadJSON), State: StatePending, MaxAttempts: input.MaxAttempts, AvailableAt: now, Version: 1, CreatedAt: now, UpdatedAt: now}
var existing int64
if err := tx.Model(&Message{}).Where("idempotency_key = ?", input.IdempotencyKey).Count(&existing).Error; err != nil {
return Message{}, fmt.Errorf("check outbox idempotency: %w", err)
}
if existing > 0 {
return Message{}, ErrDuplicateIdempotency
}
if err := tx.Create(&message).Error; err != nil {
return Message{}, fmt.Errorf("enqueue outbox message: %w", err)
}
return message, nil
}
type Relay struct {
DB *gorm.DB
Now func() time.Time
LeaseDuration time.Duration
Backoff func(int) time.Duration
}
func NewRelay(db *gorm.DB) *Relay {
return &Relay{DB: db, Now: time.Now, LeaseDuration: 30 * time.Second, Backoff: defaultBackoff}
}
func (r *Relay) Claim(worker string, limit int) ([]Message, error) {
worker = strings.TrimSpace(worker)
if worker == "" || len(worker) > 128 || limit < 1 || limit > 100 {
return nil, ErrInvalidInput
}
now, leaseUntil := r.now(), r.now().Add(r.leaseDuration())
claimed := make([]Message, 0, limit)
err := r.DB.Transaction(func(tx *gorm.DB) error {
var candidates []Message
query := tx.Where("((state IN ?) AND available_at <= ?) OR (state = ? AND lease_until < ?)", []string{StatePending, StateRetry}, now, StateProcessing, now).Order("available_at ASC, created_at ASC").Limit(limit)
if tx.Dialector.Name() == "postgres" {
query = query.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"})
}
if err := query.Find(&candidates).Error; err != nil {
return err
}
for _, item := range candidates {
result := tx.Model(&Message{}).Where("id = ? AND version = ?", item.ID, item.Version).Updates(map[string]interface{}{"state": StateProcessing, "lease_owner": worker, "lease_until": leaseUntil, "version": gorm.Expr("version + 1"), "updated_at": now})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 1 {
item.State, item.LeaseOwner, item.LeaseUntil, item.Version = StateProcessing, worker, &leaseUntil, item.Version+1
claimed = append(claimed, item)
}
}
return nil
})
if err != nil {
return nil, fmt.Errorf("claim outbox messages: %w", err)
}
return claimed, nil
}
func (r *Relay) MarkSuccess(id, worker string) error {
now := r.now()
return r.DB.Transaction(func(tx *gorm.DB) error {
var message Message
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&message, "id = ?", id).Error; err != nil {
return err
}
var delivered int64
if err := tx.Model(&DeliveryRecord{}).Where("idempotency_key = ?", message.IdempotencyKey).Count(&delivered).Error; err != nil {
return err
}
if delivered > 0 {
return nil
}
if message.State != StateProcessing || message.LeaseOwner != worker || message.LeaseUntil == nil || !message.LeaseUntil.After(now) {
return ErrLeaseLost
}
if err := tx.Create(&DeliveryRecord{IdempotencyKey: message.IdempotencyKey, MessageID: message.ID, DeliveredAt: now, CreatedAt: now}).Error; err != nil {
return err
}
number := message.AttemptCount + 1
if err := tx.Create(&Attempt{MessageID: message.ID, Number: number, Outcome: StateDelivered, Detail: "投递成功", Worker: worker, CreatedAt: now}).Error; err != nil {
return err
}
return tx.Model(&message).Updates(map[string]interface{}{"state": StateDelivered, "attempt_count": number, "delivered_at": now, "lease_owner": "", "lease_until": nil, "last_error": "", "version": gorm.Expr("version + 1"), "updated_at": now}).Error
})
}
func (r *Relay) MarkFailure(id, worker, detail string) error {
now := r.now()
detail = strings.TrimSpace(detail)
if len(detail) > 512 {
detail = detail[:512]
}
if detail == "" {
detail = "投递失败"
}
return r.DB.Transaction(func(tx *gorm.DB) error {
var message Message
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&message, "id = ?", id).Error; err != nil {
return err
}
if message.State != StateProcessing || message.LeaseOwner != worker || message.LeaseUntil == nil || !message.LeaseUntil.After(now) {
return ErrLeaseLost
}
nextAttempt := message.AttemptCount + 1
state := StateRetry
available := now.Add(r.backoff(nextAttempt))
if nextAttempt >= message.MaxAttempts {
state = StateDead
available = now
}
if err := tx.Model(&message).Updates(map[string]interface{}{"state": state, "attempt_count": nextAttempt, "available_at": available, "lease_owner": "", "lease_until": nil, "last_error": detail, "version": gorm.Expr("version + 1"), "updated_at": now}).Error; err != nil {
return err
}
return tx.Create(&Attempt{MessageID: message.ID, Number: nextAttempt, Outcome: state, Detail: detail, Worker: worker, CreatedAt: now}).Error
})
}
func (r *Relay) now() time.Time {
if r.Now != nil {
return r.Now().UTC()
}
return time.Now().UTC()
}
func (r *Relay) leaseDuration() time.Duration {
if r.LeaseDuration <= 0 {
return 30 * time.Second
}
return r.LeaseDuration
}
func (r *Relay) backoff(attempt int) time.Duration {
if r.Backoff != nil {
return r.Backoff(attempt)
}
return defaultBackoff(attempt)
}
func defaultBackoff(attempt int) time.Duration {
if attempt < 1 {
attempt = 1
}
delay := time.Second * time.Duration(1<<min(attempt-1, 8))
if delay > 5*time.Minute {
return 5 * time.Minute
}
return delay
}
+122
View File
@@ -0,0 +1,122 @@
package outbox
import (
"errors"
"testing"
"time"
"github.com/google/uuid"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
func testDB(t *testing.T) *gorm.DB {
t.Helper()
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
sqlDB, _ := db.DB()
sqlDB.SetMaxOpenConns(1)
if err = db.AutoMigrate(&Message{}, &DeliveryRecord{}, &Attempt{}); err != nil {
t.Fatal(err)
}
return db
}
func TestEnqueueClaimLeaseRecoveryBackoffAndIdempotentSuccess(t *testing.T) {
db := testDB(t)
now := time.Date(2026, 8, 28, 8, 0, 0, 0, time.UTC)
message, err := Enqueue(db, EnqueueInput{InternalType: "local_event_candidate", BusinessRef: "event-1", IdempotencyKey: "local-event:event-1:v1", PayloadJSON: []byte(`{"event":"event-1"}`), MaxAttempts: 3}, now)
if err != nil {
t.Fatal(err)
}
if _, err = Enqueue(db, EnqueueInput{InternalType: "local_event_candidate", BusinessRef: "event-1", IdempotencyKey: message.IdempotencyKey, PayloadJSON: []byte(`{}`)}, now); !errors.Is(err, ErrDuplicateIdempotency) {
t.Fatalf("expected duplicate error, got %v", err)
}
relay := NewRelay(db)
relay.Now = func() time.Time { return now }
relay.LeaseDuration = 10 * time.Second
relay.Backoff = func(int) time.Duration { return 5 * time.Second }
first, err := relay.Claim("worker-1", 1)
if err != nil || len(first) != 1 {
t.Fatalf("first claim=%#v err=%v", first, err)
}
second, err := relay.Claim("worker-2", 1)
if err != nil || len(second) != 0 {
t.Fatalf("concurrent claim=%#v err=%v", second, err)
}
now = now.Add(11 * time.Second)
recovered, err := relay.Claim("worker-2", 1)
if err != nil || len(recovered) != 1 || recovered[0].LeaseOwner != "worker-2" {
t.Fatalf("recovered=%#v err=%v", recovered, err)
}
if err = relay.MarkFailure(message.ID, "worker-2", "temporary outage"); err != nil {
t.Fatal(err)
}
now = now.Add(4 * time.Second)
waiting, _ := relay.Claim("worker-3", 1)
if len(waiting) != 0 {
t.Fatalf("claimed before backoff elapsed: %#v", waiting)
}
now = now.Add(2 * time.Second)
retry, err := relay.Claim("worker-3", 1)
if err != nil || len(retry) != 1 {
t.Fatalf("retry=%#v err=%v", retry, err)
}
if err = relay.MarkSuccess(message.ID, "worker-3"); err != nil {
t.Fatal(err)
}
if err = relay.MarkSuccess(message.ID, "worker-3"); err != nil {
t.Fatalf("idempotent success failed: %v", err)
}
var deliveries int64
db.Model(&DeliveryRecord{}).Where("idempotency_key = ?", message.IdempotencyKey).Count(&deliveries)
if deliveries != 1 {
t.Fatalf("deliveries=%d", deliveries)
}
}
func TestFailureBecomesDeadAndManualRequeuePreservesHistory(t *testing.T) {
db := testDB(t)
now := time.Date(2026, 8, 28, 9, 0, 0, 0, time.UTC)
message, err := Enqueue(db, EnqueueInput{InternalType: "audit_projection", BusinessRef: "audit-1", IdempotencyKey: "audit:audit-1:v1", PayloadJSON: []byte(`{}`), MaxAttempts: 1}, now)
if err != nil {
t.Fatal(err)
}
relay := NewRelay(db)
relay.Now = func() time.Time { return now }
claimed, _ := relay.Claim("worker", 1)
if len(claimed) != 1 {
t.Fatal("message not claimed")
}
if err = relay.MarkFailure(message.ID, "worker", "permanent failure"); err != nil {
t.Fatal(err)
}
service := NewService(db)
service.Now = func() time.Time { return now.Add(time.Minute) }
detail, err := service.Get(message.ID)
if err != nil || detail.Message.State != StateDead || len(detail.Attempts) != 1 {
t.Fatalf("detail=%#v err=%v", detail, err)
}
requeued, err := service.Requeue(message.ID, RequeueRequest{ExpectedVersion: detail.Message.Version, Reason: "出口故障已排除"}, 7)
if err != nil {
t.Fatal(err)
}
if requeued.State != StatePending {
t.Fatalf("state=%s", requeued.State)
}
detail, _ = service.Get(message.ID)
if len(detail.Attempts) != 2 || detail.Attempts[0].Outcome != "manual_requeue" || detail.Attempts[0].ActorUserID != 7 {
t.Fatalf("attempt history=%#v", detail.Attempts)
}
}
func TestProductionCannotCreateTestSink(t *testing.T) {
if _, err := NewTestSink("prod"); !errors.Is(err, ErrTestSinkForbidden) {
t.Fatalf("err=%v", err)
}
if _, err := NewTestSink("test"); err != nil {
t.Fatal(err)
}
}
+124
View File
@@ -0,0 +1,124 @@
package outbox
import (
"errors"
"fmt"
"strings"
"time"
"unicode/utf8"
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
var (
ErrInvalidInput = errors.New("可靠投递请求不符合要求")
ErrNotFound = errors.New("可靠投递记录不存在")
ErrNotDead = errors.New("仅死信记录可以重新排队")
ErrVersionConflict = errors.New("记录已变化,请刷新后重试")
)
type Service struct {
coreService.Service
Now func() time.Time
}
func NewService(db *gorm.DB) *Service {
return &Service{Service: coreService.Service{Orm: db}, Now: time.Now}
}
func (s *Service) List(request PageRequest) (PageResponse, error) {
request.State = strings.TrimSpace(request.State)
request.InternalType = strings.TrimSpace(request.InternalType)
request.Keyword = strings.TrimSpace(request.Keyword)
if request.GetPageSize() > 100 || utf8.RuneCountInString(request.Keyword) > 128 || len(request.InternalType) > 64 || (request.State != "" && !validState(request.State)) {
return PageResponse{}, ErrInvalidInput
}
query := s.Orm.Model(&Message{})
if request.State != "" {
query = query.Where("state = ?", request.State)
}
if request.InternalType != "" {
query = query.Where("internal_type = ?", request.InternalType)
}
if request.Keyword != "" {
pattern := "%" + strings.ToLower(request.Keyword) + "%"
query = query.Where("LOWER(id) LIKE ? OR LOWER(idempotency_key) LIKE ? OR LOWER(business_ref) LIKE ?", pattern, pattern, pattern)
}
var response PageResponse
if err := query.Count(&response.Count).Error; err != nil {
return PageResponse{}, fmt.Errorf("count outbox messages: %w", err)
}
if err := query.Order("created_at DESC, id DESC").Limit(request.GetPageSize()).Offset((request.GetPageIndex() - 1) * request.GetPageSize()).Find(&response.List).Error; err != nil {
return PageResponse{}, fmt.Errorf("list outbox messages: %w", err)
}
for state, target := range map[string]*int64{StatePending: &response.Summary.Pending, StateRetry: &response.Summary.Retry, StateProcessing: &response.Summary.Processing, StateDead: &response.Summary.Dead} {
if err := s.Orm.Model(&Message{}).Where("state = ?", state).Count(target).Error; err != nil {
return PageResponse{}, fmt.Errorf("summarize outbox: %w", err)
}
}
return response, nil
}
func (s *Service) Get(id string) (DetailResponse, error) {
id = strings.TrimSpace(id)
if id == "" || len(id) > 64 {
return DetailResponse{}, ErrInvalidInput
}
var message Message
if err := s.Orm.First(&message, "id = ?", id).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return DetailResponse{}, ErrNotFound
}
return DetailResponse{}, fmt.Errorf("get outbox message: %w", err)
}
var attempts []Attempt
if err := s.Orm.Where("message_id = ?", id).Order("created_at DESC, id DESC").Limit(50).Find(&attempts).Error; err != nil {
return DetailResponse{}, fmt.Errorf("list outbox attempts: %w", err)
}
return DetailResponse{Message: message, Attempts: attempts}, nil
}
func (s *Service) Requeue(id string, request RequeueRequest, actorUserID int) (Message, error) {
reason := strings.TrimSpace(request.Reason)
if utf8.RuneCountInString(reason) < 6 || utf8.RuneCountInString(reason) > 256 || request.ExpectedVersion < 1 {
return Message{}, ErrInvalidInput
}
now := s.now()
var result Message
err := s.Orm.Transaction(func(tx *gorm.DB) error {
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&result, "id = ?", strings.TrimSpace(id)).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return ErrNotFound
}
return err
}
if result.Version != request.ExpectedVersion {
return ErrVersionConflict
}
if result.State != StateDead {
return ErrNotDead
}
result.State, result.AvailableAt, result.LastError = StatePending, now, ""
result.LeaseOwner, result.LeaseUntil, result.Version = "", nil, result.Version+1
if err := tx.Save(&result).Error; err != nil {
return err
}
return tx.Create(&Attempt{MessageID: result.ID, Number: result.AttemptCount, Outcome: "manual_requeue", Detail: reason, ActorUserID: actorUserID, CreatedAt: now}).Error
})
if err != nil {
return Message{}, err
}
return result, nil
}
func (s *Service) now() time.Time {
if s.Now != nil {
return s.Now().UTC()
}
return time.Now().UTC()
}
func validState(value string) bool {
return value == StatePending || value == StateProcessing || value == StateRetry || value == StateDead || value == StateDelivered
}
+24
View File
@@ -0,0 +1,24 @@
package outbox
import (
"errors"
"strings"
)
var ErrTestSinkForbidden = errors.New("production 模式禁止启用测试接收器")
type Sink interface{ Deliver(Message) error }
type TestSink struct{ Delivered []string }
func (s *TestSink) Deliver(message Message) error {
s.Delivered = append(s.Delivered, message.IdempotencyKey)
return nil
}
func NewTestSink(applicationMode string) (*TestSink, error) {
if strings.EqualFold(strings.TrimSpace(applicationMode), "prod") || strings.EqualFold(strings.TrimSpace(applicationMode), "production") {
return nil, ErrTestSinkForbidden
}
return &TestSink{}, nil
}
@@ -0,0 +1,53 @@
package version
import (
"fmt"
"runtime"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/edge_node"
"git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration"
migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
)
func init() {
_, fileName, _, _ := runtime.Caller(0)
migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateSenseEdgeNode)
}
func migrateSenseEdgeNode(db *gorm.DB, version string) error {
return db.Transaction(func(tx *gorm.DB) error {
if err := tx.AutoMigrate(&edge_node.Node{}, &edge_node.Event{}); err != nil {
return err
}
root, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: senseLayoutMenuName, Title: "视频感知", Icon: "video-camera", Path: "/sense", MenuType: "M", Component: "Layout", Sort: 5, Visible: "0", IsFrame: "1"})
if err != nil {
return err
}
page, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseEdgeNode", Title: "边缘节点", Icon: "monitor", Path: "edge-node", Paths: fmt.Sprintf("/0/%d", root.MenuId), MenuType: "C", Permission: "sense:edge-node:list", ParentId: root.MenuId, Component: "/sense/edge-node/index", Sort: 9, Visible: "0", IsFrame: "1"})
if err != nil {
return err
}
detail, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseEdgeNodeDetail", Title: "查看节点详情", MenuType: "F", Action: "GET", Permission: "sense:edge-node:detail", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 1, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
if err = attachDeviceRole(tx, role, []migrationModels.SysMenu{page, detail}); err != nil {
return err
}
for _, policy := range [][2]string{{"/api/v1/edge-nodes", "GET"}, {"/api/v1/edge-nodes/:id", "GET"}} {
if err = tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&deviceCasbinRule{Ptype: "p", V0: role, V1: policy[0], V2: policy[1]}).Error; err != nil {
return err
}
}
}
if err = rebuildSenseMenuPaths(tx, root.MenuId, "/0"); err != nil {
return err
}
return tx.Create(&common.Migration{Version: version}).Error
})
}
@@ -0,0 +1,90 @@
package version
import (
"os"
"testing"
"gorm.io/driver/postgres"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/edge_node"
migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
)
func TestEdgeNodeMigrationAddsTablesAndReadOnlyPermissions(t *testing.T) {
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&migrationModels.SysRole{}, &migrationModels.SysMenu{}, &deviceCasbinRule{}, &common.Migration{}); err != nil {
t.Fatal(err)
}
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
if err = db.Create(&migrationModels.SysRole{RoleName: role, RoleKey: role, Status: "2"}).Error; err != nil {
t.Fatal(err)
}
}
const version = "2026082813000_edge_node.go"
if err = migrateSenseEdgeNode(db, version); err != nil {
t.Fatal(err)
}
if !db.Migrator().HasTable(&edge_node.Node{}) || !db.Migrator().HasTable(&edge_node.Event{}) {
t.Fatal("edge node tables missing")
}
var menus, reads, writes, applied int64
db.Model(&migrationModels.SysMenu{}).Where("menu_name LIKE ?", "SenseEdgeNode%").Count(&menus)
db.Model(&deviceCasbinRule{}).Where("v1 LIKE ? AND v2 = ?", "/api/v1/edge-nodes%", "GET").Count(&reads)
db.Model(&deviceCasbinRule{}).Where("v1 LIKE ? AND v2 <> ?", "/api/v1/edge-nodes%", "GET").Count(&writes)
db.Model(&common.Migration{}).Where("version = ?", version).Count(&applied)
if menus != 2 || reads != 6 || writes != 0 || applied != 1 {
t.Fatalf("menus=%d reads=%d writes=%d applied=%d", menus, reads, writes, applied)
}
var fixtures int64
db.Model(&edge_node.Node{}).Count(&fixtures)
if fixtures != 0 {
t.Fatalf("migration seeded %d fixtures", fixtures)
}
}
func TestEdgeNodeMigrationOnPostgres(t *testing.T) {
dsn := os.Getenv("SENSE_EDGE_NODE_MIGRATION_TEST_DATABASE_URL")
if dsn == "" {
t.Skip("set SENSE_EDGE_NODE_MIGRATION_TEST_DATABASE_URL to run the PostgreSQL migration test")
}
db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
const schema = "sense_edge_node_76_test"
if err = db.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE").Error; err != nil {
t.Fatal(err)
}
if err = db.Exec("CREATE SCHEMA " + schema).Error; err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = db.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE").Error })
sqlDB, err := db.DB()
if err != nil {
t.Fatal(err)
}
sqlDB.SetMaxOpenConns(1)
if err = db.Exec("SET search_path TO " + schema).Error; err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&migrationModels.SysRole{}, &migrationModels.SysMenu{}, &deviceCasbinRule{}, &common.Migration{}); err != nil {
t.Fatal(err)
}
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
if err = db.Create(&migrationModels.SysRole{RoleName: role, RoleKey: role, Status: "2"}).Error; err != nil {
t.Fatal(err)
}
}
if err = migrateSenseEdgeNode(db, "2026082813000_edge_node.go"); err != nil {
t.Fatal(err)
}
if !db.Migrator().HasTable(&edge_node.Node{}) || !db.Migrator().HasTable(&edge_node.Event{}) {
t.Fatal("edge node tables missing")
}
}
@@ -0,0 +1,57 @@
package version
import (
"fmt"
"runtime"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
"git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration"
migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
)
func init() {
_, fileName, _, _ := runtime.Caller(0)
migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateSenseMediaShard)
}
func migrateSenseMediaShard(db *gorm.DB, version string) error {
return db.Transaction(func(tx *gorm.DB) error {
if err := tx.AutoMigrate(&media_shard.Shard{}, &media_shard.Assignment{}); err != nil {
return err
}
root, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: senseLayoutMenuName, Title: "视频感知", Icon: "video-camera", Path: "/sense", MenuType: "M", Component: "Layout", Sort: 5, Visible: "0", IsFrame: "1"})
if err != nil {
return err
}
page, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseMediaShard", Title: "媒体分片", Icon: "connection", Path: "media-shard", Paths: fmt.Sprintf("/0/%d", root.MenuId), MenuType: "C", Permission: "sense:media-shard:list", ParentId: root.MenuId, Component: "/sense/media-shard/index", Sort: 10, Visible: "0", IsFrame: "1"})
if err != nil {
return err
}
detail, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseMediaShardDetail", Title: "查看分片详情", MenuType: "F", Action: "GET", Permission: "sense:media-shard:detail", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 1, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
preflight, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseMediaShardPreflight", Title: "迁移预检", MenuType: "F", Action: "GET", Permission: "sense:media-shard:preflight", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 2, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
if err = attachDeviceRole(tx, role, []migrationModels.SysMenu{page, detail, preflight}); err != nil {
return err
}
for _, policy := range [][2]string{{"/api/v1/media-shards", "GET"}, {"/api/v1/media-shards/:id", "GET"}, {"/api/v1/media-shards/:id/migration-preflight", "GET"}} {
if err = tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&deviceCasbinRule{Ptype: "p", V0: role, V1: policy[0], V2: policy[1]}).Error; err != nil {
return err
}
}
}
if err = rebuildSenseMenuPaths(tx, root.MenuId, "/0"); err != nil {
return err
}
return tx.Create(&common.Migration{Version: version}).Error
})
}
@@ -0,0 +1,86 @@
package version
import (
"os"
"testing"
"gorm.io/driver/postgres"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
)
func TestMediaShardMigrationAddsReadOnlySurfaceWithoutFixtures(t *testing.T) {
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&migrationModels.SysRole{}, &migrationModels.SysMenu{}, &deviceCasbinRule{}, &common.Migration{}); err != nil {
t.Fatal(err)
}
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
if err = db.Create(&migrationModels.SysRole{RoleName: role, RoleKey: role, Status: "2"}).Error; err != nil {
t.Fatal(err)
}
}
const version = "2026082814000_media_shard.go"
if err = migrateSenseMediaShard(db, version); err != nil {
t.Fatal(err)
}
if !db.Migrator().HasTable(&media_shard.Shard{}) || !db.Migrator().HasTable(&media_shard.Assignment{}) {
t.Fatal("media shard tables missing")
}
var menus, reads, writes, fixtures, applied int64
db.Model(&migrationModels.SysMenu{}).Where("menu_name LIKE ?", "SenseMediaShard%").Count(&menus)
db.Model(&deviceCasbinRule{}).Where("v1 LIKE ? AND v2 = ?", "/api/v1/media-shards%", "GET").Count(&reads)
db.Model(&deviceCasbinRule{}).Where("v1 LIKE ? AND v2 <> ?", "/api/v1/media-shards%", "GET").Count(&writes)
db.Model(&media_shard.Shard{}).Count(&fixtures)
db.Model(&common.Migration{}).Where("version = ?", version).Count(&applied)
if menus != 3 || reads != 9 || writes != 0 || fixtures != 0 || applied != 1 {
t.Fatalf("menus=%d reads=%d writes=%d fixtures=%d applied=%d", menus, reads, writes, fixtures, applied)
}
}
func TestMediaShardMigrationOnPostgres(t *testing.T) {
dsn := os.Getenv("SENSE_MEDIA_SHARD_MIGRATION_TEST_DATABASE_URL")
if dsn == "" {
t.Skip("set SENSE_MEDIA_SHARD_MIGRATION_TEST_DATABASE_URL to run the PostgreSQL migration test")
}
db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
const schema = "sense_media_shard_77_test"
if err = db.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE").Error; err != nil {
t.Fatal(err)
}
if err = db.Exec("CREATE SCHEMA " + schema).Error; err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = db.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE").Error })
sqlDB, err := db.DB()
if err != nil {
t.Fatal(err)
}
sqlDB.SetMaxOpenConns(1)
if err = db.Exec("SET search_path TO " + schema).Error; err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&migrationModels.SysRole{}, &migrationModels.SysMenu{}, &deviceCasbinRule{}, &common.Migration{}); err != nil {
t.Fatal(err)
}
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
if err = db.Create(&migrationModels.SysRole{RoleName: role, RoleKey: role, Status: "2"}).Error; err != nil {
t.Fatal(err)
}
}
if err = migrateSenseMediaShard(db, "2026082814000_media_shard.go"); err != nil {
t.Fatal(err)
}
if !db.Migrator().HasTable(&media_shard.Shard{}) || !db.Migrator().HasTable(&media_shard.Assignment{}) {
t.Fatal("media shard tables missing")
}
}
@@ -0,0 +1,65 @@
package version
import (
"fmt"
"runtime"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
"git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration"
migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
)
func init() {
_, fileName, _, _ := runtime.Caller(0)
migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateSenseOutbox)
}
func migrateSenseOutbox(db *gorm.DB, version string) error {
return db.Transaction(func(tx *gorm.DB) error {
if err := tx.AutoMigrate(&outbox.Message{}, &outbox.DeliveryRecord{}, &outbox.Attempt{}); err != nil {
return err
}
root, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: senseLayoutMenuName, Title: "视频感知", Icon: "video-camera", Path: "/sense", MenuType: "M", Component: "Layout", Sort: 5, Visible: "0", IsFrame: "1"})
if err != nil {
return err
}
page, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOutbox", Title: "可靠投递", Icon: "connection", Path: "outbox", Paths: fmt.Sprintf("/0/%d", root.MenuId), MenuType: "C", Permission: "sense:outbox:list", ParentId: root.MenuId, Component: "/sense/outbox/index", Sort: 11, Visible: "0", IsFrame: "1"})
if err != nil {
return err
}
detail, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOutboxDetail", Title: "查看投递详情", MenuType: "F", Action: "GET", Permission: "sense:outbox:detail", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 1, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
requeue, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOutboxRequeue", Title: "死信重新排队", MenuType: "F", Action: "POST", Permission: "sense:outbox:requeue", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 2, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
if err = attachDeviceRole(tx, role, []migrationModels.SysMenu{page, detail}); err != nil {
return err
}
for _, policy := range [][2]string{{"/api/v1/outbox", "GET"}, {"/api/v1/outbox/:id", "GET"}} {
if err = tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&deviceCasbinRule{Ptype: "p", V0: role, V1: policy[0], V2: policy[1]}).Error; err != nil {
return err
}
}
}
for _, role := range []string{"implementation_operator", "site_admin"} {
if err = attachDeviceRole(tx, role, []migrationModels.SysMenu{requeue}); err != nil {
return err
}
if err = tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&deviceCasbinRule{Ptype: "p", V0: role, V1: "/api/v1/outbox/:id/requeue", V2: "POST"}).Error; err != nil {
return err
}
}
if err = rebuildSenseMenuPaths(tx, root.MenuId, "/0"); err != nil {
return err
}
return tx.Create(&common.Migration{Version: version}).Error
})
}
@@ -0,0 +1,41 @@
package version
import (
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"testing"
)
func TestOutboxMigrationAddsRBACWithoutFixtures(t *testing.T) {
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&migrationModels.SysRole{}, &migrationModels.SysMenu{}, &deviceCasbinRule{}, &common.Migration{}); err != nil {
t.Fatal(err)
}
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
if err = db.Create(&migrationModels.SysRole{RoleName: role, RoleKey: role, Status: "2"}).Error; err != nil {
t.Fatal(err)
}
}
const version = "2026082815000_outbox.go"
if err = migrateSenseOutbox(db, version); err != nil {
t.Fatal(err)
}
if !db.Migrator().HasTable(&outbox.Message{}) || !db.Migrator().HasTable(&outbox.DeliveryRecord{}) || !db.Migrator().HasTable(&outbox.Attempt{}) {
t.Fatal("outbox tables missing")
}
var menus, reads, writes, fixtures, applied int64
db.Model(&migrationModels.SysMenu{}).Where("menu_name LIKE ?", "SenseOutbox%").Count(&menus)
db.Model(&deviceCasbinRule{}).Where("v1 LIKE ? AND v2 = ?", "/api/v1/outbox%", "GET").Count(&reads)
db.Model(&deviceCasbinRule{}).Where("v1 = ? AND v2 = ?", "/api/v1/outbox/:id/requeue", "POST").Count(&writes)
db.Model(&outbox.Message{}).Count(&fixtures)
db.Model(&common.Migration{}).Where("version = ?", version).Count(&applied)
if menus != 3 || reads != 6 || writes != 2 || fixtures != 0 || applied != 1 {
t.Fatalf("menus=%d reads=%d writes=%d fixtures=%d applied=%d", menus, reads, writes, fixtures, applied)
}
}
@@ -0,0 +1,74 @@
package version
import (
"fmt"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/ops_alert"
"git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration"
migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"runtime"
)
func init() {
_, fileName, _, _ := runtime.Caller(0)
migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateSenseOpsAlert)
}
func migrateSenseOpsAlert(db *gorm.DB, version string) error {
return db.Transaction(func(tx *gorm.DB) error {
if err := tx.AutoMigrate(&ops_alert.Alert{}, &ops_alert.Transition{}); err != nil {
return err
}
root, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: senseLayoutMenuName, Title: "视频感知", Icon: "video-camera", Path: "/sense", MenuType: "M", Component: "Layout", Sort: 5, Visible: "0", IsFrame: "1"})
if err != nil {
return err
}
page, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOpsAlert", Title: "运维告警", Icon: "warning", Path: "ops-alert", Paths: fmt.Sprintf("/0/%d", root.MenuId), MenuType: "C", Permission: "sense:ops-alert:list", ParentId: root.MenuId, Component: "/sense/ops-alert/index", Sort: 11, Visible: "0", IsFrame: "1"})
if err != nil {
return err
}
detail, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOpsAlertDetail", Title: "查看告警详情", MenuType: "F", Action: "GET", Permission: "sense:ops-alert:detail", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 1, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
refresh, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOpsAlertEvaluate", Title: "刷新告警状态", MenuType: "F", Action: "POST", Permission: "sense:ops-alert:evaluate", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 2, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
ack, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOpsAlertAcknowledge", Title: "确认运维告警", MenuType: "F", Action: "POST", Permission: "sense:ops-alert:acknowledge", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 3, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
recover, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOpsAlertRecover", Title: "确认告警恢复", MenuType: "F", Action: "POST", Permission: "sense:ops-alert:recover", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 4, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
readPolicies := [][2]string{{"/api/v1/ops-alerts", "GET"}, {"/api/v1/ops-alerts/:id", "GET"}}
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
if err = attachDeviceRole(tx, role, []migrationModels.SysMenu{page, detail}); err != nil {
return err
}
for _, policy := range readPolicies {
if err = tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&deviceCasbinRule{Ptype: "p", V0: role, V1: policy[0], V2: policy[1]}).Error; err != nil {
return err
}
}
}
for _, role := range []string{"implementation_operator", "site_admin"} {
if err = attachDeviceRole(tx, role, []migrationModels.SysMenu{refresh, ack, recover}); err != nil {
return err
}
for _, policy := range [][2]string{{"/api/v1/ops-alerts/evaluate", "POST"}, {"/api/v1/ops-alerts/:id/acknowledge", "POST"}, {"/api/v1/ops-alerts/:id/recover", "POST"}} {
if err = tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&deviceCasbinRule{Ptype: "p", V0: role, V1: policy[0], V2: policy[1]}).Error; err != nil {
return err
}
}
}
if err = rebuildSenseMenuPaths(tx, root.MenuId, "/0"); err != nil {
return err
}
return tx.Create(&common.Migration{Version: version}).Error
})
}
@@ -0,0 +1,39 @@
package version
import (
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/ops_alert"
migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"testing"
)
func TestSenseOpsAlertMigrationRBAC(t *testing.T) {
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&migrationModels.SysRole{}, &migrationModels.SysMenu{}, &deviceCasbinRule{}, &common.Migration{}); err != nil {
t.Fatal(err)
}
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
if err = db.Create(&migrationModels.SysRole{RoleName: role, RoleKey: role, Status: "2"}).Error; err != nil {
t.Fatal(err)
}
}
const version = "2026082816000_ops_alert.go"
if err = migrateSenseOpsAlert(db, version); err != nil {
t.Fatal(err)
}
var pages, viewerWrites, operatorWrites int64
db.Model(&migrationModels.SysMenu{}).Where("permission = ?", "sense:ops-alert:list").Count(&pages)
db.Model(&deviceCasbinRule{}).Where("v0 = ? AND v1 LIKE ? AND v2 = ?", "viewer", "/api/v1/ops-alerts%", "POST").Count(&viewerWrites)
db.Model(&deviceCasbinRule{}).Where("v0 = ? AND v1 LIKE ? AND v2 = ?", "implementation_operator", "/api/v1/ops-alerts%", "POST").Count(&operatorWrites)
if pages != 1 || viewerWrites != 0 || operatorWrites != 3 {
t.Fatalf("unexpected RBAC page=%d viewerWrites=%d operatorWrites=%d", pages, viewerWrites, operatorWrites)
}
if !db.Migrator().HasTable(&ops_alert.Alert{}) || !db.Migrator().HasTable(&ops_alert.Transition{}) {
t.Fatal("ops alert tables missing")
}
}
@@ -9,3 +9,5 @@ SENSE_ONVIF_ALLOWED_CIDRS=
SENSE_MEDIAMTX_BINARY=
SENSE_MEDIAMTX_CONFIG=
SENSE_MEDIAMTX_API=http://127.0.0.1:9997
SENSE_MEDIAMTX_CAPACITY=
SENSE_MEDIAMTX_SHARDS_FILE=
+79 -1
View File
@@ -10,6 +10,9 @@ import (
"time"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
func freeAddress(t *testing.T) string {
@@ -25,6 +28,81 @@ func freeAddress(t *testing.T) string {
return address
}
func startTestMediaMTX(t *testing.T, binary string) string {
t.Helper()
apiAddress, rtspAddress := freeAddress(t), freeAddress(t)
configPath := filepath.Join(t.TempDir(), "mediamtx.yml")
config := fmt.Sprintf("logLevel: warn\napi: true\napiAddress: %s\nrtspAddress: %s\nrtspTransports: [tcp]\nrtmp: false\nhls: false\nwebrtc: false\nsrt: false\nmoq: false\nplayback: false\npaths: {}\n", apiAddress, rtspAddress)
if err := os.WriteFile(configPath, []byte(config), 0o600); err != nil {
t.Fatal(err)
}
supervisor := media.NewSupervisor(binary, configPath)
if err := supervisor.Start(context.Background()); err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_ = supervisor.Stop(ctx)
})
controller, err := media.NewHTTPController("http://" + apiAddress)
if err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(8 * time.Second)
for {
if err = controller.Health(context.Background()); err == nil {
break
}
if time.Now().After(deadline) {
t.Fatalf("MediaMTX did not become ready: %v", err)
}
time.Sleep(100 * time.Millisecond)
}
return "http://" + apiAddress
}
func TestTwoRealMediaMTXShardsAreIndependentlyUsable(t *testing.T) {
binary := os.Getenv("SENSE_MEDIAMTX_TEST_BINARY")
if binary == "" {
t.Skip("set SENSE_MEDIAMTX_TEST_BINARY to run the real MediaMTX integration")
}
first, second := startTestMediaMTX(t, binary), startTestMediaMTX(t, binary)
db, err := gorm.Open(sqlite.Open("file:real_media_shards?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&media_shard.Shard{}, &media_shard.Assignment{}); err != nil {
t.Fatal(err)
}
service := media_shard.NewService(db, func(ctx context.Context, endpoint string) error {
controller, buildErr := media.NewHTTPController(endpoint)
if buildErr != nil {
return buildErr
}
return controller.Health(ctx)
})
specs := []media_shard.Spec{{ID: "one", Name: "测试分片一", Mode: "external", ControlAPI: first, Capacity: 2}, {ID: "two", Name: "测试分片二", Mode: "external", ControlAPI: second, Capacity: 2}}
if err = service.SyncSpecs(context.Background(), specs); err != nil {
t.Fatal(err)
}
if err = service.RefreshAll(context.Background()); err != nil {
t.Fatal(err)
}
for _, routeID := range []string{"route-a", "route-b", "route-c", "route-d"} {
if _, err = service.EnsureAssignment(context.Background(), routeID); err != nil {
t.Fatalf("assign %s: %v", routeID, err)
}
}
result, err := service.List(context.Background())
if err != nil {
t.Fatal(err)
}
if result.Summary.Available != 2 || result.Summary.Assigned != 4 || result.Summary.Configured != 4 {
t.Fatalf("unexpected real shard summary: %+v", result.Summary)
}
}
func TestRealMediaMTXControlLifecycle(t *testing.T) {
binary := os.Getenv("SENSE_MEDIAMTX_TEST_BINARY")
if binary == "" {
@@ -32,7 +110,7 @@ func TestRealMediaMTXControlLifecycle(t *testing.T) {
}
apiAddress, rtspAddress := freeAddress(t), freeAddress(t)
configPath := filepath.Join(t.TempDir(), "mediamtx.yml")
config := fmt.Sprintf("logLevel: warn\napi: true\napiAddress: %s\nrtspAddress: %s\nrtmp: false\nhls: false\nwebrtc: false\nsrt: false\nplayback: false\npaths: {}\n", apiAddress, rtspAddress)
config := fmt.Sprintf("logLevel: warn\napi: true\napiAddress: %s\nrtspAddress: %s\nrtspTransports: [tcp]\nrtmp: false\nhls: false\nwebrtc: false\nsrt: false\nmoq: false\nplayback: false\npaths: {}\n", apiAddress, rtspAddress)
if err := os.WriteFile(configPath, []byte(config), 0o600); err != nil {
t.Fatal(err)
}
+3
View File
@@ -34,6 +34,7 @@ try {
[IO.File]::WriteAllText((Join-Path $temporary 'web\js\runtime.fixture.js'), 'fixture', (New-Object Text.UTF8Encoding($false)))
[IO.File]::WriteAllText((Join-Path $temporary 'bin\mediamtx.exe'), 'fixture', (New-Object Text.UTF8Encoding($false)))
[IO.File]::WriteAllText((Join-Path $temporary 'config\mediamtx.yml'), 'api: true', (New-Object Text.UTF8Encoding($false)))
[IO.File]::WriteAllText((Join-Path $temporary 'config\mediamtx-shards.json'), '[]', (New-Object Text.UTF8Encoding($false)))
$webAssetAudit = Join-Path $senseRoot 'scripts\build\assert-web-assets.ps1'
& $webAssetAudit -WebRoot (Join-Path $temporary 'web')
@@ -55,6 +56,7 @@ try {
'SENSE_MEDIAMTX_BINARY=bin\mediamtx.exe',
'SENSE_MEDIAMTX_CONFIG=config\mediamtx.yml',
'SENSE_MEDIAMTX_API=http://127.0.0.1:9997',
'SENSE_MEDIAMTX_SHARDS_FILE=config\mediamtx-shards.json',
'SENSE_WEB_ROOT=web'
) -join "`n"
[IO.File]::WriteAllText((Join-Path $temporary 'config\sense.env'), $envText, (New-Object Text.UTF8Encoding($false)))
@@ -68,6 +70,7 @@ try {
Assert-True $generatedSettings.Contains('timeout: 2592000') 'generated runtime settings must keep login valid for 30 days'
Assert-True ((Get-SenseEnvironmentValue -Name 'SENSE_DATABASE_URL').Contains('p#&;=x')) 'special characters must survive env parsing'
Assert-True ($generatedSettings.Contains('password=p#\u0026;=x')) 'special characters must be safely JSON-escaped in YAML'
Assert-Equal (Join-Path $temporary 'config\mediamtx-shards.json') (Get-SenseEnvironmentValue -Name 'SENSE_MEDIAMTX_SHARDS_FILE') 'relative shard config path must resolve inside package root'
Assert-True (-not ((Get-SenseDatabaseInfo (Get-SenseEnvironmentValue -Name 'SENSE_DATABASE_URL')).Sanitized.Contains('password='))) 'PostgreSQL tool arguments must not contain password'
[Environment]::SetEnvironmentVariable('SENSE_PORT', $null, 'Process')
+9
View File
@@ -0,0 +1,9 @@
import request from '@/utils/request'
export function listEdgeNodes(query) {
return request({ url: '/api/v1/edge-nodes', method: 'get', params: query })
}
export function getEdgeNode(id) {
return request({ url: `/api/v1/edge-nodes/${encodeURIComponent(id)}`, method: 'get' })
}
+13
View File
@@ -0,0 +1,13 @@
import request from '@/utils/request'
export function listMediaShards() {
return request({ url: '/api/v1/media-shards', method: 'get' })
}
export function getMediaShard(id) {
return request({ url: `/api/v1/media-shards/${encodeURIComponent(id)}`, method: 'get' })
}
export function preflightMediaShard(id) {
return request({ url: `/api/v1/media-shards/${encodeURIComponent(id)}/migration-preflight`, method: 'get' })
}
+21
View File
@@ -0,0 +1,21 @@
import request from '@/utils/request'
export function listOpsAlerts(query) {
return request({ url: '/api/v1/ops-alerts', method: 'get', params: query })
}
export function getOpsAlert(id) {
return request({ url: `/api/v1/ops-alerts/${encodeURIComponent(id)}`, method: 'get' })
}
export function evaluateOpsAlerts() {
return request({ url: '/api/v1/ops-alerts/evaluate', method: 'post' })
}
export function acknowledgeOpsAlert(id, data) {
return request({ url: `/api/v1/ops-alerts/${encodeURIComponent(id)}/acknowledge`, method: 'post', data })
}
export function recoverOpsAlert(id, data) {
return request({ url: `/api/v1/ops-alerts/${encodeURIComponent(id)}/recover`, method: 'post', data })
}
+20
View File
@@ -0,0 +1,20 @@
import request from '@/utils/request'
export function listOutbox(query) {
return request({ url: '/api/v1/outbox', method: 'get', params: query })
}
export function getOutbox(id) {
return request({
url: `/api/v1/outbox/${encodeURIComponent(id)}`,
method: 'get'
})
}
export function requeueOutbox(id, expectedVersion, reason) {
return request({
url: `/api/v1/outbox/${encodeURIComponent(id)}/requeue`,
method: 'post',
data: { expectedVersion, reason }
})
}
@@ -0,0 +1,36 @@
const stateLabels = {
online: '在线',
offline: '离线',
recovering: '恢复中',
ready: '正常',
unavailable: '不可用',
idle: '无待回填',
pending: '等待回填'
}
const eventLabels = {
heartbeat_received: '首次心跳',
heartbeat_timeout: '心跳超时',
node_recovered: '心跳恢复',
recovery_started: '开始恢复'
}
export function stateLabel(value) { return stateLabels[value] || value || '未知' }
export function eventLabel(value) { return eventLabels[value] || value || '状态变化' }
export function stateType(value) { return ({ online: 'success', offline: 'danger', recovering: 'warning', ready: 'success', unavailable: 'danger', idle: 'info', pending: 'warning' })[value] || 'info' }
export function freshnessText(item) {
if (!item?.lastHeartbeatAt) return '从未收到心跳'
const seconds = Math.max(0, Number(item.freshnessSeconds) || 0)
if (item.stale) return `已陈旧 ${durationText(seconds)}`
return `${durationText(seconds)}前`
}
export function durationText(seconds) {
const value = Math.max(0, Number(seconds) || 0)
if (value < 60) return `${Math.floor(value)} 秒`
if (value < 3600) return `${Math.floor(value / 60)} 分钟`
if (value < 86400) return `${Math.floor(value / 3600)} 小时`
return `${Math.floor(value / 86400)} 天`
}
export function buildEdgeNodeQuery(query) {
return { pageIndex: Number(query.pageIndex) || 1, pageSize: Number(query.pageSize) || 10 }
}
@@ -0,0 +1,88 @@
<template>
<BasicLayout>
<template #wrapper>
<el-card class="box-card">
<div class="page-header">
<div><h3>边缘节点</h3><p>查看各现场节点的运行负载、连接状态和待回填数据。离线时保留最后一次已知状态。</p></div>
<el-button :icon="Refresh" :loading="loading" @click="load">刷新状态</el-button>
</div>
<el-alert v-if="summary.offline" type="warning" :closable="false" show-icon class="status-alert">
<template #title>{{ summary.offline }} 个节点离线</template>
离线阈值为 {{ summary.heartbeatTimeoutSeconds }} 秒。页面中的隧道、视频和回填数据是节点离线前的最后已知状态。
</el-alert>
<el-row :gutter="12" class="summary-row" aria-label="边缘节点概览">
<el-col v-for="item in summaryCards" :key="item.label" :xs="12" :sm="6">
<div class="summary-item"><span>{{ item.label }}</span><strong :class="item.className">{{ item.value }}</strong></div>
</el-col>
</el-row>
<el-table v-loading="loading" :data="nodes" border stripe empty-text="暂无边缘节点">
<el-table-column label="节点" min-width="180">
<template #default="scope"><strong>{{ scope.row.name }}</strong><div class="muted">{{ scope.row.id }} · {{ scope.row.location || '未填写位置' }}</div></template>
</el-table-column>
<el-table-column label="状态" width="106" align="center"><template #default="scope"><el-tag :type="stateType(scope.row.status)">{{ stateLabel(scope.row.status) }}</el-tag></template></el-table-column>
<el-table-column label="最近心跳" min-width="150"><template #default="scope"><div>{{ freshnessText(scope.row) }}</div><small>{{ formatTime(scope.row.lastHeartbeatAt) }}</small></template></el-table-column>
<el-table-column label="负载" min-width="112"><template #default="scope">{{ scope.row.loadUsed }} / {{ scope.row.loadCapacity }} 路</template></el-table-column>
<el-table-column label="控制隧道" min-width="142"><template #default="scope"><el-tag size="small" :type="stateType(scope.row.controlTunnelStatus)">{{ stateLabel(scope.row.controlTunnelStatus) }}</el-tag><div class="muted">{{ scope.row.controlLatencyMs ? `${scope.row.controlLatencyMs} ms` : scope.row.controlTunnelDetail }}</div></template></el-table-column>
<el-table-column label="视频数据面" min-width="150"><template #default="scope"><el-tag size="small" :type="stateType(scope.row.videoPlaneStatus)">{{ stateLabel(scope.row.videoPlaneStatus) }}</el-tag><div class="muted">{{ scope.row.activeStreamCount }} 路视频 · {{ scope.row.videoPlaneDetail }}</div></template></el-table-column>
<el-table-column label="回填队列" min-width="128"><template #default="scope"><el-tag size="small" :type="stateType(scope.row.backfillStatus)">{{ stateLabel(scope.row.backfillStatus) }}</el-tag><div class="muted">{{ scope.row.backfillQueueDepth }} 条待回填</div></template></el-table-column>
<el-table-column label="版本" min-width="130"><template #default="scope"><div>{{ scope.row.runtimeVersion || '—' }}</div><small>{{ scope.row.modelVersion || '未部署模型' }}</small></template></el-table-column>
<el-table-column label="操作" width="92" fixed="right"><template #default="scope"><el-button v-permisaction="['sense:edge-node:detail']" type="primary" link @click="openDetail(scope.row.id)">详情</el-button></template></el-table-column>
</el-table>
<pagination v-show="total > 0" v-model:current-page="query.pageIndex" v-model:page-size="query.pageSize" :total="total" @pagination="load" />
</el-card>
<el-dialog v-model="detailOpen" title="边缘节点详情" width="min(800px, calc(100vw - 24px))" :close-on-click-modal="false">
<template v-if="selected">
<el-alert v-if="selected.stale" title="当前显示最后已知状态" type="warning" :closable="false" show-icon class="detail-alert">最近心跳距今 {{ durationText(selected.freshnessSeconds) }};连接恢复前,下列通道数据不会继续更新。</el-alert>
<el-descriptions :column="2" border>
<el-descriptions-item label="节点">{{ selected.name }}({{ selected.id }})</el-descriptions-item><el-descriptions-item label="位置">{{ selected.location || '未填写' }}</el-descriptions-item>
<el-descriptions-item label="运行版本">{{ selected.runtimeVersion || '—' }}</el-descriptions-item><el-descriptions-item label="模型版本">{{ selected.modelVersion || '未部署' }}</el-descriptions-item>
<el-descriptions-item label="运行时长">{{ durationText(selected.uptimeSeconds) }}</el-descriptions-item><el-descriptions-item label="最近心跳">{{ formatTime(selected.lastHeartbeatAt) }}</el-descriptions-item>
<el-descriptions-item label="控制隧道">{{ stateLabel(selected.controlTunnelStatus) }} · {{ selected.controlTunnelDetail }}</el-descriptions-item><el-descriptions-item label="视频数据面">{{ stateLabel(selected.videoPlaneStatus) }} · {{ selected.videoPlaneDetail }}</el-descriptions-item>
<el-descriptions-item label="回填队列">{{ stateLabel(selected.backfillStatus) }} · {{ selected.backfillQueueDepth }} 条</el-descriptions-item><el-descriptions-item label="恢复阶段">{{ recoveryText(selected) }}</el-descriptions-item>
</el-descriptions>
<h4 class="audit-title">状态审计</h4>
<el-timeline v-if="selected.events?.length">
<el-timeline-item v-for="event in selected.events" :key="event.id" :timestamp="formatTime(event.occurredAt)" placement="top"><strong>{{ eventLabel(event.eventType) }}</strong><div>{{ stateLabel(event.fromStatus) }} → {{ stateLabel(event.toStatus) }}</div><small>{{ event.detail }}</small></el-timeline-item>
</el-timeline>
<el-empty v-else description="暂无状态变化记录" :image-size="72" />
</template>
<template #footer><el-button @click="detailOpen = false">关闭</el-button></template>
</el-dialog>
</template>
</BasicLayout>
</template>
<script setup>
import { computed, onMounted, reactive, ref } from 'vue'
import { Refresh } from '@element-plus/icons-vue'
import { ElMessage } from 'element-plus'
import { getEdgeNode, listEdgeNodes } from '@/api/sense/edge-node'
import { buildEdgeNodeQuery, durationText, eventLabel, freshnessText, stateLabel, stateType } from './edgeNodeState'
defineOptions({ name: 'SenseEdgeNode' })
const loading = ref(false)
const nodes = ref([])
const total = ref(0)
const detailOpen = ref(false)
const selected = ref(null)
const query = reactive({ pageIndex: 1, pageSize: 10 })
const summary = reactive({ total: 0, online: 0, offline: 0, recovering: 0, loadUsed: 0, loadCapacity: 0, backfillQueueDepth: 0, heartbeatTimeoutSeconds: 90 })
const summaryCards = computed(() => [
{ label: '节点总数', value: summary.total }, { label: '在线 / 恢复中', value: `${summary.online} / ${summary.recovering}`, className: 'online-number' },
{ label: '总负载', value: `${summary.loadUsed} / ${summary.loadCapacity} 路` }, { label: '待回填', value: `${summary.backfillQueueDepth} 条`, className: summary.backfillQueueDepth ? 'warning-number' : '' }
])
function unwrap(response) { return response?.data?.data ?? response?.data ?? response }
function formatTime(value) { return value ? new Date(value).toLocaleString('zh-CN', { hour12: false }) : '—' }
function recoveryText(item) { return item.status === 'recovering' ? (item.recoveryPhase || '通道收敛中') : '已收敛' }
async function load() { loading.value = true; try { const payload = unwrap(await listEdgeNodes(buildEdgeNodeQuery(query))) || {}; nodes.value = payload.list || []; total.value = payload.count || 0; Object.assign(summary, payload.summary || {}) } catch (error) { ElMessage.error(error.message || '边缘节点加载失败') } finally { loading.value = false } }
async function openDetail(id) { try { selected.value = unwrap(await getEdgeNode(id)); detailOpen.value = true } catch (error) { ElMessage.error(error.message || '节点详情加载失败') } }
onMounted(load)
</script>
<style scoped>
.page-header{display:flex;align-items:flex-start;justify-content:space-between;gap:16px}.page-header h3{margin:0 0 6px}.page-header p{margin:0;color:#909399}.status-alert{margin:16px 0}.summary-row{margin:16px 0 20px}.summary-item{display:flex;align-items:center;justify-content:space-between;min-height:72px;padding:12px 16px;border:1px solid #ebeef5;border-radius:4px}.summary-item span{color:#606266}.summary-item strong{font-size:20px}.online-number{color:#67c23a}.warning-number{color:#e6a23c}.muted,small{color:#909399;font-size:12px}.detail-alert{margin-bottom:16px}.audit-title{margin:20px 0 14px}@media(max-width:768px){.page-header{align-items:stretch;flex-direction:column}.summary-item{margin-bottom:8px;min-height:64px}.page-header .el-button{min-height:44px}}
</style>
@@ -0,0 +1,65 @@
<template>
<BasicLayout>
<template #wrapper>
<el-card class="box-card">
<div class="page-header">
<div><h3>媒体分片</h3><p>查看 MediaMTX 分片容量、路径归属与故障影响。故障不会自动迁移路径。</p></div>
<el-button :icon="Refresh" :loading="loading" @click="load">刷新状态</el-button>
</div>
<el-alert v-if="summary.failed" type="error" :closable="false" show-icon class="status-alert">
<template #title>{{ summary.failed }} 个媒体分片异常,影响 {{ summary.impactedPaths }} 条路径</template>
路径仍保持原分片归属,便于定位故障范围;请先查看详情,不要直接变更生产配置。
</el-alert>
<el-row :gutter="12" class="summary-row" aria-label="媒体分片概览">
<el-col v-for="item in summaryCards" :key="item.label" :xs="12" :sm="6"><div class="summary-item"><span>{{ item.label }}</span><strong :class="item.className">{{ item.value }}</strong></div></el-col>
</el-row>
<el-table v-loading="loading" :data="shards" border stripe empty-text="暂无媒体分片运行记录">
<el-table-column label="分片" min-width="180"><template #default="scope"><strong>{{ scope.row.name }}</strong><div class="muted">{{ scope.row.id }} · {{ stateLabel(scope.row.mode) }}</div></template></el-table-column>
<el-table-column label="状态" width="112" align="center"><template #default="scope"><el-tag :type="stateType(scope.row.status)">{{ stateLabel(scope.row.status) }}</el-tag><div class="muted">{{ freshnessText(scope.row) }}</div></template></el-table-column>
<el-table-column label="已分配 / 容量" min-width="190"><template #default="scope"><div>{{ scope.row.assignedPaths }} / {{ scope.row.capacity }} 条路径</div><el-progress :percentage="usagePercent(scope.row)" :status="usagePercent(scope.row) >= 90 ? 'warning' : ''" /></template></el-table-column>
<el-table-column label="剩余容量" width="110" align="center"><template #default="scope">{{ scope.row.remaining }}</template></el-table-column>
<el-table-column prop="detail" label="运行说明" min-width="210" show-overflow-tooltip />
<el-table-column label="最近探测" min-width="170"><template #default="scope">{{ formatTime(scope.row.lastProbeAt) }}</template></el-table-column>
<el-table-column label="操作" width="156" fixed="right"><template #default="scope"><el-button v-permisaction="['sense:media-shard:detail']" type="primary" link @click="openDetail(scope.row.id)">影响详情</el-button><el-button v-permisaction="['sense:media-shard:preflight']" type="primary" link @click="openPreflight(scope.row.id)">迁移预检</el-button></template></el-table-column>
</el-table>
</el-card>
<el-dialog v-model="detailOpen" title="分片故障影响" width="min(880px, calc(100vw - 24px))" :close-on-click-modal="false">
<template v-if="selected"><el-descriptions :column="2" border><el-descriptions-item label="分片">{{ selected.name }}({{ selected.id }})</el-descriptions-item><el-descriptions-item label="状态"><el-tag :type="stateType(selected.status)">{{ stateLabel(selected.status) }}</el-tag></el-descriptions-item><el-descriptions-item label="路径占用">{{ selected.assignedPaths }} / {{ selected.capacity }}</el-descriptions-item><el-descriptions-item label="运行说明">{{ selected.detail }}</el-descriptions-item></el-descriptions>
<h4>受影响设备与 Profile</h4><el-table :data="selected.impact || []" border empty-text="此分片暂无路径归属"><el-table-column prop="deviceName" label="设备" min-width="150"><template #default="scope">{{ scope.row.deviceName || scope.row.deviceId }}<div class="muted">{{ scope.row.location || '未填写位置' }}</div></template></el-table-column><el-table-column prop="profileToken" label="Profile" min-width="130" /><el-table-column prop="path" label="媒体路径" min-width="180" /><el-table-column prop="actual" label="最近状态" width="120" /></el-table>
</template><template #footer><el-button @click="detailOpen = false">关闭</el-button></template>
</el-dialog>
<el-dialog v-model="preflightOpen" title="跨分片迁移预检(只读)" width="min(720px, calc(100vw - 24px))" :close-on-click-modal="false">
<template v-if="preflight"><el-alert :type="preflight.ready ? 'warning' : 'error'" :title="preflightConclusion(preflight)" :closable="false" show-icon /><el-descriptions :column="2" border class="preflight-summary"><el-descriptions-item label="源分片">{{ preflight.sourceShardId }}</el-descriptions-item><el-descriptions-item label="候选目标">{{ preflight.targetShardName || '无可用目标' }}</el-descriptions-item><el-descriptions-item label="影响路径">{{ preflight.impactedPaths }}</el-descriptions-item><el-descriptions-item label="执行授权">未授权</el-descriptions-item></el-descriptions><el-table :data="preflight.checks || []" border><el-table-column label="结果" width="90"><template #default="scope"><el-tag :type="scope.row.passed ? 'success' : 'danger'">{{ scope.row.passed ? '通过' : '未通过' }}</el-tag></template></el-table-column><el-table-column prop="name" label="检查项" width="130" /><el-table-column prop="detail" label="说明" min-width="260" /></el-table></template>
<template #footer><span class="read-only-note">本页面不会修改任何路径归属。</span><el-button @click="preflightOpen = false">关闭</el-button></template>
</el-dialog>
</template>
</BasicLayout>
</template>
<script setup>
import { computed, onMounted, reactive, ref } from 'vue'
import { Refresh } from '@element-plus/icons-vue'
import { ElMessage } from 'element-plus'
import { getMediaShard, listMediaShards, preflightMediaShard } from '@/api/sense/media-shard'
import { freshnessText, preflightConclusion, stateLabel, stateType, usagePercent } from './mediaShardState'
defineOptions({ name: 'SenseMediaShard' })
const loading = ref(false); const shards = ref([]); const selected = ref(null); const preflight = ref(null); const detailOpen = ref(false); const preflightOpen = ref(false)
const summary = reactive({ total: 0, available: 0, failed: 0, configuredCapacity: 0, assignedPaths: 0, impactedPaths: 0 })
const summaryCards = computed(() => [{ label: '分片总数', value: summary.total }, { label: '可用 / 异常', value: `${summary.available} / ${summary.failed}`, className: summary.failed ? 'danger-number' : 'success-number' }, { label: '已分配 / 总容量', value: `${summary.assignedPaths} / ${summary.configuredCapacity}` }, { label: '故障影响', value: `${summary.impactedPaths} 条路径`, className: summary.impactedPaths ? 'danger-number' : '' }])
function unwrap(response) { return response?.data?.data ?? response?.data ?? response }
function formatTime(value) { return value ? new Date(value).toLocaleString('zh-CN', { hour12: false }) : '—' }
async function load() { loading.value = true; try { const payload = unwrap(await listMediaShards()) || {}; shards.value = payload.list || []; Object.assign(summary, payload.summary || {}) } catch (error) { ElMessage.error(error.message || '媒体分片加载失败') } finally { loading.value = false } }
async function openDetail(id) { try { selected.value = unwrap(await getMediaShard(id)); detailOpen.value = true } catch (error) { ElMessage.error(error.message || '分片详情加载失败') } }
async function openPreflight(id) { try { preflight.value = unwrap(await preflightMediaShard(id)); preflightOpen.value = true } catch (error) { ElMessage.error(error.message || '迁移预检失败') } }
onMounted(load)
</script>
<style scoped>
.page-header{display:flex;align-items:flex-start;justify-content:space-between;gap:16px}.page-header h3{margin:0 0 6px}.page-header p{margin:0;color:#909399}.status-alert{margin:16px 0}.summary-row{margin:16px 0 20px}.summary-item{display:flex;align-items:center;justify-content:space-between;min-height:72px;padding:12px 16px;border:1px solid #ebeef5;border-radius:4px}.summary-item span{color:#606266}.summary-item strong{font-size:20px}.success-number{color:#67c23a}.danger-number{color:#f56c6c}.muted{color:#909399;font-size:12px;margin-top:4px}.preflight-summary{margin:16px 0}.read-only-note{margin-right:16px;color:#909399}@media(max-width:768px){.page-header{align-items:stretch;flex-direction:column}.summary-item{margin-bottom:8px;min-height:64px}.page-header .el-button{min-height:44px}}
</style>
@@ -0,0 +1,6 @@
const labels = { running: '运行正常', failed: '运行异常', unknown: '等待探测', disabled: '已停用', managed: 'Sense 托管', external: '外部运行' }
export function stateLabel(value) { return labels[value] || value || '未知' }
export function stateType(value) { return ({ running: 'success', failed: 'danger', unknown: 'warning', disabled: 'info' })[value] || 'info' }
export function usagePercent(item) { const capacity = Number(item?.capacity) || 0; return capacity > 0 ? Math.min(100, Math.round((Number(item.assignedPaths) || 0) * 100 / capacity)) : 0 }
export function freshnessText(item) { if (!item?.lastProbeAt) return '尚未探测'; return item.stale ? '状态已陈旧' : '刚刚更新' }
export function preflightConclusion(result) { if (!result) return ''; if (!result.ready) return '当前不满足迁移前置条件'; return '预检通过,但仍需单独建立高风险工单并人工确认' }
@@ -0,0 +1,109 @@
<template>
<BasicLayout>
<template #wrapper>
<el-card class="box-card">
<div class="page-header">
<div>
<h3>运维告警</h3>
<p>处理设备、媒体服务和边缘节点的运行异常。</p>
</div>
<el-button v-permisaction="['sense:ops-alert:evaluate']" :icon="Refresh" :loading="refreshing" @click="refreshSources">刷新状态</el-button>
</div>
<el-alert class="boundary-alert" type="info" :closable="false" show-icon title="这里只处理系统运行问题">
运维告警不会创建本地安全事件,也不会发送到 Bell 作为业务预警;确认告警仅表示已有人处理,不代表故障已经恢复。
</el-alert>
<el-row :gutter="12" class="summary-row" aria-label="告警概览">
<el-col :xs="24" :sm="8"><div class="summary-item"><span>活动告警</span><strong>{{ summary.active }}</strong></div></el-col>
<el-col :xs="24" :sm="8"><div class="summary-item"><span>待确认</span><strong class="danger-number">{{ summary.unacknowledged }}</strong></div></el-col>
<el-col :xs="24" :sm="8"><div class="summary-item"><span>恢复观察</span><strong class="primary-number">{{ summary.recovering }}</strong></div></el-col>
</el-row>
<el-form :model="query" label-width="76px" class="filter-form" @submit.prevent="search">
<el-form-item label="告警类型">
<el-select v-model="query.alertType" clearable placeholder="全部类型">
<el-option v-for="item in typeOptions" :key="item.value" :label="item.label" :value="item.value" />
</el-select>
</el-form-item>
<el-form-item label="处理状态">
<el-select v-model="query.state" clearable placeholder="全部状态">
<el-option label="待确认" value="unacknowledged" /><el-option label="已确认" value="acknowledged" />
<el-option label="恢复观察" value="recovering" /><el-option label="已恢复" value="recovered" />
</el-select>
</el-form-item>
<el-form-item label="关键词"><el-input v-model="query.keyword" clearable placeholder="设备、位置或详情" @keyup.enter="search" /></el-form-item>
<el-form-item class="filter-actions"><el-button type="primary" :icon="Search" native-type="submit">查询</el-button><el-button :icon="RefreshLeft" @click="reset">重置</el-button></el-form-item>
</el-form>
<el-table v-loading="loading" :data="alerts" border stripe empty-text="当前筛选条件下没有运维告警">
<el-table-column label="告警" min-width="180"><template #default="scope"><strong>{{ scope.row.title }}</strong><div class="muted">{{ alertTypeLabel(scope.row.alertType) }}</div></template></el-table-column>
<el-table-column label="对象" min-width="170"><template #default="scope"><strong>{{ scope.row.objectName }}</strong><div class="muted">{{ scope.row.location || scope.row.objectId }}</div></template></el-table-column>
<el-table-column label="状态" width="110" align="center"><template #default="scope"><el-tag :type="alertStateType(scope.row.state)" size="small">{{ alertStateLabel(scope.row.state) }}</el-tag></template></el-table-column>
<el-table-column label="首次 / 最近发现" min-width="190"><template #default="scope"><div>{{ formatTime(scope.row.firstSeenAt) }}</div><small>{{ formatTime(scope.row.lastSeenAt) }}</small></template></el-table-column>
<el-table-column prop="occurrenceCount" label="重复次数" width="92" align="center" />
<el-table-column label="操作" width="184" fixed="right">
<template #default="scope">
<el-button v-permisaction="['sense:ops-alert:detail']" type="primary" link @click="openDetail(scope.row.id)">详情</el-button>
<el-button v-if="canAcknowledge(scope.row)" v-permisaction="['sense:ops-alert:acknowledge']" type="warning" link @click="openAction(scope.row, 'acknowledge')">确认</el-button>
<el-button v-if="canRecover(scope.row)" v-permisaction="['sense:ops-alert:recover']" type="success" link @click="openAction(scope.row, 'recover')">恢复</el-button>
</template>
</el-table-column>
</el-table>
<pagination v-show="total > 0" v-model:current-page="query.pageIndex" v-model:page-size="query.pageSize" :total="total" @pagination="load" />
<p class="help-text">同一对象的同类故障使用唯一指纹去重;恢复后再次发生会复用历史记录并开启新的处理周期。</p>
</el-card>
<el-dialog v-model="detailOpen" title="运维告警详情" width="min(780px, calc(100vw - 24px))">
<el-descriptions v-if="selected" :column="2" border>
<el-descriptions-item label="告警编号">{{ selected.id }}</el-descriptions-item><el-descriptions-item label="处理状态"><el-tag :type="alertStateType(selected.state)">{{ alertStateLabel(selected.state) }}</el-tag></el-descriptions-item>
<el-descriptions-item label="异常对象">{{ selected.objectName }}</el-descriptions-item><el-descriptions-item label="发现次数">{{ selected.occurrenceCount }}</el-descriptions-item>
<el-descriptions-item label="异常详情" :span="2">{{ selected.detail }}</el-descriptions-item><el-descriptions-item label="建议处理" :span="2">{{ selected.nextAction }}</el-descriptions-item>
</el-descriptions>
<el-alert class="detail-boundary" type="warning" :closable="false" show-icon title="业务隔离边界">该记录仅用于 Sense 运维,不会生成安全事件或 Bell 业务预警。</el-alert>
<el-timeline v-if="transitions.length" class="timeline"><el-timeline-item v-for="item in transitions" :key="item.id" :timestamp="formatTime(item.createdAt)"><strong>{{ actionLabel(item.action) }}</strong><div class="muted">{{ item.reason }}</div></el-timeline-item></el-timeline>
<template #footer><el-button @click="detailOpen = false">关闭</el-button></template>
</el-dialog>
<el-dialog v-model="actionOpen" :title="actionMode === 'acknowledge' ? '确认运维告警' : '确认故障恢复'" width="min(520px, calc(100vw - 24px))" :close-on-click-modal="false">
<el-alert :type="actionMode === 'acknowledge' ? 'warning' : 'success'" :closable="false" show-icon :title="actionMode === 'acknowledge' ? '确认不代表恢复' : '仅在健康状态已稳定后关闭告警'" />
<el-form ref="actionFormRef" :model="actionForm" :rules="actionRules" label-position="top" class="action-form">
<el-form-item label="处理说明" prop="reason"><el-input v-model="actionForm.reason" type="textarea" :rows="4" maxlength="256" show-word-limit placeholder="请输入至少 6 个字符,说明处理人、措施或恢复依据" /></el-form-item>
</el-form>
<template #footer><el-button @click="actionOpen = false">取消</el-button><el-button :type="actionMode === 'acknowledge' ? 'warning' : 'success'" :loading="submitting" @click="submitAction">确认提交</el-button></template>
</el-dialog>
</template>
</BasicLayout>
</template>
<script setup>
import { onMounted, reactive, ref } from 'vue'
import { Refresh, RefreshLeft, Search } from '@element-plus/icons-vue'
import { ElMessage } from 'element-plus'
import { acknowledgeOpsAlert, evaluateOpsAlerts, getOpsAlert, listOpsAlerts, recoverOpsAlert } from '@/api/sense/ops-alert'
import { alertStateLabel, alertStateType, alertTypeLabel, buildOpsAlertQuery, canAcknowledge, canRecover } from './opsAlertState'
defineOptions({ name: 'SenseOpsAlert' })
const loading = ref(false); const refreshing = ref(false); const submitting = ref(false); const detailOpen = ref(false); const actionOpen = ref(false)
const alerts = ref([]); const total = ref(0); const selected = ref(null); const transitions = ref([]); const actionMode = ref('acknowledge'); const actionFormRef = ref()
const summary = reactive({ active: 0, unacknowledged: 0, recovering: 0 })
const query = reactive({ pageIndex: 1, pageSize: 10, alertType: '', state: '', keyword: '' })
const actionForm = reactive({ reason: '', expectedVersion: 0 })
const actionRules = { reason: [{ required: true, message: '请输入处理说明', trigger: 'blur' }, { min: 6, max: 256, message: '处理说明需为 6 至 256 个字符', trigger: 'blur' }] }
const typeOptions = ['device_offline', 'authentication_failed', 'clock_drift', 'reconciliation_failed', 'media_shard_failed', 'control_tunnel_failed'].map(value => ({ value, label: alertTypeLabel(value) }))
function unwrap(response) { return response?.data?.data ?? response?.data ?? response }
function formatTime(value) { return value ? new Date(value).toLocaleString('zh-CN', { hour12: false }) : '—' }
function actionLabel(value) { return ({ detected: '发现异常', acknowledge: '人工确认', health_restored: '健康恢复', recover: '确认恢复', relapsed: '恢复失败', reopened: '再次发生' })[value] || value }
async function load() { loading.value = true; try { const payload = unwrap(await listOpsAlerts(buildOpsAlertQuery(query))) || {}; alerts.value = payload.list || []; total.value = payload.count || 0; Object.assign(summary, payload.summary || { active: 0, unacknowledged: 0, recovering: 0 }) } catch (error) { ElMessage.error(error.message || '运维告警加载失败') } finally { loading.value = false } }
async function refreshSources() { refreshing.value = true; try { await evaluateOpsAlerts(); ElMessage.success('状态已刷新'); await load() } catch (error) { ElMessage.error(error.message || '健康状态刷新失败') } finally { refreshing.value = false } }
function search() { query.pageIndex = 1; load() }
function reset() { Object.assign(query, { pageIndex: 1, alertType: '', state: '', keyword: '' }); load() }
async function openDetail(id) { try { const payload = unwrap(await getOpsAlert(id)) || {}; selected.value = payload.alert; transitions.value = payload.transitions || []; detailOpen.value = true } catch (error) { ElMessage.error(error.message || '告警详情加载失败') } }
function openAction(item, mode) { selected.value = item; actionMode.value = mode; actionForm.reason = ''; actionForm.expectedVersion = item.version; actionOpen.value = true }
async function submitAction() { if (!await actionFormRef.value.validate().catch(() => false)) return; submitting.value = true; try { const payload = { expectedVersion: actionForm.expectedVersion, reason: actionForm.reason.trim() }; if (actionMode.value === 'acknowledge') await acknowledgeOpsAlert(selected.value.id, payload); else await recoverOpsAlert(selected.value.id, payload); ElMessage.success(actionMode.value === 'acknowledge' ? '告警已确认' : '告警已恢复'); actionOpen.value = false; detailOpen.value = false; await load() } catch (error) { ElMessage.warning(error.message || '状态已变化,请刷新后重试') } finally { submitting.value = false } }
onMounted(load)
</script>
<style scoped>
.page-header{display:flex;align-items:flex-start;justify-content:space-between;gap:16px}.page-header h3{margin:0 0 6px}.page-header p{margin:0;color:#909399}.boundary-alert{margin:16px 0}.summary-row{margin-bottom:20px}.summary-item{display:flex;align-items:center;justify-content:space-between;min-height:72px;padding:12px 16px;border:1px solid #ebeef5;border-radius:4px}.summary-item span{color:#606266}.summary-item strong{font-size:22px}.danger-number{color:#f56c6c}.primary-number{color:#409eff}.filter-form{display:flex;align-items:flex-end;flex-wrap:wrap;gap:0 12px;margin:16px 0 2px}.filter-form .el-form-item{width:230px}.filter-form .filter-actions{width:auto}.filter-form :deep(.el-select){width:100%}.muted,small,.help-text{color:#909399;font-size:12px}.help-text{margin:12px 0 0}.detail-boundary{margin:16px 0}.timeline{padding-top:8px}.action-form{margin-top:16px}@media(max-width:768px){.page-header{align-items:stretch;flex-direction:column}.filter-form .el-form-item{width:100%}.summary-item{margin-bottom:8px}}
</style>
@@ -0,0 +1,30 @@
const typeLabels = {
device_offline: '设备离线',
authentication_failed: '认证失败',
clock_drift: '时间漂移',
reconciliation_failed: '状态对账失败',
media_shard_failed: '媒体分片异常',
control_tunnel_failed: '控制隧道异常'
}
const stateLabels = {
unacknowledged: '待确认',
acknowledged: '已确认',
recovering: '恢复观察',
recovered: '已恢复'
}
export function alertTypeLabel(value) { return typeLabels[value] || value || '未知类型' }
export function alertStateLabel(value) { return stateLabels[value] || value || '未知状态' }
export function alertStateType(value) { return ({ unacknowledged: 'danger', acknowledged: 'warning', recovering: 'primary', recovered: 'success' })[value] || 'info' }
export function canAcknowledge(item) { return item?.state === 'unacknowledged' && item?.version > 0 }
export function canRecover(item) { return item?.state === 'recovering' && item?.version > 0 }
export function buildOpsAlertQuery(query) {
return {
pageIndex: Number(query.pageIndex) || 1,
pageSize: Number(query.pageSize) || 10,
alertType: String(query.alertType || '').trim(),
state: String(query.state || '').trim(),
keyword: String(query.keyword || '').trim()
}
}
+447
View File
@@ -0,0 +1,447 @@
<template>
<BasicLayout>
<template #wrapper>
<el-card class="box-card">
<div class="page-header">
<div>
<h3>可靠投递</h3>
<p>查看 Sense 内部待投递记录、重试进度和人工恢复。</p>
</div>
<el-button
:icon="Refresh"
:loading="loading"
@click="load"
>刷新状态</el-button>
</div>
<el-alert
type="info"
:closable="false"
show-icon
class="boundary-alert"
>
<template #title>未配置外部投递出口不影响 Sense 核心功能</template>
内部记录会安全保留,正式 connector 由后续协调工单提供;测试接收器在
production 模式不可启用。
</el-alert>
<el-row :gutter="12" class="summary-row" aria-label="可靠投递队列概览">
<el-col
v-for="card in summaryCards"
:key="card.key"
:xs="12"
:sm="6"
><div class="summary-item">
<span>{{ card.label }}</span><strong :class="card.className">{{ summary[card.key] }}</strong>
</div></el-col>
</el-row>
<el-form
:model="query"
label-width="76px"
class="filter-form"
@submit.prevent="search"
>
<el-form-item label="关键词"><el-input
v-model="query.keyword"
clearable
placeholder="内部记录、业务引用或幂等键"
@keyup.enter="search"
/></el-form-item>
<el-form-item label="状态"><el-select
v-model="query.state"
clearable
placeholder="全部状态"
><el-option
v-for="(label, value) in stateLabels"
:key="value"
:label="label"
:value="value"
/></el-select></el-form-item>
<el-form-item label="内部类型"><el-select
v-model="query.internalType"
clearable
placeholder="全部类型"
><el-option
v-for="(label, value) in typeLabels"
:key="value"
:label="label"
:value="value"
/></el-select></el-form-item>
<el-form-item class="filter-actions"><el-button
type="primary"
:icon="Search"
native-type="submit"
>查询</el-button><el-button
:icon="RefreshLeft"
@click="reset"
>重置</el-button></el-form-item>
</el-form>
<el-table
v-loading="loading"
:data="items"
border
stripe
empty-text="当前筛选条件下没有可靠投递记录"
>
<el-table-column
label="内部记录"
min-width="255"
><template #default="scope"><strong>{{ typeLabel(scope.row.internalType) }}</strong>
<div class="code-text">{{ scope.row.id }}</div>
<small>幂等键:{{ scope.row.idempotencyKey }}</small></template></el-table-column>
<el-table-column
label="状态"
width="112"
align="center"
><template #default="scope"><el-tag :type="stateType(scope.row.state)">{{
stateLabel(scope.row.state)
}}</el-tag></template></el-table-column>
<el-table-column
label="尝试"
width="74"
align="center"
><template #default="scope">{{ scope.row.attemptCount }} 次</template></el-table-column>
<el-table-column
label="下次动作"
min-width="180"
><template #default="scope">{{
nextAction(scope.row)
}}</template></el-table-column>
<el-table-column
label="租约"
min-width="142"
><template #default="scope"><div>{{ scope.row.leaseOwner || "未领取" }}</div>
<small v-if="scope.row.leaseUntil">{{
formatTime(scope.row.leaseUntil)
}}</small></template></el-table-column>
<el-table-column
label="最近结果"
min-width="210"
><template #default="scope">{{
scope.row.lastError ||
(scope.row.state === "delivered" ? "投递成功" : "尚未失败")
}}</template></el-table-column>
<el-table-column
label="操作"
width="150"
fixed="right"
><template #default="scope"><el-button
v-permisaction="['sense:outbox:detail']"
type="primary"
link
@click="openDetail(scope.row.id)"
>详情</el-button><el-button
v-if="scope.row.state === 'dead'"
v-permisaction="['sense:outbox:requeue']"
type="warning"
link
@click="openRequeue(scope.row)"
>重新排队</el-button></template></el-table-column>
</el-table>
<pagination
v-show="total > 0"
v-model:current-page="query.pageIndex"
v-model:page-size="query.pageSize"
:total="total"
@pagination="load"
/>
<p class="boundary-text">
页面不展示内部
payload、外部凭据或机器身份;人工恢复保留原业务记录、幂等键和失败历史。
</p>
</el-card>
<el-dialog
v-model="detailOpen"
title="投递记录详情"
width="min(780px, calc(100vw - 24px))"
:close-on-click-modal="false"
>
<el-descriptions v-if="selected" :column="2" border>
<el-descriptions-item label="内部记录">{{
selected.message.id
}}</el-descriptions-item><el-descriptions-item label="内部类型">{{
typeLabel(selected.message.internalType)
}}</el-descriptions-item>
<el-descriptions-item
label="幂等键"
:span="2"
><span class="code-text">{{
selected.message.idempotencyKey
}}</span></el-descriptions-item><el-descriptions-item label="当前状态"><el-tag :type="stateType(selected.message.state)">{{
stateLabel(selected.message.state)
}}</el-tag></el-descriptions-item><el-descriptions-item label="业务引用">{{
selected.message.businessRef
}}</el-descriptions-item>
<el-descriptions-item label="最近结果" :span="2">{{
selected.message.lastError || "尚未失败"
}}</el-descriptions-item>
</el-descriptions>
<h4 class="timeline-title">处理时间线</h4>
<el-timeline v-if="selected"><el-timeline-item
v-for="attempt in selected.attempts"
:key="attempt.id"
:timestamp="formatTime(attempt.createdAt)"
placement="top"
><strong>第 {{ attempt.number }} 次 · {{ attempt.outcome }}</strong>
<div>{{ attempt.detail }}</div>
<small v-if="attempt.actorUserId">操作人编号:{{ attempt.actorUserId }}</small></el-timeline-item><el-timeline-item
v-if="!selected.attempts.length"
:timestamp="formatTime(selected.message.createdAt)"
>与业务记录在同一事务中创建</el-timeline-item></el-timeline>
<template #footer><el-button @click="detailOpen = false">关闭</el-button></template>
</el-dialog>
<el-dialog
v-model="requeueOpen"
title="将死信重新排队"
width="min(560px, calc(100vw - 24px))"
:close-on-click-modal="false"
@closed="resetRequeue"
>
<el-alert
title="这是受控恢复操作"
type="warning"
:closable="false"
show-icon
>只创建新的投递尝试,不修改原业务记录或删除失败历史。执行前请先排除失败原因。</el-alert>
<el-form
ref="requeueFormRef"
:model="requeueForm"
:rules="requeueRules"
label-position="top"
class="requeue-form"
><el-form-item
label="恢复原因"
prop="reason"
><el-input
v-model="requeueForm.reason"
type="textarea"
:rows="3"
maxlength="256"
show-word-limit
placeholder="例如:出口配置已恢复,已核对幂等键和目标状态"
/></el-form-item></el-form>
<p class="boundary-text">
原因将与操作人、对象版本和时间一起写入脱敏审计。
</p>
<template #footer><el-button @click="requeueOpen = false">取消</el-button><el-button
type="warning"
:loading="requeueLoading"
@click="confirmRequeue"
>确认重新排队</el-button></template>
</el-dialog>
</template>
</BasicLayout>
</template>
<script setup>
import { computed, onMounted, reactive, ref } from 'vue'
import { Refresh, RefreshLeft, Search } from '@element-plus/icons-vue'
import { ElMessage } from 'element-plus'
import { getOutbox, listOutbox, requeueOutbox } from '@/api/sense/outbox'
import {
buildOutboxQuery,
formatTime,
nextAction,
stateLabel,
stateLabels,
stateType,
typeLabel,
typeLabels
} from './outboxState'
defineOptions({ name: 'SenseOutbox' })
const loading = ref(false)
const items = ref([])
const total = ref(0)
const detailOpen = ref(false)
const selected = ref(null)
const requeueOpen = ref(false)
const requeueLoading = ref(false)
const requeueTarget = ref(null)
const requeueFormRef = ref(null)
const summary = reactive({ pending: 0, retry: 0, processing: 0, dead: 0 })
const query = reactive({
pageIndex: 1,
pageSize: 10,
state: '',
internalType: '',
keyword: ''
})
const requeueForm = reactive({ reason: '' })
const requeueRules = {
reason: [
{ required: true, message: '请填写恢复原因', trigger: 'blur' },
{ min: 6, max: 256, message: '请填写 6 至 256 个字符', trigger: 'blur' }
]
}
const summaryCards = computed(() => [
{ key: 'pending', label: '等待投递' },
{ key: 'retry', label: '重试等待', className: 'warning-number' },
{ key: 'processing', label: '处理中 / 租约' },
{ key: 'dead', label: '死信待处理', className: 'danger-number' }
])
function unwrap(response) {
return response?.data?.data ?? response?.data ?? response
}
async function load() {
loading.value = true
try {
const payload = unwrap(await listOutbox(buildOutboxQuery(query))) || {}
items.value = payload.list || []
total.value = payload.count || 0
Object.assign(
summary,
payload.summary || { pending: 0, retry: 0, processing: 0, dead: 0 }
)
} catch (error) {
ElMessage.error(error.message || '可靠投递状态加载失败')
} finally {
loading.value = false
}
}
function search() {
query.pageIndex = 1
load()
}
function reset() {
Object.assign(query, {
pageIndex: 1,
state: '',
internalType: '',
keyword: ''
})
load()
}
async function openDetail(id) {
try {
selected.value = unwrap(await getOutbox(id))
detailOpen.value = true
} catch (error) {
ElMessage.error(error.message || '投递详情加载失败')
}
}
function openRequeue(item) {
requeueTarget.value = item
requeueOpen.value = true
}
function resetRequeue() {
requeueForm.reason = ''
requeueTarget.value = null
requeueFormRef.value?.clearValidate()
}
async function confirmRequeue() {
if (!requeueFormRef.value || !requeueTarget.value) return
const valid = await requeueFormRef.value.validate().catch(() => false)
if (!valid) return
requeueLoading.value = true
try {
await requeueOutbox(
requeueTarget.value.id,
requeueTarget.value.version,
requeueForm.reason.trim()
)
ElMessage.success('死信已重新排队,原失败历史已保留')
requeueOpen.value = false
detailOpen.value = false
await load()
} catch (error) {
ElMessage.warning(error.message || '重新排队失败,请刷新状态后重试')
} finally {
requeueLoading.value = false
}
}
onMounted(load)
</script>
<style scoped>
.page-header {
display: flex;
align-items: flex-start;
justify-content: space-between;
gap: 16px;
}
.page-header h3 {
margin: 0 0 6px;
}
.page-header p {
margin: 0;
color: #909399;
}
.boundary-alert {
margin: 16px 0;
}
.summary-row {
margin-bottom: 18px;
}
.summary-item {
display: flex;
align-items: center;
justify-content: space-between;
min-height: 68px;
padding: 12px 16px;
border: 1px solid #ebeef5;
border-radius: 4px;
}
.summary-item span {
color: #606266;
}
.summary-item strong {
font-size: 22px;
}
.warning-number {
color: #e6a23c;
}
.danger-number {
color: #f56c6c;
}
.filter-form {
display: flex;
align-items: flex-end;
flex-wrap: wrap;
gap: 0 12px;
margin-bottom: 2px;
}
.filter-form .el-form-item {
width: 260px;
}
.filter-form .filter-actions {
width: auto;
}
.filter-form :deep(.el-select) {
width: 100%;
}
.code-text {
font-family: ui-monospace, SFMono-Regular, Consolas, monospace;
overflow-wrap: anywhere;
}
.boundary-text,
small {
color: #909399;
font-size: 12px;
}
.boundary-text {
margin: 12px 0 0;
}
.timeline-title {
margin: 20px 0 14px;
}
.requeue-form {
margin-top: 16px;
}
@media (max-width: 768px) {
.page-header {
align-items: stretch;
flex-direction: column;
}
.filter-form .el-form-item {
width: 100%;
}
.summary-item {
margin-bottom: 8px;
}
}
</style>
@@ -0,0 +1,54 @@
export const stateLabels = {
pending: '等待投递',
processing: '处理中',
retry: '重试等待',
dead: '死信',
delivered: '已投递'
}
export const typeLabels = {
local_event_candidate: '本地事件候选',
audit_projection: '审计投影'
}
export function stateLabel(value) {
return stateLabels[value] || value || '—'
}
export function stateType(value) {
return (
{
pending: 'info',
processing: 'primary',
retry: 'warning',
dead: 'danger',
delivered: 'success'
}[value] || 'info'
)
}
export function typeLabel(value) {
return typeLabels[value] || value || '—'
}
export function buildOutboxQuery(query) {
return {
pageIndex: query.pageIndex,
pageSize: query.pageSize,
state: query.state || undefined,
internalType: query.internalType || undefined,
keyword: String(query.keyword || '').trim() || undefined
}
}
export function nextAction(item, now = Date.now()) {
if (item.state === 'dead') return '等待人工处理'
if (item.state === 'processing') { return item.leaseUntil ? `租约至 ${formatTime(item.leaseUntil)}` : '处理中' }
if (item.state === 'retry') {
return item.availableAt && new Date(item.availableAt).getTime() > now
? `自动重试 ${formatTime(item.availableAt)}`
: '等待重试领取'
}
if (item.state === 'delivered') return '已完成'
return '等待 worker 领取'
}
export function formatTime(value) {
return value
? new Date(value).toLocaleString('zh-CN', { hour12: false })
: '—'
}
@@ -0,0 +1,14 @@
import request from '@/utils/request'
import { getEdgeNode, listEdgeNodes } from '@/api/sense/edge-node'
jest.mock('@/utils/request', () => jest.fn())
describe('Sense edge node API', () => {
beforeEach(() => request.mockReset())
test('uses read-only endpoints and encodes identifiers', () => {
listEdgeNodes({ pageIndex: 1, pageSize: 10 })
getEdgeNode('edge/site 01')
expect(request).toHaveBeenNthCalledWith(1, { url: '/api/v1/edge-nodes', method: 'get', params: { pageIndex: 1, pageSize: 10 } })
expect(request).toHaveBeenNthCalledWith(2, { url: '/api/v1/edge-nodes/edge%2Fsite%2001', method: 'get' })
})
})
@@ -0,0 +1,18 @@
import { buildEdgeNodeQuery, durationText, eventLabel, freshnessText, stateLabel, stateType } from '@/views/sense/edge-node/edgeNodeState'
describe('Sense edge node presentation state', () => {
test('uses operator-facing connection and channel labels', () => {
expect(stateLabel('offline')).toBe('离线')
expect(stateLabel('recovering')).toBe('恢复中')
expect(stateType('offline')).toBe('danger')
expect(eventLabel('node_recovered')).toBe('心跳恢复')
})
test('makes stale last-known data explicit', () => {
expect(freshnessText({ stale: true, freshnessSeconds: 245, lastHeartbeatAt: '2026-08-28T00:00:00Z' })).toBe('已陈旧 4 分钟')
expect(freshnessText({ stale: false, freshnessSeconds: 18, lastHeartbeatAt: '2026-08-28T00:00:00Z' })).toBe('18 秒前')
expect(durationText(172800)).toBe('2 天')
})
test('sends only allowlisted pagination fields', () => {
expect(buildEdgeNodeQuery({ pageIndex: '2', pageSize: 20, ignored: 'not-sent' })).toEqual({ pageIndex: 2, pageSize: 20 })
})
})
@@ -0,0 +1,8 @@
jest.mock('@/utils/request', () => jest.fn(config => config))
import request from '@/utils/request'
import { getMediaShard, listMediaShards, preflightMediaShard } from '@/api/sense/media-shard'
describe('Sense media shard API', () => {
beforeEach(() => request.mockClear())
test('exposes only read operations and safely encodes ids', () => { expect(listMediaShards()).toEqual({ url: '/api/v1/media-shards', method: 'get' }); expect(getMediaShard('a/b')).toEqual({ url: '/api/v1/media-shards/a%2Fb', method: 'get' }); expect(preflightMediaShard('a b')).toEqual({ url: '/api/v1/media-shards/a%20b/migration-preflight', method: 'get' }); expect(request).toHaveBeenCalledTimes(3) })
})
@@ -0,0 +1,7 @@
import { freshnessText, preflightConclusion, stateLabel, stateType, usagePercent } from '@/views/sense/media-shard/mediaShardState'
describe('Sense media shard presentation state', () => {
test('uses clear operator-facing labels and non-color-only status', () => { expect(stateLabel('failed')).toBe('运行异常'); expect(stateType('failed')).toBe('danger'); expect(stateLabel('external')).toBe('外部运行') })
test('derives usage from configured capacity instead of a fixed tier', () => { expect(usagePercent({ assignedPaths: 7, capacity: 23 })).toBe(30); expect(usagePercent({ assignedPaths: 99, capacity: 0 })).toBe(0) })
test('makes stale data and read-only preflight explicit', () => { expect(freshnessText({ lastProbeAt: '2026-08-28T00:00:00Z', stale: true })).toBe('状态已陈旧'); expect(preflightConclusion({ ready: true })).toContain('高风险工单'); expect(preflightConclusion({ ready: false })).toContain('不满足') })
})
@@ -0,0 +1,18 @@
import request from '@/utils/request'
import { acknowledgeOpsAlert, evaluateOpsAlerts, getOpsAlert, listOpsAlerts, recoverOpsAlert } from '@/api/sense/ops-alert'
jest.mock('@/utils/request', () => jest.fn(config => config))
describe('Sense operational alert API', () => {
beforeEach(() => request.mockClear())
test('exposes read, refresh and optimistic state actions', () => {
const query = { pageIndex: 1, pageSize: 10, state: 'unacknowledged' }
const action = { expectedVersion: 3, reason: '已安排现场人员处理设备故障' }
expect(listOpsAlerts(query)).toEqual({ url: '/api/v1/ops-alerts', method: 'get', params: query })
expect(getOpsAlert('alert/a b')).toEqual({ url: '/api/v1/ops-alerts/alert%2Fa%20b', method: 'get' })
expect(evaluateOpsAlerts()).toEqual({ url: '/api/v1/ops-alerts/evaluate', method: 'post' })
expect(acknowledgeOpsAlert('alert/a b', action)).toEqual({ url: '/api/v1/ops-alerts/alert%2Fa%20b/acknowledge', method: 'post', data: action })
expect(recoverOpsAlert('alert/a b', action)).toEqual({ url: '/api/v1/ops-alerts/alert%2Fa%20b/recover', method: 'post', data: action })
})
})
@@ -0,0 +1,25 @@
import { alertStateLabel, alertStateType, alertTypeLabel, buildOpsAlertQuery, canAcknowledge, canRecover } from '@/views/sense/ops-alert/opsAlertState'
describe('Sense operational alert presentation state', () => {
test('uses ordinary operations wording and six alert types', () => {
expect(alertTypeLabel('device_offline')).toBe('设备离线')
expect(alertTypeLabel('authentication_failed')).toBe('认证失败')
expect(alertTypeLabel('clock_drift')).toBe('时间漂移')
expect(alertTypeLabel('reconciliation_failed')).toBe('状态对账失败')
expect(alertTypeLabel('media_shard_failed')).toBe('媒体分片异常')
expect(alertTypeLabel('control_tunnel_failed')).toBe('控制隧道异常')
expect(alertStateLabel('recovering')).toBe('恢复观察')
expect(alertStateType('recovered')).toBe('success')
})
test('enforces the approved state action boundary', () => {
expect(canAcknowledge({ state: 'unacknowledged', version: 1 })).toBe(true)
expect(canAcknowledge({ state: 'acknowledged', version: 2 })).toBe(false)
expect(canRecover({ state: 'recovering', version: 3 })).toBe(true)
expect(canRecover({ state: 'acknowledged', version: 3 })).toBe(false)
})
test('builds an allowlisted query', () => {
expect(buildOpsAlertQuery({ pageIndex: '2', pageSize: 20, alertType: ' clock_drift ', state: ' recovering ', keyword: ' 南门 ', ignored: true })).toEqual({ pageIndex: 2, pageSize: 20, alertType: 'clock_drift', state: 'recovering', keyword: '南门' })
})
})
@@ -0,0 +1,27 @@
import request from '@/utils/request'
import { getOutbox, listOutbox, requeueOutbox } from '@/api/sense/outbox'
jest.mock('@/utils/request', () => jest.fn())
describe('Sense outbox API', () => {
beforeEach(() => request.mockReset())
test('encodes identifiers and sends only the recovery fields', () => {
listOutbox({ pageIndex: 1, pageSize: 10 })
getOutbox('outbox/id unsafe')
requeueOutbox('outbox/id unsafe', 4, '出口故障已排除')
expect(request).toHaveBeenNthCalledWith(1, {
url: '/api/v1/outbox',
method: 'get',
params: { pageIndex: 1, pageSize: 10 }
})
expect(request).toHaveBeenNthCalledWith(2, {
url: '/api/v1/outbox/outbox%2Fid%20unsafe',
method: 'get'
})
expect(request).toHaveBeenNthCalledWith(3, {
url: '/api/v1/outbox/outbox%2Fid%20unsafe/requeue',
method: 'post',
data: { expectedVersion: 4, reason: '出口故障已排除' }
})
})
})
@@ -0,0 +1,37 @@
import {
buildOutboxQuery,
nextAction,
stateLabel,
stateType,
typeLabel
} from '@/views/sense/outbox/outboxState'
describe('Sense outbox presentation state', () => {
test('uses operator-facing labels and non-color-only states', () => {
expect(stateLabel('dead')).toBe('死信')
expect(stateType('dead')).toBe('danger')
expect(typeLabel('local_event_candidate')).toBe('本地事件候选')
})
test('builds an allowlisted trimmed query', () => {
expect(
buildOutboxQuery({
pageIndex: 2,
pageSize: 20,
state: 'retry',
internalType: 'local_event_candidate',
keyword: ' key ',
ignored: 'no'
})
).toEqual({
pageIndex: 2,
pageSize: 20,
state: 'retry',
internalType: 'local_event_candidate',
keyword: 'key'
})
})
test('explains the next recovery action', () => {
expect(nextAction({ state: 'dead' })).toBe('等待人工处理')
expect(nextAction({ state: 'pending' })).toBe('等待 worker 领取')
})
})
+23 -2
View File
@@ -2,8 +2,8 @@
generated: true (请先修改 Gitea Wiki,禁止直接编辑本文件)
wiki_page: Architecture-and-Code-Map
wiki_url: https://git.ilapage.cn/ila/yovision/wiki/Architecture-and-Code-Map.-
wiki_revision: 502d61f5c765c6c9b71432cdbbf0159c9b9c1071
synchronized_at: 2026-08-28T03:50:51Z
wiki_revision: 578ddbaae3d7e846037d958085e40609cd398bef
synchronized_at: 2026-08-28T08:02:32Z
<!-- gitea-wiki-mirror:end -->
# 架构与代码地图
@@ -215,3 +215,24 @@ PostgreSQL 表 `sense_provisioning_batches` 保存幂等键、配额快照和汇
- 前端:`Sense/ui/src/views/sense/quota/` 与 `Sense/ui/src/api/sense/quota.js`;复用 BasicLayout、Axios、Element Plus Alert/Row/Progress/Form/Table/Pagination/Dialog/Tag/Button 和权限指令。
- 本模块不导入 Brain/Bell 模型,不访问其数据库,也不定义跨项目配额契约。
<!-- sense-quota:end -->
<!-- sense-edge-nodes:start -->
## Sense 边缘节点代码路径
- 领域投影、DTO、服务、审计、合成夹具与测试:`Sense/server/app/sense/edge_node/`。
- GoAdmin 路由:`Sense/server/app/admin/router/sense_edge_node.go`;只开放 `GET /api/v1/edge-nodes` 与 `GET /api/v1/edge-nodes/:id`,复用 JWT、Casbin、PermissionAction 和统一响应封装。
- PostgreSQL 表:`sense_edge_nodes` 保存最后已知投影,`sense_edge_node_events` 保存不可变状态转换;迁移 `2026082813000_edge_node.go` 建立动态菜单及 implementation_operator/site_admin/viewer 的只读权限,不写入夹具。
- go-admin-ui 页面:`Sense/ui/src/views/sense/edge-node/`;API 封装:`Sense/ui/src/api/sense/edge-node.js`。页面复用 BasicLayout、Element Plus 表格/分页/Dialog/Tag/Alert 和权限指令。
- 内部 `ApplyHeartbeat` 是未来 Sense 适配器的投影写入口,本工单不把它暴露为 HTTP API,也不定义跨项目契约。
<!-- sense-edge-nodes:end -->
<!-- sense-media-shards:start -->
## Sense MediaMTX 分片代码路径
- 分片模型、容量分配、健康探测、故障影响和只读迁移预检位于 `Sense/server/app/sense/media_shard/`;既有媒体路由仍由 `Sense/server/app/sense/media/` 管理。
- PostgreSQL 表 `sense_media_shards` 保存运行配置投影和健康状态,`sense_media_shard_assignments` 保存路径到分片的稳定归属;`sense_media_routes` 继续作为路径状态事实源。
- 新路径在健康且有剩余配置容量的分片间使用 `rendezvous-v1` 确定性分配。已有归属在分片故障时保持不变,不自动迁移。
- 主分片继续复用既有 MediaMTX Supervisor;额外分片通过仓库外 JSON 配置接入本机回环 Control API,并按 external 实例管理,不由 Sense 启停。
- GoAdmin 路由位于 `Sense/server/app/admin/router/sense_media_shard.go`,只开放列表、详情和迁移预检三个 GET 接口。go-admin-ui 页面位于 `Sense/ui/src/views/sense/media-shard/`,复用 BasicLayout、Element Plus 表格、进度、Dialog、Tag、Alert 和权限指令。
- 迁移 `2026082814000_media_shard.go` 建表、注册动态菜单并为 implementation_operator、site_admin、viewer 建立只读权限;生产迁移和启动不写入合成分片。
<!-- sense-media-shards:end -->
+24 -2
View File
@@ -2,8 +2,8 @@
generated: true (请先修改 Gitea Wiki,禁止直接编辑本文件)
wiki_page: Business-Rules-and-Glossary
wiki_url: https://git.ilapage.cn/ila/yovision/wiki/Business-Rules-and-Glossary.-
wiki_revision: f7a08ae4eaa7557f2b8a470d895b06567f77d0e1
synchronized_at: 2026-08-28T03:51:05Z
wiki_revision: 999eb1aee3558ff75cfa929acac77841af20b742
synchronized_at: 2026-08-28T08:02:42Z
<!-- gitea-wiki-mirror:end -->
# 业务规则与术语
@@ -188,3 +188,25 @@ synchronized_at: 2026-08-28T03:51:05Z
- 32/64/128 只展示当前配置和目标硬件性能测试状态。未验证不得解释为支持或稳定承载承诺。
- 配额调整必须记录旧值、新值、操作人、时间和原因;viewer 与实施/运维角色只读,只有站点管理员可调整配额和重新启用设备。
<!-- sense-quota:end -->
<!-- sense-edge-nodes:start -->
## Sense 边缘节点状态规则
- **边缘节点投影**:Sense 保存的节点最后已知运行状态,不是 Brain/Bell 的共享机器身份或跨产品事实源。
- **在线**:最近心跳距读取时刻不超过 90 秒,且没有未完成的恢复阶段。
- **离线**:最近心跳距读取时刻超过 90 秒。离线是读取时推导状态,不得用“未知”覆盖最后已知的控制隧道、视频数据面、负载和回填队列;所有这些值必须同时标明陈旧。
- **恢复中**:心跳已经恢复,但控制隧道、视频数据面或回填通道仍在独立收敛。只有恢复阶段明确收敛后才显示在线。
- 状态转换写入不可变事件;列表和详情读取写入脱敏操作审计,不记录凭据或原始心跳载荷。
- 合成夹具必须显式调用,生产迁移和启动不得自动写入节点。
<!-- sense-edge-nodes:end -->
<!-- sense-media-shards:start -->
## Sense MediaMTX 分片规则
- **媒体分片**是一个独立 MediaMTX Control API 及其配置容量的 Sense 本地投影,不是 Brain 分片,也不是跨产品共享节点身份。
- 主分片容量优先读取显式 `SENSE_MEDIAMTX_CAPACITY`;未设置时读取 Sense PostgreSQL 的当前统一配额。额外分片容量逐项来自仓库外 JSON 配置。16 和 128 均不得作为分配算法硬上限。
- 新媒体路径只在状态为运行正常或等待首次探测、且仍有配置容量的分片中确定性分配;同一路径重复处理必须保持原归属。
- 分片异常、停用或状态陈旧都不得自动改写已有路径归属。详情必须能定位受影响的设备、位置、Profile 和媒体路径,同时不返回 Control API、RTSP URI 或凭据。
- 跨分片迁移预检是只读操作,只检查源状态、全部受影响路径和候选目标容量;即使预检通过也不授予执行权限。实际迁移必须另建高风险工单并取得人工确认。
- 额外分片只能配置为 external,Control API 必须使用无用户信息、无查询参数、无路径的本机回环 HTTP 地址。Sense 不停止外部实例。
<!-- sense-media-shards:end -->
+118 -2
View File
@@ -2,8 +2,8 @@
generated: true (请先修改 Gitea Wiki,禁止直接编辑本文件)
wiki_page: Local-Development-and-Verification
wiki_url: https://git.ilapage.cn/ila/yovision/wiki/Local-Development-and-Verification.-
wiki_revision: e1755deff68c95188e6bf2f759c7ec93293c8715
synchronized_at: 2026-08-27T15:22:48Z
wiki_revision: 05435d53528271a866c525655486689afc198762
synchronized_at: 2026-08-28T08:02:52Z
<!-- gitea-wiki-mirror:end -->
# 本地开发与验证
@@ -399,3 +399,119 @@ D:\supervisor\supervisord.exe ctl /c D:\supervisor\supervisord.conf reload
当前 Go Supervisor 的 `reload` 会重新读取独立配置;实际受影响实例必须以命令输出和 reload 前后 PID 为准。切换托管前先停止占用 Sense 端口的非 Supervisor 实例,防止自动启动进入 Backoff。验证至少包含 Supervisor 状态为 Running、`http://127.0.0.1:18080/health` 与首页返回 200,以及受控重启后 Sense 和受管 MediaMTX PID 均更新。
<!-- sense-supervisor:end -->
<!-- sense-local-events:start -->
### Sense 本地事件验证
合成数据通过 `local_event.SeedSyntheticFixture` 在测试中显式装载,不存在生产自动开关,也不要求 Brain 或 Bell 进程。定向及回归命令:
```powershell
cd Sense/server
go test -race ./app/sense/local_event
go test ./...
go vet ./...
go build ./...
cd ../ui
corepack pnpm@9.15.1 lint
corepack pnpm@9.15.1 test:unit
corepack pnpm@9.15.1 build:prod
```
PostgreSQL 迁移测试使用独立 schema;连接值只放当前进程环境,不写入仓库或日志:
```powershell
$env:SENSE_LOCAL_EVENT_MIGRATION_TEST_DATABASE_URL = '<隔离 PostgreSQL 连接>'
go test ./cmd/migrate/migration/version -run TestLocalEventMigrationOnPostgres -count=1 -v
```
验证应覆盖:candidate/confirmed 与 pending/success/failed 的独立组合、分页和时间/状态/关键词筛选、详情及保留提示、viewer/实施/站点管理员只读权限、操作审计,以及 Brain/Bell 均未启动时的合成候选 smoke。没有可用 PostgreSQL 测试连接时必须明确记录为未验证,不得以 SQLite 单测替代 PostgreSQL 结论。
<!-- sense-local-events:end -->
<!-- sense-operations:start -->
### Sense 运维中心验证
定向验证:
```powershell
cd Sense/server
go test -race ./app/sense/operations
go test ./app/admin/router ./cmd/migrate/migration/version
go test ./...
go vet ./...
go build ./...
cd ../ui
corepack pnpm@9.15.1 lint
corepack pnpm@9.15.1 test:unit
corepack pnpm@9.15.1 build:prod
```
合成 smoke 应覆盖设备认证失败、设备时间漂移、媒体退避、孤儿路由、本地推理 unavailable、筛选分页、详情、设备同步探测重试、媒体排队重试、版本冲突、在途防重和脱敏审计。Brain/Bell 不应启动或成为测试依赖。真实摄像头与 MediaMTX 故障恢复仍须在获准环境验证,日志不得记录设备地址、Stream URI 或凭据。
<!-- sense-operations:end -->
<!-- sense-edge-nodes:start -->
### Sense 边缘节点验证
定向验证:
```powershell
cd Sense/server
go test -race ./app/sense/edge_node
go test ./cmd/migrate/migration/version -run EdgeNode
go test ./...
go vet ./...
cd ../ui
corepack pnpm@9.15.1 lint
corepack pnpm@9.15.1 test:unit
corepack pnpm@9.15.1 build:prod
```
PostgreSQL 迁移测试只在提供隔离连接时执行:
```powershell
$env:SENSE_EDGE_NODE_MIGRATION_TEST_DATABASE_URL = '<隔离 PostgreSQL 连接>'
go test ./cmd/migrate/migration/version -run TestEdgeNodeMigrationOnPostgres -count=1 -v
```
验证应覆盖在线、超过 90 秒的离线、心跳恢复但通道未收敛、最后已知状态与陈旧时长、负载和回填队列、不可变状态事件、只读权限、脱敏读取审计,以及 Brain/Bell 不运行时的独立性。合成节点只能由测试或显式开发调用;没有隔离 PostgreSQL 连接时必须记录迁移真库测试未执行。
<!-- sense-edge-nodes:end -->
<!-- sense-media-shards:start -->
### Sense MediaMTX 分片验证
定向与全量验证:
```powershell
cd Sense/server
go test -race ./app/sense/media ./app/sense/media_shard ./cmd/migrate/migration/version
go test ./...
go vet ./...
cd ../ui
corepack pnpm@9.15.1 lint
corepack pnpm@9.15.1 test:unit
corepack pnpm@9.15.1 build:prod
cd ../..
powershell.exe -NoProfile -File .\Sense\tests\package\run-tests.ps1
```
使用真实本机 MediaMTX 二进制验证两个独立测试实例;测试会分配临时回环端口并在结束时停止其自有进程:
```powershell
cd Sense/server
$env:SENSE_MEDIAMTX_TEST_BINARY = '<mediamtx.exe 的绝对路径>'
go test ./tests/media -run 'TestTwoRealMediaMTXShardsAreIndependentlyUsable|TestRealMediaMTXControlLifecycle' -count=1 -v
```
PostgreSQL 迁移测试仅使用专用隔离连接:
```powershell
$env:SENSE_MEDIA_SHARD_MIGRATION_TEST_DATABASE_URL = '<隔离 PostgreSQL 连接>'
go test ./cmd/migrate/migration/version -run TestMediaShardMigrationOnPostgres -count=1 -v
```
验证应覆盖任意配置容量、稳定重复分配、容量耗尽、分片故障后归属不变、设备/Profile/路径影响范围、只读迁移预检、只读 RBAC、Control API 不出现在响应,以及 Brain/Bell 均不运行。没有专用 PostgreSQL 连接时必须记录真库迁移测试未执行。
<!-- sense-media-shards:end -->
+10 -2
View File
@@ -2,8 +2,8 @@
generated: true (请先修改 Gitea Wiki,禁止直接编辑本文件)
wiki_page: Product-Requirements
wiki_url: https://git.ilapage.cn/ila/yovision/wiki/Product-Requirements.-
wiki_revision: e298b5a31fcc39e3e2d9b4df543a4bc50d94bcaa
synchronized_at: 2026-08-28T03:53:13Z
wiki_revision: dcbdcf563017a1749fa76ad5f78c74a2c3cc6be1
synchronized_at: 2026-08-28T06:16:00Z
<!-- gitea-wiki-mirror:end -->
# 产品需求
@@ -224,3 +224,11 @@ SEN-011 将默认 16 路实现为可配置的 Sense 本地交付配额。统一
容量页面只展示 16/32/64/128 的当前配置和性能测试状态。除已经验证的 16 路学校试点基线外,其余档位在完成目标硬件压测前都不得解释为单机承载承诺。
<!-- sense-quota:end -->
<!-- sense-edge-nodes:start -->
## Sense 边缘节点资产与离线状态
Sense 为网管和非技术运维人员提供只读的“边缘节点”页面,集中查看节点名称、位置、运行版本、模型版本、运行时长、承载路数、控制隧道、视频数据面和回填队列。节点最近心跳超过 90 秒即在读取时显示为离线;离线不会清空最后一次采集到的通道和队列状态,页面必须同时标明陈旧时长。心跳恢复后,连接状态与控制、视频、回填通道分别收敛,未全部收敛时显示“恢复中”。列表、详情及状态变化均可审计。
本能力只管理 Sense 自有投影,不建立 Brain/Bell 共享身份、控制协议或跨项目回填执行;Brain、Bell 未运行时仍可独立查看。合成节点仅供显式开发和测试,不由生产启动或迁移自动写入。
<!-- sense-edge-nodes:end -->
+22 -2
View File
@@ -2,8 +2,8 @@
generated: true (请先修改 Gitea Wiki,禁止直接编辑本文件)
wiki_page: Deployment-and-Operations
wiki_url: https://git.ilapage.cn/ila/yovision/wiki/Deployment-and-Operations.-
wiki_revision: 95965345e43d7fc403486c8afaecf36d79b757d0
synchronized_at: 2026-08-28T03:53:21Z
wiki_revision: ce246054849b5b797dc4da0a55116238a4c00d7e
synchronized_at: 2026-08-28T08:04:52Z
<!-- gitea-wiki-mirror:end -->
# YoVision 部署与运维
@@ -92,3 +92,23 @@ Sense\start_sense.bat
批量导入被拒绝时还应检查模板只含 line_number/name/location/address,地址必须为不带账号、查询参数或片段的 HTTP(S) ONVIF 地址。条目失败时按页面原因检查网络、获准网段、凭据和接入状态;仅重试失败项,不删除已成功设备。日志、导出和问题记录不得粘贴摄像头凭据。
<!-- sense-provisioning:end -->
<!-- sense-edge-nodes:start -->
## Sense 边缘节点状态排错
升级后看不到“边缘节点”菜单时,先确认数据库迁移 `2026082813000_edge_node.go` 已成功,再重新登录或刷新动态菜单。implementation_operator、site_admin、viewer 均只有列表和详情读取权限,本模块没有网页写入、删除或远程控制接口。
节点超过 90 秒没有心跳即显示离线。离线时页面继续显示最后已知的隧道、视频、负载和回填状态,并标明陈旧时长,这不表示通道仍实时可用。心跳恢复但通道未收敛时显示“恢复中”,应分别检查控制隧道、视频数据面和回填队列。当前版本只提供 Sense 内部投影入口,不包含外部心跳接入协议或跨节点回填执行;没有节点数据时不会自动生成演示节点。
<!-- sense-edge-nodes:end -->
<!-- sense-media-shards:start -->
## Sense MediaMTX 分片配置与排错
主 MediaMTX 继续使用 `SENSE_MEDIAMTX_MODE`、`SENSE_MEDIAMTX_BINARY`、`SENSE_MEDIAMTX_CONFIG` 和 `SENSE_MEDIAMTX_API`。可选 `SENSE_MEDIAMTX_CAPACITY` 为正整数;留空时使用数据库中的当前统一配额。
额外分片通过 `SENSE_MEDIAMTX_SHARDS_FILE` 指向仓库外 JSON 文件。相对路径按 Windows 交付包根目录解析;示例结构见包内 `config\mediamtx-shards.example.json`。每项必须包含唯一 id、名称、本机回环 Control API 和正容量,mode 只能为 external。文件不得包含摄像头账号、密码、数据库连接或其他秘密。
额外 MediaMTX 进程由部署方独立启动并为各实例分配不冲突的 API、RTSP 及其他监听端口;Sense 只做健康探测和路径管理,不启停这些外部进程。升级后看不到“媒体分片”时,确认迁移 `2026082814000_media_shard.go` 成功并重新登录刷新动态菜单。
页面“运行异常”表示最近一次 Control API 探测失败;“状态已陈旧”表示运行循环已超过 30 秒没有更新探测结果。故障时先在详情定位设备/Profile/路径,不要手工改数据库归属。迁移预检不会执行迁移;任何实际跨分片迁移都必须另建高风险工单和回退方案。
<!-- sense-media-shards:end -->