diff --git a/poetry.lock b/poetry.lock
index b56e70d7f0..0438848b16 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -54,7 +54,7 @@ version = "0.7.0"
 description = "Reusable constraint types to use with typing.Annotated"
 optional = false
 python-versions = ">=3.8"
-groups = ["main", "charm-libs"]
+groups = ["main", "charm-libs", "integration"]
 files = [
     {file = "annotated_types-0.7.0-py3-none-any.whl", hash = "sha256:1f02e8b43a8fbbc3f3e0d4f0f4bfc8131bcb4eebe8849b8e5c773f3a1c582a53"},
     {file = "annotated_types-0.7.0.tar.gz", hash = "sha256:aff07c09a53a08bc8cfccb9c85b05f1aa9a2a6f23728d790723543408344ce89"},
@@ -66,7 +66,7 @@ version = "4.14.1"
 description = "High-level concurrency and networking framework on top of asyncio or Trio"
 optional = false
 python-versions = ">=3.10"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "anyio-4.14.1-py3-none-any.whl", hash = "sha256:4e5533c5b8ff0a24f5d7a176cbe6877129cd183893f66b537f8f227d10527d72"},
     {file = "anyio-4.14.1.tar.gz", hash = "sha256:8d648a3544c1a700e3ff78615cd679e4c5c3f149904287e73687b2596963629e"},
@@ -413,7 +413,7 @@ version = "0.1.3"
 description = ""
 optional = false
 python-versions = ">=3.8"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "charm_api-0.1.3-py3-none-any.whl", hash = "sha256:4dba74e33f3de4608e88db2e08e41b0681221e7d3c2cd6913e0ebeb3dbdf8d92"},
     {file = "charm_api-0.1.3.tar.gz", hash = "sha256:a90a926b89fce834e9fb2184b96e07ec46f4dd56f3cd50e0209c0fd0ff35a29e"},
@@ -425,7 +425,7 @@ version = "0.1.2"
 description = ""
 optional = false
 python-versions = ">=3.8"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "charm_json-0.1.2-py3-none-any.whl", hash = "sha256:1cad028a92831c61d6dcb99127c5216c333f90ec58d06b9f73fd0fd7355d1b3c"},
     {file = "charm_json-0.1.2.tar.gz", hash = "sha256:f91255858cd6721ec491d09450f37a0cdecf1addbfcfbd41f3e27c58dd95092e"},
@@ -440,7 +440,7 @@ version = "3.1.1.4"
 description = "In-place rolling refreshes (upgrades) of stateful charmed applications"
 optional = false
 python-versions = ">=3.8"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "charm_refresh-3.1.1.4-py3-none-any.whl", hash = "sha256:6ff6e04170c03b3fe3f23bbf32a288a0de3da0d44bf2e8dd850ee02925bf2ec6"},
     {file = "charm_refresh-3.1.1.4.tar.gz", hash = "sha256:d33f66e5b65fd356a30bfbc634710e5538086199da93c30c86246370a66570be"},
@@ -478,7 +478,7 @@ version = "1.8.3"
 description = "The charmlibs.interfaces.tls_certificates package."
 optional = false
 python-versions = ">=3.10"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "charmlibs_interfaces_tls_certificates-1.8.3-py3-none-any.whl", hash = "sha256:fa0aa150b4e4513d439c7afb7eb0ceaf00363671143b27b09382758deea8a0cc"},
     {file = "charmlibs_interfaces_tls_certificates-1.8.3.tar.gz", hash = "sha256:6bc35ee330681f2e0c58cf67e41133b144f3281f73d27293714c9ca918ee7cbc"},
@@ -495,7 +495,7 @@ version = "1.3.0.post0"
 description = "A pathlib-like interface for Juju K8s charms to interact with files in their workload container."
 optional = false
 python-versions = ">=3.10"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "charmlibs_pathops-1.3.0.post0-py3-none-any.whl", hash = "sha256:c6fc3e3460ffa03a7d4273cc7784c9d74378609a098c728cd9bbb2ed90bdfa5b"},
     {file = "charmlibs_pathops-1.3.0.post0.tar.gz", hash = "sha256:36ac11b0c7c55b71341a549a1803f5c8d61c2279204862b8d50736041b782f59"},
@@ -510,7 +510,7 @@ version = "1.1.1"
 description = "The charmlibs.rollingops package."
 optional = false
 python-versions = ">=3.12"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "charmlibs_rollingops-1.1.1-py3-none-any.whl", hash = "sha256:1e61347c7792ec11afefa60d36f77a5afccf867832854d5aeef2da0b536afec4"},
     {file = "charmlibs_rollingops-1.1.1.tar.gz", hash = "sha256:831457d2ef12117794f89a483ee9a5c347cb209a87dafe82907e9dede7d4641c"},
@@ -531,7 +531,7 @@ version = "1.0.1.post0"
 description = "The charmlibs.snap package."
 optional = false
 python-versions = ">=3.10"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "charmlibs_snap-1.0.1.post0-py3-none-any.whl", hash = "sha256:655041096952c63e92c8d25a6183149f0594d4888be4be62a9635cb96897aa01"},
     {file = "charmlibs_snap-1.0.1.post0.tar.gz", hash = "sha256:1e8da752ee405aa981665da8c4772016236baf80d2d902b02ec8c1c640fb2844"},
@@ -546,7 +546,7 @@ version = "1.0.0.post0"
 description = "The charmlibs.systemd package."
 optional = false
 python-versions = ">=3.10"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "charmlibs_systemd-1.0.0.post0-py3-none-any.whl", hash = "sha256:9e7578b98344ab237fdc6e48491f98fd80239b8802feb72c5c0ff3a9f70019da"},
     {file = "charmlibs_systemd-1.0.0.post0.tar.gz", hash = "sha256:0f978a103524fe1ef34c350930b29b45677ae5c175368e86cea7a1530ccaa0b5"},
@@ -880,7 +880,7 @@ version = "1.1.0"
 description = ""
 optional = false
 python-versions = "<4.0,>=3.10"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "data_platform_helpers-1.1.0-py3-none-any.whl", hash = "sha256:71b2c2ba31d1bc01c55c9da82d02b0cb654b99231172916722ecaf4c469d27c4"},
     {file = "data_platform_helpers-1.1.0.tar.gz", hash = "sha256:7539ef1c1047fec36ee8298aa4c75fe2bb36419544193b691805c12ca407a797"},
@@ -913,7 +913,7 @@ version = "1.0.0"
 description = "The dpcharmlibs.interfaces package."
 optional = false
 python-versions = ">=3.10"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "dpcharmlibs_interfaces-1.0.0-py3-none-any.whl", hash = "sha256:afff80be30e3ff8c31f68557946e9d5959247f1973cbc00e96da55b095f95a21"},
     {file = "dpcharmlibs_interfaces-1.0.0.tar.gz", hash = "sha256:a177653019781a7a165be52cf4293134ee58912cd93686082cdd82da347a3bed"},
@@ -987,7 +987,7 @@ version = "0.16.0"
 description = "A pure-Python, bring-your-own-I/O implementation of HTTP/1.1"
 optional = false
 python-versions = ">=3.8"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86"},
     {file = "h11-0.16.0.tar.gz", hash = "sha256:4e35b956cf45792e4caa5885e69fba00bdbc6ffafbfa020300e549b208ee5ff1"},
@@ -999,7 +999,7 @@ version = "4.3.0"
 description = "Pure-Python HTTP/2 protocol implementation"
 optional = false
 python-versions = ">=3.9"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "h2-4.3.0-py3-none-any.whl", hash = "sha256:c438f029a25f7945c69e0ccf0fb951dc3f73a5f6412981daee861431b70e2bdd"},
     {file = "h2-4.3.0.tar.gz", hash = "sha256:6c59efe4323fa18b47a632221a1888bd7fde6249819beda254aeca909f221bf1"},
@@ -1015,7 +1015,7 @@ version = "4.2.0"
 description = "Pure-Python HPACK header encoding"
 optional = false
 python-versions = ">=3.10"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "hpack-4.2.0-py3-none-any.whl", hash = "sha256:858ac0b02280fa582b5080d68db0899c62a80375e0e5413a74970c5e518b6986"},
     {file = "hpack-4.2.0.tar.gz", hash = "sha256:0895cfa3b5531fc65fe439c05eb65144f123bf7a394fcaa56aa423548d8e45c0"},
@@ -1027,7 +1027,7 @@ version = "1.0.9"
 description = "A minimal low-level HTTP client."
 optional = false
 python-versions = ">=3.8"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55"},
     {file = "httpcore-1.0.9.tar.gz", hash = "sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8"},
@@ -1049,7 +1049,7 @@ version = "0.28.1"
 description = "The next generation HTTP client."
 optional = false
 python-versions = ">=3.8"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad"},
     {file = "httpx-0.28.1.tar.gz", hash = "sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc"},
@@ -1075,7 +1075,7 @@ version = "0.9.0"
 description = "WebSockets support for HTTPX"
 optional = false
 python-versions = ">=3.10"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "httpx_ws-0.9.0-py3-none-any.whl", hash = "sha256:71640d2fb1bf9a225775015b33cd755cfd4c5f7e21c885192fe3adc4c387b248"},
     {file = "httpx_ws-0.9.0.tar.gz", hash = "sha256:797373326f70eec1ae96f6e43ae9f12002fd7d73aee139a4985eaab964338a08"},
@@ -1111,7 +1111,7 @@ version = "6.1.0"
 description = "Pure-Python HTTP/2 framing"
 optional = false
 python-versions = ">=3.9"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "hyperframe-6.1.0-py3-none-any.whl", hash = "sha256:b03380493a519fce58ea5af42e4a42317bf9bd425596f7a0835ffce80f1a42e5"},
     {file = "hyperframe-6.1.0.tar.gz", hash = "sha256:f630908a00854a7adeabd6382b43923a4c4cd4b821fcb527e6ab9e15382a3b08"},
@@ -1399,7 +1399,7 @@ version = "0.21.0"
 description = "Lightweight kubernetes client library"
 optional = false
 python-versions = ">=3.8"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "lightkube-0.21.0-py3-none-any.whl", hash = "sha256:5218c9a7029b5ba8a3e90534141795c6bb3ec3e96894afccb55f0581ca8d59b2"},
     {file = "lightkube-0.21.0.tar.gz", hash = "sha256:720910200de4bd88e0d7b5d2edfb45a4f6be292798b8776d7b551ef5a5ccdc15"},
@@ -1420,7 +1420,7 @@ version = "1.35.0.8"
 description = "Models and Resources for lightkube module"
 optional = false
 python-versions = ">=3.9"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "lightkube_models-1.35.0.8-py3-none-any.whl", hash = "sha256:d01fce42f96baf47a77a571bff59d6a513e96ae043fc03cfaaaaf79c609c4441"},
     {file = "lightkube_models-1.35.0.8.tar.gz", hash = "sha256:dbc624596a7d94e6c43c5deda972be964202e0e8f26e2ab8e61d589d710b5e22"},
@@ -1471,7 +1471,7 @@ version = "4.2.0"
 description = "Python port of markdown-it. Markdown parsing, done right!"
 optional = false
 python-versions = ">=3.10"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "markdown_it_py-4.2.0-py3-none-any.whl", hash = "sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a"},
     {file = "markdown_it_py-4.2.0.tar.gz", hash = "sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49"},
@@ -1612,7 +1612,7 @@ version = "0.1.2"
 description = "Markdown URL utilities"
 optional = false
 python-versions = ">=3.7"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8"},
     {file = "mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba"},
@@ -1845,38 +1845,42 @@ testing = ["coverage", "pytest", "pytest-benchmark"]
 
 [[package]]
 name = "postgresql-charms-single-kernel"
-version = "16.3.2"
+version = "16.3.3"
 description = "Shared and reusable code for PostgreSQL-related charms"
 optional = false
-python-versions = "<4.0,>=3.8"
+python-versions = ">=3.8,<4.0"
 groups = ["main", "integration"]
 files = [
-    {file = "postgresql_charms_single_kernel-16.3.2-py3-none-any.whl", hash = "sha256:919773efde865ec9b8142b50e99381229c2c4d2baea5f7bc1a8e78b926fc775d"},
-    {file = "postgresql_charms_single_kernel-16.3.2.tar.gz", hash = "sha256:92cad3c11e0037d9c451d5c9552693f04a7d1a234d790bd90fc5a37bbe02f8ea"},
+    {file = "40752dd71ffa1db9c15bf4a8c4f1dc0de7bf72b9.tar.gz", hash = "sha256:dd1e52f7648422a77418fcce0197160133132032d8c5d564945ce17415e72dd3"},
 ]
 
 [package.dependencies]
