1
/**
2
 * Port for creating new organizations.
3
 * Receives a command with organization details and returns the created entity.
4
 */
5
fun interface CreateOrganizationPort {
6
    /**
7
     * Creates a new organization.
8
     * @param command The organization creation command
9
     * @return The created organization entity with generated ID and default status
10
     * @throws ValidationException if command data is invalid
11
     */
12
    fun execute(command: CreateOrganizationCommand): Organization
13
}

1
// ✅ Good Command
2
data class CreateOrganizationCommand(
3
    val name: String,
4
    val contactEmail: String,
5
    val address: String
6
)
7

8
fun CreateOrganizationCommand.toOrganization(): Organization {
9
    return Organization.create(
10
        id = UUID.randomUUID(),
11
        name = this.name,
12
        contactEmail = this.contactEmail,
13
        address = this.address
14
    )
15
}

1
// ❌ Bad Command
2
data class CreateOrganizationCommand(
3
    val id: UUID = UUID.randomUUID(),  // ❌ System field
4
    var name: String,                   // ❌ Mutable
5
    val createdAt: Instant = Instant.now()  // ❌ System field
6
) {
7
    init {
8
        require(name.isNotBlank()) { "Name required" }  // ❌ Validation here
9
    }
10
}

1
// Regular class (NOT data class): private ctor + factory + transition method
2
// keep every state change going through invariants. data class would expose copy()
3
// and let callers bypass canTransitionTo.
4
class Organization private constructor(
5
    val id: UUID,
6
    val name: OrganizationName,
7
    val status: OrganizationStatus,
8
    val createdAt: Instant,
9
    val updatedAt: Instant,
10
    val version: Long,
11
) {
12
    init { require(version >= 0) { "version must be non-negative" } }
13

14
    fun transitionTo(newStatus: OrganizationStatus, clock: Clock): Organization {
15
        require(status.canTransitionTo(newStatus)) {
16
            "Cannot transition from $status to $newStatus"
17
        }
18
        return Organization(id, name, newStatus, createdAt, clock.instant(), version + 1)
19
    }
20

21
    companion object {
22
        fun create(id: UUID, name: OrganizationName, clock: Clock): Organization {
23
            val now = clock.instant()
24
            return Organization(id, name, OrganizationStatus.PENDING_APPROVAL, now, now, version = 0)
25
        }
26
    }
27
}

1
class CreateOrganizationUseCase(
2
    private val transaction: TransactionPort<Organization>,
3
    private val repository: OrganizationRepositoryPort,
4
    private val outbox: OutboxRepositoryPort,           // SDK
5
    private val events: OutboxEventFactory,             // SDK
6
) : CreateOrganizationPort {
7

8
    companion object {
9
        private val log = LoggerFactory.getLogger(CreateOrganizationUseCase::class.java)
10
    }
11

12
    override fun execute(command: CreateOrganizationCommand): Organization =
13
        transaction.run {
14
            val saved = repository.save(command.toOrganization())
15
            outbox.save(events.create(
16
                topic     = "platform.organizations",
17
                traceId   = command.traceId,
18
                aggregate = saved.id.toString(),
19
                eventName = "OrganizationCreated",
20
                eventType = EventTypes.EVENT,
21
                payload   = OrganizationCreatedPayload.from(saved),
22
            ))
23
            log.info("Created organization: id={}", saved.id)
24
            saved
25
        }
26
}

1
@Component
2
class OrganizationJdbcAdapter(
3
    private val jdbcRepository: OrganizationJdbcRepository
4
) : OrganizationRepositoryPort {
5

6
    private val log = LoggerFactory.getLogger(OrganizationJdbcAdapter::class.java)
7

8
    override fun save(organization: Organization): Organization {
9
        try {
10
            val entity = organization.toDatabaseEntity()
11
            val saved = jdbcRepository.save(entity)
12
            log.debug("Saved organization ${organization.id} to database")
13
            return saved.toDomainModel()
14
        } catch (ex: DataAccessException) {
15
            log.error("Failed to save organization ${organization.id}", ex)
16
            throw PersistenceException("Could not save organization", ex)
17
        }
18
    }
19
}

1
// Before
2
data class Organization(
3
    val contactEmail: String,  // ❌ Primitive obsession
4
    val name: String,          // ❌ No validation
5
    val address: String        // ❌ Unstructured
6
)
7

8
// After — aggregate root: regular class (private ctor + factory), NOT data class
9
class Organization private constructor(
10
    val contactEmail: Email,           // ✅ Value object
11
    val name: OrganizationName,        // ✅ Value object
12
    val address: Address               // ✅ Value object
13
) {
14
    companion object {
15
        fun create(contactEmail: Email, name: OrganizationName, address: Address): Organization =
16
            Organization(contactEmail, name, address)
17
    }
18
}
19

