Writing a Custom Adapter

Only needed on platforms where the official adapters don't fit - Supabase/PostgREST clients, Convex, Firebase, DynamoDB, proprietary AI-builder clouds.

Decide if you even need one

• Demos / tests / a single long-lived process: @chatpack/adapter-memory.
• Any Postgres you can reach with a connection string (Neon, RDS, Railway, Fly, Replit, Vercel Postgres): @chatpack/adapter-drizzle. Do not write a custom adapter for these.
• Postgres behind an HTTP client only (e.g. Supabase JS), or a non-SQL store (Convex, Firestore, DynamoDB): write a custom adapter - continue below.

The contract

Implement the StorageAdapter interface exported from @chatpack/core and pass it as chatpack({ storage: yourAdapter() }). Ten async methods:

Exact TypeScript signatures and per-method TSDoc ship in the package's .d.ts (@chatpack/core, storage.ts module).

The entities you store are described in Conversations & Messages. Users are opaque string ids - do not add foreign keys into your users table and do not validate user existence in the adapter.

The invariants

These are database-agnostic - satisfy every one of them however your platform does atomicity. Everything else is implementation detail.

1. One conversation per pairKey, even under concurrency. Core computes pairKey (sorted user ids joined with ":") and hands it to you. Two concurrent getOrCreateDirectConversation calls with the same pairKey must converge on a single conversation (created: true for at most one caller). A JS-side "select, then insert if missing" is NOT enough - enforce uniqueness in the database (unique index + insert-on-conflict, transaction, or your platform's equivalent).
2. seq is strictly increasing per conversation and assigned atomically with the message insert. Never reused, never decreasing; gaps are allowed. Concurrent sends to the same conversation must serialize. Do NOT compute MAX(seq) + 1 in application code - that races.
3. Adapters generate ids. getOrCreateDirectConversation and addMessage create the id themselves (any unique string; the official adapters use prefixed random ids like msg_<uuid>). Ids must be unique and stable.
4. Adapters never enforce permissions. Core validates participants and permission hooks before calling you. Return whatever is asked for. Corollary for hosted platforms: the adapter must run server-side with full DB privileges (e.g. the Supabase service-role key), and the Chatpack tables must NOT be readable or writable by browser/anon clients - otherwise users can read each other's messages around Chatpack entirely.
5. Cursors are opaque strings that the adapter defines. Core (and the HTTP layer) round-trip your nextCursor back into input.cursor verbatim, without inspecting it. Pick any encoding that survives a URL query parameter. Return nextCursor: null when there are no more results. Reference encodings: listMessages - the seq of the last message on the page; listConversations - "<lastActivityMs>:<conversationId>" (keyset) or simply the last conversation id.
6. Ordering. listMessages: newest-first (descending seq). listMessagesAfterSeq: oldest-first (ascending seq), only seq > afterSeq, capped at limit. listConversations: most-recently-active first (latest message activity, falling back to conversation creation time), with a stable tiebreak so pagination never skips or repeats.
7. Return real Date instances, never ISO strings. The interface types say Date and core does not coerce. Database drivers and HTTP clients (PostgREST, Convex, Firestore) often hand back strings or numbers - convert at the adapter boundary, both directions. JSON serialization is the HTTP handler's job, never yours.
8. Soft delete is an update, not a removal. Core calls updateMessage({ body: "", deletedAt }); the row must remain (clients render a tombstone) and must keep its seq. Never hard-delete. updateMessage patches only the fields that are defined on the input and throws on unknown ids; updateLastRead overwrites lastReadMessageId unconditionally (core has already validated the message).
9. metadata is an arbitrary JSON object. Store and return it losslessly (default {}).
10. countUnread is exact and batched. For each requested conversation, count messages with seq strictly greater than the seq of the viewer's lastReadMessageId (null read-state = 0, i.e. everything) AND senderId !== userId (a viewer's own messages are never unread). Soft-deleted messages keep their seq and DO count; all roles count. Return { [conversationId]: count }; ids the viewer doesn't participate in may be omitted or 0 (core treats missing as 0). One call covers a whole page of conversations - do it in one query, not one per id. Note lastReadMessageId is a message id, not a seq: resolve it at query time (SQL: LEFT JOIN the message row + COALESCE(read.seq, 0)).

Reference schema (Postgres)

If your store speaks SQL, use the official schema as-is; if not, translate the shapes and satisfy the invariants instead. Also available programmatically: import { migrationSql, chatpackSchema } from "@chatpack/adapter-drizzle" (migrationSql is plain idempotent DDL - no Drizzle required to execute it).

