Entity Resolution and the Instability Problem

PublicEntityId -> current SenzingEntityId
PublicEntityId -> status/history (current, merged, split, deleted)
PublicEntityId -> set of records (sourceSystem + sourceRecordId)

PublicEntityId E1 -> current SenzingEntityId 123 (status: current)
PublicEntityId E2 -> (status: mergedInto E1)

{
  "id": "E0",
  "status": "split",
  "splitInto": ["E3", "E4"]
}

{
  "type": "MERGE",
  "timestamp": "2025-12-01T10:23:45Z",
  "oldEntities": ["E2", "E5"],
  "newEntity": "E1"
}

{
  "type": "SPLIT",
  "timestamp": "2025-12-01T11:05:00Z",
  "oldEntity": "E0",
  "newEntities": ["E3", "E4"]
}

{
  "type": "UPDATE",
  "timestamp": "...",
  "entity": "E1",
  "changeSummary": { ... }
}

RecordAssignment
----------------
(DATA_SOURCE, RECORD_ID) -> lastKnownEntityId

EntitySummary
-------------
EntityId -> LastKnownExists (bool), LastSeenAt (timestamp)

preExists[Eold] == true
postExists[Eold] == false
|nextEntities[Eold]| > 1

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
using System.Text.Json.Nodes;
using Senzing.Sdk;
// optional, but convenient so we can write SzWithInfo, SzEntityDefaultFlags directly:
using static Senzing.Sdk.SzFlag;

namespace YourNamespace.Senzing
{
    /// <summary>
    /// Simple value object for identifying a record in Senzing.
    /// </summary>
    public readonly record struct RecordKey(string DataSource, string RecordId);

    /// <summary>
    /// Coarse-grained classification of how an entity changed during a single event.
    /// You can expand this to whatever vocabulary you want.
    /// </summary>
    public enum EntityChangeKind
    {
        Unknown = 0,
        Birth,          // did not exist before, exists now
        Death,          // existed before, now gone, records also gone
        MergeInto,      // this entity ID disappeared, records moved into one other entity
        SplitInto,      // this entity ID disappeared, records fanned out to multiple entities
        Shrink,         // same ID, lost records only
        Grow,           // same ID, gained records only
        Changed,        // same ID, both lost and gained records (partial split/merge)
        Unchanged       // same set of records
    }

    /// <summary>
    /// Summary of how a given entity ID changed in one WithInfo event.
    /// </summary>
    public sealed class EntityChangeSummary
    {
        public long EntityId { get; init; }

        public bool PreExists { get; init; }
        public bool PostExists { get; init; }

        public IReadOnlyList<RecordKey> PreRecords { get; init; } = Array.Empty<RecordKey>();
        public IReadOnlyList<RecordKey> PostRecords { get; init; } = Array.Empty<RecordKey>();

        /// <summary>
        /// For an entity that disappeared, which entity IDs now own its records.
        /// (If 0 -> delete, 1 -> merge, >1 -> split.)
        /// </summary>
        public IReadOnlyCollection<long> NextEntities { get; init; } = Array.Empty<long>();

        /// <summary>
        /// Records that used to belong to this entity but are no longer present anywhere
        /// (as far as we can see right after this event).
        /// </summary>
        public IReadOnlyList<RecordKey> DeletedRecords { get; init; } = Array.Empty<RecordKey>();

        public EntityChangeKind ChangeKind { get; init; }

        /// <summary>
        /// Optional: entities that contributed records into this entity.
        /// Useful to describe merges from the survivor's point of view.
        /// </summary>
        public IReadOnlyCollection<long> Contributors { get; init; } = Array.Empty<long>();
    }

    /// <summary>
    /// Abstraction for your persistence of "last known entity state".
    /// Implement this using your own DB / ORM layer.
    /// </summary>
    public interface IEntityStateRepository
    {
        /// <summary>
        /// Returns true if we previously knew this entity ID to exist.
        /// If it's completely unknown, return false.
        /// </summary>
        bool WasKnownToExist(long entityId);

        /// <summary>
        /// Returns the last known records for this entity ID.
        /// If we have no history for it, return an empty list.
        /// </summary>
        IReadOnlyList<RecordKey> GetKnownRecords(long entityId);

        /// <summary>
        /// Update our snapshot of an entity's state after processing an event.
        /// If exists == false, you can either delete the row or mark as not-existing.
        /// </summary>
        void SaveEntitySnapshot(long entityId, bool exists, IReadOnlyList<RecordKey> records);
    }

    /// <summary>
    /// Wraps SzEngine and applies the "robust per-event processing pattern"
    /// for AddRecord/DeleteRecord/ProcessRedoRecord WithInfo payloads.
    /// </summary>
    public sealed class SenzingEventProcessor
    {
        private readonly SzEngine _engine;
        private readonly IEntityStateRepository _stateRepo;

        public SenzingEventProcessor(SzEngine engine, IEntityStateRepository stateRepo)
        {
            _engine = engine ?? throw new ArgumentNullException(nameof(engine));
            _stateRepo = stateRepo ?? throw new ArgumentNullException(nameof(stateRepo));
        }

        #region Public entry points around SzEngine

        public IReadOnlyList<EntityChangeSummary> ProcessAddRecord(
            string dataSource,
            string recordId,
            string recordJson)
        {
            // Call Senzing AddRecord with SZ_WITH_INFO to get AFFECTED_ENTITIES. :contentReference[oaicite:2]{index=2}
            string infoJson = _engine.AddRecord(dataSource, recordId, recordJson, SzWithInfo);
            return ProcessWithInfoPayload(infoJson, "AddRecord");
        }

        public IReadOnlyList<EntityChangeSummary> ProcessDeleteRecord(
            string dataSource,
            string recordId)
        {
            // Same idea for DeleteRecord. :contentReference[oaicite:3]{index=3}
            string infoJson = _engine.DeleteRecord(dataSource, recordId, SzWithInfo);
            return ProcessWithInfoPayload(infoJson, "DeleteRecord");
        }

        public IReadOnlyList<EntityChangeSummary> ProcessRedoRecord(string redoRecordJson)
        {
            // And for ProcessRedoRecord (redo stream). :contentReference[oaicite:4]{index=4}
            string infoJson = _engine.ProcessRedoRecord(redoRecordJson, SzWithInfo);
            return ProcessWithInfoPayload(infoJson, "ProcessRedoRecord");
        }

        #endregion

        #region Core per-event algorithm