20
// Value objects stay as data class
21
data class Email(val value: String) {
22
    init {
23
        require(value.matches(EMAIL_REGEX)) {
24
            "Invalid email format: $value"
25
        }
26
    }
27

28
    companion object {
29
        private val EMAIL_REGEX = "^[A-Za-z0-9+_.-]+@[A-Za-z0-9.-]+$".toRegex()
30
    }
31
}

1
// Workflow definition
2
enum class WorkflowState {
3
    SUBMITTED,
4
    UNDER_REVIEW,
5
    APPROVED,
6
    REJECTED,
7
    ACTIVE,
8
    SUSPENDED
9
}
10

11
// Aggregate root — the state machine owns its transition rules. NOT a data class:
12
// each `advance(...)` returns a new instance only if the transition is legal.
13
class Workflow private constructor(
14
    val id: UUID,
15
    val organizationId: UUID,
16
    val currentState: WorkflowState,
17
    val history: List<StateTransition>,
18
    val version: Long,
19
) {
20
    init { require(version >= 0) { "version must be non-negative" } }
21

22
    fun advance(target: WorkflowState, reason: String, clock: Clock): Workflow {
23
        require(currentState.canTransitionTo(target)) {
24
            "Illegal workflow transition: $currentState → $target"
25
        }
26
        val transition = StateTransition(currentState, target, reason, clock.instant())
27
        return Workflow(id, organizationId, target, history + transition, version + 1)
28
    }
29

30
    fun isTerminal(): Boolean = currentState == WorkflowState.REJECTED
31

32
    companion object {
33
        fun start(id: UUID, organizationId: UUID, clock: Clock): Workflow =
34
            Workflow(id, organizationId, WorkflowState.SUBMITTED, emptyList(), version = 0)
35
    }
36
}
37

38
// State-transition rules colocated with the enum
39
fun WorkflowState.canTransitionTo(target: WorkflowState): Boolean = when (this) {
40
    WorkflowState.SUBMITTED    -> target == WorkflowState.UNDER_REVIEW
41
    WorkflowState.UNDER_REVIEW -> target == WorkflowState.APPROVED || target == WorkflowState.REJECTED
42
    WorkflowState.APPROVED     -> target == WorkflowState.ACTIVE
43
    WorkflowState.ACTIVE       -> target == WorkflowState.SUSPENDED
44
    WorkflowState.SUSPENDED    -> target == WorkflowState.ACTIVE
45
    WorkflowState.REJECTED     -> false
46
}

1
// Controller versioning
2
@RestController
3
@RequestMapping("/api/v1/organizations")
4
class OrganizationControllerV1 { ... }
5

6
@RestController
7
@RequestMapping("/api/v2/organizations")
8
class OrganizationControllerV2 { ... }
9

10
// Command versioning
11
data class CreateOrganizationCommandV1(...)
12
data class CreateOrganizationCommandV2(...)  // Extended fields
13

14
// Mapper compatibility
15
fun CreateOrganizationCommandV2.toV1(): CreateOrganizationCommandV1 {
16
    return CreateOrganizationCommandV1(
17
        name = this.name,
18
        contactEmail = this.contactEmail
19
        // New fields omitted for V1 compatibility
20
    )
21
}

1
// Saga coordinator
2
class CreateOrganizationSaga(
3
    private val orgService: OrganizationService,
4
    private val userService: UserService,
5
    private val notificationService: NotificationService
6
) {
7
    fun execute(command: CreateOrganizationCommand): SagaResult {
8
        val sagaId = UUID.randomUUID()
9

10
        try {
11
            // Step 1: Create organization
12
            val org = orgService.create(command)
13

14
            // Step 2: Create admin user
15
            val user = userService.createAdmin(org.id)
16

17
            // Step 3: Send notification
18
            notificationService.sendWelcome(user.email)
19

20
            return SagaResult.Success(org)
21
        } catch (ex: Exception) {
22
            // Compensating transactions
23
            compensate(sagaId, ex)
24
            return SagaResult.Failure(ex)
25
        }
26
    }
27

28
    private fun compensate(sagaId: UUID, error: Exception) {
29
        // Rollback steps in reverse order
30
        // Log compensation for audit
31
    }
32
}

1
@Component
2
class PaymentGatewayClient(private val webClient: WebClient) {
3

4
    private val cb = CircuitBreaker.ofDefaults("payment-gateway")
5

6
    fun authorize(req: AuthRequest): AuthResult =
7
        cb.executeSupplier {
8
            webClient.post().uri("/authorize").bodyValue(req)
9
                .retrieve().bodyToMono(AuthResult::class.java)
10
                .block(Duration.ofSeconds(5))
11
                ?: throw PaymentGatewayException("empty response")
12
        }
13
}

1
@Component
2
class KafkaHealthIndicator(
3
    private val kafkaTemplate: KafkaTemplate<String, Any>
4
) : HealthIndicator {
5

6
    override fun health(): Health = try {
7
        // Prefer this — no synthetic message hitting the topic
8
        val producerCount = kafkaTemplate.metrics().size
9
        Health.up().withDetail("kafka.producer.metrics", producerCount).build()
10
    } catch (ex: Exception) {
11
        Health.down(ex).withDetail("kafka", "disconnected").build()
12
    }
13
}

