data-store

Getting started

Install#

npm i @dmytromykhailiuk/data-store

Initialize the network layer once, at app startup, before the store starts:

import { NetworkConnection } from "@dmytromykhailiuk/network-connection";

await NetworkConnection.init("/healthcheck.txt", { pingInterval: 15_000 });

Getting started

Quick start#

import { NetworkConnection } from "@dmytromykhailiuk/network-connection";
import { ConflictError, createDataStore, defineTable } from "@dmytromykhailiuk/data-store";

interface Issue {
  id: string;
  eventId: string;
  severity: number;
  status: "OPEN" | "RESOLVED";
  _version?: number;
}

interface SyncParams {
  eventId?: string;
}

// 1. Declare tables: model type + the slice of params they consume.
const store = createDataStore<SyncParams>({
  name: "InspectorDataStore",
  schemaVersion: 1,
  ignoreFields: ["_version", "updatedAt", "owner"],
  tables: {
    Issue: defineTable<Issue, SyncParams>({
      primaryKey: "id",
      indexes: { byEvent: { key: "eventId" } },
      scope: {
        key: (params) => params.eventId ?? null,
        keep: (issue, params) => issue.eventId === params.eventId,
      },
      push: {
        create: async (issue) => (await api.createIssue(issue)).data,
        update: async (issue) => {
          try {
            return (await api.updateIssue(issue)).data;
          } catch (error) {
            if (api.isVersionMismatch(error)) {
              throw new ConflictError({ remote: api.remoteOf(error) });
            }
            throw error; // transport errors retry with backoff on their own
          }
        },
        delete: async (issue) => {
          await api.deleteIssue(issue.id, issue._version);
        },
      },
      pull: {
        fetch: async ({ params, checkpoint, signal }) => {
          const page = await api.issuesByEvent(params.eventId!, { since: checkpoint, signal });
          return { items: page.items, checkpoint: page.startedAt, done: !page.nextToken };
        },
        fetchOne: (id) => api.getIssue(id),
        deletedKey: undefined, // set to "_deleted" for server tombstones
      },
      merge: (local, remote) => ({ ...remote, status: local.status }),
    }),
  },
});

// 2. Start: opens IndexedDB, restores the outbox, pulls the current scope.
await NetworkConnection.init("/healthcheck.txt");
await store.start({ params: { eventId: route.eventId } });
await store.whenSynced();

// 3. Work with tables — everything below survives reloads and offline.
const issues = store.table("Issue");
await issues.put({ id: crypto.randomUUID(), eventId: route.eventId!, severity: 3, status: "OPEN" });
await issues.update(issueId, { status: "RESOLVED" });
const open = await issues.query({ status: { eq: "OPEN" }, severity: { ge: 2 } });

// 4. React to changes and to the sync engine.
issues.subscribe({ eventId: { eq: route.eventId! } }, ({ type, item, origin }) => {
  render(type, item, origin); // origin: "local" | "pull" | "push" | "merge" | "evict"
});
await store.whenUploaded(); // outbox is empty — safe to log out

Concepts

How it works#

Five rules explain everything the store does. Everything else in this documentation is a consequence of them.

1. IndexedDB is the single source of truth. There is no in-memory mirror to drift out of sync. Reads are plain readonly transactions — consistent snapshots by definition.
2. Records are stored wrapped: { key, data, meta }. Your fields never mix with the bookkeeping (state, revision counter, sync scope, last error).
3. One write path. Every mutation — CRUD, a transaction, an applied pull page, a push acknowledgement — acquires the table's FIFO lock and runs one readwrite IndexedDB transaction spanning the data and the outbox. A write claims its place in that queue before any asynchronous work of its own, so encoding a blob never lets a lighter payload commit ahead of a heavier one issued earlier. Locks are never held across the network.
4. The outbox is data. A queued upload commits in the same transaction as the record it belongs to, and is rebuilt from disk on every start. The in-memory queue is just a dispatcher.
5. The transport is yours. The store calls the push / pull functions a table declares and understands exactly two special errors: ConflictError and FatalPushError.

Record states#

Every record carries a meta.state describing where it stands relative to the server:

A deletion of a server-known record does not remove it immediately: the record becomes a tombstone — hidden from every read, still holding its place in the outbox — and is physically removed only when the server confirms the delete. Records the server never saw are simply dropped, along with their queued create.

The outbox#

Every syncable mutation writes an entry into the internal _outbox store — in the same IndexedDB transaction as the data, so there is no crash window where a record exists but its upload does not (or the reverse). Entries are FIFO by a monotonic sequence number; one entry per record.

Several mutations of one record coalesce into one entry that keeps its place in the queue:

Dispatching runs on a single-handler queue: strict FIFO, one mutation in flight at a time, so a parent record always lands before the child that references it. Transport retries happen inside the handler — a flaky record never loses its place and never lets a younger mutation overtake it.

Pull & checkpoints#

A pull downloads a scope page by page: fetch() returns items plus an optional checkpoint and done flag, and is called again with the new checkpoint until done. Each page is applied in one transaction together with its checkpoint — a crash mid-sync resumes from the last durable page instead of starting over or marking a half-pulled scope synced.

The checkpoint persists per (table, scope) and is handed back on the next sync of the same scope — that is your delta-sync cursor (startedAt, a nextToken, a timestamp — whatever your API paginates by). Records with unpushed local changes are rebased by default: the table's merge runs over (local, remote) and the record stays queued with the merged data — so its eventual push carries fresh server-managed fields instead of a stale base (see Pull handlers).

Sync scopes#

A scope names which slice of server data a table mirrors right now — "issues of event 42", "objects of site A". The scope.key function derives it from the store params; null means "this table cannot sync yet" (a missing route param, say). When setParams changes a table's key, the table re-syncs: records failing scope.keep are evicted (never the ones with unpushed changes), and the new scope pulls — from its own checkpoint, if it ever synced before.

Guides

Reading & writing#

const issues = store.table("Issue");

// Reads — consistent snapshots straight from IndexedDB.
await issues.get(id);                     // T | undefined
await issues.getAll({ limit: 100 });
await issues.getByIndex("byEvent", eventId);
await issues.has(id);
await issues.count({ status: { eq: "OPEN" } });
await issues.meta(id);                    // RecordMeta | undefined

// Writes — serialized per table, atomic with the outbox.
await issues.put(issue);                  // insert or replace → queues create/update
await issues.update(id, { status: "RESOLVED" });          // partial merge
await issues.update(id, (cur) => ({ ...cur, n: cur.n + 1 })); // atomic RMW
await issues.delete(id);                  // tombstone → queues delete
await issues.evict(id);                   // local removal, tells the server nothing
await issues.clear();                     // local wipe of the table + its queue

update() and delete() of a key the table does not hold reject with RecordNotFoundError — loudly and immediately. The functional form of update() is an atomic read-modify-write: it reads its origin after it wins the table lock, so twenty concurrent updates of one record run strictly one at a time in call order, each callback receiving the previous one's committed result, running exactly once and producing exactly one revision. Fifty parallel (n) => n + 1 updates produce exactly +50. The same holds inside store.transaction(), and on tables with blobPaths under active binary encoding — there the callback and the asynchronous encode run under the lock as well (local work only, never the network).

Guides

Queries & the filter DSL#

query(), count() and filtered subscriptions accept either a plain predicate or a typed, serializable filter object. Filters run against full records — there is no hidden projection that silently sees only some fields.

// The DSL — serializable, type-checked per field.
await issues.query({
  and: [
    { status: { in: ["OPEN", "TRIAGED"] } },
    { severity: { between: [2, 4] } },
    { or: [{ title: { contains: query } }, { id: { beginsWith: query } }] },
    { not: { assignee: { eq: null } } },
  ],
}, { limit: 50 });

// A predicate — for anything the DSL cannot say.
await issues.query((issue) => issue.tags.length > 3 && isHot(issue));

By default a query is purely local — a readonly IndexedDB snapshot, no network ever. With { remote: true } it first round-trips through the table's pull.query handler: the serializable filter travels to your backend, the returned records are applied with full pull semantics (upserted as synced, queued records rebased, tombstones honoured), and only then the query runs locally — so the answer includes both the fresh server rows and anything queued locally.

