internal/infra/pg/factstore.go
internal/infra/pg · 460 lines · 22 declarations · source
Declarations
type FactStore
type FactStore struct{ pool *pgxpool.Pool }
FactStore resolves entities and asserts facts with their evidence.
func NewFactStore
func NewFactStore(pool *pgxpool.Pool) *FactStore
const ProjectionFact, ProjectionEntity
const (
ProjectionFact = "fact"
ProjectionEntity = "entity"
)
Projection kinds. Erasure walks `projection_dependency`, so these strings are what make a fact and an entity reachable from the observation they derived from.
const ExtractorVersion
const ExtractorVersion = "extract/v1"
ExtractorVersion is the code label for direct callers with no configured model identity. Runtime formation supplies a configuration fingerprint instead; this label does not identify model weights.
var ErrNotSpokenByPrincipal
var ErrNotSpokenByPrincipal = errors.New("claim is not the principal's to make")
ErrNotSpokenByPrincipal is returned when a claim would become a fact about the principal but the message it came from was not the principal speaking.
var ErrUnboundSpeaker
var ErrUnboundSpeaker = errors.New("speaker reference needs the observation's attributed user")
ErrUnboundSpeaker refuses a relative reference without matching stored user attribution.
var ErrConflictingValue
var ErrConflictingValue = errors.New("a second current value for a single-cardinality relation")
ErrConflictingValue is returned when a claim would be a second current value for a single-cardinality relation whose existing value it cannot supersede, because both start at the same instant. The exclusion constraint is what refuses it; this names the refusal so a caller can record it as one rather than retry it. The database error stays in the chain, so the constraint name is readable.
method FactStore.Assert
func (s *FactStore) Assert(ctx context.Context, schema Schema, scope, observationID string,
speaker domain.Role, dataSubjectID string, claim domain.Claim) (string, error)
Assert resolves both entity ends and writes the fact with its evidence, in one transaction.
The role policy is enforced here, not at the edge
It could be enforced in the handler, and then every future write path would have to remember to. Here it is on the only road to the fact table, so a path that forgets does not exist.
A claim from an assistant or a tool message is refused as a fact ABOUT THE PRINCIPAL. That is not the same as refusing to remember it: the message is already stored and retrievable as a passage. What is refused is the promotion of a model's sentence into something the person is recorded as having said, because once promoted nothing distinguishes it from something they actually told us.
method FactStore.AssertVersioned
func (s *FactStore) AssertVersioned(ctx context.Context, schema Schema, scope, observationID string,
speaker domain.Role, dataSubjectID string, claim domain.Claim, version string) (string, error)
AssertVersioned preserves the configured extraction identity on new evidence. Retained records and source receipts keep their original stamps; changing a source's pinned identity is refused.
func assertFactTx
func assertFactTx(ctx context.Context, tx pgx.Tx, schema Schema, scope, observationID string,
speaker domain.Role, dataSubjectID string, claim domain.Claim, extractorVersion string, knownAt time.Time) (string, error)
Human corrections share the same source, role and temporal write boundary; only provenance differs.
func assertFactVersionTx
func assertFactVersionTx(ctx context.Context, tx pgx.Tx, schema Schema, scope, observationID string,
speaker domain.Role, dataSubjectID string, claim domain.Claim, extractorVersion string, knownAt time.Time, generation uuid.UUID) (string, error)
A generation uses the same attribution, withdrawal, entity and evidence boundary as an ordinary assertion. Only its identity and temporal reconciliation differ; its caller owns atomic cutover.
func nullIfEmptyString
func nullIfEmptyString(s string) any
func resolveEntity
func resolveEntity(ctx context.Context, tx pgx.Tx, schema Schema, scope, observationID, name, entityType,
dataSubjectID string, isSpeaker bool) (string, error)
resolveEntity finds or creates the entity a name refers to, inside the caller's transaction.
Resolution is by normalised name and deliberately by nothing cleverer
Lowercase, trim, collapse whitespace. That merges the overwhelming majority of what an extractor produces for one thing, because it is reading the same words back in different sentences.
It does NOT merge "Dublin office" with "our Dublin site", and nothing here pretends to. Doing that needs an embedding comparison with a threshold nobody can justify, or a model call per entity at write time. Under-merging leaves two nodes where there should be one, costing recall on a hop. Over-merging puts two people's facts on one node, which is a governance failure. The asymmetry decides it.
Why the type is not part of identity
With the type in the key, one thing extracted once as a place and once as a thing becomes two nodes holding two halves of what is known about it — a silent under-merge on every hop, invisible because both rows look correct on their own. The type is an attribute, kept as first seen.
Why this runs in the caller's transaction
A failed assert must leave no entity behind. Resolved separately, it would leave orphans for facts that do not exist — harmless individually, and exactly the kind of drift that makes an entity count stop meaning anything.
const upsertEntitySQL
const upsertEntitySQL = `
INSERT INTO {schema}.entity (entity_id, scope, canonical_name, normalized_name, entity_type)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (scope, normalized_name) WHERE identity_kind = 'named' DO UPDATE
SET resolution_version = {schema}.entity.resolution_version
RETURNING entity_id::text`
const insertFactSQL
const insertFactSQL = `
INSERT INTO {schema}.fact
(fact_id, scope, subject_entity_id, object_entity_id, predicate, statement,
valid, confidence, data_subject_id, cardinality, source_role, known, recorded_at, superseded_by, supersession_source)
VALUES ($1, $2, $3::uuid, $4::uuid, $5, $6, tstzrange($7, $13), $8, $9, $10, $11, tstzrange($12,NULL), $12, $14::uuid, $15::uuid)`
`known` defaults to [now, ∞): we believe it from the moment it is recorded. `valid` opens at the claim's own time, because when something became true and when we heard about it are different questions and the whole point of two axes is to answer both.
const supersedeSQL
const supersedeSQL = `
WITH previous AS MATERIALIZED (
SELECT fact_id,scope,valid,known FROM {schema}.fact
WHERE scope=$1 AND subject_entity_id=$2::uuid AND predicate=$3
AND upper_inf(valid) AND upper_inf(known) AND lower(valid)<$4
FOR UPDATE
), saved AS (
INSERT INTO {schema}.fact_history(history_id,scope,fact_id,valid,known,source_observation_id)
SELECT gen_random_uuid(),p.scope,p.fact_id,p.valid,
tstzrange(lower(p.known),$5),$6::uuid FROM previous p
RETURNING history_id,fact_id
), registered AS (
INSERT INTO {schema}.projection_dependency(source_observation_id,scope,projection_kind,projection_id,data_subject_id)
SELECT d.source_observation_id,d.scope,'fact_history',s.history_id::text,d.data_subject_id
FROM saved s JOIN {schema}.projection_dependency d ON d.projection_kind='fact' AND d.projection_id=s.fact_id::text AND d.scope=$1
UNION
SELECT $6::uuid,$1,'fact_history',s.history_id::text,$8::text FROM saved s
ON CONFLICT DO NOTHING
)
UPDATE {schema}.fact f SET valid=tstzrange(lower(f.valid),$4),known=tstzrange($5,NULL),
superseded_by=$7::uuid,supersession_source=$6::uuid
WHERE f.fact_id IN (SELECT fact_id FROM saved)`
Closes whatever this subject currently holds for this relation, at the moment the new one starts.
`upper_inf(valid)` rather than "contains now": the interval being closed is the OPEN one, and a fact that was already given an end is history that must not be rewritten. Ends are set to the new fact's start rather than to now(), so the two intervals meet exactly and a question about any instant has one answer.
A range that would become empty or backwards — a fact arriving with a timestamp before the current one started — is left alone. Out-of-order arrival is real, and the right answer to it is a bitemporal correction rather than an interval this statement invents.
const stampFactRetentionSQL
const stampFactRetentionSQL = `
UPDATE {schema}.fact f
SET retention_until = s.deadline
FROM (
SELECT CASE WHEN bool_or(o.retention_until IS NULL) THEN NULL ELSE max(o.retention_until) END AS deadline
FROM {schema}.fact_evidence e
JOIN {schema}.observation o ON o.observation_id = e.source_observation_id AND o.scope = e.scope
WHERE e.scope = $2 AND e.fact_id = $1
) s
WHERE f.scope = $2 AND f.fact_id = $1
AND f.retention_until IS DISTINCT FROM s.deadline`
The latest deadline among the observations supporting this fact, or none when any of them is kept indefinitely: a fact is reachable while anything behind it is.
const insertEvidenceSQL
const insertEvidenceSQL = `
INSERT INTO {schema}.fact_evidence
(fact_id, source_observation_id, source_ordinal, quote, byte_start, byte_end, extractor_version, scope)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`
const ProjectionRejectedClaim
const ProjectionRejectedClaim = "rejected_claim"
ProjectionRejectedClaim is the projection kind a rejected claim registers as.
It registers like anything else, because a refused claim still holds somebody's verbatim words. A table of "things we did not store" that turns out to store them is the worst version of this feature, and it is invisible to erasure unless it is here.
method FactStore.Reject
func (s *FactStore) Reject(ctx context.Context, schema Schema, scope, observationID, dataSubjectID string,
rejected domain.RejectedClaim) error
Reject records a proposal that did not become a fact.
Why the refusal is written down rather than counted
Both refusals are ordinary — most messages assert nothing, and a model that paraphrases is behaving normally — so neither is an error and neither belongs in an error log. But both are claims about quality that nobody can check without the words: a rate of eleven percent cannot distinguish one missing relation asserted five hundred times, where the vocabulary is wrong and the fix is one row, from five hundred relations asserted once, which is the tail and is noise.
Why it is not part of Assert
A rejected claim never reaches Assert: it has no admitted predicate, or no locatable span, so there is no fact to write and no entity to resolve. Folding it in would mean a function whose success case sometimes writes a fact and sometimes writes the reason it did not.
method FactStore.RejectVersioned
func (s *FactStore) RejectVersioned(ctx context.Context, schema Schema, scope, observationID, dataSubjectID string,
rejected domain.RejectedClaim, version string) error
Rejected text belongs to the extraction configuration that proposed it, including stable retries.
const insertRejectedClaimSQL
const insertRejectedClaimSQL = `
INSERT INTO {schema}.rejected_claim
(rejected_claim_id, scope, source_observation_id, source_ordinal, predicate, statement, quote,
reason, extractor_version, data_subject_id)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
ON CONFLICT (rejected_claim_id) DO NOTHING`