Introduce CommandAttribute/EventAttribute with config variants:
- REST controllers, MyBatis mappers, services, and DAL managers
- gRPC client/server builders and proto message definitions
- DTOs, BOs, DOs, VOs for type-safe data transfer layer
fix: add explicit ::text casts in json_build_object parameters
PostgreSQL cannot infer parameter types inside json_build_object()
calls. Add ::text casts to #{stateExtType} and #{stateDescription}
to resolve "could not determine data type of parameter $10" errors.
refactor: restructure driver gRPC client and metadata layer
- Rename GrpcStubConfig to DriverClientStubConfig with dedicated DTOs
- Extract attribute builders from inline gRPC conversion logic
- Consolidate driver metadata, register service, and command handling
- Update device state heartbeat and command result event handling
docs: add multilingual architecture diagram and sync READMEs
- Translate architecture SVG into zh, ja, vi variants
- Update README architecture image references per language
- Add missing service-level shortcut sections to ja/vi READMEs
- Add IoT DC3 layered architecture diagram (SVG + HTML viewer)
with Application / Management / Data / Driver layers,
Security & Load and Platform & System sidebars,
Data→Driver connection flows, and color-coded sections
- Reposition README descriptions across all languages (en, zh, ja, vi)
and pom.xml to AI-ready narrative: "Connects devices, collects
data organized for AI, and orchestrates the closed loop"
- Add README.ai.md with Chinese multi-paragraph AI narrative
and closed-loop flowchart
- Optimize package.json metadata (name, license, scripts)
- Fix ambiguous runExclusive mock by using typed ArgumentMatchers.<Supplier<String>>any()
- Fix setTimeoutUnit incorrect parameter in DeviceHealthScheduleJobTest
- Fix setProfileIds to setProfileId for 1:1 migration in DeviceToolTest
Add EVENT target type to AlarmTargetTypeFlagEnum, wire event report
into the alarm rule pipeline, add RabbitMQ contract E2E tests for
command call and event report, and create design documentation.
Replace dc3_profile_bind junction table with direct profile_id column on
dc3_device, simplifying the domain model. Device now owns a single
Profile instead of maintaining a many-to-many binding.
No data migration — seed DDL is updated separately.
P0: Replace content JSON-string with sealed PointCommandPayload + record DTOs.
P0: Accept write() boolean return, suppress echo on failure.
P1: Return commandId from POST endpoints, add RESTful GET /{commandId}.
P1: Add driver-online check via EntityStateMapper, PointCommandValidator.
P2: Add DeviceLockManager for per-device serialized command execution.
P3: Add POST /list API with pagination, commandId idempotency support.
- Replace two duplicate enums with single EntityStatusEnum (same index/code)
- Rename DB columns: state_flag→entity_state_flag, state_ext→entity_state_ext
- Add stateDescription field to DeviceStateDTO/DriverStateDTO for diagnostics
- Add description field to DeviceHealthState/DriverHealthState
- UPSERT SQL now writes stateDescription into entity_state_ext on heartbeat
- Remove EntityStateStatus interface (no longer needed with single enum)
- Update MQ layer, heartbeat services, status queries, gRPC, and all tests
Phase 2-4 of the point command chain redesign (point-command.md):
- Add DLX (dead letter exchange/queue) for failed commands
- Add result exchange/queue and PointCommandResultReceiver
- Add PointCommandDeadReceiver for DLX message processing
- Add CommandDedupCache (Caffeine 5-min TTL) for driver-side idempotency
- Add PointCommandResultDTO for driver-to-center result receipts
- Persist commands (PENDING->SENT) with commandId/CorrelationData
- Enhance validateCommandScope: device/point enable flag, rwFlag for write
- Add GET /point_command/get_by_command_id query endpoint
- Update PointCommandDTO to carry commandId and tenantId
- Update PointCommandReceiver: dedup check, result sending, bounded retry
Replace Caffeine LocalCacheService + @Scheduled scanner with
lease-based timeout using RabbitMQ TTL + DLX:
- Driver: each heartbeat publishes a 45s delayed check message;
DriverTimeoutCheckReceiver performs secondary lease_version and
expire_time verification before marking offline.
- Device: a self-sustaining 10s tick queue triggers batch scanning
of expired device leases in EntityStateExpiryScanner.
- Remove OfflineExpiryListener and LocalCacheService from state chain.
- Heartbeat services now only write dc3_entity_state (source of truth).
Rename driver_id to parent_entity_id, ttl_seconds to timeout_seconds.
Add last_state_flag, last_heartbeat_time, last_alarm_id,
timeout_source_flag, and state_ext fields to EntityStateDO.
Add TimeoutSourceFlagEnum for timeout source classification.
Design fixes:
- EntityStateExpiryScanner: use atomic lambdaUpdate with lease_version
WHERE condition instead of read-then-write, preventing duplicate
alarm writes across multiple Data Center instances.
- EntityStateExpiryScanner: add LIMIT 500 to scan query to bound
memory usage after extended downtime.
- OfflineExpiryListener: atomically mark DB state as offline before
writing alarm, preventing the scanner from writing a duplicate.
- EntityStateDO: fix Javadoc to reflect actual EntityTypeFlagEnum
values (DRIVER=3, DEVICE=6).
Tests (32 total, all passing):
- EntityStateExpiryScannerTest: no expired rows, already-offline skip,
online driver/device expiry with alarm, claim failure when another
instance wins, multiple expired rows processed in order.
- DriverStateServiceImplTest: null guard, new DB row creation with
correct fields, lease version increment, status flip alarm trigger.
- DeviceStateServiceImplTest: null guard, custom TTL, lease version
increment, null driverId defaults to 0, status flip alarm.
- DriverStatusServiceImplTest: empty page, offline when DB missing,
online from DB, offline when expired, device online/offline counts.
- DeviceStatusServiceImplTest: empty page, offline when DB missing,
online from DB, offline when expired, profile query.
Replace local Caffeine cache as the source of truth for driver and
device online status with dc3_entity_state, a persistent state lease
table backed by PostgreSQL.
- Add EntityStateDO/Mapper/Manager persistence stack.
- Add EntityStateExpiryScanner that scans expire_time every 15 seconds
with lease_version recheck to avoid stale alarm writes.
- Dual-write heartbeat path: DriverStateServiceImpl and
DeviceStateServiceImpl now write dc3_entity_state first, then local
cache.
- All status queries (controllers, gRPC server, local facade,
SystemHealthServiceImpl) read from dc3_entity_state. Rows missing or
past expire_time return offline.
- OfflineExpiryListener gains a DB dedup check to skip expiry events
already handled by the scanner.
- Add @EnableScheduling to DataApplication.
- Add design doc docs/design/entity-state.md with full DDL.
This resolves restart state loss, multi-instance inconsistency, and
races between expiry checks and fresh heartbeats.
Replace hand-written all-assignment constructors with Lombok
@RequiredArgsConstructor across controllers, services, config classes,
event listeners, and tool classes. No behavioral change.
Wires the windowed evaluation path that the previous P4 commits set up.
With this commit AVG/MIN/MAX/SUM/COUNT/ALL/ANY rules actually run end-
to-end:
- ConditionEvaluator: utility extracted out of RuleEvaluatorImpl so
both the LAST path and the windowed path apply identical operator
semantics. evaluate(condition, value), recoveryConditionOf(...) and
toBigDecimal(...) are the only public surfaces.
- WindowDataSource: small interface with aggregate() (scalar fold for
AVG/MIN/MAX/SUM/COUNT) and samples() (raw rows for ALL/ANY).
AggregateOutcome carries (value, sampleCount) so callers can enforce
minSamples.
- LocalWindowDataSource: reads WindowSampleBuffer.snapshot(...) and
folds in Java; AVG uses scale=6 / HALF_UP rounding to match
PostgreSQL's AVG(numeric). Skips null numValue samples for numeric
aggregates; COUNT counts every sample.
- RepositoryWindowDataSource: pushes the aggregate to the time-series
store via RepositoryService.aggregateInWindow / samplesInWindow
introduced in the previous commit. Only POINT facts are supported.
- HybridWindowDataSource (@Primary): routes by spec.duration —
duration <= dc3.alarm.window.local-cutoff goes to local, larger
goes to repository.
- WindowedRuleEvaluator: dispatch by mode. AVG/MIN/MAX/SUM/COUNT pull
the scalar and feed it to ConditionEvaluator; ALL/ANY pull samples
and apply the rule's per-sample condition. minSamples gates both
paths. Recovery uses the synthesized condition (Recovery.operator
+ Recovery.threshold) over the same window, mirroring the LAST path.
- RuleEvaluatorImpl: rewritten to dispatch by parsed WindowSpec. LAST
stays inline (no data-source plumbing needed); everything else
delegates to the windowed evaluator. The previous warn-once gate
for non-LAST modes is removed; invalid window specs (which should
already be rejected by the save validator) are skipped with a
single log line.
Tests:
- RuleEvaluatorImplTest is rewritten to pass a mock
WindowedRuleEvaluator and verify dispatch.
- WindowedRuleEvaluatorTest covers AVG/COUNT/ALL/ANY/recovery paths
+ minSamples gate + LAST defensive guard.
- All 193 tests in dc3-common-data pass.
Long-window alarm rules need a way to fold AVG/MIN/MAX/SUM/COUNT over
a time bracket of dc3_point_value, plus a raw-sample pull for ALL/ANY
where the rule condition has to run per row. The existing
RepositoryService had neither — listHistoryPointValue is count-bound
without a from/to, listPagePointValue only takes a lower bound.
Add two methods on RepositoryService:
- aggregateInWindow(WindowAggregateRequest) -> WindowAggregateResult
pushes AVG/MIN/MAX/SUM/COUNT to PostgreSQL via a fixed-set <choose>
in the new mapper XML so the function name can't be a SQL-injection
vector. Numeric aggregates filter num_value IS NOT NULL to align
with the partial index already in place; COUNT counts every row in
the window.
- samplesInWindow(tenantId, deviceId, pointId, from, to)
pulls (num_value, cal_value, create_time) ordered by create_time.
Used by ALL/ANY long-window paths — slow by design, the doc warns
against this combination.
Both live on PointValueMapper + PointValueMapper.xml so a future TSDB
adapter can swap the implementation without touching the alarm side.
PostgresRepositoryServiceImpl wires the calls through and guards
against null arguments. RepositoryStrategyFactoryTest's stub gets the
two new no-op methods so it still satisfies the interface.
Tests: PostgresRepositoryServiceImplWindowTest covers delegation +
null guards. The actual SQL is verified by inspection — adding a real
DAL slice (Testcontainers) for the alarm side is left for later.
Short windows (≤ dc3.alarm.window.local-cutoff, default PT5M) need a
side store the rule engine can read to fold AVG/MIN/MAX/SUM/COUNT/
ALL/ANY over recent samples. PointValueLocalCacheService is latest-
only and can't satisfy this, so introduce a dedicated buffer.
Shape:
- WindowSample (record): {numValue, calValue, timestamp}. numValue is
null for non-numeric payloads — aggregators must filter null.
- WindowSampleKey (record): (tenantId, targetType, entityId). Per-
entity rather than per-rule so multiple rules on the same entity
share one time-ordered series.
- WindowSampleBuffer: Caffeine map of key -> ConcurrentLinkedDeque,
trimmed inline on each append by both retention time
(local-cutoff) and per-key sample count (max-samples-per-buffer,
default 1000). Idle keys are evicted by Caffeine after
buffer-idle-expiry (default 30min).
- AlarmWindowProperties (`@ConfigurationProperties("dc3.alarm.window")`):
exposes localCutoff / maxBufferKeys / maxSamplesPerBuffer /
bufferIdleExpiry.
AlarmRuleTriggerServiceImpl.processPointValue now appends to the
buffer before dispatching the fact, so when the engine eventually
reads a snapshot it sees the triggering sample inside the window.
Doesn't connect to the evaluator yet — that's P4-4. P4-3 lands the
long-window repository path next.
Phase 4 lifts the LAST-only gate so users can save AVG/MIN/MAX/SUM/
COUNT/ALL/ANY rules ahead of the runtime evaluator landing. RuleExt's
window block already declares a mode + ISO-8601 duration + minSamples,
but nothing parsed those fields — save validation rejected anything
non-LAST and runtime ignored the duration entirely.
Add three pieces:
- WindowMode enum (dc3-common-model) — typed mode value with case-
insensitive ofCode(), plus reducesToScalar() / requiresDuration()
classifiers used by the future evaluator.
- WindowSpec record — parsed view {mode, duration, minSamples, valid,
reason} with last() / ok() / invalid() factories.
- WindowSpecParser — turns RuleExt.Window into WindowSpec, applying
the rules consistently (null window = LAST, blank mode = LAST,
duration parse failure = invalid, non-positive duration on
aggregation modes = invalid). Future runtime evaluator + this
validator both call through the same parser.
RuleServiceImpl.validateWindowMode now routes through the parser:
unknown modes, malformed durations, and zero/negative durations on
aggregation modes get UnSupportException at save time. AVG with PT3M
is now accepted at save (the runtime path still no-ops for non-LAST
until the evaluator wiring lands).
Existing rejection tests flipped: AVG with PT3M is now accepted, the
rejection paths are exercised through unknown mode, zero duration,
and malformed duration. WindowSpecParserTest covers parser behavior
in isolation.
RuleNotificationServiceImpl was issuing four DB lookups per alarm
fan-out: notify by id, message by id, enabled-bind list by tenant +
notify, and channel by id. Notify/message/channel/bind rows mutate at
configuration time and are read on every notification, so the read-amp
is significant under sustained alarm load.
Add NotifyConfigCache, a Caffeine-backed wrapper that owns:
- notify (NotifyBO by id)
- message (MessageBO by id)
- channel (NotifyChannelBO by id), with tenant scope enforced at the
call site so cross-tenant lookups still return null
- bind list (List<NotifyChannelBindBO> by (tenantId, notifyId))
Invalidation:
- NotifyServiceImpl, MessageServiceImpl, NotifyChannelServiceImpl,
NotifyChannelBindServiceImpl call the matching invalidate hook on
add/update/delete.
- Bind updates invalidate both old and new (tenantId, notifyId) tuples
in case the parent notify changes.
RuleNotificationServiceImpl is dramatically slimmed: the four
loadNotify/loadMessage/loadEnabledBinds/loadChannel helpers now route
through the cache, and the previously-injected manager/builder fields
that only existed for those calls are dropped.
Configurable via dc3.alarm.cache.notify.{max-size,ttl-seconds};
defaults hold 5000 entries for 60 seconds across each section.
PointValueServiceImpl.save(List) used to do
pointValueBOList.forEach(alarmRuleTriggerService::processPointValue),
forcing every batch caller to know that the trigger is per-fact. With
RuleRegistry now caching candidate rules, the per-fact dispatch is
already cheap, but the call site still leaks the loop.
Add AlarmRuleTriggerService.processPointValues(List<PointValueBO>) so
batch ingest hands off the whole list in one call. The default impl
fans out per-fact internally — rule semantics (firing, recovery, dedup)
are still defined sample-by-sample, so the engine itself stays unchanged.
Tests: PointValueServiceImplTest now verifies the bulk handoff
(times(1) processPointValues), and a new AlarmRuleTriggerServiceImplTest
covers the trigger's per-fact fan-out + invalid-entry filtering for
both the single and batch entrypoints.
RuleEngineImpl previously hit dc3_rule on every fact via
RuleCandidateLookup + RuleBuilder.buildBOListByDOList. With per-second
point ingestion, that means database QPS = point upload rate.
Add RuleRegistry as a Caffeine-backed cache fronting both the lookup
and the BO conversion. Cache key is (tenantId, alarmTargetTypeFlag,
entityId), which is exactly the dimension RuleCandidateLookup already
filters on, so the cached list is reusable across repeated facts for
the same entity. Cached values are pre-built RuleBOs — no repeat
DO -> BO work either.
Invalidation:
- RuleServiceImpl.add/update/delete drops the per-tenant cache so the
next evaluation reflects the change.
- A 60-second write TTL is a safety net for missed invalidations
(e.g., a manual SQL update).
Configurable via dc3.alarm.cache.rule.{max-size,ttl-seconds}; defaults
hold 10000 keys for 60 seconds.
Outbound notification used to call adapter.send synchronously inside
the @Transactional rule pipeline, so a slow webhook or feishu adapter
would stretch the rule transaction and starve the rabbit consumers
that feed it. Move the dispatch onto its own queue:
- New dc3.q.notify.task queue bound to the alarm exchange with routing
key dc3.r.notify.task.<channel-type>; 24h TTL on the queue caps any
runaway backlog when an adapter is fully wedged.
- NotifyTaskDTO carries the rendered MessagePayload plus the PENDING
notify_history row id so the worker can mutate it in place.
- NotifyTaskSender publishes tasks; RuleNotificationServiceImpl writes
the PENDING row, advances last_notify_time for rate-limit debouncing,
and hands the task to the sender — no blocking I/O remains in the
notification entrypoint.
- NotifyWorker consumes the queue, looks up the channel/adapter, and
stamps the history row with SUCCESS / FAILED / SKIPPED. Failures
re-publish the task with retryCount++ until MAX_ATTEMPTS (3); the
history row carries RETRYING in between. We deliberately don't use
basicNack(requeue=true) — RabbitMQ would put the message back at the
head of the queue and the worker would tight-loop on it.
- No DLQ wired; terminal FAILED rows are the operator audit trail.
Add tests for both the sender (routing-key shape, history-id required)
and the worker (success/failed/retrying/terminal/skipped paths). Tests
now total 138 in dc3-common-data.
dc3_notify_history previously transitioned directly from non-existent
to its terminal status (SUCCESS / FAILED / SKIPPED) at the end of the
synchronous send path. The async dispatch refactor in commit 7 needs
the history row to exist *before* the channel send so the worker has
something to update.
Refactor the writer into three pieces with no external behavior change:
- buildHistory(...) factors out the column population shared by both
paths.
- persistPendingHistory(...) writes a PENDING row before adapter.send
so the row id is stable for the subsequent finalize update (and for
the upcoming worker).
- finalizeHistory(...) updates the same row to SUCCESS / FAILED with
response_ext / error_message / target.
- historySkipped(...) keeps writing SKIPPED rows directly — those rows
describe a *decision* to not send, so they have no PENDING phase.
This change keeps existing tests green and isolates the storage shape
needed for the queue-based worker that lands next.
Vickey