Persistence Reference

shape-only — SQLite and PostgreSQL regression-tested; no live Meta validation · reviewed 2026-09-07

import {
  CURRENT_SCHEMA_VERSION,
  PersistenceError,
  runOutboxWorkerOnce,
  startOutboxWorker,
  type MigrationReport,
  type OutboxItem,
  type PersistenceHealth,
  type PersistenceStore
} from "@wats/persistence";
import { createSqlitePersistence } from "@wats/persistence/sqlite";
import { createPostgresPersistence } from "@wats/persistence/postgres";

The SQLite adapter is for local and single-instance testing. The Postgres adapter is optional and lives behind @wats/persistence/postgres; install pg yourself when you use it. SQLite users do not need Postgres installed.

SQLite quickstart

const store = await createSqlitePersistence({ filename: "./wats.sqlite" });
await store.migrate();
const health = await store.health();
await store.close();

Use a local path or :memory:. Database paths are operationally sensitive; diagnostics report only [REDACTED_SQLITE_DATABASE].

Postgres adapter

import { createPostgresPersistence } from "@wats/persistence/postgres";

const store = await createPostgresPersistence({
  connectionString: process.env.DATABASE_URL ?? ""
});
await store.migrate();
await store.close();

pg is an optional peer dependency. The subpath is importable without pg; the driver is loaded only when createPostgresPersistence(...) runs. Postgres diagnostics report [REDACTED_POSTGRES_DATABASE] and never echo the connection string. The adapter implements the same PersistenceStore contract and schema version as SQLite, including the message projection added in schema v3.

Schema and migrations

The migration runner creates:

• wats_schema_migrations
• wats_persistence_lock
• wats_webhook_events
• wats_service_requests
• wats_outbox
• wats_messages
• wats_message_status_events

Migrations are forward-only for alpha. Already-applied migration checksums must match the package migration definitions; checksum drift fails closed with PersistenceError. 001_initial keeps its originally shipped checksum; 002_outbox_lease_id upgrades existing v1 databases with the leaseId outbox column; 003_message_projection adds the message projection tables and indexes; 004_inbound_window_index adds the composite index backing getLatestInboundMessageAt; 005_service_request_claims adds the durable claim/completion columns to wats_service_requests; 006_message_uniqueness deduplicates preexisting duplicate (direction, wa_message_id) message rows (keeping the best known state) and adds the unique indexes backing message and status-event dedup. A held migration lock fails closed with PersistenceError code migration_lock_failed.

Backend faults are always typed as PersistenceError with a code from PersistenceErrorCode — a raw bun:sqlite or pg error never escapes the public API. claim_failed and completion_failed cover service-request claim/completion lease faults; outbox_failed covers outbox and message-projection transaction faults (including recordMessage/appendMessageStatus backend failures); store_closed covers operations on a closed store; invalid_record/invalid_options/invalid_filename cover input-validation failures (never backend faults); migration_failed/migration_checksum_mismatch/migration_lock_failed cover migration faults. The validation codes are distinct from the persistence-operation codes so a caller can tell a bad input from a backend fault.

Runtime contract

PersistenceStore exposes:

• backend
• migrate(): Promise<MigrationReport>
• health(): Promise<PersistenceHealth>
• recordWebhookEvent(...)
• getServiceRequest(...)
• recordServiceRequest(...)
• enqueueOutboxItem(...)
• claimOutboxItems(...)
• markOutboxItemFailed(...)
• markOutboxItemSucceeded(...)
• recordMessage(...)
• appendMessageStatus(...)
• getMessage(...)
• listMessages(...)
• getLatestInboundMessageAt(...)
• countOutboxPending()
• close(): Promise<void>

Webhook event records store safe event keys and hashes, not raw webhook bodies. Duplicate event keys return "duplicate", so the service can acknowledge Meta retries without dispatching the same update twice.

Service request idempotency stores an idempotency key, request hash, and response JSON. A matching key/body hash replays the stored response; the same key with a different body hash conflicts.

The built-in SQLite and Postgres adapters additionally implement two optional PersistenceStore methods — claimServiceRequest({ idempotencyKey, requestHash, createdAt }) and completeServiceRequest({ idempotencyKey, requestHash, responseJson, createdAt }) — for durable atomic service request claims. claimServiceRequest returns "claimed" (new reservation), "pending" (already claimed, not yet completed), "conflict" (same key, different hash), or { responseJson } (already completed, replay). A claim reserves the request identity before the caller sends to Graph and persists across restart indefinitely (no timeout-based blind resend). completeServiceRequest stores the response only when it matches the reserved hash; the older recordServiceRequest never overwrites an existing reservation.