// Table config: the server-side counterpart of query().
pull: {
  fetch: ...,
  query: async (filter, { params, signal }) =>
    (await api.searchIssues(filter, { signal })).items,
},

// Local by default:
await issues.query({ status: { eq: "OPEN" } });

// Fetch → persist → query locally:
await issues.query({ status: { eq: "OPEN" } }, { remote: true });

// Best-effort fallback when the request may fail:
const rows = await issues
  .query(filter, { remote: true })
  .catch(() => issues.query(filter));

Several fields in one object form an implicit AND. The operators are checked against the field's type at compile time — { severity: { beginsWith: "x" } } does not build. At runtime an invalid condition throws a loud SchemaError; the AppSync-style silent pass-through (a scalar where an operator object belongs quietly matching everything) is gone. compileFilter / matchesFilter are exported, so a pull.fetch can reuse the exact local semantics server-side.

Guides

Subscriptions#

// Every change of one table…
const off = issues.subscribe(({ type, key, item, origin }) => {
  // type:   "created" | "updated" | "deleted"   (item is null for deleted)
  // origin: "local" | "pull" | "push" | "merge" | "evict"
});

// …or only the slice you render.
issues.subscribe({ eventId: { eq: currentEvent } }, rerenderList);

// Store-wide events:
store.on("change", handler);              // all tables
store.on("syncStart", handler);           // { table, scopeKey }
store.on("syncEnd", handler);             // { table, scopeKey, pulled }
store.on("syncError", handler);           // { table, scopeKey, error }
store.on("pushSuccess", handler);         // { table, key, op }
store.on("pushError", handler);           // { table, key, op, error, willRetry }
store.on("conflict", handler);            // { table, key, resolution }
store.on("paramsChanged", handler);       // { params, resyncedTables }
store.on("outboxDrained", handler);       // {}

off(); // every subscribe()/on() call returns its unsubscribe function

// A component subscribes on mount…
const offList = issues.subscribe({ eventId: { eq: id } }, rerenderList);
const offBadge = store.on("outboxDrained", hideSavingBadge);
const offErrors = store.on("pushError", ({ table, key, willRetry }) => {
  if (!willRetry) showFailedToast(table, key);
});

// …and detaches on unmount by calling what it got back.
onUnmount(() => {
  offList();
  offBadge();
  offErrors();
});

Events fire only after their transaction commits — a listener never observes state that later rolled back. The origin field tells you why a change happened: a common pattern is ignoring "push" echoes (your UI already shows that data) while reacting to "pull" and "merge" — both the pull loop and applyRemote() emit "pull"-origin events. A listener that throws is reported to onError and never breaks the other listeners. Deleted events carry item: null, so filtered subscriptions always deliver them — a gone record cannot be re-checked against the filter.

For await-shaped code there are promise forms: store.whenStarted(), store.whenSynced(), store.whenUploaded(), and per table — table.whenSynced() / table.whenUploaded(). whenUploaded() resolves when nothing is queued or in flight; records parked in the error state do not block it — they are surfaced through status.failed and resolveFailed.

Guides

Transactions#

store.transaction() runs your callback against one IndexedDB transaction spanning the listed tables and the outbox: every write inside — data and queued pushes alike — commits or rolls back as a single unit. Events fire only after the commit; a thrown error rolls everything back, including the queue.

await store.transaction(["Event", "Issue", "Image"], async (tx) => {
  const event = await tx.table("Event").get(eventId);
  if (!event) throw new Error("event vanished");   // → full rollback

  await tx.table("Event").update(eventId, { status: "COMPLETED" });
  for (const issue of resolved) await tx.table("Issue").delete(issue.id);
  await tx.table("Image").put(finalImage);
});

Table locks are acquired in sorted order, so two overlapping transactions can never deadlock — they serialize. The transaction view supports get, getAll, put, update, delete — reads inside see the transaction's own writes.

Guides

Push handlers#

A table declares up to three plain async functions. Whatever create / update resolve with — the server echo, typically carrying a fresh version — replaces the local data without marking the record dirty; resolve undefined to keep the local data as-is.

