diff --git a/modules/kafka-topics-incluster/main.tf b/modules/kafka-topics-incluster/main.tf
new file mode 100644
index 0000000..3438c57
--- /dev/null
+++ b/modules/kafka-topics-incluster/main.tf
@@ -0,0 +1,188 @@
+# In-cluster Kafka implementation (Bitnami/Strimzi etc.) via the Mongey/kafka
+# provider (Kafka Admin API). The topic/user flattening locals mirror
+# modules/kafka-topics-yc 1:1 so the same infrastructure.yaml declaration
+# produces the same logical result; only the backing resources differ:
+#   yandex_mdb_kafka_topic  -> kafka_topic
+#   yandex_mdb_kafka_user   -> kafka_user_scram_credential
+#   (YC role permissions)   -> kafka_acl (roles mapped to ACL operations)
+#
+# NOTE (design open question): Mongey/kafka SCRAM + ACL support must be verified
+# against the target Kafka version before production use.
+
+locals {
+  raw_topics = var.topics
+
+  topics = {
+    for topic in local.raw_topics : topic.name => {
+      name        = topic.name
+      owner       = topic.owner
+      cluster_ref = topic.clusterRef
+      cluster     = var.kafka_cluster_refs[topic.clusterRef]
+
+      partitions = try(
+        tonumber(topic.partitions[var.environment]),
+        try(tonumber(topic.partitions._default), try(tonumber(topic.partitions), var.kafka_cluster_refs[topic.clusterRef].default_partitions))
+      )
+
+      replication_factor = try(
+        tonumber(topic.replicationFactor[var.environment]),
+        try(tonumber(topic.replicationFactor._default), try(tonumber(topic.replicationFactor), var.kafka_cluster_refs[topic.clusterRef].default_replication_factor))
+      )
+
+      config = {
+        for key, value in merge(
+          try(topic.inheritDefaultConfig, true) ? try(var.kafka_cluster_refs[topic.clusterRef].default_topic_config, {}) : {},
+          try(topic.config, {})
+        ) : key => try(tostring(value[var.environment]), try(tostring(value._default), tostring(value)))
+      }
+
+      deletion_policy = try(topic.deletionPolicy, "orphan")
+
+      # Roles granted to the topic owner on its own topic; falls back to the
+      # platform default when ownerRoles is not declared.
+      owner_roles = try(topic.ownerRoles, null) == null ? var.default_user_roles : topic.ownerRoles
+    }
+  }
+
+  # Access grants declared per user under environments.<env>.kafka.users.
+  user_grants = flatten([
+    for user in var.users : [
+      for permission in try(user.permissions, []) : [
+        for role in permission.roles : {
+          cluster_ref = user.clusterRef
+          user        = user.name
+          topic_name  = permission.topic
+          role        = role
+        }
+      ]
+    ]
+  ])
+
+  # Every (cluster, user, topic, role) tuple: owner roles on own topics plus
+  # user-side grants.
+  permission_tuples = concat(
+    flatten([
+      for topic in values(local.topics) : [
+        for role in topic.owner_roles : {
+          cluster_ref = topic.cluster_ref
+          user        = topic.owner
+          topic_name  = topic.name
+          role        = role
+        }
+      ]
+    ]),
+    local.user_grants
+  )
+
+  user_clusters = {
+    for item in distinct(concat(
+      [for tuple in local.permission_tuples : "${tuple.cluster_ref}:${tuple.user}"],
+      [for user in var.users : "${user.clusterRef}:${user.name}"]
+      )) : item => {
+      cluster_ref = split(":", item)[0]
+      user        = split(":", item)[1]
+      cluster     = var.kafka_cluster_refs[split(":", item)[0]]
+    }
+  }
+
+  user_permissions = {
+    for user_key, user in local.user_clusters : user_key => distinct([
+      for tuple in local.permission_tuples : {
+        topic_name = tuple.topic_name
+        role       = tuple.role
+      }
+      if "${tuple.cluster_ref}:${tuple.user}" == user_key
+    ])
+  }
+
+  # Abstract role -> concrete Kafka ACL operations. Kept deliberately small;
+  # extend as the platform role vocabulary grows.
+  role_operations = {
+    ACCESS_ROLE_PRODUCER = [
+      { resource_type = "Topic", operation = "Write" },
+      { resource_type = "Topic", operation = "Describe" },
+    ]
+    ACCESS_ROLE_CONSUMER = [
+      { resource_type = "Topic", operation = "Read" },
+      { resource_type = "Topic", operation = "Describe" },
+      { resource_type = "Group", operation = "Read" },
+    ]
+    ACCESS_ROLE_TOPIC_ADMIN = [
+      { resource_type = "Topic", operation = "All" },
+    ]
+  }
+
+  # Flatten (user, topic, role) permissions into individual ACL entries.
+  acl_entries = merge([
+    for user_key, perms in local.user_permissions : {
+      for e in flatten([
+        for p in perms : [
+          for op in try(local.role_operations[p.role], []) : {
+            key           = "${user_key}:${p.topic_name}:${op.resource_type}:${op.operation}"
+            principal     = "User:${local.user_clusters[user_key].user}"
+            resource_type = op.resource_type
+            resource_name = op.resource_type == "Group" ? "*" : p.topic_name
+            operation     = op.operation
+          }
+        ]
+      ]) : e.key => e
+    }
+  ]...)
+}
+
+resource "random_password" "kafka_user" {
+  for_each = var.create_users ? local.user_clusters : {}
+
+  length  = var.user_password_length
+  special = false
+  upper   = true
+  lower   = true
+  numeric = true
+
+  lifecycle {
+    ignore_changes = all
+  }
+}
+
+resource "kafka_topic" "this" {
+  for_each = local.topics
+
+  name               = each.value.name
+  replication_factor = each.value.replication_factor
+  partitions         = each.value.partitions
+  config             = each.value.config
+
+  lifecycle {
+    prevent_destroy = true
+  }
+}
+
+resource "kafka_user_scram_credential" "this" {
+  for_each = var.create_users ? local.user_clusters : {}
+
+  username         = each.value.user
+  scram_mechanism  = try(each.value.cluster.sasl_mechanism, "SCRAM-SHA-512")
+  scram_iterations = 4096
+  password         = random_password.kafka_user[each.key].result
+
+  lifecycle {
+    prevent_destroy = true
+    ignore_changes  = [password]
+  }
+
+  depends_on = [kafka_topic.this]
+}
+
+resource "kafka_acl" "this" {
+  for_each = var.create_users ? local.acl_entries : {}
+
+  resource_name                = each.value.resource_name
+  resource_type                = each.value.resource_type
+  resource_pattern_type_filter = "Literal"
+  acl_principal                = each.value.principal
+  acl_host                     = "*"
+  acl_operation                = each.value.operation
+  acl_permission_type          = "Allow"
+
+  depends_on = [kafka_user_scram_credential.this]
+}
diff --git a/modules/kafka-topics-incluster/outputs.tf b/modules/kafka-topics-incluster/outputs.tf
new file mode 100644
index 0000000..f52fd80
--- /dev/null
+++ b/modules/kafka-topics-incluster/outputs.tf
@@ -0,0 +1,30 @@
+output "topic_names" {
+  description = "Kafka topics managed by this service manifest state."
+  value       = keys(kafka_topic.this)
+}
+
+output "user_names" {
+  description = "Kafka users managed by this service manifest state."
+  value       = [for u in kafka_user_scram_credential.this : u.username]
+}
+
+# Same shape and key format (clusterRef:user) as modules/kafka-topics-yc so the
+# shared k8s-secret module consumes both backends identically.
+output "kafka_outputs_map" {
+  description = "Map for k8s-secret module. Key format: clusterRef:user."
+  value = {
+    for key, item in local.user_clusters : key => {
+      host                   = try(item.cluster.host, "")
+      hostname               = try(item.cluster.bootstrap_servers, "")
+      port                   = tostring(try(item.cluster.port, 9092))
+      username               = item.user
+      password               = try(random_password.kafka_user[key].result, "")
+      sasl_mechanism         = try(item.cluster.sasl_mechanism, "SCRAM-SHA-512")
+      security_protocol      = try(item.cluster.security_protocol, "SASL_PLAINTEXT")
+      bootstrap_server       = try(item.cluster.bootstrap_servers, "")
+      bootstrap_servers      = try(item.cluster.bootstrap_servers, "")
+      bootstrap_servers_json = try(jsonencode([item.cluster.bootstrap_servers]), "[]")
+    }
+  }
+  sensitive = true
+}
diff --git a/modules/kafka-topics-incluster/variables.tf b/modules/kafka-topics-incluster/variables.tf
new file mode 100644
index 0000000..d2a2657
--- /dev/null
+++ b/modules/kafka-topics-incluster/variables.tf
@@ -0,0 +1,89 @@
+# Entity contract for the in-cluster Kafka implementation.
+#
+# The entity-level variables (topics, users, kafka_policy, create_users,
+# default_user_roles, user_password_length) are BYTE-IDENTICAL to
+# modules/kafka-topics-yc so that a single declaration in infrastructure.yaml
+# drives both the Yandex MDB and the in-cluster (Bitnami/Strimzi) backends.
+#
+# Only `kafka_cluster_refs` differs: instead of a Yandex MDB cluster_id it
+# carries the in-cluster bootstrap endpoint and SASL/TLS parameters. Admin
+# SASL credentials are supplied by the root unit from Vault, never hardcoded.
+
+variable "environment" {
+  description = "Target environment name, for example stage/preprod/<contour>."
+  type        = string
+}
+
+variable "topics" {
+  description = "Kafka topics declared in infrastructure.yaml under environments.<env>.kafka.topics."
+  type = list(object({
+    name                 = string
+    owner                = string
+    clusterRef           = string
+    partitions           = any
+    replicationFactor    = any
+    config               = optional(any, {})
+    inheritDefaultConfig = optional(bool, true)
+    deletionPolicy       = optional(string, "orphan")
+    ownerRoles           = optional(list(string))
+  }))
+  default = []
+}
+
+variable "users" {
+  description = "User-centric Kafka access declarations from infrastructure.yaml under environments.<env>.kafka.users. Each user lists the topics it may access and the roles it holds."
+  type = list(object({
+    name       = string
+    clusterRef = string
+    permissions = optional(list(object({
+      topic = string
+      roles = list(string)
+    })), [])
+  }))
+  default = []
+}
+
+variable "kafka_cluster_refs" {
+  description = "Environment clusterRef mapping. In-cluster shape: bootstrap endpoint + SASL/TLS, no Yandex cluster_id."
+  type = map(object({
+    bootstrap_servers          = string
+    host                       = optional(string, "")
+    port                       = optional(number, 9092)
+    sasl_mechanism             = optional(string, "SCRAM-SHA-512")
+    security_protocol          = optional(string, "SASL_PLAINTEXT")
+    tls_enabled                = optional(bool, false)
+    default_partitions         = optional(number, 3)
+    default_replication_factor = optional(number, 1)
+    max_replication_factor     = optional(number, 1)
+    default_topic_config       = optional(map(string), {})
+  }))
+}
+
+variable "kafka_policy" {
+  description = "Platform policy for service-owned Kafka declarations."
+  type = object({
+    allow_create             = optional(bool, true)
+    allow_delete             = optional(bool, false)
+    allow_partition_increase = optional(bool, true)
+    allow_partition_decrease = optional(bool, false)
+  })
+  default = {}
+}
+
+variable "create_users" {
+  description = "Create one Kafka user per topic owner."
+  type        = bool
+  default     = true
+}
+
+variable "default_user_roles" {
+  description = "Kafka roles granted to generated owner users on their topics."
+  type        = list(string)
+  default     = ["ACCESS_ROLE_PRODUCER", "ACCESS_ROLE_CONSUMER"]
+}
+
+variable "user_password_length" {
+  description = "Generated Kafka user (SCRAM) password length."
+  type        = number
+  default     = 32
+}
diff --git a/modules/kafka-topics-incluster/versions.tf b/modules/kafka-topics-incluster/versions.tf
new file mode 100644
index 0000000..1d366d7
--- /dev/null
+++ b/modules/kafka-topics-incluster/versions.tf
@@ -0,0 +1,14 @@
+terraform {
+  required_version = ">= 1.5.0"
+
+  required_providers {
+    kafka = {
+      source  = "Mongey/kafka"
+      version = ">= 0.7.0"
+    }
+    random = {
+      source  = "hashicorp/random"
+      version = ">= 3.0"
+    }
+  }
+}
diff --git a/modules/minio-buckets/main.tf b/modules/minio-buckets/main.tf
new file mode 100644
index 0000000..c0133e7
--- /dev/null
+++ b/modules/minio-buckets/main.tf
@@ -0,0 +1,68 @@
+# In-cluster MinIO implementation of the `buckets` entity contract. Per bucket:
+# create the bucket, a dedicated access user (access_key/secret_key) and a policy
+# granting that user full access to only its bucket — mirroring yc-s3, which
+# creates a per-bucket service account + static key. Output shape matches yc-s3
+# so the shared k8s-secret module resolves s3 credentials identically.
+
+locals {
+  buckets_map = { for b in var.buckets : b.name => b }
+}
+
+resource "random_password" "secret_key" {
+  for_each = local.buckets_map
+
+  length  = 40
+  special = false
+  upper   = true
+  lower   = true
+  numeric = true
+
+  lifecycle {
+    ignore_changes = all
+  }
+}
+
+resource "minio_s3_bucket" "this" {
+  for_each = local.buckets_map
+
+  bucket = each.key
+  acl    = try(each.value.acl, "private")
+}
+
+resource "minio_s3_bucket_versioning" "this" {
+  for_each = { for name, b in local.buckets_map : name => b if try(b.versioning.enabled, false) }
+
+  bucket = minio_s3_bucket.this[each.key].bucket
+  versioning_configuration {
+    status = "Enabled"
+  }
+}
+
+resource "minio_iam_user" "this" {
+  for_each = local.buckets_map
+
+  name          = "${each.key}${var.user_suffix}"
+  secret        = random_password.secret_key[each.key].result
+  force_destroy = true
+}
+
+resource "minio_iam_policy" "this" {
+  for_each = local.buckets_map
+
+  name = "${each.key}-rw"
+  policy = jsonencode({
+    Version = "2012-10-17"
+    Statement = [{
+      Effect   = "Allow"
+      Action   = ["s3:*"]
+      Resource = ["arn:aws:s3:::${each.key}", "arn:aws:s3:::${each.key}/*"]
+    }]
+  })
+}
+
+resource "minio_iam_user_policy_attachment" "this" {
+  for_each = local.buckets_map
+
+  user_name  = minio_iam_user.this[each.key].name
+  policy_name = minio_iam_policy.this[each.key].name
+}
diff --git a/modules/minio-buckets/outputs.tf b/modules/minio-buckets/outputs.tf
new file mode 100644
index 0000000..f8670b3
--- /dev/null
+++ b/modules/minio-buckets/outputs.tf
@@ -0,0 +1,39 @@
+# Output shape byte-identical to modules/yc-s3 so k8s-secret consumes it unchanged.
+# access_key = the per-bucket MinIO user, secret_key = its generated secret.
+
+output "bucket_name" {
+  description = "Name of the first bucket (for backward compatibility)"
+  value       = length(var.buckets) > 0 ? minio_s3_bucket.this[var.buckets[0].name].bucket : ""
+}
+
+output "access_key" {
+  description = "Access key for the first bucket (for backward compatibility)"
+  value       = length(var.buckets) > 0 ? minio_iam_user.this[var.buckets[0].name].name : ""
+  sensitive   = true
+}
+
+output "secret_key" {
+  description = "Secret key for the first bucket (for backward compatibility)"
+  value       = length(var.buckets) > 0 ? random_password.secret_key[var.buckets[0].name].result : ""
+  sensitive   = true
+}
+
+output "buckets" {
+  description = "Map of all buckets (name => data)"
+  value = {
+    for name, bucket in minio_s3_bucket.this : name => {
+      bucket_name        = bucket.bucket
+      access_key         = minio_iam_user.this[name].name
+      secret_key         = random_password.secret_key[name].result
+      service_account_id = minio_iam_user.this[name].id
+    }
+  }
+  sensitive = true
+}
+
+# Parity with yc-s3 (which exposes extra YC service accounts); none in-cluster.
+output "service_accounts" {
+  description = "Map of additional service accounts (none for MinIO)"
+  value       = {}
+  sensitive   = true
+}
diff --git a/modules/minio-buckets/variables.tf b/modules/minio-buckets/variables.tf
new file mode 100644
index 0000000..8bf6594
--- /dev/null
+++ b/modules/minio-buckets/variables.tf
@@ -0,0 +1,30 @@
+# `buckets` contract is byte-identical to modules/yc-s3 so a single declaration in
+# infrastructure.yaml drives both the Yandex Object Storage and the in-cluster
+# MinIO backend. YC-only fields (versioning/cors) are accepted for schema parity;
+# MinIO applies what it supports (versioning) and ignores the rest.
+
+variable "buckets" {
+  description = "List of S3 buckets to create"
+  type = list(object({
+    name = string
+    acl  = optional(string, "private")
+    role = optional(string, null)
+    versioning = optional(object({
+      enabled = bool
+    }), { enabled = false })
+    cors = optional(object({
+      enabled         = optional(bool, false)
+      allowed_headers = optional(list(string), [])
+      allowed_methods = optional(list(string), [])
+      allowed_origins = optional(list(string), [])
+      expose_headers  = optional(list(string), [])
+      max_age_seconds = optional(number, 3600)
+    }), { enabled = false })
+  }))
+}
+
+variable "user_suffix" {
+  description = "Suffix for the per-bucket MinIO access user (access_key = <bucket><suffix>)."
+  type        = string
+  default     = "-sa"
+}
diff --git a/modules/minio-buckets/versions.tf b/modules/minio-buckets/versions.tf
new file mode 100644
index 0000000..b9aa65d
--- /dev/null
+++ b/modules/minio-buckets/versions.tf
@@ -0,0 +1,14 @@
+terraform {
+  required_version = ">= 1.5.0"
+
+  required_providers {
+    minio = {
+      source  = "aminueza/minio"
+      version = "~> 2.0"
+    }
+    random = {
+      source  = "hashicorp/random"
+      version = "~> 3.1"
+    }
+  }
+}
diff --git a/modules/pg-incluster/main.tf b/modules/pg-incluster/main.tf
new file mode 100644
index 0000000..e73b7f3
--- /dev/null
+++ b/modules/pg-incluster/main.tf
@@ -0,0 +1,79 @@
+# In-cluster PostgreSQL implementation of the `databases` entity contract.
+# Same `databases` list + same output maps as modules/yc-database, so the shared
+# k8s-secret module resolves credentials identically. `cluster_id` is a logical
+# key only (all databases are created on the single server the root provider
+# points at); host/port in outputs come from the database's own host or default_*.
+
+locals {
+  databases_map = { for db in var.databases : "${db.cluster_id}:${db.database.name}:${db.user.name}" => db }
+
+  db_extensions = merge([
+    for key, db in local.databases_map : {
+      for ext in try(db.database.extensions, []) : "${key}:${ext}" => { key = key, ext = ext }
+    }
+  ]...)
+
+  db_grants = merge([
+    for key, db in local.databases_map : {
+      for perm in distinct(compact(try(db.user.permissions, []))) : "${key}:${perm}" => { key = key, db = perm }
+    }
+  ]...)
+}
+
+resource "random_password" "user_password" {
+  for_each = local.databases_map
+
+  length  = try(each.value.user.password_length, 32)
+  special = try(each.value.user.password_special, false)
+  upper   = true
+  lower   = true
+  numeric = true
+
+  lifecycle {
+    ignore_changes = all
+  }
+}
+
+resource "postgresql_role" "this" {
+  for_each = local.databases_map
+
+  name             = each.value.user.name
+  login            = true
+  password         = random_password.user_password[each.key].result
+  connection_limit = try(each.value.user.conn_limit, 10)
+
+  lifecycle {
+    ignore_changes = [password]
+  }
+}
+
+resource "postgresql_database" "this" {
+  for_each = local.databases_map
+
+  name       = each.value.database.name
+  owner      = postgresql_role.this[each.key].name
+  lc_collate = try(each.value.database.lc_collate, "en_US.UTF-8")
+  lc_ctype   = try(each.value.database.lc_type, "en_US.UTF-8")
+  # template0 is required when the collation/ctype differ from the server default.
+  template = "template0"
+
+  depends_on = [postgresql_role.this]
+}
+
+resource "postgresql_extension" "this" {
+  for_each = local.db_extensions
+
+  name     = each.value.ext
+  database = postgresql_database.this[each.value.key].name
+}
+
+# CONNECT grants to OTHER existing databases (declared in user.permissions);
+# the owner already has full access to its own database.
+resource "postgresql_grant" "connect" {
+  for_each = local.db_grants
+
+  role        = postgresql_role.this[each.value.key].name
+  database    = each.value.db
+  object_type = "database"
+  privileges  = ["CONNECT"]
+}
diff --git a/modules/pg-incluster/outputs.tf b/modules/pg-incluster/outputs.tf
new file mode 100644
index 0000000..07b7490
--- /dev/null
+++ b/modules/pg-incluster/outputs.tf
@@ -0,0 +1,30 @@
+# Output shape byte-identical to modules/yc-database so k8s-secret consumes it
+# unchanged. host = the database's own host or the module default (in-cluster svc).
+
+output "databases" {
+  description = "Map of created databases (cluster_id:db_name:user_name => data)"
+  value = {
+    for key, db in local.databases_map : key => {
+      cluster_id    = db.cluster_id
+      host          = try(db.database.host, "") != "" ? db.database.host : var.default_host
+      port          = var.default_port
+      database_name = postgresql_database.this[key].name
+      user_name     = postgresql_role.this[key].name
+      password      = random_password.user_password[key].result
+    }
+  }
+  sensitive = true
+}
+
+output "database_outputs_map" {
+  description = "Map for secrets module (cluster_id:db_name:user_name => credentials)"
+  value = {
+    for key, db in local.databases_map : key => {
+      host          = try(db.database.host, "") != "" ? db.database.host : var.default_host
+      database_name = postgresql_database.this[key].name
+      user_name     = postgresql_role.this[key].name
+      password      = random_password.user_password[key].result
+    }
+  }
+  sensitive = true
+}
diff --git a/modules/pg-incluster/variables.tf b/modules/pg-incluster/variables.tf
new file mode 100644
index 0000000..32a805a
--- /dev/null
+++ b/modules/pg-incluster/variables.tf
@@ -0,0 +1,42 @@
+# Entity contract for the in-cluster PostgreSQL implementation.
+#
+# `databases` is BYTE-IDENTICAL to modules/yc-database so a single declaration
+# in infrastructure.yaml drives both the Yandex MDB and the in-cluster backend.
+# `cluster_id` is kept as a logical key (it also keys database_outputs_map as
+# cluster_id:db:user, which the shared k8s-secret module looks up) — for the
+# in-cluster backend all databases are created on the provider configured by the
+# root unit; default_host/default_port describe that single server.
+
+variable "databases" {
+  description = "List of databases to create"
+  type = list(object({
+    cluster_id = string
+    database = object({
+      name       = string
+      host       = optional(string, "")
+      lc_collate = optional(string, "en_US.UTF-8")
+      lc_type    = optional(string, "en_US.UTF-8")
+      extensions = optional(list(string), [])
+    })
+    user = object({
+      name             = string
+      password_length  = optional(number, 32)
+      password_special = optional(bool, false)
+      conn_limit       = optional(number, 10)
+      permissions      = optional(list(string), [])
+    })
+  }))
+  default = []
+}
+
+variable "default_host" {
+  description = "In-cluster PostgreSQL host to report in outputs when a database does not set its own host (e.g. postgresql.postgresql.svc.cluster.local)."
+  type        = string
+  default     = ""
+}
+
+variable "default_port" {
+  description = "In-cluster PostgreSQL port reported in outputs."
+  type        = number
+  default     = 5432
+}
diff --git a/modules/pg-incluster/versions.tf b/modules/pg-incluster/versions.tf
new file mode 100644
index 0000000..81115e0
--- /dev/null
+++ b/modules/pg-incluster/versions.tf
@@ -0,0 +1,14 @@
+terraform {
+  required_version = ">= 1.5.0"
+
+  required_providers {
+    postgresql = {
+      source  = "cyrilgdn/postgresql"
+      version = "~> 1.21"
+    }
+    random = {
+      source  = "hashicorp/random"
+      version = "~> 3.1"
+    }
+  }
+}