Skip to main content

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
)

source

var ErrInvalidRecordPage, ErrRecordNotFound​

var (
ErrInvalidRecordPage = errors.New("invalid record page")
ErrRecordNotFound = errors.New("record not found")
)

source

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.

source

func NewRecordStore​

func NewRecordStore(pool *pgxpool.Pool, schema Schema) *RecordStore

source

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.

source

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"`
}

source

type RecordPage​

type RecordPage struct {
Records []RecordSummary `json:"records"`
Next *RecordCursor `json:"next,omitempty"`
}

source

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.

source

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.

source

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`

source

type RecordTemporalState​

type RecordTemporalState struct {
Valid CitationInterval `json:"valid"`
Known CitationInterval `json:"known"`
}

source

type RecordHistoryVersion​

type RecordHistoryVersion struct {
ID string `json:"history_id"`
RecordTemporalState
EndedByObservationID string `json:"ended_by_observation_id"`
}

source

type RecordHistoryCursor​

type RecordHistoryCursor struct {
KnownUntil time.Time `json:"known_until"`
ID string `json:"history_id"`
}

source

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"`
}

source

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.

source

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`

source