push: {
  create: (item, { attempt, params, signal }) => api.create(item, { signal }),
  update: (item, ctx) => api.update(item),
  delete: async (item, ctx) => { await api.remove(item.id, item._version); },
  retry: {
    maxRetries: Infinity,   // transport errors only — see below
    retryBaseDelay: 1000,   // 1s, 2s, 4s, … doubling
    maxDelay: 30_000,       // backoff ceiling
  },
},

Inside a handler, exactly three kinds of outcome exist:

A record whose retries are exhausted lands in the error state with the failure recorded in its meta — visible via table.failed() and status.failed, revivable via retryFailed(), resolveFailed() or simply saving newer data over it — errors surface through events and onError, never through global state.

If a table has no push.create (a read-only reference table, say), a non-local put() throws SchemaError up front — telling you to either add the handler or write with { local: true }. Nothing silently sits in a queue that can never drain.

Guides

Pull handlers#

pull: {
  // Called in a loop while done !== true; checkpoint persists per scope.
  fetch: async ({ params, scopeKey, checkpoint, isFirstPage, signal }) => ({
    items: page.items,
    checkpoint: page.cursor,   // handed to the next call / next delta sync
    done: !page.nextToken,     // default: true
  }),

  // Point lookup for the conflict path and resolveFailed(); null = gone.
  fetchOne: (key, { params, signal }) => api.byId(key),

  // Server-side counterpart of query() — enables query(f, { remote: true }).
  query: (filter, { params, signal }) => api.search(filter, { signal }),

  mode: "auto",         // "auto" (default): start(), scope change, reconnect
  once: false,          // true = sync each scope a single time (static data)
  deletedKey: "_deleted", // server tombstone field → local removal
  mergePending: true,     // default: rebase queued records onto fresh remote data
  retry: { maxRetries: 3, retryBaseDelay: 1000, maxDelay: 30_000 },
},

Applying a pulled item follows one rule table:

ignoreFields (store-level plus table-level) lists server-managed fields — updatedAt, owner, a version counter — that should not count as "the record changed". A re-pull where only those fields moved emits no change events and rewrites nothing.

When data arrives outside the pull loop — a WebSocket snapshot, an SSR payload, a shared fetch layer — apply it imperatively with the same semantics:

socket.on("issues", (items) => {
  void store.table("Issue").applyRemote(items, {
    deletedKey: "_deleted",   // optional; defaults to pull.deletedKey
    // mergePending: false,   // optional; defaults to pull.mergePending
  });
});

applyRemote() follows the rule table above exactly — items land as synced, queued records are rebased, tombstoned and error records stay untouched — but it does not mark the table as synced and does not move checkpoints: it is data application, not a sync run. A table does not even need a pull config to use it.

Failed pulls emit syncError, keep the table unsynced, and re-run on the next trigger — an explicit table.sync(), a scope change, or the automatic reconnect sweep. An explicit sync() call also rejects with the failure, so imperative flows can react.

Guides

Conflicts#

A conflict exists when the server rejects a mutation because someone changed the record first. The transport signals it by throwing ConflictError — with the current remote state attached when the response already carries it, without it otherwise (the store then calls pull.fetchOne). From there the path is automatic:

1. merge(local, remote, { params }) — the table's merge function, called with copies. The default merge keeps local data field-by-field on top of the remote record, except the ignoreFields, which always come from the remote side — list your version field there and the retried push passes optimistic concurrency for free. (The same function also powers the pull-time rebase of queued records, so write it order-independent.)
2. The merged record is persisted (a "merge"-origin change event) and pushed once more in the same run — as an update even if the original op was a create that turned out to already exist.
3. A second conflict in a row stops the loop: the record moves to error, a conflict { resolution: "surfaced" } event fires, and the record waits for a human (or your code):

const stuck = await issues.failed();
// [{ key, op, item, error }]

await issues.resolveFailed(stuck[0].key, (local, remote) => {
  if (!remote) return "discard-local";      // gone on the server — drop ours
  if (remote.status === "RESOLVED") return "keep-local";  // re-push ours as-is
  return { ...remote, note: local.note };   // or hand-merge and push that
});

Special cases the store handles for you:

