Function processChangeMessage

packages/electric-db-collection/src/electric.ts:1497–1548

(changeMessage: Message<T>)

       * Process a change message: handle tags and write the mutation
       */
      const processChangeMessage = (changeMessage: Message<T>) => {
        if (!isChangeMessage(changeMessage)) {
          return
        }

        // Process tags if present
        const tags = changeMessage.headers.tags
        const removedTags = changeMessage.headers.removed_tags
        const hasTags = tags || removedTags

        // Extract active_conditions from headers (DNF support)
        const activeConditions = changeMessage.headers.active_conditions as
          | ActiveConditions
          | undefined

        const rowId = collection.getKeyFromItem(changeMessage.value)
        const operation = changeMessage.headers.operation

        // Track synced keys and handle overlapping subset queries.
        // When multiple subset queries return the same row, the server sends
        // `insert` for each response. We convert subsequent inserts to updates
        // to avoid duplicate key errors when the row's data has changed.
        const isDelete = operation === `delete`
        const isDuplicateInsert =
          operation === `insert` && syncedKeys.has(rowId)

        if (isDelete) {
          syncedKeys.delete(rowId)
        } else {
          syncedKeys.add(rowId)
        }

        if (isDelete) {
          clearTagsForRow(rowId)
        } else if (hasTags) {
          processTagsForChangeMessage(
            tags,
            removedTags,
            rowId,
            activeConditions,
          )
        }

        write({
          type: isDuplicateInsert ? `update` : operation,
          value: changeMessage.value,
          // Include the primary key and relation info in the metadata
          metadata: {
            ...changeMessage.headers,
          },
        })
      }

      // Create deduplicated loadSubset wrapper for non-eager modes
      // This prevents redundant snapshot requests when multiple concurrent