internal/infra/pg/recordstore.go
internal/infra/pg · 235 lines · 16 declarations · source
Declarations
const DefaultRecordPage, MaxRecordPage, RecordPreviewCharacters and 1 more
const (
DefaultRecordPage = 20
MaxRecordPage = 100
RecordPreviewCharacters = 512
MaxRecordSubjectBytes = 1 << 20
)
var ErrInvalidRecordPage, ErrRecordNotFound
var (
ErrInvalidRecordPage = errors.New("invalid record page")
ErrRecordNotFound = errors.New("record not found")
)
type RecordStore
type RecordStore struct {
pool *pgxpool.Pool
schema Schema
}
RecordStore provides bounded inspection of retained facts. It does not infer new claims or verify source text; a client follows the record ID through citation resolution for that evidence.
func NewRecordStore
func NewRecordStore(pool *pgxpool.Pool, schema Schema) *RecordStore
type RecordCursor
type RecordCursor struct {
ID string `json:"id"`
}
RecordCursor is a position, never an authorization capability. Project and subject predicates are reapplied on every page, even if a caller manufactures or reuses a cursor.
type RecordSummary
type RecordSummary struct {
Version string `json:"version"`
ID string `json:"id"`
SubjectID *string `json:"subject_entity_id"`
ObjectID *string `json:"object_entity_id"`
Predicate string `json:"predicate"`
StatementPreview string `json:"statement_preview"`
StatementBytes int `json:"statement_bytes"`
PreviewTruncated bool `json:"preview_truncated"`
SourceRole string `json:"source_role"`
RecordedAt time.Time `json:"recorded_at"`
Status string `json:"status"`
}
type RecordPage
type RecordPage struct {
Records []RecordSummary `json:"records"`
Next *RecordCursor `json:"next,omitempty"`
}
method RecordStore.List
func (s *RecordStore) List(ctx context.Context, scope, subject string, after *RecordCursor, limit int) (RecordPage, error)
List orders by immutable UUID, so closing validity or deleting the cursor row cannot move a returned record into a later page. This is a live inventory: inserts behind the cursor require restarting the browse. OFFSET would both rescan earlier rows and shift pages after deletion.
const recordListSQL
const recordListSQL = `WITH page AS MATERIALIZED (
SELECT fact_id FROM {schema}.fact WHERE scope=$1 AND fact_id {comparison} $2::uuid
AND (retention_until IS NULL OR retention_until > now())
ORDER BY fact_id LIMIT $3
)
SELECT f.version::text,f.fact_id::text,f.subject_entity_id::text,f.object_entity_id::text,f.predicate,
left(f.statement,$4),octet_length(f.statement),f.source_role,f.recorded_at,
CASE WHEN NOT upper_inf(f.known) THEN CASE WHEN EXISTS(SELECT 1 FROM {schema}.record_retraction rr WHERE rr.scope=f.scope AND rr.target_fact_id=f.fact_id) THEN 'retracted' ELSE 'knowledge_closed' END WHEN NOT upper_inf(f.valid) THEN 'validity_closed' ELSE 'current' END
FROM page p JOIN {schema}.fact f ON f.scope=$1 AND f.fact_id=p.fact_id
ORDER BY f.fact_id`
The page is selected before reading statement text. The subject path first deduplicates indexed provenance references, so one shared fact appears once even with several supporting observations.
const subjectRecordListSQL
const subjectRecordListSQL = `WITH page AS MATERIALIZED (
SELECT DISTINCT fact_ref FROM {schema}.projection_dependency
WHERE scope=$1 AND data_subject_id=$5 AND projection_kind='fact' AND fact_ref {comparison} $2::uuid
ORDER BY fact_ref LIMIT $3
)
SELECT f.version::text,f.fact_id::text,f.subject_entity_id::text,f.object_entity_id::text,f.predicate,
left(f.statement,$4),octet_length(f.statement),f.source_role,f.recorded_at,
CASE WHEN NOT upper_inf(f.known) THEN CASE WHEN EXISTS(SELECT 1 FROM {schema}.record_retraction rr WHERE rr.scope=f.scope AND rr.target_fact_id=f.fact_id) THEN 'retracted' ELSE 'knowledge_closed' END WHEN NOT upper_inf(f.valid) THEN 'validity_closed' ELSE 'current' END
FROM page p JOIN {schema}.fact f ON f.scope=$1 AND f.fact_id=p.fact_ref ORDER BY f.fact_id`
type RecordTemporalState
type RecordTemporalState struct {
Valid CitationInterval `json:"valid"`
Known CitationInterval `json:"known"`
}
type RecordHistoryVersion
type RecordHistoryVersion struct {
ID string `json:"history_id"`
RecordTemporalState
EndedByObservationID string `json:"ended_by_observation_id"`
}
type RecordHistoryCursor
type RecordHistoryCursor struct {
KnownUntil time.Time `json:"known_until"`
ID string `json:"history_id"`
}
type RecordHistoryPage
type RecordHistoryPage struct {
Version string `json:"version"`
Retraction *RetractionDetails `json:"retraction,omitempty"`
ID string `json:"id"`
Current RecordTemporalState `json:"current"`
Previous []RecordHistoryVersion `json:"previous"`
Next *RecordHistoryCursor `json:"next,omitempty"`
}
method RecordStore.History
func (s *RecordStore) History(ctx context.Context, scope, id string, before *RecordHistoryCursor, limit int) (RecordHistoryPage, error)
History reads the current state and its archive in one snapshot. Across pages history is live: erasure may remove a record and a later supersession may add a newer version before the cursor. The archive contains intervals, not copies of entity labels or evidence that never changed here.
const recordHistorySQL
const recordHistorySQL = `SELECT history_id::text,lower(valid),upper(valid),lower_inc(valid),upper_inc(valid),
lower(known),upper(known),lower_inc(known),upper_inc(known),source_observation_id::text
FROM {schema}.fact_history h WHERE h.scope=$1 AND h.fact_id=$2 {cursor}
AND EXISTS (SELECT 1 FROM {schema}.fact f WHERE f.scope=h.scope AND f.fact_id=h.fact_id
AND (f.retention_until IS NULL OR f.retention_until > now()))
ORDER BY upper(known) DESC,history_id DESC LIMIT $3`