postgres

package
v0.1.0-round1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 25, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Overview

Package postgres implements SyncBase persistence against PostgreSQL/OpenSQL-compatible SQL.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Migrate

func Migrate(ctx context.Context, pool *pgxpool.Pool, profile knowledge.Profile, canonical string) error

Migrate installs the stable schema and verifies the immutable processing profile.

func Open

func Open(ctx context.Context, databaseURL string) (*pgxpool.Pool, error)

Open returns a bounded PostgreSQL connection pool after a successful ping.

Types

type SessionStore

type SessionStore struct {
	// contains filtered or unexported fields
}

SessionStore persists only a SHA-256 digest of each browser bearer token.

func NewSessionStore

func NewSessionStore(pool *pgxpool.Pool) *SessionStore

NewSessionStore returns a PostgreSQL-backed browser session store.

func (*SessionStore) Create

func (s *SessionStore) Create(ctx context.Context, token string, record sessions.Record) error

Create writes a new browser session without storing the raw bearer token.

func (*SessionStore) Delete

func (s *SessionStore) Delete(ctx context.Context, token string) error

Delete revokes token immediately. A missing token is already revoked.

func (*SessionStore) Load

func (s *SessionStore) Load(ctx context.Context, token string, now time.Time) (sessions.Record, bool, error)

Load returns an unexpired session for token. The query atomically rejects expired records so multiple web instances share the same expiry decision.

type Store

type Store struct {
	// contains filtered or unexported fields
}

Store implements document, processing, and search persistence in PostgreSQL.

func NewStore

func NewStore(pool *pgxpool.Pool) *Store

NewStore returns a PostgreSQL adapter over an initialized connection pool.

func (*Store) ClaimNext

func (s *Store) ClaimNext(ctx context.Context, workerID string) (*knowledge.ClaimedRun, error)

ClaimNext exclusively leases the oldest queued processing run. The queue is deliberately single-flight for the P0 release; fencing tokens make an old worker harmless after its lease expires and the run is reclaimed.

func (*Store) ExpireRegistration

func (s *Store) ExpireRegistration(ctx context.Context, command knowledge.ReserveUploadCommand) error

ExpireRegistration closes a PENDING reservation after a definitive input rejection. Transient failures deliberately leave PENDING intact for retry.

func (*Store) Fail

func (s *Store) Fail(ctx context.Context, claimed knowledge.ClaimedRun, stage knowledge.Stage, code string) error

Fail terminates a run without exposing unsafe error details. The fencing predicate prevents an expired worker from overwriting a newer attempt.

Retriable failure codes (TEMPORARILY_UNAVAILABLE, INTERNAL) are re-queued on attempts 1 and 2 with +1s / +5s backoff; on attempt 3 they are renamed to TRANSIENT_EXHAUSTED so the operator can trigger a manual retry. Non-retriable codes (e.g. INVALID_INPUT, PROFILE_MISMATCH) still fail immediately.

func (*Store) Finalize

func (s *Store) Finalize(ctx context.Context, claimed knowledge.ClaimedRun, pageCount, chunkCount int) error

Finalize makes a successfully indexed version active. A lower version can never replace a higher version that is already active.

func (*Store) FindDocumentsByNormalizedName

func (s *Store) FindDocumentsByNormalizedName(
	ctx context.Context,
	normalizedName string,
	limit int,
) ([]knowledge.DocumentSummary, int, error)

FindDocumentsByNormalizedName returns a bounded newest-first sample and the total number of exact normalized-name matches. Display names are not unique.

func (*Store) GetDocument

func (s *Store) GetDocument(ctx context.Context, documentID uuid.UUID) (knowledge.DocumentDetails, error)

GetDocument returns one document and its newest-first version history.

func (*Store) GetSource

func (s *Store) GetSource(
	ctx context.Context,
	documentID uuid.UUID,
	versionNumber int,
) (knowledge.SourceDocument, error)

