Trait DataStore 

pub trait DataStore: Send + Sync {
    // Required methods
    fn manifest(&self) -> &AppManifest;
    fn insert(&self, entity: &str, data: &Value) -> Result<String, DataError>;
    fn get_by_id(
        &self,
        entity: &str,
        id: &str,
    ) -> Result<Option<Value>, DataError>;
    fn list(&self, entity: &str) -> Result<Vec<Value>, DataError>;
    fn list_after(
        &self,
        entity: &str,
        after: Option<&str>,
        limit: usize,
    ) -> Result<Vec<Value>, DataError>;
    fn update(
        &self,
        entity: &str,
        id: &str,
        data: &Value,
    ) -> Result<bool, DataError>;
    fn delete(&self, entity: &str, id: &str) -> Result<bool, DataError>;
    fn lookup(
        &self,
        entity: &str,
        field: &str,
        value: &str,
    ) -> Result<Option<Value>, DataError>;
    fn link(
        &self,
        entity: &str,
        id: &str,
        relation: &str,
        target_id: &str,
    ) -> Result<bool, DataError>;
    fn unlink(
        &self,
        entity: &str,
        id: &str,
        relation: &str,
    ) -> Result<bool, DataError>;
    fn query_filtered(
        &self,
        entity: &str,
        filter: &Value,
    ) -> Result<Vec<Value>, DataError>;
    fn query_graph(&self, query: &Value) -> Result<Value, DataError>;
    fn transact(&self, ops: &[Value]) -> Result<(bool, Vec<Value>), DataError>;

    // Provided methods
    fn aggregate(
        &self,
        _entity: &str,
        _spec: &Value,
    ) -> Result<Value, DataError> { ... }
    fn search(&self, _entity: &str, _query: &Value) -> Result<Value, DataError> { ... }
    fn crdt_snapshot(
        &self,
        _entity: &str,
        _row_id: &str,
    ) -> Result<Option<Vec<u8>>, DataError> { ... }
    fn crdt_apply_update(
        &self,
        _entity: &str,
        _row_id: &str,
        _update: &[u8],
    ) -> Result<Vec<u8>, DataError> { ... }
}

Platform-agnostic data store trait.

Implemented by Runtime (SQLite, self-hosted) and D1DataStore (Workers). All methods are synchronous to keep the trait Send + Sync and simple; Workers adapters can use block_on or similar bridging.

Required Methods

fn manifest(&self) -> &AppManifest

fn insert(&self, entity: &str, data: &Value) -> Result<String, DataError>

fn get_by_id(&self, entity: &str, id: &str) -> Result<Option<Value>, DataError>

fn list(&self, entity: &str) -> Result<Vec<Value>, DataError>

fn list_after( &self, entity: &str, after: Option<&str>, limit: usize, ) -> Result<Vec<Value>, DataError>

fn update( &self, entity: &str, id: &str, data: &Value, ) -> Result<bool, DataError>

fn delete(&self, entity: &str, id: &str) -> Result<bool, DataError>

fn lookup( &self, entity: &str, field: &str, value: &str, ) -> Result<Option<Value>, DataError>

fn link( &self, entity: &str, id: &str, relation: &str, target_id: &str, ) -> Result<bool, DataError>

fn unlink( &self, entity: &str, id: &str, relation: &str, ) -> Result<bool, DataError>

fn query_filtered( &self, entity: &str, filter: &Value, ) -> Result<Vec<Value>, DataError>

fn query_graph(&self, query: &Value) -> Result<Value, DataError>

fn transact(&self, ops: &[Value]) -> Result<(bool, Vec<Value>), DataError>

Execute transactional operations. Each element is a JSON object with op (“insert”/“update”/“delete”), entity, and optionally id/data.

Returns per-operation results. The implementation decides whether to use real SQL transactions (Runtime) or sequential execution (D1).

Provided Methods

fn aggregate(&self, _entity: &str, _spec: &Value) -> Result<Value, DataError>

Run an aggregation query.

Spec shape (same vocabulary in the HTTP body):

{
  "count": "*",
  "sum": ["amount"],
  "avg": ["price"],
  "min": ["createdAt"],
  "max": ["createdAt"],
  "groupBy": ["status"],
  "where": { ...standard filter... }
}

Returns {rows: [{count, sum_amount, ...}]}. Default implementation returns NOT_SUPPORTED; Runtime overrides it.

fn search(&self, _entity: &str, _query: &Value) -> Result<Value, DataError>

Run a faceted full-text search against a searchable entity. query is a JSON object with the keys defined by SearchQuery in pylon_storage::search; returns a JSON object shaped like SearchResult ({ hits, facetCounts, total, tookMs }).

Default impl returns NOT_SUPPORTED; Runtime overrides it. The value is raw JSON (not a typed struct) so backends without a dependency on pylon-storage can still compile.

fn crdt_snapshot( &self, _entity: &str, _row_id: &str, ) -> Result<Option<Vec<u8>>, DataError>

Return the binary CRDT snapshot for a row, used by the router to ship a binary update over WebSocket after every successful write.

Return value semantics:

• Ok(Some(bytes)) — entity is CRDT-mode and bytes are the current Loro snapshot for the row.
• Ok(None) — either the entity is crdt: false (LWW opt-out) or this backend doesn’t support CRDT mode at all. Callers MUST treat both cases identically: skip the binary broadcast and rely on the JSON change event for client invalidation. The conflation is intentional — every caller today does the same thing in both cases, and a richer enum (NotCrdtMode / NotSupported) would be carried through every layer for no behavioral payoff.
• Err(_) — entity is CRDT-mode but the snapshot fetch itself failed (schema lookup, sidecar read, decode). Log and continue; the JSON change event already covers the correctness path.

Default impl returns Ok(None) so backends that don’t support CRDT mode (e.g. the Workers D1 store at time of writing) compile without ceremony. Per the Ok(None) semantics above, this is correct behavior, not a stub.

fn crdt_apply_update( &self, _entity: &str, _row_id: &str, _update: &[u8], ) -> Result<Vec<u8>, DataError>

Apply a binary CRDT update from a client to the row’s LoroDoc, project the new state into the SQLite materialized view, and return the post-merge snapshot bytes (so the caller can broadcast them to OTHER subscribed clients).

update is opaque Loro bytes — either a snapshot or an incremental delta. Loro’s import contract accepts both shapes, so the store doesn’t need to know which the client sent.

Errors:

• ENTITY_NOT_FOUND — unknown entity in the manifest.
• NOT_SUPPORTED — entity is crdt: false (LWW opt-out) or the backend doesn’t implement CRDT mode.
• CRDT_DECODE_FAILED — bytes weren’t a valid Loro update.
• Storage failures from the underlying SQLite write.

Default impl returns NOT_SUPPORTED so backends without CRDT support compile cleanly.

Implementors