Message projection

The wats_messages table records outbound send attempts and authenticated inbound message metadata. Each row is keyed by a caller-generated local row_id, carries the WhatsApp message id (wa_message_id), direction, optional phone fields, Graph type, current status, and created/updated timestamps. wats_message_status_events is an append-only log of status transitions per wa_message_id.

await store.recordMessage({
  rowId: "wats-msg-...",
  waMessageId: "wamid.HBgM...",
  direction: "outbound",
  toPhone: "15551230000",
  type: "text",
  status: "sent",
  graphMessageId: "wamid.HBgM...",
  createdAt: new Date().toISOString(),
  updatedAt: new Date().toISOString()
});

await store.appendMessageStatus({ waMessageId: "wamid.HBgM...", status: "delivered", timestamp: new Date().toISOString() });

const record = await store.getMessage({ waMessageId: "wamid.HBgM..." });
const page = await store.listMessages({ limit: 50 });

recordMessage is idempotent on row_id (INSERT OR IGNORE): re-recording the same rowId is a no-op. It also deduplicates by (direction, wa_message_id) via the unique index from schema migration 006_message_uniqueness — a second insert with a different rowId but the same (direction, wa_message_id) is a no-op, preserving the first projection. appendMessageStatus inserts an event row and best-effort updates the matching message's status and updated_at (no-op if no matching message). listMessages returns rows newest-first (created_at DESC, row_id DESC), with limit (1..100) and an optional beforeRowId cursor for paging; nextCursor is the last item's rowId when more rows may exist, otherwise null.

Status transitions are rank-based, not timestamp-only. The rank order is sent(1) < delivered(2) < read(3); pending(0) is the lowest and unknown statuses rank 0. failed may replace pending or sent but must never replace delivered or read. An event at the same second (e.g. a Meta callback timestamp truncated to whole seconds landing at an equal-or-earlier millisecond than the outbound send) may advance the rank; updated_at never rolls backward (it advances to the event timestamp only when the event is newer). This closes the timestamp-precision seam between millisecond outbound sends and whole-second Meta status callbacks. If a status event arrives before the message projection (an early webhook), recordMessage reconciles the stored events and advances the freshly-inserted row so it does not regress to the send-time status. Migration 006_message_uniqueness deduplicates preexisting duplicate (direction, wa_message_id) rows by keeping the row with the best known state (highest status rank, then smallest row_id as a deterministic tiebreak), not the smallest row_id unconditionally.

The service records outbound projections automatically after a successful Graph send and exposes them read-only through GET /messages and GET /messages/{messageId} (see the Service Reference). Projection failures are swallowed and never break the send response.

Conversation window

The 24-hour customer-service window decides whether you can send free-form text or must reach for an approved template. Meta's rule: free-form is allowed within 24h of the customer's last inbound message; outside that, only templates.

import {
  getConversationWindowState,
  canSendFreeForm
} from "@wats/persistence";

const state = await getConversationWindowState(store, {
  phone: "15551230000",
  now: new Date().toISOString()
});

if (state.open) {
  // free-form is allowed; state.remainingMs tells you how long you have
} else {
  // send a template — the window closed (or never opened)
}

const ok = await canSendFreeForm(store, {
  phone: "15551230000",
  now: new Date().toISOString()
});

ConversationWindowState is { open, lastInboundAt, expiresAt, remainingMs }. remainingMs is 0 when the window is closed or there is no inbound history. The boundary is strict: an inbound exactly windowMs ago closes the window (send a template). Override the window with windowMs (integer 1..604800000; default 24h) for non-standard rules.

Clock skew: if the customer's last inbound lands after now, the window is treated as open with the full windowMs remaining. A free-form send that Meta rejects costs a retry; wrongly forcing a template silences a live conversation. The skew branch picks the cheaper failure.

getLatestInboundMessageAt({ phone }) powers the helper: it returns the createdAt ISO of the most recent wats_messages row with direction = 'inbound' and from_phone matching, or null. Schema migration 004_inbound_window_index adds the composite index (direction, from_phone, created_at DESC) so the lookup does not table-scan as the projection grows.

