feat: add notification production capability

This commit is contained in:
donghyeon-ka
2026-07-31 23:48:58 +09:00
parent b3add0162d
commit ec4bf105c4
199 changed files with 23589 additions and 122 deletions
+30 -1
View File
@@ -56,13 +56,37 @@ Package root: `dev.caskeleton.application`.
| `usecase.QueryUseCase<Q extends Query, R>` | Inbound port for read-only use cases. Implementations MUST declare `transactionMode = READ_ONLY` and `repositoryAccess = READ_REPOSITORY`. |
| `command.Command` | Marker for write intents. Plain immutable types built from domain values. |
| `query.Query` | Marker for read intents. Plain immutable types built from domain values. |
| `transaction.TransactionPort` | Outbound port for transactional boundaries. Implemented by `adapter-persistence`. |
| `transaction.TransactionPort` | Outbound port for join-capable write/read, physical root-only write, and independent write boundaries. Implemented by `adapter-persistence`. |
| `transaction.NestedRootTransactionRejectedException` | Fail-fast signal raised before action/provider side effects when `inRootWrite` detects an actual ambient transaction. |
| `transaction.TransactionMode` | `WRITE` / `READ_ONLY` / `REQUIRES_NEW`. `NESTED` and `NEVER` are intentionally absent. |
| `transaction.Isolation` | `READ_COMMITTED` (pinned default) / `REPEATABLE_READ` / `SERIALIZABLE`. `READ_UNCOMMITTED` is forbidden (not declared); the vendor default is never used (engine defaults differ — PostgreSQL READ COMMITTED vs MySQL InnoDB REPEATABLE READ). Routing the stricter levels through `TransactionPort` is a `planned` joint change with `feature-application-port-usecase-contract`; the shipped call path pins `READ_COMMITTED`. |
| `capability.UseCaseCapability` | Mandatory annotation on every concrete use case: declares `transactionMode`, `idempotency`, `repositoryAccess`, `externalOutboundAllowed`. |
| `capability.Idempotency` | `IDEMPOTENT` / `KEYED` / `NOT_IDEMPOTENT`. |
| `capability.RepositoryAccess` | `NONE` / `READ_REPOSITORY` / `WRITE_REPOSITORY`. |
## Notification R1 application boundary
- `dev.caskeleton.application.notification` owns only framework-free semantic values, code-owned
kind policy, narrow outbound ports, dispatch/receipt/admission/reconciliation orchestration and
writer-cutover command contracts.
- Feature/application code creates a typed `NotificationIntentDraft`; `NotificationPlanPort`
returns the application-owned immutable `NotificationFrozenPlan`, which is the only planning
handoff consumed by append or inline attempt ports. Provider SDK, transport DTO, persistence
entity, compiled adapter binding and raw recipient/template payload types are forbidden here.
- Provider calls run outside database transactions. Dispatch and reconciliation use bounded
claim/authorize/finalize transactions with opaque claim/version/execution tokens; an
`INDETERMINATE` submission is terminal and must not be blindly retried.
- Receipt reduction is order-independent and keeps delivery acceptance monotonic. Only hard bounce
and complaint facts may request technical suppression; consent/unsubscribe policy is outside this
capability.
- Writer-cutover operations that must prove a physical commit use `inRootWrite`. Route/profile
registries are application-owned exact inputs; signed inventory/quiescence verification is
delegated to narrow verifier ports and the persistence operation must enforce locked durable
state/journal invariants.
- This is the R1 application contract proven with fakes. It does not claim PostgreSQL schema/locking,
provider protocol, cryptographic verifier, or runtime wiring qualification; those belong to the
notification/persistence/bootstrap adapters.
## Naming convention
- Inbound port implementations end with `UseCase` (e.g. `RegisterUserUseCase`). Enforced by ArchUnit.
@@ -103,10 +127,15 @@ application-core never self-registers with a DI framework.
| Use case shape | `transactionMode` | TransactionPort call | When |
|---|---|---|---|
| Write command | `WRITE` | `tx.inWrite(...)` | Default for `CommandUseCase`. |
| Physical-root write command | `WRITE` | `tx.inRootWrite(...)` | Only when orchestration must prove there is no ambient transaction and expose a result after commit. |
| Read-only query | `READ_ONLY` | `tx.inRead(...)` | Default for `QueryUseCase`. |
| Outbox / audit / compensation | `REQUIRES_NEW` | `tx.inNew(...)` | Only when the use case MUST commit independently of the caller. |
`NESTED` and `NEVER` propagation are forbidden.
`inRootWrite` MUST reject an actual ambient transaction before invoking its action or
`PlatformTransactionManager`; it MUST NOT emulate root-only behavior with `REQUIRES_NEW`.
Both `inWrite` and `inRootWrite` satisfy the direct boundary fitness rule for a
`WRITE_REPOSITORY + WRITE` use case. READ and REQUIRES_NEW mappings remain exclusive.
### Callback signature contract (D11)
+38 -4
View File
@@ -71,13 +71,21 @@
- **존재 이유**: application 유스케이스가 `org.springframework.transaction.annotation.Transactional`
을 import 하지 않고도 트랜잭션 의도를 선언하게 하기 위한 추상화다. 구현(보통
`SpringTransactionPort`)은 persistence adapter 가 Spring `PlatformTransactionManager` 로 제공한다.
application/domain 을 프레임워크-free 로 유지하는 핵심 장치.
- 가지 경계:
application/domain 을 프레임워크-free 로 유지하는 핵심 장치.
- 가지 경계:
- `inWrite` — REQUIRED + read-write, `READ_COMMITTED`. command 유스케이스 기본.
- `inRootWrite` — 물리 root 전용 REQUIRED + read-write, `READ_COMMITTED`. 실제 ambient
transaction 이 하나라도 있으면 action 실행 전에
`NestedRootTransactionRejectedException` 으로 거부한다. 성공 값은 commit 이 끝난 뒤에만
호출자에게 반환되며, commit 실패는 그대로 전파된다.
- `inRead` — REQUIRED + read-only, `READ_COMMITTED`. query 유스케이스 기본.
- `inNew` — REQUIRES_NEW + read-write. UseCaseCapability 에 `REQUIRES_NEW` 를 명시한
유스케이스(outbox/audit/compensation)에서만 허용.
- **콜백 시그니처(D11)**: 세 메서드 모두 `Supplier`/`Runnable` 을 받아 checked exception 을 던질
- **root-only 사용 조건**: `inRootWrite` 는 join 가능한 일반 command 경계의 대체물이 아니다.
외부 효과를 commit 이후에만 시작해야 하는 orchestration처럼 물리 root를 증명해야 하는 경우에만
쓴다. 기존 transaction 안에서 `REQUIRES_NEW` 로 몰래 분리하지 않고 fail-fast하므로, 호출자는
transaction 없는 진입점에서 이 경계를 시작해야 한다.
- **콜백 시그니처(D11)**: 네 메서드 모두 `Supplier`/`Runnable` 을 받아 checked exception 을 던질
수 없다. Spring `TransactionCallback<T>` 제약과 동일하다. 그래서 호출자는 도메인 checked
exception 을 `RuntimeException` 하위로 감싸야 한다(`DomainException extends RuntimeException`).
`IOException``UncheckedIOException`, `SQLException` 은 Spring `DataAccessException` 계층이
@@ -93,7 +101,33 @@
**금지**: 많은 레코드를 도는 루프 안에서 `inNew` 호출(예: per-row outbox dispatch). 풀 고갈 +
데드락 위험. 레코드를 한 번의 `inNew` 안에서 배치 처리하거나, 루프를 트랜잭션 경계 밖으로 빼라.
- **금지 목록**: `NESTED`/`NEVER` propagation, `READ_UNCOMMITTED` isolation, application 패키지에서
`@Transactional` 직접 사용, `inNew` 의 per-record 루프 호출.
`@Transactional` 직접 사용, `inRootWrite` 의 ambient transaction 진입, `inNew` 의 per-record
루프 호출.
---
## Notification R1 오케스트레이션 경계
`dev.caskeleton.application.notification`은 알림 vendor 구현이 아니라 알림 capability의 순수
애플리케이션 계약이다.
- 입력은 typed recipient/template value와 코드 소유 `NotificationKindPolicy`로 제한한다. feature가
만든 `NotificationIntentDraft`는 `NotificationPlanPort`에서 immutable
`NotificationFrozenPlan`으로 고정되고, append/inline 포트는 이 plan만 소비한다.
- dispatch는 claim → reserve/authorize → provider call → terminal-once finalize 순서다. 짧은 DB
transaction 사이에서 provider를 호출하며, opaque claim/version/execution token으로 stale 결과를
거부한다. submission certainty가 `INDETERMINATE`면 blind retry나 fallback을 하지 않는다.
- receipt reducer는 fact 순서와 무관한 monotonic projection을 만든다. hard bounce/complaint만
technical suppression 후보이고, business consent/unsubscribe는 다른 capability가 소유한다.
- admission/reconciliation/maintenance는 bounded batch와 주입된 `Clock`을 사용한다. scheduler는
이 유스케이스만 호출하며 store/provider 포트를 직접 조율하지 않는다.
- legacy→canonical writer cutover는 exact route/generation/profile registry, root-only commit,
서명된 inventory/quiescence evidence와 closed transition action으로 표현한다. 애플리케이션은
verifier/operation 포트의 입력 계약을 강제하고, 실제 서명 검증·행 잠금·불변 journal·provider
egress 차단은 후속 adapter 구현이 증명해야 한다.
현재 증거 등급은 **R1 application contract with fakes**다. PostgreSQL DDL/locking, provider
protocol, receipt ingress, runtime wiring을 포함한 R2/R3 완료 주장이 아니다.
### TransactionMode
@@ -0,0 +1,13 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.command.Command;
import java.util.Objects;
/** Applies one already authenticated and normalized receipt. */
public record ApplyNotificationReceiptCommand(NormalizedNotificationReceiptCommand receipt)
implements Command {
public ApplyNotificationReceiptCommand {
Objects.requireNonNull(receipt, "normalized notification receipt must be non-null");
}
}
@@ -0,0 +1,21 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/** Non-sensitive result of reducing one receipt event. */
public record ApplyNotificationReceiptResult(
Status status, NotificationReceiptProjection projection, boolean suppressionApplied) {
public ApplyNotificationReceiptResult {
Objects.requireNonNull(status, "notification receipt apply status must be non-null");
Objects.requireNonNull(projection, "notification receipt projection must be non-null");
if (status == Status.DUPLICATE && suppressionApplied) {
throw new IllegalArgumentException("duplicate receipt cannot repeat technical suppression");
}
}
public enum Status {
APPLIED,
DUPLICATE
}
}
@@ -0,0 +1,80 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.capability.Idempotency;
import dev.caskeleton.application.capability.RepositoryAccess;
import dev.caskeleton.application.capability.UseCaseCapability;
import dev.caskeleton.application.security.RequiresPermission;
import dev.caskeleton.application.transaction.TransactionMode;
import dev.caskeleton.application.transaction.TransactionPort;
import dev.caskeleton.application.usecase.CommandUseCase;
import java.time.Clock;
import java.util.Objects;
/**
* Appends one receipt fact and reduces its delivery projection in one physical root transaction.
*/
@RequiresPermission("notification:receipt")
@UseCaseCapability(
transactionMode = TransactionMode.WRITE,
idempotency = Idempotency.IDEMPOTENT,
repositoryAccess = RepositoryAccess.WRITE_REPOSITORY)
public final class ApplyNotificationReceiptUseCase
implements CommandUseCase<ApplyNotificationReceiptCommand, ApplyNotificationReceiptResult> {
private final NotificationReceiptStorePort store;
private final NotificationTechnicalSuppressionPort suppression;
private final TransactionPort transactions;
private final Clock clock;
public ApplyNotificationReceiptUseCase(
NotificationReceiptStorePort store,
NotificationTechnicalSuppressionPort suppression,
TransactionPort transactions,
Clock clock) {
this.store = Objects.requireNonNull(store, "notification receipt store must be non-null");
this.suppression =
Objects.requireNonNull(suppression, "notification suppression port must be non-null");
this.transactions = Objects.requireNonNull(transactions, "transaction port must be non-null");
this.clock = Objects.requireNonNull(clock, "clock must be non-null");
}
@Override
public ApplyNotificationReceiptResult handle(ApplyNotificationReceiptCommand command) {
Objects.requireNonNull(command, "apply notification receipt command must be non-null");
return transactions.inRootWrite(() -> applyInsideRoot(command.receipt()));
}
private ApplyNotificationReceiptResult applyInsideRoot(
NormalizedNotificationReceiptCommand command) {
NotificationReceiptStorePort.AppendResult appendResult = store.appendIfAbsent(command);
if (appendResult instanceof NotificationReceiptStorePort.Duplicate duplicate) {
return new ApplyNotificationReceiptResult(
ApplyNotificationReceiptResult.Status.DUPLICATE, duplicate.projection(), false);
}
NotificationReceiptStorePort.ReceiptAggregate aggregate =
((NotificationReceiptStorePort.Appended) appendResult).aggregate();
if (!aggregate.deliveryId().equals(command.deliveryId())) {
throw new IllegalStateException(
"receipt aggregate delivery does not match normalized command");
}
NotificationReceiptProjection projection =
NotificationReceiptProjection.reduce(aggregate.facts());
store.saveProjection(aggregate.deliveryId(), projection);
boolean suppressionApplied = shouldSuppress(command.fact());
if (suppressionApplied) {
suppression.suppress(
new NotificationTechnicalSuppressionPort.SuppressionMutation(
aggregate.recipient(), command.fact().reasonCode(), clock.instant()));
}
return new ApplyNotificationReceiptResult(
ApplyNotificationReceiptResult.Status.APPLIED, projection, suppressionApplied);
}
private static boolean shouldSuppress(NotificationReceiptFact fact) {
return fact.type() == NotificationReceiptFact.Type.COMPLAINT
|| (fact.type() == NotificationReceiptFact.Type.BOUNCE
&& fact.bounceClass() == NotificationReceiptFact.BounceClass.HARD);
}
}
@@ -0,0 +1,7 @@
package dev.caskeleton.application.notification;
/** Point at which recipient consent or preference must be established. */
public enum ConsentCheckMode {
SNAPSHOT_AT_APPEND,
RECHECK_BEFORE_EACH_DELIVERY
}
@@ -0,0 +1,19 @@
package dev.caskeleton.application.notification;
/** Opaque reference resolved to an email recipient only inside a qualified adapter. */
public record EmailRecipientReference(String reference) implements NotificationRecipientReference {
public EmailRecipientReference {
reference = NotificationIntentId.requireOpaque("email recipient reference", reference);
}
@Override
public NotificationChannel channel() {
return NotificationChannel.EMAIL;
}
@Override
public String toString() {
return "EmailRecipientReference[reference=<redacted>]";
}
}
@@ -0,0 +1,32 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.command.Command;
import java.util.Objects;
/** Reviewed all-route initialization request; partial route sets are rejected by the use case. */
public record InitializeNotificationWriterFencesCommand(
String operationToken,
NotificationCanonicalWriterRouteSet reviewedRoutes,
String reviewedRouteSetDigest,
String actorReference,
NotificationReasonCode reasonCode)
implements Command {
public InitializeNotificationWriterFencesCommand {
operationToken =
NotificationIntentId.requireOpaque("writer initialization operation token", operationToken);
Objects.requireNonNull(reviewedRoutes, "reviewed writer routes must be non-null");
reviewedRouteSetDigest = requireDigest(reviewedRouteSetDigest);
actorReference =
NotificationIntentId.requireOpaque("writer initialization actor", actorReference);
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
static String requireDigest(String digest) {
if (digest == null || !digest.matches("[0-9a-f]{64}")) {
throw new IllegalArgumentException(
"writer evidence digest must be 64 lowercase hex characters");
}
return digest;
}
}
@@ -0,0 +1,13 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
/** Atomic persistence operation for absent-fence and empty-journal initialization. */
@FunctionalInterface
public interface InitializeNotificationWriterFencesOperation {
InitializeNotificationWriterFencesResult initialize(
InitializeNotificationWriterFencesCommand command,
NotificationWriterRouteSet trustedRoutes,
Instant requestedAt);
}
@@ -0,0 +1,21 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/** Committed all-route fence initialization result. */
public record InitializeNotificationWriterFencesResult(
Status status, int initializedRouteCount, String routeSetDigest) {
public InitializeNotificationWriterFencesResult {
Objects.requireNonNull(status, "writer initialization status must be non-null");
if (initializedRouteCount < 1 || initializedRouteCount > 100) {
throw new IllegalArgumentException("initialized writer route count must be in 1..100");
}
routeSetDigest = InitializeNotificationWriterFencesCommand.requireDigest(routeSetDigest);
}
public enum Status {
INITIALIZED,
REPLAYED
}
}
@@ -0,0 +1,55 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.capability.Idempotency;
import dev.caskeleton.application.capability.RepositoryAccess;
import dev.caskeleton.application.capability.UseCaseCapability;
import dev.caskeleton.application.security.RequiresPermission;
import dev.caskeleton.application.transaction.TransactionMode;
import dev.caskeleton.application.transaction.TransactionPort;
import dev.caskeleton.application.usecase.CommandUseCase;
import java.time.Clock;
import java.util.Objects;
/** Root-commits the complete trusted writer fence and proof registry initialization batch. */
@RequiresPermission("notification:cutover")
@UseCaseCapability(
transactionMode = TransactionMode.WRITE,
idempotency = Idempotency.IDEMPOTENT,
repositoryAccess = RepositoryAccess.WRITE_REPOSITORY,
crossTenantAdmin = true)
public final class InitializeNotificationWriterFencesUseCase
implements CommandUseCase<
InitializeNotificationWriterFencesCommand, InitializeNotificationWriterFencesResult> {
private final NotificationWriterRouteSet routes;
private final InitializeNotificationWriterFencesOperation operation;
private final TransactionPort transactions;
private final Clock clock;
public InitializeNotificationWriterFencesUseCase(
NotificationWriterRouteSet routes,
InitializeNotificationWriterFencesOperation operation,
TransactionPort transactions,
Clock clock) {
this.routes = Objects.requireNonNull(routes, "notification writer route set must be non-null");
this.operation =
Objects.requireNonNull(operation, "writer initialization operation must be non-null");
this.transactions = Objects.requireNonNull(transactions, "transaction port must be non-null");
this.clock = Objects.requireNonNull(clock, "clock must be non-null");
}
@Override
public InitializeNotificationWriterFencesResult handle(
InitializeNotificationWriterFencesCommand command) {
Objects.requireNonNull(command, "writer initialization command must be non-null");
if (!command.reviewedRoutes().equals(routes.canonicalRoutes())) {
throw new IllegalArgumentException(
"reviewed writer routes must exactly equal the trusted all-route set");
}
if (!command.reviewedRouteSetDigest().equals(routes.digest())) {
throw new IllegalArgumentException(
"reviewed writer route-set digest does not match trusted registry digest");
}
return transactions.inRootWrite(() -> operation.initialize(command, routes, clock.instant()));
}
}
@@ -0,0 +1,11 @@
package dev.caskeleton.application.notification;
/**
* Executes one bounded, non-durable inline attempt over a frozen application plan. The caller must
* establish the physical root-write sequencing contract before invoking this port.
*/
@FunctionalInterface
public interface InlineNotificationAttemptPort {
NotificationRequestResult.InlineCompleted attempt(NotificationFrozenPlan plan);
}
@@ -0,0 +1,16 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/** Framework-free receipt normalized by an authenticated inbound adapter. */
public record NormalizedNotificationReceiptCommand(
NotificationReceiptEventId receiptEventId,
NotificationDeliveryId deliveryId,
NotificationReceiptFact fact) {
public NormalizedNotificationReceiptCommand {
Objects.requireNonNull(receiptEventId, "notification receipt event ID must be non-null");
Objects.requireNonNull(deliveryId, "notification delivery ID must be non-null");
Objects.requireNonNull(fact, "notification receipt fact must be non-null");
}
}
@@ -0,0 +1,8 @@
package dev.caskeleton.application.notification;
/** Code-owned dispatch admission and fairness class. */
public enum NotificationAdmissionClass {
SECURITY_CRITICAL,
TRANSACTIONAL,
BULK_LOW_VALUE
}
@@ -0,0 +1,53 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.command.Command;
import java.util.Objects;
/** Audited operator request to re-probe and resume one shared notification admission gate. */
public record NotificationAdmissionGateCommand(
String operationToken,
NotificationRouteId routeId,
int policyRevision,
NotificationFaultScope faultScope,
String scopeReference,
long expectedGeneration,
int maximumParkedLegs,
String actorReference,
NotificationReasonCode reasonCode)
implements Command {
public NotificationAdmissionGateCommand {
operationToken =
NotificationIntentId.requireOpaque("admission resume operation token", operationToken);
Objects.requireNonNull(routeId, "notification route ID must be non-null");
if (policyRevision < 1 || expectedGeneration < 0) {
throw new IllegalArgumentException(
"policy revision must be positive and expected generation non-negative");
}
if (maximumParkedLegs < 1 || maximumParkedLegs > 100) {
throw new IllegalArgumentException("maximum parked legs must be in 1..100");
}
Objects.requireNonNull(faultScope, "notification fault scope must be non-null");
if (faultScope == NotificationFaultScope.DELIVERY) {
throw new IllegalArgumentException("operator admission command cannot target DELIVERY scope");
}
scopeReference =
NotificationIntentId.requireOpaque("admission scope reference", scopeReference);
actorReference =
NotificationIntentId.requireOpaque("admission resume actor reference", actorReference);
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
NotificationAdmissionReadinessPort.ResumeRequest toResumeRequest() {
return new NotificationAdmissionReadinessPort.ResumeRequest(
operationToken,
routeId,
policyRevision,
faultScope,
scopeReference,
expectedGeneration,
maximumParkedLegs,
actorReference,
reasonCode);
}
}
@@ -0,0 +1,89 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.capability.Idempotency;
import dev.caskeleton.application.capability.RepositoryAccess;
import dev.caskeleton.application.capability.UseCaseCapability;
import dev.caskeleton.application.security.RequiresPermission;
import dev.caskeleton.application.transaction.TransactionMode;
import dev.caskeleton.application.transaction.TransactionPort;
import dev.caskeleton.application.usecase.CommandUseCase;
import java.time.Clock;
import java.util.Objects;
/** Probes readiness outside a transaction and generation-CAS resumes inside one short write. */
@RequiresPermission("notification:operate")
@UseCaseCapability(
transactionMode = TransactionMode.WRITE,
idempotency = Idempotency.IDEMPOTENT,
repositoryAccess = RepositoryAccess.WRITE_REPOSITORY,
externalOutboundAllowed = true,
crossTenantAdmin = true)
public final class NotificationAdmissionGateUseCase
implements CommandUseCase<
NotificationAdmissionGateCommand, NotificationAdmissionGateUseCase.Result> {
private final NotificationAdmissionReadinessPort admission;
private final TransactionPort transactions;
private final Clock clock;
public NotificationAdmissionGateUseCase(
NotificationAdmissionReadinessPort admission, TransactionPort transactions, Clock clock) {
this.admission =
Objects.requireNonNull(admission, "notification admission port must be non-null");
this.transactions = Objects.requireNonNull(transactions, "transaction port must be non-null");
this.clock = Objects.requireNonNull(clock, "clock must be non-null");
}
@Override
public Result handle(NotificationAdmissionGateCommand command) {
Objects.requireNonNull(command, "notification admission command must be non-null");
NotificationAdmissionReadinessPort.ResumeRequest request = command.toResumeRequest();
NotificationAdmissionReadinessPort.ReadinessProbe probe =
Objects.requireNonNull(admission.probe(request), "readiness probe must be non-null");
if (!probe.ready()) {
return new Result(Result.Status.NOT_READY, probe.reasonCode());
}
NotificationAdmissionReadinessPort.ResumeResult resumeResult =
Objects.requireNonNull(
transactions.inWrite(() -> admission.resume(request, probe, clock.instant())),
"notification admission resume result must be non-null");
validateResumeResult(request, resumeResult);
return switch (resumeResult.status()) {
case RESUMED -> new Result(Result.Status.RESUMED, probe.reasonCode());
case ALREADY_ACTIVE -> new Result(Result.Status.ALREADY_ACTIVE, probe.reasonCode());
case STALE_GENERATION ->
new Result(
Result.Status.STALE_GENERATION,
new NotificationReasonCode("STALE_ADMISSION_GENERATION"));
};
}
private static void validateResumeResult(
NotificationAdmissionReadinessPort.ResumeRequest request,
NotificationAdmissionReadinessPort.ResumeResult result) {
if (result.processedLegCount() > request.maximumParkedLegs()) {
throw new IllegalArgumentException(
"admission resume processed more parked legs than the requested bound");
}
if (result.status() == NotificationAdmissionReadinessPort.ResumeStatus.RESUMED
&& result.resultingGeneration() != request.expectedGeneration() + 1) {
throw new IllegalArgumentException(
"resumed admission gate must advance the exact expected generation");
}
}
public record Result(Status status, NotificationReasonCode reasonCode) {
public Result {
Objects.requireNonNull(status, "notification admission result status must be non-null");
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
public enum Status {
RESUMED,
ALREADY_ACTIVE,
NOT_READY,
STALE_GENERATION
}
}
}
@@ -0,0 +1,154 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
import java.util.Objects;
/** Persists shared route/provider/account admission readiness with generation-guarded CAS. */
@FunctionalInterface
public interface NotificationAdmissionReadinessPort {
ParkResult park(ParkRequest request);
default ReadinessProbe probe(ResumeRequest request) {
throw new UnsupportedOperationException("notification readiness probe is not implemented");
}
default ResumeResult resume(ResumeRequest request, ReadinessProbe probe, Instant resumedAt) {
throw new UnsupportedOperationException("notification admission resume is not implemented");
}
record ParkRequest(
NotificationRouteId routeId,
int policyRevision,
NotificationFaultScope faultScope,
String scopeReference,
long expectedGeneration,
NotificationReasonCode reasonCode,
Instant parkedAt) {
public ParkRequest {
Objects.requireNonNull(routeId, "notification route ID must be non-null");
if (policyRevision < 1 || expectedGeneration < 0) {
throw new IllegalArgumentException(
"policy revision must be positive and expected generation non-negative");
}
Objects.requireNonNull(faultScope, "notification fault scope must be non-null");
if (faultScope == NotificationFaultScope.DELIVERY) {
throw new IllegalArgumentException("shared admission gate cannot use DELIVERY fault scope");
}
scopeReference =
NotificationIntentId.requireOpaque("admission scope reference", scopeReference);
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
Objects.requireNonNull(parkedAt, "admission parked time must be non-null");
}
}
enum ParkResult {
NOT_REQUESTED,
PARKED,
ALREADY_PARKED,
STALE_GENERATION
}
record ResumeRequest(
String operationToken,
NotificationRouteId routeId,
int policyRevision,
NotificationFaultScope faultScope,
String scopeReference,
long expectedGeneration,
int maximumParkedLegs,
String actorReference,
NotificationReasonCode reasonCode) {
public ResumeRequest {
operationToken =
NotificationIntentId.requireOpaque("admission resume operation token", operationToken);
Objects.requireNonNull(routeId, "notification route ID must be non-null");
if (policyRevision < 1 || expectedGeneration < 0) {
throw new IllegalArgumentException(
"policy revision must be positive and expected generation non-negative");
}
if (maximumParkedLegs < 1 || maximumParkedLegs > 100) {
throw new IllegalArgumentException("maximum parked legs must be in 1..100");
}
Objects.requireNonNull(faultScope, "notification fault scope must be non-null");
if (faultScope == NotificationFaultScope.DELIVERY) {
throw new IllegalArgumentException("shared admission gate cannot use DELIVERY fault scope");
}
scopeReference =
NotificationIntentId.requireOpaque("admission scope reference", scopeReference);
actorReference =
NotificationIntentId.requireOpaque("admission resume actor reference", actorReference);
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
}
record ReadinessProbe(boolean ready, NotificationReasonCode reasonCode) {
public ReadinessProbe {
Objects.requireNonNull(reasonCode, "notification readiness reason must be non-null");
}
}
enum ResumeStatus {
RESUMED,
ALREADY_ACTIVE,
STALE_GENERATION
}
/**
* Audited bounded result of rechecking every selected parked leg inside the gate-resume
* transaction. Initial R1 never activates fallback while resuming a binding park.
*/
record ResumeResult(
ResumeStatus status,
long resultingGeneration,
int queuedCount,
int expiredCount,
int cancelledCount,
int technicallySuppressedCount,
int policyRejectedCount,
int activatedFallbackCount) {
public ResumeResult {
Objects.requireNonNull(status, "notification admission resume status must be non-null");
if (resultingGeneration < 0
|| queuedCount < 0
|| expiredCount < 0
|| cancelledCount < 0
|| technicallySuppressedCount < 0
|| policyRejectedCount < 0
|| activatedFallbackCount < 0) {
throw new IllegalArgumentException(
"notification admission resume generation/counts must be non-negative");
}
int processedLegCount =
queuedCount
+ expiredCount
+ cancelledCount
+ technicallySuppressedCount
+ policyRejectedCount;
if (processedLegCount > 100) {
throw new IllegalArgumentException(
"notification admission resume leg count must be bounded by 100");
}
if (activatedFallbackCount != 0) {
throw new IllegalArgumentException(
"binding-park resume must not activate initial fallback legs");
}
if (status != ResumeStatus.RESUMED && processedLegCount != 0) {
throw new IllegalArgumentException(
"non-mutating admission resume status cannot report processed legs");
}
}
public int processedLegCount() {
return queuedCount
+ expiredCount
+ cancelledCount
+ technicallySuppressedCount
+ policyRejectedCount;
}
}
}
@@ -0,0 +1,31 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/** Durable append result; neither appended nor duplicate means provider delivery succeeded. */
public sealed interface NotificationAppendResult
permits NotificationAppendResult.Appended,
NotificationAppendResult.DuplicateExisting,
NotificationAppendResult.Rejected {
record Appended(NotificationIntentId intentId) implements NotificationAppendResult {
public Appended {
Objects.requireNonNull(intentId, "notification intent ID must be non-null");
}
}
record DuplicateExisting(NotificationIntentId intentId) implements NotificationAppendResult {
public DuplicateExisting {
Objects.requireNonNull(intentId, "notification intent ID must be non-null");
}
}
record Rejected(NotificationReasonCode reasonCode) implements NotificationAppendResult {
public Rejected {
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
}
}
@@ -0,0 +1,20 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/** Framework- and provider-neutral application failure carrying only a stable reason code. */
public final class NotificationApplicationException extends RuntimeException {
private final NotificationReasonCode reasonCode;
public NotificationApplicationException(NotificationReasonCode reasonCode, Throwable cause) {
super(
Objects.requireNonNull(reasonCode, "notification reason code must be non-null").value(),
cause);
this.reasonCode = reasonCode;
}
public NotificationReasonCode reasonCode() {
return reasonCode;
}
}
@@ -0,0 +1,9 @@
package dev.caskeleton.application.notification;
/** Opaque identity of one authorized physical provider attempt. */
public record NotificationAttemptId(String value) {
public NotificationAttemptId {
value = NotificationIntentId.requireOpaque("attemptId", value);
}
}
@@ -0,0 +1,44 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/**
* Internal application collaborator asserting canonical ownership in an existing write boundary.
*/
public final class NotificationCanonicalWriterFenceGuard {
private final NotificationCanonicalWriterFencePort fence;
private final NotificationCanonicalWriterRouteSet routes;
public NotificationCanonicalWriterFenceGuard(
NotificationCanonicalWriterFencePort fence, NotificationCanonicalWriterRouteSet routes) {
this.fence = Objects.requireNonNull(fence, "canonical writer fence port must be non-null");
this.routes = Objects.requireNonNull(routes, "canonical writer route set must be non-null");
}
public void assertCanonical(
NotificationCanonicalWriterRouteSet.RouteRevision route, long expectedGeneration) {
if (!routes.contains(route)) {
throw new IllegalArgumentException(
"route is outside canonical notification writer route set");
}
NotificationCanonicalWriterFencePort.FenceSnapshot snapshot =
Objects.requireNonNull(
fence.assertCanonicalInCallerTransaction(
new NotificationCanonicalWriterFencePort.FenceRequest(route, expectedGeneration)),
"canonical writer fence snapshot must be non-null");
if (!snapshot.route().equals(route)) {
throw failure("CANONICAL_WRITER_ROUTE_MISMATCH");
}
if (snapshot.owner() != NotificationWriterOwnership.CANONICAL) {
throw failure("CANONICAL_WRITER_NOT_OWNER");
}
if (snapshot.generation() != expectedGeneration) {
throw failure("STALE_CANONICAL_WRITER_GENERATION");
}
}
private static NotificationApplicationException failure(String reason) {
return new NotificationApplicationException(new NotificationReasonCode(reason), null);
}
}
@@ -0,0 +1,38 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/**
* Acquires a transaction-scoped shared fence assertion. The persistence implementation must hold
* the share lock until the caller's physical commit or rollback.
*/
@FunctionalInterface
public interface NotificationCanonicalWriterFencePort {
FenceSnapshot assertCanonicalInCallerTransaction(FenceRequest request);
record FenceRequest(
NotificationCanonicalWriterRouteSet.RouteRevision route, long expectedGeneration) {
public FenceRequest {
Objects.requireNonNull(route, "notification writer route must be non-null");
if (expectedGeneration < 0) {
throw new IllegalArgumentException("expected writer generation must be non-negative");
}
}
}
record FenceSnapshot(
NotificationCanonicalWriterRouteSet.RouteRevision route,
NotificationWriterOwnership owner,
long generation) {
public FenceSnapshot {
Objects.requireNonNull(route, "notification writer route must be non-null");
Objects.requireNonNull(owner, "notification writer owner must be non-null");
if (generation < 0) {
throw new IllegalArgumentException("writer generation must be non-negative");
}
}
}
}
@@ -0,0 +1,79 @@
package dev.caskeleton.application.notification;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
/** Bounded, ordered canonical route-revision set that production admission is allowed to use. */
public record NotificationCanonicalWriterRouteSet(List<RouteRevision> routes) {
public NotificationCanonicalWriterRouteSet {
Objects.requireNonNull(routes, "canonical notification writer routes must be non-null");
routes =
routes.stream()
.map(route -> Objects.requireNonNull(route, "canonical route must be non-null"))
.sorted(
Comparator.comparing((RouteRevision route) -> route.routeId().value())
.thenComparingInt(RouteRevision::routeRevision))
.toList();
if (routes.isEmpty() || routes.size() > 100) {
throw new IllegalArgumentException("canonical writer route set must contain 1..100 routes");
}
if (new HashSet<>(routes).size() != routes.size()) {
throw new IllegalArgumentException("canonical writer route set contains a duplicate route");
}
long distinctRouteKeys = routes.stream().map(RouteRevision::routeId).distinct().count();
if (distinctRouteKeys != routes.size()) {
throw new IllegalArgumentException(
"canonical writer route set contains multiple revisions for one route key");
}
}
public boolean contains(RouteRevision route) {
return routes.contains(route);
}
public String digest() {
MessageDigest digest = sha256();
routes.forEach(
route -> {
update(digest, route.routeId().value());
digest.update(ByteBuffer.allocate(Integer.BYTES).putInt(route.routeRevision()).array());
digest.update(
ByteBuffer.allocate(Long.BYTES).putLong(route.predecessorGeneration()).array());
});
return java.util.HexFormat.of().formatHex(digest.digest());
}
private static MessageDigest sha256() {
try {
return MessageDigest.getInstance("SHA-256");
} catch (NoSuchAlgorithmException unavailable) {
throw new IllegalStateException(
"SHA-256 must be available on every Java runtime", unavailable);
}
}
static void update(MessageDigest digest, String value) {
byte[] encoded = value.getBytes(StandardCharsets.UTF_8);
digest.update(ByteBuffer.allocate(Integer.BYTES).putInt(encoded.length).array());
digest.update(encoded);
}
public record RouteRevision(
NotificationRouteId routeId, int routeRevision, long predecessorGeneration) {
public RouteRevision {
Objects.requireNonNull(routeId, "notification route ID must be non-null");
if (routeRevision < 1 || routeRevision > 1_000_000 || predecessorGeneration < 0) {
throw new IllegalArgumentException(
"route revision must be in 1..1000000 and predecessor generation non-negative");
}
}
}
}
@@ -0,0 +1,96 @@
package dev.caskeleton.application.notification;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
/** Pure validator over application-owned policy and provider/store/ingress capability facts. */
public final class NotificationCapabilityCompatibilityValidator {
public Compatibility validate(
NotificationKindPolicy policy,
NotificationProviderCapabilityDescriptor provider,
NotificationStoreCapabilityDescriptor store,
Optional<NotificationReceiptIngressCapabilityDescriptor> receiptIngress,
boolean receiptRequired) {
Objects.requireNonNull(policy, "notification kind policy must be non-null");
Objects.requireNonNull(provider, "notification provider descriptor must be non-null");
Objects.requireNonNull(store, "notification store descriptor must be non-null");
Objects.requireNonNull(receiptIngress, "receipt ingress container must be non-null");
List<NotificationReasonCode> reasons = new ArrayList<>();
addIf(reasons, provider.channel() != policy.channel(), "PROVIDER_CHANNEL_MISMATCH");
addIf(reasons, !provider.supportedModes().contains(policy.mode()), "PROVIDER_MODE_UNSUPPORTED");
addIf(reasons, !provider.hiddenRetriesControlled(), "PROVIDER_HIDDEN_RETRY_UNCONTROLLED");
addIf(
reasons,
provider.maximumTargets() < policy.maxTargetsPerRecipient(),
"PROVIDER_TARGET_BOUND_INSUFFICIENT");
addIf(
reasons,
policy.maxReconcileCalls() > 0 && !provider.reconciliationSupported(),
"PROVIDER_RECONCILIATION_UNSUPPORTED");
if (policy.mode() == NotificationMode.DURABLE_ASYNC) {
addIf(
reasons,
!store.durableIntentStore() || !store.attemptJournal(),
"DURABLE_STORE_UNAVAILABLE");
}
addIf(
reasons,
!store.availablePolicyRevisions().contains(policy.policyRevision()),
"POLICY_REVISION_UNAVAILABLE");
addIf(
reasons,
!store.availableTemplateRevisions().contains(policy.templateRef()),
"TEMPLATE_REVISION_UNAVAILABLE");
if (receiptRequired) {
addIf(reasons, !provider.receiptSupported(), "PROVIDER_RECEIPT_UNSUPPORTED");
addIf(reasons, !store.receiptInbox(), "RECEIPT_STORE_UNAVAILABLE");
boolean ingressUnavailable =
receiptIngress.isEmpty()
|| !receiptIngress.orElseThrow().enabled()
|| !receiptIngress.orElseThrow().authenticated()
|| receiptIngress.orElseThrow().channel() != policy.channel()
|| receiptIngress.orElseThrow().supportedFactTypes().isEmpty();
addIf(reasons, ingressUnavailable, "RECEIPT_INGRESS_UNAVAILABLE");
}
return new Compatibility(reasons.isEmpty(), reasons);
}
public void requireCompatible(
NotificationKindPolicy policy,
NotificationProviderCapabilityDescriptor provider,
NotificationStoreCapabilityDescriptor store,
Optional<NotificationReceiptIngressCapabilityDescriptor> receiptIngress,
boolean receiptRequired) {
Compatibility compatibility =
validate(policy, provider, store, receiptIngress, receiptRequired);
if (!compatibility.compatible()) {
throw new NotificationApplicationException(
new NotificationReasonCode("NOTIFICATION_CAPABILITY_INCOMPATIBLE"), null);
}
}
private static void addIf(
List<NotificationReasonCode> reasons, boolean condition, String reasonCode) {
if (condition) {
reasons.add(new NotificationReasonCode(reasonCode));
}
}
public record Compatibility(boolean compatible, List<NotificationReasonCode> reasonCodes) {
public Compatibility {
reasonCodes =
List.copyOf(
Objects.requireNonNull(
reasonCodes, "notification compatibility reasons must be non-null"));
if (compatible != reasonCodes.isEmpty()) {
throw new IllegalArgumentException(
"compatible flag must equal an empty incompatibility reason set");
}
}
}
}
@@ -0,0 +1,7 @@
package dev.caskeleton.application.notification;
/** Provider-neutral delivery medium. */
public enum NotificationChannel {
EMAIL,
SLACK
}
@@ -0,0 +1,9 @@
package dev.caskeleton.application.notification;
/** Opaque identity of one provider leg for a logical recipient. */
public record NotificationDeliveryId(String value) {
public NotificationDeliveryId {
value = NotificationIntentId.requireOpaque("deliveryId", value);
}
}
@@ -0,0 +1,267 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
import java.util.List;
import java.util.Objects;
/** Durable delivery-leg state port; provider I/O is deliberately absent from this contract. */
public interface NotificationDeliveryStorePort {
List<ClaimedDelivery> claimEligible(int maximumClaims, Instant now);
AttemptAuthorization reserveAndAuthorize(ClaimedDelivery claimed, Instant now);
FinalizationResult finalizeAttempt(
AuthorizedAttempt attempt, AttemptFinalization finalization, Instant now);
List<ReconciliationClaim> claimForReconciliation(int maximumClaims, Instant now);
ReconciliationFinalizationResult finalizeReconciliation(
ReconciliationClaim claim,
NotificationReconciliationPort.ReconciliationOutcome outcome,
Instant now);
int attachOrphanReceipts(int maximumAttachments, Instant now);
record ClaimedDelivery(
NotificationDeliveryId deliveryId,
NotificationFrozenPlan plan,
int targetOrdinal,
String claimToken,
long rowVersion,
long admissionGeneration) {
public ClaimedDelivery {
Objects.requireNonNull(deliveryId, "notification delivery ID must be non-null");
Objects.requireNonNull(plan, "notification frozen plan must be non-null");
if (targetOrdinal < 0 || targetOrdinal >= plan.policy().maxTargetsPerRecipient()) {
throw new IllegalArgumentException("target ordinal is outside the frozen plan bound");
}
claimToken = NotificationIntentId.requireOpaque("claim token", claimToken);
if (rowVersion < 0 || admissionGeneration < 0) {
throw new IllegalArgumentException(
"row version and admission generation must be non-negative");
}
}
}
sealed interface AttemptAuthorization permits Authorized, StaleClaim, NotEligible {}
record Authorized(AuthorizedAttempt attempt) implements AttemptAuthorization {
public Authorized {
Objects.requireNonNull(attempt, "authorized notification attempt must be non-null");
}
}
record StaleClaim(NotificationDeliveryId deliveryId, NotificationReasonCode reasonCode)
implements AttemptAuthorization {
public StaleClaim {
Objects.requireNonNull(deliveryId, "notification delivery ID must be non-null");
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
}
record NotEligible(NotificationDeliveryId deliveryId, NotificationReasonCode reasonCode)
implements AttemptAuthorization {
public NotEligible {
Objects.requireNonNull(deliveryId, "notification delivery ID must be non-null");
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
}
record AuthorizedAttempt(
NotificationDeliveryId deliveryId,
NotificationAttemptId attemptId,
NotificationFrozenPlan plan,
int targetOrdinal,
String claimToken,
String executionToken,
long expectedRowVersion,
long admissionGeneration,
String admissionScopeReference,
Instant absoluteDeadline) {
public AuthorizedAttempt {
Objects.requireNonNull(deliveryId, "notification delivery ID must be non-null");
Objects.requireNonNull(attemptId, "notification attempt ID must be non-null");
Objects.requireNonNull(plan, "notification frozen plan must be non-null");
if (targetOrdinal < 0 || targetOrdinal >= plan.policy().maxTargetsPerRecipient()) {
throw new IllegalArgumentException("target ordinal is outside the frozen plan bound");
}
claimToken = NotificationIntentId.requireOpaque("claim token", claimToken);
executionToken =
NotificationIntentId.requireOpaque("attempt execution token", executionToken);
if (claimToken.equals(executionToken)) {
throw new IllegalArgumentException(
"attempt execution token must be distinct from the claim token");
}
if (expectedRowVersion < 0 || admissionGeneration < 0) {
throw new IllegalArgumentException(
"row version and admission generation must be non-negative");
}
admissionScopeReference =
NotificationIntentId.requireOpaque("admission scope reference", admissionScopeReference);
Objects.requireNonNull(absoluteDeadline, "absolute attempt deadline must be non-null");
}
@Override
public String toString() {
return "AuthorizedAttempt[deliveryId=<redacted>, attemptId="
+ attemptId
+ ", plan=<redacted>, targetOrdinal="
+ targetOrdinal
+ ", claimToken=<redacted>, executionToken=<redacted>, expectedRowVersion="
+ expectedRowVersion
+ ", admissionGeneration="
+ admissionGeneration
+ ", admissionScopeReference=<redacted>, absoluteDeadline="
+ absoluteDeadline
+ "]";
}
}
record AttemptFinalization(
ProviderAttemptOutcome providerOutcome,
TerminalState terminalState,
boolean fallbackEligible,
NotificationAdmissionReadinessPort.ParkResult parkResult) {
public AttemptFinalization {
Objects.requireNonNull(providerOutcome, "provider attempt outcome must be non-null");
Objects.requireNonNull(terminalState, "notification terminal state must be non-null");
Objects.requireNonNull(parkResult, "admission park result must be non-null");
if (fallbackEligible
&& providerOutcome.submissionCertainty() != SubmissionCertainty.DEFINITELY_NOT_APPLIED) {
throw new IllegalArgumentException("fallback is eligible only for DEFINITELY_NOT_APPLIED");
}
if (terminalState == TerminalState.TERMINAL_INDETERMINATE
&& providerOutcome.submissionCertainty() != SubmissionCertainty.INDETERMINATE) {
throw new IllegalArgumentException(
"TERMINAL_INDETERMINATE requires an indeterminate provider outcome");
}
if (terminalState == TerminalState.PARKED_BINDING
&& providerOutcome.retryDisposition() != RetryDisposition.PARK_BINDING) {
throw new IllegalArgumentException("PARKED_BINDING requires PARK_BINDING disposition");
}
if (providerOutcome.retryDisposition() == RetryDisposition.PARK_BINDING) {
if (parkResult == NotificationAdmissionReadinessPort.ParkResult.NOT_REQUESTED) {
throw new IllegalArgumentException("PARK_BINDING requires an admission park result");
}
boolean parked =
parkResult == NotificationAdmissionReadinessPort.ParkResult.PARKED
|| parkResult == NotificationAdmissionReadinessPort.ParkResult.ALREADY_PARKED;
TerminalState expected =
parked ? TerminalState.PARKED_BINDING : TerminalState.RETRY_SCHEDULED;
if (terminalState != expected) {
throw new IllegalArgumentException(
"terminal state must reflect the generation-guarded admission park result");
}
}
}
}
enum TerminalState {
ACCEPTED,
RETRY_SCHEDULED,
PARKED_BINDING,
TERMINAL_FAILURE,
TERMINAL_INDETERMINATE
}
enum FinalizationResult {
APPLIED,
LATE_EXACT_APPLIED,
STALE_EXECUTION_TOKEN,
ALREADY_TERMINAL
}
record ReconciliationClaim(
NotificationDeliveryId deliveryId,
String executionToken,
long expectedRowVersion,
NotificationRouteId routeId,
int routeRevision,
String bindingDigest,
int targetOrdinal,
String targetReference,
String providerCapabilityReference,
String providerBindingRevision,
String credentialGeneration,
String lookupReference,
ReconciliationLookupKind lookupKind,
Instant absoluteDeadline) {
public ReconciliationClaim {
Objects.requireNonNull(deliveryId, "notification delivery ID must be non-null");
executionToken =
NotificationIntentId.requireOpaque("reconciliation execution token", executionToken);
if (expectedRowVersion < 0 || routeRevision < 1 || targetOrdinal < 0 || targetOrdinal > 15) {
throw new IllegalArgumentException(
"reconciliation row version, route revision and target ordinal are invalid");
}
Objects.requireNonNull(routeId, "reconciliation route ID must be non-null");
if (bindingDigest == null || !bindingDigest.matches("[0-9a-f]{64}")) {
throw new IllegalArgumentException(
"reconciliation binding digest must be a lowercase SHA-256 digest");
}
targetReference =
NotificationIntentId.requireOpaque("reconciliation target reference", targetReference);
providerCapabilityReference =
NotificationIntentId.requireOpaque(
"reconciliation provider capability reference", providerCapabilityReference);
providerBindingRevision =
NotificationIntentId.requireOpaque(
"reconciliation provider binding revision", providerBindingRevision);
credentialGeneration =
NotificationIntentId.requireSlug(
"reconciliation credential generation", credentialGeneration);
lookupReference =
NotificationIntentId.requireOpaque(
"provider reconciliation lookup reference", lookupReference);
Objects.requireNonNull(lookupKind, "provider reconciliation lookup kind must be non-null");
Objects.requireNonNull(absoluteDeadline, "reconciliation deadline must be non-null");
}
@Override
public String toString() {
return "ReconciliationClaim[deliveryId=<redacted>, executionToken=<redacted>, "
+ "expectedRowVersion="
+ expectedRowVersion
+ ", routeId="
+ routeId
+ ", routeRevision="
+ routeRevision
+ ", bindingDigest="
+ bindingDigest
+ ", targetOrdinal="
+ targetOrdinal
+ ", targetReference=<redacted>, providerCapabilityReference="
+ providerCapabilityReference
+ ", providerBindingRevision="
+ providerBindingRevision
+ ", credentialGeneration="
+ credentialGeneration
+ ", lookupReference=<redacted>, lookupKind="
+ lookupKind
+ ", absoluteDeadline="
+ absoluteDeadline
+ "]";
}
}
enum ReconciliationLookupKind {
PRE_SEND_CORRELATION,
CLIENT_OPERATION_KEY,
MESSAGE_REFERENCE
}
enum ReconciliationFinalizationResult {
APPLIED,
LATE_EXACT_APPLIED,
STALE_EXECUTION_TOKEN,
ALREADY_TERMINAL
}
}
@@ -0,0 +1,13 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.command.Command;
/** Requests one bounded cross-tenant dispatch cycle. */
public record NotificationDispatchCommand(int maximumClaims) implements Command {
public NotificationDispatchCommand {
if (maximumClaims < 1 || maximumClaims > 100) {
throw new IllegalArgumentException("maximum notification claims must be in 1..100");
}
}
}
@@ -0,0 +1,37 @@
package dev.caskeleton.application.notification;
/** Bounded non-sensitive aggregate outcome of one dispatch cycle. */
public record NotificationDispatchResult(
int claimedCount,
int authorizedCount,
int providerCallCount,
int finalizedCount,
int staleClaimCount,
int indeterminateCount,
int parkedCount) {
public NotificationDispatchResult {
int[] counts = {
claimedCount,
authorizedCount,
providerCallCount,
finalizedCount,
staleClaimCount,
indeterminateCount,
parkedCount
};
for (int count : counts) {
if (count < 0 || count > 100) {
throw new IllegalArgumentException("notification dispatch counts must be in 0..100");
}
}
if (authorizedCount > claimedCount
|| providerCallCount > authorizedCount
|| finalizedCount > providerCallCount
|| staleClaimCount > claimedCount
|| indeterminateCount > providerCallCount
|| parkedCount > providerCallCount) {
throw new IllegalArgumentException("notification dispatch counts are inconsistent");
}
}
}
@@ -0,0 +1,199 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.capability.Idempotency;
import dev.caskeleton.application.capability.RepositoryAccess;
import dev.caskeleton.application.capability.UseCaseCapability;
import dev.caskeleton.application.security.RequiresPermission;
import dev.caskeleton.application.transaction.TransactionMode;
import dev.caskeleton.application.transaction.TransactionPort;
import dev.caskeleton.application.usecase.CommandUseCase;
import java.time.Clock;
import java.util.List;
import java.util.Objects;
/** Coordinates short store transactions around provider I/O for a bounded delivery batch. */
@RequiresPermission("notification:dispatch")
@UseCaseCapability(
transactionMode = TransactionMode.WRITE,
idempotency = Idempotency.IDEMPOTENT,
repositoryAccess = RepositoryAccess.WRITE_REPOSITORY,
externalOutboundAllowed = true,
sensitiveRead = true,
crossTenantAdmin = true)
public final class NotificationDispatchUseCase
implements CommandUseCase<NotificationDispatchCommand, NotificationDispatchResult> {
private final NotificationDeliveryStorePort store;
private final NotificationProviderAttemptPort provider;
private final NotificationAdmissionReadinessPort admission;
private final TransactionPort transactions;
private final Clock clock;
public NotificationDispatchUseCase(
NotificationDeliveryStorePort store,
NotificationProviderAttemptPort provider,
NotificationAdmissionReadinessPort admission,
TransactionPort transactions,
Clock clock) {
this.store = Objects.requireNonNull(store, "notification delivery store must be non-null");
this.provider = Objects.requireNonNull(provider, "notification provider port must be non-null");
this.admission =
Objects.requireNonNull(admission, "notification admission port must be non-null");
this.transactions = Objects.requireNonNull(transactions, "transaction port must be non-null");
this.clock = Objects.requireNonNull(clock, "clock must be non-null");
}
@Override
public NotificationDispatchResult handle(NotificationDispatchCommand command) {
Objects.requireNonNull(command, "notification dispatch command must be non-null");
List<NotificationDeliveryStorePort.ClaimedDelivery> claimed =
List.copyOf(
transactions.inWrite(
() -> store.claimEligible(command.maximumClaims(), clock.instant())));
if (claimed.size() > command.maximumClaims()) {
throw new IllegalStateException("notification store returned more claims than requested");
}
MutableCounts counts = new MutableCounts(claimed.size());
for (NotificationDeliveryStorePort.ClaimedDelivery delivery : claimed) {
dispatchOne(delivery, counts);
}
return counts.toResult();
}
private void dispatchOne(
NotificationDeliveryStorePort.ClaimedDelivery delivery, MutableCounts counts) {
NotificationDeliveryStorePort.AttemptAuthorization authorization =
transactions.inWrite(() -> store.reserveAndAuthorize(delivery, clock.instant()));
if (authorization instanceof NotificationDeliveryStorePort.StaleClaim) {
counts.staleClaims++;
return;
}
if (authorization instanceof NotificationDeliveryStorePort.NotEligible) {
return;
}
NotificationDeliveryStorePort.AuthorizedAttempt attempt =
((NotificationDeliveryStorePort.Authorized) authorization).attempt();
counts.authorized++;
ProviderAttemptOutcome outcome = invokeProvider(attempt);
counts.providerCalls++;
if (outcome.submissionCertainty() == SubmissionCertainty.INDETERMINATE) {
counts.indeterminate++;
}
FinalizationExecution execution =
transactions.inWrite(() -> finalizeInsideTransaction(attempt, outcome));
boolean applied =
execution.result() == NotificationDeliveryStorePort.FinalizationResult.APPLIED
|| execution.result()
== NotificationDeliveryStorePort.FinalizationResult.LATE_EXACT_APPLIED;
if (applied) {
counts.finalized++;
}
if (applied
&& execution.finalization().terminalState()
== NotificationDeliveryStorePort.TerminalState.PARKED_BINDING) {
counts.parked++;
}
}
private ProviderAttemptOutcome invokeProvider(
NotificationDeliveryStorePort.AuthorizedAttempt attempt) {
try {
return Objects.requireNonNull(
provider.attempt(attempt), "provider attempt outcome must be non-null");
} catch (RuntimeException providerFailure) {
return new ProviderAttemptOutcome(
SubmissionCertainty.INDETERMINATE,
RetryDisposition.NOT_APPLICABLE,
NotificationFaultScope.DELIVERY,
new NotificationReasonCode("UNCLASSIFIED_PROVIDER_FAILURE"),
java.util.Optional.empty(),
attempt.executionToken(),
java.util.Optional.empty());
}
}
private FinalizationExecution finalizeInsideTransaction(
NotificationDeliveryStorePort.AuthorizedAttempt attempt, ProviderAttemptOutcome outcome) {
NotificationAdmissionReadinessPort.ParkResult parkResult =
NotificationAdmissionReadinessPort.ParkResult.NOT_REQUESTED;
if (outcome.retryDisposition() == RetryDisposition.PARK_BINDING) {
parkResult =
admission.park(
new NotificationAdmissionReadinessPort.ParkRequest(
attempt.plan().routeId(),
attempt.plan().policy().policyRevision(),
outcome.faultScope(),
attempt.admissionScopeReference(),
attempt.admissionGeneration(),
outcome.reasonCode(),
clock.instant()));
}
NotificationDeliveryStorePort.AttemptFinalization finalization =
new NotificationDeliveryStorePort.AttemptFinalization(
outcome, terminalState(outcome, parkResult), fallbackEligible(outcome), parkResult);
NotificationDeliveryStorePort.FinalizationResult result =
Objects.requireNonNull(
store.finalizeAttempt(attempt, finalization, clock.instant()),
"notification attempt finalization result must be non-null");
return new FinalizationExecution(finalization, result);
}
private static NotificationDeliveryStorePort.TerminalState terminalState(
ProviderAttemptOutcome outcome, NotificationAdmissionReadinessPort.ParkResult parkResult) {
if (outcome.submissionCertainty() == SubmissionCertainty.PROVIDER_ACCEPTED) {
return NotificationDeliveryStorePort.TerminalState.ACCEPTED;
}
if (outcome.submissionCertainty() == SubmissionCertainty.INDETERMINATE) {
return NotificationDeliveryStorePort.TerminalState.TERMINAL_INDETERMINATE;
}
return switch (outcome.retryDisposition()) {
case RETRY_AT -> NotificationDeliveryStorePort.TerminalState.RETRY_SCHEDULED;
case PARK_BINDING ->
switch (parkResult) {
case PARKED, ALREADY_PARKED ->
NotificationDeliveryStorePort.TerminalState.PARKED_BINDING;
case STALE_GENERATION -> NotificationDeliveryStorePort.TerminalState.RETRY_SCHEDULED;
case NOT_REQUESTED ->
throw new IllegalStateException(
"PARK_BINDING outcome requires an admission park result");
};
case TERMINAL -> NotificationDeliveryStorePort.TerminalState.TERMINAL_FAILURE;
case NOT_APPLICABLE ->
throw new IllegalArgumentException(
"definitely-not-applied outcome requires an explicit disposition");
};
}
private static boolean fallbackEligible(ProviderAttemptOutcome outcome) {
return outcome.submissionCertainty() == SubmissionCertainty.DEFINITELY_NOT_APPLIED
&& outcome.retryDisposition() == RetryDisposition.TERMINAL;
}
private record FinalizationExecution(
NotificationDeliveryStorePort.AttemptFinalization finalization,
NotificationDeliveryStorePort.FinalizationResult result) {}
private static final class MutableCounts {
private final int claimed;
private int authorized;
private int providerCalls;
private int finalized;
private int staleClaims;
private int indeterminate;
private int parked;
private MutableCounts(int claimed) {
this.claimed = claimed;
}
private NotificationDispatchResult toResult() {
return new NotificationDispatchResult(
claimed, authorized, providerCalls, finalized, staleClaims, indeterminate, parked);
}
}
}
@@ -0,0 +1,24 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/** Historical issuer-key decision retained with accepted signed evidence. */
public record NotificationEvidenceTrustSnapshot(
String catalogRevision, HistoricalKeyStatus historicalKeyStatus, String issuerKeyDigest) {
public NotificationEvidenceTrustSnapshot {
catalogRevision =
NotificationIntentId.requireSlug("evidence trust catalog revision", catalogRevision);
Objects.requireNonNull(historicalKeyStatus, "historical evidence key status must be non-null");
issuerKeyDigest = InitializeNotificationWriterFencesCommand.requireDigest(issuerKeyDigest);
if (historicalKeyStatus == HistoricalKeyStatus.REVOKED) {
throw new IllegalArgumentException("revoked evidence issuer key cannot be accepted");
}
}
public enum HistoricalKeyStatus {
ALLOWED,
RETIRING,
REVOKED
}
}
@@ -0,0 +1,9 @@
package dev.caskeleton.application.notification;
/** Smallest durable scope affected by a classified attempt failure. */
public enum NotificationFaultScope {
DELIVERY,
ROUTE_REVISION,
PROVIDER_BINDING,
ACCOUNT
}
@@ -0,0 +1,216 @@
package dev.caskeleton.application.notification;
import java.time.Duration;
import java.time.Instant;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Objects;
import java.util.Optional;
/** Immutable provider-neutral plan snapshot returned by the application planning boundary. */
public record NotificationFrozenPlan(
NotificationIntentId intentId,
NotificationKindPolicy policy,
Locale selectedLocale,
BindingSnapshot binding,
NotificationRecipientReference recipient,
NotificationTemplateParameters parameters,
String idempotencyScope,
String sourceOperationId,
Optional<String> tenantReference,
String correlationReference,
Optional<String> causationReference,
Instant notBefore,
Instant expiresAt) {
public NotificationFrozenPlan {
Objects.requireNonNull(intentId, "notification intent ID must be non-null");
Objects.requireNonNull(policy, "notification kind policy must be non-null");
selectedLocale = NotificationIntentDraft.requireLocale("selected locale", selectedLocale);
Objects.requireNonNull(binding, "notification binding snapshot must be non-null");
Objects.requireNonNull(recipient, "notification recipient must be non-null");
Objects.requireNonNull(parameters, "notification template parameters must be non-null");
idempotencyScope =
NotificationIntentId.requireOpaque("notification idempotency scope", idempotencyScope);
sourceOperationId =
NotificationIntentId.requireOpaque("notification source operation ID", sourceOperationId);
tenantReference = requireOptionalOpaque("tenant reference", tenantReference);
correlationReference =
NotificationIntentId.requireOpaque(
"notification correlation reference", correlationReference);
causationReference = requireOptionalOpaque("causation reference", causationReference);
Objects.requireNonNull(notBefore, "notification not-before time must be non-null");
Objects.requireNonNull(expiresAt, "notification expiry time must be non-null");
if (recipient.channel() != policy.channel()) {
throw new IllegalArgumentException("recipient channel must match notification kind channel");
}
if (binding.targets().size() != policy.maxTargetsPerRecipient()) {
throw new IllegalArgumentException(
"frozen binding target count must match the code-owned policy target bound");
}
Duration lifetime = Duration.between(notBefore, expiresAt);
if (lifetime.isZero()
|| lifetime.isNegative()
|| lifetime.compareTo(policy.maxElapsedRetryHorizon()) > 0) {
throw new IllegalArgumentException(
"notification expiry must be after not-before and within the policy retry horizon");
}
}
public static NotificationFrozenPlan from(
NotificationIntentDraft draft, Locale selectedLocale, BindingSnapshot binding) {
Objects.requireNonNull(draft, "notification intent draft must be non-null");
return new NotificationFrozenPlan(
draft.intentId(),
draft.policy(),
selectedLocale,
binding,
draft.recipient(),
draft.parameters(),
draft.idempotencyScope(),
draft.sourceOperationId(),
draft.tenantReference(),
draft.correlationReference(),
draft.causationReference(),
draft.notBefore(),
draft.expiresAt());
}
public NotificationMode mode() {
return policy.mode();
}
public NotificationRouteId routeId() {
return policy.routeId();
}
private static Optional<String> requireOptionalOpaque(String field, Optional<String> reference) {
Objects.requireNonNull(reference, field + " container must be non-null");
return reference.map(value -> NotificationIntentId.requireOpaque(field, value));
}
@Override
public String toString() {
return "NotificationFrozenPlan[intentId="
+ intentId
+ ", kindId="
+ policy.kindId()
+ ", policyRevision="
+ policy.policyRevision()
+ ", selectedLocale="
+ selectedLocale.toLanguageTag()
+ ", routeRevision="
+ binding.routeRevision()
+ ", bindingDigest="
+ binding.bindingDigest()
+ ", rendererRevision="
+ binding.rendererRevision()
+ ", targets=<redacted>"
+ ", recipient=<redacted>, parameters=<redacted>, context=<redacted>, notBefore="
+ notBefore
+ ", expiresAt="
+ expiresAt
+ "]";
}
/** Provider-neutral immutable execution graph persisted with the logical intent. */
public record BindingSnapshot(
int routeRevision,
String bindingDigest,
String templateChecksum,
String rendererRevision,
List<FrozenTarget> targets,
boolean receiptRequired,
Duration perAttemptDeadline) {
private static final Duration MAXIMUM_ATTEMPT_DEADLINE = Duration.ofMinutes(5);
public BindingSnapshot {
if (routeRevision < 1 || routeRevision > 1_000_000) {
throw new IllegalArgumentException("frozen route revision must be in 1..1000000");
}
bindingDigest = requireDigest("notification binding digest", bindingDigest);
templateChecksum = requireDigest("notification template checksum", templateChecksum);
rendererRevision =
NotificationIntentId.requireSlug("notification renderer revision", rendererRevision);
Objects.requireNonNull(targets, "frozen notification targets must be non-null");
targets =
targets.stream()
.map(target -> Objects.requireNonNull(target, "frozen target must be non-null"))
.sorted(Comparator.comparingInt(FrozenTarget::ordinal))
.toList();
if (targets.isEmpty() || targets.size() > 16) {
throw new IllegalArgumentException(
"frozen notification targets must contain 1..16 entries");
}
if (new HashSet<>(targets.stream().map(FrozenTarget::targetReference).toList()).size()
!= targets.size()) {
throw new IllegalArgumentException(
"frozen notification targets contain duplicate references");
}
for (int index = 0; index < targets.size(); index++) {
if (targets.get(index).ordinal() != index) {
throw new IllegalArgumentException(
"frozen notification target ordinals must be contiguous from zero");
}
}
Objects.requireNonNull(
perAttemptDeadline, "notification per-attempt deadline must be non-null");
if (perAttemptDeadline.isZero()
|| perAttemptDeadline.isNegative()
|| perAttemptDeadline.compareTo(MAXIMUM_ATTEMPT_DEADLINE) > 0) {
throw new IllegalArgumentException(
"notification per-attempt deadline must be positive and at most five minutes");
}
}
private static String requireDigest(String field, String digest) {
if (digest == null || !digest.matches("[0-9a-f]{64}")) {
throw new IllegalArgumentException(field + " must be a lowercase SHA-256 digest");
}
return digest;
}
}
/** Opaque provider-leg identity; credentials, endpoints and SDK types are deliberately absent. */
public record FrozenTarget(
int ordinal,
String targetReference,
String providerCapabilityReference,
String providerBindingRevision,
String credentialGeneration) {
public FrozenTarget {
if (ordinal < 0 || ordinal > 15) {
throw new IllegalArgumentException("frozen notification target ordinal must be in 0..15");
}
targetReference =
NotificationIntentId.requireOpaque(
"frozen notification target reference", targetReference);
providerCapabilityReference =
NotificationIntentId.requireOpaque(
"frozen provider capability reference", providerCapabilityReference);
providerBindingRevision =
NotificationIntentId.requireOpaque(
"frozen provider binding revision", providerBindingRevision);
credentialGeneration =
NotificationIntentId.requireSlug(
"frozen provider credential generation", credentialGeneration);
}
@Override
public String toString() {
return "FrozenTarget[ordinal="
+ ordinal
+ ", targetReference=<redacted>, providerCapabilityReference="
+ providerCapabilityReference
+ ", providerBindingRevision="
+ providerBindingRevision
+ ", credentialGeneration="
+ credentialGeneration
+ "]";
}
}
}
@@ -0,0 +1,11 @@
package dev.caskeleton.application.notification;
/**
* Appends a frozen intent to durable storage in the caller's current transaction. Implementations
* must not open an independent transaction.
*/
@FunctionalInterface
public interface NotificationIntentAppendPort {
NotificationAppendResult append(NotificationFrozenPlan plan);
}
@@ -0,0 +1,84 @@
package dev.caskeleton.application.notification;
import java.time.Duration;
import java.time.Instant;
import java.util.Locale;
import java.util.Objects;
import java.util.Optional;
/** Feature-policy output containing one logical recipient and no provider or transport types. */
public record NotificationIntentDraft(
NotificationIntentId intentId,
NotificationKindPolicy policy,
Locale requestedLocale,
NotificationRecipientReference recipient,
NotificationTemplateParameters parameters,
String idempotencyScope,
String sourceOperationId,
Optional<String> tenantReference,
String correlationReference,
Optional<String> causationReference,
Instant notBefore,
Instant expiresAt) {
public NotificationIntentDraft {
Objects.requireNonNull(intentId, "notification intent ID must be non-null");
Objects.requireNonNull(policy, "notification kind policy must be non-null");
requestedLocale = requireLocale("requested locale", requestedLocale);
Objects.requireNonNull(recipient, "notification recipient must be non-null");
Objects.requireNonNull(parameters, "notification template parameters must be non-null");
if (recipient.channel() != policy.channel()) {
throw new IllegalArgumentException("recipient channel must match notification kind channel");
}
idempotencyScope =
NotificationIntentId.requireOpaque("notification idempotency scope", idempotencyScope);
sourceOperationId =
NotificationIntentId.requireOpaque("notification source operation ID", sourceOperationId);
tenantReference = requireOptionalOpaque("tenant reference", tenantReference);
correlationReference =
NotificationIntentId.requireOpaque(
"notification correlation reference", correlationReference);
causationReference = requireOptionalOpaque("causation reference", causationReference);
Objects.requireNonNull(notBefore, "notification not-before time must be non-null");
Objects.requireNonNull(expiresAt, "notification expiry time must be non-null");
Duration lifetime = Duration.between(notBefore, expiresAt);
if (lifetime.isZero()
|| lifetime.isNegative()
|| lifetime.compareTo(policy.maxElapsedRetryHorizon()) > 0) {
throw new IllegalArgumentException(
"notification expiry must be after not-before and within the policy retry horizon");
}
}
static Locale requireLocale(String field, Locale locale) {
Objects.requireNonNull(locale, field + " must be non-null");
String languageTag = locale.toLanguageTag();
if (locale.equals(Locale.ROOT)
|| languageTag.equals("und")
|| languageTag.isBlank()
|| languageTag.length() > 35) {
throw new IllegalArgumentException(field + " must be an explicit bounded locale");
}
return Locale.forLanguageTag(languageTag);
}
private static Optional<String> requireOptionalOpaque(String field, Optional<String> reference) {
Objects.requireNonNull(reference, field + " container must be non-null");
return reference.map(value -> NotificationIntentId.requireOpaque(field, value));
}
@Override
public String toString() {
return "NotificationIntentDraft[intentId="
+ intentId
+ ", kindId="
+ policy.kindId()
+ ", requestedLocale="
+ requestedLocale.toLanguageTag()
+ ", recipient=<redacted>, parameters=<redacted>, context=<redacted>, notBefore="
+ notBefore
+ ", expiresAt="
+ expiresAt
+ "]";
}
}
@@ -0,0 +1,24 @@
package dev.caskeleton.application.notification;
/** Opaque identity of one logical business notification. */
public record NotificationIntentId(String value) {
public NotificationIntentId {
value = requireOpaque("intentId", value);
}
static String requireOpaque(String field, String value) {
if (value == null || !value.matches("[A-Za-z0-9][A-Za-z0-9._:-]{0,127}")) {
throw new IllegalArgumentException(
field + " must contain 1..128 opaque identifier characters");
}
return value;
}
static String requireSlug(String field, String value) {
if (value == null || !value.matches("[a-z][a-z0-9.-]{0,62}")) {
throw new IllegalArgumentException(field + " must match [a-z][a-z0-9.-]{0,62}");
}
return value;
}
}
@@ -0,0 +1,9 @@
package dev.caskeleton.application.notification;
/** Closed-catalog notification business kind. */
public record NotificationKindId(String value) {
public NotificationKindId {
value = NotificationIntentId.requireSlug("kindId", value);
}
}
@@ -0,0 +1,124 @@
package dev.caskeleton.application.notification;
import java.time.Duration;
import java.util.Objects;
/**
* Code-owned notification kind catalog row. Runtime configuration may assert these values and lower
* resource concurrency, but cannot replace semantic fields in this record.
*/
public record NotificationKindPolicy(
NotificationKindId kindId,
NotificationChannel channel,
NotificationRouteId routeId,
NotificationTemplateRef templateRef,
NotificationMode mode,
NotificationAdmissionClass admissionClass,
NotificationRouteStrategy routeStrategy,
ConsentCheckMode consentCheckMode,
int policyRevision,
int maxTargetsPerRecipient,
int maxPhysicalAttemptsPerDelivery,
int maxFallbackActivations,
int maxReconcileCalls,
int maxTotalProviderCallsPerIntent,
Duration maxElapsedRetryHorizon) {
private static final int MAXIMUM_TARGETS = 16;
private static final int MAXIMUM_ATTEMPTS_PER_DELIVERY = 10;
private static final int MAXIMUM_PROVIDER_CALLS = 64;
private static final Duration MAXIMUM_RETRY_HORIZON = Duration.ofDays(30);
public NotificationKindPolicy {
Objects.requireNonNull(kindId, "notification kind must be non-null");
Objects.requireNonNull(channel, "notification channel must be non-null");
Objects.requireNonNull(routeId, "notification route must be non-null");
Objects.requireNonNull(templateRef, "notification template must be non-null");
Objects.requireNonNull(mode, "notification mode must be non-null");
Objects.requireNonNull(admissionClass, "notification admission class must be non-null");
Objects.requireNonNull(routeStrategy, "notification route strategy must be non-null");
Objects.requireNonNull(consentCheckMode, "consent check mode must be non-null");
Objects.requireNonNull(maxElapsedRetryHorizon, "maximum retry horizon must be non-null");
if (policyRevision < 1 || policyRevision > 1_000_000) {
throw new IllegalArgumentException("policy revision must be in 1..1000000");
}
if (maxTargetsPerRecipient < 1 || maxTargetsPerRecipient > MAXIMUM_TARGETS) {
throw new IllegalArgumentException("maximum targets per recipient must be in 1..16");
}
if (maxPhysicalAttemptsPerDelivery < 1
|| maxPhysicalAttemptsPerDelivery > MAXIMUM_ATTEMPTS_PER_DELIVERY) {
throw new IllegalArgumentException("maximum physical attempts per delivery must be in 1..10");
}
if (maxFallbackActivations < 0 || maxFallbackActivations >= MAXIMUM_TARGETS) {
throw new IllegalArgumentException("maximum fallback activations must be in 0..15");
}
if (maxReconcileCalls < 0 || maxReconcileCalls > 10) {
throw new IllegalArgumentException("maximum reconcile calls must be in 0..10");
}
if (maxTotalProviderCallsPerIntent < 1
|| maxTotalProviderCallsPerIntent > MAXIMUM_PROVIDER_CALLS) {
throw new IllegalArgumentException("maximum total provider calls must be in 1..64");
}
if (maxElapsedRetryHorizon.isZero()
|| maxElapsedRetryHorizon.isNegative()
|| maxElapsedRetryHorizon.compareTo(MAXIMUM_RETRY_HORIZON) > 0) {
throw new IllegalArgumentException(
"maximum retry horizon must be positive and at most 30 days");
}
if (admissionClass == NotificationAdmissionClass.SECURITY_CRITICAL
&& mode == NotificationMode.BEST_EFFORT_INLINE) {
throw new IllegalArgumentException(
"SECURITY_CRITICAL notification kind cannot use BEST_EFFORT_INLINE");
}
validateStrategy(routeStrategy, maxTargetsPerRecipient, maxFallbackActivations);
if (mode == NotificationMode.BEST_EFFORT_INLINE
&& (maxPhysicalAttemptsPerDelivery != 1 || maxReconcileCalls != 0)) {
throw new IllegalArgumentException(
"BEST_EFFORT_INLINE permits one attempt per target and no reconciliation");
}
long worstCaseProviderCalls =
Math.addExact(
Math.multiplyExact(
(long) maxTargetsPerRecipient, (long) maxPhysicalAttemptsPerDelivery),
maxReconcileCalls);
if (worstCaseProviderCalls > maxTotalProviderCallsPerIntent) {
throw new IllegalArgumentException(
"worst-case provider calls exceed maximum total provider calls per intent");
}
}
public NotificationKindPolicy assertRuntimeExpectation(
NotificationMode expectedMode, NotificationAdmissionClass expectedAdmissionClass) {
Objects.requireNonNull(expectedMode, "expected notification mode must be non-null");
Objects.requireNonNull(
expectedAdmissionClass, "expected notification admission class must be non-null");
if (mode != expectedMode) {
throw new IllegalStateException(
"runtime expected mode " + expectedMode + " does not match code-owned mode " + mode);
}
if (admissionClass != expectedAdmissionClass) {
throw new IllegalStateException(
"runtime expected admission "
+ expectedAdmissionClass
+ " does not match code-owned admission "
+ admissionClass);
}
return this;
}
private static void validateStrategy(
NotificationRouteStrategy strategy, int maximumTargets, int maximumFallbacks) {
if (strategy == NotificationRouteStrategy.SINGLE
&& (maximumTargets != 1 || maximumFallbacks != 0)) {
throw new IllegalArgumentException("SINGLE requires one target and zero fallbacks");
}
if (strategy == NotificationRouteStrategy.FAN_OUT_ALL && maximumFallbacks != 0) {
throw new IllegalArgumentException("FAN_OUT_ALL cannot activate fallback targets");
}
if (strategy == NotificationRouteStrategy.ORDERED_FALLBACK
&& (maximumTargets < 2 || maximumFallbacks < 1 || maximumFallbacks > maximumTargets - 1)) {
throw new IllegalArgumentException(
"ORDERED_FALLBACK requires 2..16 targets and 1..targetCount-1 fallbacks");
}
}
}
@@ -0,0 +1,103 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.command.Command;
import java.time.Duration;
import java.util.Objects;
import java.util.Optional;
/** Root-committed acquire or release of one bounded legacy writer permit. */
public record NotificationLegacyWriterPermitCommand(
Action action,
NotificationCanonicalWriterRouteSet.RouteRevision route,
long expectedGeneration,
String transportProfileId,
String permitToken,
String holderReference,
String operationToken,
String actorReference,
NotificationReasonCode reasonCode,
Optional<Duration> wireBudget)
implements Command {
private static final Duration MAXIMUM_WIRE_BUDGET = Duration.ofSeconds(30);
public NotificationLegacyWriterPermitCommand {
Objects.requireNonNull(action, "legacy writer permit action must be non-null");
Objects.requireNonNull(route, "notification writer route must be non-null");
if (expectedGeneration < 0) {
throw new IllegalArgumentException("expected writer generation must be non-negative");
}
transportProfileId =
NotificationIntentId.requireSlug("legacy transport profile ID", transportProfileId);
permitToken = NotificationIntentId.requireOpaque("legacy writer permit token", permitToken);
holderReference =
NotificationIntentId.requireOpaque("legacy writer permit holder", holderReference);
operationToken =
NotificationIntentId.requireOpaque("legacy writer permit operation token", operationToken);
actorReference =
NotificationIntentId.requireOpaque("legacy writer permit actor", actorReference);
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
Objects.requireNonNull(wireBudget, "wire budget container must be non-null");
if (action == Action.ACQUIRE) {
Duration budget =
wireBudget.orElseThrow(
() -> new IllegalArgumentException("legacy permit acquire requires a wire budget"));
if (budget.isZero() || budget.isNegative() || budget.compareTo(MAXIMUM_WIRE_BUDGET) > 0) {
throw new IllegalArgumentException(
"legacy writer wire budget must be positive and at most 30 seconds");
}
} else if (wireBudget.isPresent()) {
throw new IllegalArgumentException("legacy permit release must not carry a wire budget");
}
}
public static NotificationLegacyWriterPermitCommand acquire(
NotificationCanonicalWriterRouteSet.RouteRevision route,
long expectedGeneration,
String transportProfileId,
String permitToken,
String holderReference,
String operationToken,
String actorReference,
NotificationReasonCode reasonCode,
Duration wireBudget) {
return new NotificationLegacyWriterPermitCommand(
Action.ACQUIRE,
route,
expectedGeneration,
transportProfileId,
permitToken,
holderReference,
operationToken,
actorReference,
reasonCode,
Optional.of(wireBudget));
}
public static NotificationLegacyWriterPermitCommand release(
NotificationCanonicalWriterRouteSet.RouteRevision route,
long expectedGeneration,
String transportProfileId,
String permitToken,
String holderReference,
String operationToken,
String actorReference,
NotificationReasonCode reasonCode) {
return new NotificationLegacyWriterPermitCommand(
Action.RELEASE,
route,
expectedGeneration,
transportProfileId,
permitToken,
holderReference,
operationToken,
actorReference,
reasonCode,
Optional.empty());
}
public enum Action {
ACQUIRE,
RELEASE
}
}
@@ -0,0 +1,45 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
import java.util.Objects;
import java.util.Optional;
/** Physically committed permit facts used by the legacy wrapper's monotonic wire deadline guard. */
public record NotificationLegacyWriterPermitResult(
Status status,
String permitToken,
Optional<Instant> acquiredAt,
Optional<Instant> wireDeadline,
Optional<Instant> expiresAt) {
public NotificationLegacyWriterPermitResult {
Objects.requireNonNull(status, "legacy writer permit status must be non-null");
permitToken = NotificationIntentId.requireOpaque("legacy writer permit token", permitToken);
Objects.requireNonNull(acquiredAt, "permit acquired-at container must be non-null");
Objects.requireNonNull(wireDeadline, "permit wire-deadline container must be non-null");
Objects.requireNonNull(expiresAt, "permit expiry container must be non-null");
if (status == Status.ACQUIRED) {
Instant acquired =
acquiredAt.orElseThrow(
() -> new IllegalArgumentException("acquired permit requires DB acquired-at"));
Instant deadline =
wireDeadline.orElseThrow(
() -> new IllegalArgumentException("acquired permit requires wire deadline"));
Instant expiry =
expiresAt.orElseThrow(
() -> new IllegalArgumentException("acquired permit requires expiry"));
if (deadline.isBefore(acquired) || expiry.isBefore(deadline)) {
throw new IllegalArgumentException(
"permit timestamps must satisfy acquiredAt <= wireDeadline <= expiresAt");
}
} else if (acquiredAt.isPresent() || wireDeadline.isPresent() || expiresAt.isPresent()) {
throw new IllegalArgumentException("non-acquired permit result must not expose wire times");
}
}
public enum Status {
ACQUIRED,
RELEASED,
REPLAYED
}
}
@@ -0,0 +1,59 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.capability.Idempotency;
import dev.caskeleton.application.capability.RepositoryAccess;
import dev.caskeleton.application.capability.UseCaseCapability;
import dev.caskeleton.application.security.RequiresPermission;
import dev.caskeleton.application.transaction.TransactionMode;
import dev.caskeleton.application.transaction.TransactionPort;
import dev.caskeleton.application.usecase.CommandUseCase;
import java.time.Clock;
import java.util.Objects;
/** Root-commits bounded legacy permit acquire/release before any caller provider I/O. */
@RequiresPermission("notification:cutover-admit")
@UseCaseCapability(
transactionMode = TransactionMode.WRITE,
idempotency = Idempotency.IDEMPOTENT,
repositoryAccess = RepositoryAccess.WRITE_REPOSITORY,
crossTenantAdmin = true)
public final class NotificationLegacyWriterPermitUseCase
implements CommandUseCase<
NotificationLegacyWriterPermitCommand, NotificationLegacyWriterPermitResult> {
private final NotificationWriterRouteSet routes;
private final NotificationWriterCutoverPort cutover;
private final TransactionPort transactions;
private final Clock clock;
public NotificationLegacyWriterPermitUseCase(
NotificationWriterRouteSet routes,
NotificationWriterCutoverPort cutover,
TransactionPort transactions,
Clock clock) {
this.routes = Objects.requireNonNull(routes, "notification writer route set must be non-null");
this.cutover =
Objects.requireNonNull(cutover, "notification writer cutover port must be non-null");
this.transactions = Objects.requireNonNull(transactions, "transaction port must be non-null");
this.clock = Objects.requireNonNull(clock, "clock must be non-null");
}
@Override
public NotificationLegacyWriterPermitResult handle(
NotificationLegacyWriterPermitCommand command) {
Objects.requireNonNull(command, "legacy writer permit command must be non-null");
NotificationWriterRouteSet.RouteProfile route = routes.requireRoute(command.route());
NotificationWriterRouteSet.TransportProfile profile =
route.requireProfile(command.transportProfileId());
if (command.action() == NotificationLegacyWriterPermitCommand.Action.ACQUIRE
&& !profile.activeAdmissionProfile()) {
throw new IllegalArgumentException(
"new legacy permit acquire requires the active admission transport profile");
}
return transactions.inRootWrite(
() ->
command.action() == NotificationLegacyWriterPermitCommand.Action.ACQUIRE
? cutover.acquireLegacyPermit(command, route, clock.instant())
: cutover.releaseLegacyPermit(command, route, clock.instant()));
}
}
@@ -0,0 +1,21 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.command.Command;
/** Bounded retention and redaction mutation request. */
public record NotificationMaintenanceCommand(
int maximumExpiredIntents, int maximumPayloadRedactions, int maximumExpiredReceipts)
implements Command {
public NotificationMaintenanceCommand {
long total = (long) maximumExpiredIntents + maximumPayloadRedactions + maximumExpiredReceipts;
if (maximumExpiredIntents < 0
|| maximumPayloadRedactions < 0
|| maximumExpiredReceipts < 0
|| total < 1
|| total > 100) {
throw new IllegalArgumentException(
"notification maintenance total mutation bound must be in 1..100");
}
}
}
@@ -0,0 +1,16 @@
package dev.caskeleton.application.notification;
/** Bounded non-sensitive maintenance outcome. */
public record NotificationMaintenanceResult(
int expiredIntentCount, int redactedPayloadCount, int expiredReceiptCount) {
public NotificationMaintenanceResult {
long total = (long) expiredIntentCount + redactedPayloadCount + expiredReceiptCount;
if (expiredIntentCount < 0
|| redactedPayloadCount < 0
|| expiredReceiptCount < 0
|| total > 100) {
throw new IllegalArgumentException("notification maintenance result must total 0..100");
}
}
}
@@ -0,0 +1,23 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
/** Performs one bounded local retention/redaction mutation batch. */
@FunctionalInterface
public interface NotificationMaintenanceStorePort {
MutationResult maintain(NotificationMaintenanceCommand command, Instant now);
record MutationResult(int expiredIntentCount, int redactedPayloadCount, int expiredReceiptCount) {
public MutationResult {
long total = (long) expiredIntentCount + redactedPayloadCount + expiredReceiptCount;
if (expiredIntentCount < 0
|| redactedPayloadCount < 0
|| expiredReceiptCount < 0
|| total > 100) {
throw new IllegalArgumentException("notification maintenance mutations must total 0..100");
}
}
}
}
@@ -0,0 +1,44 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.capability.Idempotency;
import dev.caskeleton.application.capability.RepositoryAccess;
import dev.caskeleton.application.capability.UseCaseCapability;
import dev.caskeleton.application.security.RequiresPermission;
import dev.caskeleton.application.transaction.TransactionMode;
import dev.caskeleton.application.transaction.TransactionPort;
import dev.caskeleton.application.usecase.CommandUseCase;
import java.time.Clock;
import java.util.Objects;
/** Runs one bounded notification retention/redaction mutation in a short write transaction. */
@RequiresPermission("notification:maintain")
@UseCaseCapability(
transactionMode = TransactionMode.WRITE,
idempotency = Idempotency.IDEMPOTENT,
repositoryAccess = RepositoryAccess.WRITE_REPOSITORY,
crossTenantAdmin = true)
public final class NotificationMaintenanceUseCase
implements CommandUseCase<NotificationMaintenanceCommand, NotificationMaintenanceResult> {
private final NotificationMaintenanceStorePort store;
private final TransactionPort transactions;
private final Clock clock;
public NotificationMaintenanceUseCase(
NotificationMaintenanceStorePort store, TransactionPort transactions, Clock clock) {
this.store = Objects.requireNonNull(store, "notification maintenance store must be non-null");
this.transactions = Objects.requireNonNull(transactions, "transaction port must be non-null");
this.clock = Objects.requireNonNull(clock, "clock must be non-null");
}
@Override
public NotificationMaintenanceResult handle(NotificationMaintenanceCommand command) {
Objects.requireNonNull(command, "notification maintenance command must be non-null");
NotificationMaintenanceStorePort.MutationResult mutation =
transactions.inWrite(() -> store.maintain(command, clock.instant()));
return new NotificationMaintenanceResult(
mutation.expiredIntentCount(),
mutation.redactedPayloadCount(),
mutation.expiredReceiptCount());
}
}
@@ -0,0 +1,7 @@
package dev.caskeleton.application.notification;
/** Code-owned delivery durability contract. */
public enum NotificationMode {
BEST_EFFORT_INLINE,
DURABLE_ASYNC
}
@@ -0,0 +1,57 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
import java.util.List;
import java.util.Objects;
/** Bounded non-sensitive operational aggregate returned through the application query boundary. */
public record NotificationOperationsSnapshot(
Instant observedAt,
long pendingIntentCount,
long parkedDeliveryCount,
long orphanReceiptCount,
long activeLegacyPermitCount,
List<RouteWriterStatus> writerRoutes) {
private static final long MAXIMUM_COUNT = 1_000_000_000L;
public NotificationOperationsSnapshot {
Objects.requireNonNull(observedAt, "notification snapshot time must be non-null");
validateCount(pendingIntentCount);
validateCount(parkedDeliveryCount);
validateCount(orphanReceiptCount);
validateCount(activeLegacyPermitCount);
writerRoutes =
List.copyOf(
Objects.requireNonNull(writerRoutes, "writer route snapshots must be non-null"));
if (writerRoutes.size() > 100) {
throw new IllegalArgumentException("writer route snapshot exceeds 100 entries");
}
}
private static void validateCount(long count) {
if (count < 0 || count > MAXIMUM_COUNT) {
throw new IllegalArgumentException("notification operation count is outside 0..1000000000");
}
}
public record RouteWriterStatus(
NotificationRouteId routeId,
int routeRevision,
NotificationWriterOwnership owner,
long generation,
boolean draining) {
public RouteWriterStatus {
Objects.requireNonNull(routeId, "notification route ID must be non-null");
Objects.requireNonNull(owner, "notification writer owner must be non-null");
if (routeRevision < 1 || generation < 0) {
throw new IllegalArgumentException(
"writer route revision must be positive and generation non-negative");
}
if (draining && owner != NotificationWriterOwnership.LEGACY) {
throw new IllegalArgumentException("only LEGACY writer ownership may be draining");
}
}
}
}
@@ -0,0 +1,8 @@
package dev.caskeleton.application.notification;
/** Loads bounded non-sensitive operational aggregates from the durable store. */
@FunctionalInterface
public interface NotificationOperationsSnapshotPort {
NotificationOperationsSnapshot load(NotificationOperationsSnapshotQuery query);
}
@@ -0,0 +1,13 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.query.Query;
/** Requests at most a bounded number of route writer status rows. */
public record NotificationOperationsSnapshotQuery(int maximumRoutes) implements Query {
public NotificationOperationsSnapshotQuery {
if (maximumRoutes < 1 || maximumRoutes > 100) {
throw new IllegalArgumentException("maximum notification routes must be in 1..100");
}
}
}
@@ -0,0 +1,37 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.capability.Idempotency;
import dev.caskeleton.application.capability.RepositoryAccess;
import dev.caskeleton.application.capability.UseCaseCapability;
import dev.caskeleton.application.security.RequiresPermission;
import dev.caskeleton.application.transaction.TransactionMode;
import dev.caskeleton.application.transaction.TransactionPort;
import dev.caskeleton.application.usecase.QueryUseCase;
import java.util.Objects;
/** Application query boundary for bounded notification operations visibility. */
@RequiresPermission("notification:observe")
@UseCaseCapability(
transactionMode = TransactionMode.READ_ONLY,
idempotency = Idempotency.IDEMPOTENT,
repositoryAccess = RepositoryAccess.READ_REPOSITORY,
crossTenantAdmin = true)
public final class NotificationOperationsSnapshotUseCase
implements QueryUseCase<NotificationOperationsSnapshotQuery, NotificationOperationsSnapshot> {
private final NotificationOperationsSnapshotPort snapshots;
private final TransactionPort transactions;
public NotificationOperationsSnapshotUseCase(
NotificationOperationsSnapshotPort snapshots, TransactionPort transactions) {
this.snapshots =
Objects.requireNonNull(snapshots, "notification operations snapshot port must be non-null");
this.transactions = Objects.requireNonNull(transactions, "transaction port must be non-null");
}
@Override
public NotificationOperationsSnapshot handle(NotificationOperationsSnapshotQuery query) {
Objects.requireNonNull(query, "notification operations snapshot query must be non-null");
return transactions.inRead(() -> snapshots.load(query));
}
}
@@ -0,0 +1,8 @@
package dev.caskeleton.application.notification;
/** Resolves a feature-owned draft into a provider-neutral immutable application plan. */
@FunctionalInterface
public interface NotificationPlanPort {
NotificationPlanningResult plan(NotificationIntentDraft draft);
}
@@ -0,0 +1,32 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/** Closed result of compiling a feature draft without exposing adapter binding types. */
public sealed interface NotificationPlanningResult
permits NotificationPlanningResult.Planned,
NotificationPlanningResult.Rejected,
NotificationPlanningResult.CapabilityUnavailable {
record Planned(NotificationFrozenPlan plan) implements NotificationPlanningResult {
public Planned {
Objects.requireNonNull(plan, "notification frozen plan must be non-null");
}
}
record Rejected(NotificationReasonCode reasonCode) implements NotificationPlanningResult {
public Rejected {
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
}
record CapabilityUnavailable(NotificationReasonCode reasonCode)
implements NotificationPlanningResult {
public CapabilityUnavailable {
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
}
}
@@ -0,0 +1,8 @@
package dev.caskeleton.application.notification;
/** Performs one previously authorized provider call and returns only a classified outcome. */
@FunctionalInterface
public interface NotificationProviderAttemptPort {
ProviderAttemptOutcome attempt(NotificationDeliveryStorePort.AuthorizedAttempt attempt);
}
@@ -0,0 +1,41 @@
package dev.caskeleton.application.notification;
import java.util.Collections;
import java.util.EnumSet;
import java.util.Objects;
import java.util.Set;
/** Provider-neutral capability facts consumed by the pure compatibility validator. */
public record NotificationProviderCapabilityDescriptor(
String capabilityReference,
NotificationChannel channel,
Set<NotificationMode> supportedModes,
boolean receiptSupported,
boolean reconciliationSupported,
boolean hiddenRetriesControlled,
int maximumTargets,
int maximumPayloadBytes) {
public NotificationProviderCapabilityDescriptor {
capabilityReference =
NotificationIntentId.requireOpaque(
"notification provider capability reference", capabilityReference);
Objects.requireNonNull(channel, "notification provider channel must be non-null");
Objects.requireNonNull(supportedModes, "notification provider modes must be non-null");
EnumSet<NotificationMode> modes =
supportedModes.isEmpty()
? EnumSet.noneOf(NotificationMode.class)
: EnumSet.copyOf(supportedModes);
if (modes.isEmpty()) {
throw new IllegalArgumentException("notification provider must support at least one mode");
}
supportedModes = Collections.unmodifiableSet(modes);
if (maximumTargets < 1 || maximumTargets > 16) {
throw new IllegalArgumentException("notification provider maximum targets must be in 1..16");
}
if (maximumPayloadBytes < 1 || maximumPayloadBytes > 10_000_000) {
throw new IllegalArgumentException(
"notification provider maximum payload must be in 1..10000000 bytes");
}
}
}
@@ -0,0 +1,12 @@
package dev.caskeleton.application.notification;
/** Bounded stable operational reason code; never a provider error body or SDK exception message. */
public record NotificationReasonCode(String value) {
public NotificationReasonCode {
if (value == null || !value.matches("[A-Z][A-Z0-9_]{0,63}")) {
throw new IllegalArgumentException(
"notification reason code must match [A-Z][A-Z0-9_]{0,63}");
}
}
}
@@ -0,0 +1,9 @@
package dev.caskeleton.application.notification;
/** Opaque identity used to deduplicate one normalized provider receipt event. */
public record NotificationReceiptEventId(String value) {
public NotificationReceiptEventId {
value = NotificationIntentId.requireOpaque("receiptEventId", value);
}
}
@@ -0,0 +1,34 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
import java.util.Objects;
/** One normalized, immutable provider feedback fact. */
public record NotificationReceiptFact(
Type type, BounceClass bounceClass, NotificationReasonCode reasonCode, Instant occurredAt) {
public NotificationReceiptFact {
Objects.requireNonNull(type, "notification receipt type must be non-null");
Objects.requireNonNull(bounceClass, "notification bounce class must be non-null");
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
Objects.requireNonNull(occurredAt, "notification receipt occurrence time must be non-null");
if ((type == Type.BOUNCE) != (bounceClass != BounceClass.NONE)) {
throw new IllegalArgumentException(
"BOUNCE requires HARD or SOFT classification and other facts require NONE");
}
}
public enum Type {
SEND,
DELIVERY,
BOUNCE,
COMPLAINT,
DELIVERY_DELAY
}
public enum BounceClass {
NONE,
SOFT,
HARD
}
}
@@ -0,0 +1,25 @@
package dev.caskeleton.application.notification;
import java.util.Collections;
import java.util.EnumSet;
import java.util.Objects;
import java.util.Set;
/** Authenticated normalized receipt-ingress capabilities for one channel. */
public record NotificationReceiptIngressCapabilityDescriptor(
NotificationChannel channel,
boolean enabled,
boolean authenticated,
Set<NotificationReceiptFact.Type> supportedFactTypes) {
public NotificationReceiptIngressCapabilityDescriptor {
Objects.requireNonNull(channel, "notification receipt ingress channel must be non-null");
Objects.requireNonNull(
supportedFactTypes, "notification receipt ingress fact types must be non-null");
EnumSet<NotificationReceiptFact.Type> factTypes =
supportedFactTypes.isEmpty()
? EnumSet.noneOf(NotificationReceiptFact.Type.class)
: EnumSet.copyOf(supportedFactTypes);
supportedFactTypes = Collections.unmodifiableSet(factTypes);
}
}
@@ -0,0 +1,70 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
import java.util.List;
import java.util.Objects;
/** Order-independent orthogonal projection derived only from immutable receipt facts. */
public record NotificationReceiptProjection(
boolean submissionAccepted,
boolean delivered,
boolean softBounced,
boolean hardBounced,
boolean complained,
boolean deliveryDelayed,
Instant latestFactAt,
int factCount) {
public NotificationReceiptProjection {
Objects.requireNonNull(latestFactAt, "latest receipt fact time must be non-null");
if (factCount < 1 || factCount > 100) {
throw new IllegalArgumentException("receipt projection fact count must be in 1..100");
}
if ((delivered || softBounced || hardBounced || complained || deliveryDelayed)
&& !submissionAccepted) {
throw new IllegalArgumentException(
"delivery feedback cannot erase or contradict provider acceptance");
}
}
public static NotificationReceiptProjection reduce(List<NotificationReceiptFact> facts) {
Objects.requireNonNull(facts, "notification receipt facts must be non-null");
List<NotificationReceiptFact> immutableFacts = List.copyOf(facts);
if (immutableFacts.isEmpty() || immutableFacts.size() > 100) {
throw new IllegalArgumentException("notification receipt facts must contain 1..100 entries");
}
boolean accepted = false;
boolean delivered = false;
boolean softBounced = false;
boolean hardBounced = false;
boolean complained = false;
boolean delayed = false;
Instant latest = Instant.MIN;
for (NotificationReceiptFact fact : immutableFacts) {
Objects.requireNonNull(fact, "notification receipt fact must be non-null");
accepted = true;
delivered |= fact.type() == NotificationReceiptFact.Type.DELIVERY;
softBounced |=
fact.type() == NotificationReceiptFact.Type.BOUNCE
&& fact.bounceClass() == NotificationReceiptFact.BounceClass.SOFT;
hardBounced |=
fact.type() == NotificationReceiptFact.Type.BOUNCE
&& fact.bounceClass() == NotificationReceiptFact.BounceClass.HARD;
complained |= fact.type() == NotificationReceiptFact.Type.COMPLAINT;
delayed |= fact.type() == NotificationReceiptFact.Type.DELIVERY_DELAY;
if (fact.occurredAt().isAfter(latest)) {
latest = fact.occurredAt();
}
}
return new NotificationReceiptProjection(
accepted,
delivered,
softBounced,
hardBounced,
complained,
delayed,
latest,
immutableFacts.size());
}
}
@@ -0,0 +1,52 @@
package dev.caskeleton.application.notification;
import java.util.List;
import java.util.Objects;
/** Durable receipt inbox and delivery projection mutation boundary. */
public interface NotificationReceiptStorePort {
AppendResult appendIfAbsent(NormalizedNotificationReceiptCommand command);
void saveProjection(NotificationDeliveryId deliveryId, NotificationReceiptProjection projection);
sealed interface AppendResult permits Appended, Duplicate {}
record Appended(ReceiptAggregate aggregate) implements AppendResult {
public Appended {
Objects.requireNonNull(aggregate, "notification receipt aggregate must be non-null");
}
}
record Duplicate(NotificationReceiptProjection projection) implements AppendResult {
public Duplicate {
Objects.requireNonNull(projection, "notification receipt projection must be non-null");
}
}
record ReceiptAggregate(
NotificationDeliveryId deliveryId,
NotificationRecipientReference recipient,
List<NotificationReceiptFact> facts) {
public ReceiptAggregate {
Objects.requireNonNull(deliveryId, "notification delivery ID must be non-null");
Objects.requireNonNull(recipient, "notification recipient reference must be non-null");
facts =
List.copyOf(Objects.requireNonNull(facts, "notification receipt facts must be non-null"));
if (facts.isEmpty() || facts.size() > 100) {
throw new IllegalArgumentException(
"notification receipt aggregate must contain 1..100 facts");
}
}
@Override
public String toString() {
return "ReceiptAggregate[deliveryId=<redacted>, recipient=<redacted>, factCount="
+ facts.size()
+ "]";
}
}
}
@@ -0,0 +1,8 @@
package dev.caskeleton.application.notification;
/** Channel-typed opaque recipient reference; raw addresses are forbidden at this boundary. */
public sealed interface NotificationRecipientReference
permits EmailRecipientReference, SlackAudienceReference {
NotificationChannel channel();
}
@@ -0,0 +1,19 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/** Performs one bounded provider reconciliation outside any database transaction. */
@FunctionalInterface
public interface NotificationReconciliationPort {
ReconciliationOutcome reconcile(NotificationDeliveryStorePort.ReconciliationClaim claim);
record ReconciliationOutcome(
SubmissionCertainty submissionCertainty, NotificationReasonCode reasonCode) {
public ReconciliationOutcome {
Objects.requireNonNull(submissionCertainty, "submission certainty must be non-null");
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
}
}
@@ -0,0 +1,71 @@
package dev.caskeleton.application.notification;
import java.util.List;
import java.util.Objects;
/** Closed result union for a notification request; append is not delivery success. */
public sealed interface NotificationRequestResult
permits NotificationRequestResult.InlineCompleted,
NotificationRequestResult.AppendedDurably,
NotificationRequestResult.DuplicateExistingIntent,
NotificationRequestResult.RejectedByBusinessPolicy,
NotificationRequestResult.RejectedInvalidRequest,
NotificationRequestResult.CapabilityUnavailable {
record InlineCompleted(NotificationIntentId intentId, List<TargetAttemptOutcome> outcomes)
implements NotificationRequestResult {
public InlineCompleted {
Objects.requireNonNull(intentId, "notification intent ID must be non-null");
Objects.requireNonNull(outcomes, "inline outcomes must be non-null");
outcomes = List.copyOf(outcomes);
if (outcomes.isEmpty() || outcomes.size() > 16) {
throw new IllegalArgumentException("inline outcomes must contain 1..16 targets");
}
long distinctOrdinals =
outcomes.stream().map(TargetAttemptOutcome::targetOrdinal).distinct().count();
if (distinctOrdinals != outcomes.size()) {
throw new IllegalArgumentException("inline target ordinals must be unique");
}
}
}
record AppendedDurably(NotificationIntentId intentId) implements NotificationRequestResult {
public AppendedDurably {
Objects.requireNonNull(intentId, "notification intent ID must be non-null");
}
}
record DuplicateExistingIntent(NotificationIntentId intentId)
implements NotificationRequestResult {
public DuplicateExistingIntent {
Objects.requireNonNull(intentId, "notification intent ID must be non-null");
}
}
record RejectedByBusinessPolicy(NotificationReasonCode reasonCode)
implements NotificationRequestResult {
public RejectedByBusinessPolicy {
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
}
record RejectedInvalidRequest(NotificationReasonCode reasonCode)
implements NotificationRequestResult {
public RejectedInvalidRequest {
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
}
record CapabilityUnavailable(NotificationReasonCode reasonCode)
implements NotificationRequestResult {
public CapabilityUnavailable {
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
}
}
@@ -0,0 +1,9 @@
package dev.caskeleton.application.notification;
/** Closed-catalog logical technical route; never a provider or endpoint identifier. */
public record NotificationRouteId(String value) {
public NotificationRouteId {
value = NotificationIntentId.requireSlug("routeId", value);
}
}
@@ -0,0 +1,8 @@
package dev.caskeleton.application.notification;
/** Closed provider-leg expansion strategy for one logical recipient. */
public enum NotificationRouteStrategy {
SINGLE,
FAN_OUT_ALL,
ORDERED_FALLBACK
}
@@ -0,0 +1,281 @@
package dev.caskeleton.application.notification;
import java.time.Duration;
import java.time.Instant;
import java.util.Arrays;
import java.util.Objects;
/**
* Bounded immutable signed-evidence header; cryptographic verification belongs to verifier ports.
*/
public final class NotificationSignedEvidenceHeader {
private final String canonicalProfile;
private final byte[] canonicalPayload;
private final byte[] signature;
private final String signatureAlgorithm;
private final String issuerKeyId;
private final byte[] issuerPublicKeySpki;
private final String issuerPublicKeyDigest;
private final NotificationEvidenceTrustSnapshot trustSnapshot;
private final Instant issuedAt;
private final Instant expiresAt;
private final Duration allowedClockSkew;
private final Duration acceptanceMargin;
private final String environmentId;
private final String databaseId;
private final String artifactId;
private final String consumerInventoryId;
private final String providerCallLedgerId;
private final String providerCallLedgerSnapshot;
private final int childCount;
private final String childSetDigest;
public NotificationSignedEvidenceHeader(
String canonicalProfile,
byte[] canonicalPayload,
byte[] signature,
String signatureAlgorithm,
String issuerKeyId,
byte[] issuerPublicKeySpki,
String issuerPublicKeyDigest,
NotificationEvidenceTrustSnapshot trustSnapshot,
Instant issuedAt,
Instant expiresAt,
Duration allowedClockSkew,
Duration acceptanceMargin,
String environmentId,
String databaseId,
String artifactId,
String consumerInventoryId,
String providerCallLedgerId,
String providerCallLedgerSnapshot,
int childCount,
String childSetDigest) {
this.canonicalProfile =
NotificationIntentId.requireSlug("signed evidence canonical profile", canonicalProfile);
this.canonicalPayload = copyBounded("canonical evidence payload", canonicalPayload, 1, 65_536);
this.signature = copyBounded("evidence signature", signature, 64, 128);
if (!"Ed25519".equals(signatureAlgorithm)) {
throw new IllegalArgumentException("signed evidence algorithm must be Ed25519");
}
this.signatureAlgorithm = signatureAlgorithm;
this.issuerKeyId = NotificationIntentId.requireOpaque("evidence issuer key ID", issuerKeyId);
this.issuerPublicKeySpki =
copyBounded("evidence issuer public-key SPKI", issuerPublicKeySpki, 32, 1_024);
this.issuerPublicKeyDigest =
InitializeNotificationWriterFencesCommand.requireDigest(issuerPublicKeyDigest);
this.trustSnapshot =
Objects.requireNonNull(trustSnapshot, "evidence trust snapshot must be non-null");
if (!this.issuerPublicKeyDigest.equals(trustSnapshot.issuerKeyDigest())) {
throw new IllegalArgumentException("evidence issuer key digest must match trust snapshot");
}
this.issuedAt = Objects.requireNonNull(issuedAt, "evidence issued-at must be non-null");
this.expiresAt = Objects.requireNonNull(expiresAt, "evidence expires-at must be non-null");
this.allowedClockSkew =
Objects.requireNonNull(allowedClockSkew, "evidence allowed clock skew must be non-null");
this.acceptanceMargin =
Objects.requireNonNull(acceptanceMargin, "evidence acceptance margin must be non-null");
if (!expiresAt.isAfter(issuedAt)
|| allowedClockSkew.isNegative()
|| allowedClockSkew.compareTo(Duration.ofMinutes(5)) > 0
|| acceptanceMargin.isNegative()
|| acceptanceMargin.compareTo(Duration.ofMinutes(5)) > 0
|| !expiresAt.minus(acceptanceMargin).isAfter(issuedAt.minus(allowedClockSkew))) {
throw new IllegalArgumentException("signed evidence acceptance window is invalid");
}
this.environmentId =
NotificationIntentId.requireOpaque("evidence environment identity", environmentId);
this.databaseId = NotificationIntentId.requireOpaque("evidence database identity", databaseId);
this.artifactId = NotificationIntentId.requireOpaque("evidence artifact identity", artifactId);
this.consumerInventoryId =
NotificationIntentId.requireOpaque(
"evidence consumer inventory identity", consumerInventoryId);
this.providerCallLedgerId =
NotificationIntentId.requireOpaque(
"evidence provider-call ledger identity", providerCallLedgerId);
this.providerCallLedgerSnapshot =
NotificationIntentId.requireOpaque(
"evidence provider-call ledger snapshot", providerCallLedgerSnapshot);
if (childCount < 0 || childCount > 1_000) {
throw new IllegalArgumentException("signed evidence child count must be in 0..1000");
}
this.childCount = childCount;
this.childSetDigest = InitializeNotificationWriterFencesCommand.requireDigest(childSetDigest);
}
public String canonicalProfile() {
return canonicalProfile;
}
public byte[] canonicalPayload() {
return Arrays.copyOf(canonicalPayload, canonicalPayload.length);
}
public byte[] signature() {
return Arrays.copyOf(signature, signature.length);
}
public String signatureAlgorithm() {
return signatureAlgorithm;
}
public String issuerKeyId() {
return issuerKeyId;
}
public byte[] issuerPublicKeySpki() {
return Arrays.copyOf(issuerPublicKeySpki, issuerPublicKeySpki.length);
}
public String issuerPublicKeyDigest() {
return issuerPublicKeyDigest;
}
public NotificationEvidenceTrustSnapshot trustSnapshot() {
return trustSnapshot;
}
public Instant issuedAt() {
return issuedAt;
}
public Instant expiresAt() {
return expiresAt;
}
public Duration allowedClockSkew() {
return allowedClockSkew;
}
public Duration acceptanceMargin() {
return acceptanceMargin;
}
public String environmentId() {
return environmentId;
}
public String databaseId() {
return databaseId;
}
public String artifactId() {
return artifactId;
}
public String consumerInventoryId() {
return consumerInventoryId;
}
public String providerCallLedgerId() {
return providerCallLedgerId;
}
public String providerCallLedgerSnapshot() {
return providerCallLedgerSnapshot;
}
public int childCount() {
return childCount;
}
public String childSetDigest() {
return childSetDigest;
}
@Override
public boolean equals(Object candidate) {
if (this == candidate) {
return true;
}
if (!(candidate instanceof NotificationSignedEvidenceHeader other)) {
return false;
}
return childCount == other.childCount
&& canonicalProfile.equals(other.canonicalProfile)
&& Arrays.equals(canonicalPayload, other.canonicalPayload)
&& Arrays.equals(signature, other.signature)
&& signatureAlgorithm.equals(other.signatureAlgorithm)
&& issuerKeyId.equals(other.issuerKeyId)
&& Arrays.equals(issuerPublicKeySpki, other.issuerPublicKeySpki)
&& issuerPublicKeyDigest.equals(other.issuerPublicKeyDigest)
&& trustSnapshot.equals(other.trustSnapshot)
&& issuedAt.equals(other.issuedAt)
&& expiresAt.equals(other.expiresAt)
&& allowedClockSkew.equals(other.allowedClockSkew)
&& acceptanceMargin.equals(other.acceptanceMargin)
&& environmentId.equals(other.environmentId)
&& databaseId.equals(other.databaseId)
&& artifactId.equals(other.artifactId)
&& consumerInventoryId.equals(other.consumerInventoryId)
&& providerCallLedgerId.equals(other.providerCallLedgerId)
&& providerCallLedgerSnapshot.equals(other.providerCallLedgerSnapshot)
&& childSetDigest.equals(other.childSetDigest);
}
@Override
public int hashCode() {
int result =
Objects.hash(
canonicalProfile,
signatureAlgorithm,
issuerKeyId,
issuerPublicKeyDigest,
trustSnapshot,
issuedAt,
expiresAt,
allowedClockSkew,
acceptanceMargin,
environmentId,
databaseId,
artifactId,
consumerInventoryId,
providerCallLedgerId,
providerCallLedgerSnapshot,
childCount,
childSetDigest);
result = 31 * result + Arrays.hashCode(canonicalPayload);
result = 31 * result + Arrays.hashCode(signature);
result = 31 * result + Arrays.hashCode(issuerPublicKeySpki);
return result;
}
@Override
public String toString() {
return "NotificationSignedEvidenceHeader[canonicalProfile="
+ canonicalProfile
+ ", canonicalPayload=<redacted>, signature=<redacted>, signatureAlgorithm="
+ signatureAlgorithm
+ ", issuerKeyId="
+ issuerKeyId
+ ", issuerPublicKeySpki=<redacted>, issuerPublicKeyDigest="
+ issuerPublicKeyDigest
+ ", trustSnapshot="
+ trustSnapshot
+ ", issuedAt="
+ issuedAt
+ ", expiresAt="
+ expiresAt
+ ", consumerInventoryId="
+ consumerInventoryId
+ ", providerCallLedgerId="
+ providerCallLedgerId
+ ", providerCallLedgerSnapshot="
+ providerCallLedgerSnapshot
+ ", childCount="
+ childCount
+ ", childSetDigest="
+ childSetDigest
+ "]";
}
private static byte[] copyBounded(
String field, byte[] value, int minimumLength, int maximumLength) {
if (value == null || value.length < minimumLength || value.length > maximumLength) {
throw new IllegalArgumentException(
field + " length must be in " + minimumLength + ".." + maximumLength);
}
return Arrays.copyOf(value, value.length);
}
}
@@ -0,0 +1,41 @@
package dev.caskeleton.application.notification;
import java.util.Collections;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
/** Durable store capabilities and retained frozen revisions visible to application validation. */
public record NotificationStoreCapabilityDescriptor(
boolean durableIntentStore,
boolean attemptJournal,
boolean receiptInbox,
int maximumBatch,
Set<Integer> availablePolicyRevisions,
Set<NotificationTemplateRef> availableTemplateRevisions) {
public NotificationStoreCapabilityDescriptor {
if (maximumBatch < 1 || maximumBatch > 100) {
throw new IllegalArgumentException("notification store maximum batch must be in 1..100");
}
Objects.requireNonNull(availablePolicyRevisions, "available policy revisions must be non-null");
TreeSet<Integer> policies = new TreeSet<>(availablePolicyRevisions);
if (policies.isEmpty()
|| policies.size() > 100
|| policies.stream().anyMatch(revision -> revision == null || revision < 1)) {
throw new IllegalArgumentException(
"available policy revisions must contain 1..100 positive revisions");
}
availablePolicyRevisions = Collections.unmodifiableSet(policies);
Objects.requireNonNull(
availableTemplateRevisions, "available template revisions must be non-null");
HashSet<NotificationTemplateRef> templates = new HashSet<>(availableTemplateRevisions);
if (templates.isEmpty()
|| templates.size() > 100
|| templates.stream().anyMatch(Objects::isNull)) {
throw new IllegalArgumentException("available template revisions must contain 1..100 values");
}
availableTemplateRevisions = Collections.unmodifiableSet(templates);
}
}
@@ -0,0 +1,32 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
import java.util.Objects;
/** Persists technical suppression caused by hard bounce or complaint, not business consent. */
@FunctionalInterface
public interface NotificationTechnicalSuppressionPort {
void suppress(SuppressionMutation mutation);
record SuppressionMutation(
NotificationRecipientReference recipient,
NotificationReasonCode reasonCode,
Instant suppressedAt) {
public SuppressionMutation {
Objects.requireNonNull(recipient, "notification recipient reference must be non-null");
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
Objects.requireNonNull(suppressedAt, "notification suppression time must be non-null");
}
@Override
public String toString() {
return "SuppressionMutation[recipient=<redacted>, reasonCode="
+ reasonCode
+ ", suppressedAt="
+ suppressedAt
+ "]";
}
}
}
@@ -0,0 +1,36 @@
package dev.caskeleton.application.notification;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
/** Bounded, immutable and redacted template parameter bag over a closed scalar value set. */
public record NotificationTemplateParameters(Map<String, NotificationTemplateValue> values) {
private static final int MAXIMUM_PARAMETERS = 32;
public NotificationTemplateParameters {
Objects.requireNonNull(values, "template parameters must be non-null");
if (values.size() > MAXIMUM_PARAMETERS) {
throw new IllegalArgumentException(
"template parameters exceed " + MAXIMUM_PARAMETERS + " entries");
}
LinkedHashMap<String, NotificationTemplateValue> copy = new LinkedHashMap<>();
values.forEach(
(name, value) -> {
if (name == null || !name.matches("[a-z][A-Za-z0-9]{0,63}")) {
throw new IllegalArgumentException(
"template parameter name must match [a-z][A-Za-z0-9]{0,63}");
}
copy.put(
name, Objects.requireNonNull(value, "template parameter value must be non-null"));
});
values = Collections.unmodifiableMap(copy);
}
@Override
public String toString() {
return "NotificationTemplateParameters[names=" + values.keySet() + ", values=<redacted>]";
}
}
@@ -0,0 +1,12 @@
package dev.caskeleton.application.notification;
/** Immutable checked-in template identity and version. */
public record NotificationTemplateRef(String templateId, int version) {
public NotificationTemplateRef {
templateId = NotificationIntentId.requireSlug("templateId", templateId);
if (version < 1 || version > 1_000_000) {
throw new IllegalArgumentException("template version must be in 1..1000000");
}
}
}
@@ -0,0 +1,103 @@
package dev.caskeleton.application.notification;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Currency;
import java.util.Objects;
/**
* Closed scalar set accepted by the generic template boundary. Provider objects, raw HTML, JSON,
* collections and arbitrary objects cannot implement this sealed contract.
*/
public sealed interface NotificationTemplateValue
permits NotificationTemplateValue.SafeText,
NotificationTemplateValue.TrustedAbsoluteLinkReference,
NotificationTemplateValue.LocalDateValue,
NotificationTemplateValue.LocalDateTimeValue,
NotificationTemplateValue.IntegerValue,
NotificationTemplateValue.MoneyValue {
record SafeText(String value) implements NotificationTemplateValue {
public SafeText {
if (value == null || value.isBlank() || value.length() > 4_096) {
throw new IllegalArgumentException("safe text must contain 1..4096 characters");
}
if (value.chars().anyMatch(character -> character == 0)) {
throw new IllegalArgumentException("safe text must not contain NUL");
}
}
@Override
public String toString() {
return "SafeText[value=<redacted>]";
}
}
record TrustedAbsoluteLinkReference(String value) implements NotificationTemplateValue {
public TrustedAbsoluteLinkReference {
value = NotificationIntentId.requireOpaque("trusted link reference", value);
}
@Override
public String toString() {
return "TrustedAbsoluteLinkReference[value=<redacted>]";
}
}
record LocalDateValue(LocalDate value) implements NotificationTemplateValue {
public LocalDateValue {
Objects.requireNonNull(value, "local date value must be non-null");
}
@Override
public String toString() {
return "LocalDateValue[value=<redacted>]";
}
}
record LocalDateTimeValue(LocalDateTime value, ZoneId zone) implements NotificationTemplateValue {
public LocalDateTimeValue {
Objects.requireNonNull(value, "local date-time value must be non-null");
Objects.requireNonNull(zone, "business time zone must be non-null");
if (zone.getId().length() > 64) {
throw new IllegalArgumentException("business time zone exceeds 64 characters");
}
}
@Override
public String toString() {
return "LocalDateTimeValue[value=<redacted>, zone=<redacted>]";
}
}
record IntegerValue(long value) implements NotificationTemplateValue {
@Override
public String toString() {
return "IntegerValue[value=<redacted>]";
}
}
record MoneyValue(BigDecimal amount, Currency currency) implements NotificationTemplateValue {
public MoneyValue {
Objects.requireNonNull(amount, "money amount must be non-null");
Objects.requireNonNull(currency, "money currency must be non-null");
if (amount.scale() < 0 || amount.scale() > 4 || amount.precision() > 19) {
throw new IllegalArgumentException(
"money amount must have precision at most 19 and scale in 0..4");
}
}
@Override
public String toString() {
return "MoneyValue[amount=<redacted>, currency=<redacted>]";
}
}
}
@@ -0,0 +1,17 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
/** Durable legacy permit mutation boundary; ownership operations use dedicated operation ports. */
public interface NotificationWriterCutoverPort {
NotificationLegacyWriterPermitResult acquireLegacyPermit(
NotificationLegacyWriterPermitCommand command,
NotificationWriterRouteSet.RouteProfile routeProfile,
Instant requestedAt);
NotificationLegacyWriterPermitResult releaseLegacyPermit(
NotificationLegacyWriterPermitCommand command,
NotificationWriterRouteSet.RouteProfile routeProfile,
Instant requestedAt);
}
@@ -0,0 +1,36 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
import java.util.Collections;
import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
/**
* Verified complete BEGIN inventory evidence; never construct from caller-authored digests alone.
*/
public record NotificationWriterInventoryEvidence(
NotificationCanonicalWriterRouteSet.RouteRevision route,
long generation,
Set<String> nodeIds,
String nodeSetDigest,
Instant verifiedAt) {
public NotificationWriterInventoryEvidence {
Objects.requireNonNull(route, "verified writer inventory route must be non-null");
if (generation < 0) {
throw new IllegalArgumentException(
"verified writer inventory generation must be non-negative");
}
Objects.requireNonNull(nodeIds, "verified writer inventory nodes must be non-null");
TreeSet<String> nodes = new TreeSet<>();
nodeIds.forEach(
node -> nodes.add(NotificationIntentId.requireOpaque("writer inventory node ID", node)));
if (nodes.size() > 100) {
throw new IllegalArgumentException("verified writer inventory exceeds 100 nodes");
}
nodeIds = Collections.unmodifiableSet(nodes);
nodeSetDigest = InitializeNotificationWriterFencesCommand.requireDigest(nodeSetDigest);
Objects.requireNonNull(verifiedAt, "writer inventory verification time must be non-null");
}
}
@@ -0,0 +1,14 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
/** Verifies signed BEGIN inventory against the closed issuer trust catalog and exact context. */
@FunctionalInterface
public interface NotificationWriterInventoryEvidenceVerifierPort {
NotificationWriterInventoryEvidence verify(
SignedNotificationWriterInventoryManifest manifest,
NotificationCanonicalWriterRouteSet.RouteRevision expectedRoute,
long expectedGeneration,
Instant verifiedAt);
}
@@ -0,0 +1,7 @@
package dev.caskeleton.application.notification;
/** Exclusive writer owner stored in the route-specific database fence. */
public enum NotificationWriterOwnership {
LEGACY,
CANONICAL
}
@@ -0,0 +1,34 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
import java.util.Objects;
/** Verifies signed quiescence payload, issuer trust and exact route/registry context. */
@FunctionalInterface
public interface NotificationWriterQuiescenceAttestationPort {
VerifiedQuiescenceEvidence verify(
SignedNotificationWriterQuiescenceManifest manifest,
NotificationCanonicalWriterRouteSet.RouteRevision expectedRoute,
long expectedGeneration,
NotificationWriterRouteSet.RouteProfile trustedRoute,
Instant verifiedAt);
record VerifiedQuiescenceEvidence(
NotificationCanonicalWriterRouteSet.RouteRevision route,
long generation,
String childSetDigest,
int childCount,
Instant verifiedAt) {
public VerifiedQuiescenceEvidence {
Objects.requireNonNull(route, "verified quiescence route must be non-null");
if (generation < 0 || childCount < 0 || childCount > 1_000) {
throw new IllegalArgumentException(
"verified quiescence generation/count is outside bounds");
}
childSetDigest = InitializeNotificationWriterFencesCommand.requireDigest(childSetDigest);
Objects.requireNonNull(verifiedAt, "quiescence verification time must be non-null");
}
}
}
@@ -0,0 +1,157 @@
package dev.caskeleton.application.notification;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
/**
* PRE-only route set decorating the canonical route keys with legacy aliases and the complete
* current-plus-retiring transport proof registry.
*/
public record NotificationWriterRouteSet(
NotificationCanonicalWriterRouteSet canonicalRoutes, List<RouteProfile> routeProfiles) {
public NotificationWriterRouteSet {
Objects.requireNonNull(canonicalRoutes, "canonical writer route set must be non-null");
Objects.requireNonNull(routeProfiles, "writer route profiles must be non-null");
routeProfiles =
routeProfiles.stream()
.map(
profile -> Objects.requireNonNull(profile, "writer route profile must be non-null"))
.sorted(
Comparator.comparing((RouteProfile profile) -> profile.route().routeId().value())
.thenComparingInt(profile -> profile.route().routeRevision()))
.toList();
if (routeProfiles.size() != canonicalRoutes.routes().size()) {
throw new IllegalArgumentException(
"writer route profile keys must exactly equal canonical route keys");
}
if (new HashSet<>(routeProfiles.stream().map(RouteProfile::route).toList()).size()
!= routeProfiles.size()) {
throw new IllegalArgumentException("writer route profiles contain duplicate routes");
}
if (!routeProfiles.stream()
.map(RouteProfile::route)
.toList()
.equals(canonicalRoutes.routes())) {
throw new IllegalArgumentException(
"writer route profile keys must exactly equal canonical route keys");
}
}
public RouteProfile requireRoute(NotificationCanonicalWriterRouteSet.RouteRevision route) {
return routeProfiles.stream()
.filter(candidate -> candidate.route().equals(route))
.findFirst()
.orElseThrow(
() -> new IllegalArgumentException("route is outside trusted writer route set"));
}
public String digest() {
MessageDigest digest = sha256();
NotificationCanonicalWriterRouteSet.update(digest, canonicalRoutes.digest());
routeProfiles.forEach(
route -> {
NotificationCanonicalWriterRouteSet.update(digest, route.route().routeId().value());
digest.update(
ByteBuffer.allocate(Integer.BYTES).putInt(route.route().routeRevision()).array());
NotificationCanonicalWriterRouteSet.update(digest, route.legacyAlias().orElse(""));
route
.transportProfiles()
.forEach(
profile -> {
NotificationCanonicalWriterRouteSet.update(digest, profile.profileId());
NotificationCanonicalWriterRouteSet.update(digest, profile.proofClass().name());
NotificationCanonicalWriterRouteSet.update(digest, profile.evidenceRevision());
digest.update((byte) (profile.activeAdmissionProfile() ? 1 : 0));
});
});
return java.util.HexFormat.of().formatHex(digest.digest());
}
private static MessageDigest sha256() {
try {
return MessageDigest.getInstance("SHA-256");
} catch (NoSuchAlgorithmException unavailable) {
throw new IllegalStateException(
"SHA-256 must be available on every Java runtime", unavailable);
}
}
public record RouteProfile(
NotificationCanonicalWriterRouteSet.RouteRevision route,
Optional<String> legacyAlias,
List<TransportProfile> transportProfiles) {
public RouteProfile {
Objects.requireNonNull(route, "notification writer route must be non-null");
Objects.requireNonNull(legacyAlias, "legacy alias container must be non-null");
legacyAlias =
legacyAlias.map(alias -> NotificationIntentId.requireSlug("legacy route alias", alias));
Objects.requireNonNull(
transportProfiles, "legacy transport profile registry must be non-null");
transportProfiles =
transportProfiles.stream()
.map(
profile ->
Objects.requireNonNull(profile, "legacy transport profile must be non-null"))
.sorted(Comparator.comparing(TransportProfile::profileId))
.toList();
if (transportProfiles.isEmpty() || transportProfiles.size() > 8) {
throw new IllegalArgumentException(
"legacy transport profile registry must contain 1..8 profiles");
}
if (new HashSet<>(transportProfiles.stream().map(TransportProfile::profileId).toList()).size()
!= transportProfiles.size()) {
throw new IllegalArgumentException("legacy transport profile registry contains duplicates");
}
long activeCount =
transportProfiles.stream().filter(TransportProfile::activeAdmissionProfile).count();
if (activeCount != 1) {
throw new IllegalArgumentException(
"legacy transport registry requires exactly one active admission profile");
}
}
public ProofClass proofRequirement() {
return transportProfiles.stream()
.allMatch(profile -> profile.proofClass() == ProofClass.HARD_BOUND_PROVEN)
? ProofClass.HARD_BOUND_PROVEN
: ProofClass.QUIESCENCE_REQUIRED;
}
public TransportProfile requireProfile(String profileId) {
return transportProfiles.stream()
.filter(profile -> profile.profileId().equals(profileId))
.findFirst()
.orElseThrow(
() ->
new IllegalArgumentException(
"transport profile is outside trusted writer registry"));
}
}
public record TransportProfile(
String profileId,
ProofClass proofClass,
String evidenceRevision,
boolean activeAdmissionProfile) {
public TransportProfile {
profileId = NotificationIntentId.requireSlug("legacy transport profile ID", profileId);
Objects.requireNonNull(proofClass, "legacy transport proof class must be non-null");
evidenceRevision =
NotificationIntentId.requireSlug("legacy transport evidence revision", evidenceRevision);
}
}
public enum ProofClass {
HARD_BOUND_PROVEN,
QUIESCENCE_REQUIRED
}
}
@@ -0,0 +1,92 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
import java.util.Objects;
import java.util.Optional;
/**
* Provider-neutral attempt fact with submission, coordinator action and failure scope kept as
* separate axes.
*/
public record ProviderAttemptOutcome(
SubmissionCertainty submissionCertainty,
RetryDisposition retryDisposition,
NotificationFaultScope faultScope,
NotificationReasonCode reasonCode,
Optional<Instant> retryNotBefore,
String attemptCorrelationReference,
Optional<String> providerMessageReference) {
public ProviderAttemptOutcome {
Objects.requireNonNull(submissionCertainty, "submission certainty must be non-null");
Objects.requireNonNull(retryDisposition, "retry disposition must be non-null");
Objects.requireNonNull(faultScope, "fault scope must be non-null");
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
Objects.requireNonNull(retryNotBefore, "retry-not-before container must be non-null");
attemptCorrelationReference =
NotificationIntentId.requireOpaque(
"attempt correlation reference", attemptCorrelationReference);
Objects.requireNonNull(
providerMessageReference, "provider message reference container must be non-null");
providerMessageReference =
providerMessageReference.map(
value -> NotificationIntentId.requireOpaque("provider message reference", value));
validateAxes(
submissionCertainty,
retryDisposition,
faultScope,
retryNotBefore,
providerMessageReference);
}
private static void validateAxes(
SubmissionCertainty certainty,
RetryDisposition disposition,
NotificationFaultScope scope,
Optional<Instant> retryAt,
Optional<String> providerReference) {
if ((disposition == RetryDisposition.RETRY_AT) != retryAt.isPresent()) {
throw new IllegalArgumentException(
"retryNotBefore must be present exactly when retry disposition is RETRY_AT");
}
if (certainty == SubmissionCertainty.PROVIDER_ACCEPTED
&& (disposition != RetryDisposition.NOT_APPLICABLE
|| scope != NotificationFaultScope.DELIVERY)) {
throw new IllegalArgumentException("PROVIDER_ACCEPTED requires NOT_APPLICABLE and DELIVERY");
}
if (certainty == SubmissionCertainty.INDETERMINATE
&& (disposition != RetryDisposition.NOT_APPLICABLE
|| scope != NotificationFaultScope.DELIVERY
|| providerReference.isPresent())) {
throw new IllegalArgumentException(
"INDETERMINATE requires NOT_APPLICABLE, DELIVERY and no provider message reference");
}
if (certainty == SubmissionCertainty.DEFINITELY_NOT_APPLIED
&& disposition == RetryDisposition.NOT_APPLICABLE) {
throw new IllegalArgumentException(
"DEFINITELY_NOT_APPLIED requires an explicit retry, park or terminal disposition");
}
if (certainty != SubmissionCertainty.PROVIDER_ACCEPTED && providerReference.isPresent()) {
throw new IllegalArgumentException(
"provider message reference is only valid for a provider-accepted outcome");
}
if (disposition == RetryDisposition.PARK_BINDING && scope == NotificationFaultScope.DELIVERY) {
throw new IllegalArgumentException("PARK_BINDING requires a shared non-delivery fault scope");
}
}
@Override
public String toString() {
return "ProviderAttemptOutcome[submissionCertainty="
+ submissionCertainty
+ ", retryDisposition="
+ retryDisposition
+ ", faultScope="
+ faultScope
+ ", reasonCode="
+ reasonCode
+ ", retryNotBefore="
+ retryNotBefore
+ ", attemptCorrelationReference=<redacted>, providerMessageReference=<redacted>]";
}
}
@@ -0,0 +1,19 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.command.Command;
/** Requests bounded provider reconciliation and/or local orphan-receipt attachment. */
public record ReconcileNotificationDeliveriesCommand(
int maximumClaims, int maximumOrphanAttachments) implements Command {
public ReconcileNotificationDeliveriesCommand {
if (maximumClaims < 0
|| maximumClaims > 100
|| maximumOrphanAttachments < 0
|| maximumOrphanAttachments > 100
|| maximumClaims + maximumOrphanAttachments == 0) {
throw new IllegalArgumentException(
"reconciliation bounds must each be in 0..100 and at least one must be positive");
}
}
}
@@ -0,0 +1,18 @@
package dev.caskeleton.application.notification;
/** Bounded non-sensitive aggregate outcome of one reconciliation cycle. */
public record ReconcileNotificationDeliveriesResult(
int claimedCount, int providerCallCount, int finalizedCount, int orphanAttachedCount) {
public ReconcileNotificationDeliveriesResult {
int[] counts = {claimedCount, providerCallCount, finalizedCount, orphanAttachedCount};
for (int count : counts) {
if (count < 0 || count > 100) {
throw new IllegalArgumentException("reconciliation counts must be in 0..100");
}
}
if (providerCallCount > claimedCount || finalizedCount > providerCallCount) {
throw new IllegalArgumentException("reconciliation counts are inconsistent");
}
}
}
@@ -0,0 +1,84 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.capability.Idempotency;
import dev.caskeleton.application.capability.RepositoryAccess;
import dev.caskeleton.application.capability.UseCaseCapability;
import dev.caskeleton.application.security.RequiresPermission;
import dev.caskeleton.application.transaction.TransactionMode;
import dev.caskeleton.application.transaction.TransactionPort;
import dev.caskeleton.application.usecase.CommandUseCase;
import java.time.Clock;
import java.util.List;
import java.util.Objects;
/**
* Coordinates local orphan attach and provider reconciliation with short transaction boundaries.
*/
@RequiresPermission("notification:reconcile")
@UseCaseCapability(
transactionMode = TransactionMode.WRITE,
idempotency = Idempotency.IDEMPOTENT,
repositoryAccess = RepositoryAccess.WRITE_REPOSITORY,
externalOutboundAllowed = true,
sensitiveRead = true,
crossTenantAdmin = true)
public final class ReconcileNotificationDeliveriesUseCase
implements CommandUseCase<
ReconcileNotificationDeliveriesCommand, ReconcileNotificationDeliveriesResult> {
private final NotificationDeliveryStorePort store;
private final NotificationReconciliationPort provider;
private final TransactionPort transactions;
private final Clock clock;
public ReconcileNotificationDeliveriesUseCase(
NotificationDeliveryStorePort store,
NotificationReconciliationPort provider,
TransactionPort transactions,
Clock clock) {
this.store = Objects.requireNonNull(store, "notification delivery store must be non-null");
this.provider =
Objects.requireNonNull(provider, "notification reconciliation port must be non-null");
this.transactions = Objects.requireNonNull(transactions, "transaction port must be non-null");
this.clock = Objects.requireNonNull(clock, "clock must be non-null");
}
@Override
public ReconcileNotificationDeliveriesResult handle(
ReconcileNotificationDeliveriesCommand command) {
Objects.requireNonNull(command, "reconciliation command must be non-null");
int orphanAttached =
command.maximumOrphanAttachments() == 0
? 0
: transactions.inWrite(
() ->
store.attachOrphanReceipts(
command.maximumOrphanAttachments(), clock.instant()));
List<NotificationDeliveryStorePort.ReconciliationClaim> claims =
command.maximumClaims() == 0
? List.of()
: List.copyOf(
transactions.inWrite(
() -> store.claimForReconciliation(command.maximumClaims(), clock.instant())));
if (claims.size() > command.maximumClaims()) {
throw new IllegalStateException("notification store returned too many reconciliation claims");
}
int finalized = 0;
for (NotificationDeliveryStorePort.ReconciliationClaim claim : claims) {
NotificationReconciliationPort.ReconciliationOutcome outcome =
Objects.requireNonNull(
provider.reconcile(claim), "reconciliation outcome must be non-null");
NotificationDeliveryStorePort.ReconciliationFinalizationResult finalization =
transactions.inWrite(() -> store.finalizeReconciliation(claim, outcome, clock.instant()));
if (finalization == NotificationDeliveryStorePort.ReconciliationFinalizationResult.APPLIED
|| finalization
== NotificationDeliveryStorePort.ReconciliationFinalizationResult
.LATE_EXACT_APPLIED) {
finalized++;
}
}
return new ReconcileNotificationDeliveriesResult(
claims.size(), claims.size(), finalized, orphanAttached);
}
}
@@ -0,0 +1,23 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.command.Command;
import java.util.Objects;
/** Authenticated PRE request to verify and retain one immutable signed quiescence manifest. */
public record RecordNotificationWriterQuiescenceAttestationCommand(
String operationToken,
SignedNotificationWriterQuiescenceManifest manifest,
String actorReference,
NotificationReasonCode reasonCode)
implements Command {
public RecordNotificationWriterQuiescenceAttestationCommand {
operationToken =
NotificationIntentId.requireOpaque(
"quiescence attestation operation token", operationToken);
Objects.requireNonNull(manifest, "signed quiescence manifest must be non-null");
actorReference =
NotificationIntentId.requireOpaque("quiescence attestation actor", actorReference);
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
}
@@ -0,0 +1,14 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
/** Atomically re-derives current blocking sets and retains verified signed quiescence evidence. */
@FunctionalInterface
public interface RecordNotificationWriterQuiescenceAttestationOperation {
RecordNotificationWriterQuiescenceAttestationResult record(
RecordNotificationWriterQuiescenceAttestationCommand command,
NotificationWriterQuiescenceAttestationPort.VerifiedQuiescenceEvidence verifiedEvidence,
NotificationWriterRouteSet.RouteProfile trustedRoute,
Instant requestedAt);
}
@@ -0,0 +1,29 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/** Committed retained quiescence evidence identity. */
public record RecordNotificationWriterQuiescenceAttestationResult(
Status status,
String operationToken,
NotificationCanonicalWriterRouteSet.RouteRevision route,
long generation,
String childSetDigest) {
public RecordNotificationWriterQuiescenceAttestationResult {
Objects.requireNonNull(status, "quiescence attestation status must be non-null");
operationToken =
NotificationIntentId.requireOpaque(
"quiescence attestation operation token", operationToken);
Objects.requireNonNull(route, "quiescence attestation route must be non-null");
if (generation < 0) {
throw new IllegalArgumentException("quiescence generation must be non-negative");
}
childSetDigest = InitializeNotificationWriterFencesCommand.requireDigest(childSetDigest);
}
public enum Status {
RECORDED,
REPLAYED
}
}
@@ -0,0 +1,80 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.capability.Idempotency;
import dev.caskeleton.application.capability.RepositoryAccess;
import dev.caskeleton.application.capability.UseCaseCapability;
import dev.caskeleton.application.security.RequiresPermission;
import dev.caskeleton.application.transaction.TransactionMode;
import dev.caskeleton.application.transaction.TransactionPort;
import dev.caskeleton.application.usecase.CommandUseCase;
import java.time.Clock;
import java.util.Objects;
import java.util.Set;
/**
* Verifies signed quiescence outside a transaction, then root-commits its exact retained evidence.
*/
@RequiresPermission("notification:cutover-attest")
@UseCaseCapability(
transactionMode = TransactionMode.WRITE,
idempotency = Idempotency.IDEMPOTENT,
repositoryAccess = RepositoryAccess.WRITE_REPOSITORY,
crossTenantAdmin = true)
public final class RecordNotificationWriterQuiescenceAttestationUseCase
implements CommandUseCase<
RecordNotificationWriterQuiescenceAttestationCommand,
RecordNotificationWriterQuiescenceAttestationResult> {
private final NotificationWriterRouteSet routes;
private final NotificationWriterQuiescenceAttestationPort verifier;
private final RecordNotificationWriterQuiescenceAttestationOperation operation;
private final TransactionPort transactions;
private final Clock clock;
public RecordNotificationWriterQuiescenceAttestationUseCase(
NotificationWriterRouteSet routes,
NotificationWriterQuiescenceAttestationPort verifier,
RecordNotificationWriterQuiescenceAttestationOperation operation,
TransactionPort transactions,
Clock clock) {
this.routes = Objects.requireNonNull(routes, "notification writer route set must be non-null");
this.verifier =
Objects.requireNonNull(verifier, "quiescence evidence verifier must be non-null");
this.operation =
Objects.requireNonNull(operation, "quiescence attestation operation must be non-null");
this.transactions = Objects.requireNonNull(transactions, "transaction port must be non-null");
this.clock = Objects.requireNonNull(clock, "clock must be non-null");
}
@Override
public RecordNotificationWriterQuiescenceAttestationResult handle(
RecordNotificationWriterQuiescenceAttestationCommand command) {
Objects.requireNonNull(command, "quiescence attestation command must be non-null");
SignedNotificationWriterQuiescenceManifest manifest = command.manifest();
NotificationWriterRouteSet.RouteProfile route = routes.requireRoute(manifest.route());
if (route.proofRequirement() != NotificationWriterRouteSet.ProofClass.QUIESCENCE_REQUIRED) {
throw new IllegalArgumentException(
"all-hard-bound writer route forbids quiescence attestation evidence");
}
Set<String> trustedProfileIds =
route.transportProfiles().stream()
.map(NotificationWriterRouteSet.TransportProfile::profileId)
.collect(java.util.stream.Collectors.toUnmodifiableSet());
if (!manifest.transportProfileIds().equals(trustedProfileIds)) {
throw new IllegalArgumentException(
"signed quiescence transport profiles must exactly match trusted writer registry");
}
NotificationWriterQuiescenceAttestationPort.VerifiedQuiescenceEvidence evidence =
Objects.requireNonNull(
verifier.verify(
manifest, manifest.route(), manifest.drainingGeneration(), route, clock.instant()),
"verified quiescence evidence must be non-null");
if (!evidence.route().equals(manifest.route())
|| evidence.generation() != manifest.drainingGeneration()) {
throw new IllegalArgumentException(
"verified quiescence evidence does not match command route/generation");
}
return transactions.inRootWrite(
() -> operation.record(command, evidence, route, clock.instant()));
}
}
@@ -0,0 +1,9 @@
package dev.caskeleton.application.notification;
/** Coordinator action after one provider attempt. */
public enum RetryDisposition {
RETRY_AT,
PARK_BINDING,
TERMINAL,
NOT_APPLICABLE
}
@@ -0,0 +1,46 @@
package dev.caskeleton.application.notification;
import java.util.List;
import java.util.Objects;
import java.util.Set;
/** Signed complete old-writer node inventory presented to BEGIN_DRAIN. */
public record SignedNotificationWriterInventoryManifest(
NotificationSignedEvidenceHeader header,
NotificationCanonicalWriterRouteSet.RouteRevision route,
long generation,
List<NodeInventory> nodes) {
public SignedNotificationWriterInventoryManifest {
Objects.requireNonNull(header, "signed inventory header must be non-null");
Objects.requireNonNull(route, "signed inventory route must be non-null");
if (generation < 0) {
throw new IllegalArgumentException("signed inventory generation must be non-negative");
}
nodes = List.copyOf(Objects.requireNonNull(nodes, "signed inventory nodes must be non-null"));
if (nodes.size() > 100 || nodes.size() > header.childCount()) {
throw new IllegalArgumentException(
"signed inventory node count exceeds bounded signed child count");
}
if (nodes.stream()
.map(NodeInventory::nodeId)
.collect(java.util.stream.Collectors.toSet())
.size()
!= nodes.size()) {
throw new IllegalArgumentException("signed inventory contains duplicate node IDs");
}
}
public Set<String> nodeIds() {
return java.util.Collections.unmodifiableSet(
new java.util.TreeSet<>(nodes.stream().map(NodeInventory::nodeId).toList()));
}
public record NodeInventory(String nodeId, String artifactId) {
public NodeInventory {
nodeId = NotificationIntentId.requireOpaque("writer inventory node ID", nodeId);
artifactId = NotificationIntentId.requireOpaque("writer inventory artifact ID", artifactId);
}
}
}
@@ -0,0 +1,74 @@
package dev.caskeleton.application.notification;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
/** Signed exact irreversible quiescence evidence for a QUIESCENCE_REQUIRED route generation. */
public record SignedNotificationWriterQuiescenceManifest(
NotificationSignedEvidenceHeader header,
NotificationCanonicalWriterRouteSet.RouteRevision route,
long drainingGeneration,
Set<String> transportProfileIds,
List<NodeQuiescence> nodes,
int blockingPermitCount,
String blockingPermitSetDigest,
Set<String> permitHolderIds,
int productionConsumerCount,
int providerCallOpenCount) {
public SignedNotificationWriterQuiescenceManifest {
Objects.requireNonNull(header, "signed quiescence header must be non-null");
Objects.requireNonNull(route, "signed quiescence route must be non-null");
if (drainingGeneration < 0) {
throw new IllegalArgumentException("draining generation must be non-negative");
}
Objects.requireNonNull(
transportProfileIds, "signed quiescence transport profiles must be non-null");
TreeSet<String> profiles = new TreeSet<>();
transportProfileIds.forEach(
profile ->
profiles.add(NotificationIntentId.requireSlug("legacy transport profile ID", profile)));
if (profiles.isEmpty() || profiles.size() > 8) {
throw new IllegalArgumentException(
"signed quiescence transport profile set must contain 1..8 profiles");
}
transportProfileIds = java.util.Collections.unmodifiableSet(profiles);
nodes = List.copyOf(Objects.requireNonNull(nodes, "signed quiescence nodes must be non-null"));
if (nodes.size() > 100 || nodes.size() > header.childCount()) {
throw new IllegalArgumentException(
"signed quiescence node count exceeds bounded signed child count");
}
if (blockingPermitCount < 0 || blockingPermitCount > 100) {
throw new IllegalArgumentException("blocking permit count must be in 0..100");
}
blockingPermitSetDigest =
InitializeNotificationWriterFencesCommand.requireDigest(blockingPermitSetDigest);
Objects.requireNonNull(permitHolderIds, "permit holder set must be non-null");
TreeSet<String> holders = new TreeSet<>();
permitHolderIds.forEach(
holder -> holders.add(NotificationIntentId.requireOpaque("legacy permit holder", holder)));
if (holders.size() > 100) {
throw new IllegalArgumentException("permit holder set exceeds 100 entries");
}
permitHolderIds = java.util.Collections.unmodifiableSet(holders);
if (productionConsumerCount != 0 || providerCallOpenCount != 0) {
throw new IllegalArgumentException(
"quiescence evidence requires production consumer and provider open counts of zero");
}
}
public record NodeQuiescence(
String nodeId,
boolean retired,
boolean quiesced,
boolean deploymentTombstoned,
boolean credentialRevoked,
boolean egressRevoked) {
public NodeQuiescence {
nodeId = NotificationIntentId.requireOpaque("quiescence node ID", nodeId);
}
}
}
@@ -0,0 +1,25 @@
package dev.caskeleton.application.notification;
/** Opaque Slack workspace binding and audience references; neither value is a webhook URL. */
public record SlackAudienceReference(String workspaceBindingReference, String audienceReference)
implements NotificationRecipientReference {
public SlackAudienceReference {
workspaceBindingReference =
NotificationIntentId.requireOpaque(
"Slack workspace binding reference", workspaceBindingReference);
audienceReference =
NotificationIntentId.requireOpaque("Slack audience reference", audienceReference);
}
@Override
public NotificationChannel channel() {
return NotificationChannel.SLACK;
}
@Override
public String toString() {
return "SlackAudienceReference[workspaceBindingReference=<redacted>, "
+ "audienceReference=<redacted>]";
}
}
@@ -0,0 +1,8 @@
package dev.caskeleton.application.notification;
/** Whether one provider call could have produced an external side effect. */
public enum SubmissionCertainty {
DEFINITELY_NOT_APPLIED,
PROVIDER_ACCEPTED,
INDETERMINATE
}
@@ -0,0 +1,130 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.command.Command;
import java.util.Objects;
import java.util.Optional;
/**
* Audited closed writer transition request. No target owner field exists; action determines the
* only legal result owner.
*/
public record SwitchNotificationWriterOwnershipCommand(
Action action,
NotificationCanonicalWriterRouteSet.RouteRevision route,
long expectedGeneration,
long reviewedTargetGeneration,
String operationToken,
String actorReference,
NotificationReasonCode reasonCode,
Optional<SignedNotificationWriterInventoryManifest> inventoryManifest,
Optional<String> quiescenceAttestationToken)
implements Command {
public SwitchNotificationWriterOwnershipCommand {
Objects.requireNonNull(action, "writer ownership action must be non-null");
Objects.requireNonNull(route, "notification writer route must be non-null");
if (expectedGeneration < 0 || reviewedTargetGeneration < 0) {
throw new IllegalArgumentException("writer generations must be non-negative");
}
operationToken =
NotificationIntentId.requireOpaque("writer switch operation token", operationToken);
actorReference = NotificationIntentId.requireOpaque("writer switch actor", actorReference);
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
Objects.requireNonNull(inventoryManifest, "inventory manifest container must be non-null");
Objects.requireNonNull(
quiescenceAttestationToken, "quiescence attestation token container must be non-null");
quiescenceAttestationToken =
quiescenceAttestationToken.map(
token -> NotificationIntentId.requireOpaque("quiescence attestation token", token));
switch (action) {
case BEGIN_DRAIN -> {
if (reviewedTargetGeneration != expectedGeneration
|| inventoryManifest.isEmpty()
|| quiescenceAttestationToken.isPresent()) {
throw new IllegalArgumentException(
"BEGIN_DRAIN keeps generation, requires inventory and forbids attestation token");
}
}
case COMPLETE_SWITCH -> {
if (reviewedTargetGeneration != Math.addExact(expectedGeneration, 1)
|| inventoryManifest.isPresent()) {
throw new IllegalArgumentException(
"COMPLETE_SWITCH requires exactly generation+1 and no caller inventory");
}
}
case ABORT_DRAIN -> {
if (reviewedTargetGeneration != Math.addExact(expectedGeneration, 1)
|| inventoryManifest.isPresent()
|| quiescenceAttestationToken.isPresent()) {
throw new IllegalArgumentException(
"ABORT_DRAIN requires exactly generation+1 and no evidence inputs");
}
}
default -> throw new IllegalArgumentException("unsupported writer ownership action");
}
}
public static SwitchNotificationWriterOwnershipCommand beginDrain(
NotificationCanonicalWriterRouteSet.RouteRevision route,
long expectedGeneration,
String operationToken,
String actorReference,
NotificationReasonCode reasonCode,
SignedNotificationWriterInventoryManifest inventoryManifest) {
return new SwitchNotificationWriterOwnershipCommand(
Action.BEGIN_DRAIN,
route,
expectedGeneration,
expectedGeneration,
operationToken,
actorReference,
reasonCode,
Optional.of(inventoryManifest),
Optional.empty());
}
public static SwitchNotificationWriterOwnershipCommand completeSwitch(
NotificationCanonicalWriterRouteSet.RouteRevision route,
long expectedGeneration,
long reviewedTargetGeneration,
String operationToken,
String actorReference,
NotificationReasonCode reasonCode,
Optional<String> quiescenceAttestationToken) {
return new SwitchNotificationWriterOwnershipCommand(
Action.COMPLETE_SWITCH,
route,
expectedGeneration,
reviewedTargetGeneration,
operationToken,
actorReference,
reasonCode,
Optional.empty(),
quiescenceAttestationToken);
}
public static SwitchNotificationWriterOwnershipCommand abortDrain(
NotificationCanonicalWriterRouteSet.RouteRevision route,
long expectedGeneration,
long reviewedTargetGeneration,
String operationToken,
String actorReference,
NotificationReasonCode reasonCode) {
return new SwitchNotificationWriterOwnershipCommand(
Action.ABORT_DRAIN,
route,
expectedGeneration,
reviewedTargetGeneration,
operationToken,
actorReference,
reasonCode,
Optional.empty(),
Optional.empty());
}
public enum Action {
BEGIN_DRAIN,
COMPLETE_SWITCH,
ABORT_DRAIN
}
}
@@ -0,0 +1,18 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
import java.util.Optional;
/**
* Atomic persistence operation that locks/re-verifies retained evidence and applies only the closed
* LEGACY transition matrix.
*/
@FunctionalInterface
public interface SwitchNotificationWriterOwnershipOperation {
SwitchNotificationWriterOwnershipResult switchOwnership(
SwitchNotificationWriterOwnershipCommand command,
Optional<NotificationWriterInventoryEvidence> verifiedInventory,
NotificationWriterRouteSet.RouteProfile trustedRoute,
Instant requestedAt);
}
@@ -0,0 +1,70 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/** Committed result of one closed writer-fence transition. */
public record SwitchNotificationWriterOwnershipResult(
Status status,
SwitchNotificationWriterOwnershipCommand.Action action,
NotificationCanonicalWriterRouteSet.RouteRevision route,
FenceState state,
NotificationWriterOwnership owner,
long generation,
String operationToken) {
public SwitchNotificationWriterOwnershipResult {
Objects.requireNonNull(status, "writer switch status must be non-null");
Objects.requireNonNull(action, "writer switch action must be non-null");
Objects.requireNonNull(route, "notification writer route must be non-null");
Objects.requireNonNull(state, "writer fence state must be non-null");
Objects.requireNonNull(owner, "notification writer owner must be non-null");
if (generation < 0) {
throw new IllegalArgumentException("writer generation must be non-negative");
}
operationToken =
NotificationIntentId.requireOpaque("writer switch operation token", operationToken);
validateMatrix(action, state, owner);
}
public static SwitchNotificationWriterOwnershipResult applied(
SwitchNotificationWriterOwnershipCommand.Action action,
NotificationCanonicalWriterRouteSet.RouteRevision route,
NotificationWriterOwnership owner,
long generation,
String operationToken) {
FenceState state =
action == SwitchNotificationWriterOwnershipCommand.Action.BEGIN_DRAIN
? FenceState.DRAINING
: FenceState.ACTIVE;
return new SwitchNotificationWriterOwnershipResult(
Status.APPLIED, action, route, state, owner, generation, operationToken);
}
private static void validateMatrix(
SwitchNotificationWriterOwnershipCommand.Action action,
FenceState state,
NotificationWriterOwnership owner) {
boolean valid =
switch (action) {
case BEGIN_DRAIN ->
state == FenceState.DRAINING && owner == NotificationWriterOwnership.LEGACY;
case COMPLETE_SWITCH ->
state == FenceState.ACTIVE && owner == NotificationWriterOwnership.CANONICAL;
case ABORT_DRAIN ->
state == FenceState.ACTIVE && owner == NotificationWriterOwnership.LEGACY;
};
if (!valid) {
throw new IllegalArgumentException("writer switch result violates closed transition matrix");
}
}
public enum Status {
APPLIED,
REPLAYED
}
public enum FenceState {
ACTIVE,
DRAINING
}
}
@@ -0,0 +1,100 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.capability.Idempotency;
import dev.caskeleton.application.capability.RepositoryAccess;
import dev.caskeleton.application.capability.UseCaseCapability;
import dev.caskeleton.application.security.RequiresPermission;
import dev.caskeleton.application.transaction.TransactionMode;
import dev.caskeleton.application.transaction.TransactionPort;
import dev.caskeleton.application.usecase.CommandUseCase;
import java.time.Clock;
import java.util.Objects;
import java.util.Optional;
/**
* Validates signed evidence requirements and root-commits one closed writer ownership transition.
*/
@RequiresPermission("notification:cutover")
@UseCaseCapability(
transactionMode = TransactionMode.WRITE,
idempotency = Idempotency.IDEMPOTENT,
repositoryAccess = RepositoryAccess.WRITE_REPOSITORY,
crossTenantAdmin = true)
public final class SwitchNotificationWriterOwnershipUseCase
implements CommandUseCase<
SwitchNotificationWriterOwnershipCommand, SwitchNotificationWriterOwnershipResult> {
private final NotificationWriterRouteSet routes;
private final NotificationWriterInventoryEvidenceVerifierPort inventoryVerifier;
private final SwitchNotificationWriterOwnershipOperation operation;
private final TransactionPort transactions;
private final Clock clock;
public SwitchNotificationWriterOwnershipUseCase(
NotificationWriterRouteSet routes,
NotificationWriterInventoryEvidenceVerifierPort inventoryVerifier,
SwitchNotificationWriterOwnershipOperation operation,
TransactionPort transactions,
Clock clock) {
this.routes = Objects.requireNonNull(routes, "notification writer route set must be non-null");
this.inventoryVerifier =
Objects.requireNonNull(inventoryVerifier, "writer inventory verifier must be non-null");
this.operation =
Objects.requireNonNull(operation, "writer ownership switch operation must be non-null");
this.transactions = Objects.requireNonNull(transactions, "transaction port must be non-null");
this.clock = Objects.requireNonNull(clock, "clock must be non-null");
}
@Override
public SwitchNotificationWriterOwnershipResult handle(
SwitchNotificationWriterOwnershipCommand command) {
Objects.requireNonNull(command, "writer ownership switch command must be non-null");
NotificationWriterRouteSet.RouteProfile route = routes.requireRoute(command.route());
validateEvidenceRequirement(command, route);
Optional<NotificationWriterInventoryEvidence> inventory = verifyBeginInventory(command);
return transactions.inRootWrite(
() -> operation.switchOwnership(command, inventory, route, clock.instant()));
}
private Optional<NotificationWriterInventoryEvidence> verifyBeginInventory(
SwitchNotificationWriterOwnershipCommand command) {
if (command.action() != SwitchNotificationWriterOwnershipCommand.Action.BEGIN_DRAIN) {
return Optional.empty();
}
SignedNotificationWriterInventoryManifest manifest = command.inventoryManifest().orElseThrow();
if (!manifest.route().equals(command.route())
|| manifest.generation() != command.expectedGeneration()) {
throw new IllegalArgumentException(
"signed writer inventory must match BEGIN route and generation");
}
NotificationWriterInventoryEvidence evidence =
Objects.requireNonNull(
inventoryVerifier.verify(
manifest, command.route(), command.expectedGeneration(), clock.instant()),
"verified writer inventory evidence must be non-null");
if (!evidence.route().equals(command.route())
|| evidence.generation() != command.expectedGeneration()) {
throw new IllegalArgumentException(
"verified writer inventory does not match BEGIN route and generation");
}
return Optional.of(evidence);
}
private static void validateEvidenceRequirement(
SwitchNotificationWriterOwnershipCommand command,
NotificationWriterRouteSet.RouteProfile route) {
if (command.action() != SwitchNotificationWriterOwnershipCommand.Action.COMPLETE_SWITCH) {
return;
}
if (route.proofRequirement() == NotificationWriterRouteSet.ProofClass.QUIESCENCE_REQUIRED
&& command.quiescenceAttestationToken().isEmpty()) {
throw new IllegalArgumentException(
"QUIESCENCE_REQUIRED route requires a committed attestation token");
}
if (route.proofRequirement() == NotificationWriterRouteSet.ProofClass.HARD_BOUND_PROVEN
&& command.quiescenceAttestationToken().isPresent()) {
throw new IllegalArgumentException(
"all-hard-bound route forbids quiescence attestation evidence");
}
}
}
@@ -0,0 +1,25 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/** Bounded target-ordinal outcome for one provider leg. */
public record TargetAttemptOutcome(
int targetOrdinal, NotificationDeliveryId deliveryId, ProviderAttemptOutcome providerOutcome) {
public TargetAttemptOutcome {
if (targetOrdinal < 0 || targetOrdinal >= 16) {
throw new IllegalArgumentException("target ordinal must be in 0..15");
}
Objects.requireNonNull(deliveryId, "notification delivery ID must be non-null");
Objects.requireNonNull(providerOutcome, "provider attempt outcome must be non-null");
}
@Override
public String toString() {
return "TargetAttemptOutcome[targetOrdinal="
+ targetOrdinal
+ ", deliveryId=<redacted>, providerOutcome="
+ providerOutcome
+ "]";
}
}
@@ -0,0 +1,31 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.command.Command;
import java.util.Objects;
/** PRE-only bounded DB-time terminalization request for expired ACTIVE legacy permits. */
public record TerminalizeExpiredNotificationWriterPermitsCommand(
NotificationCanonicalWriterRouteSet.RouteRevision route,
long drainingGeneration,
int maximumPermits,
String operationToken,
String actorReference,
NotificationReasonCode reasonCode)
implements Command {
public TerminalizeExpiredNotificationWriterPermitsCommand {
Objects.requireNonNull(route, "notification writer route must be non-null");
if (drainingGeneration < 0) {
throw new IllegalArgumentException("draining generation must be non-negative");
}
if (maximumPermits < 1 || maximumPermits > 100) {
throw new IllegalArgumentException("terminalization permit bound must be in 1..100");
}
operationToken =
NotificationIntentId.requireOpaque(
"permit terminalization operation token", operationToken);
actorReference =
NotificationIntentId.requireOpaque("permit terminalization actor", actorReference);
Objects.requireNonNull(reasonCode, "notification reason code must be non-null");
}
}
@@ -0,0 +1,13 @@
package dev.caskeleton.application.notification;
import java.time.Instant;
/** Atomic operation selecting and CAS-terminalizing DB-time-expired permits in canonical order. */
@FunctionalInterface
public interface TerminalizeExpiredNotificationWriterPermitsOperation {
TerminalizeExpiredNotificationWriterPermitsResult terminalize(
TerminalizeExpiredNotificationWriterPermitsCommand command,
NotificationWriterRouteSet.RouteProfile trustedRoute,
Instant requestedAt);
}
@@ -0,0 +1,21 @@
package dev.caskeleton.application.notification;
import java.util.Objects;
/** Committed terminalized permit count and canonical affected tuple-set digest. */
public record TerminalizeExpiredNotificationWriterPermitsResult(
Status status, int affectedCount, String affectedSetDigest) {
public TerminalizeExpiredNotificationWriterPermitsResult {
Objects.requireNonNull(status, "permit terminalization status must be non-null");
if (affectedCount < 0 || affectedCount > 100) {
throw new IllegalArgumentException("terminalized permit count must be in 0..100");
}
affectedSetDigest = InitializeNotificationWriterFencesCommand.requireDigest(affectedSetDigest);
}
public enum Status {
APPLIED,
REPLAYED
}
}
@@ -0,0 +1,49 @@
package dev.caskeleton.application.notification;
import dev.caskeleton.application.capability.Idempotency;
import dev.caskeleton.application.capability.RepositoryAccess;
import dev.caskeleton.application.capability.UseCaseCapability;
import dev.caskeleton.application.security.RequiresPermission;
import dev.caskeleton.application.transaction.TransactionMode;
import dev.caskeleton.application.transaction.TransactionPort;
import dev.caskeleton.application.usecase.CommandUseCase;
import java.time.Clock;
import java.util.Objects;
/** Root-commits one bounded expired-permit terminalization operation. */
@RequiresPermission("notification:cutover-terminalize")
@UseCaseCapability(
transactionMode = TransactionMode.WRITE,
idempotency = Idempotency.IDEMPOTENT,
repositoryAccess = RepositoryAccess.WRITE_REPOSITORY,
crossTenantAdmin = true)
public final class TerminalizeExpiredNotificationWriterPermitsUseCase
implements CommandUseCase<
TerminalizeExpiredNotificationWriterPermitsCommand,
TerminalizeExpiredNotificationWriterPermitsResult> {
private final NotificationWriterRouteSet routes;
private final TerminalizeExpiredNotificationWriterPermitsOperation operation;
private final TransactionPort transactions;
private final Clock clock;
public TerminalizeExpiredNotificationWriterPermitsUseCase(
NotificationWriterRouteSet routes,
TerminalizeExpiredNotificationWriterPermitsOperation operation,
TransactionPort transactions,
Clock clock) {
this.routes = Objects.requireNonNull(routes, "notification writer route set must be non-null");
this.operation =
Objects.requireNonNull(operation, "permit terminalization operation must be non-null");
this.transactions = Objects.requireNonNull(transactions, "transaction port must be non-null");
this.clock = Objects.requireNonNull(clock, "clock must be non-null");
}
@Override
public TerminalizeExpiredNotificationWriterPermitsResult handle(
TerminalizeExpiredNotificationWriterPermitsCommand command) {
Objects.requireNonNull(command, "permit terminalization command must be non-null");
NotificationWriterRouteSet.RouteProfile route = routes.requireRoute(command.route());
return transactions.inRootWrite(() -> operation.terminalize(command, route, clock.instant()));
}
}
@@ -0,0 +1,12 @@
package dev.caskeleton.application.transaction;
/**
* Raised when a root-only transaction operation is invoked while an actual transaction is already
* active on the calling thread.
*/
public final class NestedRootTransactionRejectedException extends RuntimeException {
public NestedRootTransactionRejectedException() {
super("root write transaction requires no ambient actual transaction");
}
}

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