diff --git a/homeassistant/components/recorder/core.py b/homeassistant/components/recorder/core.py
index 6ba64d4a5717e1..ebc2f71795f840 100644
--- a/homeassistant/components/recorder/core.py
+++ b/homeassistant/components/recorder/core.py
@@ -1236,9 +1236,31 @@ def _handle_sqlite_corruption(self, setup_run: bool) -> None:
         move_away_broken_database(dburl_to_path(self.db_url))
         self.recorder_runs_manager.reset()
         self._setup_recorder()
+        self._notify_db_corruption()
+
         if setup_run:
             self._setup_run()
 
+    def _notify_db_corruption(self) -> None:
+        """Notify users of SQLite database corruption.
+
+        Create a persistent notification and fire a 'recorder_database_corrupt' event to inform
+        users that their recorder database has been rebuilt due to database corruption.
+        """
+        persistent_notification.create(
+            self.hass,
+            (
+                "Corruption was detected in the recorder SQLite database and a"
+                " new database has been created. See"
+                " [this page](https://www.home-assistant.io/integrations/recorder/#handling-disk-corruption-and-hardware-failures) for more information."
+            ),
+            "Database corrupt",
+            "recorder_database_corrupt",
+        )
+
+        # Send an event to the bus
+        self.hass.bus.fire("recorder_database_corrupt")
+
     def _close_event_session(self) -> None:
         """Close the event session."""
         self.states_manager.reset()
diff --git a/homeassistant/components/recorder/strings.json b/homeassistant/components/recorder/strings.json
index 2ded6be58d6c05..3a3f3c548a5535 100644
--- a/homeassistant/components/recorder/strings.json
+++ b/homeassistant/components/recorder/strings.json
@@ -20,6 +20,17 @@
     "sqlite_too_old": {
       "title": "Update SQLite to {min_version} or later to continue using the recorder",
       "description": "Support for version {server_version} of SQLite is ending; the minimum supported version is {min_version}. Please upgrade your database software."
+    },
+    "database_corrupt": {
+      "title": "SQLite database corruption detected",
+      "fix_flow": {
+        "step": {
+          "confirm": {
+            "title": "[%key:component::recorder::issues::database_corrupt::title%]",
+            "description": "Corruption was detected in the SQLite database file. The recorder database has been re-created, and no further action is required."
+          }
+        }
+      }
     }
   },
   "services": {
diff --git a/tests/components/recorder/test_init.py b/tests/components/recorder/test_init.py
index d16712e0c7064b..a28122e321b9f1 100644
--- a/tests/components/recorder/test_init.py
+++ b/tests/components/recorder/test_init.py
@@ -1687,8 +1687,18 @@ async def test_database_corruption_while_running(
     recorder_mock: Recorder,
     recorder_db_url: str,
     caplog: pytest.LogCaptureFixture,
+    issue_registry: ir.IssueRegistry,
 ) -> None:
     """Test we can recover from sqlite3 db corruption."""
+    # Setup listener for DB corruption event
+    corruption_events = 0
+
+    def handle_corruption_event(event):
+        nonlocal corruption_events
+        corruption_events += 1
+
+    hass.bus.async_listen("recorder_database_corrupt", handle_corruption_event)
+
     await hass.async_block_till_done()
     caplog.clear()
 
@@ -1703,10 +1713,15 @@ async def test_database_corruption_while_running(
     )
 
     await async_wait_recording_done(hass)
-    with patch.object(
-        get_instance(hass).event_session,
-        "close",
-        side_effect=OperationalError("statement", {}, []),
+    with (
+        patch.object(
+            get_instance(hass).event_session,
+            "close",
+            side_effect=OperationalError("statement", {}, []),
+        ),
+        patch(
+            "homeassistant.components.recorder.core.persistent_notification.create"
+        ) as notify_create_mock,
     ):
         await async_wait_recording_done(hass)
         test_db_file = recorder_db_url.removeprefix("sqlite:///")
@@ -1728,6 +1743,16 @@ async def test_database_corruption_while_running(
     assert "The system will rename the corrupt database file" in caplog.text
     assert "Connected to recorder database" in caplog.text
 
+    # Check that we tried to create a persistent notification
+    assert notify_create_mock.call_count == 1
+    assert (
+        "Corruption was detected in the recorder SQLite database and a new database has been create"
+        in notify_create_mock.call_args.args[1]
+    )
+
+    # Check the DB corruption event was fired
+    assert corruption_events == 1
+
     # This state should go into the new database
     hass.states.async_set("test.two", "on", {})
     await async_wait_recording_done(hass)