        private IReadOnlyList<EntityChangeSummary> ProcessWithInfoPayload(
            string infoJson,
            string operationName)
        {
            if (string.IsNullOrWhiteSpace(infoJson))
            {
                // Nothing to do (e.g., flags didn't include SZ_WITH_INFO).
                return Array.Empty<EntityChangeSummary>();
            }

            // 1. Parse the WithInfo JSON and extract AFFECTED_ENTITIES.
            JsonObject? infoObject = JsonNode.Parse(infoJson)?.AsObject()
                                     ?? throw new ArgumentException("Invalid JSON in info payload.", nameof(infoJson));

            var affectedIds = ExtractAffectedEntityIds(infoObject);

            if (affectedIds.Count == 0)
            {
                return Array.Empty<EntityChangeSummary>();
            }

            // 2. Pre-state from *our* DB: preExists, preRecords.
            var preExists = new Dictionary<long, bool>();
            var preRecords = new Dictionary<long, List<RecordKey>>();

            foreach (long entityId in affectedIds)
            {
                bool existed = _stateRepo.WasKnownToExist(entityId);
                preExists[entityId] = existed;

                var records = _stateRepo.GetKnownRecords(entityId)?.ToList()
                              ?? new List<RecordKey>();

                preRecords[entityId] = records;
            }

            // 3. Post-state from Senzing GetEntity: postExists, postRecords.
            var postExists = new Dictionary<long, bool>();
            var postRecords = new Dictionary<long, List<RecordKey>>();

            foreach (long entityId in affectedIds)
            {
                try
                {
                    // GetEntity by entity ID. :contentReference[oaicite:5]{index=5}
                    string entityJson = _engine.GetEntity(entityId, SzEntityDefaultFlags);

                    List<RecordKey> records = ExtractRecordKeysFromEntityJson(entityJson);

                    postExists[entityId] = true;
                    postRecords[entityId] = records;
                }
                catch (SzNotFoundException)
                {
                    // Entity ID is no longer valid.
                    postExists[entityId] = false;
                    postRecords[entityId] = new List<RecordKey>();
                }
            }

            // 4. Build currentByRecord: (DATA_SOURCE, RECORD_ID) -> set of entity IDs that own it now.
            var currentByRecord = new Dictionary<RecordKey, HashSet<long>>();

            foreach (var kvp in postRecords)
            {
                long entityId = kvp.Key;
                foreach (RecordKey record in kvp.Value)
                {
                    if (!currentByRecord.TryGetValue(record, out var set))
                    {
                        set = new HashSet<long>();
                        currentByRecord[record] = set;
                    }
                    set.Add(entityId);
                }
            }

            // 5. For entities that *used* to exist, figure out where their records went:
            //    deletedRecords + nextEntities.
            var deletedRecords = new Dictionary<long, List<RecordKey>>();
            var nextEntities = new Dictionary<long, HashSet<long>>();

            foreach (long entityId in affectedIds)
            {
                bool existedBefore = preExists.TryGetValue(entityId, out var ex) && ex;
                if (!existedBefore)
                {
                    continue; // Nothing to compare.
                }

                List<RecordKey> oldRecords = preRecords[entityId];

                var nextSet = new HashSet<long>();
                var deletedList = new List<RecordKey>();

                foreach (RecordKey record in oldRecords)
                {
                    if (currentByRecord.TryGetValue(record, out var ownersNow) &&
                        ownersNow.Count > 0)
                    {
                        // Typically a record belongs to exactly one entity, but we don't assume.
                        foreach (long owner in ownersNow)
                        {
                            nextSet.Add(owner);
                        }
                    }
                    else
                    {
                        // We don't see this record in any entity now -> treat as deleted.
                        deletedList.Add(record);
                    }
                }

                if (nextSet.Count > 0)
                {
                    nextEntities[entityId] = nextSet;
                }

                if (deletedList.Count > 0)
                {
                    deletedRecords[entityId] = deletedList;
                }
            }

            // 6. Build contributors: for each entity that exists now, which old entity IDs fed into it?
            var contributors = new Dictionary<long, HashSet<long>>();

            foreach (var kvp in nextEntities)
            {
                long from = kvp.Key;
                foreach (long to in kvp.Value)
                {
                    if (!contributors.TryGetValue(to, out var set))
                    {
                        set = new HashSet<long>();
                        contributors[to] = set;
                    }
                    set.Add(from);
                }
            }

            // 7. Classify change kind for each affected entity ID.
            var results = new List<EntityChangeSummary>();

            foreach (long entityId in affectedIds)
            {
                bool existedBefore = preExists.TryGetValue(entityId, out var exBefore) && exBefore;
                bool existsNow = postExists.TryGetValue(entityId, out var exNow) && exNow;

                List<RecordKey> before = preRecords.TryGetValue(entityId, out var pr)
                    ? pr
                    : new List<RecordKey>();

                List<RecordKey> after = postRecords.TryGetValue(entityId, out var po)
                    ? po
                    : new List<RecordKey>();

                HashSet<long> nextSet = nextEntities.TryGetValue(entityId, out var ns)
                    ? ns
                    : new HashSet<long>();

                List<RecordKey> deleted = deletedRecords.TryGetValue(entityId, out var dr)
                    ? dr
                    : new List<RecordKey>();

                // Default classification.
                EntityChangeKind kind = EntityChangeKind.Unknown;

                if (!existedBefore && existsNow)
                {
                    // Newly created entity ID.
                    kind = EntityChangeKind.Birth;
                }
                else if (existedBefore && !existsNow)
                {
                    // Entity ID disappeared.
                    if (nextSet.Count == 0)
                    {
                        // No surviving records anywhere (or not visible yet) -> death/delete.
                        kind = EntityChangeKind.Death;
                    }
                    else if (nextSet.Count == 1)
                    {
                        // All surviving records moved into a single entity -> merge.
                        kind = EntityChangeKind.MergeInto;
                    }
                    else
                    {
                        // Surviving records fanned out across multiple entities -> split.
                        kind = EntityChangeKind.SplitInto;
                    }
                }
                else if (existedBefore && existsNow)
                {
                    // Same entity ID, but we may have gained or lost records.
                    var oldSet = new HashSet<RecordKey>(before);
                    var newSet = new HashSet<RecordKey>(after);

                    bool lostAny = oldSet.Except(newSet).Any();
                    bool gainedAny = newSet.Except(oldSet).Any();

                    if (!lostAny && !gainedAny)
                    {
                        kind = EntityChangeKind.Unchanged;
                    }
                    else if (gainedAny && !lostAny)
                    {
                        kind = EntityChangeKind.Grow;
                    }
                    else if (!gainedAny && lostAny)
                    {
                        kind = EntityChangeKind.Shrink;
                    }
                    else
                    {
                        // Both: some records left, some came in.
                        kind = EntityChangeKind.Changed;
                    }
                }

                // Contributors (from survivor's perspective).
                HashSet<long> contributorSet = contributors.TryGetValue(entityId, out var cs)
                    ? cs
                    : new HashSet<long>();

                var summary = new EntityChangeSummary
                {
                    EntityId = entityId,
                    PreExists = existedBefore,
                    PostExists = existsNow,
                    PreRecords = before,
                    PostRecords = after,
                    NextEntities = nextSet,
                    DeletedRecords = deleted,
                    ChangeKind = kind,
                    Contributors = contributorSet
                };

                results.Add(summary);
            }

            // 8. Persist our new snapshot of entity -> records.
            foreach (long entityId in affectedIds)
            {
                bool existsNow = postExists.TryGetValue(entityId, out var exNow) && exNow;
                List<RecordKey> nowRecords = postRecords.TryGetValue(entityId, out var list)
                    ? list
                    : new List<RecordKey>();

                _stateRepo.SaveEntitySnapshot(entityId, existsNow, nowRecords);
            }

            // At this point you could also publish the EntityChangeSummary list
            // to an event stream, stable-ID layer, etc.
            return results;
        }

