Persistence and Data Governance

Persistence and Data Governance

Author: Ciprian Ștefan Pleșca

This page documents src/openlongevity/db.py, src/openlongevity/repository.py, and sql/schema.sql — the layer responsible for turning a normalized Publication into durable, auditable storage with an immutable revision history.

Schema

erDiagram
    PUBLICATIONS ||--o{ PUBLICATION_REVISIONS : "has revisions"

    PUBLICATIONS {
        string identifier PK
        string provider
        string source_identifier
        text title
        json payload
        string content_hash
        int revision
        string first_retrieved_at
        string last_retrieved_at
    }

    PUBLICATION_REVISIONS {
        string publication_id FK
        int revision PK
        json payload
        string content_hash
        string retrieved_at
    }

Loading

PublicationRow carries a UniqueConstraint on (provider, source_identifier), so the same upstream record cannot be inserted twice under two different local identifiers from the same provider. PublicationRevisionRow uses a composite primary key of (publication_id, revision), and cascades on delete from its parent — deleting a publication removes its full revision trail atomically.

A parallel, simpler schema exists in sql/schema.sql for standalone SQL deployment, declaring publications and evidence_records tables with CHECK constraints mirroring the Python-level enumerations (for example, retraction_status constrained to the same five values as RetractionStatus), keeping the database schema and the application's type system aligned even when the ORM path is not used.

Save algorithm: change detection without spurious revisions

The central correctness property of PublicationRepository.save() is: re-ingesting an unchanged publication must not create a new revision, while a genuinely changed publication must. This is achieved by hashing a stabilized copy of the payload that excludes the volatile retrieved_at timestamp before comparing it to the stored content_hash.

flowchart TD
    Start(["save(publication)"]) --> Check{"provenance present\nand checksum non-empty?"}
    Check -- no --> Reject[raise ValueError]
    Check -- yes --> Stable["asdict(publication), drop provenance.retrieved_at"]
    Stable --> Digest["SHA-256(json.dumps(stable, sort_keys=True))"]
    Digest --> Tx[begin transaction]
    Tx --> Insert["INSERT ... ON CONFLICT (identifier) DO NOTHING\nRETURNING identifier"]
    Insert --> Select["SELECT ... WHERE identifier = ? FOR UPDATE"]
    Select --> RowCheck{"row is None?"}
    RowCheck -- yes --> Fatal["raise RuntimeError: insert did not produce a row"]
    RowCheck -- no --> Created{"was this INSERT the\none that created the row?"}
    Created -- yes --> Fresh["changed = True"]
    Created -- no --> Compare{"stored content_hash\n!= new digest?"}
    Compare -- yes --> Changed["changed = True"]
    Compare -- no --> Unchanged["changed = False"]
    Fresh --> Freshness
    Changed --> Bump["revision += 1\nupdate payload, content_hash, title"]
    Bump --> Freshness
    Unchanged --> Freshness["last_retrieved_at = max(existing, new)"]
    Freshness --> WriteRev{"changed?"}
    WriteRev -- yes --> RevRow["INSERT PublicationRevisionRow\n(publication_id, revision, payload, content_hash, retrieved_at)"]
    WriteRev -- no --> Commit
    RevRow --> Commit[commit transaction]
    Commit --> Serialize["return serialize(row)"]

Loading

Why ON CONFLICT DO NOTHING followed by SELECT ... FOR UPDATE

This two-step pattern handles concurrent first-time ingestion of the same publication safely: if two ingestion requests race to insert the same identifier, PostgreSQL's ON CONFLICT DO NOTHING ensures only one insert succeeds, and the subsequent SELECT ... FOR UPDATE acquires a row lock on whichever row exists — the winner's own freshly inserted row, or the row inserted by the concurrent request — before either transaction decides whether a revision bump is needed. This avoids a lost-update race where both requests believe they are the first to insert.

last_retrieved_at versus revisions

Every successful save() call updates last_retrieved_at to the maximum of the existing and new timestamps, regardless of whether content changed. This distinguishes two independently meaningful facts: "when was this record last confirmed to still exist upstream" (freshness) and "when did its content actually change" (a revision). A publication ingested identically every day for a year has one revision but a last_retrieved_at updated 365 times.

Sequence: full save with a genuine content change

sequenceDiagram
    autonumber
    participant Repo as PublicationRepository
    participant DB as PostgreSQL

    Repo->>Repo: compute stable digest (excludes retrieved_at)
    Repo->>DB: BEGIN
    Repo->>DB: INSERT ... ON CONFLICT DO NOTHING RETURNING identifier
    DB-->>Repo: no row returned (already existed)
    Repo->>DB: SELECT ... WHERE identifier = ? FOR UPDATE
    DB-->>Repo: existing row (content_hash = old digest)
    Repo->>Repo: old digest != new digest -> changed = True
    Repo->>Repo: row.revision += 1
    Repo->>DB: UPDATE payload, content_hash, title, last_retrieved_at
    Repo->>DB: INSERT INTO publication_revisions (revision, payload, content_hash, retrieved_at)
    Repo->>DB: COMMIT
    Repo-->>Repo: return serialize(row) with synthetic=False

Loading

Data governance principles reflected in the code

1. Provenance is a storage precondition, not an optional field — enforced by the ValueError guard at the top of save(), directly implementing the repository's DATA_GOVERNANCE.md commitment that "every imported record should retain source identifier, retrieval date, license, and transformation history."
2. History is append-only. Nothing in repository.py updates or deletes rows in publication_revisions; revisions accumulate for as long as the parent publication exists.
3. Search is bounded and explicit about its scope. PublicationRepository.list() performs a title-only, case-insensitive, auto-escaped ILIKE-style match (icontains(query, autoescape=True)) — the README is explicit that "search currently filters publication titles; it is not a full-text engine or a federated systematic-review search," and the implementation matches that description exactly, including the autoescaping that prevents a user-supplied % or _ from being treated as a SQL wildcard.
4. Health checks probe migration state, not just connectivity. Database.health() queries the alembic_version table rather than merely opening a connection — a database that is reachable but has never been migrated correctly reports "unavailable".

Next

See API-Reference for the HTTP routes backed by this layer, and Scientific-Methodology-and-Limitations for the boundary between "persisted metadata" and "scientific evidence."