feat(alerting): Postgres repository for alert_instances with inbox queries
Implements AlertInstanceRepository: save (upsert), findById, findOpenForRule, listForInbox (3-way OR: user/group/role via && array-overlap + ANY), countUnreadForUser (LEFT JOIN alert_reads), ack, resolve, markSilenced, deleteResolvedBefore. Integration test covers all 9 scenarios including inbox fan-out across all three target types. Also adds @JsonIgnoreProperties(ignoreUnknown=true) to SilenceMatcher to suppress Jackson serializing isWildcard() as a round-trip field. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
hsiegeln
@@ -0,0 +1,247 @@
|
||||
package com.cameleer.server.app.alerting.storage;
|
||||
|
||||
import com.cameleer.server.core.alerting.*;
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.springframework.jdbc.core.ConnectionCallback;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.core.RowMapper;
|
||||
|
||||
import java.sql.Array;
|
||||
import java.sql.SQLException;
|
||||
import java.sql.Timestamp;
|
||||
import java.time.Instant;
|
||||
import java.util.*;
|
||||
|
||||
public class PostgresAlertInstanceRepository implements AlertInstanceRepository {
|
||||
|
||||
private final JdbcTemplate jdbc;
|
||||
private final ObjectMapper om;
|
||||
|
||||
public PostgresAlertInstanceRepository(JdbcTemplate jdbc, ObjectMapper om) {
|
||||
this.jdbc = jdbc;
|
||||
this.om = om;
|
||||
}
|
||||
|
||||
@Override
|
||||
public AlertInstance save(AlertInstance i) {
|
||||
String sql = """
|
||||
INSERT INTO alert_instances (
|
||||
id, rule_id, rule_snapshot, environment_id, state, severity,
|
||||
fired_at, acked_at, acked_by, resolved_at, last_notified_at,
|
||||
silenced, current_value, threshold, context, title, message,
|
||||
target_user_ids, target_group_ids, target_role_names)
|
||||
VALUES (?, ?, ?::jsonb, ?, ?::alert_state_enum, ?::severity_enum,
|
||||
?, ?, ?, ?, ?,
|
||||
?, ?, ?, ?::jsonb, ?, ?,
|
||||
?, ?, ?)
|
||||
ON CONFLICT (id) DO UPDATE SET
|
||||
state = EXCLUDED.state,
|
||||
acked_at = EXCLUDED.acked_at,
|
||||
acked_by = EXCLUDED.acked_by,
|
||||
resolved_at = EXCLUDED.resolved_at,
|
||||
last_notified_at = EXCLUDED.last_notified_at,
|
||||
silenced = EXCLUDED.silenced,
|
||||
current_value = EXCLUDED.current_value,
|
||||
threshold = EXCLUDED.threshold,
|
||||
context = EXCLUDED.context,
|
||||
title = EXCLUDED.title,
|
||||
message = EXCLUDED.message,
|
||||
target_user_ids = EXCLUDED.target_user_ids,
|
||||
target_group_ids = EXCLUDED.target_group_ids,
|
||||
target_role_names = EXCLUDED.target_role_names
|
||||
""";
|
||||
Array userIds = toTextArray(i.targetUserIds());
|
||||
Array groupIds = toUuidArray(i.targetGroupIds());
|
||||
Array roleNames = toTextArray(i.targetRoleNames());
|
||||
|
||||
jdbc.update(sql,
|
||||
i.id(), i.ruleId(), writeJson(i.ruleSnapshot()),
|
||||
i.environmentId(), i.state().name(), i.severity().name(),
|
||||
ts(i.firedAt()), ts(i.ackedAt()), i.ackedBy(),
|
||||
ts(i.resolvedAt()), ts(i.lastNotifiedAt()),
|
||||
i.silenced(), i.currentValue(), i.threshold(),
|
||||
writeJson(i.context()), i.title(), i.message(),
|
||||
userIds, groupIds, roleNames);
|
||||
return i;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Optional<AlertInstance> findById(UUID id) {
|
||||
var list = jdbc.query("SELECT * FROM alert_instances WHERE id = ?", rowMapper(), id);
|
||||
return list.isEmpty() ? Optional.empty() : Optional.of(list.get(0));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Optional<AlertInstance> findOpenForRule(UUID ruleId) {
|
||||
var list = jdbc.query("""
|
||||
SELECT * FROM alert_instances
|
||||
WHERE rule_id = ?
|
||||
AND state IN ('PENDING','FIRING','ACKNOWLEDGED')
|
||||
LIMIT 1
|
||||
""", rowMapper(), ruleId);
|
||||
return list.isEmpty() ? Optional.empty() : Optional.of(list.get(0));
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<AlertInstance> listForInbox(UUID environmentId,
|
||||
List<String> userGroupIdFilter,
|
||||
String userId,
|
||||
List<String> userRoleNames,
|
||||
int limit) {
|
||||
// Build arrays for group UUIDs and role names
|
||||
Array groupArray = toUuidArrayFromStrings(userGroupIdFilter);
|
||||
Array roleArray = toTextArray(userRoleNames);
|
||||
|
||||
String sql = """
|
||||
SELECT * FROM alert_instances
|
||||
WHERE environment_id = ?
|
||||
AND (
|
||||
? = ANY(target_user_ids)
|
||||
OR target_group_ids && ?
|
||||
OR target_role_names && ?
|
||||
)
|
||||
ORDER BY fired_at DESC
|
||||
LIMIT ?
|
||||
""";
|
||||
return jdbc.query(sql, rowMapper(), environmentId, userId, groupArray, roleArray, limit);
|
||||
}
|
||||
|
||||
@Override
|
||||
public long countUnreadForUser(UUID environmentId, String userId) {
|
||||
String sql = """
|
||||
SELECT COUNT(*) FROM alert_instances ai
|
||||
WHERE ai.environment_id = ?
|
||||
AND ? = ANY(ai.target_user_ids)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM alert_reads ar
|
||||
WHERE ar.user_id = ? AND ar.alert_instance_id = ai.id
|
||||
)
|
||||
""";
|
||||
Long count = jdbc.queryForObject(sql, Long.class, environmentId, userId, userId);
|
||||
return count == null ? 0L : count;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void ack(UUID id, String userId, Instant when) {
|
||||
jdbc.update("""
|
||||
UPDATE alert_instances
|
||||
SET state = 'ACKNOWLEDGED'::alert_state_enum,
|
||||
acked_at = ?, acked_by = ?
|
||||
WHERE id = ?
|
||||
""", Timestamp.from(when), userId, id);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void resolve(UUID id, Instant when) {
|
||||
jdbc.update("""
|
||||
UPDATE alert_instances
|
||||
SET state = 'RESOLVED'::alert_state_enum,
|
||||
resolved_at = ?
|
||||
WHERE id = ?
|
||||
""", Timestamp.from(when), id);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void markSilenced(UUID id, boolean silenced) {
|
||||
jdbc.update("UPDATE alert_instances SET silenced = ? WHERE id = ?", silenced, id);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void deleteResolvedBefore(Instant cutoff) {
|
||||
jdbc.update("""
|
||||
DELETE FROM alert_instances
|
||||
WHERE state = 'RESOLVED'::alert_state_enum
|
||||
AND resolved_at < ?
|
||||
""", Timestamp.from(cutoff));
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private RowMapper<AlertInstance> rowMapper() {
|
||||
return (rs, i) -> {
|
||||
try {
|
||||
Map<String, Object> snapshot = om.readValue(
|
||||
rs.getString("rule_snapshot"), new TypeReference<>() {});
|
||||
Map<String, Object> context = om.readValue(
|
||||
rs.getString("context"), new TypeReference<>() {});
|
||||
|
||||
Timestamp ackedAt = rs.getTimestamp("acked_at");
|
||||
Timestamp resolvedAt = rs.getTimestamp("resolved_at");
|
||||
Timestamp lastNotifiedAt = rs.getTimestamp("last_notified_at");
|
||||
|
||||
Object cvObj = rs.getObject("current_value");
|
||||
Double currentValue = cvObj == null ? null : ((Number) cvObj).doubleValue();
|
||||
Object thObj = rs.getObject("threshold");
|
||||
Double threshold = thObj == null ? null : ((Number) thObj).doubleValue();
|
||||
|
||||
UUID ruleId = rs.getObject("rule_id") == null ? null : (UUID) rs.getObject("rule_id");
|
||||
|
||||
return new AlertInstance(
|
||||
(UUID) rs.getObject("id"),
|
||||
ruleId,
|
||||
snapshot,
|
||||
(UUID) rs.getObject("environment_id"),
|
||||
AlertState.valueOf(rs.getString("state")),
|
||||
AlertSeverity.valueOf(rs.getString("severity")),
|
||||
rs.getTimestamp("fired_at").toInstant(),
|
||||
ackedAt == null ? null : ackedAt.toInstant(),
|
||||
rs.getString("acked_by"),
|
||||
resolvedAt == null ? null : resolvedAt.toInstant(),
|
||||
lastNotifiedAt == null ? null : lastNotifiedAt.toInstant(),
|
||||
rs.getBoolean("silenced"),
|
||||
currentValue,
|
||||
threshold,
|
||||
context,
|
||||
rs.getString("title"),
|
||||
rs.getString("message"),
|
||||
readTextArray(rs.getArray("target_user_ids")),
|
||||
readUuidArray(rs.getArray("target_group_ids")),
|
||||
readTextArray(rs.getArray("target_role_names")));
|
||||
} catch (Exception e) {
|
||||
throw new IllegalStateException("Failed to map alert_instances row", e);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private String writeJson(Object o) {
|
||||
try { return om.writeValueAsString(o); }
|
||||
catch (Exception e) { throw new IllegalStateException("Failed to serialize JSON", e); }
|
||||
}
|
||||
|
||||
private Timestamp ts(Instant instant) {
|
||||
return instant == null ? null : Timestamp.from(instant);
|
||||
}
|
||||
|
||||
private Array toTextArray(List<String> items) {
|
||||
return jdbc.execute((ConnectionCallback<Array>) conn ->
|
||||
conn.createArrayOf("text", items.toArray()));
|
||||
}
|
||||
|
||||
private Array toUuidArray(List<UUID> ids) {
|
||||
return jdbc.execute((ConnectionCallback<Array>) conn ->
|
||||
conn.createArrayOf("uuid", ids.toArray()));
|
||||
}
|
||||
|
||||
private Array toUuidArrayFromStrings(List<String> ids) {
|
||||
return jdbc.execute((ConnectionCallback<Array>) conn ->
|
||||
conn.createArrayOf("uuid",
|
||||
ids.stream().map(UUID::fromString).toArray()));
|
||||
}
|
||||
|
||||
private List<String> readTextArray(Array arr) throws SQLException {
|
||||
if (arr == null) return List.of();
|
||||
Object[] raw = (Object[]) arr.getArray();
|
||||
List<String> out = new ArrayList<>(raw.length);
|
||||
for (Object o : raw) out.add((String) o);
|
||||
return out;
|
||||
}
|
||||
|
||||
private List<UUID> readUuidArray(Array arr) throws SQLException {
|
||||
if (arr == null) return List.of();
|
||||
Object[] raw = (Object[]) arr.getArray();
|
||||
List<UUID> out = new ArrayList<>(raw.length);
|
||||
for (Object o : raw) out.add((UUID) o);
|
||||
return out;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,196 @@
|
||||
package com.cameleer.server.app.alerting.storage;
|
||||
|
||||
import com.cameleer.server.app.AbstractPostgresIT;
|
||||
import com.cameleer.server.core.alerting.*;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class PostgresAlertInstanceRepositoryIT extends AbstractPostgresIT {
|
||||
|
||||
private PostgresAlertInstanceRepository repo;
|
||||
private UUID envId;
|
||||
private UUID ruleId;
|
||||
private final String userId = "inbox-user-" + UUID.randomUUID();
|
||||
private final String groupId = UUID.randomUUID().toString();
|
||||
private final String roleName = "OPERATOR";
|
||||
|
||||
@BeforeEach
|
||||
void setup() {
|
||||
repo = new PostgresAlertInstanceRepository(jdbcTemplate, new ObjectMapper());
|
||||
envId = UUID.randomUUID();
|
||||
ruleId = UUID.randomUUID();
|
||||
|
||||
jdbcTemplate.update(
|
||||
"INSERT INTO environments (id, slug, display_name) VALUES (?, ?, ?)",
|
||||
envId, "test-env-" + UUID.randomUUID(), "Test Env");
|
||||
jdbcTemplate.update(
|
||||
"INSERT INTO users (user_id, provider, email) VALUES (?, 'local', ?) ON CONFLICT (user_id) DO NOTHING",
|
||||
userId, userId + "@example.com");
|
||||
jdbcTemplate.update(
|
||||
"INSERT INTO users (user_id, provider, email) VALUES ('sys-user', 'local', 'sys@example.com') ON CONFLICT (user_id) DO NOTHING");
|
||||
jdbcTemplate.update(
|
||||
"INSERT INTO alert_rules (id, environment_id, name, severity, condition_kind, condition, " +
|
||||
"notification_title_tmpl, notification_message_tmpl, created_by, updated_by) " +
|
||||
"VALUES (?, ?, 'rule', 'WARNING', 'AGENT_STATE', '{}'::jsonb, 't', 'm', 'sys-user', 'sys-user')",
|
||||
ruleId, envId);
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void cleanup() {
|
||||
jdbcTemplate.update("DELETE FROM alert_reads WHERE user_id = ?", userId);
|
||||
jdbcTemplate.update("DELETE FROM alert_notifications WHERE alert_instance_id IN " +
|
||||
"(SELECT id FROM alert_instances WHERE environment_id = ?)", envId);
|
||||
jdbcTemplate.update("DELETE FROM alert_instances WHERE environment_id = ?", envId);
|
||||
jdbcTemplate.update("DELETE FROM alert_rules WHERE environment_id = ?", envId);
|
||||
jdbcTemplate.update("DELETE FROM environments WHERE id = ?", envId);
|
||||
jdbcTemplate.update("DELETE FROM users WHERE user_id = ?", userId);
|
||||
}
|
||||
|
||||
@Test
|
||||
void saveAndFindByIdRoundtrip() {
|
||||
var inst = newInstance(ruleId, List.of(userId), List.of(), List.of());
|
||||
repo.save(inst);
|
||||
|
||||
var found = repo.findById(inst.id()).orElseThrow();
|
||||
assertThat(found.id()).isEqualTo(inst.id());
|
||||
assertThat(found.state()).isEqualTo(AlertState.FIRING);
|
||||
assertThat(found.severity()).isEqualTo(AlertSeverity.WARNING);
|
||||
assertThat(found.targetUserIds()).containsExactly(userId);
|
||||
assertThat(found.targetGroupIds()).isEmpty();
|
||||
assertThat(found.targetRoleNames()).isEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
void listForInbox_seesAllThreeTargetTypes() {
|
||||
// Instance 1 — targeted at user directly
|
||||
var byUser = newInstance(ruleId, List.of(userId), List.of(), List.of());
|
||||
// Instance 2 — targeted at group
|
||||
var byGroup = newInstance(ruleId, List.of(), List.of(UUID.fromString(groupId)), List.of());
|
||||
// Instance 3 — targeted at role
|
||||
var byRole = newInstance(ruleId, List.of(), List.of(), List.of(roleName));
|
||||
|
||||
repo.save(byUser);
|
||||
repo.save(byGroup);
|
||||
repo.save(byRole);
|
||||
|
||||
// User is member of the group AND has the role
|
||||
var inbox = repo.listForInbox(envId, List.of(groupId), userId, List.of(roleName), 50);
|
||||
assertThat(inbox).extracting(AlertInstance::id)
|
||||
.containsExactlyInAnyOrder(byUser.id(), byGroup.id(), byRole.id());
|
||||
}
|
||||
|
||||
@Test
|
||||
void listForInbox_emptyGroupsAndRoles() {
|
||||
var byUser = newInstance(ruleId, List.of(userId), List.of(), List.of());
|
||||
repo.save(byUser);
|
||||
|
||||
var inbox = repo.listForInbox(envId, List.of(), userId, List.of(), 50);
|
||||
assertThat(inbox).hasSize(1);
|
||||
assertThat(inbox.get(0).id()).isEqualTo(byUser.id());
|
||||
}
|
||||
|
||||
@Test
|
||||
void countUnreadForUser_decreasesAfterMarkRead() {
|
||||
var inst = newInstance(ruleId, List.of(userId), List.of(), List.of());
|
||||
repo.save(inst);
|
||||
|
||||
long before = repo.countUnreadForUser(envId, userId);
|
||||
assertThat(before).isEqualTo(1L);
|
||||
|
||||
// Insert read record directly (AlertReadRepository not yet wired in this test)
|
||||
jdbcTemplate.update(
|
||||
"INSERT INTO alert_reads (user_id, alert_instance_id) VALUES (?, ?) ON CONFLICT DO NOTHING",
|
||||
userId, inst.id());
|
||||
|
||||
long after = repo.countUnreadForUser(envId, userId);
|
||||
assertThat(after).isEqualTo(0L);
|
||||
}
|
||||
|
||||
@Test
|
||||
void findOpenForRule_excludesResolved() {
|
||||
var open = newInstance(ruleId, List.of(userId), List.of(), List.of());
|
||||
repo.save(open);
|
||||
|
||||
assertThat(repo.findOpenForRule(ruleId)).isPresent();
|
||||
|
||||
repo.resolve(open.id(), Instant.now());
|
||||
|
||||
assertThat(repo.findOpenForRule(ruleId)).isEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
void ack_setsAckedAtAndState() {
|
||||
var inst = newInstance(ruleId, List.of(userId), List.of(), List.of());
|
||||
repo.save(inst);
|
||||
|
||||
Instant when = Instant.now();
|
||||
repo.ack(inst.id(), userId, when);
|
||||
|
||||
var found = repo.findById(inst.id()).orElseThrow();
|
||||
assertThat(found.state()).isEqualTo(AlertState.ACKNOWLEDGED);
|
||||
assertThat(found.ackedBy()).isEqualTo(userId);
|
||||
assertThat(found.ackedAt()).isNotNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
void resolve_setsResolvedAtAndState() {
|
||||
var inst = newInstance(ruleId, List.of(userId), List.of(), List.of());
|
||||
repo.save(inst);
|
||||
|
||||
repo.resolve(inst.id(), Instant.now());
|
||||
|
||||
var found = repo.findById(inst.id()).orElseThrow();
|
||||
assertThat(found.state()).isEqualTo(AlertState.RESOLVED);
|
||||
assertThat(found.resolvedAt()).isNotNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
void deleteResolvedBefore_deletesOnlyResolved() {
|
||||
var firing = newInstance(ruleId, List.of(userId), List.of(), List.of());
|
||||
var resolved = newInstance(ruleId, List.of(userId), List.of(), List.of());
|
||||
repo.save(firing);
|
||||
repo.save(resolved);
|
||||
|
||||
Instant resolvedTime = Instant.now().minusSeconds(10);
|
||||
repo.resolve(resolved.id(), resolvedTime);
|
||||
|
||||
repo.deleteResolvedBefore(Instant.now());
|
||||
|
||||
assertThat(repo.findById(firing.id())).isPresent();
|
||||
assertThat(repo.findById(resolved.id())).isEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
void markSilenced_togglesToTrue() {
|
||||
var inst = newInstance(ruleId, List.of(userId), List.of(), List.of());
|
||||
repo.save(inst);
|
||||
|
||||
assertThat(repo.findById(inst.id()).orElseThrow().silenced()).isFalse();
|
||||
repo.markSilenced(inst.id(), true);
|
||||
assertThat(repo.findById(inst.id()).orElseThrow().silenced()).isTrue();
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private AlertInstance newInstance(UUID ruleId,
|
||||
List<String> userIds,
|
||||
List<UUID> groupIds,
|
||||
List<String> roleNames) {
|
||||
return new AlertInstance(
|
||||
UUID.randomUUID(), ruleId, Map.of(), envId,
|
||||
AlertState.FIRING, AlertSeverity.WARNING,
|
||||
Instant.now(), null, null, null, null,
|
||||
false, null, null,
|
||||
Map.of(), "title", "message",
|
||||
userIds, groupIds, roleNames);
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,10 @@
|
||||
package com.cameleer.server.core.alerting;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
@JsonIgnoreProperties(ignoreUnknown = true)
|
||||
public record SilenceMatcher(
|
||||
UUID ruleId, String appSlug, String routeId, String agentId, AlertSeverity severity) {
|
||||
|
||||
|
||||
Reference in New Issue
Block a user