chore: fix finished processing inspections, now it tracks resolved in…

…spections properly

packages/jetbrains-plugin/src/main/kotlin/com/mongodb/jbplugin/observability/probe/InspectionStatusChangedProbe.kt

@@ -1,7 +1,7 @@
 package com.mongodb.jbplugin.observability.probe
 
 import com.intellij.codeInspection.ProblemsHolder
-import com.intellij.openapi.application.readAction
+import com.intellij.openapi.application.ApplicationManager
 import com.intellij.openapi.components.Service
 import com.intellij.openapi.diagnostic.Logger
 import com.intellij.openapi.diagnostic.logger
@@ -16,10 +16,10 @@ import com.mongodb.jbplugin.observability.useLogMessage
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import java.lang.ref.WeakReference
-import java.util.Collections
 import java.util.UUID
-import java.util.concurrent.CopyOnWriteArrayList
 
 private val inspectionTelemetry = Dispatchers.IO
 private val logger: Logger = logger<InspectionStatusChangedProbe>()
@@ -36,8 +36,10 @@ class InspectionStatusChangedProbe(
         }
     }
 
+    private val mutex: Mutex = Mutex()
+
     private val problemsByInspectionType: MutableMap<InspectionType, MutableList<UniqueInspection>> =
-        Collections.synchronizedMap(mutableMapOf())
+        mutableMapOf()
 
     fun inspectionChanged(inspectionType: InspectionType, query: Node<PsiElement>) {
         val dialect = query.component<HasSourceDialect>() ?: return
@@ -46,34 +48,34 @@ class InspectionStatusChangedProbe(
         cs.launch(inspectionTelemetry) {
             val elementsWithProblems = problemsByInspectionType(inspectionType)
 
-            // check if the element is already in the list
-            if (isElementRegistered(elementsWithProblems) { psiElement }) {
-                // do nothing, it's already registered
-                return@launch
-            }
+            mutex.withLock {
+                // check if the element is already in the list
+                if (isElementRegistered(elementsWithProblems) { psiElement }) {
+                    // do nothing, it's already registered
+                    return@launch
+                }
 
-            // it's a new error, send a telemetry event and store it
-            val inspection = UniqueInspection.new(query)
-            elementsWithProblems.add(inspection)
-
-            val telemetry by service<TelemetryService>()
-            val event = TelemetryEvent.InspectionStatusChangeEvent(
-                dialect = dialect.name,
-                inspectionType = inspectionType,
-                inspectionStatus = TelemetryEvent.InspectionStatusChangeEvent.InspectionStatus.ACTIVE,
-                null,
-                null
-            )
-
-            telemetry.sendEvent(event)
-            logger.info(
-                useLogMessage("New inspection triggered")
-                    .put("inspection_id", inspection.id.toString())
-                    .mergeTelemetryEventProperties(event)
-                    .build()
-            )
-
-            problemsByInspectionType[inspectionType] = elementsWithProblems
+                // it's a new error, send a telemetry event and store it
+                val inspection = UniqueInspection.new(query)
+                elementsWithProblems.add(inspection)
+
+                val telemetry by service<TelemetryService>()
+                val event = TelemetryEvent.InspectionStatusChangeEvent(
+                    dialect = dialect.name,
+                    inspectionType = inspectionType,
+                    inspectionStatus = TelemetryEvent.InspectionStatusChangeEvent.InspectionStatus.ACTIVE,
+                    null,
+                    null
+                )
+
+                telemetry.sendEvent(event)
+                logger.info(
+                    useLogMessage("New inspection triggered")
+                        .put("inspection_id", inspection.id.toString())
+                        .mergeTelemetryEventProperties(event)
+                        .build()
+                )
+            }
         }
     }
 
@@ -85,33 +87,33 @@ class InspectionStatusChangedProbe(
         cs.launch(inspectionTelemetry) {
             val elementsWithProblems = problemsByInspectionType(inspectionType)
 
-            if (isElementRegistered(elementsWithProblems) { psiElement }) {
-                // do nothing, it's already registered
-                return@launch
-            }
+            mutex.withLock {
+                if (isElementRegistered(elementsWithProblems) { psiElement }) {
+                    // do nothing, it's already registered
+                    return@launch
+                }
 
-            // it's a new error, send a telemetry event and store it
-            val inspection = UniqueInspection.new(query)
-            elementsWithProblems.add(inspection)
-
-            val telemetry by service<TelemetryService>()
-            val event = TelemetryEvent.InspectionStatusChangeEvent(
-                dialect = dialect.name,
-                inspectionType = inspectionType,
-                inspectionStatus = TelemetryEvent.InspectionStatusChangeEvent.InspectionStatus.ACTIVE,
-                actualFieldType = actualType,
-                expectedFieldType = expectedType,
-            )
-
-            telemetry.sendEvent(event)
-            logger.info(
-                useLogMessage("New inspection triggered")
-                    .put("inspection_id", inspection.id.toString())
-                    .mergeTelemetryEventProperties(event)
-                    .build()
-            )
-
-            problemsByInspectionType[inspectionType] = elementsWithProblems
+                // it's a new error, send a telemetry event and store it
+                val inspection = UniqueInspection.new(query)
+                elementsWithProblems.add(inspection)
+
+                val telemetry by service<TelemetryService>()
+                val event = TelemetryEvent.InspectionStatusChangeEvent(
+                    dialect = dialect.name,
+                    inspectionType = inspectionType,
+                    inspectionStatus = TelemetryEvent.InspectionStatusChangeEvent.InspectionStatus.ACTIVE,
+                    actualFieldType = actualType,
+                    expectedFieldType = expectedType,
+                )
+
+                telemetry.sendEvent(event)
+                logger.info(
+                    useLogMessage("New inspection triggered")
+                        .put("inspection_id", inspection.id.toString())
+                        .mergeTelemetryEventProperties(event)
+                        .build()
+                )
+            }
         }
     }
 
@@ -123,12 +125,15 @@ class InspectionStatusChangedProbe(
             // check all our registered problems
             // if at the end of the processing cycle it's empty
             // we will assume they are
-            for (loopResult in results) {
+            mutex.withLock {
                 for (elementWithProblem in elementsWithProblems) {
-                    if (isElementRegistered(elementsWithProblems, loopResult::getPsiElement)) {
+                    val findEquivalentProblem = results.find {
+                        isElementRegistered(elementsWithProblems, it::getPsiElement)
+                    }
+                    if (findEquivalentProblem != null) {
                         // the problem is still there, so don't do anything
                         // do nothing, it's already registered
-                        break
+                        continue
                     }
 
                     elementsWithProblems.remove(elementWithProblem)
@@ -156,24 +161,28 @@ class InspectionStatusChangedProbe(
         }
     }
 
-    private suspend fun isElementRegistered(
+    private fun isElementRegistered(
         elementsWithProblems: MutableList<UniqueInspection>,
         psiElement: () -> PsiElement
     ): Boolean = runCatching {
-        readAction<Boolean> {
+        ApplicationManager.getApplication().runReadAction<Boolean> {
             elementsWithProblems.find {
-                it.on.get()?.source == psiElement ||
-                    it.on.get()?.source?.isEquivalentTo(psiElement()) == true
+                val isStrictlyEqual = it.on.get()?.source == psiElement()
+                val isEquivalent = it.on.get()?.source?.isEquivalentTo(psiElement()) == true
+
+                isStrictlyEqual || isEquivalent
             } != null
         }
     }.getOrDefault(false)
 
-    private fun problemsByInspectionType(inspectionType: InspectionType): MutableList<UniqueInspection> {
-        val result = problemsByInspectionType.computeIfAbsent(inspectionType) {
-            CopyOnWriteArrayList()
-        }
+    private suspend fun problemsByInspectionType(inspectionType: InspectionType): MutableList<UniqueInspection> {
+        return mutex.withLock {
+            val result = problemsByInspectionType.computeIfAbsent(inspectionType) {
+                mutableListOf()
+            }
 
-        result.removeAll { it.on.get() == null }
-        return result
+            result.removeAll { it.on.get() == null }
+            result
+        }
     }
 }

packages/jetbrains-plugin/src/test/kotlin/com/mongodb/jbplugin/observability/probe/InspectionStatusChangedProbeTest.kt

@@ -0,0 +1,153 @@
+package com.mongodb.jbplugin.observability.probe
+
+import com.intellij.codeInspection.ProblemsHolder
+import com.intellij.openapi.application.Application
+import com.intellij.psi.PsiElement
+import com.mongodb.jbplugin.fixtures.IntegrationTest
+import com.mongodb.jbplugin.fixtures.mockLogMessage
+import com.mongodb.jbplugin.fixtures.withMockedService
+import com.mongodb.jbplugin.mql.Node
+import com.mongodb.jbplugin.mql.components.HasSourceDialect
+import com.mongodb.jbplugin.observability.TelemetryEvent
+import com.mongodb.jbplugin.observability.TelemetryService
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.advanceUntilIdle
+import org.junit.jupiter.api.Test
+import org.mockito.Mockito.`when`
+import org.mockito.kotlin.any
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.timeout
+import org.mockito.kotlin.times
+import org.mockito.kotlin.verify
+
+@IntegrationTest
+internal class InspectionStatusChangedProbeTest {
+    @Test
+    fun `should send a InspectionStatusChangeEvent event when found for the first time`(
+        application: Application,
+        testScope: TestScope
+    ) {
+        val telemetryService = mock<TelemetryService>()
+        val dialect = HasSourceDialect.DialectName.entries.toTypedArray().random()
+
+        val query = Node<PsiElement?>(null, listOf(HasSourceDialect(dialect))) as Node<PsiElement>
+
+        application.withMockedService(telemetryService)
+            .withMockedService(mockLogMessage())
+
+        val probe = InspectionStatusChangedProbe(testScope)
+
+        probe.inspectionChanged(
+            TelemetryEvent.InspectionStatusChangeEvent.InspectionType.FIELD_DOES_NOT_EXIST,
+            query
+        )
+        probe.inspectionChanged(
+            TelemetryEvent.InspectionStatusChangeEvent.InspectionType.FIELD_DOES_NOT_EXIST,
+            query
+        )
+
+        testScope.advanceUntilIdle()
+
+        verify(telemetryService, timeout(1000).times(1)).sendEvent(any())
+    }
+
+    @Test
+    fun `should send a InspectionStatusChangeEvent event when multiple types of events`(
+        application: Application,
+        testScope: TestScope
+    ) {
+        val telemetryService = mock<TelemetryService>()
+        val dialect = HasSourceDialect.DialectName.entries.toTypedArray().random()
+
+        val query = Node<PsiElement?>(null, listOf(HasSourceDialect(dialect))) as Node<PsiElement>
+
+        application.withMockedService(telemetryService)
+            .withMockedService(mockLogMessage())
+
+        val probe = InspectionStatusChangedProbe(testScope)
+
+        probe.inspectionChanged(
+            TelemetryEvent.InspectionStatusChangeEvent.InspectionType.FIELD_DOES_NOT_EXIST,
+            query
+        )
+        probe.inspectionChanged(
+            TelemetryEvent.InspectionStatusChangeEvent.InspectionType.FIELD_DOES_NOT_EXIST,
+            query
+        )
+        probe.inspectionChanged(
+            TelemetryEvent.InspectionStatusChangeEvent.InspectionType.NO_NAMESPACE_INFERRED,
+            query
+        )
+        probe.inspectionChanged(
+            TelemetryEvent.InspectionStatusChangeEvent.InspectionType.NO_NAMESPACE_INFERRED,
+            query
+        )
+
+        testScope.advanceUntilIdle()
+
+        verify(telemetryService, timeout(1000).times(2)).sendEvent(any())
+    }
+
+    @Test
+    fun `should send a InspectionStatusChangeEvent event with type checking issues`(
+        application: Application,
+        testScope: TestScope
+    ) {
+        val telemetryService = mock<TelemetryService>()
+        val dialect = HasSourceDialect.DialectName.entries.toTypedArray().random()
+
+        val query = Node<PsiElement?>(null, listOf(HasSourceDialect(dialect))) as Node<PsiElement>
+
+        application.withMockedService(telemetryService)
+            .withMockedService(mockLogMessage())
+
+        val probe = InspectionStatusChangedProbe(testScope)
+
+        probe.typeMismatchInspectionActive(query, "actual", "expected")
+
+        testScope.advanceUntilIdle()
+
+        verify(telemetryService, timeout(1000).times(1)).sendEvent(
+            TelemetryEvent.InspectionStatusChangeEvent(
+                dialect,
+                TelemetryEvent.InspectionStatusChangeEvent.InspectionType.TYPE_MISMATCH,
+                TelemetryEvent.InspectionStatusChangeEvent.InspectionStatus.ACTIVE,
+                "actual",
+                "expected"
+            )
+        )
+    }
+
+    @Test
+    fun `should send a resolved InspectionStatusChangeEvent when there is no problem anymore`(
+        application: Application,
+        testScope: TestScope
+    ) {
+        val telemetryService = mock<TelemetryService>()
+        val problemsHolder = mock<ProblemsHolder>()
+
+        val dialect = HasSourceDialect.DialectName.entries.toTypedArray().random()
+
+        `when`(problemsHolder.results).thenReturn(emptyList())
+
+        val query = Node<PsiElement?>(null, listOf(HasSourceDialect(dialect))) as Node<PsiElement>
+
+        application.withMockedService(telemetryService)
+            .withMockedService(mockLogMessage())
+
+        val probe = InspectionStatusChangedProbe(testScope)
+
+        probe.inspectionChanged(
+            TelemetryEvent.InspectionStatusChangeEvent.InspectionType.FIELD_DOES_NOT_EXIST,
+            query
+        )
+        probe.finishedProcessingInspections(
+            TelemetryEvent.InspectionStatusChangeEvent.InspectionType.FIELD_DOES_NOT_EXIST,
+            problemsHolder
+        )
+
+        testScope.advanceUntilIdle()
+
+        verify(telemetryService, timeout(1000).times(2)).sendEvent(any())
+    }
+}