The helper sees only locally projected inbound rows. The service projects authenticated inbound messages before handlers run and rejects invalid or future event timestamps rather than fabricating a window. Real PostgreSQL regression tests run in CI against an ephemeral database; this does not claim live Meta validation.

Outbox

At-least-once local work scheduling:

await store.enqueueOutboxItem({
  id: "send-1",
  payloadHash: "sha256:...",
  createdAt: new Date().toISOString()
});

const report = await runOutboxWorkerOnce(store, {
  now: new Date().toISOString(),
  limit: 10,
  retryDelayMs: 30_000,
  async handler(item: OutboxItem) {
    // Reconstruct/send from application-owned state keyed by item.id.
  }
});

The outbox table stores only payload hashes, status, attempt counts, leaseId, and retry timestamps — no raw webhook bodies, message text, Graph request bodies, or contacts. runOutboxWorkerOnce(...) claims due pending items, calls the handler, marks successes succeeded, and reschedules failures with nextAttemptAt = now + retryDelayMs.

Claimed items use a five-minute processing lease; stale processing rows become claimable again, so a killed worker does not strand records forever. The lease is fenced: markOutboxItemFailed(...) and markOutboxItemSucceeded(...) require the current leaseId, so a stale worker cannot complete a newer reclaimed lease.

Outbox worker

startOutboxWorker(store, options) arms a timer that fires runOutboxWorkerOnce on a fixed interval. It returns a handle with stop() (cancels the pending timer and awaits any in-flight tick) and running() (true until stop()). Overlap is skipped: a tick that arrives while a previous tick is still running is a no-op, and the in-flight run re-arms the next timer in its finally block.

import { startOutboxWorker } from "@wats/persistence";

const worker = startOutboxWorker(store, {
  handler: async (item) => {
    // Reconstruct/send from application-owned state keyed by item.id.
  },
  intervalMs: 5_000,
  batchLimit: 10,
  retryDelayMs: 30_000,
  onReport: (report) => {
    // report: { processed, succeeded, failed, pending }
  },
  onError: (error) => {
    // claim/count/store errors swallowed inside a tick land here
  }
});

const pendingDepth = await store.countOutboxPending();
await worker.stop();

onReport fires after each tick with { processed, succeeded, failed, pending }. pending comes from store.countOutboxPending(). onError fires when claim/count/store errors are swallowed inside a tick — the loop never dies on its own errors. now and schedule are injectable for a deterministic virtual clock.

Wire the tick report into the service metrics registry so /metrics exposes outbox depth and throughput:

import { createMetricsRegistry, createOutboxMetricsReporter, createWatsServiceApp } from "@wats/service";

const metrics = createMetricsRegistry();
const app = createWatsServiceApp({ profile, secrets, persistence: store, metricsRegistry: metrics });
const reporter = createOutboxMetricsReporter(metrics, { adapter: store.backend });

const worker = startOutboxWorker(store, {
  handler: async (item) => { /* ... */ },
  ...reporter
});

createOutboxMetricsReporter sets outbox_depth{adapter, state="pending"} from report.pending and increments outbox_processed_total{outcome="success"|"error"} from report.succeeded / report.failed. The depth gauge is registry-only — it is not forwarded to a user-owned TelemetrySink (the sink seam has no setGauge). The reporter sets only the pending state series; processing and succeeded are not emitted.

The worker is caller-owned and single-process. There is no distributed scheduling, multi-process coordination, or automatic send-path enqueueing — the service send routes stay synchronous, and enqueuing into the outbox is the caller's job.

Redaction boundary

Persistence diagnostics must not print: access tokens, app secrets, webhook verify tokens, service bearer tokens, authorization headers, database URLs or SQLite paths, raw webhook bodies, message text, or contact payloads.

Relationship to service and CLI

@wats/service accepts an injected PersistenceStore and does not read database environment variables directly. When persistence is injected:

• signed webhook POSTs are recorded by event key/hash and duplicates are acknowledged without redispatch;
• service send routes honor Idempotency-Key for replay/conflict behavior;
• injected stores must expose the outbox methods as part of the accepted service persistence contract.

The CLI exposes read-only local message navigation through wats messages list and wats messages show. These call the running local service /api/messages endpoint; they do not call Meta and do not require live credentials. Deeper conversation/thread navigation is not built.

Non-goals

Not implemented yet: deeper CLI thread/conversation navigation, delivery/read status UI, raw webhook body storage, automatic service send enqueueing, production hosting guarantees, and live Meta validation.

Related docs

• Config Reference
• Service Reference