-charm-refresh = {version = "*", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-charmlibs-interfaces-tls-certificates = {version = ">=1.8.3", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-charmlibs-pathops = {version = ">=1.0.1", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-charmlibs-rollingops = {version = ">=1.1.1", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-charmlibs-snap = {version = ">=1.0.1", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"vm\""}
-charmlibs-systemd = {version = ">=1.0.0.post0", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"vm\""}
-data-platform-helpers = {version = ">=0.1.7", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-httpx = {version = "*", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-jinja2 = {version = ">=3.1.6", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
+charm-refresh = {version = "*", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+charmlibs-interfaces-tls-certificates = {version = ">=1.8.3", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+charmlibs-pathops = {version = ">=1.0.1", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+charmlibs-rollingops = {version = ">=1.1.1", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+charmlibs-snap = {version = ">=1.0.1", optional = true, markers = "python_version >= \"3.12\" and extra == \"vm\""}
+charmlibs-systemd = {version = ">=1.0.0.post0", optional = true, markers = "python_version >= \"3.12\" and extra == \"vm\""}
+data-platform-helpers = {version = ">=0.1.7", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+httpx = {version = "*", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+jinja2 = {version = ">=3.1.6", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
 ops = ">=2.0.0"
-psutil = {version = ">=7.2.2", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"vm\""}
-pydantic = {version = ">=2.0", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-pysyncobj = {version = ">=0.3.15", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"vm\""}
-requests = {version = "*", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
+psutil = {version = ">=7.2.2", optional = true, markers = "python_version >= \"3.12\" and extra == \"vm\""}
+pydantic = {version = ">=2.0", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+pysyncobj = {version = ">=0.3.15", optional = true, markers = "python_version >= \"3.12\" and extra == \"vm\""}
+requests = {version = "*", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
 tenacity = ">=9.0.0"
-tomli = {version = "*", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
+tomli = {version = "*", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
 
 [package.extras]
 db-driver = ["psycopg2 (>=2.9.10)"]
-postgresql = ["charm-refresh ; python_full_version >= \"3.12.0\"", "charmlibs-interfaces-tls-certificates (>=1.8.3) ; python_full_version >= \"3.12.0\"", "charmlibs-pathops (>=1.0.1) ; python_full_version >= \"3.12.0\"", "charmlibs-rollingops (>=1.1.1) ; python_full_version >= \"3.12.0\"", "data-platform-helpers (>=0.1.7) ; python_full_version >= \"3.12.0\"", "httpx ; python_full_version >= \"3.12.0\"", "jinja2 (>=3.1.6) ; python_full_version >= \"3.12.0\"", "pydantic (>=2.0) ; python_full_version >= \"3.12.0\"", "requests ; python_full_version >= \"3.12.0\"", "tomli ; python_full_version >= \"3.12.0\""]
-vm = ["charmlibs-snap (>=1.0.1) ; python_full_version >= \"3.12.0\"", "charmlibs-systemd (>=1.0.0.post0) ; python_full_version >= \"3.12.0\"", "psutil (>=7.2.2) ; python_full_version >= \"3.12.0\"", "pysyncobj (>=0.3.15) ; python_full_version >= \"3.12.0\""]
+k8s = ["lightkube (>=0.21.0) ; python_version >= \"3.12\"", "lightkube-models (>=1.28.1.4) ; python_version >= \"3.12\""]
+postgresql = ["charm-refresh ; python_version >= \"3.12\"", "charmlibs-interfaces-tls-certificates (>=1.8.3) ; python_version >= \"3.12\"", "charmlibs-pathops (>=1.0.1) ; python_version >= \"3.12\"", "charmlibs-rollingops (>=1.1.1) ; python_version >= \"3.12\"", "data-platform-helpers (>=0.1.7) ; python_version >= \"3.12\"", "httpx ; python_version >= \"3.12\"", "jinja2 (>=3.1.6) ; python_version >= \"3.12\"", "pydantic (>=2.0) ; python_version >= \"3.12\"", "requests ; python_version >= \"3.12\"", "tomli ; python_version >= \"3.12\""]
+vm = ["charmlibs-snap (>=1.0.1) ; python_version >= \"3.12\"", "charmlibs-systemd (>=1.0.0.post0) ; python_version >= \"3.12\"", "psutil (>=7.2.2) ; python_version >= \"3.12\"", "pysyncobj (>=0.3.15) ; python_version >= \"3.12\""]
+
+[package.source]
+type = "url"
+url = "https://github.com/canonical/postgresql-single-kernel-library/archive/40752dd71ffa1db9c15bf4a8c4f1dc0de7bf72b9.tar.gz"
 
 [[package]]
 name = "prompt-toolkit"
@@ -1941,7 +1945,6 @@ files = [
     {file = "psutil-7.2.2-cp37-abi3-win_arm64.whl", hash = "sha256:8c233660f575a5a89e6d4cb65d9f938126312bca76d8fe087b947b3a1aaac9ee"},
     {file = "psutil-7.2.2.tar.gz", hash = "sha256:0746f5f8d406af344fd547f1c8daa5f5c33dbc293bb8d6a16d80b4bb88f59372"},
 ]
-markers = {integration = "sys_platform != \"emscripten\" and sys_platform != \"cygwin\""}
 
 [package.extras]
 dev = ["abi3audit", "black", "check-manifest", "colorama ; os_name == \"nt\"", "coverage", "packaging", "psleak", "pylint", "pyperf", "pypinfo", "pyreadline3 ; os_name == \"nt\"", "pytest", "pytest-cov", "pytest-instafail", "pytest-xdist", "pywin32 ; os_name == \"nt\" and implementation_name != \"pypy\"", "requests", "rstcheck", "ruff", "setuptools", "sphinx", "sphinx_rtd_theme", "toml-sort", "twine", "validate-pyproject[all]", "virtualenv", "vulture", "wheel", "wheel ; os_name == \"nt\" and implementation_name != \"pypy\"", "wmi ; os_name == \"nt\" and implementation_name != \"pypy\""]
@@ -2115,7 +2118,7 @@ version = "2.13.4"
 description = "Data validation using Python type hints"
 optional = false
 python-versions = ">=3.9"
-groups = ["main", "charm-libs"]
+groups = ["main", "charm-libs", "integration"]
 files = [
     {file = "pydantic-2.13.4-py3-none-any.whl", hash = "sha256:45a282cde31d808236fd7ea9d919b128653c8b38b393d1c4ab335c62924d9aba"},
     {file = "pydantic-2.13.4.tar.gz", hash = "sha256:c40756b57adaa8b1efeeced5c196f3f3b7c435f90e84ea7f443901bec8099ef6"},
@@ -2137,7 +2140,7 @@ version = "2.46.4"
 description = "Core functionality for Pydantic validation and serialization"
 optional = false
 python-versions = ">=3.9"
-groups = ["main", "charm-libs"]
+groups = ["main", "charm-libs", "integration"]
 files = [
     {file = "pydantic_core-2.46.4-cp310-cp310-macosx_10_12_x86_64.whl", hash = "sha256:a396dcc17e5a0b164dbe026896245a4fa9ff402edca1dff0be3d53a517f74de4"},
     {file = "pydantic_core-2.46.4-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:da4b951fe36dc7c3a1ccb4e3cd1747c3542b8c9ceede8fc86cae054e764485f5"},
@@ -2598,7 +2601,7 @@ version = "15.0.0"
 description = "Render rich text, tables, progress bars, syntax highlighting, markdown and more to the terminal"
 optional = false
 python-versions = ">=3.9.0"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb"},
     {file = "rich-15.0.0.tar.gz", hash = "sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36"},
@@ -2789,7 +2792,7 @@ version = "1.0.13"
 description = "A generator library for concise, unambiguous and URL-safe UUIDs."
 optional = false
 python-versions = ">=3.6"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "shortuuid-1.0.13-py3-none-any.whl", hash = "sha256:a482a497300b49b4953e15108a7913244e1bb0d41f9d332f5e9925dba33a3c5a"},
     {file = "shortuuid-1.0.13.tar.gz", hash = "sha256:3bb9cf07f606260584b1df46399c0b87dd84773e7b25912b7e391e30797c5e72"},
@@ -3015,7 +3018,7 @@ version = "0.4.2"
 description = "Runtime typing introspection tools"
 optional = false
 python-versions = ">=3.9"
-groups = ["main", "charm-libs"]
+groups = ["main", "charm-libs", "integration"]
 files = [
     {file = "typing_inspection-0.4.2-py3-none-any.whl", hash = "sha256:4ed1cacbdc298c220f1bd249ed5287caa16f34d44ef4e9c3d0cbad5b521545e7"},
     {file = "typing_inspection-0.4.2.tar.gz", hash = "sha256:ba561c48a67c5958007083d386c3295464928b01faa735ab8547c5692e87f464"},
@@ -3148,7 +3151,7 @@ version = "1.3.2"
 description = "Pure-Python WebSocket protocol implementation"
 optional = false
 python-versions = ">=3.10"
-groups = ["main"]
+groups = ["main", "integration"]
 files = [
     {file = "wsproto-1.3.2-py3-none-any.whl", hash = "sha256:61eea322cdf56e8cc904bd3ad7573359a242ba65688716b0710a5eb12beab584"},
     {file = "wsproto-1.3.2.tar.gz", hash = "sha256:b86885dcf294e15204919950f666e06ffc6c7c114ca900b060d6e16293528294"},
@@ -3160,4 +3163,4 @@ h11 = ">=0.16.0,<1"
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.12,<4.0"
-content-hash = "1aa8c2c1c1a5643bfca523d1b20863d65b43d7f9e4740da0fb13824838514f60"
+content-hash = "6a5ae64219b2bd6d922d9a5b7ee087a8d3cd4627ed381bb37c51472795472c57"
diff --git a/pyproject.toml b/pyproject.toml
index 8bcde03322..79207478f8 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -21,7 +21,7 @@ charm-refresh = "^3.1.0.2"
 charmlibs-snap = "^1.0.1"
 charmlibs-systemd = "^1.0.0.post0"
 charmlibs-interfaces-tls-certificates = "^1.8.3"
-postgresql-charms-single-kernel = {extras = ["postgresql", "vm"], version = "16.3.2"}
+postgresql-charms-single-kernel = {url = "https://github.com/canonical/postgresql-single-kernel-library/archive/40752dd71ffa1db9c15bf4a8c4f1dc0de7bf72b9.tar.gz", extras = ["postgresql", "vm"]}
 
 [tool.poetry.group.charm-libs.dependencies]
 # data_platform_libs/v0/data_interfaces.py
@@ -76,7 +76,7 @@ allure-pytest-default-results = "^0.1.4"
 tomli = "^2.4.1"
 tomli-w = "^1.2.0"
 pysyncobj = "*"
-postgresql-charms-single-kernel = "*"
+postgresql-charms-single-kernel = {url = "https://github.com/canonical/postgresql-single-kernel-library/archive/40752dd71ffa1db9c15bf4a8c4f1dc0de7bf72b9.tar.gz", extras = ["postgresql", "vm"]}
 
 [tool.poetry.group.build-refresh-version]
 optional = true
diff --git a/src/charm.py b/src/charm.py
index 0b9626453f..1af80d694e 100755
--- a/src/charm.py
+++ b/src/charm.py
@@ -109,10 +109,8 @@
     SNAP_USER,
     SPI_MODULE,
     SYSTEM_USERS,
-    TLS_CA_BUNDLE_FILE,
-    TLS_CA_FILE,
-    TLS_CERT_FILE,
-    TLS_KEY_FILE,
+    TLS_CLIENT_RELATION,
+    TLS_PEER_RELATION,
     TRACING_PROTOCOL,
     TRACING_RELATION_NAME,
     UNIT_SCOPE,
@@ -121,12 +119,13 @@
 )
 from single_kernel_postgresql.core.config import CharmConfig
 from single_kernel_postgresql.core.state import CharmState
+from single_kernel_postgresql.events.tls import TLS
 from single_kernel_postgresql.events.tls_transfer import TLSTransfer
 from single_kernel_postgresql.managers.cluster import ClusterManager
 from single_kernel_postgresql.managers.config import ConfigManager
 from single_kernel_postgresql.managers.patroni import PatroniManager
 from single_kernel_postgresql.managers.tls import TLSManager
-from single_kernel_postgresql.utils import label2name, new_password, render_file
+from single_kernel_postgresql.utils import label2name, new_password
 from single_kernel_postgresql.utils.postgresql import (
     ACCESS_GROUP_IDENTITY,
     ACCESS_GROUPS,
@@ -155,7 +154,6 @@
 )
 from constants import (
     MONITORING_SNAP_SERVICE,
-    PATRONI_CONF_PATH,
     PGBACKREST_MONITORING_SNAP_SERVICE,
     POSTGRESQL_DATA_DIR,
     RAFT_PARTNER_PREFIX,
@@ -167,7 +165,6 @@
 from ldap import PostgreSQLLDAP
 from relations.async_replication import PostgreSQLAsyncReplication
 from relations.postgresql_provider import PostgreSQLProvider
-from relations.tls import TLS
 from relations.watcher import PostgreSQLWatcherRelation
 from rotate_logs import RotateLogs
 
@@ -379,13 +376,11 @@ def __init__(self, *args):
 
         # TODO switch to the abstract class base
         # State
-        self.state = CharmState(charm=self, substrate=self.substrate)  # type: ignore
+        self.state = CharmState(charm=self, substrate=self.substrate)
 
         # Managers
         self.patroni_manager = PatroniManager(state=self.state, workload=self.workload)
-        self.tls_manager = TLSManager(state=self.state, workload=self.workload)
         self.cluster_manager = ClusterManager(state=self.state, workload=self.workload)
-        self.config_manager = ConfigManager(state=self.state, workload=self.workload)
 
         self._observer = ClusterTopologyObserver(self, "/usr/bin/juju-exec")
         self._rotate_logs = RotateLogs(self)
@@ -421,7 +416,40 @@ def __init__(self, *args):
         self.postgresql_client_relation = PostgreSQLProvider(self)
         self.backup = PostgreSQLBackups(self, "s3-parameters")
         self.ldap = PostgreSQLLDAP(self, "ldap")
-        self.tls = TLS(self, PEER_RELATION)
+        # TLS events handler owns the two cert requirers; build it before the TLS
+        # manager so the manager can constructor-inject them for its live-fetch getters.
+        self.tls = TLS(self, self.state)
+        self.tls_manager = TLSManager(
+            state=self.state,
+            workload=self.workload,
+            client_certificate=self.tls.client_certificate,
+            peer_certificate=self.tls.peer_certificate,
+        )
+        self.config_manager = ConfigManager(
+            state=self.state,
+            workload=self.workload,
+            tls_manager=self.tls_manager,
+            patroni_manager=self.patroni_manager,
+            request_restart=self.request_restart,
+            refresh_endpoints=self.refresh_endpoints,
+            restart_services=self.restart_services,
+        )
+        # Bridge the lib TLS handler's requirer events back into a PostgreSQL reload:
+        # the lib handler stores+pushes certs on certificate_available, then we reload.
+        # Also fires on relation_broken so detaching the TLS operator re-renders Patroni
+        # with TLS disabled.
+        self.framework.observe(
+            self.tls.client_certificate.on.certificate_available, self._reload_tls_after_push
+        )
+        self.framework.observe(
+            self.tls.peer_certificate.on.certificate_available, self._reload_tls_after_push
+        )
+        self.framework.observe(
+            self.on[TLS_CLIENT_RELATION].relation_broken, self._reload_tls_after_push
+        )
+        self.framework.observe(
+            self.on[TLS_PEER_RELATION].relation_broken, self._reload_tls_after_push
+        )
         self.tls_transfer = TLSTransfer(self, PEER_RELATION)
         self.async_replication = PostgreSQLAsyncReplication(self)
         self.watcher_offer = PostgreSQLWatcherRelation(self)
@@ -498,6 +526,49 @@ def substrate(self) -> Substrates:
         """
         return Substrates.VM
 
+    def _reload_tls_after_push(self, event) -> None:
+        """Reload PostgreSQL after the lib TLS handler stores+pushes certs.
+
+        Also fires on the TLS relations' relation_broken so detaching the TLS
+        operator re-renders Patroni with TLS disabled and reloads.
+
+        Mirror the handler's readiness guard: when the internal CA is absent the
+        handler defers its push (no files on disk), so skip the reload to avoid
+        rendering ssl:on against missing TLS files on an already-running unit.
+
+        A transient config-apply failure (Patroni API unreachable, member not
+        started) defers and retries rather than leaving stale TLS state or failing
+        the hook.
+        """
+        if not self.get_secret(APP_SCOPE, "internal-ca"):
+            return
+        # Don't enable TLS in the config until the lib has written the cert files to
+        # disk (its push can defer while this local render would still succeed, which
+        # would start Patroni ssl:on against missing files).
+        if self.is_tls_enabled and not self.tls_manager.client_tls_files_on_disk():
+            event.defer()
+            return
+        try:
+            if not self.update_config():
+                event.defer()
+        except Exception:
+            logger.exception("TLS reload (update_config) failed; deferring")
+            event.defer()
+
+    def _regenerate_internal_cert(self, *, reload: bool = True) -> None:
+        """Generate the internal peer cert, push it to the workload, and (optionally) reload.
+
+        reload=False is used at cluster bootstrap: the leader renders patroni.yml on
+        leader-elected and each replica renders it in _on_peer_relation_changed just
+        before starting Patroni, so a reload here would be redundant. The internal peer
+        cert does not toggle ssl in the config -- only the operator/client cert does, via
+        is_tls_enabled -- so skipping the reload cannot leave a stale ssl setting.
+        """
+        self.tls_manager.generate_internal_peer_cert()
+        self.tls_manager.push_tls_files()
+        if reload:
+            self.update_config()
+
     def _check_and_update_internal_cert(self) -> None:
         """Check if the internal cert CN matches the unit IP and regenerate if needed."""
         try:
@@ -509,7 +580,7 @@ def _check_and_update_internal_cert(self) -> None:
                     != self.state.unit_ip
                 )
             ):
-                self.tls.generate_internal_peer_cert()
+                self._regenerate_internal_cert()
         except Exception:
             logger.exception("Unable to check or update internal cert")
 
@@ -1548,7 +1619,7 @@ def is_standby_cluster(self) -> bool:
     @property
     def is_tls_enabled(self) -> bool:
         """Return whether TLS is enabled."""
-        return all(self.tls.get_client_tls_files())
+        return all(self.tls_manager.get_client_tls_files())
 
     @property
     def _peer_members_ips(self) -> set[str]:
@@ -1759,7 +1830,7 @@ def _on_leader_elected(self, event: LeaderElectedEvent) -> None:  # noqa: C901
             logger.debug("On leader elected failed to reconfigure cluster.")
 
         if not self.get_secret(APP_SCOPE, "internal-ca"):
-            self.tls.generate_internal_peer_ca()
+            self.tls_manager.generate_internal_peer_ca()
         self.update_config()
 
         # Don't update connection endpoints in the first time this event run for
@@ -1946,7 +2017,7 @@ def _on_start(self, event: StartEvent) -> None:
             event.defer()
             return
         if not self.get_secret(UNIT_SCOPE, "internal-cert"):
-            self.tls.generate_internal_peer_cert()
+            self._regenerate_internal_cert(reload=False)
 
         self.unit_peer_data.update({"ip": self.state.unit_ip})
         self._ensure_storage_layout()
@@ -2497,10 +2568,12 @@ def _update_certificate(self) -> None:
         # Request the certificate only if there is already one. If there isn't,
         # the certificate will be generated in the relation joined event when
         # relating to the TLS Certificates Operator.
-        if all(self.tls.get_client_tls_files()) or all(self.tls.get_peer_tls_files()):
+        if all(self.tls_manager.get_client_tls_files()) or all(
+            self.tls_manager.get_peer_tls_files()
+        ):
             self.tls.refresh_tls_certificates_event.emit()
         if self.get_secret(UNIT_SCOPE, "internal-cert"):
-            self.tls.generate_internal_peer_cert()
+            self._regenerate_internal_cert()
 
     @property
     def is_blocked(self) -> bool:
@@ -2582,37 +2655,6 @@ def _peers(self) -> Relation | None:
         """
         return self.model.get_relation(PEER_RELATION)
 
-    def push_tls_files_to_workload(self) -> bool:
-        """Move TLS files to the PostgreSQL storage path and enable TLS."""
-        key, ca, cert = self.tls.get_client_tls_files()
-        if key is not None:
-            render_file(Substrates.VM, f"{PATRONI_CONF_PATH}/{TLS_KEY_FILE}", key, 0o600)
-        if ca is not None:
-            render_file(Substrates.VM, f"{PATRONI_CONF_PATH}/{TLS_CA_FILE}", ca, 0o600)
-        if cert is not None:
-            render_file(Substrates.VM, f"{PATRONI_CONF_PATH}/{TLS_CERT_FILE}", cert, 0o600)
-
-        key, ca, cert = self.tls.get_peer_tls_files()
-        if key is not None:
-            render_file(Substrates.VM, f"{PATRONI_CONF_PATH}/peer_{TLS_KEY_FILE}", key, 0o600)
-        if ca is not None:
-            render_file(Substrates.VM, f"{PATRONI_CONF_PATH}/peer_{TLS_CA_FILE}", ca, 0o600)
-        if cert is not None:
-            render_file(Substrates.VM, f"{PATRONI_CONF_PATH}/peer_{TLS_CERT_FILE}", cert, 0o600)
-
-        render_file(
-            Substrates.VM,
-            f"{PATRONI_CONF_PATH}/{TLS_CA_BUNDLE_FILE}",
-            self.tls.get_peer_ca_bundle(),
-            0o600,
-        )
-
-        try:
-            return self.update_config()
-        except Exception:
-            logger.exception("TLS files failed to push. Error in config update")
-            return False
-
     def push_ca_file_into_workload(self, secret_name: str) -> bool:
         """Move CA certificates file into the PostgreSQL storage path."""
         certs = self.get_secret(UNIT_SCOPE, secret_name)
@@ -2699,228 +2741,21 @@ def _can_connect_to_postgresql(self) -> bool:
             return False
         return True
 
-    def _calculate_max_worker_processes(self) -> str | None:
-        """Calculate cpu_max_worker_processes configuration value."""
-        if self.config.cpu_max_worker_processes == "auto":
-            # auto = minimum(8, 2 * vCores)
-            return str(min(8, 2 * self.cpu_count))
-        elif self.config.cpu_max_worker_processes is not None:
-            value = self.config.cpu_max_worker_processes
-            cap = 10 * self.cpu_count
-            if value > cap:
-                raise ValueError(
-                    f"cpu_max_worker_processes value {value} exceeds maximum allowed "
-                    f"of {cap} (10 * vCores). Please set a value <= {cap}."
-                )
-            return str(value)
-        return None
-
-    def _validate_worker_config_value(self, param_name: str, value: int) -> str:
-        """Shared validation logic for worker process parameters.
-
-        Args:
-            param_name: The configuration parameter name (for error messages)
-            value: The integer value to validate
-
-        Returns:
-            String representation of the validated value
+    def request_restart(self) -> None:
+        """Clear the restarted flag and acquire the rolling-restart lock. Bridge for ConfigManager."""
+        self.unit_peer_data.pop("postgresql_restarted", None)
+        self.on[str(self.restart_manager.name)].acquire_lock.emit()
 
-        Raises:
-            ValidationError: If value is less than 2
-            ValueError: If value exceeds 10 * vCores
-        """
-        cap = 10 * self.cpu_count
-        if value > cap:
-            raise ValueError(
-                f"{param_name} value {value} exceeds maximum allowed "
-                f"of {cap} (10 * vCores). Please set a value <= {cap}."
-            )
-        return str(value)
-
-    def _calculate_max_parallel_workers(self, base_max_workers: int) -> str | None:
-        """Calculate cpu_max_parallel_workers configuration value."""
-        if self.config.cpu_max_parallel_workers == "auto":
-            return str(base_max_workers)
-        elif self.config.cpu_max_parallel_workers is not None:
-            # Validate the value first
-            validated_value_str = self._validate_worker_config_value(
-                "cpu_max_parallel_workers", self.config.cpu_max_parallel_workers
-            )
-            # Apply the min constraint with base_max_workers
-            return str(min(int(validated_value_str), base_max_workers))
-        return None
-
-    def _calculate_max_parallel_maintenance_workers(self, base_max_workers: int) -> str | None:
-        """Calculate cpu_max_parallel_maintenance_workers configuration value."""
-        if self.config.cpu_max_parallel_maintenance_workers == "auto":
-            return str(base_max_workers)
-        elif self.config.cpu_max_parallel_maintenance_workers is not None:
-            return self._validate_worker_config_value(
-                "cpu_max_parallel_maintenance_workers",
-                self.config.cpu_max_parallel_maintenance_workers,
-            )
-        return None
-
-    def _calculate_max_logical_replication_workers(self, base_max_workers: int) -> str | None:
-        """Calculate cpu_max_logical_replication_workers configuration value."""
-        if self.config.cpu_max_logical_replication_workers == "auto":
-            return str(base_max_workers)
-        elif self.config.cpu_max_logical_replication_workers is not None:
-            return self._validate_worker_config_value(
-                "cpu_max_logical_replication_workers",
-                self.config.cpu_max_logical_replication_workers,
-            )
-        return None
-
-    def _calculate_max_sync_workers_per_subscription(self, base_max_workers: int) -> str | None:
-        """Calculate cpu_max_sync_workers_per_subscription configuration value."""
-        if self.config.cpu_max_sync_workers_per_subscription == "auto":
-            return str(base_max_workers)
-        elif self.config.cpu_max_sync_workers_per_subscription is not None:
-            return self._validate_worker_config_value(
-                "cpu_max_sync_workers_per_subscription",
-                self.config.cpu_max_sync_workers_per_subscription,
-            )
-        return None
-
-    def _calculate_max_parallel_apply_workers_per_subscription(
-        self, base_max_workers: int
-    ) -> str | None:
-        """Calculate cpu_max_parallel_apply_workers_per_subscription configuration value."""
-        if self.config.cpu_max_parallel_apply_workers_per_subscription == "auto":
-            return str(base_max_workers)
-        elif self.config.cpu_max_parallel_apply_workers_per_subscription is not None:
-            return self._validate_worker_config_value(
-                "cpu_max_parallel_apply_workers_per_subscription",
-                self.config.cpu_max_parallel_apply_workers_per_subscription,
-            )
-        return None
-
-    def _calculate_worker_process_config(self) -> dict[str, str]:
-        """Calculate worker process configuration values.
-
-        Handles 'auto' values and capping logic for worker process parameters.
-        Returns a dictionary with the calculated values ready for PostgreSQL.
-        """
-        result: dict[str, str] = {}
-
-        # Calculate cpu_max_worker_processes (baseline for other worker configs)
-        cpu_max_worker_processes_value = self._calculate_max_worker_processes()
-        if cpu_max_worker_processes_value is not None:
-            result["max_worker_processes"] = cpu_max_worker_processes_value
-
-        # Get the effective cpu_max_worker_processes for dependent configs
-        # Use the calculated value, or fall back to PostgreSQL default (8)
-        base_max_workers = int(result.get("max_worker_processes", "8"))
-
-        # Calculate other worker parameters
-        cpu_max_parallel_workers_value = self._calculate_max_parallel_workers(base_max_workers)
-        if cpu_max_parallel_workers_value is not None:
-            result["max_parallel_workers"] = cpu_max_parallel_workers_value
-
-        cpu_max_parallel_maintenance_workers_value = (
-            self._calculate_max_parallel_maintenance_workers(base_max_workers)
-        )
-        if cpu_max_parallel_maintenance_workers_value is not None:
-            result["max_parallel_maintenance_workers"] = cpu_max_parallel_maintenance_workers_value
-
-        cpu_max_logical_replication_workers_value = (
-            self._calculate_max_logical_replication_workers(base_max_workers)
-        )
-        if cpu_max_logical_replication_workers_value is not None:
-            result["max_logical_replication_workers"] = cpu_max_logical_replication_workers_value
-
-        cpu_max_sync_workers_per_subscription_value = (
-            self._calculate_max_sync_workers_per_subscription(base_max_workers)
-        )
-        if cpu_max_sync_workers_per_subscription_value is not None:
-            result["max_sync_workers_per_subscription"] = (
-                cpu_max_sync_workers_per_subscription_value
-            )
-
-        cpu_max_parallel_apply_workers_per_subscription_value = (
-            self._calculate_max_parallel_apply_workers_per_subscription(base_max_workers)
-        )
-        if cpu_max_parallel_apply_workers_per_subscription_value is not None:
-            result["max_parallel_apply_workers_per_subscription"] = (
-                cpu_max_parallel_apply_workers_per_subscription_value
-            )
-
-        return result
-
-    def _api_update_config(self) -> bool:
-        # Use config value if set, calculate otherwise
-        max_connections = (
-            self.config.experimental_max_connections
-            if self.config.experimental_max_connections
-            else max(4 * self.cpu_count, 100)
-        )
-        cfg_patch: dict[str, int | str | None] = {
-            "max_connections": max_connections,
-            "max_prepared_transactions": self.config.memory_max_prepared_transactions,
-            "max_replication_slots": 25,
-            "max_wal_senders": 25,
-            "shared_buffers": self.config.memory_shared_buffers,
-            "wal_keep_size": self.config.durability_wal_keep_size,
-        }
-
-        # Add restart-required worker process parameters via Patroni API
-        worker_configs = self._calculate_worker_process_config()
-        if "max_worker_processes" in worker_configs:
-            cfg_patch["max_worker_processes"] = worker_configs["max_worker_processes"]
-        if "max_logical_replication_workers" in worker_configs:
-            cfg_patch["max_logical_replication_workers"] = worker_configs[
-                "max_logical_replication_workers"
-            ]
-
-        base_patch = {
-            **self.state.synchronous_configuration,
-            "maximum_lag_on_failover": self.config.durability_maximum_lag_on_failover,
-        }
-        if primary_endpoint := self.async_replication.get_primary_cluster_endpoint():
-            base_patch["standby_cluster"] = {"host": primary_endpoint}
-        try:
-            self.patroni_manager.bulk_update_parameters_controller_by_patroni(
-                cfg_patch, base_patch
-            )
-        except RetryError:
-            return False
-        return True
-
-    def _build_postgresql_parameters(self) -> dict[str, str] | None:
-        """Build PostgreSQL configuration parameters.
-
-        Returns:
-            Dictionary of PostgreSQL parameters or None if base parameters couldn't be built.
-        """
-        limit_memory = None
-        if self.config.profile_limit_memory:
-            limit_memory = self.config.profile_limit_memory * 10**6
-
-        # Build PostgreSQL parameters.
-        pg_parameters = self.postgresql.build_postgresql_parameters(
-            self.model.config, self.get_available_memory(), limit_memory
-        )
-
-        # Calculate and merge worker process configurations
-        worker_configs = self._calculate_worker_process_config()
-
-        # Add cpu_wal_compression configuration (separate from worker processes)
-        if self.config.cpu_wal_compression is not None:
-            cpu_wal_compression = "on" if self.config.cpu_wal_compression else "off"
-        else:
-            # Use config.yaml default when unset (default: true)
-            cpu_wal_compression = "on"
-
-        if pg_parameters is not None:
-            pg_parameters.update(worker_configs)
-            pg_parameters["wal_compression"] = cpu_wal_compression
-        else:
-            pg_parameters = dict(worker_configs)
-            pg_parameters["wal_compression"] = cpu_wal_compression
-            logger.debug(f"pg_parameters set to worker_configs = {pg_parameters}")
+    def refresh_endpoints(self) -> None:
+        """Refresh client-relation endpoints. Bridge for ConfigManager."""
+        self.postgresql_client_relation.update_endpoints()
 
-        return pg_parameters
+    def restart_services(self) -> None:
+        """Restart the monitoring and LDAP sync snap services if needed. Bridge for ConfigManager."""
+        cache = snap.SnapCache()
+        postgres_snap = cache[charm_refresh.snap_name()]
+        self._restart_metrics_service(postgres_snap)
+        self._restart_ldap_sync_service(postgres_snap)
 
     def update_config(
         self,
@@ -2932,102 +2767,22 @@ def update_config(
         """Updates Patroni config file based on the existence of the TLS files."""
         if refresh is None:
             refresh = self.refresh
-
-        # Build PostgreSQL parameters
-        pg_parameters = self._build_postgresql_parameters()
-
-        # replication_slots = self.logical_replication.replication_slots()
-
-        # Update and reload configuration based on TLS files availability.
-        logger.debug(f"Calling render_patroni_yml_file with parameters = {pg_parameters}")
-        # TODO move to config manager's update config
-        self.config_manager.render_patroni_yml_file(
-            connectivity=self.state.peer.is_connectivity_enabled,
+        return self.config_manager.update_config(
+            self.postgresql,
+            self.generate_user_hash,
             is_creating_backup=is_creating_backup,
-            enable_ldap=self.state.application.is_ldap_enabled,
-            # TODO add rel handler
-            enable_tls=self.is_tls_enabled,
-            backup_id=self.state.application.data.get("restoring-backup"),
-            pitr_target=self.state.application.data.get("restore-to-time"),
-            restore_timeline=self.state.application.data.get("restore-timeline"),
-            restore_to_latest=self.state.application.data.get("restore-to-time", None) == "latest",
-            stanza=self.state.application.data.get("stanza", self.state.peer.data.get("stanza")),
-            restore_stanza=self.state.application.data.get("restore-stanza"),
-            parameters=pg_parameters,
-            # TODO add rel handler
-            user_databases_map=self.relations_user_databases_map,
-            # TODO add rel handler
+            relations_user_databases_map=self.relations_user_databases_map,
             ldap_parameters=self.get_ldap_parameters(),
-            # TODO add rel handler
             async_primary_cluster_endpoint=self.async_replication.get_primary_cluster_endpoint(),
             async_partner_addresses=self.async_replication.get_partner_addresses(),
             async_standby_endpoints=self.async_replication.get_standby_endpoints(),
-            # TODO add rel handler
             watcher_raft_address=self.watcher_offer.watcher_raft_address
             if self.watcher_offer.is_active
             else None,
             no_peers=no_peers,
-            # slots=replication_slots,
-        )
-        if no_peers:
-            return True
-
-        if not self.workload.is_patroni_running():
-            # If Patroni/PostgreSQL has not started yet and TLS relations was initialised,
-            # then mark TLS as enabled. This commonly happens when the charm is deployed
-            # in a bundle together with the TLS certificates operator. This flag is used to
-            # know when to call the Patroni API using HTTP or HTTPS.
-            self.unit_peer_data.update({
-                "tls": "enabled" if self.is_tls_enabled else "",
-            })
-            self.postgresql_client_relation.update_endpoints()
-            logger.debug("Early exit update_config: Workload not started yet")
-            return True
-
-        if not self.patroni_manager.member_started:
-            if self.is_tls_enabled:
-                logger.debug(
-                    "Early exit update_config: patroni not responding but TLS is enabled."
-                )
-                self._handle_postgresql_restart_need(True)
-                return True
-            logger.debug("Early exit update_config: Patroni not started yet")
-            return False
-
-        # Try to connect
-        if not self._can_connect_to_postgresql:
-            logger.warning("Early exit update_config: Cannot connect to Postgresql")
-            return False
-
-        if not self._api_update_config():
-            logger.warning("Early exit update_config: Unable to patch Patroni API")
-            return False
-
-        # self._patroni.ensure_slots_controller_by_patroni(replication_slots)
-
-        self._handle_postgresql_restart_need(
-            self.unit_peer_data.get("config_hash") != self.generate_config_hash
+            refresh=refresh,
         )
 
-        cache = snap.SnapCache()
-        postgres_snap = cache[charm_refresh.snap_name()]
-
-        # TODO handle case of scale up while refresh in progress & `refresh` is None
-        if refresh is not None and postgres_snap.revision != refresh.pinned_snap_revision:
-            logger.debug("Early exit: snap was not refreshed to the right version yet")
-            return True
-
-        self._restart_metrics_service(postgres_snap)
-        self._restart_ldap_sync_service(postgres_snap)
-
-        self.unit_peer_data.update({
-            "user_hash": self.generate_user_hash,
-            "config_hash": self.generate_config_hash,
-        })
-        if self.unit.is_leader():
-            self.app_peer_data.update({"user_hash": self.generate_user_hash})
-        return True
-
     def _validate_config_options(self) -> None:
         """Validates specific config options that need access to the database or to the TLS status."""
         if (
@@ -3057,57 +2812,10 @@ def _validate_config_options(self) -> None:
                 "storage_default_table_access_method config option has an invalid value"
             )
 
-    def _handle_postgresql_restart_need(self, config_changed: bool) -> None:
-        """Handle PostgreSQL restart need based on the TLS configuration and configuration changes."""
-        if self._can_connect_to_postgresql:
-            restart_postgresql = self.is_tls_enabled != self.postgresql.is_tls_enabled(
-                check_current_host=True
-            )
-        else:
-            restart_postgresql = False
-
-        try:
-            self.patroni_manager.reload_patroni_configuration()
-        except Exception as e:
-            logger.error(f"Reload patroni call failed! error: {e!s}")
-
-        if config_changed and not restart_postgresql:
-            # Wait for some more time than the Patroni's loop_wait default value (10 seconds),
-            # which tells how much time Patroni will wait before checking the configuration
-            # file again to reload it.
-            try:
-                for attempt in Retrying(stop=stop_after_attempt(5), wait=wait_fixed(3)):
-                    with attempt:
-                        restart_postgresql = restart_postgresql or self.is_restart_pending()
-                        if not restart_postgresql:
-                            raise Exception
-            except RetryError:
-                # Ignore the error, as it happens only to indicate that the configuration has not changed.
-                pass
-
-        self.unit_peer_data.update({"tls": "enabled" if self.is_tls_enabled else ""})
-        self.postgresql_client_relation.update_endpoints()
-
-        # Restart PostgreSQL if TLS configuration has changed
-        # (so the both old and new connections use the configuration).
-        if restart_postgresql:
-            logger.info("PostgreSQL restart required")
-            self.unit_peer_data.pop("postgresql_restarted", None)
-            self.on[str(self.restart_manager.name)].acquire_lock.emit()
-
     def _update_relation_endpoints(self) -> None:
         """Updates endpoints and read-only endpoint in all relations."""
         self.postgresql_client_relation.update_endpoints()
 
-    def get_available_memory(self) -> int:
-        """Returns the system available memory in bytes."""
-        with open("/proc/meminfo") as meminfo:
-            for line in meminfo:
-                if "MemTotal" in line:
-                    return int(line.split()[1]) * 1024
-
-        return 0
-
     @property
     def client_relations(self) -> list[Relation]:
         """Return the list of established client relations."""
@@ -3198,11 +2906,6 @@ def generate_user_hash(self) -> str:
         """Generate expected user and database hash."""
         return shake_128(str(self._collect_user_relations()).encode()).hexdigest(16)
 
-    @cached_property
-    def generate_config_hash(self) -> str:
-        """Generate current configuration hash."""
-        return shake_128(str(self.config.model_dump()).encode()).hexdigest(16)
-
     def override_patroni_restart_condition(
         self, new_condition: str, repeat_cause: str | None
     ) -> bool:
@@ -3356,32 +3059,6 @@ def get_ldap_parameters(self) -> dict:
 
         return params
 
-    def is_restart_pending(self) -> bool:
-        """Query pg_settings for pending restart."""
-        connection = None
-        try:
-            with (
-                self.postgresql._connect_to_database(
-                    database_host=self.postgresql.current_host
-                ) as connection,
-                connection.cursor() as cursor,
-            ):
-                cursor.execute("SELECT COUNT(*) FROM pg_settings WHERE pending_restart=True;")
-                result = cursor.fetchone()
-                if result is not None:
-                    return result[0] > 0
-                else:
-                    return False
-        except psycopg2.OperationalError:
-            logger.warning("Failed to connect to PostgreSQL.")
-            return False
-        except psycopg2.Error as e:
-            logger.error(f"Failed to check if restart is pending: {e}")
-            return False
-        finally:
-            if connection:
-                connection.close()
-
 
 if __name__ == "__main__":
     main(PostgresqlOperatorCharm)
diff --git a/src/relations/postgresql_provider.py b/src/relations/postgresql_provider.py
index 7b5f0bd699..a0e47d79aa 100644
--- a/src/relations/postgresql_provider.py
+++ b/src/relations/postgresql_provider.py
@@ -485,7 +485,7 @@ def update_endpoints(self, event: DatabaseRequestedEvent | None = None) -> None:
         tls = "True" if self.charm.is_tls_enabled else "False"
         ca = None
         if tls == "True":
-            _, ca, _ = self.charm.tls.get_client_tls_files()
+            _, ca, _ = self.charm.tls_manager.get_client_tls_files()
         if not ca:
             ca = ""
 
diff --git a/src/relations/tls.py b/src/relations/tls.py
deleted file mode 100644
index 2c3bc92b5a..0000000000
--- a/src/relations/tls.py
+++ /dev/null
@@ -1,259 +0,0 @@
-# Copyright 2022 Canonical Ltd.
-# See LICENSE file for licensing details.
-
-"""TLS Handler."""
-
-import logging
-import socket
-from datetime import timedelta
-from typing import TYPE_CHECKING
-
-from charmlibs.interfaces.tls_certificates import (
-    Certificate,
-    CertificateRequestAttributes,
-    PrivateKey,
-    TLSCertificatesRequiresV4,
-    generate_ca,
-    generate_certificate,
-    generate_csr,
-    generate_private_key,
-)
-from ops import (
-    EventSource,
-)
-from ops.framework import EventBase, Object
-from ops.pebble import ConnectionError as PebbleConnectionError
-from ops.pebble import PathError, ProtocolError
-from single_kernel_postgresql.config.literals import (
-    APP_SCOPE,
-    UNIT_SCOPE,
-)
-from tenacity import RetryError
-
-if TYPE_CHECKING:
-    from charm import PostgresqlOperatorCharm
-
-logger = logging.getLogger(__name__)
-SCOPE = "unit"
-TLS_CLIENT_RELATION = "client-certificates"
-TLS_PEER_RELATION = "peer-certificates"
-
-
-class RefreshTLSCertificatesEvent(EventBase):
-    """Event for refreshing TLS certificates."""
-
-
-class TlsError(Exception):
-    """TLS implementation internal exception."""
-
-
-class TLS(Object):
-    """In this class we manage certificates relation."""
-
-    refresh_tls_certificates_event = EventSource(RefreshTLSCertificatesEvent)
-
-    def _get_client_addrs(self) -> set[str]:
-        client_addrs = set()
-        if addr := self.charm.unit_peer_data.get("database-address"):
-            client_addrs.add(addr)
-        return client_addrs
-
-    def _get_peer_addrs(self) -> set[str]:
-        peer_addrs = set()
-        if addr := self.charm.unit_peer_data.get("database-peers-address"):
-            peer_addrs.add(addr)
-        if addr := self.charm.unit_peer_data.get("replication-address"):
-            peer_addrs.add(addr)
-        if addr := self.charm.unit_peer_data.get("replication-offer-address"):
-            peer_addrs.add(addr)
-        if addr := self.charm.unit_peer_data.get(
-            "ip", self.charm.unit_peer_data.get("private-address")
-        ):
-            peer_addrs.add(addr)
-        return peer_addrs
-
-    def _get_common_name(self) -> str:
-        return self.charm.unit_peer_data.get("database-address") or self.host
-
-    def _get_peer_common_name(self) -> str:
-        return self.charm.unit_peer_data.get("database-peers-address") or self.host
-
-    def __init__(self, charm: "PostgresqlOperatorCharm", peer_relation: str):
-        super().__init__(charm, "client-relations")
-        self.charm = charm
-        self.peer_relation = peer_relation
-        unit_id = self.charm.unit.name.split("/")[1]
-        self.host = f"{self.charm.app.name}-{unit_id}"
-        if self.charm.unit_peer_data:
-            client_addresses = self._get_client_addrs()
-            peer_addresses = self._get_peer_addrs()
-        else:
-            client_addresses = set()
-            peer_addresses = set()
-        self.common_hosts = {self.host}
-        if fqdn := socket.getfqdn():
-            self.common_hosts.add(fqdn)
-
-        self.client_certificate = TLSCertificatesRequiresV4(
-            self.charm,
-            TLS_CLIENT_RELATION,
-            certificate_requests=[
-                CertificateRequestAttributes(
-                    common_name=self._get_common_name(),
-                    sans_ip=frozenset(client_addresses),
-                    sans_dns=frozenset({
-                        *self.common_hosts,
-                        # IP address need to be part of the DNS SANs list due to
-                        # https://github.com/pgbackrest/pgbackrest/issues/1977.
-                        *client_addresses,
-                    }),
-                ),
-            ],
-            refresh_events=[self.refresh_tls_certificates_event],
-        )
-        self.peer_certificate = TLSCertificatesRequiresV4(
-            self.charm,
-            TLS_PEER_RELATION,
-            certificate_requests=[
-                CertificateRequestAttributes(
-                    common_name=self._get_peer_common_name(),
-                    sans_ip=frozenset(self._get_peer_addrs()),
-                    sans_dns=frozenset({
-                        *self.common_hosts,
-                        # IP address need to be part of the DNS SANs list due to
-                        # https://github.com/pgbackrest/pgbackrest/issues/1977.
-                        *peer_addresses,
-                    }),
-                ),
-            ],
-            refresh_events=[self.refresh_tls_certificates_event],
-        )
-
-        self.framework.observe(
-            self.client_certificate.on.certificate_available, self._on_certificate_available
-        )
-        self.framework.observe(
-            self.peer_certificate.on.certificate_available, self._on_peer_certificate_available
-        )
-
-        self.framework.observe(
-            self.charm.on[TLS_CLIENT_RELATION].relation_broken, self._on_certificate_available
-        )
-        self.framework.observe(
-            self.charm.on[TLS_PEER_RELATION].relation_broken, self._on_peer_certificate_available
-        )
-
-    def _on_peer_certificate_available(self, event: EventBase) -> None:
-        certs, _ = self.peer_certificate.get_assigned_certificates()
-        new_ca = str(certs[0].ca) if certs else None
-        current_ca = self.charm.get_secret(UNIT_SCOPE, "current-ca")
-        # Stash the CAs in case of rotation
-        if new_ca != current_ca:
-            self.charm.set_secret(UNIT_SCOPE, "current-ca", new_ca)
-            self.charm.set_secret(UNIT_SCOPE, "old-ca", current_ca)
-        self._on_certificate_available(event)
-
-    def _on_certificate_available(self, event: EventBase) -> None:
-        if not self.charm.get_secret(APP_SCOPE, "internal-ca"):
-            logger.debug("Charm not ready yet")
-            event.defer()
-            return
-        try:
-            if not self.charm.push_tls_files_to_workload():
-                logger.debug("Cannot push TLS certificates at this moment")
-                event.defer()
-                return
-        except (PebbleConnectionError, PathError, ProtocolError, RetryError) as e:
-            logger.error("Cannot push TLS certificates: %r", e)
-            event.defer()
-            return
-
-    def get_client_tls_files(self) -> tuple[str | None, str | None, str | None]:
-        """Prepare TLS files in special PostgreSQL way.
-
-        PostgreSQL needs three files:
-        — CA file should have a full chain.
-        — Key file should have private key.
-        — Certificate file should have certificate without certificate chain.
-        """
-        ca_file = None
-        cert = None
-        key = None
-        certs, private_key = self.client_certificate.get_assigned_certificates()
-        if private_key:
-            key = str(private_key)
-        if certs:
-            cert = str(certs[0].certificate)
-            ca_file = str(certs[0].ca)
-        return key, ca_file, cert
-
-    def get_peer_tls_files(self) -> tuple[str | None, str | None, str | None]:
-        """Prepare TLS files in special PostgreSQL way.
-
-        PostgreSQL needs three files:
-        — CA file should have a full chain.
-        — Key file should have private key.
-        — Certificate file should have certificate without certificate chain.
-        """
-        ca_file = None
-        cert = None
-        key = None
-        certs, private_key = self.peer_certificate.get_assigned_certificates()
-        if private_key:
-            key = str(private_key)
-        if certs:
-            cert = str(certs[0].certificate)
-            ca_file = str(certs[0].ca)
-        if not all((key, ca_file, cert)):
-            key = self.charm.get_secret(UNIT_SCOPE, "internal-key")
-            cert = self.charm.get_secret(UNIT_SCOPE, "internal-cert")
-            ca_file = self.charm.get_secret(APP_SCOPE, "internal-ca")
-        return key, ca_file, cert
-
-    def get_peer_ca_bundle(self) -> str:
-        """Get bundled CA certs."""
-        certs, _ = self.peer_certificate.get_assigned_certificates()
-        operator_ca = str(certs[0].ca) if certs else ""
-        old_operator_ca = self.charm.get_secret(UNIT_SCOPE, "old-ca") or ""
-        internal_ca = self.charm.get_secret(APP_SCOPE, "internal-ca") or ""
-        return "\n".join((operator_ca, old_operator_ca, internal_ca)).strip()
-
-    def generate_internal_peer_ca(self) -> None:
-        """Generate internal peer CA using the tls lib."""
-        private_key = generate_private_key()
-        ca = generate_ca(
-            private_key,
-            common_name=f"{self.charm.app.name}-{self.charm.model.uuid}",
-            validity=timedelta(days=7300),
-        )
-        logger.warning("Internal peer CA generated. Please use a proper TLS operator if possible.")
-        self.charm.set_secret(APP_SCOPE, "internal-ca-key", str(private_key))
-        self.charm.set_secret(APP_SCOPE, "internal-ca", str(ca))
-
-    def generate_internal_peer_cert(self) -> None:
-        """Generate internal peer certificate using the tls lib."""
-        if not (ca_key_secret := self.charm.get_secret(APP_SCOPE, "internal-ca-key")):
-            raise TlsError("No CA key content.")
-        ca_key = PrivateKey.from_string(ca_key_secret)
-        if not (ca_secret := self.charm.get_secret(APP_SCOPE, "internal-ca")):
-            raise TlsError("No CA cert content.")
-        ca = Certificate.from_string(ca_secret)
-        private_key = generate_private_key()
-        csr = generate_csr(
-            private_key,
-            common_name=self._get_peer_common_name(),
-            sans_ip=frozenset(self._get_peer_addrs()),
-            sans_dns=frozenset({
-                *self.common_hosts,
-                # IP address need to be part of the DNS SANs list due to
-                # https://github.com/pgbackrest/pgbackrest/issues/1977.
-                *self._get_peer_addrs(),
-            }),
-        )
-        cert = generate_certificate(csr, ca, ca_key, validity=timedelta(days=7300))
-        self.charm.set_secret(UNIT_SCOPE, "internal-key", str(private_key))
-        self.charm.set_secret(UNIT_SCOPE, "internal-cert", str(cert))
-        self.charm.push_tls_files_to_workload()
-        logger.info(
-            "Internal peer certificate generated. Please use a proper TLS operator if possible."
-        )
diff --git a/src/relations/watcher.py b/src/relations/watcher.py
index 9cc92a3bef..4f46e5f760 100644
--- a/src/relations/watcher.py
+++ b/src/relations/watcher.py
@@ -419,7 +419,7 @@ def _update_relation_data(self, relation: Relation) -> None:
             "raft-secret-id": secret_id,
             "raft-partner-addrs": json.dumps(pg_endpoints),
             "raft-port": str(RAFT_PORT),
-            "patroni-cas": self.charm.tls.get_peer_ca_bundle(),
+            "patroni-cas": self.charm.tls_manager.get_peer_ca_bundle(),
             "standby-clusters": json.dumps(self._get_standby_clusters()),
             "tls-enabled": "true" if self.charm.is_tls_enabled else "false",
         })
diff --git a/templates/patroni.yml.j2 b/templates/patroni.yml.j2
deleted file mode 100644
index 65bdc02c6b..0000000000
--- a/templates/patroni.yml.j2
+++ /dev/null
@@ -1,250 +0,0 @@
-#########################################################################################
-# [ WARNING ]
-# patroni configuration file maintained by the postgres-operator
-# local changes may be overwritten.
-#########################################################################################
-# For a complete reference of all the options for this configuration file,
-# please refer to https://patroni.readthedocs.io/en/latest/SETTINGS.html.
-
-# Name of the cluster.
-scope: {{ scope }}
-namespace: /db/
-name: {{ member_name }}
-
-log:
-  dir: {{ log_path }}
-  dateformat: "%Y-%m-%d %H:%M:%S %Z"
-  format: "%(asctime)s [%(process)d]: %(levelname)s: %(message)s "
-  file_num: 10080
-  file_size: 600
-
-restapi:
-  listen: '{{ self_ip }}:8008'
-  connect_address: '{{ self_ip }}:8008'
-  {%- if patroni_password %}
-  authentication:
-    username: patroni
-    password: {{ patroni_password }}
-  {%- endif %}
-  cafile: {{ conf_path }}/peer_ca.pem
-  certfile: {{ conf_path }}/peer_cert.pem
-  keyfile: {{ conf_path }}/peer_key.pem
-
-ctl:
-  cacert: {{ conf_path }}/peer_ca.pem
-
-raft:
-  data_dir: {{ conf_path }}/raft
-  self_addr: '{{ self_ip }}:2222'
-  password: {{ raft_password }}
-  {% if partner_addrs or watcher -%}
-  partner_addrs:
-  {% endif -%}
-  {% for partner_addr in partner_addrs -%}
-  - {{ partner_addr }}:2222
-  {% endfor %}
-  {%- if watcher %}
-  - {{ watcher }}
-  {% endif %}
-
-bootstrap:
-  dcs:
-    # Settings related to the time without communication
-    # to the leader before triggering a failover.
-    ttl: 30
-    loop_wait: 10
-    retry_timeout: 10
-    maximum_lag_on_failover: {{ maximum_lag_on_failover }}
-    synchronous_mode: true
-    synchronous_mode_strict: false
-    max_timelines_history: 50
-    synchronous_node_count: {{ synchronous_node_count }}
-    postgresql:
-      use_pg_rewind: true
-      remove_data_directory_on_rewind_failure: false
-      remove_data_directory_on_diverged_timelines: false
-      parameters:
-        synchronous_standby_names: "*"
-        {%- if enable_pgbackrest_archiving %}
-        archive_command: 'pgbackrest {{ pgbackrest_configuration_file }} --stanza={{ stanza }} archive-push %p'
-        {% else %}
-        archive_command: /bin/true
-        {%- endif %}
-        archive_mode: on
-        autovacuum: true
-        debug_print_plan: 'off'
-        debug_print_parse: 'off'
-        debug_print_rewritten: 'off'
-        fsync: true
-        full_page_writes: true
-        lc_messages: 'en_US.UTF8'
-        log_autovacuum_min_duration: 60000
-        log_checkpoints: 'on'
-        log_destination: 'stderr'
-        log_directory: '{{ postgresql_log_path }}'
-        log_error_verbosity: 'verbose'
-        log_file_mode: '0600'
-        log_filename: 'postgresql-%w_%H%M.log'
-        log_hostname: 'off'
-        log_line_prefix: '%t [%p]: user=%u,db=%d,app=%a,client=%h,line=%l '
-        log_min_duration_sample: -1
-        log_min_error_statement: 'warning'
-        log_recovery_conflict_waits: 'on'
-        log_replication_commands: 'on'
-        log_rotation_age: 1
-        log_rotation_size: 0
-        log_statement: ddl
-        log_statement_sample_rate: 1
-        log_statement_stats: 'off'
-        log_temp_files: 1
-        log_timezone: 'UTC'
-        log_truncate_on_rotation: 'on'
-        logging_collector: 'on'
-        wal_level: logical
-        shared_preload_libraries: 'timescaledb,pgaudit,set_user,pg_stat_statements'
-        session_preload_libraries: 'login_hook'
-        set_user.block_log_statement: 'on'
-        set_user.exit_on_error: 'on'
-        set_user.superuser_allowlist: '+charmed_dba'
-    slots:
-    {%- for slot, database in slots.items() %}
-      {{slot}}:
-        database: {{database}}
-        plugin: pgoutput
-        type: logical
-    {%- endfor -%}
-
-  {%- if restoring_backup %}
-  method: pgbackrest
-  pgbackrest:
-    command: >
-      pgbackrest {{ pgbackrest_configuration_file }} --stanza={{ restore_stanza }} --pg1-path={{ data_path }}
-      {%- if backup_id %} --set={{ backup_id }} {%- endif %}
-      {%- if restore_timeline %} --target-timeline="0x{{ restore_timeline }}" {% endif %}
-      {%- if restore_to_latest %} --type=default {%- else %}
-      --target-action=promote {%- if pitr_target %} --target="{{ pitr_target }}" --type=time {%- else %} --type=immediate {%- endif %}
-      {%- endif %}
-      restore
-    no_params: True
-    keep_existing_recovery_conf: True
-  {% elif primary_cluster_endpoint %}
-    standby_cluster:
-      host: {{ primary_cluster_endpoint }}
-      port: 5432
-      create_replica_methods: ["basebackup"]
-  {% else %}
-  initdb:
-  - encoding: UTF8
-  - data-checksums
-  - waldir: {{ wal_dir }}
-  {%- endif %}
-
-postgresql:
-  listen: {% for ip in listen_ips %}{{ ip }}{%- if not loop.last %},{% endif %}{% endfor %}:5432
-  basebackup:
-    - waldir: {{ wal_dir }}
-  connect_address: '{{ self_ip }}:5432'
-  # Path to PostgreSQL binaries used in the database bootstrap process.
-  bin_dir: /snap/charmed-postgresql/current/usr/lib/postgresql/{{ version }}/bin
-  data_dir: {{ data_path }}
-  parameters:
-    shared_preload_libraries: 'timescaledb,pgaudit,set_user,pg_stat_statements'
-    session_preload_libraries: 'login_hook'
-    set_user.block_log_statement: 'on'
-    set_user.exit_on_error: 'on'
-    set_user.superuser_allowlist: '+charmed_dba'
-    {%- if enable_pgbackrest_archiving %}
-    archive_command: 'pgbackrest {{ pgbackrest_configuration_file }} --stanza={{ stanza }} archive-push %p'
-    {% else %}
-    archive_command: /bin/true
-    {%- endif %}
-    {%- if enable_tls %}
-    ssl: on
-    ssl_ca_file: {{ conf_path }}/ca.pem
-    ssl_cert_file: {{ conf_path }}/cert.pem
-    ssl_key_file: {{ conf_path }}/key.pem
-    {%- endif %}
-    temp_tablespaces: temp
-    unix_socket_directories: /tmp
-    pg_stat_statements.max: 10000
-    {%- if pg_parameters %}
-    {%- for key, value in pg_parameters.items() %}
-    {{key}}: {{value}}
-    {%- endfor -%}
-    {% endif %}
-  pgpass: /tmp/pgpass
-  pg_hba:
-    - local all backup peer map=operator
-    - local all operator scram-sha-256
-    - local all monitoring password
-    - {{ 'hostssl' if enable_tls else 'host' }} all +charmed_stats 0.0.0.0/0 scram-sha-256
-    - {{ 'hostssl' if enable_tls else 'host' }} all +charmed_read 0.0.0.0/0 scram-sha-256
-    - {{ 'hostssl' if enable_tls else 'host' }} all +charmed_dml 0.0.0.0/0 scram-sha-256
-    - {{ 'hostssl' if enable_tls else 'host' }} all +charmed_backup 0.0.0.0/0 scram-sha-256
-    - {{ 'hostssl' if enable_tls else 'host' }} all +charmed_dba 0.0.0.0/0 scram-sha-256
-    - {{ 'hostssl' if enable_tls else 'host' }} all +charmed_admin 0.0.0.0/0 scram-sha-256
-    - {{ 'hostssl' if enable_tls else 'host' }} all +charmed_databases_owner 0.0.0.0/0 scram-sha-256
-    {%- if not connectivity %}
-    - {{ 'hostssl' if enable_tls else 'host' }} all all {{ self_ip }}/32 {{ instance_password_encryption }}
-    - {{ 'hostssl' if enable_tls else 'host' }} all all 0.0.0.0/0 reject
-    {%- elif enable_ldap %}
-    - {{ 'hostssl' if enable_tls else 'host' }} all +identity_access 0.0.0.0/0 ldap {{ ldap_parameters }}
-    - {{ 'hostssl' if enable_tls else 'host' }} all +internal_access 0.0.0.0/0 {{ instance_password_encryption }}
-    {%-   for user, databases in user_databases_map.items() %}
-    - {{ 'hostssl' if enable_tls else 'host' }} {{ databases }} {{ user }} 0.0.0.0/0 {{ instance_password_encryption }}
-    {%-   endfor %}
-    {%- else %}
-    - {{ 'hostssl' if enable_tls else 'host' }} all +internal_access 0.0.0.0/0 {{ instance_password_encryption }}
-    {%-   for user, databases in user_databases_map.items() %}
-    {%-     if 'pgbouncer_auth_relation_' in user %}
-    - {{ 'hostssl' if enable_tls else 'host' }} {{ databases }} {{ user }} 0.0.0.0/0 md5
-    {%-     else %}
-    - {{ 'hostssl' if enable_tls else 'host' }} {{ databases }} {{ user }} 0.0.0.0/0 {{ instance_password_encryption }}
-    {%-     endif %}
-    {%-   endfor %}
-    {%- endif %}
-    {%- if watcher_addr %}
-    # Allow watcher to connect for health checks
-    - {{ 'hostssl' if enable_tls else 'host' }} postgres watcher {{ watcher_addr }}/32 scram-sha-256
-    {%- endif %}
-    - {{ 'hostssl' if enable_tls else 'host' }} replication replication 127.0.0.1/32 scram-sha-256
-    # Allow replications connections from other cluster members.
-    {%- for endpoint in extra_replication_endpoints %}
-    - {{ 'hostssl' if enable_tls else 'host' }} replication replication {{ endpoint }}/32 scram-sha-256
-    {%- endfor %}
-    {%- for peer_ip in peers_ips %}
-    - {{ 'hostssl' if enable_tls else 'host' }} replication replication {{ peer_ip }}/0 scram-sha-256
-    {%- endfor %}
-
-  pg_ident:
-  - operator _daemon_ backup
-  authentication:
-    replication:
-      username: replication
-      password: {{ replication_password }}
-      {%- if enable_tls %}
-      sslrootcert: {{ conf_path }}/ca.pem
-      {%- endif %}
-      sslcert: {{ conf_path }}/nonexistent_cert.pem
-      sslkey: {{ conf_path }}/nonexistent_key.pem
-    rewind:
-      username: {{ rewind_user }}
-      password: {{ rewind_password }}
-      {%- if enable_tls %}
-      sslrootcert: {{ conf_path }}/ca.pem
-      {%- endif %}
-      sslcert: {{ conf_path }}/nonexistent_cert.pem
-      sslkey: {{ conf_path }}/nonexistent_key.pem
-    superuser:
-      username: {{ superuser }}
-      password: {{ superuser_password }}
-      {%- if enable_tls %}
-      sslrootcert: {{ conf_path }}/ca.pem
-      {%- endif %}
-      sslcert: {{ conf_path }}/nonexistent_cert.pem
-      sslkey: {{ conf_path }}/nonexistent_key.pem
-use_unix_socket: true
-{%- if is_creating_backup %}
-tags:
-  is_creating_backup: {{ is_creating_backup }}
-{%- endif %}
diff --git a/tests/unit/test_charm.py b/tests/unit/test_charm.py
index 7a9db219e5..981e8877ec 100644
--- a/tests/unit/test_charm.py
+++ b/tests/unit/test_charm.py
@@ -9,8 +9,7 @@
 import platform
 import subprocess
 from datetime import UTC, datetime
-from typing import ClassVar
-from unittest.mock import MagicMock, Mock, PropertyMock, call, mock_open, patch, sentinel
+from unittest.mock import MagicMock, Mock, PropertyMock, call, patch, sentinel
 
 # from ops.framework import EventBase
 import charm_refresh
@@ -188,7 +187,7 @@ def test_on_leader_elected(harness):
         ) as _primary_endpoint,
         patch("charm.PostgresqlOperatorCharm.update_config") as _update_config,
         patch("charm.PostgresqlOperatorCharm._reconfigure_cluster"),
-        patch("charm.TLS.generate_internal_peer_cert"),
+        patch("charm.TLSManager.generate_internal_peer_cert"),
     ):
         # Assert that there is no password in the peer relation.
         assert harness.charm._peers.data[harness.charm.app].get("operator-password", None) is None
@@ -444,7 +443,7 @@ def test_on_start_bootstrap_failure(harness):
         patch("charm.PostgresqlOperatorCharm._ensure_storage_layout"),
         patch("charm.PostgresqlOperatorCharm._check_detached_storage"),
         patch("charm.PostgresqlOperatorCharm.get_secret"),
-        patch("charm.TLS.generate_internal_peer_cert"),
+        patch("charm.TLSManager.generate_internal_peer_cert"),
         patch("charm.PatroniManager.bootstrap_cluster") as _bootstrap_cluster,
         patch("charm.PostgresqlOperatorCharm.update_config"),
         patch("charm.start_raft_observer"),
@@ -475,7 +474,7 @@ def test_on_start_create_user_error(harness):
         patch("charm.PostgresqlOperatorCharm._ensure_storage_layout"),
         patch("charm.PostgresqlOperatorCharm._check_detached_storage"),
         patch("charm.PostgresqlOperatorCharm.get_secret"),
-        patch("charm.TLS.generate_internal_peer_cert"),
+        patch("charm.TLSManager.generate_internal_peer_cert"),
         patch("charm.PatroniManager.bootstrap_cluster") as _bootstrap_cluster,
         patch(
             "charm.PatroniManager.member_started",
@@ -532,7 +531,7 @@ def test_on_start_success(harness):
         patch("charm.PostgresqlOperatorCharm._ensure_storage_layout"),
         patch("charm.PostgresqlOperatorCharm._check_detached_storage"),
         patch("charm.PostgresqlOperatorCharm.get_secret"),
-        patch("charm.TLS.generate_internal_peer_cert"),
+        patch("charm.TLSManager.generate_internal_peer_cert"),
         patch("charm.PatroniManager.bootstrap_cluster") as _bootstrap_cluster,
         patch(
             "charm.PatroniManager.member_started",
@@ -661,7 +660,7 @@ def test_on_start_replica(harness):
             return_value=True,
         ) as _is_storage_attached,
         patch("charm.PostgresqlOperatorCharm.get_secret"),
-        patch("charm.TLS.generate_internal_peer_cert"),
+        patch("charm.TLSManager.generate_internal_peer_cert"),
         patch("charm.start_raft_observer"),
     ):
         _get_postgresql_version.return_value = "16.6"
@@ -718,9 +717,8 @@ def test_on_start_no_patroni_member(harness):
         patch(
             "charm.PostgresqlOperatorCharm._is_storage_attached", return_value=True
         ) as _is_storage_attached,
-        patch("charm.PostgresqlOperatorCharm.get_available_memory") as _get_available_memory,
         patch("charm.PostgresqlOperatorCharm.get_secret"),
-        patch("charm.TLS.generate_internal_peer_cert"),
+        patch("charm.TLSManager.generate_internal_peer_cert"),
         patch("charm.PostgreSQLProvider.get_username_mapping", return_value={}),
         patch("charm.PostgreSQLProvider.get_databases_prefix_mapping", return_value={}),
         patch("charm.start_raft_observer"),
@@ -1351,164 +1349,63 @@ def test_restart(harness):
         mock_event.defer.assert_not_called()
 
 
-def test_update_config(harness):
-    with pathlib.Path("refresh_versions.toml").open("rb") as file:
-        _revision = tomli.load(file)["snap"]["revisions"][platform.machine()]
+def test_request_restart(harness):
+    """Bridge for ConfigManager: clears the restarted flag and acquires the restart lock."""
+    with patch(
+        "charms.rolling_ops.v0.rollingops.RollingOpsManager._on_acquire_lock"
+    ) as _on_acquire_lock:
+        rel_id = harness.model.get_relation(PEER_RELATION).id
+        with harness.hooks_disabled():
+            harness.update_relation_data(
+                rel_id, harness.charm.unit.name, {"postgresql_restarted": "True"}
+            )
+
+        harness.charm.request_restart()
+
+        assert "postgresql_restarted" not in harness.charm.unit_peer_data
+        _on_acquire_lock.assert_called_once()
+
+
+def test_refresh_endpoints(harness):
+    """Bridge for ConfigManager: refreshes the client-relation endpoints."""
+    with patch("charm.PostgreSQLProvider.update_endpoints") as _update_endpoints:
+        harness.charm.refresh_endpoints()
 
-    class _MockSnap:
-        revision = _revision
+        _update_endpoints.assert_called_once()
 
+
+def test_restart_services(harness):
+    """Bridge for ConfigManager: restarts the metrics and LDAP sync snap services."""
     with (
-        patch("subprocess.check_output", return_value=b"C"),
-        patch("charm.snap.SnapCache", lambda: {"charmed-postgresql": _MockSnap()}),
-        patch(
-            "charm.PostgresqlOperatorCharm._handle_postgresql_restart_need"
-        ) as _handle_postgresql_restart_need,
-        patch("charm.PatroniManager.ensure_slots_controller_by_patroni"),
+        patch("charm.snap.SnapCache") as _snap_cache,
         patch(
             "charm.PostgresqlOperatorCharm._restart_metrics_service"
         ) as _restart_metrics_service,
         patch(
             "charm.PostgresqlOperatorCharm._restart_ldap_sync_service"
         ) as _restart_ldap_sync_service,
-        patch("charm.PatroniManager.bulk_update_parameters_controller_by_patroni"),
-        patch("charm.PatroniManager.member_started", new_callable=PropertyMock) as _member_started,
-        patch("charm.VMWorkload.is_patroni_running") as _is_patroni_running,
-        patch("charm.ConfigManager.render_patroni_yml_file") as _render_patroni_yml_file,
-        patch(
-            "charm.PostgresqlOperatorCharm.is_tls_enabled", new_callable=PropertyMock
-        ) as _is_tls_enabled,
-        patch.object(harness.charm, "postgresql", Mock()) as postgresql_mock,
-        patch("charm.PostgresqlOperatorCharm.get_available_memory") as _get_available_memory,
-        patch.object(
-            harness.charm,
-            "_calculate_worker_process_config",
-            return_value={
-                "wal_compression": "on",
-                "max_worker_processes": "8",
-                "max_parallel_workers": "8",
-                "max_parallel_maintenance_workers": "8",
-                "max_logical_replication_workers": "8",
-                "max_sync_workers_per_subscription": "8",
-                "max_parallel_apply_workers_per_subscription": "8",
-            },
-        ),
     ):
-        rel_id = harness.model.get_relation(PEER_RELATION).id
-        # Mock some properties.
-        postgresql_mock.is_tls_enabled = PropertyMock(side_effect=[False, False, False, False])
-        _is_patroni_running.side_effect = [True, True, False, True]
-        _member_started.side_effect = [True, True, False]
-        postgresql_mock.build_postgresql_parameters.return_value = {"test": "test"}
+        postgres_snap = _snap_cache.return_value.__getitem__.return_value
 
-        # Test without TLS files available.
-        with harness.hooks_disabled():
-            harness.update_relation_data(rel_id, harness.charm.unit.name, {"tls": ""})
-        _is_tls_enabled.return_value = False
-        harness.charm.update_config()
-        _render_patroni_yml_file.assert_called_once_with(
-            connectivity=True,
-            is_creating_backup=False,
-            enable_ldap=False,
-            enable_tls=False,
-            backup_id=None,
-            pitr_target=None,
-            restore_timeline=None,
-            restore_to_latest=False,
-            stanza=None,
-            restore_stanza=None,
-            parameters={
-                "test": "test",
-                "wal_compression": "on",
-                "max_worker_processes": "8",
-                "max_parallel_workers": "8",
-                "max_parallel_maintenance_workers": "8",
-                "max_logical_replication_workers": "8",
-                "max_sync_workers_per_subscription": "8",
-                "max_parallel_apply_workers_per_subscription": "8",
-            },
-            user_databases_map={"operator": "all", "replication": "all", "rewind": "all"},
-            ldap_parameters={},
-            async_primary_cluster_endpoint=None,
-            async_partner_addresses=[],
-            async_standby_endpoints=[],
-            watcher_raft_address=None,
-            no_peers=False,
-            # slots={},
-        )
-        _handle_postgresql_restart_need.assert_called_once_with(True)
-        _restart_ldap_sync_service.assert_called_once()
-        _restart_metrics_service.assert_called_once()
-        assert "tls" not in harness.get_relation_data(rel_id, harness.charm.unit.name)
-
-        # Test with TLS files available.
-        _handle_postgresql_restart_need.reset_mock()
-        _restart_ldap_sync_service.reset_mock()
-        _restart_metrics_service.reset_mock()
-        harness.update_relation_data(
-            rel_id, harness.charm.unit.name, {"tls": ""}
-        )  # Mock some data in the relation to test that it change.
-        _is_tls_enabled.return_value = True
-        _render_patroni_yml_file.reset_mock()
-        harness.charm.update_config()
-        _render_patroni_yml_file.assert_called_once_with(
-            connectivity=True,
-            is_creating_backup=False,
-            enable_ldap=False,
-            enable_tls=True,
-            backup_id=None,
-            pitr_target=None,
-            restore_timeline=None,
-            restore_to_latest=False,
-            stanza=None,
-            restore_stanza=None,
-            parameters={
-                "test": "test",
-                "wal_compression": "on",
-                "max_worker_processes": "8",
-                "max_parallel_workers": "8",
-                "max_parallel_maintenance_workers": "8",
-                "max_logical_replication_workers": "8",
-                "max_sync_workers_per_subscription": "8",
-                "max_parallel_apply_workers_per_subscription": "8",
-            },
-            user_databases_map={"operator": "all", "replication": "all", "rewind": "all"},
-            ldap_parameters={},
-            async_primary_cluster_endpoint=None,
-            async_partner_addresses=[],
-            async_standby_endpoints=[],
-            watcher_raft_address=None,
-            no_peers=False,
-            # slots={},
-        )
-        _handle_postgresql_restart_need.assert_called_once()
-        _restart_ldap_sync_service.assert_called_once()
-        _restart_metrics_service.assert_called_once()
-        assert "tls" not in harness.get_relation_data(
-            rel_id, harness.charm.unit.name
-        )  # The "tls" flag is set in handle_postgresql_restart_need.
-
-        # Test with workload not running yet.
-        harness.update_relation_data(
-            rel_id, harness.charm.unit.name, {"tls": ""}
-        )  # Mock some data in the relation to test that it change.
-        _handle_postgresql_restart_need.reset_mock()
-        _restart_ldap_sync_service.reset_mock()
-        _restart_metrics_service.reset_mock()
-        harness.charm.update_config()
-        _handle_postgresql_restart_need.assert_not_called()
-        assert harness.get_relation_data(rel_id, harness.charm.unit.name)["tls"] == "enabled"
-
-        # Test with member not started yet.
-        harness.update_relation_data(
-            rel_id, harness.charm.unit.name, {"tls": ""}
-        )  # Mock some data in the relation to test that it doesn't change.
-        _is_tls_enabled.return_value = False
-        harness.charm.update_config()
-        _handle_postgresql_restart_need.assert_not_called()
-        _restart_ldap_sync_service.assert_not_called()
-        _restart_metrics_service.assert_not_called()
-        assert "tls" not in harness.get_relation_data(rel_id, harness.charm.unit.name)
+        harness.charm.restart_services()
+
+        _restart_metrics_service.assert_called_once_with(postgres_snap)
+        _restart_ldap_sync_service.assert_called_once_with(postgres_snap)
+
+
+def test_update_config_delegates_to_config_manager(harness):
+    """update_config forwards to the lib's ConfigManager with the charm-owned user hash/refresh."""
+    with patch.object(harness.charm, "config_manager") as _config_manager:
+        _config_manager.update_config.return_value = True
+
+        result = harness.charm.update_config(no_peers=True)
+
+        assert result is True
+        _config_manager.update_config.assert_called_once()
+        args, kwargs = _config_manager.update_config.call_args
+        assert args[1] == harness.charm.generate_user_hash
+        assert kwargs["no_peers"] is True
+        assert kwargs["refresh"] is harness.charm.refresh
 
 
 def test_on_cluster_topology_change(harness):
@@ -1648,423 +1545,6 @@ def test_validate_config_options(harness):
         assert str(e.value).startswith(message)
 
 
-def test_calculate_worker_process_config_auto_values(harness):
-    """Test worker process config calculation with 'auto' values."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        # Test with 2 vCPUs (auto should be min(8, 2*2) = 4)
-        _cpu_count.return_value = 2
-
-        with harness.hooks_disabled():
-            harness.update_config({
-                "cpu-max-worker-processes": "auto",
-                "cpu-max-parallel-workers": "auto",
-                "cpu-max-parallel-maintenance-workers": "auto",
-                "cpu-max-logical-replication-workers": "auto",
-                "cpu-max-sync-workers-per-subscription": "auto",
-                "cpu-max-parallel-apply-workers-per-subscription": "auto",
-            })
-
-        result = harness.charm._calculate_worker_process_config()
-
-        # All auto values should resolve to 4 (min(8, 2*2))
-        assert result["max_worker_processes"] == "4"
-        assert result["max_parallel_workers"] == "4"
-        assert result["max_parallel_maintenance_workers"] == "4"
-        assert result["max_logical_replication_workers"] == "4"
-        assert result["max_sync_workers_per_subscription"] == "4"
-        assert result["max_parallel_apply_workers_per_subscription"] == "4"
-
-        # Test with 8 vCPUs (auto should be min(8, 2*8) = 8)
-        _cpu_count.return_value = 8
-        result = harness.charm._calculate_worker_process_config()
-
-        assert result["max_worker_processes"] == "8"
-        assert result["max_parallel_workers"] == "8"
-
-        # Test with 1 vCPU (auto should be min(8, 2*1) = 2)
-        _cpu_count.return_value = 1
-        result = harness.charm._calculate_worker_process_config()
-
-        assert result["max_worker_processes"] == "2"
-
-
-def test_calculate_worker_process_config_numeric_values(harness):
-    """Test worker process config calculation with numeric values."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 4
-
-        with harness.hooks_disabled():
-            harness.update_config({
-                "cpu-max-worker-processes": "10",
-                "cpu-max-parallel-workers": "8",
-                "cpu-max-parallel-maintenance-workers": "8",
-            })
-
-        result = harness.charm._calculate_worker_process_config()
-
-        assert result["max_worker_processes"] == "10"
-        assert result["max_parallel_workers"] == "8"
-        assert result["max_parallel_maintenance_workers"] == "8"
-
-
-def test_calculate_worker_process_config_all_workers_numeric(harness):
-    """Test all worker configs with numeric values within limits (not capped)."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 4  # Cap would be 40
-
-        with harness.hooks_disabled():
-            harness.update_config({
-                "cpu-max-worker-processes": "20",
-                "cpu-max-parallel-workers": "15",
-                "cpu-max-parallel-maintenance-workers": "10",
-                "cpu-max-logical-replication-workers": "12",
-                "cpu-max-sync-workers-per-subscription": "8",
-                "cpu-max-parallel-apply-workers-per-subscription": "8",
-            })
-
-        result = harness.charm._calculate_worker_process_config()
-
-        # All values should pass through unchanged (within cap of 40)
-        assert result["max_worker_processes"] == "20"
-        assert result["max_parallel_workers"] == "15"
-        assert result["max_parallel_maintenance_workers"] == "10"
-        assert result["max_logical_replication_workers"] == "12"
-        assert result["max_sync_workers_per_subscription"] == "8"
-        assert result["max_parallel_apply_workers_per_subscription"] == "8"
-
-
-def test_calculate_worker_process_config_validation_blocking_max_worker_processes(harness):
-    """Test worker process config validation blocks max_worker_processes exceeding values instead of capping."""
-    import pytest
-
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 2  # Cap should be 20
-
-        # Test max_worker_processes exceeding cap - should raise ValueError
-        with harness.hooks_disabled():
-            harness.update_config({"cpu-max-worker-processes": "50"})  # Exceeds cap
-
-        with pytest.raises(
-            ValueError, match="max_worker_processes value 50 exceeds maximum allowed of 20"
-        ):
-            harness.charm._calculate_worker_process_config()
-
-
-def test_calculate_worker_process_config_validation_blocking_max_parallel_workers(harness):
-    """Test worker process config validation blocks max_parallel_workers exceeding values instead of capping."""
-    import pytest
-
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 2  # Cap should be 20
-
-        # Test max_parallel_workers exceeding cap - should raise ValueError
-        with harness.hooks_disabled():
-            harness.update_config({"cpu-max-parallel-workers": "30"})  # Exceeds cap
-
-        with pytest.raises(
-            ValueError, match="max_parallel_workers value 30 exceeds maximum allowed of 20"
-        ):
-            harness.charm._calculate_worker_process_config()
-
-
-def test_calculate_worker_process_config_validation_blocking_valid_value(harness):
-    """Test worker process config validation accepts valid values within cap."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 2  # Cap should be 20
-
-        # Test valid value within cap - should work
-        with harness.hooks_disabled():
-            harness.update_config({"cpu-max-parallel-maintenance-workers": "15"})  # Within cap
-
-        result = harness.charm._calculate_worker_process_config()
-        assert result["max_parallel_maintenance_workers"] == "15"  # Should accept valid value
-
-
-def test_calculate_worker_process_config_edge_cases(harness):
-    """Test worker process config edge cases."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 4
-
-        # Test with no worker config set (but wal_compression has default)
-        with harness.hooks_disabled():
-            harness.update_config(
-                unset=[
-                    "max_worker_processes",
-                    "max_parallel_workers",
-                    "wal_compression",
-                ]
-            )
-        with contextlib.suppress(AttributeError):
-            del harness.charm.config
-
-        result = harness.charm._calculate_worker_process_config()
-        # All worker parameters default to auto, so they get calculated
-        expected = {
-            "max_worker_processes": "8",  # min(8, 2 * 4) = 8
-            "max_parallel_workers": "8",
-            "max_parallel_maintenance_workers": "8",
-            "max_logical_replication_workers": "8",
-            "max_sync_workers_per_subscription": "8",
-            "max_parallel_apply_workers_per_subscription": "8",
-        }
-        assert result == expected
-
-        # Test with only one worker config set
-        with harness.hooks_disabled():
-            harness.update_config({"cpu-max-worker-processes": "10"})
-        with contextlib.suppress(AttributeError):
-            del harness.charm.config
-
-        result = harness.charm._calculate_worker_process_config()
-        # Other worker parameters still get auto-calculated since they default to "auto"
-        expected = {
-            "max_worker_processes": "10",
-            "max_parallel_workers": "10",  # auto = base_max_workers (10)
-            "max_parallel_maintenance_workers": "10",
-            "max_logical_replication_workers": "10",
-            "max_sync_workers_per_subscription": "10",
-            "max_parallel_apply_workers_per_subscription": "10",
-        }
-        assert result == expected
-
-        # Test with very high vCPU count
-        _cpu_count.return_value = 128
-        with harness.hooks_disabled():
-            harness.update_config({"cpu-max-worker-processes": "auto"})
-        with contextlib.suppress(AttributeError):
-            del harness.charm.config
-
-        result = harness.charm._calculate_worker_process_config()
-        # Should be min(8, 2*128) = 8
-        expected = {
-            "max_worker_processes": "8",
-            "max_parallel_workers": "8",
-            "max_parallel_maintenance_workers": "8",
-            "max_logical_replication_workers": "8",
-            "max_sync_workers_per_subscription": "8",
-            "max_parallel_apply_workers_per_subscription": "8",
-        }
-        assert result == expected
-
-
-def test_calculate_worker_process_config_dependencies(harness):
-    """Test that dependent worker configs properly use max_worker_processes as base."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 4
-
-        # Set max_worker_processes to 12, others to auto
-        with harness.hooks_disabled():
-            harness.update_config({
-                "cpu-max-worker-processes": "12",
-                "cpu-max-parallel-workers": "auto",
-                "cpu-max-parallel-maintenance-workers": "auto",
-            })
-
-        result = harness.charm._calculate_worker_process_config()
-
-        # Auto values should use max_worker_processes (12) as base
-        assert result["max_worker_processes"] == "12"
-        assert result["max_parallel_workers"] == "12"
-        assert result["max_parallel_maintenance_workers"] == "12"
-
-
-def test_calculate_worker_process_config_wal_compression(harness):
-    """Test that wal_compression is NOT included in worker process config."""
-    # Test with wal_compression enabled - should NOT be in worker config result
-    with harness.hooks_disabled():
-        harness.update_config({"cpu-wal-compression": True})
-
-    result = harness.charm._calculate_worker_process_config()
-    assert "wal_compression" not in result  # wal_compression is not part of worker configs
-
-    # Test with wal_compression disabled - should NOT be in worker config result
-    with harness.hooks_disabled():
-        harness.update_config({"cpu-wal-compression": False})
-    del harness.charm.config
-
-    result = harness.charm._calculate_worker_process_config()
-    assert "wal_compression" not in result  # wal_compression is not part of worker configs
-
-    # Test with wal_compression not explicitly set - should NOT be in worker config result
-    with harness.hooks_disabled():
-        harness.update_config(unset=["cpu-wal-compression"])
-    del harness.charm.config
-
-    result = harness.charm._calculate_worker_process_config()
-    assert "wal_compression" not in result  # wal_compression is not part of worker configs
-
-
-def test_calculate_worker_process_config_mixed_auto_and_numeric(harness):
-    """Test worker process config with mixed auto and numeric values."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 4
-
-        with harness.hooks_disabled():
-            harness.update_config({
-                "cpu-max-worker-processes": "16",
-                "cpu-max-parallel-workers": "auto",  # Should become 16
-                "cpu-max-parallel-maintenance-workers": "8",  # Explicit value
-                "cpu-max-logical-replication-workers": "auto",  # Should become 16
-            })
-        with contextlib.suppress(AttributeError):
-            del harness.charm.config
-
-        result = harness.charm._calculate_worker_process_config()
-
-        assert result["max_worker_processes"] == "16"
-        assert result["max_parallel_workers"] == "16"  # Auto resolved to base
-        assert result["max_parallel_maintenance_workers"] == "8"  # Explicit value
-        assert result["max_logical_replication_workers"] == "16"  # Auto resolved to base
-
-
-def test_calculate_worker_process_config_zero_cpu_count(harness):
-    """Test worker process config when cpu_count is 0 (edge case)."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 0
-
-        with harness.hooks_disabled():
-            harness.update_config({"cpu-max-worker-processes": "auto"})
-        with contextlib.suppress(AttributeError):
-            del harness.charm.config
-
-        result = harness.charm._calculate_worker_process_config()
-
-        # min(8, 2*0) = 0
-        assert result["max_worker_processes"] == "0"
-
-
-def test_calculate_worker_process_config_all_workers_validation_blocking(harness):
-    """Test that worker configs exceeding limits raise ValueError instead of capping."""
-    import pytest
-
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 2  # Cap = 20
-
-        # Test each exceeding parameter individually (since first exception stops execution)
-        exceeding_configs = [
-            ("cpu-max-worker-processes", "25"),
-            ("cpu-max-parallel-maintenance-workers", "22"),
-            ("cpu-max-logical-replication-workers", "30"),
-            ("cpu-max-sync-workers-per-subscription", "25"),
-            ("cpu-max-parallel-apply-workers-per-subscription", "100"),
-        ]
-
-        for param_name, invalid_value in exceeding_configs:
-            # Reset all configs to auto first, then set the specific one we want to test
-            reset_config = {
-                "cpu-max-worker-processes": "auto",
-                "cpu-max-parallel-workers": "auto",
-                "cpu-max-parallel-maintenance-workers": "auto",
-                "cpu-max-logical-replication-workers": "auto",
-                "cpu-max-sync-workers-per-subscription": "auto",
-                "cpu-max-parallel-apply-workers-per-subscription": "auto",
-            }
-            reset_config[param_name] = invalid_value
-
-            with harness.hooks_disabled():
-                harness.update_config(reset_config)
-
-            with contextlib.suppress(AttributeError):
-                del harness.charm.config
-
-            with pytest.raises(
-                ValueError,
-                match=f"{param_name.replace('-', '_')} value {invalid_value} exceeds maximum allowed of 20",
-            ):
-                harness.charm._calculate_worker_process_config()
-
-        # Test valid value within cap works - need to set max_worker_processes high enough
-        with harness.hooks_disabled():
-            harness.update_config({
-                "cpu-max-worker-processes": "20",  # Set high enough to allow max_parallel_workers = 18
-                "cpu-max-parallel-workers": "18",  # Within both CPU cap (20) and max_worker_processes (20)
-                "cpu-max-parallel-maintenance-workers": "auto",
-                "cpu-max-logical-replication-workers": "auto",
-                "cpu-max-sync-workers-per-subscription": "auto",
-                "cpu-max-parallel-apply-workers-per-subscription": "auto",
-            })
-
-        with contextlib.suppress(AttributeError):
-            del harness.charm.config
-
-        result = harness.charm._calculate_worker_process_config()
-        assert result["max_parallel_workers"] == "18"  # Should accept valid value
-
-
-def test_update_config_integrates_worker_configs(harness):
-    """Test that update_config properly integrates worker process configurations."""
-    with pathlib.Path("refresh_versions.toml").open("rb") as file:
-        _revision = tomli.load(file)["snap"]["revisions"][platform.machine()]
-
-    class _MockSnap:
-        revision = _revision
-
-    with (
-        patch("subprocess.check_output", return_value=b"C"),
-        patch("charm.snap.SnapCache", lambda: {"charmed-postgresql": _MockSnap()}),
-        patch("charm.PostgresqlOperatorCharm._handle_postgresql_restart_need"),
-        patch("charm.PatroniManager.ensure_slots_controller_by_patroni"),
-        patch("charm.PostgresqlOperatorCharm._restart_metrics_service"),
-        patch("charm.PostgresqlOperatorCharm._restart_ldap_sync_service"),
-        patch("charm.PatroniManager.bulk_update_parameters_controller_by_patroni"),
-        patch("charm.PatroniManager.member_started", new_callable=PropertyMock, return_value=True),
-        patch("charm.VMWorkload.is_patroni_running", return_value=True),
-        patch("charm.ConfigManager.render_patroni_yml_file") as _render_patroni_yml_file,
-        patch(
-            "charm.PostgresqlOperatorCharm.is_tls_enabled",
-            new_callable=PropertyMock,
-            return_value=False,
-        ),
-        patch.object(harness.charm, "postgresql", Mock()) as postgresql_mock,
-        patch("charm.PostgresqlOperatorCharm.get_available_memory", return_value=8000000),
-        patch.object(
-            PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock, return_value=4
-        ),
-    ):
-        postgresql_mock.is_tls_enabled = PropertyMock(return_value=False)
-        postgresql_mock.build_postgresql_parameters.return_value = {"shared_buffers": "128MB"}
-
-        # Configure worker processes
-        with harness.hooks_disabled():
-            harness.update_config({
-                "cpu-max-worker-processes": "auto",
-                "cpu-wal-compression": True,
-            })
-
-        harness.charm.update_config()
-
-        # Verify render was called with merged parameters
-        call_args = _render_patroni_yml_file.call_args
-        parameters = call_args.kwargs["parameters"]
-
-        # Should have original params plus our calculated ones
-        assert "shared_buffers" in parameters
-        assert parameters["max_worker_processes"] == "8"  # min(8, 2*4)
-        assert parameters["wal_compression"] == "on"
-
-
 def test_config_validation_invalid_worker_values(harness):
     """Test that pydantic validates worker process config values."""
     # Test invalid string value (not "auto" or a number)
@@ -2123,59 +1603,6 @@ def test_config_validation_valid_worker_values(harness):
         assert harness.charm.config.cpu_max_parallel_workers == 1000
 
 
-def test_update_config_with_none_pg_parameters(harness):
-    """Test update_config when build_postgresql_parameters returns None."""
-    with pathlib.Path("refresh_versions.toml").open("rb") as file:
-        _revision = tomli.load(file)["snap"]["revisions"][platform.machine()]
-
-    class _MockSnap:
-        revision = _revision
-
-    with (
-        patch("subprocess.check_output", return_value=b"C"),
-        patch("charm.snap.SnapCache", lambda: {"charmed-postgresql": _MockSnap()}),
-        patch("charm.PostgresqlOperatorCharm._handle_postgresql_restart_need"),
-        patch("charm.PatroniManager.ensure_slots_controller_by_patroni"),
-        patch("charm.PostgresqlOperatorCharm._restart_metrics_service"),
-        patch("charm.PostgresqlOperatorCharm._restart_ldap_sync_service"),
-        patch("charm.PatroniManager.bulk_update_parameters_controller_by_patroni"),
-        patch("charm.PatroniManager.member_started", new_callable=PropertyMock, return_value=True),
-        patch("charm.VMWorkload.is_patroni_running", return_value=True),
-        patch("charm.ConfigManager.render_patroni_yml_file") as _render_patroni_yml_file,
-        patch(
-            "charm.PostgresqlOperatorCharm.is_tls_enabled",
-            new_callable=PropertyMock,
-            return_value=False,
-        ),
-        patch.object(harness.charm, "postgresql", Mock()) as postgresql_mock,
-        patch("charm.PostgresqlOperatorCharm.get_available_memory", return_value=8000000),
-        patch.object(
-            PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock, return_value=4
-        ),
-    ):
-        postgresql_mock.is_tls_enabled = PropertyMock(return_value=False)
-        # Return None to test the case where build_postgresql_parameters returns None
-        postgresql_mock.build_postgresql_parameters.return_value = None
-
-        # Configure worker processes
-        with harness.hooks_disabled():
-            harness.update_config({
-                "cpu-max-worker-processes": "auto",
-                "cpu-wal-compression": True,
-            })
-
-        harness.charm.update_config()
-
-        # Verify render was called with worker configs only (no pg_parameters)
-        call_args = _render_patroni_yml_file.call_args
-        parameters = call_args.kwargs["parameters"]
-
-        # Should only have worker configs since pg_parameters was None
-        assert parameters["max_worker_processes"] == "8"  # min(8, 2*4)
-        assert parameters["wal_compression"] == "on"
-        assert "shared_buffers" not in parameters  # pg_parameters was None
-
-
 def test_on_peer_relation_changed(harness):
     with (
         patch("charm.snap.SnapCache"),
@@ -2344,7 +1771,7 @@ def test_reconfigure_cluster(harness):
 
 def test_update_certificate(harness):
     with (
-        patch("charm.TLS.get_client_tls_files") as _get_client_tls_files,
+        patch("charm.TLSManager.get_client_tls_files") as _get_client_tls_files,
         patch("charm.TLS.refresh_tls_certificates_event") as _refresh_tls_certificates_event,
     ):
         # If there is no current TLS files, _request_certificate should be called
@@ -2402,41 +1829,6 @@ def test_update_member_ip(harness):
         _update_certificate.assert_called_once()
 
 
-def test_push_tls_files_to_workload(harness):
-    with (
-        patch("charm.PostgresqlOperatorCharm.update_config") as _update_config,
-        patch("charm.render_file") as _render_file,
-        patch("charm.TLS.get_client_tls_files") as _get_client_tls_files,
-        patch("charm.TLS.get_peer_tls_files") as _get_peer_tls_files,
-        patch(
-            "charm.PostgresqlOperatorCharm.get_secret", return_value="internal_ca"
-        ) as _get_secret,
-    ):
-        _get_client_tls_files.side_effect = [
-            ("key", "ca", "cert"),
-            ("key", "ca", None),
-            ("key", None, "cert"),
-            (None, "ca", "cert"),
-        ]
-        _get_peer_tls_files.side_effect = [
-            ("key", "ca", "cert"),
-            ("key", "ca", None),
-            ("key", None, "cert"),
-            (None, "ca", "cert"),
-        ]
-        _update_config.side_effect = [True, False, False, False]
-
-        # Test when all TLS files are available.
-        assert harness.charm.push_tls_files_to_workload()
-        assert _render_file.call_count == 7
-
-        # Test when not all TLS files are available.
-        for _ in range(3):
-            _render_file.reset_mock()
-            assert not (harness.charm.push_tls_files_to_workload())
-            assert _render_file.call_count == 5
-
-
 def test_push_ca_file_into_workload(harness):
     with (
         patch("charm.PostgresqlOperatorCharm.update_config") as _update_config,
@@ -2473,24 +1865,6 @@ def test_clean_ca_file_from_workload(harness):
         _check_call.assert_called_once_with([UPDATE_CERTS_BIN_PATH])
 
 
-def test_get_available_memory(harness):
-    meminfo = (
-        "MemTotal:       16089488 kB"
-        "MemFree:          799284 kB"
-        "MemAvailable:    3926924 kB"
-        "Buffers:          187232 kB"
-        "Cached:          4445936 kB"
-        "SwapCached:       156012 kB"
-        "Active:         11890336 kB"
-    )
-
-    with patch("builtins.open", mock_open(read_data=meminfo)):
-        assert harness.charm.get_available_memory() == 16475635712
-
-    with patch("builtins.open", mock_open(read_data="")):
-        assert harness.charm.get_available_memory() == 0
-
-
 def test_juju_run_exec(harness):
     with (
         patch("charm.ClusterTopologyObserver") as _topology_observer,
@@ -2843,62 +2217,6 @@ def test_migration_from_single_secret(harness, scope, is_leader):
         )
 
 
-def test_handle_postgresql_restart_need(harness):
-    with (
-        patch("charms.rolling_ops.v0.rollingops.RollingOpsManager._on_acquire_lock") as _restart,
-        patch("charm.wait_fixed", return_value=wait_fixed(0)),
-        patch(
-            "charm.PatroniManager.reload_patroni_configuration"
-        ) as _reload_patroni_configuration,
-        patch("charm.PostgresqlOperatorCharm.is_restart_pending") as _is_restart_pending,
-        patch("single_kernel_postgresql.core.state.CharmState.unit_ip"),
-        patch(
-            "charm.PostgresqlOperatorCharm.is_tls_enabled", new_callable=PropertyMock
-        ) as _is_tls_enabled,
-        patch.object(harness.charm, "postgresql", Mock()) as postgresql_mock,
-    ):
-        rel_id = harness.model.get_relation(PEER_RELATION).id
-        for values in itertools.product(
-            [True, False], [True, False], [True, False], [True, False]
-        ):
-            _reload_patroni_configuration.reset_mock()
-            _restart.reset_mock()
-            with harness.hooks_disabled():
-                harness.update_relation_data(rel_id, harness.charm.unit.name, {"tls": ""})
-                harness.update_relation_data(
-                    rel_id,
-                    harness.charm.unit.name,
-                    {"postgresql_restarted": ("True" if values[3] else "")},
-                )
-
-            _is_tls_enabled.return_value = values[0]
-            postgresql_mock.is_tls_enabled.return_value = values[1]
-            _is_restart_pending.return_value = values[2]
-
-            harness.charm._handle_postgresql_restart_need(True)
-            _reload_patroni_configuration.assert_called_once()
-            if values[0]:
-                assert "tls" in harness.get_relation_data(rel_id, harness.charm.unit)
-            else:
-                assert "tls" not in harness.get_relation_data(rel_id, harness.charm.unit)
-
-            if (values[0] != values[1]) or values[2]:
-                assert "postgresql_restarted" not in harness.get_relation_data(
-                    rel_id, harness.charm.unit
-                )
-                _restart.assert_called_once()
-            else:
-                if values[3]:
-                    assert "postgresql_restarted" in harness.get_relation_data(
-                        rel_id, harness.charm.unit
-                    )
-                else:
-                    assert "postgresql_restarted" not in harness.get_relation_data(
-                        rel_id, harness.charm.unit
-                    )
-                _restart.assert_not_called()
-
-
 def test_on_peer_relation_departed(harness):
     with (
         patch(
@@ -3569,382 +2887,3 @@ def test_on_secret_remove(harness):
         event.secret.label = None
         harness.charm._on_secret_remove(event)
         assert not event.remove_revision.called
-
-
-def test_worker_process_validation_minimum_values(harness):
-    """Test that explicit worker process values below 2 are rejected by Pydantic validation."""
-    import pytest
-    from pydantic import ValidationError
-
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 4
-
-        # Test each parameter with value below minimum (2) - should fail at Pydantic level
-        invalid_configs = [
-            ("cpu_max_worker_processes", 1),
-            ("cpu_max_parallel_workers", 0),
-            ("cpu_max_parallel_maintenance_workers", 1),
-            ("cpu_max_logical_replication_workers", 0),
-            ("cpu_max_sync_workers_per_subscription", 1),
-            ("cpu_max_parallel_apply_workers_per_subscription", 0),
-        ]
-
-        for param_name, invalid_value in invalid_configs:
-            with harness.hooks_disabled():
-                harness.update_config({param_name.replace("_", "-"): str(invalid_value)})
-
-            with pytest.raises(
-                ValidationError,
-                match=r"Input should be greater than or equal to 2",
-            ):
-                harness.charm._calculate_worker_process_config()
-
-
-def test_worker_process_validation_maximum_values(harness):
-    """Test that explicit worker process values above 10*vCores are rejected."""
-    import pytest
-
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 2  # So max allowed is 10*2 = 20
-
-        # Test each parameter with value above maximum (20)
-        invalid_configs = [
-            ("cpu_max_worker_processes", 25),
-            ("cpu_max_parallel_workers", 30),
-            ("cpu_max_parallel_maintenance_workers", 21),
-            ("cpu_max_logical_replication_workers", 50),
-            ("cpu_max_sync_workers_per_subscription", 100),
-            ("cpu_max_parallel_apply_workers_per_subscription", 22),
-        ]
-
-        for param_name, invalid_value in invalid_configs:
-            # Reset all configs to auto first, then set the specific one we want to test
-            reset_config = {
-                "cpu-max-worker-processes": "auto",
-                "cpu-max-parallel-workers": "auto",
-                "cpu-max-parallel-maintenance-workers": "auto",
-                "cpu-max-logical-replication-workers": "auto",
-                "cpu-max-sync-workers-per-subscription": "auto",
-                "cpu-max-parallel-apply-workers-per-subscription": "auto",
-            }
-            reset_config[param_name.replace("_", "-")] = str(invalid_value)
-
-            with harness.hooks_disabled():
-                harness.update_config(reset_config)
-
-            # Clear cached config property to ensure fresh validation
-            if hasattr(harness.charm, "config"):
-                del harness.charm.config
-
-            with pytest.raises(ValueError, match=f"value {invalid_value} exceeds maximum allowed"):
-                harness.charm._calculate_worker_process_config()
-
-
-def test_worker_process_validation_valid_range(harness):
-    """Test that valid worker process values (2 <= value <= 10*vCores) are accepted."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 4  # So valid range is 2-40
-
-        # Test valid values
-        valid_configs = {
-            "cpu-max-worker-processes": "16",
-            "cpu-max-parallel-workers": "8",
-            "cpu-max-parallel-maintenance-workers": "12",
-            "cpu-max-logical-replication-workers": "20",
-            "cpu-max-sync-workers-per-subscription": "40",
-            "cpu-max-parallel-apply-workers-per-subscription": "24",
-        }
-
-        with harness.hooks_disabled():
-            harness.update_config(valid_configs)
-        result = harness.charm._calculate_worker_process_config()
-
-        # All should accept exact values specified
-        assert result["max_worker_processes"] == "16"
-        assert result["max_parallel_workers"] == "8"
-        assert result["max_parallel_maintenance_workers"] == "12"
-        assert result["max_logical_replication_workers"] == "20"
-        assert result["max_sync_workers_per_subscription"] == "40"
-        assert result["max_parallel_apply_workers_per_subscription"] == "24"
-
-
-def test_worker_process_auto_mode_allows_values_below_2(harness):
-    """Test that auto mode can produce values below 2 (unlike explicit values)."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 1  # So auto = min(8, 2*1) = 2
-
-        with harness.hooks_disabled():
-            harness.update_config({
-                "cpu-max-worker-processes": "auto",
-                "cpu-max-parallel-workers": "auto",
-                "cpu-max-parallel-maintenance-workers": "auto",
-                "cpu-max-logical-replication-workers": "auto",
-                "cpu-max-sync-workers-per-subscription": "auto",
-                "cpu-max-parallel-apply-workers-per-subscription": "auto",
-            })
-
-        result = harness.charm._calculate_worker_process_config()
-
-        # Auto mode should allow values below 2
-        assert result["max_worker_processes"] == "2"
-        assert result["max_parallel_workers"] == "2"
-        assert result["max_parallel_maintenance_workers"] == "2"
-        assert result["max_logical_replication_workers"] == "2"
-        assert result["max_sync_workers_per_subscription"] == "2"
-        assert result["max_parallel_apply_workers_per_subscription"] == "2"
-
-
-def test_validate_worker_config_value_success(harness):
-    """Test successful validation of worker config values."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 4  # vCores = 4, so max = 40
-
-        # Test successful validation
-        result = harness.charm._validate_worker_config_value("test_param", 8)
-        assert result == "8"
-
-        # Test minimum valid value
-        result = harness.charm._validate_worker_config_value("test_param", 2)
-        assert result == "2"
-
-        # Test maximum valid value
-        result = harness.charm._validate_worker_config_value("test_param", 40)
-        assert result == "40"
-
-
-def test_validate_worker_config_value_exceeds_cpu_cap(harness):
-    """Test validation when value exceeds CPU cap."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 2  # vCores = 2, so max = 20
-
-        # Test value exceeding CPU cap
-        with pytest.raises(ValueError, match="value 25 exceeds maximum allowed of 20"):
-            harness.charm._validate_worker_config_value("test_param", 25)
-
-
-def test_calculate_max_parallel_workers_with_constraint(harness):
-    """Test max_parallel_workers constrained by base_max_workers."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 4  # vCores = 4
-
-        # Set a high value for max_parallel_workers but low base_max_workers
-        with harness.hooks_disabled():
-            harness.update_config({"cpu-max-parallel-workers": "20"})  # High value
-
-        # Should be constrained by base_max_workers (which will be 8 from base)
-        base_max_workers = 6
-        result = harness.charm._calculate_max_parallel_workers(base_max_workers)
-        # Should be min(20, 6) = 6
-        assert result == "6"
-
-
-def test_calculate_worker_process_config_partial_configs(harness):
-    """Test worker config calculation with only some configs set."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 4
-
-        # Set only some worker configs, others remain at default "auto"
-        config = {
-            "cpu-max-worker-processes": "12",
-            "cpu-max-parallel-workers": "auto",  # Should use base (12)
-            "cpu-max-parallel-maintenance-workers": "8",
-            # Others remain as "auto"
-        }
-
-        with harness.hooks_disabled():
-            harness.update_config(config)
-        result = harness.charm._calculate_worker_process_config()
-
-        assert result["max_worker_processes"] == "12"
-        assert result["max_parallel_workers"] == "12"  # auto = base_max_workers
-        assert result["max_parallel_maintenance_workers"] == "8"
-        assert result["max_logical_replication_workers"] == "12"  # auto = base
-        assert result["max_sync_workers_per_subscription"] == "12"  # auto = base
-        assert result["max_parallel_apply_workers_per_subscription"] == "12"  # auto = base
-
-
-def test_calculate_worker_process_config_all_none(harness):
-    """Test worker config when all worker configs are None."""
-    # Mock the config to return None for all worker process configs
-    with patch.object(harness.charm, "config") as mock_config:
-        mock_config.cpu_max_worker_processes = None
-        mock_config.cpu_max_parallel_workers = None
-        mock_config.cpu_max_parallel_maintenance_workers = None
-        mock_config.cpu_max_logical_replication_workers = None
-        mock_config.cpu_max_sync_workers_per_subscription = None
-        mock_config.cpu_max_parallel_apply_workers_per_subscription = None
-
-        result = harness.charm._calculate_worker_process_config()
-
-        # When all are None, should return empty dict
-        assert result == {}
-
-
-def test_api_update_config_with_worker_processes(harness):
-    """Test _api_update_config includes worker process configurations."""
-    with (
-        patch.object(
-            PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-        ) as _cpu_count,
-        patch.object(harness.charm, "patroni_manager") as mock_patroni_manager,
-        patch.object(
-            harness.charm.async_replication, "get_primary_cluster_endpoint"
-        ) as mock_endpoint,
-    ):
-        _cpu_count.return_value = 4
-        mock_endpoint.return_value = None  # No primary endpoint
-
-        # Set worker process configs
-        config = {
-            "cpu-max-worker-processes": "8",
-            "cpu-max-logical-replication-workers": "4",
-        }
-
-        with harness.hooks_disabled():
-            harness.update_config(config)
-
-        harness.charm._api_update_config()
-
-        # Verify that bulk_update_parameters_controller_by_patroni was called
-        assert mock_patroni_manager.bulk_update_parameters_controller_by_patroni.called
-
-        # Get the arguments passed to the mock
-        call_args = mock_patroni_manager.bulk_update_parameters_controller_by_patroni.call_args
-        cfg_patch = call_args[0][0]  # First argument
-
-        # Verify worker configs are included
-        assert "max_worker_processes" in cfg_patch
-        assert "max_logical_replication_workers" in cfg_patch
-        assert cfg_patch["max_worker_processes"] == "8"
-        assert cfg_patch["max_logical_replication_workers"] == "4"
-
-
-def test_api_update_config_with_experimental_max_connections(harness):
-    """Test _api_update_config with experimental_max_connections set."""
-    with (
-        patch.object(
-            PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-        ) as _cpu_count,
-        patch.object(harness.charm, "patroni_manager") as mock_patroni_manager,
-        patch.object(
-            harness.charm.async_replication, "get_primary_cluster_endpoint"
-        ) as mock_endpoint,
-    ):
-        _cpu_count.return_value = 4
-        mock_endpoint.return_value = None
-
-        # Set experimental_max_connections
-        with harness.hooks_disabled():
-            harness.update_config({"experimental-max-connections": 500})
-
-        harness.charm._api_update_config()
-
-        # Get the cfg_patch argument
-        call_args = mock_patroni_manager.bulk_update_parameters_controller_by_patroni.call_args
-        cfg_patch = call_args[0][0]
-
-        # Should use experimental value instead of calculated value
-        assert cfg_patch["max_connections"] == 500
-
-
-def test_api_update_config_with_async_replication_primary(harness):
-    """Test _api_update_config when primary endpoint is available."""
-    with (
-        patch.object(harness.charm, "patroni_manager") as mock_patroni_manager,
-        patch.object(
-            harness.charm.async_replication, "get_primary_cluster_endpoint"
-        ) as mock_endpoint,
-    ):
-        # Mock primary endpoint
-        mock_endpoint.return_value = "primary-host:5432"
-
-        harness.charm._api_update_config()
-
-        # Get the arguments
-        call_args = mock_patroni_manager.bulk_update_parameters_controller_by_patroni.call_args
-        base_patch = call_args[0][1]
-
-        # Verify standby_cluster is configured
-        assert "standby_cluster" in base_patch
-        assert base_patch["standby_cluster"]["host"] == "primary-host:5432"
-
-
-def test_calculate_max_logical_replication_workers_with_base_value(harness):
-    """Test max_logical_replication_workers uses base_max_workers consistently."""
-    with patch.object(
-        PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-    ) as _cpu_count:
-        _cpu_count.return_value = 3  # vCores = 3
-
-        # Test auto mode with specific base_max_workers value
-        with harness.hooks_disabled():
-            harness.update_config({"cpu-max-logical-replication-workers": "auto"})
-
-        # Should use base_max_workers directly (like other worker params)
-        result = harness.charm._calculate_max_logical_replication_workers(10)
-        assert result == "10"
-
-
-def test_update_config_integrates_all_worker_configs(harness):
-    """Test that update_config properly integrates all worker process configs."""
-
-    class _MockSnap:
-        present = True
-        services: ClassVar = {"patroni": {"active": True}}
-
-    with (
-        patch.object(
-            PostgresqlOperatorCharm, "cpu_count", new_callable=PropertyMock
-        ) as _cpu_count,
-        patch.object(harness.charm, "postgresql") as mock_postgresql,
-        patch.object(harness.charm, "get_available_memory") as mock_memory,
-        patch("charm.snap.SnapCache", lambda: {"charmed-postgresql": _MockSnap()}),
-    ):
-        _cpu_count.return_value = 4
-        mock_memory.return_value = 1024 * 1024 * 1024  # 1GB
-
-        # Mock postgresql.build_postgresql_parameters to return a dict
-        mock_postgresql.build_postgresql_parameters.return_value = {
-            "shared_buffers": "128MB",
-        }
-
-        # Set comprehensive worker config
-        config = {
-            "cpu-max-worker-processes": "16",
-            "cpu-max-parallel-workers": "8",
-            "cpu-max-parallel-maintenance-workers": "4",
-            "cpu-max-logical-replication-workers": "12",
-            "cpu-max-sync-workers-per-subscription": "6",
-            "cpu-max-parallel-apply-workers-per-subscription": "10",
-            "cpu-wal-compression": True,
-        }
-
-        with harness.hooks_disabled():
-            harness.update_config(config)
-
-        # Test the worker config calculation without calling the full update_config
-        worker_configs = harness.charm._calculate_worker_process_config()
-
-        # Verify all worker configs are calculated properly
-        assert worker_configs["max_worker_processes"] == "16"
-        assert worker_configs["max_parallel_workers"] == "8"
-        assert worker_configs["max_parallel_maintenance_workers"] == "4"
-        assert worker_configs["max_logical_replication_workers"] == "12"
-        assert worker_configs["max_sync_workers_per_subscription"] == "6"
-        assert worker_configs["max_parallel_apply_workers_per_subscription"] == "10"
diff --git a/tests/unit/test_tls.py b/tests/unit/test_tls.py
index c8ce927080..b5197c8efa 100644
--- a/tests/unit/test_tls.py
+++ b/tests/unit/test_tls.py
@@ -1,172 +1,152 @@
 # Copyright 2022 Canonical Ltd.
 # See LICENSE file for licensing details.
-from datetime import timedelta
-from unittest.mock import Mock, patch, sentinel
+"""TLS wiring tests for the single-kernel (lib) TLS handler.
+
+The charm consumes the library TLS handler (single_kernel_postgresql.events.tls.TLS)
+and TLS manager (single_kernel_postgresql.managers.tls.TLSManager) instead of its
+own removed src/relations/tls.py. These tests exercise the lib-backed wiring:
+state-backed cert storage via TLSManager, and the charm's reload bridge that calls
+update_config after the handler stores+pushes certificates.
+"""
+
+from unittest.mock import Mock, PropertyMock, patch
 
 import pytest
 from ops.testing import Harness
 from single_kernel_postgresql.config.literals import PEER_RELATION
+from single_kernel_postgresql.events.tls import TLS
+from single_kernel_postgresql.managers.tls import TLSManager
 
 from charm import PostgresqlOperatorCharm
 
 
 @pytest.fixture(autouse=True)
 def harness():
-    with patch("relations.tls.socket") as _socket:
-        _socket.getfqdn.return_value = "fqdn"
-        harness = Harness(PostgresqlOperatorCharm)
-
-        # Set up the initial relation and hooks.
-        peer_rel_id = harness.add_relation(PEER_RELATION, "postgresql")
-        harness.add_relation_unit(peer_rel_id, "postgresql/0")
-        harness.begin()
-        yield harness
-        harness.cleanup()
-
-
-def test_generate_internal_peer_cert(harness):
+    harness = Harness(PostgresqlOperatorCharm)
+    peer_rel_id = harness.add_relation(PEER_RELATION, "postgresql")
+    harness.add_relation_unit(peer_rel_id, "postgresql/0")
+    harness.begin()
+    yield harness
+    harness.cleanup()
+
+
+def test_tls_handler_is_lib_backed(harness):
+    """The charm wires the lib TLS handler + manager (not the removed relations.tls)."""
+    charm = harness.charm
+    assert isinstance(charm.tls, TLS)
+    assert isinstance(charm.tls_manager, TLSManager)
+    # The handler owns the operator client/peer requirers and the refresh event.
+    assert hasattr(charm.tls, "client_certificate")
+    assert hasattr(charm.tls, "peer_certificate")
+    assert hasattr(charm.tls, "refresh_tls_certificates_event")
+    # The removed method must not resurface anywhere.
+    assert not hasattr(charm, "push_tls_files_to_workload")
+
+
+def test_is_tls_enabled_reflects_tls_manager(harness):
+    """is_tls_enabled is driven by TLSManager.get_client_tls_files(), not the handler."""
+    with patch("charm.TLSManager.get_client_tls_files") as _get_client_tls_files:
+        _get_client_tls_files.return_value = (None, None, None)
+        assert harness.charm.is_tls_enabled is False
+
+        _get_client_tls_files.return_value = ("key", "ca", "cert")
+        assert harness.charm.is_tls_enabled is True
+
+
+def test_reload_bridge_calls_update_config(harness):
+    """_reload_tls_after_push delegates to update_config when internal-ca is present."""
+    with harness.hooks_disabled():
+        harness.set_leader(True)
+        harness.charm.set_secret("app", "internal-ca", "ca-content")
+
+    with patch("charm.PostgresqlOperatorCharm.update_config") as _update_config:
+        harness.charm._reload_tls_after_push(Mock())
+        _update_config.assert_called_once_with()
+
+
+def test_reload_bridge_skips_update_config_without_internal_ca(harness):
+    """_reload_tls_after_push is a no-op when internal-ca is absent (defer path).
+
+    The lib TLS handler defers its push when the internal CA isn't present yet
+    (no files written to disk).  The bridge must not call update_config in that
+    case, or it would render ssl:on against TLS files that don't exist yet.
+    """
+    with patch("charm.PostgresqlOperatorCharm.update_config") as _update_config:
+        harness.charm._reload_tls_after_push(Mock())
+        _update_config.assert_not_called()
+
+
+def test_reload_bridge_defers_when_update_config_not_ready(harness):
+    """A transient config-apply failure must defer (retry), not leave stale TLS state."""
+    with harness.hooks_disabled():
+        harness.set_leader(True)
+        harness.charm.set_secret("app", "internal-ca", "ca-content")
+
+    event = Mock()
+    with patch("charm.PostgresqlOperatorCharm.update_config", return_value=False):
+        harness.charm._reload_tls_after_push(event)
+    event.defer.assert_called_once()
+
+
+def test_reload_bridge_defers_when_tls_files_not_yet_on_disk(harness):
+    """ssl:on must not be rendered until the lib has pushed the cert files to disk."""
+    with harness.hooks_disabled():
+        harness.set_leader(True)
+        harness.charm.set_secret("app", "internal-ca", "ca-content")
+    event = Mock()
     with (
-        patch(
-            "relations.tls.generate_private_key", return_value=sentinel.cert_key
-        ) as _generate_private_key,
-        patch("relations.tls.generate_csr", return_value=sentinel.cert_csr) as _generate_csr,
-        patch(
-            "relations.tls.generate_certificate", return_value=sentinel.cert
-        ) as _generate_certificate,
-        patch("relations.tls.PrivateKey") as _private_key,
-        patch("relations.tls.Certificate") as _certificate,
-        patch("charm.PostgresqlOperatorCharm.set_secret") as _set_secret,
-        patch("charm.PostgresqlOperatorCharm.get_secret", return_value="secret value"),
-        patch(
-            "charm.PostgresqlOperatorCharm.push_tls_files_to_workload"
-        ) as _push_tls_files_to_workload,
+        patch("charm.PostgresqlOperatorCharm.update_config") as _uc,
+        # is_tls_enabled -> True via the live-fetch getter
+        patch("charm.TLSManager.get_client_tls_files", return_value=("K", "CA", "C")),
+        patch.object(harness.charm.tls_manager, "client_tls_files_on_disk", return_value=False),
     ):
-        _private_key.from_string.return_value = sentinel.ca_key
-        _certificate.from_string.return_value = sentinel.ca_cert
-
-        harness.charm.tls.generate_internal_peer_cert()
-
-        _generate_csr.assert_called_once_with(
-            sentinel.cert_key,
-            common_name="postgresql-0",
-            sans_ip=frozenset(),
-            sans_dns=frozenset({"postgresql-0", "fqdn"}),
-        )
-        _generate_certificate.assert_called_once_with(
-            sentinel.cert_csr, sentinel.ca_cert, sentinel.ca_key, validity=timedelta(days=7300)
-        )
-        assert _set_secret.call_count == 2
-        _set_secret.assert_any_call("unit", "internal-key", str(sentinel.cert_key))
-        _set_secret.assert_any_call("unit", "internal-cert", str(sentinel.cert))
-        _push_tls_files_to_workload.assert_called_once_with()
-
-
-def test_get_client_tls_files(harness):
-    with patch(
-        "relations.tls.TLSCertificatesRequiresV4.get_assigned_certificates"
-    ) as _get_assigned_certificates:
-        cert_mock = Mock()
-        cert_mock.certificate = sentinel.certificate
-        cert_mock.ca = sentinel.ca
-        _get_assigned_certificates.return_value = ([cert_mock], sentinel.private_key)
+        harness.charm._reload_tls_after_push(event)
+    event.defer.assert_called_once()
+    _uc.assert_not_called()
 
-        assert harness.charm.tls.get_client_tls_files() == (
-            "sentinel.private_key",
-            "sentinel.ca",
-            "sentinel.certificate",
-        )
 
-        _get_assigned_certificates.return_value = (None, None)
-        assert harness.charm.tls.get_client_tls_files() == (None, None, None)
+def test_reload_bridge_defers_when_update_config_raises(harness):
+    """A raising update_config must be caught and deferred, not fail the hook.
 
+    The bridge mirrors the original charm's broad push-failure guard: a transient
+    Patroni/render failure defers and retries rather than propagating out of the
+    observer.
+    """
+    with harness.hooks_disabled():
+        harness.set_leader(True)
+        harness.charm.set_secret("app", "internal-ca", "ca-content")
 
-def test_get_peer_tls_files(harness):
-    with (
-        patch(
-            "relations.tls.TLSCertificatesRequiresV4.get_assigned_certificates"
-        ) as _get_assigned_certificates,
-        patch(
-            "charm.PostgresqlOperatorCharm.get_secret", return_value=sentinel.secret
-        ) as _get_secret,
-    ):
-        cert_mock = Mock()
-        cert_mock.certificate = sentinel.certificate
-        cert_mock.ca = sentinel.ca
-        _get_assigned_certificates.return_value = ([cert_mock], sentinel.private_key)
-
-        assert harness.charm.tls.get_peer_tls_files() == (
-            "sentinel.private_key",
-            "sentinel.ca",
-            "sentinel.certificate",
-        )
-        assert not _get_secret.called
-
-        _get_assigned_certificates.return_value = (None, None)
-        assert harness.charm.tls.get_peer_tls_files() == (
-            sentinel.secret,
-            sentinel.secret,
-            sentinel.secret,
-        )
-        assert _get_secret.call_count == 3
-        _get_secret.assert_any_call("unit", "internal-key")
-        _get_secret.assert_any_call("unit", "internal-cert")
-        _get_secret.assert_any_call("app", "internal-ca")
-
-
-def test_on_client_certificate_available(harness):
-    with (
-        patch("charm.PostgresqlOperatorCharm.get_secret") as _get_secret,
-        patch(
-            "charm.PostgresqlOperatorCharm.push_tls_files_to_workload"
-        ) as _push_tls_files_to_workload,
+    event = Mock()
+    with patch(
+        "charm.PostgresqlOperatorCharm.update_config",
+        side_effect=RuntimeError("patroni render failed"),
     ):
-        # Defers if internal CA is not ready yet
-        _get_secret.return_value = None
-        event_mock = Mock()
-
-        harness.charm.tls._on_certificate_available(event_mock)
-
-        event_mock.defer.assert_called_once_with()
-        assert not _push_tls_files_to_workload.called
-        event_mock.reset_mock()
+        # Must not raise.
+        harness.charm._reload_tls_after_push(event)
+    event.defer.assert_called_once_with()
 
-        # Defers if can't push
-        _get_secret.return_value = sentinel.secret
-        _push_tls_files_to_workload.return_value = False
 
-        harness.charm.tls._on_certificate_available(event_mock)
-
-        event_mock.defer.assert_called_once_with()
-        _push_tls_files_to_workload.assert_called_once_with()
-        event_mock.reset_mock()
-
-
-def test_on_peer_certificate_available(harness):
+def test_internal_cert_path_pushes_and_reloads(harness):
+    """_check_and_update_internal_cert generates, pushes, and reloads on CN mismatch."""
     with (
+        patch("charm.CharmState.unit_ip", new_callable=PropertyMock) as _unit_ip,
         patch(
-            "relations.tls.TLSCertificatesRequiresV4.get_assigned_certificates"
-        ) as _get_assigned_certificates,
-        patch("relations.tls.TLS._on_certificate_available") as _on_certificate_available,
-        patch("charm.PostgresqlOperatorCharm.get_secret") as _get_secret,
-        patch("charm.PostgresqlOperatorCharm.set_secret") as _set_secret,
+            "charm.PostgresqlOperatorCharm.get_secret",
+            return_value="-----BEGIN CERTIFICATE-----",
+        ),
+        patch("charm.load_pem_x509_certificate") as _load_cert,
+        patch("charm.TLSManager.generate_internal_peer_cert") as _generate,
+        patch("charm.TLSManager.push_tls_files") as _push,
+        patch("charm.PostgresqlOperatorCharm.update_config") as _update_config,
     ):
-        # Same ca
-        cert_mock = Mock()
-        cert_mock.ca = sentinel.ca
-        _get_assigned_certificates.return_value = ([cert_mock], None)
-        _get_secret.return_value = "sentinel.ca"
-        event_mock = Mock()
-
-        harness.charm.tls._on_peer_certificate_available(event_mock)
-
-        _on_certificate_available.assert_called_once_with(event_mock)
-        assert not _set_secret.called
-
-        # Different ca
-        _get_secret.return_value = sentinel.old_ca
+        _unit_ip.return_value = "1.2.3.4"
+        # Make the cert CN differ from the unit IP to force regeneration.
+        attr = Mock()
+        attr.value = "9.9.9.9"
+        _load_cert.return_value.subject.get_attributes_for_oid.return_value = [attr]
 
-        harness.charm.tls._on_peer_certificate_available(event_mock)
+        harness.charm._check_and_update_internal_cert()
 
-        assert _set_secret.call_count == 2
-        _set_secret.assert_any_call("unit", "current-ca", "sentinel.ca")
-        _set_secret.assert_any_call("unit", "old-ca", sentinel.old_ca)
+        _generate.assert_called_once_with()
+        _push.assert_called_once_with()
+        _update_config.assert_called_once_with()