SurrealDB
Keep connected accounts, deliveries, and publish state in SurrealDB, and stream delivery updates to your app with live queries.
This guide stores everything the SDK needs to survive a restart in SurrealDB: connected accounts, publication records, the idempotency store, the Threads workflow store, and the webhook inbox. The SDK runs in your Node.js or Bun server and talks to SurrealDB Cloud or a self-hosted SurrealDB 3 instance over a WebSocket. SurrealDB has no job scheduler, so reconciliation runs from whatever scheduler you already use.
Install
npm install @opencoredev/social-sdk surrealdbpnpm add @opencoredev/social-sdk surrealdbyarn add @opencoredev/social-sdk surrealdbbun add @opencoredev/social-sdk surrealdbnub add @opencoredev/social-sdk surrealdbaube add @opencoredev/social-sdk surrealdbThe examples use version 2 of the surrealdb JavaScript SDK.
Connect
Create the namespace, the database, and a user for the server once, as the root user. EDITOR can read and write records and define tables in this database, but can’t manage users:
DEFINE NAMESPACE app;
USE NS app;
DEFINE DATABASE social;
USE DB social;
DEFINE USER social ON DATABASE PASSWORD "change-me" ROLES EDITOR;import { Surreal } from "surrealdb";
export const db = new Surreal();
await db.connect(process.env.SURREAL_URL!, {
namespace: "app",
database: "social",
// Passed here rather than through signin() so the SDK signs in again after a reconnect.
authentication: {
namespace: "app",
database: "social",
username: process.env.SURREAL_USER!,
password: process.env.SURREAL_PASS!,
},
});SURREAL_URL is a wss:// address for SurrealDB Cloud, or ws://127.0.0.1:8000 for a local server. Table PERMISSIONS only apply to record users, not to a database user like social, so every query must filter by the tenant you verified.
Schema
Define the tables in schema.surql:
DEFINE TABLE connected_account SCHEMAFULL;
DEFINE FIELD tenant ON connected_account TYPE string;
DEFINE FIELD platform ON connected_account TYPE string;
DEFINE FIELD account_id ON connected_account TYPE string;
DEFINE FIELD backend ON connected_account TYPE string DEFAULT "default";
DEFINE FIELD created_at ON connected_account TYPE datetime DEFAULT time::now();
DEFINE INDEX account_unique ON connected_account FIELDS tenant, backend, platform, account_id UNIQUE;
DEFINE TABLE delivery SCHEMAFULL;
DEFINE FIELD tenant ON delivery TYPE string;
DEFINE FIELD idempotency_key ON delivery TYPE string;
DEFINE FIELD account ON delivery TYPE record<connected_account>;
DEFINE FIELD state ON delivery TYPE string;
DEFINE FIELD delivery_ref ON delivery TYPE option<object> FLEXIBLE;
DEFINE FIELD updated_at ON delivery TYPE datetime VALUE time::now();
DEFINE INDEX delivery_state ON delivery FIELDS state;
DEFINE TABLE idempotency_claim SCHEMAFULL;
DEFINE FIELD scope ON idempotency_claim TYPE string;
DEFINE FIELD key ON idempotency_claim TYPE string;
DEFINE FIELD fingerprint ON idempotency_claim TYPE string;
DEFINE FIELD target_keys ON idempotency_claim TYPE array<string>;
DEFINE FIELD outcomes ON idempotency_claim TYPE object FLEXIBLE DEFAULT {};
DEFINE INDEX claim_unique ON idempotency_claim FIELDS scope, key UNIQUE;
DEFINE TABLE threads_workflow SCHEMAFULL;
DEFINE FIELD workflow ON threads_workflow TYPE object FLEXIBLE;
DEFINE FIELD claimed_until ON threads_workflow TYPE option<datetime>;
DEFINE FIELD claimed_by ON threads_workflow TYPE option<string>;
DEFINE TABLE social_event SCHEMAFULL;
DEFINE FIELD platform ON social_event TYPE string;
DEFINE FIELD payload ON social_event TYPE object FLEXIBLE;
DEFINE FIELD processed_at ON social_event TYPE option<datetime>;Apply it with the surreal CLI:
surreal sql --endpoint "$SURREAL_URL" --user root --pass "$SURREAL_ROOT_PASS" \
--namespace app --database social < schema.surqlA SCHEMAFULL table rejects any field it doesn’t define, including keys nested inside an object field. The SDK’s delivery references, outcomes, and workflows are JSON whose keys can change between SDK versions, so those fields are FLEXIBLE. Tokens do not belong in these tables. Connection attempts and stored OAuth credentials hold secrets too, such as PKCE verifiers, provider state, and refresh tokens. If you store them in SurrealDB, keep them in separate tables scoped to the tenant and principal, encrypt the credentials, and use compare-and-set revisions when you refresh them, as authentication describes.
Each delivery record will get an array ID, delivery:[tenant, draftId]. The ID is unique per tenant and draft, so the delivery table needs no separate unique index.
Durable stores
Nothing the SDK needs later can live in process memory, because a restart or a second server instance would lose it. Two stores need SurrealDB:
- The client’s
idempotencyStorestops a retried request from posting twice.MemoryIdempotencyStoreonly covers one process. - The Threads adapter’s
workflowStorekeeps each post’s progress through the Threads API calls, so a later call can continue it. The defaultMemoryThreadsWorkflowStoreloses the post when the process exits.
import { RecordId, Surreal, surql } from "surrealdb";
import type { DeliveryOutcome, IdempotencyStore } from "@opencoredev/social-sdk";
import type { ThreadsWorkflow, ThreadsWorkflowStore } from "@opencoredev/social-sdk/threads";
interface ClaimRow {
id: RecordId<"idempotency_claim", string>;
fingerprint: string;
target_keys: string[];
outcomes: Record<string, DeliveryOutcome>;
}
export function surrealIdempotencyStore(db: Surreal): IdempotencyStore {
return {
async claim({ scope, key, fingerprint, targetKeys }) {
// The unique index on (scope, key) makes INSERT IGNORE return nothing for a duplicate.
const [inserted] = await db
.query<[ClaimRow[]]>(
surql`INSERT IGNORE INTO idempotency_claim
{ scope: ${scope}, key: ${key}, fingerprint: ${fingerprint}, target_keys: ${targetKeys} }`,
)
.collect();
if (inserted[0]) return { kind: "new", claimId: inserted[0].id.id, outcomes: {} };
const [rows] = await db
.query<[ClaimRow[]]>(
surql`SELECT * FROM idempotency_claim WHERE scope = ${scope} AND key = ${key}`,
)
.collect();
const row = rows[0]!;
if (
row.fingerprint !== fingerprint ||
JSON.stringify(row.target_keys) !== JSON.stringify(targetKeys)
)
return { kind: "conflict" };
return { kind: "existing", claimId: row.id.id, outcomes: row.outcomes };
},
async saveOutcome({ claimId, targetKey, outcome }) {
// MERGE is a deep merge, so outcomes saved for other targets stay put.
await db
.update(new RecordId("idempotency_claim", claimId))
.merge({ outcomes: { [targetKey]: outcome } });
},
};
}
export function surrealThreadsWorkflowStore(db: Surreal): ThreadsWorkflowStore {
const record = (id: string) => new RecordId("threads_workflow", id);
const leases = new Map<string, string>();
return {
async create(input) {
const workflow: ThreadsWorkflow = { ...input, id: crypto.randomUUID() };
await db.create(record(workflow.id)).content({ workflow });
return workflow;
},
async get(id) {
const row = await db.select<{ workflow: ThreadsWorkflow }>(record(id));
return row?.workflow;
},
async update(id, update) {
const [rows] = await db
.query<[{ workflow: ThreadsWorkflow }[]]>(
surql`UPDATE ${record(id)} MERGE { workflow: ${update} }`,
)
.collect();
if (!rows[0]) throw new Error("Threads workflow not found");
return rows[0].workflow;
},
// A lease, not a lock, so a claim held by a crashed process expires on its own.
// The token lets release clear only this process's lease, never a newer one.
async claim(id) {
if (leases.has(id)) return false;
const token = crypto.randomUUID();
const [rows] = await db
.query<[unknown[]]>(
surql`UPDATE ${record(id)} SET claimed_until = time::now() + 5m, claimed_by = ${token}
WHERE claimed_until = NONE OR claimed_until < time::now()`,
)
.collect();
if (rows.length !== 1) return false;
leases.set(id, token);
return true;
},
async release(id) {
const token = leases.get(id);
if (!token) return;
leases.delete(id);
await db
.query(
surql`UPDATE ${record(id)} SET claimed_until = NONE, claimed_by = NONE
WHERE claimed_by = ${token}`,
)
.collect();
},
};
}In SurrealDB 3, UPDATE on a missing record returns an empty result instead of creating it, which is how update detects a missing workflow and how claim detects a lease someone else holds. Keep the lease longer than the retry budget you pass to resumePublication below, so a claim can’t expire while its process is still writing to Threads.
Publish
posts.publish stores a Threads workflow, claims it, and makes the Threads API calls before it returns. The outcome is published when Threads finishes, processing when media is still being processed or another process holds the workflow’s claim, and unknown when a write may or may not have reached Threads. For processing, publishDraft calls the adapter’s native resumePublication once more, so most posts finish before it returns. Call it from your route handler after you have verified the tenant.
import { RecordId, surql } from "surrealdb";
import { createSocial, connectedAccountRef, type DeliveryRef } from "@opencoredev/social-sdk";
import { threads } from "@opencoredev/social-sdk/threads";
import { db } from "./db";
import { surrealIdempotencyStore, surrealThreadsWorkflowStore } from "./stores";
const threadsUserId = process.env.THREADS_USER_ID!;
export const social = createSocial({
backend: threads({
auth: {
userId: threadsUserId,
accessToken: process.env.THREADS_ACCESS_TOKEN!,
},
workflowStore: surrealThreadsWorkflowStore(db),
}),
idempotencyStore: surrealIdempotencyStore(db),
});
// Native calls skip the client's tenant checks, so callers authorize the account first.
const native = social.native("default", { acknowledgeUnsafe: true })!;
interface DeliveryRow {
id: RecordId<"delivery">;
tenant: string;
state: string;
delivery_ref?: DeliveryRef;
}
export async function advance(tenantId: string, delivery: RecordId<"delivery">, ref: DeliveryRef) {
const [grants] = await db
.query<[{ id: RecordId<"connected_account"> }[]]>(
surql`SELECT id FROM connected_account
WHERE tenant = ${tenantId} AND backend = ${ref.backend}
AND platform = ${ref.platform} AND account_id = ${ref.accountId}
LIMIT 1`,
)
.collect();
if (!grants[0]) throw new Error("Account grant revoked");
const account = connectedAccountRef({
backend: ref.backend,
platform: "threads",
accountId: ref.accountId,
});
const outcome = await native.resumePublication(account, ref.deliveryId, {
backendInstance: ref.backend,
correlationId: String(delivery),
retryBudget: { maxAttempts: 3, maxElapsedMs: 60_000 },
});
await db
.query(
surql`UPDATE ${delivery} SET state = ${outcome.state}
WHERE state NOTINSIDE ["published", "failed"]`,
)
.collect();
return outcome;
}
export async function publishDraft(tenantId: string, draftId: string, text: string) {
const [accounts] = await db
.query<[{ id: RecordId<"connected_account">; account_id: string; backend: string }[]]>(
// This client only holds credentials for THREADS_USER_ID, so only that account can publish.
surql`SELECT id, account_id, backend FROM connected_account
WHERE tenant = ${tenantId} AND platform = "threads" AND account_id = ${threadsUserId}
LIMIT 1`,
)
.collect();
const acct = accounts[0];
if (!acct) throw new Error("No authorized Threads account");
const result = await social.posts.publish(
{
targets: [
{
account: connectedAccountRef({
backend: acct.backend,
platform: "threads",
accountId: acct.account_id,
}),
},
],
content: { text },
idempotencyKey: draftId,
},
// Scopes the idempotency key to this tenant, so tenants can't collide on a draft ID.
{ authorization: { tenantId } },
);
const outcome = result.outcomes[0];
// One record per tenant and draft. A concurrent duplicate can report unknown with no
// delivery reference; it must not replace a resumable row or undo a final state.
const [saved] = await db
.query<[DeliveryRow[]]>(
surql`UPSERT ${new RecordId("delivery", [tenantId, draftId])} SET
tenant = ${tenantId},
idempotency_key = ${draftId},
account = ${acct.id},
state = IF state INSIDE ["published", "failed"] { state }
ELSE IF ${outcome?.delivery} = NONE AND delivery_ref != NONE { state }
ELSE { ${outcome?.state ?? "unknown"} },
delivery_ref = ${outcome?.delivery} ?? delivery_ref`,
)
.collect();
const row = saved[0]!;
const state =
row.state === "processing" && row.delivery_ref
? (await advance(tenantId, row.id, row.delivery_ref)).state
: row.state;
return { status: result.status, state };
}This client holds one Threads account’s token, so it can only publish as THREADS_USER_ID, and the query only accepts that account. To publish for several accounts, load each account’s credentials from encrypted storage and build a client for that account.
Validate draftId and text before calling publishDraft. Threads caps a post at 500 characters. A second call with the same draft ID and text returns the saved result without posting again; the same draft ID with different text is rejected as an idempotency conflict.
Reconcile
A delivery can still be processing when publishDraft returns: Threads may not have finished processing the media, or the server may have restarted partway through. reconcile resumes those workflows.
export async function reconcile() {
// Skip deliveries whose connected account was deleted or now belongs to another tenant.
const [rows] = await db
.query<[DeliveryRow[]]>(
surql`SELECT id, tenant, state, delivery_ref FROM delivery
WHERE state = "processing" AND record::exists(account) AND account.tenant = tenant
AND account.backend = delivery_ref.backend
AND account.platform = delivery_ref.platform
AND account.account_id = delivery_ref.accountId
LIMIT 20`,
)
.collect();
for (const row of rows) if (row.delivery_ref) await advance(row.tenant, row.id, row.delivery_ref);
return rows.length;
}Run it every few minutes from a scheduler such as Trigger.dev, Inngest, Restate, or a cron job. record::exists(account) checks that the linked connected_account record still exists, and account.tenant = tenant checks that it still belongs to the delivery’s tenant. A connected_account record represents an active grant here; delete it when access is revoked. The query also matches the reference’s backend, platform, and account ID. advance repeats the full grant check before every native continuation, including the one after publishing. Native calls bypass any SDK AuthorizationPolicy. The workflow store’s claim stops two processes from resuming the same post at once, so an overlapping run does no harm. resumePublication and posts.getDelivery both continue a workflow that is still processing, from the stage it reached. A workflow whose state is unknown is never replayed: both report unknown again. Leave those rows for review, as the idempotency guide describes.
Webhooks
Verify the raw request body before parsing it, then store the event in social_event with a dedupe key as its record ID. INSERT IGNORE returns an empty result when that ID already exists, so a redelivered event is stored once:
const [inserted] = await db
.query<[unknown[]]>(
surql`INSERT IGNORE INTO social_event
{ id: ${dedupeKey}, platform: "threads", payload: ${payload} }`,
)
.collect();
const isNew = inserted.length === 1;Build dedupeKey from the platform and the event’s own ID, following the events guide, and process rows where processed_at = NONE from the same scheduler that runs reconcile.
Live delivery updates
A live query pushes each change to a delivery record to your server as it happens, so a dashboard can show processing turning into published without polling. Filter by tenant so one tenant never receives another’s records:
import { Table, eq } from "surrealdb";
import { db } from "./db";
export async function watchDeliveries(
tenantId: string,
onChange: (delivery: { idempotency_key: string; state: string }) => void,
) {
const live = await db
.live(new Table("delivery"))
.fields("idempotency_key", "state")
.where(eq("tenant", tenantId));
live.subscribe((message) => {
if (message.action === "CREATE" || message.action === "UPDATE")
onChange(message.value as { idempotency_key: string; state: string });
});
return () => live.kill();
}Forward onChange to the browser over your own server-sent events or WebSocket route. Call the returned function when the client disconnects.
Notes
- Some SurrealDB examples pass the
subscribecallback three arguments. Insurrealdb2.0.8 it receives one message withaction,value, andrecordId, as shown above. - The connection option
retrydoesn’t retry ordinary queries. It only sets the defaults for queries you mark with.retry(). Without it, a write conflict under heavy load surfaces as an error;publishDraftis safe to call again with the same draft ID. - The SurrealDB details on this page come from SurrealDB’s documentation as of September 22, 2026: connecting, authentication, executing queries, live queries,
DEFINE USER,DEFINE FIELD,INSERT,UPSERT,UPDATE, and record IDs. - The samples ran against SurrealDB 3.2.4 with
surrealdb2.0.8, connected as anEDITORdatabase user, with a stubbed Threads API. They have not been run against a live Threads account.