Plan cleanup work instead of letting clients derive it

server/src/main/kotlin/com/kakao/actionbase/server/api/control/cleanup/ControlJobController.kt

@@ -0,0 +1,36 @@
+package com.kakao.actionbase.server.api.control.cleanup
+
+import com.kakao.actionbase.server.configuration.ConditionalOnControlRole
+import com.kakao.actionbase.server.configuration.HttpHeaderConstants
+import com.kakao.actionbase.server.control.cleanup.CleanupJobService
+import com.kakao.actionbase.server.control.cleanup.JobRequest
+import com.kakao.actionbase.server.control.cleanup.JobView
+import com.kakao.actionbase.server.control.cluster.CallerHeaders
+
+import org.springframework.http.HttpHeaders
+import org.springframework.web.bind.annotation.PostMapping
+import org.springframework.web.bind.annotation.RequestBody
+import org.springframework.web.bind.annotation.RequestHeader
+import org.springframework.web.bind.annotation.RestController
+
+import reactor.core.publisher.Mono
+
+/**
+ * Cleanup work, planned. `dryRun=true` is the only mode for now and returns the steps the action
+ * would take, or the precondition that stops it.
+ *
+ * Execution will be this same endpoint with `dryRun=false`, so that the plan an operator reviewed is
+ * the plan that runs - two code paths computing the same thing is how they come to disagree.
+ */
+@RestController
+@ConditionalOnControlRole
+class ControlJobController(
+    private val jobService: CleanupJobService,
+) {
+    @PostMapping("/control/jobs")
+    fun submit(
+        @RequestBody request: JobRequest,
+        @RequestHeader(value = HttpHeaders.AUTHORIZATION, required = false) authorization: String?,
+        @RequestHeader(value = HttpHeaderConstants.ACTOR_ROLE, required = false) actorRole: String?,
+    ): Mono<JobView> = jobService.submit(request, CallerHeaders.forwarded(authorization, actorRole))
+}

server/src/main/kotlin/com/kakao/actionbase/server/control/cleanup/CleanupJobService.kt

@@ -0,0 +1,72 @@
+package com.kakao.actionbase.server.control.cleanup
+
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+/**
+ * Plans a batch of cleanup work.
+ *
+ * Targets arrive as a list on purpose. A per-table endpoint is what makes every client build its own
+ * queue, and then build it again for the next client.
+ *
+ * Only dry runs are answered for now: execution waits for authorization, so that a cross-cluster
+ * drop never becomes reachable before there is something checking who asked for it.
+ */
+class CleanupJobService(
+    private val cleanupService: CleanupService,
+) {
+    fun submit(
+        request: JobRequest,
+        headers: Map<String, String>,
+    ): Mono<JobView> {
+        val action = CleanupAction.of(request.action)
+        require(request.targets.isNotEmpty()) { "targets is empty: name at least one table to plan for" }
+        if (!request.dryRun) {
+            throw UnsupportedOperationException(
+                "execution is not available yet, only dryRun=true. The plan this returns is what execution will run.",
+            )
+        }
+
+        return Flux
+            .fromIterable(request.targets.map { it.tenant }.distinct())
+            .flatMapSequential { cleanupService.htables(it, null, headers) }
+            .collectList()
+            .map { views -> view(action, request, views) }
+    }
+
+    private fun view(
+        action: CleanupAction,
+        request: JobRequest,
+        views: List<HtablesView>,
+    ): JobView {
+        val tables = views.flatMap { it.tables }.associateBy { it.tenant to it.name }
+
+        return JobView(
+            dryRun = true,
+            action = action,
+            fanout = request.fanout,
+            plans =
+                request.targets.distinct().map { target ->
+                    tables[target.tenant to target.table]
+                        ?.let { CleanupPlanner.plan(action, it, request.fanout) }
+                        ?: unseen(target, action)
+                },
+            failures = views.flatMap { it.failures },
+        )
+    }
+
+    /**
+     * A table nobody reported. It may be gone, or its cluster may be the one that did not answer -
+     * which is why the failures travel in the same response.
+     */
+    private fun unseen(
+        target: JobTarget,
+        action: CleanupAction,
+    ) = CleanupPlan(
+        target = target,
+        action = action,
+        ok = false,
+        refusal = Precondition.TABLE_ALREADY_GONE,
+        detail = "not seen on any side, and nothing binds it - check failures before concluding it is gone",
+    )
+}

server/src/main/kotlin/com/kakao/actionbase/server/control/cleanup/CleanupPlan.kt

