Compare commits

...
Author SHA1 Message Date
QiuSW 407ffa17b2 feat: 实现 Brain 匿名检测与单路跟踪 (#14) 2026-08-28 22:39:12 +08:00
ila ba4ec28763 feat: 建立 Brain 可替换视频解码流水线 (#13)
合入 dev,#13 保持待验收。
2026-08-28 22:27:27 +08:00
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
QiuSW 12f5419f21 feat: 实现容量与配额安全闸 (#75) 2026-08-28 11:55:11 +08:00
ila debf662138 Merge PR #118: Sense 运维中心 (#74)
将 #74 实现合入 dev;工单保持待验收。
2026-08-28 11:05:11 +08:00
QiuSW 184f101a86 feat: 实现 Sense 运维中心 (#74) 2026-08-28 11:02:34 +08:00
ila 4157d90e9f Merge PR #117: Sense 本地事件候选查看
关联 #73;合入 dev,工单保持待验收。
2026-08-28 10:26:58 +08:00
QiuSW 028519a710 feat: 实现 Sense 本地事件只读查看 (#73) 2026-08-28 10:26:13 +08:00
ila 44b60b5b46 Merge PR #116: Sense 批量开通与失败项重试 (#72)
实现已合入 dev,工单 #72 保持待验收。
2026-08-28 09:54:30 +08:00
QiuSW ae19e59703 feat: 实现 Sense 批量开通与失败重试 (#72) 2026-08-28 09:53:00 +08:00
ila 2e709aa4ff merge: 记录三项目并行工单验收 (#113)
#10、#62、#71 的用户验收状态已写入 Wiki 并同步核心镜像;未验证边界保持不变。
2026-08-27 23:30:15 +08:00
168 changed files with 11018 additions and 22 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),
)
@@ -0,0 +1,16 @@
"""Anonymous detection and single-stream tracking."""
from .detector import LumaBlobDetector, TorchLumaBlobDetector
from .models import BoundingBox, Detection, Detector, DetectorMetadata, TrackedObject
from .tracker import SingleStreamTracker
__all__ = [
"BoundingBox",
"Detection",
"Detector",
"DetectorMetadata",
"LumaBlobDetector",
"SingleStreamTracker",
"TorchLumaBlobDetector",
"TrackedObject",
]
+120
View File
@@ -0,0 +1,120 @@
"""Deterministic anonymous blob detectors with no biometric semantics."""
from __future__ import annotations
from collections.abc import Sequence
from yovision_brain.decode import DecodedFrame, DecoderError
from .models import BoundingBox, Detection, DetectorMetadata
_METADATA = DetectorMetadata(
name="yovision-luma-blob",
version="1.0.0",
source="YoVision Brain first-party deterministic algorithm",
license="No external model license; no learned weights are distributed",
weights="none",
)
def _components(mask: Sequence[Sequence[bool]], minimum_area: int) -> tuple[BoundingBox, ...]:
height = len(mask)
width = len(mask[0]) if height else 0
visited: set[tuple[int, int]] = set()
boxes: list[BoundingBox] = []
for y in range(height):
for x in range(width):
if not mask[y][x] or (x, y) in visited:
continue
pending = [(x, y)]
visited.add((x, y))
points: list[tuple[int, int]] = []
while pending:
current_x, current_y = pending.pop()
points.append((current_x, current_y))
for neighbor in (
(current_x - 1, current_y),
(current_x + 1, current_y),
(current_x, current_y - 1),
(current_x, current_y + 1),
):
nx, ny = neighbor
if 0 <= nx < width and 0 <= ny < height and mask[ny][nx] and neighbor not in visited:
visited.add(neighbor)
pending.append(neighbor)
if len(points) >= minimum_area:
xs, ys = zip(*points)
boxes.append(BoundingBox(min(xs), min(ys), max(xs) + 1, max(ys) + 1))
return tuple(sorted(boxes, key=lambda box: (box.top, box.left, box.bottom, box.right)))
def _validate_frame(frame: DecodedFrame) -> None:
if frame.pixel_format not in {"rgb24", "yuv444p"}:
raise DecoderError(f"anonymous detector does not support pixel format {frame.pixel_format!r}")
expected = frame.width * frame.height * 3
if len(frame.payload) != expected:
raise DecoderError(f"vision frame has {len(frame.payload)} bytes; expected {expected}")
class LumaBlobDetector:
"""Small CPU reference detector used for deterministic integration tests."""
metadata = _METADATA
def __init__(self, *, threshold: int = 200, minimum_area: int = 1) -> None:
if not 0 <= threshold <= 255 or minimum_area < 1:
raise ValueError("invalid luma detector threshold or minimum area")
self._threshold = threshold
self._minimum_area = minimum_area
def detect(self, frame: DecodedFrame) -> tuple[Detection, ...]:
_validate_frame(frame)
if frame.pixel_format == "rgb24":
pixels = [
max(frame.payload[index : index + 3])
for index in range(0, len(frame.payload), 3)
]
else:
pixels = list(frame.payload[: frame.width * frame.height])
mask = [
[pixels[y * frame.width + x] >= self._threshold for x in range(frame.width)]
for y in range(frame.height)
]
return tuple(
Detection(box=box, category="anonymous_target", confidence=1.0)
for box in _components(mask, self._minimum_area)
)
class TorchLumaBlobDetector:
"""PyTorch CPU/GPU smoke backend; it contains no external model weights."""
metadata = DetectorMetadata(
name="yovision-torch-luma-blob",
version="1.0.0",
source="YoVision Brain first-party PyTorch tensor implementation",
license="PyTorch BSD-3-Clause; no external model weights",
weights="none",
)
def __init__(self, *, threshold: int = 200, minimum_area: int = 1, device: str = "cpu") -> None:
self._threshold = threshold
self._minimum_area = minimum_area
self._device = device
def detect(self, frame: DecodedFrame) -> tuple[Detection, ...]:
_validate_frame(frame)
try:
import torch
except ImportError as exc:
raise RuntimeError("PyTorch runtime is required for TorchLumaBlobDetector") from exc
values = torch.tensor(list(frame.payload), dtype=torch.uint8, device=self._device)
if frame.pixel_format == "rgb24":
luma = values.reshape(frame.height, frame.width, 3).amax(dim=2)
else:
luma = values[: frame.width * frame.height].reshape(frame.height, frame.width)
mask = (luma >= self._threshold).cpu().tolist()
return tuple(
Detection(box=box, category="anonymous_target", confidence=1.0)
for box in _components(mask, self._minimum_area)
)
+52
View File
@@ -0,0 +1,52 @@
"""Privacy-preserving vision ports and observations."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Protocol
from yovision_brain.decode import DecodedFrame
@dataclass(frozen=True, slots=True)
class DetectorMetadata:
name: str
version: str
source: str
license: str
weights: str
@dataclass(frozen=True, slots=True)
class BoundingBox:
left: int
top: int
right: int
bottom: int
@property
def area(self) -> int:
return max(0, self.right - self.left) * max(0, self.bottom - self.top)
@dataclass(frozen=True, slots=True)
class Detection:
box: BoundingBox
category: str
confidence: float
@dataclass(frozen=True, slots=True)
class TrackedObject:
track_id: str
box: BoundingBox
category: str
confidence: float
frame_sequence: int
timestamp_ns: int
class Detector(Protocol):
metadata: DetectorMetadata
def detect(self, frame: DecodedFrame) -> tuple[Detection, ...]: ...
@@ -0,0 +1,83 @@
"""Session-local single-stream IoU tracker."""
from __future__ import annotations
from dataclasses import dataclass
from .models import BoundingBox, Detection, TrackedObject
def _iou(first: BoundingBox, second: BoundingBox) -> float:
intersection = BoundingBox(
max(first.left, second.left),
max(first.top, second.top),
min(first.right, second.right),
min(first.bottom, second.bottom),
).area
union = first.area + second.area - intersection
return intersection / union if union else 0.0
@dataclass(slots=True)
class _Track:
track_id: str
detection: Detection
missed: int = 0
class SingleStreamTracker:
"""Tracks anonymous boxes only within one process session and one stream."""
def __init__(self, *, iou_threshold: float = 0.2, max_missed: int = 2) -> None:
if not 0.0 <= iou_threshold <= 1.0 or max_missed < 0:
raise ValueError("invalid tracker threshold or missed-frame limit")
self._iou_threshold = iou_threshold
self._max_missed = max_missed
self._tracks: dict[str, _Track] = {}
self._next_id = 1
def update(
self,
detections: tuple[Detection, ...],
*,
frame_sequence: int,
timestamp_ns: int,
) -> tuple[TrackedObject, ...]:
unmatched_tracks = set(self._tracks)
results: list[TrackedObject] = []
for detection in detections:
candidates = [
(track_id, _iou(self._tracks[track_id].detection.box, detection.box))
for track_id in unmatched_tracks
if self._tracks[track_id].detection.category == detection.category
]
track_id, score = max(candidates, key=lambda item: item[1], default=("", -1.0))
if score < self._iou_threshold:
track_id = f"track-{self._next_id:06d}"
self._next_id += 1
self._tracks[track_id] = _Track(track_id, detection)
else:
unmatched_tracks.remove(track_id)
self._tracks[track_id].detection = detection
self._tracks[track_id].missed = 0
results.append(
TrackedObject(
track_id=track_id,
box=detection.box,
category=detection.category,
confidence=detection.confidence,
frame_sequence=frame_sequence,
timestamp_ns=timestamp_ns,
)
)
for track_id in unmatched_tracks:
track = self._tracks[track_id]
track.missed += 1
if track.missed > self._max_missed:
del self._tracks[track_id]
return tuple(results)
def finish(self) -> tuple[str, ...]:
ended = tuple(sorted(self._tracks))
self._tracks.clear()
return ended
+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.
+5
View File
@@ -0,0 +1,5 @@
# Brain vision fixtures
Vision tests create anonymous geometric RGB frames in memory. Never add faces,
customer recordings, biometric templates, camera credentials, or unreviewed
model weights to this directory.
+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,69 @@
from __future__ import annotations
import pytest
from yovision_brain.decode import DecodedFrame
from yovision_brain.vision import (
BoundingBox,
Detection,
LumaBlobDetector,
SingleStreamTracker,
TorchLumaBlobDetector,
)
def frame(payload: bytes, *, sequence: int = 0, width: int = 4, height: int = 3) -> DecodedFrame:
return DecodedFrame(sequence, sequence * 40_000_000, "camera", "main", width, height, "rgb24", payload)
def rgb(values: list[int]) -> bytes:
return b"".join(bytes((value, value, value)) for value in values)
def detection(left: int, top: int, right: int, bottom: int) -> Detection:
return Detection(BoundingBox(left, top, right, bottom), "anonymous_target", 0.9)
def test_detector_emits_only_anonymous_observations() -> None:
payload = rgb([0, 255, 255, 0, 0, 255, 255, 0, 0, 0, 0, 0])
result = LumaBlobDetector(minimum_area=2).detect(frame(payload))
assert result == (Detection(BoundingBox(1, 0, 3, 2), "anonymous_target", 1.0),)
assert LumaBlobDetector.metadata.weights == "none"
assert "external model license" in LumaBlobDetector.metadata.license
def test_empty_frame_has_no_detection() -> None:
assert LumaBlobDetector().detect(frame(rgb([0] * 12))) == ()
def test_tracker_keeps_session_id_across_motion_and_short_occlusion() -> None:
tracker = SingleStreamTracker(iou_threshold=0.1, max_missed=2)
first = tracker.update((detection(0, 0, 3, 3),), frame_sequence=0, timestamp_ns=0)
assert first[0].track_id == "track-000001"
assert tracker.update((), frame_sequence=1, timestamp_ns=1) == ()
resumed = tracker.update((detection(1, 0, 4, 3),), frame_sequence=2, timestamp_ns=2)
assert resumed[0].track_id == "track-000001"
assert tracker.finish() == ("track-000001",)
def test_disappeared_track_ends_and_new_target_gets_new_id() -> None:
tracker = SingleStreamTracker(max_missed=1)
first = tracker.update((detection(0, 0, 2, 2),), frame_sequence=0, timestamp_ns=0)
tracker.update((), frame_sequence=1, timestamp_ns=1)
tracker.update((), frame_sequence=2, timestamp_ns=2)
second = tracker.update((detection(0, 0, 2, 2),), frame_sequence=3, timestamp_ns=3)
assert first[0].track_id == "track-000001"
assert second[0].track_id == "track-000002"
def test_track_ids_are_session_local() -> None:
one = SingleStreamTracker().update((detection(0, 0, 1, 1),), frame_sequence=0, timestamp_ns=0)
two = SingleStreamTracker().update((detection(0, 0, 1, 1),), frame_sequence=0, timestamp_ns=0)
assert one[0].track_id == two[0].track_id == "track-000001"
def test_torch_backend_cpu_smoke_uses_no_external_weights() -> None:
pytest.importorskip("torch")
result = TorchLumaBlobDetector().detect(frame(rgb([0, 255] + [0] * 10)))
assert result[0].category == "anonymous_target"
assert TorchLumaBlobDetector.metadata.weights == "none"
@@ -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'
@@ -22,6 +22,7 @@ func registerSenseDeviceRouter(v1 *gin.RouterGroup, authMiddleware *jwt.GinJWTMi
r.POST("", api.Insert)
r.PUT("/:id", api.Update)
r.PUT("/:id/disable", api.Disable)
r.PUT("/:id/enable", api.Enable)
r.PUT("/:id/credentials", api.UpdateCredentials)
}
}
@@ -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,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/local_event"
"git.ilapage.cn/ila/yovision/Sense/server/common/actions"
"git.ilapage.cn/ila/yovision/Sense/server/common/middleware"
)
func init() { routerCheckRole = append(routerCheckRole, registerSenseLocalEventRouter) }
func registerSenseLocalEventRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
api := &local_event.API{}
r := v1.Group("/local-events").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole()).Use(actions.PermissionAction())
r.GET("", api.List)
r.GET("/:id", api.Get)
}
@@ -0,0 +1,29 @@
package router
import (
"testing"
"github.com/gin-gonic/gin"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
)
func TestLocalEventRouterIsReadOnly(t *testing.T) {
gin.SetMode(gin.TestMode)
engine := gin.New()
registerSenseLocalEventRouter(engine.Group("/api/v1"), &jwt.GinJWTMiddleware{})
routes := make(map[string]struct{})
for _, route := range engine.Routes() {
routes[route.Method+" "+route.Path] = struct{}{}
}
for _, expected := range []string{"GET /api/v1/local-events", "GET /api/v1/local-events/:id"} {
if _, ok := routes[expected]; !ok {
t.Fatalf("missing local event route: %s", expected)
}
}
for route := range routes {
if route != "GET /api/v1/local-events" && route != "GET /api/v1/local-events/:id" {
t.Fatalf("unexpected mutable local event route: %s", route)
}
}
}
@@ -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,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/operations"
"git.ilapage.cn/ila/yovision/Sense/server/common/actions"
"git.ilapage.cn/ila/yovision/Sense/server/common/middleware"
)
func init() { routerCheckRole = append(routerCheckRole, registerSenseOperationsRouter) }
func registerSenseOperationsRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
api := &operations.API{}
r := v1.Group("/operations").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole()).Use(actions.PermissionAction())
r.GET("", api.List)
r.GET("/:id", api.Get)
r.POST("/:id/retry", api.Retry)
}
@@ -0,0 +1,31 @@
package router
import (
"net/http"
"testing"
"github.com/gin-gonic/gin"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
)
func TestSenseOperationsRoutes(t *testing.T) {
gin.SetMode(gin.TestMode)
engine := gin.New()
registerSenseOperationsRouter(engine.Group("/api/v1"), &jwt.GinJWTMiddleware{})
wanted := map[string]bool{
http.MethodGet + " /api/v1/operations": false,
http.MethodGet + " /api/v1/operations/:id": false,
http.MethodPost + " /api/v1/operations/:id/retry": 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,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)
}
}
}
@@ -0,0 +1,24 @@
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/provisioning"
"git.ilapage.cn/ila/yovision/Sense/server/common/actions"
"git.ilapage.cn/ila/yovision/Sense/server/common/middleware"
)
func init() { routerCheckRole = append(routerCheckRole, registerSenseProvisioningRouter) }
func registerSenseProvisioningRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
api := &provisioning.API{}
r := v1.Group("/provisioning/batches").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole()).Use(actions.PermissionAction())
r.GET("", api.List)
r.POST("", api.Create)
r.GET("/:id", api.Get)
r.POST("/:id/execute", api.Execute)
r.POST("/:id/retry-failed", api.RetryFailed)
r.POST("/:id/items/:itemId/retry", api.RetryItem)
r.GET("/:id/export", api.Export)
}
@@ -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/quota"
"git.ilapage.cn/ila/yovision/Sense/server/common/actions"
"git.ilapage.cn/ila/yovision/Sense/server/common/middleware"
)
func init() { routerCheckRole = append(routerCheckRole, registerSenseQuotaRouter) }
func registerSenseQuotaRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
api := &quota.API{}
r := v1.Group("/quota").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole()).Use(actions.PermissionAction())
r.GET("", api.Get)
r.PUT("", api.Update)
}
@@ -14,6 +14,7 @@ import (
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/credential"
deviceService "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/service"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/service/dto"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/quota"
)
type Device struct{ api.Api }
@@ -106,6 +107,25 @@ func (e Device) Disable(c *gin.Context) {
e.OK(response, "设备已停用")
}
func (e Device) Enable(c *gin.Context) {
service := deviceService.Device{}
if err := e.MakeContext(c).MakeOrm().MakeService(&service.Service).Errors; err != nil {
e.Error(http.StatusInternalServerError, err, "服务初始化失败")
return
}
req := dto.EnableReq{ID: c.Param("id"), UpdateBy: user.GetUserId(c)}
if err := bindStrictJSON(c, &req); err != nil {
e.Error(http.StatusBadRequest, err, "请求内容格式不正确")
return
}
var response dto.DeviceResponse
if err := service.Enable(&req, &response); err != nil {
e.writeServiceError(err)
return
}
e.OK(response, "设备已启用,请重新完成视频接入验证")
}
func (e Device) UpdateCredentials(c *gin.Context) {
service := deviceService.Device{}
if err := e.MakeContext(c).MakeOrm().MakeService(&service.Service).Errors; err != nil {
@@ -133,6 +153,10 @@ func (e Device) writeServiceError(err error) {
e.Error(http.StatusNotFound, err, err.Error())
case errors.Is(err, deviceService.ErrVersionConflict):
e.Error(http.StatusConflict, err, err.Error())
case errors.Is(err, quota.ErrExceeded):
e.Error(http.StatusConflict, err, "当前配额已满,无法新增或启用设备")
case errors.Is(err, quota.ErrUnavailable):
e.Error(http.StatusServiceUnavailable, err, "配额配置不可读取,已拒绝新增或启用设备")
case errors.Is(err, credential.ErrKeyUnavailable):
e.Error(http.StatusServiceUnavailable, err, "摄像头凭据安全配置不可用")
default:
@@ -15,6 +15,7 @@ import (
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/credential"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/service/dto"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/quota"
)
var (
@@ -108,7 +109,12 @@ func (e *Device) Insert(req *dto.CreateReq, response *dto.DeviceResponse) error
}
model.CreateBy = req.CreateBy
model.UpdateBy = req.CreateBy
if err = e.Orm.Create(&model).Error; err != nil {
if err = quota.WithAvailableSlot(e.Orm, func(tx *gorm.DB) error {
return tx.Create(&model).Error
}); err != nil {
if errors.Is(err, quota.ErrUnavailable) || errors.Is(err, quota.ErrExceeded) {
return err
}
return fmt.Errorf("create device: %w", err)
}
return e.Get(model.ID, response)
@@ -155,6 +161,35 @@ func (e *Device) Disable(req *dto.DisableReq, response *dto.DeviceResponse) erro
return e.Get(req.ID, response)
}
func (e *Device) Enable(req *dto.EnableReq, response *dto.DeviceResponse) error {
if req.Version < 1 {
return ErrInvalidDevice
}
err := quota.WithAvailableSlot(e.Orm, func(tx *gorm.DB) error {
result := tx.Model(&models.Device{}).
Where("id = ? AND version = ? AND status = ?", req.ID, req.Version, models.StatusDisabled).
Updates(map[string]any{
"status": models.StatusPending,
"version": req.Version + 1, "update_by": req.UpdateBy, "updated_at": time.Now().UTC(),
})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return e.notFoundOrConflictWith(tx, req.ID)
}
return nil
})
if err != nil {
if errors.Is(err, quota.ErrUnavailable) || errors.Is(err, quota.ErrExceeded) ||
errors.Is(err, ErrDeviceNotFound) || errors.Is(err, ErrVersionConflict) {
return err
}
return fmt.Errorf("enable device: %w", err)
}
return e.Get(req.ID, response)
}
func (e *Device) UpdateCredentials(req *dto.CredentialUpdateReq, response *dto.DeviceResponse) error {
if req.Version < 1 || strings.TrimSpace(req.ONVIFUsername) == "" || req.ONVIFPassword == "" {
return ErrInvalidDevice
@@ -13,6 +13,7 @@ import (
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/credential"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/service/dto"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/quota"
)
func testDeviceService(t *testing.T) (*Device, *gorm.DB) {
@@ -21,7 +22,10 @@ func testDeviceService(t *testing.T) (*Device, *gorm.DB) {
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&models.Device{}, &credential.DeviceCredential{}); err != nil {
if err = db.AutoMigrate(&models.Device{}, &credential.DeviceCredential{}, &quota.Setting{}, &quota.Change{}); err != nil {
t.Fatal(err)
}
if err = db.Create(&quota.Setting{ID: quota.SettingID, Limit: 32, Source: "test", Version: 1}).Error; err != nil {
t.Fatal(err)
}
key := make([]byte, 32)
@@ -36,6 +40,38 @@ func testDeviceService(t *testing.T) (*Device, *gorm.DB) {
return service, db
}
func TestCreateAndEnableUseQuotaSafetyGate(t *testing.T) {
service, db := testDeviceService(t)
if err := db.Model(&quota.Setting{}).Where("id = ?", quota.SettingID).Update("limit", 1).Error; err != nil {
t.Fatal(err)
}
var first dto.DeviceResponse
if err := service.Insert(&dto.CreateReq{Name: "第一路", Modality: "video", Capabilities: []string{"video"}}, &first); err != nil {
t.Fatal(err)
}
var rejected dto.DeviceResponse
if err := service.Insert(&dto.CreateReq{Name: "第二路", Modality: "video", Capabilities: []string{"video"}}, &rejected); err != quota.ErrExceeded {
t.Fatalf("second create error=%v", err)
}
var disabled dto.DeviceResponse
if err := service.Disable(&dto.DisableReq{ID: first.ID, Version: first.Version}, &disabled); err != nil {
t.Fatal(err)
}
var second dto.DeviceResponse
if err := service.Insert(&dto.CreateReq{Name: "第二路", Modality: "video", Capabilities: []string{"video"}}, &second); err != nil {
t.Fatal(err)
}
if err := service.Enable(&dto.EnableReq{ID: disabled.ID, Version: disabled.Version}, &rejected); err != quota.ErrExceeded {
t.Fatalf("enable over quota error=%v", err)
}
if err := db.Delete(&quota.Setting{}, quota.SettingID).Error; err != nil {
t.Fatal(err)
}
if err := service.Insert(&dto.CreateReq{Name: "第三路", Modality: "video", Capabilities: []string{"video"}}, &rejected); err != quota.ErrUnavailable {
t.Fatalf("create with unreadable quota error=%v", err)
}
}
func TestDeviceLifecycleUsesAllowlistedFieldsAndOptimisticVersion(t *testing.T) {
service, _ := testDeviceService(t)
var created dto.DeviceResponse
@@ -36,6 +36,12 @@ type DisableReq struct {
UpdateBy int `json:"-"`
}
type EnableReq struct {
ID string `json:"-"`
Version int64 `json:"version"`
UpdateBy int `json:"-"`
}
type CredentialUpdateReq struct {
ID string `json:"-"`
ONVIFUsername string `json:"onvifUsername"`
+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,82 @@
package local_event
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"
"git.ilapage.cn/ila/yovision/Sense/server/common"
)
type API struct{ api.Api }
func (e *API) service(c *gin.Context) (*Service, error) {
service := &Service{}
if err := e.MakeContext(c).MakeOrm().MakeService(&service.Service).Errors; err != nil {
return nil, err
}
return service, 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
}
list, count, err := service.List(request)
if err != nil {
e.audit(c, service, "List", auditFailure, "本地事件列表查询失败")
e.writeError(err)
return
}
e.audit(c, service, "List", auditSuccess, "读取本地事件列表")
e.PageOK(list, int(count), request.GetPageIndex(), request.GetPageSize(), "查询成功")
}
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) {
err := WriteAccessAudit(service.Orm, AccessAudit{
Action: action, Status: status, Username: user.GetUserName(c), UserID: user.GetUserId(c),
ClientIP: common.GetClientIP(c), Route: c.FullPath(), Remark: remark, At: time.Now(),
})
if err != nil {
api.GetRequestLogger(c).Errorf("local event access audit failed: %s", err.Error())
}
}
func (e *API) writeError(err error) {
switch {
case errors.Is(err, ErrInvalidFilter):
e.Error(http.StatusBadRequest, err, ErrInvalidFilter.Error())
case errors.Is(err, ErrNotFound):
e.Error(http.StatusNotFound, err, ErrNotFound.Error())
default:
e.Error(http.StatusInternalServerError, err, "本地事件查询失败")
}
}
@@ -0,0 +1,38 @@
package local_event
import (
"time"
"gorm.io/gorm"
adminModels "git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
)
const (
auditSuccess = "1"
auditFailure = "2"
)
type AccessAudit struct {
Action string
Status string
Username string
UserID int
ClientIP string
Route string
Remark string
At time.Time
}
// WriteAccessAudit persists a minimal, synchronous GoAdmin operation record.
// It deliberately excludes query values, response bodies and evidence data.
func WriteAccessAudit(db *gorm.DB, input AccessAudit) error {
model := adminModels.SysOperaLog{
Title: "本地事件", BusinessType: "query", Method: "local_event.API." + input.Action,
RequestMethod: "GET", 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
}
+36
View File
@@ -0,0 +1,36 @@
package local_event
import (
"time"
commonDTO "git.ilapage.cn/ila/yovision/Sense/server/common/dto"
)
type PageRequest struct {
commonDTO.Pagination `search:"-"`
CandidateState string `form:"candidateState"`
EvidenceState string `form:"evidenceState"`
RuleRef string `form:"ruleRef"`
Keyword string `form:"keyword"`
OccurredAfter string `form:"occurredAfter"`
OccurredBefore string `form:"occurredBefore"`
}
type Response struct {
ID string `json:"id"`
OccurredAt time.Time `json:"occurredAt"`
SourceRef string `json:"sourceRef"`
SourceLabel string `json:"sourceLabel"`
RuleRef string `json:"ruleRef"`
RuleName string `json:"ruleName"`
RuleVersion string `json:"ruleVersion"`
ZoneRef string `json:"zoneRef"`
CandidateState string `json:"candidateState"`
EvidenceState string `json:"evidenceState"`
EvidenceRef string `json:"evidenceRef,omitempty"`
EvidenceDetail string `json:"evidenceDetail,omitempty"`
RetainUntil time.Time `json:"retainUntil"`
RetentionState string `json:"retentionState"`
RemainingDays int `json:"remainingDays"`
CreatedAt time.Time `json:"createdAt"`
}
@@ -0,0 +1,20 @@
package local_event
import (
"time"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
// SeedSyntheticFixture explicitly installs anonymous development/test data.
// Production startup never calls this function.
func SeedSyntheticFixture(db *gorm.DB, now time.Time) error {
now = now.UTC().Truncate(time.Second)
items := []EventCandidate{
{ID: "00000000-0000-4000-8000-000000000001", OccurredAt: now.Add(-18 * time.Minute), SourceRef: "SEN-CAM-03", SourceLabel: "东门通道", RuleRef: "rule-zone-intrusion", RuleName: "区域闯入", RuleVersion: "v3", ZoneRef: "ZONE-EAST-01", CandidateState: CandidateStateCandidate, EvidenceState: EvidenceStateSuccess, EvidenceRef: "local-evidence/fixture-001", EvidenceDetail: "已生成事件截图与短片索引", RetainUntil: now.Add(12 * 24 * time.Hour)},
{ID: "00000000-0000-4000-8000-000000000002", OccurredAt: now.Add(-23 * time.Minute), SourceRef: "SEN-CAM-08", SourceLabel: "仓库北区", RuleRef: "rule-person-stay", RuleName: "人员滞留", RuleVersion: "v2", ZoneRef: "ZONE-WAREHOUSE-02", CandidateState: CandidateStateConfirmed, EvidenceState: EvidenceStatePending, EvidenceDetail: "证据短片正在生成", RetainUntil: now.Add(12 * 24 * time.Hour)},
{ID: "00000000-0000-4000-8000-000000000003", OccurredAt: now.Add(-31 * time.Minute), SourceRef: "SEN-CAM-11", SourceLabel: "设备间入口", RuleRef: "rule-zone-intrusion", RuleName: "区域闯入", RuleVersion: "v3", ZoneRef: "ZONE-EQUIPMENT-01", CandidateState: CandidateStateCandidate, EvidenceState: EvidenceStateFailed, EvidenceDetail: "源视频片段暂不可用", RetainUntil: now.Add(2 * 24 * time.Hour)},
}
return db.Clauses(clause.OnConflict{DoNothing: true}).Create(&items).Error
}
@@ -0,0 +1,38 @@
package local_event
import (
"time"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
)
const (
CandidateStateCandidate = "candidate"
CandidateStateConfirmed = "confirmed"
EvidenceStatePending = "pending"
EvidenceStateSuccess = "success"
EvidenceStateFailed = "failed"
)
// EventCandidate is a Sense-internal, anonymous event candidate. It is not a
// cross-product event contract and does not represent a Bell alert.
type EventCandidate struct {
ID string `gorm:"size:36;primaryKey" json:"id"`
OccurredAt time.Time `gorm:"not null;index" json:"occurredAt"`
SourceRef string `gorm:"size:128;not null;index" json:"sourceRef"`
SourceLabel string `gorm:"size:128;not null;default:''" json:"sourceLabel"`
RuleRef string `gorm:"size:128;not null;index" json:"ruleRef"`
RuleName string `gorm:"size:128;not null" json:"ruleName"`
RuleVersion string `gorm:"size:64;not null;default:''" json:"ruleVersion"`
ZoneRef string `gorm:"size:128;not null;default:''" json:"zoneRef"`
CandidateState string `gorm:"size:32;not null;index" json:"candidateState"`
EvidenceState string `gorm:"size:32;not null;index" json:"evidenceState"`
EvidenceRef string `gorm:"size:512;not null;default:''" json:"evidenceRef,omitempty"`
EvidenceDetail string `gorm:"size:512;not null;default:''" json:"evidenceDetail,omitempty"`
RetainUntil time.Time `gorm:"not null;index" json:"retainUntil"`
common.ControlBy
common.ModelTime
}
func (EventCandidate) TableName() string { return "sense_local_event_candidates" }
@@ -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)
}
}
@@ -0,0 +1,145 @@
package local_event
import (
"errors"
"fmt"
"strings"
"time"
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
"gorm.io/gorm"
)
var (
ErrInvalidFilter = errors.New("本地事件查询条件不符合要求")
ErrNotFound = errors.New("本地事件候选不存在")
)
type Service struct {
coreService.Service
Now func() time.Time
}
func (s *Service) List(request *PageRequest) ([]Response, int64, error) {
filter, err := normalizeFilter(request)
if err != nil {
return nil, 0, err
}
query := s.Orm.Model(&EventCandidate{})
if filter.CandidateState != "" {
query = query.Where("candidate_state = ?", filter.CandidateState)
}
if filter.EvidenceState != "" {
query = query.Where("evidence_state = ?", filter.EvidenceState)
}
if filter.RuleRef != "" {
query = query.Where("rule_ref = ?", filter.RuleRef)
}
if filter.occurredAfter != nil {
query = query.Where("occurred_at >= ?", *filter.occurredAfter)
}
if filter.occurredBefore != nil {
query = query.Where("occurred_at <= ?", *filter.occurredBefore)
}
if filter.Keyword != "" {
pattern := "%" + strings.ToLower(filter.Keyword) + "%"
query = query.Where("LOWER(id) LIKE ? OR LOWER(source_ref) LIKE ? OR LOWER(source_label) LIKE ? OR LOWER(rule_name) LIKE ?", pattern, pattern, pattern, pattern)
}
var count int64
if err = query.Count(&count).Error; err != nil {
return nil, 0, fmt.Errorf("count local event candidates: %w", err)
}
var models []EventCandidate
if err = query.Order("occurred_at DESC, id DESC").Limit(filter.GetPageSize()).Offset((filter.GetPageIndex() - 1) * filter.GetPageSize()).Find(&models).Error; err != nil {
return nil, 0, fmt.Errorf("list local event candidates: %w", err)
}
result := make([]Response, 0, len(models))
for _, model := range models {
result = append(result, s.response(model))
}
return result, count, nil
}
func (s *Service) Get(id string) (Response, error) {
id = strings.TrimSpace(id)
if id == "" || len(id) > 64 {
return Response{}, ErrInvalidFilter
}
var model EventCandidate
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 local event candidate: %w", err)
}
return s.response(model), nil
}
type normalizedFilter struct {
*PageRequest
occurredAfter *time.Time
occurredBefore *time.Time
}
func normalizeFilter(request *PageRequest) (*normalizedFilter, error) {
if request == nil {
request = &PageRequest{}
}
request.CandidateState = strings.TrimSpace(request.CandidateState)
request.EvidenceState = strings.TrimSpace(request.EvidenceState)
request.RuleRef = strings.TrimSpace(request.RuleRef)
request.Keyword = strings.TrimSpace(request.Keyword)
if request.GetPageSize() > 100 || len([]rune(request.Keyword)) > 128 || len(request.RuleRef) > 128 {
return nil, ErrInvalidFilter
}
if request.CandidateState != "" && request.CandidateState != CandidateStateCandidate && request.CandidateState != CandidateStateConfirmed {
return nil, ErrInvalidFilter
}
if request.EvidenceState != "" && request.EvidenceState != EvidenceStatePending && request.EvidenceState != EvidenceStateSuccess && request.EvidenceState != EvidenceStateFailed {
return nil, ErrInvalidFilter
}
result := &normalizedFilter{PageRequest: request}
var err error
if strings.TrimSpace(request.OccurredAfter) != "" {
value, parseErr := time.Parse(time.RFC3339, request.OccurredAfter)
if parseErr != nil {
return nil, ErrInvalidFilter
}
result.occurredAfter = &value
}
if strings.TrimSpace(request.OccurredBefore) != "" {
value, parseErr := time.Parse(time.RFC3339, request.OccurredBefore)
if parseErr != nil {
return nil, ErrInvalidFilter
}
result.occurredBefore = &value
}
if result.occurredAfter != nil && result.occurredBefore != nil && result.occurredAfter.After(*result.occurredBefore) {
err = ErrInvalidFilter
}
if err != nil {
return nil, err
}
return result, nil
}
func (s *Service) response(model EventCandidate) Response {
now := time.Now().UTC()
if s.Now != nil {
now = s.Now().UTC()
}
remaining := int(model.RetainUntil.Sub(now).Hours() / 24)
state := "active"
if !model.RetainUntil.After(now) {
remaining, state = 0, "expired"
} else if remaining < 3 {
state = "expiring"
}
return Response{
ID: model.ID, OccurredAt: model.OccurredAt, SourceRef: model.SourceRef, SourceLabel: model.SourceLabel,
RuleRef: model.RuleRef, RuleName: model.RuleName, RuleVersion: model.RuleVersion, ZoneRef: model.ZoneRef,
CandidateState: model.CandidateState, EvidenceState: model.EvidenceState, EvidenceRef: model.EvidenceRef,
EvidenceDetail: model.EvidenceDetail, RetainUntil: model.RetainUntil, RetentionState: state,
RemainingDays: remaining, CreatedAt: model.CreatedAt,
}
}
@@ -0,0 +1,128 @@
package local_event
import (
"encoding/json"
"strings"
"testing"
"time"
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
"github.com/google/uuid"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
adminModels "git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
commonDTO "git.ilapage.cn/ila/yovision/Sense/server/common/dto"
)
func localEventService(t *testing.T, now time.Time) *Service {
t.Helper()
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
sqlDB, err := db.DB()
if err != nil {
t.Fatal(err)
}
sqlDB.SetMaxOpenConns(1)
if err = db.AutoMigrate(&EventCandidate{}, &adminModels.SysOperaLog{}); err != nil {
t.Fatal(err)
}
return &Service{Service: coreService.Service{Orm: db}, Now: func() time.Time { return now }}
}
func TestSyntheticFixtureSupportsIndependentListFilterAndDetail(t *testing.T) {
now := time.Date(2026, 8, 28, 2, 0, 0, 0, time.UTC)
service := localEventService(t, now)
if err := SeedSyntheticFixture(service.Orm, now); err != nil {
t.Fatal(err)
}
// Seeding is explicit and idempotent; no Bell or Brain service is involved.
if err := SeedSyntheticFixture(service.Orm, now); err != nil {
t.Fatal(err)
}
items, count, err := service.List(&PageRequest{CandidateState: CandidateStateCandidate, EvidenceState: EvidenceStateFailed, Keyword: "SEN-CAM-11"})
if err != nil {
t.Fatal(err)
}
if count != 1 || len(items) != 1 || items[0].RetentionState != "expiring" || items[0].RemainingDays != 2 {
t.Fatalf("unexpected filtered fixture: count=%d items=%#v", count, items)
}
detail, err := service.Get(items[0].ID)
if err != nil {
t.Fatal(err)
}
if detail.EvidenceState != EvidenceStateFailed || detail.CandidateState != CandidateStateCandidate || detail.EvidenceDetail == "" {
t.Fatalf("unexpected detail: %#v", detail)
}
encoded, _ := json.Marshal(detail)
if strings.Contains(strings.ToLower(string(encoded)), "bell") || strings.Contains(strings.ToLower(string(encoded)), "brain") {
t.Fatalf("local response leaked a cross-project schema field: %s", encoded)
}
}
func TestListPaginatesAndOrdersNewestFirst(t *testing.T) {
now := time.Date(2026, 8, 28, 2, 0, 0, 0, time.UTC)
service := localEventService(t, now)
if err := SeedSyntheticFixture(service.Orm, now); err != nil {
t.Fatal(err)
}
first, count, err := service.List(&PageRequest{Pagination: pagination(1, 2)})
if err != nil {
t.Fatal(err)
}
second, _, err := service.List(&PageRequest{Pagination: pagination(2, 2)})
if err != nil {
t.Fatal(err)
}
if count != 3 || len(first) != 2 || len(second) != 1 || !first[0].OccurredAt.After(first[1].OccurredAt) || !first[1].OccurredAt.After(second[0].OccurredAt) {
t.Fatalf("unexpected pagination: first=%#v second=%#v count=%d", first, second, count)
}
}
func TestListRejectsInvalidStatesTimesAndOversizedPages(t *testing.T) {
now := time.Date(2026, 8, 28, 2, 0, 0, 0, time.UTC)
service := localEventService(t, now)
cases := []*PageRequest{
{CandidateState: "delivered"},
{EvidenceState: "unknown"},
{OccurredAfter: "not-a-time"},
{OccurredAfter: now.Format(time.RFC3339), OccurredBefore: now.Add(-time.Hour).Format(time.RFC3339)},
{Pagination: pagination(1, 101)},
}
for _, request := range cases {
if _, _, err := service.List(request); err != ErrInvalidFilter {
t.Fatalf("request %#v: expected ErrInvalidFilter, got %v", request, err)
}
}
}
func TestGetReportsMissingCandidate(t *testing.T) {
service := localEventService(t, time.Now().UTC())
if _, err := service.Get("00000000-0000-4000-8000-000000000099"); err != ErrNotFound {
t.Fatalf("expected ErrNotFound, got %v", err)
}
}
func TestAccessAuditIsSynchronousAndExcludesEventData(t *testing.T) {
now := time.Date(2026, 8, 28, 2, 0, 0, 0, time.UTC)
service := localEventService(t, now)
if err := WriteAccessAudit(service.Orm, AccessAudit{Action: "Get", Status: auditSuccess, Username: "operator", UserID: 7, ClientIP: "127.0.0.1", Route: "/api/v1/local-events/:id", Remark: "读取本地事件详情 fixture-1", At: now}); err != nil {
t.Fatal(err)
}
var stored adminModels.SysOperaLog
if err := service.Orm.First(&stored).Error; err != nil {
t.Fatal(err)
}
if stored.Title != "本地事件" || stored.Status != auditSuccess || stored.CreateBy != 7 || stored.OperUrl != "/api/v1/local-events/:id" {
t.Fatalf("unexpected access audit: %#v", stored)
}
if stored.OperParam != "" || stored.JsonResult != "" {
t.Fatalf("audit persisted query or response data: %#v", stored)
}
}
func pagination(index, size int) commonDTO.Pagination {
return commonDTO.Pagination{PageIndex: index, PageSize: size}
}
+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)
}
}
+107
View File
@@ -0,0 +1,107 @@
package operations
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"
)
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) Retry(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
request := RetryRequest{}
if err = e.MakeContext(c).Bind(&request, binding.JSON).Errors; err != nil {
e.audit(c, service, "Retry", auditFailure, "受控重试请求格式不正确")
e.Error(http.StatusBadRequest, err, "请求格式不正确")
return
}
item, err := service.Retry(c.Request.Context(), c.Param("id"), request.ExpectedVersion, user.GetUserId(c))
if err != nil {
e.audit(c, service, "Retry", auditFailure, "受控重试被拒绝")
e.writeError(err)
return
}
e.audit(c, service, "Retry", auditSuccess, "受控重试已排队 "+item.ID)
e.OK(item, "重试任务已排队")
}
func (e *API) audit(c *gin.Context, service *Service, action, status, remark string) {
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()})
if err != nil {
api.GetRequestLogger(c).Errorf("operations audit failed: %s", err.Error())
}
}
func (e *API) writeError(err error) {
switch {
case errors.Is(err, ErrInvalidFilter):
e.Error(http.StatusBadRequest, err, err.Error())
case errors.Is(err, ErrVersionConflict), errors.Is(err, ErrRetryInProgress), errors.Is(err, ErrRetryNotAllowed):
e.Error(http.StatusConflict, err, err.Error())
case errors.Is(err, ErrProblemNotFound):
e.Error(http.StatusNotFound, err, err.Error())
default:
e.Error(http.StatusInternalServerError, err, "运维中心操作失败")
}
}
@@ -0,0 +1,26 @@
package operations
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: "operations.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
}
+68
View File
@@ -0,0 +1,68 @@
package operations
import (
"time"
commonDto "git.ilapage.cn/ila/yovision/Sense/server/common/dto"
)
const (
ObjectDevice = "device"
ObjectMedia = "media"
ObjectLocalInference = "local_inference"
ProblemAuthentication = "authentication_failed"
ProblemBackoff = "backoff_wait"
ProblemClockDrift = "clock_drift"
ProblemOrphan = "orphan_safety_gate"
ProblemUnavailable = "capability_unavailable"
ProblemUnready = "not_converged"
)
type PageRequest struct {
commonDto.Pagination `search:"-"`
ObjectType string `form:"objectType"`
ProblemType string `form:"problemType"`
Severity string `form:"severity"`
Keyword string `form:"keyword"`
}
type RetryRequest struct {
ExpectedVersion int64 `json:"expectedVersion"`
}
type Summary struct {
ManagedCount int `json:"managedCount"`
ConvergedCount int `json:"convergedCount"`
ActionableProblems int `json:"actionableProblems"`
LocalInference string `json:"localInference"`
}
type Problem struct {
ID string `json:"id"`
ObjectType string `json:"objectType"`
ObjectID string `json:"objectId"`
ObjectName string `json:"objectName"`
Location string `json:"location,omitempty"`
ProblemType string `json:"problemType"`
Severity string `json:"severity"`
Expected string `json:"expected"`
Actual string `json:"actual"`
Difference string `json:"difference"`
NextAction string `json:"nextAction"`
NextRetryAt *time.Time `json:"nextRetryAt,omitempty"`
LastSuccessAt *time.Time `json:"lastSuccessAt,omitempty"`
AttemptCount int `json:"attemptCount"`
Retryable bool `json:"retryable"`
RetryInProgress bool `json:"retryInProgress"`
Version int64 `json:"version"`
SafetyGate string `json:"safetyGate"`
OperationalOnly bool `json:"operationalOnly"`
UpdatedAt time.Time `json:"updatedAt"`
}
type PageResponse struct {
Summary Summary `json:"summary"`
List []Problem `json:"list"`
Count int64 `json:"count"`
}
@@ -0,0 +1,359 @@
package operations
import (
"context"
"errors"
"fmt"
"sort"
"strings"
"time"
"gorm.io/gorm"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/admission"
deviceModels "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media"
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
)
var (
ErrInvalidFilter = errors.New("运维中心查询条件不符合要求")
ErrProblemNotFound = errors.New("运维问题不存在或已经收敛")
ErrVersionConflict = errors.New("状态已经变化,请刷新后重试")
ErrRetryInProgress = errors.New("该对象已有重试任务")
ErrRetryNotAllowed = errors.New("该问题不允许重试")
)
type Service struct {
DB *gorm.DB
Now func() time.Time
LocalInferenceConfigured bool
DeviceRetry func(context.Context, admission.ProbeRequest) error
}
type admissionResult struct {
DeviceID string
Address string
Status string
Detail string
CheckedAt time.Time
}
func (admissionResult) TableName() string { return "sense_admission_results" }
func NewService(db *gorm.DB) *Service {
service := &Service{DB: db, Now: time.Now, LocalInferenceConfigured: false}
service.DeviceRetry = func(ctx context.Context, request admission.ProbeRequest) error {
base := coreService.Service{}
base.Orm = db
runtimeService, err := admission.NewRuntime(base)
if err != nil {
return err
}
_, err = runtimeService.Probe(ctx, request)
if err == nil {
// Keep the existing admission boundary: route intent is best-effort and
// must not turn a successful device probe into a MediaMTX failure.
_ = media.EnsureDeviceRoutes(ctx, db, request.DeviceID)
}
return err
}
return service
}
func (s *Service) List(request PageRequest) (PageResponse, error) {
if err := validateFilter(request); err != nil {
return PageResponse{}, err
}
problems, summary, err := s.project()
if err != nil {
return PageResponse{}, err
}
filtered := make([]Problem, 0, len(problems))
keyword := strings.ToLower(strings.TrimSpace(request.Keyword))
for _, item := range problems {
if request.ObjectType != "" && item.ObjectType != request.ObjectType {
continue
}
if request.ProblemType != "" && item.ProblemType != request.ProblemType {
continue
}
if request.Severity != "" && item.Severity != request.Severity {
continue
}
searchable := strings.ToLower(strings.Join([]string{item.ObjectID, item.ObjectName, item.Location, item.Difference}, " "))
if keyword != "" && !strings.Contains(searchable, keyword) {
continue
}
filtered = append(filtered, item)
}
pageIndex, pageSize := request.GetPageIndex(), request.GetPageSize()
if pageSize > 100 {
pageSize = 100
}
start := (pageIndex - 1) * pageSize
if start > len(filtered) {
start = len(filtered)
}
end := start + pageSize
if end > len(filtered) {
end = len(filtered)
}
return PageResponse{Summary: summary, List: filtered[start:end], Count: int64(len(filtered))}, nil
}
func (s *Service) Get(id string) (Problem, error) {
problems, _, err := s.project()
if err != nil {
return Problem{}, err
}
for _, item := range problems {
if item.ID == id {
return item, nil
}
}
return Problem{}, ErrProblemNotFound
}
func (s *Service) Retry(ctx context.Context, id string, expectedVersion int64, userID int) (Problem, error) {
if expectedVersion < 1 {
return Problem{}, ErrVersionConflict
}
problem, err := s.Get(id)
if err != nil {
return Problem{}, err
}
if !problem.Retryable {
if problem.RetryInProgress {
return Problem{}, ErrRetryInProgress
}
return Problem{}, ErrRetryNotAllowed
}
now := s.now().UTC()
switch problem.ObjectType {
case ObjectDevice:
var admissionState admissionResult
if err := s.DB.First(&admissionState, "device_id = ?", problem.ObjectID).Error; err != nil || strings.TrimSpace(admissionState.Address) == "" {
return Problem{}, ErrRetryNotAllowed
}
result := s.DB.Model(&deviceModels.Device{}).
Where("id = ? AND version = ? AND retry_requested_at IS NULL", problem.ObjectID, expectedVersion).
Updates(map[string]any{"retry_requested_at": now, "version": expectedVersion + 1, "updated_at": now})
if result.Error != nil {
return Problem{}, fmt.Errorf("queue device retry: %w", result.Error)
}
if result.RowsAffected == 0 {
return Problem{}, s.retryConflict(ObjectDevice, problem.ObjectID, expectedVersion)
}
if s.DeviceRetry != nil {
err = s.DeviceRetry(ctx, admission.ProbeRequest{DeviceID: problem.ObjectID, Address: admissionState.Address, Version: expectedVersion + 1, UpdateBy: userID})
if err != nil {
clearErr := s.DB.Model(&deviceModels.Device{}).Where("id = ? AND version = ?", problem.ObjectID, expectedVersion+1).Update("retry_requested_at", nil).Error
if clearErr != nil {
return Problem{}, errors.Join(fmt.Errorf("execute device retry: %w", err), fmt.Errorf("clear device retry gate: %w", clearErr))
}
return Problem{}, fmt.Errorf("execute device retry: %w", err)
}
}
case ObjectMedia:
result := s.DB.Model(&media.Route{}).
Where("id = ? AND version = ? AND actual <> ?", problem.ObjectID, expectedVersion, "retry_pending").
Updates(map[string]any{"actual": "retry_pending", "next_retry_at": now, "detail": "已请求受控重试,等待视频服务执行", "version": expectedVersion + 1, "updated_at": now})
if result.Error != nil {
return Problem{}, fmt.Errorf("queue media retry: %w", result.Error)
}
if result.RowsAffected == 0 {
return Problem{}, s.retryConflict(ObjectMedia, problem.ObjectID, expectedVersion)
}
default:
return Problem{}, ErrRetryNotAllowed
}
updated, err := s.Get(id)
if errors.Is(err, ErrProblemNotFound) {
problem.Actual, problem.Difference, problem.NextAction = "converged", "重试完成,状态已经收敛", "无需处理"
problem.Retryable, problem.RetryInProgress, problem.Version, problem.UpdatedAt = false, false, expectedVersion+2, now
return problem, nil
}
return updated, err
}
func (s *Service) project() ([]Problem, Summary, error) {
var devices []deviceModels.Device
if err := s.DB.Order("updated_at DESC").Find(&devices).Error; err != nil {
return nil, Summary{}, fmt.Errorf("list operation devices: %w", err)
}
var admissions []admissionResult
if err := s.DB.Find(&admissions).Error; err != nil {
return nil, Summary{}, fmt.Errorf("list admission results: %w", err)
}
var routes []media.Route
if err := s.DB.Order("updated_at DESC").Find(&routes).Error; err != nil {
return nil, Summary{}, fmt.Errorf("list operation media routes: %w", err)
}
admissionByDevice := make(map[string]admissionResult, len(admissions))
for _, item := range admissions {
admissionByDevice[item.DeviceID] = item
}
deviceByID := make(map[string]deviceModels.Device, len(devices))
problems := make([]Problem, 0)
summary := Summary{ManagedCount: len(devices) + len(routes), LocalInference: "configured"}
for _, device := range devices {
deviceByID[device.ID] = device
if problem, ok := deviceProblem(device, admissionByDevice[device.ID]); ok {
problems = append(problems, problem)
summary.ActionableProblems++
} else {
summary.ConvergedCount++
}
}
for _, route := range routes {
if problem, ok := mediaProblem(route, deviceByID); ok {
problems = append(problems, problem)
summary.ActionableProblems++
} else {
summary.ConvergedCount++
}
}
if !s.LocalInferenceConfigured {
summary.LocalInference = "unavailable"
now := s.now().UTC()
problems = append(problems, Problem{
ID: "local_inference:adapter", ObjectType: ObjectLocalInference, ObjectID: "adapter",
ObjectName: "本地推理适配器", ProblemType: ProblemUnavailable, Severity: "info",
Expected: "可选", Actual: "unavailable", Difference: "未配置本地推理适配器;不影响设备和媒体运维",
NextAction: "无需处理", SafetyGate: "不读取 Brain 数据库,不阻断本页", OperationalOnly: true, UpdatedAt: now,
})
}
sort.SliceStable(problems, func(i, j int) bool {
rank := map[string]int{"high": 0, "medium": 1, "info": 2}
if rank[problems[i].Severity] != rank[problems[j].Severity] {
return rank[problems[i].Severity] < rank[problems[j].Severity]
}
return problems[i].UpdatedAt.After(problems[j].UpdatedAt)
})
return problems, summary, nil
}
func deviceProblem(device deviceModels.Device, admission admissionResult) (Problem, bool) {
if device.Status == deviceModels.StatusDisabled {
return Problem{}, false
}
actual, detail, updatedAt := device.AdapterStatus, "设备尚未完成接入验证", device.UpdatedAt
if admission.DeviceID != "" {
actual, detail, updatedAt = admission.Status, admission.Detail, admission.CheckedAt
}
if device.Status == deviceModels.StatusActive && (actual == "ready" || actual == "verified") {
return Problem{}, false
}
problemType, severity, nextAction := ProblemUnready, "medium", "检查设备和接入配置后重试"
text := strings.ToLower(actual + " " + detail)
if strings.Contains(text, "auth") || strings.Contains(text, "认证") || strings.Contains(text, "凭据") {
problemType, severity, nextAction = ProblemAuthentication, "high", "确认设备账号未变更后执行受控重试"
} else if strings.Contains(text, "clock") || strings.Contains(text, "time drift") || strings.Contains(text, "时间漂移") || strings.Contains(text, "时钟") {
problemType, nextAction = ProblemClockDrift, "检查设备 NTP 和时区后重新检测"
}
retrying := device.RetryRequestedAt != nil
retryable := !retrying && strings.TrimSpace(admission.Address) != ""
if admission.DeviceID == "" || strings.TrimSpace(admission.Address) == "" {
nextAction = "先到视频接入完成地址与凭据验证"
}
return Problem{
ID: "device:" + device.ID, ObjectType: ObjectDevice, ObjectID: device.ID, ObjectName: device.Name,
Location: device.Location, ProblemType: problemType, Severity: severity, Expected: "active / ready", Actual: actual,
Difference: detail, NextAction: nextAction, AttemptCount: boolInt(retrying), Retryable: retryable,
RetryInProgress: retrying, Version: device.Version, SafetyGate: "校验设备版本且同一设备仅允许一个在途重试",
OperationalOnly: true, UpdatedAt: updatedAt,
}, true
}
func mediaProblem(route media.Route, devices map[string]deviceModels.Device) (Problem, bool) {
device, exists := devices[route.DeviceID]
name, location := route.Path, ""
if exists {
name, location = device.Name+" / "+route.Path, device.Location
}
if !exists {
return Problem{
ID: "media:" + route.ID, ObjectType: ObjectMedia, ObjectID: route.ID, ObjectName: name,
ProblemType: ProblemOrphan, Severity: "medium", Expected: "路由关联有效设备", Actual: "已隔离待确认",
Difference: "媒体路由存在,但找不到有效设备归属", NextAction: "人工核对;不会自动删除", Version: route.Version,
SafetyGate: "孤儿资源仅隔离和提示,本接口没有删除动作", OperationalOnly: true, UpdatedAt: route.UpdatedAt,
}, true
}
converged := (route.Desired == media.DesiredRunning && (route.Actual == "ready" || route.Actual == "waiting")) || (route.Desired == media.DesiredStopped && route.Actual == "stopped")
if converged {
return Problem{}, false
}
problemType, nextAction := ProblemUnready, "检查视频服务状态后重试"
if route.NextRetryAt != nil {
problemType, nextAction = ProblemBackoff, "等待退避到期或执行受控提前重试"
}
retrying := route.Actual == "retry_pending"
return Problem{
ID: "media:" + route.ID, ObjectType: ObjectMedia, ObjectID: route.ID, ObjectName: name, Location: location,
ProblemType: problemType, Severity: "medium", Expected: route.Desired, Actual: route.Actual, Difference: route.Detail,
NextAction: nextAction, NextRetryAt: route.NextRetryAt, AttemptCount: route.FailureCount, Retryable: !retrying,
RetryInProgress: retrying, Version: route.Version, SafetyGate: "校验路由版本且只排队,不删除路径或修改凭据",
OperationalOnly: true, UpdatedAt: route.UpdatedAt,
}, true
}
func (s *Service) retryConflict(objectType, objectID string, expectedVersion int64) error {
var version int64
var inProgress bool
switch objectType {
case ObjectDevice:
var item deviceModels.Device
if err := s.DB.Select("version", "retry_requested_at").First(&item, "id = ?", objectID).Error; err != nil {
return ErrProblemNotFound
}
version, inProgress = item.Version, item.RetryRequestedAt != nil
case ObjectMedia:
var item media.Route
if err := s.DB.Select("version", "actual").First(&item, "id = ?", objectID).Error; err != nil {
return ErrProblemNotFound
}
version, inProgress = item.Version, item.Actual == "retry_pending"
}
if inProgress {
return ErrRetryInProgress
}
if version != expectedVersion {
return ErrVersionConflict
}
return ErrVersionConflict
}
func (s *Service) now() time.Time {
if s.Now != nil {
return s.Now()
}
return time.Now()
}
func validateFilter(request PageRequest) error {
valid := func(value string, values ...string) bool {
if value == "" {
return true
}
for _, candidate := range values {
if value == candidate {
return true
}
}
return false
}
if !valid(request.ObjectType, ObjectDevice, ObjectMedia, ObjectLocalInference) ||
!valid(request.ProblemType, ProblemAuthentication, ProblemBackoff, ProblemClockDrift, ProblemOrphan, ProblemUnavailable, ProblemUnready) ||
!valid(request.Severity, "high", "medium", "info") || len([]rune(request.Keyword)) > 128 {
return ErrInvalidFilter
}
return nil
}
func boolInt(value bool) int {
if value {
return 1
}
return 0
}
@@ -0,0 +1,187 @@
package operations
import (
"context"
"errors"
"testing"
"time"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
adminModels "git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/admission"
deviceModels "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media"
)
func operationsTestService(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(&deviceModels.Device{}, &admissionResult{}, &media.Route{}, &adminModels.SysOperaLog{}); err != nil {
t.Fatal(err)
}
now := time.Date(2026, 8, 28, 2, 0, 0, 0, time.UTC)
service := NewService(db)
service.Now = func() time.Time { return now }
service.LocalInferenceConfigured = false
service.DeviceRetry = nil
return service, db, now
}
func TestProjectionCoversOperationsStatesAndIndependentBoundaries(t *testing.T) {
service, db, now := operationsTestService(t)
devices := []deviceModels.Device{
{ID: "ready", Name: "东门", Location: "一层", Modality: "video", CapabilitiesJSON: "[]", Status: "active", AdapterStatus: "ready", Version: 1},
{ID: "auth", Name: "仓库", Location: "北区", Modality: "video", CapabilitiesJSON: "[]", Status: "active", AdapterStatus: "verification_failed", Version: 3},
{ID: "clock", Name: "南门", Location: "室外", Modality: "video", CapabilitiesJSON: "[]", Status: "active", AdapterStatus: "verification_failed", Version: 4},
}
for index := range devices {
devices[index].CreatedAt, devices[index].UpdatedAt = now, now
}
if err := db.Create(&devices).Error; err != nil {
t.Fatal(err)
}
admissions := []admissionResult{
{DeviceID: "ready", Address: "http://192.0.2.10/onvif", Status: "ready", Detail: "接入验证完成", CheckedAt: now},
{DeviceID: "auth", Address: "http://192.0.2.11/onvif", Status: "authentication_failed", Detail: "ONVIF 认证失败", CheckedAt: now},
{DeviceID: "clock", Address: "http://192.0.2.12/onvif", Status: "clock_drift", Detail: "设备时间漂移 96 秒", CheckedAt: now},
}
if err := db.Create(&admissions).Error; err != nil {
t.Fatal(err)
}
next := now.Add(5 * time.Minute)
routes := []media.Route{
{ID: "auth:main", DeviceID: "auth", ProfileToken: "main", Path: "sense_auth", Desired: "running", Actual: "apply_failed", Detail: "媒体路径配置失败", FailureCount: 2, NextRetryAt: &next, Version: 5, UpdatedAt: now},
{ID: "missing:main", DeviceID: "missing", ProfileToken: "main", Path: "sense_orphan", Desired: "running", Actual: "ready", Detail: "上游拉流正常", Version: 1, UpdatedAt: now},
}
if err := db.Create(&routes).Error; err != nil {
t.Fatal(err)
}
response, err := service.List(PageRequest{})
if err != nil {
t.Fatal(err)
}
if response.Summary.ManagedCount != 5 || response.Summary.ConvergedCount != 1 || response.Summary.ActionableProblems != 4 || response.Summary.LocalInference != "unavailable" || response.Count != 5 {
t.Fatalf("unexpected summary: %#v count=%d", response.Summary, response.Count)
}
wanted := map[string]bool{ProblemAuthentication: false, ProblemClockDrift: false, ProblemBackoff: false, ProblemOrphan: false, ProblemUnavailable: false}
for _, item := range response.List {
wanted[item.ProblemType] = true
if !item.OperationalOnly {
t.Fatalf("problem can be mistaken for Bell alert: %#v", item)
}
}
for state, found := range wanted {
if !found {
t.Fatalf("missing problem state %s: %#v", state, response.List)
}
}
filtered, err := service.List(PageRequest{ObjectType: ObjectDevice, ProblemType: ProblemClockDrift, Keyword: "南门"})
if err != nil || filtered.Count != 1 || filtered.List[0].ObjectID != "clock" {
t.Fatalf("filtered=%#v err=%v", filtered, err)
}
}
func TestControlledRetryUsesVersionAndInProgressGate(t *testing.T) {
service, db, now := operationsTestService(t)
device := deviceModels.Device{ID: "auth", Name: "东门", Modality: "video", CapabilitiesJSON: "[]", Status: "active", AdapterStatus: "verification_failed", Version: 3}
device.CreatedAt, device.UpdatedAt = now, now
if err := db.Create(&device).Error; err != nil {
t.Fatal(err)
}
if err := db.Create(&admissionResult{DeviceID: "auth", Address: "http://192.0.2.11/onvif", Status: "authentication_failed", Detail: "认证失败", CheckedAt: now}).Error; err != nil {
t.Fatal(err)
}
item, err := service.Retry(context.Background(), "device:auth", 3, 7)
if err != nil || !item.RetryInProgress || item.Retryable || item.Version != 4 {
t.Fatalf("item=%#v err=%v", item, err)
}
if _, err = service.Retry(context.Background(), "device:auth", 3, 7); !errors.Is(err, ErrRetryInProgress) {
t.Fatalf("expected in-progress gate, got %v", err)
}
var stored deviceModels.Device
if err = db.First(&stored, "id = ?", "auth").Error; err != nil || stored.RetryRequestedAt == nil {
t.Fatalf("stored=%#v err=%v", stored, err)
}
}
func TestMediaRetryQueuesWithoutDeletingOrChangingCredentials(t *testing.T) {
service, db, now := operationsTestService(t)
device := deviceModels.Device{ID: "camera", Name: "仓库", Modality: "video", CapabilitiesJSON: "[]", Status: "active", AdapterStatus: "ready", Version: 1}
device.CreatedAt, device.UpdatedAt = now, now
if err := db.Create(&device).Error; err != nil {
t.Fatal(err)
}
if err := db.Create(&admissionResult{DeviceID: "camera", Address: "http://192.0.2.20/onvif", Status: "ready", Detail: "接入完成", CheckedAt: now}).Error; err != nil {
t.Fatal(err)
}
next := now.Add(time.Minute)
route := media.Route{ID: "camera:main", DeviceID: "camera", ProfileToken: "main", Path: "sense_camera", Desired: "running", Actual: "apply_failed", FailureCount: 2, NextRetryAt: &next, Detail: "配置失败", Version: 6, UpdatedAt: now}
if err := db.Create(&route).Error; err != nil {
t.Fatal(err)
}
item, err := service.Retry(context.Background(), "media:camera:main", 6, 7)
if err != nil || item.Actual != "retry_pending" || item.Version != 7 || !item.RetryInProgress {
t.Fatalf("item=%#v err=%v", item, err)
}
var count int64
if err = db.Model(&media.Route{}).Where("id = ?", route.ID).Count(&count).Error; err != nil || count != 1 {
t.Fatalf("route was removed: count=%d err=%v", count, err)
}
}
func TestOrphanAndUnavailableCannotRetry(t *testing.T) {
service, db, now := operationsTestService(t)
route := media.Route{ID: "missing:main", DeviceID: "missing", ProfileToken: "main", Path: "orphan", Desired: "running", Actual: "ready", Version: 1, UpdatedAt: now}
if err := db.Create(&route).Error; err != nil {
t.Fatal(err)
}
if _, err := service.Retry(context.Background(), "media:missing:main", 1, 7); !errors.Is(err, ErrRetryNotAllowed) {
t.Fatalf("orphan retry should be rejected: %v", err)
}
if _, err := service.Retry(context.Background(), "local_inference:adapter", 1, 7); !errors.Is(err, ErrRetryNotAllowed) {
t.Fatalf("optional adapter retry should be rejected: %v", err)
}
}
func TestDeviceRetryUsesExistingAdmissionPortAndOperatorIdentity(t *testing.T) {
service, db, now := operationsTestService(t)
device := deviceModels.Device{ID: "clock", Name: "南门", Modality: "video", CapabilitiesJSON: "[]", Status: "active", AdapterStatus: "verification_failed", Version: 3}
device.CreatedAt, device.UpdatedAt = now, now
if err := db.Create(&device).Error; err != nil {
t.Fatal(err)
}
if err := db.Create(&admissionResult{DeviceID: "clock", Address: "http://192.0.2.12/onvif", Status: "clock_drift", Detail: "设备时间漂移", CheckedAt: now}).Error; err != nil {
t.Fatal(err)
}
called := false
service.DeviceRetry = func(_ context.Context, request admission.ProbeRequest) error {
called = true
if request.DeviceID != "clock" || request.Address != "http://192.0.2.12/onvif" || request.Version != 4 || request.UpdateBy != 9 {
t.Fatalf("unexpected retry request: %#v", request)
}
return db.Model(&deviceModels.Device{}).Where("id = ? AND version = ?", request.DeviceID, request.Version).Updates(map[string]any{"retry_requested_at": nil, "version": request.Version + 1}).Error
}
item, err := service.Retry(context.Background(), "device:clock", 3, 9)
if err != nil || !called || item.Version != 5 || item.RetryInProgress {
t.Fatalf("item=%#v called=%v err=%v", item, called, err)
}
}
func TestAuditIsMinimalAndDesensitized(t *testing.T) {
_, db, now := operationsTestService(t)
if err := WriteAudit(db, Audit{Action: "Retry", Method: "POST", Status: "1", Username: "operator", UserID: 7, ClientIP: "127.0.0.1", Route: "/api/v1/operations/:id/retry", Remark: "受控重试已排队 device:synthetic", At: now}); err != nil {
t.Fatal(err)
}
var stored adminModels.SysOperaLog
if err := db.First(&stored).Error; err != nil {
t.Fatal(err)
}
if stored.Title != "运维中心" || stored.RequestMethod != "POST" || stored.OperParam != "" || stored.JsonResult != "" || stored.CreateBy != 7 {
t.Fatalf("unexpected audit: %#v", stored)
}
}
+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
}
+176
View File
@@ -0,0 +1,176 @@
package provisioning
import (
"encoding/csv"
"encoding/json"
"errors"
"io"
"net/http"
"strconv"
"strings"
"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"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/quota"
)
type API struct{ api.Api }
func (e *API) service(c *gin.Context) (*Service, error) {
service := &Service{}
if err := e.MakeContext(c).MakeOrm().MakeService(&service.Service).Errors; err != nil {
return nil, err
}
return service, nil
}
func (e *API) List(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
request := BatchPageRequest{}
if err = e.MakeContext(c).Bind(&request).Errors; err != nil {
e.Error(http.StatusBadRequest, err, "查询条件格式不正确")
return
}
list, count, err := service.ListBatches(&request)
if err != nil {
e.writeError(err)
return
}
e.PageOK(list, int(count), request.GetPageIndex(), request.GetPageSize(), "查询成功")
}
func (e *API) Get(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
response, err := service.GetBatch(c.Param("id"))
if err != nil {
e.writeError(err)
return
}
e.OK(response, "查询成功")
}
func (e *API) Create(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
request := CreateBatchRequest{CreateBy: user.GetUserId(c)}
if err = bindStrictJSON(c, &request); err != nil {
e.Error(http.StatusBadRequest, err, "导入内容格式不正确")
return
}
response, err := service.CreateBatch(request)
if err != nil {
e.writeError(err)
return
}
e.OK(response, "批次已导入并完成预校验")
}
func (e *API) Execute(c *gin.Context) { e.execute(c, false, "") }
func (e *API) RetryFailed(c *gin.Context) { e.execute(c, true, "") }
func (e *API) RetryItem(c *gin.Context) { e.execute(c, true, c.Param("itemId")) }
func (e *API) execute(c *gin.Context, retryFailed bool, itemID string) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
request := ExecuteRequest{UpdateBy: user.GetUserId(c)}
defer func() { clearCredentials(request.Credentials) }()
if err = bindStrictJSON(c, &request); err != nil {
e.Error(http.StatusBadRequest, err, "执行内容格式不正确")
return
}
response, err := service.Execute(c.Request.Context(), c.Param("id"), itemID, retryFailed, request)
if err != nil {
e.writeError(err)
return
}
e.OK(response, "批量开通处理完成")
}
func (e *API) Export(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
batch, err := service.GetBatch(c.Param("id"))
if err != nil {
e.writeError(err)
return
}
c.Header("Content-Type", "text/csv; charset=utf-8")
c.Header("Content-Disposition", "attachment; filename=provisioning-"+batch.ID+".csv")
c.Status(http.StatusOK)
writer := csv.NewWriter(c.Writer)
if err = writer.Write([]string{"line_number", "name", "location", "address", "status", "failure_code", "detail", "device_id"}); err != nil {
e.Logger.Error(err)
return
}
for _, item := range batch.Items {
if err = writer.Write([]string{strconv.Itoa(item.LineNumber), item.Name, item.Location, item.Address, item.Status, item.FailureCode, item.Detail, item.DeviceID}); err != nil {
e.Logger.Error(err)
return
}
}
writer.Flush()
if err = writer.Error(); err != nil {
e.Logger.Error(err)
}
}
func (e *API) writeError(err error) {
switch {
case errors.Is(err, ErrInvalidRequest):
e.Error(http.StatusBadRequest, err, err.Error())
case errors.Is(err, ErrBatchNotFound), errors.Is(err, ErrItemNotFound):
e.Error(http.StatusNotFound, err, err.Error())
case errors.Is(err, quota.ErrExceeded):
e.Error(http.StatusConflict, err, "当前配额已满,无法继续批量开通")
case errors.Is(err, quota.ErrUnavailable):
e.Error(http.StatusServiceUnavailable, err, "配额配置不可读取,已拒绝批量开通写入")
default:
e.Error(http.StatusInternalServerError, err, "批量开通操作失败")
}
}
func bindStrictJSON(c *gin.Context, target any) error {
if !strings.HasPrefix(strings.ToLower(strings.TrimSpace(c.GetHeader("Content-Type"))), "application/json") {
return errors.New("content type must be application/json")
}
decoder := json.NewDecoder(http.MaxBytesReader(c.Writer, c.Request.Body, 2<<20))
decoder.DisallowUnknownFields()
if err := decoder.Decode(target); err != nil {
return err
}
if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
if err == nil {
return errors.New("request body must contain one JSON object")
}
return err
}
return nil
}
func clearCredentials(values []CredentialInput) {
for index := range values {
values[index].ONVIFUsername = ""
values[index].ONVIFPassword = ""
values[index].RTSPUsername = ""
values[index].RTSPPassword = ""
}
}
@@ -0,0 +1,32 @@
package provisioning
import (
"net/http/httptest"
"strings"
"testing"
"github.com/gin-gonic/gin"
)
func TestCreateRequestRejectsCredentialColumns(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
context, _ := gin.CreateTestContext(recorder)
context.Request = httptest.NewRequest("POST", "/api/v1/provisioning/batches", strings.NewReader(`{
"idempotencyKey":"import-1",
"rows":[{"lineNumber":1,"name":"东门摄像机","location":"东门","address":"http://192.0.2.10/onvif","password":"must-not-be-accepted"}]
}`))
context.Request.Header.Set("Content-Type", "application/json")
var request CreateBatchRequest
if err := bindStrictJSON(context, &request); err == nil || !strings.Contains(err.Error(), "unknown field") {
t.Fatalf("expected unknown credential field to be rejected, got %v", err)
}
}
func TestClearCredentialsOverwritesTransientValues(t *testing.T) {
values := []CredentialInput{{ONVIFUsername: "installer", ONVIFPassword: "temporary-secret", RTSPUsername: "stream", RTSPPassword: "stream-secret"}}
clearCredentials(values)
if values[0].ONVIFUsername != "" || values[0].ONVIFPassword != "" || values[0].RTSPUsername != "" || values[0].RTSPPassword != "" {
t.Fatalf("credentials were not cleared: %#v", values[0])
}
}
@@ -0,0 +1,68 @@
package provisioning
import (
"time"
commonDTO "git.ilapage.cn/ila/yovision/Sense/server/common/dto"
)
type BatchPageRequest struct {
commonDTO.Pagination `search:"-"`
Status string `form:"status"`
}
type ImportRow struct {
LineNumber int `json:"lineNumber"`
Name string `json:"name"`
Location string `json:"location"`
Address string `json:"address"`
}
type CreateBatchRequest struct {
IdempotencyKey string `json:"idempotencyKey"`
Rows []ImportRow `json:"rows"`
CreateBy int `json:"-"`
}
type CredentialInput struct {
ItemID string `json:"itemId"`
ONVIFUsername string `json:"onvifUsername"`
ONVIFPassword string `json:"onvifPassword"`
RTSPSameAsONVIF bool `json:"rtspSameAsOnvif"`
RTSPUsername string `json:"rtspUsername"`
RTSPPassword string `json:"rtspPassword"`
}
type ExecuteRequest struct {
Credentials []CredentialInput `json:"credentials"`
UpdateBy int `json:"-"`
}
type ItemResponse struct {
ID string `json:"id"`
LineNumber int `json:"lineNumber"`
Name string `json:"name"`
Location string `json:"location"`
Address string `json:"address"`
Status string `json:"status"`
FailureCode string `json:"failureCode,omitempty"`
Detail string `json:"detail,omitempty"`
DeviceID string `json:"deviceId,omitempty"`
Attempts int `json:"attempts"`
LastTriedAt *time.Time `json:"lastTriedAt,omitempty"`
}
type BatchResponse struct {
ID string `json:"id"`
IdempotencyKey string `json:"idempotencyKey"`
Status string `json:"status"`
QuotaLimit int `json:"quotaLimit"`
ExistingCount int `json:"existingCount"`
TotalCount int `json:"totalCount"`
ReadyCount int `json:"readyCount"`
SuccessCount int `json:"successCount"`
FailureCount int `json:"failureCount"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
Items []ItemResponse `json:"items,omitempty"`
}
@@ -0,0 +1,58 @@
package provisioning
import (
"time"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
)
const (
BatchReady = "ready"
BatchRunning = "running"
BatchSucceeded = "succeeded"
BatchPartial = "partial"
BatchFailed = "failed"
ItemInvalid = "invalid"
ItemReady = "ready"
ItemQuotaExceeded = "quota_exceeded"
ItemRunning = "running"
ItemSucceeded = "succeeded"
ItemFailed = "failed"
)
type Batch struct {
ID string `gorm:"size:36;primaryKey" json:"id"`
IdempotencyKey string `gorm:"size:128;not null;uniqueIndex" json:"idempotencyKey"`
Status string `gorm:"size:32;not null;index" json:"status"`
QuotaLimit int `gorm:"not null" json:"quotaLimit"`
ExistingCount int `gorm:"not null" json:"existingCount"`
TotalCount int `gorm:"not null" json:"totalCount"`
ReadyCount int `gorm:"not null" json:"readyCount"`
SuccessCount int `gorm:"not null" json:"successCount"`
FailureCount int `gorm:"not null" json:"failureCount"`
common.ControlBy
common.ModelTime
Items []Item `gorm:"foreignKey:BatchID" json:"items,omitempty"`
}
func (Batch) TableName() string { return "sense_provisioning_batches" }
type Item struct {
ID string `gorm:"size:36;primaryKey" json:"id"`
BatchID string `gorm:"size:36;not null;uniqueIndex:batch_line;index" json:"batchId"`
LineNumber int `gorm:"not null;uniqueIndex:batch_line" json:"lineNumber"`
Name string `gorm:"size:128;not null" json:"name"`
Location string `gorm:"size:255;not null;default:''" json:"location"`
Address string `gorm:"size:1024;not null" json:"address"`
Status string `gorm:"size:32;not null;index" json:"status"`
FailureCode string `gorm:"size:64;not null;default:''" json:"failureCode,omitempty"`
Detail string `gorm:"size:512;not null;default:''" json:"detail,omitempty"`
DeviceID string `gorm:"size:36;not null;default:'';index" json:"deviceId,omitempty"`
Attempts int `gorm:"not null;default:0" json:"attempts"`
LastTriedAt *time.Time `json:"lastTriedAt,omitempty"`
common.ControlBy
common.ModelTime
}
func (Item) TableName() string { return "sense_provisioning_items" }
@@ -0,0 +1,360 @@
package provisioning
import (
"context"
"errors"
"fmt"
"net/url"
"sort"
"strings"
"time"
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgconn"
"gorm.io/gorm"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/admission"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/credential"
deviceService "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/service"
deviceDTO "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/service/dto"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/quota"
)
var (
ErrInvalidRequest = errors.New("批量开通请求不符合要求")
ErrBatchNotFound = errors.New("批量开通批次不存在")
ErrItemNotFound = errors.New("批量开通条目不存在")
)
type Activator func(context.Context, *Service, *Item, CredentialInput, int) (string, string, string, error)
type Service struct {
coreService.Service
Quota int
Activator Activator
}
func (s *Service) quotaLimit() (int, error) {
if s.Quota > 0 {
return s.Quota, nil
}
return quota.ReadLimit(s.Orm)
}
func (s *Service) CreateBatch(request CreateBatchRequest) (BatchResponse, error) {
request.IdempotencyKey = strings.TrimSpace(request.IdempotencyKey)
if request.IdempotencyKey == "" || len(request.IdempotencyKey) > 128 || len(request.Rows) == 0 || len(request.Rows) > 1000 {
return BatchResponse{}, ErrInvalidRequest
}
var existing Batch
if err := s.Orm.Preload("Items", func(db *gorm.DB) *gorm.DB { return db.Order("line_number") }).First(&existing, "idempotency_key = ?", request.IdempotencyKey).Error; err == nil {
return batchResponse(existing), nil
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
return BatchResponse{}, fmt.Errorf("read provisioning idempotency key: %w", err)
}
var existingDevices int64
if err := s.Orm.Table("sense_devices").Where("status <> ?", "disabled").Count(&existingDevices).Error; err != nil {
return BatchResponse{}, fmt.Errorf("count provisioned devices: %w", err)
}
quotaLimit, err := s.quotaLimit()
if err != nil {
return BatchResponse{}, err
}
available := quotaLimit - int(existingDevices)
if available < 0 {
available = 0
}
batch := Batch{ID: uuid.NewString(), IdempotencyKey: request.IdempotencyKey, Status: BatchReady, QuotaLimit: quotaLimit, ExistingCount: int(existingDevices), TotalCount: len(request.Rows)}
batch.CreateBy, batch.UpdateBy = request.CreateBy, request.CreateBy
seenLines := map[int]bool{}
seenAddresses := map[string]bool{}
ready := 0
for _, row := range request.Rows {
item := Item{ID: uuid.NewString(), BatchID: batch.ID, LineNumber: row.LineNumber, Name: strings.TrimSpace(row.Name), Location: strings.TrimSpace(row.Location), Address: strings.TrimSpace(row.Address), Status: ItemReady}
item.CreateBy, item.UpdateBy = request.CreateBy, request.CreateBy
code, detail := validateImportRow(item, seenLines, seenAddresses)
if code != "" {
item.Status, item.FailureCode, item.Detail = ItemInvalid, code, detail
} else if ready >= available {
item.Status, item.FailureCode, item.Detail = ItemQuotaExceeded, "quota_exceeded", "超出当前可用配额,请调整配额或批次后重试"
} else {
ready++
}
batch.Items = append(batch.Items, item)
}
batch.ReadyCount = ready
if ready == 0 {
batch.Status = BatchFailed
}
if err := s.Orm.Transaction(func(tx *gorm.DB) error {
if err := tx.Omit("Items").Create(&batch).Error; err != nil {
return err
}
return tx.Create(&batch.Items).Error
}); err != nil {
if isDuplicateKey(err) {
return s.GetBatchByKey(request.IdempotencyKey)
}
return BatchResponse{}, fmt.Errorf("create provisioning batch: %w", err)
}
return s.GetBatch(batch.ID)
}
func validateImportRow(item Item, seenLines map[int]bool, seenAddresses map[string]bool) (string, string) {
if item.LineNumber < 1 || seenLines[item.LineNumber] {
return "invalid_line_number", "行号必须为正整数且批次内唯一"
}
seenLines[item.LineNumber] = true
if item.Name == "" || len([]rune(item.Name)) > 128 || len([]rune(item.Location)) > 255 {
return "invalid_device", "设备名称不能为空,名称或安装位置长度不能超过限制"
}
parsed, err := url.Parse(item.Address)
if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Hostname() == "" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" {
return "invalid_address", "设备地址必须是无账号、查询参数和片段的 HTTP(S) 地址"
}
key := strings.ToLower(parsed.String())
if seenAddresses[key] {
return "duplicate_address", "同一批次中设备地址不能重复"
}
seenAddresses[key] = true
return "", ""
}
func (s *Service) ListBatches(request *BatchPageRequest) ([]BatchResponse, int64, error) {
query := s.Orm.Model(&Batch{})
if request.Status != "" {
query = query.Where("status = ?", request.Status)
}
var count int64
if err := query.Count(&count).Error; err != nil {
return nil, 0, err
}
pageSize := request.GetPageSize()
if pageSize > 100 {
pageSize = 100
}
var batches []Batch
if err := query.Order("created_at DESC").Limit(pageSize).Offset((request.GetPageIndex() - 1) * pageSize).Find(&batches).Error; err != nil {
return nil, 0, err
}
result := make([]BatchResponse, 0, len(batches))
for _, batch := range batches {
result = append(result, batchResponse(batch))
}
return result, count, nil
}
func (s *Service) GetBatch(id string) (BatchResponse, error) {
var batch Batch
if err := s.Orm.Preload("Items", func(db *gorm.DB) *gorm.DB { return db.Order("line_number") }).First(&batch, "id = ?", id).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return BatchResponse{}, ErrBatchNotFound
}
return BatchResponse{}, err
}
return batchResponse(batch), nil
}
func (s *Service) GetBatchByKey(key string) (BatchResponse, error) {
var batch Batch
if err := s.Orm.Preload("Items", func(db *gorm.DB) *gorm.DB { return db.Order("line_number") }).First(&batch, "idempotency_key = ?", key).Error; err != nil {
return BatchResponse{}, err
}
return batchResponse(batch), nil
}
func (s *Service) Execute(ctx context.Context, batchID string, itemID string, retryFailed bool, request ExecuteRequest) (BatchResponse, error) {
credentialByItem := make(map[string]CredentialInput, len(request.Credentials))
for _, input := range request.Credentials {
if input.ItemID == "" || credentialByItem[input.ItemID].ItemID != "" {
return BatchResponse{}, ErrInvalidRequest
}
credentialByItem[input.ItemID] = input
}
var batch Batch
if err := s.Orm.First(&batch, "id = ?", batchID).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return BatchResponse{}, ErrBatchNotFound
}
return BatchResponse{}, err
}
query := s.Orm.Where("batch_id = ?", batchID)
if itemID != "" {
query = query.Where("id = ?", itemID)
}
if retryFailed {
query = query.Where("status = ?", ItemFailed)
} else {
query = query.Where("status = ?", ItemReady)
}
var items []Item
if err := query.Order("line_number").Find(&items).Error; err != nil {
return BatchResponse{}, err
}
if itemID != "" && len(items) == 0 {
return BatchResponse{}, ErrItemNotFound
}
if len(items) == 0 {
return s.GetBatch(batchID)
}
if err := s.Orm.Model(&Batch{}).Where("id = ?", batchID).Updates(map[string]any{"status": BatchRunning, "update_by": request.UpdateBy}).Error; err != nil {
return BatchResponse{}, fmt.Errorf("mark provisioning batch running: %w", err)
}
activate := s.Activator
if activate == nil {
activate = defaultActivate
}
for index := range items {
item := &items[index]
now := time.Now().UTC()
expectedStatus := ItemReady
if retryFailed {
expectedStatus = ItemFailed
}
claim := s.Orm.Model(&Item{}).Where("id = ? AND status = ?", item.ID, expectedStatus).Updates(map[string]any{"status": ItemRunning, "attempts": gorm.Expr("attempts + 1"), "last_tried_at": now, "update_by": request.UpdateBy})
if claim.Error != nil {
return BatchResponse{}, fmt.Errorf("mark provisioning item running: %w", claim.Error)
}
if claim.RowsAffected == 0 {
continue
}
input, ok := credentialByItem[item.ID]
if !ok || strings.TrimSpace(input.ONVIFUsername) == "" || input.ONVIFPassword == "" {
if err := s.failItem(item.ID, request.UpdateBy, "credentials_required", "请安全填写该设备的账号密码后重试"); err != nil {
return BatchResponse{}, err
}
continue
}
deviceID, status, detail, err := activate(ctx, s, item, input, request.UpdateBy)
if deviceID != "" && deviceID != item.DeviceID {
item.DeviceID = deviceID
if err := s.Orm.Model(&Item{}).Where("id = ?", item.ID).Update("device_id", deviceID).Error; err != nil {
return BatchResponse{}, fmt.Errorf("link provisioning item to device: %w", err)
}
}
if err != nil {
code, safeDetail := safeFailure(err)
if updateErr := s.failItem(item.ID, request.UpdateBy, code, safeDetail); updateErr != nil {
return BatchResponse{}, updateErr
}
continue
}
if status != "ready" {
if detail == "" {
detail = "设备或视频流尚未通过验证"
}
if err := s.failItem(item.ID, request.UpdateBy, status, detail); err != nil {
return BatchResponse{}, err
}
continue
}
if err := s.Orm.Model(&Item{}).Where("id = ?", item.ID).Updates(map[string]any{"status": ItemSucceeded, "failure_code": "", "detail": "设备已开通并通过视频验证", "update_by": request.UpdateBy}).Error; err != nil {
return BatchResponse{}, fmt.Errorf("mark provisioning item succeeded: %w", err)
}
}
if err := s.refreshBatch(batchID, request.UpdateBy); err != nil {
return BatchResponse{}, err
}
return s.GetBatch(batchID)
}
func isDuplicateKey(err error) bool {
if errors.Is(err, gorm.ErrDuplicatedKey) {
return true
}
var postgresError *pgconn.PgError
return errors.As(err, &postgresError) && postgresError.Code == "23505"
}
func (s *Service) failItem(id string, updateBy int, code, detail string) error {
if err := s.Orm.Model(&Item{}).Where("id = ?", id).Updates(map[string]any{"status": ItemFailed, "failure_code": code, "detail": detail, "update_by": updateBy}).Error; err != nil {
return fmt.Errorf("mark provisioning item failed: %w", err)
}
return nil
}
func (s *Service) refreshBatch(batchID string, updateBy int) error {
var items []Item
if err := s.Orm.Where("batch_id = ?", batchID).Find(&items).Error; err != nil {
return err
}
ready, success, failure := 0, 0, 0
for _, item := range items {
switch item.Status {
case ItemReady, ItemRunning:
ready++
case ItemSucceeded:
success++
case ItemFailed:
failure++
}
}
status := BatchFailed
if success == len(items) {
status = BatchSucceeded
} else if success > 0 {
status = BatchPartial
} else if ready > 0 {
status = BatchReady
}
return s.Orm.Model(&Batch{}).Where("id = ?", batchID).Updates(map[string]any{"status": status, "ready_count": ready, "success_count": success, "failure_count": failure, "update_by": updateBy}).Error
}
func defaultActivate(ctx context.Context, service *Service, item *Item, input CredentialInput, updateBy int) (string, string, string, error) {
device := deviceService.Device{Service: service.Service}
deviceID := item.DeviceID
var response deviceDTO.DeviceResponse
if deviceID == "" {
if err := device.Insert(&deviceDTO.CreateReq{Name: item.Name, Location: item.Location, Modality: "video", Capabilities: []string{"video"}, CreateBy: updateBy}, &response); err != nil {
return "", "", "", err
}
deviceID = response.ID
if err := service.Orm.Model(&Item{}).Where("id = ?", item.ID).Update("device_id", deviceID).Error; err != nil {
return deviceID, "", "", err
}
} else if err := device.Get(deviceID, &response); err != nil {
return deviceID, "", "", err
}
if err := device.UpdateCredentials(&deviceDTO.CredentialUpdateReq{ID: deviceID, ONVIFUsername: input.ONVIFUsername, ONVIFPassword: input.ONVIFPassword, RTSPSameAsONVIF: input.RTSPSameAsONVIF, RTSPUsername: input.RTSPUsername, RTSPPassword: input.RTSPPassword, Version: response.Version, UpdateBy: updateBy}, &response); err != nil {
return deviceID, "", "", err
}
probeService, err := admission.NewRuntime(service.Service)
if err != nil {
return deviceID, "", "", err
}
result, err := probeService.Probe(ctx, admission.ProbeRequest{DeviceID: deviceID, Address: item.Address, Version: response.Version, UpdateBy: updateBy})
if err != nil {
return deviceID, "", "", err
}
return deviceID, result.Status, result.Detail, nil
}
func safeFailure(err error) (string, string) {
switch {
case errors.Is(err, credential.ErrKeyUnavailable):
return "credential_key_unavailable", "摄像头凭据安全配置不可用"
case errors.Is(err, credential.ErrCredentialNotConfigured):
return "credentials_required", "请安全填写摄像头账号密码后重试"
case errors.Is(err, deviceService.ErrInvalidDevice), errors.Is(err, admission.ErrInvalid):
return "invalid_device", "设备信息或凭据不符合要求"
case errors.Is(err, deviceService.ErrVersionConflict), errors.Is(err, admission.ErrConflict):
return "version_conflict", "设备已被其他操作更新,请重试"
case errors.Is(err, quota.ErrExceeded):
return "quota_exceeded", "当前配额已满,请调整配额或停用其他设备后重试"
case errors.Is(err, quota.ErrUnavailable):
return "quota_unavailable", "配额配置不可读取,已拒绝新增或启用设备"
default:
return "activation_failed", "设备开通失败,请检查网络、地址和凭据后重试"
}
}
func batchResponse(batch Batch) BatchResponse {
response := BatchResponse{ID: batch.ID, IdempotencyKey: batch.IdempotencyKey, Status: batch.Status, QuotaLimit: batch.QuotaLimit, ExistingCount: batch.ExistingCount, TotalCount: batch.TotalCount, ReadyCount: batch.ReadyCount, SuccessCount: batch.SuccessCount, FailureCount: batch.FailureCount, CreatedAt: batch.CreatedAt, UpdatedAt: batch.UpdatedAt, Items: make([]ItemResponse, 0, len(batch.Items))}
sort.Slice(batch.Items, func(i, j int) bool { return batch.Items[i].LineNumber < batch.Items[j].LineNumber })
for _, item := range batch.Items {
response.Items = append(response.Items, ItemResponse{ID: item.ID, LineNumber: item.LineNumber, Name: item.Name, Location: item.Location, Address: item.Address, Status: item.Status, FailureCode: item.FailureCode, Detail: item.Detail, DeviceID: item.DeviceID, Attempts: item.Attempts, LastTriedAt: item.LastTriedAt})
}
return response
}
@@ -0,0 +1,267 @@
package provisioning
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
"sync/atomic"
"testing"
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgconn"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
deviceModels "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
)
func provisioningService(t *testing.T, quota int) *Service {
t.Helper()
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{TranslateError: true})
if err != nil {
t.Fatal(err)
}
sqlDB, err := db.DB()
if err != nil {
t.Fatal(err)
}
sqlDB.SetMaxOpenConns(1)
if err = db.AutoMigrate(&deviceModels.Device{}, &Batch{}, &Item{}); err != nil {
t.Fatal(err)
}
return &Service{Service: coreService.Service{Orm: db}, Quota: quota}
}
func TestCreateBatchValidatesQuotaAndIsIdempotent(t *testing.T) {
service := provisioningService(t, 2)
if err := service.Orm.Create(&deviceModels.Device{ID: "existing", Name: "已接入摄像机", Modality: "video", Status: "active", AdapterStatus: "ready", Version: 1}).Error; err != nil {
t.Fatal(err)
}
request := CreateBatchRequest{
IdempotencyKey: "import-001",
CreateBy: 7,
Rows: []ImportRow{
{LineNumber: 1, Name: "东门摄像机", Location: "东门", Address: "http://192.0.2.10/onvif"},
{LineNumber: 2, Name: "重复地址", Location: "东门", Address: "http://192.0.2.10/onvif"},
{LineNumber: 3, Name: "西门摄像机", Location: "西门", Address: "https://192.0.2.11/onvif"},
},
}
created, err := service.CreateBatch(request)
if err != nil {
t.Fatal(err)
}
if created.QuotaLimit != 2 || created.ExistingCount != 1 || created.ReadyCount != 1 || len(created.Items) != 3 {
t.Fatalf("unexpected batch: %#v", created)
}
if created.Items[0].Status != ItemReady || created.Items[1].Status != ItemInvalid || created.Items[2].Status != ItemQuotaExceeded {
t.Fatalf("unexpected item statuses: %#v", created.Items)
}
request.Rows = []ImportRow{{LineNumber: 1, Name: "不应覆盖", Address: "http://192.0.2.99/onvif"}}
repeated, err := service.CreateBatch(request)
if err != nil {
t.Fatal(err)
}
if repeated.ID != created.ID || repeated.TotalCount != 3 || repeated.Items[0].Name != "东门摄像机" {
t.Fatalf("idempotent replay changed the batch: %#v", repeated)
}
}
func TestExecuteSupportsPartialSuccessAndFailedOnlyRetry(t *testing.T) {
service := provisioningService(t, 16)
created, err := service.CreateBatch(CreateBatchRequest{
IdempotencyKey: "execute-001",
Rows: []ImportRow{
{LineNumber: 1, Name: "东门摄像机", Address: "http://192.0.2.10/onvif"},
{LineNumber: 2, Name: "西门摄像机", Address: "http://192.0.2.11/onvif"},
},
})
if err != nil {
t.Fatal(err)
}
calls := map[int]int{}
service.Activator = func(_ context.Context, _ *Service, item *Item, input CredentialInput, _ int) (string, string, string, error) {
calls[item.LineNumber]++
if input.ONVIFPassword != "temporary-secret" {
t.Fatalf("activator did not receive the transient credential")
}
if item.LineNumber == 2 && calls[item.LineNumber] == 1 {
return "device-2", "", "", errors.New("synthetic network failure containing temporary-secret")
}
return "device-" + string(rune('0'+item.LineNumber)), "ready", "验证通过", nil
}
credentials := make([]CredentialInput, 0, len(created.Items))
for _, item := range created.Items {
credentials = append(credentials, CredentialInput{ItemID: item.ID, ONVIFUsername: "installer", ONVIFPassword: "temporary-secret", RTSPSameAsONVIF: true})
}
partial, err := service.Execute(context.Background(), created.ID, "", false, ExecuteRequest{Credentials: credentials, UpdateBy: 8})
if err != nil {
t.Fatal(err)
}
if partial.Status != BatchPartial || partial.SuccessCount != 1 || partial.FailureCount != 1 {
t.Fatalf("expected partial success, got %#v", partial)
}
if partial.Items[1].Detail == "" || strings.Contains(partial.Items[1].Detail, "temporary-secret") {
t.Fatalf("unsafe failure detail: %q", partial.Items[1].Detail)
}
retried, err := service.Execute(context.Background(), created.ID, "", true, ExecuteRequest{Credentials: []CredentialInput{{ItemID: created.Items[1].ID, ONVIFUsername: "installer", ONVIFPassword: "temporary-secret", RTSPSameAsONVIF: true}}, UpdateBy: 8})
if err != nil {
t.Fatal(err)
}
if retried.Status != BatchSucceeded || retried.SuccessCount != 2 || calls[1] != 1 || calls[2] != 2 {
t.Fatalf("failed-only retry was not idempotent: response=%#v calls=%#v", retried, calls)
}
encoded, err := json.Marshal(retried)
if err != nil {
t.Fatal(err)
}
if strings.Contains(string(encoded), "temporary-secret") || strings.Contains(string(encoded), "installer") {
t.Fatalf("response contains credentials: %s", encoded)
}
var stored []Item
if err = service.Orm.Find(&stored).Error; err != nil {
t.Fatal(err)
}
storedJSON, _ := json.Marshal(stored)
if strings.Contains(string(storedJSON), "temporary-secret") || strings.Contains(string(storedJSON), "installer") {
t.Fatalf("provisioning rows contain credentials: %s", storedJSON)
}
}
func TestExecuteRequiresCredentialsWithoutCallingActivator(t *testing.T) {
service := provisioningService(t, 16)
created, err := service.CreateBatch(CreateBatchRequest{IdempotencyKey: "missing-credentials", Rows: []ImportRow{{LineNumber: 1, Name: "东门摄像机", Address: "http://192.0.2.10/onvif"}}})
if err != nil {
t.Fatal(err)
}
called := false
service.Activator = func(context.Context, *Service, *Item, CredentialInput, int) (string, string, string, error) {
called = true
return "", "", "", nil
}
result, err := service.Execute(context.Background(), created.ID, "", false, ExecuteRequest{})
if err != nil {
t.Fatal(err)
}
if called || result.Status != BatchFailed || result.Items[0].FailureCode != "credentials_required" {
t.Fatalf("missing credentials were not handled safely: %#v called=%v", result, called)
}
}
func TestConcurrentCreateUsesOneIdempotentBatch(t *testing.T) {
service := provisioningService(t, 16)
sqlDB, err := service.Orm.DB()
if err != nil {
t.Fatal(err)
}
sqlDB.SetMaxOpenConns(1)
request := CreateBatchRequest{IdempotencyKey: "concurrent-import", Rows: []ImportRow{{LineNumber: 1, Name: "东门摄像机", Address: "http://192.0.2.10/onvif"}}}
const workers = 8
ids := make(chan string, workers)
errs := make(chan error, workers)
var wait sync.WaitGroup
for index := 0; index < workers; index++ {
wait.Add(1)
go func() {
defer wait.Done()
batch, createErr := service.CreateBatch(request)
if createErr != nil {
errs <- createErr
return
}
ids <- batch.ID
}()
}
wait.Wait()
close(ids)
close(errs)
for createErr := range errs {
t.Fatalf("concurrent create failed: %v", createErr)
}
var first string
for id := range ids {
if first == "" {
first = id
} else if id != first {
t.Fatalf("idempotent creates returned different batches: %q and %q", first, id)
}
}
var count int64
if err = service.Orm.Model(&Batch{}).Count(&count).Error; err != nil || count != 1 {
t.Fatalf("batch count=%d err=%v", count, err)
}
}
func TestSixteenDeviceBaselineSmoke(t *testing.T) {
service := provisioningService(t, 16)
rows := make([]ImportRow, 0, 16)
for index := 1; index <= 16; index++ {
rows = append(rows, ImportRow{LineNumber: index, Name: fmt.Sprintf("摄像机-%02d", index), Address: fmt.Sprintf("http://192.0.2.%d/onvif", index)})
}
created, err := service.CreateBatch(CreateBatchRequest{IdempotencyKey: "sixteen-device-smoke", Rows: rows})
if err != nil {
t.Fatal(err)
}
if created.ReadyCount != 16 {
t.Fatalf("ready count=%d", created.ReadyCount)
}
service.Activator = func(_ context.Context, _ *Service, item *Item, _ CredentialInput, _ int) (string, string, string, error) {
return "device-" + item.ID, "ready", "验证通过", nil
}
credentials := make([]CredentialInput, 0, 16)
for _, item := range created.Items {
credentials = append(credentials, CredentialInput{ItemID: item.ID, ONVIFUsername: "installer", ONVIFPassword: "temporary-secret", RTSPSameAsONVIF: true})
}
completed, err := service.Execute(context.Background(), created.ID, "", false, ExecuteRequest{Credentials: credentials})
if err != nil {
t.Fatal(err)
}
if completed.Status != BatchSucceeded || completed.SuccessCount != 16 || completed.FailureCount != 0 {
t.Fatalf("unexpected 16-device result: %#v", completed)
}
}
func TestConcurrentExecuteClaimsAnItemOnce(t *testing.T) {
service := provisioningService(t, 16)
created, err := service.CreateBatch(CreateBatchRequest{IdempotencyKey: "concurrent-execute", Rows: []ImportRow{{LineNumber: 1, Name: "东门摄像机", Address: "http://192.0.2.10/onvif"}}})
if err != nil {
t.Fatal(err)
}
var calls atomic.Int32
service.Activator = func(_ context.Context, _ *Service, item *Item, _ CredentialInput, _ int) (string, string, string, error) {
calls.Add(1)
return "device-" + item.ID, "ready", "验证通过", nil
}
request := ExecuteRequest{Credentials: []CredentialInput{{ItemID: created.Items[0].ID, ONVIFUsername: "installer", ONVIFPassword: "temporary-secret", RTSPSameAsONVIF: true}}}
var wait sync.WaitGroup
errs := make(chan error, 2)
for index := 0; index < 2; index++ {
wait.Add(1)
go func() {
defer wait.Done()
_, executeErr := service.Execute(context.Background(), created.ID, "", false, request)
errs <- executeErr
}()
}
wait.Wait()
close(errs)
for executeErr := range errs {
if executeErr != nil {
t.Fatal(executeErr)
}
}
if calls.Load() != 1 {
t.Fatalf("activator calls=%d", calls.Load())
}
}
func TestPostgresDuplicateKeyDetection(t *testing.T) {
if !isDuplicateKey(&pgconn.PgError{Code: "23505"}) || isDuplicateKey(&pgconn.PgError{Code: "23503"}) {
t.Fatal("PostgreSQL duplicate-key detection is incorrect")
}
}
+73
View File
@@ -0,0 +1,73 @@
package quota
import (
"errors"
"net/http"
"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"
)
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 &Service{DB: base.Orm}, nil
}
func (e *API) Get(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.Error(http.StatusBadRequest, err, "查询条件格式不正确")
return
}
response, err := service.Overview(request)
if err != nil {
e.writeError(err)
return
}
e.OK(response, "查询成功")
}
func (e *API) Update(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
request := UpdateRequest{UpdateBy: user.GetUserId(c)}
if err = e.MakeContext(c).Bind(&request, binding.JSON).Errors; err != nil {
e.Error(http.StatusBadRequest, err, "请求内容格式不正确")
return
}
response, err := service.Update(request)
if err != nil {
e.writeError(err)
return
}
e.OK(response, "配额已更新")
}
func (e *API) writeError(err error) {
switch {
case errors.Is(err, ErrInvalid):
e.Error(http.StatusBadRequest, err, err.Error())
case errors.Is(err, ErrExceeded), errors.Is(err, ErrVersionConflict):
e.Error(http.StatusConflict, err, err.Error())
case errors.Is(err, ErrUnavailable):
e.Error(http.StatusServiceUnavailable, err, err.Error())
default:
e.Error(http.StatusInternalServerError, err, "容量与配额操作失败")
}
}
+60
View File
@@ -0,0 +1,60 @@
package quota
import (
"time"
commonDto "git.ilapage.cn/ila/yovision/Sense/server/common/dto"
)
const (
ReadStatusReadable = "readable"
ReadStatusUnreadable = "unreadable"
)
type PageRequest struct {
commonDto.Pagination `search:"-"`
Keyword string `form:"keyword"`
Status string `form:"status"`
}
type UpdateRequest struct {
Limit int `json:"limit"`
Reason string `json:"reason"`
Version int64 `json:"version"`
UpdateBy int `json:"-"`
}
type Summary struct {
Limit int `json:"limit"`
Used int `json:"used"`
Remaining int `json:"remaining"`
ReadStatus string `json:"readStatus"`
Source string `json:"source"`
Version int64 `json:"version"`
LastReadAt time.Time `json:"lastReadAt"`
LastChanged time.Time `json:"lastChangedAt,omitempty"`
}
type DeviceOccupancy struct {
ID string `json:"id"`
Name string `json:"name"`
Location string `json:"location"`
Status string `json:"status"`
Occupied bool `json:"occupied"`
Version int64 `json:"version"`
UpdatedAt time.Time `json:"updatedAt"`
}
type Tier struct {
Limit int `json:"limit"`
Configured bool `json:"configured"`
Validation string `json:"validation"`
DeliveryStatus string `json:"deliveryStatus"`
}
type Overview struct {
Summary Summary `json:"summary"`
List []DeviceOccupancy `json:"list"`
Count int64 `json:"count"`
Tiers []Tier `json:"tiers"`
}

Some files were not shown because too many files have changed in this diff Show More