Add a write correlator to group all attributes of a single write across TRoE and MongoDB

[cfreyfh]

Problem

TRoE rows have no key that links all attributes written by a single operation
("entity snapshot"):

• instanceId is unique per attribute
• observedAt is a business timestamp, sometimes NULL and not unique
• ts (request time) is shared within a request but not guaranteed unique

So a query like "give me all attributes written together with
OriginSource = OPC_Server_01" cannot be answered reliably.

Solution

Introduce a write correlator: a single value per write operation, shared by
every row that write produces.

• The correlator is taken from the NGSILD-Correlator request header.
• If the header is absent, one is generated per request
  (urn:ngsi-ld:correlator:<uuid>), so the grouping key always exists.
• It is resolved exactly once per request (correlatorGet()) and then written
  into every row of that write.

Where it is stored

• TRoE / PostgreSQL — a new correlator column on entities, attributes
  and subAttributes. All rows of one write share the same value.
• MongoDB — the entity document stores the correlator of the last write in
  the lastCorrelator field. It is updated by every write path: entity create,
  attribute append, attribute patch/replace and attribute delete.

This provides a guaranteed-unique key grouping all attributes of one snapshot:

SELECT a.* FROM attributes a
WHERE a.correlator IN (
  SELECT correlator FROM attributes
  WHERE id = '<expanded OriginSource name>' AND text = 'OPC_Server_01'
);

[github-actions]

CLA Assistant Lite bot:
Thank you for your submission, we really appreciate it. Like many open-source projects, we ask that you all sign our Contributor License Agreement before we can accept your contribution. You can sign the CLA by just posting a Pull Request Comment same as the below format.

---

I have read the CLA Document and I hereby sign the CLA

---

2 out of 3 committers have signed the CLA.
✅ @carstenfrey-de
✅ @cfreyfh
❌ @claude
_{You can retrigger this bot by commenting recheckcla in this Pull Request}

[kzangeli]

One very important detail.
The postgres DB layout is changed in the PR, so, all existing DBs need a migration.
Would be great to automate this inside orion-ld (the broker starts, sees the DB layout is old and migrates automatically).
This will not br the last change in DB layout, so, an automatic mechanism like that would be quite useful.

[kzangeli]

So, all tests pass, all good.
BUT, as I said in a comment, we need to merge the DB model. Any other deployment out there will fail (if they use TRoE) probably at startup once this modification is in, so, we need an automatic "merge" (whatever it's called) of the DB layout.
Can't ship this PR without it.

[cfreyfh]

What do you think about this?

automatic, versioned schema-migration mechanism for the TRoE PostgreSQL database. On startup (and for every tenant database), the broker detects an outdated DB layout and migrates it automatically — no external tooling or manual step required.

This is built as a general mechanism (not a one-off for the correlator column), since the DB layout will keep evolving: future changes just bump a version and append a step.

How it works

• Each TRoE database carries a metadata table with a schemaVersion row.
• A database created before this mechanism existed has no such row and is treated as the baseline (1).
• An ordered, idempotent migration registry lives in code (pgSchemaMigrate). Each step uses ADD COLUMN IF NOT EXISTS / CREATE INDEX IF NOT EXISTS, bumps schemaVersion, and runs in its own transaction (rollback on failure).
• A PostgreSQL advisory lock wraps the whole migration, so multiple broker instances pointing at the same database never migrate in parallel.
• Hooked into pgDatabasePrepare, which already runs per database at startup and per tenant.

schemaVersion	Change
1	Baseline (released TRoE layout)
2	Adds the write correlator column to entities, attributes, subAttributes (+ index)

Freshly created databases already carry the latest layout, so the steps run as no-ops on them; only pre-existing databases are actually upgraded.

[kzangeli]

That's perfect. Exactly what I had in mind

[sonarqubecloud]

Quality Gate failed

Failed conditions
7 Security Hotspots
B Maintainability Rating on New Code (required ≥ A)

See analysis details on SonarQube Cloud

Catch issues before they fail your Quality Gate with our IDE extension SonarQube for IDE