diff --git a/src/sentry/feedback/usecases/create_feedback.py b/src/sentry/feedback/usecases/create_feedback.py
index 70e5afa4e707cf..2f4e26b5a9b9f6 100644
--- a/src/sentry/feedback/usecases/create_feedback.py
+++ b/src/sentry/feedback/usecases/create_feedback.py
@@ -283,7 +283,15 @@ def should_filter_feedback(
     return False, None
 
 
-def create_feedback_issue(event, project_id: int, source: FeedbackCreationSource):
+def create_feedback_issue(
+    event: dict[str, Any], project_id: int, source: FeedbackCreationSource
+) -> dict[str, Any] | None:
+    """
+    Produces a feedback issue occurrence to kafka for issues processing. Applies filters, spam filters, and event validation.
+
+    Returns the formatted event data that was sent to issue platform.
+    """
+
     metrics.incr(
         "feedback.create_feedback_issue.entered",
         tags={
@@ -306,7 +314,7 @@ def create_feedback_issue(event, project_id: int, source: FeedbackCreationSource
             category=DataCategory.USER_REPORT_V2,
             quantity=1,
         )
-        return
+        return None
 
     feedback_message = event["contexts"]["feedback"]["message"]
 
@@ -409,6 +417,8 @@ def create_feedback_issue(event, project_id: int, source: FeedbackCreationSource
         quantity=1,
     )
 
+    return event_fixed
+
 
 def auto_ignore_spam_feedbacks(project, issue_fingerprint):
     """
diff --git a/src/sentry/feedback/usecases/save_feedback_event.py b/src/sentry/feedback/usecases/save_feedback_event.py
new file mode 100644
index 00000000000000..b97468d97444a4
--- /dev/null
+++ b/src/sentry/feedback/usecases/save_feedback_event.py
@@ -0,0 +1,66 @@
+import logging
+from collections.abc import Mapping
+from datetime import UTC, datetime
+from typing import Any
+
+from sentry.feedback.usecases.create_feedback import FeedbackCreationSource, create_feedback_issue
+from sentry.ingest.userreport import save_userreport
+from sentry.models.project import Project
+from sentry.utils import metrics
+
+logger = logging.getLogger(__name__)
+
+
+def save_feedback_event(event_data: Mapping[str, Any], project_id: int):
+    """Saves a feedback from a feedback event envelope.
+
+    If the save is successful and the `associated_event_id` field is present, this will
+    also save a UserReport in Postgres. This is to ensure the feedback can be queried by
+    group_id, which is hard to associate in clickhouse.
+    """
+    if not isinstance(event_data, dict):
+        event_data = dict(event_data)
+
+    # Produce to issue platform
+    fixed_event_data = create_feedback_issue(
+        event_data, project_id, FeedbackCreationSource.NEW_FEEDBACK_ENVELOPE
+    )
+    if not fixed_event_data:
+        return
+
+    try:
+        # Shim to UserReport
+        feedback_context = fixed_event_data["contexts"]["feedback"]
+        associated_event_id = feedback_context.get("associated_event_id")
+        if associated_event_id:
+            project = Project.objects.get_from_cache(id=project_id)
+            save_userreport(
+                project,
+                {
+                    "event_id": associated_event_id,
+                    "project_id": project_id,
+                    # XXX(aliu): including environment ensures the update_user_reports task
+                    # will not shim the report back to feedback.
+                    "environment_id": fixed_event_data.get("environment", "production"),
+                    "name": feedback_context["name"],
+                    "email": feedback_context["contact_email"],
+                    "comments": feedback_context["message"],
+                },
+                FeedbackCreationSource.NEW_FEEDBACK_ENVELOPE,
+                start_time=datetime.fromtimestamp(fixed_event_data["timestamp"], UTC),
+            )
+            metrics.incr("feedback.shim_to_userreport.success")
+
+    except Exception:
+        metrics.incr("feedback.shim_to_userreport.failed")
+        logger.exception(
+            "Error shimming from feedback event to user report.",
+            extra={
+                "associated_event_id": associated_event_id,
+                "project_id": project_id,
+                "environment_id": fixed_event_data.get("environment"),
+                "username": feedback_context.get("name"),
+                "email": feedback_context.get("contact_email"),
+                "comments": feedback_context.get("message"),
+            },
+        )
diff --git a/src/sentry/ingest/userreport.py b/src/sentry/ingest/userreport.py
index d97dd4e84b7878..dbc3c10ad67aad 100644
--- a/src/sentry/ingest/userreport.py
+++ b/src/sentry/ingest/userreport.py
@@ -83,7 +83,10 @@ def save_userreport(
         report["event_id"] = report["event_id"].lower()
         report["project_id"] = project.id
 
-        event = eventstore.backend.get_event_by_id(project.id, report["event_id"])
+        # Use the associated event to validate and update the report.
+        event: Event | GroupEvent | None = eventstore.backend.get_event_by_id(
+            project.id, report["event_id"]
+        )
 
         euser = find_event_user(event)
 
@@ -99,6 +102,7 @@ def save_userreport(
             report["environment_id"] = event.get_environment().id
             report["group_id"] = event.group_id
 
+        # Save the report.
         try:
             with atomic_transaction(using=router.db_for_write(UserReport)):
                 report_instance = UserReport.objects.create(**report)
@@ -136,6 +140,7 @@ def save_userreport(
             if report_instance.group_id:
                 report_instance.notify()
 
+        # Additionally processing if save is successful.
         user_feedback_received.send_robust(project=project, sender=save_userreport)
 
         logger.info(
@@ -150,12 +155,13 @@ def save_userreport(
             "user_report.create_user_report.saved",
             tags={"has_event": bool(event), "referrer": source.value},
         )
-        if event:
+        if event and source.value in FeedbackCreationSource.old_feedback_category_values():
             logger.info(
                 "ingest.user_report.shim_to_feedback",
                 extra={"project_id": project.id, "event_id": report["event_id"]},
             )
             shim_to_feedback(report, event, project, source)
+            # XXX(aliu): the update_user_reports task will still try to shim the report after a period, unless group_id or environment is set.
 
         return report_instance
 
diff --git a/src/sentry/tasks/post_process.py b/src/sentry/tasks/post_process.py
index 6e0edcfce65ac7..ec0a51fe2c6603 100644
--- a/src/sentry/tasks/post_process.py
+++ b/src/sentry/tasks/post_process.py
@@ -1410,25 +1410,26 @@ def link_event_to_user_report(job: PostProcessJob) -> None:
         event = job["event"]
         project = event.project
         user_reports_without_group = UserReport.objects.filter(
-            project_id=project.id,
-            event_id=event.event_id,
-            group_id__isnull=True,
-            environment_id__isnull=True,
+            project_id=project.id, event_id=event.event_id, group_id__isnull=True
         )
         for report in user_reports_without_group:
-            shim_to_feedback(
-                {
-                    "name": report.name,
-                    "email": report.email,
-                    "comments": report.comments,
-                    "event_id": report.event_id,
-                    "level": "error",
-                },
-                event,
-                project,
-                FeedbackCreationSource.USER_REPORT_ENVELOPE,
-            )
-            metrics.incr("event_manager.save._update_user_reports_with_event_link.shim_to_feedback")
+            if report.environment_id is None:
+                shim_to_feedback(
+                    {
+                        "name": report.name,
+                        "email": report.email,
+                        "comments": report.comments,
+                        "event_id": report.event_id,
+                        "level": "error",
+                    },
+                    event,
+                    project,
+                    FeedbackCreationSource.USER_REPORT_ENVELOPE,
+                )
+                metrics.incr(
+                    "event_manager.save._update_user_reports_with_event_link.shim_to_feedback"
+                )
+        # If environment is set, this report was already shimmed from new feedback.
 
         user_reports_updated = user_reports_without_group.update(
             group_id=group.id, environment_id=event.get_environment().id
diff --git a/src/sentry/tasks/store.py b/src/sentry/tasks/store.py
index b480ac5a9e9bbe..e918c66a112672 100644
--- a/src/sentry/tasks/store.py
+++ b/src/sentry/tasks/store.py
@@ -16,7 +16,7 @@
 from sentry.constants import DEFAULT_STORE_NORMALIZER_ARGS
 from sentry.datascrubbing import scrub_data
 from sentry.eventstore import processing
-from sentry.feedback.usecases.create_feedback import FeedbackCreationSource, create_feedback_issue
+from sentry.feedback.usecases.save_feedback_event import save_feedback_event
 from sentry.ingest.types import ConsumerType
 from sentry.killswitches import killswitch_matches_context
 from sentry.lang.native.symbolicator import SymbolicatorTaskKind
@@ -690,7 +690,7 @@ def save_event_feedback(
     project_id: int,
     **kwargs: Any,
 ) -> None:
-    create_feedback_issue(data, project_id, FeedbackCreationSource.NEW_FEEDBACK_ENVELOPE)
+    save_feedback_event(data, project_id)
 
 
 @instrumented_task(
diff --git a/src/sentry/tasks/update_user_reports.py b/src/sentry/tasks/update_user_reports.py
index 0075f46dd8e02e..123edc957a5062 100644
--- a/src/sentry/tasks/update_user_reports.py
+++ b/src/sentry/tasks/update_user_reports.py
@@ -38,13 +38,12 @@ def update_user_reports(**kwargs: Any) -> None:
     # ingestion time
     user_reports = UserReport.objects.filter(
         group_id__isnull=True,
-        environment_id__isnull=True,
         date_added__gte=start,
         date_added__lte=end,
     )
 
     # We do one query per project, just to avoid the small case that two projects have the same event ID
-    project_map: dict[int, Any] = {}
+    project_map: dict[int, list[UserReport]] = {}
     for r in user_reports:
         project_map.setdefault(r.project_id, []).append(r)
 
@@ -90,24 +89,26 @@ def update_user_reports(**kwargs: Any) -> None:
         for event in events:
             report = report_by_event.get(event.event_id)
             if report:
-                if not is_in_feedback_denylist(project.organization):
-                    logger.info(
-                        "update_user_reports.shim_to_feedback",
-                        extra={"report_id": report.id, "event_id": event.event_id},
-                    )
-                    metrics.incr("tasks.update_user_reports.shim_to_feedback")
-                    shim_to_feedback(
-                        {
-                            "name": report.name,
-                            "email": report.email,
-                            "comments": report.comments,
-                            "event_id": report.event_id,
-                            "level": "error",
-                        },
-                        event,
-                        project,
-                        FeedbackCreationSource.UPDATE_USER_REPORTS_TASK,
-                    )
+                if report.environment_id is None:
+                    if not is_in_feedback_denylist(project.organization):
+                        logger.info(
+                            "update_user_reports.shim_to_feedback",
+                            extra={"report_id": report.id, "event_id": event.event_id},
+                        )
+                        metrics.incr("tasks.update_user_reports.shim_to_feedback")
+                        shim_to_feedback(
+                            {
+                                "name": report.name,
+                                "email": report.email,
+                                "comments": report.comments,
+                                "event_id": report.event_id,
+                                "level": "error",
+                            },
+                            event,
+                            project,
+                            FeedbackCreationSource.UPDATE_USER_REPORTS_TASK,
+                        )
+                # XXX(aliu): If a report has environment_id but not group_id, this report was shimmed from a feedback issue, so no need to shim again.
                 report.update(group_id=event.group_id, environment_id=event.get_environment().id)
                 updated_reports += 1
 
diff --git a/tests/sentry/feedback/usecases/test_create_feedback.py b/tests/sentry/feedback/usecases/test_create_feedback.py
index f02afaec9a0c39..e496979f219739 100644
--- a/tests/sentry/feedback/usecases/test_create_feedback.py
+++ b/tests/sentry/feedback/usecases/test_create_feedback.py
@@ -74,7 +74,10 @@ def create_dummy_response(*args, **kwargs):
     )
 
 
-def mock_feedback_event(project_id: int, dt: datetime):
+def mock_feedback_event(project_id: int, dt: datetime | None = None):
+    if dt is None:
+        dt = datetime.now(UTC)
+
     return {
         "project_id": project_id,
         "request": {
@@ -773,7 +776,7 @@ def test_create_feedback_adds_associated_event_id(
 @django_db_all
 def test_create_feedback_tags(default_project, mock_produce_occurrence_to_kafka):
     """We want to surface these tags in the UI. We also use user.email for alert conditions."""
-    event = mock_feedback_event(default_project.id, datetime.now(UTC))
+    event = mock_feedback_event(default_project.id)
     event["user"]["email"] = "josh.ferge@sentry.io"
     event["contexts"]["feedback"]["contact_email"] = "andrew@sentry.io"
     event["contexts"]["trace"] = {"trace_id": "abc123"}
@@ -796,7 +799,7 @@ def test_create_feedback_tags(default_project, mock_produce_occurrence_to_kafka)
 
 @django_db_all
 def test_create_feedback_tags_skips_if_empty(default_project, mock_produce_occurrence_to_kafka):
-    event = mock_feedback_event(default_project.id, datetime.now(UTC))
+    event = mock_feedback_event(default_project.id)
     event["user"].pop("email", None)
     event["contexts"]["feedback"].pop("contact_email", None)
     create_feedback_issue(event, default_project.id, FeedbackCreationSource.NEW_FEEDBACK_ENVELOPE)
@@ -826,7 +829,7 @@ def test_create_feedback_filters_large_message(
     monkeypatch.setattr("sentry.llm.usecases.complete_prompt", mock_complete_prompt)
 
     with Feature(features), set_sentry_option("feedback.message.max-size", 4096):
-        event = mock_feedback_event(default_project.id, datetime.now(UTC))
+        event = mock_feedback_event(default_project.id)
         event["contexts"]["feedback"]["message"] = "a" * 7007
         create_feedback_issue(
             event, default_project.id, FeedbackCreationSource.NEW_FEEDBACK_ENVELOPE
@@ -839,7 +842,7 @@ def test_create_feedback_filters_large_message(
 @django_db_all
 def test_create_feedback_evidence_has_source(default_project, mock_produce_occurrence_to_kafka):
     """We need this evidence field in post process, to determine if we should send alerts."""
-    event = mock_feedback_event(default_project.id, datetime.now(UTC))
+    event = mock_feedback_event(default_project.id)
     source = FeedbackCreationSource.NEW_FEEDBACK_ENVELOPE
     create_feedback_issue(event, default_project.id, source)
 
@@ -857,7 +860,7 @@ def test_create_feedback_evidence_has_spam(
     default_project.update_option("sentry:feedback_ai_spam_detection", True)
 
     with Feature({"organizations:user-feedback-spam-filter-ingest": True}):
-        event = mock_feedback_event(default_project.id, datetime.now(UTC))
+        event = mock_feedback_event(default_project.id)
         source = FeedbackCreationSource.NEW_FEEDBACK_ENVELOPE
         create_feedback_issue(event, default_project.id, source)
 
diff --git a/tests/sentry/feedback/usecases/test_save_feedback_event.py b/tests/sentry/feedback/usecases/test_save_feedback_event.py
new file mode 100644
index 00000000000000..e2922bf384e8fd
--- /dev/null
+++ b/tests/sentry/feedback/usecases/test_save_feedback_event.py
@@ -0,0 +1,66 @@
+from datetime import UTC, datetime
+from unittest import mock
+
+import pytest
+
+from sentry.feedback.usecases.create_feedback import FeedbackCreationSource
+from sentry.feedback.usecases.save_feedback_event import save_feedback_event
+from sentry.testutils.pytest.fixtures import django_db_all
+from tests.sentry.feedback.usecases.test_create_feedback import mock_feedback_event
+
+
+@pytest.fixture
+def mock_create_feedback_issue():
+    with mock.patch("sentry.feedback.usecases.save_feedback_event.create_feedback_issue") as m:
+        yield m
+
+
+@pytest.fixture
+def mock_save_userreport():
+    with mock.patch("sentry.feedback.usecases.save_feedback_event.save_userreport") as m:
+        yield m
+
+
+@django_db_all
+def test_save_feedback_event_no_associated_error(
+    default_project, mock_create_feedback_issue, mock_save_userreport
+):
+    event_data = mock_feedback_event(default_project.id)
+    mock_create_feedback_issue.return_value = None
+
+    save_feedback_event(event_data, default_project.id)
+
+    mock_create_feedback_issue.assert_called_once_with(
+        event_data, default_project.id, FeedbackCreationSource.NEW_FEEDBACK_ENVELOPE
+    )
+    mock_save_userreport.assert_not_called()
+
+
+@django_db_all
+def test_save_feedback_event_with_associated_error(
+    default_project, mock_create_feedback_issue, mock_save_userreport
+):
+    event_data = mock_feedback_event(default_project.id)
+    event_data["contexts"]["feedback"]["associated_event_id"] = "abcd" * 8
+    mock_create_feedback_issue.return_value = event_data
+
+    save_feedback_event(event_data, default_project.id)
+
+    mock_create_feedback_issue.assert_called_once_with(
+        event_data, default_project.id, FeedbackCreationSource.NEW_FEEDBACK_ENVELOPE
+    )
+
+    feedback_context = event_data["contexts"]["feedback"]
+    mock_save_userreport.assert_called_once_with(
+        default_project,
+        {
+            "event_id": "abcd" * 8,
+            "project_id": default_project.id,
+            "environment_id": event_data["environment"],
+            "name": feedback_context["name"],
+            "email": feedback_context["contact_email"],
+            "comments": feedback_context["message"],
+        },
+        FeedbackCreationSource.NEW_FEEDBACK_ENVELOPE,
+        start_time=datetime.fromtimestamp(event_data["timestamp"], UTC),
+    )
diff --git a/tests/sentry/ingest/test_userreport.py b/tests/sentry/ingest/test_userreport.py
index c49bbb457d7b5e..6654064842c16e 100644
--- a/tests/sentry/ingest/test_userreport.py
+++ b/tests/sentry/ingest/test_userreport.py
@@ -1,13 +1,24 @@
 from unittest.mock import Mock
 
+from django.utils import timezone
+
 from sentry.feedback.usecases.create_feedback import (
     UNREAL_FEEDBACK_UNATTENDED_MESSAGE,
     FeedbackCreationSource,
 )
 from sentry.ingest.userreport import save_userreport, should_filter_user_report
 from sentry.models.userreport import UserReport
+from sentry.testutils.factories import Factories
 from sentry.testutils.pytest.fixtures import django_db_all
 
+
+def _mock_event(project_id: int, environment: str):
+    return Factories.store_event(
+        data={"timestamp": timezone.now().isoformat(), "environment": environment},
+        project_id=project_id,
+    )
+
+
 #################################
 # should_filter_user_report tests
 #################################
@@ -116,3 +127,67 @@ def test_save_user_report_filters_large_message(default_project, monkeypatch):
     result = save_userreport(default_project, report, FeedbackCreationSource.USER_REPORT_ENVELOPE)
     assert result is None
     assert UserReport.objects.count() == 0
+
+
+@django_db_all
+def test_save_user_report_shims_if_event_found(default_project, monkeypatch):
+    monkeypatch.setattr("sentry.ingest.userreport.is_in_feedback_denylist", lambda org: False)
+    monkeypatch.setattr(
+        "sentry.ingest.userreport.should_filter_user_report", Mock(return_value=(False, None))
+    )
+
+    event = _mock_event(default_project.id, "prod")
+    monkeypatch.setattr(
+        "sentry.eventstore.backend.get_event_by_id",
+        lambda project_id, event_id: event,
+    )
+
+    mock_shim_to_feedback = Mock()
+    monkeypatch.setattr("sentry.ingest.userreport.shim_to_feedback", mock_shim_to_feedback)
+
+    report = {
+        "event_id": event.event_id,
+        "name": "Test User",
+        "email": "test@example.com",
+        "comments": "This is a test feedback",
+        "project_id": default_project.id,
+    }
+
+    save_userreport(default_project, report, FeedbackCreationSource.USER_REPORT_ENVELOPE)
+    mock_shim_to_feedback.assert_called_once()
+
+
+@django_db_all
+def test_save_user_report_does_not_shim_if_event_found_but_source_is_new_feedback(
+    default_project, monkeypatch
+):
+    # Exact same setup as test_save_user_report_shims_if_event_found
+    monkeypatch.setattr("sentry.ingest.userreport.is_in_feedback_denylist", lambda org: False)
+    monkeypatch.setattr(
+        "sentry.ingest.userreport.should_filter_user_report", Mock(return_value=(False, None))
+    )
+
+    event = _mock_event(default_project.id, "prod")
+    monkeypatch.setattr(
+        "sentry.eventstore.backend.get_event_by_id",
+        lambda project_id, event_id: event,
+    )
+
+    mock_shim_to_feedback = Mock()
+    monkeypatch.setattr("sentry.ingest.userreport.shim_to_feedback", mock_shim_to_feedback)
+
+    report = {
+        "event_id": event.event_id,
+        "name": "Test User",
+        "email": "test@example.com",
+        "comments": "This is a test feedback",
+        "project_id": default_project.id,
+    }
+
+    # Source is new feedback, so no shim
+    save_userreport(
+        default_project,
+        report,
+        FeedbackCreationSource.NEW_FEEDBACK_ENVELOPE,
+    )
+    assert mock_shim_to_feedback.call_count == 0
diff --git a/tests/sentry/tasks/test_update_user_reports.py b/tests/sentry/tasks/test_update_user_reports.py
index dec98b5a8631f3..a02d50c8be265c 100644
--- a/tests/sentry/tasks/test_update_user_reports.py
+++ b/tests/sentry/tasks/test_update_user_reports.py
@@ -123,7 +123,7 @@ def test_event_timerange(self):
         assert report4.environment_id is None
 
     @patch("sentry.feedback.usecases.create_feedback.produce_occurrence_to_kafka")
-    def test_simple_calls_feedback_shim_if_ff_enabled(self, mock_produce_occurrence_to_kafka):
+    def test_calls_feedback_shim_if_ff_enabled(self, mock_produce_occurrence_to_kafka):
         project = self.create_project()
         event1 = self.store_event(
             data={
@@ -162,6 +162,37 @@ def test_simple_calls_feedback_shim_if_ff_enabled(self, mock_produce_occurrence_
         assert mock_event_data["contexts"]["feedback"]["associated_event_id"] == event1.event_id
         assert mock_event_data["level"] == "error"
 
+    @patch("sentry.feedback.usecases.create_feedback.produce_occurrence_to_kafka")
+    def test_does_not_call_feedback_shim_if_environment_is_set(
+        self, mock_produce_occurrence_to_kafka
+    ):
+        project = self.create_project()
+        event1 = self.store_event(
+            data={
+                "environment": self.environment.name,
+                "tags": {"foo": "bar"},
+            },
+            project_id=project.id,
+        )
+        UserReport.objects.create(
+            project_id=project.id,
+            event_id=event1.event_id,
+            comments="It broke!",
+            email="foo@example.com",
+            name="Foo Bar",
+            environment_id=self.environment.id,
+        )
+        with self.tasks():
+            update_user_reports(max_events=2)
+
+        # group_id is still updated
+        report1 = UserReport.objects.get(project_id=project.id, event_id=event1.event_id)
+        assert report1.group_id == event1.group_id
+        assert report1.environment_id == event1.get_environment().id
+
+        # no shim
+        assert len(mock_produce_occurrence_to_kafka.mock_calls) == 0
+
     @patch("sentry.quotas.backend.get_event_retention")
     def test_event_retention(self, mock_get_event_retention):
         retention_days = 21