• Delete vs. delete — the remote is already gone: counted as success, the tombstone is removed.
• Delete vs. update — the merge refreshes server-managed fields and the delete retries once with them.
• Update vs. delete — the record vanished server-side: re-created through push.create when the table has one, surfaced otherwise.
• Create vs. exists — escalated to update with the merged data.

Guides

Changing scope#

// The user navigated to another event:
await store.setParams({ eventId: route.eventId });
await store.whenSynced();

store.on("paramsChanged", ({ params, resyncedTables }) => {
  // resyncedTables: the tables whose scope key changed
});

For every table whose scope.key changed, in order:

1. An in-flight pull of the old scope is aborted.
2. Synced records failing scope.keep(item, newParams) are evicted in one transaction. Records with unpushed changes (pending, pushing, error) are never evicted — a scope change cannot lose data on its way to the server.
3. If the eviction removed anything, the old scope's checkpoint and completion mark are dropped too — a delta pull cannot refill holes, so returning to that scope refetches it. Untouched scopes keep their marks: navigating back is instant.
4. The new scope pulls, resuming from its own checkpoint if it synced before.

Guides

Migrations#

The configured schemaVersion drives IndexedDB upgrades. Structure is reconciled declaratively: on a version bump the store diffs the configured tables and indexes against what exists on disk and creates or drops object stores and indexes to match — adding a table or an index is just "declare it, bump the version". A table added without a bump fails fast at start() with a SchemaError telling you exactly that.

Data migrations transform records, keyed by the version they migrate to, running inside the upgrade transaction in ascending order over the crossed versions:

createDataStore({
  name: "InspectorDataStore",
  schemaVersion: 3,          // v2 added byState; v3 renames a status
  migrations: {
    3: async ({ table }) => {
      await table("Issue").updateEach((issue) =>
        issue.status === "OPENED" ? { ...issue, status: "OPEN" } : issue,
      );
    },
  },
  tables: { /* … */ },
});

updateEach(fn) replaces each record's data with the returned object, deletes on null, keeps on the same object; getAll(), put(), delete() cover the rest. A migration that throws aborts the whole upgrade — the database stays at the old version, nothing half-commits. Do not await anything external inside a migration: the upgrade transaction auto-commits.

Guides

Local-only tables#

Draft: defineTable<DraftNote>({ primaryKey: "id", local: true }),

A local: true table is a typed, transactional, indexed IndexedDB table with no sync machinery at all: records live in the local state, deletions are physical, synced is always true, and combining local with push / pull / scope is a configuration error. Perfect for drafts, device caches and UI state that participates in transactions with synced tables.

Guides

Binary fields#

Some WebKit builds — old iOS Safari most famously — throw DataCloneError the moment a Blob is stored in IndexedDB. Declare where your blobs live and the store works around it the way localForage does: on affected browsers the blobs are transparently stored as ArrayBuffers and turned back into real Blobs on every read. Your code never sees the difference.

const store = createDataStore({
  binary: { mode: "auto" },   // default; "always" / "never" to force
  tables: {
    Photo: defineTable<Photo>({
      primaryKey: "id",
      // Typed dot-paths: autocompleted, must point at Blob values.
      blobPaths: ["preview", "attachment.file", "frames"], // frames: Blob[]
      push: { /* handlers receive real Blobs */ },
    }),
  },
});

mode: "auto" runs a feature probe at start(): a tiny Blob is written to a service store and read back. Both failure modes — the loud DataCloneError and environments that "store" the blob but hand back garbage — turn encoding on. The probe costs about a millisecond, and reads tolerate both stored forms regardless of the mode, so a browser update that fixes the bug needs no migration.

The contract, everywhere:

• get / getAll / query / events / pending() / failed() — always return real Blobs;
• push handlers receive real Blobs and may return them in the echo;
• merge sees real Blobs but must treat blob fields as opaque: pick the local or the remote value whole — fabricating new binary data inside merge throws a SchemaError;
• blob fields are compared by (MIME type, size) for change detection and cannot be filtered on;
• writes with blobs work through put / update / applyRemote / pulled pages — encoding always happens before the transaction opens.