        #endregion

        #region Helper methods

        /// <summary>
        /// Extracts entity IDs from the AFFECTED_ENTITIES array
        /// in an AddRecord/DeleteRecord/ProcessRedoRecord WithInfo response.
        /// </summary>
        private static HashSet<long> ExtractAffectedEntityIds(JsonObject infoObject)
        {
            var result = new HashSet<long>();

            JsonArray? affectedArr = infoObject["AFFECTED_ENTITIES"]?.AsArray();
            if (affectedArr == null) return result;

            foreach (JsonNode? node in affectedArr)
            {
                JsonObject? obj = node?.AsObject();
                if (obj == null) continue;

                long id = obj["ENTITY_ID"]?.GetValue<long>() ?? 0L;
                if (id > 0)
                {
                    result.Add(id);
                }
            }

            return result;
        }

        /// <summary>
        /// Parses a GetEntity(...) JSON response and returns the list of record keys
        /// in RESOLVED_ENTITY.RECORDS[*].(DATA_SOURCE, RECORD_ID). :contentReference[oaicite:6]{index=6}
        /// </summary>
        private static List<RecordKey> ExtractRecordKeysFromEntityJson(string entityJson)
        {
            var result = new List<RecordKey>();

            if (string.IsNullOrWhiteSpace(entityJson))
                return result;

            JsonObject? root = JsonNode.Parse(entityJson)?.AsObject();
            if (root == null) return result;

            JsonObject? resolved = root["RESOLVED_ENTITY"]?.AsObject();
            if (resolved == null) return result;

            JsonArray? recordsArr = resolved["RECORDS"]?.AsArray();
            if (recordsArr == null) return result;

            foreach (JsonNode? node in recordsArr)
            {
                JsonObject? rec = node?.AsObject();
                if (rec == null) continue;

                string? ds = rec["DATA_SOURCE"]?.GetValue<string>();
                string? rid = rec["RECORD_ID"]?.GetValue<string>();

                if (!string.IsNullOrEmpty(ds) && !string.IsNullOrEmpty(rid))
                {
                    result.Add(new RecordKey(ds!, rid!));
                }
            }

            return result;
        }

        #endregion
    }
}

// Somewhere during startup:
SzEnvironment env = /* obtain from Senzing setup */;
SzEngine engine = env.GetEngine();
IEntityStateRepository stateRepo = new MyEntityStateRepository(/* your DB stuff */);

var processor = new SenzingEventProcessor(engine, stateRepo);

// Adding a record:
var changes = processor.ProcessAddRecord("CUSTOMERS", "12345", recordJson);

// 'changes' now tells you, per entity ID:
// - did it appear/disappear
// - old vs new record membership
// - whether it merged/split/shrank/grew
// - where its records moved to (NextEntities)
// - which old entities fed into each survivor (Contributors)