CREATE TABLE IF NOT EXISTS "chatpack_conversations" (
  "id" text PRIMARY KEY,
  "pair_key" text NOT NULL,
  "created_at" timestamptz NOT NULL,
  "metadata" jsonb NOT NULL DEFAULT '{}',
  "last_seq" integer NOT NULL DEFAULT 0,
  "last_activity_at" timestamptz NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS "chatpack_conversations_pair_key_idx"
  ON "chatpack_conversations" ("pair_key");
CREATE INDEX IF NOT EXISTS "chatpack_conversations_activity_idx"
  ON "chatpack_conversations" ("last_activity_at", "id");

CREATE TABLE IF NOT EXISTS "chatpack_conversation_participants" (
  "conversation_id" text NOT NULL REFERENCES "chatpack_conversations"("id") ON DELETE CASCADE,
  "user_id" text NOT NULL,
  "joined_at" timestamptz NOT NULL,
  "last_read_message_id" text
);
CREATE UNIQUE INDEX IF NOT EXISTS "chatpack_participants_conv_user_idx"
  ON "chatpack_conversation_participants" ("conversation_id", "user_id");
CREATE INDEX IF NOT EXISTS "chatpack_participants_user_idx"
  ON "chatpack_conversation_participants" ("user_id");

CREATE TABLE IF NOT EXISTS "chatpack_messages" (
  "id" text PRIMARY KEY,
  "conversation_id" text NOT NULL REFERENCES "chatpack_conversations"("id") ON DELETE CASCADE,
  "sender_id" text NOT NULL,
  "body" text NOT NULL,
  "role" text NOT NULL DEFAULT 'user',
  "seq" bigint NOT NULL,
  "created_at" timestamptz NOT NULL,
  "edited_at" timestamptz,
  "deleted_at" timestamptz,
  "metadata" jsonb NOT NULL DEFAULT '{}'
);
CREATE UNIQUE INDEX IF NOT EXISTS "chatpack_messages_conv_seq_idx"
  ON "chatpack_messages" ("conversation_id", "seq");

How the reference implementation satisfies the two hard invariants:

• Atomic seq - one statement, row-locked by Postgres:

  UPDATE chatpack_conversations
  SET last_seq = last_seq + 1, last_activity_at = $now
  WHERE id = $conversationId
  RETURNING last_seq;

  then insert the message with the returned value. The unique index on (conversation_id, seq) backstops the invariant. On Supabase/PostgREST you cannot express this UPDATE through the client - wrap it in a SQL function and call it via RPC.

• Idempotent pair creation - INSERT ... ON CONFLICT (pair_key) DO NOTHING, then re-select by pair_key; zero inserted rows means a concurrent call won and both converge on the same row.

• Batched unread counts - one GROUP BY query per page; a LEFT JOIN resolves last_read_message_id to its seq (invariant 10). The unique index on (conversation_id, seq) makes each count an index range scan:

  SELECT m.conversation_id, count(*)
  FROM chatpack_messages m
  JOIN chatpack_conversation_participants p
    ON p.conversation_id = m.conversation_id AND p.user_id = $userId
  LEFT JOIN chatpack_messages r ON r.id = p.last_read_message_id
  WHERE m.conversation_id = ANY($conversationIds)
    AND m.sender_id <> $userId
    AND m.seq > COALESCE(r.seq, 0)
  GROUP BY m.conversation_id;

  Conversations absent from the result have zero unread.

Non-SQL platforms (Convex, Firestore, ...)

Ignore the SQL; keep the three-collection shape and satisfy the invariants with your platform's transaction primitive. Example - on Convex, each adapter method becomes a Convex query/mutation (DB access only exists inside Convex functions) and the StorageAdapter object calls them through the Convex client; addMessage is a single mutation (read counter → increment → insert), which is automatically a serializable transaction, satisfying invariant 2 with no extra machinery. Enforce invariant 1 with an index/lookup on pairKey inside one mutation.

Skeleton

import type { StorageAdapter } from "@chatpack/core";

export function myAdapter(client: MyDbClient): StorageAdapter {
  return {
    async getOrCreateDirectConversation({ pairKey, userIds, metadata }) {
      // 1. Attempt atomic create (unique on pairKey; generate the id here).
      // 2. On conflict, fetch the existing conversation by pairKey.
      // return { conversation, created };
    },
    async getConversation(conversationId) {
      // Fetch conversation + BOTH participant rows, or return null.
    },
    async listConversations({ userId, limit, cursor }) {
      // Conversations where userId is a participant, most-recently-active
      // first; keyset-paginate; you define the cursor encoding.
      // return { conversations, nextCursor }; // nextCursor: string | null
    },
    async addMessage({ conversationId, senderId, body, role, metadata }) {
      // Atomically bump the conversation's seq counter and insert the
      // message with that seq (single transaction / RPC / mutation).
      // return message; // with adapter-generated id and real Date fields
    },
    async getMessage(messageId) {
      // Fetch by id, or return null.
    },
    async listMessages({ conversationId, limit, cursor }) {
      // Descending seq; cursor = e.g. String(lastMessage.seq).
      // return { messages, nextCursor };
    },
    async listMessagesAfterSeq({ conversationId, afterSeq, limit }) {
      // Ascending seq, seq > afterSeq, up to limit. Powers gap-fill.
      // return messages;
    },
    async updateMessage({ messageId, body, editedAt, deletedAt }) {
      // Patch ONLY the defined fields; throw if messageId is unknown.
      // return updatedMessage;
    },
    async updateLastRead({ conversationId, userId, messageId }) {
      // Set that participant's lastReadMessageId.
    },
    async countUnread({ userId, conversationIds }) {
      // For each conversation: count messages with seq > the seq of this
      // user's lastReadMessageId (null -> 0) AND senderId !== userId.
      // Tombstones count. One batched query, not one per id (invariant 10).
      // return { [conversationId]: count };
    },
  };
}

Pitfalls

Each of these has bitten a real integration:

• Checking pair existence in JS before inserting (races; use DB uniqueness).
• MAX(seq) + 1 computed in application code (races; use an atomic counter).
• Returning ISO strings where the contract says Date.
• Hard-deleting messages, or dropping seq/row on delete (must tombstone).
• listMessages returned oldest-first (must be newest-first), or listMessagesAfterSeq returned newest-first (must be oldest-first).
• Stubbing nextCursor as always-null - pagination silently breaks once history exceeds one page.
• Exposing chatpack_* tables to browser/anon database clients (bypasses Chatpack's permission layer; adapter must be server-only with privileged credentials).
• Adding a foreign key to the app's users table (Chatpack never owns users).
• Enforcing "is sender a participant?" in the adapter (core already did).
• Counting the viewer's own messages as unread in countUnread (they never are), or issuing one count query per conversation (N+1; batch the page).

Verify your adapter

Run these checks before calling it done. All of them go through the normal Chatpack API (chat.api.* or the HTTP routes) - if any fails, the bug is in the adapter.

1. Round trip. Create a conversation (A→B), send a message, restart the process (or hit a fresh serverless isolate), list messages - the message is still there with the same id, seq: 1, and createdAt as a valid timestamp.
2. Pair idempotency, both directions. getOrCreateConversation(A→B) then (B→A) - same conversation id both times, and only one row/document exists.
3. Pair idempotency, concurrent. Fire 5+ parallel getOrCreateConversation calls for the same new pair - exactly one conversation exists afterwards.
4. Seq under concurrent sends. Fire 10 parallel sendMessage calls into one conversation - afterwards all 10 seq values are distinct and strictly increasing; no duplicates, no failures.
5. Pagination walk. Send 25 messages, list with limit: 10, follow nextCursor until null - you get exactly 25 unique messages, newest-first, no repeats, no gaps. Same walk for listConversations.
6. Gap-fill. After sending messages 1..N, listMessagesAfterSeq with afterSeq: 3 returns 4..N oldest-first.
7. Edit + soft delete. Edit a message (body changes, editedAt set); delete it (body "", deletedAt set, still present in listMessages with its original seq); editing the deleted message now fails with MESSAGE_DELETED.
8. Read-state. markRead then re-fetch the conversation - that participant's lastReadMessageId is updated and survives a restart. Bonus: markRead with an OLDER message afterwards is a no-op (core enforces monotonicity; your adapter just stores what it's given).
9. Unread counts. B sends 3 messages; getConversation as A shows unreadCount: 3, as B shows 0 (own messages never count). A marks the 2nd read - A now sees 1. Delete the 3rd as B - A still sees 1 (tombstones count).
10. Types at the boundary. createdAt instanceof Date is true on returned conversations and messages (catches ISO-string leakage).
11. Privilege check (hosted DBs). With the browser/anon credentials, a direct read of chatpack_messages returns nothing/denied.

Real-time plugins and custom adapters

typing(), presence(), receipts() publish ephemeral events on the SSE stream - never stored, never replayed - so custom storage adapters need zero changes to support them.