GetSource returns the immutable original metadata for an exact version.

func (*Store) HasInactiveMatch

func (s *Store) HasInactiveMatch(
	ctx context.Context,
	profile knowledge.Profile,
	query []float32,
) (bool, error)

HasInactiveMatch reports whether the same above-policy vector query matches only evidence that is not the document's currently active version. It returns a boolean only: inactive snippets and source locations never cross the adapter boundary.

func (*Store) Heartbeat

func (s *Store) Heartbeat(ctx context.Context, runID uuid.UUID, fence int64, workerID string) error

Heartbeat extends the lease only for its current fenced owner.

func (*Store) ListDocuments

func (s *Store) ListDocuments(ctx context.Context, limit, offset int) ([]knowledge.DocumentSummary, error)

ListDocuments returns a deterministic, bounded page of document summaries.

func (*Store) LoadCheckpoint

func (s *Store) LoadCheckpoint(
	ctx context.Context,
	claimed knowledge.ClaimedRun,
	stage knowledge.Stage,
	inputSHA256 string,
) (*knowledge.Checkpoint, error)

LoadCheckpoint returns only an artifact whose input chain matches the currently fenced run. The artifact bytes and output digest are verified by the worker before use.

func (*Store) Ready

func (s *Store) Ready(ctx context.Context) error

Ready reports whether PostgreSQL accepts a request within the caller's context.

func (*Store) RecoverRegistration

func (s *Store) RecoverRegistration(ctx context.Context, requestKey string) (knowledge.UploadRecovery, error)

RecoverRegistration resolves the durable state of an idempotent upload key.

func (*Store) Register

Register atomically creates an idempotent document version and processing run. It deliberately does not lock queue_control: worker fencing and upload admission are separate concerns, and registration must not wait behind a worker's singleton claim transaction.

func (*Store) ReserveRegistration

func (s *Store) ReserveRegistration(ctx context.Context, command knowledge.ReserveUploadCommand) (knowledge.UploadRecovery, error)

ReserveRegistration durably records the idempotency identity before parsing or Original storage. A recovery request can therefore distinguish a genuine in-flight registration from a request that never reached the API.

func (*Store) Retry

func (s *Store) Retry(ctx context.Context, parentRunID uuid.UUID, requestKey string) (uuid.UUID, error)

Retry creates one idempotent child run for a failed version. It deliberately restarts the deterministic pipeline from metadata; content-addressed originals and staged-chunk replacement make that replay safe.

func (*Store) SaveCheckpoint

func (s *Store) SaveCheckpoint(
	ctx context.Context,
	claimed knowledge.ClaimedRun,
	checkpoint knowledge.Checkpoint,
) error

SaveCheckpoint atomically attaches a content-addressed artifact to the currently fenced run. Reclaims may replace an older fence's checkpoint.

func (*Store) Search

func (s *Store) Search(
	ctx context.Context,
	profile knowledge.Profile,
	query []float32,
	limit int,
	baseURL string,
) ([]knowledge.SearchHit, error)

Search returns chunks from active versions only, using the immutable active profile and the public, bounded score formula.

func (*Store) SetStage

func (s *Store) SetStage(ctx context.Context, runID uuid.UUID, fence int64, stage knowledge.Stage) error

SetStage advances the observable processing stage when the caller still owns the current lease.

func (*Store) StorageKeyReferenced

func (s *Store) StorageKeyReferenced(ctx context.Context, storageKey string) (bool, error)

StorageKeyReferenced reports whether a committed version references an object.

func (*Store) StoreChunks

func (s *Store) StoreChunks(ctx context.Context, claimed knowledge.ClaimedRun, chunks []knowledge.IndexedChunk) error

StoreChunks atomically replaces staged chunks for one version and profile.

func (*Store) VerifyProfile

func (s *Store) VerifyProfile(ctx context.Context, profile knowledge.Profile) error

VerifyProfile rejects a runtime whose processing profile differs from the database.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL