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