const encoded = await photos.encodeBlobs({ id, name, preview: file });
await store.transaction(["Photo", "Event"], async (tx) => {
  await tx.table("Photo").put(encoded);          // ok — already encoded
  await tx.table("Event").update(eventId, { photoCount: n + 1 });
});

Guides

Testing#

The store runs unmodified on fake-indexeddb. Two pieces make tests deterministic: reset the IndexedDB universe between tests, and alias @dmytromykhailiuk/network-connection to a controllable double — through the bundler alias, so retry-request sees the same state.

// vitest.config.ts
export default defineConfig({
  resolve: { alias: { "@dmytromykhailiuk/network-connection": "./tests/network-mock.ts" } },
  test: {
    environment: "jsdom",
    setupFiles: ["tests/setup.ts"],
    server: { deps: { inline: ["@dmytromykhailiuk/retry-request"] } },
  },
});

// tests/setup.ts
import "fake-indexeddb/auto";
import { IDBFactory } from "fake-indexeddb";
beforeEach(() => {
  indexedDB = new IDBFactory();
});

This package's own suite — 160+ tests across CRUD, transactions, the push pipeline, conflicts, pulls, scope changes, outbox persistence across "reloads" and concurrency stress — is written exactly this way and doubles as a cookbook.

Reference

Required vs. optional#

Formally almost everything is optional: requirements are enforced at the moment of the operation that needs them, with a SchemaError naming exactly what to add — never a mutation silently stuck in a queue nothing can drain. The one compile-time exception: a declared pull block must contain fetch.

Store config#

Table config#

push.*#

pull.*#

Typical configurations#

// Local-only: push / pull / scope are FORBIDDEN (SchemaError at creation).
Draft: defineTable<Draft>({ primaryKey: "id", local: true }),

// Read-only reference data:
Site: defineTable<Site>({ primaryKey: "id",
  pull: { fetch, once: true } }),

// Write-only, immutable after creation:
Vehicle: defineTable<Vehicle>({ primaryKey: "id",
  push: { create } }),          // update()/delete() throw — by design

// Full CRUD sync — the recommended minimum with server versioning:
Issue: defineTable<Issue, P>({ primaryKey: "id",
  push: { create, update, delete },
  pull: { fetch, fetchOne },    // fetchOne: conflicts resolve instead of surfacing
  merge,                        // optional; default is LWW + ignoreFields
  scope,                        // optional; without it — one unnamed scope
}),

Reference

The store#

Reference

TableStore#

Reference

Errors#

Every failure is a DataStoreError with a stable code — branch on instanceof or on the code, never on message text.

Reference

Design FAQ#

Why are unique indexes not supported?

A unique index turns every put into a constraint that can abort a sync transaction halfway through a pulled page. Uniqueness belongs to the primary key; everything else is a query.

Why does a pull rebase pending records instead of overwriting them?

Overwriting would silently drop the user's unpushed edit; skipping (available via mergePending: false) leaves the queued push built on a stale base — on a last-write-wins backend it would then blindly erase other clients' changes. The rebase runs the same merge the conflict path uses: local fields win, server-managed fields (ignoreFields) come from the remote, and the record stays queued. Tombstones and error-state records are never rebased — the first is decided by the delete push, the second by a human.

Can I run several stores?

Yes — every createDataStore() is fully isolated: its own database, locks, queue and events. Use different names.

What happens on a second tab?

IndexedDB itself is shared and every write is transactional, so data never corrupts. But each tab runs its own outbox dispatcher — for now run the store in one tab (or a SharedWorker / leader-election layer on top). Cross-tab leadership is on the roadmap.

Reference

Exports#

createDataStore · defineTable · compileFilter · matchesFilter · validateFilter · toPredicate · DataStoreError · ConflictError · FatalPushError · RecordNotFoundError · TransactionInactiveError · StoreNotStartedError · SchemaError

Types: DataStore · DataStoreConfig · DataStoreEvents · DataStoreStatus · TableConfig · TableDefinition · TableStore · StoreTransaction · TransactionTable · PushConfig · PushContext · PullConfig · PullContext · PullResult · ScopeConfig · Filter · FieldCondition · ChangeEvent · RecordMeta · RecordState · PendingOp · MigrationContext · QueryOptions · FailedEntry · PendingEntry and more.