@@ -0,0 +1,140 @@
+package com.kakao.actionbase.server.control.cleanup
+
+import com.kakao.actionbase.server.control.topology.Side
+
+/**
+ * The lifecycle of a datastore table, as operators run it.
+ *
+ * Teardown goes `REPLICATION_OFF -> UNLINK -> DISABLE_TABLE -> DROP_TABLE -> DELETE_METADATA`;
+ * setup goes `ENABLE_TABLE -> REPLICATION_ON`. The order is not a style preference - disabling a
+ * table while replication is still on breaks standby sync, and dropping one a collection still
+ * points at leaves the metadata dangling.
+ *
+ * Per #470 `table` is the physical HBase table and `collection` the logical v3 entity, so a name
+ * carrying `TABLE` never means metadata.
+ */
+enum class CleanupAction {
+    /** Replication scope 1 -> 0. */
+    REPLICATION_OFF,
+
+    /** Replication scope 0 -> 1. */
+    REPLICATION_ON,
+
+    /** Deactivate the collection and storage bound to the table. */
+    UNLINK,
+
+    DISABLE_TABLE,
+
+    ENABLE_TABLE,
+
+    DROP_TABLE,
+
+    /** Delete the metadata left behind once the table is gone. */
+    DELETE_METADATA,
+    ;
+
+    /** Metadata is not replicated, so the actions that touch it only ever run on the active side. */
+    val metadataOnly: Boolean get() = this == UNLINK || this == DELETE_METADATA
+
+    /**
+     * Whether the action takes something that cannot be put back: [DROP_TABLE] loses the rows,
+     * [DELETE_METADATA] the definition. Everything else is a staging step an operator can undo,
+     * which is the line a confirmation prompt belongs on.
+     */
+    val destructive: Boolean get() = this == DROP_TABLE || this == DELETE_METADATA
+
+    companion object {
+        fun of(value: String): CleanupAction =
+            entries.firstOrNull { it.name.equals(value, ignoreCase = true) }
+                ?: throw IllegalArgumentException("unknown action '$value', expected one of ${entries.map { it.name.lowercase() }}")
+    }
+}
+
+/** A single call the executor will make. */
+enum class StepOp {
+    REPLICATION_DISABLE,
+    REPLICATION_ENABLE,
+    DEACTIVATE_COLLECTION,
+    DEACTIVATE_STORAGE,
+    DISABLE_TABLE,
+    ENABLE_TABLE,
+    DROP_TABLE,
+    DELETE_COLLECTION,
+    DELETE_STORAGE,
+}
+
+/**
+ * Why a plan is empty. Typed rather than prose: an operator staring at a disabled button needs to
+ * know which precondition to satisfy, and a test needs to assert that it named the right one.
+ */
+enum class Precondition {
+    /** Every side is already where the action would take it. */
+    ALREADY_SATISFIED,
+
+    /** `DISABLE_TABLE` and `DROP_TABLE` need replication off first, or the standby stops tracking. */
+    NEEDS_LOCAL_REPLICATION,
+
+    /** `REPLICATION_ON` needs the table enabled first. */
+    NEEDS_ENABLED,
+
+    /** `DROP_TABLE` needs the metadata unlinked first. */
+    NEEDS_UNLINK,
+
+    /** `DROP_TABLE` needs the table disabled first. */
+    NEEDS_DISABLE,
+
+    /** `DELETE_METADATA` is for what is already gone. */
+    TABLE_STILL_EXISTS,
+
+    /** `DROP_TABLE` is for what still exists. */
+    TABLE_ALREADY_GONE,
+
+    /** Another tenant's metadata still points at this table. */
+    HELD_BY_ANOTHER_TENANT,
+
+    /** Nothing binds the table, so there is nothing to unlink or purge. */
+    NOTHING_BOUND,
+}
+
+data class JobTarget(
+    val tenant: String,
+    val table: String,
+)
+
+data class PlanStep(
+    val side: Side,
+    val op: StepOp,
+    /** The metadata entity a metadata step acts on. */
+    val name: String? = null,
+)
+
+/**
+ * What would happen, or why nothing would.
+ *
+ * A refusal is a *prediction*: the data plane enforces these invariants itself and remains the one
+ * that says no. Repeating the checks here is how a client learns before it asks, not a second source
+ * of truth.
+ */
+data class CleanupPlan(
+    val target: JobTarget,
+    val action: CleanupAction,
+    val ok: Boolean,
+    val steps: List<PlanStep> = emptyList(),
+    val refusal: Precondition? = null,
+    val detail: String? = null,
+)
+
+data class JobRequest(
+    val targets: List<JobTarget> = emptyList(),
+    val action: String,
+    val fanout: Boolean = false,
+    val dryRun: Boolean = true,
+)
+
+data class JobView(
+    val dryRun: Boolean,
+    val action: CleanupAction,
+    val fanout: Boolean,
+    val plans: List<CleanupPlan>,
+    val failures: List<SideFailure>,
+)