1
// Event store
2
interface EventStore {
3
    fun append(aggregateId: UUID, events: List<DomainEvent>)
4
    fun loadEvents(aggregateId: UUID): List<DomainEvent>
5
}
6

7
// Domain events
8
sealed class OrganizationEvent : DomainEvent {
9
    data class Created(...) : OrganizationEvent()
10
    data class StatusChanged(...) : OrganizationEvent()
11
    data class ContactUpdated(...) : OrganizationEvent()
12
}
13

14
// Rebuild state from events
15
fun loadOrganization(id: UUID, eventStore: EventStore): Organization {
16
    val events = eventStore.loadEvents(id)
17
    return events.fold(Organization.empty()) { org, event ->
18
        org.apply(event)
19
    }
20
}

1
// Write side (commands)
2
interface OrganizationWriteRepositoryPort {
3
    fun save(organization: Organization): Organization
4
}
5

6
// Read side (queries)
7
interface OrganizationReadRepositoryPort {
8
    fun findById(id: UUID): OrganizationReadModel?
9
    fun search(query: SearchQuery): List<OrganizationReadModel>
10
}
11

12
// Projection (keeps read model in sync) — MUST be idempotent (see messaging.md §7)
13
@Component
14
class OrganizationProjection(
15
    private val elasticsearchRepo: ElasticsearchRepository,
16
    private val processed: ProcessedEventPort,           // idempotency-key table
17
) {
18
    @KafkaListener(topics = ["organization-events"])
19
    @Transactional
20
    fun project(@Payload event: EventMessage<OrganizationEvent>) {
21
        val eventId = event.header.traceId      // or a dedicated event id field
22
        if (!processed.tryRecord(consumer = "organization-projection", eventId = eventId)) {
23
            return                                // duplicate — skip silently
24
        }
25
        when (val body = event.body.payload) {
26
            is OrganizationCreated -> elasticsearchRepo.index(body.toReadModel())
27
            is OrganizationUpdated -> elasticsearchRepo.update(body.id, body.toReadModel())
28
        }
29
    }
30
}

1
application/src/main/kotlin/.../port/inbound/
2
├── CreateOrganizationPort.kt
3
├── GetOrganizationByIdPort.kt
4
├── ChangeOrganizationStatusPort.kt
5
└── SendOutboxMessagesPort.kt          # provided by codehuntersio-sdk-event-messaging

1
application/src/main/kotlin/.../port/outbound/
2
├── repository/
3
│   ├── OrganizationRepositoryPort.kt
4
│   ├── UserRepositoryPort.kt
5
│   ├── OutboxRepositoryPort.kt
6
│   └── support/
7
│       ├── TransactionPort.kt
8
│       ├── TransactionCallbackPort.kt
9
│       └── ChainablePort.kt
10
└── # NO publisher/ folder — outbox port + factory provided by codehuntersio-sdk-event-messaging
11
    # (OutboxRepositoryPort, MessagePublisherPort, SendOutboxMessagesPort, OutboxEventFactory)

1
application/src/main/kotlin/.../usecase/
2
└── organization/
3
    ├── CreateOrganizationUseCase.kt
4
    ├── GetOrganizationByIdUseCase.kt
5
    ├── ChangeOrganizationStatusUseCase.kt
6
    └── command/
7
        ├── CreateOrganizationCommand.kt
8
        └── mapper/
9
            └── CommandMapper.kt

1
domain/src/main/kotlin/.../domain/
2
├── Organization.kt
3
├── User.kt
4
├── enums/
5
│   ├── OrganizationStatus.kt
6
│   └── Roles.kt
7
├── exception/
8
│   └── OrganizationNotFoundException.kt
9
└── util/
10
    ├── MdcUtil.kt
11
    └── TraceContextUtil.kt

1
infrastructure/entry-points/
2
├── rest-controller/
3
│   └── .../controller/organization/
4
│       ├── OrganizationController.kt
5
│       └── model/
6
│           ├── request/
7
│           │   └── CreateOrganizationRequest.kt
8
│           └── response/
9
│               ├── CreatedOrganizationResponse.kt
10
│               └── OrganizationResponse.kt
11
└── scheduler-task/
12
    └── OutboxTaskScheduler.kt

1
infrastructure/driven-adapters/
2
├── jdbc-repository/
3
│   ├── OrganizationRepositoryAdapter.kt
4
│   └── repository/transaction/
5
│       └── JdbcTransaction.kt
6
├── kafka-producer/
7
│   ├── OrganizationKafkaPublisherAdapter.kt
8
│   └── publisher/properties/
9
│       └── KafKaTopicProperties.kt
10
├── redis-cache/
11
│   └── OrganizationCacheAdapter.kt
12
└── metrics-publisher/
13
    └── OrganizationMetricsAdapter.kt