package com.octaldata.endpoints

import com.octaldata.domain.BrokerId
import com.octaldata.domain.DeploymentId
import com.octaldata.domain.integrations.IntegrationId
import com.octaldata.domain.metrics.MetricCollectorId
import com.octaldata.domain.registry.RegistryId
import com.octaldata.domain.schemas.SchemaSubject
import com.octaldata.domain.schemas.SchemaVersion
import com.octaldata.domain.topics.ConsumerGroupId
import com.octaldata.domain.topics.MemberId
import com.octaldata.domain.topics.NamespaceName
import com.octaldata.domain.topics.PartitionId
import com.octaldata.domain.topics.TenantName
import com.octaldata.streamops.ktor.extractPathParameter
import io.ktor.application.ApplicationCall
import io.ktor.util.pipeline.PipelineContext
import mu.KotlinLogging

val logger = KotlinLogging.logger { }

suspend fun PipelineContext<*, ApplicationCall>.withNamespaceName(f: suspend (NamespaceName) -> Unit) {
   extractPathParameter("namespaceName", ::NamespaceName, f)
}

suspend fun PipelineContext<*, ApplicationCall>.withTenantName(f: suspend (TenantName) -> Unit) {
   extractPathParameter("tenantName", ::TenantName, f)
}

suspend fun PipelineContext<*, ApplicationCall>.withPartitionId(f: suspend (PartitionId) -> Unit) {
   extractPathParameter("partitionId", ::PartitionId, f)
}

suspend fun PipelineContext<*, ApplicationCall>.withConsumerGroupId(f: suspend (ConsumerGroupId) -> Unit) {
   extractPathParameter("consumerGroupId", ::ConsumerGroupId, f)
}

suspend fun PipelineContext<*, ApplicationCall>.withMemberId(f: suspend (MemberId) -> Unit) {
   extractPathParameter("memberId", ::MemberId, f)
}

suspend fun PipelineContext<*, ApplicationCall>.withRegistryId(f: suspend (RegistryId) -> Unit) {
   extractPathParameter("registryId", ::RegistryId, f)
}

suspend fun PipelineContext<*, ApplicationCall>.withSchemaSubject(f: suspend (SchemaSubject) -> Unit) {
   extractPathParameter("subject", ::SchemaSubject, f)
}

suspend fun PipelineContext<*, ApplicationCall>.withSchemaVersion(f: suspend (SchemaVersion) -> Unit) {
   extractPathParameter("version", ::SchemaVersion, f)
}

suspend fun PipelineContext<*, ApplicationCall>.withMetricCollectorId(f: suspend (MetricCollectorId) -> Unit) {
   extractPathParameter("metricCollectorId", ::MetricCollectorId, f)
}

suspend fun PipelineContext<*, ApplicationCall>.withIntegrationId(f: suspend (IntegrationId) -> Unit) {
   extractPathParameter("integrationId", ::IntegrationId, f)
}

/**
 * Extracts the [DeploymentId] from this request, or returns a 500,
 * and then invokes the given function with that cluster id.
 */
suspend fun PipelineContext<*, ApplicationCall>.withBrokerId(f: suspend (BrokerId) -> Unit) {
   extractPathParameter("brokerId", ::BrokerId, f)
}